#include "uv.h"
#include "internal.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/uio.h>
#include <sys/un.h>
#include <unistd.h>
#include <limits.h>

Macros
#define	RETRY_ON_WRITE_ERROR(errno) (errno == EINTR)

#define	IS_TRANSIENT_WRITE_ERROR(errno, send_handle) (errno == EAGAIN \|\| errno == EWOULDBLOCK \|\| errno == ENOBUFS)

#define	UV_DEC_BACKLOG(w) /* no-op */

#define	UV__CMSG_FD_COUNT 64

#define	UV__CMSG_FD_SIZE (UV__CMSG_FD_COUNT * sizeof(int))

Functions
static void	uv__stream_connect (uv_stream_t *)

static void	uv__write (uv_stream_t *stream)

static void	uv__read (uv_stream_t *stream)

static void	uv__stream_io (uv_loop_t loop, uv__io_t w, unsigned int events)

static void	uv__write_callbacks (uv_stream_t *stream)

static size_t	uv__write_req_size (uv_write_t *req)

void	uv__stream_init (uv_loop_t loop, uv_stream_t stream, uv_handle_type type)

static void	uv__stream_osx_interrupt_select (uv_stream_t *stream)

int	uv__stream_open (uv_stream_t *stream, int fd, int flags)

void	uv__stream_flush_write_queue (uv_stream_t *stream, int error)

void	uv__stream_destroy (uv_stream_t *stream)

static int	uv__emfile_trick (uv_loop_t *loop, int accept_fd)

void	uv__server_io (uv_loop_t loop, uv__io_t w, unsigned int events)

int	uv_accept (uv_stream_t server, uv_stream_t client)

int	uv_listen (uv_stream_t *stream, int backlog, uv_connection_cb cb)

static void	uv__drain (uv_stream_t *stream)

static ssize_t	uv__writev (int fd, struct iovec *vec, size_t n)

static int	uv__write_req_update (uv_stream_t stream, uv_write_t req, size_t n)

static void	uv__write_req_finish (uv_write_t *req)

static int	uv__handle_fd (uv_handle_t *handle)

uv_handle_type	uv__handle_type (int fd)

static void	uv__stream_eof (uv_stream_t stream, const uv_buf_t buf)

static int	uv__stream_queue_fd (uv_stream_t *stream, int fd)

static int	uv__stream_recv_cmsg (uv_stream_t stream, struct msghdr msg)

int	uv_shutdown (uv_shutdown_t req, uv_stream_t stream, uv_shutdown_cb cb)

int	uv_write2 (uv_write_t req, uv_stream_t stream, const uv_buf_t bufs[], unsigned int nbufs, uv_stream_t *send_handle, uv_write_cb cb)

int	uv_write (uv_write_t req, uv_stream_t handle, const uv_buf_t bufs[], unsigned int nbufs, uv_write_cb cb)

void	uv_try_write_cb (uv_write_t *req, int status)

int	uv_try_write (uv_stream_t *stream, const uv_buf_t bufs[], unsigned int nbufs)

int	uv_read_start (uv_stream_t *stream, uv_alloc_cb alloc_cb, uv_read_cb read_cb)

int	uv_read_stop (uv_stream_t *stream)

int	uv_is_readable (const uv_stream_t *stream)

int	uv_is_writable (const uv_stream_t *stream)

void	uv__stream_close (uv_stream_t *handle)

int	uv_stream_set_blocking (uv_stream_t *handle, int blocking)

Macro Definition Documentation

◆ IS_TRANSIENT_WRITE_ERROR

#define IS_TRANSIENT_WRITE_ERROR	(	errno,
		send_handle
	)	(errno == EAGAIN \|\| errno == EWOULDBLOCK \|\| errno == ENOBUFS)

Definition at line 73 of file stream.c.

◆ RETRY_ON_WRITE_ERROR

#define RETRY_ON_WRITE_ERROR ( errno ) (errno == EINTR)

Definition at line 72 of file stream.c.

◆ UV__CMSG_FD_COUNT

#define UV__CMSG_FD_COUNT 64

Definition at line 1055 of file stream.c.

◆ UV__CMSG_FD_SIZE

#define UV__CMSG_FD_SIZE (UV__CMSG_FD_COUNT * sizeof(int))

Definition at line 1057 of file stream.c.

◆ UV_DEC_BACKLOG

#define UV_DEC_BACKLOG ( w ) /* no-op */

Definition at line 524 of file stream.c.

Function Documentation

◆ uv__drain()

static void uv__drain ( uv_stream_t * stream )

static

Definition at line 679 of file stream.c.

                                            {
   uv_shutdown_t* req;
   int err;
  
   assert(QUEUE_EMPTY(&stream->write_queue));
   uv__io_stop(stream->loop, &stream->io_watcher, POLLOUT);
   uv__stream_osx_interrupt_select(stream);
  
   /* Shutdown? */
   if ((stream->flags & UV_HANDLE_SHUTTING) &&
       !(stream->flags & UV_HANDLE_CLOSING) &&
       !(stream->flags & UV_HANDLE_SHUT)) {
     assert(stream->shutdown_req);
  
     req = stream->shutdown_req;
     stream->shutdown_req = NULL;
     stream->flags &= ~UV_HANDLE_SHUTTING;
     uv__req_unregister(stream->loop, req);
  
     err = 0;
     if (shutdown(uv__stream_fd(stream), SHUT_WR))
       err = UV__ERR(errno);
  
     if (err == 0)
       stream->flags |= UV_HANDLE_SHUT;
  
     if (req->cb != NULL)
       req->cb(req, err);
   }
 }

References assert(), err, NULL, QUEUE_EMPTY, req, UV__ERR, uv__io_stop(), uv__req_unregister, uv__stream_fd, uv__stream_osx_interrupt_select(), UV_HANDLE_CLOSING, UV_HANDLE_SHUT, and UV_HANDLE_SHUTTING.

Referenced by uv__stream_io().

◆ uv__emfile_trick()

static int uv__emfile_trick	(	uv_loop_t *	loop,
		int	accept_fd
	)

static

Definition at line 497 of file stream.c.

                                                             {
   int err;
   int emfile_fd;
  
   if (loop->emfile_fd == -1)
     return UV_EMFILE;
  
   uv__close(loop->emfile_fd);
   loop->emfile_fd = -1;
  
   do {
     err = uv__accept(accept_fd);
     if (err >= 0)
       uv__close(err);
   } while (err >= 0 || err == UV_EINTR);
  
   emfile_fd = uv__open_cloexec("/", O_RDONLY);
   if (emfile_fd >= 0)
     loop->emfile_fd = emfile_fd;
  
   return err;
 }

References err, loop, O_RDONLY, uv__accept(), uv__close(), and uv__open_cloexec().

Referenced by uv__server_io().

◆ uv__handle_fd()

static int uv__handle_fd ( uv_handle_t * handle )

static

Definition at line 788 of file stream.c.

                                               {
   switch (handle->type) {
     case UV_NAMED_PIPE:
     case UV_TCP:
       return ((uv_stream_t*) handle)->io_watcher.fd;
  
     case UV_UDP:
       return ((uv_udp_t*) handle)->io_watcher.fd;
  
     default:
       return -1;
   }
 }

