iolog.c

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