Add FiberScheduler::connect

keep SOCK_NONBLOCK

accept and tests

bb fmt

Author: mstetsyuk

include/silk/fibers/fiber.h

@@ -9,6 +9,7 @@
 #include <memory>
 #include <utility>
 
+#include <sys/socket.h>
 #include <sys/uio.h>
 
 namespace silk
@@ -412,6 +413,62 @@ class FiberScheduler
      */
     static void poll(int fd, uint32_t events, uint64_t * triggeredEvents, IoFuture * future) noexcept;
 
+    /**
+     * Blocking connect: connect @p fd to @p addr.
+     *
+     * @param fd       File descriptor to connect.
+     * @param addr     Destination address.
+     * @param addrlen  Length of @p addr.
+     * @return         0 on success, or a errno on failure.
+     */
+    static int connect(int fd, const sockaddr * addr, socklen_t addrlen) noexcept
+    {
+        IoFuture future;
+        connect(fd, addr, addrlen, &future);
+        return future.wait();
+    }
+
+    /**
+     * Async connect. Returns immediately; the caller must wait on @p future
+     * for the result.
+     *
+     * @param fd       File descriptor to connect.
+     * @param addr     Destination address; must remain valid until @p future completes.
+     * @param addrlen  Length of @p addr.
+     * @param future   Completion handle; wait() returns 0 on success or a errno on failure.
+     */
+    static void connect(int fd, const sockaddr * addr, socklen_t addrlen, IoFuture * future) noexcept;
+
+    /**
+     * Blocking accept: accept a connection on the listening socket @p fd.
+     *
+     * @param fd          Listening socket to accept on.
+     * @param addr        If not null, receives the peer address.
+     * @param addrlen     In/out length of @p addr; ignored if @p addr is null.
+     * @param flags       accept4() flags applied to the accepted socket (e.g. SOCK_CLOEXEC).
+     * @param acceptedFd  If not null, receives the accepted socket fd on success.
+     * @return            0 on success, or a errno on failure.
+     */
+    static int accept(int fd, sockaddr * addr, socklen_t * addrlen, int flags, uint64_t * acceptedFd = nullptr) noexcept
+    {
+        IoFuture future;
+        accept(fd, addr, addrlen, flags, acceptedFd, &future);
+        return future.wait();
+    }
+
+    /**
+     * Async accept. Returns immediately; the caller must wait on @p future
+     * for the result.
+     *
+     * @param fd          Listening socket to accept on.
+     * @param addr        If not null, receives the peer address; must remain valid until @p future completes.
+     * @param addrlen     In/out length of @p addr; must remain valid until @p future completes; ignored if @p addr is null.
+     * @param flags       accept4() flags applied to the accepted socket (e.g. SOCK_CLOEXEC).
+     * @param acceptedFd  If not null, receives the accepted socket fd on success.
+     * @param future      Completion handle; wait() returns 0 on success or a errno on failure.
+     */
+    static void accept(int fd, sockaddr * addr, socklen_t * addrlen, int flags, uint64_t * acceptedFd, IoFuture * future) noexcept;
+
     /**
      * Completion handle for an async sleep submitted via sleep().
      */

src/fibers/fiber.cpp

@@ -1436,6 +1436,18 @@ void FiberScheduler::poll(int fd, uint32_t events, uint64_t * triggeredEvents, I
     enqueueIo(future, [=](io_uring_sqe * sqe) noexcept { ::io_uring_prep_poll_add(sqe, fd, events); });
 }
 
