Oleg Efimov <o.efimov@corp.badoo.com>
Lars Gierth <larsg@systemli.org>
rcp <zerhacken@yahoo.com>
+Alexis Campailla <alexis@janeasystems.com>
+StarWing <weasley.wx@gmail.com>
+thierry-FreeBSD <thierry@FreeBSD.org>
+Isaiah Norton <isaiah.norton@gmail.com>
+Raul Martins <raulms.martins@gmail.com>
+David Capello <davidcapello@gmail.com>
+Paul Tan <pyokagan@gmail.com>
-2014.02.28, Version 0.11.21 (Unstable)
+2014.03.11, Version 0.11.22 (Unstable)
+
+Changes since version 0.11.21:
+
+* unix, windows: map ERANGE errno (Saúl Ibarra Corretgé)
+
+* unix, windows: make uv_cwd be consistent with uv_exepath (Saúl Ibarra
+ Corretgé)
+
+* process: remove debug perror() prints (Fedor Indutny)
+
+* windows: fall back for volume info query (Isaiah Norton)
+
+* pipe: allow queueing pending handles (Fedor Indutny)
+
+* windows: fix winsock status codes for address errors (Raul Martins)
+
+* windows: Remove unused variable from uv__pipe_insert_pending_socket (David
+ Capello)
+
+* unix: workaround broken pthread_sigmask on Android (Paul Tan)
+
+* error: add ENXIO for O_NONBLOCK FIFO open() (Fedor Indutny)
+
+* freebsd: use accept4, introduced in version 10 (Saúl Ibarra Corretgé)
+
+* windows: fix warnings of MinGW -Wall -O3 (StarWing)
+
+* openbsd, osx: fix compilation warning on scandir (Saúl Ibarra Corretgé)
+
+* linux: always deregister closing fds from epoll (Geoffry Song)
+
+* unix: reopen tty as /dev/tty (Saúl Ibarra Corretgé)
+
+* kqueue: invalidate fd in uv_fs_event_t (Fedor Indutny)
+
+
+2014.02.28, Version 0.11.21 (Unstable), 3ef958158ae1019e027ebaa93114160099db5206
Changes since version 0.11.20:
* linux: fix C99/C++ comment (Fedor Indutny)
+2014.02.19, Version 0.10.25 (Stable), d778dc588507588b12b9f9d2905078db542ed751
+
+Changes since version 0.10.24:
+
+* stream: start thread after assignments (Oguz Bastemur)
+
+* unix: correct error when calling uv_shutdown twice (Saúl Ibarra Corretgé)
+
2014.01.30, Version 0.10.24 (Stable), aecd296b6bce9b40f06a61c5c94e43d45ac7308a
Changes since version 0.10.23:
test/test-pipe-bind-error.c \
test/test-pipe-connect-error.c \
test/test-pipe-getsockname.c \
+ test/test-pipe-sendmsg.c \
test/test-pipe-server-close.c \
test/test-platform-output.c \
test/test-poll-close.c \
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-CC = gcc
+CC ?= gcc
CFLAGS += -Wall \
-Wextra \
all: libuv.a
+clean:
+ -$(RM) $(OBJS) libuv.a
+
libuv.a: $(OBJS)
$(AR) crs $@ $^
-# FIXME(bnoordhuis) Don't rebuild everything when a source file changes.
-$(OBJS): $(OBJS:.o=.c) $(INCLUDES)
+$(OBJS): %.o : %.c $(INCLUDES)
+ $(CC) $(CFLAGS) -c -o $@ $<
test/test-ping-pong.c
test/test-pipe-bind-error.c
test/test-pipe-connect-error.c
+test/test-pipe-sendmsg.c
test/test-pipe-server-close.c
test/test-platform-output.c
test/test-poll-close.c
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
AC_PREREQ(2.57)
-AC_INIT([libuv], [0.11.21], [https://github.com/joyent/libuv/issues])
+AC_INIT([libuv], [0.11.22], [https://github.com/joyent/libuv/issues])
AC_CONFIG_MACRO_DIR([m4])
m4_include([m4/libuv-extra-automake-flags.m4])
AM_INIT_AUTOMAKE([-Wall -Werror foreign subdir-objects] UV_EXTRA_AUTOMAKE_FLAGS)
AM_CONDITIONAL([WINNT], [AS_CASE([$host_os], [mingw*], [true], [false])])
PANDORA_ENABLE_DTRACE
AC_CHECK_PROG(PKG_CONFIG, pkg-config, yes)
-AM_CONDITIONAL([HAVE_PKG_CONFIG], [test "x$PKG_CONFIG" = "xyes"])
-AS_IF([test "x$PKG_CONFIG" = "xyes"], [
+AM_CONDITIONAL([HAVE_PKG_CONFIG], [test "x$PKG_CONFIG" != "x"])
+AS_IF([test "x$PKG_CONFIG" != "x"], [
AC_CONFIG_FILES([libuv.pc])
])
AC_CONFIG_FILES([Makefile])
#endif /* defined(PTHREAD_BARRIER_SERIAL_THREAD) */
int pthread_yield(void);
+
+/* Workaround pthread_sigmask() returning EINVAL on versions < 4.1 by
+ * replacing all calls to pthread_sigmask with sigprocmask. See:
+ * https://android.googlesource.com/platform/bionic/+/9bf330b5
+ * https://code.google.com/p/android/issues/detail?id=15337
+ */
+int uv__pthread_sigmask(int how, const sigset_t* set, sigset_t* oset);
+
+#ifdef pthread_sigmask
+#undef pthread_sigmask
+#endif
+#define pthread_sigmask(how, set, oldset) uv__pthread_sigmask(how, set, oldset)
+
#endif /* GOOGLE_BREAKPAD_COMMON_ANDROID_TESTING_PTHREAD_FIXES_H */
# define UV__ERANGE (-4034)
#endif
+#if defined(ENXIO) && !defined(_WIN32)
+# define UV__ENXIO (-ENXIO)
+#else
+# define UV__ENXIO (-4033)
+#endif
+
#endif /* UV_ERRNO_H_ */
uv_connection_cb connection_cb; \
int delayed_error; \
int accepted_fd; \
+ void* queued_fds; \
UV_STREAM_PRIVATE_PLATFORM_FIELDS \
#define UV_TCP_PRIVATE_FIELDS /* empty */
#define UV_VERSION_MAJOR 0
#define UV_VERSION_MINOR 11
-#define UV_VERSION_PATCH 21
+#define UV_VERSION_PATCH 22
#define UV_VERSION_IS_RELEASE 1
#endif /* UV_VERSION_H */
uv_write_t ipc_header_write_req; \
int ipc_pid; \
uint64_t remaining_ipc_rawdata_bytes; \
- unsigned char reserved[sizeof(void*)]; \
struct { \
- WSAPROTOCOL_INFOW* socket_info; \
- int tcp_connection; \
+ void* queue[2]; \
+ int queue_len; \
} pending_ipc_info; \
uv_write_t* non_overlapped_writes_tail;
XX(EXDEV, "cross-device link not permitted") \
XX(UNKNOWN, "unknown error") \
XX(EOF, "end of file") \
+ XX(ENXIO, "no such device or address") \
#define UV_HANDLE_TYPE_MAP(XX) \
XX(ASYNC, async) \
typedef void (*uv_read_cb)(uv_stream_t* stream,
ssize_t nread,
const uv_buf_t* buf);
-
-/*
- * Just like the uv_read_cb except that if the pending parameter is true
- * then you can use uv_accept() to pull the new handle into the process.
- * If no handle is pending then pending will be UV_UNKNOWN_HANDLE.
- */
-typedef void (*uv_read2_cb)(uv_pipe_t* pipe,
- ssize_t nread,
- const uv_buf_t* buf,
- uv_handle_type pending);
-
typedef void (*uv_write_cb)(uv_write_t* req, int status);
typedef void (*uv_connect_cb)(uv_connect_t* req, int status);
typedef void (*uv_shutdown_cb)(uv_shutdown_t* req, int status);
size_t write_queue_size; \
uv_alloc_cb alloc_cb; \
uv_read_cb read_cb; \
- uv_read2_cb read2_cb; \
/* private */ \
UV_STREAM_PRIVATE_FIELDS
UV_EXTERN int uv_read_stop(uv_stream_t*);
-/*
- * Extended read methods for receiving handles over a pipe. The pipe must be
- * initialized with ipc == 1.
- */
-UV_EXTERN int uv_read2_start(uv_stream_t*, uv_alloc_cb alloc_cb,
- uv_read2_cb read_cb);
-
/*
* Write data to stream. Buffers are written in order. Example:
*/
UV_EXTERN void uv_pipe_pending_instances(uv_pipe_t* handle, int count);
+/*
+ * Used to receive handles over ipc pipes.
+ *
+ * First - call `uv_pipe_pending_count`, if it is > 0 - initialize handle
+ * using type, returned by `uv_pipe_pending_type` and call
+ * `uv_accept(pipe, handle)`.
+ */
+UV_EXTERN int uv_pipe_pending_count(uv_pipe_t* handle);
+UV_EXTERN uv_handle_type uv_pipe_pending_type(uv_pipe_t* handle);
/*
* uv_poll_t is a subclass of uv_handle_t.
UV_EXTERN int uv_exepath(char* buffer, size_t* size);
/* Gets the current working directory */
-UV_EXTERN int uv_cwd(char* buffer, size_t size);
+UV_EXTERN int uv_cwd(char* buffer, size_t* size);
/* Changes the current working directory */
UV_EXTERN int uv_chdir(const char* dir);
Version: @PACKAGE_VERSION@
Description: multi-platform support library with a focus on asynchronous I/O.
-Libs: -L${libdir} -luv
+Libs: -L${libdir} -luv @LIBS@
Cflags: -I${includedir}
# include <sys/filio.h>
# include <sys/ioctl.h>
# include <sys/wait.h>
+# define UV__O_CLOEXEC O_CLOEXEC
+# if __FreeBSD__ >= 10
+# define uv__accept4 accept4
+# define UV__SOCK_NONBLOCK SOCK_NONBLOCK
+# define UV__SOCK_CLOEXEC SOCK_CLOEXEC
+# endif
+# if !defined(F_DUP2FD_CLOEXEC) && defined(_F_DUP2FD_CLOEXEC)
+# define F_DUP2FD_CLOEXEC _F_DUP2FD_CLOEXEC
+# endif
#endif
static void uv__run_pending(uv_loop_t* loop);
assert(sockfd >= 0);
while (1) {
-#if defined(__linux__)
+#if defined(__linux__) || __FreeBSD__ >= 10
static int no_accept4;
if (no_accept4)
}
-int uv_cwd(char* buffer, size_t size) {
- if (buffer == NULL)
- return -EINVAL;
-
- if (size == 0)
+int uv_cwd(char* buffer, size_t* size) {
+ if (buffer == NULL || size == NULL)
return -EINVAL;
- if (getcwd(buffer, size) == NULL)
+ if (getcwd(buffer, *size) == NULL)
return -errno;
+ *size = strlen(buffer);
return 0;
}
return 0;
}
+
+
+int uv__open_cloexec(const char* path, int flags) {
+ int err;
+ int fd;
+
+#if defined(__linux__) || (defined(__FreeBSD__) && __FreeBSD__ >= 9)
+ static int no_cloexec;
+
+ if (!no_cloexec) {
+ fd = open(path, flags | UV__O_CLOEXEC);
+ if (fd != -1)
+ return fd;
+
+ if (errno != EINVAL)
+ return -errno;
+
+ /* O_CLOEXEC not supported. */
+ no_cloexec = 1;
+ }
+#endif
+
+ fd = open(path, flags);
+ if (fd == -1)
+ return -errno;
+
+ err = uv__cloexec(fd, 1);
+ if (err) {
+ uv__close(fd);
+ return err;
+ }
+
+ return fd;
+}
+
+
+int uv__dup2_cloexec(int oldfd, int newfd) {
+ int r;
+#if defined(__FreeBSD__) && __FreeBSD__ >= 10
+ do
+ r = dup3(oldfd, newfd, O_CLOEXEC);
+ while (r == -1 && errno == EINTR);
+ if (r == -1)
+ return -errno;
+ return r;
+#elif defined(__FreeBSD__) && defined(F_DUP2FD_CLOEXEC)
+ do
+ r = fcntl(oldfd, F_DUP2FD_CLOEXEC, newfd);
+ while (r == -1 && errno == EINTR);
+ if (r != -1)
+ return r;
+ if (errno != EINVAL)
+ return -errno;
+ /* Fall through. */
+#elif defined(__linux__)
+ static int no_dup3;
+ if (!no_dup3) {
+ do
+ r = uv__dup3(oldfd, newfd, UV__O_CLOEXEC);
+ while (r == -1 && (errno == EINTR || errno == EBUSY));
+ if (r != -1)
+ return r;
+ if (errno != ENOSYS)
+ return -errno;
+ /* Fall through. */
+ no_dup3 = 1;
+ }
+#endif
+ {
+ int err;
+ do
+ r = dup2(oldfd, newfd);
+#if defined(__linux__)
+ while (r == -1 && (errno == EINTR || errno == EBUSY));
+#else
+ while (r == -1 && errno == EINTR);
+#endif
+
+ if (r == -1)
+ return -errno;
+
+ err = uv__cloexec(newfd, 1);
+ if (err) {
+ uv__close(newfd);
+ return err;
+ }
+
+ return r;
+ }
+}
}
+#if defined(__OpenBSD__) || (defined(__APPLE__) && !defined(MAC_OS_X_VERSION_10_8))
+static int uv__fs_readdir_filter(struct dirent* dent) {
+#else
static int uv__fs_readdir_filter(const struct dirent* dent) {
+#endif
return strcmp(dent->d_name, ".") != 0 && strcmp(dent->d_name, "..") != 0;
}
# define O_CLOEXEC 0x00100000
#endif
+typedef struct uv__stream_queued_fds_s uv__stream_queued_fds_t;
+
/* handle flags */
enum {
UV_CLOSING = 0x01, /* uv_close() called but not finished. */
UV_CLOCK_FAST = 1 /* Use the fastest clock with <= 1ms granularity. */
} uv_clocktype_t;
+struct uv__stream_queued_fds_s {
+ unsigned int size;
+ unsigned int offset;
+ int fds[1];
+};
+
+
/* core */
int uv__nonblock(int fd, int set);
int uv__close(int fd);
#endif /* defined(__APPLE__) */
void uv__server_io(uv_loop_t* loop, uv__io_t* w, unsigned int events);
int uv__accept(int sockfd);
+int uv__dup2_cloexec(int oldfd, int newfd);
+int uv__open_cloexec(const char* path, int flags);
/* tcp */
int uv_tcp_listen(uv_tcp_t* tcp, int backlog, uv_connection_cb cb);
void uv__timer_close(uv_timer_t* handle);
void uv__udp_close(uv_udp_t* handle);
void uv__udp_finish_close(uv_udp_t* handle);
+uv_handle_type uv__handle_type(int fd);
#if defined(__APPLE__)
int uv___stream_fd(uv_stream_t* handle);
#if defined(__APPLE__)
if (uv__fsevents_close(handle))
- uv__io_stop(handle->loop, &handle->event_watcher, UV__POLLIN);
-#else
- uv__io_stop(handle->loop, &handle->event_watcher, UV__POLLIN);
#endif /* defined(__APPLE__) */
+ {
+ uv__io_close(handle->loop, &handle->event_watcher);
+ }
free(handle->path);
handle->path = NULL;
# endif
#endif /* __NR_pwritev */
+#ifndef __NR_dup3
+# if defined(__x86_64__)
+# define __NR_dup3 292
+# elif defined(__i386__)
+# define __NR_dup3 330
+# elif defined(__arm__)
+# define __NR_dup3 (UV_SYSCALL_BASE + 358)
+# endif
+#endif /* __NR_pwritev */
+
int uv__accept4(int fd, struct sockaddr* addr, socklen_t* addrlen, int flags) {
#if defined(__i386__)
#endif
}
+
ssize_t uv__preadv(int fd, const struct iovec *iov, int iovcnt, off_t offset) {
#if defined(__NR_preadv)
return syscall(__NR_preadv, fd, iov, iovcnt, offset);
#endif
}
+
ssize_t uv__pwritev(int fd, const struct iovec *iov, int iovcnt, off_t offset) {
#if defined(__NR_pwritev)
return syscall(__NR_pwritev, fd, iov, iovcnt, offset);
return errno = ENOSYS, -1;
#endif
}
+
+
+int uv__dup3(int oldfd, int newfd, int flags) {
+#if defined(__NR_dup3)
+ return syscall(__NR_dup3, oldfd, newfd, flags);
+#else
+ return errno = ENOSYS, -1;
+#endif
+}
int flags);
ssize_t uv__preadv(int fd, const struct iovec *iov, int iovcnt, off_t offset);
ssize_t uv__pwritev(int fd, const struct iovec *iov, int iovcnt, off_t offset);
+int uv__dup3(int oldfd, int newfd, int flags);
#endif /* UV_LINUX_SYSCALL_H_ */
void uv_pipe_pending_instances(uv_pipe_t* handle, int count) {
}
+
+
+int uv_pipe_pending_count(uv_pipe_t* handle) {
+ uv__stream_queued_fds_t* queued_fds;
+
+ if (!handle->ipc)
+ return 0;
+
+ if (handle->accepted_fd == -1)
+ return 0;
+
+ if (handle->queued_fds == NULL)
+ return 1;
+
+ queued_fds = handle->queued_fds;
+ return queued_fds->offset + 1;
+}
+
+
+uv_handle_type uv_pipe_pending_type(uv_pipe_t* handle) {
+ if (!handle->ipc)
+ return UV_UNKNOWN_HANDLE;
+
+ if (handle->accepted_fd == -1)
+ return UV_UNKNOWN_HANDLE;
+ else
+ return uv__handle_type(handle->accepted_fd);
+}
close_fd = use_fd;
if (use_fd == -1) {
- uv__write_int(error_fd, -errno);
- perror("failed to open stdio");
+ uv__write_int(error_fd, -errno);
_exit(127);
}
}
if (options->cwd != NULL && chdir(options->cwd)) {
uv__write_int(error_fd, -errno);
- perror("chdir()");
_exit(127);
}
if ((options->flags & UV_PROCESS_SETGID) && setgid(options->gid)) {
uv__write_int(error_fd, -errno);
- perror("setgid()");
_exit(127);
}
if ((options->flags & UV_PROCESS_SETUID) && setuid(options->uid)) {
uv__write_int(error_fd, -errno);
- perror("setuid()");
_exit(127);
}
execvp(options->file, options->args);
uv__write_int(error_fd, -errno);
- perror("execvp()");
_exit(127);
}
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
+/* Android versions < 4.1 have a broken pthread_sigmask.
+ * Note that this block of code must come before any inclusion of
+ * pthread-fixes.h so that the real pthread_sigmask can be referenced.
+ * */
+#include <errno.h>
+#include <pthread.h>
+
+int uv__pthread_sigmask(int how, const sigset_t* set, sigset_t* oset) {
+ static int workaround;
+
+ if (workaround) {
+ return sigprocmask(how, set, oset);
+ } else if (pthread_sigmask(how, set, oset)) {
+ if (errno == EINVAL && sigprocmask(how, set, oset) == 0) {
+ workaround = 1;
+ return 0;
+ } else {
+ return -1;
+ }
+ } else {
+ return 0;
+ }
+}
/*Android doesn't provide pthread_barrier_t for now.*/
#ifndef PTHREAD_BARRIER_SERIAL_THREAD
static size_t uv__write_req_size(uv_write_t* req);
-/* Used by the accept() EMFILE party trick. */
-static int uv__open_cloexec(const char* path, int flags) {
- int err;
- int fd;
-
-#if defined(__linux__)
- fd = open(path, flags | UV__O_CLOEXEC);
- if (fd != -1)
- return fd;
-
- if (errno != EINVAL)
- return -errno;
-
- /* O_CLOEXEC not supported. */
-#endif
-
- fd = open(path, flags);
- if (fd == -1)
- return -errno;
-
- err = uv__cloexec(fd, 1);
- if (err) {
- uv__close(fd);
- return err;
- }
-
- return fd;
-}
-
-
static size_t uv_count_bufs(const uv_buf_t bufs[], unsigned int nbufs) {
unsigned int i;
size_t bytes;
uv__handle_init(loop, (uv_handle_t*)stream, type);
stream->read_cb = NULL;
- stream->read2_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);
if (server->accepted_fd == -1)
return -EAGAIN;
+ err = 0;
switch (client->type) {
case UV_NAMED_PIPE:
case UV_TCP:
if (err) {
/* TODO handle error */
uv__close(server->accepted_fd);
- server->accepted_fd = -1;
- return err;
+ goto done;
}
break;
err = uv_udp_open((uv_udp_t*) client, server->accepted_fd);
if (err) {
uv__close(server->accepted_fd);
- server->accepted_fd = -1;
- return err;
+ goto done;
}
break;
assert(0);
}
- uv__io_start(server->loop, &server->io_watcher, UV__POLLIN);
- server->accepted_fd = -1;
- return 0;
+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) {
+ 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, UV__POLLIN);
+ }
+ return err;
}
msg.msg_flags = 0;
msg.msg_control = (void*) scratch;
- msg.msg_controllen = CMSG_LEN(sizeof(fd_to_send));
+ 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 = msg.msg_controllen;
+ cmsg->cmsg_len = CMSG_LEN(sizeof(fd_to_send));
/* silence aliasing warning */
{
}
-static uv_handle_type uv__handle_type(int fd) {
+uv_handle_type uv__handle_type(int fd) {
struct sockaddr_storage ss;
socklen_t len;
int type;
}
-static void uv__stream_read_cb(uv_stream_t* stream,
- int status,
- const uv_buf_t* buf,
- uv_handle_type type) {
- if (stream->read_cb != NULL)
- stream->read_cb(stream, status, buf);
- else
- stream->read2_cb((uv_pipe_t*) stream, status, buf, type);
-}
-
-
static void uv__stream_eof(uv_stream_t* stream, const uv_buf_t* buf) {
stream->flags |= UV_STREAM_READ_EOF;
uv__io_stop(stream->loop, &stream->io_watcher, UV__POLLIN);
if (!uv__io_active(&stream->io_watcher, UV__POLLOUT))
uv__handle_stop(stream);
uv__stream_osx_interrupt_select(stream);
- uv__stream_read_cb(stream, UV_EOF, buf, UV_UNKNOWN_HANDLE);
+ stream->read_cb(stream, UV_EOF, buf);
+}
+
+
+static int uv__stream_queue_fd(uv_stream_t* stream, int fd) {
+ 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 = malloc((queue_size - 1) * sizeof(*queued_fds->fds) +
+ sizeof(*queued_fds));
+ if (queued_fds == NULL)
+ return -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 = 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 -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;
+}
+
+
+#define UV__CMSG_FD_COUNT 64
+#define UV__CMSG_FD_SIZE (UV__CMSG_FD_COUNT * sizeof(int))
+
+
+static int uv__stream_recv_cmsg(uv_stream_t* stream, struct msghdr* msg) {
+ 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;
}
uv_buf_t buf;
ssize_t nread;
struct msghdr msg;
- struct cmsghdr* cmsg;
- char cmsg_space[64];
+ char cmsg_space[CMSG_SPACE(UV__CMSG_FD_SIZE)];
int count;
+ int err;
+ int is_ipc;
stream->flags &= ~UV_STREAM_READ_PARTIAL;
*/
count = 32;
+ is_ipc = stream->type == UV_NAMED_PIPE && ((uv_pipe_t*) stream)->ipc;
+
/* XXX: Maybe instead of having UV_STREAM_READING we just test if
* tcp->read_cb is NULL or not?
*/
- while ((stream->read_cb || stream->read2_cb)
+ while (stream->read_cb
&& (stream->flags & UV_STREAM_READING)
&& (count-- > 0)) {
assert(stream->alloc_cb != NULL);
stream->alloc_cb((uv_handle_t*)stream, 64 * 1024, &buf);
if (buf.len == 0) {
/* User indicates it can't or won't handle the read. */
- uv__stream_read_cb(stream, UV_ENOBUFS, &buf, UV_UNKNOWN_HANDLE);
+ stream->read_cb(stream, UV_ENOBUFS, &buf);
return;
}
assert(buf.base != NULL);
assert(uv__stream_fd(stream) >= 0);
- if (stream->read_cb) {
+ if (!is_ipc) {
do {
nread = read(uv__stream_fd(stream), buf.base, buf.len);
}
while (nread < 0 && errno == EINTR);
} else {
- assert(stream->read2_cb);
- /* read2_cb uses recvmsg */
+ /* 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 = 64;
- msg.msg_control = (void*) cmsg_space;
+ msg.msg_controllen = sizeof(cmsg_space);
+ msg.msg_control = cmsg_space;
do {
nread = uv__recvmsg(uv__stream_fd(stream), &msg, 0);
uv__io_start(stream->loop, &stream->io_watcher, UV__POLLIN);
uv__stream_osx_interrupt_select(stream);
}
- uv__stream_read_cb(stream, 0, &buf, UV_UNKNOWN_HANDLE);
+ stream->read_cb(stream, 0, &buf);
} else {
/* Error. User should call uv_close(). */
- uv__stream_read_cb(stream, -errno, &buf, UV_UNKNOWN_HANDLE);
+ stream->read_cb(stream, -errno, &buf);
assert(!uv__io_active(&stream->io_watcher, UV__POLLIN) &&
"stream->read_cb(status=-1) did not call uv_close()");
}
/* Successful read */
ssize_t buflen = buf.len;
- if (stream->read_cb) {
- stream->read_cb(stream, nread, &buf);
- } else {
- assert(stream->read2_cb);
-
- /*
- * XXX: Some implementations can send multiple file descriptors in a
- * single message. We should be using CMSG_NXTHDR() to walk the
- * chain to get at them all. This would require changing the API to
- * hand these back up the caller, is a pain.
- */
-
- for (cmsg = CMSG_FIRSTHDR(&msg);
- msg.msg_controllen > 0 && cmsg != NULL;
- cmsg = CMSG_NXTHDR(&msg, cmsg)) {
-
- if (cmsg->cmsg_type == SCM_RIGHTS) {
- if (stream->accepted_fd != -1) {
- fprintf(stderr, "(libuv) ignoring extra FD received\n");
- }
-
- /* silence aliasing warning */
- {
- void* pv = CMSG_DATA(cmsg);
- int* pi = pv;
- stream->accepted_fd = *pi;
- }
-
- } else {
- fprintf(stderr, "ignoring non-SCM_RIGHTS ancillary data: %d\n",
- cmsg->cmsg_type);
- }
- }
-
-
- if (stream->accepted_fd >= 0) {
- stream->read2_cb((uv_pipe_t*) stream,
- nread,
- &buf,
- uv__handle_type(stream->accepted_fd));
- } else {
- stream->read2_cb((uv_pipe_t*) stream, nread, &buf, UV_UNKNOWN_HANDLE);
+ if (is_ipc) {
+ err = uv__stream_recv_cmsg(stream, &msg);
+ if (err != 0) {
+ stream->read_cb(stream, err, NULL);
+ return;
}
}
+ stream->read_cb(stream, nread, &buf);
/* Return if we didn't fill the buffer, there is no more data to read. */
if (nread < buflen) {
}
+#undef UV__CMSG_FD_COUNT
+#undef UV__CMSG_FD_SIZE
+
+
int uv_shutdown(uv_shutdown_t* req, uv_stream_t* stream, uv_shutdown_cb cb) {
assert((stream->type == UV_TCP || stream->type == UV_NAMED_PIPE) &&
"uv_shutdown (unix) only supports uv_handle_t right now");
}
-static int uv__read_start_common(uv_stream_t* stream,
- uv_alloc_cb alloc_cb,
- uv_read_cb read_cb,
- uv_read2_cb read2_cb) {
+int uv_read_start(uv_stream_t* stream,
+ uv_alloc_cb alloc_cb,
+ uv_read_cb read_cb) {
assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE ||
stream->type == UV_TTY);
assert(alloc_cb);
stream->read_cb = read_cb;
- stream->read2_cb = read2_cb;
stream->alloc_cb = alloc_cb;
uv__io_start(stream->loop, &stream->io_watcher, UV__POLLIN);
}
-int uv_read_start(uv_stream_t* stream, uv_alloc_cb alloc_cb,
- uv_read_cb read_cb) {
- return uv__read_start_common(stream, alloc_cb, read_cb, NULL);
-}
-
-
-int uv_read2_start(uv_stream_t* stream, uv_alloc_cb alloc_cb,
- uv_read2_cb read_cb) {
- return uv__read_start_common(stream, alloc_cb, NULL, read_cb);
-}
-
-
int uv_read_stop(uv_stream_t* stream) {
/* Sanity check. We're going to stop the handle unless it's primed for
* writing but that means there should be some kind of write action in
uv__stream_osx_interrupt_select(stream);
stream->read_cb = NULL;
- stream->read2_cb = NULL;
stream->alloc_cb = NULL;
return 0;
}
void uv__stream_close(uv_stream_t* handle) {
+ unsigned int i;
+ uv__stream_queued_fds_t* queued_fds;
+
#if defined(__APPLE__)
/* Terminate select loop first */
if (handle->select != NULL) {
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]);
+ free(handle->queued_fds);
+ handle->queued_fds = NULL;
+ }
+
assert(!uv__io_active(&handle->io_watcher, UV__POLLIN | UV__POLLOUT));
}
int uv_tty_init(uv_loop_t* loop, uv_tty_t* tty, int fd, int readable) {
- uv__stream_init(loop, (uv_stream_t*)tty, UV_TTY);
+ int flags;
+ int newfd;
+ int r;
+
+ newfd = -1;
+
+ uv__stream_init(loop, (uv_stream_t*) tty, UV_TTY);
+
+ /* Reopen the file descriptor when it refers to a tty. This lets us put the
+ * tty in non-blocking mode without affecting other processes that share it
+ * with us.
+ *
+ * Example: `node | cat` - if we put our fd 0 in non-blocking mode, it also
+ * affects fd 1 of `cat` because both file descriptors refer to the same
+ * struct file in the kernel. When we reopen our fd 0, it points to a
+ * different struct file, hence changing its properties doesn't affect
+ * other processes.
+ */
+ if (isatty(fd)) {
+ newfd = uv__open_cloexec("/dev/tty", O_RDWR);
+
+ if (newfd == -1)
+ return -errno;
-#if defined(__APPLE__)
- {
- int err = uv__stream_try_select((uv_stream_t*) tty, &fd);
- if (err)
- return err;
+ r = uv__dup2_cloexec(newfd, fd);
+ if (r < 0 && r != -EINVAL) {
+ /* EINVAL means newfd == fd which could conceivably happen if another
+ * thread called close(fd) between our calls to isatty() and open().
+ * That's a rather unlikely event but let's handle it anyway.
+ */
+ uv__close(newfd);
+ return r;
+ }
+
+ fd = newfd;
}
-#endif /* defined(__APPLE__) */
-
- if (readable) {
- uv__nonblock(fd, 1);
- uv__stream_open((uv_stream_t*)tty, fd, UV_STREAM_READABLE);
- } else {
- /* Note: writable tty we set to blocking mode. */
- uv__stream_open((uv_stream_t*)tty, fd, UV_STREAM_WRITABLE);
- tty->flags |= UV_STREAM_BLOCKING;
+
+#if defined(__APPLE__)
+ r = uv__stream_try_select((uv_stream_t*) tty, &fd);
+ if (r) {
+ if (newfd != -1)
+ uv__close(newfd);
+ return r;
}
+#endif
+ if (readable)
+ flags = UV_STREAM_READABLE;
+ else
+ flags = UV_STREAM_WRITABLE;
+
+ uv__nonblock(fd, 1);
+ uv__stream_open((uv_stream_t*) tty, fd, flags);
tty->mode = 0;
+
return 0;
}
#endif
+#if !defined(__MINGW32__) || __MSVCRT_VERSION__ >= 0x800
static void uv__crt_invalid_parameter_handler(const wchar_t* expression,
const wchar_t* function, const wchar_t * file, unsigned int line,
uintptr_t reserved) {
/* No-op. */
}
+#endif
static void uv_init(void) {
return UV_EBUSY;
}
if (loop != &uv_default_loop_) {
- int i;
+ size_t i;
for (i = 0; i < ARRAY_SIZE(loop->poll_peer_sockets); i++) {
SOCKET sock = loop->poll_peer_sockets[i];
if (sock != 0 && sock != INVALID_SOCKET)
#define TIME_T_TO_FILETIME(time, filetime_ptr) \
do { \
- *(uint64_t*) (filetime_ptr) = ((int64_t) (time) * 10000000LL) + \
+ uint64_t bigtime = ((int64_t) (time) * 10000000LL) + \
116444736000000000ULL; \
+ (filetime_ptr)->dwLowDateTime = bigtime & 0xFFFFFFFF; \
+ (filetime_ptr)->dwHighDateTime = bigtime >> 32; \
} while(0)
#define IS_SLASH(c) ((c) == L'\\' || (c) == L'/')
const char* path, const char* new_path, const int copy_path) {
char* buf;
char* pos;
- ssize_t buf_sz = 0, path_len, pathw_len, new_pathw_len;
+ ssize_t buf_sz = 0, path_len, pathw_len = 0, new_pathw_len = 0;
/* new_path can only be set if path is also set. */
assert(new_path == NULL || path != NULL);
-1,
(WCHAR*) pos,
pathw_len);
- assert(r == pathw_len);
+ assert(r == (DWORD) pathw_len);
req->pathw = (WCHAR*) pos;
pos += r * sizeof(WCHAR);
} else {
-1,
(WCHAR*) pos,
new_pathw_len);
- assert(r == new_pathw_len);
+ assert(r == (DWORD) new_pathw_len);
req->new_pathw = (WCHAR*) pos;
pos += r * sizeof(WCHAR);
} else {
FileFsVolumeInformation);
/* Buffer overflow (a warning status code) is expected here. */
- if (NT_ERROR(nt_status)) {
+ if (io_status.Status == STATUS_NOT_IMPLEMENTED) {
+ statbuf->st_dev = 0;
+ } else if (NT_ERROR(nt_status)) {
SetLastError(pRtlNtStatusToDosError(nt_status));
return -1;
+ } else {
+ statbuf->st_dev = volume_info.VolumeSerialNumber;
}
/* Todo: st_mode should probably always be 0666 for everyone. We might also
statbuf->st_nlink = file_info.StandardInformation.NumberOfLinks;
- statbuf->st_dev = volume_info.VolumeSerialNumber;
-
/* The st_blksize is supposed to be the 'optimal' number of bytes for reading
* and writing to the disk. That is, for any definition of 'optimal' - it's
* supposed to at least avoid read-update-write behavior when writing to the
int uv_pipe_accept(uv_pipe_t* server, uv_stream_t* client);
int uv_pipe_read_start(uv_pipe_t* handle, uv_alloc_cb alloc_cb,
uv_read_cb read_cb);
-int uv_pipe_read2_start(uv_pipe_t* handle, uv_alloc_cb alloc_cb,
- uv_read2_cb read_cb);
int uv_pipe_write(uv_loop_t* loop, uv_write_t* req, uv_pipe_t* handle,
const uv_buf_t bufs[], unsigned int nbufs, uv_write_cb cb);
int uv_pipe_write2(uv_loop_t* loop, uv_write_t* req, uv_pipe_t* handle,
void uv__util_init();
int uv_parent_pid();
-void uv_fatal_error(const int errorno, const char* syscall);
+__declspec(noreturn) void uv_fatal_error(const int errorno, const char* syscall);
/*
#include "stream-inl.h"
#include "req-inl.h"
+typedef struct uv__ipc_queue_item_s uv__ipc_queue_item_t;
+
+struct uv__ipc_queue_item_s {
+ /*
+ * NOTE: It is important for socket_info to be the first field,
+ * because we will we assigning it to the pending_ipc_info.socket_info
+ */
+ WSAPROTOCOL_INFOW socket_info;
+ QUEUE member;
+ int tcp_connection;
+};
/* A zero-size buffer for use by uv_pipe_read */
static char uv_zero_[] = "";
handle->name = NULL;
handle->ipc_pid = 0;
handle->remaining_ipc_rawdata_bytes = 0;
- handle->pending_ipc_info.socket_info = NULL;
- handle->pending_ipc_info.tcp_connection = 0;
+ QUEUE_INIT(&handle->pending_ipc_info.queue);
+ handle->pending_ipc_info.queue_len = 0;
handle->ipc = ipc;
handle->non_overlapped_writes_tail = NULL;
NTSTATUS nt_status;
IO_STATUS_BLOCK io_status;
FILE_PIPE_LOCAL_INFORMATION pipe_info;
+ uv__ipc_queue_item_t* item;
if ((handle->flags & UV_HANDLE_CONNECTION) &&
handle->shutdown_req != NULL &&
assert(!(handle->flags & UV_HANDLE_CLOSED));
if (handle->flags & UV_HANDLE_CONNECTION) {
- if (handle->pending_ipc_info.socket_info) {
- free(handle->pending_ipc_info.socket_info);
- handle->pending_ipc_info.socket_info = NULL;
+ /* Free pending sockets */
+ while (!QUEUE_EMPTY(&handle->pending_ipc_info.queue)) {
+ QUEUE* q;
+ SOCKET socket;
+
+ q = QUEUE_HEAD(&handle->pending_ipc_info.queue);
+ QUEUE_REMOVE(q);
+ item = QUEUE_DATA(q, uv__ipc_queue_item_t, member);
+
+ /* Materialize socket and close it */
+ socket = WSASocketW(FROM_PROTOCOL_INFO,
+ FROM_PROTOCOL_INFO,
+ FROM_PROTOCOL_INFO,
+ &item->socket_info,
+ 0,
+ WSA_FLAG_OVERLAPPED);
+ free(item);
+
+ if (socket != INVALID_SOCKET)
+ closesocket(socket);
}
+ handle->pending_ipc_info.queue_len = 0;
if (handle->flags & UV_HANDLE_EMULATE_IOCP) {
if (handle->read_req.wait_handle != INVALID_HANDLE_VALUE) {
uv_loop_t* loop = server->loop;
uv_pipe_t* pipe_client;
uv_pipe_accept_t* req;
+ QUEUE* q;
+ uv__ipc_queue_item_t* item;
+ int err;
if (server->ipc) {
- if (!server->pending_ipc_info.socket_info) {
+ if (QUEUE_EMPTY(&server->pending_ipc_info.queue)) {
/* No valid pending sockets. */
return WSAEWOULDBLOCK;
}
- return uv_tcp_import((uv_tcp_t*)client, server->pending_ipc_info.socket_info,
- server->pending_ipc_info.tcp_connection);
+ q = QUEUE_HEAD(&server->pending_ipc_info.queue);
+ QUEUE_REMOVE(q);
+ server->pending_ipc_info.queue_len--;
+ item = QUEUE_DATA(q, uv__ipc_queue_item_t, member);
+
+ err = uv_tcp_import((uv_tcp_t*)client,
+ &item->socket_info,
+ item->tcp_connection);
+ if (err != 0)
+ return err;
+
+ free(item);
+
} else {
pipe_client = (uv_pipe_t*)client;
}
-static int uv_pipe_read_start_impl(uv_pipe_t* handle, uv_alloc_cb alloc_cb,
- uv_read_cb read_cb, uv_read2_cb read2_cb) {
+int uv_pipe_read_start(uv_pipe_t* handle,
+ uv_alloc_cb alloc_cb,
+ uv_read_cb read_cb) {
uv_loop_t* loop = handle->loop;
handle->flags |= UV_HANDLE_READING;
INCREASE_ACTIVE_COUNT(loop, handle);
handle->read_cb = read_cb;
- handle->read2_cb = read2_cb;
handle->alloc_cb = alloc_cb;
/* If reading was stopped and then started again, there could still be a */
}
-int uv_pipe_read_start(uv_pipe_t* handle, uv_alloc_cb alloc_cb,
- uv_read_cb read_cb) {
- return uv_pipe_read_start_impl(handle, alloc_cb, read_cb, NULL);
-}
-
-
-int uv_pipe_read2_start(uv_pipe_t* handle, uv_alloc_cb alloc_cb,
- uv_read2_cb read_cb) {
- return uv_pipe_read_start_impl(handle, alloc_cb, NULL, read_cb);
-}
-
-
static void uv_insert_non_overlapped_write_req(uv_pipe_t* handle,
uv_write_t* req) {
req->next_req = NULL;
int err;
int result;
uv_tcp_t* tcp_send_handle;
- uv_write_t* ipc_header_req;
+ uv_write_t* ipc_header_req = NULL;
uv_ipc_frame_uv_stream ipc_frame;
if (nbufs != 1 && (nbufs != 0 || !send_handle)) {
/* Request completed immediately. */
req->queued_bytes = 0;
} else {
+ assert(ipc_header_req != NULL);
/* Request queued by the kernel. */
if (WaitForSingleObject(ipc_header_req->overlapped.hEvent, INFINITE) !=
WAIT_OBJECT_0) {
handle->flags &= ~UV_HANDLE_READABLE;
uv_read_stop((uv_stream_t*) handle);
- if (handle->read2_cb) {
- handle->read2_cb(handle, UV_EOF, &uv_null_buf_, UV_UNKNOWN_HANDLE);
- } else {
- handle->read_cb((uv_stream_t*) handle, UV_EOF, &uv_null_buf_);
- }
+ handle->read_cb((uv_stream_t*) handle, UV_EOF, &uv_null_buf_);
}
uv_read_stop((uv_stream_t*) handle);
- if (handle->read2_cb) {
- handle->read2_cb(handle,
- uv_translate_sys_error(error),
- &buf,
- UV_UNKNOWN_HANDLE);
- } else {
- handle->read_cb((uv_stream_t*)handle, uv_translate_sys_error(error), &buf);
- }
+ handle->read_cb((uv_stream_t*)handle, uv_translate_sys_error(error), &buf);
}
}
+void uv__pipe_insert_pending_socket(uv_pipe_t* handle,
+ WSAPROTOCOL_INFOW* info,
+ int tcp_connection) {
+ uv__ipc_queue_item_t* item;
+
+ item = (uv__ipc_queue_item_t*) malloc(sizeof(*item));
+ if (item == NULL)
+ uv_fatal_error(ERROR_OUTOFMEMORY, "malloc");
+
+ memcpy(&item->socket_info, info, sizeof(item->socket_info));
+ item->tcp_connection = tcp_connection;
+ QUEUE_INSERT_TAIL(&handle->pending_ipc_info.queue, &item->member);
+ handle->pending_ipc_info.queue_len++;
+}
+
+
void uv_process_pipe_read_req(uv_loop_t* loop, uv_pipe_t* handle,
uv_req_t* req) {
DWORD bytes, avail;
assert(bytes == sizeof(ipc_frame) - sizeof(ipc_frame.header));
/* Store the pending socket info. */
- assert(!handle->pending_ipc_info.socket_info);
- handle->pending_ipc_info.socket_info =
- (WSAPROTOCOL_INFOW*)malloc(sizeof(*(handle->pending_ipc_info.socket_info)));
- if (!handle->pending_ipc_info.socket_info) {
- uv_fatal_error(ERROR_OUTOFMEMORY, "malloc");
- }
-
- *(handle->pending_ipc_info.socket_info) = ipc_frame.socket_info;
- handle->pending_ipc_info.tcp_connection =
- ipc_frame.header.flags & UV_IPC_TCP_CONNECTION;
+ uv__pipe_insert_pending_socket(
+ handle,
+ &ipc_frame.socket_info,
+ ipc_frame.header.flags & UV_IPC_TCP_CONNECTION);
}
if (ipc_frame.header.flags & UV_IPC_RAW_DATA) {
handle->alloc_cb((uv_handle_t*) handle, avail, &buf);
if (buf.len == 0) {
- if (handle->read2_cb) {
- handle->read2_cb(handle, UV_ENOBUFS, &buf, UV_UNKNOWN_HANDLE);
- } else if (handle->read_cb) {
- handle->read_cb((uv_stream_t*) handle, UV_ENOBUFS, &buf);
- }
+ handle->read_cb((uv_stream_t*) handle, UV_ENOBUFS, &buf);
break;
}
assert(buf.base != NULL);
assert(handle->remaining_ipc_rawdata_bytes >= bytes);
handle->remaining_ipc_rawdata_bytes =
handle->remaining_ipc_rawdata_bytes - bytes;
- if (handle->read2_cb) {
- handle->read2_cb(handle, bytes, &buf,
- handle->pending_ipc_info.socket_info ? UV_TCP : UV_UNKNOWN_HANDLE);
- } else if (handle->read_cb) {
- handle->read_cb((uv_stream_t*)handle, bytes, &buf);
- }
-
- if (handle->pending_ipc_info.socket_info) {
- free(handle->pending_ipc_info.socket_info);
- handle->pending_ipc_info.socket_info = NULL;
- }
- } else {
- handle->read_cb((uv_stream_t*)handle, bytes, &buf);
}
+ handle->read_cb((uv_stream_t*)handle, bytes, &buf);
/* Read again only if bytes == buf.len */
if (bytes <= buf.len) {
free(name_info);
return err;
}
+
+
+int uv_pipe_pending_count(uv_pipe_t* handle) {
+ if (!handle->ipc)
+ return 0;
+ return handle->pending_ipc_info.queue_len;
+}
+
+
+uv_handle_type uv_pipe_pending_type(uv_pipe_t* handle) {
+ if (!handle->ipc)
+ return UV_UNKNOWN_HANDLE;
+ if (handle->pending_ipc_info.queue_len == 0)
+ return UV_UNKNOWN_HANDLE;
+ else
+ return UV_TCP;
+}
handle->mask_events_2 = 0;
} else {
assert(0);
+ return;
}
/* Setting Exclusive to TRUE makes the other poll request return if there */
static int uv__fast_poll_cancel_poll_req(uv_loop_t* loop, uv_poll_t* handle) {
AFD_POLL_INFO afd_poll_info;
- DWORD result;
+ int result;
afd_poll_info.Exclusive = TRUE;
afd_poll_info.NumberOfHandles = 1;
mask_events = handle->mask_events_2;
} else {
assert(0);
+ return;
}
/* Report an error unless the select was just interrupted. */
SOCKET peer_socket;
index = -1;
- for (i = 0; i < ARRAY_SIZE(uv_msafd_provider_ids); i++) {
+ for (i = 0; (size_t) i < ARRAY_SIZE(uv_msafd_provider_ids); i++) {
if (memcmp((void*) &protocol_info->ProviderId,
(void*) &uv_msafd_provider_ids[i],
sizeof protocol_info->ProviderId) == 0) {
handle->mask_events_2 = 0;
} else {
assert(0);
+ return;
}
if (!QueueUserWorkItem(uv__slow_poll_thread_proc,
mask_events = handle->mask_events_2;
} else {
assert(0);
+ return;
}
if (!REQ_SUCCESS(req)) {
/* an uv_pipe_t for use by the parent. The other one is given to */
/* the child. */
uv_pipe_t* parent_pipe = (uv_pipe_t*) fdopt.data.stream;
- HANDLE child_pipe;
+ HANDLE child_pipe = INVALID_HANDLE_VALUE;
/* Create a new, connected pipe pair. stdio[i].stream should point */
/* to an uninitialized, but not connected pipe handle. */
default:
assert(0);
+ return -1;
}
CHILD_STDIO_HANDLE(buffer, i) = child_handle;
}
/* Make an inheritable copy of the handle. */
- if (uv__duplicate_handle(loop,
- stream_handle,
- &child_handle) < 0) {
+ err = uv__duplicate_handle(loop, stream_handle, &child_handle);
+ if (err)
goto error;
- }
CHILD_STDIO_HANDLE(buffer, i) = child_handle;
CHILD_STDIO_CRT_FLAGS(buffer, i) = crt_flags;
default:
assert(0);
+ return -1;
}
}
char** env;
size_t env_len = 1; /* room for closing null */
int len;
- int i;
+ size_t i;
DWORD var_size;
env_var_t required_vars[] = {
for (handle = RB_NFIND(uv_signal_tree_s, &uv__signal_tree, &lookup);
handle != NULL && handle->signum == signum;
handle = RB_NEXT(uv_signal_tree_s, &uv__signal_tree, handle)) {
- unsigned long previous = InterlockedExchange(&handle->pending_signum, signum);
+ unsigned long previous = InterlockedExchange(
+ (volatile LONG*) &handle->pending_signum, signum);
if (!previous) {
POST_COMPLETION_FOR_REQ(handle->loop, &handle->signal_req);
void uv_process_signal_req(uv_loop_t* loop, uv_signal_t* handle,
uv_req_t* req) {
- unsigned long dispatched_signum;
+ long dispatched_signum;
assert(handle->type == UV_SIGNAL);
assert(req->type == UV_SIGNAL_REQ);
- dispatched_signum = InterlockedExchange(&handle->pending_signum, 0);
+ dispatched_signum = InterlockedExchange(
+ (volatile LONG*) &handle->pending_signum, 0);
assert(dispatched_signum != 0);
/* Check if the pending signal equals the signum that we are watching for. */
}
-int uv_read2_start(uv_stream_t* handle, uv_alloc_cb alloc_cb,
- uv_read2_cb read_cb) {
- int err;
-
- if (handle->flags & UV_HANDLE_READING) {
- return UV_EALREADY;
- }
-
- if (!(handle->flags & UV_HANDLE_READABLE)) {
- return UV_ENOTCONN;
- }
-
- err = ERROR_INVALID_PARAMETER;
- switch (handle->type) {
- case UV_NAMED_PIPE:
- err = uv_pipe_read2_start((uv_pipe_t*)handle, alloc_cb, read_cb);
- break;
- default:
- assert(0);
- }
-
- return uv_translate_sys_error(err);
-}
-
-
int uv_read_stop(uv_stream_t* handle) {
int err;
int tcp_connection) {
int err;
- SOCKET socket = WSASocketW(AF_INET,
- SOCK_STREAM,
- IPPROTO_IP,
+ SOCKET socket = WSASocketW(FROM_PROTOCOL_INFO,
+ FROM_PROTOCOL_INFO,
+ FROM_PROTOCOL_INFO,
socket_protocol_info,
0,
WSA_FLAG_OVERLAPPED);
buf.base[buf_used++] = handle->last_key[handle->last_key_offset++];
/* If the buffer is full, emit it */
- if (buf_used == buf.len) {
+ if ((size_t) buf_used == buf.len) {
handle->read_cb((uv_stream_t*) handle, buf_used, &buf);
buf = uv_null_buf_;
buf_used = 0;
}
-int uv_cwd(char* buffer, size_t size) {
+int uv_cwd(char* buffer, size_t* size) {
DWORD utf16_len;
WCHAR utf16_buffer[MAX_PATH];
int r;
- if (buffer == NULL || size == 0) {
+ if (buffer == NULL || size == NULL) {
return UV_EINVAL;
}
utf16_buffer[utf16_len] = L'\0';
}
+ /* Check how much space we need */
+ r = WideCharToMultiByte(CP_UTF8,
+ 0,
+ utf16_buffer,
+ -1,
+ NULL,
+ 0,
+ NULL,
+ NULL);
+ if (r == 0) {
+ return uv_translate_sys_error(GetLastError());
+ } else if (r > (int) *size) {
+ *size = r;
+ return UV_ENOBUFS;
+ }
+
/* Convert to UTF-8 */
r = WideCharToMultiByte(CP_UTF8,
0,
utf16_buffer,
-1,
buffer,
- size > INT_MAX ? INT_MAX : (int) size,
+ *size > INT_MAX ? INT_MAX : (int) *size,
NULL,
NULL);
if (r == 0) {
return uv_translate_sys_error(GetLastError());
}
+ *size = r;
return 0;
}
int parent_pid = -1;
HANDLE handle;
PROCESSENTRY32 pe;
- int current_pid = GetCurrentProcessId();
+ DWORD current_pid = GetCurrentProcessId();
pe.dwSize = sizeof(PROCESSENTRY32);
handle = CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0);
DWORD cpu_speed_size = sizeof(cpu_speed);
WCHAR cpu_brand[256];
DWORD cpu_brand_size = sizeof(cpu_brand);
- int len;
+ size_t len;
len = _snwprintf(key_name,
ARRAY_SIZE(key_name),
*/
static BOOL uv_get_extension_function(SOCKET socket, GUID guid,
void **target) {
- DWORD result, bytes;
+ int result;
+ DWORD bytes;
result = WSAIoctl(socket,
SIO_GET_EXTENSION_FUNCTION_POINTER,
case STATUS_COMMITMENT_LIMIT:
case STATUS_WORKING_SET_QUOTA:
case STATUS_NO_MEMORY:
- case STATUS_CONFLICTING_ADDRESSES:
case STATUS_QUOTA_EXCEEDED:
case STATUS_TOO_MANY_PAGING_FILES:
case STATUS_REMOTE_RESOURCES:
- case STATUS_TOO_MANY_ADDRESSES:
return WSAENOBUFS;
+ case STATUS_TOO_MANY_ADDRESSES:
case STATUS_SHARING_VIOLATION:
case STATUS_ADDRESS_ALREADY_EXISTS:
return WSAEADDRINUSE;
case STATUS_PIPE_DISCONNECTED:
return WSAESHUTDOWN;
+ case STATUS_CONFLICTING_ADDRESSES:
case STATUS_INVALID_ADDRESS:
case STATUS_INVALID_ADDRESS_COMPONENT:
return WSAEADDRNOTAVAIL;
static void ipc_write_cb(uv_write_t* req, int status);
static void ipc_close_cb(uv_handle_t* handle);
static void ipc_connect_cb(uv_connect_t* req, int status);
-static void ipc_read2_cb(uv_pipe_t* ipc_pipe,
- ssize_t nread,
- const uv_buf_t* buf,
- uv_handle_type type);
+static void ipc_read_cb(uv_stream_t* handle,
+ ssize_t nread,
+ const uv_buf_t* buf);
static void ipc_alloc_cb(uv_handle_t* handle,
size_t suggested_size,
uv_buf_t* buf);
struct ipc_client_ctx* ctx;
ctx = container_of(req, struct ipc_client_ctx, connect_req);
ASSERT(0 == status);
- ASSERT(0 == uv_read2_start((uv_stream_t*) &ctx->ipc_pipe,
- ipc_alloc_cb,
- ipc_read2_cb));
+ ASSERT(0 == uv_read_start((uv_stream_t*) &ctx->ipc_pipe,
+ ipc_alloc_cb,
+ ipc_read_cb));
}
}
-static void ipc_read2_cb(uv_pipe_t* ipc_pipe,
- ssize_t nread,
- const uv_buf_t* buf,
- uv_handle_type type) {
+static void ipc_read_cb(uv_stream_t* handle,
+ ssize_t nread,
+ const uv_buf_t* buf) {
struct ipc_client_ctx* ctx;
uv_loop_t* loop;
+ uv_handle_type type;
+ uv_pipe_t* ipc_pipe;
+ ipc_pipe = (uv_pipe_t*) handle;
ctx = container_of(ipc_pipe, struct ipc_client_ctx, ipc_pipe);
loop = ipc_pipe->loop;
+ ASSERT(1 == uv_pipe_pending_count(ipc_pipe));
+ type = uv_pipe_pending_type(ipc_pipe);
if (type == UV_TCP)
ASSERT(0 == uv_tcp_init(loop, (uv_tcp_t*) ctx->server_handle));
else if (type == UV_NAMED_PIPE)
else
ASSERT(0);
- ASSERT(0 == uv_accept((uv_stream_t*) &ctx->ipc_pipe, ctx->server_handle));
+ ASSERT(0 == uv_accept(handle, ctx->server_handle));
uv_close((uv_handle_t*) &ctx->ipc_pipe, NULL);
}
#ifndef TASK_H_
#define TASK_H_
+#include "uv.h"
+
#include <stdio.h>
#include <stddef.h>
#include <stdlib.h>
/* This macro cleans up the main loop. This is used to avoid valgrind
* warnings about memory being "leaked" by the main event loop.
*/
-#define MAKE_VALGRIND_HAPPY() \
- uv_loop_delete(uv_default_loop())
+#define MAKE_VALGRIND_HAPPY() \
+ do { \
+ close_loop(uv_default_loop()); \
+ uv_loop_delete(uv_default_loop()); \
+ } while (0)
/* Just sugar for wrapping the main() for a task or helper. */
#define TEST_IMPL(name) \
#endif
+#if defined(__clang__) || \
+ defined(__GNUC__) || \
+ defined(__INTEL_COMPILER) || \
+ defined(__SUNPRO_C)
+# define UNUSED __attribute__((unused))
+#else
+# define UNUSED
+#endif
+
+/* Fully close a loop */
+static void close_walk_cb(uv_handle_t* handle, void* arg) {
+ if (!uv_is_closing(handle))
+ uv_close(handle, NULL);
+}
+
+UNUSED static void close_loop(uv_loop_t* loop) {
+ uv_walk(loop, close_walk_cb, NULL);
+ uv_run(loop, UV_RUN_DEFAULT);
+}
+
#endif /* TASK_H_ */
char* last_slash;
int err;
- size = sizeof(buffer_orig) / sizeof(buffer_orig[0]);
- err = uv_cwd(buffer_orig, size);
+ size = sizeof(buffer_orig);
+ err = uv_cwd(buffer_orig, &size);
ASSERT(err == 0);
/* Remove trailing slash unless at a root directory. */
err = uv_chdir(buffer_orig);
ASSERT(err == 0);
- err = uv_cwd(buffer_new, size);
+ err = uv_cwd(buffer_new, &size);
ASSERT(err == 0);
ASSERT(strcmp(buffer_orig, buffer_new) == 0);
#ifdef _WIN32
{
static char src_path_buf[PATHMAX];
+ size_t size;
+ size = sizeof(src_path_buf);
strcpy(src_path_buf, "\\\\?\\");
- uv_cwd(src_path_buf + 4, sizeof(src_path_buf));
+ uv_cwd(src_path_buf + 4, &size);
strcat(src_path_buf, "\\test_dir\\");
test_dir = src_path_buf;
}
}
-static void recv_cb(uv_pipe_t* handle,
+static void recv_cb(uv_stream_t* handle,
ssize_t nread,
- const uv_buf_t* buf,
- uv_handle_type pending) {
+ const uv_buf_t* buf) {
+ uv_handle_type pending;
+ uv_pipe_t* pipe;
int r;
- ASSERT(pending == ctx.expected_type);
- ASSERT(handle == &ctx.channel);
+ pipe = (uv_pipe_t*) handle;
+ ASSERT(pipe == &ctx.channel);
ASSERT(nread >= 0);
+ ASSERT(1 == uv_pipe_pending_count(pipe));
+
+ pending = uv_pipe_pending_type(pipe);
+ ASSERT(pending == ctx.expected_type);
if (pending == UV_NAMED_PIPE)
r = uv_pipe_init(ctx.channel.loop, &ctx.recv.pipe, 0);
abort();
ASSERT(r == 0);
- r = uv_accept((uv_stream_t*)&ctx.channel, &ctx.recv.stream);
+ r = uv_accept(handle, &ctx.recv.stream);
ASSERT(r == 0);
uv_close((uv_handle_t*)&ctx.channel, NULL);
NULL);
ASSERT(r == 0);
- r = uv_read2_start((uv_stream_t*)&ctx.channel, alloc_cb, recv_cb);
+ r = uv_read_start((uv_stream_t*)&ctx.channel, alloc_cb, recv_cb);
ASSERT(r == 0);
r = uv_run(uv_default_loop(), UV_RUN_DEFAULT);
}
-static void read2_cb(uv_pipe_t* handle,
- ssize_t nread,
- const uv_buf_t* rdbuf,
- uv_handle_type pending) {
+static void read_cb(uv_stream_t* handle,
+ ssize_t nread,
+ const uv_buf_t* rdbuf) {
uv_buf_t wrbuf;
+ uv_pipe_t* pipe;
+ uv_handle_type pending;
int r;
- ASSERT(pending == UV_NAMED_PIPE || pending == UV_TCP);
- ASSERT(handle == &ctx.channel);
+ pipe = (uv_pipe_t*) handle;
+ ASSERT(pipe == &ctx.channel);
ASSERT(nread >= 0);
+ ASSERT(1 == uv_pipe_pending_count(pipe));
+
+ pending = uv_pipe_pending_type(pipe);
+ ASSERT(pending == UV_NAMED_PIPE || pending == UV_TCP);
wrbuf = uv_buf_init(".", 1);
abort();
ASSERT(r == 0);
- r = uv_accept((uv_stream_t*)handle, &ctx.recv.stream);
+ r = uv_accept(handle, &ctx.recv.stream);
ASSERT(r == 0);
r = uv_write2(&ctx.write_req,
ASSERT(1 == uv_is_writable((uv_stream_t*)&ctx.channel));
ASSERT(0 == uv_is_closing((uv_handle_t*)&ctx.channel));
- r = uv_read2_start((uv_stream_t*)&ctx.channel, alloc_cb, read2_cb);
+ r = uv_read_start((uv_stream_t*)&ctx.channel, alloc_cb, read_cb);
ASSERT(r == 0);
r = uv_run(uv_default_loop(), UV_RUN_DEFAULT);
static uv_tcp_t tcp_connection;
static int exit_cb_called;
-static int read2_cb_called;
+static int read_cb_called;
static int tcp_write_cb_called;
static int tcp_read_cb_called;
static int on_pipe_read_called;
}
-static void on_read(uv_pipe_t* pipe,
+static void on_read(uv_stream_t* handle,
ssize_t nread,
- const uv_buf_t* buf,
- uv_handle_type pending) {
+ const uv_buf_t* buf) {
int r;
+ uv_pipe_t* pipe;
+ uv_handle_type pending;
uv_buf_t outbuf;
+ pipe = (uv_pipe_t*) handle;
+
if (nread == 0) {
/* Everything OK, but nothing read. */
free(buf->base);
fprintf(stderr, "got %d bytes\n", (int)nread);
+ pending = uv_pipe_pending_type(pipe);
if (!tcp_server_listening) {
+ ASSERT(1 == uv_pipe_pending_count(pipe));
ASSERT(nread > 0 && buf->base && pending != UV_UNKNOWN_HANDLE);
- read2_cb_called++;
+ read_cb_called++;
/* Accept the pending TCP server, and start listening on it. */
ASSERT(pending == UV_TCP);
make_many_connections();
} else if (memcmp("accepted_connection\n", buf->base, nread) == 0) {
/* Remote server has accepted a connection. Close the channel. */
+ ASSERT(0 == uv_pipe_pending_count(pipe));
ASSERT(pending == UV_UNKNOWN_HANDLE);
remote_conn_accepted = 1;
uv_close((uv_handle_t*)&channel, NULL);
}
-static void on_read_connection(uv_pipe_t* pipe,
+static void on_read_connection(uv_stream_t* handle,
ssize_t nread,
- const uv_buf_t* buf,
- uv_handle_type pending) {
+ const uv_buf_t* buf) {
int r;
uv_buf_t outbuf;
+ uv_pipe_t* pipe;
+ uv_handle_type pending;
+ pipe = (uv_pipe_t*) handle;
if (nread == 0) {
/* Everything OK, but nothing read. */
free(buf->base);
fprintf(stderr, "got %d bytes\n", (int)nread);
+ ASSERT(1 == uv_pipe_pending_count(pipe));
+ pending = uv_pipe_pending_type(pipe);
+
ASSERT(nread > 0 && buf->base && pending != UV_UNKNOWN_HANDLE);
- read2_cb_called++;
+ read_cb_called++;
/* Accept the pending TCP connection */
ASSERT(pending == UV_TCP);
r = uv_tcp_init(uv_default_loop(), &tcp_connection);
ASSERT(r == 0);
- r = uv_accept((uv_stream_t*)pipe, (uv_stream_t*)&tcp_connection);
+ r = uv_accept(handle, (uv_stream_t*)&tcp_connection);
ASSERT(r == 0);
/* Make sure that the expected data is correctly multiplexed. */
}
-static int run_ipc_test(const char* helper, uv_read2_cb read_cb) {
+static int run_ipc_test(const char* helper, uv_read_cb read_cb) {
uv_process_t process;
int r;
spawn_helper(&channel, &process, helper);
- uv_read2_start((uv_stream_t*)&channel, on_alloc, read_cb);
+ uv_read_start((uv_stream_t*)&channel, on_alloc, read_cb);
r = uv_run(uv_default_loop(), UV_RUN_DEFAULT);
ASSERT(r == 0);
int r = run_ipc_test("ipc_helper_listen_before_write", on_read);
ASSERT(local_conn_accepted == 1);
ASSERT(remote_conn_accepted == 1);
- ASSERT(read2_cb_called == 1);
+ ASSERT(read_cb_called == 1);
ASSERT(exit_cb_called == 1);
return r;
}
int r = run_ipc_test("ipc_helper_listen_after_write", on_read);
ASSERT(local_conn_accepted == 1);
ASSERT(remote_conn_accepted == 1);
- ASSERT(read2_cb_called == 1);
+ ASSERT(read_cb_called == 1);
ASSERT(exit_cb_called == 1);
return r;
}
TEST_IMPL(ipc_tcp_connection) {
int r = run_ipc_test("ipc_helper_tcp_connection", on_read_connection);
- ASSERT(read2_cb_called == 1);
+ ASSERT(read_cb_called == 1);
ASSERT(tcp_write_cb_called == 1);
ASSERT(tcp_read_cb_called == 1);
ASSERT(exit_cb_called == 1);
TEST_DECLARE (pipe_connect_to_file)
TEST_DECLARE (pipe_getsockname)
TEST_DECLARE (pipe_getsockname_abstract)
+TEST_DECLARE (pipe_sendmsg)
TEST_DECLARE (pipe_server_close)
TEST_DECLARE (connection_fail)
TEST_DECLARE (connection_fail_doesnt_auto_close)
TEST_ENTRY (pipe_listen_without_bind)
TEST_ENTRY (pipe_getsockname)
TEST_ENTRY (pipe_getsockname_abstract)
+ TEST_ENTRY (pipe_sendmsg)
TEST_ENTRY (connection_fail)
TEST_ENTRY (connection_fail_doesnt_auto_close)
--- /dev/null
+/* Copyright Joyent, Inc. and other Node contributors. All rights reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to
+ * deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+ * sell copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+ * IN THE SOFTWARE.
+ */
+
+#include "uv.h"
+#include "task.h"
+
+
+#ifndef _WIN32
+
+#include <fcntl.h>
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+
+/* NOTE: size should be divisible by 2 */
+static uv_pipe_t incoming[4];
+static unsigned int incoming_count;
+static unsigned int close_called;
+
+
+static void set_nonblocking(uv_os_sock_t sock) {
+ int r;
+#ifdef _WIN32
+ unsigned long on = 1;
+ r = ioctlsocket(sock, FIONBIO, &on);
+ ASSERT(r == 0);
+#else
+ int flags = fcntl(sock, F_GETFL, 0);
+ ASSERT(flags >= 0);
+ r = fcntl(sock, F_SETFL, flags | O_NONBLOCK);
+ ASSERT(r >= 0);
+#endif
+}
+
+
+
+
+static void close_cb(uv_handle_t* handle) {
+ close_called++;
+}
+
+
+static void alloc_cb(uv_handle_t* handle, size_t size, uv_buf_t* buf) {
+ static char base[1];
+
+ buf->base = base;
+ buf->len = sizeof(base);
+}
+
+
+static void read_cb(uv_stream_t* handle,
+ ssize_t nread,
+ const uv_buf_t* buf) {
+ uv_pipe_t* p;
+ uv_pipe_t* inc;
+ uv_handle_type pending;
+ unsigned int i;
+
+ p = (uv_pipe_t*) handle;
+ ASSERT(nread >= 0);
+
+ while (uv_pipe_pending_count(p) != 0) {
+ pending = uv_pipe_pending_type(p);
+ ASSERT(pending == UV_NAMED_PIPE);
+
+ ASSERT(incoming_count < ARRAY_SIZE(incoming));
+ inc = &incoming[incoming_count++];
+ ASSERT(0 == uv_pipe_init(p->loop, inc, 0));
+ ASSERT(0 == uv_accept(handle, (uv_stream_t*) inc));
+ }
+
+ if (incoming_count != ARRAY_SIZE(incoming))
+ return;
+
+ ASSERT(0 == uv_read_stop((uv_stream_t*) p));
+ uv_close((uv_handle_t*) p, close_cb);
+ for (i = 0; i < ARRAY_SIZE(incoming); i++)
+ uv_close((uv_handle_t*) &incoming[i], close_cb);
+}
+
+
+TEST_IMPL(pipe_sendmsg) {
+ uv_pipe_t p;
+ int r;
+ int fds[2];
+ int send_fds[ARRAY_SIZE(incoming)];
+ struct msghdr msg;
+ char scratch[64];
+ struct cmsghdr *cmsg;
+ unsigned int i;
+ uv_buf_t buf;
+
+ ASSERT(0 == socketpair(AF_UNIX, SOCK_STREAM, 0, fds));
+ for (i = 0; i < ARRAY_SIZE(send_fds); i += 2)
+ ASSERT(0 == socketpair(AF_UNIX, SOCK_STREAM, 0, send_fds + i));
+ ASSERT(i == ARRAY_SIZE(send_fds));
+ ASSERT(0 == uv_pipe_init(uv_default_loop(), &p, 1));
+ ASSERT(0 == uv_pipe_open(&p, fds[1]));
+
+ buf = uv_buf_init("X", 1);
+ memset(&msg, 0, sizeof(msg));
+ msg.msg_iov = (struct iovec*) &buf;
+ msg.msg_iovlen = 1;
+ msg.msg_flags = 0;
+
+ msg.msg_control = (void*) scratch;
+ msg.msg_controllen = CMSG_LEN(sizeof(send_fds));
+ ASSERT(sizeof(scratch) >= msg.msg_controllen);
+
+ cmsg = CMSG_FIRSTHDR(&msg);
+ cmsg->cmsg_level = SOL_SOCKET;
+ cmsg->cmsg_type = SCM_RIGHTS;
+ cmsg->cmsg_len = msg.msg_controllen;
+
+ /* silence aliasing warning */
+ {
+ void* pv = CMSG_DATA(cmsg);
+ int* pi = pv;
+ for (i = 0; i < ARRAY_SIZE(send_fds); i++)
+ pi[i] = send_fds[i];
+ }
+
+ set_nonblocking(fds[1]);
+ ASSERT(0 == uv_read_start((uv_stream_t*) &p, alloc_cb, read_cb));
+
+ do
+ r = sendmsg(fds[0], &msg, 0);
+ while (r == -1 && errno == EINTR);
+ ASSERT(r == 1);
+
+ uv_run(uv_default_loop(), UV_RUN_DEFAULT);
+ ASSERT(ARRAY_SIZE(incoming) == incoming_count);
+ ASSERT(ARRAY_SIZE(incoming) + 1 == close_called);
+ close(fds[0]);
+
+ MAKE_VALGRIND_HAPPY();
+ return 0;
+}
+
+#else /* !_WIN32 */
+
+TEST_IMPL(pipe_sendmsg) {
+ MAKE_VALGRIND_HAPPY();
+ return 0;
+}
+
+#endif /* _WIN32 */
'test/test-pipe-bind-error.c',
'test/test-pipe-connect-error.c',
'test/test-pipe-getsockname.c',
+ 'test/test-pipe-sendmsg.c',
'test/test-pipe-server-close.c',
'test/test-platform-output.c',
'test/test-poll.c',
projects / platform / upstream / nodejs.git / commitdiff