index/records.c

   1 /* $Id: records.c,v 1.4 2008-03-05 09:18:51 adam Exp $
   2    Copyright (C) 1995-2008
   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 this program; if not, write to the Free Software
  19 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  20
  21 */
  22
  23 /*
  24  *  Format of first block
  25  *      next       (8 bytes)
  26  *      ref_count  (2 bytes)
  27  *      block      (500 bytes)
  28  *
  29  *  Format of subsequent blocks
  30  *      next  (8 bytes)
  31  *      block (502 bytes)
  32  *
  33  *  Format of each record
  34  *      sysno
  35  *      (length, data) - pairs
  36  *      length = 0 if same as previous
  37  */
  38 #include <stdio.h>
  39 #include <stdlib.h>
  40 #include <assert.h>
  41 #include <string.h>
  42
  43 #include <yaz/yaz-util.h>
  44 #include <idzebra/bfile.h>
  45 #include "recindex.h"
  46
  47 #if HAVE_BZLIB_H
  48 #include <bzlib.h>
  49 #endif
  50
  51 #define REC_BLOCK_TYPES 2
  52 #define REC_HEAD_MAGIC "recindex"
  53 #define REC_VERSION 5
  54
  55 struct records_info {
  56     int rw;
  57     int compression_method;
  58
  59     recindex_t recindex;
  60
  61     char *data_fname[REC_BLOCK_TYPES];
  62     BFile data_BFile[REC_BLOCK_TYPES];
  63
  64     char *tmp_buf;
  65     int tmp_size;
  66
  67     struct record_cache_entry *record_cache;
  68     int cache_size;
  69     int cache_cur;
  70     int cache_max;
  71
  72     int compression_chunk_size;
  73
  74     Zebra_mutex mutex;
  75
  76     struct records_head {
  77         char magic[8];
  78         char version[4];
  79         zint block_size[REC_BLOCK_TYPES];
  80         zint block_free[REC_BLOCK_TYPES];
  81         zint block_last[REC_BLOCK_TYPES];
  82         zint block_used[REC_BLOCK_TYPES];
  83         zint block_move[REC_BLOCK_TYPES];
  84
  85         zint total_bytes;
  86         zint index_last;
  87         zint index_free;
  88         zint no_records;
  89
  90     } head;
  91 };
  92
  93 enum recordCacheFlag { recordFlagNop, recordFlagWrite, recordFlagNew,
  94                        recordFlagDelete };
  95
  96 struct record_cache_entry {
  97     Record rec;
  98     enum recordCacheFlag flag;
  99 };
 100
 101 struct record_index_entry {
 102     zint next;         /* first block of record info / next free entry */
 103     int size;          /* size of record or 0 if free entry */
 104 };
 105
 106 Record rec_cp(Record rec);
 107
 108 /* Modify argument to if below: 1=normal, 0=sysno testing */
 109 #if 1
 110 /* If this is used sysno are not converted (no testing) */
 111 #define FAKE_OFFSET 0
 112 #define USUAL_RANGE 6000000000LL
 113
 114 #else
 115 /* Use a fake > 2^32 offset so we can test for proper 64-bit handling */
 116 #define FAKE_OFFSET 6000000000LL
 117 #define USUAL_RANGE 2000000000LL
 118 #endif
 119
 120 static zint rec_sysno_to_ext(zint sysno)
 121 {
 122     assert(sysno >= 0 && sysno <= USUAL_RANGE);
 123     return sysno + FAKE_OFFSET;
 124 }
 125
 126 zint rec_sysno_to_int(zint sysno)
 127 {
 128     assert(sysno >= FAKE_OFFSET && sysno <= FAKE_OFFSET + USUAL_RANGE);
 129     return sysno - FAKE_OFFSET;
 130 }
 131
 132 static void rec_tmp_expand(Records p, int size)
 133 {
 134     if (p->tmp_size < size + 2048 ||
 135         p->tmp_size < p->head.block_size[REC_BLOCK_TYPES-1]*2)
 136     {
 137         xfree(p->tmp_buf);
 138         p->tmp_size = size + (int)
 139                         (p->head.block_size[REC_BLOCK_TYPES-1])*2 + 2048;
 140         p->tmp_buf = (char *) xmalloc(p->tmp_size);
 141     }
 142 }
 143
 144 static ZEBRA_RES rec_release_blocks(Records p, zint sysno)
 145 {
 146     struct record_index_entry entry;
 147     zint freeblock;
 148     char block_and_ref[sizeof(zint) + sizeof(short)];
 149     int dst_type;
 150     int first = 1;
 151
 152     if (recindex_read_indx(p->recindex, sysno, &entry, sizeof(entry), 1) != 1)
 153         return ZEBRA_FAIL;
 154
 155     freeblock = entry.next;
 156     assert(freeblock > 0);
 157     dst_type = CAST_ZINT_TO_INT(freeblock & 7);
 158     assert(dst_type < REC_BLOCK_TYPES);
 159     freeblock = freeblock / 8;
 160     while (freeblock)
 161     {
 162         if (bf_read(p->data_BFile[dst_type], freeblock, 0,
 163                      first ? sizeof(block_and_ref) : sizeof(zint),
 164                      block_and_ref) != 1)
 165         {
 166             yaz_log(YLOG_FATAL|YLOG_ERRNO, "read in rec_del_single");
 167             return ZEBRA_FAIL;
 168         }
 169         if (first)
 170         {
 171             short ref;
 172             memcpy(&ref, block_and_ref + sizeof(freeblock), sizeof(ref));
 173             --ref;
 174             memcpy(block_and_ref + sizeof(freeblock), &ref, sizeof(ref));
 175             if (ref)
 176             {
 177                 if (bf_write(p->data_BFile[dst_type], freeblock, 0,
 178                               sizeof(block_and_ref), block_and_ref))
 179                 {
 180                     yaz_log(YLOG_FATAL|YLOG_ERRNO, "write in rec_del_single");
 181                     return ZEBRA_FAIL;
 182                 }
 183                 return ZEBRA_OK;
 184             }
 185             first = 0;
 186         }
 187
 188         if (bf_write(p->data_BFile[dst_type], freeblock, 0, sizeof(freeblock),
 189                       &p->head.block_free[dst_type]))
 190         {
 191             yaz_log(YLOG_FATAL|YLOG_ERRNO, "write in rec_del_single");
 192             return ZEBRA_FAIL;
 193         }
 194         p->head.block_free[dst_type] = freeblock;
 195         memcpy(&freeblock, block_and_ref, sizeof(freeblock));
 196
 197         p->head.block_used[dst_type]--;
 198     }
 199     p->head.total_bytes -= entry.size;
 200     return ZEBRA_OK;
 201 }
 202
 203 static ZEBRA_RES rec_delete_single(Records p, Record rec)
 204 {
 205     struct record_index_entry entry;
 206
 207     /* all data in entry must be reset, since it's written verbatim */
 208     memset(&entry, '\0', sizeof(entry));
 209     if (rec_release_blocks(p, rec_sysno_to_int(rec->sysno)) != ZEBRA_OK)
 210         return ZEBRA_FAIL;
 211
 212     entry.next = p->head.index_free;
 213     entry.size = 0;
 214     p->head.index_free = rec_sysno_to_int(rec->sysno);
 215     recindex_write_indx(p->recindex, rec_sysno_to_int(rec->sysno), &entry, sizeof(entry));
 216     return ZEBRA_OK;
 217 }
 218
 219 static ZEBRA_RES rec_write_tmp_buf(Records p, int size, zint *sysnos)
 220 {
 221     struct record_index_entry entry;
 222     int no_written = 0;
 223     char *cptr = p->tmp_buf;
 224     zint block_prev = -1, block_free;
 225     int dst_type = 0;
 226     int i;
 227
 228     /* all data in entry must be reset, since it's written verbatim */
 229     memset(&entry, '\0', sizeof(entry));
 230
 231     for (i = 1; i<REC_BLOCK_TYPES; i++)
 232         if (size >= p->head.block_move[i])
 233             dst_type = i;
 234     while (no_written < size)
 235     {
 236         block_free = p->head.block_free[dst_type];
 237         if (block_free)
 238         {
 239             if (bf_read(p->data_BFile[dst_type],
 240                          block_free, 0, sizeof(*p->head.block_free),
 241                          &p->head.block_free[dst_type]) != 1)
 242             {
 243                 yaz_log(YLOG_FATAL|YLOG_ERRNO, "read in %s at free block "
 244                          ZINT_FORMAT,
 245                          p->data_fname[dst_type], block_free);
 246                 return ZEBRA_FAIL;
 247             }
 248         }
 249         else
 250             block_free = p->head.block_last[dst_type]++;
 251         if (block_prev == -1)
 252         {
 253             entry.next = block_free*8 + dst_type;
 254             entry.size = size;
 255             p->head.total_bytes += size;
 256             while (*sysnos > 0)
 257             {
 258                 recindex_write_indx(p->recindex, *sysnos, &entry, sizeof(entry));
 259                 sysnos++;
 260             }
 261         }
 262         else
 263         {
 264             memcpy(cptr, &block_free, sizeof(block_free));
 265             bf_write(p->data_BFile[dst_type], block_prev, 0, 0, cptr);
 266             cptr = p->tmp_buf + no_written;
 267         }
 268         block_prev = block_free;
 269         no_written += CAST_ZINT_TO_INT(p->head.block_size[dst_type])
 270             - sizeof(zint);
 271         p->head.block_used[dst_type]++;
 272     }
 273     assert(block_prev != -1);
 274     block_free = 0;
 275     memcpy(cptr, &block_free, sizeof(block_free));
 276     bf_write(p->data_BFile[dst_type], block_prev, 0,
 277               sizeof(block_free) + (p->tmp_buf+size) - cptr, cptr);
 278     return ZEBRA_OK;
 279 }
 280
 281 Records rec_open(BFiles bfs, int rw, int compression_method)
 282 {
 283     Records p;
 284     int i, r;
 285     int version;
 286     ZEBRA_RES ret = ZEBRA_OK;
 287
 288     p = (Records) xmalloc(sizeof(*p));
 289     memset(&p->head, '\0', sizeof(p->head));
 290     p->compression_method = compression_method;
 291     p->rw = rw;
 292     p->tmp_size = 1024;
 293     p->tmp_buf = (char *) xmalloc(p->tmp_size);
 294     p->compression_chunk_size = 0;
 295     p->recindex = recindex_open(bfs, rw, 0 /* 1=isamb for recindex */);
 296     r = recindex_read_head(p->recindex, p->tmp_buf);
 297     switch (r)
 298     {
 299     case 0:
 300         memcpy(p->head.magic, REC_HEAD_MAGIC, sizeof(p->head.magic));
 301         sprintf(p->head.version, "%3d", REC_VERSION);
 302         p->head.index_free = 0;
 303         p->head.index_last = 1;
 304         p->head.no_records = 0;
 305         p->head.total_bytes = 0;
 306         for (i = 0; i<REC_BLOCK_TYPES; i++)
 307         {
 308             p->head.block_free[i] = 0;
 309             p->head.block_last[i] = 1;
 310             p->head.block_used[i] = 0;
 311         }
 312         p->head.block_size[0] = 128;
 313         p->head.block_move[0] = 0;
 314         for (i = 1; i<REC_BLOCK_TYPES; i++)
 315         {
 316             p->head.block_size[i] = p->head.block_size[i-1] * 4;
 317             p->head.block_move[i] = p->head.block_size[i] * 24;
 318         }
 319         if (rw)
 320         {
 321             if (recindex_write_head(p->recindex,
 322                                     &p->head, sizeof(p->head)) != ZEBRA_OK)
 323                 ret = ZEBRA_FAIL;
 324         }
 325         break;
 326     case 1:
 327         memcpy(&p->head, p->tmp_buf, sizeof(p->head));
 328         if (memcmp(p->head.magic, REC_HEAD_MAGIC, sizeof(p->head.magic)))
 329         {
 330             yaz_log(YLOG_FATAL, "file %s has bad format",
 331                     recindex_get_fname(p->recindex));
 332             ret = ZEBRA_FAIL;
 333         }
 334         version = atoi(p->head.version);
 335         if (version != REC_VERSION)
 336         {
 337             yaz_log(YLOG_FATAL, "file %s is version %d, but version"
 338                   " %d is required",
 339                     recindex_get_fname(p->recindex), version, REC_VERSION);
 340             ret = ZEBRA_FAIL;
 341         }
 342         p->compression_chunk_size = 90000; /* good for BZIP2 */
 343         break;
 344     }
 345     for (i = 0; i<REC_BLOCK_TYPES; i++)
 346     {
 347         char str[80];
 348         sprintf(str, "recd%c", i + 'A');
 349         p->data_fname[i] = (char *) xmalloc(strlen(str)+1);
 350         strcpy(p->data_fname[i], str);
 351         p->data_BFile[i] = NULL;
 352     }
 353     for (i = 0; i<REC_BLOCK_TYPES; i++)
 354     {
 355         if (!(p->data_BFile[i] =
 356               bf_open(bfs, p->data_fname[i],
 357                       CAST_ZINT_TO_INT(p->head.block_size[i]), rw)))
 358         {
 359             yaz_log(YLOG_FATAL|YLOG_ERRNO, "bf_open %s", p->data_fname[i]);
 360             ret = ZEBRA_FAIL;
 361             break;
 362         }
 363     }
 364     p->cache_max = 400;
 365     p->cache_cur = 0;
 366     p->record_cache = (struct record_cache_entry *)
 367         xmalloc(sizeof(*p->record_cache)*p->cache_max);
 368     zebra_mutex_init(&p->mutex);
 369     if (ret == ZEBRA_FAIL)
 370         rec_close(&p);
 371     return p;
 372 }
 373
 374 static void rec_encode_unsigned(unsigned n, unsigned char *buf, int *len)
 375 {
 376     (*len) = 0;
 377     while (n > 127)
 378     {
 379         buf[*len] = 128 + (n & 127);
 380         n = n >> 7;
 381         (*len)++;
 382     }
 383     buf[*len] = n;
 384     (*len)++;
 385 }
 386
 387 static void rec_decode_unsigned(unsigned *np, unsigned char *buf, int *len)
 388 {
 389     unsigned n = 0;
 390     unsigned w = 1;
 391     (*len) = 0;
 392
 393     while (buf[*len] > 127)
 394     {
 395         n += w*(buf[*len] & 127);
 396         w = w << 7;
 397         (*len)++;
 398     }
 399     n += w * buf[*len];
 400     (*len)++;
 401     *np = n;
 402 }
 403
 404 static void rec_encode_zint(zint n, unsigned char *buf, int *len)
 405 {
 406     (*len) = 0;
 407     while (n > 127)
 408     {
 409         buf[*len] = (unsigned) (128 + (n & 127));
 410         n = n >> 7;
 411         (*len)++;
 412     }
 413     buf[*len] = (unsigned) n;
 414     (*len)++;
 415 }
 416
 417 static void rec_decode_zint(zint *np, unsigned char *buf, int *len)
 418 {
 419     zint  n = 0;
 420     zint w = 1;
 421     (*len) = 0;
 422
 423     while (buf[*len] > 127)
 424     {
 425         n += w*(buf[*len] & 127);
 426         w = w << 7;
 427         (*len)++;
 428     }
 429     n += w * buf[*len];
 430     (*len)++;
 431     *np = n;
 432 }
 433
 434 static void rec_cache_flush_block1(Records p, Record rec, Record last_rec,
 435                                    char **out_buf, int *out_size,
 436                                    int *out_offset)
 437 {
 438     int i;
 439     int len;
 440
 441     for (i = 0; i<REC_NO_INFO; i++)
 442     {
 443         if (*out_offset + CAST_ZINT_TO_INT(rec->size[i]) + 20 > *out_size)
 444         {
 445             int new_size = *out_offset + rec->size[i] + 65536;
 446             char *np = (char *) xmalloc(new_size);
 447             if (*out_offset)
 448                 memcpy(np, *out_buf, *out_offset);
 449             xfree(*out_buf);
 450             *out_size = new_size;
 451             *out_buf = np;
 452         }
 453         if (i == 0)
 454         {
 455             rec_encode_zint(rec_sysno_to_int(rec->sysno),
 456                             (unsigned char *) *out_buf + *out_offset, &len);
 457             (*out_offset) += len;
 458         }
 459         if (rec->size[i] == 0)
 460         {
 461             rec_encode_unsigned(1, (unsigned char *) *out_buf + *out_offset,
 462                                 &len);
 463             (*out_offset) += len;
 464         }
 465         else if (last_rec && rec->size[i] == last_rec->size[i] &&
 466                  !memcmp(rec->info[i], last_rec->info[i], rec->size[i]))
 467         {
 468             rec_encode_unsigned(0, (unsigned char *) *out_buf + *out_offset,
 469                                 &len);
 470             (*out_offset) += len;
 471         }
 472         else
 473         {
 474             rec_encode_unsigned(rec->size[i]+1,
 475                                 (unsigned char *) *out_buf + *out_offset,
 476                                 &len);
 477             (*out_offset) += len;
 478             memcpy(*out_buf + *out_offset, rec->info[i], rec->size[i]);
 479             (*out_offset) += rec->size[i];
 480         }
 481     }
 482 }
 483
 484 static ZEBRA_RES rec_write_multiple(Records p, int saveCount)
 485 {
 486     int i;
 487     short ref_count = 0;
 488     char compression_method;
 489     Record last_rec = 0;
 490     int out_size = 1000;
 491     int out_offset = 0;
 492     char *out_buf = (char *) xmalloc(out_size);
 493     zint *sysnos = (zint *) xmalloc(sizeof(*sysnos) * (p->cache_cur + 1));
 494     zint *sysnop = sysnos;
 495     ZEBRA_RES ret = ZEBRA_OK;
 496
 497     for (i = 0; i<p->cache_cur - saveCount; i++)
 498     {
 499         struct record_cache_entry *e = p->record_cache + i;
 500         switch (e->flag)
 501         {
 502         case recordFlagNew:
 503             rec_cache_flush_block1(p, e->rec, last_rec, &out_buf,
 504                                     &out_size, &out_offset);
 505             *sysnop++ = rec_sysno_to_int(e->rec->sysno);
 506             ref_count++;
 507             e->flag = recordFlagNop;
 508             last_rec = e->rec;
 509             break;
 510         case recordFlagWrite:
 511             if (rec_release_blocks(p, rec_sysno_to_int(e->rec->sysno))
 512                 != ZEBRA_OK)
 513                 ret = ZEBRA_FAIL;
 514
 515             rec_cache_flush_block1(p, e->rec, last_rec, &out_buf,
 516                                     &out_size, &out_offset);
 517             *sysnop++ = rec_sysno_to_int(e->rec->sysno);
 518             ref_count++;
 519             e->flag = recordFlagNop;
 520             last_rec = e->rec;
 521             break;
 522         case recordFlagDelete:
 523             if (rec_delete_single(p, e->rec) != ZEBRA_OK)
 524                 ret = ZEBRA_FAIL;
 525
 526             e->flag = recordFlagNop;
 527             break;
 528         default:
 529             break;
 530         }
 531     }
 532
 533     *sysnop = -1;
 534     if (ref_count)
 535     {
 536         unsigned int csize = 0;  /* indicate compression "not performed yet" */
 537         compression_method = p->compression_method;
 538         switch (compression_method)
 539         {
 540         case REC_COMPRESS_BZIP2:
 541 #if HAVE_BZLIB_H
 542             csize = out_offset + (out_offset >> 6) + 620;
 543             rec_tmp_expand(p, csize);
 544 #ifdef BZ_CONFIG_ERROR
 545             i = BZ2_bzBuffToBuffCompress
 546 #else
 547             i = bzBuffToBuffCompress
 548 #endif
 549                                     (p->tmp_buf+sizeof(zint)+sizeof(short)+
 550                                       sizeof(char),
 551                                       &csize, out_buf, out_offset, 1, 0, 30);
 552             if (i != BZ_OK)
 553             {
 554                 yaz_log(YLOG_WARN, "bzBuffToBuffCompress error code=%d", i);
 555                 csize = 0;
 556             }
 557             yaz_log(YLOG_LOG, "compress %4d %5d %5d", ref_count, out_offset,
 558                   csize);
 559 #endif
 560             break;
 561         case REC_COMPRESS_NONE:
 562             break;
 563         }
 564         if (!csize)
 565         {
 566             /* either no compression or compression not supported ... */
 567             csize = out_offset;
 568             rec_tmp_expand(p, csize);
 569             memcpy(p->tmp_buf + sizeof(zint) + sizeof(short) + sizeof(char),
 570                     out_buf, out_offset);
 571             csize = out_offset;
 572             compression_method = REC_COMPRESS_NONE;
 573         }
 574         memcpy(p->tmp_buf + sizeof(zint), &ref_count, sizeof(ref_count));
 575         memcpy(p->tmp_buf + sizeof(zint)+sizeof(short),
 576                 &compression_method, sizeof(compression_method));
 577
 578         /* -------- compression */
 579         if (rec_write_tmp_buf(p, csize + sizeof(short) + sizeof(char), sysnos)
 580             != ZEBRA_OK)
 581             ret = ZEBRA_FAIL;
 582     }
 583     xfree(out_buf);
 584     xfree(sysnos);
 585     return ret;
 586 }
 587
 588 static ZEBRA_RES rec_cache_flush(Records p, int saveCount)
 589 {
 590     int i, j;
 591     ZEBRA_RES ret;
 592
 593     if (saveCount >= p->cache_cur)
 594         saveCount = 0;
 595
 596     ret = rec_write_multiple(p, saveCount);
 597
 598     for (i = 0; i<p->cache_cur - saveCount; i++)
 599     {
 600         struct record_cache_entry *e = p->record_cache + i;
 601         rec_free(&e->rec);
 602     }
 603     /* i still being used ... */
 604     for (j = 0; j<saveCount; j++, i++)
 605         memcpy(p->record_cache+j, p->record_cache+i,
 606                 sizeof(*p->record_cache));
 607     p->cache_cur = saveCount;
 608     return ret;
 609 }
 610
 611 static Record *rec_cache_lookup(Records p, zint sysno,
 612                                 enum recordCacheFlag flag)
 613 {
 614     int i;
 615     for (i = 0; i<p->cache_cur; i++)
 616     {
 617         struct record_cache_entry *e = p->record_cache + i;
 618         if (e->rec->sysno == sysno)
 619         {
 620             if (flag != recordFlagNop && e->flag == recordFlagNop)
 621                 e->flag = flag;
 622             return &e->rec;
 623         }
 624     }
 625     return NULL;
 626 }
 627
 628 static ZEBRA_RES rec_cache_insert(Records p, Record rec, enum recordCacheFlag flag)
 629 {
 630     struct record_cache_entry *e;
 631     ZEBRA_RES ret = ZEBRA_OK;
 632
 633     if (p->cache_cur == p->cache_max)
 634         ret = rec_cache_flush(p, 1);
 635     else if (p->cache_cur > 0)
 636     {
 637         int i, j;
 638         int used = 0;
 639         for (i = 0; i<p->cache_cur; i++)
 640         {
 641             Record r = (p->record_cache + i)->rec;
 642             for (j = 0; j<REC_NO_INFO; j++)
 643                 used += r->size[j];
 644         }
 645         if (used > p->compression_chunk_size)
 646             ret = rec_cache_flush(p, 1);
 647     }
 648     assert(p->cache_cur < p->cache_max);
 649
 650     e = p->record_cache + (p->cache_cur)++;
 651     e->flag = flag;
 652     e->rec = rec_cp(rec);
 653     return ret;
 654 }
 655
 656 ZEBRA_RES rec_close(Records *pp)
 657 {
 658     Records p = *pp;
 659     int i;
 660     ZEBRA_RES ret = ZEBRA_OK;
 661
 662     if (!p)
 663         return ret;
 664
 665     zebra_mutex_destroy(&p->mutex);
 666     if (rec_cache_flush(p, 0) != ZEBRA_OK)
 667         ret = ZEBRA_FAIL;
 668
 669     xfree(p->record_cache);
 670
 671     if (p->rw)
 672     {
 673         if (recindex_write_head(p->recindex, &p->head, sizeof(p->head)) != ZEBRA_OK)
 674             ret = ZEBRA_FAIL;
 675     }
 676
 677     recindex_close(p->recindex);
 678
 679     for (i = 0; i<REC_BLOCK_TYPES; i++)
 680     {
 681         if (p->data_BFile[i])
 682             bf_close(p->data_BFile[i]);
 683         xfree(p->data_fname[i]);
 684     }
 685     xfree(p->tmp_buf);
 686     xfree(p);
 687     *pp = NULL;
 688     return ret;
 689 }
 690
 691 static Record rec_get_int(Records p, zint sysno)
 692 {
 693     int i, in_size, r;
 694     Record rec, *recp;
 695     struct record_index_entry entry;
 696     zint freeblock;
 697     int dst_type;
 698     char *nptr, *cptr;
 699     char *in_buf = 0;
 700     char *bz_buf = 0;
 701 #if HAVE_BZLIB_H
 702     unsigned int bz_size;
 703 #endif
 704     char compression_method;
 705
 706     assert(sysno > 0);
 707     assert(p);
 708
 709     if ((recp = rec_cache_lookup(p, sysno, recordFlagNop)))
 710         return rec_cp(*recp);
 711
 712     if (recindex_read_indx(p->recindex, rec_sysno_to_int(sysno), &entry, sizeof(entry), 1) < 1)
 713         return NULL;       /* record is not there! */
 714
 715     if (!entry.size)
 716         return NULL;       /* record is deleted */
 717
 718     dst_type = (int) (entry.next & 7);
 719     assert(dst_type < REC_BLOCK_TYPES);
 720     freeblock = entry.next / 8;
 721
 722     assert(freeblock > 0);
 723
 724     rec_tmp_expand(p, entry.size);
 725
 726     cptr = p->tmp_buf;
 727     r = bf_read(p->data_BFile[dst_type], freeblock, 0, 0, cptr);
 728     if (r < 0)
 729         return 0;
 730     memcpy(&freeblock, cptr, sizeof(freeblock));
 731
 732     while (freeblock)
 733     {
 734         zint tmp;
 735
 736         cptr += p->head.block_size[dst_type] - sizeof(freeblock);
 737
 738         memcpy(&tmp, cptr, sizeof(tmp));
 739         r = bf_read(p->data_BFile[dst_type], freeblock, 0, 0, cptr);
 740         if (r < 0)
 741             return 0;
 742         memcpy(&freeblock, cptr, sizeof(freeblock));
 743         memcpy(cptr, &tmp, sizeof(tmp));
 744     }
 745
 746     rec = (Record) xmalloc(sizeof(*rec));
 747     rec->sysno = sysno;
 748     memcpy(&compression_method, p->tmp_buf + sizeof(zint) + sizeof(short),
 749             sizeof(compression_method));
 750     in_buf = p->tmp_buf + sizeof(zint) + sizeof(short) + sizeof(char);
 751     in_size = entry.size - sizeof(short) - sizeof(char);
 752     switch (compression_method)
 753     {
 754     case REC_COMPRESS_BZIP2:
 755 #if HAVE_BZLIB_H
 756         bz_size = entry.size * 20 + 100;
 757         while (1)
 758         {
 759             bz_buf = (char *) xmalloc(bz_size);
 760 #ifdef BZ_CONFIG_ERROR
 761             i = BZ2_bzBuffToBuffDecompress
 762 #else
 763             i = bzBuffToBuffDecompress
 764 #endif
 765                 (bz_buf, &bz_size, in_buf, in_size, 0, 0);
 766             yaz_log(YLOG_LOG, "decompress %5d %5d", in_size, bz_size);
 767             if (i == BZ_OK)
 768                 break;
 769             yaz_log(YLOG_LOG, "failed");
 770             xfree(bz_buf);
 771             bz_size *= 2;
 772         }
 773         in_buf = bz_buf;
 774         in_size = bz_size;
 775 #else
 776         yaz_log(YLOG_FATAL, "cannot decompress record(s) in BZIP2 format");
 777         return 0;
 778 #endif
 779         break;
 780     case REC_COMPRESS_NONE:
 781         break;
 782     }
 783     for (i = 0; i<REC_NO_INFO; i++)
 784         rec->info[i] = 0;
 785
 786     nptr = in_buf;                /* skip ref count */
 787     while (nptr < in_buf + in_size)
 788     {
 789         zint this_sysno;
 790         int len;
 791         rec_decode_zint(&this_sysno, (unsigned char *) nptr, &len);
 792         nptr += len;
 793
 794         for (i = 0; i < REC_NO_INFO; i++)
 795         {
 796             unsigned int this_size;
 797             rec_decode_unsigned(&this_size, (unsigned char *) nptr, &len);
 798             nptr += len;
 799
 800             if (this_size == 0)
 801                 continue;
 802             rec->size[i] = this_size-1;
 803
 804             if (rec->size[i])
 805             {
 806                 rec->info[i] = nptr;
 807                 nptr += rec->size[i];
 808             }
 809             else
 810                 rec->info[i] = NULL;
 811         }
 812         if (this_sysno == rec_sysno_to_int(sysno))
 813             break;
 814     }
 815     for (i = 0; i<REC_NO_INFO; i++)
 816     {
 817         if (rec->info[i] && rec->size[i])
 818         {
 819             char *np = xmalloc(rec->size[i]+1);
 820             memcpy(np, rec->info[i], rec->size[i]);
 821             np[rec->size[i]] = '\0';
 822             rec->info[i] = np;
 823         }
 824         else
 825         {
 826             assert(rec->info[i] == 0);
 827             assert(rec->size[i] == 0);
 828         }
 829     }
 830     xfree(bz_buf);
 831     if (rec_cache_insert(p, rec, recordFlagNop) != ZEBRA_OK)
 832         return 0;
 833     return rec;
 834 }
 835
 836 Record rec_get(Records p, zint sysno)
 837 {
 838     Record rec;
 839     zebra_mutex_lock(&p->mutex);
 840
 841     rec = rec_get_int(p, sysno);
 842     zebra_mutex_unlock(&p->mutex);
 843     return rec;
 844 }
 845
 846 Record rec_get_root(Records p)
 847 {
 848     return rec_get(p, rec_sysno_to_ext(1));
 849 }
 850
 851 static Record rec_new_int(Records p)
 852 {
 853     int i;
 854     zint sysno;
 855     Record rec;
 856
 857     assert(p);
 858     rec = (Record) xmalloc(sizeof(*rec));
 859     if (1 || p->head.index_free == 0)
 860         sysno = (p->head.index_last)++;
 861     else
 862     {
 863         struct record_index_entry entry;
 864
 865         if (recindex_read_indx(p->recindex, p->head.index_free, &entry, sizeof(entry), 0) < 1)
 866         {
 867             xfree(rec);
 868             return 0;
 869         }
 870         sysno = p->head.index_free;
 871         p->head.index_free = entry.next;
 872     }
 873     (p->head.no_records)++;
 874     rec->sysno = rec_sysno_to_ext(sysno);
 875     for (i = 0; i < REC_NO_INFO; i++)
 876     {
 877         rec->info[i] = NULL;
 878         rec->size[i] = 0;
 879     }
 880     rec_cache_insert(p, rec, recordFlagNew);
 881     return rec;
 882 }
 883
 884 Record rec_new(Records p)
 885 {
 886     Record rec;
 887     zebra_mutex_lock(&p->mutex);
 888
 889     rec = rec_new_int(p);
 890     zebra_mutex_unlock(&p->mutex);
 891     return rec;
 892 }
 893
 894 ZEBRA_RES rec_del(Records p, Record *recpp)
 895 {
 896     Record *recp;
 897     ZEBRA_RES ret = ZEBRA_OK;
 898
 899     zebra_mutex_lock(&p->mutex);
 900     (p->head.no_records)--;
 901     if ((recp = rec_cache_lookup(p, (*recpp)->sysno, recordFlagDelete)))
 902     {
 903         rec_free(recp);
 904         *recp = *recpp;
 905     }
 906     else
 907     {
 908         ret = rec_cache_insert(p, *recpp, recordFlagDelete);
 909         rec_free(recpp);
 910     }
 911     zebra_mutex_unlock(&p->mutex);
 912     *recpp = NULL;
 913     return ret;
 914 }
 915
 916 ZEBRA_RES rec_put(Records p, Record *recpp)
 917 {
 918     Record *recp;
 919     ZEBRA_RES ret = ZEBRA_OK;
 920
 921     zebra_mutex_lock(&p->mutex);
 922     if ((recp = rec_cache_lookup(p, (*recpp)->sysno, recordFlagWrite)))
 923     {
 924         rec_free(recp);
 925         *recp = *recpp;
 926     }
 927     else
 928     {
 929         ret = rec_cache_insert(p, *recpp, recordFlagWrite);
 930         rec_free(recpp);
 931     }
 932     zebra_mutex_unlock(&p->mutex);
 933     *recpp = NULL;
 934     return ret;
 935 }
 936
 937 void rec_free(Record *recpp)
 938 {
 939     int i;
 940
 941     if (!*recpp)
 942         return ;
 943     for (i = 0; i < REC_NO_INFO; i++)
 944         xfree((*recpp)->info[i]);
 945     xfree(*recpp);
 946     *recpp = NULL;
 947 }
 948
 949 Record rec_cp(Record rec)
 950 {
 951     Record n;
 952     int i;
 953
 954     n = (Record) xmalloc(sizeof(*n));
 955     n->sysno = rec->sysno;
 956     for (i = 0; i < REC_NO_INFO; i++)
 957         if (!rec->info[i])
 958         {
 959             n->info[i] = NULL;
 960             n->size[i] = 0;
 961         }
 962         else
 963         {
 964             n->size[i] = rec->size[i];
 965             n->info[i] = (char *) xmalloc(rec->size[i]+1);
 966             memcpy(n->info[i], rec->info[i], rec->size[i]);
 967             n->info[i][rec->size[i]] = '\0';
 968         }
 969     return n;
 970 }
 971
 972
 973 char *rec_strdup(const char *s, size_t *len)
 974 {
 975     char *p;
 976
 977     if (!s)
 978     {
 979         *len = 0;
 980         return NULL;
 981     }
 982     *len = strlen(s)+1;
 983     p = (char *) xmalloc(*len);
 984     strcpy(p, s);
 985     return p;
 986 }
 987
 988 void rec_prstat(Records records)
 989 {
 990     int i;
 991     zint total_bytes = 0;
 992
 993     yaz_log (YLOG_LOG,
 994           "Total records                        %8" ZINT_FORMAT0,
 995           records->head.no_records);
 996
 997     for (i = 0; i< REC_BLOCK_TYPES; i++)
 998     {
 999         yaz_log (YLOG_LOG, "Record blocks of size "ZINT_FORMAT,
1000               records->head.block_size[i]);
1001         yaz_log (YLOG_LOG,
1002           " Used/Total/Bytes used            "
1003               ZINT_FORMAT "/" ZINT_FORMAT "/" ZINT_FORMAT,
1004               records->head.block_used[i], records->head.block_last[i]-1,
1005               records->head.block_used[i] * records->head.block_size[i]);
1006         total_bytes +=
1007             records->head.block_used[i] * records->head.block_size[i];
1008     }
1009     yaz_log (YLOG_LOG,
1010           "Total size of record index in bytes  %8" ZINT_FORMAT0,
1011           records->head.total_bytes);
1012     yaz_log (YLOG_LOG,
1013           "Total size with overhead             %8" ZINT_FORMAT0,
1014           total_bytes);
1015 }
1016
1017 /*
1018  * Local variables:
1019  * c-basic-offset: 4
1020  * indent-tabs-mode: nil
1021  * End:
1022  * vim: shiftwidth=4 tabstop=8 expandtab
1023  */
1024