neovim/src/nvim/event/stream.c

#include <assert.h>
#include <stdbool.h>
#include <stddef.h>
#include <uv.h>
#include <uv/version.h>

#include "nvim/event/defs.h"
#include "nvim/event/loop.h"
#include "nvim/event/stream.h"
#include "nvim/log.h"
#include "nvim/rbuffer.h"
#ifdef MSWIN
# include "nvim/os/os_win_console.h"
#endif

#ifdef INCLUDE_GENERATED_DECLARATIONS
# include "event/stream.c.generated.h"
#endif

// For compatibility with libuv < 1.19.0 (tested on 1.18.0)
#if UV_VERSION_MINOR < 19
# define uv_stream_get_write_queue_size(stream) stream->write_queue_size
#endif

/// Sets the stream associated with `fd` to "blocking" mode.
///
/// @return `0` on success, or libuv error code on failure.
int stream_set_blocking(int fd, bool blocking)
{
  // Private loop to avoid conflict with existing watcher(s):
  //    uv__io_stop: Assertion `loop->watchers[w->fd] == w' failed.
  uv_loop_t loop;
  uv_pipe_t stream;
  uv_loop_init(&loop);
  uv_pipe_init(&loop, &stream, 0);
  uv_pipe_open(&stream, fd);
  int retval = uv_stream_set_blocking((uv_stream_t *)&stream, blocking);
  uv_close((uv_handle_t *)&stream, NULL);
  uv_run(&loop, UV_RUN_NOWAIT);  // not necessary, but couldn't hurt.
  uv_loop_close(&loop);
  return retval;
}

void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream)
  FUNC_ATTR_NONNULL_ARG(2)
{
  stream->uvstream = uvstream;

  if (fd >= 0) {
    uv_handle_type type = uv_guess_handle(fd);
    stream->fd = fd;

    if (type == UV_FILE) {
      // Non-blocking file reads are simulated with an idle handle that reads in
      // chunks of the ring buffer size, giving time for other events to be
      // processed between reads.
      uv_idle_init(&loop->uv, &stream->uv.idle);
      stream->uv.idle.data = stream;
    } else {
      assert(type == UV_NAMED_PIPE || type == UV_TTY);
#ifdef MSWIN
      if (type == UV_TTY) {
        uv_tty_init(&loop->uv, &stream->uv.tty, fd, 0);
        uv_tty_set_mode(&stream->uv.tty, UV_TTY_MODE_RAW);
        DWORD dwMode;
        if (GetConsoleMode(stream->uv.tty.handle, &dwMode)) {
          dwMode |= ENABLE_VIRTUAL_TERMINAL_INPUT;
          SetConsoleMode(stream->uv.tty.handle, dwMode);
        }
        stream->uvstream = (uv_stream_t *)&stream->uv.tty;
      } else {
#endif
      uv_pipe_init(&loop->uv, &stream->uv.pipe, 0);
      uv_pipe_open(&stream->uv.pipe, fd);
      stream->uvstream = (uv_stream_t *)&stream->uv.pipe;
#ifdef MSWIN
    }
#endif
    }
  }

  if (stream->uvstream) {
    stream->uvstream->data = stream;
  }

  stream->internal_data = NULL;
  stream->fpos = 0;
  stream->curmem = 0;
  stream->maxmem = 0;
  stream->pending_reqs = 0;
  stream->read_cb = NULL;
  stream->write_cb = NULL;
  stream->close_cb = NULL;
  stream->internal_close_cb = NULL;
  stream->closed = false;
  stream->buffer = NULL;
  stream->events = NULL;
  stream->num_bytes = 0;
}

void stream_close(Stream *stream, stream_close_cb on_stream_close, void *data)
  FUNC_ATTR_NONNULL_ARG(1)
{
  assert(!stream->closed);
  DLOG("closing Stream: %p", (void *)stream);
  stream->closed = true;
  stream->close_cb = on_stream_close;
  stream->close_cb_data = data;

#ifdef MSWIN
  if (UV_TTY == uv_guess_handle(stream->fd)) {
    // Undo UV_TTY_MODE_RAW from stream_init(). #10801
    uv_tty_set_mode(&stream->uv.tty, UV_TTY_MODE_NORMAL);
  }
#endif

  if (!stream->pending_reqs) {
    stream_close_handle(stream);
  }
}

void stream_may_close(Stream *stream)
{
  if (!stream->closed) {
    stream_close(stream, NULL, NULL);
  }
}

void stream_close_handle(Stream *stream)
  FUNC_ATTR_NONNULL_ALL
{
  uv_handle_t *handle = NULL;
  if (stream->uvstream) {
    if (uv_stream_get_write_queue_size(stream->uvstream) > 0) {
      WLOG("closed Stream (%p) with %zu unwritten bytes",
           (void *)stream,
           uv_stream_get_write_queue_size(stream->uvstream));
    }
    handle = (uv_handle_t *)stream->uvstream;
  } else {
    handle = (uv_handle_t *)&stream->uv.idle;
  }

  assert(handle != NULL);

  if (!uv_is_closing(handle)) {
    uv_close(handle, close_cb);
  }
}

static void close_cb(uv_handle_t *handle)
{
  Stream *stream = handle->data;
  if (stream->buffer) {
    rbuffer_free(stream->buffer);
  }
  if (stream->close_cb) {
    stream->close_cb(stream, stream->close_cb_data);
  }
  if (stream->internal_close_cb) {
    stream->internal_close_cb(stream, stream->internal_data);
  }
}