From ba2f615cd40d5d809d1a141c7b098e3bd22ff7bb Mon Sep 17 00:00:00 2001
From: ZyX <kp-pav@yandex.ru>
Date: Sat, 16 Jul 2016 02:26:04 +0300
Subject: functests: Test for error conditions
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

During testing found the following bugs:

1. msgpack-gen.lua script is completely unprepared for Float values either in
   return type or in arguments. Specifically:

   1. At the time of writing relevant code FLOAT_OBJ did not exist as well as
      FLOATING_OBJ, but it would be used by msgpack-gen.lua should return type
      be Float. I added FLOATING_OBJ macros later because did not know that
      msgpack-gen.lua uses these _OBJ macros, otherwise it would be FLOAT_OBJ.
   2. msgpack-gen.lua should use .data.floating in place of .data.float. But it
      did not expect that .data subattribute may have name different from
      lowercased type name.

2. vim_replace_termcodes returned its argument as-is if it receives an empty
   string (as well as _vim_id*() functions did). But if something in returned
   argument lives in an allocated memory such action will cause double free:
   once when freeing arguments, then when freeing return value. It did not cause
   problems yet because msgpack bindings return empty string as {NULL, 0} and
   nothing was actually allocated.
3. New code in msgpack-gen.lua popped arguments in reversed order, making lua
   bindings’ signatures be different from API ones.
---
 src/nvim/msgpack_rpc/helpers.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 5137b375f0..64a018f5c3 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -117,7 +117,7 @@ bool msgpack_rpc_to_object(const msgpack_object *const obj, Object *const arg)
       case MSGPACK_OBJECT_FLOAT: {
         STATIC_ASSERT(sizeof(Float) == sizeof(cur.mobj->via.f64),
                       "Msgpack floating-point size does not match API integer");
-        *cur.aobj = FLOATING_OBJ(cur.mobj->via.f64);
+        *cur.aobj = FLOAT_OBJ(cur.mobj->via.f64);
         break;
       }
 #define STR_CASE(type, attr, obj, dest, conv) \
-- 
cgit 


From ca4c8b7f8a7b9543ef01157cb5ab94783f624ac6 Mon Sep 17 00:00:00 2001
From: ZyX <kp-pav@yandex.ru>
Date: Sun, 22 Jan 2017 04:55:26 +0300
Subject: api: Allow kObjectTypeNil to be zero without breaking compatibility

---
 src/nvim/msgpack_rpc/helpers.c | 26 +++++++++++++++++++++++---
 1 file changed, 23 insertions(+), 3 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 64a018f5c3..972d5b3441 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -20,13 +20,20 @@
 static msgpack_zone zone;
 static msgpack_sbuffer sbuffer;
 
