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