sbin/jscan/jstream.c

   1 /*
   2  * Copyright (c) 2004,2005 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/sbin/jscan/jstream.c,v 1.3 2005/07/05 02:38:34 dillon Exp $
  35  */
  36
  37 #include "jscan.h"
  38
  39 static struct jhash     *JHashAry[JHASH_SIZE];
  40
  41 static struct jstream *jaddrecord(struct jfile *jf, struct jstream *js);
  42 static void jnormalize(struct jstream *js);
  43
  44 /*
  45  * Locate the next (or previous) complete virtual stream transaction given a
  46  * file descriptor and direction.  Keep track of partial stream records as
  47  * a side effect.
  48  *
  49  * Note that a transaction might represent a huge I/O operation, resulting
  50  * in an overall node structure that spans gigabytes, but individual
  51  * subrecord leaf nodes are limited in size and we depend on this to simplify
  52  * the handling of leaf records.
  53  *
  54  * A transaction may cover several raw records.  The jstream collection for
  55  * a transaction is only returned when the entire transaction has been
  56  * successfully scanned.  Due to the interleaving of transactions the ordering
  57  * of returned JS's may be different (not exactly reversed) when scanning a
  58  * journal backwards verses forwards.  Since parallel operations are
  59  * theoretically non-conflicting, this should not present a problem.
  60  */
  61 struct jstream *
  62 jscan_stream(struct jfile *jf)
  63 {
  64     struct journal_rawrecbeg head;
  65     struct journal_rawrecend tail;
  66     int recsize;
  67     int search;
  68     int error;
  69     struct jstream *js;
  70
  71     /*
  72      * Get the current offset and make sure it is 16-byte aligned.  If it
  73      * isn't, align it and enter search mode.
  74      */
  75     if (jf->jf_pos & 15) {
  76         jf_warn(jf, "realigning bad offset and entering search mode");
  77         jalign(jf);
  78         search = 1;
  79     } else {
  80         search = 0;
  81     }
  82
  83     error = 0;
  84     js = NULL;
  85
  86     if (jf->jf_direction == JF_FORWARDS) {
  87         /*
  88          * Scan the journal forwards.  Note that the file pointer might not
  89          * be seekable.
  90          */
  91         while ((error = jread(jf, &head, sizeof(head))) == 0) {
  92             if (head.begmagic != JREC_BEGMAGIC) {
  93                 if (search == 0)
  94                     jf_warn(jf, "bad beginmagic, searching for new record");
  95                 search = 1;
  96                 jalign(jf);
  97                 continue;
  98             }
  99             recsize = (head.recsize + 15) & ~15;
 100             if (recsize <= 0) {
 101                 jf_warn(jf, "bad recordsize: %d\n", recsize);
 102                 search = 1;
 103                 jalign(jf);
 104                 continue;
 105             }
 106             jset(jf);
 107             js = malloc(offsetof(struct jstream, js_data[recsize]));
 108             bzero(js, sizeof(struct jstream));
 109             bcopy(&head, js->js_data, sizeof(head));
 110             error = jread(jf, js->js_data + sizeof(head), recsize - sizeof(head));
 111             if (error) {
 112                 jf_warn(jf, "Incomplete stream record\n");
 113                 jreturn(jf);
 114                 free(js);
 115                 js = NULL;
 116                 break;
 117             }
 118
 119             /*
 120              * note: recsize is aligned (the actual record size),
 121              * head.recsize is unaligned (the actual payload size).
 122              */
 123             js->js_size = head.recsize;
 124             bcopy(js->js_data + recsize - sizeof(tail), &tail, sizeof(tail));
 125             if (tail.endmagic != JREC_ENDMAGIC) {
 126                 jf_warn(jf, "bad endmagic, searching for new record");
 127                 search = 1;
 128                 jreturn(jf);
 129                 free(js);
 130                 js = NULL;
 131                 continue;
 132             }
 133             jflush(jf);
 134             if ((js = jaddrecord(jf, js)) != NULL)
 135                 break;
 136         }
 137     } else {
 138         /*
 139          * Scan the journal backwards.  Note that jread()'s reverse-seek and
 140          * read.  The data read will be forward ordered, however.
 141          */
 142         while ((error = jread(jf, &tail, sizeof(tail))) == 0) {
 143             if (tail.endmagic != JREC_ENDMAGIC) {
 144                 if (search == 0)
 145                     jf_warn(jf, "bad endmagic, searching for new record");
 146                 search = 1;
 147                 jalign(jf);
 148                 continue;
 149             }
 150             recsize = (tail.recsize + 15) & ~15;
 151             if (recsize <= 0) {
 152                 jf_warn(jf, "bad recordsize: %d\n", recsize);
 153                 search = 1;
 154                 jalign(jf);
 155                 continue;
 156             }
 157             jset(jf);
 158             js = malloc(offsetof(struct jstream, js_data[recsize]));
 159             bzero(js, sizeof(struct jstream));
 160             bcopy(&tail, js->js_data + recsize - sizeof(tail), sizeof(tail));
 161             error = jread(jf, js->js_data, recsize - sizeof(tail));
 162
 163             if (error) {
 164                 jf_warn(jf, "Incomplete stream record\n");
 165                 jreturn(jf);
 166                 free(js);
 167                 js = NULL;
 168                 break;
 169             }
 170             js->js_size = tail.recsize;
 171             bcopy(js->js_data + recsize - sizeof(tail), &tail, sizeof(tail));
 172             bcopy(js->js_data, &head, sizeof(head));
 173             if (head.begmagic != JREC_BEGMAGIC) {
 174                 jf_warn(jf, "bad begmagic, searching for new record");
 175                 search = 1;
 176                 jreturn(jf);
 177                 free(js);
 178                 continue;
 179             }
 180             jflush(jf);
 181             if ((js = jaddrecord(jf, js)) != NULL)
 182                 break;
 183         }
 184     }
 185     jf->jf_error = error;
 186     return(js);
 187 }
 188
 189 /*
 190  * Integrate a jstream record.  Deal with the transaction begin and end flags
 191  * to create a forward-referenced collection of jstream records.  If we are
 192  * able to complete a transaction, the first js associated with that
 193  * transaction is returned.
 194  *
 195  * XXX we need to store the data for very large multi-record transactions
 196  * separately since it might not fit into memory.
 197  */
 198 static struct jstream *
 199 jaddrecord(struct jfile *jf, struct jstream *js)
 200 {
 201     struct journal_rawrecbeg *head = (void *)js->js_data;
 202     struct jhash *jh;
 203     struct jhash **jhp;
 204
 205     /*
 206      * Check for a completely self-contained transaction, just return the
 207      * js if possible.
 208      */
 209     if ((head->streamid & (JREC_STREAMCTL_BEGIN|JREC_STREAMCTL_END)) ==
 210         (JREC_STREAMCTL_BEGIN|JREC_STREAMCTL_END)
 211     ) {
 212         jnormalize(js);
 213         return (js);
 214     }
 215
 216     /*
 217      * Check for an open transaction in the hash table, create a new one
 218      * if necessary.
 219      */
 220     jhp = &JHashAry[head->streamid & JHASH_MASK];
 221     while ((jh = *jhp) != NULL) {
 222         if (((jh->jh_transid ^ head->streamid) & JREC_STREAMID_MASK) == 0)
 223             break;
 224         jhp = &jh->jh_hash;
 225     }
 226     if (jh == NULL) {
 227         jh = malloc(sizeof(*jh));
 228         bzero(jh, sizeof(*jh));
 229         *jhp = jh;
 230         jh->jh_first = js;
 231         jh->jh_last = js;
 232         jh->jh_transid = head->streamid;
 233         return (NULL);
 234     }
 235
 236     /*
 237      * Emplace the stream segment
 238      */
 239     jh->jh_transid |= head->streamid & JREC_STREAMCTL_MASK;
 240     if (jf->jf_direction == JF_FORWARDS) {
 241         jh->jh_last->js_next = js;
 242         jh->jh_last = js;
 243     } else {
 244         js->js_next = jh->jh_first;
 245         jh->jh_first = js;
 246     }
 247
 248     /*
 249      * If the transaction is complete, remove the hash entry and return the
 250      * js representing the beginning of the transaction.
 251      */
 252     if ((jh->jh_transid & (JREC_STREAMCTL_BEGIN|JREC_STREAMCTL_END)) ==
 253         (JREC_STREAMCTL_BEGIN|JREC_STREAMCTL_END)
 254     ) {
 255         *jhp = jh->jh_hash;
 256         js = jh->jh_first;
 257         free(jh);
 258
 259         jnormalize(js);
 260     } else {
 261         js = NULL;
 262     }
 263     return (js);
 264 }
 265
 266 /*
 267  * Renormalize the jscan list to remove all the meta record headers
 268  * and trailers except for the very first one.
 269  */
 270 static
 271 void
 272 jnormalize(struct jstream *js)
 273 {
 274     struct jstream *jscan;
 275     off_t off;
 276
 277     js->js_normalized_off = 0;
 278     js->js_normalized_base = js->js_data;
 279     js->js_normalized_size = ((struct journal_rawrecbeg *)js->js_data)->recsize - sizeof(struct journal_rawrecend);
 280     js->js_normalized_total = js->js_normalized_size;
 281     off = js->js_normalized_size;
 282     for (jscan = js->js_next; jscan; jscan = jscan->js_next) {
 283         jscan->js_normalized_off = off;
 284         jscan->js_normalized_base = jscan->js_data +
 285                 sizeof(struct journal_rawrecbeg);
 286         jscan->js_normalized_size = jscan->js_size -
 287                sizeof(struct journal_rawrecbeg) -
 288                sizeof(struct journal_rawrecend);
 289         off += jscan->js_normalized_size;
 290         js->js_normalized_total += jscan->js_normalized_size;
 291     }
 292 }
 293
 294 void
 295 jscan_dispose(struct jstream *js)
 296 {
 297     struct jstream *jnext;
 298
 299     if (js->js_alloc_buf) {
 300         free(js->js_alloc_buf);
 301         js->js_alloc_buf = NULL;
 302         js->js_alloc_size = 0;
 303     }
 304
 305     while (js) {
 306         jnext = js->js_next;
 307         free(js);
 308         js = jnext;
 309     }
 310 }
 311
 312 /*
 313  * Read the specified block of data out of a linked set of jstream
 314  * structures.  Returns 0 on success or an error code on error.
 315  */
 316 int
 317 jsread(struct jstream *js, off_t off, void *buf, int bytes)
 318 {
 319     const void *ptr;
 320     int n;
 321
 322     while (bytes) {
 323         n = jsreadany(js, off, &ptr);
 324         if (n == 0)
 325             return (ENOENT);
 326         if (n > bytes)
 327             n = bytes;
 328         bcopy(ptr, buf, n);
 329         buf = (char *)buf + n;
 330         off += n;
 331         bytes -= n;
 332     }
 333     return(0);
 334 }
 335
 336 /*
 337  * Read the specified block of data out of a linked set of jstream
 338  * structures.  Attempt to return a pointer into the data set but
 339  * allocate and copy if that is not possible.  Returns 0 on success
 340  * or an error code on error.
 341  */
 342 int
 343 jsreadp(struct jstream *js, off_t off, const void **bufp,
 344         int bytes)
 345 {
 346     int error = 0;
 347     int n;
 348
 349     n = jsreadany(js, off, bufp);
 350     if (n < bytes) {
 351         if (js->js_alloc_size < bytes) {
 352             if (js->js_alloc_buf)
 353                 free(js->js_alloc_buf);
 354             js->js_alloc_buf = malloc(bytes);
 355             js->js_alloc_size = bytes;
 356             assert(js->js_alloc_buf != NULL);
 357         }
 358         error = jsread(js, off, js->js_alloc_buf, bytes);
 359         if (error) {
 360             *bufp = NULL;
 361         } else {
 362             *bufp = js->js_alloc_buf;
 363         }
 364     }
 365     return(error);
 366 }
 367
 368 int
 369 jsreadcallback(struct jstream *js, ssize_t (*func)(int, const void *, size_t),
 370                 int fd, off_t off, int bytes)
 371 {
 372     const void *bufp;
 373     int res;
 374     int n;
 375     int r;
 376
 377     res = 0;
 378     while (bytes && (n = jsreadany(js, off, &bufp)) > 0) {
 379         if (n > bytes)
 380             n = bytes;
 381         r = func(fd, bufp, n);
 382         if (r != n) {
 383             if (res == 0)
 384                 res = -1;
 385         }
 386         res += n;
 387         bytes -= n;
 388         off += n;
 389     }
 390     return(res);
 391 }
 392
 393 /*
 394  * Return the largest contiguous buffer starting at the specified offset,
 395  * or 0.
 396  */
 397 int
 398 jsreadany(struct jstream *js, off_t off, const void **bufp)
 399 {
 400     struct jstream *scan;
 401     int n;
 402
 403     if ((scan = js->js_cache) == NULL || scan->js_normalized_off > off)
 404         scan = js;
 405     while (scan && scan->js_normalized_off <= off) {
 406         js->js_cache = scan;
 407         if (scan->js_normalized_off + scan->js_normalized_size > off) {
 408             n = (int)(off - scan->js_normalized_off);
 409             *bufp = scan->js_normalized_base + n;
 410             return(scan->js_normalized_size - n);
 411         }
 412         scan = scan->js_next;
 413     }
 414     return(0);
 415 }
 416