/src/unbound/util/tube.c

Source (jump to first uncovered line)
/*
 * util/tube.c - pipe service
 *
 * Copyright (c) 2008, NLnet Labs. All rights reserved.
 *
 * This software is open source.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 
 * Redistributions of source code must retain the above copyright notice,
 * this list of conditions and the following disclaimer.
 * 
 * Redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the following disclaimer in the documentation
 * and/or other materials provided with the distribution.
 * 
 * Neither the name of the NLNET LABS nor the names of its contributors may
 * be used to endorse or promote products derived from this software without
 * specific prior written permission.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
 * TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

/**
 * \file
 *
 * This file contains pipe service functions.
 */
#include "config.h"
#include "util/tube.h"
#include "util/log.h"
#include "util/net_help.h"
#include "util/netevent.h"
#include "util/fptr_wlist.h"
#include "util/ub_event.h"
#ifdef HAVE_POLL_H
#include <poll.h>
#endif

#ifndef USE_WINSOCK
/* on unix */

#ifndef HAVE_SOCKETPAIR
/** no socketpair() available, like on Minix 3.1.7, use pipe */
#define socketpair(f, t, p, sv) pipe(sv) 
#endif /* HAVE_SOCKETPAIR */

struct tube* tube_create(void)
{
  struct tube* tube = (struct tube*)calloc(1, sizeof(*tube));
  int sv[2];
  if(!tube) {
    int err = errno;
    log_err("tube_create: out of memory");
    errno = err;
    return NULL;
  }
  tube->sr = -1;
  tube->sw = -1;
  if(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == -1) {
    int err = errno;
    log_err("socketpair: %s", strerror(errno));
    free(tube);
    errno = err;
    return NULL;
  }
  tube->sr = sv[0];
  tube->sw = sv[1];
  if(!fd_set_nonblock(tube->sr) || !fd_set_nonblock(tube->sw)) {
    int err = errno;
    log_err("tube: cannot set nonblocking");
    tube_delete(tube);
    errno = err;
    return NULL;
  }
  return tube;
}

void tube_delete(struct tube* tube)
{
  if(!tube) return;
  tube_remove_bg_listen(tube);
  tube_remove_bg_write(tube);
  /* close fds after deleting commpoints, to be sure.
   *            Also epoll does not like closing fd before event_del */
  tube_close_read(tube);
  tube_close_write(tube);
  free(tube);
}

void tube_close_read(struct tube* tube)
{
  if(tube->sr != -1) {
    close(tube->sr);
    tube->sr = -1;
  }
}

void tube_close_write(struct tube* tube)
{
  if(tube->sw != -1) {
    close(tube->sw);
    tube->sw = -1;
  }
}

void tube_remove_bg_listen(struct tube* tube)
{
  if(tube->listen_com) {
    comm_point_delete(tube->listen_com);
    tube->listen_com = NULL;
  }
  free(tube->cmd_msg);
  tube->cmd_msg = NULL;
}

void tube_remove_bg_write(struct tube* tube)
{
  if(tube->res_com) {
    comm_point_delete(tube->res_com);
    tube->res_com = NULL;
  }
  if(tube->res_list) {
    struct tube_res_list* np, *p = tube->res_list;
    tube->res_list = NULL;
    tube->res_last = NULL;
    while(p) {
      np = p->next;
      free(p->buf);
      free(p);
      p = np;
    }
  }
}

int
tube_handle_listen(struct comm_point* c, void* arg, int error,
        struct comm_reply* ATTR_UNUSED(reply_info))
{
  struct tube* tube = (struct tube*)arg;
  ssize_t r;
  if(error != NETEVENT_NOERROR) {
    fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
    (*tube->listen_cb)(tube, NULL, 0, error, tube->listen_arg);
    return 0;
  }

