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 *
| 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 $
| 26 * $FreeBSD: head/sys/kern/kern_alq.c 103995 2002-09-26 07:38:56Z 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;
| 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;
| 77static int ald_shutingdown = 0;
|
| 78struct thread *ald_thread;
79static struct proc *ald_proc;
|
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
| 80
81#define ALD_LOCK() mtx_lock(&ald_mtx)
82#define ALD_UNLOCK() mtx_unlock(&ald_mtx)
83
84/* Daemon functions */
85static int ald_add(struct alq *);
86static int ald_rem(struct alq *);
87static void ald_startup(void *);
88static void ald_daemon(void);
89static void ald_shutdown(void *, int);
90static void ald_activate(struct alq *);
91static void ald_deactivate(struct alq *);
92
93/* Internal queue functions */
94static void alq_shutdown(struct alq *);
95static int alq_doio(struct alq *);
96
97
98/*
99 * Add a new queue to the global list. Fail if we're shutting down.
100 */
101static int
102ald_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);
114done:
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 */
123static int
124ald_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);
136done:
137 ALD_UNLOCK();
138 return (error);
139}
140
141/*
142 * Put a queue on the active list. This will schedule it for writing.
143 */
144static void
145ald_activate(struct alq *alq)
146{
147 LIST_INSERT_HEAD(&ald_active, alq, aq_act);
148 wakeup(&ald_active);
149}
150
151static void
152ald_deactivate(struct alq *alq)
153{
154 LIST_REMOVE(alq, aq_act);
155 alq->aq_flags &= ~AQ_ACTIVE;
156}
157
158static void
159ald_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
166static void
167ald_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
|
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
| 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
196static void
197ald_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
213static void
214alq_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
|
227 vn_close(alq->aq_vp, FREAD|FWRITE, alq->aq_cred,
| 230 vn_close(alq->aq_vp, 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,
| 231 curthread);
232 crfree(alq->aq_cred);
233}
234
235/*
236 * Flush all pending data to disk. This operation will block.
237 */
238static int
239alq_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
313static struct kproc_desc ald_kp = {
314 "ALQ Daemon",
315 ald_daemon,
|
313 &ald_thread
| 316 &ald_proc
|
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);
| 317};
318
319SYSINIT(aldthread, SI_SUB_KTHREAD_IDLE, SI_ORDER_ANY, kproc_start, &ald_kp)
320SYSINIT(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 */
328int
329alq_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);
|
342 flags = FREAD | FWRITE | O_NOFOLLOW | O_CREAT;
| 345 flags = 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}
| 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 */
393int
394alq_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
407struct ale *
408alq_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
438void
439alq_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
462void
463alq_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 */
484void
485alq_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}
|