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 } 504