ServerCore/EPoll.cpp

#include "Thread.h"
#include "EPoll.h"
#include "Command.h"
#include "Exception.h"
#include <signal.h>
#include <sys/epoll.h>

namespace core {

   EPoll::EPoll() : Command() {

      coreutils::Log(coreutils::LOG_DEBUG_2) << "EPoll starting.";

      maxSockets = 1000;
      epfd = epoll_create1(0);
      terminateThreads = false;
      started = false;
   }

   // Was empty - if this EPoll's worker pool was started but stop() was
   // never explicitly called (e.g. an exception unwinding the stack
   // during startup, before the caller's own normal shutdown path could
   // run), the pool just kept running against this object's - and every
   // Socket registered with it's - soon-to-be-freed memory. stop() itself
   // is what actually joins every worker thread before anything gets
   // torn down; see Thread::~Thread()'s doc comment for the crash this
   // was confirmed to cause. Guarded by `started` so a never-started
   // EPoll (numberOfThreads/threads still default) doesn't call stop()
   // over a garbage thread count, and by `!terminateThreads` so an
   // EPoll that already went through a normal stop() doesn't do it
   // (and close an already-closed epfd) twice.
   EPoll::~EPoll() {
      if (started && !terminateThreads)
        stop();
   }

   bool EPoll::start(int numberOfThreads, int maxSockets, ThreadScope *scope) {

      coreutils::Log(coreutils::LOG_DEBUG_2) << "Starting epoll event processing.";

      this->numberOfThreads = numberOfThreads;

      coreutils::Log(coreutils::LOG_DEBUG_3) << "Number of threads starting is " << numberOfThreads << ".";
      coreutils::Log(coreutils::LOG_DEBUG_3) << "Maximum connections is " << maxSockets << ".";

      // TODO: Set the number of maximum open files to the maxSockets value.
      //
      //----------------------------------------------------------------------
      // Create thread objects into vector for number of threads requested.
      // Hand all the threads a pointer to the EPoll object so they can run
      // the socket handlers - and, if given, the same shared ThreadScope
      // so each one runs its onThreadStart()/onThreadEnd() hooks (see
      // ThreadScope's own doc comment for why - e.g. MySQL client library
      // per-thread init).
      //----------------------------------------------------------------------

      for(int ix = 0; ix < numberOfThreads; ++ix)
        threads.emplace_back(*this, scope);

      for(int ix = 0; ix < numberOfThreads; ++ix)
        threads[ix].start();

      started = true;

      return true;
   }

   bool EPoll::stop() {

      terminateThreads = true;

      //--------------------------------------------------------
      // Kill and join all the threads that have been started.
      //--------------------------------------------------------

      for(int ix = 0; ix < numberOfThreads; ++ix)
        threads[ix].join();

      //--------------------------
      // Close the epoll socket.
      //--------------------------

      close(epfd);

      return true;
   }

   bool EPoll::isStopping() {
      return terminateThreads;
   }

   int EPoll::getDescriptor() {
      return epfd;
   }

   int EPoll::processCommand(coreutils::ZString &request, TCPSession &session) {
      int sequence = 0;
      // Was `for(auto threadx : threads)` - copied each live, running
      // Thread by value. Harmless while ~Thread() was empty, but now
      // that it actually joins and deletes the underlying std::thread*
      // (see its own doc comment), a copy going out of scope at the end
      // of each iteration would join and free the SAME thread the real
      // vector entry still owns and expects to keep running - a
      // double-free the next time anything lists threads on a live
      // server (this method is reachable via the console's own command
      // dispatch).
      for(auto &threadx : threads) {
         session.out << "|" << ++sequence;
         threadx.output(session.out);
         session.out << "|" << std::endl;
      }
      return 1;
   }

}