citadel/modules/eventclient/serv_eventclient.c

   1 /*
   2  * Copyright (c) 1998-2009 by the citadel.org team
   3  *
   4  *  This program is free software; you can redistribute it and/or modify
   5  *  it under the terms of the GNU General Public License as published by
   6  *  the Free Software Foundation; either version 3 of the License, or
   7  *  (at your option) any later version.
   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  *  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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  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
  64 long EvIDSource = 1;
  65 /*****************************************************************************
  66  *                   libevent / curl integration                             *
  67  *****************************************************************************/
  68 #define MOPT(s, v)                                                      \
  69         do {                                                            \
  70                 sta = curl_multi_setopt(mhnd, (CURLMOPT_##s), (v));     \
  71                 if (sta) {                                              \
  72                         syslog(LOG_ERR, "EVCURL: error setting option " #s " on curl multi handle: %s\n", curl_easy_strerror(sta)); \
  73                         exit (1);                                       \
  74                 }                                                       \
  75         } while (0)
  76
  77 typedef struct _evcurl_global_data {
  78         int magic;
  79         CURLM *mhnd;
  80         ev_timer timeev;
  81         int nrun;
  82 } evcurl_global_data;
  83
  84 ev_async WakeupCurl;
  85 evcurl_global_data global;
  86
  87 static void
  88 gotstatus(int nnrun)
  89 {
  90         CURLMsg *msg;
  91         int nmsg;
  92
  93         global.nrun = nnrun;
  94
  95         syslog(LOG_DEBUG, "CURLEV: gotstatus(): about to call curl_multi_info_read\n");
  96         while ((msg = curl_multi_info_read(global.mhnd, &nmsg))) {
  97                 syslog(LOG_ERR, "EVCURL: got curl multi_info message msg=%d\n", msg->msg);
  98                 if (CURLMSG_DONE == msg->msg) {
  99                         CURL *chnd;
 100                         char *chandle;
 101                         CURLcode sta;
 102                         CURLMcode msta;
 103                         AsyncIO  *IO;
 104
 105                         chandle = NULL;;
 106                         chnd = msg->easy_handle;
 107                         sta = curl_easy_getinfo(chnd, CURLINFO_PRIVATE, &chandle);
 108                         syslog(LOG_ERR, "EVCURL: request complete\n");
 109                         if (sta)
 110                                 syslog(LOG_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta));
 111                         IO = (AsyncIO *)chandle;
 112
 113                         ev_io_stop(event_base, &IO->recv_event);
 114                         ev_io_stop(event_base, &IO->send_event);
 115
 116                         sta = msg->data.result;
 117                         if (sta) {
 118                                 EV_syslog(LOG_ERR, "EVCURL: error description: %s\n", IO->HttpReq.errdesc);
 119                                 EV_syslog(LOG_ERR, "EVCURL: error performing request: %s\n", curl_easy_strerror(sta));
 120                         }
 121                         sta = curl_easy_getinfo(chnd, CURLINFO_RESPONSE_CODE, &IO->HttpReq.httpcode);
 122                         if (sta)
 123                                 EV_syslog(LOG_ERR, "EVCURL: error asking curl for response code from request: %s\n", curl_easy_strerror(sta));
 124                         EV_syslog(LOG_ERR, "EVCURL: http response code was %ld\n", (long)IO->HttpReq.httpcode);
 125
 126
 127                         curl_slist_free_all(IO->HttpReq.headers);
 128                         msta = curl_multi_remove_handle(global.mhnd, chnd);
 129                         if (msta)
 130                                 EV_syslog(LOG_ERR, "EVCURL: warning problem detaching completed handle from curl multi: %s\n", curl_multi_strerror(msta));
 131
 132                         ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
 133
 134                         IO->HttpReq.attached = 0;
 135                         switch(IO->SendDone(IO))
 136                         {
 137                         case eDBQuery:
 138                                 curl_easy_cleanup(IO->HttpReq.chnd);
 139                                 IO->HttpReq.chnd = NULL;
 140                                 break;
 141                         case eSendDNSQuery:
 142                         case eReadDNSReply:
 143                         case eConnect:
 144                         case eSendReply:
 145                         case eSendMore:
 146                         case eSendFile:
 147                         case eReadMessage:
 148                         case eReadMore:
 149                         case eReadPayload:
 150                         case eReadFile:
 151                                 curl_easy_cleanup(IO->HttpReq.chnd);
 152                                 IO->HttpReq.chnd = NULL;
 153                                 break;
 154                         case eTerminateConnection:
 155                         case eAbort:
 156                                 curl_easy_cleanup(IO->HttpReq.chnd);
 157                                 IO->HttpReq.chnd = NULL;
 158                                 FreeStrBuf(&IO->HttpReq.ReplyData);
 159                                 FreeURL(&IO->ConnectMe);
 160                                 RemoveContext(IO->CitContext);
 161                                 IO->Terminate(IO);
 162                         }
 163                 }
 164         }
 165 }
 166
 167 static void
 168 stepmulti(void *data, curl_socket_t fd, int which)
 169 {
 170         int running_handles = 0;
 171         CURLMcode msta;
 172
 173         msta = curl_multi_socket_action(global.mhnd, fd, which, &running_handles);
 174         syslog(LOG_DEBUG, "EVCURL: stepmulti(): calling gotstatus()\n");
 175         if (msta)
 176                 syslog(LOG_ERR, "EVCURL: error in curl processing events on multi handle, fd %d: %s\n", (int)fd, curl_multi_strerror(msta));
 177         if (global.nrun != running_handles)
 178                 gotstatus(running_handles);
 179 }
 180
 181 static void
 182 gottime(struct ev_loop *loop, ev_timer *timeev, int events) {
 183         syslog(LOG_DEBUG, "EVCURL: waking up curl for timeout\n");
 184         stepmulti(NULL, CURL_SOCKET_TIMEOUT, 0);
 185 }
 186
 187 static void
 188 got_in(struct ev_loop *loop, ev_io *ioev, int events) {
 189         syslog(LOG_DEBUG, "EVCURL: waking up curl for io on fd %d\n", (int)ioev->fd);
 190         stepmulti(ioev->data, ioev->fd, CURL_CSELECT_IN);
 191 }
 192
 193 static void
 194 got_out(struct ev_loop *loop, ev_io *ioev, int events) {
 195         syslog(LOG_DEBUG, "EVCURL: waking up curl for io on fd %d\n", (int)ioev->fd);
 196         stepmulti(ioev->data, ioev->fd, CURL_CSELECT_OUT);
 197 }
 198
 199 static size_t
 200 gotdata(void *data, size_t size, size_t nmemb, void *cglobal) {
 201         AsyncIO *IO = (AsyncIO*) cglobal;
 202
 203         if (IO->HttpReq.ReplyData == NULL)
 204         {
 205             IO->HttpReq.ReplyData = NewStrBufPlain(NULL, SIZ);
 206         }
 207         return CurlFillStrBuf_callback(data, size, nmemb, IO->HttpReq.ReplyData);
 208 }
 209
 210 static int
 211 gotwatchtime(CURLM *multi, long tblock_ms, void *cglobal) {
 212         syslog(LOG_DEBUG, "EVCURL: gotwatchtime called %ld ms\n", tblock_ms);
 213         evcurl_global_data *global = cglobal;
 214         ev_timer_stop(EV_DEFAULT, &global->timeev);
 215         if (tblock_ms < 0 || 14000 < tblock_ms)
 216                 tblock_ms = 14000;
 217         ev_timer_set(&global->timeev, 0.5e-3 + 1.0e-3 * tblock_ms, 14.0);
 218         ev_timer_start(EV_DEFAULT_UC, &global->timeev);
 219         curl_multi_perform(global, &global->nrun);
 220         return 0;
 221 }
 222
 223 static int
 224 gotwatchsock(CURL *easy, curl_socket_t fd, int action, void *cglobal, void *vIO) {
 225         evcurl_global_data *global = cglobal;
 226         CURLM *mhnd = global->mhnd;
 227         char *f;
 228         AsyncIO *IO = (AsyncIO*) vIO;
 229         CURLcode sta;
 230         const char *Action;
 231
 232         if (IO == NULL) {
 233                 sta = curl_easy_getinfo(easy, CURLINFO_PRIVATE, &f);
 234                 if (sta) {
 235                         EV_syslog(LOG_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta));
 236                         return -1;
 237                 }
 238                 IO = (AsyncIO *) f;
 239                 EV_syslog(LOG_DEBUG, "EVCURL: got socket for URL: %s\n", IO->ConnectMe->PlainUrl);
 240                 if (IO->SendBuf.fd != 0)
 241                 {
 242                         ev_io_stop(event_base, &IO->recv_event);
 243                         ev_io_stop(event_base, &IO->send_event);
 244                 }
 245                 IO->SendBuf.fd = fd;
 246                 ev_io_init(&IO->recv_event, &got_in, fd, EV_READ);
 247                 ev_io_init(&IO->send_event, &got_out, fd, EV_WRITE);
 248                 curl_multi_assign(mhnd, fd, IO);
 249         }
 250
 251         Action = "";
 252         switch (action)
 253         {
 254         case CURL_POLL_NONE:
 255                 Action = "CURL_POLL_NONE";
 256                 break;
 257         case CURL_POLL_REMOVE:
 258                 Action = "CURL_POLL_REMOVE";
 259                 break;
 260         case CURL_POLL_IN:
 261                 Action = "CURL_POLL_IN";
 262                 break;
 263         case CURL_POLL_OUT:
 264                 Action = "CURL_POLL_OUT";
 265                 break;
 266         case CURL_POLL_INOUT:
 267                 Action = "CURL_POLL_INOUT";
 268                 break;
 269         }
 270
 271
 272         EV_syslog(LOG_DEBUG, "EVCURL: gotwatchsock called fd=%d action=%s[%d]\n", (int)fd, Action, action);
 273
 274         switch (action)
 275         {
 276         case CURL_POLL_NONE:
 277                 EVM_syslog(LOG_ERR,"EVCURL: called first time to register this sockwatcker\n");
 278                 break;
 279         case CURL_POLL_REMOVE:
 280                 EVM_syslog(LOG_ERR,"EVCURL: called last time to unregister this sockwatcher\n");
 281                 ev_io_stop(event_base, &IO->recv_event);
 282                 ev_io_stop(event_base, &IO->send_event);
 283                 break;
 284         case CURL_POLL_IN:
 285                 ev_io_start(event_base, &IO->recv_event);
 286                 ev_io_stop(event_base, &IO->send_event);
 287                 break;
 288         case CURL_POLL_OUT:
 289                 ev_io_start(event_base, &IO->send_event);
 290                 ev_io_stop(event_base, &IO->recv_event);
 291                 break;
 292         case CURL_POLL_INOUT:
 293                 ev_io_start(event_base, &IO->send_event);
 294                 ev_io_start(event_base, &IO->recv_event);
 295                 break;
 296         }
 297         return 0;
 298 }
 299
 300 void curl_init_connectionpool(void)
 301 {
 302         CURLM *mhnd ;
 303
 304         ev_timer_init(&global.timeev, &gottime, 14.0, 14.0);
 305         global.timeev.data = (void *)&global;
 306         global.nrun = -1;
 307         CURLcode sta = curl_global_init(CURL_GLOBAL_ALL);
 308
 309         if (sta)
 310         {
 311                 syslog(LOG_ERR,"EVCURL: error initializing curl library: %s\n", curl_easy_strerror(sta));
 312                 exit(1);
 313         }
 314         mhnd = global.mhnd = curl_multi_init();
 315         if (!mhnd)
 316         {
 317                 syslog(LOG_ERR,"EVCURL: error initializing curl multi handle\n");
 318                 exit(3);
 319         }
 320
 321         MOPT(SOCKETFUNCTION, &gotwatchsock);
 322         MOPT(SOCKETDATA, (void *)&global);
 323         MOPT(TIMERFUNCTION, &gotwatchtime);
 324         MOPT(TIMERDATA, (void *)&global);
 325
 326         return;
 327 }
 328
 329
 330
 331
 332 int evcurl_init(AsyncIO *IO,
 333                 void *CustomData,
 334                 const char* Desc,
 335                 IO_CallBack CallBack,
 336                 IO_CallBack Terminate,
 337                 IO_CallBack ShutdownAbort)
 338 {
 339         CURLcode sta;
 340         CURL *chnd;
 341
 342         EVM_syslog(LOG_DEBUG, "EVCURL: evcurl_init called ms\n");
 343         IO->HttpReq.attached = 0;
 344         IO->SendDone = CallBack;
 345         IO->Terminate = Terminate;
 346         IO->ShutdownAbort = ShutdownAbort;
 347         chnd = IO->HttpReq.chnd = curl_easy_init();
 348         if (!chnd)
 349         {
 350                 EVM_syslog(LOG_ERR, "EVCURL: error initializing curl handle\n");
 351                 return 1;
 352         }
 353
 354         strcpy(IO->HttpReq.errdesc, Desc);
 355
 356         OPT(VERBOSE, (long)1);
 357                 /* unset in production */
 358         OPT(NOPROGRESS, 1L);
 359         OPT(NOSIGNAL, 1L);
 360         OPT(FAILONERROR, (long)1);
 361         OPT(ENCODING, "");
 362         OPT(FOLLOWLOCATION, (long)1);
 363         OPT(MAXREDIRS, (long)7);
 364         OPT(USERAGENT, CITADEL);
 365
 366         OPT(TIMEOUT, (long)1800);
 367         OPT(LOW_SPEED_LIMIT, (long)64);
 368         OPT(LOW_SPEED_TIME, (long)600);
 369         OPT(CONNECTTIMEOUT, (long)600);
 370         OPT(PRIVATE, (void *)IO);
 371
 372         OPT(FORBID_REUSE, 1);
 373         OPT(WRITEFUNCTION, &gotdata);
 374         OPT(WRITEDATA, (void *)IO);
 375         OPT(ERRORBUFFER, IO->HttpReq.errdesc);
 376
 377         if (
 378                 (!IsEmptyStr(config.c_ip_addr))
 379                 && (strcmp(config.c_ip_addr, "*"))
 380                 && (strcmp(config.c_ip_addr, "::"))
 381                 && (strcmp(config.c_ip_addr, "0.0.0.0"))
 382         ) {
 383                 OPT(INTERFACE, config.c_ip_addr);
 384         }
 385                 /* point to a structure that points back to the perl structure and stuff */
 386         EV_syslog(LOG_DEBUG, "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl);
 387         OPT(URL, IO->ConnectMe->PlainUrl);
 388         if (StrLength(IO->ConnectMe->CurlCreds))
 389         {
 390                 OPT(HTTPAUTH, (long)CURLAUTH_BASIC);
 391                 OPT(USERPWD, ChrPtr(IO->ConnectMe->CurlCreds));
 392         }
 393 #ifdef CURLOPT_HTTP_CONTENT_DECODING
 394         OPT(HTTP_CONTENT_DECODING, 1);
 395         OPT(ENCODING, "");
 396 #endif
 397         if (StrLength(IO->HttpReq.PostData) > 0)
 398         {
 399                 OPT(POSTFIELDS, ChrPtr(IO->HttpReq.PostData));
 400                 OPT(POSTFIELDSIZE, StrLength(IO->HttpReq.PostData));
 401
 402         }
 403         else if ((IO->HttpReq.PlainPostDataLen != 0) && (IO->HttpReq.PlainPostData != NULL))
 404         {
 405                 OPT(POSTFIELDS, IO->HttpReq.PlainPostData);
 406                 OPT(POSTFIELDSIZE, IO->HttpReq.PlainPostDataLen);
 407         }
 408
 409         IO->HttpReq.headers = curl_slist_append(IO->HttpReq.headers, "Connection: close");
 410         OPT(HTTPHEADER, IO->HttpReq.headers);
 411
 412         return 1;
 413 }
 414
 415
 416 static void IOcurl_abort_shutdown_callback(struct ev_loop *loop, ev_cleanup *watcher, int revents)
 417 {
 418         CURLMcode msta;
 419         AsyncIO *IO = watcher->data;
 420         EV_syslog(LOG_DEBUG, "EVENT Curl: %s\n", __FUNCTION__);
 421
 422         curl_slist_free_all(IO->HttpReq.headers);
 423         msta = curl_multi_remove_handle(global.mhnd, IO->HttpReq.chnd);
 424         if (msta)
 425                 EV_syslog(LOG_ERR, "EVCURL: warning problem detaching completed handle from curl multi: %s\n", curl_multi_strerror(msta));
 426
 427         curl_easy_cleanup(IO->HttpReq.chnd);
 428         IO->HttpReq.chnd = NULL;
 429         ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
 430         ev_io_stop(event_base, &IO->recv_event);
 431         ev_io_stop(event_base, &IO->send_event);
 432         assert(IO->ShutdownAbort);
 433         IO->ShutdownAbort(IO);
 434 }
 435 eNextState
 436 evcurl_handle_start(AsyncIO *IO)
 437 {
 438         CURLMcode msta;
 439         IO->NextState = eConnect;
 440         EVM_syslog(LOG_DEBUG, "EVCURL: attaching to curl multi handle\n");
 441         msta = curl_multi_add_handle(global.mhnd, IO->HttpReq.chnd);
 442         if (msta)
 443                 EV_syslog(LOG_ERR, "EVCURL: error attaching to curl multi handle: %s\n", curl_multi_strerror(msta));
 444         IO->HttpReq.attached = 1;
 445         ev_async_send (event_base, &WakeupCurl);
 446         ev_cleanup_init(&IO->abort_by_shutdown,
 447                         IOcurl_abort_shutdown_callback);
 448         ev_cleanup_start(event_base, &IO->abort_by_shutdown);
 449         return eReadMessage;
 450 }
 451
 452 static void WakeupCurlCallback(EV_P_ ev_async *w, int revents)
 453 {
 454         syslog(LOG_DEBUG, "EVCURL: waking up curl multi handle\n");
 455
 456         curl_multi_perform(&global, CURL_POLL_NONE);
 457 }
 458
 459 static void evcurl_shutdown (void)
 460 {
 461         curl_multi_cleanup(global.mhnd);
 462 }
 463 /*****************************************************************************
 464  *                       libevent integration                                *
 465  *****************************************************************************/
 466 /*
 467  * client event queue plus its methods.
 468  * this currently is the main loop (which may change in some future?)
 469  */
 470 int evbase_count = 0;
 471 int event_add_pipe[2] = {-1, -1};
 472 pthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */
 473 HashList *QueueEvents = NULL;
 474 HashList *InboundEventQueue = NULL;
 475 HashList *InboundEventQueues[2] = { NULL, NULL };
 476
 477 ev_async AddJob;
 478 ev_async ExitEventLoop;
 479
 480 static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
 481 {
 482         HashList *q;
 483         void *v;
 484         HashPos  *It;
 485         long len;
 486         const char *Key;
 487
 488         /* get the control command... */
 489         pthread_mutex_lock(&EventQueueMutex);
 490
 491         if (InboundEventQueues[0] == InboundEventQueue) {
 492                 InboundEventQueue = InboundEventQueues[1];
 493                 q = InboundEventQueues[0];
 494         }
 495         else {
 496                 InboundEventQueue = InboundEventQueues[0];
 497                 q = InboundEventQueues[1];
 498         }
 499         pthread_mutex_unlock(&EventQueueMutex);
 500
 501         It = GetNewHashPos(q, 0);
 502         while (GetNextHashPos(q, It, &len, &Key, &v))
 503         {
 504                 IOAddHandler *h = v;
 505                 if (h->IO->ID == 0)
 506                         h->IO->ID = EvIDSource++;
 507                 h->EvAttch(h->IO);
 508         }
 509         DeleteHashPos(&It);
 510         DeleteHashContent(&q);
 511         syslog(LOG_DEBUG, "EVENT Q Add done.\n");
 512 }
 513
 514
 515 static void EventExitCallback(EV_P_ ev_async *w, int revents)
 516 {
 517         ev_break(event_base, EVBREAK_ALL);
 518
 519         syslog(LOG_DEBUG, "EVENT Q exiting.\n");
 520 }
 521
 522
 523
 524 void InitEventQueue(void)
 525 {
 526         struct rlimit LimitSet;
 527
 528         pthread_mutex_init(&EventQueueMutex, NULL);
 529
 530         if (pipe(event_add_pipe) != 0) {
 531                 syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
 532                 abort();
 533         }
 534         LimitSet.rlim_cur = 1;
 535         LimitSet.rlim_max = 1;
 536         setrlimit(event_add_pipe[1], &LimitSet);
 537
 538         QueueEvents = NewHash(1, Flathash);
 539         InboundEventQueues[0] = NewHash(1, Flathash);
 540         InboundEventQueues[1] = NewHash(1, Flathash);
 541         InboundEventQueue = InboundEventQueues[0];
 542 }
 543 /*
 544  * this thread operates the select() etc. via libev.
 545  *
 546  *
 547  */
 548 void *client_event_thread(void *arg)
 549 {
 550         struct CitContext libev_client_CC;
 551
 552         CtdlFillSystemContext(&libev_client_CC, "LibEv Thread");
 553 //      citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 554         syslog(LOG_DEBUG, "client_ev_thread() initializing\n");
 555
 556         event_base = ev_default_loop (EVFLAG_AUTO);
 557         ev_async_init(&AddJob, QueueEventAddCallback);
 558         ev_async_start(event_base, &AddJob);
 559         ev_async_init(&ExitEventLoop, EventExitCallback);
 560         ev_async_start(event_base, &ExitEventLoop);
 561         ev_async_init(&WakeupCurl, WakeupCurlCallback);
 562         ev_async_start(event_base, &WakeupCurl);
 563
 564         curl_init_connectionpool();
 565
 566         ev_run (event_base, 0);
 567
 568
 569 ///what todo here?      CtdlClearSystemContext();
 570         ev_loop_destroy (EV_DEFAULT_UC);
 571         curl_global_cleanup();
 572         curl_multi_cleanup(global.mhnd);
 573         DeleteHash(&QueueEvents);
 574         InboundEventQueue = NULL;
 575         DeleteHash(&InboundEventQueues[0]);
 576         DeleteHash(&InboundEventQueues[1]);
 577 /*      citthread_mutex_destroy(&EventQueueMutex); TODO */
 578         evcurl_shutdown();
 579
 580         return(NULL);
 581 }
 582 /*------------------------------------------------------------------------------*/
 583 /*
 584  * DB-Queue; does async bdb operations.
 585  * has its own set of handlers.
 586  */
 587 ev_loop *event_db;
 588 int evdb_count = 0;
 589 int evdb_add_pipe[2] = {-1, -1};
 590 pthread_mutex_t DBEventQueueMutex; /* locks the access to the following vars: */
 591 HashList *DBQueueEvents = NULL;
 592 HashList *DBInboundEventQueue = NULL;
 593 HashList *DBInboundEventQueues[2] = { NULL, NULL };
 594
 595 ev_async DBAddJob;
 596 ev_async DBExitEventLoop;
 597
 598 extern void ShutDownDBCLient(AsyncIO *IO);
 599
 600 static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents)
 601 {
 602         HashList *q;
 603         void *v;
 604         HashPos  *It;
 605         long len;
 606         const char *Key;
 607
 608         /* get the control command... */
 609         pthread_mutex_lock(&DBEventQueueMutex);
 610
 611         if (DBInboundEventQueues[0] == DBInboundEventQueue) {
 612                 DBInboundEventQueue = DBInboundEventQueues[1];
 613                 q = DBInboundEventQueues[0];
 614         }
 615         else {
 616                 DBInboundEventQueue = DBInboundEventQueues[0];
 617                 q = DBInboundEventQueues[1];
 618         }
 619         pthread_mutex_unlock(&DBEventQueueMutex);
 620
 621         It = GetNewHashPos(q, 0);
 622         while (GetNextHashPos(q, It, &len, &Key, &v))
 623         {
 624                 IOAddHandler *h = v;
 625                 eNextState rc;
 626                 if (h->IO->ID == 0)
 627                         h->IO->ID = EvIDSource++;
 628                 rc = h->EvAttch(h->IO);
 629                 switch (rc)
 630                 {
 631                 case eAbort:
 632                     ShutDownDBCLient(h->IO);
 633                 default:
 634                     break;
 635                 }
 636         }
 637         DeleteHashPos(&It);
 638         DeleteHashContent(&q);
 639         syslog(LOG_DEBUG, "DBEVENT Q Add done.\n");
 640 }
 641
 642
 643 static void DBEventExitCallback(EV_P_ ev_async *w, int revents)
 644 {
 645         syslog(LOG_DEBUG, "EVENT Q exiting.\n");
 646         ev_break(event_db, EVBREAK_ALL);
 647 }
 648
 649
 650
 651 void DBInitEventQueue(void)
 652 {
 653         struct rlimit LimitSet;
 654
 655         pthread_mutex_init(&DBEventQueueMutex, NULL);
 656
 657         if (pipe(evdb_add_pipe) != 0) {
 658                 syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
 659                 abort();
 660         }
 661         LimitSet.rlim_cur = 1;
 662         LimitSet.rlim_max = 1;
 663         setrlimit(evdb_add_pipe[1], &LimitSet);
 664
 665         DBQueueEvents = NewHash(1, Flathash);
 666         DBInboundEventQueues[0] = NewHash(1, Flathash);
 667         DBInboundEventQueues[1] = NewHash(1, Flathash);
 668         DBInboundEventQueue = DBInboundEventQueues[0];
 669 }
 670
 671 /*
 672  * this thread operates writing to the message database via libev.
 673  *
 674  *
 675  */
 676 void *db_event_thread(void *arg)
 677 {
 678         struct CitContext libev_msg_CC;
 679
 680         CtdlFillSystemContext(&libev_msg_CC, "LibEv DB IO Thread");
 681 //      citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 682         syslog(LOG_DEBUG, "client_msgev_thread() initializing\n");
 683
 684         event_db = ev_loop_new (EVFLAG_AUTO);
 685
 686         ev_async_init(&DBAddJob, DBQueueEventAddCallback);
 687         ev_async_start(event_db, &DBAddJob);
 688         ev_async_init(&DBExitEventLoop, DBEventExitCallback);
 689         ev_async_start(event_db, &DBExitEventLoop);
 690
 691         ev_run (event_db, 0);
 692
 693
 694 //// what to do here?   CtdlClearSystemContext();
 695         ev_loop_destroy (event_db);
 696
 697         DeleteHash(&DBQueueEvents);
 698         DBInboundEventQueue = NULL;
 699         DeleteHash(&DBInboundEventQueues[0]);
 700         DeleteHash(&DBInboundEventQueues[1]);
 701 /*      citthread_mutex_destroy(&DBEventQueueMutex); TODO */
 702
 703         return(NULL);
 704 }
 705
 706 CTDL_MODULE_INIT(event_client)
 707 {
 708         if (!threading)
 709         {
 710                 InitEventQueue();
 711                 DBInitEventQueue();
 712                 CtdlThreadCreate(/*"Client event", */ client_event_thread);
 713                 CtdlThreadCreate(/*"DB event", */db_event_thread);
 714 /// todo register shutdown callback.
 715         }
 716         return "event";
 717 }