[NaCl SDK] Support non blocking TCP/UDP

native_client_sdk/src/tests/nacl_io_test/event_test.cc

 /* Copyright (c) 2013 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 <errno.h>
 #include <fcntl.h>
+#include <pthread.h>
 #include <stdio.h>
 #include <sys/ioctl.h>
 #include <sys/stat.h>
 #include <sys/time.h>
 
 #include "gtest/gtest.h"
 
 #include "nacl_io/event_emitter.h"
 #include "nacl_io/event_listener.h"
+#include "nacl_io/event_listener.h"
+#include "nacl_io/event_listener.h"
 #include "nacl_io/kernel_intercept.h"
 #include "nacl_io/kernel_proxy.h"
 #include "nacl_io/kernel_wrap.h"
+#include "nacl_io/mount_node_pipe.h"
+#include "nacl_io/mount_stream.h"
+
+#include "ppapi_simple/ps.h"
 
 
 using namespace nacl_io;
 using namespace sdk_util;
 
-class EventEmitterTester : public MountNode {
- public:
-  EventEmitterTester() : MountNode(NULL), event_status_(0), event_cnt_(0) {}
-
-  void SetEventStatus(uint32_t bits) { event_status_ = bits; }
-  uint32_t GetEventStatus() { return event_status_; }
-
-  Error Ioctl(int request, char* arg) {
-    event_status_ = static_cast<uint32_t>(request);
-    return 0;
-  }
-
-  int GetType() { return S_IFSOCK; }
-  int NumEvents() { return event_cnt_; }
-
- public:
-  // Make this function public for testing
-  void RaiseEvent(uint32_t events) {
-    EventEmitter::RaiseEvent(events);
-  }
-
-    // Called after registering locally, but while lock is still held.
-  void ChainRegisterEventInfo(const ScopedEventInfo& event) {
-    event_cnt_++;
-  }
-
-  // Called before unregistering locally, but while lock is still held.
-  void ChainUnregisterEventInfo(const ScopedEventInfo& event) {
-    event_cnt_--;
+
+class EventListenerTester : public EventListener {
+ public:
+  EventListenerTester() : EventListener(), events_(0) {};
+
+  virtual void ReceiveEvents(EventEmitter* emitter, uint32_t events) {
+    events_ |= events;
+  }
+
+  uint32_t Events() {
+    return events_;
+  }
+
+  void Clear() {
+    events_ = 0;
+  }
+
+  uint32_t events_;
+};
+
+
+TEST(EmitterBasic, SingleThread) {
+  EventListenerTester listener_a;
+  EventListenerTester listener_b;
+  EventEmitter emitter;
+
+  emitter.RegisterListener(&listener_a, POLLIN | POLLOUT | POLLERR);
+  emitter.RegisterListener(&listener_b, POLLIN | POLLOUT | POLLERR);
+
+  EXPECT_EQ(0, emitter.GetEventStatus());
+  EXPECT_EQ(0, listener_a.Events());
+
+  {
+    AUTO_LOCK(emitter.GetLock())
+    emitter.RaiseEvents_Locked(POLLIN);
+  }
+  EXPECT_EQ(POLLIN, listener_a.Events());
+
+  listener_a.Clear();
+
+  {
+    AUTO_LOCK(emitter.GetLock())
+    emitter.RaiseEvents_Locked(POLLOUT);
+  }
+  EXPECT_EQ(POLLOUT, listener_a.Events());
+  EXPECT_EQ(POLLIN | POLLOUT, listener_b.Events());
+}
+
+class EmitterTest : public ::testing::Test {
+ public:
+  void SetUp() {
+    pthread_cond_destroy(&multi_cond_);

[binji, 2013/09/15 22:18:58]

pthread_cond_init

[noelallen1, 2013/09/17 21:21:54]

Done.

+    waiting_ = 0;
+    signaled_ = 0;
+  }
+
+  void TearDown() {
+    pthread_cond_destroy(&multi_cond_);
+  }
+
+  static void* ThreadThunk(void *ptr) {
+    return static_cast<EmitterTest*>(ptr)->ThreadEntry();
+  }
+
+  void CreateThread() {

[binji, 2013/09/15 22:18:58]

It would be clearer if the function order were: CreateThread, ThreadThunk,
ThreadEntry.

[noelallen1, 2013/09/17 21:21:54]

Done.

+    pthread_t id;
+    EXPECT_EQ(0, pthread_create(&id, NULL, ThreadThunk, this));
+  }
+
+  void* ThreadEntry() {
+    EventListenerLock listener(&emitter_);
+
+    pthread_cond_signal(&multi_cond_);
+    waiting_++;

[binji, 2013/09/15 22:18:58]

atomic increment?

[noelallen1, 2013/09/17 21:21:54]

I'm holding the emitter lock.

+    EXPECT_EQ(0, listener.WaitOnEvent(POLLIN, -1));
+    emitter_.ClearEvents_Locked(POLLIN);
+    signaled_ ++;
+    return NULL;
   }
 
  protected:
-  uint32_t event_status_;
-  uint32_t event_cnt_;
-};
-
-
-const int MAX_EVENTS = 8;
-
-// IDs for Emitters
-const int ID_EMITTER = 5;
-const int ID_LISTENER = 6;
-const int ID_EMITTER_DUP = 7;
-
-// Kernel Event values
-const uint32_t KE_EXPECTED = 4;
-const uint32_t KE_FILTERED = 2;
-const uint32_t KE_NONE = 0;
-
-// User Data values
-const uint64_t USER_DATA_A = 1;
-const uint64_t USER_DATA_B = 5;
-
-// Timeout durations
-const int TIMEOUT_IMMEDIATE = 0;
-const int TIMEOUT_SHORT= 100;
-const int TIMEOUT_LONG = 500;
-const int TIMEOUT_NEVER = -1;
-const int TIMEOUT_VERY_LONG = 1000;
-
-// We subtract TIMEOUT_SLOP from the expected minimum timed due to rounding
-// and clock drift converting between absolute and relative time.  This should
-// only be 1 for Less Than, and 1 for rounding, but we use 10 since we don't
-// care about real precision, aren't testing of the underlying
-// implementations and don't want flakiness.
-const int TIMEOUT_SLOP = 10;
-
-TEST(EventTest, EmitterBasic) {
-  ScopedRef<EventEmitterTester> emitter(new EventEmitterTester());
-  ScopedRef<EventEmitter> null_emitter;
-
-  ScopedEventListener listener(new EventListener);
-
-  // Verify construction
-  EXPECT_EQ(0, emitter->NumEvents());
-  EXPECT_EQ(0, emitter->GetEventStatus());
-
-  // Verify status
-  emitter->SetEventStatus(KE_EXPECTED);
-  EXPECT_EQ(KE_EXPECTED, emitter->GetEventStatus());
-
-  // Fail to update or free an ID not in the set
-  EXPECT_EQ(ENOENT, listener->Update(ID_EMITTER, KE_EXPECTED, USER_DATA_A));
-  EXPECT_EQ(ENOENT, listener->Free(ID_EMITTER));
-
-  // Fail to Track self
-  EXPECT_EQ(EINVAL, listener->Track(ID_LISTENER,
-                                    listener,
-                                    KE_EXPECTED,
-                                    USER_DATA_A));
-
-  // Set the emitter filter and data
-  EXPECT_EQ(0, listener->Track(ID_EMITTER, emitter, KE_EXPECTED, USER_DATA_A));
-  EXPECT_EQ(1, emitter->NumEvents());
-
-  // Fail to add the same ID
-  EXPECT_EQ(EEXIST,
-            listener->Track(ID_EMITTER, emitter, KE_EXPECTED, USER_DATA_A));
-  EXPECT_EQ(1, emitter->NumEvents());
-
-  int event_cnt = 0;
-  EventData ev[MAX_EVENTS];
-
-  // Do not allow a wait with a zero events count.
-  EXPECT_EQ(EINVAL, listener->Wait(ev, 0, TIMEOUT_IMMEDIATE, &event_cnt));
-
-  // Do not allow a wait with a negative events count.
-  EXPECT_EQ(EINVAL, listener->Wait(ev, -1, TIMEOUT_IMMEDIATE, &event_cnt));
-
-  // Do not allow a wait with a NULL EventData pointer
-  EXPECT_EQ(EFAULT,
-            listener->Wait(NULL, MAX_EVENTS, TIMEOUT_IMMEDIATE, &event_cnt));
-
-  // Return with no events if the Emitter has no signals set.
-  memset(ev, 0, sizeof(ev));
-  event_cnt = 100;
-  emitter->SetEventStatus(KE_NONE);
-  EXPECT_EQ(0, listener->Wait(ev, MAX_EVENTS, TIMEOUT_IMMEDIATE, &event_cnt));
-  EXPECT_EQ(0, event_cnt);
-
-  // Return with no events if the Emitter has a filtered signals set.
-  memset(ev, 0, sizeof(ev));
-  event_cnt = 100;
-  emitter->SetEventStatus(KE_FILTERED);
-  EXPECT_EQ(0, listener->Wait(ev, MAX_EVENTS, TIMEOUT_IMMEDIATE, &event_cnt));
-  EXPECT_EQ(0, event_cnt);
-
-  // Return with one event if the Emitter has the expected signal set.
-  memset(ev, 0, sizeof(ev));
-  event_cnt = 100;
-  emitter->SetEventStatus(KE_EXPECTED);
-  EXPECT_EQ(0, listener->Wait(ev, MAX_EVENTS, TIMEOUT_IMMEDIATE, &event_cnt));
-  EXPECT_EQ(1, event_cnt);
-  EXPECT_EQ(USER_DATA_A, ev[0].user_data);
-  EXPECT_EQ(KE_EXPECTED, ev[0].events);
-
-  // Return with one event containing only the expected signal.
-  memset(ev, 0, sizeof(ev));
-  event_cnt = 100;
-  emitter->SetEventStatus(KE_EXPECTED | KE_FILTERED);
-  EXPECT_EQ(0, listener->Wait(ev, MAX_EVENTS, TIMEOUT_IMMEDIATE, &event_cnt));
-  EXPECT_EQ(1, event_cnt);
-  EXPECT_EQ(USER_DATA_A, ev[0].user_data);
-  EXPECT_EQ(KE_EXPECTED, ev[0].events);
-
-  // Change the USER_DATA on an existing event
-  EXPECT_EQ(0, listener->Update(ID_EMITTER, KE_EXPECTED, USER_DATA_B));
-
-  // Return with one event signaled with the alternate USER DATA
-  memset(ev, 0, sizeof(ev));
-  event_cnt = 100;
-  emitter->SetEventStatus(KE_EXPECTED | KE_FILTERED);
-  EXPECT_EQ(0, listener->Wait(ev, MAX_EVENTS, 0, &event_cnt));
-  EXPECT_EQ(1, event_cnt);
-  EXPECT_EQ(USER_DATA_B, ev[0].user_data);
-  EXPECT_EQ(KE_EXPECTED, ev[0].events);
-
-  // Reset the USER_DATA.
-  EXPECT_EQ(0, listener->Update(ID_EMITTER, KE_EXPECTED, USER_DATA_A));
-
-  // Support adding a DUP.
-  EXPECT_EQ(0, listener->Track(ID_EMITTER_DUP,
-                               emitter,
-                               KE_EXPECTED,
-                               USER_DATA_A));
-  EXPECT_EQ(2, emitter->NumEvents());
-
-  // Return unsignaled.
-  memset(ev, 0, sizeof(ev));
-  emitter->SetEventStatus(KE_NONE);
-  event_cnt = 100;
-  EXPECT_EQ(0, listener->Wait(ev, MAX_EVENTS, TIMEOUT_IMMEDIATE, &event_cnt));
-  EXPECT_EQ(0, event_cnt);
-
-  // Return with two event signaled with expected data.
-  memset(ev, 0, sizeof(ev));
-  emitter->SetEventStatus(KE_EXPECTED);
-  event_cnt = 100;
-  EXPECT_EQ(0, listener->Wait(ev, MAX_EVENTS, TIMEOUT_IMMEDIATE, &event_cnt));
-  EXPECT_EQ(2, event_cnt);
-  EXPECT_EQ(USER_DATA_A, ev[0].user_data);
-  EXPECT_EQ(KE_EXPECTED, ev[0].events);
-  EXPECT_EQ(USER_DATA_A, ev[1].user_data);
-  EXPECT_EQ(KE_EXPECTED, ev[1].events);
-}
-
-long Duration(struct timeval* start, struct timeval* end) {
-  if (start->tv_usec > end->tv_usec) {
-    end->tv_sec -= 1;
-    end->tv_usec += 1000000;
-  }
-  long cur_time = 1000 * (end->tv_sec - start->tv_sec);
-  cur_time += (end->tv_usec - start->tv_usec) / 1000;
-  return cur_time;
-}
-
-
-// Run a timed wait, and return the average of 8 iterations to reduce
-// chance of false negative on outlier.
-const int TRIES_TO_AVERAGE = 8;
-bool TimedListen(ScopedEventListener& listen,
-                 EventData* ev,
-                 int ev_max,
-                 int ev_expect,
-                 int ms_wait,
-                 long* duration) {
-
-  struct timeval start;
-  struct timeval end;
-  long total_time = 0;
-
-  for (int a=0; a < TRIES_TO_AVERAGE; a++) {
-    gettimeofday(&start, NULL);
-
-    int signaled;
-
-    EXPECT_EQ(0, listen->Wait(ev, ev_max, ms_wait, &signaled));
-    EXPECT_EQ(signaled, ev_expect);
-
-    if (signaled != ev_expect) {
-      return false;
+  pthread_cond_t multi_cond_;
+  EventEmitter emitter_;
+
+  uint32_t waiting_;
+  uint32_t signaled_;
+};
+
+
+const int NUM_THREADS = 10;
+TEST_F(EmitterTest, MultiThread) {
+  for (int a=0; a <NUM_THREADS; a++)
+    CreateThread();
+
+  sleep(1);

[binji, 2013/09/15 22:18:58]

why?

[noelallen1, 2013/09/17 21:21:54]

Tests are written such that threads that should make progress, must make
progress for the test to move forward, otherwise the tester timeout triggers. 
The sleep give threads that shouldn't move forward a chance to move forward and
trigger a failure.   Removing the sleep would make the test run faster, but
reduces the chance of catching a race.

+  EXPECT_EQ(0, signaled_);
+
+  {
+    AUTO_LOCK(emitter_.GetLock());
+    // Wait for all threads to wait
+    while(waiting_ < NUM_THREADS)
+      pthread_cond_wait(&multi_cond_, emitter_.GetLock().mutex());
+
+    emitter_.RaiseEvents_Locked(POLLIN);
+  }
+
+  sleep(1);

[binji, 2013/09/15 22:18:58]

can we do better than sleeping here?

[noelallen1, 2013/09/17 21:21:54]

Not without serious changes to pthread_cond, and/or adding a wrapping class that
supports debugging.

+  EXPECT_EQ(1, signaled_);
+
+  {
+    AUTO_LOCK(emitter_.GetLock());
+    emitter_.RaiseEvents_Locked(POLLIN);
+  }
+
+  sleep(1);
+  EXPECT_EQ(2, signaled_);

[binji, 2013/09/15 22:18:58]

clean up other threads?

[noelallen1, 2013/09/17 21:21:54]

The other threads just return after running the test.

[binji, 2013/09/19 00:48:54]

Aren't they all blocked on WaitOnEvent?

[noelallen1, 2013/09/19 21:29:27]

Done.

+}
+
+
+TEST(PipeTest, Listener) {
+  const char hello[] = "Hello World.";
+  char tmp[64] = "Goodbye";
+
+  EventEmitterPipe pipe(32);
+
+  // Expect to time out on input.
+  {
+    EventListenerLock locker(&pipe);
+    EXPECT_EQ(ETIMEDOUT, locker.WaitOnEvent(POLLIN, 0));
+  }
+
+  // Output should be ready to go.
+  {
+    EventListenerLock locker(&pipe);
+    EXPECT_EQ(0, locker.WaitOnEvent(POLLOUT, 0));
+    EXPECT_EQ(sizeof(hello), pipe.Write_Locked(hello, sizeof(hello)));
+  }
+
+  // We should now be able to poll
+  {
+    EventListenerLock locker(&pipe);
+    EXPECT_EQ(0, locker.WaitOnEvent(POLLIN, 0));
+    EXPECT_EQ(sizeof(hello), pipe.Read_Locked(tmp, sizeof(tmp)));
+  }
+
+  // Verify we can read it correctly.
+  EXPECT_EQ(0, strcmp(hello, tmp));
+}
+
+
+class TestMountStream : public MountStream {
+ public:
+  TestMountStream() {}
+};
+
+TEST(PipeNodeTest, Basic) {
+  ScopedMount mnt(new TestMountStream());
+
+  MountNodePipe* pipe_node = new MountNodePipe(mnt.get());
+  ScopedRef<MountNodePipe> pipe(pipe_node);
+
+  EXPECT_EQ(POLLOUT, pipe_node->GetEventStatus());
+}
+
+const int MAX_FDS = 32;
+class SelectPollTest : public ::testing::Test {
+ public:
+  void SetUp() {
+    kp = new KernelProxy();
+    kp->Init(NULL);
+    EXPECT_EQ(0, kp->umount("/"));
+    EXPECT_EQ(0, kp->mount("", "/", "memfs", 0, NULL));
+
+    memset(&tv, 0, sizeof(tv));
+  }
+
+  void TearDown() {
+    delete kp;
+  }
+
+  void SetFDs(int* fds, int cnt) {
+    FD_ZERO(&rd_set);
+    FD_ZERO(&wr_set);
+    FD_ZERO(&ex_set);
+
+    for (int index = 0; index < cnt; index++) {
+      EXPECT_NE(-1, fds[index]);
+      FD_SET(fds[index], &rd_set);
+      FD_SET(fds[index], &wr_set);
+      FD_SET(fds[index], &ex_set);
+
+      pollfds[index].fd = fds[index];
+      pollfds[index].events = POLLIN | POLLOUT;
+      pollfds[index].revents = -1;
     }
-
-    gettimeofday(&end, NULL);
-
-    long cur_time = Duration(&start, &end);
-    total_time += cur_time;
-  }
-
-  *duration = total_time / TRIES_TO_AVERAGE;
-  return true;
-}
-
-
-// NOTE:  These timing tests are potentially flaky, the real test is
-// for the zero timeout should be, has the ConditionVariable been waited on?
-// Once we provide a debuggable SimpleCond and SimpleLock we can actually test
-// the correct thing.
-
-// Normal scheduling would expect us to see ~10ms accuracy, but we'll
-// use a much bigger number (yet smaller than the MAX_MS_TIMEOUT).
-const int SCHEDULING_GRANULARITY = 100;
-
-const int EXPECT_ONE_EVENT = 1;
-const int EXPECT_NO_EVENT = 0;
-
-TEST(EventTest, EmitterTimeout) {
-  ScopedRef<EventEmitterTester> emitter(new EventEmitterTester());
-  ScopedEventListener listener(new EventListener());
-  long duration;
-
-  EventData ev[MAX_EVENTS];
-  memset(ev, 0, sizeof(ev));
-  EXPECT_EQ(0, listener->Track(ID_EMITTER, emitter, KE_EXPECTED, USER_DATA_A));
-
-  // Return immediately when emitter is signaled, with no timeout
-  emitter->SetEventStatus(KE_EXPECTED);
-  memset(ev, 0, sizeof(ev));
-  EXPECT_TRUE(TimedListen(listener, ev, MAX_EVENTS, EXPECT_ONE_EVENT,
-                          TIMEOUT_IMMEDIATE, &duration));
-  EXPECT_EQ(USER_DATA_A, ev[0].user_data);
-  EXPECT_EQ(KE_EXPECTED, ev[0].events);
-  EXPECT_EQ(0, duration);
-
-  // Return immediately when emitter is signaled, even with timeout
-  emitter->SetEventStatus(KE_EXPECTED);
-  memset(ev, 0, sizeof(ev));
-  EXPECT_TRUE(TimedListen(listener, ev, MAX_EVENTS, EXPECT_ONE_EVENT,
-                          TIMEOUT_LONG, &duration));
-  EXPECT_EQ(USER_DATA_A, ev[0].user_data);
-  EXPECT_EQ(KE_EXPECTED, ev[0].events);
-  EXPECT_GT(SCHEDULING_GRANULARITY, duration);
-
-  // Return immediately if Emiiter is already signaled when blocking forever.
-  emitter->SetEventStatus(KE_EXPECTED);
-  memset(ev, 0, sizeof(ev));
-  EXPECT_TRUE(TimedListen(listener, ev, MAX_EVENTS, EXPECT_ONE_EVENT,
-                          TIMEOUT_NEVER, &duration));
-  EXPECT_EQ(USER_DATA_A, ev[0].user_data);
-  EXPECT_EQ(KE_EXPECTED, ev[0].events);
-  EXPECT_GT(SCHEDULING_GRANULARITY, duration);
-
-  // Return immediately if Emitter is no signaled when not blocking.
-  emitter->SetEventStatus(KE_NONE);
-  memset(ev, 0, sizeof(ev));
-  EXPECT_TRUE(TimedListen(listener, ev, MAX_EVENTS, EXPECT_NO_EVENT,
-                          TIMEOUT_IMMEDIATE, &duration));
-  EXPECT_EQ(0, duration);
-
-  // Wait TIMEOUT_LONG if the emitter is not in a signaled state.
-  emitter->SetEventStatus(KE_NONE);
-  memset(ev, 0, sizeof(ev));
-  EXPECT_TRUE(TimedListen(listener, ev, MAX_EVENTS, EXPECT_NO_EVENT,
-                          TIMEOUT_LONG, &duration));
-  EXPECT_LT(TIMEOUT_LONG - TIMEOUT_SLOP, duration);
-  EXPECT_GT(TIMEOUT_LONG + SCHEDULING_GRANULARITY, duration);
-}
-
-struct SignalInfo {
-  EventEmitterTester* em;
-  unsigned int ms_wait;
-  uint32_t events;
-};
-
-static void *SignalEmitterThread(void *ptr) {
-  SignalInfo* info = (SignalInfo*) ptr;
-  struct timespec ts;
-  ts.tv_sec = 0;
-  ts.tv_nsec = info->ms_wait * 1000000;
-
-  nanosleep(&ts, NULL);
-
-  info->em->RaiseEvent(info->events);
-  return NULL;
-}
-
-TEST(EventTest, EmitterSignalling) {
-  ScopedRef<EventEmitterTester> emitter(new EventEmitterTester());
-  ScopedEventListener listener(new EventListener);
-
-  SignalInfo siginfo;
-  struct timeval start;
-  struct timeval end;
-  long duration;
-
-  EventData ev[MAX_EVENTS];
-  memset(ev, 0, sizeof(ev));
-  EXPECT_EQ(0, listener->Track(ID_EMITTER, emitter, KE_EXPECTED, USER_DATA_A));
-
-  // Setup another thread to wait 1/4 of the max time, and signal both
-  // an expected, and unexpected value.
-  siginfo.em = emitter.get();
-  siginfo.ms_wait = TIMEOUT_SHORT;
-  siginfo.events = KE_EXPECTED | KE_FILTERED;
-  pthread_t tid;
-  pthread_create(&tid, NULL, SignalEmitterThread, &siginfo);
-
-  // Wait for the signal from the other thread and time it.
-  gettimeofday(&start, NULL);
-  int cnt = 0;
-  EXPECT_EQ(0, listener->Wait(ev, MAX_EVENTS, TIMEOUT_VERY_LONG, &cnt));
-  EXPECT_EQ(1, cnt);
-  gettimeofday(&end, NULL);
-
-  // Verify the wait duration, and that we only recieved the expected signal.
-  duration = Duration(&start, &end);
-  EXPECT_GT(TIMEOUT_SHORT + SCHEDULING_GRANULARITY, duration);
-  EXPECT_LT(TIMEOUT_SHORT - TIMEOUT_SLOP, duration);
-  EXPECT_EQ(USER_DATA_A, ev[0].user_data);
-  EXPECT_EQ(KE_EXPECTED, ev[0].events);
-}
-
-
-namespace {
-
-class KernelProxyPolling : public KernelProxy {
- public:
-  virtual int socket(int domain, int type, int protocol) {
-    ScopedMount mnt;
-    ScopedMountNode node(new EventEmitterTester());
-    ScopedKernelHandle handle(new KernelHandle(mnt, node));
-
-    Error error = handle->Init(0);
-    if (error) {
-      errno = error;
-      return -1;
-    }
-
-    return AllocateFD(handle);
-  }
-};
-
-class KernelProxyPollingTest : public ::testing::Test {
- public:
-  void SetUp() {
-    ki_init(&kp_);
-  }
-
-  void TearDown() {
-    ki_uninit();
+  }
+
+  void CloseFDs(int* fds, int cnt) {
+    for (int index = 0; index < cnt; index++)
+      kp->close(fds[index]);
   }
 
  protected:
-  KernelProxyPolling kp_;
-};
-
-}  // namespace
-
-
-#define SOCKET_CNT 4
-void SetFDs(fd_set* set, int* fds) {
-  FD_ZERO(set);
-
-  FD_SET(0, set);
-  FD_SET(1, set);
-  FD_SET(2, set);
-
-  for (int index = 0; index < SOCKET_CNT; index++)
-    FD_SET(fds[index], set);
-}
-
-TEST_F(KernelProxyPollingTest, Select) {
-  int fds[SOCKET_CNT];
-
+  KernelProxy* kp;
+
+  timeval tv;
   fd_set rd_set;
   fd_set wr_set;
-
-  FD_ZERO(&rd_set);
-  FD_ZERO(&wr_set);
-
-  FD_SET(0, &rd_set);
-  FD_SET(1, &rd_set);
-  FD_SET(2, &rd_set);
-
-  FD_SET(0, &wr_set);
-  FD_SET(1, &wr_set);
-  FD_SET(2, &wr_set);
-
-  // Expect normal files to select as read, write, and error
-  int cnt = select(4, &rd_set, &rd_set, &rd_set, NULL);
-  EXPECT_EQ(3 * 3, cnt);
-  EXPECT_NE(0, FD_ISSET(0, &rd_set));
-  EXPECT_NE(0, FD_ISSET(1, &rd_set));
-  EXPECT_NE(0, FD_ISSET(2, &rd_set));
-
-  for (int index = 0 ; index < SOCKET_CNT; index++) {
-    fds[index] = socket(0, 0, 0);
-    EXPECT_NE(-1, fds[index]);
-  }
-
-  // Highest numbered fd
-  const int fdnum = fds[SOCKET_CNT - 1] + 1;
-
-  // Expect only the normal files to select
-  SetFDs(&rd_set, fds);
-  cnt = select(fds[SOCKET_CNT-1] + 1, &rd_set, NULL, NULL, NULL);
-  EXPECT_EQ(3, cnt);
-  EXPECT_NE(0, FD_ISSET(0, &rd_set));
-  EXPECT_NE(0, FD_ISSET(1, &rd_set));
-  EXPECT_NE(0, FD_ISSET(2, &rd_set));
-  for (int index = 0 ; index < SOCKET_CNT; index++) {
-    EXPECT_EQ(0, FD_ISSET(fds[index], &rd_set));
-  }
-
-  // Poke one of the pollable nodes to be READ ready
-  ioctl(fds[0], POLLIN, NULL);
-
-  // Expect normal files to be read/write and one pollable node to be read.
-  SetFDs(&rd_set, fds);
-  SetFDs(&wr_set, fds);
-  cnt = select(fdnum, &rd_set, &wr_set, NULL, NULL);
-  EXPECT_EQ(7, cnt);
+  fd_set ex_set;
+  struct pollfd pollfds[MAX_FDS];
+};
+
+TEST_F(SelectPollTest, PollMemPipe) {
+  int fds[2];
+
+  // Both FDs for regular files should be read/write but not exception.
+  fds[0] = kp->open("/test.txt", O_CREAT | O_WRONLY);
+  fds[1] = kp->open("/test.txt", O_RDONLY);
+
+  SetFDs(fds, 2);
+
+  EXPECT_EQ(2, kp->poll(pollfds, 2, 0));
+  EXPECT_EQ(POLLIN | POLLOUT, pollfds[0].revents);
+  EXPECT_EQ(POLLIN | POLLOUT, pollfds[1].revents);
+  CloseFDs(fds, 2);
+
+  // The write FD should select for write-only, read FD should not select
+  EXPECT_EQ(0, kp->pipe(fds));
+  SetFDs(fds, 2);
+
+  EXPECT_EQ(2, kp->poll(pollfds, 2, 0));
+  // TODO(noelallen) fix poll based on open mode
+  // EXPECT_EQ(0, pollfds[0].revents);
+  // Bug 291018
+  EXPECT_EQ(POLLOUT, pollfds[1].revents);
+
+  CloseFDs(fds, 2);
+}
+
+TEST_F(SelectPollTest, SelectMemPipe) {
+  int fds[2];
+
+  // Both FDs for regular files should be read/write but not exception.
+  fds[0] = kp->open("/test.txt", O_CREAT | O_WRONLY);
+  fds[1] = kp->open("/test.txt", O_RDONLY);
+  SetFDs(fds, 2);
+
+  EXPECT_EQ(4, kp->select(fds[1] + 1, &rd_set, &wr_set, &ex_set, &tv));
   EXPECT_NE(0, FD_ISSET(fds[0], &rd_set));
-  EXPECT_EQ(0, FD_ISSET(fds[0], &wr_set));
-}
-
-
+  EXPECT_NE(0, FD_ISSET(fds[1], &rd_set));
+  EXPECT_NE(0, FD_ISSET(fds[0], &wr_set));
+  EXPECT_NE(0, FD_ISSET(fds[1], &wr_set));
+  EXPECT_EQ(0, FD_ISSET(fds[0], &ex_set));
+  EXPECT_EQ(0, FD_ISSET(fds[1], &ex_set));
+
+  CloseFDs(fds, 2);
+
+  // The write FD should select for write-only, read FD should not select
+  EXPECT_EQ(0, kp->pipe(fds));
+  SetFDs(fds, 2);
+
+  EXPECT_EQ(2, kp->select(fds[1] + 1, &rd_set, &wr_set, &ex_set, &tv));
+  EXPECT_EQ(0, FD_ISSET(fds[0], &rd_set));
+  EXPECT_EQ(0, FD_ISSET(fds[1], &rd_set));
+  // TODO(noelallen) fix poll based on open mode
+  // EXPECT_EQ(0, FD_ISSET(fds[0], &wr_set));
+  // Bug 291018
+  EXPECT_NE(0, FD_ISSET(fds[1], &wr_set));
+  EXPECT_EQ(0, FD_ISSET(fds[0], &ex_set));
+  EXPECT_EQ(0, FD_ISSET(fds[1], &ex_set));
+}
+
+