lib/bucket.c

   1 /*
   2  *      Sherlock Library -- Object Buckets
   3  *
   4  *      (c) 2001--2003 Martin Mares <mj@ucw.cz>
   5  *
   6  *      This software may be freely distributed and used according to the terms
   7  *      of the GNU Lesser General Public License.
   8  */
   9
  10 #include "lib/lib.h"
  11 #include "lib/bucket.h"
  12 #include "lib/fastbuf.h"
  13 #include "lib/lfs.h"
  14 #include "lib/conf.h"
  15
  16 #include <string.h>
  17 #include <stdlib.h>
  18 #include <fcntl.h>
  19 #include <unistd.h>
  20 #include <sys/file.h>
  21
  22 static int obuck_fd;
  23 static unsigned int obuck_remains, obuck_check_pad;
  24 static struct fastbuf *obuck_fb;
  25 static struct obuck_header obuck_hdr;
  26 static sh_off_t bucket_start, bucket_current;
  27
  28 /*** Configuration ***/
  29
  30 byte *obuck_name = "not/configured";
  31 static uns obuck_io_buflen = 65536;
  32 static int obuck_shake_buflen = 1048576;
  33 static uns obuck_slurp_buflen = 65536;
  34
  35 static struct cfitem obuck_config[] = {
  36   { "Buckets",          CT_SECTION,     NULL },
  37   { "BucketFile",       CT_STRING,      &obuck_name },
  38   { "BufSize",          CT_INT,         &obuck_io_buflen },
  39   { "ShakeBufSize",     CT_INT,         &obuck_shake_buflen },
  40   { "SlurpBufSize",     CT_INT,         &obuck_slurp_buflen },
  41   { NULL,               CT_STOP,        NULL }
  42 };
  43
  44 static void CONSTRUCTOR obuck_init_config(void)
  45 {
  46   cf_register(obuck_config);
  47 }
  48
  49 /*** Internal operations ***/
  50
  51 static void
  52 obuck_broken(char *msg)
  53 {
  54   die("Object pool corrupted: %s (pos=%Lx)", msg, (long long) bucket_start);
  55 }
  56
  57 /*
  58  *  Unfortunately we cannot use flock() here since it happily permits
  59  *  locking a shared fd (e.g., after fork()) multiple times. The fcntl
  60  *  locks are very ugly and they don't support 64-bit offsets, but we
  61  *  can work around the problem by always locking the first header
  62  *  in the file.
  63  */
  64
  65 static inline void
  66 obuck_do_lock(int type)
  67 {
  68   struct flock fl;
  69
  70   fl.l_type = type;
  71   fl.l_whence = SEEK_SET;
  72   fl.l_start = 0;
  73   fl.l_len = sizeof(struct obuck_header);
  74   if (fcntl(obuck_fd, F_SETLKW, &fl) < 0)
  75     die("fcntl lock: %m");
  76 }
  77
  78 inline void
  79 obuck_lock_read(void)
  80 {
  81   obuck_do_lock(F_RDLCK);
  82 }
  83
  84 inline void
  85 obuck_lock_write(void)
  86 {
  87   obuck_do_lock(F_WRLCK);
  88 }
  89
  90 inline void
  91 obuck_unlock(void)
  92 {
  93   obuck_do_lock(F_UNLCK);
  94 }
  95
  96 /*** FastIO emulation ***/
  97
  98 /* We need to use pread/pwrite since we work on fd's shared between processes */
  99
 100 static int
 101 obuck_fb_refill(struct fastbuf *f)
 102 {
 103   unsigned limit = (obuck_io_buflen < obuck_remains) ? obuck_io_buflen : obuck_remains;
 104   unsigned size = (limit == obuck_remains) ? (limit+obuck_check_pad+4) : limit;
 105   int l;
 106
 107   if (!limit)
 108     return 0;
 109   l = sh_pread(obuck_fd, f->buffer, size, bucket_current);
 110   if (l < 0)
 111     die("Error reading bucket: %m");
 112   if ((unsigned) l != size)
 113     obuck_broken("Short read");
 114   f->bptr = f->buffer;
 115   f->bstop = f->buffer + limit;
 116   bucket_current += limit;
 117   f->pos = bucket_current - bucket_start - sizeof(obuck_hdr);
 118   obuck_remains -= limit;
 119   if (!obuck_remains)   /* Should check the trailer */
 120     {
 121       if (GET_U32(f->buffer + size - 4) != OBUCK_TRAILER)
 122         obuck_broken("Missing trailer");
 123     }
 124   return limit;
 125 }
 126
 127 static void
 128 obuck_fb_spout(struct fastbuf *f)
 129 {
 130   int l = f->bptr - f->buffer;
 131   char *c = f->buffer;
 132
 133   while (l)
 134     {
 135       int z = sh_pwrite(obuck_fd, c, l, bucket_current);
 136       if (z <= 0)
 137         die("Error writing bucket: %m");
 138       bucket_current += z;
 139       l -= z;
 140       c += z;
 141     }
 142   f->bptr = f->buffer;
 143   f->pos = bucket_current - bucket_start - sizeof(obuck_hdr);
 144 }
 145
 146 /*** Exported functions ***/
 147
 148 void
 149 obuck_init(int writeable)
 150 {
 151   struct fastbuf *b;
 152   sh_off_t size;
 153
 154   obuck_fd = sh_open(obuck_name, (writeable ? O_RDWR | O_CREAT : O_RDONLY), 0666);
 155   if (obuck_fd < 0)
 156     die("Unable to open bucket file %s: %m", obuck_name);
 157   obuck_fb = b = xmalloc_zero(sizeof(struct fastbuf) + obuck_io_buflen + OBUCK_ALIGN + 4);
 158   b->buffer = (char *)(b+1);
 159   b->bptr = b->bstop = b->buffer;
 160   b->bufend = b->buffer + obuck_io_buflen;
 161   b->name = "bucket";
 162   b->refill = obuck_fb_refill;
 163   b->spout = obuck_fb_spout;
 164   obuck_lock_read();
 165   size = sh_seek(obuck_fd, 0, SEEK_END);
 166   if (size)
 167     {
 168       /* If the bucket pool is not empty, check consistency of its end */
 169       u32 check;
 170       bucket_start = size - 4;  /* for error reporting */
 171       if (sh_pread(obuck_fd, &check, 4, size-4) != 4 ||
 172           check != OBUCK_TRAILER)
 173         obuck_broken("Missing trailer of last object");
 174     }
 175   obuck_unlock();
 176 }
 177
 178 void
 179 obuck_cleanup(void)
 180 {
 181   bclose(obuck_fb);
 182   close(obuck_fd);
 183   xfree(obuck_fb);
 184 }
 185
 186 void
 187 obuck_sync(void)
 188 {
 189   bflush(obuck_fb);
 190   fsync(obuck_fd);
 191 }
 192
 193 static void
 194 obuck_get(oid_t oid)
 195 {
 196   struct fastbuf *b = obuck_fb;
 197
 198   bucket_start = obuck_get_pos(oid);
 199   bflush(b);
 200   if (sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start) != sizeof(obuck_hdr))
 201     obuck_broken("Short header read");
 202   bucket_current = bucket_start + sizeof(obuck_hdr);
 203   if (obuck_hdr.magic != OBUCK_MAGIC)
 204     obuck_broken("Missing magic number");
 205   if (obuck_hdr.oid == OBUCK_OID_DELETED)
 206     obuck_broken("Access to deleted bucket");
 207   if (obuck_hdr.oid != oid)
 208     obuck_broken("Invalid backlink");
 209 }
 210
 211 void
 212 obuck_find_by_oid(struct obuck_header *hdrp)
 213 {
 214   oid_t oid = hdrp->oid;
 215
 216   ASSERT(oid < OBUCK_OID_FIRST_SPECIAL);
 217   obuck_lock_read();
 218   obuck_get(oid);
 219   obuck_unlock();
 220   memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 221 }
 222
 223 int
 224 obuck_find_first(struct obuck_header *hdrp, int full)
 225 {
 226   bucket_start = 0;
 227   obuck_hdr.magic = 0;
 228   return obuck_find_next(hdrp, full);
 229 }
 230
 231 int
 232 obuck_find_next(struct obuck_header *hdrp, int full)
 233 {
 234   int c;
 235   struct fastbuf *b = obuck_fb;
 236
 237   for(;;)
 238     {
 239       if (obuck_hdr.magic)
 240         bucket_start = (bucket_start + sizeof(obuck_hdr) + obuck_hdr.length +
 241                         4 + OBUCK_ALIGN - 1) & ~((sh_off_t)(OBUCK_ALIGN - 1));
 242       bflush(b);
 243       obuck_lock_read();
 244       c = sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 245       obuck_unlock();
 246       if (!c)
 247         return 0;
 248       if (c != sizeof(obuck_hdr))
 249         obuck_broken("Short header read");
 250       bucket_current = bucket_start + sizeof(obuck_hdr);
 251       if (obuck_hdr.magic != OBUCK_MAGIC)
 252         obuck_broken("Missing magic number");
 253       if (obuck_hdr.oid != OBUCK_OID_DELETED || full)
 254         {
 255           memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 256           return 1;
 257         }
 258     }
 259 }
 260
 261 struct fastbuf *
 262 obuck_fetch(void)
 263 {
 264   obuck_fb->pos = 0;
 265   obuck_remains = obuck_hdr.length;
 266   obuck_check_pad = (OBUCK_ALIGN - sizeof(obuck_hdr) - obuck_hdr.length - 4) & (OBUCK_ALIGN - 1);
 267   return obuck_fb;
 268 }
 269
 270 void
 271 obuck_fetch_end(struct fastbuf *b UNUSED)
 272 {
 273 }
 274
 275 oid_t
 276 obuck_predict_last_oid(void)
 277 {
 278   sh_off_t size = sh_seek(obuck_fd, 0, SEEK_END);
 279   return size >> OBUCK_SHIFT;
 280 }
 281
 282 struct fastbuf *
 283 obuck_create(u32 type)
 284 {
 285   obuck_lock_write();
 286   bflush(obuck_fb);
 287   bucket_start = sh_seek(obuck_fd, 0, SEEK_END);
 288   if (bucket_start & (OBUCK_ALIGN - 1))
 289     obuck_broken("Misaligned file");
 290   obuck_hdr.magic = OBUCK_INCOMPLETE_MAGIC;
 291   obuck_hdr.oid = bucket_start >> OBUCK_SHIFT;
 292   obuck_hdr.length = 0;
 293   obuck_hdr.type = type;
 294   bucket_current = bucket_start;
 295   bwrite(obuck_fb, &obuck_hdr, sizeof(obuck_hdr));
 296   obuck_fb->pos = -sizeof(obuck_hdr);
 297   return obuck_fb;
 298 }
 299
 300 void
 301 obuck_create_end(struct fastbuf *b UNUSED, struct obuck_header *hdrp)
 302 {
 303   int pad;
 304   obuck_hdr.magic = OBUCK_MAGIC;
 305   obuck_hdr.length = btell(obuck_fb);
 306   pad = (OBUCK_ALIGN - sizeof(obuck_hdr) - obuck_hdr.length - 4) & (OBUCK_ALIGN - 1);
 307   while (pad--)
 308     bputc(obuck_fb, 0);
 309   bputl(obuck_fb, OBUCK_TRAILER);
 310   bflush(obuck_fb);
 311   ASSERT(!(bucket_current & (OBUCK_ALIGN - 1)));
 312   sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 313   obuck_unlock();
 314   memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
 315 }
 316
 317 void
 318 obuck_delete(oid_t oid)
 319 {
 320   obuck_lock_write();
 321   obuck_get(oid);
 322   obuck_hdr.oid = OBUCK_OID_DELETED;
 323   sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
 324   obuck_unlock();
 325 }
 326
 327 /*** Fast reading of the whole pool ***/
 328
 329 static struct fastbuf *obuck_rpf;
 330
 331 static int
 332 obuck_slurp_refill(struct fastbuf *f)
 333 {
 334   uns l;
 335
 336   if (!obuck_remains)
 337     return 0;
 338   l = bdirect_read_prepare(obuck_rpf, &f->buffer);
 339   if (!l)
 340     obuck_broken("Incomplete object");
 341   l = MIN(l, obuck_remains);
 342   bdirect_read_commit(obuck_rpf, f->buffer + l);
 343   obuck_remains -= l;
 344   f->bptr = f->buffer;
 345   f->bufend = f->bstop = f->buffer + l;
 346   return 1;
 347 }
 348
 349 struct fastbuf *
 350 obuck_slurp_pool(struct obuck_header *hdrp)
 351 {
 352   static struct fastbuf limiter;
 353   uns l;
 354
 355   do
 356     {
 357       if (!obuck_rpf)
 358         {
 359           obuck_lock_read();
 360           obuck_rpf = bopen(obuck_name, O_RDONLY, obuck_slurp_buflen);
 361         }
 362       else
 363         {
 364           bsetpos(obuck_rpf, bucket_current - 4);
 365           if (bgetl(obuck_rpf) != OBUCK_TRAILER)
 366             obuck_broken("Missing trailer");
 367         }
 368       bucket_start = btell(obuck_rpf);
 369       l = bread(obuck_rpf, hdrp, sizeof(struct obuck_header));
 370       if (!l)
 371         {
 372           bclose(obuck_rpf);
 373           obuck_rpf = NULL;
 374           obuck_unlock();
 375           return NULL;
 376         }
 377       if (l != sizeof(struct obuck_header))
 378         obuck_broken("Short header read");
 379       if (hdrp->magic != OBUCK_MAGIC)
 380         obuck_broken("Missing magic number");
 381       bucket_current = (bucket_start + sizeof(obuck_hdr) + hdrp->length +
 382                         4 + OBUCK_ALIGN - 1) & ~((sh_off_t)(OBUCK_ALIGN - 1));
 383     }
 384   while (hdrp->oid == OBUCK_OID_DELETED);
 385   if (obuck_get_pos(hdrp->oid) != bucket_start)
 386     obuck_broken("Invalid backlink");
 387   obuck_remains = hdrp->length;
 388   limiter.bptr = limiter.bstop = limiter.buffer = limiter.bufend = NULL;
 389   limiter.name = "Bucket";
 390   limiter.pos = 0;
 391   limiter.refill = obuck_slurp_refill;
 392   return &limiter;
 393 }
 394
 395 /*** Shakedown ***/
 396
 397 void
 398 obuck_shakedown(int (*kibitz)(struct obuck_header *old, oid_t new, byte *buck))
 399 {
 400   byte *rbuf, *wbuf, *msg;
 401   sh_off_t rstart, wstart, w_bucket_start;
 402   int roff, woff, rsize, l;
 403   struct obuck_header *rhdr, *whdr;
 404
 405   rbuf = xmalloc(obuck_shake_buflen);
 406   wbuf = xmalloc(obuck_shake_buflen);
 407   rstart = wstart = 0;
 408   roff = woff = rsize = 0;
 409
 410   /* We need to be the only accessor, all the object ID's are becoming invalid */
 411   obuck_lock_write();
 412
 413   for(;;)
 414     {
 415       bucket_start = rstart + roff;
 416       w_bucket_start = wstart + woff;
 417       if (rsize - roff < OBUCK_ALIGN)
 418         goto reread;
 419       rhdr = (struct obuck_header *)(rbuf + roff);
 420       if (rhdr->magic != OBUCK_MAGIC ||
 421           rhdr->oid != OBUCK_OID_DELETED && rhdr->oid != (oid_t)(bucket_start >> OBUCK_SHIFT))
 422         {
 423           msg = "header mismatch";
 424           goto broken;
 425         }
 426       l = (sizeof(struct obuck_header) + rhdr->length + 4 + OBUCK_ALIGN - 1) & ~(OBUCK_ALIGN-1);
 427       if (l > obuck_shake_buflen)
 428         {
 429           if (rhdr->oid != OBUCK_OID_DELETED)
 430             {
 431               msg = "bucket longer than ShakeBufSize";
 432               goto broken;
 433             }
 434           rstart = bucket_start + l;
 435           roff = 0;
 436           rsize = 0;
 437           goto reread;
 438         }
 439       if (rsize - roff < l)
 440         goto reread;
 441       if (GET_U32(rbuf + roff + l - 4) != OBUCK_TRAILER)
 442         {
 443           msg = "missing trailer";
 444           goto broken;
 445         }
 446       if (rhdr->oid != OBUCK_OID_DELETED)
 447         {
 448           if (kibitz(rhdr, w_bucket_start >> OBUCK_SHIFT, (byte *)(rhdr+1)))
 449             {
 450               if (bucket_start == w_bucket_start)
 451                 {
 452                   /* No copying needed now nor ever in the past, hence woff==0 */
 453                   wstart += l;
 454                 }
 455               else
 456                 {
 457                   if (obuck_shake_buflen - woff < l)
 458                     {
 459                       if (sh_pwrite(obuck_fd, wbuf, woff, wstart) != woff)
 460                         die("obuck_shakedown write failed: %m");
 461                       wstart += woff;
 462                       woff = 0;
 463                     }
 464                   whdr = (struct obuck_header *)(wbuf+woff);
 465                   memcpy(whdr, rhdr, l);
 466                   whdr->oid = w_bucket_start >> OBUCK_SHIFT;
 467                   woff += l;
 468                 }
 469             }
 470         }
 471       else
 472         kibitz(rhdr, OBUCK_OID_DELETED, NULL);
 473       roff += l;
 474       continue;
 475
 476     reread:
 477       if (roff)
 478         {
 479           memmove(rbuf, rbuf+roff, rsize-roff);
 480           rsize -= roff;
 481           rstart += roff;
 482           roff = 0;
 483         }
 484       l = sh_pread(obuck_fd, rbuf+rsize, obuck_shake_buflen-rsize, rstart+rsize);
 485       if (l < 0)
 486         die("obuck_shakedown read error: %m");
 487       if (!l)
 488         {
 489           if (!rsize)
 490             break;
 491           msg = "unexpected EOF";
 492           goto broken;
 493         }
 494       rsize += l;
 495     }
 496   if (woff)
 497     {
 498       if (sh_pwrite(obuck_fd, wbuf, woff, wstart) != woff)
 499         die("obuck_shakedown write failed: %m");
 500       wstart += woff;
 501     }
 502   sh_ftruncate(obuck_fd, wstart);
 503
 504   obuck_unlock();
 505   xfree(rbuf);
 506   xfree(wbuf);
 507   return;
 508
 509  broken:
 510   log(L_ERROR, "Error during object pool shakedown: %s (pos=%Ld, id=%x), gathering debris", msg, (long long) bucket_start, (uns)(bucket_start >> OBUCK_SHIFT));
 511   if (woff)
 512     {
 513       sh_pwrite(obuck_fd, wbuf, woff, wstart);
 514       wstart += woff;
 515     }
 516   while (wstart + OBUCK_ALIGN <= bucket_start)
 517     {
 518       u32 check = OBUCK_TRAILER;
 519       obuck_hdr.magic = OBUCK_MAGIC;
 520       obuck_hdr.oid = OBUCK_OID_DELETED;
 521       if (bucket_start - wstart < 0x40000000)
 522         obuck_hdr.length = bucket_start - wstart - sizeof(obuck_hdr) - 4;
 523       else
 524         obuck_hdr.length = 0x40000000 - sizeof(obuck_hdr) - 4;
 525       sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), wstart);
 526       wstart += sizeof(obuck_hdr) + obuck_hdr.length + 4;
 527       sh_pwrite(obuck_fd, &check, 4, wstart-4);
 528     }
 529   die("Fatal error during object pool shakedown");
 530 }
 531
 532 /*** Testing ***/
 533
 534 #ifdef TEST
 535
 536 #define COUNT 5000
 537 #define MAXLEN 10000
 538 #define KILLPERC 13
 539 #define LEN(i) ((259309*(i))%MAXLEN)
 540
 541 int main(int argc, char **argv)
 542 {
 543   int ids[COUNT];
 544   unsigned int i, j, cnt;
 545   struct obuck_header h;
 546   struct fastbuf *b;
 547
 548   log_init(NULL);
 549   if (cf_getopt(argc, argv, CF_SHORT_OPTS, CF_NO_LONG_OPTS, NULL) >= 0 ||
 550       optind < argc)
 551   {
 552     fputs("This program supports only the following command-line arguments:\n" CF_USAGE, stderr);
 553     exit(1);
 554   }
 555
 556   unlink(obuck_name);
 557   obuck_init(1);
 558   for(j=0; j<COUNT; j++)
 559     {
 560       b = obuck_create();
 561       for(i=0; i<LEN(j); i++)
 562         bputc(b, (i+j) % 256);
 563       obuck_create_end(b, &h);
 564       printf("Writing %08x %d\n", h.oid, h.length);
 565       ids[j] = h.oid;
 566     }
 567   for(j=0; j<COUNT; j++)
 568     if (j % 100 < KILLPERC)
 569       {
 570         printf("Deleting %08x\n", ids[j]);
 571         obuck_delete(ids[j]);
 572       }
 573   cnt = 0;
 574   for(j=0; j<COUNT; j++)
 575     if (j % 100 >= KILLPERC)
 576       {
 577         cnt++;
 578         h.oid = ids[j];
 579         obuck_find_by_oid(&h);
 580         b = obuck_fetch();
 581         printf("Reading %08x %d\n", h.oid, h.length);
 582         if (h.length != LEN(j))
 583           die("Invalid length");
 584         for(i=0; i<h.length; i++)
 585           if ((unsigned) bgetc(b) != (i+j) % 256)
 586             die("Contents mismatch");
 587         if (bgetc(b) != EOF)
 588           die("EOF mismatch");
 589         obuck_fetch_end(b);
 590       }
 591   if (obuck_find_first(&h, 0))
 592     do
 593       {
 594         printf("<<< %08x\t%d\n", h.oid, h.length);
 595         cnt--;
 596       }
 597     while (obuck_find_next(&h, 0));
 598   if (cnt)
 599     die("Walk mismatch");
 600   obuck_cleanup();
 601   return 0;
 602 }
 603
 604 #endif