References handle.

Referenced by uv__write(), and uv_write2().

◆ uv__handle_type()

uv_handle_type uv__handle_type ( int fd )

Definition at line 958 of file stream.c.

                                        {
   struct sockaddr_storage ss;
   socklen_t sslen;
   socklen_t len;
   int type;
  
   memset(&ss, 0, sizeof(ss));
   sslen = sizeof(ss);
  
   if (getsockname(fd, (struct sockaddr*)&ss, &sslen))
     return UV_UNKNOWN_HANDLE;
  
   len = sizeof type;
  
   if (getsockopt(fd, SOL_SOCKET, SO_TYPE, &type, &len))
     return UV_UNKNOWN_HANDLE;
  
   if (type == SOCK_STREAM) {
 #if defined(_AIX) || defined(__DragonFly__)
     /* on AIX/DragonFly the getsockname call returns an empty sa structure
      * for sockets of type AF_UNIX.  For all other types it will
      * return a properly filled in structure.
      */
     if (sslen == 0)
       return UV_NAMED_PIPE;
 #endif
     switch (ss.ss_family) {
       case AF_UNIX:
         return UV_NAMED_PIPE;
       case AF_INET:
       case AF_INET6:
         return UV_TCP;
       }
   }
  
   if (type == SOCK_DGRAM &&
       (ss.ss_family == AF_INET || ss.ss_family == AF_INET6))
     return UV_UDP;
  
   return UV_UNKNOWN_HANDLE;
 }

References AF_INET, AF_INET6, AF_UNIX, fd, len, memset(), SO_TYPE, SOCK_DGRAM, SOCK_STREAM, SOL_SOCKET, type, and UV_UNKNOWN_HANDLE.

Referenced by uv_pipe_pending_type().

◆ uv__read()

static void uv__read ( uv_stream_t * stream )

static

Definition at line 1116 of file stream.c.

                                           {
   uv_buf_t buf;
   ssize_t nread;
   struct msghdr msg;
   char cmsg_space[CMSG_SPACE(UV__CMSG_FD_SIZE)];
   int count;
   int err;
   int is_ipc;
  
   stream->flags &= ~UV_HANDLE_READ_PARTIAL;
  
   /* Prevent loop starvation when the data comes in as fast as (or faster than)
    * we can read it. XXX Need to rearm fd if we switch to edge-triggered I/O.
    */
   count = 32;
  
   is_ipc = stream->type == UV_NAMED_PIPE && ((uv_pipe_t*) stream)->ipc;
  
   /* XXX: Maybe instead of having UV_HANDLE_READING we just test if
    * tcp->read_cb is NULL or not?
    */
   while (stream->read_cb
       && (stream->flags & UV_HANDLE_READING)
       && (count-- > 0)) {
     assert(stream->alloc_cb != NULL);
  
     buf = uv_buf_init(NULL, 0);
     stream->alloc_cb((uv_handle_t*)stream, 64 * 1024, &buf);
     if (buf.base == NULL || buf.len == 0) {
       /* User indicates it can't or won't handle the read. */
       stream->read_cb(stream, UV_ENOBUFS, &buf);
       return;
     }
  
     assert(buf.base != NULL);
     assert(uv__stream_fd(stream) >= 0);
  
     if (!is_ipc) {
       do {
         nread = read(uv__stream_fd(stream), buf.base, buf.len);
       }
       while (nread < 0 && errno == EINTR);
     } else {
       /* ipc uses recvmsg */
       msg.msg_flags = 0;
       msg.msg_iov = (struct iovec*) &buf;
       msg.msg_iovlen = 1;
       msg.msg_name = NULL;
       msg.msg_namelen = 0;
       /* Set up to receive a descriptor even if one isn't in the message */
       msg.msg_controllen = sizeof(cmsg_space);
       msg.msg_control = cmsg_space;
  
       do {
         nread = uv__recvmsg(uv__stream_fd(stream), &msg, 0);
       }
       while (nread < 0 && errno == EINTR);
     }
  
     if (nread < 0) {
       /* Error */
       if (errno == EAGAIN || errno == EWOULDBLOCK) {
         /* Wait for the next one. */
         if (stream->flags & UV_HANDLE_READING) {
           uv__io_start(stream->loop, &stream->io_watcher, POLLIN);
           uv__stream_osx_interrupt_select(stream);
         }
         stream->read_cb(stream, 0, &buf);
 #if defined(__CYGWIN__) || defined(__MSYS__)
       } else if (errno == ECONNRESET && stream->type == UV_NAMED_PIPE) {
         uv__stream_eof(stream, &buf);
         return;
 #endif
       } else {
         /* Error. User should call uv_close(). */
         stream->read_cb(stream, UV__ERR(errno), &buf);
         if (stream->flags & UV_HANDLE_READING) {
           stream->flags &= ~UV_HANDLE_READING;
           uv__io_stop(stream->loop, &stream->io_watcher, POLLIN);
           if (!uv__io_active(&stream->io_watcher, POLLOUT))
             uv__handle_stop(stream);
           uv__stream_osx_interrupt_select(stream);
         }
       }
       return;
     } else if (nread == 0) {
       uv__stream_eof(stream, &buf);
       return;
     } else {
       /* Successful read */
       ssize_t buflen = buf.len;
  
       if (is_ipc) {
         err = uv__stream_recv_cmsg(stream, &msg);
         if (err != 0) {
           stream->read_cb(stream, err, &buf);
           return;
         }
       }
  
 #if defined(__MVS__)
       if (is_ipc && msg.msg_controllen > 0) {
         uv_buf_t blankbuf;
         int nread;
         struct iovec *old;
  
         blankbuf.base = 0;
         blankbuf.len = 0;
         old = msg.msg_iov;
         msg.msg_iov = (struct iovec*) &blankbuf;
         nread = 0;
         do {
           nread = uv__recvmsg(uv__stream_fd(stream), &msg, 0);
           err = uv__stream_recv_cmsg(stream, &msg);
           if (err != 0) {
             stream->read_cb(stream, err, &buf);
             msg.msg_iov = old;
             return;
           }
         } while (nread == 0 && msg.msg_controllen > 0);
         msg.msg_iov = old;
       }
 #endif
       stream->read_cb(stream, nread, &buf);
  
       /* Return if we didn't fill the buffer, there is no more data to read. */
       if (nread < buflen) {
         stream->flags |= UV_HANDLE_READ_PARTIAL;
         return;
       }
     }
   }
 }

References assert(), uv_buf_t::base, buflen, CMSG_SPACE, count, EAGAIN, ECONNRESET, EINTR, err, uv_buf_t::len, msg, NULL, read(), UV__CMSG_FD_SIZE, UV__ERR, uv__handle_stop, uv__io_active(), uv__io_start(), uv__io_stop(), uv__recvmsg(), uv__stream_eof(), uv__stream_fd, uv__stream_osx_interrupt_select(), uv__stream_recv_cmsg(), uv_buf_init(), UV_HANDLE_READ_PARTIAL, and UV_HANDLE_READING.

Referenced by uv__stream_io().

◆ uv__server_io()

