xref: /illumos-gate/usr/src/uts/common/os/serializer.c (revision 2d6eb4a5e0a47d30189497241345dc5466bb68ab)
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