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
  22 static const char iolog_ver2[] = "fio version 2 iolog";
  23
  24 void 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
  30 void log_io_u(const struct thread_data *td, const struct io_u *io_u)
  31 {
  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,
  36                                                 io_ddir_name(io_u->ddir),
  37                                                 io_u->offset, io_u->buflen);
  38 }
  39
  40 void 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
  60 static void iolog_delay(struct thread_data *td, unsigned long delay)
  61 {
  62         uint64_t usec = utime_since_now(&td->last_issue);
  63         uint64_t this_delay;
  64         struct timeval tv;
  65
  66         if (delay < td->time_offset) {
  67                 td->time_offset = 0;
  68                 return;
  69         }
  70
  71         delay -= td->time_offset;
  72         if (delay < usec)
  73                 return;
  74
  75         delay -= usec;
  76
  77         fio_gettime(&tv, NULL);
  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         }
  86
  87         usec = utime_since_now(&tv);
  88         if (usec > delay)
  89                 td->time_offset = usec - delay;
  90         else
  91                 td->time_offset = 0;
  92 }
  93
  94 static 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:
 118                 td_io_unlink_file(td, f);
 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
 128 int read_iolog_get(struct thread_data *td, struct io_u *io_u)
 129 {
 130         struct io_piece *ipo;
 131         unsigned long elapsed;
 132
 133         while (!flist_empty(&td->io_log_list)) {
 134                 int ret;
 135
 136                 ipo = flist_first_entry(&td->io_log_list, struct io_piece, list);
 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);
 163                 }
 164
 165                 free(ipo);
 166
 167                 if (io_u->ddir != DDIR_WAIT)
 168                         return 0;
 169         }
 170
 171         td->done = 1;
 172         return 1;
 173 }
 174
 175 void 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)) {
 189                 ipo = flist_first_entry(&td->io_hist_list, struct io_piece, list);
 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  */
 200 void 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;
 210         ipo->numberio = io_u->numberio;
 211         ipo->flags = IP_F_IN_FLIGHT;
 212
 213         io_u->ipo = ipo;
 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          */
 234         if (((!td->o.verifysort) || !td_random(td) || !td->o.overwrite) &&
 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          */
 248 restart:
 249         p = &td->io_hist_tree.rb_node;
 250         parent = NULL;
 251         while (*p) {
 252                 int overlap = 0;
 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;
 260                 else if (ipo->offset < __ipo->offset) {
 261                         p = &(*p)->rb_left;
 262                         overlap = ipo->offset + ipo->len > __ipo->offset;
 263                 }
 264                 else if (ipo->offset > __ipo->offset) {
 265                         p = &(*p)->rb_right;
 266                         overlap = __ipo->offset + __ipo->len > ipo->offset;
 267                 }
 268                 else
 269                         overlap = 1;
 270
 271                 if (overlap) {
 272                         dprint(FD_IO, "iolog: overlap %llu/%lu, %llu/%lu",
 273                                 __ipo->offset, __ipo->len,
 274                                 ipo->offset, ipo->len);
 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
 289 void unlog_io_piece(struct thread_data *td, struct io_u *io_u)
 290 {
 291         struct io_piece *ipo = io_u->ipo;
 292
 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
 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
 318 void trim_io_piece(struct thread_data *td, const struct io_u *io_u)
 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
 328 void 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  */
 341 static 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                         }
 388                         fileno = get_fileno(td, fname);
 389                 } else if (r == 2) {
 390                         rw = DDIR_INVAL;
 391                         if (!strcmp(act, "add")) {
 392                                 fileno = add_file(td, fname, 0, 1);
 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;
 439                         if (rw != DDIR_INVAL && bytes > td->o.max_bs[rw])
 440                                 td->o.max_bs[rw] = bytes;
 441                         ipo->fileno = fileno;
 442                         ipo->file_action = file_action;
 443                         td->o.size += bytes;
 444                 }
 445
 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  */
 474 static 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  */
 512 static 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
 548 int init_iolog(struct thread_data *td)
 549 {
 550         int ret = 0;
 551
 552         if (td->o.read_iolog_file) {
 553                 int need_swap;
 554
 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                  */
 559                 if (is_blktrace(td->o.read_iolog_file, &need_swap))
 560                         ret = load_blktrace(td, td->o.read_iolog_file, need_swap);
 561                 else
 562                         ret = init_iolog_read(td);
 563         } else if (td->o.write_iolog_file)
 564                 ret = init_iolog_write(td);
 565
 566         if (ret)
 567                 td_verror(td, EINVAL, "failed initializing iolog");
 568
 569         return ret;
 570 }
 571
 572 void setup_log(struct io_log **log, struct log_params *p,
 573                const char *filename)
 574 {
 575         struct io_log *l;
 576
 577         l = calloc(1, sizeof(*l));
 578         l->nr_samples = 0;
 579         l->max_samples = 1024;
 580         l->log_type = p->log_type;
 581         l->log_offset = p->log_offset;
 582         l->log_gz = p->log_gz;
 583         l->log_gz_store = p->log_gz_store;
 584         l->log = malloc(l->max_samples * log_entry_sz(l));
 585         l->avg_msec = p->avg_msec;
 586         l->filename = strdup(filename);
 587         l->td = p->td;
 588
 589         if (l->log_offset)
 590                 l->log_ddir_mask = LOG_OFFSET_SAMPLE_BIT;
 591
 592         INIT_FLIST_HEAD(&l->chunk_list);
 593
 594         if (l->log_gz && !p->td)
 595                 l->log_gz = 0;
 596         else if (l->log_gz || l->log_gz_store) {
 597                 pthread_mutex_init(&l->chunk_lock, NULL);
 598                 p->td->flags |= TD_F_COMPRESS_LOG;
 599         }
 600
 601         *log = l;
 602 }
 603
 604 #ifdef CONFIG_SETVBUF
 605 static 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
 615 static void clear_file_buffer(void *buf)
 616 {
 617         free(buf);
 618 }
 619 #else
 620 static void *set_file_buffer(FILE *f)
 621 {
 622         return NULL;
 623 }
 624
 625 static void clear_file_buffer(void *buf)
 626 {
 627 }
 628 #endif
 629
 630 void free_log(struct io_log *log)
 631 {
 632         free(log->log);
 633         free(log->filename);
 634         free(log);
 635 }
 636
 637 void flush_samples(FILE *f, void *samples, uint64_t sample_size)
 638 {
 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);
 647         log_offset = (s->__ddir & LOG_OFFSET_SAMPLE_BIT) != 0;
 648
 649         nr_samples = sample_size / __log_entry_sz(log_offset);
 650
 651         for (i = 0; i < nr_samples; i++) {
 652                 s = __get_sample(samples, log_offset, i);
 653
 654                 if (!log_offset) {
 655                         fprintf(f, "%lu, %lu, %u, %u\n",
 656                                         (unsigned long) s->time,
 657                                         (unsigned long) s->val,
 658                                         io_sample_ddir(s), s->bs);
 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,
 665                                         io_sample_ddir(s), s->bs,
 666                                         (unsigned long long) so->offset);
 667                 }
 668         }
 669 }
 670
 671 #ifdef CONFIG_ZLIB
 672
 673 struct iolog_flush_data {
 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;
 680         struct io_log *log;
 681         void *samples;
 682         uint64_t nr_samples;
 683 };
 684
 685 #define GZ_CHUNK        131072
 686
 687 static struct iolog_compress *get_new_chunk(unsigned int seq)
 688 {
 689         struct iolog_compress *c;
 690
 691         c = malloc(sizeof(*c));
 692         INIT_FLIST_HEAD(&c->list);
 693         c->buf = malloc(GZ_CHUNK);
 694         c->len = 0;
 695         c->seq = seq;
 696         return c;
 697 }
 698
 699 static void free_chunk(struct iolog_compress *ic)
 700 {
 701         free(ic->buf);
 702         free(ic);
 703 }
 704
 705 static int z_stream_init(z_stream *stream, int gz_hdr)
 706 {
 707         int wbits = 15;
 708
 709         stream->zalloc = Z_NULL;
 710         stream->zfree = Z_NULL;
 711         stream->opaque = Z_NULL;
 712         stream->next_in = Z_NULL;
 713
 714         /*
 715          * zlib magic - add 32 for auto-detection of gz header or not,
 716          * if we decide to store files in a gzip friendly format.
 717          */
 718         if (gz_hdr)
 719                 wbits += 32;
 720
 721         if (inflateInit2(stream, wbits) != Z_OK)
 722                 return 1;
 723
 724         return 0;
 725 }
 726
 727 struct inflate_chunk_iter {
 728         unsigned int seq;
 729         int err;
 730         void *buf;
 731         size_t buf_size;
 732         size_t buf_used;
 733         size_t chunk_sz;
 734 };
 735
 736 static void finish_chunk(z_stream *stream, FILE *f,
 737                          struct inflate_chunk_iter *iter)
 738 {
 739         int ret;
 740
 741         ret = inflateEnd(stream);
 742         if (ret != Z_OK)
 743                 log_err("fio: failed to end log inflation (%d)\n", ret);
 744
 745         flush_samples(f, iter->buf, iter->buf_used);
 746         free(iter->buf);
 747         iter->buf = NULL;
 748         iter->buf_size = iter->buf_used = 0;
 749 }
 750
 751 /*
 752  * Iterative chunk inflation. Handles cases where we cross into a new
 753  * sequence, doing flush finish of previous chunk if needed.
 754  */
 755 static size_t inflate_chunk(struct iolog_compress *ic, int gz_hdr, FILE *f,
 756                             z_stream *stream, struct inflate_chunk_iter *iter)
 757 {
 758         size_t ret;
 759
 760         dprint(FD_COMPRESS, "inflate chunk size=%lu, seq=%u",
 761                                 (unsigned long) ic->len, ic->seq);
 762
 763         if (ic->seq != iter->seq) {
 764                 if (iter->seq)
 765                         finish_chunk(stream, f, iter);
 766
 767                 z_stream_init(stream, gz_hdr);
 768                 iter->seq = ic->seq;
 769         }
 770
 771         stream->avail_in = ic->len;
 772         stream->next_in = ic->buf;
 773
 774         if (!iter->buf_size) {
 775                 iter->buf_size = iter->chunk_sz;
 776                 iter->buf = malloc(iter->buf_size);
 777         }
 778
 779         while (stream->avail_in) {
 780                 size_t this_out = iter->buf_size - iter->buf_used;
 781                 int err;
 782
 783                 stream->avail_out = this_out;
 784                 stream->next_out = iter->buf + iter->buf_used;
 785
 786                 err = inflate(stream, Z_NO_FLUSH);
 787                 if (err < 0) {
 788                         log_err("fio: failed inflating log: %d\n", err);
 789                         iter->err = err;
 790                         break;
 791                 }
 792
 793                 iter->buf_used += this_out - stream->avail_out;
 794
 795                 if (!stream->avail_out) {
 796                         iter->buf_size += iter->chunk_sz;
 797                         iter->buf = realloc(iter->buf, iter->buf_size);
 798                         continue;
 799                 }
 800
 801                 if (err == Z_STREAM_END)
 802                         break;
 803         }
 804
 805         ret = (void *) stream->next_in - ic->buf;
 806
 807         dprint(FD_COMPRESS, "inflated to size=%lu\n", (unsigned long) ret);
 808
 809         return ret;
 810 }
 811
 812 /*
 813  * Inflate stored compressed chunks, or write them directly to the log
 814  * file if so instructed.
 815  */
 816 static int inflate_gz_chunks(struct io_log *log, FILE *f)
 817 {
 818         struct inflate_chunk_iter iter = { .chunk_sz = log->log_gz, };
 819         z_stream stream;
 820
 821         while (!flist_empty(&log->chunk_list)) {
 822                 struct iolog_compress *ic;
 823
 824                 ic = flist_first_entry(&log->chunk_list, struct iolog_compress, list);
 825                 flist_del(&ic->list);
 826
 827                 if (log->log_gz_store) {
 828                         size_t ret;
 829
 830                         dprint(FD_COMPRESS, "log write chunk size=%lu, "
 831                                 "seq=%u\n", (unsigned long) ic->len, ic->seq);
 832
 833                         ret = fwrite(ic->buf, ic->len, 1, f);
 834                         if (ret != 1 || ferror(f)) {
 835                                 iter.err = errno;
 836                                 log_err("fio: error writing compressed log\n");
 837                         }
 838                 } else
 839                         inflate_chunk(ic, log->log_gz_store, f, &stream, &iter);
 840
 841                 free_chunk(ic);
 842         }
 843
 844         if (iter.seq) {
 845                 finish_chunk(&stream, f, &iter);
 846                 free(iter.buf);
 847         }
 848
 849         return iter.err;
 850 }
 851
 852 /*
 853  * Open compressed log file and decompress the stored chunks and
 854  * write them to stdout. The chunks are stored sequentially in the
 855  * file, so we iterate over them and do them one-by-one.
 856  */
 857 int iolog_file_inflate(const char *file)
 858 {
 859         struct inflate_chunk_iter iter = { .chunk_sz = 64 * 1024 * 1024, };
 860         struct iolog_compress ic;
 861         z_stream stream;
 862         struct stat sb;
 863         ssize_t ret;
 864         size_t total;
 865         void *buf;
 866         FILE *f;
 867
 868         f = fopen(file, "r");
 869         if (!f) {
 870                 perror("fopen");
 871                 return 1;
 872         }
 873
 874         if (stat(file, &sb) < 0) {
 875                 fclose(f);
 876                 perror("stat");
 877                 return 1;
 878         }
 879
 880         ic.buf = buf = malloc(sb.st_size);
 881         ic.len = sb.st_size;
 882         ic.seq = 1;
 883
 884         ret = fread(ic.buf, ic.len, 1, f);
 885         if (ret < 0) {
 886                 perror("fread");
 887                 fclose(f);
 888                 free(buf);
 889                 return 1;
 890         } else if (ret != 1) {
 891                 log_err("fio: short read on reading log\n");
 892                 fclose(f);
 893                 free(buf);
 894                 return 1;
 895         }
 896
 897         fclose(f);
 898
 899         /*
 900          * Each chunk will return Z_STREAM_END. We don't know how many
 901          * chunks are in the file, so we just keep looping and incrementing
 902          * the sequence number until we have consumed the whole compressed
 903          * file.
 904          */
 905         total = ic.len;
 906         do {
 907                 size_t iret;
 908
 909                 iret = inflate_chunk(&ic,  1, stdout, &stream, &iter);
 910                 total -= iret;
 911                 if (!total)
 912                         break;
 913                 if (iter.err)
 914                         break;
 915
 916                 ic.seq++;
 917                 ic.len -= iret;
 918                 ic.buf += iret;
 919         } while (1);
 920
 921         if (iter.seq) {
 922                 finish_chunk(&stream, stdout, &iter);
 923                 free(iter.buf);
 924         }
 925
 926         free(buf);
 927         return iter.err;
 928 }
 929
 930 #else
 931
 932 static int inflate_gz_chunks(struct io_log *log, FILE *f)
 933 {
 934         return 0;
 935 }
 936
 937 int iolog_file_inflate(const char *file)
 938 {
 939         log_err("fio: log inflation not possible without zlib\n");
 940         return 1;
 941 }
 942
 943 #endif
 944
 945 void flush_log(struct io_log *log, int do_append)
 946 {
 947         void *buf;
 948         FILE *f;
 949
 950         if (!do_append)
 951                 f = fopen(log->filename, "w");
 952         else
 953                 f = fopen(log->filename, "a");
 954         if (!f) {
 955                 perror("fopen log");
 956                 return;
 957         }
 958
 959         buf = set_file_buffer(f);
 960
 961         inflate_gz_chunks(log, f);
 962
 963         flush_samples(f, log->log, log->nr_samples * log_entry_sz(log));
 964
 965         fclose(f);
 966         clear_file_buffer(buf);
 967 }
 968
 969 static int finish_log(struct thread_data *td, struct io_log *log, int trylock)
 970 {
 971         if (td->flags & TD_F_COMPRESS_LOG)
 972                 iolog_flush(log, 1);
 973
 974         if (trylock) {
 975                 if (fio_trylock_file(log->filename))
 976                         return 1;
 977         } else
 978                 fio_lock_file(log->filename);
 979
 980         if (td->client_type == FIO_CLIENT_TYPE_GUI || is_backend)
 981                 fio_send_iolog(td, log, log->filename);
 982         else
 983                 flush_log(log, !td->o.per_job_logs);
 984
 985         fio_unlock_file(log->filename);
 986         free_log(log);
 987         return 0;
 988 }
 989
 990 size_t log_chunk_sizes(struct io_log *log)
 991 {
 992         struct flist_head *entry;
 993         size_t ret;
 994
 995         if (flist_empty(&log->chunk_list))
 996                 return 0;
 997
 998         ret = 0;
 999         pthread_mutex_lock(&log->chunk_lock);
1000         flist_for_each(entry, &log->chunk_list) {
1001                 struct iolog_compress *c;
1002
1003                 c = flist_entry(entry, struct iolog_compress, list);
1004                 ret += c->len;
1005         }
1006         pthread_mutex_unlock(&log->chunk_lock);
1007         return ret;
1008 }
1009
1010 #ifdef CONFIG_ZLIB
1011
1012 static void drop_data_unlock(struct iolog_flush_data *data)
1013 {
1014         int refs;
1015
1016         refs = --data->refs;
1017         pthread_mutex_unlock(&data->lock);
1018
1019         if (!refs) {
1020                 free(data);
1021                 pthread_mutex_destroy(&data->lock);
1022                 pthread_cond_destroy(&data->cv);
1023         }
1024 }
1025
1026 /*
1027  * Invoked from our compress helper thread, when logging would have exceeded
1028  * the specified memory limitation. Compresses the previously stored
1029  * entries.
1030  */
1031 static int gz_work(struct submit_worker *sw, struct workqueue_work *work)
1032 {
1033         struct iolog_flush_data *data;
1034         struct iolog_compress *c;
1035         struct flist_head list;
1036         unsigned int seq;
1037         z_stream stream;
1038         size_t total = 0;
1039         int ret;
1040
1041         INIT_FLIST_HEAD(&list);
1042
1043         data = container_of(work, struct iolog_flush_data, work);
1044
1045         stream.zalloc = Z_NULL;
1046         stream.zfree = Z_NULL;
1047         stream.opaque = Z_NULL;
1048
1049         ret = deflateInit(&stream, Z_DEFAULT_COMPRESSION);
1050         if (ret != Z_OK) {
1051                 log_err("fio: failed to init gz stream\n");
1052                 return 0;
1053         }
1054
1055         seq = ++data->log->chunk_seq;
1056
1057         stream.next_in = (void *) data->samples;
1058         stream.avail_in = data->nr_samples * log_entry_sz(data->log);
1059
1060         dprint(FD_COMPRESS, "deflate input size=%lu, seq=%u\n",
1061                                 (unsigned long) stream.avail_in, seq);
1062         do {
1063                 c = get_new_chunk(seq);
1064                 stream.avail_out = GZ_CHUNK;
1065                 stream.next_out = c->buf;
1066                 ret = deflate(&stream, Z_NO_FLUSH);
1067                 if (ret < 0) {
1068                         log_err("fio: deflate log (%d)\n", ret);
1069                         free_chunk(c);
1070                         goto err;
1071                 }
1072
1073                 c->len = GZ_CHUNK - stream.avail_out;
1074                 flist_add_tail(&c->list, &list);
1075                 total += c->len;
1076         } while (stream.avail_in);
1077
1078         stream.next_out = c->buf + c->len;
1079         stream.avail_out = GZ_CHUNK - c->len;
1080
1081         ret = deflate(&stream, Z_FINISH);
1082         if (ret == Z_STREAM_END)
1083                 c->len = GZ_CHUNK - stream.avail_out;
1084         else {
1085                 do {
1086                         c = get_new_chunk(seq);
1087                         stream.avail_out = GZ_CHUNK;
1088                         stream.next_out = c->buf;
1089                         ret = deflate(&stream, Z_FINISH);
1090                         c->len = GZ_CHUNK - stream.avail_out;
1091                         total += c->len;
1092                         flist_add_tail(&c->list, &list);
1093                 } while (ret != Z_STREAM_END);
1094         }
1095
1096         dprint(FD_COMPRESS, "deflated to size=%lu\n", (unsigned long) total);
1097
1098         ret = deflateEnd(&stream);
1099         if (ret != Z_OK)
1100                 log_err("fio: deflateEnd %d\n", ret);
1101
1102         free(data->samples);
1103
1104         if (!flist_empty(&list)) {
1105                 pthread_mutex_lock(&data->log->chunk_lock);
1106                 flist_splice_tail(&list, &data->log->chunk_list);
1107                 pthread_mutex_unlock(&data->log->chunk_lock);
1108         }
1109
1110         ret = 0;
1111 done:
1112         if (data->wait) {
1113                 pthread_mutex_lock(&data->lock);
1114                 data->done = 1;
1115                 pthread_cond_signal(&data->cv);
1116
1117                 drop_data_unlock(data);
1118         } else
1119                 free(data);
1120         return ret;
1121 err:
1122         while (!flist_empty(&list)) {
1123                 c = flist_first_entry(list.next, struct iolog_compress, list);
1124                 flist_del(&c->list);
1125                 free_chunk(c);
1126         }
1127         ret = 1;
1128         goto done;
1129 }
1130
1131 static int gz_init_worker(struct submit_worker *sw)
1132 {
1133         struct thread_data *td = sw->wq->td;
1134
1135         if (!fio_option_is_set(&td->o, log_gz_cpumask))
1136                 return 0;
1137
1138         if (fio_setaffinity(gettid(), td->o.log_gz_cpumask) == -1) {
1139                 log_err("gz: failed to set CPU affinity\n");
1140                 return 1;
1141         }
1142
1143         return 0;
1144 }
1145
1146 static struct workqueue_ops log_compress_wq_ops = {
1147         .fn             = gz_work,
1148         .init_worker_fn = gz_init_worker,
1149         .nice           = 1,
1150 };
1151
1152 int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out)
1153 {
1154         if (!(td->flags & TD_F_COMPRESS_LOG))
1155                 return 0;
1156
1157         workqueue_init(td, &td->log_compress_wq, &log_compress_wq_ops, 1, sk_out);
1158         return 0;
1159 }
1160
1161 void iolog_compress_exit(struct thread_data *td)
1162 {
1163         if (!(td->flags & TD_F_COMPRESS_LOG))
1164                 return;
1165
1166         workqueue_exit(&td->log_compress_wq);
1167 }
1168
1169 /*
1170  * Queue work item to compress the existing log entries. We reset the
1171  * current log to a small size, and reference the existing log in the
1172  * data that we queue for compression. Once compression has been done,
1173  * this old log is freed. If called with wait == 1, will not return until
1174  * the log compression has completed.
1175  */
1176 int iolog_flush(struct io_log *log, int wait)
1177 {
1178         struct iolog_flush_data *data;
1179
1180         io_u_quiesce(log->td);
1181
1182         data = malloc(sizeof(*data));
1183         if (!data)
1184                 return 1;
1185
1186         data->log = log;
1187
1188         data->samples = log->log;
1189         data->nr_samples = log->nr_samples;
1190
1191         log->nr_samples = 0;
1192         log->max_samples = 128;
1193         log->log = malloc(log->max_samples * log_entry_sz(log));
1194
1195         data->wait = wait;
1196         if (data->wait) {
1197                 pthread_mutex_init(&data->lock, NULL);
1198                 pthread_cond_init(&data->cv, NULL);
1199                 data->done = 0;
1200                 data->refs = 2;
1201         }
1202
1203         workqueue_enqueue(&log->td->log_compress_wq, &data->work);
1204
1205         if (wait) {
1206                 pthread_mutex_lock(&data->lock);
1207                 while (!data->done)
1208                         pthread_cond_wait(&data->cv, &data->lock);
1209
1210                 drop_data_unlock(data);
1211         }
1212
1213         return 0;
1214 }
1215
1216 #else
1217
1218 int iolog_flush(struct io_log *log, int wait)
1219 {
1220         return 1;
1221 }
1222
1223 int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out)
1224 {
1225         return 0;
1226 }
1227
1228 void iolog_compress_exit(struct thread_data *td)
1229 {
1230 }
1231
1232 #endif
1233
1234 static int __write_log(struct thread_data *td, struct io_log *log, int try)
1235 {
1236         if (log)
1237                 return finish_log(td, log, try);
1238
1239         return 0;
1240 }
1241
1242 static int write_iops_log(struct thread_data *td, int try)
1243 {
1244         return __write_log(td, td->iops_log, try);
1245 }
1246
1247 static int write_slat_log(struct thread_data *td, int try)
1248 {
1249         return __write_log(td, td->slat_log, try);
1250 }
1251
1252 static int write_clat_log(struct thread_data *td, int try)
1253 {
1254         return __write_log(td, td->clat_log, try);
1255 }
1256
1257 static int write_lat_log(struct thread_data *td, int try)
1258 {
1259         return __write_log(td, td->lat_log, try);
1260 }
1261
1262 static int write_bandw_log(struct thread_data *td, int try)
1263 {
1264         return __write_log(td, td->bw_log, try);
1265 }
1266
1267 enum {
1268         BW_LOG_MASK     = 1,
1269         LAT_LOG_MASK    = 2,
1270         SLAT_LOG_MASK   = 4,
1271         CLAT_LOG_MASK   = 8,
1272         IOPS_LOG_MASK   = 16,
1273
1274         ALL_LOG_NR      = 5,
1275 };
1276
1277 struct log_type {
1278         unsigned int mask;
1279         int (*fn)(struct thread_data *, int);
1280 };
1281
1282 static struct log_type log_types[] = {
1283         {
1284                 .mask   = BW_LOG_MASK,
1285                 .fn     = write_bandw_log,
1286         },
1287         {
1288                 .mask   = LAT_LOG_MASK,
1289                 .fn     = write_lat_log,
1290         },
1291         {
1292                 .mask   = SLAT_LOG_MASK,
1293                 .fn     = write_slat_log,
1294         },
1295         {
1296                 .mask   = CLAT_LOG_MASK,
1297                 .fn     = write_clat_log,
1298         },
1299         {
1300                 .mask   = IOPS_LOG_MASK,
1301                 .fn     = write_iops_log,
1302         },
1303 };
1304
1305 void fio_writeout_logs(struct thread_data *td)
1306 {
1307         unsigned int log_mask = 0;
1308         unsigned int log_left = ALL_LOG_NR;
1309         int old_state, i;
1310
1311         old_state = td_bump_runstate(td, TD_FINISHING);
1312
1313         finalize_logs(td);
1314
1315         while (log_left) {
1316                 int prev_log_left = log_left;
1317
1318                 for (i = 0; i < ALL_LOG_NR && log_left; i++) {
1319                         struct log_type *lt = &log_types[i];
1320                         int ret;
1321
1322                         if (!(log_mask & lt->mask)) {
1323                                 ret = lt->fn(td, log_left != 1);
1324                                 if (!ret) {
1325                                         log_left--;
1326                                         log_mask |= lt->mask;
1327                                 }
1328                         }
1329                 }
1330
1331                 if (prev_log_left == log_left)
1332                         usleep(5000);
1333         }
1334
1335         td_restore_runstate(td, old_state);
1336 }