tube.c revision 238106
11901Swollman/* 21901Swollman * util/tube.c - pipe service 31901Swollman * 41901Swollman * Copyright (c) 2008, NLnet Labs. All rights reserved. 51901Swollman * 61901Swollman * This software is open source. 71901Swollman * 81901Swollman * Redistribution and use in source and binary forms, with or without 91901Swollman * modification, are permitted provided that the following conditions 101901Swollman * are met: 111901Swollman * 121901Swollman * Redistributions of source code must retain the above copyright notice, 131901Swollman * this list of conditions and the following disclaimer. 141901Swollman * 151901Swollman * Redistributions in binary form must reproduce the above copyright notice, 161901Swollman * this list of conditions and the following disclaimer in the documentation 171901Swollman * and/or other materials provided with the distribution. 181901Swollman * 191901Swollman * Neither the name of the NLNET LABS nor the names of its contributors may 201901Swollman * be used to endorse or promote products derived from this software without 211901Swollman * specific prior written permission. 221901Swollman * 231901Swollman * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 241901Swollman * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED 251901Swollman * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR 261901Swollman * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE 271901Swollman * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 281901Swollman * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 291901Swollman * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 301901Swollman * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 311901Swollman * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 321901Swollman * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 331901Swollman * POSSIBILITY OF SUCH DAMAGE. 341901Swollman */ 351901Swollman 361901Swollman/** 371901Swollman * \file 381901Swollman * 391901Swollman * This file contains pipe service functions. 401901Swollman */ 411901Swollman#include "config.h" 421901Swollman#include "util/tube.h" 431901Swollman#include "util/log.h" 441901Swollman#include "util/net_help.h" 451901Swollman#include "util/netevent.h" 461901Swollman#include "util/fptr_wlist.h" 471901Swollman 481901Swollman#ifndef USE_WINSOCK 491901Swollman/* on unix */ 501901Swollman 511901Swollman#ifndef HAVE_SOCKETPAIR 521901Swollman/** no socketpair() available, like on Minix 3.1.7, use pipe */ 531901Swollman#define socketpair(f, t, p, sv) pipe(sv) 541901Swollman#endif /* HAVE_SOCKETPAIR */ 551901Swollman 561901Swollmanstruct tube* tube_create(void) 571901Swollman{ 581901Swollman struct tube* tube = (struct tube*)calloc(1, sizeof(*tube)); 591901Swollman int sv[2]; 601901Swollman if(!tube) { 611901Swollman int err = errno; 621901Swollman log_err("tube_create: out of memory"); 631901Swollman errno = err; 641901Swollman return NULL; 651901Swollman } 661901Swollman tube->sr = -1; 671901Swollman tube->sw = -1; 681901Swollman if(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == -1) { 691901Swollman int err = errno; 701901Swollman log_err("socketpair: %s", strerror(errno)); 711901Swollman free(tube); 721901Swollman errno = err; 731901Swollman return NULL; 741901Swollman } 751901Swollman tube->sr = sv[0]; 761901Swollman tube->sw = sv[1]; 771901Swollman if(!fd_set_nonblock(tube->sr) || !fd_set_nonblock(tube->sw)) { 781901Swollman int err = errno; 791901Swollman log_err("tube: cannot set nonblocking"); 801901Swollman tube_delete(tube); 811901Swollman errno = err; 821901Swollman return NULL; 831901Swollman } 841901Swollman return tube; 851901Swollman} 861901Swollman 871901Swollmanvoid tube_delete(struct tube* tube) 881901Swollman{ 891901Swollman if(!tube) return; 901901Swollman tube_remove_bg_listen(tube); 911901Swollman tube_remove_bg_write(tube); 921901Swollman /* close fds after deleting commpoints, to be sure. 931901Swollman * Also epoll does not like closing fd before event_del */ 941901Swollman tube_close_read(tube); 951901Swollman tube_close_write(tube); 961901Swollman free(tube); 971901Swollman} 981901Swollman 991901Swollmanvoid tube_close_read(struct tube* tube) 1001901Swollman{ 1011901Swollman if(tube->sr != -1) { 1021901Swollman close(tube->sr); 1031901Swollman tube->sr = -1; 1041901Swollman } 1051901Swollman} 1061901Swollman 1071901Swollmanvoid tube_close_write(struct tube* tube) 1081901Swollman{ 1091901Swollman if(tube->sw != -1) { 1101901Swollman close(tube->sw); 1111901Swollman tube->sw = -1; 1121901Swollman } 1131901Swollman} 1141901Swollman 1151901Swollmanvoid tube_remove_bg_listen(struct tube* tube) 1161901Swollman{ 1171901Swollman if(tube->listen_com) { 1181901Swollman comm_point_delete(tube->listen_com); 1191901Swollman tube->listen_com = NULL; 1201901Swollman } 1211901Swollman if(tube->cmd_msg) { 1221901Swollman free(tube->cmd_msg); 1231901Swollman tube->cmd_msg = NULL; 1241901Swollman } 1251901Swollman} 1261901Swollman 1271901Swollmanvoid tube_remove_bg_write(struct tube* tube) 1281901Swollman{ 1291901Swollman if(tube->res_com) { 1301901Swollman comm_point_delete(tube->res_com); 1311901Swollman tube->res_com = NULL; 1321901Swollman } 1331901Swollman if(tube->res_list) { 1341901Swollman struct tube_res_list* np, *p = tube->res_list; 1351901Swollman 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 146int 147tube_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 222int 223tube_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 286int 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 328int 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 *buf = (uint8_t*)malloc(*len); 364 if(!*buf) { 365 log_err("tube read out of memory"); 366 (void)fd_set_nonblock(fd); 367 return 0; 368 } 369 d = 0; 370 while(d != (ssize_t)*len) { 371 if((r=read(fd, (*buf)+d, (size_t)((ssize_t)*len)-d)) == -1) { 372 log_err("tube msg read failed: %s", strerror(errno)); 373 (void)fd_set_nonblock(fd); 374 free(*buf); 375 return 0; 376 } 377 if(r == 0) { /* EOF */ 378 (void)fd_set_nonblock(fd); 379 free(*buf); 380 return 0; 381 } 382 d += r; 383 } 384 if(!fd_set_nonblock(fd)) { 385 free(*buf); 386 return 0; 387 } 388 return 1; 389} 390 391/** perform a select() on the fd */ 392static int 393pollit(int fd, struct timeval* t) 394{ 395 fd_set r; 396#ifndef S_SPLINT_S 397 FD_ZERO(&r); 398 FD_SET(FD_SET_T fd, &r); 399#endif 400 if(select(fd+1, &r, NULL, NULL, t) == -1) { 401 return 0; 402 } 403 errno = 0; 404 return (int)(FD_ISSET(fd, &r)); 405} 406 407int tube_poll(struct tube* tube) 408{ 409 struct timeval t; 410 memset(&t, 0, sizeof(t)); 411 return pollit(tube->sr, &t); 412} 413 414int tube_wait(struct tube* tube) 415{ 416 return pollit(tube->sr, NULL); 417} 418 419int tube_read_fd(struct tube* tube) 420{ 421 return tube->sr; 422} 423 424int tube_setup_bg_listen(struct tube* tube, struct comm_base* base, 425 tube_callback_t* cb, void* arg) 426{ 427 tube->listen_cb = cb; 428 tube->listen_arg = arg; 429 if(!(tube->listen_com = comm_point_create_raw(base, tube->sr, 430 0, tube_handle_listen, tube))) { 431 int err = errno; 432 log_err("tube_setup_bg_l: commpoint creation failed"); 433 errno = err; 434 return 0; 435 } 436 return 1; 437} 438 439int tube_setup_bg_write(struct tube* tube, struct comm_base* base) 440{ 441 if(!(tube->res_com = comm_point_create_raw(base, tube->sw, 442 1, tube_handle_write, tube))) { 443 int err = errno; 444 log_err("tube_setup_bg_w: commpoint creation failed"); 445 errno = err; 446 return 0; 447 } 448 return 1; 449} 450 451int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len) 452{ 453 struct tube_res_list* item = 454 (struct tube_res_list*)malloc(sizeof(*item)); 455 if(!item) { 456 free(msg); 457 log_err("out of memory for async answer"); 458 return 0; 459 } 460 item->buf = msg; 461 item->len = len; 462 item->next = NULL; 463 /* add at back of list, since the first one may be partially written */ 464 if(tube->res_last) 465 tube->res_last->next = item; 466 else tube->res_list = item; 467 tube->res_last = item; 468 if(tube->res_list == tube->res_last) { 469 /* first added item, start the write process */ 470 comm_point_start_listening(tube->res_com, -1, -1); 471 } 472 return 1; 473} 474 475void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events), 476 void* ATTR_UNUSED(arg)) 477{ 478 log_assert(0); 479} 480 481#else /* USE_WINSOCK */ 482/* on windows */ 483 484 485struct tube* tube_create(void) 486{ 487 /* windows does not have forks like unix, so we only support 488 * threads on windows. And thus the pipe need only connect 489 * threads. We use a mutex and a list of datagrams. */ 490 struct tube* tube = (struct tube*)calloc(1, sizeof(*tube)); 491 if(!tube) { 492 int err = errno; 493 log_err("tube_create: out of memory"); 494 errno = err; 495 return NULL; 496 } 497 tube->event = WSACreateEvent(); 498 if(tube->event == WSA_INVALID_EVENT) { 499 free(tube); 500 log_err("WSACreateEvent: %s", wsa_strerror(WSAGetLastError())); 501 } 502 if(!WSAResetEvent(tube->event)) { 503 log_err("WSAResetEvent: %s", wsa_strerror(WSAGetLastError())); 504 } 505 lock_basic_init(&tube->res_lock); 506 verbose(VERB_ALGO, "tube created"); 507 return tube; 508} 509 510void tube_delete(struct tube* tube) 511{ 512 if(!tube) return; 513 tube_remove_bg_listen(tube); 514 tube_remove_bg_write(tube); 515 tube_close_read(tube); 516 tube_close_write(tube); 517 if(!WSACloseEvent(tube->event)) 518 log_err("WSACloseEvent: %s", wsa_strerror(WSAGetLastError())); 519 lock_basic_destroy(&tube->res_lock); 520 verbose(VERB_ALGO, "tube deleted"); 521 free(tube); 522} 523 524void tube_close_read(struct tube* ATTR_UNUSED(tube)) 525{ 526 verbose(VERB_ALGO, "tube close_read"); 527} 528 529void tube_close_write(struct tube* ATTR_UNUSED(tube)) 530{ 531 verbose(VERB_ALGO, "tube close_write"); 532 /* wake up waiting reader with an empty queue */ 533 if(!WSASetEvent(tube->event)) { 534 log_err("WSASetEvent: %s", wsa_strerror(WSAGetLastError())); 535 } 536} 537 538void tube_remove_bg_listen(struct tube* tube) 539{ 540 verbose(VERB_ALGO, "tube remove_bg_listen"); 541 winsock_unregister_wsaevent(&tube->ev_listen); 542} 543 544void tube_remove_bg_write(struct tube* tube) 545{ 546 verbose(VERB_ALGO, "tube remove_bg_write"); 547 if(tube->res_list) { 548 struct tube_res_list* np, *p = tube->res_list; 549 tube->res_list = NULL; 550 tube->res_last = NULL; 551 while(p) { 552 np = p->next; 553 free(p->buf); 554 free(p); 555 p = np; 556 } 557 } 558} 559 560int tube_write_msg(struct tube* tube, uint8_t* buf, uint32_t len, 561 int ATTR_UNUSED(nonblock)) 562{ 563 uint8_t* a; 564 verbose(VERB_ALGO, "tube write_msg len %d", (int)len); 565 a = (uint8_t*)memdup(buf, len); 566 if(!a) { 567 log_err("out of memory in tube_write_msg"); 568 return 0; 569 } 570 /* always nonblocking, this pipe cannot get full */ 571 return tube_queue_item(tube, a, len); 572} 573 574int tube_read_msg(struct tube* tube, uint8_t** buf, uint32_t* len, 575 int nonblock) 576{ 577 struct tube_res_list* item = NULL; 578 verbose(VERB_ALGO, "tube read_msg %s", nonblock?"nonblock":"blocking"); 579 *buf = NULL; 580 if(!tube_poll(tube)) { 581 verbose(VERB_ALGO, "tube read_msg nodata"); 582 /* nothing ready right now, wait if we want to */ 583 if(nonblock) 584 return -1; /* would block waiting for items */ 585 if(!tube_wait(tube)) 586 return 0; 587 } 588 lock_basic_lock(&tube->res_lock); 589 if(tube->res_list) { 590 item = tube->res_list; 591 tube->res_list = item->next; 592 if(tube->res_last == item) { 593 /* the list is now empty */ 594 tube->res_last = NULL; 595 verbose(VERB_ALGO, "tube read_msg lastdata"); 596 if(!WSAResetEvent(tube->event)) { 597 log_err("WSAResetEvent: %s", 598 wsa_strerror(WSAGetLastError())); 599 } 600 } 601 } 602 lock_basic_unlock(&tube->res_lock); 603 if(!item) 604 return 0; /* would block waiting for items */ 605 *buf = item->buf; 606 *len = item->len; 607 free(item); 608 verbose(VERB_ALGO, "tube read_msg len %d", (int)*len); 609 return 1; 610} 611 612int tube_poll(struct tube* tube) 613{ 614 struct tube_res_list* item = NULL; 615 lock_basic_lock(&tube->res_lock); 616 item = tube->res_list; 617 lock_basic_unlock(&tube->res_lock); 618 if(item) 619 return 1; 620 return 0; 621} 622 623int tube_wait(struct tube* tube) 624{ 625 /* block on eventhandle */ 626 DWORD res = WSAWaitForMultipleEvents( 627 1 /* one event in array */, 628 &tube->event /* the event to wait for, our pipe signal */, 629 0 /* wait for all events is false */, 630 WSA_INFINITE /* wait, no timeout */, 631 0 /* we are not alertable for IO completion routines */ 632 ); 633 if(res == WSA_WAIT_TIMEOUT) { 634 return 0; 635 } 636 if(res == WSA_WAIT_IO_COMPLETION) { 637 /* a bit unexpected, since we were not alertable */ 638 return 0; 639 } 640 return 1; 641} 642 643int tube_read_fd(struct tube* ATTR_UNUSED(tube)) 644{ 645 /* nothing sensible on Windows */ 646 return -1; 647} 648 649int 650tube_handle_listen(struct comm_point* ATTR_UNUSED(c), void* ATTR_UNUSED(arg), 651 int ATTR_UNUSED(error), struct comm_reply* ATTR_UNUSED(reply_info)) 652{ 653 log_assert(0); 654 return 0; 655} 656 657int 658tube_handle_write(struct comm_point* ATTR_UNUSED(c), void* ATTR_UNUSED(arg), 659 int ATTR_UNUSED(error), struct comm_reply* ATTR_UNUSED(reply_info)) 660{ 661 log_assert(0); 662 return 0; 663} 664 665int tube_setup_bg_listen(struct tube* tube, struct comm_base* base, 666 tube_callback_t* cb, void* arg) 667{ 668 tube->listen_cb = cb; 669 tube->listen_arg = arg; 670 if(!comm_base_internal(base)) 671 return 1; /* ignore when no comm base - testing */ 672 return winsock_register_wsaevent(comm_base_internal(base), 673 &tube->ev_listen, tube->event, &tube_handle_signal, tube); 674} 675 676int tube_setup_bg_write(struct tube* ATTR_UNUSED(tube), 677 struct comm_base* ATTR_UNUSED(base)) 678{ 679 /* the queue item routine performs the signaling */ 680 return 1; 681} 682 683int tube_queue_item(struct tube* tube, uint8_t* msg, size_t len) 684{ 685 struct tube_res_list* item = 686 (struct tube_res_list*)malloc(sizeof(*item)); 687 verbose(VERB_ALGO, "tube queue_item len %d", (int)len); 688 if(!item) { 689 free(msg); 690 log_err("out of memory for async answer"); 691 return 0; 692 } 693 item->buf = msg; 694 item->len = len; 695 item->next = NULL; 696 lock_basic_lock(&tube->res_lock); 697 /* add at back of list, since the first one may be partially written */ 698 if(tube->res_last) 699 tube->res_last->next = item; 700 else tube->res_list = item; 701 tube->res_last = item; 702 /* signal the eventhandle */ 703 if(!WSASetEvent(tube->event)) { 704 log_err("WSASetEvent: %s", wsa_strerror(WSAGetLastError())); 705 } 706 lock_basic_unlock(&tube->res_lock); 707 return 1; 708} 709 710void tube_handle_signal(int ATTR_UNUSED(fd), short ATTR_UNUSED(events), 711 void* arg) 712{ 713 struct tube* tube = (struct tube*)arg; 714 uint8_t* buf; 715 uint32_t len = 0; 716 verbose(VERB_ALGO, "tube handle_signal"); 717 while(tube_poll(tube)) { 718 if(tube_read_msg(tube, &buf, &len, 1)) { 719 fptr_ok(fptr_whitelist_tube_listen(tube->listen_cb)); 720 (*tube->listen_cb)(tube, buf, len, NETEVENT_NOERROR, 721 tube->listen_arg); 722 } 723 } 724} 725 726#endif /* USE_WINSOCK */ 727