/src/bind9/lib/isc/netmgr/tcp.c

Source
/*
 * Copyright (C) Internet Systems Consortium, Inc. ("ISC")
 *
 * SPDX-License-Identifier: MPL-2.0
 *
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, you can obtain one at https://mozilla.org/MPL/2.0/.
 *
 * See the COPYRIGHT file distributed with this work for additional
 * information regarding copyright ownership.
 */

#include <libgen.h>
#include <unistd.h>

#include <isc/async.h>
#include <isc/atomic.h>
#include <isc/barrier.h>
#include <isc/buffer.h>
#include <isc/errno.h>
#include <isc/log.h>
#include <isc/magic.h>
#include <isc/mem.h>
#include <isc/netmgr.h>
#include <isc/quota.h>
#include <isc/random.h>
#include <isc/refcount.h>
#include <isc/region.h>
#include <isc/result.h>
#include <isc/sockaddr.h>
#include <isc/stdtime.h>
#include <isc/thread.h>
#include <isc/util.h>
#include <isc/uv.h>

#include "../loop_p.h"
#include "netmgr-int.h"

static atomic_uint_fast32_t last_tcpquota_log = 0;

static bool
can_log_tcp_quota(void) {
  isc_stdtime_t last;
  isc_stdtime_t now = isc_stdtime_now();
  last = atomic_exchange_relaxed(&last_tcpquota_log, now);
  if (now != last) {
    return true;
  }

  return false;
}

static isc_result_t
tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req);

static isc_result_t
tcp_send_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req);
static void
tcp_connect_cb(uv_connect_t *uvreq, int status);
static void
tcp_stop_cb(uv_handle_t *handle);

static void
tcp_connection_cb(uv_stream_t *server, int status);

static void
tcp_close_cb(uv_handle_t *uvhandle);

static isc_result_t
accept_connection(isc_nmsocket_t *ssock);

static void
quota_accept_cb(void *arg);

static void
tcp_dbg_log(const isc_nmsocket_t *sock, const isc_result_t result,
      const char *msg) {
  const int level = ISC_LOG_DEBUG(99);

  if (!isc_log_wouldlog(level)) {
    return;
  }

  char err_msg[256];
  char peer_sabuf[ISC_SOCKADDR_FORMATSIZE];
  char local_sabuf[ISC_SOCKADDR_FORMATSIZE];
  const bool has_peer_info = !sock->accepting && sock->recv_cb != NULL;

  err_msg[0] = peer_sabuf[0] = local_sabuf[0] = '\0';

  isc_sockaddr_format(&sock->iface, local_sabuf, sizeof(local_sabuf));

  if (has_peer_info) {
    isc_sockaddr_format(&sock->peer, peer_sabuf,
            sizeof(peer_sabuf));
  }

  if (result != ISC_R_SUCCESS) {
    snprintf(err_msg, sizeof(err_msg), " (error: %s)",
       isc_result_totext(result));
  }

  isc_log_write(NS_LOGCATEGORY_GENERAL, ISC_LOGMODULE_NETMGR, level,
          " (%s%son %s): %s%s", peer_sabuf,
          has_peer_info ? " " : "", local_sabuf, msg, err_msg);
}

static isc_result_t
tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
  isc__networker_t *worker = NULL;
  isc_result_t result = ISC_R_UNSET;
  int r;

  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(VALID_UVREQ(req));

  REQUIRE(sock->tid == isc_tid());

  worker = sock->worker;

  sock->connecting = true;

  /* 2 minute timeout */
  result = isc__nm_socket_connectiontimeout(sock->fd, 120 * 1000);
  RUNTIME_CHECK(result == ISC_R_SUCCESS);

  r = uv_tcp_init(&worker->loop->loop, &sock->uv_handle.tcp);
  UV_RUNTIME_CHECK(uv_tcp_init, r);
  uv_handle_set_data(&sock->uv_handle.handle, sock);

  r = uv_timer_init(&worker->loop->loop, &sock->read_timer);
  UV_RUNTIME_CHECK(uv_timer_init, r);
  uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);

  r = uv_tcp_open(&sock->uv_handle.tcp, sock->fd);
  if (r != 0) {
    isc__nm_closesocket(sock->fd);
    isc__nm_incstats(sock, STATID_OPENFAIL);
    return isc_uverr2result(r);
  }
  isc__nm_incstats(sock, STATID_OPEN);

  if (req->local.length != 0) {
    r = uv_tcp_bind(&sock->uv_handle.tcp, &req->local.type.sa, 0);
    if (r != 0) {
      isc__nm_incstats(sock, STATID_BINDFAIL);
      return isc_uverr2result(r);
    }
  }

  isc__nm_set_network_buffers(&sock->uv_handle.handle);

  uv_handle_set_data(&req->uv_req.handle, req);
  r = uv_tcp_connect(&req->uv_req.connect, &sock->uv_handle.tcp,
         &req->peer.type.sa, tcp_connect_cb);
  if (r != 0) {
    isc__nm_incstats(sock, STATID_CONNECTFAIL);
    return isc_uverr2result(r);
  }

  uv_handle_set_data((uv_handle_t *)&sock->read_timer,
         &req->uv_req.connect);
  isc__nmsocket_timer_start(sock);

  return ISC_R_SUCCESS;
}

