isamb/isamb.c

   1 /* $Id: isamb.c,v 1.31 2004-06-01 13:46:41 adam Exp $
   2    Copyright (C) 1995,1996,1997,1998,1999,2000,2001,2002,2003,2004
   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 #include <string.h>
  24 #include <yaz/xmalloc.h>
  25 #include <yaz/log.h>
  26 #include <isamb.h>
  27 #include <assert.h>
  28
  29 #ifndef ISAMB_DEBUG
  30 #define ISAMB_DEBUG 0
  31 #endif
  32
  33 struct ISAMB_head {
  34     int first_block;
  35     int last_block;
  36     int block_size;
  37     int block_max;
  38     int free_list;
  39 };
  40
  41 #define ISAMB_DATA_OFFSET 3
  42
  43 /* maximum size of encoded buffer */
  44 #define DST_ITEM_MAX 256
  45
  46 #define ISAMB_MAX_LEVEL 10
  47 /* approx 2*max page + max size of item */
  48 #define DST_BUF_SIZE 16840
  49
  50 #define ISAMB_CACHE_ENTRY_SIZE 4096
  51
  52 /* CAT_MAX: _must_ be power of 2 */
  53 #define CAT_MAX 4
  54 #define CAT_MASK (CAT_MAX-1)
  55 /* CAT_NO: <= CAT_MAX */
  56 #define CAT_NO 4
  57
  58 /* ISAMB_PTR_CODEC=1 var, =0 fixed */
  59 #define ISAMB_PTR_CODEC  1
  60
  61 struct ISAMB_cache_entry {
  62     ISAMB_P pos;
  63     unsigned char *buf;
  64     int dirty;
  65     int hits;
  66     struct ISAMB_cache_entry *next;
  67 };
  68
  69 struct ISAMB_file {
  70     BFile bf;
  71     int head_dirty;
  72     struct ISAMB_head head;
  73     struct ISAMB_cache_entry *cache_entries;
  74 };
  75
  76 struct ISAMB_s {
  77     BFiles bfs;
  78     ISAMC_M *method;
  79
  80     struct ISAMB_file *file;
  81     int no_cat;
  82     int cache; /* 0=no cache, 1=use cache, -1=dummy isam (for testing only) */
  83     int log_io;        /* log level for bf_read/bf_write calls */
  84     int log_freelist;  /* log level for freelist handling */
  85     int skipped_numbers; /* on a leaf node */
  86     int returned_numbers;
  87     int skipped_nodes[ISAMB_MAX_LEVEL]; /* [0]=skipped leaves, 1=higher etc */
  88     int accessed_nodes[ISAMB_MAX_LEVEL]; /* nodes we did not skip */
  89 };
  90
  91 struct ISAMB_block {
  92     ISAMB_P pos;
  93     int cat;
  94     int size;
  95     int leaf;
  96     int dirty;
  97     int deleted;
  98     int offset;
  99     char *bytes;
 100     unsigned char *buf;
 101     void *decodeClientData;
 102     int log_rw;
 103 };
 104
 105 struct ISAMB_PP_s {
 106     ISAMB isamb;
 107     ISAMB_P pos;
 108     int level;
 109     int maxlevel; /* total depth */
 110     int total_size;
 111     int no_blocks;
 112     int skipped_numbers; /* on a leaf node */
 113     int returned_numbers;
 114     int skipped_nodes[ISAMB_MAX_LEVEL]; /* [0]=skipped leaves, 1=higher etc */
 115     int accessed_nodes[ISAMB_MAX_LEVEL]; /* nodes we did not skip */
 116     struct ISAMB_block **block;
 117 };
 118
 119 #if ISAMB_PTR_CODEC
 120 static void encode_ptr (char **dst, unsigned pos)
 121 {
 122     unsigned char *bp = (unsigned char*) *dst;
 123
 124     while (pos > 127)
 125     {
 126          *bp++ = 128 | (pos & 127);
 127          pos = pos >> 7;
 128     }
 129     *bp++ = pos;
 130     *dst = (char *) bp;
 131 }
 132 #else
 133 static void encode_ptr (char **dst, unsigned pos)
 134 {
 135     memcpy(*dst, &pos, sizeof(pos));
 136     (*dst) += sizeof(pos);
 137 }
 138 #endif
 139
 140 #if ISAMB_PTR_CODEC
 141 static void decode_ptr (char **src1, int *pos)
 142 {
 143     unsigned char **src = (unsigned char **) src1;
 144     unsigned d = 0;
 145     unsigned char c;
 146     unsigned r = 0;
 147
 148     while (((c = *(*src)++) & 128))
 149     {
 150         d += ((c & 127) << r);
 151         r += 7;
 152     }
 153     d += (c << r);
 154     *pos = d;
 155 }
 156 #else
 157 static void decode_ptr (char **src, int *pos)
 158 {
 159      memcpy (pos, *src, sizeof(*pos));
 160      (*src) += sizeof(*pos);
 161 }
 162 #endif
 163
 164 ISAMB isamb_open (BFiles bfs, const char *name, int writeflag, ISAMC_M *method,
 165                   int cache)
 166 {
 167     ISAMB isamb = xmalloc (sizeof(*isamb));
 168     int i, b_size = 32;
 169
 170     isamb->bfs = bfs;
 171     isamb->method = (ISAMC_M *) xmalloc (sizeof(*method));
 172     memcpy (isamb->method, method, sizeof(*method));
 173     isamb->no_cat = CAT_NO;
 174     isamb->log_io = 0;
 175     isamb->log_freelist = 0;
 176     isamb->cache = cache;
 177     isamb->skipped_numbers=0;
 178     isamb->returned_numbers=0;
 179     for (i=0;i<ISAMB_MAX_LEVEL;i++)
 180       isamb->skipped_nodes[i]= isamb->accessed_nodes[i]=0;
 181
 182     assert (cache == 0);
 183     isamb->file = xmalloc (sizeof(*isamb->file) * isamb->no_cat);
 184     for (i = 0; i<isamb->no_cat; i++)
 185     {
 186         char fname[DST_BUF_SIZE];
 187         isamb->file[i].cache_entries = 0;
 188         isamb->file[i].head_dirty = 0;
 189         sprintf (fname, "%s%c", name, i+'A');
 190         if (cache)
 191             isamb->file[i].bf = bf_open (bfs, fname, ISAMB_CACHE_ENTRY_SIZE,
 192                                          writeflag);
 193         else
 194             isamb->file[i].bf = bf_open (bfs, fname, b_size, writeflag);
 195
 196
 197         if (!bf_read (isamb->file[i].bf, 0, 0, sizeof(struct ISAMB_head),
 198                       &isamb->file[i].head))
 199         {
 200             isamb->file[i].head.first_block = ISAMB_CACHE_ENTRY_SIZE/b_size+1;
 201             isamb->file[i].head.last_block = isamb->file[i].head.first_block;
 202             isamb->file[i].head.block_size = b_size;
 203             isamb->file[i].head.block_max = b_size - ISAMB_DATA_OFFSET;
 204             isamb->file[i].head.free_list = 0;
 205         }
 206         assert (isamb->file[i].head.block_size >= ISAMB_DATA_OFFSET);
 207         isamb->file[i].head_dirty = 0;
 208         assert(isamb->file[i].head.block_size == b_size);
 209         b_size = b_size * 4;
 210     }
 211     return isamb;
 212 }
 213
 214 static void flush_blocks (ISAMB b, int cat)
 215 {
 216     while (b->file[cat].cache_entries)
 217     {
 218         struct ISAMB_cache_entry *ce_this = b->file[cat].cache_entries;
 219         b->file[cat].cache_entries = ce_this->next;
 220
 221         if (ce_this->dirty)
 222         {
 223             yaz_log (b->log_io, "bf_write: flush_blocks");
 224             bf_write (b->file[cat].bf, ce_this->pos, 0, 0, ce_this->buf);
 225         }
 226         xfree (ce_this->buf);
 227         xfree (ce_this);
 228     }
 229 }
 230
 231 static int get_block (ISAMB b, ISAMC_P pos, char *userbuf, int wr)
 232 {
 233     int cat = pos&CAT_MASK;
 234     int off = ((pos/CAT_MAX) &
 235                (ISAMB_CACHE_ENTRY_SIZE / b->file[cat].head.block_size - 1))
 236         * b->file[cat].head.block_size;
 237     int norm = pos / (CAT_MASK*ISAMB_CACHE_ENTRY_SIZE / b->file[cat].head.block_size);
 238     int no = 0;
 239     struct ISAMB_cache_entry **ce, *ce_this = 0, **ce_last = 0;
 240
 241     if (!b->cache)
 242         return 0;
 243
 244     assert (ISAMB_CACHE_ENTRY_SIZE >= b->file[cat].head.block_size);
 245     for (ce = &b->file[cat].cache_entries; *ce; ce = &(*ce)->next, no++)
 246     {
 247         ce_last = ce;
 248         if ((*ce)->pos == norm)
 249         {
 250             ce_this = *ce;
 251             *ce = (*ce)->next;   /* remove from list */
 252
 253             ce_this->next = b->file[cat].cache_entries;  /* move to front */
 254             b->file[cat].cache_entries = ce_this;
 255
 256             if (wr)
 257             {
 258                 memcpy (ce_this->buf + off, userbuf,
 259                         b->file[cat].head.block_size);
 260                 ce_this->dirty = 1;
 261             }
 262             else
 263                 memcpy (userbuf, ce_this->buf + off,
 264                         b->file[cat].head.block_size);
 265             return 1;
 266         }
 267     }
 268     if (no >= 40)
 269     {
 270         assert (no == 40);
 271         assert (ce_last && *ce_last);
 272         ce_this = *ce_last;
 273         *ce_last = 0;  /* remove the last entry from list */
 274         if (ce_this->dirty)
 275         {
 276             yaz_log (b->log_io, "bf_write: get_block");
 277             bf_write (b->file[cat].bf, ce_this->pos, 0, 0, ce_this->buf);
 278         }
 279         xfree (ce_this->buf);
 280         xfree (ce_this);
 281     }
 282     ce_this = xmalloc (sizeof(*ce_this));
 283     ce_this->next = b->file[cat].cache_entries;
 284     b->file[cat].cache_entries = ce_this;
 285     ce_this->buf = xmalloc (ISAMB_CACHE_ENTRY_SIZE);
 286     ce_this->pos = norm;
 287     yaz_log (b->log_io, "bf_read: get_block");
 288     if (!bf_read (b->file[cat].bf, norm, 0, 0, ce_this->buf))
 289         memset (ce_this->buf, 0, ISAMB_CACHE_ENTRY_SIZE);
 290     if (wr)
 291     {
 292         memcpy (ce_this->buf + off, userbuf, b->file[cat].head.block_size);
 293         ce_this->dirty = 1;
 294     }
 295     else
 296     {
 297         ce_this->dirty = 0;
 298         memcpy (userbuf, ce_this->buf + off, b->file[cat].head.block_size);
 299     }
 300     return 1;
 301 }
 302
 303
 304 void isamb_close (ISAMB isamb)
 305 {
 306     int i;
 307     for (i=0;isamb->accessed_nodes[i];i++)
 308         logf(LOG_DEBUG,"isamb_close  level leaf-%d: %d read, %d skipped",
 309              i, isamb->accessed_nodes[i], isamb->skipped_nodes[i]);
 310     logf(LOG_DEBUG,"isamb_close returned %d values, skipped %d",
 311          isamb->skipped_numbers, isamb->returned_numbers);
 312     for (i = 0; i<isamb->no_cat; i++)
 313     {
 314         flush_blocks (isamb, i);
 315         if (isamb->file[i].head_dirty)
 316             bf_write (isamb->file[i].bf, 0, 0,
 317                       sizeof(struct ISAMB_head), &isamb->file[i].head);
 318
 319         bf_close (isamb->file[i].bf);
 320     }
 321     xfree (isamb->file);
 322     xfree (isamb->method);
 323     xfree (isamb);
 324 }
 325
 326
 327 struct ISAMB_block *open_block (ISAMB b, ISAMC_P pos)
 328 {
 329     int cat = pos&CAT_MASK;
 330     struct ISAMB_block *p;
 331     if (!pos)
 332         return 0;
 333     p = xmalloc (sizeof(*p));
 334     p->pos = pos;
 335     p->cat = pos & CAT_MASK;
 336     p->buf = xmalloc (b->file[cat].head.block_size);
 337
 338     if (!get_block (b, pos, p->buf, 0))
 339     {
 340         yaz_log (b->log_io, "bf_read: open_block");
 341         if (!bf_read (b->file[cat].bf, pos/CAT_MAX, 0, 0, p->buf))
 342         {
 343             yaz_log (LOG_FATAL, "isamb: read fail for pos=%ld block=%ld",
 344                      (long) pos, (long) pos/CAT_MAX);
 345             abort();
 346         }
 347     }
 348     p->bytes = p->buf + ISAMB_DATA_OFFSET;
 349     p->leaf = p->buf[0];
 350     p->size = (p->buf[1] + 256 * p->buf[2]) - ISAMB_DATA_OFFSET;
 351     if (p->size < 0)
 352     {
 353         yaz_log (LOG_FATAL, "Bad block size %d in pos=%d\n", p->size, pos);
 354     }
 355     assert (p->size >= 0);
 356     p->offset = 0;
 357     p->dirty = 0;
 358     p->deleted = 0;
 359     p->decodeClientData = (*b->method->code_start)(ISAMC_DECODE);
 360     yaz_log (LOG_DEBUG, "isamb_open_block: Opened block %d ofs=%d",pos, p->offset);
 361     return p;
 362 }
 363
 364 struct ISAMB_block *new_block (ISAMB b, int leaf, int cat)
 365 {
 366     struct ISAMB_block *p;
 367
 368     p = xmalloc (sizeof(*p));
 369     p->buf = xmalloc (b->file[cat].head.block_size);
 370
 371     if (!b->file[cat].head.free_list)
 372     {
 373         int block_no;
 374         block_no = b->file[cat].head.last_block++;
 375         p->pos = block_no * CAT_MAX + cat;
 376     }
 377     else
 378     {
 379         p->pos = b->file[cat].head.free_list;
 380         assert((p->pos & CAT_MASK) == cat);
 381         if (!get_block (b, p->pos, p->buf, 0))
 382         {
 383             yaz_log (b->log_io, "bf_read: new_block");
 384             if (!bf_read (b->file[cat].bf, p->pos/CAT_MAX, 0, 0, p->buf))
 385             {
 386                 yaz_log (LOG_FATAL, "isamb: read fail for pos=%ld block=%ld",
 387                          (long) p->pos/CAT_MAX, (long) p->pos/CAT_MAX);
 388                 abort ();
 389             }
 390         }
 391         yaz_log (b->log_freelist, "got block %d from freelist %d:%d", p->pos,
 392                  cat, p->pos/CAT_MAX);
 393         memcpy (&b->file[cat].head.free_list, p->buf, sizeof(int));
 394     }
 395     p->cat = cat;
 396     b->file[cat].head_dirty = 1;
 397     memset (p->buf, 0, b->file[cat].head.block_size);
 398     p->bytes = p->buf + ISAMB_DATA_OFFSET;
 399     p->leaf = leaf;
 400     p->size = 0;
 401     p->dirty = 1;
 402     p->deleted = 0;
 403     p->offset = 0;
 404     p->decodeClientData = (*b->method->code_start)(ISAMC_DECODE);
 405     return p;
 406 }
 407
 408 struct ISAMB_block *new_leaf (ISAMB b, int cat)
 409 {
 410     return new_block (b, 1, cat);
 411 }
 412
 413
 414 struct ISAMB_block *new_int (ISAMB b, int cat)
 415 {
 416     return new_block (b, 0, cat);
 417 }
 418
 419 static void check_block (ISAMB b, struct ISAMB_block *p)
 420 {
 421     if (p->leaf)
 422     {
 423         ;
 424     }
 425     else
 426     {
 427         /* sanity check */
 428         char *startp = p->bytes;
 429         char *src = startp;
 430         char *endp = p->bytes + p->size;
 431         int pos;
 432
 433         decode_ptr (&src, &pos);
 434         assert ((pos&CAT_MASK) == p->cat);
 435         while (src != endp)
 436         {
 437             int item_len;
 438             decode_ptr (&src, &item_len);
 439             assert (item_len > 0 && item_len < 30);
 440             src += item_len;
 441             decode_ptr (&src, &pos);
 442             assert ((pos&CAT_MASK) == p->cat);
 443         }
 444     }
 445 }
 446
 447 void close_block (ISAMB b, struct ISAMB_block *p)
 448 {
 449     if (!p)
 450         return;
 451     if (p->deleted)
 452     {
 453         yaz_log (b->log_freelist, "release block %d from freelist %d:%d",
 454                  p->pos, p->cat, p->pos/CAT_MAX);
 455         memcpy (p->buf, &b->file[p->cat].head.free_list, sizeof(int));
 456         b->file[p->cat].head.free_list = p->pos;
 457         if (!get_block (b, p->pos, p->buf, 1))
 458         {
 459             yaz_log (b->log_io, "bf_write: close_block (deleted)");
 460             bf_write (b->file[p->cat].bf, p->pos/CAT_MAX, 0, 0, p->buf);
 461         }
 462     }
 463     else if (p->dirty)
 464     {
 465         int size = p->size + ISAMB_DATA_OFFSET;
 466         assert (p->size >= 0);
 467         p->buf[0] = p->leaf;
 468         p->buf[1] = size & 255;
 469         p->buf[2] = size >> 8;
 470         check_block(b, p);
 471         if (!get_block (b, p->pos, p->buf, 1))
 472         {
 473             yaz_log (b->log_io, "bf_write: close_block");
 474             bf_write (b->file[p->cat].bf, p->pos/CAT_MAX, 0, 0, p->buf);
 475         }
 476     }
 477     (*b->method->code_stop)(ISAMC_DECODE, p->decodeClientData);
 478     xfree (p->buf);
 479     xfree (p);
 480 }
 481
 482 int insert_sub (ISAMB b, struct ISAMB_block **p,
 483                 void *new_item, int *mode,
 484                 ISAMC_I *stream,
 485                 struct ISAMB_block **sp,
 486                 void *sub_item, int *sub_size,
 487                 void *max_item);
 488
 489 int insert_int (ISAMB b, struct ISAMB_block *p, void *lookahead_item,
 490                 int *mode,
 491                 ISAMC_I *stream, struct ISAMB_block **sp,
 492                 void *split_item, int *split_size, void *last_max_item)
 493 {
 494     char *startp = p->bytes;
 495     char *src = startp;
 496     char *endp = p->bytes + p->size;
 497     int pos;
 498     struct ISAMB_block *sub_p1 = 0, *sub_p2 = 0;
 499     char sub_item[DST_ITEM_MAX];
 500     int sub_size;
 501     int more;
 502
 503     *sp = 0;
 504
 505     assert(p->size >= 0);
 506     decode_ptr (&src, &pos);
 507     while (src != endp)
 508     {
 509         int item_len;
 510         int d;
 511         char *src0 = src;
 512         decode_ptr (&src, &item_len);
 513         d = (*b->method->compare_item)(src, lookahead_item);
 514         if (d > 0)
 515         {
 516             sub_p1 = open_block (b, pos);
 517             assert (sub_p1);
 518             more = insert_sub (b, &sub_p1, lookahead_item, mode,
 519                                stream, &sub_p2,
 520                                sub_item, &sub_size, src);
 521             src = src0;
 522             break;
 523         }
 524         src += item_len;
 525         decode_ptr (&src, &pos);
 526     }
 527     if (!sub_p1)
 528     {
 529         sub_p1 = open_block (b, pos);
 530         assert (sub_p1);
 531         more = insert_sub (b, &sub_p1, lookahead_item, mode, stream, &sub_p2,
 532                            sub_item, &sub_size, last_max_item);
 533     }
 534     if (sub_p2)
 535     {
 536         /* there was a split - must insert pointer in this one */
 537         char dst_buf[DST_BUF_SIZE];
 538         char *dst = dst_buf;
 539
 540         assert (sub_size < 30 && sub_size > 1);
 541
 542         memcpy (dst, startp, src - startp);
 543
 544         dst += src - startp;
 545
 546         encode_ptr (&dst, sub_size);      /* sub length and item */
 547         memcpy (dst, sub_item, sub_size);
 548         dst += sub_size;
 549
 550         encode_ptr (&dst, sub_p2->pos);   /* pos */
 551
 552         if (endp - src)                   /* remaining data */
 553         {
 554             memcpy (dst, src, endp - src);
 555             dst += endp - src;
 556         }
 557         p->size = dst - dst_buf;
 558         assert (p->size >= 0);
 559         if (p->size <= b->file[p->cat].head.block_max)
 560         {
 561             memcpy (startp, dst_buf, dst - dst_buf);
 562         }
 563         else
 564         {
 565             int p_new_size;
 566             char *half;
 567             src = dst_buf;
 568             endp = dst;
 569
 570             half = src + b->file[p->cat].head.block_size/2;
 571             decode_ptr (&src, &pos);
 572             while (src <= half)
 573             {
 574                 decode_ptr (&src, split_size);
 575                 src += *split_size;
 576                 decode_ptr (&src, &pos);
 577             }
 578             p_new_size = src - dst_buf;
 579             memcpy (p->bytes, dst_buf, p_new_size);
 580
 581             decode_ptr (&src, split_size);
 582             memcpy (split_item, src, *split_size);
 583             src += *split_size;
 584
 585             *sp = new_int (b, p->cat);
 586             (*sp)->size = endp - src;
 587             memcpy ((*sp)->bytes, src, (*sp)->size);
 588
 589             p->size = p_new_size;
 590         }
 591         p->dirty = 1;
 592         close_block (b, sub_p2);
 593     }
 594     close_block (b, sub_p1);
 595     return more;
 596 }
 597
 598
 599 int insert_leaf (ISAMB b, struct ISAMB_block **sp1, void *lookahead_item,
 600                  int *lookahead_mode, ISAMC_I *stream,
 601                  struct ISAMB_block **sp2,
 602                  void *sub_item, int *sub_size,
 603                  void *max_item)
 604 {
 605     struct ISAMB_block *p = *sp1;
 606     char *src = 0, *endp = 0;
 607     char dst_buf[DST_BUF_SIZE], *dst = dst_buf;
 608     int new_size;
 609     void *c1 = (*b->method->code_start)(ISAMC_DECODE);
 610     void *c2 = (*b->method->code_start)(ISAMC_ENCODE);
 611     int more = 1;
 612     int quater = b->file[b->no_cat-1].head.block_max / CAT_MAX;
 613     char *cut = dst_buf + quater * 2;
 614     char *maxp = dst_buf + b->file[b->no_cat-1].head.block_max;
 615     char *half1 = 0;
 616     char *half2 = 0;
 617     char cut_item_buf[DST_ITEM_MAX];
 618     int cut_item_size = 0;
 619
 620     if (p && p->size)
 621     {
 622         char file_item_buf[DST_ITEM_MAX];
 623         char *file_item = file_item_buf;
 624
 625         src = p->bytes;
 626         endp = p->bytes + p->size;
 627         (*b->method->code_item)(ISAMC_DECODE, c1, &file_item, &src);
 628         while (1)
 629         {
 630             char *dst_item = 0;
 631             char *dst_0 = dst;
 632             char *lookahead_next;
 633             int d = -1;
 634
 635             if (lookahead_item)
 636                 d = (*b->method->compare_item)(file_item_buf, lookahead_item);
 637
 638             if (d > 0)
 639             {
 640                 dst_item = lookahead_item;
 641                 if (!*lookahead_mode)
 642                 {
 643                     yaz_log (LOG_WARN, "isamb: Inconsistent register (1)");
 644                     assert (*lookahead_mode);
 645                 }
 646             }
 647             else
 648                 dst_item = file_item_buf;
 649             if (!*lookahead_mode && d == 0)
 650             {
 651                 p->dirty = 1;
 652             }
 653             else if (!half1 && dst > cut)
 654             {
 655                 char *dst_item_0 = dst_item;
 656                 half1 = dst; /* candidate for splitting */
 657
 658                 (*b->method->code_item)(ISAMC_ENCODE, c2, &dst, &dst_item);
 659
 660                 cut_item_size = dst_item - dst_item_0;
 661                 memcpy (cut_item_buf, dst_item_0, cut_item_size);
 662
 663                 half2 = dst;
 664             }
 665             else
 666                 (*b->method->code_item)(ISAMC_ENCODE, c2, &dst, &dst_item);
 667             if (d > 0)
 668             {
 669                 if (dst > maxp)
 670                 {
 671                     dst = dst_0;
 672                     lookahead_item = 0;
 673                 }
 674                 else
 675                 {
 676                     lookahead_next = lookahead_item;
 677                     if (!(*stream->read_item)(stream->clientData,
 678                                               &lookahead_next,
 679                                               lookahead_mode))
 680                     {
 681                         lookahead_item = 0;
 682                         more = 0;
 683                     }
 684                     if (lookahead_item && max_item &&
 685                         (*b->method->compare_item)(max_item, lookahead_item) <= 0)
 686                     {
 687                         /* max_item 1 */
 688                         lookahead_item = 0;
 689                     }
 690
 691                     p->dirty = 1;
 692                 }
 693             }
 694             else if (d == 0)
 695             {
 696                 lookahead_next = lookahead_item;
 697                 if (!(*stream->read_item)(stream->clientData,
 698                                           &lookahead_next, lookahead_mode))
 699                 {
 700                     lookahead_item = 0;
 701                     more = 0;
 702                 }
 703                 if (src == endp)
 704                     break;
 705                 file_item = file_item_buf;
 706                 (*b->method->code_item)(ISAMC_DECODE, c1, &file_item, &src);
 707             }
 708             else
 709             {
 710                 if (src == endp)
 711                     break;
 712                 file_item = file_item_buf;
 713                 (*b->method->code_item)(ISAMC_DECODE, c1, &file_item, &src);
 714             }
 715         }
 716     }
 717     maxp = dst_buf + b->file[b->no_cat-1].head.block_max + quater;
 718     while (lookahead_item)
 719     {
 720         char *dst_item = lookahead_item;
 721         char *dst_0 = dst;
 722
 723         if (max_item &&
 724             (*b->method->compare_item)(max_item, lookahead_item) <= 0)
 725         {
 726             /* max_item 2 */
 727             break;
 728         }
 729         if (!*lookahead_mode)
 730         {
 731             yaz_log (LOG_WARN, "isamb: Inconsistent register (2)");
 732             abort();
 733         }
 734         else if (!half1 && dst > cut)
 735         {
 736             char *dst_item_0 = dst_item;
 737             half1 = dst; /* candidate for splitting */
 738
 739             (*b->method->code_item)(ISAMC_ENCODE, c2, &dst, &dst_item);
 740
 741             cut_item_size = dst_item - dst_item_0;
 742             memcpy (cut_item_buf, dst_item_0, cut_item_size);
 743
 744             half2 = dst;
 745         }
 746         else
 747             (*b->method->code_item)(ISAMC_ENCODE, c2, &dst, &dst_item);
 748
 749         if (dst > maxp)
 750         {
 751             dst = dst_0;
 752             break;
 753         }
 754         if (p)
 755             p->dirty = 1;
 756         dst_item = lookahead_item;
 757         if (!(*stream->read_item)(stream->clientData, &dst_item,
 758                                   lookahead_mode))
 759         {
 760             lookahead_item = 0;
 761             more = 0;
 762         }
 763     }
 764     new_size = dst - dst_buf;
 765     if (p && p->cat != b->no_cat-1 &&
 766         new_size > b->file[p->cat].head.block_max)
 767     {
 768         /* non-btree block will be removed */
 769         p->deleted = 1;
 770         close_block (b, p);
 771         /* delete it too!! */
 772         p = 0; /* make a new one anyway */
 773     }
 774     if (!p)
 775     {   /* must create a new one */
 776         int i;
 777         for (i = 0; i < b->no_cat; i++)
 778             if (new_size <= b->file[i].head.block_max)
 779                 break;
 780         if (i == b->no_cat)
 781             i = b->no_cat - 1;
 782         p = new_leaf (b, i);
 783     }
 784     if (new_size > b->file[p->cat].head.block_max)
 785     {
 786         char *first_dst;
 787         char *cut_item = cut_item_buf;
 788
 789         assert (half1);
 790         assert (half2);
 791
 792        /* first half */
 793         p->size = half1 - dst_buf;
 794         memcpy (p->bytes, dst_buf, half1 - dst_buf);
 795
 796         /* second half */
 797         *sp2 = new_leaf (b, p->cat);
 798
 799         (*b->method->code_reset)(c2);
 800
 801         first_dst = (*sp2)->bytes;
 802
 803         (*b->method->code_item)(ISAMC_ENCODE, c2, &first_dst, &cut_item);
 804
 805         memcpy (first_dst, half2, dst - half2);
 806
 807         (*sp2)->size = (first_dst - (*sp2)->bytes) + (dst - half2);
 808         (*sp2)->dirty = 1;
 809         p->dirty = 1;
 810         memcpy (sub_item, cut_item_buf, cut_item_size);
 811         *sub_size = cut_item_size;
 812     }
 813     else
 814     {
 815         memcpy (p->bytes, dst_buf, dst - dst_buf);
 816         p->size = new_size;
 817     }
 818     (*b->method->code_stop)(ISAMC_DECODE, c1);
 819     (*b->method->code_stop)(ISAMC_ENCODE, c2);
 820     *sp1 = p;
 821     return more;
 822 }
 823
 824 int insert_sub (ISAMB b, struct ISAMB_block **p, void *new_item,
 825                 int *mode,
 826                 ISAMC_I *stream,
 827                 struct ISAMB_block **sp,
 828                 void *sub_item, int *sub_size,
 829                 void *max_item)
 830 {
 831     if (!*p || (*p)->leaf)
 832         return insert_leaf (b, p, new_item, mode, stream, sp, sub_item,
 833                             sub_size, max_item);
 834     else
 835         return insert_int (b, *p, new_item, mode, stream, sp, sub_item,
 836                            sub_size, max_item);
 837 }
 838
 839 int isamb_unlink (ISAMB b, ISAMC_P pos)
 840 {
 841     struct ISAMB_block *p1;
 842
 843     if (!pos)
 844         return 0;
 845     p1 = open_block(b, pos);
 846     p1->deleted = 1;
 847     if (!p1->leaf)
 848     {
 849         int sub_p;
 850         int item_len;
 851         char *src = p1->bytes + p1->offset;
 852
 853         decode_ptr(&src, &sub_p);
 854         isamb_unlink(b, sub_p);
 855
 856         while (src != p1->bytes + p1->size)
 857         {
 858             decode_ptr(&src, &item_len);
 859             src += item_len;
 860             decode_ptr(&src, &sub_p);
 861             isamb_unlink(b, sub_p);
 862         }
 863     }
 864     close_block(b, p1);
 865     return 0;
 866 }
 867
 868 int isamb_merge (ISAMB b, ISAMC_P pos, ISAMC_I *stream)
 869 {
 870     char item_buf[DST_ITEM_MAX];
 871     char *item_ptr;
 872     int i_mode;
 873     int more;
 874
 875     if (b->cache < 0)
 876     {
 877         int more = 1;
 878         while (more)
 879         {
 880             item_ptr = item_buf;
 881             more =
 882                 (*stream->read_item)(stream->clientData, &item_ptr, &i_mode);
 883         }
 884         return 1;
 885     }
 886     item_ptr = item_buf;
 887     more = (*stream->read_item)(stream->clientData, &item_ptr, &i_mode);
 888     while (more)
 889     {
 890         struct ISAMB_block *p = 0, *sp = 0;
 891         char sub_item[DST_ITEM_MAX];
 892         int sub_size;
 893
 894         if (pos)
 895             p = open_block (b, pos);
 896         more = insert_sub (b, &p, item_buf, &i_mode, stream, &sp,
 897                             sub_item, &sub_size, 0);
 898         if (sp)
 899         {    /* increase level of tree by one */
 900             struct ISAMB_block *p2 = new_int (b, p->cat);
 901             char *dst = p2->bytes + p2->size;
 902
 903             encode_ptr (&dst, p->pos);
 904             assert (sub_size < 20);
 905             encode_ptr (&dst, sub_size);
 906             memcpy (dst, sub_item, sub_size);
 907             dst += sub_size;
 908             encode_ptr (&dst, sp->pos);
 909
 910             p2->size = dst - p2->bytes;
 911             pos = p2->pos;  /* return new super page */
 912             close_block (b, sp);
 913             close_block (b, p2);
 914         }
 915         else
 916             pos = p->pos;   /* return current one (again) */
 917         close_block (b, p);
 918     }
 919     return pos;
 920 }
 921
 922 ISAMB_PP isamb_pp_open_x (ISAMB isamb, ISAMB_P pos, int *level)
 923 {
 924     ISAMB_PP pp = xmalloc (sizeof(*pp));
 925     int i;
 926
 927     pp->isamb = isamb;
 928     pp->block = xmalloc (ISAMB_MAX_LEVEL * sizeof(*pp->block));
 929
 930     pp->pos = pos;
 931     pp->level = 0;
 932     pp->maxlevel=0;
 933     pp->total_size = 0;
 934     pp->no_blocks = 0;
 935     pp->skipped_numbers=0;
 936     pp->returned_numbers=0;
 937     for (i=0;i<ISAMB_MAX_LEVEL;i++)
 938         pp->skipped_nodes[i] = pp->accessed_nodes[i]=0;
 939     while (1)
 940     {
 941         struct ISAMB_block *p = open_block (isamb, pos);
 942         char *src = p->bytes + p->offset;
 943         pp->block[pp->level] = p;
 944
 945         pp->total_size += p->size;
 946         pp->no_blocks++;
 947         if (p->leaf)
 948             break;
 949
 950
 951         decode_ptr (&src, &pos);
 952         p->offset = src - p->bytes;
 953         pp->level++;
 954         pp->accessed_nodes[pp->level]++;
 955     }
 956     pp->block[pp->level+1] = 0;
 957     pp->maxlevel=pp->level;
 958     if (level)
 959         *level = pp->level;
 960     return pp;
 961 }
 962
 963 ISAMB_PP isamb_pp_open (ISAMB isamb, ISAMB_P pos)
 964 {
 965     return isamb_pp_open_x (isamb, pos, 0);
 966 }
 967
 968 void isamb_pp_close_x (ISAMB_PP pp, int *size, int *blocks)
 969 {
 970     int i;
 971     if (!pp)
 972         return;
 973     logf(LOG_DEBUG,"isamb_pp_close lev=%d returned %d values, skipped %d",
 974         pp->maxlevel, pp->skipped_numbers, pp->returned_numbers);
 975     for (i=pp->maxlevel;i>=0;i--)
 976         if ( pp->skipped_nodes[i] || pp->accessed_nodes[i])
 977             logf(LOG_DEBUG,"isamb_pp_close  level leaf-%d: %d read, %d skipped", i,
 978                  pp->accessed_nodes[i], pp->skipped_nodes[i]);
 979     pp->isamb->skipped_numbers += pp->skipped_numbers;
 980     pp->isamb->returned_numbers += pp->returned_numbers;
 981     for (i=pp->maxlevel;i>=0;i--)
 982     {
 983         pp->isamb->accessed_nodes[i] += pp->accessed_nodes[i];
 984         pp->isamb->skipped_nodes[i] += pp->skipped_nodes[i];
 985     }
 986     if (size)
 987         *size = pp->total_size;
 988     if (blocks)
 989         *blocks = pp->no_blocks;
 990     for (i = 0; i <= pp->level; i++)
 991         close_block (pp->isamb, pp->block[i]);
 992     xfree (pp->block);
 993     xfree (pp);
 994 }
 995
 996 int isamb_block_info (ISAMB isamb, int cat)
 997 {
 998     if (cat >= 0 && cat < isamb->no_cat)
 999         return isamb->file[cat].head.block_size;
1000     return -1;
1001 }
1002
1003 void isamb_pp_close (ISAMB_PP pp)
1004 {
1005     isamb_pp_close_x (pp, 0, 0);
1006 }
1007
1008
1009 #if 0
1010 /* Old isamb_pp_read that Adam wrote, kept as a reference in case we need to
1011    debug the more complex pp_read that also forwards. May be deleted near end
1012    of 2004, if it has not shown to be useful */
1013
1014 int isamb_pp_read (ISAMB_PP pp, void *buf)
1015 {
1016     char *dst = buf;
1017     char *src;
1018     struct ISAMB_block *p = pp->block[pp->level];
1019     if (!p)
1020         return 0;
1021
1022     while (p->offset == p->size)
1023     {
1024         int pos, item_len;
1025         while (p->offset == p->size)
1026         {
1027             if (pp->level == 0)
1028                 return 0;
1029             close_block (pp->isamb, pp->block[pp->level]);
1030             pp->block[pp->level] = 0;
1031             (pp->level)--;
1032             p = pp->block[pp->level];
1033             assert (!p->leaf);
1034         }
1035         src = p->bytes + p->offset;
1036
1037         decode_ptr (&src, &item_len);
1038         src += item_len;
1039         decode_ptr (&src, &pos);
1040
1041         p->offset = src - (char*) p->bytes;
1042
1043         ++(pp->level);
1044
1045         while (1)
1046         {
1047             pp->block[pp->level] = p = open_block (pp->isamb, pos);
1048
1049             pp->total_size += p->size;
1050             pp->no_blocks++;
1051
1052             if (p->leaf)
1053             {
1054                 break;
1055             }
1056             src = p->bytes + p->offset;
1057             decode_ptr (&src, &pos);
1058             p->offset = src - (char*) p->bytes;
1059             pp->level++;
1060         }
1061     }
1062     assert (p->offset < p->size);
1063     assert (p->leaf);
1064     src = p->bytes + p->offset;
1065     (*pp->isamb->method->code_item)(ISAMC_DECODE, p->decodeClientData,
1066                                     &dst, &src);
1067     p->offset = src - (char*) p->bytes;
1068     key_logdump_txt(LOG_DEBUG,buf, "isamb_pp_read returning 1");
1069     return 1;
1070 }
1071
1072 #else
1073 int isamb_pp_read (ISAMB_PP pp, void *buf)
1074 {
1075     return isamb_pp_forward(pp,buf,0);
1076 }
1077 #endif
1078
1079 int isamb_pp_forward (ISAMB_PP pp, void *buf, const void *untilbuf)
1080 {
1081     /* pseudocode:
1082      *   while 1
1083      *     while at end of node
1084      *       climb higher. If out, return 0
1085      *     while not on a leaf (and not at its end)
1086      *       decode next
1087      *       if cmp
1088      *         descend to node
1089      *     decode next
1090      *     if cmp
1091      *       return 1
1092      */
1093         /*
1094          * The upper nodes consist of a sequence of nodenumbers and keys
1095          * When opening a block,  the first node number is read in, and
1096          * offset points to the first key, which is the upper limit of keys
1097          * in the node just read.
1098          */
1099     char *dst = buf;
1100     char *src;
1101     struct ISAMB_block *p = pp->block[pp->level];
1102     int cmp;
1103     int item_len;
1104     int pos;
1105     int nxtpos;
1106     if (!p)
1107         return 0;
1108 #if ISAMB_DEBUG
1109     logf(LOG_DEBUG,"isamb_pp_forward starting [%p] p=%d",pp,p->pos);
1110
1111     (*pp->isamb->method->log_item)(LOG_DEBUG, untilbuf, "until");
1112     (*pp->isamb->method->log_item)(LOG_DEBUG, buf, "buf");
1113 #endif
1114
1115     while (1)
1116     {
1117         while ( p->offset == p->size)
1118         {  /* end of this block - climb higher */
1119 #if ISAMB_DEBUG
1120             logf(LOG_DEBUG,"isamb_pp_forward climbing from l=%d",
1121                             pp->level);
1122 #endif
1123             if (pp->level == 0)
1124             {
1125 #if ISAMB_DEBUG
1126                 logf(LOG_DEBUG,"isamb_pp_forward returning 0 at root");
1127 #endif
1128                 return 0; /* at end of the root, nothing left */
1129             }
1130             close_block(pp->isamb, pp->block[pp->level]);
1131             pp->block[pp->level]=0;
1132             (pp->level)--;
1133             p=pp->block[pp->level];
1134 #if ISAMB_DEBUG
1135             logf(LOG_DEBUG,"isamb_pp_forward climbed to node %d off=%d",
1136                             p->pos, p->offset);
1137 #endif
1138             assert(!p->leaf);
1139             /* skip the child we have handled */
1140             if (p->offset != p->size)
1141             {
1142                 src = p->bytes + p->offset;
1143                 decode_ptr(&src, &item_len);
1144 #if ISAMB_DEBUG
1145                 (*pp->isamb->method->log_item)(LOG_DEBUG, src,
1146                                                " isamb_pp_forward "
1147                                                "climb skipping old key");
1148 #endif
1149                 src += item_len;
1150                 decode_ptr(&src,&pos);
1151                 p->offset = src - (char*) p->bytes;
1152                 break; /* even if this puts us at the end of the block, we need to */
1153                        /* descend to the last pos. UGLY coding, clean up some */
1154                        /* day */
1155             }
1156         }
1157         if (!p->leaf)
1158         {
1159             src = p->bytes + p->offset;
1160             if (p->offset == p->size)
1161                 cmp=-2 ; /* descend to the last node, as we have no value to cmp */
1162             else
1163             {
1164                 decode_ptr(&src, &item_len);
1165 #if ISAMB_DEBUG
1166                 logf(LOG_DEBUG,"isamb_pp_forward (B) on a high node. ofs=%d sz=%d nxtpos=%d ",
1167                         p->offset,p->size,pos);
1168                 (*pp->isamb->method->log_item)(LOG_DEBUG, src, "");
1169 #endif
1170                 if (untilbuf)
1171                     cmp=(*pp->isamb->method->compare_item)(untilbuf,src);
1172                 else
1173                     cmp=-2;
1174                 src += item_len;
1175                 decode_ptr(&src,&nxtpos);
1176             }
1177             if (cmp<2)
1178             {
1179 #if ISAMB_DEBUG
1180                 logf(LOG_DEBUG,"isambb_pp_forward descending l=%d p=%d ",
1181                             pp->level, pos);
1182 #endif
1183                 ++(pp->level);
1184                 p = open_block(pp->isamb,pos);
1185                 pp->block[pp->level] = p ;
1186                 pp->total_size += p->size;
1187                 (pp->accessed_nodes[pp->maxlevel - pp->level])++;
1188                 pp->no_blocks++;
1189                 if ( !p->leaf)
1190                 { /* block starts with a pos */
1191                     src = p->bytes + p->offset;
1192                     decode_ptr(&src,&pos);
1193                     p->offset=src-(char*) p->bytes;
1194 #if ISAMB_DEBUG
1195                     logf(LOG_DEBUG,"isamb_pp_forward: block %d starts with %d",
1196                                     p->pos, pos);
1197 #endif
1198                 }
1199             } /* descend to the node */
1200             else
1201             { /* skip the node */
1202                 p->offset = src - (char*) p->bytes;
1203                 pos=nxtpos;
1204                 (pp->skipped_nodes[pp->maxlevel - pp->level -1])++;
1205 #if ISAMB_DEBUG
1206                 logf(LOG_DEBUG,
1207                     "isamb_pp_forward: skipping block on level %d, noting on %d (%d)",
1208                     pp->level, pp->maxlevel - pp->level-1 ,
1209                     pp->skipped_nodes[pp->maxlevel - pp->level-1 ]);
1210 #endif
1211                 /* 0 is always leafs, 1 is one level above leafs etc, no
1212                  * matter how high tree */
1213             }
1214         } /* not on a leaf */
1215         else
1216         { /* on a leaf */
1217             src = p->bytes + p->offset;
1218             dst=buf;
1219             (*pp->isamb->method->code_item)(ISAMC_DECODE, p->decodeClientData,
1220                                             &dst, &src);
1221             p->offset = src - (char*) p->bytes;
1222             if (untilbuf)
1223                 cmp=(*pp->isamb->method->compare_item)(untilbuf,buf);
1224             else
1225                 cmp=-2;
1226 #if ISAMB_DEBUG
1227             logf(LOG_DEBUG,"isamb_pp_forward on a leaf. cmp=%d",
1228                               cmp);
1229             (*pp->isamb->method->log_item)(LOG_DEBUG, buf, "");
1230 #endif
1231             if (cmp <2)
1232             {
1233 #if ISAMB_DEBUG
1234                 if (untilbuf)
1235                 {
1236                     (*pp->isamb->method->log_item)(LOG_DEBUG, buf,
1237                                                    "isamb_pp_forward returning 1");
1238                 }
1239                 else
1240                 {
1241                     (*pp->isamb->method->log_item)(LOG_DEBUG, buf,
1242                                                    "isamb_pp_read returning 1 (fwd)");
1243                 }
1244 #endif
1245                 pp->returned_numbers++;
1246                 return 1;
1247             }
1248             else
1249                 pp->skipped_numbers++;
1250         } /* leaf */
1251     } /* main loop */
1252 }
1253
1254
1255 int isamb_pp_num (ISAMB_PP pp)
1256 {
1257     return 1;
1258 }