  if(tube->cmd_read < sizeof(tube->cmd_len)) {
    /* complete reading the length of control msg */
    r = read(c->fd, ((uint8_t*)&tube->cmd_len) + tube->cmd_read,
      sizeof(tube->cmd_len) - tube->cmd_read);
    if(r==0) {
      /* error has happened or */
      /* parent closed pipe, must have exited somehow */
      fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
      (*tube->listen_cb)(tube, NULL, 0, NETEVENT_CLOSED, 
        tube->listen_arg);
      return 0;
    }
    if(r==-1) {
      if(errno != EAGAIN && errno != EINTR) {
        log_err("rpipe error: %s", strerror(errno));
      }
      /* nothing to read now, try later */
      return 0;
    }
    tube->cmd_read += r;
    if(tube->cmd_read < sizeof(tube->cmd_len)) {
      /* not complete, try later */
      return 0;
    }
    tube->cmd_msg = (uint8_t*)calloc(1, tube->cmd_len);
    if(!tube->cmd_msg) {
      log_err("malloc failure");
      tube->cmd_read = 0;
      return 0;
    }
  }
  /* cmd_len has been read, read remainder */
  r = read(c->fd, tube->cmd_msg+tube->cmd_read-sizeof(tube->cmd_len),
    tube->cmd_len - (tube->cmd_read - sizeof(tube->cmd_len)));
  if(r==0) {
    /* error has happened or */
    /* parent closed pipe, must have exited somehow */
    fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
    (*tube->listen_cb)(tube, NULL, 0, NETEVENT_CLOSED, 
      tube->listen_arg);
    return 0;
  }
  if(r==-1) {
    /* nothing to read now, try later */
    if(errno != EAGAIN && errno != EINTR) {
      log_err("rpipe error: %s", strerror(errno));
    }
    return 0;
  }
  tube->cmd_read += r;
  if(tube->cmd_read < sizeof(tube->cmd_len) + tube->cmd_len) {
    /* not complete, try later */
    return 0;
  }
  tube->cmd_read = 0;

  fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
  (*tube->listen_cb)(tube, tube->cmd_msg, tube->cmd_len, 
    NETEVENT_NOERROR, tube->listen_arg);
    /* also frees the buf */
  tube->cmd_msg = NULL;
  return 0;
}

int
tube_handle_write(struct comm_point* c, void* arg, int error,
        struct comm_reply* ATTR_UNUSED(reply_info))
{
  struct tube* tube = (struct tube*)arg;
  struct tube_res_list* item = tube->res_list;
  ssize_t r;
  if(error != NETEVENT_NOERROR) {
    log_err("tube_handle_write net error %d", error);
    return 0;
  }

  if(!item) {
    comm_point_stop_listening(c);
    return 0;
  }

  if(tube->res_write < sizeof(item->len)) {
    r = write(c->fd, ((uint8_t*)&item->len) + tube->res_write,
      sizeof(item->len) - tube->res_write);
    if(r == -1) {
      if(errno != EAGAIN && errno != EINTR) {
        log_err("wpipe error: %s", strerror(errno));
      }
      return 0; /* try again later */
    }
    if(r == 0) {
      /* error on pipe, must have exited somehow */
      /* cannot signal this to pipe user */
      return 0;
    }
    tube->res_write += r;
    if(tube->res_write < sizeof(item->len))
      return 0;
  }
  r = write(c->fd, item->buf + tube->res_write - sizeof(item->len),
    item->len - (tube->res_write - sizeof(item->len)));
  if(r == -1) {
    if(errno != EAGAIN && errno != EINTR) {
      log_err("wpipe error: %s", strerror(errno));
    }
    return 0; /* try again later */
  }
  if(r == 0) {
    /* error on pipe, must have exited somehow */
    /* cannot signal this to pipe user */
    return 0;
  }
  tube->res_write += r;
  if(tube->res_write < sizeof(item->len) + item->len)
    return 0;
  /* done this result, remove it */
  free(item->buf);
  item->buf = NULL;
  tube->res_list = tube->res_list->next;
  free(item);
  if(!tube->res_list) {
    tube->res_last = NULL;
    comm_point_stop_listening(c);
  }
  tube->res_write = 0;
  return 0;
}

int tube_write_msg(struct tube* tube, uint8_t* buf, uint32_t len, 
        int nonblock)
{
  ssize_t r, d;
  int fd = tube->sw;

  /* test */
  if(nonblock) {
    r = write(fd, &len, sizeof(len));
    if(r == -1) {
      if(errno==EINTR || errno==EAGAIN)
        return -1;
      log_err("tube msg write failed: %s", strerror(errno));
      return -1; /* can still continue, perhaps */
    }
  } else r = 0;
  if(!fd_set_block(fd))
    return 0;
  /* write remainder */
  d = r;
  while(d != (ssize_t)sizeof(len)) {
    if((r=write(fd, ((char*)&len)+d, sizeof(len)-d)) == -1) {
      if(errno == EAGAIN)
        continue; /* temporarily unavail: try again*/
      log_err("tube msg write failed: %s", strerror(errno));
      (void)fd_set_nonblock(fd);
      return 0;
    }
    d += r;
  }
  d = 0;
  while(d != (ssize_t)len) {
    if((r=write(fd, buf+d, len-d)) == -1) {
      if(errno == EAGAIN)
        continue; /* temporarily unavail: try again*/
      log_err("tube msg write failed: %s", strerror(errno));
      (void)fd_set_nonblock(fd);
      return 0;
    }
    d += r;
  }
  if(!fd_set_nonblock(fd))
    return 0;
  return 1;
}

