index/recindex.c

   1 /* $Id: recindex.c,v 1.41 2005-01-15 19:38:26 adam Exp $
   2    Copyright (C) 1995-2005
   3    Index Data ApS
   4
   5 This file is part of the Zebra server.
   6
   7 Zebra is free software; you can redistribute it and/or modify it under
   8 the terms of the GNU General Public License as published by the Free
   9 Software Foundation; either version 2, or (at your option) any later
  10 version.
  11
  12 Zebra is distributed in the hope that it will be useful, but WITHOUT ANY
  13 WARRANTY; without even the implied warranty of MERCHANTABILITY or
  14 FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  15 for more details.
  16
  17 You should have received a copy of the GNU General Public License
  18 along with Zebra; see the file LICENSE.zebra.  If not, write to the
  19 Free Software Foundation, 59 Temple Place - Suite 330, Boston, MA
  20 02111-1307, USA.
  21 */
  22
  23 #define RIDX_CHUNK 128
  24
  25 /*
  26  *  Format of first block
  27  *      next       (8 bytes)
  28  *      ref_count  (2 bytes)
  29  *      block      (500 bytes)
  30  *
  31  *  Format of subsequent blocks
  32  *      next  (8 bytes)
  33  *      block (502 bytes)
  34  *
  35  *  Format of each record
  36  *      sysno
  37  *      (length, data) - pairs
  38  *      length = 0 if same as previous
  39  */
  40 #include <stdio.h>
  41 #include <assert.h>
  42 #include <string.h>
  43
  44 #include "recindxp.h"
  45
  46 #if HAVE_BZLIB_H
  47 #include <bzlib.h>
  48 #endif
  49 static void rec_write_head (Records p)
  50 {
  51     int r;
  52
  53     assert (p);
  54     assert (p->index_BFile);
  55
  56     r = bf_write (p->index_BFile, 0, 0, sizeof(p->head), &p->head);
  57     if (r)
  58     {
  59         yaz_log (YLOG_FATAL|YLOG_ERRNO, "write head of %s", p->index_fname);
  60         exit (1);
  61     }
  62 }
  63
  64 static void rec_tmp_expand (Records p, int size)
  65 {
  66     if (p->tmp_size < size + 2048 ||
  67         p->tmp_size < p->head.block_size[REC_BLOCK_TYPES-1]*2)
  68     {
  69         xfree (p->tmp_buf);
  70         p->tmp_size = size + (int)
  71                         (p->head.block_size[REC_BLOCK_TYPES-1])*2 + 2048;
  72         p->tmp_buf = (char *) xmalloc (p->tmp_size);
  73     }
  74 }
  75
  76 static int read_indx (Records p, SYSNO sysno, void *buf, int itemsize,
  77                       int ignoreError)
  78 {
  79     int r;
  80     zint pos = (sysno-1)*itemsize;
  81     int off = (int) (pos%RIDX_CHUNK);
  82     int sz1 = RIDX_CHUNK - off;    /* sz1 is size of buffer to read.. */
  83
  84     if (sz1 > itemsize)
  85         sz1 = itemsize;  /* no more than itemsize bytes */
  86
  87     r = bf_read (p->index_BFile, 1+pos/RIDX_CHUNK, off, sz1, buf);
  88     if (r == 1 && sz1 < itemsize) /* boundary? - must read second part */
  89         r = bf_read (p->index_BFile, 2+pos/RIDX_CHUNK, 0, itemsize - sz1,
  90                         (char*) buf + sz1);
  91     if (r != 1 && !ignoreError)
  92     {
  93         yaz_log (YLOG_FATAL|YLOG_ERRNO, "read in %s at pos %ld",
  94               p->index_fname, (long) pos);
  95         exit (1);
  96     }
  97     return r;
  98 }
  99
 100 static void write_indx (Records p, SYSNO sysno, void *buf, int itemsize)
 101 {
 102     zint pos = (sysno-1)*itemsize;
 103     int off = (int) (pos%RIDX_CHUNK);
 104     int sz1 = RIDX_CHUNK - off;    /* sz1 is size of buffer to read.. */
 105
 106     if (sz1 > itemsize)
 107         sz1 = itemsize;  /* no more than itemsize bytes */
 108
 109     bf_write(p->index_BFile, 1+pos/RIDX_CHUNK, off, sz1, buf);
 110     if (sz1 < itemsize)   /* boundary? must write second part */
 111         bf_write(p->index_BFile, 2+pos/RIDX_CHUNK, 0, itemsize - sz1,
 112                 (char*) buf + sz1);
 113 }
 114
 115 static void rec_release_blocks (Records p, SYSNO sysno)
 116 {
 117     struct record_index_entry entry;
 118     zint freeblock;
 119     char block_and_ref[sizeof(zint) + sizeof(short)];
 120     int dst_type;
 121     int first = 1;
 122
 123     if (read_indx (p, sysno, &entry, sizeof(entry), 1) != 1)
 124         return ;
 125
 126     freeblock = entry.next;
 127     assert (freeblock > 0);
 128     dst_type = (int) (freeblock & 7);
 129     assert (dst_type < REC_BLOCK_TYPES);
 130     freeblock = freeblock / 8;
 131     while (freeblock)
 132     {
 133         if (bf_read (p->data_BFile[dst_type], freeblock, 0,
 134                      first ? sizeof(block_and_ref) : sizeof(zint),
 135                      block_and_ref) != 1)
 136         {
 137             yaz_log (YLOG_FATAL|YLOG_ERRNO, "read in rec_del_single");
 138             exit (1);
 139         }
 140         if (first)
 141         {
 142             short ref;
 143             memcpy (&ref, block_and_ref + sizeof(freeblock), sizeof(ref));
 144             --ref;
 145             memcpy (block_and_ref + sizeof(freeblock), &ref, sizeof(ref));
 146             if (ref)
 147             {
 148                 if (bf_write (p->data_BFile[dst_type], freeblock, 0,
 149                               sizeof(block_and_ref), block_and_ref))
 150                 {
 151                     yaz_log (YLOG_FATAL|YLOG_ERRNO, "write in rec_del_single");
 152                     exit (1);
 153                 }
 154                 return;
 155             }
 156             first = 0;
 157         }
 158
 159         if (bf_write (p->data_BFile[dst_type], freeblock, 0, sizeof(freeblock),
 160                       &p->head.block_free[dst_type]))
 161         {
 162             yaz_log (YLOG_FATAL|YLOG_ERRNO, "write in rec_del_single");
 163             exit (1);
 164         }
 165         p->head.block_free[dst_type] = freeblock;
 166         memcpy (&freeblock, block_and_ref, sizeof(freeblock));
 167
 168         p->head.block_used[dst_type]--;
 169     }
 170     p->head.total_bytes -= entry.size;
 171 }
 172
 173 static void rec_delete_single (Records p, Record rec)
 174 {
 175     struct record_index_entry entry;
 176
 177     rec_release_blocks (p, rec->sysno);
 178
 179     entry.next = p->head.index_free;
 180     entry.size = 0;
 181     p->head.index_free = rec->sysno;
 182     write_indx (p, rec->sysno, &entry, sizeof(entry));
 183 }
 184
 185 static void rec_write_tmp_buf (Records p, int size, SYSNO *sysnos)
 186 {
 187     struct record_index_entry entry;
 188     int no_written = 0;
 189     char *cptr = p->tmp_buf;
 190     zint block_prev = -1, block_free;
 191     int dst_type = 0;
 192     int i;
 193
 194     for (i = 1; i<REC_BLOCK_TYPES; i++)
 195         if (size >= p->head.block_move[i])
 196             dst_type = i;
 197     while (no_written < size)
 198     {
 199         block_free = p->head.block_free[dst_type];
 200         if (block_free)
 201         {
 202             if (bf_read (p->data_BFile[dst_type],
 203                          block_free, 0, sizeof(*p->head.block_free),
 204                          &p->head.block_free[dst_type]) != 1)
 205             {
 206                 yaz_log (YLOG_FATAL|YLOG_ERRNO, "read in %s at free block " ZINT_FORMAT,
 207                       p->data_fname[dst_type], block_free);
 208                 exit (1);
 209             }
 210         }
 211         else
 212             block_free = p->head.block_last[dst_type]++;
 213         if (block_prev == -1)
 214         {
 215             entry.next = block_free*8 + dst_type;
 216             entry.size = size;
 217             p->head.total_bytes += size;
 218             while (*sysnos > 0)
 219             {
 220                 write_indx (p, *sysnos, &entry, sizeof(entry));
 221                 sysnos++;
 222             }
 223         }
 224         else
 225         {
 226             memcpy (cptr, &block_free, sizeof(block_free));
 227             bf_write (p->data_BFile[dst_type], block_prev, 0, 0, cptr);
 228             cptr = p->tmp_buf + no_written;
 229         }
 230         block_prev = block_free;
 231         no_written += (int)(p->head.block_size[dst_type]) - sizeof(zint);
 232         p->head.block_used[dst_type]++;
 233     }
 234     assert (block_prev != -1);
 235     block_free = 0;
 236     memcpy (cptr, &block_free, sizeof(block_free));
 237     bf_write (p->data_BFile[dst_type], block_prev, 0,
 238               sizeof(block_free) + (p->tmp_buf+size) - cptr, cptr);
 239 }
 240
 241 Records rec_open (BFiles bfs, int rw, int compression_method)
 242 {
 243     Records p;
 244     int i, r;
 245     int version;
 246
 247     p = (Records) xmalloc (sizeof(*p));
 248     p->compression_method = compression_method;
 249     p->rw = rw;
 250     p->tmp_size = 1024;
 251     p->tmp_buf = (char *) xmalloc (p->tmp_size);
 252     p->index_fname = "reci";
 253     p->index_BFile = bf_open (bfs, p->index_fname, RIDX_CHUNK, rw);
 254     if (p->index_BFile == NULL)
 255     {
 256         yaz_log (YLOG_FATAL|YLOG_ERRNO, "open %s", p->index_fname);
 257         exit (1);
 258     }
 259     r = bf_read (p->index_BFile, 0, 0, 0, p->tmp_buf);
 260     switch (r)
 261     {
 262     case 0:
 263         memcpy (p->head.magic, REC_HEAD_MAGIC, sizeof(p->head.magic));
 264         sprintf (p->head.version, "%3d", REC_VERSION);
 265         p->head.index_free = 0;
 266         p->head.index_last = 1;
 267         p->head.no_records = 0;
 268         p->head.total_bytes = 0;
 269         for (i = 0; i<REC_BLOCK_TYPES; i++)
 270         {
 271             p->head.block_free[i] = 0;
 272             p->head.block_last[i] = 1;
 273             p->head.block_used[i] = 0;
 274         }
 275         p->head.block_size[0] = 128;
 276         p->head.block_move[0] = 0;
 277         for (i = 1; i<REC_BLOCK_TYPES; i++)
 278         {
 279             p->head.block_size[i] = p->head.block_size[i-1] * 4;
 280             p->head.block_move[i] = p->head.block_size[i] * 24;
 281         }
 282         if (rw)
 283             rec_write_head (p);
 284         break;
 285     case 1:
 286         memcpy (&p->head, p->tmp_buf, sizeof(p->head));
 287         if (memcmp (p->head.magic, REC_HEAD_MAGIC, sizeof(p->head.magic)))
 288         {
 289             yaz_log (YLOG_FATAL, "file %s has bad format", p->index_fname);
 290             exit (1);
 291         }
 292         version = atoi (p->head.version);
 293         if (version != REC_VERSION)
 294         {
 295             yaz_log (YLOG_FATAL, "file %s is version %d, but version"
 296                   " %d is required", p->index_fname, version, REC_VERSION);
 297             exit (1);
 298         }
 299         break;
 300     }
 301     for (i = 0; i<REC_BLOCK_TYPES; i++)
 302     {
 303         char str[80];
 304         sprintf (str, "recd%c", i + 'A');
 305         p->data_fname[i] = (char *) xmalloc (strlen(str)+1);
 306         strcpy (p->data_fname[i], str);
 307         p->data_BFile[i] = NULL;
 308     }
 309     for (i = 0; i<REC_BLOCK_TYPES; i++)
 310     {
 311         if (!(p->data_BFile[i] = bf_open (bfs, p->data_fname[i],
 312                                           (int) (p->head.block_size[i]),
 313                                           rw)))
 314         {
 315             yaz_log (YLOG_FATAL|YLOG_ERRNO, "bf_open %s", p->data_fname[i]);
 316             exit (1);
 317         }
 318     }
 319     p->cache_max = 400;
 320     p->cache_cur = 0;
 321     p->record_cache = (struct record_cache_entry *)
 322         xmalloc (sizeof(*p->record_cache)*p->cache_max);
 323     zebra_mutex_init (&p->mutex);
 324     return p;
 325 }
 326
 327 static void rec_encode_unsigned (unsigned n, unsigned char *buf, int *len)
 328 {
 329     (*len) = 0;
 330     while (n > 127)
 331     {
 332         buf[*len] = 128 + (n & 127);
 333         n = n >> 7;
 334         (*len)++;
 335     }
 336     buf[*len] = n;
 337     (*len)++;
 338 }
 339
 340 static void rec_decode_unsigned(unsigned *np, unsigned char *buf, int *len)
 341 {
 342     unsigned n = 0;
 343     unsigned w = 1;
 344     (*len) = 0;
 345
 346     while (buf[*len] > 127)
 347     {
 348         n += w*(buf[*len] & 127);
 349         w = w << 7;
 350         (*len)++;
 351     }
 352     n += w * buf[*len];
 353     (*len)++;
 354     *np = n;
 355 }
 356
 357 static void rec_encode_zint (zint n, unsigned char *buf, int *len)
 358 {
 359     (*len) = 0;
 360     while (n > 127)
 361     {
 362         buf[*len] = (unsigned) (128 + (n & 127));
 363         n = n >> 7;
 364         (*len)++;
 365     }
 366     buf[*len] = (unsigned) n;
 367     (*len)++;
 368 }
 369
 370 static void rec_decode_zint(zint *np, unsigned char *buf, int *len)
 371 {
 372     zint  n = 0;
 373     zint w = 1;
 374     (*len) = 0;
 375
 376     while (buf[*len] > 127)
 377     {
 378         n += w*(buf[*len] & 127);
 379         w = w << 7;
 380         (*len)++;
 381     }
 382     n += w * buf[*len];
 383     (*len)++;
 384     *np = n;
 385 }
 386
 387 static void rec_cache_flush_block1 (Records p, Record rec, Record last_rec,
 388                                     char **out_buf, int *out_size,
 389                                     int *out_offset)
 390 {
 391     int i;
 392     int len;
 393
 394     for (i = 0; i<REC_NO_INFO; i++)
 395     {
 396         if (*out_offset + (int) rec->size[i] + 20 > *out_size)
 397         {
 398             int new_size = *out_offset + rec->size[i] + 65536;
 399             char *np = (char *) xmalloc (new_size);
 400             if (*out_offset)
 401                 memcpy (np, *out_buf, *out_offset);
 402             xfree (*out_buf);
 403             *out_size = new_size;
 404             *out_buf = np;
 405         }
 406         if (i == 0)
 407         {
 408             rec_encode_zint (rec->sysno, *out_buf + *out_offset, &len);
 409             (*out_offset) += len;
 410         }
 411         if (rec->size[i] == 0)
 412         {
 413             rec_encode_unsigned (1, *out_buf + *out_offset, &len);
 414             (*out_offset) += len;
 415         }
 416         else if (last_rec && rec->size[i] == last_rec->size[i] &&
 417                  !memcmp (rec->info[i], last_rec->info[i], rec->size[i]))
 418         {
 419             rec_encode_unsigned (0, *out_buf + *out_offset, &len);
 420             (*out_offset) += len;
 421         }
 422         else
 423         {
 424             rec_encode_unsigned (rec->size[i]+1, *out_buf + *out_offset, &len);
 425             (*out_offset) += len;
 426             memcpy (*out_buf + *out_offset, rec->info[i], rec->size[i]);
 427             (*out_offset) += rec->size[i];
 428         }
 429     }
 430 }
 431
 432 static void rec_write_multiple (Records p, int saveCount)
 433 {
 434     int i;
 435     short ref_count = 0;
 436     char compression_method;
 437     Record last_rec = 0;
 438     int out_size = 1000;
 439     int out_offset = 0;
 440     char *out_buf = (char *) xmalloc (out_size);
 441     SYSNO *sysnos = (SYSNO *) xmalloc (sizeof(*sysnos) * (p->cache_cur + 1));
 442     SYSNO *sysnop = sysnos;
 443
 444     for (i = 0; i<p->cache_cur - saveCount; i++)
 445     {
 446         struct record_cache_entry *e = p->record_cache + i;
 447         switch (e->flag)
 448         {
 449         case recordFlagNew:
 450             rec_cache_flush_block1 (p, e->rec, last_rec, &out_buf,
 451                                     &out_size, &out_offset);
 452             *sysnop++ = e->rec->sysno;
 453             ref_count++;
 454             e->flag = recordFlagNop;
 455             last_rec = e->rec;
 456             break;
 457         case recordFlagWrite:
 458             rec_release_blocks (p, e->rec->sysno);
 459             rec_cache_flush_block1 (p, e->rec, last_rec, &out_buf,
 460                                     &out_size, &out_offset);
 461             *sysnop++ = e->rec->sysno;
 462             ref_count++;
 463             e->flag = recordFlagNop;
 464             last_rec = e->rec;
 465             break;
 466         case recordFlagDelete:
 467             rec_delete_single (p, e->rec);
 468             e->flag = recordFlagNop;
 469             break;
 470         default:
 471             break;
 472         }
 473     }
 474
 475     *sysnop = -1;
 476     if (ref_count)
 477     {
 478         int csize = 0;  /* indicate compression "not performed yet" */
 479         compression_method = p->compression_method;
 480         switch (compression_method)
 481         {
 482         case REC_COMPRESS_BZIP2:
 483 #if HAVE_BZLIB_H
 484             csize = out_offset + (out_offset >> 6) + 620;
 485             rec_tmp_expand (p, csize);
 486 #ifdef BZ_CONFIG_ERROR
 487             i = BZ2_bzBuffToBuffCompress
 488 #else
 489             i = bzBuffToBuffCompress
 490 #endif
 491                                      (p->tmp_buf+sizeof(zint)+sizeof(short)+
 492                                       sizeof(char),
 493                                       &csize, out_buf, out_offset, 1, 0, 30);
 494             if (i != BZ_OK)
 495             {
 496                 yaz_log (YLOG_WARN, "bzBuffToBuffCompress error code=%d", i);
 497                 csize = 0;
 498             }
 499             yaz_log (YLOG_LOG, "compress %4d %5d %5d", ref_count, out_offset,
 500                   csize);
 501 #endif
 502             break;
 503         case REC_COMPRESS_NONE:
 504             break;
 505         }
 506         if (!csize)
 507         {
 508             /* either no compression or compression not supported ... */
 509             csize = out_offset;
 510             rec_tmp_expand (p, csize);
 511             memcpy (p->tmp_buf + sizeof(zint) + sizeof(short) + sizeof(char),
 512                     out_buf, out_offset);
 513             csize = out_offset;
 514             compression_method = REC_COMPRESS_NONE;
 515         }
 516         memcpy (p->tmp_buf + sizeof(zint), &ref_count, sizeof(ref_count));
 517         memcpy (p->tmp_buf + sizeof(zint)+sizeof(short),
 518                 &compression_method, sizeof(compression_method));
 519
 520         /* -------- compression */
 521         rec_write_tmp_buf (p, csize + sizeof(short) + sizeof(char), sysnos);
 522     }
 523     xfree (out_buf);
 524     xfree (sysnos);
 525 }
 526
 527 static void rec_cache_flush (Records p, int saveCount)
 528 {
 529     int i, j;
 530
 531     if (saveCount >= p->cache_cur)
 532         saveCount = 0;
 533
 534     rec_write_multiple (p, saveCount);
 535
 536     for (i = 0; i<p->cache_cur - saveCount; i++)
 537     {
 538         struct record_cache_entry *e = p->record_cache + i;
 539         rec_rm (&e->rec);
 540     }
 541     /* i still being used ... */
 542     for (j = 0; j<saveCount; j++, i++)
 543         memcpy (p->record_cache+j, p->record_cache+i,
 544                 sizeof(*p->record_cache));
 545     p->cache_cur = saveCount;
 546 }
 547
 548 static Record *rec_cache_lookup (Records p, SYSNO sysno,
 549                                  enum recordCacheFlag flag)
 550 {
 551     int i;
 552     for (i = 0; i<p->cache_cur; i++)
 553     {
 554         struct record_cache_entry *e = p->record_cache + i;
 555         if (e->rec->sysno == sysno)
 556         {
 557             if (flag != recordFlagNop && e->flag == recordFlagNop)
 558                 e->flag = flag;
 559             return &e->rec;
 560         }
 561     }
 562     return NULL;
 563 }
 564
 565 static void rec_cache_insert (Records p, Record rec, enum recordCacheFlag flag)
 566 {
 567     struct record_cache_entry *e;
 568
 569     if (p->cache_cur == p->cache_max)
 570         rec_cache_flush (p, 1);
 571     else if (p->cache_cur > 0)
 572     {
 573         int i, j;
 574         int used = 0;
 575         for (i = 0; i<p->cache_cur; i++)
 576         {
 577             Record r = (p->record_cache + i)->rec;
 578             for (j = 0; j<REC_NO_INFO; j++)
 579                 used += r->size[j];
 580         }
 581         if (used > 90000)
 582             rec_cache_flush (p, 1);
 583     }
 584     assert (p->cache_cur < p->cache_max);
 585
 586     e = p->record_cache + (p->cache_cur)++;
 587     e->flag = flag;
 588     e->rec = rec_cp (rec);
 589 }
 590
 591 void rec_close (Records *pp)
 592 {
 593     Records p = *pp;
 594     int i;
 595
 596     assert (p);
 597
 598     zebra_mutex_destroy (&p->mutex);
 599     rec_cache_flush (p, 0);
 600     xfree (p->record_cache);
 601
 602     if (p->rw)
 603         rec_write_head (p);
 604
 605     if (p->index_BFile)
 606         bf_close (p->index_BFile);
 607
 608     for (i = 0; i<REC_BLOCK_TYPES; i++)
 609     {
 610         if (p->data_BFile[i])
 611             bf_close (p->data_BFile[i]);
 612         xfree (p->data_fname[i]);
 613     }
 614     xfree (p->tmp_buf);
 615     xfree (p);
 616     *pp = NULL;
 617 }
 618
 619 static Record rec_get_int (Records p, SYSNO sysno)
 620 {
 621     int i, in_size, r;
 622     Record rec, *recp;
 623     struct record_index_entry entry;
 624     zint freeblock;
 625     int dst_type;
 626     char *nptr, *cptr;
 627     char *in_buf = 0;
 628     char *bz_buf = 0;
 629 #if HAVE_BZLIB_H
 630     int bz_size;
 631 #endif
 632     char compression_method;
 633
 634     assert (sysno > 0);
 635     assert (p);
 636
 637     if ((recp = rec_cache_lookup (p, sysno, recordFlagNop)))
 638         return rec_cp (*recp);
 639
 640     if (read_indx (p, sysno, &entry, sizeof(entry), 1) < 1)
 641         return NULL;       /* record is not there! */
 642
 643     if (!entry.size)
 644         return NULL;       /* record is deleted */
 645
 646     dst_type = (int) (entry.next & 7);
 647     assert (dst_type < REC_BLOCK_TYPES);
 648     freeblock = entry.next / 8;
 649
 650     assert (freeblock > 0);
 651
 652     rec_tmp_expand (p, entry.size);
 653
 654     cptr = p->tmp_buf;
 655     r = bf_read (p->data_BFile[dst_type], freeblock, 0, 0, cptr);
 656     if (r < 0)
 657         return 0;
 658     memcpy (&freeblock, cptr, sizeof(freeblock));
 659
 660     while (freeblock)
 661     {
 662         zint tmp;
 663
 664         cptr += p->head.block_size[dst_type] - sizeof(freeblock);
 665
 666         memcpy (&tmp, cptr, sizeof(tmp));
 667         r = bf_read (p->data_BFile[dst_type], freeblock, 0, 0, cptr);
 668         if (r < 0)
 669             return 0;
 670         memcpy (&freeblock, cptr, sizeof(freeblock));
 671         memcpy (cptr, &tmp, sizeof(tmp));
 672     }
 673
 674     rec = (Record) xmalloc (sizeof(*rec));
 675     rec->sysno = sysno;
 676     memcpy (&compression_method, p->tmp_buf + sizeof(zint) + sizeof(short),
 677             sizeof(compression_method));
 678     in_buf = p->tmp_buf + sizeof(zint) + sizeof(short) + sizeof(char);
 679     in_size = entry.size - sizeof(short) - sizeof(char);
 680     switch (compression_method)
 681     {
 682     case REC_COMPRESS_BZIP2:
 683 #if HAVE_BZLIB_H
 684         bz_size = entry.size * 20 + 100;
 685         while (1)
 686         {
 687             bz_buf = (char *) xmalloc (bz_size);
 688 #ifdef BZ_CONFIG_ERROR
 689             i = BZ2_bzBuffToBuffDecompress
 690 #else
 691             i = bzBuffToBuffDecompress
 692 #endif
 693                  (bz_buf, &bz_size, in_buf, in_size, 0, 0);
 694             yaz_log (YLOG_LOG, "decompress %5d %5d", in_size, bz_size);
 695             if (i == BZ_OK)
 696                 break;
 697             yaz_log (YLOG_LOG, "failed");
 698             xfree (bz_buf);
 699             bz_size *= 2;
 700         }
 701         in_buf = bz_buf;
 702         in_size = bz_size;
 703 #else
 704         yaz_log (YLOG_FATAL, "cannot decompress record(s) in BZIP2 format");
 705         exit (1);
 706 #endif
 707         break;
 708     case REC_COMPRESS_NONE:
 709         break;
 710     }
 711     for (i = 0; i<REC_NO_INFO; i++)
 712         rec->info[i] = 0;
 713
 714     nptr = in_buf;                /* skip ref count */
 715     while (nptr < in_buf + in_size)
 716     {
 717         zint this_sysno;
 718         int len;
 719         rec_decode_zint (&this_sysno, nptr, &len);
 720         nptr += len;
 721
 722         for (i = 0; i < REC_NO_INFO; i++)
 723         {
 724             int this_size;
 725             rec_decode_unsigned (&this_size, nptr, &len);
 726             nptr += len;
 727
 728             if (this_size == 0)
 729                 continue;
 730             rec->size[i] = this_size-1;
 731
 732             if (rec->size[i])
 733             {
 734                 rec->info[i] = nptr;
 735                 nptr += rec->size[i];
 736             }
 737             else
 738                 rec->info[i] = NULL;
 739         }
 740         if (this_sysno == sysno)
 741             break;
 742     }
 743     for (i = 0; i<REC_NO_INFO; i++)
 744     {
 745         if (rec->info[i] && rec->size[i])
 746         {
 747             char *np = xmalloc (rec->size[i]+1);
 748             memcpy (np, rec->info[i], rec->size[i]);
 749             np[rec->size[i]] = '\0';
 750             rec->info[i] = np;
 751         }
 752         else
 753         {
 754             assert (rec->info[i] == 0);
 755             assert (rec->size[i] == 0);
 756         }
 757     }
 758     xfree (bz_buf);
 759     rec_cache_insert (p, rec, recordFlagNop);
 760     return rec;
 761 }
 762
 763 Record rec_get (Records p, SYSNO sysno)
 764 {
 765     Record rec;
 766     zebra_mutex_lock (&p->mutex);
 767
 768     rec = rec_get_int (p, sysno);
 769     zebra_mutex_unlock (&p->mutex);
 770     return rec;
 771 }
 772
 773 static Record rec_new_int (Records p)
 774 {
 775     int i;
 776     SYSNO sysno;
 777     Record rec;
 778
 779     assert (p);
 780     rec = (Record) xmalloc (sizeof(*rec));
 781     if (1 || p->head.index_free == 0)
 782         sysno = (p->head.index_last)++;
 783     else
 784     {
 785         struct record_index_entry entry;
 786
 787         read_indx (p, p->head.index_free, &entry, sizeof(entry), 0);
 788         sysno = p->head.index_free;
 789         p->head.index_free = entry.next;
 790     }
 791     (p->head.no_records)++;
 792     rec->sysno = sysno;
 793     for (i = 0; i < REC_NO_INFO; i++)
 794     {
 795         rec->info[i] = NULL;
 796         rec->size[i] = 0;
 797     }
 798     rec_cache_insert (p, rec, recordFlagNew);
 799     return rec;
 800 }
 801
 802 Record rec_new (Records p)
 803 {
 804     Record rec;
 805     zebra_mutex_lock (&p->mutex);
 806
 807     rec = rec_new_int (p);
 808     zebra_mutex_unlock (&p->mutex);
 809     return rec;
 810 }
 811
 812 void rec_del (Records p, Record *recpp)
 813 {
 814     Record *recp;
 815
 816     zebra_mutex_lock (&p->mutex);
 817     (p->head.no_records)--;
 818     if ((recp = rec_cache_lookup (p, (*recpp)->sysno, recordFlagDelete)))
 819     {
 820         rec_rm (recp);
 821         *recp = *recpp;
 822     }
 823     else
 824     {
 825         rec_cache_insert (p, *recpp, recordFlagDelete);
 826         rec_rm (recpp);
 827     }
 828     zebra_mutex_unlock (&p->mutex);
 829     *recpp = NULL;
 830 }
 831
 832 void rec_put (Records p, Record *recpp)
 833 {
 834     Record *recp;
 835
 836     zebra_mutex_lock (&p->mutex);
 837     if ((recp = rec_cache_lookup (p, (*recpp)->sysno, recordFlagWrite)))
 838     {
 839         rec_rm (recp);
 840         *recp = *recpp;
 841     }
 842     else
 843     {
 844         rec_cache_insert (p, *recpp, recordFlagWrite);
 845         rec_rm (recpp);
 846     }
 847     zebra_mutex_unlock (&p->mutex);
 848     *recpp = NULL;
 849 }
 850
 851 void rec_rm (Record *recpp)
 852 {
 853     int i;
 854
 855     if (!*recpp)
 856         return ;
 857     for (i = 0; i < REC_NO_INFO; i++)
 858         xfree ((*recpp)->info[i]);
 859     xfree (*recpp);
 860     *recpp = NULL;
 861 }
 862
 863 Record rec_cp (Record rec)
 864 {
 865     Record n;
 866     int i;
 867
 868     n = (Record) xmalloc (sizeof(*n));
 869     n->sysno = rec->sysno;
 870     for (i = 0; i < REC_NO_INFO; i++)
 871         if (!rec->info[i])
 872         {
 873             n->info[i] = NULL;
 874             n->size[i] = 0;
 875         }
 876         else
 877         {
 878             n->size[i] = rec->size[i];
 879             n->info[i] = (char *) xmalloc (rec->size[i]);
 880             memcpy (n->info[i], rec->info[i], rec->size[i]);
 881         }
 882     return n;
 883 }
 884
 885
 886 char *rec_strdup (const char *s, size_t *len)
 887 {
 888     char *p;
 889
 890     if (!s)
 891     {
 892         *len = 0;
 893         return NULL;
 894     }
 895     *len = strlen(s)+1;
 896     p = (char *) xmalloc (*len);
 897     strcpy (p, s);
 898     return p;
 899 }
 900