From 4fdc7029ccef0500fd8f968413c2b82db8fe2f29 Mon Sep 17 00:00:00 2001 From: Wilfried Goesgens Date: Fri, 6 Apr 2012 17:36:29 +0200 Subject: [PATCH] SMTP-Client: move the client-shutdown procedure into the DB-Thread; we musn't block the network thread for db IO. --- citadel/event_client.c | 12 ++- citadel/event_client.h | 3 + citadel/modules/extnotify/funambol65.c | 7 ++ citadel/modules/network/serv_networkclient.c | 9 ++ citadel/modules/pop3client/serv_pop3client.c | 9 ++ citadel/modules/rssclient/serv_rssclient.c | 13 +++ citadel/modules/smtp/serv_smtpeventclient.c | 96 +++++++++++++++---- citadel/modules/smtp/smtp_clienthandlers.h | 5 + .../urldeshortener/serv_expand_shorter_urls.c | 6 ++ 9 files changed, 135 insertions(+), 25 deletions(-) diff --git a/citadel/event_client.c b/citadel/event_client.c index 5087dd9eb..e2b3ba69a 100644 --- a/citadel/event_client.c +++ b/citadel/event_client.c @@ -127,8 +127,8 @@ void ShutDownDBCLient(AsyncIO *IO) EVM_syslog(LOG_DEBUG, "DBEVENT Terminating.\n"); ev_cleanup_stop(event_db, &IO->db_abort_by_shutdown); - assert(IO->Terminate); - IO->Terminate(IO); + assert(IO->DBTerminate); + IO->DBTerminate(IO); } void @@ -906,6 +906,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) @@ -920,6 +921,7 @@ void InitIOStruct(AsyncIO *IO, IO->SendDone = SendDone; IO->ReadDone = ReadDone; IO->Terminate = Terminate; + IO->DBTerminate = DBTerminate; IO->LineReader = LineReader; IO->ConnFail = ConnFail; IO->Timeout = Timeout; @@ -943,6 +945,7 @@ int InitcURLIOStruct(AsyncIO *IO, const char* Desc, IO_CallBack SendDone, IO_CallBack Terminate, + IO_CallBack DBTerminate, IO_CallBack ShutdownAbort) { IO->Data = Data; @@ -950,8 +953,9 @@ int InitcURLIOStruct(AsyncIO *IO, IO->CitContext = CloneContext(CC); ((CitContext *)IO->CitContext)->session_specific_data = (char*) Data; - IO->SendDone = SendDone; - IO->Terminate = Terminate; + IO->SendDone = SendDone; + IO->Terminate = Terminate; + IO->DBTerminate = DBTerminate; IO->ShutdownAbort = ShutdownAbort; strcpy(IO->HttpReq.errdesc, Desc); diff --git a/citadel/event_client.h b/citadel/event_client.h index 217ffc26e..6ae8b27d3 100644 --- a/citadel/event_client.h +++ b/citadel/event_client.h @@ -146,6 +146,7 @@ struct AsyncIO { IO_CallBack ReadDone, /* Theres new data to read... */ SendDone, /* we may send more data */ Terminate, /* shutting down... */ + DBTerminate, /* shutting down... */ Timeout, /* Timeout handler;may also be conn. timeout */ ConnFail, /* What to do when one connection failed? */ ShutdownAbort,/* we're going down. make your piece. */ @@ -274,6 +275,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); @@ -283,6 +285,7 @@ int InitcURLIOStruct(AsyncIO *IO, const char* Desc, IO_CallBack SendDone, IO_CallBack Terminate, + IO_CallBack DBTerminate, IO_CallBack ShutdownAbort); eNextState ReAttachIO(AsyncIO *IO, diff --git a/citadel/modules/extnotify/funambol65.c b/citadel/modules/extnotify/funambol65.c index 03f478ca6..f4b0d8d38 100644 --- a/citadel/modules/extnotify/funambol65.c +++ b/citadel/modules/extnotify/funambol65.c @@ -49,6 +49,7 @@ eNextState EvaluateResult(AsyncIO *IO); eNextState ExtNotifyTerminate(AsyncIO *IO); +eNextState ExtNotifyTerminateDB(AsyncIO *IO); eNextState ExtNotifyShutdownAbort(AsyncIO *IO); /* @@ -81,6 +82,7 @@ int notify_http_server(char *remoteurl, "Citadel ExtNotify", EvaluateResult, ExtNotifyTerminate, + ExtNotifyTerminateDB, ExtNotifyShutdownAbort)) { syslog(LOG_ALERT, "Unable to initialize libcurl.\n"); @@ -295,6 +297,11 @@ eNextState EvaluateResult(AsyncIO *IO) return 0; } +eNextState ExtNotifyTerminateDB(AsyncIO *IO) +{ + free(IO); + return eAbort; +} eNextState ExtNotifyTerminate(AsyncIO *IO) { free(IO); diff --git a/citadel/modules/network/serv_networkclient.c b/citadel/modules/network/serv_networkclient.c index 0023d8363..51db45198 100644 --- a/citadel/modules/network/serv_networkclient.c +++ b/citadel/modules/network/serv_networkclient.c @@ -157,6 +157,7 @@ void DeleteNetworker(void *vptr) FreeStrBuf(&NW->port); FreeStrBuf(&NW->secret); FreeStrBuf(&NW->Url); + FreeStrBuf(&NW->IO.ErrMsg); FreeAsyncIOContents(&NW->IO); free(NW); } @@ -808,6 +809,13 @@ eNextState NWC_Terminate(AsyncIO *IO) return eAbort; } +eNextState NWC_TerminateDB(AsyncIO *IO) +{ + EVN_syslog(LOG_DEBUG, "%s\n", __FUNCTION__); + FinalizeNetworker(IO); + return eAbort; +} + eNextState NWC_Timeout(AsyncIO *IO) { AsyncNetworker *NW = IO->Data; @@ -881,6 +889,7 @@ void RunNetworker(AsyncNetworker *NW) NWC_DispatchWriteDone, NWC_DispatchReadDone, NWC_Terminate, + NWC_TerminateDB, NWC_ConnFail, NWC_Timeout, NWC_Shutdown); diff --git a/citadel/modules/pop3client/serv_pop3client.c b/citadel/modules/pop3client/serv_pop3client.c index 0a97df2cc..fa151f669 100644 --- a/citadel/modules/pop3client/serv_pop3client.c +++ b/citadel/modules/pop3client/serv_pop3client.c @@ -706,6 +706,14 @@ eNextState POP3_C_Terminate(AsyncIO *IO) FinalizePOP3AggrRun(IO); return eAbort; } +eNextState POP3_C_TerminateDB(AsyncIO *IO) +{ +/// pop3aggr *pMsg = (pop3aggr *)IO->Data; + + syslog(LOG_DEBUG, "POP3: %s\n", __FUNCTION__); + FinalizePOP3AggrRun(IO); + return eAbort; +} eNextState POP3_C_Timeout(AsyncIO *IO) { pop3aggr *pMsg = IO->Data; @@ -883,6 +891,7 @@ int pop3_do_fetching(pop3aggr *cpptr) POP3_C_DispatchWriteDone, POP3_C_DispatchReadDone, POP3_C_Terminate, + POP3_C_TerminateDB, POP3_C_ConnFail, POP3_C_Timeout, POP3_C_Shutdown); diff --git a/citadel/modules/rssclient/serv_rssclient.c b/citadel/modules/rssclient/serv_rssclient.c index 8a3ba6354..6d4392393 100644 --- a/citadel/modules/rssclient/serv_rssclient.c +++ b/citadel/modules/rssclient/serv_rssclient.c @@ -69,6 +69,7 @@ HashList *RSSQueueRooms = NULL; /* rss_room_counter */ HashList *RSSFetchUrls = NULL; /*->rss_aggregator;->RefCount access locked*/ eNextState RSSAggregator_Terminate(AsyncIO *IO); +eNextState RSSAggregator_TerminateDB(AsyncIO *IO); eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO); struct CitContext rss_CC; @@ -180,6 +181,17 @@ eNextState RSSAggregator_Terminate(AsyncIO *IO) return eAbort; } +eNextState RSSAggregator_TerminateDB(AsyncIO *IO) +{ + rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data; + + EVM_syslog(LOG_DEBUG, "RSS: Terminating.\n"); + + + UnlinkRSSAggregator(RSSAggr); + return eAbort; +} + eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO) { const char *pUrl; @@ -294,6 +306,7 @@ int rss_do_fetching(rss_aggregator *Cfg) "Citadel RSS Client", RSSAggregator_ParseReply, RSSAggregator_Terminate, + RSSAggregator_TerminateDB, RSSAggregator_ShutdownAbort)) { syslog(LOG_ALERT, "Unable to initialize libcurl.\n"); diff --git a/citadel/modules/smtp/serv_smtpeventclient.c b/citadel/modules/smtp/serv_smtpeventclient.c index 90a36b235..e5c1cee00 100644 --- a/citadel/modules/smtp/serv_smtpeventclient.c +++ b/citadel/modules/smtp/serv_smtpeventclient.c @@ -117,10 +117,15 @@ eNextState SMTP_C_DispatchReadDone(AsyncIO *IO); eNextState SMTP_C_DispatchWriteDone(AsyncIO *IO); eNextState SMTP_C_DNSFail(AsyncIO *IO); eNextState SMTP_C_Terminate(AsyncIO *IO); +eNextState SMTP_C_TerminateDB(AsyncIO *IO); eReadState SMTP_C_ReadServerStatus(AsyncIO *IO); eNextState mx_connect_ip(AsyncIO *IO); eNextState get_one_mx_host_ip(AsyncIO *IO); +eNextState FinalizeMessageSendDB(AsyncIO *IO); +eNextState FinalizeMessageSend_DB1(AsyncIO *IO); +eNextState FinalizeMessageSend_DB2(AsyncIO *IO); +eNextState FinalizeMessageSend_DB3(AsyncIO *IO); /****************************************************************************** * So, we're finished with sending (regardless of success or failure) * @@ -128,16 +133,12 @@ eNextState get_one_mx_host_ip(AsyncIO *IO); * we need to free the memory and send bounce messages (on terminal failure) * * else we just free our SMTP-Message struct. * ******************************************************************************/ -void FinalizeMessageSend(SmtpOutMsg *Msg) +inline void FinalizeMessageSend_1(AsyncIO *IO) { - int IDestructQueItem; - int nRemain; - StrBuf *MsgData; - AsyncIO *IO = &Msg->IO; - - IDestructQueItem = DecreaseQReference(Msg->MyQItem); + SmtpOutMsg *Msg = IO->Data; + Msg->IDestructQueItem = DecreaseQReference(Msg->MyQItem); - nRemain = CountActiveQueueEntries(Msg->MyQItem); + Msg->nRemain = CountActiveQueueEntries(Msg->MyQItem); if (Msg->MyQEntry->Active && CheckQEntryIsBounce(Msg->MyQEntry)) @@ -146,10 +147,19 @@ void FinalizeMessageSend(SmtpOutMsg *Msg) Msg->MyQItem->SendBounceMail |= (1<MyQEntry->Status); } - if ((nRemain > 0) || IDestructQueItem) - MsgData = SerializeQueueItem(Msg->MyQItem); + if ((Msg->nRemain > 0) || Msg->IDestructQueItem) + Msg->QMsgData = SerializeQueueItem(Msg->MyQItem); else - MsgData = NULL; + Msg->QMsgData = NULL; +} +eNextState FinalizeMessageSend(SmtpOutMsg *Msg) +{ + return QueueDBOperation(&Msg->IO, FinalizeMessageSend_DB1); +} + +inline void FinalizeMessageSend_DB_1(AsyncIO *IO) +{ + SmtpOutMsg *Msg = IO->Data; /* * Uncompleted delivery instructions remain, so delete the old @@ -157,11 +167,35 @@ void FinalizeMessageSend(SmtpOutMsg *Msg) */ EVS_syslog(LOG_DEBUG, "SMTPQD: %ld", Msg->MyQItem->QueMsgID); CtdlDeleteMessages(SMTP_SPOOLOUT_ROOM, &Msg->MyQItem->QueMsgID, 1, ""); +} +eNextState FinalizeMessageSend_DB1(AsyncIO *IO) +{ + FinalizeMessageSend_1(IO); + FinalizeMessageSend_DB_1(IO); + return NextDBOperation(IO, FinalizeMessageSend_DB2); +} + - if (IDestructQueItem) +inline void FinalizeMessageSend_DB_2(AsyncIO *IO) +{ + SmtpOutMsg *Msg = IO->Data; + + if (Msg->IDestructQueItem) smtpq_do_bounce(Msg->MyQItem, Msg->msgtext); +} +eNextState FinalizeMessageSend_DB2(AsyncIO *IO) +{ + FinalizeMessageSend_DB_2(IO); + + return NextDBOperation(IO, FinalizeMessageSend_DB3); +} + + +inline void FinalizeMessageSend_DB_3(AsyncIO *IO) +{ + SmtpOutMsg *Msg = IO->Data; - if (nRemain > 0) + if (Msg->nRemain > 0) { struct CtdlMessage *msg; msg = malloc(sizeof(struct CtdlMessage)); @@ -169,7 +203,7 @@ void FinalizeMessageSend(SmtpOutMsg *Msg) msg->cm_magic = CTDLMESSAGE_MAGIC; msg->cm_anon_type = MES_NORMAL; msg->cm_format_type = FMT_RFC822; - msg->cm_fields['M'] = SmashStrBuf(&MsgData); + msg->cm_fields['M'] = SmashStrBuf(&Msg->QMsgData); msg->cm_fields['U'] = strdup("QMSG"); Msg->MyQItem->QueMsgID = CtdlSubmitMsg(msg, NULL, SMTP_SPOOLOUT_ROOM, QP_EADDR); @@ -181,12 +215,24 @@ void FinalizeMessageSend(SmtpOutMsg *Msg) &Msg->MyQItem->MessageID, 1, ""); - FreeStrBuf(&MsgData); + FreeStrBuf(&Msg->QMsgData); } +} +eNextState FinalizeMessageSend_DB3(AsyncIO *IO) +{ + FinalizeMessageSend_DB_3(IO); + return eAbort; +} + +eNextState FinalizeMessageSend_DB(AsyncIO *IO) +{ + SmtpOutMsg *Msg = IO->Data; + RemoveContext(Msg->IO.CitContext); - if (IDestructQueItem) + if (Msg->IDestructQueItem) RemoveQItem(Msg->MyQItem); DeleteSmtpOutMsg(Msg); + return eAbort; } eNextState FailOneAttempt(AsyncIO *IO) @@ -502,6 +548,7 @@ SmtpOutMsg *new_smtp_outmsg(OneQueItem *MyQItem, SMTP_C_DispatchWriteDone, SMTP_C_DispatchReadDone, SMTP_C_Terminate, + SMTP_C_TerminateDB, SMTP_C_ConnFail, SMTP_C_Timeout, SMTP_C_Shutdown); @@ -562,7 +609,11 @@ void smtp_try_one_queue_entry(OneQueItem *MyQItem, StrBufPlain(Msg->MyQEntry->StatusMessage, HKEY("Invalid Recipient!")); } - FinalizeMessageSend(Msg); + FinalizeMessageSend_1(&Msg->IO); + FinalizeMessageSend_DB_1(&Msg->IO); + FinalizeMessageSend_DB_2(&Msg->IO); + FinalizeMessageSend_DB_3(&Msg->IO); + FinalizeMessageSend_DB(&Msg->IO); } } @@ -651,8 +702,12 @@ eNextState SMTP_C_Terminate(AsyncIO *IO) SmtpOutMsg *Msg = IO->Data; EVS_syslog(LOG_DEBUG, "SMTP: %s\n", __FUNCTION__); - FinalizeMessageSend(Msg); - return eAbort; + return FinalizeMessageSend(Msg); +} +eNextState SMTP_C_TerminateDB(AsyncIO *IO) +{ + EVS_syslog(LOG_DEBUG, "SMTP: %s\n", __FUNCTION__); + return FinalizeMessageSend_DB(IO); } eNextState SMTP_C_Timeout(AsyncIO *IO) { @@ -690,8 +745,7 @@ eNextState SMTP_C_Shutdown(AsyncIO *IO) Msg->MyQEntry->Status = 3; StrBufPlain(Msg->MyQEntry->StatusMessage, HKEY("server shutdown during message submit.")); - FinalizeMessageSend(Msg); - return eAbort; + return FinalizeMessageSend(Msg); } diff --git a/citadel/modules/smtp/smtp_clienthandlers.h b/citadel/modules/smtp/smtp_clienthandlers.h index 02b84e75f..38a7f58f4 100644 --- a/citadel/modules/smtp/smtp_clienthandlers.h +++ b/citadel/modules/smtp/smtp_clienthandlers.h @@ -38,6 +38,8 @@ typedef struct _stmp_out_msg { long n; AsyncIO IO; long CXFlags; + int IDestructQueItem; + int nRemain; eSMTP_C_States State; @@ -57,11 +59,14 @@ typedef struct _stmp_out_msg { ParsedURL *Relay; ParsedURL *pCurrRelay; StrBuf *msgtext; + StrBuf *QMsgData; const char *envelope_from; + char user[1024]; char node[1024]; char name[1024]; char mailfrom[1024]; + long Flags; } SmtpOutMsg; diff --git a/citadel/modules/urldeshortener/serv_expand_shorter_urls.c b/citadel/modules/urldeshortener/serv_expand_shorter_urls.c index bb6d1734a..ef08b6775 100644 --- a/citadel/modules/urldeshortener/serv_expand_shorter_urls.c +++ b/citadel/modules/urldeshortener/serv_expand_shorter_urls.c @@ -103,6 +103,11 @@ eNextState TerminateLookupUrl(AsyncIO *IO) //TOOD return eAbort; } +eNextState TerminateLookupUrlDB(AsyncIO *IO) +{ +//TOOD + return eAbort; +} eNextState LookupUrlResult(AsyncIO *IO) { return eTerminateConnection; /// /TODO @@ -128,6 +133,7 @@ int LookupUrl(StrBuf *ShorterUrlStr) "Citadel RSS ShorterURL Expander", LookupUrlResult, TerminateLookupUrl, + TerminateLookupUrlDB, ShutdownLookuUrl)) { syslog(LOG_ALERT, "Unable to initialize libcurl.\n"); -- 2.30.2