int tube_read_msg(struct tube* tube, uint8_t** buf, uint32_t* len, 
        int nonblock)
{
  ssize_t r, d;
  int fd = tube->sr;

  /* test */
  *len = 0;
  if(nonblock) {
    r = read(fd, len, sizeof(*len));
    if(r == -1) {
      if(errno==EINTR || errno==EAGAIN)
        return -1;
      log_err("tube msg read failed: %s", strerror(errno));
      return -1; /* we can still continue, perhaps */
    }
    if(r == 0) /* EOF */
      return 0;
  } else r = 0;
  if(!fd_set_block(fd))
    return 0;
  /* read remainder */
  d = r;
  while(d != (ssize_t)sizeof(*len)) {
    if((r=read(fd, ((char*)len)+d, sizeof(*len)-d)) == -1) {
      log_err("tube msg read failed: %s", strerror(errno));
      (void)fd_set_nonblock(fd);
      return 0;
    }
    if(r == 0) /* EOF */ {
      (void)fd_set_nonblock(fd);
      return 0;
    }
    d += r;
  }
  if (*len >= 65536*2) {
    log_err("tube msg length %u is too big", (unsigned)*len);
    (void)fd_set_nonblock(fd);
    return 0;
  }
  *buf = (uint8_t*)malloc(*len);
  if(!*buf) {
    log_err("tube read out of memory");
    (void)fd_set_nonblock(fd);
    return 0;
  }
  d = 0;
  while(d < (ssize_t)*len) {
    if((r=read(fd, (*buf)+d, (size_t)((ssize_t)*len)-d)) == -1) {
      log_err("tube msg read failed: %s", strerror(errno));
      (void)fd_set_nonblock(fd);
      free(*buf);
      return 0;
    }
    if(r == 0) { /* EOF */
      (void)fd_set_nonblock(fd);
      free(*buf);
      return 0;
    }
    d += r;
  }
  if(!fd_set_nonblock(fd)) {
    free(*buf);
    return 0;
  }
  return 1;
}

/** perform poll() on the fd */
static int
pollit(int fd, struct timeval* t)
{
  struct pollfd fds;
  int pret;
  int msec = -1;
  memset(&fds, 0, sizeof(fds));
  fds.fd = fd;
  fds.events = POLLIN | POLLERR | POLLHUP;
#ifndef S_SPLINT_S
  if(t)
    msec = t->tv_sec*1000 + t->tv_usec/1000;
#endif

  pret = poll(&fds, 1, msec);

  if(pret == -1)
    return 0;
  if(pret != 0)
    return 1;
  return 0;
}

int tube_poll(struct tube* tube)
{
  struct timeval t;
  memset(&t, 0, sizeof(t));
  return pollit(tube->sr, &t);
}

int tube_wait(struct tube* tube)
{
  return pollit(tube->sr, NULL);
}

int tube_wait_timeout(struct tube* tube, int msec)
{
  int ret = 0;

  while(1) {
    struct pollfd fds;
    memset(&fds, 0, sizeof(fds));

    fds.fd = tube->sr;
    fds.events = POLLIN | POLLERR | POLLHUP;
    ret = poll(&fds, 1, msec);

    if(ret == -1) {
      if(errno == EAGAIN || errno == EINTR)
        continue;
      return -1;
    }
    break;
  }

  if(ret != 0)
    return 1;
  return 0;
}

int tube_read_fd(struct tube* tube)
{
  return tube->sr;
}

int tube_setup_bg_listen(struct tube* tube, struct comm_base* base,
        tube_callback_type* cb, void* arg)
{
  tube->listen_cb = cb;
  tube->listen_arg = arg;
  if(!(tube->listen_com = comm_point_create_raw(base, tube->sr, 
    0, tube_handle_listen, tube))) {
    int err = errno;
    log_err("tube_setup_bg_l: commpoint creation failed");
    errno = err;
    return 0;
  }
  return 1;
}

int tube_setup_bg_write(struct tube* tube, struct comm_base* base)
{
  if(!(tube->res_com = comm_point_create_raw(base, tube->sw, 
    1, tube_handle_write, tube))) {
    int err = errno;
    log_err("tube_setup_bg_w: commpoint creation failed");
    errno = err;
    return 0;
  }
  return 1;
}

