Let the smtpd process handle the enqueueing of offline messages at

startup, rather than playing tricks with the runner. This will allow
further simplifications and improvements in the runner/queue.

ok gilles@

5 files changed, 91 insertions, 106 deletions

diff --git a/usr.sbin/smtpd/queue.c b/usr.sbin/smtpd/queue.c
index c041225b4a9..9d18574c42f 100644
--- a/usr.sbin/smtpd/queue.c
+++ b/usr.sbin/smtpd/queue.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: queue.c,v 1.109 2011/10/26 20:47:31 gilles Exp $	*/
+/*	$OpenBSD: queue.c,v 1.110 2011/11/07 11:14:10 eric Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -196,10 +196,6 @@ queue_imsg(struct imsgev *iev, struct imsg *imsg)
 
 	if (iev->proc == PROC_PARENT) {
 		switch (imsg->hdr.type) {
-		case IMSG_PARENT_ENQUEUE_OFFLINE:
-			queue_pass_to_runner(iev, imsg);
-			return;
-
 		case IMSG_CTL_VERBOSE:
 			log_verbose(*(int *)imsg->data);
 			queue_pass_to_runner(iev, imsg);
diff --git a/usr.sbin/smtpd/ramqueue.c b/usr.sbin/smtpd/ramqueue.c
index ff95dfef1ed..570b0cf62fd 100644
--- a/usr.sbin/smtpd/ramqueue.c
+++ b/usr.sbin/smtpd/ramqueue.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: ramqueue.c,v 1.24 2011/10/27 14:32:57 chl Exp $	*/
+/*	$OpenBSD: ramqueue.c,v 1.25 2011/11/07 11:14:10 eric Exp $	*/
 
 /*
  * Copyright (c) 2011 Gilles Chehade <gilles@openbsd.org>
@@ -45,7 +45,6 @@ void ramqueue_insert(struct ramqueue *, struct envelope *, time_t);
 int ramqueue_host_cmp(struct ramqueue_host *, struct ramqueue_host *);
 void ramqueue_put_host(struct ramqueue *, struct ramqueue_host *);
 void ramqueue_put_batch(struct ramqueue *, struct ramqueue_batch *);
-int ramqueue_load_offline(struct ramqueue *);
 
 static int ramqueue_expire(struct envelope *, time_t);
 static time_t ramqueue_next_schedule(struct envelope *, time_t);
@@ -111,28 +110,6 @@ ramqueue_batch_first_envelope(struct ramqueue_batch *rq_batch)
 }
 
 int
-ramqueue_load_offline(struct ramqueue *rqueue)
-{
-	char		 path[MAXPATHLEN];
-	static struct qwalk    *q = NULL;
-
-	log_debug("ramqueue: offline queue loading in progress");
-	if (q == NULL)
-		q = qwalk_new(PATH_OFFLINE);
-	while (qwalk(q, path)) {
-		imsg_compose_event(env->sc_ievs[PROC_QUEUE],
-		    IMSG_PARENT_ENQUEUE_OFFLINE, PROC_PARENT, 0, -1, path,
-		    strlen(path) + 1);
-		log_debug("ramqueue: offline queue loading interrupted");
-		return 0;
-	}
-	qwalk_close(q);
-	q = NULL;
-	log_debug("ramqueue: offline queue loading over");
-	return 1;
-}
-
-int
 ramqueue_load(struct ramqueue *rqueue, time_t *nsched)
 {
 	char			path[MAXPATHLEN];
diff --git a/usr.sbin/smtpd/runner.c b/usr.sbin/smtpd/runner.c
index cad84049f80..3ec10dcd932 100644
--- a/usr.sbin/smtpd/runner.c
+++ b/usr.sbin/smtpd/runner.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: runner.c,v 1.122 2011/10/27 14:32:57 chl Exp $	*/
+/*	$OpenBSD: runner.c,v 1.123 2011/11/07 11:14:10 eric Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -128,10 +128,6 @@ runner_imsg(struct imsgev *iev, struct imsg *imsg)
 		stat_decrement(STATS_MTA_SESSION);
 		return;
 
-	case IMSG_PARENT_ENQUEUE_OFFLINE:
-		/*		runner_process_offline();*/
-		return;
-
 	case IMSG_SMTP_ENQUEUE:
 		e = imsg->data;
 		if (imsg->fd < 0 || !bounce_session(imsg->fd, e)) {
@@ -306,7 +302,6 @@ runner_timeout(int fd, short event, void *p)
 	struct ramqueue_envelope *rq_evp;
 	struct timeval		 tv;
 	static int		 rq_done = 0;
-	static int		 rq_off_done = 0;
 	time_t			 nsched;
 	time_t			 curtm;
 
@@ -317,15 +312,6 @@ runner_timeout(int fd, short event, void *p)
 	if (rq_evp)
 		nsched = rq_evp->sched;
 
-
-	/* fetch one offline message at a time to prevent a huge
-	 * offline queue from hogging the deliveries of incoming
-	 * messages.
-	 */
-	if (! rq_off_done)
-		rq_off_done = ramqueue_load_offline(rqueue);
-
-
 	/* load as many envelopes as possible from disk-queue to
 	 * ram-queue until a schedulable envelope is found.
 	 */
@@ -346,13 +332,13 @@ runner_timeout(int fd, short event, void *p)
 	}
 
 	if (rq_evp == NULL ||
-	    (rq_done && rq_off_done && ramqueue_is_empty(rqueue))) {
+	    (rq_done && ramqueue_is_empty(rqueue))) {
 		log_debug("runner: nothing to schedule, wake me up. zZzZzZ");
 		return;
 	}
 
 	/* disk-queues not fully loaded, no time for sleeping */
