workqueue: add a workqueue_work type

[fio.git] / workqueue.c

diff --git a/workqueue.c b/workqueue.c
index 8f6963f58aaadac55ef37d378f8c7f05c928516e..11791209b54da02502d26c06ed0d60be25d304ee 100644 (file)
--- a/workqueue.c
+++ b/workqueue.c
@@ -7,7 +7,6 @@
 #include <unistd.h>
 
 #include "fio.h"
-#include "ioengine.h"
 #include "flist.h"
 #include "workqueue.h"
 #include "lib/getrusage.h"
@@ -79,7 +78,7 @@ static struct submit_worker *get_submit_worker(struct workqueue *wq)
        return sw;
 }
 
-static int all_sw_idle(struct workqueue *wq)
+static bool all_sw_idle(struct workqueue *wq)
 {
        int i;
 
@@ -87,10 +86,10 @@ static int all_sw_idle(struct workqueue *wq)
                struct submit_worker *sw = &wq->workers[i];
 
                if (!(sw->flags & SW_F_IDLE))
-                       return 0;
+                       return false;
        }
 
-       return 1;
+       return true;
 }
 
 /*
@@ -110,44 +109,36 @@ void workqueue_flush(struct workqueue *wq)
 }
 
 /*
- * Must be serialized by caller.
+ * Must be serialized by caller. Returns true for queued, false for busy.
  */
-int workqueue_enqueue(struct workqueue *wq, struct io_u *io_u)
+bool workqueue_enqueue(struct workqueue *wq, struct workqueue_work *work)
 {
        struct submit_worker *sw;
 
        sw = get_submit_worker(wq);
        if (sw) {
-               const enum fio_ddir ddir = acct_ddir(io_u);
-               struct thread_data *parent = wq->td;
-
-               if (ddir_rw(ddir)) {
-                       parent->io_issues[ddir]++;
-                       parent->io_issue_bytes[ddir] += io_u->xfer_buflen;
-               }
-
                pthread_mutex_lock(&sw->lock);
-               flist_add_tail(&io_u->verify_list, &sw->work_list);
+               flist_add_tail(&work->list, &sw->work_list);
                sw->seq = ++wq->work_seq;
                sw->flags &= ~SW_F_IDLE;
                pthread_mutex_unlock(&sw->lock);
 
                pthread_cond_signal(&sw->cond);
-               return FIO_Q_QUEUED;
+               return true;
        }
 
-       return FIO_Q_BUSY;
+       return false;
 }
 
 static void handle_list(struct submit_worker *sw, struct flist_head *list)
 {
        struct workqueue *wq = sw->wq;
-       struct io_u *io_u;
+       struct workqueue_work *work;
 
        while (!flist_empty(list)) {
-               io_u = flist_first_entry(list, struct io_u, verify_list);
-               flist_del_init(&io_u->verify_list);
-               wq->fn(&sw->td, io_u);
+               work = flist_first_entry(list, struct workqueue_work, list);
+               flist_del_init(&work->list);
+               wq->fn(&sw->td, work);
        }
 }
 
@@ -161,6 +152,7 @@ static int init_submit_worker(struct submit_worker *sw)
        memcpy(&td->ts, &parent->ts, sizeof(td->ts));
        td->o.uid = td->o.gid = -1U;
        dup_files(td, parent);
+       td->eo = parent->eo;
        fio_options_mem_dupe(td);
 
        if (ioengine_load(td))
@@ -184,7 +176,7 @@ static int init_submit_worker(struct submit_worker *sw)
 
        fio_gettime(&td->epoch, NULL);
        fio_getrusage(&td->ru_start);
-       clear_io_state(td);
+       clear_io_state(td, 1);
 
        td_set_runstate(td, TD_RUNNING);
        td->flags |= TD_F_CHILD;
@@ -197,7 +189,7 @@ err:
        return 1;
 }
 
-#ifdef CONFIG_SFA
+#ifdef CONFIG_SFAA
 static void sum_val(uint64_t *dst, uint64_t *src)
 {
        if (*src) {
@@ -215,13 +207,32 @@ static void sum_val(uint64_t *dst, uint64_t *src)
 }
 #endif
 
+static void pthread_double_unlock(pthread_mutex_t *lock1,
+                                 pthread_mutex_t *lock2)
+{
+#ifndef CONFIG_SFAA
+       pthread_mutex_unlock(lock1);
+       pthread_mutex_unlock(lock2);
+#endif
+}
+
+static void pthread_double_lock(pthread_mutex_t *lock1, pthread_mutex_t *lock2)
+{
+#ifndef CONFIG_SFAA
+       if (lock1 < lock2) {
+               pthread_mutex_lock(lock1);
+               pthread_mutex_lock(lock2);
+       } else {
+               pthread_mutex_lock(lock2);
+               pthread_mutex_lock(lock1);
+       }
+#endif
+}
+
 static void sum_ddir(struct thread_data *dst, struct thread_data *src,
                     enum fio_ddir ddir)
 {
-#ifndef CONFIG_SFA
-       pthread_mutex_lock(&dst->io_wq.stat_lock);
-       pthread_mutex_lock(&src->io_wq.stat_lock);
-#endif
+       pthread_double_lock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
 
        sum_val(&dst->io_bytes[ddir], &src->io_bytes[ddir]);
        sum_val(&dst->io_blocks[ddir], &src->io_blocks[ddir]);
@@ -229,10 +240,7 @@ static void sum_ddir(struct thread_data *dst, struct thread_data *src,
        sum_val(&dst->this_io_bytes[ddir], &src->this_io_bytes[ddir]);
        sum_val(&dst->bytes_done[ddir], &src->bytes_done[ddir]);
 
-#ifndef CONFIG_SFA
-       pthread_mutex_unlock(&src->io_wq.stat_lock);
-       pthread_mutex_unlock(&dst->io_wq.stat_lock);
-#endif
+       pthread_double_unlock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
 }
 
 static void update_accounting(struct submit_worker *sw)
@@ -281,8 +289,12 @@ static void *worker_thread(void *data)
 
                        if (td->io_u_queued || td->cur_depth ||
                            td->io_u_in_flight) {
+                               int ret;
+
                                pthread_mutex_unlock(&sw->lock);
-                               io_u_quiesce(td);
+                               ret = io_u_quiesce(td);
+                               if (ret > 0)
+                                       td->cur_depth -= ret;
                                pthread_mutex_lock(&sw->lock);
                        }
 
@@ -341,7 +353,7 @@ static void shutdown_worker(struct submit_worker *sw, unsigned int *sum_cnt)
 
        pthread_join(sw->thread, NULL);
        (*sum_cnt)++;
-       sum_thread_stats(&parent->ts, &sw->td.ts, *sum_cnt);
+       sum_thread_stats(&parent->ts, &sw->td.ts, *sum_cnt == 1);
        free_worker(sw);
 }
 
@@ -366,7 +378,9 @@ void workqueue_exit(struct workqueue *wq)
                        sw = &wq->workers[i];
                        if (sw->flags & SW_F_ACCOUNTED)
                                continue;
+                       pthread_mutex_lock(&sw->lock);
                        sw->flags |= SW_F_ACCOUNTED;
+                       pthread_mutex_unlock(&sw->lock);
                        shutdown_worker(sw, &sum_cnt);
                        shutdown++;
                }