- teach smtpctl remove about the new ramqueue structure

- bonus #1: O(log n) removal of envelopes
- bonus #2: removing all envelopes that have the same msgid works again

4 files changed, 82 insertions, 49 deletions

diff --git a/usr.sbin/smtpd/ramqueue.c b/usr.sbin/smtpd/ramqueue.c
index 04dac5193cb..292dc6e0b37 100644
--- a/usr.sbin/smtpd/ramqueue.c
+++ b/usr.sbin/smtpd/ramqueue.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: ramqueue.c,v 1.12 2011/08/17 19:36:23 gilles Exp $	*/
+/*	$OpenBSD: ramqueue.c,v 1.13 2011/08/17 20:04:43 gilles Exp $	*/
 
 /*
  * Copyright (c) 2011 Gilles Chehade <gilles@openbsd.org>
@@ -223,22 +223,6 @@ ramqueue_insert(struct ramqueue *rqueue, struct envelope *envelope, time_t curtm
 	    env->stats->ramqueue.envelopes_max);
 }
 
-void
-ramqueue_remove(struct ramqueue *rqueue, struct ramqueue_envelope *rq_evp)
-{
-	struct ramqueue_batch *rq_batch = rq_evp->batch;
-	struct ramqueue_message *rq_message = rq_evp->message;
-
-	if (rq_evp == rqueue->current_evp)
-		rqueue->current_evp = TAILQ_NEXT(rqueue->current_evp, queue_entry);
-
-	RB_REMOVE(evptree, &rq_message->evptree, rq_evp);
-	TAILQ_REMOVE(&rq_batch->envelope_queue, rq_evp, batchqueue_entry);
-	TAILQ_REMOVE(&rqueue->queue, rq_evp, queue_entry);
-
-	env->stats->ramqueue.envelopes--;
-}
-
 static int
 ramqueue_expire(struct envelope *envelope, time_t curtm)
 {
@@ -383,7 +367,7 @@ ramqueue_remove_message(struct ramqueue *rqueue, struct ramqueue_message *rq_msg
 }
 
 void
-ramqueue_reschedule(struct ramqueue *rq, u_int64_t id)
+ramqueue_schedule(struct ramqueue *rq, u_int64_t id)
 {
 	struct ramqueue_message *rq_msg;
 	struct ramqueue_envelope *rq_evp;
@@ -393,7 +377,7 @@ ramqueue_reschedule(struct ramqueue *rq, u_int64_t id)
 		rq_evp = ramqueue_lookup_envelope(rq, id);
 		if (rq_evp == NULL)
 			return;
-		ramqueue_reschedule_envelope(rq, rq_evp);
+		ramqueue_schedule_envelope(rq, rq_evp);
 		return;
 	}
 
@@ -403,12 +387,12 @@ ramqueue_reschedule(struct ramqueue *rq, u_int64_t id)
 
 	/* scheduling by msgid */
 	RB_FOREACH(rq_evp, evptree, &rq_msg->evptree) {
-		ramqueue_reschedule_envelope(rq, rq_evp);
+		ramqueue_schedule_envelope(rq, rq_evp);
 	}
 }
 
 void
-ramqueue_reschedule_envelope(struct ramqueue *rq, struct ramqueue_envelope *rq_evp)
+ramqueue_schedule_envelope(struct ramqueue *rq, struct ramqueue_envelope *rq_evp)
 {
 	rq_evp->sched = 0;
 	TAILQ_REMOVE(&rq->queue, rq_evp, queue_entry);
@@ -482,6 +466,23 @@ ramqueue_lookup_envelope(struct ramqueue *rq, u_int64_t evpid)
 	return RB_FIND(evptree, &rq_msg->evptree, &evpkey);
 }
 
+void
+ramqueue_remove_envelope(struct ramqueue *rq, struct ramqueue_envelope *rq_evp)
+{
+	struct ramqueue_batch *rq_batch = rq_evp->batch;
+	struct ramqueue_message *rq_msg = rq_evp->message;
+
+	if (rq_evp == rq->current_evp)
+		rq->current_evp = TAILQ_NEXT(rq->current_evp, queue_entry);
+
+	RB_REMOVE(evptree, &rq_msg->evptree, rq_evp);
+	TAILQ_REMOVE(&rq_batch->envelope_queue, rq_evp, batchqueue_entry);
+	TAILQ_REMOVE(&rq->queue, rq_evp, queue_entry);
+
+	env->stats->ramqueue.envelopes--;
+}
+
+
 RB_GENERATE(hosttree, ramqueue_host, hosttree_entry, ramqueue_host_cmp);
 RB_GENERATE(msgtree, ramqueue_message, msgtree_entry, ramqueue_msg_cmp);
 RB_GENERATE(evptree, ramqueue_envelope, evptree_entry, ramqueue_evp_cmp);
