X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fmodules%2Feventclient%2Fserv_eventclient.c;h=7761b662e6a50da621a63a01611f3e14ea9d03f4;hb=5e122974be0e2fda97d106dfb9f8ffec98225582;hp=05f8e6f1c298d679af96fe4e81b1a40353485cdf;hpb=67d954e97ccee1cd9c9ae8a969eece1383a04d46;p=citadel.git diff --git a/citadel/modules/eventclient/serv_eventclient.c b/citadel/modules/eventclient/serv_eventclient.c index 05f8e6f1c..7761b662e 100644 --- a/citadel/modules/eventclient/serv_eventclient.c +++ b/citadel/modules/eventclient/serv_eventclient.c @@ -1,19 +1,19 @@ /* - * Copyright (c) 1998-2009 by the citadel.org team + * Copyright (c) 1998-2012 by the citadel.org team * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation; either version 3 of the License, or - * (at your option) any later version. + * This program is open source software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 3. + * + * * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * - * You should have received a copy of the GNU General Public License - * along with this program; if not, write to the Free Software - * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * + * + * */ #include "sysdep.h" @@ -44,6 +44,7 @@ #include #include #include +#include #include #include #include @@ -55,25 +56,46 @@ #include "ctdl_module.h" -#ifdef EXPERIMENTAL_SMTP_EVENT_CLIENT - #include "event_client.h" #include "serv_curl.h" ev_loop *event_base; +int DebugEventLoop = 0; +int DebugEventLoopBacktrace = 0; +int DebugCurl = 0; +long EvIDSource = 1; /***************************************************************************** * libevent / curl integration * *****************************************************************************/ +#define DBGLOG(LEVEL) if ((LEVEL != LOG_DEBUG) || (DebugCurl != 0)) + +#define EVCURL_syslog(LEVEL, FORMAT, ...) \ + DBGLOG (LEVEL) syslog(LEVEL, "EVCURL:IO[%ld]CC[%d] " FORMAT, \ + IO->ID, CCID, __VA_ARGS__) + +#define EVCURLM_syslog(LEVEL, FORMAT) \ + DBGLOG (LEVEL) syslog(LEVEL, "EVCURL:IO[%ld]CC[%d] " FORMAT, \ + IO->ID, CCID) + +#define CURL_syslog(LEVEL, FORMAT, ...) \ + DBGLOG (LEVEL) syslog(LEVEL, "CURL: " FORMAT, __VA_ARGS__) + +#define CURLM_syslog(LEVEL, FORMAT) \ + DBGLOG (LEVEL) syslog(LEVEL, "CURL: " FORMAT) + #define MOPT(s, v) \ do { \ sta = curl_multi_setopt(mhnd, (CURLMOPT_##s), (v)); \ if (sta) { \ - CtdlLogPrintf(CTDL_ERR, "EVCURL: error setting option " #s " on curl multi handle: %s\n", curl_easy_strerror(sta)); \ + EVQ_syslog(LOG_ERR, "error setting option " \ + #s " on curl multi handle: %s\n", \ + curl_easy_strerror(sta)); \ exit (1); \ } \ } while (0) + typedef struct _evcurl_global_data { int magic; CURLM *mhnd; @@ -81,163 +103,292 @@ typedef struct _evcurl_global_data { int nrun; } evcurl_global_data; -typedef struct _sockwatcher_data -{ - evcurl_global_data *global; - ev_io ioev; -} sockwatcher_data; - ev_async WakeupCurl; evcurl_global_data global; static void -gotstatus(evcurl_global_data *global, int nnrun) +gotstatus(int nnrun) { - CURLM *mhnd; CURLMsg *msg; int nmsg; -/* - if (EVCURL_GLOBAL_MAGIC != global.magic) - { - CtdlLogPrintf(CTDL_ERR, "internal error: gotstatus on wrong struct"); - return; - } -*/ - global->nrun = nnrun; - mhnd = global->mhnd; - - CtdlLogPrintf(CTDL_DEBUG, "CURLEV: gotstatus(): about to call curl_multi_info_read\n"); - while ((msg = curl_multi_info_read(mhnd, &nmsg))) { - CtdlLogPrintf(CTDL_ERR, "EVCURL: got curl multi_info message msg=%d\n", msg->msg); + + global.nrun = nnrun; + + CURLM_syslog(LOG_DEBUG, + "gotstatus(): about to call curl_multi_info_read\n"); + while ((msg = curl_multi_info_read(global.mhnd, &nmsg))) { + CURL_syslog(LOG_DEBUG, + "got curl multi_info message msg=%d\n", + msg->msg); + if (CURLMSG_DONE == msg->msg) { CURL *chnd; - char *chandle; + char *chandle = NULL; CURLcode sta; CURLMcode msta; - AsyncIO *IO; + AsyncIO*IO; chandle = NULL;; chnd = msg->easy_handle; - sta = curl_easy_getinfo(chnd, CURLINFO_PRIVATE, &chandle); - CtdlLogPrintf(CTDL_ERR, "EVCURL: request complete\n"); - if (sta) - CtdlLogPrintf(CTDL_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta)); + sta = curl_easy_getinfo(chnd, + CURLINFO_PRIVATE, + &chandle); + EVCURLM_syslog(LOG_DEBUG, "request complete\n"); + if (sta) { + EVCURL_syslog(LOG_ERR, + "error asking curl for private" + " cookie of curl handle: %s\n", + curl_easy_strerror(sta)); + continue; + } IO = (AsyncIO *)chandle; - + + IO->Now = ev_now(event_base); + + ev_io_stop(event_base, &IO->recv_event); + ev_io_stop(event_base, &IO->send_event); + sta = msg->data.result; if (sta) { - CtdlLogPrintf(CTDL_ERR, "EVCURL: error description: %s\n", IO->HttpReq.errdesc); - CtdlLogPrintf(CTDL_ERR, "EVCURL: error performing request: %s\n", curl_easy_strerror(sta)); + EVCURL_syslog(LOG_ERR, + "error description: %s\n", + IO->HttpReq.errdesc); + EVCURL_syslog(LOG_ERR, + "error performing request: %s\n", + curl_easy_strerror(sta)); } - sta = curl_easy_getinfo(chnd, CURLINFO_RESPONSE_CODE, &IO->HttpReq.httpcode); + sta = curl_easy_getinfo(chnd, + CURLINFO_RESPONSE_CODE, + &IO->HttpReq.httpcode); if (sta) - CtdlLogPrintf(CTDL_ERR, "EVCURL: error asking curl for response code from request: %s\n", curl_easy_strerror(sta)); - CtdlLogPrintf(CTDL_ERR, "EVCURL: http response code was %ld\n", (long)IO->HttpReq.httpcode); - msta = curl_multi_remove_handle(mhnd, chnd); + EVCURL_syslog(LOG_ERR, + "error asking curl for " + "response code from request: %s\n", + curl_easy_strerror(sta)); + EVCURL_syslog(LOG_ERR, + "http response code was %ld\n", + (long)IO->HttpReq.httpcode); + + + curl_slist_free_all(IO->HttpReq.headers); + msta = curl_multi_remove_handle(global.mhnd, chnd); if (msta) - CtdlLogPrintf(CTDL_ERR, "EVCURL: warning problem detaching completed handle from curl multi: %s\n", curl_multi_strerror(msta)); + EVCURL_syslog(LOG_ERR, + "warning problem detaching " + "completed handle from curl multi: " + "%s\n", + curl_multi_strerror(msta)); + + ev_cleanup_stop(event_base, &IO->abort_by_shutdown); IO->HttpReq.attached = 0; - IO->SendDone(IO); - curl_easy_cleanup(IO->HttpReq.chnd); - curl_slist_free_all(IO->HttpReq.headers); - FreeStrBuf(&IO->HttpReq.ReplyData); - FreeURL(&IO->ConnectMe); - RemoveContext(IO->CitContext); - IO->Terminate(IO); + switch(IO->SendDone(IO)) + { + case eDBQuery: + curl_easy_cleanup(IO->HttpReq.chnd); + IO->HttpReq.chnd = NULL; + break; + case eSendDNSQuery: + case eReadDNSReply: + case eConnect: + case eSendReply: + case eSendMore: + case eSendFile: + case eReadMessage: + case eReadMore: + case eReadPayload: + case eReadFile: + curl_easy_cleanup(IO->HttpReq.chnd); + IO->HttpReq.chnd = NULL; + break; + case eTerminateConnection: + case eAbort: + curl_easy_cleanup(IO->HttpReq.chnd); + IO->HttpReq.chnd = NULL; + FreeStrBuf(&IO->HttpReq.ReplyData); + FreeURL(&IO->ConnectMe); + RemoveContext(IO->CitContext); + IO->Terminate(IO); + } } } } static void -stepmulti(evcurl_global_data *global, curl_socket_t fd) { - int nnrun; +stepmulti(void *data, curl_socket_t fd, int which) +{ + int running_handles = 0; CURLMcode msta; - - if (global == NULL) { - CtdlLogPrintf(CTDL_DEBUG, "EVCURL: stepmulti(NULL): wtf?\n"); - return; - } - msta = curl_multi_socket_action(global->mhnd, fd, 0, &nnrun); - CtdlLogPrintf(CTDL_DEBUG, "EVCURL: stepmulti(): calling gotstatus()\n"); + + msta = curl_multi_socket_action(global.mhnd, + fd, + which, + &running_handles); + + CURLM_syslog(LOG_DEBUG, "stepmulti(): calling gotstatus()\n"); if (msta) - CtdlLogPrintf(CTDL_ERR, "EVCURL: error in curl processing events on multi handle, fd %d: %s\n", (int)fd, curl_multi_strerror(msta)); - if (global->nrun != nnrun) - gotstatus(global, nnrun); + CURL_syslog(LOG_ERR, + "error in curl processing events" + "on multi handle, fd %d: %s\n", + (int)fd, + curl_multi_strerror(msta)); + + if (global.nrun != running_handles) + gotstatus(running_handles); } static void -gottime(struct ev_loop *loop, ev_timer *timeev, int events) { - CtdlLogPrintf(CTDL_DEBUG, "EVCURL: waking up curl for timeout\n"); - evcurl_global_data *global = (void *)timeev->data; - stepmulti(global, CURL_SOCKET_TIMEOUT); +gottime(struct ev_loop *loop, ev_timer *timeev, int events) +{ + CURLM_syslog(LOG_DEBUG, "EVCURL: waking up curl for timeout\n"); + stepmulti(NULL, CURL_SOCKET_TIMEOUT, 0); +} + +static void +got_in(struct ev_loop *loop, ev_io *ioev, int events) +{ + CURL_syslog(LOG_DEBUG, + "EVCURL: waking up curl for io on fd %d\n", + (int)ioev->fd); + + stepmulti(ioev->data, ioev->fd, CURL_CSELECT_IN); } static void -gotio(struct ev_loop *loop, ev_io *ioev, int events) { - CtdlLogPrintf(CTDL_DEBUG, "EVCURL: waking up curl for io on fd %d\n", (int)ioev->fd); - sockwatcher_data *sockwatcher = (void *)ioev->data; - stepmulti(sockwatcher->global, ioev->fd); +got_out(struct ev_loop *loop, ev_io *ioev, int events) +{ + CURL_syslog(LOG_DEBUG, + "waking up curl for io on fd %d\n", + (int)ioev->fd); + + stepmulti(ioev->data, ioev->fd, CURL_CSELECT_OUT); } static size_t gotdata(void *data, size_t size, size_t nmemb, void *cglobal) { AsyncIO *IO = (AsyncIO*) cglobal; - //evcurl_request_data *D = (evcurl_request_data*) data; - CtdlLogPrintf(CTDL_DEBUG, "EVCURL: gotdata(): calling CurlFillStrBuf_callback()\n"); if (IO->HttpReq.ReplyData == NULL) { - IO->HttpReq.ReplyData = NewStrBufPlain(NULL, SIZ); + IO->HttpReq.ReplyData = NewStrBufPlain(NULL, SIZ); } - return CurlFillStrBuf_callback(data, size, nmemb, IO->HttpReq.ReplyData); + IO->Now = ev_now(event_base); + return CurlFillStrBuf_callback(data, + size, + nmemb, + IO->HttpReq.ReplyData); } static int gotwatchtime(CURLM *multi, long tblock_ms, void *cglobal) { - CtdlLogPrintf(CTDL_DEBUG, "EVCURL: gotwatchtime called %ld ms\n", tblock_ms); + CURL_syslog(LOG_DEBUG, "EVCURL: gotwatchtime called %ld ms\n", tblock_ms); evcurl_global_data *global = cglobal; ev_timer_stop(EV_DEFAULT, &global->timeev); if (tblock_ms < 0 || 14000 < tblock_ms) tblock_ms = 14000; ev_timer_set(&global->timeev, 0.5e-3 + 1.0e-3 * tblock_ms, 14.0); ev_timer_start(EV_DEFAULT_UC, &global->timeev); - curl_multi_perform(global, CURL_POLL_NONE); + curl_multi_perform(global, &global->nrun); return 0; } static int -gotwatchsock(CURL *easy, curl_socket_t fd, int action, void *cglobal, void *csockwatcher) { +gotwatchsock(CURL *easy, + curl_socket_t fd, + int action, + void *cglobal, + void *vIO) +{ evcurl_global_data *global = cglobal; CURLM *mhnd = global->mhnd; - sockwatcher_data *sockwatcher = csockwatcher; + char *f; + AsyncIO *IO = (AsyncIO*) vIO; + CURLcode sta; + const char *Action; + + if (IO == NULL) { + sta = curl_easy_getinfo(easy, CURLINFO_PRIVATE, &f); + if (sta) { + EVCURL_syslog(LOG_ERR, + "EVCURL: error asking curl for private " + "cookie of curl handle: %s\n", + curl_easy_strerror(sta)); + return -1; + } + IO = (AsyncIO *) f; + EVCURL_syslog(LOG_DEBUG, + "EVCURL: got socket for URL: %s\n", + IO->ConnectMe->PlainUrl); + + if (IO->SendBuf.fd != 0) + { + ev_io_stop(event_base, &IO->recv_event); + ev_io_stop(event_base, &IO->send_event); + } + IO->SendBuf.fd = fd; + ev_io_init(&IO->recv_event, &got_in, fd, EV_READ); + ev_io_init(&IO->send_event, &got_out, fd, EV_WRITE); + curl_multi_assign(mhnd, fd, IO); + } - CtdlLogPrintf(CTDL_DEBUG,"EVCURL: gotwatchsock called fd=%d action=%d\n", (int)fd, action); + IO->Now = ev_now(event_base); - if (!sockwatcher) { - CtdlLogPrintf(CTDL_ERR,"EVCURL: called first time to register this sockwatcker\n"); - sockwatcher = malloc(sizeof(sockwatcher_data)); - sockwatcher->global = global; - ev_init(&sockwatcher->ioev, &gotio); - sockwatcher->ioev.data = (void *)sockwatcher; - curl_multi_assign(mhnd, fd, sockwatcher); + Action = ""; + switch (action) + { + case CURL_POLL_NONE: + Action = "CURL_POLL_NONE"; + break; + case CURL_POLL_REMOVE: + Action = "CURL_POLL_REMOVE"; + break; + case CURL_POLL_IN: + Action = "CURL_POLL_IN"; + break; + case CURL_POLL_OUT: + Action = "CURL_POLL_OUT"; + break; + case CURL_POLL_INOUT: + Action = "CURL_POLL_INOUT"; + break; } - if (CURL_POLL_REMOVE == action) { - CtdlLogPrintf(CTDL_ERR,"EVCURL: called last time to unregister this sockwatcher\n"); - ev_io_stop(event_base, &sockwatcher->ioev); - free(sockwatcher); - } else { - int events = (action & CURL_POLL_IN ? EV_READ : 0) | (action & CURL_POLL_OUT ? EV_WRITE : 0); - ev_io_stop(EV_DEFAULT, &sockwatcher->ioev); - if (events) { - ev_io_set(&sockwatcher->ioev, fd, events); - ev_io_start(EV_DEFAULT, &sockwatcher->ioev); - } + + + EVCURL_syslog(LOG_DEBUG, + "EVCURL: gotwatchsock called fd=%d action=%s[%d]\n", + (int)fd, Action, action); + + switch (action) + { + case CURL_POLL_NONE: + EVCURLM_syslog(LOG_DEBUG, + "called first time " + "to register this sockwatcker\n"); + break; + case CURL_POLL_REMOVE: + EVCURLM_syslog(LOG_DEBUG, + "called last time to unregister " + "this sockwatcher\n"); + ev_io_stop(event_base, &IO->recv_event); + ev_io_stop(event_base, &IO->send_event); + break; + case CURL_POLL_IN: + ev_io_start(event_base, &IO->recv_event); + ev_io_stop(event_base, &IO->send_event); + break; + case CURL_POLL_OUT: + ev_io_start(event_base, &IO->send_event); + ev_io_stop(event_base, &IO->recv_event); + break; + case CURL_POLL_INOUT: + ev_io_start(event_base, &IO->send_event); + ev_io_start(event_base, &IO->recv_event); + break; } return 0; } -void curl_init_connectionpool(void) +void curl_init_connectionpool(void) { CURLM *mhnd ; @@ -246,15 +397,19 @@ void curl_init_connectionpool(void) global.nrun = -1; CURLcode sta = curl_global_init(CURL_GLOBAL_ALL); - if (sta) + if (sta) { - CtdlLogPrintf(CTDL_ERR,"EVCURL: error initializing curl library: %s\n", curl_easy_strerror(sta)); + CURL_syslog(LOG_ERR, + "error initializing curl library: %s\n", + curl_easy_strerror(sta)); + exit(1); } mhnd = global.mhnd = curl_multi_init(); if (!mhnd) { - CtdlLogPrintf(CTDL_ERR,"EVCURL: error initializing curl multi handle\n"); + CURLM_syslog(LOG_ERR, + "error initializing curl multi handle\n"); exit(3); } @@ -266,113 +421,152 @@ void curl_init_connectionpool(void) return; } - - - -int evcurl_init(AsyncIO *IO, - void *CustomData, - const char* Desc, - IO_CallBack CallBack, - IO_CallBack Terminate) +int evcurl_init(AsyncIO *IO) { CURLcode sta; CURL *chnd; - CtdlLogPrintf(CTDL_DEBUG,"EVCURL: evcurl_init called ms\n"); + EVCURLM_syslog(LOG_DEBUG, "EVCURL: evcurl_init called ms\n"); IO->HttpReq.attached = 0; - IO->SendDone = CallBack; - IO->Terminate = Terminate; chnd = IO->HttpReq.chnd = curl_easy_init(); if (!chnd) { - CtdlLogPrintf(CTDL_ERR, "EVCURL: error initializing curl handle\n"); - return 1; + EVCURLM_syslog(LOG_ERR, "EVCURL: error initializing curl handle\n"); + return 0; } - strcpy(IO->HttpReq.errdesc, Desc); - +#if DEBUG OPT(VERBOSE, (long)1); - /* unset in production */ - OPT(NOPROGRESS, (long)1); - OPT(NOSIGNAL, (long)1); +#endif + OPT(NOPROGRESS, 1L); + + OPT(NOSIGNAL, 1L); OPT(FAILONERROR, (long)1); OPT(ENCODING, ""); - OPT(FOLLOWLOCATION, (long)1); - OPT(MAXREDIRS, (long)7); + OPT(FOLLOWLOCATION, (long)0); + OPT(MAXREDIRS, (long)0); OPT(USERAGENT, CITADEL); OPT(TIMEOUT, (long)1800); OPT(LOW_SPEED_LIMIT, (long)64); OPT(LOW_SPEED_TIME, (long)600); - OPT(CONNECTTIMEOUT, (long)600); + OPT(CONNECTTIMEOUT, (long)600); OPT(PRIVATE, (void *)IO); - - OPT(WRITEFUNCTION, &gotdata); + OPT(FORBID_REUSE, 1); + OPT(WRITEFUNCTION, &gotdata); OPT(WRITEDATA, (void *)IO); OPT(ERRORBUFFER, IO->HttpReq.errdesc); - if ( - (!IsEmptyStr(config.c_ip_addr)) + if ((!IsEmptyStr(config.c_ip_addr)) && (strcmp(config.c_ip_addr, "*")) && (strcmp(config.c_ip_addr, "::")) && (strcmp(config.c_ip_addr, "0.0.0.0")) - ) { + ) + { OPT(INTERFACE, config.c_ip_addr); } - /* point to a structure that points back to the perl structure and stuff */ - CtdlLogPrintf(CTDL_DEBUG, "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl); + +#ifdef CURLOPT_HTTP_CONTENT_DECODING + OPT(HTTP_CONTENT_DECODING, 1); + OPT(ENCODING, ""); +#endif + + IO->HttpReq.headers = curl_slist_append(IO->HttpReq.headers, + "Connection: close"); + + return 1; +} + + +static void IOcurl_abort_shutdown_callback(struct ev_loop *loop, + ev_cleanup *watcher, + int revents) +{ + CURLMcode msta; + AsyncIO *IO = watcher->data; + IO->Now = ev_now(event_base); + EVCURL_syslog(LOG_DEBUG, "EVENT Curl: %s\n", __FUNCTION__); + + curl_slist_free_all(IO->HttpReq.headers); + msta = curl_multi_remove_handle(global.mhnd, IO->HttpReq.chnd); + if (msta) + { + EVCURL_syslog(LOG_ERR, + "EVCURL: warning problem detaching completed handle " + "from curl multi: %s\n", + curl_multi_strerror(msta)); + } + + curl_easy_cleanup(IO->HttpReq.chnd); + IO->HttpReq.chnd = NULL; + ev_cleanup_stop(event_base, &IO->abort_by_shutdown); + ev_io_stop(event_base, &IO->recv_event); + ev_io_stop(event_base, &IO->send_event); + assert(IO->ShutdownAbort); + IO->ShutdownAbort(IO); +} +eNextState +evcurl_handle_start(AsyncIO *IO) +{ + CURLMcode msta; + CURLcode sta; + CURL *chnd; + + chnd = IO->HttpReq.chnd; + EVCURL_syslog(LOG_DEBUG, + "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl); OPT(URL, IO->ConnectMe->PlainUrl); if (StrLength(IO->ConnectMe->CurlCreds)) { OPT(HTTPAUTH, (long)CURLAUTH_BASIC); OPT(USERPWD, ChrPtr(IO->ConnectMe->CurlCreds)); } -#ifdef CURLOPT_HTTP_CONTENT_DECODING - OPT(HTTP_CONTENT_DECODING, 1); - OPT(ENCODING, ""); -#endif if (StrLength(IO->HttpReq.PostData) > 0) - { + { OPT(POSTFIELDS, ChrPtr(IO->HttpReq.PostData)); OPT(POSTFIELDSIZE, StrLength(IO->HttpReq.PostData)); } - else if ((IO->HttpReq.PlainPostDataLen != 0) && (IO->HttpReq.PlainPostData != NULL)) + else if ((IO->HttpReq.PlainPostDataLen != 0) && + (IO->HttpReq.PlainPostData != NULL)) { OPT(POSTFIELDS, IO->HttpReq.PlainPostData); OPT(POSTFIELDSIZE, IO->HttpReq.PlainPostDataLen); } + OPT(HTTPHEADER, IO->HttpReq.headers); - if (IO->HttpReq.headers != NULL) - OPT(HTTPHEADER, IO->HttpReq.headers); - - return 1; -} - -void -evcurl_handle_start(AsyncIO *IO) -{ - CURLMcode msta; - - CtdlLogPrintf(CTDL_DEBUG, "EVCURL: attaching to curl multi handle\n"); + IO->NextState = eConnect; + EVCURLM_syslog(LOG_DEBUG, "EVCURL: attaching to curl multi handle\n"); msta = curl_multi_add_handle(global.mhnd, IO->HttpReq.chnd); if (msta) - CtdlLogPrintf(CTDL_ERR, "EVCURL: error attaching to curl multi handle: %s\n", curl_multi_strerror(msta)); + { + EVCURL_syslog(LOG_ERR, + "EVCURL: error attaching to curl multi handle: %s\n", + curl_multi_strerror(msta)); + } + IO->HttpReq.attached = 1; ev_async_send (event_base, &WakeupCurl); + ev_cleanup_init(&IO->abort_by_shutdown, + IOcurl_abort_shutdown_callback); + + ev_cleanup_start(event_base, &IO->abort_by_shutdown); + return eReadMessage; } static void WakeupCurlCallback(EV_P_ ev_async *w, int revents) { - CtdlLogPrintf(CTDL_DEBUG, "EVCURL: waking up curl multi handle\n"); + CURLM_syslog(LOG_DEBUG, "waking up curl multi handle\n"); curl_multi_perform(&global, CURL_POLL_NONE); } static void evcurl_shutdown (void) { + curl_global_cleanup(); curl_multi_cleanup(global.mhnd); + CURLM_syslog(LOG_DEBUG, "exiting\n"); } /***************************************************************************** * libevent integration * @@ -383,24 +577,25 @@ static void evcurl_shutdown (void) */ int evbase_count = 0; int event_add_pipe[2] = {-1, -1}; -citthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */ +pthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */ HashList *QueueEvents = NULL; HashList *InboundEventQueue = NULL; HashList *InboundEventQueues[2] = { NULL, NULL }; -ev_async AddJob; +ev_async AddJob; ev_async ExitEventLoop; static void QueueEventAddCallback(EV_P_ ev_async *w, int revents) { + ev_tstamp Now; HashList *q; void *v; - HashPos *It; + HashPos*It; long len; const char *Key; /* get the control command... */ - citthread_mutex_lock(&EventQueueMutex); + pthread_mutex_lock(&EventQueueMutex); if (InboundEventQueues[0] == InboundEventQueue) { InboundEventQueue = InboundEventQueues[1]; @@ -410,17 +605,23 @@ static void QueueEventAddCallback(EV_P_ ev_async *w, int revents) InboundEventQueue = InboundEventQueues[0]; q = InboundEventQueues[1]; } - citthread_mutex_unlock(&EventQueueMutex); - + pthread_mutex_unlock(&EventQueueMutex); + Now = ev_now (event_base); It = GetNewHashPos(q, 0); while (GetNextHashPos(q, It, &len, &Key, &v)) { IOAddHandler *h = v; + if (h->IO->ID == 0) { + h->IO->ID = EvIDSource++; + } + if (h->IO->StartIO == 0.0) + h->IO->StartIO = Now; + h->IO->Now = Now; h->EvAttch(h->IO); } DeleteHashPos(&It); DeleteHashContent(&q); - CtdlLogPrintf(CTDL_DEBUG, "EVENT Q Read done.\n"); + EVQM_syslog(LOG_DEBUG, "EVENT Q Add done.\n"); } @@ -428,7 +629,7 @@ static void EventExitCallback(EV_P_ ev_async *w, int revents) { ev_break(event_base, EVBREAK_ALL); - CtdlLogPrintf(CTDL_DEBUG, "EVENT Q exiting.\n"); + EVQM_syslog(LOG_DEBUG, "EVENT Q exiting.\n"); } @@ -437,10 +638,12 @@ void InitEventQueue(void) { struct rlimit LimitSet; - citthread_mutex_init(&EventQueueMutex, NULL); + pthread_mutex_init(&EventQueueMutex, NULL); if (pipe(event_add_pipe) != 0) { - CtdlLogPrintf(CTDL_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno)); + syslog(LOG_EMERG, + "Unable to create pipe for libev queueing: %s\n", + strerror(errno)); abort(); } LimitSet.rlim_cur = 1; @@ -452,10 +655,11 @@ void InitEventQueue(void) InboundEventQueues[1] = NewHash(1, Flathash); InboundEventQueue = InboundEventQueues[0]; } +extern void CtdlDestroyEVCleanupHooks(void); + +extern int EVQShutDown; /* * this thread operates the select() etc. via libev. - * - * */ void *client_event_thread(void *arg) { @@ -463,10 +667,9 @@ void *client_event_thread(void *arg) CtdlFillSystemContext(&libev_client_CC, "LibEv Thread"); // citthread_setspecific(MyConKey, (void *)&smtp_queue_CC); - CtdlLogPrintf(CTDL_DEBUG, "client_ev_thread() initializing\n"); + EVQM_syslog(LOG_DEBUG, "client_event_thread() initializing\n"); event_base = ev_default_loop (EVFLAG_AUTO); - ev_async_init(&AddJob, QueueEventAddCallback); ev_async_start(event_base, &AddJob); ev_async_init(&ExitEventLoop, EventExitCallback); @@ -478,20 +681,26 @@ void *client_event_thread(void *arg) ev_run (event_base, 0); + EVQM_syslog(LOG_DEBUG, "client_event_thread() exiting\n"); ///what todo here? CtdlClearSystemContext(); ev_loop_destroy (EV_DEFAULT_UC); - DeleteHash(&QueueEvents); InboundEventQueue = NULL; DeleteHash(&InboundEventQueues[0]); DeleteHash(&InboundEventQueues[1]); - citthread_mutex_destroy(&EventQueueMutex); +/* citthread_mutex_destroy(&EventQueueMutex); TODO */ evcurl_shutdown(); + close(event_add_pipe[0]); + close(event_add_pipe[1]); + + CtdlDestroyEVCleanupHooks(); + EVQShutDown = 1; return(NULL); } -/*------------------------------------------------------------------------------*/ + +/*----------------------------------------------------------------------------*/ /* * DB-Queue; does async bdb operations. * has its own set of handlers. @@ -499,26 +708,27 @@ void *client_event_thread(void *arg) ev_loop *event_db; int evdb_count = 0; int evdb_add_pipe[2] = {-1, -1}; -citthread_mutex_t DBEventQueueMutex; /* locks the access to the following vars: */ +pthread_mutex_t DBEventQueueMutex; /* locks the access to the following vars: */ HashList *DBQueueEvents = NULL; HashList *DBInboundEventQueue = NULL; HashList *DBInboundEventQueues[2] = { NULL, NULL }; -ev_async DBAddJob; +ev_async DBAddJob; ev_async DBExitEventLoop; extern void ShutDownDBCLient(AsyncIO *IO); static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents) { + ev_tstamp Now; HashList *q; void *v; - HashPos *It; + HashPos *It; long len; const char *Key; /* get the control command... */ - citthread_mutex_lock(&DBEventQueueMutex); + pthread_mutex_lock(&DBEventQueueMutex); if (DBInboundEventQueues[0] == DBInboundEventQueue) { DBInboundEventQueue = DBInboundEventQueues[1]; @@ -528,31 +738,37 @@ static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents) DBInboundEventQueue = DBInboundEventQueues[0]; q = DBInboundEventQueues[1]; } - citthread_mutex_unlock(&DBEventQueueMutex); + pthread_mutex_unlock(&DBEventQueueMutex); + Now = ev_now (event_db); It = GetNewHashPos(q, 0); while (GetNextHashPos(q, It, &len, &Key, &v)) { IOAddHandler *h = v; eNextState rc; + if (h->IO->ID == 0) + h->IO->ID = EvIDSource++; + if (h->IO->StartDB == 0.0) + h->IO->StartDB = Now; + h->IO->Now = Now; rc = h->EvAttch(h->IO); switch (rc) { case eAbort: - ShutDownDBCLient(h->IO); + ShutDownDBCLient(h->IO); default: - break; + break; } } DeleteHashPos(&It); DeleteHashContent(&q); - CtdlLogPrintf(CTDL_DEBUG, "DBEVENT Q Read done.\n"); + EVQM_syslog(LOG_DEBUG, "DBEVENT Q Add done.\n"); } static void DBEventExitCallback(EV_P_ ev_async *w, int revents) { - CtdlLogPrintf(CTDL_DEBUG, "EVENT Q exiting.\n"); + EVQM_syslog(LOG_DEBUG, "DB EVENT Q exiting.\n"); ev_break(event_db, EVBREAK_ALL); } @@ -562,10 +778,10 @@ void DBInitEventQueue(void) { struct rlimit LimitSet; - citthread_mutex_init(&DBEventQueueMutex, NULL); + pthread_mutex_init(&DBEventQueueMutex, NULL); if (pipe(evdb_add_pipe) != 0) { - CtdlLogPrintf(CTDL_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno)); + EVQ_syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno)); abort(); } LimitSet.rlim_cur = 1; @@ -580,16 +796,14 @@ void DBInitEventQueue(void) /* * this thread operates writing to the message database via libev. - * - * */ -void *db_event_thread(void *arg) +void *db_event_thread(void *arg) { struct CitContext libev_msg_CC; CtdlFillSystemContext(&libev_msg_CC, "LibEv DB IO Thread"); // citthread_setspecific(MyConKey, (void *)&smtp_queue_CC); - CtdlLogPrintf(CTDL_DEBUG, "client_msgev_thread() initializing\n"); + EVQM_syslog(LOG_DEBUG, "dbevent_thread() initializing\n"); event_db = ev_loop_new (EVFLAG_AUTO); @@ -600,6 +814,7 @@ void *db_event_thread(void *arg) ev_run (event_db, 0); + EVQM_syslog(LOG_DEBUG, "dbevent_thread() exiting\n"); //// what to do here? CtdlClearSystemContext(); ev_loop_destroy (event_db); @@ -608,24 +823,52 @@ void *db_event_thread(void *arg) DBInboundEventQueue = NULL; DeleteHash(&DBInboundEventQueues[0]); DeleteHash(&DBInboundEventQueues[1]); - citthread_mutex_destroy(&DBEventQueueMutex); + + close(evdb_add_pipe[0]); + close(evdb_add_pipe[1]); +/* citthread_mutex_destroy(&DBEventQueueMutex); TODO */ return(NULL); } -#endif +void ShutDownEventQueues(void) +{ + EVQM_syslog(LOG_DEBUG, "EVENT Qs triggering exits.\n"); + + pthread_mutex_lock(&DBEventQueueMutex); + ev_async_send (event_db, &DBExitEventLoop); + pthread_mutex_unlock(&DBEventQueueMutex); + + pthread_mutex_lock(&EventQueueMutex); + ev_async_send (EV_DEFAULT_ &ExitEventLoop); + pthread_mutex_unlock(&EventQueueMutex); +} + +void DebugEventloopEnable(const int n) +{ + DebugEventLoop = n; +} +void DebugEventloopBacktraceEnable(const int n) +{ + DebugEventLoopBacktrace = n; +} + +void DebugCurlEnable(const int n) +{ + DebugCurl = n; +} CTDL_MODULE_INIT(event_client) { -#ifdef EXPERIMENTAL_SMTP_EVENT_CLIENT if (!threading) { + CtdlRegisterDebugFlagHook(HKEY("eventloop"), DebugEventloopEnable, &DebugEventLoop); + CtdlRegisterDebugFlagHook(HKEY("eventloopbacktrace"), DebugEventloopBacktraceEnable, &DebugEventLoopBacktrace); + CtdlRegisterDebugFlagHook(HKEY("curl"), DebugCurlEnable, &DebugCurl); InitEventQueue(); DBInitEventQueue(); - CtdlThreadCreate("Client event", CTDLTHREAD_BIGSTACK, client_event_thread, NULL); - CtdlThreadCreate("DB event", CTDLTHREAD_BIGSTACK, db_event_thread, NULL); -/// todo register shutdown callback. + CtdlThreadCreate(client_event_thread); + CtdlThreadCreate(db_event_thread); } -#endif return "event"; }