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