avoid blocking accept calls

this is to increase portability to platforms not implementing the
close-convention for (server) sockets.

also set all accepted sockets to blocking mode to avoid issues related
to maybe inheriting this setting from the server socket.

3 files changed, 58 insertions, 7 deletions

diff --git a/vespalib/src/tests/net/socket/socket_test.cpp b/vespalib/src/tests/net/socket/socket_test.cpp
index c7cb2a0b6d9..88146cd4fb2 100644
--- a/vespalib/src/tests/net/socket/socket_test.cpp
+++ b/vespalib/src/tests/net/socket/socket_test.cpp
@@ -324,6 +324,15 @@ TEST("require that sockets can be set blocking and non-blocking") {
     TEST_DO(verifier.verify_blocking(false));
 }
 
+TEST("require that server sockets use non-blocking underlying socket") {
+    ServerSocket tcp_server("tcp/0");
+    ServerSocket ipc_server("ipc/file:my_socket");
+    test::SocketOptionsVerifier tcp_verifier(tcp_server.get_fd());
+    test::SocketOptionsVerifier ipc_verifier(ipc_server.get_fd());
+    TEST_DO(tcp_verifier.verify_blocking(false));
+    TEST_DO(ipc_verifier.verify_blocking(false));
+}
+
 TEST("require that tcp nodelay can be enabled and disabled") {
     SocketHandle handle(socket(my_inet(), SOCK_STREAM, 0));
     test::SocketOptionsVerifier verifier(handle.get());
diff --git a/vespalib/src/vespa/vespalib/net/server_socket.cpp b/vespalib/src/vespa/vespalib/net/server_socket.cpp
index cc363f49763..2928a9a72b0 100644
--- a/vespalib/src/vespa/vespalib/net/server_socket.cpp
+++ b/vespalib/src/vespa/vespalib/net/server_socket.cpp
@@ -4,12 +4,27 @@
 #include "socket_spec.h"
 #include <sys/stat.h>
 #include <dirent.h>
+#include <errno.h>
+#include <chrono>
+#include <thread>
 
 #include <vespa/log/log.h>
 LOG_SETUP(".vespalib.net.server_socket");
 
 namespace vespalib {
 
+namespace {
+
+SocketHandle adjust_blocking(SocketHandle handle, bool value) {
+    if (handle.valid() && handle.set_blocking(value)) {
+        return handle;
+    } else {
+        return SocketHandle();
+    }
+}
+
+bool is_blocked(int err) { return ((err == EWOULDBLOCK) || (err == EAGAIN)); }
+
 bool is_socket(const vespalib::string &path) {
     struct stat info;
     if (path.empty() || (lstat(path.c_str(), &info) != 0)) {
@@ -18,6 +33,8 @@ bool is_socket(const vespalib::string &path) {
     return S_ISSOCK(info.st_mode);
 }
 
+}
+
 void
 ServerSocket::cleanup()
 {
@@ -27,8 +44,10 @@ ServerSocket::cleanup()
 }
 
 ServerSocket::ServerSocket(const SocketSpec &spec)
-    : _handle(spec.server_address().listen()),
-      _path(spec.path())
+    : _handle(adjust_blocking(spec.server_address().listen(), false)),
+      _path(spec.path()),
+      _blocking(true),
+      _shutdown(false)
 {
     if (!_handle.valid() && is_socket(_path)) {
         if (!spec.client_address().connect_async().valid()) {
@@ -54,7 +73,9 @@ ServerSocket::ServerSocket(int port)
 
 ServerSocket::ServerSocket(ServerSocket &&rhs)
     : _handle(std::move(rhs._handle)),
-      _path(std::move(rhs._path))
+      _path(std::move(rhs._path)),
+      _blocking(rhs._blocking),
+      _shutdown(rhs._shutdown.load(std::memory_order_acquire))
 {
     rhs._path.clear();
 }
@@ -65,6 +86,8 @@ ServerSocket::operator=(ServerSocket &&rhs)
     cleanup();
     _handle = std::move(rhs._handle);
     _path = std::move(rhs._path);
+    _blocking = rhs._blocking;
+    _shutdown.store(rhs._shutdown.load(std::memory_order_acquire), std::memory_order_release);
     rhs._path.clear();
     return *this;
 }
@@ -78,13 +101,28 @@ ServerSocket::address() const
 void
 ServerSocket::shutdown()
 {
+    _shutdown.store(true, std::memory_order_release);
     _handle.shutdown();
 }
 
 SocketHandle
 ServerSocket::accept()
 {
-    return _handle.accept();
+    if (!_blocking) {
+        return adjust_blocking(_handle.accept(), true);
+    } else {
+        for (;;) {
+            if (_shutdown.load(std::memory_order_acquire)) {
+                errno = EIO;
+                return SocketHandle();
+            }
+            SocketHandle res = _handle.accept();
+            if (res.valid() || !is_blocked(errno)) {
+                return adjust_blocking(std::move(res), true);
+            }
+            std::this_thread::sleep_for(std::chrono::milliseconds(10));
+        }
+    }
 }
 
 } // namespace vespalib
diff --git a/vespalib/src/vespa/vespalib/net/server_socket.h b/vespalib/src/vespa/vespalib/net/server_socket.h
index 1ad0a738117..af6f0a31175 100644
--- a/vespalib/src/vespa/vespalib/net/server_socket.h
+++ b/vespalib/src/vespa/vespalib/net/server_socket.h
@@ -4,6 +4,7 @@
 
 #include "socket_handle.h"
 #include "socket_address.h"
+#include <atomic>
 
 namespace vespalib {
 
@@ -14,9 +15,9 @@ class ServerSocket
 private:
     SocketHandle _handle;
     vespalib::string _path;
+    bool _blocking;
+    std::atomic<bool> _shutdown;
 
-    explicit ServerSocket(SocketHandle handle);
-    static ServerSocket listen(const SocketSpec &spec);
     void cleanup();
 public:
     ServerSocket() : _handle(), _path() {}
@@ -30,7 +31,10 @@ public:
     int get_fd() const { return _handle.get(); }
     SocketAddress address() const;
     void shutdown();
-    bool set_blocking(bool value) { return _handle.set_blocking(value); }
+    bool set_blocking(bool value) {
+        _blocking = value;
+        return true;
+    }
     SocketHandle accept();
 };