index/records.c

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