rset/rsmultior.c

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