projects / fio.git / blame_incremental
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (non-incremental) | history | HEAD
Merge branch 'enable-dataplacement-while-replaying-io' of https://github.com/parkvibe...
[fio.git] / iolog.c
... / ...
CommitLineData
1/*
2 * Code related to writing an iolog of what a thread is doing, and to
3 * later read that back and replay
4 */
5#include <stdio.h>
6#include <stdlib.h>
7#include <assert.h>
8#include <sys/types.h>
9#include <sys/stat.h>
10#include <unistd.h>
11#ifdef CONFIG_ZLIB
12#include <zlib.h>
13#endif
14
15#include "flist.h"
16#include "fio.h"
17#include "trim.h"
18#include "filelock.h"
19#include "smalloc.h"
20#include "blktrace.h"
21#include "pshared.h"
22#include "lib/roundup.h"
23
24#include <netinet/in.h>
25#include <netinet/tcp.h>
26#include <arpa/inet.h>
27#include <sys/stat.h>
28#include <sys/socket.h>
29#include <sys/un.h>
30
31static int iolog_flush(struct io_log *log);
32
33static const char iolog_ver2[] = "fio version 2 iolog";
34static const char iolog_ver3[] = "fio version 3 iolog";
35
36void queue_io_piece(struct thread_data *td, struct io_piece *ipo)
37{
38 flist_add_tail(&ipo->list, &td->io_log_list);
39 td->total_io_size += ipo->len;
40}
41
42void log_io_u(const struct thread_data *td, const struct io_u *io_u)
43{
44 struct timespec now;
45
46 if (!td->o.write_iolog_file)
47 return;
48
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,
53 io_u->buflen);
54
55}
56
57void log_file(struct thread_data *td, struct fio_file *f,
58 enum file_log_act what)
59{
60 const char *act[] = { "add", "open", "close" };
61 struct timespec now;
62
63 assert(what < 3);
64
65 if (!td->o.write_iolog_file)
66 return;
67
68
69 /*
70 * this happens on the pre-open/close done before the job starts
71 */
72 if (!td->iolog_f)
73 return;
74
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]);
79}
80
81static void iolog_delay(struct thread_data *td, unsigned long delay)
82{
83 uint64_t usec = utime_since_now(&td->last_issue);
84 unsigned long orig_delay = delay;
85 struct timespec ts;
86 int ret = 0;
87
88 if (delay < td->time_offset) {
89 td->time_offset = 0;
90 return;
91 }
92
93 delay -= td->time_offset;
94 if (delay < usec)
95 return;
96
97 delay -= usec;
98
99 fio_gettime(&ts, NULL);
100
101 while (delay && !td->terminate) {
102 ret = io_u_queued_complete(td, 0);
103 if (ret < 0)
104 td_verror(td, -ret, "io_u_queued_complete");
105 if (td->flags & TD_F_REGROW_LOGS)
106 regrow_logs(td);
107 if (utime_since_now(&ts) > delay)
108 break;
109 }
110
111 usec = utime_since_now(&ts);
112 if (usec > orig_delay)
113 td->time_offset = usec - orig_delay;
114 else
115 td->time_offset = 0;
116}
117
118static int ipo_special(struct thread_data *td, struct io_piece *ipo)
119{
120 struct fio_file *f;
121 int ret;
122
123 /*
124 * Not a special ipo
125 */
126 if (ipo->ddir != DDIR_INVAL)
127 return 0;
128
129 f = td->files[ipo->fileno];
130
131 if (ipo->delay)
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",
139 f->file_name);
140 break;
141 }
142 ret = td_io_open_file(td, f);
143 if (!ret) {
144 if (td->o.dp_type != FIO_DP_NONE) {
145 int dp_init_ret = dp_init(td);
146
147 if (dp_init_ret != 0) {
148 td_verror(td, dp_init_ret, "dp_init");
149 return -1;
150 }
151 }
152 break;
153 }
154 td_verror(td, ret, "iolog open file");
155 return -1;
156 case FIO_LOG_CLOSE_FILE:
157 td_io_close_file(td, f);
158 break;
159 case FIO_LOG_UNLINK_FILE:
160 td_io_unlink_file(td, f);
161 break;
162 case FIO_LOG_ADD_FILE:
163 /*
164 * Nothing to do
165 */
166 break;
167 default:
168 log_err("fio: bad file action %d\n", ipo->file_action);
169 break;
170 }
171
172 return 1;
173}
174
175static bool read_iolog(struct thread_data *td);
176
177unsigned long long delay_since_ttime(const struct thread_data *td,
178 unsigned long long time)
179{
180 double tmp;
181 double scale;
182 const unsigned long long *last_ttime = &td->io_log_last_ttime;
183
184 if (!*last_ttime || td->o.no_stall || time < *last_ttime)
185 return 0;
186 else if (td->o.replay_time_scale == 100)
187 return time - *last_ttime;
188
189
190 scale = (double) 100.0 / (double) td->o.replay_time_scale;
191 tmp = time - *last_ttime;
192 return tmp * scale;
193}
194
195int read_iolog_get(struct thread_data *td, struct io_u *io_u)
196{
197 struct io_piece *ipo;
198 unsigned long elapsed;
199
200 while (!flist_empty(&td->io_log_list)) {
201 int ret;
202
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))
207 return 1;
208 } else {
209 if (!read_iolog(td))
210 return 1;
211 }
212 }
213 td->io_log_current--;
214 }
215 ipo = flist_first_entry(&td->io_log_list, struct io_piece, list);
216 flist_del(&ipo->list);
217 remove_trim_entry(td, ipo);
218
219 ret = ipo_special(td, ipo);
220 if (ret < 0) {
221 free(ipo);
222 break;
223 } else if (ret > 0) {
224 free(ipo);
225 continue;
226 }
227
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);
237 if (ipo->delay)
238 iolog_delay(td, ipo->delay);
239 } else {
240 elapsed = mtime_since_genesis();
241 if (ipo->delay > elapsed)
242 usec_sleep(td, (ipo->delay - elapsed) * 1000);
243 }
244
245 if (td->o.dp_type != FIO_DP_NONE)
246 dp_fill_dspec_data(td, io_u);
247
248 free(ipo);
249
250 if (io_u->ddir != DDIR_WAIT)
251 return 0;
252 }
253
254 td->done = 1;
255 return 1;
256}
257
258void prune_io_piece_log(struct thread_data *td)
259{
260 struct io_piece *ipo;
261 struct fio_rb_node *n;
262
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);
267 td->io_hist_len--;
268 free(ipo);
269 }
270
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);
275 td->io_hist_len--;
276 free(ipo);
277 }
278}
279
280/*
281 * log a successful write, so we can unwind the log for verify
282 */
283void log_io_piece(struct thread_data *td, struct io_u *io_u)
284{
285 struct fio_rb_node **p, *parent;
286 struct io_piece *ipo, *__ipo;
287
288 ipo = calloc(1, sizeof(struct io_piece));
289 init_ipo(ipo);
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;
295
296 io_u->ipo = ipo;
297
298 if (io_u_should_trim(td, io_u)) {
299 flist_add_tail(&ipo->trim_list, &td->trim_list);
300 td->trim_entries++;
301 }
302
303 /*
304 * Only sort writes if we don't have a random map in which case we need
305 * to check for duplicate blocks and drop the old one, which we rely on
306 * the rb insert/lookup for handling.
307 */
308 if (file_randommap(td, ipo->file)) {
309 INIT_FLIST_HEAD(&ipo->list);
310 flist_add_tail(&ipo->list, &td->io_hist_list);
311 ipo->flags |= IP_F_ONLIST;
312 td->io_hist_len++;
313 return;
314 }
315
316 RB_CLEAR_NODE(&ipo->rb_node);
317
318 /*
319 * Sort the entry into the verification list
320 */
321restart:
322 p = &td->io_hist_tree.rb_node;
323 parent = NULL;
324 while (*p) {
325 int overlap = 0;
326 parent = *p;
327
328 __ipo = rb_entry(parent, struct io_piece, rb_node);
329 if (ipo->file < __ipo->file)
330 p = &(*p)->rb_left;
331 else if (ipo->file > __ipo->file)
332 p = &(*p)->rb_right;
333 else if (ipo->offset < __ipo->offset) {
334 p = &(*p)->rb_left;
335 overlap = ipo->offset + ipo->len > __ipo->offset;
336 }
337 else if (ipo->offset > __ipo->offset) {
338 p = &(*p)->rb_right;
339 overlap = __ipo->offset + __ipo->len > ipo->offset;
340 }
341 else
342 overlap = 1;
343
344 if (overlap) {
345 dprint(FD_IO, "iolog: overlap %llu/%lu, %llu/%lu\n",
346 __ipo->offset, __ipo->len,
347 ipo->offset, ipo->len);
348 td->io_hist_len--;
349 rb_erase(parent, &td->io_hist_tree);
350 remove_trim_entry(td, __ipo);
351 if (!(__ipo->flags & IP_F_IN_FLIGHT))
352 free(__ipo);
353 goto restart;
354 }
355 }
356
357 rb_link_node(&ipo->rb_node, parent, p);
358 rb_insert_color(&ipo->rb_node, &td->io_hist_tree);
359 ipo->flags |= IP_F_ONRB;
360 td->io_hist_len++;
361}
362
363void unlog_io_piece(struct thread_data *td, struct io_u *io_u)
364{
365 struct io_piece *ipo = io_u->ipo;
366
367 if (td->ts.nr_block_infos) {
368 uint32_t *info = io_u_block_info(td, io_u);
369 if (BLOCK_INFO_STATE(*info) < BLOCK_STATE_TRIM_FAILURE) {
370 if (io_u->ddir == DDIR_TRIM)
371 *info = BLOCK_INFO_SET_STATE(*info,
372 BLOCK_STATE_TRIM_FAILURE);
373 else if (io_u->ddir == DDIR_WRITE)
374 *info = BLOCK_INFO_SET_STATE(*info,
375 BLOCK_STATE_WRITE_FAILURE);
376 }
377 }
378
379 if (!ipo)
380 return;
381
382 if (ipo->flags & IP_F_ONRB)
383 rb_erase(&ipo->rb_node, &td->io_hist_tree);
384 else if (ipo->flags & IP_F_ONLIST)
385 flist_del(&ipo->list);
386
387 free(ipo);
388 io_u->ipo = NULL;
389 td->io_hist_len--;
390}
391
392void trim_io_piece(const struct io_u *io_u)
393{
394 struct io_piece *ipo = io_u->ipo;
395
396 if (!ipo)
397 return;
398
399 ipo->len = io_u->xfer_buflen - io_u->resid;
400}
401
402void write_iolog_close(struct thread_data *td)
403{
404 if (!td->iolog_f)
405 return;
406
407 fflush(td->iolog_f);
408 fclose(td->iolog_f);
409 free(td->iolog_buf);
410 td->iolog_f = NULL;
411 td->iolog_buf = NULL;
412}
413
414int64_t iolog_items_to_fetch(struct thread_data *td)
415{
416 struct timespec now;
417 uint64_t elapsed;
418 uint64_t for_1s;
419 int64_t items_to_fetch;
420
421 if (!td->io_log_highmark)
422 return 10;
423
424
425 fio_gettime(&now, NULL);
426 elapsed = ntime_since(&td->io_log_highmark_time, &now);
427 if (elapsed) {
428 for_1s = (td->io_log_highmark - td->io_log_current) * 1000000000 / elapsed;
429 items_to_fetch = for_1s - td->io_log_current;
430 if (items_to_fetch < 0)
431 items_to_fetch = 0;
432 } else
433 items_to_fetch = 0;
434
435 td->io_log_highmark = td->io_log_current + items_to_fetch;
436 td->io_log_checkmark = (td->io_log_highmark + 1) / 2;
437 fio_gettime(&td->io_log_highmark_time, NULL);
438
439 return items_to_fetch;
440}
441
442#define io_act(_td, _r) (((_td)->io_log_version == 3 && (r) == 5) || \
443 ((_td)->io_log_version == 2 && (r) == 4))
444#define file_act(_td, _r) (((_td)->io_log_version == 3 && (r) == 3) || \
445 ((_td)->io_log_version == 2 && (r) == 2))
446
447/*
448 * Read version 2 and 3 iolog data. It is enhanced to include per-file logging,
449 * syncs, etc.
450 */
451static bool read_iolog(struct thread_data *td)
452{
453 unsigned long long offset;
454 unsigned int bytes;
455 unsigned long long delay = 0;
456 int reads, writes, trims, waits, fileno = 0, file_action = 0; /* stupid gcc */
457 char *rfname, *fname, *act;
458 char *str, *p;
459 enum fio_ddir rw;
460 bool realloc = false;
461 int64_t items_to_fetch = 0;
462 int syncs;
463
464 if (td->o.read_iolog_chunked) {
465 items_to_fetch = iolog_items_to_fetch(td);
466 if (!items_to_fetch)
467 return true;
468 }
469
470 /*
471 * Read in the read iolog and store it, reuse the infrastructure
472 * for doing verifications.
473 */
474 str = malloc(4096);
475 rfname = fname = malloc(256+16);
476 act = malloc(256+16);
477
478 syncs = reads = writes = trims = waits = 0;
479 while ((p = fgets(str, 4096, td->io_log_rfile)) != NULL) {
480 struct io_piece *ipo;
481 int r;
482 unsigned long long ttime;
483
484 if (td->io_log_version == 3) {
485 r = sscanf(p, "%llu %256s %256s %llu %u", &ttime, rfname, act,
486 &offset, &bytes);
487 delay = delay_since_ttime(td, ttime);
488 td->io_log_last_ttime = ttime;
489 /*
490 * "wait" is not allowed with version 3
491 */
492 if (!strcmp(act, "wait")) {
493 log_err("iolog: ignoring wait command with"
494 " version 3 for file %s\n", fname);
495 continue;
496 }
497 } else /* version 2 */
498 r = sscanf(p, "%256s %256s %llu %u", rfname, act, &offset, &bytes);
499
500 if (td->o.replay_redirect)
501 fname = td->o.replay_redirect;
502
503 if (io_act(td, r)) {
504 /*
505 * Check action first
506 */
507 if (!strcmp(act, "wait"))
508 rw = DDIR_WAIT;
509 else if (!strcmp(act, "read")) {
510 if (td->o.replay_skip & (1u << DDIR_READ))
511 continue;
512 rw = DDIR_READ;
513 } else if (!strcmp(act, "write")) {
514 if (td->o.replay_skip & (1u << DDIR_WRITE))
515 continue;
516 rw = DDIR_WRITE;
517 } else if (!strcmp(act, "sync")) {
518 if (td->o.replay_skip & (1u << DDIR_SYNC))
519 continue;
520 rw = DDIR_SYNC;
521 } else if (!strcmp(act, "datasync"))
522 rw = DDIR_DATASYNC;
523 else if (!strcmp(act, "trim")) {
524 if (td->o.replay_skip & (1u << DDIR_TRIM))
525 continue;
526 rw = DDIR_TRIM;
527 } else {
528 log_err("fio: bad iolog file action: %s\n",
529 act);
530 continue;
531 }
532 fileno = get_fileno(td, fname);
533 } else if (file_act(td, r)) {
534 rw = DDIR_INVAL;
535 if (!strcmp(act, "add")) {
536 if (td->o.replay_redirect &&
537 get_fileno(td, fname) != -1) {
538 dprint(FD_FILE, "iolog: ignoring"
539 " re-add of file %s\n", fname);
540 } else {
541 fileno = add_file(td, fname, td->subjob_number, 1);
542 file_action = FIO_LOG_ADD_FILE;
543 }
544 } else if (!strcmp(act, "open")) {
545 fileno = get_fileno(td, fname);
546 file_action = FIO_LOG_OPEN_FILE;
547 } else if (!strcmp(act, "close")) {
548 fileno = get_fileno(td, fname);
549 file_action = FIO_LOG_CLOSE_FILE;
550 } else {
551 log_err("fio: bad iolog file action: %s\n",
552 act);
553 continue;
554 }
555 } else {
556 log_err("bad iolog%d: %s\n", td->io_log_version, p);
557 continue;
558 }
559
560 if (rw == DDIR_READ)
561 reads++;
562 else if (rw == DDIR_WRITE) {
563 /*
564 * Don't add a write for ro mode
565 */
566 if (read_only)
567 continue;
568 writes++;
569 } else if (rw == DDIR_TRIM) {
570 /*
571 * Don't add a trim for ro mode
572 */
573 if (read_only)
574 continue;
575 trims++;
576 } else if (rw == DDIR_WAIT) {
577 if (td->o.no_stall)
578 continue;
579 waits++;
580 } else if (rw == DDIR_INVAL) {
581 } else if (ddir_sync(rw)) {
582 syncs++;
583 } else {
584 log_err("bad ddir: %d\n", rw);
585 continue;
586 }
587
588 /*
589 * Make note of file
590 */
591 ipo = calloc(1, sizeof(*ipo));
592 init_ipo(ipo);
593 ipo->ddir = rw;
594 if (td->io_log_version == 3)
595 ipo->delay = delay;
596 if (rw == DDIR_WAIT) {
597 ipo->delay = offset;
598 } else {
599 if (td->o.replay_scale)
600 ipo->offset = offset / td->o.replay_scale;
601 else
602 ipo->offset = offset;
603 ipo_bytes_align(td->o.replay_align, ipo);
604
605 ipo->len = bytes;
606 if (rw != DDIR_INVAL && bytes > td->o.max_bs[rw]) {
607 realloc = true;
608 td->o.max_bs[rw] = bytes;
609 }
610 ipo->fileno = fileno;
611 ipo->file_action = file_action;
612 td->o.size += bytes;
613 }
614
615 queue_io_piece(td, ipo);
616
617 if (td->o.read_iolog_chunked) {
618 td->io_log_current++;
619 items_to_fetch--;
620 if (items_to_fetch == 0)
621 break;
622 }
623 }
624
625 free(str);
626 free(act);
627 free(rfname);
628
629 if (td->o.read_iolog_chunked) {
630 td->io_log_highmark = td->io_log_current;
631 td->io_log_checkmark = (td->io_log_highmark + 1) / 2;
632 fio_gettime(&td->io_log_highmark_time, NULL);
633 }
634
635 if (writes && read_only) {
636 log_err("fio: <%s> skips replay of %d writes due to"
637 " read-only\n", td->o.name, writes);
638 writes = 0;
639 }
640 if (syncs)
641 td->flags |= TD_F_SYNCS;
642
643 if (td->o.read_iolog_chunked) {
644 if (td->io_log_current == 0) {
645 return false;
646 }
647 td->o.td_ddir = TD_DDIR_RW;
648 if (realloc && td->orig_buffer)
649 {
650 io_u_quiesce(td);
651 free_io_mem(td);
652 if (init_io_u_buffers(td))
653 return false;
654 }
655 return true;
656 }
657
658 if (!reads && !writes && !waits && !trims)
659 return false;
660
661 td->o.td_ddir = 0;
662 if (reads)
663 td->o.td_ddir |= TD_DDIR_READ;
664 if (writes)
665 td->o.td_ddir |= TD_DDIR_WRITE;
666 if (trims)
667 td->o.td_ddir |= TD_DDIR_TRIM;
668
669 return true;
670}
671
672static bool is_socket(const char *path)
673{
674 struct stat buf;
675 int r;
676
677 r = stat(path, &buf);
678 if (r == -1)
679 return false;
680
681 return S_ISSOCK(buf.st_mode);
682}
683
684static int open_socket(const char *path)
685{
686 struct sockaddr_un addr;
687 int ret, fd;
688
689 fd = socket(AF_UNIX, SOCK_STREAM, 0);
690 if (fd < 0)
691 return fd;
692
693 addr.sun_family = AF_UNIX;
694 if (snprintf(addr.sun_path, sizeof(addr.sun_path), "%s", path) >=
695 sizeof(addr.sun_path)) {
696 log_err("%s: path name %s is too long for a Unix socket\n",
697 __func__, path);
698 }
699
700 ret = connect(fd, (const struct sockaddr *)&addr, strlen(path) + sizeof(addr.sun_family));
701 if (!ret)
702 return fd;
703
704 close(fd);
705 return -1;
706}
707
708/*
709 * open iolog, check version, and call appropriate parser
710 */
711static bool init_iolog_read(struct thread_data *td, char *fname)
712{
713 char buffer[256], *p;
714 FILE *f = NULL;
715
716 dprint(FD_IO, "iolog: name=%s\n", fname);
717
718 if (is_socket(fname)) {
719 int fd;
720
721 fd = open_socket(fname);
722 if (fd >= 0)
723 f = fdopen(fd, "r");
724 } else if (!strcmp(fname, "-")) {
725 f = stdin;
726 } else
727 f = fopen(fname, "r");
728
729 if (!f) {
730 perror("fopen read iolog");
731 return false;
732 }
733
734 p = fgets(buffer, sizeof(buffer), f);
735 if (!p) {
736 td_verror(td, errno, "iolog read");
737 log_err("fio: unable to read iolog\n");
738 fclose(f);
739 return false;
740 }
741
742 /*
743 * versions 2 and 3 of the iolog store a specific string as the
744 * first line, check for that
745 */
746 if (!strncmp(iolog_ver2, buffer, strlen(iolog_ver2)))
747 td->io_log_version = 2;
748 else if (!strncmp(iolog_ver3, buffer, strlen(iolog_ver3)))
749 td->io_log_version = 3;
750 else {
751 log_err("fio: iolog version 1 is no longer supported\n");
752 fclose(f);
753 return false;
754 }
755
756 free_release_files(td);
757 td->io_log_rfile = f;
758 return read_iolog(td);
759}
760
761/*
762 * Set up a log for storing io patterns.
763 */
764static bool init_iolog_write(struct thread_data *td)
765{
766 struct fio_file *ff;
767 FILE *f;
768 unsigned int i;
769
770 f = fopen(td->o.write_iolog_file, "a");
771 if (!f) {
772 perror("fopen write iolog");
773 return false;
774 }
775
776 /*
777 * That's it for writing, setup a log buffer and we're done.
778 */
779 td->iolog_f = f;
780 td->iolog_buf = malloc(8192);
781 setvbuf(f, td->iolog_buf, _IOFBF, 8192);
782 fio_gettime(&td->io_log_start_time, NULL);
783
784 /*
785 * write our version line
786 */
787 if (fprintf(f, "%s\n", iolog_ver3) < 0) {
788 perror("iolog init\n");
789 return false;
790 }
791
792 /*
793 * add all known files
794 */
795 for_each_file(td, ff, i)
796 log_file(td, ff, FIO_LOG_ADD_FILE);
797
798 return true;
799}
800
801bool init_iolog(struct thread_data *td)
802{
803 bool ret;
804
805 if (td->o.read_iolog_file) {
806 int need_swap;
807 char * fname = get_name_by_idx(td->o.read_iolog_file, td->subjob_number);
808
809 /*
810 * Check if it's a blktrace file and load that if possible.
811 * Otherwise assume it's a normal log file and load that.
812 */
813 if (is_blktrace(fname, &need_swap)) {
814 td->io_log_blktrace = 1;
815 ret = init_blktrace_read(td, fname, need_swap);
816 } else {
817 td->io_log_blktrace = 0;
818 ret = init_iolog_read(td, fname);
819 }
820 free(fname);
821 } else if (td->o.write_iolog_file)
822 ret = init_iolog_write(td);
823 else
824 ret = true;
825
826 if (!ret)
827 td_verror(td, EINVAL, "failed initializing iolog");
828
829 init_disk_util(td);
830
831 return ret;
832}
833
834void setup_log(struct io_log **log, struct log_params *p,
835 const char *filename)
836{
837 struct io_log *l;
838 int i;
839 struct io_u_plat_entry *entry;
840 struct flist_head *list;
841
842 l = scalloc(1, sizeof(*l));
843 INIT_FLIST_HEAD(&l->io_logs);
844 l->log_type = p->log_type;
845 l->log_offset = p->log_offset;
846 l->log_prio = p->log_prio;
847 l->log_gz = p->log_gz;
848 l->log_gz_store = p->log_gz_store;
849 l->avg_msec = p->avg_msec;
850 l->hist_msec = p->hist_msec;
851 l->hist_coarseness = p->hist_coarseness;
852 l->filename = strdup(filename);
853 l->td = p->td;
854
855 /* Initialize histogram lists for each r/w direction,
856 * with initial io_u_plat of all zeros:
857 */
858 for (i = 0; i < DDIR_RWDIR_CNT; i++) {
859 list = &l->hist_window[i].list;
860 INIT_FLIST_HEAD(list);
861 entry = calloc(1, sizeof(struct io_u_plat_entry));
862 flist_add(&entry->list, list);
863 }
864
865 if (l->td && l->td->o.io_submit_mode != IO_MODE_OFFLOAD) {
866 unsigned int def_samples = DEF_LOG_ENTRIES;
867 struct io_logs *__p;
868
869 __p = calloc(1, sizeof(*l->pending));
870 if (l->td->o.iodepth > DEF_LOG_ENTRIES)
871 def_samples = roundup_pow2(l->td->o.iodepth);
872 __p->max_samples = def_samples;
873 __p->log = calloc(__p->max_samples, log_entry_sz(l));
874 l->pending = __p;
875 }
876
877 if (l->log_offset)
878 l->log_ddir_mask = LOG_OFFSET_SAMPLE_BIT;
879 if (l->log_prio)
880 l->log_ddir_mask |= LOG_PRIO_SAMPLE_BIT;
881 /*
882 * The bandwidth-log option generates agg-read_bw.log,
883 * agg-write_bw.log and agg-trim_bw.log for which l->td is NULL.
884 * Check if l->td is valid before dereferencing it.
885 */
886 if (l->td && l->td->o.log_max == IO_LOG_SAMPLE_BOTH)
887 l->log_ddir_mask |= LOG_AVG_MAX_SAMPLE_BIT;
888
889 INIT_FLIST_HEAD(&l->chunk_list);
890
891 if (l->log_gz && !p->td)
892 l->log_gz = 0;
893 else if (l->log_gz || l->log_gz_store) {
894 mutex_init_pshared(&l->chunk_lock);
895 mutex_init_pshared(&l->deferred_free_lock);
896 p->td->flags |= TD_F_COMPRESS_LOG;
897 }
898
899 *log = l;
900}
901
902#ifdef CONFIG_SETVBUF
903static void *set_file_buffer(FILE *f)
904{
905 size_t size = 1048576;
906 void *buf;
907
908 buf = malloc(size);
909 setvbuf(f, buf, _IOFBF, size);
910 return buf;
911}
912
913static void clear_file_buffer(void *buf)
914{
915 free(buf);
916}
917#else
918static void *set_file_buffer(FILE *f)
919{
920 return NULL;
921}
922
923static void clear_file_buffer(void *buf)
924{
925}
926#endif
927
928void free_log(struct io_log *log)
929{
930 while (!flist_empty(&log->io_logs)) {
931 struct io_logs *cur_log;
932
933 cur_log = flist_first_entry(&log->io_logs, struct io_logs, list);
934 flist_del_init(&cur_log->list);
935 free(cur_log->log);
936 sfree(cur_log);
937 }
938
939 if (log->pending) {
940 free(log->pending->log);
941 free(log->pending);
942 log->pending = NULL;
943 }
944
945 free(log->pending);
946 free(log->filename);
947 sfree(log);
948}
949
950uint64_t hist_sum(int j, int stride, uint64_t *io_u_plat,
951 uint64_t *io_u_plat_last)
952{
953 uint64_t sum;
954 int k;
955
956 if (io_u_plat_last) {
957 for (k = sum = 0; k < stride; k++)
958 sum += io_u_plat[j + k] - io_u_plat_last[j + k];
959 } else {
960 for (k = sum = 0; k < stride; k++)
961 sum += io_u_plat[j + k];
962 }
963
964 return sum;
965}
966
967static void flush_hist_samples(FILE *f, int hist_coarseness, void *samples,
968 uint64_t sample_size)
969{
970 struct io_sample *s;
971 int log_offset;
972 uint64_t i, j, nr_samples;
973 struct io_u_plat_entry *entry, *entry_before;
974 uint64_t *io_u_plat;
975 uint64_t *io_u_plat_before;
976
977 int stride = 1 << hist_coarseness;
978
979 if (!sample_size)
980 return;
981
982 s = __get_sample(samples, 0, 0);
983 log_offset = (s->__ddir & LOG_OFFSET_SAMPLE_BIT) != 0;
984
985 nr_samples = sample_size / __log_entry_sz(log_offset);
986
987 for (i = 0; i < nr_samples; i++) {
988 s = __get_sample(samples, log_offset, i);
989
990 entry = s->data.plat_entry;
991 io_u_plat = entry->io_u_plat;
992
993 entry_before = flist_first_entry(&entry->list, struct io_u_plat_entry, list);
994 io_u_plat_before = entry_before->io_u_plat;
995
996 fprintf(f, "%lu, %u, %llu, ", (unsigned long) s->time,
997 io_sample_ddir(s), (unsigned long long) s->bs);
998 for (j = 0; j < FIO_IO_U_PLAT_NR - stride; j += stride) {
999 fprintf(f, "%llu, ", (unsigned long long)
1000 hist_sum(j, stride, io_u_plat, io_u_plat_before));
1001 }
1002 fprintf(f, "%llu\n", (unsigned long long)
1003 hist_sum(FIO_IO_U_PLAT_NR - stride, stride, io_u_plat,
1004 io_u_plat_before));
1005
1006 flist_del(&entry_before->list);
1007 free(entry_before);
1008 }
1009}
1010
1011void flush_samples(FILE *f, void *samples, uint64_t sample_size)
1012{
1013 struct io_sample *s;
1014 int log_offset, log_prio, log_avg_max;
1015 uint64_t i, nr_samples;
1016 unsigned int prio_val;
1017 const char *fmt;
1018
1019 if (!sample_size)
1020 return;
1021
1022 s = __get_sample(samples, 0, 0);
1023 log_offset = (s->__ddir & LOG_OFFSET_SAMPLE_BIT) != 0;
1024 log_prio = (s->__ddir & LOG_PRIO_SAMPLE_BIT) != 0;
1025 log_avg_max = (s->__ddir & LOG_AVG_MAX_SAMPLE_BIT) != 0;
1026
1027 if (log_offset) {
1028 if (log_prio) {
1029 if (log_avg_max)
1030 fmt = "%" PRIu64 ", %" PRId64 ", %" PRId64 ", %u, %llu, %llu, 0x%04x\n";
1031 else
1032 fmt = "%" PRIu64 ", %" PRId64 ", %u, %llu, %llu, 0x%04x\n";
1033 } else {
1034 if (log_avg_max)
1035 fmt = "%" PRIu64 ", %" PRId64 ", %" PRId64 ", %u, %llu, %llu, %u\n";
1036 else
1037 fmt = "%" PRIu64 ", %" PRId64 ", %u, %llu, %llu, %u\n";
1038 }
1039 } else {
1040 if (log_prio) {
1041 if (log_avg_max)
1042 fmt = "%" PRIu64 ", %" PRId64 ", %" PRId64 ", %u, %llu, 0x%04x\n";
1043 else
1044 fmt = "%" PRIu64 ", %" PRId64 ", %u, %llu, 0x%04x\n";
1045 } else {
1046 if (log_avg_max)
1047 fmt = "%" PRIu64 ", %" PRId64 ", %" PRId64 ", %u, %llu, %u\n";
1048 else
1049 fmt = "%" PRIu64 ", %" PRId64 ", %u, %llu, %u\n";
1050 }
1051 }
1052
1053 nr_samples = sample_size / __log_entry_sz(log_offset);
1054
1055 for (i = 0; i < nr_samples; i++) {
1056 s = __get_sample(samples, log_offset, i);
1057
1058 if (log_prio)
1059 prio_val = s->priority;
1060 else
1061 prio_val = ioprio_value_is_class_rt(s->priority);
1062
1063 if (!log_offset) {
1064 if (log_avg_max)
1065 fprintf(f, fmt,
1066 s->time,
1067 s->data.val.val0,
1068 s->data.val.val1,
1069 io_sample_ddir(s), (unsigned long long) s->bs,
1070 prio_val);
1071 else
1072 fprintf(f, fmt,
1073 s->time,
1074 s->data.val.val0,
1075 io_sample_ddir(s), (unsigned long long) s->bs,
1076 prio_val);
1077 } else {
1078 struct io_sample_offset *so = (void *) s;
1079
1080 if (log_avg_max)
1081 fprintf(f, fmt,
1082 s->time,
1083 s->data.val.val0,
1084 s->data.val.val1,
1085 io_sample_ddir(s), (unsigned long long) s->bs,
1086 (unsigned long long) so->offset,
1087 prio_val);
1088 else
1089 fprintf(f, fmt,
1090 s->time,
1091 s->data.val.val0,
1092 io_sample_ddir(s), (unsigned long long) s->bs,
1093 (unsigned long long) so->offset,
1094 prio_val);
1095 }
1096 }
1097}
1098
1099#ifdef CONFIG_ZLIB
1100
1101struct iolog_flush_data {
1102 struct workqueue_work work;
1103 struct io_log *log;
1104 void *samples;
1105 uint32_t nr_samples;
1106 bool free;
1107};
1108
1109#define GZ_CHUNK 131072
1110
1111static struct iolog_compress *get_new_chunk(unsigned int seq)
1112{
1113 struct iolog_compress *c;
1114
1115 c = malloc(sizeof(*c));
1116 INIT_FLIST_HEAD(&c->list);
1117 c->buf = malloc(GZ_CHUNK);
1118 c->len = 0;
1119 c->seq = seq;
1120 return c;
1121}
1122
1123static void free_chunk(struct iolog_compress *ic)
1124{
1125 free(ic->buf);
1126 free(ic);
1127}
1128
1129static int z_stream_init(z_stream *stream, int gz_hdr)
1130{
1131 int wbits = 15;
1132
1133 memset(stream, 0, sizeof(*stream));
1134 stream->zalloc = Z_NULL;
1135 stream->zfree = Z_NULL;
1136 stream->opaque = Z_NULL;
1137 stream->next_in = Z_NULL;
1138
1139 /*
1140 * zlib magic - add 32 for auto-detection of gz header or not,
1141 * if we decide to store files in a gzip friendly format.
1142 */
1143 if (gz_hdr)
1144 wbits += 32;
1145
1146 if (inflateInit2(stream, wbits) != Z_OK)
1147 return 1;
1148
1149 return 0;
1150}
1151
1152struct inflate_chunk_iter {
1153 unsigned int seq;
1154 int err;
1155 void *buf;
1156 size_t buf_size;
1157 size_t buf_used;
1158 size_t chunk_sz;
1159};
1160
1161static void finish_chunk(z_stream *stream, FILE *f,
1162 struct inflate_chunk_iter *iter)
1163{
1164 int ret;
1165
1166 ret = inflateEnd(stream);
1167 if (ret != Z_OK)
1168 log_err("fio: failed to end log inflation seq %d (%d)\n",
1169 iter->seq, ret);
1170
1171 flush_samples(f, iter->buf, iter->buf_used);
1172 free(iter->buf);
1173 iter->buf = NULL;
1174 iter->buf_size = iter->buf_used = 0;
1175}
1176
1177/*
1178 * Iterative chunk inflation. Handles cases where we cross into a new
1179 * sequence, doing flush finish of previous chunk if needed.
1180 */
1181static size_t inflate_chunk(struct iolog_compress *ic, int gz_hdr, FILE *f,
1182 z_stream *stream, struct inflate_chunk_iter *iter)
1183{
1184 size_t ret;
1185
1186 dprint(FD_COMPRESS, "inflate chunk size=%lu, seq=%u\n",
1187 (unsigned long) ic->len, ic->seq);
1188
1189 if (ic->seq != iter->seq) {
1190 if (iter->seq)
1191 finish_chunk(stream, f, iter);
1192
1193 z_stream_init(stream, gz_hdr);
1194 iter->seq = ic->seq;
1195 }
1196
1197 stream->avail_in = ic->len;
1198 stream->next_in = ic->buf;
1199
1200 if (!iter->buf_size) {
1201 iter->buf_size = iter->chunk_sz;
1202 iter->buf = malloc(iter->buf_size);
1203 }
1204
1205 while (stream->avail_in) {
1206 size_t this_out = iter->buf_size - iter->buf_used;
1207 int err;
1208
1209 stream->avail_out = this_out;
1210 stream->next_out = iter->buf + iter->buf_used;
1211
1212 err = inflate(stream, Z_NO_FLUSH);
1213 if (err < 0) {
1214 log_err("fio: failed inflating log: %d\n", err);
1215 iter->err = err;
1216 break;
1217 }
1218
1219 iter->buf_used += this_out - stream->avail_out;
1220
1221 if (!stream->avail_out) {
1222 iter->buf_size += iter->chunk_sz;
1223 iter->buf = realloc(iter->buf, iter->buf_size);
1224 continue;
1225 }
1226
1227 if (err == Z_STREAM_END)
1228 break;
1229 }
1230
1231 ret = (void *) stream->next_in - ic->buf;
1232
1233 dprint(FD_COMPRESS, "inflated to size=%lu\n", (unsigned long) iter->buf_size);
1234
1235 return ret;
1236}
1237
1238/*
1239 * Inflate stored compressed chunks, or write them directly to the log
1240 * file if so instructed.
1241 */
1242static int inflate_gz_chunks(struct io_log *log, FILE *f)
1243{
1244 struct inflate_chunk_iter iter = { .chunk_sz = log->log_gz, };
1245 z_stream stream;
1246
1247 while (!flist_empty(&log->chunk_list)) {
1248 struct iolog_compress *ic;
1249
1250 ic = flist_first_entry(&log->chunk_list, struct iolog_compress, list);
1251 flist_del(&ic->list);
1252
1253 if (log->log_gz_store) {
1254 size_t ret;
1255
1256 dprint(FD_COMPRESS, "log write chunk size=%lu, "
1257 "seq=%u\n", (unsigned long) ic->len, ic->seq);
1258
1259 ret = fwrite(ic->buf, ic->len, 1, f);
1260 if (ret != 1 || ferror(f)) {
1261 iter.err = errno;
1262 log_err("fio: error writing compressed log\n");
1263 }
1264 } else
1265 inflate_chunk(ic, log->log_gz_store, f, &stream, &iter);
1266
1267 free_chunk(ic);
1268 }
1269
1270 if (iter.seq) {
1271 finish_chunk(&stream, f, &iter);
1272 free(iter.buf);
1273 }
1274
1275 return iter.err;
1276}
1277
1278/*
1279 * Open compressed log file and decompress the stored chunks and
1280 * write them to stdout. The chunks are stored sequentially in the
1281 * file, so we iterate over them and do them one-by-one.
1282 */
1283int iolog_file_inflate(const char *file)
1284{
1285 struct inflate_chunk_iter iter = { .chunk_sz = 64 * 1024 * 1024, };
1286 struct iolog_compress ic;
1287 z_stream stream;
1288 struct stat sb;
1289 size_t ret;
1290 size_t total;
1291 void *buf;
1292 FILE *f;
1293
1294 f = fopen(file, "rb");
1295 if (!f) {
1296 perror("fopen");
1297 return 1;
1298 }
1299
1300 if (stat(file, &sb) < 0) {
1301 fclose(f);
1302 perror("stat");
1303 return 1;
1304 }
1305
1306 ic.buf = buf = malloc(sb.st_size);
1307 ic.len = sb.st_size;
1308 ic.seq = 1;
1309
1310 ret = fread(ic.buf, ic.len, 1, f);
1311 if (ret == 0 && ferror(f)) {
1312 perror("fread");
1313 fclose(f);
1314 free(buf);
1315 return 1;
1316 } else if (ferror(f) || (!feof(f) && ret != 1)) {
1317 log_err("fio: short read on reading log\n");
1318 fclose(f);
1319 free(buf);
1320 return 1;
1321 }
1322
1323 fclose(f);
1324
1325 /*
1326 * Each chunk will return Z_STREAM_END. We don't know how many
1327 * chunks are in the file, so we just keep looping and incrementing
1328 * the sequence number until we have consumed the whole compressed
1329 * file.
1330 */
1331 total = ic.len;
1332 do {
1333 size_t iret;
1334
1335 iret = inflate_chunk(&ic, 1, stdout, &stream, &iter);
1336 total -= iret;
1337 if (!total)
1338 break;
1339 if (iter.err)
1340 break;
1341
1342 ic.seq++;
1343 ic.len -= iret;
1344 ic.buf += iret;
1345 } while (1);
1346
1347 if (iter.seq) {
1348 finish_chunk(&stream, stdout, &iter);
1349 free(iter.buf);
1350 }
1351
1352 free(buf);
1353 return iter.err;
1354}
1355
1356#else
1357
1358static int inflate_gz_chunks(struct io_log *log, FILE *f)
1359{
1360 return 0;
1361}
1362
1363int iolog_file_inflate(const char *file)
1364{
1365 log_err("fio: log inflation not possible without zlib\n");
1366 return 1;
1367}
1368
1369#endif
1370
1371void flush_log(struct io_log *log, bool do_append)
1372{
1373 void *buf;
1374 FILE *f;
1375
1376 /*
1377 * If log_gz_store is true, we are writing a binary file.
1378 * Set the mode appropriately (on all platforms) to avoid issues
1379 * on windows (line-ending conversions, etc.)
1380 */
1381 if (!do_append)
1382 if (log->log_gz_store)
1383 f = fopen(log->filename, "wb");
1384 else
1385 f = fopen(log->filename, "w");
1386 else
1387 if (log->log_gz_store)
1388 f = fopen(log->filename, "ab");
1389 else
1390 f = fopen(log->filename, "a");
1391 if (!f) {
1392 perror("fopen log");
1393 return;
1394 }
1395
1396 buf = set_file_buffer(f);
1397
1398 inflate_gz_chunks(log, f);
1399
1400 while (!flist_empty(&log->io_logs)) {
1401 struct io_logs *cur_log;
1402
1403 cur_log = flist_first_entry(&log->io_logs, struct io_logs, list);
1404 flist_del_init(&cur_log->list);
1405
1406 if (log->td && log == log->td->clat_hist_log)
1407 flush_hist_samples(f, log->hist_coarseness, cur_log->log,
1408 log_sample_sz(log, cur_log));
1409 else
1410 flush_samples(f, cur_log->log, log_sample_sz(log, cur_log));
1411
1412 sfree(cur_log);
1413 }
1414
1415 fclose(f);
1416 clear_file_buffer(buf);
1417}
1418
1419static int finish_log(struct thread_data *td, struct io_log *log, int trylock)
1420{
1421 if (td->flags & TD_F_COMPRESS_LOG)
1422 iolog_flush(log);
1423
1424 if (trylock) {
1425 if (fio_trylock_file(log->filename))
1426 return 1;
1427 } else
1428 fio_lock_file(log->filename);
1429
1430 if (td->client_type == FIO_CLIENT_TYPE_GUI || is_backend)
1431 fio_send_iolog(td, log, log->filename);
1432 else
1433 flush_log(log, !td->o.per_job_logs);
1434
1435 fio_unlock_file(log->filename);
1436 free_log(log);
1437 return 0;
1438}
1439
1440size_t log_chunk_sizes(struct io_log *log)
1441{
1442 struct flist_head *entry;
1443 size_t ret;
1444
1445 if (flist_empty(&log->chunk_list))
1446 return 0;
1447
1448 ret = 0;
1449 pthread_mutex_lock(&log->chunk_lock);
1450 flist_for_each(entry, &log->chunk_list) {
1451 struct iolog_compress *c;
1452
1453 c = flist_entry(entry, struct iolog_compress, list);
1454 ret += c->len;
1455 }
1456 pthread_mutex_unlock(&log->chunk_lock);
1457 return ret;
1458}
1459
1460#ifdef CONFIG_ZLIB
1461
1462static void iolog_put_deferred(struct io_log *log, void *ptr)
1463{
1464 if (!ptr)
1465 return;
1466
1467 pthread_mutex_lock(&log->deferred_free_lock);
1468 if (log->deferred < IOLOG_MAX_DEFER) {
1469 log->deferred_items[log->deferred] = ptr;
1470 log->deferred++;
1471 } else if (!fio_did_warn(FIO_WARN_IOLOG_DROP))
1472 log_err("fio: had to drop log entry free\n");
1473 pthread_mutex_unlock(&log->deferred_free_lock);
1474}
1475
1476static void iolog_free_deferred(struct io_log *log)
1477{
1478 int i;
1479
1480 if (!log->deferred)
1481 return;
1482
1483 pthread_mutex_lock(&log->deferred_free_lock);
1484
1485 for (i = 0; i < log->deferred; i++) {
1486 free(log->deferred_items[i]);
1487 log->deferred_items[i] = NULL;
1488 }
1489
1490 log->deferred = 0;
1491 pthread_mutex_unlock(&log->deferred_free_lock);
1492}
1493
1494static int gz_work(struct iolog_flush_data *data)
1495{
1496 struct iolog_compress *c = NULL;
1497 struct flist_head list;
1498 unsigned int seq;
1499 z_stream stream;
1500 size_t total = 0;
1501 int ret;
1502
1503 INIT_FLIST_HEAD(&list);
1504
1505 memset(&stream, 0, sizeof(stream));
1506 stream.zalloc = Z_NULL;
1507 stream.zfree = Z_NULL;
1508 stream.opaque = Z_NULL;
1509
1510 ret = deflateInit(&stream, Z_DEFAULT_COMPRESSION);
1511 if (ret != Z_OK) {
1512 log_err("fio: failed to init gz stream\n");
1513 goto err;
1514 }
1515
1516 seq = ++data->log->chunk_seq;
1517
1518 stream.next_in = (void *) data->samples;
1519 stream.avail_in = data->nr_samples * log_entry_sz(data->log);
1520
1521 dprint(FD_COMPRESS, "deflate input size=%lu, seq=%u, log=%s\n",
1522 (unsigned long) stream.avail_in, seq,
1523 data->log->filename);
1524 do {
1525 if (c)
1526 dprint(FD_COMPRESS, "seq=%d, chunk=%lu\n", seq,
1527 (unsigned long) c->len);
1528 c = get_new_chunk(seq);
1529 stream.avail_out = GZ_CHUNK;
1530 stream.next_out = c->buf;
1531 ret = deflate(&stream, Z_NO_FLUSH);
1532 if (ret < 0) {
1533 log_err("fio: deflate log (%d)\n", ret);
1534 free_chunk(c);
1535 goto err;
1536 }
1537
1538 c->len = GZ_CHUNK - stream.avail_out;
1539 flist_add_tail(&c->list, &list);
1540 total += c->len;
1541 } while (stream.avail_in);
1542
1543 stream.next_out = c->buf + c->len;
1544 stream.avail_out = GZ_CHUNK - c->len;
1545
1546 ret = deflate(&stream, Z_FINISH);
1547 if (ret < 0) {
1548 /*
1549 * Z_BUF_ERROR is special, it just means we need more
1550 * output space. We'll handle that below. Treat any other
1551 * error as fatal.
1552 */
1553 if (ret != Z_BUF_ERROR) {
1554 log_err("fio: deflate log (%d)\n", ret);
1555 flist_del(&c->list);
1556 free_chunk(c);
1557 goto err;
1558 }
1559 }
1560
1561 total -= c->len;
1562 c->len = GZ_CHUNK - stream.avail_out;
1563 total += c->len;
1564 dprint(FD_COMPRESS, "seq=%d, chunk=%lu\n", seq, (unsigned long) c->len);
1565
1566 if (ret != Z_STREAM_END) {
1567 do {
1568 c = get_new_chunk(seq);
1569 stream.avail_out = GZ_CHUNK;
1570 stream.next_out = c->buf;
1571 ret = deflate(&stream, Z_FINISH);
1572 c->len = GZ_CHUNK - stream.avail_out;
1573 total += c->len;
1574 flist_add_tail(&c->list, &list);
1575 dprint(FD_COMPRESS, "seq=%d, chunk=%lu\n", seq,
1576 (unsigned long) c->len);
1577 } while (ret != Z_STREAM_END);
1578 }
1579
1580 dprint(FD_COMPRESS, "deflated to size=%lu\n", (unsigned long) total);
1581
1582 ret = deflateEnd(&stream);
1583 if (ret != Z_OK)
1584 log_err("fio: deflateEnd %d\n", ret);
1585
1586 iolog_put_deferred(data->log, data->samples);
1587
1588 if (!flist_empty(&list)) {
1589 pthread_mutex_lock(&data->log->chunk_lock);
1590 flist_splice_tail(&list, &data->log->chunk_list);
1591 pthread_mutex_unlock(&data->log->chunk_lock);
1592 }
1593
1594 ret = 0;
1595done:
1596 if (data->free)
1597 sfree(data);
1598 return ret;
1599err:
1600 while (!flist_empty(&list)) {
1601 c = flist_first_entry(list.next, struct iolog_compress, list);
1602 flist_del(&c->list);
1603 free_chunk(c);
1604 }
1605 ret = 1;
1606 goto done;
1607}
1608
1609/*
1610 * Invoked from our compress helper thread, when logging would have exceeded
1611 * the specified memory limitation. Compresses the previously stored
1612 * entries.
1613 */
1614static int gz_work_async(struct submit_worker *sw, struct workqueue_work *work)
1615{
1616 return gz_work(container_of(work, struct iolog_flush_data, work));
1617}
1618
1619static int gz_init_worker(struct submit_worker *sw)
1620{
1621 struct thread_data *td = sw->wq->td;
1622
1623 if (!fio_option_is_set(&td->o, log_gz_cpumask))
1624 return 0;
1625
1626 if (fio_setaffinity(gettid(), td->o.log_gz_cpumask) == -1) {
1627 log_err("gz: failed to set CPU affinity\n");
1628 return 1;
1629 }
1630
1631 return 0;
1632}
1633
1634static struct workqueue_ops log_compress_wq_ops = {
1635 .fn = gz_work_async,
1636 .init_worker_fn = gz_init_worker,
1637 .nice = 1,
1638};
1639
1640int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out)
1641{
1642 if (!(td->flags & TD_F_COMPRESS_LOG))
1643 return 0;
1644
1645 workqueue_init(td, &td->log_compress_wq, &log_compress_wq_ops, 1, sk_out);
1646 return 0;
1647}
1648
1649void iolog_compress_exit(struct thread_data *td)
1650{
1651 if (!(td->flags & TD_F_COMPRESS_LOG))
1652 return;
1653
1654 workqueue_exit(&td->log_compress_wq);
1655}
1656
1657/*
1658 * Queue work item to compress the existing log entries. We reset the
1659 * current log to a small size, and reference the existing log in the
1660 * data that we queue for compression. Once compression has been done,
1661 * this old log is freed. Will not return until the log compression
1662 * has completed, and will flush all previous logs too
1663 */
1664static int iolog_flush(struct io_log *log)
1665{
1666 struct iolog_flush_data *data;
1667
1668 workqueue_flush(&log->td->log_compress_wq);
1669 data = malloc(sizeof(*data));
1670 if (!data)
1671 return 1;
1672
1673 data->log = log;
1674 data->free = false;
1675
1676 while (!flist_empty(&log->io_logs)) {
1677 struct io_logs *cur_log;
1678
1679 cur_log = flist_first_entry(&log->io_logs, struct io_logs, list);
1680 flist_del_init(&cur_log->list);
1681
1682 data->samples = cur_log->log;
1683 data->nr_samples = cur_log->nr_samples;
1684
1685 sfree(cur_log);
1686
1687 gz_work(data);
1688 }
1689
1690 free(data);
1691 return 0;
1692}
1693
1694int iolog_cur_flush(struct io_log *log, struct io_logs *cur_log)
1695{
1696 struct iolog_flush_data *data;
1697
1698 data = smalloc(sizeof(*data));
1699 if (!data)
1700 return 1;
1701
1702 data->log = log;
1703
1704 data->samples = cur_log->log;
1705 data->nr_samples = cur_log->nr_samples;
1706 data->free = true;
1707
1708 cur_log->nr_samples = cur_log->max_samples = 0;
1709 cur_log->log = NULL;
1710
1711 workqueue_enqueue(&log->td->log_compress_wq, &data->work);
1712
1713 iolog_free_deferred(log);
1714
1715 return 0;
1716}
1717#else
1718
1719static int iolog_flush(struct io_log *log)
1720{
1721 return 1;
1722}
1723
1724int iolog_cur_flush(struct io_log *log, struct io_logs *cur_log)
1725{
1726 return 1;
1727}
1728
1729int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out)
1730{
1731 return 0;
1732}
1733
1734void iolog_compress_exit(struct thread_data *td)
1735{
1736}
1737
1738#endif
1739
1740struct io_logs *iolog_cur_log(struct io_log *log)
1741{
1742 if (flist_empty(&log->io_logs))
1743 return NULL;
1744
1745 return flist_last_entry(&log->io_logs, struct io_logs, list);
1746}
1747
1748uint64_t iolog_nr_samples(struct io_log *iolog)
1749{
1750 struct flist_head *entry;
1751 uint64_t ret = 0;
1752
1753 flist_for_each(entry, &iolog->io_logs) {
1754 struct io_logs *cur_log;
1755
1756 cur_log = flist_entry(entry, struct io_logs, list);
1757 ret += cur_log->nr_samples;
1758 }
1759
1760 return ret;
1761}
1762
1763static int __write_log(struct thread_data *td, struct io_log *log, int try)
1764{
1765 if (log)
1766 return finish_log(td, log, try);
1767
1768 return 0;
1769}
1770
1771static int write_iops_log(struct thread_data *td, int try, bool unit_log)
1772{
1773 int ret;
1774
1775 if (per_unit_log(td->iops_log) != unit_log)
1776 return 0;
1777
1778 ret = __write_log(td, td->iops_log, try);
1779 if (!ret)
1780 td->iops_log = NULL;
1781
1782 return ret;
1783}
1784
1785static int write_slat_log(struct thread_data *td, int try, bool unit_log)
1786{
1787 int ret;
1788
1789 if (!unit_log)
1790 return 0;
1791
1792 ret = __write_log(td, td->slat_log, try);
1793 if (!ret)
1794 td->slat_log = NULL;
1795
1796 return ret;
1797}
1798
1799static int write_clat_log(struct thread_data *td, int try, bool unit_log)
1800{
1801 int ret;
1802
1803 if (!unit_log)
1804 return 0;
1805
1806 ret = __write_log(td, td->clat_log, try);
1807 if (!ret)
1808 td->clat_log = NULL;
1809
1810 return ret;
1811}
1812
1813static int write_clat_hist_log(struct thread_data *td, int try, bool unit_log)
1814{
1815 int ret;
1816
1817 if (!unit_log)
1818 return 0;
1819
1820 ret = __write_log(td, td->clat_hist_log, try);
1821 if (!ret)
1822 td->clat_hist_log = NULL;
1823
1824 return ret;
1825}
1826
1827static int write_lat_log(struct thread_data *td, int try, bool unit_log)
1828{
1829 int ret;
1830
1831 if (!unit_log)
1832 return 0;
1833
1834 ret = __write_log(td, td->lat_log, try);
1835 if (!ret)
1836 td->lat_log = NULL;
1837
1838 return ret;
1839}
1840
1841static int write_bandw_log(struct thread_data *td, int try, bool unit_log)
1842{
1843 int ret;
1844
1845 if (per_unit_log(td->bw_log) != unit_log)
1846 return 0;
1847
1848 ret = __write_log(td, td->bw_log, try);
1849 if (!ret)
1850 td->bw_log = NULL;
1851
1852 return ret;
1853}
1854
1855enum {
1856 BW_LOG_MASK = 1,
1857 LAT_LOG_MASK = 2,
1858 SLAT_LOG_MASK = 4,
1859 CLAT_LOG_MASK = 8,
1860 IOPS_LOG_MASK = 16,
1861 CLAT_HIST_LOG_MASK = 32,
1862
1863 ALL_LOG_NR = 6,
1864};
1865
1866struct log_type {
1867 unsigned int mask;
1868 int (*fn)(struct thread_data *, int, bool);
1869};
1870
1871static struct log_type log_types[] = {
1872 {
1873 .mask = BW_LOG_MASK,
1874 .fn = write_bandw_log,
1875 },
1876 {
1877 .mask = LAT_LOG_MASK,
1878 .fn = write_lat_log,
1879 },
1880 {
1881 .mask = SLAT_LOG_MASK,
1882 .fn = write_slat_log,
1883 },
1884 {
1885 .mask = CLAT_LOG_MASK,
1886 .fn = write_clat_log,
1887 },
1888 {
1889 .mask = IOPS_LOG_MASK,
1890 .fn = write_iops_log,
1891 },
1892 {
1893 .mask = CLAT_HIST_LOG_MASK,
1894 .fn = write_clat_hist_log,
1895 }
1896};
1897
1898void td_writeout_logs(struct thread_data *td, bool unit_logs)
1899{
1900 unsigned int log_mask = 0;
1901 unsigned int log_left = ALL_LOG_NR;
1902 int old_state, i;
1903
1904 old_state = td_bump_runstate(td, TD_FINISHING);
1905
1906 finalize_logs(td, unit_logs);
1907
1908 while (log_left) {
1909 int prev_log_left = log_left;
1910
1911 for (i = 0; i < ALL_LOG_NR && log_left; i++) {
1912 struct log_type *lt = &log_types[i];
1913 int ret;
1914
1915 if (!(log_mask & lt->mask)) {
1916 ret = lt->fn(td, log_left != 1, unit_logs);
1917 if (!ret) {
1918 log_left--;
1919 log_mask |= lt->mask;
1920 }
1921 }
1922 }
1923
1924 if (prev_log_left == log_left)
1925 usleep(5000);
1926 }
1927
1928 td_restore_runstate(td, old_state);
1929}
1930
1931void fio_writeout_logs(bool unit_logs)
1932{
1933 for_each_td(td) {
1934 td_writeout_logs(td, unit_logs);
1935 } end_for_each();
1936}
fio - Flexible IO Tester