Support code for control mode flow control: allow clients to have

separate offsets (used and acknowleged) into the pane buffers; turn off
reading from panes when no clients can accept the data; and add a -A
flag to refresh-client to let clients turn receiving a pane on and off.

11 files changed, 357 insertions, 95 deletions

diff --git a/cmd-pipe-pane.c b/cmd-pipe-pane.c
index 185bdc12..d9d9f436 100644
--- a/cmd-pipe-pane.c
+++ b/cmd-pipe-pane.c
@@ -56,16 +56,17 @@ const struct cmd_entry cmd_pipe_pane_entry = {
 static enum cmd_retval
 cmd_pipe_pane_exec(struct cmd *self, struct cmdq_item *item)
 {
-	struct args		*args = cmd_get_args(self);
-	struct cmd_find_state	*target = cmdq_get_target(item);
-	struct client		*tc = cmdq_get_target_client(item);
-	struct window_pane	*wp = target->wp;
-	struct session		*s = target->s;
-	struct winlink		*wl = target->wl;
-	char			*cmd;
-	int			 old_fd, pipe_fd[2], null_fd, in, out;
-	struct format_tree	*ft;
-	sigset_t		 set, oldset;
+	struct args			*args = cmd_get_args(self);
+	struct cmd_find_state		*target = cmdq_get_target(item);
+	struct client			*tc = cmdq_get_target_client(item);
+	struct window_pane		*wp = target->wp;
+	struct session			*s = target->s;
+	struct winlink			*wl = target->wl;
+	struct window_pane_offset	*wpo = &wp->pipe_offset;
+	char				*cmd;
+	int				 old_fd, pipe_fd[2], null_fd, in, out;
+	struct format_tree		*ft;
+	sigset_t			 set, oldset;
 
 	/* Destroy the old pipe. */
 	old_fd = wp->pipe_fd;
@@ -159,10 +160,7 @@ cmd_pipe_pane_exec(struct cmd *self, struct cmdq_item *item)
 		close(pipe_fd[1]);
 
 		wp->pipe_fd = pipe_fd[0];
-		if (wp->fd != -1)
-			wp->pipe_off = EVBUFFER_LENGTH(wp->event->input);
-		else
-			wp->pipe_off = 0;
+		memcpy(wpo, &wp->offset, sizeof *wpo);
 
 		setblocking(wp->pipe_fd, 0);
 		wp->pipe_event = bufferevent_new(wp->pipe_fd,
diff --git a/cmd-refresh-client.c b/cmd-refresh-client.c
index d31bd2cc..65537917 100644
--- a/cmd-refresh-client.c
+++ b/cmd-refresh-client.c
@@ -34,23 +34,55 @@ const struct cmd_entry cmd_refresh_client_entry = {
 	.name = "refresh-client",
 	.alias = "refresh",
 
-	.args = { "cC:Df:F:lLRSt:U", 0, 1 },
-	.usage = "[-cDlLRSU] [-C XxY] [-f flags] " CMD_TARGET_CLIENT_USAGE
-		" [adjustment]",
+	.args = { "A:cC:Df:F:lLRSt:U", 0, 1 },
+	.usage = "[-cDlLRSU] [-A pane:state] [-C XxY] [-f flags] "
+		 CMD_TARGET_CLIENT_USAGE " [adjustment]",
 
 	.flags = CMD_AFTERHOOK|CMD_CLIENT_TFLAG,
 	.exec = cmd_refresh_client_exec
 };
 
+static void
+cmd_refresh_client_update_offset(struct client *tc, const char *value)
+{
+	struct window_pane	*wp;
+	struct client_offset	*co;
+	char			*copy, *colon;
+	u_int			 pane;
+
+	if (*value != '%')
+		return;
+	copy = xstrdup(value);
+	if ((colon = strchr(copy, ':')) == NULL)
+		goto out;
+	*colon++ = '\0';
+
+	if (sscanf(copy, "%%%u", &pane) != 1)
+		goto out;
+	wp = window_pane_find_by_id(pane);
+	if (wp == NULL)
+		goto out;
+
+	co = server_client_add_pane_offset(tc, wp);
+	if (strcmp(colon, "on") == 0)
+		co->flags &= ~CLIENT_OFFSET_OFF;
+	else if (strcmp(colon, "off") == 0)
+		co->flags |= CLIENT_OFFSET_OFF;
+
+out:
+	free(copy);
+}
+
 static enum cmd_retval
 cmd_refresh_client_exec(struct cmd *self, struct cmdq_item *item)
 {
-	struct args	*args = cmd_get_args(self);
-	struct client	*tc = cmdq_get_target_client(item);
-	struct tty	*tty = &tc->tty;
-	struct window	*w;
-	const char	*size, *errstr;
-	u_int		 x, y, adjust;
+	struct args		*args = cmd_get_args(self);
+	struct client		*tc = cmdq_get_target_client(item);
+	struct tty		*tty = &tc->tty;
+	struct window		*w;
+	const char		*size, *errstr, *value;
+	u_int			 x, y, adjust;
+	struct args_value	*av;
 
 	if (args_has(args, 'c') ||
 	    args_has(args, 'L') ||
@@ -112,11 +144,19 @@ cmd_refresh_client_exec(struct cmd *self, struct cmdq_item *item)
 	if (args_has(args, 'f'))
 		server_client_set_flags(tc, args_get(args, 'f'));
 
-	if (args_has(args, 'C')) {
-		if (~tc->flags & CLIENT_CONTROL) {
-			cmdq_error(item, "not a control client");
-			return (CMD_RETURN_ERROR);
+	if (args_has(args, 'A')) {
+		if (~tc->flags & CLIENT_CONTROL)
+			goto not_control_client;
+		value = args_first_value(args, 'A', &av);
+		while (value != NULL) {
+			cmd_refresh_client_update_offset(tc, value);
+			value = args_next_value(&av);
 		}
+		return (CMD_RETURN_NORMAL);
+	}
+	if (args_has(args, 'C')) {
+		if (~tc->flags & CLIENT_CONTROL)
+			goto not_control_client;
 		size = args_get(args, 'C');
 		if (sscanf(size, "%u,%u", &x, &y) != 2 &&
 		    sscanf(size, "%ux%u", &x, &y) != 2) {
@@ -142,4 +182,8 @@ cmd_refresh_client_exec(struct cmd *self, struct cmdq_item *item)
 		server_redraw_client(tc);
 	}
 	return (CMD_RETURN_NORMAL);
+
+not_control_client:
+	cmdq_error(item, "not a control client");
+	return (CMD_RETURN_ERROR);
 }
diff --git a/control-notify.c b/control-notify.c
index a513c147..a1735d57 100644
--- a/control-notify.c
+++ b/control-notify.c
@@ -27,40 +27,6 @@
 	((c) != NULL && ((c)->flags & CLIENT_CONTROL))
 
 void
-control_notify_input(struct client *c, struct window_pane *wp,
-    const u_char *buf, size_t len)
-{
-	struct evbuffer *message;
-	u_int		 i;
-
-	if (c->session == NULL)
-	    return;
-
-	if (c->flags & CLIENT_CONTROL_NOOUTPUT)
-		return;
-
-	/*
-	 * Only write input if the window pane is linked to a window belonging
-	 * to the client's session.
-	 */
-	if (winlink_find_by_window(&c->session->windows, wp->window) != NULL) {
-		message = evbuffer_new();
-		if (message == NULL)
-			fatalx("out of memory");
-		evbuffer_add_printf(message, "%%output %%%u ", wp->id);
-		for (i = 0; i < len; i++) {
-			if (buf[i] < ' ' || buf[i] == '\\')
-			    evbuffer_add_printf(message, "\\%03o", buf[i]);
-			else
-			    evbuffer_add_printf(message, "%c", buf[i]);
-		}
-		evbuffer_add(message, "", 1);
-		control_write(c, "%s", EVBUFFER_DATA(message));
-		evbuffer_free(message);
-	}
-}
-
-void
 control_notify_pane_mode_changed(int pane)
 {
 	struct client	*c;
diff --git a/control.c b/control.c
index bdc89de4..b1bcd2c1 100644
--- a/control.c
+++ b/control.c
@@ -38,6 +38,53 @@ control_write(struct client *c, const char *fmt, ...)
 	va_end(ap);
 }
 
+/* Write output from a pane. */
+void
+control_write_output(struct client *c, struct window_pane *wp)
+{
+	struct client_offset	*co;
+	struct evbuffer		*message;
+	u_char			*new_data;
+	size_t			 new_size, i;
+
+	if (c->flags & CLIENT_CONTROL_NOOUTPUT)
+		return;
+
+	/*
+	 * Only write input if the window pane is linked to a window belonging
+	 * to the client's session.
+	 */
+	if (winlink_find_by_window(&c->session->windows, wp->window) == NULL)
+		return;
+
+	co = server_client_add_pane_offset(c, wp);
+	if (co->flags & CLIENT_OFFSET_OFF) {
+		window_pane_update_used_data(wp, &co->offset, SIZE_MAX, 1);
+		return;
+	}
+	new_data = window_pane_get_new_data(wp, &co->offset, &new_size);
+	if (new_size == 0)
+		return;
+
+	message = evbuffer_new();
+	if (message == NULL)
+		fatalx("out of memory");
+	evbuffer_add_printf(message, "%%output %%%u ", wp->id);
+
+	for (i = 0; i < new_size; i++) {
+		if (new_data[i] < ' ' || new_data[i] == '\\')
+			evbuffer_add_printf(message, "\\%03o", new_data[i]);
+		else
+			evbuffer_add_printf(message, "%c", new_data[i]);
+	}
+	evbuffer_add(message, "", 1);
+
+	control_write(c, "%s", EVBUFFER_DATA(message));
+	evbuffer_free(message);
+
+	window_pane_update_used_data(wp, &co->offset, new_size, 1);
+}
+
 /* Control error callback. */
 static enum cmd_retval
 control_error(struct cmdq_item *item, void *data)
diff --git a/input.c b/input.c
index 44b1b948..7bb69663 100644
--- a/input.c
+++ b/input.c
@@ -942,10 +942,12 @@ input_parse(struct input_ctx *ictx, u_char *buf, size_t len)
 void
 input_parse_pane(struct window_pane *wp)
 {
-	struct evbuffer	*evb = wp->event->input;
+	void	*new_data;
+	size_t	 new_size;
 
-	input_parse_buffer(wp, EVBUFFER_DATA(evb), EVBUFFER_LENGTH(evb));
-	evbuffer_drain(evb, EVBUFFER_LENGTH(evb));
+	new_data = window_pane_get_new_data(wp, &wp->offset, &new_size);
+	input_parse_buffer(wp, new_data, new_size);
+	window_pane_update_used_data(wp, &wp->offset, new_size, 1);
 }
 
 /* Parse given input. */
@@ -960,7 +962,6 @@ input_parse_buffer(struct window_pane *wp, u_char *buf, size_t len)
 
 	window_update_activity(wp->window);
 	wp->flags |= PANE_CHANGED;
-	notify_input(wp, buf, len);
 
 	/* NULL wp if there is a mode set as don't want to update the tty. */
 	if (TAILQ_EMPTY(&wp->modes))
diff --git a/notify.c b/notify.c
index 54254354..a50eb264 100644
--- a/notify.c
+++ b/notify.c
@@ -210,17 +210,6 @@ notify_hook(struct cmdq_item *item, const char *name)
 }
 
 void
-notify_input(struct window_pane *wp, const u_char *buf, size_t len)
-{
-	struct client	*c;
-
-	TAILQ_FOREACH(c, &clients, entry) {
-		if (c->flags & CLIENT_CONTROL)
-			control_notify_input(c, wp, buf, len);
-	}
-}
-
-void
 notify_client(const char *name, struct client *c)
 {
 	struct cmd_find_state	fs;
diff --git a/server-client.c b/server-client.c
index 3cd2083d..02abd73a 100644
--- a/server-client.c
+++ b/server-client.c
@@ -35,6 +35,7 @@
 static void	server_client_free(int, short, void *);
 static void	server_client_check_pane_focus(struct window_pane *);
 static void	server_client_check_pane_resize(struct window_pane *);
+static void	server_client_check_pane_buffer(struct window_pane *);
 static void	server_client_check_window_resize(struct window *);
 static key_code	server_client_check_mouse(struct client *, struct key_event *);
 static void	server_client_repeat_timer(int, short, void *);
@@ -70,6 +71,43 @@ server_client_window_cmp(struct client_window *cw1,
 }
 RB_GENERATE(client_windows, client_window, entry, server_client_window_cmp);
 
+/* Compare client offsets. */
+static int
+server_client_offset_cmp(struct client_offset *co1, struct client_offset *co2)
+{
+	if (co1->pane < co2->pane)
+		return (-1);
+	if (co1->pane > co2->pane)
+		return (1);
+	return (0);
+}
+RB_GENERATE(client_offsets, client_offset, entry, server_client_offset_cmp);
+
+/* Get pane offsets for this client. */
+struct client_offset *
+server_client_get_pane_offset(struct client *c, struct window_pane *wp)
+{
+	struct client_offset	co = { .pane = wp->id };
+
+	return (RB_FIND(client_offsets, &c->offsets, &co));
+}
+
+/* Add pane offsets for this client. */
+struct client_offset *
+server_client_add_pane_offset(struct client *c, struct window_pane *wp)
+{
+	struct client_offset	*co;
+
+	co = server_client_get_pane_offset(c, wp);
+	if (co != NULL)
+		return (co);
+	co = xcalloc(1, sizeof *co);
+	co->pane = wp->id;
+	RB_INSERT(client_offsets, &c->offsets, co);
+	memcpy(&co->offset, &wp->offset, sizeof co->offset);
+	return (co);
+}
+
 /* Number of attached clients. */
 u_int
 server_client_how_many(void)
@@ -226,15 +264,14 @@ server_client_create(int fd)
 
 	c->queue = cmdq_new();
 	RB_INIT(&c->windows);
+	RB_INIT(&c->offsets);
+	RB_INIT(&c->files);
 
 	c->tty.fd = -1;
 	c->tty.sx = 80;
 	c->tty.sy = 24;
 
 	status_init(c);
-
-	RB_INIT(&c->files);
-
 	c->flags |= CLIENT_FOCUSED;
 
 	c->keytable = key_bindings_get_table("root", 1);
@@ -288,6 +325,7 @@ server_client_lost(struct client *c)
 {
 	struct client_file	*cf, *cf1;
 	struct client_window	*cw, *cw1;
+	struct client_offset	*co, *co1;
 
 	c->flags |= CLIENT_DEAD;
 
@@ -303,6 +341,10 @@ server_client_lost(struct client *c)
 		RB_REMOVE(client_windows, &c->windows, cw);
 		free(cw);
 	}
+	RB_FOREACH_SAFE(co, client_offsets, &c->offsets, co1) {
+		RB_REMOVE(client_offsets, &c->offsets, co);
+		free(co);
+	}
 
 	TAILQ_REMOVE(&clients, c, entry);
 	log_debug("lost client %p", c);
@@ -1368,6 +1410,7 @@ server_client_loop(void)
 				if (focus)
 					server_client_check_pane_focus(wp);
 				server_client_check_pane_resize(wp);
+				server_client_check_pane_buffer(wp);
 			}
 			wp->flags &= ~PANE_REDRAW;
 		}
@@ -1492,6 +1535,88 @@ server_client_check_pane_resize(struct window_pane *wp)
 		log_debug("%s: %%%u timer running", __func__, wp->id);
 }
 
+/* Check pane buffer size. */
+static void
+server_client_check_pane_buffer(struct window_pane *wp)
+{
+	struct evbuffer			*evb = wp->event->input;
+	size_t				 minimum;
+	struct client			*c;
+	struct client_offset		*co;
+	int				 off = !TAILQ_EMPTY(&clients);
+
+	/*
+	 * Work out the minimum acknowledged size. This is the most that can be
+	 * removed from the buffer.
+	 */
+	minimum = wp->offset.acknowledged;
+	if (wp->pipe_fd != -1 && wp->pipe_offset.acknowledged < minimum)
+		minimum = wp->pipe_offset.acknowledged;
+	TAILQ_FOREACH(c, &clients, entry) {
+		if (c->session == NULL)
+			continue;
+		if ((~c->flags & CLIENT_CONTROL) ||
+		    (c->flags & CLIENT_CONTROL_NOOUTPUT) ||
+		    (co = server_client_get_pane_offset(c, wp)) == NULL) {
+			off = 0;
+			continue;
+		}
+		if (~co->flags & CLIENT_OFFSET_OFF)
+			off = 0;
+		log_debug("%s: %s has %zu bytes used, %zu bytes acknowledged "
+		    "for %%%u", __func__, c->name, co->offset.used,
+		    co->offset.acknowledged, wp->id);
+		if (co->offset.acknowledged < minimum)
+			minimum = co->offset.acknowledged;
+	}
+	minimum -= wp->base_offset;
+	if (minimum == 0)
+		goto out;
+
+	/* Drain the buffer. */
+	log_debug("%s: %%%u has %zu minimum (of %zu) bytes acknowledged",
+	    __func__, wp->id, minimum, EVBUFFER_LENGTH(evb));
+	evbuffer_drain(evb, minimum);
+
+	/*
+	 * Adjust the base offset. If it would roll over, all the offsets into
+	 * the buffer need to be adjusted.
+	 */
+	if (wp->base_offset > SIZE_MAX - minimum) {
+		log_debug("%s: %%%u base offset has wrapped", __func__, wp->id);
+		wp->offset.acknowledged -= wp->base_offset;
+		wp->offset.used -= wp->base_offset;
+		if (wp->pipe_fd != -1) {
+			wp->pipe_offset.acknowledged -= wp->base_offset;
+			wp->pipe_offset.used -= wp->base_offset;
+		}
+		TAILQ_FOREACH(c, &clients, entry) {
+			if (c->session == NULL || (~c->flags & CLIENT_CONTROL))
+				continue;
+			co = server_client_get_pane_offset(c, wp);
+			if (co != NULL) {
+				co->offset.acknowledged -= wp->base_offset;
+				co->offset.used -= wp->base_offset;
+			}
+		}
+		wp->base_offset = minimum;
+	} else
+		wp->base_offset += minimum;
+
+out:
+	/*
+	 * If there is data remaining, and there are no clients able to consume
+	 * it, do not read any more. This is true when 1) there are attached
+	 * clients 2) all the clients are control clients 3) all of them have
+	 * either the OFF flag set, or are otherwise not able to accept any
+	 * more data for this pane.
+	 */
+	if (off)
+		bufferevent_disable(wp->event, EV_READ);
+	else
+		bufferevent_enable(wp->event, EV_READ);
+}
+
 /* Check whether pane should be focused. */
 static void
 server_client_check_pane_focus(struct window_pane *wp)
diff --git a/spawn.c b/spawn.c
index 01f19097..d5b52ffa 100644
--- a/spawn.c
+++ b/spawn.c
@@ -439,7 +439,6 @@ spawn_pane(struct spawn_context *sc, char **cause)
 	_exit(1);
 
 complete:
-	new_wp->pipe_off = 0;
 	new_wp->flags &= ~PANE_EXITED;
 
 	sigprocmask(SIG_SETMASK, &oldset, NULL);
diff --git a/tmux.1 b/tmux.1
index 9f687bd2..e7131829 100644
--- a/tmux.1
+++ b/tmux.1
@@ -1251,6 +1251,7 @@ and sets an environment variable for the newly created session; it may be
 specified multiple times.
 .It Xo Ic refresh-client
 .Op Fl cDlLRSU
+.Op Fl A Ar pane:state
 .Op Fl C Ar XxY
 .Op Fl f Ar flags
 .Op Fl t Ar target-client
@@ -1295,7 +1296,25 @@ window, changing the current window in the attached session will reset
 it.
 .Pp
 .Fl C
-sets the width and height of a control client.
+sets the width and height of a control mode client.
+.Fl A
+informs
+.Nm
+of a control mode client's interest in a pane.
+The argument is a pane ID (with leading
+.Ql % ) ,
+a colon, then one of
+.Ql on
+or
+.Ql off .
+If
+.Ql off ,
+.Nm
+will not send output from the pane to the client and if all clients have turned
+the pane off, will stop reading from the pane.
+.Fl A
+may be given multiple times.
+.Pp
 .Fl f
 sets a comma-separated list of client flags, see
 .Ic attach-session .
diff --git a/tmux.h b/tmux.h
index cdc57474..cfb63e36 100644
--- a/tmux.h
+++ b/tmux.h
@@ -896,6 +896,12 @@ struct window_mode_entry {
 	TAILQ_ENTRY (window_mode_entry)	 entry;
 };
 
+/* Offsets into pane buffer. */
+struct window_pane_offset {
+	size_t	used;
+	size_t	acknowledged;
+};
+
 /* Child window structure. */
 struct window_pane {
 	u_int		 id;
@@ -946,6 +952,8 @@ struct window_pane {
 
 	int		 fd;
 	struct bufferevent *event;
+	struct window_pane_offset offset;
+	size_t		 base_offset;
 
 	struct event	 resize_timer;
 
@@ -957,7 +965,7 @@ struct window_pane {
 
 	int		 pipe_fd;
 	struct bufferevent *pipe_event;
-	size_t		 pipe_off;
+	struct window_pane_offset pipe_offset;
 
 	struct screen	*screen;
 	struct screen	 base;
@@ -1541,6 +1549,18 @@ struct client_window {
 };
 RB_HEAD(client_windows, client_window);
 
+/* Client offsets. */
+struct client_offset {
+	u_int				pane;
+
+	struct window_pane_offset	offset;
+	int				flags;
+#define CLIENT_OFFSET_OFF 0x1
+
+	RB_ENTRY(client_offset)		entry;
+};
+RB_HEAD(client_offsets, client_offset);
+
 /* Client connection. */
 typedef int (*prompt_input_cb)(struct client *, void *, const char *, int);
 typedef void (*prompt_free_cb)(void *);
@@ -1555,6 +1575,7 @@ struct client {
 	struct cmdq_list *queue;
 
 	struct client_windows windows;
+	struct client_offsets offsets;
 
 	pid_t		 pid;
 	int		 fd;
@@ -1927,7 +1948,6 @@ char		*format_trim_right(const char *, u_int);
 
 /* notify.c */
 void	notify_hook(struct cmdq_item *, const char *);
-void	notify_input(struct window_pane *, const u_char *, size_t);
 void	notify_client(const char *, struct client *);
 void	notify_session(const char *, struct session *);
 void	notify_winlink(const char *, struct winlink *);
@@ -2339,6 +2359,11 @@ void printflike(1, 2) server_add_message(const char *, ...);
 
 /* server-client.c */
 RB_PROTOTYPE(client_windows, client_window, entry, server_client_window_cmp);
+RB_PROTOTYPE(client_offsets, client_offset, entry, server_client_offset_cmp);
+struct client_offset *server_client_get_pane_offset(struct client *,
+	     struct window_pane *);
+struct client_offset *server_client_add_pane_offset(struct client *,
+	     struct window_pane *);
 u_int	 server_client_how_many(void);
 void	 server_client_set_overlay(struct client *, u_int, overlay_check_cb,
 	     overlay_mode_cb, overlay_draw_cb, overlay_key_cb,
@@ -2684,6 +2709,12 @@ void		 winlink_clear_flags(struct winlink *);
 int		 winlink_shuffle_up(struct session *, struct winlink *);
 int		 window_pane_start_input(struct window_pane *,
 		     struct cmdq_item *, char **);
+void		*window_pane_get_new_data(struct window_pane *,
+		     struct window_pane_offset *, size_t *);
+void		 window_pane_update_used_data(struct window_pane *,
+		     struct window_pane_offset *, size_t, int);
+void		 window_pane_acknowledge_data(struct window_pane *,
+		     struct window_pane_offset *, size_t);
 
 /* layout.c */
 u_int		 layout_count_cells(struct layout_cell *);
@@ -2800,6 +2831,7 @@ char	*parse_window_name(const char *);
 /* control.c */
 void	control_start(struct client *);
 void printflike(2, 3) control_write(struct client *, const char *, ...);
+void	control_write_output(struct client *, struct window_pane *);
 
 /* control-notify.c */
 void	control_notify_input(struct client *, struct window_pane *,
diff --git a/window.c b/window.c
index 2427bd6e..c7a17551 100644
--- a/window.c
+++ b/window.c
@@ -881,7 +881,6 @@ window_pane_create(struct window *w, u_int sx, u_int sy, u_int hlimit)
 	wp->sy = wp->osx = sy;
 
 	wp->pipe_fd = -1;
-	wp->pipe_off = 0;
 	wp->pipe_event = NULL;
 
 	screen_init(&wp->base, sx, sy, hlimit);
@@ -933,22 +932,28 @@ window_pane_destroy(struct window_pane *wp)
 static void
 window_pane_read_callback(__unused struct bufferevent *bufev, void *data)
 {
-	struct window_pane	*wp = data;
-	struct evbuffer		*evb = wp->event->input;
-	size_t			 size = EVBUFFER_LENGTH(evb);
-	char			*new_data;
-	size_t			 new_size;
+	struct window_pane		*wp = data;
+	struct evbuffer			*evb = wp->event->input;
+	struct window_pane_offset	*wpo = &wp->pipe_offset;
+	size_t				 size = EVBUFFER_LENGTH(evb);
+	char				*new_data;
+	size_t				 new_size;
+	struct client			*c;
 
-	new_size = size - wp->pipe_off;
-	if (wp->pipe_fd != -1 && new_size > 0) {
-		new_data = EVBUFFER_DATA(evb) + wp->pipe_off;
-		bufferevent_write(wp->pipe_event, new_data, new_size);
+	if (wp->pipe_fd != -1) {
+		new_data = window_pane_get_new_data(wp, wpo, &new_size);
+		if (new_size > 0) {
+			bufferevent_write(wp->pipe_event, new_data, new_size);
+			window_pane_update_used_data(wp, wpo, new_size, 1);
+		}
 	}
 
 	log_debug("%%%u has %zu bytes", wp->id, size);
+	TAILQ_FOREACH(c, &clients, entry) {
+		if (c->session != NULL && c->flags & CLIENT_CONTROL)
+			control_write_output(c, wp);
+	}
 	input_parse_pane(wp);
-
-	wp->pipe_off = EVBUFFER_LENGTH(evb);
 }
 
 static void
@@ -1541,3 +1546,40 @@ window_pane_start_input(struct window_pane *wp, struct cmdq_item *item,
 
 	return (0);
 }
+
+void *
+window_pane_get_new_data(struct window_pane *wp,
+    struct window_pane_offset *wpo, size_t *size)
+{
+	size_t	used = wpo->used - wp->base_offset;
+
+	*size = EVBUFFER_LENGTH(wp->event->input) - used;
+	return (EVBUFFER_DATA(wp->event->input) + used);
+}
+
+void
+window_pane_update_used_data(struct window_pane *wp,
+    struct window_pane_offset *wpo, size_t size, int acknowledge)
+{
+	size_t	used = wpo->used - wp->base_offset;
+
+	if (size > EVBUFFER_LENGTH(wp->event->input) - used)
+		size = EVBUFFER_LENGTH(wp->event->input) - used;
+	wpo->used += size;
+
+	if (acknowledge)
+		window_pane_acknowledge_data(wp, wpo, size);
+}
+
+void
+window_pane_acknowledge_data(struct window_pane *wp,
+    struct window_pane_offset *wpo, size_t size)
+{
+	size_t	acknowledged = wpo->acknowledged - wp->base_offset;
+
+	if (size > EVBUFFER_LENGTH(wp->event->input) - acknowledged)
+		size = EVBUFFER_LENGTH(wp->event->input) - acknowledged;
+	wpo->acknowledged += size;
+	if (wpo->acknowledged > wpo->used)
+		wpo->acknowledged = wpo->used;
+}