+/// Value by which objects represented as EXT type are shifted
+///
+/// Subtracted when packing, added when unpacking. Used to allow moving
+/// buffer/window/tabpage block inside ObjectType enum. This block yet cannot be
+/// split or reordered.
+#define EXT_OBJECT_TYPE_SHIFT kObjectTypeBuffer
+
 #define HANDLE_TYPE_CONVERSION_IMPL(t, lt) \
   bool msgpack_rpc_to_##lt(const msgpack_object *const obj, \
                            Integer *const arg) \
     FUNC_ATTR_NONNULL_ALL \
   { \
     if (obj->type != MSGPACK_OBJECT_EXT \
-        || obj->via.ext.type != kObjectType##t) { \
+        || obj->via.ext.type + EXT_OBJECT_TYPE_SHIFT != kObjectType##t) { \
       return false; \
     } \
     \
@@ -51,7 +58,8 @@ static msgpack_sbuffer sbuffer;
     msgpack_packer pac; \
     msgpack_packer_init(&pac, &sbuffer, msgpack_sbuffer_write); \
     msgpack_pack_int64(&pac, (handle_T)o); \
-    msgpack_pack_ext(res, sbuffer.size, kObjectType##t); \
+    msgpack_pack_ext(res, sbuffer.size, \
+                     kObjectType##t - EXT_OBJECT_TYPE_SHIFT); \
     msgpack_pack_ext_body(res, sbuffer.data, sbuffer.size); \
     msgpack_sbuffer_clear(&sbuffer); \
   }
@@ -211,7 +219,7 @@ bool msgpack_rpc_to_object(const msgpack_object *const obj, Object *const arg)
         break;
       }
       case MSGPACK_OBJECT_EXT: {
-        switch (cur.mobj->via.ext.type) {
+        switch ((ObjectType)(cur.mobj->via.ext.type + EXT_OBJECT_TYPE_SHIFT)) {
           case kObjectTypeBuffer: {
             cur.aobj->type = kObjectTypeBuffer;
             ret = msgpack_rpc_to_buffer(cur.mobj, &cur.aobj->data.integer);
@@ -227,6 +235,15 @@ bool msgpack_rpc_to_object(const msgpack_object *const obj, Object *const arg)
             ret = msgpack_rpc_to_tabpage(cur.mobj, &cur.aobj->data.integer);
             break;
           }
+          case kObjectTypeNil:
+          case kObjectTypeBoolean:
+          case kObjectTypeInteger:
+          case kObjectTypeFloat:
+          case kObjectTypeString:
+          case kObjectTypeArray:
+          case kObjectTypeDictionary: {
+            break;
+          }
         }
         break;
       }
@@ -350,6 +367,9 @@ void msgpack_rpc_from_object(const Object result, msgpack_packer *const res)
   kv_push(stack, ((APIToMPObjectStackItem) { &result, false, 0 }));
   while (kv_size(stack)) {
     APIToMPObjectStackItem cur = kv_last(stack);
+    STATIC_ASSERT(kObjectTypeWindow == kObjectTypeBuffer + 1
+                  && kObjectTypeTabpage == kObjectTypeWindow + 1,
+                  "Buffer, window and tabpage enum items are in order");
     switch (cur.aobj->type) {
       case kObjectTypeNil: {
         msgpack_pack_nil(res);
-- 
cgit 


From 22d3ce9c29c163470dbb652199acbb7c82e3dca1 Mon Sep 17 00:00:00 2001
From: ZyX <kp-pav@yandex.ru>
Date: Sun, 22 Jan 2017 05:35:48 +0300
Subject: msgpack_rpc: Fix #HANDLE_TYPE_CONVERSION_IMPL

Function declarations generator is able to handle properly only the *first*
function definition that is in macros, and only if it is the first entity in the
macros. So msgpack_rpc_from_* was already really a static function, additionally
its attributes were useless. This commit switches to explicit declarations and
makes generated functions static.
---
 src/nvim/msgpack_rpc/helpers.c | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 972d5b3441..94a4f5ce3e 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -28,9 +28,11 @@ static msgpack_sbuffer sbuffer;
 #define EXT_OBJECT_TYPE_SHIFT kObjectTypeBuffer
 
 #define HANDLE_TYPE_CONVERSION_IMPL(t, lt) \
-  bool msgpack_rpc_to_##lt(const msgpack_object *const obj, \
-                           Integer *const arg) \
-    FUNC_ATTR_NONNULL_ALL \
+  static bool msgpack_rpc_to_##lt(const msgpack_object *const obj, \
+                                  Integer *const arg) \
+    REAL_FATTR_NONNULL_ALL REAL_FATTR_WARN_UNUSED_RESULT; \
+  static bool msgpack_rpc_to_##lt(const msgpack_object *const obj, \
+                                  Integer *const arg) \
   { \
     if (obj->type != MSGPACK_OBJECT_EXT \
         || obj->via.ext.type + EXT_OBJECT_TYPE_SHIFT != kObjectType##t) { \
@@ -52,8 +54,9 @@ static msgpack_sbuffer sbuffer;
     return true; \
   } \
   \
-  void msgpack_rpc_from_##lt(Integer o, msgpack_packer *res) \
-    FUNC_ATTR_NONNULL_ARG(2) \
+  static void msgpack_rpc_from_##lt(Integer o, msgpack_packer *res) \
+    REAL_FATTR_NONNULL_ARG(2); \
+  static void msgpack_rpc_from_##lt(Integer o, msgpack_packer *res) \
   { \
     msgpack_packer pac; \
     msgpack_packer_init(&pac, &sbuffer, msgpack_sbuffer_write); \
-- 
cgit 


From ae4adcc70735a89bffb110bcf9d5a993b0786c4d Mon Sep 17 00:00:00 2001
From: ZyX <kp-pav@yandex.ru>
Date: Sat, 28 Jan 2017 03:47:15 +0300
Subject: gendeclarations: Make declarations generator work with macros funcs

Now it checks functions also after every semicolon and closing figure brace,
possibly preceded by whitespaces (tabs and spaces). This should make messing
with declarations in macros not needed.
---
 src/nvim/msgpack_rpc/helpers.c | 7 ++-----
 1 file changed, 2 insertions(+), 5 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 94a4f5ce3e..ec35d56978 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -30,9 +30,7 @@ static msgpack_sbuffer sbuffer;
 #define HANDLE_TYPE_CONVERSION_IMPL(t, lt) \
   static bool msgpack_rpc_to_##lt(const msgpack_object *const obj, \
                                   Integer *const arg) \
-    REAL_FATTR_NONNULL_ALL REAL_FATTR_WARN_UNUSED_RESULT; \
-  static bool msgpack_rpc_to_##lt(const msgpack_object *const obj, \
-                                  Integer *const arg) \
+    FUNC_ATTR_NONNULL_ALL FUNC_ATTR_WARN_UNUSED_RESULT \
   { \
     if (obj->type != MSGPACK_OBJECT_EXT \
         || obj->via.ext.type + EXT_OBJECT_TYPE_SHIFT != kObjectType##t) { \
@@ -55,8 +53,7 @@ static msgpack_sbuffer sbuffer;
   } \
   \
   static void msgpack_rpc_from_##lt(Integer o, msgpack_packer *res) \
-    REAL_FATTR_NONNULL_ARG(2); \
-  static void msgpack_rpc_from_##lt(Integer o, msgpack_packer *res) \
+    FUNC_ATTR_NONNULL_ARG(2) \
   { \
     msgpack_packer pac; \
     msgpack_packer_init(&pac, &sbuffer, msgpack_sbuffer_write); \
-- 
cgit 


From 62fde319360e27a86c0deba0053ff230f80ca772 Mon Sep 17 00:00:00 2001
From: ZyX <kp-pav@yandex.ru>
Date: Sun, 29 Jan 2017 01:29:51 +0300
Subject: api: Also shift numbers in api_metadata output
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Fixes problem introduced by “api: Allow kObjectTypeNil to be zero without
breaking compatibility”: apparently there are clients which use metadata and
there are which aren’t. For the first that commit would not be needed, for the
second that commit misses this critical piece.
---
 src/nvim/msgpack_rpc/helpers.c | 7 -------
 src/nvim/msgpack_rpc/helpers.h | 7 +++++++
 2 files changed, 7 insertions(+), 7 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index ec35d56978..21acd6a394 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -20,13 +20,6 @@
 static msgpack_zone zone;
 static msgpack_sbuffer sbuffer;
 
-/// Value by which objects represented as EXT type are shifted
-///
-/// Subtracted when packing, added when unpacking. Used to allow moving
-/// buffer/window/tabpage block inside ObjectType enum. This block yet cannot be
-/// split or reordered.
-#define EXT_OBJECT_TYPE_SHIFT kObjectTypeBuffer
-
 #define HANDLE_TYPE_CONVERSION_IMPL(t, lt) \
   static bool msgpack_rpc_to_##lt(const msgpack_object *const obj, \
                                   Integer *const arg) \
diff --git a/src/nvim/msgpack_rpc/helpers.h b/src/nvim/msgpack_rpc/helpers.h
index 7d9f114140..0e4cd1be6d 100644
--- a/src/nvim/msgpack_rpc/helpers.h
+++ b/src/nvim/msgpack_rpc/helpers.h
@@ -9,6 +9,13 @@
 #include "nvim/event/wstream.h"
 #include "nvim/api/private/defs.h"
 
+/// Value by which objects represented as EXT type are shifted
+///
+/// Subtracted when packing, added when unpacking. Used to allow moving
+/// buffer/window/tabpage block inside ObjectType enum. This block yet cannot be
+/// split or reordered.
+#define EXT_OBJECT_TYPE_SHIFT kObjectTypeBuffer
+
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "msgpack_rpc/helpers.h.generated.h"
 #endif
-- 
cgit 


From 28dafe3ff0b0dc082fb62b2251fd64a167ce7188 Mon Sep 17 00:00:00 2001
From: ZyX <kp-pav@yandex.ru>
Date: Sun, 21 Aug 2016 08:16:47 +0300
Subject: eval,*: Move get_tv_string to typval.c

Function was renamed and changed to return `const char *`.
---
 src/nvim/msgpack_rpc/channel.c | 20 ++++++++++----------
 src/nvim/msgpack_rpc/helpers.c |  4 ++--
 2 files changed, 12 insertions(+), 12 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index de6167c7fc..259dcc523c 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -147,7 +147,7 @@ void channel_from_connection(SocketWatcher *watcher)
 /// @param name The event name, an arbitrary string
 /// @param args Array with event arguments
 /// @return True if the event was sent successfully, false otherwise.
-bool channel_send_event(uint64_t id, char *name, Array args)
+bool channel_send_event(uint64_t id, const char *name, Array args)
 {
   Channel *channel = NULL;
 
@@ -160,7 +160,7 @@ bool channel_send_event(uint64_t id, char *name, Array args)
   if (channel) {
     if (channel->pending_requests) {
       // Pending request, queue the notification for later sending.
-      String method = cstr_as_string(name);
+      const String method = cstr_as_string((char *)name);
       WBuffer *buffer = serialize_request(id, 0, method, args, &out_buffer, 1);
       kv_push(channel->delayed_notifications, buffer);
     } else {
@@ -182,7 +182,7 @@ bool channel_send_event(uint64_t id, char *name, Array args)
 /// @param[out] error True if the return value is an error
 /// @return Whatever the remote method returned
 Object channel_send_call(uint64_t id,
-                         char *method_name,
+                         const char *method_name,
                          Array args,
                          Error *err)
 {
@@ -519,10 +519,10 @@ static void send_error(Channel *channel, uint64_t id, char *err)
 
 static void send_request(Channel *channel,
                          uint64_t id,
-                         char *name,
+                         const char *name,
                          Array args)
 {
-  String method = {.size = strlen(name), .data = name};
+  const String method = cstr_as_string((char *)name);
   channel_write(channel, serialize_request(channel->id,
                                            id,
                                            method,
@@ -532,10 +532,10 @@ static void send_request(Channel *channel,
 }
 
 static void send_event(Channel *channel,
-                       char *name,
+                       const char *name,
                        Array args)
 {
-  String method = {.size = strlen(name), .data = name};
+  const String method = cstr_as_string((char *)name);
   channel_write(channel, serialize_request(channel->id,
                                            0,
                                            method,
@@ -544,7 +544,7 @@ static void send_event(Channel *channel,
                                            1));
 }
 
-static void broadcast_event(char *name, Array args)
+static void broadcast_event(const char *name, Array args)
 {
   kvec_t(Channel *) subscribed = KV_INITIAL_VALUE;
   Channel *channel;
@@ -560,7 +560,7 @@ static void broadcast_event(char *name, Array args)
     goto end;
   }
 
-  String method = {.size = strlen(name), .data = name};
+  const String method = cstr_as_string((char *)name);
   WBuffer *buffer = serialize_request(0,
                                       0,
                                       method,
@@ -728,7 +728,7 @@ static void call_set_error(Channel *channel, char *msg)
 
 static WBuffer *serialize_request(uint64_t channel_id,
                                   uint64_t request_id,
-                                  String method,
+                                  const String method,
                                   Array args,
                                   msgpack_sbuffer *sbuffer,
                                   size_t refcount)
diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 5137b375f0..808bb863fd 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -322,7 +322,7 @@ void msgpack_rpc_from_float(Float result, msgpack_packer *res)
   msgpack_pack_double(res, result);
 }
 
-void msgpack_rpc_from_string(String result, msgpack_packer *res)
+void msgpack_rpc_from_string(const String result, msgpack_packer *res)
   FUNC_ATTR_NONNULL_ARG(2)
 {
   msgpack_pack_str(res, result.size);
@@ -478,7 +478,7 @@ Object msgpack_rpc_handle_invalid_arguments(uint64_t channel_id,
 
 /// Serializes a msgpack-rpc request or notification(id == 0)
 void msgpack_rpc_serialize_request(uint64_t request_id,
-                                   String method,
+                                   const String method,
                                    Array args,
                                    msgpack_packer *pac)
   FUNC_ATTR_NONNULL_ARG(4)
-- 
cgit 


From f4a3a96b6852f2eb5cf68d26b2bf58123c39c602 Mon Sep 17 00:00:00 2001
From: James McCoy <jamessan@jamessan.com>
Date: Thu, 30 Mar 2017 11:06:26 -0400
Subject: Add handling for MSGPACK_OBJECT_FLOAT{32,64}

msgpack-c previously only had MSGPACK_OBJECT_FLOAT, which was a 64-bit
value.  Now, 32-bit and 64-bit floats are supported as distinct types,
but we'll simply continue to treat everything as 64-bit types.
---
 src/nvim/msgpack_rpc/helpers.c | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 808bb863fd..4d8a9984e1 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -114,7 +114,13 @@ bool msgpack_rpc_to_object(const msgpack_object *const obj, Object *const arg)
         }
         break;
       }
-      case MSGPACK_OBJECT_FLOAT: {
+#ifdef NVIM_MSGPACK_HAS_FLOAT32
+      case MSGPACK_OBJECT_FLOAT32:
+      case MSGPACK_OBJECT_FLOAT64:
+#else
+      case MSGPACK_OBJECT_FLOAT:
+#endif
+      {
         STATIC_ASSERT(sizeof(Float) == sizeof(cur.mobj->via.f64),
                       "Msgpack floating-point size does not match API integer");
         *cur.aobj = FLOATING_OBJ(cur.mobj->via.f64);
@@ -181,7 +187,12 @@ bool msgpack_rpc_to_object(const msgpack_object *const obj, Object *const arg)
               case MSGPACK_OBJECT_BOOLEAN:
               case MSGPACK_OBJECT_POSITIVE_INTEGER:
               case MSGPACK_OBJECT_NEGATIVE_INTEGER:
+#ifdef NVIM_MSGPACK_HAS_FLOAT32
+              case MSGPACK_OBJECT_FLOAT32:
+              case MSGPACK_OBJECT_FLOAT64:
+#else
               case MSGPACK_OBJECT_FLOAT:
+#endif
               case MSGPACK_OBJECT_EXT:
               case MSGPACK_OBJECT_MAP:
               case MSGPACK_OBJECT_ARRAY: {
-- 
cgit 


From c2f3e361c52ec4e7149ea1d8c6a1202e0873da8e Mon Sep 17 00:00:00 2001
From: ZyX <kp-pav@yandex.ru>
Date: Wed, 19 Apr 2017 19:11:50 +0300
Subject: *: Add comment to all C files

---
 src/nvim/msgpack_rpc/channel.c | 3 +++
 src/nvim/msgpack_rpc/helpers.c | 3 +++
 src/nvim/msgpack_rpc/server.c  | 3 +++
 3 files changed, 9 insertions(+)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 259dcc523c..b2c6ef852c 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -1,3 +1,6 @@
+// This is an open source non-commercial project. Dear PVS-Studio, please check
+// it. PVS-Studio Static Code Analyzer for C, C++ and C#: http://www.viva64.com
+
 #include <stdbool.h>
 #include <string.h>
 #include <inttypes.h>
diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 4d8a9984e1..967ce31c1b 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -1,3 +1,6 @@
+// This is an open source non-commercial project. Dear PVS-Studio, please check
+// it. PVS-Studio Static Code Analyzer for C, C++ and C#: http://www.viva64.com
+
 #include <stdint.h>
 #include <stdbool.h>
 #include <inttypes.h>
diff --git a/src/nvim/msgpack_rpc/server.c b/src/nvim/msgpack_rpc/server.c
index d7c2926a0f..b6958088ca 100644
--- a/src/nvim/msgpack_rpc/server.c
+++ b/src/nvim/msgpack_rpc/server.c
@@ -1,3 +1,6 @@
+// This is an open source non-commercial project. Dear PVS-Studio, please check
+// it. PVS-Studio Static Code Analyzer for C, C++ and C#: http://www.viva64.com
+
 #include <assert.h>
 #include <stdlib.h>
 #include <string.h>
-- 
cgit 


From 5c9860a0a2bf27d409c986673f0a74542561c4c3 Mon Sep 17 00:00:00 2001
From: Sander Bosma <sanderbosma@gmail.com>
Date: Wed, 1 Mar 2017 10:43:47 +0100
Subject: api: Do not truncate errors <1 MB. #6237

Closes #5984
---
 src/nvim/msgpack_rpc/channel.c | 73 +++++++++++++++++++++---------------------
 src/nvim/msgpack_rpc/helpers.c | 22 ++++++-------
 2 files changed, 47 insertions(+), 48 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index b2c6ef852c..06cdf7d1f1 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -192,7 +192,7 @@ Object channel_send_call(uint64_t id,
   Channel *channel = NULL;
 
   if (!(channel = pmap_get(uint64_t)(channels, id)) || channel->closed) {
-    api_set_error(err, Exception, _("Invalid channel \"%" PRIu64 "\""), id);
+    _api_set_error(err, kErrorTypeException, _("Invalid channel \"%" PRIu64 "\""), id);
     api_free_array(args);
     return NIL;
   }
@@ -212,7 +212,7 @@ Object channel_send_call(uint64_t id,
 
   if (frame.errored) {
     if (frame.result.type == kObjectTypeString) {
-      api_set_error(err, Exception, "%s", frame.result.data.string.data);
+      _api_set_error(err, kErrorTypeException, "%s", frame.result.data.string.data);
     } else if (frame.result.type == kObjectTypeArray) {
       // Should be an error in the form [type, message]
       Array array = frame.result.data.array;
@@ -220,14 +220,13 @@ Object channel_send_call(uint64_t id,
           && (array.items[0].data.integer == kErrorTypeException
               || array.items[0].data.integer == kErrorTypeValidation)
           && array.items[1].type == kObjectTypeString) {
-        err->type = (ErrorType) array.items[0].data.integer;
-        xstrlcpy(err->msg, array.items[1].data.string.data, sizeof(err->msg));
-        err->set = true;
+        _api_set_error(err, (ErrorType)array.items[0].data.integer, "%s",
+                       array.items[1].data.string.data);
       } else {
-        api_set_error(err, Exception, "%s", "unknown error");
+        _api_set_error(err, kErrorTypeException, "%s", "unknown error");
       }
     } else {
-      api_set_error(err, Exception, "%s", "unknown error");
+      _api_set_error(err, kErrorTypeException, "%s", "unknown error");
     }
 
     api_free_object(frame.result);
@@ -412,38 +411,38 @@ static void handle_request(Channel *channel, msgpack_object *request)
                channel->id);
       call_set_error(channel, buf);
     }
-    return;
-  }
-
-  // Retrieve the request handler
-  MsgpackRpcRequestHandler handler;
-  msgpack_object *method = msgpack_rpc_method(request);
-
-  if (method) {
-    handler = msgpack_rpc_get_handler_for(method->via.bin.ptr,
-                                          method->via.bin.size);
   } else {
-    handler.fn = msgpack_rpc_handle_missing_method;
-    handler.async = true;
-  }
+    // Retrieve the request handler
+    MsgpackRpcRequestHandler handler;
+    msgpack_object *method = msgpack_rpc_method(request);
 
-  Array args = ARRAY_DICT_INIT;
-  if (!msgpack_rpc_to_array(msgpack_rpc_args(request), &args)) {
-    handler.fn = msgpack_rpc_handle_invalid_arguments;
-    handler.async = true;
-  }
+    if (method) {
+      handler = msgpack_rpc_get_handler_for(method->via.bin.ptr,
+                                            method->via.bin.size);
+    } else {
+      handler.fn = msgpack_rpc_handle_missing_method;
+      handler.async = true;
+    }
 
-  RequestEvent *event_data = xmalloc(sizeof(RequestEvent));
-  event_data->channel = channel;
-  event_data->handler = handler;
-  event_data->args = args;
-  event_data->request_id = request_id;
-  incref(channel);
-  if (handler.async) {
-    on_request_event((void **)&event_data);
-  } else {
-    multiqueue_put(channel->events, on_request_event, 1, event_data);
+    Array args = ARRAY_DICT_INIT;
+    if (!msgpack_rpc_to_array(msgpack_rpc_args(request), &args)) {
+      handler.fn = msgpack_rpc_handle_invalid_arguments;
+      handler.async = true;
+    }
+
+    RequestEvent *event_data = xmalloc(sizeof(RequestEvent));
+    event_data->channel = channel;
+    event_data->handler = handler;
+    event_data->args = args;
+    event_data->request_id = request_id;
+    incref(channel);
+    if (handler.async) {
+      on_request_event((void **)&event_data);
+    } else {
+      multiqueue_put(channel->events, on_request_event, 1, event_data);
+    }
   }
+  xfree(error.msg);
 }
 
 static void on_request_event(void **argv)
@@ -470,6 +469,7 @@ static void on_request_event(void **argv)
   api_free_array(args);
   decref(channel);
   xfree(e);
+  xfree(error.msg);
 }
 
 static bool channel_write(Channel *channel, WBuffer *buffer)
@@ -512,12 +512,13 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
 static void send_error(Channel *channel, uint64_t id, char *err)
 {
   Error e = ERROR_INIT;
-  api_set_error(&e, Exception, "%s", err);
+  _api_set_error(&e, kErrorTypeException, "%s", err);
   channel_write(channel, serialize_response(channel->id,
                                             id,
                                             &e,
                                             NIL,
                                             &out_buffer));
+  xfree(e.msg);
 }
 
 static void send_request(Channel *channel,
diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 967ce31c1b..c273343b41 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -475,8 +475,7 @@ Object msgpack_rpc_handle_missing_method(uint64_t channel_id,
                                          Array args,
                                          Error *error)
 {
-  snprintf(error->msg, sizeof(error->msg), "Invalid method name");
-  error->set = true;
+  _api_set_error(error, error->type, "Invalid method name");
   return NIL;
 }
 
@@ -485,8 +484,7 @@ Object msgpack_rpc_handle_invalid_arguments(uint64_t channel_id,
                                             Array args,
                                             Error *error)
 {
-  snprintf(error->msg, sizeof(error->msg), "Invalid method arguments");
-  error->set = true;
+  _api_set_error(error, error->type, "Invalid method arguments");
   return NIL;
 }
 
@@ -572,29 +570,29 @@ void msgpack_rpc_validate(uint64_t *response_id,
   *response_id = NO_RESPONSE;
   // Validate the basic structure of the msgpack-rpc payload
   if (req->type != MSGPACK_OBJECT_ARRAY) {
-    api_set_error(err, Validation, _("Message is not an array"));
+    _api_set_error(err, kErrorTypeValidation, _("Message is not an array"));
     return;
   }
 
   if (req->via.array.size == 0) {
-    api_set_error(err, Validation, _("Message is empty"));
+    _api_set_error(err, kErrorTypeValidation, _("Message is empty"));
     return;
   }
 
   if (req->via.array.ptr[0].type != MSGPACK_OBJECT_POSITIVE_INTEGER) {
-    api_set_error(err, Validation, _("Message type must be an integer"));
+    _api_set_error(err, kErrorTypeValidation, _("Message type must be an integer"));
     return;
   }
 
   uint64_t type = req->via.array.ptr[0].via.u64;
   if (type != kMessageTypeRequest && type != kMessageTypeNotification) {
-    api_set_error(err, Validation, _("Unknown message type"));
+    _api_set_error(err, kErrorTypeValidation, _("Unknown message type"));
     return;
   }
 
   if ((type == kMessageTypeRequest && req->via.array.size != 4)
       || (type == kMessageTypeNotification && req->via.array.size != 3)) {
-    api_set_error(err, Validation, _("Request array size should be 4 (request) "
+    _api_set_error(err, kErrorTypeValidation, _("Request array size should be 4 (request) "
                                      "or 3 (notification)"));
     return;
   }
@@ -602,19 +600,19 @@ void msgpack_rpc_validate(uint64_t *response_id,
   if (type == kMessageTypeRequest) {
     msgpack_object *id_obj = msgpack_rpc_msg_id(req);
     if (!id_obj) {
-      api_set_error(err, Validation, _("ID must be a positive integer"));
+      _api_set_error(err, kErrorTypeValidation, _("ID must be a positive integer"));
       return;
     }
     *response_id = id_obj->via.u64;
   }
 
   if (!msgpack_rpc_method(req)) {
-    api_set_error(err, Validation, _("Method must be a string"));
+    _api_set_error(err, kErrorTypeValidation, _("Method must be a string"));
     return;
   }
 
   if (!msgpack_rpc_args(req)) {
-    api_set_error(err, Validation, _("Parameters must be an array"));
+    _api_set_error(err, kErrorTypeValidation, _("Parameters must be an array"));
     return;
   }
 }
-- 
cgit 


From 2a49163103827465f25810f5f4e3d4305159f209 Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Sun, 23 Apr 2017 15:59:59 +0200
Subject: api_clear_error()

---
 src/nvim/msgpack_rpc/channel.c | 60 +++++++++++++++++++++---------------------
 1 file changed, 30 insertions(+), 30 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 06cdf7d1f1..bd7f84fed9 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -411,38 +411,38 @@ static void handle_request(Channel *channel, msgpack_object *request)
                channel->id);
       call_set_error(channel, buf);
     }
-  } else {
-    // Retrieve the request handler
-    MsgpackRpcRequestHandler handler;
-    msgpack_object *method = msgpack_rpc_method(request);
+    api_clear_error(&error);
+    return;
+  }
+  // Retrieve the request handler
+  MsgpackRpcRequestHandler handler;
+  msgpack_object *method = msgpack_rpc_method(request);
 
-    if (method) {
-      handler = msgpack_rpc_get_handler_for(method->via.bin.ptr,
-                                            method->via.bin.size);
-    } else {
-      handler.fn = msgpack_rpc_handle_missing_method;
-      handler.async = true;
-    }
+  if (method) {
+    handler = msgpack_rpc_get_handler_for(method->via.bin.ptr,
+                                          method->via.bin.size);
+  } else {
+    handler.fn = msgpack_rpc_handle_missing_method;
+    handler.async = true;
+  }
 
-    Array args = ARRAY_DICT_INIT;
-    if (!msgpack_rpc_to_array(msgpack_rpc_args(request), &args)) {
-      handler.fn = msgpack_rpc_handle_invalid_arguments;
-      handler.async = true;
-    }
+  Array args = ARRAY_DICT_INIT;
+  if (!msgpack_rpc_to_array(msgpack_rpc_args(request), &args)) {
+    handler.fn = msgpack_rpc_handle_invalid_arguments;
+    handler.async = true;
+  }
 
-    RequestEvent *event_data = xmalloc(sizeof(RequestEvent));
-    event_data->channel = channel;
-    event_data->handler = handler;
-    event_data->args = args;
-    event_data->request_id = request_id;
-    incref(channel);
-    if (handler.async) {
-      on_request_event((void **)&event_data);
-    } else {
-      multiqueue_put(channel->events, on_request_event, 1, event_data);
-    }
+  RequestEvent *event_data = xmalloc(sizeof(RequestEvent));
+  event_data->channel = channel;
+  event_data->handler = handler;
+  event_data->args = args;
+  event_data->request_id = request_id;
+  incref(channel);
+  if (handler.async) {
+    on_request_event((void **)&event_data);
+  } else {
+    multiqueue_put(channel->events, on_request_event, 1, event_data);
   }
-  xfree(error.msg);
 }
 
 static void on_request_event(void **argv)
@@ -469,7 +469,7 @@ static void on_request_event(void **argv)
   api_free_array(args);
   decref(channel);
   xfree(e);
-  xfree(error.msg);
+  api_clear_error(&error);
 }
 
 static bool channel_write(Channel *channel, WBuffer *buffer)
@@ -518,7 +518,7 @@ static void send_error(Channel *channel, uint64_t id, char *err)
                                             &e,
                                             NIL,
                                             &out_buffer));
-  xfree(e.msg);
+  api_clear_error(&e);
 }
 
 static void send_request(Channel *channel,
-- 
cgit 


From 2ed91f222f1dddda10fbdc9cb80df2be7d4c2da3 Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Sun, 23 Apr 2017 19:58:13 +0200
Subject: api/internal: Remove `set` field from Error type.

---
 src/nvim/msgpack_rpc/channel.c | 2 +-
 src/nvim/msgpack_rpc/helpers.c | 6 +++---
 2 files changed, 4 insertions(+), 4 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index bd7f84fed9..36a3210c70 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -397,7 +397,7 @@ static void handle_request(Channel *channel, msgpack_object *request)
   Error error = ERROR_INIT;
   msgpack_rpc_validate(&request_id, request, &error);
 
-  if (error.set) {
+  if (ERROR_SET(&error)) {
     // Validation failed, send response with error
     if (channel_write(channel,
                       serialize_response(channel->id,
diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index c273343b41..800f5edb85 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -475,7 +475,7 @@ Object msgpack_rpc_handle_missing_method(uint64_t channel_id,
                                          Array args,
                                          Error *error)
 {
-  _api_set_error(error, error->type, "Invalid method name");
+  _api_set_error(error, kErrorTypeException, "Invalid method name");
   return NIL;
 }
 
@@ -484,7 +484,7 @@ Object msgpack_rpc_handle_invalid_arguments(uint64_t channel_id,
                                             Array args,
                                             Error *error)
 {
-  _api_set_error(error, error->type, "Invalid method arguments");
+  _api_set_error(error, kErrorTypeException, "Invalid method arguments");
   return NIL;
 }
 
@@ -517,7 +517,7 @@ void msgpack_rpc_serialize_response(uint64_t response_id,
   msgpack_pack_int(pac, 1);
   msgpack_pack_uint64(pac, response_id);
 
-  if (err->set) {
+  if (ERROR_SET(err)) {
     // error represented by a [type, message] array
     msgpack_pack_array(pac, 2);
     msgpack_rpc_from_integer(err->type, pac);
-- 
cgit 


From 3fbc660d57f4726044662bde1bf52c527e45fb98 Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Sun, 23 Apr 2017 21:54:44 +0200
Subject: api_set_error(): rename

---
 src/nvim/msgpack_rpc/channel.c | 12 ++++++------
 src/nvim/msgpack_rpc/helpers.c | 20 ++++++++++----------
 2 files changed, 16 insertions(+), 16 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 36a3210c70..3cf85316d9 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -192,7 +192,7 @@ Object channel_send_call(uint64_t id,
   Channel *channel = NULL;
 
   if (!(channel = pmap_get(uint64_t)(channels, id)) || channel->closed) {
-    _api_set_error(err, kErrorTypeException, _("Invalid channel \"%" PRIu64 "\""), id);
+    api_set_error(err, kErrorTypeException, _("Invalid channel \"%" PRIu64 "\""), id);
     api_free_array(args);
     return NIL;
   }
@@ -212,7 +212,7 @@ Object channel_send_call(uint64_t id,
 
   if (frame.errored) {
     if (frame.result.type == kObjectTypeString) {
-      _api_set_error(err, kErrorTypeException, "%s", frame.result.data.string.data);
+      api_set_error(err, kErrorTypeException, "%s", frame.result.data.string.data);
     } else if (frame.result.type == kObjectTypeArray) {
       // Should be an error in the form [type, message]
       Array array = frame.result.data.array;
@@ -220,13 +220,13 @@ Object channel_send_call(uint64_t id,
           && (array.items[0].data.integer == kErrorTypeException
               || array.items[0].data.integer == kErrorTypeValidation)
           && array.items[1].type == kObjectTypeString) {
-        _api_set_error(err, (ErrorType)array.items[0].data.integer, "%s",
+        api_set_error(err, (ErrorType)array.items[0].data.integer, "%s",
                        array.items[1].data.string.data);
       } else {
-        _api_set_error(err, kErrorTypeException, "%s", "unknown error");
+        api_set_error(err, kErrorTypeException, "%s", "unknown error");
       }
     } else {
-      _api_set_error(err, kErrorTypeException, "%s", "unknown error");
+      api_set_error(err, kErrorTypeException, "%s", "unknown error");
     }
 
     api_free_object(frame.result);
@@ -512,7 +512,7 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
 static void send_error(Channel *channel, uint64_t id, char *err)
 {
   Error e = ERROR_INIT;
-  _api_set_error(&e, kErrorTypeException, "%s", err);
+  api_set_error(&e, kErrorTypeException, "%s", err);
   channel_write(channel, serialize_response(channel->id,
                                             id,
                                             &e,
diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 800f5edb85..0ad1d9ae4b 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -475,7 +475,7 @@ Object msgpack_rpc_handle_missing_method(uint64_t channel_id,
                                          Array args,
                                          Error *error)
 {
-  _api_set_error(error, kErrorTypeException, "Invalid method name");
+  api_set_error(error, kErrorTypeException, "Invalid method name");
   return NIL;
 }
 
@@ -484,7 +484,7 @@ Object msgpack_rpc_handle_invalid_arguments(uint64_t channel_id,
                                             Array args,
                                             Error *error)
 {
-  _api_set_error(error, kErrorTypeException, "Invalid method arguments");
+  api_set_error(error, kErrorTypeException, "Invalid method arguments");
   return NIL;
 }
 
@@ -570,29 +570,29 @@ void msgpack_rpc_validate(uint64_t *response_id,
   *response_id = NO_RESPONSE;
   // Validate the basic structure of the msgpack-rpc payload
   if (req->type != MSGPACK_OBJECT_ARRAY) {
-    _api_set_error(err, kErrorTypeValidation, _("Message is not an array"));
+    api_set_error(err, kErrorTypeValidation, _("Message is not an array"));
     return;
   }
 
   if (req->via.array.size == 0) {
-    _api_set_error(err, kErrorTypeValidation, _("Message is empty"));
+    api_set_error(err, kErrorTypeValidation, _("Message is empty"));
     return;
   }
 
   if (req->via.array.ptr[0].type != MSGPACK_OBJECT_POSITIVE_INTEGER) {
-    _api_set_error(err, kErrorTypeValidation, _("Message type must be an integer"));
+    api_set_error(err, kErrorTypeValidation, _("Message type must be an integer"));
     return;
   }
 
   uint64_t type = req->via.array.ptr[0].via.u64;
   if (type != kMessageTypeRequest && type != kMessageTypeNotification) {
-    _api_set_error(err, kErrorTypeValidation, _("Unknown message type"));
+    api_set_error(err, kErrorTypeValidation, _("Unknown message type"));
     return;
   }
 
   if ((type == kMessageTypeRequest && req->via.array.size != 4)
       || (type == kMessageTypeNotification && req->via.array.size != 3)) {
-    _api_set_error(err, kErrorTypeValidation, _("Request array size should be 4 (request) "
+    api_set_error(err, kErrorTypeValidation, _("Request array size should be 4 (request) "
                                      "or 3 (notification)"));
     return;
   }
@@ -600,19 +600,19 @@ void msgpack_rpc_validate(uint64_t *response_id,
   if (type == kMessageTypeRequest) {
     msgpack_object *id_obj = msgpack_rpc_msg_id(req);
     if (!id_obj) {
-      _api_set_error(err, kErrorTypeValidation, _("ID must be a positive integer"));
+      api_set_error(err, kErrorTypeValidation, _("ID must be a positive integer"));
       return;
     }
     *response_id = id_obj->via.u64;
   }
 
   if (!msgpack_rpc_method(req)) {
-    _api_set_error(err, kErrorTypeValidation, _("Method must be a string"));
+    api_set_error(err, kErrorTypeValidation, _("Method must be a string"));
     return;
   }
 
   if (!msgpack_rpc_args(req)) {
-    _api_set_error(err, kErrorTypeValidation, _("Parameters must be an array"));
+    api_set_error(err, kErrorTypeValidation, _("Parameters must be an array"));
     return;
   }
 }
-- 
cgit 


From 086c354a0aad2769042dc91bf5bad021109f56e4 Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Sun, 23 Apr 2017 22:30:08 +0200
Subject: api: Do not translate error messages.

Also re-word some error messages:
- "Key does not exist: %s"
- "Invalid channel: %<PRIu64>"
- "Request array size must be 4 (request) or 3 (notification)"
- "String cannot contain newlines"

References #6150
---
 src/nvim/msgpack_rpc/channel.c |  7 ++++---
 src/nvim/msgpack_rpc/helpers.c | 18 +++++++++---------
 2 files changed, 13 insertions(+), 12 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 3cf85316d9..59594357de 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -192,7 +192,7 @@ Object channel_send_call(uint64_t id,
   Channel *channel = NULL;
 
   if (!(channel = pmap_get(uint64_t)(channels, id)) || channel->closed) {
-    api_set_error(err, kErrorTypeException, _("Invalid channel \"%" PRIu64 "\""), id);
+    api_set_error(err, kErrorTypeException, "Invalid channel: %" PRIu64, id);
     api_free_array(args);
     return NIL;
   }
@@ -212,7 +212,8 @@ Object channel_send_call(uint64_t id,
 
   if (frame.errored) {
     if (frame.result.type == kObjectTypeString) {
-      api_set_error(err, kErrorTypeException, "%s", frame.result.data.string.data);
+      api_set_error(err, kErrorTypeException, "%s",
+                    frame.result.data.string.data);
     } else if (frame.result.type == kObjectTypeArray) {
       // Should be an error in the form [type, message]
       Array array = frame.result.data.array;
@@ -221,7 +222,7 @@ Object channel_send_call(uint64_t id,
               || array.items[0].data.integer == kErrorTypeValidation)
           && array.items[1].type == kObjectTypeString) {
         api_set_error(err, (ErrorType)array.items[0].data.integer, "%s",
-                       array.items[1].data.string.data);
+                      array.items[1].data.string.data);
       } else {
         api_set_error(err, kErrorTypeException, "%s", "unknown error");
       }
diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 0ad1d9ae4b..0228582d37 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -570,49 +570,49 @@ void msgpack_rpc_validate(uint64_t *response_id,
   *response_id = NO_RESPONSE;
   // Validate the basic structure of the msgpack-rpc payload
   if (req->type != MSGPACK_OBJECT_ARRAY) {
-    api_set_error(err, kErrorTypeValidation, _("Message is not an array"));
+    api_set_error(err, kErrorTypeValidation, "Message is not an array");
     return;
   }
 
   if (req->via.array.size == 0) {
-    api_set_error(err, kErrorTypeValidation, _("Message is empty"));
+    api_set_error(err, kErrorTypeValidation, "Message is empty");
     return;
   }
 
   if (req->via.array.ptr[0].type != MSGPACK_OBJECT_POSITIVE_INTEGER) {
-    api_set_error(err, kErrorTypeValidation, _("Message type must be an integer"));
+    api_set_error(err, kErrorTypeValidation, "Message type must be an integer");
     return;
   }
 
   uint64_t type = req->via.array.ptr[0].via.u64;
   if (type != kMessageTypeRequest && type != kMessageTypeNotification) {
-    api_set_error(err, kErrorTypeValidation, _("Unknown message type"));
+    api_set_error(err, kErrorTypeValidation, "Unknown message type");
     return;
   }
 
   if ((type == kMessageTypeRequest && req->via.array.size != 4)
       || (type == kMessageTypeNotification && req->via.array.size != 3)) {
-    api_set_error(err, kErrorTypeValidation, _("Request array size should be 4 (request) "
-                                     "or 3 (notification)"));
+    api_set_error(err, kErrorTypeValidation,
+                  "Request array size must be 4 (request) or 3 (notification)");
     return;
   }
 
   if (type == kMessageTypeRequest) {
     msgpack_object *id_obj = msgpack_rpc_msg_id(req);
     if (!id_obj) {
-      api_set_error(err, kErrorTypeValidation, _("ID must be a positive integer"));
+      api_set_error(err, kErrorTypeValidation, "ID must be a positive integer");
       return;
     }
     *response_id = id_obj->via.u64;
   }
 
   if (!msgpack_rpc_method(req)) {
-    api_set_error(err, kErrorTypeValidation, _("Method must be a string"));
+    api_set_error(err, kErrorTypeValidation, "Method must be a string");
     return;
   }
 
   if (!msgpack_rpc_args(req)) {
-    api_set_error(err, kErrorTypeValidation, _("Parameters must be an array"));
+    api_set_error(err, kErrorTypeValidation, "Parameters must be an array");
     return;
   }
 }
-- 
cgit 


From 3ea10077534cb1dcb1597ffcf85e601fa0c0e27b Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Mon, 13 Mar 2017 15:02:37 +0100
Subject: api: nvim_get_mode()
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Asynchronous API functions are served immediately, which means pending
input could change the state of Nvim shortly after an async API function
result is returned.

nvim_get_mode() is different:
  - If RPCs are known to be blocked, it responds immediately (without
    flushing the input/event queue)
  - else it is handled just-in-time before waiting for input, after
    pending input was processed. This makes the result more reliable
    (but not perfect).

Internally this is handled as a special case, but _semantically_ nothing
has changed: API users never know when input flushes, so this internal
special-case doesn't violate that. As far as API users are concerned,
nvim_get_mode() is just another asynchronous API function.

In all cases nvim_get_mode() never blocks for more than the time it
takes to flush the input/event queue (~µs).

Note: This doesn't address #6166; nvim_get_mode() will provoke #6166 if
e.g. `d` is operator-pending.

Closes #6159
---
 src/nvim/msgpack_rpc/channel.c | 10 ++++++++++
 src/nvim/msgpack_rpc/helpers.c |  2 +-
 2 files changed, 11 insertions(+), 1 deletion(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 59594357de..799e6eadef 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -28,7 +28,9 @@
 #include "nvim/map.h"
 #include "nvim/log.h"
 #include "nvim/misc1.h"
+#include "nvim/state.h"
 #include "nvim/lib/kvec.h"
+#include "nvim/os/input.h"
 
 #define CHANNEL_BUFFER_SIZE 0xffff
 
@@ -433,6 +435,14 @@ static void handle_request(Channel *channel, msgpack_object *request)
     handler.async = true;
   }
 
+  if (handler.async) {
+    char buf[256] = { 0 };
+    memcpy(buf, method->via.bin.ptr, MIN(255, method->via.bin.size));
+    if (strcmp("nvim_get_mode", buf) == 0) {
+      handler.async = input_blocking();
+    }
+  }
+
   RequestEvent *event_data = xmalloc(sizeof(RequestEvent));
   event_data->channel = channel;
   event_data->handler = handler;
diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 0228582d37..91ef5524ea 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -76,7 +76,7 @@ typedef struct {
   size_t idx;
 } MPToAPIObjectStackItem;
 
-/// Convert type used by msgpack parser to Neovim own API type
+/// Convert type used by msgpack parser to Nvim API type.
 ///
 /// @param[in]  obj  Msgpack value to convert.
 /// @param[out]  arg  Location where result of conversion will be saved.
-- 
cgit 


From acfd2a2a29ae852ecc965ca888eb5049400bf39d Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Tue, 14 Mar 2017 00:44:03 +0100
Subject: input.c: Process only safe events before blocking.

Introduce multiqueue_process_priority() to process only events at or
above a certain priority.
---
 src/nvim/msgpack_rpc/channel.c | 32 +++++++++++++++++---------------
 1 file changed, 17 insertions(+), 15 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 799e6eadef..9e43924db1 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -435,24 +435,26 @@ static void handle_request(Channel *channel, msgpack_object *request)
     handler.async = true;
   }
 
-  if (handler.async) {
-    char buf[256] = { 0 };
-    memcpy(buf, method->via.bin.ptr, MIN(255, method->via.bin.size));
-    if (strcmp("nvim_get_mode", buf) == 0) {
-      handler.async = input_blocking();
-    }
-  }
-
-  RequestEvent *event_data = xmalloc(sizeof(RequestEvent));
-  event_data->channel = channel;
-  event_data->handler = handler;
-  event_data->args = args;
-  event_data->request_id = request_id;
+  RequestEvent *evdata = xmalloc(sizeof(RequestEvent));
+  evdata->channel = channel;
+  evdata->handler = handler;
+  evdata->args = args;
+  evdata->request_id = request_id;
   incref(channel);
   if (handler.async) {
-    on_request_event((void **)&event_data);
+    bool is_get_mode = sizeof("nvim_get_mode") - 1 == method->via.bin.size
+      && !strncmp("nvim_get_mode", method->via.bin.ptr, method->via.bin.size);
+
+    if (is_get_mode && !input_blocking()) {
+      // Schedule on the main loop with special priority. #6247
+      Event ev = event_create(kEvPriorityAsync, on_request_event, 1, evdata);
+      multiqueue_put_event(channel->events, ev);
+    } else {
+      // Invoke immediately.
+      on_request_event((void **)&evdata);
+    }
   } else {
-    multiqueue_put(channel->events, on_request_event, 1, event_data);
+    multiqueue_put(channel->events, on_request_event, 1, evdata);
   }
 }
 
-- 
cgit 


From f17a8185191b778960953508a5bf9b5f95b0560c Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Thu, 27 Apr 2017 13:54:54 +0200
Subject: api/nvim_get_mode: Use child-queue instead of "priority".

---
 src/nvim/msgpack_rpc/channel.c | 7 +++----
 src/nvim/msgpack_rpc/channel.h | 5 +++++
 2 files changed, 8 insertions(+), 4 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 9e43924db1..911f2a6fa4 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -28,7 +28,6 @@
 #include "nvim/map.h"
 #include "nvim/log.h"
 #include "nvim/misc1.h"
-#include "nvim/state.h"
 #include "nvim/lib/kvec.h"
 #include "nvim/os/input.h"
 
@@ -91,6 +90,7 @@ static msgpack_sbuffer out_buffer;
 /// Initializes the module
 void channel_init(void)
 {
+  ch_before_blocking_events = multiqueue_new_child(main_loop.events);
   channels = pmap_new(uint64_t)();
   event_strings = pmap_new(cstr_t)();
   msgpack_sbuffer_init(&out_buffer);
@@ -446,9 +446,8 @@ static void handle_request(Channel *channel, msgpack_object *request)
       && !strncmp("nvim_get_mode", method->via.bin.ptr, method->via.bin.size);
 
     if (is_get_mode && !input_blocking()) {
-      // Schedule on the main loop with special priority. #6247
-      Event ev = event_create(kEvPriorityAsync, on_request_event, 1, evdata);
-      multiqueue_put_event(channel->events, ev);
+      // Defer the event to a special queue used by os/input.c. #6247
+      multiqueue_put(ch_before_blocking_events, on_request_event, 1, evdata);
     } else {
       // Invoke immediately.
       on_request_event((void **)&evdata);
diff --git a/src/nvim/msgpack_rpc/channel.h b/src/nvim/msgpack_rpc/channel.h
index 0d92976d02..f8fe6f129b 100644
--- a/src/nvim/msgpack_rpc/channel.h
+++ b/src/nvim/msgpack_rpc/channel.h
@@ -11,6 +11,11 @@
 
 #define METHOD_MAXLEN 512
 
+/// HACK: os/input.c drains this queue immediately before blocking for input.
+///       Events on this queue are async-safe, but they need the resolved state
+///       of os_inchar(), so they are processed "just-in-time".
+MultiQueue *ch_before_blocking_events;
+
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "msgpack_rpc/channel.h.generated.h"
 #endif
-- 
cgit 


From 1483800cdff3bd00faabe66436a1618f567f6754 Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Mon, 1 May 2017 17:04:56 +0200
Subject: coverity/161682: FP: Dereference after null check (#6630)

---
 src/nvim/msgpack_rpc/channel.c | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 911f2a6fa4..cd64e14976 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -442,8 +442,7 @@ static void handle_request(Channel *channel, msgpack_object *request)
   evdata->request_id = request_id;
   incref(channel);
   if (handler.async) {
-    bool is_get_mode = sizeof("nvim_get_mode") - 1 == method->via.bin.size
-      && !strncmp("nvim_get_mode", method->via.bin.ptr, method->via.bin.size);
+    bool is_get_mode = handler.fn == handle_nvim_get_mode;
 
     if (is_get_mode && !input_blocking()) {
       // Defer the event to a special queue used by os/input.c. #6247
-- 
cgit 


From aace622ca5ffe4deb72989452dc60877304d71aa Mon Sep 17 00:00:00 2001
From: Carlo Abelli <carlo.abelli@gmail.com>
Date: Mon, 8 May 2017 09:08:12 -0400
Subject: refactor/single-include (#6687)

---
 src/nvim/msgpack_rpc/server.h | 2 ++
 1 file changed, 2 insertions(+)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/server.h b/src/nvim/msgpack_rpc/server.h
index f1a6703938..5446e40e0b 100644
--- a/src/nvim/msgpack_rpc/server.h
+++ b/src/nvim/msgpack_rpc/server.h
@@ -1,6 +1,8 @@
 #ifndef NVIM_MSGPACK_RPC_SERVER_H
 #define NVIM_MSGPACK_RPC_SERVER_H
 
+#include <stdio.h>
+
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "msgpack_rpc/server.h.generated.h"
 #endif
-- 
cgit 


From 3efc82cbb28a1fd6f1c10a4fc91e8ce749ad384f Mon Sep 17 00:00:00 2001
From: Marco Hinz <mh.codebro@gmail.com>
Date: Thu, 4 May 2017 17:59:51 +0200
Subject: Server: use uv_getaddrinfo() for $NVIM_LISTEN_ADDRESS

This change implicitly adds IPv6 support.

If the address contains ":", we try to use a TCP socket instead of a Unix domain
socket. Everything in front of the last occurrence of ":" is the hostname and
everything after it the port.

If the hostname lookup fails, we fall back to using a Unix domain socket.

If the port is empty ("localhost:"), a random port will be assigned.

Examples:

  NVIM_LISTEN_ADDRESS=localhost:12345 -> TCP (IPv4 or IPv6), port: 12345
  NVIM_LISTEN_ADDRESS=localhost:      -> TCP (IPv4 or IPv6), port: random (> 1024)
  NVIM_LISTEN_ADDRESS=localhost:0     -> TCP (IPv4 or IPv6), port: random (> 1024)
  NVIM_LISTEN_ADDRESS=localhost       -> Unix domain socket "localhost" in current dir
---
 src/nvim/msgpack_rpc/server.c | 27 ++++++++++++++++-----------
 1 file changed, 16 insertions(+), 11 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/server.c b/src/nvim/msgpack_rpc/server.c
index b6958088ca..ee96fa1a74 100644
--- a/src/nvim/msgpack_rpc/server.c
+++ b/src/nvim/msgpack_rpc/server.c
@@ -97,31 +97,36 @@ char *server_address_new(void)
 #endif
 }
 
-/// Starts listening for API calls on the TCP address or pipe path `endpoint`.
+/// Starts listening for API calls.
+///
 /// The socket type is determined by parsing `endpoint`: If it's a valid IPv4
-/// address in 'ip[:port]' format, then it will be TCP socket. The port is
-/// optional and if omitted defaults to NVIM_DEFAULT_TCP_PORT. Otherwise it
-/// will be a unix socket or named pipe.
+/// or IPv6 address in 'ip:[port]' format, then it will be a TCP socket.
+/// Otherwise it will be a Unix socket or named pipe (Windows).
+///
+/// If no port is given, a random one will be assigned.
 ///
-/// @param endpoint Address of the server. Either a 'ip[:port]' string or an
-///        arbitrary identifier (trimmed to 256 bytes) for the unix socket or
-///        named pipe.
+/// @param endpoint Address of the server. Either a 'ip:[port]' string or an
+///                 arbitrary identifier (trimmed to 256 bytes) for the Unix
+///                 socket or named pipe.
 /// @returns 0 on success, 1 on a regular error, and negative errno
-///          on failure to bind or connect.
+///          on failure to bind or listen.
 int server_start(const char *endpoint)
 {
-  if (endpoint == NULL) {
-    ELOG("Attempting to start server on NULL endpoint");
+  if (endpoint == NULL || endpoint[0] == '\0') {
+    ELOG("Empty or NULL endpoint");
     return 1;
   }
 
   SocketWatcher *watcher = xmalloc(sizeof(SocketWatcher));
-  socket_watcher_init(&main_loop, watcher, endpoint, NULL);
+  socket_watcher_init(&main_loop, watcher, endpoint);
 
   // Check if a watcher for the endpoint already exists
   for (int i = 0; i < watchers.ga_len; i++) {
     if (!strcmp(watcher->addr, ((SocketWatcher **)watchers.ga_data)[i]->addr)) {
       ELOG("Already listening on %s", watcher->addr);
+      if (watcher->stream->type == UV_TCP) {
+        uv_freeaddrinfo(watcher->uv.tcp.addrinfo);
+      }
       socket_watcher_close(watcher, free_server);
       return 1;
     }
-- 
cgit 


From fd5e4e2e4c6b231980effc3055fb93300bd055b0 Mon Sep 17 00:00:00 2001
From: Marco Hinz <mh.codebro@gmail.com>
Date: Tue, 16 May 2017 14:56:08 +0200
Subject: Server: don't fall back to Unix sockets

---
 src/nvim/msgpack_rpc/server.c | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/server.c b/src/nvim/msgpack_rpc/server.c
index ee96fa1a74..bae5a32850 100644
--- a/src/nvim/msgpack_rpc/server.c
+++ b/src/nvim/msgpack_rpc/server.c
@@ -118,7 +118,12 @@ int server_start(const char *endpoint)
   }
 
   SocketWatcher *watcher = xmalloc(sizeof(SocketWatcher));
-  socket_watcher_init(&main_loop, watcher, endpoint);
+
+  int result = socket_watcher_init(&main_loop, watcher, endpoint);
+  if (result < 0) {
+    xfree(watcher);
+    return result;
+  }
 
   // Check if a watcher for the endpoint already exists
   for (int i = 0; i < watchers.ga_len; i++) {
@@ -132,7 +137,7 @@ int server_start(const char *endpoint)
     }
   }
 
-  int result = socket_watcher_start(watcher, MAX_CONNECTIONS, connection_cb);
+  result = socket_watcher_start(watcher, MAX_CONNECTIONS, connection_cb);
   if (result < 0) {
     ELOG("Failed to start server: %s", uv_strerror(result));
     socket_watcher_close(watcher, free_server);
-- 
cgit 


From 6a7593875827374c15484dd4eecd31a88f8c6f77 Mon Sep 17 00:00:00 2001
From: Björn Linse <bjorn.linse@gmail.com>
Date: Wed, 26 Apr 2017 13:10:21 +0200
Subject: channels: implement sockopen() to connect to socket

Helped-By: oni-link <knil.ino@gmail.com>
---
 src/nvim/msgpack_rpc/channel.c | 19 +++++++++++++++++++
 1 file changed, 19 insertions(+)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index cd64e14976..0ff749649b 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -145,6 +145,25 @@ void channel_from_connection(SocketWatcher *watcher)
   rstream_start(&channel->data.stream, parse_msgpack, channel);
 }
 
+uint64_t channel_connect(bool tcp, const char *address,
+                         int timeout, const char **error)
+{
+  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL);
+  if (!socket_connect(&main_loop, &channel->data.stream,
+                      tcp, address, timeout, error)) {
+    decref(channel);
+    return 0;
+  }
+
+  incref(channel);  // close channel only after the stream is closed
+  channel->data.stream.internal_close_cb = close_cb;
+  channel->data.stream.internal_data = channel;
+  wstream_init(&channel->data.stream, 0);
+  rstream_init(&channel->data.stream, CHANNEL_BUFFER_SIZE);
+  rstream_start(&channel->data.stream, parse_msgpack, channel);
+  return channel->id;
+}
+
 /// Sends event/arguments to channel
 ///
 /// @param id The channel id. If 0, the event will be sent to all
-- 
cgit 


From 5a151555c8dce70bbf235e7f6d5bd1ced5e7c46c Mon Sep 17 00:00:00 2001
From: Björn Linse <bjorn.linse@gmail.com>
Date: Sat, 6 May 2017 09:31:36 +0200
Subject: sockets: don't deadlock when connecting to own pipe address

---
 src/nvim/msgpack_rpc/channel.c | 78 ++++++++++++++++++++++++++++++++++--------
 src/nvim/msgpack_rpc/server.c  | 12 +++++++
 2 files changed, 75 insertions(+), 15 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 0ff749649b..e8ee0ede75 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -12,6 +12,7 @@
 #include "nvim/api/vim.h"
 #include "nvim/api/ui.h"
 #include "nvim/msgpack_rpc/channel.h"
+#include "nvim/msgpack_rpc/server.h"
 #include "nvim/event/loop.h"
 #include "nvim/event/libuv_process.h"
 #include "nvim/event/rstream.h"
@@ -28,6 +29,7 @@
 #include "nvim/map.h"
 #include "nvim/log.h"
 #include "nvim/misc1.h"
+#include "nvim/path.h"
 #include "nvim/lib/kvec.h"
 #include "nvim/os/input.h"
 
@@ -41,7 +43,8 @@
 typedef enum {
   kChannelTypeSocket,
   kChannelTypeProc,
-  kChannelTypeStdio
+  kChannelTypeStdio,
+  kChannelTypeInternal
 } ChannelType;
 
 typedef struct {
@@ -125,7 +128,7 @@ uint64_t channel_from_process(Process *proc, uint64_t id)
 
   wstream_init(proc->in, 0);
   rstream_init(proc->out, 0);
-  rstream_start(proc->out, parse_msgpack, channel);
+  rstream_start(proc->out, receive_msgpack, channel);
 
   return channel->id;
 }
@@ -142,12 +145,22 @@ void channel_from_connection(SocketWatcher *watcher)
   channel->data.stream.internal_data = channel;
   wstream_init(&channel->data.stream, 0);
   rstream_init(&channel->data.stream, CHANNEL_BUFFER_SIZE);
-  rstream_start(&channel->data.stream, parse_msgpack, channel);
+  rstream_start(&channel->data.stream, receive_msgpack, channel);
 }
 
 uint64_t channel_connect(bool tcp, const char *address,
                          int timeout, const char **error)
 {
+  if (!tcp) {
+    char *path = fix_fname(address);
+    if (server_owns_pipe_address(path)) {
+      // avoid deadlock
+      xfree(path);
+      return channel_create_internal();
+    }
+    xfree(path);
+  }
+
   Channel *channel = register_channel(kChannelTypeSocket, 0, NULL);
   if (!socket_connect(&main_loop, &channel->data.stream,
                       tcp, address, timeout, error)) {
@@ -160,7 +173,7 @@ uint64_t channel_connect(bool tcp, const char *address,
   channel->data.stream.internal_data = channel;
   wstream_init(&channel->data.stream, 0);
   rstream_init(&channel->data.stream, CHANNEL_BUFFER_SIZE);
-  rstream_start(&channel->data.stream, parse_msgpack, channel);
+  rstream_start(&channel->data.stream, receive_msgpack, channel);
   return channel->id;
 }
 
@@ -324,11 +337,20 @@ void channel_from_stdio(void)
   incref(channel);  // stdio channels are only closed on exit
   // read stream
   rstream_init_fd(&main_loop, &channel->data.std.in, 0, CHANNEL_BUFFER_SIZE);
-  rstream_start(&channel->data.std.in, parse_msgpack, channel);
+  rstream_start(&channel->data.std.in, receive_msgpack, channel);
   // write stream
   wstream_init_fd(&main_loop, &channel->data.std.out, 1, 0);
 }
 
+/// Creates a loopback channel. This is used to avoid deadlock
+/// when an instance connects to its own named pipe.
+uint64_t channel_create_internal(void)
+{
+  Channel *channel = register_channel(kChannelTypeInternal, 0, NULL);
+  incref(channel);  // internal channel lives until process exit
+  return channel->id;
+}
+
 void channel_process_exit(uint64_t id, int status)
 {
   Channel *channel = pmap_get(uint64_t)(channels, id);
@@ -337,8 +359,8 @@ void channel_process_exit(uint64_t id, int status)
   decref(channel);
 }
 
-static void parse_msgpack(Stream *stream, RBuffer *rbuf, size_t c, void *data,
-    bool eof)
+static void receive_msgpack(Stream *stream, RBuffer *rbuf, size_t c,
+                            void *data, bool eof)
 {
   Channel *channel = data;
   incref(channel);
@@ -360,6 +382,14 @@ static void parse_msgpack(Stream *stream, RBuffer *rbuf, size_t c, void *data,
   rbuffer_read(rbuf, msgpack_unpacker_buffer(channel->unpacker), count);
   msgpack_unpacker_buffer_consumed(channel->unpacker, count);
 
+  parse_msgpack(channel);
+
+end:
+  decref(channel);
+}
+
+static void parse_msgpack(Channel *channel)
+{
   msgpack_unpacked unpacked;
   msgpack_unpacked_init(&unpacked);
   msgpack_unpack_return result;
@@ -383,7 +413,7 @@ static void parse_msgpack(Stream *stream, RBuffer *rbuf, size_t c, void *data,
       }
       msgpack_unpacked_destroy(&unpacked);
       // Bail out from this event loop iteration
-      goto end;
+      return;
     }
 
     handle_request(channel, &unpacked.data);
@@ -407,11 +437,9 @@ static void parse_msgpack(Stream *stream, RBuffer *rbuf, size_t c, void *data,
                            "This error can also happen when deserializing "
                            "an object with high level of nesting");
   }
-
-end:
-  decref(channel);
 }
 
+
 static void handle_request(Channel *channel, msgpack_object *request)
   FUNC_ATTR_NONNULL_ALL
 {
@@ -521,8 +549,11 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
     case kChannelTypeStdio:
       success = wstream_write(&channel->data.std.out, buffer);
       break;
-    default:
-      abort();
+    case kChannelTypeInternal:
+      incref(channel);
+      CREATE_EVENT(channel->events, internal_read_event, 2, channel, buffer);
+      success = true;
+      break;
   }
 
   if (!success) {
@@ -539,6 +570,22 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
   return success;
 }
 
+static void internal_read_event(void **argv)
+{
+  Channel *channel = argv[0];
+  WBuffer *buffer = argv[1];
+
+  msgpack_unpacker_reserve_buffer(channel->unpacker, buffer->size);
+  memcpy(msgpack_unpacker_buffer(channel->unpacker),
+         buffer->data, buffer->size);
+  msgpack_unpacker_buffer_consumed(channel->unpacker, buffer->size);
+
+  parse_msgpack(channel);
+
+  decref(channel);
+  wstream_release_wbuffer(buffer);
+}
+
 static void send_error(Channel *channel, uint64_t id, char *err)
 {
   Error e = ERROR_INIT;
@@ -655,8 +702,9 @@ static void close_channel(Channel *channel)
       stream_close(&channel->data.std.out, NULL, NULL);
       multiqueue_put(main_loop.fast_events, exit_event, 1, channel);
       return;
-    default:
-      abort();
+    case kChannelTypeInternal:
+      // nothing to free.
+      break;
   }
 
   decref(channel);
diff --git a/src/nvim/msgpack_rpc/server.c b/src/nvim/msgpack_rpc/server.c
index bae5a32850..c9edd05dc2 100644
--- a/src/nvim/msgpack_rpc/server.c
+++ b/src/nvim/msgpack_rpc/server.c
@@ -97,6 +97,18 @@ char *server_address_new(void)
 #endif
 }
 
+/// Check if this instance owns a pipe address.
+/// The argument must already be resolved to an absolute path!
+bool server_owns_pipe_address(const char *path)
+{
+  for (int i = 0; i < watchers.ga_len; i++) {
+    if (!strcmp(path, ((SocketWatcher **)watchers.ga_data)[i]->addr)) {
+      return true;
+    }
+  }
+  return false;
+}
+
 /// Starts listening for API calls.
 ///
 /// The socket type is determined by parsing `endpoint`: If it's a valid IPv4
-- 
cgit 


From fcc9d999670c6fb29ac01e9c8d0a7e787ccfabea Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Wed, 31 May 2017 03:37:29 +0200
Subject: channel_write: fix compiler warning

---
 src/nvim/msgpack_rpc/channel.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index e8ee0ede75..413b800af5 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -532,7 +532,7 @@ static void on_request_event(void **argv)
 
 static bool channel_write(Channel *channel, WBuffer *buffer)
 {
-  bool success;
+  bool success = false;
 
   if (channel->closed) {
     wstream_release_wbuffer(buffer);
-- 
cgit 


From fe1af9c2bc020cdb9e1e86e1f16d6060246f957a Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Tue, 30 May 2017 02:03:08 +0200
Subject: log: Always enable; remove DISABLE_LOG

- Establish ERROR log level as "critical". Such errors are rare and will
  be valuable when users encounter unusual circumstances.
- Set -DMIN_LOG_LEVEL=3 for release-type builds
---
 src/nvim/msgpack_rpc/channel.c | 12 ++++++------
 src/nvim/msgpack_rpc/server.c  |  4 ++--
 2 files changed, 8 insertions(+), 8 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 413b800af5..68ac35bc4e 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -370,7 +370,7 @@ 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);
+    call_set_error(channel, buf, WARNING_LOG_LEVEL);
     goto end;
   }
 
@@ -409,7 +409,7 @@ static void parse_msgpack(Channel *channel)
                  "ch %" PRIu64 " returned a response with an unknown request "
                  "id. Ensure the client is properly synchronized",
                  channel->id);
-        call_set_error(channel, buf);
+        call_set_error(channel, buf, ERROR_LOG_LEVEL);
       }
       msgpack_unpacked_destroy(&unpacked);
       // Bail out from this event loop iteration
@@ -459,7 +459,7 @@ static void handle_request(Channel *channel, msgpack_object *request)
       snprintf(buf, sizeof(buf),
                "ch %" PRIu64 " sent an invalid message, closed.",
                channel->id);
-      call_set_error(channel, buf);
+      call_set_error(channel, buf, ERROR_LOG_LEVEL);
     }
     api_clear_error(&error);
     return;
@@ -564,7 +564,7 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
              "Before returning from a RPC call, ch %" PRIu64 " was "
              "closed due to a failed write",
              channel->id);
-    call_set_error(channel, buf);
+    call_set_error(channel, buf, ERROR_LOG_LEVEL);
   }
 
   return success;
@@ -795,9 +795,9 @@ static void complete_call(msgpack_object *obj, Channel *channel)
   }
 }
 
-static void call_set_error(Channel *channel, char *msg)
+static void call_set_error(Channel *channel, char *msg, int loglevel)
 {
-  ELOG("RPC: %s", msg);
+  LOG(loglevel, "RPC: %s", msg);
   for (size_t i = 0; i < kv_size(channel->call_stack); i++) {
     ChannelCallFrame *frame = kv_A(channel->call_stack, i);
     frame->returned = true;
diff --git a/src/nvim/msgpack_rpc/server.c b/src/nvim/msgpack_rpc/server.c
index c9edd05dc2..1e0cc27886 100644
--- a/src/nvim/msgpack_rpc/server.c
+++ b/src/nvim/msgpack_rpc/server.c
@@ -125,7 +125,7 @@ bool server_owns_pipe_address(const char *path)
 int server_start(const char *endpoint)
 {
   if (endpoint == NULL || endpoint[0] == '\0') {
-    ELOG("Empty or NULL endpoint");
+    WLOG("Empty or NULL endpoint");
     return 1;
   }
 
@@ -151,7 +151,7 @@ int server_start(const char *endpoint)
 
   result = socket_watcher_start(watcher, MAX_CONNECTIONS, connection_cb);
   if (result < 0) {
-    ELOG("Failed to start server: %s", uv_strerror(result));
+    WLOG("Failed to start server: %s", uv_strerror(result));
     socket_watcher_close(watcher, free_server);
     return result;
   }
-- 
cgit 


From 502af39f6261e1d1b71df7fa3169e943707ee7f2 Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Wed, 19 Jul 2017 02:24:12 +0200
Subject: log: channel registration

---
 src/nvim/msgpack_rpc/channel.c | 44 +++++++++++++++++++++++++++++++++---------
 1 file changed, 35 insertions(+), 9 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 68ac35bc4e..d7748925aa 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -117,12 +117,11 @@ void channel_teardown(void)
 /// Creates an API channel by starting a process and connecting to its
 /// stdin/stdout. stderr is handled by the job infrastructure.
 ///
-/// @param argv The argument vector for the process. [consumed]
-/// @return The channel id (> 0), on success.
-///         0, on error.
+/// @return Channel id (> 0), on success. 0, on error.
 uint64_t channel_from_process(Process *proc, uint64_t id)
 {
-  Channel *channel = register_channel(kChannelTypeProc, id, proc->events);
+  Channel *channel = register_channel(kChannelTypeProc, id, proc->events,
+                                      proc->argv);
   incref(channel);  // process channels are only closed by the exit_cb
   channel->data.proc = proc;
 
@@ -138,7 +137,7 @@ uint64_t channel_from_process(Process *proc, uint64_t id)
 /// @param watcher The SocketWatcher ready to accept the connection
 void channel_from_connection(SocketWatcher *watcher)
 {
-  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL);
+  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL, NULL);
   socket_watcher_accept(watcher, &channel->data.stream);
   incref(channel);  // close channel only after the stream is closed
   channel->data.stream.internal_close_cb = close_cb;
@@ -161,7 +160,7 @@ uint64_t channel_connect(bool tcp, const char *address,
     xfree(path);
   }
 
-  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL);
+  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL, NULL);
   if (!socket_connect(&main_loop, &channel->data.stream,
                       tcp, address, timeout, error)) {
     decref(channel);
@@ -333,7 +332,7 @@ bool channel_close(uint64_t id)
 /// Neovim
 void channel_from_stdio(void)
 {
-  Channel *channel = register_channel(kChannelTypeStdio, 0, NULL);
+  Channel *channel = register_channel(kChannelTypeStdio, 0, NULL, NULL);
   incref(channel);  // stdio channels are only closed on exit
   // read stream
   rstream_init_fd(&main_loop, &channel->data.std.in, 0, CHANNEL_BUFFER_SIZE);
@@ -346,7 +345,7 @@ void channel_from_stdio(void)
 /// when an instance connects to its own named pipe.
 uint64_t channel_create_internal(void)
 {
-  Channel *channel = register_channel(kChannelTypeInternal, 0, NULL);
+  Channel *channel = register_channel(kChannelTypeInternal, 0, NULL, NULL);
   incref(channel);  // internal channel lives until process exit
   return channel->id;
 }
@@ -746,7 +745,7 @@ static void close_cb(Stream *stream, void *data)
 }
 
 static Channel *register_channel(ChannelType type, uint64_t id,
-                                 MultiQueue *events)
+                                 MultiQueue *events, char **argv)
 {
   Channel *rv = xmalloc(sizeof(Channel));
   rv->events = events ? events : multiqueue_new_child(main_loop.events);
@@ -761,6 +760,33 @@ static Channel *register_channel(ChannelType type, uint64_t id,
   kv_init(rv->call_stack);
   kv_init(rv->delayed_notifications);
   pmap_put(uint64_t)(channels, rv->id, rv);
+
+#if MIN_LOG_LEVEL <= DEBUG_LOG_LEVEL
+  switch(type) {
+    case kChannelTypeSocket:
+      DLOG("register ch %" PRIu64 " type=socket", rv->id);
+      break;
+    case kChannelTypeProc: {
+      char buf[IOSIZE];
+      buf[0] = '\0';
+      char **p = argv;
+      while (p != NULL && *p != NULL) {
+        xstrlcat(buf, *p, sizeof(buf));
+        xstrlcat(buf, " ", sizeof(buf));
+        p++;
+      }
+      DLOG("register ch %" PRIu64 " type=proc argv=%s", rv->id, buf);
+      }
+      break;
+    case kChannelTypeStdio:
+      DLOG("register ch %" PRIu64 " type=stdio", rv->id);
+      break;
+    case kChannelTypeInternal:
+      DLOG("register ch %" PRIu64 " type=internal", rv->id);
+      break;
+  }
+#endif
+
   return rv;
 }
 
-- 
cgit 


From bc6a3fe78445f41851f30c6f55e11b6b1ed5feaf Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Wed, 19 Jul 2017 10:27:34 +0200
Subject: log: caller provides the source details

---
 src/nvim/msgpack_rpc/channel.c | 57 +++++++++++++++++-------------------------
 1 file changed, 23 insertions(+), 34 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index d7748925aa..6fd1af1ba6 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -117,11 +117,15 @@ void channel_teardown(void)
 /// Creates an API channel by starting a process and connecting to its
 /// stdin/stdout. stderr is handled by the job infrastructure.
 ///
+/// @param proc   process object
+/// @param id     (optional) channel id
+/// @param source description of source function, rplugin name, TCP addr, etc
+///
 /// @return Channel id (> 0), on success. 0, on error.
-uint64_t channel_from_process(Process *proc, uint64_t id)
+uint64_t channel_from_process(Process *proc, uint64_t id, char *source)
 {
   Channel *channel = register_channel(kChannelTypeProc, id, proc->events,
-                                      proc->argv);
+                                      source);
   incref(channel);  // process channels are only closed by the exit_cb
   channel->data.proc = proc;
 
@@ -137,7 +141,8 @@ uint64_t channel_from_process(Process *proc, uint64_t id)
 /// @param watcher The SocketWatcher ready to accept the connection
 void channel_from_connection(SocketWatcher *watcher)
 {
-  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL, NULL);
+  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL,
+                                      watcher->addr);
   socket_watcher_accept(watcher, &channel->data.stream);
   incref(channel);  // close channel only after the stream is closed
   channel->data.stream.internal_close_cb = close_cb;
@@ -147,8 +152,9 @@ void channel_from_connection(SocketWatcher *watcher)
   rstream_start(&channel->data.stream, receive_msgpack, channel);
 }
 
-uint64_t channel_connect(bool tcp, const char *address,
-                         int timeout, const char **error)
+/// @param source description of source function, rplugin name, TCP addr, etc
+uint64_t channel_connect(bool tcp, const char *address, int timeout,
+                         char *source, const char **error)
 {
   if (!tcp) {
     char *path = fix_fname(address);
@@ -160,7 +166,7 @@ uint64_t channel_connect(bool tcp, const char *address,
     xfree(path);
   }
 
-  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL, NULL);
+  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL, source);
   if (!socket_connect(&main_loop, &channel->data.stream,
                       tcp, address, timeout, error)) {
     decref(channel);
@@ -328,8 +334,7 @@ bool channel_close(uint64_t id)
   return true;
 }
 
-/// Creates an API channel from stdin/stdout. This is used when embedding
-/// Neovim
+/// Creates an API channel from stdin/stdout. Used to embed Nvim.
 void channel_from_stdio(void)
 {
   Channel *channel = register_channel(kChannelTypeStdio, 0, NULL, NULL);
@@ -744,9 +749,12 @@ static void close_cb(Stream *stream, void *data)
   decref(data);
 }
 
+/// @param source description of source function, rplugin name, TCP addr, etc
 static Channel *register_channel(ChannelType type, uint64_t id,
-                                 MultiQueue *events, char **argv)
+                                 MultiQueue *events, char *source)
 {
+  // Jobs and channels share the same id namespace.
+  assert(id == 0 || !pmap_get(uint64_t)(channels, id));
   Channel *rv = xmalloc(sizeof(Channel));
   rv->events = events ? events : multiqueue_new_child(main_loop.events);
   rv->type = type;
@@ -761,31 +769,12 @@ static Channel *register_channel(ChannelType type, uint64_t id,
   kv_init(rv->delayed_notifications);
   pmap_put(uint64_t)(channels, rv->id, rv);
 
-#if MIN_LOG_LEVEL <= DEBUG_LOG_LEVEL
-  switch(type) {
-    case kChannelTypeSocket:
-      DLOG("register ch %" PRIu64 " type=socket", rv->id);
-      break;
-    case kChannelTypeProc: {
-      char buf[IOSIZE];
-      buf[0] = '\0';
-      char **p = argv;
-      while (p != NULL && *p != NULL) {
-        xstrlcat(buf, *p, sizeof(buf));
-        xstrlcat(buf, " ", sizeof(buf));
-        p++;
-      }
-      DLOG("register ch %" PRIu64 " type=proc argv=%s", rv->id, buf);
-      }
-      break;
-    case kChannelTypeStdio:
-      DLOG("register ch %" PRIu64 " type=stdio", rv->id);
-      break;
-    case kChannelTypeInternal:
-      DLOG("register ch %" PRIu64 " type=internal", rv->id);
-      break;
-  }
-#endif
+  ILOG("new channel %" PRIu64 " (%s): %s", rv->id,
+       (type == kChannelTypeProc ? "proc"
+        : (type == kChannelTypeSocket ? "socket"
+           : (type == kChannelTypeStdio ? "stdio"
+              : (type == kChannelTypeInternal ? "internal" : "?")))),
+       (source ? source : "?"));
 
   return rv;
 }
-- 
cgit 


From e006b1d98d92aead6aff565ed5d8b034772aa16c Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Thu, 27 Jul 2017 00:30:57 +0200
Subject: log: some DEBUG-level stream logging

---
 src/nvim/msgpack_rpc/channel.c | 16 +++++++++++++---
 1 file changed, 13 insertions(+), 3 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 6fd1af1ba6..789cf1d4a8 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -133,6 +133,9 @@ uint64_t channel_from_process(Process *proc, uint64_t id, char *source)
   rstream_init(proc->out, 0);
   rstream_start(proc->out, receive_msgpack, channel);
 
+  DLOG("ch %" PRIu64 " in-stream=%p out-stream=%p", channel->id, proc->in,
+       proc->out);
+
   return channel->id;
 }
 
@@ -150,6 +153,8 @@ void channel_from_connection(SocketWatcher *watcher)
   wstream_init(&channel->data.stream, 0);
   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);
 }
 
 /// @param source description of source function, rplugin name, TCP addr, etc
@@ -344,6 +349,9 @@ void channel_from_stdio(void)
   rstream_start(&channel->data.std.in, receive_msgpack, channel);
   // write stream
   wstream_init_fd(&main_loop, &channel->data.std.out, 1, 0);
+
+  DLOG("ch %" PRIu64 " in-stream=%p out-stream=%p", channel->id,
+       &channel->data.std.in, &channel->data.std.out);
 }
 
 /// Creates a loopback channel. This is used to avoid deadlock
@@ -363,6 +371,7 @@ void channel_process_exit(uint64_t id, int status)
   decref(channel);
 }
 
+// rstream.c:read_event() invokes this as stream->read_cb().
 static void receive_msgpack(Stream *stream, RBuffer *rbuf, size_t c,
                             void *data, bool eof)
 {
@@ -379,7 +388,8 @@ static void receive_msgpack(Stream *stream, RBuffer *rbuf, size_t c,
   }
 
   size_t count = rbuffer_size(rbuf);
-  DLOG("parsing %u bytes of msgpack data from Stream(%p)", count, stream);
+  DLOG("ch %" PRIu64 ": parsing %u bytes from msgpack Stream: %p",
+       channel->id, count, stream);
 
   // Feed the unpacker with data
   msgpack_unpacker_reserve_buffer(channel->unpacker, count);
@@ -565,8 +575,8 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
     char buf[256];
     snprintf(buf,
              sizeof(buf),
-             "Before returning from a RPC call, ch %" PRIu64 " was "
-             "closed due to a failed write",
+             "ch %" PRIu64 ": stream write failed. "
+             "RPC canceled; closing channel",
              channel->id);
     call_set_error(channel, buf, ERROR_LOG_LEVEL);
   }
-- 
cgit 


From af993da4351d579aa24de08d806c8c1b90813106 Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Thu, 27 Jul 2017 01:08:50 +0200
Subject: 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)
---
 src/nvim/msgpack_rpc/channel.c | 61 +++++++++++++++++++++++++++++++++++-------
 src/nvim/msgpack_rpc/helpers.c |  2 +-
 2 files changed, 52 insertions(+), 11 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

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.
-- 
cgit 


From 0f442c328e33a0c742df9283c1944c077d202a77 Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Fri, 18 Aug 2017 14:13:28 +0200
Subject: channel.c:call_set_error(): fix memory leak

---
 src/nvim/msgpack_rpc/channel.c | 1 +
 src/nvim/msgpack_rpc/helpers.c | 7 ++++++-
 2 files changed, 7 insertions(+), 1 deletion(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index ed97f298d9..02f3854f47 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -868,6 +868,7 @@ static void call_set_error(Channel *channel, char *msg, int loglevel)
     ChannelCallFrame *frame = kv_A(channel->call_stack, i);
     frame->returned = true;
     frame->errored = true;
+    api_free_object(frame->result);
     frame->result = STRING_OBJ(cstr_to_string(msg));
   }
 
diff --git a/src/nvim/msgpack_rpc/helpers.c b/src/nvim/msgpack_rpc/helpers.c
index 98e56dee3b..fecae11d45 100644
--- a/src/nvim/msgpack_rpc/helpers.c
+++ b/src/nvim/msgpack_rpc/helpers.c
@@ -88,7 +88,12 @@ bool msgpack_rpc_to_object(const msgpack_object *const obj, Object *const arg)
 {
   bool ret = true;
   kvec_t(MPToAPIObjectStackItem) stack = KV_INITIAL_VALUE;
-  kv_push(stack, ((MPToAPIObjectStackItem) { obj, arg, false, 0 }));
+  kv_push(stack, ((MPToAPIObjectStackItem) {
+    .mobj = obj,
+    .aobj = arg,
+    .container = false,
+    .idx = 0,
+  }));
   while (ret && kv_size(stack)) {
     MPToAPIObjectStackItem cur = kv_last(stack);
     if (!cur.container) {
-- 
cgit 


From cdd9e868efdad1f1eb9febfabb5f8671e75b95b9 Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Sat, 2 Sep 2017 19:51:49 +0200
Subject: doc: channel, eventloop

---
 src/nvim/msgpack_rpc/channel.c | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 02f3854f47..88232a55de 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -188,12 +188,11 @@ uint64_t channel_connect(bool tcp, const char *address, int timeout,
   return channel->id;
 }
 
-/// Sends event/arguments to channel
+/// Publishes an event to a channel.
 ///
-/// @param id The channel id. If 0, the event will be sent to all
-///        channels that have subscribed to the event type
-/// @param name The event name, an arbitrary string
-/// @param args Array with event arguments
+/// @param id Channel id. 0 means "broadcast to all subscribed channels"
+/// @param name Event name (application-defined)
+/// @param args Array of event arguments
 /// @return True if the event was sent successfully, false otherwise.
 bool channel_send_event(uint64_t id, const char *name, Array args)
 {
@@ -215,7 +214,6 @@ bool channel_send_event(uint64_t id, const char *name, Array args)
       send_event(channel, name, args);
     }
   }  else {
-    // TODO(tarruda): Implement event broadcasting in vimscript
     broadcast_event(name, args);
   }
 
-- 
cgit 


From 2a3bcd1ff883429a3dd17e7ae5ddc1396abbad17 Mon Sep 17 00:00:00 2001
From: Björn Linse <bjorn.linse@gmail.com>
Date: Sun, 29 Oct 2017 03:06:53 +0100
Subject: rpc: Don't delay notifications when request is pending (#6544)

With the old behavior, if a GUI makes a blocking request that requires user
interaction (like nvim_input()), it would not get any screen updates.

The client, not nvim, should decide how to handle notifications during a
pending request. If an rplugin wants to avoid async calls while a sync call is
busy, it likely wants to avoid processing async calls while another async call
also is handled as well.

This may break the expectation of some existing rplugins. For compatibility,
remote/define.vim reimplements the old behavior. Clients can opt-out by
specifying `sync=urgent`.

- Legacy hosts should be updated to use `sync=urgent`. They could add a flag
  indicating which async methods are always safe to call and which must wait
  until the main loop returns.
- New hosts can expose the full asyncness, they don't need to offer both
  behaviors.

ref #6532
ref #1398 d83868fe9071af1b4866594eac12f7aa0fa71b53
---
 src/nvim/msgpack_rpc/channel.c | 36 ++----------------------------------
 1 file changed, 2 insertions(+), 34 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 88232a55de..5efdb9a194 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -56,7 +56,6 @@ typedef struct {
 typedef struct {
   uint64_t id;
   size_t refcount;
-  size_t pending_requests;
   PMap(cstr_t) *subscribed_events;
   bool closed;
   ChannelType type;
@@ -71,7 +70,6 @@ typedef struct {
   } data;
   uint64_t next_request_id;
   kvec_t(ChannelCallFrame *) call_stack;
-  kvec_t(WBuffer *) delayed_notifications;
   MultiQueue *events;
 } Channel;
 
@@ -205,14 +203,7 @@ bool channel_send_event(uint64_t id, const char *name, Array args)
   }
 
   if (channel) {
-    if (channel->pending_requests) {
-      // Pending request, queue the notification for later sending.
-      const String method = cstr_as_string((char *)name);
-      WBuffer *buffer = serialize_request(id, 0, method, args, &out_buffer, 1);
-      kv_push(channel->delayed_notifications, buffer);
-    } else {
-      send_event(channel, name, args);
-    }
+    send_event(channel, name, args);
   }  else {
     broadcast_event(name, args);
   }
@@ -248,10 +239,8 @@ Object channel_send_call(uint64_t id,
   // Push the frame
   ChannelCallFrame frame = { request_id, false, false, NIL };
   kv_push(channel->call_stack, &frame);
-  channel->pending_requests++;
   LOOP_PROCESS_EVENTS_UNTIL(&main_loop, channel->events, -1, frame.returned);
   (void)kv_pop(channel->call_stack);
-  channel->pending_requests--;
 
   if (frame.errored) {
     if (frame.result.type == kObjectTypeString) {
@@ -276,10 +265,6 @@ Object channel_send_call(uint64_t id,
     api_free_object(frame.result);
   }
 
-  if (!channel->pending_requests) {
-    send_delayed_notifications(channel);
-  }
-
   decref(channel);
 
   return frame.errored ? NIL : frame.result;
@@ -704,11 +689,7 @@ static void broadcast_event(const char *name, Array args)
 
   for (size_t i = 0; i < kv_size(subscribed); i++) {
     Channel *channel = kv_A(subscribed, i);
-    if (channel->pending_requests) {
-      kv_push(channel->delayed_notifications, buffer);
-    } else {
-      channel_write(channel, buffer);
-    }
+    channel_write(channel, buffer);
   }
 
 end:
@@ -786,7 +767,6 @@ static void free_channel(Channel *channel)
 
   pmap_free(cstr_t)(channel->subscribed_events);
   kv_destroy(channel->call_stack);
-  kv_destroy(channel->delayed_notifications);
   if (channel->type != kChannelTypeProc) {
     multiqueue_free(channel->events);
   }
@@ -811,11 +791,9 @@ static Channel *register_channel(ChannelType type, uint64_t id,
   rv->closed = false;
   rv->unpacker = msgpack_unpacker_new(MSGPACK_UNPACKER_INIT_BUFFER_SIZE);
   rv->id = id > 0 ? id : next_chan_id++;
-  rv->pending_requests = 0;
   rv->subscribed_events = pmap_new(cstr_t)();
   rv->next_request_id = 1;
   kv_init(rv->call_stack);
-  kv_init(rv->delayed_notifications);
   pmap_put(uint64_t)(channels, rv->id, rv);
 
   ILOG("new channel %" PRIu64 " (%s): %s", rv->id,
@@ -912,16 +890,6 @@ static WBuffer *serialize_response(uint64_t channel_id,
   return rv;
 }
 
-static void send_delayed_notifications(Channel* channel)
-{
-  for (size_t i = 0; i < kv_size(channel->delayed_notifications); i++) {
-    WBuffer *buffer = kv_A(channel->delayed_notifications, i);
-    channel_write(channel, buffer);
-  }
-
-  kv_size(channel->delayed_notifications) = 0;
-}
-
 static void incref(Channel *channel)
 {
   channel->refcount++;
-- 
cgit 


From df107149913f82e8b3b6b2060d6dbed3d90e68fe Mon Sep 17 00:00:00 2001
From: Phlosioneer <mattmdrr2@gmail.com>
Date: Sun, 19 Nov 2017 19:55:28 -0500
Subject: server.c: Fix bug in release mode (#7594)

When compiling with CMAKE_BUILD_TYPE=RelWithDebInfo, several
-Wmaybe-uninitialized warnings are printed. These were thought to
be false positives (#5061); there are no control paths that lead
to an uninitialized value. However, when gcc is run in -O2 mode,
it makes a mistake while generating the necessary logic.

Specifically, for the code:
...
  int = 0; // Index of the server whose address equals addr.
  for (; i < watchers.ga_len; i++) {
    watcher = ((SocketWatcher **)watchers.ga_data)[i];
    // <snip>
  }
  if (i >= watchers.ga_len) {
    ELOG("Not listening on %s", addr);
    return;
  }
...

Gcc generates:
...
<+98>:  cmp  %ebx, %ebp
<+100>: jg   0x530f13   <server_stop+55>
<+102>: cmp  %ebp, ebx
<+104>: jl   0x530f7e   <server_stop+162>
...

Normally, the if statement should catch the only control path
where watcher is not assigned: watchers.ga_len <= 0. When
compiled, the assembly lines 98 and 100 correspond to checking if
i < watchers.ga_len, and the lines 102 and 104 correspond to
checking if i >= watchers.ga_len. The assembly seems to compare
ebp (which is watchers.ga_len) with ebx (which is i), and jump
if greater; then do the same comparison and jump if less. This is
where gcc makes a mistake: it flips the order of the cmp
instruction. This means that the REAL behavior is first check if
i < watchers.ga_len and then check if i < watchers.ga_len. Which
means the code inside the if statement is NEVER executed; no
combination of i and watchers.ga_len will ever trigger ELOG().

So not only is this a use of an uninitialized value if
watchers.ga_len == 0 (or technically, if it's less than zero too),
it also clobbers any error detection if the for loop reaches the
last entry (which would normally cause i == watchers.ga_len too).

This commit fixes this issue by adding a bool to keep track of
whether a watcher was found during the loop. This makes gcc
generate the correct code, avoiding both bugs.
---
 src/nvim/msgpack_rpc/server.c | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/server.c b/src/nvim/msgpack_rpc/server.c
index 1e0cc27886..9bf122f4db 100644
--- a/src/nvim/msgpack_rpc/server.c
+++ b/src/nvim/msgpack_rpc/server.c
@@ -180,6 +180,7 @@ int server_start(const char *endpoint)
 void server_stop(char *endpoint)
 {
   SocketWatcher *watcher;
+  bool watcher_found = false;
   char addr[ADDRESS_MAX_SIZE];
 
   // Trim to `ADDRESS_MAX_SIZE`
@@ -189,11 +190,12 @@ void server_stop(char *endpoint)
   for (; i < watchers.ga_len; i++) {
     watcher = ((SocketWatcher **)watchers.ga_data)[i];
     if (strcmp(addr, watcher->addr) == 0) {
+      watcher_found = true;
       break;
     }
   }
 
-  if (i >= watchers.ga_len) {
+  if (!watcher_found) {
     ELOG("Not listening on %s", addr);
     return;
   }
-- 
cgit 


From 3717e2157f2d45ce23dbe4ac03085fea2d956dc4 Mon Sep 17 00:00:00 2001
From: Björn Linse <bjorn.linse@gmail.com>
Date: Sun, 23 Jul 2017 18:04:14 +0200
Subject: Revert channel logging, rebased on new code below

---
 src/nvim/msgpack_rpc/channel.c | 43 ++++++++++++++----------------------------
 1 file changed, 14 insertions(+), 29 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 5efdb9a194..18f6334780 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -115,15 +115,12 @@ void channel_teardown(void)
 /// Creates an API channel by starting a process and connecting to its
 /// stdin/stdout. stderr is handled by the job infrastructure.
 ///
-/// @param proc   process object
-/// @param id     (optional) channel id
-/// @param source description of source function, rplugin name, TCP addr, etc
-///
-/// @return Channel id (> 0), on success. 0, on error.
-uint64_t channel_from_process(Process *proc, uint64_t id, char *source)
+/// @param argv The argument vector for the process. [consumed]
+/// @return The channel id (> 0), on success.
+///         0, on error.
+uint64_t channel_from_process(Process *proc, uint64_t id)
 {
-  Channel *channel = register_channel(kChannelTypeProc, id, proc->events,
-                                      source);
+  Channel *channel = register_channel(kChannelTypeProc, id, proc->events);
   incref(channel);  // process channels are only closed by the exit_cb
   channel->data.proc = proc;
 
@@ -142,8 +139,7 @@ uint64_t channel_from_process(Process *proc, uint64_t id, char *source)
 /// @param watcher The SocketWatcher ready to accept the connection
 void channel_from_connection(SocketWatcher *watcher)
 {
-  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL,
-                                      watcher->addr);
+  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL);
   socket_watcher_accept(watcher, &channel->data.stream);
   incref(channel);  // close channel only after the stream is closed
   channel->data.stream.internal_close_cb = close_cb;
@@ -156,9 +152,8 @@ void channel_from_connection(SocketWatcher *watcher)
        &channel->data.stream);
 }
 
-/// @param source description of source function, rplugin name, TCP addr, etc
-uint64_t channel_connect(bool tcp, const char *address, int timeout,
-                         char *source, const char **error)
+uint64_t channel_connect(bool tcp, const char *address,
+                         int timeout, const char **error)
 {
   if (!tcp) {
     char *path = fix_fname(address);
@@ -170,7 +165,7 @@ uint64_t channel_connect(bool tcp, const char *address, int timeout,
     xfree(path);
   }
 
-  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL, source);
+  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL);
   if (!socket_connect(&main_loop, &channel->data.stream,
                       tcp, address, timeout, error)) {
     decref(channel);
@@ -323,10 +318,11 @@ bool channel_close(uint64_t id)
   return true;
 }
 
-/// Creates an API channel from stdin/stdout. Used to embed Nvim.
+/// Creates an API channel from stdin/stdout. This is used when embedding
+/// Neovim
 void channel_from_stdio(void)
 {
-  Channel *channel = register_channel(kChannelTypeStdio, 0, NULL, NULL);
+  Channel *channel = register_channel(kChannelTypeStdio, 0, NULL);
   incref(channel);  // stdio channels are only closed on exit
   // read stream
   rstream_init_fd(&main_loop, &channel->data.std.in, 0, CHANNEL_BUFFER_SIZE);
@@ -342,7 +338,7 @@ void channel_from_stdio(void)
 /// when an instance connects to its own named pipe.
 uint64_t channel_create_internal(void)
 {
-  Channel *channel = register_channel(kChannelTypeInternal, 0, NULL, NULL);
+  Channel *channel = register_channel(kChannelTypeInternal, 0, NULL);
   incref(channel);  // internal channel lives until process exit
   return channel->id;
 }
@@ -778,12 +774,9 @@ static void close_cb(Stream *stream, void *data)
   decref(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)
+                                 MultiQueue *events)
 {
-  // Jobs and channels share the same id namespace.
-  assert(id == 0 || !pmap_get(uint64_t)(channels, id));
   Channel *rv = xmalloc(sizeof(Channel));
   rv->events = events ? events : multiqueue_new_child(main_loop.events);
   rv->type = type;
@@ -795,14 +788,6 @@ static Channel *register_channel(ChannelType type, uint64_t id,
   rv->next_request_id = 1;
   kv_init(rv->call_stack);
   pmap_put(uint64_t)(channels, rv->id, rv);
-
-  ILOG("new channel %" PRIu64 " (%s): %s", rv->id,
-       (type == kChannelTypeProc ? "proc"
-        : (type == kChannelTypeSocket ? "socket"
-           : (type == kChannelTypeStdio ? "stdio"
-              : (type == kChannelTypeInternal ? "internal" : "?")))),
-       (source ? source : "?"));
-
   return rv;
 }
 
-- 
cgit 


From 5215e3205a07b85e4e4cf1f8a8ca6be2b9556459 Mon Sep 17 00:00:00 2001
From: Björn Linse <bjorn.linse@gmail.com>
Date: Sun, 27 Aug 2017 11:59:33 +0200
Subject: channels: refactor

---
 src/nvim/msgpack_rpc/channel.c      | 388 ++++++++++++------------------------
 src/nvim/msgpack_rpc/channel.h      |   2 +
 src/nvim/msgpack_rpc/channel_defs.h |  36 ++++
 3 files changed, 166 insertions(+), 260 deletions(-)
 create mode 100644 src/nvim/msgpack_rpc/channel_defs.h

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 18f6334780..42b1f63830 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -11,6 +11,7 @@
 #include "nvim/api/private/helpers.h"
 #include "nvim/api/vim.h"
 #include "nvim/api/ui.h"
+#include "nvim/channel.h"
 #include "nvim/msgpack_rpc/channel.h"
 #include "nvim/msgpack_rpc/server.h"
 #include "nvim/event/loop.h"
@@ -40,47 +41,6 @@
 #define log_server_msg(...)
 #endif
 
-typedef enum {
-  kChannelTypeSocket,
-  kChannelTypeProc,
-  kChannelTypeStdio,
-  kChannelTypeInternal
-} ChannelType;
-
-typedef struct {
-  uint64_t request_id;
-  bool returned, errored;
-  Object result;
-} ChannelCallFrame;
-
-typedef struct {
-  uint64_t id;
-  size_t refcount;
-  PMap(cstr_t) *subscribed_events;
-  bool closed;
-  ChannelType type;
-  msgpack_unpacker *unpacker;
-  union {
-    Stream stream;  // bidirectional (socket)
-    Process *proc;
-    struct {
-      Stream in;
-      Stream out;
-    } std;
-  } data;
-  uint64_t next_request_id;
-  kvec_t(ChannelCallFrame *) call_stack;
-  MultiQueue *events;
-} Channel;
-
-typedef struct {
-  Channel *channel;
-  MsgpackRpcRequestHandler handler;
-  Array args;
-  uint64_t request_id;
-} RequestEvent;
-
-static PMap(uint64_t) *channels = NULL;
 static PMap(cstr_t) *event_strings = NULL;
 static msgpack_sbuffer out_buffer;
 
@@ -88,50 +48,32 @@ static msgpack_sbuffer out_buffer;
 # include "msgpack_rpc/channel.c.generated.h"
 #endif
 
-/// Initializes the module
-void channel_init(void)
+void rpc_init(void)
 {
   ch_before_blocking_events = multiqueue_new_child(main_loop.events);
-  channels = pmap_new(uint64_t)();
   event_strings = pmap_new(cstr_t)();
   msgpack_sbuffer_init(&out_buffer);
-  remote_ui_init();
 }
 
-/// Teardown the module
-void channel_teardown(void)
-{
-  if (!channels) {
-    return;
-  }
-
-  Channel *channel;
 
-  map_foreach_value(channels, channel, {
-    close_channel(channel);
-  });
-}
-
-/// Creates an API channel by starting a process and connecting to its
-/// stdin/stdout. stderr is handled by the job infrastructure.
-///
-/// @param argv The argument vector for the process. [consumed]
-/// @return The channel id (> 0), on success.
-///         0, on error.
-uint64_t channel_from_process(Process *proc, uint64_t id)
+void rpc_start(Channel *channel)
 {
-  Channel *channel = register_channel(kChannelTypeProc, id, proc->events);
-  incref(channel);  // process channels are only closed by the exit_cb
-  channel->data.proc = proc;
+  channel->is_rpc = true;
+  RpcState *rpc = &channel->rpc;
+  rpc->closed = false;
+  rpc->unpacker = msgpack_unpacker_new(MSGPACK_UNPACKER_INIT_BUFFER_SIZE);
+  rpc->subscribed_events = pmap_new(cstr_t)();
+  rpc->next_request_id = 1;
+  kv_init(rpc->call_stack);
 
-  wstream_init(proc->in, 0);
-  rstream_init(proc->out, 0);
-  rstream_start(proc->out, receive_msgpack, channel);
+  Stream *in = channel_instream(channel);
+  Stream *out = channel_outstream(channel);
 
-  DLOG("ch %" PRIu64 " in-stream=%p out-stream=%p", channel->id, proc->in,
-       proc->out);
+  DLOG("rpc ch %" PRIu64 " in-stream=%p out-stream=%p", channel->id, in, out);
 
-  return channel->id;
+  wstream_init(in, 0);
+  rstream_init(out, CHANNEL_BUFFER_SIZE);
+  rstream_start(out, receive_msgpack, channel);
 }
 
 /// Creates an API channel from a tcp/pipe socket connection
@@ -139,19 +81,15 @@ uint64_t channel_from_process(Process *proc, uint64_t id)
 /// @param watcher The SocketWatcher ready to accept the connection
 void channel_from_connection(SocketWatcher *watcher)
 {
-  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL);
-  socket_watcher_accept(watcher, &channel->data.stream);
-  incref(channel);  // close channel only after the stream is closed
-  channel->data.stream.internal_close_cb = close_cb;
-  channel->data.stream.internal_data = channel;
-  wstream_init(&channel->data.stream, 0);
-  rstream_init(&channel->data.stream, CHANNEL_BUFFER_SIZE);
-  rstream_start(&channel->data.stream, receive_msgpack, channel);
-
-  DLOG("ch %" PRIu64 " in/out-stream=%p", channel->id,
-       &channel->data.stream);
+  Channel *channel = channel_alloc(kChannelStreamSocket);
+  socket_watcher_accept(watcher, &channel->stream.socket);
+  channel_incref(channel);  // close channel only after the stream is closed
+  channel->stream.socket.internal_close_cb = close_cb;
+  channel->stream.socket.internal_data = channel;
+  rpc_start(channel);
 }
 
+/// TODO: move to eval.c, also support bytes
 uint64_t channel_connect(bool tcp, const char *address,
                          int timeout, const char **error)
 {
@@ -165,34 +103,40 @@ uint64_t channel_connect(bool tcp, const char *address,
     xfree(path);
   }
 
-  Channel *channel = register_channel(kChannelTypeSocket, 0, NULL);
-  if (!socket_connect(&main_loop, &channel->data.stream,
+  Channel *channel = channel_alloc(kChannelStreamSocket);
+  if (!socket_connect(&main_loop, &channel->stream.socket,
                       tcp, address, timeout, error)) {
-    decref(channel);
+    channel_decref(channel);
     return 0;
   }
 
-  incref(channel);  // close channel only after the stream is closed
-  channel->data.stream.internal_close_cb = close_cb;
-  channel->data.stream.internal_data = channel;
-  wstream_init(&channel->data.stream, 0);
-  rstream_init(&channel->data.stream, CHANNEL_BUFFER_SIZE);
-  rstream_start(&channel->data.stream, receive_msgpack, channel);
+  channel_incref(channel);  // close channel only after the stream is closed
+  channel->stream.socket.internal_close_cb = close_cb;
+  channel->stream.socket.internal_data = channel;
+  rpc_start(channel);
   return channel->id;
 }
 
+static Channel *find_rpc_channel(uint64_t id)
+{
+  Channel *chan = find_channel(id);
+  if (!chan || !chan->is_rpc || chan->rpc.closed) {
+    return NULL;
+  }
+  return chan;
+}
+
 /// Publishes an event to a channel.
 ///
 /// @param id Channel id. 0 means "broadcast to all subscribed channels"
 /// @param name Event name (application-defined)
 /// @param args Array of event arguments
 /// @return True if the event was sent successfully, false otherwise.
-bool channel_send_event(uint64_t id, const char *name, Array args)
+bool rpc_send_event(uint64_t id, const char *name, Array args)
 {
   Channel *channel = NULL;
 
-  if (id && (!(channel = pmap_get(uint64_t)(channels, id))
-            || channel->closed)) {
+  if (id && (!(channel = find_rpc_channel(id)))) {
     api_free_array(args);
     return false;
   }
@@ -213,29 +157,30 @@ bool channel_send_event(uint64_t id, const char *name, Array args)
 /// @param args Array with method arguments
 /// @param[out] error True if the return value is an error
 /// @return Whatever the remote method returned
-Object channel_send_call(uint64_t id,
-                         const char *method_name,
-                         Array args,
-                         Error *err)
+Object rpc_send_call(uint64_t id,
+                     const char *method_name,
+                     Array args,
+                     Error *err)
 {
   Channel *channel = NULL;
 
-  if (!(channel = pmap_get(uint64_t)(channels, id)) || channel->closed) {
+  if (!(channel = find_rpc_channel(id))) {
     api_set_error(err, kErrorTypeException, "Invalid channel: %" PRIu64, id);
     api_free_array(args);
     return NIL;
   }
 
-  incref(channel);
-  uint64_t request_id = channel->next_request_id++;
+  channel_incref(channel);
+  RpcState *rpc = &channel->rpc;
+  uint64_t request_id = rpc->next_request_id++;
   // Send the msgpack-rpc request
   send_request(channel, request_id, method_name, args);
 
   // Push the frame
   ChannelCallFrame frame = { request_id, false, false, NIL };
-  kv_push(channel->call_stack, &frame);
+  kv_push(rpc->call_stack, &frame);
   LOOP_PROCESS_EVENTS_UNTIL(&main_loop, channel->events, -1, frame.returned);
-  (void)kv_pop(channel->call_stack);
+  (void)kv_pop(rpc->call_stack);
 
   if (frame.errored) {
     if (frame.result.type == kObjectTypeString) {
@@ -260,7 +205,7 @@ Object channel_send_call(uint64_t id,
     api_free_object(frame.result);
   }
 
-  decref(channel);
+  channel_decref(channel);
 
   return frame.errored ? NIL : frame.result;
 }
@@ -269,11 +214,11 @@ Object channel_send_call(uint64_t id,
 ///
 /// @param id The channel id
 /// @param event The event type string
-void channel_subscribe(uint64_t id, char *event)
+void rpc_subscribe(uint64_t id, char *event)
 {
   Channel *channel;
 
-  if (!(channel = pmap_get(uint64_t)(channels, id)) || channel->closed) {
+  if (!(channel = find_rpc_channel(id))) {
     abort();
   }
 
@@ -284,18 +229,18 @@ void channel_subscribe(uint64_t id, char *event)
     pmap_put(cstr_t)(event_strings, event_string, event_string);
   }
 
-  pmap_put(cstr_t)(channel->subscribed_events, event_string, event_string);
+  pmap_put(cstr_t)(channel->rpc.subscribed_events, event_string, event_string);
 }
 
 /// Unsubscribes to event broadcasts
 ///
 /// @param id The channel id
 /// @param event The event type string
-void channel_unsubscribe(uint64_t id, char *event)
+void rpc_unsubscribe(uint64_t id, char *event)
 {
   Channel *channel;
 
-  if (!(channel = pmap_get(uint64_t)(channels, id)) || channel->closed) {
+  if (!(channel = find_rpc_channel(id))) {
     abort();
   }
 
@@ -310,7 +255,7 @@ bool channel_close(uint64_t id)
 {
   Channel *channel;
 
-  if (!(channel = pmap_get(uint64_t)(channels, id)) || channel->closed) {
+  if (!(channel = find_rpc_channel(id))) {
     return false;
   }
 
@@ -322,24 +267,22 @@ bool channel_close(uint64_t id)
 /// Neovim
 void channel_from_stdio(void)
 {
-  Channel *channel = register_channel(kChannelTypeStdio, 0, NULL);
-  incref(channel);  // stdio channels are only closed on exit
+  Channel *channel = channel_alloc(kChannelStreamStdio);
+  channel_incref(channel);  // stdio channels are only closed on exit
   // read stream
-  rstream_init_fd(&main_loop, &channel->data.std.in, 0, CHANNEL_BUFFER_SIZE);
-  rstream_start(&channel->data.std.in, receive_msgpack, channel);
-  // write stream
-  wstream_init_fd(&main_loop, &channel->data.std.out, 1, 0);
+  rstream_init_fd(&main_loop, &channel->stream.stdio.in, 0, CHANNEL_BUFFER_SIZE);
+  wstream_init_fd(&main_loop, &channel->stream.stdio.out, 1, 0);
 
-  DLOG("ch %" PRIu64 " in-stream=%p out-stream=%p", channel->id,
-       &channel->data.std.in, &channel->data.std.out);
+  rpc_start(channel);
 }
 
 /// Creates a loopback channel. This is used to avoid deadlock
 /// when an instance connects to its own named pipe.
 uint64_t channel_create_internal(void)
 {
-  Channel *channel = register_channel(kChannelTypeInternal, 0, NULL);
-  incref(channel);  // internal channel lives until process exit
+  Channel *channel = channel_alloc(kChannelStreamInternal);
+  channel_incref(channel);  // internal channel lives until process exit
+  rpc_start(channel);
   return channel->id;
 }
 
@@ -347,8 +290,8 @@ void channel_process_exit(uint64_t id, int status)
 {
   Channel *channel = pmap_get(uint64_t)(channels, id);
 
-  channel->closed = true;
-  decref(channel);
+  // channel_decref(channel); remove??
+  channel->rpc.closed = true;
 }
 
 // rstream.c:read_event() invokes this as stream->read_cb().
@@ -356,7 +299,7 @@ static void receive_msgpack(Stream *stream, RBuffer *rbuf, size_t c,
                             void *data, bool eof)
 {
   Channel *channel = data;
-  incref(channel);
+  channel_incref(channel);
 
   if (eof) {
     close_channel(channel);
@@ -367,30 +310,19 @@ 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",
+  DLOG("ch %" PRIu64 ": parsing %zu bytes from msgpack Stream: %p",
        channel->id, count, stream);
 
   // Feed the unpacker with data
-  msgpack_unpacker_reserve_buffer(channel->unpacker, count);
-  rbuffer_read(rbuf, msgpack_unpacker_buffer(channel->unpacker), count);
-  msgpack_unpacker_buffer_consumed(channel->unpacker, count);
+  msgpack_unpacker_reserve_buffer(channel->rpc.unpacker, count);
+  rbuffer_read(rbuf, msgpack_unpacker_buffer(channel->rpc.unpacker), count);
+  msgpack_unpacker_buffer_consumed(channel->rpc.unpacker, count);
 
   parse_msgpack(channel);
 
 end:
-  decref(channel);
+  channel_decref(channel);
 }
 
 static void parse_msgpack(Channel *channel)
@@ -400,8 +332,8 @@ static void parse_msgpack(Channel *channel)
   msgpack_unpack_return result;
 
   // Deserialize everything we can.
-  while ((result = msgpack_unpacker_next(channel->unpacker, &unpacked)) ==
-      MSGPACK_UNPACK_SUCCESS) {
+  while ((result = msgpack_unpacker_next(channel->rpc.unpacker, &unpacked)) ==
+         MSGPACK_UNPACK_SUCCESS) {
     bool is_response = is_rpc_response(&unpacked.data);
     log_client_msg(channel->id, !is_response, unpacked.data);
 
@@ -427,7 +359,7 @@ static void parse_msgpack(Channel *channel)
   if (result == MSGPACK_UNPACK_NOMEM_ERROR) {
     mch_errmsg(e_outofmem);
     mch_errmsg("\n");
-    decref(channel);
+    channel_decref(channel);
     preserve_exit();
   }
 
@@ -492,7 +424,7 @@ static void handle_request(Channel *channel, msgpack_object *request)
   evdata->handler = handler;
   evdata->args = args;
   evdata->request_id = request_id;
-  incref(channel);
+  channel_incref(channel);
   if (handler.async) {
     bool is_get_mode = handler.fn == handle_nvim_get_mode;
 
@@ -530,66 +462,30 @@ static void on_request_event(void **argv)
     api_free_object(result);
   }
   api_free_array(args);
-  decref(channel);
+  channel_decref(channel);
   xfree(e);
   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;
+  bool success;
 
-  if (channel->closed) {
+  if (channel->rpc.closed) {
     wstream_release_wbuffer(buffer);
     return false;
   }
 
-  switch (channel->type) {
-    case kChannelTypeSocket:
-    case kChannelTypeProc:
-    case kChannelTypeStdio:
-      success = wstream_write(chan_wstream(channel), buffer);
-      break;
-    case kChannelTypeInternal:
-      incref(channel);
-      CREATE_EVENT(channel->events, internal_read_event, 2, channel, buffer);
-      success = true;
-      break;
+  if (channel->streamtype == kChannelStreamInternal) {
+    channel_incref(channel);
+    CREATE_EVENT(channel->events, internal_read_event, 2, channel, buffer);
+    success = true;
+  } else {
+    Stream *in = channel_instream(channel);
+    success = wstream_write(in, buffer);
   }
 
+
   if (!success) {
     // If the write failed for any reason, close the channel
     char buf[256];
@@ -609,14 +505,14 @@ static void internal_read_event(void **argv)
   Channel *channel = argv[0];
   WBuffer *buffer = argv[1];
 
-  msgpack_unpacker_reserve_buffer(channel->unpacker, buffer->size);
-  memcpy(msgpack_unpacker_buffer(channel->unpacker),
+  msgpack_unpacker_reserve_buffer(channel->rpc.unpacker, buffer->size);
+  memcpy(msgpack_unpacker_buffer(channel->rpc.unpacker),
          buffer->data, buffer->size);
-  msgpack_unpacker_buffer_consumed(channel->unpacker, buffer->size);
+  msgpack_unpacker_buffer_consumed(channel->rpc.unpacker, buffer->size);
 
   parse_msgpack(channel);
 
-  decref(channel);
+  channel_decref(channel);
   wstream_release_wbuffer(buffer);
 }
 
@@ -665,7 +561,8 @@ static void broadcast_event(const char *name, Array args)
   Channel *channel;
 
   map_foreach_value(channels, channel, {
-    if (pmap_has(cstr_t)(channel->subscribed_events, name)) {
+    if (channel->is_rpc
+        && pmap_has(cstr_t)(channel->rpc.subscribed_events, name)) {
       kv_push(subscribed, channel);
     }
   });
@@ -695,10 +592,11 @@ end:
 static void unsubscribe(Channel *channel, char *event)
 {
   char *event_string = pmap_get(cstr_t)(event_strings, event);
-  pmap_del(cstr_t)(channel->subscribed_events, event_string);
+  pmap_del(cstr_t)(channel->rpc.subscribed_events, event_string);
 
   map_foreach_value(channels, channel, {
-    if (pmap_has(cstr_t)(channel->subscribed_events, event_string)) {
+    if (channel->is_rpc
+        && pmap_has(cstr_t)(channel->rpc.subscribed_events, event_string)) {
       return;
     }
   });
@@ -709,86 +607,65 @@ static void unsubscribe(Channel *channel, char *event)
 }
 
 /// Close the channel streams/process and free the channel resources.
+/// TODO: move to channel.h
 static void close_channel(Channel *channel)
 {
-  if (channel->closed) {
+  if (channel->rpc.closed) {
     return;
   }
 
-  channel->closed = true;
+  channel->rpc.closed = true;
 
-  switch (channel->type) {
-    case kChannelTypeSocket:
-      stream_close(&channel->data.stream, NULL, NULL);
+  switch (channel->streamtype) {
+    case kChannelStreamSocket:
+      stream_close(&channel->stream.socket, NULL, NULL);
       break;
-    case kChannelTypeProc:
+    case kChannelStreamProc:
       // Only close the rpc channel part,
       // there could be an error message on the stderr stream
-      process_close_in(channel->data.proc);
-      process_close_out(channel->data.proc);
+      process_close_in(&channel->stream.proc);
+      process_close_out(&channel->stream.proc);
       break;
-    case kChannelTypeStdio:
-      stream_close(&channel->data.std.in, NULL, NULL);
-      stream_close(&channel->data.std.out, NULL, NULL);
+    case kChannelStreamStdio:
+      stream_close(&channel->stream.stdio.in, NULL, NULL);
+      stream_close(&channel->stream.stdio.out, NULL, NULL);
       multiqueue_put(main_loop.fast_events, exit_event, 1, channel);
       return;
-    case kChannelTypeInternal:
+    case kChannelStreamInternal:
       // nothing to free.
       break;
   }
 
-  decref(channel);
+  channel_decref(channel);
 }
 
 static void exit_event(void **argv)
 {
-  decref(argv[0]);
+  channel_decref(argv[0]);
 
   if (!exiting) {
     mch_exit(0);
   }
 }
 
-static void free_channel(Channel *channel)
+void rpc_free(Channel *channel)
 {
   remote_ui_disconnect(channel->id);
-  pmap_del(uint64_t)(channels, channel->id);
-  msgpack_unpacker_free(channel->unpacker);
+  msgpack_unpacker_free(channel->rpc.unpacker);
 
   // Unsubscribe from all events
   char *event_string;
-  map_foreach_value(channel->subscribed_events, event_string, {
+  map_foreach_value(channel->rpc.subscribed_events, event_string, {
     unsubscribe(channel, event_string);
   });
 
-  pmap_free(cstr_t)(channel->subscribed_events);
-  kv_destroy(channel->call_stack);
-  if (channel->type != kChannelTypeProc) {
-    multiqueue_free(channel->events);
-  }
-  xfree(channel);
+  pmap_free(cstr_t)(channel->rpc.subscribed_events);
+  kv_destroy(channel->rpc.call_stack);
 }
 
 static void close_cb(Stream *stream, void *data)
 {
-  decref(data);
-}
-
-static Channel *register_channel(ChannelType type, uint64_t id,
-                                 MultiQueue *events)
-{
-  Channel *rv = xmalloc(sizeof(Channel));
-  rv->events = events ? events : multiqueue_new_child(main_loop.events);
-  rv->type = type;
-  rv->refcount = 1;
-  rv->closed = false;
-  rv->unpacker = msgpack_unpacker_new(MSGPACK_UNPACKER_INIT_BUFFER_SIZE);
-  rv->id = id > 0 ? id : next_chan_id++;
-  rv->subscribed_events = pmap_new(cstr_t)();
-  rv->next_request_id = 1;
-  kv_init(rv->call_stack);
-  pmap_put(uint64_t)(channels, rv->id, rv);
-  return rv;
+  channel_decref(data);
 }
 
 static bool is_rpc_response(msgpack_object *obj)
@@ -803,15 +680,18 @@ static bool is_rpc_response(msgpack_object *obj)
 static bool is_valid_rpc_response(msgpack_object *obj, Channel *channel)
 {
   uint64_t response_id = obj->via.array.ptr[1].via.u64;
+  if (kv_size(channel->rpc.call_stack) == 0) {
+    return false;
+  }
+
   // Must be equal to the frame at the stack's bottom
-  return kv_size(channel->call_stack) && response_id
-    == kv_A(channel->call_stack, kv_size(channel->call_stack) - 1)->request_id;
+  ChannelCallFrame *frame = kv_last(channel->rpc.call_stack);
+  return response_id == frame->request_id;
 }
 
 static void complete_call(msgpack_object *obj, Channel *channel)
 {
-  ChannelCallFrame *frame = kv_A(channel->call_stack,
-                             kv_size(channel->call_stack) - 1);
+  ChannelCallFrame *frame = kv_last(channel->rpc.call_stack);
   frame->returned = true;
   frame->errored = obj->via.array.ptr[2].type != MSGPACK_OBJECT_NIL;
 
@@ -825,8 +705,8 @@ static void complete_call(msgpack_object *obj, Channel *channel)
 static void call_set_error(Channel *channel, char *msg, int loglevel)
 {
   LOG(loglevel, "RPC: %s", msg);
-  for (size_t i = 0; i < kv_size(channel->call_stack); i++) {
-    ChannelCallFrame *frame = kv_A(channel->call_stack, i);
+  for (size_t i = 0; i < kv_size(channel->rpc.call_stack); i++) {
+    ChannelCallFrame *frame = kv_A(channel->rpc.call_stack, i);
     frame->returned = true;
     frame->errored = true;
     api_free_object(frame->result);
@@ -875,18 +755,6 @@ static WBuffer *serialize_response(uint64_t channel_id,
   return rv;
 }
 
-static void incref(Channel *channel)
-{
-  channel->refcount++;
-}
-
-static void decref(Channel *channel)
-{
-  if (!(--channel->refcount)) {
-    free_channel(channel);
-  }
-}
-
 #if MIN_LOG_LEVEL <= DEBUG_LOG_LEVEL
 #define REQ "[request]  "
 #define RES "[response] "
diff --git a/src/nvim/msgpack_rpc/channel.h b/src/nvim/msgpack_rpc/channel.h
index f8fe6f129b..9ff5abdc5f 100644
--- a/src/nvim/msgpack_rpc/channel.h
+++ b/src/nvim/msgpack_rpc/channel.h
@@ -8,6 +8,7 @@
 #include "nvim/event/socket.h"
 #include "nvim/event/process.h"
 #include "nvim/vim.h"
+#include "nvim/channel.h"
 
 #define METHOD_MAXLEN 512
 
@@ -16,6 +17,7 @@
 ///       of os_inchar(), so they are processed "just-in-time".
 MultiQueue *ch_before_blocking_events;
 
+
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "msgpack_rpc/channel.h.generated.h"
 #endif
diff --git a/src/nvim/msgpack_rpc/channel_defs.h b/src/nvim/msgpack_rpc/channel_defs.h
new file mode 100644
index 0000000000..6d8362e8b7
--- /dev/null
+++ b/src/nvim/msgpack_rpc/channel_defs.h
@@ -0,0 +1,36 @@
+#ifndef NVIM_MSGPACK_RPC_CHANNEL_DEFS_H
+#define NVIM_MSGPACK_RPC_CHANNEL_DEFS_H
+
+#include <stdbool.h>
+#include <uv.h>
+#include <msgpack.h>
+
+#include "nvim/api/private/defs.h"
+#include "nvim/event/socket.h"
+#include "nvim/event/process.h"
+#include "nvim/vim.h"
+
+typedef struct Channel Channel;
+
+typedef struct {
+  uint64_t request_id;
+  bool returned, errored;
+  Object result;
+} ChannelCallFrame;
+
+typedef struct {
+  Channel *channel;
+  MsgpackRpcRequestHandler handler;
+  Array args;
+  uint64_t request_id;
+} RequestEvent;
+
+typedef struct {
+  PMap(cstr_t) *subscribed_events;
+  bool closed;
+  msgpack_unpacker *unpacker;
+  uint64_t next_request_id;
+  kvec_t(ChannelCallFrame *) call_stack;
+} RpcState;
+
+#endif  // NVIM_MSGPACK_RPC_CHANNEL_DEFS_H
-- 
cgit 


From 1ebc96fe10fbdbec22caa26d5d52a9f095da9687 Mon Sep 17 00:00:00 2001
From: Björn Linse <bjorn.linse@gmail.com>
Date: Mon, 5 Jun 2017 08:29:10 +0200
Subject: channels: allow bytes sockets and stdio, and buffered bytes output

---
 src/nvim/msgpack_rpc/channel.c | 87 ++++--------------------------------------
 1 file changed, 8 insertions(+), 79 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 42b1f63830..56af4fa791 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -13,7 +13,6 @@
 #include "nvim/api/ui.h"
 #include "nvim/channel.h"
 #include "nvim/msgpack_rpc/channel.h"
-#include "nvim/msgpack_rpc/server.h"
 #include "nvim/event/loop.h"
 #include "nvim/event/libuv_process.h"
 #include "nvim/event/rstream.h"
@@ -30,12 +29,9 @@
 #include "nvim/map.h"
 #include "nvim/log.h"
 #include "nvim/misc1.h"
-#include "nvim/path.h"
 #include "nvim/lib/kvec.h"
 #include "nvim/os/input.h"
 
-#define CHANNEL_BUFFER_SIZE 0xffff
-
 #if MIN_LOG_LEVEL > DEBUG_LOG_LEVEL
 #define log_client_msg(...)
 #define log_server_msg(...)
@@ -66,57 +62,18 @@ void rpc_start(Channel *channel)
   rpc->next_request_id = 1;
   kv_init(rpc->call_stack);
 
-  Stream *in = channel_instream(channel);
-  Stream *out = channel_outstream(channel);
-
-  DLOG("rpc ch %" PRIu64 " in-stream=%p out-stream=%p", channel->id, in, out);
-
-  wstream_init(in, 0);
-  rstream_init(out, CHANNEL_BUFFER_SIZE);
-  rstream_start(out, receive_msgpack, channel);
-}
-
-/// Creates an API channel from a tcp/pipe socket connection
-///
-/// @param watcher The SocketWatcher ready to accept the connection
-void channel_from_connection(SocketWatcher *watcher)
-{
-  Channel *channel = channel_alloc(kChannelStreamSocket);
-  socket_watcher_accept(watcher, &channel->stream.socket);
-  channel_incref(channel);  // close channel only after the stream is closed
-  channel->stream.socket.internal_close_cb = close_cb;
-  channel->stream.socket.internal_data = channel;
-  rpc_start(channel);
-}
-
-/// TODO: move to eval.c, also support bytes
-uint64_t channel_connect(bool tcp, const char *address,
-                         int timeout, const char **error)
-{
-  if (!tcp) {
-    char *path = fix_fname(address);
-    if (server_owns_pipe_address(path)) {
-      // avoid deadlock
-      xfree(path);
-      return channel_create_internal();
-    }
-    xfree(path);
-  }
+  if (channel->streamtype != kChannelStreamInternal) {
+    Stream *out = channel_outstream(channel);
+#if MIN_LOG_LEVEL <= DEBUG_LOG_LEVEL
+    Stream *in = channel_instream(channel);
+    DLOG("rpc ch %" PRIu64 " in-stream=%p out-stream=%p", channel->id, in, out);
+#endif
 
-  Channel *channel = channel_alloc(kChannelStreamSocket);
-  if (!socket_connect(&main_loop, &channel->stream.socket,
-                      tcp, address, timeout, error)) {
-    channel_decref(channel);
-    return 0;
+    rstream_start(out, receive_msgpack, channel);
   }
-
-  channel_incref(channel);  // close channel only after the stream is closed
-  channel->stream.socket.internal_close_cb = close_cb;
-  channel->stream.socket.internal_data = channel;
-  rpc_start(channel);
-  return channel->id;
 }
 
+
 static Channel *find_rpc_channel(uint64_t id)
 {
   Channel *chan = find_channel(id);
@@ -263,29 +220,6 @@ bool channel_close(uint64_t id)
   return true;
 }
 
-/// Creates an API channel from stdin/stdout. This is used when embedding
-/// Neovim
-void channel_from_stdio(void)
-{
-  Channel *channel = channel_alloc(kChannelStreamStdio);
-  channel_incref(channel);  // stdio channels are only closed on exit
-  // read stream
-  rstream_init_fd(&main_loop, &channel->stream.stdio.in, 0, CHANNEL_BUFFER_SIZE);
-  wstream_init_fd(&main_loop, &channel->stream.stdio.out, 1, 0);
-
-  rpc_start(channel);
-}
-
-/// Creates a loopback channel. This is used to avoid deadlock
-/// when an instance connects to its own named pipe.
-uint64_t channel_create_internal(void)
-{
-  Channel *channel = channel_alloc(kChannelStreamInternal);
-  channel_incref(channel);  // internal channel lives until process exit
-  rpc_start(channel);
-  return channel->id;
-}
-
 void channel_process_exit(uint64_t id, int status)
 {
   Channel *channel = pmap_get(uint64_t)(channels, id);
@@ -663,11 +597,6 @@ void rpc_free(Channel *channel)
   kv_destroy(channel->rpc.call_stack);
 }
 
-static void close_cb(Stream *stream, void *data)
-{
-  channel_decref(data);
-}
-
 static bool is_rpc_response(msgpack_object *obj)
 {
   return obj->type == MSGPACK_OBJECT_ARRAY
-- 
cgit 


From 90e5cc5484ceeb410ae2a2706e09ed475cade4a5 Mon Sep 17 00:00:00 2001
From: Björn Linse <bjorn.linse@gmail.com>
Date: Thu, 8 Jun 2017 17:15:53 +0200
Subject: channels: generalize jobclose()

---
 src/nvim/msgpack_rpc/channel.c | 62 +++++++-----------------------------------
 1 file changed, 10 insertions(+), 52 deletions(-)

(limited to 'src/nvim/msgpack_rpc')

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 56af4fa791..32781cf4d9 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -54,6 +54,7 @@ void rpc_init(void)
 
 void rpc_start(Channel *channel)
 {
+  channel_incref(channel);
   channel->is_rpc = true;
   RpcState *rpc = &channel->rpc;
   rpc->closed = false;
@@ -204,31 +205,6 @@ void rpc_unsubscribe(uint64_t id, char *event)
   unsubscribe(channel, event);
 }
 
-/// Closes a channel
-///
-/// @param id The channel id
-/// @return true if successful, false otherwise
-bool channel_close(uint64_t id)
-{
-  Channel *channel;
-
-  if (!(channel = find_rpc_channel(id))) {
-    return false;
-  }
-
-  close_channel(channel);
-  return true;
-}
-
-void channel_process_exit(uint64_t id, int status)
-{
-  Channel *channel = pmap_get(uint64_t)(channels, id);
-
-  // channel_decref(channel); remove??
-  channel->rpc.closed = true;
-}
-
-// rstream.c:read_event() invokes this as stream->read_cb().
 static void receive_msgpack(Stream *stream, RBuffer *rbuf, size_t c,
                             void *data, bool eof)
 {
@@ -236,7 +212,7 @@ static void receive_msgpack(Stream *stream, RBuffer *rbuf, size_t c,
   channel_incref(channel);
 
   if (eof) {
-    close_channel(channel);
+    channel_close(channel->id, kChannelPartRpc, NULL);
     char buf[256];
     snprintf(buf, sizeof(buf), "ch %" PRIu64 " was closed by the client",
              channel->id);
@@ -540,43 +516,25 @@ static void unsubscribe(Channel *channel, char *event)
   xfree(event_string);
 }
 
-/// Close the channel streams/process and free the channel resources.
-/// TODO: move to channel.h
-static void close_channel(Channel *channel)
+
+/// Mark rpc state as closed, and release its reference to the channel.
+/// Don't call this directly, call channel_close(id, kChannelPartRpc, &error)
+void rpc_close(Channel *channel)
 {
   if (channel->rpc.closed) {
     return;
   }
 
   channel->rpc.closed = true;
+  channel_decref(channel);
 
-  switch (channel->streamtype) {
-    case kChannelStreamSocket:
-      stream_close(&channel->stream.socket, NULL, NULL);
-      break;
-    case kChannelStreamProc:
-      // Only close the rpc channel part,
-      // there could be an error message on the stderr stream
-      process_close_in(&channel->stream.proc);
-      process_close_out(&channel->stream.proc);
-      break;
-    case kChannelStreamStdio:
-      stream_close(&channel->stream.stdio.in, NULL, NULL);
-      stream_close(&channel->stream.stdio.out, NULL, NULL);
-      multiqueue_put(main_loop.fast_events, exit_event, 1, channel);
-      return;
-    case kChannelStreamInternal:
-      // nothing to free.
-      break;
+  if (channel->streamtype == kChannelStreamStdio) {
+    multiqueue_put(main_loop.fast_events, exit_event, 0);
   }
-
-  channel_decref(channel);
 }
 
 static void exit_event(void **argv)
 {
-  channel_decref(argv[0]);
-
   if (!exiting) {
     mch_exit(0);
   }
@@ -642,7 +600,7 @@ static void call_set_error(Channel *channel, char *msg, int loglevel)
     frame->result = STRING_OBJ(cstr_to_string(msg));
   }
 
-  close_channel(channel);
+  channel_close(channel->id, kChannelPartRpc, NULL);
 }
 
 static WBuffer *serialize_request(uint64_t channel_id,
-- 
cgit