cconv.c lib/prio_tree.c json.c lib/zipf.c lib/axmap.c \
lib/lfsr.c gettime-thread.c helpers.c lib/flist_sort.c \
lib/hweight.c lib/getrusage.c idletime.c td_error.c \
- profiles/tiobench.c profiles/act.c
+ profiles/tiobench.c profiles/act.c io_u_queue.c
ifdef CONFIG_64BIT_LLP64
CFLAGS += -DBITS_PER_LONG=32
*/
static void cleanup_pending_aio(struct thread_data *td)
{
- struct flist_head *entry, *n;
- struct io_u *io_u;
int r;
/*
* now cancel remaining active events
*/
if (td->io_ops->cancel) {
- flist_for_each_safe(entry, n, &td->io_u_busylist) {
- io_u = flist_entry(entry, struct io_u, list);
+ struct io_u *io_u;
+ int i;
- /*
- * if the io_u isn't in flight, then that generally
- * means someone leaked an io_u. complain but fix
- * it up, so we don't stall here.
- */
- if ((io_u->flags & IO_U_F_FLIGHT) == 0) {
- log_err("fio: non-busy IO on busy list\n");
- put_io_u(td, io_u);
- } else {
+ io_u_qiter(&td->io_u_all, io_u, i) {
+ if (io_u->flags & IO_U_F_FLIGHT) {
r = td->io_ops->cancel(td, io_u);
if (!r)
put_io_u(td, io_u);
static void cleanup_io_u(struct thread_data *td)
{
- struct flist_head *entry, *n;
struct io_u *io_u;
- flist_for_each_safe(entry, n, &td->io_u_freelist) {
- io_u = flist_entry(entry, struct io_u, list);
-
- flist_del(&io_u->list);
+ while ((io_u = io_u_qpop(&td->io_u_freelist)) != NULL) {
if (td->io_ops->io_u_free)
td->io_ops->io_u_free(td, io_u);
}
free_io_mem(td);
+
+ io_u_rexit(&td->io_u_requeues);
+ io_u_qexit(&td->io_u_freelist);
+ io_u_qexit(&td->io_u_all);
}
static int init_io_u(struct thread_data *td)
struct io_u *io_u;
unsigned int max_bs, min_write;
int cl_align, i, max_units;
- int data_xfer = 1;
+ int data_xfer = 1, err;
char *p;
max_units = td->o.iodepth;
if ((td->io_ops->flags & FIO_NOIO) || !(td_read(td) || td_write(td)))
data_xfer = 0;
+ err = 0;
+ err += io_u_rinit(&td->io_u_requeues, td->o.iodepth);
+ err += io_u_qinit(&td->io_u_freelist, td->o.iodepth);
+ err += io_u_qinit(&td->io_u_all, td->o.iodepth);
+
+ if (err) {
+ log_err("fio: failed setting up IO queues\n");
+ return 1;
+ }
+
/*
* if we may later need to do address alignment, then add any
* possible adjustment here so that we don't cause a buffer
io_u = ptr;
memset(io_u, 0, sizeof(*io_u));
- INIT_FLIST_HEAD(&io_u->list);
+ INIT_FLIST_HEAD(&io_u->verify_list);
dprint(FD_MEM, "io_u alloc %p, index %u\n", io_u, i);
if (data_xfer) {
io_u->index = i;
io_u->flags = IO_U_F_FREE;
- flist_add(&io_u->list, &td->io_u_freelist);
+ io_u_qpush(&td->io_u_freelist, io_u);
+
+ /*
+ * io_u never leaves this stack, used for iteration of all
+ * io_u buffers.
+ */
+ io_u_qpush(&td->io_u_all, io_u);
if (td->io_ops->io_u_init) {
int ret = td->io_ops->io_u_init(td, io_u);
if (is_backend)
fio_server_send_start(td);
- INIT_FLIST_HEAD(&td->io_u_freelist);
- INIT_FLIST_HEAD(&td->io_u_busylist);
- INIT_FLIST_HEAD(&td->io_u_requeues);
INIT_FLIST_HEAD(&td->io_log_list);
INIT_FLIST_HEAD(&td->io_hist_list);
INIT_FLIST_HEAD(&td->verify_list);
{
struct posixaio_data *pd = td->io_ops->data;
os_aiocb_t *suspend_list[SUSPEND_ENTRIES];
- struct flist_head *entry;
struct timespec start;
int have_timeout = 0;
int suspend_entries;
+ struct io_u *io_u;
unsigned int r;
+ int i;
if (t && !fill_timespec(&start))
have_timeout = 1;
restart:
memset(suspend_list, 0, sizeof(*suspend_list));
suspend_entries = 0;
- flist_for_each(entry, &td->io_u_busylist) {
- struct io_u *io_u = flist_entry(entry, struct io_u, list);
+ io_u_qiter(&td->io_u_all, io_u, i) {
int err;
- if (io_u->seen)
+ if (io_u->seen || !(io_u->flags & IO_U_F_FLIGHT))
continue;
err = aio_error(&io_u->aiocb);
#include "server.h"
#include "stat.h"
#include "flow.h"
+#include "io_u_queue.h"
#ifdef CONFIG_SOLARISAIO
#include <sys/asynch.h>
/*
* List of free and busy io_u's
*/
- struct flist_head io_u_freelist;
- struct flist_head io_u_busylist;
- struct flist_head io_u_requeues;
+ struct io_u_ring io_u_requeues;
+ struct io_u_queue io_u_freelist;
+ struct io_u_queue io_u_all;
pthread_mutex_t io_u_lock;
pthread_cond_t free_cond;
if (io_u->flags & IO_U_F_IN_CUR_DEPTH)
td->cur_depth--;
- flist_del_init(&io_u->list);
- flist_add(&io_u->list, &td->io_u_freelist);
+ io_u_qpush(&td->io_u_freelist, io_u);
td_io_u_unlock(td);
td_io_u_free_notify(td);
}
__io_u->flags &= ~IO_U_F_FLIGHT;
if (__io_u->flags & IO_U_F_IN_CUR_DEPTH)
td->cur_depth--;
- flist_del(&__io_u->list);
- flist_add_tail(&__io_u->list, &td->io_u_requeues);
+
+ io_u_rpush(&td->io_u_requeues, __io_u);
td_io_u_unlock(td);
*io_u = NULL;
}
struct io_u *__get_io_u(struct thread_data *td)
{
- struct io_u *io_u = NULL;
+ struct io_u *io_u;
td_io_u_lock(td);
again:
- if (!flist_empty(&td->io_u_requeues))
- io_u = flist_entry(td->io_u_requeues.next, struct io_u, list);
- else if (!queue_full(td)) {
- io_u = flist_entry(td->io_u_freelist.next, struct io_u, list);
+ if (!io_u_rempty(&td->io_u_requeues))
+ io_u = io_u_rpop(&td->io_u_requeues);
+ else if (!io_u_qempty(&td->io_u_freelist))
+ io_u = io_u_qpop(&td->io_u_freelist);
+ if (io_u) {
io_u->buflen = 0;
io_u->resid = 0;
io_u->file = NULL;
io_u->error = 0;
io_u->acct_ddir = -1;
- flist_del(&io_u->list);
- flist_add_tail(&io_u->list, &td->io_u_busylist);
td->cur_depth++;
io_u->flags |= IO_U_F_IN_CUR_DEPTH;
} else if (td->o.verify_async) {
--- /dev/null
+#include <stdlib.h>
+#include "io_u_queue.h"
+
+int io_u_qinit(struct io_u_queue *q, unsigned int nr)
+{
+ q->io_us = calloc(sizeof(struct io_u *), nr);
+ if (!q->io_us)
+ return 1;
+
+ q->nr = 0;
+ return 0;
+}
+
+void io_u_qexit(struct io_u_queue *q)
+{
+ free(q->io_us);
+}
+
+int io_u_rinit(struct io_u_ring *ring, unsigned int nr)
+{
+ ring->max = nr + 1;
+ if (ring->max & (ring->max - 1)) {
+ ring->max--;
+ ring->max |= ring->max >> 1;
+ ring->max |= ring->max >> 2;
+ ring->max |= ring->max >> 4;
+ ring->max |= ring->max >> 8;
+ ring->max |= ring->max >> 16;
+ ring->max++;
+ }
+
+ ring->ring = calloc(sizeof(struct io_u *), ring->max);
+ if (!ring->ring)
+ return 1;
+
+ ring->head = ring->tail = 0;
+ return 0;
+}
+
+void io_u_rexit(struct io_u_ring *ring)
+{
+ free(ring->ring);
+}
--- /dev/null
+#ifndef FIO_IO_U_QUEUE
+#define FIO_IO_U_QUEUE
+
+#include <assert.h>
+
+struct io_u;
+
+struct io_u_queue {
+ struct io_u **io_us;
+ unsigned int nr;
+};
+
+static inline struct io_u *io_u_qpop(struct io_u_queue *q)
+{
+ if (q->nr)
+ return q->io_us[--q->nr];
+
+ return NULL;
+}
+
+static inline void io_u_qpush(struct io_u_queue *q, struct io_u *io_u)
+{
+ q->io_us[q->nr++] = io_u;
+}
+
+static inline int io_u_qempty(struct io_u_queue *q)
+{
+ return !q->nr;
+}
+
+#define io_u_qiter(q, io_u, i) \
+ for (i = 0, io_u = (q)->io_us[0]; i < (q)->nr; i++, io_u = (q)->io_us[i])
+
+int io_u_qinit(struct io_u_queue *q, unsigned int nr);
+void io_u_qexit(struct io_u_queue *q);
+
+struct io_u_ring {
+ unsigned int head;
+ unsigned int tail;
+ unsigned int max;
+ struct io_u **ring;
+};
+
+int io_u_rinit(struct io_u_ring *ring, unsigned int nr);
+void io_u_rexit(struct io_u_ring *ring);
+
+static inline void io_u_rpush(struct io_u_ring *r, struct io_u *io_u)
+{
+ if (r->head + 1 != r->tail) {
+ r->ring[r->head] = io_u;
+ r->head = (r->head + 1) & (r->max - 1);
+ return;
+ }
+
+ assert(0);
+}
+
+static inline struct io_u *io_u_rpop(struct io_u_ring *r)
+{
+ if (r->head != r->tail) {
+ struct io_u *io_u = r->ring[r->tail];
+
+ r->tail = (r->tail + 1) & (r->max - 1);
+ return io_u;
+ }
+
+ return NULL;
+}
+
+static inline int io_u_rempty(struct io_u_ring *ring)
+{
+ return ring->head == ring->tail;
+}
+
+#endif
* The io unit
*/
struct io_u {
- union {
-#ifdef CONFIG_LIBAIO
- struct iocb iocb;
-#endif
-#ifdef CONFIG_POSIXAIO
- os_aiocb_t aiocb;
-#endif
-#ifdef FIO_HAVE_SGIO
- struct sg_io_hdr hdr;
-#endif
-#ifdef CONFIG_GUASI
- guasi_req_t greq;
-#endif
-#ifdef CONFIG_SOLARISAIO
- aio_result_t resultp;
-#endif
-#ifdef FIO_HAVE_BINJECT
- struct b_user_cmd buc;
-#endif
-#ifdef CONFIG_RDMA
- struct ibv_mr *mr;
-#endif
- void *mmap_data;
- };
struct timeval start_time;
struct timeval issue_time;
*/
unsigned long buf_filled_len;
+ union {
+#ifdef CONFIG_LIBAIO
+ struct iocb iocb;
+#endif
+#ifdef CONFIG_POSIXAIO
+ os_aiocb_t aiocb;
+#endif
+#ifdef FIO_HAVE_SGIO
+ struct sg_io_hdr hdr;
+#endif
+#ifdef CONFIG_GUASI
+ guasi_req_t greq;
+#endif
+#ifdef CONFIG_SOLARISAIO
+ aio_result_t resultp;
+#endif
+#ifdef FIO_HAVE_BINJECT
+ struct b_user_cmd buc;
+#endif
+#ifdef CONFIG_RDMA
+ struct ibv_mr *mr;
+#endif
+ void *mmap_data;
+ };
+
unsigned int resid;
unsigned int error;
void *engine_data;
};
- struct flist_head list;
+ struct flist_head verify_list;
/*
* Callback for io completion
/*
* io unit handling
*/
-#define queue_full(td) flist_empty(&(td)->io_u_freelist)
+#define queue_full(td) io_u_qempty(&(td)->io_u_freelist)
extern struct io_u *__get_io_u(struct thread_data *);
extern struct io_u *get_io_u(struct thread_data *);
extern void put_io_u(struct thread_data *, struct io_u *);
td->cur_depth--;
io_u->flags &= ~IO_U_F_IN_CUR_DEPTH;
}
- flist_del(&io_u->list);
- flist_add_tail(&io_u->list, &td->verify_list);
+ flist_add_tail(&io_u->verify_list, &td->verify_list);
io_u->flags |= IO_U_F_FREE_DEF;
pthread_mutex_unlock(&td->io_u_lock);
continue;
while (!flist_empty(&list)) {
- io_u = flist_entry(list.next, struct io_u, list);
- flist_del_init(&io_u->list);
+ io_u = flist_entry(list.next, struct io_u, verify_list);
+ flist_del(&io_u->verify_list);
ret = verify_io_u(td, io_u);
put_io_u(td, io_u);