The Design and Implementation of the FreeBSD Operating System, Second Edition
Now available: The Design and Implementation of the FreeBSD Operating System (Second Edition)


[ source navigation ] [ diff markup ] [ identifier search ] [ freetext search ] [ file search ] [ list types ] [ track identifier ]

FreeBSD/Linux Kernel Cross Reference
sys/kern/kern_alq.c

Version: -  FREEBSD  -  FREEBSD-13-STABLE  -  FREEBSD-13-0  -  FREEBSD-12-STABLE  -  FREEBSD-12-0  -  FREEBSD-11-STABLE  -  FREEBSD-11-0  -  FREEBSD-10-STABLE  -  FREEBSD-10-0  -  FREEBSD-9-STABLE  -  FREEBSD-9-0  -  FREEBSD-8-STABLE  -  FREEBSD-8-0  -  FREEBSD-7-STABLE  -  FREEBSD-7-0  -  FREEBSD-6-STABLE  -  FREEBSD-6-0  -  FREEBSD-5-STABLE  -  FREEBSD-5-0  -  FREEBSD-4-STABLE  -  FREEBSD-3-STABLE  -  FREEBSD22  -  l41  -  OPENBSD  -  linux-2.6  -  MK84  -  PLAN9  -  xnu-8792 
SearchContext: -  none  -  3  -  10 

    1 /*-
    2  * SPDX-License-Identifier: BSD-2-Clause-FreeBSD
    3  *
    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
    7  * All rights reserved.
    8  *
    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.
   12  *
   13  * Redistribution and use in source and binary forms, with or without
   14  * modification, are permitted provided that the following conditions
   15  * are met:
   16  * 1. Redistributions of source code must retain the above copyright
   17  *    notice unmodified, this list of conditions, and the following
   18  *    disclaimer.
   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.
   22  *
   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.
   33  */
   34 
   35 #include <sys/cdefs.h>
   36 __FBSDID("$FreeBSD$");
   37 
   38 #include "opt_mac.h"
   39 
   40 #include <sys/param.h>
   41 #include <sys/systm.h>
   42 #include <sys/kernel.h>
   43 #include <sys/kthread.h>
   44 #include <sys/lock.h>
   45 #include <sys/mount.h>
   46 #include <sys/mutex.h>
   47 #include <sys/namei.h>
   48 #include <sys/proc.h>
   49 #include <sys/vnode.h>
   50 #include <sys/alq.h>
   51 #include <sys/malloc.h>
   52 #include <sys/unistd.h>
   53 #include <sys/fcntl.h>
   54 #include <sys/eventhandler.h>
   55 
   56 #include <security/mac/mac_framework.h>
   57 
   58 /* Async. Logging Queue */
   59 struct alq {
   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
   72                                          */
   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 */
   79 };
   80 
   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) */
   87 
   88 #define ALQ_LOCK(alq)   mtx_lock_spin(&(alq)->aq_mtx)
   89 #define ALQ_UNLOCK(alq) mtx_unlock_spin(&(alq)->aq_mtx)
   90 
   91 #define HAS_PENDING_DATA(alq) ((alq)->aq_freebytes != (alq)->aq_buflen)
   92 
   93 static MALLOC_DEFINE(M_ALD, "ALD", "ALD");
   94 
   95 /*
   96  * The ald_mtx protects the ald_queues list and the ald_active list.
   97  */
   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;
  105 
  106 #define ALD_LOCK()      mtx_lock(&ald_mtx)
  107 #define ALD_UNLOCK()    mtx_unlock(&ald_mtx)
  108 
  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 *);
  117 
  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 *);
  122 
  123 /*
  124  * Add a new queue to the global list.  Fail if we're shutting down.
  125  */
  126 static int
  127 ald_add(struct alq *alq)
  128 {
  129         int error;
  130 
  131         error = 0;
  132 
  133         ALD_LOCK();
  134         if (ald_shutingdown) {
  135                 error = EBUSY;
  136                 goto done;
  137         }
  138         LIST_INSERT_HEAD(&ald_queues, alq, aq_link);
  139 done:
  140         ALD_UNLOCK();
  141         return (error);
  142 }
  143 
  144 /*
  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.
  147  */
  148 static int
  149 ald_rem(struct alq *alq)
  150 {
  151         int error;
  152 
  153         error = 0;
  154 
  155         ALD_LOCK();
  156         if (ald_shutingdown) {
  157                 error = EBUSY;
  158                 goto done;
  159         }
  160         LIST_REMOVE(alq, aq_link);
  161 done:
  162         ALD_UNLOCK();
  163         return (error);
  164 }
  165 
  166 /*
  167  * Put a queue on the active list.  This will schedule it for writing.
  168  */
  169 static void
  170 ald_activate(struct alq *alq)
  171 {
  172         LIST_INSERT_HEAD(&ald_active, alq, aq_act);
  173         wakeup(&ald_active);
  174 }
  175 
  176 static void
  177 ald_deactivate(struct alq *alq)
  178 {
  179         LIST_REMOVE(alq, aq_act);
  180         alq->aq_flags &= ~AQ_ACTIVE;
  181 }
  182 
  183 static void
  184 ald_startup(void *unused)
  185 {
  186         mtx_init(&ald_mtx, "ALDmtx", NULL, MTX_DEF|MTX_QUIET);
  187         LIST_INIT(&ald_queues);
  188         LIST_INIT(&ald_active);
  189 }
  190 
  191 static void
  192 ald_daemon(void)
  193 {
  194         int needwakeup;
  195         struct alq *alq;
  196 
  197         ald_thread = FIRST_THREAD_IN_PROC(ald_proc);
  198 
  199         alq_eventhandler_tag = EVENTHANDLER_REGISTER(shutdown_pre_sync,
  200             ald_shutdown, NULL, SHUTDOWN_PRI_FIRST);
  201 
  202         ALD_LOCK();
  203 
  204         for (;;) {
  205                 while ((alq = LIST_FIRST(&ald_active)) == NULL &&
  206                     !ald_shutingdown)
  207                         mtx_sleep(&ald_active, &ald_mtx, PWAIT, "aldslp", 0);
  208 
  209                 /* Don't shutdown until all active ALQs are flushed. */
  210                 if (ald_shutingdown && alq == NULL) {
  211                         ALD_UNLOCK();
  212                         break;
  213                 }
  214 
  215                 ALQ_LOCK(alq);
  216                 ald_deactivate(alq);
  217                 ALD_UNLOCK();
  218                 needwakeup = alq_doio(alq);
  219                 ALQ_UNLOCK(alq);
  220                 if (needwakeup)
  221                         wakeup_one(alq);
  222                 ALD_LOCK();
  223         }
  224 
  225         kproc_exit(0);
  226 }
  227 
  228 static void
  229 ald_shutdown(void *arg, int howto)
  230 {
  231         struct alq *alq;
  232 
  233         ALD_LOCK();
  234 
  235         /* Ensure no new queues can be created. */
  236         ald_shutingdown = 1;
  237 
  238         /* Shutdown all ALQs prior to terminating the ald_daemon. */
  239         while ((alq = LIST_FIRST(&ald_queues)) != NULL) {
  240                 LIST_REMOVE(alq, aq_link);
  241                 ALD_UNLOCK();
  242                 alq_shutdown(alq);
  243                 ALD_LOCK();
  244         }
  245 
  246         /* At this point, all ALQs are flushed and shutdown. */
  247 
  248         /*
  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.
  251          */
  252         wakeup(&ald_active);
  253 
  254         /* Wait for ald_daemon to exit. */
  255         mtx_sleep(ald_proc, &ald_mtx, PWAIT, "aldslp", 0);
  256 
  257         ALD_UNLOCK();
  258 }
  259 
  260 static void
  261 alq_shutdown(struct alq *alq)
  262 {
  263         ALQ_LOCK(alq);
  264 
  265         /* Stop any new writers. */
  266         alq->aq_flags |= AQ_SHUTDOWN;
  267 
  268         /*
  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.
  272          */
  273         if (!(alq->aq_flags & AQ_ACTIVE) && HAS_PENDING_DATA(alq)) {
  274                 alq->aq_flags |= AQ_ACTIVE;
  275                 ALQ_UNLOCK(alq);
  276                 ALD_LOCK();
  277                 ald_activate(alq);
  278                 ALD_UNLOCK();
  279                 ALQ_LOCK(alq);
  280         }
  281 
  282         /* Drain IO */
  283         while (alq->aq_flags & AQ_ACTIVE) {
  284                 alq->aq_flags |= AQ_WANTED;
  285                 msleep_spin(alq, &alq->aq_mtx, "aldclose", 0);
  286         }
  287         ALQ_UNLOCK(alq);
  288 
  289         vn_close(alq->aq_vp, FWRITE, alq->aq_cred,
  290             curthread);
  291         crfree(alq->aq_cred);
  292 }
  293 
  294 void
  295 alq_destroy(struct alq *alq)
  296 {
  297         /* Drain all pending IO. */
  298         alq_shutdown(alq);
  299 
  300         mtx_destroy(&alq->aq_mtx);
  301         free(alq->aq_entbuf, M_ALD);
  302         free(alq, M_ALD);
  303 }
  304 
  305 /*
  306  * Flush all pending data to disk.  This operation will block.
  307  */
  308 static int
  309 alq_doio(struct alq *alq)
  310 {
  311         struct thread *td;
  312         struct mount *mp;
  313         struct vnode *vp;
  314         struct uio auio;
  315         struct iovec aiov[2];
  316         int totlen;
  317         int iov;
  318         int wrapearly;
  319 
  320         KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__));
  321 
  322         vp = alq->aq_vp;
  323         td = curthread;
  324         totlen = 0;
  325         iov = 1;
  326         wrapearly = alq->aq_wrapearly;
  327 
  328         bzero(&aiov, sizeof(aiov));
  329         bzero(&auio, sizeof(auio));
  330 
  331         /* Start the write from the location of our buffer tail pointer. */
  332         aiov[0].iov_base = alq->aq_entbuf + alq->aq_writetail;
  333 
  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 -
  340                     wrapearly;
  341         } else {
  342                 /*
  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
  346                  */
  347                 aiov[0].iov_len = alq->aq_buflen - alq->aq_writetail -
  348                     wrapearly;
  349                 iov++;
  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;
  353         }
  354 
  355         alq->aq_flags |= AQ_FLUSHING;
  356         ALQ_UNLOCK(alq);
  357 
  358         auio.uio_iov = &aiov[0];
  359         auio.uio_offset = 0;
  360         auio.uio_segflg = UIO_SYSSPACE;
  361         auio.uio_rw = UIO_WRITE;
  362         auio.uio_iovcnt = iov;
  363         auio.uio_resid = totlen;
  364         auio.uio_td = td;
  365 
  366         /*
  367          * Do all of the junk required to write now.
  368          */
  369         vn_start_write(vp, &mp, V_WAIT);
  370         vn_lock(vp, LK_EXCLUSIVE | LK_RETRY);
  371         /*
  372          * XXX: VOP_WRITE error checks are ignored.
  373          */
  374 #ifdef MAC
  375         if (mac_vnode_check_write(alq->aq_cred, NOCRED, vp) == 0)
  376 #endif
  377                 VOP_WRITE(vp, &auio, IO_UNIT | IO_APPEND, alq->aq_cred);
  378         VOP_UNLOCK(vp);
  379         vn_finished_write(mp);
  380 
  381         ALQ_LOCK(alq);
  382         alq->aq_flags &= ~AQ_FLUSHING;
  383 
  384         /* Adjust writetail as required, taking into account wrapping. */
  385         alq->aq_writetail = (alq->aq_writetail + totlen + wrapearly) %
  386             alq->aq_buflen;
  387         alq->aq_freebytes += totlen + wrapearly;
  388 
  389         /*
  390          * If we just flushed part of the buffer which wrapped, reset the
  391          * wrapearly indicator.
  392          */
  393         if (wrapearly)
  394                 alq->aq_wrapearly = 0;
  395 
  396         /*
  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.
  400          */
  401         if (!HAS_PENDING_DATA(alq))
  402                 alq->aq_writehead = alq->aq_writetail = 0;
  403 
  404         KASSERT((alq->aq_writetail >= 0 && alq->aq_writetail < alq->aq_buflen),
  405             ("%s: aq_writetail < 0 || aq_writetail >= aq_buflen", __func__));
  406 
  407         if (alq->aq_flags & AQ_WANTED) {
  408                 alq->aq_flags &= ~AQ_WANTED;
  409                 return (1);
  410         }
  411 
  412         return(0);
  413 }
  414 
  415 static struct kproc_desc ald_kp = {
  416         "ALQ Daemon",
  417         ald_daemon,
  418         &ald_proc
  419 };
  420 
  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);
  423 
  424 /* User visible queue functions */
  425 
  426 /*
  427  * Create the queue data structure, allocate the buffer, and open the file.
  428  */
  429 
  430 int
  431 alq_open_flags(struct alq **alqp, const char *file, struct ucred *cred, int cmode,
  432     int size, int flags)
  433 {
  434         struct thread *td;
  435         struct nameidata nd;
  436         struct alq *alq;
  437         int oflags;
  438         int error;
  439 
  440         KASSERT((size > 0), ("%s: size <= 0", __func__));
  441 
  442         *alqp = NULL;
  443         td = curthread;
  444 
  445         NDINIT(&nd, LOOKUP, NOFOLLOW, UIO_SYSSPACE, file, td);
  446         oflags = FWRITE | O_NOFOLLOW | O_CREAT;
  447 
  448         error = vn_open_cred(&nd, &oflags, cmode, 0, cred, NULL);
  449         if (error)
  450                 return (error);
  451 
  452         NDFREE(&nd, NDF_ONLY_PNBUF);
  453         /* We just unlock so we hold a reference */
  454         VOP_UNLOCK(nd.ni_vp);
  455 
  456         alq = malloc(sizeof(*alq), M_ALD, M_WAITOK|M_ZERO);
  457         alq->aq_vp = nd.ni_vp;
  458         alq->aq_cred = crhold(cred);
  459 
  460         mtx_init(&alq->aq_mtx, "ALD Queue", NULL, MTX_SPIN|MTX_QUIET);
  461 
  462         alq->aq_buflen = size;
  463         alq->aq_entmax = 0;
  464         alq->aq_entlen = 0;
  465 
  466         alq->aq_freebytes = alq->aq_buflen;
  467         alq->aq_entbuf = malloc(alq->aq_buflen, M_ALD, M_WAITOK|M_ZERO);
  468         alq->aq_writehead = alq->aq_writetail = 0;
  469         if (flags & ALQ_ORDERED)
  470                 alq->aq_flags |= AQ_ORDERED;
  471 
  472         if ((error = ald_add(alq)) != 0) {
  473                 alq_destroy(alq);
  474                 return (error);
  475         }
  476 
  477         *alqp = alq;
  478 
  479         return (0);
  480 }
  481 
  482 int
  483 alq_open(struct alq **alqp, const char *file, struct ucred *cred, int cmode,
  484     int size, int count)
  485 {
  486         int ret;
  487 
  488         KASSERT((count >= 0), ("%s: count < 0", __func__));
  489 
  490         if (count > 0) {
  491                 if ((ret = alq_open_flags(alqp, file, cred, cmode,
  492                     size*count, 0)) == 0) {
  493                         (*alqp)->aq_flags |= AQ_LEGACY;
  494                         (*alqp)->aq_entmax = count;
  495                         (*alqp)->aq_entlen = size;
  496                 }
  497         } else
  498                 ret = alq_open_flags(alqp, file, cred, cmode, size, 0);
  499 
  500         return (ret);
  501 }
  502 
  503 /*
  504  * Copy a new entry into the queue.  If the operation would block either
  505  * wait or return an error depending on the value of waitok.
  506  */
  507 int
  508 alq_writen(struct alq *alq, void *data, int len, int flags)
  509 {
  510         int activate, copy, ret;
  511         void *waitchan;
  512 
  513         KASSERT((len > 0 && len <= alq->aq_buflen),
  514             ("%s: len <= 0 || len > aq_buflen", __func__));
  515 
  516         activate = ret = 0;
  517         copy = len;
  518         waitchan = NULL;
  519 
  520         ALQ_LOCK(alq);
  521 
  522         /*
  523          * Fail to perform the write and return EWOULDBLOCK if:
  524          * - The message is larger than our underlying buffer.
  525          * - The ALQ is being shutdown.
  526          * - There is insufficient free space in our underlying buffer
  527          *   to accept the message and the user can't wait for space.
  528          * - There is insufficient free space in our underlying buffer
  529          *   to accept the message and the alq is inactive due to prior
  530          *   use of the ALQ_NOACTIVATE flag (which would lead to deadlock).
  531          */
  532         if (len > alq->aq_buflen ||
  533             alq->aq_flags & AQ_SHUTDOWN ||
  534             (((flags & ALQ_NOWAIT) || (!(alq->aq_flags & AQ_ACTIVE) &&
  535             HAS_PENDING_DATA(alq))) && alq->aq_freebytes < len)) {
  536                 ALQ_UNLOCK(alq);
  537                 return (EWOULDBLOCK);
  538         }
  539 
  540         /*
  541          * If we want ordered writes and there is already at least one thread
  542          * waiting for resources to become available, sleep until we're woken.
  543          */
  544         if (alq->aq_flags & AQ_ORDERED && alq->aq_waiters > 0) {
  545                 KASSERT(!(flags & ALQ_NOWAIT),
  546                     ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__));
  547                 alq->aq_waiters++;
  548                 msleep_spin(&alq->aq_waiters, &alq->aq_mtx, "alqwnord", 0);
  549                 alq->aq_waiters--;
  550         }
  551 
  552         /*
  553          * (ALQ_WAITOK && aq_freebytes < len) or aq_freebytes >= len, either
  554          * enter while loop and sleep until we have enough free bytes (former)
  555          * or skip (latter). If AQ_ORDERED is set, only 1 thread at a time will
  556          * be in this loop. Otherwise, multiple threads may be sleeping here
  557          * competing for ALQ resources.
  558          */
  559         while (alq->aq_freebytes < len && !(alq->aq_flags & AQ_SHUTDOWN)) {
  560                 KASSERT(!(flags & ALQ_NOWAIT),
  561                     ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__));
  562                 alq->aq_flags |= AQ_WANTED;
  563                 alq->aq_waiters++;
  564                 if (waitchan)
  565                         wakeup(waitchan);
  566                 msleep_spin(alq, &alq->aq_mtx, "alqwnres", 0);
  567                 alq->aq_waiters--;
  568 
  569                 /*
  570                  * If we're the first thread to wake after an AQ_WANTED wakeup
  571                  * but there isn't enough free space for us, we're going to loop
  572                  * and sleep again. If there are other threads waiting in this
  573                  * loop, schedule a wakeup so that they can see if the space
  574                  * they require is available.
  575                  */
  576                 if (alq->aq_waiters > 0 && !(alq->aq_flags & AQ_ORDERED) &&
  577                     alq->aq_freebytes < len && !(alq->aq_flags & AQ_WANTED))
  578                         waitchan = alq;
  579                 else
  580                         waitchan = NULL;
  581         }
  582 
  583         /*
  584          * If there are waiters, we need to signal the waiting threads after we
  585          * complete our work. The alq ptr is used as a wait channel for threads
  586          * requiring resources to be freed up. In the AQ_ORDERED case, threads
  587          * are not allowed to concurrently compete for resources in the above
  588          * while loop, so we use a different wait channel in this case.
  589          */
  590         if (alq->aq_waiters > 0) {
  591                 if (alq->aq_flags & AQ_ORDERED)
  592                         waitchan = &alq->aq_waiters;
  593                 else
  594                         waitchan = alq;
  595         } else
  596                 waitchan = NULL;
  597 
  598         /* Bail if we're shutting down. */
  599         if (alq->aq_flags & AQ_SHUTDOWN) {
  600                 ret = EWOULDBLOCK;
  601                 goto unlock;
  602         }
  603 
  604         /*
  605          * If we need to wrap the buffer to accommodate the write,
  606          * we'll need 2 calls to bcopy.
  607          */
  608         if ((alq->aq_buflen - alq->aq_writehead) < len)
  609                 copy = alq->aq_buflen - alq->aq_writehead;
  610 
  611         /* Copy message (or part thereof if wrap required) to the buffer. */
  612         bcopy(data, alq->aq_entbuf + alq->aq_writehead, copy);
  613         alq->aq_writehead += copy;
  614 
  615         if (alq->aq_writehead >= alq->aq_buflen) {
  616                 KASSERT((alq->aq_writehead == alq->aq_buflen),
  617                     ("%s: alq->aq_writehead (%d) > alq->aq_buflen (%d)",
  618                     __func__,
  619                     alq->aq_writehead,
  620                     alq->aq_buflen));
  621                 alq->aq_writehead = 0;
  622         }
  623 
  624         if (copy != len) {
  625                 /*
  626                  * Wrap the buffer by copying the remainder of our message
  627                  * to the start of the buffer and resetting aq_writehead.
  628                  */
  629                 bcopy(((uint8_t *)data)+copy, alq->aq_entbuf, len - copy);
  630                 alq->aq_writehead = len - copy;
  631         }
  632 
  633         KASSERT((alq->aq_writehead >= 0 && alq->aq_writehead < alq->aq_buflen),
  634             ("%s: aq_writehead < 0 || aq_writehead >= aq_buflen", __func__));
  635 
  636         alq->aq_freebytes -= len;
  637 
  638         if (!(alq->aq_flags & AQ_ACTIVE) && !(flags & ALQ_NOACTIVATE)) {
  639                 alq->aq_flags |= AQ_ACTIVE;
  640                 activate = 1;
  641         }
  642 
  643         KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__));
  644 
  645 unlock:
  646         ALQ_UNLOCK(alq);
  647 
  648         if (activate) {
  649                 ALD_LOCK();
  650                 ald_activate(alq);
  651                 ALD_UNLOCK();
  652         }
  653 
  654         /* NB: We rely on wakeup_one waking threads in a FIFO manner. */
  655         if (waitchan != NULL)
  656                 wakeup_one(waitchan);
  657 
  658         return (ret);
  659 }
  660 
  661 int
  662 alq_write(struct alq *alq, void *data, int flags)
  663 {
  664         /* Should only be called in fixed length message (legacy) mode. */
  665         KASSERT((alq->aq_flags & AQ_LEGACY),
  666             ("%s: fixed length write on variable length queue", __func__));
  667         return (alq_writen(alq, data, alq->aq_entlen, flags));
  668 }
  669 
  670 /*
  671  * Retrieve a pointer for the ALQ to write directly into, avoiding bcopy.
  672  */
  673 struct ale *
  674 alq_getn(struct alq *alq, int len, int flags)
  675 {
  676         int contigbytes;
  677         void *waitchan;
  678 
  679         KASSERT((len > 0 && len <= alq->aq_buflen),
  680             ("%s: len <= 0 || len > alq->aq_buflen", __func__));
  681 
  682         waitchan = NULL;
  683 
  684         ALQ_LOCK(alq);
  685 
  686         /*
  687          * Determine the number of free contiguous bytes.
  688          * We ensure elsewhere that if aq_writehead == aq_writetail because
  689          * the buffer is empty, they will both be set to 0 and therefore
  690          * aq_freebytes == aq_buflen and is fully contiguous.
  691          * If they are equal and the buffer is not empty, aq_freebytes will
  692          * be 0 indicating the buffer is full.
  693          */
  694         if (alq->aq_writehead <= alq->aq_writetail)
  695                 contigbytes = alq->aq_freebytes;
  696         else {
  697                 contigbytes = alq->aq_buflen - alq->aq_writehead;
  698 
  699                 if (contigbytes < len) {
  700                         /*
  701                          * Insufficient space at end of buffer to handle a
  702                          * contiguous write. Wrap early if there's space at
  703                          * the beginning. This will leave a hole at the end
  704                          * of the buffer which we will have to skip over when
  705                          * flushing the buffer to disk.
  706                          */
  707                         if (alq->aq_writetail >= len || flags & ALQ_WAITOK) {
  708                                 /* Keep track of # bytes left blank. */
  709                                 alq->aq_wrapearly = contigbytes;
  710                                 /* Do the wrap and adjust counters. */
  711                                 contigbytes = alq->aq_freebytes =
  712                                     alq->aq_writetail;
  713                                 alq->aq_writehead = 0;
  714                         }
  715                 }
  716         }
  717 
  718         /*
  719          * Return a NULL ALE if:
  720          * - The message is larger than our underlying buffer.
  721          * - The ALQ is being shutdown.
  722          * - There is insufficient free space in our underlying buffer
  723          *   to accept the message and the user can't wait for space.
  724          * - There is insufficient free space in our underlying buffer
  725          *   to accept the message and the alq is inactive due to prior
  726          *   use of the ALQ_NOACTIVATE flag (which would lead to deadlock).
  727          */
  728         if (len > alq->aq_buflen ||
  729             alq->aq_flags & AQ_SHUTDOWN ||
  730             (((flags & ALQ_NOWAIT) || (!(alq->aq_flags & AQ_ACTIVE) &&
  731             HAS_PENDING_DATA(alq))) && contigbytes < len)) {
  732                 ALQ_UNLOCK(alq);
  733                 return (NULL);
  734         }
  735 
  736         /*
  737          * If we want ordered writes and there is already at least one thread
  738          * waiting for resources to become available, sleep until we're woken.
  739          */
  740         if (alq->aq_flags & AQ_ORDERED && alq->aq_waiters > 0) {
  741                 KASSERT(!(flags & ALQ_NOWAIT),
  742                     ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__));
  743                 alq->aq_waiters++;
  744                 msleep_spin(&alq->aq_waiters, &alq->aq_mtx, "alqgnord", 0);
  745                 alq->aq_waiters--;
  746         }
  747 
  748         /*
  749          * (ALQ_WAITOK && contigbytes < len) or contigbytes >= len, either enter
  750          * while loop and sleep until we have enough contiguous free bytes
  751          * (former) or skip (latter). If AQ_ORDERED is set, only 1 thread at a
  752          * time will be in this loop. Otherwise, multiple threads may be
  753          * sleeping here competing for ALQ resources.
  754          */
  755         while (contigbytes < len && !(alq->aq_flags & AQ_SHUTDOWN)) {
  756                 KASSERT(!(flags & ALQ_NOWAIT),
  757                     ("%s: ALQ_NOWAIT set but incorrectly ignored!", __func__));
  758                 alq->aq_flags |= AQ_WANTED;
  759                 alq->aq_waiters++;
  760                 if (waitchan)
  761                         wakeup(waitchan);
  762                 msleep_spin(alq, &alq->aq_mtx, "alqgnres", 0);
  763                 alq->aq_waiters--;
  764 
  765                 if (alq->aq_writehead <= alq->aq_writetail)
  766                         contigbytes = alq->aq_freebytes;
  767                 else
  768                         contigbytes = alq->aq_buflen - alq->aq_writehead;
  769 
  770                 /*
  771                  * If we're the first thread to wake after an AQ_WANTED wakeup
  772                  * but there isn't enough free space for us, we're going to loop
  773                  * and sleep again. If there are other threads waiting in this
  774                  * loop, schedule a wakeup so that they can see if the space
  775                  * they require is available.
  776                  */
  777                 if (alq->aq_waiters > 0 && !(alq->aq_flags & AQ_ORDERED) &&
  778                     contigbytes < len && !(alq->aq_flags & AQ_WANTED))
  779                         waitchan = alq;
  780                 else
  781                         waitchan = NULL;
  782         }
  783 
  784         /*
  785          * If there are waiters, we need to signal the waiting threads after we
  786          * complete our work. The alq ptr is used as a wait channel for threads
  787          * requiring resources to be freed up. In the AQ_ORDERED case, threads
  788          * are not allowed to concurrently compete for resources in the above
  789          * while loop, so we use a different wait channel in this case.
  790          */
  791         if (alq->aq_waiters > 0) {
  792                 if (alq->aq_flags & AQ_ORDERED)
  793                         waitchan = &alq->aq_waiters;
  794                 else
  795                         waitchan = alq;
  796         } else
  797                 waitchan = NULL;
  798 
  799         /* Bail if we're shutting down. */
  800         if (alq->aq_flags & AQ_SHUTDOWN) {
  801                 ALQ_UNLOCK(alq);
  802                 if (waitchan != NULL)
  803                         wakeup_one(waitchan);
  804                 return (NULL);
  805         }
  806 
  807         /*
  808          * If we are here, we have a contiguous number of bytes >= len
  809          * available in our buffer starting at aq_writehead.
  810          */
  811         alq->aq_getpost.ae_data = alq->aq_entbuf + alq->aq_writehead;
  812         alq->aq_getpost.ae_bytesused = len;
  813 
  814         return (&alq->aq_getpost);
  815 }
  816 
  817 struct ale *
  818 alq_get(struct alq *alq, int flags)
  819 {
  820         /* Should only be called in fixed length message (legacy) mode. */
  821         KASSERT((alq->aq_flags & AQ_LEGACY),
  822             ("%s: fixed length get on variable length queue", __func__));
  823         return (alq_getn(alq, alq->aq_entlen, flags));
  824 }
  825 
  826 void
  827 alq_post_flags(struct alq *alq, struct ale *ale, int flags)
  828 {
  829         int activate;
  830         void *waitchan;
  831 
  832         activate = 0;
  833 
  834         if (ale->ae_bytesused > 0) {
  835                 if (!(alq->aq_flags & AQ_ACTIVE) &&
  836                     !(flags & ALQ_NOACTIVATE)) {
  837                         alq->aq_flags |= AQ_ACTIVE;
  838                         activate = 1;
  839                 }
  840 
  841                 alq->aq_writehead += ale->ae_bytesused;
  842                 alq->aq_freebytes -= ale->ae_bytesused;
  843 
  844                 /* Wrap aq_writehead if we filled to the end of the buffer. */
  845                 if (alq->aq_writehead == alq->aq_buflen)
  846                         alq->aq_writehead = 0;
  847 
  848                 KASSERT((alq->aq_writehead >= 0 &&
  849                     alq->aq_writehead < alq->aq_buflen),
  850                     ("%s: aq_writehead < 0 || aq_writehead >= aq_buflen",
  851                     __func__));
  852 
  853                 KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__));
  854         }
  855 
  856         /*
  857          * If there are waiters, we need to signal the waiting threads after we
  858          * complete our work. The alq ptr is used as a wait channel for threads
  859          * requiring resources to be freed up. In the AQ_ORDERED case, threads
  860          * are not allowed to concurrently compete for resources in the
  861          * alq_getn() while loop, so we use a different wait channel in this case.
  862          */
  863         if (alq->aq_waiters > 0) {
  864                 if (alq->aq_flags & AQ_ORDERED)
  865                         waitchan = &alq->aq_waiters;
  866                 else
  867                         waitchan = alq;
  868         } else
  869                 waitchan = NULL;
  870 
  871         ALQ_UNLOCK(alq);
  872 
  873         if (activate) {
  874                 ALD_LOCK();
  875                 ald_activate(alq);
  876                 ALD_UNLOCK();
  877         }
  878 
  879         /* NB: We rely on wakeup_one waking threads in a FIFO manner. */
  880         if (waitchan != NULL)
  881                 wakeup_one(waitchan);
  882 }
  883 
  884 void
  885 alq_flush(struct alq *alq)
  886 {
  887         int needwakeup = 0;
  888 
  889         ALD_LOCK();
  890         ALQ_LOCK(alq);
  891 
  892         /*
  893          * Pull the lever iff there is data to flush and we're
  894          * not already in the middle of a flush operation.
  895          */
  896         if (HAS_PENDING_DATA(alq) && !(alq->aq_flags & AQ_FLUSHING)) {
  897                 if (alq->aq_flags & AQ_ACTIVE)
  898                         ald_deactivate(alq);
  899 
  900                 ALD_UNLOCK();
  901                 needwakeup = alq_doio(alq);
  902         } else
  903                 ALD_UNLOCK();
  904 
  905         ALQ_UNLOCK(alq);
  906 
  907         if (needwakeup)
  908                 wakeup_one(alq);
  909 }
  910 
  911 /*
  912  * Flush remaining data, close the file and free all resources.
  913  */
  914 void
  915 alq_close(struct alq *alq)
  916 {
  917         /* Only flush and destroy alq if not already shutting down. */
  918         if (ald_rem(alq) == 0)
  919                 alq_destroy(alq);
  920 }
  921 
  922 static int
  923 alq_load_handler(module_t mod, int what, void *arg)
  924 {
  925         int ret;
  926 
  927         ret = 0;
  928 
  929         switch (what) {
  930         case MOD_LOAD:
  931         case MOD_SHUTDOWN:
  932                 break;
  933 
  934         case MOD_QUIESCE:
  935                 ALD_LOCK();
  936                 /* Only allow unload if there are no open queues. */
  937                 if (LIST_FIRST(&ald_queues) == NULL) {
  938                         ald_shutingdown = 1;
  939                         ALD_UNLOCK();
  940                         EVENTHANDLER_DEREGISTER(shutdown_pre_sync,
  941                             alq_eventhandler_tag);
  942                         ald_shutdown(NULL, 0);
  943                         mtx_destroy(&ald_mtx);
  944                 } else {
  945                         ALD_UNLOCK();
  946                         ret = EBUSY;
  947                 }
  948                 break;
  949 
  950         case MOD_UNLOAD:
  951                 /* If MOD_QUIESCE failed we must fail here too. */
  952                 if (ald_shutingdown == 0)
  953                         ret = EBUSY;
  954                 break;
  955 
  956         default:
  957                 ret = EINVAL;
  958                 break;
  959         }
  960 
  961         return (ret);
  962 }
  963 
  964 static moduledata_t alq_mod =
  965 {
  966         "alq",
  967         alq_load_handler,
  968         NULL
  969 };
  970 
  971 DECLARE_MODULE(alq, alq_mod, SI_SUB_LAST, SI_ORDER_ANY);
  972 MODULE_VERSION(alq, 1);

Cache object: 1f303f76503ee31eb14156e88085ccb6


[ source navigation ] [ diff markup ] [ identifier search ] [ freetext search ] [ file search ] [ list types ] [ track identifier ]


This page is part of the FreeBSD/Linux Linux Kernel Cross-Reference, and was automatically generated using a modified version of the LXR engine.