static void
tcp_connect_cb(uv_connect_t *uvreq, int status) {
  isc_result_t result = ISC_R_UNSET;
  isc__nm_uvreq_t *req = NULL;
  isc_nmsocket_t *sock = uv_handle_get_data((uv_handle_t *)uvreq->handle);
  struct sockaddr_storage ss;
  isc__networker_t *worker = NULL;
  int r;

  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(sock->tid == isc_tid());

  worker = sock->worker;

  req = uv_handle_get_data((uv_handle_t *)uvreq);

  REQUIRE(VALID_UVREQ(req));
  REQUIRE(VALID_NMHANDLE(req->handle));

  INSIST(sock->connecting);

  if (sock->timedout || status == UV_ETIMEDOUT) {
    /* Connection timed-out */
    result = ISC_R_TIMEDOUT;
    goto error;
  } else if (isc__nm_closing(worker)) {
    /* Network manager shutting down */
    result = ISC_R_SHUTTINGDOWN;
    goto error;
  } else if (isc__nmsocket_closing(sock)) {
    /* Connection canceled */
    result = ISC_R_CANCELED;
    goto error;
  } else if (status == UV_EADDRINUSE) {
    /*
     * On FreeBSD the TCP connect() call sometimes results in a
     * spurious transient EADDRINUSE. Try a few more times before
     * giving up.
     */
    if (--req->connect_tries > 0) {
      r = uv_tcp_connect(&req->uv_req.connect,
             &sock->uv_handle.tcp,
             &req->peer.type.sa, tcp_connect_cb);
      if (r != 0) {
        result = isc_uverr2result(r);
        goto error;
      }
      return;
    }
    result = isc_uverr2result(status);
    goto error;
  } else if (status != 0) {
    result = isc_uverr2result(status);
    goto error;
  }

  isc__nmsocket_timer_stop(sock);
  uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);

  isc__nm_incstats(sock, STATID_CONNECT);
  r = uv_tcp_getpeername(&sock->uv_handle.tcp, (struct sockaddr *)&ss,
             &(int){ sizeof(ss) });
  if (r != 0) {
    result = isc_uverr2result(r);
    goto error;
  }

  sock->connecting = false;
  sock->connected = true;

  result = isc_sockaddr_fromsockaddr(&sock->peer, (struct sockaddr *)&ss);
  RUNTIME_CHECK(result == ISC_R_SUCCESS);

  isc__nm_connectcb(sock, req, ISC_R_SUCCESS, false);

  return;
error:
  isc__nm_failed_connect_cb(sock, req, result, false);
}

void
isc_nm_tcpconnect(isc_sockaddr_t *local, isc_sockaddr_t *peer,
      isc_nm_cb_t connect_cb, void *connect_cbarg,
      unsigned int timeout) {
  isc_result_t result = ISC_R_SUCCESS;
  isc_nmsocket_t *sock = NULL;
  isc__nm_uvreq_t *req = NULL;
  sa_family_t sa_family;
  isc__networker_t *worker = isc__networker_current();
  uv_os_sock_t fd = -1;

  REQUIRE(local != NULL);
  REQUIRE(peer != NULL);

  if (isc__nm_closing(worker)) {
    connect_cb(NULL, ISC_R_SHUTTINGDOWN, connect_cbarg);
    return;
  }

  sa_family = peer->type.sa.sa_family;

  result = isc__nm_socket(sa_family, SOCK_STREAM, 0, &fd);
  if (result != ISC_R_SUCCESS) {
    connect_cb(NULL, result, connect_cbarg);
    return;
  }

  sock = isc_mempool_get(worker->nmsocket_pool);
  isc__nmsocket_init(sock, worker, isc_nm_tcpsocket, local, NULL);

  sock->connect_timeout = timeout;
  sock->fd = fd;
  sock->client = true;

  req = isc__nm_uvreq_get(sock);
  req->cb.connect = connect_cb;
  req->cbarg = connect_cbarg;
  req->peer = *peer;
  req->local = *local;
  req->handle = isc__nmhandle_get(sock, &req->peer, &sock->iface);

  (void)isc__nm_socket_min_mtu(sock->fd, sa_family);
  (void)isc__nm_socket_tcp_maxseg(sock->fd, NM_MAXSEG);

  sock->active = true;

  result = tcp_connect_direct(sock, req);
  if (result != ISC_R_SUCCESS) {
    sock->active = false;
    isc__nm_tcp_close(sock);
    isc__nm_connectcb(sock, req, result, true);
  }

  /*
   * The sock is now attached to the handle.
   */
  isc__nmsocket_detach(&sock);
}

static uv_os_sock_t
isc__nm_tcp_lb_socket(sa_family_t sa_family) {
  isc_result_t result;
  uv_os_sock_t sock;

  result = isc__nm_socket(sa_family, SOCK_STREAM, 0, &sock);
  RUNTIME_CHECK(result == ISC_R_SUCCESS);

  (void)isc__nm_socket_v6only(sock, sa_family);

  /* FIXME: set mss */

  result = isc__nm_socket_reuse(sock, 1);
  RUNTIME_CHECK(result == ISC_R_SUCCESS);

  if (isc__netmgr->load_balance_sockets) {
    result = isc__nm_socket_reuse_lb(sock);
    RUNTIME_CHECK(result == ISC_R_SUCCESS);
  }

  return sock;
}

