X-Git-Url: https://git.camperquake.de/gitweb.cgi?a=blobdiff_plain;f=lib%2Flibzpool%2Ftaskq.c;h=36c0ec7dfc78e8401659a984bde5fb77e2dfe1b7;hb=2959d94a0a53612cc1ca9ce9d17df26c3d69a513;hp=93acdcf8e4e37e136a8ef9c89c7d606f28595e02;hpb=172bb4bd5e4afef721dd4d2972d8680d983f144b;p=zfs.git diff --git a/lib/libzpool/taskq.c b/lib/libzpool/taskq.c index 93acdcf..36c0ec7 100644 --- a/lib/libzpool/taskq.c +++ b/lib/libzpool/taskq.c @@ -19,7 +19,7 @@ * CDDL HEADER END */ /* - * Copyright 2008 Sun Microsystems, Inc. All rights reserved. + * Copyright 2010 Sun Microsystems, Inc. All rights reserved. * Use is subject to license terms. */ @@ -42,13 +42,15 @@ struct taskq { krwlock_t tq_threadlock; kcondvar_t tq_dispatch_cv; kcondvar_t tq_wait_cv; - thread_t *tq_threadlist; + kthread_t **tq_threadlist; int tq_flags; int tq_active; int tq_nthreads; int tq_nalloc; int tq_minalloc; int tq_maxalloc; + kcondvar_t tq_maxalloc_cv; + int tq_maxalloc_wait; task_t *tq_freelist; task_t tq_task; }; @@ -57,26 +59,36 @@ static task_t * task_alloc(taskq_t *tq, int tqflags) { task_t *t; + int rv; - if ((t = tq->tq_freelist) != NULL && tq->tq_nalloc >= tq->tq_minalloc) { +again: if ((t = tq->tq_freelist) != NULL && tq->tq_nalloc >= tq->tq_minalloc) { tq->tq_freelist = t->task_next; } else { - mutex_exit(&tq->tq_lock); if (tq->tq_nalloc >= tq->tq_maxalloc) { - if (!(tqflags & KM_SLEEP)) { - mutex_enter(&tq->tq_lock); + if (!(tqflags & KM_SLEEP)) return (NULL); - } + /* * We don't want to exceed tq_maxalloc, but we can't * wait for other tasks to complete (and thus free up * task structures) without risking deadlock with * the caller. So, we just delay for one second - * to throttle the allocation rate. + * to throttle the allocation rate. If we have tasks + * complete before one second timeout expires then + * taskq_ent_free will signal us and we will + * immediately retry the allocation. */ - delay(hz); + tq->tq_maxalloc_wait++; + rv = cv_timedwait(&tq->tq_maxalloc_cv, + &tq->tq_lock, ddi_get_lbolt() + hz); + tq->tq_maxalloc_wait--; + if (rv > 0) + goto again; /* signaled */ } + mutex_exit(&tq->tq_lock); + t = kmem_alloc(sizeof (task_t), tqflags); + mutex_enter(&tq->tq_lock); if (t != NULL) tq->tq_nalloc++; @@ -96,6 +108,9 @@ task_free(taskq_t *tq, task_t *t) kmem_free(t, sizeof (task_t)); mutex_enter(&tq->tq_lock); } + + if (tq->tq_maxalloc_wait) + cv_signal(&tq->tq_maxalloc_cv); } taskqid_t @@ -114,8 +129,13 @@ taskq_dispatch(taskq_t *tq, task_func_t func, void *arg, uint_t tqflags) mutex_exit(&tq->tq_lock); return (0); } - t->task_next = &tq->tq_task; - t->task_prev = tq->tq_task.task_prev; + if (tqflags & TQ_FRONT) { + t->task_next = tq->tq_task.task_next; + t->task_prev = &tq->tq_task; + } else { + t->task_next = &tq->tq_task; + t->task_prev = tq->tq_task.task_prev; + } t->task_next->task_prev = t; t->task_prev->task_next = t; t->task_func = func; @@ -134,7 +154,7 @@ taskq_wait(taskq_t *tq) mutex_exit(&tq->tq_lock); } -static void * +static void taskq_thread(void *arg) { taskq_t *tq = arg; @@ -163,7 +183,7 @@ taskq_thread(void *arg) tq->tq_nthreads--; cv_broadcast(&tq->tq_wait_cv); mutex_exit(&tq->tq_lock); - return (NULL); + thread_exit(); } /*ARGSUSED*/ @@ -174,10 +194,24 @@ taskq_create(const char *name, int nthreads, pri_t pri, taskq_t *tq = kmem_zalloc(sizeof (taskq_t), KM_SLEEP); int t; + if (flags & TASKQ_THREADS_CPU_PCT) { + int pct; + ASSERT3S(nthreads, >=, 0); + ASSERT3S(nthreads, <=, 100); + pct = MIN(nthreads, 100); + pct = MAX(pct, 0); + + nthreads = (sysconf(_SC_NPROCESSORS_ONLN) * pct) / 100; + nthreads = MAX(nthreads, 1); /* need at least 1 thread */ + } else { + ASSERT3S(nthreads, >=, 1); + } + rw_init(&tq->tq_threadlock, NULL, RW_DEFAULT, NULL); mutex_init(&tq->tq_lock, NULL, MUTEX_DEFAULT, NULL); cv_init(&tq->tq_dispatch_cv, NULL, CV_DEFAULT, NULL); cv_init(&tq->tq_wait_cv, NULL, CV_DEFAULT, NULL); + cv_init(&tq->tq_maxalloc_cv, NULL, CV_DEFAULT, NULL); tq->tq_flags = flags | TASKQ_ACTIVE; tq->tq_active = nthreads; tq->tq_nthreads = nthreads; @@ -185,7 +219,7 @@ taskq_create(const char *name, int nthreads, pri_t pri, tq->tq_maxalloc = maxalloc; tq->tq_task.task_next = &tq->tq_task; tq->tq_task.task_prev = &tq->tq_task; - tq->tq_threadlist = kmem_alloc(nthreads * sizeof (thread_t), KM_SLEEP); + tq->tq_threadlist = kmem_alloc(nthreads*sizeof(kthread_t *), KM_SLEEP); if (flags & TASKQ_PREPOPULATE) { mutex_enter(&tq->tq_lock); @@ -195,8 +229,8 @@ taskq_create(const char *name, int nthreads, pri_t pri, } for (t = 0; t < nthreads; t++) - (void) thr_create(0, 0, taskq_thread, - tq, THR_BOUND, &tq->tq_threadlist[t]); + VERIFY((tq->tq_threadlist[t] = thread_create(NULL, 0, + taskq_thread, tq, TS_RUN, NULL, 0, 0)) != NULL); return (tq); } @@ -204,7 +238,6 @@ taskq_create(const char *name, int nthreads, pri_t pri, void taskq_destroy(taskq_t *tq) { - int t; int nthreads = tq->tq_nthreads; taskq_wait(tq); @@ -225,21 +258,19 @@ taskq_destroy(taskq_t *tq) mutex_exit(&tq->tq_lock); - for (t = 0; t < nthreads; t++) - (void) thr_join(tq->tq_threadlist[t], NULL, NULL); - - kmem_free(tq->tq_threadlist, nthreads * sizeof (thread_t)); + kmem_free(tq->tq_threadlist, nthreads * sizeof (kthread_t *)); rw_destroy(&tq->tq_threadlock); mutex_destroy(&tq->tq_lock); cv_destroy(&tq->tq_dispatch_cv); cv_destroy(&tq->tq_wait_cv); + cv_destroy(&tq->tq_maxalloc_cv); kmem_free(tq, sizeof (taskq_t)); } int -taskq_member(taskq_t *tq, void *t) +taskq_member(taskq_t *tq, kthread_t *t) { int i; @@ -247,7 +278,7 @@ taskq_member(taskq_t *tq, void *t) return (1); for (i = 0; i < tq->tq_nthreads; i++) - if (tq->tq_threadlist[i] == (thread_t)(uintptr_t)t) + if (tq->tq_threadlist[i] == t) return (1); return (0); @@ -259,3 +290,10 @@ system_taskq_init(void) system_taskq = taskq_create("system_taskq", 64, minclsyspri, 4, 512, TASKQ_DYNAMIC | TASKQ_PREPOPULATE); } + +void +system_taskq_fini(void) +{ + taskq_destroy(system_taskq); + system_taskq = NULL; /* defensive */ +}