patch 7.4.1669

Problem:    When writing buffer lines to a pipe Vim may block.
Solution:   Avoid blocking, write more lines later.

src/channel.c

@@ -434,7 +434,6 @@ channel_read_fd(int fd)
 
 /*
  * Read a command from netbeans.
- * TODO: instead of channel ID use the FD.
  */
 #ifdef FEAT_GUI_X11
     static void
@@ -974,6 +973,7 @@ channel_set_job(channel_T *channel, job_T *job, jobopt_T *options)
 		/* Special mode: send last-but-one line when appending a line
 		 * to the buffer. */
 		in_part->ch_buffer->b_write_to_channel = TRUE;
+		in_part->ch_buf_append = TRUE;
 		in_part->ch_buf_top =
 				   in_part->ch_buffer->b_ml.ml_line_count + 1;
 	    }
@@ -1048,6 +1048,8 @@ channel_set_options(channel_T *channel, jobopt_T *opt)
 	channel->ch_part[PART_OUT].ch_timeout = opt->jo_out_timeout;
     if (opt->jo_set & JO_ERR_TIMEOUT)
 	channel->ch_part[PART_ERR].ch_timeout = opt->jo_err_timeout;
+    if (opt->jo_set & JO_BLOCK_WRITE)
+	channel->ch_part[PART_IN].ch_block_write = 1;
 
     if (opt->jo_set & JO_CALLBACK)
     {
@@ -1184,7 +1186,6 @@ write_buf_line(buf_T *buf, linenr_T lnum, channel_T *channel)
     int	    len = (int)STRLEN(line);
     char_u  *p;
 
-    /* TODO: check if channel can be written to, do not block on write */
     if ((p = alloc(len + 2)) == NULL)
 	return;
     STRCPY(p, line);
@@ -1194,10 +1195,79 @@ write_buf_line(buf_T *buf, linenr_T lnum, channel_T *channel)
     vim_free(p);
 }
 
+/*
+ * Return TRUE if "channel" can be written to.
+ * Returns FALSE if the input is closed or the write would block.
+ */
+    static int
+can_write_buf_line(channel_T *channel)
+{
+    chanpart_T *in_part = &channel->ch_part[PART_IN];
+
+    if (in_part->ch_fd == INVALID_FD)
+	return FALSE;  /* pipe was closed */
+
+    /* for testing: block every other attempt to write */
+    if (in_part->ch_block_write == 1)
+	in_part->ch_block_write = -1;
+    else if (in_part->ch_block_write == -1)
+	in_part->ch_block_write = 1;
+
+    /* TODO: Win32 implementation, probably using WaitForMultipleObjects() */
+#ifndef WIN32
+    {
+# if defined(HAVE_SELECT)
+	struct timeval	tval;
+	fd_set		wfds;
+	int		ret;
+
+	FD_ZERO(&wfds);
+	FD_SET((int)in_part->ch_fd, &wfds);
+	tval.tv_sec = 0;
+	tval.tv_usec = 0;
+	for (;;)
+	{
+	    ret = select((int)in_part->ch_fd + 1, NULL, &wfds, NULL, &tval);
+#  ifdef EINTR
+	    SOCK_ERRNO;
+	    if (ret == -1 && errno == EINTR)
+		continue;
+#  endif
+	    if (ret <= 0 || in_part->ch_block_write == 1)
+	    {
+		if (ret > 0)
+		    ch_log(channel, "FAKED Input not ready for writing");
+		else
+		    ch_log(channel, "Input not ready for writing");
+		return FALSE;
+	    }
+	    break;
+	}
+# else
+	struct pollfd	fds;
+
+	fds.fd = in_part->ch_fd;
+	fds.events = POLLOUT;
+	if (poll(&fds, 1, 0) <= 0)
+	{
+	    ch_log(channel, "Input not ready for writing");
+	    return FALSE;
+	}
+	if (in_part->ch_block_write == 1)
+	{
+	    ch_log(channel, "FAKED Input not ready for writing");
+	    return FALSE;
+	}
+# endif
+    }
+#endif
+    return TRUE;
+}
+
 /*
  * Write any lines to the input channel.
  */
