Merge pull request #4565 from neobrain/fix_async_timeout

FEXServer: Don't time out while clients are still connected

Source/Common/Async.h

@@ -54,6 +54,7 @@ private:
   std::vector<pollfd> PollFDs;
   std::optional<int> CurrentFD; // FD that is currently being processed
 
+  bool is_stopped = false;
   int AsyncStopRequest[2] = {-1, -1};
 
   // Maps FD to callback
@@ -91,12 +92,22 @@ public:
     ::close(AsyncStopRequest[1]);
   }
 
-  error run(std::optional<std::chrono::nanoseconds> Timeout = std::nullopt) {
+  void cleanup() {
+    callbacks.clear();
+  }
+
+  [[nodiscard]]
+  bool stopped() const {
+    return is_stopped;
+  }
+
+  error run_one(std::optional<std::chrono::nanoseconds> Timeout = std::nullopt) {
     // Process events queued before entering wait loop
     update_fd_list();
 
     timespec ts = to_timespec(Timeout.value_or(std::chrono::nanoseconds {0}));
 
+    // ppoll may return EINTR/EAGAIN, so a loop is used here. Normally, we return in the first iteration.
     while (true) {
       int Result = ::ppoll(PollFDs.data(), PollFDs.size(), Timeout ? &ts : nullptr, nullptr);
 
@@ -104,14 +115,10 @@ public:
         if (errno == EINTR || errno == EAGAIN) {
           continue;
         }
-        callbacks.clear();
         return error::generic_errno;
       } else if (Result == 0) {
-        callbacks.clear();
         return error::timeout;
       } else {
-        bool exit_requested = false;
-
         // Walk the FDs and see if we got any results
         for (auto& ActiveFD : PollFDs) {
           if (ActiveFD.revents == 0) {
@@ -139,14 +146,14 @@ public:
                 ActiveFD.revents = 0;
               }
             } else if (Ret == post_callback::stop_reactor) {
-              exit_requested = true;
+              is_stopped = true;
             }
             CurrentFD.reset();
           }
           if (ActiveFD.revents & (POLLHUP | POLLERR | POLLNVAL | POLLRDHUP)) {
             auto Callback = std::move(callbacks[ActiveFD.fd]);
             if (Callback) {
-              exit_requested |= (Callback(error::eof) == post_callback::stop_reactor);
+              is_stopped |= (Callback(error::eof) == post_callback::stop_reactor);
             }
             // Error or hangup, erase the socket from our list
             QueuedEvents.push_back(Event {.FD = {.fd = ActiveFD.fd}, .Erase = true});
@@ -155,12 +162,23 @@ public:
           ActiveFD.revents = 0;
         }
 
-        if (exit_requested) {
-          callbacks.clear();
+        if (is_stopped) {
+          cleanup();
           return error::success;
         }
 
         update_fd_list();
+        return error::success;
+      }
+    }
+  }
+
+  error run(std::optional<std::chrono::nanoseconds> Timeout = std::nullopt) {
+    while (true) {
+      auto Result = run_one(Timeout);
+      if (Result != error::success || is_stopped) {
+        cleanup();
+        return Result;
       }
     }
   }

Source/Tools/FEXServer/ProcessPipe.cpp

@@ -24,6 +24,7 @@ constexpr int USER_PERMS = S_IRWXU | S_IRWXG | S_IRWXO;
 int ServerLockFD {-1};
 std::optional<fasio::tcp_acceptor> ServerAcceptor;
 std::optional<fasio::tcp_acceptor> ServerFSAcceptor;
+int NumClients = 0;
 time_t RequestTimeout {10};
 bool Foreground {false};
 std::vector<struct pollfd> PollFDs {};
@@ -210,6 +211,7 @@ bool InitializeServerSocket(bool abstract) {
     }
 
     int FD = Socket->FD;
+    ++NumClients;
     Reactor.bind_handler(
       pollfd {
         .fd = FD,
@@ -219,6 +221,7 @@ bool InitializeServerSocket(bool abstract) {
       [Socket = std::move(Socket).value()](fasio::error ec) mutable {
       if (ec != fasio::error::success) {
         close(Socket.FD);
+        --NumClients;
         return fasio::post_callback::drop;
       }
       HandleSocketData(Socket);
@@ -389,7 +392,18 @@ void CloseConnections() {
 
 void WaitForRequests() {
   Reactor.enable_async_stop();
-  Reactor.run(Foreground ? std::nullopt : std::optional {std::chrono::seconds {RequestTimeout}});
+
+  while (true) {
+    std::optional Timeout = std::chrono::seconds {RequestTimeout};
+    if (Foreground || NumClients > 0) {
+      Timeout.reset();
+    }
+    auto Result = Reactor.run_one(Timeout);
+    if (Result != fasio::error::success || Reactor.stopped()) {
+      Reactor.cleanup();
+      break;
+    }
+  }
 
   LogMan::Msg::DFmt("[FEXServer] Shutting Down");