Rework the envelope flushing loops in mta to avoid sending all delivery

notifications in one go to the queue. Simplify code in the process.

3 files changed, 83 insertions, 71 deletions

diff --git a/usr.sbin/smtpd/mta.c b/usr.sbin/smtpd/mta.c
index 2e17834f80a..4c7861b05f0 100644
--- a/usr.sbin/smtpd/mta.c
+++ b/usr.sbin/smtpd/mta.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: mta.c,v 1.174 2013/11/30 10:21:16 eric Exp $	*/
+/*	$OpenBSD: mta.c,v 1.175 2013/12/03 09:06:26 eric Exp $	*/
 
 /*
  * Copyright (c) 2008 Pierre-Yves Ritschard <pyr@openbsd.org>
@@ -73,6 +73,7 @@ static void mta_connect(struct mta_connector *);
 static void mta_route_enable(struct mta_route *);
 static void mta_route_disable(struct mta_route *, int, int);
 static void mta_drain(struct mta_relay *);
+static void mta_delivery_flush_event(int, short, void *);
 static void mta_flush(struct mta_relay *, int, const char *);
 static struct mta_route *mta_find_route(struct mta_connector *, time_t, int*,
     time_t*);
@@ -134,6 +135,8 @@ static struct tree wait_mx;
 static struct tree wait_preference;
 static struct tree wait_secret;
 static struct tree wait_source;
+static struct tree flush_evp;
+static struct event ev_flush_evp;
 
 static struct runq *runq_relay;
 static struct runq *runq_connector;
@@ -533,11 +536,14 @@ mta(void)
 	tree_init(&wait_mx);
 	tree_init(&wait_preference);
 	tree_init(&wait_source);
+	tree_init(&flush_evp);
 	dict_init(&hoststat);
 
 	imsg_callback = mta_imsg;
 	event_init();
 
+	evtimer_set(&ev_flush_evp, mta_delivery_flush_event, NULL);
+
 	runq_init(&runq_relay, mta_on_timeout);
 	runq_init(&runq_connector, mta_on_timeout);
 	runq_init(&runq_route, mta_on_timeout);
@@ -678,52 +684,75 @@ mta_route_next_task(struct mta_relay *relay, struct mta_route *route)
 	return (task);
 }
 
+static void
+mta_delivery_flush_event(int fd, short event, void *arg)
+{
+	struct mta_envelope	*e;
+	struct timeval		 tv;
+
+	if (tree_poproot(&flush_evp, NULL, (void**)(&e))) {
+
+		if (e->delivery == IMSG_DELIVERY_OK)
+			queue_ok(e->id);
+		else if (e->delivery == IMSG_DELIVERY_TEMPFAIL)
+			queue_tempfail(e->id, e->penalty, e->status);
+		else if (e->delivery == IMSG_DELIVERY_PERMFAIL)
+			queue_permfail(e->id, e->status);
+		else if (e->delivery == IMSG_DELIVERY_LOOP)
+			queue_loop(e->id);
+		else {
+			log_warnx("warn: bad delivery type %i for %016" PRIx64,
+			    e->delivery, e->id);
+			fatalx("aborting");
+		}
+
+		log_debug("debug: mta: flush for %016"PRIx64" (-> %s)", e->id, e->dest);
+
+		free(e->dest);
+		free(e->rcpt);
+		free(e);
+
+		tv.tv_sec = 0;
+		tv.tv_usec = 0;
+		evtimer_add(&ev_flush_evp, &tv);
+	}
+}
+
 void
 mta_delivery_log(struct mta_envelope *e, const char *source, const char *relay,
     int delivery, const char *status)
 {
-	if (delivery == IMSG_DELIVERY_OK) {
+	if (delivery == IMSG_DELIVERY_OK)
 		mta_log(e, "Ok", source, relay, status);
-	}
-	else if (delivery == IMSG_DELIVERY_TEMPFAIL) {
+	else if (delivery == IMSG_DELIVERY_TEMPFAIL)
 		mta_log(e, "TempFail", source, relay, status);
-	}
-	else if (delivery == IMSG_DELIVERY_PERMFAIL) {
+	else if (delivery == IMSG_DELIVERY_PERMFAIL)
 		mta_log(e, "PermFail", source, relay, status);
-	}
-	else if (delivery == IMSG_DELIVERY_LOOP) {
+	else if (delivery == IMSG_DELIVERY_LOOP)
 		mta_log(e, "PermFail", source, relay, "Loop detected");
+	else {
+		log_warnx("warn: bad delivery type %i for %016" PRIx64,
+		    delivery, e->id);
+		fatalx("aborting");
 	}
-	else
-		errx(1, "bad delivery");
-}
 
-void
-mta_delivery_notify(struct mta_envelope *e, int delivery, const char *status,
-    uint32_t penalty)
-{
-	if (delivery == IMSG_DELIVERY_OK) {
-		queue_ok(e->id);
-	}
-	else if (delivery == IMSG_DELIVERY_TEMPFAIL) {
-		queue_tempfail(e->id, penalty, status);
-	}
-	else if (delivery == IMSG_DELIVERY_PERMFAIL) {
-		queue_permfail(e->id, status);
-	}
-	else if (delivery == IMSG_DELIVERY_LOOP) {
-		queue_loop(e->id);
-	}
-	else
-		errx(1, "bad delivery");
+	e->delivery = delivery;
+	if (status)
+		strlcpy(e->status, status, sizeof(e->status));
 }
 
 void
-mta_delivery(struct mta_envelope *e, const char *source, const char *relay,
-    int delivery, const char *status, uint32_t penalty)
+mta_delivery_notify(struct mta_envelope *e, uint32_t penalty)
 {
-	mta_delivery_log(e, source, relay, delivery, status);
-	mta_delivery_notify(e, delivery, status, penalty);
+	struct timeval	tv;
+
+	e->penalty = penalty;
+	tree_xset(&flush_evp, e->id, e);
+	if (tree_count(&flush_evp) == 1) {
+		tv.tv_sec = 0;
+		tv.tv_usec = 0;
+		evtimer_add(&ev_flush_evp, &tv);
+	}
 }
 
 static void
@@ -1276,8 +1305,7 @@ mta_flush(struct mta_relay *relay, int fail, const char *error)
 	const char     		*domain;
 	void			*iter;
 	struct mta_connector	*c;
-	size_t			 n;
-	size_t			 r;
+	size_t			 n, r;
 
 	log_debug("debug: mta_flush(%s, %d, \"%s\")",
 	    mta_relay_to_text(relay), fail, error);
@@ -1290,8 +1318,7 @@ mta_flush(struct mta_relay *relay, int fail, const char *error)
 		TAILQ_REMOVE(&relay->tasks, task, entry);
 		while ((e = TAILQ_FIRST(&task->envelopes))) {
 			TAILQ_REMOVE(&task->envelopes, e, entry);
-			mta_delivery(e, NULL, relay->domain->name, fail, error, 0);
-			
+
 			/*
 			 * host was suspended, cache envelope id in hoststat tree
 			 * so that it can be retried when a delivery succeeds for
@@ -1310,9 +1337,9 @@ mta_flush(struct mta_relay *relay, int fail, const char *error)
 					mta_hoststat_cache(domain+1, e->id);
 			}
 
-			free(e->dest);
-			free(e->rcpt);
-			free(e);
+			mta_delivery_log(e, NULL, relay->domain->name, fail, error);
+			mta_delivery_notify(e, 0);
+
 			n++;
 		}
 		free(task->sender);
diff --git a/usr.sbin/smtpd/mta_session.c b/usr.sbin/smtpd/mta_session.c
index d1c569c5ddb..bcc5606d758 100644
--- a/usr.sbin/smtpd/mta_session.c
+++ b/usr.sbin/smtpd/mta_session.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: mta_session.c,v 1.48 2013/11/28 10:43:37 eric Exp $	*/
+/*	$OpenBSD: mta_session.c,v 1.49 2013/12/03 09:06:26 eric Exp $	*/
 
 /*
  * Copyright (c) 2008 Pierre-Yves Ritschard <pyr@openbsd.org>
@@ -98,12 +98,6 @@ enum mta_state {
 #define MTA_EXT_AUTH_LOGIN     	0x10
 
 
-struct failed_evp {
-	int			 delivery;
-	char			 error[SMTPD_MAXLINESIZE];
-	struct mta_envelope     *evp;
-};
-
 struct mta_session {
 	uint64_t		 id;
 	struct mta_relay	*relay;
@@ -133,7 +127,7 @@ struct mta_session {
 	FILE			*datafp;
 
 #define	MAX_FAILED_ENVELOPES	15
-	struct failed_evp	 failed[MAX_FAILED_ENVELOPES];
+	struct mta_envelope	*failed[MAX_FAILED_ENVELOPES];
 	int			 failedcount;
 };
 
@@ -842,7 +836,6 @@ static void
 mta_response(struct mta_session *s, char *line)
 {
 	struct mta_envelope	*e;
-	struct failed_evp	*fevp;
 	struct sockaddr_storage	 ss;
 	struct sockaddr		*sa;
 	const char		*domain;
@@ -1007,11 +1000,7 @@ mta_response(struct mta_session *s, char *line)
 				    buf, delivery, line);
 
 			/* push failed envelope to the session fail queue */
