Merge pull request #4223 from Sonicadvance1/netstream_timeout

GdbServer: Implement new netstream that can be interrupted

Source/Tools/LinuxEmulation/LinuxSyscalls/GdbServer.cpp

@@ -64,7 +64,7 @@ namespace FEX {
 #ifndef _WIN32
 void GdbServer::Break(FEXCore::Core::InternalThreadState* Thread, int signal) {
   std::lock_guard lk(sendMutex);
-  if (!CommsStream) {
+  if (!CommsStream.HasSocket()) {
     return;
   }
 
@@ -73,7 +73,7 @@ void GdbServer::Break(FEXCore::Core::InternalThreadState* Thread, int signal) {
   CurrentDebuggingThread = ThreadObject->ThreadInfo.TID.load();
 
   const auto str = fextl::fmt::format("T{:02x}thread:{:x};", signal, CurrentDebuggingThread);
-  SendPacket(*CommsStream, str);
+  SendPacket(str);
 }
 
 void GdbServer::WaitForThreadWakeup() {
@@ -178,7 +178,7 @@ static fextl::string encodeHex(std::string_view str) {
 // Takes a serial stream and reads a single packet
 // Un-escapes chars, checks the checksum and request a retransmit if it fails.
 // Once the checksum is validated, it acknowledges and returns the packet in a string
-fextl::string GdbServer::ReadPacket(std::iostream& stream) {
+fextl::string GdbServer::ReadPacket() {
   fextl::string packet {};
 
   // The GDB "Remote Serial Protocal" was originally 7bit clean for use on serial ports.
@@ -190,9 +190,9 @@ fextl::string GdbServer::ReadPacket(std::iostream& stream) {
   // where any $ or # in the packet body are escaped ('}' followed by the char XORed with 0x20)
   // The checksum is a single unsigned byte sum of the data, hex encoded.
 
-  int c;
-  while ((c = stream.get()) > 0) {
-    switch (c) {
+  Utils::NetStream::ReturnGet c;
+  while ((c = CommsStream.get()).HasData()) {
+    switch (c.GetData()) {
     case '$': // start of packet
       if (packet.size() != 0) {
         LogMan::Msg::EFmt("Dropping unexpected data: \"{}\"", packet);
@@ -203,15 +203,23 @@ fextl::string GdbServer::ReadPacket(std::iostream& stream) {
       break;
     case '}': // escape char
     {
-      char escaped;
-      stream >> escaped;
-      packet.push_back(escaped ^ 0x20);
+      Utils::NetStream::ReturnGet escaped;
+
+      do {
+        escaped = CommsStream.get();
+      } while (!escaped.HasData() && !escaped.HasHangup());
+
+      if (escaped.HasData()) {
+        packet.push_back(escaped.GetData() ^ 0x20);
+      } else {
+        LogMan::Msg::EFmt("Received Invalid escape char: ${}", packet);
+      }
       break;
     }
     case '#': // end of packet
     {
       char hexString[3] = {0, 0, 0};
-      stream.read(hexString, 2);
+      CommsStream.read(hexString, 2, true);
       int expected_checksum = std::strtoul(hexString, nullptr, 16);
 
       if (calculateChecksum(packet) == expected_checksum) {
@@ -221,7 +229,7 @@ fextl::string GdbServer::ReadPacket(std::iostream& stream) {
       }
       break;
     }
-    default: packet.push_back((char)c); break;
+    default: packet.push_back(c.GetData()); break;
     }
   }
 
@@ -248,22 +256,22 @@ static fextl::string escapePacket(const fextl::string& packet) {
   return ss.str();
 }
 
-void GdbServer::SendPacket(std::ostream& stream, const fextl::string& packet) {
+void GdbServer::SendPacket(const fextl::string& packet) {
   const auto escaped = escapePacket(packet);
   const auto str = fextl::fmt::format("${}#{:02x}", escaped, calculateChecksum(escaped));
 
-  stream << str << std::flush;
+  CommsStream.SendPacket(str);
 }
 
-void GdbServer::SendACK(std::ostream& stream, bool NACK) {
+void GdbServer::SendACK(bool NACK) {
   if (NoAckMode) {
     return;
   }
 
   if (NACK) {
-    stream << "-" << std::flush;
+    CommsStream.SendPacket("-");
   } else {
-    stream << "+" << std::flush;
+    CommsStream.SendPacket("+");
   }
 
   if (SettingNoAckMode) {
@@ -1341,16 +1349,16 @@ GdbServer::HandledPacketType GdbServer::ProcessPacket(const fextl::string& packe
 void GdbServer::SendPacketPair(const HandledPacketType& response) {
   std::lock_guard lk(sendMutex);
   if (response.TypeResponse == HandledPacketType::TYPE_ACK || response.TypeResponse == HandledPacketType::TYPE_ONLYACK) {
-    SendACK(*CommsStream, false);
+    SendACK(false);
   } else if (response.TypeResponse == HandledPacketType::TYPE_NACK || response.TypeResponse == HandledPacketType::TYPE_ONLYNACK) {
-    SendACK(*CommsStream, true);
+    SendACK(true);
   }
 
   if (response.TypeResponse == HandledPacketType::TYPE_UNKNOWN) {
-    SendPacket(*CommsStream, "");
+    SendPacket("");
   } else if (response.TypeResponse != HandledPacketType::TYPE_ONLYNACK && response.TypeResponse != HandledPacketType::TYPE_ONLYACK &&
              response.TypeResponse != HandledPacketType::TYPE_NONE) {
-    SendPacket(*CommsStream, response.Response);
+    SendPacket(response.Response);
   }
 }
 
@@ -1362,7 +1370,7 @@ GdbServer::WaitForConnectionResult GdbServer::WaitForConnection() {
     int Result = ppoll(&PollFD, 1, nullptr, nullptr);
     if (Result > 0) {
       if (PollFD.revents & POLLIN) {
-        CommsStream = OpenSocket();
+        OpenSocket();
         return WaitForConnectionResult::CONNECTION;
       } else if (PollFD.revents & (POLLHUP | POLLERR | POLLNVAL)) {
         // Listen socket error or shutting down
@@ -1392,47 +1400,52 @@ void GdbServer::GdbServerLoop() {
 
     HandledPacketType response {};
 
-    // Outer server loop. Handles packet start, ACK/NAK and break
+    while (!CoreShuttingDown.load()) {
+      // Outer server loop. Handles packet start, ACK/NAK and break
+      Utils::NetStream::ReturnGet c;
+      while ((c = CommsStream.get()).HasData()) {
+        switch (c.GetData()) {
+        case '$': {
+          auto packet = ReadPacket();
+          response = ProcessPacket(packet);
+          SendPacketPair(response);
+          if (response.TypeResponse == HandledPacketType::TYPE_UNKNOWN) {
+            LogMan::Msg::DFmt("Unknown packet {}", packet);
+          }
+          break;
+        }
+        case '+':
+          // ACK, do nothing.
+          break;
+        case '-':
+          // NAK, Resend requested
+          {
+            std::lock_guard lk(sendMutex);
+            SendPacket(response.Response);
+          }
+          break;
+        case '\x03': { // ASCII EOT
+          SyscallHandler->TM.Pause();
+          fextl::string str = fextl::fmt::format("T02thread:{:02x};", getpid());
+          if (LibraryMapChanged) {
+            // If libraries have changed then let gdb know
+            str += "library:1;";
+          }
+          SendPacketPair({std::move(str), HandledPacketType::TYPE_ACK});
+          break;
+        }
+        default: LogMan::Msg::DFmt("GdbServer: Unexpected byte {} ({:02x})", c.GetData(), c.GetData());
+        }
+      }
 
-    int c;
-    while ((c = CommsStream->get()) >= 0) {
-      switch (c) {
-      case '$': {
-        auto packet = ReadPacket(*CommsStream);
-        response = ProcessPacket(packet);
-        SendPacketPair(response);
-        if (response.TypeResponse == HandledPacketType::TYPE_UNKNOWN) {
-          LogMan::Msg::DFmt("Unknown packet {}", packet);
-        }
+      if (c.HasHangup()) {
         break;
       }
-      case '+':
-        // ACK, do nothing.
-        break;
-      case '-':
-        // NAK, Resend requested
-        {
-          std::lock_guard lk(sendMutex);
-          SendPacket(*CommsStream, response.Response);
-        }
-        break;
-      case '\x03': { // ASCII EOT
-        SyscallHandler->TM.Pause();
-        fextl::string str = fextl::fmt::format("T02thread:{:02x};", getpid());
-        if (LibraryMapChanged) {
-          // If libraries have changed then let gdb know
-          str += "library:1;";
-        }
-        SendPacketPair({std::move(str), HandledPacketType::TYPE_ACK});
-        break;
-      }
-      default: LogMan::Msg::DFmt("GdbServer: Unexpected byte {} ({:02x})", static_cast<char>(c), c);
-      }
     }
 
     {
       std::lock_guard lk(sendMutex);
-      CommsStream.reset();
+      CommsStream.InvalidateSocket();
     }
   }
 
@@ -1504,14 +1517,14 @@ void GdbServer::CloseListenSocket() {
   unlink(GdbUnixSocketPath.c_str());
 }
 
-fextl::unique_ptr<std::iostream> GdbServer::OpenSocket() {
+void GdbServer::OpenSocket() {
   // Block until a connection arrives
   struct sockaddr_storage their_addr {};
   socklen_t addr_size {};
 
   int new_fd = accept(ListenSocket, (struct sockaddr*)&their_addr, &addr_size);
 
-  return fextl::make_unique<FEXCore::Utils::NetStream>(new_fd);
+  CommsStream.OpenSocket(new_fd);
 }
 
 #endif

Source/Tools/LinuxEmulation/LinuxSyscalls/GdbServer.h

@@ -14,11 +14,11 @@ $end_info$
 #include <FEXCore/fextl/string.h>
 
 #include <atomic>
-#include <istream>
 #include <memory>
 #include <mutex>
 #include <stdint.h>
 
+#include "LinuxSyscalls/NetStream.h"
 #include "LinuxSyscalls/SignalDelegator.h"
 
 namespace FEX {
@@ -45,12 +45,12 @@ private:
     ERROR,
   };
   WaitForConnectionResult WaitForConnection();
-  fextl::unique_ptr<std::iostream> OpenSocket();
+  void OpenSocket();
   void StartThread();
-  fextl::string ReadPacket(std::iostream& stream);
-  void SendPacket(std::ostream& stream, const fextl::string& packet);
+  fextl::string ReadPacket();
+  void SendPacket(const fextl::string& packet);
 
-  void SendACK(std::ostream& stream, bool NACK);
+  void SendACK(bool NACK);
 
   Event ThreadBreakEvent {};
   void WaitForThreadWakeup();
@@ -147,7 +147,7 @@ private:
   FEX::HLE::SyscallHandler* const SyscallHandler;
   FEX::HLE::SignalDelegator* SignalDelegation;
   fextl::unique_ptr<FEXCore::Threads::Thread> gdbServerThread;
-  fextl::unique_ptr<std::iostream> CommsStream;
+  FEX::Utils::NetStream CommsStream;
   std::mutex sendMutex;
   bool SettingNoAckMode {false};
   bool NoAckMode {false};

Source/Tools/LinuxEmulation/LinuxSyscalls/NetStream.cpp

@@ -4,128 +4,104 @@
 #include <FEXCore/Utils/Allocator.h>
 #include <FEXCore/Utils/LogManager.h>
 
-#include <array>
 #include <cstring>
-#include <iterator>
-#ifndef _WIN32
+#include <sys/eventfd.h>
 #include <sys/socket.h>
-#endif
+#include <poll.h>
 #include <unistd.h>
 
-namespace FEXCore::Utils {
-namespace {
-  class NetBuf final : public std::streambuf, public FEXCore::Allocator::FEXAllocOperators {
-  public:
-    explicit NetBuf(int socketfd)
-      : socket {socketfd} {
-      reset_output_buffer();
-    }
-    ~NetBuf() override {
-      close(socket);
-    }
+namespace FEX::Utils {
 
-  private:
-    std::streamsize xsputn(const char* buffer, std::streamsize size) override;
-
-    std::streambuf::int_type underflow() override;
-    std::streambuf::int_type overflow(std::streambuf::int_type ch) override;
-    int sync() override;
-
-    void reset_output_buffer() {
-      // we always leave room for one extra char
-      setp(std::begin(output_buffer), std::end(output_buffer) - 1);
-    }
-
-    int flushBuffer(const char* buffer, size_t size);
-
-    int socket;
-    std::array<char, 1400> output_buffer;
-    std::array<char, 1500> input_buffer; // enough for a typical packet
-  };
-
-  int NetBuf::flushBuffer(const char* buffer, size_t size) {
-#ifndef _WIN32
-    size_t total = 0;
-
-    // Send data
-    while (total < size) {
-      size_t sent = send(socket, (const void*)(buffer + total), size - total, MSG_NOSIGNAL);
-      if (sent == -1) {
-        // lets just assume all errors are end of file.
-        return -1;
-      }
-      total += sent;
-    }
-
-    return 0;
-#else
-    ERROR_AND_DIE_FMT("Unsupported");
-#endif
-  }
-
-  std::streamsize NetBuf::xsputn(const char* buffer, std::streamsize size) {
-    size_t buf_remaining = epptr() - pptr();
-
-    // Check if the string fits neatly in our buffer
-    if (size <= buf_remaining) {
-      ::memcpy(pptr(), buffer, size);
-      pbump(size);
-      return size;
-    }
-
-    // Otherwise, flush the buffer first
-    if (sync() < 0) {
-      return traits_type::eof();
-    }
-
-    if (size > sizeof(output_buffer) / 2) {
-      // If we have a large string, bypass the buffer
-      flushBuffer(buffer, size);
-      return size;
-    } else {
-      return xsputn(buffer, size);
-    }
-  }
-
-  std::streambuf::int_type NetBuf::overflow(std::streambuf::int_type ch) {
-    // we always leave room for one extra char
-    *pptr() = (char)ch;
-    pbump(1);
-    return sync();
-  }
-
-  int NetBuf::sync() {
-    // Flush and reset output buffer to zero
-    if (flushBuffer(pbase(), pptr() - pbase()) < 0) {
-      return -1;
-    }
-    reset_output_buffer();
-    return 0;
-  }
-
-  std::streambuf::int_type NetBuf::underflow() {
-#ifndef _WIN32
-    ssize_t size = recv(socket, (void*)std::begin(input_buffer), sizeof(input_buffer), 0);
-
-    if (size <= 0) {
-      setg(nullptr, nullptr, nullptr);
-      return traits_type::eof();
-    }
-
-    setg(&input_buffer[0], &input_buffer[0], &input_buffer[size]);
-
-    return traits_type::to_int_type(*gptr());
-#else
-    ERROR_AND_DIE_FMT("Unsupported");
-#endif
-  }
-} // Anonymous namespace
-
-NetStream::NetStream(int socketfd)
-  : std::iostream(new NetBuf(socketfd)) {}
-
-NetStream::~NetStream() {
-  delete rdbuf();
+NetStream::NetStream()
+  : receive_buffer(1500) {
+  eventfd = ::eventfd(0, EFD_CLOEXEC);
 }
 
-} // namespace FEXCore::Utils
+NetStream::ReturnGet NetStream::get() {
+  if (read_offset != receive_buffer.size() && read_offset != receive_offset) {
+    auto Result = receive_buffer.at(read_offset);
+    ++read_offset;
+    return NetStream::ReturnGet {Result};
+  }
+
+  if (read_offset == receive_buffer.size()) {
+    read_offset = 0;
+    receive_offset = 0;
+  }
+
+  struct pollfd pfds[2] = {
+    {
+      .fd = socketfd,
+      .events = POLLIN,
+      .revents = 0,
+    },
+
+    {
+      .fd = eventfd,
+      .events = POLLIN,
+      .revents = 0,
+    },
+  };
+
+  auto Result = poll(pfds, sizeof(pfds), -1);
+  if (Result > 0) {
+    for (auto& pfd : pfds) {
+      if (pfd.fd == eventfd && (pfd.revents & POLLIN)) {
+        // Interrupted by eventfd.
+        uint64_t data;
+        ::read(pfd.fd, &data, sizeof(data));
+        break;
+      }
+
+      if (pfd.revents & POLLHUP) {
+        return NetStream::ReturnGet {true};
+      }
+
+      const auto remaining_size = receive_buffer.size() - receive_offset;
+      Result = ::recv(socketfd, &receive_buffer.at(receive_offset), remaining_size, 0);
+      if (Result > 0) {
+        receive_offset += Result;
+        auto Result = receive_buffer.at(read_offset);
+        ++read_offset;
+        return NetStream::ReturnGet {Result};
+      }
+    }
+  }
+
+  return NetStream::ReturnGet {false};
+}
+
+size_t NetStream::read(char* buf, size_t size, bool ContinueOnInterrupt) {
+  size_t Read {};
+  while (Read < size) {
+    auto Result = get();
+    if (Result.HasData()) {
+      buf[Read] = Result.GetData();
+      ++Read;
+    } else if ((!Result.HasData() && !ContinueOnInterrupt) || Result.HasHangup()) {
+      return Read;
+    }
+  }
+  return Read;
+}
+
+void NetStream::InterruptReader() {
+  uint64_t data {1};
+  ::write(eventfd, &data, sizeof(data));
+}
+
+bool NetStream::SendPacket(const fextl::string& packet) {
+  size_t Total {};
+  while (Total < packet.size()) {
+    size_t Remaining = packet.size() - Total;
+    size_t sent = ::send(socketfd, &packet.at(Total), Remaining, MSG_NOSIGNAL);
+    if (sent == -1) {
+      return false;
+    }
+    Total += sent;
+  }
+
+  return Total == packet.size();
+}
+
+} // namespace FEX::Utils

Source/Tools/LinuxEmulation/LinuxSyscalls/NetStream.h

@@ -2,13 +2,52 @@
 #pragma once
 
 #include <FEXCore/Utils/CompilerDefs.h>
+#include <FEXCore/fextl/string.h>
+#include <FEXCore/fextl/vector.h>
 
-#include <iostream>
+#include <optional>
+#include <variant>
 
-namespace FEXCore::Utils {
-class FEX_DEFAULT_VISIBILITY NetStream : public std::iostream {
+namespace FEX::Utils {
+class NetStream final {
 public:
-  explicit NetStream(int socketfd);
-  ~NetStream() override;
+  NetStream();
+
+  void OpenSocket(int _socketfd) {
+    socketfd = _socketfd;
+  }
+
+  void InvalidateSocket() {
+    socketfd = -1;
+  }
+
+  bool HasSocket() const {
+    return socketfd != -1;
+  }
+
+  struct ReturnGet final : public std::variant<char, bool> {
+    bool HasHangup() const {
+      return std::holds_alternative<bool>(*this) && std::get<bool>(*this);
+    }
+    bool HasData() const {
+      return std::holds_alternative<char>(*this);
+    }
+    char GetData() const {
+      return std::get<char>(*this);
+    }
+  };
+
+  ReturnGet get();
+  size_t read(char* buf, size_t size, bool ContinueOnInterrupt);
+
+  void InterruptReader();
+
+  bool SendPacket(const fextl::string& packet);
+private:
+  int socketfd {-1};
+  int eventfd {-1};
+  size_t read_offset {};
+  size_t receive_offset {};
+  fextl::vector<char> receive_buffer;
 };
-} // namespace FEXCore::Utils
+} // namespace FEX::Utils