void uv__server_io	(	uv_loop_t *	loop,
		uv__io_t *	w,
		unsigned int	events
	)

Definition at line 528 of file stream.c.

                                                                       {
   uv_stream_t* stream;
   int err;
  
   stream = container_of(w, uv_stream_t, io_watcher);
   assert(events & POLLIN);
   assert(stream->accepted_fd == -1);
   assert(!(stream->flags & UV_HANDLE_CLOSING));
  
   uv__io_start(stream->loop, &stream->io_watcher, POLLIN);
  
   /* connection_cb can close the server socket while we're
    * in the loop so check it on each iteration.
    */
   while (uv__stream_fd(stream) != -1) {
     assert(stream->accepted_fd == -1);
  
 #if defined(UV_HAVE_KQUEUE)
     if (w->rcount <= 0)
       return;
 #endif /* defined(UV_HAVE_KQUEUE) */
  
     err = uv__accept(uv__stream_fd(stream));
     if (err < 0) {
       if (err == UV_EAGAIN || err == UV__ERR(EWOULDBLOCK))
         return;  /* Not an error. */
  
       if (err == UV_ECONNABORTED)
         continue;  /* Ignore. Nothing we can do about that. */
  
       if (err == UV_EMFILE || err == UV_ENFILE) {
         err = uv__emfile_trick(loop, uv__stream_fd(stream));
         if (err == UV_EAGAIN || err == UV__ERR(EWOULDBLOCK))
           break;
       }
  
       stream->connection_cb(stream, err);
       continue;
     }
  
     UV_DEC_BACKLOG(w)
     stream->accepted_fd = err;
     stream->connection_cb(stream, 0);
  
     if (stream->accepted_fd != -1) {
       /* The user hasn't yet accepted called uv_accept() */
       uv__io_stop(loop, &stream->io_watcher, POLLIN);
       return;
     }
  
     if (stream->type == UV_TCP &&
         (stream->flags & UV_HANDLE_TCP_SINGLE_ACCEPT)) {
       /* Give other processes a chance to accept connections. */
       struct timespec timeout = { 0, 1 };
       nanosleep(&timeout, NULL);
     }
   }
 }

References assert(), container_of, err, loop, nanosleep, NULL, timeout, uv__accept(), uv__emfile_trick(), UV__ERR, uv__io_start(), uv__io_stop(), uv__stream_fd, UV_DEC_BACKLOG, UV_HANDLE_CLOSING, UV_HANDLE_TCP_SINGLE_ACCEPT, and w.

Referenced by uv_pipe_listen(), and uv_tcp_listen().

◆ uv__stream_close()

void uv__stream_close ( uv_stream_t * handle )

Definition at line 1633 of file stream.c.

                                            {
   unsigned int i;
   uv__stream_queued_fds_t* queued_fds;
  
 #if defined(__APPLE__)
   /* Terminate select loop first */
   if (handle->select != NULL) {
     uv__stream_select_t* s;
  
     s = handle->select;
  
     uv_sem_post(&s->close_sem);
     uv_sem_post(&s->async_sem);
     uv__stream_osx_interrupt_select(handle);
     uv_thread_join(&s->thread);
     uv_sem_destroy(&s->close_sem);
     uv_sem_destroy(&s->async_sem);
     uv__close(s->fake_fd);
     uv__close(s->int_fd);
     uv_close((uv_handle_t*) &s->async, uv__stream_osx_cb_close);
  
     handle->select = NULL;
   }
 #endif /* defined(__APPLE__) */
  
   uv__io_close(handle->loop, &handle->io_watcher);
   uv_read_stop(handle);
   uv__handle_stop(handle);
   handle->flags &= ~(UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);
  
   if (handle->io_watcher.fd != -1) {
     /* Don't close stdio file descriptors.  Nothing good comes from it. */
     if (handle->io_watcher.fd > STDERR_FILENO)
       uv__close(handle->io_watcher.fd);
     handle->io_watcher.fd = -1;
   }
  
   if (handle->accepted_fd != -1) {
     uv__close(handle->accepted_fd);
     handle->accepted_fd = -1;
   }
  
   /* Close all queued fds */
   if (handle->queued_fds != NULL) {
     queued_fds = handle->queued_fds;
     for (i = 0; i < queued_fds->offset; i++)
       uv__close(queued_fds->fds[i]);
     uv__free(handle->queued_fds);
     handle->queued_fds = NULL;
   }
  
   assert(!uv__io_active(&handle->io_watcher, POLLIN | POLLOUT));
 }

References assert(), uv__stream_queued_fds_s::fds, handle, i, NULL, uv__stream_queued_fds_s::offset, s, STDERR_FILENO, uv__close(), uv__free(), uv__handle_stop, uv__io_active(), uv__io_close(), uv__stream_osx_interrupt_select(), uv_close(), UV_HANDLE_READABLE, UV_HANDLE_WRITABLE, uv_read_stop(), uv_sem_destroy(), uv_sem_post(), and uv_thread_join().

Referenced by uv__pipe_close(), uv__process_close_stream(), uv__tcp_close(), and uv_close().

◆ uv__stream_connect()

static void uv__stream_connect ( uv_stream_t * stream )

static

We get called here from directly following a call to connect(2). In order to determine if we've errored out or succeeded must call getsockopt.

Definition at line 1344 of file stream.c.

                                                     {
   int error;
   uv_connect_t* req = stream->connect_req;
   socklen_t errorsize = sizeof(int);
  
   assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE);
   assert(req);
  
   if (stream->delayed_error) {
     /* To smooth over the differences between unixes errors that
      * were reported synchronously on the first connect can be delayed
      * until the next tick--which is now.
      */
     error = stream->delayed_error;
     stream->delayed_error = 0;
   } else {
     /* Normal situation: we need to get the socket error from the kernel. */
     assert(uv__stream_fd(stream) >= 0);
     getsockopt(uv__stream_fd(stream),
                SOL_SOCKET,
                SO_ERROR,
                &error,
                &errorsize);
     error = UV__ERR(error);
   }
  
   if (error == UV__ERR(EINPROGRESS))
     return;
  
   stream->connect_req = NULL;
   uv__req_unregister(stream->loop, req);
  
   if (error < 0 || QUEUE_EMPTY(&stream->write_queue)) {
     uv__io_stop(stream->loop, &stream->io_watcher, POLLOUT);
   }
  
   if (req->cb)
     req->cb(req, error);
  
   if (uv__stream_fd(stream) == -1)
     return;
  
   if (error < 0) {
     uv__stream_flush_write_queue(stream, UV_ECANCELED);
     uv__write_callbacks(stream);
   }
 }

References assert(), EINPROGRESS, error(), int, NULL, QUEUE_EMPTY, req, SO_ERROR, SOL_SOCKET, UV__ERR, uv__io_stop(), uv__req_unregister, uv__stream_fd, uv__stream_flush_write_queue(), and uv__write_callbacks().

Referenced by uv__stream_io().

◆ uv__stream_destroy()

void uv__stream_destroy ( uv_stream_t * stream )

