2 * Code related to writing an iolog of what a thread is doing, and to
3 * later read that back and replay
22 #include "lib/roundup.h"
24 #include <netinet/in.h>
25 #include <netinet/tcp.h>
26 #include <arpa/inet.h>
28 #include <sys/socket.h>
31 static int iolog_flush(struct io_log *log);
33 static const char iolog_ver2[] = "fio version 2 iolog";
34 static const char iolog_ver3[] = "fio version 3 iolog";
36 void queue_io_piece(struct thread_data *td, struct io_piece *ipo)
38 flist_add_tail(&ipo->list, &td->io_log_list);
39 td->total_io_size += ipo->len;
42 void log_io_u(const struct thread_data *td, const struct io_u *io_u)
46 if (!td->o.write_iolog_file)
49 fio_gettime(&now, NULL);
50 fprintf(td->iolog_f, "%llu %s %s %llu %llu\n",
51 (unsigned long long) utime_since_now(&td->io_log_start_time),
52 io_u->file->file_name, io_ddir_name(io_u->ddir), io_u->offset,
57 void log_file(struct thread_data *td, struct fio_file *f,
58 enum file_log_act what)
60 const char *act[] = { "add", "open", "close" };
65 if (!td->o.write_iolog_file)
70 * this happens on the pre-open/close done before the job starts
75 fio_gettime(&now, NULL);
76 fprintf(td->iolog_f, "%llu %s %s\n",
77 (unsigned long long) utime_since_now(&td->io_log_start_time),
78 f->file_name, act[what]);
81 static void iolog_delay(struct thread_data *td, unsigned long delay)
83 uint64_t usec = utime_since_now(&td->last_issue);
84 unsigned long orig_delay = delay;
88 if (delay < td->time_offset) {
93 delay -= td->time_offset;
99 fio_gettime(&ts, NULL);
101 while (delay && !td->terminate) {
102 ret = io_u_queued_complete(td, 0);
104 td_verror(td, -ret, "io_u_queued_complete");
105 if (td->flags & TD_F_REGROW_LOGS)
107 if (utime_since_now(&ts) > delay)
111 usec = utime_since_now(&ts);
112 if (usec > orig_delay)
113 td->time_offset = usec - orig_delay;
118 static int ipo_special(struct thread_data *td, struct io_piece *ipo)
126 if (ipo->ddir != DDIR_INVAL)
129 f = td->files[ipo->fileno];
132 iolog_delay(td, ipo->delay);
133 if (fio_fill_issue_time(td))
134 fio_gettime(&td->last_issue, NULL);
135 switch (ipo->file_action) {
136 case FIO_LOG_OPEN_FILE:
137 if (td->o.replay_redirect && fio_file_open(f)) {
138 dprint(FD_FILE, "iolog: ignoring re-open of file %s\n",
142 ret = td_io_open_file(td, f);
144 if (td->o.dp_type != FIO_DP_NONE) {
145 int dp_init_ret = dp_init(td);
147 if (dp_init_ret != 0) {
148 td_verror(td, abs(dp_init_ret), "dp_init");
154 td_verror(td, ret, "iolog open file");
156 case FIO_LOG_CLOSE_FILE:
157 td_io_close_file(td, f);
159 case FIO_LOG_UNLINK_FILE:
160 td_io_unlink_file(td, f);
162 case FIO_LOG_ADD_FILE:
168 log_err("fio: bad file action %d\n", ipo->file_action);
175 static bool read_iolog(struct thread_data *td);
177 unsigned long long delay_since_ttime(const struct thread_data *td,
178 unsigned long long time)
182 const unsigned long long *last_ttime = &td->io_log_last_ttime;
184 if (!*last_ttime || td->o.no_stall || time < *last_ttime)
186 else if (td->o.replay_time_scale == 100)
187 return time - *last_ttime;
190 scale = (double) 100.0 / (double) td->o.replay_time_scale;
191 tmp = time - *last_ttime;
195 int read_iolog_get(struct thread_data *td, struct io_u *io_u)
197 struct io_piece *ipo;
198 unsigned long elapsed;
200 while (!flist_empty(&td->io_log_list)) {
203 if (td->o.read_iolog_chunked) {
204 if (td->io_log_checkmark == td->io_log_current) {
205 if (td->io_log_blktrace) {
206 if (!read_blktrace(td))
213 td->io_log_current--;
215 ipo = flist_first_entry(&td->io_log_list, struct io_piece, list);
216 flist_del(&ipo->list);
217 remove_trim_entry(td, ipo);
219 ret = ipo_special(td, ipo);
223 } else if (ret > 0) {
228 io_u->ddir = ipo->ddir;
229 if (ipo->ddir != DDIR_WAIT) {
230 io_u->offset = ipo->offset;
231 io_u->verify_offset = ipo->offset;
232 io_u->buflen = ipo->len;
233 io_u->file = td->files[ipo->fileno];
234 get_file(io_u->file);
235 dprint(FD_IO, "iolog: get %llu/%llu/%s\n", io_u->offset,
236 io_u->buflen, io_u->file->file_name);
238 iolog_delay(td, ipo->delay);
240 if (td->o.dp_type != FIO_DP_NONE)
241 dp_fill_dspec_data(td, io_u);
243 elapsed = mtime_since_genesis();
244 if (ipo->delay > elapsed)
245 usec_sleep(td, (ipo->delay - elapsed) * 1000);
250 if (io_u->ddir != DDIR_WAIT)
258 void prune_io_piece_log(struct thread_data *td)
260 struct io_piece *ipo;
261 struct fio_rb_node *n;
263 while ((n = rb_first(&td->io_hist_tree)) != NULL) {
264 ipo = rb_entry(n, struct io_piece, rb_node);
265 rb_erase(n, &td->io_hist_tree);
266 remove_trim_entry(td, ipo);
271 while (!flist_empty(&td->io_hist_list)) {
272 ipo = flist_first_entry(&td->io_hist_list, struct io_piece, list);
273 flist_del(&ipo->list);
274 remove_trim_entry(td, ipo);
281 * log a successful write, so we can unwind the log for verify
283 void log_io_piece(struct thread_data *td, struct io_u *io_u)
285 struct fio_rb_node **p, *parent;
286 struct io_piece *ipo, *__ipo;
288 ipo = calloc(1, sizeof(struct io_piece));
290 ipo->file = io_u->file;
291 ipo->offset = io_u->offset;
292 ipo->len = io_u->buflen;
293 ipo->numberio = io_u->numberio;
294 ipo->flags = IP_F_IN_FLIGHT;
298 if (io_u_should_trim(td, io_u)) {
299 flist_add_tail(&ipo->trim_list, &td->trim_list);
304 * Sort writes if we don't have a random map in which case we need to
305 * check for duplicate blocks and drop the old one, which we rely on
306 * the rb insert/lookup for handling. Sort writes if we have offset
307 * modifier which can also create duplicate blocks.
309 if (!fio_offset_overlap_risk(td)) {
310 INIT_FLIST_HEAD(&ipo->list);
311 flist_add_tail(&ipo->list, &td->io_hist_list);
312 ipo->flags |= IP_F_ONLIST;
317 RB_CLEAR_NODE(&ipo->rb_node);
320 * Sort the entry into the verification list
323 p = &td->io_hist_tree.rb_node;
329 __ipo = rb_entry(parent, struct io_piece, rb_node);
330 if (ipo->file < __ipo->file)
332 else if (ipo->file > __ipo->file)
334 else if (ipo->offset < __ipo->offset) {
336 overlap = ipo->offset + ipo->len > __ipo->offset;
338 else if (ipo->offset > __ipo->offset) {
340 overlap = __ipo->offset + __ipo->len > ipo->offset;
346 dprint(FD_IO, "iolog: overlap %llu/%lu, %llu/%lu\n",
347 __ipo->offset, __ipo->len,
348 ipo->offset, ipo->len);
350 rb_erase(parent, &td->io_hist_tree);
351 remove_trim_entry(td, __ipo);
352 if (!(__ipo->flags & IP_F_IN_FLIGHT))
358 rb_link_node(&ipo->rb_node, parent, p);
359 rb_insert_color(&ipo->rb_node, &td->io_hist_tree);
360 ipo->flags |= IP_F_ONRB;
364 void unlog_io_piece(struct thread_data *td, struct io_u *io_u)
366 struct io_piece *ipo = io_u->ipo;
368 if (td->ts.nr_block_infos) {
369 uint32_t *info = io_u_block_info(td, io_u);
370 if (BLOCK_INFO_STATE(*info) < BLOCK_STATE_TRIM_FAILURE) {
371 if (io_u->ddir == DDIR_TRIM)
372 *info = BLOCK_INFO_SET_STATE(*info,
373 BLOCK_STATE_TRIM_FAILURE);
374 else if (io_u->ddir == DDIR_WRITE)
375 *info = BLOCK_INFO_SET_STATE(*info,
376 BLOCK_STATE_WRITE_FAILURE);
383 if (ipo->flags & IP_F_ONRB)
384 rb_erase(&ipo->rb_node, &td->io_hist_tree);
385 else if (ipo->flags & IP_F_ONLIST)
386 flist_del(&ipo->list);
393 void trim_io_piece(const struct io_u *io_u)
395 struct io_piece *ipo = io_u->ipo;
400 ipo->len = io_u->xfer_buflen - io_u->resid;
403 void write_iolog_close(struct thread_data *td)
412 td->iolog_buf = NULL;
415 int64_t iolog_items_to_fetch(struct thread_data *td)
420 int64_t items_to_fetch;
422 if (!td->io_log_highmark)
426 fio_gettime(&now, NULL);
427 elapsed = ntime_since(&td->io_log_highmark_time, &now);
429 for_1s = (td->io_log_highmark - td->io_log_current) * 1000000000 / elapsed;
430 items_to_fetch = for_1s - td->io_log_current;
431 if (items_to_fetch < 0)
436 td->io_log_highmark = td->io_log_current + items_to_fetch;
437 td->io_log_checkmark = (td->io_log_highmark + 1) / 2;
438 fio_gettime(&td->io_log_highmark_time, NULL);
440 return items_to_fetch;
443 #define io_act(_td, _r) (((_td)->io_log_version == 3 && (r) == 5) || \
444 ((_td)->io_log_version == 2 && (r) == 4))
445 #define file_act(_td, _r) (((_td)->io_log_version == 3 && (r) == 3) || \
446 ((_td)->io_log_version == 2 && (r) == 2))
449 * Read version 2 and 3 iolog data. It is enhanced to include per-file logging,
452 static bool read_iolog(struct thread_data *td)
454 unsigned long long offset;
456 unsigned long long delay = 0;
457 int reads, writes, trims, waits, fileno = 0, file_action = 0; /* stupid gcc */
458 char *rfname, *fname, *act;
461 bool realloc = false;
462 int64_t items_to_fetch = 0;
465 if (td->o.read_iolog_chunked) {
466 items_to_fetch = iolog_items_to_fetch(td);
472 * Read in the read iolog and store it, reuse the infrastructure
473 * for doing verifications.
476 rfname = fname = malloc(256+16);
477 act = malloc(256+16);
479 syncs = reads = writes = trims = waits = 0;
480 while ((p = fgets(str, 4096, td->io_log_rfile)) != NULL) {
481 struct io_piece *ipo;
483 unsigned long long ttime;
485 if (td->io_log_version == 3) {
486 r = sscanf(p, "%llu %256s %256s %llu %u", &ttime, rfname, act,
488 delay = delay_since_ttime(td, ttime);
489 td->io_log_last_ttime = ttime;
491 * "wait" is not allowed with version 3
493 if (!strcmp(act, "wait")) {
494 log_err("iolog: ignoring wait command with"
495 " version 3 for file %s\n", fname);
498 } else /* version 2 */
499 r = sscanf(p, "%256s %256s %llu %u", rfname, act, &offset, &bytes);
501 if (td->o.replay_redirect)
502 fname = td->o.replay_redirect;
508 if (!strcmp(act, "wait"))
510 else if (!strcmp(act, "read")) {
511 if (td->o.replay_skip & (1u << DDIR_READ))
514 } else if (!strcmp(act, "write")) {
515 if (td->o.replay_skip & (1u << DDIR_WRITE))
518 } else if (!strcmp(act, "sync")) {
519 if (td->o.replay_skip & (1u << DDIR_SYNC))
522 } else if (!strcmp(act, "datasync"))
524 else if (!strcmp(act, "trim")) {
525 if (td->o.replay_skip & (1u << DDIR_TRIM))
529 log_err("fio: bad iolog file action: %s\n",
533 fileno = get_fileno(td, fname);
534 } else if (file_act(td, r)) {
536 if (!strcmp(act, "add")) {
537 if (td->o.replay_redirect &&
538 get_fileno(td, fname) != -1) {
539 dprint(FD_FILE, "iolog: ignoring"
540 " re-add of file %s\n", fname);
542 fileno = add_file(td, fname, td->subjob_number, 1);
543 file_action = FIO_LOG_ADD_FILE;
545 } else if (!strcmp(act, "open")) {
546 fileno = get_fileno(td, fname);
547 file_action = FIO_LOG_OPEN_FILE;
548 } else if (!strcmp(act, "close")) {
549 fileno = get_fileno(td, fname);
550 file_action = FIO_LOG_CLOSE_FILE;
552 log_err("fio: bad iolog file action: %s\n",
557 log_err("bad iolog%d: %s\n", td->io_log_version, p);
563 else if (rw == DDIR_WRITE) {
565 * Don't add a write for ro mode
570 } else if (rw == DDIR_TRIM) {
572 * Don't add a trim for ro mode
577 } else if (rw == DDIR_WAIT) {
581 } else if (rw == DDIR_INVAL) {
582 } else if (ddir_sync(rw)) {
585 log_err("bad ddir: %d\n", rw);
592 ipo = calloc(1, sizeof(*ipo));
595 if (td->io_log_version == 3)
597 if (rw == DDIR_WAIT) {
600 if (td->o.replay_scale)
601 ipo->offset = offset / td->o.replay_scale;
603 ipo->offset = offset;
604 ipo_bytes_align(td->o.replay_align, ipo);
607 if (rw != DDIR_INVAL && bytes > td->o.max_bs[rw]) {
609 td->o.max_bs[rw] = bytes;
611 ipo->fileno = fileno;
612 ipo->file_action = file_action;
616 queue_io_piece(td, ipo);
618 if (td->o.read_iolog_chunked) {
619 td->io_log_current++;
621 if (items_to_fetch == 0)
630 if (td->o.read_iolog_chunked) {
631 td->io_log_highmark = td->io_log_current;
632 td->io_log_checkmark = (td->io_log_highmark + 1) / 2;
633 fio_gettime(&td->io_log_highmark_time, NULL);
636 if (writes && read_only) {
637 log_err("fio: <%s> skips replay of %d writes due to"
638 " read-only\n", td->o.name, writes);
642 td->flags |= TD_F_SYNCS;
644 if (td->o.read_iolog_chunked) {
645 if (td->io_log_current == 0) {
648 td->o.td_ddir = TD_DDIR_RW;
649 if (realloc && td->orig_buffer)
653 if (init_io_u_buffers(td))
659 if (!reads && !writes && !waits && !trims)
664 td->o.td_ddir |= TD_DDIR_READ;
666 td->o.td_ddir |= TD_DDIR_WRITE;
668 td->o.td_ddir |= TD_DDIR_TRIM;
673 static bool is_socket(const char *path)
678 r = stat(path, &buf);
682 return S_ISSOCK(buf.st_mode);
685 static int open_socket(const char *path)
687 struct sockaddr_un addr;
690 fd = socket(AF_UNIX, SOCK_STREAM, 0);
694 addr.sun_family = AF_UNIX;
695 if (snprintf(addr.sun_path, sizeof(addr.sun_path), "%s", path) >=
696 sizeof(addr.sun_path)) {
697 log_err("%s: path name %s is too long for a Unix socket\n",
701 ret = connect(fd, (const struct sockaddr *)&addr, strlen(path) + sizeof(addr.sun_family));
710 * open iolog, check version, and call appropriate parser
712 static bool init_iolog_read(struct thread_data *td, char *fname)
714 char buffer[256], *p;
717 dprint(FD_IO, "iolog: name=%s\n", fname);
719 if (is_socket(fname)) {
722 fd = open_socket(fname);
725 } else if (!strcmp(fname, "-")) {
728 f = fopen(fname, "r");
731 perror("fopen read iolog");
735 p = fgets(buffer, sizeof(buffer), f);
737 td_verror(td, errno, "iolog read");
738 log_err("fio: unable to read iolog\n");
744 * versions 2 and 3 of the iolog store a specific string as the
745 * first line, check for that
747 if (!strncmp(iolog_ver2, buffer, strlen(iolog_ver2)))
748 td->io_log_version = 2;
749 else if (!strncmp(iolog_ver3, buffer, strlen(iolog_ver3)))
750 td->io_log_version = 3;
752 log_err("fio: iolog version 1 is no longer supported\n");
757 free_release_files(td);
758 td->io_log_rfile = f;
759 return read_iolog(td);
763 * Set up a log for storing io patterns.
765 static bool init_iolog_write(struct thread_data *td)
771 f = fopen(td->o.write_iolog_file, "a");
773 perror("fopen write iolog");
778 * That's it for writing, setup a log buffer and we're done.
781 td->iolog_buf = malloc(8192);
782 setvbuf(f, td->iolog_buf, _IOFBF, 8192);
783 fio_gettime(&td->io_log_start_time, NULL);
786 * write our version line
788 if (fprintf(f, "%s\n", iolog_ver3) < 0) {
789 perror("iolog init\n");
794 * add all known files
796 for_each_file(td, ff, i)
797 log_file(td, ff, FIO_LOG_ADD_FILE);
802 bool init_iolog(struct thread_data *td)
806 if (td->o.read_iolog_file) {
808 char * fname = get_name_by_idx(td->o.read_iolog_file, td->subjob_number);
811 * Check if it's a blktrace file and load that if possible.
812 * Otherwise assume it's a normal log file and load that.
814 if (is_blktrace(fname, &need_swap)) {
815 td->io_log_blktrace = 1;
816 ret = init_blktrace_read(td, fname, need_swap);
818 td->io_log_blktrace = 0;
819 ret = init_iolog_read(td, fname);
822 } else if (td->o.write_iolog_file)
823 ret = init_iolog_write(td);
828 td_verror(td, EINVAL, "failed initializing iolog");
835 void setup_log(struct io_log **log, struct log_params *p,
836 const char *filename)
840 struct io_u_plat_entry *entry;
841 struct flist_head *list;
843 l = scalloc(1, sizeof(*l));
845 INIT_FLIST_HEAD(&l->io_logs);
846 l->log_type = p->log_type;
847 l->log_offset = p->log_offset;
848 l->log_prio = p->log_prio;
849 l->log_issue_time = p->log_issue_time;
850 l->log_gz = p->log_gz;
851 l->log_gz_store = p->log_gz_store;
852 l->avg_msec = p->avg_msec;
853 l->hist_msec = p->hist_msec;
854 l->hist_coarseness = p->hist_coarseness;
855 l->filename = strdup(filename);
858 /* Initialize histogram lists for each r/w direction,
859 * with initial io_u_plat of all zeros:
861 for (i = 0; i < DDIR_RWDIR_CNT; i++) {
862 list = &l->hist_window[i].list;
863 INIT_FLIST_HEAD(list);
864 entry = calloc(1, sizeof(struct io_u_plat_entry));
865 flist_add(&entry->list, list);
868 if (l->td && l->td->o.io_submit_mode != IO_MODE_OFFLOAD) {
869 unsigned int def_samples = DEF_LOG_ENTRIES;
872 __p = calloc(1, sizeof(*l->pending));
873 if (l->td->o.iodepth > DEF_LOG_ENTRIES)
874 def_samples = roundup_pow2(l->td->o.iodepth);
875 __p->max_samples = def_samples;
876 __p->log = calloc(__p->max_samples, log_entry_sz(l));
881 l->log_ddir_mask = LOG_OFFSET_SAMPLE_BIT;
883 l->log_ddir_mask |= LOG_PRIO_SAMPLE_BIT;
885 * The bandwidth-log option generates agg-read_bw.log,
886 * agg-write_bw.log and agg-trim_bw.log for which l->td is NULL.
887 * Check if l->td is valid before dereferencing it.
889 if (l->td && l->td->o.log_max == IO_LOG_SAMPLE_BOTH)
890 l->log_ddir_mask |= LOG_AVG_MAX_SAMPLE_BIT;
892 if (l->log_issue_time)
893 l->log_ddir_mask |= LOG_ISSUE_TIME_SAMPLE_BIT;
895 INIT_FLIST_HEAD(&l->chunk_list);
897 if (l->log_gz && !p->td)
899 else if (l->log_gz || l->log_gz_store) {
900 mutex_init_pshared(&l->chunk_lock);
901 mutex_init_pshared(&l->deferred_free_lock);
902 p->td->flags |= TD_F_COMPRESS_LOG;
908 #ifdef CONFIG_SETVBUF
909 static void *set_file_buffer(FILE *f)
911 size_t size = 1048576;
915 setvbuf(f, buf, _IOFBF, size);
919 static void clear_file_buffer(void *buf)
924 static void *set_file_buffer(FILE *f)
929 static void clear_file_buffer(void *buf)
934 void free_log(struct io_log *log)
936 while (!flist_empty(&log->io_logs)) {
937 struct io_logs *cur_log;
939 cur_log = flist_first_entry(&log->io_logs, struct io_logs, list);
940 flist_del_init(&cur_log->list);
946 free(log->pending->log);
956 uint64_t hist_sum(int j, int stride, uint64_t *io_u_plat,
957 uint64_t *io_u_plat_last)
962 if (io_u_plat_last) {
963 for (k = sum = 0; k < stride; k++)
964 sum += io_u_plat[j + k] - io_u_plat_last[j + k];
966 for (k = sum = 0; k < stride; k++)
967 sum += io_u_plat[j + k];
973 static void flush_hist_samples(FILE *f, int hist_coarseness, void *samples,
974 uint64_t sample_size)
977 bool log_offset, log_issue_time;
978 uint64_t i, j, nr_samples;
979 struct io_u_plat_entry *entry, *entry_before;
981 uint64_t *io_u_plat_before;
983 int stride = 1 << hist_coarseness;
988 s = __get_sample(samples, 0, 0, 0);
989 log_offset = (s->__ddir & LOG_OFFSET_SAMPLE_BIT) != 0;
990 log_issue_time = (s->__ddir & LOG_ISSUE_TIME_SAMPLE_BIT) != 0;
992 nr_samples = sample_size / __log_entry_sz(log_offset, log_issue_time);
994 for (i = 0; i < nr_samples; i++) {
995 s = __get_sample(samples, log_offset, log_issue_time, i);
997 entry = s->data.plat_entry;
998 io_u_plat = entry->io_u_plat;
1000 entry_before = flist_first_entry(&entry->list, struct io_u_plat_entry, list);
1001 io_u_plat_before = entry_before->io_u_plat;
1003 fprintf(f, "%lu, %u, %llu, ", (unsigned long) s->time,
1004 io_sample_ddir(s), (unsigned long long) s->bs);
1005 for (j = 0; j < FIO_IO_U_PLAT_NR - stride; j += stride) {
1006 fprintf(f, "%llu, ", (unsigned long long)
1007 hist_sum(j, stride, io_u_plat, io_u_plat_before));
1009 fprintf(f, "%llu\n", (unsigned long long)
1010 hist_sum(FIO_IO_U_PLAT_NR - stride, stride, io_u_plat,
1013 flist_del(&entry_before->list);
1018 static int print_sample_fields(char **p, size_t *left, const char *fmt, ...) {
1023 ret = vsnprintf(*p, *left, fmt, ap);
1024 if (ret < 0 || ret >= *left) {
1025 log_err("sample file write failed: %d\n", ret);
1038 * flush_samples - Generate output for log samples
1039 * Each sample output is built using a temporary buffer. This buffer size
1041 * - Each sample has less than 10 fields
1042 * - Each sample field fits in 25 characters (20 digits for 64 bit number
1043 * and a few bytes delimiter)
1045 void flush_samples(FILE *f, void *samples, uint64_t sample_size)
1047 struct io_sample *s;
1048 bool log_offset, log_prio, log_avg_max, log_issue_time;
1049 uint64_t i, nr_samples;
1058 s = __get_sample(samples, 0, 0, 0);
1059 log_offset = (s->__ddir & LOG_OFFSET_SAMPLE_BIT) != 0;
1060 log_prio = (s->__ddir & LOG_PRIO_SAMPLE_BIT) != 0;
1061 log_avg_max = (s->__ddir & LOG_AVG_MAX_SAMPLE_BIT) != 0;
1062 log_issue_time = (s->__ddir & LOG_ISSUE_TIME_SAMPLE_BIT) != 0;
1064 nr_samples = sample_size / __log_entry_sz(log_offset, log_issue_time);
1066 for (i = 0; i < nr_samples; i++) {
1067 s = __get_sample(samples, log_offset, log_issue_time, i);
1071 ret = print_sample_fields(&p, &left, "%" PRIu64 ", %" PRId64,
1072 s->time, s->data.val.val0);
1077 ret = print_sample_fields(&p, &left, ", %" PRId64,
1083 ret = print_sample_fields(&p, &left, ", %u, %llu",
1085 (unsigned long long) s->bs);
1090 ret = print_sample_fields(&p, &left, ", %llu",
1091 (unsigned long long) s->aux[IOS_AUX_OFFSET_INDEX]);
1097 ret = print_sample_fields(&p, &left, ", 0x%04x",
1100 ret = print_sample_fields(&p, &left, ", %u",
1101 ioprio_value_is_class_rt(s->priority));
1105 if (log_issue_time) {
1106 ret = print_sample_fields(&p, &left, ", %llu",
1107 (unsigned long long) s->aux[IOS_AUX_ISSUE_TIME_INDEX]);
1112 fprintf(f, "%s\n", buf);
1118 struct iolog_flush_data {
1119 struct workqueue_work work;
1122 uint32_t nr_samples;
1126 #define GZ_CHUNK 131072
1128 static struct iolog_compress *get_new_chunk(unsigned int seq)
1130 struct iolog_compress *c;
1132 c = malloc(sizeof(*c));
1133 INIT_FLIST_HEAD(&c->list);
1134 c->buf = malloc(GZ_CHUNK);
1140 static void free_chunk(struct iolog_compress *ic)
1146 static int z_stream_init(z_stream *stream, int gz_hdr)
1150 memset(stream, 0, sizeof(*stream));
1151 stream->zalloc = Z_NULL;
1152 stream->zfree = Z_NULL;
1153 stream->opaque = Z_NULL;
1154 stream->next_in = Z_NULL;
1157 * zlib magic - add 32 for auto-detection of gz header or not,
1158 * if we decide to store files in a gzip friendly format.
1163 if (inflateInit2(stream, wbits) != Z_OK)
1169 struct inflate_chunk_iter {
1178 static void finish_chunk(z_stream *stream, FILE *f,
1179 struct inflate_chunk_iter *iter)
1183 ret = inflateEnd(stream);
1185 log_err("fio: failed to end log inflation seq %d (%d)\n",
1188 flush_samples(f, iter->buf, iter->buf_used);
1191 iter->buf_size = iter->buf_used = 0;
1195 * Iterative chunk inflation. Handles cases where we cross into a new
1196 * sequence, doing flush finish of previous chunk if needed.
1198 static size_t inflate_chunk(struct iolog_compress *ic, int gz_hdr, FILE *f,
1199 z_stream *stream, struct inflate_chunk_iter *iter)
1203 dprint(FD_COMPRESS, "inflate chunk size=%lu, seq=%u\n",
1204 (unsigned long) ic->len, ic->seq);
1206 if (ic->seq != iter->seq) {
1208 finish_chunk(stream, f, iter);
1210 z_stream_init(stream, gz_hdr);
1211 iter->seq = ic->seq;
1214 stream->avail_in = ic->len;
1215 stream->next_in = ic->buf;
1217 if (!iter->buf_size) {
1218 iter->buf_size = iter->chunk_sz;
1219 iter->buf = malloc(iter->buf_size);
1222 while (stream->avail_in) {
1223 size_t this_out = iter->buf_size - iter->buf_used;
1226 stream->avail_out = this_out;
1227 stream->next_out = iter->buf + iter->buf_used;
1229 err = inflate(stream, Z_NO_FLUSH);
1231 log_err("fio: failed inflating log: %d\n", err);
1236 iter->buf_used += this_out - stream->avail_out;
1238 if (!stream->avail_out) {
1239 iter->buf_size += iter->chunk_sz;
1240 iter->buf = realloc(iter->buf, iter->buf_size);
1244 if (err == Z_STREAM_END)
1248 ret = (void *) stream->next_in - ic->buf;
1250 dprint(FD_COMPRESS, "inflated to size=%lu\n", (unsigned long) iter->buf_size);
1256 * Inflate stored compressed chunks, or write them directly to the log
1257 * file if so instructed.
1259 static int inflate_gz_chunks(struct io_log *log, FILE *f)
1261 struct inflate_chunk_iter iter = { .chunk_sz = log->log_gz, };
1264 while (!flist_empty(&log->chunk_list)) {
1265 struct iolog_compress *ic;
1267 ic = flist_first_entry(&log->chunk_list, struct iolog_compress, list);
1268 flist_del(&ic->list);
1270 if (log->log_gz_store) {
1273 dprint(FD_COMPRESS, "log write chunk size=%lu, "
1274 "seq=%u\n", (unsigned long) ic->len, ic->seq);
1276 ret = fwrite(ic->buf, ic->len, 1, f);
1277 if (ret != 1 || ferror(f)) {
1279 log_err("fio: error writing compressed log\n");
1282 inflate_chunk(ic, log->log_gz_store, f, &stream, &iter);
1288 finish_chunk(&stream, f, &iter);
1296 * Open compressed log file and decompress the stored chunks and
1297 * write them to stdout. The chunks are stored sequentially in the
1298 * file, so we iterate over them and do them one-by-one.
1300 int iolog_file_inflate(const char *file)
1302 struct inflate_chunk_iter iter = { .chunk_sz = 64 * 1024 * 1024, };
1303 struct iolog_compress ic;
1311 f = fopen(file, "rb");
1317 if (stat(file, &sb) < 0) {
1323 ic.buf = buf = malloc(sb.st_size);
1324 ic.len = sb.st_size;
1327 ret = fread(ic.buf, ic.len, 1, f);
1328 if (ret == 0 && ferror(f)) {
1333 } else if (ferror(f) || (!feof(f) && ret != 1)) {
1334 log_err("fio: short read on reading log\n");
1343 * Each chunk will return Z_STREAM_END. We don't know how many
1344 * chunks are in the file, so we just keep looping and incrementing
1345 * the sequence number until we have consumed the whole compressed
1352 iret = inflate_chunk(&ic, 1, stdout, &stream, &iter);
1365 finish_chunk(&stream, stdout, &iter);
1375 static int inflate_gz_chunks(struct io_log *log, FILE *f)
1380 int iolog_file_inflate(const char *file)
1382 log_err("fio: log inflation not possible without zlib\n");
1388 void flush_log(struct io_log *log, bool do_append)
1394 * If log_gz_store is true, we are writing a binary file.
1395 * Set the mode appropriately (on all platforms) to avoid issues
1396 * on windows (line-ending conversions, etc.)
1399 if (log->log_gz_store)
1400 f = fopen(log->filename, "wb");
1402 f = fopen(log->filename, "w");
1404 if (log->log_gz_store)
1405 f = fopen(log->filename, "ab");
1407 f = fopen(log->filename, "a");
1409 perror("fopen log");
1413 buf = set_file_buffer(f);
1415 inflate_gz_chunks(log, f);
1417 while (!flist_empty(&log->io_logs)) {
1418 struct io_logs *cur_log;
1420 cur_log = flist_first_entry(&log->io_logs, struct io_logs, list);
1421 flist_del_init(&cur_log->list);
1423 if (log->td && log == log->td->clat_hist_log)
1424 flush_hist_samples(f, log->hist_coarseness, cur_log->log,
1425 log_sample_sz(log, cur_log));
1427 flush_samples(f, cur_log->log, log_sample_sz(log, cur_log));
1433 clear_file_buffer(buf);
1436 static int finish_log(struct thread_data *td, struct io_log *log, int trylock)
1438 if (td->flags & TD_F_COMPRESS_LOG)
1442 if (fio_trylock_file(log->filename))
1445 fio_lock_file(log->filename);
1447 if (td->client_type == FIO_CLIENT_TYPE_GUI || is_backend)
1448 fio_send_iolog(td, log, log->filename);
1450 flush_log(log, !td->o.per_job_logs);
1452 fio_unlock_file(log->filename);
1457 size_t log_chunk_sizes(struct io_log *log)
1459 struct flist_head *entry;
1462 if (flist_empty(&log->chunk_list))
1466 pthread_mutex_lock(&log->chunk_lock);
1467 flist_for_each(entry, &log->chunk_list) {
1468 struct iolog_compress *c;
1470 c = flist_entry(entry, struct iolog_compress, list);
1473 pthread_mutex_unlock(&log->chunk_lock);
1479 static void iolog_put_deferred(struct io_log *log, void *ptr)
1484 pthread_mutex_lock(&log->deferred_free_lock);
1485 if (log->deferred < IOLOG_MAX_DEFER) {
1486 log->deferred_items[log->deferred] = ptr;
1488 } else if (!fio_did_warn(FIO_WARN_IOLOG_DROP))
1489 log_err("fio: had to drop log entry free\n");
1490 pthread_mutex_unlock(&log->deferred_free_lock);
1493 static void iolog_free_deferred(struct io_log *log)
1500 pthread_mutex_lock(&log->deferred_free_lock);
1502 for (i = 0; i < log->deferred; i++) {
1503 free(log->deferred_items[i]);
1504 log->deferred_items[i] = NULL;
1508 pthread_mutex_unlock(&log->deferred_free_lock);
1511 static int gz_work(struct iolog_flush_data *data)
1513 struct iolog_compress *c = NULL;
1514 struct flist_head list;
1520 INIT_FLIST_HEAD(&list);
1522 memset(&stream, 0, sizeof(stream));
1523 stream.zalloc = Z_NULL;
1524 stream.zfree = Z_NULL;
1525 stream.opaque = Z_NULL;
1527 ret = deflateInit(&stream, Z_DEFAULT_COMPRESSION);
1529 log_err("fio: failed to init gz stream\n");
1533 seq = ++data->log->chunk_seq;
1535 stream.next_in = (void *) data->samples;
1536 stream.avail_in = data->nr_samples * log_entry_sz(data->log);
1538 dprint(FD_COMPRESS, "deflate input size=%lu, seq=%u, log=%s\n",
1539 (unsigned long) stream.avail_in, seq,
1540 data->log->filename);
1543 dprint(FD_COMPRESS, "seq=%d, chunk=%lu\n", seq,
1544 (unsigned long) c->len);
1545 c = get_new_chunk(seq);
1546 stream.avail_out = GZ_CHUNK;
1547 stream.next_out = c->buf;
1548 ret = deflate(&stream, Z_NO_FLUSH);
1550 log_err("fio: deflate log (%d)\n", ret);
1555 c->len = GZ_CHUNK - stream.avail_out;
1556 flist_add_tail(&c->list, &list);
1558 } while (stream.avail_in);
1560 stream.next_out = c->buf + c->len;
1561 stream.avail_out = GZ_CHUNK - c->len;
1563 ret = deflate(&stream, Z_FINISH);
1566 * Z_BUF_ERROR is special, it just means we need more
1567 * output space. We'll handle that below. Treat any other
1570 if (ret != Z_BUF_ERROR) {
1571 log_err("fio: deflate log (%d)\n", ret);
1572 flist_del(&c->list);
1579 c->len = GZ_CHUNK - stream.avail_out;
1581 dprint(FD_COMPRESS, "seq=%d, chunk=%lu\n", seq, (unsigned long) c->len);
1583 if (ret != Z_STREAM_END) {
1585 c = get_new_chunk(seq);
1586 stream.avail_out = GZ_CHUNK;
1587 stream.next_out = c->buf;
1588 ret = deflate(&stream, Z_FINISH);
1589 c->len = GZ_CHUNK - stream.avail_out;
1591 flist_add_tail(&c->list, &list);
1592 dprint(FD_COMPRESS, "seq=%d, chunk=%lu\n", seq,
1593 (unsigned long) c->len);
1594 } while (ret != Z_STREAM_END);
1597 dprint(FD_COMPRESS, "deflated to size=%lu\n", (unsigned long) total);
1599 ret = deflateEnd(&stream);
1601 log_err("fio: deflateEnd %d\n", ret);
1603 iolog_put_deferred(data->log, data->samples);
1605 if (!flist_empty(&list)) {
1606 pthread_mutex_lock(&data->log->chunk_lock);
1607 flist_splice_tail(&list, &data->log->chunk_list);
1608 pthread_mutex_unlock(&data->log->chunk_lock);
1617 while (!flist_empty(&list)) {
1618 c = flist_first_entry(list.next, struct iolog_compress, list);
1619 flist_del(&c->list);
1627 * Invoked from our compress helper thread, when logging would have exceeded
1628 * the specified memory limitation. Compresses the previously stored
1631 static int gz_work_async(struct submit_worker *sw, struct workqueue_work *work)
1633 return gz_work(container_of(work, struct iolog_flush_data, work));
1636 static int gz_init_worker(struct submit_worker *sw)
1638 struct thread_data *td = sw->wq->td;
1640 if (!fio_option_is_set(&td->o, log_gz_cpumask))
1643 if (fio_setaffinity(gettid(), td->o.log_gz_cpumask) == -1) {
1644 log_err("gz: failed to set CPU affinity\n");
1651 static struct workqueue_ops log_compress_wq_ops = {
1652 .fn = gz_work_async,
1653 .init_worker_fn = gz_init_worker,
1657 int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out)
1659 if (!(td->flags & TD_F_COMPRESS_LOG))
1662 workqueue_init(td, &td->log_compress_wq, &log_compress_wq_ops, 1, sk_out);
1666 void iolog_compress_exit(struct thread_data *td)
1668 if (!(td->flags & TD_F_COMPRESS_LOG))
1671 workqueue_exit(&td->log_compress_wq);
1675 * Queue work item to compress the existing log entries. We reset the
1676 * current log to a small size, and reference the existing log in the
1677 * data that we queue for compression. Once compression has been done,
1678 * this old log is freed. Will not return until the log compression
1679 * has completed, and will flush all previous logs too
1681 static int iolog_flush(struct io_log *log)
1683 struct iolog_flush_data *data;
1685 workqueue_flush(&log->td->log_compress_wq);
1686 data = malloc(sizeof(*data));
1693 while (!flist_empty(&log->io_logs)) {
1694 struct io_logs *cur_log;
1696 cur_log = flist_first_entry(&log->io_logs, struct io_logs, list);
1697 flist_del_init(&cur_log->list);
1699 data->samples = cur_log->log;
1700 data->nr_samples = cur_log->nr_samples;
1711 int iolog_cur_flush(struct io_log *log, struct io_logs *cur_log)
1713 struct iolog_flush_data *data;
1715 data = smalloc(sizeof(*data));
1721 data->samples = cur_log->log;
1722 data->nr_samples = cur_log->nr_samples;
1725 cur_log->nr_samples = cur_log->max_samples = 0;
1726 cur_log->log = NULL;
1728 workqueue_enqueue(&log->td->log_compress_wq, &data->work);
1730 iolog_free_deferred(log);
1736 static int iolog_flush(struct io_log *log)
1741 int iolog_cur_flush(struct io_log *log, struct io_logs *cur_log)
1746 int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out)
1751 void iolog_compress_exit(struct thread_data *td)
1757 struct io_logs *iolog_cur_log(struct io_log *log)
1759 if (flist_empty(&log->io_logs))
1762 return flist_last_entry(&log->io_logs, struct io_logs, list);
1765 uint64_t iolog_nr_samples(struct io_log *iolog)
1767 struct flist_head *entry;
1770 flist_for_each(entry, &iolog->io_logs) {
1771 struct io_logs *cur_log;
1773 cur_log = flist_entry(entry, struct io_logs, list);
1774 ret += cur_log->nr_samples;
1780 static int __write_log(struct thread_data *td, struct io_log *log, int try)
1783 return finish_log(td, log, try);
1788 static int write_iops_log(struct thread_data *td, int try, bool unit_log)
1792 if (per_unit_log(td->iops_log) != unit_log)
1795 ret = __write_log(td, td->iops_log, try);
1797 td->iops_log = NULL;
1802 static int write_slat_log(struct thread_data *td, int try, bool unit_log)
1809 ret = __write_log(td, td->slat_log, try);
1811 td->slat_log = NULL;
1816 static int write_clat_log(struct thread_data *td, int try, bool unit_log)
1823 ret = __write_log(td, td->clat_log, try);
1825 td->clat_log = NULL;
1830 static int write_clat_hist_log(struct thread_data *td, int try, bool unit_log)
1837 ret = __write_log(td, td->clat_hist_log, try);
1839 td->clat_hist_log = NULL;
1844 static int write_lat_log(struct thread_data *td, int try, bool unit_log)
1851 ret = __write_log(td, td->lat_log, try);
1858 static int write_bandw_log(struct thread_data *td, int try, bool unit_log)
1862 if (per_unit_log(td->bw_log) != unit_log)
1865 ret = __write_log(td, td->bw_log, try);
1878 CLAT_HIST_LOG_MASK = 32,
1885 int (*fn)(struct thread_data *, int, bool);
1888 static struct log_type log_types[] = {
1890 .mask = BW_LOG_MASK,
1891 .fn = write_bandw_log,
1894 .mask = LAT_LOG_MASK,
1895 .fn = write_lat_log,
1898 .mask = SLAT_LOG_MASK,
1899 .fn = write_slat_log,
1902 .mask = CLAT_LOG_MASK,
1903 .fn = write_clat_log,
1906 .mask = IOPS_LOG_MASK,
1907 .fn = write_iops_log,
1910 .mask = CLAT_HIST_LOG_MASK,
1911 .fn = write_clat_hist_log,
1915 void td_writeout_logs(struct thread_data *td, bool unit_logs)
1917 unsigned int log_mask = 0;
1918 unsigned int log_left = ALL_LOG_NR;
1921 old_state = td_bump_runstate(td, TD_FINISHING);
1923 finalize_logs(td, unit_logs);
1926 int prev_log_left = log_left;
1928 for (i = 0; i < ALL_LOG_NR && log_left; i++) {
1929 struct log_type *lt = &log_types[i];
1932 if (!(log_mask & lt->mask)) {
1933 ret = lt->fn(td, log_left != 1, unit_logs);
1936 log_mask |= lt->mask;
1941 if (prev_log_left == log_left)
1945 td_restore_runstate(td, old_state);
1948 void fio_writeout_logs(bool unit_logs)
1951 td_writeout_logs(td, unit_logs);