unbound/util/tube.c

   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
  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.
  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 */