kern_threadpool.c revision 1.3.2.3
1 1.3.2.3 pgoyette /* $NetBSD: kern_threadpool.c,v 1.3.2.3 2019/01/18 08:50:57 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.3 pgoyette * struct threadpool_job 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.3 pgoyette __KERNEL_RCSID(0, "$NetBSD: kern_threadpool.c,v 1.3.2.3 2019/01/18 08:50:57 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.3 pgoyette #include <sys/sysctl.h> 102 1.3.2.2 pgoyette #include <sys/threadpool.h> 103 1.3.2.2 pgoyette 104 1.3.2.2 pgoyette /* Data structures */ 105 1.3.2.2 pgoyette 106 1.3.2.3 pgoyette TAILQ_HEAD(job_head, threadpool_job); 107 1.3.2.2 pgoyette TAILQ_HEAD(thread_head, threadpool_thread); 108 1.3.2.2 pgoyette 109 1.3.2.2 pgoyette struct threadpool_thread { 110 1.3.2.2 pgoyette struct lwp *tpt_lwp; 111 1.3.2.3 pgoyette char *tpt_lwp_savedname; 112 1.3.2.3 pgoyette struct threadpool *tpt_pool; 113 1.3.2.3 pgoyette struct threadpool_job *tpt_job; 114 1.3.2.2 pgoyette kcondvar_t tpt_cv; 115 1.3.2.2 pgoyette TAILQ_ENTRY(threadpool_thread) tpt_entry; 116 1.3.2.2 pgoyette }; 117 1.3.2.2 pgoyette 118 1.3.2.2 pgoyette struct threadpool { 119 1.3.2.2 pgoyette kmutex_t tp_lock; 120 1.3.2.2 pgoyette struct threadpool_thread tp_overseer; 121 1.3.2.2 pgoyette struct job_head tp_jobs; 122 1.3.2.2 pgoyette struct thread_head tp_idle_threads; 123 1.3.2.3 pgoyette uint64_t tp_refcnt; 124 1.3.2.2 pgoyette int tp_flags; 125 1.3.2.2 pgoyette #define THREADPOOL_DYING 0x01 126 1.3.2.2 pgoyette struct cpu_info *tp_cpu; 127 1.3.2.2 pgoyette pri_t tp_pri; 128 1.3.2.2 pgoyette }; 129 1.3.2.2 pgoyette 130 1.3.2.3 pgoyette static void threadpool_hold(struct threadpool *); 131 1.3.2.3 pgoyette static void threadpool_rele(struct threadpool *); 132 1.3.2.2 pgoyette 133 1.3.2.3 pgoyette static int threadpool_percpu_create(struct threadpool_percpu **, pri_t); 134 1.3.2.3 pgoyette static void threadpool_percpu_destroy(struct threadpool_percpu *); 135 1.3.2.2 pgoyette 136 1.3.2.3 pgoyette static threadpool_job_fn_t threadpool_job_dead; 137 1.3.2.2 pgoyette 138 1.3.2.3 pgoyette static void threadpool_job_hold(struct threadpool_job *); 139 1.3.2.3 pgoyette static void threadpool_job_rele(struct threadpool_job *); 140 1.3.2.2 pgoyette 141 1.3.2.2 pgoyette static void threadpool_overseer_thread(void *) __dead; 142 1.3.2.2 pgoyette static void threadpool_thread(void *) __dead; 143 1.3.2.2 pgoyette 144 1.3.2.2 pgoyette static pool_cache_t threadpool_thread_pc __read_mostly; 145 1.3.2.2 pgoyette 146 1.3.2.2 pgoyette static kmutex_t threadpools_lock __cacheline_aligned; 147 1.3.2.2 pgoyette 148 1.3.2.3 pgoyette /* Default to 30 second idle timeout for pool threads. */ 149 1.3.2.3 pgoyette static int threadpool_idle_time_ms = 30 * 1000; 150 1.3.2.2 pgoyette 151 1.3.2.2 pgoyette struct threadpool_unbound { 152 1.3.2.2 pgoyette struct threadpool tpu_pool; 153 1.3.2.2 pgoyette 154 1.3.2.2 pgoyette /* protected by threadpools_lock */ 155 1.3.2.2 pgoyette LIST_ENTRY(threadpool_unbound) tpu_link; 156 1.3.2.3 pgoyette uint64_t tpu_refcnt; 157 1.3.2.2 pgoyette }; 158 1.3.2.2 pgoyette 159 1.3.2.2 pgoyette static LIST_HEAD(, threadpool_unbound) unbound_threadpools; 160 1.3.2.2 pgoyette 161 1.3.2.2 pgoyette static struct threadpool_unbound * 162 1.3.2.2 pgoyette threadpool_lookup_unbound(pri_t pri) 163 1.3.2.2 pgoyette { 164 1.3.2.2 pgoyette struct threadpool_unbound *tpu; 165 1.3.2.2 pgoyette 166 1.3.2.2 pgoyette LIST_FOREACH(tpu, &unbound_threadpools, tpu_link) { 167 1.3.2.2 pgoyette if (tpu->tpu_pool.tp_pri == pri) 168 1.3.2.2 pgoyette return tpu; 169 1.3.2.2 pgoyette } 170 1.3.2.2 pgoyette return NULL; 171 1.3.2.2 pgoyette } 172 1.3.2.2 pgoyette 173 1.3.2.2 pgoyette static void 174 1.3.2.2 pgoyette threadpool_insert_unbound(struct threadpool_unbound *tpu) 175 1.3.2.2 pgoyette { 176 1.3.2.2 pgoyette KASSERT(threadpool_lookup_unbound(tpu->tpu_pool.tp_pri) == NULL); 177 1.3.2.2 pgoyette LIST_INSERT_HEAD(&unbound_threadpools, tpu, tpu_link); 178 1.3.2.2 pgoyette } 179 1.3.2.2 pgoyette 180 1.3.2.2 pgoyette static void 181 1.3.2.2 pgoyette threadpool_remove_unbound(struct threadpool_unbound *tpu) 182 1.3.2.2 pgoyette { 183 1.3.2.2 pgoyette KASSERT(threadpool_lookup_unbound(tpu->tpu_pool.tp_pri) == tpu); 184 1.3.2.2 pgoyette LIST_REMOVE(tpu, tpu_link); 185 1.3.2.2 pgoyette } 186 1.3.2.2 pgoyette 187 1.3.2.2 pgoyette struct threadpool_percpu { 188 1.3.2.2 pgoyette percpu_t * tpp_percpu; 189 1.3.2.2 pgoyette pri_t tpp_pri; 190 1.3.2.2 pgoyette 191 1.3.2.2 pgoyette /* protected by threadpools_lock */ 192 1.3.2.2 pgoyette LIST_ENTRY(threadpool_percpu) tpp_link; 193 1.3.2.3 pgoyette uint64_t tpp_refcnt; 194 1.3.2.2 pgoyette }; 195 1.3.2.2 pgoyette 196 1.3.2.2 pgoyette static LIST_HEAD(, threadpool_percpu) percpu_threadpools; 197 1.3.2.2 pgoyette 198 1.3.2.3 pgoyette static struct threadpool_percpu * 199 1.3.2.2 pgoyette threadpool_lookup_percpu(pri_t pri) 200 1.3.2.2 pgoyette { 201 1.3.2.3 pgoyette struct threadpool_percpu *tpp; 202 1.3.2.2 pgoyette 203 1.3.2.2 pgoyette LIST_FOREACH(tpp, &percpu_threadpools, tpp_link) { 204 1.3.2.2 pgoyette if (tpp->tpp_pri == pri) 205 1.3.2.2 pgoyette return tpp; 206 1.3.2.2 pgoyette } 207 1.3.2.2 pgoyette return NULL; 208 1.3.2.2 pgoyette } 209 1.3.2.2 pgoyette 210 1.3.2.2 pgoyette static void 211 1.3.2.3 pgoyette threadpool_insert_percpu(struct threadpool_percpu *tpp) 212 1.3.2.2 pgoyette { 213 1.3.2.2 pgoyette KASSERT(threadpool_lookup_percpu(tpp->tpp_pri) == NULL); 214 1.3.2.2 pgoyette LIST_INSERT_HEAD(&percpu_threadpools, tpp, tpp_link); 215 1.3.2.2 pgoyette } 216 1.3.2.2 pgoyette 217 1.3.2.2 pgoyette static void 218 1.3.2.3 pgoyette threadpool_remove_percpu(struct threadpool_percpu *tpp) 219 1.3.2.2 pgoyette { 220 1.3.2.2 pgoyette KASSERT(threadpool_lookup_percpu(tpp->tpp_pri) == tpp); 221 1.3.2.2 pgoyette LIST_REMOVE(tpp, tpp_link); 222 1.3.2.2 pgoyette } 223 1.3.2.2 pgoyette 224 1.3.2.2 pgoyette #ifdef THREADPOOL_VERBOSE 225 1.3.2.2 pgoyette #define TP_LOG(x) printf x 226 1.3.2.2 pgoyette #else 227 1.3.2.2 pgoyette #define TP_LOG(x) /* nothing */ 228 1.3.2.2 pgoyette #endif /* THREADPOOL_VERBOSE */ 229 1.3.2.2 pgoyette 230 1.3.2.2 pgoyette static int 231 1.3.2.3 pgoyette sysctl_kern_threadpool_idle_ms(SYSCTLFN_ARGS) 232 1.3.2.3 pgoyette { 233 1.3.2.3 pgoyette struct sysctlnode node; 234 1.3.2.3 pgoyette int val, error; 235 1.3.2.3 pgoyette 236 1.3.2.3 pgoyette node = *rnode; 237 1.3.2.3 pgoyette 238 1.3.2.3 pgoyette val = threadpool_idle_time_ms; 239 1.3.2.3 pgoyette node.sysctl_data = &val; 240 1.3.2.3 pgoyette error = sysctl_lookup(SYSCTLFN_CALL(&node)); 241 1.3.2.3 pgoyette if (error == 0 && newp != NULL) { 242 1.3.2.3 pgoyette /* Disallow negative values and 0 (forever). */ 243 1.3.2.3 pgoyette if (val < 1) 244 1.3.2.3 pgoyette error = EINVAL; 245 1.3.2.3 pgoyette else 246 1.3.2.3 pgoyette threadpool_idle_time_ms = val; 247 1.3.2.3 pgoyette } 248 1.3.2.3 pgoyette 249 1.3.2.3 pgoyette return error; 250 1.3.2.3 pgoyette } 251 1.3.2.3 pgoyette 252 1.3.2.3 pgoyette SYSCTL_SETUP_PROTO(sysctl_threadpool_setup); 253 1.3.2.3 pgoyette 254 1.3.2.3 pgoyette SYSCTL_SETUP(sysctl_threadpool_setup, 255 1.3.2.3 pgoyette "sysctl kern.threadpool subtree setup") 256 1.3.2.3 pgoyette { 257 1.3.2.3 pgoyette const struct sysctlnode *rnode, *cnode; 258 1.3.2.3 pgoyette int error __diagused; 259 1.3.2.3 pgoyette 260 1.3.2.3 pgoyette error = sysctl_createv(clog, 0, NULL, &rnode, 261 1.3.2.3 pgoyette CTLFLAG_PERMANENT, 262 1.3.2.3 pgoyette CTLTYPE_NODE, "threadpool", 263 1.3.2.3 pgoyette SYSCTL_DESCR("threadpool subsystem options"), 264 1.3.2.3 pgoyette NULL, 0, NULL, 0, 265 1.3.2.3 pgoyette CTL_KERN, CTL_CREATE, CTL_EOL); 266 1.3.2.3 pgoyette KASSERT(error == 0); 267 1.3.2.3 pgoyette 268 1.3.2.3 pgoyette error = sysctl_createv(clog, 0, &rnode, &cnode, 269 1.3.2.3 pgoyette CTLFLAG_PERMANENT | CTLFLAG_READWRITE, 270 1.3.2.3 pgoyette CTLTYPE_INT, "idle_ms", 271 1.3.2.3 pgoyette SYSCTL_DESCR("idle thread timeout in ms"), 272 1.3.2.3 pgoyette sysctl_kern_threadpool_idle_ms, 0, NULL, 0, 273 1.3.2.3 pgoyette CTL_CREATE, CTL_EOL); 274 1.3.2.3 pgoyette KASSERT(error == 0); 275 1.3.2.3 pgoyette } 276 1.3.2.3 pgoyette 277 1.3.2.3 pgoyette void 278 1.3.2.2 pgoyette threadpools_init(void) 279 1.3.2.2 pgoyette { 280 1.3.2.2 pgoyette 281 1.3.2.2 pgoyette threadpool_thread_pc = 282 1.3.2.2 pgoyette pool_cache_init(sizeof(struct threadpool_thread), 0, 0, 0, 283 1.3.2.2 pgoyette "thplthrd", NULL, IPL_NONE, NULL, NULL, NULL); 284 1.3.2.2 pgoyette 285 1.3.2.2 pgoyette LIST_INIT(&unbound_threadpools); 286 1.3.2.2 pgoyette LIST_INIT(&percpu_threadpools); 287 1.3.2.2 pgoyette mutex_init(&threadpools_lock, MUTEX_DEFAULT, IPL_NONE); 288 1.3.2.2 pgoyette } 289 1.3.2.2 pgoyette 290 1.3.2.2 pgoyette /* Thread pool creation */ 291 1.3.2.2 pgoyette 292 1.3.2.2 pgoyette static bool 293 1.3.2.2 pgoyette threadpool_pri_is_valid(pri_t pri) 294 1.3.2.2 pgoyette { 295 1.3.2.2 pgoyette return (pri == PRI_NONE || (pri >= PRI_USER && pri < PRI_COUNT)); 296 1.3.2.2 pgoyette } 297 1.3.2.2 pgoyette 298 1.3.2.2 pgoyette static int 299 1.3.2.3 pgoyette threadpool_create(struct threadpool *const pool, struct cpu_info *ci, 300 1.3.2.3 pgoyette pri_t pri) 301 1.3.2.2 pgoyette { 302 1.3.2.2 pgoyette struct lwp *lwp; 303 1.3.2.2 pgoyette int ktflags; 304 1.3.2.2 pgoyette int error; 305 1.3.2.2 pgoyette 306 1.3.2.2 pgoyette KASSERT(threadpool_pri_is_valid(pri)); 307 1.3.2.2 pgoyette 308 1.3.2.2 pgoyette mutex_init(&pool->tp_lock, MUTEX_DEFAULT, IPL_VM); 309 1.3.2.2 pgoyette /* XXX overseer */ 310 1.3.2.2 pgoyette TAILQ_INIT(&pool->tp_jobs); 311 1.3.2.2 pgoyette TAILQ_INIT(&pool->tp_idle_threads); 312 1.3.2.3 pgoyette pool->tp_refcnt = 1; /* overseer's reference */ 313 1.3.2.2 pgoyette pool->tp_flags = 0; 314 1.3.2.2 pgoyette pool->tp_cpu = ci; 315 1.3.2.2 pgoyette pool->tp_pri = pri; 316 1.3.2.2 pgoyette 317 1.3.2.2 pgoyette pool->tp_overseer.tpt_lwp = NULL; 318 1.3.2.2 pgoyette pool->tp_overseer.tpt_pool = pool; 319 1.3.2.2 pgoyette pool->tp_overseer.tpt_job = NULL; 320 1.3.2.2 pgoyette cv_init(&pool->tp_overseer.tpt_cv, "poolover"); 321 1.3.2.2 pgoyette 322 1.3.2.2 pgoyette ktflags = 0; 323 1.3.2.2 pgoyette ktflags |= KTHREAD_MPSAFE; 324 1.3.2.2 pgoyette if (pri < PRI_KERNEL) 325 1.3.2.2 pgoyette ktflags |= KTHREAD_TS; 326 1.3.2.2 pgoyette error = kthread_create(pri, ktflags, ci, &threadpool_overseer_thread, 327 1.3.2.2 pgoyette &pool->tp_overseer, &lwp, 328 1.3.2.2 pgoyette "pooloverseer/%d@%d", (ci ? cpu_index(ci) : -1), (int)pri); 329 1.3.2.2 pgoyette if (error) 330 1.3.2.2 pgoyette goto fail0; 331 1.3.2.2 pgoyette 332 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 333 1.3.2.2 pgoyette pool->tp_overseer.tpt_lwp = lwp; 334 1.3.2.2 pgoyette cv_broadcast(&pool->tp_overseer.tpt_cv); 335 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 336 1.3.2.2 pgoyette 337 1.3.2.2 pgoyette return 0; 338 1.3.2.2 pgoyette 339 1.3.2.2 pgoyette fail0: KASSERT(error); 340 1.3.2.2 pgoyette KASSERT(pool->tp_overseer.tpt_job == NULL); 341 1.3.2.2 pgoyette KASSERT(pool->tp_overseer.tpt_pool == pool); 342 1.3.2.2 pgoyette KASSERT(pool->tp_flags == 0); 343 1.3.2.2 pgoyette KASSERT(pool->tp_refcnt == 0); 344 1.3.2.2 pgoyette KASSERT(TAILQ_EMPTY(&pool->tp_idle_threads)); 345 1.3.2.2 pgoyette KASSERT(TAILQ_EMPTY(&pool->tp_jobs)); 346 1.3.2.2 pgoyette KASSERT(!cv_has_waiters(&pool->tp_overseer.tpt_cv)); 347 1.3.2.2 pgoyette cv_destroy(&pool->tp_overseer.tpt_cv); 348 1.3.2.2 pgoyette mutex_destroy(&pool->tp_lock); 349 1.3.2.2 pgoyette return error; 350 1.3.2.2 pgoyette } 351 1.3.2.2 pgoyette 352 1.3.2.2 pgoyette /* Thread pool destruction */ 353 1.3.2.2 pgoyette 354 1.3.2.2 pgoyette static void 355 1.3.2.3 pgoyette threadpool_destroy(struct threadpool *pool) 356 1.3.2.2 pgoyette { 357 1.3.2.2 pgoyette struct threadpool_thread *thread; 358 1.3.2.2 pgoyette 359 1.3.2.2 pgoyette /* Mark the pool dying and wait for threads to commit suicide. */ 360 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 361 1.3.2.2 pgoyette KASSERT(TAILQ_EMPTY(&pool->tp_jobs)); 362 1.3.2.2 pgoyette pool->tp_flags |= THREADPOOL_DYING; 363 1.3.2.2 pgoyette cv_broadcast(&pool->tp_overseer.tpt_cv); 364 1.3.2.2 pgoyette TAILQ_FOREACH(thread, &pool->tp_idle_threads, tpt_entry) 365 1.3.2.2 pgoyette cv_broadcast(&thread->tpt_cv); 366 1.3.2.2 pgoyette while (0 < pool->tp_refcnt) { 367 1.3.2.3 pgoyette TP_LOG(("%s: draining %" PRIu64 " references...\n", __func__, 368 1.3.2.2 pgoyette pool->tp_refcnt)); 369 1.3.2.2 pgoyette cv_wait(&pool->tp_overseer.tpt_cv, &pool->tp_lock); 370 1.3.2.2 pgoyette } 371 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 372 1.3.2.2 pgoyette 373 1.3.2.2 pgoyette KASSERT(pool->tp_overseer.tpt_job == NULL); 374 1.3.2.2 pgoyette KASSERT(pool->tp_overseer.tpt_pool == pool); 375 1.3.2.2 pgoyette KASSERT(pool->tp_flags == THREADPOOL_DYING); 376 1.3.2.2 pgoyette KASSERT(pool->tp_refcnt == 0); 377 1.3.2.2 pgoyette KASSERT(TAILQ_EMPTY(&pool->tp_idle_threads)); 378 1.3.2.2 pgoyette KASSERT(TAILQ_EMPTY(&pool->tp_jobs)); 379 1.3.2.2 pgoyette KASSERT(!cv_has_waiters(&pool->tp_overseer.tpt_cv)); 380 1.3.2.2 pgoyette cv_destroy(&pool->tp_overseer.tpt_cv); 381 1.3.2.2 pgoyette mutex_destroy(&pool->tp_lock); 382 1.3.2.2 pgoyette } 383 1.3.2.2 pgoyette 384 1.3.2.3 pgoyette static void 385 1.3.2.3 pgoyette threadpool_hold(struct threadpool *pool) 386 1.3.2.2 pgoyette { 387 1.3.2.2 pgoyette 388 1.3.2.3 pgoyette KASSERT(mutex_owned(&pool->tp_lock)); 389 1.3.2.3 pgoyette pool->tp_refcnt++; 390 1.3.2.3 pgoyette KASSERT(pool->tp_refcnt != 0); 391 1.3.2.2 pgoyette } 392 1.3.2.2 pgoyette 393 1.3.2.2 pgoyette static void 394 1.3.2.3 pgoyette threadpool_rele(struct threadpool *pool) 395 1.3.2.2 pgoyette { 396 1.3.2.2 pgoyette 397 1.3.2.3 pgoyette KASSERT(mutex_owned(&pool->tp_lock)); 398 1.3.2.3 pgoyette KASSERT(0 < pool->tp_refcnt); 399 1.3.2.3 pgoyette if (--pool->tp_refcnt == 0) 400 1.3.2.3 pgoyette cv_broadcast(&pool->tp_overseer.tpt_cv); 401 1.3.2.2 pgoyette } 402 1.3.2.2 pgoyette 403 1.3.2.2 pgoyette /* Unbound thread pools */ 404 1.3.2.2 pgoyette 405 1.3.2.2 pgoyette int 406 1.3.2.3 pgoyette threadpool_get(struct threadpool **poolp, pri_t pri) 407 1.3.2.2 pgoyette { 408 1.3.2.2 pgoyette struct threadpool_unbound *tpu, *tmp = NULL; 409 1.3.2.2 pgoyette int error; 410 1.3.2.2 pgoyette 411 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 412 1.3.2.2 pgoyette 413 1.3.2.2 pgoyette if (! threadpool_pri_is_valid(pri)) 414 1.3.2.2 pgoyette return EINVAL; 415 1.3.2.2 pgoyette 416 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 417 1.3.2.2 pgoyette tpu = threadpool_lookup_unbound(pri); 418 1.3.2.2 pgoyette if (tpu == NULL) { 419 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 420 1.3.2.2 pgoyette TP_LOG(("%s: No pool for pri=%d, creating one.\n", 421 1.3.2.3 pgoyette __func__, (int)pri)); 422 1.3.2.3 pgoyette tmp = kmem_zalloc(sizeof(*tmp), KM_SLEEP); 423 1.3.2.3 pgoyette error = threadpool_create(&tmp->tpu_pool, NULL, pri); 424 1.3.2.3 pgoyette if (error) { 425 1.3.2.3 pgoyette kmem_free(tmp, sizeof(*tmp)); 426 1.3.2.2 pgoyette return error; 427 1.3.2.3 pgoyette } 428 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 429 1.3.2.2 pgoyette tpu = threadpool_lookup_unbound(pri); 430 1.3.2.2 pgoyette if (tpu == NULL) { 431 1.3.2.2 pgoyette TP_LOG(("%s: Won the creation race for pri=%d.\n", 432 1.3.2.3 pgoyette __func__, (int)pri)); 433 1.3.2.2 pgoyette tpu = tmp; 434 1.3.2.2 pgoyette tmp = NULL; 435 1.3.2.2 pgoyette threadpool_insert_unbound(tpu); 436 1.3.2.2 pgoyette } 437 1.3.2.2 pgoyette } 438 1.3.2.2 pgoyette KASSERT(tpu != NULL); 439 1.3.2.2 pgoyette tpu->tpu_refcnt++; 440 1.3.2.3 pgoyette KASSERT(tpu->tpu_refcnt != 0); 441 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 442 1.3.2.2 pgoyette 443 1.3.2.3 pgoyette if (tmp != NULL) { 444 1.3.2.3 pgoyette threadpool_destroy(&tmp->tpu_pool); 445 1.3.2.3 pgoyette kmem_free(tmp, sizeof(*tmp)); 446 1.3.2.3 pgoyette } 447 1.3.2.2 pgoyette KASSERT(tpu != NULL); 448 1.3.2.2 pgoyette *poolp = &tpu->tpu_pool; 449 1.3.2.2 pgoyette return 0; 450 1.3.2.2 pgoyette } 451 1.3.2.2 pgoyette 452 1.3.2.2 pgoyette void 453 1.3.2.3 pgoyette threadpool_put(struct threadpool *pool, pri_t pri) 454 1.3.2.2 pgoyette { 455 1.3.2.2 pgoyette struct threadpool_unbound *tpu = 456 1.3.2.2 pgoyette container_of(pool, struct threadpool_unbound, tpu_pool); 457 1.3.2.2 pgoyette 458 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 459 1.3.2.2 pgoyette 460 1.3.2.2 pgoyette KASSERT(threadpool_pri_is_valid(pri)); 461 1.3.2.2 pgoyette 462 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 463 1.3.2.2 pgoyette KASSERT(tpu == threadpool_lookup_unbound(pri)); 464 1.3.2.2 pgoyette KASSERT(0 < tpu->tpu_refcnt); 465 1.3.2.2 pgoyette if (--tpu->tpu_refcnt == 0) { 466 1.3.2.2 pgoyette TP_LOG(("%s: Last reference for pri=%d, destroying pool.\n", 467 1.3.2.3 pgoyette __func__, (int)pri)); 468 1.3.2.2 pgoyette threadpool_remove_unbound(tpu); 469 1.3.2.3 pgoyette } else { 470 1.3.2.2 pgoyette tpu = NULL; 471 1.3.2.3 pgoyette } 472 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 473 1.3.2.2 pgoyette 474 1.3.2.3 pgoyette if (tpu) { 475 1.3.2.3 pgoyette threadpool_destroy(&tpu->tpu_pool); 476 1.3.2.3 pgoyette kmem_free(tpu, sizeof(*tpu)); 477 1.3.2.3 pgoyette } 478 1.3.2.2 pgoyette } 479 1.3.2.2 pgoyette 480 1.3.2.2 pgoyette /* Per-CPU thread pools */ 481 1.3.2.2 pgoyette 482 1.3.2.2 pgoyette int 483 1.3.2.3 pgoyette threadpool_percpu_get(struct threadpool_percpu **pool_percpup, pri_t pri) 484 1.3.2.2 pgoyette { 485 1.3.2.3 pgoyette struct threadpool_percpu *pool_percpu, *tmp = NULL; 486 1.3.2.2 pgoyette int error; 487 1.3.2.2 pgoyette 488 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 489 1.3.2.2 pgoyette 490 1.3.2.2 pgoyette if (! threadpool_pri_is_valid(pri)) 491 1.3.2.2 pgoyette return EINVAL; 492 1.3.2.2 pgoyette 493 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 494 1.3.2.2 pgoyette pool_percpu = threadpool_lookup_percpu(pri); 495 1.3.2.2 pgoyette if (pool_percpu == NULL) { 496 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 497 1.3.2.2 pgoyette TP_LOG(("%s: No pool for pri=%d, creating one.\n", 498 1.3.2.3 pgoyette __func__, (int)pri)); 499 1.3.2.2 pgoyette error = threadpool_percpu_create(&tmp, pri); 500 1.3.2.2 pgoyette if (error) 501 1.3.2.2 pgoyette return error; 502 1.3.2.2 pgoyette KASSERT(tmp != NULL); 503 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 504 1.3.2.2 pgoyette pool_percpu = threadpool_lookup_percpu(pri); 505 1.3.2.2 pgoyette if (pool_percpu == NULL) { 506 1.3.2.2 pgoyette TP_LOG(("%s: Won the creation race for pri=%d.\n", 507 1.3.2.3 pgoyette __func__, (int)pri)); 508 1.3.2.2 pgoyette pool_percpu = tmp; 509 1.3.2.2 pgoyette tmp = NULL; 510 1.3.2.2 pgoyette threadpool_insert_percpu(pool_percpu); 511 1.3.2.2 pgoyette } 512 1.3.2.2 pgoyette } 513 1.3.2.2 pgoyette KASSERT(pool_percpu != NULL); 514 1.3.2.2 pgoyette pool_percpu->tpp_refcnt++; 515 1.3.2.3 pgoyette KASSERT(pool_percpu->tpp_refcnt != 0); 516 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 517 1.3.2.2 pgoyette 518 1.3.2.2 pgoyette if (tmp != NULL) 519 1.3.2.2 pgoyette threadpool_percpu_destroy(tmp); 520 1.3.2.2 pgoyette KASSERT(pool_percpu != NULL); 521 1.3.2.2 pgoyette *pool_percpup = pool_percpu; 522 1.3.2.2 pgoyette return 0; 523 1.3.2.2 pgoyette } 524 1.3.2.2 pgoyette 525 1.3.2.2 pgoyette void 526 1.3.2.3 pgoyette threadpool_percpu_put(struct threadpool_percpu *pool_percpu, pri_t pri) 527 1.3.2.2 pgoyette { 528 1.3.2.2 pgoyette 529 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 530 1.3.2.2 pgoyette 531 1.3.2.2 pgoyette KASSERT(threadpool_pri_is_valid(pri)); 532 1.3.2.2 pgoyette 533 1.3.2.2 pgoyette mutex_enter(&threadpools_lock); 534 1.3.2.2 pgoyette KASSERT(pool_percpu == threadpool_lookup_percpu(pri)); 535 1.3.2.2 pgoyette KASSERT(0 < pool_percpu->tpp_refcnt); 536 1.3.2.2 pgoyette if (--pool_percpu->tpp_refcnt == 0) { 537 1.3.2.2 pgoyette TP_LOG(("%s: Last reference for pri=%d, destroying pool.\n", 538 1.3.2.3 pgoyette __func__, (int)pri)); 539 1.3.2.2 pgoyette threadpool_remove_percpu(pool_percpu); 540 1.3.2.3 pgoyette } else { 541 1.3.2.2 pgoyette pool_percpu = NULL; 542 1.3.2.3 pgoyette } 543 1.3.2.2 pgoyette mutex_exit(&threadpools_lock); 544 1.3.2.2 pgoyette 545 1.3.2.2 pgoyette if (pool_percpu) 546 1.3.2.2 pgoyette threadpool_percpu_destroy(pool_percpu); 547 1.3.2.2 pgoyette } 548 1.3.2.2 pgoyette 549 1.3.2.3 pgoyette struct threadpool * 550 1.3.2.3 pgoyette threadpool_percpu_ref(struct threadpool_percpu *pool_percpu) 551 1.3.2.2 pgoyette { 552 1.3.2.3 pgoyette struct threadpool **poolp, *pool; 553 1.3.2.2 pgoyette 554 1.3.2.2 pgoyette poolp = percpu_getref(pool_percpu->tpp_percpu); 555 1.3.2.2 pgoyette pool = *poolp; 556 1.3.2.2 pgoyette percpu_putref(pool_percpu->tpp_percpu); 557 1.3.2.2 pgoyette 558 1.3.2.2 pgoyette return pool; 559 1.3.2.2 pgoyette } 560 1.3.2.2 pgoyette 561 1.3.2.3 pgoyette struct threadpool * 562 1.3.2.3 pgoyette threadpool_percpu_ref_remote(struct threadpool_percpu *pool_percpu, 563 1.3.2.2 pgoyette struct cpu_info *ci) 564 1.3.2.2 pgoyette { 565 1.3.2.3 pgoyette struct threadpool **poolp, *pool; 566 1.3.2.2 pgoyette 567 1.3.2.2 pgoyette percpu_traverse_enter(); 568 1.3.2.2 pgoyette poolp = percpu_getptr_remote(pool_percpu->tpp_percpu, ci); 569 1.3.2.2 pgoyette pool = *poolp; 570 1.3.2.2 pgoyette percpu_traverse_exit(); 571 1.3.2.2 pgoyette 572 1.3.2.2 pgoyette return pool; 573 1.3.2.2 pgoyette } 574 1.3.2.2 pgoyette 575 1.3.2.2 pgoyette static int 576 1.3.2.3 pgoyette threadpool_percpu_create(struct threadpool_percpu **pool_percpup, pri_t pri) 577 1.3.2.2 pgoyette { 578 1.3.2.3 pgoyette struct threadpool_percpu *pool_percpu; 579 1.3.2.2 pgoyette struct cpu_info *ci; 580 1.3.2.2 pgoyette CPU_INFO_ITERATOR cii; 581 1.3.2.2 pgoyette unsigned int i, j; 582 1.3.2.2 pgoyette int error; 583 1.3.2.2 pgoyette 584 1.3.2.2 pgoyette pool_percpu = kmem_zalloc(sizeof(*pool_percpu), KM_SLEEP); 585 1.3.2.2 pgoyette if (pool_percpu == NULL) { 586 1.3.2.2 pgoyette error = ENOMEM; 587 1.3.2.2 pgoyette goto fail0; 588 1.3.2.2 pgoyette } 589 1.3.2.2 pgoyette pool_percpu->tpp_pri = pri; 590 1.3.2.2 pgoyette 591 1.3.2.3 pgoyette pool_percpu->tpp_percpu = percpu_alloc(sizeof(struct threadpool *)); 592 1.3.2.2 pgoyette if (pool_percpu->tpp_percpu == NULL) { 593 1.3.2.2 pgoyette error = ENOMEM; 594 1.3.2.2 pgoyette goto fail1; 595 1.3.2.2 pgoyette } 596 1.3.2.2 pgoyette 597 1.3.2.2 pgoyette for (i = 0, CPU_INFO_FOREACH(cii, ci), i++) { 598 1.3.2.3 pgoyette struct threadpool *pool; 599 1.3.2.2 pgoyette 600 1.3.2.3 pgoyette pool = kmem_zalloc(sizeof(*pool), KM_SLEEP); 601 1.3.2.3 pgoyette error = threadpool_create(pool, ci, pri); 602 1.3.2.3 pgoyette if (error) { 603 1.3.2.3 pgoyette kmem_free(pool, sizeof(*pool)); 604 1.3.2.2 pgoyette goto fail2; 605 1.3.2.3 pgoyette } 606 1.3.2.2 pgoyette percpu_traverse_enter(); 607 1.3.2.3 pgoyette struct threadpool **const poolp = 608 1.3.2.2 pgoyette percpu_getptr_remote(pool_percpu->tpp_percpu, ci); 609 1.3.2.2 pgoyette *poolp = pool; 610 1.3.2.2 pgoyette percpu_traverse_exit(); 611 1.3.2.2 pgoyette } 612 1.3.2.2 pgoyette 613 1.3.2.2 pgoyette /* Success! */ 614 1.3.2.3 pgoyette *pool_percpup = (struct threadpool_percpu *)pool_percpu; 615 1.3.2.2 pgoyette return 0; 616 1.3.2.2 pgoyette 617 1.3.2.2 pgoyette fail2: for (j = 0, CPU_INFO_FOREACH(cii, ci), j++) { 618 1.3.2.2 pgoyette if (i <= j) 619 1.3.2.2 pgoyette break; 620 1.3.2.2 pgoyette percpu_traverse_enter(); 621 1.3.2.3 pgoyette struct threadpool **const poolp = 622 1.3.2.2 pgoyette percpu_getptr_remote(pool_percpu->tpp_percpu, ci); 623 1.3.2.3 pgoyette struct threadpool *const pool = *poolp; 624 1.3.2.2 pgoyette percpu_traverse_exit(); 625 1.3.2.3 pgoyette threadpool_destroy(pool); 626 1.3.2.3 pgoyette kmem_free(pool, sizeof(*pool)); 627 1.3.2.2 pgoyette } 628 1.3.2.2 pgoyette percpu_free(pool_percpu->tpp_percpu, sizeof(struct taskthread_pool *)); 629 1.3.2.2 pgoyette fail1: kmem_free(pool_percpu, sizeof(*pool_percpu)); 630 1.3.2.2 pgoyette fail0: return error; 631 1.3.2.2 pgoyette } 632 1.3.2.2 pgoyette 633 1.3.2.2 pgoyette static void 634 1.3.2.3 pgoyette threadpool_percpu_destroy(struct threadpool_percpu *pool_percpu) 635 1.3.2.2 pgoyette { 636 1.3.2.2 pgoyette struct cpu_info *ci; 637 1.3.2.2 pgoyette CPU_INFO_ITERATOR cii; 638 1.3.2.2 pgoyette 639 1.3.2.2 pgoyette for (CPU_INFO_FOREACH(cii, ci)) { 640 1.3.2.2 pgoyette percpu_traverse_enter(); 641 1.3.2.3 pgoyette struct threadpool **const poolp = 642 1.3.2.2 pgoyette percpu_getptr_remote(pool_percpu->tpp_percpu, ci); 643 1.3.2.3 pgoyette struct threadpool *const pool = *poolp; 644 1.3.2.2 pgoyette percpu_traverse_exit(); 645 1.3.2.3 pgoyette threadpool_destroy(pool); 646 1.3.2.3 pgoyette kmem_free(pool, sizeof(*pool)); 647 1.3.2.2 pgoyette } 648 1.3.2.2 pgoyette 649 1.3.2.3 pgoyette percpu_free(pool_percpu->tpp_percpu, sizeof(struct threadpool *)); 650 1.3.2.2 pgoyette kmem_free(pool_percpu, sizeof(*pool_percpu)); 651 1.3.2.2 pgoyette } 652 1.3.2.2 pgoyette 653 1.3.2.2 pgoyette /* Thread pool jobs */ 654 1.3.2.2 pgoyette 655 1.3.2.2 pgoyette void __printflike(4,5) 656 1.3.2.3 pgoyette threadpool_job_init(struct threadpool_job *job, threadpool_job_fn_t fn, 657 1.3.2.2 pgoyette kmutex_t *lock, const char *fmt, ...) 658 1.3.2.2 pgoyette { 659 1.3.2.2 pgoyette va_list ap; 660 1.3.2.2 pgoyette 661 1.3.2.2 pgoyette va_start(ap, fmt); 662 1.3.2.2 pgoyette (void)vsnprintf(job->job_name, sizeof(job->job_name), fmt, ap); 663 1.3.2.2 pgoyette va_end(ap); 664 1.3.2.2 pgoyette 665 1.3.2.2 pgoyette job->job_lock = lock; 666 1.3.2.2 pgoyette job->job_thread = NULL; 667 1.3.2.2 pgoyette job->job_refcnt = 0; 668 1.3.2.2 pgoyette cv_init(&job->job_cv, job->job_name); 669 1.3.2.2 pgoyette job->job_fn = fn; 670 1.3.2.2 pgoyette } 671 1.3.2.2 pgoyette 672 1.3.2.2 pgoyette static void 673 1.3.2.3 pgoyette threadpool_job_dead(struct threadpool_job *job) 674 1.3.2.2 pgoyette { 675 1.3.2.2 pgoyette 676 1.3.2.3 pgoyette panic("threadpool job %p ran after destruction", job); 677 1.3.2.2 pgoyette } 678 1.3.2.2 pgoyette 679 1.3.2.2 pgoyette void 680 1.3.2.3 pgoyette threadpool_job_destroy(struct threadpool_job *job) 681 1.3.2.2 pgoyette { 682 1.3.2.2 pgoyette 683 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 684 1.3.2.2 pgoyette 685 1.3.2.2 pgoyette KASSERTMSG((job->job_thread == NULL), "job %p still running", job); 686 1.3.2.2 pgoyette 687 1.3.2.2 pgoyette mutex_enter(job->job_lock); 688 1.3.2.2 pgoyette while (0 < job->job_refcnt) 689 1.3.2.2 pgoyette cv_wait(&job->job_cv, job->job_lock); 690 1.3.2.2 pgoyette mutex_exit(job->job_lock); 691 1.3.2.2 pgoyette 692 1.3.2.2 pgoyette job->job_lock = NULL; 693 1.3.2.2 pgoyette KASSERT(job->job_thread == NULL); 694 1.3.2.2 pgoyette KASSERT(job->job_refcnt == 0); 695 1.3.2.2 pgoyette KASSERT(!cv_has_waiters(&job->job_cv)); 696 1.3.2.2 pgoyette cv_destroy(&job->job_cv); 697 1.3.2.2 pgoyette job->job_fn = threadpool_job_dead; 698 1.3.2.2 pgoyette (void)strlcpy(job->job_name, "deadjob", sizeof(job->job_name)); 699 1.3.2.2 pgoyette } 700 1.3.2.2 pgoyette 701 1.3.2.3 pgoyette static void 702 1.3.2.3 pgoyette threadpool_job_hold(struct threadpool_job *job) 703 1.3.2.2 pgoyette { 704 1.3.2.2 pgoyette unsigned int refcnt; 705 1.3.2.3 pgoyette 706 1.3.2.2 pgoyette do { 707 1.3.2.2 pgoyette refcnt = job->job_refcnt; 708 1.3.2.3 pgoyette KASSERT(refcnt != UINT_MAX); 709 1.3.2.2 pgoyette } while (atomic_cas_uint(&job->job_refcnt, refcnt, (refcnt + 1)) 710 1.3.2.2 pgoyette != refcnt); 711 1.3.2.2 pgoyette } 712 1.3.2.2 pgoyette 713 1.3.2.2 pgoyette static void 714 1.3.2.3 pgoyette threadpool_job_rele(struct threadpool_job *job) 715 1.3.2.2 pgoyette { 716 1.3.2.2 pgoyette unsigned int refcnt; 717 1.3.2.2 pgoyette 718 1.3.2.3 pgoyette KASSERT(mutex_owned(job->job_lock)); 719 1.3.2.3 pgoyette 720 1.3.2.2 pgoyette do { 721 1.3.2.2 pgoyette refcnt = job->job_refcnt; 722 1.3.2.2 pgoyette KASSERT(0 < refcnt); 723 1.3.2.2 pgoyette if (refcnt == 1) { 724 1.3.2.2 pgoyette refcnt = atomic_dec_uint_nv(&job->job_refcnt); 725 1.3.2.2 pgoyette KASSERT(refcnt != UINT_MAX); 726 1.3.2.2 pgoyette if (refcnt == 0) 727 1.3.2.2 pgoyette cv_broadcast(&job->job_cv); 728 1.3.2.2 pgoyette return; 729 1.3.2.2 pgoyette } 730 1.3.2.2 pgoyette } while (atomic_cas_uint(&job->job_refcnt, refcnt, (refcnt - 1)) 731 1.3.2.2 pgoyette != refcnt); 732 1.3.2.2 pgoyette } 733 1.3.2.2 pgoyette 734 1.3.2.2 pgoyette void 735 1.3.2.3 pgoyette threadpool_job_done(struct threadpool_job *job) 736 1.3.2.2 pgoyette { 737 1.3.2.2 pgoyette 738 1.3.2.2 pgoyette KASSERT(mutex_owned(job->job_lock)); 739 1.3.2.2 pgoyette KASSERT(job->job_thread != NULL); 740 1.3.2.2 pgoyette KASSERT(job->job_thread->tpt_lwp == curlwp); 741 1.3.2.2 pgoyette 742 1.3.2.3 pgoyette /* 743 1.3.2.3 pgoyette * We can safely read this field; it's only modified right before 744 1.3.2.3 pgoyette * we call the job work function, and we are only preserving it 745 1.3.2.3 pgoyette * to use here; no one cares if it contains junk afterward. 746 1.3.2.3 pgoyette */ 747 1.3.2.3 pgoyette lwp_lock(curlwp); 748 1.3.2.3 pgoyette curlwp->l_name = job->job_thread->tpt_lwp_savedname; 749 1.3.2.3 pgoyette lwp_unlock(curlwp); 750 1.3.2.3 pgoyette 751 1.3.2.3 pgoyette /* 752 1.3.2.3 pgoyette * Inline the work of threadpool_job_rele(); the job is already 753 1.3.2.3 pgoyette * locked, the most likely scenario (XXXJRT only scenario?) is 754 1.3.2.3 pgoyette * that we're dropping the last reference (the one taken in 755 1.3.2.3 pgoyette * threadpool_schedule_job()), and we always do the cv_broadcast() 756 1.3.2.3 pgoyette * anyway. 757 1.3.2.3 pgoyette */ 758 1.3.2.3 pgoyette KASSERT(0 < job->job_refcnt); 759 1.3.2.3 pgoyette unsigned int refcnt __diagused = atomic_dec_uint_nv(&job->job_refcnt); 760 1.3.2.3 pgoyette KASSERT(refcnt != UINT_MAX); 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.3 pgoyette threadpool_schedule_job(struct threadpool *pool, struct threadpool_job *job) 767 1.3.2.2 pgoyette { 768 1.3.2.2 pgoyette 769 1.3.2.2 pgoyette KASSERT(mutex_owned(job->job_lock)); 770 1.3.2.2 pgoyette 771 1.3.2.2 pgoyette /* 772 1.3.2.2 pgoyette * If the job's already running, let it keep running. The job 773 1.3.2.2 pgoyette * is guaranteed by the interlock not to end early -- if it had 774 1.3.2.2 pgoyette * ended early, threadpool_job_done would have set job_thread 775 1.3.2.2 pgoyette * to NULL under the interlock. 776 1.3.2.2 pgoyette */ 777 1.3.2.2 pgoyette if (__predict_true(job->job_thread != NULL)) { 778 1.3.2.2 pgoyette TP_LOG(("%s: job '%s' already runnining.\n", 779 1.3.2.3 pgoyette __func__, job->job_name)); 780 1.3.2.2 pgoyette return; 781 1.3.2.2 pgoyette } 782 1.3.2.2 pgoyette 783 1.3.2.3 pgoyette threadpool_job_hold(job); 784 1.3.2.3 pgoyette 785 1.3.2.2 pgoyette /* Otherwise, try to assign a thread to the job. */ 786 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 787 1.3.2.2 pgoyette if (__predict_false(TAILQ_EMPTY(&pool->tp_idle_threads))) { 788 1.3.2.2 pgoyette /* Nobody's idle. Give it to the overseer. */ 789 1.3.2.2 pgoyette TP_LOG(("%s: giving job '%s' to overseer.\n", 790 1.3.2.3 pgoyette __func__, job->job_name)); 791 1.3.2.2 pgoyette job->job_thread = &pool->tp_overseer; 792 1.3.2.2 pgoyette TAILQ_INSERT_TAIL(&pool->tp_jobs, job, job_entry); 793 1.3.2.2 pgoyette } else { 794 1.3.2.2 pgoyette /* Assign it to the first idle thread. */ 795 1.3.2.2 pgoyette job->job_thread = TAILQ_FIRST(&pool->tp_idle_threads); 796 1.3.2.2 pgoyette TP_LOG(("%s: giving job '%s' to idle thread %p.\n", 797 1.3.2.3 pgoyette __func__, job->job_name, job->job_thread)); 798 1.3.2.2 pgoyette TAILQ_REMOVE(&pool->tp_idle_threads, job->job_thread, 799 1.3.2.2 pgoyette tpt_entry); 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.3 pgoyette threadpool_cancel_job_async(struct threadpool *pool, struct threadpool_job *job) 811 1.3.2.2 pgoyette { 812 1.3.2.2 pgoyette 813 1.3.2.2 pgoyette KASSERT(mutex_owned(job->job_lock)); 814 1.3.2.2 pgoyette 815 1.3.2.2 pgoyette /* 816 1.3.2.2 pgoyette * XXXJRT This fails (albeit safely) when all of the following 817 1.3.2.2 pgoyette * are true: 818 1.3.2.2 pgoyette * 819 1.3.2.2 pgoyette * => "pool" is something other than what the job was 820 1.3.2.2 pgoyette * scheduled on. This can legitimately occur if, 821 1.3.2.2 pgoyette * for example, a job is percpu-scheduled on CPU0 822 1.3.2.2 pgoyette * and then CPU1 attempts to cancel it without taking 823 1.3.2.2 pgoyette * a remote pool reference. (this might happen by 824 1.3.2.2 pgoyette * "luck of the draw"). 825 1.3.2.2 pgoyette * 826 1.3.2.2 pgoyette * => "job" is not yet running, but is assigned to the 827 1.3.2.2 pgoyette * overseer. 828 1.3.2.2 pgoyette * 829 1.3.2.2 pgoyette * When this happens, this code makes the determination that 830 1.3.2.2 pgoyette * the job is already running. The failure mode is that the 831 1.3.2.2 pgoyette * caller is told the job is running, and thus has to wait. 832 1.3.2.2 pgoyette * The overseer will eventually get to it and the job will 833 1.3.2.2 pgoyette * proceed as if it had been already running. 834 1.3.2.2 pgoyette */ 835 1.3.2.2 pgoyette 836 1.3.2.2 pgoyette if (job->job_thread == NULL) { 837 1.3.2.2 pgoyette /* Nothing to do. Guaranteed not running. */ 838 1.3.2.2 pgoyette return true; 839 1.3.2.2 pgoyette } else if (job->job_thread == &pool->tp_overseer) { 840 1.3.2.2 pgoyette /* Take it off the list to guarantee it won't run. */ 841 1.3.2.2 pgoyette job->job_thread = NULL; 842 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 843 1.3.2.2 pgoyette TAILQ_REMOVE(&pool->tp_jobs, job, job_entry); 844 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 845 1.3.2.3 pgoyette threadpool_job_rele(job); 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.3 pgoyette threadpool_cancel_job(struct threadpool *pool, struct threadpool_job *job) 855 1.3.2.2 pgoyette { 856 1.3.2.2 pgoyette 857 1.3.2.2 pgoyette ASSERT_SLEEPABLE(); 858 1.3.2.2 pgoyette 859 1.3.2.2 pgoyette KASSERT(mutex_owned(job->job_lock)); 860 1.3.2.2 pgoyette 861 1.3.2.3 pgoyette if (threadpool_cancel_job_async(pool, job)) 862 1.3.2.2 pgoyette return; 863 1.3.2.2 pgoyette 864 1.3.2.2 pgoyette /* Already running. Wait for it to complete. */ 865 1.3.2.2 pgoyette while (job->job_thread != NULL) 866 1.3.2.2 pgoyette cv_wait(&job->job_cv, job->job_lock); 867 1.3.2.2 pgoyette } 868 1.3.2.2 pgoyette 869 1.3.2.2 pgoyette /* Thread pool overseer thread */ 870 1.3.2.2 pgoyette 871 1.3.2.2 pgoyette static void __dead 872 1.3.2.2 pgoyette threadpool_overseer_thread(void *arg) 873 1.3.2.2 pgoyette { 874 1.3.2.2 pgoyette struct threadpool_thread *const overseer = arg; 875 1.3.2.3 pgoyette struct threadpool *const pool = overseer->tpt_pool; 876 1.3.2.2 pgoyette struct lwp *lwp = NULL; 877 1.3.2.2 pgoyette int ktflags; 878 1.3.2.2 pgoyette int error; 879 1.3.2.2 pgoyette 880 1.3.2.2 pgoyette KASSERT((pool->tp_cpu == NULL) || (pool->tp_cpu == curcpu())); 881 1.3.2.2 pgoyette 882 1.3.2.2 pgoyette /* Wait until we're initialized. */ 883 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 884 1.3.2.2 pgoyette while (overseer->tpt_lwp == NULL) 885 1.3.2.2 pgoyette cv_wait(&overseer->tpt_cv, &pool->tp_lock); 886 1.3.2.2 pgoyette 887 1.3.2.2 pgoyette TP_LOG(("%s: starting.\n", __func__)); 888 1.3.2.2 pgoyette 889 1.3.2.2 pgoyette for (;;) { 890 1.3.2.2 pgoyette /* Wait until there's a job. */ 891 1.3.2.2 pgoyette while (TAILQ_EMPTY(&pool->tp_jobs)) { 892 1.3.2.2 pgoyette if (ISSET(pool->tp_flags, THREADPOOL_DYING)) { 893 1.3.2.2 pgoyette TP_LOG(("%s: THREADPOOL_DYING\n", 894 1.3.2.3 pgoyette __func__)); 895 1.3.2.2 pgoyette break; 896 1.3.2.2 pgoyette } 897 1.3.2.2 pgoyette cv_wait(&overseer->tpt_cv, &pool->tp_lock); 898 1.3.2.2 pgoyette } 899 1.3.2.2 pgoyette if (__predict_false(TAILQ_EMPTY(&pool->tp_jobs))) 900 1.3.2.2 pgoyette break; 901 1.3.2.2 pgoyette 902 1.3.2.2 pgoyette /* If there are no threads, we'll have to try to start one. */ 903 1.3.2.2 pgoyette if (TAILQ_EMPTY(&pool->tp_idle_threads)) { 904 1.3.2.2 pgoyette TP_LOG(("%s: Got a job, need to create a thread.\n", 905 1.3.2.3 pgoyette __func__)); 906 1.3.2.3 pgoyette threadpool_hold(pool); 907 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 908 1.3.2.2 pgoyette 909 1.3.2.2 pgoyette struct threadpool_thread *const thread = 910 1.3.2.2 pgoyette pool_cache_get(threadpool_thread_pc, PR_WAITOK); 911 1.3.2.2 pgoyette thread->tpt_lwp = NULL; 912 1.3.2.2 pgoyette thread->tpt_pool = pool; 913 1.3.2.2 pgoyette thread->tpt_job = NULL; 914 1.3.2.2 pgoyette cv_init(&thread->tpt_cv, "poolthrd"); 915 1.3.2.2 pgoyette 916 1.3.2.2 pgoyette ktflags = 0; 917 1.3.2.2 pgoyette ktflags |= KTHREAD_MPSAFE; 918 1.3.2.2 pgoyette if (pool->tp_pri < PRI_KERNEL) 919 1.3.2.2 pgoyette ktflags |= KTHREAD_TS; 920 1.3.2.2 pgoyette error = kthread_create(pool->tp_pri, ktflags, 921 1.3.2.2 pgoyette pool->tp_cpu, &threadpool_thread, thread, &lwp, 922 1.3.2.2 pgoyette "poolthread/%d@%d", 923 1.3.2.2 pgoyette (pool->tp_cpu ? cpu_index(pool->tp_cpu) : -1), 924 1.3.2.2 pgoyette (int)pool->tp_pri); 925 1.3.2.2 pgoyette 926 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 927 1.3.2.2 pgoyette if (error) { 928 1.3.2.2 pgoyette pool_cache_put(threadpool_thread_pc, thread); 929 1.3.2.2 pgoyette threadpool_rele(pool); 930 1.3.2.2 pgoyette /* XXX What to do to wait for memory? */ 931 1.3.2.2 pgoyette (void)kpause("thrdplcr", false, hz, 932 1.3.2.2 pgoyette &pool->tp_lock); 933 1.3.2.2 pgoyette continue; 934 1.3.2.2 pgoyette } 935 1.3.2.3 pgoyette /* 936 1.3.2.3 pgoyette * New kthread now owns the reference to the pool 937 1.3.2.3 pgoyette * taken above. 938 1.3.2.3 pgoyette */ 939 1.3.2.2 pgoyette KASSERT(lwp != NULL); 940 1.3.2.2 pgoyette TAILQ_INSERT_TAIL(&pool->tp_idle_threads, thread, 941 1.3.2.2 pgoyette tpt_entry); 942 1.3.2.2 pgoyette thread->tpt_lwp = lwp; 943 1.3.2.2 pgoyette lwp = NULL; 944 1.3.2.2 pgoyette cv_broadcast(&thread->tpt_cv); 945 1.3.2.2 pgoyette continue; 946 1.3.2.2 pgoyette } 947 1.3.2.2 pgoyette 948 1.3.2.2 pgoyette /* There are idle threads, so try giving one a job. */ 949 1.3.2.3 pgoyette struct threadpool_job *const job = TAILQ_FIRST(&pool->tp_jobs); 950 1.3.2.2 pgoyette TAILQ_REMOVE(&pool->tp_jobs, job, job_entry); 951 1.3.2.3 pgoyette /* 952 1.3.2.3 pgoyette * Take an extra reference on the job temporarily so that 953 1.3.2.3 pgoyette * it won't disappear on us while we have both locks dropped. 954 1.3.2.3 pgoyette */ 955 1.3.2.3 pgoyette threadpool_job_hold(job); 956 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 957 1.3.2.2 pgoyette 958 1.3.2.2 pgoyette mutex_enter(job->job_lock); 959 1.3.2.2 pgoyette /* If the job was cancelled, we'll no longer be its thread. */ 960 1.3.2.2 pgoyette if (__predict_true(job->job_thread == overseer)) { 961 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 962 1.3.2.2 pgoyette if (__predict_false( 963 1.3.2.2 pgoyette TAILQ_EMPTY(&pool->tp_idle_threads))) { 964 1.3.2.2 pgoyette /* 965 1.3.2.2 pgoyette * Someone else snagged the thread 966 1.3.2.2 pgoyette * first. We'll have to try again. 967 1.3.2.2 pgoyette */ 968 1.3.2.2 pgoyette TP_LOG(("%s: '%s' lost race to use idle thread.\n", 969 1.3.2.3 pgoyette __func__, job->job_name)); 970 1.3.2.2 pgoyette TAILQ_INSERT_HEAD(&pool->tp_jobs, job, 971 1.3.2.2 pgoyette job_entry); 972 1.3.2.2 pgoyette } else { 973 1.3.2.2 pgoyette /* 974 1.3.2.2 pgoyette * Assign the job to the thread and 975 1.3.2.2 pgoyette * wake the thread so it starts work. 976 1.3.2.2 pgoyette */ 977 1.3.2.2 pgoyette struct threadpool_thread *const thread = 978 1.3.2.2 pgoyette TAILQ_FIRST(&pool->tp_idle_threads); 979 1.3.2.2 pgoyette 980 1.3.2.2 pgoyette TP_LOG(("%s: '%s' gets thread %p\n", 981 1.3.2.3 pgoyette __func__, job->job_name, thread)); 982 1.3.2.2 pgoyette KASSERT(thread->tpt_job == NULL); 983 1.3.2.2 pgoyette TAILQ_REMOVE(&pool->tp_idle_threads, thread, 984 1.3.2.2 pgoyette tpt_entry); 985 1.3.2.2 pgoyette thread->tpt_job = job; 986 1.3.2.2 pgoyette job->job_thread = thread; 987 1.3.2.2 pgoyette cv_broadcast(&thread->tpt_cv); 988 1.3.2.2 pgoyette } 989 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 990 1.3.2.2 pgoyette } 991 1.3.2.3 pgoyette threadpool_job_rele(job); 992 1.3.2.2 pgoyette mutex_exit(job->job_lock); 993 1.3.2.2 pgoyette 994 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 995 1.3.2.2 pgoyette } 996 1.3.2.3 pgoyette threadpool_rele(pool); 997 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 998 1.3.2.2 pgoyette 999 1.3.2.2 pgoyette TP_LOG(("%s: exiting.\n", __func__)); 1000 1.3.2.2 pgoyette 1001 1.3.2.2 pgoyette kthread_exit(0); 1002 1.3.2.2 pgoyette } 1003 1.3.2.2 pgoyette 1004 1.3.2.2 pgoyette /* Thread pool thread */ 1005 1.3.2.2 pgoyette 1006 1.3.2.2 pgoyette static void __dead 1007 1.3.2.2 pgoyette threadpool_thread(void *arg) 1008 1.3.2.2 pgoyette { 1009 1.3.2.2 pgoyette struct threadpool_thread *const thread = arg; 1010 1.3.2.3 pgoyette struct threadpool *const pool = thread->tpt_pool; 1011 1.3.2.2 pgoyette 1012 1.3.2.2 pgoyette KASSERT((pool->tp_cpu == NULL) || (pool->tp_cpu == curcpu())); 1013 1.3.2.2 pgoyette 1014 1.3.2.2 pgoyette /* Wait until we're initialized and on the queue. */ 1015 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 1016 1.3.2.2 pgoyette while (thread->tpt_lwp == NULL) 1017 1.3.2.2 pgoyette cv_wait(&thread->tpt_cv, &pool->tp_lock); 1018 1.3.2.2 pgoyette 1019 1.3.2.2 pgoyette TP_LOG(("%s: starting.\n", __func__)); 1020 1.3.2.2 pgoyette 1021 1.3.2.2 pgoyette KASSERT(thread->tpt_lwp == curlwp); 1022 1.3.2.2 pgoyette for (;;) { 1023 1.3.2.2 pgoyette /* Wait until we are assigned a job. */ 1024 1.3.2.2 pgoyette while (thread->tpt_job == NULL) { 1025 1.3.2.2 pgoyette if (ISSET(pool->tp_flags, THREADPOOL_DYING)) { 1026 1.3.2.2 pgoyette TP_LOG(("%s: THREADPOOL_DYING\n", 1027 1.3.2.3 pgoyette __func__)); 1028 1.3.2.2 pgoyette break; 1029 1.3.2.2 pgoyette } 1030 1.3.2.2 pgoyette if (cv_timedwait(&thread->tpt_cv, &pool->tp_lock, 1031 1.3.2.3 pgoyette mstohz(threadpool_idle_time_ms))) 1032 1.3.2.2 pgoyette break; 1033 1.3.2.2 pgoyette } 1034 1.3.2.2 pgoyette if (__predict_false(thread->tpt_job == NULL)) { 1035 1.3.2.2 pgoyette TAILQ_REMOVE(&pool->tp_idle_threads, thread, 1036 1.3.2.2 pgoyette tpt_entry); 1037 1.3.2.2 pgoyette break; 1038 1.3.2.2 pgoyette } 1039 1.3.2.2 pgoyette 1040 1.3.2.3 pgoyette struct threadpool_job *const job = thread->tpt_job; 1041 1.3.2.2 pgoyette KASSERT(job != NULL); 1042 1.3.2.2 pgoyette 1043 1.3.2.2 pgoyette /* Set our lwp name to reflect what job we're doing. */ 1044 1.3.2.2 pgoyette lwp_lock(curlwp); 1045 1.3.2.3 pgoyette char *const lwp_name __diagused = curlwp->l_name; 1046 1.3.2.3 pgoyette thread->tpt_lwp_savedname = curlwp->l_name; 1047 1.3.2.2 pgoyette curlwp->l_name = job->job_name; 1048 1.3.2.2 pgoyette lwp_unlock(curlwp); 1049 1.3.2.2 pgoyette 1050 1.3.2.3 pgoyette mutex_spin_exit(&pool->tp_lock); 1051 1.3.2.2 pgoyette 1052 1.3.2.3 pgoyette TP_LOG(("%s: running job '%s' on thread %p.\n", 1053 1.3.2.3 pgoyette __func__, job->job_name, thread)); 1054 1.3.2.2 pgoyette 1055 1.3.2.3 pgoyette /* Run the job. */ 1056 1.3.2.3 pgoyette (*job->job_fn)(job); 1057 1.3.2.3 pgoyette 1058 1.3.2.3 pgoyette /* lwp name restored in threadpool_job_done(). */ 1059 1.3.2.3 pgoyette KASSERTMSG((curlwp->l_name == lwp_name), 1060 1.3.2.3 pgoyette "someone forgot to call threadpool_job_done()!"); 1061 1.3.2.3 pgoyette 1062 1.3.2.3 pgoyette /* 1063 1.3.2.3 pgoyette * We can compare pointers, but we can no longer deference 1064 1.3.2.3 pgoyette * job after this because threadpool_job_done() drops the 1065 1.3.2.3 pgoyette * last reference on the job while the job is locked. 1066 1.3.2.3 pgoyette */ 1067 1.3.2.2 pgoyette 1068 1.3.2.2 pgoyette mutex_spin_enter(&pool->tp_lock); 1069 1.3.2.2 pgoyette KASSERT(thread->tpt_job == job); 1070 1.3.2.2 pgoyette thread->tpt_job = NULL; 1071 1.3.2.2 pgoyette TAILQ_INSERT_TAIL(&pool->tp_idle_threads, thread, tpt_entry); 1072 1.3.2.2 pgoyette } 1073 1.3.2.3 pgoyette threadpool_rele(pool); 1074 1.3.2.2 pgoyette mutex_spin_exit(&pool->tp_lock); 1075 1.3.2.2 pgoyette 1076 1.3.2.2 pgoyette TP_LOG(("%s: thread %p exiting.\n", __func__, thread)); 1077 1.3.2.2 pgoyette 1078 1.3.2.2 pgoyette KASSERT(!cv_has_waiters(&thread->tpt_cv)); 1079 1.3.2.2 pgoyette cv_destroy(&thread->tpt_cv); 1080 1.3.2.2 pgoyette pool_cache_put(threadpool_thread_pc, thread); 1081 1.3.2.2 pgoyette kthread_exit(0); 1082 1.3.2.2 pgoyette } 1083
Indexes created Mon Sep 22 10:09:38 GMT 2025