diff --git a/usr.sbin/smtpd/runner.c b/usr.sbin/smtpd/runner.c
index e966c0ebc3d..caf51d56048 100644
--- a/usr.sbin/smtpd/runner.c
+++ b/usr.sbin/smtpd/runner.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: runner.c,v 1.109 2011/08/16 19:02:03 gilles Exp $	*/
+/*	$OpenBSD: runner.c,v 1.110 2011/08/17 20:04:43 gilles Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -155,27 +155,14 @@ runner_imsg(struct imsgev *iev, struct imsg *imsg)
 		return;
 
 	case IMSG_RUNNER_SCHEDULE:
-		ramqueue_reschedule(&env->sc_rqueue,
+		runner_schedule(&env->sc_rqueue,
 		    *(u_int64_t *)imsg->data);
 		runner_reset_events();		
 		return;
 
 	case IMSG_RUNNER_REMOVE: {
-		u_int64_t ullval = *(u_int64_t *)imsg->data;
-		struct envelope	 envelope;
-		struct ramqueue_envelope *rq_evp;
-		
-		if (! queue_envelope_load(Q_QUEUE, ullval, &envelope))
-			return;
-		
-		rq_evp = ramqueue_envelope_by_id(&env->sc_rqueue, ullval);
-		if (rq_evp == NULL)
-			return;
-
-		ramqueue_remove(&env->sc_rqueue, rq_evp);
-
-		queue_envelope_delete(Q_QUEUE, &envelope);
-
+		runner_remove(&env->sc_rqueue,
+		    *(u_int64_t *)imsg->data);
 		runner_reset_events();
 		return;
 	}
@@ -437,7 +424,7 @@ runner_process_batch(struct ramqueue_envelope *rq_evp, time_t curtm)
 			imsg_compose_event(env->sc_ievs[PROC_QUEUE],
 			    IMSG_SMTP_ENQUEUE, PROC_SMTP, 0, -1, &envelope,
 			    sizeof envelope);
-			ramqueue_remove(&env->sc_rqueue, rq_evp);
+			ramqueue_remove_envelope(&env->sc_rqueue, rq_evp);
 			free(rq_evp);
 		}
 		env->stats->runner.bounces_active++;
@@ -459,7 +446,7 @@ runner_process_batch(struct ramqueue_envelope *rq_evp, time_t curtm)
 		imsg_compose_event(env->sc_ievs[PROC_QUEUE],
 		    IMSG_MDA_SESS_NEW, PROC_MDA, 0, fd, &envelope,
 		    sizeof envelope);
-		ramqueue_remove(&env->sc_rqueue, rq_evp);
+		ramqueue_remove_envelope(&env->sc_rqueue, rq_evp);
 		free(rq_evp);
 
 		env->stats->mda.sessions_active++;
@@ -484,7 +471,7 @@ runner_process_batch(struct ramqueue_envelope *rq_evp, time_t curtm)
 			imsg_compose_event(env->sc_ievs[PROC_QUEUE],
 			    IMSG_BATCH_APPEND, PROC_MTA, 0, -1, &envelope,
 			    sizeof envelope);
-			ramqueue_remove(&env->sc_rqueue, rq_evp);
+			ramqueue_remove_envelope(&env->sc_rqueue, rq_evp);
 			free(rq_evp);
 			env->stats->runner.active++;
 			SET_IF_GREATER(env->stats->runner.active,
@@ -706,3 +693,46 @@ runner_check_loop(struct envelope *ep)
 	fclose(fp);
 	return ret;
 }
+
+void
+runner_schedule(struct ramqueue *rq, u_int64_t id)
+{
+	ramqueue_schedule(rq, id);
+}
+
+
+void
+runner_remove(struct ramqueue *rq, u_int64_t id)
+{
+	struct ramqueue_message *rq_msg;
+	struct ramqueue_envelope *rq_evp;
+
+	/* removing by evpid */
+	if (id > 0xffffffffL) {
+		rq_evp = ramqueue_lookup_envelope(rq, id);
+		if (rq_evp == NULL)
+			return;
+		runner_remove_envelope(rq, rq_evp);
+		return;
+	}
+
+	rq_msg = ramqueue_lookup_message(rq, id);
+	if (rq_msg == NULL)
+		return;
+
+	/* scheduling by msgid */
+	RB_FOREACH(rq_evp, evptree, &rq_msg->evptree) {
+		runner_remove_envelope(rq, rq_evp);
+	}
+}
+
+void
+runner_remove_envelope(struct ramqueue *rq, struct ramqueue_envelope *rq_evp)
+{
+	struct envelope evp;
+
+	if (queue_envelope_load(Q_QUEUE, rq_evp->evpid, &evp))
+		queue_envelope_delete(Q_QUEUE, &evp);
+
+	ramqueue_remove_envelope(rq, rq_evp);
+}
diff --git a/usr.sbin/smtpd/smtpctl.8 b/usr.sbin/smtpd/smtpctl.8
index 27015b7a046..f53d62eb7b5 100644
--- a/usr.sbin/smtpd/smtpctl.8
+++ b/usr.sbin/smtpd/smtpctl.8
@@ -1,4 +1,4 @@
-.\" $OpenBSD: smtpctl.8,v 1.20 2011/07/22 06:33:54 jmc Exp $
+.\" $OpenBSD: smtpctl.8,v 1.21 2011/08/17 20:04:43 gilles Exp $
 .\"
 .\" Copyright (c) 2006 Pierre-Yves Ritschard <pyr@openbsd.org>
 .\"
