X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=io_u.c;h=c93896001296419f1a348f2b5f9f46f660b0a366;hp=ebcc8bb283eb6a65e93db799cfe9e2f83993f85e;hb=fa01d139c558a3788154c5b0f094bfdb5325728f;hpb=3af6ef399a9df324ffe4a8c8e03b52a42f587229 diff --git a/io_u.c b/io_u.c index ebcc8bb2..c9389600 100644 --- a/io_u.c +++ b/io_u.c @@ -3,10 +3,20 @@ #include #include #include +#include #include "fio.h" #include "os.h" +struct io_completion_data { + int nr; /* input */ + endio_handler *handler; /* input */ + + int error; /* output */ + unsigned long bytes_done[2]; /* output */ + struct timeval time; /* output */ +}; + /* * The ->file_map[] contains a map of blocks we have or have not done io * to yet. Used to make sure we cover the entire range in a fair fashion. @@ -184,12 +194,25 @@ static enum fio_ddir get_rw_ddir(struct thread_data *td) void put_io_u(struct thread_data *td, struct io_u *io_u) { + assert((io_u->flags & IO_U_F_FREE) == 0); + io_u->flags |= IO_U_F_FREE; + io_u->file = NULL; list_del(&io_u->list); list_add(&io_u->list, &td->io_u_freelist); td->cur_depth--; } +void requeue_io_u(struct thread_data *td, struct io_u **io_u) +{ + struct io_u *__io_u = *io_u; + + list_del(&__io_u->list); + list_add_tail(&__io_u->list, &td->io_u_requeues); + td->cur_depth--; + *io_u = NULL; +} + static int fill_io_u(struct thread_data *td, struct fio_file *f, struct io_u *io_u) { @@ -202,8 +225,8 @@ static int fill_io_u(struct thread_data *td, struct fio_file *f, /* * see if it's time to sync */ - if (td->fsync_blocks && !(td->io_blocks[DDIR_WRITE] % td->fsync_blocks) - && should_fsync(td)) { + if (td->fsync_blocks && !(td->io_issues[DDIR_WRITE] % td->fsync_blocks) + && td->io_issues[DDIR_WRITE] && should_fsync(td)) { io_u->ddir = DDIR_SYNC; io_u->file = f; return 0; @@ -270,23 +293,23 @@ static void io_u_mark_latency(struct thread_data *td, unsigned long msec) switch (msec) { default: index++; - case 1024 ... 2047: + case 1000 ... 1999: index++; - case 512 ... 1023: + case 750 ... 999: index++; - case 256 ... 511: + case 500 ... 749: index++; - case 128 ... 255: + case 250 ... 499: index++; - case 64 ... 127: + case 100 ... 249: index++; - case 32 ... 63: + case 50 ... 99: index++; - case 16 ... 31: + case 20 ... 49: index++; - case 8 ... 15: + case 10 ... 19: index++; - case 4 ... 7: + case 4 ... 9: index++; case 2 ... 3: index++; @@ -297,16 +320,46 @@ static void io_u_mark_latency(struct thread_data *td, unsigned long msec) td->io_u_lat[index]++; } +static struct fio_file *get_next_file(struct thread_data *td) +{ + unsigned int old_next_file = td->next_file; + struct fio_file *f; + + do { + f = &td->files[td->next_file]; + + td->next_file++; + if (td->next_file >= td->nr_files) + td->next_file = 0; + + if (f->fd != -1) + break; + + f = NULL; + } while (td->next_file != old_next_file); + + return f; +} + struct io_u *__get_io_u(struct thread_data *td) { struct io_u *io_u = NULL; - if (!queue_full(td)) { + if (!list_empty(&td->io_u_requeues)) + io_u = list_entry(td->io_u_requeues.next, struct io_u, list); + else if (!queue_full(td)) { io_u = list_entry(td->io_u_freelist.next, struct io_u, list); io_u->buflen = 0; - io_u->error = 0; io_u->resid = 0; + io_u->file = NULL; + } + + if (io_u) { + assert(io_u->flags & IO_U_F_FREE); + io_u->flags &= ~IO_U_F_FREE; + + io_u->error = 0; list_del(&io_u->list); list_add(&io_u->list, &td->io_u_busylist); td->cur_depth++; @@ -320,14 +373,29 @@ struct io_u *__get_io_u(struct thread_data *td) * Return an io_u to be processed. Gets a buflen and offset, sets direction, * etc. The returned io_u is fully ready to be prepped and submitted. */ -struct io_u *get_io_u(struct thread_data *td, struct fio_file *f) +struct io_u *get_io_u(struct thread_data *td) { + struct fio_file *f; struct io_u *io_u; io_u = __get_io_u(td); if (!io_u) return NULL; + /* + * from a requeue, io_u already setup + */ + if (io_u->file) + goto out; + + f = get_next_file(td); + if (!f) { + put_io_u(td, io_u); + return NULL; + } + + io_u->file = f; + if (td->zone_bytes >= td->zone_size) { td->zone_bytes = 0; f->last_pos += td->zone_skip; @@ -362,6 +430,7 @@ struct io_u *get_io_u(struct thread_data *td, struct fio_file *f) /* * Set io data pointers. */ +out: io_u->xfer_buf = io_u->buf; io_u->xfer_buflen = io_u->buflen; @@ -374,11 +443,14 @@ struct io_u *get_io_u(struct thread_data *td, struct fio_file *f) return io_u; } -void io_completed(struct thread_data *td, struct io_u *io_u, - struct io_completion_data *icd) +static void io_completed(struct thread_data *td, struct io_u *io_u, + struct io_completion_data *icd) { unsigned long msec; + assert(io_u->flags & IO_U_F_FLIGHT); + io_u->flags &= ~IO_U_F_FLIGHT; + if (io_u->ddir == DDIR_SYNC) { td->last_was_sync = 1; return; @@ -418,7 +490,8 @@ void io_completed(struct thread_data *td, struct io_u *io_u, icd->error = io_u->error; } -void init_icd(struct io_completion_data *icd, icd_handler *handler, int nr) +static void init_icd(struct io_completion_data *icd, endio_handler *handler, + int nr) { fio_gettime(&icd->time, NULL); @@ -429,7 +502,8 @@ void init_icd(struct io_completion_data *icd, icd_handler *handler, int nr) icd->bytes_done[0] = icd->bytes_done[1] = 0; } -void ios_completed(struct thread_data *td, struct io_completion_data *icd) +static void ios_completed(struct thread_data *td, + struct io_completion_data *icd) { struct io_u *io_u; int i; @@ -441,3 +515,131 @@ void ios_completed(struct thread_data *td, struct io_completion_data *icd) put_io_u(td, io_u); } } + +/* + * Complete a single io_u for the sync engines. + */ +long io_u_sync_complete(struct thread_data *td, struct io_u *io_u, + endio_handler *handler) +{ + struct io_completion_data icd; + + init_icd(&icd, handler, 1); + io_completed(td, io_u, &icd); + put_io_u(td, io_u); + + if (!icd.error) + return icd.bytes_done[0] + icd.bytes_done[1]; + + return -1; +} + +/* + * Called to complete min_events number of io for the async engines. + */ +long io_u_queued_complete(struct thread_data *td, int min_events, + endio_handler *handler) + +{ + struct io_completion_data icd; + struct timespec *tvp = NULL; + int ret; + + if (min_events > 0) { + ret = td_io_commit(td); + if (ret < 0) { + td_verror(td, -ret); + return ret; + } + } else { + struct timespec ts = { .tv_sec = 0, .tv_nsec = 0, }; + + tvp = &ts; + } + + ret = td_io_getevents(td, min_events, td->cur_depth, tvp); + if (ret < 0) { + td_verror(td, -ret); + return ret; + } else if (!ret) + return ret; + + init_icd(&icd, handler, ret); + ios_completed(td, &icd); + if (!icd.error) + return icd.bytes_done[0] + icd.bytes_done[1]; + + return -1; +} + +/* + * Call when io_u is really queued, to update the submission latency. + */ +void io_u_queued(struct thread_data *td, struct io_u *io_u) +{ + unsigned long slat_time; + + slat_time = mtime_since(&io_u->start_time, &io_u->issue_time); + add_slat_sample(td, io_u->ddir, slat_time); +} + +#ifdef FIO_USE_TIMEOUT +void io_u_set_timeout(struct thread_data *td) +{ + assert(td->cur_depth); + + td->timer.it_interval.tv_sec = 0; + td->timer.it_interval.tv_usec = 0; + td->timer.it_value.tv_sec = IO_U_TIMEOUT + IO_U_TIMEOUT_INC; + td->timer.it_value.tv_usec = 0; + setitimer(ITIMER_REAL, &td->timer, NULL); + fio_gettime(&td->timeout_end, NULL); +} +#else +void io_u_set_timeout(struct thread_data fio_unused *td) +{ +} +#endif + +#ifdef FIO_USE_TIMEOUT +static void io_u_timeout_handler(int fio_unused sig) +{ + struct thread_data *td, *__td; + pid_t pid = getpid(); + int i; + + log_err("fio: io_u timeout\n"); + + /* + * TLS would be nice... + */ + td = NULL; + for_each_td(__td, i) { + if (__td->pid == pid) { + td = __td; + break; + } + } + + if (!td) { + log_err("fio: io_u timeout, can't find job\n"); + exit(1); + } + + if (!td->cur_depth) { + log_err("fio: timeout without pending work?\n"); + return; + } + + log_err("fio: io_u timeout: job=%s, pid=%d\n", td->name, td->pid); + td->error = ETIMEDOUT; + exit(1); +} +#endif + +void io_u_init_timeout(void) +{ +#ifdef FIO_USE_TIMEOUT + signal(SIGALRM, io_u_timeout_handler); +#endif +}