lib/bucket.c

   1 /*
   2  *      Sherlock Library -- Object Buckets
   3  *
   4  *      (c) 2001 Martin Mares <mj@ucw.cz>
   5  */
   6
   7 #include "lib/lib.h"
   8 #include "lib/bucket.h"
   9 #include "lib/fastbuf.h"
  10 #include "lib/lfs.h"
  11
  12 #include <string.h>
  13 #include <fcntl.h>
  14 #include <unistd.h>
  15 #include <sys/file.h>
  16
  17 static int obuck_fd;
  18 static unsigned int obuck_remains, obuck_check_pad;
  19 static struct fastbuf *obuck_fb;
  20 static struct obuck_header obuck_hdr;
  21 static sh_off_t bucket_start;
  22 byte *obuck_name = "db/objects";                /* FIXME */
  23
  24 /*** Internal operations ***/
  25
  26 static void
  27 obuck_broken(char *msg)
  28 {
  29   die("Object pool corrupted: %s (pos=%Lx)", msg, (long long) bucket_start);    /* FIXME */
  30 }
  31
  32 static inline void
  33 obuck_lock_read(void)
  34 {
  35   flock(obuck_fd, LOCK_SH);
  36 }
  37
  38 static inline void
  39 obuck_lock_write(void)
  40 {
  41   flock(obuck_fd, LOCK_EX);
  42 }
  43
  44 static inline void
  45 obuck_unlock(void)
  46 {
  47   flock(obuck_fd, LOCK_UN);
  48 }
  49
  50 /*** FastIO emulation ***/
  51
  52 /* We need to use pread/pwrite since we work on fd's shared between processes */
  53
  54 static int
  55 obuck_fb_refill(struct fastbuf *f)
  56 {
  57   unsigned limit = (f->buflen < obuck_remains) ? f->buflen : obuck_remains;
  58   unsigned size = (limit == obuck_remains) ? (limit+obuck_check_pad+4) : limit;
  59   int l;
  60
  61   if (!limit)
  62     return 0;
  63   l = sh_pread(f->fd, f->buffer, size, f->fdpos);
  64   if (l < 0)
  65     die("Error reading bucket: %m");
  66   if ((unsigned) l != size)
  67     obuck_broken("Short read");
  68   f->bptr = f->buffer;
  69   f->bstop = f->buffer + limit;
  70   f->pos = f->fdpos;
  71   f->fdpos += limit;
  72   obuck_remains -= limit;
  73   if (!obuck_remains)   /* Should check the trailer */
  74     {
  75       u32 check;
  76       memcpy(&check, f->buffer + size - 4, 4);
  77       if (check != OBUCK_TRAILER)
  78         obuck_broken("Missing trailer");
  79     }
  80   return limit;
  81 }
  82
  83 static void
  84 obuck_fb_spout(struct fastbuf *f)
  85 {
  86   int l = f->bptr - f->buffer;
  87   char *c = f->buffer;
  88
  89   while (l)
  90     {
  91       int z = sh_pwrite(f->fd, c, l, f->fdpos);
  92       if (z <= 0)
  93         die("Error writing bucket: %m");
  94       f->fdpos += z;
  95       l -= z;
  96       c += z;
  97     }
  98   f->bptr = f->buffer;
  99   f->pos = f->fdpos;
 100 }
 101
 102 static void
 103 obuck_fb_close(struct fastbuf *f)
 104 {
 105   close(f->fd);
 106 }
 107
 108 /*** Exported functions ***/
 109
 110 void
 111 obuck_init(int writeable)
 112 {
 113   struct fastbuf *b;
 114   int buflen = 65536;
 115   sh_off_t size;
 116
 117   obuck_fd = open(obuck_name, (writeable ? O_RDWR | O_CREAT : O_RDONLY), 0666);
 118   obuck_fb = b = xmalloc(sizeof(struct fastbuf) + buflen + OBUCK_ALIGN + 4);
 119   bzero(b, sizeof(struct fastbuf));
 120   b->buflen = buflen;
 121   b->buffer = (char *)(b+1);
 122   b->bptr = b->bstop = b->buffer;
 123   b->bufend = b->buffer + buflen;
 124   b->name = "bucket";
 125   b->fd = obuck_fd;
 126   b->refill = obuck_fb_refill;
 127   b->spout = obuck_fb_spout;
 128   b->close = obuck_fb_close;
 129   obuck_lock_read();
 130   size = sh_seek(obuck_fd, 0, SEEK_END);
 131   if (size)
 132     {
 133       /* If the bucket pool is not empty, check consistency of its end */
 134       u32 check;
 135       bucket_start = size - 4;  /* for error reporting */
 136       if (sh_pread(obuck_fd, &check, 4, size-4) != 4 ||
 137           check != OBUCK_TRAILER)
 138         obuck_broken("Missing trailer of last object");
 139     }
 140   obuck_unlock();
 141 }
 142
 143 void
 144 obuck_cleanup(void)
 145 {
 146   bclose(obuck_fb);
 147 }
 148
 149 void                                    /* FIXME: Call somewhere :) */
 150 obuck_sync(void)
 151 {
 152   bflush(obuck_fb);
 153   fsync(obuck_fd);
 154 }
 155
 156 static void
 157 obuck_get(oid_t oid)
 158 {
 159   struct fastbuf *b = obuck_fb;
 160
 161   bucket_start = ((sh_off_t) oid) << OBUCK_SHIFT;
 162   bflush(b);
 163   if (sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start) != sizeof(obuck_hdr))
 164     obuck_broken("Short header read");
 165   b->fdpos = bucket_start + sizeof(obuck_hdr);
 166   if (obuck_hdr.magic != OBUCK_MAGIC)
 167     obuck_broken("Missing magic number");
 168   if (obuck_hdr.oid == OBUCK_OID_DELETED)
 169     obuck_broken("Access to deleted bucket");
 170   if (obuck_hdr.oid != oid)
 171     obuck_broken("Invalid backlink");
 172 }
 173
 174 void
 175 obuck_find_by_oid(struct obuck_header *hdrp)
 176 {
 177   oid_t oid = hdrp->oid;
 178
 179   obuck_lock_read();
 180   obuck_get(oid);
 181   obuck_unlock();
 182   memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 183 }
 184
 185 int
 186 obuck_find_first(struct obuck_header *hdrp, int full)
 187 {
 188   bucket_start = 0;
 189   obuck_hdr.magic = 0;
 190   return obuck_find_next(hdrp, full);
 191 }
 192
 193 int
 194 obuck_find_next(struct obuck_header *hdrp, int full)
 195 {
 196   int c;
 197   struct fastbuf *b = obuck_fb;
 198
 199   for(;;)
 200     {
 201       if (obuck_hdr.magic)
 202         bucket_start = (bucket_start + sizeof(obuck_hdr) + obuck_hdr.length +
 203                         4 + OBUCK_ALIGN - 1) & ~((sh_off_t)(OBUCK_ALIGN - 1));
 204       bflush(b);
 205       obuck_lock_read();
 206       c = sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 207       obuck_unlock();
 208       if (!c)
 209         return 0;
 210       if (c != sizeof(obuck_hdr))
 211         obuck_broken("Short header read");
 212       b->fdpos = bucket_start + sizeof(obuck_hdr);
 213       if (obuck_hdr.magic != OBUCK_MAGIC)
 214         obuck_broken("Missing magic number");
 215       if (obuck_hdr.oid != OBUCK_OID_DELETED || full)
 216         {
 217           memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 218           return 1;
 219         }
 220     }
 221 }
 222
 223 struct fastbuf *
 224 obuck_fetch(void)
 225 {
 226   obuck_remains = obuck_hdr.length;
 227   obuck_check_pad = (OBUCK_ALIGN - sizeof(obuck_hdr) - obuck_hdr.length - 4) & (OBUCK_ALIGN - 1);
 228   return obuck_fb;
 229 }
 230
 231 void
 232 obuck_fetch_end(struct fastbuf *b UNUSED)
 233 {
 234 }
 235
 236 struct fastbuf *
 237 obuck_create(void)
 238 {
 239   obuck_lock_write();
 240   bflush(obuck_fb);
 241   bucket_start = sh_seek(obuck_fd, 0, SEEK_END);
 242   if (bucket_start & (OBUCK_ALIGN - 1))
 243     obuck_broken("Misaligned file");
 244   obuck_hdr.magic = OBUCK_INCOMPLETE_MAGIC;
 245   obuck_hdr.oid = bucket_start >> OBUCK_SHIFT;
 246   obuck_hdr.length = obuck_hdr.orig_length = 0;
 247   obuck_fb->fdpos = obuck_fb->pos = bucket_start;
 248   bwrite(obuck_fb, &obuck_hdr, sizeof(obuck_hdr));
 249   return obuck_fb;
 250 }
 251
 252 void
 253 obuck_create_end(struct fastbuf *b UNUSED, struct obuck_header *hdrp)
 254 {
 255   int pad;
 256   obuck_hdr.magic = OBUCK_MAGIC;
 257   obuck_hdr.length = obuck_hdr.orig_length = btell(obuck_fb) - bucket_start - sizeof(obuck_hdr);
 258   pad = (OBUCK_ALIGN - sizeof(obuck_hdr) - obuck_hdr.length - 4) & (OBUCK_ALIGN - 1);
 259   while (pad--)
 260     bputc(obuck_fb, 0);
 261   bputl(obuck_fb, OBUCK_TRAILER);
 262   bflush(obuck_fb);
 263   ASSERT(!(btell(obuck_fb) & (OBUCK_ALIGN - 1)));
 264   sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 265   obuck_unlock();
 266   memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 267 }
 268
 269 void
 270 obuck_delete(oid_t oid)
 271 {
 272   obuck_lock_write();
 273   obuck_get(oid);
 274   obuck_hdr.oid = OBUCK_OID_DELETED;
 275   sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 276   obuck_unlock();
 277 }
 278
 279 /*** Testing ***/
 280
 281 #ifdef TEST
 282
 283 #define COUNT 100
 284 #define MAXLEN 10000
 285 #define KILLPERC 13
 286 #define LEN(i) ((259309*(i))%MAXLEN)
 287
 288 int main(void)
 289 {
 290   int ids[COUNT];
 291   unsigned int i, j, cnt;
 292   struct obuck_header h;
 293   struct fastbuf *b;
 294   unlink(obuck_name);
 295   obuck_init(1);
 296   for(j=0; j<COUNT; j++)
 297     {
 298       b = obuck_create();
 299       for(i=0; i<LEN(j); i++)
 300         bputc(b, (i+j) % 256);
 301       obuck_create_end(b, &h);
 302       printf("Writing %08x %d -> %d\n", h.oid, h.orig_length, h.length);
 303       ids[j] = h.oid;
 304     }
 305   for(j=0; j<COUNT; j++)
 306     if (j % 100 < KILLPERC)
 307       {
 308         printf("Deleting %08x\n", ids[j]);
 309         obuck_delete(ids[j]);
 310       }
 311   cnt = 0;
 312   for(j=0; j<COUNT; j++)
 313     if (j % 100 >= KILLPERC)
 314       {
 315         cnt++;
 316         h.oid = ids[j];
 317         obuck_find_by_oid(&h);
 318         b = obuck_fetch();
 319         printf("Reading %08x %d -> %d\n", h.oid, h.orig_length, h.length);
 320         if (h.orig_length != LEN(j))
 321           die("Invalid length");
 322         for(i=0; i<h.orig_length; i++)
 323           if ((unsigned) bgetc(b) != (i+j) % 256)
 324             die("Contents mismatch");
 325         if (bgetc(b) != EOF)
 326           die("EOF mismatch");
 327         obuck_fetch_end(b);
 328       }
 329   if (obuck_find_first(&h, 0))
 330     do
 331       {
 332         printf("<<< %08x\t%d\n", h.oid, h.orig_length);
 333         cnt--;
 334       }
 335     while (obuck_find_next(&h, 0));
 336   if (cnt)
 337     die("Walk mismatch");
 338   obuck_cleanup();
 339   return 0;
 340 }
 341
 342 #endif