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 nameidata nd;
  435         struct alq *alq;
  436         int oflags;
  437         int error;
  438 
  439         KASSERT((size > 0), ("%s: size <= 0", __func__));
  440 
  441         *alqp = NULL;
  442 
  443         NDINIT(&nd, LOOKUP, NOFOLLOW, UIO_SYSSPACE, file);
  444         oflags = FWRITE | O_NOFOLLOW | O_CREAT;
  445 
  446         error = vn_open_cred(&nd, &oflags, cmode, 0, cred, NULL);
  447         if (error)
  448                 return (error);
  449 
  450         NDFREE_PNBUF(&nd);
  451         /* We just unlock so we hold a reference */
  452         VOP_UNLOCK(nd.ni_vp);
  453 
  454         alq = malloc(sizeof(*alq), M_ALD, M_WAITOK|M_ZERO);
  455         alq->aq_vp = nd.ni_vp;
  456         alq->aq_cred = crhold(cred);
  457 
  458         mtx_init(&alq->aq_mtx, "ALD Queue", NULL, MTX_SPIN|MTX_QUIET);
  459 
  460         alq->aq_buflen = size;
  461         alq->aq_entmax = 0;
  462         alq->aq_entlen = 0;
  463 
  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;
  469 
  470         if ((error = ald_add(alq)) != 0) {
  471                 alq_destroy(alq);
  472                 return (error);
  473         }
  474 
  475         *alqp = alq;
  476 
  477         return (0);
  478 }
  479 
  480 int
  481 alq_open(struct alq **alqp, const char *file, struct ucred *cred, int cmode,
  482     int size, int count)
  483 {
  484         int ret;
  485 
  486         KASSERT((count >= 0), ("%s: count < 0", __func__));
  487 
  488         if (count > 0) {
  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;
  494                 }
  495         } else
  496                 ret = alq_open_flags(alqp, file, cred, cmode, size, 0);
  497 
  498         return (ret);
  499 }
  500 
  501 /*
  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.
  504  */
  505 int
  506 alq_writen(struct alq *alq, void *data, int len, int flags)
  507 {
  508         int activate, copy, ret;
  509         void *waitchan;
  510 
  511         KASSERT((len > 0 && len <= alq->aq_buflen),
  512             ("%s: len <= 0 || len > aq_buflen", __func__));
  513 
  514         activate = ret = 0;
  515         copy = len;
  516         waitchan = NULL;
  517 
  518         ALQ_LOCK(alq);
  519 
  520         /*
  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).
  529          */
  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)) {
  534                 ALQ_UNLOCK(alq);
  535                 return (EWOULDBLOCK);
  536         }
  537 
  538         /*
  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.
  541          */
  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__));
  545                 alq->aq_waiters++;
  546                 msleep_spin(&alq->aq_waiters, &alq->aq_mtx, "alqwnord", 0);
  547                 alq->aq_waiters--;
  548         }
  549 
  550         /*
  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.
  556          */
  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;
  561                 alq->aq_waiters++;
  562                 if (waitchan)
  563                         wakeup(waitchan);
  564                 msleep_spin(alq, &alq->aq_mtx, "alqwnres", 0);
  565                 alq->aq_waiters--;
  566 
  567                 /*
  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.
  573                  */
  574                 if (alq->aq_waiters > 0 && !(alq->aq_flags & AQ_ORDERED) &&
  575                     alq->aq_freebytes < len && !(alq->aq_flags & AQ_WANTED))
  576                         waitchan = alq;
  577                 else
  578                         waitchan = NULL;
  579         }
  580 
  581         /*
  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.
  587          */
  588         if (alq->aq_waiters > 0) {
  589                 if (alq->aq_flags & AQ_ORDERED)
  590                         waitchan = &alq->aq_waiters;
  591                 else
  592                         waitchan = alq;
  593         } else
  594                 waitchan = NULL;
  595 
  596         /* Bail if we're shutting down. */
  597         if (alq->aq_flags & AQ_SHUTDOWN) {
  598                 ret = EWOULDBLOCK;
  599                 goto unlock;
  600         }
  601 
  602         /*
  603          * If we need to wrap the buffer to accommodate the write,
  604          * we'll need 2 calls to bcopy.
  605          */
  606         if ((alq->aq_buflen - alq->aq_writehead) < len)
  607                 copy = alq->aq_buflen - alq->aq_writehead;
  608 
  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;
  612 
  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)",
  616                     __func__,
  617                     alq->aq_writehead,
  618                     alq->aq_buflen));
  619                 alq->aq_writehead = 0;
  620         }
  621 
  622         if (copy != len) {
  623                 /*
  624                  * Wrap the buffer by copying the remainder of our message
  625                  * to the start of the buffer and resetting aq_writehead.
  626                  */
  627                 bcopy(((uint8_t *)data)+copy, alq->aq_entbuf, len - copy);
  628                 alq->aq_writehead = len - copy;
  629         }
  630 
  631         KASSERT((alq->aq_writehead >= 0 && alq->aq_writehead < alq->aq_buflen),
  632             ("%s: aq_writehead < 0 || aq_writehead >= aq_buflen", __func__));
  633 
  634         alq->aq_freebytes -= len;
  635 
  636         if (!(alq->aq_flags & AQ_ACTIVE) && !(flags & ALQ_NOACTIVATE)) {
  637                 alq->aq_flags |= AQ_ACTIVE;
  638                 activate = 1;
  639         }
  640 
  641         KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__));
  642 
  643 unlock:
  644         ALQ_UNLOCK(alq);
  645 
  646         if (activate) {
  647                 ALD_LOCK();
  648                 ald_activate(alq);
  649                 ALD_UNLOCK();
  650         }
  651 
  652         /* NB: We rely on wakeup_one waking threads in a FIFO manner. */
  653         if (waitchan != NULL)
  654                 wakeup_one(waitchan);
  655 
  656         return (ret);
  657 }
  658 
  659 int
  660 alq_write(struct alq *alq, void *data, int flags)
  661 {
  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));
  666 }
  667 
  668 /*
  669  * Retrieve a pointer for the ALQ to write directly into, avoiding bcopy.
  670  */
  671 struct ale *
  672 alq_getn(struct alq *alq, int len, int flags)
  673 {
  674         int contigbytes;
  675         void *waitchan;
  676 
  677         KASSERT((len > 0 && len <= alq->aq_buflen),
  678             ("%s: len <= 0 || len > alq->aq_buflen", __func__));
  679 
  680         waitchan = NULL;
  681 
  682         ALQ_LOCK(alq);
  683 
  684         /*
  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.
  691          */
  692         if (alq->aq_writehead <= alq->aq_writetail)
  693                 contigbytes = alq->aq_freebytes;
  694         else {
  695                 contigbytes = alq->aq_buflen - alq->aq_writehead;
  696 
  697                 if (contigbytes < len) {
  698                         /*
  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.
  704                          */
  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 =
  710                                     alq->aq_writetail;
  711                                 alq->aq_writehead = 0;
  712                         }
  713                 }
  714         }
  715 
  716         /*
  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).
  725          */
  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)) {
  730                 ALQ_UNLOCK(alq);
  731                 return (NULL);
  732         }
  733 
  734         /*
  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.
  737          */
  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__));
  741                 alq->aq_waiters++;
  742                 msleep_spin(&alq->aq_waiters, &alq->aq_mtx, "alqgnord", 0);
  743                 alq->aq_waiters--;
  744         }
  745 
  746         /*
  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.
  752          */
  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;
  757                 alq->aq_waiters++;
  758                 if (waitchan)
  759                         wakeup(waitchan);
  760                 msleep_spin(alq, &alq->aq_mtx, "alqgnres", 0);
  761                 alq->aq_waiters--;
  762 
  763                 if (alq->aq_writehead <= alq->aq_writetail)
  764                         contigbytes = alq->aq_freebytes;
  765                 else
  766                         contigbytes = alq->aq_buflen - alq->aq_writehead;
  767 
  768                 /*
  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.
  774                  */
  775                 if (alq->aq_waiters > 0 && !(alq->aq_flags & AQ_ORDERED) &&
  776                     contigbytes < len && !(alq->aq_flags & AQ_WANTED))
  777                         waitchan = alq;
  778                 else
  779                         waitchan = NULL;
  780         }
  781 
  782         /*
  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.
  788          */
  789         if (alq->aq_waiters > 0) {
  790                 if (alq->aq_flags & AQ_ORDERED)
  791                         waitchan = &alq->aq_waiters;
  792                 else
  793                         waitchan = alq;
  794         } else
  795                 waitchan = NULL;
  796 
  797         /* Bail if we're shutting down. */
  798         if (alq->aq_flags & AQ_SHUTDOWN) {
  799                 ALQ_UNLOCK(alq);
  800                 if (waitchan != NULL)
  801                         wakeup_one(waitchan);
  802                 return (NULL);
  803         }
  804 
  805         /*
  806          * If we are here, we have a contiguous number of bytes >= len
  807          * available in our buffer starting at aq_writehead.
  808          */
  809         alq->aq_getpost.ae_data = alq->aq_entbuf + alq->aq_writehead;
  810         alq->aq_getpost.ae_bytesused = len;
  811 
  812         return (&alq->aq_getpost);
  813 }
  814 
  815 struct ale *
  816 alq_get(struct alq *alq, int flags)
  817 {
  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));
  822 }
  823 
  824 void
  825 alq_post_flags(struct alq *alq, struct ale *ale, int flags)
  826 {
  827         int activate;
  828         void *waitchan;
  829 
  830         activate = 0;
  831 
  832         if (ale->ae_bytesused > 0) {
  833                 if (!(alq->aq_flags & AQ_ACTIVE) &&
  834                     !(flags & ALQ_NOACTIVATE)) {
  835                         alq->aq_flags |= AQ_ACTIVE;
  836                         activate = 1;
  837                 }
  838 
  839                 alq->aq_writehead += ale->ae_bytesused;
  840                 alq->aq_freebytes -= ale->ae_bytesused;
  841 
  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;
  845 
  846                 KASSERT((alq->aq_writehead >= 0 &&
  847                     alq->aq_writehead < alq->aq_buflen),
  848                     ("%s: aq_writehead < 0 || aq_writehead >= aq_buflen",
  849                     __func__));
  850 
  851                 KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__));
  852         }
  853 
  854         /*
  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.
  860          */
  861         if (alq->aq_waiters > 0) {
  862                 if (alq->aq_flags & AQ_ORDERED)
  863                         waitchan = &alq->aq_waiters;
  864                 else
  865                         waitchan = alq;
  866         } else
  867                 waitchan = NULL;
  868 
  869         ALQ_UNLOCK(alq);
  870 
  871         if (activate) {
  872                 ALD_LOCK();
  873                 ald_activate(alq);
  874                 ALD_UNLOCK();
  875         }
  876 
  877         /* NB: We rely on wakeup_one waking threads in a FIFO manner. */
  878         if (waitchan != NULL)
  879                 wakeup_one(waitchan);
  880 }
  881 
  882 void
  883 alq_flush(struct alq *alq)
  884 {
  885         int needwakeup = 0;
  886 
  887         ALD_LOCK();
  888         ALQ_LOCK(alq);
  889 
  890         /*
  891          * Pull the lever iff there is data to flush and we're
  892          * not already in the middle of a flush operation.
  893          */
  894         if (HAS_PENDING_DATA(alq) && !(alq->aq_flags & AQ_FLUSHING)) {
  895                 if (alq->aq_flags & AQ_ACTIVE)
  896                         ald_deactivate(alq);
  897 
  898                 ALD_UNLOCK();
  899                 needwakeup = alq_doio(alq);
  900         } else
  901                 ALD_UNLOCK();
  902 
  903         ALQ_UNLOCK(alq);
  904 
  905         if (needwakeup)
  906                 wakeup_one(alq);
  907 }
  908 
  909 /*
  910  * Flush remaining data, close the file and free all resources.
  911  */
  912 void
  913 alq_close(struct alq *alq)
  914 {
  915         /* Only flush and destroy alq if not already shutting down. */
  916         if (ald_rem(alq) == 0)
  917                 alq_destroy(alq);
  918 }
  919 
  920 static int
  921 alq_load_handler(module_t mod, int what, void *arg)
  922 {
  923         int ret;
  924 
  925         ret = 0;
  926 
  927         switch (what) {
  928         case MOD_LOAD:
  929         case MOD_SHUTDOWN:
  930                 break;
  931 
  932         case MOD_QUIESCE:
  933                 ALD_LOCK();
  934                 /* Only allow unload if there are no open queues. */
  935                 if (LIST_FIRST(&ald_queues) == NULL) {
  936                         ald_shutingdown = 1;
  937                         ALD_UNLOCK();
  938                         EVENTHANDLER_DEREGISTER(shutdown_pre_sync,
  939                             alq_eventhandler_tag);
  940                         ald_shutdown(NULL, 0);
  941                         mtx_destroy(&ald_mtx);
  942                 } else {
  943                         ALD_UNLOCK();
  944                         ret = EBUSY;
  945                 }
  946                 break;
  947 
  948         case MOD_UNLOAD:
  949                 /* If MOD_QUIESCE failed we must fail here too. */
  950                 if (ald_shutingdown == 0)
  951                         ret = EBUSY;
  952                 break;
  953 
  954         default:
  955                 ret = EINVAL;
  956                 break;
  957         }
  958 
  959         return (ret);
  960 }
  961 
  962 static moduledata_t alq_mod =
  963 {
  964         "alq",
  965         alq_load_handler,
  966         NULL
  967 };
  968 
  969 DECLARE_MODULE(alq, alq_mod, SI_SUB_LAST, SI_ORDER_ANY);
  970 MODULE_VERSION(alq, 1);
Cache object: e02d4f7151bea4b40212bea2fd57993b
[ 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.
FreeBSD/Linux Kernel Cross Reference sys/kern/kern_alq.c

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