Definition at line 458 of file stream.c.

                                              {
   assert(!uv__io_active(&stream->io_watcher, POLLIN | POLLOUT));
   assert(stream->flags & UV_HANDLE_CLOSED);
  
   if (stream->connect_req) {
     uv__req_unregister(stream->loop, stream->connect_req);
     stream->connect_req->cb(stream->connect_req, UV_ECANCELED);
     stream->connect_req = NULL;
   }
  
   uv__stream_flush_write_queue(stream, UV_ECANCELED);
   uv__write_callbacks(stream);
  
   if (stream->shutdown_req) {
     /* The ECANCELED error code is a lie, the shutdown(2) syscall is a
      * fait accompli at this point. Maybe we should revisit this in v0.11.
      * A possible reason for leaving it unchanged is that it informs the
      * callee that the handle has been destroyed.
      */
     uv__req_unregister(stream->loop, stream->shutdown_req);
     stream->shutdown_req->cb(stream->shutdown_req, UV_ECANCELED);
     stream->shutdown_req = NULL;
   }
  
   assert(stream->write_queue_size == 0);
 }

References assert(), NULL, uv__io_active(), uv__req_unregister, uv__stream_flush_write_queue(), uv__write_callbacks(), and UV_HANDLE_CLOSED.

Referenced by uv__finish_close().

◆ uv__stream_eof()

static void uv__stream_eof	(	uv_stream_t *	stream,
		const uv_buf_t *	buf
	)

static

Definition at line 1001 of file stream.c.

                                                                      {
   stream->flags |= UV_HANDLE_READ_EOF;
   stream->flags &= ~UV_HANDLE_READING;
   uv__io_stop(stream->loop, &stream->io_watcher, POLLIN);
   if (!uv__io_active(&stream->io_watcher, POLLOUT))
     uv__handle_stop(stream);
   uv__stream_osx_interrupt_select(stream);
   stream->read_cb(stream, UV_EOF, buf);
 }

References uv__handle_stop, uv__io_active(), uv__io_stop(), uv__stream_osx_interrupt_select(), UV_HANDLE_READ_EOF, and UV_HANDLE_READING.

Referenced by uv__read(), and uv__stream_io().

◆ uv__stream_flush_write_queue()

void uv__stream_flush_write_queue	(	uv_stream_t *	stream,
		int	error
	)

Definition at line 443 of file stream.c.

                                                                   {
   uv_write_t* req;
   QUEUE* q;
   while (!QUEUE_EMPTY(&stream->write_queue)) {
     q = QUEUE_HEAD(&stream->write_queue);
     QUEUE_REMOVE(q);
  
     req = QUEUE_DATA(q, uv_write_t, queue);
     req->error = error;
  
     QUEUE_INSERT_TAIL(&stream->write_completed_queue, &req->queue);
   }
 }

References error(), queue, QUEUE_DATA, QUEUE_EMPTY, QUEUE_HEAD, QUEUE_INSERT_TAIL, QUEUE_REMOVE, and req.

Referenced by uv__stream_connect(), and uv__stream_destroy().

◆ uv__stream_init()

void uv__stream_init	(	uv_loop_t *	loop,
		uv_stream_t *	stream,
		uv_handle_type	type
	)

Definition at line 85 of file stream.c.

                                           {
   int err;
  
   uv__handle_init(loop, (uv_handle_t*)stream, type);
   stream->read_cb = NULL;
   stream->alloc_cb = NULL;
   stream->close_cb = NULL;
   stream->connection_cb = NULL;
   stream->connect_req = NULL;
   stream->shutdown_req = NULL;
   stream->accepted_fd = -1;
   stream->queued_fds = NULL;
   stream->delayed_error = 0;
   QUEUE_INIT(&stream->write_queue);
   QUEUE_INIT(&stream->write_completed_queue);
   stream->write_queue_size = 0;
  
   if (loop->emfile_fd == -1) {
     err = uv__open_cloexec("/dev/null", O_RDONLY);
     if (err < 0)
         /* In the rare case that "/dev/null" isn't mounted open "/"
          * instead.
          */
         err = uv__open_cloexec("/", O_RDONLY);
     if (err >= 0)
       loop->emfile_fd = err;
   }
  
 #if defined(__APPLE__)
   stream->select = NULL;
 #endif /* defined(__APPLE_) */
  
   uv__io_init(&stream->io_watcher, uv__stream_io, -1);
 }

References err, loop, NULL, O_RDONLY, QUEUE_INIT, type, uv__handle_init, uv__io_init(), uv__open_cloexec(), and uv__stream_io().

Referenced by uv_pipe_init(), uv_tcp_init_ex(), and uv_tty_init().

◆ uv__stream_io()

static void uv__stream_io	(	uv_loop_t *	loop,
		uv__io_t *	w,
		unsigned int	events
	)

static

Definition at line 1287 of file stream.c.

                                                                              {
   uv_stream_t* stream;
  
   stream = container_of(w, uv_stream_t, io_watcher);
  
   assert(stream->type == UV_TCP ||
          stream->type == UV_NAMED_PIPE ||
          stream->type == UV_TTY);
   assert(!(stream->flags & UV_HANDLE_CLOSING));
  
   if (stream->connect_req) {
     uv__stream_connect(stream);
     return;
   }
  
   assert(uv__stream_fd(stream) >= 0);
  
   /* Ignore POLLHUP here. Even if it's set, there may still be data to read. */
   if (events & (POLLIN | POLLERR | POLLHUP))
     uv__read(stream);
  
   if (uv__stream_fd(stream) == -1)
     return;  /* read_cb closed stream. */
  
   /* Short-circuit iff POLLHUP is set, the user is still interested in read
    * events and uv__read() reported a partial read but not EOF. If the EOF
    * flag is set, uv__read() called read_cb with err=UV_EOF and we don't
    * have to do anything. If the partial read flag is not set, we can't
    * report the EOF yet because there is still data to read.
    */
   if ((events & POLLHUP) &&
       (stream->flags & UV_HANDLE_READING) &&
       (stream->flags & UV_HANDLE_READ_PARTIAL) &&
       !(stream->flags & UV_HANDLE_READ_EOF)) {
     uv_buf_t buf = { NULL, 0 };
     uv__stream_eof(stream, &buf);
   }
  
   if (uv__stream_fd(stream) == -1)
     return;  /* read_cb closed stream. */
  
   if (events & (POLLOUT | POLLERR | POLLHUP)) {
     uv__write(stream);
     uv__write_callbacks(stream);
  
     /* Write queue drained. */
     if (QUEUE_EMPTY(&stream->write_queue))
       uv__drain(stream);
   }
 }

References assert(), container_of, NULL, QUEUE_EMPTY, uv__drain(), uv__read(), uv__stream_connect(), uv__stream_eof(), uv__stream_fd, uv__write(), uv__write_callbacks(), UV_HANDLE_CLOSING, UV_HANDLE_READ_EOF, UV_HANDLE_READ_PARTIAL, UV_HANDLE_READING, and w.

Referenced by uv__stream_init().

◆ uv__stream_open()

int uv__stream_open	(	uv_stream_t *	stream,
		int	fd,
		int	flags
	)

