rset/rsmultior.c

   1 /* $Id: rsmultior.c,v 1.8 2004-08-31 10:43:39 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 RSFD r_open (RSET ct, int flag);
  36 static void r_close (RSFD rfd);
  37 static void r_delete (RSET ct);
  38 static void r_rewind (RSFD rfd);
  39 static int r_read (RSFD rfd, void *buf);
  40 static int r_write (RSFD rfd, const void *buf);
  41 static int r_forward(RSFD rfd, void *buf,
  42                      int (*cmpfunc)(const void *p1, const void *p2),
  43                      const void *untilbuf);
  44 static void r_pos (RSFD rfd, double *current, double *total);
  45
  46 static const struct rset_control control =
  47 {
  48     "multi-or",
  49     r_delete,
  50     r_open,
  51     r_close,
  52     r_rewind,
  53     r_forward,
  54     r_pos,
  55     r_read,
  56     r_write,
  57 };
  58
  59 const struct rset_control *rset_kind_multior = &control;
  60
  61 /* The heap structure:
  62  * The rset contains a list or rsets we are ORing together
  63  * The rfd contains a heap of heap-items, which contain
  64  * a rfd opened to those rsets, and a buffer for one key.
  65  * They also contain a ptr to the rset list in the rset
  66  * itself, for practical reasons.
  67  */
  68
  69 struct heap_item {
  70     RSFD fd;
  71     void *buf;
  72     RSET rset;
  73 };
  74
  75 struct heap {
  76     int heapnum;
  77     int heapmax;
  78     int keysize;
  79     int     (*cmp)(const void *p1, const void *p2);
  80     struct heap_item **heap; /* ptrs to the rfd */
  81 };
  82 typedef struct heap *HEAP;
  83
  84
  85 struct rset_multior_info {
  86     int     key_size;
  87     int     no_rec;
  88     int     (*cmp)(const void *p1, const void *p2);
  89     int     no_rsets;
  90     RSET    *rsets;
  91 };
  92
  93
  94 struct rset_multior_rfd {
  95     int flag;
  96     struct heap_item *items; /* we alloc and free them here */
  97     HEAP h;
  98     zint hits; /* returned so far */
  99     char *prevvalue; /* to see if we are in another record */
 100       /* FIXME - is this really needed? */
 101 };
 102
 103 #if 0
 104 static void heap_dump_item( HEAP h, int i, int level) {
 105     double cur,tot;
 106     if (i>h->heapnum)
 107         return;
 108     (void)rset_pos(h->heap[i]->rset,h->heap[i]->fd, &cur, &tot);
 109     logf(LOG_LOG," %d %*s i=%p buf=%p %0.1f/%0.1f",i, level, "",
 110                     &(h->heap[i]), h->heap[i]->buf, cur,tot );
 111     heap_dump_item(h, 2*i, level+1);
 112     heap_dump_item(h, 2*i+1, level+1);
 113 }
 114 static void heap_dump( HEAP h,char *msg) {
 115     logf(LOG_LOG, "heap dump: %s num=%d max=%d",msg, h->heapnum, h->heapmax);
 116     heap_dump_item(h,1,1);
 117 }
 118 #endif
 119
 120 static void heap_swap (HEAP h, int x, int y)
 121 {
 122     struct heap_item *swap;
 123     swap = h->heap[x];
 124     h->heap[x]=h->heap[y];
 125     h->heap[y]=swap;
 126 }
 127
 128 static int heap_cmp(HEAP h, int x, int y)
 129 {
 130     return (*h->cmp)(h->heap[x]->buf,h->heap[y]->buf);
 131 }
 132
 133 static int heap_empty(HEAP h)
 134 {
 135     return ( 0==h->heapnum );
 136 }
 137
 138 static void heap_delete (HEAP h)
 139 { /* deletes the first item in the heap, and balances the rest */
 140     int cur = 1, child = 2;
 141     h->heap[1]=0; /* been deleted */
 142     heap_swap (h, 1, h->heapnum--);
 143     while (child <= h->heapnum) {
 144         if (child < h->heapnum && heap_cmp(h,child,1+child)>0 )
 145             child++;
 146         if (heap_cmp(h,cur,child) > 0)
 147         {
 148             heap_swap (h, cur, child);
 149             cur = child;
 150             child = 2*cur;
 151         }
 152         else
 153             break;
 154     }
 155 }
 156
 157 static void heap_balance (HEAP h)
 158 { /* The heap root element has changed value (to bigger) */
 159   /* swap downwards until the heap is ordered again */
 160     int cur = 1, child = 2;
 161     while (child <= h->heapnum) {
 162         if (child < h->heapnum && heap_cmp(h,child,1+child)>0 )
 163             child++;
 164         if (heap_cmp(h,cur,child) > 0)
 165         {
 166             heap_swap (h, cur, child);
 167             cur = child;
 168             child = 2*cur;
 169         }
 170         else
 171             break;
 172     }
 173 }
 174
 175
 176 static void heap_insert (HEAP h, struct heap_item *hi)
 177 {
 178     int cur, parent;
 179
 180     cur = ++(h->heapnum);
 181     assert(cur <= h->heapmax);
 182     h->heap[cur]=hi;
 183     parent = cur/2;
 184     while (parent && (heap_cmp(h,parent,cur) > 0))
 185     {
 186         assert(parent>0);
 187         heap_swap (h, cur, parent);
 188         cur = parent;
 189         parent = cur/2;
 190     }
 191 }
 192
 193
 194 static
 195 HEAP heap_create (NMEM nmem, int size, int key_size,
 196       int (*cmp)(const void *p1, const void *p2))
 197 {
 198     HEAP h = (HEAP) nmem_malloc (nmem, sizeof(*h));
 199
 200     ++size; /* heap array starts at 1 */
 201     h->heapnum = 0;
 202     h->heapmax = size;
 203     h->keysize = key_size;
 204     h->cmp = cmp;
 205     h->heap = (struct heap_item**) nmem_malloc(nmem,(size)*sizeof(*h->heap));
 206     h->heap[0]=0; /* not used */
 207     return h;
 208 }
 209
 210 static void heap_clear( HEAP h)
 211 {
 212     assert(h);
 213     h->heapnum=0;
 214 }
 215
 216 static void heap_destroy (HEAP h)
 217 {
 218     /* nothing to delete, all is nmem'd, and will go away in due time */
 219 }
 220
 221
 222 RSET rsmultior_create( NMEM nmem, int key_size,
 223             int (*cmp)(const void *p1, const void *p2),
 224             int no_rsets, RSET* rsets)
 225 {
 226     RSET rnew=rset_create_base(&control, nmem);
 227     struct rset_multior_info *info;
 228     info = (struct rset_multior_info *) nmem_malloc(rnew->nmem,sizeof(*info));
 229     info->key_size = key_size;
 230     info->cmp = cmp;
 231     info->no_rsets=no_rsets;
 232     info->rsets=(RSET*)nmem_malloc(rnew->nmem, no_rsets*sizeof(*rsets));
 233     memcpy(info->rsets,rsets,no_rsets*sizeof(*rsets));
 234     rnew->priv=info;
 235     return rnew;
 236 }
 237
 238 static void r_delete (RSET ct)
 239 {
 240     struct rset_multior_info *info = (struct rset_multior_info *) ct->priv;
 241     int i;
 242     for(i=0;i<info->no_rsets;i++)
 243         rset_delete(info->rsets[i]);
 244 }
 245
 246 static RSFD r_open (RSET ct, int flag)
 247 {
 248     RSFD rfd;
 249     struct rset_multior_rfd *p;
 250     struct rset_multior_info *info = (struct rset_multior_info *) ct->priv;
 251     int i;
 252
 253     if (flag & RSETF_WRITE)
 254     {
 255         logf (LOG_FATAL, "multior set type is read-only");
 256         return NULL;
 257     }
 258     rfd=rfd_create_base(ct);
 259     if (rfd->priv) {
 260         p=(struct rset_multior_rfd *)rfd->priv;
 261         heap_clear(p->h);
 262         assert(p->items);
 263         /* all other pointers shouls already be allocated, in right sizes! */
 264     }
 265     else {
 266         p = (struct rset_multior_rfd *) nmem_malloc (ct->nmem,sizeof(*p));
 267         rfd->priv=p;
 268         p->h = heap_create( ct->nmem, info->no_rsets,
 269                               info->key_size, info->cmp);
 270         p->items=(struct heap_item *) nmem_malloc(ct->nmem,
 271                               info->no_rsets*sizeof(*p->items));
 272         for (i=0; i<info->no_rsets; i++){
 273             p->items[i].rset=info->rsets[i];
 274             p->items[i].buf=nmem_malloc(ct->nmem,info->key_size);
 275         }
 276     }
 277     p->flag = flag;
 278     p->prevvalue=0;
 279     p->hits=0;
 280     for (i=0; i<info->no_rsets; i++){
 281         p->items[i].fd=rset_open(info->rsets[i],RSETF_READ);
 282         if ( rset_read(p->items[i].fd, p->items[i].buf) )
 283             heap_insert(p->h, &(p->items[i]));
 284     }
 285     return rfd;
 286 }
 287
 288 static void r_close (RSFD rfd)
 289 {
 290     struct rset_multior_info *info=(struct rset_multior_info *)(rfd->rset->priv);
 291     struct rset_multior_rfd *p=(struct rset_multior_rfd *)(rfd->priv);
 292     int i;
 293
 294     heap_destroy (p->h);
 295     for (i = 0; i<info->no_rsets; i++)
 296         if (p->items[i].fd)
 297             rset_close(p->items[i].fd);
 298     rfd_delete_base(rfd);
 299 }
 300
 301
 302 static void r_rewind (RSFD rfd)
 303 {
 304     assert(!"rewind not implemented yet");
 305 }
 306
 307
 308 static int r_forward(RSFD rfd, void *buf,
 309                      int (*cmpfunc)(const void *p1, const void *p2),
 310                      const void *untilbuf)
 311 {
 312     struct rset_multior_info *info=(struct rset_multior_info *)(rfd->rset->priv);
 313     struct rset_multior_rfd *mrfd=(struct rset_multior_rfd *)(rfd->priv);
 314     struct heap_item it;
 315     int rdres;
 316     if (heap_empty(mrfd->h))
 317         return 0;
 318     if (cmpfunc)
 319         assert(cmpfunc==info->cmp);
 320     it = *(mrfd->h->heap[1]);
 321     memcpy(buf,it.buf, info->key_size);
 322     (mrfd->hits)++;
 323     if (untilbuf)
 324         rdres=rset_forward(it.fd, it.buf, cmpfunc,untilbuf);
 325     else
 326         rdres=rset_read(it.fd, it.buf);
 327     if ( rdres )
 328         heap_balance(mrfd->h);
 329     else
 330         heap_delete(mrfd->h);
 331     return 1;
 332
 333 }
 334
 335 static int r_read (RSFD rfd, void *buf)
 336 {
 337     return r_forward(rfd, buf,0,0);
 338 }
 339
 340 static void r_pos (RSFD rfd, double *current, double *total)
 341 {
 342     struct rset_multior_info *info=(struct rset_multior_info *)(rfd->rset->priv);
 343     struct rset_multior_rfd *mrfd=(struct rset_multior_rfd *)(rfd->priv);
 344     double cur, tot;
 345     double scur=0.0, stot=0.0;
 346     int i;
 347     for (i=0; i<info->no_rsets; i++){
 348         rset_pos(mrfd->items[i].fd, &cur, &tot);
 349         logf(LOG_LOG, "r_pos: %d %0.1f %0.1f", i, cur,tot);
 350         scur += cur;
 351         stot += tot;
 352     }
 353     if (stot <1.0) { /* nothing there */
 354         *current=0;
 355         *total=0;
 356         return;
 357     }
 358     *current=mrfd->hits;
 359     *total=*current*stot/scur;
 360 }
 361
 362 static int r_write (RSFD rfd, const void *buf)
 363 {
 364     logf (LOG_FATAL, "multior set type is read-only");
 365     return -1;
 366 }