If the admin issues a remove request for an envelope which is inflight, do not

ignore the request. Instead defer the removal until the envelope returns to the
scheduler. Simplify code by a great deal in the process.

1 files changed, 107 insertions, 102 deletions

diff --git a/usr.sbin/smtpd/scheduler_ramqueue.c b/usr.sbin/smtpd/scheduler_ramqueue.c
index 4efd88d132b..e26b91831c6 100644
--- a/usr.sbin/smtpd/scheduler_ramqueue.c
+++ b/usr.sbin/smtpd/scheduler_ramqueue.c
@@ -1,8 +1,8 @@
-/*	$OpenBSD: scheduler_ramqueue.c,v 1.30 2013/07/19 21:34:31 eric Exp $	*/
+/*	$OpenBSD: scheduler_ramqueue.c,v 1.31 2013/10/25 19:18:43 eric Exp $	*/
 
 /*
  * Copyright (c) 2012 Gilles Chehade <gilles@poolp.org>
- * Copyright (c) 2012 Eric Faurot <eric@openbsd.org>
+ * Copyright (c) 2012-2013 Eric Faurot <eric@openbsd.org>
  *
  * Permission to use, copy, modify, and distribute this software for any
  * purpose with or without fee is hereby granted, provided that the above
@@ -41,22 +41,22 @@ TAILQ_HEAD(evplist, rq_envelope);
 struct rq_message {
 	uint32_t		 msgid;
 	struct tree		 envelopes;
-	struct rq_message	*q_next;
-	struct evplist		 q_mta;
 };
 
 struct rq_envelope {
 	TAILQ_ENTRY(rq_envelope) entry;
 
 	uint64_t		 evpid;
-	int			 type;
-
-#define	RQ_ENVELOPE_PENDING	 0x01
-#define	RQ_ENVELOPE_SCHEDULED	 0x02
-#define	RQ_ENVELOPE_EXPIRED	 0x04
-#define	RQ_ENVELOPE_REMOVED	 0x08
-#define	RQ_ENVELOPE_INFLIGHT	 0x10
-#define	RQ_ENVELOPE_SUSPEND	 0x20
+	enum delivery_type	 type;
+
+#define	RQ_EVPSTATE_PENDING	 0
+#define	RQ_EVPSTATE_SCHEDULED	 1
+#define	RQ_EVPSTATE_INFLIGHT	 2
+	uint8_t			 state;
+
+#define	RQ_ENVELOPE_EXPIRED	 0x01
+#define	RQ_ENVELOPE_REMOVED	 0x02
+#define	RQ_ENVELOPE_SUSPEND	 0x04
 	uint8_t			 flags;
 
 	time_t			 sched;
@@ -75,7 +75,7 @@ struct rq_queue {
 	struct evplist		 q_pending;
 	struct evplist		 q_inflight;
 
-	struct rq_message	*q_mtabatch;
+	struct evplist		 q_mta;
 	struct evplist		 q_mda;
 	struct evplist		 q_bounce;
 	struct evplist		 q_expired;
@@ -169,7 +169,6 @@ scheduler_ram_insert(struct scheduler_info *si)
 	if ((message = tree_get(&update->messages, msgid)) == NULL) {
 		message = xcalloc(1, sizeof *message, "scheduler_insert");
 		message->msgid = msgid;
-		TAILQ_INIT(&message->q_mta);
 		tree_init(&message->envelopes);
 		tree_xset(&update->messages, msgid, message);
 		stat_increment("scheduler.ramqueue.message", 1);
@@ -187,7 +186,7 @@ scheduler_ram_insert(struct scheduler_info *si)
 	update->evpcount++;
 	stat_increment("scheduler.ramqueue.envelope", 1);
 
-	envelope->flags = RQ_ENVELOPE_PENDING;
+	envelope->state = RQ_EVPSTATE_PENDING;
 	sorted_insert(&update->q_pending, envelope);
 
 	si->nexttry = envelope->sched;
@@ -260,15 +259,26 @@ scheduler_ram_update(struct scheduler_info *si)
 	evp = tree_xget(&msg->envelopes, si->evpid);
 
 	/* it *must* be in-flight */
