sys/kern/kern_alq.c

   1 /*
   2  * Copyright (c) 2002, Jeffrey Roberson <jeff@freebsd.org>
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions
   7  * are met:
   8  * 1. Redistributions of source code must retain the above copyright
   9  *    notice unmodified, this list of conditions, and the following
  10  *    disclaimer.
  11  * 2. Redistributions in binary form must reproduce the above copyright
  12  *    notice, this list of conditions and the following disclaimer in the
  13  *    documentation and/or other materials provided with the distribution.
  14  *
  15  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
  16  * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
  17  * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
  18  * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
  19  * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
  20  * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  21  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  22  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  23  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
  24  * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  25  *
  26  * $FreeBSD$
  27  *
  28  */
  29
  30 #include <sys/param.h>
  31 #include <sys/systm.h>
  32 #include <sys/kernel.h>
  33 #include <sys/kthread.h>
  34 #include <sys/lock.h>
  35 #include <sys/mutex.h>
  36 #include <sys/namei.h>
  37 #include <sys/proc.h>
  38 #include <sys/vnode.h>
  39 #include <sys/alq.h>
  40 #include <sys/malloc.h>
  41 #include <sys/unistd.h>
  42 #include <sys/fcntl.h>
  43 #include <sys/eventhandler.h>
  44
  45 /* Async. Logging Queue */
  46 struct alq {
  47         int     aq_entmax;              /* Max entries */
  48         int     aq_entlen;              /* Entry length */
  49         char    *aq_entbuf;             /* Buffer for stored entries */
  50         int     aq_flags;               /* Queue flags */
  51         struct mtx      aq_mtx;         /* Queue lock */
  52         struct vnode    *aq_vp;         /* Open vnode handle */
  53         struct ucred    *aq_cred;       /* Credentials of the opening thread */
  54         struct ale      *aq_first;      /* First ent */
  55         struct ale      *aq_entfree;    /* First free ent */
  56         struct ale      *aq_entvalid;   /* First ent valid for writing */
  57         LIST_ENTRY(alq) aq_act;         /* List of active queues */
  58         LIST_ENTRY(alq) aq_link;        /* List of all queues */
  59 };
  60
  61 #define AQ_WANTED       0x0001          /* Wakeup sleeper when io is done */
  62 #define AQ_ACTIVE       0x0002          /* on the active list */
  63 #define AQ_FLUSHING     0x0004          /* doing IO */
  64 #define AQ_SHUTDOWN     0x0008          /* Queue no longer valid */
  65
  66 #define ALQ_LOCK(alq)   mtx_lock_spin(&(alq)->aq_mtx)
  67 #define ALQ_UNLOCK(alq) mtx_unlock_spin(&(alq)->aq_mtx)
  68
  69 static MALLOC_DEFINE(M_ALD, "ALD", "ALD");
  70
  71 /*
  72  * The ald_mtx protects the ald_queues list and the ald_active list.
  73  */
  74 static struct mtx ald_mtx;
  75 static LIST_HEAD(, alq) ald_queues;
  76 static LIST_HEAD(, alq) ald_active;
  77 static int ald_shutingdown = 0;
  78 struct thread *ald_thread;
  79 static struct proc *ald_proc;
  80
  81 #define ALD_LOCK()      mtx_lock(&ald_mtx)
  82 #define ALD_UNLOCK()    mtx_unlock(&ald_mtx)
  83
  84 /* Daemon functions */
  85 static int ald_add(struct alq *);
  86 static int ald_rem(struct alq *);
  87 static void ald_startup(void *);
  88 static void ald_daemon(void);
  89 static void ald_shutdown(void *, int);
  90 static void ald_activate(struct alq *);
  91 static void ald_deactivate(struct alq *);
  92
  93 /* Internal queue functions */
  94 static void alq_shutdown(struct alq *);
  95 static int alq_doio(struct alq *);
  96
  97
  98 /*
  99  * Add a new queue to the global list.  Fail if we're shutting down.
 100  */
 101 static int
 102 ald_add(struct alq *alq)
 103 {
 104         int error;
 105
 106         error = 0;
 107
 108         ALD_LOCK();
 109         if (ald_shutingdown) {
 110                 error = EBUSY;
 111                 goto done;
 112         }
 113         LIST_INSERT_HEAD(&ald_queues, alq, aq_link);
 114 done:
 115         ALD_UNLOCK();
 116         return (error);
 117 }
 118
 119 /*
 120  * Remove a queue from the global list unless we're shutting down.  If so,
 121  * the ald will take care of cleaning up it's resources.
 122  */
 123 static int
 124 ald_rem(struct alq *alq)
 125 {
 126         int error;
 127
 128         error = 0;
 129
 130         ALD_LOCK();
 131         if (ald_shutingdown) {
 132                 error = EBUSY;
 133                 goto done;
 134         }
 135         LIST_REMOVE(alq, aq_link);
 136 done:
 137         ALD_UNLOCK();
 138         return (error);
 139 }
 140
 141 /*
 142  * Put a queue on the active list.  This will schedule it for writing.
 143  */
 144 static void
 145 ald_activate(struct alq *alq)
 146 {
 147         LIST_INSERT_HEAD(&ald_active, alq, aq_act);
 148         wakeup(&ald_active);
 149 }
 150
 151 static void
 152 ald_deactivate(struct alq *alq)
 153 {
 154         LIST_REMOVE(alq, aq_act);
 155         alq->aq_flags &= ~AQ_ACTIVE;
 156 }
 157
 158 static void
 159 ald_startup(void *unused)
 160 {
 161         mtx_init(&ald_mtx, "ALDmtx", NULL, MTX_DEF|MTX_QUIET);
 162         LIST_INIT(&ald_queues);
 163         LIST_INIT(&ald_active);
 164 }
 165
 166 static void
 167 ald_daemon(void)
 168 {
 169         int needwakeup;
 170         struct alq *alq;
 171
 172         mtx_lock(&Giant);
 173
 174         ald_thread = FIRST_THREAD_IN_PROC(ald_proc);
 175
 176         EVENTHANDLER_REGISTER(shutdown_pre_sync, ald_shutdown, NULL,
 177             SHUTDOWN_PRI_FIRST);
 178
 179         ALD_LOCK();
 180
 181         for (;;) {
 182                 while ((alq = LIST_FIRST(&ald_active)) == NULL)
 183                         msleep(&ald_active, &ald_mtx, PWAIT, "aldslp", 0);
 184
 185                 ALQ_LOCK(alq);
 186                 ald_deactivate(alq);
 187                 ALD_UNLOCK();
 188                 needwakeup = alq_doio(alq);
 189                 ALQ_UNLOCK(alq);
 190                 if (needwakeup)
 191                         wakeup(alq);
 192                 ALD_LOCK();
 193         }
 194 }
 195
 196 static void
 197 ald_shutdown(void *arg, int howto)
 198 {
 199         struct alq *alq;
 200
 201         ALD_LOCK();
 202         ald_shutingdown = 1;
 203
 204         while ((alq = LIST_FIRST(&ald_queues)) != NULL) {
 205                 LIST_REMOVE(alq, aq_link);
 206                 ALD_UNLOCK();
 207                 alq_shutdown(alq);
 208                 ALD_LOCK();
 209         }
 210         ALD_UNLOCK();
 211 }
 212
 213 static void
 214 alq_shutdown(struct alq *alq)
 215 {
 216         ALQ_LOCK(alq);
 217
 218         /* Stop any new writers. */
 219         alq->aq_flags |= AQ_SHUTDOWN;
 220
 221         /* Drain IO */
 222         while (alq->aq_flags & (AQ_FLUSHING|AQ_ACTIVE)) {
 223                 alq->aq_flags |= AQ_WANTED;
 224                 ALQ_UNLOCK(alq);
 225                 tsleep(alq, PWAIT, "aldclose", 0);
 226                 ALQ_LOCK(alq);
 227         }
 228         ALQ_UNLOCK(alq);
 229
 230         vn_close(alq->aq_vp, FWRITE, alq->aq_cred,
 231             curthread);
 232         crfree(alq->aq_cred);
 233 }
 234
 235 /*
 236  * Flush all pending data to disk.  This operation will block.
 237  */
 238 static int
 239 alq_doio(struct alq *alq)
 240 {
 241         struct thread *td;
 242         struct mount *mp;
 243         struct vnode *vp;
 244         struct uio auio;
 245         struct iovec aiov[2];
 246         struct ale *ale;
 247         struct ale *alstart;
 248         int totlen;
 249         int iov;
 250
 251         vp = alq->aq_vp;
 252         td = curthread;
 253         totlen = 0;
 254         iov = 0;
 255
 256         alstart = ale = alq->aq_entvalid;
 257         alq->aq_entvalid = NULL;
 258
 259         bzero(&aiov, sizeof(aiov));
 260         bzero(&auio, sizeof(auio));
 261
 262         do {
 263                 if (aiov[iov].iov_base == NULL)
 264                         aiov[iov].iov_base = ale->ae_data;
 265                 aiov[iov].iov_len += alq->aq_entlen;
 266                 totlen += alq->aq_entlen;
 267                 /* Check to see if we're wrapping the buffer */
 268                 if (ale->ae_data + alq->aq_entlen != ale->ae_next->ae_data)
 269                         iov++;
 270                 ale->ae_flags &= ~AE_VALID;
 271                 ale = ale->ae_next;
 272         } while (ale->ae_flags & AE_VALID);
 273
 274         alq->aq_flags |= AQ_FLUSHING;
 275         ALQ_UNLOCK(alq);
 276
 277         if (iov == 2 || aiov[iov].iov_base == NULL)
 278                 iov--;
 279
 280         auio.uio_iov = &aiov[0];
 281         auio.uio_offset = 0;
 282         auio.uio_segflg = UIO_SYSSPACE;
 283         auio.uio_rw = UIO_WRITE;
 284         auio.uio_iovcnt = iov + 1;
 285         auio.uio_resid = totlen;
 286         auio.uio_td = td;
 287
 288         /*
 289          * Do all of the junk required to write now.
 290          */
 291         vn_start_write(vp, &mp, V_WAIT);
 292         vn_lock(vp, LK_EXCLUSIVE | LK_RETRY, td);
 293         VOP_LEASE(vp, td, alq->aq_cred, LEASE_WRITE);
 294         /* XXX error ignored */
 295         VOP_WRITE(vp, &auio, IO_UNIT | IO_APPEND, alq->aq_cred);
 296         VOP_UNLOCK(vp, 0, td);
 297         vn_finished_write(mp);
 298
 299         ALQ_LOCK(alq);
 300         alq->aq_flags &= ~AQ_FLUSHING;
 301
 302         if (alq->aq_entfree == NULL)
 303                 alq->aq_entfree = alstart;
 304
 305         if (alq->aq_flags & AQ_WANTED) {
 306                 alq->aq_flags &= ~AQ_WANTED;
 307                 return (1);
 308         }
 309
 310         return(0);
 311 }
 312
 313 static struct kproc_desc ald_kp = {
 314         "ALQ Daemon",
 315         ald_daemon,
 316         &ald_proc
 317 };
 318
 319 SYSINIT(aldthread, SI_SUB_KTHREAD_IDLE, SI_ORDER_ANY, kproc_start, &ald_kp)
 320 SYSINIT(ald, SI_SUB_LOCK, SI_ORDER_ANY, ald_startup, NULL)
 321
 322
 323 /* User visible queue functions */
 324
 325 /*
 326  * Create the queue data structure, allocate the buffer, and open the file.
 327  */
 328 int
 329 alq_open(struct alq **alqp, const char *file, int size, int count)
 330 {
 331         struct thread *td;
 332         struct nameidata nd;
 333         struct ale *ale;
 334         struct ale *alp;
 335         struct alq *alq;
 336         char *bufp;
 337         int flags;
 338         int error;
 339         int i;
 340
 341         *alqp = NULL;
 342         td = curthread;
 343
 344         NDINIT(&nd, LOOKUP, NOFOLLOW, UIO_SYSSPACE, file, td);
 345         flags = FWRITE | O_NOFOLLOW | O_CREAT;
 346
 347         error = vn_open(&nd, &flags, 0);
 348         if (error)
 349                 return (error);
 350
 351         NDFREE(&nd, NDF_ONLY_PNBUF);
 352         /* We just unlock so we hold a reference */
 353         VOP_UNLOCK(nd.ni_vp, 0, td);
 354
 355         alq = malloc(sizeof(*alq), M_ALD, M_WAITOK|M_ZERO);
 356         alq->aq_entbuf = malloc(count * size, M_ALD, M_WAITOK|M_ZERO);
 357         alq->aq_first = malloc(sizeof(*ale) * count, M_ALD, M_WAITOK|M_ZERO);
 358         alq->aq_vp = nd.ni_vp;
 359         alq->aq_cred = crhold(td->td_ucred);
 360         alq->aq_entmax = count;
 361         alq->aq_entlen = size;
 362         alq->aq_entfree = alq->aq_first;
 363
 364         mtx_init(&alq->aq_mtx, "ALD Queue", NULL, MTX_SPIN|MTX_QUIET);
 365
 366         bufp = alq->aq_entbuf;
 367         ale = alq->aq_first;
 368         alp = NULL;
 369
 370         /* Match up entries with buffers */
 371         for (i = 0; i < count; i++) {
 372                 if (alp)
 373                         alp->ae_next = ale;
 374                 ale->ae_data = bufp;
 375                 alp = ale;
 376                 ale++;
 377                 bufp += size;
 378         }
 379
 380         alp->ae_next = alq->aq_first;
 381
 382         if ((error = ald_add(alq)) != 0)
 383                 return (error);
 384         *alqp = alq;
 385
 386         return (0);
 387 }
 388
 389 /*
 390  * Copy a new entry into the queue.  If the operation would block either
 391  * wait or return an error depending on the value of waitok.
 392  */
 393 int
 394 alq_write(struct alq *alq, void *data, int waitok)
 395 {
 396         struct ale *ale;
 397
 398         if ((ale = alq_get(alq, waitok)) == NULL)
 399                 return (EWOULDBLOCK);
 400
 401         bcopy(data, ale->ae_data, alq->aq_entlen);
 402         alq_post(alq, ale);
 403
 404         return (0);
 405 }
 406
 407 struct ale *
 408 alq_get(struct alq *alq, int waitok)
 409 {
 410         struct ale *ale;
 411         struct ale *aln;
 412
 413         ale = NULL;
 414
 415         ALQ_LOCK(alq);
 416
 417         /* Loop until we get an entry or we're shutting down */
 418         while ((alq->aq_flags & AQ_SHUTDOWN) == 0 &&
 419             (ale = alq->aq_entfree) == NULL &&
 420             (waitok & ALQ_WAITOK)) {
 421                 alq->aq_flags |= AQ_WANTED;
 422                 ALQ_UNLOCK(alq);
 423                 tsleep(alq, PWAIT, "alqget", 0);
 424                 ALQ_LOCK(alq);
 425         }
 426
 427         if (ale != NULL) {
 428                 aln = ale->ae_next;
 429                 if ((aln->ae_flags & AE_VALID) == 0)
 430                         alq->aq_entfree = aln;
 431         } else
 432                 ALQ_UNLOCK(alq);
 433
 434
 435         return (ale);
 436 }
 437
 438 void
 439 alq_post(struct alq *alq, struct ale *ale)
 440 {
 441         int activate;
 442
 443         ale->ae_flags |= AE_VALID;
 444
 445         if (alq->aq_entvalid == NULL)
 446                 alq->aq_entvalid = ale;
 447
 448         if ((alq->aq_flags & AQ_ACTIVE) == 0) {
 449                 alq->aq_flags |= AQ_ACTIVE;
 450                 activate = 1;
 451         } else
 452                 activate = 0;
 453
 454         ALQ_UNLOCK(alq);
 455         if (activate) {
 456                 ALD_LOCK();
 457                 ald_activate(alq);
 458                 ALD_UNLOCK();
 459         }
 460 }
 461
 462 void
 463 alq_flush(struct alq *alq)
 464 {
 465         int needwakeup = 0;
 466
 467         ALD_LOCK();
 468         ALQ_LOCK(alq);
 469         if (alq->aq_flags & AQ_ACTIVE) {
 470                 ald_deactivate(alq);
 471                 ALD_UNLOCK();
 472                 needwakeup = alq_doio(alq);
 473         } else
 474                 ALD_UNLOCK();
 475         ALQ_UNLOCK(alq);
 476
 477         if (needwakeup)
 478                 wakeup(alq);
 479 }
 480
 481 /*
 482  * Flush remaining data, close the file and free all resources.
 483  */
 484 void
 485 alq_close(struct alq *alq)
 486 {
 487         /*
 488          * If we're already shuting down someone else will flush and close
 489          * the vnode.
 490          */
 491         if (ald_rem(alq) != 0)
 492                 return;
 493
 494         /*
 495          * Drain all pending IO.
 496          */
 497         alq_shutdown(alq);
 498
 499         mtx_destroy(&alq->aq_mtx);
 500         free(alq->aq_first, M_ALD);
 501         free(alq->aq_entbuf, M_ALD);
 502         free(alq, M_ALD);
 503 }