static void
start_tcp_child_job(void *arg) {
  isc_nmsocket_t *sock = arg;

  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(VALID_NMSOCK(sock->parent));
  REQUIRE(sock->type == isc_nm_tcpsocket);
  REQUIRE(sock->tid == isc_tid());

  sa_family_t sa_family = sock->iface.type.sa.sa_family;
  int r, flags = 0;
  isc_result_t result = ISC_R_UNSET;
  isc_loop_t *loop = sock->worker->loop;
  struct sockaddr_storage ss;

  (void)isc__nm_socket_min_mtu(sock->fd, sa_family);
  (void)isc__nm_socket_tcp_maxseg(sock->fd, NM_MAXSEG);

  r = uv_tcp_init(&loop->loop, &sock->uv_handle.tcp);
  UV_RUNTIME_CHECK(uv_tcp_init, r);
  uv_handle_set_data(&sock->uv_handle.handle, sock);
  /* This keeps the socket alive after everything else is gone */
  isc__nmsocket_attach(sock, &(isc_nmsocket_t *){ NULL });

  r = uv_timer_init(&loop->loop, &sock->read_timer);
  UV_RUNTIME_CHECK(uv_timer_init, r);
  uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);

  r = uv_tcp_open(&sock->uv_handle.tcp, sock->fd);
  if (r < 0) {
    isc__nm_closesocket(sock->fd);
    isc__nm_incstats(sock, STATID_OPENFAIL);
    goto done;
  }
  isc__nm_incstats(sock, STATID_OPEN);

  if (sa_family == AF_INET6) {
    flags = UV_TCP_IPV6ONLY;
  }

  if (isc__netmgr->load_balance_sockets) {
    r = isc__nm_tcp_freebind(&sock->uv_handle.tcp,
           &sock->iface.type.sa, flags);
    if (r < 0) {
      isc__nm_incstats(sock, STATID_BINDFAIL);
      goto done;
    }
  } else if (sock->tid == 0) {
    r = isc__nm_tcp_freebind(&sock->uv_handle.tcp,
           &sock->iface.type.sa, flags);
    if (r < 0) {
      isc__nm_incstats(sock, STATID_BINDFAIL);
      goto done;
    }
    sock->parent->uv_handle.tcp.flags = sock->uv_handle.tcp.flags;
  } else {
    /* The socket is already bound, just copy the flags */
    sock->uv_handle.tcp.flags = sock->parent->uv_handle.tcp.flags;
  }

  isc__nm_set_network_buffers(&sock->uv_handle.handle);

  /*
   * The callback will run in the same thread uv_listen() was called
   * from, so a race with tcp_connection_cb() isn't possible.
   */
  r = uv_listen((uv_stream_t *)&sock->uv_handle.tcp, sock->backlog,
          tcp_connection_cb);
  if (r != 0) {
    isc__nmsocket_log(sock, ISC_LOG_ERROR, "uv_listen failed: %s",
          isc_result_totext(isc_uverr2result(r)));
    isc__nm_incstats(sock, STATID_BINDFAIL);
    goto done;
  }

  if (sock->tid == 0) {
    r = uv_tcp_getsockname(&sock->uv_handle.tcp,
               (struct sockaddr *)&ss,
               &(int){ sizeof(ss) });
    if (r != 0) {
      goto done;
    }

    result = isc_sockaddr_fromsockaddr(&sock->parent->iface,
               (struct sockaddr *)&ss);
    if (result != ISC_R_SUCCESS) {
      goto done_result;
    }
  }

done:
  result = isc_uverr2result(r);

done_result:
  if (result != ISC_R_SUCCESS) {
    sock->pquota = NULL;
  }

  sock->result = result;

  REQUIRE(!loop->paused);

  if (sock->tid != 0) {
    isc_barrier_wait(&sock->parent->listen_barrier);
  }
}

static void
start_tcp_child(isc_sockaddr_t *iface, isc_nmsocket_t *sock, uv_os_sock_t fd,
    isc_tid_t tid) {
  isc_nmsocket_t *csock = &sock->children[tid];
  isc__networker_t *worker = isc__networker_get(tid);

  isc__nmsocket_init(csock, worker, isc_nm_tcpsocket, iface, sock);
  csock->accept_cb = sock->accept_cb;
  csock->accept_cbarg = sock->accept_cbarg;
  csock->backlog = sock->backlog;

  /*
   * Quota isn't attached, just assigned.
   */
  csock->pquota = sock->pquota;

  if (isc__netmgr->load_balance_sockets) {
    UNUSED(fd);
    csock->fd = isc__nm_tcp_lb_socket(iface->type.sa.sa_family);
  } else {
    csock->fd = dup(fd);
  }
  REQUIRE(csock->fd >= 0);

  if (tid == 0) {
    start_tcp_child_job(csock);
  } else {
    isc_async_run(worker->loop, start_tcp_child_job, csock);
  }
}