-	if (!(evp->flags & RQ_ENVELOPE_INFLIGHT))
+	if (evp->state != RQ_EVPSTATE_INFLIGHT)
 		errx(1, "evp:%016" PRIx64 " not in-flight", si->evpid);
 
+	TAILQ_REMOVE(&ramqueue.q_inflight, evp, entry);
+
+	/*
+	 * If the envelope was removed while inflight,  schedule it for
+	 * removal immediatly.
+	 */
+	if (evp->flags & RQ_ENVELOPE_REMOVED) {
+		TAILQ_INSERT_TAIL(&ramqueue.q_removed, evp, entry);
+		evp->state = RQ_EVPSTATE_SCHEDULED;
+		evp->t_scheduled = currtime;
+		return (1);
+	}
+
 	while ((evp->sched = scheduler_compute_schedule(si)) <= currtime)
 		si->retry += 1;
 
-	TAILQ_REMOVE(&ramqueue.q_inflight, evp, entry);
-	evp->flags &= ~RQ_ENVELOPE_INFLIGHT;
-	evp->flags |= RQ_ENVELOPE_PENDING;
+	evp->state = RQ_EVPSTATE_PENDING;
 	if (!(evp->flags & RQ_ENVELOPE_SUSPEND))
 		sorted_insert(&ramqueue.q_pending, evp);
 
@@ -291,11 +301,11 @@ scheduler_ram_delete(uint64_t evpid)
 	evp = tree_xget(&msg->envelopes, evpid);
 
 	/* it *must* be in-flight */
-	if (!(evp->flags & RQ_ENVELOPE_INFLIGHT))
+	if (evp->state != RQ_EVPSTATE_INFLIGHT)
 		errx(1, "evp:%016" PRIx64 " not in-flight", evpid);
 
 	TAILQ_REMOVE(&ramqueue.q_inflight, evp, entry);
-	evp->flags &= ~RQ_ENVELOPE_INFLIGHT;
+
 	rq_envelope_delete(&ramqueue, evp);
 
 	return (1);
@@ -306,7 +316,6 @@ scheduler_ram_batch(int typemask, struct scheduler_batch *ret)
 {
 	struct evplist		*q;
 	struct rq_envelope	*evp;
-	struct rq_message	*msg;
 	size_t			 n;
 
 	currtime = time(NULL);
@@ -331,11 +340,8 @@ scheduler_ram_batch(int typemask, struct scheduler_batch *ret)
 		q = &ramqueue.q_mda;
 		ret->type = SCHED_MDA;
 	}
