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