rset/rsmultior.c

   1 /* $Id: rsmultior.c,v 1.2 2004-08-19 12:49:15 heikki Exp $
   2    Copyright (C) 1995,1996,1997,1998,1999,2000,2001,2002
   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
  24
  25
  26 #include <assert.h>
  27 #include <stdio.h>
  28 #include <stdlib.h>
  29 #include <string.h>
  30
  31 #include <zebrautl.h>
  32 #include <isamc.h>
  33 #include <rsmultior.h>
  34
  35 static void *r_create(RSET ct, const struct rset_control *sel, void *parms);
  36 static RSFD r_open (RSET ct, int flag);
  37 static void r_close (RSFD rfd);
  38 static void r_delete (RSET ct);
  39 static void r_rewind (RSFD rfd);
  40 static int r_read (RSFD rfd, void *buf, int *term_index);
  41 static int r_write (RSFD rfd, const void *buf);
  42 static int r_forward(RSET ct, RSFD rfd, void *buf,  int *term_index,
  43                      int (*cmpfunc)(const void *p1, const void *p2),
  44                      const void *untilbuf);
  45
  46 static const struct rset_control control =
  47 {
  48     "multi-or",
  49     r_create,
  50     r_open,
  51     r_close,
  52     r_delete,
  53     r_rewind,
  54     r_forward,
  55     rset_default_pos, /* FIXME */
  56     r_read,
  57     r_write,
  58 };
  59
  60 const struct rset_control *rset_kind_multior = &control;
  61
  62 /* The heap structure:
  63  * The rset contains a list or rsets we are ORing together
  64  * The rfd contains a heap of heap-items, which contain
  65  * a rfd opened to those rsets, and a buffer for one key.
  66  * They also contain a ptr to the rset list in the rset
  67  * itself, for practical reasons.
  68  */
  69
  70 struct heap_item {
  71     RSFD fd;
  72     void *buf;
  73     RSET rset;
  74 };
  75
  76 struct heap {
  77     int heapnum;
  78     int heapmax;
  79     int keysize;
  80     int     (*cmp)(const void *p1, const void *p2);
  81     struct heap_item **heap; /* ptrs to the rfd */
  82 };
  83 typedef struct heap *HEAP;
  84
  85
  86 struct rset_multior_info {
  87     int     key_size;
  88     int     no_rec;
  89     int     (*cmp)(const void *p1, const void *p2);
  90     int     no_rsets;
  91     RSET    *rsets;
  92     int     term_index;
  93     struct rset_multior_rfd *rfd_list;
  94 };
  95
  96
  97 struct rset_multior_rfd {
  98     int flag;
  99     struct heap_item *items; /* we alloc and free them here */
 100     HEAP h;
 101     struct rset_multior_rfd *next;
 102     struct rset_multior_info *info;
 103     zint *countp;
 104     char *prevvalue; /* to see if we are in another record */
 105 };
 106
 107 #if 0
 108 static void heap_dump_item( HEAP h, int i, int level) {
 109     double cur,tot;
 110     if (i>h->heapnum)
 111         return;
 112     (void)rset_pos(h->heap[i]->rset,h->heap[i]->fd, &cur, &tot);
 113     logf(LOG_LOG," %d %*s i=%p buf=%p %0.1f/%0.1f",i, level, "",
 114                     &(h->heap[i]), h->heap[i]->buf, cur,tot );
 115     heap_dump_item(h, 2*i, level+1);
 116     heap_dump_item(h, 2*i+1, level+1);
 117 }
 118 static void heap_dump( HEAP h,char *msg) {
 119     logf(LOG_LOG, "heap dump: %s num=%d max=%d",msg, h->heapnum, h->heapmax);
 120     heap_dump_item(h,1,1);
 121 }
 122 #endif
 123
 124 static void heap_swap (HEAP h, int x, int y)
 125 {
 126     struct heap_item *swap;
 127     swap = h->heap[x];
 128     h->heap[x]=h->heap[y];
 129     h->heap[y]=swap;
 130 }
 131
 132 static int heap_cmp(HEAP h, int x, int y)
 133 {
 134     return (*h->cmp)(h->heap[x]->buf,h->heap[y]->buf);
 135 }
 136
 137 static int heap_empty(HEAP h)
 138 {
 139     return ( 0==h->heapnum );
 140 }
 141
 142 static void heap_delete (HEAP h)
 143 { /* deletes the first item in the heap, and balances the rest */
 144     int cur = 1, child = 2;
 145     h->heap[1]=0; /* been deleted */
 146     heap_swap (h, 1, h->heapnum--);
 147     while (child <= h->heapnum) {
 148         if (child < h->heapnum && heap_cmp(h,child,1+child)>0 )
 149             child++;
 150         if (heap_cmp(h,cur,child) > 0)
 151         {
 152             heap_swap (h, cur, child);
 153             cur = child;
 154             child = 2*cur;
 155         }
 156         else
 157             break;
 158     }
 159 }
 160
 161 static void heap_balance (HEAP h)
 162 { /* The heap root element has changed value (to bigger) */
 163   /* swap downwards until the heap is ordered again */
 164     int cur = 1, child = 2;
 165     while (child <= h->heapnum) {
 166         if (child < h->heapnum && heap_cmp(h,child,1+child)>0 )
 167             child++;
 168         if (heap_cmp(h,cur,child) > 0)
 169         {
 170             heap_swap (h, cur, child);
 171             cur = child;
 172             child = 2*cur;
 173         }
 174         else
 175             break;
 176     }
 177 }
 178
 179
 180 static void heap_insert (HEAP h, struct heap_item *hi)
 181 {
 182     int cur, parent;
 183
 184     cur = ++(h->heapnum);
 185     assert(cur <= h->heapmax);
 186     h->heap[cur]=hi;
 187     parent = cur/2;
 188     while (parent && (heap_cmp(h,parent,cur) > 0))
 189     {
 190         assert(parent>0);
 191         heap_swap (h, cur, parent);
 192         cur = parent;
 193         parent = cur/2;
 194     }
 195 }
 196
 197
 198 static
 199 HEAP heap_create (int size, int key_size,
 200       int (*cmp)(const void *p1, const void *p2))
 201 {
 202     HEAP h = (HEAP) xmalloc (sizeof(*h));
 203
 204     ++size; /* heap array starts at 1 */
 205     h->heapnum = 0;
 206     h->heapmax = size;
 207     h->keysize = key_size;
 208     h->cmp = cmp;
 209     h->heap = (struct heap_item**) xmalloc((size)*sizeof(*h->heap));
 210     h->heap[0]=0; /* not used */
 211     return h;
 212 }
 213
 214 static void heap_destroy (HEAP h)
 215 {
 216     xfree (h->heap); /* safe, they all point to the rfd */
 217     xfree (h);
 218 }
 219
 220
 221 static void *r_create (RSET ct, const struct rset_control *sel, void *parms)
 222 {
 223     rset_multior_parms *r_parms = (rset_multior_parms *) parms;
 224     struct rset_multior_info *info;
 225
 226     ct->flags |= RSET_FLAG_VOLATILE;
 227         /* FIXME - Remove the whole flags thing, from all rsets */
 228     info = (struct rset_multior_info *) xmalloc (sizeof(*info));
 229     info->key_size = r_parms->key_size;
 230     assert (info->key_size > 1);
 231     info->cmp = r_parms->cmp;
 232     info->no_rsets= r_parms->no_rsets;
 233     info->rsets=r_parms->rsets; /* now we own it! */
 234     info->rfd_list=0;
 235     info->term_index=0 ; /* r_parms->rset_term; */ /*??*/ /*FIXME */
 236     ct->no_rset_terms = 1;
 237     ct->rset_terms = (RSET_TERM *) xmalloc (sizeof(*ct->rset_terms));
 238     ct->rset_terms[0] = r_parms->rset_term;
 239     return info;
 240 }
 241
 242 static RSFD r_open (RSET ct, int flag)
 243 {
 244     struct rset_multior_rfd *rfd;
 245     struct rset_multior_info *info = (struct rset_multior_info *) ct->buf;
 246     int i;
 247     int dummy_termindex;
 248
 249     if (flag & RSETF_WRITE)
 250     {
 251         logf (LOG_FATAL, "multior set type is read-only");
 252         return NULL;
 253     }
 254     rfd = (struct rset_multior_rfd *) xmalloc (sizeof(*rfd));
 255     rfd->flag = flag;
 256     rfd->next = info->rfd_list;
 257     rfd->info = info;
 258     info->rfd_list = rfd;
 259     if (ct->no_rset_terms==1)
 260         rfd->countp=&ct->rset_terms[0]->count;
 261     else
 262         rfd->countp=0;
 263     rfd->h = heap_create( info->no_rsets, info->key_size, info->cmp);
 264     rfd->prevvalue=0;
 265     rfd->items=(struct heap_item *) xmalloc(info->no_rsets*sizeof(*rfd->items));
 266     for (i=0; i<info->no_rsets; i++){
 267        rfd->items[i].rset=info->rsets[i];
 268        rfd->items[i].buf=xmalloc(info->key_size);
 269        rfd->items[i].fd=rset_open(info->rsets[i],RSETF_READ);
 270 /*       if (item_readbuf(&(rfd->items[i]))) */
 271        if ( rset_read(rfd->items[i].rset, rfd->items[i].fd,
 272                       rfd->items[i].buf, &dummy_termindex) )
 273            heap_insert(rfd->h, &(rfd->items[i]));
 274     }
 275     return rfd;
 276 }
 277
 278 static void r_close (RSFD rfd)
 279 {
 280     struct rset_multior_rfd *mrfd = (struct rset_multior_rfd *) rfd;
 281     struct rset_multior_info *info = mrfd->info;
 282     struct rset_multior_rfd **rfdp;
 283     int i;
 284
 285     for (rfdp = &info->rfd_list; *rfdp; rfdp = &(*rfdp)->next)
 286         if (*rfdp == rfd)
 287         {
 288             *rfdp = (*rfdp)->next;
 289
 290             heap_destroy (mrfd->h);
 291             for (i = 0; i<info->no_rsets; i++) {
 292                 if (mrfd->items[i].fd)
 293                     rset_close(info->rsets[i],mrfd->items[i].fd);
 294                 xfree(mrfd->items[i].buf);
 295             }
 296             xfree(mrfd->items);
 297             if (mrfd->prevvalue)
 298                 xfree(mrfd->prevvalue);
 299             xfree(mrfd);
 300             return;
 301         }
 302     logf (LOG_FATAL, "r_close but no rfd match!");
 303     assert (0);
 304 }
 305
 306 static void r_delete (RSET ct)
 307 {
 308     struct rset_multior_info *info = (struct rset_multior_info *) ct->buf;
 309     int i;
 310
 311     assert (info->rfd_list == NULL);
 312     for(i=0;i<info->no_rsets;i++)
 313         rset_delete(info->rsets[i]);
 314     xfree(info->rsets);
 315     xfree(info);
 316
 317     for (i = 0; i<ct->no_rset_terms; i++) /* usually only 1 */
 318         rset_term_destroy (ct->rset_terms[i]);
 319     xfree (ct->rset_terms);
 320 }
 321
 322 static void r_rewind (RSFD rfd)
 323 {
 324     assert(!"rewind not implemented yet");
 325 }
 326
 327
 328 static int r_forward(RSET ct, RSFD rfd, void *buf,  int *term_index,
 329                      int (*cmpfunc)(const void *p1, const void *p2),
 330                      const void *untilbuf)
 331 {
 332     struct rset_multior_rfd *mrfd = (struct rset_multior_rfd *) rfd;
 333     struct rset_multior_info *info = mrfd->info;
 334     struct heap_item it;
 335     int rdres;
 336     int dummycount=0;
 337     if (heap_empty(mrfd->h))
 338         return 0;
 339     if (cmpfunc)
 340         assert(cmpfunc==mrfd->info->cmp);
 341     *term_index=0;
 342     it = *(mrfd->h->heap[1]);
 343     memcpy(buf,it.buf, info->key_size);
 344     if (mrfd->countp) {
 345         if (mrfd->prevvalue) {  /* in another record */
 346             if ( (*mrfd->info->cmp)(mrfd->prevvalue,it.buf) < -1)
 347                 (*mrfd->countp)++;
 348         } else {
 349             mrfd->prevvalue=xmalloc(info->key_size);
 350             (*mrfd->countp)++;
 351         }
 352         memcpy(mrfd->prevvalue,it.buf, info->key_size);
 353     }
 354     if (untilbuf)
 355         rdres=rset_forward(it.rset, it.fd, it.buf, &dummycount,
 356                            cmpfunc,untilbuf);
 357     else
 358         rdres=rset_read(it.rset, it.fd, it.buf, &dummycount);
 359     if ( rdres )
 360         heap_balance(mrfd->h);
 361     else
 362         heap_delete(mrfd->h);
 363     return 1;
 364
 365 }
 366
 367 static int r_read (RSFD rfd, void *buf, int *term_index)
 368 {
 369     return r_forward(0,rfd, buf, term_index,0,0);
 370 }
 371
 372 #if 0
 373 static int old_read
 374 {
 375     struct rset_multior_rfd *mrfd = (struct rset_multior_rfd *) rfd;
 376     struct trunc_info *ti = mrfd->ti;
 377     int n = ti->indx[ti->ptr[1]];
 378
 379     if (!ti->heapnum)
 380         return 0;
 381     *term_index = 0;
 382     memcpy (buf, ti->heap[ti->ptr[1]], ti->keysize);
 383     if (((struct rset_multior_rfd *) rfd)->position)
 384     {
 385         if (isc_pp_read (((struct rset_multior_rfd *) rfd)->ispt[n], ti->tmpbuf))
 386         {
 387             heap_delete (ti);
 388             if ((*ti->cmp)(ti->tmpbuf, ti->heap[ti->ptr[1]]) > 1)
 389                  ((struct rset_multior_rfd *) rfd)->position--;
 390             heap_insert (ti, ti->tmpbuf, n);
 391         }
 392         else
 393             heap_delete (ti);
 394         if (mrfd->countp && (
 395                 *mrfd->countp == 0 || (*ti->cmp)(buf, mrfd->pbuf) > 1))
 396         {
 397             memcpy (mrfd->pbuf, buf, ti->keysize);
 398             (*mrfd->countp)++;
 399         }
 400         return 1;
 401     }
 402     while (1)
 403     {
 404         if (!isc_pp_read (((struct rset_multior_rfd *) rfd)->ispt[n], ti->tmpbuf))
 405         {
 406             heap_delete (ti);
 407             break;
 408         }
 409         if ((*ti->cmp)(ti->tmpbuf, ti->heap[ti->ptr[1]]) > 1)
 410         {
 411             heap_delete (ti);
 412             heap_insert (ti, ti->tmpbuf, n);
 413             break;
 414         }
 415     }
 416     if (mrfd->countp && (
 417             *mrfd->countp == 0 || (*ti->cmp)(buf, mrfd->pbuf) > 1))
 418     {
 419         memcpy (mrfd->pbuf, buf, ti->keysize);
 420         (*mrfd->countp)++;
 421     }
 422     return 1;
 423 }
 424
 425 #endif
 426 /*!*/ /* FIXME Forward */
 427 /*!*/ /* FIXME pos */
 428
 429 static int r_write (RSFD rfd, const void *buf)
 430 {
 431     logf (LOG_FATAL, "multior set type is read-only");
 432     return -1;
 433 }