lib/db.c

   1 /*
   2  *      Sherlock Library -- Fast Database Management Routines
   3  *
   4  *      (c) 1999--2000 Martin Mares <mj@ucw.cz>
   5  */
   6
   7 /*
   8  *  This library uses the standard algorithm for external hashing (page directory
   9  *  mapping topmost K bits of hash value to page address, directory splits and
  10  *  so on). Peculiarities of this implementation (aka design decisions):
  11  *
  12  *   o  We allow both fixed and variable length keys and values (this includes
  13  *      zero size values for cases you want to represent only a set of keys).
  14  *   o  We assume that key_size + val_size < page_size.
  15  *   o  We never shrink the directory nor free empty pages. (The reason is that
  16  *      if the database was once large, it's likely it will again become large soon.)
  17  *   o  The only pages which can be freed are those of the directory (during
  18  *      directory split), so we keep only a simple 32-entry free block list
  19  *      and we assume it's sorted.
  20  */
  21
  22 #include "lib/lib.h"
  23 #include "lib/lfs.h"
  24 #include "lib/pagecache.h"
  25 #include "lib/db.h"
  26 #include "lib/db_internal.h"
  27
  28 #include <stdio.h>
  29 #include <stdlib.h>
  30 #include <string.h>
  31 #include <fcntl.h>
  32 #include <unistd.h>
  33
  34 struct sdbm *
  35 sdbm_open(struct sdbm_options *o)
  36 {
  37   struct sdbm *d;
  38   struct sdbm_root root, *r;
  39   uns cache_size = o->cache_size ? o->cache_size : 16;
  40
  41   d = xmalloc_zero(sizeof(struct sdbm));
  42   d->flags = o->flags;
  43   d->fd = sh_open(o->name, ((d->flags & SDBM_WRITE) ? O_RDWR : O_RDONLY), 0666);
  44   if (d->fd >= 0)                       /* Already exists, let's check it */
  45     {
  46       if (read(d->fd, &root, sizeof(root)) != sizeof(root))
  47         goto bad;
  48       if (root.magic != SDBM_MAGIC || root.version != SDBM_VERSION)
  49         goto bad;
  50       d->file_size = sh_seek(d->fd, 0, SEEK_END);
  51       d->page_size = 1 << root.page_order;
  52       d->cache = pgc_open(d->page_size, cache_size);
  53       d->root_page = pgc_read(d->cache, d->fd, 0);
  54       d->root = (void *) d->root_page->data;
  55     }
  56   else if ((d->flags & SDBM_CREAT) && (d->fd = sh_open(o->name, O_RDWR | O_CREAT, 0666)) >= 0)
  57     {
  58       struct page *q;
  59       uns page_order = o->page_order;
  60       if (page_order < 10)
  61         page_order = 10;
  62       d->page_size = 1 << page_order;
  63       d->cache = pgc_open(d->page_size, cache_size);
  64       d->root_page = pgc_get_zero(d->cache, d->fd, 0);
  65       r = d->root = (void *) d->root_page->data;                 /* Build root page */
  66       r->magic = SDBM_MAGIC;
  67       r->version = SDBM_VERSION;
  68       r->page_order = page_order;
  69       r->key_size = o->key_size;
  70       r->val_size = o->val_size;
  71       r->dir_start = d->page_size;
  72       r->dir_order = 0;
  73       d->file_size = 3*d->page_size;
  74       q = pgc_get_zero(d->cache, d->fd, d->page_size);          /* Build page directory */
  75       ((u32 *)q->data)[0] = 2*d->page_size;
  76       pgc_put(d->cache, q);
  77       q = pgc_get_zero(d->cache, d->fd, 2*d->page_size);        /* Build single data page */
  78       pgc_put(d->cache, q);
  79     }
  80   else
  81     goto bad;
  82   d->dir_size = 1 << d->root->dir_order;
  83   d->dir_shift = 32 - d->root->dir_order;
  84   d->page_order = d->root->page_order;
  85   d->page_mask = d->page_size - 1;
  86   d->key_size = d->root->key_size;
  87   d->val_size = d->root->val_size;
  88   return d;
  89
  90 bad:
  91   sdbm_close(d);
  92   return NULL;
  93 }
  94
  95 void
  96 sdbm_close(struct sdbm *d)
  97 {
  98   if (d->root_page)
  99     pgc_put(d->cache, d->root_page);
 100   if (d->cache)
 101     pgc_close(d->cache);
 102   if (d->fd >= 0)
 103     close(d->fd);
 104   xfree(d);
 105 }
 106
 107 static uns
 108 sdbm_alloc_pages(struct sdbm *d, uns number)
 109 {
 110   uns where = d->file_size;
 111   uns size = number << d->page_order;
 112   if (d->file_size + size < d->file_size)       /* Wrap around? */
 113     die("SDB: Database file too large, giving up");
 114   d->file_size += size;
 115   return where;
 116 }
 117
 118 static uns
 119 sdbm_alloc_page(struct sdbm *d)
 120 {
 121   uns pos;
 122
 123   if (!d->root->free_pool[0].count)
 124     return sdbm_alloc_pages(d, 1);
 125   pos = d->root->free_pool[0].first;
 126   d->root->free_pool[0].first += d->page_size;
 127   if (!--d->root->free_pool[0].count)
 128     {
 129       memmove(d->root->free_pool, d->root->free_pool+1, SDBM_NUM_FREE_PAGE_POOLS * sizeof(d->root->free_pool[0]));
 130       d->root->free_pool[SDBM_NUM_FREE_PAGE_POOLS-1].count = 0;
 131     }
 132   pgc_mark_dirty(d->cache, d->root_page);
 133   return pos;
 134 }
 135
 136 static void
 137 sdbm_free_pages(struct sdbm *d, uns start, uns number)
 138 {
 139   uns i = 0;
 140
 141   while (d->root->free_pool[i].count)
 142     i++;
 143   d->root->free_pool[i].first = start;
 144   d->root->free_pool[i].count = number;
 145   pgc_mark_dirty(d->cache, d->root_page);
 146 }
 147
 148 static u32
 149 sdbm_hash(byte *key, uns keylen)
 150 {
 151   /*
 152    *  This used to be the same hash function as GDBM uses,
 153    *  but it turned out that it tends to give the same results
 154    *  on similar keys. Damn it.
 155    */
 156   u32 value = 0x238F13AF * keylen;
 157   while (keylen--)
 158     value = 37*value + *key++;
 159   return (1103515243 * value + 12345);
 160 }
 161
 162 static int
 163 sdbm_get_entry(struct sdbm *d, byte *pos, byte **key, uns *keylen, byte **val, uns *vallen)
 164 {
 165   byte *p = pos;
 166
 167   if (d->key_size >= 0)
 168     *keylen = d->key_size;
 169   else
 170     {
 171       *keylen = (p[0] << 8) | p[1];
 172       p += 2;
 173     }
 174   *key = p;
 175   p += *keylen;
 176   if (d->val_size >= 0)
 177     *vallen = d->val_size;
 178   else
 179     {
 180       *vallen = (p[0] << 8) | p[1];
 181       p += 2;
 182     }
 183   *val = p;
 184   p += *vallen;
 185   return p - pos;
 186 }
 187
 188 static int
 189 sdbm_entry_len(struct sdbm *d, uns keylen, uns vallen)
 190 {
 191   uns len = keylen + vallen;
 192   if (d->key_size < 0)
 193     len += 2;
 194   if (d->val_size < 0)
 195     len += 2;
 196   return len;
 197 }
 198
 199 static void
 200 sdbm_store_entry(struct sdbm *d, byte *pos, byte *key, uns keylen, byte *val, uns vallen)
 201 {
 202   if (d->key_size < 0)
 203     {
 204       *pos++ = keylen >> 8;
 205       *pos++ = keylen;
 206     }
 207   memmove(pos, key, keylen);
 208   pos += keylen;
 209   if (d->val_size < 0)
 210     {
 211       *pos++ = vallen >> 8;
 212       *pos++ = vallen;
 213     }
 214   memmove(pos, val, vallen);
 215 }
 216
 217 static uns
 218 sdbm_page_rank(struct sdbm *d, uns dirpos)
 219 {
 220   struct page *b;
 221   u32 pg, x;
 222   uns l, r;
 223   uns pm = d->page_mask;
 224
 225   b = pgc_read(d->cache, d->fd, d->root->dir_start + (dirpos & ~pm));
 226   pg = GET32(b->data, dirpos & pm);
 227   l = dirpos;
 228   while ((l & pm) && GET32(b->data, (l - 4) & pm) == pg)
 229     l -= 4;
 230   r = dirpos + 4;
 231   /* We heavily depend on unused directory entries being zero */
 232   while ((r & pm) && GET32(b->data, r & pm) == pg)
 233     r += 4;
 234   pgc_put(d->cache, b);
 235
 236   if (!(l & pm) && !(r & pm))
 237     {
 238       /* Note that if it spans page boundary, it must contain an integer number of pages */
 239       while (l)
 240         {
 241           b = pgc_read(d->cache, d->fd, d->root->dir_start + ((l - 4) & ~pm));
 242           x = GET32(b->data, 0);
 243           pgc_put(d->cache, b);
 244           if (x != pg)
 245             break;
 246           l -= d->page_size;
 247         }
 248       while (r < 4*d->dir_size)
 249         {
 250           b = pgc_read(d->cache, d->fd, d->root->dir_start + (r & ~pm));
 251           x = GET32(b->data, 0);
 252           pgc_put(d->cache, b);
 253           if (x != pg)
 254             break;
 255           r += d->page_size;
 256         }
 257     }
 258   return (r - l) >> 2;
 259 }
 260
 261 static void
 262 sdbm_expand_directory(struct sdbm *d)
 263 {
 264   struct page *b, *c;
 265   int i, ent;
 266   u32 *dir, *t;
 267
 268   if (4*d->dir_size < d->page_size)
 269     {
 270       /* It still fits within single page */
 271       b = pgc_read(d->cache, d->fd, d->root->dir_start);
 272       dir = (u32 *) b->data;
 273       for(i=d->dir_size-1; i>=0; i--)
 274         dir[2*i] = dir[2*i+1] = dir[i];
 275       pgc_mark_dirty(d->cache, b);
 276       pgc_put(d->cache, b);
 277     }
 278   else
 279     {
 280       uns old_dir = d->root->dir_start;
 281       uns old_dir_pages = 1 << (d->root->dir_order + 2 - d->page_order);
 282       uns page, new_dir;
 283       new_dir = d->root->dir_start = sdbm_alloc_pages(d, 2*old_dir_pages);
 284       ent = 1 << (d->page_order - 3);
 285       for(page=0; page < old_dir_pages; page++)
 286         {
 287           b = pgc_read(d->cache, d->fd, old_dir + (page << d->page_order));
 288           dir = (u32 *) b->data;
 289           c = pgc_get(d->cache, d->fd, new_dir + (page << (d->page_order + 1)));
 290           t = (u32 *) c->data;
 291           for(i=0; i<ent; i++)
 292             t[2*i] = t[2*i+1] = dir[i];
 293           pgc_put(d->cache, c);
 294           c = pgc_get(d->cache, d->fd, new_dir + (page << (d->page_order + 1)) + d->page_size);
 295           t = (u32 *) c->data;
 296           for(i=0; i<ent; i++)
 297             t[2*i] = t[2*i+1] = dir[ent+i];
 298           pgc_put(d->cache, c);
 299           pgc_put(d->cache, b);
 300         }
 301       if (!(d->flags & SDBM_FAST))
 302         {
 303           /*
 304            *  Unless in super-fast mode, fill old directory pages with zeroes.
 305            *  This slows us down a bit, but allows database reconstruction after
 306            *  the free list is lost.
 307            */
 308           for(page=0; page < old_dir_pages; page++)
 309             {
 310               b = pgc_get_zero(d->cache, d->fd, old_dir + (page << d->page_order));
 311               pgc_put(d->cache, b);
 312             }
 313         }
 314       sdbm_free_pages(d, old_dir, old_dir_pages);
 315     }
 316
 317   d->root->dir_order++;
 318   d->dir_size = 1 << d->root->dir_order;
 319   d->dir_shift = 32 - d->root->dir_order;
 320   pgc_mark_dirty(d->cache, d->root_page);
 321   if (!(d->flags & SDBM_FAST))
 322     sdbm_sync(d);
 323 }
 324
 325 static void
 326 sdbm_split_data(struct sdbm *d, struct sdbm_bucket *s, struct sdbm_bucket *d0, struct sdbm_bucket *d1, uns sigbit)
 327 {
 328   byte *sp = s->data;
 329   byte *dp[2] = { d0->data, d1->data };
 330   byte *K, *D;
 331   uns Kl, Dl, sz, i;
 332
 333   while (sp < s->data + s->used)
 334     {
 335       sz = sdbm_get_entry(d, sp, &K, &Kl, &D, &Dl);
 336       sp += sz;
 337       i = (sdbm_hash(K, Kl) & (1 << sigbit)) ? 1 : 0;
 338       sdbm_store_entry(d, dp[i], K, Kl, D, Dl);
 339       dp[i] += sz;
 340     }
 341   d0->used = dp[0] - d0->data;
 342   d1->used = dp[1] - d1->data;
 343 }
 344
 345 static void
 346 sdbm_split_dir(struct sdbm *d, uns dirpos, uns count, uns pos)
 347 {
 348   struct page *b;
 349   uns i;
 350
 351   count *= 4;
 352   while (count)
 353     {
 354       b = pgc_read(d->cache, d->fd, d->root->dir_start + (dirpos & ~d->page_mask));
 355       i = d->page_size - (dirpos & d->page_mask);
 356       if (i > count)
 357         i = count;
 358       count -= i;
 359       while (i)
 360         {
 361           GET32(b->data, dirpos & d->page_mask) = pos;
 362           dirpos += 4;
 363           i -= 4;
 364         }
 365       pgc_mark_dirty(d->cache, b);
 366       pgc_put(d->cache, b);
 367     }
 368 }
 369
 370 static inline uns
 371 sdbm_dirpos(struct sdbm *d, uns hash)
 372 {
 373   if (d->dir_shift != 32)               /* avoid shifting by 32 bits */
 374     return (hash >> d->dir_shift) << 2; /* offset in the directory */
 375   else
 376     return 0;
 377 }
 378
 379 static struct page *
 380 sdbm_split_page(struct sdbm *d, struct page *b, u32 hash)
 381 {
 382   struct page *p[2];
 383   uns i, rank, sigbit, rank_log, dirpos;
 384
 385   dirpos = sdbm_dirpos(d, hash);
 386   rank = sdbm_page_rank(d, dirpos);     /* rank = # of pointers to this page */
 387   if (rank == 1)
 388     {
 389       sdbm_expand_directory(d);
 390       rank = 2;
 391       dirpos *= 2;
 392     }
 393   rank_log = 1;                         /* rank_log = log2(rank) */
 394   while ((1U << rank_log) < rank)
 395     rank_log++;
 396   sigbit = d->dir_shift + rank_log - 1; /* sigbit = bit we split on */
 397   p[0] = b;
 398   p[1] = pgc_get(d->cache, d->fd, sdbm_alloc_page(d));
 399   sdbm_split_data(d, (void *) b->data, (void *) p[0]->data, (void *) p[1]->data, sigbit);
 400   sdbm_split_dir(d, (dirpos & ~(4*rank - 1))+2*rank, rank/2, p[1]->pos);
 401   pgc_mark_dirty(d->cache, p[0]);
 402   i = (hash & (1 << sigbit)) ? 1 : 0;
 403   pgc_put(d->cache, p[!i]);
 404   return p[i];
 405 }
 406
 407 static int
 408 sdbm_put_user(byte *D, uns Dl, byte *val, uns *vallen)
 409 {
 410   if (vallen)
 411     {
 412       if (*vallen < Dl)
 413         return 1;
 414       *vallen = Dl;
 415     }
 416   if (val)
 417     memcpy(val, D, Dl);
 418   return 0;
 419 }
 420
 421 static int
 422 sdbm_access(struct sdbm *d, byte *key, uns keylen, byte *val, uns *vallen, uns mode)    /* 0=read, 1=store, 2=replace */
 423 {
 424   struct page *p, *q;
 425   u32 hash, h, pos, size;
 426   struct sdbm_bucket *b;
 427   byte *c, *e;
 428   int rc;
 429
 430   if ((d->key_size >= 0 && keylen != (uns) d->key_size) || keylen > 65535)
 431     return SDBM_ERROR_BAD_KEY_SIZE;
 432   if (val && ((d->val_size >= 0 && *vallen != (uns) d->val_size) || *vallen >= 65535) && mode)
 433     return SDBM_ERROR_BAD_VAL_SIZE;
 434   if (!mode && !(d->flags & SDBM_WRITE))
 435     return SDBM_ERROR_READ_ONLY;
 436   hash = sdbm_hash(key, keylen);
 437   h = sdbm_dirpos(d, hash);
 438   p = pgc_read(d->cache, d->fd, d->root->dir_start + (h & ~d->page_mask));
 439   pos = GET32(p->data, h & d->page_mask);
 440   pgc_put(d->cache, p);
 441   q = pgc_read(d->cache, d->fd, pos);
 442   b = (void *) q->data;
 443   c = b->data;
 444   e = c + b->used;
 445   while (c < e)
 446     {
 447       byte *K, *D;
 448       uns Kl, Dl, s;
 449       s = sdbm_get_entry(d, c, &K, &Kl, &D, &Dl);
 450       if (Kl == keylen && !memcmp(K, key, Kl))
 451         {
 452           /* Gotcha! */
 453           switch (mode)
 454             {
 455             case 0:                     /* fetch: found */
 456               rc = sdbm_put_user(D, Dl, val, vallen);
 457               pgc_put(d->cache, q);
 458               return rc ? SDBM_ERROR_TOO_LARGE : 1;
 459             case 1:                     /* store: already present */
 460               pgc_put(d->cache, q);
 461               return 0;
 462             default:                    /* replace: delete the old one */
 463               memmove(c, c+s, e-(c+s));
 464               b->used -= s;
 465               goto insert;
 466             }
 467         }
 468       c += s;
 469     }
 470   if (!mode || !val)            /* fetch or delete: no success */
 471     {
 472       pgc_put(d->cache, q);
 473       return 0;
 474     }
 475
 476 insert:
 477   if (val)
 478     {
 479       size = sdbm_entry_len(d, keylen, *vallen);
 480       while (b->used + size > d->page_size - sizeof(struct sdbm_bucket))
 481         {
 482           /* Page overflow, need to split */
 483           if (size >= d->page_size - sizeof(struct sdbm_bucket))
 484             {
 485               pgc_put(d->cache, q);
 486               return SDBM_ERROR_GIANT;
 487             }
 488           q = sdbm_split_page(d, q, hash);
 489           b = (void *) q->data;
 490         }
 491       sdbm_store_entry(d, b->data + b->used, key, keylen, val, *vallen);
 492       b->used += size;
 493     }
 494   pgc_mark_dirty(d->cache, q);
 495   pgc_put(d->cache, q);
 496   if (d->flags & SDBM_SYNC)
 497     sdbm_sync(d);
 498   return 1;
 499 }
 500
 501 int
 502 sdbm_store(struct sdbm *d, byte *key, uns keylen, byte *val, uns vallen)
 503 {
 504   return sdbm_access(d, key, keylen, val, &vallen, 1);
 505 }
 506
 507 int
 508 sdbm_replace(struct sdbm *d, byte *key, uns keylen, byte *val, uns vallen)
 509 {
 510   return sdbm_access(d, key, keylen, val, &vallen, 2);
 511 }
 512
 513 int
 514 sdbm_delete(struct sdbm *d, byte *key, uns keylen)
 515 {
 516   return sdbm_access(d, key, keylen, NULL, NULL, 2);
 517 }
 518
 519 int
 520 sdbm_fetch(struct sdbm *d, byte *key, uns keylen, byte *val, uns *vallen)
 521 {
 522   return sdbm_access(d, key, keylen, val, vallen, 0);
 523 }
 524
 525 void
 526 sdbm_rewind(struct sdbm *d)
 527 {
 528   d->find_pos = d->page_size;
 529   d->find_free_list = 0;
 530 }
 531
 532 int
 533 sdbm_get_next(struct sdbm *d, byte *key, uns *keylen, byte *val, uns *vallen)
 534 {
 535   uns pos = d->find_pos;
 536   byte *K, *V;
 537   uns c, Kl, Vl;
 538   struct page *p;
 539   struct sdbm_bucket *b;
 540
 541   for(;;)
 542     {
 543       c = pos & d->page_mask;
 544       if (!c)
 545         {
 546           if (pos >= d->file_size)
 547             break;
 548           if (pos == d->root->dir_start)
 549             pos += (4*d->dir_size + d->page_size - 1) & ~d->page_mask;
 550           else if (pos == d->root->free_pool[d->find_free_list].first)
 551             pos += d->root->free_pool[d->find_free_list++].count << d->page_order;
 552           else
 553             pos += 4;
 554           continue;
 555         }
 556       p = pgc_read(d->cache, d->fd, pos & ~d->page_mask);
 557       b = (void *) p->data;
 558       if (c - 4 >= b->used)
 559         {
 560           pos = (pos & ~d->page_mask) + d->page_size;
 561           pgc_put(d->cache, p);
 562           continue;
 563         }
 564       c = sdbm_get_entry(d, p->data + c, &K, &Kl, &V, &Vl);
 565       d->find_pos = pos + c;
 566       c = sdbm_put_user(K, Kl, key, keylen) ||
 567           sdbm_put_user(V, Vl, val, vallen);
 568       pgc_put(d->cache, p);
 569       return c ? SDBM_ERROR_TOO_LARGE : 1;
 570     }
 571   d->find_pos = pos;
 572   return 0;
 573 }
 574
 575 void
 576 sdbm_sync(struct sdbm *d)
 577 {
 578   pgc_flush(d->cache);
 579   if (d->flags & SDBM_FSYNC)
 580     fsync(d->fd);
 581 }