OSEC

Neohapsis is currently accepting applications for employment. For more information, please visit our website www.neohapsis.com or email hr@neohapsis.com
Postfix Archives: [patch] for separate per-entry rcpt lists

[patch] for separate per-entry rcpt lists

Patrik Rak (patrikein.cz)
Mon, 22 Nov 1999 21:02:14 +0100 (CET)

Hi!

Some time ago I posted a message to this list about possible problems
caused by messages with large number of recipients. Such message(s) can
completely block delivery of any other messages for significant
amount of time.

The patch below attempts to defeat this. If you are technically inclined
and/or want to know how this is accomplished, please read the notes at the
end of this message.

Index: postfix/qmgr/qmgr.h
diff -u postfix/qmgr/qmgr.h:1.1 postfix/qmgr/qmgr.h:1.2
--- postfix/qmgr/qmgr.h:1.1 Tue Aug 31 20:08:37 1999
+++ postfix/qmgr/qmgr.h Sun Nov 21 22:10:21 1999
-179,17 +179,20
     long offset; /* REC_TYPE_RCPT byte */
     char *address; /* complete address */
     QMGR_QUEUE *queue; /* resolved queue */
+ QMGR_RCPT *next; /* linkage */
 };
 
 struct QMGR_RCPT_LIST {
- QMGR_RCPT *info;
+ QMGR_RCPT *rcpt; /* first of the list */
     int len;
- int avail;
 };
 
 extern void qmgr_rcpt_list_init(QMGR_RCPT_LIST *);
 extern void qmgr_rcpt_list_add(QMGR_RCPT_LIST *, long, const char *);
 extern void qmgr_rcpt_list_free(QMGR_RCPT_LIST *);
+
+extern void qmgr_rcpt_list_free_first(QMGR_RCPT_LIST *);
+extern void qmgr_rcpt_list_move_first(QMGR_RCPT_LIST *, QMGR_RCPT_LIST *);
 
  /*
   * Structure of one next-hop queue entry. In order to save some copying
Index: postfix/qmgr/qmgr_defer.c
diff -u postfix/qmgr/qmgr_defer.c:1.1 postfix/qmgr/qmgr_defer.c:1.2
--- postfix/qmgr/qmgr_defer.c:1.1 Fri Dec 11 19:55:45 1998
+++ postfix/qmgr/qmgr_defer.c Sun Nov 21 22:10:22 1999
-113,7 +113,6
     QMGR_ENTRY *next;
     QMGR_MESSAGE *message;
     QMGR_RCPT *recipient;
- int nrcpt;
 
     /*
      * Sanity checks.
-129,8 +128,7
     for (entry = queue->todo.next; entry != 0; entry = next) {
         next = entry->peers.next;
         message = entry->message;
- for (nrcpt = 0; nrcpt < entry->rcpt_list.len; nrcpt++) {
- recipient = entry->rcpt_list.info + nrcpt;
+ for (recipient = entry->rcpt_list.rcpt; recipient; recipient = recipient->next) {
             qmgr_defer_recipient(message, recipient->address, reason);
         }
         qmgr_entry_done(entry, QMGR_QUEUE_TODO);
Index: postfix/qmgr/qmgr_deliver.c
diff -u postfix/qmgr/qmgr_deliver.c:1.1 postfix/qmgr/qmgr_deliver.c:1.2
--- postfix/qmgr/qmgr_deliver.c:1.1 Sat Mar 20 01:45:02 1999
+++ postfix/qmgr/qmgr_deliver.c Sun Nov 21 22:10:22 1999
-113,7 +113,6
 
 static int qmgr_deliver_send_request(QMGR_ENTRY *entry, VSTREAM *stream)
 {
- QMGR_RCPT_LIST list = entry->rcpt_list;
     QMGR_RCPT *recipient;
     QMGR_MESSAGE *message = entry->message;
 
-123,7 +122,7
                entry->queue->name, message->sender,
                message->errors_to, message->return_receipt,
                message->arrival_time);
- for (recipient = list.info; recipient < list.info + list.len; recipient++)
+ for (recipient = entry->rcpt_list.rcpt; recipient; recipient = recipient->next)
         mail_print(stream, "%ld %s", recipient->offset, recipient->address);
     mail_print(stream, "%s", "0");
     if (vstream_fflush(stream) != 0) {
Index: postfix/qmgr/qmgr_entry.c
diff -u postfix/qmgr/qmgr_entry.c:1.1 postfix/qmgr/qmgr_entry.c:1.2
--- postfix/qmgr/qmgr_entry.c:1.1 Sat Feb 6 23:06:45 1999
+++ postfix/qmgr/qmgr_entry.c Sun Nov 21 22:10:22 1999
-135,6 +135,13
     } else {
         msg_panic("qmgr_entry_done: bad queue spec: %d", which);
     }
+
+ /*
+ * Free the recipient list and decrease in-core recipient count accordingly.
+ */
+ qmgr_recipient_count -= entry->rcpt_list.len;
+ qmgr_rcpt_list_free(&entry->rcpt_list);
+
     myfree((char *) entry);
 
     /*
-174,8 +181,7
     entry = (QMGR_ENTRY *) mymalloc(sizeof(QMGR_ENTRY));
     entry->stream = 0;
     entry->message = message;
- entry->rcpt_list.len = 0;
- entry->rcpt_list.info = 0;
+ qmgr_rcpt_list_init(&entry->rcpt_list);
     message->refcount++;
     entry->queue = queue;
     QMGR_LIST_APPEND(queue->todo, entry);
Index: postfix/qmgr/qmgr_message.c
diff -u postfix/qmgr/qmgr_message.c:1.1 postfix/qmgr/qmgr_message.c:1.2
--- postfix/qmgr/qmgr_message.c:1.1 Sat Sep 4 22:37:30 1999
+++ postfix/qmgr/qmgr_message.c Sun Nov 21 22:10:22 1999
-206,12 +206,11
      * already looked at, and reset the in-core recipient address list.
      */
     if (message->rcpt_offset) {
+ if (message->rcpt_list.rcpt || message->rcpt_list.len)
+ msg_panic("%s: recipient list not empty on recipient reload", message->queue_id);
         if (vstream_fseek(message->fp, message->rcpt_offset, SEEK_SET) < 0)
             msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
         message->rcpt_offset = 0;
- qmgr_recipient_count -= message->rcpt_list.len;
- qmgr_rcpt_list_free(&message->rcpt_list);
- qmgr_rcpt_list_init(&message->rcpt_list);
     }
 
     /*
-281,8 +280,6
         }
     } while (rec_type > 0 && rec_type != REC_TYPE_END);
 
- qmgr_recipient_count += message->rcpt_list.len;
-
     /*
      * If there is no size record, use the queue file size instead.
      */
