Switch tmux to use imsg. This is the last major change to make the

client-server protocol more resilient and make the protocol versioning work
properly. In future, the only things requiring a protocol version bump will be
changes in the message structs, and (when both client and server have this
change) mixing different versions should nicely report an error message.

As a side effect this also makes the code tidier, fixes a problem with the way
errors reported during server startup were handled, and supports fd passing
(which will be used in future).

Looked over by eric@, thanks.

Please note that mixing a client with this change with an older server or vice
versa may cause tmux to crash or hang - tmux should be completely exited before
upgrading.

12 files changed, 869 insertions, 137 deletions

diff --git a/Makefile b/Makefile
index bec2a8bb..97d39afa 100644
--- a/Makefile
+++ b/Makefile
@@ -28,7 +28,8 @@ SRCS=	attributes.c buffer-poll.c buffer.c cfg.c client-fn.c \
 	cmd-set-environment.c cmd-show-environment.c \
 	cmd-up-pane.c cmd-display-message.c cmd.c \
 	colour.c environ.c grid-view.c grid.c input-keys.c \
-	input.c key-bindings.c key-string.c layout-set.c layout.c log.c \
+	imsg.c imsg-buffer.c input.c key-bindings.c key-string.c \
+	layout-set.c layout.c log.c \
 	mode-key.c names.c options-cmd.c options.c paste.c procname.c \
 	resize.c screen-redraw.c screen-write.c screen.c server-fn.c \
 	server-msg.c server.c session.c status.c tmux.c tty-keys.c tty-term.c \
diff --git a/client-fn.c b/client-fn.c
index 129c6871..789ac427 100644
--- a/client-fn.c
+++ b/client-fn.c
@@ -66,14 +66,7 @@ void
 client_write_server(
     struct client_ctx *cctx, enum msgtype type, void *buf, size_t len)
 {
-	struct hdr	hdr;
-
-	hdr.type = type;
-	hdr.size = len;
-	buffer_write(cctx->srv_out, &hdr, sizeof hdr);
-
-	if (buf != NULL && len > 0)
-		buffer_write(cctx->srv_out, buf, len);
+	imsg_compose(&cctx->ibuf, type, PROTOCOL_VERSION, -1, -1, buf, len);
 }
 
 void
diff --git a/client.c b/client.c
index 4e756981..a56af84e 100644
--- a/client.c
+++ b/client.c
@@ -92,16 +92,13 @@ server_started:
 		fatal("fcntl failed");
 	if (fcntl(fd, F_SETFL, mode|O_NONBLOCK) == -1)
 		fatal("fcntl failed");