-    void
+    static void
 channel_write_in(channel_T *channel)
 {
     chanpart_T *in_part = &channel->ch_part[PART_IN];
@@ -1205,21 +1275,20 @@ channel_write_in(channel_T *channel)
     buf_T	*buf = in_part->ch_buffer;
     int		written = 0;
 
-    if (buf == NULL)
-	return;
+    if (buf == NULL || in_part->ch_buf_append)
+	return;  /* no buffer or using appending */
     if (!buf_valid(buf) || buf->b_ml.ml_mfp == NULL)
     {
 	/* buffer was wiped out or unloaded */
 	in_part->ch_buffer = NULL;
 	return;
     }
-    if (in_part->ch_fd == INVALID_FD)
-	/* pipe was closed */
-	return;
 
     for (lnum = in_part->ch_buf_top; lnum <= in_part->ch_buf_bot
 				   && lnum <= buf->b_ml.ml_line_count; ++lnum)
     {
+	if (!can_write_buf_line(channel))
+	    break;
 	write_buf_line(buf, lnum, channel);
 	++written;
     }
@@ -1230,6 +1299,37 @@ channel_write_in(channel_T *channel)
 	ch_logn(channel, "written %d lines to channel", written);
 
     in_part->ch_buf_top = lnum;
+    if (lnum > buf->b_ml.ml_line_count)
+    {
+	/* Writing is done, no longer need the buffer. */
+	in_part->ch_buffer = NULL;
+	ch_log(channel, "Finished writing all lines to channel");
+    }
+    else
+	ch_logn(channel, "Still %d more lines to write",
+					  buf->b_ml.ml_line_count - lnum + 1);
+}
+
+/*
+ * Write any lines waiting to be written to a channel.
+ */
+    void
+channel_write_any_lines()
+{
+    channel_T	*channel;
+
+    for (channel = first_channel; channel != NULL; channel = channel->ch_next)
+    {
+	chanpart_T  *in_part = &channel->ch_part[PART_IN];
+
+	if (in_part->ch_buffer != NULL)
+	{
+	    if (in_part->ch_buf_append)
+		channel_write_new_lines(in_part->ch_buffer);
+	    else
+		channel_write_in(channel);
+	}
+    }
 }
 
 /*
@@ -1249,15 +1349,16 @@ channel_write_new_lines(buf_T *buf)
 	linenr_T    lnum;
 	int	    written = 0;
 
-	if (in_part->ch_buffer == buf)
+	if (in_part->ch_buffer == buf && in_part->ch_buf_append)
 	{
 	    if (in_part->ch_fd == INVALID_FD)
-		/* pipe was closed */
-		continue;
+		continue;  /* pipe was closed */
 	    found_one = TRUE;
 	    for (lnum = in_part->ch_buf_bot; lnum < buf->b_ml.ml_line_count;
 								       ++lnum)
 	    {
+		if (!can_write_buf_line(channel))
+		    break;
 		write_buf_line(buf, lnum, channel);
 		++written;
 	    }
@@ -1266,6 +1367,9 @@ channel_write_new_lines(buf_T *buf)
 		ch_logn(channel, "written line %d to channel", (int)lnum - 1);
 	    else if (written > 1)
 		ch_logn(channel, "written %d lines to channel", written);
+	    if (lnum < buf->b_ml.ml_line_count)
+		ch_logn(channel, "Still %d more lines to write",
+					      buf->b_ml.ml_line_count - lnum);
 
 	    in_part->ch_buf_bot = lnum;
 	}
@@ -1325,11 +1429,34 @@ channel_get(channel_T *channel, int part)
     static char_u *
 channel_get_all(channel_T *channel, int part)
 {
-    /* Concatenate everything into one buffer.
-     * TODO: avoid multiple allocations. */
-    while (channel_collapse(channel, part) == OK)
-	;
-    return channel_get(channel, part);
+    readq_T *head = &channel->ch_part[part].ch_head;
+    readq_T *node = head->rq_next;
+    long_u  len = 1;
+    char_u  *res;
+    char_u  *p;
+
+    /* If there is only one buffer just get that one. */
+    if (head->rq_next == NULL || head->rq_next->rq_next == NULL)
+	return channel_get(channel, part);
+
+    /* Concatenate everything into one buffer. */
+    for (node = head->rq_next; node != NULL; node = node->rq_next)
+	len += (long_u)STRLEN(node->rq_buffer);
+    res = lalloc(len, TRUE);
+    if (res == NULL)
+	return NULL;
+    *res = NUL;
+    for (node = head->rq_next; node != NULL; node = node->rq_next)
+	STRCAT(res, node->rq_buffer);
+
+    /* Free all buffers */
+    do
+    {
+	p = channel_get(channel, part);
+	vim_free(p);
+    } while (p != NULL);
+
+    return res;
 }
 
 /*
@@ -1365,10 +1492,12 @@ channel_collapse(channel_T *channel, int part)
 
 /*
  * Store "buf[len]" on "channel"/"part".
+ * When "prepend" is TRUE put in front, otherwise append at the end.
  * Returns OK or FAIL.
  */
     static int
