Improve reporting in smtpctl and a few fixes.

When sending a request to the scheduler, wait for the success/failure
report from the scheduler.  Simplify the code by introducing generic
functions for interruptible iteration over envelopes.  Report the total
number of affected envelopes for schedule, pause, resume and remove
envelope operations.

4 files changed, 164 insertions, 150 deletions

diff --git a/usr.sbin/smtpd/control.c b/usr.sbin/smtpd/control.c
index 5c92ae0243c..d816fd09307 100644
--- a/usr.sbin/smtpd/control.c
+++ b/usr.sbin/smtpd/control.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: control.c,v 1.89 2013/07/19 21:14:52 eric Exp $	*/
+/*	$OpenBSD: control.c,v 1.90 2013/10/25 18:58:10 eric Exp $	*/
 
 /*
  * Copyright (c) 2012 Gilles Chehade <gilles@poolp.org>
@@ -98,10 +98,13 @@ control_imsg(struct mproc *p, struct imsg *imsg)
 	}
 	if (p->proc == PROC_SCHEDULER) {
 		switch (imsg->hdr.type) {
+		case IMSG_CTL_OK:
+		case IMSG_CTL_FAIL:
 		case IMSG_CTL_LIST_MESSAGES:
 			c = tree_get(&ctl_conns, imsg->hdr.peerid);
 			if (c == NULL)
 				return;
+			imsg->hdr.peerid = 0;
 			m_forward(&c->mproc, imsg);
 			return;
 		}
@@ -119,15 +122,11 @@ control_imsg(struct mproc *p, struct imsg *imsg)
 	if (p->proc == PROC_MTA) {
 		switch (imsg->hdr.type) {
 		case IMSG_CTL_MTA_SHOW_ROUTES:
-			c = tree_get(&ctl_conns, imsg->hdr.peerid);
-			if (c == NULL)
-				return;
-			m_forward(&c->mproc, imsg);
-			return;
 		case IMSG_CTL_MTA_SHOW_HOSTSTATS:
 			c = tree_get(&ctl_conns, imsg->hdr.peerid);
 			if (c == NULL)
 				return;
+			imsg->hdr.peerid = 0;
 			m_forward(&c->mproc, imsg);
 			return;
 		}
@@ -579,8 +578,8 @@ control_dispatch_ext(struct mproc *p, struct imsg *imsg)
 		if (c->euid)
 			goto badcred;
 
+		imsg->hdr.peerid = c->id;
 		m_forward(p_scheduler, imsg);
-		m_compose(p, IMSG_CTL_OK, 0, 0, -1, NULL, 0);
 		return;
 
 	case IMSG_CTL_PAUSE_MDA:
@@ -629,8 +628,8 @@ control_dispatch_ext(struct mproc *p, struct imsg *imsg)
 		if (c->euid)
 			goto badcred;
 
+		imsg->hdr.peerid = c->id;
 		m_forward(p_scheduler, imsg);
-		m_compose(p, IMSG_CTL_OK, 0, 0, -1, NULL, 0);
 		return;
 
 	case IMSG_CTL_RESUME_MDA:
@@ -699,33 +698,28 @@ control_dispatch_ext(struct mproc *p, struct imsg *imsg)
 		return;
 
 	case IMSG_CTL_MTA_SHOW_ROUTES:
-		if (c->euid)
-			goto badcred;
-		m_compose(p_mta, IMSG_CTL_MTA_SHOW_ROUTES, c->id, 0, -1,
-		    imsg->data, imsg->hdr.len - sizeof(imsg->hdr));
-		return;
-
 	case IMSG_CTL_MTA_SHOW_HOSTSTATS:
 		if (c->euid)
 			goto badcred;
-		m_compose(p_mta, IMSG_CTL_MTA_SHOW_HOSTSTATS, c->id, 0, -1,
-		    imsg->data, imsg->hdr.len - sizeof(imsg->hdr));
+
+		imsg->hdr.peerid = c->id;
+		m_forward(p_mta, imsg);
 		return;
 
 	case IMSG_CTL_SCHEDULE:
 		if (c->euid)
 			goto badcred;
 
+		imsg->hdr.peerid = c->id;
 		m_forward(p_scheduler, imsg);
-		m_compose(p, IMSG_CTL_OK, 0, 0, -1, NULL, 0);
 		return;
 
 	case IMSG_CTL_REMOVE:
 		if (c->euid)
 			goto badcred;
 
+		imsg->hdr.peerid = c->id;
 		m_forward(p_scheduler, imsg);
-		m_compose(p, IMSG_CTL_OK, 0, 0, -1, NULL, 0);
 		return;
 
 	case IMSG_LKA_UPDATE_TABLE:
diff --git a/usr.sbin/smtpd/parser.c b/usr.sbin/smtpd/parser.c
index a8d174a9956..aea11b37921 100644
--- a/usr.sbin/smtpd/parser.c
+++ b/usr.sbin/smtpd/parser.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: parser.c,v 1.35 2013/07/19 13:41:23 eric Exp $	*/
+/*	$OpenBSD: parser.c,v 1.36 2013/10/25 18:58:10 eric Exp $	*/
 
 /*
  * Copyright (c) 2013 Eric Faurot	<eric@openbsd.org>
@@ -215,7 +215,7 @@ cmd_run(int argc, char **argv)
 	if (node->cmd == NULL)
 		goto fail;
 
-	return (node->cmd(np, param));
+	return (node->cmd(np, np ? param : NULL));
 
 fail:
 	fprintf(stderr, "possibilities are:\n");
diff --git a/usr.sbin/smtpd/scheduler.c b/usr.sbin/smtpd/scheduler.c
index fb06b6bc016..cc691d87cc0 100644
--- a/usr.sbin/smtpd/scheduler.c
+++ b/usr.sbin/smtpd/scheduler.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: scheduler.c,v 1.32 2013/07/19 21:14:52 eric Exp $	*/
+/*	$OpenBSD: scheduler.c,v 1.33 2013/10/25 18:58:10 eric Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@poolp.org>
@@ -79,7 +79,7 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 	uint32_t       		 penalty;
 	size_t			 n, i;
 	time_t			 timestamp;
-	int			 v;
+	int			 v, r;
 
 	switch (imsg->hdr.type) {
 
@@ -263,8 +263,10 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 		else
 			log_debug("debug: scheduler: "
 			    "scheduling evp:%016" PRIx64, id);
-		backend->schedule(id);
+		r = backend->schedule(id);
 		scheduler_reset_events();
+		m_compose(p, r ? IMSG_CTL_OK : IMSG_CTL_FAIL, imsg->hdr.peerid,
+		    0, -1, NULL, 0);
 		return;
 
 	case IMSG_MTA_SCHEDULE:
@@ -281,8 +283,10 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 		else
 			log_debug("debug: scheduler: "
 			    "removing evp:%016" PRIx64, id);
-		backend->remove(id);
+		r = backend->remove(id);
 		scheduler_reset_events();
+		m_compose(p, r ? IMSG_CTL_OK : IMSG_CTL_FAIL, imsg->hdr.peerid,
+		    0, -1, NULL, 0);
 		return;
 
 	case IMSG_CTL_PAUSE_EVP:
@@ -293,8 +297,10 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 		else
 			log_debug("debug: scheduler: "
 			    "suspending evp:%016" PRIx64, id);
-		backend->suspend(id);
+		r = backend->suspend(id);
 		scheduler_reset_events();
+		m_compose(p, r ? IMSG_CTL_OK : IMSG_CTL_FAIL, imsg->hdr.peerid,
+		    0, -1, NULL, 0);
 		return;
 
 	case IMSG_CTL_RESUME_EVP:
@@ -305,8 +311,10 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 		else
 			log_debug("debug: scheduler: "
 			    "resuming evp:%016" PRIx64, id);
-		backend->resume(id);
+		r = backend->resume(id);
 		scheduler_reset_events();
+		m_compose(p, r ? IMSG_CTL_OK : IMSG_CTL_FAIL, imsg->hdr.peerid,
+		    0, -1, NULL, 0);
 		return;
 	}
 
diff --git a/usr.sbin/smtpd/smtpctl.c b/usr.sbin/smtpd/smtpctl.c
index 7d453410690..006fed00cac 100644
--- a/usr.sbin/smtpd/smtpctl.c
+++ b/usr.sbin/smtpd/smtpctl.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpctl.c,v 1.108 2013/07/19 21:14:52 eric Exp $	*/
+/*	$OpenBSD: smtpctl.c,v 1.109 2013/10/25 18:58:10 eric Exp $	*/
 
 /*
  * Copyright (c) 2013 Eric Faurot <eric@openbsd.org>
@@ -49,6 +49,8 @@
 #define	PATH_CAT	"/bin/cat"
 #define PATH_QUEUE	"/queue"
 
+int srv_connect(void);
+
 void usage(void);
 static void show_queue_envelope(struct envelope *, int);
 static void getflag(uint *, int, char *, char *, size_t);
@@ -96,7 +98,7 @@ void stat_decrement(const char *k, size_t v)
 {
 }
 
-static int
+int
 srv_connect(void)
 {
 	struct sockaddr_un	sun;
@@ -189,20 +191,23 @@ srv_end(void)
 }
 
 static int
-srv_check_result(void)
+srv_check_result(int verbose_)
 {
 	srv_recv(-1);
 	srv_end();
 
 	switch (imsg.hdr.type) {
 	case IMSG_CTL_OK:
-		printf("command succeeded\n");
+		if (verbose_)
+			printf("command succeeded\n");
 		return (0);
 	case IMSG_CTL_FAIL:
-		if (rlen)
-			printf("command failed: %s\n", rdata);
-		else
-			printf("command failed\n");
+		if (verbose_) {
+			if (rlen)
+				printf("command failed: %s\n", rdata);
+			else
+				printf("command failed\n");
+		}
 		return (1);
 	default:
 		errx(1, "wrong message in response: %u", imsg.hdr.type);
@@ -294,12 +299,100 @@ srv_iter_envelopes(uint32_t msgid, struct envelope *evp)
 }
 
 static int
+srv_iter_evpids(uint32_t msgid, uint64_t *evpid, int *offset)
+{
+	static uint64_t	*evpids = NULL;
+	static int	 n, alloc = 0;
+	struct envelope	 evp;
+
+	if (evpids == NULL) {
+		alloc = 1000;
+		evpids = malloc(alloc * sizeof(*evpids));
+		if (evpids == NULL)
+			err(1, "malloc");
+	}
+
+	if (*offset == 0) {
+		n = 0;
+		while (srv_iter_envelopes(msgid, &evp)) {
+			if (n == alloc) {
+				alloc += 256;
+				evpids = realloc(evpids, alloc * sizeof(*evpids));
+				if (evpids == NULL)
+					err(1, "realloc");
+			}
+			evpids[n++] = evp.id;
+		}
+	}
+
+	if (*offset >= n)
+		return (0);
+	*evpid = evpids[*offset];
+	*offset += 1;
+	return (1);
+}
+
+static void
+srv_foreach_envelope(struct parameter *argv, int ctl, size_t *total, size_t *ok)
+{
+	uint32_t	msgid;
+	uint64_t	evpid;
+	int		i;
+
+	*total = 0;
+	*ok = 0;
+
+	if (argv == NULL) {
+		while (srv_iter_messages(&msgid)) {
+			i = 0;
+			while (srv_iter_evpids(msgid, &evpid, &i)) {
+				*total += 1;
+				srv_send(ctl, &evpid, sizeof(evpid));
+				if (srv_check_result(0) == 0)
+					*ok += 1;
+			}
+		}
+	} else if (argv->type == P_MSGID) {
+		i = 0;
+		while (srv_iter_evpids(argv->u.u_msgid, &evpid, &i)) {
+			srv_send(ctl, &evpid, sizeof(evpid));
+			if (srv_check_result(0) == 0)
+				*ok += 1;
+		}
+	} else {
+		*total += 1;
+		srv_send(ctl, &argv->u.u_evpid, sizeof(evpid));
+		if (srv_check_result(0) == 0)
+			*ok += 1;
+	}
+}
+
+static void
+srv_show_cmd(int cmd, const void *data, size_t len)
+{
+	int	done = 0;
+
+	srv_send(cmd, data, len);
+
+	do {
+		srv_recv(cmd);
+		if (rlen) {
+			printf("%s\n", rdata);
+			srv_read(NULL, rlen);
+		}
+		else
+			done = 1;
+		srv_end();
+	} while (!done);
+}
+
+static int
 do_log_brief(int argc, struct parameter *argv)
 {
 	int	v = 0;
 
 	srv_send(IMSG_CTL_VERBOSE, &v, sizeof(v));
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
@@ -308,7 +401,7 @@ do_log_verbose(int argc, struct parameter *argv)
 	int	v = TRACE_DEBUG;
 
 	srv_send(IMSG_CTL_VERBOSE, &v, sizeof(v));
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
@@ -370,26 +463,10 @@ do_monitor(int argc, struct parameter *argv)
 static int
 do_pause_envelope(int argc, struct parameter *argv)
 {
-	struct envelope	 evp;
-	uint32_t	 msgid;
+	size_t	total, ok;
 
-	if (argc == 0) {
-		while (srv_iter_messages(&msgid)) {
-			while (srv_iter_envelopes(msgid, &evp)) {
-				srv_send(IMSG_CTL_PAUSE_EVP, &evp.id,
-				    sizeof(evp.id));
-				srv_check_result();
-			}
-		}
-	} else if (argv[0].type == P_MSGID) {
-		while (srv_iter_envelopes(argv[0].u.u_msgid, &evp)) {
-			srv_send(IMSG_CTL_PAUSE_EVP, &evp.id, sizeof(evp.id));
-			srv_check_result();
-		}
-	} else {
-		srv_send(IMSG_CTL_PAUSE_EVP, &argv[0].u.u_evpid, sizeof(evp.id));
-		srv_check_result();
-	}
+	srv_foreach_envelope(argv, IMSG_CTL_PAUSE_EVP, &total, &ok);
+	printf("%zu envelope%s paused\n", ok, (ok > 1) ? "s" : "");
 
 	return (0);
 }
@@ -398,21 +475,21 @@ static int
 do_pause_mda(int argc, struct parameter *argv)
 {
 	srv_send(IMSG_CTL_PAUSE_MDA, NULL, 0);
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
 do_pause_mta(int argc, struct parameter *argv)
 {
 	srv_send(IMSG_CTL_PAUSE_MTA, NULL, 0);
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
 do_pause_smtp(int argc, struct parameter *argv)
 {
 	srv_send(IMSG_CTL_PAUSE_SMTP, NULL, 0);
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
@@ -423,32 +500,16 @@ do_profile(int argc, struct parameter *argv)
 	v = str_to_profile(argv[0].u.u_str);
 
 	srv_send(IMSG_CTL_PROFILE, &v, sizeof(v));
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
 do_remove(int argc, struct parameter *argv)
 {
-	struct envelope	 evp;
-	uint32_t	 msgid;
+	size_t	total, ok;
 
-	if (argc == 0) {
-		while (srv_iter_messages(&msgid)) {
-			while (srv_iter_envelopes(msgid, &evp)) {
-				srv_send(IMSG_CTL_REMOVE, &evp.id,
-				    sizeof(evp.id));
-				srv_check_result();
-			}
-		}
-	} else if (argv[0].type == P_MSGID) {
-		while (srv_iter_envelopes(argv[0].u.u_msgid, &evp)) {
-			srv_send(IMSG_CTL_REMOVE, &evp.id, sizeof(evp.id));
-			srv_check_result();
-		}
-	} else {
-		srv_send(IMSG_CTL_REMOVE, &argv[0].u.u_evpid, sizeof(evp.id));
-		srv_check_result();
-	}
+	srv_foreach_envelope(argv, IMSG_CTL_REMOVE, &total, &ok);
+	printf("%zu envelope%s removed\n", ok, (ok > 1) ? "s" : "");
 
 	return (0);
 }
@@ -456,26 +517,10 @@ do_remove(int argc, struct parameter *argv)
 static int
 do_resume_envelope(int argc, struct parameter *argv)
 {
-	struct envelope	 evp;
-	uint32_t	 msgid;
+	size_t	total, ok;
 
-	if (argc == 0) {
-		while (srv_iter_messages(&msgid)) {
-			while (srv_iter_envelopes(msgid, &evp)) {
-				srv_send(IMSG_CTL_RESUME_EVP, &evp.id,
-				    sizeof(evp.id));
-				srv_check_result();
-			}
-		}
-	} else if (argv[0].type == P_MSGID) {
-		while (srv_iter_envelopes(argv[0].u.u_msgid, &evp)) {
-			srv_send(IMSG_CTL_RESUME_EVP, &evp.id, sizeof(evp.id));
-			srv_check_result();
-		}
-	} else {
-		srv_send(IMSG_CTL_RESUME_EVP, &argv[0].u.u_evpid, sizeof(evp.id));
-		srv_check_result();
-	}
+	srv_foreach_envelope(argv, IMSG_CTL_RESUME_EVP, &total, &ok);
+	printf("%zu envelope%s resumed\n", ok, (ok > 1) ? "s" : "");
 
 	return (0);
 }
@@ -484,14 +529,14 @@ static int
 do_resume_mda(int argc, struct parameter *argv)
 {
 	srv_send(IMSG_CTL_RESUME_MDA, NULL, 0);
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
 do_resume_mta(int argc, struct parameter *argv)
 {
 	srv_send(IMSG_CTL_RESUME_MTA, NULL, 0);
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
@@ -505,40 +550,23 @@ do_resume_route(int argc, struct parameter *argv)
 		v = argv[0].u.u_routeid;
 
 	srv_send(IMSG_CTL_RESUME_ROUTE, &v, sizeof(v));
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
 do_resume_smtp(int argc, struct parameter *argv)
 {
 	srv_send(IMSG_CTL_RESUME_SMTP, NULL, 0);
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
 do_schedule(int argc, struct parameter *argv)
 {
-	struct envelope	 evp;
-	uint32_t	 msgid;
+	size_t	total, ok;
 
-	if (argc == 0) {
-		while (srv_iter_messages(&msgid)) {
-			while (srv_iter_envelopes(msgid, &evp)) {
-				srv_send(IMSG_CTL_SCHEDULE, &evp.id,
-				    sizeof(evp.id));
-				srv_check_result();
-			}
-		}
-
-	} else if (argv[0].type == P_MSGID) {
-		while (srv_iter_envelopes(argv[0].u.u_msgid, &evp)) {
-			srv_send(IMSG_CTL_SCHEDULE, &evp.id, sizeof(evp.id));
-			srv_check_result();
-		}
-	} else {
-		srv_send(IMSG_CTL_SCHEDULE, &argv[0].u.u_evpid, sizeof(evp.id));
-		srv_check_result();
-	}
+	srv_foreach_envelope(argv, IMSG_CTL_SCHEDULE, &total, &ok);
+	printf("%zu envelope%s scheduled\n", ok, (ok > 1) ? "s" : "");
 
 	return (0);
 }
@@ -564,16 +592,7 @@ do_show_envelope(int argc, struct parameter *argv)
 static int
 do_show_hoststats(int argc, struct parameter *argv)
 {
-	srv_send(IMSG_CTL_MTA_SHOW_HOSTSTATS, NULL, 0);
-
-	do {
-		srv_recv(IMSG_CTL_MTA_SHOW_HOSTSTATS);
-		if (rlen) {
-			printf("%s\n", rdata);
-			srv_read(NULL, rlen);
-		}
-		srv_end();
-	} while (rlen);
+	srv_show_cmd(IMSG_CTL_MTA_SHOW_HOSTSTATS, NULL, 0);
 
 	return (0);
 }
@@ -662,16 +681,7 @@ do_show_queue(int argc, struct parameter *argv)
 static int
 do_show_routes(int argc, struct parameter *argv)
 {
-	srv_send(IMSG_CTL_MTA_SHOW_ROUTES, NULL, 0);
-
-	do {
-		srv_recv(IMSG_CTL_MTA_SHOW_ROUTES);
-		if (rlen) {
-			printf("%s\n", rdata);
-			srv_read(NULL, rlen);
-		}
-		srv_end();
-	} while (rlen);
+	srv_show_cmd(IMSG_CTL_MTA_SHOW_ROUTES, NULL, 0);
 
 	return (0);
 }
@@ -732,7 +742,7 @@ static int
 do_stop(int argc, struct parameter *argv)
 {
 	srv_send(IMSG_CTL_SHUTDOWN, NULL, 0);
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
@@ -743,7 +753,7 @@ do_trace(int argc, struct parameter *argv)
 	v = str_to_trace(argv[0].u.u_str);
 
 	srv_send(IMSG_CTL_TRACE, &v, sizeof(v));
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
@@ -754,7 +764,7 @@ do_unprofile(int argc, struct parameter *argv)
 	v = str_to_profile(argv[0].u.u_str);
 
 	srv_send(IMSG_CTL_UNPROFILE, &v, sizeof(v));
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
@@ -765,7 +775,7 @@ do_untrace(int argc, struct parameter *argv)
 	v = str_to_trace(argv[0].u.u_str);
 
 	srv_send(IMSG_CTL_UNTRACE, &v, sizeof(v));
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 static int
@@ -774,7 +784,7 @@ do_update_table(int argc, struct parameter *argv)
 	const char	*name = argv[0].u.u_str;
 
 	srv_send(IMSG_LKA_UPDATE_TABLE, name, strlen(name) + 1);
-	return srv_check_result();
+	return srv_check_result(1);
 }
 
 int
@@ -937,9 +947,10 @@ show_offline_envelope(uint64_t evpid)
 	if (fp == NULL)
 		goto end;
 
-	buflen = fread(buffer, 1, sizeof buffer, fp);
+	buflen = fread(buffer, 1, sizeof (buffer) - 1, fp);
 	p = buffer;
 	plen = buflen;
+	buffer[buflen] = '\0';
 
 	if (is_encrypted_buffer(p)) {
 		warnx("offline encrypted queue is not supported yet");
@@ -1003,10 +1014,11 @@ display(const char *s)
 
 		fclose(fp);
 		fp = ofp;
-		fseek(fp, SEEK_SET, 0);
+		fseek(fp, 0, SEEK_SET);
 	}
 	gzipped = is_gzip_fp(fp);
 
+	lseek(fileno(fp), 0, SEEK_SET);
 	(void)dup2(fileno(fp), STDIN_FILENO);
 	if (gzipped)
 		execl(PATH_GZCAT, gzcat_argv0, NULL);
@@ -1080,7 +1092,7 @@ is_gzip_fp(FILE *fp)
 
 	ret = is_gzip_buffer((const char *)&magic);
 end:
-	fseek(fp, SEEK_SET, 0);
+	fseek(fp, 0, SEEK_SET);
 	return ret;
 }
 
@@ -1114,6 +1126,6 @@ is_encrypted_fp(FILE *fp)
 
 	ret = is_encrypted_buffer((const char *)&magic);
 end:
-	fseek(fp, SEEK_SET, 0);
+	fseek(fp, 0, SEEK_SET);
 	return ret;
 }