Add a limit on the number of inflight envelopes. The scheduler suspends

scheduling of mta/mda envelopes until the number of inflight envelopes
falls below that line.

4 files changed, 58 insertions, 14 deletions

diff --git a/usr.sbin/smtpd/parse.y b/usr.sbin/smtpd/parse.y
index 3f647ca3254..722f92aeef2 100644
--- a/usr.sbin/smtpd/parse.y
+++ b/usr.sbin/smtpd/parse.y
@@ -1,4 +1,4 @@
-/*	$OpenBSD: parse.y,v 1.127 2013/11/13 08:39:33 eric Exp $	*/
+/*	$OpenBSD: parse.y,v 1.128 2013/11/19 10:01:20 eric Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@poolp.org>
@@ -144,7 +144,7 @@ typedef struct {
 %token	AS QUEUE COMPRESSION ENCRYPTION MAXMESSAGESIZE LISTEN ON ANY PORT EXPIRE
 %token	TABLE SECURE SMTPS CERTIFICATE DOMAIN BOUNCEWARN LIMIT INET4 INET6
 %token  RELAY BACKUP VIA DELIVER TO LMTP MAILDIR MBOX HOSTNAME HOSTNAMES
-%token	ACCEPT REJECT INCLUDE ERROR MDA FROM FOR SOURCE MTA PKI
+%token	ACCEPT REJECT INCLUDE ERROR MDA FROM FOR SOURCE MTA PKI SCHEDULER
 %token	ARROW AUTH TLS LOCAL VIRTUAL TAG TAGGED ALIAS FILTER FILTERCHAIN KEY CA DHPARAMS
 %token	AUTH_OPTIONAL TLS_REQUIRE USERBASE SENDER MASK_SOURCE VERIFY FORWARDONLY RECIPIENT
 %token	<v.string>	STRING
@@ -256,7 +256,7 @@ bouncedelays	: bouncedelays ',' bouncedelay
 		| /* EMPTY */
 		;
 
