/*
* 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 */

You are viewing proxied material from ftp.icm.edu.pl. The copyright of proxied material belongs to its original authors. Any comments or complaints in relation to proxied material should be directed to the original authors of the content concerned. Please see the disclaimer for more details.