Here it is, the new thread interface.
[citadel.git] / citadel / modules / fulltext / serv_fulltext.c
1 /*
2  * $Id$
3  *
4  * This module handles fulltext indexing of the message base.
5  *
6  */
7
8
9 #include "sysdep.h"
10 #include <stdlib.h>
11 #include <unistd.h>
12 #include <stdio.h>
13 #include <fcntl.h>
14 #include <signal.h>
15 #include <pwd.h>
16 #include <errno.h>
17 #include <sys/types.h>
18
19 #if TIME_WITH_SYS_TIME
20 # include <sys/time.h>
21 # include <time.h>
22 #else
23 # if HAVE_SYS_TIME_H
24 #  include <sys/time.h>
25 # else
26 #  include <time.h>
27 # endif
28 #endif
29
30 #include <sys/wait.h>
31 #include <string.h>
32 #include <limits.h>
33 #include <libcitadel.h>
34 #include "citadel.h"
35 #include "server.h"
36 #include "citserver.h"
37 #include "support.h"
38 #include "config.h"
39 #include "database.h"
40 #include "msgbase.h"
41 #include "control.h"
42 #include "room_ops.h"
43 #include "serv_fulltext.h"
44 #include "ft_wordbreaker.h"
45
46
47 #include "ctdl_module.h"
48
49
50
51 long ft_newhighest = 0L;
52 long *ft_newmsgs = NULL;
53 int ft_num_msgs = 0;
54 int ft_num_alloc = 0;
55
56
57 int ftc_num_msgs[65536];
58 long *ftc_msgs[65536];
59
60
61 /*
62  * Compare function
63  */
64 int longcmp(const void *rec1, const void *rec2) {
65         long i1, i2;
66
67         i1 = *(const long *)rec1;
68         i2 = *(const long *)rec2;
69
70         if (i1 > i2) return(1);
71         if (i1 < i2) return(-1);
72         return(0);
73 }
74
75 /*
76  * Flush our index cache out to disk.
77  */
78 void ft_flush_cache(void) {
79         int i;
80         time_t last_update = 0;
81
82         for (i=0; i<65536; ++i) {
83                 if ((time(NULL) - last_update) >= 10) {
84                         lprintf(CTDL_INFO,
85                                 "Flushing index cache to disk (%d%% complete)\n",
86                                 (i * 100 / 65536)
87                         );
88                         last_update = time(NULL);
89                 }
90                 if (ftc_msgs[i] != NULL) {
91                         cdb_store(CDB_FULLTEXT, &i, sizeof(int), ftc_msgs[i],
92                                 (ftc_num_msgs[i] * sizeof(long)));
93                         ftc_num_msgs[i] = 0;
94                         free(ftc_msgs[i]);
95                         ftc_msgs[i] = NULL;
96                 }
97         }
98         lprintf(CTDL_INFO, "Flushed index cache to disk (100%% complete)\n");
99 }
100
101
102 /*
103  * Index or de-index a message.  (op == 1 to index, 0 to de-index)
104  */
105 void ft_index_message(long msgnum, int op) {
106         int num_tokens = 0;
107         int *tokens = NULL;
108         int i, j;
109         struct cdbdata *cdb_bucket;
110         char *msgtext;
111         int tok;
112
113         lprintf(CTDL_DEBUG, "ft_index_message() %s msg %ld\n",
114                 (op ? "adding" : "removing") , msgnum
115         );
116
117         /* Output the message as text before indexing it, so we don't end up
118          * indexing a bunch of encoded base64, etc.
119          */
120         CC->redirect_buffer = malloc(SIZ);
121         CC->redirect_len = 0;
122         CC->redirect_alloc = SIZ;
123         CtdlOutputMsg(msgnum, MT_CITADEL, HEADERS_ALL, 0, 1, NULL);
124         msgtext = CC->redirect_buffer;
125         CC->redirect_buffer = NULL;
126         CC->redirect_len = 0;
127         CC->redirect_alloc = 0;
128         lprintf(CTDL_DEBUG, "Wordbreaking message %ld...\n", msgnum);
129         wordbreaker(msgtext, &num_tokens, &tokens);
130         free(msgtext);
131
132         lprintf(CTDL_DEBUG, "Indexing message %ld [%d tokens]\n", msgnum, num_tokens);
133         if (num_tokens > 0) {
134                 for (i=0; i<num_tokens; ++i) {
135
136                         /* Add the message to the relevant token bucket */
137
138                         /* search for tokens[i] */
139                         tok = tokens[i];
140
141                         if ( (tok >= 0) && (tok <= 65535) ) {
142                                 /* fetch the bucket, Liza */
143                                 if (ftc_msgs[tok] == NULL) {
144                                         cdb_bucket = cdb_fetch(CDB_FULLTEXT, &tok, sizeof(int));
145                                         if (cdb_bucket != NULL) {
146                                                 ftc_num_msgs[tok] = cdb_bucket->len / sizeof(long);
147                                                 ftc_msgs[tok] = (long *)cdb_bucket->ptr;
148                                                 cdb_bucket->ptr = NULL;
149                                                 cdb_free(cdb_bucket);
150                                         }
151                                         else {
152                                                 ftc_num_msgs[tok] = 0;
153                                                 ftc_msgs[tok] = malloc(sizeof(long));
154                                         }
155                                 }
156         
157         
158                                 if (op == 1) {  /* add to index */
159                                         ++ftc_num_msgs[tok];
160                                         ftc_msgs[tok] = realloc(ftc_msgs[tok],
161                                                                 ftc_num_msgs[tok]*sizeof(long));
162                                         ftc_msgs[tok][ftc_num_msgs[tok] - 1] = msgnum;
163                                 }
164         
165                                 if (op == 0) {  /* remove from index */
166                                         if (ftc_num_msgs[tok] >= 1) {
167                                                 for (j=0; j<ftc_num_msgs[tok]; ++j) {
168                                                         if (ftc_msgs[tok][j] == msgnum) {
169                                                                 memmove(&ftc_msgs[tok][j], &ftc_msgs[tok][j+1], ((ftc_num_msgs[tok] - j - 1)*sizeof(long)));
170                                                                 --ftc_num_msgs[tok];
171                                                                 --j;
172                                                         }
173                                                 }
174                                         }
175                                 }
176                         }
177                         else {
178                                 lprintf(CTDL_ALERT, "Invalid token %d !!\n", tok);
179                         }
180                 }
181
182                 free(tokens);
183         }
184 }
185
186
187
188 /*
189  * Add a message to the list of those to be indexed.
190  */
191 void ft_index_msg(long msgnum, void *userdata) {
192
193         if ((msgnum > CitControl.MMfulltext) && (msgnum <= ft_newhighest)) {
194                 ++ft_num_msgs;
195                 if (ft_num_msgs > ft_num_alloc) {
196                         ft_num_alloc += 1024;
197                         ft_newmsgs = realloc(ft_newmsgs,
198                                 (ft_num_alloc * sizeof(long)));
199                 }
200                 ft_newmsgs[ft_num_msgs - 1] = msgnum;
201         }
202
203 }
204
205 /*
206  * Scan a room for messages to index.
207  */
208 void ft_index_room(struct ctdlroom *qrbuf, void *data)
209 {
210         getroom(&CC->room, qrbuf->QRname);
211         CtdlForEachMessage(MSGS_ALL, 0L, NULL, NULL, NULL, ft_index_msg, NULL);
212 }
213
214
215 /*
216  * Begin the fulltext indexing process.  (Called as an EVT_TIMER event)
217  */
218 void do_fulltext_indexing(void) {
219         int i;
220         static time_t last_index = 0L;
221         static time_t last_progress = 0L;
222         time_t run_time = 0L;
223         time_t end_time = 0L;
224         
225         /*
226          * Don't do this if the site doesn't have it enabled.
227          */
228         if (!config.c_enable_fulltext) {
229                 return;
230         }
231
232         /*
233          * Make sure we don't run the indexer too frequently.
234          * FIXME move the setting into config
235          */
236 /*
237  * The thread sleeps for 300 seconds so we don't need this here any more
238  
239         if ( (time(NULL) - last_index) < 300L) {
240                 return;
241         }
242 */
243
244         /*
245          * Check to see whether the fulltext index is up to date; if there
246          * are no messages to index, don't waste any more time trying.
247          */
248         if (CitControl.MMfulltext >= CitControl.MMhighest) {
249                 return;         /* nothing to do! */
250         }
251         
252         run_time = time(NULL);
253         lprintf(CTDL_DEBUG, "do_fulltext_indexing() started (%ld)\n", run_time);
254         
255         /*
256          * If we've switched wordbreaker modules, burn the index and start
257          * over.
258          */
259         begin_critical_section(S_CONTROL);
260         if (CitControl.fulltext_wordbreaker != FT_WORDBREAKER_ID) {
261                 lprintf(CTDL_DEBUG, "wb ver on disk = %d, code ver = %d\n",
262                         CitControl.fulltext_wordbreaker, FT_WORDBREAKER_ID);
263                 lprintf(CTDL_INFO, "(re)initializing full text index\n");
264                 cdb_trunc(CDB_FULLTEXT);
265                 CitControl.MMfulltext = 0L;
266                 put_control();
267         }
268         end_critical_section(S_CONTROL);
269
270         /*
271          * Now go through each room and find messages to index.
272          */
273         ft_newhighest = CitControl.MMhighest;
274         ForEachRoom(ft_index_room, NULL);       /* load all msg pointers */
275
276         if (ft_num_msgs > 0) {
277                 qsort(ft_newmsgs, ft_num_msgs, sizeof(long), longcmp);
278                 for (i=0; i<(ft_num_msgs-1); ++i) { /* purge dups */
279                         if (ft_newmsgs[i] == ft_newmsgs[i+1]) {
280                                 memmove(&ft_newmsgs[i], &ft_newmsgs[i+1],
281                                         ((ft_num_msgs - i - 1)*sizeof(long)));
282                                 --ft_num_msgs;
283                                 --i;
284                         }
285                 }
286
287                 /* Here it is ... do each message! */
288                 for (i=0; i<ft_num_msgs; ++i) {
289                         if (time(NULL) != last_progress) {
290                                 lprintf(CTDL_DEBUG,
291                                         "Indexed %d of %d messages (%d%%)\n",
292                                                 i, ft_num_msgs,
293                                                 ((i*100) / ft_num_msgs)
294                                 );
295                                 last_progress = time(NULL);
296                         }
297                         ft_index_message(ft_newmsgs[i], 1);
298
299                         /* Check to see if we need to quit early */
300                         if (CtdlThreadCheckStop()) {
301                                 lprintf(CTDL_DEBUG, "Indexer quitting early\n");
302                                 ft_newhighest = ft_newmsgs[i];
303                                 break;
304                         }
305
306                         /* Check to see if we have to maybe flush to disk */
307                         if (i >= FT_MAX_CACHE) {
308                                 lprintf(CTDL_DEBUG, "Time to flush.\n");
309                                 ft_newhighest = ft_newmsgs[i];
310                                 break;
311                         }
312
313                 }
314
315                 free(ft_newmsgs);
316                 ft_num_msgs = 0;
317                 ft_num_alloc = 0;
318                 ft_newmsgs = NULL;
319         }
320         end_time = time(NULL);
321         lprintf(CTDL_DEBUG, "do_fulltext_indexing() duration (%ld)\n", end_time - run_time);
322
323         /* Save our place so we don't have to do this again */
324         ft_flush_cache();
325         begin_critical_section(S_CONTROL);
326         CitControl.MMfulltext = ft_newhighest;
327         CitControl.fulltext_wordbreaker = FT_WORDBREAKER_ID;
328         put_control();
329         end_critical_section(S_CONTROL);
330         last_index = time(NULL);
331
332         lprintf(CTDL_DEBUG, "do_fulltext_indexing() finished\n");
333         return;
334 }
335
336 /*
337  * Main loop for the indexer thread.
338  */
339 void *indexer_thread(void *arg) {
340         struct CitContext indexerCC;
341
342         lprintf(CTDL_DEBUG, "indexer_thread() initializing\n");
343
344         memset(&indexerCC, 0, sizeof(struct CitContext));
345         indexerCC.internal_pgm = 1;
346         indexerCC.cs_pid = 0;
347         pthread_setspecific(MyConKey, (void *)&indexerCC );
348
349         cdb_allocate_tsd();
350
351         while (!CtdlThreadCheckStop()) {
352                 do_fulltext_indexing();
353                 CtdlThreadSleep(300);
354         }
355
356         lprintf(CTDL_DEBUG, "indexer_thread() exiting\n");
357         return NULL;
358 }
359
360
361
362 /*
363  * API call to perform searches.
364  * (This one does the "all of these words" search.)
365  * Caller is responsible for freeing the message list.
366  */
367 void ft_search(int *fts_num_msgs, long **fts_msgs, char *search_string) {
368         int num_tokens = 0;
369         int *tokens = NULL;
370         int i, j;
371         struct cdbdata *cdb_bucket;
372         int num_all_msgs = 0;
373         long *all_msgs = NULL;
374         int num_ret_msgs = 0;
375         int num_ret_alloc = 0;
376         long *ret_msgs = NULL;
377         int tok;
378
379         wordbreaker(search_string, &num_tokens, &tokens);
380         if (num_tokens > 0) {
381                 for (i=0; i<num_tokens; ++i) {
382
383                         /* search for tokens[i] */
384                         tok = tokens[i];
385
386                         /* fetch the bucket, Liza */
387                         if (ftc_msgs[tok] == NULL) {
388                                 cdb_bucket = cdb_fetch(CDB_FULLTEXT, &tok, sizeof(int));
389                                 if (cdb_bucket != NULL) {
390                                         ftc_num_msgs[tok] = cdb_bucket->len / sizeof(long);
391                                         ftc_msgs[tok] = (long *)cdb_bucket->ptr;
392                                         cdb_bucket->ptr = NULL;
393                                         cdb_free(cdb_bucket);
394                                 }
395                                 else {
396                                         ftc_num_msgs[tok] = 0;
397                                         ftc_msgs[tok] = malloc(sizeof(long));
398                                 }
399                         }
400
401                         num_all_msgs += ftc_num_msgs[tok];
402                         if (num_all_msgs > 0) {
403                                 all_msgs = realloc(all_msgs, num_all_msgs*sizeof(long) );
404                                 memcpy(&all_msgs[num_all_msgs-ftc_num_msgs[tok]],
405                                         ftc_msgs[tok], ftc_num_msgs[tok]*sizeof(long) );
406                         }
407
408                 }
409                 free(tokens);
410                 qsort(all_msgs, num_all_msgs, sizeof(long), longcmp);
411
412                 /*
413                  * At this point, if a message appears num_tokens times in the
414                  * list, then it contains all of the search tokens.
415                  */
416                 if (num_all_msgs >= num_tokens)
417                    for (j=0; j<(num_all_msgs-num_tokens+1); ++j) {
418                         if (all_msgs[j] == all_msgs[j+num_tokens-1]) {
419
420                                 ++num_ret_msgs;
421                                 if (num_ret_msgs > num_ret_alloc) {
422                                         num_ret_alloc += 64;
423                                         ret_msgs = realloc(ret_msgs,
424                                                 (num_ret_alloc*sizeof(long)) );
425                                 }
426                                 ret_msgs[num_ret_msgs - 1] = all_msgs[j];
427
428                         }
429                 }
430
431                 free(all_msgs);
432         }
433
434         *fts_num_msgs = num_ret_msgs;
435         *fts_msgs = ret_msgs;
436 }
437
438
439 /*
440  * This search command is for diagnostic purposes and may be removed or replaced.
441  */
442 void cmd_srch(char *argbuf) {
443         int num_msgs = 0;
444         long *msgs = NULL;
445         int i;
446         char search_string[256];
447
448         if (CtdlAccessCheck(ac_logged_in)) return;
449
450         if (!config.c_enable_fulltext) {
451                 cprintf("%d Full text index is not enabled on this server.\n",
452                         ERROR + CMD_NOT_SUPPORTED);
453                 return;
454         }
455
456         extract_token(search_string, argbuf, 0, '|', sizeof search_string);
457         ft_search(&num_msgs, &msgs, search_string);
458
459         cprintf("%d %d msgs match all search words:\n",
460                 LISTING_FOLLOWS, num_msgs);
461         if (num_msgs > 0) {
462                 for (i=0; i<num_msgs; ++i) {
463                         cprintf("%ld\n", msgs[i]);
464                 }
465         }
466         if (msgs != NULL) free(msgs);
467         cprintf("000\n");
468 }
469
470 /*
471  * Zero out our index cache.
472  */
473 void initialize_ft_cache(void) {
474         memset(ftc_num_msgs, 0, (65536 * sizeof(int)));
475         memset(ftc_msgs, 0, (65536 * sizeof(long *)));
476 }
477
478
479 void ft_delete_remove(char *room, long msgnum)
480 {
481         if (room) return;
482         
483         /* Remove from fulltext index */
484         if (config.c_enable_fulltext) {
485                 ft_index_message(msgnum, 0);
486         }
487 }
488
489 /*****************************************************************************/
490
491 CTDL_MODULE_INIT(fulltext)
492 {
493         if (!threading)
494         {
495                 initialize_ft_cache();
496                 CtdlRegisterProtoHook(cmd_srch, "SRCH", "Full text search");
497                 CtdlRegisterDeleteHook(ft_delete_remove);
498                 CtdlRegisterSearchFuncHook(ft_search, "fulltext");
499         }
500         else
501         {
502                 CtdlThreadCreate("indexer", CTDLTHREAD_BIGSTACK, indexer_thread, NULL);
503         }
504         /* return our Subversion id for the Log */
505         return "$Id$";
506 }