X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fevent_client.c;h=999c9e4546c1b989efc6fb93c6f6ab8df8394b8f;hb=8357d67fb22adec3b854d61bdbd898dcfcc91959;hp=d0e0fd9aefe9f20295a8827fda8fb5dabcf3f26b;hpb=b1493ab199a2cf39b919cb8d0536263ae1d280aa;p=citadel.git diff --git a/citadel/event_client.c b/citadel/event_client.c index d0e0fd9ae..999c9e454 100644 --- a/citadel/event_client.c +++ b/citadel/event_client.c @@ -1,76 +1,75 @@ /* - * * 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" -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#if TIME_WITH_SYS_TIME -# include -# include -#else -# if HAVE_SYS_TIME_H -# include -# else -# include -# endif -#endif -#include -#include +#include #include -#include +#include +#include #include #include #include -#include #if HAVE_BACKTRACE #include #endif #include -#include "citadel.h" -#include "server.h" -#include "citserver.h" -#include "support.h" -#include "config.h" -#include "control.h" -#include "user_ops.h" -#include "database.h" -#include "msgbase.h" -#include "internet_addressing.h" -#include "genstamp.h" -#include "domain.h" -#include "clientsocket.h" -#include "locate_host.h" -#include "citadel_dirs.h" -#include "event_client.h" #include "ctdl_module.h" +#include "event_client.h" +#include "citserver.h" + +ConstStr IOStates[] = { + {HKEY("DB Queue")}, + {HKEY("DB Q Next")}, + {HKEY("DB Attach")}, + {HKEY("DB Next")}, + {HKEY("DB Stop")}, + {HKEY("DB Exit")}, + {HKEY("DB Terminate")}, + {HKEY("IO Queue")}, + {HKEY("IO Attach")}, + {HKEY("IO Connect Socket")}, + {HKEY("IO Abort")}, + {HKEY("IO Timeout")}, + {HKEY("IO ConnFail")}, + {HKEY("IO ConnFail Now")}, + {HKEY("IO Conn Now")}, + {HKEY("IO Conn Wait")}, + {HKEY("Curl Q")}, + {HKEY("Curl Start")}, + {HKEY("Curl Shotdown")}, + {HKEY("Curl More IO")}, + {HKEY("Curl Got IO")}, + {HKEY("Curl Got Data")}, + {HKEY("Curl Got Status")}, + {HKEY("C-Ares Start")}, + {HKEY("C-Ares IO Done")}, + {HKEY("C-Ares Finished")}, + {HKEY("C-Ares exit")}, + {HKEY("Killing")}, + {HKEY("Exit")} +}; + +void SetEVState(AsyncIO *IO, eIOState State) +{ + + CitContext* CCC = IO->CitContext; + if (CCC != NULL) + memcpy(CCC->lastcmdname, IOStates[State].Key, IOStates[State].len + 1); + +} +eNextState QueueAnEventContext(AsyncIO *IO); 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, @@ -82,42 +81,70 @@ 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; extern ev_async DBExitEventLoop; -eNextState QueueDBOperation(AsyncIO *IO, IO_CallBack CB) +eNextState QueueAnDBOperation(AsyncIO *IO) { IOAddHandler *h; int i; + SetEVState(IO, eDBQ); h = (IOAddHandler*)malloc(sizeof(IOAddHandler)); h->IO = IO; - h->EvAttch = CB; + + assert(IO->ReAttachCB != NULL); + + h->EvAttch = IO->ReAttachCB; 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); - EVM_syslog(LOG_DEBUG, "DBEVENT Q Done.\n"); + pthread_mutex_unlock(&DBEventExitQueueMutex); + + EVQM_syslog(LOG_DEBUG, "DBEVENT Q Done.\n"); return eDBQuery; } +void StopDBWatchers(AsyncIO *IO) +{ + SetEVState(IO, eDBStop); + 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); + SetEVState(IO, eDBTerm); 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); @@ -127,6 +154,8 @@ void DB_PerformNext(struct ev_loop *loop, ev_idle *watcher, int revents) { AsyncIO *IO = watcher->data; + + SetEVState(IO, eDBNext); IO->Now = ev_now(event_db); EV_syslog(LOG_DEBUG, "%s()", __FUNCTION__); become_session(IO->CitContext); @@ -136,12 +165,15 @@ DB_PerformNext(struct ev_loop *loop, ev_idle *watcher, int revents) assert(IO->NextDBOperation); switch (IO->NextDBOperation(IO)) { + case eSendReply: + ev_cleanup_stop(loop, &IO->db_abort_by_shutdown); + QueueAnEventContext(IO); + break; case eDBQuery: break; case eSendDNSQuery: case eReadDNSReply: case eConnect: - case eSendReply: case eSendMore: case eSendFile: case eReadMessage: @@ -160,6 +192,7 @@ DB_PerformNext(struct ev_loop *loop, ev_idle *watcher, int revents) eNextState NextDBOperation(AsyncIO *IO, IO_CallBack CB) { + SetEVState(IO, eQDBNext); IO->NextDBOperation = CB; ev_idle_init(&IO->db_unwind_stack, DB_PerformNext); @@ -173,6 +206,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; @@ -183,6 +217,8 @@ static void IO_abort_shutdown_callback(struct ev_loop *loop, int revents) { AsyncIO *IO = watcher->data; + + SetEVState(IO, eIOAbort); EV_syslog(LOG_DEBUG, "EVENT Q: %s\n", __FUNCTION__); IO->Now = ev_now(event_base); assert(IO->ShutdownAbort); @@ -190,30 +226,67 @@ static void IO_abort_shutdown_callback(struct ev_loop *loop, } -eNextState QueueEventContext(AsyncIO *IO, IO_CallBack CB) +eNextState QueueAnEventContext(AsyncIO *IO) { IOAddHandler *h; int i; + SetEVState(IO, eIOQ); h = (IOAddHandler*)malloc(sizeof(IOAddHandler)); h->IO = IO; - h->EvAttch = CB; + + assert(IO->ReAttachCB != NULL); + + h->EvAttch = IO->ReAttachCB; + 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; } +eNextState EventQueueDBOperation(AsyncIO *IO, IO_CallBack CB, int CloseFDs) +{ + StopClientWatchers(IO, CloseFDs); + IO->ReAttachCB = CB; + return eDBQuery; +} +eNextState DBQueueEventContext(AsyncIO *IO, IO_CallBack CB) +{ + StopDBWatchers(IO); + IO->ReAttachCB = CB; + return eSendReply; +} + +eNextState QueueEventContext(AsyncIO *IO, IO_CallBack CB) +{ + IO->ReAttachCB = CB; + return QueueAnEventContext(IO); +} + extern eNextState evcurl_handle_start(AsyncIO *IO); eNextState QueueCurlContext(AsyncIO *IO) @@ -221,22 +294,46 @@ eNextState QueueCurlContext(AsyncIO *IO) IOAddHandler *h; int i; + SetEVState(IO, eCurlQ); h = (IOAddHandler*)malloc(sizeof(IOAddHandler)); h->IO = 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; } -void DestructCAres(AsyncIO *IO); +eNextState CurlQueueDBOperation(AsyncIO *IO, IO_CallBack CB) +{ + StopCurlWatchers(IO); + IO->ReAttachCB = CB; + return eDBQuery; +} + + void FreeAsyncIOContents(AsyncIO *IO) { CitContext *Ctx = IO->CitContext; @@ -245,28 +342,56 @@ void FreeAsyncIOContents(AsyncIO *IO) FreeStrBuf(&IO->SendBuf.Buf); FreeStrBuf(&IO->RecvBuf.Buf); - DestructCAres(IO); - FreeURL(&IO->ConnectMe); FreeStrBuf(&IO->HttpReq.ReplyData); if (Ctx) { Ctx->state = CON_IDLE; Ctx->kill_me = 1; + IO->CitContext = NULL; } } -void StopClientWatchers(AsyncIO *IO) +void DestructCAres(AsyncIO *IO); +void StopClientWatchers(AsyncIO *IO, int CloseFD) +{ + EVM_syslog(LOG_DEBUG, "EVENT StopClientWatchers"); + + DestructCAres(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 (CloseFD && (IO->SendBuf.fd > 0)) { + close(IO->SendBuf.fd); + IO->SendBuf.fd = 0; + IO->RecvBuf.fd = 0; + } +} + +void StopCurlWatchers(AsyncIO *IO) { + EVM_syslog(LOG_DEBUG, "EVENT StopCurlWatchers \n"); + 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); + curl_easy_cleanup(IO->HttpReq.chnd); + IO->HttpReq.chnd = NULL; + if (IO->SendBuf.fd != 0) { close(IO->SendBuf.fd); } @@ -274,15 +399,16 @@ void StopClientWatchers(AsyncIO *IO) IO->RecvBuf.fd = 0; } -void ShutDownCLient(AsyncIO *IO) +eNextState ShutDownCLient(AsyncIO *IO) { CitContext *Ctx =IO->CitContext; + + SetEVState(IO, eExit); become_session(Ctx); 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); @@ -293,11 +419,12 @@ void ShutDownCLient(AsyncIO *IO) IO->DNS.Channel = NULL; } assert(IO->Terminate); - IO->Terminate(IO); + return IO->Terminate(IO); } void PostInbound(AsyncIO *IO) { + switch (IO->NextState) { case eSendFile: ev_io_start(event_base, &IO->send_event); @@ -306,7 +433,23 @@ void PostInbound(AsyncIO *IO) case eSendMore: assert(IO->SendDone); IO->NextState = IO->SendDone(IO); - ev_io_start(event_base, &IO->send_event); + 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); + QueueAnDBOperation(IO); + default: + break; + } break; case eReadPayload: case eReadMore: @@ -314,17 +457,18 @@ void PostInbound(AsyncIO *IO) ev_io_start(event_base, &IO->recv_event); break; case eTerminateConnection: - ShutDownCLient(IO); - break; case eAbort: - ShutDownCLient(IO); + if (ShutDownCLient(IO) == eDBQuery) { + QueueAnDBOperation(IO); + } break; case eSendDNSQuery: case eReadDNSReply: - case eDBQuery: case eConnect: case eReadMessage: break; + case eDBQuery: + QueueAnDBOperation(IO); } } eReadState HandleInbound(AsyncIO *IO) @@ -372,10 +516,17 @@ eReadState HandleInbound(AsyncIO *IO) } if (Finished != eMustReadMore) { - assert(IO->ReadDone); ev_io_stop(event_base, &IO->recv_event); IO->NextState = IO->ReadDone(IO); - Finished = StrBufCheckBuffer(&IO->RecvBuf); + if (IO->NextState == eDBQuery) { + if (QueueAnDBOperation(IO) == eAbort) + return eReadFail; + else + return eReadSuccess; + } + else { + Finished = StrBufCheckBuffer(&IO->RecvBuf); + } } } @@ -412,6 +563,11 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents) IO->SendBuf.fd); fd = fopen(fn, "a+"); + if (fd == NULL) { + syslog(LOG_EMERG, "failed to open file %s: %s", fn, strerror(errno)); + cit_backtrace(); + exit(1); + } fprintf(fd, "Send: BufSize: %ld BufContent: [", nbytes); rv = fwrite(pchh, nbytes, 1, fd); @@ -425,8 +581,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); } @@ -514,17 +670,30 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents) } } else if (rc < 0) { - IO_Timeout_callback(loop, &IO->rw_timeout, revents); + 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: @@ -550,6 +719,7 @@ IO_Timeout_callback(struct ev_loop *loop, ev_timer *watcher, int revents) { AsyncIO *IO = watcher->data; + SetEVState(IO, eIOTimeout); IO->Now = ev_now(event_base); ev_timer_stop (event_base, &IO->rw_timeout); become_session(IO->CitContext); @@ -578,6 +748,7 @@ IO_connfail_callback(struct ev_loop *loop, ev_timer *watcher, int revents) { AsyncIO *IO = watcher->data; + SetEVState(IO, eIOConnfail); IO->Now = ev_now(event_base); ev_timer_stop (event_base, &IO->conn_fail); @@ -610,6 +781,7 @@ IO_connfailimmediate_callback(struct ev_loop *loop, { AsyncIO *IO = watcher->data; + SetEVState(IO, eIOConnfailNow); IO->Now = ev_now(event_base); ev_idle_stop (event_base, &IO->conn_fail_immediate); @@ -634,13 +806,39 @@ 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; + + SetEVState(IO, eIOConnNow); + 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) { @@ -669,8 +867,8 @@ IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents) } 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; } @@ -693,6 +891,11 @@ IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents) IO->SendBuf.fd); fd = fopen(fn, "a+"); + if (fd == NULL) { + syslog(LOG_EMERG, "failed to open file %s: %s", fn, strerror(errno)); + cit_backtrace(); + exit(1); + } fprintf(fd, "Read: BufSize: %ld BufContent: [", nbytes); rv = fwrite(pchh, nbytes, 1, fd); @@ -704,15 +907,21 @@ IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents) if (nbytes > 0) { HandleInbound(IO); } else if (nbytes == 0) { - IO_Timeout_callback(loop, &IO->rw_timeout, revents); + StopClientWatchers(IO, 1); + SetNextTimeout(IO, 0.01); return; } else if (nbytes == -1) { - // FD is gone. kick it. - StopClientWatchers(IO); - 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; } } @@ -721,6 +930,8 @@ void IO_postdns_callback(struct ev_loop *loop, ev_idle *watcher, int revents) { AsyncIO *IO = watcher->data; + + SetEVState(IO, eCaresFinished); IO->Now = ev_now(event_base); EV_syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__); become_session(IO->CitContext); @@ -733,9 +944,18 @@ IO_postdns_callback(struct ev_loop *loop, ev_idle *watcher, int revents) case eAbort: //// StopClientWatchers(IO); ShutDownCLient(IO); + break; + case eDBQuery: + StopClientWatchers(IO, 0); + QueueAnDBOperation(IO); + break; default: break; } + case eDBQuery: + StopClientWatchers(IO, 0); + QueueAnDBOperation(IO); + break; default: break; } @@ -751,6 +971,7 @@ eNextState EvConnectSock(AsyncIO *IO, int fdflags; int rc = -1; + SetEVState(IO, eIOConnectSock); become_session(IO->CitContext); if (ReadFirst) { @@ -779,11 +1000,13 @@ eNextState EvConnectSock(AsyncIO *IO, } 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; @@ -792,8 +1015,9 @@ eNextState EvConnectSock(AsyncIO *IO, 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", @@ -850,13 +1074,14 @@ eNextState EvConnectSock(AsyncIO *IO, } if (rc >= 0){ - EVM_syslog(LOG_DEBUG, "connect() immediate success.\n"); + SetEVState(IO, eIOConnNow); + 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"); + SetEVState(IO, eIOConnWait); + EV_syslog(LOG_DEBUG, "connect() = %d have to wait now.\n", IO->SendBuf.fd); ev_io_init(&IO->conn_event, IO_connestd_callback, @@ -870,12 +1095,17 @@ eNextState EvConnectSock(AsyncIO *IO, return IO->NextState; } else { + SetEVState(IO, eIOConnfail); ev_idle_init(&IO->conn_fail_immediate, IO_connfailimmediate_callback); 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)); @@ -895,6 +1125,7 @@ eNextState ReAttachIO(AsyncIO *IO, void *pData, int ReadFirst) { + SetEVState(IO, eIOAttach); IO->Data = pData; become_session(IO->CitContext); ev_cleanup_start(event_base, &IO->abort_by_shutdown); @@ -925,7 +1156,8 @@ 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; @@ -962,7 +1194,8 @@ int InitcURLIOStruct(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->SendDone = SendDone; IO->Terminate = Terminate; @@ -976,6 +1209,98 @@ int InitcURLIOStruct(AsyncIO *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; + + SetEVState(IO, eKill); + + if (Ctx->OtherOne->ShutdownAbort != NULL) { + Ctx->OtherOne->NextState = eAbort; + if (Ctx->OtherOne->ShutdownAbort(Ctx->OtherOne) == eDBQuery) { + StopClientWatchers(Ctx->OtherOne, 0); + QueueAnDBOperation(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: + Ctx->IO.ReAttachCB = KillOtherContextNow; + QueueAnEventContext(&Ctx->IO); + break; + case eDBQuery: + Ctx->IO.ReAttachCB = KillOtherContextNow; + QueueAnDBOperation(&Ctx->IO); + 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 @@ -983,14 +1308,17 @@ void EV_backtrace(AsyncIO *IO) 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) + if (strings != NULL) { EV_syslog(LOG_ALERT, " BT %s\n", strings[i]); - else + } + else { EV_syslog(LOG_ALERT, " BT %p\n", stack_frames[i]); + } } free(strings); #endif