neovim

socket.c (11241B)

---

#include <assert.h>
#include <inttypes.h>
#include <stdbool.h>
#include <stdio.h>
#include <string.h>
#include <uv.h>

#include "nvim/ascii_defs.h"
#include "nvim/charset.h"
#include "nvim/event/defs.h"
#include "nvim/event/loop.h"
#include "nvim/event/multiqueue.h"
#include "nvim/event/socket.h"
#include "nvim/event/stream.h"
#include "nvim/gettext_defs.h"
#include "nvim/log.h"
#include "nvim/main.h"
#include "nvim/memory.h"
#include "nvim/os/fs.h"
#include "nvim/os/os_defs.h"
#include "nvim/path.h"
#include "nvim/types_defs.h"

#include "event/socket.c.generated.h"

/// Checks if an address string looks like a TCP endpoint, and returns the end of the host part.
///
/// @param address Address string
/// @return pointer to the end of the host part of the address, or NULL if it is not a TCP address
char *socket_address_tcp_host_end(const char *address)
{
 if (address == NULL) {
   return NULL;
 }

 // Windows drive letter path: "X:\..." or "X:/..." is a local path, not TCP.
 if (ASCII_ISALPHA((uint8_t)address[0]) && address[1] == ':'
     && (address[2] == '\\' || address[2] == '/')) {
   return NULL;
 }

 char *colon = strrchr(address, ':');
 return colon != NULL && colon != address ? colon : NULL;
}

int socket_watcher_init(Loop *loop, SocketWatcher *watcher, const char *endpoint)
 FUNC_ATTR_NONNULL_ALL
{
 xstrlcpy(watcher->addr, endpoint, sizeof(watcher->addr));
 char *addr = watcher->addr;
 char *host_end = socket_address_tcp_host_end(addr);

 if (host_end) {
   // Split user specified address into two strings, addr (hostname) and port.
   // The port part in watcher->addr will be updated later.
   *host_end = NUL;
   char *port = host_end + 1;
   intmax_t iport;

   int ok = try_getdigits(&(char *){ port }, &iport);
   if (!ok || iport < 0 || iport > UINT16_MAX) {
     ELOG("Invalid port: %s", port);
     return UV_EINVAL;
   }

   if (*port == NUL) {
     // When no port is given, (uv_)getaddrinfo expects NULL otherwise the
     // implementation may attempt to lookup the service by name (and fail)
     port = NULL;
   }

   uv_getaddrinfo_t request;

   int retval = uv_getaddrinfo(&loop->uv, &request, NULL, addr, port,
                               &(struct addrinfo){ .ai_family = AF_UNSPEC,
                                                   .ai_socktype = SOCK_STREAM, });
   if (retval != 0) {
     ELOG("Host lookup failed: %s", endpoint);
     return retval;
   }
   watcher->uv.tcp.addrinfo = request.addrinfo;

   uv_tcp_init(&loop->uv, &watcher->uv.tcp.handle);
   uv_tcp_nodelay(&watcher->uv.tcp.handle, true);
   watcher->stream = (uv_stream_t *)(&watcher->uv.tcp.handle);
 } else {
   uv_pipe_init(&loop->uv, &watcher->uv.pipe.handle, 0);
   watcher->stream = (uv_stream_t *)(&watcher->uv.pipe.handle);
 }

 watcher->stream->data = watcher;
 watcher->cb = NULL;
 watcher->close_cb = NULL;
 watcher->events = NULL;
 watcher->data = NULL;

 return 0;
}

/// Callback for closing a handle initialized by socket_connect().
static void connect_close_cb(uv_handle_t *handle)
{
 bool *closed = handle->data;
 *closed = true;
}

/// Check if a socket is alive by attempting to connect to it.
/// @param loop Event loop
/// @param addr Socket address to probe
/// @return true if socket is alive (connection succeeded), false otherwise
static bool socket_alive(Loop *loop, const char *addr)
{
 RStream stream;
 const char *error = NULL;

 // Try to connect with a 500ms timeout (fast failure for dead sockets)
 bool connected = socket_connect(loop, &stream, false, addr, 500, &error);
 if (!connected) {
   return false;
 }

 // Connection succeeded - socket is alive. Close the probe connection properly.
 bool closed = false;
 stream.s.uv.pipe.data = &closed;
 uv_close((uv_handle_t *)&stream.s.uv.pipe, connect_close_cb);
 LOOP_PROCESS_EVENTS_UNTIL(&main_loop, NULL, -1, closed);

 return true;
}

