Handle >1 parallelism in serialize db entries

4 files changed, 40 insertions, 18 deletions

diff --git a/src/src/enq.c b/src/src/enq.c
index 6cd243c4c..7f7bebc68 100644
--- a/src/src/enq.c
+++ b/src/src/enq.c
@@ -20,17 +20,20 @@
 /* This function is called when a host is listed for serialization of
 connections. It is also called when ETRN is listed for serialization. We open
 the misc database and look for a record, which implies an existing connection
-or ETRN run. If not found, create one and return TRUE.
+or ETRN run. If increasing the count would take us past the given limit
+value return FALSE.  If not, bump it and return TRUE.  If not found, create
+one with value 1 and return TRUE.
 
 Arguments:
   key            string on which to serialize
+  lim            parallelism limit
 
 Returns:         TRUE if OK to proceed; FALSE otherwise
 */
 
 
 BOOL
-enq_start(uschar *key)
+enq_start(uschar *key, unsigned lim)
 {
 dbdata_serialize *serial_record;
 dbdata_serialize new_record;
@@ -44,25 +47,31 @@ deliberate; the dbfn_open() function - which is an Exim function - always tries
 to create if it can't open a read/write file. It expects only O_RDWR or
 O_RDONLY as its argument. */
 
-dbm_file = dbfn_open(US"misc", O_RDWR, &dbblock, TRUE);
-if (dbm_file == NULL) return FALSE;
+if (!(dbm_file = dbfn_open(US"misc", O_RDWR, &dbblock, TRUE)))
+  return FALSE;
 
 /* See if there is a record for this host or queue run; if there is, we cannot
 proceed with the connection unless the record is very old. */
 
 serial_record = dbfn_read(dbm_file, key);
-if (serial_record != NULL && time(NULL) - serial_record->time_stamp < 6*60*60)
+if (serial_record && time(NULL) - serial_record->time_stamp < 6*60*60)
   {
-  dbfn_close(dbm_file);
-  DEBUG(D_transport) debug_printf("outstanding serialization record for %s\n",
-    key);
-  return FALSE;
+  if (serial_record->count >= lim)
+    {
+    dbfn_close(dbm_file);
+    DEBUG(D_transport) debug_printf("outstanding serialization record for %s\n",
+      key);
+    return FALSE;
+    }
+  new_record.count = serial_record->count + 1;
   }
+else
+  new_record.count = 1;
 
-/* We can proceed - insert a new record or update the old one. At present
-the count field is not used; just set it to 1. */
+/* We can proceed - insert a new record or update the old one. */
 
-new_record.count = 1;
+DEBUG(D_transport) debug_printf("write serialization record for %s val %d\n",
+      key, new_record.count);
 dbfn_write(dbm_file, key, &new_record, (int)sizeof(dbdata_serialize));
 dbfn_close(dbm_file);
 return TRUE;
@@ -88,12 +97,25 @@ enq_end(uschar *key)
 {
 open_db dbblock;
 open_db *dbm_file;
+dbdata_serialize *serial_record;
 
 DEBUG(D_transport) debug_printf("end serialized: %s\n", key);
 
-dbm_file = dbfn_open(US"misc", O_RDWR, &dbblock, TRUE);
-if (dbm_file == NULL) return;
-dbfn_delete(dbm_file, key);
+if (  !(dbm_file = dbfn_open(US"misc", O_RDWR, &dbblock, TRUE))
+   || !(serial_record = dbfn_read(dbm_file, key))
+   )
+  return;
+if (--serial_record->count > 0)
+  {
+  DEBUG(D_transport) debug_printf("write serialization record for %s val %d\n",
+      key, serial_record->count);
+  dbfn_write(dbm_file, key, serial_record, (int)sizeof(dbdata_serialize));
+  }
+else
+  {
+  DEBUG(D_transport) debug_printf("remove serialization record for %s\n", key);
+  dbfn_delete(dbm_file, key);
+  }
 dbfn_close(dbm_file);
 }
 
diff --git a/src/src/functions.h b/src/src/functions.h
index fcfe74675..9baa32d59 100644
--- a/src/src/functions.h
+++ b/src/src/functions.h
@@ -163,7 +163,7 @@ extern void    dscp_list_to_stream(FILE *);
 extern BOOL    dscp_lookup(const uschar *, int, int *, int *, int *);
 
 extern void    enq_end(uschar *);
-extern BOOL    enq_start(uschar *);
+extern BOOL    enq_start(uschar *, unsigned);
 #ifdef EXPERIMENTAL_EVENT
 extern uschar *event_raise(uschar *, const uschar *, uschar *);
 #endif
diff --git a/src/src/smtp_in.c b/src/src/smtp_in.c
index abbcad62c..9aacef1fc 100644
--- a/src/src/smtp_in.c
+++ b/src/src/smtp_in.c
@@ -4979,7 +4979,7 @@ while (done <= 0)
     /* If ETRN queue runs are to be serialized, check the database to
     ensure one isn't already running. */
 
-    if (smtp_etrn_serialize && !enq_start(etrn_serialize_key))
+    if (smtp_etrn_serialize && !enq_start(etrn_serialize_key, 1))
       {
       smtp_printf("458 Already processing %s\r\n", smtp_cmd_data);
       break;
diff --git a/src/src/transports/smtp.c b/src/src/transports/smtp.c
index f129cce9b..eae54764b 100644
--- a/src/src/transports/smtp.c
+++ b/src/src/transports/smtp.c
@@ -3532,7 +3532,7 @@ for (cutoff_retry = 0; expired &&
         verify_check_given_host(&ob->serialize_hosts, host) == OK)
       {
       serialize_key = string_sprintf("host-serialize-%s", host->name);
-      if (!enq_start(serialize_key))
+      if (!enq_start(serialize_key, 1))
         {
         DEBUG(D_transport)
           debug_printf("skipping host %s because another Exim process "