isc_result_t
isc_nm_listentcp(uint32_t workers, isc_sockaddr_t *iface,
     isc_nm_accept_cb_t accept_cb, void *accept_cbarg, int backlog,
     isc_quota_t *quota, isc_nmsocket_t **sockp) {
  isc_nmsocket_t *sock = NULL;
  uv_os_sock_t fd = -1;
  isc_result_t result = ISC_R_UNSET;
  isc__networker_t *worker = isc__networker_get(0);

  REQUIRE(isc_tid() == 0);

  if (workers == 0) {
    workers = isc__netmgr->nloops;
  }
  REQUIRE(workers <= isc__netmgr->nloops);

  sock = isc_mempool_get(worker->nmsocket_pool);
  isc__nmsocket_init(sock, worker, isc_nm_tcplistener, iface, NULL);

  sock->nchildren = (workers == ISC_NM_LISTEN_ALL)
          ? (uint32_t)isc__netmgr->nloops
          : workers;
  sock->children = isc_mem_cget(worker->mctx, sock->nchildren,
              sizeof(sock->children[0]));

  isc__nmsocket_barrier_init(sock);

  sock->accept_cb = accept_cb;
  sock->accept_cbarg = accept_cbarg;
  sock->backlog = backlog;
  sock->pquota = quota;

  if (!isc__netmgr->load_balance_sockets) {
    fd = isc__nm_tcp_lb_socket(iface->type.sa.sa_family);
  }

  start_tcp_child(iface, sock, fd, 0);
  result = sock->children[0].result;
  INSIST(result != ISC_R_UNSET);

  for (size_t i = 1; i < sock->nchildren; i++) {
    start_tcp_child(iface, sock, fd, i);
  }

  isc_barrier_wait(&sock->listen_barrier);

  if (!isc__netmgr->load_balance_sockets) {
    isc__nm_closesocket(fd);
  }

  /*
   * If any of the child sockets have failed then isc_nm_listentcp
   * fails.
   */
  for (size_t i = 1; i < sock->nchildren; i++) {
    if (result == ISC_R_SUCCESS &&
        sock->children[i].result != ISC_R_SUCCESS)
    {
      result = sock->children[i].result;
    }
  }

  if (result != ISC_R_SUCCESS) {
    sock->active = false;
    isc__nm_tcp_stoplistening(sock);
    isc_nmsocket_close(&sock);

    return result;
  }

  sock->active = true;

  *sockp = sock;
  return ISC_R_SUCCESS;
}

static void
tcp_connection_cb(uv_stream_t *server, int status) {
  isc_nmsocket_t *ssock = uv_handle_get_data((uv_handle_t *)server);
  isc_result_t result;

  REQUIRE(ssock->accept_cb != NULL);

  if (status != 0) {
    result = isc_uverr2result(status);
    tcp_dbg_log(ssock, result,
          "TCP peer connection attempt early failure");
    goto done;
  }

  REQUIRE(VALID_NMSOCK(ssock));
  REQUIRE(ssock->tid == isc_tid());

  if (isc__nmsocket_closing(ssock)) {
    result = ISC_R_CANCELED;
    goto done;
  }

  /* Prepare the child socket */
  isc_nmsocket_t *csock = isc_mempool_get(ssock->worker->nmsocket_pool);
  isc__nmsocket_init(csock, ssock->worker, isc_nm_tcpsocket,
         &ssock->iface, NULL);
  isc__nmsocket_attach(ssock, &csock->server);

  tcp_dbg_log(csock, ISC_R_SUCCESS, "TCP peer connection attempt");

  if (csock->server->pquota != NULL) {
    result = isc_quota_acquire_cb(csock->server->pquota,
                &csock->quotacb, quota_accept_cb,
                csock);
    if (result == ISC_R_QUOTA) {
      csock->quota_accept_ts = isc_time_monotonic();
      isc__nm_incstats(ssock, STATID_ACCEPTFAIL);
      goto done;
    }
  }

  result = accept_connection(csock);
done:
  isc__nm_accept_connection_log(ssock, result, can_log_tcp_quota());
}

static void
stop_tcp_child_job(void *arg) {
  isc_nmsocket_t *sock = arg;

  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(sock->tid == isc_tid());
  REQUIRE(sock->parent != NULL);
  REQUIRE(sock->type == isc_nm_tcpsocket);
  REQUIRE(!sock->closing);

  sock->active = false;
  sock->closing = true;

  /*
   * The order of the close operation is important here, the uv_close()
   * gets scheduled in the reverse order, so we need to close the timer
   * last, so its gone by the time we destroy the socket
   */

  /* 2. close the listening socket */
  isc__nmsocket_clearcb(sock);
  isc__nm_stop_reading(sock);
  uv_close(&sock->uv_handle.handle, tcp_stop_cb);

  /* 1. close the read timer */
  isc__nmsocket_timer_stop(sock);
  uv_close(&sock->read_timer, NULL);

  REQUIRE(!sock->worker->loop->paused);
  isc_barrier_wait(&sock->parent->stop_barrier);
}

static void
stop_tcp_child(isc_nmsocket_t *sock) {
  REQUIRE(VALID_NMSOCK(sock));

  if (sock->tid == 0) {
    stop_tcp_child_job(sock);
  } else {
    isc_async_run(sock->worker->loop, stop_tcp_child_job, sock);
  }
}

void
isc__nm_tcp_stoplistening(isc_nmsocket_t *sock) {
  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(sock->type == isc_nm_tcplistener);
  REQUIRE(sock->tid == isc_tid());
  REQUIRE(sock->tid == 0);
  REQUIRE(!sock->closing);

  sock->closing = true;

  /* Mark the parent socket inactive */
  sock->active = false;

  /* Stop all the other threads' children */
  for (size_t i = 1; i < sock->nchildren; i++) {
    stop_tcp_child(&sock->children[i]);
  }

  /* Stop the child for the main thread */
  stop_tcp_child(&sock->children[0]);

  /* Stop the parent */
  sock->closed = true;

  isc__nmsocket_prep_destroy(sock);
}

static void
tcp_stop_cb(uv_handle_t *handle) {
  isc_nmsocket_t *sock = uv_handle_get_data(handle);
  uv_handle_set_data(handle, NULL);

  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(sock->tid == isc_tid());
  REQUIRE(sock->closing);
  REQUIRE(sock->type == isc_nm_tcpsocket);
  REQUIRE(!sock->closed);

  sock->closed = true;

  isc__nm_incstats(sock, STATID_CLOSE);

  isc__nmsocket_detach(&sock);
}