int socket_watcher_start(SocketWatcher *watcher, int backlog, socket_cb cb)
 FUNC_ATTR_NONNULL_ALL
{
 watcher->cb = cb;
 int result = UV_EINVAL;

 if (watcher->stream->type == UV_TCP) {
   struct addrinfo *ai = watcher->uv.tcp.addrinfo;

   for (; ai; ai = ai->ai_next) {
     result = uv_tcp_bind(&watcher->uv.tcp.handle, ai->ai_addr, 0);
     if (result != 0) {
       continue;
     }
     result = uv_listen(watcher->stream, backlog, connection_cb);
     if (result == 0) {
       struct sockaddr_storage sas;

       // When the endpoint in socket_watcher_init() didn't specify a port
       // number, a free random port number will be assigned. sin_port will
       // contain 0 in this case, unless uv_tcp_getsockname() is used first.
       uv_tcp_getsockname(&watcher->uv.tcp.handle, (struct sockaddr *)&sas,
                          &(int){ sizeof(sas) });
       uint16_t port = (sas.ss_family == AF_INET) ? ((struct sockaddr_in *)(&sas))->sin_port
                                                  : ((struct sockaddr_in6 *)(&sas))->sin6_port;
       // v:servername uses the string from watcher->addr
       size_t len = strlen(watcher->addr);
       snprintf(watcher->addr + len, sizeof(watcher->addr) - len, ":%" PRIu16,
                ntohs(port));
       break;
     }
   }
   uv_freeaddrinfo(watcher->uv.tcp.addrinfo);
 } else {
   result = uv_pipe_bind(&watcher->uv.pipe.handle, watcher->addr);

   // If bind failed with EACCES/EADDRINUSE, check if socket is stale
   if (result == UV_EACCES || result == UV_EADDRINUSE) {
     Loop *loop = watcher->stream->loop->data;

     if (!socket_alive(loop, watcher->addr)) {
       // Socket exists but is dead - remove it
       ILOG("Removing stale socket: %s", watcher->addr);
       int rm_result = os_remove(watcher->addr);

       if (rm_result != 0) {
         WLOG("Failed to remove stale socket %s: %s",
              watcher->addr, uv_strerror(rm_result));
       } else {
         // Close and reinit the pipe handle before retrying bind
         uv_loop_t *uv_loop = watcher->uv.pipe.handle.loop;
         bool closed = false;
         watcher->uv.pipe.handle.data = &closed;
         uv_close((uv_handle_t *)&watcher->uv.pipe.handle, connect_close_cb);
         LOOP_PROCESS_EVENTS_UNTIL(&main_loop, NULL, -1, closed);

         uv_pipe_init(uv_loop, &watcher->uv.pipe.handle, 0);
         watcher->stream = (uv_stream_t *)(&watcher->uv.pipe.handle);
         watcher->stream->data = watcher;

         // Retry bind with fresh handle
         result = uv_pipe_bind(&watcher->uv.pipe.handle, watcher->addr);
       }
     } else {
       // Socket is alive - this is a real error
       ELOG("Socket already in use by another Nvim instance: %s", watcher->addr);
     }
   }

   if (result == 0) {
     result = uv_listen(watcher->stream, backlog, connection_cb);
   }
 }

 assert(result <= 0);  // libuv should return negative error code or zero.
 if (result < 0) {
   if (result == UV_EACCES) {
     // Libuv converts ENOENT to EACCES for Windows compatibility, but if
     // the parent directory does not exist, ENOENT would be more accurate.
     *path_tail(watcher->addr) = NUL;
     if (!os_path_exists(watcher->addr)) {
       result = UV_ENOENT;
     }
   }
   return result;
 }

 return 0;
}

int socket_watcher_accept(SocketWatcher *watcher, RStream *stream)
 FUNC_ATTR_NONNULL_ARG(1) FUNC_ATTR_NONNULL_ARG(2)
{
 uv_stream_t *client;

 if (watcher->stream->type == UV_TCP) {
   client = (uv_stream_t *)(&stream->s.uv.tcp);
   uv_tcp_init(watcher->uv.tcp.handle.loop, (uv_tcp_t *)client);
   uv_tcp_nodelay((uv_tcp_t *)client, true);
 } else {
   client = (uv_stream_t *)&stream->s.uv.pipe;
   uv_pipe_init(watcher->uv.pipe.handle.loop, (uv_pipe_t *)client, 0);
 }

 int result = uv_accept(watcher->stream, client);

 if (result) {
   uv_close((uv_handle_t *)client, NULL);
   return result;
 }

 stream_init(NULL, &stream->s, -1, false, client);
 return 0;
}

