Indents are 8 characters wide and are expressed as a tab character.
[citadel.git] / citadel / modules / eventclient / serv_eventclient.c
index 97e217de9dd43c6ca3b9e171a47970b36dacaa5a..520ae7fc8704270ac7149fbaad31d6db1f4665e6 100644 (file)
@@ -1,19 +1,13 @@
 /*
- * Copyright (c) 1998-2012 by the citadel.org team
+ * Copyright (c) 1998-2015 by the citadel.org team
  *
- *  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 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.
- *
- *  
- *  
- *  
+ * 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.
  */
 
 #include "sysdep.h"
 #include "server.h"
 #include "citserver.h"
 #include "support.h"
-
 #include "ctdl_module.h"
-
+#include "config.h"
 #include "event_client.h"
 #include "serv_curl.h"
 
 ev_loop *event_base;
 int DebugEventLoop = 0;
+int DebugEventLoopBacktrace = 0;
 int DebugCurl = 0;
+pthread_key_t evConKey;
 
 long EvIDSource = 1;
 /*****************************************************************************
  *                   libevent / curl integration                             *
  *****************************************************************************/
-#define MOPT(s, v)\
+
+#define MOPT(s, v)                                                     \
        do {                                                            \
                sta = curl_multi_setopt(mhnd, (CURLMOPT_##s), (v));     \
                if (sta) {                                              \
-                       syslog(LOG_ERR, "EVCURL: error setting option " \
+                       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;
@@ -88,6 +85,8 @@ typedef struct _evcurl_global_data {
 ev_async WakeupCurl;
 evcurl_global_data global;
 
+eNextState QueueAnDBOperation(AsyncIO *IO);
+
 static void
 gotstatus(int nnrun)
 {
@@ -97,15 +96,15 @@ gotstatus(int nnrun)
        global.nrun = nnrun;
 
        syslog(LOG_DEBUG,
-              "CURLEV: gotstatus(): about to call curl_multi_info_read\n");
+                    "gotstatus(): about to call curl_multi_info_read\n");
        while ((msg = curl_multi_info_read(global.mhnd, &nmsg))) {
-               syslog(LOG_ERR,
-                      "EVCURL: got curl multi_info message msg=%d\n",
-                      msg->msg);
+               syslog(LOG_DEBUG,
+                           "got curl multi_info message msg=%d\n",
+                           msg->msg);
 
                if (CURLMSG_DONE == msg->msg) {
                        CURL *chnd;
-                       char *chandle;
+                       void *chandle = NULL;
                        CURLcode sta;
                        CURLMcode msta;
                        AsyncIO*IO;
@@ -115,49 +114,66 @@ gotstatus(int nnrun)
                        sta = curl_easy_getinfo(chnd,
                                                CURLINFO_PRIVATE,
                                                &chandle);
-                       syslog(LOG_ERR, "EVCURL: request complete\n");
-                       if (sta)
+                       if (sta) {
                                syslog(LOG_ERR,
-                                      "EVCURL: error asking curl for private"
+                                      "error asking curl for private"
                                       " cookie of curl handle: %s\n",
                                       curl_easy_strerror(sta));
+                               continue;
+                       }
                        IO = (AsyncIO *)chandle;
+                       if (IO->ID == 0) {
+                               syslog(LOG_ERR,
+                                             "Error, invalid IO context %p\n",
+                                             IO);
+                               continue;
+                       }
+                       SetEVState(IO, eCurlGotStatus);
 
-                       IO->Now = ev_now(event_base);
+                       syslog(LOG_DEBUG, "request complete\n");
+
+                       IO->CitContext->lastcmd = 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) {
-                               EV_syslog(LOG_ERR,
-                                         "EVCURL: error description: %s\n",
-                                         IO->HttpReq.errdesc);
-                               EV_syslog(LOG_ERR,
-                                         "EVCURL: error performing request: %s\n",
-                                         curl_easy_strerror(sta));
+                               syslog(LOG_ERR,
+                                             "error description: %s\n",
+                                             IO->HttpReq.errdesc);
+                               IO->HttpReq.CurlError = curl_easy_strerror(sta);
+                               syslog(LOG_ERR,
+                                             "error performing request: %s\n",
+                                             IO->HttpReq.CurlError);
+                               if (sta == CURLE_OPERATION_TIMEDOUT)
+                               {
+                                       IO->SendBuf.fd = 0;
+                                       IO->RecvBuf.fd = 0;
+                               }
                        }
                        sta = curl_easy_getinfo(chnd,
                                                CURLINFO_RESPONSE_CODE,
                                                &IO->HttpReq.httpcode);
                        if (sta)
-                               EV_syslog(LOG_ERR,
-                                         "EVCURL: error asking curl for "
-                                         "response code from request: %s\n",
-                                         curl_easy_strerror(sta));
-                       EV_syslog(LOG_ERR,
-                                 "EVCURL: http response code was %ld\n",
-                                 (long)IO->HttpReq.httpcode);
+                               syslog(LOG_ERR,
+                                             "error asking curl for "
+                                             "response code from request: %s\n",
+                                             curl_easy_strerror(sta));
+                       syslog(LOG_DEBUG,
+                                     "http response code was %ld\n",
+                                     (long)IO->HttpReq.httpcode);
 
 
                        curl_slist_free_all(IO->HttpReq.headers);
+                       IO->HttpReq.headers = NULL;
                        msta = curl_multi_remove_handle(global.mhnd, chnd);
                        if (msta)
-                               EV_syslog(LOG_ERR,
-                                         "EVCURL: warning problem detaching "
-                                         "completed handle from curl multi: "
-                                         "%s\n",
-                                         curl_multi_strerror(msta));
+                               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);
 
@@ -165,8 +181,8 @@ gotstatus(int nnrun)
                        switch(IO->SendDone(IO))
                        {
                        case eDBQuery:
-                               curl_easy_cleanup(IO->HttpReq.chnd);
-                               IO->HttpReq.chnd = NULL;
+                               FreeURL(&IO->ConnectMe);
+                               QueueAnDBOperation(IO);
                                break;
                        case eSendDNSQuery:
                        case eReadDNSReply:
@@ -178,8 +194,6 @@ gotstatus(int nnrun)
                        case eReadMore:
                        case eReadPayload:
                        case eReadFile:
-                               curl_easy_cleanup(IO->HttpReq.chnd);
-                               IO->HttpReq.chnd = NULL;
                                break;
                        case eTerminateConnection:
                        case eAbort:
@@ -205,13 +219,13 @@ stepmulti(void *data, curl_socket_t fd, int which)
                                        which,
                                        &running_handles);
 
-       syslog(LOG_DEBUG, "EVCURL: stepmulti(): calling gotstatus()\n");
+       syslog(LOG_DEBUG, "stepmulti(): calling gotstatus()\n");
        if (msta)
                syslog(LOG_ERR,
-                      "EVCURL: error in curl processing events"
-                      "on multi handle, fd %d: %s\n",
-                      (int)fd,
-                      curl_multi_strerror(msta));
+                           "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);
@@ -228,8 +242,8 @@ static void
 got_in(struct ev_loop *loop, ev_io *ioev, int events)
 {
        syslog(LOG_DEBUG,
-              "EVCURL: waking up curl for io on fd %d\n",
-              (int)ioev->fd);
+                   "EVCURL: waking up curl for io on fd %d\n",
+                   (int)ioev->fd);
 
        stepmulti(ioev->data, ioev->fd, CURL_CSELECT_IN);
 }
@@ -238,21 +252,23 @@ static void
 got_out(struct ev_loop *loop, ev_io *ioev, int events)
 {
        syslog(LOG_DEBUG,
-              "EVCURL: waking up curl for io on fd %d\n",
-              (int)ioev->fd);
+                   "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) {
+gotdata(void *data, size_t size, size_t nmemb, void *cglobal)
+{
        AsyncIO *IO = (AsyncIO*) cglobal;
 
+       SetEVState(IO, eCurlGotData);
        if (IO->HttpReq.ReplyData == NULL)
        {
                IO->HttpReq.ReplyData = NewStrBufPlain(NULL, SIZ);
        }
-       IO->Now = ev_now(event_base);
+       IO->CitContext->lastcmd = IO->Now = ev_now(event_base);
        return CurlFillStrBuf_callback(data,
                                       size,
                                       nmemb,
@@ -281,7 +297,7 @@ gotwatchsock(CURL *easy,
 {
        evcurl_global_data *global = cglobal;
        CURLM *mhnd = global->mhnd;
-       char *f;
+       void *f;
        AsyncIO *IO = (AsyncIO*) vIO;
        CURLcode sta;
        const char *Action;
@@ -289,16 +305,17 @@ gotwatchsock(CURL *easy,
        if (IO == NULL) {
                sta = curl_easy_getinfo(easy, CURLINFO_PRIVATE, &f);
                if (sta) {
-                       EV_syslog(LOG_ERR,
-                                 "EVCURL: error asking curl for private "
-                                 "cookie of curl handle: %s\n",
-                                 curl_easy_strerror(sta));
+                       syslog(LOG_ERR,
+                                   "EVCURL: error asking curl for private "
+                                   "cookie of curl handle: %s\n",
+                                   curl_easy_strerror(sta));
                        return -1;
                }
                IO = (AsyncIO *) f;
-               EV_syslog(LOG_DEBUG,
-                         "EVCURL: got socket for URL: %s\n",
-                         IO->ConnectMe->PlainUrl);
+               SetEVState(IO, eCurlNewIO);
+               syslog(LOG_DEBUG,
+                             "EVCURL: got socket for URL: %s\n",
+                             IO->ConnectMe->PlainUrl);
 
                if (IO->SendBuf.fd != 0)
                {
@@ -311,7 +328,8 @@ gotwatchsock(CURL *easy,
                curl_multi_assign(mhnd, fd, IO);
        }
 
-       IO->Now = ev_now(event_base);
+       SetEVState(IO, eCurlGotIO);
+       IO->CitContext->lastcmd = IO->Now = ev_now(event_base);
 
        Action = "";
        switch (action)
@@ -334,21 +352,21 @@ gotwatchsock(CURL *easy,
        }
 
 
-       EV_syslog(LOG_DEBUG,
-                 "EVCURL: gotwatchsock called fd=%d action=%s[%d]\n",
-                 (int)fd, Action, action);
+       syslog(LOG_DEBUG,
+                     "EVCURL: gotwatchsock called fd=%d action=%s[%d]\n",
+                     (int)fd, Action, action);
 
        switch (action)
        {
        case CURL_POLL_NONE:
-               EVM_syslog(LOG_ERR,
-                          "EVCURL: called first time "
-                          "to register this sockwatcker\n");
+               syslog(LOG_DEBUG,
+                              "called first time "
+                              "to register this sockwatcker\n");
                break;
        case CURL_POLL_REMOVE:
-               EVM_syslog(LOG_ERR,
-                          "EVCURL: called last time to unregister "
-                          "this sockwatcher\n");
+               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;
@@ -380,8 +398,8 @@ void curl_init_connectionpool(void)
        if (sta)
        {
                syslog(LOG_ERR,
-                      "EVCURL: error initializing curl library: %s\n",
-                      curl_easy_strerror(sta));
+                           "error initializing curl library: %s\n",
+                           curl_easy_strerror(sta));
 
                exit(1);
        }
@@ -389,7 +407,7 @@ void curl_init_connectionpool(void)
        if (!mhnd)
        {
                syslog(LOG_ERR,
-                      "EVCURL: error initializing curl multi handle\n");
+                            "error initializing curl multi handle\n");
                exit(3);
        }
 
@@ -406,12 +424,12 @@ int evcurl_init(AsyncIO *IO)
        CURLcode sta;
        CURL *chnd;
 
-       EVM_syslog(LOG_DEBUG, "EVCURL: evcurl_init called ms\n");
+       syslog(LOG_DEBUG, "EVCURL: evcurl_init called ms\n");
        IO->HttpReq.attached = 0;
        chnd = IO->HttpReq.chnd = curl_easy_init();
        if (!chnd)
        {
-               EVM_syslog(LOG_ERR, "EVCURL: error initializing curl handle\n");
+               syslog(LOG_ERR, "EVCURL: error initializing curl handle\n");
                return 0;
        }
 
@@ -438,13 +456,13 @@ int evcurl_init(AsyncIO *IO)
        OPT(WRITEDATA, (void *)IO);
        OPT(ERRORBUFFER, IO->HttpReq.errdesc);
 
-       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"))
+       if ((!IsEmptyStr(CtdlGetConfigStr("c_ip_addr")))
+               && (strcmp(CtdlGetConfigStr("c_ip_addr"), "*"))
+               && (strcmp(CtdlGetConfigStr("c_ip_addr"), "::"))
+               && (strcmp(CtdlGetConfigStr("c_ip_addr"), "0.0.0.0"))
                )
        {
-               OPT(INTERFACE, config.c_ip_addr);
+               OPT(INTERFACE, CtdlGetConfigStr("c_ip_addr"));
        }
 
 #ifdef CURLOPT_HTTP_CONTENT_DECODING
@@ -465,17 +483,23 @@ static void IOcurl_abort_shutdown_callback(struct ev_loop *loop,
 {
        CURLMcode msta;
        AsyncIO *IO = watcher->data;
-       IO->Now = ev_now(event_base);
-       EV_syslog(LOG_DEBUG, "EVENT Curl: %s\n", __FUNCTION__);
+
+       if (IO == NULL)
+               return;
+
+       SetEVState(IO, eCurlShutdown);
+       IO->CitContext->lastcmd = IO->Now = ev_now(event_base);
+       syslog(LOG_DEBUG, "EVENT Curl: %s\n", __FUNCTION__);
 
        curl_slist_free_all(IO->HttpReq.headers);
+       IO->HttpReq.headers = NULL;
        msta = curl_multi_remove_handle(global.mhnd, IO->HttpReq.chnd);
        if (msta)
        {
-               EV_syslog(LOG_ERR,
-                         "EVCURL: warning problem detaching completed handle "
-                         "from curl multi: %s\n",
-                         curl_multi_strerror(msta));
+               syslog(LOG_ERR,
+                             "EVCURL: warning problem detaching completed handle "
+                             "from curl multi: %s\n",
+                             curl_multi_strerror(msta));
        }
 
        curl_easy_cleanup(IO->HttpReq.chnd);
@@ -486,6 +510,7 @@ static void IOcurl_abort_shutdown_callback(struct ev_loop *loop,
        assert(IO->ShutdownAbort);
        IO->ShutdownAbort(IO);
 }
+
 eNextState
 evcurl_handle_start(AsyncIO *IO)
 {
@@ -493,8 +518,9 @@ evcurl_handle_start(AsyncIO *IO)
        CURLcode sta;
        CURL *chnd;
 
+       SetEVState(IO, eCurlStart);
        chnd = IO->HttpReq.chnd;
-       EV_syslog(LOG_DEBUG,
+       syslog(LOG_DEBUG,
                  "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl);
        OPT(URL, IO->ConnectMe->PlainUrl);
        if (StrLength(IO->ConnectMe->CurlCreds))
@@ -517,27 +543,29 @@ evcurl_handle_start(AsyncIO *IO)
        OPT(HTTPHEADER, IO->HttpReq.headers);
 
        IO->NextState = eConnect;
-       EVM_syslog(LOG_DEBUG, "EVCURL: attaching to curl multi handle\n");
+       syslog(LOG_DEBUG, "EVCURL: attaching to curl multi handle\n");
        msta = curl_multi_add_handle(global.mhnd, IO->HttpReq.chnd);
        if (msta)
        {
-               EV_syslog(LOG_ERR,
+               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)
 {
-       syslog(LOG_DEBUG, "EVCURL: waking up curl multi handle\n");
+       syslog(LOG_DEBUG, "waking up curl multi handle\n");
 
        curl_multi_perform(&global, CURL_POLL_NONE);
 }
@@ -546,7 +574,7 @@ static void evcurl_shutdown (void)
 {
        curl_global_cleanup();
        curl_multi_cleanup(global.mhnd);
-       syslog(LOG_DEBUG, "client_event_thread() initializing\n");
+       syslog(LOG_DEBUG, "exiting\n");
 }
 /*****************************************************************************
  *                       libevent integration                                *
@@ -556,17 +584,21 @@ static void evcurl_shutdown (void)
  * this currently is the main loop (which may change in some future?)
  */
 int evbase_count = 0;
-int event_add_pipe[2] = {-1, -1};
 pthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */
+pthread_mutex_t EventExitQueueMutex; /* locks the access to the event queue pointer on exit. */
 HashList *QueueEvents = NULL;
 HashList *InboundEventQueue = NULL;
 HashList *InboundEventQueues[2] = { NULL, NULL };
+extern void ShutDownCLient(AsyncIO *IO);
 
 ev_async AddJob;
 ev_async ExitEventLoop;
 
 static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
 {
+       CitContext *Ctx;
+       long IOID = -1;
+       long count = 0;
        ev_tstamp Now;
        HashList *q;
        void *v;
@@ -591,17 +623,43 @@ static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
        while (GetNextHashPos(q, It, &len, &Key, &v))
        {
                IOAddHandler *h = v;
+               count ++;
                if (h->IO->ID == 0) {
                        h->IO->ID = EvIDSource++;
                }
+               IOID = h->IO->ID;
                if (h->IO->StartIO == 0.0)
                        h->IO->StartIO = Now;
-               h->IO->Now = Now;
-               h->EvAttch(h->IO);
+
+               SetEVState(h->IO, eIOAttach);
+
+               Ctx = h->IO->CitContext;
+               become_session(Ctx);
+
+               h->IO->CitContext->lastcmd = h->IO->Now = Now;
+               switch (h->EvAttch(h->IO))
+               {
+               case eReadMore:
+               case eReadMessage:
+               case eReadFile:
+               case eSendReply:
+               case eSendMore:
+               case eReadPayload:
+               case eSendFile:
+               case eDBQuery:
+               case eSendDNSQuery:
+               case eReadDNSReply:
+               case eConnect:
+                       break;
+               case eTerminateConnection:
+               case eAbort:
+                       ShutDownCLient(h->IO);
+               break;
+               }
        }
        DeleteHashPos(&It);
        DeleteHashContent(&q);
-       syslog(LOG_DEBUG, "EVENT Q Add done.\n");
+       syslog(LOG_DEBUG, "%s CC[%ld] EVENT Q Add %ld  done.", IOSTR, IOID, count);
 }
 
 
@@ -616,19 +674,8 @@ static void EventExitCallback(EV_P_ ev_async *w, int revents)
 
 void InitEventQueue(void)
 {
-       struct rlimit LimitSet;
-
        pthread_mutex_init(&EventQueueMutex, NULL);
-
-       if (pipe(event_add_pipe) != 0) {
-               syslog(LOG_EMERG,
-                      "Unable to create pipe for libev queueing: %s\n",
-                      strerror(errno));
-               abort();
-       }
-       LimitSet.rlim_cur = 1;
-       LimitSet.rlim_max = 1;
-       setrlimit(event_add_pipe[1], &LimitSet);
+       pthread_mutex_init(&EventExitQueueMutex, NULL);
 
        QueueEvents = NewHash(1, Flathash);
        InboundEventQueues[0] = NewHash(1, Flathash);
@@ -638,6 +685,7 @@ void InitEventQueue(void)
 extern void CtdlDestroyEVCleanupHooks(void);
 
 extern int EVQShutDown;
+const char *IOLog = "IO";
 /*
  * this thread operates the select() etc. via libev.
  */
@@ -646,7 +694,9 @@ void *client_event_thread(void *arg)
        struct CitContext libev_client_CC;
 
        CtdlFillSystemContext(&libev_client_CC, "LibEv Thread");
-//     citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
+
+       pthread_setspecific(evConKey, IOLog);
+
        syslog(LOG_DEBUG, "client_event_thread() initializing\n");
 
        event_base = ev_default_loop (EVFLAG_AUTO);
@@ -664,19 +714,20 @@ void *client_event_thread(void *arg)
        syslog(LOG_DEBUG, "client_event_thread() exiting\n");
 
 ///what todo here?     CtdlClearSystemContext();
+       pthread_mutex_lock(&EventExitQueueMutex);
        ev_loop_destroy (EV_DEFAULT_UC);
+       event_base = NULL;
        DeleteHash(&QueueEvents);
        InboundEventQueue = NULL;
        DeleteHash(&InboundEventQueues[0]);
        DeleteHash(&InboundEventQueues[1]);
 /*     citthread_mutex_destroy(&EventQueueMutex); TODO */
        evcurl_shutdown();
-       close(event_add_pipe[0]);
-       close(event_add_pipe[1]);
 
        CtdlDestroyEVCleanupHooks();
 
-       EVQShutDown = 1;        
+       pthread_mutex_unlock(&EventExitQueueMutex);
+       EVQShutDown = 1;
        return(NULL);
 }
 
@@ -687,8 +738,8 @@ void *client_event_thread(void *arg)
  */
 ev_loop *event_db;
 int evdb_count = 0;
-int evdb_add_pipe[2] = {-1, -1};
 pthread_mutex_t DBEventQueueMutex; /* locks the access to the following vars: */
+pthread_mutex_t DBEventExitQueueMutex; /* locks the access to the db-event queue pointer on exit. */
 HashList *DBQueueEvents = NULL;
 HashList *DBInboundEventQueue = NULL;
 HashList *DBInboundEventQueues[2] = { NULL, NULL };
@@ -700,6 +751,9 @@ extern void ShutDownDBCLient(AsyncIO *IO);
 
 static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents)
 {
+       CitContext *Ctx;
+       long IOID = -1;
+       long count = 0;;
        ev_tstamp Now;
        HashList *q;
        void *v;
@@ -726,11 +780,18 @@ static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents)
        {
                IOAddHandler *h = v;
                eNextState rc;
+               count ++;
                if (h->IO->ID == 0)
                        h->IO->ID = EvIDSource++;
+               IOID = h->IO->ID;
                if (h->IO->StartDB == 0.0)
                        h->IO->StartDB = Now;
-               h->IO->Now = Now;
+               h->IO->CitContext->lastcmd = h->IO->Now = Now;
+
+               SetEVState(h->IO, eDBAttach);
+               Ctx = h->IO->CitContext;
+               become_session(Ctx);
+               ev_cleanup_start(event_db, &h->IO->db_abort_by_shutdown);
                rc = h->EvAttch(h->IO);
                switch (rc)
                {
@@ -742,7 +803,7 @@ static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents)
        }
        DeleteHashPos(&It);
        DeleteHashContent(&q);
-       syslog(LOG_DEBUG, "DBEVENT Q Add done.\n");
+       syslog(LOG_DEBUG, "%s CC[%ld] DBEVENT Q Add %ld done.", IOSTR, IOID, count);
 }
 
 
@@ -756,17 +817,8 @@ static void DBEventExitCallback(EV_P_ ev_async *w, int revents)
 
 void DBInitEventQueue(void)
 {
-       struct rlimit LimitSet;
-
        pthread_mutex_init(&DBEventQueueMutex, NULL);
-
-       if (pipe(evdb_add_pipe) != 0) {
-               syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
-               abort();
-       }
-       LimitSet.rlim_cur = 1;
-       LimitSet.rlim_max = 1;
-       setrlimit(evdb_add_pipe[1], &LimitSet);
+       pthread_mutex_init(&DBEventExitQueueMutex, NULL);
 
        DBQueueEvents = NewHash(1, Flathash);
        DBInboundEventQueues[0] = NewHash(1, Flathash);
@@ -774,18 +826,23 @@ void DBInitEventQueue(void)
        DBInboundEventQueue = DBInboundEventQueues[0];
 }
 
+const char *DBLog = "BD";
+
 /*
  * this thread operates writing to the message database via libev.
  */
 void *db_event_thread(void *arg)
 {
+       ev_loop *tmp;
        struct CitContext libev_msg_CC;
 
+       pthread_setspecific(evConKey, DBLog);
+
        CtdlFillSystemContext(&libev_msg_CC, "LibEv DB IO Thread");
-//     citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
+
        syslog(LOG_DEBUG, "dbevent_thread() initializing\n");
 
-       event_db = ev_loop_new (EVFLAG_AUTO);
+       tmp = event_db = ev_loop_new (EVFLAG_AUTO);
 
        ev_async_init(&DBAddJob, DBQueueEventAddCallback);
        ev_async_start(event_db, &DBAddJob);
@@ -794,20 +851,20 @@ void *db_event_thread(void *arg)
 
        ev_run (event_db, 0);
 
-       syslog(LOG_DEBUG, "dbevent_thread() exiting\n");
+       pthread_mutex_lock(&DBEventExitQueueMutex);
 
-//// what to do here?  CtdlClearSystemContext();
-       ev_loop_destroy (event_db);
+       event_db = NULL;
+       syslog(LOG_INFO, "dbevent_thread() exiting\n");
 
        DeleteHash(&DBQueueEvents);
        DBInboundEventQueue = NULL;
        DeleteHash(&DBInboundEventQueues[0]);
        DeleteHash(&DBInboundEventQueues[1]);
 
-       close(evdb_add_pipe[0]);
-       close(evdb_add_pipe[1]);
 /*     citthread_mutex_destroy(&DBEventQueueMutex); TODO */
 
+       ev_loop_destroy (tmp);
+       pthread_mutex_unlock(&DBEventExitQueueMutex);
        return(NULL);
 }
 
@@ -824,22 +881,33 @@ void ShutDownEventQueues(void)
        pthread_mutex_unlock(&EventQueueMutex);
 }
 
-void DebugEventloopEnable(void)
+void DebugEventloopEnable(const int n)
 {
-       DebugEventLoop = 1;
+       DebugEventLoop = n;
+}
+void DebugEventloopBacktraceEnable(const int n)
+{
+       DebugEventLoopBacktrace = n;
 }
 
-void DebugCurlEnable(void)
+void DebugCurlEnable(const int n)
 {
-       DebugCurl = 1;
+       DebugCurl = n;
 }
 
+const char *WLog = "WX";
 CTDL_MODULE_INIT(event_client)
 {
        if (!threading)
        {
-               CtdlRegisterDebugFlagHook(HKEY("eventloop"), DebugEventloopEnable);
-               CtdlRegisterDebugFlagHook(HKEY("curl"), DebugCurlEnable);
+               if (pthread_key_create(&evConKey, NULL) != 0) {
+                       syslog(LOG_CRIT, "Can't create TSD key: %s", strerror(errno));
+               }
+               pthread_setspecific(evConKey, WLog);
+
+               CtdlRegisterDebugFlagHook(HKEY("eventloop"), DebugEventloopEnable, &DebugEventLoop);
+               CtdlRegisterDebugFlagHook(HKEY("eventloopbacktrace"), DebugEventloopBacktraceEnable, &DebugEventLoopBacktrace);
+               CtdlRegisterDebugFlagHook(HKEY("curl"), DebugCurlEnable, &DebugCurl);
                InitEventQueue();
                DBInitEventQueue();
                CtdlThreadCreate(client_event_thread);