added SingleFdSelector utility

3 files changed, 152 insertions, 0 deletions

diff --git a/vespalib/src/tests/net/selector/selector_test.cpp b/vespalib/src/tests/net/selector/selector_test.cpp
index 964e37adf3b..302de712ef8 100644
--- a/vespalib/src/tests/net/selector/selector_test.cpp
+++ b/vespalib/src/tests/net/selector/selector_test.cpp
@@ -199,4 +199,53 @@ TEST_MT_FF("require that selection sources can be added while waiting for events
     }
 }
 
+TEST_MT_FFF("require that single fd selector can wait for read events while handling wakeups correctly",
+            2, SocketPair(SocketPair::create()), SingleFdSelector(f1.a.get()), TimeBomb(60))
+{
+    if (thread_id == 0) {
+        EXPECT_EQUAL(f2.wait_readable(), false); // wakeup only
+        TEST_BARRIER(); // #1
+        EXPECT_EQUAL(f2.wait_readable(), true); // read only
+        TEST_BARRIER(); // #2
+        TEST_BARRIER(); // #3
+        EXPECT_EQUAL(f2.wait_readable(), true); // read and wakeup
+    } else {
+        std::this_thread::sleep_for(std::chrono::milliseconds(20));
+        f2.wakeup();
+        TEST_BARRIER(); // #1
+        vespalib::string msg("test");
+        std::this_thread::sleep_for(std::chrono::milliseconds(20));
+        ASSERT_EQUAL(f1.b.write(msg.data(), msg.size()), ssize_t(msg.size()));
+        TEST_BARRIER(); // #2
+        f2.wakeup();
+        TEST_BARRIER(); // #3
+    }
+}
+
+TEST_MT_FFF("require that single fd selector can wait for write events while handling wakeups correctly",
+            2, SocketPair(SocketPair::create()), SingleFdSelector(f1.a.get()), TimeBomb(60))
+{
+    if (thread_id == 0) {
+        EXPECT_EQUAL(f2.wait_writable(), true); // write only
+        TEST_BARRIER(); // #1
+        TEST_BARRIER(); // #2
+        EXPECT_EQUAL(f2.wait_writable(), true); // write and wakeup
+        size_t buffer_size = 0;
+        while (f1.a.write("x", 1) == 1) {
+            ++buffer_size;
+        }
+        EXPECT_TRUE((errno == EWOULDBLOCK) || (errno == EAGAIN));
+        fprintf(stderr, "buffer size: %zu\n", buffer_size);
+        TEST_BARRIER(); // #3
+        EXPECT_EQUAL(f2.wait_readable(), false); // wakeup only
+    } else {
+        TEST_BARRIER(); // #1
+        f2.wakeup();
+        TEST_BARRIER(); // #2
+        TEST_BARRIER(); // #3
+        std::this_thread::sleep_for(std::chrono::milliseconds(20));
+        f2.wakeup();
+    }
+}
+
 TEST_MAIN() { TEST_RUN_ALL(); }
diff --git a/vespalib/src/vespa/vespalib/net/selector.cpp b/vespalib/src/vespa/vespalib/net/selector.cpp
index 5d73396bc7d..762849b81f0 100644
--- a/vespalib/src/vespa/vespalib/net/selector.cpp
+++ b/vespalib/src/vespa/vespalib/net/selector.cpp
@@ -15,6 +15,28 @@ namespace {
 
 //-----------------------------------------------------------------------------
 
+struct SingleFdHandler {
+    int my_fd;
+    bool got_wakeup;
+    bool got_read;
+    bool got_write;
+    SingleFdHandler(int my_fd_in)
+        : my_fd(my_fd_in), got_wakeup(false), got_read(false), got_write(false) {}
+    void handle_wakeup() {
+        got_wakeup = true;
+    }
+    void handle_event(int &ctx, bool read, bool write) {
+        if ((ctx == my_fd) && read) {
+            got_read = true;
+        }
+        if ((ctx == my_fd) && write) {
+            got_write = true;
+        }
+    }
+};
+
+//-----------------------------------------------------------------------------
+
 uint32_t maybe(uint32_t value, bool yes) { return yes ? value : 0; }
 
 void check(int res) {
@@ -104,4 +126,58 @@ Epoll::wait(epoll_event *events, size_t max_events, int timeout_ms)
 
 //-----------------------------------------------------------------------------
 
+SingleFdSelector::SingleFdSelector(int fd)
+    : _fd(fd),
+      _selector()
+{
+    _selector.add(_fd, _fd, false, false);
+}
+
+SingleFdSelector::~SingleFdSelector()
+{
+    _selector.remove(_fd);
+}
+
+bool
+SingleFdSelector::wait_readable()
+{
+    _selector.update(_fd, _fd, true, false);
+    for (;;) {
+        _selector.poll(-1);
+        SingleFdHandler handler(_fd);
+        _selector.dispatch(handler);
+        if (handler.got_read) {
+            return true;
+        }
+        if (handler.got_wakeup) {
+            return false;
+        }
+    }
+}
+
+bool
+SingleFdSelector::wait_writable()
+{
+    _selector.update(_fd, _fd, false, true);
+    for (;;) {
+        _selector.poll(-1);
+        SingleFdHandler handler(_fd);
+        _selector.dispatch(handler);
+        if (handler.got_write) {
+            return true;
+        }
+        if (handler.got_wakeup) {
+            return false;
+        }
+    }
+}
+
+void
+SingleFdSelector::wakeup()
+{
+    _selector.wakeup();
+}
+
+//-----------------------------------------------------------------------------
+
 } // namespace vespalib
diff --git a/vespalib/src/vespa/vespalib/net/selector.h b/vespalib/src/vespa/vespalib/net/selector.h
index 28fff89a78c..8b9d1c5f5d8 100644
--- a/vespalib/src/vespa/vespalib/net/selector.h
+++ b/vespalib/src/vespa/vespalib/net/selector.h
@@ -108,4 +108,31 @@ public:
 
 //-----------------------------------------------------------------------------
 
+/**
+ * Selector used to wait for events on a single file
+ * descriptor. Useful for testing or sync wrappers. Note: do not use
+ * for performance-critical code.
+ **/
+class SingleFdSelector
+{
+private:
+    int _fd;
+    Selector<int> _selector;
+
+public:
+    SingleFdSelector(int fd);
+    ~SingleFdSelector();
+
+    // returns true when readable or false on wakeup
+    bool wait_readable();
+
+    // returns true when writable or false on wakeup
+    bool wait_writable();
+
+    // make wait_readable/wait_writable return false immediately
+    void wakeup();
+};
+
+//-----------------------------------------------------------------------------
+
 } // namespace vespalib