FEX-Emu--FEX/Source/Tools/FEXServer/Logger.cpp

#include "Common/FEXServerClient.h"

#include <atomic>
#include <mutex>
#include <poll.h>
#include <thread>
#include <vector>

namespace Logging {
  void ClientMsgHandler(int FD, uint64_t Timestamp, uint32_t PID, uint32_t TID, uint32_t Level, const char* Msg);
}

namespace Logger {
  std::vector<struct pollfd> PollFDs{};
  std::mutex IncomingPollFDsLock{};
  std::vector<struct pollfd> IncomingPollFDs{};
  std::thread LogThread;
  std::atomic<bool> ShouldShutdown {false};
  std::atomic<int32_t> LoggerThreadTID{};

  void HandleLogData(int Socket) {
    std::vector<uint8_t> Data(1500);
    size_t CurrentRead{};
    while (true) {
      int Read = read(Socket, &Data.at(CurrentRead), Data.size() - CurrentRead);
      if (Read > 0) {
        CurrentRead += Read;
        if (CurrentRead == Data.size()) {
          Data.resize(Data.size() << 1);
        }
        else {
          // No more to read
          break;
        }
      }
      else {
        if (errno == EWOULDBLOCK) {
          // no error
        }
        else {
          perror("read");
        }
        break;
      }
    }

    size_t CurrentOffset{};
    while (CurrentOffset < CurrentRead) {
      FEXServerClient::Logging::PacketHeader *Header = reinterpret_cast<FEXServerClient::Logging::PacketHeader*>(&Data[CurrentOffset]);
      if (Header->PacketType == FEXServerClient::Logging::PacketTypes::TYPE_MSG) {
        FEXServerClient::Logging::PacketMsg *Msg = reinterpret_cast<FEXServerClient::Logging::PacketMsg*>(&Data[CurrentOffset]);
        const char *MsgText = reinterpret_cast<const char*>(&Data[CurrentOffset + sizeof(FEXServerClient::Logging::PacketMsg)]);
        Logging::ClientMsgHandler(Socket, Msg->Header.Timestamp, Msg->Header.PID, Msg->Header.TID, Msg->Level, MsgText);

        CurrentOffset += sizeof(FEXServerClient::Logging::PacketMsg) + Msg->MessageLength;
      }
      else {
        CurrentOffset = CurrentRead;
      }
    }
  }

  void LogThreadFunc() {
    LoggerThreadTID = ::gettid();

    while (!ShouldShutdown) {
      struct timespec ts{};
      ts.tv_sec = 5;

      {
        std::unique_lock lk {IncomingPollFDsLock};
        PollFDs.insert(PollFDs.end(), std::make_move_iterator(IncomingPollFDs.begin()), std::make_move_iterator(IncomingPollFDs.end()));
        IncomingPollFDs.clear();
      }
      if (PollFDs.size() == 0) {
        pselect(0, nullptr, nullptr, nullptr, &ts, nullptr);
      }
      else {
        int Result = ppoll(&PollFDs.at(0), PollFDs.size(), &ts, nullptr);
        if (Result > 0) {
          // Walk the FDs and see if we got any results
          for (auto it = PollFDs.begin(); it != PollFDs.end(); ) {
            bool Erase{};
            if (it->revents != 0) {
              if (it->revents & POLLIN) {
                // Data from the socket
                HandleLogData(it->fd);
              }
              else if (it->revents & (POLLHUP | POLLERR | POLLNVAL | POLLRDHUP)) {
                // Error or hangup, close the socket and erase it from our list
                Erase = true;
                close(it->fd);
              }

              it->revents = 0;
              --Result;
            }

            if (Erase) {
              it = PollFDs.erase(it);
            }
            else {
              ++it;
            }

            if (Result == 0) {
              // Early break if we've consumed all the results
              break;
            }
          }
        }
      }
    }
  }

  void StartLogThread() {
    LogThread = std::thread(LogThreadFunc);
  }

  void AppendLogFD(int FD) {
    {
      std::unique_lock lk {IncomingPollFDsLock};
      IncomingPollFDs.emplace_back(pollfd {
        .fd = FD,
        .events = POLLIN,
        .revents = 0,
      });
    }

    // Wake up the thread immediately
    tgkill(::getpid(), LoggerThreadTID, SIGUSR1);
  }

  bool LogThreadRunning() {
    return LogThread.joinable();
  }

  void Shutdown() {
    ShouldShutdown = true;

    // Wake up the thread immediately
    tgkill(::getpid(), LoggerThreadTID, SIGUSR1);

    if (LogThread.joinable()) {
      LogThread.join();
    }
  }
}