[NaCl SDK] nacl_io: Add support for basic socketpairs.

This is a very basic implementation of a unix socketpair. The socket
nodes have type S_IFSOCK.

The event emitter are set up in a master/slave configuration, the master
owns the fifos and slave handles. The slave has a strong reference to
its master (since it depends on it, cannot survive alone). Each emitter
in the pair has their own status, but share the pair of fifos and use a
single lock in the master to synchronize their access.

Shutdown should be rather simple to bolt on to this implementation.

CQ_EXTRA_TRYBOTS=tryserver.chromium.linux:linux_nacl_sdk;tryserver.chromium.mac:mac_nacl_sdk
BUG=532095

Review URL: https://codereview.chromium.org/1335783005

Cr-Commit-Position: refs/heads/master@{#350849}

10 files changed, 482 insertions, 25 deletions

diff --git a/native_client_sdk/src/libraries/nacl_io/kernel_proxy.cc b/native_client_sdk/src/libraries/nacl_io/kernel_proxy.cc
index b554ade..a141923 100644
--- a/native_client_sdk/src/libraries/nacl_io/kernel_proxy.cc
+++ b/native_client_sdk/src/libraries/nacl_io/kernel_proxy.cc
@@ -39,6 +39,7 @@
 #include "nacl_io/pipe/pipe_node.h"
 #include "nacl_io/socket/tcp_node.h"
 #include "nacl_io/socket/udp_node.h"
+#include "nacl_io/socket/unix_node.h"
 #include "nacl_io/stream/stream_fs.h"
 #include "nacl_io/typed_fs_factory.h"
 #include "sdk_util/auto_lock.h"
@@ -1813,21 +1814,73 @@ int KernelProxy::socketpair(int domain, int type, int protocol, int* sv) {
     return -1;
   }
 
-  // Catch-22: We don't support AF_UNIX, but any other AF doesn't support
-  // socket pairs. Thus, this function always fails.
-  if (AF_UNIX != domain) {
-    errno = EPROTONOSUPPORT;
+  if (AF_INET == domain || AF_INET6 == domain) {
+    errno = EOPNOTSUPP;
     return -1;
   }
 
-  if (AF_INET != domain && AF_INET6 != domain) {
+  if (AF_UNIX != domain) {
     errno = EAFNOSUPPORT;
     return -1;
   }
 
-  // We cannot reach this point.
-  errno = ENOSYS;
-  return -1;
+  if (SOCK_STREAM != type) {
+    errno = EPROTOTYPE;
+    return -1;
+  }
+
+  int open_flags = O_RDWR;
+
+#if defined(SOCK_CLOEXEC)
+  if (type & SOCK_CLOEXEC) {
+#if defined(O_CLOEXEC)
+    // The NaCl newlib version of fcntl.h doesn't currently define
+    // O_CLOEXEC.
+    // TODO(sbc): remove this guard once it gets added.
+    open_flags |= O_CLOEXEC;
+#endif
+    type &= ~SOCK_CLOEXEC;
+  }
+#endif
+
+#if defined(SOCK_NONBLOCK)
+  if (type & SOCK_NONBLOCK) {
+    open_flags |= O_NONBLOCK;
+    type &= ~SOCK_NONBLOCK;
+  }
+#endif
+
+  UnixNode* socket = new UnixNode(stream_fs_.get());
+  Error rtn = socket->Init(O_RDWR);
+  if (rtn != 0) {
+    errno = rtn;
+    return -1;
+  }
+  ScopedNode node0(socket);
+  socket = new UnixNode(stream_fs_.get(), *socket);
+  rtn = socket->Init(O_RDWR);
+  if (rtn != 0) {
+    errno = rtn;
+    return -1;
+  }
+  ScopedNode node1(socket);
+  ScopedKernelHandle handle0(new KernelHandle(stream_fs_, node0));
+  rtn = handle0->Init(open_flags);
+  if (rtn != 0) {
+    errno = rtn;
+    return -1;
+  }
+  ScopedKernelHandle handle1(new KernelHandle(stream_fs_, node1));
+  rtn = handle1->Init(open_flags);
+  if (rtn != 0) {
+    errno = rtn;
+    return -1;
+  }
+
+  sv[0] = AllocateFD(handle0);
+  sv[1] = AllocateFD(handle1);
+
+  return 0;
 }
 
 int KernelProxy::AcquireSocketHandle(int fd, ScopedKernelHandle* handle) {
diff --git a/native_client_sdk/src/libraries/nacl_io/library.dsc b/native_client_sdk/src/libraries/nacl_io/library.dsc
index 733d804..0000497 100644
--- a/native_client_sdk/src/libraries/nacl_io/library.dsc
+++ b/native_client_sdk/src/libraries/nacl_io/library.dsc
@@ -63,6 +63,8 @@
         "socket/tcp_node.cc",
         "socket/udp_event_emitter.cc",
         "socket/udp_node.cc",
+        "socket/unix_event_emitter.cc",
+        "socket/unix_node.cc",
         "stream/stream_event_emitter.cc",
         "stream/stream_fs.cc",
         "stream/stream_node.cc",
@@ -210,6 +212,8 @@
         "socket/tcp_node.h",
         "socket/udp_event_emitter.h",
         "socket/udp_node.h",
+        "socket/unix_event_emitter.h",
+        "socket/unix_node.h",
         "stream/stream_event_emitter.h",
         "stream/stream_fs.h",
         "stream/stream_node.h",
diff --git a/native_client_sdk/src/libraries/nacl_io/socket/socket_node.cc b/native_client_sdk/src/libraries/nacl_io/socket/socket_node.cc
index 5ff6f9d..b3517de 100644
--- a/native_client_sdk/src/libraries/nacl_io/socket/socket_node.cc
+++ b/native_client_sdk/src/libraries/nacl_io/socket/socket_node.cc
@@ -375,6 +375,8 @@ Error SocketNode::RecvFrom(const HandleAttr& attr,
                            socklen_t* addrlen,
                            int* out_len) {
   PP_Resource addr = 0;
+  if (0 == socket_resource_)
+    return EBADF;
   Error err = RecvHelper(attr, buf, len, flags, &addr, out_len);
   if (0 == err && 0 != addr) {
     if (src_addr)
@@ -392,9 +394,6 @@ Error SocketNode::RecvHelper(const HandleAttr& attr,
                              int flags,
                              PP_Resource* addr,
                              int* out_len) {
-  if (0 == socket_resource_)
-    return EBADF;
-
   if (TestStreamFlags(SSF_RECV_ENDOFSTREAM)) {
     *out_len = 0;
     return 0;
@@ -429,6 +428,11 @@ Error SocketNode::Send(const HandleAttr& attr,
                        size_t len,
                        int flags,
                        int* out_len) {
+  if (0 == socket_resource_)
+    return EBADF;
+
+  if (0 == remote_addr_)
+    return ENOTCONN;
   return SendHelper(attr, buf, len, flags, remote_addr_, out_len);
 }
 
@@ -446,6 +450,9 @@ Error SocketNode::SendTo(const HandleAttr& attr,
   if (0 == addr)
     return EINVAL;
 
+  if (0 == socket_resource_)
+    return EBADF;
+
   Error err = SendHelper(attr, buf, len, flags, addr, out_len);
   filesystem_->ppapi()->ReleaseResource(addr);
   return err;
@@ -457,12 +464,6 @@ Error SocketNode::SendHelper(const HandleAttr& attr,
                              int flags,
                              PP_Resource addr,
                              int* out_len) {
-  if (0 == socket_resource_)
-    return EBADF;
-
-  if (0 == addr)
-    return ENOTCONN;
-
   int ms = write_timeout_;
   if ((flags & MSG_DONTWAIT) || !attr.IsBlocking())
     ms = 0;
diff --git a/native_client_sdk/src/libraries/nacl_io/socket/unix_event_emitter.cc b/native_client_sdk/src/libraries/nacl_io/socket/unix_event_emitter.cc
new file mode 100644
index 0000000..1c3a523
--- /dev/null
+++ b/native_client_sdk/src/libraries/nacl_io/socket/unix_event_emitter.cc
@@ -0,0 +1,102 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "nacl_io/socket/unix_event_emitter.h"
+
+#include <stdlib.h>
+
+#include "nacl_io/fifo_char.h"
+#include "sdk_util/scoped_ref.h"
+
+namespace nacl_io {
+
+class UnixChildEventEmitter;
+class UnixMasterEventEmitter;
+
+typedef sdk_util::ScopedRef<UnixMasterEventEmitter>
+    ScopedUnixMasterEventEmitter;
+
+class UnixMasterEventEmitter : public UnixEventEmitter {
+ public:
+  explicit UnixMasterEventEmitter(size_t size)
+      : in_fifo_(size),
+        out_fifo_(size),
+        child_emitter_created_(false),
+        child_emitter_(NULL) {
+    UpdateStatus_Locked();
+  }
+
+  virtual ScopedUnixEventEmitter GetPeerEmitter();
+
+ protected:
+  virtual FIFOChar* in_fifoc() { return &in_fifo_; }
+  virtual FIFOChar* out_fifoc() { return &out_fifo_; }
+  virtual const sdk_util::SimpleLock& GetFifoLock() { return fifo_lock_; }
+
+ private:
+  FIFOChar in_fifo_;
+  FIFOChar out_fifo_;
+  sdk_util::SimpleLock fifo_lock_;
+  bool child_emitter_created_;
+  UnixChildEventEmitter* child_emitter_;
+
+  friend class UnixChildEventEmitter;
+};
+
+class UnixChildEventEmitter : public UnixEventEmitter {
+ public:
+  explicit UnixChildEventEmitter(UnixMasterEventEmitter* parent)
+      : parent_emitter_(parent) {
+    UpdateStatus_Locked();
+  }
+  virtual ScopedUnixEventEmitter GetPeerEmitter() { return parent_emitter_; }
+
+ protected:
+  virtual void Destroy() { parent_emitter_->child_emitter_ = NULL; }
+
+  virtual FIFOChar* in_fifoc() { return parent_emitter_->out_fifoc(); }
+  virtual FIFOChar* out_fifoc() { return parent_emitter_->in_fifoc(); }
+  virtual const sdk_util::SimpleLock& GetFifoLock() {
+    return parent_emitter_->GetFifoLock();
+  }
+
+ private:
+  ScopedUnixMasterEventEmitter parent_emitter_;
+};
+
+ScopedUnixEventEmitter UnixMasterEventEmitter::GetPeerEmitter() {
+  if (!child_emitter_created_) {
+    child_emitter_created_ = true;
+    child_emitter_ = new UnixChildEventEmitter(this);
+  }
+  return sdk_util::ScopedRef<UnixChildEventEmitter>(child_emitter_);
+}
+
+uint32_t UnixEventEmitter::ReadIn_Locked(char* data, uint32_t len) {
+  AUTO_LOCK(GetFifoLock());
+  uint32_t count = in_fifoc()->Read(data, len);
+  ScopedUnixEventEmitter peer = GetPeerEmitter();
+  if (peer) {
+    peer->UpdateStatus_Locked();
+  }
+  UpdateStatus_Locked();
+  return count;
+}
+
+uint32_t UnixEventEmitter::WriteOut_Locked(const char* data, uint32_t len) {
+  AUTO_LOCK(GetFifoLock());
+  uint32_t count = out_fifoc()->Write(data, len);
+  ScopedUnixEventEmitter peer = GetPeerEmitter();
+  if (peer) {
+    peer->UpdateStatus_Locked();
+  }
+  UpdateStatus_Locked();
+  return count;
+}
+
+ScopedUnixEventEmitter UnixEventEmitter::MakeUnixEventEmitter(size_t size) {
+  return ScopedUnixEventEmitter(new UnixMasterEventEmitter(size));
+}
+
+}  // namespace nacl_io
diff --git a/native_client_sdk/src/libraries/nacl_io/socket/unix_event_emitter.h b/native_client_sdk/src/libraries/nacl_io/socket/unix_event_emitter.h
new file mode 100644
index 0000000..f8c952d
--- /dev/null
+++ b/native_client_sdk/src/libraries/nacl_io/socket/unix_event_emitter.h
@@ -0,0 +1,50 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef LIBRARIES_NACL_IO_SOCKET_UNIX_EVENT_EMITTER_H_
+#define LIBRARIES_NACL_IO_SOCKET_UNIX_EVENT_EMITTER_H_
+
+#include "nacl_io/fifo_char.h"
+#include "nacl_io/stream/stream_event_emitter.h"
+
+#include "sdk_util/macros.h"
+#include "sdk_util/scoped_ref.h"
+#include "sdk_util/simple_lock.h"
+
+namespace nacl_io {
+
+class UnixEventEmitter;
+class Node;
+
+typedef sdk_util::ScopedRef<UnixEventEmitter> ScopedUnixEventEmitter;
+
+class UnixEventEmitter : public StreamEventEmitter {
+ public:
+  uint32_t ReadIn_Locked(char* buffer, uint32_t len);
+  uint32_t WriteOut_Locked(const char* buffer, uint32_t len);
+
+  uint32_t BytesInOutputFIFO();
+  uint32_t SpaceInInputFIFO();
+
+  virtual ScopedUnixEventEmitter GetPeerEmitter() = 0;
+
+  static ScopedUnixEventEmitter MakeUnixEventEmitter(size_t size);
+
+ protected:
+  UnixEventEmitter() {}
+
+  // Probably only need the master's lock.
+  virtual const sdk_util::SimpleLock& GetFifoLock() = 0;
+  virtual FIFOInterface* in_fifo() { return in_fifoc(); }
+  virtual FIFOInterface* out_fifo() { return out_fifoc(); }
+  virtual FIFOChar* in_fifoc() = 0;
+  virtual FIFOChar* out_fifoc() = 0;
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(UnixEventEmitter);
+};
+
+}  // namespace nacl_io
+
+#endif  // LIBRARIES_NACL_IO_SOCKET_UNIX_EVENT_EMITTER_H_
diff --git a/native_client_sdk/src/libraries/nacl_io/socket/unix_node.cc b/native_client_sdk/src/libraries/nacl_io/socket/unix_node.cc
new file mode 100644
index 0000000..c5675a6
--- /dev/null
+++ b/native_client_sdk/src/libraries/nacl_io/socket/unix_node.cc
@@ -0,0 +1,94 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "nacl_io/ossocket.h"
+#ifdef PROVIDES_SOCKET_API
+
+#include <assert.h>
+#include <string.h>
+#include <algorithm>
+
+#include "nacl_io/socket/unix_event_emitter.h"
+#include "nacl_io/socket/unix_node.h"
+
+namespace nacl_io {
+
+UnixNode::UnixNode(Filesystem* filesystem)
+    : SocketNode(filesystem),
+      emitter_(UnixEventEmitter::MakeUnixEventEmitter(65536)) {
+  emitter_->AttachStream(this);
+}
+
+UnixNode::UnixNode(Filesystem* filesystem, const UnixNode& peer)
+    : SocketNode(filesystem), emitter_(peer.emitter_->GetPeerEmitter()) {
+  emitter_->AttachStream(this);
+}
+
+EventEmitter* UnixNode::GetEventEmitter() {
+  return emitter_.get();
+}
+
+Error UnixNode::Recv_Locked(void* buffer,
+                            size_t len,
+                            PP_Resource* out_addr,
+                            int* out_len) {
+  assert(emitter_.get());
+  *out_len = emitter_->ReadIn_Locked((char*)buffer, len);
+  *out_addr = 0;
+  return 0;
+}
+
+Error UnixNode::Send_Locked(const void* buffer,
+                            size_t len,
+                            PP_Resource out_addr,
+                            int* out_len) {
+  assert(emitter_.get());
+  *out_len = emitter_->WriteOut_Locked((char*)buffer, len);
+  return 0;
+}
+
+Error UnixNode::RecvFrom(const HandleAttr& attr,
+                         void* buf,
+                         size_t len,
+                         int flags,
+                         struct sockaddr* src_addr,
+                         socklen_t* addrlen,
+                         int* out_len) {
+  PP_Resource addr = 0;
+  Error err = RecvHelper(attr, buf, len, flags, &addr, out_len);
+  if (0 == err) {
+    if (src_addr) {
+      unsigned short family = AF_UNIX;
+      memcpy(src_addr, &family,
+             std::min(*addrlen, static_cast<socklen_t>(sizeof(family))));
+      *addrlen = sizeof(family);
+    }
+  }
+
+  return err;
+}
+
+Error UnixNode::Send(const HandleAttr& attr,
+                     const void* buf,
+                     size_t len,
+                     int flags,
+                     int* out_len) {
+  PP_Resource addr = 0;
+  return SendHelper(attr, buf, len, flags, addr, out_len);
+}
+
+Error UnixNode::SendTo(const HandleAttr& attr,
+                       const void* buf,
+                       size_t len,
+                       int flags,
+                       const struct sockaddr* dest_addr,
+                       socklen_t addrlen,
+                       int* out_len) {
+  PP_Resource addr = 0;
+  return SendHelper(attr, buf, len, flags, addr, out_len);
+}
+
+}  // namespace nacl_io
+
+#endif  // PROVIDES_SOCKET_API
diff --git a/native_client_sdk/src/libraries/nacl_io/socket/unix_node.h b/native_client_sdk/src/libraries/nacl_io/socket/unix_node.h
new file mode 100644
index 0000000..c8b7f30
--- /dev/null
+++ b/native_client_sdk/src/libraries/nacl_io/socket/unix_node.h
@@ -0,0 +1,60 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef LIBRARIES_NACL_IO_SOCKET_UNIX_NODE_H_
+#define LIBRARIES_NACL_IO_SOCKET_UNIX_NODE_H_
+
+#include "nacl_io/ossocket.h"
+#ifdef PROVIDES_SOCKET_API
+
+#include "nacl_io/node.h"
+#include "nacl_io/socket/socket_node.h"
+#include "nacl_io/socket/unix_event_emitter.h"
+
+namespace nacl_io {
+
+class UnixNode : public SocketNode {
+ public:
+  explicit UnixNode(Filesystem* filesystem);
+  UnixNode(Filesystem* filesystem, const UnixNode& peer);
+
+  virtual EventEmitter* GetEventEmitter();
+
+ protected:
+  virtual Error Recv_Locked(void* buffer,
+                            size_t len,
+                            PP_Resource* out_addr,
+                            int* out_len);
+  virtual Error Send_Locked(const void* buffer,
+                            size_t len,
+                            PP_Resource addr,
+                            int* out_len);
+  virtual Error RecvFrom(const HandleAttr& attr,
+                         void* buf,
+                         size_t len,
+                         int flags,
+                         struct sockaddr* src_addr,
+                         socklen_t* addrlen,
+                         int* out_len);
+  virtual Error Send(const HandleAttr& attr,
+                     const void* buf,
+                     size_t len,
+                     int flags,
+                     int* out_len);
+  virtual Error SendTo(const HandleAttr& attr,
+                       const void* buf,
+                       size_t len,
+                       int flags,
+                       const struct sockaddr* dest_addr,
+                       socklen_t addrlen,
+                       int* out_len);
+
+ private:
+  ScopedUnixEventEmitter emitter_;
+};
+
+}  // namespace nacl_io
+
+#endif  // PROVIDES_SOCKET_API
+#endif  // LIBRARIES_NACL_IO_SOCKET_UNIX_NODE_H_
diff --git a/native_client_sdk/src/tests/nacl_io_test/kernel_proxy_test.cc b/native_client_sdk/src/tests/nacl_io_test/kernel_proxy_test.cc
index 24a0fd1..52f6eeb 100644
--- a/native_client_sdk/src/tests/nacl_io_test/kernel_proxy_test.cc
+++ b/native_client_sdk/src/tests/nacl_io_test/kernel_proxy_test.cc
@@ -1116,6 +1116,7 @@ TEST_F(KernelProxyErrorTest, ReadError) {
       .WillOnce(DoAll(SetArgPointee<3>(0),  // Read 0 bytes.
                       Return(1234)));       // Returned error 1234.
 
+  EXPECT_CALL(*mock_node, GetType()).WillRepeatedly(Return(S_IFDIR));
   EXPECT_CALL(*mock_node, Destroy()).Times(1);
 
   int fd = ki_open("/dummy", O_RDONLY, 0);
diff --git a/native_client_sdk/src/tests/nacl_io_test/mock_node.cc b/native_client_sdk/src/tests/nacl_io_test/mock_node.cc
index cc658c7..02a752b 100644
--- a/native_client_sdk/src/tests/nacl_io_test/mock_node.cc
+++ b/native_client_sdk/src/tests/nacl_io_test/mock_node.cc
@@ -4,6 +4,8 @@
 
 #include "mock_node.h"
 
+using ::testing::Return;
+
 MockNode::MockNode(nacl_io::Filesystem* mount) : Node(mount) {}
 
 MockNode::~MockNode() {}
diff --git a/native_client_sdk/src/tests/nacl_io_test/socket_test.cc b/native_client_sdk/src/tests/nacl_io_test/socket_test.cc
index 03d9637..e2d86d5 100644
--- a/native_client_sdk/src/tests/nacl_io_test/socket_test.cc
+++ b/native_client_sdk/src/tests/nacl_io_test/socket_test.cc
@@ -243,23 +243,113 @@ TEST_F(SocketTest, Shutdown) {
   EXPECT_EQ(errno, ENOTSOCK);
 }
 
-TEST_F(SocketTest, Socket) {
-  EXPECT_LT(ki_socket(AF_UNIX, SOCK_STREAM, 0), 0);
-  EXPECT_EQ(errno, EAFNOSUPPORT);
+TEST_F(SocketTest, SocketInetRawUnsupported) {
   EXPECT_LT(ki_socket(AF_INET, SOCK_RAW, 0), 0);
   EXPECT_EQ(errno, EPROTONOSUPPORT);
 }
 
-TEST_F(SocketTest, Socketpair) {
+TEST_F(SocketTest, SocketpairUnsupported) {
   int sv[2];
   EXPECT_LT(ki_socketpair(AF_INET, SOCK_STREAM, 0, NULL), 0);
   EXPECT_EQ(errno, EFAULT);
-  EXPECT_LT(ki_socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
-  EXPECT_EQ(errno, EAFNOSUPPORT);
   EXPECT_LT(ki_socketpair(AF_INET, SOCK_STREAM, 0, sv), 0);
-  EXPECT_EQ(errno, EPROTONOSUPPORT);
+  EXPECT_EQ(errno, EOPNOTSUPP);
   EXPECT_LT(ki_socketpair(AF_INET6, SOCK_STREAM, 0, sv), 0);
-  EXPECT_EQ(errno, EPROTONOSUPPORT);
+  EXPECT_EQ(errno, EOPNOTSUPP);
+  EXPECT_LT(ki_socketpair(AF_UNIX, SOCK_DGRAM, 0, sv), 0);
+  EXPECT_EQ(errno, EPROTOTYPE);
+  EXPECT_LT(ki_socketpair(AF_MAX, SOCK_STREAM, 0, sv), 0);
+  EXPECT_EQ(errno, EAFNOSUPPORT);
+}
+
+class UnixSocketTest : public ::testing::Test {
+ public:
+  UnixSocketTest() { sv_[0] = sv_[1] = -1; }
+
+  void SetUp() {
+    ASSERT_EQ(0, ki_push_state_for_testing());
+    ASSERT_EQ(0, ki_init(&kp_));
+  }
+
+  void TearDown() {
+    if (sv_[0] != -1)
+      EXPECT_EQ(0, ki_close(sv_[0]));
+    if (sv_[1] != -1)
+      EXPECT_EQ(0, ki_close(sv_[1]));
+    ki_uninit();
+  }
+
+ protected:
+  KernelProxy kp_;
+
+  int sv_[2];
+};
+
+TEST_F(UnixSocketTest, Socket) {
+  EXPECT_EQ(-1, ki_socket(AF_UNIX, SOCK_STREAM, 0));
+  EXPECT_EQ(EAFNOSUPPORT, errno);
+}
+
+TEST_F(UnixSocketTest, Socketpair) {
+  errno = 0;
+  EXPECT_EQ(0, ki_socketpair(AF_UNIX, SOCK_STREAM, 0, sv_));
+  EXPECT_EQ(0, errno);
+  EXPECT_LE(0, sv_[0]);
+  EXPECT_LE(0, sv_[1]);
+}
+
+TEST_F(UnixSocketTest, SendRecv) {
+  char outbuf[256];
+  char inbuf[512];
+
+  memset(outbuf, 0xA5, sizeof(outbuf));
+  memset(inbuf, 0x3C, sizeof(inbuf));
+
+  EXPECT_EQ(0, ki_socketpair(AF_UNIX, SOCK_STREAM, 0, sv_));
+
+  int len1 = ki_send(sv_[0], outbuf, sizeof(outbuf), /* flags */ 0);
+  EXPECT_EQ(sizeof(outbuf), len1);
+
+  // The buffers should be different.
+  EXPECT_NE(0, memcmp(outbuf, inbuf, sizeof(outbuf)));
+
+  int len2 = ki_recv(sv_[1], inbuf, sizeof(inbuf), /* flags */ 0);
+  EXPECT_EQ(sizeof(outbuf), len2);
+
+  EXPECT_EQ(0, memcmp(outbuf, inbuf, sizeof(outbuf)));
+
+  // A reader should block after to read at this point.
+  EXPECT_EQ(-1, ki_recv(sv_[1], inbuf, sizeof(inbuf), MSG_DONTWAIT));
+  EXPECT_EQ(EAGAIN, errno);
+
+  // Send data back in the opposite direction.
+  memset(inbuf, 0x3C, sizeof(inbuf));
+  EXPECT_NE(0, memcmp(outbuf, inbuf, sizeof(outbuf)));
+  len1 = ki_send(sv_[1], outbuf, sizeof(outbuf), /* flags */ 0);
+  EXPECT_EQ(sizeof(outbuf), len1);
+
+  EXPECT_NE(0, memcmp(outbuf, inbuf, sizeof(outbuf)));
+
+  len2 = ki_recv(sv_[0], inbuf, sizeof(inbuf), /* flags */ 0);
+  EXPECT_EQ(sizeof(outbuf), len2);
+
+  EXPECT_EQ(0, memcmp(outbuf, inbuf, sizeof(outbuf)));
+  EXPECT_EQ(-1, ki_recv(sv_[0], inbuf, sizeof(inbuf), MSG_DONTWAIT));
+  EXPECT_EQ(EAGAIN, errno);
+}
+
+TEST_F(UnixSocketTest, RecvNonBlocking) {
+  char buf[128];
+
+  EXPECT_EQ(0, ki_socketpair(AF_UNIX, SOCK_STREAM, 0, sv_));
+
+  EXPECT_EQ(-1, ki_recv(sv_[0], buf, sizeof(buf), MSG_DONTWAIT));
+  EXPECT_EQ(EAGAIN, errno);
+
+  struct pollfd pollfd = {sv_[0], POLLIN | POLLOUT, 0};
+  EXPECT_EQ(1, ki_poll(&pollfd, 1, 0));
+  EXPECT_EQ(POLLOUT, pollfd.revents & POLLOUT);
+  EXPECT_NE(POLLIN, pollfd.revents & POLLIN);
 }
 
 TEST(SocketUtilityFunctions, Htonl) {