-			e->delivery = delivery;
-			fevp = &s->failed[s->failedcount];
-			fevp->delivery = delivery;
-			fevp->evp = e;
-			strlcpy(fevp->error, line, sizeof fevp->error);
+			s->failed[s->failedcount] = e;
 			s->failedcount++;
 
 			/*
@@ -1368,10 +1357,12 @@ mta_flush_task(struct mta_session *s, int delivery, const char *error, size_t co
 		sa = (struct sockaddr *)&ss;
 		sa_len = sizeof(ss);
 		if (getsockname(s->io.sock, sa, &sa_len) < 0)
-			mta_delivery(e, NULL, relay, delivery, error, 0);
+			mta_delivery_log(e, NULL, relay, delivery, error);
 		else
-			mta_delivery(e, sa_to_text(sa),
-			    relay, delivery, error, 0);
+			mta_delivery_log(e, sa_to_text(sa),
+			    relay, delivery, error);
+
+		mta_delivery_notify(e, 0);
 
 		domain = strchr(e->dest, '@');
 		if (domain) {
@@ -1380,9 +1371,6 @@ mta_flush_task(struct mta_session *s, int delivery, const char *error, size_t co
 				mta_hoststat_cache(domain + 1, e->id);
 		}
 
-		free(e->dest);
-		free(e->rcpt);
-		free(e);
 		n++;
 	}
 
@@ -1404,25 +1392,21 @@ static void
 mta_flush_failedqueue(struct mta_session *s)
 {
 	int			 i;
-	struct failed_evp	*fevp;
 	struct mta_envelope	*e;
 	const char		*domain;
 	uint32_t		 penalty;
 
 	penalty = s->failedcount == MAX_FAILED_ENVELOPES ? 1 : 0;
 	for (i = 0; i < s->failedcount; ++i) {
-		fevp = &s->failed[i];
-		e = fevp->evp;
-		mta_delivery_notify(e, fevp->delivery, fevp->error, penalty);
+		e = s->failed[i];
 
 		domain = strchr(e->dest, '@');
 		if (domain)
-			mta_hoststat_update(domain + 1, fevp->error);
+			mta_hoststat_update(domain + 1, e->status);
 
-		free(e->dest);
-		free(e->rcpt);
-		free(e);
+		mta_delivery_notify(e, penalty);
 	}
+
 	s->failedcount = 0;
 }
 
diff --git a/usr.sbin/smtpd/smtpd.h b/usr.sbin/smtpd/smtpd.h
index ba7b32b02dc..d84c0fa66b9 100644
--- a/usr.sbin/smtpd/smtpd.h
+++ b/usr.sbin/smtpd/smtpd.h
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpd.h,v 1.438 2013/11/30 10:11:57 eric Exp $	*/
+/*	$OpenBSD: smtpd.h,v 1.439 2013/12/03 09:06:26 eric Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@poolp.org>
@@ -760,6 +760,8 @@ struct mta_envelope {
 	char				*rcpt;
 	struct mta_task			*task;
 	int				 delivery;
+	int				 penalty;
+	char				 status[SMTPD_MAXLINESIZE];
 };
 
 struct mta_task {
@@ -1186,8 +1188,7 @@ void mta_route_down(struct mta_relay *, struct mta_route *);
 void mta_route_collect(struct mta_relay *, struct mta_route *);
 void mta_source_error(struct mta_relay *, struct mta_route *, const char *);
 void mta_delivery_log(struct mta_envelope *, const char *, const char *, int, const char *);
-void mta_delivery_notify(struct mta_envelope *, int, const char *, uint32_t);
-void mta_delivery(struct mta_envelope *, const char *, const char *, int, const char *, uint32_t);
+void mta_delivery_notify(struct mta_envelope *, uint32_t);
 struct mta_task *mta_route_next_task(struct mta_relay *, struct mta_route *);
 const char *mta_host_to_text(struct mta_host *);
 const char *mta_relay_to_text(struct mta_relay *);