1 /*
2 * CDDL HEADER START
3 *
4 * The contents of this file are subject to the terms of the
5 * Common Development and Distribution License, Version 1.0 only
6 * (the "License"). You may not use this file except in compliance
7 * with the License.
8 *
9 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
10 * or http://www.opensolaris.org/os/licensing.
11 * See the License for the specific language governing permissions
12 * and limitations under the License.
13 *
14 * When distributing Covered Code, include this CDDL HEADER in each
15 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
16 * If applicable, add the following below this CDDL HEADER, with the
17 * fields enclosed by brackets "[]" replaced with your own identifying
18 * information: Portions Copyright [yyyy] [name of copyright owner]
19 *
20 * CDDL HEADER END
21 */
22 /*
23 * Copyright 2004 Sun Microsystems, Inc. All rights reserved.
24 * Use is subject to license terms.
25 */
26
27 #pragma ident "%Z%%M% %I% %E% SMI"
28
29 /*
30 * Kernel protection serializers: general purpose synchronization mechanism.
31 *
32 * Serializers provide a simple way to serialize access to some resource. They
33 * can be used as an alternative to locks or STREAMS perimeters. They scale
34 * much better than STREAMS outer serializers.
35 *
36 * Serializer is an abstraction that guarantees that all functions executed
37 * within the serializer are serialized: they are executed in the order they
38 * entered serializer one at a time.
39 *
40 * INTERFACES:
41 *
42 * serializer_t *serializer_create(flags);
43 *
44 * Create a serializer. The flags may be either SER_SLEEP or SER_NOSLEEP
45 * which are the same as KM_SLEEP and KM_NOSLEEP respectively.
46 *
47 * serializer_enter(serializer, proc, mblk, arg);
48 *
49 * Execute 'proc(mblk, arg)' within the serializer.
50 *
51 * serializer_wait(serializer);
52 *
53 * Wait for pending serializer jobs to complete. This function should never
54 * be called within the serializer or it will deadlock.
55 *
56 * serializer_destroy(serializer);
57 *
58 * Destroy serializer.
59 *
60 * Serializers export three DTrace SDT probes:
61 *
62 * serializer-enqueue(serializer, mblk, arg, proc)
63 *
64 * The probe triggers when serializer is busy and the request is
65 * queued.
66 *
67 * serializer-exec-start(serializer, mblk, arg, proc)
68 *
69 * The probe triggers before the request is executed
70 *
71 * serializer-exec-end(serializer, mblk, arg, proc)
72 *
73 * The probe triggers after the request is executed
74 *
75 *
76 * IMPLEMENTATION.
77 *
78 * Serializer consists of a "owner" and a list of queued jobs. The first thread
79 * entering serializer sets the owner and executes its job directly without
80 * context switch. Then it processes jobs which may have been enqueued while it
81 * was executing a job and drops the owner, leaving the serializer empty. Any
82 * thread entering an owned serializer enqueues its job and returns immediately.
83 *
84 * Serializer data structure holds several fields used for debugging only. They
85 * are not relevant for the proper serializer functioning.
86 *
87 * When new requests arrive faster then they are processed it is possible that a
88 * thread that started processing serializer will continue doing so for a long
89 * time. To avoid such pathological behavior the amount of requests drained by
90 * serializer_enter() is limited by `serializer_credit' value. After the credit
91 * is expired serializer_enter() schedules a taskq request to continue draining.
92 * The taskq thread draining is not limited by serializer_credit. Note that it
93 * is possible that another serializer_enter() will drain the serializer before
94 * a taskq thread will get to it.
95 */
96
97 #include <sys/types.h>
98 #include <sys/kmem.h>
99 #include <sys/thread.h>
100 #include <sys/mutex.h>
101 #include <sys/systm.h>
102 #include <sys/debug.h>
103 #include <sys/taskq.h>
104 #include <sys/sdt.h>
105 #include <sys/serializer.h>
106
107 #define SERIALIZER_NAMELEN 31
108
109 /*
110 * Serializer abstraction.
111 * Fields marked (D) are used for debugging purposes only.
112 */
113 struct serializer_s {
114 kmutex_t ser_lock; /* Protects state and the list */
115 kthread_t *ser_owner; /* Thread executing serializer */
116 ushort_t ser_taskq; /* Serializer scheduled for taskq */
117 kcondvar_t ser_cv; /* For serializer-wait */
118 uint_t ser_count; /* # of queued requests (D) */
119 mblk_t *ser_first; /* First message in the queue */
120 mblk_t *ser_last; /* Last message in the queue */
121 srproc_t *ser_proc; /* Currently executing proc (D) */
122 mblk_t *ser_curr; /* Currently executing msg (D) */
123 void *ser_arg; /* Currently executing arg (D) */
124 };
125
126 static kmem_cache_t *serializer_cache;
127
128 /*
129 * How many drains are allowed before we switch to taskq processing.
130 */
131 #define SERIALIZER_CREDIT 10
132 static int serializer_credit = SERIALIZER_CREDIT;
133
134 /* Statistics for debugging */
135 static int perim_context_swtch = 0;
136
137 static int serializer_constructor(void *, void *, int);
138 static void serializer_destructor(void *, void *);
139 static void serializer_exec(serializer_t *, srproc_t, mblk_t *, void *);
140 static void serializer_enqueue(serializer_t *, srproc_t, mblk_t *, void *);
141 static void serializer_drain(serializer_t *, int);
142 static void serializer_drain_completely(serializer_t *);
143
144 /*
145 * SERIALIZER Implementation.
146 */
147
148 /*
149 * Record debugging information and execute single request.
150 */
151 static void
serializer_exec(serializer_t * s,srproc_t proc,mblk_t * mp,void * arg)152 serializer_exec(serializer_t *s, srproc_t proc, mblk_t *mp, void *arg)
153 {
154 ASSERT(MUTEX_NOT_HELD(&s->ser_lock));
155 ASSERT(s->ser_owner == curthread);
156
157 ASSERT(proc != NULL);
158 ASSERT(mp != NULL);
159
160 s->ser_curr = mp;
161 s->ser_arg = arg;
162 s->ser_proc = proc;
163 proc(mp, arg);
164 }
165
166 /*
167 * Enqueue a single request on serializer.
168 */
169 static void
serializer_enqueue(serializer_t * s,srproc_t proc,mblk_t * mp,void * arg)170 serializer_enqueue(serializer_t *s, srproc_t proc, mblk_t *mp, void *arg)
171 {
172 ASSERT(MUTEX_HELD(&s->ser_lock));
173
174 DTRACE_PROBE4(serializer__enqueue, serializer_t *, s,
175 mblk_t *, mp, void *, arg, srproc_t, proc);
176 s->ser_count++;
177 mp->b_queue = (queue_t *)proc;
178 mp->b_prev = (mblk_t *)arg;
179 if (s->ser_last != NULL)
180 s->ser_last->b_next = mp;
181 else
182 s->ser_first = mp;
183 s->ser_last = mp;
184 }
185
186 /*
187 * Drain serializer, limiting drain to `credit' requests at most.
188 */
189 static void
serializer_drain(serializer_t * s,int credit)190 serializer_drain(serializer_t *s, int credit)
191 {
192 mblk_t *mp = s->ser_first;
193
194 ASSERT(MUTEX_HELD(&s->ser_lock));
195 ASSERT(s->ser_owner == curthread);
196
197 for (; mp != NULL && credit-- != 0; mp = s->ser_first) {
198 srproc_t *proc = (srproc_t *)mp->b_queue;
199 void *arg = mp->b_prev;
200
201 if ((s->ser_first = s->ser_first->b_next) == NULL) {
202 s->ser_last = NULL;
203 } else {
204 mp->b_next = NULL;
205 }
206 ASSERT(s->ser_count != 0);
207 s->ser_count--;
208 mp->b_queue = NULL;
209 mp->b_prev = NULL;
210 mutex_exit(&s->ser_lock);
211
212 DTRACE_PROBE4(serializer__exec__start, serializer_t *, s,
213 mblk_t *, mp, void *, arg, srproc_t, proc);
214 serializer_exec(s, proc, mp, arg);
215 DTRACE_PROBE4(serializer__exec__end, serializer_t *, s,
216 mblk_t *, mp, void *, arg, srproc_t, proc);
217
218 mutex_enter(&s->ser_lock);
219 }
220 }
221
222 /*
223 * Drain serializer completely if serializer is free.
224 */
225 static void
serializer_drain_completely(serializer_t * s)226 serializer_drain_completely(serializer_t *s)
227 {
228 mutex_enter(&s->ser_lock);
229 ASSERT(s->ser_taskq);
230 if (s->ser_owner == NULL) {
231 s->ser_owner = curthread;
232 while (s->ser_first != NULL)
233 serializer_drain(s, INT_MAX);
234 s->ser_owner = NULL;
235 s->ser_curr = NULL;
236 s->ser_proc = NULL;
237 s->ser_arg = NULL;
238 }
239 s->ser_taskq = B_FALSE;
240 /*
241 * Wake up serializer_wait().
242 */
243 cv_signal(&s->ser_cv);
244 mutex_exit(&s->ser_lock);
245 }
246
247 /*
248 * Call proc(mp, arg) within serializer.
249 *
250 * If serializer is empty and not owned, proc(mp, arg) is called right
251 * away. Otherwise the request is queued.
252 */
253 void
serializer_enter(serializer_t * s,srproc_t proc,mblk_t * mp,void * arg)254 serializer_enter(serializer_t *s, srproc_t proc, mblk_t *mp, void *arg)
255 {
256 ASSERT(proc != NULL);
257 ASSERT(mp != NULL);
258 ASSERT(mp->b_next == NULL);
259 ASSERT(mp->b_prev == NULL);
260
261 ASSERT(MUTEX_NOT_HELD(&s->ser_lock));
262
263 mutex_enter(&s->ser_lock);
264 if (s->ser_owner != NULL) {
265 /*
266 * Serializer is owned. Enqueue and return.
267 */
268 serializer_enqueue(s, proc, mp, arg);
269 } else {
270 taskqid_t tid = 0;
271
272 /*
273 * If the request list is empty, can process right away,
274 * otherwise enqueue and process.
275 */
276 s->ser_owner = curthread;
277
278 if (s->ser_first != NULL) {
279 ASSERT(s->ser_count != 0);
280 serializer_enqueue(s, proc, mp, arg);
281 } else {
282 ASSERT(s->ser_count == 0);
283 mutex_exit(&s->ser_lock);
284 /*
285 * Execute request
286 */
287 DTRACE_PROBE4(serializer__exec__start,
288 serializer_t *, s, mblk_t *, mp,
289 void *, arg, srproc_t, proc);
290 serializer_exec(s, proc, mp, arg);
291 DTRACE_PROBE4(serializer__exec__end,
292 serializer_t *, s, mblk_t *, mp,
293 void *, arg, srproc_t, proc);
294 mutex_enter(&s->ser_lock);
295 }
296
297 /*
298 * Drain whatever has arrived in the meantime.
299 * If we spend too much time draining, continue draining by the
300 * taskq thread.
301 */
302 while ((s->ser_first != NULL) && (tid == 0)) {
303 serializer_drain(s, serializer_credit);
304 if (s->ser_first != NULL) {
305 perim_context_swtch++;
306 /*
307 * If there is a taskq pending for this
308 * serializer, no need to schedule a new one.
309 */
310 if (s->ser_taskq) {
311 break;
312 } else {
313 tid = taskq_dispatch(system_taskq,
314 (task_func_t *)
315 serializer_drain_completely,
316 s, TQ_NOSLEEP | TQ_NOQUEUE);
317 if (tid != 0)
318 s->ser_taskq = B_TRUE;
319 }
320 }
321 }
322 s->ser_owner = NULL;
323 s->ser_curr = NULL;
324 s->ser_proc = NULL;
325 s->ser_arg = NULL;
326 }
327 /*
328 * Wakeup serializer_wait().
329 */
330 cv_signal(&s->ser_cv);
331 mutex_exit(&s->ser_lock);
332 }
333
334 /*
335 * Wait for pending serializer jobs to complete. This function should never be
336 * called within the serializer or it will deadlock.
337 */
338 void
serializer_wait(serializer_t * s)339 serializer_wait(serializer_t *s)
340 {
341 mutex_enter(&s->ser_lock);
342
343 ASSERT(s->ser_owner != curthread);
344
345 while ((s->ser_owner != NULL) || s->ser_taskq || (s->ser_first != NULL))
346 cv_wait(&s->ser_cv, &s->ser_lock);
347 ASSERT((s->ser_first == NULL) && (s->ser_last == NULL));
348 /*
349 * Wakeup other potential waiters.
350 */
351 cv_signal(&s->ser_cv);
352 mutex_exit(&s->ser_lock);
353 }
354
355 /*
356 * Create a new serializer.
357 */
358 serializer_t *
serializer_create(int flags)359 serializer_create(int flags)
360 {
361 return (kmem_cache_alloc(serializer_cache, flags));
362 }
363
364 /*
365 * Wait for all pending entries to drain and then destroy serializer.
366 */
367 void
serializer_destroy(serializer_t * s)368 serializer_destroy(serializer_t *s)
369 {
370 serializer_wait(s);
371
372 ASSERT(s->ser_owner == NULL);
373 ASSERT(s->ser_taskq == 0);
374 ASSERT(s->ser_count == 0);
375 ASSERT(s->ser_first == NULL);
376 ASSERT(s->ser_last == NULL);
377
378 kmem_cache_free(serializer_cache, s);
379 }
380
381 /*ARGSUSED*/
382 static int
serializer_constructor(void * buf,void * cdrarg,int kmflags)383 serializer_constructor(void *buf, void *cdrarg, int kmflags)
384 {
385 serializer_t *s = buf;
386
387 mutex_init(&s->ser_lock, NULL, MUTEX_DEFAULT, NULL);
388 cv_init(&s->ser_cv, NULL, CV_DEFAULT, NULL);
389
390 s->ser_taskq = 0;
391 s->ser_count = 0;
392 s->ser_first = s->ser_last = s->ser_curr = NULL;
393 s->ser_proc = NULL;
394 s->ser_arg = NULL;
395 s->ser_owner = NULL;
396 return (0);
397 }
398
399 /*ARGSUSED*/
400 static void
serializer_destructor(void * buf,void * cdrarg)401 serializer_destructor(void *buf, void *cdrarg)
402 {
403 serializer_t *s = buf;
404
405 mutex_destroy(&s->ser_lock);
406 cv_destroy(&s->ser_cv);
407 }
408
409 void
serializer_init(void)410 serializer_init(void)
411 {
412 serializer_cache = kmem_cache_create("serializer_cache",
413 sizeof (serializer_t), 0, serializer_constructor,
414 serializer_destructor, NULL, NULL, NULL, 0);
415 }
416