Partial send on blocking sockets is not available in Windows

.github/workflows/build_and_test.yml

@@ -21,11 +21,11 @@ jobs:
             # which fails at link time
             type: Release
 
-          # uses pre-installed OpenSSL 1.1.1u and fails sockpuppet_tls_buffered_test with 'SSL routines:ssl3_write_bytes:bad length:ssl'
-          #- name: "Windows MSVC17 shared library using OpenSSL"
-          #  os: windows-2022
-          #  generator: -G "Visual Studio 17 2022" -A x64
-          #  type: Debug
+          - name: "Windows MSVC17 shared library using OpenSSL"
+            os: windows-2022
+            generator: -G "Visual Studio 17 2022" -A x64
+            # uses pre-installed OpenSSL 1.1.1u
+            type: Debug
 
           - name: "Unix GCC shared library using LibreSSL"
             os: ubuntu-latest
@@ -36,6 +36,7 @@ jobs:
           - name: "Unix GCC shared library using OpenSSL"
             os: ubuntu-latest
             generator: -G "Unix Makefiles"
+            # uses pre-installed OpenSSL 3.0.2
             type: Debug
 
           - name: "OSX clang shared library using LibreSSL"
@@ -47,6 +48,7 @@ jobs:
           - name: "OSX clang shared library using OpenSSL"
             os: macos-latest
             generator: -G "Unix Makefiles"
+            # uses pre-installed OpenSSL 3.1.1
             type: Debug
     runs-on: ${{ matrix.os }}
     steps:

examples/sockpuppet_chat_client.cpp

@@ -45,13 +45,13 @@ struct ReconnectClient
           to_string(client->LocalAddress()) +
           " -> " +
           to_string(remoteAddress));
-    } catch(std::exception const &) {
+    } catch(std::exception const &e) {
       ioBuf.Print(
           "failed to (re)connect to " +
           to_string(remoteAddress) +
+          " (" + e.what() + ")"
           " will retry in " +
-          std::to_string(delay.count()) +
-          "s");
+          std::to_string(delay.count()) + "s");
 
       // schedule a reconnect attempt with increasing backoff delay
       ToDo(driver,

include/sockpuppet/socket_async.h

@@ -136,8 +136,9 @@ using ConnectHandler = std::function<void(SocketTcp, Address)>;
 /// @param  Address of peer that just disconnected from local socket.
 ///         Matches connect address the socket was created with or
 ///         obtained peer address of incoming connection.
+/// @param  Error/reason message.
 /// @note  After peer disconnect the socket is invalid and should be released.
-using DisconnectHandler = std::function<void(Address)>;
+using DisconnectHandler = std::function<void(Address, char const *)>;
 
 /// UDP (unreliable communication) socket class that adds an interface for
 /// an external socket driver to the buffered UDP class.

src/driver_impl.cpp

@@ -18,7 +18,7 @@ struct FdEqual
 
   bool operator()(SocketAsyncImpl const &async) const
   {
-    return (async.buff->sock->fd == fd);
+    return (async.DriverGetFd() == fd);
   }
 
   bool operator()(pollfd const &pfd) const
@@ -102,7 +102,7 @@ void Driver::DriverImpl::Step(Duration timeout)
   StepGuard lock(*this);
 
   if(todos.empty()) {
-    StepFds(timeout);
+    StepSockets(timeout);
   } else {
     // execute due ToDos while keeping track of the time
     auto remaining =
@@ -113,7 +113,7 @@ void Driver::DriverImpl::Step(Duration timeout)
               StepTodos(DeadlineLimited(timeout))));
 
     // run sockets with remaining time
-    StepFds(remaining);
+    StepSockets(remaining);
   }
 }
 
@@ -144,8 +144,12 @@ Duration Driver::DriverImpl::StepTodos(Deadline deadline)
   return Duration(0);
 }
 
