1 /* Copyright Joyent, Inc. and other Node contributors. All rights reserved. 2 * 3 * Permission is hereby granted, free of charge, to any person obtaining a copy 4 * of this software and associated documentation files (the "Software"), to 5 * deal in the Software without restriction, including without limitation the 6 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or 7 * sell copies of the Software, and to permit persons to whom the Software is 8 * furnished to do so, subject to the following conditions: 9 * 10 * The above copyright notice and this permission notice shall be included in 11 * all copies or substantial portions of the Software. 12 * 13 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 14 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 15 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 16 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 17 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING 18 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS 19 * IN THE SOFTWARE. 20 */ 21 22 #include "uv.h" 23 #include "internal.h" 24 25 #include <stdio.h> 26 #include <stdlib.h> 27 #include <string.h> 28 #include <assert.h> 29 #include <errno.h> 30 31 #include <sys/types.h> 32 #include <sys/socket.h> 33 #include <sys/uio.h> 34 #include <sys/un.h> 35 #include <unistd.h> 36 #include <limits.h> /* IOV_MAX */ 37 38 #if defined(__APPLE__) 39 # include <sys/event.h> 40 # include <sys/time.h> 41 # include <sys/select.h> 42 43 /* Forward declaration */ 44 typedef struct uv__stream_select_s uv__stream_select_t; 45 46 struct uv__stream_select_s { 47 uv_stream_t* stream; 48 uv_thread_t thread; 49 uv_sem_t close_sem; 50 uv_sem_t async_sem; 51 uv_async_t async; 52 int events; 53 int fake_fd; 54 int int_fd; 55 int fd; 56 fd_set* sread; 57 size_t sread_sz; 58 fd_set* swrite; 59 size_t swrite_sz; 60 }; 61 #endif /* defined(__APPLE__) */ 62 63 union uv__cmsg { 64 struct cmsghdr hdr; 65 /* This cannot be larger because of the IBMi PASE limitation that 66 * the total size of control messages cannot exceed 256 bytes. 67 */ 68 char pad[256]; 69 }; 70 71 STATIC_ASSERT(256 == sizeof(union uv__cmsg)); 72 73 static void uv__stream_connect(uv_stream_t*); 74 static void uv__write(uv_stream_t* stream); 75 static void uv__read(uv_stream_t* stream); 76 static void uv__stream_io(uv_loop_t* loop, uv__io_t* w, unsigned int events); 77 static void uv__write_callbacks(uv_stream_t* stream); 78 static size_t uv__write_req_size(uv_write_t* req); 79 static void uv__drain(uv_stream_t* stream); 80 81 82 void uv__stream_init(uv_loop_t* loop, 83 uv_stream_t* stream, 84 uv_handle_type type) { 85 int err; 86 87 uv__handle_init(loop, (uv_handle_t*)stream, type); 88 stream->read_cb = NULL; 89 stream->alloc_cb = NULL; 90 stream->close_cb = NULL; 91 stream->connection_cb = NULL; 92 stream->connect_req = NULL; 93 stream->shutdown_req = NULL; 94 stream->accepted_fd = -1; 95 stream->queued_fds = NULL; 96 stream->delayed_error = 0; 97 uv__queue_init(&stream->write_queue); 98 uv__queue_init(&stream->write_completed_queue); 99 stream->write_queue_size = 0; 100 101 if (loop->emfile_fd == -1) { 102 err = uv__open_cloexec("/dev/null", O_RDONLY); 103 if (err < 0) 104 /* In the rare case that "/dev/null" isn't mounted open "/" 105 * instead. 106 */ 107 err = uv__open_cloexec("/", O_RDONLY); 108 if (err >= 0) 109 loop->emfile_fd = err; 110 } 111 112 #if defined(__APPLE__) 113 stream->select = NULL; 114 #endif /* defined(__APPLE_) */ 115 116 uv__io_init(&stream->io_watcher, uv__stream_io, -1); 117 } 118 119 120 static void uv__stream_osx_interrupt_select(uv_stream_t* stream) { 121 #if defined(__APPLE__) 122 /* Notify select() thread about state change */ 123 uv__stream_select_t* s; 124 int r; 125 126 s = stream->select; 127 if (s == NULL) 128 return; 129 130 /* Interrupt select() loop 131 * NOTE: fake_fd and int_fd are socketpair(), thus writing to one will 132 * emit read event on other side 133 */ 134 do 135 r = write(s->fake_fd, "x", 1); 136 while (r == -1 && errno == EINTR); 137 138 assert(r == 1); 139 #else /* !defined(__APPLE__) */ 140 /* No-op on any other platform */ 141 #endif /* !defined(__APPLE__) */ 142 } 143 144 145 #if defined(__APPLE__) 146 static void uv__stream_osx_select(void* arg) { 147 uv_stream_t* stream; 148 uv__stream_select_t* s; 149 char buf[1024]; 150 int events; 151 int fd; 152 int r; 153 int max_fd; 154 155 stream = arg; 156 s = stream->select; 157 fd = s->fd; 158 159 if (fd > s->int_fd) 160 max_fd = fd; 161 else 162 max_fd = s->int_fd; 163 164 for (;;) { 165 /* Terminate on semaphore */ 166 if (uv_sem_trywait(&s->close_sem) == 0) 167 break; 168 169 /* Watch fd using select(2) */ 170 memset(s->sread, 0, s->sread_sz); 171 memset(s->swrite, 0, s->swrite_sz); 172 173 if (uv__io_active(&stream->io_watcher, POLLIN)) 174 FD_SET(fd, s->sread); 175 if (uv__io_active(&stream->io_watcher, POLLOUT)) 176 FD_SET(fd, s->swrite); 177 FD_SET(s->int_fd, s->sread); 178 179 /* Wait indefinitely for fd events */ 180 r = select(max_fd + 1, s->sread, s->swrite, NULL, NULL); 181 if (r == -1) { 182 if (errno == EINTR) 183 continue; 184 185 /* XXX: Possible?! */ 186 abort(); 187 } 188 189 /* Ignore timeouts */ 190 if (r == 0) 191 continue; 192 193 /* Empty socketpair's buffer in case of interruption */ 194 if (FD_ISSET(s->int_fd, s->sread)) 195 for (;;) { 196 r = read(s->int_fd, buf, sizeof(buf)); 197 198 if (r == sizeof(buf)) 199 continue; 200 201 if (r != -1) 202 break; 203 204 if (errno == EAGAIN || errno == EWOULDBLOCK) 205 break; 206 207 if (errno == EINTR) 208 continue; 209 210 abort(); 211 } 212 213 /* Handle events */ 214 events = 0; 215 if (FD_ISSET(fd, s->sread)) 216 events |= POLLIN; 217 if (FD_ISSET(fd, s->swrite)) 218 events |= POLLOUT; 219 220 assert(events != 0 || FD_ISSET(s->int_fd, s->sread)); 221 if (events != 0) { 222 ACCESS_ONCE(int, s->events) = events; 223 224 uv_async_send(&s->async); 225 uv_sem_wait(&s->async_sem); 226 227 /* Should be processed at this stage */ 228 assert((s->events == 0) || (stream->flags & UV_HANDLE_CLOSING)); 229 } 230 } 231 } 232 233 234 static void uv__stream_osx_select_cb(uv_async_t* handle) { 235 uv__stream_select_t* s; 236 uv_stream_t* stream; 237 int events; 238 239 s = container_of(handle, uv__stream_select_t, async); 240 stream = s->stream; 241 242 /* Get and reset stream's events */ 243 events = s->events; 244 ACCESS_ONCE(int, s->events) = 0; 245 246 assert(events != 0); 247 assert(events == (events & (POLLIN | POLLOUT))); 248 249 /* Invoke callback on event-loop */ 250 if ((events & POLLIN) && uv__io_active(&stream->io_watcher, POLLIN)) 251 uv__stream_io(stream->loop, &stream->io_watcher, POLLIN); 252 253 if ((events & POLLOUT) && uv__io_active(&stream->io_watcher, POLLOUT)) 254 uv__stream_io(stream->loop, &stream->io_watcher, POLLOUT); 255 256 if (stream->flags & UV_HANDLE_CLOSING) 257 return; 258 259 /* NOTE: It is important to do it here, otherwise `select()` might be called 260 * before the actual `uv__read()`, leading to the blocking syscall 261 */ 262 uv_sem_post(&s->async_sem); 263 } 264 265 266 static void uv__stream_osx_cb_close(uv_handle_t* async) { 267 uv__stream_select_t* s; 268 269 s = container_of(async, uv__stream_select_t, async); 270 uv__free(s); 271 } 272 273 274 int uv__stream_try_select(uv_stream_t* stream, int* fd) { 275 /* 276 * kqueue doesn't work with some files from /dev mount on osx. 277 * select(2) in separate thread for those fds 278 */ 279 280 struct kevent filter[1]; 281 struct kevent events[1]; 282 struct timespec timeout; 283 uv__stream_select_t* s; 284 int fds[2]; 285 int err; 286 int ret; 287 int kq; 288 int old_fd; 289 int max_fd; 290 size_t sread_sz; 291 size_t swrite_sz; 292 293 kq = kqueue(); 294 if (kq == -1) { 295 perror("(libuv) kqueue()"); 296 return UV__ERR(errno); 297 } 298 299 EV_SET(&filter[0], *fd, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, 0); 300 301 /* Use small timeout, because we only want to capture EINVALs */ 302 timeout.tv_sec = 0; 303 timeout.tv_nsec = 1; 304 305 do 306 ret = kevent(kq, filter, 1, events, 1, &timeout); 307 while (ret == -1 && errno == EINTR); 308 309 uv__close(kq); 310 311 if (ret == -1) 312 return UV__ERR(errno); 313 314 if (ret == 0 || (events[0].flags & EV_ERROR) == 0 || events[0].data != EINVAL) 315 return 0; 316 317 /* At this point we definitely know that this fd won't work with kqueue */ 318 319 /* 320 * Create fds for io watcher and to interrupt the select() loop. 321 * NOTE: do it ahead of malloc below to allocate enough space for fd_sets 322 */ 323 if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds)) 324 return UV__ERR(errno); 325 326 max_fd = *fd; 327 if (fds[1] > max_fd) 328 max_fd = fds[1]; 329 330 sread_sz = ROUND_UP(max_fd + 1, sizeof(uint32_t) * NBBY) / NBBY; 331 swrite_sz = sread_sz; 332 333 s = uv__malloc(sizeof(*s) + sread_sz + swrite_sz); 334 if (s == NULL) { 335 err = UV_ENOMEM; 336 goto failed_malloc; 337 } 338 339 s->events = 0; 340 s->fd = *fd; 341 s->sread = (fd_set*) ((char*) s + sizeof(*s)); 342 s->sread_sz = sread_sz; 343 s->swrite = (fd_set*) ((char*) s->sread + sread_sz); 344 s->swrite_sz = swrite_sz; 345 346 err = uv_async_init(stream->loop, &s->async, uv__stream_osx_select_cb); 347 if (err) 348 goto failed_async_init; 349 350 s->async.flags |= UV_HANDLE_INTERNAL; 351 uv__handle_unref(&s->async); 352 353 err = uv_sem_init(&s->close_sem, 0); 354 if (err != 0) 355 goto failed_close_sem_init; 356 357 err = uv_sem_init(&s->async_sem, 0); 358 if (err != 0) 359 goto failed_async_sem_init; 360 361 s->fake_fd = fds[0]; 362 s->int_fd = fds[1]; 363 364 old_fd = *fd; 365 s->stream = stream; 366 stream->select = s; 367 *fd = s->fake_fd; 368 369 err = uv_thread_create(&s->thread, uv__stream_osx_select, stream); 370 if (err != 0) 371 goto failed_thread_create; 372 373 return 0; 374 375 failed_thread_create: 376 s->stream = NULL; 377 stream->select = NULL; 378 *fd = old_fd; 379 380 uv_sem_destroy(&s->async_sem); 381 382 failed_async_sem_init: 383 uv_sem_destroy(&s->close_sem); 384 385 failed_close_sem_init: 386 uv__close(fds[0]); 387 uv__close(fds[1]); 388 uv_close((uv_handle_t*) &s->async, uv__stream_osx_cb_close); 389 return err; 390 391 failed_async_init: 392 uv__free(s); 393 394 failed_malloc: 395 uv__close(fds[0]); 396 uv__close(fds[1]); 397 398 return err; 399 } 400 #endif /* defined(__APPLE__) */ 401 402 403 int uv__stream_open(uv_stream_t* stream, int fd, int flags) { 404 #if defined(__APPLE__) 405 int enable; 406 #endif 407 408 if (!(stream->io_watcher.fd == -1 || stream->io_watcher.fd == fd)) 409 return UV_EBUSY; 410 411 assert(fd >= 0); 412 stream->flags |= flags; 413 414 if (stream->type == UV_TCP) { 415 if ((stream->flags & UV_HANDLE_TCP_NODELAY) && uv__tcp_nodelay(fd, 1)) 416 return UV__ERR(errno); 417 418 /* TODO Use delay the user passed in. */ 419 if ((stream->flags & UV_HANDLE_TCP_KEEPALIVE) && 420 uv__tcp_keepalive(fd, 1, 60)) { 421 return UV__ERR(errno); 422 } 423 } 424 425 #if defined(__APPLE__) 426 enable = 1; 427 if (setsockopt(fd, SOL_SOCKET, SO_OOBINLINE, &enable, sizeof(enable)) && 428 errno != ENOTSOCK && 429 errno != EINVAL) { 430 return UV__ERR(errno); 431 } 432 #endif 433 434 stream->io_watcher.fd = fd; 435 436 return 0; 437 } 438 439 440 void uv__stream_flush_write_queue(uv_stream_t* stream, int error) { 441 uv_write_t* req; 442 struct uv__queue* q; 443 while (!uv__queue_empty(&stream->write_queue)) { 444 q = uv__queue_head(&stream->write_queue); 445 uv__queue_remove(q); 446 447 req = uv__queue_data(q, uv_write_t, queue); 448 req->error = error; 449 450 uv__queue_insert_tail(&stream->write_completed_queue, &req->queue); 451 } 452 } 453 454 455 void uv__stream_destroy(uv_stream_t* stream) { 456 assert(!uv__io_active(&stream->io_watcher, POLLIN | POLLOUT)); 457 assert(stream->flags & UV_HANDLE_CLOSED); 458 459 if (stream->connect_req) { 460 uv__req_unregister(stream->loop); 461 stream->connect_req->cb(stream->connect_req, UV_ECANCELED); 462 stream->connect_req = NULL; 463 } 464 465 uv__stream_flush_write_queue(stream, UV_ECANCELED); 466 uv__write_callbacks(stream); 467 uv__drain(stream); 468 469 assert(stream->write_queue_size == 0); 470 } 471 472 473 /* Implements a best effort approach to mitigating accept() EMFILE errors. 474 * We have a spare file descriptor stashed away that we close to get below 475 * the EMFILE limit. Next, we accept all pending connections and close them 476 * immediately to signal the clients that we're overloaded - and we are, but 477 * we still keep on trucking. 478 * 479 * There is one caveat: it's not reliable in a multi-threaded environment. 480 * The file descriptor limit is per process. Our party trick fails if another 481 * thread opens a file or creates a socket in the time window between us 482 * calling close() and accept(). 483 */ 484 static int uv__emfile_trick(uv_loop_t* loop, int accept_fd) { 485 int err; 486 int emfile_fd; 487 488 if (loop->emfile_fd == -1) 489 return UV_EMFILE; 490 491 uv__close(loop->emfile_fd); 492 loop->emfile_fd = -1; 493 494 do { 495 err = uv__accept(accept_fd); 496 if (err >= 0) 497 uv__close(err); 498 } while (err >= 0 || err == UV_EINTR); 499 500 emfile_fd = uv__open_cloexec("/", O_RDONLY); 501 if (emfile_fd >= 0) 502 loop->emfile_fd = emfile_fd; 503 504 return err; 505 } 506 507 508 void uv__server_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) { 509 uv_stream_t* stream; 510 int err; 511 int fd; 512 513 stream = container_of(w, uv_stream_t, io_watcher); 514 assert(events & POLLIN); 515 assert(stream->accepted_fd == -1); 516 assert(!(stream->flags & UV_HANDLE_CLOSING)); 517 518 fd = uv__stream_fd(stream); 519 err = uv__accept(fd); 520 521 if (err == UV_EMFILE || err == UV_ENFILE) 522 err = uv__emfile_trick(loop, fd); /* Shed load. */ 523 524 if (err < 0) 525 return; 526 527 stream->accepted_fd = err; 528 stream->connection_cb(stream, 0); 529 530 if (stream->accepted_fd != -1) 531 /* The user hasn't yet accepted called uv_accept() */ 532 uv__io_stop(loop, &stream->io_watcher, POLLIN); 533 } 534 535 536 int uv_accept(uv_stream_t* server, uv_stream_t* client) { 537 int err; 538 539 assert(server->loop == client->loop); 540 541 if (server->accepted_fd == -1) 542 return UV_EAGAIN; 543 544 switch (client->type) { 545 case UV_NAMED_PIPE: 546 case UV_TCP: 547 err = uv__stream_open(client, 548 server->accepted_fd, 549 UV_HANDLE_READABLE | UV_HANDLE_WRITABLE); 550 if (err) { 551 /* TODO handle error */ 552 uv__close(server->accepted_fd); 553 goto done; 554 } 555 break; 556 557 case UV_UDP: 558 err = uv_udp_open((uv_udp_t*) client, server->accepted_fd); 559 if (err) { 560 uv__close(server->accepted_fd); 561 goto done; 562 } 563 break; 564 565 default: 566 return UV_EINVAL; 567 } 568 569 client->flags |= UV_HANDLE_BOUND; 570 571 done: 572 /* Process queued fds */ 573 if (server->queued_fds != NULL) { 574 uv__stream_queued_fds_t* queued_fds; 575 576 queued_fds = server->queued_fds; 577 578 /* Read first */ 579 server->accepted_fd = queued_fds->fds[0]; 580 581 /* All read, free */ 582 assert(queued_fds->offset > 0); 583 if (--queued_fds->offset == 0) { 584 uv__free(queued_fds); 585 server->queued_fds = NULL; 586 } else { 587 /* Shift rest */ 588 memmove(queued_fds->fds, 589 queued_fds->fds + 1, 590 queued_fds->offset * sizeof(*queued_fds->fds)); 591 } 592 } else { 593 server->accepted_fd = -1; 594 if (err == 0) 595 uv__io_start(server->loop, &server->io_watcher, POLLIN); 596 } 597 return err; 598 } 599 600 601 int uv_listen(uv_stream_t* stream, int backlog, uv_connection_cb cb) { 602 int err; 603 if (uv__is_closing(stream)) { 604 return UV_EINVAL; 605 } 606 switch (stream->type) { 607 case UV_TCP: 608 err = uv__tcp_listen((uv_tcp_t*)stream, backlog, cb); 609 break; 610 611 case UV_NAMED_PIPE: 612 err = uv__pipe_listen((uv_pipe_t*)stream, backlog, cb); 613 break; 614 615 default: 616 err = UV_EINVAL; 617 } 618 619 if (err == 0) 620 uv__handle_start(stream); 621 622 return err; 623 } 624 625 626 static void uv__drain(uv_stream_t* stream) { 627 uv_shutdown_t* req; 628 int err; 629 630 assert(uv__queue_empty(&stream->write_queue)); 631 if (!(stream->flags & UV_HANDLE_CLOSING)) { 632 uv__io_stop(stream->loop, &stream->io_watcher, POLLOUT); 633 uv__stream_osx_interrupt_select(stream); 634 } 635 636 if (!uv__is_stream_shutting(stream)) 637 return; 638 639 req = stream->shutdown_req; 640 assert(req); 641 642 if ((stream->flags & UV_HANDLE_CLOSING) || 643 !(stream->flags & UV_HANDLE_SHUT)) { 644 stream->shutdown_req = NULL; 645 uv__req_unregister(stream->loop); 646 647 err = 0; 648 if (stream->flags & UV_HANDLE_CLOSING) 649 /* The user destroyed the stream before we got to do the shutdown. */ 650 err = UV_ECANCELED; 651 else if (shutdown(uv__stream_fd(stream), SHUT_WR)) 652 err = UV__ERR(errno); 653 else /* Success. */ 654 stream->flags |= UV_HANDLE_SHUT; 655 656 if (req->cb != NULL) 657 req->cb(req, err); 658 } 659 } 660 661 662 static ssize_t uv__writev(int fd, struct iovec* vec, size_t n) { 663 if (n == 1) 664 return write(fd, vec->iov_base, vec->iov_len); 665 else 666 return writev(fd, vec, n); 667 } 668 669 670 static size_t uv__write_req_size(uv_write_t* req) { 671 size_t size; 672 673 assert(req->bufs != NULL); 674 size = uv__count_bufs(req->bufs + req->write_index, 675 req->nbufs - req->write_index); 676 assert(req->handle->write_queue_size >= size); 677 678 return size; 679 } 680 681 682 /* Returns 1 if all write request data has been written, or 0 if there is still 683 * more data to write. 684 * 685 * Note: the return value only says something about the *current* request. 686 * There may still be other write requests sitting in the queue. 687 */ 688 static int uv__write_req_update(uv_stream_t* stream, 689 uv_write_t* req, 690 size_t n) { 691 uv_buf_t* buf; 692 size_t len; 693 694 assert(n <= stream->write_queue_size); 695 stream->write_queue_size -= n; 696 697 buf = req->bufs + req->write_index; 698 699 do { 700 len = n < buf->len ? n : buf->len; 701 if (buf->len != 0) 702 buf->base += len; 703 buf->len -= len; 704 buf += (buf->len == 0); /* Advance to next buffer if this one is empty. */ 705 n -= len; 706 } while (n > 0); 707 708 req->write_index = buf - req->bufs; 709 710 return req->write_index == req->nbufs; 711 } 712 713 714 static void uv__write_req_finish(uv_write_t* req) { 715 uv_stream_t* stream = req->handle; 716 717 /* Pop the req off tcp->write_queue. */ 718 uv__queue_remove(&req->queue); 719 720 /* Only free when there was no error. On error, we touch up write_queue_size 721 * right before making the callback. The reason we don't do that right away 722 * is that a write_queue_size > 0 is our only way to signal to the user that 723 * they should stop writing - which they should if we got an error. Something 724 * to revisit in future revisions of the libuv API. 725 */ 726 if (req->error == 0) { 727 if (req->bufs != req->bufsml) 728 uv__free(req->bufs); 729 req->bufs = NULL; 730 } 731 732 /* Add it to the write_completed_queue where it will have its 733 * callback called in the near future. 734 */ 735 uv__queue_insert_tail(&stream->write_completed_queue, &req->queue); 736 uv__io_feed(stream->loop, &stream->io_watcher); 737 } 738 739 740 static int uv__handle_fd(uv_handle_t* handle) { 741 switch (handle->type) { 742 case UV_NAMED_PIPE: 743 case UV_TCP: 744 return ((uv_stream_t*) handle)->io_watcher.fd; 745 746 case UV_UDP: 747 return ((uv_udp_t*) handle)->io_watcher.fd; 748 749 default: 750 return -1; 751 } 752 } 753 754 static int uv__try_write(uv_stream_t* stream, 755 const uv_buf_t bufs[], 756 unsigned int nbufs, 757 uv_stream_t* send_handle) { 758 struct iovec* iov; 759 int iovmax; 760 int iovcnt; 761 ssize_t n; 762 763 /* 764 * Cast to iovec. We had to have our own uv_buf_t instead of iovec 765 * because Windows's WSABUF is not an iovec. 766 */ 767 iov = (struct iovec*) bufs; 768 iovcnt = nbufs; 769 770 iovmax = uv__getiovmax(); 771 772 /* Limit iov count to avoid EINVALs from writev() */ 773 if (iovcnt > iovmax) 774 iovcnt = iovmax; 775 776 /* 777 * Now do the actual writev. Note that we've been updating the pointers 778 * inside the iov each time we write. So there is no need to offset it. 779 */ 780 if (send_handle != NULL) { 781 int fd_to_send; 782 struct msghdr msg; 783 union uv__cmsg cmsg; 784 785 if (uv__is_closing(send_handle)) 786 return UV_EBADF; 787 788 fd_to_send = uv__handle_fd((uv_handle_t*) send_handle); 789 790 memset(&cmsg, 0, sizeof(cmsg)); 791 792 assert(fd_to_send >= 0); 793 794 msg.msg_name = NULL; 795 msg.msg_namelen = 0; 796 msg.msg_iov = iov; 797 msg.msg_iovlen = iovcnt; 798 msg.msg_flags = 0; 799 800 msg.msg_control = &cmsg.hdr; 801 msg.msg_controllen = CMSG_SPACE(sizeof(fd_to_send)); 802 803 cmsg.hdr.cmsg_level = SOL_SOCKET; 804 cmsg.hdr.cmsg_type = SCM_RIGHTS; 805 cmsg.hdr.cmsg_len = CMSG_LEN(sizeof(fd_to_send)); 806 memcpy(CMSG_DATA(&cmsg.hdr), &fd_to_send, sizeof(fd_to_send)); 807 808 do 809 n = sendmsg(uv__stream_fd(stream), &msg, 0); 810 while (n == -1 && errno == EINTR); 811 } else { 812 do 813 n = uv__writev(uv__stream_fd(stream), iov, iovcnt); 814 while (n == -1 && errno == EINTR); 815 } 816 817 if (n >= 0) 818 return n; 819 820 if (errno == EAGAIN || errno == EWOULDBLOCK || errno == ENOBUFS) 821 return UV_EAGAIN; 822 823 #ifdef __APPLE__ 824 /* macOS versions 10.10 and 10.15 - and presumbaly 10.11 to 10.14, too - 825 * have a bug where a race condition causes the kernel to return EPROTOTYPE 826 * because the socket isn't fully constructed. It's probably the result of 827 * the peer closing the connection and that is why libuv translates it to 828 * ECONNRESET. Previously, libuv retried until the EPROTOTYPE error went 829 * away but some VPN software causes the same behavior except the error is 830 * permanent, not transient, turning the retry mechanism into an infinite 831 * loop. See https://github.com/libuv/libuv/pull/482. 832 */ 833 if (errno == EPROTOTYPE) 834 return UV_ECONNRESET; 835 #endif /* __APPLE__ */ 836 837 return UV__ERR(errno); 838 } 839 840 static void uv__write(uv_stream_t* stream) { 841 struct uv__queue* q; 842 uv_write_t* req; 843 ssize_t n; 844 int count; 845 846 assert(uv__stream_fd(stream) >= 0); 847 848 /* Prevent loop starvation when the consumer of this stream read as fast as 849 * (or faster than) we can write it. This `count` mechanism does not need to 850 * change even if we switch to edge-triggered I/O. 851 */ 852 count = 32; 853 854 for (;;) { 855 if (uv__queue_empty(&stream->write_queue)) 856 return; 857 858 q = uv__queue_head(&stream->write_queue); 859 req = uv__queue_data(q, uv_write_t, queue); 860 assert(req->handle == stream); 861 862 n = uv__try_write(stream, 863 &(req->bufs[req->write_index]), 864 req->nbufs - req->write_index, 865 req->send_handle); 866 867 /* Ensure the handle isn't sent again in case this is a partial write. */ 868 if (n >= 0) { 869 req->send_handle = NULL; 870 if (uv__write_req_update(stream, req, n)) { 871 uv__write_req_finish(req); 872 if (count-- > 0) 873 continue; /* Start trying to write the next request. */ 874 875 return; 876 } 877 } else if (n != UV_EAGAIN) 878 goto error; 879 880 /* If this is a blocking stream, try again. */ 881 if (stream->flags & UV_HANDLE_BLOCKING_WRITES) 882 continue; 883 884 /* We're not done. */ 885 uv__io_start(stream->loop, &stream->io_watcher, POLLOUT); 886 887 /* Notify select() thread about state change */ 888 uv__stream_osx_interrupt_select(stream); 889 890 return; 891 } 892 893 error: 894 req->error = n; 895 uv__write_req_finish(req); 896 uv__io_stop(stream->loop, &stream->io_watcher, POLLOUT); 897 uv__stream_osx_interrupt_select(stream); 898 } 899 900 901 static void uv__write_callbacks(uv_stream_t* stream) { 902 uv_write_t* req; 903 struct uv__queue* q; 904 struct uv__queue pq; 905 906 if (uv__queue_empty(&stream->write_completed_queue)) 907 return; 908 909 uv__queue_move(&stream->write_completed_queue, &pq); 910 911 while (!uv__queue_empty(&pq)) { 912 /* Pop a req off write_completed_queue. */ 913 q = uv__queue_head(&pq); 914 req = uv__queue_data(q, uv_write_t, queue); 915 uv__queue_remove(q); 916 uv__req_unregister(stream->loop); 917 918 if (req->bufs != NULL) { 919 stream->write_queue_size -= uv__write_req_size(req); 920 if (req->bufs != req->bufsml) 921 uv__free(req->bufs); 922 req->bufs = NULL; 923 } 924 925 /* NOTE: call callback AFTER freeing the request data. */ 926 if (req->cb) 927 req->cb(req, req->error); 928 } 929 } 930 931 932 static void uv__stream_eof(uv_stream_t* stream, const uv_buf_t* buf) { 933 stream->flags |= UV_HANDLE_READ_EOF; 934 stream->flags &= ~UV_HANDLE_READING; 935 uv__io_stop(stream->loop, &stream->io_watcher, POLLIN); 936 uv__handle_stop(stream); 937 uv__stream_osx_interrupt_select(stream); 938 stream->read_cb(stream, UV_EOF, buf); 939 } 940 941 942 static int uv__stream_queue_fd(uv_stream_t* stream, int fd) { 943 uv__stream_queued_fds_t* queued_fds; 944 unsigned int queue_size; 945 946 queued_fds = stream->queued_fds; 947 if (queued_fds == NULL) { 948 queue_size = 8; 949 queued_fds = uv__malloc((queue_size - 1) * sizeof(*queued_fds->fds) + 950 sizeof(*queued_fds)); 951 if (queued_fds == NULL) 952 return UV_ENOMEM; 953 queued_fds->size = queue_size; 954 queued_fds->offset = 0; 955 stream->queued_fds = queued_fds; 956 957 /* Grow */ 958 } else if (queued_fds->size == queued_fds->offset) { 959 queue_size = queued_fds->size + 8; 960 queued_fds = uv__realloc(queued_fds, 961 (queue_size - 1) * sizeof(*queued_fds->fds) + 962 sizeof(*queued_fds)); 963 964 /* 965 * Allocation failure, report back. 966 * NOTE: if it is fatal - sockets will be closed in uv__stream_close 967 */ 968 if (queued_fds == NULL) 969 return UV_ENOMEM; 970 queued_fds->size = queue_size; 971 stream->queued_fds = queued_fds; 972 } 973 974 /* Put fd in a queue */ 975 queued_fds->fds[queued_fds->offset++] = fd; 976 977 return 0; 978 } 979 980 981 static int uv__stream_recv_cmsg(uv_stream_t* stream, struct msghdr* msg) { 982 struct cmsghdr* cmsg; 983 char* p; 984 char* pe; 985 int fd; 986 int err; 987 size_t count; 988 989 err = 0; 990 for (cmsg = CMSG_FIRSTHDR(msg); cmsg != NULL; cmsg = CMSG_NXTHDR(msg, cmsg)) { 991 if (cmsg->cmsg_type != SCM_RIGHTS) { 992 fprintf(stderr, "ignoring non-SCM_RIGHTS ancillary data: %d\n", 993 cmsg->cmsg_type); 994 continue; 995 } 996 997 assert(cmsg->cmsg_len >= CMSG_LEN(0)); 998 count = cmsg->cmsg_len - CMSG_LEN(0); 999 assert(count % sizeof(fd) == 0); 1000 count /= sizeof(fd); 1001 1002 p = (void*) CMSG_DATA(cmsg); 1003 pe = p + count * sizeof(fd); 1004 1005 while (p < pe) { 1006 memcpy(&fd, p, sizeof(fd)); 1007 p += sizeof(fd); 1008 1009 if (err == 0) { 1010 if (stream->accepted_fd == -1) 1011 stream->accepted_fd = fd; 1012 else 1013 err = uv__stream_queue_fd(stream, fd); 1014 } 1015 1016 if (err != 0) 1017 uv__close(fd); 1018 } 1019 } 1020 1021 return err; 1022 } 1023 1024 1025 static void uv__read(uv_stream_t* stream) { 1026 uv_buf_t buf; 1027 ssize_t nread; 1028 struct msghdr msg; 1029 union uv__cmsg cmsg; 1030 int count; 1031 int err; 1032 int is_ipc; 1033 1034 stream->flags &= ~UV_HANDLE_READ_PARTIAL; 1035 1036 /* Prevent loop starvation when the data comes in as fast as (or faster than) 1037 * we can read it. XXX Need to rearm fd if we switch to edge-triggered I/O. 1038 */ 1039 count = 32; 1040 1041 is_ipc = stream->type == UV_NAMED_PIPE && ((uv_pipe_t*) stream)->ipc; 1042 1043 /* XXX: Maybe instead of having UV_HANDLE_READING we just test if 1044 * tcp->read_cb is NULL or not? 1045 */ 1046 while (stream->read_cb 1047 && (stream->flags & UV_HANDLE_READING) 1048 && (count-- > 0)) { 1049 assert(stream->alloc_cb != NULL); 1050 1051 buf = uv_buf_init(NULL, 0); 1052 stream->alloc_cb((uv_handle_t*)stream, 64 * 1024, &buf); 1053 if (buf.base == NULL || buf.len == 0) { 1054 /* User indicates it can't or won't handle the read. */ 1055 stream->read_cb(stream, UV_ENOBUFS, &buf); 1056 return; 1057 } 1058 1059 assert(buf.base != NULL); 1060 assert(uv__stream_fd(stream) >= 0); 1061 1062 if (!is_ipc) { 1063 do { 1064 nread = read(uv__stream_fd(stream), buf.base, buf.len); 1065 } 1066 while (nread < 0 && errno == EINTR); 1067 } else { 1068 /* ipc uses recvmsg */ 1069 msg.msg_flags = 0; 1070 msg.msg_iov = (struct iovec*) &buf; 1071 msg.msg_iovlen = 1; 1072 msg.msg_name = NULL; 1073 msg.msg_namelen = 0; 1074 /* Set up to receive a descriptor even if one isn't in the message */ 1075 msg.msg_controllen = sizeof(cmsg); 1076 msg.msg_control = &cmsg.hdr; 1077 1078 do { 1079 nread = uv__recvmsg(uv__stream_fd(stream), &msg, 0); 1080 } 1081 while (nread < 0 && errno == EINTR); 1082 } 1083 1084 if (nread < 0) { 1085 /* Error */ 1086 if (errno == EAGAIN || errno == EWOULDBLOCK) { 1087 /* Wait for the next one. */ 1088 if (stream->flags & UV_HANDLE_READING) { 1089 uv__io_start(stream->loop, &stream->io_watcher, POLLIN); 1090 uv__stream_osx_interrupt_select(stream); 1091 } 1092 stream->read_cb(stream, 0, &buf); 1093 #if defined(__CYGWIN__) || defined(__MSYS__) 1094 } else if (errno == ECONNRESET && stream->type == UV_NAMED_PIPE) { 1095 uv__stream_eof(stream, &buf); 1096 return; 1097 #endif 1098 } else { 1099 /* Error. User should call uv_close(). */ 1100 stream->flags &= ~(UV_HANDLE_READABLE | UV_HANDLE_WRITABLE); 1101 stream->read_cb(stream, UV__ERR(errno), &buf); 1102 if (stream->flags & UV_HANDLE_READING) { 1103 stream->flags &= ~UV_HANDLE_READING; 1104 uv__io_stop(stream->loop, &stream->io_watcher, POLLIN); 1105 uv__handle_stop(stream); 1106 uv__stream_osx_interrupt_select(stream); 1107 } 1108 } 1109 return; 1110 } else if (nread == 0) { 1111 uv__stream_eof(stream, &buf); 1112 return; 1113 } else { 1114 /* Successful read */ 1115 ssize_t buflen = buf.len; 1116 1117 if (is_ipc) { 1118 err = uv__stream_recv_cmsg(stream, &msg); 1119 if (err != 0) { 1120 stream->read_cb(stream, err, &buf); 1121 return; 1122 } 1123 } 1124 1125 #if defined(__MVS__) 1126 if (is_ipc && msg.msg_controllen > 0) { 1127 uv_buf_t blankbuf; 1128 int nread; 1129 struct iovec *old; 1130 1131 blankbuf.base = 0; 1132 blankbuf.len = 0; 1133 old = msg.msg_iov; 1134 msg.msg_iov = (struct iovec*) &blankbuf; 1135 nread = 0; 1136 do { 1137 nread = uv__recvmsg(uv__stream_fd(stream), &msg, 0); 1138 err = uv__stream_recv_cmsg(stream, &msg); 1139 if (err != 0) { 1140 stream->read_cb(stream, err, &buf); 1141 msg.msg_iov = old; 1142 return; 1143 } 1144 } while (nread == 0 && msg.msg_controllen > 0); 1145 msg.msg_iov = old; 1146 } 1147 #endif 1148 stream->read_cb(stream, nread, &buf); 1149 1150 /* Return if we didn't fill the buffer, there is no more data to read. */ 1151 if (nread < buflen) { 1152 stream->flags |= UV_HANDLE_READ_PARTIAL; 1153 return; 1154 } 1155 } 1156 } 1157 } 1158 1159 1160 int uv_shutdown(uv_shutdown_t* req, uv_stream_t* stream, uv_shutdown_cb cb) { 1161 assert(stream->type == UV_TCP || 1162 stream->type == UV_TTY || 1163 stream->type == UV_NAMED_PIPE); 1164 1165 if (!(stream->flags & UV_HANDLE_WRITABLE) || 1166 stream->flags & UV_HANDLE_SHUT || 1167 uv__is_stream_shutting(stream) || 1168 uv__is_closing(stream)) { 1169 return UV_ENOTCONN; 1170 } 1171 1172 assert(uv__stream_fd(stream) >= 0); 1173 1174 /* Initialize request. The `shutdown(2)` call will always be deferred until 1175 * `uv__drain`, just before the callback is run. */ 1176 uv__req_init(stream->loop, req, UV_SHUTDOWN); 1177 req->handle = stream; 1178 req->cb = cb; 1179 stream->shutdown_req = req; 1180 stream->flags &= ~UV_HANDLE_WRITABLE; 1181 1182 if (uv__queue_empty(&stream->write_queue)) 1183 uv__io_feed(stream->loop, &stream->io_watcher); 1184 1185 return 0; 1186 } 1187 1188 1189 static void uv__stream_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) { 1190 uv_stream_t* stream; 1191 1192 stream = container_of(w, uv_stream_t, io_watcher); 1193 1194 assert(stream->type == UV_TCP || 1195 stream->type == UV_NAMED_PIPE || 1196 stream->type == UV_TTY); 1197 assert(!(stream->flags & UV_HANDLE_CLOSING)); 1198 1199 if (stream->connect_req) { 1200 uv__stream_connect(stream); 1201 return; 1202 } 1203 1204 assert(uv__stream_fd(stream) >= 0); 1205 1206 /* Ignore POLLHUP here. Even if it's set, there may still be data to read. */ 1207 if (events & (POLLIN | POLLERR | POLLHUP)) 1208 uv__read(stream); 1209 1210 if (uv__stream_fd(stream) == -1) 1211 return; /* read_cb closed stream. */ 1212 1213 /* Short-circuit iff POLLHUP is set, the user is still interested in read 1214 * events and uv__read() reported a partial read but not EOF. If the EOF 1215 * flag is set, uv__read() called read_cb with err=UV_EOF and we don't 1216 * have to do anything. If the partial read flag is not set, we can't 1217 * report the EOF yet because there is still data to read. 1218 */ 1219 if ((events & POLLHUP) && 1220 (stream->flags & UV_HANDLE_READING) && 1221 (stream->flags & UV_HANDLE_READ_PARTIAL) && 1222 !(stream->flags & UV_HANDLE_READ_EOF)) { 1223 uv_buf_t buf = { NULL, 0 }; 1224 uv__stream_eof(stream, &buf); 1225 } 1226 1227 if (uv__stream_fd(stream) == -1) 1228 return; /* read_cb closed stream. */ 1229 1230 if (events & (POLLOUT | POLLERR | POLLHUP)) { 1231 uv__write(stream); 1232 uv__write_callbacks(stream); 1233 1234 /* Write queue drained. */ 1235 if (uv__queue_empty(&stream->write_queue)) 1236 uv__drain(stream); 1237 } 1238 } 1239 1240 1241 /** 1242 * We get called here from directly following a call to connect(2). 1243 * In order to determine if we've errored out or succeeded must call 1244 * getsockopt. 1245 */ 1246 static void uv__stream_connect(uv_stream_t* stream) { 1247 int error; 1248 uv_connect_t* req = stream->connect_req; 1249 socklen_t errorsize = sizeof(int); 1250 1251 assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE); 1252 assert(req); 1253 1254 if (stream->delayed_error) { 1255 /* To smooth over the differences between unixes errors that 1256 * were reported synchronously on the first connect can be delayed 1257 * until the next tick--which is now. 1258 */ 1259 error = stream->delayed_error; 1260 stream->delayed_error = 0; 1261 } else { 1262 /* Normal situation: we need to get the socket error from the kernel. */ 1263 assert(uv__stream_fd(stream) >= 0); 1264 getsockopt(uv__stream_fd(stream), 1265 SOL_SOCKET, 1266 SO_ERROR, 1267 &error, 1268 &errorsize); 1269 error = UV__ERR(error); 1270 } 1271 1272 if (error == UV__ERR(EINPROGRESS)) 1273 return; 1274 1275 stream->connect_req = NULL; 1276 uv__req_unregister(stream->loop); 1277 1278 if (error < 0 || uv__queue_empty(&stream->write_queue)) { 1279 uv__io_stop(stream->loop, &stream->io_watcher, POLLOUT); 1280 } 1281 1282 if (req->cb) 1283 req->cb(req, error); 1284 1285 if (uv__stream_fd(stream) == -1) 1286 return; 1287 1288 if (error < 0) { 1289 uv__stream_flush_write_queue(stream, UV_ECANCELED); 1290 uv__write_callbacks(stream); 1291 } 1292 } 1293 1294 1295 static int uv__check_before_write(uv_stream_t* stream, 1296 unsigned int nbufs, 1297 uv_stream_t* send_handle) { 1298 assert(nbufs > 0); 1299 assert((stream->type == UV_TCP || 1300 stream->type == UV_NAMED_PIPE || 1301 stream->type == UV_TTY) && 1302 "uv_write (unix) does not yet support other types of streams"); 1303 1304 if (uv__stream_fd(stream) < 0) 1305 return UV_EBADF; 1306 1307 if (!(stream->flags & UV_HANDLE_WRITABLE)) 1308 return UV_EPIPE; 1309 1310 if (send_handle != NULL) { 1311 if (stream->type != UV_NAMED_PIPE || !((uv_pipe_t*)stream)->ipc) 1312 return UV_EINVAL; 1313 1314 /* XXX We abuse uv_write2() to send over UDP handles to child processes. 1315 * Don't call uv__stream_fd() on those handles, it's a macro that on OS X 1316 * evaluates to a function that operates on a uv_stream_t with a couple of 1317 * OS X specific fields. On other Unices it does (handle)->io_watcher.fd, 1318 * which works but only by accident. 1319 */ 1320 if (uv__handle_fd((uv_handle_t*) send_handle) < 0) 1321 return UV_EBADF; 1322 1323 #if defined(__CYGWIN__) || defined(__MSYS__) 1324 /* Cygwin recvmsg always sets msg_controllen to zero, so we cannot send it. 1325 See https://github.com/mirror/newlib-cygwin/blob/86fc4bf0/winsup/cygwin/fhandler_socket.cc#L1736-L1743 */ 1326 return UV_ENOSYS; 1327 #endif 1328 } 1329 1330 return 0; 1331 } 1332 1333 int uv_write2(uv_write_t* req, 1334 uv_stream_t* stream, 1335 const uv_buf_t bufs[], 1336 unsigned int nbufs, 1337 uv_stream_t* send_handle, 1338 uv_write_cb cb) { 1339 int empty_queue; 1340 int err; 1341 1342 err = uv__check_before_write(stream, nbufs, send_handle); 1343 if (err < 0) 1344 return err; 1345 1346 /* It's legal for write_queue_size > 0 even when the write_queue is empty; 1347 * it means there are error-state requests in the write_completed_queue that 1348 * will touch up write_queue_size later, see also uv__write_req_finish(). 1349 * We could check that write_queue is empty instead but that implies making 1350 * a write() syscall when we know that the handle is in error mode. 1351 */ 1352 empty_queue = (stream->write_queue_size == 0); 1353 1354 /* Initialize the req */ 1355 uv__req_init(stream->loop, req, UV_WRITE); 1356 req->cb = cb; 1357 req->handle = stream; 1358 req->error = 0; 1359 req->send_handle = send_handle; 1360 uv__queue_init(&req->queue); 1361 1362 req->bufs = req->bufsml; 1363 if (nbufs > ARRAY_SIZE(req->bufsml)) 1364 req->bufs = uv__malloc(nbufs * sizeof(bufs[0])); 1365 1366 if (req->bufs == NULL) 1367 return UV_ENOMEM; 1368 1369 memcpy(req->bufs, bufs, nbufs * sizeof(bufs[0])); 1370 req->nbufs = nbufs; 1371 req->write_index = 0; 1372 stream->write_queue_size += uv__count_bufs(bufs, nbufs); 1373 1374 /* Append the request to write_queue. */ 1375 uv__queue_insert_tail(&stream->write_queue, &req->queue); 1376 1377 /* If the queue was empty when this function began, we should attempt to 1378 * do the write immediately. Otherwise start the write_watcher and wait 1379 * for the fd to become writable. 1380 */ 1381 if (stream->connect_req) { 1382 /* Still connecting, do nothing. */ 1383 } 1384 else if (empty_queue) { 1385 uv__write(stream); 1386 } 1387 else { 1388 /* 1389 * blocking streams should never have anything in the queue. 1390 * if this assert fires then somehow the blocking stream isn't being 1391 * sufficiently flushed in uv__write. 1392 */ 1393 assert(!(stream->flags & UV_HANDLE_BLOCKING_WRITES)); 1394 uv__io_start(stream->loop, &stream->io_watcher, POLLOUT); 1395 uv__stream_osx_interrupt_select(stream); 1396 } 1397 1398 return 0; 1399 } 1400 1401 1402 /* The buffers to be written must remain valid until the callback is called. 1403 * This is not required for the uv_buf_t array. 1404 */ 1405 int uv_write(uv_write_t* req, 1406 uv_stream_t* handle, 1407 const uv_buf_t bufs[], 1408 unsigned int nbufs, 1409 uv_write_cb cb) { 1410 return uv_write2(req, handle, bufs, nbufs, NULL, cb); 1411 } 1412 1413 1414 int uv_try_write(uv_stream_t* stream, 1415 const uv_buf_t bufs[], 1416 unsigned int nbufs) { 1417 return uv_try_write2(stream, bufs, nbufs, NULL); 1418 } 1419 1420 1421 int uv_try_write2(uv_stream_t* stream, 1422 const uv_buf_t bufs[], 1423 unsigned int nbufs, 1424 uv_stream_t* send_handle) { 1425 int err; 1426 1427 /* Connecting or already writing some data */ 1428 if (stream->connect_req != NULL || stream->write_queue_size != 0) 1429 return UV_EAGAIN; 1430 1431 err = uv__check_before_write(stream, nbufs, NULL); 1432 if (err < 0) 1433 return err; 1434 1435 return uv__try_write(stream, bufs, nbufs, send_handle); 1436 } 1437 1438 1439 int uv__read_start(uv_stream_t* stream, 1440 uv_alloc_cb alloc_cb, 1441 uv_read_cb read_cb) { 1442 assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE || 1443 stream->type == UV_TTY); 1444 1445 /* The UV_HANDLE_READING flag is irrelevant of the state of the stream - it 1446 * just expresses the desired state of the user. */ 1447 stream->flags |= UV_HANDLE_READING; 1448 stream->flags &= ~UV_HANDLE_READ_EOF; 1449 1450 /* TODO: try to do the read inline? */ 1451 assert(uv__stream_fd(stream) >= 0); 1452 assert(alloc_cb); 1453 1454 stream->read_cb = read_cb; 1455 stream->alloc_cb = alloc_cb; 1456 1457 uv__io_start(stream->loop, &stream->io_watcher, POLLIN); 1458 uv__handle_start(stream); 1459 uv__stream_osx_interrupt_select(stream); 1460 1461 return 0; 1462 } 1463 1464 1465 int uv_read_stop(uv_stream_t* stream) { 1466 if (!(stream->flags & UV_HANDLE_READING)) 1467 return 0; 1468 1469 stream->flags &= ~UV_HANDLE_READING; 1470 uv__io_stop(stream->loop, &stream->io_watcher, POLLIN); 1471 uv__handle_stop(stream); 1472 uv__stream_osx_interrupt_select(stream); 1473 1474 stream->read_cb = NULL; 1475 stream->alloc_cb = NULL; 1476 return 0; 1477 } 1478 1479 1480 int uv_is_readable(const uv_stream_t* stream) { 1481 return !!(stream->flags & UV_HANDLE_READABLE); 1482 } 1483 1484 1485 int uv_is_writable(const uv_stream_t* stream) { 1486 return !!(stream->flags & UV_HANDLE_WRITABLE); 1487 } 1488 1489 1490 #if defined(__APPLE__) 1491 int uv___stream_fd(const uv_stream_t* handle) { 1492 const uv__stream_select_t* s; 1493 1494 assert(handle->type == UV_TCP || 1495 handle->type == UV_TTY || 1496 handle->type == UV_NAMED_PIPE); 1497 1498 s = handle->select; 1499 if (s != NULL) 1500 return s->fd; 1501 1502 return handle->io_watcher.fd; 1503 } 1504 #endif /* defined(__APPLE__) */ 1505 1506 1507 void uv__stream_close(uv_stream_t* handle) { 1508 unsigned int i; 1509 uv__stream_queued_fds_t* queued_fds; 1510 1511 #if defined(__APPLE__) 1512 /* Terminate select loop first */ 1513 if (handle->select != NULL) { 1514 uv__stream_select_t* s; 1515 1516 s = handle->select; 1517 1518 uv_sem_post(&s->close_sem); 1519 uv_sem_post(&s->async_sem); 1520 uv__stream_osx_interrupt_select(handle); 1521 uv_thread_join(&s->thread); 1522 uv_sem_destroy(&s->close_sem); 1523 uv_sem_destroy(&s->async_sem); 1524 uv__close(s->fake_fd); 1525 uv__close(s->int_fd); 1526 uv_close((uv_handle_t*) &s->async, uv__stream_osx_cb_close); 1527 1528 handle->select = NULL; 1529 } 1530 #endif /* defined(__APPLE__) */ 1531 1532 uv__io_close(handle->loop, &handle->io_watcher); 1533 uv_read_stop(handle); 1534 uv__handle_stop(handle); 1535 handle->flags &= ~(UV_HANDLE_READABLE | UV_HANDLE_WRITABLE); 1536 1537 if (handle->io_watcher.fd != -1) { 1538 /* Don't close stdio file descriptors. Nothing good comes from it. */ 1539 if (handle->io_watcher.fd > STDERR_FILENO) 1540 uv__close(handle->io_watcher.fd); 1541 handle->io_watcher.fd = -1; 1542 } 1543 1544 if (handle->accepted_fd != -1) { 1545 uv__close(handle->accepted_fd); 1546 handle->accepted_fd = -1; 1547 } 1548 1549 /* Close all queued fds */ 1550 if (handle->queued_fds != NULL) { 1551 queued_fds = handle->queued_fds; 1552 for (i = 0; i < queued_fds->offset; i++) 1553 uv__close(queued_fds->fds[i]); 1554 uv__free(handle->queued_fds); 1555 handle->queued_fds = NULL; 1556 } 1557 1558 assert(!uv__io_active(&handle->io_watcher, POLLIN | POLLOUT)); 1559 } 1560 1561 1562 int uv_stream_set_blocking(uv_stream_t* handle, int blocking) { 1563 /* Don't need to check the file descriptor, uv__nonblock() 1564 * will fail with EBADF if it's not valid. 1565 */ 1566 return uv__nonblock(uv__stream_fd(handle), !blocking); 1567 } 1568
Indexes created Fri Jun 19 00:25:02 UTC 2026