+void FiberScheduler::connect(int fd, const sockaddr * addr, socklen_t addrlen, IoFuture * future) noexcept
+{
+    future->result = nullptr;
+    enqueueIo(future, [=](io_uring_sqe * sqe) noexcept { ::io_uring_prep_connect(sqe, fd, addr, addrlen); });
+}
+
+void FiberScheduler::accept(int fd, sockaddr * addr, socklen_t * addrlen, int flags, uint64_t * acceptedFd, IoFuture * future) noexcept
+{
+    future->result = acceptedFd;
+    enqueueIo(future, [=](io_uring_sqe * sqe) noexcept { ::io_uring_prep_accept(sqe, fd, addr, addrlen, flags); });
+}
+
 void FiberScheduler::cancelIo(IoFuture * future) noexcept
 {
     future->result = nullptr;

src/fibers/tests/fiber-test.cpp

@@ -13,6 +13,10 @@
 #include <sched.h>
 #include <unistd.h>
 
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+
 namespace silk
 {
 
@@ -408,6 +412,60 @@ TEST(Fiber, cancelRead)
     ::close(fds[1]);
 }
 
+// connect + accept: client connect and listener accepts on loopback.
+TEST(Fiber, connectAccept)
+{
+    int listenFd = ::socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0);
+    ASSERT_GE(listenFd, 0);
+
+    sockaddr_in addr{};
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+
+    int r = ::bind(listenFd, reinterpret_cast<const sockaddr *>(&addr), sizeof(addr));
+    ASSERT_EQ(r, 0);
+    r = ::listen(listenFd, 1);
+    ASSERT_EQ(r, 0);
+
+    socklen_t len = sizeof(addr);
+    r = ::getsockname(listenFd, reinterpret_cast<sockaddr *>(&addr), &len);
+    ASSERT_EQ(r, 0);
+
+    int clientFd = ::socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0);
+    ASSERT_GE(clientFd, 0);
+
+    struct Params
+    {
+        int listenFd;
+        int clientFd;
+        sockaddr_in addr;
+
+        static int fiberMain(Params * p) noexcept
+        {
+            FiberScheduler::IoFuture connectFuture;
+            FiberScheduler::connect(p->clientFd, reinterpret_cast<const sockaddr *>(&p->addr), sizeof(p->addr), &connectFuture);
+
+            uint64_t acceptedFd = 0;
+            int a = FiberScheduler::accept(p->listenFd, nullptr, nullptr, SOCK_CLOEXEC, &acceptedFd);
+            EXPECT_EQ(a, 0);
+            EXPECT_GE(static_cast<int>(acceptedFd), 0);
+
+            EXPECT_EQ(connectFuture.wait(), 0);
+
+            ::close(static_cast<int>(acceptedFd));
+
+            return 0;
+        }
+    };
+
+    r = FiberScheduler::run(Params::fiberMain, {listenFd, clientFd, addr});
+    ASSERT_EQ(r, 0);
+
+    ::close(clientFd);
+    ::close(listenFd);
+}
+
+
 // Stress: many fibers each doing a write+read through their own pipe.
 TEST(Fiber, concurrentReadWrite)
 {

src/perf/net-perf.cpp

@@ -134,38 +134,10 @@ int TcpConnection::connect(const char * host, uint16_t port, TcpConnection * out
         return r;
     }
 
-    int r = ::connect(fd, reinterpret_cast<const sockaddr *>(&addr), sizeof(addr));
-    if (r < 0)
-    {
-        r = errno;
-        if (r != EINPROGRESS)
-        {
-            SILK_ERROR("connect failed: %s", std::strerror(r));
-            ::close(fd);
-            return r;
-        }
-
-        r = silk::FiberScheduler::poll(fd, POLLOUT);
-        if (r)
-        {
-            SILK_ERROR("poll failed: %s", std::strerror(r));
-            ::close(fd);
-            return r;
-        }
-    }
-
-    r = 0;
-    socklen_t len = sizeof(r);
-    if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &r, &len))
-    {
-        r = errno;
-        SILK_ERROR("getsockopt SO_ERROR failed: %s", std::strerror(r));
-        ::close(fd);
-        return r;
-    }
+    int r = silk::FiberScheduler::connect(fd, reinterpret_cast<const sockaddr *>(&addr), sizeof(addr));
     if (r)
     {
-        SILK_ERROR("connect error: %s", std::strerror(r));
+        SILK_ERROR("connect failed: %s", std::strerror(r));
         ::close(fd);
         return r;
     }
@@ -237,26 +209,13 @@ int TcpConnection::listen(const char * host, uint16_t port, int backlog, TcpConn
 
 int TcpConnection::accept(TcpConnection * out) noexcept
 {
-    int fd;
-    for (;;)
+    uint64_t acceptedFd = 0;
+    int r = silk::FiberScheduler::accept(connFd, nullptr, nullptr, SOCK_NONBLOCK | SOCK_CLOEXEC, &acceptedFd);
+    if (r)
     {
-        fd = ::accept4(connFd, nullptr, nullptr, SOCK_NONBLOCK | SOCK_CLOEXEC);
-        if (fd >= 0)
-        {
-            break;
-        }
-
-        int r = errno;
-        if (r == EAGAIN)
-        {
-            r = silk::FiberScheduler::poll(connFd, POLLIN);
-            if (!r)
-            {
-                continue;
-            }
-        }
         return r;
     }
+    int fd = static_cast<int>(acceptedFd);
 
     int value = 1;
     if (::setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &value, sizeof(value)))