-	else if (typemask & SCHED_MTA && ramqueue.q_mtabatch) {
-		msg = ramqueue.q_mtabatch;
-		ramqueue.q_mtabatch = msg->q_next;
-		msg->q_next = NULL;
-		q = &msg->q_mta;
+	else if (typemask & SCHED_MTA && TAILQ_FIRST(&ramqueue.q_mta)) {
+		q = &ramqueue.q_mta;
 		ret->type = SCHED_MTA;
 	}
 	else if ((evp = TAILQ_FIRST(&ramqueue.q_pending))) {
@@ -357,8 +363,8 @@ scheduler_ram_batch(int typemask, struct scheduler_batch *ret)
 
 		TAILQ_REMOVE(q, evp, entry);
 
-		/* consistency check */ 
-		if (!(evp->flags & RQ_ENVELOPE_SCHEDULED))
+		/* consistency check */
+		if (evp->state != RQ_EVPSTATE_SCHEDULED)
 			errx(1, "evp:%016" PRIx64 " not scheduled", evp->evpid);
 
 		ret->evpids[n] = evp->evpid;
@@ -367,8 +373,7 @@ scheduler_ram_batch(int typemask, struct scheduler_batch *ret)
 			rq_envelope_delete(&ramqueue, evp);
 		else {
 			TAILQ_INSERT_TAIL(&ramqueue.q_inflight, evp, entry);
-			evp->flags &= ~RQ_ENVELOPE_SCHEDULED;
-			evp->flags |= RQ_ENVELOPE_INFLIGHT;
+			evp->state = RQ_EVPSTATE_INFLIGHT;
 			evp->t_inflight = currtime;
 		}
 	}
@@ -414,22 +419,26 @@ scheduler_ram_envelopes(uint64_t from, struct evpstate *dst, size_t size)
 		if (evp->flags & (RQ_ENVELOPE_REMOVED | RQ_ENVELOPE_EXPIRED))
 			continue;
 
-		dst[n].retry = 0;
 		dst[n].evpid = evp->evpid;
-		if (evp->flags & RQ_ENVELOPE_PENDING) {
+		dst[n].flags = 0;
+		dst[n].retry = 0;
+		dst[n].time = 0;
+
+		if (evp->state == RQ_EVPSTATE_PENDING) {
 			dst[n].time = evp->sched;
 			dst[n].flags = EF_PENDING;
 		}
-		else if (evp->flags & RQ_ENVELOPE_SCHEDULED) {
+		else if (evp->state == RQ_EVPSTATE_SCHEDULED) {
 			dst[n].time = evp->t_scheduled;
 			dst[n].flags = EF_PENDING;
 		}
-		else if (evp->flags & RQ_ENVELOPE_INFLIGHT) {
+		else if (evp->state == RQ_EVPSTATE_INFLIGHT) {
 			dst[n].time = evp->t_inflight;
 			dst[n].flags = EF_INFLIGHT;
 		}
 		if (evp->flags & RQ_ENVELOPE_SUSPEND)
 			dst[n].flags |= EF_SUSPEND;
+
 		n++;
 	}
 
@@ -453,11 +462,10 @@ scheduler_ram_schedule(uint64_t evpid)
 			return (0);
 		if ((evp = tree_get(&msg->envelopes, evpid)) == NULL)
 			return (0);
-		if (evp->flags & RQ_ENVELOPE_PENDING) {
-			rq_envelope_schedule(&ramqueue, evp);
-			return (1);
-		}
-		return (0);
+		if (evp->state == RQ_EVPSTATE_INFLIGHT)
+			return (0);
+		rq_envelope_schedule(&ramqueue, evp);
+		return (1);
 	}
 	else {
 		msgid = evpid;
@@ -465,11 +473,12 @@ scheduler_ram_schedule(uint64_t evpid)
 			return (0);
 		i = NULL;
 		r = 0;
-		while (tree_iter(&msg->envelopes, &i, NULL, (void*)(&evp)))
-			if (evp->flags & RQ_ENVELOPE_PENDING) {
-				rq_envelope_schedule(&ramqueue, evp);
-				r++;
-			}
+		while (tree_iter(&msg->envelopes, &i, NULL, (void*)(&evp))) {
+			if (evp->state == RQ_EVPSTATE_INFLIGHT)
+				continue;
+			rq_envelope_schedule(&ramqueue, evp);
+			r++;
+		}
 		return (r);
 	}
 }
@@ -611,6 +620,7 @@ rq_queue_init(struct rq_queue *rq)
 	tree_init(&rq->messages);
 	TAILQ_INIT(&rq->q_pending);
 	TAILQ_INIT(&rq->q_inflight);
+	TAILQ_INIT(&rq->q_mta);
 	TAILQ_INIT(&rq->q_mda);
 	TAILQ_INIT(&rq->q_bounce);
 	TAILQ_INIT(&rq->q_expired);
@@ -654,16 +664,15 @@ rq_queue_schedule(struct rq_queue *rq)
 		if (evp->sched > currtime && evp->expire > currtime)
 			break;
 
-		if (evp->flags != RQ_ENVELOPE_PENDING)
+		if (evp->state != RQ_EVPSTATE_PENDING)
 			errx(1, "evp:%016" PRIx64 " flags=0x%x", evp->evpid,
 			    evp->flags);
 
 		if (evp->expire <= currtime) {
 			TAILQ_REMOVE(&rq->q_pending, evp, entry);
 			TAILQ_INSERT_TAIL(&rq->q_expired, evp, entry);
-			evp->flags &= ~RQ_ENVELOPE_PENDING;
+			evp->state = RQ_EVPSTATE_SCHEDULED;
 			evp->flags |= RQ_ENVELOPE_EXPIRED;
-			evp->flags |= RQ_ENVELOPE_SCHEDULED;
 			evp->t_scheduled = currtime;
 			continue;
 		}
@@ -674,25 +683,28 @@ rq_queue_schedule(struct rq_queue *rq)
 static struct evplist *
 rq_envelope_list(struct rq_queue *rq, struct rq_envelope *evp)
 {
-	if (evp->flags & RQ_ENVELOPE_SCHEDULED) {
+	switch (evp->state) {
+	case RQ_EVPSTATE_PENDING:
+		return &rq->q_pending;
+
+	case RQ_EVPSTATE_SCHEDULED:
 		if (evp->flags & RQ_ENVELOPE_EXPIRED)
 			return &rq->q_expired;
 		if (evp->flags & RQ_ENVELOPE_REMOVED)
 			return &rq->q_removed;
 		if (evp->type == D_MTA)
-			return &evp->message->q_mta;
+			return &rq->q_mta;
 		if (evp->type == D_MDA)
 			return &rq->q_mda;
 		if (evp->type == D_BOUNCE)
 			return &rq->q_bounce;
-	}
-
-	if (evp->flags & RQ_ENVELOPE_PENDING)
-		return &rq->q_pending;
+		errx(1, "%016" PRIx64 " bad evp type %d", evp->evpid, evp->type);
 
-	if (evp->flags & RQ_ENVELOPE_INFLIGHT)
+	case RQ_EVPSTATE_INFLIGHT:
 		return &rq->q_inflight;
+	}
 
+	errx(1, "%016" PRIx64 " bad state %d", evp->evpid, evp->state);
 	return (NULL);
 }
 
@@ -701,66 +713,48 @@ rq_envelope_schedule(struct rq_queue *rq, struct rq_envelope *evp)
 {
 	struct evplist	*q = NULL;
 
-	if (evp->type == D_MTA) {
-		if (TAILQ_EMPTY(&evp->message->q_mta)) {
-			evp->message->q_next = rq->q_mtabatch;
-			rq->q_mtabatch = evp->message;
-		}
-		q = &evp->message->q_mta;
-	}
-	else if (evp->type == D_MDA)
+	switch (evp->type) {
+	case D_MTA:
+		q = &rq->q_mta;
+		break;
+	case D_MDA:
 		q = &rq->q_mda;
-	else if (evp->type == D_BOUNCE)
+		break;
+	case D_BOUNCE:
 		q = &rq->q_bounce;
+		break;
+	}
+
+	if (!(evp->flags & RQ_ENVELOPE_SUSPEND))
+		TAILQ_REMOVE(&rq->q_pending, evp, entry);
 
-	TAILQ_REMOVE(&rq->q_pending, evp, entry);
 	TAILQ_INSERT_TAIL(q, evp, entry);
-	evp->flags &= ~RQ_ENVELOPE_PENDING;
-	evp->flags |= RQ_ENVELOPE_SCHEDULED;
+	evp->state = RQ_EVPSTATE_SCHEDULED;
 	evp->t_scheduled = currtime;
 }
 
 static int
 rq_envelope_remove(struct rq_queue *rq, struct rq_envelope *evp)
 {
-	struct rq_message	*m;
-	struct evplist		*q = NULL;
-
 	if (evp->flags & (RQ_ENVELOPE_REMOVED | RQ_ENVELOPE_EXPIRED))
 		return (0);
 	/*
-	 * For now we just ignore it, but we could mark the envelope for
-	 * removal and possibly send a cancellation to the agent.
+	 * If envelope is inflight, mark it envelope for removal.
 	 */
-	if (evp->flags & (RQ_ENVELOPE_INFLIGHT))
-		return (0);
+	if (evp->state == RQ_EVPSTATE_INFLIGHT) {
+		evp->flags |= RQ_ENVELOPE_REMOVED;
+		return (1);
+	}
 
-	q = rq_envelope_list(rq, evp);
+	if (!(evp->flags & RQ_ENVELOPE_SUSPEND)) {
+		TAILQ_REMOVE(rq_envelope_list(rq, evp), evp, entry);
+	}
 
-	TAILQ_REMOVE(q, evp, entry);
 	TAILQ_INSERT_TAIL(&rq->q_removed, evp, entry);
-	evp->flags &= ~RQ_ENVELOPE_PENDING;
+	evp->state = RQ_EVPSTATE_SCHEDULED;
 	evp->flags |= RQ_ENVELOPE_REMOVED;
-	evp->flags |= RQ_ENVELOPE_SCHEDULED;
 	evp->t_scheduled = currtime;
 
-	/*
-	 * We might need to unschedule the message if it was the only
-	 * scheduled envelope
-	 */
-	if (q == &evp->message->q_mta && TAILQ_EMPTY(q)) {
-		if (rq->q_mtabatch == evp->message)
-			rq->q_mtabatch = evp->message->q_next;
-		else {
-			for (m = rq->q_mtabatch; m->q_next; m = m->q_next)
-				if (m->q_next == evp->message) {
-					m->q_next = evp->message->q_next;
-					break;
-				}
-		}
-		evp->message->q_next = NULL;
-	}
-
 	return (1);
 }
 
@@ -770,8 +764,9 @@ rq_envelope_suspend(struct rq_queue *rq, struct rq_envelope *evp)
 	if (evp->flags & RQ_ENVELOPE_SUSPEND)
 		return (0);
 
-	if (!(evp->flags & RQ_ENVELOPE_INFLIGHT))
+	if (evp->state != RQ_EVPSTATE_INFLIGHT) {
 		TAILQ_REMOVE(rq_envelope_list(rq, evp), evp, entry);
+	}
 
 	evp->flags |= RQ_ENVELOPE_SUSPEND;
 
@@ -784,10 +779,11 @@ rq_envelope_resume(struct rq_queue *rq, struct rq_envelope *evp)
 	if (!(evp->flags & RQ_ENVELOPE_SUSPEND))
 		return (0);
 
-	if (!(evp->flags & RQ_ENVELOPE_INFLIGHT))
+	if (evp->state != RQ_EVPSTATE_INFLIGHT)
 		sorted_insert(rq_envelope_list(rq, evp), evp);
 
 	evp->flags &= ~RQ_ENVELOPE_SUSPEND;
+
 	return (1);
 }
 
