diff options
Diffstat (limited to 'sys/kern/kern_alq.c')
| -rw-r--r-- | sys/kern/kern_alq.c | 968 |
1 files changed, 968 insertions, 0 deletions
diff --git a/sys/kern/kern_alq.c b/sys/kern/kern_alq.c new file mode 100644 index 0000000..893d2f8 --- /dev/null +++ b/sys/kern/kern_alq.c @@ -0,0 +1,968 @@ +/*- + * Copyright (c) 2002, Jeffrey Roberson <jeff@freebsd.org> + * Copyright (c) 2008-2009, Lawrence Stewart <lstewart@freebsd.org> + * Copyright (c) 2009-2010, The FreeBSD Foundation + * All rights reserved. + * + * Portions of this software were developed at the Centre for Advanced + * Internet Architectures, Swinburne University of Technology, Melbourne, + * Australia by Lawrence Stewart under sponsorship from the FreeBSD Foundation. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions + * are met: + * 1. Redistributions of source code must retain the above copyright + * notice unmodified, this list of conditions, and the following + * disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * + * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR + * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES + * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. + * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, + * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT + * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF + * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ + +#include <sys/cdefs.h> +__FBSDID("$FreeBSD$"); + +#include "opt_mac.h" + +#include <sys/param.h> +#include <sys/systm.h> +#include <sys/kernel.h> +#include <sys/kthread.h> +#include <sys/lock.h> +#include <sys/mount.h> +#include <sys/mutex.h> +#include <sys/namei.h> +#include <sys/proc.h> +#include <sys/vnode.h> +#include <sys/alq.h> +#include <sys/malloc.h> +#include <sys/unistd.h> +#include <sys/fcntl.h> +#include <sys/eventhandler.h> + +#include <security/mac/mac_framework.h> + +/* Async. Logging Queue */ +struct alq { + char *aq_entbuf; /* Buffer for stored entries */ + int aq_entmax; /* Max entries */ + int aq_entlen; /* Entry length */ + int aq_freebytes; /* Bytes available in buffer */ + int aq_buflen; /* Total length of our buffer */ + int aq_writehead; /* Location for next write */ + int aq_writetail; /* Flush starts at this location */ + int aq_wrapearly; /* # bytes left blank at end of buf */ + int aq_flags; /* Queue flags */ + int aq_waiters; /* Num threads waiting for resources + * NB: Used as a wait channel so must + * not be first field in the alq struct + */ + struct ale aq_getpost; /* ALE for use by get/post */ + struct mtx aq_mtx; /* Queue lock */ + struct vnode *aq_vp; /* Open vnode handle */ + struct ucred *aq_cred; /* Credentials of the opening thread */ + LIST_ENTRY(alq) aq_act; /* List of active queues */ + LIST_ENTRY(alq) aq_link; /* List of all queues */ +}; + +#define AQ_WANTED 0x0001 /* Wakeup sleeper when io is done */ +#define AQ_ACTIVE 0x0002 /* on the active list */ +#define AQ_FLUSHING 0x0004 /* doing IO */ +#define AQ_SHUTDOWN 0x0008 /* Queue no longer valid */ +#define AQ_ORDERED 0x0010 /* Queue enforces ordered writes */ +#define AQ_LEGACY 0x0020 /* Legacy queue (fixed length writes) */ + +#define ALQ_LOCK(alq) mtx_lock_spin(&(alq)->aq_mtx) +#define ALQ_UNLOCK(alq) mtx_unlock_spin(&(alq)->aq_mtx) + +#define HAS_PENDING_DATA(alq) ((alq)->aq_freebytes != (alq)->aq_buflen) + +static MALLOC_DEFINE(M_ALD, "ALD", "ALD"); + +/* + * The ald_mtx protects the ald_queues list and the ald_active list. + */ +static struct mtx ald_mtx; +static LIST_HEAD(, alq) ald_queues; +static LIST_HEAD(, alq) ald_active; +static int ald_shutingdown = 0; +struct thread *ald_thread; +static struct proc *ald_proc; + +#define ALD_LOCK() mtx_lock(&ald_mtx) +#define ALD_UNLOCK() mtx_unlock(&ald_mtx) + +/* Daemon functions */ +static int ald_add(struct alq *); +static int ald_rem(struct alq *); +static void ald_startup(void *); +static void ald_daemon(void); +static void ald_shutdown(void *, int); +static void ald_activate(struct alq *); +static void ald_deactivate(struct alq *); + +/* Internal queue functions */ +static void alq_shutdown(struct alq *); +static void alq_destroy(struct alq *); +static int alq_doio(struct alq *); + + +/* + * Add a new queue to the global list. Fail if we're shutting down. + */ +static int +ald_add(struct alq *alq) +{ + int error; + + error = 0; + + ALD_LOCK(); + if (ald_shutingdown) { + error = EBUSY; + goto done; + } + LIST_INSERT_HEAD(&ald_queues, alq, aq_link); +done: + ALD_UNLOCK(); + return (error); +} + +/* + * Remove a queue from the global list unless we're shutting down. If so, + * the ald will take care of cleaning up it's resources. + */ +static int +ald_rem(struct alq *alq) +{ + int error; + + error = 0; + + ALD_LOCK(); + if (ald_shutingdown) { + error = EBUSY; + goto done; + } + LIST_REMOVE(alq, aq_link); +done: + ALD_UNLOCK(); + return (error); +} + +/* + * Put a queue on the active list. This will schedule it for writing. + */ +static void +ald_activate(struct alq *alq) +{ + LIST_INSERT_HEAD(&ald_active, alq, aq_act); + wakeup(&ald_active); +} + +static void +ald_deactivate(struct alq *alq) +{ + LIST_REMOVE(alq, aq_act); + alq->aq_flags &= ~AQ_ACTIVE; +} + +static void +ald_startup(void *unused) +{ + mtx_init(&ald_mtx, "ALDmtx", NULL, MTX_DEF|MTX_QUIET); + LIST_INIT(&ald_queues); + LIST_INIT(&ald_active); +} + +static void +ald_daemon(void) +{ + int needwakeup; + struct alq *alq; + + ald_thread = FIRST_THREAD_IN_PROC(ald_proc); + + EVENTHANDLER_REGISTER(shutdown_pre_sync, ald_shutdown, NULL, + SHUTDOWN_PRI_FIRST); + + ALD_LOCK(); + + for (;;) { + while ((alq = LIST_FIRST(&ald_active)) == NULL && + !ald_shutingdown) + mtx_sleep(&ald_active, &ald_mtx, PWAIT, "aldslp", 0); + + /* Don't shutdown until all active ALQs are flushed. */ + if (ald_shutingdown && alq == NULL) { + ALD_UNLOCK(); + break; + } + + ALQ_LOCK(alq); + ald_deactivate(alq); + ALD_UNLOCK(); + needwakeup = alq_doio(alq); + ALQ_UNLOCK(alq); + if (needwakeup) + wakeup_one(alq); + ALD_LOCK(); + } + + kproc_exit(0); +} + +static void +ald_shutdown(void *arg, int howto) +{ + struct alq *alq; + + ALD_LOCK(); + + /* Ensure no new queues can be created. */ + ald_shutingdown = 1; + + /* Shutdown all ALQs prior to terminating the ald_daemon. */ + while ((alq = LIST_FIRST(&ald_queues)) != NULL) { + LIST_REMOVE(alq, aq_link); + ALD_UNLOCK(); + alq_shutdown(alq); + ALD_LOCK(); + } + + /* At this point, all ALQs are flushed and shutdown. */ + + /* + * Wake ald_daemon so that it exits. It won't be able to do + * anything until we mtx_sleep because we hold the ald_mtx. + */ + wakeup(&ald_active); + + /* Wait for ald_daemon to exit. */ + mtx_sleep(ald_proc, &ald_mtx, PWAIT, "aldslp", 0); + + ALD_UNLOCK(); +} + +static void +alq_shutdown(struct alq *alq) +{ + ALQ_LOCK(alq); + + /* Stop any new writers. */ + alq->aq_flags |= AQ_SHUTDOWN; + + /* + * If the ALQ isn't active but has unwritten data (possible if + * the ALQ_NOACTIVATE flag has been used), explicitly activate the + * ALQ here so that the pending data gets flushed by the ald_daemon. + */ + if (!(alq->aq_flags & AQ_ACTIVE) && HAS_PENDING_DATA(alq)) { + alq->aq_flags |= AQ_ACTIVE; + ALQ_UNLOCK(alq); + ALD_LOCK(); + ald_activate(alq); + ALD_UNLOCK(); + ALQ_LOCK(alq); + } + + /* Drain IO */ + while (alq->aq_flags & AQ_ACTIVE) { + alq->aq_flags |= AQ_WANTED; + msleep_spin(alq, &alq->aq_mtx, "aldclose", 0); + } + ALQ_UNLOCK(alq); + + vn_close(alq->aq_vp, FWRITE, alq->aq_cred, + curthread); + crfree(alq->aq_cred); +} + +void +alq_destroy(struct alq *alq) +{ + /* Drain all pending IO. */ + alq_shutdown(alq); + + mtx_destroy(&alq->aq_mtx); + free(alq->aq_entbuf, M_ALD); + free(alq, M_ALD); +} + +/* + * Flush all pending data to disk. This operation will block. + */ +static int +alq_doio(struct alq *alq) +{ + struct thread *td; + struct mount *mp; + struct vnode *vp; + struct uio auio; + struct iovec aiov[2]; + int totlen; + int iov; + int wrapearly; + + KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__)); + + vp = alq->aq_vp; + td = curthread; + totlen = 0; + iov = 1; + wrapearly = alq->aq_wrapearly; + + bzero(&aiov, sizeof(aiov)); + bzero(&auio, sizeof(auio)); + + /* Start the write from the location of our buffer tail pointer. */ + aiov[0].iov_base = alq->aq_entbuf + alq->aq_writetail; + + if (alq->aq_writetail < alq->aq_writehead) { + /* Buffer not wrapped. */ + totlen = aiov[0].iov_len = alq->aq_writehead - alq->aq_writetail; + } else if (alq->aq_writehead == 0) { + /* Buffer not wrapped (special case to avoid an empty iov). */ + totlen = aiov[0].iov_len = alq->aq_buflen - alq->aq_writetail - + wrapearly; + } else { + /* + * Buffer wrapped, requires 2 aiov entries: + * - first is from writetail to end of buffer + * - second is from start of buffer to writehead + */ + aiov[0].iov_len = alq->aq_buflen - alq->aq_writetail - + wrapearly; + iov++; + aiov[1].iov_base = alq->aq_entbuf; + aiov[1].iov_len = alq->aq_writehead; + totlen = aiov[0].iov_len + aiov[1].iov_len; + } + + alq->aq_flags |= AQ_FLUSHING; + ALQ_UNLOCK(alq); + + auio.uio_iov = &aiov[0]; + auio.uio_offset = 0; + auio.uio_segflg = UIO_SYSSPACE; + auio.uio_rw = UIO_WRITE; + auio.uio_iovcnt = iov; + auio.uio_resid = totlen; + auio.uio_td = td; + + /* + * Do all of the junk required to write now. + */ + vn_start_write(vp, &mp, V_WAIT); + vn_lock(vp, LK_EXCLUSIVE | LK_RETRY); + /* + * XXX: VOP_WRITE error checks are ignored. + */ +#ifdef MAC + if (mac_vnode_check_write(alq->aq_cred, NOCRED, vp) == 0) +#endif + VOP_WRITE(vp, &auio, IO_UNIT | IO_APPEND, alq->aq_cred); + VOP_UNLOCK(vp, 0); + vn_finished_write(mp); + + ALQ_LOCK(alq); + alq->aq_flags &= ~AQ_FLUSHING; + + /* Adjust writetail as required, taking into account wrapping. */ + alq->aq_writetail = (alq->aq_writetail + totlen + wrapearly) % + alq->aq_buflen; + alq->aq_freebytes += totlen + wrapearly; + + /* + * If we just flushed part of the buffer which wrapped, reset the + * wrapearly indicator. + */ + if (wrapearly) + alq->aq_wrapearly = 0; + + /* + * If we just flushed the buffer completely, reset indexes to 0 to + * minimise buffer wraps. + * This is also required to ensure alq_getn() can't wedge itself. + */ + if (!HAS_PENDING_DATA(alq)) + alq->aq_writehead = alq->aq_writetail = 0; + + KASSERT((alq->aq_writetail >= 0 && alq->aq_writetail < alq->aq_buflen), + ("%s: aq_writetail < 0 || aq_writetail >= aq_buflen", __func__)); + + if (alq->aq_flags & AQ_WANTED) { + alq->aq_flags &= ~AQ_WANTED; + return (1); + } + + return(0); +} + +static struct kproc_desc ald_kp = { + "ALQ Daemon", + ald_daemon, + &ald_proc +}; + +SYSINIT(aldthread, SI_SUB_KTHREAD_IDLE, SI_ORDER_ANY, kproc_start, &ald_kp); +SYSINIT(ald, SI_SUB_LOCK, SI_ORDER_ANY, ald_startup, NULL); + + +/* User visible queue functions */ + +/* + * Create the queue data structure, allocate the buffer, and open the file. + */ + +int +alq_open_flags(struct alq **alqp, const char *file, struct ucred *cred, int cmode, + int size, int flags) +{ + struct thread *td; + struct nameidata nd; + struct alq *alq; + int oflags; + int error; + + KASSERT((size > 0), ("%s: size <= 0", __func__)); + + *alqp = NULL; + td = curthread; + + NDINIT(&nd, LOOKUP, NOFOLLOW, UIO_SYSSPACE, file, td); + oflags = FWRITE | O_NOFOLLOW | O_CREAT; + + error = vn_open_cred(&nd, &oflags, cmode, 0, cred, NULL); + if (error) + return (error); + + NDFREE(&nd, NDF_ONLY_PNBUF); + /* We just unlock so we hold a reference */ + VOP_UNLOCK(nd.ni_vp, 0); + + alq = malloc(sizeof(*alq), M_ALD, M_WAITOK|M_ZERO); + alq->aq_vp = nd.ni_vp; + alq->aq_cred = crhold(cred); + + mtx_init(&alq->aq_mtx, "ALD Queue", NULL, MTX_SPIN|MTX_QUIET); + + alq->aq_buflen = size; + alq->aq_entmax = 0; + alq->aq_entlen = 0; + + alq->aq_freebytes = alq->aq_buflen; + alq->aq_entbuf = malloc(alq->aq_buflen, M_ALD, M_WAITOK|M_ZERO); + alq->aq_writehead = alq->aq_writetail = 0; + if (flags & ALQ_ORDERED) + alq->aq_flags |= AQ_ORDERED; + + if ((error = ald_add(alq)) != 0) { + alq_destroy(alq); + return (error); + } + + *alqp = alq; + + return (0); +} + +int +alq_open(struct alq **alqp, const char *file, struct ucred *cred, int cmode, + int size, int count) +{ + int ret; + + KASSERT((count >= 0), ("%s: count < 0", __func__)); + + if (count > 0) { + ret = alq_open_flags(alqp, file, cred, cmode, size*count, 0); + (*alqp)->aq_flags |= AQ_LEGACY; + (*alqp)->aq_entmax = count; + (*alqp)->aq_entlen = size; + } else + ret = alq_open_flags(alqp, file, cred, cmode, size, 0); + + return (ret); +} + + +/* + * Copy a new entry into the queue. If the operation would block either + * wait or return an error depending on the value of waitok. + */ +int +alq_writen(struct alq *alq, void *data, int len, int flags) +{ + int activate, copy, ret; + void *waitchan; + + KASSERT((len > 0 && len <= alq->aq_buflen), + ("%s: len <= 0 || len > aq_buflen", __func__)); + + activate = ret = 0; + copy = len; + waitchan = NULL; + + ALQ_LOCK(alq); + + /* + * Fail to perform the write and return EWOULDBLOCK if: + * - The message is larger than our underlying buffer. + * - The ALQ is being shutdown. + * - There is insufficient free space in our underlying buffer + * to accept the message and the user can't wait for space. + * - There is insufficient free space in our underlying buffer + * to accept the message and the alq is inactive due to prior + * use of the ALQ_NOACTIVATE flag (which would lead to deadlock). + */ + if (len > alq->aq_buflen || + alq->aq_flags & AQ_SHUTDOWN || + (((flags & ALQ_NOWAIT) || (!(alq->aq_flags & AQ_ACTIVE) && + HAS_PENDING_DATA(alq))) && alq->aq_freebytes < len)) { + ALQ_UNLOCK(alq); + return (EWOULDBLOCK); + } + + /* + * If we want ordered writes and there is already at least one thread + * waiting for resources to become available, sleep until we're woken. + */ + if (alq->aq_flags & AQ_ORDERED && alq->aq_waiters > 0) { + KASSERT(!(flags & ALQ_NOWAIT), + ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__)); + alq->aq_waiters++; + msleep_spin(&alq->aq_waiters, &alq->aq_mtx, "alqwnord", 0); + alq->aq_waiters--; + } + + /* + * (ALQ_WAITOK && aq_freebytes < len) or aq_freebytes >= len, either + * enter while loop and sleep until we have enough free bytes (former) + * or skip (latter). If AQ_ORDERED is set, only 1 thread at a time will + * be in this loop. Otherwise, multiple threads may be sleeping here + * competing for ALQ resources. + */ + while (alq->aq_freebytes < len && !(alq->aq_flags & AQ_SHUTDOWN)) { + KASSERT(!(flags & ALQ_NOWAIT), + ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__)); + alq->aq_flags |= AQ_WANTED; + alq->aq_waiters++; + if (waitchan) + wakeup(waitchan); + msleep_spin(alq, &alq->aq_mtx, "alqwnres", 0); + alq->aq_waiters--; + + /* + * If we're the first thread to wake after an AQ_WANTED wakeup + * but there isn't enough free space for us, we're going to loop + * and sleep again. If there are other threads waiting in this + * loop, schedule a wakeup so that they can see if the space + * they require is available. + */ + if (alq->aq_waiters > 0 && !(alq->aq_flags & AQ_ORDERED) && + alq->aq_freebytes < len && !(alq->aq_flags & AQ_WANTED)) + waitchan = alq; + else + waitchan = NULL; + } + + /* + * If there are waiters, we need to signal the waiting threads after we + * complete our work. The alq ptr is used as a wait channel for threads + * requiring resources to be freed up. In the AQ_ORDERED case, threads + * are not allowed to concurrently compete for resources in the above + * while loop, so we use a different wait channel in this case. + */ + if (alq->aq_waiters > 0) { + if (alq->aq_flags & AQ_ORDERED) + waitchan = &alq->aq_waiters; + else + waitchan = alq; + } else + waitchan = NULL; + + /* Bail if we're shutting down. */ + if (alq->aq_flags & AQ_SHUTDOWN) { + ret = EWOULDBLOCK; + goto unlock; + } + + /* + * If we need to wrap the buffer to accommodate the write, + * we'll need 2 calls to bcopy. + */ + if ((alq->aq_buflen - alq->aq_writehead) < len) + copy = alq->aq_buflen - alq->aq_writehead; + + /* Copy message (or part thereof if wrap required) to the buffer. */ + bcopy(data, alq->aq_entbuf + alq->aq_writehead, copy); + alq->aq_writehead += copy; + + if (alq->aq_writehead >= alq->aq_buflen) { + KASSERT((alq->aq_writehead == alq->aq_buflen), + ("%s: alq->aq_writehead (%d) > alq->aq_buflen (%d)", + __func__, + alq->aq_writehead, + alq->aq_buflen)); + alq->aq_writehead = 0; + } + + if (copy != len) { + /* + * Wrap the buffer by copying the remainder of our message + * to the start of the buffer and resetting aq_writehead. + */ + bcopy(((uint8_t *)data)+copy, alq->aq_entbuf, len - copy); + alq->aq_writehead = len - copy; + } + + KASSERT((alq->aq_writehead >= 0 && alq->aq_writehead < alq->aq_buflen), + ("%s: aq_writehead < 0 || aq_writehead >= aq_buflen", __func__)); + + alq->aq_freebytes -= len; + + if (!(alq->aq_flags & AQ_ACTIVE) && !(flags & ALQ_NOACTIVATE)) { + alq->aq_flags |= AQ_ACTIVE; + activate = 1; + } + + KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__)); + +unlock: + ALQ_UNLOCK(alq); + + if (activate) { + ALD_LOCK(); + ald_activate(alq); + ALD_UNLOCK(); + } + + /* NB: We rely on wakeup_one waking threads in a FIFO manner. */ + if (waitchan != NULL) + wakeup_one(waitchan); + + return (ret); +} + +int +alq_write(struct alq *alq, void *data, int flags) +{ + /* Should only be called in fixed length message (legacy) mode. */ + KASSERT((alq->aq_flags & AQ_LEGACY), + ("%s: fixed length write on variable length queue", __func__)); + return (alq_writen(alq, data, alq->aq_entlen, flags)); +} + +/* + * Retrieve a pointer for the ALQ to write directly into, avoiding bcopy. + */ +struct ale * +alq_getn(struct alq *alq, int len, int flags) +{ + int contigbytes; + void *waitchan; + + KASSERT((len > 0 && len <= alq->aq_buflen), + ("%s: len <= 0 || len > alq->aq_buflen", __func__)); + + waitchan = NULL; + + ALQ_LOCK(alq); + + /* + * Determine the number of free contiguous bytes. + * We ensure elsewhere that if aq_writehead == aq_writetail because + * the buffer is empty, they will both be set to 0 and therefore + * aq_freebytes == aq_buflen and is fully contiguous. + * If they are equal and the buffer is not empty, aq_freebytes will + * be 0 indicating the buffer is full. + */ + if (alq->aq_writehead <= alq->aq_writetail) + contigbytes = alq->aq_freebytes; + else { + contigbytes = alq->aq_buflen - alq->aq_writehead; + + if (contigbytes < len) { + /* + * Insufficient space at end of buffer to handle a + * contiguous write. Wrap early if there's space at + * the beginning. This will leave a hole at the end + * of the buffer which we will have to skip over when + * flushing the buffer to disk. + */ + if (alq->aq_writetail >= len || flags & ALQ_WAITOK) { + /* Keep track of # bytes left blank. */ + alq->aq_wrapearly = contigbytes; + /* Do the wrap and adjust counters. */ + contigbytes = alq->aq_freebytes = + alq->aq_writetail; + alq->aq_writehead = 0; + } + } + } + + /* + * Return a NULL ALE if: + * - The message is larger than our underlying buffer. + * - The ALQ is being shutdown. + * - There is insufficient free space in our underlying buffer + * to accept the message and the user can't wait for space. + * - There is insufficient free space in our underlying buffer + * to accept the message and the alq is inactive due to prior + * use of the ALQ_NOACTIVATE flag (which would lead to deadlock). + */ + if (len > alq->aq_buflen || + alq->aq_flags & AQ_SHUTDOWN || + (((flags & ALQ_NOWAIT) || (!(alq->aq_flags & AQ_ACTIVE) && + HAS_PENDING_DATA(alq))) && contigbytes < len)) { + ALQ_UNLOCK(alq); + return (NULL); + } + + /* + * If we want ordered writes and there is already at least one thread + * waiting for resources to become available, sleep until we're woken. + */ + if (alq->aq_flags & AQ_ORDERED && alq->aq_waiters > 0) { + KASSERT(!(flags & ALQ_NOWAIT), + ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__)); + alq->aq_waiters++; + msleep_spin(&alq->aq_waiters, &alq->aq_mtx, "alqgnord", 0); + alq->aq_waiters--; + } + + /* + * (ALQ_WAITOK && contigbytes < len) or contigbytes >= len, either enter + * while loop and sleep until we have enough contiguous free bytes + * (former) or skip (latter). If AQ_ORDERED is set, only 1 thread at a + * time will be in this loop. Otherwise, multiple threads may be + * sleeping here competing for ALQ resources. + */ + while (contigbytes < len && !(alq->aq_flags & AQ_SHUTDOWN)) { + KASSERT(!(flags & ALQ_NOWAIT), + ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__)); + alq->aq_flags |= AQ_WANTED; + alq->aq_waiters++; + if (waitchan) + wakeup(waitchan); + msleep_spin(alq, &alq->aq_mtx, "alqgnres", 0); + alq->aq_waiters--; + + if (alq->aq_writehead <= alq->aq_writetail) + contigbytes = alq->aq_freebytes; + else + contigbytes = alq->aq_buflen - alq->aq_writehead; + + /* + * If we're the first thread to wake after an AQ_WANTED wakeup + * but there isn't enough free space for us, we're going to loop + * and sleep again. If there are other threads waiting in this + * loop, schedule a wakeup so that they can see if the space + * they require is available. + */ + if (alq->aq_waiters > 0 && !(alq->aq_flags & AQ_ORDERED) && + contigbytes < len && !(alq->aq_flags & AQ_WANTED)) + waitchan = alq; + else + waitchan = NULL; + } + + /* + * If there are waiters, we need to signal the waiting threads after we + * complete our work. The alq ptr is used as a wait channel for threads + * requiring resources to be freed up. In the AQ_ORDERED case, threads + * are not allowed to concurrently compete for resources in the above + * while loop, so we use a different wait channel in this case. + */ + if (alq->aq_waiters > 0) { + if (alq->aq_flags & AQ_ORDERED) + waitchan = &alq->aq_waiters; + else + waitchan = alq; + } else + waitchan = NULL; + + /* Bail if we're shutting down. */ + if (alq->aq_flags & AQ_SHUTDOWN) { + ALQ_UNLOCK(alq); + if (waitchan != NULL) + wakeup_one(waitchan); + return (NULL); + } + + /* + * If we are here, we have a contiguous number of bytes >= len + * available in our buffer starting at aq_writehead. + */ + alq->aq_getpost.ae_data = alq->aq_entbuf + alq->aq_writehead; + alq->aq_getpost.ae_bytesused = len; + + return (&alq->aq_getpost); +} + +struct ale * +alq_get(struct alq *alq, int flags) +{ + /* Should only be called in fixed length message (legacy) mode. */ + KASSERT((alq->aq_flags & AQ_LEGACY), + ("%s: fixed length get on variable length queue", __func__)); + return (alq_getn(alq, alq->aq_entlen, flags)); +} + +void +alq_post_flags(struct alq *alq, struct ale *ale, int flags) +{ + int activate; + void *waitchan; + + activate = 0; + + if (ale->ae_bytesused > 0) { + if (!(alq->aq_flags & AQ_ACTIVE) && + !(flags & ALQ_NOACTIVATE)) { + alq->aq_flags |= AQ_ACTIVE; + activate = 1; + } + + alq->aq_writehead += ale->ae_bytesused; + alq->aq_freebytes -= ale->ae_bytesused; + + /* Wrap aq_writehead if we filled to the end of the buffer. */ + if (alq->aq_writehead == alq->aq_buflen) + alq->aq_writehead = 0; + + KASSERT((alq->aq_writehead >= 0 && + alq->aq_writehead < alq->aq_buflen), + ("%s: aq_writehead < 0 || aq_writehead >= aq_buflen", + __func__)); + + KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__)); + } + + /* + * If there are waiters, we need to signal the waiting threads after we + * complete our work. The alq ptr is used as a wait channel for threads + * requiring resources to be freed up. In the AQ_ORDERED case, threads + * are not allowed to concurrently compete for resources in the + * alq_getn() while loop, so we use a different wait channel in this case. + */ + if (alq->aq_waiters > 0) { + if (alq->aq_flags & AQ_ORDERED) + waitchan = &alq->aq_waiters; + else + waitchan = alq; + } else + waitchan = NULL; + + ALQ_UNLOCK(alq); + + if (activate) { + ALD_LOCK(); + ald_activate(alq); + ALD_UNLOCK(); + } + + /* NB: We rely on wakeup_one waking threads in a FIFO manner. */ + if (waitchan != NULL) + wakeup_one(waitchan); +} + +void +alq_flush(struct alq *alq) +{ + int needwakeup = 0; + + ALD_LOCK(); + ALQ_LOCK(alq); + + /* + * Pull the lever iff there is data to flush and we're + * not already in the middle of a flush operation. + */ + if (HAS_PENDING_DATA(alq) && !(alq->aq_flags & AQ_FLUSHING)) { + if (alq->aq_flags & AQ_ACTIVE) + ald_deactivate(alq); + + ALD_UNLOCK(); + needwakeup = alq_doio(alq); + } else + ALD_UNLOCK(); + + ALQ_UNLOCK(alq); + + if (needwakeup) + wakeup_one(alq); +} + +/* + * Flush remaining data, close the file and free all resources. + */ +void +alq_close(struct alq *alq) +{ + /* Only flush and destroy alq if not already shutting down. */ + if (ald_rem(alq) == 0) + alq_destroy(alq); +} + +static int +alq_load_handler(module_t mod, int what, void *arg) +{ + int ret; + + ret = 0; + + switch (what) { + case MOD_LOAD: + case MOD_SHUTDOWN: + break; + + case MOD_QUIESCE: + ALD_LOCK(); + /* Only allow unload if there are no open queues. */ + if (LIST_FIRST(&ald_queues) == NULL) { + ald_shutingdown = 1; + ALD_UNLOCK(); + ald_shutdown(NULL, 0); + mtx_destroy(&ald_mtx); + } else { + ALD_UNLOCK(); + ret = EBUSY; + } + break; + + case MOD_UNLOAD: + /* If MOD_QUIESCE failed we must fail here too. */ + if (ald_shutingdown == 0) + ret = EBUSY; + break; + + default: + ret = EINVAL; + break; + } + + return (ret); +} + +static moduledata_t alq_mod = +{ + "alq", + alq_load_handler, + NULL +}; + +DECLARE_MODULE(alq, alq_mod, SI_SUB_SMP, SI_ORDER_ANY); +MODULE_VERSION(alq, 1); |
