EVENT: we memset the context, remember & re-set the ID so we can find the earlier...
[citadel.git] / citadel / event_client.c
index 5a32cea9bca45878d533fe68cb1cec7f68700147..90c5e5dd986619f4f9f5426bea3393133a9cb7c1 100644 (file)
@@ -1,20 +1,13 @@
 /*
- *
  * Copyright (c) 1998-2012 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 as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  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 open source software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License, version 3.
  *
- *  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
+ * 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"
@@ -47,6 +40,9 @@
 #include <netinet/in.h>
 #include <arpa/inet.h>
 #include <assert.h>
+#if HAVE_BACKTRACE
+#include <execinfo.h>
+#endif
 
 #include <libcitadel.h>
 #include "citadel.h"
 #include "citadel_dirs.h"
 
 #include "event_client.h"
+#include "ctdl_module.h"
 
+static void IO_Timeout_callback(struct ev_loop *loop, ev_timer *watcher, int revents);
 static void IO_abort_shutdown_callback(struct ev_loop *loop,
                                       ev_cleanup *watcher,
-                                      int revents)
-{
-       AsyncIO *IO = watcher->data;
-       EV_syslog(LOG_DEBUG, "EVENT Q: %s\n", __FUNCTION__);
-
-       assert(IO->ShutdownAbort);
-       IO->ShutdownAbort(IO);
-}
+                                      int revents);
 
 
 /*------------------------------------------------------------------------------
@@ -84,6 +75,7 @@ static void IO_abort_shutdown_callback(struct ev_loop *loop,
  *----------------------------------------------------------------------------*/
 extern int evdb_count;
 extern pthread_mutex_t DBEventQueueMutex;
+extern pthread_mutex_t DBEventExitQueueMutex;
 extern HashList *DBInboundEventQueue;
 extern struct ev_loop *event_db;
 extern ev_async DBAddJob;
@@ -100,36 +92,58 @@ eNextState QueueDBOperation(AsyncIO *IO, IO_CallBack CB)
        ev_cleanup_init(&IO->db_abort_by_shutdown,
                        IO_abort_shutdown_callback);
        IO->db_abort_by_shutdown.data = IO;
-       ev_cleanup_start(event_db, &IO->db_abort_by_shutdown);
 
        pthread_mutex_lock(&DBEventQueueMutex);
+       if (DBInboundEventQueue == NULL)
+       {
+               /* shutting down... */
+               free(h);
+               EVM_syslog(LOG_DEBUG, "DBEVENT Q exiting.\n");
+               pthread_mutex_unlock(&DBEventQueueMutex);
+               return eAbort;
+       }
        EVM_syslog(LOG_DEBUG, "DBEVENT Q\n");
        i = ++evdb_count ;
        Put(DBInboundEventQueue, IKEY(i), h, NULL);
        pthread_mutex_unlock(&DBEventQueueMutex);
 
+       pthread_mutex_lock(&DBEventExitQueueMutex);
+       if (event_db == NULL)
+       {
+               pthread_mutex_unlock(&DBEventExitQueueMutex);
+               return eAbort;
+       }
        ev_async_send (event_db, &DBAddJob);
+       pthread_mutex_unlock(&DBEventExitQueueMutex);
+
        EVM_syslog(LOG_DEBUG, "DBEVENT Q Done.\n");
        return eDBQuery;
 }
 
+void StopDBWatchers(AsyncIO *IO)
+{
+       ev_cleanup_stop(event_db, &IO->db_abort_by_shutdown);
+       ev_idle_stop(event_db, &IO->db_unwind_stack);
+}
+
 void ShutDownDBCLient(AsyncIO *IO)
 {
        CitContext *Ctx =IO->CitContext;
        become_session(Ctx);
 
        EVM_syslog(LOG_DEBUG, "DBEVENT Terminating.\n");
-       ev_cleanup_stop(event_db, &IO->db_abort_by_shutdown);
+       StopDBWatchers(IO);
 
-       assert(IO->Terminate);
-       IO->Terminate(IO);
+       assert(IO->DBTerminate);
+       IO->DBTerminate(IO);
 }
 
 void
 DB_PerformNext(struct ev_loop *loop, ev_idle *watcher, int revents)
 {
        AsyncIO *IO = watcher->data;
-       EV_syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__);
+       IO->Now = ev_now(event_db);
+       EV_syslog(LOG_DEBUG, "%s()", __FUNCTION__);
        become_session(IO->CitContext);
 
        ev_idle_stop(event_db, &IO->db_unwind_stack);
