kern_threadpool.c revision 1.3.2.2
1 1.3.2.2 pgoyette /* $NetBSD: kern_threadpool.c,v 1.3.2.2 2018/12/26 14:02:04 pgoyette Exp $ */ 2 1.3.2.2 pgoyette 3 1.3.2.2 pgoyette /*- 4 1.3.2.2 pgoyette * Copyright (c) 2014, 2018 The NetBSD Foundation, Inc. 5 1.3.2.2 pgoyette * All rights reserved. 6 1.3.2.2 pgoyette * 7 1.3.2.2 pgoyette * This code is derived from software contributed to The NetBSD Foundation 8 1.3.2.2 pgoyette * by Taylor R. Campbell and Jason R. Thorpe. 9 1.3.2.2 pgoyette * 10 1.3.2.2 pgoyette * Redistribution and use in source and binary forms, with or without 11 1.3.2.2 pgoyette * modification, are permitted provided that the following conditions 12 1.3.2.2 pgoyette * are met: 13 1.3.2.2 pgoyette * 1. Redistributions of source code must retain the above copyright 14 1.3.2.2 pgoyette * notice, this list of conditions and the following disclaimer. 15 1.3.2.2 pgoyette * 2. Redistributions in binary form must reproduce the above copyright 16 1.3.2.2 pgoyette * notice, this list of conditions and the following disclaimer in the 17 1.3.2.2 pgoyette * documentation and/or other materials provided with the distribution. 18 1.3.2.2 pgoyette * 19 1.3.2.2 pgoyette * THIS SOFTWARE IS PROVIDED BY THE NETBSD FOUNDATION, INC. AND CONTRIBUTORS 20 1.3.2.2 pgoyette * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED 21 1.3.2.2 pgoyette * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR 22 1.3.2.2 pgoyette * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE FOUNDATION OR CONTRIBUTORS 23 1.3.2.2 pgoyette * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 24 1.3.2.2 pgoyette * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 25 1.3.2.2 pgoyette * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 26 1.3.2.2 pgoyette * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 27 1.3.2.2 pgoyette * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 28 1.3.2.2 pgoyette * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 29 1.3.2.2 pgoyette * POSSIBILITY OF SUCH DAMAGE. 30 1.3.2.2 pgoyette */ 31 1.3.2.2 pgoyette 32 1.3.2.2 pgoyette /* 33 1.3.2.2 pgoyette * Thread pools. 34 1.3.2.2 pgoyette * 35 1.3.2.2 pgoyette * A thread pool is a collection of worker threads idle or running 36 1.3.2.2 pgoyette * jobs, together with an overseer thread that does not run jobs but 37 1.3.2.2 pgoyette * can be given jobs to assign to a worker thread. Scheduling a job in 38 1.3.2.2 pgoyette * a thread pool does not allocate or even sleep at all, except perhaps 39 1.3.2.2 pgoyette * on an adaptive lock, unlike kthread_create. Jobs reuse threads, so 40 1.3.2.2 pgoyette * they do not incur the expense of creating and destroying kthreads 41 1.3.2.2 pgoyette * unless there is not much work to be done. 42 1.3.2.2 pgoyette * 43 1.3.2.2 pgoyette * A per-CPU thread pool (threadpool_percpu) is a collection of thread 44 1.3.2.2 pgoyette * pools, one per CPU bound to that CPU. For each priority level in 45 1.3.2.2 pgoyette * use, there is one shared unbound thread pool (i.e., pool of threads 46 1.3.2.2 pgoyette * not bound to any CPU) and one shared per-CPU thread pool. 47 1.3.2.2 pgoyette * 48 1.3.2.2 pgoyette * To use the unbound thread pool at priority pri, call 49 1.3.2.2 pgoyette * threadpool_get(&pool, pri). When you're done, call 50 1.3.2.2 pgoyette * threadpool_put(pool, pri). 51 1.3.2.2 pgoyette * 52 1.3.2.2 pgoyette * To use the per-CPU thread pools at priority pri, call 53 1.3.2.2 pgoyette * threadpool_percpu_get(&pool_percpu, pri), and then use the thread 54 1.3.2.2 pgoyette * pool returned by threadpool_percpu_ref(pool_percpu) for the current 55 1.3.2.2 pgoyette * CPU, or by threadpool_percpu_ref_remote(pool_percpu, ci) for another 56 1.3.2.2 pgoyette * CPU. When you're done, call threadpool_percpu_put(pool_percpu, 57 1.3.2.2 pgoyette * pri). 58 1.3.2.2 pgoyette * 59 1.3.2.2 pgoyette * +--MACHINE-----------------------------------------------+ 60 1.3.2.2 pgoyette * | +--CPU 0-------+ +--CPU 1-------+ +--CPU n-------+ | 61 1.3.2.2 pgoyette * | | <overseer 0> | | <overseer 1> | ... | <overseer n> | | 62 1.3.2.2 pgoyette * | | <idle 0a> | | <running 1a> | ... | <idle na> | | 63 1.3.2.2 pgoyette * | | <running 0b> | | <running 1b> | ... | <idle nb> | | 64 1.3.2.2 pgoyette * | | . | | . | ... | . | | 65 1.3.2.2 pgoyette * | | . | | . | ... | . | | 66 1.3.2.2 pgoyette * | | . | | . | ... | . | | 67 1.3.2.2 pgoyette * | +--------------+ +--------------+ +--------------+ | 68 1.3.2.2 pgoyette * | +--unbound---------+ | 69 1.3.2.2 pgoyette * | | <overseer n+1> | | 70 1.3.2.2 pgoyette * | | <idle (n+1)a> | | 71 1.3.2.2 pgoyette * | | <running (n+1)b> | | 72 1.3.2.2 pgoyette * | +------------------+ | 73 1.3.2.2 pgoyette * +--------------------------------------------------------+ 74 1.3.2.2 pgoyette * 75 1.3.2.2 pgoyette * XXX Why one overseer per CPU? I did that originally to avoid 76 1.3.2.2 pgoyette * touching remote CPUs' memory when scheduling a job, but that still 77 1.3.2.2 pgoyette * requires interprocessor synchronization. Perhaps we could get by 78 1.3.2.2 pgoyette * with a single overseer thread, at the expense of another pointer in 79 1.3.2.2 pgoyette * struct threadpool_job_impl to identify the CPU on which it must run 80 1.3.2.2 pgoyette * in order for the overseer to schedule it correctly. 81 1.3.2.2 pgoyette */ 82 1.3.2.2 pgoyette 83 1.3.2.2 pgoyette #include <sys/cdefs.h> 84 1.3.2.2 pgoyette __KERNEL_RCSID(0, "$NetBSD: kern_threadpool.c,v 1.3.2.2 2018/12/26 14:02:04 pgoyette Exp $"); 85 1.3.2.2 pgoyette 86 1.3.2.2 pgoyette #include <sys/types.h> 87 1.3.2.2 pgoyette #include <sys/param.h> 88 1.3.2.2 pgoyette #include <sys/atomic.h> 89 1.3.2.2 pgoyette #include <sys/condvar.h> 90 1.3.2.2 pgoyette #include <sys/cpu.h> 91 1.3.2.2 pgoyette #include <sys/kernel.h> 92 1.3.2.2 pgoyette #include <sys/kmem.h> 93 1.3.2.2 pgoyette #include <sys/kthread.h> 94 1.3.2.2 pgoyette #include <sys/mutex.h> 95 1.3.2.2 pgoyette #include <sys/once.h> 96 1.3.2.2 pgoyette #include <sys/percpu.h> 97 1.3.2.2 pgoyette #include <sys/pool.h> 98 1.3.2.2 pgoyette #include <sys/proc.h> 99 1.3.2.2 pgoyette #include <sys/queue.h> 100 1.3.2.2 pgoyette #include <sys/systm.h> 101 1.3.2.2 pgoyette #include <sys/threadpool.h> 102 1.3.2.2 pgoyette 103 1.3.2.2 pgoyette static ONCE_DECL(threadpool_init_once) 104 1.3.2.2 pgoyette 105 1.3.2.2 pgoyette #define THREADPOOL_INIT() \ 106 1.3.2.2 pgoyette do { \ 107 1.3.2.2 pgoyette int threadpool_init_error __diagused = \ 108 1.3.2.2 pgoyette RUN_ONCE(&threadpool_init_once, threadpools_init); \ 109 1.3.2.2 pgoyette KASSERT(threadpool_init_error == 0); \ 110 1.3.2.2 pgoyette } while (/*CONSTCOND*/0) 111 1.3.2.2 pgoyette 112 1.3.2.2 pgoyette 113 1.3.2.2 pgoyette /* Data structures */ 114 1.3.2.2 pgoyette 115 1.3.2.2 pgoyette TAILQ_HEAD(job_head, threadpool_job_impl); 116 1.3.2.2 pgoyette TAILQ_HEAD(thread_head, threadpool_thread); 117 1.3.2.2 pgoyette 118 1.3.2.2 pgoyette typedef struct threadpool_job_impl { 119 1.3.2.2 pgoyette kmutex_t *job_lock; /* 1 */ 120 1.3.2.2 pgoyette struct threadpool_thread *job_thread; /* 1 */ 121 1.3.2.2 pgoyette TAILQ_ENTRY(threadpool_job_impl) job_entry; /* 2 */ 122 1.3.2.2 pgoyette volatile unsigned int job_refcnt; /* 1 */ 123 1.3.2.2 pgoyette /* implicit pad on _LP64 */ 124 1.3.2.2 pgoyette kcondvar_t job_cv; /* 3 */ 125 1.3.2.2 pgoyette threadpool_job_fn_t job_fn; /* 1 */ 126 1.3.2.2 pgoyette /* ILP32 / LP64 */ 127 1.3.2.2 pgoyette char job_name[MAXCOMLEN]; /* 4 / 2 */ 128 1.3.2.2 pgoyette } threadpool_job_impl_t; 129 1.3.2.2 pgoyette 130 1.3.2.2 pgoyette CTASSERT(sizeof(threadpool_job_impl_t) <= sizeof(threadpool_job_t)); 131 1.3.2.2 pgoyette #define THREADPOOL_JOB_TO_IMPL(j) ((threadpool_job_impl_t *)(j)) 132 1.3.2.2 pgoyette #define THREADPOOL_IMPL_TO_JOB(j) ((threadpool_job_t *)(j)) 133 1.3.2.2 pgoyette 134 1.3.2.2 pgoyette struct threadpool_thread { 135 1.3.2.2 pgoyette struct lwp *tpt_lwp; 136 1.3.2.2 pgoyette threadpool_t *tpt_pool; 137 1.3.2.2 pgoyette threadpool_job_impl_t *tpt_job; 138 1.3.2.2 pgoyette kcondvar_t tpt_cv; 139 1.3.2.2 pgoyette TAILQ_ENTRY(threadpool_thread) tpt_entry; 140 1.3.2.2 pgoyette }; 141 1.3.2.2 pgoyette 142 1.3.2.2 pgoyette struct threadpool { 143 1.3.2.2 pgoyette kmutex_t tp_lock; 144 1.3.2.2 pgoyette struct threadpool_thread tp_overseer; 145 1.3.2.2 pgoyette struct job_head tp_jobs; 146 1.3.2.2 pgoyette struct thread_head tp_idle_threads; 147 1.3.2.2 pgoyette unsigned int tp_refcnt; 148 1.3.2.2 pgoyette int tp_flags; 149 1.3.2.2 pgoyette #define THREADPOOL_DYING 0x01 150 1.3.2.2 pgoyette struct cpu_info *tp_cpu; 151 1.3.2.2 pgoyette pri_t tp_pri; 152 1.3.2.2 pgoyette }; 153 1.3.2.2 pgoyette 154 1.3.2.2 pgoyette static int threadpool_hold(threadpool_t *); 155 1.3.2.2 pgoyette static void threadpool_rele(threadpool_t *); 156 1.3.2.2 pgoyette 157 1.3.2.2 pgoyette static int threadpool_percpu_create(threadpool_percpu_t **, pri_t); 158 1.3.2.2 pgoyette static void threadpool_percpu_destroy(threadpool_percpu_t *); 159 1.3.2.2 pgoyette 160 1.3.2.2 pgoyette static void threadpool_job_dead(threadpool_job_t *); 161 1.3.2.2 pgoyette 162 1.3.2.2 pgoyette static int threadpool_job_hold(threadpool_job_impl_t *); 163 1.3.2.2 pgoyette static void threadpool_job_rele(threadpool_job_impl_t *); 164 1.3.2.2 pgoyette 165 1.3.2.2 pgoyette static void threadpool_overseer_thread(void *) __dead; 166 1.3.2.2 pgoyette static void threadpool_thread(void *) __dead; 167 1.3.2.2 pgoyette 168 1.3.2.2 pgoyette static pool_cache_t threadpool_thread_pc __read_mostly; 169 1.3.2.2 pgoyette 170 1.3.2.2 pgoyette static kmutex_t threadpools_lock __cacheline_aligned; 171 1.3.2.2 pgoyette 172 1.3.2.2 pgoyette /* Idle out threads after 30 seconds */ 173 1.3.2.2 pgoyette #define THREADPOOL_IDLE_TICKS mstohz(30 * 1000) 174 1.3.2.2 pgoyette 175 1.3.2.2 pgoyette struct threadpool_unbound { 176 1.3.2.2 pgoyette /* must be first; see threadpool_create() */ 177 1.3.2.2 pgoyette struct threadpool tpu_pool; 178 1.3.2.2 pgoyette 179 1.3.2.2 pgoyette /* protected by threadpools_lock */ 180 1.3.2.2 pgoyette LIST_ENTRY(threadpool_unbound) tpu_link; 181 1.3.2.2 pgoyette unsigned int tpu_refcnt; 182 1.3.2.2 pgoyette }; 183 1.3.2.2 pgoyette 184 1.3.2.2 pgoyette static LIST_HEAD(, threadpool_unbound) unbound_threadpools; 185 1.3.2.2 pgoyette 186 1.3.2.2 pgoyette static struct threadpool_unbound * 187 1.3.2.2 pgoyette threadpool_lookup_unbound(pri_t pri) 188 1.3.2.2 pgoyette { 189 1.3.2.2 pgoyette struct threadpool_unbound *tpu; 190 1.3.2.2 pgoyette 191 1.3.2.2 pgoyette LIST_FOREACH(tpu, &unbound_threadpools, tpu_link) { 192 1.3.2.2 pgoyette if (tpu->tpu_pool.tp_pri == pri) 193 1.3.2.2 pgoyette return tpu; 194 1.3.2.2 pgoyette } 195 1.3.2.2 pgoyette return NULL; 196 1.3.2.2 pgoyette } 197 1.3.2.2 pgoyette 198 1.3.2.2 pgoyette static void 199 1.3.2.2 pgoyette threadpool_insert_unbound(struct threadpool_unbound *tpu) 200 1.3.2.2 pgoyette { 201 1.3.2.2 pgoyette KASSERT(threadpool_lookup_unbound(tpu->tpu_pool.tp_pri) == NULL); 202 1.3.2.2 pgoyette LIST_INSERT_HEAD(&unbound_threadpools, tpu, tpu_link); 203 1.3.2.2 pgoyette } 204 1.3.2.2 pgoyette 205 1.3.2.2 pgoyette static void 206 1.3.2.2 pgoyette threadpool_remove_unbound(struct threadpool_unbound *tpu) 207 1.3.2.2 pgoyette { 208 1.3.2.2 pgoyette KASSERT(threadpool_lookup_unbound(tpu->tpu_pool.tp_pri) == tpu); 209 1.3.2.2 pgoyette LIST_REMOVE(tpu, tpu_link); 210 1.3.2.2 pgoyette } 211 1.3.2.2 pgoyette 212 1.3.2.2 pgoyette struct threadpool_percpu { 213 1.3.2.2 pgoyette percpu_t * tpp_percpu; 214 1.3.2.2 pgoyette pri_t tpp_pri; 215 1.3.2.2 pgoyette 216 1.3.2.2 pgoyette /* protected by threadpools_lock */ 217 1.3.2.2 pgoyette LIST_ENTRY(threadpool_percpu) tpp_link; 218 1.3.2.2 pgoyette unsigned int tpp_refcnt; 219 1.3.2.2 pgoyette }; 220 1.3.2.2 pgoyette 221 1.3.2.2 pgoyette static LIST_HEAD(, threadpool_percpu) percpu_threadpools; 222 1.3.2.2 pgoyette 223 1.3.2.2 pgoyette static threadpool_percpu_t * 224 1.3.2.2 pgoyette threadpool_lookup_percpu(pri_t pri) 225 1.3.2.2 pgoyette { 226 1.3.2.2 pgoyette threadpool_percpu_t *tpp; 227 1.3.2.2 pgoyette 228 1.3.2.2 pgoyette LIST_FOREACH(tpp, &percpu_threadpools, tpp_link) { 229 1.3.2.2 pgoyette if (tpp->tpp_pri == pri) 230 1.3.2.2 pgoyette return tpp; 231 1.3.2.2 pgoyette } 232 1.3.2.2 pgoyette return NULL; 233 1.3.2.2 pgoyette } 234 1.3.2.2 pgoyette 235 1.3.2.2 pgoyette static void 236 1.3.2.2 pgoyette threadpool_insert_percpu(threadpool_percpu_t *tpp) 237 1.3.2.2 pgoyette { 238 1.3.2.2 pgoyette KASSERT(threadpool_lookup_percpu(tpp->tpp_pri) == NULL); 239 1.3.2.2 pgoyette LIST_INSERT_HEAD(&percpu_threadpools, tpp, tpp_link); 240 1.3.2.2 pgoyette } 241 1.3.2.2 pgoyette 242 1.3.2.2 pgoyette static void 243 1.3.2.2 pgoyette threadpool_remove_percpu(threadpool_percpu_t *tpp) 244 1.3.2.2 pgoyette { 245 1.3.2.2 pgoyette KASSERT(threadpool_lookup_percpu(tpp->tpp_pri) == tpp); 246 1.3.2.2 pgoyette LIST_REMOVE(tpp, tpp_link); 247 1.3.2.2 pgoyette } 248 1.3.2.2 pgoyette 249 1.3.2.2 pgoyette #ifdef THREADPOOL_VERBOSE 250 1.3.2.2 pgoyette #define TP_LOG(x) printf x 251 1.3.2.2 pgoyette #else 252 1.3.2.2 pgoyette #define TP_LOG(x) /* nothing */ 253 1.3.2.2 pgoyette #endif /* THREADPOOL_VERBOSE */ 254 1.3.2.2 pgoyette 255 1.3.2.2 pgoyette static int 256 1.3.2.2 pgoyette threadpools_init(void) 257 1.3.2.2 pgoyette { 258 1.3.2.2 pgoyette 259 1.3.2.2 pgoyette threadpool_thread_pc = 260 1.3.2.2 pgoyette pool_cache_init(sizeof(struct threadpool_thread), 0, 0, 0, 261 1.3.2.2 pgoyette "thplthrd", NULL, IPL_NONE, NULL, NULL, NULL); 262 1.3.2.2 pgoyette 263 1.3.2.2 pgoyette LIST_INIT(&unbound_threadpools); 264 1.3.2.2 pgoyette LIST_INIT(&percpu_threadpools); 265 1.3.2.2 pgoyette mutex_init(&threadpools_lock, MUTEX_DEFAULT, IPL_NONE); 266 1.3.2.2 pgoyette 267 1.3.2.2 pgoyette TP_LOG(("%s: sizeof(threadpool_job) = %zu\n", 268 1.3.2.2 pgoyette __func__, sizeof(threadpool_job_t))); 269 1.3.2.2 pgoyette 270 1.3.2.2 pgoyette return 0; 271 1.3.2.2 pgoyette } 272 1.3.2.2 pgoyette 273 1.3.2.2 pgoyette /* Thread pool creation */ 274 1.3.2.2 pgoyette 275 1.3.2.2 pgoyette static bool 276 1.3.2.2 pgoyette threadpool_pri_is_valid(pri_t pri) 277 1.3.2.2 pgoyette { 278 1.3.2.2 pgoyette return (pri == PRI_NONE || (pri >= PRI_USER && pri < PRI_COUNT)); 279 1.3.2.2 pgoyette } 280 1.3.2.2 pgoyette 281 1.3.2.2 pgoyette static int 282 1.3.2.2 pgoyette threadpool_create(threadpool_t **poolp, struct cpu_info *ci, pri_t pri, 283 1.3.2.2 pgoyette size_t size) 284 1.3.2.2 pgoyette { 285 1.3.2.2 pgoyette threadpool_t *const pool = kmem_zalloc(size, KM_SLEEP); 286 1.3.2.2 pgoyette struct lwp *lwp; 287 1.3.2.2 pgoyette int ktflags; 288 1.3.2.2 pgoyette int error; 289 1.3.2.2 pgoyette 290 1.3.2.2 pgoyette KASSERT(threadpool_pri_is_valid(pri)); 291 1.3.2.2 pgoyette 292 1.3.2.2 pgoyette mutex_init(&pool->tp_lock, MUTEX_DEFAULT, IPL_VM); 293 1.3.2.2 pgoyette /* XXX overseer */ 294 1.3.2.2 pgoyette TAILQ_INIT(&pool->tp_jobs); 295 1.3.2.2 pgoyette TAILQ_INIT(&pool->tp_idle_threads); 296 1.3.2.2 pgoyette pool->tp_refcnt = 0; 297 1.3.2.2 pgoyette pool->tp_flags = 0; 298 1.3.2.2 pgoyette pool->tp_cpu = ci; 299 1.3.2.2 pgoyette pool->tp_pri = pri; 300 1.3.2.2 pgoyette 301 1.3.2.2 pgoyette error = threadpool_hold(pool); 302 1.3.2.2 pgoyette KASSERT(error == 0); 303 1.3.2.2 pgoyette pool->tp_overseer.tpt_lwp = NULL; 304 1.3.2.2 pgoyette pool->tp_overseer.tpt_pool = pool; 305 1.3.2.2 pgoyette pool->tp_overseer.tpt_job = NULL; 306 1.3.2.2 pgoyette cv_init(&pool->tp_overseer.tpt_cv, "poolover"); 307 1.3.2.2 pgoyette 308 1.3.2.2 pgoyette ktflags = 0; 309 1.3.2.2 pgoyette ktflags |= KTHREAD_MPSAFE; 310 1.3.2.2 pgoyette if (pri < PRI_KERNEL) 311 1.3.2.2 pgoyette ktflags |= KTHREAD_TS; 312 1.3.2.2 pgoyette error = kthread_create(pri, ktflags, ci, &threadpool_overseer_thread, 313 1.3.2.2 pgoyette &pool->tp_overseer, &lwp, 314 1.3.2.2 pgoyette "pooloverseer/%d@%d", (ci ? cpu_index(ci) : -1), (int)pri); 315 1.3.2.2 pgoyette if (error) 316 1.3.2.2 pgoyette goto fail0; 317 1.3.2.2 pgoyette 318 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 319 1.3.2.2 pgoyette pool->tp_overseer.tpt_lwp = lwp; 320 1.3.2.2 pgoyette cv_broadcast(&pool->tp_overseer.tpt_cv); 321 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 322 1.3.2.2 pgoyette 323 1.3.2.2 pgoyette *poolp = pool; 324 1.3.2.2 pgoyette return 0; 325 1.3.2.2 pgoyette 326 1.3.2.2 pgoyette fail0: KASSERT(error); 327 1.3.2.2 pgoyette KASSERT(pool->tp_overseer.tpt_job == NULL); 328 1.3.2.2 pgoyette KASSERT(pool->tp_overseer.tpt_pool == pool); 329 1.3.2.2 pgoyette KASSERT(pool->tp_flags == 0); 330 1.3.2.2 pgoyette KASSERT(pool->tp_refcnt == 0); 331 1.3.2.2 pgoyette KASSERT(TAILQ_EMPTY(&pool->tp_idle_threads)); 332 1.3.2.2 pgoyette KASSERT(TAILQ_EMPTY(&pool->tp_jobs)); 333 1.3.2.2 pgoyette KASSERT(!cv_has_waiters(&pool->tp_overseer.tpt_cv)); 334 1.3.2.2 pgoyette cv_destroy(&pool->tp_overseer.tpt_cv); 335 1.3.2.2 pgoyette mutex_destroy(&pool->tp_lock); 336 1.3.2.2 pgoyette kmem_free(pool, size); 337 1.3.2.2 pgoyette return error; 338 1.3.2.2 pgoyette } 339 1.3.2.2 pgoyette 340 1.3.2.2 pgoyette /* Thread pool destruction */ 341 1.3.2.2 pgoyette 342 1.3.2.2 pgoyette static void 343 1.3.2.2 pgoyette threadpool_destroy(threadpool_t *pool, size_t size) 344 1.3.2.2 pgoyette { 345 1.3.2.2 pgoyette struct threadpool_thread *thread; 346 1.3.2.2 pgoyette 347 1.3.2.2 pgoyette /* Mark the pool dying and wait for threads to commit suicide. */ 348 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 349 1.3.2.2 pgoyette KASSERT(TAILQ_EMPTY(&pool->tp_jobs)); 350 1.3.2.2 pgoyette pool->tp_flags |= THREADPOOL_DYING; 351 1.3.2.2 pgoyette cv_broadcast(&pool->tp_overseer.tpt_cv); 352 1.3.2.2 pgoyette TAILQ_FOREACH(thread, &pool->tp_idle_threads, tpt_entry) 353 1.3.2.2 pgoyette cv_broadcast(&thread->tpt_cv); 354 1.3.2.2 pgoyette while (0 < pool->tp_refcnt) { 355 1.3.2.2 pgoyette TP_LOG(("%s: draining %u references...\n", __func__, 356 1.3.2.2 pgoyette pool->tp_refcnt)); 357 1.3.2.2 pgoyette cv_wait(&pool->tp_overseer.tpt_cv, &pool->tp_lock); 358 1.3.2.2 pgoyette } 359 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 360 1.3.2.2 pgoyette 361 1.3.2.2 pgoyette KASSERT(pool->tp_overseer.tpt_job == NULL); 362 1.3.2.2 pgoyette KASSERT(pool->tp_overseer.tpt_pool == pool); 363 1.3.2.2 pgoyette KASSERT(pool->tp_flags == THREADPOOL_DYING); 364 1.3.2.2 pgoyette KASSERT(pool->tp_refcnt == 0); 365 1.3.2.2 pgoyette KASSERT(TAILQ_EMPTY(&pool->tp_idle_threads)); 366 1.3.2.2 pgoyette KASSERT(TAILQ_EMPTY(&pool->tp_jobs)); 367 1.3.2.2 pgoyette KASSERT(!cv_has_waiters(&pool->tp_overseer.tpt_cv)); 368 1.3.2.2 pgoyette cv_destroy(&pool->tp_overseer.tpt_cv); 369 1.3.2.2 pgoyette mutex_destroy(&pool->tp_lock); 370 1.3.2.2 pgoyette kmem_free(pool, size); 371 1.3.2.2 pgoyette } 372 1.3.2.2 pgoyette 373 1.3.2.2 pgoyette static int 374 1.3.2.2 pgoyette threadpool_hold(threadpool_t *pool) 375 1.3.2.2 pgoyette { 376 1.3.2.2 pgoyette unsigned int refcnt; 377 1.3.2.2 pgoyette 378 1.3.2.2 pgoyette do { 379 1.3.2.2 pgoyette refcnt = pool->tp_refcnt; 380 1.3.2.2 pgoyette if (refcnt == UINT_MAX) 381 1.3.2.2 pgoyette return EBUSY; 382 1.3.2.2 pgoyette } while (atomic_cas_uint(&pool->tp_refcnt, refcnt, (refcnt + 1)) 383 1.3.2.2 pgoyette != refcnt); 384 1.3.2.2 pgoyette 385 1.3.2.2 pgoyette return 0; 386 1.3.2.2 pgoyette } 387 1.3.2.2 pgoyette 388 1.3.2.2 pgoyette static void 389 1.3.2.2 pgoyette threadpool_rele(threadpool_t *pool) 390 1.3.2.2 pgoyette { 391 1.3.2.2 pgoyette unsigned int refcnt; 392 1.3.2.2 pgoyette 393 1.3.2.2 pgoyette do { 394 1.3.2.2 pgoyette refcnt = pool->tp_refcnt; 395 1.3.2.2 pgoyette KASSERT(0 < refcnt); 396 1.3.2.2 pgoyette if (refcnt == 1) { 397 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 398 1.3.2.2 pgoyette refcnt = atomic_dec_uint_nv(&pool->tp_refcnt); 399 1.3.2.2 pgoyette KASSERT(refcnt != UINT_MAX); 400 1.3.2.2 pgoyette if (refcnt == 0) 401 1.3.2.2 pgoyette cv_broadcast(&pool->tp_overseer.tpt_cv); 402 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 403 1.3.2.2 pgoyette return; 404 1.3.2.2 pgoyette } 405 1.3.2.2 pgoyette } while (atomic_cas_uint(&pool->tp_refcnt, refcnt, (refcnt - 1)) 406 1.3.2.2 pgoyette != refcnt); 407 1.3.2.2 pgoyette } 408 1.3.2.2 pgoyette 409 1.3.2.2 pgoyette /* Unbound thread pools */ 410 1.3.2.2 pgoyette 411 1.3.2.2 pgoyette int 412 1.3.2.2 pgoyette threadpool_get(threadpool_t **poolp, pri_t pri) 413 1.3.2.2 pgoyette { 414 1.3.2.2 pgoyette struct threadpool_unbound *tpu, *tmp = NULL; 415 1.3.2.2 pgoyette int error; 416 1.3.2.2 pgoyette 417 1.3.2.2 pgoyette THREADPOOL_INIT(); 418 1.3.2.2 pgoyette 419 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 420 1.3.2.2 pgoyette 421 1.3.2.2 pgoyette if (! threadpool_pri_is_valid(pri)) 422 1.3.2.2 pgoyette return EINVAL; 423 1.3.2.2 pgoyette 424 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 425 1.3.2.2 pgoyette tpu = threadpool_lookup_unbound(pri); 426 1.3.2.2 pgoyette if (tpu == NULL) { 427 1.3.2.2 pgoyette threadpool_t *new_pool; 428 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 429 1.3.2.2 pgoyette TP_LOG(("%s: No pool for pri=%d, creating one.\n", 430 1.3.2.2 pgoyette __func__, (int)pri)); 431 1.3.2.2 pgoyette error = threadpool_create(&new_pool, NULL, pri, sizeof(*tpu)); 432 1.3.2.2 pgoyette if (error) 433 1.3.2.2 pgoyette return error; 434 1.3.2.2 pgoyette KASSERT(new_pool != NULL); 435 1.3.2.2 pgoyette tmp = container_of(new_pool, struct threadpool_unbound, 436 1.3.2.2 pgoyette tpu_pool); 437 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 438 1.3.2.2 pgoyette tpu = threadpool_lookup_unbound(pri); 439 1.3.2.2 pgoyette if (tpu == NULL) { 440 1.3.2.2 pgoyette TP_LOG(("%s: Won the creation race for pri=%d.\n", 441 1.3.2.2 pgoyette __func__, (int)pri)); 442 1.3.2.2 pgoyette tpu = tmp; 443 1.3.2.2 pgoyette tmp = NULL; 444 1.3.2.2 pgoyette threadpool_insert_unbound(tpu); 445 1.3.2.2 pgoyette } 446 1.3.2.2 pgoyette } 447 1.3.2.2 pgoyette KASSERT(tpu != NULL); 448 1.3.2.2 pgoyette if (tpu->tpu_refcnt == UINT_MAX) { 449 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 450 1.3.2.2 pgoyette if (tmp != NULL) 451 1.3.2.2 pgoyette threadpool_destroy(&tmp->tpu_pool, sizeof(*tpu)); 452 1.3.2.2 pgoyette return EBUSY; 453 1.3.2.2 pgoyette } 454 1.3.2.2 pgoyette tpu->tpu_refcnt++; 455 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 456 1.3.2.2 pgoyette 457 1.3.2.2 pgoyette if (tmp != NULL) 458 1.3.2.2 pgoyette threadpool_destroy((threadpool_t *)tmp, sizeof(*tpu)); 459 1.3.2.2 pgoyette KASSERT(tpu != NULL); 460 1.3.2.2 pgoyette *poolp = &tpu->tpu_pool; 461 1.3.2.2 pgoyette return 0; 462 1.3.2.2 pgoyette } 463 1.3.2.2 pgoyette 464 1.3.2.2 pgoyette void 465 1.3.2.2 pgoyette threadpool_put(threadpool_t *pool, pri_t pri) 466 1.3.2.2 pgoyette { 467 1.3.2.2 pgoyette struct threadpool_unbound *tpu = 468 1.3.2.2 pgoyette container_of(pool, struct threadpool_unbound, tpu_pool); 469 1.3.2.2 pgoyette 470 1.3.2.2 pgoyette THREADPOOL_INIT(); 471 1.3.2.2 pgoyette 472 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 473 1.3.2.2 pgoyette 474 1.3.2.2 pgoyette KASSERT(threadpool_pri_is_valid(pri)); 475 1.3.2.2 pgoyette 476 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 477 1.3.2.2 pgoyette KASSERT(tpu == threadpool_lookup_unbound(pri)); 478 1.3.2.2 pgoyette KASSERT(0 < tpu->tpu_refcnt); 479 1.3.2.2 pgoyette if (--tpu->tpu_refcnt == 0) { 480 1.3.2.2 pgoyette TP_LOG(("%s: Last reference for pri=%d, destroying pool.\n", 481 1.3.2.2 pgoyette __func__, (int)pri)); 482 1.3.2.2 pgoyette threadpool_remove_unbound(tpu); 483 1.3.2.2 pgoyette } else 484 1.3.2.2 pgoyette tpu = NULL; 485 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 486 1.3.2.2 pgoyette 487 1.3.2.2 pgoyette if (tpu) 488 1.3.2.2 pgoyette threadpool_destroy(pool, sizeof(*tpu)); 489 1.3.2.2 pgoyette } 490 1.3.2.2 pgoyette 491 1.3.2.2 pgoyette /* Per-CPU thread pools */ 492 1.3.2.2 pgoyette 493 1.3.2.2 pgoyette int 494 1.3.2.2 pgoyette threadpool_percpu_get(threadpool_percpu_t **pool_percpup, pri_t pri) 495 1.3.2.2 pgoyette { 496 1.3.2.2 pgoyette threadpool_percpu_t *pool_percpu, *tmp = NULL; 497 1.3.2.2 pgoyette int error; 498 1.3.2.2 pgoyette 499 1.3.2.2 pgoyette THREADPOOL_INIT(); 500 1.3.2.2 pgoyette 501 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 502 1.3.2.2 pgoyette 503 1.3.2.2 pgoyette if (! threadpool_pri_is_valid(pri)) 504 1.3.2.2 pgoyette return EINVAL; 505 1.3.2.2 pgoyette 506 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 507 1.3.2.2 pgoyette pool_percpu = threadpool_lookup_percpu(pri); 508 1.3.2.2 pgoyette if (pool_percpu == NULL) { 509 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 510 1.3.2.2 pgoyette TP_LOG(("%s: No pool for pri=%d, creating one.\n", 511 1.3.2.2 pgoyette __func__, (int)pri)); 512 1.3.2.2 pgoyette error = threadpool_percpu_create(&tmp, pri); 513 1.3.2.2 pgoyette if (error) 514 1.3.2.2 pgoyette return error; 515 1.3.2.2 pgoyette KASSERT(tmp != NULL); 516 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 517 1.3.2.2 pgoyette pool_percpu = threadpool_lookup_percpu(pri); 518 1.3.2.2 pgoyette if (pool_percpu == NULL) { 519 1.3.2.2 pgoyette TP_LOG(("%s: Won the creation race for pri=%d.\n", 520 1.3.2.2 pgoyette __func__, (int)pri)); 521 1.3.2.2 pgoyette pool_percpu = tmp; 522 1.3.2.2 pgoyette tmp = NULL; 523 1.3.2.2 pgoyette threadpool_insert_percpu(pool_percpu); 524 1.3.2.2 pgoyette } 525 1.3.2.2 pgoyette } 526 1.3.2.2 pgoyette KASSERT(pool_percpu != NULL); 527 1.3.2.2 pgoyette if (pool_percpu->tpp_refcnt == UINT_MAX) { 528 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 529 1.3.2.2 pgoyette if (tmp != NULL) 530 1.3.2.2 pgoyette threadpool_percpu_destroy(tmp); 531 1.3.2.2 pgoyette return EBUSY; 532 1.3.2.2 pgoyette } 533 1.3.2.2 pgoyette pool_percpu->tpp_refcnt++; 534 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 535 1.3.2.2 pgoyette 536 1.3.2.2 pgoyette if (tmp != NULL) 537 1.3.2.2 pgoyette threadpool_percpu_destroy(tmp); 538 1.3.2.2 pgoyette KASSERT(pool_percpu != NULL); 539 1.3.2.2 pgoyette *pool_percpup = pool_percpu; 540 1.3.2.2 pgoyette return 0; 541 1.3.2.2 pgoyette } 542 1.3.2.2 pgoyette 543 1.3.2.2 pgoyette void 544 1.3.2.2 pgoyette threadpool_percpu_put(threadpool_percpu_t *pool_percpu, pri_t pri) 545 1.3.2.2 pgoyette { 546 1.3.2.2 pgoyette 547 1.3.2.2 pgoyette THREADPOOL_INIT(); 548 1.3.2.2 pgoyette 549 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 550 1.3.2.2 pgoyette 551 1.3.2.2 pgoyette KASSERT(threadpool_pri_is_valid(pri)); 552 1.3.2.2 pgoyette 553 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 554 1.3.2.2 pgoyette KASSERT(pool_percpu == threadpool_lookup_percpu(pri)); 555 1.3.2.2 pgoyette KASSERT(0 < pool_percpu->tpp_refcnt); 556 1.3.2.2 pgoyette if (--pool_percpu->tpp_refcnt == 0) { 557 1.3.2.2 pgoyette TP_LOG(("%s: Last reference for pri=%d, destroying pool.\n", 558 1.3.2.2 pgoyette __func__, (int)pri)); 559 1.3.2.2 pgoyette threadpool_remove_percpu(pool_percpu); 560 1.3.2.2 pgoyette } else 561 1.3.2.2 pgoyette pool_percpu = NULL; 562 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 563 1.3.2.2 pgoyette 564 1.3.2.2 pgoyette if (pool_percpu) 565 1.3.2.2 pgoyette threadpool_percpu_destroy(pool_percpu); 566 1.3.2.2 pgoyette } 567 1.3.2.2 pgoyette 568 1.3.2.2 pgoyette threadpool_t * 569 1.3.2.2 pgoyette threadpool_percpu_ref(threadpool_percpu_t *pool_percpu) 570 1.3.2.2 pgoyette { 571 1.3.2.2 pgoyette threadpool_t **poolp, *pool; 572 1.3.2.2 pgoyette 573 1.3.2.2 pgoyette poolp = percpu_getref(pool_percpu->tpp_percpu); 574 1.3.2.2 pgoyette pool = *poolp; 575 1.3.2.2 pgoyette percpu_putref(pool_percpu->tpp_percpu); 576 1.3.2.2 pgoyette 577 1.3.2.2 pgoyette return pool; 578 1.3.2.2 pgoyette } 579 1.3.2.2 pgoyette 580 1.3.2.2 pgoyette threadpool_t * 581 1.3.2.2 pgoyette threadpool_percpu_ref_remote(threadpool_percpu_t *pool_percpu, 582 1.3.2.2 pgoyette struct cpu_info *ci) 583 1.3.2.2 pgoyette { 584 1.3.2.2 pgoyette threadpool_t **poolp, *pool; 585 1.3.2.2 pgoyette 586 1.3.2.2 pgoyette percpu_traverse_enter(); 587 1.3.2.2 pgoyette poolp = percpu_getptr_remote(pool_percpu->tpp_percpu, ci); 588 1.3.2.2 pgoyette pool = *poolp; 589 1.3.2.2 pgoyette percpu_traverse_exit(); 590 1.3.2.2 pgoyette 591 1.3.2.2 pgoyette return pool; 592 1.3.2.2 pgoyette } 593 1.3.2.2 pgoyette 594 1.3.2.2 pgoyette static int 595 1.3.2.2 pgoyette threadpool_percpu_create(threadpool_percpu_t **pool_percpup, pri_t pri) 596 1.3.2.2 pgoyette { 597 1.3.2.2 pgoyette threadpool_percpu_t *pool_percpu; 598 1.3.2.2 pgoyette struct cpu_info *ci; 599 1.3.2.2 pgoyette CPU_INFO_ITERATOR cii; 600 1.3.2.2 pgoyette unsigned int i, j; 601 1.3.2.2 pgoyette int error; 602 1.3.2.2 pgoyette 603 1.3.2.2 pgoyette pool_percpu = kmem_zalloc(sizeof(*pool_percpu), KM_SLEEP); 604 1.3.2.2 pgoyette if (pool_percpu == NULL) { 605 1.3.2.2 pgoyette error = ENOMEM; 606 1.3.2.2 pgoyette goto fail0; 607 1.3.2.2 pgoyette } 608 1.3.2.2 pgoyette pool_percpu->tpp_pri = pri; 609 1.3.2.2 pgoyette 610 1.3.2.2 pgoyette pool_percpu->tpp_percpu = percpu_alloc(sizeof(threadpool_t *)); 611 1.3.2.2 pgoyette if (pool_percpu->tpp_percpu == NULL) { 612 1.3.2.2 pgoyette error = ENOMEM; 613 1.3.2.2 pgoyette goto fail1; 614 1.3.2.2 pgoyette } 615 1.3.2.2 pgoyette 616 1.3.2.2 pgoyette for (i = 0, CPU_INFO_FOREACH(cii, ci), i++) { 617 1.3.2.2 pgoyette threadpool_t *pool; 618 1.3.2.2 pgoyette 619 1.3.2.2 pgoyette error = threadpool_create(&pool, ci, pri, sizeof(*pool)); 620 1.3.2.2 pgoyette if (error) 621 1.3.2.2 pgoyette goto fail2; 622 1.3.2.2 pgoyette percpu_traverse_enter(); 623 1.3.2.2 pgoyette threadpool_t **const poolp = 624 1.3.2.2 pgoyette percpu_getptr_remote(pool_percpu->tpp_percpu, ci); 625 1.3.2.2 pgoyette *poolp = pool; 626 1.3.2.2 pgoyette percpu_traverse_exit(); 627 1.3.2.2 pgoyette } 628 1.3.2.2 pgoyette 629 1.3.2.2 pgoyette /* Success! */ 630 1.3.2.2 pgoyette *pool_percpup = (threadpool_percpu_t *)pool_percpu; 631 1.3.2.2 pgoyette return 0; 632 1.3.2.2 pgoyette 633 1.3.2.2 pgoyette fail2: for (j = 0, CPU_INFO_FOREACH(cii, ci), j++) { 634 1.3.2.2 pgoyette if (i <= j) 635 1.3.2.2 pgoyette break; 636 1.3.2.2 pgoyette percpu_traverse_enter(); 637 1.3.2.2 pgoyette threadpool_t **const poolp = 638 1.3.2.2 pgoyette percpu_getptr_remote(pool_percpu->tpp_percpu, ci); 639 1.3.2.2 pgoyette threadpool_t *const pool = *poolp; 640 1.3.2.2 pgoyette percpu_traverse_exit(); 641 1.3.2.2 pgoyette threadpool_destroy(pool, sizeof(*pool)); 642 1.3.2.2 pgoyette } 643 1.3.2.2 pgoyette percpu_free(pool_percpu->tpp_percpu, sizeof(struct taskthread_pool *)); 644 1.3.2.2 pgoyette fail1: kmem_free(pool_percpu, sizeof(*pool_percpu)); 645 1.3.2.2 pgoyette fail0: return error; 646 1.3.2.2 pgoyette } 647 1.3.2.2 pgoyette 648 1.3.2.2 pgoyette static void 649 1.3.2.2 pgoyette threadpool_percpu_destroy(threadpool_percpu_t *pool_percpu) 650 1.3.2.2 pgoyette { 651 1.3.2.2 pgoyette struct cpu_info *ci; 652 1.3.2.2 pgoyette CPU_INFO_ITERATOR cii; 653 1.3.2.2 pgoyette 654 1.3.2.2 pgoyette for (CPU_INFO_FOREACH(cii, ci)) { 655 1.3.2.2 pgoyette percpu_traverse_enter(); 656 1.3.2.2 pgoyette threadpool_t **const poolp = 657 1.3.2.2 pgoyette percpu_getptr_remote(pool_percpu->tpp_percpu, ci); 658 1.3.2.2 pgoyette threadpool_t *const pool = *poolp; 659 1.3.2.2 pgoyette percpu_traverse_exit(); 660 1.3.2.2 pgoyette threadpool_destroy(pool, sizeof(*pool)); 661 1.3.2.2 pgoyette } 662 1.3.2.2 pgoyette 663 1.3.2.2 pgoyette percpu_free(pool_percpu->tpp_percpu, sizeof(threadpool_t *)); 664 1.3.2.2 pgoyette kmem_free(pool_percpu, sizeof(*pool_percpu)); 665 1.3.2.2 pgoyette } 666 1.3.2.2 pgoyette 667 1.3.2.2 pgoyette /* Thread pool jobs */ 668 1.3.2.2 pgoyette 669 1.3.2.2 pgoyette void __printflike(4,5) 670 1.3.2.2 pgoyette threadpool_job_init(threadpool_job_t *ext_job, threadpool_job_fn_t fn, 671 1.3.2.2 pgoyette kmutex_t *lock, const char *fmt, ...) 672 1.3.2.2 pgoyette { 673 1.3.2.2 pgoyette threadpool_job_impl_t *job = THREADPOOL_JOB_TO_IMPL(ext_job); 674 1.3.2.2 pgoyette va_list ap; 675 1.3.2.2 pgoyette 676 1.3.2.2 pgoyette va_start(ap, fmt); 677 1.3.2.2 pgoyette (void)vsnprintf(job->job_name, sizeof(job->job_name), fmt, ap); 678 1.3.2.2 pgoyette va_end(ap); 679 1.3.2.2 pgoyette 680 1.3.2.2 pgoyette job->job_lock = lock; 681 1.3.2.2 pgoyette job->job_thread = NULL; 682 1.3.2.2 pgoyette job->job_refcnt = 0; 683 1.3.2.2 pgoyette cv_init(&job->job_cv, job->job_name); 684 1.3.2.2 pgoyette job->job_fn = fn; 685 1.3.2.2 pgoyette } 686 1.3.2.2 pgoyette 687 1.3.2.2 pgoyette static void 688 1.3.2.2 pgoyette threadpool_job_dead(threadpool_job_t *ext_job) 689 1.3.2.2 pgoyette { 690 1.3.2.2 pgoyette 691 1.3.2.2 pgoyette panic("threadpool job %p ran after destruction", ext_job); 692 1.3.2.2 pgoyette } 693 1.3.2.2 pgoyette 694 1.3.2.2 pgoyette void 695 1.3.2.2 pgoyette threadpool_job_destroy(threadpool_job_t *ext_job) 696 1.3.2.2 pgoyette { 697 1.3.2.2 pgoyette threadpool_job_impl_t *job = THREADPOOL_JOB_TO_IMPL(ext_job); 698 1.3.2.2 pgoyette 699 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 700 1.3.2.2 pgoyette 701 1.3.2.2 pgoyette KASSERTMSG((job->job_thread == NULL), "job %p still running", job); 702 1.3.2.2 pgoyette 703 1.3.2.2 pgoyette mutex_enter(job->job_lock); 704 1.3.2.2 pgoyette while (0 < job->job_refcnt) 705 1.3.2.2 pgoyette cv_wait(&job->job_cv, job->job_lock); 706 1.3.2.2 pgoyette mutex_exit(job->job_lock); 707 1.3.2.2 pgoyette 708 1.3.2.2 pgoyette job->job_lock = NULL; 709 1.3.2.2 pgoyette KASSERT(job->job_thread == NULL); 710 1.3.2.2 pgoyette KASSERT(job->job_refcnt == 0); 711 1.3.2.2 pgoyette KASSERT(!cv_has_waiters(&job->job_cv)); 712 1.3.2.2 pgoyette cv_destroy(&job->job_cv); 713 1.3.2.2 pgoyette job->job_fn = threadpool_job_dead; 714 1.3.2.2 pgoyette (void)strlcpy(job->job_name, "deadjob", sizeof(job->job_name)); 715 1.3.2.2 pgoyette } 716 1.3.2.2 pgoyette 717 1.3.2.2 pgoyette static int 718 1.3.2.2 pgoyette threadpool_job_hold(threadpool_job_impl_t *job) 719 1.3.2.2 pgoyette { 720 1.3.2.2 pgoyette unsigned int refcnt; 721 1.3.2.2 pgoyette do { 722 1.3.2.2 pgoyette refcnt = job->job_refcnt; 723 1.3.2.2 pgoyette if (refcnt == UINT_MAX) 724 1.3.2.2 pgoyette return EBUSY; 725 1.3.2.2 pgoyette } while (atomic_cas_uint(&job->job_refcnt, refcnt, (refcnt + 1)) 726 1.3.2.2 pgoyette != refcnt); 727 1.3.2.2 pgoyette 728 1.3.2.2 pgoyette return 0; 729 1.3.2.2 pgoyette } 730 1.3.2.2 pgoyette 731 1.3.2.2 pgoyette static void 732 1.3.2.2 pgoyette threadpool_job_rele(threadpool_job_impl_t *job) 733 1.3.2.2 pgoyette { 734 1.3.2.2 pgoyette unsigned int refcnt; 735 1.3.2.2 pgoyette 736 1.3.2.2 pgoyette do { 737 1.3.2.2 pgoyette refcnt = job->job_refcnt; 738 1.3.2.2 pgoyette KASSERT(0 < refcnt); 739 1.3.2.2 pgoyette if (refcnt == 1) { 740 1.3.2.2 pgoyette mutex_enter(job->job_lock); 741 1.3.2.2 pgoyette refcnt = atomic_dec_uint_nv(&job->job_refcnt); 742 1.3.2.2 pgoyette KASSERT(refcnt != UINT_MAX); 743 1.3.2.2 pgoyette if (refcnt == 0) 744 1.3.2.2 pgoyette cv_broadcast(&job->job_cv); 745 1.3.2.2 pgoyette mutex_exit(job->job_lock); 746 1.3.2.2 pgoyette return; 747 1.3.2.2 pgoyette } 748 1.3.2.2 pgoyette } while (atomic_cas_uint(&job->job_refcnt, refcnt, (refcnt - 1)) 749 1.3.2.2 pgoyette != refcnt); 750 1.3.2.2 pgoyette } 751 1.3.2.2 pgoyette 752 1.3.2.2 pgoyette void 753 1.3.2.2 pgoyette threadpool_job_done(threadpool_job_t *ext_job) 754 1.3.2.2 pgoyette { 755 1.3.2.2 pgoyette threadpool_job_impl_t *job = THREADPOOL_JOB_TO_IMPL(ext_job); 756 1.3.2.2 pgoyette 757 1.3.2.2 pgoyette KASSERT(mutex_owned(job->job_lock)); 758 1.3.2.2 pgoyette KASSERT(job->job_thread != NULL); 759 1.3.2.2 pgoyette KASSERT(job->job_thread->tpt_lwp == curlwp); 760 1.3.2.2 pgoyette 761 1.3.2.2 pgoyette cv_broadcast(&job->job_cv); 762 1.3.2.2 pgoyette job->job_thread = NULL; 763 1.3.2.2 pgoyette } 764 1.3.2.2 pgoyette 765 1.3.2.2 pgoyette void 766 1.3.2.2 pgoyette threadpool_schedule_job(threadpool_t *pool, threadpool_job_t *ext_job) 767 1.3.2.2 pgoyette { 768 1.3.2.2 pgoyette threadpool_job_impl_t *job = THREADPOOL_JOB_TO_IMPL(ext_job); 769 1.3.2.2 pgoyette 770 1.3.2.2 pgoyette KASSERT(mutex_owned(job->job_lock)); 771 1.3.2.2 pgoyette 772 1.3.2.2 pgoyette /* 773 1.3.2.2 pgoyette * If the job's already running, let it keep running. The job 774 1.3.2.2 pgoyette * is guaranteed by the interlock not to end early -- if it had 775 1.3.2.2 pgoyette * ended early, threadpool_job_done would have set job_thread 776 1.3.2.2 pgoyette * to NULL under the interlock. 777 1.3.2.2 pgoyette */ 778 1.3.2.2 pgoyette if (__predict_true(job->job_thread != NULL)) { 779 1.3.2.2 pgoyette TP_LOG(("%s: job '%s' already runnining.\n", 780 1.3.2.2 pgoyette __func__, job->job_name)); 781 1.3.2.2 pgoyette return; 782 1.3.2.2 pgoyette } 783 1.3.2.2 pgoyette 784 1.3.2.2 pgoyette /* Otherwise, try to assign a thread to the job. */ 785 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 786 1.3.2.2 pgoyette if (__predict_false(TAILQ_EMPTY(&pool->tp_idle_threads))) { 787 1.3.2.2 pgoyette /* Nobody's idle. Give it to the overseer. */ 788 1.3.2.2 pgoyette TP_LOG(("%s: giving job '%s' to overseer.\n", 789 1.3.2.2 pgoyette __func__, job->job_name)); 790 1.3.2.2 pgoyette job->job_thread = &pool->tp_overseer; 791 1.3.2.2 pgoyette TAILQ_INSERT_TAIL(&pool->tp_jobs, job, job_entry); 792 1.3.2.2 pgoyette } else { 793 1.3.2.2 pgoyette /* Assign it to the first idle thread. */ 794 1.3.2.2 pgoyette job->job_thread = TAILQ_FIRST(&pool->tp_idle_threads); 795 1.3.2.2 pgoyette TP_LOG(("%s: giving job '%s' to idle thread %p.\n", 796 1.3.2.2 pgoyette __func__, job->job_name, job->job_thread)); 797 1.3.2.2 pgoyette TAILQ_REMOVE(&pool->tp_idle_threads, job->job_thread, 798 1.3.2.2 pgoyette tpt_entry); 799 1.3.2.2 pgoyette threadpool_job_hold(job); 800 1.3.2.2 pgoyette job->job_thread->tpt_job = job; 801 1.3.2.2 pgoyette } 802 1.3.2.2 pgoyette 803 1.3.2.2 pgoyette /* Notify whomever we gave it to, overseer or idle thread. */ 804 1.3.2.2 pgoyette KASSERT(job->job_thread != NULL); 805 1.3.2.2 pgoyette cv_broadcast(&job->job_thread->tpt_cv); 806 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 807 1.3.2.2 pgoyette } 808 1.3.2.2 pgoyette 809 1.3.2.2 pgoyette bool 810 1.3.2.2 pgoyette threadpool_cancel_job_async(threadpool_t *pool, threadpool_job_t *ext_job) 811 1.3.2.2 pgoyette { 812 1.3.2.2 pgoyette threadpool_job_impl_t *job = THREADPOOL_JOB_TO_IMPL(ext_job); 813 1.3.2.2 pgoyette 814 1.3.2.2 pgoyette KASSERT(mutex_owned(job->job_lock)); 815 1.3.2.2 pgoyette 816 1.3.2.2 pgoyette /* 817 1.3.2.2 pgoyette * XXXJRT This fails (albeit safely) when all of the following 818 1.3.2.2 pgoyette * are true: 819 1.3.2.2 pgoyette * 820 1.3.2.2 pgoyette * => "pool" is something other than what the job was 821 1.3.2.2 pgoyette * scheduled on. This can legitimately occur if, 822 1.3.2.2 pgoyette * for example, a job is percpu-scheduled on CPU0 823 1.3.2.2 pgoyette * and then CPU1 attempts to cancel it without taking 824 1.3.2.2 pgoyette * a remote pool reference. (this might happen by 825 1.3.2.2 pgoyette * "luck of the draw"). 826 1.3.2.2 pgoyette * 827 1.3.2.2 pgoyette * => "job" is not yet running, but is assigned to the 828 1.3.2.2 pgoyette * overseer. 829 1.3.2.2 pgoyette * 830 1.3.2.2 pgoyette * When this happens, this code makes the determination that 831 1.3.2.2 pgoyette * the job is already running. The failure mode is that the 832 1.3.2.2 pgoyette * caller is told the job is running, and thus has to wait. 833 1.3.2.2 pgoyette * The overseer will eventually get to it and the job will 834 1.3.2.2 pgoyette * proceed as if it had been already running. 835 1.3.2.2 pgoyette */ 836 1.3.2.2 pgoyette 837 1.3.2.2 pgoyette if (job->job_thread == NULL) { 838 1.3.2.2 pgoyette /* Nothing to do. Guaranteed not running. */ 839 1.3.2.2 pgoyette return true; 840 1.3.2.2 pgoyette } else if (job->job_thread == &pool->tp_overseer) { 841 1.3.2.2 pgoyette /* Take it off the list to guarantee it won't run. */ 842 1.3.2.2 pgoyette job->job_thread = NULL; 843 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 844 1.3.2.2 pgoyette TAILQ_REMOVE(&pool->tp_jobs, job, job_entry); 845 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 846 1.3.2.2 pgoyette return true; 847 1.3.2.2 pgoyette } else { 848 1.3.2.2 pgoyette /* Too late -- already running. */ 849 1.3.2.2 pgoyette return false; 850 1.3.2.2 pgoyette } 851 1.3.2.2 pgoyette } 852 1.3.2.2 pgoyette 853 1.3.2.2 pgoyette void 854 1.3.2.2 pgoyette threadpool_cancel_job(threadpool_t *pool, threadpool_job_t *ext_job) 855 1.3.2.2 pgoyette { 856 1.3.2.2 pgoyette threadpool_job_impl_t *job = THREADPOOL_JOB_TO_IMPL(ext_job); 857 1.3.2.2 pgoyette 858 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 859 1.3.2.2 pgoyette 860 1.3.2.2 pgoyette KASSERT(mutex_owned(job->job_lock)); 861 1.3.2.2 pgoyette 862 1.3.2.2 pgoyette if (threadpool_cancel_job_async(pool, ext_job)) 863 1.3.2.2 pgoyette return; 864 1.3.2.2 pgoyette 865 1.3.2.2 pgoyette /* Already running. Wait for it to complete. */ 866 1.3.2.2 pgoyette while (job->job_thread != NULL) 867 1.3.2.2 pgoyette cv_wait(&job->job_cv, job->job_lock); 868 1.3.2.2 pgoyette } 869 1.3.2.2 pgoyette 870 1.3.2.2 pgoyette /* Thread pool overseer thread */ 871 1.3.2.2 pgoyette 872 1.3.2.2 pgoyette static void __dead 873 1.3.2.2 pgoyette threadpool_overseer_thread(void *arg) 874 1.3.2.2 pgoyette { 875 1.3.2.2 pgoyette struct threadpool_thread *const overseer = arg; 876 1.3.2.2 pgoyette threadpool_t *const pool = overseer->tpt_pool; 877 1.3.2.2 pgoyette struct lwp *lwp = NULL; 878 1.3.2.2 pgoyette int ktflags; 879 1.3.2.2 pgoyette int error; 880 1.3.2.2 pgoyette 881 1.3.2.2 pgoyette KASSERT((pool->tp_cpu == NULL) || (pool->tp_cpu == curcpu())); 882 1.3.2.2 pgoyette 883 1.3.2.2 pgoyette /* Wait until we're initialized. */ 884 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 885 1.3.2.2 pgoyette while (overseer->tpt_lwp == NULL) 886 1.3.2.2 pgoyette cv_wait(&overseer->tpt_cv, &pool->tp_lock); 887 1.3.2.2 pgoyette 888 1.3.2.2 pgoyette TP_LOG(("%s: starting.\n", __func__)); 889 1.3.2.2 pgoyette 890 1.3.2.2 pgoyette for (;;) { 891 1.3.2.2 pgoyette /* Wait until there's a job. */ 892 1.3.2.2 pgoyette while (TAILQ_EMPTY(&pool->tp_jobs)) { 893 1.3.2.2 pgoyette if (ISSET(pool->tp_flags, THREADPOOL_DYING)) { 894 1.3.2.2 pgoyette TP_LOG(("%s: THREADPOOL_DYING\n", 895 1.3.2.2 pgoyette __func__)); 896 1.3.2.2 pgoyette break; 897 1.3.2.2 pgoyette } 898 1.3.2.2 pgoyette cv_wait(&overseer->tpt_cv, &pool->tp_lock); 899 1.3.2.2 pgoyette } 900 1.3.2.2 pgoyette if (__predict_false(TAILQ_EMPTY(&pool->tp_jobs))) 901 1.3.2.2 pgoyette break; 902 1.3.2.2 pgoyette 903 1.3.2.2 pgoyette /* If there are no threads, we'll have to try to start one. */ 904 1.3.2.2 pgoyette if (TAILQ_EMPTY(&pool->tp_idle_threads)) { 905 1.3.2.2 pgoyette TP_LOG(("%s: Got a job, need to create a thread.\n", 906 1.3.2.2 pgoyette __func__)); 907 1.3.2.2 pgoyette error = threadpool_hold(pool); 908 1.3.2.2 pgoyette if (error) { 909 1.3.2.2 pgoyette (void)kpause("thrdplrf", false, hz, 910 1.3.2.2 pgoyette &pool->tp_lock); 911 1.3.2.2 pgoyette continue; 912 1.3.2.2 pgoyette } 913 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 914 1.3.2.2 pgoyette 915 1.3.2.2 pgoyette struct threadpool_thread *const thread = 916 1.3.2.2 pgoyette pool_cache_get(threadpool_thread_pc, PR_WAITOK); 917 1.3.2.2 pgoyette thread->tpt_lwp = NULL; 918 1.3.2.2 pgoyette thread->tpt_pool = pool; 919 1.3.2.2 pgoyette thread->tpt_job = NULL; 920 1.3.2.2 pgoyette cv_init(&thread->tpt_cv, "poolthrd"); 921 1.3.2.2 pgoyette 922 1.3.2.2 pgoyette ktflags = 0; 923 1.3.2.2 pgoyette ktflags |= KTHREAD_MPSAFE; 924 1.3.2.2 pgoyette if (pool->tp_pri < PRI_KERNEL) 925 1.3.2.2 pgoyette ktflags |= KTHREAD_TS; 926 1.3.2.2 pgoyette error = kthread_create(pool->tp_pri, ktflags, 927 1.3.2.2 pgoyette pool->tp_cpu, &threadpool_thread, thread, &lwp, 928 1.3.2.2 pgoyette "poolthread/%d@%d", 929 1.3.2.2 pgoyette (pool->tp_cpu ? cpu_index(pool->tp_cpu) : -1), 930 1.3.2.2 pgoyette (int)pool->tp_pri); 931 1.3.2.2 pgoyette 932 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 933 1.3.2.2 pgoyette if (error) { 934 1.3.2.2 pgoyette pool_cache_put(threadpool_thread_pc, thread); 935 1.3.2.2 pgoyette threadpool_rele(pool); 936 1.3.2.2 pgoyette /* XXX What to do to wait for memory? */ 937 1.3.2.2 pgoyette (void)kpause("thrdplcr", false, hz, 938 1.3.2.2 pgoyette &pool->tp_lock); 939 1.3.2.2 pgoyette continue; 940 1.3.2.2 pgoyette } 941 1.3.2.2 pgoyette KASSERT(lwp != NULL); 942 1.3.2.2 pgoyette TAILQ_INSERT_TAIL(&pool->tp_idle_threads, thread, 943 1.3.2.2 pgoyette tpt_entry); 944 1.3.2.2 pgoyette thread->tpt_lwp = lwp; 945 1.3.2.2 pgoyette lwp = NULL; 946 1.3.2.2 pgoyette cv_broadcast(&thread->tpt_cv); 947 1.3.2.2 pgoyette continue; 948 1.3.2.2 pgoyette } 949 1.3.2.2 pgoyette 950 1.3.2.2 pgoyette /* There are idle threads, so try giving one a job. */ 951 1.3.2.2 pgoyette bool rele_job = true; 952 1.3.2.2 pgoyette threadpool_job_impl_t *const job = TAILQ_FIRST(&pool->tp_jobs); 953 1.3.2.2 pgoyette TAILQ_REMOVE(&pool->tp_jobs, job, job_entry); 954 1.3.2.2 pgoyette error = threadpool_job_hold(job); 955 1.3.2.2 pgoyette if (error) { 956 1.3.2.2 pgoyette TAILQ_INSERT_HEAD(&pool->tp_jobs, job, job_entry); 957 1.3.2.2 pgoyette (void)kpause("pooljob", false, hz, &pool->tp_lock); 958 1.3.2.2 pgoyette continue; 959 1.3.2.2 pgoyette } 960 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 961 1.3.2.2 pgoyette 962 1.3.2.2 pgoyette mutex_enter(job->job_lock); 963 1.3.2.2 pgoyette /* If the job was cancelled, we'll no longer be its thread. */ 964 1.3.2.2 pgoyette if (__predict_true(job->job_thread == overseer)) { 965 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 966 1.3.2.2 pgoyette if (__predict_false( 967 1.3.2.2 pgoyette TAILQ_EMPTY(&pool->tp_idle_threads))) { 968 1.3.2.2 pgoyette /* 969 1.3.2.2 pgoyette * Someone else snagged the thread 970 1.3.2.2 pgoyette * first. We'll have to try again. 971 1.3.2.2 pgoyette */ 972 1.3.2.2 pgoyette TP_LOG(("%s: '%s' lost race to use idle thread.\n", 973 1.3.2.2 pgoyette __func__, job->job_name)); 974 1.3.2.2 pgoyette TAILQ_INSERT_HEAD(&pool->tp_jobs, job, 975 1.3.2.2 pgoyette job_entry); 976 1.3.2.2 pgoyette } else { 977 1.3.2.2 pgoyette /* 978 1.3.2.2 pgoyette * Assign the job to the thread and 979 1.3.2.2 pgoyette * wake the thread so it starts work. 980 1.3.2.2 pgoyette */ 981 1.3.2.2 pgoyette struct threadpool_thread *const thread = 982 1.3.2.2 pgoyette TAILQ_FIRST(&pool->tp_idle_threads); 983 1.3.2.2 pgoyette 984 1.3.2.2 pgoyette TP_LOG(("%s: '%s' gets thread %p\n", 985 1.3.2.2 pgoyette __func__, job->job_name, thread)); 986 1.3.2.2 pgoyette KASSERT(thread->tpt_job == NULL); 987 1.3.2.2 pgoyette TAILQ_REMOVE(&pool->tp_idle_threads, thread, 988 1.3.2.2 pgoyette tpt_entry); 989 1.3.2.2 pgoyette thread->tpt_job = job; 990 1.3.2.2 pgoyette job->job_thread = thread; 991 1.3.2.2 pgoyette cv_broadcast(&thread->tpt_cv); 992 1.3.2.2 pgoyette /* Gave the thread our job reference. */ 993 1.3.2.2 pgoyette rele_job = false; 994 1.3.2.2 pgoyette } 995 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 996 1.3.2.2 pgoyette } 997 1.3.2.2 pgoyette mutex_exit(job->job_lock); 998 1.3.2.2 pgoyette if (__predict_false(rele_job)) 999 1.3.2.2 pgoyette threadpool_job_rele(job); 1000 1.3.2.2 pgoyette 1001 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 1002 1.3.2.2 pgoyette } 1003 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 1004 1.3.2.2 pgoyette 1005 1.3.2.2 pgoyette TP_LOG(("%s: exiting.\n", __func__)); 1006 1.3.2.2 pgoyette 1007 1.3.2.2 pgoyette threadpool_rele(pool); 1008 1.3.2.2 pgoyette kthread_exit(0); 1009 1.3.2.2 pgoyette } 1010 1.3.2.2 pgoyette 1011 1.3.2.2 pgoyette /* Thread pool thread */ 1012 1.3.2.2 pgoyette 1013 1.3.2.2 pgoyette static void __dead 1014 1.3.2.2 pgoyette threadpool_thread(void *arg) 1015 1.3.2.2 pgoyette { 1016 1.3.2.2 pgoyette struct threadpool_thread *const thread = arg; 1017 1.3.2.2 pgoyette threadpool_t *const pool = thread->tpt_pool; 1018 1.3.2.2 pgoyette 1019 1.3.2.2 pgoyette KASSERT((pool->tp_cpu == NULL) || (pool->tp_cpu == curcpu())); 1020 1.3.2.2 pgoyette 1021 1.3.2.2 pgoyette /* Wait until we're initialized and on the queue. */ 1022 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 1023 1.3.2.2 pgoyette while (thread->tpt_lwp == NULL) 1024 1.3.2.2 pgoyette cv_wait(&thread->tpt_cv, &pool->tp_lock); 1025 1.3.2.2 pgoyette 1026 1.3.2.2 pgoyette TP_LOG(("%s: starting.\n", __func__)); 1027 1.3.2.2 pgoyette 1028 1.3.2.2 pgoyette KASSERT(thread->tpt_lwp == curlwp); 1029 1.3.2.2 pgoyette for (;;) { 1030 1.3.2.2 pgoyette /* Wait until we are assigned a job. */ 1031 1.3.2.2 pgoyette while (thread->tpt_job == NULL) { 1032 1.3.2.2 pgoyette if (ISSET(pool->tp_flags, THREADPOOL_DYING)) { 1033 1.3.2.2 pgoyette TP_LOG(("%s: THREADPOOL_DYING\n", 1034 1.3.2.2 pgoyette __func__)); 1035 1.3.2.2 pgoyette break; 1036 1.3.2.2 pgoyette } 1037 1.3.2.2 pgoyette if (cv_timedwait(&thread->tpt_cv, &pool->tp_lock, 1038 1.3.2.2 pgoyette THREADPOOL_IDLE_TICKS)) 1039 1.3.2.2 pgoyette break; 1040 1.3.2.2 pgoyette } 1041 1.3.2.2 pgoyette if (__predict_false(thread->tpt_job == NULL)) { 1042 1.3.2.2 pgoyette TAILQ_REMOVE(&pool->tp_idle_threads, thread, 1043 1.3.2.2 pgoyette tpt_entry); 1044 1.3.2.2 pgoyette break; 1045 1.3.2.2 pgoyette } 1046 1.3.2.2 pgoyette 1047 1.3.2.2 pgoyette threadpool_job_impl_t *const job = thread->tpt_job; 1048 1.3.2.2 pgoyette KASSERT(job != NULL); 1049 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 1050 1.3.2.2 pgoyette 1051 1.3.2.2 pgoyette TP_LOG(("%s: running job '%s' on thread %p.\n", 1052 1.3.2.2 pgoyette __func__, job->job_name, thread)); 1053 1.3.2.2 pgoyette 1054 1.3.2.2 pgoyette /* Set our lwp name to reflect what job we're doing. */ 1055 1.3.2.2 pgoyette lwp_lock(curlwp); 1056 1.3.2.2 pgoyette char *const lwp_name = curlwp->l_name; 1057 1.3.2.2 pgoyette curlwp->l_name = job->job_name; 1058 1.3.2.2 pgoyette lwp_unlock(curlwp); 1059 1.3.2.2 pgoyette 1060 1.3.2.2 pgoyette /* Run the job. */ 1061 1.3.2.2 pgoyette (*job->job_fn)(THREADPOOL_IMPL_TO_JOB(job)); 1062 1.3.2.2 pgoyette 1063 1.3.2.2 pgoyette /* Restore our lwp name. */ 1064 1.3.2.2 pgoyette lwp_lock(curlwp); 1065 1.3.2.2 pgoyette curlwp->l_name = lwp_name; 1066 1.3.2.2 pgoyette lwp_unlock(curlwp); 1067 1.3.2.2 pgoyette 1068 1.3.2.2 pgoyette /* Job is done and its name is unreferenced. Release it. */ 1069 1.3.2.2 pgoyette threadpool_job_rele(job); 1070 1.3.2.2 pgoyette 1071 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 1072 1.3.2.2 pgoyette KASSERT(thread->tpt_job == job); 1073 1.3.2.2 pgoyette thread->tpt_job = NULL; 1074 1.3.2.2 pgoyette TAILQ_INSERT_TAIL(&pool->tp_idle_threads, thread, tpt_entry); 1075 1.3.2.2 pgoyette } 1076 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 1077 1.3.2.2 pgoyette 1078 1.3.2.2 pgoyette TP_LOG(("%s: thread %p exiting.\n", __func__, thread)); 1079 1.3.2.2 pgoyette 1080 1.3.2.2 pgoyette KASSERT(!cv_has_waiters(&thread->tpt_cv)); 1081 1.3.2.2 pgoyette cv_destroy(&thread->tpt_cv); 1082 1.3.2.2 pgoyette pool_cache_put(threadpool_thread_pc, thread); 1083 1.3.2.2 pgoyette threadpool_rele(pool); 1084 1.3.2.2 pgoyette kthread_exit(0); 1085 1.3.2.2 pgoyette } 1086
Indexes created Tue Sep 23 07:09:52 GMT 2025