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