]> code.citadel.org Git - citadel.git/blobdiff - citadel/modules/smtp/serv_smtpqueue.c
Revert "Revert "SMTP-Client: make shure everything is finished before we terminate...
[citadel.git] / citadel / modules / smtp / serv_smtpqueue.c
index f56ad9a9a24e74185b181c9b444ce550f686cb84..92bc9d625d0826398c0dab5636f7395f3cfa37d8 100644 (file)
  * The VRFY and EXPN commands have been removed from this implementation
  * because nobody uses these commands anymore, except for spammers.
  *
- * Copyright (c) 1998-2009 by the citadel.org team
+ * Copyright (c) 1998-2012 by the citadel.org team
  *
- *  This program is free 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 open source software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License version 3.
+ *  
+ *  
  *
  *  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.
  *
- *  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
+ *  
+ *  
+ *  
  */
 
 #include "sysdep.h"
@@ -93,11 +93,23 @@ struct CitContext smtp_queue_CC;
 pthread_mutex_t ActiveQItemsLock;
 HashList *ActiveQItems  = NULL;
 HashList *QItemHandlers = NULL;
+int max_sessions_for_outbound_smtp = 500; /* how many sessions might be active till we stop adding more smtp jobs */
+int ndelay_count = 50; /* every n queued messages we will sleep... */
+int delay_msec = 5000; /* this many seconds. */
 
 static const long MaxRetry = SMTP_RETRY_INTERVAL * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2;
 int MsgCount            = 0;
 int run_queue_now       = 0;   /* Set to 1 to ignore SMTP send retry times */
 
+void RegisterQItemHandler(const char *Key, long Len, QItemHandler H)
+{
+       QItemHandlerStruct *HS = (QItemHandlerStruct*)malloc(sizeof(QItemHandlerStruct));
+       HS->H = H;
+       Put(QItemHandlers, Key, Len, HS, NULL);
+}
+
+
+
 void smtp_try_one_queue_entry(OneQueItem *MyQItem,
                              MailQEntry *MyQEntry,
                              StrBuf *MsgText,
@@ -129,6 +141,22 @@ int DecreaseQReference(OneQueItem *MyQItem)
        return IDestructQueItem;
 }
 