-channel_save(channel_T *channel, int part, char_u *buf, int len, char *lead)
+channel_save(channel_T *channel, int part, char_u *buf, int len,
+						      int prepend, char *lead)
 {
     readq_T *node;
     readq_T *head = &channel->ch_part[part].ch_head;
@@ -1400,14 +1529,28 @@ channel_save(channel_T *channel, int part, char_u *buf, int len, char *lead)
 	node->rq_buffer[len] = NUL;
     }
 
-    /* append node to the tail of the queue */
-    node->rq_next = NULL;
-    node->rq_prev = head->rq_prev;
-    if (head->rq_prev == NULL)
+    if (prepend)
+    {
+	/* preend node to the head of the queue */
+	node->rq_next = head->rq_next;
+	node->rq_prev = NULL;
+	if (head->rq_next == NULL)
+	    head->rq_prev = node;
+	else
+	    head->rq_next->rq_prev = node;
 	head->rq_next = node;
+    }
     else
-	head->rq_prev->rq_next = node;
-    head->rq_prev = node;
+    {
+	/* append node to the tail of the queue */
+	node->rq_next = NULL;
+	node->rq_prev = head->rq_prev;
+	if (head->rq_prev == NULL)
+	    head->rq_next = node;
+	else
+	    head->rq_prev->rq_next = node;
+	head->rq_prev = node;
+    }
 
     if (log_fd != NULL && lead != NULL)
     {
@@ -1420,6 +1563,42 @@ channel_save(channel_T *channel, int part, char_u *buf, int len, char *lead)
     return OK;
 }
 
+    static int
+channel_fill(js_read_T *reader)
+{
+    channel_T	*channel = (channel_T *)reader->js_cookie;
+    int		part = reader->js_cookie_arg;
+    char_u	*next = channel_get(channel, part);
+    int		unused;
+    int		len;
+    char_u	*p;
+
+    if (next == NULL)
+	return FALSE;
+
+    unused = reader->js_end - reader->js_buf - reader->js_used;
+    if (unused > 0)
+    {
+	/* Prepend unused text. */
+	len = (int)STRLEN(next);
+	p = alloc(unused + len + 1);
+	if (p == NULL)
+	{
+	    vim_free(next);
+	    return FALSE;
+	}
+	mch_memmove(p, reader->js_buf + reader->js_used, unused);
+	mch_memmove(p + unused, next, len + 1);
+	vim_free(next);
+	next = p;
+    }
+
+    vim_free(reader->js_buf);
+    reader->js_buf = next;
+    reader->js_used = 0;
+    return TRUE;
+}
+
 /*
  * Use the read buffer of "channel"/"part" and parse a JSON message that is
  * complete.  The messages are added to the queue.
@@ -1439,19 +1618,17 @@ channel_parse_json(channel_T *channel, int part)
     if (channel_peek(channel, part) == NULL)
 	return FALSE;
 
-    /* TODO: make reader work properly */
-    /* reader.js_buf = channel_peek(channel, part); */
-    reader.js_buf = channel_get_all(channel, part);
+    reader.js_buf = channel_get(channel, part);
     reader.js_used = 0;
-    reader.js_fill = NULL;
-    /* reader.js_fill = channel_fill; */
+    reader.js_fill = channel_fill;
     reader.js_cookie = channel;
+    reader.js_cookie_arg = part;
 
     /* When a message is incomplete we wait for a short while for more to
      * arrive.  After the delay drop the input, otherwise a truncated string
      * or list will make us hang.  */
     status = json_decode(&reader, &listtv,
-		     chanpart->ch_mode == MODE_JS ? JSON_JS : 0);
+				  chanpart->ch_mode == MODE_JS ? JSON_JS : 0);
     if (status == OK)
     {
 	/* Only accept the response when it is a list with at least two
@@ -1552,10 +1729,10 @@ channel_parse_json(channel_T *channel, int part)
     }
     else if (reader.js_buf[reader.js_used] != NUL)
     {
-	/* Put the unread part back into the channel.
-	 * TODO: insert in front */
+	/* Put the unread part back into the channel. */
 	channel_save(channel, part, reader.js_buf + reader.js_used,
-		(int)(reader.js_end - reader.js_buf) - reader.js_used, NULL);
+			(int)(reader.js_end - reader.js_buf) - reader.js_used,
+								  TRUE, NULL);
 	ret = status == MAYBE ? FALSE: TRUE;
     }
     else
@@ -2307,6 +2484,57 @@ channel_free_all(void)
 /* Buffer size for reading incoming messages. */
 #define MAXMSGSIZE 4096
 
