X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=iolog.c;h=b29684a85eb69c30d219b95eff122c33df9ce5f6;hp=7c05d3284a724fefdb56f2a181c16721a80b05c2;hb=55312f9f5572611037b6254912460f64173df574;hpb=aee2ab6775d96609a4632703827c409a7f9abcca diff --git a/iolog.c b/iolog.c index 7c05d328..b29684a8 100644 --- a/iolog.c +++ b/iolog.c @@ -6,6 +6,9 @@ #include #include #include +#include +#include +#include #ifdef CONFIG_ZLIB #include #endif @@ -15,60 +18,24 @@ #include "verify.h" #include "trim.h" #include "filelock.h" -#include "tp.h" +#include "lib/tp.h" static const char iolog_ver2[] = "fio version 2 iolog"; -#ifdef CONFIG_ZLIB - -struct iolog_compress { - struct flist_head list; - void *buf; - size_t len; - unsigned int seq; -}; - -#define GZ_CHUNK 131072 - -static struct iolog_compress *get_new_chunk(unsigned int seq) -{ - struct iolog_compress *c; - - c = malloc(sizeof(*c)); - INIT_FLIST_HEAD(&c->list); - c->buf = malloc(GZ_CHUNK); - c->len = 0; - c->seq = seq; - return c; -} - -static void free_chunk(struct iolog_compress *ic) -{ - free(ic->buf); - free(ic); -} - -#endif - void queue_io_piece(struct thread_data *td, struct io_piece *ipo) { flist_add_tail(&ipo->list, &td->io_log_list); td->total_io_size += ipo->len; } -void log_io_u(struct thread_data *td, struct io_u *io_u) +void log_io_u(const struct thread_data *td, const struct io_u *io_u) { - const char *act[] = { "read", "write", "sync", "datasync", - "sync_file_range", "wait", "trim" }; - - assert(io_u->ddir <= 6); - if (!td->o.write_iolog_file) return; fprintf(td->iolog_f, "%s %s %llu %lu\n", io_u->file->file_name, - act[io_u->ddir], io_u->offset, - io_u->buflen); + io_ddir_name(io_u->ddir), + io_u->offset, io_u->buflen); } void log_file(struct thread_data *td, struct fio_file *f, @@ -93,20 +60,22 @@ void log_file(struct thread_data *td, struct fio_file *f, static void iolog_delay(struct thread_data *td, unsigned long delay) { - unsigned long usec = utime_since_now(&td->last_issue); - unsigned long this_delay; + uint64_t usec = utime_since_now(&td->last_issue); + uint64_t this_delay; + struct timeval tv; + if (delay < td->time_offset) { + td->time_offset = 0; + return; + } + + delay -= td->time_offset; if (delay < usec) return; delay -= usec; - /* - * less than 100 usec delay, just regard it as noise - */ - if (delay < 100) - return; - + fio_gettime(&tv, NULL); while (delay && !td->terminate) { this_delay = delay; if (this_delay > 500000) @@ -115,6 +84,12 @@ static void iolog_delay(struct thread_data *td, unsigned long delay) usec_sleep(td, this_delay); delay -= this_delay; } + + usec = utime_since_now(&tv); + if (usec > delay) + td->time_offset = usec - delay; + else + td->time_offset = 0; } static int ipo_special(struct thread_data *td, struct io_piece *ipo) @@ -141,7 +116,7 @@ static int ipo_special(struct thread_data *td, struct io_piece *ipo) td_io_close_file(td, f); break; case FIO_LOG_UNLINK_FILE: - unlink(f->file_name); + td_io_unlink_file(td, f); break; default: log_err("fio: bad file action %d\n", ipo->file_action); @@ -159,7 +134,7 @@ int read_iolog_get(struct thread_data *td, struct io_u *io_u) while (!flist_empty(&td->io_log_list)) { int ret; - ipo = flist_entry(td->io_log_list.next, struct io_piece, list); + ipo = flist_first_entry(&td->io_log_list, struct io_piece, list); flist_del(&ipo->list); remove_trim_entry(td, ipo); @@ -212,7 +187,7 @@ void prune_io_piece_log(struct thread_data *td) } while (!flist_empty(&td->io_hist_list)) { - ipo = flist_entry(td->io_hist_list.next, struct io_piece, list); + ipo = flist_first_entry(&td->io_hist_list, struct io_piece, list); flist_del(&ipo->list); remove_trim_entry(td, ipo); td->io_hist_len--; @@ -275,6 +250,7 @@ restart: p = &td->io_hist_tree.rb_node; parent = NULL; while (*p) { + int overlap = 0; parent = *p; __ipo = rb_entry(parent, struct io_piece, rb_node); @@ -282,11 +258,18 @@ restart: p = &(*p)->rb_left; else if (ipo->file > __ipo->file) p = &(*p)->rb_right; - else if (ipo->offset < __ipo->offset) + else if (ipo->offset < __ipo->offset) { p = &(*p)->rb_left; - else if (ipo->offset > __ipo->offset) + overlap = ipo->offset + ipo->len > __ipo->offset; + } + else if (ipo->offset > __ipo->offset) { p = &(*p)->rb_right; - else { + overlap = __ipo->offset + __ipo->len > ipo->offset; + } + else + overlap = 1; + + if (overlap) { dprint(FD_IO, "iolog: overlap %llu/%lu, %llu/%lu", __ipo->offset, __ipo->len, ipo->offset, ipo->len); @@ -308,6 +291,18 @@ void unlog_io_piece(struct thread_data *td, struct io_u *io_u) { struct io_piece *ipo = io_u->ipo; + if (td->ts.nr_block_infos) { + uint32_t *info = io_u_block_info(td, io_u); + if (BLOCK_INFO_STATE(*info) < BLOCK_STATE_TRIM_FAILURE) { + if (io_u->ddir == DDIR_TRIM) + *info = BLOCK_INFO_SET_STATE(*info, + BLOCK_STATE_TRIM_FAILURE); + else if (io_u->ddir == DDIR_WRITE) + *info = BLOCK_INFO_SET_STATE(*info, + BLOCK_STATE_WRITE_FAILURE); + } + } + if (!ipo) return; @@ -321,7 +316,7 @@ void unlog_io_piece(struct thread_data *td, struct io_u *io_u) td->io_hist_len--; } -void trim_io_piece(struct thread_data *td, struct io_u *io_u) +void trim_io_piece(struct thread_data *td, const struct io_u *io_u) { struct io_piece *ipo = io_u->ipo; @@ -578,19 +573,23 @@ int init_iolog(struct thread_data *td) void setup_log(struct io_log **log, struct log_params *p, const char *filename) { - struct io_log *l = malloc(sizeof(*l)); + struct io_log *l; - memset(l, 0, sizeof(*l)); + l = calloc(1, sizeof(*l)); l->nr_samples = 0; l->max_samples = 1024; l->log_type = p->log_type; l->log_offset = p->log_offset; l->log_gz = p->log_gz; + l->log_gz_store = p->log_gz_store; l->log = malloc(l->max_samples * log_entry_sz(l)); l->avg_msec = p->avg_msec; l->filename = strdup(filename); l->td = p->td; + if (l->log_offset) + l->log_ddir_mask = LOG_OFFSET_SAMPLE_BIT; + INIT_FLIST_HEAD(&l->chunk_list); if (l->log_gz && !p->td) @@ -636,78 +635,139 @@ void free_log(struct io_log *log) free(log); } -static void flush_samples(FILE *f, void *samples, uint64_t nr_samples, - int log_offset) +static void flush_samples(FILE *f, void *samples, uint64_t sample_size) { - uint64_t i; + struct io_sample *s; + int log_offset; + uint64_t i, nr_samples; + + if (!sample_size) + return; + + s = __get_sample(samples, 0, 0); + log_offset = (s->__ddir & LOG_OFFSET_SAMPLE_BIT) != 0; + + nr_samples = sample_size / __log_entry_sz(log_offset); for (i = 0; i < nr_samples; i++) { - struct io_sample *s = __get_sample(samples, log_offset, i); + s = __get_sample(samples, log_offset, i); if (!log_offset) { fprintf(f, "%lu, %lu, %u, %u\n", (unsigned long) s->time, (unsigned long) s->val, - s->ddir, s->bs); + io_sample_ddir(s), s->bs); } else { struct io_sample_offset *so = (void *) s; fprintf(f, "%lu, %lu, %u, %u, %llu\n", (unsigned long) s->time, (unsigned long) s->val, - s->ddir, s->bs, + io_sample_ddir(s), s->bs, (unsigned long long) so->offset); } } } #ifdef CONFIG_ZLIB -static int z_stream_init(z_stream *stream) + +struct iolog_flush_data { + struct tp_work work; + struct io_log *log; + void *samples; + uint64_t nr_samples; +}; + +struct iolog_compress { + struct flist_head list; + void *buf; + size_t len; + unsigned int seq; +}; + +#define GZ_CHUNK 131072 + +static struct iolog_compress *get_new_chunk(unsigned int seq) +{ + struct iolog_compress *c; + + c = malloc(sizeof(*c)); + INIT_FLIST_HEAD(&c->list); + c->buf = malloc(GZ_CHUNK); + c->len = 0; + c->seq = seq; + return c; +} + +static void free_chunk(struct iolog_compress *ic) +{ + free(ic->buf); + free(ic); +} + +static int z_stream_init(z_stream *stream, int gz_hdr) { + int wbits = 15; + stream->zalloc = Z_NULL; stream->zfree = Z_NULL; stream->opaque = Z_NULL; stream->next_in = Z_NULL; - if (inflateInit(stream) != Z_OK) + /* + * zlib magic - add 32 for auto-detection of gz header or not, + * if we decide to store files in a gzip friendly format. + */ + if (gz_hdr) + wbits += 32; + + if (inflateInit2(stream, wbits) != Z_OK) return 1; return 0; } -struct flush_chunk_iter { +struct inflate_chunk_iter { unsigned int seq; + int err; void *buf; size_t buf_size; size_t buf_used; size_t chunk_sz; }; -static void finish_chunk(z_stream *stream, int log_offset, FILE *f, - struct flush_chunk_iter *iter) +static void finish_chunk(z_stream *stream, FILE *f, + struct inflate_chunk_iter *iter) { - uint64_t nr_samples; int ret; ret = inflateEnd(stream); if (ret != Z_OK) log_err("fio: failed to end log inflation (%d)\n", ret); - nr_samples = iter->buf_used / __log_entry_sz(log_offset); - flush_samples(f, iter->buf, nr_samples, log_offset); + flush_samples(f, iter->buf, iter->buf_used); free(iter->buf); iter->buf = NULL; iter->buf_size = iter->buf_used = 0; } -static int flush_chunk(struct iolog_compress *ic, int log_offset, FILE *f, - z_stream *stream, struct flush_chunk_iter *iter) +/* + * Iterative chunk inflation. Handles cases where we cross into a new + * sequence, doing flush finish of previous chunk if needed. + */ +static size_t inflate_chunk(struct iolog_compress *ic, int gz_hdr, FILE *f, + z_stream *stream, struct inflate_chunk_iter *iter) { + size_t ret; + + dprint(FD_COMPRESS, "inflate chunk size=%lu, seq=%u", + (unsigned long) ic->len, ic->seq); + if (ic->seq != iter->seq) { if (iter->seq) - finish_chunk(stream, log_offset, f, iter); + finish_chunk(stream, f, iter); - z_stream_init(stream); + z_stream_init(stream, gz_hdr); iter->seq = ic->seq; } @@ -720,49 +780,167 @@ static int flush_chunk(struct iolog_compress *ic, int log_offset, FILE *f, } while (stream->avail_in) { + size_t this_out = iter->buf_size - iter->buf_used; int err; - stream->avail_out = iter->buf_size - iter->buf_used; + stream->avail_out = this_out; stream->next_out = iter->buf + iter->buf_used; err = inflate(stream, Z_NO_FLUSH); if (err < 0) { log_err("fio: failed inflating log: %d\n", err); + iter->err = err; break; } - iter->buf_used += iter->buf_size - iter->buf_used - stream->avail_out; + iter->buf_used += this_out - stream->avail_out; + + if (!stream->avail_out) { + iter->buf_size += iter->chunk_sz; + iter->buf = realloc(iter->buf, iter->buf_size); + continue; + } + + if (err == Z_STREAM_END) + break; } - free_chunk(ic); - return 0; + ret = (void *) stream->next_in - ic->buf; + + dprint(FD_COMPRESS, "inflated to size=%lu\n", (unsigned long) ret); + + return ret; } -static void flush_gz_chunks(struct io_log *log, FILE *f) +/* + * Inflate stored compressed chunks, or write them directly to the log + * file if so instructed. + */ +static int inflate_gz_chunks(struct io_log *log, FILE *f) { - struct flush_chunk_iter iter = { .chunk_sz = log->log_gz, }; - struct flist_head *node; + struct inflate_chunk_iter iter = { .chunk_sz = log->log_gz, }; z_stream stream; while (!flist_empty(&log->chunk_list)) { struct iolog_compress *ic; - node = log->chunk_list.next; - ic = flist_entry(node, struct iolog_compress, list); + ic = flist_first_entry(&log->chunk_list, struct iolog_compress, list); flist_del(&ic->list); - flush_chunk(ic, log->log_offset, f, &stream, &iter); + + if (log->log_gz_store) { + size_t ret; + + dprint(FD_COMPRESS, "log write chunk size=%lu, " + "seq=%u\n", (unsigned long) ic->len, ic->seq); + + ret = fwrite(ic->buf, ic->len, 1, f); + if (ret != 1 || ferror(f)) { + iter.err = errno; + log_err("fio: error writing compressed log\n"); + } + } else + inflate_chunk(ic, log->log_gz_store, f, &stream, &iter); + + free_chunk(ic); + } + + if (iter.seq) { + finish_chunk(&stream, f, &iter); + free(iter.buf); + } + + return iter.err; +} + +/* + * Open compressed log file and decompress the stored chunks and + * write them to stdout. The chunks are stored sequentially in the + * file, so we iterate over them and do them one-by-one. + */ +int iolog_file_inflate(const char *file) +{ + struct inflate_chunk_iter iter = { .chunk_sz = 64 * 1024 * 1024, }; + struct iolog_compress ic; + z_stream stream; + struct stat sb; + ssize_t ret; + size_t total; + void *buf; + FILE *f; + + f = fopen(file, "r"); + if (!f) { + perror("fopen"); + return 1; + } + + if (stat(file, &sb) < 0) { + fclose(f); + perror("stat"); + return 1; } + ic.buf = buf = malloc(sb.st_size); + ic.len = sb.st_size; + ic.seq = 1; + + ret = fread(ic.buf, ic.len, 1, f); + if (ret < 0) { + perror("fread"); + fclose(f); + free(buf); + return 1; + } else if (ret != 1) { + log_err("fio: short read on reading log\n"); + fclose(f); + free(buf); + return 1; + } + + fclose(f); + + /* + * Each chunk will return Z_STREAM_END. We don't know how many + * chunks are in the file, so we just keep looping and incrementing + * the sequence number until we have consumed the whole compressed + * file. + */ + total = ic.len; + do { + size_t iret; + + iret = inflate_chunk(&ic, 1, stdout, &stream, &iter); + total -= iret; + if (!total) + break; + if (iter.err) + break; + + ic.seq++; + ic.len -= iret; + ic.buf += iret; + } while (1); + if (iter.seq) { - finish_chunk(&stream, log->log_offset, f, &iter); + finish_chunk(&stream, stdout, &iter); free(iter.buf); } + + free(buf); + return iter.err; } #else -static void flush_gz_chunks(struct io_log *log, FILE *f) +static int inflate_gz_chunks(struct io_log *log, FILE *f) { + return 0; +} + +int iolog_file_inflate(const char *file) +{ + log_err("fio: log inflation not possible without zlib\n"); + return 1; } #endif @@ -780,9 +958,9 @@ void flush_log(struct io_log *log) buf = set_file_buffer(f); - flush_gz_chunks(log, f); + inflate_gz_chunks(log, f); - flush_samples(f, log->log, log->nr_samples, log->log_offset); + flush_samples(f, log->log, log->nr_samples * log_entry_sz(log)); fclose(f); clear_file_buffer(buf); @@ -811,13 +989,11 @@ static int finish_log(struct thread_data *td, struct io_log *log, int trylock) #ifdef CONFIG_ZLIB -struct iolog_flush_data { - struct tp_work work; - struct io_log *log; - void *samples; - uint64_t nr_samples; -}; - +/* + * Invoked from our compress helper thread, when logging would have exceeded + * the specified memory limitation. Compresses the previously stored + * entries. + */ static int gz_work(struct tp_work *work) { struct iolog_flush_data *data; @@ -836,15 +1012,19 @@ static int gz_work(struct tp_work *work) stream.zfree = Z_NULL; stream.opaque = Z_NULL; - if (deflateInit(&stream, Z_DEFAULT_COMPRESSION) != Z_OK) { + ret = deflateInit(&stream, Z_DEFAULT_COMPRESSION); + if (ret != Z_OK) { log_err("fio: failed to init gz stream\n"); return 0; } seq = ++data->log->chunk_seq; + stream.next_in = (void *) data->samples; stream.avail_in = data->nr_samples * log_entry_sz(data->log); + dprint(FD_COMPRESS, "deflate input size=%lu, seq=%u\n", + (unsigned long) stream.avail_in, seq); do { c = get_new_chunk(seq); stream.avail_out = GZ_CHUNK; @@ -852,7 +1032,8 @@ static int gz_work(struct tp_work *work) ret = deflate(&stream, Z_NO_FLUSH); if (ret < 0) { log_err("fio: deflate log (%d)\n", ret); - break; + free_chunk(c); + goto err; } c->len = GZ_CHUNK - stream.avail_out; @@ -873,10 +1054,13 @@ static int gz_work(struct tp_work *work) stream.next_out = c->buf; ret = deflate(&stream, Z_FINISH); c->len = GZ_CHUNK - stream.avail_out; + total += c->len; flist_add_tail(&c->list, &list); } while (ret != Z_STREAM_END); } + dprint(FD_COMPRESS, "deflated to size=%lu\n", (unsigned long) total); + ret = deflateEnd(&stream); if (ret != Z_OK) log_err("fio: deflateEnd %d\n", ret); @@ -889,18 +1073,34 @@ static int gz_work(struct tp_work *work) pthread_mutex_unlock(&data->log->chunk_lock); } + ret = 0; +done: if (work->wait) { work->done = 1; pthread_cond_signal(&work->cv); } else free(data); - return 0; + return ret; +err: + while (!flist_empty(&list)) { + c = flist_first_entry(list.next, struct iolog_compress, list); + flist_del(&c->list); + free_chunk(c); + } + ret = 1; + goto done; } +/* + * Queue work item to compress the existing log entries. We copy the + * samples, and reset the log sample count to 0 (so the logging will + * continue to use the memory associated with the log). If called with + * wait == 1, will not return until the log compression has completed. + */ int iolog_flush(struct io_log *log, int wait) { - struct thread_data *td = log->td; + struct tp_data *tdat = log->td->tp_data; struct iolog_flush_data *data; size_t sample_size; @@ -929,7 +1129,8 @@ int iolog_flush(struct io_log *log, int wait) } else data->work.wait = 0; - tp_queue_work(td->tp_data, &data->work); + data->work.prio = 1; + tp_queue_work(tdat, &data->work); if (wait) { pthread_mutex_lock(&data->work.lock);