kern_alq.c revision 103830
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: head/sys/kern/kern_alq.c 103830 2002-09-23 05:20:00Z jeff $ 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 */ 46struct 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 69static MALLOC_DEFINE(M_ALD, "ALD", "ALD"); 70 71/* 72 * The ald_mtx protects the ald_queues list and the ald_active list. 73 */ 74static struct mtx ald_mtx; 75static LIST_HEAD(, alq) ald_queues; 76static LIST_HEAD(, alq) ald_active; 77static struct proc *ald_thread; 78static int ald_shutingdown = 0; 79 80#define ALD_LOCK() mtx_lock(&ald_mtx) 81#define ALD_UNLOCK() mtx_unlock(&ald_mtx) 82 83/* Daemon functions */ 84static int ald_add(struct alq *); 85static int ald_rem(struct alq *); 86static void ald_startup(void *); 87static void ald_daemon(void); 88static void ald_shutdown(void *, int); 89static void ald_activate(struct alq *); 90static void ald_deactivate(struct alq *); 91 92/* Internal queue functions */ 93static void alq_shutdown(struct alq *); 94static int alq_doio(struct alq *); 95 96 97/* 98 * Add a new queue to the global list. Fail if we're shutting down. 99 */ 100static int 101ald_add(struct alq *alq) 102{ 103 int error; 104 105 error = 0; 106 107 ALD_LOCK(); 108 if (ald_shutingdown) { 109 error = EBUSY; 110 goto done; 111 } 112 LIST_INSERT_HEAD(&ald_queues, alq, aq_link); 113done: 114 ALD_UNLOCK(); 115 return (error); 116} 117 118/* 119 * Remove a queue from the global list unless we're shutting down. If so, 120 * the ald will take care of cleaning up it's resources. 121 */ 122static int 123ald_rem(struct alq *alq) 124{ 125 int error; 126 127 error = 0; 128 129 ALD_LOCK(); 130 if (ald_shutingdown) { 131 error = EBUSY; 132 goto done; 133 } 134 LIST_REMOVE(alq, aq_link); 135done: 136 ALD_UNLOCK(); 137 return (error); 138} 139 140/* 141 * Put a queue on the active list. This will schedule it for writing. 142 */ 143static void 144ald_activate(struct alq *alq) 145{ 146 LIST_INSERT_HEAD(&ald_active, alq, aq_act); 147 wakeup(&ald_active); 148} 149 150static void 151ald_deactivate(struct alq *alq) 152{ 153 LIST_REMOVE(alq, aq_act); 154 alq->aq_flags &= ~AQ_ACTIVE; 155} 156 157static void 158ald_startup(void *unused) 159{ 160 mtx_init(&ald_mtx, "ALDmtx", NULL, MTX_DEF|MTX_QUIET); 161 LIST_INIT(&ald_queues); 162 LIST_INIT(&ald_active); 163} 164 165static void 166ald_daemon(void) 167{ 168 int needwakeup; 169 struct alq *alq; 170 171 mtx_lock(&Giant); 172 173 EVENTHANDLER_REGISTER(shutdown_pre_sync, ald_shutdown, NULL, 174 SHUTDOWN_PRI_FIRST); 175 176 ALD_LOCK(); 177 178 for (;;) { 179 while ((alq = LIST_FIRST(&ald_active)) == NULL) 180 msleep(&ald_active, &ald_mtx, PWAIT, "aldslp", 0); 181 182 ALQ_LOCK(alq); 183 ald_deactivate(alq); 184 ALD_UNLOCK(); 185 needwakeup = alq_doio(alq); 186 ALQ_UNLOCK(alq); 187 if (needwakeup) 188 wakeup(alq); 189 ALD_LOCK(); 190 } 191} 192 193static void 194ald_shutdown(void *arg, int howto) 195{ 196 struct alq *alq; 197 198 ALD_LOCK(); 199 ald_shutingdown = 1; 200 201 while ((alq = LIST_FIRST(&ald_queues)) != NULL) { 202 LIST_REMOVE(alq, aq_link); 203 ALD_UNLOCK(); 204 alq_shutdown(alq); 205 ALD_LOCK(); 206 } 207 ALD_UNLOCK(); 208} 209 210static void 211alq_shutdown(struct alq *alq) 212{ 213 ALQ_LOCK(alq); 214 215 /* Stop any new writers. */ 216 alq->aq_flags |= AQ_SHUTDOWN; 217 218 /* Drain IO */ 219 while (alq->aq_flags & (AQ_FLUSHING|AQ_ACTIVE)) { 220 alq->aq_flags |= AQ_WANTED; 221 ALQ_UNLOCK(alq); 222 tsleep(alq, PWAIT, "aldclose", 0); 223 ALQ_LOCK(alq); 224 } 225 ALQ_UNLOCK(alq); 226 227 vn_close(alq->aq_vp, FREAD|FWRITE, alq->aq_cred, 228 curthread); 229 crfree(alq->aq_cred); 230} 231 232/* 233 * Flush all pending data to disk. This operation will block. 234 */ 235static int 236alq_doio(struct alq *alq) 237{ 238 struct thread *td; 239 struct mount *mp; 240 struct vnode *vp; 241 struct uio auio; 242 struct iovec aiov[2]; 243 struct ale *ale; 244 struct ale *alstart; 245 int totlen; 246 int iov; 247 248 vp = alq->aq_vp; 249 td = curthread; 250 totlen = 0; 251 iov = 0; 252 253 alstart = ale = alq->aq_entvalid; 254 alq->aq_entvalid = NULL; 255 256 bzero(&aiov, sizeof(aiov)); 257 bzero(&auio, sizeof(auio)); 258 259 do { 260 if (aiov[iov].iov_base == NULL) 261 aiov[iov].iov_base = ale->ae_data; 262 aiov[iov].iov_len += alq->aq_entlen; 263 totlen += alq->aq_entlen; 264 /* Check to see if we're wrapping the buffer */ 265 if (ale->ae_data + alq->aq_entlen != ale->ae_next->ae_data) 266 iov++; 267 ale->ae_flags &= ~AE_VALID; 268 ale = ale->ae_next; 269 } while (ale->ae_flags & AE_VALID); 270 271 alq->aq_flags |= AQ_FLUSHING; 272 ALQ_UNLOCK(alq); 273 274 if (iov == 2 || aiov[iov].iov_base == NULL) 275 iov--; 276 277 auio.uio_iov = &aiov[0]; 278 auio.uio_offset = 0; 279 auio.uio_segflg = UIO_SYSSPACE; 280 auio.uio_rw = UIO_WRITE; 281 auio.uio_iovcnt = iov + 1; 282 auio.uio_resid = totlen; 283 auio.uio_td = td; 284 285 /* 286 * Do all of the junk required to write now. 287 */ 288 vn_start_write(vp, &mp, V_WAIT); 289 vn_lock(vp, LK_EXCLUSIVE | LK_RETRY, td); 290 VOP_LEASE(vp, td, alq->aq_cred, LEASE_WRITE); 291 /* XXX error ignored */ 292 VOP_WRITE(vp, &auio, IO_UNIT | IO_APPEND, alq->aq_cred); 293 VOP_UNLOCK(vp, 0, td); 294 vn_finished_write(mp); 295 296 ALQ_LOCK(alq); 297 alq->aq_flags &= ~AQ_FLUSHING; 298 299 if (alq->aq_entfree == NULL) 300 alq->aq_entfree = alstart; 301 302 if (alq->aq_flags & AQ_WANTED) { 303 alq->aq_flags &= ~AQ_WANTED; 304 return (1); 305 } 306 307 return(0); 308} 309 310static struct kproc_desc ald_kp = { 311 "ALQ Daemon", 312 ald_daemon, 313 &ald_thread 314}; 315 316SYSINIT(aldthread, SI_SUB_KTHREAD_IDLE, SI_ORDER_ANY, kproc_start, &ald_kp) 317SYSINIT(ald, SI_SUB_LOCK, SI_ORDER_ANY, ald_startup, NULL) 318 319 320/* User visible queue functions */ 321 322/* 323 * Create the queue data structure, allocate the buffer, and open the file. 324 */ 325int 326alq_open(struct alq **alqp, const char *file, int size, int count) 327{ 328 struct thread *td; 329 struct nameidata nd; 330 struct ale *ale; 331 struct ale *alp; 332 struct alq *alq; 333 char *bufp; 334 int flags; 335 int error; 336 int i; 337 338 *alqp = NULL; 339 td = curthread; 340 341 NDINIT(&nd, LOOKUP, NOFOLLOW, UIO_SYSSPACE, file, td); 342 flags = FREAD | FWRITE | O_NOFOLLOW | O_CREAT; 343 344 error = vn_open(&nd, &flags, 0); 345 if (error) 346 return (error); 347 348 NDFREE(&nd, NDF_ONLY_PNBUF); 349 /* We just unlock so we hold a reference */ 350 VOP_UNLOCK(nd.ni_vp, 0, td); 351 352 alq = malloc(sizeof(*alq), M_ALD, M_WAITOK|M_ZERO); 353 alq->aq_entbuf = malloc(count * size, M_ALD, M_WAITOK|M_ZERO); 354 alq->aq_first = malloc(sizeof(*ale) * count, M_ALD, M_WAITOK|M_ZERO); 355 alq->aq_vp = nd.ni_vp; 356 alq->aq_cred = crhold(td->td_ucred); 357 alq->aq_entmax = count; 358 alq->aq_entlen = size; 359 alq->aq_entfree = alq->aq_first; 360 361 mtx_init(&alq->aq_mtx, "ALD Queue", NULL, MTX_SPIN|MTX_QUIET); 362 363 bufp = alq->aq_entbuf; 364 ale = alq->aq_first; 365 alp = NULL; 366 367 /* Match up entries with buffers */ 368 for (i = 0; i < count; i++) { 369 if (alp) 370 alp->ae_next = ale; 371 ale->ae_data = bufp; 372 alp = ale; 373 ale++; 374 bufp += size; 375 } 376 377 alp->ae_next = alq->aq_first; 378 379 if ((error = ald_add(alq)) != 0) 380 return (error); 381 *alqp = alq; 382 383 return (0); 384} 385 386/* 387 * Copy a new entry into the queue. If the operation would block either 388 * wait or return an error depending on the value of waitok. 389 */ 390int 391alq_write(struct alq *alq, void *data, int waitok) 392{ 393 struct ale *ale; 394 395 if ((ale = alq_get(alq, waitok)) == NULL) 396 return (EWOULDBLOCK); 397 398 bcopy(data, ale->ae_data, alq->aq_entlen); 399 alq_post(alq, ale); 400 401 return (0); 402} 403 404struct ale * 405alq_get(struct alq *alq, int waitok) 406{ 407 struct ale *ale; 408 struct ale *aln; 409 410 ale = NULL; 411 412 ALQ_LOCK(alq); 413 414 /* Loop until we get an entry or we're shutting down */ 415 while ((alq->aq_flags & AQ_SHUTDOWN) == 0 && 416 (ale = alq->aq_entfree) == NULL && 417 (waitok & ALQ_WAITOK)) { 418 alq->aq_flags |= AQ_WANTED; 419 ALQ_UNLOCK(alq); 420 tsleep(alq, PWAIT, "alqget", 0); 421 ALQ_LOCK(alq); 422 } 423 424 if (ale != NULL) { 425 aln = ale->ae_next; 426 if ((aln->ae_flags & AE_VALID) == 0) 427 alq->aq_entfree = aln; 428 } else 429 ALQ_UNLOCK(alq); 430 431 432 return (ale); 433} 434 435void 436alq_post(struct alq *alq, struct ale *ale) 437{ 438 int activate; 439 440 ale->ae_flags |= AE_VALID; 441 442 if (alq->aq_entvalid == NULL) 443 alq->aq_entvalid = ale; 444 445 if ((alq->aq_flags & AQ_ACTIVE) == 0) { 446 alq->aq_flags |= AQ_ACTIVE; 447 activate = 1; 448 } else 449 activate = 0; 450 451 ALQ_UNLOCK(alq); 452 if (activate) { 453 ALD_LOCK(); 454 ald_activate(alq); 455 ALD_UNLOCK(); 456 } 457} 458 459void 460alq_flush(struct alq *alq) 461{ 462 int needwakeup = 0; 463 464 ALD_LOCK(); 465 ALQ_LOCK(alq); 466 if (alq->aq_flags & AQ_ACTIVE) { 467 ald_deactivate(alq); 468 ALD_UNLOCK(); 469 needwakeup = alq_doio(alq); 470 } else 471 ALD_UNLOCK(); 472 ALQ_UNLOCK(alq); 473 474 if (needwakeup) 475 wakeup(alq); 476} 477 478/* 479 * Flush remaining data, close the file and free all resources. 480 */ 481void 482alq_close(struct alq *alq) 483{ 484 /* 485 * If we're already shuting down someone else will flush and close 486 * the vnode. 487 */ 488 if (ald_rem(alq) != 0) 489 return; 490 491 /* 492 * Drain all pending IO. 493 */ 494 alq_shutdown(alq); 495 496 mtx_destroy(&alq->aq_mtx); 497 free(alq->aq_first, M_ALD); 498 free(alq->aq_entbuf, M_ALD); 499 free(alq, M_ALD); 500} 501