X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=iolog.c;h=cf264916a9ecd83f7b3079bd186573441dba65fb;hp=eb38027e681821866460b08e07f3e17c05212834;hb=a59b12d2a5eb92c1128a5d8ebcd03b1831962ce5;hpb=a6d7953002576946bd7d6703fca698a16ad454a5

diff --git a/iolog.c b/iolog.c
index eb38027e..cf264916 100644
--- a/iolog.c
+++ b/iolog.c
@@ -19,6 +19,7 @@
 #include "smalloc.h"
 #include "blktrace.h"
 #include "pshared.h"
+#include "lib/roundup.h"
 
 #include <netinet/in.h>
 #include <netinet/tcp.h>
@@ -141,6 +142,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;
@@ -148,7 +151,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);
@@ -165,6 +174,7 @@ int read_iolog_get(struct thread_data *td, struct io_u *io_u)
 		io_u->ddir = ipo->ddir;
 		if (ipo->ddir != DDIR_WAIT) {
 			io_u->offset = ipo->offset;
+			io_u->verify_offset = ipo->offset;
 			io_u->buflen = ipo->len;
 			io_u->file = td->files[ipo->fileno];
 			get_file(io_u->file);
@@ -334,6 +344,9 @@ void trim_io_piece(const struct io_u *io_u)
 
 void write_iolog_close(struct thread_data *td)
 {
+	if (!td->iolog_f)
+		return;
+
 	fflush(td->iolog_f);
 	fclose(td->iolog_f);
 	free(td->iolog_buf);
@@ -341,11 +354,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;
@@ -353,8 +394,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
@@ -365,7 +412,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;
 
@@ -405,7 +452,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;
@@ -460,26 +507,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)
@@ -495,7 +571,9 @@ static bool read_iolog2(struct thread_data *td, FILE *f)
 static bool is_socket(const char *path)
 {
 	struct stat buf;
-	int r = stat(path, &buf);
+	int r;
+
+	r = stat(path, &buf);
 	if (r == -1)
 		return false;
 
@@ -504,34 +582,51 @@ static bool is_socket(const char *path)
 
 static int open_socket(const char *path)
 {
-	int fd = socket(AF_UNIX, SOCK_STREAM, 0);
 	struct sockaddr_un addr;
+	int ret, fd;
+
+	fd = socket(AF_UNIX, SOCK_STREAM, 0);
 	if (fd < 0)
 		return fd;
+
 	addr.sun_family = AF_UNIX;
-	strncpy(addr.sun_path, path, sizeof(addr.sun_path));
-	if (connect(fd, (const struct sockaddr *)&addr, strlen(path) + sizeof(addr.sun_family)) == 0)
+	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;
-	else
-		close(fd);
+
+	close(fd);
 	return -1;
 }
 
 /*
  * open iolog, check version, and call appropriate parser
  */
-static bool init_iolog_read(struct thread_data *td)
+static bool init_iolog_read(struct thread_data *td, char *fname)
 {
 	char buffer[256], *p;
 	FILE *f = NULL;
-	bool ret;
-	if (is_socket(td->o.read_iolog_file)) {
-		int fd = open_socket(td->o.read_iolog_file);
-		if (fd >= 0) {
+
+	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 if (!strcmp(fname, "-")) {
+		f = stdin;
 	} else
-		f = fopen(td->o.read_iolog_file, "r");
+		f = fopen(fname, "r");
+
+	free(fname);
+
 	if (!f) {
 		perror("fopen read iolog");
 		return false;
@@ -549,15 +644,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;
 }
 
 /*
@@ -605,15 +700,16 @@ bool init_iolog(struct thread_data *td)
 
 	if (td->o.read_iolog_file) {
 		int need_swap;
+		char * fname = get_name_by_idx(td->o.read_iolog_file, td->subjob_number);
 
 		/*
 		 * Check if it's a blktrace file and load that if possible.
 		 * Otherwise assume it's a normal log file and load that.
 		 */
-		if (is_blktrace(td->o.read_iolog_file, &need_swap))
-			ret = load_blktrace(td, td->o.read_iolog_file, need_swap);
+		if (is_blktrace(fname, &need_swap))
+			ret = load_blktrace(td, fname, need_swap);
 		else
-			ret = init_iolog_read(td);
+			ret = init_iolog_read(td, fname);
 	} else if (td->o.write_iolog_file)
 		ret = init_iolog_write(td);
 	else
@@ -656,10 +752,13 @@ void setup_log(struct io_log **log, struct log_params *p,
 	}
 
 	if (l->td && l->td->o.io_submit_mode != IO_MODE_OFFLOAD) {
+		unsigned int def_samples = DEF_LOG_ENTRIES;
 		struct io_logs *__p;
 
 		__p = calloc(1, sizeof(*l->pending));
-		__p->max_samples = DEF_LOG_ENTRIES;
+		if (l->td->o.iodepth > DEF_LOG_ENTRIES)
+			def_samples = roundup_pow2(l->td->o.iodepth);
+		__p->max_samples = def_samples;
 		__p->log = calloc(__p->max_samples, log_entry_sz(l));
 		l->pending = __p;
 	}
@@ -807,18 +906,18 @@ void flush_samples(FILE *f, void *samples, uint64_t sample_size)
 		s = __get_sample(samples, log_offset, i);
 
 		if (!log_offset) {
-			fprintf(f, "%lu, %" PRId64 ", %u, %llu\n",
+			fprintf(f, "%lu, %" PRId64 ", %u, %llu, %u\n",
 					(unsigned long) s->time,
 					s->data.val,
-					io_sample_ddir(s), (unsigned long long) s->bs);
+					io_sample_ddir(s), (unsigned long long) s->bs, s->priority_bit);
 		} else {
 			struct io_sample_offset *so = (void *) s;
 
-			fprintf(f, "%lu, %" PRId64 ", %u, %llu, %llu\n",
+			fprintf(f, "%lu, %" PRId64 ", %u, %llu, %llu, %u\n",
 					(unsigned long) s->time,
 					s->data.val,
 					io_sample_ddir(s), (unsigned long long) s->bs,
-					(unsigned long long) so->offset);
+					(unsigned long long) so->offset, s->priority_bit);
 		}
 	}
 }