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)0);
 363         OPT(MAXREDIRS, (long)0);
 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_global_cleanup();
 462         curl_multi_cleanup(global.mhnd);
 463         syslog(LOG_DEBUG, "client_event_thread() initializing\n");
 464 }
 465 /*****************************************************************************
 466  *                       libevent integration                                *
 467  *****************************************************************************/
 468 /*
 469  * client event queue plus its methods.
 470  * this currently is the main loop (which may change in some future?)
 471  */
 472 int evbase_count = 0;
 473 int event_add_pipe[2] = {-1, -1};
 474 pthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */
 475 HashList *QueueEvents = NULL;
 476 HashList *InboundEventQueue = NULL;
 477 HashList *InboundEventQueues[2] = { NULL, NULL };
 478
 479 ev_async AddJob;
 480 ev_async ExitEventLoop;
 481
 482 static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
 483 {
 484         HashList *q;
 485         void *v;
 486         HashPos  *It;
 487         long len;
 488         const char *Key;
 489
 490         /* get the control command... */
 491         pthread_mutex_lock(&EventQueueMutex);
 492
 493         if (InboundEventQueues[0] == InboundEventQueue) {
 494                 InboundEventQueue = InboundEventQueues[1];
 495                 q = InboundEventQueues[0];
 496         }
 497         else {
 498                 InboundEventQueue = InboundEventQueues[0];
 499                 q = InboundEventQueues[1];
 500         }
 501         pthread_mutex_unlock(&EventQueueMutex);
 502
 503         It = GetNewHashPos(q, 0);
 504         while (GetNextHashPos(q, It, &len, &Key, &v))
 505         {
 506                 IOAddHandler *h = v;
 507                 if (h->IO->ID == 0)
 508                         h->IO->ID = EvIDSource++;
 509                 h->EvAttch(h->IO);
 510         }
 511         DeleteHashPos(&It);
 512         DeleteHashContent(&q);
 513         syslog(LOG_DEBUG, "EVENT Q Add done.\n");
 514 }
 515
 516
 517 static void EventExitCallback(EV_P_ ev_async *w, int revents)
 518 {
 519         ev_break(event_base, EVBREAK_ALL);
 520
 521         syslog(LOG_DEBUG, "EVENT Q exiting.\n");
 522 }
 523
 524
 525
 526 void InitEventQueue(void)
 527 {
 528         struct rlimit LimitSet;
 529
 530         pthread_mutex_init(&EventQueueMutex, NULL);
 531
 532         if (pipe(event_add_pipe) != 0) {
 533                 syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
 534                 abort();
 535         }
 536         LimitSet.rlim_cur = 1;
 537         LimitSet.rlim_max = 1;
 538         setrlimit(event_add_pipe[1], &LimitSet);
 539
 540         QueueEvents = NewHash(1, Flathash);
 541         InboundEventQueues[0] = NewHash(1, Flathash);
 542         InboundEventQueues[1] = NewHash(1, Flathash);
 543         InboundEventQueue = InboundEventQueues[0];
 544 }
 545 /*
 546  * this thread operates the select() etc. via libev.
 547  *
 548  *
 549  */
 550 void *client_event_thread(void *arg)
 551 {
 552         struct CitContext libev_client_CC;
 553
 554         CtdlFillSystemContext(&libev_client_CC, "LibEv Thread");
 555 //      citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 556         syslog(LOG_DEBUG, "client_event_thread() initializing\n");
 557
 558         event_base = ev_default_loop (EVFLAG_AUTO);
 559         ev_async_init(&AddJob, QueueEventAddCallback);
 560         ev_async_start(event_base, &AddJob);
 561         ev_async_init(&ExitEventLoop, EventExitCallback);
 562         ev_async_start(event_base, &ExitEventLoop);
 563         ev_async_init(&WakeupCurl, WakeupCurlCallback);
 564         ev_async_start(event_base, &WakeupCurl);
 565
 566         curl_init_connectionpool();
 567
 568         ev_run (event_base, 0);
 569
 570         syslog(LOG_DEBUG, "client_event_thread() exiting\n");
 571
 572 ///what todo here?      CtdlClearSystemContext();
 573         ev_loop_destroy (EV_DEFAULT_UC);
 574         DeleteHash(&QueueEvents);
 575         InboundEventQueue = NULL;
 576         DeleteHash(&InboundEventQueues[0]);
 577         DeleteHash(&InboundEventQueues[1]);
 578 /*      citthread_mutex_destroy(&EventQueueMutex); TODO */
 579         evcurl_shutdown();
 580
 581         return(NULL);
 582 }
 583 /*------------------------------------------------------------------------------*/
 584 /*
 585  * DB-Queue; does async bdb operations.
 586  * has its own set of handlers.
 587  */
 588 ev_loop *event_db;
 589 int evdb_count = 0;
 590 int evdb_add_pipe[2] = {-1, -1};
 591 pthread_mutex_t DBEventQueueMutex; /* locks the access to the following vars: */
 592 HashList *DBQueueEvents = NULL;
 593 HashList *DBInboundEventQueue = NULL;
 594 HashList *DBInboundEventQueues[2] = { NULL, NULL };
 595
 596 ev_async DBAddJob;
 597 ev_async DBExitEventLoop;
 598
 599 extern void ShutDownDBCLient(AsyncIO *IO);
 600
 601 static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents)
 602 {
 603         HashList *q;
 604         void *v;
 605         HashPos  *It;
 606         long len;
 607         const char *Key;
 608
 609         /* get the control command... */
 610         pthread_mutex_lock(&DBEventQueueMutex);
 611
 612         if (DBInboundEventQueues[0] == DBInboundEventQueue) {
 613                 DBInboundEventQueue = DBInboundEventQueues[1];
 614                 q = DBInboundEventQueues[0];
 615         }
 616         else {
 617                 DBInboundEventQueue = DBInboundEventQueues[0];
 618                 q = DBInboundEventQueues[1];
 619         }
 620         pthread_mutex_unlock(&DBEventQueueMutex);
 621
 622         It = GetNewHashPos(q, 0);
 623         while (GetNextHashPos(q, It, &len, &Key, &v))
 624         {
 625                 IOAddHandler *h = v;
 626                 eNextState rc;
 627                 if (h->IO->ID == 0)
 628                         h->IO->ID = EvIDSource++;
 629                 rc = h->EvAttch(h->IO);
 630                 switch (rc)
 631                 {
 632                 case eAbort:
 633                     ShutDownDBCLient(h->IO);
 634                 default:
 635                     break;
 636                 }
 637         }
 638         DeleteHashPos(&It);
 639         DeleteHashContent(&q);
 640         syslog(LOG_DEBUG, "DBEVENT Q Add done.\n");
 641 }
 642
 643
 644 static void DBEventExitCallback(EV_P_ ev_async *w, int revents)
 645 {
 646         syslog(LOG_DEBUG, "DB EVENT Q exiting.\n");
 647         ev_break(event_db, EVBREAK_ALL);
 648 }
 649
 650
 651
 652 void DBInitEventQueue(void)
 653 {
 654         struct rlimit LimitSet;
 655
 656         pthread_mutex_init(&DBEventQueueMutex, NULL);
 657
 658         if (pipe(evdb_add_pipe) != 0) {
 659                 syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
 660                 abort();
 661         }
 662         LimitSet.rlim_cur = 1;
 663         LimitSet.rlim_max = 1;
 664         setrlimit(evdb_add_pipe[1], &LimitSet);
 665
 666         DBQueueEvents = NewHash(1, Flathash);
 667         DBInboundEventQueues[0] = NewHash(1, Flathash);
 668         DBInboundEventQueues[1] = NewHash(1, Flathash);
 669         DBInboundEventQueue = DBInboundEventQueues[0];
 670 }
 671
 672 /*
 673  * this thread operates writing to the message database via libev.
 674  *
 675  *
 676  */
 677 void *db_event_thread(void *arg)
 678 {
 679         struct CitContext libev_msg_CC;
 680
 681         CtdlFillSystemContext(&libev_msg_CC, "LibEv DB IO Thread");
 682 //      citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 683         syslog(LOG_DEBUG, "dbevent_thread() initializing\n");
 684
 685         event_db = ev_loop_new (EVFLAG_AUTO);
 686
 687         ev_async_init(&DBAddJob, DBQueueEventAddCallback);
 688         ev_async_start(event_db, &DBAddJob);
 689         ev_async_init(&DBExitEventLoop, DBEventExitCallback);
 690         ev_async_start(event_db, &DBExitEventLoop);
 691
 692         ev_run (event_db, 0);
 693
 694         syslog(LOG_DEBUG, "dbevent_thread() exiting\n");
 695
 696 //// what to do here?   CtdlClearSystemContext();
 697         ev_loop_destroy (event_db);
 698
 699         DeleteHash(&DBQueueEvents);
 700         DBInboundEventQueue = NULL;
 701         DeleteHash(&DBInboundEventQueues[0]);
 702         DeleteHash(&DBInboundEventQueues[1]);
 703 /*      citthread_mutex_destroy(&DBEventQueueMutex); TODO */
 704
 705         return(NULL);
 706 }
 707
 708 void ShutDownEventQueues(void)
 709 {
 710         syslog(LOG_DEBUG, "EVENT Qs triggering exits.\n");
 711
 712         pthread_mutex_lock(&DBEventQueueMutex);
 713         ev_async_send (event_db, &DBExitEventLoop);
 714         pthread_mutex_unlock(&DBEventQueueMutex);
 715
 716         pthread_mutex_lock(&EventQueueMutex);
 717         ev_async_send (EV_DEFAULT_ &ExitEventLoop);
 718         pthread_mutex_unlock(&EventQueueMutex);
 719 }
 720
 721 CTDL_MODULE_INIT(event_client)
 722 {
 723         if (!threading)
 724         {
 725                 CtdlRegisterCleanupHook(ShutDownEventQueues);
 726                 InitEventQueue();
 727                 DBInitEventQueue();
 728                 CtdlThreadCreate(/*"Client event", */ client_event_thread);
 729                 CtdlThreadCreate(/*"DB event", */db_event_thread);
 730 /// todo register shutdown callback.
 731         }
 732         return "event";
 733 }