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 static void
 156 obuck_get(oid_t oid)
 157 {
 158   struct fastbuf *b = obuck_fb;
 159
 160   bucket_start = ((sh_off_t) oid) << OBUCK_SHIFT;
 161   bflush(b);
 162   if (sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start) != sizeof(obuck_hdr))
 163     obuck_broken("Short header read");
 164   b->fdpos = bucket_start + sizeof(obuck_hdr);
 165   if (obuck_hdr.magic != OBUCK_MAGIC)
 166     obuck_broken("Missing magic number");
 167   if (obuck_hdr.oid == OBUCK_OID_DELETED)
 168     obuck_broken("Access to deleted bucket");
 169   if (obuck_hdr.oid != oid)
 170     obuck_broken("Invalid backlink");
 171 }
 172
 173 void
 174 obuck_find_by_oid(struct obuck_header *hdrp)
 175 {
 176   oid_t oid = hdrp->oid;
 177
 178   obuck_lock_read();
 179   obuck_get(oid);
 180   obuck_unlock();
 181   memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 182 }
 183
 184 int
 185 obuck_find_first(struct obuck_header *hdrp, int full)
 186 {
 187   bucket_start = 0;
 188   obuck_hdr.magic = 0;
 189   return obuck_find_next(hdrp, full);
 190 }
 191
 192 int
 193 obuck_find_next(struct obuck_header *hdrp, int full)
 194 {
 195   int c;
 196   struct fastbuf *b = obuck_fb;
 197
 198   for(;;)
 199     {
 200       if (obuck_hdr.magic)
 201         bucket_start = (bucket_start + sizeof(obuck_hdr) + obuck_hdr.length +
 202                         4 + OBUCK_ALIGN - 1) & ~((sh_off_t)(OBUCK_ALIGN - 1));
 203       bflush(b);
 204       obuck_lock_read();
 205       c = sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 206       obuck_unlock();
 207       if (!c)
 208         return 0;
 209       if (c != sizeof(obuck_hdr))
 210         obuck_broken("Short header read");
 211       b->fdpos = bucket_start + sizeof(obuck_hdr);
 212       if (obuck_hdr.magic != OBUCK_MAGIC)
 213         obuck_broken("Missing magic number");
 214       if (obuck_hdr.oid != OBUCK_OID_DELETED || full)
 215         {
 216           memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 217           return 1;
 218         }
 219     }
 220 }
 221
 222 struct fastbuf *
 223 obuck_fetch(void)
 224 {
 225   obuck_remains = obuck_hdr.length;
 226   obuck_check_pad = (OBUCK_ALIGN - sizeof(obuck_hdr) - obuck_hdr.length - 4) & (OBUCK_ALIGN - 1);
 227   return obuck_fb;
 228 }
 229
 230 void
 231 obuck_fetch_end(struct fastbuf *b UNUSED)
 232 {
 233 }
 234
 235 struct fastbuf *
 236 obuck_create(void)
 237 {
 238   obuck_lock_write();
 239   bflush(obuck_fb);
 240   bucket_start = sh_seek(obuck_fd, 0, SEEK_END);
 241   if (bucket_start & (OBUCK_ALIGN - 1))
 242     obuck_broken("Misaligned file");
 243   obuck_hdr.magic = OBUCK_INCOMPLETE_MAGIC;
 244   obuck_hdr.oid = bucket_start >> OBUCK_SHIFT;
 245   obuck_hdr.length = obuck_hdr.orig_length = 0;
 246   obuck_fb->fdpos = obuck_fb->pos = bucket_start;
 247   bwrite(obuck_fb, &obuck_hdr, sizeof(obuck_hdr));
 248   return obuck_fb;
 249 }
 250
 251 void
 252 obuck_create_end(struct fastbuf *b UNUSED, struct obuck_header *hdrp)
 253 {
 254   int pad;
 255   obuck_hdr.magic = OBUCK_MAGIC;
 256   obuck_hdr.length = obuck_hdr.orig_length = btell(obuck_fb) - bucket_start - sizeof(obuck_hdr);
 257   pad = (OBUCK_ALIGN - sizeof(obuck_hdr) - obuck_hdr.length - 4) & (OBUCK_ALIGN - 1);
 258   while (pad--)
 259     bputc(obuck_fb, 0);
 260   bputl(obuck_fb, OBUCK_TRAILER);
 261   bflush(obuck_fb);
 262   ASSERT(!(btell(obuck_fb) & (OBUCK_ALIGN - 1)));
 263   sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 264   obuck_unlock();
 265   memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 266 }
 267
 268 void
 269 obuck_delete(oid_t oid)
 270 {
 271   obuck_lock_write();
 272   obuck_get(oid);
 273   obuck_hdr.oid = OBUCK_OID_DELETED;
 274   sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 275   obuck_unlock();
 276 }
 277
 278 /*** Testing ***/
 279
 280 #ifdef TEST
 281
 282 #define COUNT 100
 283 #define MAXLEN 10000
 284 #define KILLPERC 13
 285 #define LEN(i) ((259309*(i))%MAXLEN)
 286
 287 int main(void)
 288 {
 289   int ids[COUNT];
 290   unsigned int i, j, cnt;
 291   struct obuck_header h;
 292   struct fastbuf *b;
 293   unlink(obuck_name);
 294   obuck_init(1);
 295   for(j=0; j<COUNT; j++)
 296     {
 297       b = obuck_create();
 298       for(i=0; i<LEN(j); i++)
 299         bputc(b, (i+j) % 256);
 300       obuck_create_end(b, &h);
 301       printf("Writing %08x %d -> %d\n", h.oid, h.orig_length, h.length);
 302       ids[j] = h.oid;
 303     }
 304   for(j=0; j<COUNT; j++)
 305     if (j % 100 < KILLPERC)
 306       {
 307         printf("Deleting %08x\n", ids[j]);
 308         obuck_delete(ids[j]);
 309       }
 310   cnt = 0;
 311   for(j=0; j<COUNT; j++)
 312     if (j % 100 >= KILLPERC)
 313       {
 314         cnt++;
 315         h.oid = ids[j];
 316         obuck_find_by_oid(&h);
 317         b = obuck_fetch();
 318         printf("Reading %08x %d -> %d\n", h.oid, h.orig_length, h.length);
 319         if (h.orig_length != LEN(j))
 320           die("Invalid length");
 321         for(i=0; i<h.orig_length; i++)
 322           if ((unsigned) bgetc(b) != (i+j) % 256)
 323             die("Contents mismatch");
 324         if (bgetc(b) != EOF)
 325           die("EOF mismatch");
 326         obuck_fetch_end(b);
 327       }
 328   if (obuck_find_first(&h, 0))
 329     do
 330       {
 331         printf("<<< %08x\t%d\n", h.oid, h.orig_length);
 332         cnt--;
 333       }
 334     while (obuck_find_next(&h, 0));
 335   if (cnt)
 336     die("Walk mismatch");
 337   obuck_cleanup();
 338   return 0;
 339 }
 340
 341 #endif