Merge branch 'dev' of https://github.com/smartxworks/fio
[fio.git] / iolog.c
diff --git a/iolog.c b/iolog.c
index d51e49c..b72dcf9 100644 (file)
--- a/iolog.c
+++ b/iolog.c
 #include "blktrace.h"
 #include "pshared.h"
 
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <arpa/inet.h>
+#include <sys/stat.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+
 static int iolog_flush(struct io_log *log);
 
 static const char iolog_ver2[] = "fio version 2 iolog";
@@ -134,6 +141,8 @@ static int ipo_special(struct thread_data *td, struct io_piece *ipo)
        return 1;
 }
 
+static bool read_iolog2(struct thread_data *td);
+
 int read_iolog_get(struct thread_data *td, struct io_u *io_u)
 {
        struct io_piece *ipo;
@@ -141,7 +150,13 @@ int read_iolog_get(struct thread_data *td, struct io_u *io_u)
 
        while (!flist_empty(&td->io_log_list)) {
                int ret;
-
+               if (td->o.read_iolog_chunked) {
+                       if (td->io_log_checkmark == td->io_log_current) {
+                               if (!read_iolog2(td))
+                                       return 1;
+                       }
+                       td->io_log_current--;
+               }
                ipo = flist_first_entry(&td->io_log_list, struct io_piece, list);
                flist_del(&ipo->list);
                remove_trim_entry(td, ipo);
@@ -334,11 +349,39 @@ void write_iolog_close(struct thread_data *td)
        td->iolog_buf = NULL;
 }
 
+static int64_t iolog_items_to_fetch(struct thread_data *td)
+{
+       struct timespec now;
+       uint64_t elapsed;
+       uint64_t for_1s;
+       int64_t items_to_fetch;
+
+       if (!td->io_log_highmark)
+               return 10;
+
+
+       fio_gettime(&now, NULL);
+       elapsed = ntime_since(&td->io_log_highmark_time, &now);
+       if (elapsed) {
+               for_1s = (td->io_log_highmark - td->io_log_current) * 1000000000 / elapsed;
+               items_to_fetch = for_1s - td->io_log_current;
+               if (items_to_fetch < 0)
+                       items_to_fetch = 0;
+       } else
+               items_to_fetch = 0;
+
+       td->io_log_highmark = td->io_log_current + items_to_fetch;
+       td->io_log_checkmark = (td->io_log_highmark + 1) / 2;
+       fio_gettime(&td->io_log_highmark_time, NULL);
+
+       return items_to_fetch;
+}
+
 /*
  * Read version 2 iolog data. It is enhanced to include per-file logging,
  * syncs, etc.
  */
-static bool read_iolog2(struct thread_data *td, FILE *f)
+static bool read_iolog2(struct thread_data *td)
 {
        unsigned long long offset;
        unsigned int bytes;
@@ -346,8 +389,14 @@ static bool read_iolog2(struct thread_data *td, FILE *f)
        char *rfname, *fname, *act;
        char *str, *p;
        enum fio_ddir rw;
+       bool realloc = false;
+       int64_t items_to_fetch = 0;
 
-       free_release_files(td);
+       if (td->o.read_iolog_chunked) {
+               items_to_fetch = iolog_items_to_fetch(td);
+               if (!items_to_fetch)
+                       return true;
+       }
 
        /*
         * Read in the read iolog and store it, reuse the infrastructure
@@ -358,7 +407,7 @@ static bool read_iolog2(struct thread_data *td, FILE *f)
        act = malloc(256+16);
 
        reads = writes = waits = 0;
-       while ((p = fgets(str, 4096, f)) != NULL) {
+       while ((p = fgets(str, 4096, td->io_log_rfile)) != NULL) {
                struct io_piece *ipo;
                int r;
 
@@ -398,7 +447,7 @@ static bool read_iolog2(struct thread_data *td, FILE *f)
                                        dprint(FD_FILE, "iolog: ignoring"
                                                " re-add of file %s\n", fname);
                                } else {
-                                       fileno = add_file(td, fname, 0, 1);
+                                       fileno = add_file(td, fname, td->subjob_number, 1);
                                        file_action = FIO_LOG_ADD_FILE;
                                }
                                continue;
@@ -453,26 +502,55 @@ static bool read_iolog2(struct thread_data *td, FILE *f)
                        ipo_bytes_align(td->o.replay_align, ipo);
 
                        ipo->len = bytes;
-                       if (rw != DDIR_INVAL && bytes > td->o.max_bs[rw])
+                       if (rw != DDIR_INVAL && bytes > td->o.max_bs[rw]) {
+                               realloc = true;
                                td->o.max_bs[rw] = bytes;
+                       }
                        ipo->fileno = fileno;
                        ipo->file_action = file_action;
                        td->o.size += bytes;
                }
 
                queue_io_piece(td, ipo);
+
+               if (td->o.read_iolog_chunked) {
+                       td->io_log_current++;
+                       items_to_fetch--;
+                       if (items_to_fetch == 0)
+                               break;
+               }
        }
 
        free(str);
        free(act);
        free(rfname);
 
+       if (td->o.read_iolog_chunked) {
+               td->io_log_highmark = td->io_log_current;
+               td->io_log_checkmark = (td->io_log_highmark + 1) / 2;
+               fio_gettime(&td->io_log_highmark_time, NULL);
+       }
+
        if (writes && read_only) {
                log_err("fio: <%s> skips replay of %d writes due to"
                        " read-only\n", td->o.name, writes);
                writes = 0;
        }
 
+       if (td->o.read_iolog_chunked) {
+               if (td->io_log_current == 0) {
+                       return false;
+               }
+               td->o.td_ddir = TD_DDIR_RW;
+               if (realloc && td->orig_buffer)
+               {
+                       io_u_quiesce(td);
+                       free_io_mem(td);
+                       init_io_u_buffers(td);
+               }
+               return true;
+       }
+
        if (!reads && !writes && !waits)
                return false;
        else if (reads && !writes)
@@ -485,16 +563,64 @@ static bool read_iolog2(struct thread_data *td, FILE *f)
        return true;
 }
 
+static bool is_socket(const char *path)
+{
+       struct stat buf;
+       int r;
+
+       r = stat(path, &buf);
+       if (r == -1)
+               return false;
+
+       return S_ISSOCK(buf.st_mode);
+}
+
+static int open_socket(const char *path)
+{
+       struct sockaddr_un addr;
+       int ret, fd;
+
+       fd = socket(AF_UNIX, SOCK_STREAM, 0);
+       if (fd < 0)
+               return fd;
+
+       addr.sun_family = AF_UNIX;
+       if (snprintf(addr.sun_path, sizeof(addr.sun_path), "%s", path) >=
+           sizeof(addr.sun_path)) {
+               log_err("%s: path name %s is too long for a Unix socket\n",
+                       __func__, path);
+       }
+
+       ret = connect(fd, (const struct sockaddr *)&addr, strlen(path) + sizeof(addr.sun_family));
+       if (!ret)
+               return fd;
+
+       close(fd);
+       return -1;
+}
+
 /*
  * open iolog, check version, and call appropriate parser
  */
 static bool init_iolog_read(struct thread_data *td)
 {
-       char buffer[256], *p;
-       FILE *f;
-       bool ret;
+       char buffer[256], *p, *fname;
+       FILE *f = NULL;
+
+       fname = get_name_by_idx(td->o.read_iolog_file, td->subjob_number);
+       dprint(FD_IO, "iolog: name=%s\n", fname);
+
+       if (is_socket(fname)) {
+               int fd;
+
+               fd = open_socket(fname);
+               if (fd >= 0)
+                       f = fdopen(fd, "r");
+       } else
+               f = fopen(fname, "r");
+
+       free(fname);
 
-       f = fopen(td->o.read_iolog_file, "r");
        if (!f) {
                perror("fopen read iolog");
                return false;
@@ -512,15 +638,15 @@ static bool init_iolog_read(struct thread_data *td)
         * version 2 of the iolog stores a specific string as the
         * first line, check for that
         */
-       if (!strncmp(iolog_ver2, buffer, strlen(iolog_ver2)))
-               ret = read_iolog2(td, f);
-       else {
-               log_err("fio: iolog version 1 is no longer supported\n");
-               ret = false;
+       if (!strncmp(iolog_ver2, buffer, strlen(iolog_ver2))) {
+               free_release_files(td);
+               td->io_log_rfile = f;
+               return read_iolog2(td);
        }
 
+       log_err("fio: iolog version 1 is no longer supported\n");
        fclose(f);
-       return ret;
+       return false;
 }
 
 /*