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 static 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         if (td->client_type == FIO_CLIENT_TYPE_GUI)
 988                 fio_send_iolog(td, log, log->filename);
 989         else
 990                 flush_log(log, !td->o.per_job_logs);
 991
 992         fio_unlock_file(log->filename);
 993         free_log(log);
 994         return 0;
 995 }
 996
 997 #ifdef CONFIG_ZLIB
 998
 999 static void drop_data_unlock(struct iolog_flush_data *data)
1000 {
1001         int refs;
1002
1003         refs = --data->refs;
1004         pthread_mutex_unlock(&data->lock);
1005
1006         if (!refs) {
1007                 free(data);
1008                 pthread_mutex_destroy(&data->lock);
1009                 pthread_cond_destroy(&data->cv);
1010         }
1011 }
1012
1013 /*
1014  * Invoked from our compress helper thread, when logging would have exceeded
1015  * the specified memory limitation. Compresses the previously stored
1016  * entries.
1017  */
1018 static int gz_work(struct submit_worker *sw, struct workqueue_work *work)
1019 {
1020         struct iolog_flush_data *data;
1021         struct iolog_compress *c;
1022         struct flist_head list;
1023         unsigned int seq;
1024         z_stream stream;
1025         size_t total = 0;
1026         int ret;
1027
1028         INIT_FLIST_HEAD(&list);
1029
1030         data = container_of(work, struct iolog_flush_data, work);
1031
1032         stream.zalloc = Z_NULL;
1033         stream.zfree = Z_NULL;
1034         stream.opaque = Z_NULL;
1035
1036         ret = deflateInit(&stream, Z_DEFAULT_COMPRESSION);
1037         if (ret != Z_OK) {
1038                 log_err("fio: failed to init gz stream\n");
1039                 return 0;
1040         }
1041
1042         seq = ++data->log->chunk_seq;
1043
1044         stream.next_in = (void *) data->samples;
1045         stream.avail_in = data->nr_samples * log_entry_sz(data->log);
1046
1047         dprint(FD_COMPRESS, "deflate input size=%lu, seq=%u\n",
1048                                 (unsigned long) stream.avail_in, seq);
1049         do {
1050                 c = get_new_chunk(seq);
1051                 stream.avail_out = GZ_CHUNK;
1052                 stream.next_out = c->buf;
1053                 ret = deflate(&stream, Z_NO_FLUSH);
1054                 if (ret < 0) {
1055                         log_err("fio: deflate log (%d)\n", ret);
1056                         free_chunk(c);
1057                         goto err;
1058                 }
1059
1060                 c->len = GZ_CHUNK - stream.avail_out;
1061                 flist_add_tail(&c->list, &list);
1062                 total += c->len;
1063         } while (stream.avail_in);
1064
1065         stream.next_out = c->buf + c->len;
1066         stream.avail_out = GZ_CHUNK - c->len;
1067
1068         ret = deflate(&stream, Z_FINISH);
1069         if (ret == Z_STREAM_END)
1070                 c->len = GZ_CHUNK - stream.avail_out;
1071         else {
1072                 do {
1073                         c = get_new_chunk(seq);
1074                         stream.avail_out = GZ_CHUNK;
1075                         stream.next_out = c->buf;
1076                         ret = deflate(&stream, Z_FINISH);
1077                         c->len = GZ_CHUNK - stream.avail_out;
1078                         total += c->len;
1079                         flist_add_tail(&c->list, &list);
1080                 } while (ret != Z_STREAM_END);
1081         }
1082
1083         dprint(FD_COMPRESS, "deflated to size=%lu\n", (unsigned long) total);
1084
1085         ret = deflateEnd(&stream);
1086         if (ret != Z_OK)
1087                 log_err("fio: deflateEnd %d\n", ret);
1088
1089         free(data->samples);
1090
1091         if (!flist_empty(&list)) {
1092                 pthread_mutex_lock(&data->log->chunk_lock);
1093                 flist_splice_tail(&list, &data->log->chunk_list);
1094                 pthread_mutex_unlock(&data->log->chunk_lock);
1095         }
1096
1097         ret = 0;
1098 done:
1099         if (data->wait) {
1100                 pthread_mutex_lock(&data->lock);
1101                 data->done = 1;
1102                 pthread_cond_signal(&data->cv);
1103
1104                 drop_data_unlock(data);
1105         } else
1106                 free(data);
1107         return ret;
1108 err:
1109         while (!flist_empty(&list)) {
1110                 c = flist_first_entry(list.next, struct iolog_compress, list);
1111                 flist_del(&c->list);
1112                 free_chunk(c);
1113         }
1114         ret = 1;
1115         goto done;
1116 }
1117
1118 static int gz_init_worker(struct submit_worker *sw)
1119 {
1120         struct thread_data *td = sw->wq->td;
1121
1122         if (!fio_option_is_set(&td->o, log_gz_cpumask))
1123                 return 0;
1124
1125         if (fio_setaffinity(gettid(), td->o.log_gz_cpumask) == -1) {
1126                 log_err("gz: failed to set CPU affinity\n");
1127                 return 1;
1128         }
1129
1130         return 0;
1131 }
1132
1133 static struct workqueue_ops log_compress_wq_ops = {
1134         .fn             = gz_work,
1135         .init_worker_fn = gz_init_worker,
1136         .nice   = 1,
1137 };
1138
1139 int iolog_compress_init(struct thread_data *td)
1140 {
1141         if (!(td->flags & TD_F_COMPRESS_LOG))
1142                 return 0;
1143
1144         workqueue_init(td, &td->log_compress_wq, &log_compress_wq_ops, 1);
1145         return 0;
1146 }
1147
1148 void iolog_compress_exit(struct thread_data *td)
1149 {
1150         if (!(td->flags & TD_F_COMPRESS_LOG))
1151                 return;
1152
1153         workqueue_exit(&td->log_compress_wq);
1154 }
1155
1156 /*
1157  * Queue work item to compress the existing log entries. We copy the
1158  * samples, and reset the log sample count to 0 (so the logging will
1159  * continue to use the memory associated with the log). If called with
1160  * wait == 1, will not return until the log compression has completed.
1161  */
1162 int iolog_flush(struct io_log *log, int wait)
1163 {
1164         struct iolog_flush_data *data;
1165         size_t sample_size;
1166
1167         data = malloc(sizeof(*data));
1168         if (!data)
1169                 return 1;
1170
1171         data->log = log;
1172
1173         sample_size = log->nr_samples * log_entry_sz(log);
1174         data->samples = malloc(sample_size);
1175         if (!data->samples) {
1176                 free(data);
1177                 return 1;
1178         }
1179
1180         memcpy(data->samples, log->log, sample_size);
1181         data->nr_samples = log->nr_samples;
1182         log->nr_samples = 0;
1183
1184         data->wait = wait;
1185         if (data->wait) {
1186                 pthread_mutex_init(&data->lock, NULL);
1187                 pthread_cond_init(&data->cv, NULL);
1188                 data->done = 0;
1189                 data->refs = 2;
1190         }
1191
1192         workqueue_enqueue(&log->td->log_compress_wq, &data->work);
1193
1194         if (wait) {
1195                 pthread_mutex_lock(&data->lock);
1196                 while (!data->done)
1197                         pthread_cond_wait(&data->cv, &data->lock);
1198
1199                 drop_data_unlock(data);
1200         }
1201
1202         return 0;
1203 }
1204
1205 #else
1206
1207 int iolog_flush(struct io_log *log, int wait)
1208 {
1209         return 1;
1210 }
1211
1212 int iolog_compress_init(struct thread_data *td)
1213 {
1214         return 0;
1215 }
1216
1217 void iolog_compress_exit(struct thread_data *td)
1218 {
1219 }
1220
1221 #endif
1222
1223 static int __write_log(struct thread_data *td, struct io_log *log, int try)
1224 {
1225         if (log)
1226                 return finish_log(td, log, try);
1227
1228         return 0;
1229 }
1230
1231 static int write_iops_log(struct thread_data *td, int try)
1232 {
1233         return __write_log(td, td->iops_log, try);
1234 }
1235
1236 static int write_slat_log(struct thread_data *td, int try)
1237 {
1238         return __write_log(td, td->slat_log, try);
1239 }
1240
1241 static int write_clat_log(struct thread_data *td, int try)
1242 {
1243         return __write_log(td, td->clat_log, try);
1244 }
1245
1246 static int write_lat_log(struct thread_data *td, int try)
1247 {
1248         return __write_log(td, td->lat_log, try);
1249 }
1250
1251 static int write_bandw_log(struct thread_data *td, int try)
1252 {
1253         return __write_log(td, td->bw_log, try);
1254 }
1255
1256 enum {
1257         BW_LOG_MASK     = 1,
1258         LAT_LOG_MASK    = 2,
1259         SLAT_LOG_MASK   = 4,
1260         CLAT_LOG_MASK   = 8,
1261         IOPS_LOG_MASK   = 16,
1262
1263         ALL_LOG_NR      = 5,
1264 };
1265
1266 struct log_type {
1267         unsigned int mask;
1268         int (*fn)(struct thread_data *, int);
1269 };
1270
1271 static struct log_type log_types[] = {
1272         {
1273                 .mask   = BW_LOG_MASK,
1274                 .fn     = write_bandw_log,
1275         },
1276         {
1277                 .mask   = LAT_LOG_MASK,
1278                 .fn     = write_lat_log,
1279         },
1280         {
1281                 .mask   = SLAT_LOG_MASK,
1282                 .fn     = write_slat_log,
1283         },
1284         {
1285                 .mask   = CLAT_LOG_MASK,
1286                 .fn     = write_clat_log,
1287         },
1288         {
1289                 .mask   = IOPS_LOG_MASK,
1290                 .fn     = write_iops_log,
1291         },
1292 };
1293
1294 void fio_writeout_logs(struct thread_data *td)
1295 {
1296         unsigned int log_mask = 0;
1297         unsigned int log_left = ALL_LOG_NR;
1298         int old_state, i;
1299
1300         old_state = td_bump_runstate(td, TD_FINISHING);
1301
1302         finalize_logs(td);
1303
1304         while (log_left) {
1305                 int prev_log_left = log_left;
1306
1307                 for (i = 0; i < ALL_LOG_NR && log_left; i++) {
1308                         struct log_type *lt = &log_types[i];
1309                         int ret;
1310
1311                         if (!(log_mask & lt->mask)) {
1312                                 ret = lt->fn(td, log_left != 1);
1313                                 if (!ret) {
1314                                         log_left--;
1315                                         log_mask |= lt->mask;
1316                                 }
1317                         }
1318                 }
1319
1320                 if (prev_log_left == log_left)
1321                         usleep(5000);
1322         }
1323
1324         td_restore_runstate(td, old_state);
1325 }