Definition at line 406 of file stream.c.

                                                             {
 #if defined(__APPLE__)
   int enable;
 #endif
  
   if (!(stream->io_watcher.fd == -1 || stream->io_watcher.fd == fd))
     return UV_EBUSY;
  
   assert(fd >= 0);
   stream->flags |= flags;
  
   if (stream->type == UV_TCP) {
     if ((stream->flags & UV_HANDLE_TCP_NODELAY) && uv__tcp_nodelay(fd, 1))
       return UV__ERR(errno);
  
     /* TODO Use delay the user passed in. */
     if ((stream->flags & UV_HANDLE_TCP_KEEPALIVE) &&
         uv__tcp_keepalive(fd, 1, 60)) {
       return UV__ERR(errno);
     }
   }
  
 #if defined(__APPLE__)
   enable = 1;
   if (setsockopt(fd, SOL_SOCKET, SO_OOBINLINE, &enable, sizeof(enable)) &&
       errno != ENOTSOCK &&
       errno != EINVAL) {
     return UV__ERR(errno);
   }
 #endif
  
   stream->io_watcher.fd = fd;
  
   return 0;
 }

References assert(), EINVAL, ENOTSOCK, fd, flags, SO_OOBINLINE, SOL_SOCKET, UV__ERR, uv__tcp_keepalive(), uv__tcp_nodelay(), UV_HANDLE_TCP_KEEPALIVE, and UV_HANDLE_TCP_NODELAY.

Referenced by new_socket(), uv__process_open_stream(), uv_accept(), uv_pipe_connect(), uv_pipe_open(), uv_tcp_open(), and uv_tty_init().

◆ uv__stream_osx_interrupt_select()

static void uv__stream_osx_interrupt_select ( uv_stream_t * stream )

static

Definition at line 123 of file stream.c.

                                                                  {
 #if defined(__APPLE__)
   /* Notify select() thread about state change */
   uv__stream_select_t* s;
   int r;
  
   s = stream->select;
   if (s == NULL)
     return;
  
   /* Interrupt select() loop
    * NOTE: fake_fd and int_fd are socketpair(), thus writing to one will
    * emit read event on other side
    */
   do
     r = write(s->fake_fd, "x", 1);
   while (r == -1 && errno == EINTR);
  
   assert(r == 1);
 #else  /* !defined(__APPLE__) */
   /* No-op on any other platform */
 #endif  /* !defined(__APPLE__) */
 }

References assert(), EINTR, NULL, r, s, and write.

Referenced by uv__drain(), uv__read(), uv__stream_close(), uv__stream_eof(), uv__write(), uv_read_start(), uv_read_stop(), uv_shutdown(), uv_try_write(), and uv_write2().

◆ uv__stream_queue_fd()

static int uv__stream_queue_fd	(	uv_stream_t *	stream,
		int	fd
	)

static

Definition at line 1012 of file stream.c.

                                                             {
   uv__stream_queued_fds_t* queued_fds;
   unsigned int queue_size;
  
   queued_fds = stream->queued_fds;
   if (queued_fds == NULL) {
     queue_size = 8;
     queued_fds = uv__malloc((queue_size - 1) * sizeof(*queued_fds->fds) +
                             sizeof(*queued_fds));
     if (queued_fds == NULL)
       return UV_ENOMEM;
     queued_fds->size = queue_size;
     queued_fds->offset = 0;
     stream->queued_fds = queued_fds;
  
     /* Grow */
   } else if (queued_fds->size == queued_fds->offset) {
     queue_size = queued_fds->size + 8;
     queued_fds = uv__realloc(queued_fds,
                              (queue_size - 1) * sizeof(*queued_fds->fds) +
                               sizeof(*queued_fds));
  
     /*
      * Allocation failure, report back.
      * NOTE: if it is fatal - sockets will be closed in uv__stream_close
      */
     if (queued_fds == NULL)
       return UV_ENOMEM;
     queued_fds->size = queue_size;
     stream->queued_fds = queued_fds;
   }
  
   /* Put fd in a queue */
   queued_fds->fds[queued_fds->offset++] = fd;
  
   return 0;
 }

References fd, uv__stream_queued_fds_s::fds, NULL, uv__stream_queued_fds_s::offset, uv__stream_queued_fds_s::size, uv__malloc(), and uv__realloc().

Referenced by uv__stream_recv_cmsg().

◆ uv__stream_recv_cmsg()

static int uv__stream_recv_cmsg	(	uv_stream_t *	stream,
		struct msghdr *	msg
	)

static

Definition at line 1060 of file stream.c.

                                                                          {
   struct cmsghdr* cmsg;
  
   for (cmsg = CMSG_FIRSTHDR(msg); cmsg != NULL; cmsg = CMSG_NXTHDR(msg, cmsg)) {
     char* start;
     char* end;
     int err;
     void* pv;
     int* pi;
     unsigned int i;
     unsigned int count;
  
     if (cmsg->cmsg_type != SCM_RIGHTS) {
       fprintf(stderr, "ignoring non-SCM_RIGHTS ancillary data: %d\n",
           cmsg->cmsg_type);
       continue;
     }
  
     /* silence aliasing warning */
     pv = CMSG_DATA(cmsg);
     pi = pv;
  
     /* Count available fds */
     start = (char*) cmsg;
     end = (char*) cmsg + cmsg->cmsg_len;
     count = 0;
     while (start + CMSG_LEN(count * sizeof(*pi)) < end)
       count++;
     assert(start + CMSG_LEN(count * sizeof(*pi)) == end);
  
     for (i = 0; i < count; i++) {
       /* Already has accepted fd, queue now */
       if (stream->accepted_fd != -1) {
         err = uv__stream_queue_fd(stream, pi[i]);
         if (err != 0) {
           /* Close rest */
           for (; i < count; i++)
             uv__close(pi[i]);
           return err;
         }
       } else {
         stream->accepted_fd = pi[i];
       }
     }
   }
  
   return 0;
 }

References assert(), CMSG_DATA, CMSG_FIRSTHDR, cmsghdr::cmsg_len, CMSG_LEN, CMSG_NXTHDR, cmsghdr::cmsg_type, count, test_evm::end, err, i, msg, NULL, SCM_RIGHTS, start, uv__close(), uv__stream_queue_fd(), and while().

Referenced by uv__read().

◆ uv__write()

static void uv__write ( uv_stream_t * stream )

static

