unix: use select() for specific fds on OS X

kqueue(2) on osx doesn't work (emits EINVAL error) with specific fds
(i.e. /dev/tty, /dev/null, etc). When given such descriptors - start
select(2) watcher thread that will emit io events.

Author: indutny

include/uv-private/uv-darwin.h

@@ -49,4 +49,7 @@
   uv_sem_t cf_sem;                                                            \
   uv_mutex_t cf_mutex;                                                        \
 
+#define UV_STREAM_PRIVATE_PLATFORM_FIELDS                                     \
+  void* select;                                                               \
+
 #endif /* UV_DARWIN_H */

include/uv-private/uv-unix.h

@@ -90,6 +90,10 @@ struct uv__work {
 # define UV_PLATFORM_FS_EVENT_FIELDS /* empty */
 #endif
 
+#ifndef UV_STREAM_PRIVATE_PLATFORM_FIELDS
+# define UV_STREAM_PRIVATE_PLATFORM_FIELDS /* empty */
+#endif
+
 /* Note: May be cast to struct iovec. See writev(2). */
 typedef struct {
   char* base;
@@ -209,6 +213,7 @@ typedef struct {
   uv_connection_cb connection_cb;                                             \
   int delayed_error;                                                          \
   int accepted_fd;                                                            \
+  UV_STREAM_PRIVATE_PLATFORM_FIELDS                                           \
 
 #define UV_TCP_PRIVATE_FIELDS /* empty */
 

src/unix/async.c

@@ -84,6 +84,8 @@ int uv_async_send(uv_async_t* handle) {
     r = write(handle->loop->async_pipefd[1], "x", 1);
   while (r == -1 && errno == EINTR);
 
+  assert(r == -1 || r == 1);
+
   if (r == -1 && errno != EAGAIN && errno != EWOULDBLOCK)
     return uv__set_sys_error(handle->loop, errno);
 

src/unix/internal.h

@@ -197,6 +197,13 @@ 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);
 
+#if defined(__APPLE__)
+int uv___stream_fd(uv_stream_t* handle);
+#define uv__stream_fd(handle) (uv___stream_fd((uv_stream_t*) (handle)))
+#else
+#define uv__stream_fd(handle) ((handle)->io_watcher.fd)
+#endif /* defined(__APPLE__) */
+
 #ifdef UV__O_NONBLOCK
 # define UV__F_NONBLOCK UV__O_NONBLOCK
 #else

src/unix/pipe.c

@@ -57,7 +57,7 @@ int uv_pipe_bind(uv_pipe_t* handle, const char* name) {
   bound = 0;
 
   /* Already bound? */
-  if (handle->io_watcher.fd >= 0) {
+  if (uv__stream_fd(handle) >= 0) {
     uv__set_artificial_error(handle->loop, UV_EINVAL);
     goto out;
   }
@@ -117,13 +117,13 @@ int uv_pipe_listen(uv_pipe_t* handle, int backlog, uv_connection_cb cb) {
   saved_errno = errno;
   status = -1;
 
-  if (handle->io_watcher.fd == -1) {
+  if (uv__stream_fd(handle) == -1) {
     uv__set_artificial_error(handle->loop, UV_EINVAL);
     goto out;
   }
-  assert(handle->io_watcher.fd >= 0);
+  assert(uv__stream_fd(handle) >= 0);
 
-  if ((status = listen(handle->io_watcher.fd, backlog)) == -1) {
+  if ((status = listen(uv__stream_fd(handle), backlog)) == -1) {
     uv__set_sys_error(handle->loop, errno);
   } else {
     handle->connection_cb = cb;
@@ -172,7 +172,7 @@ void uv_pipe_connect(uv_connect_t* req,
   int r;
 
   saved_errno = errno;
-  new_sock = (handle->io_watcher.fd == -1);
+  new_sock = (uv__stream_fd(handle) == -1);
   err = -1;
 
   if (new_sock)
@@ -187,7 +187,8 @@ void uv_pipe_connect(uv_connect_t* req,
    * is either there or not.
    */
   do {
-    r = connect(handle->io_watcher.fd, (struct sockaddr*)&saddr, sizeof saddr);
+    r = connect(uv__stream_fd(handle),
+                (struct sockaddr*)&saddr, sizeof saddr);
   }
   while (r == -1 && errno == EINTR);
 
@@ -196,7 +197,7 @@ void uv_pipe_connect(uv_connect_t* req,
 
   if (new_sock)
     if (uv__stream_open((uv_stream_t*)handle,
-                        handle->io_watcher.fd,
+                        uv__stream_fd(handle),
                         UV_STREAM_READABLE | UV_STREAM_WRITABLE))
       goto out;
 
@@ -233,7 +234,7 @@ static void uv__pipe_accept(uv_loop_t* loop, uv__io_t* w, unsigned int events) {
 
   assert(pipe->type == UV_NAMED_PIPE);
 
-  sockfd = uv__accept(pipe->io_watcher.fd);
+  sockfd = uv__accept(uv__stream_fd(pipe));
   if (sockfd == -1) {
     if (errno != EAGAIN && errno != EWOULDBLOCK) {
       uv__set_sys_error(pipe->loop, errno);

src/unix/process.c

@@ -204,7 +204,7 @@ static int uv__process_init_stdio(uv_stdio_container_t* container, int fds[2]) {
       if (container->flags & UV_INHERIT_FD) {
         fd = container->data.fd;
       } else {
-        fd = container->data.stream->io_watcher.fd;
+        fd = uv__stream_fd(container->data.stream);
       }
 
       if (fd == -1) {