Implementation of RFC 2920 PIPELINING extension, client side only for now.

This restructures the client_* API internals significantly.  The code becomes
pipelining in nature.  All SMTP commands are put on the output queue and
dequeued as quickly as possible.  Once dequeued, they're moved to the receive
queue so that replies can be matched with previous commands.

Dequeuing commands from the output queue halts when the count of commands
currently in-pipeline (``cmdi'') is equal to the command send window (``cmdw'').
There are three cmdw values useful in practice:

0               clear pipeline, ie. inhibit all future sends
1               disable pipelining, ie. use old ``one-request-one-reply`` mode
SIZE_T_MAX      enable pipelining, ie. dequeue as many commands as possible

At the beginning of session cmdw is 1.  When it is found that peer supports
PIPELINING, it grows to SIZE_T_MAX.  After dequeing DATA it is again 1.  After
sending QUIT it is 0.

Each command dequeued from the output queue becomes a buf in a msgbuf.  The act
of combining multiple commands into a single send operation did not need to be
implemented: buf_write() already combines bufs using iovec and sends them at
once using sendmsg(2).

Tested by todd@ and oga@

"looks good" to gilles@

1 files changed, 17 insertions, 15 deletions

diff --git a/usr.sbin/smtpd/mta.c b/usr.sbin/smtpd/mta.c
index 7f014b64e82..dc64b81ccab 100644
--- a/usr.sbin/smtpd/mta.c
+++ b/usr.sbin/smtpd/mta.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: mta.c,v 1.82 2009/12/14 16:44:14 jacekm Exp $	*/
+/*	$OpenBSD: mta.c,v 1.83 2009/12/23 17:16:03 jacekm Exp $	*/
 
 /*
  * Copyright (c) 2008 Pierre-Yves Ritschard <pyr@openbsd.org>
@@ -670,6 +670,8 @@ mta_enter_state(struct mta_session *s, int newstate, void *p)
 		if (m->sender.user[0] && m->sender.domain[0])
 			client_sender(pcb, "%s@%s", m->sender.user,
 			    m->sender.domain);
+		else
+			client_sender(pcb, "");
 			
 		/* set envelope recipients */
 		TAILQ_FOREACH(m, &s->recipients, entry)
@@ -677,7 +679,7 @@ mta_enter_state(struct mta_session *s, int newstate, void *p)
 			    m->recipient.domain);
 
 		s->pcb = pcb;
-		event_set(&s->ev, s->fd, EV_WRITE, mta_event, s);
+		event_set(&s->ev, s->fd, EV_READ|EV_WRITE, mta_event, s);
 		event_add(&s->ev, &pcb->timeout);
 		break;
 
@@ -792,16 +794,16 @@ mta_event(int fd, short event, void *p)
 		goto out;
 	}
 
-	switch (client_talk(pcb)) {
-	case CLIENT_WANT_READ:
-		goto read;
+	switch (client_talk(pcb, event & EV_WRITE)) {
 	case CLIENT_WANT_WRITE:
-		goto write;
+		goto rw;
+	case CLIENT_STOP_WRITE:
+		goto ro;
 	case CLIENT_RCPT_FAIL:
-		mta_message_status(pcb->rcptfail->p, pcb->reply);
-		mta_message_log(s, pcb->rcptfail->p);
-		mta_message_done(s, pcb->rcptfail->p);
-		goto write;
+		mta_message_status(pcb->rcptfail, pcb->reply);
+		mta_message_log(s, pcb->rcptfail);
+		mta_message_done(s, pcb->rcptfail);
+		goto rw;
 	case CLIENT_DONE:
 		mta_status(s, "%s", pcb->status);
 		break;
@@ -811,7 +813,7 @@ mta_event(int fd, short event, void *p)
 
 out:
 	client_close(pcb);
-	pcb = NULL;
+	s->pcb = NULL;
 
 	if (TAILQ_EMPTY(&s->recipients))
 		mta_enter_state(s, MTA_DONE, NULL);
@@ -819,13 +821,13 @@ out:
 		mta_enter_state(s, MTA_CONNECT, NULL);
 	return;
 
-read:
-	event_set(&s->ev, fd, EV_READ, mta_event, s);
+rw:
+	event_set(&s->ev, fd, EV_READ|EV_WRITE, mta_event, s);
 	event_add(&s->ev, &pcb->timeout);
 	return;
 
-write:
-	event_set(&s->ev, fd, EV_WRITE, mta_event, s);
+ro:
+	event_set(&s->ev, fd, EV_READ, mta_event, s);
 	event_add(&s->ev, &pcb->timeout);
 }