+#if defined(HAVE_SELECT)
+/*
+ * Add write fds where we are waiting for writing to be possible.
+ */
+    static int
+channel_fill_wfds(int maxfd_arg, fd_set *wfds)
+{
+    int		maxfd = maxfd_arg;
+    channel_T	*ch;
+
+    for (ch = first_channel; ch != NULL; ch = ch->ch_next)
+    {
+	chanpart_T  *in_part = &ch->ch_part[PART_IN];
+
+	if (in_part->ch_fd != INVALID_FD && in_part->ch_buffer != NULL)
+	{
+	    FD_SET((int)in_part->ch_fd, wfds);
+	    if ((int)in_part->ch_fd >= maxfd)
+		maxfd = (int)in_part->ch_fd + 1;
+	}
+    }
+    return maxfd;
+}
+#else
+/*
+ * Add write fds where we are waiting for writing to be possible.
+ */
+    static int
+channel_fill_poll_write(int nfd_in, struct pollfd *fds)
+{
+    int		nfd = nfd_in;
+    channel_T	*ch;
+
+    for (ch = first_channel; ch != NULL; ch = ch->ch_next)
+    {
+	chanpart_T  *in_part = &ch->ch_part[PART_IN];
+
+	if (in_part->ch_fd != INVALID_FD && in_part->ch_buffer != NULL)
+	{
+	    in_part->ch_poll_idx = nfd;
+	    fds[nfd].fd = in_part->ch_fd;
+	    fds[nfd].events = POLLOUT;
+	    ++nfd;
+	}
+	else
+	    in_part->ch_poll_idx = -1;
+    }
+    return nfd;
+}
+#endif
+
 /*
  * Check for reading from "fd" with "timeout" msec.
  * Return FAIL when there is nothing to read.
@@ -2321,8 +2549,9 @@ channel_wait(channel_T *channel, sock_T fd, int timeout)
     if (fd != channel->CH_SOCK_FD)
     {
 	DWORD	nread;
-	int	diff;
+	int	sleep_time;
 	DWORD	deadline = GetTickCount() + timeout;
+	int	delay = 1;
 
 	/* reading from a pipe, not a socket */
 	while (TRUE)
@@ -2330,12 +2559,21 @@ channel_wait(channel_T *channel, sock_T fd, int timeout)
 	    if (PeekNamedPipe((HANDLE)fd, NULL, 0, NULL, &nread, NULL)
 								 && nread > 0)
 		return OK;
-	    diff = deadline - GetTickCount();
-	    if (diff <= 0)
+
+	    /* perhaps write some buffer lines */
+	    channel_write_any_lines();
+
+	    sleep_time = deadline - GetTickCount();
+	    if (sleep_time <= 0)
 		break;
-	    /* Wait for 5 msec.
-	     * TODO: increase the sleep time when looping more often */
-	    Sleep(5);
+	    /* Wait for a little while.  Very short at first, up to 10 msec
+	     * after looping a few times. */
+	    if (sleep_time > delay)
+		sleep_time = delay;
+	    Sleep(sleep_time);
+	    delay = delay * 2;
+	    if (delay > 10)
+		delay = 10;
 	}
     }
     else
@@ -2344,31 +2582,56 @@ channel_wait(channel_T *channel, sock_T fd, int timeout)
 #if defined(HAVE_SELECT)
 	struct timeval	tval;
 	fd_set		rfds;
-	int			ret;
+	fd_set		wfds;
+	int		ret;
+	int		maxfd;
 
-	FD_ZERO(&rfds);
-	FD_SET((int)fd, &rfds);
 	tval.tv_sec = timeout / 1000;
 	tval.tv_usec = (timeout % 1000) * 1000;
 	for (;;)
 	{
-	    ret = select((int)fd + 1, &rfds, NULL, NULL, &tval);
+	    FD_ZERO(&rfds);
+	    FD_SET((int)fd, &rfds);
+
+	    /* Write lines to a pipe when a pipe can be written to.  Need to
+	     * set this every time, some buffers may be done. */
+	    maxfd = (int)fd + 1;
+	    FD_ZERO(&wfds);
+	    maxfd = channel_fill_wfds(maxfd, &wfds);
+
+	    ret = select(maxfd, &rfds, &wfds, NULL, &tval);
 # ifdef EINTR
 	    SOCK_ERRNO;
 	    if (ret == -1 && errno == EINTR)
 		continue;
 # endif
 	    if (ret > 0)
-		return OK;
+	    {
+		if (FD_ISSET(fd, &rfds))
+		    return OK;
+		channel_write_any_lines();
+		continue;
+	    }
 	    break;
 	}
 #else