int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len)
{
  struct tube_res_list* item;
  if(!tube || !tube->res_com) return 0;
  item = (struct tube_res_list*)malloc(sizeof(*item));
  if(!item) {
    free(msg);
    log_err("out of memory for async answer");
    return 0;
  }
  item->buf = msg;
  item->len = len;
  item->next = NULL;
  /* add at back of list, since the first one may be partially written */
  if(tube->res_last)
    tube->res_last->next = item;
  else    tube->res_list = item;
  tube->res_last = item;
  if(tube->res_list == tube->res_last) {
    /* first added item, start the write process */
    comm_point_start_listening(tube->res_com, -1, -1);
  }
  return 1;
}

void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events), 
  void* ATTR_UNUSED(arg))
{
  log_assert(0);
}

#else /* USE_WINSOCK */
/* on windows */


struct tube* tube_create(void)
{
  /* windows does not have forks like unix, so we only support
   * threads on windows. And thus the pipe need only connect
   * threads. We use a mutex and a list of datagrams. */
  struct tube* tube = (struct tube*)calloc(1, sizeof(*tube));
  if(!tube) {
    int err = errno;
    log_err("tube_create: out of memory");
    errno = err;
    return NULL;
  }
  tube->event = WSACreateEvent();
  if(tube->event == WSA_INVALID_EVENT) {
    free(tube);
    log_err("WSACreateEvent: %s", wsa_strerror(WSAGetLastError()));
    return NULL;
  }
  if(!WSAResetEvent(tube->event)) {
    log_err("WSAResetEvent: %s", wsa_strerror(WSAGetLastError()));
  }
  lock_basic_init(&tube->res_lock);
  verbose(VERB_ALGO, "tube created");
  return tube;
}

void tube_delete(struct tube* tube)
{
  if(!tube) return;
  tube_remove_bg_listen(tube);
  tube_remove_bg_write(tube);
  tube_close_read(tube);
  tube_close_write(tube);
  if(!WSACloseEvent(tube->event))
    log_err("WSACloseEvent: %s", wsa_strerror(WSAGetLastError()));
  lock_basic_destroy(&tube->res_lock);
  verbose(VERB_ALGO, "tube deleted");
  free(tube);
}

void tube_close_read(struct tube* ATTR_UNUSED(tube))
{
  verbose(VERB_ALGO, "tube close_read");
}

void tube_close_write(struct tube* ATTR_UNUSED(tube))
{
  verbose(VERB_ALGO, "tube close_write");
  /* wake up waiting reader with an empty queue */
  if(!WSASetEvent(tube->event)) {
    log_err("WSASetEvent: %s", wsa_strerror(WSAGetLastError()));
  }
}

void tube_remove_bg_listen(struct tube* tube)
{
  verbose(VERB_ALGO, "tube remove_bg_listen");
  if (tube->ev_listen != NULL) {
    ub_winsock_unregister_wsaevent(tube->ev_listen);
    tube->ev_listen = NULL;
  }
}

void tube_remove_bg_write(struct tube* tube)
{
  verbose(VERB_ALGO, "tube remove_bg_write");
  if(tube->res_list) {
    struct tube_res_list* np, *p = tube->res_list;
    tube->res_list = NULL;
    tube->res_last = NULL;
    while(p) {
      np = p->next;
      free(p->buf);
      free(p);
      p = np;
    }
  }
}

int tube_write_msg(struct tube* tube, uint8_t* buf, uint32_t len, 
        int ATTR_UNUSED(nonblock))
{
  uint8_t* a;
  verbose(VERB_ALGO, "tube write_msg len %d", (int)len);
  a = (uint8_t*)memdup(buf, len);
  if(!a) {
    log_err("out of memory in tube_write_msg");
    return 0;
  }
  /* always nonblocking, this pipe cannot get full */
  return tube_queue_item(tube, a, len);
}

int tube_read_msg(struct tube* tube, uint8_t** buf, uint32_t* len, 
        int nonblock)
{
  struct tube_res_list* item = NULL;
  verbose(VERB_ALGO, "tube read_msg %s", nonblock?"nonblock":"blocking");
  *buf = NULL;
  if(!tube_poll(tube)) {
    verbose(VERB_ALGO, "tube read_msg nodata");
    /* nothing ready right now, wait if we want to */
    if(nonblock)
      return -1; /* would block waiting for items */
    if(!tube_wait(tube))
      return 0;
  }
  lock_basic_lock(&tube->res_lock);
  if(tube->res_list) {
    item = tube->res_list;
    tube->res_list = item->next;
    if(tube->res_last == item) {
      /* the list is now empty */
      tube->res_last = NULL;
      verbose(VERB_ALGO, "tube read_msg lastdata");
      if(!WSAResetEvent(tube->event)) {
        log_err("WSAResetEvent: %s", 
          wsa_strerror(WSAGetLastError()));
      }
    }
  }
  lock_basic_unlock(&tube->res_lock);
  if(!item)
    return 0; /* would block waiting for items */
  *buf = item->buf;
  *len = item->len;
  free(item);
  verbose(VERB_ALGO, "tube read_msg len %d", (int)*len);
  return 1;
}

