2 * Copyright (c) 2002, Jeffrey Roberson <jeff@freebsd.org>
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions
8 * 1. Redistributions of source code must retain the above copyright
9 * notice unmodified, this list of conditions, and the following
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.
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.
27 #include <sys/cdefs.h>
28 __FBSDID("$FreeBSD$");
32 #include <sys/param.h>
33 #include <sys/systm.h>
34 #include <sys/kernel.h>
35 #include <sys/kthread.h>
38 #include <sys/mount.h>
39 #include <sys/mutex.h>
40 #include <sys/namei.h>
42 #include <sys/vnode.h>
44 #include <sys/malloc.h>
45 #include <sys/unistd.h>
46 #include <sys/fcntl.h>
47 #include <sys/eventhandler.h>
49 /* Async. Logging Queue */
51 int aq_entmax; /* Max entries */
52 int aq_entlen; /* Entry length */
53 char *aq_entbuf; /* Buffer for stored entries */
54 int aq_flags; /* Queue flags */
55 struct mtx aq_mtx; /* Queue lock */
56 struct vnode *aq_vp; /* Open vnode handle */
57 struct ucred *aq_cred; /* Credentials of the opening thread */
58 struct ale *aq_first; /* First ent */
59 struct ale *aq_entfree; /* First free ent */
60 struct ale *aq_entvalid; /* First ent valid for writing */
61 LIST_ENTRY(alq) aq_act; /* List of active queues */
62 LIST_ENTRY(alq) aq_link; /* List of all queues */
65 #define AQ_WANTED 0x0001 /* Wakeup sleeper when io is done */
66 #define AQ_ACTIVE 0x0002 /* on the active list */
67 #define AQ_FLUSHING 0x0004 /* doing IO */
68 #define AQ_SHUTDOWN 0x0008 /* Queue no longer valid */
70 #define ALQ_LOCK(alq) mtx_lock_spin(&(alq)->aq_mtx)
71 #define ALQ_UNLOCK(alq) mtx_unlock_spin(&(alq)->aq_mtx)
73 static MALLOC_DEFINE(M_ALD, "ALD", "ALD");
76 * The ald_mtx protects the ald_queues list and the ald_active list.
78 static struct mtx ald_mtx;
79 static LIST_HEAD(, alq) ald_queues;
80 static LIST_HEAD(, alq) ald_active;
81 static int ald_shutingdown = 0;
82 struct thread *ald_thread;
83 static struct proc *ald_proc;
85 #define ALD_LOCK() mtx_lock(&ald_mtx)
86 #define ALD_UNLOCK() mtx_unlock(&ald_mtx)
88 /* Daemon functions */
89 static int ald_add(struct alq *);
90 static int ald_rem(struct alq *);
91 static void ald_startup(void *);
92 static void ald_daemon(void);
93 static void ald_shutdown(void *, int);
94 static void ald_activate(struct alq *);
95 static void ald_deactivate(struct alq *);
97 /* Internal queue functions */
98 static void alq_shutdown(struct alq *);
99 static int alq_doio(struct alq *);
103 * Add a new queue to the global list. Fail if we're shutting down.
106 ald_add(struct alq *alq)
113 if (ald_shutingdown) {
117 LIST_INSERT_HEAD(&ald_queues, alq, aq_link);
124 * Remove a queue from the global list unless we're shutting down. If so,
125 * the ald will take care of cleaning up it's resources.
128 ald_rem(struct alq *alq)
135 if (ald_shutingdown) {
139 LIST_REMOVE(alq, aq_link);
146 * Put a queue on the active list. This will schedule it for writing.
149 ald_activate(struct alq *alq)
151 LIST_INSERT_HEAD(&ald_active, alq, aq_act);
156 ald_deactivate(struct alq *alq)
158 LIST_REMOVE(alq, aq_act);
159 alq->aq_flags &= ~AQ_ACTIVE;
163 ald_startup(void *unused)
165 mtx_init(&ald_mtx, "ALDmtx", NULL, MTX_DEF|MTX_QUIET);
166 LIST_INIT(&ald_queues);
167 LIST_INIT(&ald_active);
176 ald_thread = FIRST_THREAD_IN_PROC(ald_proc);
178 EVENTHANDLER_REGISTER(shutdown_pre_sync, ald_shutdown, NULL,
184 while ((alq = LIST_FIRST(&ald_active)) == NULL)
185 msleep(&ald_active, &ald_mtx, PWAIT, "aldslp", 0);
190 needwakeup = alq_doio(alq);
199 ald_shutdown(void *arg, int howto)
206 while ((alq = LIST_FIRST(&ald_queues)) != NULL) {
207 LIST_REMOVE(alq, aq_link);
216 alq_shutdown(struct alq *alq)
220 /* Stop any new writers. */
221 alq->aq_flags |= AQ_SHUTDOWN;
224 while (alq->aq_flags & (AQ_FLUSHING|AQ_ACTIVE)) {
225 alq->aq_flags |= AQ_WANTED;
227 tsleep(alq, PWAIT, "aldclose", 0);
232 vn_close(alq->aq_vp, FWRITE, alq->aq_cred,
234 crfree(alq->aq_cred);
238 * Flush all pending data to disk. This operation will block.
241 alq_doio(struct alq *alq)
247 struct iovec aiov[2];
259 alstart = ale = alq->aq_entvalid;
260 alq->aq_entvalid = NULL;
262 bzero(&aiov, sizeof(aiov));
263 bzero(&auio, sizeof(auio));
266 if (aiov[iov].iov_base == NULL)
267 aiov[iov].iov_base = ale->ae_data;
268 aiov[iov].iov_len += alq->aq_entlen;
269 totlen += alq->aq_entlen;
270 /* Check to see if we're wrapping the buffer */
271 if (ale->ae_data + alq->aq_entlen != ale->ae_next->ae_data)
273 ale->ae_flags &= ~AE_VALID;
275 } while (ale->ae_flags & AE_VALID);
277 alq->aq_flags |= AQ_FLUSHING;
280 if (iov == 2 || aiov[iov].iov_base == NULL)
283 auio.uio_iov = &aiov[0];
285 auio.uio_segflg = UIO_SYSSPACE;
286 auio.uio_rw = UIO_WRITE;
287 auio.uio_iovcnt = iov + 1;
288 auio.uio_resid = totlen;
292 * Do all of the junk required to write now.
294 vfslocked = VFS_LOCK_GIANT(vp->v_mount);
295 vn_start_write(vp, &mp, V_WAIT);
296 vn_lock(vp, LK_EXCLUSIVE | LK_RETRY, td);
297 VOP_LEASE(vp, td, alq->aq_cred, LEASE_WRITE);
299 * XXX: VOP_WRITE error checks are ignored.
302 if (mac_check_vnode_write(alq->aq_cred, NOCRED, vp) == 0)
304 VOP_WRITE(vp, &auio, IO_UNIT | IO_APPEND, alq->aq_cred);
305 VOP_UNLOCK(vp, 0, td);
306 vn_finished_write(mp);
307 VFS_UNLOCK_GIANT(vfslocked);
310 alq->aq_flags &= ~AQ_FLUSHING;
312 if (alq->aq_entfree == NULL)
313 alq->aq_entfree = alstart;
315 if (alq->aq_flags & AQ_WANTED) {
316 alq->aq_flags &= ~AQ_WANTED;
323 static struct kproc_desc ald_kp = {
329 SYSINIT(aldthread, SI_SUB_KTHREAD_IDLE, SI_ORDER_ANY, kproc_start, &ald_kp)
330 SYSINIT(ald, SI_SUB_LOCK, SI_ORDER_ANY, ald_startup, NULL)
333 /* User visible queue functions */
336 * Create the queue data structure, allocate the buffer, and open the file.
339 alq_open(struct alq **alqp, const char *file, struct ucred *cred, int cmode,
355 NDINIT(&nd, LOOKUP, NOFOLLOW | MPSAFE, UIO_SYSSPACE, file, td);
356 flags = FWRITE | O_NOFOLLOW | O_CREAT;
358 error = vn_open_cred(&nd, &flags, cmode, cred, -1);
362 vfslocked = NDHASGIANT(&nd);
363 NDFREE(&nd, NDF_ONLY_PNBUF);
364 /* We just unlock so we hold a reference */
365 VOP_UNLOCK(nd.ni_vp, 0, td);
366 VFS_UNLOCK_GIANT(vfslocked);
368 alq = malloc(sizeof(*alq), M_ALD, M_WAITOK|M_ZERO);
369 alq->aq_entbuf = malloc(count * size, M_ALD, M_WAITOK|M_ZERO);
370 alq->aq_first = malloc(sizeof(*ale) * count, M_ALD, M_WAITOK|M_ZERO);
371 alq->aq_vp = nd.ni_vp;
372 alq->aq_cred = crhold(cred);
373 alq->aq_entmax = count;
374 alq->aq_entlen = size;
375 alq->aq_entfree = alq->aq_first;
377 mtx_init(&alq->aq_mtx, "ALD Queue", NULL, MTX_SPIN|MTX_QUIET);
379 bufp = alq->aq_entbuf;
383 /* Match up entries with buffers */
384 for (i = 0; i < count; i++) {
393 alp->ae_next = alq->aq_first;
395 if ((error = ald_add(alq)) != 0)
403 * Copy a new entry into the queue. If the operation would block either
404 * wait or return an error depending on the value of waitok.
407 alq_write(struct alq *alq, void *data, int waitok)
411 if ((ale = alq_get(alq, waitok)) == NULL)
412 return (EWOULDBLOCK);
414 bcopy(data, ale->ae_data, alq->aq_entlen);
421 alq_get(struct alq *alq, int waitok)
430 /* Loop until we get an entry or we're shutting down */
431 while ((alq->aq_flags & AQ_SHUTDOWN) == 0 &&
432 (ale = alq->aq_entfree) == NULL &&
433 (waitok & ALQ_WAITOK)) {
434 alq->aq_flags |= AQ_WANTED;
436 tsleep(alq, PWAIT, "alqget", 0);
442 if ((aln->ae_flags & AE_VALID) == 0)
443 alq->aq_entfree = aln;
445 alq->aq_entfree = NULL;
454 alq_post(struct alq *alq, struct ale *ale)
458 ale->ae_flags |= AE_VALID;
460 if (alq->aq_entvalid == NULL)
461 alq->aq_entvalid = ale;
463 if ((alq->aq_flags & AQ_ACTIVE) == 0) {
464 alq->aq_flags |= AQ_ACTIVE;
478 alq_flush(struct alq *alq)
484 if (alq->aq_flags & AQ_ACTIVE) {
487 needwakeup = alq_doio(alq);
497 * Flush remaining data, close the file and free all resources.
500 alq_close(struct alq *alq)
503 * If we're already shuting down someone else will flush and close
506 if (ald_rem(alq) != 0)
510 * Drain all pending IO.
514 mtx_destroy(&alq->aq_mtx);
515 free(alq->aq_first, M_ALD);
516 free(alq->aq_entbuf, M_ALD);