kern_alq.c revision 103785 
1/*
2 * Copyright (c) 2002, Jeffrey Roberson <jeff@freebsd.org>
3 * All rights reserved.
4 *
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions
7 * are met:
8 * 1. Redistributions of source code must retain the above copyright
9 *    notice unmodified, this list of conditions, and the following
10 *    disclaimer.
11 * 2. Redistributions in binary form must reproduce the above copyright
12 *    notice, this list of conditions and the following disclaimer in the
13 *    documentation and/or other materials provided with the distribution.
14 *
15 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
16 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
17 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
18 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
19 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
20 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
21 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
22 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
23 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
24 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
25 *
26 * $FreeBSD: head/sys/kern/kern_alq.c 103785 2002-09-22 07:11:14Z jeff $
27 *
28 */
29
30#include <sys/param.h>
31#include <sys/systm.h>
32#include <sys/kernel.h>
33#include <sys/kthread.h>
34#include <sys/lock.h>
35#include <sys/mutex.h>
36#include <sys/namei.h>
37#include <sys/proc.h>
38#include <sys/vnode.h>
39#include <sys/alq.h>
40#include <sys/malloc.h>
41#include <sys/unistd.h>
42#include <sys/fcntl.h>
43#include <sys/eventhandler.h>
44
45/* Async. Logging Queue */
46struct alq {
47	int	aq_entmax;		/* Max entries */
48	int	aq_entlen;		/* Entry length */
49	char	*aq_entbuf;		/* Buffer for stored entries */
50	int	aq_flags;		/* Queue flags */
51	struct mtx	aq_mtx;		/* Queue lock */
52	struct vnode	*aq_vp;		/* Open vnode handle */
53	struct thread	*aq_td;		/* Thread that opened the vnode */
54	struct ale	*aq_first;	/* First ent */
55	struct ale	*aq_entfree;	/* First free ent */
56	struct ale	*aq_entvalid;	/* First ent valid for writing */
57	LIST_ENTRY(alq)	aq_act;		/* List of active queues */
58	LIST_ENTRY(alq)	aq_link;	/* List of all queues */
59};
60
61#define	AQ_WANTED	0x0001		/* Wakeup sleeper when io is done */
62#define	AQ_ACTIVE	0x0002		/* on the active list */
63#define	AQ_FLUSHING	0x0004		/* doing IO */
64#define	AQ_SHUTDOWN	0x0008		/* Queue no longer valid */
65
66#define	ALQ_LOCK(alq)	mtx_lock_spin(&(alq)->aq_mtx)
67#define	ALQ_UNLOCK(alq)	mtx_unlock_spin(&(alq)->aq_mtx)
68
69static MALLOC_DEFINE(M_ALD, "ALD", "ALD");
70
71/*
72 * The ald_mtx protects the ald_queues list and the ald_active list.
73 */
74static struct mtx ald_mtx;
75static LIST_HEAD(, alq) ald_queues;
76static LIST_HEAD(, alq) ald_active;
77static struct proc *ald_thread;
78static int ald_shutingdown = 0;
79
80#define	ALD_LOCK()	mtx_lock(&ald_mtx)
81#define	ALD_UNLOCK()	mtx_unlock(&ald_mtx)
82
83/* Daemon functions */
84static int ald_add(struct alq *);
85static int ald_rem(struct alq *);
86static void ald_startup(void *);
87static void ald_daemon(void);
88static void ald_shutdown(void *, int);
89static void ald_activate(struct alq *);
90static void ald_deactivate(struct alq *);
91
92/* Internal queue functions */
93static void alq_shutdown(struct alq *);
94static int alq_doio(struct alq *);
95
96
97/*
98 * Add a new queue to the global list.  Fail if we're shutting down.
99 */
100static int
101ald_add(struct alq *alq)
102{
103	int error;
104
105	error = 0;
106
107	ALD_LOCK();
108	if (ald_shutingdown) {
109		error = EBUSY;
110		goto done;
111	}
112	LIST_INSERT_HEAD(&ald_queues, alq, aq_link);
113done:
114	ALD_UNLOCK();
115	return (error);
116}
117
118/*
119 * Remove a queue from the global list unless we're shutting down.  If so,
120 * the ald will take care of cleaning up it's resources.
121 */
122static int
123ald_rem(struct alq *alq)
124{
125	int error;
126
127	error = 0;
128
129	ALD_LOCK();
130	if (ald_shutingdown) {
131		error = EBUSY;
132		goto done;
133	}
134	LIST_REMOVE(alq, aq_link);
135done:
136	ALD_UNLOCK();
137	return (error);
138}
139
140/*
141 * Put a queue on the active list.  This will schedule it for writing.
142 */
143static void
144ald_activate(struct alq *alq)
145{
146	LIST_INSERT_HEAD(&ald_active, alq, aq_act);
147	wakeup(&ald_active);
148}
149
150static void
151ald_deactivate(struct alq *alq)
152{
153	LIST_REMOVE(alq, aq_act);
154	alq->aq_flags &= ~AQ_ACTIVE;
155}
156
157static void
158ald_startup(void *unused)
159{
160	mtx_init(&ald_mtx, "ALDmtx", NULL, MTX_DEF|MTX_QUIET);
161	LIST_INIT(&ald_queues);
162	LIST_INIT(&ald_active);
163}
164
165static void
166ald_daemon(void)
167{
168	int needwakeup;
169	struct alq *alq;
170
171	mtx_lock(&Giant);
172
173	EVENTHANDLER_REGISTER(shutdown_pre_sync, ald_shutdown, NULL,
174	    SHUTDOWN_PRI_FIRST);
175
176	ALD_LOCK();
177
178	for (;;) {
179		while ((alq = LIST_FIRST(&ald_active)) == NULL)
180			msleep(&ald_active, &ald_mtx, PWAIT, "aldslp", 0);
181
182		ALQ_LOCK(alq);
183		ald_deactivate(alq);
184		ALD_UNLOCK();
185		needwakeup = alq_doio(alq);
186		ALQ_UNLOCK(alq);
187		if (needwakeup)
188			wakeup(alq);
189		ALD_LOCK();
190	}
191}
192
193static void
194ald_shutdown(void *arg, int howto)
195{
196	struct alq *alq;
197
198	ALD_LOCK();
199	ald_shutingdown = 1;
200
201	while ((alq = LIST_FIRST(&ald_queues)) != NULL) {
202		LIST_REMOVE(alq, aq_link);
203		ALD_UNLOCK();
204		alq_shutdown(alq);
205		ALD_LOCK();
206	}
207	ALD_UNLOCK();
208}
209
210static void
211alq_shutdown(struct alq *alq)
212{
213	ALQ_LOCK(alq);
214
215	/* Stop any new writers. */
216	alq->aq_flags |= AQ_SHUTDOWN;
217
218	/* Drain IO */
219	while (alq->aq_flags & (AQ_FLUSHING|AQ_ACTIVE)) {
220		alq->aq_flags |= AQ_WANTED;
221		ALQ_UNLOCK(alq);
222		tsleep(alq, PWAIT, "aldclose", 0);
223		ALQ_LOCK(alq);
224	}
225	ALQ_UNLOCK(alq);
226
227	vn_close(alq->aq_vp, FREAD|FWRITE, alq->aq_td->td_ucred,
228	    alq->aq_td);
229}
230
231/*
232 * Flush all pending data to disk.  This operation will block.
233 */
234static int
235alq_doio(struct alq *alq)
236{
237	struct thread *td;
238	struct mount *mp;
239	struct vnode *vp;
240	struct uio auio;
241	struct iovec aiov[2];
242	struct ale *ale;
243	struct ale *alstart;
244	int totlen;
245	int iov;
246
247	vp = alq->aq_vp;
248	td = curthread;
249	totlen = 0;
250	iov = 0;
251
252	alstart = ale = alq->aq_entvalid;
253	alq->aq_entvalid = NULL;
254
255	bzero(&aiov, sizeof(aiov));
256	bzero(&auio, sizeof(auio));
257
258	do {
259		if (aiov[iov].iov_base == NULL)
260			aiov[iov].iov_base = ale->ae_data;
261		aiov[iov].iov_len += alq->aq_entlen;
262		totlen += alq->aq_entlen;
263		/* Check to see if we're wrapping the buffer */
264		if (ale->ae_data + alq->aq_entlen != ale->ae_next->ae_data)
265			iov++;
266		ale->ae_flags &= ~AE_VALID;
267		ale = ale->ae_next;
268	} while (ale->ae_flags & AE_VALID);
269
270	alq->aq_flags |= AQ_FLUSHING;
271	ALQ_UNLOCK(alq);
272
273	if (iov == 2 || aiov[iov].iov_base == NULL)
274		iov--;
275
276	auio.uio_iov = &aiov[0];
277	auio.uio_offset = 0;
278	auio.uio_segflg = UIO_SYSSPACE;
279	auio.uio_rw = UIO_WRITE;
280	auio.uio_iovcnt = iov + 1;
281	auio.uio_resid = totlen;
282	auio.uio_td = td;
283
284	/*
285	 * Do all of the junk required to write now.
286	 */
287	vn_start_write(vp, &mp, V_WAIT);
288	vn_lock(vp, LK_EXCLUSIVE | LK_RETRY, td);
289	VOP_LEASE(vp, td, td->td_ucred, LEASE_WRITE);
290	/* XXX error ignored */
291	VOP_WRITE(vp, &auio, IO_UNIT | IO_APPEND, td->td_ucred);
292	VOP_UNLOCK(vp, 0, td);
293	vn_finished_write(mp);
294
295	ALQ_LOCK(alq);
296	alq->aq_flags &= ~AQ_FLUSHING;
297
298	if (alq->aq_entfree == NULL)
299		alq->aq_entfree = alstart;
300
301	if (alq->aq_flags & AQ_WANTED) {
302		alq->aq_flags &= ~AQ_WANTED;
303		return (1);
304	}
305
306	return(0);
307}
308
309static struct kproc_desc ald_kp = {
310        "ALQ Daemon",
311        ald_daemon,
312        &ald_thread
313};
314
315SYSINIT(aldthread, SI_SUB_KTHREAD_IDLE, SI_ORDER_ANY, kproc_start, &ald_kp)
316SYSINIT(ald, SI_SUB_LOCK, SI_ORDER_ANY, ald_startup, NULL)
317
318
319/* User visible queue functions */
320
321/*
322 * Create the queue data structure, allocate the buffer, and open the file.
323 */
324int
325alq_open(struct alq **alqp, const char *file, int size, int count)
326{
327	struct thread *td;
328	struct nameidata nd;
329	struct ale *ale;
330	struct ale *alp;
331	struct alq *alq;
332	char *bufp;
333	int flags;
334	int error;
335	int i;
336
337	*alqp = NULL;
338	td = curthread;
339
340	NDINIT(&nd, LOOKUP, NOFOLLOW, UIO_SYSSPACE, file, td);
341	flags = FREAD | FWRITE | O_NOFOLLOW | O_CREAT;
342
343	error = vn_open(&nd, &flags, 0);
344	if (error)
345		return (error);
346
347	NDFREE(&nd, NDF_ONLY_PNBUF);
348	/* We just unlock so we hold a reference */
349	VOP_UNLOCK(nd.ni_vp, 0, td);
350
351	alq = malloc(sizeof(*alq), M_ALD, M_WAITOK|M_ZERO);
352	alq->aq_entbuf = malloc(count * size, M_ALD, M_WAITOK|M_ZERO);
353	alq->aq_first = malloc(sizeof(*ale) * count, M_ALD, M_WAITOK|M_ZERO);
354	alq->aq_vp = nd.ni_vp;
355	alq->aq_td = td;
356	alq->aq_entmax = count;
357	alq->aq_entlen = size;
358	alq->aq_entfree = alq->aq_first;
359
360	mtx_init(&alq->aq_mtx, "ALD Queue", NULL, MTX_SPIN|MTX_QUIET);
361
362	bufp = alq->aq_entbuf;
363	ale = alq->aq_first;
364	alp = NULL;
365
366	/* Match up entries with buffers */
367	for (i = 0; i < count; i++) {
368		if (alp)
369			alp->ae_next = ale;
370		ale->ae_data = bufp;
371		alp = ale;
372		ale++;
373		bufp += size;
374	}
375
376	alp->ae_next = alq->aq_first;
377
378	if ((error = ald_add(alq)) != 0)
379		return (error);
380	*alqp = alq;
381
382	return (0);
383}
384
385/*
386 * Copy a new entry into the queue.  If the operation would block either
387 * wait or return an error depending on the value of waitok.
388 */
389int
390alq_write(struct alq *alq, void *data, int waitok)
391{
392	struct ale *ale;
393
394	if ((ale = alq_get(alq, waitok)) == NULL)
395		return (EWOULDBLOCK);
396
397	bcopy(data, ale->ae_data, alq->aq_entlen);
398	alq_post(alq, ale);
399
400	return (0);
401}
402
403struct ale *
404alq_get(struct alq *alq, int waitok)
405{
406	struct ale *ale;
407	struct ale *aln;
408
409	ale = NULL;
410
411	ALQ_LOCK(alq);
412
413	/* Loop until we get an entry or we're shutting down */
414	while ((alq->aq_flags & AQ_SHUTDOWN) == 0 &&
415	    (ale = alq->aq_entfree) == NULL &&
416	    (waitok & ALQ_WAITOK)) {
417		alq->aq_flags |= AQ_WANTED;
418		ALQ_UNLOCK(alq);
419		tsleep(alq, PWAIT, "alqget", 0);
420		ALQ_LOCK(alq);
421	}
422
423	if (ale != NULL) {
424		aln = ale->ae_next;
425		if ((aln->ae_flags & AE_VALID) == 0)
426			alq->aq_entfree = aln;
427	} else
428		ALQ_UNLOCK(alq);
429
430
431	return (ale);
432}
433
434void
435alq_post(struct alq *alq, struct ale *ale)
436{
437	int activate;
438
439	ale->ae_flags |= AE_VALID;
440
441	if (alq->aq_entvalid == NULL)
442		alq->aq_entvalid = ale;
443
444	if ((alq->aq_flags & AQ_ACTIVE) == 0) {
445		alq->aq_flags |= AQ_ACTIVE;
446		activate = 1;
447	} else
448		activate = 0;
449
450	ALQ_UNLOCK(alq);
451	if (activate) {
452		ALD_LOCK();
453		ald_activate(alq);
454		ALD_UNLOCK();
455	}
456}
457
458void
459alq_flush(struct alq *alq)
460{
461	int needwakeup = 0;
462
463	ALD_LOCK();
464	ALQ_LOCK(alq);
465	if (alq->aq_flags & AQ_ACTIVE) {
466		ald_deactivate(alq);
467		ALD_UNLOCK();
468		needwakeup = alq_doio(alq);
469	} else
470		ALD_UNLOCK();
471	ALQ_UNLOCK(alq);
472
473	if (needwakeup)
474		wakeup(alq);
475}
476
477/*
478 * Flush remaining data, close the file and free all resources.
479 */
480void
481alq_close(struct alq *alq)
482{
483	/*
484	 * If we're already shuting down someone else will flush and close
485	 * the vnode.
486	 */
487	if (ald_rem(alq) != 0)
488		return;
489
490	/*
491	 * Drain all pending IO.
492	 */
493	alq_shutdown(alq);
494
495	mtx_destroy(&alq->aq_mtx);
496	free(alq->aq_first, M_ALD);
497	free(alq->aq_entbuf, M_ALD);
498	free(alq, M_ALD);
499}
500