isamb/isamb.c

   1 /* $Id: isamb.c,v 1.23 2003-03-02 23:12:50 adam Exp $
   2    Copyright (C) 1995,1996,1997,1998,1999,2000,2001,2002,2003
   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 struct ISAMB_head {
  30     int first_block;
  31     int last_block;
  32     int block_size;
  33     int block_max;
  34     int free_list;
  35 };
  36
  37 #define ISAMB_DATA_OFFSET 3
  38
  39 #define DST_ITEM_MAX 256
  40
  41 /* approx 2*4 K + max size of item */
  42 #define DST_BUF_SIZE 8448
  43
  44 #define ISAMB_CACHE_ENTRY_SIZE 4096
  45
  46 struct ISAMB_cache_entry {
  47     ISAMB_P pos;
  48     char *buf;
  49     int dirty;
  50     int hits;
  51     struct ISAMB_cache_entry *next;
  52 };
  53
  54
  55 struct ISAMB_file {
  56     BFile bf;
  57     int head_dirty;
  58     struct ISAMB_head head;
  59     struct ISAMB_cache_entry *cache_entries;
  60 };
  61
  62 struct ISAMB_s {
  63     BFiles bfs;
  64     ISAMC_M method;
  65
  66     struct ISAMB_file *file;
  67     int no_cat;
  68     int cache; /* 0=no cache, 1=use cache, -1=dummy isam (for testing only) */
  69     int log_io;        /* log level for bf_read/bf_write calls */
  70     int log_freelist;  /* log level for freelist handling */
  71 };
  72
  73 struct ISAMB_block {
  74     ISAMB_P pos;
  75     int cat;
  76     int size;
  77     int leaf;
  78     int dirty;
  79     int deleted;
  80     int offset;
  81     char *bytes;
  82     unsigned char *buf;
  83     void *decodeClientData;
  84     int log_rw;
  85 };
  86
  87 struct ISAMB_PP_s {
  88     ISAMB isamb;
  89     ISAMB_P pos;
  90     int level;
  91     int total_size;
  92     int no_blocks;
  93     struct ISAMB_block **block;
  94 };
  95
  96 void encode_ptr (char **dst, int pos)
  97 {
  98     memcpy (*dst, &pos, sizeof(pos));
  99     (*dst) += sizeof(pos);
 100 }
 101
 102 void decode_ptr (char **src, int *pos)
 103 {
 104     memcpy (pos, *src, sizeof(*pos));
 105     (*src) += sizeof(*pos);
 106 }
 107
 108 ISAMB isamb_open (BFiles bfs, const char *name, int writeflag, ISAMC_M method,
 109                   int cache)
 110 {
 111     ISAMB isamb = xmalloc (sizeof(*isamb));
 112     int i, b_size = 32;
 113
 114     isamb->bfs = bfs;
 115     isamb->method = (ISAMC_M) xmalloc (sizeof(*method));
 116     memcpy (isamb->method, method, sizeof(*method));
 117     isamb->no_cat = 4;
 118     isamb->log_io = 0;
 119     isamb->cache = cache;
 120
 121     isamb->file = xmalloc (sizeof(*isamb->file) * isamb->no_cat);
 122     for (i = 0; i<isamb->no_cat; i++)
 123     {
 124         char fname[DST_BUF_SIZE];
 125         isamb->file[i].cache_entries = 0;
 126         isamb->file[i].head_dirty = 0;
 127         sprintf (fname, "%s%c", name, i+'A');
 128         if (cache)
 129             isamb->file[i].bf = bf_open (bfs, fname, ISAMB_CACHE_ENTRY_SIZE,
 130                                          writeflag);
 131         else
 132             isamb->file[i].bf = bf_open (bfs, fname, b_size, writeflag);
 133
 134
 135         if (!bf_read (isamb->file[i].bf, 0, 0, sizeof(struct ISAMB_head),
 136                       &isamb->file[i].head))
 137         {
 138             isamb->file[i].head.first_block = ISAMB_CACHE_ENTRY_SIZE/b_size+1;
 139             isamb->file[i].head.last_block = isamb->file[i].head.first_block;
 140             isamb->file[i].head.block_size = b_size;
 141             isamb->file[i].head.block_max = b_size - ISAMB_DATA_OFFSET;
 142             isamb->file[i].head.free_list = 0;
 143         }
 144         assert (isamb->file[i].head.block_size >= ISAMB_DATA_OFFSET);
 145         isamb->file[i].head_dirty = 0;
 146         assert(isamb->file[i].head.block_size == b_size);
 147         b_size = b_size * 4;
 148     }
 149     return isamb;
 150 }
 151
 152 static void flush_blocks (ISAMB b, int cat)
 153 {
 154     while (b->file[cat].cache_entries)
 155     {
 156         struct ISAMB_cache_entry *ce_this = b->file[cat].cache_entries;
 157         b->file[cat].cache_entries = ce_this->next;
 158
 159         if (ce_this->dirty)
 160         {
 161             yaz_log (b->log_io, "bf_write: flush_blocks");
 162             bf_write (b->file[cat].bf, ce_this->pos, 0, 0, ce_this->buf);
 163         }
 164         xfree (ce_this->buf);
 165         xfree (ce_this);
 166     }
 167 }
 168
 169 static int get_block (ISAMB b, ISAMC_P pos, char *userbuf, int wr)
 170 {
 171     int cat = pos&3;
 172     int off = ((pos/4) &
 173                (ISAMB_CACHE_ENTRY_SIZE / b->file[cat].head.block_size - 1))
 174         * b->file[cat].head.block_size;
 175     int norm = pos / (4*ISAMB_CACHE_ENTRY_SIZE / b->file[cat].head.block_size);
 176     int no = 0;
 177     struct ISAMB_cache_entry **ce, *ce_this = 0, **ce_last = 0;
 178
 179     if (!b->cache)
 180         return 0;
 181
 182     assert (ISAMB_CACHE_ENTRY_SIZE >= b->file[cat].head.block_size);
 183     for (ce = &b->file[cat].cache_entries; *ce; ce = &(*ce)->next, no++)
 184     {
 185         ce_last = ce;
 186         if ((*ce)->pos == norm)
 187         {
 188             ce_this = *ce;
 189             *ce = (*ce)->next;   /* remove from list */
 190
 191             ce_this->next = b->file[cat].cache_entries;  /* move to front */
 192             b->file[cat].cache_entries = ce_this;
 193
 194             if (wr)
 195             {
 196                 memcpy (ce_this->buf + off, userbuf,
 197                         b->file[cat].head.block_size);
 198                 ce_this->dirty = 1;
 199             }
 200             else
 201                 memcpy (userbuf, ce_this->buf + off,
 202                         b->file[cat].head.block_size);
 203             return 1;
 204         }
 205     }
 206     if (no >= 40)
 207     {
 208         assert (no == 40);
 209         assert (ce_last && *ce_last);
 210         ce_this = *ce_last;
 211         *ce_last = 0;  /* remove the last entry from list */
 212         if (ce_this->dirty)
 213         {
 214             yaz_log (b->log_io, "bf_write: get_block");
 215             bf_write (b->file[cat].bf, ce_this->pos, 0, 0, ce_this->buf);
 216         }
 217         xfree (ce_this->buf);
 218         xfree (ce_this);
 219     }
 220     ce_this = xmalloc (sizeof(*ce_this));
 221     ce_this->next = b->file[cat].cache_entries;
 222     b->file[cat].cache_entries = ce_this;
 223     ce_this->buf = xmalloc (ISAMB_CACHE_ENTRY_SIZE);
 224     ce_this->pos = norm;
 225     yaz_log (b->log_io, "bf_read: get_block");
 226     if (!bf_read (b->file[cat].bf, norm, 0, 0, ce_this->buf))
 227         memset (ce_this->buf, 0, ISAMB_CACHE_ENTRY_SIZE);
 228     if (wr)
 229     {
 230         memcpy (ce_this->buf + off, userbuf, b->file[cat].head.block_size);
 231         ce_this->dirty = 1;
 232     }
 233     else
 234     {
 235         ce_this->dirty = 0;
 236         memcpy (userbuf, ce_this->buf + off, b->file[cat].head.block_size);
 237     }
 238     return 1;
 239 }
 240
 241
 242 void isamb_close (ISAMB isamb)
 243 {
 244     int i;
 245     for (i = 0; i<isamb->no_cat; i++)
 246     {
 247         flush_blocks (isamb, i);
 248         if (isamb->file[i].head_dirty)
 249             bf_write (isamb->file[i].bf, 0, 0,
 250                       sizeof(struct ISAMB_head), &isamb->file[i].head);
 251
 252         bf_close (isamb->file[i].bf);
 253     }
 254     xfree (isamb->file);
 255     xfree (isamb->method);
 256     xfree (isamb);
 257 }
 258
 259
 260 struct ISAMB_block *open_block (ISAMB b, ISAMC_P pos)
 261 {
 262     int cat = pos&3;
 263     struct ISAMB_block *p;
 264     if (!pos)
 265         return 0;
 266     p = xmalloc (sizeof(*p));
 267     p->pos = pos;
 268     p->cat = pos & 3;
 269     p->buf = xmalloc (b->file[cat].head.block_size);
 270
 271     if (!get_block (b, pos, p->buf, 0))
 272     {
 273         yaz_log (b->log_io, "bf_read: open_block");
 274         if (!bf_read (b->file[cat].bf, pos/4, 0, 0, p->buf))
 275         {
 276             yaz_log (LOG_FATAL, "read failure for pos=%ld block=%ld",
 277                      (long) pos, (long) pos/4);
 278             abort();
 279         }
 280     }
 281     p->bytes = p->buf + ISAMB_DATA_OFFSET;
 282     p->leaf = p->buf[0];
 283     p->size = p->buf[1] + 256 * p->buf[2] - ISAMB_DATA_OFFSET;
 284     p->offset = 0;
 285     p->dirty = 0;
 286     p->deleted = 0;
 287     p->decodeClientData = (*b->method->code_start)(ISAMC_DECODE);
 288     return p;
 289 }
 290
 291 struct ISAMB_block *new_block (ISAMB b, int leaf, int cat)
 292 {
 293     struct ISAMB_block *p;
 294
 295     p = xmalloc (sizeof(*p));
 296     p->buf = xmalloc (b->file[cat].head.block_size);
 297
 298     if (!b->file[cat].head.free_list)
 299     {
 300         int block_no;
 301         block_no = b->file[cat].head.last_block++;
 302         p->pos = block_no * 4 + cat;
 303     }
 304     else
 305     {
 306         p->pos = b->file[cat].head.free_list;
 307         if (!get_block (b, p->pos, p->buf, 0))
 308         {
 309             yaz_log (b->log_io, "bf_read: new_block");
 310             if (!bf_read (b->file[cat].bf, p->pos/4, 0, 0, p->buf))
 311             {
 312                 yaz_log (LOG_FATAL, "read failure for pos=%ld block=%ld",
 313                          (long) p->pos/4, (long) p->pos/4);
 314                 abort ();
 315             }
 316         }
 317         yaz_log (b->log_freelist, "got block %d from freelist %d:%d", p->pos,
 318                  cat, p->pos/4);
 319         memcpy (&b->file[cat].head.free_list, p->buf, sizeof(int));
 320     }
 321     p->cat = cat;
 322     b->file[cat].head_dirty = 1;
 323     memset (p->buf, 0, b->file[cat].head.block_size);
 324     p->bytes = p->buf + ISAMB_DATA_OFFSET;
 325     p->leaf = leaf;
 326     p->size = 0;
 327     p->dirty = 1;
 328     p->deleted = 0;
 329     p->offset = 0;
 330     p->decodeClientData = (*b->method->code_start)(ISAMC_DECODE);
 331     return p;
 332 }
 333
 334 struct ISAMB_block *new_leaf (ISAMB b, int cat)
 335 {
 336     return new_block (b, 1, cat);
 337 }
 338
 339
 340 struct ISAMB_block *new_int (ISAMB b, int cat)
 341 {
 342     return new_block (b, 0, cat);
 343 }
 344
 345 void close_block (ISAMB b, struct ISAMB_block *p)
 346 {
 347     if (!p)
 348         return;
 349     if (p->deleted)
 350     {
 351         yaz_log (b->log_freelist, "release block %d from freelist %d:%d",
 352                  p->pos, p->cat, p->pos/4);
 353         memcpy (p->buf, &b->file[p->cat].head.free_list, sizeof(int));
 354         b->file[p->cat].head.free_list = p->pos;
 355         if (!get_block (b, p->pos, p->buf, 1))
 356         {
 357             yaz_log (b->log_io, "bf_write: close_block");
 358             bf_write (b->file[p->cat].bf, p->pos/4, 0, 0, p->buf);
 359         }
 360     }
 361     else if (p->dirty)
 362     {
 363         int size = p->size + ISAMB_DATA_OFFSET;
 364         p->buf[0] = p->leaf;
 365         p->buf[1] = size & 255;
 366         p->buf[2] = size >> 8;
 367         if (!get_block (b, p->pos, p->buf, 1))
 368         {
 369             yaz_log (b->log_io, "bf_write: close_block");
 370             bf_write (b->file[p->cat].bf, p->pos/4, 0, 0, p->buf);
 371         }
 372     }
 373     (*b->method->code_stop)(ISAMC_DECODE, p->decodeClientData);
 374     xfree (p->buf);
 375     xfree (p);
 376 }
 377
 378 int insert_sub (ISAMB b, struct ISAMB_block **p,
 379                 void *new_item, int *mode,
 380                 ISAMC_I stream,
 381                 struct ISAMB_block **sp,
 382                 void *sub_item, int *sub_size,
 383                 void *max_item);
 384
 385 int insert_int (ISAMB b, struct ISAMB_block *p, void *lookahead_item,
 386                 int *mode,
 387                 ISAMC_I stream, struct ISAMB_block **sp,
 388                 void *split_item, int *split_size, void *last_max_item)
 389 {
 390     char *startp = p->bytes;
 391     char *src = startp;
 392     char *endp = p->bytes + p->size;
 393     int pos;
 394     struct ISAMB_block *sub_p1 = 0, *sub_p2 = 0;
 395     char sub_item[DST_ITEM_MAX];
 396     int sub_size;
 397     int more;
 398
 399     *sp = 0;
 400
 401     decode_ptr (&src, &pos);
 402     while (src != endp)
 403     {
 404         int item_len;
 405         int d;
 406         decode_ptr (&src, &item_len);
 407         d = (*b->method->compare_item)(src, lookahead_item);
 408         if (d > 0)
 409         {
 410             sub_p1 = open_block (b, pos);
 411             assert (sub_p1);
 412             more = insert_sub (b, &sub_p1, lookahead_item, mode,
 413                                stream, &sub_p2,
 414                                sub_item, &sub_size, src);
 415             break;
 416         }
 417         src += item_len;
 418         decode_ptr (&src, &pos);
 419     }
 420     if (!sub_p1)
 421     {
 422         sub_p1 = open_block (b, pos);
 423         assert (sub_p1);
 424         more = insert_sub (b, &sub_p1, lookahead_item, mode, stream, &sub_p2,
 425                            sub_item, &sub_size, last_max_item);
 426     }
 427     if (sub_p2)
 428     {
 429         /* there was a split - must insert pointer in this one */
 430         char dst_buf[DST_BUF_SIZE];
 431         char *dst = dst_buf;
 432
 433         assert (sub_size < 30 && sub_size > 1);
 434
 435         memcpy (dst, startp, src - startp);
 436
 437         dst += src - startp;
 438
 439         encode_ptr (&dst, sub_size);      /* sub length and item */
 440         memcpy (dst, sub_item, sub_size);
 441         dst += sub_size;
 442
 443         encode_ptr (&dst, sub_p2->pos);   /* pos */
 444
 445         if (endp - src)                   /* remaining data */
 446         {
 447             memcpy (dst, src, endp - src);
 448             dst += endp - src;
 449         }
 450         p->size = dst - dst_buf;
 451         if (p->size <= b->file[p->cat].head.block_max)
 452         {
 453             memcpy (startp, dst_buf, dst - dst_buf);
 454         }
 455         else
 456         {
 457             int p_new_size;
 458             char *half;
 459             src = dst_buf;
 460             endp = dst;
 461
 462             half = src + b->file[p->cat].head.block_size/2;
 463             decode_ptr (&src, &pos);
 464             while (src <= half)
 465             {
 466                 decode_ptr (&src, split_size);
 467                 src += *split_size;
 468                 decode_ptr (&src, &pos);
 469             }
 470             p_new_size = src - dst_buf;
 471             memcpy (p->bytes, dst_buf, p_new_size);
 472
 473             decode_ptr (&src, split_size);
 474             memcpy (split_item, src, *split_size);
 475             src += *split_size;
 476
 477             *sp = new_int (b, p->cat);
 478             (*sp)->size = endp - src;
 479             memcpy ((*sp)->bytes, src, (*sp)->size);
 480
 481             p->size = p_new_size;
 482         }
 483         p->dirty = 1;
 484         close_block (b, sub_p2);
 485     }
 486     close_block (b, sub_p1);
 487     return more;
 488 }
 489
 490
 491 int insert_leaf (ISAMB b, struct ISAMB_block **sp1, void *lookahead_item,
 492                  int *lookahead_mode, ISAMC_I stream, struct ISAMB_block **sp2,
 493                  void *sub_item, int *sub_size,
 494                  void *max_item)
 495 {
 496     struct ISAMB_block *p = *sp1;
 497     char *src = 0, *endp = 0;
 498     char dst_buf[DST_BUF_SIZE], *dst = dst_buf;
 499     int new_size;
 500     void *c1 = (*b->method->code_start)(ISAMC_DECODE);
 501     void *c2 = (*b->method->code_start)(ISAMC_ENCODE);
 502     int more = 1;
 503     int quater = b->file[b->no_cat-1].head.block_max / 4;
 504     char *cut = dst_buf + quater * 2;
 505     char *maxp = dst_buf + b->file[b->no_cat-1].head.block_max;
 506     char *half1 = 0;
 507     char *half2 = 0;
 508     char cut_item_buf[DST_ITEM_MAX];
 509     int cut_item_size = 0;
 510
 511     if (p && p->size)
 512     {
 513         char file_item_buf[DST_ITEM_MAX];
 514         char *file_item = file_item_buf;
 515
 516         src = p->bytes;
 517         endp = p->bytes + p->size;
 518         (*b->method->code_item)(ISAMC_DECODE, c1, &file_item, &src);
 519         while (1)
 520         {
 521             char *dst_item = 0;
 522             char *dst_0 = dst;
 523             char *lookahead_next;
 524             int d = -1;
 525
 526             if (lookahead_item)
 527                 d = (*b->method->compare_item)(file_item_buf, lookahead_item);
 528
 529             if (d > 0)
 530             {
 531                 dst_item = lookahead_item;
 532                 assert (*lookahead_mode);
 533             }
 534             else
 535                 dst_item = file_item_buf;
 536             if (!*lookahead_mode && d == 0)
 537             {
 538                 p->dirty = 1;
 539             }
 540             else if (!half1 && dst > cut)
 541             {
 542                 char *dst_item_0 = dst_item;
 543                 half1 = dst; /* candidate for splitting */
 544
 545                 (*b->method->code_item)(ISAMC_ENCODE, c2, &dst, &dst_item);
 546
 547                 cut_item_size = dst_item - dst_item_0;
 548                 memcpy (cut_item_buf, dst_item_0, cut_item_size);
 549
 550                 half2 = dst;
 551             }
 552             else
 553                 (*b->method->code_item)(ISAMC_ENCODE, c2, &dst, &dst_item);
 554             if (d > 0)
 555             {
 556                 if (dst > maxp)
 557                 {
 558                     dst = dst_0;
 559                     lookahead_item = 0;
 560                 }
 561                 else
 562                 {
 563                     lookahead_next = lookahead_item;
 564                     if (!(*stream->read_item)(stream->clientData,
 565                                               &lookahead_next,
 566                                               lookahead_mode))
 567                     {
 568                         lookahead_item = 0;
 569                         more = 0;
 570                     }
 571                     if (lookahead_item && max_item &&
 572                         (*b->method->compare_item)(max_item, lookahead_item) <= 0)
 573                     {
 574                         /* max_item 1 */
 575                         lookahead_item = 0;
 576                     }
 577
 578                     p->dirty = 1;
 579                 }
 580             }
 581             else if (d == 0)
 582             {
 583                 lookahead_next = lookahead_item;
 584                 if (!(*stream->read_item)(stream->clientData,
 585                                           &lookahead_next, lookahead_mode))
 586                 {
 587                     lookahead_item = 0;
 588                     more = 0;
 589                 }
 590                 if (src == endp)
 591                     break;
 592                 file_item = file_item_buf;
 593                 (*b->method->code_item)(ISAMC_DECODE, c1, &file_item, &src);
 594             }
 595             else
 596             {
 597                 if (src == endp)
 598                     break;
 599                 file_item = file_item_buf;
 600                 (*b->method->code_item)(ISAMC_DECODE, c1, &file_item, &src);
 601             }
 602         }
 603     }
 604     maxp = dst_buf + b->file[b->no_cat-1].head.block_max + quater;
 605     while (lookahead_item)
 606     {
 607         char *dst_item = lookahead_item;
 608         char *dst_0 = dst;
 609
 610         if (max_item &&
 611             (*b->method->compare_item)(max_item, lookahead_item) <= 0)
 612         {
 613             /* max_item 2 */
 614             break;
 615         }
 616         if (!*lookahead_mode)
 617         {
 618             yaz_log (LOG_WARN, "Inconsistent register (2)");
 619             abort();
 620         }
 621         else if (!half1 && dst > cut)
 622         {
 623             char *dst_item_0 = dst_item;
 624             half1 = dst; /* candidate for splitting */
 625
 626             (*b->method->code_item)(ISAMC_ENCODE, c2, &dst, &dst_item);
 627
 628             cut_item_size = dst_item - dst_item_0;
 629             memcpy (cut_item_buf, dst_item_0, cut_item_size);
 630
 631             half2 = dst;
 632         }
 633         else
 634             (*b->method->code_item)(ISAMC_ENCODE, c2, &dst, &dst_item);
 635
 636         if (dst > maxp)
 637         {
 638             dst = dst_0;
 639             break;
 640         }
 641         if (p)
 642             p->dirty = 1;
 643         dst_item = lookahead_item;
 644         if (!(*stream->read_item)(stream->clientData, &dst_item,
 645                                   lookahead_mode))
 646         {
 647             lookahead_item = 0;
 648             more = 0;
 649         }
 650     }
 651     new_size = dst - dst_buf;
 652     if (p && p->cat != b->no_cat-1 &&
 653         new_size > b->file[p->cat].head.block_max)
 654     {
 655         /* non-btree block will be removed */
 656         p->deleted = 1;
 657         close_block (b, p);
 658         /* delete it too!! */
 659         p = 0; /* make a new one anyway */
 660     }
 661     if (!p)
 662     {   /* must create a new one */
 663         int i;
 664         for (i = 0; i < b->no_cat; i++)
 665             if (new_size <= b->file[i].head.block_max)
 666                 break;
 667         if (i == b->no_cat)
 668             i = b->no_cat - 1;
 669         p = new_leaf (b, i);
 670     }
 671     if (new_size > b->file[p->cat].head.block_max)
 672     {
 673         char *first_dst;
 674         char *cut_item = cut_item_buf;
 675
 676         assert (half1);
 677         assert (half2);
 678
 679        /* first half */
 680         p->size = half1 - dst_buf;
 681         memcpy (p->bytes, dst_buf, half1 - dst_buf);
 682
 683         /* second half */
 684         *sp2 = new_leaf (b, p->cat);
 685
 686         (*b->method->code_reset)(c2);
 687
 688         first_dst = (*sp2)->bytes;
 689
 690         (*b->method->code_item)(ISAMC_ENCODE, c2, &first_dst, &cut_item);
 691
 692         memcpy (first_dst, half2, dst - half2);
 693
 694         (*sp2)->size = (first_dst - (*sp2)->bytes) + (dst - half2);
 695         (*sp2)->dirty = 1;
 696         p->dirty = 1;
 697         memcpy (sub_item, cut_item_buf, cut_item_size);
 698         *sub_size = cut_item_size;
 699     }
 700     else
 701     {
 702         memcpy (p->bytes, dst_buf, dst - dst_buf);
 703         p->size = new_size;
 704     }
 705     (*b->method->code_stop)(ISAMC_DECODE, c1);
 706     (*b->method->code_stop)(ISAMC_ENCODE, c2);
 707     *sp1 = p;
 708     return more;
 709 }
 710
 711 int insert_sub (ISAMB b, struct ISAMB_block **p, void *new_item,
 712                 int *mode,
 713                 ISAMC_I stream,
 714                 struct ISAMB_block **sp,
 715                 void *sub_item, int *sub_size,
 716                 void *max_item)
 717 {
 718     if (!*p || (*p)->leaf)
 719         return insert_leaf (b, p, new_item, mode, stream, sp, sub_item,
 720                             sub_size, max_item);
 721     else
 722         return insert_int (b, *p, new_item, mode, stream, sp, sub_item,
 723                            sub_size, max_item);
 724 }
 725
 726 int isamb_merge (ISAMB b, ISAMC_P pos, ISAMC_I stream)
 727 {
 728     char item_buf[DST_ITEM_MAX];
 729     char *item_ptr;
 730     int i_mode;
 731     int more;
 732
 733     if (b->cache < 0)
 734     {
 735         int more = 1;
 736         while (more)
 737         {
 738             item_ptr = item_buf;
 739             more =
 740                 (*stream->read_item)(stream->clientData, &item_ptr, &i_mode);
 741         }
 742         return 1;
 743     }
 744     item_ptr = item_buf;
 745     more = (*stream->read_item)(stream->clientData, &item_ptr, &i_mode);
 746     while (more)
 747     {
 748         struct ISAMB_block *p = 0, *sp = 0;
 749         char sub_item[DST_ITEM_MAX];
 750         int sub_size;
 751
 752         if (pos)
 753             p = open_block (b, pos);
 754         more = insert_sub (b, &p, item_buf, &i_mode, stream, &sp,
 755                             sub_item, &sub_size, 0);
 756         if (sp)
 757         {    /* increase level of tree by one */
 758             struct ISAMB_block *p2 = new_int (b, p->cat);
 759             char *dst = p2->bytes + p2->size;
 760
 761             encode_ptr (&dst, p->pos);
 762             assert (sub_size < 20);
 763             encode_ptr (&dst, sub_size);
 764             memcpy (dst, sub_item, sub_size);
 765             dst += sub_size;
 766             encode_ptr (&dst, sp->pos);
 767
 768             p2->size = dst - p2->bytes;
 769             pos = p2->pos;  /* return new super page */
 770             close_block (b, sp);
 771             close_block (b, p2);
 772         }
 773         else
 774             pos = p->pos;   /* return current one (again) */
 775         close_block (b, p);
 776     }
 777     return pos;
 778 }
 779
 780 ISAMB_PP isamb_pp_open_x (ISAMB isamb, ISAMB_P pos, int *level)
 781 {
 782     ISAMB_PP pp = xmalloc (sizeof(*pp));
 783
 784     pp->isamb = isamb;
 785     pp->block = xmalloc (10 * sizeof(*pp->block));
 786
 787     pp->pos = pos;
 788     pp->level = 0;
 789     pp->total_size = 0;
 790     pp->no_blocks = 0;
 791     while (1)
 792     {
 793         struct ISAMB_block *p = open_block (isamb, pos);
 794         char *src = p->bytes + p->offset;
 795         pp->block[pp->level] = p;
 796
 797         pp->total_size += p->size;
 798         pp->no_blocks++;
 799
 800         if (p->leaf)
 801             break;
 802
 803         decode_ptr (&src, &pos);
 804         p->offset = src - p->bytes;
 805         pp->level++;
 806     }
 807     pp->block[pp->level+1] = 0;
 808     if (level)
 809         *level = pp->level;
 810     return pp;
 811 }
 812
 813 ISAMB_PP isamb_pp_open (ISAMB isamb, ISAMB_P pos)
 814 {
 815     return isamb_pp_open_x (isamb, pos, 0);
 816 }
 817
 818 void isamb_pp_close_x (ISAMB_PP pp, int *size, int *blocks)
 819 {
 820     int i;
 821     if (!pp)
 822         return;
 823     if (size)
 824         *size = pp->total_size;
 825     if (blocks)
 826         *blocks = pp->no_blocks;
 827     for (i = 0; i <= pp->level; i++)
 828         close_block (pp->isamb, pp->block[i]);
 829     xfree (pp->block);
 830     xfree (pp);
 831 }
 832
 833 int isamb_block_info (ISAMB isamb, int cat)
 834 {
 835     if (cat >= 0 && cat < isamb->no_cat)
 836         return isamb->file[cat].head.block_size;
 837     return -1;
 838 }
 839
 840 void isamb_pp_close (ISAMB_PP pp)
 841 {
 842     isamb_pp_close_x (pp, 0, 0);
 843 }
 844
 845 int isamb_pp_read (ISAMB_PP pp, void *buf)
 846 {
 847     char *dst = buf;
 848     char *src;
 849     struct ISAMB_block *p = pp->block[pp->level];
 850     if (!p)
 851         return 0;
 852
 853     while (p->offset == p->size)
 854     {
 855         int pos, item_len;
 856         while (p->offset == p->size)
 857         {
 858             if (pp->level == 0)
 859                 return 0;
 860             close_block (pp->isamb, pp->block[pp->level]);
 861             pp->block[pp->level] = 0;
 862             (pp->level)--;
 863             p = pp->block[pp->level];
 864             assert (!p->leaf);  /* must be int */
 865         }
 866         src = p->bytes + p->offset;
 867
 868         decode_ptr (&src, &item_len);
 869         src += item_len;
 870         decode_ptr (&src, &pos);
 871
 872         p->offset = src - (char*) p->bytes;
 873
 874         ++(pp->level);
 875
 876         while (1)
 877         {
 878             pp->block[pp->level] = p = open_block (pp->isamb, pos);
 879
 880             pp->total_size += p->size;
 881             pp->no_blocks++;
 882
 883             if (p->leaf) /* leaf */
 884             {
 885                 break;
 886             }
 887             src = p->bytes + p->offset;
 888             decode_ptr (&src, &pos);
 889             p->offset = src - (char*) p->bytes;
 890             pp->level++;
 891         }
 892     }
 893     assert (p->offset < p->size);
 894     assert (p->leaf);
 895     src = p->bytes + p->offset;
 896     (*pp->isamb->method->code_item)(ISAMC_DECODE, p->decodeClientData,
 897                                     &dst, &src);
 898     p->offset = src - (char*) p->bytes;
 899     return 1;
 900 }
 901
 902 int isamb_pp_num (ISAMB_PP pp)
 903 {
 904     return 1;
 905 }