-384,18 +381,99
     return (strcasecmp(rcpt1->address, rcpt2->address));
 }
 
+/* qmgr_message_sort_merge - merge two sorted recipient lists together */
+
+static void qmgr_message_sort_merge( QMGR_RCPT ** first, QMGR_RCPT ** last,
+ QMGR_RCPT * f1, QMGR_RCPT * l1,
+ QMGR_RCPT * f2, QMGR_RCPT * l2,
+ QMGR_RCPT * next )
+{
+ QMGR_RCPT ** pp;
+
+ /*
+ * Try joining the lists if possible.
+ */
+ if( qmgr_message_sort_compare( l1, f2 ) <= 0 ) {
+ *first = f1;
+ l1->next = f2;
+ *last = l2;
+ l2->next = next;
+ return;
+ }
+ if( qmgr_message_sort_compare( l2, f1 ) <= 0 ) {
+ *first = f2;
+ l2->next = f1;
+ *last = l1;
+ l1->next = next;
+ return;
+ }
+
+ /*
+ * It didn't work, so let's do it the hard way - merge the lists.
+ */
+ pp = first;
+ for( ; ; ) {
+ if( qmgr_message_sort_compare( f1, f2 ) < 0 ) {
+ *pp = f1;
+ pp = &f1->next;
+ if( f1 != l1 ) {
+ f1 = f1->next;
+ }
+ else {
+ *pp = f2;
+ *last = l2;
+ l2->next = next;
+ break;
+ }
+ }
+ else {
+ *pp = f2;
+ pp = &f2->next;
+ if( f2 != l2 ) {
+ f2 = f2->next;
+ }
+ else {
+ *pp = f1;
+ *last = l1;
+ l1->next = next;
+ break;
+ }
+ }
+ }
+}
+
+/* qmgr_message_mergesort - sort specified part of the recipient list */
+
+static void qmgr_message_mergesort( QMGR_RCPT ** first, QMGR_RCPT ** last, QMGR_RCPT * list, int len )
+{
+ QMGR_RCPT *f1, *l1, *f2, *l2;
+ int half;
+
+ if( len < 2 ) {
+ *first = *last = list;
+ return;
+ }
+
+ half = len / 2;
+ qmgr_message_mergesort( &f1, &l1, list, half );
+ qmgr_message_mergesort( &f2, &l2, l1->next, len - half );
+ qmgr_message_sort_merge( first, last, f1, l1, f2, l2, l2->next );
+}
+
 /* qmgr_message_sort - sort message recipient addresses by domain */
 
 static void qmgr_message_sort(QMGR_MESSAGE *message)
 {
- qsort((char *) message->rcpt_list.info, message->rcpt_list.len,
- sizeof(message->rcpt_list.info[0]), qmgr_message_sort_compare);
+ QMGR_RCPT_LIST *list = &message->rcpt_list;
+ QMGR_RCPT *dummy;
+
+ qmgr_message_mergesort(&list->rcpt, &dummy, list->rcpt, list->len);
+
     if (msg_verbose) {
- QMGR_RCPT_LIST list = message->rcpt_list;
         QMGR_RCPT *rcpt;
 
         msg_info("start sorted recipient list");
- for (rcpt = list.info; rcpt < list.info + list.len; rcpt++)
+ for (rcpt = list->rcpt; rcpt; rcpt = rcpt->next)
             msg_info("qmgr_message_sort: %s", rcpt->address);
         msg_info("end sorted recipient list");
     }
-406,7 +484,6
 static void qmgr_message_resolve(QMGR_MESSAGE *message)
 {
     static ARGV *defer_xport_argv;
- QMGR_RCPT_LIST list = message->rcpt_list;
     QMGR_RCPT *recipient;
     QMGR_TRANSPORT *transport = 0;
     QMGR_QUEUE *queue = 0;
-422,7 +499,7
 #define UPDATE(ptr,new) { myfree(ptr); ptr = mystrdup(new); }
 
     resolve_clnt_init(&reply);
- for (recipient = list.info; recipient < list.info + list.len; recipient++) {
+ for (recipient = message->rcpt_list.rcpt; recipient; recipient = recipient->next) {
 
         /*
          * This may be a bit late in the game, but it is the most convenient
-593,7 +670,7
 
 static void qmgr_message_assign(QMGR_MESSAGE *message)
 {
- QMGR_RCPT_LIST list = message->rcpt_list;
+ QMGR_RCPT_LIST *list = &message->rcpt_list;
     QMGR_RCPT *recipient;
     QMGR_ENTRY *entry = 0;
     QMGR_QUEUE *queue;
-601,24 +678,25
     /*
      * Try to bundle as many recipients in a delivery request as we can. When
      * the recipient resolves to the same site and transport as the previous
- * recipient, do not create a new queue entry, just bump the count of
- * recipients for the existing queue entry. All this provided that we do
+ * recipient, do not create a new queue entry, just move that recipient
+ * to the recipient list of the existing queue entry. All this provided that we do
      * not exceed the transport-specific limit on the number of recipients
- * per transaction. Skip recipients with a dead transport or destination.
+ * per transaction. Dispose of the recipients with a dead transport or destination.
      */
 #define LIMIT_OK(limit, count) ((limit) == 0 || ((count) < (limit)))
 
- for (recipient = list.info; recipient < list.info + list.len; recipient++) {
+ while ((recipient = list->rcpt) != 0) {
         if ((queue = recipient->queue) != 0) {
             if (message->single_rcpt || entry == 0 || entry->queue != queue
                 || !LIMIT_OK(entry->queue->transport->recipient_limit,
                              entry->rcpt_list.len)) {
                 entry = qmgr_entry_create(queue, message);
- entry->rcpt_list.info = recipient;
- entry->rcpt_list.len = 1;
- } else {
- entry->rcpt_list.len++;
             }
+ qmgr_rcpt_list_move_first(&entry->rcpt_list,list);
+ qmgr_recipient_count++;
+ }
+ else {
+ qmgr_rcpt_list_free_first(list);
         }
     }
 }
-639,7 +717,6
         myfree(message->errors_to);
     if (message->return_receipt)
         myfree(message->return_receipt);
- qmgr_recipient_count -= message->rcpt_list.len;
     qmgr_rcpt_list_free(&message->rcpt_list);
     qmgr_message_count--;
     myfree((char *) message);
Index: postfix/qmgr/qmgr_rcpt_list.c
diff -u postfix/qmgr/qmgr_rcpt_list.c:1.1 postfix/qmgr/qmgr_rcpt_list.c:1.2
--- postfix/qmgr/qmgr_rcpt_list.c:1.1 Fri Dec 11 19:55:45 1998
+++ postfix/qmgr/qmgr_rcpt_list.c Sun Nov 21 22:10:22 1999
-16,6 +16,14
 /*
 /* void qmgr_rcpt_list_free(list)
 /* QMGR_RCPT_LIST *list;
+/*
+/* void qmgr_rcpt_list_free_first(list)
+/* QMGR_RCPT_LIST *list;
+/*
+/* void qmgr_rcpt_list_move_first(destination,source)
+/* QMGR_RCPT_LIST *destination;
+/* QMGR_RCPT_LIST *source;
+/*
 /* DESCRIPTION
 /* This module maintains lists of queue manager recipient structures.
 /* These structures are extended versions of the structures maintained
-31,7 +39,18
 /* The recipient name is copied.
 /*
 /* qmgr_rcpt_list_free() releases memory for the specified list
-/* of recipient structures.
+/* of recipient structures. Actually, the structure is kept in cache
+/* for later use, as the total number of these structures is known
+/* to be currently bounded by var_qmgr_rcpt_limit.
+/*
+/* qmgr_rcpt_list_free_first() removes first recipient of the specified
+/* list and releases memory of its structure. It is an error to call
+/* this function with an empty list.
+/*
+/* qmgr_rcpt_list_move_first() moves first recipient from one specified
+/* list to another. It is an error to call this function with an empty
+/* source list.
+/*
 /* SEE ALSO
 /* qmgr_rcpt_list(3h) data structure
 /* recipient_list(3) same code, different data structure.
-60,37 +79,80
 
 #include "qmgr.h"
 
+/* qmgr_rcpt_cache - pool of available QMGR_RCPT structures */
+
+static QMGR_RCPT *qmgr_rcpt_cache;
+
 /* qmgr_rcpt_list_init - initialize */
 
 void qmgr_rcpt_list_init(QMGR_RCPT_LIST *list)
 {
- list->avail = 1;
+ list->rcpt = 0;
     list->len = 0;
- list->info = (QMGR_RCPT *) mymalloc(sizeof(QMGR_RCPT));
 }
 
 /* qmgr_rcpt_list_add - add rcpt to list */
 
-void qmgr_rcpt_list_add(QMGR_RCPT_LIST *list, long offset, const char *rcpt)
+void qmgr_rcpt_list_add(QMGR_RCPT_LIST *list, long offset, const char *address)
 {
- if (list->len >= list->avail) {
- list->avail *= 2;
- list->info = (QMGR_RCPT *)
- myrealloc((char *) list->info, list->avail * sizeof(QMGR_RCPT));
+ QMGR_RCPT *rcpt;
+
+ if (qmgr_rcpt_cache != 0) {
+ rcpt = qmgr_rcpt_cache;
+ qmgr_rcpt_cache = qmgr_rcpt_cache->next;
+ }
+ else {
+ rcpt = (QMGR_RCPT *) mymalloc(sizeof(QMGR_RCPT));
     }
- list->info[list->len].address = mystrdup(rcpt);
- list->info[list->len].offset = offset;
- list->info[list->len].queue = 0;
+ rcpt->address = mystrdup(address);
+ rcpt->offset = offset;
+ rcpt->queue = 0;
+ rcpt->next = list->rcpt;
+ list->rcpt = rcpt;
     list->len++;
 }
 
+/* qmgr_rcpt_list_move_first - move first recipient from one list to another */
+
+void qmgr_rcpt_list_move_first(QMGR_RCPT_LIST *dst, QMGR_RCPT_LIST *src)
+{
+ QMGR_RCPT *rcpt;
+
+ rcpt = src->rcpt;
+ src->rcpt = rcpt->next;
+ src->len--;
+ rcpt->next = dst->rcpt;
+ dst->rcpt = rcpt;
+ dst->len++;
+}
+
+/* qmgr_rcpt_list_free_first - remove first recipient from the list */
+
+void qmgr_rcpt_list_free_first(QMGR_RCPT_LIST *list)
+{
+ QMGR_RCPT *rcpt;
+
+ rcpt = list->rcpt;
+ list->rcpt = rcpt->next;
+ list->len--;
+ myfree(rcpt->address);
+ rcpt->next = qmgr_rcpt_cache;
+ qmgr_rcpt_cache = rcpt;
+}
+
 /* qmgr_rcpt_list_free - release memory for in-core recipient structure */
 
 void qmgr_rcpt_list_free(QMGR_RCPT_LIST *list)
 {
     QMGR_RCPT *rcpt;
 
- for (rcpt = list->info; rcpt < list->info + list->len; rcpt++)
- myfree(rcpt->address);
- myfree((char *) list->info);
+ if ((rcpt = list->rcpt) != 0) {
+ QMGR_RCPT *last;
+ do {
+ myfree(rcpt->address);
+ last = rcpt;
+ } while((rcpt = rcpt->next) != 0) ;
+ last->next = qmgr_rcpt_cache;
+ qmgr_rcpt_cache = list->rcpt;
+ }
 }

Ok, here are the promised notes.

Problem:

There is an overall limit on total number of recipient structures which can
be held in memory at any given time. Every time a message arrives, as
much as possible recipients is read into memory. Once the message is
delivered to all of them, next bunch of recipients is read in and
the process repeats, until there are no more recipients to handle.

The problem is that the recipient structures are all freed at once,
after the message has been delivered to all of them (i.e., those which
fitted in memory). It means that even if the message has already been
delivered to all but one recipients, the structures can't be reused.

The impact is that once the total limit is reached, no more messages new
messages can make it into the active queue. And because as soon as the
first bunch of recipients is delivered, whole new bunch is read in,
the active queue is blocked until all of the message recipients, even
those stored in queue files, are taken care of. And it's even worse
when you realize that this limit applies to all transfer types at once
- so for example one big message being sent out via SMTP can block local
delivery completely for a significant amount of time.

As this seems of little concern for those with moderate mail server
configuration, those handling big mailing lists might know now what I
am talking about.

Solution:

The solution is obvious. As soon as we are done with some recipient (or
group of recipients, if there were more per site postfix was talking to),
we recycle the recipient structures, making them available for other
messages immediately.

This way we make sure that one message can't block all other ones. New
messages will have a chance to make it in and everything will adjust
in a fair manner. And, most remarkably, any other low traffic transports
won't be affected at all. (Hint: Why not try setting another SMTP transport
for sites you want to reach without delays? Yes, now it will make difference.)

Source changes:

The changes can be split into following groups:

1) The QMGR_RCPT_LIST structure has been changed to point to single
linked list of QMGR_RCPT structures instead of an array. This seemed
appropriate in order to handle the assignment of recipients to
transport queue entries efficiently. The (de)allocation routines of these
structures has been changed and so was the way how the lists are
traversed.

2) The way how qmgr_recipient_count is treated has changed (this is the
whole point of this patch). Instead of changing on per-message basis, it
is increased when recipients are assigned to transport queue entries and
decreased whenever an transport queue entry is discarded.

3) To sort the recipient list, qsort has been replaced by mergesort.

If you are really into technicalities, read ahead...

Implementation details:

ad 1) The QMGR_RCPT structures are not actually freed when no longer used.
They are kept in its own cache for later use. There are several things how
the code might be modified in the future:

        a) The cache might be completely abandoned, if underlying
           allocation routines are considered efficient enough.
           
        b) The cache might be prefilled with full set of these structures,
           allocated all at once as an array. This would work nice because
           we know the upper limit of the structures required ever, and
           would also bind all structures together in one place in memory.
           
        c) Some cache management might be implemented, to maintain
           reasonable used/unused ratio, if it ever becomes an issue.

