#include "Thread.h" #include "EPoll.h" #include "Command.h" #include "Exception.h" #include #include 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; } }