@@ -174,11 +188,23 @@ eNextState NextDBOperation(AsyncIO *IO, IO_CallBack CB)
  *----------------------------------------------------------------------------*/
 extern int evbase_count;
 extern pthread_mutex_t EventQueueMutex;
+extern pthread_mutex_t EventExitQueueMutex; 
 extern HashList *InboundEventQueue;
 extern struct ev_loop *event_base;
 extern ev_async AddJob;
 extern ev_async ExitEventLoop;
 
+static void IO_abort_shutdown_callback(struct ev_loop *loop,
+                                      ev_cleanup *watcher,
+                                      int revents)
+{
+       AsyncIO *IO = watcher->data;
+       EV_syslog(LOG_DEBUG, "EVENT Q: %s\n", __FUNCTION__);
+       IO->Now = ev_now(event_base);
+       assert(IO->ShutdownAbort);
+       IO->ShutdownAbort(IO);
+}
+
 
 eNextState QueueEventContext(AsyncIO *IO, IO_CallBack CB)
 {
@@ -191,15 +217,28 @@ eNextState QueueEventContext(AsyncIO *IO, IO_CallBack CB)
        ev_cleanup_init(&IO->abort_by_shutdown,
                        IO_abort_shutdown_callback);
        IO->abort_by_shutdown.data = IO;
-       ev_cleanup_start(event_base, &IO->abort_by_shutdown);
 
        pthread_mutex_lock(&EventQueueMutex);
+       if (InboundEventQueue == NULL)
+       {
+               free(h);
+               /* shutting down... */
+               EVM_syslog(LOG_DEBUG, "EVENT Q exiting.\n");
+               pthread_mutex_unlock(&EventQueueMutex);
+               return eAbort;
+       }
        EVM_syslog(LOG_DEBUG, "EVENT Q\n");
        i = ++evbase_count;
        Put(InboundEventQueue, IKEY(i), h, NULL);
        pthread_mutex_unlock(&EventQueueMutex);
 
+       pthread_mutex_lock(&EventExitQueueMutex);
+       if (event_base == NULL) {
+               pthread_mutex_unlock(&EventExitQueueMutex);
+               return eAbort;
+       }
        ev_async_send (event_base, &AddJob);
+       pthread_mutex_unlock(&EventExitQueueMutex);
        EVM_syslog(LOG_DEBUG, "EVENT Q Done.\n");
        return eSendReply;
 }
@@ -216,12 +255,28 @@ eNextState QueueCurlContext(AsyncIO *IO)
        h->EvAttch = evcurl_handle_start;
 
        pthread_mutex_lock(&EventQueueMutex);
+       if (InboundEventQueue == NULL)
+       {
+               /* shutting down... */
+               free(h);
+               EVM_syslog(LOG_DEBUG, "EVENT Q exiting.\n");
+               pthread_mutex_unlock(&EventQueueMutex);
+               return eAbort;
+       }
+
        EVM_syslog(LOG_DEBUG, "EVENT Q\n");
        i = ++evbase_count;
        Put(InboundEventQueue, IKEY(i), h, NULL);
        pthread_mutex_unlock(&EventQueueMutex);
 
+       pthread_mutex_lock(&EventExitQueueMutex);
+       if (event_base == NULL) {
+               pthread_mutex_unlock(&EventExitQueueMutex);
+               return eAbort;
+       }
        ev_async_send (event_base, &AddJob);
+       pthread_mutex_unlock(&EventExitQueueMutex);
+
        EVM_syslog(LOG_DEBUG, "EVENT Q Done.\n");
        return eSendReply;
 }
