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"
47 #include "util/ub_event.h"
52 #ifndef HAVE_SOCKETPAIR
53 /** no socketpair() available, like on Minix 3.1.7, use pipe */
54 #define socketpair(f, t, p, sv) pipe(sv)
55 #endif /* HAVE_SOCKETPAIR */
57 struct tube* tube_create(void)
59 struct tube* tube = (struct tube*)calloc(1, sizeof(*tube));
63 log_err("tube_create: out of memory");
69 if(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == -1) {
71 log_err("socketpair: %s", strerror(errno));
78 if(!fd_set_nonblock(tube->sr) || !fd_set_nonblock(tube->sw)) {
80 log_err("tube: cannot set nonblocking");
88 void tube_delete(struct tube* tube)
91 tube_remove_bg_listen(tube);
92 tube_remove_bg_write(tube);
93 /* close fds after deleting commpoints, to be sure.
94 * Also epoll does not like closing fd before event_del */
95 tube_close_read(tube);
96 tube_close_write(tube);
100 void tube_close_read(struct tube* tube)
108 void tube_close_write(struct tube* tube)
116 void tube_remove_bg_listen(struct tube* tube)
118 if(tube->listen_com) {
119 comm_point_delete(tube->listen_com);
120 tube->listen_com = NULL;
123 tube->cmd_msg = NULL;
126 void tube_remove_bg_write(struct tube* tube)
129 comm_point_delete(tube->res_com);
130 tube->res_com = NULL;
133 struct tube_res_list* np, *p = tube->res_list;
134 tube->res_list = NULL;
135 tube->res_last = NULL;
146 tube_handle_listen(struct comm_point* c, void* arg, int error,
147 struct comm_reply* ATTR_UNUSED(reply_info))
149 struct tube* tube = (struct tube*)arg;
151 if(error != NETEVENT_NOERROR) {
152 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
153 (*tube->listen_cb)(tube, NULL, 0, error, tube->listen_arg);
157 if(tube->cmd_read < sizeof(tube->cmd_len)) {
158 /* complete reading the length of control msg */
159 r = read(c->fd, ((uint8_t*)&tube->cmd_len) + tube->cmd_read,
160 sizeof(tube->cmd_len) - tube->cmd_read);
162 /* error has happened or */
163 /* parent closed pipe, must have exited somehow */
164 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
165 (*tube->listen_cb)(tube, NULL, 0, NETEVENT_CLOSED,
170 if(errno != EAGAIN && errno != EINTR) {
171 log_err("rpipe error: %s", strerror(errno));
173 /* nothing to read now, try later */
177 if(tube->cmd_read < sizeof(tube->cmd_len)) {
178 /* not complete, try later */
181 tube->cmd_msg = (uint8_t*)calloc(1, tube->cmd_len);
183 log_err("malloc failure");
188 /* cmd_len has been read, read remainder */
189 r = read(c->fd, tube->cmd_msg+tube->cmd_read-sizeof(tube->cmd_len),
190 tube->cmd_len - (tube->cmd_read - sizeof(tube->cmd_len)));
192 /* error has happened or */
193 /* parent closed pipe, must have exited somehow */
194 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
195 (*tube->listen_cb)(tube, NULL, 0, NETEVENT_CLOSED,
200 /* nothing to read now, try later */
201 if(errno != EAGAIN && errno != EINTR) {
202 log_err("rpipe error: %s", strerror(errno));
207 if(tube->cmd_read < sizeof(tube->cmd_len) + tube->cmd_len) {
208 /* not complete, try later */
213 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
214 (*tube->listen_cb)(tube, tube->cmd_msg, tube->cmd_len,
215 NETEVENT_NOERROR, tube->listen_arg);
216 /* also frees the buf */
217 tube->cmd_msg = NULL;
222 tube_handle_write(struct comm_point* c, void* arg, int error,
223 struct comm_reply* ATTR_UNUSED(reply_info))
225 struct tube* tube = (struct tube*)arg;
226 struct tube_res_list* item = tube->res_list;
228 if(error != NETEVENT_NOERROR) {
229 log_err("tube_handle_write net error %d", error);
234 comm_point_stop_listening(c);
238 if(tube->res_write < sizeof(item->len)) {
239 r = write(c->fd, ((uint8_t*)&item->len) + tube->res_write,
240 sizeof(item->len) - tube->res_write);
242 if(errno != EAGAIN && errno != EINTR) {
243 log_err("wpipe error: %s", strerror(errno));
245 return 0; /* try again later */
248 /* error on pipe, must have exited somehow */
249 /* cannot signal this to pipe user */
252 tube->res_write += r;
253 if(tube->res_write < sizeof(item->len))
256 r = write(c->fd, item->buf + tube->res_write - sizeof(item->len),
257 item->len - (tube->res_write - sizeof(item->len)));
259 if(errno != EAGAIN && errno != EINTR) {
260 log_err("wpipe error: %s", strerror(errno));
262 return 0; /* try again later */
265 /* error on pipe, must have exited somehow */
266 /* cannot signal this to pipe user */
269 tube->res_write += r;
270 if(tube->res_write < sizeof(item->len) + item->len)
272 /* done this result, remove it */
275 tube->res_list = tube->res_list->next;
277 if(!tube->res_list) {
278 tube->res_last = NULL;
279 comm_point_stop_listening(c);
285 int tube_write_msg(struct tube* tube, uint8_t* buf, uint32_t len,
293 r = write(fd, &len, sizeof(len));
295 if(errno==EINTR || errno==EAGAIN)
297 log_err("tube msg write failed: %s", strerror(errno));
298 return -1; /* can still continue, perhaps */
301 if(!fd_set_block(fd))
303 /* write remainder */
305 while(d != (ssize_t)sizeof(len)) {
306 if((r=write(fd, ((char*)&len)+d, sizeof(len)-d)) == -1) {
308 continue; /* temporarily unavail: try again*/
309 log_err("tube msg write failed: %s", strerror(errno));
310 (void)fd_set_nonblock(fd);
316 while(d != (ssize_t)len) {
317 if((r=write(fd, buf+d, len-d)) == -1) {
319 continue; /* temporarily unavail: try again*/
320 log_err("tube msg write failed: %s", strerror(errno));
321 (void)fd_set_nonblock(fd);
326 if(!fd_set_nonblock(fd))
331 int tube_read_msg(struct tube* tube, uint8_t** buf, uint32_t* len,
340 r = read(fd, len, sizeof(*len));
342 if(errno==EINTR || errno==EAGAIN)
344 log_err("tube msg read failed: %s", strerror(errno));
345 return -1; /* we can still continue, perhaps */
350 if(!fd_set_block(fd))
354 while(d != (ssize_t)sizeof(*len)) {
355 if((r=read(fd, ((char*)len)+d, sizeof(*len)-d)) == -1) {
356 log_err("tube msg read failed: %s", strerror(errno));
357 (void)fd_set_nonblock(fd);
360 if(r == 0) /* EOF */ {
361 (void)fd_set_nonblock(fd);
366 log_assert(*len < 65536*2);
367 *buf = (uint8_t*)malloc(*len);
369 log_err("tube read out of memory");
370 (void)fd_set_nonblock(fd);
374 while(d < (ssize_t)*len) {
375 if((r=read(fd, (*buf)+d, (size_t)((ssize_t)*len)-d)) == -1) {
376 log_err("tube msg read failed: %s", strerror(errno));
377 (void)fd_set_nonblock(fd);
381 if(r == 0) { /* EOF */
382 (void)fd_set_nonblock(fd);
388 if(!fd_set_nonblock(fd)) {
395 /** perform a select() on the fd */
397 pollit(int fd, struct timeval* t)
402 FD_SET(FD_SET_T fd, &r);
404 if(select(fd+1, &r, NULL, NULL, t) == -1) {
408 return (int)(FD_ISSET(fd, &r));
411 int tube_poll(struct tube* tube)
414 memset(&t, 0, sizeof(t));
415 return pollit(tube->sr, &t);
418 int tube_wait(struct tube* tube)
420 return pollit(tube->sr, NULL);
423 int tube_read_fd(struct tube* tube)
428 int tube_setup_bg_listen(struct tube* tube, struct comm_base* base,
429 tube_callback_t* cb, void* arg)
431 tube->listen_cb = cb;
432 tube->listen_arg = arg;
433 if(!(tube->listen_com = comm_point_create_raw(base, tube->sr,
434 0, tube_handle_listen, tube))) {
436 log_err("tube_setup_bg_l: commpoint creation failed");
443 int tube_setup_bg_write(struct tube* tube, struct comm_base* base)
445 if(!(tube->res_com = comm_point_create_raw(base, tube->sw,
446 1, tube_handle_write, tube))) {
448 log_err("tube_setup_bg_w: commpoint creation failed");
455 int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len)
457 struct tube_res_list* item =
458 (struct tube_res_list*)malloc(sizeof(*item));
461 log_err("out of memory for async answer");
467 /* add at back of list, since the first one may be partially written */
469 tube->res_last->next = item;
470 else tube->res_list = item;
471 tube->res_last = item;
472 if(tube->res_list == tube->res_last) {
473 /* first added item, start the write process */
474 comm_point_start_listening(tube->res_com, -1, -1);
479 void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events),
480 void* ATTR_UNUSED(arg))
485 #else /* USE_WINSOCK */
489 struct tube* tube_create(void)
491 /* windows does not have forks like unix, so we only support
492 * threads on windows. And thus the pipe need only connect
493 * threads. We use a mutex and a list of datagrams. */
494 struct tube* tube = (struct tube*)calloc(1, sizeof(*tube));
497 log_err("tube_create: out of memory");
501 tube->event = WSACreateEvent();
502 if(tube->event == WSA_INVALID_EVENT) {
504 log_err("WSACreateEvent: %s", wsa_strerror(WSAGetLastError()));
506 if(!WSAResetEvent(tube->event)) {
507 log_err("WSAResetEvent: %s", wsa_strerror(WSAGetLastError()));
509 lock_basic_init(&tube->res_lock);
510 verbose(VERB_ALGO, "tube created");
514 void tube_delete(struct tube* tube)
517 tube_remove_bg_listen(tube);
518 tube_remove_bg_write(tube);
519 tube_close_read(tube);
520 tube_close_write(tube);
521 if(!WSACloseEvent(tube->event))
522 log_err("WSACloseEvent: %s", wsa_strerror(WSAGetLastError()));
523 lock_basic_destroy(&tube->res_lock);
524 verbose(VERB_ALGO, "tube deleted");
528 void tube_close_read(struct tube* ATTR_UNUSED(tube))
530 verbose(VERB_ALGO, "tube close_read");
533 void tube_close_write(struct tube* ATTR_UNUSED(tube))
535 verbose(VERB_ALGO, "tube close_write");
536 /* wake up waiting reader with an empty queue */
537 if(!WSASetEvent(tube->event)) {
538 log_err("WSASetEvent: %s", wsa_strerror(WSAGetLastError()));
542 void tube_remove_bg_listen(struct tube* tube)
544 verbose(VERB_ALGO, "tube remove_bg_listen");
545 ub_winsock_unregister_wsaevent(tube->ev_listen);
548 void tube_remove_bg_write(struct tube* tube)
550 verbose(VERB_ALGO, "tube remove_bg_write");
552 struct tube_res_list* np, *p = tube->res_list;
553 tube->res_list = NULL;
554 tube->res_last = NULL;
564 int tube_write_msg(struct tube* tube, uint8_t* buf, uint32_t len,
565 int ATTR_UNUSED(nonblock))
568 verbose(VERB_ALGO, "tube write_msg len %d", (int)len);
569 a = (uint8_t*)memdup(buf, len);
571 log_err("out of memory in tube_write_msg");
574 /* always nonblocking, this pipe cannot get full */
575 return tube_queue_item(tube, a, len);
578 int tube_read_msg(struct tube* tube, uint8_t** buf, uint32_t* len,
581 struct tube_res_list* item = NULL;
582 verbose(VERB_ALGO, "tube read_msg %s", nonblock?"nonblock":"blocking");
584 if(!tube_poll(tube)) {
585 verbose(VERB_ALGO, "tube read_msg nodata");
586 /* nothing ready right now, wait if we want to */
588 return -1; /* would block waiting for items */
592 lock_basic_lock(&tube->res_lock);
594 item = tube->res_list;
595 tube->res_list = item->next;
596 if(tube->res_last == item) {
597 /* the list is now empty */
598 tube->res_last = NULL;
599 verbose(VERB_ALGO, "tube read_msg lastdata");
600 if(!WSAResetEvent(tube->event)) {
601 log_err("WSAResetEvent: %s",
602 wsa_strerror(WSAGetLastError()));
606 lock_basic_unlock(&tube->res_lock);
608 return 0; /* would block waiting for items */
612 verbose(VERB_ALGO, "tube read_msg len %d", (int)*len);
616 int tube_poll(struct tube* tube)
618 struct tube_res_list* item = NULL;
619 lock_basic_lock(&tube->res_lock);
620 item = tube->res_list;
621 lock_basic_unlock(&tube->res_lock);
627 int tube_wait(struct tube* tube)
629 /* block on eventhandle */
630 DWORD res = WSAWaitForMultipleEvents(
631 1 /* one event in array */,
632 &tube->event /* the event to wait for, our pipe signal */,
633 0 /* wait for all events is false */,
634 WSA_INFINITE /* wait, no timeout */,
635 0 /* we are not alertable for IO completion routines */
637 if(res == WSA_WAIT_TIMEOUT) {
640 if(res == WSA_WAIT_IO_COMPLETION) {
641 /* a bit unexpected, since we were not alertable */
647 int tube_read_fd(struct tube* ATTR_UNUSED(tube))
649 /* nothing sensible on Windows */
654 tube_handle_listen(struct comm_point* ATTR_UNUSED(c), void* ATTR_UNUSED(arg),
655 int ATTR_UNUSED(error), struct comm_reply* ATTR_UNUSED(reply_info))
662 tube_handle_write(struct comm_point* ATTR_UNUSED(c), void* ATTR_UNUSED(arg),
663 int ATTR_UNUSED(error), struct comm_reply* ATTR_UNUSED(reply_info))
669 int tube_setup_bg_listen(struct tube* tube, struct comm_base* base,
670 tube_callback_t* cb, void* arg)
672 tube->listen_cb = cb;
673 tube->listen_arg = arg;
674 if(!comm_base_internal(base))
675 return 1; /* ignore when no comm base - testing */
676 tube->ev_listen = ub_winsock_register_wsaevent(
677 comm_base_internal(base), tube->event, &tube_handle_signal, tube);
678 return tube->ev_listen ? 1 : 0;
681 int tube_setup_bg_write(struct tube* ATTR_UNUSED(tube),
682 struct comm_base* ATTR_UNUSED(base))
684 /* the queue item routine performs the signaling */
688 int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len)
690 struct tube_res_list* item =
691 (struct tube_res_list*)malloc(sizeof(*item));
692 verbose(VERB_ALGO, "tube queue_item len %d", (int)len);
695 log_err("out of memory for async answer");
701 lock_basic_lock(&tube->res_lock);
702 /* add at back of list, since the first one may be partially written */
704 tube->res_last->next = item;
705 else tube->res_list = item;
706 tube->res_last = item;
707 /* signal the eventhandle */
708 if(!WSASetEvent(tube->event)) {
709 log_err("WSASetEvent: %s", wsa_strerror(WSAGetLastError()));
711 lock_basic_unlock(&tube->res_lock);
715 void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events),
718 struct tube* tube = (struct tube*)arg;
721 verbose(VERB_ALGO, "tube handle_signal");
722 while(tube_poll(tube)) {
723 if(tube_read_msg(tube, &buf, &len, 1)) {
724 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
725 (*tube->listen_cb)(tube, buf, len, NETEVENT_NOERROR,
731 #endif /* USE_WINSOCK */