rpc: close channel if stream was closed

f_jobstop()/f_rpcstop() .. process_stop() .. process_close_in(proc)
closes the write-stream of a RPC channel. But there might be
a pending RPC notification on the queue, which may get processed just
before the channel is closed.

To handle that case, check the Stream.closed in
channel.c:receive_msgpack().

Before this change, the above scenario could trigger
this assert(!stream->closed) in wstream_write():

    0x00007f96e1cd3428 in __GI_raise (sig=sig@entry=6) at ../sysdeps/unix/sysv/linux/raise.c:54
    0x00007f96e1cd502a in __GI_abort () at abort.c:89
    0x00007f96e1ccbbd7 in __assert_fail_base (fmt=<optimized out>, assertion=assertion@entry=0x768f9b "!stream->closed",
    file=file@entry=0x768f70 "../src/nvim/event/wstream.c", line=line@entry=77,
    function=function@entry=0x768fb0 <__PRETTY_FUNCTION__.13735> "wstream_write") at assert.c:92
    0x00007f96e1ccbc82 in __GI___assert_fail (assertion=0x768f9b "!stream->closed", file=0x768f70 "../src/nvim/event/wstream.c", line=77,
    function=0x768fb0 <__PRETTY_FUNCTION__.13735> "wstream_write") at assert.c:101
    0x00000000004d2c1f in wstream_write (stream=0x7f96e0a35078, buffer=0x7f96e09f9b40) at ../src/nvim/event/wstream.c:77
    0x00000000005857b2 in channel_write (channel=0x7f96e0ae5800, buffer=0x7f96e09f9b40) at ../src/nvim/msgpack_rpc/channel.c:551
    0x000000000058567d in on_request_event (argv=0x7ffed792efa0) at ../src/nvim/msgpack_rpc/channel.c:523
    0x00000000005854c8 in handle_request (channel=0x7f96e0ae5800, request=0x7ffed792f1b8) at ../src/nvim/msgpack_rpc/channel.c:503
    0x00000000005850cb in parse_msgpack (channel=0x7f96e0ae5800) at ../src/nvim/msgpack_rpc/channel.c:423
    0x0000000000584f90 in receive_msgpack (stream=0x7f96e0a35218, rbuf=0x7f96e0d1d4c0, c=22, data=0x7f96e0ae5800, eof=false)
    at ../src/nvim/msgpack_rpc/channel.c:389
    0x00000000004d0b20 in read_event (argv=0x7ffed792f4a8) at ../src/nvim/event/rstream.c:190
    0x00000000004ce462 in multiqueue_process_events (this=0x7f96e18172d0) at ../src/nvim/event/multiqueue.c:150
    0x000000000059b630 in nv_event (cap=0x7ffed792f620) at ../src/nvim/normal.c:7908
    0x000000000058be69 in normal_execute (state=0x7ffed792f580, key=-25341) at ../src/nvim/normal.c:1137
    0x0000000000652463 in state_enter (s=0x7ffed792f580) at ../src/nvim/state.c:61
    0x000000000058a1fe in normal_enter (cmdwin=false, noexmode=false) at ../src/nvim/normal.c:467
    0x00000000005500c2 in main (argc=2, argv=0x7ffed792f8d8) at ../src/nvim/main.c:554