void
isc__nm_tcp_failed_read_cb(isc_nmsocket_t *sock, isc_result_t result,
         bool async) {
  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(result != ISC_R_SUCCESS);

  isc__nmsocket_timer_stop(sock);
  isc__nm_stop_reading(sock);
  sock->reading = false;

  if (sock->recv_cb != NULL) {
    isc__nm_uvreq_t *req = isc__nm_get_read_req(sock, NULL);
    isc__nmsocket_clearcb(sock);
    isc__nm_readcb(sock, req, result, async);
  }

  isc__nmsocket_prep_destroy(sock);
}

void
isc__nm_tcp_read(isc_nmhandle_t *handle, isc_nm_recv_cb_t cb, void *cbarg) {
  isc_nmsocket_t *sock;
  isc_result_t result;

  REQUIRE(VALID_NMHANDLE(handle));
  REQUIRE(VALID_NMSOCK(handle->sock));

  sock = handle->sock;

  REQUIRE(sock->type == isc_nm_tcpsocket);
  REQUIRE(sock->statichandle == handle);

  sock->recv_cb = cb;
  sock->recv_cbarg = cbarg;

  /* Initialize the timer */
  if (sock->read_timeout == 0) {
    sock->read_timeout =
      sock->keepalive
        ? atomic_load_relaxed(&isc__netmgr->keepalive)
        : atomic_load_relaxed(&isc__netmgr->idle);
  }

  if (isc__nmsocket_closing(sock)) {
    CLEANUP(ISC_R_CANCELED);
  }

  if (!sock->reading_throttled) {
    CHECK(isc__nm_start_reading(sock));
  }

  sock->reading = true;

  if (!sock->manual_read_timer) {
    isc__nmsocket_timer_start(sock);
  }

  return;
cleanup:
  isc__nm_tcp_failed_read_cb(sock, result, true);
}

void
isc__nm_tcp_read_stop(isc_nmhandle_t *handle) {
  REQUIRE(VALID_NMHANDLE(handle));
  REQUIRE(VALID_NMSOCK(handle->sock));

  isc_nmsocket_t *sock = handle->sock;

  if (!sock->manual_read_timer) {
    isc__nmsocket_timer_stop(sock);
  }
  isc__nm_stop_reading(sock);
  sock->reading = false;

  return;
}

void
isc__nm_tcp_read_cb(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf) {
  isc_nmsocket_t *sock = uv_handle_get_data((uv_handle_t *)stream);
  isc__nm_uvreq_t *req = NULL;

  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(sock->tid == isc_tid());
  REQUIRE(buf != NULL);

  if (isc__nmsocket_closing(sock)) {
    isc__nm_tcp_failed_read_cb(sock, ISC_R_CANCELED, false);
    goto free;
  }

  if (nread < 0) {
    if (nread != UV_EOF) {
      isc__nm_incstats(sock, STATID_RECVFAIL);
    }

    isc__nm_tcp_failed_read_cb(sock, isc_uverr2result(nread),
             false);

    goto free;
  }

  req = isc__nm_get_read_req(sock, NULL);

  /*
   * The callback will be called synchronously because the
   * result is ISC_R_SUCCESS, so we don't need to retain
   * the buffer
   */
  req->uvbuf.base = buf->base;
  req->uvbuf.len = nread;

  if (!sock->client) {
    sock->read_timeout =
      sock->keepalive
        ? atomic_load_relaxed(&isc__netmgr->keepalive)
        : atomic_load_relaxed(&isc__netmgr->idle);
  }

  isc__nm_readcb(sock, req, ISC_R_SUCCESS, false);

  if (!sock->client && sock->reading) {
    /*
     * Stop reading if we have accumulated enough bytes in the send
     * queue; this means that the TCP client is not reading back the
     * data we sending to it, and there's no reason to continue
     * processing more incoming DNS messages, if the client is not
     * reading back the responses.
     */
    size_t write_queue_size =
      uv_stream_get_write_queue_size(&sock->uv_handle.stream);

    if (write_queue_size >= ISC_NETMGR_TCP_SENDBUF_SIZE) {
      isc__nmsocket_log(
        sock, ISC_LOG_DEBUG(3),
        "throttling TCP connection, the other side is "
        "not reading the data (%zu)",
        write_queue_size);
      sock->reading_throttled = true;
      isc__nm_stop_reading(sock);
    }
  } else if (uv_is_active(&sock->uv_handle.handle) &&
       !sock->manual_read_timer)
  {
    /* The readcb could have paused the reading */
    /* The timer will be updated */
    isc__nmsocket_timer_restart(sock);
  }

free:
  if (nread < 0) {
    /*
     * The buffer may be a null buffer on error.
     */
    if (buf->base == NULL && buf->len == 0) {
      return;
    }
  }

  isc__nm_free_uvbuf(sock, buf);
}

/*
 * This is called after we get a quota_accept_cb() callback.
 */
static void
tcpaccept_cb(void *arg) {
  isc_nmsocket_t *csock = arg;
  isc_nmsocket_t *ssock = csock->server;

  REQUIRE(VALID_NMSOCK(csock));
  REQUIRE(csock->tid == isc_tid());

  isc_result_t result = accept_connection(csock);
  isc__nm_accept_connection_log(ssock, result, can_log_tcp_quota());
  isc__nmsocket_detach(&csock);
}

