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_zero(sizeof(struct fastbuf) + buflen + OBUCK_ALIGN + 4);
 119   b->buflen = buflen;
 120   b->buffer = (char *)(b+1);
 121   b->bptr = b->bstop = b->buffer;
 122   b->bufend = b->buffer + buflen;
 123   b->name = "bucket";
 124   b->fd = obuck_fd;
 125   b->refill = obuck_fb_refill;
 126   b->spout = obuck_fb_spout;
 127   b->close = obuck_fb_close;
 128   obuck_lock_read();
 129   size = sh_seek(obuck_fd, 0, SEEK_END);
 130   if (size)
 131     {
 132       /* If the bucket pool is not empty, check consistency of its end */
 133       u32 check;
 134       bucket_start = size - 4;  /* for error reporting */
 135       if (sh_pread(obuck_fd, &check, 4, size-4) != 4 ||
 136           check != OBUCK_TRAILER)
 137         obuck_broken("Missing trailer of last object");
 138     }
 139   obuck_unlock();
 140 }
 141
 142 void
 143 obuck_cleanup(void)
 144 {
 145   bclose(obuck_fb);
 146 }
 147
 148 void                                    /* FIXME: Call somewhere :) */
 149 obuck_sync(void)
 150 {
 151   bflush(obuck_fb);
 152   fsync(obuck_fd);
 153 }
 154
 155 sh_off_t
 156 obuck_size(void)
 157 {
 158   return sh_seek(obuck_fd, 0, SEEK_END);
 159 }
 160
 161 static void
 162 obuck_get(oid_t oid)
 163 {
 164   struct fastbuf *b = obuck_fb;
 165
 166   bucket_start = ((sh_off_t) oid) << OBUCK_SHIFT;
 167   bflush(b);
 168   if (sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start) != sizeof(obuck_hdr))
 169     obuck_broken("Short header read");
 170   b->fdpos = bucket_start + sizeof(obuck_hdr);
 171   if (obuck_hdr.magic != OBUCK_MAGIC)
 172     obuck_broken("Missing magic number");
 173   if (obuck_hdr.oid == OBUCK_OID_DELETED)
 174     obuck_broken("Access to deleted bucket");
 175   if (obuck_hdr.oid != oid)
 176     obuck_broken("Invalid backlink");
 177 }
 178
 179 void
 180 obuck_find_by_oid(struct obuck_header *hdrp)
 181 {
 182   oid_t oid = hdrp->oid;
 183
 184   ASSERT(oid < OBUCK_OID_FIRST_SPECIAL);
 185   obuck_lock_read();
 186   obuck_get(oid);
 187   obuck_unlock();
 188   memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 189 }
 190
 191 int
 192 obuck_find_first(struct obuck_header *hdrp, int full)
 193 {
 194   bucket_start = 0;
 195   obuck_hdr.magic = 0;
 196   return obuck_find_next(hdrp, full);
 197 }
 198
 199 int
 200 obuck_find_next(struct obuck_header *hdrp, int full)
 201 {
 202   int c;
 203   struct fastbuf *b = obuck_fb;
 204
 205   for(;;)
 206     {
 207       if (obuck_hdr.magic)
 208         bucket_start = (bucket_start + sizeof(obuck_hdr) + obuck_hdr.length +
 209                         4 + OBUCK_ALIGN - 1) & ~((sh_off_t)(OBUCK_ALIGN - 1));
 210       bflush(b);
 211       obuck_lock_read();
 212       c = sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 213       obuck_unlock();
 214       if (!c)
 215         return 0;
 216       if (c != sizeof(obuck_hdr))
 217         obuck_broken("Short header read");
 218       b->fdpos = bucket_start + sizeof(obuck_hdr);
 219       if (obuck_hdr.magic != OBUCK_MAGIC)
 220         obuck_broken("Missing magic number");
 221       if (obuck_hdr.oid != OBUCK_OID_DELETED || full)
 222         {
 223           memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 224           return 1;
 225         }
 226     }
 227 }
 228
 229 struct fastbuf *
 230 obuck_fetch(void)
 231 {
 232   obuck_remains = obuck_hdr.length;
 233   obuck_check_pad = (OBUCK_ALIGN - sizeof(obuck_hdr) - obuck_hdr.length - 4) & (OBUCK_ALIGN - 1);
 234   return obuck_fb;
 235 }
 236
 237 void
 238 obuck_fetch_end(struct fastbuf *b UNUSED)
 239 {
 240 }
 241
 242 struct fastbuf *
 243 obuck_create(void)
 244 {
 245   obuck_lock_write();
 246   bflush(obuck_fb);
 247   bucket_start = sh_seek(obuck_fd, 0, SEEK_END);
 248   if (bucket_start & (OBUCK_ALIGN - 1))
 249     obuck_broken("Misaligned file");
 250   obuck_hdr.magic = OBUCK_INCOMPLETE_MAGIC;
 251   obuck_hdr.oid = bucket_start >> OBUCK_SHIFT;
 252   obuck_hdr.length = obuck_hdr.orig_length = 0;
 253   obuck_fb->fdpos = obuck_fb->pos = bucket_start;
 254   bwrite(obuck_fb, &obuck_hdr, sizeof(obuck_hdr));
 255   return obuck_fb;
 256 }
 257
 258 void
 259 obuck_create_end(struct fastbuf *b UNUSED, struct obuck_header *hdrp)
 260 {
 261   int pad;
 262   obuck_hdr.magic = OBUCK_MAGIC;
 263   obuck_hdr.length = obuck_hdr.orig_length = btell(obuck_fb) - bucket_start - sizeof(obuck_hdr);
 264   pad = (OBUCK_ALIGN - sizeof(obuck_hdr) - obuck_hdr.length - 4) & (OBUCK_ALIGN - 1);
 265   while (pad--)
 266     bputc(obuck_fb, 0);
 267   bputl(obuck_fb, OBUCK_TRAILER);
 268   bflush(obuck_fb);
 269   ASSERT(!(btell(obuck_fb) & (OBUCK_ALIGN - 1)));
 270   sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 271   obuck_unlock();
 272   memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 273 }
 274
 275 void
 276 obuck_delete(oid_t oid)
 277 {
 278   obuck_lock_write();
 279   obuck_get(oid);
 280   obuck_hdr.oid = OBUCK_OID_DELETED;
 281   sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 282   obuck_unlock();
 283 }
 284
 285 /*** Testing ***/
 286
 287 #ifdef TEST
 288
 289 #define COUNT 100
 290 #define MAXLEN 10000
 291 #define KILLPERC 13
 292 #define LEN(i) ((259309*(i))%MAXLEN)
 293
 294 int main(void)
 295 {
 296   int ids[COUNT];
 297   unsigned int i, j, cnt;
 298   struct obuck_header h;
 299   struct fastbuf *b;
 300   unlink(obuck_name);
 301   obuck_init(1);
 302   for(j=0; j<COUNT; j++)
 303     {
 304       b = obuck_create();
 305       for(i=0; i<LEN(j); i++)
 306         bputc(b, (i+j) % 256);
 307       obuck_create_end(b, &h);
 308       printf("Writing %08x %d -> %d\n", h.oid, h.orig_length, h.length);
 309       ids[j] = h.oid;
 310     }
 311   for(j=0; j<COUNT; j++)
 312     if (j % 100 < KILLPERC)
 313       {
 314         printf("Deleting %08x\n", ids[j]);
 315         obuck_delete(ids[j]);
 316       }
 317   cnt = 0;
 318   for(j=0; j<COUNT; j++)
 319     if (j % 100 >= KILLPERC)
 320       {
 321         cnt++;
 322         h.oid = ids[j];
 323         obuck_find_by_oid(&h);
 324         b = obuck_fetch();
 325         printf("Reading %08x %d -> %d\n", h.oid, h.orig_length, h.length);
 326         if (h.orig_length != LEN(j))
 327           die("Invalid length");
 328         for(i=0; i<h.orig_length; i++)
 329           if ((unsigned) bgetc(b) != (i+j) % 256)
 330             die("Contents mismatch");
 331         if (bgetc(b) != EOF)
 332           die("EOF mismatch");
 333         obuck_fetch_end(b);
 334       }
 335   if (obuck_find_first(&h, 0))
 336     do
 337       {
 338         printf("<<< %08x\t%d\n", h.oid, h.orig_length);
 339         cnt--;
 340       }
 341     while (obuck_find_next(&h, 0));
 342   if (cnt)
 343     die("Walk mismatch");
 344   obuck_cleanup();
 345   return 0;
 346 }
 347
 348 #endif