+void DecreaseShutdownDeliveries(OneQueItem *MyQItem)
+{
+       pthread_mutex_lock(&ActiveQItemsLock);
+       MyQItem->NotYetShutdownDeliveries--;
+       pthread_mutex_unlock(&ActiveQItemsLock);
+}
+
+int GetShutdownDeliveries(OneQueItem *MyQItem)
+{
+       int DestructNow;
+
+       pthread_mutex_lock(&ActiveQItemsLock);
+       DestructNow = MyQItem->ActiveDeliveries == 0;
+       pthread_mutex_unlock(&ActiveQItemsLock);
+       return DestructNow;
+}
 void RemoveQItem(OneQueItem *MyQItem)
 {
        long len;
@@ -158,16 +186,28 @@ void RemoveQItem(OneQueItem *MyQItem)
 void FreeMailQEntry(void *qv)
 {
        MailQEntry *Q = qv;
+/*
+       syslog(LOG_DEBUG, "---------------%s--------------", __FUNCTION__);
+       cit_backtrace();
+*/
        FreeStrBuf(&Q->Recipient);
        FreeStrBuf(&Q->StatusMessage);
+
+       memset(Q, 0, sizeof(MailQEntry));
        free(Q);
 }
 void FreeQueItem(OneQueItem **Item)
 {
+/*
+       syslog(LOG_DEBUG, "---------------%s--------------", __FUNCTION__);
+       cit_backtrace();
+*/
        DeleteHash(&(*Item)->MailQEntries);
        FreeStrBuf(&(*Item)->EnvelopeFrom);
        FreeStrBuf(&(*Item)->BounceTo);
+       FreeStrBuf(&(*Item)->SenderRoom);
        FreeURL(&(*Item)->URL);
+       memset(*Item, 0, sizeof(OneQueItem));
        free(*Item);
        Item = NULL;
 }
@@ -181,6 +221,29 @@ void HFreeQueItem(void *Item)
  * - 3/4 (transient errors
  *        were experienced and it's time to try again)
  */
+int CheckQEntryActive(MailQEntry *ThisItem)
+{
+       if ((ThisItem->Status == 0) ||
+           (ThisItem->Status == 3) ||
+           (ThisItem->Status == 4))
+       {
+               return 1;
+       }
+       else
+               return 0;
+}
+int CheckQEntryIsBounce(MailQEntry *ThisItem)
+{
+       if ((ThisItem->Status == 3) ||
+           (ThisItem->Status == 4) ||
+           (ThisItem->Status == 5))
+       {
+               return 1;
+       }
+       else
+               return 0;
+}      
+
 int CountActiveQueueEntries(OneQueItem *MyQItem)
 {
        HashPos  *It;
@@ -194,9 +257,8 @@ int CountActiveQueueEntries(OneQueItem *MyQItem)
        while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
        {
                MailQEntry *ThisItem = vQE;
-               if ((ThisItem->Status == 0) ||
-                   (ThisItem->Status == 3) ||
-                   (ThisItem->Status == 4))
+
+               if (CheckQEntryActive(ThisItem))
                {
                        ActiveDeliveries++;
                        ThisItem->Active = 1;
@@ -233,9 +295,9 @@ OneQueItem *DeserializeQueueItem(StrBuf *RawQItem, long QueMsgID)
                StrBufExtract_NextToken(Token, Line, &pItemPart, '|');
                if (GetHash(QItemHandlers, SKEY(Token), &vHandler))
                {
-                       QItemHandler H;
-                       H = (QItemHandler) vHandler;
-                       H(Item, Line, &pItemPart);
+                       QItemHandlerStruct *HS;
+                       HS = (QItemHandlerStruct*) vHandler;
+                       HS->H(Item, Line, &pItemPart);
                }
        }
        FreeStrBuf(&Line);
@@ -293,24 +355,24 @@ StrBuf *SerializeQueueItem(OneQueItem *MyQItem)
                StrBufAppendBuf(QMessage, MyQItem->EnvelopeFrom, 0);
        }
 
+       if (StrLength(MyQItem->SenderRoom) > 0) {
+               StrBufAppendBufPlain(QMessage, HKEY("\nsource_room|"), 0);
+               StrBufAppendBuf(QMessage, MyQItem->SenderRoom, 0);
+       }
+
        StrBufAppendBufPlain(QMessage, HKEY("\nretry|"), 0);
        StrBufAppendPrintf(QMessage, "%ld",
                           MyQItem->Retry);
 
        StrBufAppendBufPlain(QMessage, HKEY("\nattempted|"), 0);
        StrBufAppendPrintf(QMessage, "%ld",
-                          MyQItem->ReattemptWhen);
+                          time(NULL) /*ctdl_ev_now()*/ + MyQItem->Retry);
 
        It = GetNewHashPos(MyQItem->MailQEntries, 0);
        while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
        {
                MailQEntry *ThisItem = vQE;
 
-               if (!ThisItem->Active)
-               {
-                       /* skip already sent ones from the spoolfile. */
-                       continue;
-               }
                StrBufAppendBufPlain(QMessage, HKEY("\nremote|"), 0);
                StrBufAppendBuf(QMessage, ThisItem->Recipient, 0);
                StrBufAppendBufPlain(QMessage, HKEY("|"), 0);
@@ -361,6 +423,13 @@ void QItem_Handle_BounceTo(OneQueItem *Item, StrBuf *Line, const char **Pos)
        StrBufExtract_NextToken(Item->BounceTo, Line, Pos, '|');
 }
 
+void QItem_Handle_SenderRoom(OneQueItem *Item, StrBuf *Line, const char **Pos)
+{
+       if (Item->SenderRoom == NULL)
+               Item->SenderRoom = NewStrBufPlain(NULL, StrLength(Line));
+       StrBufExtract_NextToken(Item->SenderRoom, Line, Pos, '|');
+}
+
 void QItem_Handle_Recipient(OneQueItem *Item, StrBuf *Line, const char **Pos)
 {
        if (Item->Current == NULL)
@@ -378,7 +447,10 @@ void QItem_Handle_retry(OneQueItem *Item, StrBuf *Line, const char **Pos)
 {
        Item->Retry =
                StrBufExtractNext_int(Line, Pos, '|');
-       Item->Retry *= 2;
+       if (Item->Retry == 0)
+               Item->Retry = SMTP_RETRY_INTERVAL;
+       else
+               Item->Retry *= 2;
 }
 
 
@@ -439,36 +511,46 @@ void smtpq_do_bounce(OneQueItem *MyQItem, StrBuf *OMsgTxt)
        StrBuf *Msg = NULL;
        StrBuf *BounceMB;
        struct recptypes *valid;
+       time_t now;
 
        HashPos *It;
        void *vQE;
        long len;
        const char *Key;
 
+       int first_attempt = 0;
        int successful_bounce = 0;
        int num_bounces = 0;
        int give_up = 0;
 
        syslog(LOG_DEBUG, "smtp_do_bounce() called\n");
 
-       if ( (ev_time() - MyQItem->Submitted) > SMTP_GIVE_UP ) {
-               give_up = 1;/// TODO: replace time by libevq timer get
+       if (MyQItem->SendBounceMail == 0)
+               return;
+
+       now = time (NULL); //ev_time();
+
+       if ( (now - MyQItem->Submitted) > SMTP_GIVE_UP ) {
+               give_up = 1;
+       }
+
+       if (MyQItem->Retry == SMTP_RETRY_INTERVAL) {
+               first_attempt = 1;
        }
 
        /*
         * Now go through the instructions checking for stuff.
         */
+       Msg = NewStrBufPlain(NULL, 1024);
        It = GetNewHashPos(MyQItem->MailQEntries, 0);
        while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
        {
                MailQEntry *ThisItem = vQE;
-               if ((ThisItem->Status == 5) || /* failed now? */
-                   ((give_up == 1) &&
-                    (ThisItem->Status != 2)))
+               if ((ThisItem->Active && (ThisItem->Status == 5)) || /* failed now? */
+                   ((give_up == 1) && (ThisItem->Status != 2)) ||
+                   ((first_attempt == 1) && (ThisItem->Status != 2)))
                        /* giving up after failed attempts... */
                {
-                       if (num_bounces == 0)
-                               Msg = NewStrBufPlain(NULL, 1024);
                        ++num_bounces;
 
                        StrBufAppendBuf(Msg, ThisItem->Recipient, 0);
@@ -552,8 +634,19 @@ void smtpq_do_bounce(OneQueItem *MyQItem, StrBuf *OMsgTxt)
        StrBufAppendBuf(BounceMB, Msg, 0);
        FreeStrBuf(&Msg);
 
+       if (StrLength(MyQItem->SenderRoom) > 0)
+       {
+               StrBufAppendBufPlain(
+                       BounceMB,
+                       HKEY("The message was originaly posted in: "), 0);
+               StrBufAppendBuf(BounceMB, MyQItem->SenderRoom, 0);
+               StrBufAppendBufPlain(
+                       BounceMB,
+                       HKEY("\n"), 0);
+       }
+
        /* Attach the original message */
-       StrBufAppendBufPlain(BounceMB, HKEY("--"), 0);
+       StrBufAppendBufPlain(BounceMB, HKEY("\r\n--"), 0);
        StrBufAppendBuf(BounceMB, boundary, 0);
        StrBufAppendBufPlain(BounceMB, HKEY("\r\n"), 0);
        StrBufAppendBufPlain(BounceMB,
@@ -612,6 +705,8 @@ void smtpq_do_bounce(OneQueItem *MyQItem, StrBuf *OMsgTxt)
  * Called by smtp_do_queue() to handle an individual message.
  */
 void smtp_do_procmsg(long msgnum, void *userdata) {
+       time_t now;
+       int mynumsessions = num_sessions;
        struct CtdlMessage *msg = NULL;
        char *instr = NULL;
        StrBuf *PlainQItem;
@@ -626,6 +721,13 @@ void smtp_do_procmsg(long msgnum, void *userdata) {
        int HaveBuffers = 0;
        StrBuf *Msg =NULL;
 
+       if (mynumsessions > max_sessions_for_outbound_smtp) {
+               syslog(LOG_DEBUG,
+                      "SMTP Queue: skipping because of num jobs %d > %d max_sessions_for_outbound_smtp",
+                      mynumsessions,
+                      max_sessions_for_outbound_smtp);
+       }
+
        syslog(LOG_DEBUG, "SMTP Queue: smtp_do_procmsg(%ld)\n", msgnum);
        ///strcpy(envelope_from, "");
 
@@ -661,11 +763,12 @@ void smtp_do_procmsg(long msgnum, void *userdata) {
        /*
         * Postpone delivery if we've already tried recently.
         */
+       now = time(NULL);
        if ((MyQItem->ReattemptWhen != 0) && 
-           (time(NULL) < MyQItem->ReattemptWhen) &&
+           (now < MyQItem->ReattemptWhen) &&
            (run_queue_now == 0))
        {
-               syslog(LOG_DEBUG, "SMTP client: Retry time not yet reached.\n");
+               syslog(LOG_DEBUG, "SMTP client: Retry time not yet reached. %ld seconds left.",  MyQItem->ReattemptWhen - now);
 
                It = GetNewHashPos(MyQItem->MailQEntries, 0);
                pthread_mutex_lock(&ActiveQItemsLock);
@@ -769,9 +872,34 @@ void smtp_do_procmsg(long msgnum, void *userdata) {
        }
        DeleteHashPos(&It);
 
+       MyQItem->NotYetShutdownDeliveries = 
        MyQItem->ActiveDeliveries = CountActiveQueueEntries(MyQItem);
+
+       /* failsafe against overload: 
+        * will we exceed the limit set? 
+        */
+       if ((MyQItem->ActiveDeliveries + mynumsessions > max_sessions_for_outbound_smtp) && 
+           /* if yes, did we reach more than half of the quota? */
+           ((mynumsessions * 2) > max_sessions_for_outbound_smtp) && 
+           /* if... would we ever fit into half of the quota?? */
+           (((MyQItem->ActiveDeliveries * 2)  < max_sessions_for_outbound_smtp)))
+       {
+               /* abort delivery for another time. */
+               syslog(LOG_DEBUG,
+                      "SMTP Queue: skipping because of num jobs %d + %ld > %d max_sessions_for_outbound_smtp",
+                      mynumsessions,
+                      MyQItem->ActiveDeliveries,
+                      max_sessions_for_outbound_smtp);
+
+               FreeQueItem(&MyQItem);
+
+               return;
+       }
+
+
        if (MyQItem->ActiveDeliveries > 0)
        {
+               int nActivated = 0;
                int n = MsgCount++;
                int m = MyQItem->ActiveDeliveries;
                int i = 1;
@@ -786,6 +914,11 @@ void smtp_do_procmsg(long msgnum, void *userdata) {
                        if (ThisItem->Active == 1)
                        {
                                int KeepBuffers = (i == m);
+
+                               nActivated++;
+                               if (nActivated % ndelay_count == 0)
+                                       usleep(delay_msec);
+
                                if (i > 1) n = MsgCount++;
                                syslog(LOG_DEBUG,
                                       "SMTPQ: Trying <%ld> <%s> %d / %d \n",
@@ -904,7 +1037,7 @@ void smtp_init_spoolout(void) {
         * Create the room.  This will silently fail if the room already
         * exists, and that's perfectly ok, because we want it to exist.
         */
-       CtdlCreateRoom(SMTP_SPOOLOUT_ROOM, 3, "", 0, 1, 0, VIEW_MAILBOX);
+       CtdlCreateRoom(SMTP_SPOOLOUT_ROOM, 3, "", 0, 1, 0, VIEW_QUEUE);
 
        /*
         * Make sure it's set to be a "system room" so it doesn't show up
@@ -962,25 +1095,43 @@ void cmd_smtp(char *argbuf) {
 
 CTDL_MODULE_INIT(smtp_queu)
 {
+       char *pstr;
+
        if (!threading)
        {
+               pstr = getenv("CITSERVER_n_session_max");
+               if ((pstr != NULL) && (*pstr != '\0'))
+                       max_sessions_for_outbound_smtp = atol(pstr); /* how many sessions might be active till we stop adding more smtp jobs */
+
+               pstr = getenv("CITSERVER_smtp_n_delay_count");
+               if ((pstr != NULL) && (*pstr != '\0'))
+                       ndelay_count = atol(pstr); /* every n queued messages we will sleep... */
+
+               pstr = getenv("CITSERVER_smtp_delay");
+               if ((pstr != NULL) && (*pstr != '\0'))
+                       delay_msec = atol(pstr) * 1000; /* this many seconds. */
+
+
+
+
                CtdlFillSystemContext(&smtp_queue_CC, "SMTP_Send");
                ActiveQItems = NewHash(1, lFlathash);
                pthread_mutex_init(&ActiveQItemsLock, NULL);
 
                QItemHandlers = NewHash(0, NULL);
 
-               Put(QItemHandlers, HKEY("msgid"), QItem_Handle_MsgID, reference_free_handler);
-               Put(QItemHandlers, HKEY("envelope_from"), QItem_Handle_EnvelopeFrom, reference_free_handler);
-               Put(QItemHandlers, HKEY("retry"), QItem_Handle_retry, reference_free_handler);
-               Put(QItemHandlers, HKEY("attempted"), QItem_Handle_Attempted, reference_free_handler);
-               Put(QItemHandlers, HKEY("remote"), QItem_Handle_Recipient, reference_free_handler);
-               Put(QItemHandlers, HKEY("bounceto"), QItem_Handle_BounceTo, reference_free_handler);
-               Put(QItemHandlers, HKEY("submitted"), QItem_Handle_Submitted, reference_free_handler);
+               RegisterQItemHandler(HKEY("msgid"),             QItem_Handle_MsgID);
+               RegisterQItemHandler(HKEY("envelope_from"),     QItem_Handle_EnvelopeFrom);
+               RegisterQItemHandler(HKEY("retry"),             QItem_Handle_retry);
+               RegisterQItemHandler(HKEY("attempted"),         QItem_Handle_Attempted);
+               RegisterQItemHandler(HKEY("remote"),            QItem_Handle_Recipient);
+               RegisterQItemHandler(HKEY("bounceto"),          QItem_Handle_BounceTo);
+               RegisterQItemHandler(HKEY("source_room"),       QItem_Handle_SenderRoom);
+               RegisterQItemHandler(HKEY("submitted"),         QItem_Handle_Submitted);
 
                smtp_init_spoolout();
 
-               CtdlRegisterCleanupHook(smtp_evq_cleanup);
+               CtdlRegisterEVCleanupHook(smtp_evq_cleanup);
 
                CtdlRegisterProtoHook(cmd_smtp, "SMTP", "SMTP utility commands");
                CtdlRegisterSessionHook(smtp_do_queue, EVT_TIMER);