-void Driver::DriverImpl::StepFds(Duration timeout)
+void Driver::DriverImpl::StepSockets(Duration timeout)
 {
+  // query sockets whether they request/suppress write poll
+  // the TLS socket uses this override for the TLS handshake
+  QuerySockets();
+
   if(!Wait(pfds, timeout)) {
     return; // timeout exceeded
   }
@@ -157,7 +161,7 @@ void Driver::DriverImpl::StepFds(Duration timeout)
   } else if(pfds.front().revents != 0) {
     throw std::logic_error("unexpected signalling pipe poll result");
   } else {
-    DoOneFdTask();
+    DoOneSocketTask();
   }
 }
 
@@ -193,26 +197,25 @@ void Driver::DriverImpl::ToDoMove(ToDoShared todo, TimePoint when)
   todos.Move(std::move(todo), when);
 }
 
-void Driver::DriverImpl::AsyncRegister(
-    SocketAsyncImpl &sock)
+void Driver::DriverImpl::AsyncRegister(SocketAsyncImpl &sock)
 {
   PauseGuard lock(*this);
 
   sockets.emplace_back(sock);
-  pfds.emplace_back(pollfd{sock.buff->sock->fd, POLLIN, 0});
+  pfds.emplace_back(pollfd{sock.DriverGetFd(), POLLIN, 0});
 }
 
 void Driver::DriverImpl::AsyncUnregister(SOCKET fd)
 {
   PauseGuard lock(*this);
 
-  auto itSocket = std::find_if(begin(sockets), end(sockets), FdEqual{fd});
-  assert(itSocket != end(sockets));
-  sockets.erase(itSocket);
+  if(auto it = std::find_if(begin(sockets), end(sockets), FdEqual{fd}); it != end(sockets)) {
+    sockets.erase(it);
+  }
 
-  auto itPfd = std::find_if(begin(pfds), end(pfds), FdEqual{fd});
-  assert(itPfd != end(pfds));
-  pfds.erase(itPfd);
+  if(auto it = std::find_if(begin(pfds), end(pfds), FdEqual{fd}); it != end(pfds)) {
+    pfds.erase(it);
+  }
 }
 
 void Driver::DriverImpl::AsyncWantSend(SOCKET fd)
@@ -240,21 +243,37 @@ void Driver::DriverImpl::Unbump()
   (void)pipeTo.ReceiveFrom(dump, sizeof(dump));
 }
 
-void Driver::DriverImpl::DoOneFdTask()
+void Driver::DriverImpl::QuerySockets()
+{
+#ifdef SOCKPUPPET_WITH_TLS
+  assert(sockets.size() + 1U == pfds.size());
+
+  for(size_t i = 0U; i < sockets.size(); ++i) {
+    auto &&pfd = pfds[i + 1U];
+    auto &&sock = sockets[i].get();
+    assert(pfd.fd == sock.DriverGetFd());
+
+    sock.DriverQuery(pfd.events);
+  }
+#endif // SOCKPUPPET_WITH_TLS
+}
+
+void Driver::DriverImpl::DoOneSocketTask()
 {
   assert(sockets.size() + 1U == pfds.size());
 
   // user task may unregister/destroy a socket -> handle only one
   for(size_t i = 0U; i < sockets.size(); ++i) {
     auto &&pfd = pfds[i + 1U];
     auto &&sock = sockets[i].get();
-    assert(pfd.fd == sock.buff->sock->fd);
+    assert(pfd.fd == sock.DriverGetFd());
 
     if(pfd.revents & POLLIN) {
       sock.DriverOnReadable();
       return;
     } else if(pfd.revents & POLLOUT) {
       if(sock.DriverOnWritable()) {
+        // send queue emptied -> stop write poll
         pfd.events &= ~POLLOUT;
       }
       return;

src/driver_impl.h

@@ -53,18 +53,19 @@ struct Driver::DriverImpl
     PauseGuard &operator=(PauseGuard &&) = delete;
   };
 
-  /// internal signalling pipe for cancelling Step()
+  /// Internal signalling pipe for cancelling Step()
   AddressShared pipeToAddr;
   SocketImpl pipeFrom;
   SocketImpl pipeTo;
 
+  /// Lists of managed ToDos and Sockets protected by mutex
   std::recursive_mutex stepMtx;
   std::mutex pauseMtx;
   ToDos todos; // guarded by stepMtx
   std::vector<SocketRef> sockets; // guarded by stepMtx
   std::vector<pollfd> pfds; // front element belongs to internal signalling pipe; guarded by stepMtx
 
-  std::atomic<bool> shouldStop; ///< flag for cancelling Run()
+  std::atomic<bool> shouldStop; ///< Flag for cancelling Run()
 
   DriverImpl();
   DriverImpl(DriverImpl const &) = delete;
@@ -76,7 +77,7 @@ struct Driver::DriverImpl
   void Step(Duration timeout);
   template<typename Deadline>
   Duration StepTodos(Deadline deadline);
-  void StepFds(Duration timeout);
+  void StepSockets(Duration timeout);
 
   void Run();
   void Stop();
@@ -91,10 +92,13 @@ struct Driver::DriverImpl
   void AsyncUnregister(SOCKET fd);
   void AsyncWantSend(SOCKET fd);
 
+  // interactions with signalling pipe
   void Bump();
   void Unbump();
 
-  void DoOneFdTask();
+  // interactions with sockets
+  void QuerySockets();
+  void DoOneSocketTask();
 };
 
 } // namespace sockpuppet

