isamc/merge.c

   1 /* $Id: merge.c,v 1.25 2004-08-06 12:28:23 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 <stdlib.h>
  24 #include <assert.h>
  25 #include <string.h>
  26 #include <stdio.h>
  27 #include <yaz/log.h>
  28 #include "isamc-p.h"
  29
  30 struct isc_merge_block {
  31     int offset;       /* offset in r_buf */
  32     zint block;        /* block number of file (0 if none) */
  33     int dirty;        /* block is different from that on file */
  34 };
  35
  36 #if 0
  37 static void opt_blocks (ISAMC is, struct isc_merge_block *mb, int ptr,
  38                         int last)
  39 {
  40     int i, no_dirty = 0;
  41     for (i = 0; i<ptr; i++)
  42         if (mb[i].dirty)
  43             no_dirty++;
  44     if (no_dirty*4 < ptr*3)
  45         return;
  46     /* bubble-sort it */
  47     for (i = 0; i<ptr; i++)
  48     {
  49         int tmp, j, j_min = -1;
  50         for (j = i; j<ptr; j++)
  51         {
  52             if (j_min < 0 || mb[j_min].block > mb[j].block)
  53                 j_min = j;
  54         }
  55         assert (j_min >= 0);
  56         tmp = mb[j_min].block;
  57         mb[j_min].block = mb[i].block;
  58         mb[i].block = tmp;
  59         mb[i].dirty = 1;
  60     }
  61     if (!last)
  62         mb[i].dirty = 1;
  63 }
  64 #endif
  65
  66 static void flush_blocks (ISAMC is, struct isc_merge_block *mb, int ptr,
  67                           char *r_buf, zint *firstpos, int cat, int last,
  68                           zint *numkeys)
  69 {
  70     int i;
  71
  72     for (i = 0; i<ptr; i++)
  73     {
  74         /* consider this block number */
  75         if (!mb[i].block)
  76         {
  77             mb[i].block = isc_alloc_block (is, cat);
  78             mb[i].dirty = 1;
  79         }
  80
  81         /* consider next block pointer */
  82         if (last && i == ptr-1)
  83             mb[i+1].block = 0;
  84         else if (!mb[i+1].block)
  85         {
  86             mb[i+1].block = isc_alloc_block (is, cat);
  87             mb[i+1].dirty = 1;
  88             mb[i].dirty = 1;
  89         }
  90     }
  91
  92     for (i = 0; i<ptr; i++)
  93     {
  94         char *src;
  95         ISAMC_BLOCK_SIZE ssize = mb[i+1].offset - mb[i].offset;
  96
  97         assert (ssize);
  98
  99         /* skip rest if not dirty */
 100         if (!mb[i].dirty)
 101         {
 102             assert (mb[i].block);
 103             if (!*firstpos)
 104                 *firstpos = mb[i].block;
 105             if (is->method->debug > 2)
 106                 logf (LOG_LOG, "isc: skip ptr=%d size=%d %d " ZINT_FORMAT,
 107                      i, ssize, cat, mb[i].block);
 108             ++(is->files[cat].no_skip_writes);
 109             continue;
 110         }
 111         /* write block */
 112
 113         if (!*firstpos)
 114         {
 115             *firstpos = mb[i].block;
 116             src = r_buf + mb[i].offset - ISAMC_BLOCK_OFFSET_1;
 117             ssize += ISAMC_BLOCK_OFFSET_1;
 118
 119             memcpy (src+sizeof(zint)+sizeof(ssize), numkeys, sizeof(*numkeys));
 120             if (is->method->debug > 2)
 121                 logf (LOG_LOG, "isc: flush ptr=%d numk=" ZINT_FORMAT " size=%d nextpos="
 122                       ZINT_FORMAT, i, *numkeys, (int) ssize, mb[i+1].block);
 123         }
 124         else
 125         {
 126             src = r_buf + mb[i].offset - ISAMC_BLOCK_OFFSET_N;
 127             ssize += ISAMC_BLOCK_OFFSET_N;
 128             if (is->method->debug > 2)
 129                 logf (LOG_LOG, "isc: flush ptr=%d size=%d nextpos=" ZINT_FORMAT,
 130                      i, (int) ssize, mb[i+1].block);
 131         }
 132         memcpy (src, &mb[i+1].block, sizeof(zint));
 133         memcpy (src+sizeof(zint), &ssize, sizeof(ssize));
 134         isc_write_block (is, cat, mb[i].block, src);
 135     }
 136 }
 137
 138 static int get_border (ISAMC is, struct isc_merge_block *mb, zint ptr,
 139                        int cat, zint firstpos)
 140 {
 141    /* Border set to initial fill or block size depending on
 142       whether we are creating a new one or updating and old one.
 143     */
 144
 145     int fill = mb[ptr].block ? is->method->filecat[cat].bsize :
 146                                is->method->filecat[cat].ifill;
 147     int off = (ptr||firstpos) ? ISAMC_BLOCK_OFFSET_N : ISAMC_BLOCK_OFFSET_1;
 148
 149     assert (ptr < 199);
 150
 151     return mb[ptr].offset + fill - off;
 152 }
 153
 154 ISAMC_P isc_merge (ISAMC is, ISAMC_P ipos, ISAMC_I *data)
 155 {
 156
 157     char i_item[128], *i_item_ptr;
 158     int i_more, i_mode, i;
 159
 160     ISAMC_PP pp;
 161     char f_item[128], *f_item_ptr;
 162     int f_more;
 163     int last_dirty = 0;
 164     int debug = is->method->debug;
 165
 166     struct isc_merge_block mb[200];
 167
 168     zint firstpos = 0;
 169     int cat = 0;
 170     char r_item_buf[128]; /* temporary result output */
 171     char *r_buf;          /* block with resulting data */
 172     int r_offset = 0;     /* current offset in r_buf */
 173     int ptr = 0;          /* pointer */
 174     void *r_clientData;   /* encode client data */
 175     zint border;
 176     zint numKeys = 0;
 177
 178     r_clientData = (*is->method->codec.start)();
 179     r_buf = is->merge_buf + 128;
 180
 181     pp = isc_pp_open (is, ipos);
 182     /* read first item from file. make sure f_more indicates no boundary */
 183     f_item_ptr = f_item;
 184     f_more = isc_read_item (pp, &f_item_ptr);
 185     if (f_more > 0)
 186         f_more = 1;
 187     cat = pp->cat;
 188
 189     if (debug > 1)
 190         logf (LOG_LOG, "isc: isc_merge begin %d " ZINT_FORMAT, cat, pp->pos);
 191
 192     /* read first item from i */
 193     i_item_ptr = i_item;
 194     i_more = (*data->read_item)(data->clientData, &i_item_ptr, &i_mode);
 195
 196     mb[ptr].block = pp->pos;     /* is zero if no block on disk */
 197     mb[ptr].dirty = 0;
 198     mb[ptr].offset = 0;
 199
 200     border = get_border (is, mb, ptr, cat, firstpos);
 201     while (i_more || f_more)
 202     {
 203         char *r_item = r_item_buf;
 204         int cmp;
 205
 206         if (f_more > 1)
 207         {
 208             /* block to block boundary in the original file. */
 209             f_more = 1;
 210             if (cat == pp->cat)
 211             {
 212                 /* the resulting output is of the same category as the
 213                    the original
 214                 */
 215                 if (r_offset <= mb[ptr].offset +is->method->filecat[cat].mfill)
 216                 {
 217                     /* the resulting output block is too small/empty. Delete
 218                        the original (if any)
 219                     */
 220                     if (debug > 3)
 221                         logf (LOG_LOG, "isc: release A");
 222                     if (mb[ptr].block)
 223                         isc_release_block (is, pp->cat, mb[ptr].block);
 224                     mb[ptr].block = pp->pos;
 225                     if (!mb[ptr].dirty)
 226                         mb[ptr].dirty = 1;
 227                     if (ptr > 0)
 228                         mb[ptr-1].dirty = 1;
 229                 }
 230                 else
 231                 {
 232                     /* indicate new boundary based on the original file */
 233                     mb[++ptr].block = pp->pos;
 234                     mb[ptr].dirty = last_dirty;
 235                     mb[ptr].offset = r_offset;
 236                     if (debug > 3)
 237                         logf (LOG_LOG, "isc: bound ptr=%d,offset=%d",
 238                             ptr, r_offset);
 239                     if (cat==is->max_cat && ptr >= is->method->max_blocks_mem)
 240                     {
 241                         /* We are dealing with block(s) of max size. Block(s)
 242                            except 1 will be flushed.
 243                          */
 244                         if (debug > 2)
 245                             logf (LOG_LOG, "isc: flush A %d sections", ptr);
 246                         flush_blocks (is, mb, ptr-1, r_buf, &firstpos, cat,
 247                                       0, &pp->numKeys);
 248                         mb[0].block = mb[ptr-1].block;
 249                         mb[0].dirty = mb[ptr-1].dirty;
 250                         memcpy (r_buf, r_buf + mb[ptr-1].offset,
 251                                 mb[ptr].offset - mb[ptr-1].offset);
 252                         mb[0].offset = 0;
 253
 254                         mb[1].block = mb[ptr].block;
 255                         mb[1].dirty = mb[ptr].dirty;
 256                         mb[1].offset = mb[ptr].offset - mb[ptr-1].offset;
 257                         ptr = 1;
 258                         r_offset = mb[ptr].offset;
 259                     }
 260                 }
 261             }
 262             border = get_border (is, mb, ptr, cat, firstpos);
 263         }
 264         last_dirty = 0;
 265         if (!f_more)
 266             cmp = -1;
 267         else if (!i_more)
 268             cmp = 1;
 269         else
 270             cmp = (*is->method->compare_item)(i_item, f_item);
 271         if (cmp == 0)                   /* insert i=f */
 272         {
 273             if (!i_mode)   /* delete item? */
 274             {
 275                 /* move i */
 276                 i_item_ptr = i_item;
 277                 i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 278                                            &i_mode);
 279                 /* is next input item the same as current except
 280                    for the delete flag? */
 281                 cmp = (*is->method->compare_item)(i_item, f_item);
 282                 if (!cmp && i_mode)   /* delete/insert nop? */
 283                 {
 284                     /* yes! insert as if it was an insert only */
 285                     memcpy (r_item, i_item, i_item_ptr - i_item);
 286                     i_item_ptr = i_item;
 287                     i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 288                                                 &i_mode);
 289                 }
 290                 else
 291                 {
 292                     /* no! delete the item */
 293                     r_item = NULL;
 294                     last_dirty = 1;
 295                     mb[ptr].dirty = 2;
 296                 }
 297             }
 298             else
 299             {
 300                 memcpy (r_item, f_item, f_item_ptr - f_item);
 301
 302                 /* move i */
 303                 i_item_ptr = i_item;
 304                 i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 305                                            &i_mode);
 306             }
 307             /* move f */
 308             f_item_ptr = f_item;
 309             f_more = isc_read_item (pp, &f_item_ptr);
 310         }
 311         else if (cmp > 0)               /* insert f */
 312         {
 313             memcpy (r_item, f_item, f_item_ptr - f_item);
 314             /* move f */
 315             f_item_ptr = f_item;
 316             f_more = isc_read_item (pp, &f_item_ptr);
 317         }
 318         else                            /* insert i */
 319         {
 320             if (!i_mode)                /* delete item which isn't there? */
 321             {
 322                 logf (LOG_FATAL, "Inconsistent register at offset %d",
 323                                  r_offset);
 324                 abort ();
 325             }
 326             memcpy (r_item, i_item, i_item_ptr - i_item);
 327             mb[ptr].dirty = 2;
 328             last_dirty = 1;
 329             /* move i */
 330             i_item_ptr = i_item;
 331             i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 332                                         &i_mode);
 333         }
 334         if (r_item)  /* insert resulting item? */
 335         {
 336             char *r_out_ptr = r_buf + r_offset;
 337             const char *src = r_item;
 338             int new_offset;
 339
 340             (*is->method->codec.encode)(r_clientData, &r_out_ptr, &src);
 341             new_offset = r_out_ptr - r_buf;
 342
 343             numKeys++;
 344
 345             if (border < new_offset && border >= r_offset)
 346             {
 347                 if (debug > 2)
 348                     logf (LOG_LOG, "isc: border %d %d", ptr, border);
 349                 /* Max size of current block category reached ...
 350                    make new virtual block entry */
 351                 mb[++ptr].block = 0;
 352                 mb[ptr].dirty = 1;
 353                 mb[ptr].offset = r_offset;
 354                 if (cat == is->max_cat && ptr >= is->method->max_blocks_mem)
 355                 {
 356                     /* We are dealing with block(s) of max size. Block(s)
 357                        except one will be flushed. Note: the block(s) are
 358                        surely not the last one(s).
 359                      */
 360                     if (debug > 2)
 361                         logf (LOG_LOG, "isc: flush B %d sections", ptr-1);
 362                     flush_blocks (is, mb, ptr-1, r_buf, &firstpos, cat,
 363                                   0, &pp->numKeys);
 364                     mb[0].block = mb[ptr-1].block;
 365                     mb[0].dirty = mb[ptr-1].dirty;
 366                     memcpy (r_buf, r_buf + mb[ptr-1].offset,
 367                             mb[ptr].offset - mb[ptr-1].offset);
 368                     mb[0].offset = 0;
 369
 370                     mb[1].block = mb[ptr].block;
 371                     mb[1].dirty = mb[0].dirty;
 372                     mb[1].offset = mb[ptr].offset - mb[ptr-1].offset;
 373                     memcpy (r_buf + mb[1].offset, r_buf + r_offset,
 374                             new_offset - r_offset);
 375                     new_offset = (new_offset - r_offset) + mb[1].offset;
 376                     ptr = 1;
 377                 }
 378                 border = get_border (is, mb, ptr, cat, firstpos);
 379             }
 380             r_offset = new_offset;
 381         }
 382         if (cat < is->max_cat && ptr >= is->method->filecat[cat].mblocks)
 383         {
 384             /* Max number blocks in current category reached ->
 385                must switch to next category (with larger block size)
 386             */
 387             int j = 0;
 388
 389             (is->files[cat].no_remap)++;
 390             /* delete all original block(s) read so far */
 391             for (i = 0; i < ptr; i++)
 392                 if (mb[i].block)
 393                     isc_release_block (is, pp->cat, mb[i].block);
 394             /* also delete all block to be read in the future */
 395             pp->deleteFlag = 1;
 396
 397             /* remap block offsets */
 398             assert (mb[j].offset == 0);
 399             cat++;
 400             mb[j].dirty = 1;
 401             mb[j].block = 0;
 402             mb[ptr].offset = r_offset;
 403             for (i = 1; i < ptr; i++)
 404             {
 405                 int border = is->method->filecat[cat].ifill -
 406                          ISAMC_BLOCK_OFFSET_1 + mb[j].offset;
 407                 if (debug > 3)
 408                     logf (LOG_LOG, "isc: remap %d border=%d", i, border);
 409                 if (mb[i+1].offset > border && mb[i].offset <= border)
 410                 {
 411                     if (debug > 3)
 412                         logf (LOG_LOG, "isc:  to %d %d", j, mb[i].offset);
 413                     mb[++j].dirty = 1;
 414                     mb[j].block = 0;
 415                     mb[j].offset = mb[i].offset;
 416                 }
 417             }
 418             if (debug > 2)
 419                 logf (LOG_LOG, "isc: remap from %d to %d sections to cat %d",
 420                       ptr, j, cat);
 421             ptr = j;
 422             border = get_border (is, mb, ptr, cat, firstpos);
 423             if (debug > 3)
 424                 logf (LOG_LOG, "isc: border=%d r_offset=%d", border, r_offset);
 425         }
 426     }
 427     if (mb[ptr].offset < r_offset)
 428     {   /* make the final boundary offset */
 429         mb[++ptr].dirty = 1;
 430         mb[ptr].block = 0;
 431         mb[ptr].offset = r_offset;
 432     }
 433     else
 434     {   /* empty output. Release last block if any */
 435         if (cat == pp->cat && mb[ptr].block)
 436         {
 437             if (debug > 3)
 438                 logf (LOG_LOG, "isc: release C");
 439             isc_release_block (is, pp->cat, mb[ptr].block);
 440             mb[ptr].block = 0;
 441             if (ptr > 0)
 442                 mb[ptr-1].dirty = 1;
 443         }
 444     }
 445
 446     if (debug > 2)
 447         logf (LOG_LOG, "isc: flush C, %d sections", ptr);
 448
 449     if (firstpos)
 450     {
 451         /* we have to patch initial block with num keys if that
 452            has changed */
 453         if (numKeys != isc_pp_num (pp))
 454         {
 455             if (debug > 2)
 456                 logf (LOG_LOG, "isc: patch num keys firstpos=" ZINT_FORMAT " num=" ZINT_FORMAT,
 457                                 firstpos, numKeys);
 458             bf_write (is->files[cat].bf, firstpos, ISAMC_BLOCK_OFFSET_N,
 459                       sizeof(numKeys), &numKeys);
 460         }
 461     }
 462     else if (ptr > 0)
 463     {   /* we haven't flushed initial block yet and there surely are some
 464            blocks to flush. Make first block dirty if numKeys differ */
 465         if (numKeys != isc_pp_num (pp))
 466             mb[0].dirty = 1;
 467     }
 468     /* flush rest of block(s) in r_buf */
 469     flush_blocks (is, mb, ptr, r_buf, &firstpos, cat, 1, &numKeys);
 470
 471     (*is->method->codec.stop)(r_clientData);
 472     if (!firstpos)
 473         cat = 0;
 474     if (debug > 1)
 475         logf (LOG_LOG, "isc: isc_merge return %d " ZINT_FORMAT, cat, firstpos);
 476     isc_pp_close (pp);
 477     return cat + firstpos * 8;
 478 }
 479