xref: /freebsd/contrib/unbound/util/tube.c (revision 38f0b757fd84d17d0fc24739a7cda160c4516d81)
1 /*
2  * util/tube.c - pipe service
3  *
4  * Copyright (c) 2008, NLnet Labs. All rights reserved.
5  *
6  * This software is open source.
7  *
8  * Redistribution and use in source and binary forms, with or without
9  * modification, are permitted provided that the following conditions
10  * are met:
11  *
12  * Redistributions of source code must retain the above copyright notice,
13  * this list of conditions and the following disclaimer.
14  *
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.
18  *
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.
22  *
23  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
24  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
25  * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
26  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE
27  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
28  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
29  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
30  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
31  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
32  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
33  * POSSIBILITY OF SUCH DAMAGE.
34  */
35 
36 /**
37  * \file
38  *
39  * This file contains pipe service functions.
40  */
41 #include "config.h"
42 #include "util/tube.h"
43 #include "util/log.h"
44 #include "util/net_help.h"
45 #include "util/netevent.h"
46 #include "util/fptr_wlist.h"
47 
48 #ifndef USE_WINSOCK
49 /* on unix */
50 
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 */
55 
56 struct tube* tube_create(void)
57 {
58 	struct tube* tube = (struct tube*)calloc(1, sizeof(*tube));
59 	int sv[2];
60 	if(!tube) {
61 		int err = errno;
62 		log_err("tube_create: out of memory");
63 		errno = err;
64 		return NULL;
65 	}
66 	tube->sr = -1;
67 	tube->sw = -1;
68 	if(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == -1) {
69 		int err = errno;
70 		log_err("socketpair: %s", strerror(errno));
71 		free(tube);
72 		errno = err;
73 		return NULL;
74 	}
75 	tube->sr = sv[0];
76 	tube->sw = sv[1];
77 	if(!fd_set_nonblock(tube->sr) || !fd_set_nonblock(tube->sw)) {
78 		int err = errno;
79 		log_err("tube: cannot set nonblocking");
80 		tube_delete(tube);
81 		errno = err;
82 		return NULL;
83 	}
84 	return tube;
85 }
86 
87 void tube_delete(struct tube* tube)
88 {
89 	if(!tube) return;
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);
96 	free(tube);
97 }
98 
99 void tube_close_read(struct tube* tube)
100 {
101 	if(tube->sr != -1) {
102 		close(tube->sr);
103 		tube->sr = -1;
104 	}
105 }
106 
107 void tube_close_write(struct tube* tube)
108 {
109 	if(tube->sw != -1) {
110 		close(tube->sw);
111 		tube->sw = -1;
112 	}
113 }
114 
115 void tube_remove_bg_listen(struct tube* tube)
116 {
117 	if(tube->listen_com) {
118 		comm_point_delete(tube->listen_com);
119 		tube->listen_com = NULL;
120 	}
121 	if(tube->cmd_msg) {
122 		free(tube->cmd_msg);
123 		tube->cmd_msg = NULL;
124 	}
125 }
126 
127 void tube_remove_bg_write(struct tube* tube)
128 {
129 	if(tube->res_com) {
130 		comm_point_delete(tube->res_com);
131 		tube->res_com = NULL;
132 	}
133 	if(tube->res_list) {
134 		struct tube_res_list* np, *p = tube->res_list;
135 		tube->res_list = NULL;
136 		tube->res_last = NULL;
137 		while(p) {
138 			np = p->next;
139 			free(p->buf);
140 			free(p);
141 			p = np;
142 		}
143 	}
144 }
145 
146 int
147 tube_handle_listen(struct comm_point* c, void* arg, int error,
148         struct comm_reply* ATTR_UNUSED(reply_info))
149 {
150 	struct tube* tube = (struct tube*)arg;
151 	ssize_t r;
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);
155 		return 0;
156 	}
157 
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);
162 		if(r==0) {
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,
167 				tube->listen_arg);
168 			return 0;
169 		}
170 		if(r==-1) {
171 			if(errno != EAGAIN && errno != EINTR) {
172 				log_err("rpipe error: %s", strerror(errno));
173 			}
174 			/* nothing to read now, try later */
175 			return 0;
176 		}
177 		tube->cmd_read += r;
178 		if(tube->cmd_read < sizeof(tube->cmd_len)) {
179 			/* not complete, try later */
180 			return 0;
181 		}
182 		tube->cmd_msg = (uint8_t*)calloc(1, tube->cmd_len);
183 		if(!tube->cmd_msg) {
184 			log_err("malloc failure");
185 			tube->cmd_read = 0;
186 			return 0;
187 		}
188 	}
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)));
192 	if(r==0) {
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,
197 			tube->listen_arg);
198 		return 0;
199 	}
200 	if(r==-1) {
201 		/* nothing to read now, try later */
202 		if(errno != EAGAIN && errno != EINTR) {
203 			log_err("rpipe error: %s", strerror(errno));
204 		}
205 		return 0;
206 	}
207 	tube->cmd_read += r;
208 	if(tube->cmd_read < sizeof(tube->cmd_len) + tube->cmd_len) {
209 		/* not complete, try later */
210 		return 0;
211 	}
212 	tube->cmd_read = 0;
213 
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;
219 	return 0;
220 }
221 
222 int
223 tube_handle_write(struct comm_point* c, void* arg, int error,
224         struct comm_reply* ATTR_UNUSED(reply_info))
225 {
226 	struct tube* tube = (struct tube*)arg;
227 	struct tube_res_list* item = tube->res_list;
228 	ssize_t r;
229 	if(error != NETEVENT_NOERROR) {
230 		log_err("tube_handle_write net error %d", error);
231 		return 0;
232 	}
233 
234 	if(!item) {
235 		comm_point_stop_listening(c);
236 		return 0;
237 	}
238 
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);
242 		if(r == -1) {
243 			if(errno != EAGAIN && errno != EINTR) {
244 				log_err("wpipe error: %s", strerror(errno));
245 			}
246 			return 0; /* try again later */
247 		}
248 		if(r == 0) {
249 			/* error on pipe, must have exited somehow */
250 			/* cannot signal this to pipe user */
251 			return 0;
252 		}
253 		tube->res_write += r;
254 		if(tube->res_write < sizeof(item->len))
255 			return 0;
256 	}
257 	r = write(c->fd, item->buf + tube->res_write - sizeof(item->len),
258 		item->len - (tube->res_write - sizeof(item->len)));
259 	if(r == -1) {
260 		if(errno != EAGAIN && errno != EINTR) {
261 			log_err("wpipe error: %s", strerror(errno));
262 		}
263 		return 0; /* try again later */
264 	}
265 	if(r == 0) {
266 		/* error on pipe, must have exited somehow */
267 		/* cannot signal this to pipe user */
268 		return 0;
269 	}
270 	tube->res_write += r;
271 	if(tube->res_write < sizeof(item->len) + item->len)
272 		return 0;
273 	/* done this result, remove it */
274 	free(item->buf);
275 	item->buf = NULL;
276 	tube->res_list = tube->res_list->next;
277 	free(item);
278 	if(!tube->res_list) {
279 		tube->res_last = NULL;
280 		comm_point_stop_listening(c);
281 	}
282 	tube->res_write = 0;
283 	return 0;
284 }
285 
286 int tube_write_msg(struct tube* tube, uint8_t* buf, uint32_t len,
287         int nonblock)
288 {
289 	ssize_t r, d;
290 	int fd = tube->sw;
291 
292 	/* test */
293 	if(nonblock) {
294 		r = write(fd, &len, sizeof(len));
295 		if(r == -1) {
296 			if(errno==EINTR || errno==EAGAIN)
297 				return -1;
298 			log_err("tube msg write failed: %s", strerror(errno));
299 			return -1; /* can still continue, perhaps */
300 		}
301 	} else r = 0;
302 	if(!fd_set_block(fd))
303 		return 0;
304 	/* write remainder */
305 	d = r;
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);
310 			return 0;
311 		}
312 		d += r;
313 	}
314 	d = 0;
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);
319 			return 0;
320 		}
321 		d += r;
322 	}
323 	if(!fd_set_nonblock(fd))
324 		return 0;
325 	return 1;
326 }
327 
328 int tube_read_msg(struct tube* tube, uint8_t** buf, uint32_t* len,
329         int nonblock)
330 {
331 	ssize_t r, d;
332 	int fd = tube->sr;
333 
334 	/* test */
335 	*len = 0;
336 	if(nonblock) {
337 		r = read(fd, len, sizeof(*len));
338 		if(r == -1) {
339 			if(errno==EINTR || errno==EAGAIN)
340 				return -1;
341 			log_err("tube msg read failed: %s", strerror(errno));
342 			return -1; /* we can still continue, perhaps */
343 		}
344 		if(r == 0) /* EOF */
345 			return 0;
346 	} else r = 0;
347 	if(!fd_set_block(fd))
348 		return 0;
349 	/* read remainder */
350 	d = r;
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);
355 			return 0;
356 		}
357 		if(r == 0) /* EOF */ {
358 			(void)fd_set_nonblock(fd);
359 			return 0;
360 		}
361 		d += r;
362 	}
363 	log_assert(*len < 65536*2);
364 	*buf = (uint8_t*)malloc(*len);
365 	if(!*buf) {
366 		log_err("tube read out of memory");
367 		(void)fd_set_nonblock(fd);
368 		return 0;
369 	}
370 	d = 0;
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);
375 			free(*buf);
376 			return 0;
377 		}
378 		if(r == 0) { /* EOF */
379 			(void)fd_set_nonblock(fd);
380 			free(*buf);
381 			return 0;
382 		}
383 		d += r;
384 	}
385 	if(!fd_set_nonblock(fd)) {
386 		free(*buf);
387 		return 0;
388 	}
389 	return 1;
390 }
391 
392 /** perform a select() on the fd */
393 static int
394 pollit(int fd, struct timeval* t)
395 {
396 	fd_set r;
397 #ifndef S_SPLINT_S
398 	FD_ZERO(&r);
399 	FD_SET(FD_SET_T fd, &r);
400 #endif
401 	if(select(fd+1, &r, NULL, NULL, t) == -1) {
402 		return 0;
403 	}
404 	errno = 0;
405 	return (int)(FD_ISSET(fd, &r));
406 }
407 
408 int tube_poll(struct tube* tube)
409 {
410 	struct timeval t;
411 	memset(&t, 0, sizeof(t));
412 	return pollit(tube->sr, &t);
413 }
414 
415 int tube_wait(struct tube* tube)
416 {
417 	return pollit(tube->sr, NULL);
418 }
419 
420 int tube_read_fd(struct tube* tube)
421 {
422 	return tube->sr;
423 }
424 
425 int tube_setup_bg_listen(struct tube* tube, struct comm_base* base,
426         tube_callback_t* cb, void* arg)
427 {
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))) {
432 		int err = errno;
433 		log_err("tube_setup_bg_l: commpoint creation failed");
434 		errno = err;
435 		return 0;
436 	}
437 	return 1;
438 }
439 
440 int tube_setup_bg_write(struct tube* tube, struct comm_base* base)
441 {
442 	if(!(tube->res_com = comm_point_create_raw(base, tube->sw,
443 		1, tube_handle_write, tube))) {
444 		int err = errno;
445 		log_err("tube_setup_bg_w: commpoint creation failed");
446 		errno = err;
447 		return 0;
448 	}
449 	return 1;
450 }
451 
452 int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len)
453 {
454 	struct tube_res_list* item =
455 		(struct tube_res_list*)malloc(sizeof(*item));
456 	if(!item) {
457 		free(msg);
458 		log_err("out of memory for async answer");
459 		return 0;
460 	}
461 	item->buf = msg;
462 	item->len = len;
463 	item->next = NULL;
464 	/* add at back of list, since the first one may be partially written */
465 	if(tube->res_last)
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);
472 	}
473 	return 1;
474 }
475 
476 void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events),
477 	void* ATTR_UNUSED(arg))
478 {
479 	log_assert(0);
480 }
481 
482 #else /* USE_WINSOCK */
483 /* on windows */
484 
485 
486 struct tube* tube_create(void)
487 {
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));
492 	if(!tube) {
493 		int err = errno;
494 		log_err("tube_create: out of memory");
495 		errno = err;
496 		return NULL;
497 	}
498 	tube->event = WSACreateEvent();
499 	if(tube->event == WSA_INVALID_EVENT) {
500 		free(tube);
501 		log_err("WSACreateEvent: %s", wsa_strerror(WSAGetLastError()));
502 	}
503 	if(!WSAResetEvent(tube->event)) {
504 		log_err("WSAResetEvent: %s", wsa_strerror(WSAGetLastError()));
505 	}
506 	lock_basic_init(&tube->res_lock);
507 	verbose(VERB_ALGO, "tube created");
508 	return tube;
509 }
510 
511 void tube_delete(struct tube* tube)
512 {
513 	if(!tube) return;
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");
522 	free(tube);
523 }
524 
525 void tube_close_read(struct tube* ATTR_UNUSED(tube))
526 {
527 	verbose(VERB_ALGO, "tube close_read");
528 }
529 
530 void tube_close_write(struct tube* ATTR_UNUSED(tube))
531 {
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()));
536 	}
537 }
538 
539 void tube_remove_bg_listen(struct tube* tube)
540 {
541 	verbose(VERB_ALGO, "tube remove_bg_listen");
542 	winsock_unregister_wsaevent(&tube->ev_listen);
543 }
544 
545 void tube_remove_bg_write(struct tube* tube)
546 {
547 	verbose(VERB_ALGO, "tube remove_bg_write");
548 	if(tube->res_list) {
549 		struct tube_res_list* np, *p = tube->res_list;
550 		tube->res_list = NULL;
551 		tube->res_last = NULL;
552 		while(p) {
553 			np = p->next;
554 			free(p->buf);
555 			free(p);
556 			p = np;
557 		}
558 	}
559 }
560 
561 int tube_write_msg(struct tube* tube, uint8_t* buf, uint32_t len,
562         int ATTR_UNUSED(nonblock))
563 {
564 	uint8_t* a;
565 	verbose(VERB_ALGO, "tube write_msg len %d", (int)len);
566 	a = (uint8_t*)memdup(buf, len);
567 	if(!a) {
568 		log_err("out of memory in tube_write_msg");
569 		return 0;
570 	}
571 	/* always nonblocking, this pipe cannot get full */
572 	return tube_queue_item(tube, a, len);
573 }
574 
575 int tube_read_msg(struct tube* tube, uint8_t** buf, uint32_t* len,
576         int nonblock)
577 {
578 	struct tube_res_list* item = NULL;
579 	verbose(VERB_ALGO, "tube read_msg %s", nonblock?"nonblock":"blocking");
580 	*buf = NULL;
581 	if(!tube_poll(tube)) {
582 		verbose(VERB_ALGO, "tube read_msg nodata");
583 		/* nothing ready right now, wait if we want to */
584 		if(nonblock)
585 			return -1; /* would block waiting for items */
586 		if(!tube_wait(tube))
587 			return 0;
588 	}
589 	lock_basic_lock(&tube->res_lock);
590 	if(tube->res_list) {
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()));
600 			}
601 		}
602 	}
603 	lock_basic_unlock(&tube->res_lock);
604 	if(!item)
605 		return 0; /* would block waiting for items */
606 	*buf = item->buf;
607 	*len = item->len;
608 	free(item);
609 	verbose(VERB_ALGO, "tube read_msg len %d", (int)*len);
610 	return 1;
611 }
612 
613 int tube_poll(struct tube* tube)
614 {
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);
619 	if(item)
620 		return 1;
621 	return 0;
622 }
623 
624 int tube_wait(struct tube* tube)
625 {
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 */
633 		);
634 	if(res == WSA_WAIT_TIMEOUT) {
635 		return 0;
636 	}
637 	if(res == WSA_WAIT_IO_COMPLETION) {
638 		/* a bit unexpected, since we were not alertable */
639 		return 0;
640 	}
641 	return 1;
642 }
643 
644 int tube_read_fd(struct tube* ATTR_UNUSED(tube))
645 {
646 	/* nothing sensible on Windows */
647 	return -1;
648 }
649 
650 int
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))
653 {
654 	log_assert(0);
655 	return 0;
656 }
657 
658 int
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))
661 {
662 	log_assert(0);
663 	return 0;
664 }
665 
666 int tube_setup_bg_listen(struct tube* tube, struct comm_base* base,
667         tube_callback_t* cb, void* arg)
668 {
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);
675 }
676 
677 int tube_setup_bg_write(struct tube* ATTR_UNUSED(tube),
678 	struct comm_base* ATTR_UNUSED(base))
679 {
680 	/* the queue item routine performs the signaling */
681 	return 1;
682 }
683 
684 int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len)
685 {
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);
689 	if(!item) {
690 		free(msg);
691 		log_err("out of memory for async answer");
692 		return 0;
693 	}
694 	item->buf = msg;
695 	item->len = len;
696 	item->next = NULL;
697 	lock_basic_lock(&tube->res_lock);
698 	/* add at back of list, since the first one may be partially written */
699 	if(tube->res_last)
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()));
706 	}
707 	lock_basic_unlock(&tube->res_lock);
708 	return 1;
709 }
710 
711 void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events),
712 	void* arg)
713 {
714 	struct tube* tube = (struct tube*)arg;
715 	uint8_t* buf;
716 	uint32_t len = 0;
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,
722 				tube->listen_arg);
723 		}
724 	}
725 }
726 
727 #endif /* USE_WINSOCK */
728