@@ -243,20 +298,43 @@ void FreeAsyncIOContents(AsyncIO *IO)
        if (Ctx) {
                Ctx->state = CON_IDLE;
                Ctx->kill_me = 1;
+               IO->CitContext = NULL;
        }
 }
 
 
-void StopClientWatchers(AsyncIO *IO)
+void StopClientWatchers(AsyncIO *IO, int CloseFD)
+{
+       ev_timer_stop (event_base, &IO->rw_timeout);
+       ev_timer_stop(event_base, &IO->conn_fail);
+       ev_idle_stop(event_base, &IO->unwind_stack);
+       ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
+
+       ev_io_stop(event_base, &IO->conn_event);
+       ev_io_stop(event_base, &IO->send_event);
+       ev_io_stop(event_base, &IO->recv_event);
+
+       if (CloseFD && (IO->SendBuf.fd > 0)) {
+               close(IO->SendBuf.fd);
+               IO->SendBuf.fd = 0;
+               IO->RecvBuf.fd = 0;
+       }
+}
+
+void StopCurlWatchers(AsyncIO *IO)
 {
        ev_timer_stop (event_base, &IO->rw_timeout);
        ev_timer_stop(event_base, &IO->conn_fail);
        ev_idle_stop(event_base, &IO->unwind_stack);
+       ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
 
        ev_io_stop(event_base, &IO->conn_event);
        ev_io_stop(event_base, &IO->send_event);
        ev_io_stop(event_base, &IO->recv_event);
-       close(IO->SendBuf.fd);
+
+       if (IO->SendBuf.fd != 0) {
+               close(IO->SendBuf.fd);
+       }
        IO->SendBuf.fd = 0;
        IO->RecvBuf.fd = 0;
 }
@@ -268,11 +346,12 @@ void ShutDownCLient(AsyncIO *IO)
 
        EVM_syslog(LOG_DEBUG, "EVENT Terminating \n");
 
-       ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
-       StopClientWatchers(IO);
+       StopClientWatchers(IO, 1);
 
        if (IO->DNS.Channel != NULL) {
                ares_destroy(IO->DNS.Channel);
+               EV_DNS_LOG_STOP(DNS.recv_event);
+               EV_DNS_LOG_STOP(DNS.send_event);
                ev_io_stop(event_base, &IO->DNS.recv_event);
                ev_io_stop(event_base, &IO->DNS.send_event);
                IO->DNS.Channel = NULL;
@@ -281,6 +360,52 @@ void ShutDownCLient(AsyncIO *IO)
        IO->Terminate(IO);
 }
 
+void PostInbound(AsyncIO *IO)
+{
+       switch (IO->NextState) {
+       case eSendFile:
+               ev_io_start(event_base, &IO->send_event);
+               break;
+       case eSendReply:
+       case eSendMore:
+               assert(IO->SendDone);
+               IO->NextState = IO->SendDone(IO);
+               switch (IO->NextState)
+               {
+               case eSendFile:
+               case eSendReply:
+               case eSendMore:
+               case eReadMessage:
+               case eReadPayload:
+               case eReadMore:
+               case eReadFile:
+                       ev_io_start(event_base, &IO->send_event);
+                       break;
+               case eDBQuery:
+                       StopClientWatchers(IO, 0);
+               default:
+                       break;
+               }
+               break;
+       case eReadPayload:
+       case eReadMore:
+       case eReadFile:
+               ev_io_start(event_base, &IO->recv_event);
+               break;
+       case eTerminateConnection:
+               ShutDownCLient(IO);
+               break;
+       case eAbort:
+               ShutDownCLient(IO);
+               break;
+       case eSendDNSQuery:
+       case eReadDNSReply:
+       case eDBQuery:
+       case eConnect:
+       case eReadMessage:
+               break;
+       }
+}
 eReadState HandleInbound(AsyncIO *IO)
 {
        const char *Err = NULL;
@@ -333,34 +458,8 @@ eReadState HandleInbound(AsyncIO *IO)
                }
        }
 
