initial commit0.1

14 files changed, 1915 insertions, 0 deletions

diff --git a/ioreplay/src/replay/replay.c b/ioreplay/src/replay/replay.c
new file mode 100644
index 0000000..89f5fee
--- /dev/null
+++ b/ioreplay/src/replay/replay.c
@@ -0,0 +1,191 @@
+// Copyright 2018 Mimecast Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "replay.h"
+
+#include "../datas/amap.h"
+#include "../meta/meta.h"
+#include "../mounts.h"
+#include "rworker.h"
+#include "rstats.h"
+
+void replay_extract_header(options_s *opts, FILE *replay_fd, long *num_vsizes,
+                           long *num_pids, long *num_fds, long *num_lines)
+{
+    meta_s *m = meta_new(replay_fd);
+    meta_read_start(m);
+
+    long version = 0;
+    if (meta_read_l(m, "version", &version)) {
+        Put("Replay version is '%ld'", version);
+        if (version != REPLAY_VERSION) {
+            Error(".replay file of incompatible version, got %x, expected %x",
+                  (int)version, REPLAY_VERSION);
+        }
+    }
+
+    char *user;
+    if (meta_read_s(m, "user", &user)) {
+        Put("Setting user to '%s'", user);
+        opts->user = user;
+    }
+
+    char *name;
+    if (meta_read_s(m, "name", &name)) {
+        Put("Setting name to '%s'", name);
+        opts->name = name;
+    }
+
+    if (meta_read_l(m, "num_vsizes", num_vsizes)) {
+        if (*num_vsizes < 0) {
+            Error("Lamport vsize overflow");
+        }
+        Put("Setting num of vsizes to '%ld'", *num_vsizes);
+    }
+
+    if (meta_read_l(m, "num_mapped_pids", num_pids)) {
+        if (*num_pids < 0) {
+            Error("Process overflow (too many process IDs in .replay)");
+        }
+        Put("Setting num of PIDs to '%ld'", *num_pids);
+    }
+
+    if (meta_read_l(m, "num_mapped_fds", num_fds)) {
+        if (*num_fds < 0) {
+            Error("FD overflow (too many FDs in .replay)");
+        }
+        Put("Setting num of FDs to '%ld'", *num_fds);
+    }
+
+    if (meta_read_l(m, "num_lines", num_lines)) {
+        if (*num_fds < 0) {
+            Error("Overflow (too many lines in .replay)");
+        }
+        Put("Setting num of lines to '%ld'", *num_lines);
+    }
+
+    meta_destroy(m);
+}
+
+status_e replay_run(options_s *opts)
+{
+    status_e status = SUCCESS;
+
+    if (opts->drop_caches) {
+        drop_caches();
+        //cache_file(opts->replay_file);
+    }
+
+    // Extract information from the meta header
+    FILE *replay_fd = Fopen(opts->replay_file, "r");
+    long num_vsizes = 0, num_pids = 0, num_fds = 0, num_lines = 0;
+    replay_extract_header(opts, replay_fd, &num_vsizes, &num_pids,
+                          &num_fds, &num_lines);
+    fclose(replay_fd);
+
+    // A map of all file descriptors used.
+    Out("Creating FD map...");
+    amap_s *fds_map = NULL;
+    if (opts->num_workers > 1) {
+        fds_map = amap_new_mmapped(num_fds);
+    } else {
+        fds_map = amap_new(num_fds);
+    }
+    Put("done");
+
+    // To collect all individual worker's stats into the global
+    // stats object.
+    stack_s *all_worker_stats = stack_new();
+
+    // The global stats object
+    rstats_s *stats = rstats_new(opts);
+    rstats_start(stats);
+
+    // Fork worker processes, each worker process will read the .replay file
+    // individually.
+
+    if (opts->num_workers > 1) {
+        for (int i = 0; i < opts->num_workers; ++i) {
+            rworker_stats_s *worker_stats = rworker_stats_new_mmap();
+            stack_push(all_worker_stats, worker_stats);
+
+            pid_t pid = fork();
+
+            if (pid == 0) {
+                // One worker object per fork
+                rworker_s *w = rworker_new(i, fds_map, num_vsizes, num_pids, opts,
+                                           worker_stats);
+
+                // Process the .replay journal line by line
+                status_e status = rworker_process_lines(w, num_lines);
+                Put("worker(%d): Exiting from %d with status %d", i,
+                    pid, status);
+                rworker_destroy(w);
+
+                // Exit sub-process
+                exit(status);
+
+            } else if (pid < 0) {
+                Errno("worker(%d): Unable to create worker process! :'-(", i);
+
+            } else {
+                Put("worker(%d): Process with pid %d forked", i, pid);
+            }
+        }
+
+        drop_root(opts->user);
+
+        Put("Waiting for worker processes to finish");
+        pid_t pid;
+        int rworker_status = SUCCESS;
+
+       while ((pid = wait(&rworker_status)) > 0) {
+            if (rworker_status != SUCCESS)
+                status = rworker_status;
+
+            Put("Process with pid %d exited with status %d",
+                pid, rworker_status);
+        }
+
+        Put("All workers finished (%d)!", status);
+
+    } else {
+        Put("Only one worker, don't fork sub-processes");
+
+        rworker_stats_s *worker_stats = rworker_stats_new_mmap();
+        stack_push(all_worker_stats, worker_stats);
+
+        rworker_s *w = rworker_new(0, fds_map, num_vsizes, num_pids,
+                                   opts, worker_stats);
+        status = rworker_process_lines(w, num_lines);
+        rworker_destroy(w);
+
+        Put("Worker finished work!");
+    }
+
+    // Collect all statistics
+    rstats_stop(stats);
+    while (!stack_is_empty(all_worker_stats)) {
+        rworker_stats_s *worker_stats = stack_pop(all_worker_stats);
+        rstats_add_from_worker(stats, worker_stats);
+        rworker_stats_destroy(worker_stats);
+    }
+    stack_destroy(all_worker_stats);
+
+    rstats_print(stats);
+    rstats_destroy(stats);
+
+    amap_destroy(fds_map);
+    return status;
+}
diff --git a/ioreplay/src/replay/replay.h b/ioreplay/src/replay/replay.h
new file mode 100644
index 0000000..dcc3d84
--- /dev/null
+++ b/ioreplay/src/replay/replay.h
@@ -0,0 +1,46 @@
+// Copyright 2018 Mimecast Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef REPLAY_H
+#define REPLAY_H
+
+#include "../defaults.h"
+#include "../utils/futils.h"
+#include "../opcodes.h"
+#include "../options.h"
+#include "rioop.h"
+#include "rprocess.h"
+
+/**
+ * @brief Replays the given .replay file
+ *
+ * @param opts The options object
+ * @return SUCCESS if everything went fine
+ */
+status_e replay_run(options_s *opts);
+
+/**
+ * @brief Extract required meta data from .replay's meta header
+ *
+ * @param opts The options object
+ * @param replay_fd The file handle to the .replay file
+ * @param num_vsizes The amount of virtual sizes/paths
+ * @param num_pids The amount of process IDs
+ * @param num_fds The amount of virtual file descriptors
+ * @param num_lines The amount of .replay lines with I/O ops
+ */
+void replay_extract_header(options_s *opts, FILE *replay_fd, long *num_vsizes,
+                           long *num_pids, long *num_fds,long *num_lines);
+
+#endif // REPLAY_H
diff --git a/ioreplay/src/replay/rioop.c b/ioreplay/src/replay/rioop.c
new file mode 100644
index 0000000..2e16c94
--- /dev/null
+++ b/ioreplay/src/replay/rioop.c
@@ -0,0 +1,425 @@
+// Copyright 2018 Mimecast Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "rioop.h"
+
+#include "../vfd.h"
+#include "rworker.h"
+
+// Printing error messages
+#define _Error(...) \
+  fprintf(stderr, "%s:%d ERROR: ", __FILE__, __LINE__); \
+  fprintf(stderr, __VA_ARGS__); \
+  fprintf(stderr, "\nlineno:%ld path:%s\n", task->lineno, vfd->path); \
+  fflush(stdout); \
+  fflush(stderr); \
+  exit(ERROR);
+
+#define _Errno(...) \
+  fprintf(stderr, "%s:%d ERROR: %s (%d). ", __FILE__, __LINE__, \
+      strerror(errno), errno); \
+  fprintf(stderr, __VA_ARGS__); \
+  fprintf(stderr, "\nlineno:%ld path:%s\n", task->lineno, vfd->path); \
+  fflush(stdout); \
+  fflush(stderr); \
+  exit(ERROR);
+
+#define _Init_arg(num) int arg = atoi(task->toks[num])
+#define _Init_cmd(num) int cmd = atoi(task->toks[num])
+#define _Init_fd(num) long fd = atol(task->toks[num])
+#define _Init_flags(num) int flags = atoi(task->toks[num])
+//#define _Init_mode(num) int mode = atoi(task->toks[num])
+#define _Init_offset(num) long offset = atol(task->toks[num])
+#define _Init_op(num) int op = atoi(task->toks[num])
+#define _Init_path2(num) char *path2 = task->toks[num]
+#define _Init_path(num) char *path = task->toks[num]
+#define _Init_rc(num) int rc = atoi(task->toks[num])
+#define _Init_whence(num) long whence = atol(task->toks[num])
+
+#define _Init_bytes(num) \
+    int bytes = atoi(task->toks[num]); \
+    if (bytes <= 0) return
+
+#define _Init_virtfd \
+    vfd_s *vfd = amap_get(p->fds_map, fd); \
+    if (vfd == NULL) return
+
+void rioop_run(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_op(2);
+
+    switch (op) {
+    // stat() syscalls
+    case FSTAT:
+        rioop_fstat(p, t, task);
+        break;
+    case FSTATFS:
+    case FSTATFS64:
+        //Error("op(%d) not implemented", op);
+        break;
+    case FSTAT_AT:
+    case LSTAT:
+    case STAT:
+        rioop_stat(p, t, task);
+        break;
+    case STATFS:
+    case STATFS64:
+        //Error("op(%d) not implemented", op);
+        break;
+
+    // read() syscalls
+    case READ:
+    case READV:
+        rioop_read(p, t, task);
+        break;
+    case READAHEAD:
+        //Error("op(%d) not implemented", op);
+        break;
+    case READLINK:
+    case READLINK_AT:
+        //Error("op(%d) not implemented", op);
+        break;
+
+    // write() syscalls
+    case WRITE:
+    case WRITEV:
+        rioop_write(p, t, task);
+        break;
+
+    // open() and other syscalls which may creat
+    case OPEN:
+    case OPEN_AT:
+        rioop_open(p, t, task, -1);
+        break;
+    case CREAT:
+        // A call to crat() is equivalent to calling open() with flags..
+        rioop_open(p, t, task, O_CREAT|O_WRONLY|O_TRUNC);
+        break;
+    case MKDIR:
+    case MKDIR_AT:
+        rioop_mkdir(p, t, task);
+        break;
+
+    // rename() syscalls
+    case RENAME:
+    case RENAME_AT:
+    case RENAME_AT2:
+        rioop_rename(p, t, task);
+        break;
+
+    // close() and unlink() syscalls
+    case CLOSE:
+        rioop_close(p, t, task);
+        break;
+    case UNLINK:
+    case UNLINK_AT:
+        rioop_unlink(p, t, task);
+        break;
+    case RMDIR:
+        rioop_rmdir(p, t, task);
+        break;
+
+    // sync() syscalls
+    case FSYNC:
+        rioop_fsync(p, t, task);
+        break;
+    case FDATASYNC:
+        rioop_fdatasync(p, t, task);
+        break;
+    case SYNC:
+    case SYNCFS:
+    case SYNC_FILE_RANGE:
+        //Error("op(%d) not implemented", op);
+        break;
+
+    // Other syscalls
+    case FCNTL:
+        rioop_fcntl(p, t, task);
+        break;
+    case GETDENTS:
+        rioop_getdents(p, t, task);
+        break;
+    case LSEEK:
+        rioop_lseek(p, t, task);
+        break;
+
+    // chmod() syscalls
+    case CHMOD:
+        rioop_chmod(p, t, task);
+        break;
+    case FCHMOD:
+        rioop_fchmod(p, t, task);
+        break;
+
+    // chown() syscalls
+    case CHOWN:
+        rioop_chown(p, t, task);
+        break;
+    case FCHOWN:
+    case FCHOWNAT:
+        rioop_fchown(p, t, task);
+        break;
+    case LCHOWN:
+        rioop_lchown(p, t, task);
+        break;
+
+    // Meta operations (I/O replay internal use only).
+    case META_EXIT_GROUP:
+        break;
+    case META_TIMELINE:
+        break;
+
+    default:
+        Error("op(%d) not implemented", op);
+        break;
+    }
+}
+
+void rioop_stat(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_path(3);
+    struct stat buf;
+    stat(path, &buf);
+}
+
+void rioop_fstat(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_fd(3);
+    _Init_virtfd;
+    struct stat buf;
+    fstat(vfd->fd, &buf);
+}
+
+void rioop_rename(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_path(3);
+    _Init_path2(4);
+    rename(path, path2);
+}
+
+void rioop_read(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_fd(3);
+    _Init_bytes(4);
+    _Init_virtfd;
+
+    char *buf = Calloc(bytes+1, char);
+    read(vfd->fd, buf, bytes);
+    free(buf);
+}
+
+void rioop_write(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_fd(3);
+    _Init_bytes(4);
+    _Init_virtfd;
+
+    char *buf = Calloc(bytes+1, char);
+    sprintf(buf, "%ld", task->lineno);
+    Fill_with_stuff(buf, bytes);
+    if (vfd->fd == 0) {
+        Debug("%d %d %ld", vfd->fd, vfd->debug, task->lineno);
+        _Error("ERROR");
+    }
+    write(vfd->fd, buf, bytes);
+    free(buf);
+}
+
+void rioop_open(rprocess_s *p, rthread_s *t, rtask_s *task, int flags_)
+{
+    _Init_fd(3);
+    _Init_path(4);
+    _Init_flags(6);
+
+    // Special case as this is creat() now
+    if (flags_ != -1)
+        flags = flags_;
+
+    bool directory = Has(flags, O_DIRECTORY);
+
+    if (fd > 0) {
+        if (directory) {
+            // We can not open a directory via open() otherwise!
+            flags &= (O_RDONLY & ~(O_RDWR|O_WRONLY|O_CREAT));
+        } else {
+            // We don't want to open the file in read only mode.
+            // SystemTap could have skipped syscalls to fcntl or open
+            flags &= ~O_RDONLY;
+        }
+        //    flags |= O_DIRECT|O_SYNC;
+        flags &= ~O_EXCL;
+    }
+
+    int ret = open(path, flags, S_IRWXU|S_IRWXG|S_IRWXO);
+
+    if (fd < 0 && ret > 0) {
+        close(ret);
+#ifdef THREAD_DEBUG
+        fprintf(t->rthread_fd, "TRACE OPEN|open+close|%s|\n", path);
+        fflush(t->rthread_fd);
+#endif
+    }
+
+    if (fd > 0 && ret > 0) {
+        vfd_s *vfd = vfd_new(ret, fd, path);
+        amap_set(p->fds_map, fd, vfd);
+
+#ifdef THREAD_DEBUG
+        fprintf(t->rthread_fd, "TRACE OPEN|open|%s|\n", path);
+        fflush(t->rthread_fd);
+#endif
+    }
+}
+
+void rioop_close(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_fd(3);
+    _Init_virtfd;
+
+    amap_unset(p->fds_map, fd);
+    if (vfd->dirfd) {
+        closedir(vfd->dirfd);
+#ifdef THREAD_DEBUG
+        fprintf(t->rthread_fd, "TRACE OPEN|closedir|%s|\n", vfd->path);
+        fflush(t->rthread_fd);
+#endif
+    } else {
+        close(vfd->fd);
+#ifdef THREAD_DEBUG
+        fprintf(t->rthread_fd, "TRACE OPEN|close|%s|\n", vfd->path);
+        fflush(t->rthread_fd);
+#endif
+    }
+    vfd_destroy(vfd);
+}
+
+void rioop_getdents(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_fd(3);
+    _Init_virtfd;
+
+    // getdents expects a dirfd
+    DIR *dirfd = fdopendir(vfd->fd);
+    if (dirfd) {
+        vfd->dirfd = dirfd;
+        readdir(dirfd);
+#ifdef THREAD_DEBUG
+        fprintf(t->rthread_fd, "TRACE OPEN|fdopendir|%s|\n", vfd->path);
+        fflush(t->rthread_fd);
+#endif
+    }
+}
+
+void rioop_mkdir(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_path(3);
+    mkdir(path, S_IRWXU|S_IRWXG|S_IRWXO);
+}
+
+void rioop_unlink(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_path(3);
+    unlink(path);
+}
+
+void rioop_rmdir(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_path(3);
+    rmdir(path);
+}
+
+void rioop_lseek(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_fd(3);
+    _Init_bytes(6);
+    _Init_virtfd;
+    lseek(vfd->fd, bytes, SEEK_SET);
+}
+
+void rioop_fsync(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_fd(3);
+    _Init_virtfd;
+    fsync(vfd->fd);
+}
+
+void rioop_fdatasync(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_fd(3);
+    _Init_virtfd;
+    fdatasync(vfd->fd);
+}
+
+void rioop_fcntl(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_fd(3);
+    _Init_cmd(4);
+    _Init_arg(5);
+    _Init_virtfd;
+
+    switch (cmd) {
+    case F_GETFD:
+    case F_GETFL:
+        fcntl(vfd->fd, cmd);
+        break;
+    case F_SETFD:
+    case F_SETFL:
+        fcntl(vfd->fd, cmd, arg);
+        break;
+    default:
+        break;
+    }
+}
+
+void rioop_chmod(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_path(3);
+    chmod(path, S_IRWXU|S_IRWXG|S_IRWXO);
+}
+
+void rioop_fchmod(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_fd(3);
+    _Init_virtfd;
+    fchmod(vfd->fd, S_IRWXU|S_IRWXG|S_IRWXO);
+}
+
+void rioop_chown(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_path(3);
+    rworker_s *w = t->worker;
+    options_s *opts = w->opts;
+    struct passwd *pwd = getpwnam(opts->user);
+    chown(path, pwd->pw_uid, -1);
+}
+
+void rioop_fchown(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_fd(3);
+    _Init_virtfd;
+    rworker_s *w = t->worker;
+    options_s *opts = w->opts;
+    struct passwd *pwd = getpwnam(opts->user);
+    fchown(vfd->fd, pwd->pw_uid, -1);
+}
+
+void rioop_lchown(rprocess_s *p, rthread_s *t, rtask_s *task)
+{
+    _Init_path(3);
+    rworker_s *w = t->worker;
+    options_s *opts = w->opts;
+    struct passwd *pwd = getpwnam(opts->user);
+    lchown(path, pwd->pw_uid, -1);
+}
+
diff --git a/ioreplay/src/replay/rioop.h b/ioreplay/src/replay/rioop.h
new file mode 100644
index 0000000..4db4284
--- /dev/null
+++ b/ioreplay/src/replay/rioop.h
@@ -0,0 +1,54 @@
+// Copyright 2018 Mimecast Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef RIOOP_H
+#define RIOOP_H
+
+#include "../defaults.h"
+#include "../utils/futils.h"
+#include "../opcodes.h"
+#include "rprocess.h"
+#include "rthread.h"
+
+/**
+ * @brief Replays the responsible I/O operation of a given task
+ *
+ * @param p The virtual replay process object
+ * @param t The thread object
+ * @param task The replay task object
+ */
+void rioop_run(rprocess_s *p, rthread_s *t, rtask_s *task);
+
+void rioop_close(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_fcntl(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_fdatasync(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_fstat(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_fsync(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_getdents(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_mkdir(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_open(rprocess_s *p, rthread_s *t, rtask_s *task, int flags_);
+void rioop_read(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_rename(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_stat(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_lseek(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_unlink(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_rmdir(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_write(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_chmod(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_fchmod(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_chown(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_fchown(rprocess_s *p, rthread_s *t, rtask_s *task);
+void rioop_lchown(rprocess_s *p, rthread_s *t, rtask_s *task);
+
+#endif // RIOOP_H
diff --git a/ioreplay/src/replay/rprocess.c b/ioreplay/src/replay/rprocess.c
new file mode 100644
index 0000000..4efd835
--- /dev/null
+++ b/ioreplay/src/replay/rprocess.c
@@ -0,0 +1,34 @@
+// Copyright 2018 Mimecast Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "rprocess.h"
+
+rprocess_s* rprocess_new(const int pid, amap_s *fds_map)
+{
+    rprocess_s *p = Malloc(rprocess_s);
+
+    p->fds_map = fds_map;
+    p->pid = pid;
+    p->terminate = 0;
+    p->lineno = 0;
+
+    return p;
+}
+
+void rprocess_destroy(rprocess_s *p)
+{
+    if (!p)
+        return;
+    free(p);
+}
diff --git a/ioreplay/src/replay/rprocess.h b/ioreplay/src/replay/rprocess.h
new file mode 100644
index 0000000..739dd89
--- /dev/null
+++ b/ioreplay/src/replay/rprocess.h
@@ -0,0 +1,40 @@
+// Copyright 2018 Mimecast Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef RPROCESS_H
+#define RPROCESS_H
+
+#include "../datas/hmap.h"
+#include "../datas/amap.h"
+#include "../defaults.h"
+#include "rthread.h"
+
+/**
+ * @brief The virtual replay process object definition
+ *
+ * This defines a virtual process in replay context.
+ */
+typedef struct rprocess_s_ {
+    int terminate; /**< Indicates whether the worker is terminating or not */
+    int rworker_num; /**< The worker number of the responsible worker */
+    int pid; /**< The virtual process ID */
+    unsigned long lineno; /**< Holding the current .replay line number */
+    bool initm; /**< Indicates whether ioreplay is in init mode or not */
+    amap_s *fds_map; /**< Holding all file descriptors */
+} rprocess_s;
+
+rprocess_s* rprocess_new(const int pid, amap_s *fds_map);
+void rprocess_destroy(rprocess_s* p);
+
+#endif // RPROCESS_H
diff --git a/ioreplay/src/replay/rstats.c b/ioreplay/src/replay/rstats.c
new file mode 100644
index 0000000..c3e6e38
--- /dev/null
+++ b/ioreplay/src/replay/rstats.c
@@ -0,0 +1,108 @@
+// Copyright 2018 Mimecast Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "rstats.h"
+
+#include <sys/types.h>
+
+rstats_s* rstats_new(options_s *opts)
+{
+    rstats_s *s = Malloc(rstats_s);
+
+    s->opts = opts;
+    s->loadavg_high = 0;
+    s->ioops = 0;
+    s->duration = 0;
+    s->time_ahead = -1;
+
+    if (opts->stats_file)
+        s->stats_fd = Fopen(opts->stats_file, "w");
+    else
+        s->stats_fd = stdout;
+
+    return s;
+}
+
+void rstats_destroy(rstats_s *s)
+{
+    if (s->stats_fd != stdout)
+        fclose(s->stats_fd);
+
+    free(s);
+}
+
+rworker_stats_s* rworker_stats_new_mmap(options_s *opts)
+{
+    // Share this object between processes, so that the stats cann be
+    // collected by the master process!
+    rworker_stats_s *s = Mmapshared(rworker_stats_s);
+
+    s->loadavg_high = 0;
+    s->ioops = 0;
+    s->time_ahead = -1;
+
+    return s;
+}
+
+void rworker_stats_destroy(rworker_stats_s *s)
+{
+    munmap(s, sizeof(rworker_stats_s));
+}
+
+
+void rstats_start(rstats_s* s)
+{
+    gettimeofday(&s->start_time, NULL);
+}
+
+void rstats_stop(rstats_s* s)
+{
+    gettimeofday(&s->end_time, NULL);
+    s->duration= ((s->end_time.tv_sec - s->start_time.tv_sec) * 1000
+                   + (s->end_time.tv_usec - s->start_time.tv_usec) / 1000) / 1000;
+
+}
+
+void rstats_add_from_worker(rstats_s* s, rworker_stats_s* w)
+{
+    if (s->loadavg_high < w->loadavg_high)
+        s->loadavg_high = w->loadavg_high;
+
+    if (s->time_ahead == -1 || s->time_ahead > w->time_ahead)
+        s->time_ahead = w->time_ahead;
+
+    s->ioops += w->ioops;
+}
+
+void rstats_print(rstats_s* s)
+{
+    options_s *opts = s->opts;
+
+    if (opts->stats_file) {
+        Put("Writing stats to '%s'", opts->stats_file);
+    }
+
+    fprintf(s->stats_fd, "Stats of test '%s':\n", opts->name);
+    fprintf(s->stats_fd, "\tNum workers: %d\n", opts->num_workers);
+    fprintf(s->stats_fd, "\tThreads per worker: %d\n", opts->num_threads_per_worker);
+    fprintf(s->stats_fd, "\tThreads total: %d\n",
+            opts->num_threads_per_worker * opts->num_workers);
+    fprintf(s->stats_fd, "\tHighest loadavg: %.2f\n", s->loadavg_high);
+    fprintf(s->stats_fd, "\tPerformed ioops: %ld\n", s->ioops);
+    if (s->duration > 0)
+        fprintf(s->stats_fd, "\tAverage ioops/s: %.2f\n", s->ioops/s->duration);
+    fprintf(s->stats_fd, "\tTime ahead: %lds\n", s->time_ahead/1000);
+    fprintf(s->stats_fd, "\tTotal time: %.2fs\n", s->duration);
+}
+
diff --git a/ioreplay/src/replay/rstats.h b/ioreplay/src/replay/rstats.h
new file mode 100644
index 0000000..1ce3f27
--- /dev/null
+++ b/ioreplay/src/replay/rstats.h
@@ -0,0 +1,117 @@
+// Copyright 2018 Mimecast Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/**
+ * @file rstats.h
+ * @author Paul Buetow
+ *
+ * @brief For collecting replay stats
+ */
+
+#ifndef RSTATS_H
+#define RSTATS_H
+
+#include "../defaults.h"
+#include "../options.h"
+
+#include <pthread.h>
+
+/**
+ * @brief Definition of the rstats object
+ *
+ * Used to store global statistics.
+ */
+typedef struct rstats_s_ {
+    double loadavg_high; /**< Highest load average */
+    long ioops; /**< Total amount if io operations */
+    double duration; /**< Duration of the test */
+    long time_ahead; /**< Time ahead of the original speed */
+    struct timeval start_time; /**< Start time of the test */
+    struct timeval end_time; /**< End time of the test */
+    options_s *opts; /**< The I/O replay options object */
+    FILE *stats_fd; /**< The file descriptor for writing the stats */
+} rstats_s;
+
+/**
+ * @brief Definition of the per worker stats object
+ *
+ * Used to store per worker process I/O stats
+ */
+typedef struct rworker_stats_s_ {
+    double loadavg_high; /**< Highest amount of io ops per second */
+    long ioops; /**< Total amount if io operations */
+    long time_ahead; /**< Time ahead of the original speed */
+} rworker_stats_s;
+
+/**
+ * @brief Creates a new stats object
+ *
+ * @return The new stats object
+ */
+rstats_s* rstats_new(options_s *opts);
+
+/**
+ * @brief Destroys the stats object
+ *
+ * @param s The stats object
+ */
+void rstats_destroy(rstats_s* s);
+
+/**
+ * @brief Creates a new per worker stats object
+ *
+ * The memory is mapped into shared memory so it can be shared across multiple
+ * processes.
+ *
+ * @return The new stats object
+ */
+rworker_stats_s* rworker_stats_new_mmap();
+
+/**
+ * @brief Destroys the per worker stats object
+ *
+ * @param s The stats object
+ */
+void rworker_stats_destroy(rworker_stats_s* s);
+
+/**
+ * @brief Starts the stats
+ *
+ * @param s The stats object
+ */
+void rstats_start(rstats_s* s);
+
+/**
+ * @brief Finalises the stats
+ *
+ * @param s The stats object
+ */
+void rstats_stop(rstats_s* s);
+
+/**
+ * @brief Prints the stats
+ *
+ * @param s The stats object
+ */