src/socket.cpp

@@ -17,6 +17,7 @@ SocketUdp::SocketUdp(Address const &bindAddress)
 {
   impl->Bind(bindAddress.impl->ForUdp());
   impl->SetSockOptBroadcast();
+  impl->SetSockOptNonBlocking();
 }
 
 size_t SocketUdp::SendTo(char const *data, size_t size,
@@ -54,6 +55,7 @@ SocketTcp::SocketTcp(Address const &connectAddress)
 {
   impl->SetSockOptNoSigPipe();
   impl->Connect(connectAddress.impl->ForTcp());
+  impl->SetSockOptNonBlocking();
 }
 
 #ifdef SOCKPUPPET_WITH_TLS
@@ -65,6 +67,7 @@ SocketTcp::SocketTcp(Address const &connectAddress,
 {
   impl->SetSockOptNoSigPipe();
   impl->Connect(connectAddress.impl->ForTcp());
+  impl->SetSockOptNonBlocking();
 }
 #endif // SOCKPUPPET_WITH_TLS
 
@@ -97,6 +100,7 @@ SocketTcp::SocketTcp(std::unique_ptr<SocketImpl> &&other)
   : impl(std::move(other))
 {
   impl->SetSockOptNoSigPipe();
+  impl->SetSockOptNonBlocking();
 }
 
 SocketTcp::SocketTcp(SocketTcp &&other) noexcept = default;
@@ -112,6 +116,7 @@ Acceptor::Acceptor(Address const &bindAddress)
 {
   impl->SetSockOptReuseAddr();
   impl->Bind(bindAddress.impl->ForTcp());
+  impl->SetSockOptNonBlocking();
 }
 
 #ifdef SOCKPUPPET_WITH_TLS
@@ -123,6 +128,7 @@ Acceptor::Acceptor(Address const &bindAddress,
 {
   impl->SetSockOptReuseAddr();
   impl->Bind(bindAddress.impl->ForTcp());
+  impl->SetSockOptNonBlocking();
 }
 #endif // SOCKPUPPET_WITH_TLS
 

src/socket_async_impl.cpp

@@ -106,11 +106,20 @@ bool SocketAsyncImpl::DoSendEnqueue(std::promise<void> promise, Args&&... args)
 
   auto &q = std::get<Queue>(sendQ);
   bool wasEmpty = q.empty();
-  assert(!pendingTlsSend || !wasEmpty); // queue cannot be empty while TLS send is pending
   q.emplace(std::move(promise), std::forward<Args>(args)...);
   return wasEmpty;
 }
 
+SOCKET SocketAsyncImpl::DriverGetFd() const
+{
+  return buff->sock->fd;
+}
+
+void SocketAsyncImpl::DriverQuery(short &events)
+{
+  buff->sock->DriverQuery(events);
+}
+
 void SocketAsyncImpl::DriverOnReadable()
 {
   onReadable();
@@ -130,21 +139,6 @@ void SocketAsyncImpl::DriverConnect(ConnectHandler const &onConnect)
 
 void SocketAsyncImpl::DriverReceive(ReceiveHandler const &onReceive)
 {
-  if(pendingTlsSend) {
-    // a previous TLS send failed because handshake receipt was pending
-    // which probably arrived now: repeat the same send call to handle
-    // the handshake and continue where it left off sending
-    // see https://www.openssl.org/docs/man1.1.1/man3/SSL_write.html
-    bool isSendQueueEmpty = DriverOnWritable();
-    if(!isSendQueueEmpty) {
-      if(auto ptr = driver.lock()) {
-        ptr->AsyncWantSend(buff->sock->fd);
-      }
-    }
-
-    return;
-  }
-
   try {
     auto buffer = buff->Receive();
     if(buffer->empty()) {
@@ -189,30 +183,21 @@ bool SocketAsyncImpl::DriverSend(SendQ &q)
 {
   auto const sendQSize = q.size();
   if(!sendQSize) {
-    throw std::logic_error("uncalled send");
+    // TLS socket has requested write for TLS handshake
+    buff->sock->DriverPending();
+    return true;
   }
-
   auto &&[promise, buffer] = q.front();
-  if(pendingTlsSend) {
-    assert(pendingTlsSend == buffer->data());
-    pendingTlsSend = nullptr;
-  }
 
   try {
-    if(auto sent = buff->sock->SendSome(buffer->data(), buffer->size())) {
-      if(sent == buffer->size()) {
-        promise.set_value();
-      } else {
-        // allow partial send to avoid starving other driver's sockets if this one is rate limited
-        buffer->erase(0, sent);
-        return false;
-      }
-    } else { // zero-size sent data
-      // TLS can't send while handshake receipt pending:
-      // give up for now by proclaiming the send queue is empty,
-      // but actually keep the data queued and retry the exact same call on readable
-      pendingTlsSend = buffer->data();
-      return true;
+    auto sent = buff->sock->SendSome(buffer->data(), buffer->size());
+    if(sent == buffer->size()) {
+      promise.set_value();
+    } else {
+      // allow partial send to avoid starving other driver's sockets if this one is rate limited
+      // (may also be zero sent despite socket being writable when TLS handshake is pending)
+      buffer->erase(0, sent);
+      return false;
     }
   } catch(std::runtime_error const &e) {
     promise.set_exception(std::make_exception_ptr(e));
@@ -248,9 +233,14 @@ void SocketAsyncImpl::DriverOnError(char const *message)
 }
 
 void SocketAsyncImpl::DriverDisconnect(DisconnectHandler const &onDisconnect,
-    AddressShared peerAddr, char const *)
+    AddressShared peerAddr, char const *reason)
 {
-  onDisconnect(Address(std::move(peerAddr)));
+  // don't try to do any further receives that would just lead to more errors
+  if(auto ptr = driver.lock()) {
+    ptr->AsyncUnregister(buff->sock->fd);
+  }
+
+  onDisconnect(Address(std::move(peerAddr)), reason);
 }
 
 

src/socket_async_impl.h

@@ -30,7 +30,6 @@ struct SocketAsyncImpl
   std::function<void(char const *)> onError; // contains use-case-dependent data as bound arguments
   mutable std::mutex sendQMtx;
   std::variant<SendQ, SendToQ> sendQ; // use-case dependent queue type
-  char const *pendingTlsSend = nullptr; // flag to satisfy OpenSSL_write retry requirements
 
   SocketAsyncImpl(std::unique_ptr<SocketBufferedImpl> &&buff,
                   DriverShared &driver,
@@ -57,6 +56,8 @@ struct SocketAsyncImpl
   bool DoSendEnqueue(std::promise<void> promise, Args&&... args);
 
   // in thread context of DriverImpl
+  SOCKET DriverGetFd() const;
+  void DriverQuery(short &events);
   void DriverOnReadable();
   void DriverConnect(ConnectHandler const &onConnect);
   void DriverReceive(ReceiveHandler const &onReceive);