1 /* 2 * Code related to writing an iolog of what a thread is doing, and to 3 * later read that back and replay 4 */ 5 #include <stdio.h> 6 #include <stdlib.h> 7 #include <libgen.h> 8 #include <assert.h> 9 #include <sys/types.h> 10 #include <sys/stat.h> 11 #include <unistd.h> 12 #ifdef CONFIG_ZLIB 13 #include <zlib.h> 14 #endif 15 16 #include "flist.h" 17 #include "fio.h" 18 #include "verify.h" 19 #include "trim.h" 20 #include "filelock.h" 21 #include "smalloc.h" 22 23 static int iolog_flush(struct io_log *log); 24 25 static const char iolog_ver2[] = "fio version 2 iolog"; 26 27 void queue_io_piece(struct thread_data *td, struct io_piece *ipo) 28 { 29 flist_add_tail(&ipo->list, &td->io_log_list); 30 td->total_io_size += ipo->len; 31 } 32 33 void log_io_u(const struct thread_data *td, const struct io_u *io_u) 34 { 35 if (!td->o.write_iolog_file) 36 return; 37 38 fprintf(td->iolog_f, "%s %s %llu %lu\n", io_u->file->file_name, 39 io_ddir_name(io_u->ddir), 40 io_u->offset, io_u->buflen); 41 } 42 43 void log_file(struct thread_data *td, struct fio_file *f, 44 enum file_log_act what) 45 { 46 const char *act[] = { "add", "open", "close" }; 47 48 assert(what < 3); 49 50 if (!td->o.write_iolog_file) 51 return; 52 53 54 /* 55 * this happens on the pre-open/close done before the job starts 56 */ 57 if (!td->iolog_f) 58 return; 59 60 fprintf(td->iolog_f, "%s %s\n", f->file_name, act[what]); 61 } 62 63 static void iolog_delay(struct thread_data *td, unsigned long delay) 64 { 65 uint64_t usec = utime_since_now(&td->last_issue); 66 uint64_t this_delay; 67 struct timeval tv; 68 69 if (delay < td->time_offset) { 70 td->time_offset = 0; 71 return; 72 } 73 74 delay -= td->time_offset; 75 if (delay < usec) 76 return; 77 78 delay -= usec; 79 80 fio_gettime(&tv, NULL); 81 while (delay && !td->terminate) { 82 this_delay = delay; 83 if (this_delay > 500000) 84 this_delay = 500000; 85 86 usec_sleep(td, this_delay); 87 delay -= this_delay; 88 } 89 90 usec = utime_since_now(&tv); 91 if (usec > delay) 92 td->time_offset = usec - delay; 93 else 94 td->time_offset = 0; 95 } 96 97 static int ipo_special(struct thread_data *td, struct io_piece *ipo) 98 { 99 struct fio_file *f; 100 int ret; 101 102 /* 103 * Not a special ipo 104 */ 105 if (ipo->ddir != DDIR_INVAL) 106 return 0; 107 108 f = td->files[ipo->fileno]; 109 110 switch (ipo->file_action) { 111 case FIO_LOG_OPEN_FILE: 112 if (td->o.replay_redirect && fio_file_open(f)) { 113 dprint(FD_FILE, "iolog: ignoring re-open of file %s\n", 114 f->file_name); 115 break; 116 } 117 ret = td_io_open_file(td, f); 118 if (!ret) 119 break; 120 td_verror(td, ret, "iolog open file"); 121 return -1; 122 case FIO_LOG_CLOSE_FILE: 123 td_io_close_file(td, f); 124 break; 125 case FIO_LOG_UNLINK_FILE: 126 td_io_unlink_file(td, f); 127 break; 128 default: 129 log_err("fio: bad file action %d\n", ipo->file_action); 130 break; 131 } 132 133 return 1; 134 } 135 136 int read_iolog_get(struct thread_data *td, struct io_u *io_u) 137 { 138 struct io_piece *ipo; 139 unsigned long elapsed; 140 141 while (!flist_empty(&td->io_log_list)) { 142 int ret; 143 144 ipo = flist_first_entry(&td->io_log_list, struct io_piece, list); 145 flist_del(&ipo->list); 146 remove_trim_entry(td, ipo); 147 148 ret = ipo_special(td, ipo); 149 if (ret < 0) { 150 free(ipo); 151 break; 152 } else if (ret > 0) { 153 free(ipo); 154 continue; 155 } 156 157 io_u->ddir = ipo->ddir; 158 if (ipo->ddir != DDIR_WAIT) { 159 io_u->offset = ipo->offset; 160 io_u->buflen = ipo->len; 161 io_u->file = td->files[ipo->fileno]; 162 get_file(io_u->file); 163 dprint(FD_IO, "iolog: get %llu/%lu/%s\n", io_u->offset, 164 io_u->buflen, io_u->file->file_name); 165 if (ipo->delay) 166 iolog_delay(td, ipo->delay); 167 } else { 168 elapsed = mtime_since_genesis(); 169 if (ipo->delay > elapsed) 170 usec_sleep(td, (ipo->delay - elapsed) * 1000); 171 } 172 173 free(ipo); 174 175 if (io_u->ddir != DDIR_WAIT) 176 return 0; 177 } 178 179 td->done = 1; 180 return 1; 181 } 182 183 void prune_io_piece_log(struct thread_data *td) 184 { 185 struct io_piece *ipo; 186 struct rb_node *n; 187 188 while ((n = rb_first(&td->io_hist_tree)) != NULL) { 189 ipo = rb_entry(n, struct io_piece, rb_node); 190 rb_erase(n, &td->io_hist_tree); 191 remove_trim_entry(td, ipo); 192 td->io_hist_len--; 193 free(ipo); 194 } 195 196 while (!flist_empty(&td->io_hist_list)) { 197 ipo = flist_first_entry(&td->io_hist_list, struct io_piece, list); 198 flist_del(&ipo->list); 199 remove_trim_entry(td, ipo); 200 td->io_hist_len--; 201 free(ipo); 202 } 203 } 204 205 /* 206 * log a successful write, so we can unwind the log for verify 207 */ 208 void log_io_piece(struct thread_data *td, struct io_u *io_u) 209 { 210 struct rb_node **p, *parent; 211 struct io_piece *ipo, *__ipo; 212 213 ipo = malloc(sizeof(struct io_piece)); 214 init_ipo(ipo); 215 ipo->file = io_u->file; 216 ipo->offset = io_u->offset; 217 ipo->len = io_u->buflen; 218 ipo->numberio = io_u->numberio; 219 ipo->flags = IP_F_IN_FLIGHT; 220 221 io_u->ipo = ipo; 222 223 if (io_u_should_trim(td, io_u)) { 224 flist_add_tail(&ipo->trim_list, &td->trim_list); 225 td->trim_entries++; 226 } 227 228 /* 229 * We don't need to sort the entries, if: 230 * 231 * Sequential writes, or 232 * Random writes that lay out the file as it goes along 233 * 234 * For both these cases, just reading back data in the order we 235 * wrote it out is the fastest. 236 * 237 * One exception is if we don't have a random map AND we are doing 238 * verifies, in that case we need to check for duplicate blocks and 239 * drop the old one, which we rely on the rb insert/lookup for 240 * handling. 241 */ 242 if (((!td->o.verifysort) || !td_random(td) || !td->o.overwrite) && 243 (file_randommap(td, ipo->file) || td->o.verify == VERIFY_NONE)) { 244 INIT_FLIST_HEAD(&ipo->list); 245 flist_add_tail(&ipo->list, &td->io_hist_list); 246 ipo->flags |= IP_F_ONLIST; 247 td->io_hist_len++; 248 return; 249 } 250 251 RB_CLEAR_NODE(&ipo->rb_node); 252 253 /* 254 * Sort the entry into the verification list 255 */ 256 restart: 257 p = &td->io_hist_tree.rb_node; 258 parent = NULL; 259 while (*p) { 260 int overlap = 0; 261 parent = *p; 262 263 __ipo = rb_entry(parent, struct io_piece, rb_node); 264 if (ipo->file < __ipo->file) 265 p = &(*p)->rb_left; 266 else if (ipo->file > __ipo->file) 267 p = &(*p)->rb_right; 268 else if (ipo->offset < __ipo->offset) { 269 p = &(*p)->rb_left; 270 overlap = ipo->offset + ipo->len > __ipo->offset; 271 } 272 else if (ipo->offset > __ipo->offset) { 273 p = &(*p)->rb_right; 274 overlap = __ipo->offset + __ipo->len > ipo->offset; 275 } 276 else 277 overlap = 1; 278 279 if (overlap) { 280 dprint(FD_IO, "iolog: overlap %llu/%lu, %llu/%lu\n", 281 __ipo->offset, __ipo->len, 282 ipo->offset, ipo->len); 283 td->io_hist_len--; 284 rb_erase(parent, &td->io_hist_tree); 285 remove_trim_entry(td, __ipo); 286 free(__ipo); 287 goto restart; 288 } 289 } 290 291 rb_link_node(&ipo->rb_node, parent, p); 292 rb_insert_color(&ipo->rb_node, &td->io_hist_tree); 293 ipo->flags |= IP_F_ONRB; 294 td->io_hist_len++; 295 } 296 297 void unlog_io_piece(struct thread_data *td, struct io_u *io_u) 298 { 299 struct io_piece *ipo = io_u->ipo; 300 301 if (td->ts.nr_block_infos) { 302 uint32_t *info = io_u_block_info(td, io_u); 303 if (BLOCK_INFO_STATE(*info) < BLOCK_STATE_TRIM_FAILURE) { 304 if (io_u->ddir == DDIR_TRIM) 305 *info = BLOCK_INFO_SET_STATE(*info, 306 BLOCK_STATE_TRIM_FAILURE); 307 else if (io_u->ddir == DDIR_WRITE) 308 *info = BLOCK_INFO_SET_STATE(*info, 309 BLOCK_STATE_WRITE_FAILURE); 310 } 311 } 312 313 if (!ipo) 314 return; 315 316 if (ipo->flags & IP_F_ONRB) 317 rb_erase(&ipo->rb_node, &td->io_hist_tree); 318 else if (ipo->flags & IP_F_ONLIST) 319 flist_del(&ipo->list); 320 321 free(ipo); 322 io_u->ipo = NULL; 323 td->io_hist_len--; 324 } 325 326 void trim_io_piece(struct thread_data *td, const struct io_u *io_u) 327 { 328 struct io_piece *ipo = io_u->ipo; 329 330 if (!ipo) 331 return; 332 333 ipo->len = io_u->xfer_buflen - io_u->resid; 334 } 335 336 void write_iolog_close(struct thread_data *td) 337 { 338 fflush(td->iolog_f); 339 fclose(td->iolog_f); 340 free(td->iolog_buf); 341 td->iolog_f = NULL; 342 td->iolog_buf = NULL; 343 } 344 345 /* 346 * Read version 2 iolog data. It is enhanced to include per-file logging, 347 * syncs, etc. 348 */ 349 static int read_iolog2(struct thread_data *td, FILE *f) 350 { 351 unsigned long long offset; 352 unsigned int bytes; 353 int reads, writes, waits, fileno = 0, file_action = 0; /* stupid gcc */ 354 char *rfname, *fname, *act; 355 char *str, *p; 356 enum fio_ddir rw; 357 358 free_release_files(td); 359 360 /* 361 * Read in the read iolog and store it, reuse the infrastructure 362 * for doing verifications. 363 */ 364 str = malloc(4096); 365 rfname = fname = malloc(256+16); 366 act = malloc(256+16); 367 368 reads = writes = waits = 0; 369 while ((p = fgets(str, 4096, f)) != NULL) { 370 struct io_piece *ipo; 371 int r; 372 373 r = sscanf(p, "%256s %256s %llu %u", rfname, act, &offset, 374 &bytes); 375 376 if (td->o.replay_redirect) 377 fname = td->o.replay_redirect; 378 379 if (r == 4) { 380 /* 381 * Check action first 382 */ 383 if (!strcmp(act, "wait")) 384 rw = DDIR_WAIT; 385 else if (!strcmp(act, "read")) 386 rw = DDIR_READ; 387 else if (!strcmp(act, "write")) 388 rw = DDIR_WRITE; 389 else if (!strcmp(act, "sync")) 390 rw = DDIR_SYNC; 391 else if (!strcmp(act, "datasync")) 392 rw = DDIR_DATASYNC; 393 else if (!strcmp(act, "trim")) 394 rw = DDIR_TRIM; 395 else { 396 log_err("fio: bad iolog file action: %s\n", 397 act); 398 continue; 399 } 400 fileno = get_fileno(td, fname); 401 } else if (r == 2) { 402 rw = DDIR_INVAL; 403 if (!strcmp(act, "add")) { 404 if (td->o.replay_redirect && 405 get_fileno(td, fname) != -1) { 406 dprint(FD_FILE, "iolog: ignoring" 407 " re-add of file %s\n", fname); 408 } else { 409 fileno = add_file(td, fname, 0, 1); 410 file_action = FIO_LOG_ADD_FILE; 411 } 412 continue; 413 } else if (!strcmp(act, "open")) { 414 fileno = get_fileno(td, fname); 415 file_action = FIO_LOG_OPEN_FILE; 416 } else if (!strcmp(act, "close")) { 417 fileno = get_fileno(td, fname); 418 file_action = FIO_LOG_CLOSE_FILE; 419 } else { 420 log_err("fio: bad iolog file action: %s\n", 421 act); 422 continue; 423 } 424 } else { 425 log_err("bad iolog2: %s\n", p); 426 continue; 427 } 428 429 if (rw == DDIR_READ) 430 reads++; 431 else if (rw == DDIR_WRITE) { 432 /* 433 * Don't add a write for ro mode 434 */ 435 if (read_only) 436 continue; 437 writes++; 438 } else if (rw == DDIR_WAIT) { 439 if (td->o.no_stall) 440 continue; 441 waits++; 442 } else if (rw == DDIR_INVAL) { 443 } else if (!ddir_sync(rw)) { 444 log_err("bad ddir: %d\n", rw); 445 continue; 446 } 447 448 /* 449 * Make note of file 450 */ 451 ipo = malloc(sizeof(*ipo)); 452 init_ipo(ipo); 453 ipo->ddir = rw; 454 if (rw == DDIR_WAIT) { 455 ipo->delay = offset; 456 } else { 457 if (td->o.replay_scale) 458 ipo->offset = offset / td->o.replay_scale; 459 else 460 ipo->offset = offset; 461 ipo_bytes_align(td->o.replay_align, ipo); 462 463 ipo->len = bytes; 464 if (rw != DDIR_INVAL && bytes > td->o.max_bs[rw]) 465 td->o.max_bs[rw] = bytes; 466 ipo->fileno = fileno; 467 ipo->file_action = file_action; 468 td->o.size += bytes; 469 } 470 471 queue_io_piece(td, ipo); 472 } 473 474 free(str); 475 free(act); 476 free(rfname); 477 478 if (writes && read_only) { 479 log_err("fio: <%s> skips replay of %d writes due to" 480 " read-only\n", td->o.name, writes); 481 writes = 0; 482 } 483 484 if (!reads && !writes && !waits) 485 return 1; 486 else if (reads && !writes) 487 td->o.td_ddir = TD_DDIR_READ; 488 else if (!reads && writes) 489 td->o.td_ddir = TD_DDIR_WRITE; 490 else 491 td->o.td_ddir = TD_DDIR_RW; 492 493 return 0; 494 } 495 496 /* 497 * open iolog, check version, and call appropriate parser 498 */ 499 static int init_iolog_read(struct thread_data *td) 500 { 501 char buffer[256], *p; 502 FILE *f; 503 int ret; 504 505 f = fopen(td->o.read_iolog_file, "r"); 506 if (!f) { 507 perror("fopen read iolog"); 508 return 1; 509 } 510 511 p = fgets(buffer, sizeof(buffer), f); 512 if (!p) { 513 td_verror(td, errno, "iolog read"); 514 log_err("fio: unable to read iolog\n"); 515 fclose(f); 516 return 1; 517 } 518 519 /* 520 * version 2 of the iolog stores a specific string as the 521 * first line, check for that 522 */ 523 if (!strncmp(iolog_ver2, buffer, strlen(iolog_ver2))) 524 ret = read_iolog2(td, f); 525 else { 526 log_err("fio: iolog version 1 is no longer supported\n"); 527 ret = 1; 528 } 529 530 fclose(f); 531 return ret; 532 } 533 534 /* 535 * Set up a log for storing io patterns. 536 */ 537 static int init_iolog_write(struct thread_data *td) 538 { 539 struct fio_file *ff; 540 FILE *f; 541 unsigned int i; 542 543 f = fopen(td->o.write_iolog_file, "a"); 544 if (!f) { 545 perror("fopen write iolog"); 546 return 1; 547 } 548 549 /* 550 * That's it for writing, setup a log buffer and we're done. 551 */ 552 td->iolog_f = f; 553 td->iolog_buf = malloc(8192); 554 setvbuf(f, td->iolog_buf, _IOFBF, 8192); 555 556 /* 557 * write our version line 558 */ 559 if (fprintf(f, "%s\n", iolog_ver2) < 0) { 560 perror("iolog init\n"); 561 return 1; 562 } 563 564 /* 565 * add all known files 566 */ 567 for_each_file(td, ff, i) 568 log_file(td, ff, FIO_LOG_ADD_FILE); 569 570 return 0; 571 } 572 573 int init_iolog(struct thread_data *td) 574 { 575 int ret = 0; 576 577 if (td->o.read_iolog_file) { 578 int need_swap; 579 580 /* 581 * Check if it's a blktrace file and load that if possible. 582 * Otherwise assume it's a normal log file and load that. 583 */ 584 if (is_blktrace(td->o.read_iolog_file, &need_swap)) 585 ret = load_blktrace(td, td->o.read_iolog_file, need_swap); 586 else 587 ret = init_iolog_read(td); 588 } else if (td->o.write_iolog_file) 589 ret = init_iolog_write(td); 590 591 if (ret) 592 td_verror(td, EINVAL, "failed initializing iolog"); 593 594 return ret; 595 } 596 597 void setup_log(struct io_log **log, struct log_params *p, 598 const char *filename) 599 { 600 struct io_log *l; 601 int i; 602 struct io_u_plat_entry *entry; 603 struct flist_head *list; 604 605 l = scalloc(1, sizeof(*l)); 606 INIT_FLIST_HEAD(&l->io_logs); 607 l->log_type = p->log_type; 608 l->log_offset = p->log_offset; 609 l->log_gz = p->log_gz; 610 l->log_gz_store = p->log_gz_store; 611 l->avg_msec = p->avg_msec; 612 l->hist_msec = p->hist_msec; 613 l->hist_coarseness = p->hist_coarseness; 614 l->filename = strdup(filename); 615 l->td = p->td; 616 617 /* Initialize histogram lists for each r/w direction, 618 * with initial io_u_plat of all zeros: 619 */ 620 for (i = 0; i < DDIR_RWDIR_CNT; i++) { 621 list = &l->hist_window[i].list; 622 INIT_FLIST_HEAD(list); 623 entry = calloc(1, sizeof(struct io_u_plat_entry)); 624 flist_add(&entry->list, list); 625 } 626 627 if (l->td && l->td->o.io_submit_mode != IO_MODE_OFFLOAD) { 628 struct io_logs *p; 629 630 p = calloc(1, sizeof(*l->pending)); 631 p->max_samples = DEF_LOG_ENTRIES; 632 p->log = calloc(p->max_samples, log_entry_sz(l)); 633 l->pending = p; 634 } 635 636 if (l->log_offset) 637 l->log_ddir_mask = LOG_OFFSET_SAMPLE_BIT; 638 639 INIT_FLIST_HEAD(&l->chunk_list); 640 641 if (l->log_gz && !p->td) 642 l->log_gz = 0; 643 else if (l->log_gz || l->log_gz_store) { 644 mutex_init_pshared(&l->chunk_lock); 645 p->td->flags |= TD_F_COMPRESS_LOG; 646 } 647 648 *log = l; 649 } 650 651 #ifdef CONFIG_SETVBUF 652 static void *set_file_buffer(FILE *f) 653 { 654 size_t size = 1048576; 655 void *buf; 656 657 buf = malloc(size); 658 setvbuf(f, buf, _IOFBF, size); 659 return buf; 660 } 661 662 static void clear_file_buffer(void *buf) 663 { 664 free(buf); 665 } 666 #else 667 static void *set_file_buffer(FILE *f) 668 { 669 return NULL; 670 } 671 672 static void clear_file_buffer(void *buf) 673 { 674 } 675 #endif 676 677 void free_log(struct io_log *log) 678 { 679 while (!flist_empty(&log->io_logs)) { 680 struct io_logs *cur_log; 681 682 cur_log = flist_first_entry(&log->io_logs, struct io_logs, list); 683 flist_del_init(&cur_log->list); 684 free(cur_log->log); 685 sfree(cur_log); 686 } 687 688 if (log->pending) { 689 free(log->pending->log); 690 free(log->pending); 691 log->pending = NULL; 692 } 693 694 free(log->pending); 695 free(log->filename); 696 sfree(log); 697 } 698 699 unsigned long hist_sum(int j, int stride, unsigned int *io_u_plat, 700 unsigned int *io_u_plat_last) 701 { 702 unsigned long sum; 703 int k; 704 705 if (io_u_plat_last) { 706 for (k = sum = 0; k < stride; k++) 707 sum += io_u_plat[j + k] - io_u_plat_last[j + k]; 708 } else { 709 for (k = sum = 0; k < stride; k++) 710 sum += io_u_plat[j + k]; 711 } 712 713 return sum; 714 } 715 716 static void flush_hist_samples(FILE *f, int hist_coarseness, void *samples, 717 uint64_t sample_size) 718 { 719 struct io_sample *s; 720 int log_offset; 721 uint64_t i, j, nr_samples; 722 struct io_u_plat_entry *entry, *entry_before; 723 unsigned int *io_u_plat; 724 unsigned int *io_u_plat_before; 725 726 int stride = 1 << hist_coarseness; 727 728 if (!sample_size) 729 return; 730 731 s = __get_sample(samples, 0, 0); 732 log_offset = (s->__ddir & LOG_OFFSET_SAMPLE_BIT) != 0; 733 734 nr_samples = sample_size / __log_entry_sz(log_offset); 735 736 for (i = 0; i < nr_samples; i++) { 737 s = __get_sample(samples, log_offset, i); 738 739 entry = s->data.plat_entry; 740 io_u_plat = entry->io_u_plat; 741 742 entry_before = flist_first_entry(&entry->list, struct io_u_plat_entry, list); 743 io_u_plat_before = entry_before->io_u_plat; 744 745 fprintf(f, "%lu, %u, %u, ", (unsigned long) s->time, 746 io_sample_ddir(s), s->bs); 747 for (j = 0; j < FIO_IO_U_PLAT_NR - stride; j += stride) { 748 fprintf(f, "%lu, ", hist_sum(j, stride, io_u_plat, 749 io_u_plat_before)); 750 } 751 fprintf(f, "%lu\n", (unsigned long) 752 hist_sum(FIO_IO_U_PLAT_NR - stride, stride, io_u_plat, 753 io_u_plat_before)); 754 755 flist_del(&entry_before->list); 756 free(entry_before); 757 } 758 } 759 760 void flush_samples(FILE *f, void *samples, uint64_t sample_size) 761 { 762 struct io_sample *s; 763 int log_offset; 764 uint64_t i, nr_samples; 765 766 if (!sample_size) 767 return; 768 769 s = __get_sample(samples, 0, 0); 770 log_offset = (s->__ddir & LOG_OFFSET_SAMPLE_BIT) != 0; 771 772 nr_samples = sample_size / __log_entry_sz(log_offset); 773 774 for (i = 0; i < nr_samples; i++) { 775 s = __get_sample(samples, log_offset, i); 776 777 if (!log_offset) { 778 fprintf(f, "%lu, %" PRId64 ", %u, %u\n", 779 (unsigned long) s->time, 780 s->data.val, 781 io_sample_ddir(s), s->bs); 782 } else { 783 struct io_sample_offset *so = (void *) s; 784 785 fprintf(f, "%lu, %" PRId64 ", %u, %u, %llu\n", 786 (unsigned long) s->time, 787 s->data.val, 788 io_sample_ddir(s), s->bs, 789 (unsigned long long) so->offset); 790 } 791 } 792 } 793 794 #ifdef CONFIG_ZLIB 795 796 struct iolog_flush_data { 797 struct workqueue_work work; 798 struct io_log *log; 799 void *samples; 800 uint32_t nr_samples; 801 bool free; 802 }; 803 804 #define GZ_CHUNK 131072 805 806 static struct iolog_compress *get_new_chunk(unsigned int seq) 807 { 808 struct iolog_compress *c; 809 810 c = malloc(sizeof(*c)); 811 INIT_FLIST_HEAD(&c->list); 812 c->buf = malloc(GZ_CHUNK); 813 c->len = 0; 814 c->seq = seq; 815 return c; 816 } 817 818 static void free_chunk(struct iolog_compress *ic) 819 { 820 free(ic->buf); 821 free(ic); 822 } 823 824 static int z_stream_init(z_stream *stream, int gz_hdr) 825 { 826 int wbits = 15; 827 828 memset(stream, 0, sizeof(*stream)); 829 stream->zalloc = Z_NULL; 830 stream->zfree = Z_NULL; 831 stream->opaque = Z_NULL; 832 stream->next_in = Z_NULL; 833 834 /* 835 * zlib magic - add 32 for auto-detection of gz header or not, 836 * if we decide to store files in a gzip friendly format. 837 */ 838 if (gz_hdr) 839 wbits += 32; 840 841 if (inflateInit2(stream, wbits) != Z_OK) 842 return 1; 843 844 return 0; 845 } 846 847 struct inflate_chunk_iter { 848 unsigned int seq; 849 int err; 850 void *buf; 851 size_t buf_size; 852 size_t buf_used; 853 size_t chunk_sz; 854 }; 855 856 static void finish_chunk(z_stream *stream, FILE *f, 857 struct inflate_chunk_iter *iter) 858 { 859 int ret; 860 861 ret = inflateEnd(stream); 862 if (ret != Z_OK) 863 log_err("fio: failed to end log inflation seq %d (%d)\n", 864 iter->seq, ret); 865 866 flush_samples(f, iter->buf, iter->buf_used); 867 free(iter->buf); 868 iter->buf = NULL; 869 iter->buf_size = iter->buf_used = 0; 870 } 871 872 /* 873 * Iterative chunk inflation. Handles cases where we cross into a new 874 * sequence, doing flush finish of previous chunk if needed. 875 */ 876 static size_t inflate_chunk(struct iolog_compress *ic, int gz_hdr, FILE *f, 877 z_stream *stream, struct inflate_chunk_iter *iter) 878 { 879 size_t ret; 880 881 dprint(FD_COMPRESS, "inflate chunk size=%lu, seq=%u\n", 882 (unsigned long) ic->len, ic->seq); 883 884 if (ic->seq != iter->seq) { 885 if (iter->seq) 886 finish_chunk(stream, f, iter); 887 888 z_stream_init(stream, gz_hdr); 889 iter->seq = ic->seq; 890 } 891 892 stream->avail_in = ic->len; 893 stream->next_in = ic->buf; 894 895 if (!iter->buf_size) { 896 iter->buf_size = iter->chunk_sz; 897 iter->buf = malloc(iter->buf_size); 898 } 899 900 while (stream->avail_in) { 901 size_t this_out = iter->buf_size - iter->buf_used; 902 int err; 903 904 stream->avail_out = this_out; 905 stream->next_out = iter->buf + iter->buf_used; 906 907 err = inflate(stream, Z_NO_FLUSH); 908 if (err < 0) { 909 log_err("fio: failed inflating log: %d\n", err); 910 iter->err = err; 911 break; 912 } 913 914 iter->buf_used += this_out - stream->avail_out; 915 916 if (!stream->avail_out) { 917 iter->buf_size += iter->chunk_sz; 918 iter->buf = realloc(iter->buf, iter->buf_size); 919 continue; 920 } 921 922 if (err == Z_STREAM_END) 923 break; 924 } 925 926 ret = (void *) stream->next_in - ic->buf; 927 928 dprint(FD_COMPRESS, "inflated to size=%lu\n", (unsigned long) iter->buf_size); 929 930 return ret; 931 } 932 933 /* 934 * Inflate stored compressed chunks, or write them directly to the log 935 * file if so instructed. 936 */ 937 static int inflate_gz_chunks(struct io_log *log, FILE *f) 938 { 939 struct inflate_chunk_iter iter = { .chunk_sz = log->log_gz, }; 940 z_stream stream; 941 942 while (!flist_empty(&log->chunk_list)) { 943 struct iolog_compress *ic; 944 945 ic = flist_first_entry(&log->chunk_list, struct iolog_compress, list); 946 flist_del(&ic->list); 947 948 if (log->log_gz_store) { 949 size_t ret; 950 951 dprint(FD_COMPRESS, "log write chunk size=%lu, " 952 "seq=%u\n", (unsigned long) ic->len, ic->seq); 953 954 ret = fwrite(ic->buf, ic->len, 1, f); 955 if (ret != 1 || ferror(f)) { 956 iter.err = errno; 957 log_err("fio: error writing compressed log\n"); 958 } 959 } else 960 inflate_chunk(ic, log->log_gz_store, f, &stream, &iter); 961 962 free_chunk(ic); 963 } 964 965 if (iter.seq) { 966 finish_chunk(&stream, f, &iter); 967 free(iter.buf); 968 } 969 970 return iter.err; 971 } 972 973 /* 974 * Open compressed log file and decompress the stored chunks and 975 * write them to stdout. The chunks are stored sequentially in the 976 * file, so we iterate over them and do them one-by-one. 977 */ 978 int iolog_file_inflate(const char *file) 979 { 980 struct inflate_chunk_iter iter = { .chunk_sz = 64 * 1024 * 1024, }; 981 struct iolog_compress ic; 982 z_stream stream; 983 struct stat sb; 984 ssize_t ret; 985 size_t total; 986 void *buf; 987 FILE *f; 988 989 f = fopen(file, "r"); 990 if (!f) { 991 perror("fopen"); 992 return 1; 993 } 994 995 if (stat(file, &sb) < 0) { 996 fclose(f); 997 perror("stat"); 998 return 1; 999 } 1000 1001 ic.buf = buf = malloc(sb.st_size); 1002 ic.len = sb.st_size; 1003 ic.seq = 1; 1004 1005 ret = fread(ic.buf, ic.len, 1, f); 1006 if (ret < 0) { 1007 perror("fread"); 1008 fclose(f); 1009 free(buf); 1010 return 1; 1011 } else if (ret != 1) { 1012 log_err("fio: short read on reading log\n"); 1013 fclose(f); 1014 free(buf); 1015 return 1; 1016 } 1017 1018 fclose(f); 1019 1020 /* 1021 * Each chunk will return Z_STREAM_END. We don't know how many 1022 * chunks are in the file, so we just keep looping and incrementing 1023 * the sequence number until we have consumed the whole compressed 1024 * file. 1025 */ 1026 total = ic.len; 1027 do { 1028 size_t iret; 1029 1030 iret = inflate_chunk(&ic, 1, stdout, &stream, &iter); 1031 total -= iret; 1032 if (!total) 1033 break; 1034 if (iter.err) 1035 break; 1036 1037 ic.seq++; 1038 ic.len -= iret; 1039 ic.buf += iret; 1040 } while (1); 1041 1042 if (iter.seq) { 1043 finish_chunk(&stream, stdout, &iter); 1044 free(iter.buf); 1045 } 1046 1047 free(buf); 1048 return iter.err; 1049 } 1050 1051 #else 1052 1053 static int inflate_gz_chunks(struct io_log *log, FILE *f) 1054 { 1055 return 0; 1056 } 1057 1058 int iolog_file_inflate(const char *file) 1059 { 1060 log_err("fio: log inflation not possible without zlib\n"); 1061 return 1; 1062 } 1063 1064 #endif 1065 1066 void flush_log(struct io_log *log, bool do_append) 1067 { 1068 void *buf; 1069 FILE *f; 1070 1071 if (!do_append) 1072 f = fopen(log->filename, "w"); 1073 else 1074 f = fopen(log->filename, "a"); 1075 if (!f) { 1076 perror("fopen log"); 1077 return; 1078 } 1079 1080 buf = set_file_buffer(f); 1081 1082 inflate_gz_chunks(log, f); 1083 1084 while (!flist_empty(&log->io_logs)) { 1085 struct io_logs *cur_log; 1086 1087 cur_log = flist_first_entry(&log->io_logs, struct io_logs, list); 1088 flist_del_init(&cur_log->list); 1089 1090 if (log->td && log == log->td->clat_hist_log) 1091 flush_hist_samples(f, log->hist_coarseness, cur_log->log, 1092 log_sample_sz(log, cur_log)); 1093 else 1094 flush_samples(f, cur_log->log, log_sample_sz(log, cur_log)); 1095 1096 sfree(cur_log); 1097 } 1098 1099 fclose(f); 1100 clear_file_buffer(buf); 1101 } 1102 1103 static int finish_log(struct thread_data *td, struct io_log *log, int trylock) 1104 { 1105 if (td->flags & TD_F_COMPRESS_LOG) 1106 iolog_flush(log); 1107 1108 if (trylock) { 1109 if (fio_trylock_file(log->filename)) 1110 return 1; 1111 } else 1112 fio_lock_file(log->filename); 1113 1114 if (td->client_type == FIO_CLIENT_TYPE_GUI || is_backend) 1115 fio_send_iolog(td, log, log->filename); 1116 else 1117 flush_log(log, !td->o.per_job_logs); 1118 1119 fio_unlock_file(log->filename); 1120 free_log(log); 1121 return 0; 1122 } 1123 1124 size_t log_chunk_sizes(struct io_log *log) 1125 { 1126 struct flist_head *entry; 1127 size_t ret; 1128 1129 if (flist_empty(&log->chunk_list)) 1130 return 0; 1131 1132 ret = 0; 1133 pthread_mutex_lock(&log->chunk_lock); 1134 flist_for_each(entry, &log->chunk_list) { 1135 struct iolog_compress *c; 1136 1137 c = flist_entry(entry, struct iolog_compress, list); 1138 ret += c->len; 1139 } 1140 pthread_mutex_unlock(&log->chunk_lock); 1141 return ret; 1142 } 1143 1144 #ifdef CONFIG_ZLIB 1145 1146 static int gz_work(struct iolog_flush_data *data) 1147 { 1148 struct iolog_compress *c = NULL; 1149 struct flist_head list; 1150 unsigned int seq; 1151 z_stream stream; 1152 size_t total = 0; 1153 int ret; 1154 1155 INIT_FLIST_HEAD(&list); 1156 1157 memset(&stream, 0, sizeof(stream)); 1158 stream.zalloc = Z_NULL; 1159 stream.zfree = Z_NULL; 1160 stream.opaque = Z_NULL; 1161 1162 ret = deflateInit(&stream, Z_DEFAULT_COMPRESSION); 1163 if (ret != Z_OK) { 1164 log_err("fio: failed to init gz stream\n"); 1165 goto err; 1166 } 1167 1168 seq = ++data->log->chunk_seq; 1169 1170 stream.next_in = (void *) data->samples; 1171 stream.avail_in = data->nr_samples * log_entry_sz(data->log); 1172 1173 dprint(FD_COMPRESS, "deflate input size=%lu, seq=%u, log=%s\n", 1174 (unsigned long) stream.avail_in, seq, 1175 data->log->filename); 1176 do { 1177 if (c) 1178 dprint(FD_COMPRESS, "seq=%d, chunk=%lu\n", seq, 1179 (unsigned long) c->len); 1180 c = get_new_chunk(seq); 1181 stream.avail_out = GZ_CHUNK; 1182 stream.next_out = c->buf; 1183 ret = deflate(&stream, Z_NO_FLUSH); 1184 if (ret < 0) { 1185 log_err("fio: deflate log (%d)\n", ret); 1186 free_chunk(c); 1187 goto err; 1188 } 1189 1190 c->len = GZ_CHUNK - stream.avail_out; 1191 flist_add_tail(&c->list, &list); 1192 total += c->len; 1193 } while (stream.avail_in); 1194 1195 stream.next_out = c->buf + c->len; 1196 stream.avail_out = GZ_CHUNK - c->len; 1197 1198 ret = deflate(&stream, Z_FINISH); 1199 if (ret < 0) { 1200 /* 1201 * Z_BUF_ERROR is special, it just means we need more 1202 * output space. We'll handle that below. Treat any other 1203 * error as fatal. 1204 */ 1205 if (ret != Z_BUF_ERROR) { 1206 log_err("fio: deflate log (%d)\n", ret); 1207 flist_del(&c->list); 1208 free_chunk(c); 1209 goto err; 1210 } 1211 } 1212 1213 total -= c->len; 1214 c->len = GZ_CHUNK - stream.avail_out; 1215 total += c->len; 1216 dprint(FD_COMPRESS, "seq=%d, chunk=%lu\n", seq, (unsigned long) c->len); 1217 1218 if (ret != Z_STREAM_END) { 1219 do { 1220 c = get_new_chunk(seq); 1221 stream.avail_out = GZ_CHUNK; 1222 stream.next_out = c->buf; 1223 ret = deflate(&stream, Z_FINISH); 1224 c->len = GZ_CHUNK - stream.avail_out; 1225 total += c->len; 1226 flist_add_tail(&c->list, &list); 1227 dprint(FD_COMPRESS, "seq=%d, chunk=%lu\n", seq, 1228 (unsigned long) c->len); 1229 } while (ret != Z_STREAM_END); 1230 } 1231 1232 dprint(FD_COMPRESS, "deflated to size=%lu\n", (unsigned long) total); 1233 1234 ret = deflateEnd(&stream); 1235 if (ret != Z_OK) 1236 log_err("fio: deflateEnd %d\n", ret); 1237 1238 free(data->samples); 1239 1240 if (!flist_empty(&list)) { 1241 pthread_mutex_lock(&data->log->chunk_lock); 1242 flist_splice_tail(&list, &data->log->chunk_list); 1243 pthread_mutex_unlock(&data->log->chunk_lock); 1244 } 1245 1246 ret = 0; 1247 done: 1248 if (data->free) 1249 free(data); 1250 return ret; 1251 err: 1252 while (!flist_empty(&list)) { 1253 c = flist_first_entry(list.next, struct iolog_compress, list); 1254 flist_del(&c->list); 1255 free_chunk(c); 1256 } 1257 ret = 1; 1258 goto done; 1259 } 1260 1261 /* 1262 * Invoked from our compress helper thread, when logging would have exceeded 1263 * the specified memory limitation. Compresses the previously stored 1264 * entries. 1265 */ 1266 static int gz_work_async(struct submit_worker *sw, struct workqueue_work *work) 1267 { 1268 return gz_work(container_of(work, struct iolog_flush_data, work)); 1269 } 1270 1271 static int gz_init_worker(struct submit_worker *sw) 1272 { 1273 struct thread_data *td = sw->wq->td; 1274 1275 if (!fio_option_is_set(&td->o, log_gz_cpumask)) 1276 return 0; 1277 1278 if (fio_setaffinity(gettid(), td->o.log_gz_cpumask) == -1) { 1279 log_err("gz: failed to set CPU affinity\n"); 1280 return 1; 1281 } 1282 1283 return 0; 1284 } 1285 1286 static struct workqueue_ops log_compress_wq_ops = { 1287 .fn = gz_work_async, 1288 .init_worker_fn = gz_init_worker, 1289 .nice = 1, 1290 }; 1291 1292 int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out) 1293 { 1294 if (!(td->flags & TD_F_COMPRESS_LOG)) 1295 return 0; 1296 1297 workqueue_init(td, &td->log_compress_wq, &log_compress_wq_ops, 1, sk_out); 1298 return 0; 1299 } 1300 1301 void iolog_compress_exit(struct thread_data *td) 1302 { 1303 if (!(td->flags & TD_F_COMPRESS_LOG)) 1304 return; 1305 1306 workqueue_exit(&td->log_compress_wq); 1307 } 1308 1309 /* 1310 * Queue work item to compress the existing log entries. We reset the 1311 * current log to a small size, and reference the existing log in the 1312 * data that we queue for compression. Once compression has been done, 1313 * this old log is freed. If called with finish == true, will not return 1314 * until the log compression has completed, and will flush all previous 1315 * logs too 1316 */ 1317 static int iolog_flush(struct io_log *log) 1318 { 1319 struct iolog_flush_data *data; 1320 1321 data = malloc(sizeof(*data)); 1322 if (!data) 1323 return 1; 1324 1325 data->log = log; 1326 data->free = false; 1327 1328 while (!flist_empty(&log->io_logs)) { 1329 struct io_logs *cur_log; 1330 1331 cur_log = flist_first_entry(&log->io_logs, struct io_logs, list); 1332 flist_del_init(&cur_log->list); 1333 1334 data->samples = cur_log->log; 1335 data->nr_samples = cur_log->nr_samples; 1336 1337 sfree(cur_log); 1338 1339 gz_work(data); 1340 } 1341 1342 free(data); 1343 return 0; 1344 } 1345 1346 int iolog_cur_flush(struct io_log *log, struct io_logs *cur_log) 1347 { 1348 struct iolog_flush_data *data; 1349 1350 data = malloc(sizeof(*data)); 1351 if (!data) 1352 return 1; 1353 1354 data->log = log; 1355 1356 data->samples = cur_log->log; 1357 data->nr_samples = cur_log->nr_samples; 1358 data->free = true; 1359 1360 cur_log->nr_samples = cur_log->max_samples = 0; 1361 cur_log->log = NULL; 1362 1363 workqueue_enqueue(&log->td->log_compress_wq, &data->work); 1364 return 0; 1365 } 1366 #else 1367 1368 static int iolog_flush(struct io_log *log) 1369 { 1370 return 1; 1371 } 1372 1373 int iolog_cur_flush(struct io_log *log, struct io_logs *cur_log) 1374 { 1375 return 1; 1376 } 1377 1378 int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out) 1379 { 1380 return 0; 1381 } 1382 1383 void iolog_compress_exit(struct thread_data *td) 1384 { 1385 } 1386 1387 #endif 1388 1389 struct io_logs *iolog_cur_log(struct io_log *log) 1390 { 1391 if (flist_empty(&log->io_logs)) 1392 return NULL; 1393 1394 return flist_last_entry(&log->io_logs, struct io_logs, list); 1395 } 1396 1397 uint64_t iolog_nr_samples(struct io_log *iolog) 1398 { 1399 struct flist_head *entry; 1400 uint64_t ret = 0; 1401 1402 flist_for_each(entry, &iolog->io_logs) { 1403 struct io_logs *cur_log; 1404 1405 cur_log = flist_entry(entry, struct io_logs, list); 1406 ret += cur_log->nr_samples; 1407 } 1408 1409 return ret; 1410 } 1411 1412 static int __write_log(struct thread_data *td, struct io_log *log, int try) 1413 { 1414 if (log) 1415 return finish_log(td, log, try); 1416 1417 return 0; 1418 } 1419 1420 static int write_iops_log(struct thread_data *td, int try, bool unit_log) 1421 { 1422 int ret; 1423 1424 if (per_unit_log(td->iops_log) != unit_log) 1425 return 0; 1426 1427 ret = __write_log(td, td->iops_log, try); 1428 if (!ret) 1429 td->iops_log = NULL; 1430 1431 return ret; 1432 } 1433 1434 static int write_slat_log(struct thread_data *td, int try, bool unit_log) 1435 { 1436 int ret; 1437 1438 if (!unit_log) 1439 return 0; 1440 1441 ret = __write_log(td, td->slat_log, try); 1442 if (!ret) 1443 td->slat_log = NULL; 1444 1445 return ret; 1446 } 1447 1448 static int write_clat_log(struct thread_data *td, int try, bool unit_log) 1449 { 1450 int ret; 1451 1452 if (!unit_log) 1453 return 0; 1454 1455 ret = __write_log(td, td->clat_log, try); 1456 if (!ret) 1457 td->clat_log = NULL; 1458 1459 return ret; 1460 } 1461 1462 static int write_clat_hist_log(struct thread_data *td, int try, bool unit_log) 1463 { 1464 int ret; 1465 1466 if (!unit_log) 1467 return 0; 1468 1469 ret = __write_log(td, td->clat_hist_log, try); 1470 if (!ret) 1471 td->clat_hist_log = NULL; 1472 1473 return ret; 1474 } 1475 1476 static int write_lat_log(struct thread_data *td, int try, bool unit_log) 1477 { 1478 int ret; 1479 1480 if (!unit_log) 1481 return 0; 1482 1483 ret = __write_log(td, td->lat_log, try); 1484 if (!ret) 1485 td->lat_log = NULL; 1486 1487 return ret; 1488 } 1489 1490 static int write_bandw_log(struct thread_data *td, int try, bool unit_log) 1491 { 1492 int ret; 1493 1494 if (per_unit_log(td->bw_log) != unit_log) 1495 return 0; 1496 1497 ret = __write_log(td, td->bw_log, try); 1498 if (!ret) 1499 td->bw_log = NULL; 1500 1501 return ret; 1502 } 1503 1504 enum { 1505 BW_LOG_MASK = 1, 1506 LAT_LOG_MASK = 2, 1507 SLAT_LOG_MASK = 4, 1508 CLAT_LOG_MASK = 8, 1509 IOPS_LOG_MASK = 16, 1510 CLAT_HIST_LOG_MASK = 32, 1511 1512 ALL_LOG_NR = 6, 1513 }; 1514 1515 struct log_type { 1516 unsigned int mask; 1517 int (*fn)(struct thread_data *, int, bool); 1518 }; 1519 1520 static struct log_type log_types[] = { 1521 { 1522 .mask = BW_LOG_MASK, 1523 .fn = write_bandw_log, 1524 }, 1525 { 1526 .mask = LAT_LOG_MASK, 1527 .fn = write_lat_log, 1528 }, 1529 { 1530 .mask = SLAT_LOG_MASK, 1531 .fn = write_slat_log, 1532 }, 1533 { 1534 .mask = CLAT_LOG_MASK, 1535 .fn = write_clat_log, 1536 }, 1537 { 1538 .mask = IOPS_LOG_MASK, 1539 .fn = write_iops_log, 1540 }, 1541 { 1542 .mask = CLAT_HIST_LOG_MASK, 1543 .fn = write_clat_hist_log, 1544 } 1545 }; 1546 1547 void td_writeout_logs(struct thread_data *td, bool unit_logs) 1548 { 1549 unsigned int log_mask = 0; 1550 unsigned int log_left = ALL_LOG_NR; 1551 int old_state, i; 1552 1553 old_state = td_bump_runstate(td, TD_FINISHING); 1554 1555 finalize_logs(td, unit_logs); 1556 1557 while (log_left) { 1558 int prev_log_left = log_left; 1559 1560 for (i = 0; i < ALL_LOG_NR && log_left; i++) { 1561 struct log_type *lt = &log_types[i]; 1562 int ret; 1563 1564 if (!(log_mask & lt->mask)) { 1565 ret = lt->fn(td, log_left != 1, unit_logs); 1566 if (!ret) { 1567 log_left--; 1568 log_mask |= lt->mask; 1569 } 1570 } 1571 } 1572 1573 if (prev_log_left == log_left) 1574 usleep(5000); 1575 } 1576 1577 td_restore_runstate(td, old_state); 1578 } 1579 1580 void fio_writeout_logs(bool unit_logs) 1581 { 1582 struct thread_data *td; 1583 int i; 1584 1585 for_each_td(td, i) 1586 td_writeout_logs(td, unit_logs); 1587 } 1588