index/recindex.c

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