Definition at line 802 of file stream.c.

                                            {
   struct iovec* iov;
   QUEUE* q;
   uv_write_t* req;
   int iovmax;
   int iovcnt;
   ssize_t n;
   int err;
  
 start:
  
   assert(uv__stream_fd(stream) >= 0);
  
   if (QUEUE_EMPTY(&stream->write_queue))
     return;
  
   q = QUEUE_HEAD(&stream->write_queue);
   req = QUEUE_DATA(q, uv_write_t, queue);
   assert(req->handle == stream);
  
   /*
    * Cast to iovec. We had to have our own uv_buf_t instead of iovec
    * because Windows's WSABUF is not an iovec.
    */
   assert(sizeof(uv_buf_t) == sizeof(struct iovec));
   iov = (struct iovec*) &(req->bufs[req->write_index]);
   iovcnt = req->nbufs - req->write_index;
  
   iovmax = uv__getiovmax();
  
   /* Limit iov count to avoid EINVALs from writev() */
   if (iovcnt > iovmax)
     iovcnt = iovmax;
  
   /*
    * Now do the actual writev. Note that we've been updating the pointers
    * inside the iov each time we write. So there is no need to offset it.
    */
  
   if (req->send_handle) {
     int fd_to_send;
     struct msghdr msg;
     struct cmsghdr *cmsg;
     union {
       char data[64];
       struct cmsghdr alias;
     } scratch;
  
     if (uv__is_closing(req->send_handle)) {
       err = UV_EBADF;
       goto error;
     }
  
     fd_to_send = uv__handle_fd((uv_handle_t*) req->send_handle);
  
     memset(&scratch, 0, sizeof(scratch));
  
     assert(fd_to_send >= 0);
  
     msg.msg_name = NULL;
     msg.msg_namelen = 0;
     msg.msg_iov = iov;
     msg.msg_iovlen = iovcnt;
     msg.msg_flags = 0;
  
     msg.msg_control = &scratch.alias;
     msg.msg_controllen = CMSG_SPACE(sizeof(fd_to_send));
  
     cmsg = CMSG_FIRSTHDR(&msg);
     cmsg->cmsg_level = SOL_SOCKET;
     cmsg->cmsg_type = SCM_RIGHTS;
     cmsg->cmsg_len = CMSG_LEN(sizeof(fd_to_send));
  
     /* silence aliasing warning */
     {
       void* pv = CMSG_DATA(cmsg);
       int* pi = pv;
       *pi = fd_to_send;
     }
  
     do
       n = sendmsg(uv__stream_fd(stream), &msg, 0);
     while (n == -1 && RETRY_ON_WRITE_ERROR(errno));
  
     /* Ensure the handle isn't sent again in case this is a partial write. */
     if (n >= 0)
       req->send_handle = NULL;
   } else {
     do
       n = uv__writev(uv__stream_fd(stream), iov, iovcnt);
     while (n == -1 && RETRY_ON_WRITE_ERROR(errno));
   }
  
   if (n == -1 && !IS_TRANSIENT_WRITE_ERROR(errno, req->send_handle)) {
     err = UV__ERR(errno);
     goto error;
   }
  
   if (n >= 0 && uv__write_req_update(stream, req, n)) {
     uv__write_req_finish(req);
     return;  /* TODO(bnoordhuis) Start trying to write the next request. */
   }
  
   /* If this is a blocking stream, try again. */
   if (stream->flags & UV_HANDLE_BLOCKING_WRITES)
     goto start;
  
   /* We're not done. */
   uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);
  
   /* Notify select() thread about state change */
   uv__stream_osx_interrupt_select(stream);
  
   return;
  
 error:
   req->error = err;
   uv__write_req_finish(req);
   uv__io_stop(stream->loop, &stream->io_watcher, POLLOUT);
   if (!uv__io_active(&stream->io_watcher, POLLIN))
     uv__handle_stop(stream);
   uv__stream_osx_interrupt_select(stream);
 }

References assert(), CMSG_DATA, CMSG_FIRSTHDR, cmsghdr::cmsg_len, CMSG_LEN, cmsghdr::cmsg_level, CMSG_SPACE, cmsghdr::cmsg_type, err, error(), iov, IS_TRANSIENT_WRITE_ERROR, memset(), msg, n, NULL, queue, QUEUE_DATA, QUEUE_EMPTY, QUEUE_HEAD, req, RETRY_ON_WRITE_ERROR, SCM_RIGHTS, SOL_SOCKET, start, UV__ERR, uv__getiovmax(), uv__handle_fd(), uv__handle_stop, uv__io_active(), uv__io_start(), uv__io_stop(), uv__is_closing, uv__stream_fd, uv__stream_osx_interrupt_select(), uv__write_req_finish(), uv__write_req_update(), uv__writev(), and UV_HANDLE_BLOCKING_WRITES.

Referenced by uv__stream_io(), and uv_write2().

◆ uv__write_callbacks()

static void uv__write_callbacks ( uv_stream_t * stream )

static

Definition at line 927 of file stream.c.

                                                      {
   uv_write_t* req;
   QUEUE* q;
   QUEUE pq;
  
   if (QUEUE_EMPTY(&stream->write_completed_queue))
     return;
  
   QUEUE_MOVE(&stream->write_completed_queue, &pq);
  
   while (!QUEUE_EMPTY(&pq)) {
     /* Pop a req off write_completed_queue. */
     q = QUEUE_HEAD(&pq);
     req = QUEUE_DATA(q, uv_write_t, queue);
     QUEUE_REMOVE(q);
     uv__req_unregister(stream->loop, req);
  
     if (req->bufs != NULL) {
       stream->write_queue_size -= uv__write_req_size(req);
       if (req->bufs != req->bufsml)
         uv__free(req->bufs);
       req->bufs = NULL;
     }
  
     /* NOTE: call callback AFTER freeing the request data. */
     if (req->cb)
       req->cb(req, req->error);
   }
 }

References NULL, queue, QUEUE_DATA, QUEUE_EMPTY, QUEUE_HEAD, QUEUE_MOVE, QUEUE_REMOVE, req, uv__free(), uv__req_unregister, and uv__write_req_size().

Referenced by uv__stream_connect(), uv__stream_destroy(), and uv__stream_io().

◆ uv__write_req_finish()

static void uv__write_req_finish ( uv_write_t * req )

static

Definition at line 762 of file stream.c.

                                                   {
   uv_stream_t* stream = req->handle;
  
   /* Pop the req off tcp->write_queue. */
   QUEUE_REMOVE(&req->queue);
  
   /* Only free when there was no error. On error, we touch up write_queue_size
    * right before making the callback. The reason we don't do that right away
    * is that a write_queue_size > 0 is our only way to signal to the user that
    * they should stop writing - which they should if we got an error. Something
    * to revisit in future revisions of the libuv API.
    */
   if (req->error == 0) {
     if (req->bufs != req->bufsml)
       uv__free(req->bufs);
     req->bufs = NULL;
   }
  
   /* Add it to the write_completed_queue where it will have its
    * callback called in the near future.
    */
   QUEUE_INSERT_TAIL(&stream->write_completed_queue, &req->queue);
   uv__io_feed(stream->loop, &stream->io_watcher);
 }

References NULL, QUEUE_INSERT_TAIL, QUEUE_REMOVE, req, uv__free(), and uv__io_feed().

Referenced by uv__write().

◆ uv__write_req_size()

static size_t uv__write_req_size ( uv_write_t * req )

static

Definition at line 719 of file stream.c.

                                                   {
   size_t size;
  
   assert(req->bufs != NULL);
   size = uv__count_bufs(req->bufs + req->write_index,
                         req->nbufs - req->write_index);
   assert(req->handle->write_queue_size >= size);
  
   return size;
 }

References assert(), NULL, req, and uv__count_bufs().

Referenced by uv__write_callbacks(), and uv_try_write().

◆ uv__write_req_update()

static int uv__write_req_update	(	uv_stream_t *	stream,
		uv_write_t *	req,
		size_t	n
	)

static

