lib/db.c

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