-opt_limit	: INET4 {
+opt_limit_mta	: INET4 {
 			limits->family = AF_INET;
 		}
 		| INET6 {
@@ -264,7 +264,7 @@ opt_limit	: INET4 {
 		}
 		| STRING NUMBER {
 			if (!limit_mta_set(limits, $1, $2)) {
-				yyerror("invalid limit keyword");
+				yyerror("invalid mta limit keyword: %s", $1);
 				free($1);
 				YYERROR;
 			}
@@ -272,7 +272,24 @@ opt_limit	: INET4 {
 		}
 		;
 
-limits		: opt_limit limits
+limits_mta	: opt_limit_mta limits_mta
+		| /* empty */
+		;
+
+opt_limit_scheduler : STRING NUMBER {
+			if (!strcmp($1, "max-inflight")) {
+				conf->sc_scheduler_max_inflight = $2;
+			}
+			else {
+				yyerror("invalid scheduler limit keyword: %s", $1);
+				free($1);
+				YYERROR;
+			}
+			free($1);
+		}
+		;
+
+limits_scheduler: opt_limit_scheduler limits_scheduler
 		| /* empty */
 		;
 
@@ -542,10 +559,11 @@ main		: BOUNCEWARN {
 				memmove(limits, d, sizeof(*limits));
 			}
 			free($5);
-		} limits
+		} limits_mta
 		| LIMIT MTA {
 			limits = dict_get(conf->sc_limits_dict, "default");
-		} limits
+		} limits_mta
+		| LIMIT SCHEDULER limits_scheduler
 		| LISTEN {
 			bzero(&l, sizeof l);
 			bzero(&listen_opts, sizeof listen_opts);
@@ -1106,6 +1124,7 @@ lookup(char *s)
 		{ "recipient",		RECIPIENT },
 		{ "reject",		REJECT },
 		{ "relay",		RELAY },
+		{ "scheduler",		SCHEDULER },
 		{ "secure",		SECURE },
 		{ "sender",    		SENDER },
 		{ "smtps",		SMTPS },
@@ -1506,6 +1525,8 @@ parse_config(struct smtpd *x_conf, const char *filename, int opts)
 	conf->sc_qexpire = SMTPD_QUEUE_EXPIRY;
 	conf->sc_opts = opts;
 
+	conf->sc_scheduler_max_inflight = 5000;
+
 	if ((file = pushfile(filename, 0)) == NULL) {
 		purge_config(PURGE_EVERYTHING);
 		return (-1);
diff --git a/usr.sbin/smtpd/scheduler.c b/usr.sbin/smtpd/scheduler.c
index 7971ae3be24..efe2cbf7622 100644
--- a/usr.sbin/smtpd/scheduler.c
+++ b/usr.sbin/smtpd/scheduler.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: scheduler.c,v 1.35 2013/10/27 17:47:53 eric Exp $	*/
+/*	$OpenBSD: scheduler.c,v 1.36 2013/11/19 10:01:20 eric Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@poolp.org>
@@ -57,6 +57,7 @@ static void scheduler_process_mta(struct scheduler_batch *);
 
 static struct scheduler_backend *backend = NULL;
 static struct event		 ev;
+static size_t			 ninflight;
 
 extern const char *backend_scheduler;
 
@@ -128,8 +129,11 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 		stat_decrement("scheduler.envelope", 1);
 		if (! inflight)
 			backend->remove(evpid);
-		else
+		else {
 			backend->delete(evpid);
+			ninflight -= 1;
+			stat_decrement("scheduler.envelope.inflight", 1);
+		}
 
 		scheduler_reset_events();
 		return;
@@ -141,6 +145,7 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 		log_trace(TRACE_SCHEDULER,
 		    "scheduler: deleting evp:%016" PRIx64 " (ok)", evpid);
 		backend->delete(evpid);
+		ninflight -= 1;
 		stat_increment("scheduler.delivery.ok", 1);
 		stat_decrement("scheduler.envelope.inflight", 1);
 		stat_decrement("scheduler.envelope", 1);
@@ -156,6 +161,7 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 		    "scheduler: updating evp:%016" PRIx64, evp.id);
 		scheduler_info(&si, &evp, penalty);
 		backend->update(&si);
+		ninflight -= 1;
 		stat_increment("scheduler.delivery.tempfail", 1);
 		stat_decrement("scheduler.envelope.inflight", 1);
 
@@ -185,6 +191,7 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 		log_trace(TRACE_SCHEDULER,
 		    "scheduler: deleting evp:%016" PRIx64 " (fail)", evpid);
 		backend->delete(evpid);
+		ninflight -= 1;
 		stat_increment("scheduler.delivery.permfail", 1);
 		stat_decrement("scheduler.envelope.inflight", 1);
 		stat_decrement("scheduler.envelope", 1);
@@ -198,6 +205,7 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 		log_trace(TRACE_SCHEDULER,
 		    "scheduler: deleting evp:%016" PRIx64 " (loop)", evpid);
 		backend->delete(evpid);
+		ninflight -= 1;
 		stat_increment("scheduler.delivery.loop", 1);
 		stat_decrement("scheduler.envelope.inflight", 1);
 		stat_decrement("scheduler.envelope", 1);
@@ -213,6 +221,7 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 		    "scheduler: holding evp:%016" PRIx64 " on %016" PRIx64,
 		    evpid, holdq);
 		backend->hold(evpid, holdq);
+		ninflight -= 1;
 		stat_decrement("scheduler.envelope.inflight", 1);
 		scheduler_reset_events();
 		return;
@@ -458,9 +467,11 @@ scheduler_timeout(int fd, short event, void *p)
 	tv.tv_usec = 0;
 
 	typemask = SCHED_REMOVE | SCHED_EXPIRE | SCHED_BOUNCE;
-	if (!(env->sc_flags & SMTPD_MDA_PAUSED))
+	if (ninflight < env->sc_scheduler_max_inflight &&
+	    !(env->sc_flags & SMTPD_MDA_PAUSED))
 		typemask |= SCHED_MDA;
-	if (!(env->sc_flags & SMTPD_MTA_PAUSED))
+	if (ninflight < env->sc_scheduler_max_inflight &&
+	    !(env->sc_flags & SMTPD_MTA_PAUSED))
 		typemask |= SCHED_MTA;
 
 	bzero(&batch, sizeof (batch));
@@ -564,6 +575,7 @@ scheduler_process_bounce(struct scheduler_batch *batch)
 		m_close(p_queue);
 	}
 
+	ninflight += batch->evpcount;
 	stat_increment("scheduler.envelope.inflight", batch->evpcount);
 }
 
@@ -580,6 +592,7 @@ scheduler_process_mda(struct scheduler_batch *batch)
 		m_close(p_queue);
 	}
 
+	ninflight += batch->evpcount;
 	stat_increment("scheduler.envelope.inflight", batch->evpcount);
 }
 
@@ -596,5 +609,6 @@ scheduler_process_mta(struct scheduler_batch *batch)
 		m_close(p_queue);
 	}
 
+	ninflight += batch->evpcount;
 	stat_increment("scheduler.envelope.inflight", batch->evpcount);
 }
diff --git a/usr.sbin/smtpd/smtpd.conf.5 b/usr.sbin/smtpd/smtpd.conf.5
index 9787309ef0c..9e07449142c 100644
--- a/usr.sbin/smtpd/smtpd.conf.5
+++ b/usr.sbin/smtpd/smtpd.conf.5
@@ -1,4 +1,4 @@
-.\"	$OpenBSD: smtpd.conf.5,v 1.108 2013/11/18 12:24:26 eric Exp $
+.\"	$OpenBSD: smtpd.conf.5,v 1.109 2013/11/19 10:01:20 eric Exp $
 .\"
 .\" Copyright (c) 2008 Janne Johansson <jj@openbsd.org>
 .\" Copyright (c) 2009 Jacek Masiulaniec <jacekm@dobremiasto.net>
@@ -17,7 +17,7 @@
 .\" OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 .\"
 .\"
-.Dd $Mdocdate: November 18 2013 $
+.Dd $Mdocdate: November 19 2013 $
 .Dt SMTPD.CONF 5
 .Os
 .Sh NAME
@@ -550,6 +550,14 @@ If a
 is specified, the restriction only applies when connecting
 to MXs for this domain.
 .It Xo
+.Ic limit scheduler max-inflight
+.Ar num
+.Xc
+Suspend the scheduling of envelopes for deliver/relay until the number
+of inflight envelopes falls below
+.Ar num .
+Changing the default value might degrade performances.
+.It Xo
 .Bk -words
 .Ic listen on Ar interface
 .Op Ar family
diff --git a/usr.sbin/smtpd/smtpd.h b/usr.sbin/smtpd/smtpd.h
index d9b26c77688..38d5d1bf19a 100644
--- a/usr.sbin/smtpd/smtpd.h
+++ b/usr.sbin/smtpd/smtpd.h
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpd.h,v 1.434 2013/11/18 12:24:26 eric Exp $	*/
+/*	$OpenBSD: smtpd.h,v 1.435 2013/11/19 10:01:20 eric Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@poolp.org>
@@ -505,6 +505,7 @@ struct smtpd {
 	char			       *sc_queue_key;
 	size_t				sc_queue_evpcache_size;
 
+	size_t				sc_scheduler_max_inflight;
 	int				sc_qexpire;
 #define MAX_BOUNCE_WARN			4
 	time_t				sc_bounce_warn[MAX_BOUNCE_WARN];