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