-	struct pollfd	fds;
+	for (;;)
+	{
+	    struct pollfd   fds[MAX_OPEN_CHANNELS + 1];
+	    int		    nfd = 1;
 
-	fds.fd = fd;
-	fds.events = POLLIN;
-	if (poll(&fds, 1, timeout) > 0)
-	    return OK;
+	    fds[0].fd = fd;
+	    fds[0].events = POLLIN;
+	    nfd = channel_fill_poll_write(nfd, fds);
+	    if (poll(fds, nfd, timeout) > 0)
+	    {
+		if (fds[0].revents & POLLIN)
+		    return OK;
+		channel_write_any_lines();
+		continue;
+	    }
+	    break;
+	}
 #endif
     }
     return FAIL;
@@ -2419,7 +2682,7 @@ channel_read(channel_T *channel, int part, char *func)
 	    break;	/* error or nothing more to read */
 
 	/* Store the read message in the queue. */
-	channel_save(channel, part, buf, len, "RECV ");
+	channel_save(channel, part, buf, len, FALSE, "RECV ");
 	readlen += len;
 	if (len < MAXMSGSIZE)
 	    break;	/* did read everything that's available */
@@ -2446,11 +2709,10 @@ channel_read(channel_T *channel, int part, char *func)
 	if (channel->ch_part[part].ch_mode == MODE_RAW
 				 || channel->ch_part[part].ch_mode == MODE_NL)
 	    channel_save(channel, part, (char_u *)DETACH_MSG_RAW,
-					 (int)STRLEN(DETACH_MSG_RAW), "PUT ");
+				  (int)STRLEN(DETACH_MSG_RAW), FALSE, "PUT ");
 
-	/* TODO: When reading from stdout is not possible, should we try to
-	 * keep stdin and stderr open?  Probably not, assume the other side
-	 * has died. */
+	/* When reading from stdout is not possible, assume the other side has
+	 * died. */
 	channel_close(channel, TRUE);
 	if (channel->ch_nb_close_cb != NULL)
 	    (*channel->ch_nb_close_cb)();
@@ -2933,10 +3195,12 @@ channel_poll_setup(int nfd_in, void *fds_in)
     {
 	for (part = PART_SOCK; part < PART_IN; ++part)
 	{
-	    if (channel->ch_part[part].ch_fd != INVALID_FD)
+	    chanpart_T	*ch_part = &channel->ch_part[part];
+
+	    if (ch_part->ch_fd != INVALID_FD)
 	    {
-		channel->ch_part[part].ch_poll_idx = nfd;
-		fds[nfd].fd = channel->ch_part[part].ch_fd;
+		ch_part->ch_poll_idx = nfd;
+		fds[nfd].fd = ch_part->ch_fd;
 		fds[nfd].events = POLLIN;
 		nfd++;
 	    }
@@ -2945,6 +3209,8 @@ channel_poll_setup(int nfd_in, void *fds_in)
 	}
     }
 
+    nfd = channel_fill_poll_write(nfd, fds);
+
     return nfd;
 }
 
@@ -2958,19 +3224,35 @@ channel_poll_check(int ret_in, void *fds_in)
     channel_T	*channel;
     struct	pollfd *fds = fds_in;
     int		part;
+    int		idx;
+    chanpart_T	*in_part;
 
     for (channel = first_channel; channel != NULL; channel = channel->ch_next)
     {
 	for (part = PART_SOCK; part < PART_IN; ++part)
 	{
-	    int idx = channel->ch_part[part].ch_poll_idx;
+	    idx = channel->ch_part[part].ch_poll_idx;
 
-	    if (ret > 0 && idx != -1 && fds[idx].revents & POLLIN)
+	    if (ret > 0 && idx != -1 && (fds[idx].revents & POLLIN))
 	    {
 		channel_read(channel, part, "channel_poll_check");
 		--ret;
 	    }
 	}
+
+	in_part = &channel->ch_part[PART_IN];
+	idx = in_part->ch_poll_idx;
+	if (ret > 0 && idx != -1 && (fds[idx].revents & POLLOUT))
+	{
+	    if (in_part->ch_buf_append)
+	    {
+		if (in_part->ch_buffer != NULL)
+		    channel_write_new_lines(in_part->ch_buffer);
+	    }
+	    else
+		channel_write_in(channel);
+	    --ret;
+	}
     }
 
     return ret;
@@ -2979,14 +3261,15 @@ channel_poll_check(int ret_in, void *fds_in)
 
 # if (!defined(WIN32) && defined(HAVE_SELECT)) || defined(PROTO)
 /*
- * The type of "rfds" is hidden to avoid problems with the function proto.
+ * The "fd_set" type is hidden to avoid problems with the function proto.
  */
     int
