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;
122 tube->cmd_msg = NULL;
125 void tube_remove_bg_write(struct tube* tube)
128 comm_point_delete(tube->res_com);
129 tube->res_com = NULL;
132 struct tube_res_list* np, *p = tube->res_list;
133 tube->res_list = NULL;
134 tube->res_last = NULL;
145 tube_handle_listen(struct comm_point* c, void* arg, int error,
146 struct comm_reply* ATTR_UNUSED(reply_info))
148 struct tube* tube = (struct tube*)arg;
150 if(error != NETEVENT_NOERROR) {
151 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
152 (*tube->listen_cb)(tube, NULL, 0, error, tube->listen_arg);
156 if(tube->cmd_read < sizeof(tube->cmd_len)) {
157 /* complete reading the length of control msg */
158 r = read(c->fd, ((uint8_t*)&tube->cmd_len) + tube->cmd_read,
159 sizeof(tube->cmd_len) - tube->cmd_read);
161 /* error has happened or */
162 /* parent closed pipe, must have exited somehow */
163 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
164 (*tube->listen_cb)(tube, NULL, 0, NETEVENT_CLOSED,
169 if(errno != EAGAIN && errno != EINTR) {
170 log_err("rpipe error: %s", strerror(errno));
172 /* nothing to read now, try later */
176 if(tube->cmd_read < sizeof(tube->cmd_len)) {
177 /* not complete, try later */
180 tube->cmd_msg = (uint8_t*)calloc(1, tube->cmd_len);
182 log_err("malloc failure");
187 /* cmd_len has been read, read remainder */
188 r = read(c->fd, tube->cmd_msg+tube->cmd_read-sizeof(tube->cmd_len),
189 tube->cmd_len - (tube->cmd_read - sizeof(tube->cmd_len)));
191 /* error has happened or */
192 /* parent closed pipe, must have exited somehow */
193 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
194 (*tube->listen_cb)(tube, NULL, 0, NETEVENT_CLOSED,
199 /* nothing to read now, try later */
200 if(errno != EAGAIN && errno != EINTR) {
201 log_err("rpipe error: %s", strerror(errno));
206 if(tube->cmd_read < sizeof(tube->cmd_len) + tube->cmd_len) {
207 /* not complete, try later */
212 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
213 (*tube->listen_cb)(tube, tube->cmd_msg, tube->cmd_len,
214 NETEVENT_NOERROR, tube->listen_arg);
215 /* also frees the buf */
216 tube->cmd_msg = NULL;
221 tube_handle_write(struct comm_point* c, void* arg, int error,
222 struct comm_reply* ATTR_UNUSED(reply_info))
224 struct tube* tube = (struct tube*)arg;
225 struct tube_res_list* item = tube->res_list;
227 if(error != NETEVENT_NOERROR) {
228 log_err("tube_handle_write net error %d", error);
233 comm_point_stop_listening(c);
237 if(tube->res_write < sizeof(item->len)) {
238 r = write(c->fd, ((uint8_t*)&item->len) + tube->res_write,
239 sizeof(item->len) - tube->res_write);
241 if(errno != EAGAIN && errno != EINTR) {
242 log_err("wpipe error: %s", strerror(errno));
244 return 0; /* try again later */
247 /* error on pipe, must have exited somehow */
248 /* cannot signal this to pipe user */
251 tube->res_write += r;
252 if(tube->res_write < sizeof(item->len))
255 r = write(c->fd, item->buf + tube->res_write - sizeof(item->len),
256 item->len - (tube->res_write - sizeof(item->len)));
258 if(errno != EAGAIN && errno != EINTR) {
259 log_err("wpipe error: %s", strerror(errno));
261 return 0; /* try again later */
264 /* error on pipe, must have exited somehow */
265 /* cannot signal this to pipe user */
268 tube->res_write += r;
269 if(tube->res_write < sizeof(item->len) + item->len)
271 /* done this result, remove it */
274 tube->res_list = tube->res_list->next;
276 if(!tube->res_list) {
277 tube->res_last = NULL;
278 comm_point_stop_listening(c);
284 int tube_write_msg(struct tube* tube, uint8_t* buf, uint32_t len,
292 r = write(fd, &len, sizeof(len));
294 if(errno==EINTR || errno==EAGAIN)
296 log_err("tube msg write failed: %s", strerror(errno));
297 return -1; /* can still continue, perhaps */
300 if(!fd_set_block(fd))
302 /* write remainder */
304 while(d != (ssize_t)sizeof(len)) {
305 if((r=write(fd, ((char*)&len)+d, sizeof(len)-d)) == -1) {
306 log_err("tube msg write failed: %s", strerror(errno));
307 (void)fd_set_nonblock(fd);
313 while(d != (ssize_t)len) {
314 if((r=write(fd, buf+d, len-d)) == -1) {
315 log_err("tube msg write failed: %s", strerror(errno));
316 (void)fd_set_nonblock(fd);
321 if(!fd_set_nonblock(fd))
326 int tube_read_msg(struct tube* tube, uint8_t** buf, uint32_t* len,
335 r = read(fd, len, sizeof(*len));
337 if(errno==EINTR || errno==EAGAIN)
339 log_err("tube msg read failed: %s", strerror(errno));
340 return -1; /* we can still continue, perhaps */
345 if(!fd_set_block(fd))
349 while(d != (ssize_t)sizeof(*len)) {
350 if((r=read(fd, ((char*)len)+d, sizeof(*len)-d)) == -1) {
351 log_err("tube msg read failed: %s", strerror(errno));
352 (void)fd_set_nonblock(fd);
355 if(r == 0) /* EOF */ {
356 (void)fd_set_nonblock(fd);
361 log_assert(*len < 65536*2);
362 *buf = (uint8_t*)malloc(*len);
364 log_err("tube read out of memory");
365 (void)fd_set_nonblock(fd);
369 while(d < (ssize_t)*len) {
370 if((r=read(fd, (*buf)+d, (size_t)((ssize_t)*len)-d)) == -1) {
371 log_err("tube msg read failed: %s", strerror(errno));
372 (void)fd_set_nonblock(fd);
376 if(r == 0) { /* EOF */
377 (void)fd_set_nonblock(fd);
383 if(!fd_set_nonblock(fd)) {
390 /** perform a select() on the fd */
392 pollit(int fd, struct timeval* t)
397 FD_SET(FD_SET_T fd, &r);
399 if(select(fd+1, &r, NULL, NULL, t) == -1) {
403 return (int)(FD_ISSET(fd, &r));
406 int tube_poll(struct tube* tube)
409 memset(&t, 0, sizeof(t));
410 return pollit(tube->sr, &t);
413 int tube_wait(struct tube* tube)
415 return pollit(tube->sr, NULL);
418 int tube_read_fd(struct tube* tube)
423 int tube_setup_bg_listen(struct tube* tube, struct comm_base* base,
424 tube_callback_t* cb, void* arg)
426 tube->listen_cb = cb;
427 tube->listen_arg = arg;
428 if(!(tube->listen_com = comm_point_create_raw(base, tube->sr,
429 0, tube_handle_listen, tube))) {
431 log_err("tube_setup_bg_l: commpoint creation failed");
438 int tube_setup_bg_write(struct tube* tube, struct comm_base* base)
440 if(!(tube->res_com = comm_point_create_raw(base, tube->sw,
441 1, tube_handle_write, tube))) {
443 log_err("tube_setup_bg_w: commpoint creation failed");
450 int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len)
452 struct tube_res_list* item =
453 (struct tube_res_list*)malloc(sizeof(*item));
456 log_err("out of memory for async answer");
462 /* add at back of list, since the first one may be partially written */
464 tube->res_last->next = item;
465 else tube->res_list = item;
466 tube->res_last = item;
467 if(tube->res_list == tube->res_last) {
468 /* first added item, start the write process */
469 comm_point_start_listening(tube->res_com, -1, -1);
474 void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events),
475 void* ATTR_UNUSED(arg))
480 #else /* USE_WINSOCK */
484 struct tube* tube_create(void)
486 /* windows does not have forks like unix, so we only support
487 * threads on windows. And thus the pipe need only connect
488 * threads. We use a mutex and a list of datagrams. */
489 struct tube* tube = (struct tube*)calloc(1, sizeof(*tube));
492 log_err("tube_create: out of memory");
496 tube->event = WSACreateEvent();
497 if(tube->event == WSA_INVALID_EVENT) {
499 log_err("WSACreateEvent: %s", wsa_strerror(WSAGetLastError()));
501 if(!WSAResetEvent(tube->event)) {
502 log_err("WSAResetEvent: %s", wsa_strerror(WSAGetLastError()));
504 lock_basic_init(&tube->res_lock);
505 verbose(VERB_ALGO, "tube created");
509 void tube_delete(struct tube* tube)
512 tube_remove_bg_listen(tube);
513 tube_remove_bg_write(tube);
514 tube_close_read(tube);
515 tube_close_write(tube);
516 if(!WSACloseEvent(tube->event))
517 log_err("WSACloseEvent: %s", wsa_strerror(WSAGetLastError()));
518 lock_basic_destroy(&tube->res_lock);
519 verbose(VERB_ALGO, "tube deleted");
523 void tube_close_read(struct tube* ATTR_UNUSED(tube))
525 verbose(VERB_ALGO, "tube close_read");
528 void tube_close_write(struct tube* ATTR_UNUSED(tube))
530 verbose(VERB_ALGO, "tube close_write");
531 /* wake up waiting reader with an empty queue */
532 if(!WSASetEvent(tube->event)) {
533 log_err("WSASetEvent: %s", wsa_strerror(WSAGetLastError()));
537 void tube_remove_bg_listen(struct tube* tube)
539 verbose(VERB_ALGO, "tube remove_bg_listen");
540 winsock_unregister_wsaevent(&tube->ev_listen);
543 void tube_remove_bg_write(struct tube* tube)
545 verbose(VERB_ALGO, "tube remove_bg_write");
547 struct tube_res_list* np, *p = tube->res_list;
548 tube->res_list = NULL;
549 tube->res_last = NULL;
559 int tube_write_msg(struct tube* tube, uint8_t* buf, uint32_t len,
560 int ATTR_UNUSED(nonblock))
563 verbose(VERB_ALGO, "tube write_msg len %d", (int)len);
564 a = (uint8_t*)memdup(buf, len);
566 log_err("out of memory in tube_write_msg");
569 /* always nonblocking, this pipe cannot get full */
570 return tube_queue_item(tube, a, len);
573 int tube_read_msg(struct tube* tube, uint8_t** buf, uint32_t* len,
576 struct tube_res_list* item = NULL;
577 verbose(VERB_ALGO, "tube read_msg %s", nonblock?"nonblock":"blocking");
579 if(!tube_poll(tube)) {
580 verbose(VERB_ALGO, "tube read_msg nodata");
581 /* nothing ready right now, wait if we want to */
583 return -1; /* would block waiting for items */
587 lock_basic_lock(&tube->res_lock);
589 item = tube->res_list;
590 tube->res_list = item->next;
591 if(tube->res_last == item) {
592 /* the list is now empty */
593 tube->res_last = NULL;
594 verbose(VERB_ALGO, "tube read_msg lastdata");
595 if(!WSAResetEvent(tube->event)) {
596 log_err("WSAResetEvent: %s",
597 wsa_strerror(WSAGetLastError()));
601 lock_basic_unlock(&tube->res_lock);
603 return 0; /* would block waiting for items */
607 verbose(VERB_ALGO, "tube read_msg len %d", (int)*len);
611 int tube_poll(struct tube* tube)
613 struct tube_res_list* item = NULL;
614 lock_basic_lock(&tube->res_lock);
615 item = tube->res_list;
616 lock_basic_unlock(&tube->res_lock);
622 int tube_wait(struct tube* tube)
624 /* block on eventhandle */
625 DWORD res = WSAWaitForMultipleEvents(
626 1 /* one event in array */,
627 &tube->event /* the event to wait for, our pipe signal */,
628 0 /* wait for all events is false */,
629 WSA_INFINITE /* wait, no timeout */,
630 0 /* we are not alertable for IO completion routines */
632 if(res == WSA_WAIT_TIMEOUT) {
635 if(res == WSA_WAIT_IO_COMPLETION) {
636 /* a bit unexpected, since we were not alertable */
642 int tube_read_fd(struct tube* ATTR_UNUSED(tube))
644 /* nothing sensible on Windows */
649 tube_handle_listen(struct comm_point* ATTR_UNUSED(c), void* ATTR_UNUSED(arg),
650 int ATTR_UNUSED(error), struct comm_reply* ATTR_UNUSED(reply_info))
657 tube_handle_write(struct comm_point* ATTR_UNUSED(c), void* ATTR_UNUSED(arg),
658 int ATTR_UNUSED(error), struct comm_reply* ATTR_UNUSED(reply_info))
664 int tube_setup_bg_listen(struct tube* tube, struct comm_base* base,
665 tube_callback_t* cb, void* arg)
667 tube->listen_cb = cb;
668 tube->listen_arg = arg;
669 if(!comm_base_internal(base))
670 return 1; /* ignore when no comm base - testing */
671 return winsock_register_wsaevent(comm_base_internal(base),
672 &tube->ev_listen, tube->event, &tube_handle_signal, tube);
675 int tube_setup_bg_write(struct tube* ATTR_UNUSED(tube),
676 struct comm_base* ATTR_UNUSED(base))
678 /* the queue item routine performs the signaling */
682 int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len)
684 struct tube_res_list* item =
685 (struct tube_res_list*)malloc(sizeof(*item));
686 verbose(VERB_ALGO, "tube queue_item len %d", (int)len);
689 log_err("out of memory for async answer");
695 lock_basic_lock(&tube->res_lock);
696 /* add at back of list, since the first one may be partially written */
698 tube->res_last->next = item;
699 else tube->res_list = item;
700 tube->res_last = item;
701 /* signal the eventhandle */
702 if(!WSASetEvent(tube->event)) {
703 log_err("WSASetEvent: %s", wsa_strerror(WSAGetLastError()));
705 lock_basic_unlock(&tube->res_lock);
709 void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events),
712 struct tube* tube = (struct tube*)arg;
715 verbose(VERB_ALGO, "tube handle_signal");
716 while(tube_poll(tube)) {
717 if(tube_read_msg(tube, &buf, &len, 1)) {
718 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb));
719 (*tube->listen_cb)(tube, buf, len, NETEVENT_NOERROR,
725 #endif /* USE_WINSOCK */