2 * SPDX-License-Identifier: BSD-2-Clause-FreeBSD
4 * Copyright (c) 2002, Jeffrey Roberson <jeff@freebsd.org>
5 * Copyright (c) 2008-2009, Lawrence Stewart <lstewart@freebsd.org>
6 * Copyright (c) 2009-2010, The FreeBSD Foundation
9 * Portions of this software were developed at the Centre for Advanced
10 * Internet Architectures, Swinburne University of Technology, Melbourne,
11 * Australia by Lawrence Stewart under sponsorship from the FreeBSD Foundation.
13 * Redistribution and use in source and binary forms, with or without
14 * modification, are permitted provided that the following conditions
16 * 1. Redistributions of source code must retain the above copyright
17 * notice unmodified, this list of conditions, and the following
19 * 2. Redistributions in binary form must reproduce the above copyright
20 * notice, this list of conditions and the following disclaimer in the
21 * documentation and/or other materials provided with the distribution.
23 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
24 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
25 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
26 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
27 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
28 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
29 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
30 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
31 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
32 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
35 #include <sys/cdefs.h>
36 __FBSDID("$FreeBSD$");
40 #include <sys/param.h>
41 #include <sys/systm.h>
42 #include <sys/kernel.h>
43 #include <sys/kthread.h>
45 #include <sys/mount.h>
46 #include <sys/mutex.h>
47 #include <sys/namei.h>
49 #include <sys/vnode.h>
51 #include <sys/malloc.h>
52 #include <sys/unistd.h>
53 #include <sys/fcntl.h>
54 #include <sys/eventhandler.h>
56 #include <security/mac/mac_framework.h>
58 /* Async. Logging Queue */
60 char *aq_entbuf; /* Buffer for stored entries */
61 int aq_entmax; /* Max entries */
62 int aq_entlen; /* Entry length */
63 int aq_freebytes; /* Bytes available in buffer */
64 int aq_buflen; /* Total length of our buffer */
65 int aq_writehead; /* Location for next write */
66 int aq_writetail; /* Flush starts at this location */
67 int aq_wrapearly; /* # bytes left blank at end of buf */
68 int aq_flags; /* Queue flags */
69 int aq_waiters; /* Num threads waiting for resources
70 * NB: Used as a wait channel so must
71 * not be first field in the alq struct
73 struct ale aq_getpost; /* ALE for use by get/post */
74 struct mtx aq_mtx; /* Queue lock */
75 struct vnode *aq_vp; /* Open vnode handle */
76 struct ucred *aq_cred; /* Credentials of the opening thread */
77 LIST_ENTRY(alq) aq_act; /* List of active queues */
78 LIST_ENTRY(alq) aq_link; /* List of all queues */
81 #define AQ_WANTED 0x0001 /* Wakeup sleeper when io is done */
82 #define AQ_ACTIVE 0x0002 /* on the active list */
83 #define AQ_FLUSHING 0x0004 /* doing IO */
84 #define AQ_SHUTDOWN 0x0008 /* Queue no longer valid */
85 #define AQ_ORDERED 0x0010 /* Queue enforces ordered writes */
86 #define AQ_LEGACY 0x0020 /* Legacy queue (fixed length writes) */
88 #define ALQ_LOCK(alq) mtx_lock_spin(&(alq)->aq_mtx)
89 #define ALQ_UNLOCK(alq) mtx_unlock_spin(&(alq)->aq_mtx)
91 #define HAS_PENDING_DATA(alq) ((alq)->aq_freebytes != (alq)->aq_buflen)
93 static MALLOC_DEFINE(M_ALD, "ALD", "ALD");
96 * The ald_mtx protects the ald_queues list and the ald_active list.
98 static struct mtx ald_mtx;
99 static LIST_HEAD(, alq) ald_queues;
100 static LIST_HEAD(, alq) ald_active;
101 static int ald_shutingdown = 0;
102 struct thread *ald_thread;
103 static struct proc *ald_proc;
104 static eventhandler_tag alq_eventhandler_tag = NULL;
106 #define ALD_LOCK() mtx_lock(&ald_mtx)
107 #define ALD_UNLOCK() mtx_unlock(&ald_mtx)
109 /* Daemon functions */
110 static int ald_add(struct alq *);
111 static int ald_rem(struct alq *);
112 static void ald_startup(void *);
113 static void ald_daemon(void);
114 static void ald_shutdown(void *, int);
115 static void ald_activate(struct alq *);
116 static void ald_deactivate(struct alq *);
118 /* Internal queue functions */
119 static void alq_shutdown(struct alq *);
120 static void alq_destroy(struct alq *);
121 static int alq_doio(struct alq *);
124 * Add a new queue to the global list. Fail if we're shutting down.
127 ald_add(struct alq *alq)
134 if (ald_shutingdown) {
138 LIST_INSERT_HEAD(&ald_queues, alq, aq_link);
145 * Remove a queue from the global list unless we're shutting down. If so,
146 * the ald will take care of cleaning up it's resources.
149 ald_rem(struct alq *alq)
156 if (ald_shutingdown) {
160 LIST_REMOVE(alq, aq_link);
167 * Put a queue on the active list. This will schedule it for writing.
170 ald_activate(struct alq *alq)
172 LIST_INSERT_HEAD(&ald_active, alq, aq_act);
177 ald_deactivate(struct alq *alq)
179 LIST_REMOVE(alq, aq_act);
180 alq->aq_flags &= ~AQ_ACTIVE;
184 ald_startup(void *unused)
186 mtx_init(&ald_mtx, "ALDmtx", NULL, MTX_DEF|MTX_QUIET);
187 LIST_INIT(&ald_queues);
188 LIST_INIT(&ald_active);
197 ald_thread = FIRST_THREAD_IN_PROC(ald_proc);
199 alq_eventhandler_tag = EVENTHANDLER_REGISTER(shutdown_pre_sync,
200 ald_shutdown, NULL, SHUTDOWN_PRI_FIRST);
205 while ((alq = LIST_FIRST(&ald_active)) == NULL &&
207 mtx_sleep(&ald_active, &ald_mtx, PWAIT, "aldslp", 0);
209 /* Don't shutdown until all active ALQs are flushed. */
210 if (ald_shutingdown && alq == NULL) {
218 needwakeup = alq_doio(alq);
229 ald_shutdown(void *arg, int howto)
235 /* Ensure no new queues can be created. */
238 /* Shutdown all ALQs prior to terminating the ald_daemon. */
239 while ((alq = LIST_FIRST(&ald_queues)) != NULL) {
240 LIST_REMOVE(alq, aq_link);
246 /* At this point, all ALQs are flushed and shutdown. */
249 * Wake ald_daemon so that it exits. It won't be able to do
250 * anything until we mtx_sleep because we hold the ald_mtx.
254 /* Wait for ald_daemon to exit. */
255 mtx_sleep(ald_proc, &ald_mtx, PWAIT, "aldslp", 0);
261 alq_shutdown(struct alq *alq)
265 /* Stop any new writers. */
266 alq->aq_flags |= AQ_SHUTDOWN;
269 * If the ALQ isn't active but has unwritten data (possible if
270 * the ALQ_NOACTIVATE flag has been used), explicitly activate the
271 * ALQ here so that the pending data gets flushed by the ald_daemon.
273 if (!(alq->aq_flags & AQ_ACTIVE) && HAS_PENDING_DATA(alq)) {
274 alq->aq_flags |= AQ_ACTIVE;
283 while (alq->aq_flags & AQ_ACTIVE) {
284 alq->aq_flags |= AQ_WANTED;
285 msleep_spin(alq, &alq->aq_mtx, "aldclose", 0);
289 vn_close(alq->aq_vp, FWRITE, alq->aq_cred,
291 crfree(alq->aq_cred);
295 alq_destroy(struct alq *alq)
297 /* Drain all pending IO. */
300 mtx_destroy(&alq->aq_mtx);
301 free(alq->aq_entbuf, M_ALD);
306 * Flush all pending data to disk. This operation will block.
309 alq_doio(struct alq *alq)
315 struct iovec aiov[2];
320 KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__));
326 wrapearly = alq->aq_wrapearly;
328 bzero(&aiov, sizeof(aiov));
329 bzero(&auio, sizeof(auio));
331 /* Start the write from the location of our buffer tail pointer. */
332 aiov[0].iov_base = alq->aq_entbuf + alq->aq_writetail;
334 if (alq->aq_writetail < alq->aq_writehead) {
335 /* Buffer not wrapped. */
336 totlen = aiov[0].iov_len = alq->aq_writehead - alq->aq_writetail;
337 } else if (alq->aq_writehead == 0) {
338 /* Buffer not wrapped (special case to avoid an empty iov). */
339 totlen = aiov[0].iov_len = alq->aq_buflen - alq->aq_writetail -
343 * Buffer wrapped, requires 2 aiov entries:
344 * - first is from writetail to end of buffer
345 * - second is from start of buffer to writehead
347 aiov[0].iov_len = alq->aq_buflen - alq->aq_writetail -
350 aiov[1].iov_base = alq->aq_entbuf;
351 aiov[1].iov_len = alq->aq_writehead;
352 totlen = aiov[0].iov_len + aiov[1].iov_len;
355 alq->aq_flags |= AQ_FLUSHING;
358 auio.uio_iov = &aiov[0];
360 auio.uio_segflg = UIO_SYSSPACE;
361 auio.uio_rw = UIO_WRITE;
362 auio.uio_iovcnt = iov;
363 auio.uio_resid = totlen;
367 * Do all of the junk required to write now.
369 vn_start_write(vp, &mp, V_WAIT);
370 vn_lock(vp, LK_EXCLUSIVE | LK_RETRY);
372 * XXX: VOP_WRITE error checks are ignored.
375 if (mac_vnode_check_write(alq->aq_cred, NOCRED, vp) == 0)
377 VOP_WRITE(vp, &auio, IO_UNIT | IO_APPEND, alq->aq_cred);
379 vn_finished_write(mp);
382 alq->aq_flags &= ~AQ_FLUSHING;
384 /* Adjust writetail as required, taking into account wrapping. */
385 alq->aq_writetail = (alq->aq_writetail + totlen + wrapearly) %
387 alq->aq_freebytes += totlen + wrapearly;
390 * If we just flushed part of the buffer which wrapped, reset the
391 * wrapearly indicator.
394 alq->aq_wrapearly = 0;
397 * If we just flushed the buffer completely, reset indexes to 0 to
398 * minimise buffer wraps.
399 * This is also required to ensure alq_getn() can't wedge itself.
401 if (!HAS_PENDING_DATA(alq))
402 alq->aq_writehead = alq->aq_writetail = 0;
404 KASSERT((alq->aq_writetail >= 0 && alq->aq_writetail < alq->aq_buflen),
405 ("%s: aq_writetail < 0 || aq_writetail >= aq_buflen", __func__));
407 if (alq->aq_flags & AQ_WANTED) {
408 alq->aq_flags &= ~AQ_WANTED;
415 static struct kproc_desc ald_kp = {
421 SYSINIT(aldthread, SI_SUB_KTHREAD_IDLE, SI_ORDER_ANY, kproc_start, &ald_kp);
422 SYSINIT(ald, SI_SUB_LOCK, SI_ORDER_ANY, ald_startup, NULL);
424 /* User visible queue functions */
427 * Create the queue data structure, allocate the buffer, and open the file.
431 alq_open_flags(struct alq **alqp, const char *file, struct ucred *cred, int cmode,
439 KASSERT((size > 0), ("%s: size <= 0", __func__));
443 NDINIT(&nd, LOOKUP, NOFOLLOW, UIO_SYSSPACE, file);
444 oflags = FWRITE | O_NOFOLLOW | O_CREAT;
446 error = vn_open_cred(&nd, &oflags, cmode, 0, cred, NULL);
450 NDFREE(&nd, NDF_ONLY_PNBUF);
451 /* We just unlock so we hold a reference */
452 VOP_UNLOCK(nd.ni_vp);
454 alq = malloc(sizeof(*alq), M_ALD, M_WAITOK|M_ZERO);
455 alq->aq_vp = nd.ni_vp;
456 alq->aq_cred = crhold(cred);
458 mtx_init(&alq->aq_mtx, "ALD Queue", NULL, MTX_SPIN|MTX_QUIET);
460 alq->aq_buflen = size;
464 alq->aq_freebytes = alq->aq_buflen;
465 alq->aq_entbuf = malloc(alq->aq_buflen, M_ALD, M_WAITOK|M_ZERO);
466 alq->aq_writehead = alq->aq_writetail = 0;
467 if (flags & ALQ_ORDERED)
468 alq->aq_flags |= AQ_ORDERED;
470 if ((error = ald_add(alq)) != 0) {
481 alq_open(struct alq **alqp, const char *file, struct ucred *cred, int cmode,
486 KASSERT((count >= 0), ("%s: count < 0", __func__));
489 if ((ret = alq_open_flags(alqp, file, cred, cmode,
490 size*count, 0)) == 0) {
491 (*alqp)->aq_flags |= AQ_LEGACY;
492 (*alqp)->aq_entmax = count;
493 (*alqp)->aq_entlen = size;
496 ret = alq_open_flags(alqp, file, cred, cmode, size, 0);
502 * Copy a new entry into the queue. If the operation would block either
503 * wait or return an error depending on the value of waitok.
506 alq_writen(struct alq *alq, void *data, int len, int flags)
508 int activate, copy, ret;
511 KASSERT((len > 0 && len <= alq->aq_buflen),
512 ("%s: len <= 0 || len > aq_buflen", __func__));
521 * Fail to perform the write and return EWOULDBLOCK if:
522 * - The message is larger than our underlying buffer.
523 * - The ALQ is being shutdown.
524 * - There is insufficient free space in our underlying buffer
525 * to accept the message and the user can't wait for space.
526 * - There is insufficient free space in our underlying buffer
527 * to accept the message and the alq is inactive due to prior
528 * use of the ALQ_NOACTIVATE flag (which would lead to deadlock).
530 if (len > alq->aq_buflen ||
531 alq->aq_flags & AQ_SHUTDOWN ||
532 (((flags & ALQ_NOWAIT) || (!(alq->aq_flags & AQ_ACTIVE) &&
533 HAS_PENDING_DATA(alq))) && alq->aq_freebytes < len)) {
535 return (EWOULDBLOCK);
539 * If we want ordered writes and there is already at least one thread
540 * waiting for resources to become available, sleep until we're woken.
542 if (alq->aq_flags & AQ_ORDERED && alq->aq_waiters > 0) {
543 KASSERT(!(flags & ALQ_NOWAIT),
544 ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__));
546 msleep_spin(&alq->aq_waiters, &alq->aq_mtx, "alqwnord", 0);
551 * (ALQ_WAITOK && aq_freebytes < len) or aq_freebytes >= len, either
552 * enter while loop and sleep until we have enough free bytes (former)
553 * or skip (latter). If AQ_ORDERED is set, only 1 thread at a time will
554 * be in this loop. Otherwise, multiple threads may be sleeping here
555 * competing for ALQ resources.
557 while (alq->aq_freebytes < len && !(alq->aq_flags & AQ_SHUTDOWN)) {
558 KASSERT(!(flags & ALQ_NOWAIT),
559 ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__));
560 alq->aq_flags |= AQ_WANTED;
564 msleep_spin(alq, &alq->aq_mtx, "alqwnres", 0);
568 * If we're the first thread to wake after an AQ_WANTED wakeup
569 * but there isn't enough free space for us, we're going to loop
570 * and sleep again. If there are other threads waiting in this
571 * loop, schedule a wakeup so that they can see if the space
572 * they require is available.
574 if (alq->aq_waiters > 0 && !(alq->aq_flags & AQ_ORDERED) &&
575 alq->aq_freebytes < len && !(alq->aq_flags & AQ_WANTED))
582 * If there are waiters, we need to signal the waiting threads after we
583 * complete our work. The alq ptr is used as a wait channel for threads
584 * requiring resources to be freed up. In the AQ_ORDERED case, threads
585 * are not allowed to concurrently compete for resources in the above
586 * while loop, so we use a different wait channel in this case.
588 if (alq->aq_waiters > 0) {
589 if (alq->aq_flags & AQ_ORDERED)
590 waitchan = &alq->aq_waiters;
596 /* Bail if we're shutting down. */
597 if (alq->aq_flags & AQ_SHUTDOWN) {
603 * If we need to wrap the buffer to accommodate the write,
604 * we'll need 2 calls to bcopy.
606 if ((alq->aq_buflen - alq->aq_writehead) < len)
607 copy = alq->aq_buflen - alq->aq_writehead;
609 /* Copy message (or part thereof if wrap required) to the buffer. */
610 bcopy(data, alq->aq_entbuf + alq->aq_writehead, copy);
611 alq->aq_writehead += copy;
613 if (alq->aq_writehead >= alq->aq_buflen) {
614 KASSERT((alq->aq_writehead == alq->aq_buflen),
615 ("%s: alq->aq_writehead (%d) > alq->aq_buflen (%d)",
619 alq->aq_writehead = 0;
624 * Wrap the buffer by copying the remainder of our message
625 * to the start of the buffer and resetting aq_writehead.
627 bcopy(((uint8_t *)data)+copy, alq->aq_entbuf, len - copy);
628 alq->aq_writehead = len - copy;
631 KASSERT((alq->aq_writehead >= 0 && alq->aq_writehead < alq->aq_buflen),
632 ("%s: aq_writehead < 0 || aq_writehead >= aq_buflen", __func__));
634 alq->aq_freebytes -= len;
636 if (!(alq->aq_flags & AQ_ACTIVE) && !(flags & ALQ_NOACTIVATE)) {
637 alq->aq_flags |= AQ_ACTIVE;
641 KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__));
652 /* NB: We rely on wakeup_one waking threads in a FIFO manner. */
653 if (waitchan != NULL)
654 wakeup_one(waitchan);
660 alq_write(struct alq *alq, void *data, int flags)
662 /* Should only be called in fixed length message (legacy) mode. */
663 KASSERT((alq->aq_flags & AQ_LEGACY),
664 ("%s: fixed length write on variable length queue", __func__));
665 return (alq_writen(alq, data, alq->aq_entlen, flags));
669 * Retrieve a pointer for the ALQ to write directly into, avoiding bcopy.
672 alq_getn(struct alq *alq, int len, int flags)
677 KASSERT((len > 0 && len <= alq->aq_buflen),
678 ("%s: len <= 0 || len > alq->aq_buflen", __func__));
685 * Determine the number of free contiguous bytes.
686 * We ensure elsewhere that if aq_writehead == aq_writetail because
687 * the buffer is empty, they will both be set to 0 and therefore
688 * aq_freebytes == aq_buflen and is fully contiguous.
689 * If they are equal and the buffer is not empty, aq_freebytes will
690 * be 0 indicating the buffer is full.
692 if (alq->aq_writehead <= alq->aq_writetail)
693 contigbytes = alq->aq_freebytes;
695 contigbytes = alq->aq_buflen - alq->aq_writehead;
697 if (contigbytes < len) {
699 * Insufficient space at end of buffer to handle a
700 * contiguous write. Wrap early if there's space at
701 * the beginning. This will leave a hole at the end
702 * of the buffer which we will have to skip over when
703 * flushing the buffer to disk.
705 if (alq->aq_writetail >= len || flags & ALQ_WAITOK) {
706 /* Keep track of # bytes left blank. */
707 alq->aq_wrapearly = contigbytes;
708 /* Do the wrap and adjust counters. */
709 contigbytes = alq->aq_freebytes =
711 alq->aq_writehead = 0;
717 * Return a NULL ALE if:
718 * - The message is larger than our underlying buffer.
719 * - The ALQ is being shutdown.
720 * - There is insufficient free space in our underlying buffer
721 * to accept the message and the user can't wait for space.
722 * - There is insufficient free space in our underlying buffer
723 * to accept the message and the alq is inactive due to prior
724 * use of the ALQ_NOACTIVATE flag (which would lead to deadlock).
726 if (len > alq->aq_buflen ||
727 alq->aq_flags & AQ_SHUTDOWN ||
728 (((flags & ALQ_NOWAIT) || (!(alq->aq_flags & AQ_ACTIVE) &&
729 HAS_PENDING_DATA(alq))) && contigbytes < len)) {
735 * If we want ordered writes and there is already at least one thread
736 * waiting for resources to become available, sleep until we're woken.
738 if (alq->aq_flags & AQ_ORDERED && alq->aq_waiters > 0) {
739 KASSERT(!(flags & ALQ_NOWAIT),
740 ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__));
742 msleep_spin(&alq->aq_waiters, &alq->aq_mtx, "alqgnord", 0);
747 * (ALQ_WAITOK && contigbytes < len) or contigbytes >= len, either enter
748 * while loop and sleep until we have enough contiguous free bytes
749 * (former) or skip (latter). If AQ_ORDERED is set, only 1 thread at a
750 * time will be in this loop. Otherwise, multiple threads may be
751 * sleeping here competing for ALQ resources.
753 while (contigbytes < len && !(alq->aq_flags & AQ_SHUTDOWN)) {
754 KASSERT(!(flags & ALQ_NOWAIT),
755 ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__));
756 alq->aq_flags |= AQ_WANTED;
760 msleep_spin(alq, &alq->aq_mtx, "alqgnres", 0);
763 if (alq->aq_writehead <= alq->aq_writetail)
764 contigbytes = alq->aq_freebytes;
766 contigbytes = alq->aq_buflen - alq->aq_writehead;
769 * If we're the first thread to wake after an AQ_WANTED wakeup
770 * but there isn't enough free space for us, we're going to loop
771 * and sleep again. If there are other threads waiting in this
772 * loop, schedule a wakeup so that they can see if the space
773 * they require is available.
775 if (alq->aq_waiters > 0 && !(alq->aq_flags & AQ_ORDERED) &&
776 contigbytes < len && !(alq->aq_flags & AQ_WANTED))
783 * If there are waiters, we need to signal the waiting threads after we
784 * complete our work. The alq ptr is used as a wait channel for threads
785 * requiring resources to be freed up. In the AQ_ORDERED case, threads
786 * are not allowed to concurrently compete for resources in the above
787 * while loop, so we use a different wait channel in this case.
789 if (alq->aq_waiters > 0) {
790 if (alq->aq_flags & AQ_ORDERED)
791 waitchan = &alq->aq_waiters;
797 /* Bail if we're shutting down. */
798 if (alq->aq_flags & AQ_SHUTDOWN) {
800 if (waitchan != NULL)
801 wakeup_one(waitchan);
806 * If we are here, we have a contiguous number of bytes >= len
807 * available in our buffer starting at aq_writehead.
809 alq->aq_getpost.ae_data = alq->aq_entbuf + alq->aq_writehead;
810 alq->aq_getpost.ae_bytesused = len;
812 return (&alq->aq_getpost);
816 alq_get(struct alq *alq, int flags)
818 /* Should only be called in fixed length message (legacy) mode. */
819 KASSERT((alq->aq_flags & AQ_LEGACY),
820 ("%s: fixed length get on variable length queue", __func__));
821 return (alq_getn(alq, alq->aq_entlen, flags));
825 alq_post_flags(struct alq *alq, struct ale *ale, int flags)
832 if (ale->ae_bytesused > 0) {
833 if (!(alq->aq_flags & AQ_ACTIVE) &&
834 !(flags & ALQ_NOACTIVATE)) {
835 alq->aq_flags |= AQ_ACTIVE;
839 alq->aq_writehead += ale->ae_bytesused;
840 alq->aq_freebytes -= ale->ae_bytesused;
842 /* Wrap aq_writehead if we filled to the end of the buffer. */
843 if (alq->aq_writehead == alq->aq_buflen)
844 alq->aq_writehead = 0;
846 KASSERT((alq->aq_writehead >= 0 &&
847 alq->aq_writehead < alq->aq_buflen),
848 ("%s: aq_writehead < 0 || aq_writehead >= aq_buflen",
851 KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__));
855 * If there are waiters, we need to signal the waiting threads after we
856 * complete our work. The alq ptr is used as a wait channel for threads
857 * requiring resources to be freed up. In the AQ_ORDERED case, threads
858 * are not allowed to concurrently compete for resources in the
859 * alq_getn() while loop, so we use a different wait channel in this case.
861 if (alq->aq_waiters > 0) {
862 if (alq->aq_flags & AQ_ORDERED)
863 waitchan = &alq->aq_waiters;
877 /* NB: We rely on wakeup_one waking threads in a FIFO manner. */
878 if (waitchan != NULL)
879 wakeup_one(waitchan);
883 alq_flush(struct alq *alq)
891 * Pull the lever iff there is data to flush and we're
892 * not already in the middle of a flush operation.
894 if (HAS_PENDING_DATA(alq) && !(alq->aq_flags & AQ_FLUSHING)) {
895 if (alq->aq_flags & AQ_ACTIVE)
899 needwakeup = alq_doio(alq);
910 * Flush remaining data, close the file and free all resources.
913 alq_close(struct alq *alq)
915 /* Only flush and destroy alq if not already shutting down. */
916 if (ald_rem(alq) == 0)
921 alq_load_handler(module_t mod, int what, void *arg)
934 /* Only allow unload if there are no open queues. */
935 if (LIST_FIRST(&ald_queues) == NULL) {
938 EVENTHANDLER_DEREGISTER(shutdown_pre_sync,
939 alq_eventhandler_tag);
940 ald_shutdown(NULL, 0);
941 mtx_destroy(&ald_mtx);
949 /* If MOD_QUIESCE failed we must fail here too. */
950 if (ald_shutingdown == 0)
962 static moduledata_t alq_mod =
969 DECLARE_MODULE(alq, alq_mod, SI_SUB_LAST, SI_ORDER_ANY);
970 MODULE_VERSION(alq, 1);