-channel_select_setup(int maxfd_in, void *rfds_in)
+channel_select_setup(int maxfd_in, void *rfds_in, void *wfds_in)
 {
     int		maxfd = maxfd_in;
     channel_T	*channel;
     fd_set	*rfds = rfds_in;
+    fd_set	*wfds = wfds_in;
     int		part;
 
     for (channel = first_channel; channel != NULL; channel = channel->ch_next)
@@ -3004,19 +3287,23 @@ channel_select_setup(int maxfd_in, void *rfds_in)
 	}
     }
 
+    maxfd = channel_fill_wfds(maxfd, wfds);
+
     return maxfd;
 }
 
 /*
- * The type of "rfds" is hidden to avoid problems with the function proto.
+ * The "fd_set" type is hidden to avoid problems with the function proto.
  */
     int
-channel_select_check(int ret_in, void *rfds_in)
+channel_select_check(int ret_in, void *rfds_in, void *wfds_in)
 {
     int		ret = ret_in;
     channel_T	*channel;
     fd_set	*rfds = rfds_in;
+    fd_set	*wfds = wfds_in;
     int		part;
+    chanpart_T	*in_part;
 
     for (channel = first_channel; channel != NULL; channel = channel->ch_next)
     {
@@ -3030,6 +3317,20 @@ channel_select_check(int ret_in, void *rfds_in)
 		--ret;
 	    }
 	}
+
+	in_part = &channel->ch_part[PART_IN];
+	if (ret > 0 && in_part->ch_fd != INVALID_FD
+					    && FD_ISSET(in_part->ch_fd, wfds))
+	{
+	    if (in_part->ch_buf_append)
+	    {
+		if (in_part->ch_buffer != NULL)
+		    channel_write_new_lines(in_part->ch_buffer);
+	    }
+	    else
+		channel_write_in(channel);
+	    --ret;
+	}
     }
 
     return ret;
@@ -3531,6 +3832,13 @@ get_job_options(typval_T *tv, jobopt_T *opt, int supported)
 		    return FAIL;
 		}
 	    }
+	    else if (STRCMP(hi->hi_key, "block_write") == 0)
+	    {
+		if (!(supported & JO_BLOCK_WRITE))
+		    break;
+		opt->jo_set |= JO_BLOCK_WRITE;
+		opt->jo_block_write = get_tv_number(item);
+	    }
 	    else
 		break;
 	    --todo;
@@ -3750,8 +4058,8 @@ job_start(typval_T *argvars)
     clear_job_options(&opt);
     opt.jo_mode = MODE_NL;
     if (get_job_options(&argvars[1], &opt,
-	    JO_MODE_ALL + JO_CB_ALL + JO_TIMEOUT_ALL
-			    + JO_STOPONEXIT + JO_EXIT_CB + JO_OUT_IO) == FAIL)
+	    JO_MODE_ALL + JO_CB_ALL + JO_TIMEOUT_ALL + JO_STOPONEXIT
+			   + JO_EXIT_CB + JO_OUT_IO + JO_BLOCK_WRITE) == FAIL)
 	return job;
 
     /* Check that when io is "file" that there is a file name. */
@@ -3812,6 +4120,11 @@ job_start(typval_T *argvars)
     {
 	/* Command is a string. */
 	cmd = argvars[0].vval.v_string;
+	if (cmd == NULL || *cmd == NUL)
+	{
+	    EMSG(_(e_invarg));
+	    return job;
+	}
 #ifdef USE_ARGV
 	if (mch_parse_cmd(cmd, FALSE, &argv, &argc) == FAIL)
 	    return job;

src/json.c

@@ -350,8 +350,10 @@ json_skip_white(js_read_T *reader)
 	if (reader->js_fill != NULL && c == NUL)
 	{
 	    if (reader->js_fill(reader))
+	    {
 		reader->js_end = reader->js_buf + STRLEN(reader->js_buf);
-	    continue;
+		continue;
+	    }
 	}
 	if (c == NUL || c > ' ')
 	    break;

src/misc2.c

@@ -6230,6 +6230,9 @@ parse_queued_messages(void)
     netbeans_parse_messages();
 # endif
 # ifdef FEAT_JOB_CHANNEL
+    /* Write any buffer lines still to be written. */
+    channel_write_any_lines();
+
     /* Process the messages queued on channels. */
     channel_parse_messages();
 # endif

src/os_unix.c

@@ -5539,7 +5539,8 @@ RealWaitForChar(int fd, long msec, int *check_for_gpm UNUSED, int *break_loop)
 # endif
 #endif
 #ifndef HAVE_SELECT
-	struct pollfd   fds[6 + MAX_OPEN_CHANNELS];
+			/* each channel may use in, out and err */
+	struct pollfd   fds[6 + 3 * MAX_OPEN_CHANNELS];
 	int		nfd;
 # ifdef FEAT_XCLIPBOARD
 	int		xterm_idx = -1;
@@ -5652,7 +5653,7 @@ RealWaitForChar(int fd, long msec, int *check_for_gpm UNUSED, int *break_loop)
 
 	struct timeval  tv;
 	struct timeval	*tvp;
-	fd_set		rfds, efds;
+	fd_set		rfds, wfds, efds;
 	int		maxfd;
 	long		towait = msec;
 
@@ -5685,6 +5686,7 @@ RealWaitForChar(int fd, long msec, int *check_for_gpm UNUSED, int *break_loop)
 	 */
 select_eintr:
 	FD_ZERO(&rfds);
+	FD_ZERO(&wfds);
 	FD_ZERO(&efds);
 	FD_SET(fd, &rfds);
 # if !defined(__QNX__) && !defined(__CYGWIN32__)
@@ -5725,10 +5727,10 @@ select_eintr:
 	}
 # endif
 # ifdef FEAT_JOB_CHANNEL
