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