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