Definition at line 737 of file stream.c.

                                           {
   uv_buf_t* buf;
   size_t len;
  
   assert(n <= stream->write_queue_size);
   stream->write_queue_size -= n;
  
   buf = req->bufs + req->write_index;
  
   do {
     len = n < buf->len ? n : buf->len;
     buf->base += len;
     buf->len -= len;
     buf += (buf->len == 0);  /* Advance to next buffer if this one is empty. */
     n -= len;
   } while (n > 0);
  
   req->write_index = buf - req->bufs;
  
   return req->write_index == req->nbufs;
 }

References assert(), len, n, and req.

Referenced by uv__write().

◆ uv__writev()

static ssize_t uv__writev	(	int	fd,
		struct iovec *	vec,
		size_t	n
	)

static

Definition at line 711 of file stream.c.

                                                                {
   if (n == 1)
     return write(fd, vec->iov_base, vec->iov_len);
   else
     return writev(fd, vec, n);
 }

References fd, n, write, and writev.

Referenced by uv__write().

◆ uv_accept()

int uv_accept	(	uv_stream_t *	server,
		uv_stream_t *	client
	)

Definition at line 591 of file stream.c.

                                                         {
   int err;
  
   assert(server->loop == client->loop);
  
   if (server->accepted_fd == -1)
     return UV_EAGAIN;
  
   switch (client->type) {
     case UV_NAMED_PIPE:
     case UV_TCP:
       err = uv__stream_open(client,
                             server->accepted_fd,
                             UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);
       if (err) {
         /* TODO handle error */
         uv__close(server->accepted_fd);
         goto done;
       }
       break;
  
     case UV_UDP:
       err = uv_udp_open((uv_udp_t*) client, server->accepted_fd);
       if (err) {
         uv__close(server->accepted_fd);
         goto done;
       }
       break;
  
     default:
       return UV_EINVAL;
   }
  
   client->flags |= UV_HANDLE_BOUND;
  
 done:
   /* Process queued fds */
   if (server->queued_fds != NULL) {
     uv__stream_queued_fds_t* queued_fds;
  
     queued_fds = server->queued_fds;
  
     /* Read first */
     server->accepted_fd = queued_fds->fds[0];
  
     /* All read, free */
     assert(queued_fds->offset > 0);
     if (--queued_fds->offset == 0) {
       uv__free(queued_fds);
       server->queued_fds = NULL;
     } else {
       /* Shift rest */
       memmove(queued_fds->fds,
               queued_fds->fds + 1,
               queued_fds->offset * sizeof(*queued_fds->fds));
     }
   } else {
     server->accepted_fd = -1;
     if (err == 0)
       uv__io_start(server->loop, &server->io_watcher, POLLIN);
   }
   return err;
 }

◆ uv_is_readable()

int uv_is_readable ( const uv_stream_t * stream )

Definition at line 1606 of file stream.c.

                                               {
   return !!(stream->flags & UV_HANDLE_READABLE);
 }

◆ uv_is_writable()

int uv_is_writable ( const uv_stream_t * stream )

Definition at line 1611 of file stream.c.

                                               {
   return !!(stream->flags & UV_HANDLE_WRITABLE);
 }

◆ uv_listen()

int uv_listen	(	uv_stream_t *	stream,
		int	backlog,
		uv_connection_cb	cb
	)

Definition at line 656 of file stream.c.

                                                                      {
   int err;
  
   switch (stream->type) {
   case UV_TCP:
     err = uv_tcp_listen((uv_tcp_t*)stream, backlog, cb);
     break;
  
   case UV_NAMED_PIPE:
     err = uv_pipe_listen((uv_pipe_t*)stream, backlog, cb);
     break;
  
   default:
     err = UV_EINVAL;
   }
  
   if (err == 0)
     uv__handle_start(stream);
  
   return err;
 }

◆ uv_read_start()

int uv_read_start	(	uv_stream_t *	stream,
		uv_alloc_cb	alloc_cb,
		uv_read_cb	read_cb
	)

Definition at line 1555 of file stream.c.

                                       {
   assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE ||
       stream->type == UV_TTY);
  
   if (stream->flags & UV_HANDLE_CLOSING)
     return UV_EINVAL;
  
   if (!(stream->flags & UV_HANDLE_READABLE))
     return UV_ENOTCONN;
  
   /* The UV_HANDLE_READING flag is irrelevant of the state of the tcp - it just
    * expresses the desired state of the user.
    */
   stream->flags |= UV_HANDLE_READING;
  
   /* TODO: try to do the read inline? */
   /* TODO: keep track of tcp state. If we've gotten a EOF then we should
    * not start the IO watcher.
    */
   assert(uv__stream_fd(stream) >= 0);
   assert(alloc_cb);
  
   stream->read_cb = read_cb;
   stream->alloc_cb = alloc_cb;
  
   uv__io_start(stream->loop, &stream->io_watcher, POLLIN);
   uv__handle_start(stream);
   uv__stream_osx_interrupt_select(stream);
  
   return 0;
 }

◆ uv_read_stop()

int uv_read_stop ( uv_stream_t * stream )

Definition at line 1590 of file stream.c.

                                       {
   if (!(stream->flags & UV_HANDLE_READING))
     return 0;
  
   stream->flags &= ~UV_HANDLE_READING;
   uv__io_stop(stream->loop, &stream->io_watcher, POLLIN);
   if (!uv__io_active(&stream->io_watcher, POLLOUT))
     uv__handle_stop(stream);
   uv__stream_osx_interrupt_select(stream);
  
   stream->read_cb = NULL;
   stream->alloc_cb = NULL;
   return 0;
 }

Referenced by uv__stream_close().

◆ uv_shutdown()

int uv_shutdown	(	uv_shutdown_t *	req,
		uv_stream_t *	stream,
		uv_shutdown_cb	cb
	)

Definition at line 1259 of file stream.c.

                                                                             {
   assert(stream->type == UV_TCP ||
          stream->type == UV_TTY ||
          stream->type == UV_NAMED_PIPE);
  
   if (!(stream->flags & UV_HANDLE_WRITABLE) ||
       stream->flags & UV_HANDLE_SHUT ||
       stream->flags & UV_HANDLE_SHUTTING ||
       uv__is_closing(stream)) {
     return UV_ENOTCONN;
   }
  
   assert(uv__stream_fd(stream) >= 0);
  
   /* Initialize request */
   uv__req_init(stream->loop, req, UV_SHUTDOWN);
   req->handle = stream;
   req->cb = cb;
   stream->shutdown_req = req;
   stream->flags |= UV_HANDLE_SHUTTING;
  
   uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);
   uv__stream_osx_interrupt_select(stream);
  
   return 0;
 }

◆ uv_stream_set_blocking()

int uv_stream_set_blocking	(	uv_stream_t *	handle,
		int	blocking
	)

Definition at line 1688 of file stream.c.

                                                               {
   /* Don't need to check the file descriptor, uv__nonblock()
    * will fail with EBADF if it's not valid.
    */
   return uv__nonblock(uv__stream_fd(handle), !blocking);
 }

◆ uv_try_write()

int uv_try_write	(	uv_stream_t *	stream,
		const uv_buf_t	bufs[],
		unsigned int	nbufs
	)