-	maxfd = channel_select_setup(maxfd, &rfds);
+	maxfd = channel_select_setup(maxfd, &rfds, &wfds);
 # endif
 
-	ret = select(maxfd + 1, &rfds, NULL, &efds, tvp);
+	ret = select(maxfd + 1, &rfds, &wfds, &efds, tvp);
 	result = ret > 0 && FD_ISSET(fd, &rfds);
 	if (result)
 	    --ret;
@@ -5810,7 +5812,7 @@ select_eintr:
 # endif
 #ifdef FEAT_JOB_CHANNEL
 	if (ret > 0)
-	    ret = channel_select_check(ret, &rfds);
+	    ret = channel_select_check(ret, &rfds, &wfds);
 #endif
 
 #endif /* HAVE_SELECT */

src/proto/channel.pro

@@ -13,7 +13,7 @@ void channel_set_pipes(channel_T *channel, sock_T in, sock_T out, sock_T err);
 void channel_set_job(channel_T *channel, job_T *job, jobopt_T *options);
 void channel_set_options(channel_T *channel, jobopt_T *opt);
 void channel_set_req_callback(channel_T *channel, int part, char_u *callback, partial_T *partial, int id);
-void channel_write_in(channel_T *channel);
+void channel_write_any_lines(void);
 void channel_write_new_lines(buf_T *buf);
 char_u *channel_get(channel_T *channel, int part);
 int channel_collapse(channel_T *channel, int part);
@@ -37,8 +37,8 @@ void ch_expr_common(typval_T *argvars, typval_T *rettv, int eval);
 void ch_raw_common(typval_T *argvars, typval_T *rettv, int eval);
 int channel_poll_setup(int nfd_in, void *fds_in);
 int channel_poll_check(int ret_in, void *fds_in);
-int channel_select_setup(int maxfd_in, void *rfds_in);
-int channel_select_check(int ret_in, void *rfds_in);
+int channel_select_setup(int maxfd_in, void *rfds_in, void *wfds_in);
+int channel_select_check(int ret_in, void *rfds_in, void *wfds_in);
 int channel_parse_messages(void);
 int set_ref_in_channel(int copyID);
 int channel_part_send(channel_T *channel);

src/quickfix.c

