index/key_block.c

   1 /* This file is part of the Zebra server.
   2    Copyright (C) 1994-2011 Index Data
   3
   4 Zebra is free software; you can redistribute it and/or modify it under
   5 the terms of the GNU General Public License as published by the Free
   6 Software Foundation; either version 2, or (at your option) any later
   7 version.
   8
   9 Zebra is distributed in the hope that it will be useful, but WITHOUT ANY
  10 WARRANTY; without even the implied warranty of MERCHANTABILITY or
  11 FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  12 for more details.
  13
  14 You should have received a copy of the GNU General Public License
  15 along with this program; if not, write to the Free Software
  16 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  17
  18 */
  19
  20 #if HAVE_CONFIG_H
  21 #include <config.h>
  22 #endif
  23 #include <stdio.h>
  24 #include <stdlib.h>
  25 #include <string.h>
  26 #include <assert.h>
  27 #include <ctype.h>
  28
  29 #if YAZ_POSIX_THREADS
  30 #include <pthread.h>
  31 #endif
  32
  33 #include "key_block.h"
  34 #include <yaz/nmem.h>
  35 #include <yaz/xmalloc.h>
  36
  37 struct zebra_key_block {
  38     char **key_buf;
  39     size_t ptr_top;
  40     size_t ptr_i;
  41     size_t key_buf_used;
  42     int key_file_no;
  43     char *key_tmp_dir;
  44     int use_threads;
  45     char **alt_buf;
  46 #if YAZ_POSIX_THREADS
  47     char **thread_key_buf;
  48     size_t thread_ptr_top;
  49     size_t thread_ptr_i;
  50     int exit_flag;
  51     pthread_t thread_id;
  52     pthread_mutex_t mutex;
  53
  54     pthread_cond_t work_available;
  55
  56     pthread_cond_t cond_sorting;
  57     int is_sorting;
  58 #endif
  59 };
  60
  61 #define ENCODE_BUFLEN 768
  62 struct encode_info {
  63     void *encode_handle;
  64     void *decode_handle;
  65     char buf[ENCODE_BUFLEN];
  66 };
  67
  68 #define USE_SHELLSORT 0
  69
  70 #if USE_SHELLSORT
  71 static void shellsort(void *ar, int r, size_t s,
  72                       int (*cmp)(const void *a, const void *b))
  73 {
  74     char *a = ar;
  75     char v[100];
  76     int h, i, j, k;
  77     static const int incs[16] = { 1391376, 463792, 198768, 86961, 33936,
  78                                   13776, 4592, 1968, 861, 336,
  79                                   112, 48, 21, 7, 3, 1 };
  80     for ( k = 0; k < 16; k++)
  81         for (h = incs[k], i = h; i < r; i++)
  82         {
  83             memcpy (v, a+s*i, s);
  84             j = i;
  85             while (j > h && (*cmp)(a + s*(j-h), v) > 0)
  86             {
  87                 memcpy (a + s*j, a + s*(j-h), s);
  88                 j -= h;
  89             }
  90             memcpy (a+s*j, v, s);
  91         }
  92 }
  93 #endif
  94
  95
  96 static void encode_key_init(struct encode_info *i)
  97 {
  98     i->encode_handle = iscz1_start();
  99     i->decode_handle = iscz1_start();
 100 }
 101
 102 static void encode_key_write(const char *k, struct encode_info *i, FILE *outf)
 103 {
 104     struct it_key key;
 105     char *bp = i->buf, *bp0;
 106     const char *src = (char *) &key;
 107     size_t klen = strlen(k);
 108
 109     if (fwrite (k, klen+1, 1, outf) != 1)
 110     {
 111         yaz_log (YLOG_FATAL|YLOG_ERRNO, "fwrite");
 112         zebra_exit("encode_key_write");
 113     }
 114
 115     k = k + klen+1;
 116
 117     /* and copy & align key so we can mangle */
 118     memcpy (&key, k+1, sizeof(struct it_key));  /* *k is insert/delete */
 119
 120 #if 0
 121     /* debugging */
 122     key_logdump_txt(YLOG_LOG, &key, *k ? "i" : "d");
 123 #endif
 124     assert(key.mem[0] >= 0);
 125
 126     bp0 = bp++;
 127     iscz1_encode(i->encode_handle, &bp, &src);
 128
 129     *bp0 = (*k * 128) + bp - bp0 - 1; /* length and insert/delete combined */
 130     if (fwrite (i->buf, bp - i->buf, 1, outf) != 1)
 131     {
 132         yaz_log (YLOG_FATAL|YLOG_ERRNO, "fwrite");
 133         zebra_exit("encode_key_write");
 134     }
 135
 136 #if 0
 137     /* debugging */
 138     if (1)
 139     {
 140         struct it_key key2;
 141         const char *src = bp0+1;
 142         char *dst = (char*) &key2;
 143         iscz1_decode(i->decode_handle, &dst, &src);
 144
 145         key_logdump_txt(YLOG_LOG, &key2, *k ? "i" : "d");
 146
 147         assert(key2.mem[1]);
 148     }
 149 #endif
 150 }
 151
 152 static void encode_key_flush (struct encode_info *i, FILE *outf)
 153 {
 154     iscz1_stop(i->encode_handle);
 155     iscz1_stop(i->decode_handle);
 156 }
 157
 158 void key_block_flush_int(zebra_key_block_t p,
 159                          char **key_buf, size_t ptr_top, size_t ptr_i);
 160
 161 #if YAZ_POSIX_THREADS
 162 static void *thread_func(void *vp)
 163 {
 164     zebra_key_block_t p = (zebra_key_block_t) vp;
 165     while (1)
 166     {
 167         pthread_mutex_lock(&p->mutex);
 168
 169         while (!p->is_sorting && !p->exit_flag)
 170             pthread_cond_wait(&p->work_available, &p->mutex);
 171
 172         if (p->exit_flag)
 173             break;
 174
 175         pthread_mutex_unlock(&p->mutex);
 176
 177         key_block_flush_int(p, p->thread_key_buf,
 178                             p->thread_ptr_top, p->thread_ptr_i);
 179
 180         pthread_mutex_lock(&p->mutex);
 181         p->is_sorting = 0;
 182         pthread_cond_signal(&p->cond_sorting);
 183         pthread_mutex_unlock(&p->mutex);
 184     }
 185     pthread_mutex_unlock(&p->mutex);
 186     return 0;
 187 }
 188 #endif
 189
 190 zebra_key_block_t key_block_create(int mem, const char *key_tmp_dir,
 191                                    int use_threads)
 192 {
 193     zebra_key_block_t p = xmalloc(sizeof(*p));
 194
 195 #if YAZ_POSIX_THREADS
 196     /* we'll be making two memory areas so cut in half */
 197     if (use_threads)
 198         mem = mem / 2;
 199 #endif
 200     p->key_buf = (char**) xmalloc (mem);
 201     p->ptr_top = mem/sizeof(char*);
 202     p->ptr_i = 0;
 203     p->key_buf_used = 0;
 204     p->key_tmp_dir = xstrdup(key_tmp_dir);
 205     p->key_file_no = 0;
 206     p->alt_buf = 0;
 207     p->use_threads = 0;
 208     if (use_threads)
 209     {
 210 #if YAZ_POSIX_THREADS
 211         p->use_threads = use_threads;
 212         p->is_sorting = 0;
 213         p->exit_flag = 0;
 214         pthread_mutex_init(&p->mutex, 0);
 215         pthread_cond_init(&p->work_available, 0);
 216         pthread_cond_init(&p->cond_sorting, 0);
 217         pthread_create(&p->thread_id, 0, thread_func, p);
 218         p->alt_buf = (char**) xmalloc (mem);
 219 #endif
 220     }
 221     yaz_log(YLOG_DEBUG, "key_block_create t=%d", p->use_threads);
 222     return p;
 223 }
 224
 225 void key_block_destroy(zebra_key_block_t *pp)
 226 {
 227     zebra_key_block_t p = *pp;
 228     if (p)
 229     {
 230         if (p->use_threads)
 231         {
 232 #if YAZ_POSIX_THREADS
 233             pthread_mutex_lock(&p->mutex);
 234
 235             while (p->is_sorting)
 236                 pthread_cond_wait(&p->cond_sorting, &p->mutex);
 237
 238             p->exit_flag = 1;
 239
 240             pthread_cond_broadcast(&p->work_available);
 241
 242             pthread_mutex_unlock(&p->mutex);
 243             pthread_join(p->thread_id, 0);
 244             pthread_cond_destroy(&p->work_available);
 245             pthread_cond_destroy(&p->cond_sorting);
 246             pthread_mutex_destroy(&p->mutex);
 247
 248 #endif
 249             xfree(p->alt_buf);
 250         }
 251         xfree(p->key_buf);
 252         xfree(p->key_tmp_dir);
 253         xfree(p);
 254         *pp = 0;
 255     }
 256 }
 257
 258 void key_block_write(zebra_key_block_t p, zint sysno, struct it_key *key_in,
 259                      int cmd, const char *str_buf, size_t str_len,
 260                      zint staticrank, int static_rank_enable)
 261 {
 262     int ch;
 263     int i, j = 0;
 264     struct it_key key_out;
 265
 266     if (p->key_buf_used + 1024 > (p->ptr_top -p->ptr_i)*sizeof(char*))
 267         key_block_flush(p, 0);
 268     ++(p->ptr_i);
 269     assert(p->ptr_i > 0);
 270     (p->key_buf)[p->ptr_top - p->ptr_i] =
 271         (char*)p->key_buf + p->key_buf_used;
 272
 273     /* key_in->mem[0] ord/ch */
 274     /* key_in->mem[1] filter specified record ID */
 275
 276     /* encode the ordinal value (field/use/attribute) .. */
 277     ch = CAST_ZINT_TO_INT(key_in->mem[0]);
 278     p->key_buf_used +=
 279         key_SU_encode(ch, (char*)p->key_buf +
 280                       p->key_buf_used);
 281
 282     /* copy the 0-terminated stuff from str to output */
 283     memcpy((char*)p->key_buf + p->key_buf_used, str_buf, str_len);
 284     p->key_buf_used += str_len;
 285     ((char*)p->key_buf)[(p->key_buf_used)++] = '\0';
 286
 287     /* the delete/insert indicator */
 288     ((char*)p->key_buf)[(p->key_buf_used)++] = cmd;
 289
 290     if (static_rank_enable)
 291     {
 292         assert(staticrank >= 0);
 293         key_out.mem[j++] = staticrank;
 294     }
 295
 296     if (key_in->mem[1]) /* filter specified record ID */
 297         key_out.mem[j++] = key_in->mem[1];
 298     else
 299         key_out.mem[j++] = sysno;
 300     for (i = 2; i < key_in->len; i++)
 301         key_out.mem[j++] = key_in->mem[i];
 302     key_out.len = j;
 303
 304     memcpy((char*)p->key_buf + p->key_buf_used,
 305            &key_out, sizeof(key_out));
 306     (p->key_buf_used) += sizeof(key_out);
 307 }
 308
 309
 310 void key_block_flush_int(zebra_key_block_t p,
 311                          char **key_buf, size_t ptr_top, size_t ptr_i)
 312 {
 313     FILE *outf;
 314     char out_fname[200];
 315     char *prevcp, *cp;
 316     struct encode_info encode_info;
 317
 318     if (ptr_i == 0)
 319         return ;
 320
 321     (p->key_file_no)++;
 322     yaz_log(YLOG_DEBUG, "sorting section %d", (p->key_file_no));
 323
 324     assert(ptr_i > 0);
 325
 326 #if USE_SHELLSORT
 327     shellsort(key_buf + ptr_top - ptr_i, ptr_i,
 328               sizeof(char*), key_qsort_compare);
 329 #else
 330     qsort(key_buf + ptr_top - ptr_i, ptr_i,
 331           sizeof(char*), key_qsort_compare);
 332 #endif
 333     sprintf(out_fname, "%s/key%d.tmp", p->key_tmp_dir, p->key_file_no);
 334
 335     if (!(outf = fopen (out_fname, "wb")))
 336     {
 337         yaz_log (YLOG_FATAL|YLOG_ERRNO, "fopen %s", out_fname);
 338         zebra_exit("key_block_flush");
 339     }
 340     yaz_log(YLOG_DEBUG, "writing section %d", p->key_file_no);
 341     prevcp = cp = (key_buf)[ptr_top - ptr_i];
 342
 343     encode_key_init (&encode_info);
 344     encode_key_write (cp, &encode_info, outf);
 345
 346     while (--ptr_i > 0)
 347     {
 348         cp = (key_buf)[ptr_top - ptr_i];
 349         if (strcmp (cp, prevcp))
 350         {
 351             encode_key_flush ( &encode_info, outf);
 352             encode_key_init (&encode_info);
 353             encode_key_write (cp, &encode_info, outf);
 354             prevcp = cp;
 355         }
 356         else
 357             encode_key_write (cp + strlen(cp), &encode_info, outf);
 358     }
 359     encode_key_flush ( &encode_info, outf);
 360     if (fclose (outf))
 361     {
 362         yaz_log (YLOG_FATAL|YLOG_ERRNO, "fclose %s", out_fname);
 363         zebra_exit("key_block_flush");
 364     }
 365     yaz_log(YLOG_DEBUG, "finished section %d", p->key_file_no);
 366 }
 367
 368 void key_block_flush(zebra_key_block_t p, int is_final)
 369 {
 370     if (!p)
 371         return;
 372
 373     if (p->use_threads)
 374     {
 375 #if YAZ_POSIX_THREADS
 376         char **tmp;
 377
 378         pthread_mutex_lock(&p->mutex);
 379
 380         while (p->is_sorting)
 381             pthread_cond_wait(&p->cond_sorting, &p->mutex);
 382
 383         p->is_sorting = 1;
 384
 385         p->thread_ptr_top = p->ptr_top;
 386         p->thread_ptr_i = p->ptr_i;
 387         p->thread_key_buf = p->key_buf;
 388
 389         tmp = p->key_buf;
 390         p->key_buf = p->alt_buf;
 391         p->alt_buf = tmp;
 392
 393         pthread_cond_signal(&p->work_available);
 394
 395         if (is_final)
 396         {
 397             while (p->is_sorting)
 398                 pthread_cond_wait(&p->cond_sorting, &p->mutex);
 399         }
 400         pthread_mutex_unlock(&p->mutex);
 401 #endif
 402     }
 403     else
 404         key_block_flush_int(p, p->key_buf, p->ptr_top, p->ptr_i);
 405     p->ptr_i = 0;
 406     p->key_buf_used = 0;
 407 }
 408
 409 int key_block_get_no_files(zebra_key_block_t p)
 410 {
 411     if (p)
 412         return p->key_file_no;
 413     return 0;
 414 }
 415
 416 /*
 417  * Local variables:
 418  * c-basic-offset: 4
 419  * c-file-style: "Stroustrup"
 420  * indent-tabs-mode: nil
 421  * End:
 422  * vim: shiftwidth=4 tabstop=8 expandtab
 423  */
 424