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.5 2005/09/06 06:42:44 dillon Exp $
  35  */
  36
  37 #include "jscan.h"
  38
  39 static struct jhash     *JHashAry[JHASH_SIZE];
  40
  41 static void jnormalize(struct jstream *js);
  42
  43 /*
  44  * Integrate a raw record.  Deal with the transaction begin and end flags
  45  * to create a forward-referenced collection of jstream records.  If we are
  46  * able to complete a transaction, the first js associated with that
  47  * transaction is returned.
  48  *
  49  * XXX we need to store the data for very large multi-record transactions
  50  * separately since it might not fit into memory.
  51  */
  52 struct jstream *
  53 jaddrecord(struct jfile *jf, struct jdata *jd)
  54 {
  55     struct journal_rawrecbeg *head;
  56     struct jstream *js;
  57     struct jhash *jh;
  58     struct jhash **jhp;
  59
  60     js = malloc(sizeof(struct jstream));
  61     bzero(js, sizeof(struct jstream));
  62     js->js_jdata = jref(jd);
  63     js->js_head = (void *)jd->jd_data;
  64     js->js_jfile = jf;
  65     head = js->js_head;
  66
  67     /*
  68      * Check for a completely self-contained transaction, just return the
  69      * js if possible.
  70      */
  71     if ((head->streamid & (JREC_STREAMCTL_BEGIN|JREC_STREAMCTL_END)) ==
  72         (JREC_STREAMCTL_BEGIN|JREC_STREAMCTL_END)
  73     ) {
  74         jnormalize(js);
  75         return (js);
  76     }
  77
  78     /*
  79      * Check for an open transaction in the hash table, create a new one
  80      * if necessary.
  81      */
  82     jhp = &JHashAry[head->streamid & JHASH_MASK];
  83     while ((jh = *jhp) != NULL) {
  84         if (((jh->jh_transid ^ head->streamid) & JREC_STREAMID_MASK) == 0)
  85             break;
  86         jhp = &jh->jh_hash;
  87     }
  88     if (jh == NULL) {
  89         jh = malloc(sizeof(*jh));
  90         bzero(jh, sizeof(*jh));
  91         *jhp = jh;
  92         jh->jh_first = js;
  93         jh->jh_last = js;
  94         jh->jh_transid = head->streamid;
  95         return (NULL);
  96     }
  97
  98     /*
  99      * Emplace the stream segment
 100      */
 101     jh->jh_transid |= head->streamid & JREC_STREAMCTL_MASK;
 102     if (jf->jf_direction == JD_FORWARDS) {
 103         jh->jh_last->js_next = js;
 104         jh->jh_last = js;
 105     } else {
 106         js->js_next = jh->jh_first;
 107         jh->jh_first = js;
 108     }
 109
 110     /*
 111      * If the transaction is complete, remove the hash entry and return the
 112      * js representing the beginning of the transaction.
 113      */
 114     if ((jh->jh_transid & (JREC_STREAMCTL_BEGIN|JREC_STREAMCTL_END)) ==
 115         (JREC_STREAMCTL_BEGIN|JREC_STREAMCTL_END)
 116     ) {
 117         *jhp = jh->jh_hash;
 118         js = jh->jh_first;
 119         free(jh);
 120
 121         jnormalize(js);
 122     } else {
 123         js = NULL;
 124     }
 125     return (js);
 126 }
 127
 128 /*
 129  * Renormalize the jscan list to remove all the meta record headers
 130  * and trailers except for the very first one.
 131  */
 132 static
 133 void
 134 jnormalize(struct jstream *js)
 135 {
 136     struct jstream *jscan;
 137     off_t off;
 138
 139     js->js_normalized_off = 0;
 140     js->js_normalized_base = (void *)js->js_head;
 141     js->js_normalized_size = js->js_head->recsize - sizeof(struct journal_rawrecend);
 142     js->js_normalized_total = js->js_normalized_size;
 143     off = js->js_normalized_size;
 144     for (jscan = js->js_next; jscan; jscan = jscan->js_next) {
 145         jscan->js_normalized_off = off;
 146         jscan->js_normalized_base = (char *)jscan->js_head +
 147                 sizeof(struct journal_rawrecbeg);
 148         jscan->js_normalized_size = jscan->js_head->recsize -
 149                sizeof(struct journal_rawrecbeg) -
 150                sizeof(struct journal_rawrecend);
 151         off += jscan->js_normalized_size;
 152         js->js_normalized_total += jscan->js_normalized_size;
 153     }
 154 }
 155
 156 void
 157 jscan_dispose(struct jstream *js)
 158 {
 159     struct jstream *jnext;
 160
 161     if (js->js_alloc_buf) {
 162         free(js->js_alloc_buf);
 163         js->js_alloc_buf = NULL;
 164         js->js_alloc_size = 0;
 165     }
 166
 167     while (js) {
 168         jnext = js->js_next;
 169         jfree(js->js_jfile, js->js_jdata);
 170         js->js_jdata = NULL;
 171         free(js);
 172         js = jnext;
 173     }
 174 }
 175
 176 /*
 177  * Read the specified block of data out of a linked set of jstream
 178  * structures.  Returns 0 on success or an error code on error.
 179  */
 180 int
 181 jsread(struct jstream *js, off_t off, void *buf, int bytes)
 182 {
 183     const void *ptr;
 184     int n;
 185
 186     while (bytes) {
 187         n = jsreadany(js, off, &ptr);
 188         if (n == 0)
 189             return (ENOENT);
 190         if (n > bytes)
 191             n = bytes;
 192         bcopy(ptr, buf, n);
 193         buf = (char *)buf + n;
 194         off += n;
 195         bytes -= n;
 196     }
 197     return(0);
 198 }
 199
 200 /*
 201  * Read the specified block of data out of a linked set of jstream
 202  * structures.  Attempt to return a pointer into the data set but
 203  * allocate and copy if that is not possible.  Returns 0 on success
 204  * or an error code on error.
 205  */
 206 int
 207 jsreadp(struct jstream *js, off_t off, const void **bufp,
 208         int bytes)
 209 {
 210     int error = 0;
 211     int n;
 212
 213     n = jsreadany(js, off, bufp);
 214     if (n < bytes) {
 215         if (js->js_alloc_size < bytes) {
 216             if (js->js_alloc_buf)
 217                 free(js->js_alloc_buf);
 218             js->js_alloc_buf = malloc(bytes);
 219             js->js_alloc_size = bytes;
 220             assert(js->js_alloc_buf != NULL);
 221         }
 222         error = jsread(js, off, js->js_alloc_buf, bytes);
 223         if (error) {
 224             *bufp = NULL;
 225         } else {
 226             *bufp = js->js_alloc_buf;
 227         }
 228     }
 229     return(error);
 230 }
 231
 232 int
 233 jsreadcallback(struct jstream *js, ssize_t (*func)(int, const void *, size_t),
 234                 int fd, off_t off, int bytes)
 235 {
 236     const void *bufp;
 237     int res;
 238     int n;
 239     int r;
 240
 241     res = 0;
 242     while (bytes && (n = jsreadany(js, off, &bufp)) > 0) {
 243         if (n > bytes)
 244             n = bytes;
 245         r = func(fd, bufp, n);
 246         if (r != n) {
 247             if (res == 0)
 248                 res = -1;
 249         }
 250         res += n;
 251         bytes -= n;
 252         off += n;
 253     }
 254     return(res);
 255 }
 256
 257 /*
 258  * Return the largest contiguous buffer starting at the specified offset,
 259  * or 0.
 260  */
 261 int
 262 jsreadany(struct jstream *js, off_t off, const void **bufp)
 263 {
 264     struct jstream *scan;
 265     int n;
 266
 267     if ((scan = js->js_cache) == NULL || scan->js_normalized_off > off)
 268         scan = js;
 269     while (scan && scan->js_normalized_off <= off) {
 270         js->js_cache = scan;
 271         if (scan->js_normalized_off + scan->js_normalized_size > off) {
 272             n = (int)(off - scan->js_normalized_off);
 273             *bufp = scan->js_normalized_base + n;
 274             return(scan->js_normalized_size - n);
 275         }
 276         scan = scan->js_next;
 277     }
 278     return(0);
 279 }
 280