workqueue: grab sw->lock for flag manipulation

[fio.git] / workqueue.c

diff --git a/workqueue.c b/workqueue.c
index 92088bab0a66cdfa34d9a1ba3badecab0dc6ba4b..27570b2a0b20e5e7cd4ad0a9ccaacf697f6962c7 100644 (file)
--- a/workqueue.c
+++ b/workqueue.c
@@ -79,7 +79,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 +87,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;
 }
 
 /*
@@ -124,6 +124,7 @@ int workqueue_enqueue(struct workqueue *wq, struct io_u *io_u)
                if (ddir_rw(ddir)) {
                        parent->io_issues[ddir]++;
                        parent->io_issue_bytes[ddir] += io_u->xfer_buflen;
+                       parent->rate_io_issue_bytes[ddir] += io_u->xfer_buflen;
                }
 
                pthread_mutex_lock(&sw->lock);
@@ -161,6 +162,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 +186,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,6 +199,7 @@ err:
        return 1;
 }
 
+#ifdef CONFIG_SFAA
 static void sum_val(uint64_t *dst, uint64_t *src)
 {
        if (*src) {
@@ -204,15 +207,50 @@ static void sum_val(uint64_t *dst, uint64_t *src)
                *src = 0;
        }
 }
+#else
+static void sum_val(uint64_t *dst, uint64_t *src)
+{
+       if (*src) {
+               *dst += *src;
+               *src = 0;
+       }
+}
+#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)
 {
+       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]);
        sum_val(&dst->this_io_blocks[ddir], &src->this_io_blocks[ddir]);
        sum_val(&dst->this_io_bytes[ddir], &src->this_io_bytes[ddir]);
        sum_val(&dst->bytes_done[ddir], &src->bytes_done[ddir]);
+
+       pthread_double_unlock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
 }
 
 static void update_accounting(struct submit_worker *sw)
@@ -261,8 +299,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);
                        }
 
@@ -346,7 +388,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++;
                }
@@ -355,6 +399,7 @@ void workqueue_exit(struct workqueue *wq)
        free(wq->workers);
        pthread_mutex_destroy(&wq->flush_lock);
        pthread_cond_destroy(&wq->flush_cond);
+       pthread_mutex_destroy(&wq->stat_lock);
 }
 
 static int start_worker(struct workqueue *wq, unsigned int index)
@@ -393,6 +438,7 @@ int workqueue_init(struct thread_data *td, struct workqueue *wq,
        wq->next_free_worker = 0;
        pthread_cond_init(&wq->flush_cond, NULL);
        pthread_mutex_init(&wq->flush_lock, NULL);
+       pthread_mutex_init(&wq->stat_lock, NULL);
 
        wq->workers = calloc(wq->max_workers, sizeof(struct submit_worker));
 
@@ -401,13 +447,8 @@ int workqueue_init(struct thread_data *td, struct workqueue *wq,
                        break;
 
        wq->max_workers = i;
-       if (!wq->max_workers) {
-err:
-               log_err("Can't create rate workqueue\n");
-               td_verror(td, ESRCH, "workqueue_init");
-               workqueue_exit(wq);
-               return 1;
-       }
+       if (!wq->max_workers)
+               goto err;
 
        /*
         * Wait for them all to be started and initialized
@@ -437,8 +478,12 @@ err:
                pthread_mutex_unlock(&wq->flush_lock);
        } while (1);
 
-       if (error)
-               goto err;
+       if (!error)
+               return 0;
 
-       return 0;
+err:
+       log_err("Can't create rate workqueue\n");
+       td_verror(td, ESRCH, "workqueue_init");
+       workqueue_exit(wq);
+       return 1;
 }