void socket_watcher_close(SocketWatcher *watcher, socket_close_cb cb)
 FUNC_ATTR_NONNULL_ARG(1)
{
 watcher->close_cb = cb;
 uv_close((uv_handle_t *)watcher->stream, close_cb);
}

static void connection_event(void **argv)
{
 SocketWatcher *watcher = argv[0];
 int status = (int)(uintptr_t)(argv[1]);
 watcher->cb(watcher, status, watcher->data);
}

static void connection_cb(uv_stream_t *handle, int status)
{
 SocketWatcher *watcher = handle->data;
 CREATE_EVENT(watcher->events, connection_event, watcher, (void *)(uintptr_t)status);
}

static void close_cb(uv_handle_t *handle)
{
 SocketWatcher *watcher = handle->data;
 if (watcher->close_cb) {
   watcher->close_cb(watcher, watcher->data);
 }
}

static void connect_cb(uv_connect_t *req, int status)
{
 int *ret_status = req->data;
 *ret_status = status;
 uv_handle_t *handle = (uv_handle_t *)req->handle;
 if (status != 0 && !uv_is_closing(handle)) {
   uv_close(handle, connect_close_cb);
 }
}

bool socket_connect(Loop *loop, RStream *stream, bool is_tcp, const char *address, int timeout,
                   const char **error)
{
 bool success = false;
 bool closed;
 int status;
 uv_connect_t req;
 req.data = &status;
 uv_stream_t *uv_stream;

 uv_tcp_t *tcp = &stream->s.uv.tcp;
 uv_getaddrinfo_t addr_req;
 addr_req.addrinfo = NULL;
 const struct addrinfo *addrinfo = NULL;
 char *addr = NULL;
 if (is_tcp) {
   addr = xstrdup(address);
   char *host_end = strrchr(addr, ':');
   if (!host_end) {
     *error = _("tcp address must be host:port");
     goto cleanup;
   }
   *host_end = NUL;

   const struct addrinfo hints = { .ai_family = AF_UNSPEC,
                                   .ai_socktype = SOCK_STREAM,
                                   .ai_flags = AI_NUMERICSERV };
   int retval = uv_getaddrinfo(&loop->uv, &addr_req, NULL,
                               addr, host_end + 1, &hints);
   if (retval != 0) {
     *error = _("failed to lookup host or port");
     goto cleanup;
   }
   addrinfo = addr_req.addrinfo;

tcp_retry:
   uv_tcp_init(&loop->uv, tcp);
   uv_tcp_nodelay(tcp, true);
   uv_tcp_connect(&req,  tcp, addrinfo->ai_addr, connect_cb);
   uv_stream = (uv_stream_t *)tcp;
 } else {
   uv_pipe_t *pipe = &stream->s.uv.pipe;
   uv_pipe_init(&loop->uv, pipe, 0);
   uv_pipe_connect(&req,  pipe, address, connect_cb);
   uv_stream = (uv_stream_t *)pipe;
 }
 uv_stream->data = &closed;
 closed = false;
 status = 1;
 LOOP_PROCESS_EVENTS_UNTIL(&main_loop, NULL, timeout, status != 1);
 if (status == 0) {
   stream_init(NULL, &stream->s, -1, false, uv_stream);
   assert(uv_stream->data != &closed);  // Should have been set by stream_init().
   success = true;
 } else {
   if (!uv_is_closing((uv_handle_t *)uv_stream)) {
     uv_close((uv_handle_t *)uv_stream, connect_close_cb);
   }
   // Wait for the close callback to arrive before retrying or returning, otherwise
   // it may lead to a hang or stack-use-after-return.
   LOOP_PROCESS_EVENTS_UNTIL(&main_loop, NULL, -1, closed);

   if (is_tcp && addrinfo->ai_next) {
     addrinfo = addrinfo->ai_next;
     goto tcp_retry;
   } else {
     *error = _("connection refused");
   }
 }

cleanup:
 xfree(addr);
 uv_freeaddrinfo(addr_req.addrinfo);
 return success;
}