rset/rsmultior.c

   1 /* $Id: rsmultior.c,v 1.6 2004-08-24 14:25:16 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(RSET ct, 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_open,
  50     r_close,
  51     r_delete,
  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     struct rset_multior_rfd *rfd_list;
  92 };
  93
  94
  95 struct rset_multior_rfd {
  96     int flag;
  97     struct heap_item *items; /* we alloc and free them here */
  98     HEAP h;
  99     struct rset_multior_rfd *next;
 100     struct rset_multior_info *info;
 101     zint hits; /* returned so far */
 102     char *prevvalue; /* to see if we are in another record */
 103       /* FIXME - is this really needed? */
 104 };
 105
 106 #if 0
 107 static void heap_dump_item( HEAP h, int i, int level) {
 108     double cur,tot;
 109     if (i>h->heapnum)
 110         return;
 111     (void)rset_pos(h->heap[i]->rset,h->heap[i]->fd, &cur, &tot);
 112     logf(LOG_LOG," %d %*s i=%p buf=%p %0.1f/%0.1f",i, level, "",
 113                     &(h->heap[i]), h->heap[i]->buf, cur,tot );
 114     heap_dump_item(h, 2*i, level+1);
 115     heap_dump_item(h, 2*i+1, level+1);
 116 }
 117 static void heap_dump( HEAP h,char *msg) {
 118     logf(LOG_LOG, "heap dump: %s num=%d max=%d",msg, h->heapnum, h->heapmax);
 119     heap_dump_item(h,1,1);
 120 }
 121 #endif
 122
 123 static void heap_swap (HEAP h, int x, int y)
 124 {
 125     struct heap_item *swap;
 126     swap = h->heap[x];
 127     h->heap[x]=h->heap[y];
 128     h->heap[y]=swap;
 129 }
 130
 131 static int heap_cmp(HEAP h, int x, int y)
 132 {
 133     return (*h->cmp)(h->heap[x]->buf,h->heap[y]->buf);
 134 }
 135
 136 static int heap_empty(HEAP h)
 137 {
 138     return ( 0==h->heapnum );
 139 }
 140
 141 static void heap_delete (HEAP h)
 142 { /* deletes the first item in the heap, and balances the rest */
 143     int cur = 1, child = 2;
 144     h->heap[1]=0; /* been deleted */
 145     heap_swap (h, 1, h->heapnum--);
 146     while (child <= h->heapnum) {
 147         if (child < h->heapnum && heap_cmp(h,child,1+child)>0 )
 148             child++;
 149         if (heap_cmp(h,cur,child) > 0)
 150         {
 151             heap_swap (h, cur, child);
 152             cur = child;
 153             child = 2*cur;
 154         }
 155         else
 156             break;
 157     }
 158 }
 159
 160 static void heap_balance (HEAP h)
 161 { /* The heap root element has changed value (to bigger) */
 162   /* swap downwards until the heap is ordered again */
 163     int cur = 1, child = 2;
 164     while (child <= h->heapnum) {
 165         if (child < h->heapnum && heap_cmp(h,child,1+child)>0 )
 166             child++;
 167         if (heap_cmp(h,cur,child) > 0)
 168         {
 169             heap_swap (h, cur, child);
 170             cur = child;
 171             child = 2*cur;
 172         }
 173         else
 174             break;
 175     }
 176 }
 177
 178
 179 static void heap_insert (HEAP h, struct heap_item *hi)
 180 {
 181     int cur, parent;
 182
 183     cur = ++(h->heapnum);
 184     assert(cur <= h->heapmax);
 185     h->heap[cur]=hi;
 186     parent = cur/2;
 187     while (parent && (heap_cmp(h,parent,cur) > 0))
 188     {
 189         assert(parent>0);
 190         heap_swap (h, cur, parent);
 191         cur = parent;
 192         parent = cur/2;
 193     }
 194 }
 195
 196
 197 static
 198 HEAP heap_create (int size, int key_size,
 199       int (*cmp)(const void *p1, const void *p2))
 200 {
 201     HEAP h = (HEAP) xmalloc (sizeof(*h));
 202
 203     ++size; /* heap array starts at 1 */
 204     h->heapnum = 0;
 205     h->heapmax = size;
 206     h->keysize = key_size;
 207     h->cmp = cmp;
 208     h->heap = (struct heap_item**) xmalloc((size)*sizeof(*h->heap));
 209     h->heap[0]=0; /* not used */
 210     return h;
 211 }
 212
 213 static void heap_destroy (HEAP h)
 214 {
 215     xfree (h->heap); /* safe, they all point to the rfd */
 216     xfree (h);
 217 }
 218
 219
 220 RSET rsmultior_create( NMEM nmem, int key_size,
 221             int (*cmp)(const void *p1, const void *p2),
 222             int no_rsets, RSET* rsets)
 223 {
 224     RSET rnew=rset_create_base(&control, nmem);
 225     struct rset_multior_info *info;
 226     info = (struct rset_multior_info *) nmem_malloc(rnew->nmem,sizeof(*info));
 227     info->key_size = key_size;
 228     info->cmp = cmp;
 229     info->no_rsets=no_rsets;
 230     info->rsets=(RSET*)nmem_malloc(rnew->nmem, no_rsets*sizeof(*rsets));
 231     memcpy(info->rsets,rsets,no_rsets*sizeof(*rsets));
 232     info->rfd_list = NULL;
 233     rnew->priv=info;
 234     return rnew;
 235 }
 236
 237 static void r_delete (RSET ct)
 238 {
 239     struct rset_multior_info *info = (struct rset_multior_info *) ct->priv;
 240     int i;
 241
 242     assert (info->rfd_list == NULL);
 243     for(i=0;i<info->no_rsets;i++)
 244         rset_delete(info->rsets[i]);
 245 /*    xfree(info->rsets); */ /* nmem'd */
 246 /*    xfree(info); */  /* nmem'd */
 247 }
 248 #if 0
 249 static void *r_create (RSET ct, const struct rset_control *sel, void *parms)
 250 {
 251     rset_multior_parms *r_parms = (rset_multior_parms *) parms;
 252     struct rset_multior_info *info;
 253     info = (struct rset_multior_info *) xmalloc (sizeof(*info));
 254     info->key_size = r_parms->key_size;
 255     assert (info->key_size > 1);
 256     info->cmp = r_parms->cmp;
 257     info->no_rsets= r_parms->no_rsets;
 258     info->rsets=r_parms->rsets; /* now we own it! */
 259     info->rfd_list=0;
 260     return info;
 261 }
 262 #endif
 263
 264 static RSFD r_open (RSET ct, int flag)
 265 {
 266     struct rset_multior_rfd *rfd;
 267     struct rset_multior_info *info = (struct rset_multior_info *) ct->priv;
 268     int i;
 269
 270     if (flag & RSETF_WRITE)
 271     {
 272         logf (LOG_FATAL, "multior set type is read-only");
 273         return NULL;
 274     }
 275     rfd = (struct rset_multior_rfd *) xmalloc (sizeof(*rfd));
 276     rfd->flag = flag;
 277     rfd->next = info->rfd_list;
 278     rfd->info = info;
 279     info->rfd_list = rfd;
 280     rfd->h = heap_create( info->no_rsets, info->key_size, info->cmp);
 281     rfd->prevvalue=0;
 282     rfd->hits=0;
 283     rfd->items=(struct heap_item *) xmalloc(info->no_rsets*sizeof(*rfd->items));
 284     for (i=0; i<info->no_rsets; i++){
 285        rfd->items[i].rset=info->rsets[i];
 286        rfd->items[i].buf=xmalloc(info->key_size);
 287        rfd->items[i].fd=rset_open(info->rsets[i],RSETF_READ);
 288 /*       if (item_readbuf(&(rfd->items[i]))) */
 289        if ( rset_read(rfd->items[i].rset, rfd->items[i].fd,
 290                       rfd->items[i].buf) )
 291            heap_insert(rfd->h, &(rfd->items[i]));
 292     }
 293     return rfd;
 294 }
 295
 296 static void r_close (RSFD rfd)
 297 {
 298     struct rset_multior_rfd *mrfd = (struct rset_multior_rfd *) rfd;
 299     struct rset_multior_info *info = mrfd->info;
 300     struct rset_multior_rfd **rfdp;
 301     int i;
 302
 303     for (rfdp = &info->rfd_list; *rfdp; rfdp = &(*rfdp)->next)
 304         if (*rfdp == rfd)
 305         {
 306             *rfdp = (*rfdp)->next;
 307
 308             heap_destroy (mrfd->h);
 309             for (i = 0; i<info->no_rsets; i++) {
 310                 if (mrfd->items[i].fd)
 311                     rset_close(info->rsets[i],mrfd->items[i].fd);
 312                 xfree(mrfd->items[i].buf);
 313             }
 314             xfree(mrfd->items);
 315             if (mrfd->prevvalue)
 316                 xfree(mrfd->prevvalue);
 317             xfree(mrfd);
 318             return;
 319         }
 320     logf (LOG_FATAL, "r_close but no rfd match!");
 321     assert (0);
 322 }
 323
 324
 325 static void r_rewind (RSFD rfd)
 326 {
 327     assert(!"rewind not implemented yet");
 328 }
 329
 330
 331 static int r_forward(RSET ct, RSFD rfd, void *buf,
 332                      int (*cmpfunc)(const void *p1, const void *p2),
 333                      const void *untilbuf)
 334 {
 335     struct rset_multior_rfd *mrfd = (struct rset_multior_rfd *) rfd;
 336     struct rset_multior_info *info = mrfd->info;
 337     struct heap_item it;
 338     int rdres;
 339     if (heap_empty(mrfd->h))
 340         return 0;
 341     if (cmpfunc)
 342         assert(cmpfunc==mrfd->info->cmp);
 343     it = *(mrfd->h->heap[1]);
 344     memcpy(buf,it.buf, info->key_size);
 345     (mrfd->hits)++;
 346     if (untilbuf)
 347         rdres=rset_forward(it.rset, it.fd, it.buf, cmpfunc,untilbuf);
 348     else
 349         rdres=rset_read(it.rset, it.fd, it.buf);
 350     if ( rdres )
 351         heap_balance(mrfd->h);
 352     else
 353         heap_delete(mrfd->h);
 354     return 1;
 355
 356 }
 357
 358 static int r_read (RSFD rfd, void *buf)
 359 {
 360     return r_forward(0,rfd, buf,0,0);
 361 }
 362
 363 static void r_pos (RSFD rfd, double *current, double *total)
 364 {
 365     struct rset_multior_rfd *mrfd = (struct rset_multior_rfd *) rfd;
 366     struct rset_multior_info *info = mrfd->info;
 367     double cur, tot;
 368     double scur=0.0, stot=0.0;
 369     int i;
 370     for (i=0; i<info->no_rsets; i++){
 371         rset_pos(mrfd->items[i].rset, mrfd->items[i].fd, &cur, &tot);
 372         logf(LOG_LOG, "r_pos: %d %0.1f %0.1f", i, cur,tot);
 373         scur += cur;
 374         stot += tot;
 375     }
 376     if (stot <1.0) { /* nothing there */
 377         *current=0;
 378         *total=0;
 379         return;
 380     }
 381     *current=mrfd->hits;
 382     *total=*current*stot/scur;
 383 }
 384
 385 static int r_write (RSFD rfd, const void *buf)
 386 {
 387     logf (LOG_FATAL, "multior set type is read-only");
 388     return -1;
 389 }