static void
quota_accept_cb(void *arg) {
  isc_nmsocket_t *csock = arg;
  isc_nmsocket_t *ssock = csock->server;

  REQUIRE(VALID_NMSOCK(csock));

  /*
   * This needs to be asynchronous, because the quota might have been
   * released by a different child socket.
   */
  if (csock->tid == isc_tid()) {
    isc_result_t result = accept_connection(csock);
    isc__nm_accept_connection_log(ssock, result,
                can_log_tcp_quota());
  } else {
    isc__nmsocket_attach(csock, &(isc_nmsocket_t *){ NULL });
    isc_async_run(csock->worker->loop, tcpaccept_cb, csock);
  }
}

static isc_result_t
accept_connection(isc_nmsocket_t *csock) {
  int r;
  isc_result_t result;
  struct sockaddr_storage ss;
  isc_sockaddr_t local;
  isc_nmhandle_t *handle = NULL;

  REQUIRE(VALID_NMSOCK(csock));
  REQUIRE(VALID_NMSOCK(csock->server));
  REQUIRE(csock->tid == isc_tid());

  csock->accepting = true;
  csock->accept_cb = csock->server->accept_cb;
  csock->accept_cbarg = csock->server->accept_cbarg;
  csock->recv_cb = csock->server->recv_cb;
  csock->recv_cbarg = csock->server->recv_cbarg;
  csock->read_timeout = atomic_load_relaxed(&isc__netmgr->init);

  r = uv_tcp_init(&csock->worker->loop->loop, &csock->uv_handle.tcp);
  UV_RUNTIME_CHECK(uv_tcp_init, r);
  uv_handle_set_data(&csock->uv_handle.handle, csock);

  r = uv_timer_init(&csock->worker->loop->loop, &csock->read_timer);
  UV_RUNTIME_CHECK(uv_timer_init, r);
  uv_handle_set_data((uv_handle_t *)&csock->read_timer, csock);

  if (csock->server->pquota != NULL) {
    isc__nm_incstats(csock, STATID_CLIENTS);
  }

  /*
   * We need to initialize the tcp and timer before failing because
   * isc__nm_tcp_close() can't handle uninitalized TCP nmsocket.
   */
  if (isc__nmsocket_closing(csock)) {
    CLEANUP(ISC_R_CANCELED);
  }

  r = uv_accept(&csock->server->uv_handle.stream,
          &csock->uv_handle.stream);
  if (r != 0) {
    result = isc_uverr2result(r);
    goto cleanup;
  }

  /* Check if the connection is not expired */
  if (csock->quota_accept_ts != 0) {
    /* The timestamp is given in nanoseconds */
    const uint64_t time_elapsed_ms =
      (isc_time_monotonic() - csock->quota_accept_ts) /
      NS_PER_MS;

    if (time_elapsed_ms >= csock->read_timeout) {
      /*
       * At this point we have received a connection from a
       * queue of accepted connections (via uv_accept()), but
       * it has expired. We cannot do anything better than
       * drop it on the floor at this point.
       */
      CLEANUP(ISC_R_TIMEDOUT);
    } else {
      /* Adjust the initial read timeout accordingly */
      csock->read_timeout -= time_elapsed_ms;
    }
  }

  r = uv_tcp_getpeername(&csock->uv_handle.tcp, (struct sockaddr *)&ss,
             &(int){ sizeof(ss) });
  if (r != 0) {
    result = isc_uverr2result(r);
    goto cleanup;
  }

  CHECK(isc_sockaddr_fromsockaddr(&csock->peer, (struct sockaddr *)&ss));

  r = uv_tcp_getsockname(&csock->uv_handle.tcp, (struct sockaddr *)&ss,
             &(int){ sizeof(ss) });
  if (r != 0) {
    result = isc_uverr2result(r);
    goto cleanup;
  }

  CHECK(isc_sockaddr_fromsockaddr(&local, (struct sockaddr *)&ss));

  handle = isc__nmhandle_get(csock, NULL, &local);

  result = csock->accept_cb(handle, ISC_R_SUCCESS, csock->accept_cbarg);
  if (result != ISC_R_SUCCESS) {
    isc_nmhandle_detach(&handle);
    goto cleanup;
  }

  csock->accepting = false;

  tcp_dbg_log(csock, ISC_R_SUCCESS, "TCP connection has been accepted");

  isc__nm_incstats(csock, STATID_ACCEPT);

  /*
   * The acceptcb needs to attach to the handle if it wants to keep the
   * connection alive
   */
  isc_nmhandle_detach(&handle);

  /*
   * sock is now attached to the handle.
   */
  isc__nmsocket_detach(&csock);

  return ISC_R_SUCCESS;

cleanup:
  csock->active = false;
  csock->accepting = false;

  if (result != ISC_R_NOTCONNECTED) {
    /* IGNORE: The client disconnected before we could accept */
    isc__nmsocket_log(csock, ISC_LOG_ERROR,
          "Accepting TCP connection failed: %s",
          isc_result_totext(result));
  }

  tcp_dbg_log(csock, result, "TCP connection has NOT been accepted");

  isc__nmsocket_prep_destroy(csock);

  isc__nmsocket_detach(&csock);

  return result;
}

