make socket selection more flexible

freely specify read/write selection
handler no longer expected to map context to file descriptor
selector no longer bound to handler
selector templated on context directly
also added listen failure warning for server sockets

1 files changed, 77 insertions, 47 deletions

diff --git a/vespalib/src/tests/net/selector/selector_test.cpp b/vespalib/src/tests/net/selector/selector_test.cpp
index 82494f5423f..f21e29b586d 100644
--- a/vespalib/src/tests/net/selector/selector_test.cpp
+++ b/vespalib/src/tests/net/selector/selector_test.cpp
@@ -34,54 +34,65 @@ struct Context {
     }
 };
 
-struct Handler {
-    bool wakeup;
-    using context_type = Context;
-    int get_fd(Context &ctx) const { return ctx.fd; }
-    void handle_wakeup() { wakeup = true; }
-    void handle_event(Context &ctx, bool read, bool write) {
-        ctx.can_read = read;
-        ctx.can_write = write;
-    }
-    void reset() {
-        wakeup = false;
-    }
-};
-
 struct Fixture {
-    Handler handler;
-    Selector<Handler> selector;
+    bool wakeup;
+    Selector<Context> selector;
     std::vector<SocketPair> sockets;
     std::vector<Context> contexts;
-    Fixture(size_t size, bool write_enabled) : handler(), selector(handler, 1024), sockets(), contexts() {
+    Fixture(size_t size, bool read_enabled, bool write_enabled) : wakeup(false), selector(), sockets(), contexts() {
         for (size_t i = 0; i < size; ++i) {
             sockets.push_back(SocketPair::create());
             contexts.push_back(Context(sockets.back().a.get()));
         }
         for (auto &ctx: contexts) {
-            selector.add(ctx, write_enabled);
+            selector.add(ctx.fd, ctx, read_enabled, write_enabled);
         }
     }
+    void update(size_t idx, bool read, bool write) {
+        Context &ctx = contexts[idx];
+        selector.update(ctx.fd, ctx, read, write);
+    }
+    bool write(size_t idx, const char *str) {
+        size_t len = strlen(str);
+        ssize_t res = ::write(sockets[idx].b.get(), str, len);
+        return (res == ssize_t(len));
+    }
+    bool write_self(size_t idx, const char *str) {
+        size_t len = strlen(str);
+        ssize_t res = ::write(sockets[idx].a.get(), str, len);
+        return (res == ssize_t(len));
+    }
+    bool read(size_t idx, size_t len) {
+        char buf[128];
+        ssize_t res = ::read(sockets[idx].a.get(), buf, len);
+        return (res == ssize_t(len));
+    }
     Fixture &reset() {
-        handler.reset();
+        wakeup = false;
         for (auto &ctx: contexts) {
             ctx.reset();
         }
         return *this;
     }
-    Fixture &poll(int timeout_ms = 250000) {
+    Fixture &poll(int timeout_ms = 60000) {
         selector.poll(timeout_ms);
-        selector.dispatch();
+        selector.dispatch(*this);
         return *this;
     }
     void verify(bool expect_wakeup, std::vector<std::pair<bool,bool> > expect_events) {
-        EXPECT_EQUAL(expect_wakeup, handler.wakeup);
+        EXPECT_EQUAL(expect_wakeup, wakeup);
         ASSERT_EQUAL(expect_events.size(), contexts.size());
         for (size_t i = 0; i < expect_events.size(); ++i) {
             EXPECT_EQUAL(expect_events[i].first, contexts[i].can_read);
             EXPECT_EQUAL(expect_events[i].second, contexts[i].can_write);
         }
     }
+    // selector callbacks
+    void handle_wakeup() { wakeup = true; }
+    void handle_event(Context &ctx, bool read, bool write) {
+        ctx.can_read = read;
+        ctx.can_write = write;
+    }
 };
 
 constexpr std::pair<bool,bool> none = std::make_pair(false, false);
@@ -89,54 +100,73 @@ constexpr std::pair<bool,bool> in   = std::make_pair(true,  false);
 constexpr std::pair<bool,bool> out  = std::make_pair(false, true);
 constexpr std::pair<bool,bool> both = std::make_pair(true,  true);
 
-TEST_F("require that basic events trigger correctly", Fixture(1, true)) {
+TEST_F("require that basic events trigger correctly", Fixture(1, true, true)) {
     TEST_DO(f1.reset().poll().verify(false, {out}));
-    EXPECT_EQUAL(write(f1.sockets[0].b.get(), "test", 4), 4);
+    EXPECT_TRUE(f1.write(0, "test"));
     TEST_DO(f1.reset().poll().verify(false, {both}));
-    f1.selector.disable_write(f1.contexts[0]);
+    f1.update(0, true, false);
     TEST_DO(f1.reset().poll().verify(false, {in}));
-    f1.selector.enable_write(f1.contexts[0]);
-    TEST_DO(f1.reset().poll().verify(false, {both}));
+    f1.update(0, false, true);
+    TEST_DO(f1.reset().poll().verify(false, {out}));
+    f1.update(0, false, false);
+    TEST_DO(f1.reset().poll(10).verify(false, {none}));
+    f1.update(0, true, true);
     f1.selector.wakeup();
     TEST_DO(f1.reset().poll().verify(true, {both}));
     TEST_DO(f1.reset().poll().verify(false, {both}));
 }
 
-TEST_F("require that multiple sources can be selected on", Fixture(5, false)) {
-    char buf[128];
+TEST_FFF("require that sources can be added with some events disabled",
+         Fixture(1, true, false), Fixture(1, false, true), Fixture(1, false, false))
+{
+    EXPECT_TRUE(f1.write(0, "test"));
+    EXPECT_TRUE(f2.write(0, "test"));
+    EXPECT_TRUE(f3.write(0, "test"));
+    TEST_DO(f1.reset().poll().verify(false, {in}));
+    TEST_DO(f2.reset().poll().verify(false, {out}));
+    TEST_DO(f3.reset().poll(10).verify(false, {none}));
+    f1.update(0, true, true);
+    f2.update(0, true, true);
+    f3.update(0, true, true);
+    TEST_DO(f1.reset().poll().verify(false, {both}));
+    TEST_DO(f2.reset().poll().verify(false, {both}));
+    TEST_DO(f3.reset().poll().verify(false, {both}));
+}
+
+TEST_F("require that multiple sources can be selected on", Fixture(5, true, false)) {
     TEST_DO(f1.reset().poll(10).verify(false, {none, none, none, none, none}));
-    EXPECT_EQUAL(write(f1.sockets[1].b.get(), "test", 4), 4);
-    EXPECT_EQUAL(write(f1.sockets[3].b.get(), "test", 4), 4);
+    EXPECT_TRUE(f1.write(1, "test"));
+    EXPECT_TRUE(f1.write(3, "test"));
     TEST_DO(f1.reset().poll().verify(false, {none, in, none, in, none}));
-    EXPECT_EQUAL(read(f1.sockets[1].a.get(), buf, sizeof(buf)), 4);
-    EXPECT_EQUAL(read(f1.sockets[3].a.get(), buf, 2), 2);
+    EXPECT_TRUE(f1.read(1, strlen("test")));
+    EXPECT_TRUE(f1.read(3, strlen("te")));
     TEST_DO(f1.reset().poll().verify(false, {none, none, none, in, none}));
-    EXPECT_EQUAL(read(f1.sockets[3].a.get(), buf, sizeof(buf)), 2);
+    EXPECT_TRUE(f1.read(3, strlen("st")));
     TEST_DO(f1.reset().poll(10).verify(false, {none, none, none, none, none}));
 }
 
-TEST_F("require that removed sources no longer produce events", Fixture(2, true)) {
+TEST_F("require that removed sources no longer produce events", Fixture(2, true, true)) {
     TEST_DO(f1.reset().poll().verify(false, {out, out}));
-    EXPECT_EQUAL(write(f1.sockets[0].b.get(), "test", 4), 4);
-    EXPECT_EQUAL(write(f1.sockets[1].b.get(), "test", 4), 4);
+    EXPECT_TRUE(f1.write(0, "test"));
+    EXPECT_TRUE(f1.write(1, "test"));
     TEST_DO(f1.reset().poll().verify(false, {both, both}));
-    f1.selector.remove(f1.contexts[0]);
+    f1.selector.remove(f1.contexts[0].fd);
     TEST_DO(f1.reset().poll().verify(false, {none, both}));
 }
 
-TEST_F("require that filling the output buffer disables write events", Fixture(1, true)) {
-    EXPECT_EQUAL(write(f1.sockets[0].b.get(), "test", 4), 4);
+TEST_F("require that filling the output buffer disables write events", Fixture(1, true, true)) {
+    EXPECT_TRUE(f1.write(0, "test"));
     TEST_DO(f1.reset().poll().verify(false, {both}));
     size_t buffer_size = 0;
-    while (write(f1.sockets[0].a.get(), "x", 1) == 1) {
+    while (f1.write_self(0, "x")) {
         ++buffer_size;
     }
-    EXPECT_EQUAL(errno, EWOULDBLOCK);
+    EXPECT_TRUE((errno == EWOULDBLOCK) || (errno == EAGAIN));
     fprintf(stderr, "buffer size: %zu\n", buffer_size);
     TEST_DO(f1.reset().poll().verify(false, {in}));
 }
 
-TEST_MT_FF("require that selector can be woken while waiting for events", 2, Fixture(0, false), TimeBomb(60)) {
+TEST_MT_FF("require that selector can be woken while waiting for events", 2, Fixture(0, true, false), TimeBomb(60)) {
     if (thread_id == 0) {
         TEST_DO(f1.reset().poll().verify(true, {}));
     } else {
@@ -145,16 +175,16 @@ TEST_MT_FF("require that selector can be woken while waiting for events", 2, Fix
     }
 }
 
-TEST_MT_FF("require that selection criteria can be changed while waiting for events", 2, Fixture(1, false), TimeBomb(60)) {
+TEST_MT_FF("require that selection criteria can be changed while waiting for events", 2, Fixture(1, true, false), TimeBomb(60)) {
     if (thread_id == 0) {
         TEST_DO(f1.reset().poll().verify(false, {out}));
     } else {
         std::this_thread::sleep_for(std::chrono::milliseconds(20));
-        f1.selector.enable_write(f1.contexts[0]);        
+        f1.update(0, true, true);
     }
 }
 
-TEST_MT_FF("require that selection sources can be added while waiting for events", 2, Fixture(0, false), TimeBomb(60)) {
+TEST_MT_FF("require that selection sources can be added while waiting for events", 2, Fixture(0, true, false), TimeBomb(60)) {
     if (thread_id == 0) {
         TEST_DO(f1.reset().poll().verify(false, {}));
         TEST_BARRIER();
@@ -162,7 +192,7 @@ TEST_MT_FF("require that selection sources can be added while waiting for events
         SocketPair pair = SocketPair::create();
         Context ctx(pair.a.get());
         std::this_thread::sleep_for(std::chrono::milliseconds(20));
-        f1.selector.add(ctx, true);
+        f1.selector.add(ctx.fd, ctx, true, true);
         TEST_BARRIER();
         EXPECT_TRUE(ctx.can_write);
     }