2 * util/tube.c - pipe service
4 * Copyright (c) 2008, NLnet Labs. All rights reserved.
6 * This software is open source.
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions
12 * Redistributions of source code must retain the above copyright notice,
13 * this list of conditions and the following disclaimer.
15 * Redistributions in binary form must reproduce the above copyright notice,
16 * this list of conditions and the following disclaimer in the documentation
17 * and/or other materials provided with the distribution.
19 * Neither the name of the NLNET LABS nor the names of its contributors may
20 * be used to endorse or promote products derived from this software without
21 * specific prior written permission.
23 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
24 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
25 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
26 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
27 * HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
28 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
29 * TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
30 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
31 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
32 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
33 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
39 * This file contains pipe service functions.
42 #include "util/tube.h"
44 #include "util/net_help.h"
45 #include "util/netevent.h"
46 #include "util/fptr_wlist.h"
51 #ifndef HAVE_SOCKETPAIR
52 /** no socketpair() available, like on Minix 3.1.7, use pipe */
53 #define socketpair(f, t, p, sv) pipe(sv)
54 #endif /* HAVE_SOCKETPAIR */
56 struct tube* tube_create(void)
58 struct tube* tube = (struct tube*)calloc(1, sizeof(*tube));
62 log_err("tube_create: out of memory");
68 if(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == -1) {
70 log_err("socketpair: %s", strerror(errno));
77 if(!fd_set_nonblock(tube->sr) || !fd_set_nonblock(tube->sw)) {
79 log_err("tube: cannot set nonblocking");
87 void tube_delete(struct tube* tube)
90 tube_remove_bg_listen(tube);
91 tube_remove_bg_write(tube);
92 /* close fds after deleting commpoints, to be sure.
93 * Also epoll does not like closing fd before event_del */
94 tube_close_read(tube);
95 tube_close_write(tube);
99 void tube_close_read(struct tube* tube)
107 void tube_close_write(struct tube* tube)
115 void tube_remove_bg_listen(struct tube* tube)
117 if(tube->listen_com) {
118 comm_point_delete(tube->listen_com);
119 tube->listen_com = NULL;
123 tube->cmd_msg = NULL;
127 void tube_remove_bg_write(struct tube* tube)
130 comm_point_delete(tube->res_com);
131 tube->res_com = NULL;
134 struct tube_res_list* np, *p = tube->res_list;
135 tube->res_list = NULL;
136 tube->res_last = NULL;
147 tube_handle_listen(struct comm_point* c, void* arg, int error,
148 struct comm_reply* ATTR_UNUSED(reply_info))
150 struct tube* tube = (struct tube*)arg;
152 if(error != NETEVENT_NOERROR) {
153 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
154 (*tube->listen_cb)(tube, NULL, 0, error, tube->listen_arg);
158 if(tube->cmd_read < sizeof(tube->cmd_len)) {
159 /* complete reading the length of control msg */
160 r = read(c->fd, ((uint8_t*)&tube->cmd_len) + tube->cmd_read,
161 sizeof(tube->cmd_len) - tube->cmd_read);
163 /* error has happened or */
164 /* parent closed pipe, must have exited somehow */
165 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
166 (*tube->listen_cb)(tube, NULL, 0, NETEVENT_CLOSED,
171 if(errno != EAGAIN && errno != EINTR) {
172 log_err("rpipe error: %s", strerror(errno));
174 /* nothing to read now, try later */
178 if(tube->cmd_read < sizeof(tube->cmd_len)) {
179 /* not complete, try later */
182 tube->cmd_msg = (uint8_t*)calloc(1, tube->cmd_len);
184 log_err("malloc failure");
189 /* cmd_len has been read, read remainder */
190 r = read(c->fd, tube->cmd_msg+tube->cmd_read-sizeof(tube->cmd_len),
191 tube->cmd_len - (tube->cmd_read - sizeof(tube->cmd_len)));
193 /* error has happened or */
194 /* parent closed pipe, must have exited somehow */
195 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
196 (*tube->listen_cb)(tube, NULL, 0, NETEVENT_CLOSED,
201 /* nothing to read now, try later */
202 if(errno != EAGAIN && errno != EINTR) {
203 log_err("rpipe error: %s", strerror(errno));
208 if(tube->cmd_read < sizeof(tube->cmd_len) + tube->cmd_len) {
209 /* not complete, try later */
214 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
215 (*tube->listen_cb)(tube, tube->cmd_msg, tube->cmd_len,
216 NETEVENT_NOERROR, tube->listen_arg);
217 /* also frees the buf */
218 tube->cmd_msg = NULL;
223 tube_handle_write(struct comm_point* c, void* arg, int error,
224 struct comm_reply* ATTR_UNUSED(reply_info))
226 struct tube* tube = (struct tube*)arg;
227 struct tube_res_list* item = tube->res_list;
229 if(error != NETEVENT_NOERROR) {
230 log_err("tube_handle_write net error %d", error);
235 comm_point_stop_listening(c);
239 if(tube->res_write < sizeof(item->len)) {
240 r = write(c->fd, ((uint8_t*)&item->len) + tube->res_write,
241 sizeof(item->len) - tube->res_write);
243 if(errno != EAGAIN && errno != EINTR) {
244 log_err("wpipe error: %s", strerror(errno));
246 return 0; /* try again later */
249 /* error on pipe, must have exited somehow */
250 /* cannot signal this to pipe user */
253 tube->res_write += r;
254 if(tube->res_write < sizeof(item->len))
257 r = write(c->fd, item->buf + tube->res_write - sizeof(item->len),
258 item->len - (tube->res_write - sizeof(item->len)));
260 if(errno != EAGAIN && errno != EINTR) {
261 log_err("wpipe error: %s", strerror(errno));
263 return 0; /* try again later */
266 /* error on pipe, must have exited somehow */
267 /* cannot signal this to pipe user */
270 tube->res_write += r;
271 if(tube->res_write < sizeof(item->len) + item->len)
273 /* done this result, remove it */
276 tube->res_list = tube->res_list->next;
278 if(!tube->res_list) {
279 tube->res_last = NULL;
280 comm_point_stop_listening(c);
286 int tube_write_msg(struct tube* tube, uint8_t* buf, uint32_t len,
294 r = write(fd, &len, sizeof(len));
296 if(errno==EINTR || errno==EAGAIN)
298 log_err("tube msg write failed: %s", strerror(errno));
299 return -1; /* can still continue, perhaps */
302 if(!fd_set_block(fd))
304 /* write remainder */
306 while(d != (ssize_t)sizeof(len)) {
307 if((r=write(fd, ((char*)&len)+d, sizeof(len)-d)) == -1) {
308 log_err("tube msg write failed: %s", strerror(errno));
309 (void)fd_set_nonblock(fd);
315 while(d != (ssize_t)len) {
316 if((r=write(fd, buf+d, len-d)) == -1) {
317 log_err("tube msg write failed: %s", strerror(errno));
318 (void)fd_set_nonblock(fd);
323 if(!fd_set_nonblock(fd))
328 int tube_read_msg(struct tube* tube, uint8_t** buf, uint32_t* len,
337 r = read(fd, len, sizeof(*len));
339 if(errno==EINTR || errno==EAGAIN)
341 log_err("tube msg read failed: %s", strerror(errno));
342 return -1; /* we can still continue, perhaps */
347 if(!fd_set_block(fd))
351 while(d != (ssize_t)sizeof(*len)) {
352 if((r=read(fd, ((char*)len)+d, sizeof(*len)-d)) == -1) {
353 log_err("tube msg read failed: %s", strerror(errno));
354 (void)fd_set_nonblock(fd);
357 if(r == 0) /* EOF */ {
358 (void)fd_set_nonblock(fd);
363 log_assert(*len < 65536*2);
364 *buf = (uint8_t*)malloc(*len);
366 log_err("tube read out of memory");
367 (void)fd_set_nonblock(fd);
371 while(d < (ssize_t)*len) {
372 if((r=read(fd, (*buf)+d, (size_t)((ssize_t)*len)-d)) == -1) {
373 log_err("tube msg read failed: %s", strerror(errno));
374 (void)fd_set_nonblock(fd);
378 if(r == 0) { /* EOF */
379 (void)fd_set_nonblock(fd);
385 if(!fd_set_nonblock(fd)) {
392 /** perform a select() on the fd */
394 pollit(int fd, struct timeval* t)
399 FD_SET(FD_SET_T fd, &r);
401 if(select(fd+1, &r, NULL, NULL, t) == -1) {
405 return (int)(FD_ISSET(fd, &r));
408 int tube_poll(struct tube* tube)
411 memset(&t, 0, sizeof(t));
412 return pollit(tube->sr, &t);
415 int tube_wait(struct tube* tube)
417 return pollit(tube->sr, NULL);
420 int tube_read_fd(struct tube* tube)
425 int tube_setup_bg_listen(struct tube* tube, struct comm_base* base,
426 tube_callback_t* cb, void* arg)
428 tube->listen_cb = cb;
429 tube->listen_arg = arg;
430 if(!(tube->listen_com = comm_point_create_raw(base, tube->sr,
431 0, tube_handle_listen, tube))) {
433 log_err("tube_setup_bg_l: commpoint creation failed");
440 int tube_setup_bg_write(struct tube* tube, struct comm_base* base)
442 if(!(tube->res_com = comm_point_create_raw(base, tube->sw,
443 1, tube_handle_write, tube))) {
445 log_err("tube_setup_bg_w: commpoint creation failed");
452 int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len)
454 struct tube_res_list* item =
455 (struct tube_res_list*)malloc(sizeof(*item));
458 log_err("out of memory for async answer");
464 /* add at back of list, since the first one may be partially written */
466 tube->res_last->next = item;
467 else tube->res_list = item;
468 tube->res_last = item;
469 if(tube->res_list == tube->res_last) {
470 /* first added item, start the write process */
471 comm_point_start_listening(tube->res_com, -1, -1);
476 void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events),
477 void* ATTR_UNUSED(arg))
482 #else /* USE_WINSOCK */
486 struct tube* tube_create(void)
488 /* windows does not have forks like unix, so we only support
489 * threads on windows. And thus the pipe need only connect
490 * threads. We use a mutex and a list of datagrams. */
491 struct tube* tube = (struct tube*)calloc(1, sizeof(*tube));
494 log_err("tube_create: out of memory");
498 tube->event = WSACreateEvent();
499 if(tube->event == WSA_INVALID_EVENT) {
501 log_err("WSACreateEvent: %s", wsa_strerror(WSAGetLastError()));
503 if(!WSAResetEvent(tube->event)) {
504 log_err("WSAResetEvent: %s", wsa_strerror(WSAGetLastError()));
506 lock_basic_init(&tube->res_lock);
507 verbose(VERB_ALGO, "tube created");
511 void tube_delete(struct tube* tube)
514 tube_remove_bg_listen(tube);
515 tube_remove_bg_write(tube);
516 tube_close_read(tube);
517 tube_close_write(tube);
518 if(!WSACloseEvent(tube->event))
519 log_err("WSACloseEvent: %s", wsa_strerror(WSAGetLastError()));
520 lock_basic_destroy(&tube->res_lock);
521 verbose(VERB_ALGO, "tube deleted");
525 void tube_close_read(struct tube* ATTR_UNUSED(tube))
527 verbose(VERB_ALGO, "tube close_read");
530 void tube_close_write(struct tube* ATTR_UNUSED(tube))
532 verbose(VERB_ALGO, "tube close_write");
533 /* wake up waiting reader with an empty queue */
534 if(!WSASetEvent(tube->event)) {
535 log_err("WSASetEvent: %s", wsa_strerror(WSAGetLastError()));
539 void tube_remove_bg_listen(struct tube* tube)
541 verbose(VERB_ALGO, "tube remove_bg_listen");
542 winsock_unregister_wsaevent(&tube->ev_listen);
545 void tube_remove_bg_write(struct tube* tube)
547 verbose(VERB_ALGO, "tube remove_bg_write");
549 struct tube_res_list* np, *p = tube->res_list;
550 tube->res_list = NULL;
551 tube->res_last = NULL;
561 int tube_write_msg(struct tube* tube, uint8_t* buf, uint32_t len,
562 int ATTR_UNUSED(nonblock))
565 verbose(VERB_ALGO, "tube write_msg len %d", (int)len);
566 a = (uint8_t*)memdup(buf, len);
568 log_err("out of memory in tube_write_msg");
571 /* always nonblocking, this pipe cannot get full */
572 return tube_queue_item(tube, a, len);
575 int tube_read_msg(struct tube* tube, uint8_t** buf, uint32_t* len,
578 struct tube_res_list* item = NULL;
579 verbose(VERB_ALGO, "tube read_msg %s", nonblock?"nonblock":"blocking");
581 if(!tube_poll(tube)) {
582 verbose(VERB_ALGO, "tube read_msg nodata");
583 /* nothing ready right now, wait if we want to */
585 return -1; /* would block waiting for items */
589 lock_basic_lock(&tube->res_lock);
591 item = tube->res_list;
592 tube->res_list = item->next;
593 if(tube->res_last == item) {
594 /* the list is now empty */
595 tube->res_last = NULL;
596 verbose(VERB_ALGO, "tube read_msg lastdata");
597 if(!WSAResetEvent(tube->event)) {
598 log_err("WSAResetEvent: %s",
599 wsa_strerror(WSAGetLastError()));
603 lock_basic_unlock(&tube->res_lock);
605 return 0; /* would block waiting for items */
609 verbose(VERB_ALGO, "tube read_msg len %d", (int)*len);
613 int tube_poll(struct tube* tube)
615 struct tube_res_list* item = NULL;
616 lock_basic_lock(&tube->res_lock);
617 item = tube->res_list;
618 lock_basic_unlock(&tube->res_lock);
624 int tube_wait(struct tube* tube)
626 /* block on eventhandle */
627 DWORD res = WSAWaitForMultipleEvents(
628 1 /* one event in array */,
629 &tube->event /* the event to wait for, our pipe signal */,
630 0 /* wait for all events is false */,
631 WSA_INFINITE /* wait, no timeout */,
632 0 /* we are not alertable for IO completion routines */
634 if(res == WSA_WAIT_TIMEOUT) {
637 if(res == WSA_WAIT_IO_COMPLETION) {
638 /* a bit unexpected, since we were not alertable */
644 int tube_read_fd(struct tube* ATTR_UNUSED(tube))
646 /* nothing sensible on Windows */
651 tube_handle_listen(struct comm_point* ATTR_UNUSED(c), void* ATTR_UNUSED(arg),
652 int ATTR_UNUSED(error), struct comm_reply* ATTR_UNUSED(reply_info))
659 tube_handle_write(struct comm_point* ATTR_UNUSED(c), void* ATTR_UNUSED(arg),
660 int ATTR_UNUSED(error), struct comm_reply* ATTR_UNUSED(reply_info))
666 int tube_setup_bg_listen(struct tube* tube, struct comm_base* base,
667 tube_callback_t* cb, void* arg)
669 tube->listen_cb = cb;
670 tube->listen_arg = arg;
671 if(!comm_base_internal(base))
672 return 1; /* ignore when no comm base - testing */
673 return winsock_register_wsaevent(comm_base_internal(base),
674 &tube->ev_listen, tube->event, &tube_handle_signal, tube);
677 int tube_setup_bg_write(struct tube* ATTR_UNUSED(tube),
678 struct comm_base* ATTR_UNUSED(base))
680 /* the queue item routine performs the signaling */
684 int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len)
686 struct tube_res_list* item =
687 (struct tube_res_list*)malloc(sizeof(*item));
688 verbose(VERB_ALGO, "tube queue_item len %d", (int)len);
691 log_err("out of memory for async answer");
697 lock_basic_lock(&tube->res_lock);
698 /* add at back of list, since the first one may be partially written */
700 tube->res_last->next = item;
701 else tube->res_list = item;
702 tube->res_last = item;
703 /* signal the eventhandle */
704 if(!WSASetEvent(tube->event)) {
705 log_err("WSASetEvent: %s", wsa_strerror(WSAGetLastError()));
707 lock_basic_unlock(&tube->res_lock);
711 void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events),
714 struct tube* tube = (struct tube*)arg;
717 verbose(VERB_ALGO, "tube handle_signal");
718 while(tube_poll(tube)) {
719 if(tube_read_msg(tube, &buf, &len, 1)) {
720 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
721 (*tube->listen_cb)(tube, buf, len, NETEVENT_NOERROR,
727 #endif /* USE_WINSOCK */