iolog: reinstate log compression nice level
[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>
7#include <libgen.h>
8#include <assert.h>
b26317c9
JA
9#include <sys/types.h>
10#include <sys/stat.h>
11#include <unistd.h>
aee2ab67
JA
12#ifdef CONFIG_ZLIB
13#include <zlib.h>
14#endif
15
ac9b9101
JA
16#include "flist.h"
17#include "fio.h"
18#include "verify.h"
19#include "trim.h"
243bfe19 20#include "filelock.h"
ac9b9101
JA
21
22static const char iolog_ver2[] = "fio version 2 iolog";
23
24void queue_io_piece(struct thread_data *td, struct io_piece *ipo)
25{
26 flist_add_tail(&ipo->list, &td->io_log_list);
27 td->total_io_size += ipo->len;
28}
29
1f440ece 30void log_io_u(const struct thread_data *td, const struct io_u *io_u)
ac9b9101 31{
ac9b9101
JA
32 if (!td->o.write_iolog_file)
33 return;
34
35 fprintf(td->iolog_f, "%s %s %llu %lu\n", io_u->file->file_name,
173375c2
JA
36 io_ddir_name(io_u->ddir),
37 io_u->offset, io_u->buflen);
ac9b9101
JA
38}
39
40void log_file(struct thread_data *td, struct fio_file *f,
41 enum file_log_act what)
42{
43 const char *act[] = { "add", "open", "close" };
44
45 assert(what < 3);
46
47 if (!td->o.write_iolog_file)
48 return;
49
50
51 /*
52 * this happens on the pre-open/close done before the job starts
53 */
54 if (!td->iolog_f)
55 return;
56
57 fprintf(td->iolog_f, "%s %s\n", f->file_name, act[what]);
58}
59
60static void iolog_delay(struct thread_data *td, unsigned long delay)
61{
df8fb609
JA
62 uint64_t usec = utime_since_now(&td->last_issue);
63 uint64_t this_delay;
64 struct timeval tv;
ac9b9101 65
df8fb609
JA
66 if (delay < td->time_offset) {
67 td->time_offset = 0;
68 return;
69 }
70
71 delay -= td->time_offset;
ac9b9101
JA
72 if (delay < usec)
73 return;
74
75 delay -= usec;
76
df8fb609 77 fio_gettime(&tv, NULL);
30b18672
JA
78 while (delay && !td->terminate) {
79 this_delay = delay;
80 if (this_delay > 500000)
81 this_delay = 500000;
82
83 usec_sleep(td, this_delay);
84 delay -= this_delay;
85 }
df8fb609
JA
86
87 usec = utime_since_now(&tv);
88 if (usec > delay)
89 td->time_offset = usec - delay;
90 else
91 td->time_offset = 0;
ac9b9101
JA
92}
93
94static int ipo_special(struct thread_data *td, struct io_piece *ipo)
95{
96 struct fio_file *f;
97 int ret;
98
99 /*
100 * Not a special ipo
101 */
102 if (ipo->ddir != DDIR_INVAL)
103 return 0;
104
105 f = td->files[ipo->fileno];
106
107 switch (ipo->file_action) {
108 case FIO_LOG_OPEN_FILE:
109 ret = td_io_open_file(td, f);
110 if (!ret)
111 break;
112 td_verror(td, ret, "iolog open file");
113 return -1;
114 case FIO_LOG_CLOSE_FILE:
115 td_io_close_file(td, f);
116 break;
117 case FIO_LOG_UNLINK_FILE:
38ef9c90 118 td_io_unlink_file(td, f);
ac9b9101
JA
119 break;
120 default:
121 log_err("fio: bad file action %d\n", ipo->file_action);
122 break;
123 }
124
125 return 1;
126}
127
128int read_iolog_get(struct thread_data *td, struct io_u *io_u)
129{
130 struct io_piece *ipo;
131 unsigned long elapsed;
3c3ed070 132
ac9b9101
JA
133 while (!flist_empty(&td->io_log_list)) {
134 int ret;
135
9342d5f8 136 ipo = flist_first_entry(&td->io_log_list, struct io_piece, list);
ac9b9101
JA
137 flist_del(&ipo->list);
138 remove_trim_entry(td, ipo);
139
140 ret = ipo_special(td, ipo);
141 if (ret < 0) {
142 free(ipo);
143 break;
144 } else if (ret > 0) {
145 free(ipo);
146 continue;
147 }
148
149 io_u->ddir = ipo->ddir;
150 if (ipo->ddir != DDIR_WAIT) {
151 io_u->offset = ipo->offset;
152 io_u->buflen = ipo->len;
153 io_u->file = td->files[ipo->fileno];
154 get_file(io_u->file);
155 dprint(FD_IO, "iolog: get %llu/%lu/%s\n", io_u->offset,
156 io_u->buflen, io_u->file->file_name);
157 if (ipo->delay)
158 iolog_delay(td, ipo->delay);
159 } else {
160 elapsed = mtime_since_genesis();
161 if (ipo->delay > elapsed)
162 usec_sleep(td, (ipo->delay - elapsed) * 1000);
ac9b9101
JA
163 }
164
165 free(ipo);
3c3ed070 166
ac9b9101
JA
167 if (io_u->ddir != DDIR_WAIT)
168 return 0;
169 }
170
171 td->done = 1;
172 return 1;
173}
174
175void prune_io_piece_log(struct thread_data *td)
176{
177 struct io_piece *ipo;
178 struct rb_node *n;
179
180 while ((n = rb_first(&td->io_hist_tree)) != NULL) {
181 ipo = rb_entry(n, struct io_piece, rb_node);
182 rb_erase(n, &td->io_hist_tree);
183 remove_trim_entry(td, ipo);
184 td->io_hist_len--;
185 free(ipo);
186 }
187
188 while (!flist_empty(&td->io_hist_list)) {
4e1020f8 189 ipo = flist_first_entry(&td->io_hist_list, struct io_piece, list);
ac9b9101
JA
190 flist_del(&ipo->list);
191 remove_trim_entry(td, ipo);
192 td->io_hist_len--;
193 free(ipo);
194 }
195}
196
197/*
198 * log a successful write, so we can unwind the log for verify
199 */
200void log_io_piece(struct thread_data *td, struct io_u *io_u)
201{
202 struct rb_node **p, *parent;
203 struct io_piece *ipo, *__ipo;
204
205 ipo = malloc(sizeof(struct io_piece));
206 init_ipo(ipo);
207 ipo->file = io_u->file;
208 ipo->offset = io_u->offset;
209 ipo->len = io_u->buflen;
da0a7bd2 210 ipo->numberio = io_u->numberio;
f9401285
JA
211 ipo->flags = IP_F_IN_FLIGHT;
212
213 io_u->ipo = ipo;
ac9b9101
JA
214
215 if (io_u_should_trim(td, io_u)) {
216 flist_add_tail(&ipo->trim_list, &td->trim_list);
217 td->trim_entries++;
218 }
219
220 /*
221 * We don't need to sort the entries, if:
222 *
223 * Sequential writes, or
224 * Random writes that lay out the file as it goes along
225 *
226 * For both these cases, just reading back data in the order we
227 * wrote it out is the fastest.
228 *
229 * One exception is if we don't have a random map AND we are doing
230 * verifies, in that case we need to check for duplicate blocks and
231 * drop the old one, which we rely on the rb insert/lookup for
232 * handling.
233 */
c4b6117b 234 if (((!td->o.verifysort) || !td_random(td) || !td->o.overwrite) &&
ac9b9101
JA
235 (file_randommap(td, ipo->file) || td->o.verify == VERIFY_NONE)) {
236 INIT_FLIST_HEAD(&ipo->list);
237 flist_add_tail(&ipo->list, &td->io_hist_list);
238 ipo->flags |= IP_F_ONLIST;
239 td->io_hist_len++;
240 return;
241 }
242
243 RB_CLEAR_NODE(&ipo->rb_node);
244
245 /*
246 * Sort the entry into the verification list
247 */
248restart:
249 p = &td->io_hist_tree.rb_node;
250 parent = NULL;
251 while (*p) {
3b1ff869 252 int overlap = 0;
ac9b9101
JA
253 parent = *p;
254
255 __ipo = rb_entry(parent, struct io_piece, rb_node);
256 if (ipo->file < __ipo->file)
257 p = &(*p)->rb_left;
258 else if (ipo->file > __ipo->file)
259 p = &(*p)->rb_right;
3b1ff869 260 else if (ipo->offset < __ipo->offset) {
ac9b9101 261 p = &(*p)->rb_left;
3b1ff869
JE
262 overlap = ipo->offset + ipo->len > __ipo->offset;
263 }
264 else if (ipo->offset > __ipo->offset) {
ac9b9101 265 p = &(*p)->rb_right;
3b1ff869
JE
266 overlap = __ipo->offset + __ipo->len > ipo->offset;
267 }
268 else
269 overlap = 1;
270
271 if (overlap) {
885ac623
JA
272 dprint(FD_IO, "iolog: overlap %llu/%lu, %llu/%lu",
273 __ipo->offset, __ipo->len,
274 ipo->offset, ipo->len);
ac9b9101
JA
275 td->io_hist_len--;
276 rb_erase(parent, &td->io_hist_tree);
277 remove_trim_entry(td, __ipo);
278 free(__ipo);
279 goto restart;
280 }
281 }
282
283 rb_link_node(&ipo->rb_node, parent, p);
284 rb_insert_color(&ipo->rb_node, &td->io_hist_tree);
285 ipo->flags |= IP_F_ONRB;
286 td->io_hist_len++;
287}
288
890b6656
JA
289void unlog_io_piece(struct thread_data *td, struct io_u *io_u)
290{
291 struct io_piece *ipo = io_u->ipo;
292
66347cfa
DE
293 if (td->ts.nr_block_infos) {
294 uint32_t *info = io_u_block_info(td, io_u);
295 if (BLOCK_INFO_STATE(*info) < BLOCK_STATE_TRIM_FAILURE) {
296 if (io_u->ddir == DDIR_TRIM)
297 *info = BLOCK_INFO_SET_STATE(*info,
298 BLOCK_STATE_TRIM_FAILURE);
299 else if (io_u->ddir == DDIR_WRITE)
300 *info = BLOCK_INFO_SET_STATE(*info,
301 BLOCK_STATE_WRITE_FAILURE);
302 }
303 }
304
890b6656
JA
305 if (!ipo)
306 return;
307
308 if (ipo->flags & IP_F_ONRB)
309 rb_erase(&ipo->rb_node, &td->io_hist_tree);
310 else if (ipo->flags & IP_F_ONLIST)
311 flist_del(&ipo->list);
312
313 free(ipo);
314 io_u->ipo = NULL;
315 td->io_hist_len--;
316}
317
1f440ece 318void trim_io_piece(struct thread_data *td, const struct io_u *io_u)
890b6656
JA
319{
320 struct io_piece *ipo = io_u->ipo;
321
322 if (!ipo)
323 return;
324
325 ipo->len = io_u->xfer_buflen - io_u->resid;
326}
327
ac9b9101
JA
328void write_iolog_close(struct thread_data *td)
329{
330 fflush(td->iolog_f);
331 fclose(td->iolog_f);
332 free(td->iolog_buf);
333 td->iolog_f = NULL;
334 td->iolog_buf = NULL;
335}
336
337/*
338 * Read version 2 iolog data. It is enhanced to include per-file logging,
339 * syncs, etc.
340 */
341static int read_iolog2(struct thread_data *td, FILE *f)
342{
343 unsigned long long offset;
344 unsigned int bytes;
345 int reads, writes, waits, fileno = 0, file_action = 0; /* stupid gcc */
346 char *fname, *act;
347 char *str, *p;
348 enum fio_ddir rw;
349
350 free_release_files(td);
351
352 /*
353 * Read in the read iolog and store it, reuse the infrastructure
354 * for doing verifications.
355 */
356 str = malloc(4096);
357 fname = malloc(256+16);
358 act = malloc(256+16);
359
360 reads = writes = waits = 0;
361 while ((p = fgets(str, 4096, f)) != NULL) {
362 struct io_piece *ipo;
363 int r;
364
365 r = sscanf(p, "%256s %256s %llu %u", fname, act, &offset,
366 &bytes);
367 if (r == 4) {
368 /*
369 * Check action first
370 */
371 if (!strcmp(act, "wait"))
372 rw = DDIR_WAIT;
373 else if (!strcmp(act, "read"))
374 rw = DDIR_READ;
375 else if (!strcmp(act, "write"))
376 rw = DDIR_WRITE;
377 else if (!strcmp(act, "sync"))
378 rw = DDIR_SYNC;
379 else if (!strcmp(act, "datasync"))
380 rw = DDIR_DATASYNC;
381 else if (!strcmp(act, "trim"))
382 rw = DDIR_TRIM;
383 else {
384 log_err("fio: bad iolog file action: %s\n",
385 act);
386 continue;
387 }
033ace1e 388 fileno = get_fileno(td, fname);
ac9b9101
JA
389 } else if (r == 2) {
390 rw = DDIR_INVAL;
391 if (!strcmp(act, "add")) {
5903e7b7 392 fileno = add_file(td, fname, 0, 1);
ac9b9101
JA
393 file_action = FIO_LOG_ADD_FILE;
394 continue;
395 } else if (!strcmp(act, "open")) {
396 fileno = get_fileno(td, fname);
397 file_action = FIO_LOG_OPEN_FILE;
398 } else if (!strcmp(act, "close")) {
399 fileno = get_fileno(td, fname);
400 file_action = FIO_LOG_CLOSE_FILE;
401 } else {
402 log_err("fio: bad iolog file action: %s\n",
403 act);
404 continue;
405 }
406 } else {
407 log_err("bad iolog2: %s", p);
408 continue;
409 }
410
411 if (rw == DDIR_READ)
412 reads++;
413 else if (rw == DDIR_WRITE) {
414 /*
415 * Don't add a write for ro mode
416 */
417 if (read_only)
418 continue;
419 writes++;
420 } else if (rw == DDIR_WAIT) {
421 waits++;
422 } else if (rw == DDIR_INVAL) {
423 } else if (!ddir_sync(rw)) {
424 log_err("bad ddir: %d\n", rw);
425 continue;
426 }
427
428 /*
429 * Make note of file
430 */
431 ipo = malloc(sizeof(*ipo));
432 init_ipo(ipo);
433 ipo->ddir = rw;
434 if (rw == DDIR_WAIT) {
435 ipo->delay = offset;
436 } else {
437 ipo->offset = offset;
438 ipo->len = bytes;
42793d94 439 if (rw != DDIR_INVAL && bytes > td->o.max_bs[rw])
ac9b9101
JA
440 td->o.max_bs[rw] = bytes;
441 ipo->fileno = fileno;
442 ipo->file_action = file_action;
70afff59 443 td->o.size += bytes;
ac9b9101 444 }
3c3ed070 445
ac9b9101
JA
446 queue_io_piece(td, ipo);
447 }
448
449 free(str);
450 free(act);
451 free(fname);
452
453 if (writes && read_only) {
454 log_err("fio: <%s> skips replay of %d writes due to"
455 " read-only\n", td->o.name, writes);
456 writes = 0;
457 }
458
459 if (!reads && !writes && !waits)
460 return 1;
461 else if (reads && !writes)
462 td->o.td_ddir = TD_DDIR_READ;
463 else if (!reads && writes)
464 td->o.td_ddir = TD_DDIR_WRITE;
465 else
466 td->o.td_ddir = TD_DDIR_RW;
467
468 return 0;
469}
470
471/*
472 * open iolog, check version, and call appropriate parser
473 */
474static int init_iolog_read(struct thread_data *td)
475{
476 char buffer[256], *p;
477 FILE *f;
478 int ret;
479
480 f = fopen(td->o.read_iolog_file, "r");
481 if (!f) {
482 perror("fopen read iolog");
483 return 1;
484 }
485
486 p = fgets(buffer, sizeof(buffer), f);
487 if (!p) {
488 td_verror(td, errno, "iolog read");
489 log_err("fio: unable to read iolog\n");
490 fclose(f);
491 return 1;
492 }
493
494 /*
495 * version 2 of the iolog stores a specific string as the
496 * first line, check for that
497 */
498 if (!strncmp(iolog_ver2, buffer, strlen(iolog_ver2)))
499 ret = read_iolog2(td, f);
500 else {
501 log_err("fio: iolog version 1 is no longer supported\n");
502 ret = 1;
503 }
504
505 fclose(f);
506 return ret;
507}
508
509/*
510 * Set up a log for storing io patterns.
511 */
512static int init_iolog_write(struct thread_data *td)
513{
514 struct fio_file *ff;
515 FILE *f;
516 unsigned int i;
517
518 f = fopen(td->o.write_iolog_file, "a");
519 if (!f) {
520 perror("fopen write iolog");
521 return 1;
522 }
523
524 /*
525 * That's it for writing, setup a log buffer and we're done.
526 */
527 td->iolog_f = f;
528 td->iolog_buf = malloc(8192);
529 setvbuf(f, td->iolog_buf, _IOFBF, 8192);
530
531 /*
532 * write our version line
533 */
534 if (fprintf(f, "%s\n", iolog_ver2) < 0) {
535 perror("iolog init\n");
536 return 1;
537 }
538
539 /*
540 * add all known files
541 */
542 for_each_file(td, ff, i)
543 log_file(td, ff, FIO_LOG_ADD_FILE);
544
545 return 0;
546}
547
548int init_iolog(struct thread_data *td)
549{
550 int ret = 0;
551
552 if (td->o.read_iolog_file) {
d95b34a6
JA
553 int need_swap;
554
ac9b9101
JA
555 /*
556 * Check if it's a blktrace file and load that if possible.
557 * Otherwise assume it's a normal log file and load that.
558 */
d95b34a6
JA
559 if (is_blktrace(td->o.read_iolog_file, &need_swap))
560 ret = load_blktrace(td, td->o.read_iolog_file, need_swap);
ac9b9101
JA
561 else
562 ret = init_iolog_read(td);
563 } else if (td->o.write_iolog_file)
564 ret = init_iolog_write(td);
565
f01b34ae
JA
566 if (ret)
567 td_verror(td, EINVAL, "failed initializing iolog");
568
ac9b9101
JA
569 return ret;
570}
571
aee2ab67
JA
572void setup_log(struct io_log **log, struct log_params *p,
573 const char *filename)
ac9b9101 574{
e75cc8f3 575 struct io_log *l;
ac9b9101 576
e75cc8f3 577 l = calloc(1, sizeof(*l));
ac9b9101
JA
578 l->nr_samples = 0;
579 l->max_samples = 1024;
aee2ab67
JA
580 l->log_type = p->log_type;
581 l->log_offset = p->log_offset;
582 l->log_gz = p->log_gz;
b26317c9 583 l->log_gz_store = p->log_gz_store;
ae588852 584 l->log = malloc(l->max_samples * log_entry_sz(l));
aee2ab67 585 l->avg_msec = p->avg_msec;
cb7e0ace 586 l->filename = strdup(filename);
aee2ab67
JA
587 l->td = p->td;
588
b26317c9 589 if (l->log_offset)
49e98daa 590 l->log_ddir_mask = LOG_OFFSET_SAMPLE_BIT;
b26317c9 591
aee2ab67
JA
592 INIT_FLIST_HEAD(&l->chunk_list);
593
594 if (l->log_gz && !p->td)
595 l->log_gz = 0;
19417eda 596 else if (l->log_gz || l->log_gz_store) {
aee2ab67
JA
597 pthread_mutex_init(&l->chunk_lock, NULL);
598 p->td->flags |= TD_F_COMPRESS_LOG;
599 }
600
ac9b9101
JA
601 *log = l;
602}
603
2e802282
JA
604#ifdef CONFIG_SETVBUF
605static void *set_file_buffer(FILE *f)
606{
607 size_t size = 1048576;
608 void *buf;
609
610 buf = malloc(size);
611 setvbuf(f, buf, _IOFBF, size);
612 return buf;
613}
614
615static void clear_file_buffer(void *buf)
616{
617 free(buf);
618}
619#else
620static void *set_file_buffer(FILE *f)
621{
622 return NULL;
623}
624
625static void clear_file_buffer(void *buf)
626{
627}
628#endif
629
518dac09 630void free_log(struct io_log *log)
cb7e0ace
JA
631{
632 free(log->log);
633 free(log->filename);
634 free(log);
635}
636
b26317c9 637static void flush_samples(FILE *f, void *samples, uint64_t sample_size)
ac9b9101 638{
b26317c9
JA
639 struct io_sample *s;
640 int log_offset;
641 uint64_t i, nr_samples;
642
643 if (!sample_size)
644 return;
645
646 s = __get_sample(samples, 0, 0);
49e98daa 647 log_offset = (s->__ddir & LOG_OFFSET_SAMPLE_BIT) != 0;
b26317c9
JA
648
649 nr_samples = sample_size / __log_entry_sz(log_offset);
ac9b9101 650
aee2ab67 651 for (i = 0; i < nr_samples; i++) {
b26317c9 652 s = __get_sample(samples, log_offset, i);
ac9b9101 653
aee2ab67 654 if (!log_offset) {
ae588852
JA
655 fprintf(f, "%lu, %lu, %u, %u\n",
656 (unsigned long) s->time,
657 (unsigned long) s->val,
b26317c9 658 io_sample_ddir(s), s->bs);
ae588852
JA
659 } else {
660 struct io_sample_offset *so = (void *) s;
661
662 fprintf(f, "%lu, %lu, %u, %u, %llu\n",
663 (unsigned long) s->time,
664 (unsigned long) s->val,
b26317c9 665 io_sample_ddir(s), s->bs,
ae588852
JA
666 (unsigned long long) so->offset);
667 }
ac9b9101 668 }
aee2ab67
JA
669}
670
671#ifdef CONFIG_ZLIB
97eabb2a
JA
672
673struct iolog_flush_data {
155f2f02
JA
674 struct workqueue_work work;
675 pthread_mutex_t lock;
676 pthread_cond_t cv;
677 int wait;
678 volatile int done;
679 volatile int refs;
97eabb2a
JA
680 struct io_log *log;
681 void *samples;
682 uint64_t nr_samples;
683};
684
685struct iolog_compress {
686 struct flist_head list;
687 void *buf;
688 size_t len;
689 unsigned int seq;
97eabb2a
JA
690};
691
692#define GZ_CHUNK 131072
693
694static struct iolog_compress *get_new_chunk(unsigned int seq)
695{
696 struct iolog_compress *c;
697
698 c = malloc(sizeof(*c));
699 INIT_FLIST_HEAD(&c->list);
700 c->buf = malloc(GZ_CHUNK);
701 c->len = 0;
702 c->seq = seq;
97eabb2a
JA
703 return c;
704}
705
706static void free_chunk(struct iolog_compress *ic)
707{
c07826cd
JA
708 free(ic->buf);
709 free(ic);
97eabb2a
JA
710}
711
b26317c9 712static int z_stream_init(z_stream *stream, int gz_hdr)
aee2ab67 713{
b26317c9
JA
714 int wbits = 15;
715
aee2ab67
JA
716 stream->zalloc = Z_NULL;
717 stream->zfree = Z_NULL;
718 stream->opaque = Z_NULL;
719 stream->next_in = Z_NULL;
720
1a8e7458
JA
721 /*
722 * zlib magic - add 32 for auto-detection of gz header or not,
723 * if we decide to store files in a gzip friendly format.
724 */
b26317c9
JA
725 if (gz_hdr)
726 wbits += 32;
727
728 if (inflateInit2(stream, wbits) != Z_OK)
aee2ab67
JA
729 return 1;
730
731 return 0;
732}
733
1a8e7458 734struct inflate_chunk_iter {
aee2ab67 735 unsigned int seq;
dad95da1 736 int err;
aee2ab67
JA
737 void *buf;
738 size_t buf_size;
739 size_t buf_used;
740 size_t chunk_sz;
741};
742
b26317c9 743static void finish_chunk(z_stream *stream, FILE *f,
1a8e7458 744 struct inflate_chunk_iter *iter)
aee2ab67 745{
aee2ab67
JA
746 int ret;
747
748 ret = inflateEnd(stream);
749 if (ret != Z_OK)
750 log_err("fio: failed to end log inflation (%d)\n", ret);
751
b26317c9 752 flush_samples(f, iter->buf, iter->buf_used);
aee2ab67
JA
753 free(iter->buf);
754 iter->buf = NULL;
755 iter->buf_size = iter->buf_used = 0;
756}
757
1a8e7458
JA
758/*
759 * Iterative chunk inflation. Handles cases where we cross into a new
760 * sequence, doing flush finish of previous chunk if needed.
761 */
762static size_t inflate_chunk(struct iolog_compress *ic, int gz_hdr, FILE *f,
763 z_stream *stream, struct inflate_chunk_iter *iter)
aee2ab67 764{
0c56718d
JA
765 size_t ret;
766
767 dprint(FD_COMPRESS, "inflate chunk size=%lu, seq=%u",
768 (unsigned long) ic->len, ic->seq);
769
aee2ab67
JA
770 if (ic->seq != iter->seq) {
771 if (iter->seq)
b26317c9 772 finish_chunk(stream, f, iter);
aee2ab67 773
b26317c9 774 z_stream_init(stream, gz_hdr);
aee2ab67
JA
775 iter->seq = ic->seq;
776 }
777
778 stream->avail_in = ic->len;
779 stream->next_in = ic->buf;
780
781 if (!iter->buf_size) {
782 iter->buf_size = iter->chunk_sz;
783 iter->buf = malloc(iter->buf_size);
784 }
785
786 while (stream->avail_in) {
b26317c9 787 size_t this_out = iter->buf_size - iter->buf_used;
aee2ab67
JA
788 int err;
789
b26317c9 790 stream->avail_out = this_out;
aee2ab67
JA
791 stream->next_out = iter->buf + iter->buf_used;
792
793 err = inflate(stream, Z_NO_FLUSH);
794 if (err < 0) {
795 log_err("fio: failed inflating log: %d\n", err);
dad95da1 796 iter->err = err;
aee2ab67
JA
797 break;
798 }
799
b26317c9
JA
800 iter->buf_used += this_out - stream->avail_out;
801
802 if (!stream->avail_out) {
803 iter->buf_size += iter->chunk_sz;
804 iter->buf = realloc(iter->buf, iter->buf_size);
805 continue;
806 }
807
808 if (err == Z_STREAM_END)
809 break;
aee2ab67
JA
810 }
811
0c56718d
JA
812 ret = (void *) stream->next_in - ic->buf;
813
814 dprint(FD_COMPRESS, "inflated to size=%lu\n", (unsigned long) ret);
815
816 return ret;
aee2ab67
JA
817}
818
1a8e7458
JA
819/*
820 * Inflate stored compressed chunks, or write them directly to the log
821 * file if so instructed.
822 */
dad95da1 823static int inflate_gz_chunks(struct io_log *log, FILE *f)
aee2ab67 824{
1a8e7458 825 struct inflate_chunk_iter iter = { .chunk_sz = log->log_gz, };
aee2ab67
JA
826 z_stream stream;
827
828 while (!flist_empty(&log->chunk_list)) {
829 struct iolog_compress *ic;
830
9342d5f8 831 ic = flist_first_entry(&log->chunk_list, struct iolog_compress, list);
aee2ab67 832 flist_del(&ic->list);
b26317c9 833
1a8e7458
JA
834 if (log->log_gz_store) {
835 size_t ret;
836
0c56718d
JA
837 dprint(FD_COMPRESS, "log write chunk size=%lu, "
838 "seq=%u\n", (unsigned long) ic->len, ic->seq);
839
1a8e7458 840 ret = fwrite(ic->buf, ic->len, 1, f);
dad95da1
JA
841 if (ret != 1 || ferror(f)) {
842 iter.err = errno;
1a8e7458 843 log_err("fio: error writing compressed log\n");
dad95da1 844 }
1a8e7458
JA
845 } else
846 inflate_chunk(ic, log->log_gz_store, f, &stream, &iter);
c07826cd
JA
847
848 free_chunk(ic);
aee2ab67
JA
849 }
850
851 if (iter.seq) {
b26317c9 852 finish_chunk(&stream, f, &iter);
aee2ab67
JA
853 free(iter.buf);
854 }
dad95da1
JA
855
856 return iter.err;
aee2ab67
JA
857}
858
1a8e7458
JA
859/*
860 * Open compressed log file and decompress the stored chunks and
861 * write them to stdout. The chunks are stored sequentially in the
862 * file, so we iterate over them and do them one-by-one.
863 */
b26317c9
JA
864int iolog_file_inflate(const char *file)
865{
1a8e7458 866 struct inflate_chunk_iter iter = { .chunk_sz = 64 * 1024 * 1024, };
b26317c9
JA
867 struct iolog_compress ic;
868 z_stream stream;
869 struct stat sb;
fd771971 870 ssize_t ret;
f302710c
JA
871 size_t total;
872 void *buf;
b26317c9
JA
873 FILE *f;
874
b26317c9
JA
875 f = fopen(file, "r");
876 if (!f) {
877 perror("fopen");
878 return 1;
879 }
880
fd771971
JA
881 if (stat(file, &sb) < 0) {
882 fclose(f);
883 perror("stat");
884 return 1;
885 }
886
f302710c 887 ic.buf = buf = malloc(sb.st_size);
b26317c9 888 ic.len = sb.st_size;
b26317c9
JA
889 ic.seq = 1;
890
891 ret = fread(ic.buf, ic.len, 1, f);
892 if (ret < 0) {
893 perror("fread");
894 fclose(f);
bb1116fe 895 free(buf);
b26317c9
JA
896 return 1;
897 } else if (ret != 1) {
898 log_err("fio: short read on reading log\n");
899 fclose(f);
bb1116fe 900 free(buf);
b26317c9
JA
901 return 1;
902 }
903
904 fclose(f);
905
1a8e7458
JA
906 /*
907 * Each chunk will return Z_STREAM_END. We don't know how many
908 * chunks are in the file, so we just keep looping and incrementing
909 * the sequence number until we have consumed the whole compressed
910 * file.
911 */
f302710c
JA
912 total = ic.len;
913 do {
8a1db9a1 914 size_t iret;
f302710c 915
8a1db9a1
JA
916 iret = inflate_chunk(&ic, 1, stdout, &stream, &iter);
917 total -= iret;
f302710c
JA
918 if (!total)
919 break;
dad95da1
JA
920 if (iter.err)
921 break;
f302710c
JA
922
923 ic.seq++;
8a1db9a1
JA
924 ic.len -= iret;
925 ic.buf += iret;
f302710c 926 } while (1);
b26317c9
JA
927
928 if (iter.seq) {
929 finish_chunk(&stream, stdout, &iter);
930 free(iter.buf);
931 }
932
f302710c 933 free(buf);
dad95da1 934 return iter.err;
b26317c9
JA
935}
936
aee2ab67
JA
937#else
938
dad95da1 939static int inflate_gz_chunks(struct io_log *log, FILE *f)
aee2ab67 940{
b7364e21 941 return 0;
aee2ab67
JA
942}
943
a68c66b2
JA
944int iolog_file_inflate(const char *file)
945{
946 log_err("fio: log inflation not possible without zlib\n");
947 return 1;
948}
949
aee2ab67
JA
950#endif
951
3a5db920 952void flush_log(struct io_log *log, int do_append)
aee2ab67
JA
953{
954 void *buf;
955 FILE *f;
956
3a5db920
JA
957 if (!do_append)
958 f = fopen(log->filename, "w");
959 else
960 f = fopen(log->filename, "a");
aee2ab67
JA
961 if (!f) {
962 perror("fopen log");
963 return;
964 }
965
966 buf = set_file_buffer(f);
967
1a8e7458 968 inflate_gz_chunks(log, f);
aee2ab67 969
b26317c9 970 flush_samples(f, log->log, log->nr_samples * log_entry_sz(log));
ac9b9101
JA
971
972 fclose(f);
2e802282 973 clear_file_buffer(buf);
ac9b9101
JA
974}
975
cb7e0ace 976static int finish_log(struct thread_data *td, struct io_log *log, int trylock)
ac9b9101 977{
155f2f02 978 if (td->flags & TD_F_COMPRESS_LOG)
aee2ab67
JA
979 iolog_flush(log, 1);
980
243bfe19 981 if (trylock) {
cb7e0ace 982 if (fio_trylock_file(log->filename))
243bfe19
JA
983 return 1;
984 } else
cb7e0ace 985 fio_lock_file(log->filename);
243bfe19 986
aee2ab67 987 if (td->client_type == FIO_CLIENT_TYPE_GUI)
cb7e0ace 988 fio_send_iolog(td, log, log->filename);
aee2ab67 989 else
3a5db920 990 flush_log(log, !td->o.per_job_logs);
243bfe19 991
cb7e0ace 992 fio_unlock_file(log->filename);
518dac09 993 free_log(log);
243bfe19 994 return 0;
ac9b9101
JA
995}
996
aee2ab67
JA
997#ifdef CONFIG_ZLIB
998
1a8e7458
JA
999/*
1000 * Invoked from our compress helper thread, when logging would have exceeded
1001 * the specified memory limitation. Compresses the previously stored
1002 * entries.
1003 */
155f2f02 1004static int gz_work(struct submit_worker *sw, struct workqueue_work *work)
aee2ab67
JA
1005{
1006 struct iolog_flush_data *data;
1007 struct iolog_compress *c;
1008 struct flist_head list;
1009 unsigned int seq;
1010 z_stream stream;
1011 size_t total = 0;
d73ac887 1012 int ret;
aee2ab67
JA
1013
1014 INIT_FLIST_HEAD(&list);
1015
1016 data = container_of(work, struct iolog_flush_data, work);
1017
1018 stream.zalloc = Z_NULL;
1019 stream.zfree = Z_NULL;
1020 stream.opaque = Z_NULL;
1021
b26317c9 1022 ret = deflateInit(&stream, Z_DEFAULT_COMPRESSION);
b26317c9 1023 if (ret != Z_OK) {
aee2ab67
JA
1024 log_err("fio: failed to init gz stream\n");
1025 return 0;
1026 }
1027
1028 seq = ++data->log->chunk_seq;
b26317c9 1029
aee2ab67
JA
1030 stream.next_in = (void *) data->samples;
1031 stream.avail_in = data->nr_samples * log_entry_sz(data->log);
1032
0c56718d
JA
1033 dprint(FD_COMPRESS, "deflate input size=%lu, seq=%u\n",
1034 (unsigned long) stream.avail_in, seq);
aee2ab67
JA
1035 do {
1036 c = get_new_chunk(seq);
1037 stream.avail_out = GZ_CHUNK;
1038 stream.next_out = c->buf;
1039 ret = deflate(&stream, Z_NO_FLUSH);
1040 if (ret < 0) {
1041 log_err("fio: deflate log (%d)\n", ret);
dad95da1
JA
1042 free_chunk(c);
1043 goto err;
aee2ab67
JA
1044 }
1045
1046 c->len = GZ_CHUNK - stream.avail_out;
1047 flist_add_tail(&c->list, &list);
1048 total += c->len;
1049 } while (stream.avail_in);
1050
1051 stream.next_out = c->buf + c->len;
1052 stream.avail_out = GZ_CHUNK - c->len;
1053
1054 ret = deflate(&stream, Z_FINISH);
1055 if (ret == Z_STREAM_END)
1056 c->len = GZ_CHUNK - stream.avail_out;
1057 else {
1058 do {
1059 c = get_new_chunk(seq);
1060 stream.avail_out = GZ_CHUNK;
1061 stream.next_out = c->buf;
1062 ret = deflate(&stream, Z_FINISH);
1063 c->len = GZ_CHUNK - stream.avail_out;
0c56718d 1064 total += c->len;
aee2ab67
JA
1065 flist_add_tail(&c->list, &list);
1066 } while (ret != Z_STREAM_END);
1067 }
1068
0c56718d
JA
1069 dprint(FD_COMPRESS, "deflated to size=%lu\n", (unsigned long) total);
1070
aee2ab67
JA
1071 ret = deflateEnd(&stream);
1072 if (ret != Z_OK)
1073 log_err("fio: deflateEnd %d\n", ret);
1074
1075 free(data->samples);
1076
1077 if (!flist_empty(&list)) {
1078 pthread_mutex_lock(&data->log->chunk_lock);
1079 flist_splice_tail(&list, &data->log->chunk_list);
1080 pthread_mutex_unlock(&data->log->chunk_lock);
1081 }
1082
dad95da1
JA
1083 ret = 0;
1084done:
155f2f02
JA
1085 if (data->wait) {
1086 int refs;
1087
1088 pthread_mutex_lock(&data->lock);
1089 data->done = 1;
1090 pthread_cond_signal(&data->cv);
1091 refs = --data->refs;
1092 pthread_mutex_unlock(&data->lock);
1093 if (!refs)
1094 free(data);
aee2ab67
JA
1095 } else
1096 free(data);
dad95da1
JA
1097 return ret;
1098err:
1099 while (!flist_empty(&list)) {
1100 c = flist_first_entry(list.next, struct iolog_compress, list);
1101 flist_del(&c->list);
1102 free_chunk(c);
1103 }
1104 ret = 1;
1105 goto done;
aee2ab67
JA
1106}
1107
155f2f02
JA
1108static struct workqueue_ops log_compress_wq_ops = {
1109 .fn = gz_work,
ff84ff9a 1110 .nice = 1,
155f2f02
JA
1111};
1112
1113int iolog_compress_init(struct thread_data *td)
1114{
1115 if (!(td->flags & TD_F_COMPRESS_LOG))
1116 return 0;
1117
1118 workqueue_init(td, &td->log_compress_wq, &log_compress_wq_ops, 1);
1119 return 0;
1120}
1121
1122void iolog_compress_exit(struct thread_data *td)
1123{
1124 if (!(td->flags & TD_F_COMPRESS_LOG))
1125 return;
1126
1127 workqueue_exit(&td->log_compress_wq);
1128}
1129
1a8e7458
JA
1130/*
1131 * Queue work item to compress the existing log entries. We copy the
1132 * samples, and reset the log sample count to 0 (so the logging will
1133 * continue to use the memory associated with the log). If called with
1134 * wait == 1, will not return until the log compression has completed.
1135 */
aee2ab67
JA
1136int iolog_flush(struct io_log *log, int wait)
1137{
aee2ab67
JA
1138 struct iolog_flush_data *data;
1139 size_t sample_size;
1140
1141 data = malloc(sizeof(*data));
1142 if (!data)
1143 return 1;
1144
1145 data->log = log;
1146
1147 sample_size = log->nr_samples * log_entry_sz(log);
1148 data->samples = malloc(sample_size);
1149 if (!data->samples) {
1150 free(data);
1151 return 1;
1152 }
1153
1154 memcpy(data->samples, log->log, sample_size);
1155 data->nr_samples = log->nr_samples;
aee2ab67
JA
1156 log->nr_samples = 0;
1157
1158 if (wait) {
155f2f02
JA
1159 pthread_mutex_init(&data->lock, NULL);
1160 pthread_cond_init(&data->cv, NULL);
1161 data->done = 0;
1162 data->wait = 1;
1163 data->refs = 2;
aee2ab67 1164 } else
155f2f02 1165 data->wait = 0;
aee2ab67 1166
155f2f02 1167 workqueue_enqueue(&log->td->log_compress_wq, &data->work);
aee2ab67
JA
1168
1169 if (wait) {
155f2f02
JA
1170 int refs;
1171
1172 pthread_mutex_lock(&data->lock);
1173 while (!data->done)
1174 pthread_cond_wait(&data->cv, &data->lock);
1175 refs = --data->refs;
1176 pthread_mutex_unlock(&data->lock);
1177 if (!refs)
1178 free(data);
aee2ab67
JA
1179 }
1180
1181 return 0;
1182}
1183
1184#else
1185
1186int iolog_flush(struct io_log *log, int wait)
1187{
1188 return 1;
1189}
1190
1191#endif
1192
cb7e0ace 1193static int write_iops_log(struct thread_data *td, int try)
905e3d4f 1194{
cb7e0ace 1195 struct io_log *log = td->iops_log;
905e3d4f
JA
1196
1197 if (!log)
1198 return 0;
1199
cb7e0ace 1200 return finish_log(td, log, try);
905e3d4f
JA
1201}
1202
1203static int write_slat_log(struct thread_data *td, int try)
1204{
cb7e0ace 1205 struct io_log *log = td->slat_log;
905e3d4f 1206
cb7e0ace
JA
1207 if (!log)
1208 return 0;
1209
1210 return finish_log(td, log, try);
905e3d4f
JA
1211}
1212
1213static int write_clat_log(struct thread_data *td, int try)
1214{
cb7e0ace 1215 struct io_log *log = td->clat_log;
905e3d4f 1216
cb7e0ace
JA
1217 if (!log)
1218 return 0;
1219
1220 return finish_log(td, log, try);
905e3d4f
JA
1221}
1222
1223static int write_lat_log(struct thread_data *td, int try)
1224{
cb7e0ace
JA
1225 struct io_log *log = td->lat_log;
1226
1227 if (!log)
1228 return 0;
905e3d4f 1229
cb7e0ace 1230 return finish_log(td, log, try);
905e3d4f
JA
1231}
1232
1233static int write_bandw_log(struct thread_data *td, int try)
1234{
cb7e0ace
JA
1235 struct io_log *log = td->bw_log;
1236
1237 if (!log)
1238 return 0;
905e3d4f 1239
cb7e0ace 1240 return finish_log(td, log, try);
905e3d4f
JA
1241}
1242
1243enum {
1244 BW_LOG_MASK = 1,
1245 LAT_LOG_MASK = 2,
1246 SLAT_LOG_MASK = 4,
1247 CLAT_LOG_MASK = 8,
1248 IOPS_LOG_MASK = 16,
1249
905e3d4f
JA
1250 ALL_LOG_NR = 5,
1251};
1252
1253struct log_type {
1254 unsigned int mask;
1255 int (*fn)(struct thread_data *, int);
1256};
1257
1258static struct log_type log_types[] = {
1259 {
1260 .mask = BW_LOG_MASK,
1261 .fn = write_bandw_log,
1262 },
1263 {
1264 .mask = LAT_LOG_MASK,
1265 .fn = write_lat_log,
1266 },
1267 {
1268 .mask = SLAT_LOG_MASK,
1269 .fn = write_slat_log,
1270 },
1271 {
1272 .mask = CLAT_LOG_MASK,
1273 .fn = write_clat_log,
1274 },
1275 {
1276 .mask = IOPS_LOG_MASK,
1277 .fn = write_iops_log,
1278 },
1279};
1280
1281void fio_writeout_logs(struct thread_data *td)
1282{
ea5409f9 1283 unsigned int log_mask = 0;
905e3d4f
JA
1284 unsigned int log_left = ALL_LOG_NR;
1285 int old_state, i;
1286
1287 old_state = td_bump_runstate(td, TD_FINISHING);
1288
1289 finalize_logs(td);
1290
1291 while (log_left) {
1292 int prev_log_left = log_left;
1293
1294 for (i = 0; i < ALL_LOG_NR && log_left; i++) {
1295 struct log_type *lt = &log_types[i];
1296 int ret;
1297
ea5409f9 1298 if (!(log_mask & lt->mask)) {
905e3d4f
JA
1299 ret = lt->fn(td, log_left != 1);
1300 if (!ret) {
1301 log_left--;
ea5409f9 1302 log_mask |= lt->mask;
905e3d4f
JA
1303 }
1304 }
1305 }
1306
1307 if (prev_log_left == log_left)
1308 usleep(5000);
1309 }
1310
1311 td_restore_runstate(td, old_state);
1312}