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 }