@@ -14,7 +14,7 @@
 .\" ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 .\" OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 .\"
-.Dd $Mdocdate: July 22 2011 $
+.Dd $Mdocdate: August 17 2011 $
 .Dt SMTPCTL 8
 .Os
 .Sh NAME
@@ -47,8 +47,8 @@ Temporarily stop deliveries to local users.
 .It Cm pause outgoing
 Temporarily stop relaying and deliveries to
 remote users.
-.It Cm remove Ar envelope-id
-Removes a single envelope from queue.
+.It Cm remove Ar envelope-id | message-id
+Removes a single envelope, or all envelopes with the same message ID.
 .It Cm resume incoming
 Resume accepting incoming sessions.
 .It Cm resume local
diff --git a/usr.sbin/smtpd/smtpd.h b/usr.sbin/smtpd/smtpd.h
index af592ea8e03..3346af51df1 100644
--- a/usr.sbin/smtpd/smtpd.h
+++ b/usr.sbin/smtpd/smtpd.h
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpd.h,v 1.231 2011/08/17 19:36:23 gilles Exp $	*/
+/*	$OpenBSD: smtpd.h,v 1.232 2011/08/17 20:04:43 gilles Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -1095,7 +1095,6 @@ int ramqueue_load_offline(struct ramqueue *);
 int ramqueue_host_cmp(struct ramqueue_host *, struct ramqueue_host *);
 int ramqueue_msg_cmp(struct ramqueue_message *, struct ramqueue_message *);
 int ramqueue_evp_cmp(struct ramqueue_envelope *, struct ramqueue_envelope *);
-void ramqueue_remove(struct ramqueue *, struct ramqueue_envelope *);
 int ramqueue_is_empty(struct ramqueue *);
 int ramqueue_is_empty(struct ramqueue *);
 int ramqueue_batch_is_empty(struct ramqueue_batch *);
@@ -1114,8 +1113,11 @@ struct ramqueue_host *ramqueue_lookup_host(struct ramqueue *, char *);
 struct ramqueue_message *ramqueue_lookup_message(struct ramqueue *, u_int32_t);
 struct ramqueue_envelope *ramqueue_lookup_envelope(struct ramqueue *, u_int64_t);
 
-void ramqueue_reschedule(struct ramqueue *, u_int64_t);
-void ramqueue_reschedule_envelope(struct ramqueue *, struct ramqueue_envelope *);
+void ramqueue_schedule(struct ramqueue *, u_int64_t);
+void ramqueue_schedule_envelope(struct ramqueue *, struct ramqueue_envelope *);
+
+void ramqueue_remove_renvelope(struct ramqueue *, struct ramqueue_envelope *);
+
 
 RB_PROTOTYPE(hosttree, ramqueue_host, hosttree_entry, ramqueue_host_cmp);
 RB_PROTOTYPE(msgtree,  ramqueue_message, msg_entry, ramqueue_msg_cmp);