Networker: remove unneeded assignment

[citadel.git] / citadel / event_client.c

diff --git a/citadel/event_client.c b/citadel/event_client.c
index 697fd7bcd87a0e171e54fd24090a11b82e2a9a5d..c02f3bbea048bdde5a01d9b36ef353d1f4fc84a5 100644 (file)
--- a/citadel/event_client.c
+++ b/citadel/event_client.c
@@ -82,6 +82,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;
@@ -98,26 +99,47 @@ 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->DBTerminate);
        IO->DBTerminate(IO);
@@ -173,6 +195,7 @@ 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;
@@ -201,15 +224,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;
 }
@@ -226,12 +262,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;
 }
@@ -262,6 +314,25 @@ void StopClientWatchers(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);
+
+       if (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);
@@ -281,7 +352,6 @@ void ShutDownCLient(AsyncIO *IO)
 
        EVM_syslog(LOG_DEBUG, "EVENT Terminating \n");
 
-       ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
        StopClientWatchers(IO);
 
        if (IO->DNS.Channel != NULL) {
@@ -521,10 +591,14 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
 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:
@@ -634,13 +708,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;
-
-       IO->Now = ev_now(event_base);
-       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)
 {
@@ -779,7 +878,7 @@ eNextState EvConnectSock(AsyncIO *IO,
        }
        fdflags = fcntl(IO->SendBuf.fd, F_GETFL);
        if (fdflags < 0) {
-               EV_syslog(LOG_DEBUG,
+               EV_syslog(LOG_ERR,
                          "EVENT: unable to get socket flags! %s \n",
                          strerror(errno));
                StrBufPrintf(IO->ErrMsg,
@@ -792,7 +891,7 @@ eNextState EvConnectSock(AsyncIO *IO,
        fdflags = fdflags | O_NONBLOCK;
        if (fcntl(IO->SendBuf.fd, F_SETFL, fdflags) < 0) {
                EV_syslog(
-                       LOG_DEBUG,
+                       LOG_ERR,
                        "EVENT: unable to set socket nonblocking flags! %s \n",
                        strerror(errno));
                StrBufPrintf(IO->ErrMsg,
@@ -852,7 +951,6 @@ eNextState EvConnectSock(AsyncIO *IO,
        if (rc >= 0){
                EVM_syslog(LOG_DEBUG, "connect() immediate success.\n");
                set_start_callback(event_base, IO, 0);
-               ev_timer_start(event_base, &IO->rw_timeout);
                return IO->NextState;
        }
        else if (errno == EINPROGRESS) {