Alternative approach suggested by bfredl is to use close_cb of the
process. My unsuccessful attempt is below. (It seems close_cb is queued
too late, which is the similar problem addressed by this commit):

    commit 75fc12c6ab15711bdb7b18c6d42ec9d157f5145e
    Author: Justin M. Keyes <justinkz@gmail.com>
    Date:   Fri Aug 18 01:30:41 2017 +0200

        rpc: use Stream's close_cb instead of explicit check in receive_msgpack()

    diff --git a/src/nvim/event/process.c b/src/nvim/event/process.c
    index 8371d3cd482e..e52da23cdc40 100644
    --- a/src/nvim/event/process.c
    +++ b/src/nvim/event/process.c
    @@ -416,6 +416,10 @@ static void on_process_exit(Process *proc)
     static void on_process_stream_close(Stream *stream, void *data)
     {
       Process *proc = data;
    +  ILOG("on_process_stream_close");
    +  if (proc->stream_close_cb != NULL) {
    +    proc->stream_close_cb(stream, proc->stream_close_data);
    +  }
       decref(proc);
     }

    diff --git a/src/nvim/event/process.h b/src/nvim/event/process.h
    index 5c00e8e7ecd5..34a8d54f6f8c 100644
    --- a/src/nvim/event/process.h
    +++ b/src/nvim/event/process.h
    @@ -26,6 +26,11 @@ struct process {
       Stream *in, *out, *err;
       process_exit_cb cb;
       internal_process_cb internal_exit_cb, internal_close_cb;
    +
    +  // Called when any of the process streams (in/out/err) closes.
    +  stream_close_cb stream_close_cb;
    +  void *stream_close_data;
    +
       bool closed, detach;
       MultiQueue *events;
     };
    @@ -50,6 +55,8 @@ static inline Process process_init(Loop *loop, ProcessType type, void *data)
         .closed = false,
         .internal_close_cb = NULL,
         .internal_exit_cb = NULL,
    +    .stream_close_cb = NULL,
    +    .stream_close_data = NULL,
         .detach = false
       };
     }
    diff --git a/src/nvim/event/stream.c b/src/nvim/event/stream.c
    index 7c865bfe1e8c..c8720d1e45d9 100644
    --- a/src/nvim/event/stream.c
    +++ b/src/nvim/event/stream.c
    @@ -95,7 +95,11 @@ void stream_close(Stream *stream, stream_close_cb on_stream_close, void *data)
     void stream_close_handle(Stream *stream)
       FUNC_ATTR_NONNULL_ALL
     {
    +  ILOG("stream=%d", stream);
    +  // LOG_CALLSTACK();
       if (stream->uvstream) {
    +    // problem:  this schedules on the queue, but channel.c:receive_msgpack may
    +    // be processed before close_cb is called by libuv.
         uv_close((uv_handle_t *)stream->uvstream, close_cb);
       } else {
         uv_close((uv_handle_t *)&stream->uv.idle, close_cb);
    @@ -105,6 +109,7 @@ void stream_close_handle(Stream *stream)
     static void close_cb(uv_handle_t *handle)
     {
       Stream *stream = handle->data;
    +  ILOG(">>>>>>>>>>>>>>>>>>>>>>> stream=%p stream->internal_close_cb=%p", stream, stream->internal_close_cb);
       if (stream->buffer) {
         rbuffer_free(stream->buffer);
       }
    diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
    index 782eabe04e4a..dc2b794e366a 100644
    --- a/src/nvim/msgpack_rpc/channel.c
    +++ b/src/nvim/msgpack_rpc/channel.c
    @@ -128,6 +128,8 @@ uint64_t channel_from_process(Process *proc, uint64_t id, char *source)
                                           source);
       incref(channel);  // process channels are only closed by the exit_cb
       channel->data.proc = proc;
    +  channel->data.proc->stream_close_cb = close_cb2;
    +  channel->data.proc->stream_close_data = channel;

       wstream_init(proc->in, 0);
       rstream_init(proc->out, 0);
    @@ -387,17 +389,6 @@ static void receive_msgpack(Stream *stream, RBuffer *rbuf, size_t c,
         goto end;
       }

    -  if ((chan_wstream(channel) != NULL && chan_wstream(channel)->closed)
    -      || (chan_rstream(channel) != NULL && chan_rstream(channel)->closed)) {
    -    char buf[256];
    -    snprintf(buf, sizeof(buf),
    -             "ch %" PRIu64 ": stream closed unexpectedly. "
    -             "closing channel",
    -             channel->id);
    -    call_set_error(channel, buf, WARN_LOG_LEVEL);
    -    goto end;
    -  }
    -
       size_t count = rbuffer_size(rbuf);
       DLOG("ch %" PRIu64 ": parsing %u bytes from msgpack Stream: %p",
            channel->id, count, stream);
    @@ -571,23 +562,6 @@ static Stream *chan_wstream(Channel *chan)
       abort();
     }

    -/// Returns the Stream that a Channel reads from.
    -static Stream *chan_rstream(Channel *chan)
    -{
    -  switch (chan->type) {
    -    case kChannelTypeSocket:
    -      return &chan->data.stream;
    -    case kChannelTypeProc:
    -      return chan->data.proc->out;
    -    case kChannelTypeStdio:
    -      return &chan->data.std.in;
    -    case kChannelTypeInternal:
    -      return NULL;
    -  }
    -  abort();
    -}
    -
    -
     static bool channel_write(Channel *channel, WBuffer *buffer)
     {
       bool success = false;
    @@ -799,6 +773,12 @@ static void close_cb(Stream *stream, void *data)
       decref(data);
     }

    +static void close_cb2(Stream *stream, void *data)
    +{
    +  ILOG("close_cb2");
    +  close_channel(data);
    +}
    +
     /// @param source description of source function, rplugin name, TCP addr, etc
     static Channel *register_channel(ChannelType type, uint64_t id,
                                      MultiQueue *events, char *source)

5 files changed, 80 insertions, 23 deletions

diff --git a/src/nvim/log.c b/src/nvim/log.c
index 3baf0b2ebd..436a8a4079 100644
--- a/src/nvim/log.c
+++ b/src/nvim/log.c
@@ -250,7 +250,7 @@ static bool v_do_log_to_file(FILE *log_file, int log_level,
   static const char *log_levels[] = {
     [DEBUG_LOG_LEVEL]   = "DEBUG",
     [INFO_LOG_LEVEL]    = "INFO ",
-    [WARNING_LOG_LEVEL] = "WARN ",
+    [WARN_LOG_LEVEL]    = "WARN ",
     [ERROR_LOG_LEVEL]   = "ERROR",
   };
   assert(log_level >= DEBUG_LOG_LEVEL && log_level <= ERROR_LOG_LEVEL);
diff --git a/src/nvim/log.h b/src/nvim/log.h
index 5064d9333b..d63bcc366c 100644
--- a/src/nvim/log.h
+++ b/src/nvim/log.h
@@ -6,7 +6,7 @@
 
 #define DEBUG_LOG_LEVEL 0
 #define INFO_LOG_LEVEL 1
-#define WARNING_LOG_LEVEL 2
+#define WARN_LOG_LEVEL 2
 #define ERROR_LOG_LEVEL 3
 
 #define DLOG(...)
@@ -43,12 +43,12 @@
                            __VA_ARGS__)
 #endif
 
-#if MIN_LOG_LEVEL <= WARNING_LOG_LEVEL
+#if MIN_LOG_LEVEL <= WARN_LOG_LEVEL
 # undef WLOG
 # undef WLOGN
-# define WLOG(...) do_log(WARNING_LOG_LEVEL, __func__, __LINE__, true, \
+# define WLOG(...) do_log(WARN_LOG_LEVEL, __func__, __LINE__, true, \
                           __VA_ARGS__)
