ServerCore/Thread.cpp

#include "Thread.h"
#include "EPoll.h"
#include "TCPServer.h"
#include "INotify.h"
#include <sys/epoll.h>

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;
	       // events[ix].data.ptr is a raw pointer of UNKNOWN validity at
	       // this point - it could be a live TCPSession, a listening
	       // TCPServer/ConsoleServer (immortal, never shared_ptr-owned),
	       // or - the case this whole block exists to handle safely - a
	       // stale/duplicate event for a TCPSession something else
	       // already fully destroyed and freed on another thread.
	       //
	       // A first version of this fix used dynamic_cast<TCPSession*>
	       // to tell these apart, and a version before that cast
	       // straight to Socket* unconditionally (the original code).
	       // BOTH require reading the pointed-to object's own memory
	       // (its vtable, at minimum) to work - confirmed live via
	       // AddressSanitizer that dynamic_cast itself segfaults on a
	       // stale pointer, before any pin()-based safety logic even
	       // runs. TCPSession::tryGet() and TCPServer::isKnownListening
	       // Socket() below never dereference this raw pointer for their
	       // own check - only compare its address against a registry -
	       // so they're safe to call on literally any value here,
	       // known-good or not.
	       void *rawPtr = events[ix].data.ptr;
	       std::shared_ptr<TCPSession> pinned = TCPSession::tryGet(rawPtr);
	       if(pinned) {
		  if(pinned->eventReceived(events[ix], ++ePoll.eventId)) {
		     // keep alive - pinned goes out of scope at the end of
		     // this iteration, releasing this thread's temporary
		     // reference; real ownership (TCPServer::sessions) is
		     // unaffected.
		  } else {
		     pinned->shutDown = true;
		     // Was `delete (Socket *)events[ix].data.ptr;`
		     // unconditionally, for every socket type, including
		     // real sessions another thread might be mid-call on
		     // right now via its own pin()'d reference. Dropping
		     // the one real owning reference is what actually
		     // frees the object - immediately, if this thread's
		     // own `pinned` (about to go out of scope) was the
		     // last one outstanding; safely deferred otherwise.
		     pinned->server->removeFromSessionList(pinned.get());
		  }
	       } else if(TCPServer::isKnownListeningSocket(rawPtr)) {
		  // Known-immortal (see its own doc comment) - safe to call
		  // directly, unchanged from the original behavior. A
		  // listening socket's own eventReceived() returning false
		  // isn't something that happens in normal operation, but
		  // preserved as-is for parity with the pre-existing code.
		  Socket *rawSocket = (Socket *)rawPtr;
		  if(rawSocket->eventReceived(events[ix], ++ePoll.eventId)) {
		     // keep alive
		  } else {
		     rawSocket->shutDown = true;
		     delete rawSocket;
		  }
	       } else if(INotify::isKnownINotify(rawPtr)) {
		  // A third category of long-lived, non-TCPSession Socket -
		  // see INotify::isKnownINotify()'s own doc comment for the
		  // real, confirmed-live bug this fixes: every INotify event
		  // (e.g. BARANTMail's mail-queue delivery watch) previously
		  // fell all the way through to the "unrecognized" branch
		  // below and was silently dropped forever.
		  Socket *rawSocket = (Socket *)rawPtr;
		  if(rawSocket->eventReceived(events[ix], ++ePoll.eventId)) {
		     // keep alive
		  } else {
		     rawSocket->shutDown = true;
		     delete rawSocket;
		  }
	       } else {
		  // Neither registry recognizes this pointer - almost
		  // certainly a stale/duplicate epoll event for a
		  // TCPSession that has already been fully destroyed and
		  // unregistered elsewhere. There is nothing safe to do
		  // with it except leave it alone entirely.
		  coreutils::Log(coreutils::LOG_DEBUG_1) << "Ignoring stale/unrecognized epoll event (pointer " << rawPtr << ")";
	       }
            }
	 }
      }

      if(thread)
	thread->onThreadEnd();

      coreutils::Log(coreutils::LOG_DEBUG_1) << "Thread ending with thread id " << threadId << ".";
   }
   
}