@@ -532,7 +532,8 @@ qf_init_ext(
 		else if (tv->v_type == VAR_LIST)
 		{
 		    /* Get the next line from the supplied list */
-		    while (p_li && p_li->li_tv.v_type != VAR_STRING)
+		    while (p_li && (p_li->li_tv.v_type != VAR_STRING
+				|| p_li->li_tv.vval.v_string == NULL))
 			p_li = p_li->li_next;	/* Skip non-string items */
 
 		    if (!p_li)			/* End of the list */

src/structs.h

@@ -1383,12 +1383,15 @@ typedef struct {
 #else
     struct timeval ch_deadline;
 #endif
+    int		ch_block_write;	/* for testing: 0 when not used, -1 when write
+				 * does not block, 1 simulate blocking */
 
     cbq_T	ch_cb_head;	/* dummy node for per-request callbacks */
     char_u	*ch_callback;	/* call when a msg is not handled */
     partial_T	*ch_partial;
 
     buf_T	*ch_buffer;	/* buffer to read from or write to */
+    int		ch_buf_append;	/* write appended lines instead top-bot */
     linenr_T	ch_buf_top;	/* next line to send */
     linenr_T	ch_buf_bot;	/* last line to send */
 } chanpart_T;
@@ -1457,7 +1460,8 @@ struct channel_S {
 #define JO_ERR_BUF	    0x2000000	/* "err_buf" (JO_OUT_BUF << 1) */
 #define JO_IN_BUF	    0x4000000	/* "in_buf" (JO_OUT_BUF << 2) */
 #define JO_CHANNEL	    0x8000000	/* "channel" */
-#define JO_ALL		    0xfffffff
+#define JO_BLOCK_WRITE	    0x10000000	/* "block_write" */
+#define JO_ALL		    0x7fffffff
 
 #define JO_MODE_ALL	(JO_MODE + JO_IN_MODE + JO_OUT_MODE + JO_ERR_MODE)
 #define JO_CB_ALL \
@@ -1499,6 +1503,7 @@ typedef struct
     int		jo_timeout;
     int		jo_out_timeout;
     int		jo_err_timeout;
+    int		jo_block_write;	/* for testing only */
     int		jo_part;
     int		jo_id;
     char_u	jo_soe_buf[NUMBUFLEN];
@@ -2971,6 +2976,7 @@ struct js_reader
 				/* function to fill the buffer or NULL;
                                  * return TRUE when the buffer was filled */
     void	*js_cookie;	/* can be used by js_fill */
+    int		js_cookie_arg;	/* can be used by js_fill */
 };
 typedef struct js_reader js_read_T;
 

src/testdir/test_channel.vim

@@ -791,7 +791,7 @@ func Run_test_pipe_from_buffer(use_name)
 
   sp pipe-input
   call setline(1, ['echo one', 'echo two', 'echo three'])
-  let options = {'in_io': 'buffer'}
+  let options = {'in_io': 'buffer', 'block_write': 1}
   if a:use_name
     let options['in_name'] = 'pipe-input'
   else
@@ -885,7 +885,8 @@ func Test_pipe_io_two_buffers()
 
   let job = job_start(s:python . " test_channel_pipe.py",
 	\ {'in_io': 'buffer', 'in_name': 'pipe-input', 'in_top': 0,
-	\  'out_io': 'buffer', 'out_name': 'pipe-output'})
+	\  'out_io': 'buffer', 'out_name': 'pipe-output',
+	\  'block_write': 1})
   call assert_equal("run", job_status(job))
   try
     exe "normal Gaecho hello\<CR>"
@@ -920,7 +921,8 @@ func Test_pipe_io_one_buffer()
 
   let job = job_start(s:python . " test_channel_pipe.py",
 	\ {'in_io': 'buffer', 'in_name': 'pipe-io', 'in_top': 0,
-	\  'out_io': 'buffer', 'out_name': 'pipe-io'})
+	\  'out_io': 'buffer', 'out_name': 'pipe-io',
+	\  'block_write': 1})
   call assert_equal("run", job_status(job))
   try
     exe "normal Goecho hello\<CR>"
@@ -1201,5 +1203,10 @@ func Test_close_callback()
   call s:run_server('s:test_close_callback')
 endfunc
 
+func Test_job_start_invalid()
+  call assert_fails('call job_start($x)', 'E474:')
+  call assert_fails('call job_start("")', 'E474:')
+endfunc
+
 " Uncomment this to see what happens, output is in src/testdir/channellog.
 " call ch_logfile('channellog', 'w')

src/testdir/test_quickfix.vim

@@ -692,3 +692,8 @@ func Test_caddbuffer_to_empty()
   endtry
   quit!
 endfunc
+
+func Test_cgetexpr_works()
+  " this must not crash Vim
+  cgetexpr [$x]
+endfunc

src/version.c

@@ -748,6 +748,18 @@ static char *(features[]) =
 
 static int included_patches[] =
 {   /* Add new patch number below this line */
+/**/
+    1669,
+/**/
+    1668,
+/**/
+    1667,
+/**/
+    1666,
+/**/
+    1665,
+/**/
+    1664,
 /**/
     1663,
 /**/

src/vim.h

@@ -493,13 +493,11 @@ typedef unsigned long u8char_T;	    /* long should be 32 bits or more */
 #ifndef HAVE_SELECT
 # ifdef HAVE_SYS_POLL_H
 #  include <sys/poll.h>
-#  define HAVE_POLL
 # elif defined(WIN32)
 #  define HAVE_SELECT
 # else
 #  ifdef HAVE_POLL_H
 #   include <poll.h>
-#   define HAVE_POLL
 #  endif
 # endif
 #endif