Definition at line 1507 of file stream.c.

                                      {
   int r;
   int has_pollout;
   size_t written;
   size_t req_size;
   uv_write_t req;
  
   /* Connecting or already writing some data */
   if (stream->connect_req != NULL || stream->write_queue_size != 0)
     return UV_EAGAIN;
  
   has_pollout = uv__io_active(&stream->io_watcher, POLLOUT);
  
   r = uv_write(&req, stream, bufs, nbufs, uv_try_write_cb);
   if (r != 0)
     return r;
  
   /* Remove not written bytes from write queue size */
   written = uv__count_bufs(bufs, nbufs);
   if (req.bufs != NULL)
     req_size = uv__write_req_size(&req);
   else
     req_size = 0;
   written -= req_size;
   stream->write_queue_size -= req_size;
  
   /* Unqueue request, regardless of immediateness */
   QUEUE_REMOVE(&req.queue);
   uv__req_unregister(stream->loop, &req);
   if (req.bufs != req.bufsml)
     uv__free(req.bufs);
   req.bufs = NULL;
  
   /* Do not poll for writable, if we wasn't before calling this */
   if (!has_pollout) {
     uv__io_stop(stream->loop, &stream->io_watcher, POLLOUT);
     uv__stream_osx_interrupt_select(stream);
   }
  
   if (written == 0 && req_size != 0)
     return req.error < 0 ? req.error : UV_EAGAIN;
   else
     return written;
 }

◆ uv_try_write_cb()

void uv_try_write_cb	(	uv_write_t *	req,
		int	status
	)

Definition at line 1501 of file stream.c.

                                                   {
   /* Should not be called */
   abort();
 }

Referenced by uv_try_write().

◆ uv_write()

int uv_write	(	uv_write_t *	req,
		uv_stream_t *	handle,
		const uv_buf_t	bufs[],
		unsigned int	nbufs,
		uv_write_cb	cb
	)

Definition at line 1492 of file stream.c.

                              {
   return uv_write2(req, handle, bufs, nbufs, NULL, cb);
 }

Referenced by uv_try_write(), and uv_write2().

◆ uv_write2()

int uv_write2	(	uv_write_t *	req,
		uv_stream_t *	stream,
		const uv_buf_t	bufs[],
		unsigned int	nbufs,
		uv_stream_t *	send_handle,
		uv_write_cb	cb
	)

Definition at line 1393 of file stream.c.

                               {
   int empty_queue;
  
   assert(nbufs > 0);
   assert((stream->type == UV_TCP ||
           stream->type == UV_NAMED_PIPE ||
           stream->type == UV_TTY) &&
          "uv_write (unix) does not yet support other types of streams");
  
   if (uv__stream_fd(stream) < 0)
     return UV_EBADF;
  
   if (!(stream->flags & UV_HANDLE_WRITABLE))
     return UV_EPIPE;
  
   if (send_handle) {
     if (stream->type != UV_NAMED_PIPE || !((uv_pipe_t*)stream)->ipc)
       return UV_EINVAL;
  
     /* XXX We abuse uv_write2() to send over UDP handles to child processes.
      * Don't call uv__stream_fd() on those handles, it's a macro that on OS X
      * evaluates to a function that operates on a uv_stream_t with a couple of
      * OS X specific fields. On other Unices it does (handle)->io_watcher.fd,
      * which works but only by accident.
      */
     if (uv__handle_fd((uv_handle_t*) send_handle) < 0)
       return UV_EBADF;
  
 #if defined(__CYGWIN__) || defined(__MSYS__)
     /* Cygwin recvmsg always sets msg_controllen to zero, so we cannot send it.
        See https://github.com/mirror/newlib-cygwin/blob/86fc4bf0/winsup/cygwin/fhandler_socket.cc#L1736-L1743 */
     return UV_ENOSYS;
 #endif
   }
  
   /* It's legal for write_queue_size > 0 even when the write_queue is empty;
    * it means there are error-state requests in the write_completed_queue that
    * will touch up write_queue_size later, see also uv__write_req_finish().
    * We could check that write_queue is empty instead but that implies making
    * a write() syscall when we know that the handle is in error mode.
    */
   empty_queue = (stream->write_queue_size == 0);
  
   /* Initialize the req */
   uv__req_init(stream->loop, req, UV_WRITE);
   req->cb = cb;
   req->handle = stream;
   req->error = 0;
   req->send_handle = send_handle;
   QUEUE_INIT(&req->queue);
  
   req->bufs = req->bufsml;
   if (nbufs > ARRAY_SIZE(req->bufsml))
     req->bufs = uv__malloc(nbufs * sizeof(bufs[0]));
  
   if (req->bufs == NULL)
     return UV_ENOMEM;
  
   memcpy(req->bufs, bufs, nbufs * sizeof(bufs[0]));
   req->nbufs = nbufs;
   req->write_index = 0;
   stream->write_queue_size += uv__count_bufs(bufs, nbufs);
  
   /* Append the request to write_queue. */
   QUEUE_INSERT_TAIL(&stream->write_queue, &req->queue);
  
   /* If the queue was empty when this function began, we should attempt to
    * do the write immediately. Otherwise start the write_watcher and wait
    * for the fd to become writable.
    */
   if (stream->connect_req) {
     /* Still connecting, do nothing. */
   }
   else if (empty_queue) {
     uv__write(stream);
   }
   else {
     /*
      * blocking streams should never have anything in the queue.
      * if this assert fires then somehow the blocking stream isn't being
      * sufficiently flushed in uv__write.
      */
     assert(!(stream->flags & UV_HANDLE_BLOCKING_WRITES));
     uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);
     uv__stream_osx_interrupt_select(stream);
   }
  
   return 0;
 }

Referenced by uv_write().

Macros

Functions

Macro Definition Documentation

◆ IS_TRANSIENT_WRITE_ERROR

◆ RETRY_ON_WRITE_ERROR

◆ UV__CMSG_FD_COUNT

◆ UV__CMSG_FD_SIZE

◆ UV_DEC_BACKLOG

Function Documentation

◆ uv__drain()

◆ uv__emfile_trick()

◆ uv__handle_fd()

◆ uv__handle_type()

◆ uv__read()

◆ uv__server_io()

◆ uv__stream_close()

◆ uv__stream_connect()

◆ uv__stream_destroy()

◆ uv__stream_eof()

◆ uv__stream_flush_write_queue()

◆ uv__stream_init()

◆ uv__stream_io()

◆ uv__stream_open()

◆ uv__stream_osx_interrupt_select()

◆ uv__stream_queue_fd()

◆ uv__stream_recv_cmsg()

◆ uv__write()

◆ uv__write_callbacks()

◆ uv__write_req_finish()

◆ uv__write_req_size()

◆ uv__write_req_update()

◆ uv__writev()

◆ uv_accept()

◆ uv_is_readable()

◆ uv_is_writable()

◆ uv_listen()

◆ uv_read_start()

◆ uv_read_stop()

◆ uv_shutdown()

◆ uv_stream_set_blocking()

◆ uv_try_write()

◆ uv_try_write_cb()

◆ uv_write()

◆ uv_write2()