58bd1208b04ebb78cb000dc9af1d9b1f828d064a
[dragonfly.git] / sys / kern / vfs_journal.c
1 /*
2  * Copyright (c) 2004-2006 The DragonFly Project.  All rights reserved.
3  * 
4  * This code is derived from software contributed to The DragonFly Project
5  * by Matthew Dillon <dillon@backplane.com>
6  * 
7  * Redistribution and use in source and binary forms, with or without
8  * modification, are permitted provided that the following conditions
9  * are met:
10  * 
11  * 1. Redistributions of source code must retain the above copyright
12  *    notice, this list of conditions and the following disclaimer.
13  * 2. Redistributions in binary form must reproduce the above copyright
14  *    notice, this list of conditions and the following disclaimer in
15  *    the documentation and/or other materials provided with the
16  *    distribution.
17  * 3. Neither the name of The DragonFly Project nor the names of its
18  *    contributors may be used to endorse or promote products derived
19  *    from this software without specific, prior written permission.
20  * 
21  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22  * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
24  * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE
25  * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
26  * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
27  * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
28  * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
29  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
30  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
31  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
32  * SUCH DAMAGE.
33  *
34  * $DragonFly: src/sys/kern/vfs_journal.c,v 1.33 2007/05/09 00:53:34 dillon Exp $
35  */
36 /*
37  * The journaling protocol is intended to evolve into a two-way stream
38  * whereby transaction IDs can be acknowledged by the journaling target
39  * when the data has been committed to hard storage.  Both implicit and
40  * explicit acknowledgement schemes will be supported, depending on the
41  * sophistication of the journaling stream, plus resynchronization and
42  * restart when a journaling stream is interrupted.  This information will
43  * also be made available to journaling-aware filesystems to allow better
44  * management of their own physical storage synchronization mechanisms as
45  * well as to allow such filesystems to take direct advantage of the kernel's
46  * journaling layer so they don't have to roll their own.
47  *
48  * In addition, the worker thread will have access to much larger 
49  * spooling areas then the memory buffer is able to provide by e.g. 
50  * reserving swap space, in order to absorb potentially long interruptions
51  * of off-site journaling streams, and to prevent 'slow' off-site linkages
52  * from radically slowing down local filesystem operations.  
53  *
54  * Because of the non-trivial algorithms the journaling system will be
55  * required to support, use of a worker thread is mandatory.  Efficiencies
56  * are maintained by utilitizing the memory FIFO to batch transactions when
57  * possible, reducing the number of gratuitous thread switches and taking
58  * advantage of cpu caches through the use of shorter batched code paths
59  * rather then trying to do everything in the context of the process
60  * originating the filesystem op.  In the future the memory FIFO can be
61  * made per-cpu to remove BGL or other locking requirements.
62  */
63 #include <sys/param.h>
64 #include <sys/systm.h>
65 #include <sys/buf.h>
66 #include <sys/conf.h>
67 #include <sys/kernel.h>
68 #include <sys/queue.h>
69 #include <sys/lock.h>
70 #include <sys/malloc.h>
71 #include <sys/mount.h>
72 #include <sys/unistd.h>
73 #include <sys/vnode.h>
74 #include <sys/poll.h>
75 #include <sys/mountctl.h>
76 #include <sys/journal.h>
77 #include <sys/file.h>
78 #include <sys/proc.h>
79 #include <sys/msfbuf.h>
80 #include <sys/socket.h>
81 #include <sys/socketvar.h>
82
83 #include <machine/limits.h>
84
85 #include <vm/vm.h>
86 #include <vm/vm_object.h>
87 #include <vm/vm_page.h>
88 #include <vm/vm_pager.h>
89 #include <vm/vnode_pager.h>
90
91 #include <sys/file2.h>
92 #include <sys/thread2.h>
93
94 static void journal_wthread(void *info);
95 static void journal_rthread(void *info);
96
97 static void *journal_reserve(struct journal *jo,
98                         struct journal_rawrecbeg **rawpp,
99                         int16_t streamid, int bytes);
100 static void *journal_extend(struct journal *jo,
101                         struct journal_rawrecbeg **rawpp,
102                         int truncbytes, int bytes, int *newstreamrecp);
103 static void journal_abort(struct journal *jo,
104                         struct journal_rawrecbeg **rawpp);
105 static void journal_commit(struct journal *jo,
106                         struct journal_rawrecbeg **rawpp,
107                         int bytes, int closeout);
108
109
110 MALLOC_DEFINE(M_JOURNAL, "journal", "Journaling structures");
111 MALLOC_DEFINE(M_JFIFO, "journal-fifo", "Journal FIFO");
112
113 void
114 journal_create_threads(struct journal *jo)
115 {
116         jo->flags &= ~(MC_JOURNAL_STOP_REQ | MC_JOURNAL_STOP_IMM);
117         jo->flags |= MC_JOURNAL_WACTIVE;
118         lwkt_create(journal_wthread, jo, NULL, &jo->wthread,
119                         TDF_STOPREQ, -1, "journal w:%.*s", JIDMAX, jo->id);
120         lwkt_setpri(&jo->wthread, TDPRI_KERN_DAEMON);
121         lwkt_schedule(&jo->wthread);
122
123         if (jo->flags & MC_JOURNAL_WANT_FULLDUPLEX) {
124             jo->flags |= MC_JOURNAL_RACTIVE;
125             lwkt_create(journal_rthread, jo, NULL, &jo->rthread,
126                         TDF_STOPREQ, -1, "journal r:%.*s", JIDMAX, jo->id);
127             lwkt_setpri(&jo->rthread, TDPRI_KERN_DAEMON);
128             lwkt_schedule(&jo->rthread);
129         }
130 }
131
132 void
133 journal_destroy_threads(struct journal *jo, int flags)
134 {
135     int wcount;
136
137     jo->flags |= MC_JOURNAL_STOP_REQ | (flags & MC_JOURNAL_STOP_IMM);
138     wakeup(&jo->fifo);
139     wcount = 0;
140     while (jo->flags & (MC_JOURNAL_WACTIVE | MC_JOURNAL_RACTIVE)) {
141         tsleep(jo, 0, "jwait", hz);
142         if (++wcount % 10 == 0) {
143             kprintf("Warning: journal %s waiting for descriptors to close\n",
144                 jo->id);
145         }
146     }
147
148     /*
149      * XXX SMP - threads should move to cpu requesting the restart or
150      * termination before finishing up to properly interlock.
151      */
152     tsleep(jo, 0, "jwait", hz);
153     lwkt_free_thread(&jo->wthread);
154     if (jo->flags & MC_JOURNAL_WANT_FULLDUPLEX)
155         lwkt_free_thread(&jo->rthread);
156 }
157
158 /*
159  * The per-journal worker thread is responsible for writing out the
160  * journal's FIFO to the target stream.
161  */
162 static void
163 journal_wthread(void *info)
164 {
165     struct journal *jo = info;
166     struct journal_rawrecbeg *rawp;
167     int error;
168     size_t avail;
169     size_t bytes;
170     size_t res;
171
172     for (;;) {
173         /*
174          * Calculate the number of bytes available to write.  This buffer
175          * area may contain reserved records so we can't just write it out
176          * without further checks.
177          */
178         bytes = jo->fifo.windex - jo->fifo.rindex;
179
180         /*
181          * sleep if no bytes are available or if an incomplete record is
182          * encountered (it needs to be filled in before we can write it
183          * out), and skip any pad records that we encounter.
184          */
185         if (bytes == 0) {
186             if (jo->flags & MC_JOURNAL_STOP_REQ)
187                 break;
188             tsleep(&jo->fifo, 0, "jfifo", hz);
189             continue;
190         }
191
192         /*
193          * Sleep if we can not go any further due to hitting an incomplete
194          * record.  This case should occur rarely but may have to be better
195          * optimized XXX.
196          */
197         rawp = (void *)(jo->fifo.membase + (jo->fifo.rindex & jo->fifo.mask));
198         if (rawp->begmagic == JREC_INCOMPLETEMAGIC) {
199             tsleep(&jo->fifo, 0, "jpad", hz);
200             continue;
201         }
202
203         /*
204          * Skip any pad records.  We do not write out pad records if we can
205          * help it. 
206          */
207         if (rawp->streamid == JREC_STREAMID_PAD) {
208             if ((jo->flags & MC_JOURNAL_WANT_FULLDUPLEX) == 0) {
209                 if (jo->fifo.rindex == jo->fifo.xindex) {
210                     jo->fifo.xindex += (rawp->recsize + 15) & ~15;
211                     jo->total_acked += (rawp->recsize + 15) & ~15;
212                 }
213             }
214             jo->fifo.rindex += (rawp->recsize + 15) & ~15;
215             jo->total_acked += bytes;
216             KKASSERT(jo->fifo.windex - jo->fifo.rindex >= 0);
217             continue;
218         }
219
220         /*
221          * 'bytes' is the amount of data that can potentially be written out.  
222          * Calculate 'res', the amount of data that can actually be written
223          * out.  res is bounded either by hitting the end of the physical
224          * memory buffer or by hitting an incomplete record.  Incomplete
225          * records often occur due to the way the space reservation model
226          * works.
227          */
228         res = 0;
229         avail = jo->fifo.size - (jo->fifo.rindex & jo->fifo.mask);
230         while (res < bytes && rawp->begmagic == JREC_BEGMAGIC) {
231             res += (rawp->recsize + 15) & ~15;
232             if (res >= avail) {
233                 KKASSERT(res == avail);
234                 break;
235             }
236             rawp = (void *)((char *)rawp + ((rawp->recsize + 15) & ~15));
237         }
238
239         /*
240          * Issue the write and deal with any errors or other conditions.
241          * For now assume blocking I/O.  Since we are record-aware the
242          * code cannot yet handle partial writes.
243          *
244          * We bump rindex prior to issuing the write to avoid racing
245          * the acknowledgement coming back (which could prevent the ack
246          * from bumping xindex).  Restarts are always based on xindex so
247          * we do not try to undo the rindex if an error occurs.
248          *
249          * XXX EWOULDBLOCK/NBIO
250          * XXX notification on failure
251          * XXX permanent verses temporary failures
252          * XXX two-way acknowledgement stream in the return direction / xindex
253          */
254         bytes = res;
255         jo->fifo.rindex += bytes;
256         error = fp_write(jo->fp, 
257                         jo->fifo.membase +
258                          ((jo->fifo.rindex - bytes) & jo->fifo.mask),
259                         bytes, &res, UIO_SYSSPACE);
260         if (error) {
261             kprintf("journal_thread(%s) write, error %d\n", jo->id, error);
262             /* XXX */
263         } else {
264             KKASSERT(res == bytes);
265         }
266
267         /*
268          * Advance rindex.  If the journal stream is not full duplex we also
269          * advance xindex, otherwise the rjournal thread is responsible for
270          * advancing xindex.
271          */
272         if ((jo->flags & MC_JOURNAL_WANT_FULLDUPLEX) == 0) {
273             jo->fifo.xindex += bytes;
274             jo->total_acked += bytes;
275         }
276         KKASSERT(jo->fifo.windex - jo->fifo.rindex >= 0);
277         if ((jo->flags & MC_JOURNAL_WANT_FULLDUPLEX) == 0) {
278             if (jo->flags & MC_JOURNAL_WWAIT) {
279                 jo->flags &= ~MC_JOURNAL_WWAIT; /* XXX hysteresis */
280                 wakeup(&jo->fifo.windex);
281             }
282         }
283     }
284     fp_shutdown(jo->fp, SHUT_WR);
285     jo->flags &= ~MC_JOURNAL_WACTIVE;
286     wakeup(jo);
287     wakeup(&jo->fifo.windex);
288 }
289
290 /*
291  * A second per-journal worker thread is created for two-way journaling
292  * streams to deal with the return acknowledgement stream.
293  */
294 static void
295 journal_rthread(void *info)
296 {
297     struct journal_rawrecbeg *rawp;
298     struct journal_ackrecord ack;
299     struct journal *jo = info;
300     int64_t transid;
301     int error;
302     size_t count;
303     size_t bytes;
304
305     transid = 0;
306     error = 0;
307
308     for (;;) {
309         /*
310          * We have been asked to stop
311          */
312         if (jo->flags & MC_JOURNAL_STOP_REQ)
313                 break;
314
315         /*
316          * If we have no active transaction id, get one from the return
317          * stream.
318          */
319         if (transid == 0) {
320             error = fp_read(jo->fp, &ack, sizeof(ack), &count, 
321                             1, UIO_SYSSPACE);
322 #if 0
323             kprintf("fp_read ack error %d count %d\n", error, count);
324 #endif
325             if (error || count != sizeof(ack))
326                 break;
327             if (error) {
328                 kprintf("read error %d on receive stream\n", error);
329                 break;
330             }
331             if (ack.rbeg.begmagic != JREC_BEGMAGIC ||
332                 ack.rend.endmagic != JREC_ENDMAGIC
333             ) {
334                 kprintf("bad begmagic or endmagic on receive stream\n");
335                 break;
336             }
337             transid = ack.rbeg.transid;
338         }
339
340         /*
341          * Calculate the number of unacknowledged bytes.  If there are no
342          * unacknowledged bytes then unsent data was acknowledged, report,
343          * sleep a bit, and loop in that case.  This should not happen 
344          * normally.  The ack record is thrown away.
345          */
346         bytes = jo->fifo.rindex - jo->fifo.xindex;
347
348         if (bytes == 0) {
349             kprintf("warning: unsent data acknowledged transid %08llx\n",
350                     (long long)transid);
351             tsleep(&jo->fifo.xindex, 0, "jrseq", hz);
352             transid = 0;
353             continue;
354         }
355
356         /*
357          * Since rindex has advanced, the record pointed to by xindex
358          * must be a valid record.
359          */
360         rawp = (void *)(jo->fifo.membase + (jo->fifo.xindex & jo->fifo.mask));
361         KKASSERT(rawp->begmagic == JREC_BEGMAGIC);
362         KKASSERT(rawp->recsize <= bytes);
363
364         /*
365          * The target can acknowledge several records at once.
366          */
367         if (rawp->transid < transid) {
368 #if 1
369             kprintf("ackskip %08llx/%08llx\n",
370                     (long long)rawp->transid,
371                     (long long)transid);
372 #endif
373             jo->fifo.xindex += (rawp->recsize + 15) & ~15;
374             jo->total_acked += (rawp->recsize + 15) & ~15;
375             if (jo->flags & MC_JOURNAL_WWAIT) {
376                 jo->flags &= ~MC_JOURNAL_WWAIT; /* XXX hysteresis */
377                 wakeup(&jo->fifo.windex);
378             }
379             continue;
380         }
381         if (rawp->transid == transid) {
382 #if 1
383             kprintf("ackskip %08llx/%08llx\n",
384                     (long long)rawp->transid,
385                     (long long)transid);
386 #endif
387             jo->fifo.xindex += (rawp->recsize + 15) & ~15;
388             jo->total_acked += (rawp->recsize + 15) & ~15;
389             if (jo->flags & MC_JOURNAL_WWAIT) {
390                 jo->flags &= ~MC_JOURNAL_WWAIT; /* XXX hysteresis */
391                 wakeup(&jo->fifo.windex);
392             }
393             transid = 0;
394             continue;
395         }
396         kprintf("warning: unsent data(2) acknowledged transid %08llx\n",
397                 (long long)transid);
398         transid = 0;
399     }
400     jo->flags &= ~MC_JOURNAL_RACTIVE;
401     wakeup(jo);
402     wakeup(&jo->fifo.windex);
403 }
404
405 /*
406  * This builds a pad record which the journaling thread will skip over.  Pad
407  * records are required when we are unable to reserve sufficient stream space
408  * due to insufficient space at the end of the physical memory fifo.
409  *
410  * Even though the record is not transmitted, a normal transid must be 
411  * assigned to it so link recovery operations after a failure work properly.
412  */
413 static
414 void
415 journal_build_pad(struct journal_rawrecbeg *rawp, int recsize, int64_t transid)
416 {
417     struct journal_rawrecend *rendp;
418     
419     KKASSERT((recsize & 15) == 0 && recsize >= 16);
420
421     rawp->streamid = JREC_STREAMID_PAD;
422     rawp->recsize = recsize;    /* must be 16-byte aligned */
423     rawp->transid = transid;
424     /*
425      * WARNING, rendp may overlap rawp->transid.  This is necessary to
426      * allow PAD records to fit in 16 bytes.  Use cpu_ccfence() to
427      * hopefully cause the compiler to not make any assumptions.
428      */
429     rendp = (void *)((char *)rawp + rawp->recsize - sizeof(*rendp));
430     rendp->endmagic = JREC_ENDMAGIC;
431     rendp->check = 0;
432     rendp->recsize = rawp->recsize;
433
434     /*
435      * Set the begin magic last.  This is what will allow the journal
436      * thread to write the record out.  Use a store fence to prevent
437      * compiler and cpu reordering of the writes.
438      */
439     cpu_sfence();
440     rawp->begmagic = JREC_BEGMAGIC;
441 }
442
443 /*
444  * Wake up the worker thread if the FIFO is more then half full or if
445  * someone is waiting for space to be freed up.  Otherwise let the 
446  * heartbeat deal with it.  Being able to avoid waking up the worker
447  * is the key to the journal's cpu performance.
448  */
449 static __inline
450 void
451 journal_commit_wakeup(struct journal *jo)
452 {
453     int avail;
454
455     avail = jo->fifo.size - (jo->fifo.windex - jo->fifo.xindex);
456     KKASSERT(avail >= 0);
457     if ((avail < (jo->fifo.size >> 1)) || (jo->flags & MC_JOURNAL_WWAIT))
458         wakeup(&jo->fifo);
459 }
460
461 /*
462  * Create a new BEGIN stream record with the specified streamid and the
463  * specified amount of payload space.  *rawpp will be set to point to the
464  * base of the new stream record and a pointer to the base of the payload
465  * space will be returned.  *rawpp does not need to be pre-NULLd prior to
466  * making this call.  The raw record header will be partially initialized.
467  *
468  * A stream can be extended, aborted, or committed by other API calls
469  * below.  This may result in a sequence of potentially disconnected
470  * stream records to be output to the journaling target.  The first record
471  * (the one created by this function) will be marked JREC_STREAMCTL_BEGIN,
472  * while the last record on commit or abort will be marked JREC_STREAMCTL_END
473  * (and possibly also JREC_STREAMCTL_ABORTED).  The last record could wind
474  * up being the same as the first, in which case the bits are all set in
475  * the first record.
476  *
477  * The stream record is created in an incomplete state by setting the begin
478  * magic to JREC_INCOMPLETEMAGIC.  This prevents the worker thread from
479  * flushing the fifo past our record until we have finished populating it.
480  * Other threads can reserve and operate on their own space without stalling
481  * but the stream output will stall until we have completed operations.  The
482  * memory FIFO is intended to be large enough to absorb such situations
483  * without stalling out other threads.
484  */
485 static
486 void *
487 journal_reserve(struct journal *jo, struct journal_rawrecbeg **rawpp,
488                 int16_t streamid, int bytes)
489 {
490     struct journal_rawrecbeg *rawp;
491     int avail;
492     int availtoend;
493     int req;
494
495     /*
496      * Add header and trailer overheads to the passed payload.  Note that
497      * the passed payload size need not be aligned in any way.
498      */
499     bytes += sizeof(struct journal_rawrecbeg);
500     bytes += sizeof(struct journal_rawrecend);
501
502     for (;;) {
503         /*
504          * First, check boundary conditions.  If the request would wrap around
505          * we have to skip past the ending block and return to the beginning
506          * of the FIFO's buffer.  Calculate 'req' which is the actual number
507          * of bytes being reserved, including wrap-around dead space.
508          *
509          * Neither 'bytes' or 'req' are aligned.
510          *
511          * Note that availtoend is not truncated to avail and so cannot be
512          * used to determine whether the reservation is possible by itself.
513          * Also, since all fifo ops are 16-byte aligned, we can check
514          * the size before calculating the aligned size.
515          */
516         availtoend = jo->fifo.size - (jo->fifo.windex & jo->fifo.mask);
517         KKASSERT((availtoend & 15) == 0);
518         if (bytes > availtoend) 
519             req = bytes + availtoend;   /* add pad to end */
520         else
521             req = bytes;
522
523         /*
524          * Next calculate the total available space and see if it is
525          * sufficient.  We cannot overwrite previously buffered data
526          * past xindex because otherwise we would not be able to restart
527          * a broken link at the target's last point of commit.
528          */
529         avail = jo->fifo.size - (jo->fifo.windex - jo->fifo.xindex);
530         KKASSERT(avail >= 0 && (avail & 15) == 0);
531
532         if (avail < req) {
533             /* XXX MC_JOURNAL_STOP_IMM */
534             jo->flags |= MC_JOURNAL_WWAIT;
535             ++jo->fifostalls;
536             tsleep(&jo->fifo.windex, 0, "jwrite", 0);
537             continue;
538         }
539
540         /*
541          * Create a pad record for any dead space and create an incomplete
542          * record for the live space, then return a pointer to the
543          * contiguous buffer space that was requested.
544          *
545          * NOTE: The worker thread will not flush past an incomplete
546          * record, so the reserved space can be filled in at-will.  The
547          * journaling code must also be aware the reserved sections occuring
548          * after this one will also not be written out even if completed
549          * until this one is completed.
550          *
551          * The transaction id must accomodate real and potential pad creation.
552          */
553         rawp = (void *)(jo->fifo.membase + (jo->fifo.windex & jo->fifo.mask));
554         if (req != bytes) {
555             journal_build_pad(rawp, availtoend, jo->transid);
556             ++jo->transid;
557             rawp = (void *)jo->fifo.membase;
558         }
559         rawp->begmagic = JREC_INCOMPLETEMAGIC;  /* updated by abort/commit */
560         rawp->recsize = bytes;                  /* (unaligned size) */
561         rawp->streamid = streamid | JREC_STREAMCTL_BEGIN;
562         rawp->transid = jo->transid;
563         jo->transid += 2;
564
565         /*
566          * Issue a memory barrier to guarentee that the record data has been
567          * properly initialized before we advance the write index and return
568          * a pointer to the reserved record.  Otherwise the worker thread
569          * could accidently run past us.
570          *
571          * Note that stream records are always 16-byte aligned.
572          */
573         cpu_sfence();
574         jo->fifo.windex += (req + 15) & ~15;
575         *rawpp = rawp;
576         return(rawp + 1);
577     }
578     /* not reached */
579     *rawpp = NULL;
580     return(NULL);
581 }
582
583 /*
584  * Attempt to extend the stream record by <bytes> worth of payload space.
585  *
586  * If it is possible to extend the existing stream record no truncation
587  * occurs and the record is extended as specified.  A pointer to the 
588  * truncation offset within the payload space is returned.
589  *
590  * If it is not possible to do this the existing stream record is truncated
591  * and committed, and a new stream record of size <bytes> is created.  A
592  * pointer to the base of the new stream record's payload space is returned.
593  *
594  * *rawpp is set to the new reservation in the case of a new record but
595  * the caller cannot depend on a comparison with the old rawp to determine if
596  * this case occurs because we could end up using the same memory FIFO
597  * offset for the new stream record.  Use *newstreamrecp instead.
598  */
599 static void *
600 journal_extend(struct journal *jo, struct journal_rawrecbeg **rawpp, 
601                 int truncbytes, int bytes, int *newstreamrecp)
602 {
603     struct journal_rawrecbeg *rawp;
604     int16_t streamid;
605     int availtoend;
606     int avail;
607     int osize;
608     int nsize;
609     int wbase;
610     void *rptr;
611
612     *newstreamrecp = 0;
613     rawp = *rawpp;
614     osize = (rawp->recsize + 15) & ~15;
615     nsize = (rawp->recsize + bytes + 15) & ~15;
616     wbase = (char *)rawp - jo->fifo.membase;
617
618     /*
619      * If the aligned record size does not change we can trivially adjust
620      * the record size.
621      */
622     if (nsize == osize) {
623         rawp->recsize += bytes;
624         return((char *)(rawp + 1) + truncbytes);
625     }
626
627     /*
628      * If the fifo's write index hasn't been modified since we made the
629      * reservation and we do not hit any boundary conditions, we can 
630      * trivially make the record smaller or larger.
631      */
632     if ((jo->fifo.windex & jo->fifo.mask) == wbase + osize) {
633         availtoend = jo->fifo.size - wbase;
634         avail = jo->fifo.size - (jo->fifo.windex - jo->fifo.xindex) + osize;
635         KKASSERT((availtoend & 15) == 0);
636         KKASSERT((avail & 15) == 0);
637         if (nsize <= avail && nsize <= availtoend) {
638             jo->fifo.windex += nsize - osize;
639             rawp->recsize += bytes;
640             return((char *)(rawp + 1) + truncbytes);
641         }
642     }
643
644     /*
645      * It was not possible to extend the buffer.  Commit the current
646      * buffer and create a new one.  We manually clear the BEGIN mark that
647      * journal_reserve() creates (because this is a continuing record, not
648      * the start of a new stream).
649      */
650     streamid = rawp->streamid & JREC_STREAMID_MASK;
651     journal_commit(jo, rawpp, truncbytes, 0);
652     rptr = journal_reserve(jo, rawpp, streamid, bytes);
653     rawp = *rawpp;
654     rawp->streamid &= ~JREC_STREAMCTL_BEGIN;
655     *newstreamrecp = 1;
656     return(rptr);
657 }
658
659 /*
660  * Abort a journal record.  If the transaction record represents a stream
661  * BEGIN and we can reverse the fifo's write index we can simply reverse
662  * index the entire record, as if it were never reserved in the first place.
663  *
664  * Otherwise we set the JREC_STREAMCTL_ABORTED bit and commit the record
665  * with the payload truncated to 0 bytes.
666  */
667 static void
668 journal_abort(struct journal *jo, struct journal_rawrecbeg **rawpp)
669 {
670     struct journal_rawrecbeg *rawp;
671     int osize;
672
673     rawp = *rawpp;
674     osize = (rawp->recsize + 15) & ~15;
675
676     if ((rawp->streamid & JREC_STREAMCTL_BEGIN) &&
677         (jo->fifo.windex & jo->fifo.mask) == 
678          (char *)rawp - jo->fifo.membase + osize)
679     {
680         jo->fifo.windex -= osize;
681         *rawpp = NULL;
682     } else {
683         rawp->streamid |= JREC_STREAMCTL_ABORTED;
684         journal_commit(jo, rawpp, 0, 1);
685     }
686 }
687
688 /*
689  * Commit a journal record and potentially truncate it to the specified
690  * number of payload bytes.  If you do not want to truncate the record,
691  * simply pass -1 for the bytes parameter.  Do not pass rawp->recsize, that
692  * field includes header and trailer and will not be correct.  Note that
693  * passing 0 will truncate the entire data payload of the record.
694  *
695  * The logical stream is terminated by this function.
696  *
697  * If truncation occurs, and it is not possible to physically optimize the
698  * memory FIFO due to other threads having reserved space after ours,
699  * the remaining reserved space will be covered by a pad record.
700  */
701 static void
702 journal_commit(struct journal *jo, struct journal_rawrecbeg **rawpp,
703                 int bytes, int closeout)
704 {
705     struct journal_rawrecbeg *rawp;
706     struct journal_rawrecend *rendp;
707     int osize;
708     int nsize;
709
710     rawp = *rawpp;
711     *rawpp = NULL;
712
713     KKASSERT((char *)rawp >= jo->fifo.membase &&
714              (char *)rawp + rawp->recsize <= jo->fifo.membase + jo->fifo.size);
715     KKASSERT(((intptr_t)rawp & 15) == 0);
716
717     /*
718      * Truncate the record if necessary.  If the FIFO write index as still
719      * at the end of our record we can optimally backindex it.  Otherwise
720      * we have to insert a pad record to cover the dead space.
721      *
722      * We calculate osize which is the 16-byte-aligned original recsize.
723      * We calculate nsize which is the 16-byte-aligned new recsize.
724      *
725      * Due to alignment issues or in case the passed truncation bytes is
726      * the same as the original payload, nsize may be equal to osize even
727      * if the committed bytes is less then the originally reserved bytes.
728      */
729     if (bytes >= 0) {
730         KKASSERT(bytes >= 0 && bytes <= rawp->recsize - sizeof(struct journal_rawrecbeg) - sizeof(struct journal_rawrecend));
731         osize = (rawp->recsize + 15) & ~15;
732         rawp->recsize = bytes + sizeof(struct journal_rawrecbeg) +
733                         sizeof(struct journal_rawrecend);
734         nsize = (rawp->recsize + 15) & ~15;
735         KKASSERT(nsize <= osize);
736         if (osize == nsize) {
737             /* do nothing */
738         } else if ((jo->fifo.windex & jo->fifo.mask) == (char *)rawp - jo->fifo.membase + osize) {
739             /* we are able to backindex the fifo */
740             jo->fifo.windex -= osize - nsize;
741         } else {
742             /* we cannot backindex the fifo, emplace a pad in the dead space */
743             journal_build_pad((void *)((char *)rawp + nsize), osize - nsize,
744                                 rawp->transid + 1);
745         }
746     }
747
748     /*
749      * Fill in the trailer.  Note that unlike pad records, the trailer will
750      * never overlap the header.
751      */
752     rendp = (void *)((char *)rawp + 
753             ((rawp->recsize + 15) & ~15) - sizeof(*rendp));
754     rendp->endmagic = JREC_ENDMAGIC;
755     rendp->recsize = rawp->recsize;
756     rendp->check = 0;           /* XXX check word, disabled for now */
757
758     /*
759      * Fill in begmagic last.  This will allow the worker thread to proceed.
760      * Use a memory barrier to guarentee write ordering.  Mark the stream
761      * as terminated if closeout is set.  This is the typical case.
762      */
763     if (closeout)
764         rawp->streamid |= JREC_STREAMCTL_END;
765     cpu_sfence();               /* memory and compiler barrier */
766     rawp->begmagic = JREC_BEGMAGIC;
767
768     journal_commit_wakeup(jo);
769 }
770
771 /************************************************************************
772  *                      TRANSACTION SUPPORT ROUTINES                    *
773  ************************************************************************
774  *
775  * JRECORD_*() - routines to create subrecord transactions and embed them
776  *               in the logical streams managed by the journal_*() routines.
777  */
778
779 /*
780  * Initialize the passed jrecord structure and start a new stream transaction
781  * by reserving an initial build space in the journal's memory FIFO.
782  */
783 void
784 jrecord_init(struct journal *jo, struct jrecord *jrec, int16_t streamid)
785 {
786     bzero(jrec, sizeof(*jrec));
787     jrec->jo = jo;
788     jrec->streamid = streamid;
789     jrec->stream_residual = JREC_DEFAULTSIZE;
790     jrec->stream_reserved = jrec->stream_residual;
791     jrec->stream_ptr = 
792         journal_reserve(jo, &jrec->rawp, streamid, jrec->stream_reserved);
793 }
794
795 /*
796  * Push a recursive record type.  All pushes should have matching pops.
797  * The old parent is returned and the newly pushed record becomes the
798  * new parent.  Note that the old parent's pointer may already be invalid
799  * or may become invalid if jrecord_write() had to build a new stream
800  * record, so the caller should not mess with the returned pointer in
801  * any way other then to save it.
802  */
803 struct journal_subrecord *
804 jrecord_push(struct jrecord *jrec, int16_t rectype)
805 {
806     struct journal_subrecord *save;
807
808     save = jrec->parent;
809     jrec->parent = jrecord_write(jrec, rectype|JMASK_NESTED, 0);
810     jrec->last = NULL;
811     KKASSERT(jrec->parent != NULL);
812     ++jrec->pushcount;
813     ++jrec->pushptrgood;        /* cleared on flush */
814     return(save);
815 }
816
817 /*
818  * Pop a previously pushed sub-transaction.  We must set JMASK_LAST
819  * on the last record written within the subtransaction.  If the last 
820  * record written is not accessible or if the subtransaction is empty,
821  * we must write out a pad record with JMASK_LAST set before popping.
822  *
823  * When popping a subtransaction the parent record's recsize field
824  * will be properly set.  If the parent pointer is no longer valid
825  * (which can occur if the data has already been flushed out to the
826  * stream), the protocol spec allows us to leave it 0.
827  *
828  * The saved parent pointer which we restore may or may not be valid,
829  * and if not valid may or may not be NULL, depending on the value
830  * of pushptrgood.
831  */
832 void
833 jrecord_pop(struct jrecord *jrec, struct journal_subrecord *save)
834 {
835     struct journal_subrecord *last;
836
837     KKASSERT(jrec->pushcount > 0);
838     KKASSERT(jrec->residual == 0);
839
840     /*
841      * Set JMASK_LAST on the last record we wrote at the current
842      * level.  If last is NULL we either no longer have access to the
843      * record or the subtransaction was empty and we must write out a pad
844      * record.
845      */
846     if ((last = jrec->last) == NULL) {
847         jrecord_write(jrec, JLEAF_PAD|JMASK_LAST, 0);
848         last = jrec->last;      /* reload after possible flush */
849     } else {
850         last->rectype |= JMASK_LAST;
851     }
852
853     /*
854      * pushptrgood tells us how many levels of parent record pointers
855      * are valid.  The jrec only stores the current parent record pointer
856      * (and it is only valid if pushptrgood != 0).  The higher level parent
857      * record pointers are saved by the routines calling jrecord_push() and
858      * jrecord_pop().  These pointers may become stale and we determine
859      * that fact by tracking the count of valid parent pointers with 
860      * pushptrgood.  Pointers become invalid when their related stream
861      * record gets pushed out.
862      *
863      * If no pointer is available (the data has already been pushed out),
864      * then no fixup of e.g. the length field is possible for non-leaf
865      * nodes.  The protocol allows for this situation by placing a larger
866      * burden on the program scanning the stream on the other end.
867      *
868      * [parentA]
869      *    [node X]
870      *    [parentB]
871      *       [node Y]
872      *       [node Z]
873      *    (pop B)       see NOTE B
874      * (pop A)          see NOTE A
875      *
876      * NOTE B:  This pop sets LAST in node Z if the node is still accessible,
877      *          else a PAD record is appended and LAST is set in that.
878      *
879      *          This pop sets the record size in parentB if parentB is still
880      *          accessible, else the record size is left 0 (the scanner must
881      *          deal with that).
882      *
883      *          This pop sets the new 'last' record to parentB, the pointer
884      *          to which may or may not still be accessible.
885      *
886      * NOTE A:  This pop sets LAST in parentB if the node is still accessible,
887      *          else a PAD record is appended and LAST is set in that.
888      *
889      *          This pop sets the record size in parentA if parentA is still
890      *          accessible, else the record size is left 0 (the scanner must
891      *          deal with that).
892      *
893      *          This pop sets the new 'last' record to parentA, the pointer
894      *          to which may or may not still be accessible.
895      *
896      * Also note that the last record in the stream transaction, which in
897      * the above example is parentA, does not currently have the LAST bit
898      * set.
899      *
900      * The current parent becomes the last record relative to the
901      * saved parent passed into us.  It's validity is based on 
902      * whether pushptrgood is non-zero prior to decrementing.  The saved
903      * parent becomes the new parent, and its validity is based on whether
904      * pushptrgood is non-zero after decrementing.
905      *
906      * The old jrec->parent may be NULL if it is no longer accessible.
907      * If pushptrgood is non-zero, however, it is guarenteed to not
908      * be NULL (since no flush occured).
909      */
910     jrec->last = jrec->parent;
911     --jrec->pushcount;
912     if (jrec->pushptrgood) {
913         KKASSERT(jrec->last != NULL && last != NULL);
914         if (--jrec->pushptrgood == 0) {
915             jrec->parent = NULL;        /* 'save' contains garbage or NULL */
916         } else {
917             KKASSERT(save != NULL);
918             jrec->parent = save;        /* 'save' must not be NULL */
919         }
920
921         /*
922          * Set the record size in the old parent.  'last' still points to
923          * the original last record in the subtransaction being popped,
924          * jrec->last points to the old parent (which became the last
925          * record relative to the new parent being popped into).
926          */
927         jrec->last->recsize = (char *)last + last->recsize - (char *)jrec->last;
928     } else {
929         jrec->parent = NULL;
930         KKASSERT(jrec->last == NULL);
931     }
932 }
933
934 /*
935  * Write out a leaf record, including associated data.
936  */
937 void
938 jrecord_leaf(struct jrecord *jrec, int16_t rectype, void *ptr, int bytes)
939 {
940     jrecord_write(jrec, rectype, bytes);
941     jrecord_data(jrec, ptr, bytes);
942 }
943
944 /*
945  * Write a leaf record out and return a pointer to its base.  The leaf
946  * record may contain potentially megabytes of data which is supplied
947  * in jrecord_data() calls.  The exact amount must be specified in this
948  * call.
949  *
950  * THE RETURNED SUBRECORD POINTER IS ONLY VALID IMMEDIATELY AFTER THE
951  * CALL AND MAY BECOME INVALID AT ANY TIME.  ONLY THE PUSH/POP CODE SHOULD
952  * USE THE RETURN VALUE.
953  */
954 struct journal_subrecord *
955 jrecord_write(struct jrecord *jrec, int16_t rectype, int bytes)
956 {
957     struct journal_subrecord *last;
958     int pusheditout;
959
960     /*
961      * Try to catch some obvious errors.  Nesting records must specify a
962      * size of 0, and there should be no left-overs from previous operations
963      * (such as incomplete data writeouts).
964      */
965     KKASSERT(bytes == 0 || (rectype & JMASK_NESTED) == 0);
966     KKASSERT(jrec->residual == 0);
967
968     /*
969      * Check to see if the current stream record has enough room for
970      * the new subrecord header.  If it doesn't we extend the current
971      * stream record.
972      *
973      * This may have the side effect of pushing out the current stream record
974      * and creating a new one.  We must adjust our stream tracking fields
975      * accordingly.
976      */
977     if (jrec->stream_residual < sizeof(struct journal_subrecord)) {
978         jrec->stream_ptr = journal_extend(jrec->jo, &jrec->rawp,
979                                 jrec->stream_reserved - jrec->stream_residual,
980                                 JREC_DEFAULTSIZE, &pusheditout);
981         if (pusheditout) {
982             /*
983              * If a pushout occured, the pushed out stream record was
984              * truncated as specified and the new record is exactly the
985              * extension size specified.
986              */
987             jrec->stream_reserved = JREC_DEFAULTSIZE;
988             jrec->stream_residual = JREC_DEFAULTSIZE;
989             jrec->parent = NULL;        /* no longer accessible */
990             jrec->pushptrgood = 0;      /* restored parents in pops no good */
991         } else {
992             /*
993              * If no pushout occured the stream record is NOT truncated and
994              * IS extended.
995              */
996             jrec->stream_reserved += JREC_DEFAULTSIZE;
997             jrec->stream_residual += JREC_DEFAULTSIZE;
998         }
999     }
1000     last = (void *)jrec->stream_ptr;
1001     last->rectype = rectype;
1002     last->reserved = 0;
1003
1004     /*
1005      * We may not know the record size for recursive records and the 
1006      * header may become unavailable due to limited FIFO space.  Write
1007      * -1 to indicate this special case.
1008      */
1009     if ((rectype & JMASK_NESTED) && bytes == 0)
1010         last->recsize = -1;
1011     else
1012         last->recsize = sizeof(struct journal_subrecord) + bytes;
1013     jrec->last = last;
1014     jrec->residual = bytes;             /* remaining data to be posted */
1015     jrec->residual_align = -bytes & 7;  /* post-data alignment required */
1016     jrec->stream_ptr += sizeof(*last);  /* current write pointer */
1017     jrec->stream_residual -= sizeof(*last); /* space remaining in stream */
1018     return(last);
1019 }
1020
1021 /*
1022  * Write out the data associated with a leaf record.  Any number of calls
1023  * to this routine may be made as long as the byte count adds up to the
1024  * amount originally specified in jrecord_write().
1025  *
1026  * The act of writing out the leaf data may result in numerous stream records
1027  * being pushed out.   Callers should be aware that even the associated
1028  * subrecord header may become inaccessible due to stream record pushouts.
1029  */
1030 void
1031 jrecord_data(struct jrecord *jrec, const void *buf, int bytes)
1032 {
1033     int pusheditout;
1034     int extsize;
1035
1036     KKASSERT(bytes >= 0 && bytes <= jrec->residual);
1037
1038     /*
1039      * Push out stream records as long as there is insufficient room to hold
1040      * the remaining data.
1041      */
1042     while (jrec->stream_residual < bytes) {
1043         /*
1044          * Fill in any remaining space in the current stream record.
1045          */
1046         bcopy(buf, jrec->stream_ptr, jrec->stream_residual);
1047         buf = (const char *)buf + jrec->stream_residual;
1048         bytes -= jrec->stream_residual;
1049         /*jrec->stream_ptr += jrec->stream_residual;*/
1050         jrec->residual -= jrec->stream_residual;
1051         jrec->stream_residual = 0;
1052
1053         /*
1054          * Try to extend the current stream record, but no more then 1/4
1055          * the size of the FIFO.
1056          */
1057         extsize = jrec->jo->fifo.size >> 2;
1058         if (extsize > bytes)
1059             extsize = (bytes + 15) & ~15;
1060
1061         jrec->stream_ptr = journal_extend(jrec->jo, &jrec->rawp,
1062                                 jrec->stream_reserved - jrec->stream_residual,
1063                                 extsize, &pusheditout);
1064         if (pusheditout) {
1065             jrec->stream_reserved = extsize;
1066             jrec->stream_residual = extsize;
1067             jrec->parent = NULL;        /* no longer accessible */
1068             jrec->last = NULL;          /* no longer accessible */
1069             jrec->pushptrgood = 0;      /* restored parents in pops no good */
1070         } else {
1071             jrec->stream_reserved += extsize;
1072             jrec->stream_residual += extsize;
1073         }
1074     }
1075
1076     /*
1077      * Push out any remaining bytes into the current stream record.
1078      */
1079     if (bytes) {
1080         bcopy(buf, jrec->stream_ptr, bytes);
1081         jrec->stream_ptr += bytes;
1082         jrec->stream_residual -= bytes;
1083         jrec->residual -= bytes;
1084     }
1085
1086     /*
1087      * Handle data alignment requirements for the subrecord.  Because the
1088      * stream record's data space is more strictly aligned, it must already
1089      * have sufficient space to hold any subrecord alignment slop.
1090      */
1091     if (jrec->residual == 0 && jrec->residual_align) {
1092         KKASSERT(jrec->residual_align <= jrec->stream_residual);
1093         bzero(jrec->stream_ptr, jrec->residual_align);
1094         jrec->stream_ptr += jrec->residual_align;
1095         jrec->stream_residual -= jrec->residual_align;
1096         jrec->residual_align = 0;
1097     }
1098 }
1099
1100 /*
1101  * We are finished with the transaction.  This closes the transaction created
1102  * by jrecord_init().
1103  *
1104  * NOTE: If abortit is not set then we must be at the top level with no
1105  *       residual subrecord data left to output.
1106  *
1107  *       If abortit is set then we can be in any state, all pushes will be 
1108  *       popped and it is ok for there to be residual data.  This works 
1109  *       because the virtual stream itself is truncated.  Scanners must deal
1110  *       with this situation.
1111  *
1112  * The stream record will be committed or aborted as specified and jrecord
1113  * resources will be cleaned up.
1114  */
1115 void
1116 jrecord_done(struct jrecord *jrec, int abortit)
1117 {
1118     KKASSERT(jrec->rawp != NULL);
1119
1120     if (abortit) {
1121         journal_abort(jrec->jo, &jrec->rawp);
1122     } else {
1123         KKASSERT(jrec->pushcount == 0 && jrec->residual == 0);
1124         journal_commit(jrec->jo, &jrec->rawp, 
1125                         jrec->stream_reserved - jrec->stream_residual, 1);
1126     }
1127
1128     /*
1129      * jrec should not be used beyond this point without another init,
1130      * but clean up some fields to ensure that we panic if it is.
1131      *
1132      * Note that jrec->rawp is NULLd out by journal_abort/journal_commit.
1133      */
1134     jrec->jo = NULL;
1135     jrec->stream_ptr = NULL;
1136 }
1137
1138 /************************************************************************
1139  *                      LOW LEVEL RECORD SUPPORT ROUTINES               *
1140  ************************************************************************
1141  *
1142  * These routine create low level recursive and leaf subrecords representing
1143  * common filesystem structures.
1144  */
1145
1146 /*
1147  * Write out a filename path relative to the base of the mount point.
1148  * rectype is typically JLEAF_PATH{1,2,3,4}.
1149  */
1150 void
1151 jrecord_write_path(struct jrecord *jrec, int16_t rectype, struct namecache *ncp)
1152 {
1153     char buf[64];       /* local buffer if it fits, else malloced */
1154     char *base;
1155     int pathlen;
1156     int index;
1157     struct namecache *scan;
1158
1159     /*
1160      * Pass 1 - figure out the number of bytes required.  Include terminating
1161      *         \0 on last element and '/' separator on other elements.
1162      *
1163      * The namecache topology terminates at the root of the filesystem
1164      * (the normal lookup code would then continue by using the mount
1165      * structure to figure out what it was mounted on).
1166      */
1167 again:
1168     pathlen = 0;
1169     for (scan = ncp; scan; scan = scan->nc_parent) {
1170         if (scan->nc_nlen > 0)
1171             pathlen += scan->nc_nlen + 1;
1172     }
1173
1174     if (pathlen <= sizeof(buf))
1175         base = buf;
1176     else
1177         base = kmalloc(pathlen, M_TEMP, M_INTWAIT);
1178
1179     /*
1180      * Pass 2 - generate the path buffer
1181      */
1182     index = pathlen;
1183     for (scan = ncp; scan; scan = scan->nc_parent) {
1184         if (scan->nc_nlen == 0)
1185             continue;
1186         if (scan->nc_nlen >= index) {
1187             if (base != buf)
1188                 kfree(base, M_TEMP);
1189             goto again;
1190         }
1191         if (index == pathlen)
1192             base[--index] = 0;
1193         else
1194             base[--index] = '/';
1195         index -= scan->nc_nlen;
1196         bcopy(scan->nc_name, base + index, scan->nc_nlen);
1197     }
1198     jrecord_leaf(jrec, rectype, base + index, pathlen - index);
1199     if (base != buf)
1200         kfree(base, M_TEMP);
1201 }
1202
1203 /*
1204  * Write out a file attribute structure.  While somewhat inefficient, using
1205  * a recursive data structure is the most portable and extensible way.
1206  */
1207 void
1208 jrecord_write_vattr(struct jrecord *jrec, struct vattr *vat)
1209 {
1210     void *save;
1211
1212     save = jrecord_push(jrec, JTYPE_VATTR);
1213     if (vat->va_type != VNON)
1214         jrecord_leaf(jrec, JLEAF_VTYPE, &vat->va_type, sizeof(vat->va_type));
1215     if (vat->va_mode != (mode_t)VNOVAL)
1216         jrecord_leaf(jrec, JLEAF_MODES, &vat->va_mode, sizeof(vat->va_mode));
1217     if (vat->va_nlink != VNOVAL)
1218         jrecord_leaf(jrec, JLEAF_NLINK, &vat->va_nlink, sizeof(vat->va_nlink));
1219     if (vat->va_uid != VNOVAL)
1220         jrecord_leaf(jrec, JLEAF_UID, &vat->va_uid, sizeof(vat->va_uid));
1221     if (vat->va_gid != VNOVAL)
1222         jrecord_leaf(jrec, JLEAF_GID, &vat->va_gid, sizeof(vat->va_gid));
1223     if (vat->va_fsid != VNOVAL)
1224         jrecord_leaf(jrec, JLEAF_FSID, &vat->va_fsid, sizeof(vat->va_fsid));
1225     if (vat->va_fileid != VNOVAL)
1226         jrecord_leaf(jrec, JLEAF_INUM, &vat->va_fileid, sizeof(vat->va_fileid));
1227     if (vat->va_size != VNOVAL)
1228         jrecord_leaf(jrec, JLEAF_SIZE, &vat->va_size, sizeof(vat->va_size));
1229     if (vat->va_atime.tv_sec != VNOVAL)
1230         jrecord_leaf(jrec, JLEAF_ATIME, &vat->va_atime, sizeof(vat->va_atime));
1231     if (vat->va_mtime.tv_sec != VNOVAL)
1232         jrecord_leaf(jrec, JLEAF_MTIME, &vat->va_mtime, sizeof(vat->va_mtime));
1233     if (vat->va_ctime.tv_sec != VNOVAL)
1234         jrecord_leaf(jrec, JLEAF_CTIME, &vat->va_ctime, sizeof(vat->va_ctime));
1235     if (vat->va_gen != VNOVAL)
1236         jrecord_leaf(jrec, JLEAF_GEN, &vat->va_gen, sizeof(vat->va_gen));
1237     if (vat->va_flags != VNOVAL)
1238         jrecord_leaf(jrec, JLEAF_FLAGS, &vat->va_flags, sizeof(vat->va_flags));
1239     if (vat->va_rmajor != VNOVAL) {
1240         udev_t rdev = makeudev(vat->va_rmajor, vat->va_rminor);
1241         jrecord_leaf(jrec, JLEAF_UDEV, &rdev, sizeof(rdev));
1242         jrecord_leaf(jrec, JLEAF_UMAJOR, &vat->va_rmajor, sizeof(vat->va_rmajor));
1243         jrecord_leaf(jrec, JLEAF_UMINOR, &vat->va_rminor, sizeof(vat->va_rminor));
1244     }
1245 #if 0
1246     if (vat->va_filerev != VNOVAL)
1247         jrecord_leaf(jrec, JLEAF_FILEREV, &vat->va_filerev, sizeof(vat->va_filerev));
1248 #endif
1249     jrecord_pop(jrec, save);
1250 }
1251
1252 /*
1253  * Write out the creds used to issue a file operation.  If a process is
1254  * available write out additional tracking information related to the 
1255  * process.
1256  *
1257  * XXX additional tracking info
1258  * XXX tty line info
1259  */
1260 void
1261 jrecord_write_cred(struct jrecord *jrec, struct thread *td, struct ucred *cred)
1262 {
1263     void *save;
1264     struct proc *p;
1265
1266     save = jrecord_push(jrec, JTYPE_CRED);
1267     jrecord_leaf(jrec, JLEAF_UID, &cred->cr_uid, sizeof(cred->cr_uid));
1268     jrecord_leaf(jrec, JLEAF_GID, &cred->cr_gid, sizeof(cred->cr_gid));
1269     if (td && (p = td->td_proc) != NULL) {
1270         jrecord_leaf(jrec, JLEAF_PID, &p->p_pid, sizeof(p->p_pid));
1271         jrecord_leaf(jrec, JLEAF_COMM, p->p_comm, sizeof(p->p_comm));
1272     }
1273     jrecord_pop(jrec, save);
1274 }
1275
1276 /*
1277  * Write out information required to identify a vnode
1278  *
1279  * XXX this needs work.  We should write out the inode number as well,
1280  * and in fact avoid writing out the file path for seqential writes
1281  * occuring within e.g. a certain period of time.
1282  */
1283 void
1284 jrecord_write_vnode_ref(struct jrecord *jrec, struct vnode *vp)
1285 {
1286     struct namecache *ncp;
1287
1288     TAILQ_FOREACH(ncp, &vp->v_namecache, nc_vnode) {
1289         if ((ncp->nc_flag & (NCF_UNRESOLVED|NCF_DESTROYED)) == 0)
1290             break;
1291     }
1292     if (ncp)
1293         jrecord_write_path(jrec, JLEAF_PATH_REF, ncp);
1294 }
1295
1296 void
1297 jrecord_write_vnode_link(struct jrecord *jrec, struct vnode *vp, 
1298                          struct namecache *notncp)
1299 {
1300     struct namecache *ncp;
1301
1302     TAILQ_FOREACH(ncp, &vp->v_namecache, nc_vnode) {
1303         if (ncp == notncp)
1304             continue;
1305         if ((ncp->nc_flag & (NCF_UNRESOLVED|NCF_DESTROYED)) == 0)
1306             break;
1307     }
1308     if (ncp)
1309         jrecord_write_path(jrec, JLEAF_PATH_REF, ncp);
1310 }
1311
1312 /*
1313  * Write out the data represented by a pagelist
1314  */
1315 void
1316 jrecord_write_pagelist(struct jrecord *jrec, int16_t rectype,
1317                         struct vm_page **pglist, int *rtvals, int pgcount,
1318                         off_t offset)
1319 {
1320     struct msf_buf *msf;
1321     int error;
1322     int b;
1323     int i;
1324
1325     i = 0;
1326     while (i < pgcount) {
1327         /*
1328          * Find the next valid section.  Skip any invalid elements
1329          */
1330         if (rtvals[i] != VM_PAGER_OK) {
1331             ++i;
1332             offset += PAGE_SIZE;
1333             continue;
1334         }
1335
1336         /*
1337          * Figure out how big the valid section is, capping I/O at what the
1338          * MSFBUF can represent.
1339          */
1340         b = i;
1341         while (i < pgcount && i - b != XIO_INTERNAL_PAGES && 
1342                rtvals[i] == VM_PAGER_OK
1343         ) {
1344             ++i;
1345         }
1346
1347         /*
1348          * And write it out.
1349          */
1350         if (i - b) {
1351             error = msf_map_pagelist(&msf, pglist + b, i - b, 0);
1352             if (error == 0) {
1353                 kprintf("RECORD PUTPAGES %d\n", msf_buf_bytes(msf));
1354                 jrecord_leaf(jrec, JLEAF_SEEKPOS, &offset, sizeof(offset));
1355                 jrecord_leaf(jrec, rectype, 
1356                              msf_buf_kva(msf), msf_buf_bytes(msf));
1357                 msf_buf_free(msf);
1358             } else {
1359                 kprintf("jrecord_write_pagelist: mapping failure\n");
1360             }
1361             offset += (off_t)(i - b) << PAGE_SHIFT;
1362         }
1363     }
1364 }
1365
1366 /*
1367  * Write out the data represented by a UIO.
1368  */
1369 struct jwuio_info {
1370     struct jrecord *jrec;
1371     int16_t rectype;
1372 };
1373
1374 static int jrecord_write_uio_callback(void *info, char *buf, int bytes);
1375
1376 void
1377 jrecord_write_uio(struct jrecord *jrec, int16_t rectype, struct uio *uio)
1378 {
1379     struct jwuio_info info = { jrec, rectype };
1380     int error;
1381
1382     if (uio->uio_segflg != UIO_NOCOPY) {
1383         jrecord_leaf(jrec, JLEAF_SEEKPOS, &uio->uio_offset, 
1384                      sizeof(uio->uio_offset));
1385         error = msf_uio_iterate(uio, jrecord_write_uio_callback, &info);
1386         if (error)
1387             kprintf("XXX warning uio iterate failed %d\n", error);
1388     }
1389 }
1390
1391 static int
1392 jrecord_write_uio_callback(void *info_arg, char *buf, int bytes)
1393 {
1394     struct jwuio_info *info = info_arg;
1395
1396     jrecord_leaf(info->jrec, info->rectype, buf, bytes);
1397     return(0);
1398 }
1399
1400 void
1401 jrecord_file_data(struct jrecord *jrec, struct vnode *vp, 
1402                   off_t off, off_t bytes)
1403 {
1404     const int bufsize = 8192;
1405     char *buf;
1406     int error;
1407     int n;
1408
1409     buf = kmalloc(bufsize, M_JOURNAL, M_WAITOK);
1410     jrecord_leaf(jrec, JLEAF_SEEKPOS, &off, sizeof(off));
1411     while (bytes) {
1412         n = (bytes > bufsize) ? bufsize : (int)bytes;
1413         error = vn_rdwr(UIO_READ, vp, buf, n, off, UIO_SYSSPACE, IO_NODELOCKED,
1414                         proc0.p_ucred, NULL);
1415         if (error) {
1416             jrecord_leaf(jrec, JLEAF_ERROR, &error, sizeof(error));
1417             break;
1418         }
1419         jrecord_leaf(jrec, JLEAF_FILEDATA, buf, n);
1420         bytes -= n;
1421         off += n;
1422     }
1423     kfree(buf, M_JOURNAL);
1424 }
1425