#include "Thread.h" #include "EPoll.h" #include namespace core { Thread::Thread(EPoll &ePoll) : ePoll(ePoll), _thread(NULL), thread(NULL) {} // Previously discarded `scope` entirely (never assigned to the `thread` // member) - a real, live bug found while adding the first actual caller // of this constructor (APIServer's MysqlThreadScope): the whole point // of passing a scope in was to have it invoked from run() below, which // silently never happened. Thread::Thread(EPoll &ePoll, ThreadScope *scope) : ePoll(ePoll), _thread(NULL), thread(scope) {} // Was empty - _thread (heap-allocated in start()) was never joined or // freed here. A joinable std::thread destroyed without join()/detach() // calls std::terminate() immediately, which this leaked-pointer // approach silently sidestepped: the wrapped OS thread just kept // running - polling this Thread's EPoll reference, touching Socket // objects registered with it - even after this object, its EPoll, and // every one of those sockets had already been destroyed out from under // it (confirmed live: an exception thrown during BARANTMail startup, // after EPoll::start() had already spun up its worker pool, unwound // the stack destructing EPoll and its registered sockets while those // still-running threads kept dereferencing them, producing a real // "double free or corruption" crash). join() first (skipped if already // joined via EPoll::stop() having run normally) so the OS thread is // actually gone before anything it might touch gets freed. Thread::~Thread() { if (_thread) { if (_thread->joinable()) _thread->join(); delete _thread; } } void Thread::start() { _thread = new std::thread(&Thread::run, this); } void Thread::join() { _thread->join(); } std::string Thread::getStatus() { return status; } pid_t Thread::getThreadId() { return threadId; } int Thread::getCount() { return count; } void Thread::output(std::stringstream &data) { data << "|" << getThreadId(); data << "|" << getStatus(); data << "|" << getCount(); } void Thread::run() { threadId = syscall(SYS_gettid); coreutils::Log(coreutils::LOG_DEBUG_1) << "Thread started with thread id " << threadId << "."; if(thread) thread->onThreadStart(); count = 0; struct epoll_event events[50]; while(1) { if(ePoll.isStopping()) break; status = "WAITING"; // A finite timeout, not the previous -1 (wait forever), is what // makes EPoll::stop() able to return at all: a thread idly // blocked here with no socket activity would otherwise never // re-check isStopping() above, so join()ing it (both // EPoll::stop() and, now, ~Thread() itself) would hang forever. // Costs up to a one-second shutdown latency. int rc = epoll_wait(ePoll.getDescriptor(), events, 50, 1000); status = "RUNNING"; if(rc < 0) { // TODO: Make log entry indicating status received and ignore for now. } else if(rc == 0) { // Plain timeout, no socket activity - not itself a signal to // stop (isStopping() above already decides that). Previously // broke out of the loop unconditionally, which the old // infinite wait never actually triggered in practice, but // would have silently ended this thread after any single // idle second once the wait became finite. } else if(rc > 0) { for(int ix = 0; ix < rc; ++ix) { ++count; if(((Socket *)events[ix].data.ptr)->eventReceived(events[ix], ++ePoll.eventId)) { // coreutils::Log(coreutils::LOG_DEBUG_4) << "return true"; // ePoll.resetSocket((Socket *)events[ix].data.ptr); } else { ((Socket *)events[ix].data.ptr)->shutDown = true; // coreutils::Log(coreutils::LOG_DEBUG_4) << "return false"; delete (Socket *)events[ix].data.ptr; } } } } if(thread) thread->onThreadEnd(); coreutils::Log(coreutils::LOG_DEBUG_1) << "Thread ending with thread id " << threadId << "."; } }