static void
tcp_send(isc_nmhandle_t *handle, const isc_region_t *region, isc_nm_cb_t cb,
   void *cbarg, const bool dnsmsg) {
  REQUIRE(VALID_NMHANDLE(handle));
  REQUIRE(VALID_NMSOCK(handle->sock));

  isc_nmsocket_t *sock = handle->sock;
  isc_result_t result;
  isc__nm_uvreq_t *uvreq = NULL;

  REQUIRE(sock->type == isc_nm_tcpsocket);
  REQUIRE(sock->tid == isc_tid());

  uvreq = isc__nm_uvreq_get(sock);
  if (dnsmsg) {
    *(uint16_t *)uvreq->tcplen = htons(region->length);
  }
  uvreq->uvbuf.base = (char *)region->base;
  uvreq->uvbuf.len = region->length;

  isc_nmhandle_attach(handle, &uvreq->handle);

  uvreq->cb.send = cb;
  uvreq->cbarg = cbarg;

  if (sock->write_timeout == 0) {
    sock->write_timeout =
      sock->keepalive
        ? atomic_load_relaxed(&isc__netmgr->keepalive)
        : atomic_load_relaxed(&isc__netmgr->idle);
  }

  result = tcp_send_direct(sock, uvreq);
  if (result != ISC_R_SUCCESS) {
    isc__nm_incstats(sock, STATID_SENDFAIL);
    isc__nm_failed_send_cb(sock, uvreq, result, true);
  }

  return;
}

void
isc__nm_tcp_send(isc_nmhandle_t *handle, const isc_region_t *region,
     isc_nm_cb_t cb, void *cbarg) {
  tcp_send(handle, region, cb, cbarg, false);
}

void
isc__nm_tcp_senddns(isc_nmhandle_t *handle, const isc_region_t *region,
        isc_nm_cb_t cb, void *cbarg) {
  tcp_send(handle, region, cb, cbarg, true);
}

static void
tcp_maybe_restart_reading(isc_nmsocket_t *sock) {
  if (!sock->client && sock->reading &&
      !uv_is_active(&sock->uv_handle.handle))
  {
    /*
     * Restart reading if we have less data in the send queue than
     * the send buffer size, this means that the TCP client has
     * started reading some data again.  Starting reading when we go
     * under the limit instead of waiting for all data has been
     * flushed allows faster recovery (in case there was a
     * congestion and now there isn't).
     */
    size_t write_queue_size =
      uv_stream_get_write_queue_size(&sock->uv_handle.stream);
    if (write_queue_size < ISC_NETMGR_TCP_SENDBUF_SIZE) {
      isc__nmsocket_log(
        sock, ISC_LOG_DEBUG(3),
        "resuming TCP connection, the other side  "
        "is reading the data again (%zu)",
        write_queue_size);
      isc__nm_start_reading(sock);
      sock->reading_throttled = false;
    }
  }
}

static void
tcp_send_cb(uv_write_t *req, int status) {
  isc__nm_uvreq_t *uvreq = (isc__nm_uvreq_t *)req->data;
  isc_nmsocket_t *sock = NULL;

  REQUIRE(VALID_UVREQ(uvreq));
  REQUIRE(VALID_NMSOCK(uvreq->sock));

  sock = uvreq->sock;

  isc_nm_timer_stop(uvreq->timer);
  isc_nm_timer_detach(&uvreq->timer);

  if (status < 0) {
    isc__nm_incstats(sock, STATID_SENDFAIL);
    isc__nm_failed_send_cb(sock, uvreq, isc_uverr2result(status),
               false);
    if (!sock->client && sock->reading) {
      /*
       * As we are resuming reading, it is not throttled
       * anymore (technically).
       */
      sock->reading_throttled = false;
      isc__nm_start_reading(sock);
      isc__nmsocket_reset(sock);
    }
    return;
  }

  isc__nm_sendcb(sock, uvreq, ISC_R_SUCCESS, false);
  tcp_maybe_restart_reading(sock);
}

static isc_result_t
tcp_send_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(VALID_UVREQ(req));
  REQUIRE(sock->tid == isc_tid());
  REQUIRE(sock->type == isc_nm_tcpsocket);

  int r;
  uv_buf_t bufs[2] = { { 0 }, { 0 } }; /* ugly, but required for old GCC
            versions */
  size_t nbufs = 1;

  if (isc__nmsocket_closing(sock)) {
    return ISC_R_CANCELED;
  }

  /* Check if we are not trying to send a DNS message */
  if (*(uint16_t *)req->tcplen == 0) {
    bufs[0].base = req->uvbuf.base;
    bufs[0].len = req->uvbuf.len;

    r = uv_try_write(&sock->uv_handle.stream, bufs, nbufs);

    if (r == (int)(bufs[0].len)) {
      /* Wrote everything */
      isc__nm_sendcb(sock, req, ISC_R_SUCCESS, true);
      tcp_maybe_restart_reading(sock);
      return ISC_R_SUCCESS;
    } else if (r > 0) {
      bufs[0].base += (size_t)r;
      bufs[0].len -= (size_t)r;
    } else if (!(r == UV_ENOSYS || r == UV_EAGAIN)) {
      return isc_uverr2result(r);
    }
  } else {
    nbufs = 2;
    bufs[0].base = req->tcplen;
    bufs[0].len = 2;
    bufs[1].base = req->uvbuf.base;
    bufs[1].len = req->uvbuf.len;

    r = uv_try_write(&sock->uv_handle.stream, bufs, nbufs);

    if (r == (int)(bufs[0].len + bufs[1].len)) {
      /* Wrote everything */
      isc__nm_sendcb(sock, req, ISC_R_SUCCESS, true);
      tcp_maybe_restart_reading(sock);
      return ISC_R_SUCCESS;
    } else if (r == 1) {
      /* Partial write of DNSMSG length */
      bufs[0].base = req->tcplen + 1;
      bufs[0].len = 1;
    } else if (r > 0) {
      /* Partial write of DNSMSG */
      nbufs = 1;
      bufs[0].base = req->uvbuf.base + (r - 2);
      bufs[0].len = req->uvbuf.len - (r - 2);
    } else if (!(r == UV_ENOSYS || r == UV_EAGAIN)) {
      return isc_uverr2result(r);
    }
  }

  if (!sock->client && sock->reading) {
    sock->reading_throttled = true;
    isc__nm_stop_reading(sock);
  }

  isc__nmsocket_log(sock, ISC_LOG_DEBUG(3),
        "%sthe other side is not "
        "reading the data, switching to uv_write()",
        !sock->client && sock->reading
          ? "throttling TCP connection, "
          : "");

  r = uv_write(&req->uv_req.write, &sock->uv_handle.stream, bufs, nbufs,
         tcp_send_cb);
  if (r < 0) {
    return isc_uverr2result(r);
  }

  isc_nm_timer_create(req->handle, isc__nmsocket_writetimeout_cb, req,
          &req->timer);
  if (sock->write_timeout > 0) {
    isc_nm_timer_start(req->timer, sock->write_timeout);
  }

  return ISC_R_SUCCESS;
}

