1/*- 2 * SPDX-License-Identifier: BSD-2-Clause 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#include "opt_mac.h" 37 38#include <sys/param.h> 39#include <sys/systm.h> 40#include <sys/alq.h> 41#include <sys/eventhandler.h> 42#include <sys/fcntl.h> 43#include <sys/kernel.h> 44#include <sys/kthread.h> 45#include <sys/lock.h> 46#include <sys/malloc.h> 47#include <sys/mount.h> 48#include <sys/mutex.h> 49#include <sys/namei.h> 50#include <sys/proc.h> 51#include <sys/reboot.h> 52#include <sys/unistd.h> 53#include <sys/vnode.h> 54 55#include <security/mac/mac_framework.h> 56 57/* Async. Logging Queue */ 58struct alq { 59 char *aq_entbuf; /* Buffer for stored entries */ 60 int aq_entmax; /* Max entries */ 61 int aq_entlen; /* Entry length */ 62 int aq_freebytes; /* Bytes available in buffer */ 63 int aq_buflen; /* Total length of our buffer */ 64 int aq_writehead; /* Location for next write */ 65 int aq_writetail; /* Flush starts at this location */ 66 int aq_wrapearly; /* # bytes left blank at end of buf */ 67 int aq_flags; /* Queue flags */ 68 int aq_waiters; /* Num threads waiting for resources 69 * NB: Used as a wait channel so must 70 * not be first field in the alq struct 71 */ 72 struct ale aq_getpost; /* ALE for use by get/post */ 73 struct mtx aq_mtx; /* Queue lock */ 74 struct vnode *aq_vp; /* Open vnode handle */ 75 struct ucred *aq_cred; /* Credentials of the opening thread */ 76 LIST_ENTRY(alq) aq_act; /* List of active queues */ 77 LIST_ENTRY(alq) aq_link; /* List of all queues */ 78}; 79 80#define AQ_WANTED 0x0001 /* Wakeup sleeper when io is done */ 81#define AQ_ACTIVE 0x0002 /* on the active list */ 82#define AQ_FLUSHING 0x0004 /* doing IO */ 83#define AQ_SHUTDOWN 0x0008 /* Queue no longer valid */ 84#define AQ_ORDERED 0x0010 /* Queue enforces ordered writes */ 85#define AQ_LEGACY 0x0020 /* Legacy queue (fixed length writes) */ 86 87#define ALQ_LOCK(alq) mtx_lock_spin(&(alq)->aq_mtx) 88#define ALQ_UNLOCK(alq) mtx_unlock_spin(&(alq)->aq_mtx) 89 90#define HAS_PENDING_DATA(alq) ((alq)->aq_freebytes != (alq)->aq_buflen) 91 92static MALLOC_DEFINE(M_ALD, "ALD", "ALD"); 93 94/* 95 * The ald_mtx protects the ald_queues list and the ald_active list. 96 */ 97static struct mtx ald_mtx; 98static LIST_HEAD(, alq) ald_queues; 99static LIST_HEAD(, alq) ald_active; 100static int ald_shutingdown = 0; 101struct thread *ald_thread; 102static struct proc *ald_proc; 103static eventhandler_tag alq_eventhandler_tag = NULL; 104 105#define ALD_LOCK() mtx_lock(&ald_mtx) 106#define ALD_UNLOCK() mtx_unlock(&ald_mtx) 107 108/* Daemon functions */ 109static int ald_add(struct alq *); 110static int ald_rem(struct alq *); 111static void ald_startup(void *); 112static void ald_daemon(void); 113static void ald_shutdown(void *, int); 114static void ald_activate(struct alq *); 115static void ald_deactivate(struct alq *); 116 117/* Internal queue functions */ 118static void alq_shutdown(struct alq *); 119static void alq_destroy(struct alq *); 120static int alq_doio(struct alq *); 121 122/* 123 * Add a new queue to the global list. Fail if we're shutting down. 124 */ 125static int 126ald_add(struct alq *alq) 127{ 128 int error; 129 130 error = 0; 131 132 ALD_LOCK(); 133 if (ald_shutingdown) { 134 error = EBUSY; 135 goto done; 136 } 137 LIST_INSERT_HEAD(&ald_queues, alq, aq_link); 138done: 139 ALD_UNLOCK(); 140 return (error); 141} 142 143/* 144 * Remove a queue from the global list unless we're shutting down. If so, 145 * the ald will take care of cleaning up it's resources. 146 */ 147static int 148ald_rem(struct alq *alq) 149{ 150 int error; 151 152 error = 0; 153 154 ALD_LOCK(); 155 if (ald_shutingdown) { 156 error = EBUSY; 157 goto done; 158 } 159 LIST_REMOVE(alq, aq_link); 160done: 161 ALD_UNLOCK(); 162 return (error); 163} 164 165/* 166 * Put a queue on the active list. This will schedule it for writing. 167 */ 168static void 169ald_activate(struct alq *alq) 170{ 171 LIST_INSERT_HEAD(&ald_active, alq, aq_act); 172 wakeup(&ald_active); 173} 174 175static void 176ald_deactivate(struct alq *alq) 177{ 178 LIST_REMOVE(alq, aq_act); 179 alq->aq_flags &= ~AQ_ACTIVE; 180} 181 182static void 183ald_startup(void *unused) 184{ 185 mtx_init(&ald_mtx, "ALDmtx", NULL, MTX_DEF|MTX_QUIET); 186 LIST_INIT(&ald_queues); 187 LIST_INIT(&ald_active); 188} 189 190static void 191ald_daemon(void) 192{ 193 int needwakeup; 194 struct alq *alq; 195 196 ald_thread = FIRST_THREAD_IN_PROC(ald_proc); 197 198 alq_eventhandler_tag = EVENTHANDLER_REGISTER(shutdown_pre_sync, 199 ald_shutdown, NULL, SHUTDOWN_PRI_FIRST); 200 201 ALD_LOCK(); 202 203 for (;;) { 204 while ((alq = LIST_FIRST(&ald_active)) == NULL && 205 !ald_shutingdown) 206 mtx_sleep(&ald_active, &ald_mtx, PWAIT, "aldslp", 0); 207 208 /* Don't shutdown until all active ALQs are flushed. */ 209 if (ald_shutingdown && alq == NULL) { 210 ALD_UNLOCK(); 211 break; 212 } 213 214 ALQ_LOCK(alq); 215 ald_deactivate(alq); 216 ALD_UNLOCK(); 217 needwakeup = alq_doio(alq); 218 ALQ_UNLOCK(alq); 219 if (needwakeup) 220 wakeup_one(alq); 221 ALD_LOCK(); 222 } 223 224 kproc_exit(0); 225} 226 227static void 228ald_shutdown(void *arg, int howto) 229{ 230 struct alq *alq; 231 232 if ((howto & RB_NOSYNC) != 0 || SCHEDULER_STOPPED()) 233 return; 234 235 ALD_LOCK(); 236 237 /* Ensure no new queues can be created. */ 238 ald_shutingdown = 1; 239 240 /* Shutdown all ALQs prior to terminating the ald_daemon. */ 241 while ((alq = LIST_FIRST(&ald_queues)) != NULL) { 242 LIST_REMOVE(alq, aq_link); 243 ALD_UNLOCK(); 244 alq_shutdown(alq); 245 ALD_LOCK(); 246 } 247 248 /* At this point, all ALQs are flushed and shutdown. */ 249 250 /* 251 * Wake ald_daemon so that it exits. It won't be able to do 252 * anything until we mtx_sleep because we hold the ald_mtx. 253 */ 254 wakeup(&ald_active); 255 256 /* Wait for ald_daemon to exit. */ 257 mtx_sleep(ald_proc, &ald_mtx, PWAIT, "aldslp", 0); 258 259 ALD_UNLOCK(); 260} 261 262static void 263alq_shutdown(struct alq *alq) 264{ 265 ALQ_LOCK(alq); 266 267 /* Stop any new writers. */ 268 alq->aq_flags |= AQ_SHUTDOWN; 269 270 /* 271 * If the ALQ isn't active but has unwritten data (possible if 272 * the ALQ_NOACTIVATE flag has been used), explicitly activate the 273 * ALQ here so that the pending data gets flushed by the ald_daemon. 274 */ 275 if (!(alq->aq_flags & AQ_ACTIVE) && HAS_PENDING_DATA(alq)) { 276 alq->aq_flags |= AQ_ACTIVE; 277 ALQ_UNLOCK(alq); 278 ALD_LOCK(); 279 ald_activate(alq); 280 ALD_UNLOCK(); 281 ALQ_LOCK(alq); 282 } 283 284 /* Drain IO */ 285 while (alq->aq_flags & AQ_ACTIVE) { 286 alq->aq_flags |= AQ_WANTED; 287 msleep_spin(alq, &alq->aq_mtx, "aldclose", 0); 288 } 289 ALQ_UNLOCK(alq); 290 291 vn_close(alq->aq_vp, FWRITE, alq->aq_cred, 292 curthread); 293 crfree(alq->aq_cred); 294} 295 296void 297alq_destroy(struct alq *alq) 298{ 299 /* Drain all pending IO. */ 300 alq_shutdown(alq); 301 302 mtx_destroy(&alq->aq_mtx); 303 free(alq->aq_entbuf, M_ALD); 304 free(alq, M_ALD); 305} 306 307/* 308 * Flush all pending data to disk. This operation will block. 309 */ 310static int 311alq_doio(struct alq *alq) 312{ 313 struct thread *td; 314 struct mount *mp; 315 struct vnode *vp; 316 struct uio auio; 317 struct iovec aiov[2]; 318 int totlen; 319 int iov; 320 int wrapearly; 321 322 KASSERT((HAS_PENDING_DATA(alq)), ("%s: queue empty!", __func__)); 323 324 vp = alq->aq_vp; 325 td = curthread; 326 totlen = 0; 327 iov = 1; 328 wrapearly = alq->aq_wrapearly; 329 330 bzero(&aiov, sizeof(aiov)); 331 bzero(&auio, sizeof(auio)); 332 333 /* Start the write from the location of our buffer tail pointer. */ 334 aiov[0].iov_base = alq->aq_entbuf + alq->aq_writetail; 335 336 if (alq->aq_writetail < alq->aq_writehead) { 337 /* Buffer not wrapped. */ 338 totlen = aiov[0].iov_len = alq->aq_writehead - alq->aq_writetail; 339 } else if (alq->aq_writehead == 0) { 340 /* Buffer not wrapped (special case to avoid an empty iov). */ 341 totlen = aiov[0].iov_len = alq->aq_buflen - alq->aq_writetail - 342 wrapearly; 343 } else { 344 /* 345 * Buffer wrapped, requires 2 aiov entries: 346 * - first is from writetail to end of buffer 347 * - second is from start of buffer to writehead 348 */ 349 aiov[0].iov_len = alq->aq_buflen - alq->aq_writetail - 350 wrapearly; 351 iov++; 352 aiov[1].iov_base = alq->aq_entbuf; 353 aiov[1].iov_len = alq->aq_writehead; 354 totlen = aiov[0].iov_len + aiov[1].iov_len; 355 } 356 357 alq->aq_flags |= AQ_FLUSHING; 358 ALQ_UNLOCK(alq); 359 360 auio.uio_iov = &aiov[0]; 361 auio.uio_offset = 0; 362 auio.uio_segflg = UIO_SYSSPACE; 363 auio.uio_rw = UIO_WRITE; 364 auio.uio_iovcnt = iov; 365 auio.uio_resid = totlen; 366 auio.uio_td = td; 367 368 /* 369 * Do all of the junk required to write now. 370 */ 371 vn_start_write(vp, &mp, V_WAIT); 372 vn_lock(vp, LK_EXCLUSIVE | LK_RETRY); 373 /* 374 * XXX: VOP_WRITE error checks are ignored. 375 */ 376#ifdef MAC 377 if (mac_vnode_check_write(alq->aq_cred, NOCRED, vp) == 0) 378#endif 379 VOP_WRITE(vp, &auio, IO_UNIT | IO_APPEND, alq->aq_cred); 380 VOP_UNLOCK(vp); 381 vn_finished_write(mp); 382 383 ALQ_LOCK(alq); 384 alq->aq_flags &= ~AQ_FLUSHING; 385 386 /* Adjust writetail as required, taking into account wrapping. */ 387 alq->aq_writetail = (alq->aq_writetail + totlen + wrapearly) % 388 alq->aq_buflen; 389 alq->aq_freebytes += totlen + wrapearly; 390 391 /* 392 * If we just flushed part of the buffer which wrapped, reset the 393 * wrapearly indicator. 394 */ 395 if (wrapearly) 396 alq->aq_wrapearly = 0; 397 398 /* 399 * If we just flushed the buffer completely, reset indexes to 0 to 400 * minimise buffer wraps. 401 * This is also required to ensure alq_getn() can't wedge itself. 402 */ 403 if (!HAS_PENDING_DATA(alq)) 404 alq->aq_writehead = alq->aq_writetail = 0; 405 406 KASSERT((alq->aq_writetail >= 0 && alq->aq_writetail < alq->aq_buflen), 407 ("%s: aq_writetail < 0 || aq_writetail >= aq_buflen", __func__)); 408 409 if (alq->aq_flags & AQ_WANTED) { 410 alq->aq_flags &= ~AQ_WANTED; 411 return (1); 412 } 413 414 return(0); 415} 416 417static struct kproc_desc ald_kp = { 418 "ALQ Daemon", 419 ald_daemon, 420 &ald_proc 421}; 422 423SYSINIT(aldthread, SI_SUB_KTHREAD_IDLE, SI_ORDER_ANY, kproc_start, &ald_kp); 424SYSINIT(ald, SI_SUB_LOCK, SI_ORDER_ANY, ald_startup, NULL); 425 426/* User visible queue functions */ 427 428/* 429 * Create the queue data structure, allocate the buffer, and open the file. 430 */ 431 432int 433alq_open_flags(struct alq **alqp, const char *file, struct ucred *cred, int cmode, 434 int size, int flags) 435{ 436 struct nameidata nd; 437 struct alq *alq; 438 int oflags; 439 int error; 440 441 KASSERT((size > 0), ("%s: size <= 0", __func__)); 442 443 *alqp = NULL; 444 445 NDINIT(&nd, LOOKUP, NOFOLLOW, UIO_SYSSPACE, file); 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_PNBUF(&nd); 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 482int 483alq_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 */ 507int 508alq_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 645unlock: 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 661int 662alq_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 */ 673struct ale * 674alq_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 817struct ale * 818alq_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 826void 827alq_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 884void 885alq_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 */ 914void 915alq_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 922static int 923alq_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 964static moduledata_t alq_mod = 965{ 966 "alq", 967 alq_load_handler, 968 NULL 969}; 970 971DECLARE_MODULE(alq, alq_mod, SI_SUB_LAST, SI_ORDER_ANY); 972MODULE_VERSION(alq, 1); 973