-       switch (IO->NextState) {
-       case eSendFile:
-               ev_io_start(event_base, &IO->send_event);
-               break;
-       case eSendReply:
-       case eSendMore:
-               assert(IO->SendDone);
-               IO->NextState = IO->SendDone(IO);
-               ev_io_start(event_base, &IO->send_event);
-               break;
-       case eReadPayload:
-       case eReadMore:
-       case eReadFile:
-               ev_io_start(event_base, &IO->recv_event);
-               break;
-       case eTerminateConnection:
-//////TODOxxxx
-               break;
-       case eAbort:
-               ShutDownCLient(IO);
-               break;
-       case eSendDNSQuery:
-       case eReadDNSReply:
-       case eDBQuery:
-       case eConnect:
-       case eReadMessage:
-               break;
-       }
+       PostInbound(IO);
+
        return Finished;
 }
 
@@ -372,6 +471,7 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
        AsyncIO *IO = watcher->data;
        const char *errmsg = NULL;
 
+       IO->Now = ev_now(event_base);
        become_session(IO->CitContext);
 #ifdef BIGBAD_IODBG
        {
@@ -404,8 +504,8 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
                                StrBufPlain(IO->ErrMsg, errmsg, -1);
                        break;
                default:
-                       rc = StrBuf_write_one_chunk_callback(watcher->fd,
-                                                            0/*TODO*/,
+                       rc = StrBuf_write_one_chunk_callback(IO->SendBuf.fd,
+                                                            0,
                                                             &IO->SendBuf);
                }
 
@@ -432,6 +532,8 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
                case eSendFile:
                        if (IO->IOB.ChunkSendRemain > 0) {
                                ev_io_start(event_base, &IO->recv_event);
+                               SetNextTimeout(IO, 100.0);
+
                        } else {
                                assert(IO->ReadDone);
                                IO->NextState = IO->ReadDone(IO);
@@ -491,18 +593,30 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
                }
        }
        else if (rc < 0) {
-               assert(IO->Timeout);
-               IO->Timeout(IO);
+               if (errno != EAGAIN) {
+                       StopClientWatchers(IO, 1);
+                       EV_syslog(LOG_DEBUG,
+                                 "IO_send_callback(): Socket Invalid! [%d] [%s] [%d]\n",
+                                 errno, strerror(errno), IO->SendBuf.fd);
+                       StrBufPrintf(IO->ErrMsg,
+                                    "Socket Invalid! [%s]",
+                                    strerror(errno));
+                       SetNextTimeout(IO, 0.01);
+               }
        }
        /* else : must write more. */
 }
 static void
 set_start_callback(struct ev_loop *loop, AsyncIO *IO, int revents)
 {
+       ev_timer_stop(event_base, &IO->conn_fail);
+       ev_timer_start(event_base, &IO->rw_timeout);
+
        switch(IO->NextState) {
        case eReadMore:
        case eReadMessage:
        case eReadFile:
+               StrBufAppendBufPlain(IO->ErrMsg, HKEY("[while waiting for greeting]"), 0);
                ev_io_start(event_base, &IO->recv_event);
                break;
        case eSendReply:
@@ -528,6 +642,7 @@ IO_Timeout_callback(struct ev_loop *loop, ev_timer *watcher, int revents)
 {
        AsyncIO *IO = watcher->data;
 
+       IO->Now = ev_now(event_base);
        ev_timer_stop (event_base, &IO->rw_timeout);
        become_session(IO->CitContext);
 
@@ -555,6 +670,7 @@ IO_connfail_callback(struct ev_loop *loop, ev_timer *watcher, int revents)
 {
        AsyncIO *IO = watcher->data;
 
+       IO->Now = ev_now(event_base);
        ev_timer_stop (event_base, &IO->conn_fail);
 
        if (IO->SendBuf.fd != 0)
@@ -586,6 +702,7 @@ IO_connfailimmediate_callback(struct ev_loop *loop,
 {
        AsyncIO *IO = watcher->data;
 
+       IO->Now = ev_now(event_base);
        ev_idle_stop (event_base, &IO->conn_fail_immediate);
 
        if (IO->SendBuf.fd != 0)
@@ -609,12 +726,38 @@ IO_connfailimmediate_callback(struct ev_loop *loop,
 static void
 IO_connestd_callback(struct ev_loop *loop, ev_io *watcher, int revents)
 {
-       AsyncIO *IO = watcher->data;
-
-       ev_io_stop(loop, &IO->conn_event);
-       ev_timer_stop (event_base, &IO->conn_fail);
-       set_start_callback(loop, IO, revents);
+        AsyncIO *IO = watcher->data;
+        int             so_err = 0;
+        socklen_t       lon = sizeof(so_err);
+        int             err;
+
+        IO->Now = ev_now(event_base);
+        EVM_syslog(LOG_DEBUG, "connect() succeeded.\n");
+
+        ev_io_stop(loop, &IO->conn_event);
+        ev_timer_stop(event_base, &IO->conn_fail);
+
+        err = getsockopt(IO->SendBuf.fd,
+                         SOL_SOCKET,
+                         SO_ERROR,
+                         (void*)&so_err,
+                         &lon);
+
+        if ((err == 0) && (so_err != 0))
+        {
+                EV_syslog(LOG_DEBUG, "connect() failed [%d][%s]\n",
+                          so_err,
+                          strerror(so_err));
+                IO_connfail_callback(loop, &IO->conn_fail, revents);
+
+        }
+        else
+        {
+                EVM_syslog(LOG_DEBUG, "connect() succeeded\n");
+                set_start_callback(loop, IO, revents);
+        }
 }
+
 static void
 IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents)
 {
@@ -622,6 +765,7 @@ IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents)
        ssize_t nbytes;
        AsyncIO *IO = watcher->data;
 
+       IO->Now = ev_now(event_base);
        switch (IO->NextState) {
        case eReadFile:
                nbytes = FileRecvChunked(&IO->IOB, &errmsg);
@@ -632,14 +776,18 @@ IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents)
                        if (IO->IOB.ChunkSendRemain == 0)
                        {
                                IO->NextState = eSendReply;
+                               assert(IO->ReadDone);
+                               ev_io_stop(event_base, &IO->recv_event);
+                               PostInbound(IO);
+                               return;
                        }
                        else
                                return;
                }
                break;
        default:
-               nbytes = StrBuf_read_one_chunk_callback(watcher->fd,
-                                                       0 /*TODO */,
+               nbytes = StrBuf_read_one_chunk_callback(IO->RecvBuf.fd,
+                                                       0,
                                                        &IO->RecvBuf);
                break;
        }
@@ -673,22 +821,21 @@ IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents)
        if (nbytes > 0) {
                HandleInbound(IO);
        } else if (nbytes == 0) {
-               assert(IO->Timeout);
-
-               switch (IO->Timeout(IO))
-               {
-               case eAbort:
-                       ShutDownCLient(IO);
-               default:
-                       break;
-               }
+               StopClientWatchers(IO, 1);
+               SetNextTimeout(IO, 0.01);
                return;
        } else if (nbytes == -1) {
-/// TODO: FD is gone. kick it.        sock_buff_invoke_free(sb, errno);
-               EV_syslog(LOG_DEBUG,
-                         "EVENT: Socket Invalid! %s \n",
-                         strerror(errno));
-               ShutDownCLient(IO);
+               if (errno != EAGAIN) {
+                       // FD is gone. kick it. 
+                       StopClientWatchers(IO, 1);
+                       EV_syslog(LOG_DEBUG,
+                                 "IO_recv_callback(): Socket Invalid! [%d] [%s] [%d]\n",
+                                 errno, strerror(errno), IO->SendBuf.fd);
+                       StrBufPrintf(IO->ErrMsg,
+                                    "Socket Invalid! [%s]",
+                                    strerror(errno));
+                       SetNextTimeout(IO, 0.01);
+               }
                return;
        }
 }
@@ -697,15 +844,17 @@ void
 IO_postdns_callback(struct ev_loop *loop, ev_idle *watcher, int revents)
 {
        AsyncIO *IO = watcher->data;
+       IO->Now = ev_now(event_base);
        EV_syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__);
        become_session(IO->CitContext);
-       assert(IO->DNS.Fail);
        assert(IO->DNS.Query->PostDNS);
        switch (IO->DNS.Query->PostDNS(IO))
        {
        case eAbort:
+               assert(IO->DNS.Fail);
                switch (IO->DNS.Fail(IO)) {
                case eAbort:
+////                   StopClientWatchers(IO);
                        ShutDownCLient(IO);
                default:
                        break;
@@ -721,6 +870,7 @@ eNextState EvConnectSock(AsyncIO *IO,
                         double first_rw_timeout,
                         int ReadFirst)
 {
+       struct sockaddr_in egress_sin;
        int fdflags;
        int rc = -1;
 
@@ -747,29 +897,35 @@ eNextState EvConnectSock(AsyncIO *IO,
                StrBufPrintf(IO->ErrMsg,
                             "Failed to create socket: %s",
                             strerror(errno));
+               IO->SendBuf.fd = IO->RecvBuf.fd = 0;
                return eAbort;
        }
        fdflags = fcntl(IO->SendBuf.fd, F_GETFL);
        if (fdflags < 0) {
-               EV_syslog(LOG_DEBUG,
-                         "EVENT: unable to get socket flags! %s \n",
+               EV_syslog(LOG_ERR,
+                         "EVENT: unable to get socket %d flags! %s \n",
+                         IO->SendBuf.fd,
                          strerror(errno));
                StrBufPrintf(IO->ErrMsg,
-                            "Failed to get socket flags: %s",
+                            "Failed to get socket %d flags: %s",
+                            IO->SendBuf.fd,
                             strerror(errno));
+               close(IO->SendBuf.fd);
+               IO->SendBuf.fd = IO->RecvBuf.fd = 0;
                return eAbort;
        }
        fdflags = fdflags | O_NONBLOCK;
        if (fcntl(IO->SendBuf.fd, F_SETFL, fdflags) < 0) {
                EV_syslog(
-                       LOG_DEBUG,
-                       "EVENT: unable to set socket nonblocking flags! %s \n",
+                       LOG_ERR,
+                       "EVENT: unable to set socket %d nonblocking flags! %s \n",
+                       IO->SendBuf.fd,
                        strerror(errno));
                StrBufPrintf(IO->ErrMsg,
                             "Failed to set socket flags: %s",
                             strerror(errno));
                close(IO->SendBuf.fd);
-               IO->SendBuf.fd = IO->RecvBuf.fd = -1;
+               IO->SendBuf.fd = IO->RecvBuf.fd = 0;
                return eAbort;
        }
 /* TODO: maye we could use offsetof() to calc the position of data...
@@ -786,28 +942,46 @@ eNextState EvConnectSock(AsyncIO *IO,
        IO->rw_timeout.data = IO;
 
 
+
+
        /* for debugging you may bypass it like this:
         * IO->Addr.sin_addr.s_addr = inet_addr("127.0.0.1");
         * ((struct sockaddr_in)IO->ConnectMe->Addr).sin_addr.s_addr =
         *   inet_addr("127.0.0.1");
         */
-       if (IO->ConnectMe->IPv6)
+       if (IO->ConnectMe->IPv6) {
                rc = connect(IO->SendBuf.fd,
                             &IO->ConnectMe->Addr,
                             sizeof(struct sockaddr_in6));
-       else
+       }
+       else {
+               /* If citserver is bound to a specific IP address on the host, make
+                * sure we use that address for outbound connections.
+                */
+       
+               memset(&egress_sin, 0, sizeof(egress_sin));
+               egress_sin.sin_family = AF_INET;
+               if (!IsEmptyStr(config.c_ip_addr)) {
+                       egress_sin.sin_addr.s_addr = inet_addr(config.c_ip_addr);
+                       if (egress_sin.sin_addr.s_addr == !INADDR_ANY) {
+                               egress_sin.sin_addr.s_addr = INADDR_ANY;
+                       }
+
+                       /* If this bind fails, no problem; we can still use INADDR_ANY */
+                       bind(IO->SendBuf.fd, (struct sockaddr *)&egress_sin, sizeof(egress_sin));
+               }
                rc = connect(IO->SendBuf.fd,
                             (struct sockaddr_in *)&IO->ConnectMe->Addr,
                             sizeof(struct sockaddr_in));
+       }
 
        if (rc >= 0){
-               EVM_syslog(LOG_DEBUG, "connect() immediate success.\n");
+               EV_syslog(LOG_DEBUG, "connect() = %d immediate success.\n", IO->SendBuf.fd);
                set_start_callback(event_base, IO, 0);
-               ev_timer_start(event_base, &IO->rw_timeout);
                return IO->NextState;
        }
        else if (errno == EINPROGRESS) {
-               EVM_syslog(LOG_DEBUG, "connect() have to wait now.\n");
+               EV_syslog(LOG_DEBUG, "connect() = %d have to wait now.\n", IO->SendBuf.fd);
 
                ev_io_init(&IO->conn_event,
                           IO_connestd_callback,
@@ -826,7 +1000,11 @@ eNextState EvConnectSock(AsyncIO *IO,
                IO->conn_fail_immediate.data = IO;
                ev_idle_start(event_base, &IO->conn_fail_immediate);
 
-               EV_syslog(LOG_ERR, "connect() failed: %s\n", strerror(errno));
+               EV_syslog(LOG_ERR,
+                         "connect() = %d failed: %s\n",
+                         IO->SendBuf.fd,
+                         strerror(errno));
+
                StrBufPrintf(IO->ErrMsg,
                             "Failed to connect: %s",
                             strerror(errno));
@@ -868,6 +1046,7 @@ void InitIOStruct(AsyncIO *IO,
                  IO_CallBack SendDone,
                  IO_CallBack ReadDone,
                  IO_CallBack Terminate,
+                 IO_CallBack DBTerminate,
                  IO_CallBack ConnFail,
                  IO_CallBack Timeout,
                  IO_CallBack ShutdownAbort)
@@ -875,13 +1054,15 @@ void InitIOStruct(AsyncIO *IO,
        IO->Data          = Data;
 
        IO->CitContext    = CloneContext(CC);
-       ((CitContext *)IO->CitContext)->session_specific_data = (char*) Data;
+       IO->CitContext->session_specific_data = Data;
+       IO->CitContext->IO = IO;
 
        IO->NextState     = NextState;
 
        IO->SendDone      = SendDone;
        IO->ReadDone      = ReadDone;
        IO->Terminate     = Terminate;
+       IO->DBTerminate   = DBTerminate;
        IO->LineReader    = LineReader;
        IO->ConnFail      = ConnFail;
        IO->Timeout       = Timeout;
@@ -892,6 +1073,9 @@ void InitIOStruct(AsyncIO *IO,
        IO->SendBuf.Buf   = NewStrBufPlain(NULL, 1024);
        IO->RecvBuf.Buf   = NewStrBufPlain(NULL, 1024);
        IO->IOBuf         = NewStrBuf();
+       EV_syslog(LOG_DEBUG,
+                 "EVENT: Session lives at %p IO at %p \n",
+                 Data, IO);
 
 }
 
@@ -902,15 +1086,18 @@ int InitcURLIOStruct(AsyncIO *IO,
                     const char* Desc,
                     IO_CallBack SendDone,
                     IO_CallBack Terminate,
+                    IO_CallBack DBTerminate,
                     IO_CallBack ShutdownAbort)
 {
        IO->Data          = Data;
 
        IO->CitContext    = CloneContext(CC);
-       ((CitContext *)IO->CitContext)->session_specific_data = (char*) Data;
+       IO->CitContext->session_specific_data = Data;
+       IO->CitContext->IO = IO;
 
-       IO->SendDone = SendDone;
-       IO->Terminate = Terminate;
+       IO->SendDone      = SendDone;
+       IO->Terminate     = Terminate;
+       IO->DBTerminate   = DBTerminate;
        IO->ShutdownAbort = ShutdownAbort;
 
        strcpy(IO->HttpReq.errdesc, Desc);
@@ -919,3 +1106,115 @@ int InitcURLIOStruct(AsyncIO *IO,
        return  evcurl_init(IO);
 
 }
+
+
+typedef struct KillOtherSessionContext {
+       AsyncIO IO;
+       AsyncIO *OtherOne;
+}KillOtherSessionContext;
+
+eNextState KillTerminate(AsyncIO *IO)
+{
+       long id;
+       KillOtherSessionContext *Ctx = (KillOtherSessionContext*)IO->Data;
+       EV_syslog(LOG_DEBUG, "%s Exit\n", __FUNCTION__);
+       id = IO->ID;
+       FreeAsyncIOContents(IO);
+       memset(Ctx, 0, sizeof(KillOtherSessionContext));
+       IO->ID = id; /* just for the case we want to analyze it in a coredump */
+       free(Ctx);
+       return eAbort;
+
+}
+
+eNextState KillShutdown(AsyncIO *IO)
+{
+       return eTerminateConnection;
+}
+
+eNextState KillOtherContextNow(AsyncIO *IO)
+{
+       KillOtherSessionContext *Ctx = IO->Data;
+
+       if (Ctx->OtherOne->ShutdownAbort != NULL)
+               Ctx->OtherOne->ShutdownAbort(Ctx->OtherOne);
+       return eTerminateConnection;
+}
+
+void KillAsyncIOContext(AsyncIO *IO)
+{
+       KillOtherSessionContext *Ctx;
+
+       Ctx = (KillOtherSessionContext*) malloc(sizeof(KillOtherSessionContext));
+       memset(Ctx, 0, sizeof(KillOtherSessionContext));
+       
+       InitIOStruct(&Ctx->IO,
+                    Ctx,
+                    eReadMessage,
+                    NULL,
+                    NULL,
+                    NULL,
+                    NULL,
+                    KillTerminate,
+                    NULL,
+                    NULL,
+                    NULL,
+                    KillShutdown);
+
+       Ctx->OtherOne = IO;
+
+       switch(IO->NextState) {
+       case eSendDNSQuery:
+       case eReadDNSReply:
+
+       case eConnect:
+       case eSendReply:
+       case eSendMore:
+       case eSendFile:
+
+       case eReadMessage:
+       case eReadMore:
+       case eReadPayload:
+       case eReadFile:
+               QueueEventContext(&Ctx->IO, KillOtherContextNow);
+               break;
+       case eDBQuery:
+               QueueDBOperation(&Ctx->IO, KillOtherContextNow);
+               break;
+       case eTerminateConnection:
+       case eAbort:
+               /*hm, its already dying, dunno which Queue its in... */
+               free(Ctx);
+       }
+       
+}
+
+extern int DebugEventLoopBacktrace;
+void EV_backtrace(AsyncIO *IO)
+{
+#ifdef HAVE_BACKTRACE
+       void *stack_frames[50];
+       size_t size, i;
+       char **strings;
+
+       if ((IO == NULL) || (DebugEventLoopBacktrace == 0))
+               return;
+       size = backtrace(stack_frames, sizeof(stack_frames) / sizeof(void*));
+       strings = backtrace_symbols(stack_frames, size);
+       for (i = 0; i < size; i++) {
+               if (strings != NULL) {
+                       EV_syslog(LOG_ALERT, " BT %s\n", strings[i]);
+               }
+               else {
+                       EV_syslog(LOG_ALERT, " BT %p\n", stack_frames[i]);
+               }
+       }
+       free(strings);
+#endif
+}
+
+
+ev_tstamp ctdl_ev_now (void)
+{
+       return ev_now(event_base);
+}