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 struct iolog_compress {
 686         struct flist_head list;
 687         void *buf;
 688         size_t len;
 689         unsigned int seq;
 690 };
 691
 692 #define GZ_CHUNK        131072
 693
 694 static struct iolog_compress *get_new_chunk(unsigned int seq)
 695 {
 696         struct iolog_compress *c;
 697
 698         c = malloc(sizeof(*c));
 699         INIT_FLIST_HEAD(&c->list);
 700         c->buf = malloc(GZ_CHUNK);
 701         c->len = 0;
 702         c->seq = seq;
 703         return c;
 704 }
 705
 706 static void free_chunk(struct iolog_compress *ic)
 707 {
 708         free(ic->buf);
 709         free(ic);
 710 }
 711
 712 static int z_stream_init(z_stream *stream, int gz_hdr)
 713 {
 714         int wbits = 15;
 715
 716         stream->zalloc = Z_NULL;
 717         stream->zfree = Z_NULL;
 718         stream->opaque = Z_NULL;
 719         stream->next_in = Z_NULL;
 720
 721         /*
 722          * zlib magic - add 32 for auto-detection of gz header or not,
 723          * if we decide to store files in a gzip friendly format.
 724          */
 725         if (gz_hdr)
 726                 wbits += 32;
 727
 728         if (inflateInit2(stream, wbits) != Z_OK)
 729                 return 1;
 730
 731         return 0;
 732 }
 733
 734 struct inflate_chunk_iter {
 735         unsigned int seq;
 736         int err;
 737         void *buf;
 738         size_t buf_size;
 739         size_t buf_used;
 740         size_t chunk_sz;
 741 };
 742
 743 static void finish_chunk(z_stream *stream, FILE *f,
 744                          struct inflate_chunk_iter *iter)
 745 {
 746         int ret;
 747
 748         ret = inflateEnd(stream);
 749         if (ret != Z_OK)
 750                 log_err("fio: failed to end log inflation (%d)\n", ret);
 751
 752         flush_samples(f, iter->buf, iter->buf_used);
 753         free(iter->buf);
 754         iter->buf = NULL;
 755         iter->buf_size = iter->buf_used = 0;
 756 }
 757
 758 /*
 759  * Iterative chunk inflation. Handles cases where we cross into a new
 760  * sequence, doing flush finish of previous chunk if needed.
 761  */
 762 static size_t inflate_chunk(struct iolog_compress *ic, int gz_hdr, FILE *f,
 763                             z_stream *stream, struct inflate_chunk_iter *iter)
 764 {
 765         size_t ret;
 766
 767         dprint(FD_COMPRESS, "inflate chunk size=%lu, seq=%u",
 768                                 (unsigned long) ic->len, ic->seq);
 769
 770         if (ic->seq != iter->seq) {
 771                 if (iter->seq)
 772                         finish_chunk(stream, f, iter);
 773
 774                 z_stream_init(stream, gz_hdr);
 775                 iter->seq = ic->seq;
 776         }
 777
 778         stream->avail_in = ic->len;
 779         stream->next_in = ic->buf;
 780
 781         if (!iter->buf_size) {
 782                 iter->buf_size = iter->chunk_sz;
 783                 iter->buf = malloc(iter->buf_size);
 784         }
 785
 786         while (stream->avail_in) {
 787                 size_t this_out = iter->buf_size - iter->buf_used;
 788                 int err;
 789
 790                 stream->avail_out = this_out;
 791                 stream->next_out = iter->buf + iter->buf_used;
 792
 793                 err = inflate(stream, Z_NO_FLUSH);
 794                 if (err < 0) {
 795                         log_err("fio: failed inflating log: %d\n", err);
 796                         iter->err = err;
 797                         break;
 798                 }
 799
 800                 iter->buf_used += this_out - stream->avail_out;
 801
 802                 if (!stream->avail_out) {
 803                         iter->buf_size += iter->chunk_sz;
 804                         iter->buf = realloc(iter->buf, iter->buf_size);
 805                         continue;
 806                 }
 807
 808                 if (err == Z_STREAM_END)
 809                         break;
 810         }
 811
 812         ret = (void *) stream->next_in - ic->buf;
 813
 814         dprint(FD_COMPRESS, "inflated to size=%lu\n", (unsigned long) ret);
 815
 816         return ret;
 817 }
 818
 819 /*
 820  * Inflate stored compressed chunks, or write them directly to the log
 821  * file if so instructed.
 822  */
 823 static int inflate_gz_chunks(struct io_log *log, FILE *f)
 824 {
 825         struct inflate_chunk_iter iter = { .chunk_sz = log->log_gz, };
 826         z_stream stream;
 827
 828         while (!flist_empty(&log->chunk_list)) {
 829                 struct iolog_compress *ic;
 830
 831                 ic = flist_first_entry(&log->chunk_list, struct iolog_compress, list);
 832                 flist_del(&ic->list);
 833
 834                 if (log->log_gz_store) {
 835                         size_t ret;
 836
 837                         dprint(FD_COMPRESS, "log write chunk size=%lu, "
 838                                 "seq=%u\n", (unsigned long) ic->len, ic->seq);
 839
 840                         ret = fwrite(ic->buf, ic->len, 1, f);
 841                         if (ret != 1 || ferror(f)) {
 842                                 iter.err = errno;
 843                                 log_err("fio: error writing compressed log\n");
 844                         }
 845                 } else
 846                         inflate_chunk(ic, log->log_gz_store, f, &stream, &iter);
 847
 848                 free_chunk(ic);
 849         }
 850
 851         if (iter.seq) {
 852                 finish_chunk(&stream, f, &iter);
 853                 free(iter.buf);
 854         }
 855
 856         return iter.err;
 857 }
 858
 859 /*
 860  * Open compressed log file and decompress the stored chunks and
 861  * write them to stdout. The chunks are stored sequentially in the
 862  * file, so we iterate over them and do them one-by-one.
 863  */
 864 int iolog_file_inflate(const char *file)
 865 {
 866         struct inflate_chunk_iter iter = { .chunk_sz = 64 * 1024 * 1024, };
 867         struct iolog_compress ic;
 868         z_stream stream;
 869         struct stat sb;
 870         ssize_t ret;
 871         size_t total;
 872         void *buf;
 873         FILE *f;
 874
 875         f = fopen(file, "r");
 876         if (!f) {
 877                 perror("fopen");
 878                 return 1;
 879         }
 880
 881         if (stat(file, &sb) < 0) {
 882                 fclose(f);
 883                 perror("stat");
 884                 return 1;
 885         }
 886
 887         ic.buf = buf = malloc(sb.st_size);
 888         ic.len = sb.st_size;
 889         ic.seq = 1;
 890
 891         ret = fread(ic.buf, ic.len, 1, f);
 892         if (ret < 0) {
 893                 perror("fread");
 894                 fclose(f);
 895                 free(buf);
 896                 return 1;
 897         } else if (ret != 1) {
 898                 log_err("fio: short read on reading log\n");
 899                 fclose(f);
 900                 free(buf);
 901                 return 1;
 902         }
 903
 904         fclose(f);
 905
 906         /*
 907          * Each chunk will return Z_STREAM_END. We don't know how many
 908          * chunks are in the file, so we just keep looping and incrementing
 909          * the sequence number until we have consumed the whole compressed
 910          * file.
 911          */
 912         total = ic.len;
 913         do {
 914                 size_t iret;
 915
 916                 iret = inflate_chunk(&ic,  1, stdout, &stream, &iter);
 917                 total -= iret;
 918                 if (!total)
 919                         break;
 920                 if (iter.err)
 921                         break;
 922
 923                 ic.seq++;
 924                 ic.len -= iret;
 925                 ic.buf += iret;
 926         } while (1);
 927
 928         if (iter.seq) {
 929                 finish_chunk(&stream, stdout, &iter);
 930                 free(iter.buf);
 931         }
 932
 933         free(buf);
 934         return iter.err;
 935 }
 936
 937 #else
 938
 939 static int inflate_gz_chunks(struct io_log *log, FILE *f)
 940 {
 941         return 0;
 942 }
 943
 944 int iolog_file_inflate(const char *file)
 945 {
 946         log_err("fio: log inflation not possible without zlib\n");
 947         return 1;
 948 }
 949
 950 #endif
 951
 952 void flush_log(struct io_log *log, int do_append)
 953 {
 954         void *buf;
 955         FILE *f;
 956
 957         if (!do_append)
 958                 f = fopen(log->filename, "w");
 959         else
 960                 f = fopen(log->filename, "a");
 961         if (!f) {
 962                 perror("fopen log");
 963                 return;
 964         }
 965
 966         buf = set_file_buffer(f);
 967
 968         inflate_gz_chunks(log, f);
 969
 970         flush_samples(f, log->log, log->nr_samples * log_entry_sz(log));
 971
 972         fclose(f);
 973         clear_file_buffer(buf);
 974 }
 975
 976 static int finish_log(struct thread_data *td, struct io_log *log, int trylock)
 977 {
 978         if (td->flags & TD_F_COMPRESS_LOG)
 979                 iolog_flush(log, 1);
 980
 981         if (trylock) {
 982                 if (fio_trylock_file(log->filename))
 983                         return 1;
 984         } else
 985                 fio_lock_file(log->filename);
 986
 987         /*
 988          * We should do this for any networked client. Will enable when
 989          * the kinks are ironed out.
 990          *
 991          * if (td->client_type == FIO_CLIENT_TYPE_GUI || is_backed)
 992          */
 993         if (td->client_type == FIO_CLIENT_TYPE_GUI)
 994                 fio_send_iolog(td, log, log->filename);
 995         else
 996                 flush_log(log, !td->o.per_job_logs);
 997
 998         fio_unlock_file(log->filename);
 999         free_log(log);
1000         return 0;
1001 }
1002
1003 #ifdef CONFIG_ZLIB
1004
1005 static void drop_data_unlock(struct iolog_flush_data *data)
1006 {
1007         int refs;
1008
1009         refs = --data->refs;
1010         pthread_mutex_unlock(&data->lock);
1011
1012         if (!refs) {
1013                 free(data);
1014                 pthread_mutex_destroy(&data->lock);
1015                 pthread_cond_destroy(&data->cv);
1016         }
1017 }
1018
1019 /*
1020  * Invoked from our compress helper thread, when logging would have exceeded
1021  * the specified memory limitation. Compresses the previously stored
1022  * entries.
1023  */
1024 static int gz_work(struct submit_worker *sw, struct workqueue_work *work)
1025 {
1026         struct iolog_flush_data *data;
1027         struct iolog_compress *c;
1028         struct flist_head list;
1029         unsigned int seq;
1030         z_stream stream;
1031         size_t total = 0;
1032         int ret;
1033
1034         INIT_FLIST_HEAD(&list);
1035
1036         data = container_of(work, struct iolog_flush_data, work);
1037
1038         stream.zalloc = Z_NULL;
1039         stream.zfree = Z_NULL;
1040         stream.opaque = Z_NULL;
1041
1042         ret = deflateInit(&stream, Z_DEFAULT_COMPRESSION);
1043         if (ret != Z_OK) {
1044                 log_err("fio: failed to init gz stream\n");
1045                 return 0;
1046         }
1047
1048         seq = ++data->log->chunk_seq;
1049
1050         stream.next_in = (void *) data->samples;
1051         stream.avail_in = data->nr_samples * log_entry_sz(data->log);
1052
1053         dprint(FD_COMPRESS, "deflate input size=%lu, seq=%u\n",
1054                                 (unsigned long) stream.avail_in, seq);
1055         do {
1056                 c = get_new_chunk(seq);
1057                 stream.avail_out = GZ_CHUNK;
1058                 stream.next_out = c->buf;
1059                 ret = deflate(&stream, Z_NO_FLUSH);
1060                 if (ret < 0) {
1061                         log_err("fio: deflate log (%d)\n", ret);
1062                         free_chunk(c);
1063                         goto err;
1064                 }
1065
1066                 c->len = GZ_CHUNK - stream.avail_out;
1067                 flist_add_tail(&c->list, &list);
1068                 total += c->len;
1069         } while (stream.avail_in);
1070
1071         stream.next_out = c->buf + c->len;
1072         stream.avail_out = GZ_CHUNK - c->len;
1073
1074         ret = deflate(&stream, Z_FINISH);
1075         if (ret == Z_STREAM_END)
1076                 c->len = GZ_CHUNK - stream.avail_out;
1077         else {
1078                 do {
1079                         c = get_new_chunk(seq);
1080                         stream.avail_out = GZ_CHUNK;
1081                         stream.next_out = c->buf;
1082                         ret = deflate(&stream, Z_FINISH);
1083                         c->len = GZ_CHUNK - stream.avail_out;
1084                         total += c->len;
1085                         flist_add_tail(&c->list, &list);
1086                 } while (ret != Z_STREAM_END);
1087         }
1088
1089         dprint(FD_COMPRESS, "deflated to size=%lu\n", (unsigned long) total);
1090
1091         ret = deflateEnd(&stream);
1092         if (ret != Z_OK)
1093                 log_err("fio: deflateEnd %d\n", ret);
1094
1095         free(data->samples);
1096
1097         if (!flist_empty(&list)) {
1098                 pthread_mutex_lock(&data->log->chunk_lock);
1099                 flist_splice_tail(&list, &data->log->chunk_list);
1100                 pthread_mutex_unlock(&data->log->chunk_lock);
1101         }
1102
1103         ret = 0;
1104 done:
1105         if (data->wait) {
1106                 pthread_mutex_lock(&data->lock);
1107                 data->done = 1;
1108                 pthread_cond_signal(&data->cv);
1109
1110                 drop_data_unlock(data);
1111         } else
1112                 free(data);
1113         return ret;
1114 err:
1115         while (!flist_empty(&list)) {
1116                 c = flist_first_entry(list.next, struct iolog_compress, list);
1117                 flist_del(&c->list);
1118                 free_chunk(c);
1119         }
1120         ret = 1;
1121         goto done;
1122 }
1123
1124 static int gz_init_worker(struct submit_worker *sw)
1125 {
1126         struct thread_data *td = sw->wq->td;
1127
1128         if (!fio_option_is_set(&td->o, log_gz_cpumask))
1129                 return 0;
1130
1131         if (fio_setaffinity(gettid(), td->o.log_gz_cpumask) == -1) {
1132                 log_err("gz: failed to set CPU affinity\n");
1133                 return 1;
1134         }
1135
1136         return 0;
1137 }
1138
1139 static struct workqueue_ops log_compress_wq_ops = {
1140         .fn             = gz_work,
1141         .init_worker_fn = gz_init_worker,
1142         .nice           = 1,
1143 };
1144
1145 int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out)
1146 {
1147         if (!(td->flags & TD_F_COMPRESS_LOG))
1148                 return 0;
1149
1150         workqueue_init(td, &td->log_compress_wq, &log_compress_wq_ops, 1, sk_out);
1151         return 0;
1152 }
1153
1154 void iolog_compress_exit(struct thread_data *td)
1155 {
1156         if (!(td->flags & TD_F_COMPRESS_LOG))
1157                 return;
1158
1159         workqueue_exit(&td->log_compress_wq);
1160 }
1161
1162 /*
1163  * Queue work item to compress the existing log entries. We reset the
1164  * current log to a small size, and reference the existing log in the
1165  * data that we queue for compression. Once compression has been done,
1166  * this old log is freed. If called with wait == 1, will not return until
1167  * the log compression has completed.
1168  */
1169 int iolog_flush(struct io_log *log, int wait)
1170 {
1171         struct iolog_flush_data *data;
1172
1173         io_u_quiesce(log->td);
1174
1175         data = malloc(sizeof(*data));
1176         if (!data)
1177                 return 1;
1178
1179         data->log = log;
1180
1181         data->samples = log->log;
1182         data->nr_samples = log->nr_samples;
1183
1184         log->nr_samples = 0;
1185         log->max_samples = 128;
1186         log->log = malloc(log->max_samples * log_entry_sz(log));
1187
1188         data->wait = wait;
1189         if (data->wait) {
1190                 pthread_mutex_init(&data->lock, NULL);
1191                 pthread_cond_init(&data->cv, NULL);
1192                 data->done = 0;
1193                 data->refs = 2;
1194         }
1195
1196         workqueue_enqueue(&log->td->log_compress_wq, &data->work);
1197
1198         if (wait) {
1199                 pthread_mutex_lock(&data->lock);
1200                 while (!data->done)
1201                         pthread_cond_wait(&data->cv, &data->lock);
1202
1203                 drop_data_unlock(data);
1204         }
1205
1206         return 0;
1207 }
1208
1209 #else
1210
1211 int iolog_flush(struct io_log *log, int wait)
1212 {
1213         return 1;
1214 }
1215
1216 int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out)
1217 {
1218         return 0;
1219 }
1220
1221 void iolog_compress_exit(struct thread_data *td)
1222 {
1223 }
1224
1225 #endif
1226
1227 static int __write_log(struct thread_data *td, struct io_log *log, int try)
1228 {
1229         if (log)
1230                 return finish_log(td, log, try);
1231
1232         return 0;
1233 }
1234
1235 static int write_iops_log(struct thread_data *td, int try)
1236 {
1237         return __write_log(td, td->iops_log, try);
1238 }
1239
1240 static int write_slat_log(struct thread_data *td, int try)
1241 {
1242         return __write_log(td, td->slat_log, try);
1243 }
1244
1245 static int write_clat_log(struct thread_data *td, int try)
1246 {
1247         return __write_log(td, td->clat_log, try);
1248 }
1249
1250 static int write_lat_log(struct thread_data *td, int try)
1251 {
1252         return __write_log(td, td->lat_log, try);
1253 }
1254
1255 static int write_bandw_log(struct thread_data *td, int try)
1256 {
1257         return __write_log(td, td->bw_log, try);
1258 }
1259
1260 enum {
1261         BW_LOG_MASK     = 1,
1262         LAT_LOG_MASK    = 2,
1263         SLAT_LOG_MASK   = 4,
1264         CLAT_LOG_MASK   = 8,
1265         IOPS_LOG_MASK   = 16,
1266
1267         ALL_LOG_NR      = 5,
1268 };
1269
1270 struct log_type {
1271         unsigned int mask;
1272         int (*fn)(struct thread_data *, int);
1273 };
1274
1275 static struct log_type log_types[] = {
1276         {
1277                 .mask   = BW_LOG_MASK,
1278                 .fn     = write_bandw_log,
1279         },
1280         {
1281                 .mask   = LAT_LOG_MASK,
1282                 .fn     = write_lat_log,
1283         },
1284         {
1285                 .mask   = SLAT_LOG_MASK,
1286                 .fn     = write_slat_log,
1287         },
1288         {
1289                 .mask   = CLAT_LOG_MASK,
1290                 .fn     = write_clat_log,
1291         },
1292         {
1293                 .mask   = IOPS_LOG_MASK,
1294                 .fn     = write_iops_log,
1295         },
1296 };
1297
1298 void fio_writeout_logs(struct thread_data *td)
1299 {
1300         unsigned int log_mask = 0;
1301         unsigned int log_left = ALL_LOG_NR;
1302         int old_state, i;
1303
1304         old_state = td_bump_runstate(td, TD_FINISHING);
1305
1306         finalize_logs(td);
1307
1308         while (log_left) {
1309                 int prev_log_left = log_left;
1310
1311                 for (i = 0; i < ALL_LOG_NR && log_left; i++) {
1312                         struct log_type *lt = &log_types[i];
1313                         int ret;
1314
1315                         if (!(log_mask & lt->mask)) {
1316                                 ret = lt->fn(td, log_left != 1);
1317                                 if (!ret) {
1318                                         log_left--;
1319                                         log_mask |= lt->mask;
1320                                 }
1321                         }
1322                 }
1323
1324                 if (prev_log_left == log_left)
1325                         usleep(5000);
1326         }
1327
1328         td_restore_runstate(td, old_state);
1329 }