EV: expose current event queue state to the RWHO command
[citadel] / citadel / modules / eventclient / serv_eventclient.c
1 /*
2  * Copyright (c) 1998-2012 by the citadel.org team
3  *
4  *  This program is open source software; you can redistribute it and/or modify
5  *  it under the terms of the GNU General Public License version 3.
6  *  
7  *  
8  *
9  *  This program is distributed in the hope that it will be useful,
10  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
11  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12  *  GNU General Public License for more details.
13  *
14  *  
15  *  
16  *  
17  */
18
19 #include "sysdep.h"
20 #include <stdlib.h>
21 #include <unistd.h>
22 #include <stdio.h>
23 #include <termios.h>
24 #include <fcntl.h>
25 #include <signal.h>
26 #include <pwd.h>
27 #include <errno.h>
28 #include <sys/types.h>
29 #include <syslog.h>
30
31 #if TIME_WITH_SYS_TIME
32 # include <sys/time.h>
33 # include <time.h>
34 #else
35 # if HAVE_SYS_TIME_H
36 #  include <sys/time.h>
37 # else
38 #  include <time.h>
39 # endif
40 #endif
41 #include <sys/wait.h>
42 #include <ctype.h>
43 #include <string.h>
44 #include <limits.h>
45 #include <sys/socket.h>
46 #include <netinet/in.h>
47 #include <assert.h>
48 #include <arpa/inet.h>
49 #include <libcitadel.h>
50 #include <curl/curl.h>
51 #include <curl/multi.h>
52 #include "citadel.h"
53 #include "server.h"
54 #include "citserver.h"
55 #include "support.h"
56
57 #include "ctdl_module.h"
58
59 #include "event_client.h"
60 #include "serv_curl.h"
61
62 ev_loop *event_base;
63 int DebugEventLoop = 0;
64 int DebugEventLoopBacktrace = 0;
65 int DebugCurl = 0;
66
67 long EvIDSource = 1;
68 /*****************************************************************************
69  *                   libevent / curl integration                             *
70  *****************************************************************************/
71 #define DBGLOG(LEVEL) if ((LEVEL != LOG_DEBUG) || (DebugCurl != 0))
72
73 #define EVCURL_syslog(LEVEL, FORMAT, ...)                               \
74         DBGLOG (LEVEL) syslog(LEVEL, "EVCURL:IO[%ld]CC[%d] " FORMAT,    \
75                               IO->ID, CCID, __VA_ARGS__)
76
77 #define EVCURLM_syslog(LEVEL, FORMAT)                                   \
78         DBGLOG (LEVEL) syslog(LEVEL, "EVCURL:IO[%ld]CC[%d] " FORMAT,    \
79                               IO->ID, CCID)
80
81 #define CURL_syslog(LEVEL, FORMAT, ...)                                 \
82         DBGLOG (LEVEL) syslog(LEVEL, "CURL: " FORMAT, __VA_ARGS__)
83
84 #define CURLM_syslog(LEVEL, FORMAT)                     \
85         DBGLOG (LEVEL) syslog(LEVEL, "CURL: " FORMAT)
86
87 #define MOPT(s, v)                                                      \
88         do {                                                            \
89                 sta = curl_multi_setopt(mhnd, (CURLMOPT_##s), (v));     \
90                 if (sta) {                                              \
91                         EVQ_syslog(LOG_ERR, "error setting option "     \
92                                #s " on curl multi handle: %s\n",        \
93                                curl_easy_strerror(sta));                \
94                         exit (1);                                       \
95                 }                                                       \
96         } while (0)
97
98
99 typedef struct _evcurl_global_data {
100         int magic;
101         CURLM *mhnd;
102         ev_timer timeev;
103         int nrun;
104 } evcurl_global_data;
105
106 ev_async WakeupCurl;
107 evcurl_global_data global;
108
109 static void
110 gotstatus(int nnrun)
111 {
112         CURLMsg *msg;
113         int nmsg;
114
115         global.nrun = nnrun;
116
117         CURLM_syslog(LOG_DEBUG,
118                      "gotstatus(): about to call curl_multi_info_read\n");
119         while ((msg = curl_multi_info_read(global.mhnd, &nmsg))) {
120                 CURL_syslog(LOG_DEBUG,
121                             "got curl multi_info message msg=%d\n",
122                             msg->msg);
123
124                 if (CURLMSG_DONE == msg->msg) {
125                         CURL *chnd;
126                         char *chandle = NULL;
127                         CURLcode sta;
128                         CURLMcode msta;
129                         AsyncIO*IO;
130
131                         chandle = NULL;;
132                         chnd = msg->easy_handle;
133                         sta = curl_easy_getinfo(chnd,
134                                                 CURLINFO_PRIVATE,
135                                                 &chandle);
136                         if (sta) {
137                                 syslog(LOG_ERR,
138                                        "error asking curl for private"
139                                        " cookie of curl handle: %s\n",
140                                        curl_easy_strerror(sta));
141                                 continue;
142                         }
143                         IO = (AsyncIO *)chandle;
144                         if (IO->ID == 0) {
145                                 EVCURL_syslog(LOG_ERR,
146                                               "Error, invalid IO context %p\n",
147                                               IO);
148                                 continue;
149                         }
150                         SetEVState(IO, eCurlGotStatus);
151
152                         EVCURLM_syslog(LOG_DEBUG, "request complete\n");
153
154                         IO->Now = ev_now(event_base);
155
156                         ev_io_stop(event_base, &IO->recv_event);
157                         ev_io_stop(event_base, &IO->send_event);
158
159                         sta = msg->data.result;
160                         if (sta) {
161                                 EVCURL_syslog(LOG_ERR,
162                                               "error description: %s\n",
163                                               IO->HttpReq.errdesc);
164                                 EVCURL_syslog(LOG_ERR,
165                                               "error performing request: %s\n",
166                                               curl_easy_strerror(sta));
167                         }
168                         sta = curl_easy_getinfo(chnd,
169                                                 CURLINFO_RESPONSE_CODE,
170                                                 &IO->HttpReq.httpcode);
171                         if (sta)
172                                 EVCURL_syslog(LOG_ERR,
173                                               "error asking curl for "
174                                               "response code from request: %s\n",
175                                               curl_easy_strerror(sta));
176                         EVCURL_syslog(LOG_DEBUG,
177                                       "http response code was %ld\n",
178                                       (long)IO->HttpReq.httpcode);
179
180
181                         curl_slist_free_all(IO->HttpReq.headers);
182                         msta = curl_multi_remove_handle(global.mhnd, chnd);
183                         if (msta)
184                                 EVCURL_syslog(LOG_ERR,
185                                               "warning problem detaching "
186                                               "completed handle from curl multi: "
187                                               "%s\n",
188                                               curl_multi_strerror(msta));
189
190                         ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
191
192                         IO->HttpReq.attached = 0;
193                         switch(IO->SendDone(IO))
194                         {
195                         case eDBQuery:
196                                 curl_easy_cleanup(IO->HttpReq.chnd);
197                                 IO->HttpReq.chnd = NULL;
198                                 break;
199                         case eSendDNSQuery:
200                         case eReadDNSReply:
201                         case eConnect:
202                         case eSendReply:
203                         case eSendMore:
204                         case eSendFile:
205                         case eReadMessage:
206                         case eReadMore:
207                         case eReadPayload:
208                         case eReadFile:
209                                 curl_easy_cleanup(IO->HttpReq.chnd);
210                                 IO->HttpReq.chnd = NULL;
211                                 break;
212                         case eTerminateConnection:
213                         case eAbort:
214                                 curl_easy_cleanup(IO->HttpReq.chnd);
215                                 IO->HttpReq.chnd = NULL;
216                                 FreeStrBuf(&IO->HttpReq.ReplyData);
217                                 FreeURL(&IO->ConnectMe);
218                                 RemoveContext(IO->CitContext);
219                                 IO->Terminate(IO);
220                         }
221                 }
222         }
223 }
224
225 static void
226 stepmulti(void *data, curl_socket_t fd, int which)
227 {
228         int running_handles = 0;
229         CURLMcode msta;
230
231         msta = curl_multi_socket_action(global.mhnd,
232                                         fd,
233                                         which,
234                                         &running_handles);
235
236         CURLM_syslog(LOG_DEBUG, "stepmulti(): calling gotstatus()\n");
237         if (msta)
238                 CURL_syslog(LOG_ERR,
239                             "error in curl processing events"
240                             "on multi handle, fd %d: %s\n",
241                             (int)fd,
242                             curl_multi_strerror(msta));
243
244         if (global.nrun != running_handles)
245                 gotstatus(running_handles);
246 }
247
248 static void
249 gottime(struct ev_loop *loop, ev_timer *timeev, int events)
250 {
251         CURLM_syslog(LOG_DEBUG, "EVCURL: waking up curl for timeout\n");
252         stepmulti(NULL, CURL_SOCKET_TIMEOUT, 0);
253 }
254
255 static void
256 got_in(struct ev_loop *loop, ev_io *ioev, int events)
257 {
258         CURL_syslog(LOG_DEBUG,
259                     "EVCURL: waking up curl for io on fd %d\n",
260                     (int)ioev->fd);
261
262         stepmulti(ioev->data, ioev->fd, CURL_CSELECT_IN);
263 }
264
265 static void
266 got_out(struct ev_loop *loop, ev_io *ioev, int events)
267 {
268         CURL_syslog(LOG_DEBUG,
269                     "waking up curl for io on fd %d\n",
270                     (int)ioev->fd);
271
272         stepmulti(ioev->data, ioev->fd, CURL_CSELECT_OUT);
273 }
274
275 static size_t
276 gotdata(void *data, size_t size, size_t nmemb, void *cglobal)
277 {
278         AsyncIO *IO = (AsyncIO*) cglobal;
279
280         SetEVState(IO, eCurlGotData);
281         if (IO->HttpReq.ReplyData == NULL)
282         {
283                 IO->HttpReq.ReplyData = NewStrBufPlain(NULL, SIZ);
284         }
285         IO->Now = ev_now(event_base);
286         return CurlFillStrBuf_callback(data,
287                                        size,
288                                        nmemb,
289                                        IO->HttpReq.ReplyData);
290 }
291
292 static int
293 gotwatchtime(CURLM *multi, long tblock_ms, void *cglobal) {
294         CURL_syslog(LOG_DEBUG, "EVCURL: gotwatchtime called %ld ms\n", tblock_ms);
295         evcurl_global_data *global = cglobal;
296         ev_timer_stop(EV_DEFAULT, &global->timeev);
297         if (tblock_ms < 0 || 14000 < tblock_ms)
298                 tblock_ms = 14000;
299         ev_timer_set(&global->timeev, 0.5e-3 + 1.0e-3 * tblock_ms, 14.0);
300         ev_timer_start(EV_DEFAULT_UC, &global->timeev);
301         curl_multi_perform(global, &global->nrun);
302         return 0;
303 }
304
305 static int
306 gotwatchsock(CURL *easy,
307              curl_socket_t fd,
308              int action,
309              void *cglobal,
310              void *vIO)
311 {
312         evcurl_global_data *global = cglobal;
313         CURLM *mhnd = global->mhnd;
314         char *f;
315         AsyncIO *IO = (AsyncIO*) vIO;
316         CURLcode sta;
317         const char *Action;
318
319         if (IO == NULL) {
320                 sta = curl_easy_getinfo(easy, CURLINFO_PRIVATE, &f);
321                 if (sta) {
322                         CURL_syslog(LOG_ERR,
323                                     "EVCURL: error asking curl for private "
324                                     "cookie of curl handle: %s\n",
325                                     curl_easy_strerror(sta));
326                         return -1;
327                 }
328                 IO = (AsyncIO *) f;
329                 SetEVState(IO, eCurlNewIO);
330                 EVCURL_syslog(LOG_DEBUG,
331                               "EVCURL: got socket for URL: %s\n",
332                               IO->ConnectMe->PlainUrl);
333
334                 if (IO->SendBuf.fd != 0)
335                 {
336                         ev_io_stop(event_base, &IO->recv_event);
337                         ev_io_stop(event_base, &IO->send_event);
338                 }
339                 IO->SendBuf.fd = fd;
340                 ev_io_init(&IO->recv_event, &got_in, fd, EV_READ);
341                 ev_io_init(&IO->send_event, &got_out, fd, EV_WRITE);
342                 curl_multi_assign(mhnd, fd, IO);
343         }
344
345         SetEVState(IO, eCurlGotIO);
346         IO->Now = ev_now(event_base);
347
348         Action = "";
349         switch (action)
350         {
351         case CURL_POLL_NONE:
352                 Action = "CURL_POLL_NONE";
353                 break;
354         case CURL_POLL_REMOVE:
355                 Action = "CURL_POLL_REMOVE";
356                 break;
357         case CURL_POLL_IN:
358                 Action = "CURL_POLL_IN";
359                 break;
360         case CURL_POLL_OUT:
361                 Action = "CURL_POLL_OUT";
362                 break;
363         case CURL_POLL_INOUT:
364                 Action = "CURL_POLL_INOUT";
365                 break;
366         }
367
368
369         EVCURL_syslog(LOG_DEBUG,
370                       "EVCURL: gotwatchsock called fd=%d action=%s[%d]\n",
371                       (int)fd, Action, action);
372
373         switch (action)
374         {
375         case CURL_POLL_NONE:
376                 EVCURLM_syslog(LOG_DEBUG,
377                                "called first time "
378                                "to register this sockwatcker\n");
379                 break;
380         case CURL_POLL_REMOVE:
381                 EVCURLM_syslog(LOG_DEBUG,
382                                "called last time to unregister "
383                                "this sockwatcher\n");
384                 ev_io_stop(event_base, &IO->recv_event);
385                 ev_io_stop(event_base, &IO->send_event);
386                 break;
387         case CURL_POLL_IN:
388                 ev_io_start(event_base, &IO->recv_event);
389                 ev_io_stop(event_base, &IO->send_event);
390                 break;
391         case CURL_POLL_OUT:
392                 ev_io_start(event_base, &IO->send_event);
393                 ev_io_stop(event_base, &IO->recv_event);
394                 break;
395         case CURL_POLL_INOUT:
396                 ev_io_start(event_base, &IO->send_event);
397                 ev_io_start(event_base, &IO->recv_event);
398                 break;
399         }
400         return 0;
401 }
402
403 void curl_init_connectionpool(void)
404 {
405         CURLM *mhnd ;
406
407         ev_timer_init(&global.timeev, &gottime, 14.0, 14.0);
408         global.timeev.data = (void *)&global;
409         global.nrun = -1;
410         CURLcode sta = curl_global_init(CURL_GLOBAL_ALL);
411
412         if (sta)
413         {
414                 CURL_syslog(LOG_ERR,
415                             "error initializing curl library: %s\n",
416                             curl_easy_strerror(sta));
417
418                 exit(1);
419         }
420         mhnd = global.mhnd = curl_multi_init();
421         if (!mhnd)
422         {
423                 CURLM_syslog(LOG_ERR,
424                              "error initializing curl multi handle\n");
425                 exit(3);
426         }
427
428         MOPT(SOCKETFUNCTION, &gotwatchsock);
429         MOPT(SOCKETDATA, (void *)&global);
430         MOPT(TIMERFUNCTION, &gotwatchtime);
431         MOPT(TIMERDATA, (void *)&global);
432
433         return;
434 }
435
436 int evcurl_init(AsyncIO *IO)
437 {
438         CURLcode sta;
439         CURL *chnd;
440
441         EVCURLM_syslog(LOG_DEBUG, "EVCURL: evcurl_init called ms\n");
442         IO->HttpReq.attached = 0;
443         chnd = IO->HttpReq.chnd = curl_easy_init();
444         if (!chnd)
445         {
446                 EVCURLM_syslog(LOG_ERR, "EVCURL: error initializing curl handle\n");
447                 return 0;
448         }
449
450 #if DEBUG
451         OPT(VERBOSE, (long)1);
452 #endif
453         OPT(NOPROGRESS, 1L);
454
455         OPT(NOSIGNAL, 1L);
456         OPT(FAILONERROR, (long)1);
457         OPT(ENCODING, "");
458         OPT(FOLLOWLOCATION, (long)0);
459         OPT(MAXREDIRS, (long)0);
460         OPT(USERAGENT, CITADEL);
461
462         OPT(TIMEOUT, (long)1800);
463         OPT(LOW_SPEED_LIMIT, (long)64);
464         OPT(LOW_SPEED_TIME, (long)600);
465         OPT(CONNECTTIMEOUT, (long)600);
466         OPT(PRIVATE, (void *)IO);
467
468         OPT(FORBID_REUSE, 1);
469         OPT(WRITEFUNCTION, &gotdata);
470         OPT(WRITEDATA, (void *)IO);
471         OPT(ERRORBUFFER, IO->HttpReq.errdesc);
472
473         if ((!IsEmptyStr(config.c_ip_addr))
474                 && (strcmp(config.c_ip_addr, "*"))
475                 && (strcmp(config.c_ip_addr, "::"))
476                 && (strcmp(config.c_ip_addr, "0.0.0.0"))
477                 )
478         {
479                 OPT(INTERFACE, config.c_ip_addr);
480         }
481
482 #ifdef CURLOPT_HTTP_CONTENT_DECODING
483         OPT(HTTP_CONTENT_DECODING, 1);
484         OPT(ENCODING, "");
485 #endif
486
487         IO->HttpReq.headers = curl_slist_append(IO->HttpReq.headers,
488                                                 "Connection: close");
489
490         return 1;
491 }
492
493
494 static void IOcurl_abort_shutdown_callback(struct ev_loop *loop,
495                                            ev_cleanup *watcher,
496                                            int revents)
497 {
498         CURLMcode msta;
499         AsyncIO *IO = watcher->data;
500
501         if (IO == NULL)
502                 return;
503
504         SetEVState(IO, eCurlShutdown);
505         IO->Now = ev_now(event_base);
506         EVCURL_syslog(LOG_DEBUG, "EVENT Curl: %s\n", __FUNCTION__);
507
508         curl_slist_free_all(IO->HttpReq.headers);
509         msta = curl_multi_remove_handle(global.mhnd, IO->HttpReq.chnd);
510         if (msta)
511         {
512                 EVCURL_syslog(LOG_ERR,
513                               "EVCURL: warning problem detaching completed handle "
514                               "from curl multi: %s\n",
515                               curl_multi_strerror(msta));
516         }
517
518         curl_easy_cleanup(IO->HttpReq.chnd);
519         IO->HttpReq.chnd = NULL;
520         ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
521         ev_io_stop(event_base, &IO->recv_event);
522         ev_io_stop(event_base, &IO->send_event);
523         assert(IO->ShutdownAbort);
524         IO->ShutdownAbort(IO);
525 }
526 eNextState
527 evcurl_handle_start(AsyncIO *IO)
528 {
529         CURLMcode msta;
530         CURLcode sta;
531         CURL *chnd;
532
533         SetEVState(IO, eCurlStart);
534         chnd = IO->HttpReq.chnd;
535         EVCURL_syslog(LOG_DEBUG,
536                   "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl);
537         OPT(URL, IO->ConnectMe->PlainUrl);
538         if (StrLength(IO->ConnectMe->CurlCreds))
539         {
540                 OPT(HTTPAUTH, (long)CURLAUTH_BASIC);
541                 OPT(USERPWD, ChrPtr(IO->ConnectMe->CurlCreds));
542         }
543         if (StrLength(IO->HttpReq.PostData) > 0)
544         {
545                 OPT(POSTFIELDS, ChrPtr(IO->HttpReq.PostData));
546                 OPT(POSTFIELDSIZE, StrLength(IO->HttpReq.PostData));
547
548         }
549         else if ((IO->HttpReq.PlainPostDataLen != 0) &&
550                  (IO->HttpReq.PlainPostData != NULL))
551         {
552                 OPT(POSTFIELDS, IO->HttpReq.PlainPostData);
553                 OPT(POSTFIELDSIZE, IO->HttpReq.PlainPostDataLen);
554         }
555         OPT(HTTPHEADER, IO->HttpReq.headers);
556
557         IO->NextState = eConnect;
558         EVCURLM_syslog(LOG_DEBUG, "EVCURL: attaching to curl multi handle\n");
559         msta = curl_multi_add_handle(global.mhnd, IO->HttpReq.chnd);
560         if (msta)
561         {
562                 EVCURL_syslog(LOG_ERR,
563                           "EVCURL: error attaching to curl multi handle: %s\n",
564                           curl_multi_strerror(msta));
565         }
566
567         IO->HttpReq.attached = 1;
568         ev_async_send (event_base, &WakeupCurl);
569
570         ev_cleanup_init(&IO->abort_by_shutdown,
571                         IOcurl_abort_shutdown_callback);
572
573         ev_cleanup_start(event_base, &IO->abort_by_shutdown);
574
575         return eReadMessage;
576 }
577
578 static void WakeupCurlCallback(EV_P_ ev_async *w, int revents)
579 {
580         CURLM_syslog(LOG_DEBUG, "waking up curl multi handle\n");
581
582         curl_multi_perform(&global, CURL_POLL_NONE);
583 }
584
585 static void evcurl_shutdown (void)
586 {
587         curl_global_cleanup();
588         curl_multi_cleanup(global.mhnd);
589         CURLM_syslog(LOG_DEBUG, "exiting\n");
590 }
591 /*****************************************************************************
592  *                       libevent integration                                *
593  *****************************************************************************/
594 /*
595  * client event queue plus its methods.
596  * this currently is the main loop (which may change in some future?)
597  */
598 int evbase_count = 0;
599 pthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */
600 pthread_mutex_t EventExitQueueMutex; /* locks the access to the event queue pointer on exit. */
601 HashList *QueueEvents = NULL;
602 HashList *InboundEventQueue = NULL;
603 HashList *InboundEventQueues[2] = { NULL, NULL };
604 extern void ShutDownCLient(AsyncIO *IO);
605
606 ev_async AddJob;
607 ev_async ExitEventLoop;
608
609 static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
610 {
611         CitContext *Ctx;
612         ev_tstamp Now;
613         HashList *q;
614         void *v;
615         HashPos*It;
616         long len;
617         const char *Key;
618
619         /* get the control command... */
620         pthread_mutex_lock(&EventQueueMutex);
621
622         if (InboundEventQueues[0] == InboundEventQueue) {
623                 InboundEventQueue = InboundEventQueues[1];
624                 q = InboundEventQueues[0];
625         }
626         else {
627                 InboundEventQueue = InboundEventQueues[0];
628                 q = InboundEventQueues[1];
629         }
630         pthread_mutex_unlock(&EventQueueMutex);
631         Now = ev_now (event_base);
632         It = GetNewHashPos(q, 0);
633         while (GetNextHashPos(q, It, &len, &Key, &v))
634         {
635                 IOAddHandler *h = v;
636                 if (h->IO->ID == 0) {
637                         h->IO->ID = EvIDSource++;
638                 }
639                 if (h->IO->StartIO == 0.0)
640                         h->IO->StartIO = Now;
641
642                 SetEVState(h->IO, eIOAttach);
643
644                 Ctx = h->IO->CitContext;
645                 become_session(Ctx);
646
647                 h->IO->Now = Now;
648                 switch (h->EvAttch(h->IO))
649                 {
650                 case eReadMore:
651                 case eReadMessage:
652                 case eReadFile:
653                 case eSendReply:
654                 case eSendMore:
655                 case eReadPayload:
656                 case eSendFile:
657                 case eDBQuery:
658                 case eSendDNSQuery:
659                 case eReadDNSReply:
660                 case eConnect:
661                         break;
662                 case eTerminateConnection:
663                 case eAbort:
664                         ShutDownCLient(h->IO);
665                 break;
666                 }
667         }
668         DeleteHashPos(&It);
669         DeleteHashContent(&q);
670         EVQM_syslog(LOG_DEBUG, "EVENT Q Add done.\n");
671 }
672
673
674 static void EventExitCallback(EV_P_ ev_async *w, int revents)
675 {
676         ev_break(event_base, EVBREAK_ALL);
677
678         EVQM_syslog(LOG_DEBUG, "EVENT Q exiting.\n");
679 }
680
681
682
683 void InitEventQueue(void)
684 {
685         pthread_mutex_init(&EventQueueMutex, NULL);
686         pthread_mutex_init(&EventExitQueueMutex, NULL);
687
688         QueueEvents = NewHash(1, Flathash);
689         InboundEventQueues[0] = NewHash(1, Flathash);
690         InboundEventQueues[1] = NewHash(1, Flathash);
691         InboundEventQueue = InboundEventQueues[0];
692 }
693 extern void CtdlDestroyEVCleanupHooks(void);
694
695 extern int EVQShutDown;
696 /*
697  * this thread operates the select() etc. via libev.
698  */
699 void *client_event_thread(void *arg) 
700 {
701         struct CitContext libev_client_CC;
702
703         CtdlFillSystemContext(&libev_client_CC, "LibEv Thread");
704 //      citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
705         EVQM_syslog(LOG_DEBUG, "client_event_thread() initializing\n");
706
707         event_base = ev_default_loop (EVFLAG_AUTO);
708         ev_async_init(&AddJob, QueueEventAddCallback);
709         ev_async_start(event_base, &AddJob);
710         ev_async_init(&ExitEventLoop, EventExitCallback);
711         ev_async_start(event_base, &ExitEventLoop);
712         ev_async_init(&WakeupCurl, WakeupCurlCallback);
713         ev_async_start(event_base, &WakeupCurl);
714
715         curl_init_connectionpool();
716
717         ev_run (event_base, 0);
718
719         EVQM_syslog(LOG_DEBUG, "client_event_thread() exiting\n");
720
721 ///what todo here?      CtdlClearSystemContext();
722         pthread_mutex_lock(&EventExitQueueMutex);
723         ev_loop_destroy (EV_DEFAULT_UC);
724         event_base = NULL;
725         DeleteHash(&QueueEvents);
726         InboundEventQueue = NULL;
727         DeleteHash(&InboundEventQueues[0]);
728         DeleteHash(&InboundEventQueues[1]);
729 /*      citthread_mutex_destroy(&EventQueueMutex); TODO */
730         evcurl_shutdown();
731
732         CtdlDestroyEVCleanupHooks();
733
734         pthread_mutex_unlock(&EventExitQueueMutex);
735         EVQShutDown = 1;
736         return(NULL);
737 }
738
739 /*----------------------------------------------------------------------------*/
740 /*
741  * DB-Queue; does async bdb operations.
742  * has its own set of handlers.
743  */
744 ev_loop *event_db;
745 int evdb_count = 0;
746 pthread_mutex_t DBEventQueueMutex; /* locks the access to the following vars: */
747 pthread_mutex_t DBEventExitQueueMutex; /* locks the access to the db-event queue pointer on exit. */
748 HashList *DBQueueEvents = NULL;
749 HashList *DBInboundEventQueue = NULL;
750 HashList *DBInboundEventQueues[2] = { NULL, NULL };
751
752 ev_async DBAddJob;
753 ev_async DBExitEventLoop;
754
755 extern void ShutDownDBCLient(AsyncIO *IO);
756
757 static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents)
758 {
759         CitContext *Ctx;
760         ev_tstamp Now;
761         HashList *q;
762         void *v;
763         HashPos *It;
764         long len;
765         const char *Key;
766
767         /* get the control command... */
768         pthread_mutex_lock(&DBEventQueueMutex);
769
770         if (DBInboundEventQueues[0] == DBInboundEventQueue) {
771                 DBInboundEventQueue = DBInboundEventQueues[1];
772                 q = DBInboundEventQueues[0];
773         }
774         else {
775                 DBInboundEventQueue = DBInboundEventQueues[0];
776                 q = DBInboundEventQueues[1];
777         }
778         pthread_mutex_unlock(&DBEventQueueMutex);
779
780         Now = ev_now (event_db);
781         It = GetNewHashPos(q, 0);
782         while (GetNextHashPos(q, It, &len, &Key, &v))
783         {
784                 IOAddHandler *h = v;
785                 eNextState rc;
786                 if (h->IO->ID == 0)
787                         h->IO->ID = EvIDSource++;
788                 if (h->IO->StartDB == 0.0)
789                         h->IO->StartDB = Now;
790                 h->IO->Now = Now;
791
792                 SetEVState(h->IO, eDBAttach);
793                 Ctx = h->IO->CitContext;
794                 become_session(Ctx);
795                 ev_cleanup_start(event_db, &h->IO->db_abort_by_shutdown);
796                 rc = h->EvAttch(h->IO);
797                 switch (rc)
798                 {
799                 case eAbort:
800                         ShutDownDBCLient(h->IO);
801                 default:
802                         break;
803                 }
804         }
805         DeleteHashPos(&It);
806         DeleteHashContent(&q);
807         EVQM_syslog(LOG_DEBUG, "DBEVENT Q Add done.\n");
808 }
809
810
811 static void DBEventExitCallback(EV_P_ ev_async *w, int revents)
812 {
813         EVQM_syslog(LOG_DEBUG, "DB EVENT Q exiting.\n");
814         ev_break(event_db, EVBREAK_ALL);
815 }
816
817
818
819 void DBInitEventQueue(void)
820 {
821         pthread_mutex_init(&DBEventQueueMutex, NULL);
822         pthread_mutex_init(&DBEventExitQueueMutex, NULL);
823
824         DBQueueEvents = NewHash(1, Flathash);
825         DBInboundEventQueues[0] = NewHash(1, Flathash);
826         DBInboundEventQueues[1] = NewHash(1, Flathash);
827         DBInboundEventQueue = DBInboundEventQueues[0];
828 }
829
830 /*
831  * this thread operates writing to the message database via libev.
832  */
833 void *db_event_thread(void *arg)
834 {
835         ev_loop *tmp;
836         struct CitContext libev_msg_CC;
837
838         CtdlFillSystemContext(&libev_msg_CC, "LibEv DB IO Thread");
839
840         EVQM_syslog(LOG_DEBUG, "dbevent_thread() initializing\n");
841
842         tmp = event_db = ev_loop_new (EVFLAG_AUTO);
843
844         ev_async_init(&DBAddJob, DBQueueEventAddCallback);
845         ev_async_start(event_db, &DBAddJob);
846         ev_async_init(&DBExitEventLoop, DBEventExitCallback);
847         ev_async_start(event_db, &DBExitEventLoop);
848
849         ev_run (event_db, 0);
850
851         pthread_mutex_lock(&DBEventExitQueueMutex);
852
853         event_db = NULL;
854         EVQM_syslog(LOG_INFO, "dbevent_thread() exiting\n");
855
856         DeleteHash(&DBQueueEvents);
857         DBInboundEventQueue = NULL;
858         DeleteHash(&DBInboundEventQueues[0]);
859         DeleteHash(&DBInboundEventQueues[1]);
860
861 /*      citthread_mutex_destroy(&DBEventQueueMutex); TODO */
862
863         ev_loop_destroy (tmp);
864         pthread_mutex_unlock(&DBEventExitQueueMutex);
865         return(NULL);
866 }
867
868 void ShutDownEventQueues(void)
869 {
870         EVQM_syslog(LOG_DEBUG, "EVENT Qs triggering exits.\n");
871
872         pthread_mutex_lock(&DBEventQueueMutex);
873         ev_async_send (event_db, &DBExitEventLoop);
874         pthread_mutex_unlock(&DBEventQueueMutex);
875
876         pthread_mutex_lock(&EventQueueMutex);
877         ev_async_send (EV_DEFAULT_ &ExitEventLoop);
878         pthread_mutex_unlock(&EventQueueMutex);
879 }
880
881 void DebugEventloopEnable(const int n)
882 {
883         DebugEventLoop = n;
884 }
885 void DebugEventloopBacktraceEnable(const int n)
886 {
887         DebugEventLoopBacktrace = n;
888 }
889
890 void DebugCurlEnable(const int n)
891 {
892         DebugCurl = n;
893 }
894
895 CTDL_MODULE_INIT(event_client)
896 {
897         if (!threading)
898         {
899                 CtdlRegisterDebugFlagHook(HKEY("eventloop"), DebugEventloopEnable, &DebugEventLoop);
900                 CtdlRegisterDebugFlagHook(HKEY("eventloopbacktrace"), DebugEventloopBacktraceEnable, &DebugEventLoopBacktrace);
901                 CtdlRegisterDebugFlagHook(HKEY("curl"), DebugCurlEnable, &DebugCurl);
902                 InitEventQueue();
903                 DBInitEventQueue();
904                 CtdlThreadCreate(client_event_thread);
905                 CtdlThreadCreate(db_event_thread);
906         }
907         return "event";
908 }