X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fevent_client.c;h=c633f5357a1e845520eec5d9eee2a8f86439e696;hb=149b7a41b77a3cc0752fe88090e60ce7dae90a85;hp=79a2f0f9c213feb6d4b8517f0de0d7cbef88aae3;hpb=aa7365c86de8e26e796d3aa3fd605c85d8c26220;p=citadel.git diff --git a/citadel/event_client.c b/citadel/event_client.c index 79a2f0f9c..c633f5357 100644 --- a/citadel/event_client.c +++ b/citadel/event_client.c @@ -298,6 +298,7 @@ void FreeAsyncIOContents(AsyncIO *IO) if (Ctx) { Ctx->state = CON_IDLE; Ctx->kill_me = 1; + IO->CitContext = NULL; } } @@ -315,8 +316,8 @@ void StopClientWatchers(AsyncIO *IO, int CloseFD) if (CloseFD && (IO->SendBuf.fd > 0)) { close(IO->SendBuf.fd); - IO->SendBuf.fd = -1; - IO->RecvBuf.fd = -1; + IO->SendBuf.fd = 0; + IO->RecvBuf.fd = 0; } } @@ -369,7 +370,22 @@ 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); + default: + break; + } break; case eReadPayload: case eReadMore: @@ -580,12 +596,12 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents) if (errno != EAGAIN) { StopClientWatchers(IO, 1); EV_syslog(LOG_DEBUG, - "EVENT: Socket Invalid! [%d] [%s] [%d]\n", + "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.0); + SetNextTimeout(IO, 0.01); } } /* else : must write more. */ @@ -805,19 +821,20 @@ IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents) if (nbytes > 0) { HandleInbound(IO); } else if (nbytes == 0) { - SetNextTimeout(IO, 0.0); + StopClientWatchers(IO, 1); + SetNextTimeout(IO, 0.01); return; } else if (nbytes == -1) { if (errno != EAGAIN) { // FD is gone. kick it. StopClientWatchers(IO, 1); EV_syslog(LOG_DEBUG, - "EVENT: Socket Invalid! [%d] [%s] [%d]\n", + "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.0); + SetNextTimeout(IO, 0.01); } return; } @@ -886,10 +903,12 @@ eNextState EvConnectSock(AsyncIO *IO, fdflags = fcntl(IO->SendBuf.fd, F_GETFL); if (fdflags < 0) { EV_syslog(LOG_ERR, - "EVENT: unable to get socket flags! %s \n", + "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; @@ -899,7 +918,8 @@ eNextState EvConnectSock(AsyncIO *IO, if (fcntl(IO->SendBuf.fd, F_SETFL, fdflags) < 0) { EV_syslog( LOG_ERR, - "EVENT: unable to set socket nonblocking flags! %s \n", + "EVENT: unable to set socket %d nonblocking flags! %s \n", + IO->SendBuf.fd, strerror(errno)); StrBufPrintf(IO->ErrMsg, "Failed to set socket flags: %s", @@ -956,12 +976,12 @@ eNextState EvConnectSock(AsyncIO *IO, } 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); 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, @@ -980,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)); @@ -1030,7 +1054,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; @@ -1067,7 +1092,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; @@ -1081,6 +1107,85 @@ int InitcURLIOStruct(AsyncIO *IO, } + +typedef struct KillOtherSessionContext { + AsyncIO IO; + AsyncIO *OtherOne; +}KillOtherSessionContext; + +eNextState KillTerminate(AsyncIO *IO) +{ + KillOtherSessionContext *Ctx = (KillOtherSessionContext*)IO->Data; + EV_syslog(LOG_DEBUG, "%s Exit\n", __FUNCTION__); + FreeAsyncIOContents(IO); + memset(Ctx, 0, sizeof(KillOtherSessionContext)); + 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) {