ad 2) It would be nice if there was a way how to read in some more
recipients whenever there is enough space available, even if all those
currently in memory are not delivered yet. Unfortunately, even if the
changes I made are ready for such usage, it doesn't seem it's possible
to call qmgr_message_realloc() from qmgr_entry_done(), because of the queue
file locking issues.

Also, one might like to replace the qmgr_rcpt_list_{move,free}_first() calls
in qmgr_message_assign() by more straightforward and efficient code which
deals with the list directly. I thought about it for a while, but then I
decided to go with readability instead, because this is one of the key
points of this patch and I didn't want to make it too tricky.

ad 3) Unlike qsort, mergesort is very handy when sorting linked lists. It
goes without saying that the time spent is O(n*log(n)), too. My implementation
above uses join instead of merge whenever possible, making it O(n) for
already sorted cases (to be exact, (n-1) comparisons for the forward sorted
list and 2*(n-1) comparisons for the reverse sorted list). This is quite
useful because postfix actually does two sort passes in an attempt to
improve the efficiency of the resolver, and as long as one doesn't use
the transport tables, the order before the second pass rarely ever changes.

Well, that's it. If you have any comments or questions, feel free to
contact me at either <patrikraxoft.cz> or <patrikein.cz>. The latter is
somewhat faster, but only the first one is guaranteed to last forever.

Patrik

P.S. If some of you knows how to get around Wietse's robotic autoresponder,
I would appreciate if you could mention this patch so he doesn't miss it.

This archive was generated by hypermail 2.0b3 on Mon Nov 22 1999 - 14:03:59 CST