-	cctx->srv_fd = fd;
-	cctx->srv_in = buffer_create(BUFSIZ);
-	cctx->srv_out = buffer_create(BUFSIZ);
+	imsg_init(&cctx->ibuf, fd);
 
 	if (cmdflags & CMD_SENDENVIRON)
 		client_send_environ(cctx);
 	if (isatty(STDIN_FILENO)) {
 		if (ioctl(STDIN_FILENO, TIOCGWINSZ, &ws) == -1)
 			fatal("ioctl(TIOCGWINSZ)");
-		data.version = PROTOCOL_VERSION;
 		data.flags = flags;
 		data.sx = ws.ws_col;
 		data.sy = ws.ws_row;
@@ -153,6 +150,7 @@ int
 client_main(struct client_ctx *cctx)
 {
 	struct pollfd	 pfd;
+	int		 nfds;
 
 	siginit();
 
@@ -173,24 +171,33 @@ client_main(struct client_ctx *cctx)
 			sigcont = 0;
 		}
 
-		pfd.fd = cctx->srv_fd;
+		pfd.fd = cctx->ibuf.fd;
 		pfd.events = POLLIN;
-		if (BUFFER_USED(cctx->srv_out) > 0)
+		if (cctx->ibuf.w.queued > 0)
 			pfd.events |= POLLOUT;
 
-		if (poll(&pfd, 1, INFTIM) == -1) {
+		if ((nfds = poll(&pfd, 1, INFTIM)) == -1) {
 			if (errno == EAGAIN || errno == EINTR)
 				continue;
 			fatal("poll failed");
 		}
+		if (nfds == 0)
+			continue;
 
-		if (buffer_poll(&pfd, cctx->srv_in, cctx->srv_out) != 0) {
-			cctx->exittype = CCTX_DIED;
-			break;
+		if (pfd.revents & (POLLERR|POLLHUP|POLLNVAL))
+			fatalx("socket error");
+
+		if (pfd.revents & POLLIN) {
+			if (client_msg_dispatch(cctx) != 0)
+				break;
 		}
 
-		if (client_msg_dispatch(cctx) != 0)
-			break;
+		if (pfd.revents & POLLOUT) {
+			if (msgbuf_write(&cctx->ibuf.w) < 0) {
+				cctx->exittype = CCTX_DIED;
+				break;
+			}
+		}
 	}
 
  	if (sigterm) {
@@ -235,54 +242,61 @@ client_handle_winch(struct client_ctx *cctx)
 int
 client_msg_dispatch(struct client_ctx *cctx)
 {
-	struct hdr		 hdr;
+	struct imsg		 imsg;
 	struct msg_print_data	 printdata;
+	ssize_t			 n, datalen;
+
+	if ((n = imsg_read(&cctx->ibuf)) == -1 || n == 0) {
+		cctx->exittype = CCTX_DIED;
+		return (-1);
+	}
 
 	for (;;) {
-		if (BUFFER_USED(cctx->srv_in) < sizeof hdr)
-			return (0);
-		memcpy(&hdr, BUFFER_OUT(cctx->srv_in), sizeof hdr);
-		if (BUFFER_USED(cctx->srv_in) < (sizeof hdr) + hdr.size)
+		if ((n = imsg_get(&cctx->ibuf, &imsg)) == -1)
+			fatalx("imsg_get failed");
+		if (n == 0)
 			return (0);
-		buffer_remove(cctx->srv_in, sizeof hdr);
+		datalen = imsg.hdr.len - IMSG_HEADER_SIZE;
 
-		switch (hdr.type) {
+		switch (imsg.hdr.type) {
 		case MSG_DETACH:
-			if (hdr.size != 0)
+			if (datalen != 0)
 				fatalx("bad MSG_DETACH size");
 
 			client_write_server(cctx, MSG_EXITING, NULL, 0);
 			cctx->exittype = CCTX_DETACH;
 			break;
 		case MSG_ERROR:
-			if (hdr.size != sizeof printdata)
-				fatalx("bad MSG_PRINT size");
-			buffer_read(cctx->srv_in, &printdata, sizeof printdata);
+			if (datalen != sizeof printdata)
+				fatalx("bad MSG_ERROR size");
+			memcpy(&printdata, imsg.data, sizeof printdata);
 
 			printdata.msg[(sizeof printdata.msg) - 1] = '\0';
 			cctx->errstr = xstrdup(printdata.msg);
+			imsg_free(&imsg);
 			return (-1);
 		case MSG_EXIT:
-			if (hdr.size != 0)
+			if (datalen != 0)
 				fatalx("bad MSG_EXIT size");
-		
+
 			client_write_server(cctx, MSG_EXITING, NULL, 0);
 			cctx->exittype = CCTX_EXIT;
 			break;
 		case MSG_EXITED:
-			if (hdr.size != 0)
+			if (datalen != 0)
 				fatalx("bad MSG_EXITED size");
 
+			imsg_free(&imsg);
 			return (-1);
 		case MSG_SHUTDOWN:
-			if (hdr.size != 0)
+			if (datalen != 0)
 				fatalx("bad MSG_SHUTDOWN size");
 
 			client_write_server(cctx, MSG_EXITING, NULL, 0);
 			cctx->exittype = CCTX_SHUTDOWN;
 			break;
 		case MSG_SUSPEND:
-			if (hdr.size != 0)
+			if (datalen != 0)
 				fatalx("bad MSG_SUSPEND size");
 
 			client_suspend();
@@ -290,5 +304,7 @@ client_msg_dispatch(struct client_ctx *cctx)
 		default:
 			fatalx("unexpected message");
 		}
+
+		imsg_free(&imsg);
 	}
 }
diff --git a/cmd-server-info.c b/cmd-server-info.c
index a79f2520..396d15b0 100644
--- a/cmd-server-info.c
+++ b/cmd-server-info.c
@@ -90,7 +90,7 @@ cmd_server_info_exec(unused struct cmd *self, struct cmd_ctx *ctx)
 			continue;
 
 		ctx->print(ctx, "%2d: %s (%d, %d): %s [%ux%u %s] "
-		    "[flags=0x%x/0x%x]", i, c->tty.path, c->fd, c->tty.fd,
+		    "[flags=0x%x/0x%x]", i, c->tty.path, c->ibuf.fd, c->tty.fd,
 		    c->session->name, c->tty.sx, c->tty.sy, c->tty.termname,
 		    c->flags, c->tty.flags);
 	}
diff --git a/imsg-buffer.c b/imsg-buffer.c
new file mode 100644
index 00000000..06fe0b1a
--- /dev/null
+++ b/imsg-buffer.c
@@ -0,0 +1,305 @@
+/*	$OpenBSD$	*/
+
+/*
+ * Copyright (c) 2003, 2004 Henning Brauer <henning@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
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+ * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+
+#include <sys/param.h>
+#include <sys/queue.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include "imsg.h"
+
+int	buf_realloc(struct buf *, size_t);
+void	buf_enqueue(struct msgbuf *, struct buf *);
+void	buf_dequeue(struct msgbuf *, struct buf *);
+
+struct buf *
+buf_open(size_t len)
+{
+	struct buf	*buf;
+
+	if ((buf = calloc(1, sizeof(struct buf))) == NULL)
+		return (NULL);
+	if ((buf->buf = malloc(len)) == NULL) {
+		free(buf);
+		return (NULL);
+	}
+	buf->size = buf->max = len;
+	buf->fd = -1;
+
+	return (buf);
+}
+
+struct buf *
+buf_dynamic(size_t len, size_t max)
+{
+	struct buf	*buf;
+
+	if (max < len)
+		return (NULL);
+
+	if ((buf = buf_open(len)) == NULL)
+		return (NULL);
+
+	if (max > 0)
+		buf->max = max;
+
+	return (buf);
+}
+
+int
+buf_realloc(struct buf *buf, size_t len)
+{
+	u_char	*b;
+
+	/* on static buffers max is eq size and so the following fails */
+	if (buf->wpos + len > buf->max) {
+		errno = ENOMEM;
+		return (-1);
+	}
+
+	b = realloc(buf->buf, buf->wpos + len);
+	if (b == NULL)
+		return (-1);
+	buf->buf = b;
+	buf->size = buf->wpos + len;
+
+	return (0);
+}
+
+int
+buf_add(struct buf *buf, const void *data, size_t len)
+{
+	if (buf->wpos + len > buf->size)
+		if (buf_realloc(buf, len) == -1)
+			return (-1);
+
+	memcpy(buf->buf + buf->wpos, data, len);
+	buf->wpos += len;
+	return (0);
+}
+
+void *
+buf_reserve(struct buf *buf, size_t len)
+{
+	void	*b;
+
+	if (buf->wpos + len > buf->size)
+		if (buf_realloc(buf, len) == -1)
+			return (NULL);
+
+	b = buf->buf + buf->wpos;
+	buf->wpos += len;
+	return (b);
+}
+
+void *
+buf_seek(struct buf *buf, size_t pos, size_t len)
+{
+	/* only allowed to seek in already written parts */
+	if (pos + len > buf->wpos)
+		return (NULL);
+
+	return (buf->buf + pos);
+}
+
+size_t
+buf_size(struct buf *buf)
+{
+	return (buf->wpos);
+}
+
+size_t
+buf_left(struct buf *buf)
+{
+	return (buf->max - buf->wpos);
+}
+
+void
+buf_close(struct msgbuf *msgbuf, struct buf *buf)
+{
+	buf_enqueue(msgbuf, buf);
+}
+
+int
+buf_write(struct msgbuf *msgbuf)
+{
+	struct iovec	 iov[IOV_MAX];
+	struct buf	*buf, *next;
+	unsigned int	 i = 0;
+	ssize_t	n;
+
+	bzero(&iov, sizeof(iov));
+	TAILQ_FOREACH(buf, &msgbuf->bufs, entry) {
+		if (i >= IOV_MAX)
+			break;
+		iov[i].iov_base = buf->buf + buf->rpos;
+		iov[i].iov_len = buf->wpos - buf->rpos;
+		i++;
+	}
+
+	if ((n = writev(msgbuf->fd, iov, i)) == -1) {
+		if (errno == EAGAIN || errno == ENOBUFS ||
+		    errno == EINTR)	/* try later */
+			return (0);
+		else
+			return (-1);
+	}
+
+	if (n == 0) {			/* connection closed */
+		errno = 0;
+		return (-2);
+	}
+
+	for (buf = TAILQ_FIRST(&msgbuf->bufs); buf != NULL && n > 0;
+	    buf = next) {
+		next = TAILQ_NEXT(buf, entry);
+		if (buf->rpos + n >= buf->wpos) {
+			n -= buf->wpos - buf->rpos;
+			buf_dequeue(msgbuf, buf);
+		} else {
+			buf->rpos += n;
+			n = 0;
+		}
+	}
+
+	return (0);
+}
+
+void
+buf_free(struct buf *buf)
+{
+	free(buf->buf);
+	free(buf);
+}
+
+void
+msgbuf_init(struct msgbuf *msgbuf)
+{
+	msgbuf->queued = 0;
+	msgbuf->fd = -1;
+	TAILQ_INIT(&msgbuf->bufs);
+}
+
+void
+msgbuf_clear(struct msgbuf *msgbuf)
+{
+	struct buf	*buf;
+
+	while ((buf = TAILQ_FIRST(&msgbuf->bufs)) != NULL)
+		buf_dequeue(msgbuf, buf);
+}
+
+int
+msgbuf_write(struct msgbuf *msgbuf)
+{
+	struct iovec	 iov[IOV_MAX];
+	struct buf	*buf, *next;
+	unsigned int	 i = 0;
+	ssize_t		 n;
+	struct msghdr	 msg;
+	struct cmsghdr	*cmsg;
+	union {
+		struct cmsghdr	hdr;
+		char		buf[CMSG_SPACE(sizeof(int))];
+	} cmsgbuf;
+
+	bzero(&iov, sizeof(iov));
+	bzero(&msg, sizeof(msg));
+	TAILQ_FOREACH(buf, &msgbuf->bufs, entry) {
+		if (i >= IOV_MAX)
+			break;
+		iov[i].iov_base = buf->buf + buf->rpos;
+		iov[i].iov_len = buf->wpos - buf->rpos;
+		i++;
+		if (buf->fd != -1)
+			break;
+	}
+
+	msg.msg_iov = iov;
+	msg.msg_iovlen = i;
+
+	if (buf != NULL && buf->fd != -1) {
+		msg.msg_control = (caddr_t)&cmsgbuf.buf;
+		msg.msg_controllen = sizeof(cmsgbuf.buf);
+		cmsg = CMSG_FIRSTHDR(&msg);
+		cmsg->cmsg_len = CMSG_LEN(sizeof(int));
+		cmsg->cmsg_level = SOL_SOCKET;
+		cmsg->cmsg_type = SCM_RIGHTS;
+		*(int *)CMSG_DATA(cmsg) = buf->fd;
+	}
+
+	if ((n = sendmsg(msgbuf->fd, &msg, 0)) == -1) {
+		if (errno == EAGAIN || errno == ENOBUFS ||
+		    errno == EINTR)	/* try later */
+			return (0);
+		else
+			return (-1);
+	}
+
+	if (n == 0) {			/* connection closed */
+		errno = 0;
+		return (-2);
+	}
+
+	/*
+	 * assumption: fd got sent if sendmsg sent anything
+	 * this works because fds are passed one at a time
+	 */
+	if (buf != NULL && buf->fd != -1) {
+		close(buf->fd);
+		buf->fd = -1;
+	}
+
+	for (buf = TAILQ_FIRST(&msgbuf->bufs); buf != NULL && n > 0;
+	    buf = next) {
+		next = TAILQ_NEXT(buf, entry);
+		if (buf->rpos + n >= buf->wpos) {
+			n -= buf->wpos - buf->rpos;
+			buf_dequeue(msgbuf, buf);
+		} else {
+			buf->rpos += n;
+			n = 0;
+		}
+	}
+
+	return (0);
+}
+
+void
+buf_enqueue(struct msgbuf *msgbuf, struct buf *buf)
+{
+	TAILQ_INSERT_TAIL(&msgbuf->bufs, buf, entry);
+	msgbuf->queued++;
+}
+
+void
+buf_dequeue(struct msgbuf *msgbuf, struct buf *buf)
+{
+	TAILQ_REMOVE(&msgbuf->bufs, buf, entry);
+
+	if (buf->fd != -1)
+		close(buf->fd);
+
+	msgbuf->queued--;
+	buf_free(buf);
+}
diff --git a/imsg.c b/imsg.c
new file mode 100644
index 00000000..263c7c8c
--- /dev/null
+++ b/imsg.c
@@ -0,0 +1,271 @@
+/*	$OpenBSD$	*/
+
+/*
+ * Copyright (c) 2003, 2004 Henning Brauer <henning@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
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+ * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+
+#include <sys/param.h>
+#include <sys/queue.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include "imsg.h"
+
+int	 imsg_get_fd(struct imsgbuf *);
+
+void
+imsg_init(struct imsgbuf *ibuf, int fd)
+{
+	msgbuf_init(&ibuf->w);
+	bzero(&ibuf->r, sizeof(ibuf->r));
+	ibuf->fd = fd;
+	ibuf->w.fd = fd;
+	ibuf->pid = getpid();
+	TAILQ_INIT(&ibuf->fds);
+}
+
+ssize_t
+imsg_read(struct imsgbuf *ibuf)
+{
+	struct msghdr		 msg;
+	struct cmsghdr		*cmsg;
+	union {
+		struct cmsghdr hdr;
+		char	buf[CMSG_SPACE(sizeof(int) * 16)];
+	} cmsgbuf;
+	struct iovec		 iov;
+	ssize_t			 n;
+	int			 fd;
+	struct imsg_fd		*ifd;
+
+	bzero(&msg, sizeof(msg));
+
+	iov.iov_base = ibuf->r.buf + ibuf->r.wpos;
+	iov.iov_len = sizeof(ibuf->r.buf) - ibuf->r.wpos;
+	msg.msg_iov = &iov;
+	msg.msg_iovlen = 1;
+	msg.msg_control = &cmsgbuf.buf;
+	msg.msg_controllen = sizeof(cmsgbuf.buf);
+
+	if ((n = recvmsg(ibuf->fd, &msg, 0)) == -1) {
+		if (errno != EINTR && errno != EAGAIN) {
+			return (-1);
+		}
+		return (-2);
+	}
+
+	ibuf->r.wpos += n;
+
+	for (cmsg = CMSG_FIRSTHDR(&msg); cmsg != NULL;
+	    cmsg = CMSG_NXTHDR(&msg, cmsg)) {
+		if (cmsg->cmsg_level == SOL_SOCKET &&
+		    cmsg->cmsg_type == SCM_RIGHTS) {
+			fd = (*(int *)CMSG_DATA(cmsg));
+			if ((ifd = calloc(1, sizeof(struct imsg_fd))) == NULL) {
+				/* XXX: this return can leak */
+				return (-1);
+			}
+			ifd->fd = fd;
+			TAILQ_INSERT_TAIL(&ibuf->fds, ifd, entry);
+		}
+		/* we do not handle other ctl data level */
+	}
+
+	return (n);
+}
+
+ssize_t
+imsg_get(struct imsgbuf *ibuf, struct imsg *imsg)
+{
+	size_t			 av, left, datalen;
+
+	av = ibuf->r.wpos;
+
+	if (IMSG_HEADER_SIZE > av)
+		return (0);
+
+	memcpy(&imsg->hdr, ibuf->r.buf, sizeof(imsg->hdr));
+	if (imsg->hdr.len < IMSG_HEADER_SIZE ||
+	    imsg->hdr.len > MAX_IMSGSIZE) {
+		errno = ERANGE;
+		return (-1);
+	}
+	if (imsg->hdr.len > av)
+		return (0);
+	datalen = imsg->hdr.len - IMSG_HEADER_SIZE;
+	ibuf->r.rptr = ibuf->r.buf + IMSG_HEADER_SIZE;
+	if ((imsg->data = malloc(datalen)) == NULL)
+		return (-1);
+
+	if (imsg->hdr.flags & IMSGF_HASFD)
+		imsg->fd = imsg_get_fd(ibuf);
+	else
+		imsg->fd = -1;
+
+	memcpy(imsg->data, ibuf->r.rptr, datalen);
+
+	if (imsg->hdr.len < av) {
+		left = av - imsg->hdr.len;
+		memmove(&ibuf->r.buf, ibuf->r.buf + imsg->hdr.len, left);
+		ibuf->r.wpos = left;
+	} else
+		ibuf->r.wpos = 0;
+
+	return (datalen + IMSG_HEADER_SIZE);
+}
+
+int
+imsg_compose(struct imsgbuf *ibuf, u_int32_t type, u_int32_t peerid,
+    pid_t pid, int fd, void *data, u_int16_t datalen)
+{
+	struct buf	*wbuf;
+
+	if ((wbuf = imsg_create(ibuf, type, peerid, pid, datalen)) == NULL)
+		return (-1);
+
+	if (imsg_add(wbuf, data, datalen) == -1)
+		return (-1);
+
+	wbuf->fd = fd;
+
+	imsg_close(ibuf, wbuf);
+
+	return (1);
+}
+
+int
+imsg_composev(struct imsgbuf *ibuf, u_int32_t type, u_int32_t peerid,
+    pid_t pid, int fd, const struct iovec *iov, int iovcnt)
+{
+	struct buf	*wbuf;
+	int		 i, datalen = 0;
+
+	for (i = 0; i < iovcnt; i++)
+		datalen += iov[i].iov_len;
+
+	if ((wbuf = imsg_create(ibuf, type, peerid, pid, datalen)) == NULL)
+		return (-1);
+
+	for (i = 0; i < iovcnt; i++)
+		if (imsg_add(wbuf, iov[i].iov_base, iov[i].iov_len) == -1)
+			return (-1);
+
+	wbuf->fd = fd;
+
+	imsg_close(ibuf, wbuf);
+
+	return (1);
+}
+
+/* ARGSUSED */
+struct buf *
+imsg_create(struct imsgbuf *ibuf, u_int32_t type, u_int32_t peerid,
+    pid_t pid, u_int16_t datalen)
+{
+	struct buf	*wbuf;
+	struct imsg_hdr	 hdr;
+
+	datalen += IMSG_HEADER_SIZE;
+	if (datalen > MAX_IMSGSIZE) {
+		errno = ERANGE;
+		return (NULL);
+	}
+
+	hdr.type = type;
+	hdr.flags = 0;
+	hdr.peerid = peerid;
+	if ((hdr.pid = pid) == 0)
+		hdr.pid = ibuf->pid;
+	if ((wbuf = buf_dynamic(datalen, MAX_IMSGSIZE)) == NULL) {
+		return (NULL);
+	}
+	if (imsg_add(wbuf, &hdr, sizeof(hdr)) == -1)
+		return (NULL);
+
+	return (wbuf);
+}
+
+int
+imsg_add(struct buf *msg, void *data, u_int16_t datalen)
+{
+	if (datalen)
+		if (buf_add(msg, data, datalen) == -1) {
+			buf_free(msg);
+			return (-1);
+		}
+	return (datalen);
+}
+
+void
+imsg_close(struct imsgbuf *ibuf, struct buf *msg)
+{
+	struct imsg_hdr	*hdr;
+
+	hdr = (struct imsg_hdr *)msg->buf;
+
+	hdr->flags &= ~IMSGF_HASFD;
+	if (msg->fd != -1)
+		hdr->flags |= IMSGF_HASFD;
+
+	hdr->len = (u_int16_t)msg->wpos;
+
+	buf_close(&ibuf->w, msg);
+}
+
+void
+imsg_free(struct imsg *imsg)
+{
+	free(imsg->data);
+}
+
+int
+imsg_get_fd(struct imsgbuf *ibuf)
+{
+	int		 fd;
+	struct imsg_fd	*ifd;
+
+	if ((ifd = TAILQ_FIRST(&ibuf->fds)) == NULL)
+		return (-1);
+
+	fd = ifd->fd;
+	TAILQ_REMOVE(&ibuf->fds, ifd, entry);
+	free(ifd);
+
+	return (fd);
+}
+
+int
+imsg_flush(struct imsgbuf *ibuf)
+{
+	while (ibuf->w.queued)
+		if (msgbuf_write(&ibuf->w) < 0)
+			return (-1);
+	return (0);
+}
+
+void
+imsg_clear(struct imsgbuf *ibuf)
+{
+	int	fd;
+
+	msgbuf_clear(&ibuf->w);
+	while ((fd = imsg_get_fd(ibuf)) != -1)
+		close(fd);
+}
diff --git a/imsg.h b/imsg.h
new file mode 100644
index 00000000..5de73f0c
--- /dev/null
+++ b/imsg.h
@@ -0,0 +1,108 @@
+/*	$OpenBSD$	*/
+
+/*
+ * Copyright (c) 2006, 2007 Pierre-Yves Ritschard <pyr@openbsd.org>
+ * Copyright (c) 2006, 2007, 2008 Reyk Floeter <reyk@openbsd.org>
+ * Copyright (c) 2003, 2004 Henning Brauer <henning@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
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+ * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+
+#include <sys/tree.h>
+
+#define READ_BUF_SIZE		65535
+#define IMSG_HEADER_SIZE	sizeof(struct imsg_hdr)
+#define MAX_IMSGSIZE		16384
+
+struct buf {
+	TAILQ_ENTRY(buf)	 entry;
+	u_char			*buf;
+	size_t			 size;
+	size_t			 max;
+	size_t			 wpos;
+	size_t			 rpos;
+	int			 fd;
+};
+
+struct msgbuf {
+	TAILQ_HEAD(, buf)	 bufs;
+	u_int32_t		 queued;
+	int			 fd;
+};
+
+struct buf_read {
+	u_char			 buf[READ_BUF_SIZE];
+	u_char			*rptr;
+	size_t			 wpos;
+};
+
+struct imsg_fd {
+	TAILQ_ENTRY(imsg_fd)	entry;
+	int			fd;
+};
+
+struct imsgbuf {
+	TAILQ_HEAD(, imsg_fd)	 fds;
+	struct buf_read		 r;
+	struct msgbuf		 w;
+	int			 fd;
+	pid_t			 pid;
+};
+
+#define IMSGF_HASFD	1
+
+struct imsg_hdr {
+	u_int32_t	 type;
+	u_int16_t	 len;
+	u_int16_t	 flags;
+	u_int32_t	 peerid;
+	u_int32_t	 pid;
+};
+
+struct imsg {
+	struct imsg_hdr	 hdr;
+	int		 fd;
+	void		*data;
+};
+
+
+/* buffer.c */
+struct buf	*buf_open(size_t);
+struct buf	*buf_dynamic(size_t, size_t);
+int		 buf_add(struct buf *, const void *, size_t);
+void		*buf_reserve(struct buf *, size_t);
+void		*buf_seek(struct buf *, size_t, size_t);
+size_t		 buf_size(struct buf *);
+size_t		 buf_left(struct buf *);
+void		 buf_close(struct msgbuf *, struct buf *);
+int		 buf_write(struct msgbuf *);
+void		 buf_free(struct buf *);
+void		 msgbuf_init(struct msgbuf *);
+void		 msgbuf_clear(struct msgbuf *);
+int		 msgbuf_write(struct msgbuf *);
+
+/* imsg.c */
+void	 imsg_init(struct imsgbuf *, int);
+ssize_t	 imsg_read(struct imsgbuf *);
+ssize_t	 imsg_get(struct imsgbuf *, struct imsg *);
+int	 imsg_compose(struct imsgbuf *, u_int32_t, u_int32_t, pid_t,
+	    int, void *, u_int16_t);
+int	 imsg_composev(struct imsgbuf *, u_int32_t, u_int32_t,  pid_t,
+	    int, const struct iovec *, int);
+struct buf *imsg_create(struct imsgbuf *, u_int32_t, u_int32_t, pid_t,
+	    u_int16_t);
+int	 imsg_add(struct buf *, void *, u_int16_t);
+void	 imsg_close(struct imsgbuf *, struct buf *);
+void	 imsg_free(struct imsg *);
+int	 imsg_flush(struct imsgbuf *);
+void	 imsg_clear(struct imsgbuf *);
diff --git a/server-fn.c b/server-fn.c
index 6f6ed2c8..8d5d06bd 100644
--- a/server-fn.c
+++ b/server-fn.c
@@ -55,18 +55,12 @@ void
 server_write_client(
     struct client *c, enum msgtype type, const void *buf, size_t len)
 {
-	struct hdr	 hdr;
+	struct imsgbuf	*ibuf = &c->ibuf;
 
 	if (c->flags & CLIENT_BAD)
 		return;
-	log_debug("writing %d to client %d", type, c->fd);
-
-	hdr.type = type;
-	hdr.size = len;
-
-	buffer_write(c->out, &hdr, sizeof hdr);
-	if (buf != NULL && len > 0)
-		buffer_write(c->out, buf, len);
+	log_debug("writing %d to client %d", type, c->ibuf.fd);
+	imsg_compose(ibuf, type, PROTOCOL_VERSION, -1, -1, (void *) buf, len);
 }
 
 void
