26 files changed, 855 insertions, 313 deletions

diff --git a/src/nvim/CMakeLists.txt b/src/nvim/CMakeLists.txt
index 2e45b5e9d6..880ff42ba5 100644
--- a/src/nvim/CMakeLists.txt
+++ b/src/nvim/CMakeLists.txt
@@ -40,6 +40,7 @@ file(MAKE_DIRECTORY ${GENERATED_INCLUDES_DIR}/tui)
 file(GLOB NEOVIM_SOURCES *.c os/*.c api/*.c api/private/*.c msgpack_rpc/*.c
   tui/*.c)
 file(GLOB_RECURSE NEOVIM_HEADERS *.h)
+file(GLOB UNIT_TEST_FIXTURES ${PROJECT_SOURCE_DIR}/test/unit/fixtures/*.c)
 
 foreach(sfile ${NEOVIM_SOURCES})
   get_filename_component(f ${sfile} NAME)
@@ -205,7 +206,7 @@ set_target_properties(libnvim PROPERTIES
 set_property(TARGET libnvim APPEND_STRING PROPERTY COMPILE_FLAGS " -DMAKE_LIB ")
 
 add_library(nvim-test MODULE EXCLUDE_FROM_ALL ${NEOVIM_GENERATED_SOURCES}
-    ${NEOVIM_SOURCES} ${NEOVIM_HEADERS})
+    ${NEOVIM_SOURCES} ${UNIT_TEST_FIXTURES} ${NEOVIM_HEADERS})
 target_link_libraries(nvim-test ${NVIM_LINK_LIBRARIES})
 set_property(TARGET nvim-test APPEND_STRING PROPERTY COMPILE_FLAGS -DUNIT_TESTING)
 
diff --git a/src/nvim/eval.c b/src/nvim/eval.c
index 93590445c9..d3ab47a505 100644
--- a/src/nvim/eval.c
+++ b/src/nvim/eval.c
@@ -20341,19 +20341,19 @@ static inline void push_job_event(Job *job, ufunc_T *callback,
   }, !disable_job_defer);
 }
 
-static void on_job_stdout(RStream *rstream, void *job, bool eof)
+static void on_job_stdout(RStream *rstream, RBuffer *buf, void *job, bool eof)
 {
   TerminalJobData *data = job_data(job);
-  on_job_output(rstream, job, eof, data->on_stdout, "stdout");
+  on_job_output(rstream, job, buf, eof, data->on_stdout, "stdout");
 }
 
-static void on_job_stderr(RStream *rstream, void *job, bool eof)
+static void on_job_stderr(RStream *rstream, RBuffer *buf, void *job, bool eof)
 {
   TerminalJobData *data = job_data(job);
-  on_job_output(rstream, job, eof, data->on_stderr, "stderr");
+  on_job_output(rstream, job, buf, eof, data->on_stderr, "stderr");
 }
 
-static void on_job_output(RStream *rstream, Job *job, bool eof,
+static void on_job_output(RStream *rstream, Job *job, RBuffer *buf, bool eof,
     ufunc_T *callback, const char *type)
 {
   if (eof) {
@@ -20361,20 +20361,19 @@ static void on_job_output(RStream *rstream, Job *job, bool eof,
   }
 
   TerminalJobData *data = job_data(job);
-  char *ptr = rstream_read_ptr(rstream);
-  size_t len = rstream_pending(rstream);
+  RBUFFER_UNTIL_EMPTY(buf, ptr, len) {
+    // The order here matters, the terminal must receive the data first because
+    // push_job_event will modify the read buffer(convert NULs into NLs)
+    if (data->term) {
+      terminal_receive(data->term, ptr, len);
+    }
 
-  // The order here matters, the terminal must receive the data first because
-  // push_job_event will modify the read buffer(convert NULs into NLs)
-  if (data->term) {
-    terminal_receive(data->term, ptr, len);
-  }
+    if (callback) {
+      push_job_event(job, callback, type, ptr, len, 0);
+    }
 
-  if (callback) {
-    push_job_event(job, callback, type, ptr, len, 0);
+    rbuffer_consumed(buf, len);
   }
-
-  rbuffer_consumed(rstream_buffer(rstream), len);
 }
 
 static void on_job_exit(Job *job, int status, void *d)
diff --git a/src/nvim/lib/klist.h b/src/nvim/lib/klist.h
index 7df809f07b..10d6846133 100644
--- a/src/nvim/lib/klist.h
+++ b/src/nvim/lib/klist.h
@@ -27,10 +27,12 @@
 #define _AC_KLIST_H
 
 #include <stdlib.h>
+#include <assert.h>
 
 #include "nvim/memory.h"
 #include "nvim/func_attr.h"
 
+
 #define KMEMPOOL_INIT(name, kmptype_t, kmpfree_f)                       \
     typedef struct {                                                    \
         size_t cnt, n, max;                                             \
@@ -95,23 +97,27 @@
             kmp_free(name, kl->mp, p);                                  \
         kmp_free(name, kl->mp, p);                                      \
         kmp_destroy(name, kl->mp);                                      \
-        xfree(kl);                                                       \
+        xfree(kl);                                                      \
     }                                                                   \
-    static inline kltype_t *kl_pushp_##name(kl_##name##_t *kl) {        \
+    static inline void kl_push_##name(kl_##name##_t *kl, kltype_t d) {  \
         kl1_##name *q, *p = kmp_alloc(name, kl->mp);                    \
         q = kl->tail; p->next = 0; kl->tail->next = p; kl->tail = p;    \
         ++kl->size;                                                     \
-        return &q->data;                                                \
+        q->data = d;                                                    \
     }                                                                   \
-    static inline int kl_shift_##name(kl_##name##_t *kl, kltype_t *d) { \
+    static inline kltype_t kl_shift_at_##name(kl_##name##_t *kl,        \
+                                              kl1_##name **n) {         \
+        assert((*n)->next);                                             \
         kl1_##name *p;                                                  \
-        if (kl->head->next == 0) return -1;                             \
         --kl->size;                                                     \
-        p = kl->head; kl->head = kl->head->next;                        \
-        if (d) *d = p->data;                                            \
+        p = *n;                                                         \
+        *n = (*n)->next;                                                \
+        if (p == kl->head) kl->head = *n;                               \
+        kltype_t d = p->data;                                           \
         kmp_free(name, kl->mp, p);                                      \
-        return 0;                                                       \
-    }
+        return d;                                                       \
+    }                                                                   \
+    
 
 #define kliter_t(name) kl1_##name
 #define klist_t(name) kl_##name##_t
@@ -122,7 +128,14 @@
 
 #define kl_init(name) kl_init_##name()
 #define kl_destroy(name, kl) kl_destroy_##name(kl)
-#define kl_pushp(name, kl) kl_pushp_##name(kl)
-#define kl_shift(name, kl, d) kl_shift_##name(kl, d)
+#define kl_push(name, kl, d) kl_push_##name(kl, d)
+#define kl_shift_at(name, kl, node) kl_shift_at_##name(kl, node)
+#define kl_shift(name, kl) kl_shift_at(name, kl, &kl->head)
 #define kl_empty(kl) ((kl)->size == 0)
+// Iteration macros. It's ok to modify the list while iterating as long as a
+// `break` statement is executed before the next iteration.
+#define kl_iter(name, kl, p) kl_iter_at(name, kl, p, NULL)
+#define kl_iter_at(name, kl, p, h) \
+  for (kl1_##name *p = h ? h : kl->head; p != kl->tail; p = p->next)
+
 #endif
diff --git a/src/nvim/main.c b/src/nvim/main.c
index e1bb2d0b66..50c16c51d6 100644
--- a/src/nvim/main.c
+++ b/src/nvim/main.c
@@ -138,6 +138,7 @@ static const char *err_extra_cmd =
 /// Needed for unit tests. Must be called after `time_init()`.
 void early_init(void)
 {
+  fs_init();
   handle_init();
 
   (void)mb_init();      // init mb_bytelen_tab[] to ones
diff --git a/src/nvim/memory.h b/src/nvim/memory.h
index 4ff31ff732..7b477da2f5 100644
--- a/src/nvim/memory.h
+++ b/src/nvim/memory.h
@@ -1,6 +1,7 @@
 #ifndef NVIM_MEMORY_H
 #define NVIM_MEMORY_H
 
+#include <stdint.h>  // for uint8_t
 #include <stddef.h>  // for size_t
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index df78f822d6..2a81b4f160 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -328,19 +328,17 @@ static void channel_from_stdio(void)
   channel->data.streams.uv = NULL;
 }
 
-static void job_out(RStream *rstream, void *data, bool eof)
+static void job_out(RStream *rstream, RBuffer *buf, void *data, bool eof)
 {
   Job *job = data;
-  parse_msgpack(rstream, job_data(job), eof);
+  parse_msgpack(rstream, buf, job_data(job), eof);
 }
 
-static void job_err(RStream *rstream, void *data, bool eof)
+static void job_err(RStream *rstream, RBuffer *rbuf, void *data, bool eof)
 {
-  size_t count;
-  char buf[256];
-
-  while ((count = rstream_pending(rstream))) {
-    size_t read = rstream_read(rstream, buf, sizeof(buf) - 1);
+  while (rbuffer_size(rbuf)) {
+    char buf[256];
+    size_t read = rbuffer_read(rbuf, buf, sizeof(buf) - 1);
     buf[read] = NUL;
     ELOG("Channel %" PRIu64 " stderr: %s",
          ((Channel *)job_data(data))->id, buf);
@@ -352,7 +350,7 @@ static void job_exit(Job *job, int status, void *data)
   decref(data);
 }
 
-static void parse_msgpack(RStream *rstream, void *data, bool eof)
+static void parse_msgpack(RStream *rstream, RBuffer *rbuf, void *data, bool eof)
 {
   Channel *channel = data;
   incref(channel);
@@ -363,14 +361,14 @@ static void parse_msgpack(RStream *rstream, void *data, bool eof)
     goto end;
   }
 
-  size_t count = rstream_pending(rstream);
+  size_t count = rbuffer_size(rbuf);
   DLOG("Feeding the msgpack parser with %u bytes of data from RStream(%p)",
        count,
        rstream);
 
   // Feed the unpacker with data
   msgpack_unpacker_reserve_buffer(channel->unpacker, count);
-  rstream_read(rstream, msgpack_unpacker_buffer(channel->unpacker), count);
+  rbuffer_read(rbuf, msgpack_unpacker_buffer(channel->unpacker), count);
   msgpack_unpacker_buffer_consumed(channel->unpacker, count);
 
   msgpack_unpacked unpacked;
diff --git a/src/nvim/os/event.c b/src/nvim/os/event.c
index 4c3a4581c3..56874b495d 100644
--- a/src/nvim/os/event.c
+++ b/src/nvim/os/event.c
@@ -149,7 +149,7 @@ void event_push(Event event, bool deferred)
   // returns(user hits a key for example). To avoid this scenario, we call
   // uv_stop when a event is enqueued.
   uv_stop(uv_default_loop());
-  *kl_pushp(Event, deferred ? deferred_events : immediate_events) = event;
+  kl_push(Event, deferred ? deferred_events : immediate_events, event);
 }
 
 void event_process(void)
@@ -159,9 +159,8 @@ void event_process(void)
 
 static void process_events_from(klist_t(Event) *queue)
 {
-  Event event;
-
-  while (kl_shift(Event, queue, &event) == 0) {
+  while (!kl_empty(queue)) {
+    Event event = kl_shift(Event, queue);
     event.handler(event);
   }
 }
diff --git a/src/nvim/os/fs.c b/src/nvim/os/fs.c
index 52c10d0ca7..553dda5e88 100644
--- a/src/nvim/os/fs.c
+++ b/src/nvim/os/fs.c
@@ -19,6 +19,15 @@
 
 // Many fs functions from libuv return that value on success.
 static const int kLibuvSuccess = 0;
+static uv_loop_t fs_loop;
+
+
+// Initialize the fs module
+void fs_init(void)
+{
+  uv_loop_init(&fs_loop);
+}
+
 
 /// Change to the given directory.
 ///
@@ -184,7 +193,7 @@ int os_open(const char* path, int flags, int mode)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_fs_t open_req;
-  int r = uv_fs_open(uv_default_loop(), &open_req, path, flags, mode, NULL);
+  int r = uv_fs_open(&fs_loop, &open_req, path, flags, mode, NULL);
   uv_fs_req_cleanup(&open_req);
   // r is the same as open_req.result (except for OOM: then only r is set).
   return r;
@@ -197,7 +206,7 @@ static bool os_stat(const char *name, uv_stat_t *statbuf)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_fs_t request;
-  int result = uv_fs_stat(uv_default_loop(), &request, name, NULL);
+  int result = uv_fs_stat(&fs_loop, &request, name, NULL);
   *statbuf = request.statbuf;
   uv_fs_req_cleanup(&request);
   return (result == kLibuvSuccess);
@@ -224,7 +233,7 @@ int os_setperm(const char_u *name, int perm)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_fs_t request;
-  int result = uv_fs_chmod(uv_default_loop(), &request,
+  int result = uv_fs_chmod(&fs_loop, &request,
                            (const char*)name, perm, NULL);
   uv_fs_req_cleanup(&request);
 
@@ -245,7 +254,7 @@ int os_fchown(int file_descriptor, uv_uid_t owner, uv_gid_t group)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_fs_t request;
-  int result = uv_fs_fchown(uv_default_loop(), &request, file_descriptor,
+  int result = uv_fs_fchown(&fs_loop, &request, file_descriptor,
                             owner, group, NULL);
   uv_fs_req_cleanup(&request);
   return result;
@@ -294,7 +303,7 @@ int os_rename(const char_u *path, const char_u *new_path)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_fs_t request;
-  int result = uv_fs_rename(uv_default_loop(), &request,
+  int result = uv_fs_rename(&fs_loop, &request,
                             (const char *)path, (const char *)new_path, NULL);
   uv_fs_req_cleanup(&request);
 
@@ -312,7 +321,7 @@ int os_mkdir(const char *path, int32_t mode)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_fs_t request;
-  int result = uv_fs_mkdir(uv_default_loop(), &request, path, mode, NULL);
+  int result = uv_fs_mkdir(&fs_loop, &request, path, mode, NULL);
   uv_fs_req_cleanup(&request);
   return result;
 }
@@ -328,7 +337,7 @@ int os_mkdtemp(const char *template, char *path)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_fs_t request;
-  int result = uv_fs_mkdtemp(uv_default_loop(), &request, template, NULL);
+  int result = uv_fs_mkdtemp(&fs_loop, &request, template, NULL);
   if (result == kLibuvSuccess) {
     STRNCPY(path, request.path, TEMP_FILE_PATH_MAXLEN);
   }
@@ -343,7 +352,7 @@ int os_rmdir(const char *path)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_fs_t request;
-  int result = uv_fs_rmdir(uv_default_loop(), &request, path, NULL);
+  int result = uv_fs_rmdir(&fs_loop, &request, path, NULL);
   uv_fs_req_cleanup(&request);
   return result;
 }
@@ -356,7 +365,7 @@ int os_rmdir(const char *path)
 bool os_scandir(Directory *dir, const char *path)
   FUNC_ATTR_NONNULL_ALL
 {
-  int r = uv_fs_scandir(uv_default_loop(), &dir->request, path, 0, NULL);
+  int r = uv_fs_scandir(&fs_loop, &dir->request, path, 0, NULL);
   if (r <= 0) {
     os_closedir(dir);
   }
@@ -388,7 +397,7 @@ int os_remove(const char *path)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_fs_t request;
-  int result = uv_fs_unlink(uv_default_loop(), &request, path, NULL);
+  int result = uv_fs_unlink(&fs_loop, &request, path, NULL);
   uv_fs_req_cleanup(&request);
   return result;
 }
@@ -413,7 +422,7 @@ bool os_fileinfo_link(const char *path, FileInfo *file_info)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_fs_t request;
-  int result = uv_fs_lstat(uv_default_loop(), &request, path, NULL);
+  int result = uv_fs_lstat(&fs_loop, &request, path, NULL);
   file_info->stat = request.statbuf;
   uv_fs_req_cleanup(&request);
   return (result == kLibuvSuccess);
@@ -428,7 +437,7 @@ bool os_fileinfo_fd(int file_descriptor, FileInfo *file_info)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_fs_t request;
-  int result = uv_fs_fstat(uv_default_loop(), &request, file_descriptor, NULL);
+  int result = uv_fs_fstat(&fs_loop, &request, file_descriptor, NULL);
   file_info->stat = request.statbuf;
   uv_fs_req_cleanup(&request);
   return (result == kLibuvSuccess);
diff --git a/src/nvim/os/input.c b/src/nvim/os/input.c
index 74a5d3bc2e..726335bd9a 100644
--- a/src/nvim/os/input.c
+++ b/src/nvim/os/input.c
@@ -79,7 +79,7 @@ void input_stop(void)
 // Low level input function
 int os_inchar(uint8_t *buf, int maxlen, int ms, int tb_change_cnt)
 {
-  if (rbuffer_pending(input_buffer)) {
+  if (rbuffer_size(input_buffer)) {
     return (int)rbuffer_read(input_buffer, (char *)buf, (size_t)maxlen);
   }
 
@@ -108,7 +108,7 @@ int os_inchar(uint8_t *buf, int maxlen, int ms, int tb_change_cnt)
     return 0;
   }
 
-  if (rbuffer_pending(input_buffer)) {
+  if (rbuffer_size(input_buffer)) {
     // Safe to convert rbuffer_read to int, it will never overflow since we use
     // relatively small buffers.
     return (int)rbuffer_read(input_buffer, (char *)buf, (size_t)maxlen);
@@ -153,7 +153,7 @@ size_t input_enqueue(String keys)
 {
   char *ptr = keys.data, *end = ptr + keys.size;
 
-  while (rbuffer_available(input_buffer) >= 6 && ptr < end) {
+  while (rbuffer_space(input_buffer) >= 6 && ptr < end) {
     uint8_t buf[6] = {0};
     unsigned int new_size = trans_special((uint8_t **)&ptr, buf, true);
 
@@ -309,16 +309,17 @@ static InbufPollResult inbuf_poll(int ms)
   return input_eof ? kInputEof : kInputNone;
 }
 
-static void read_cb(RStream *rstream, void *data, bool at_eof)
+static void read_cb(RStream *rstream, RBuffer *buf, void *data, bool at_eof)
 {
   if (at_eof) {
     input_eof = true;
   }
 
-  char *buf = rbuffer_read_ptr(read_buffer);
-  size_t buf_size = rbuffer_pending(read_buffer);
-  (void)rbuffer_write(input_buffer, buf, buf_size);
-  rbuffer_consumed(read_buffer, buf_size);
+  assert(rbuffer_space(input_buffer) >= rbuffer_size(read_buffer));
+  RBUFFER_UNTIL_EMPTY(read_buffer, ptr, len) {
+    (void)rbuffer_write(input_buffer, ptr, len);
+    rbuffer_consumed(read_buffer, len);
+  }
 }
 
 static void process_interrupts(void)
@@ -327,18 +328,16 @@ static void process_interrupts(void)
     return;
   }
 
-  char *inbuf = rbuffer_read_ptr(input_buffer);
-  size_t count = rbuffer_pending(input_buffer), consume_count = 0;
-
-  for (int i = (int)count - 1; i >= 0; i--) {
-    if (inbuf[i] == 3) {
+  size_t consume_count = 0;
+  RBUFFER_EACH_REVERSE(input_buffer, c, i) {
+    if ((uint8_t)c == 3) {
       got_int = true;
-      consume_count = (size_t)i;
+      consume_count = i;
       break;
     }
   }
 
-  if (got_int) {
+  if (got_int && consume_count) {
     // Remove everything typed before the CTRL-C
     rbuffer_consumed(input_buffer, consume_count);
   }
@@ -362,7 +361,7 @@ static int push_event_key(uint8_t *buf, int maxlen)
 static bool input_ready(void)
 {
   return typebuf_was_filled ||                 // API call filled typeahead
-         rbuffer_pending(input_buffer) > 0 ||  // Input buffer filled
+         rbuffer_size(input_buffer) ||         // Input buffer filled
          event_has_deferred();                 // Events must be processed
 }
 
diff --git a/src/nvim/os/job.c b/src/nvim/os/job.c
index 038d0e3c26..4769ee4d2f 100644
--- a/src/nvim/os/job.c
+++ b/src/nvim/os/job.c
@@ -404,17 +404,17 @@ static void job_stop_timer_cb(uv_timer_t *handle)
 }
 
 // Wraps the call to std{out,err}_cb and emits a JobExit event if necessary.
-static void read_cb(RStream *rstream, void *data, bool eof)
+static void read_cb(RStream *rstream, RBuffer *buf, void *data, bool eof)
 {
   Job *job = data;
 
   if (rstream == job->out) {
-    job->opts.stdout_cb(rstream, data, eof);
+    job->opts.stdout_cb(rstream, buf, data, eof);
     if (eof) {
       close_job_out(job);
     }
   } else {
-    job->opts.stderr_cb(rstream, data, eof);
+    job->opts.stderr_cb(rstream, buf, data, eof);
     if (eof) {
       close_job_err(job);
     }
diff --git a/src/nvim/os/rstream.c b/src/nvim/os/rstream.c
index 702f282d53..af84288f0f 100644
--- a/src/nvim/os/rstream.c
+++ b/src/nvim/os/rstream.c
@@ -14,12 +14,6 @@
 #include "nvim/log.h"
 #include "nvim/misc1.h"
 
-struct rbuffer {
-  char *data;
-  size_t capacity, rpos, wpos;
-  RStream *rstream;
-};
-
 struct rstream {
   void *data;
   uv_buf_t uvbuf;
@@ -37,135 +31,6 @@ struct rstream {
 # include "os/rstream.c.generated.h"
 #endif
 
-/// Creates a new `RBuffer` instance.
-RBuffer *rbuffer_new(size_t capacity)
-{
-  RBuffer *rv = xmalloc(sizeof(RBuffer));
-  rv->data = xmalloc(capacity);
-  rv->capacity = capacity;
-  rv->rpos = rv->wpos = 0;
-  rv->rstream = NULL;
-  return rv;
-}
-
-/// Advances `rbuffer` read pointers to consume data. If the associated
-/// RStream had stopped because the buffer was full, this will restart it.
-///
-/// This is called automatically by rbuffer_read, but when using
-/// `rbuffer_read_ptr` directly, this needs to called after the data was
-/// consumed.
-void rbuffer_consumed(RBuffer *rbuffer, size_t count)
-{
-  rbuffer->rpos += count;
-  if (count && rbuffer->wpos == rbuffer->capacity) {
-    // `wpos` is at the end of the buffer, so free some space by moving unread
-    // data...
-    rbuffer_relocate(rbuffer);
-    if (rbuffer->rstream) {
-      // restart the associated RStream
-      rstream_start(rbuffer->rstream);
-    }
-  }
-}
-
-/// Advances `rbuffer` write pointers. If the internal buffer becomes full,
-/// this will stop the associated RStream instance.
-void rbuffer_produced(RBuffer *rbuffer, size_t count)
-{
-  rbuffer->wpos += count;
-  DLOG("Received %u bytes from RStream(%p)", (size_t)count, rbuffer->rstream);
-
-  rbuffer_relocate(rbuffer);
-  if (rbuffer->rstream && rbuffer->wpos == rbuffer->capacity) {
-    // The last read filled the buffer, stop reading for now
-    //
-    rstream_stop(rbuffer->rstream);
-    DLOG("Buffer for RStream(%p) is full, stopping it", rbuffer->rstream);
-  }
-}
-
-/// Reads data from a `RBuffer` instance into a raw buffer.
-///
-/// @param rbuffer The `RBuffer` instance
-/// @param buffer The buffer which will receive the data
-/// @param count Number of bytes that `buffer` can accept
-/// @return The number of bytes copied into `buffer`
-size_t rbuffer_read(RBuffer *rbuffer, char *buffer, size_t count)
-{
-  size_t read_count = rbuffer_pending(rbuffer);
-
-  if (count < read_count) {
-    read_count = count;
-  }
-
-  if (read_count > 0) {
-    memcpy(buffer, rbuffer_read_ptr(rbuffer), read_count);
-    rbuffer_consumed(rbuffer, read_count);
-  }
-
-  return read_count;
-}
-
-/// Copies data to `rbuffer` read queue.
-///
-/// @param rbuffer the `RBuffer` instance
-/// @param buffer The buffer containing data to be copied
-/// @param count Number of bytes that should be copied
-/// @return The number of bytes actually copied
-size_t rbuffer_write(RBuffer *rbuffer, char *buffer, size_t count)
-{
-  size_t write_count = rbuffer_available(rbuffer);
-
-  if (count < write_count) {
-    write_count = count;
-  }
-
-  if (write_count > 0) {
-    memcpy(rbuffer_write_ptr(rbuffer), buffer, write_count);
-    rbuffer_produced(rbuffer, write_count);
-  }
-
-  return write_count;
-}
-
-/// Returns a pointer to a raw buffer containing the first byte available for
-/// reading.
-char *rbuffer_read_ptr(RBuffer *rbuffer)
-{
-  return rbuffer->data + rbuffer->rpos;
-}
-
-/// Returns a pointer to a raw buffer containing the first byte available for
-/// write.
-char *rbuffer_write_ptr(RBuffer *rbuffer)
-{
-  return rbuffer->data + rbuffer->wpos;
-}
-
-/// Returns the number of bytes ready for consumption in `rbuffer`
-///
-/// @param rbuffer The `RBuffer` instance
-/// @return The number of bytes ready for consumption
-size_t rbuffer_pending(RBuffer *rbuffer)
-{
-  return rbuffer->wpos - rbuffer->rpos;
-}
-
-/// Returns available space in `rbuffer`
-///
-/// @param rbuffer The `RBuffer` instance
-/// @return The space available in number of bytes
-size_t rbuffer_available(RBuffer *rbuffer)
-{
-  return rbuffer->capacity - rbuffer->wpos;
-}
-
-void rbuffer_free(RBuffer *rbuffer)
-{
-  xfree(rbuffer->data);
-  xfree(rbuffer);
-}
-
 /// Creates a new RStream instance. A RStream encapsulates all the boilerplate
 /// necessary for reading from a libuv stream.
 ///
@@ -177,8 +42,10 @@ void rbuffer_free(RBuffer *rbuffer)
 RStream * rstream_new(rstream_cb cb, RBuffer *buffer, void *data)
 {
   RStream *rv = xmalloc(sizeof(RStream));
+  buffer->data = rv;
+  buffer->full_cb = on_rbuffer_full;
+  buffer->nonfull_cb = on_rbuffer_nonfull;
   rv->buffer = buffer;
-  rv->buffer->rstream = rv;
   rv->fpos = 0;
   rv->data = data;
   rv->cb = cb;
@@ -190,16 +57,14 @@ RStream * rstream_new(rstream_cb cb, RBuffer *buffer, void *data)
   return rv;
 }
 
-/// Returns the read pointer used by the rstream.
-char *rstream_read_ptr(RStream *rstream)
+static void on_rbuffer_full(RBuffer *buf, void *data)
 {
-  return rbuffer_read_ptr(rstream->buffer);
+  rstream_stop(data);
 }
 
-/// Returns the number of bytes before the rstream is full.
-size_t rstream_available(RStream *rstream)
+static void on_rbuffer_nonfull(RBuffer *buf, void *data)
 {
-  return rbuffer_available(rstream->buffer);
+  rstream_start(data);
 }
 
 /// Frees all memory allocated for a RStream instance
@@ -297,37 +162,13 @@ void rstream_stop(RStream *rstream)
   }
 }
 
-/// Returns the number of bytes ready for consumption in `rstream`
-size_t rstream_pending(RStream *rstream)
-{
-  return rbuffer_pending(rstream->buffer);
-}
-
-/// Reads data from a `RStream` instance into a buffer.
-///
-/// @param rstream The `RStream` instance
-/// @param buffer The buffer which will receive the data
-/// @param count Number of bytes that `buffer` can accept
-/// @return The number of bytes copied into `buffer`
-size_t rstream_read(RStream *rstream, char *buffer, size_t count)
-{
-  return rbuffer_read(rstream->buffer, buffer, count);
-}
-
-RBuffer *rstream_buffer(RStream *rstream)
-{
-  return rstream->buffer;
-}
-
 // Callbacks used by libuv
 
 // Called by libuv to allocate memory for reading.
 static void alloc_cb(uv_handle_t *handle, size_t suggested, uv_buf_t *buf)
 {
   RStream *rstream = handle_get_rstream(handle);
-
-  buf->len = rbuffer_available(rstream->buffer);
-  buf->base = rbuffer_write_ptr(rstream->buffer);
+  buf->base = rbuffer_write_ptr(rstream->buffer, &buf->len);
 }
 
 // Callback invoked by libuv after it copies the data into the buffer provided
@@ -338,23 +179,30 @@ static void read_cb(uv_stream_t *stream, ssize_t cnt, const uv_buf_t *buf)
   RStream *rstream = handle_get_rstream((uv_handle_t *)stream);
 
   if (cnt <= 0) {
-    if (cnt != UV_ENOBUFS) {
-      DLOG("Closing RStream(%p)", rstream);
+    if (cnt != UV_ENOBUFS
+        // cnt == 0 means libuv asked for a buffer and decided it wasn't needed:
+        // http://docs.libuv.org/en/latest/stream.html#c.uv_read_start.
+        //
+        // We don't need to do anything with the RBuffer because the next call
+        // to `alloc_cb` will return the same unused pointer(`rbuffer_produced`
+        // won't be called)
+        && cnt != 0) {
+      DLOG("Closing RStream(%p) because of %s(%zd)", rstream,
+           uv_strerror((int)cnt), cnt);
       // Read error or EOF, either way stop the stream and invoke the callback
       // with eof == true
       uv_read_stop(stream);
-      rstream->cb(rstream, rstream->data, true);
+      rstream->cb(rstream, rstream->buffer, rstream->data, true);
     }
     return;
   }
 
   // at this point we're sure that cnt is positive, no error occurred
-  size_t nread = (size_t) cnt;
-
+  size_t nread = (size_t)cnt;
   // Data was already written, so all we need is to update 'wpos' to reflect
   // the space actually used in the buffer.
   rbuffer_produced(rstream->buffer, nread);
-  rstream->cb(rstream, rstream->data, false);
+  rstream->cb(rstream, rstream->buffer, rstream->data, false);
 }
 
 // Called by the by the 'idle' handle to emulate a reading event
@@ -363,8 +211,7 @@ static void fread_idle_cb(uv_idle_t *handle)
   uv_fs_t req;
   RStream *rstream = handle_get_rstream((uv_handle_t *)handle);
 
-  rstream->uvbuf.len = rbuffer_available(rstream->buffer);
-  rstream->uvbuf.base = rbuffer_write_ptr(rstream->buffer);
+  rstream->uvbuf.base = rbuffer_write_ptr(rstream->buffer, &rstream->uvbuf.len);
 
   // the offset argument to uv_fs_read is int64_t, could someone really try
   // to read more than 9 quintillion (9e18) bytes?
@@ -389,7 +236,7 @@ static void fread_idle_cb(uv_idle_t *handle)
 
   if (req.result <= 0) {
     uv_idle_stop(rstream->fread_idle);
-    rstream->cb(rstream, rstream->data, true);
+    rstream->cb(rstream, rstream->buffer, rstream->data, true);
     return;
   }
 
@@ -404,15 +251,3 @@ static void close_cb(uv_handle_t *handle)
   xfree(handle->data);
   xfree(handle);
 }
-
-static void rbuffer_relocate(RBuffer *rbuffer)
-{
-  assert(rbuffer->rpos <= rbuffer->wpos);
-  // Move data ...
-  memmove(
-      rbuffer->data,  // ...to the beginning of the buffer(rpos 0)
-      rbuffer->data + rbuffer->rpos,  // ...From the first unread position
-      rbuffer->wpos - rbuffer->rpos);  // ...By the number of unread bytes
-  rbuffer->wpos -= rbuffer->rpos;
-  rbuffer->rpos = 0;
-}
diff --git a/src/nvim/os/rstream.h b/src/nvim/os/rstream.h
index 713d1e77e6..3e24724573 100644
--- a/src/nvim/os/rstream.h
+++ b/src/nvim/os/rstream.h
@@ -5,7 +5,6 @@
 #include <stdint.h>
 #include <uv.h>
 #include "nvim/os/event_defs.h"
-
 #include "nvim/os/rstream_defs.h"
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
diff --git a/src/nvim/os/rstream_defs.h b/src/nvim/os/rstream_defs.h
index 1d71160963..45dced0b62 100644
--- a/src/nvim/os/rstream_defs.h
+++ b/src/nvim/os/rstream_defs.h
@@ -3,15 +3,18 @@
 
 #include <stdbool.h>
 
-typedef struct rbuffer RBuffer;
+#include "nvim/rbuffer.h"
+
 typedef struct rstream RStream;
 
 /// Type of function called when the RStream receives data
 ///
 /// @param rstream The RStream instance
+/// @param rbuffer The associated RBuffer instance
 /// @param data State associated with the RStream instance
 /// @param eof If the stream reached EOF.
-typedef void (*rstream_cb)(RStream *rstream, void *data, bool eof);
+typedef void (*rstream_cb)(RStream *rstream, RBuffer *buf, void *data,
+    bool eof);
 
 #endif  // NVIM_OS_RSTREAM_DEFS_H
 
diff --git a/src/nvim/os/shell.c b/src/nvim/os/shell.c
index 2de3b1aeed..48174533a6 100644
--- a/src/nvim/os/shell.c
+++ b/src/nvim/os/shell.c
@@ -283,25 +283,28 @@ static void dynamic_buffer_ensure(DynamicBuffer *buf, size_t desired)
   buf->data = xrealloc(buf->data, buf->cap);
 }
 
-static void system_data_cb(RStream *rstream, void *data, bool eof)
+static void system_data_cb(RStream *rstream, RBuffer *buf, void *data, bool eof)
 {
   Job *job = data;
-  DynamicBuffer *buf = job_data(job);
+  DynamicBuffer *dbuf = job_data(job);
 
-  size_t nread = rstream_pending(rstream);
-
-  dynamic_buffer_ensure(buf, buf->len + nread + 1);
-  rstream_read(rstream, buf->data + buf->len, nread);
-
-  buf->len += nread;
+  size_t nread = buf->size;
+  dynamic_buffer_ensure(dbuf, dbuf->len + nread + 1);
+  rbuffer_read(buf, dbuf->data + dbuf->len, nread);
+  dbuf->len += nread;
 }
 
-static void out_data_cb(RStream *rstream, void *data, bool eof)
+static void out_data_cb(RStream *rstream, RBuffer *buf, void *data, bool eof)
 {
-  RBuffer *rbuffer = rstream_buffer(rstream);
-  size_t written = write_output(rbuffer_read_ptr(rbuffer),
-                                rbuffer_pending(rbuffer), false, eof);
-  rbuffer_consumed(rbuffer, written);
+  RBUFFER_UNTIL_EMPTY(buf, ptr, len) {
+    size_t written = write_output(ptr, len, false,
+        eof && len <= rbuffer_size(buf));
+    if (written) {
+      rbuffer_consumed(buf, written);
+    } else {
+      break;
+    }
+  }
 }
 
 /// Parses a command string into a sequence of words, taking quotes into
diff --git a/src/nvim/rbuffer.c b/src/nvim/rbuffer.c
new file mode 100644
index 0000000000..9cf681585b
--- /dev/null
+++ b/src/nvim/rbuffer.c
@@ -0,0 +1,205 @@
+#include <assert.h>
+#include <stddef.h>
+#include <string.h>
+
+#include "nvim/memory.h"
+#include "nvim/vim.h"
+#include "nvim/rbuffer.h"
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "rbuffer.c.generated.h"
+#endif
+
+/// Creates a new `RBuffer` instance.
+RBuffer *rbuffer_new(size_t capacity)
+  FUNC_ATTR_WARN_UNUSED_RESULT FUNC_ATTR_NONNULL_RET
+{
+  if (!capacity) {
+    capacity = 0xffff;
+  }
+
+  RBuffer *rv = xmalloc(sizeof(RBuffer) + capacity);
+  rv->full_cb = rv->nonfull_cb = NULL;
+  rv->data = NULL;
+  rv->size = 0;
+  rv->write_ptr = rv->read_ptr = rv->start_ptr;
+  rv->end_ptr = rv->start_ptr + capacity;
+  return rv;
+}
+
+void rbuffer_free(RBuffer *buf)
+{
+  xfree(buf);
+}
+
+size_t rbuffer_size(RBuffer *buf) FUNC_ATTR_NONNULL_ALL
+{
+  return buf->size;
+}
+
+size_t rbuffer_capacity(RBuffer *buf) FUNC_ATTR_NONNULL_ALL
+{
+  return (size_t)(buf->end_ptr - buf->start_ptr);
+}
+
+size_t rbuffer_space(RBuffer *buf) FUNC_ATTR_NONNULL_ALL
+{
+  return rbuffer_capacity(buf) - buf->size;
+}
+
+/// Return a pointer to a raw buffer containing the first empty slot available
+/// for writing. The second argument is a pointer to the maximum number of
+/// bytes that could be written.
+///
+/// It is necessary to call this function twice to ensure all empty space was
+/// used. See RBUFFER_UNTIL_FULL for a macro that simplifies this task.
+char *rbuffer_write_ptr(RBuffer *buf, size_t *write_count) FUNC_ATTR_NONNULL_ALL
+{
+  if (buf->size == rbuffer_capacity(buf)) {
+    *write_count = 0;
+    return NULL;
+  }
+
+  if (buf->write_ptr >= buf->read_ptr) {
+    *write_count = (size_t)(buf->end_ptr - buf->write_ptr);
+  } else {
+    *write_count = (size_t)(buf->read_ptr - buf->write_ptr);
+  }
+
+  return buf->write_ptr;
+}
+
+/// Adjust `rbuffer` write pointer to reflect produced data. This is called
+/// automatically by `rbuffer_write`, but when using `rbuffer_write_ptr`
+/// directly, this needs to called after the data was copied to the internal
+/// buffer. The write pointer will be wrapped if required.
+void rbuffer_produced(RBuffer *buf, size_t count) FUNC_ATTR_NONNULL_ALL
+{
+  assert(count && count <= rbuffer_space(buf));
+
+  buf->write_ptr += count;
+  if (buf->write_ptr >= buf->end_ptr) {
+    // wrap around
+    buf->write_ptr -= rbuffer_capacity(buf);
+  }
+
+  buf->size += count;
+  if (buf->full_cb && !rbuffer_space(buf)) {
+    buf->full_cb(buf, buf->data);
+  }
+}
+
+/// Return a pointer to a raw buffer containing the first byte available
+/// for reading. The second argument is a pointer to the maximum number of
+/// bytes that could be read.
+///
+/// It is necessary to call this function twice to ensure all available bytes
+/// were read. See RBUFFER_UNTIL_EMPTY for a macro that simplifies this task.
+char *rbuffer_read_ptr(RBuffer *buf, size_t *read_count) FUNC_ATTR_NONNULL_ALL
+{
+  if (!buf->size) {
+    *read_count = 0;
+    return NULL;
+  }
+
+  if (buf->read_ptr < buf->write_ptr) {
+    *read_count = (size_t)(buf->write_ptr - buf->read_ptr);
+  } else {
+    *read_count = (size_t)(buf->end_ptr - buf->read_ptr);
+  }
+
+  return buf->read_ptr;
+}
+
+/// Adjust `rbuffer` read pointer to reflect consumed data. This is called
+/// automatically by `rbuffer_read`, but when using `rbuffer_read_ptr`
+/// directly, this needs to called after the data was copied from the internal
+/// buffer. The read pointer will be wrapped if required.
+void rbuffer_consumed(RBuffer *buf, size_t count)
+  FUNC_ATTR_NONNULL_ALL
+{
+  assert(count && count <= buf->size);
+
+  buf->read_ptr += count;
+  if (buf->read_ptr >= buf->end_ptr) {
+      buf->read_ptr -= rbuffer_capacity(buf);
+  }
+
+  bool was_full = buf->size == rbuffer_capacity(buf);
+  buf->size -= count;
+  if (buf->nonfull_cb && was_full) {
+    buf->nonfull_cb(buf, buf->data);
+  }
+}
+
+// Higher level functions for copying from/to RBuffer instances and data
+// pointers
+size_t rbuffer_write(RBuffer *buf, char *src, size_t src_size)
+  FUNC_ATTR_NONNULL_ALL
+{
+  size_t size = src_size;
+
+  RBUFFER_UNTIL_FULL(buf, wptr, wcnt) {
+    size_t copy_count = MIN(src_size, wcnt);
+    memcpy(wptr, src, copy_count);
+    rbuffer_produced(buf, copy_count);
+
+    if (!(src_size -= copy_count)) {
+      return size;
+    }
+
+    src += copy_count;
+  }
+
+  return size - src_size;
+}
+
+size_t rbuffer_read(RBuffer *buf, char *dst, size_t dst_size)
+  FUNC_ATTR_NONNULL_ALL
+{
+  size_t size = dst_size;
+
+  RBUFFER_UNTIL_EMPTY(buf, rptr, rcnt) {
+    size_t copy_count = MIN(dst_size, rcnt);
+    memcpy(dst, rptr, copy_count);
+    rbuffer_consumed(buf, copy_count);
+
+    if (!(dst_size -= copy_count)) {
+      return size;
+    }
+
+    dst += copy_count;
+  }
+
+  return size - dst_size;
+}
+
+char *rbuffer_get(RBuffer *buf, size_t index)
+    FUNC_ATTR_NONNULL_ALL FUNC_ATTR_NONNULL_RET
+{
+  assert(index < buf->size);
+  char *rptr = buf->read_ptr + index;
+  if (rptr >= buf->end_ptr) {
+    rptr -= rbuffer_capacity(buf);
+  }
+  return rptr;
+}
+
+int rbuffer_cmp(RBuffer *buf, const char *str, size_t count)
+  FUNC_ATTR_NONNULL_ALL
+{
+  assert(count <= buf->size);
+  size_t rcnt;
+  (void)rbuffer_read_ptr(buf, &rcnt);
+  size_t n = MIN(count, rcnt);
+  int rv = memcmp(str, buf->read_ptr, n);
+  count -= n;
+  size_t remaining = buf->size - rcnt;
+
+  if (rv || !count || !remaining) {
+    return rv;
+  }
+
+  return memcmp(str + n, buf->start_ptr, count);
+}
+
diff --git a/src/nvim/rbuffer.h b/src/nvim/rbuffer.h
new file mode 100644
index 0000000000..b205db0b5a
--- /dev/null
+++ b/src/nvim/rbuffer.h
@@ -0,0 +1,83 @@
+// Ring buffer implementation. This is basically an array that wraps read/write
+// pointers around the memory region. It should be more efficient than the old
+// RBuffer which required memmove() calls to relocate read/write positions.
+//
+// The main purpose of RBuffer is simplify memory management when reading from
+// uv_stream_t instances:
+//
+// - The event loop writes data to a RBuffer, advancing the write pointer
+// - The main loop reads data, advancing the read pointer
+// - If the buffer becomes full(size == capacity) the rstream is temporarily
+//   stopped(automatic backpressure handling)
+//
+// Reference: http://en.wikipedia.org/wiki/Circular_buffer
+#ifndef NVIM_RBUFFER_H
+#define NVIM_RBUFFER_H
+
+#include <stddef.h>
+#include <stdint.h>
+
+// Macros that simplify working with the read/write pointers directly by hiding
+// ring buffer wrap logic. Some examples:
+//
+// - Pass the write pointer to a function(write_data) that incrementally
+//   produces data, returning the number of bytes actually written to the
+//   ring buffer:
+//
+//       RBUFFER_UNTIL_FULL(rbuf, ptr, cnt)
+//         rbuffer_produced(rbuf, write_data(state, ptr, cnt));
+//
+// - Pass the read pointer to a function(read_data) that incrementally
+//   consumes data, returning the number of bytes actually read from the
+//   ring buffer:
+//
+//       RBUFFER_UNTIL_EMPTY(rbuf, ptr, cnt)
+//         rbuffer_consumed(rbuf, read_data(state, ptr, cnt));
+//
+// Note that the rbuffer_{produced,consumed} calls are necessary or these macros
+// create infinite loops
+#define RBUFFER_UNTIL_EMPTY(buf, rptr, rcnt)                                 \
+  for (size_t rcnt = 0, _r = 1; _r; _r = 0)                                  \
+    for (char *rptr = rbuffer_read_ptr(buf, &rcnt);                          \
+         buf->size;                                                          \
+         rptr = rbuffer_read_ptr(buf, &rcnt))
+
+#define RBUFFER_UNTIL_FULL(buf, wptr, wcnt)                                  \
+  for (size_t wcnt = 0, _r = 1; _r; _r = 0)                                  \
+    for (char *wptr = rbuffer_write_ptr(buf, &wcnt);                         \
+         rbuffer_space(buf);                                                 \
+         wptr = rbuffer_write_ptr(buf, &wcnt))
+
+
+// Iteration
+#define RBUFFER_EACH(buf, c, i)                                              \
+  for (size_t i = 0; i < buf->size; i = buf->size)                           \
+      for (char c = 0;                                                       \
+           i < buf->size ? ((int)(c = *rbuffer_get(buf, i))) || 1 : 0;       \
+           i++)
+
+#define RBUFFER_EACH_REVERSE(buf, c, i)                                      \
+  for (size_t i = buf->size; i != SIZE_MAX; i = SIZE_MAX)                    \
+      for (char c = 0;                                                       \
+           i-- > 0 ? ((int)(c = *rbuffer_get(buf, i))) || 1 : 0;             \
+           )
+
+typedef struct rbuffer RBuffer;
+/// Type of function invoked during certain events:
+///   - When the RBuffer switches to the full state
+///   - When the RBuffer switches to the non-full state
+typedef void(*rbuffer_callback)(RBuffer *buf, void *data);
+
+struct rbuffer {
+  rbuffer_callback full_cb, nonfull_cb;
+  void *data;
+  size_t size;
+  char *end_ptr, *read_ptr, *write_ptr;
+  char start_ptr[];
+};
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "rbuffer.h.generated.h"
+#endif
+
+#endif  // NVIM_RBUFFER_H
diff --git a/src/nvim/testdir/test86.in b/src/nvim/testdir/test86.in
index 958bd57e29..41d9a2fa32 100644
--- a/src/nvim/testdir/test86.in
+++ b/src/nvim/testdir/test86.in
@@ -1,5 +1,8 @@
 Tests for various python features.     vim: set ft=vim :
 
+This test is not run in Neovim(see the has('nvim') check below) because the
+python compatibility layer is not complete.
+
 NOTE: This will cause errors when run under valgrind.
 This would require recompiling Python with:
   ./configure --without-pymalloc
diff --git a/src/nvim/testdir/test87.in b/src/nvim/testdir/test87.in
index cad778e858..58f9ac6418 100644
--- a/src/nvim/testdir/test87.in
+++ b/src/nvim/testdir/test87.in
@@ -1,9 +1,12 @@
 Tests for various python features.     vim: set ft=vim :
 
+This test is not run in Neovim(see the has('nvim') check below) because the
+python3 compatibility layer is not complete.
+
 STARTTEST
 :so small.vim
 :set noswapfile
-:if !has('python3') | e! test.ok | wq! test.out | endif
+:if !has('python3') || has('nvim') | e! test.ok | wq! test.out | endif
 :lang C
 :fun Test()
 :py3 import vim
diff --git a/src/nvim/tui/term_input.inl b/src/nvim/tui/term_input.inl
index d25cbb7ba1..451ac195f2 100644
--- a/src/nvim/tui/term_input.inl
+++ b/src/nvim/tui/term_input.inl
@@ -162,11 +162,10 @@ static void timer_cb(uv_timer_t *handle)
 
 static bool handle_bracketed_paste(TermInput *input)
 {
-  char *ptr = rbuffer_read_ptr(input->read_buffer);
-  size_t len = rbuffer_pending(input->read_buffer);
-  if (len > 5 && (!strncmp(ptr, "\x1b[200~", 6)
-        || !strncmp(ptr, "\x1b[201~", 6))) {
-    bool enable = ptr[4] == '0';
+  if (rbuffer_size(input->read_buffer) > 5 &&
+      (!rbuffer_cmp(input->read_buffer, "\x1b[200~", 6) ||
+       !rbuffer_cmp(input->read_buffer, "\x1b[201~", 6))) {
+    bool enable = *rbuffer_get(input->read_buffer, 4) == '0';
     // Advance past the sequence
     rbuffer_consumed(input->read_buffer, 6);
     if (input->paste_enabled == enable) {
@@ -195,11 +194,11 @@ static bool handle_bracketed_paste(TermInput *input)
 
 static bool handle_forced_escape(TermInput *input)
 {
-  char *ptr = rbuffer_read_ptr(input->read_buffer);
-  size_t len = rbuffer_pending(input->read_buffer);
-  if (len > 1 && ptr[0] == ESC && ptr[1] == NUL) {
+  if (rbuffer_size(input->read_buffer) > 1
+      && !rbuffer_cmp(input->read_buffer, "\x1b\x00", 2)) {
     // skip the ESC and NUL and push one <esc> to the input buffer
-    termkey_push_bytes(input->tk, ptr, 1);
+    size_t rcnt;
+    termkey_push_bytes(input->tk, rbuffer_read_ptr(input->read_buffer, &rcnt), 1);
     rbuffer_consumed(input->read_buffer, 2);
     tk_getkeys(input, true);
     return true;
@@ -207,9 +206,9 @@ static bool handle_forced_escape(TermInput *input)
   return false;
 }
 
-static void read_cb(RStream *rstream, void *rstream_data, bool eof)
+static void read_cb(RStream *rstream, RBuffer *buf, void *data, bool eof)
 {
-  TermInput *input = rstream_data;
+  TermInput *input = data;
 
   if (eof) {
     if (input->in_fd == 0 && !os_isatty(0) && os_isatty(2)) {
@@ -236,19 +235,33 @@ static void read_cb(RStream *rstream, void *rstream_data, bool eof)
     if (handle_bracketed_paste(input) || handle_forced_escape(input)) {
       continue;
     }
-    char *ptr = rbuffer_read_ptr(input->read_buffer);
-    size_t len = rbuffer_pending(input->read_buffer);
-    // Find the next 'esc' and push everything up to it(excluding)
-    size_t i;
-    for (i = ptr[0] == ESC ? 1 : 0; i < len; i++) {
-      if (ptr[i] == '\x1b') {
+
+    // Find the next 'esc' and push everything up to it(excluding). This is done
+    // so the `handle_bracketed_paste`/`handle_forced_escape` calls above work
+    // as expected.
+    size_t count = 0;
+    RBUFFER_EACH(input->read_buffer, c, i) {
+      count = i + 1;
+      if (c == '\x1b' && count > 1) {
+        break;
+      }
+    }
+
+    RBUFFER_UNTIL_EMPTY(input->read_buffer, ptr, len) {
+      size_t consumed = termkey_push_bytes(input->tk, ptr, MIN(count, len));
+      // termkey_push_bytes can return (size_t)-1, so it is possible that
+      // `consumed > input->read_buffer->size`, but since tk_getkeys is called
+      // soon, it shouldn't happen
+      assert(consumed <= input->read_buffer->size);
+      rbuffer_consumed(input->read_buffer, consumed);
+      // Need to process the keys now since there's no guarantee "count" will
+      // fit into libtermkey's input buffer.
+      tk_getkeys(input, false);
+      if (!(count -= consumed)) {
         break;
       }
     }
-    size_t consumed = termkey_push_bytes(input->tk, ptr, i);
-    rbuffer_consumed(input->read_buffer, consumed);
-    tk_getkeys(input, false);
-  } while (rbuffer_pending(input->read_buffer));
+  } while (rbuffer_size(input->read_buffer));
 }
 
 static TermInput *term_input_new(void)
diff --git a/test/unit/buffer_spec.lua b/test/unit/buffer_spec.lua
index 5244c2af86..e0e2b827e9 100644
--- a/test/unit/buffer_spec.lua
+++ b/test/unit/buffer_spec.lua
@@ -3,8 +3,6 @@ local helpers = require("test.unit.helpers")
 local to_cstr = helpers.to_cstr
 local eq      = helpers.eq
 
-helpers.vim_init()
-
 local buffer = helpers.cimport("./src/nvim/buffer.h")
 local window = helpers.cimport("./src/nvim/window.h")
 local option = helpers.cimport("./src/nvim/option.h")
diff --git a/test/unit/fixtures/rbuffer.c b/test/unit/fixtures/rbuffer.c
new file mode 100644
index 0000000000..d587d6b054
--- /dev/null
+++ b/test/unit/fixtures/rbuffer.c
@@ -0,0 +1,28 @@
+#include "nvim/rbuffer.h"
+#include "rbuffer.h"
+
+
+void ut_rbuffer_each_read_chunk(RBuffer *buf, each_ptr_cb cb)
+{
+  RBUFFER_UNTIL_EMPTY(buf, rptr, rcnt) {
+    cb(rptr, rcnt);
+    rbuffer_consumed(buf, rcnt);
+  }
+}
+
+void ut_rbuffer_each_write_chunk(RBuffer *buf, each_ptr_cb cb)
+{
+  RBUFFER_UNTIL_FULL(buf, wptr, wcnt) {
+    cb(wptr, wcnt);
+    rbuffer_produced(buf, wcnt);
+  }
+}
+void ut_rbuffer_each(RBuffer *buf, each_cb cb)
+{
+  RBUFFER_EACH(buf, c, i) cb(c, i);
+}
+
+void ut_rbuffer_each_reverse(RBuffer *buf, each_cb cb)
+{
+  RBUFFER_EACH_REVERSE(buf, c, i) cb(c, i);
+}
diff --git a/test/unit/fixtures/rbuffer.h b/test/unit/fixtures/rbuffer.h
new file mode 100644
index 0000000000..640092c627
--- /dev/null
+++ b/test/unit/fixtures/rbuffer.h
@@ -0,0 +1,9 @@
+#include "nvim/rbuffer.h"
+
+typedef void(*each_ptr_cb)(char *ptr, size_t cnt);
+typedef void(*each_cb)(char c, size_t i);
+
+void ut_rbuffer_each_read_chunk(RBuffer *buf, each_ptr_cb cb);
+void ut_rbuffer_each_write_chunk(RBuffer *buf, each_ptr_cb cb);
+void ut_rbuffer_each(RBuffer *buf, each_cb cb);
+void ut_rbuffer_each_reverse(RBuffer *buf, each_cb cb);
diff --git a/test/unit/helpers.lua b/test/unit/helpers.lua
index 708e7a94ab..5bcc661226 100644
--- a/test/unit/helpers.lua
+++ b/test/unit/helpers.lua
@@ -136,15 +136,11 @@ end
 
 -- initialize some global variables, this is still necessary to unit test
 -- functions that rely on global state.
-local function vim_init()
-  if vim_init_called ~= nil then
-    return 
-  end
+do
   local main = cimport('./src/nvim/main.h')
   local time = cimport('./src/nvim/os/time.h')
   time.time_init()
   main.early_init()
-  vim_init_called = true
 end
 
 -- C constants.
@@ -167,7 +163,6 @@ return {
   lib = libnvim,
   cstr = cstr,
   to_cstr = to_cstr,
-  vim_init = vim_init,
   NULL = NULL,
   OK = OK,
   FAIL = FAIL
diff --git a/test/unit/os/env_spec.lua b/test/unit/os/env_spec.lua
index 9d936c2564..8e18c599d9 100644
--- a/test/unit/os/env_spec.lua
+++ b/test/unit/os/env_spec.lua
@@ -12,9 +12,6 @@ local NULL = helpers.NULL
 
 require('lfs')
 
--- Needed because expand_env_esc uses the char table
-helpers.vim_init()
-
 local env = cimport('./src/nvim/os/os.h')
 
 describe('env function', function()
diff --git a/test/unit/rbuffer_spec.lua b/test/unit/rbuffer_spec.lua
new file mode 100644
index 0000000000..89136410d3
--- /dev/null
+++ b/test/unit/rbuffer_spec.lua
@@ -0,0 +1,350 @@
+local helpers = require("test.unit.helpers")
+
+local ffi     = helpers.ffi
+local eq      = helpers.eq
+local cstr    = helpers.cstr
+local to_cstr = helpers.to_cstr
+
+local rbuffer = helpers.cimport("./test/unit/fixtures/rbuffer.h")
+
+describe('rbuffer functions', function()
+  local capacity = 16
+  local rbuf
+
+  local function inspect()
+    return ffi.string(rbuf.start_ptr, capacity)
+  end
+
+  local function write(str)
+    local buf = to_cstr(str)
+    return rbuffer.rbuffer_write(rbuf, buf, #str)
+  end
+
+  local function read(len)
+    local buf = cstr(len)
+    len = rbuffer.rbuffer_read(rbuf, buf, len)
+    return ffi.string(buf, len)
+  end
+
+  local function get(idx)
+    return ffi.string(rbuffer.rbuffer_get(rbuf, idx), 1)
+  end
+
+  before_each(function()
+    rbuf = ffi.gc(rbuffer.rbuffer_new(capacity), rbuffer.rbuffer_free)
+    -- fill the internal buffer with the character '0' to simplify inspecting
+    ffi.C.memset(rbuf.start_ptr, string.byte('0'), capacity)
+  end)
+
+  describe('RBUFFER_UNTIL_FULL', function()
+    local chunks
+
+    local function collect_write_chunks()
+      rbuffer.ut_rbuffer_each_write_chunk(rbuf, function(wptr, wcnt)
+        table.insert(chunks, ffi.string(wptr, wcnt))
+      end)
+    end
+
+    before_each(function()
+      chunks = {}
+    end)
+
+    describe('with empty buffer in one contiguous chunk', function()
+      it('is called once with the empty chunk', function()
+        collect_write_chunks()
+        eq({'0000000000000000'}, chunks)
+      end)
+    end)
+
+    describe('with partially empty buffer in one contiguous chunk', function()
+      before_each(function()
+        write('string')
+      end)
+
+      it('is called once with the empty chunk', function()
+        collect_write_chunks()
+        eq({'0000000000'}, chunks)
+      end)
+    end)
+
+    describe('with filled buffer in one contiguous chunk', function()
+      before_each(function()
+        write('abcdefghijklmnopq')
+      end)
+
+      it('is not called', function()
+        collect_write_chunks()
+        eq({}, chunks)
+      end)
+    end)
+
+    describe('with buffer partially empty in two contiguous chunks', function()
+      before_each(function()
+        write('1234567890')
+        read(8)
+      end)
+
+      it('is called twice with each filled chunk', function()
+        collect_write_chunks()
+        eq({'000000', '12345678'}, chunks)
+      end)
+    end)
+
+    describe('with buffer empty in two contiguous chunks', function()
+      before_each(function()
+        write('12345678')
+        read(8)
+      end)
+
+      it('is called twice with each filled chunk', function()
+        collect_write_chunks()
+        eq({'00000000', '12345678'}, chunks)
+      end)
+    end)
+
+    describe('with buffer filled in two contiguous chunks', function()
+      before_each(function()
+        write('12345678')
+        read(8)
+        write('abcdefghijklmnopq')
+      end)
+
+      it('is not called', function()
+        collect_write_chunks()
+        eq({}, chunks)
+      end)
+    end)
+  end)
+
+  describe('RBUFFER_UNTIL_EMPTY', function()
+    local chunks
+
+    local function collect_read_chunks()
+      rbuffer.ut_rbuffer_each_read_chunk(rbuf, function(rptr, rcnt)
+        table.insert(chunks, ffi.string(rptr, rcnt))
+      end)
+    end
+
+    before_each(function()
+      chunks = {}
+    end)
+
+    describe('with empty buffer', function()
+      it('is not called', function()
+        collect_read_chunks()
+        eq({}, chunks)
+      end)
+    end)
+
+    describe('with partially filled buffer in one contiguous chunk', function()
+      before_each(function()
+        write('string')
+      end)
+
+      it('is called once with the filled chunk', function()
+        collect_read_chunks()
+        eq({'string'}, chunks)
+      end)
+    end)
+
+    describe('with filled buffer in one contiguous chunk', function()
+      before_each(function()
+        write('abcdefghijklmnopq')
+      end)
+
+      it('is called once with the filled chunk', function()
+        collect_read_chunks()
+        eq({'abcdefghijklmnop'}, chunks)
+      end)
+    end)
+
+    describe('with buffer partially filled in two contiguous chunks', function()
+      before_each(function()
+        write('1234567890')
+        read(10)
+        write('long string')
+      end)
+
+      it('is called twice with each filled chunk', function()
+        collect_read_chunks()
+        eq({'long s', 'tring'}, chunks)
+      end)
+    end)
+
+    describe('with buffer filled in two contiguous chunks', function()
+      before_each(function()
+        write('12345678')
+        read(8)
+        write('abcdefghijklmnopq')
+      end)
+
+      it('is called twice with each filled chunk', function()
+        collect_read_chunks()
+        eq({'abcdefgh', 'ijklmnop'}, chunks)
+      end)
+    end)
+  end)
+
+  describe('RBUFFER_EACH', function()
+    local chars
+
+    local function collect_chars()
+      rbuffer.ut_rbuffer_each(rbuf, function(c, i)
+        table.insert(chars, {string.char(c), tonumber(i)})
+      end)
+    end
+    before_each(function()
+      chars = {}
+    end)
+
+    describe('with empty buffer', function()
+      it('is not called', function()
+        collect_chars()
+        eq({}, chars)
+      end)
+    end)
+
+    describe('with buffer filled in two contiguous chunks', function()
+      before_each(function()
+        write('1234567890')
+        read(10)
+        write('long string')
+      end)
+
+      it('collects each character and index', function()
+        collect_chars()
+        eq({{'l', 0}, {'o', 1}, {'n', 2}, {'g', 3}, {' ', 4}, {'s', 5},
+            {'t', 6}, {'r', 7}, {'i', 8}, {'n', 9}, {'g', 10}}, chars)
+      end)
+    end)
+  end)
+
+  describe('RBUFFER_EACH_REVERSE', function()
+    local chars
+
+    local function collect_chars()
+      rbuffer.ut_rbuffer_each_reverse(rbuf, function(c, i)
+        table.insert(chars, {string.char(c), tonumber(i)})
+      end)
+    end
+    before_each(function()
+      chars = {}
+    end)
+
+    describe('with empty buffer', function()
+      it('is not called', function()
+        collect_chars()
+        eq({}, chars)
+      end)
+    end)
+
+    describe('with buffer filled in two contiguous chunks', function()
+      before_each(function()
+        write('1234567890')
+        read(10)
+        write('long string')
+      end)
+
+      it('collects each character and index', function()
+        collect_chars()
+        eq({{'g', 10}, {'n', 9}, {'i', 8}, {'r', 7}, {'t', 6}, {'s', 5},
+            {' ', 4}, {'g', 3}, {'n', 2}, {'o', 1}, {'l', 0}}, chars)
+      end)
+    end)
+  end)
+
+  describe('rbuffer_cmp', function()
+    local function cmp(str)
+      local rv = rbuffer.rbuffer_cmp(rbuf, to_cstr(str), #str)
+      if rv == 0 then
+        return 0
+      else
+        return rv / math.abs(rv)
+      end
+    end
+
+    describe('with buffer filled in two contiguous chunks', function()
+      before_each(function()
+        write('1234567890')
+        read(10)
+        write('long string')
+      end)
+
+      it('compares the common longest sequence', function()
+        eq(0, cmp('long string'))
+        eq(0, cmp('long strin'))
+        eq(-1, cmp('long striM'))
+        eq(1, cmp('long strio'))
+        eq(0, cmp('long'))
+        eq(-1, cmp('lonG'))
+        eq(1, cmp('lonh'))
+      end)
+    end)
+
+    describe('with empty buffer', function()
+      it('returns 0 since no characters are compared', function()
+        eq(0, cmp(''))
+      end)
+    end)
+  end)
+
+  describe('rbuffer_write', function()
+    it('fills the internal buffer and returns the write count', function()
+      eq(12, write('short string'))
+      eq('short string0000', inspect())
+    end)
+
+    it('wont write beyond capacity', function()
+      eq(16, write('very very long string'))
+      eq('very very long s', inspect())
+    end)
+  end)
+
+  describe('rbuffer_read', function()
+    it('reads what was previously written', function()
+      write('to read')
+      eq('to read', read(20))
+    end)
+
+    it('reads nothing if the buffer is empty', function()
+      eq('', read(20))
+      write('empty')
+      eq('empty', read(20))
+      eq('', read(20))
+    end)
+  end)
+
+  describe('rbuffer_get', function()
+    it('fetch the pointer at offset, wrapping if required', function()
+      write('1234567890')
+      read(10)
+      write('long string')
+      eq('l', get(0))
+      eq('o', get(1))
+      eq('n', get(2))
+      eq('g', get(3))
+      eq(' ', get(4))
+      eq('s', get(5))
+      eq('t', get(6))
+      eq('r', get(7))
+      eq('i', get(8))
+      eq('n', get(9))
+      eq('g', get(10))
+    end)
+  end)
+
+  describe('wrapping behavior', function()
+    it('writing/reading wraps across the end of the internal buffer', function()
+      write('1234567890')
+      eq('1234', read(4))
+      eq('5678', read(4))
+      write('987654321')
+      eq('3214567890987654', inspect())
+      eq('90987654321', read(20))
+      eq('', read(4))
+      write('abcdefghijklmnopqrs')
+      eq('nopabcdefghijklm', inspect())
+      eq('abcdefghijklmnop', read(20))
+    end)
+  end)
+end)
diff --git a/test/unit/tempfile_spec.lua b/test/unit/tempfile_spec.lua
index 6484a98b8f..e558ff04c8 100644
--- a/test/unit/tempfile_spec.lua
+++ b/test/unit/tempfile_spec.lua
@@ -4,8 +4,6 @@ local helpers = require 'test.unit.helpers'
 local os = helpers.cimport './src/nvim/os/os.h'
 local tempfile = helpers.cimport './src/nvim/tempfile.h'
 
-helpers.vim_init()
-
 describe('tempfile related functions', function()
   after_each(function()
     tempfile.vim_deltempdir()