@@ -825,21 +821,30 @@ rq_envelope_to_text(struct rq_envelope *e)
 	    duration_to_text(e->expire - currtime));
 	strlcat(buf, t, sizeof buf);
 
-	if (e->flags & RQ_ENVELOPE_PENDING) {
+
+	switch (e->state) {
+	case RQ_EVPSTATE_PENDING:
 		snprintf(t, sizeof t, ",pending=%s",
 		    duration_to_text(e->sched - currtime));
 		strlcat(buf, t, sizeof buf);
-	}
-	if (e->flags & RQ_ENVELOPE_SCHEDULED) {
+		break;
+
+	case RQ_EVPSTATE_SCHEDULED:
 		snprintf(t, sizeof t, ",scheduled=%s",
 		    duration_to_text(currtime - e->t_scheduled));
 		strlcat(buf, t, sizeof buf);
-	}
-	if (e->flags & RQ_ENVELOPE_INFLIGHT) {
+		break;
+
+	case RQ_EVPSTATE_INFLIGHT:
 		snprintf(t, sizeof t, ",inflight=%s",
 		    duration_to_text(currtime - e->t_inflight));
 		strlcat(buf, t, sizeof buf);
+		break;
+
+	default:
+		errx(1, "%016" PRIx64 " bad state %d", e->evpid, e->state);
 	}
+
 	if (e->flags & RQ_ENVELOPE_REMOVED)
 		strlcat(buf, ",removed", sizeof buf);
 	if (e->flags & RQ_ENVELOPE_EXPIRED)