X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=workqueue.c;h=9e6c41ff2f399172703b6e438061236670962df8;hp=6e67f3e7df4dc8ac0910a2c96ef28d39d9e04bb2;hb=465767437da1bcf99a375295799d33948d9fc5f0;hpb=2466096336bd0fbc1a94811aa338926af6baf42f diff --git a/workqueue.c b/workqueue.c index 6e67f3e7..9e6c41ff 100644 --- a/workqueue.c +++ b/workqueue.c @@ -9,14 +9,15 @@ #include "fio.h" #include "flist.h" #include "workqueue.h" +#include "smalloc.h" +#include "pshared.h" enum { SW_F_IDLE = 1 << 0, SW_F_RUNNING = 1 << 1, SW_F_EXIT = 1 << 2, - SW_F_EXITED = 1 << 3, - SW_F_ACCOUNTED = 1 << 4, - SW_F_ERROR = 1 << 5, + SW_F_ACCOUNTED = 1 << 3, + SW_F_ERROR = 1 << 4, }; static struct submit_worker *__get_submit_worker(struct workqueue *wq, @@ -84,19 +85,18 @@ static bool all_sw_idle(struct workqueue *wq) */ void workqueue_flush(struct workqueue *wq) { + pthread_mutex_lock(&wq->flush_lock); wq->wake_idle = 1; - while (!all_sw_idle(wq)) { - pthread_mutex_lock(&wq->flush_lock); + while (!all_sw_idle(wq)) pthread_cond_wait(&wq->flush_cond, &wq->flush_lock); - pthread_mutex_unlock(&wq->flush_lock); - } wq->wake_idle = 0; + pthread_mutex_unlock(&wq->flush_lock); } /* - * Must be serialized by caller. Returns true for queued, false for busy. + * Must be serialized by caller. */ void workqueue_enqueue(struct workqueue *wq, struct workqueue_work *work) { @@ -109,9 +109,9 @@ void workqueue_enqueue(struct workqueue *wq, struct workqueue_work *work) 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); + pthread_mutex_unlock(&sw->lock); } static void handle_list(struct submit_worker *sw, struct flist_head *list) @@ -130,7 +130,7 @@ static void *worker_thread(void *data) { struct submit_worker *sw = data; struct workqueue *wq = sw->wq; - unsigned int eflags = 0, ret = 0; + unsigned int ret = 0; FLIST_HEAD(local_list); sk_out_assign(sw->sk_out); @@ -158,12 +158,10 @@ static void *worker_thread(void *data) if (sw->flags & SW_F_ERROR) goto done; + pthread_mutex_lock(&sw->lock); while (1) { - pthread_mutex_lock(&sw->lock); - if (flist_empty(&sw->work_list)) { if (sw->flags & SW_F_EXIT) { - pthread_mutex_unlock(&sw->lock); break; } @@ -172,42 +170,43 @@ static void *worker_thread(void *data) workqueue_pre_sleep(sw); pthread_mutex_lock(&sw->lock); } - - /* - * We dropped and reaquired the lock, check - * state again. - */ - if (!flist_empty(&sw->work_list)) - goto handle_work; - + } + /* + * We may have dropped and reaquired the lock, check state + * again. + */ + if (flist_empty(&sw->work_list)) { if (sw->flags & SW_F_EXIT) { - pthread_mutex_unlock(&sw->lock); break; - } else if (!(sw->flags & SW_F_IDLE)) { + } + if (!(sw->flags & SW_F_IDLE)) { sw->flags |= SW_F_IDLE; wq->next_free_worker = sw->index; + pthread_mutex_unlock(&sw->lock); + pthread_mutex_lock(&wq->flush_lock); if (wq->wake_idle) pthread_cond_signal(&wq->flush_cond); + pthread_mutex_unlock(&wq->flush_lock); + pthread_mutex_lock(&sw->lock); + } + } + if (flist_empty(&sw->work_list)) { + if (sw->flags & SW_F_EXIT) { + break; } - if (wq->ops.update_acct_fn) - wq->ops.update_acct_fn(sw); - pthread_cond_wait(&sw->cond, &sw->lock); } else { -handle_work: flist_splice_init(&sw->work_list, &local_list); } pthread_mutex_unlock(&sw->lock); handle_list(sw, &local_list); + if (wq->ops.update_acct_fn) + wq->ops.update_acct_fn(sw); + pthread_mutex_lock(&sw->lock); } - - if (wq->ops.update_acct_fn) - wq->ops.update_acct_fn(sw); + pthread_mutex_unlock(&sw->lock); done: - pthread_mutex_lock(&sw->lock); - sw->flags |= (SW_F_EXITED | eflags); - pthread_mutex_unlock(&sw->lock); sk_out_drop(); return NULL; } @@ -263,7 +262,7 @@ void workqueue_exit(struct workqueue *wq) } } while (shutdown && shutdown != wq->max_workers); - free(wq->workers); + sfree(wq->workers); wq->workers = NULL; pthread_mutex_destroy(&wq->flush_lock); pthread_cond_destroy(&wq->flush_cond); @@ -277,8 +276,11 @@ static int start_worker(struct workqueue *wq, unsigned int index, int ret; INIT_FLIST_HEAD(&sw->work_list); - pthread_cond_init(&sw->cond, NULL); - pthread_mutex_init(&sw->lock, NULL); + + ret = mutex_cond_init_pshared(&sw->lock, &sw->cond); + if (ret) + return ret; + sw->wq = wq; sw->index = index; sw->sk_out = sk_out; @@ -307,17 +309,24 @@ int workqueue_init(struct thread_data *td, struct workqueue *wq, { unsigned int running; int i, error; + int ret; wq->max_workers = max_workers; wq->td = td; wq->ops = *ops; wq->work_seq = 0; 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)); + ret = mutex_cond_init_pshared(&wq->flush_lock, &wq->flush_cond); + if (ret) + goto err; + ret = mutex_init_pshared(&wq->stat_lock); + if (ret) + goto err; + + wq->workers = smalloc(wq->max_workers * sizeof(struct submit_worker)); + if (!wq->workers) + goto err; for (i = 0; i < wq->max_workers; i++) if (start_worker(wq, i, sk_out)) @@ -331,11 +340,11 @@ int workqueue_init(struct thread_data *td, struct workqueue *wq, * Wait for them all to be started and initialized */ error = 0; + pthread_mutex_lock(&wq->flush_lock); do { struct submit_worker *sw; running = 0; - pthread_mutex_lock(&wq->flush_lock); for (i = 0; i < wq->max_workers; i++) { sw = &wq->workers[i]; pthread_mutex_lock(&sw->lock); @@ -346,14 +355,12 @@ int workqueue_init(struct thread_data *td, struct workqueue *wq, pthread_mutex_unlock(&sw->lock); } - if (error || running == wq->max_workers) { - pthread_mutex_unlock(&wq->flush_lock); + if (error || running == wq->max_workers) break; - } pthread_cond_wait(&wq->flush_cond, &wq->flush_lock); - pthread_mutex_unlock(&wq->flush_lock); } while (1); + pthread_mutex_unlock(&wq->flush_lock); if (!error) return 0;