diff --git a/server-msg.c b/server-msg.c
index b23ec4c4..6cc69a0a 100644
--- a/server-msg.c
+++ b/server-msg.c
@@ -37,45 +37,56 @@ void printflike2 server_msg_command_info(struct cmd_ctx *, const char *, ...);
 int
 server_msg_dispatch(struct client *c)
 {
-	struct hdr		 hdr;
+	struct imsg		 imsg;
 	struct msg_command_data	 commanddata;
 	struct msg_identify_data identifydata;
 	struct msg_resize_data	 resizedata;
 	struct msg_unlock_data	 unlockdata;
 	struct msg_environ_data	 environdata;
+	ssize_t			 n, datalen;
+
+        if ((n = imsg_read(&c->ibuf)) == -1 || n == 0)
+                return (-1);
 
 	for (;;) {
-		if (BUFFER_USED(c->in) < sizeof hdr)
-			return (0);
-		memcpy(&hdr, BUFFER_OUT(c->in), sizeof hdr);
-		if (BUFFER_USED(c->in) < (sizeof hdr) + hdr.size)
+		if ((n = imsg_get(&c->ibuf, &imsg)) == -1)
+			return (-1);
+		if (n == 0)
 			return (0);
-		buffer_remove(c->in, sizeof hdr);
+		datalen = imsg.hdr.len - IMSG_HEADER_SIZE;
+
+		if (imsg.hdr.peerid != PROTOCOL_VERSION) {
+			server_write_client(c, MSG_VERSION, NULL, 0);
+			c->flags |= CLIENT_BAD;
+			imsg_free(&imsg);
+			continue;
+		}
 
-		switch (hdr.type) {
+		log_debug("got %d from client %d", imsg.hdr.type, c->ibuf.fd);
+		switch (imsg.hdr.type) {
 		case MSG_COMMAND:
-			if (hdr.size != sizeof commanddata)
+			if (datalen != sizeof commanddata)
 				fatalx("bad MSG_COMMAND size");
-			buffer_read(c->in, &commanddata, sizeof commanddata);
+			memcpy(&commanddata, imsg.data, sizeof commanddata);
 
 			server_msg_command(c, &commanddata);
 			break;
 		case MSG_IDENTIFY:
-			if (hdr.size != sizeof identifydata)
+			if (datalen != sizeof identifydata)
 				fatalx("bad MSG_IDENTIFY size");
-			buffer_read(c->in, &identifydata, sizeof identifydata);
+			memcpy(&identifydata, imsg.data, sizeof identifydata);
 
 			server_msg_identify(c, &identifydata);
 			break;
 		case MSG_RESIZE:
-			if (hdr.size != sizeof resizedata)
+			if (datalen != sizeof resizedata)
 				fatalx("bad MSG_RESIZE size");
-			buffer_read(c->in, &resizedata, sizeof resizedata);
+			memcpy(&resizedata, imsg.data, sizeof resizedata);
 
 			server_msg_resize(c, &resizedata);
 			break;
 		case MSG_EXITING:
-			if (hdr.size != 0)
+			if (datalen != 0)
 				fatalx("bad MSG_EXITING size");
 
 			c->session = NULL;
@@ -83,9 +94,9 @@ server_msg_dispatch(struct client *c)
 			server_write_client(c, MSG_EXITED, NULL, 0);
 			break;
 		case MSG_UNLOCK:
-			if (hdr.size != sizeof unlockdata)
+			if (datalen != sizeof unlockdata)
 				fatalx("bad MSG_UNLOCK size");
-			buffer_read(c->in, &unlockdata, sizeof unlockdata);
+			memcpy(&unlockdata, imsg.data, sizeof unlockdata);
 
 			unlockdata.pass[(sizeof unlockdata.pass) - 1] = '\0';
 			if (server_unlock(unlockdata.pass) != 0)
@@ -94,7 +105,7 @@ server_msg_dispatch(struct client *c)
 			server_write_client(c, MSG_EXIT, NULL, 0);
 			break;
 		case MSG_WAKEUP:
-			if (hdr.size != 0)
+			if (datalen != 0)
 				fatalx("bad MSG_WAKEUP size");
 
 			c->flags &= ~CLIENT_SUSPENDED;
@@ -102,9 +113,9 @@ server_msg_dispatch(struct client *c)
 			server_redraw_client(c);
 			break;
 		case MSG_ENVIRON:
-			if (hdr.size != sizeof environdata)
+			if (datalen != sizeof environdata)
 				fatalx("bad MSG_ENVIRON size");
-			buffer_read(c->in, &environdata, sizeof environdata);
+			memcpy(&environdata, imsg.data, sizeof environdata);
 
 			environdata.var[(sizeof environdata.var) - 1] = '\0';
 			if (strchr(environdata.var, '=') != NULL)
@@ -113,6 +124,8 @@ server_msg_dispatch(struct client *c)
 		default:
 			fatalx("unexpected message");
 		}
+
+		imsg_free(&imsg);
 	}
 }
 
@@ -224,11 +237,6 @@ error:
 void
 server_msg_identify(struct client *c, struct msg_identify_data *data)
 {
-	if (data->version != PROTOCOL_VERSION) {
-		server_write_error(c, "protocol version mismatch");
-		return;
-	}
-
 	c->tty.sx = data->sx;
 	c->tty.sy = data->sy;
 
diff --git a/server.c b/server.c
index ab407a04..acaf5207 100644
--- a/server.c
+++ b/server.c
@@ -85,9 +85,7 @@ server_create_client(int fd)
 		fatal("fcntl failed");
 
 	c = xcalloc(1, sizeof *c);
-	c->fd = fd;
-	c->in = buffer_create(BUFSIZ);
-	c->out = buffer_create(BUFSIZ);
+	imsg_init(&c->ibuf, fd);
 
 	ARRAY_INIT(&c->prompt_hdata);
 
@@ -672,10 +670,10 @@ server_fill_clients(struct pollfd **pfd)
 		if (c == NULL)
 			(*pfd)->fd = -1;
 		else {
-			(*pfd)->fd = c->fd;
+			(*pfd)->fd = c->ibuf.fd;
 			if (!(c->flags & CLIENT_BAD))
-				(*pfd)->events = POLLIN;
-			if (BUFFER_USED(c->out) > 0)
+				(*pfd)->events |= POLLIN;
+			if (c->ibuf.w.queued > 0)
 				(*pfd)->events |= POLLOUT;
 		}
 		(*pfd)++;
@@ -718,17 +716,32 @@ server_handle_clients(struct pollfd **pfd)
 		c = ARRAY_ITEM(&clients, i);
 
 		if (c != NULL) {
-			if (buffer_poll(*pfd, c->in, c->out) != 0) {
+			if ((*pfd)->revents & (POLLERR|POLLNVAL|POLLHUP)) {
 				server_lost_client(c);
 				(*pfd) += 2;
 				continue;
-			} else if (c->flags & CLIENT_BAD) {
-				if (BUFFER_USED(c->out) == 0)
+			}
+
+			if ((*pfd)->revents & POLLOUT) {
+				if (msgbuf_write(&c->ibuf.w) < 0) {
+					server_lost_client(c);
+					(*pfd) += 2;
+					continue;
+				}
+			}
+
+			if (c->flags & CLIENT_BAD) {
+				if (c->ibuf.w.queued == 0)
 					server_lost_client(c);
 				(*pfd) += 2;
-				continue;			
-			} else
-				server_msg_dispatch(c);
+				continue;
+			} else if ((*pfd)->revents & POLLIN) {
+				if (server_msg_dispatch(c) != 0) {
+					server_lost_client(c);
+					(*pfd) += 2;
+					continue;
+				}
+			}
 		}
 		(*pfd)++;
 
@@ -910,9 +923,8 @@ server_lost_client(struct client *c)
 	if (c->cwd != NULL)
 		xfree(c->cwd);
 
-	close(c->fd);
-	buffer_destroy(c->in);
-	buffer_destroy(c->out);
+	close(c->ibuf.fd);
+	imsg_clear(&c->ibuf);
 	xfree(c);
 
 	recalculate_sizes();
diff --git a/tmux.c b/tmux.c
index af099e3f..34416e30 100644
--- a/tmux.c
+++ b/