int tube_poll(struct tube* tube)
{
  struct tube_res_list* item = NULL;
  lock_basic_lock(&tube->res_lock);
  item = tube->res_list;
  lock_basic_unlock(&tube->res_lock);
  if(item)
    return 1;
  return 0;
}

int tube_wait(struct tube* tube)
{
  /* block on eventhandle */
  DWORD res = WSAWaitForMultipleEvents(
    1 /* one event in array */, 
    &tube->event /* the event to wait for, our pipe signal */, 
    0 /* wait for all events is false */, 
    WSA_INFINITE /* wait, no timeout */,
    0 /* we are not alertable for IO completion routines */
    );
  if(res == WSA_WAIT_TIMEOUT) {
    return 0;
  }
  if(res == WSA_WAIT_IO_COMPLETION) {
    /* a bit unexpected, since we were not alertable */
    return 0;
  }
  return 1;
}

int tube_wait_timeout(struct tube* tube, int msec)
{
  /* block on eventhandle */
  DWORD res = WSAWaitForMultipleEvents(
    1 /* one event in array */,
    &tube->event /* the event to wait for, our pipe signal */,
    0 /* wait for all events is false */,
    msec /* wait for timeout */,
    0 /* we are not alertable for IO completion routines */
    );
  if(res == WSA_WAIT_TIMEOUT) {
    return 0;
  }
  if(res == WSA_WAIT_IO_COMPLETION) {
    /* a bit unexpected, since we were not alertable */
    return -1;
  }
  return 1;
}

int tube_read_fd(struct tube* ATTR_UNUSED(tube))
{
  /* nothing sensible on Windows */
  return -1;
}

int
tube_handle_listen(struct comm_point* ATTR_UNUSED(c), void* ATTR_UNUSED(arg), 
  int ATTR_UNUSED(error), struct comm_reply* ATTR_UNUSED(reply_info))
{
  log_assert(0);
  return 0;
}

int
tube_handle_write(struct comm_point* ATTR_UNUSED(c), void* ATTR_UNUSED(arg), 
  int ATTR_UNUSED(error), struct comm_reply* ATTR_UNUSED(reply_info))
{
  log_assert(0);
  return 0;
}

int tube_setup_bg_listen(struct tube* tube, struct comm_base* base,
        tube_callback_type* cb, void* arg)
{
  tube->listen_cb = cb;
  tube->listen_arg = arg;
  if(!comm_base_internal(base))
    return 1; /* ignore when no comm base - testing */
  tube->ev_listen = ub_winsock_register_wsaevent(
      comm_base_internal(base), tube->event, &tube_handle_signal, tube);
  return tube->ev_listen ? 1 : 0;
}

int tube_setup_bg_write(struct tube* ATTR_UNUSED(tube), 
  struct comm_base* ATTR_UNUSED(base))
{
  /* the queue item routine performs the signaling */
  return 1;
}

int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len)
{
  struct tube_res_list* item;
  if(!tube) return 0;
  item = (struct tube_res_list*)malloc(sizeof(*item));
  verbose(VERB_ALGO, "tube queue_item len %d", (int)len);
  if(!item) {
    free(msg);
    log_err("out of memory for async answer");
    return 0;
  }
  item->buf = msg;
  item->len = len;
  item->next = NULL;
  lock_basic_lock(&tube->res_lock);
  /* add at back of list, since the first one may be partially written */
  if(tube->res_last)
    tube->res_last->next = item;
  else    tube->res_list = item;
  tube->res_last = item;
  /* signal the eventhandle */
  if(!WSASetEvent(tube->event)) {
    log_err("WSASetEvent: %s", wsa_strerror(WSAGetLastError()));
  }
  lock_basic_unlock(&tube->res_lock);
  return 1;
}

void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events), 
  void* arg)
{
  struct tube* tube = (struct tube*)arg;
  uint8_t* buf;
  uint32_t len = 0;
  verbose(VERB_ALGO, "tube handle_signal");
  while(tube_poll(tube)) {
    if(tube_read_msg(tube, &buf, &len, 1)) {
      fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
      (*tube->listen_cb)(tube, buf, len, NETEVENT_NOERROR, 
        tube->listen_arg);
    }
  }
}

#endif /* USE_WINSOCK */

Coverage Report

Created: 2025-08-28 06:16