-# define WLOGN(...) do_log(WARNING_LOG_LEVEL, __func__, __LINE__, false, \
+# define WLOGN(...) do_log(WARN_LOG_LEVEL, __func__, __LINE__, false, \
                            __VA_ARGS__)
 #endif
 
diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 789cf1d4a8..ed97f298d9 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -62,7 +62,7 @@ typedef struct {
   ChannelType type;
   msgpack_unpacker *unpacker;
   union {
-    Stream stream;
+    Stream stream;  // bidirectional (socket)
     Process *proc;
     struct {
       Stream in;
@@ -154,7 +154,8 @@ void channel_from_connection(SocketWatcher *watcher)
   rstream_init(&channel->data.stream, CHANNEL_BUFFER_SIZE);
   rstream_start(&channel->data.stream, receive_msgpack, channel);
 
-  DLOG("ch %" PRIu64 " in/out-stream=%p", &channel->data.stream);
+  DLOG("ch %" PRIu64 " in/out-stream=%p", channel->id,
+       &channel->data.stream);
 }
 
 /// @param source description of source function, rplugin name, TCP addr, etc
@@ -383,7 +384,18 @@ static void receive_msgpack(Stream *stream, RBuffer *rbuf, size_t c,
     char buf[256];
     snprintf(buf, sizeof(buf), "ch %" PRIu64 " was closed by the client",
              channel->id);
-    call_set_error(channel, buf, WARNING_LOG_LEVEL);
+    call_set_error(channel, buf, WARN_LOG_LEVEL);
+    goto end;
+  }
+
+  if ((chan_wstream(channel) != NULL && chan_wstream(channel)->closed)
+      || (chan_rstream(channel) != NULL && chan_rstream(channel)->closed)) {
+    char buf[256];
+    snprintf(buf, sizeof(buf),
+             "ch %" PRIu64 ": stream closed unexpectedly. "
+             "closing channel",
+             channel->id);
+    call_set_error(channel, buf, WARN_LOG_LEVEL);
     goto end;
   }
 
@@ -445,8 +457,8 @@ static void parse_msgpack(Channel *channel)
     // causes for this error(search for 'goto _failed')
     //
     // A not so uncommon cause for this might be deserializing objects with
-    // a high nesting level: msgpack will break when it's internal parse stack
-    // size exceeds MSGPACK_EMBED_STACK_SIZE(defined as 32 by default)
+    // a high nesting level: msgpack will break when its internal parse stack
+    // size exceeds MSGPACK_EMBED_STACK_SIZE (defined as 32 by default)
     send_error(channel, 0, "Invalid msgpack payload. "
                            "This error can also happen when deserializing "
                            "an object with high level of nesting");
@@ -544,6 +556,39 @@ static void on_request_event(void **argv)
   api_clear_error(&error);
 }
 
+/// Returns the Stream that a Channel writes to.
+static Stream *chan_wstream(Channel *chan)
+{
+  switch (chan->type) {
+    case kChannelTypeSocket:
+      return &chan->data.stream;
+    case kChannelTypeProc:
+      return chan->data.proc->in;
+    case kChannelTypeStdio:
+      return &chan->data.std.out;
+    case kChannelTypeInternal:
+      return NULL;
+  }
+  abort();
+}
+
+/// Returns the Stream that a Channel reads from.
+static Stream *chan_rstream(Channel *chan)
+{
+  switch (chan->type) {
+    case kChannelTypeSocket:
+      return &chan->data.stream;
+    case kChannelTypeProc:
+      return chan->data.proc->out;
+    case kChannelTypeStdio:
+      return &chan->data.std.in;
+    case kChannelTypeInternal:
+      return NULL;
+  }
+  abort();
+}
+
+
 static bool channel_write(Channel *channel, WBuffer *buffer)
 {
   bool success = false;
@@ -555,13 +600,9 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
 
   switch (channel->type) {
     case kChannelTypeSocket:
-      success = wstream_write(&channel->data.stream, buffer);
-      break;
     case kChannelTypeProc:
-      success = wstream_write(channel->data.proc->in, buffer);
-      break;
     case kChannelTypeStdio:
-      success = wstream_write(&channel->data.std.out, buffer);
+      success = wstream_write(chan_wstream(channel), buffer);
       break;
     case kChannelTypeInternal:
       incref(channel);
diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 444c6cc256..98e56dee3b 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -361,7 +361,7 @@ typedef struct {
   size_t idx;
 } APIToMPObjectStackItem;
 
-/// Convert type used by Neovim API to msgpack
+/// Convert type used by Nvim API to msgpack type.
 ///
 /// @param[in]  result  Object to convert.
 /// @param[out]  res  Structure that defines where conversion results are saved.
diff --git a/test/functional/api/server_requests_spec.lua b/test/functional/api/server_requests_spec.lua
index 6a32f979ea..9f245d913b 100644
--- a/test/functional/api/server_requests_spec.lua
+++ b/test/functional/api/server_requests_spec.lua
@@ -20,6 +20,22 @@ describe('server -> client', function()
     cid = nvim('get_api_info')[1]
   end)
 
+  it('handles unexpected closed stream while preparing RPC response', function()
+    source([[
+      let g:_nvim_args = [v:progpath, '--embed', '-n', '-u', 'NONE', '-i', 'NONE', ]
+      let ch1 = jobstart(g:_nvim_args, {'rpc': v:true})
+      let child1_ch = rpcrequest(ch1, "nvim_get_api_info")[0]
+      call rpcnotify(ch1, 'nvim_eval', 'rpcrequest('.child1_ch.', "nvim_get_api_info")')
+
+      let ch2 = jobstart(g:_nvim_args, {'rpc': v:true})
+      let child2_ch = rpcrequest(ch2, "nvim_get_api_info")[0]
+      call rpcnotify(ch2, 'nvim_eval', 'rpcrequest('.child2_ch.', "nvim_get_api_info")')
+
+      call jobstop(ch1)
+    ]])
+    eq(2, eval("1+1"))  -- Still alive?
+  end)
+
   describe('simple call', function()
     it('works', function()
       local function on_setup()
@@ -141,7 +157,7 @@ describe('server -> client', function()
     end)
   end)
 
-  describe('when the client is a recursive vim instance', function()
+  describe('recursive (child) nvim client', function()
     if os.getenv("TRAVIS") and helpers.os_name() == "osx" then
       -- XXX: Hangs Travis macOS since e9061117a5b8f195c3f26a5cb94e18ddd7752d86.
       pending("[Hangs on Travis macOS. #5002]", function() end)
@@ -155,7 +171,7 @@ describe('server -> client', function()
 
     after_each(function() command('call rpcstop(vim)') end)
 
-    it('can send/recieve notifications and make requests', function()
+    it('can send/receive notifications and make requests', function()
       nvim('command', "call rpcnotify(vim, 'vim_set_current_line', 'SOME TEXT')")
 
       -- Wait for the notification to complete.
@@ -188,7 +204,7 @@ describe('server -> client', function()
     end)
   end)
 
-  describe('when using jobstart', function()
+  describe('jobstart()', function()
     local jobid
     before_each(function()
       local channel = nvim('get_api_info')[1]
@@ -227,7 +243,7 @@ describe('server -> client', function()
     end)
   end)
 
-  describe('when connecting to another nvim instance', function()
+  describe('connecting to another (peer) nvim', function()
     local function connect_test(server, mode, address)
       local serverpid = funcs.getpid()
       local client = spawn(nvim_argv)
@@ -256,7 +272,7 @@ describe('server -> client', function()
       client:close()
     end
 
-    it('over a named pipe', function()
+    it('via named pipe', function()
       local server = spawn(nvim_argv)
       set_session(server)
       local address = funcs.serverlist()[1]
@@ -265,7 +281,7 @@ describe('server -> client', function()
       connect_test(server, 'pipe', address)
     end)
 
-    it('to an ip adress', function()
+    it('via ip address', function()
       local server = spawn(nvim_argv)
       set_session(server)
       local address = funcs.serverstart("127.0.0.1:")
@@ -273,7 +289,7 @@ describe('server -> client', function()
       connect_test(server, 'tcp', address)
     end)
 
-    it('to a hostname', function()
+    it('via hostname', function()
       local server = spawn(nvim_argv)
       set_session(server)
       local address = funcs.serverstart("localhost:")