static void
tcp_close_sock(isc_nmsocket_t *sock) {
  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(sock->tid == isc_tid());
  REQUIRE(sock->closing);
  REQUIRE(!sock->closed);

  sock->closed = true;
  sock->connected = false;

  isc__nm_incstats(sock, STATID_CLOSE);

  if (sock->server != NULL) {
    if (sock->server->pquota != NULL) {
      isc__nm_decstats(sock, STATID_CLIENTS);
      isc_quota_release(sock->server->pquota);
    }
    isc__nmsocket_detach(&sock->server);
  }

  tcp_dbg_log(sock, ISC_R_SUCCESS, "TCP connection closed");

  isc__nmsocket_prep_destroy(sock);
}

static void
tcp_close_cb(uv_handle_t *handle) {
  isc_nmsocket_t *sock = uv_handle_get_data(handle);
  uv_handle_set_data(handle, NULL);

  tcp_close_sock(sock);
}

void
isc__nm_tcp_close(isc_nmsocket_t *sock) {
  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(sock->type == isc_nm_tcpsocket);
  REQUIRE(!isc__nmsocket_active(sock));
  REQUIRE(sock->tid == isc_tid());
  REQUIRE(sock->parent == NULL);
  REQUIRE(!sock->closing);

  sock->closing = true;

  /*
   * The order of the close operation is important here, the uv_close()
   * gets scheduled in the reverse order, so we need to close the timer
   * last, so its gone by the time we destroy the socket
   */

  if (!uv_is_closing(&sock->uv_handle.handle)) {
    /* Normal order of operation */

    /* 2. close the socket + destroy the socket in callback */
    isc__nmsocket_clearcb(sock);
    isc__nm_stop_reading(sock);
    sock->reading = false;
    uv_close(&sock->uv_handle.handle, tcp_close_cb);

    /* 1. close the timer */
    isc__nmsocket_timer_stop(sock);
    uv_close((uv_handle_t *)&sock->read_timer, NULL);
  } else {
    /* The socket was already closed elsewhere */

    /* 1. close the timer + destroy the socket in callback */
    isc__nmsocket_timer_stop(sock);
    uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
    uv_close((uv_handle_t *)&sock->read_timer, tcp_close_cb);
  }
}

static void
tcp_close_connect_cb(uv_handle_t *handle) {
  isc_nmsocket_t *sock = uv_handle_get_data(handle);

  REQUIRE(VALID_NMSOCK(sock));

  REQUIRE(sock->tid == isc_tid());

  isc__nmsocket_prep_destroy(sock);
  isc__nmsocket_detach(&sock);
}

void
isc__nm_tcp_shutdown(isc_nmsocket_t *sock) {
  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(sock->tid == isc_tid());
  REQUIRE(sock->type == isc_nm_tcpsocket);

  /*
   * If the socket is active, mark it inactive and
   * continue. If it isn't active, stop now.
   */
  if (!sock->active) {
    return;
  }
  sock->active = false;

  INSIST(!sock->accepting);

  if (sock->connecting) {
    isc_nmsocket_t *tsock = NULL;
    isc__nmsocket_attach(sock, &tsock);
    uv_close(&sock->uv_handle.handle, tcp_close_connect_cb);
    return;
  }

  /* There's a handle attached to the socket (from accept or connect) */
  if (sock->statichandle) {
    isc__nm_failed_read_cb(sock, ISC_R_SHUTTINGDOWN, false);
    return;
  }

  /* Destroy the non-listening socket */
  if (sock->parent == NULL) {
    isc__nmsocket_prep_destroy(sock);
    return;
  }

  /* Destroy the listening socket if on the same loop */
  if (sock->tid == sock->parent->tid) {
    isc__nmsocket_prep_destroy(sock->parent);
  }
}

void
isc__nmhandle_tcp_set_manual_timer(isc_nmhandle_t *handle, const bool manual) {
  isc_nmsocket_t *sock;

  REQUIRE(VALID_NMHANDLE(handle));
  sock = handle->sock;
  REQUIRE(VALID_NMSOCK(sock));
  REQUIRE(sock->type == isc_nm_tcpsocket);
  REQUIRE(sock->tid == isc_tid());
  REQUIRE(!uv_is_active(&sock->uv_handle.handle));

  sock->manual_read_timer = manual;
}

Coverage Report

Created: 2026-01-24 06:24