isamc/merge.c

   1 /*
   2  * Copyright (c) 1996, Index Data.
   3  * See the file LICENSE for details.
   4  * Sebastian Hammer, Adam Dickmeiss
   5  *
   6  * $Log: merge.c,v $
   7  * Revision 1.2  1996-11-01 13:36:46  adam
   8  * New element, max_blocks_mem, that control how many blocks of max size
   9  * to store in memory during isc_merge.
  10  * Function isc_merge now ignoreds delete/update of identical keys and
  11  * the proper blocks are then non-dirty and not written in flush_blocks.
  12  *
  13  * Revision 1.1  1996/11/01  08:59:15  adam
  14  * First version of isc_merge that supports update/delete.
  15  *
  16  */
  17
  18 #include <stdlib.h>
  19 #include <assert.h>
  20 #include <string.h>
  21 #include <stdio.h>
  22
  23 #include <log.h>
  24 #include "isamc-p.h"
  25
  26 struct isc_merge_block {
  27     int offset;       /* offset in r_buf */
  28     int block;        /* block number of file (0 if none) */
  29     int dirty;        /* block is different from that on file */
  30 };
  31
  32 static void flush_blocks (ISAMC is, struct isc_merge_block *mb, int ptr,
  33                           char *r_buf, int *firstpos, int cat, int last)
  34 {
  35     int i;
  36
  37     for (i = 0; i<ptr; i++)
  38     {
  39         /* skip rest if not dirty */
  40         if (!mb[i].dirty)
  41         {
  42             assert (mb[i].block);
  43             if (!*firstpos)
  44                 *firstpos = mb[i].block;
  45             if (is->method->debug > 2)
  46                 logf (LOG_LOG, "isc: skip block %d %d", cat, mb[i].block);
  47             ++(is->files[cat].no_skip_writes);
  48             continue;
  49         }
  50         /* consider this block number */
  51
  52         if (!mb[i].block)
  53             mb[i].block = isc_alloc_block (is, cat);
  54         if (!*firstpos)
  55             *firstpos = mb[i].block;
  56
  57         /* consider next block pointer */
  58         if (last && i == ptr-1)
  59             mb[i+1].block = 0;
  60         else if (!mb[i+1].block)
  61             mb[i+1].block = isc_alloc_block (is, cat);
  62
  63         /* write block */
  64         assert (mb[i+1].offset > mb[i].offset);
  65         isc_write_dblock (is, cat, mb[i].block, r_buf + mb[i].offset,
  66                           mb[i+1].block, mb[i+1].offset - mb[i].offset);
  67     }
  68 }
  69
  70 ISAMC_P isc_merge (ISAMC is, ISAMC_P ipos, ISAMC_I data)
  71 {
  72
  73     char i_item[128], *i_item_ptr;
  74     int i_more, i_mode, i;
  75
  76     ISAMC_PP pp;
  77     char f_item[128], *f_item_ptr;
  78     int f_more;
  79
  80     struct isc_merge_block mb[100];
  81     int firstpos = 0;
  82     int cat = 0;
  83     char r_item_buf[128]; /* temporary result output */
  84     char *r_buf;          /* block with resulting data */
  85     int r_offset = 0;     /* current offset in r_buf */
  86     int ptr = 0;          /* pointer */
  87     void *r_clientData;   /* encode client data */
  88
  89     r_clientData = (*is->method->code_start)(ISAMC_ENCODE);
  90     r_buf = is->merge_buf + ISAMC_BLOCK_OFFSET;
  91
  92     pp = isc_pp_open (is, ipos);
  93     /* read first item from file. make sure f_more indicates no boundary */
  94     f_item_ptr = f_item;
  95     f_more = isc_read_item (pp, &f_item_ptr);
  96     if (f_more > 0)
  97         f_more = 1;
  98     cat = pp->cat;
  99
 100     if (is->method->debug > 1)
 101         logf (LOG_LOG, "isc: isc_merge begin %d %d", cat, pp->pos);
 102
 103     /* read first item from i */
 104     i_item_ptr = i_item;
 105     i_more = (*data->read_item)(data->clientData, &i_item_ptr, &i_mode);
 106
 107     mb[ptr].block = pp->pos;     /* is zero if no block on disk */
 108     mb[ptr].dirty = 0;
 109     mb[ptr].offset = 0;
 110
 111     while (i_more || f_more)
 112     {
 113         char *r_item = r_item_buf;
 114         int cmp;
 115
 116         if (f_more > 1)
 117         {
 118             /* block to block boundary in the original file. */
 119             f_more = 1;
 120             if (cat == pp->cat)
 121             {
 122                 /* the resulting output is of the same category as the
 123                    the original
 124                  */
 125                 if (mb[ptr].offset == r_offset)
 126                 {
 127                     /* the resulting output block is empty. Delete
 128                        the original (if any)
 129                      */
 130                     if (is->method->debug > 3)
 131                         logf (LOG_LOG, "isc: release A");
 132                     if (mb[ptr].block)
 133                         isc_release_block (is, pp->cat, mb[ptr].block);
 134                     mb[ptr].block = pp->pos;
 135                     mb[ptr].dirty = 0;
 136                 }
 137                 else
 138                 {
 139                     /* indicate new boundary based on the original file */
 140                     mb[++ptr].block = pp->pos;
 141                     mb[ptr].dirty = 0;
 142                     mb[ptr].offset = r_offset;
 143 #if 0
 144                     if (r_item && cat == is->max_cat)
 145                     {
 146                         /* We are dealing with block(s) of max size. Block(s)
 147                            will be flushed. Note: the block(s) are surely not
 148                            the last one(s).
 149                          */
 150                         if (is->method->debug > 2)
 151                             logf (LOG_LOG, "isc: flush A %d sections", ptr);
 152                         flush_blocks (is, mb, ptr-1, r_buf, &firstpos, cat, 0);
 153                         ptr = 0;
 154                         mb[ptr].block = pp->pos;
 155                         mb[ptr].dirty = 0;
 156                         mb[ptr].offset = 0;
 157                     }
 158 #endif
 159                 }
 160             }
 161         }
 162         if (!f_more)
 163             cmp = -1;
 164         else if (!i_more)
 165             cmp = 1;
 166         else
 167             cmp = (*is->method->compare_item)(i_item, f_item);
 168         if (cmp == 0)                   /* insert i=f */
 169         {
 170             if (!i_mode)   /* delete item? */
 171             {
 172                 /* move i */
 173                 i_item_ptr = i_item;
 174                 i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 175                                            &i_mode);
 176                 /* it next input item the same as current except
 177                    for the delete flag? */
 178                 cmp = (*is->method->compare_item)(i_item, f_item);
 179                 if (!cmp && i_mode)
 180                 {
 181                     /* yes! insert as if it was an insert only */
 182                     memcpy (r_item, i_item, i_item_ptr - i_item);
 183                     i_item_ptr = i_item;
 184                     i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 185                                                 &i_mode);
 186                 }
 187                 else
 188                 {
 189                     /* no! delete the item */
 190                     r_item = NULL;
 191                     mb[ptr].dirty = 1;
 192                 }
 193             }
 194             else
 195             {
 196                 memcpy (r_item, f_item, f_item_ptr - f_item);
 197
 198                 /* move i */
 199                 i_item_ptr = i_item;
 200                 i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 201                                            &i_mode);
 202             }
 203             /* move f */
 204             f_item_ptr = f_item;
 205             f_more = isc_read_item (pp, &f_item_ptr);
 206         }
 207         else if (cmp > 0)               /* insert f */
 208         {
 209             memcpy (r_item, f_item, f_item_ptr - f_item);
 210             /* move f */
 211             f_item_ptr = f_item;
 212             f_more = isc_read_item (pp, &f_item_ptr);
 213         }
 214         else                            /* insert i */
 215         {
 216             if (!i_mode)                /* delete item which isn't there? */
 217             {
 218                 logf (LOG_FATAL, "Inconsistent register at offset %d",
 219                                  r_offset);
 220                 abort ();
 221             }
 222             memcpy (r_item, i_item, i_item_ptr - i_item);
 223             mb[ptr].dirty = 1;
 224             /* move i */
 225             i_item_ptr = i_item;
 226             i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 227                                         &i_mode);
 228         }
 229         if (r_item)  /* insert resulting item? */
 230         {
 231             char *r_out_ptr = r_buf + r_offset;
 232             int new_offset;
 233             int border;
 234
 235             /* border set to initial fill or block size depending on
 236                whether we are creating a new one or updating and old one
 237              */
 238             if (mb[ptr].block)
 239                 border = mb[ptr].offset + is->method->filecat[cat].bsize
 240                          -ISAMC_BLOCK_OFFSET;
 241             else
 242                 border = mb[ptr].offset + is->method->filecat[cat].ifill
 243                          -ISAMC_BLOCK_OFFSET;
 244
 245             (*is->method->code_item)(ISAMC_ENCODE, r_clientData,
 246                                      &r_out_ptr, &r_item);
 247             new_offset = r_out_ptr - r_buf;
 248
 249             if (border >= r_offset && border < new_offset)
 250             {
 251                 if (is->method->debug > 2)
 252                     logf (LOG_LOG, "isc: border %d %d", ptr, border);
 253                 /* Max size of current block category reached ...
 254                    make new virtual block entry */
 255                 mb[++ptr].block = 0;
 256                 mb[ptr].dirty = 1;
 257                 mb[ptr].offset = r_offset;
 258                 if (cat == is->max_cat && ptr >= is->method->max_blocks_mem)
 259                 {
 260                     /* We are dealing with block(s) of max size. Block(s)
 261                        except one will be flushed. Note: the block(s) are
 262                        surely not the last one(s).
 263                      */
 264                     if (is->method->debug > 2)
 265                         logf (LOG_LOG, "isc: flush B %d sections", ptr-1);
 266                     flush_blocks (is, mb, ptr-1, r_buf, &firstpos, cat, 0);
 267
 268                     mb[0].block = mb[ptr-1].block;
 269                     mb[0].dirty = mb[ptr-1].dirty;
 270                     memcpy (r_buf, r_buf + mb[ptr-1].offset,
 271                             mb[ptr].offset - mb[ptr-1].offset);
 272                     mb[0].offset = 0;
 273
 274                     mb[1].block = mb[ptr].block;
 275                     mb[1].dirty = mb[0].dirty;
 276                     mb[1].offset = mb[ptr].offset - mb[ptr-1].offset;
 277                     memcpy (r_buf + mb[1].offset, r_buf + r_offset,
 278                             new_offset - r_offset);
 279                     new_offset = (new_offset - r_offset) + mb[1].offset;
 280                     ptr = 1;
 281                }
 282             }
 283             r_offset = new_offset;
 284         }
 285         if (cat < is->max_cat && ptr >= is->method->filecat[cat].mblocks)
 286         {
 287             /* Max number blocks in current category reached ->
 288                must switch to next category (with larger block size)
 289             */
 290             int j = 0;
 291
 292             (is->files[cat].no_remap)++;
 293             /* delete all original block(s) read so far */
 294             for (i = 0; i < ptr; i++)
 295                 if (mb[i].block)
 296                     isc_release_block (is, pp->cat, mb[i].block);
 297             /* also delete all block to be read in the future */
 298             pp->deleteFlag = 1;
 299
 300             /* remap block offsets */
 301             assert (mb[j].offset == 0);
 302             cat++;
 303             mb[j].dirty = 1;
 304             mb[j].block = 0;
 305             for (i = 1; i < ptr; i++)
 306             {
 307                 int border = is->method->filecat[cat].ifill -
 308                          ISAMC_BLOCK_OFFSET + mb[j].offset;
 309                 if (is->method->debug > 3)
 310                     logf (LOG_LOG, "isc: remap %d border=%d", i, border);
 311                 if (mb[i+1].offset > border && mb[i].offset <= border)
 312                 {
 313                     if (is->method->debug > 3)
 314                         logf (LOG_LOG, "isc:  to %d %d", j, mb[i].offset);
 315                     mb[++j].dirty = 1;
 316                     mb[j].block = 0;
 317                     mb[j].offset = mb[i].offset;
 318                 }
 319             }
 320             if (is->method->debug > 2)
 321                 logf (LOG_LOG, "isc: remap from %d to %d sections to cat %d",
 322                       ptr, j, cat);
 323             ptr = j;
 324         }
 325     }
 326     if (mb[ptr].offset < r_offset)
 327     {   /* make the final boundary offset */
 328         mb[++ptr].dirty = 1;         /* ignored by flush_blocks */
 329         mb[ptr].block = 0;         /* ignored by flush_blocks */
 330         mb[ptr].offset = r_offset;
 331     }
 332     else
 333     {   /* empty output. Release last block if any */
 334         if (cat == pp->cat && mb[ptr].block)
 335         {
 336             if (is->method->debug > 3)
 337                 logf (LOG_LOG, "isc: release C");
 338             isc_release_block (is, pp->cat, mb[ptr].block);
 339             mb[ptr].block = 0;
 340             mb[ptr].dirty = 1;
 341         }
 342     }
 343
 344     if (is->method->debug > 2)
 345         logf (LOG_LOG, "isc: flush C, %d sections", ptr);
 346
 347     /* flush rest of block(s) in r_buf */
 348     flush_blocks (is, mb, ptr, r_buf, &firstpos, cat, 1);
 349
 350     (*is->method->code_stop)(ISAMC_ENCODE, r_clientData);
 351     if (!firstpos)
 352         cat = 0;
 353     if (is->method->debug > 1)
 354         logf (LOG_LOG, "isc: isc_merge return %d %d", cat, firstpos);
 355     return cat + firstpos * 8;
 356 }
 357