-	if (!rq_done || !rq_off_done)
+	if (!rq_done)
 		nsched = 0;
 	else {
 		nsched = nsched - curtm;
diff --git a/usr.sbin/smtpd/smtpd.c b/usr.sbin/smtpd/smtpd.c
index 60fb1190e21..90d95a2b6e9 100644
--- a/usr.sbin/smtpd/smtpd.c
+++ b/usr.sbin/smtpd/smtpd.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpd.c,v 1.134 2011/10/26 20:47:31 gilles Exp $	*/
+/*	$OpenBSD: smtpd.c,v 1.135 2011/11/07 11:14:10 eric Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -28,6 +28,7 @@
 #include <sys/uio.h>
 #include <sys/mman.h>
 
+#include <dirent.h>
 #include <err.h>
 #include <errno.h>
 #include <fcntl.h>
@@ -54,7 +55,6 @@ static void parent_send_config_client_certs(void);
 static void parent_send_config_ruleset(int);
 static void parent_sig_handler(int, short, void *);
 static void forkmda(struct imsgev *, u_int32_t, struct deliver *);
-static int parent_enqueue_offline(char *);
 static int parent_forward_open(char *);
 static int path_starts_with(char *, char *);
 static void fork_peers(void);
@@ -62,25 +62,30 @@ static struct child *child_lookup(pid_t);
 static struct child *child_add(pid_t, int, int);
 static void child_del(pid_t);
 
-static int	queueing_add(char *);
-static void	queueing_done(void);
+static void	offline_scan(int, short, void *);
+static int	offline_add(char *);
+static void	offline_done(void);
+static int	offline_enqueue(char *);
 
-struct queueing {
-	TAILQ_ENTRY(queueing)	 entry;
+struct offline {
+	TAILQ_ENTRY(offline)	 entry;
 	char			*path;
 };
 
-#define QUEUEING_MAX 5
-static size_t			queueing_running = 0;
-TAILQ_HEAD(, queueing)		queueing_q;
+#define OFFLINE_READMAX		20
+#define OFFLINE_QUEUEMAX	5
+static size_t			offline_running = 0;
+TAILQ_HEAD(, offline)		offline_q;
+
+static struct event		offline_ev;
+static struct timeval		offline_timeout;
+
 
 extern char	**environ;
 void		(*imsg_callback)(struct imsgev *, struct imsg *);
 
 struct smtpd	*env = NULL;
 
-int __b64_pton(char const *, unsigned char *, size_t);
-
 static void
 parent_imsg(struct imsgev *iev, struct imsg *imsg)
 {
@@ -126,17 +131,6 @@ parent_imsg(struct imsgev *iev, struct imsg *imsg)
 		}
 	}
 
-	if (iev->proc == PROC_QUEUE) {
-		switch (imsg->hdr.type) {
-		case IMSG_PARENT_ENQUEUE_OFFLINE:
-			if (! queueing_add(imsg->data))
-				imsg_compose_event(iev,
-				    IMSG_PARENT_ENQUEUE_OFFLINE, 0, 0, -1,
-				    NULL, 0);
-			return;
-		}
-	}
-
 	if (iev->proc == PROC_MDA) {
 		switch (imsg->hdr.type) {
 		case IMSG_PARENT_FORK_MDA:
@@ -383,14 +377,9 @@ parent_sig_handler(int sig, short event, void *p)
 
 			case CHILD_ENQUEUE_OFFLINE:
 				if (fail)
-					log_warnx("couldn't enqueue offline "
+					log_warnx("smtpd: couldn't enqueue offline "
 					    "message; smtpctl %s", cause);
-				else
-					log_debug("offline message enqueued");
-				imsg_compose_event(env->sc_ievs[PROC_QUEUE],
-				    IMSG_PARENT_ENQUEUE_OFFLINE, 0, 0, -1,
-				    NULL, 0);
-				queueing_done();
+				offline_done();
 				break;
 
 			default:
@@ -440,7 +429,7 @@ main(int argc, char *argv[])
 
 	log_init(1);
 
-	TAILQ_INIT(&queueing_q);
+	TAILQ_INIT(&offline_q);
 
 	while ((c = getopt(argc, argv, "dD:nf:T:v")) != -1) {
 		switch (c) {
@@ -549,6 +538,12 @@ main(int argc, char *argv[])
 	bzero(&tv, sizeof(tv));
 	evtimer_add(&env->sc_ev, &tv);
 
+	/* defer offline scanning for a second */
+	evtimer_set(&offline_ev, offline_scan, NULL);
+	offline_timeout.tv_sec = 1;
+	offline_timeout.tv_usec = 0;
+	evtimer_add(&offline_ev, &offline_timeout);
+
 	if (event_dispatch() < 0)
 		fatal("event_dispatch");
 
@@ -880,51 +875,84 @@ forkmda(struct imsgev *iev, u_int32_t id,
 #undef error
 #undef error2
 
+static void
+offline_scan(int fd, short ev, void *arg)
+{
+	DIR		*dir = arg;
+	struct dirent	*d;
+	int		 n = 0;
+
+	if (dir == NULL) {
+		log_debug("smtpd: scanning offline queue...");
+		if ((dir = opendir(PATH_SPOOL PATH_OFFLINE)) == NULL)
+			errx(1, "smtpd: opendir");
+	}
+
+	while((d = readdir(dir)) != NULL) {
+		if (d->d_type != DT_REG)
+			continue;
+
+		offline_add(d->d_name);
+
+		if ((n++) == OFFLINE_READMAX) {
+			evtimer_set(&offline_ev, offline_scan, dir);
+			offline_timeout.tv_sec = 0;
+			offline_timeout.tv_usec = 100000;
+			evtimer_add(&offline_ev, &offline_timeout);
+			return;
+		}
+	}
+
+	log_debug("smtpd: offline scanning done");
+	closedir(dir);
+}
+
 static int
-parent_enqueue_offline(char *runner_path)
+offline_enqueue(char *name)
 {
-	char		 path[MAXPATHLEN];
+	char			path[MAXPATHLEN];
 	struct user_backend *ub;
 	struct user	 u;
 	struct stat	 sb;
 	pid_t		 pid;
 
-	log_debug("parent_enqueue_offline: path %s", runner_path);
-
-	if (! bsnprintf(path, sizeof(path), "%s%s", PATH_SPOOL, runner_path))
-		fatalx("parent_enqueue_offline: filename too long");
+	strlcpy(path, PATH_SPOOL PATH_OFFLINE "/", sizeof path);
+	if (strlcat(path, name, sizeof path) >= sizeof path)
+		fatalx("smtpd: path name too long");
 
 	if (! path_starts_with(path, PATH_SPOOL PATH_OFFLINE))
-		fatalx("parent_enqueue_offline: path outside offline dir");
+		fatalx("smtpd: path outside offline dir");
+
+	log_debug("smtpd: enqueueing offline message %s", path);
 
 	if (lstat(path, &sb) == -1) {
 		if (errno == ENOENT) {
-			log_warn("parent_enqueue_offline: %s", path);
+			log_warn("smtpd: lstat: %s", path);
 			return (0);
 		}
-		fatal("parent_enqueue_offline: lstat");
+		fatal("lstat");
 	}
 
 	if (chflags(path, 0) == -1) {
 		if (errno == ENOENT) {
-			log_warn("parent_enqueue_offline: %s", path);
+			log_warn("smtpd: chflags: %s", path);
 			return (0);
 		}
-		fatal("parent_enqueue_offline: chflags");
+		fatal("chflags");
 	}
 
 	ub = user_backend_lookup(USER_GETPWNAM);
 	bzero(&u, sizeof (u));
 	errno = 0;
 	if (! ub->getbyuid(&u, sb.st_uid)) {
-		log_warn("parent_enqueue_offline: getpwuid for uid %d failed",
+		log_warn("smtpd: getpwuid for uid %d failed",
 		    sb.st_uid);
 		unlink(path);
 		return (0);
 	}
 
 	if (! S_ISREG(sb.st_mode)) {
-		log_warnx("file %s (uid %d) not regular, removing", path, sb.st_uid);
+		log_warnx("smtpd: file %s (uid %d) not regular, removing", path, sb.st_uid);
 		if (S_ISDIR(sb.st_mode))
 			rmdir(path);
 		else
@@ -933,7 +961,7 @@ parent_enqueue_offline(char *runner_path)
 	}
 
 	if ((pid = fork()) == -1)
-		fatal("parent_enqueue_offline: fork");
+		fatal("offline_enqueue: fork");
 
 	if (pid == 0) {
 		char	*envp[2], *p, *tmp;
@@ -988,43 +1016,43 @@ parent_enqueue_offline(char *runner_path)
 		_exit(1);
 	}
 
-	queueing_running++;
+	offline_running++;
 	child_add(pid, CHILD_ENQUEUE_OFFLINE, -1);
 
 	return (1);
 }
 
 static int
-queueing_add(char *path)
+offline_add(char *path)
 {
-	struct queueing	*q;
+	struct offline	*q;
 
-	if (queueing_running < QUEUEING_MAX)
+	if (offline_running < OFFLINE_QUEUEMAX)
 		/* skip queue */
-		return parent_enqueue_offline(path);
+		return offline_enqueue(path);
 
 	q = malloc(sizeof(*q) + strlen(path) + 1);
 	if (q == NULL)
 		return (-1);
 	q->path = (char *)q + sizeof(*q);
 	memmove(q->path, path, strlen(path) + 1);
-	TAILQ_INSERT_TAIL(&queueing_q, q, entry);
+	TAILQ_INSERT_TAIL(&offline_q, q, entry);
 
 	return (1);
 }
 
 static void
-queueing_done(void)
+offline_done(void)
 {
-	struct queueing	*q;
+	struct offline	*q;
 
-	queueing_running--;
+	offline_running--;
 
-	while(queueing_running < QUEUEING_MAX) {
-		if ((q = TAILQ_FIRST(&queueing_q)) == NULL)
+	while(offline_running < OFFLINE_QUEUEMAX) {
+		if ((q = TAILQ_FIRST(&offline_q)) == NULL)
 			break; /* all done */
-		TAILQ_REMOVE(&queueing_q, q, entry);
-		parent_enqueue_offline(q->path);
+		TAILQ_REMOVE(&offline_q, q, entry);
+		offline_enqueue(q->path);
 		free(q);
 	}
 }
@@ -1211,7 +1239,6 @@ imsg_to_str(int type)
 	CASE(IMSG_BATCH_CLOSE);
 	CASE(IMSG_BATCH_DONE);
 
-	CASE(IMSG_PARENT_ENQUEUE_OFFLINE);
 	CASE(IMSG_PARENT_FORWARD_OPEN);
 	CASE(IMSG_PARENT_FORK_MDA);
 
diff --git a/usr.sbin/smtpd/smtpd.h b/usr.sbin/smtpd/smtpd.h
index 2738b63b6bd..bf16fbbbe8d 100644
--- a/usr.sbin/smtpd/smtpd.h
+++ b/usr.sbin/smtpd/smtpd.h
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpd.h,v 1.247 2011/10/26 20:47:31 gilles Exp $	*/
+/*	$OpenBSD: smtpd.h,v 1.248 2011/11/07 11:14:10 eric Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -174,7 +174,6 @@ enum imsg_type {
 	IMSG_BATCH_CLOSE,
 	IMSG_BATCH_DONE,
 
-	IMSG_PARENT_ENQUEUE_OFFLINE,
 	IMSG_PARENT_FORWARD_OPEN,
 	IMSG_PARENT_FORK_MDA,