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 */
58 struct 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
92 static MALLOC_DEFINE(M_ALD, "ALD", "ALD");
93
94 /*
95 * The ald_mtx protects the ald_queues list and the ald_active list.
96 */
97 static struct mtx ald_mtx;
98 static LIST_HEAD(, alq) ald_queues;
99 static LIST_HEAD(, alq) ald_active;
100 static int ald_shutingdown = 0;
101 struct thread *ald_thread;
102 static struct proc *ald_proc;
103 static 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 */
109 static int ald_add(struct alq *);
110 static int ald_rem(struct alq *);
111 static void ald_startup(void *);
112 static void ald_daemon(void);
113 static void ald_shutdown(void *, int);
114 static void ald_activate(struct alq *);
115 static void ald_deactivate(struct alq *);
116
117 /* Internal queue functions */
118 static void alq_shutdown(struct alq *);
119 static void alq_destroy(struct alq *);
120 static int alq_doio(struct alq *);
121
122 /*
123 * Add a new queue to the global list. Fail if we're shutting down.
124 */
125 static int
ald_add(struct alq * alq)126 ald_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);
138 done:
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 */
147 static int
ald_rem(struct alq * alq)148 ald_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);
160 done:
161 ALD_UNLOCK();
162 return (error);
163 }
164
165 /*
166 * Put a queue on the active list. This will schedule it for writing.
167 */
168 static void
ald_activate(struct alq * alq)169 ald_activate(struct alq *alq)
170 {
171 LIST_INSERT_HEAD(&ald_active, alq, aq_act);
172 wakeup(&ald_active);
173 }
174
175 static void
ald_deactivate(struct alq * alq)176 ald_deactivate(struct alq *alq)
177 {
178 LIST_REMOVE(alq, aq_act);
179 alq->aq_flags &= ~AQ_ACTIVE;
180 }
181
182 static void
ald_startup(void * unused)183 ald_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
190 static void
ald_daemon(void)191 ald_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
227 static void
ald_shutdown(void * arg,int howto)228 ald_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
262 static void
alq_shutdown(struct alq * alq)263 alq_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
296 void
alq_destroy(struct alq * alq)297 alq_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 */
310 static int
alq_doio(struct alq * alq)311 alq_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
417 static struct kproc_desc ald_kp = {
418 "ALQ Daemon",
419 ald_daemon,
420 &ald_proc
421 };
422
423 SYSINIT(aldthread, SI_SUB_KTHREAD_IDLE, SI_ORDER_ANY, kproc_start, &ald_kp);
424 SYSINIT(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
432 int
alq_open_flags(struct alq ** alqp,const char * file,struct ucred * cred,int cmode,int size,int flags)433 alq_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
482 int
alq_open(struct alq ** alqp,const char * file,struct ucred * cred,int cmode,int size,int count)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
alq_writen(struct alq * alq,void * data,int len,int flags)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
alq_write(struct alq * alq,void * data,int flags)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 *
alq_getn(struct alq * alq,int len,int flags)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 *
alq_get(struct alq * alq,int flags)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
alq_post_flags(struct alq * alq,struct ale * ale,int flags)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
alq_flush(struct alq * alq)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
alq_close(struct alq * alq)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
alq_load_handler(module_t mod,int what,void * arg)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);
973