ServerCore/Socket.cpp

#include "Socket.h"
#include "EPoll.h"
#include "Exception.h"
#include "Log.h"
#include "ZString.h"
#include <iostream>

namespace core
{

   void sigpipe_handler(int unused) {}

   Socket::Socket(EPoll &ePoll, std::string text) : ePoll(ePoll), text(text) {
      coreutils::Log(coreutils::LOG_DEBUG_2) << "Socket object created [" << text << "].";
      buffer = (char *)malloc(4096);
      length = 4096;
   }

   Socket::~Socket() {
      shutDown = true;
      onUnregister();
      disableSocket();
      coreutils::Log(coreutils::LOG_DEBUG_4) << "Free on socket " << descriptor;
      free(buffer);
      if(descriptor == -1)
		  return;
      close(descriptor);
      coreutils::Log(coreutils::LOG_DEBUG_1) << text << " has ended (" << descriptor << ").";
   }

   void Socket::setDescriptor(int descriptor)
   {
      if ((descriptor == -1) && (errno == 24))
      {
         shutdown("Too many files open");
         throw coreutils::Exception("Too many files open. Refusing connection.");
      }
      coreutils::Log(coreutils::LOG_DEBUG_1) << text << " has started (" << descriptor << ").";
      if(descriptor < 3)
        throw coreutils::Exception("Descriptor out of range", __FILE__, __LINE__);
      this->descriptor = descriptor;
      onRegister();
      enableSocket();
      onRegistered();
   }

   int Socket::getDescriptor()
   {
      return descriptor;
   }

   void Socket::setBufferSize(int length)
   {
      this->length = length;
      buffer = (char *)realloc(buffer, length);
   }

   int Socket::getBufferSize()
   {
      return length;
   }

   void Socket::onRegister() {}

   void Socket::onRegistered() {}

   void Socket::onUnregister() {}

   void Socket::onUnregistered() {}

   bool Socket::eventReceived(struct epoll_event event, long long eventId) {
      coreutils::Log(coreutils::LOG_DEBUG_1) << "Event process beginning for socket " << getDescriptor();
      if(inHandler)
         coreutils::Log(coreutils::LOG_DEBUG_2) << "inHandler was already true.";
      inHandler = true;
      if(event.events & EPOLLRDHUP) {
         coreutils::Log(coreutils::LOG_DEBUG_2) << "EPOLLRDHUP";
         readHangup = true;
         shutdown("hangup received");
      }
      if(event.events & EPOLLIN) {
         coreutils::Log(coreutils::LOG_DEBUG_2) << "EPOLLIN";
         // A read just happened (or at least an opportunity for OpenSSL
         // to make progress on whatever it needed a read for) - give any
         // write that was waiting on that a fresh attempt via this same
         // call's own trailing resetSocket(), rather than leaving
         // EPOLLWRNORM permanently un-armed from a single earlier stall.
         waitingOnReadToWrite = false;
         coreutils::ZString zbuffer(buffer, length);
         // receiveData() calls all the way down through
         // onDataReceived()/HTTPRequest::doParse()/application request
         // handling with NO exception protection anywhere in that chain -
         // confirmed live: a single thrown coreutils::Exception (e.g. a
         // malformed request hitting a parse-time throw, well before
         // ApiSession::handleComplete()'s own try/catch around
         // match.render() ever runs) reached std::terminate() and killed
         // the entire server process, not just this one connection, since
         // this runs on an EPoll worker thread and C++ terminates the
         // whole program on an exception escaping a thread. The old
         // lock.lock()/lock.unlock() pair (no RAII) also leaked the lock
         // permanently on this same path - std::lock_guard fixes that as
         // a side effect of adding the try/catch here.
         try {
            std::lock_guard<std::mutex> guard(lock);
            receiveData(zbuffer);
         } catch(coreutils::Exception &e) {
            // Also to stderr, not just Log() - this process's Log class
            // writes through a static listener/file neither this server
            // nor its main() ever configures (its own startup banner
            // uses plain std::cout, not Log at all), so a Log()-only
            // message here would be silently dropped. stderr reliably
            // lands in the log file given how this process is started
            // (both stdout and stderr redirected to the same file).
            std::cerr << "Exception processing socket " << getDescriptor() << ": " << e.text << std::endl;
            coreutils::Log(coreutils::LOG_EXCEPT) << "Exception processing socket " << getDescriptor() << ": " << e.text;
            shutdown("exception during receiveData: " + e.text);
         } catch(std::exception &e) {
            std::cerr << "std::exception processing socket " << getDescriptor() << ": " << e.what() << std::endl;
            coreutils::Log(coreutils::LOG_EXCEPT) << "std::exception processing socket " << getDescriptor() << ": " << e.what();
            shutdown(std::string("exception during receiveData: ") + e.what());
         } catch(...) {
            std::cerr << "Unknown exception processing socket " << getDescriptor() << std::endl;
            coreutils::Log(coreutils::LOG_EXCEPT) << "Unknown exception processing socket " << getDescriptor();
            shutdown("unknown exception during receiveData");
         }
      }
      if(event.events & EPOLLWRNORM) {
         coreutils::Log(coreutils::LOG_DEBUG_2) << "EPOLLWRNORM";
	 writeSocket();
      }
      // A graceful shutdown() (session.terminate() -> "goodbye" reply
      // queued via write(), then shutdown() called - see e.g. IMAP's
      // LOGOUT/POP3's QUIT/SMTP's QUIT) used to make keepAlive false
      // the moment shutDown was set, in this SAME event cycle that had
      // just queued that reply - Thread::run() then deletes this object
      // immediately on a false return, before the reply was ever
      // written to the actual socket (write() only queues into fifo;
      // the real send() happens in writeSocket(), only reachable via a
      // LATER EPOLLWRNORM event this object never lived to receive).
      // Confirmed live: LOGOUT/QUIT connections closed correctly but
      // their own final reply line never reached the client. Staying
      // alive while shutDown is set but fifo still has data - as long
      // as the peer hasn't already hung up (readHangup - trying to
      // drain to a peer that's already gone is pointless, not graceful)
      // - gives that queued reply a real chance to go out before this
      // socket actually closes.
      bool keepAlive = !shutDown || (fifo.size() > 0 && !readHangup);

      // resetSocket() re-arms this fd via EPOLL_CTL_MOD, which makes it
      // immediately eligible for delivery to ANY worker thread's
      // epoll_wait() - including one already blocked and about to wake
      // up right away. The old code called resetSocket() from inside
      // the branches above while several more statements (an
      // unconditional inHandler write, this Log call, and `return
      // !shutDown`) still touched `this` afterward - a second thread
      // could legitimately receive the re-armed event, finish handling
      // it, and delete this object (Thread::run() does exactly that
      // when eventReceived() returns false) before this thread's own
      // remaining statements ran, a heap-use-after-free confirmed via
      // AddressSanitizer under concurrent load. resetSocket() must be
      // the last thing that ever touches `this`, with the return value
      // already captured in a local (`keepAlive`) beforehand.
      coreutils::Log(coreutils::LOG_DEBUG_1) << "Event process ending for socket " << getDescriptor();
      // Was cleared several statements above (right after the EPOLLIN/
      // EPOLLWRNORM handling) - a genuinely separate race from the one
      // documented above: Socket::write() re-arms via this same
      // resetSocket() (see its own doc comment there) whenever it can
      // grab `lock`, from whatever thread called it (e.g. a broadcast
      // hitting this session while ITS OWN eventReceived() call is still
      // running, past receiveData()'s lock_guard but before reaching
      // here). Clearing inHandler that early left exactly that window
      // open for write() to re-arm this fd a second, premature time,
      // letting a third thread pick up a fresh event - and potentially
      // delete this object outright - while this call was still
      // mid-flight. Keeping inHandler true until the instant before
      // resetSocket() actually runs (write() checks it, see its own
      // comment) closes that window down to the two adjacent statements
      // below, as tight as achievable without restructuring how write()
      // re-arms at all.
      inHandler = false;
      if(keepAlive)
        resetSocket();
      return keepAlive;
   }

   void Socket::connectionRequest() {}
      
   void Socket::onDataReceived(std::string data) {
      throw coreutils::Exception("Need to override onDataReceived.", __FILE__, __LINE__, -1);
   }
   
   void Socket::onDataReceived(coreutils::ZString &data) {
      onDataReceived(std::string(data.getData(), data.getLength()));
   }
   
   void Socket::receiveData(coreutils::ZString &buffer) {
      coreutils::ZString blank("");
      if(buffer.getLength() <= 0)
        throw coreutils::Exception("Request to receive data with a zero buffer length.", __FILE__, __LINE__, -1);
      int len;
      int error = -1;
      if((len = ::read(getDescriptor(), buffer.getData(), buffer.getLength())) >= 0) {
         coreutils::ZString zbuffer(buffer.getData(), len);
	 //         coreutils::Log(coreutils::LOG_DEBUG_1) << zbuffer;
         onDataReceived(zbuffer);
      } else {	   
	 error = errno;	   
	 switch (error) {
	    
            // When a listening socket receives a connection
	    // request we get one of these.
           
	  case ENOTCONN:
            connectionRequest();
            break;
	    
	  case ECONNRESET:
            break;
	    
	  default:
            throw coreutils::Exception("Error in read of data from socket.", __FILE__, __LINE__, error);
         }
      }
   }
   
   int Socket::sendData(const char *data, int length) {
      return ::write(descriptor, data, length);
   }

   void Socket::writeSocket() {
      outlock.lock();
      //      coreutils::Log(coreutils::LOG_DEBUG_3) << "writing data to socket " << getDescriptor();
      if(fifo.size() > 0) {
	 // No longer gated on !shutDown: writeSocket() only ever runs in
	 // response to an EPOLLWRNORM event this object explicitly
	 // re-armed for (see resetSocket()'s own doc comment) - refusing
	 // to actually perform that queued write once shutDown was set
	 // was exactly what left a graceful QUIT/LOGOUT's own reply
	 // stuck in the fifo forever, never reaching the client.
	 {
	    // sendData()'s return value was previously discarded and the
	    // fifo entry popped unconditionally after ONE call - correct
	    // only by accident, since a single write()/SSL_write() call is
	    // never guaranteed to accept an entire buffer. A response
	    // small enough to fit the kernel's socket send buffer in one
	    // shot (true for almost everything on this site, and always
	    // true testing over loopback, where that buffer is
	    // effectively unlimited) went out fine either way - but a
	    // large one (confirmed with a real 208KB response, mostly an
	    // embedded base64 font: fine over loopback, but only ~65KB of
	    // it ever reached a real remote client before the rest was
	    // silently dropped here) left the client waiting forever for
	    // bytes matching its own Content-Length that were never
	    // coming. Now: pop only once the FULL entry has gone out;
	    // otherwise trim off just what was actually sent and leave
	    // the remainder queued - resetSocket() (called at the end of
	    // every eventReceived(), see its own doc comment) keeps
	    // re-arming EPOLLWRNORM as long as the fifo isn't empty, so
	    // the next write-ready event retries this same entry
	    // automatically. A return <= 0 (EAGAIN or a transient error)
	    // sends nothing this round and is handled identically - just
	    // leave the entry queued for the next attempt.
	    int sent = sendData(fifo.front().c_str(), fifo.front().length());
	    if(sent > 0) {
	       consecutiveWriteStalls = 0;
	       waitingOnReadToWrite = false;
	       if((size_t)sent >= fifo.front().length())
		 fifo.pop();
	       else
		 fifo.front().erase(0, sent);
	    } else if(sendNeedsRead()) {
	       // Not a stall in the CPU-spinning sense consecutiveWriteStalls
	       // exists to catch - resetSocket() below simply won't re-arm
	       // EPOLLWRNORM while this is true, so there's nothing to busy-
	       // retry until a real EPOLLIN event (already always watched
	       // for) brings in whatever OpenSSL is actually waiting to read.
	       waitingOnReadToWrite = true;
	    } else if(++consecutiveWriteStalls > 50) {
	       // sendData() keeps failing for a reason that has nothing to
	       // do with the socket's own write-readiness (confirmed real
	       // case: TLS::sendData() returning SSL_ERROR_WANT_READ,
	       // which needs a READ to make progress, not another write
	       // attempt) - resetSocket() re-arms EPOLLWRNORM regardless,
	       // and an already-writable TCP socket makes it fire again
	       // immediately, so this cycle would otherwise spin forever
	       // doing real work at the CPU's own top speed. See
	       // consecutiveWriteStalls' own doc comment in Socket.h for
	       // the confirmed real damage (a 55GB, then 12.9GB debug log)
	       // this caused before being noticed.
	       coreutils::Log(coreutils::LOG_EXCEPT) << "Socket " << getDescriptor() << ": " << consecutiveWriteStalls << " consecutive writes made zero progress - giving up and shutting down.";
	       // shutdown() alone only sets shutDown=true and expects the
	       // fifo to drain normally before the connection actually
	       // closes (see eventReceived()'s keepAlive: "!shutDown ||
	       // (fifo.size() > 0 && !readHangup)") - exactly backwards
	       // for THIS path, since the whole reason we're here is that
	       // the front of the fifo can never be sent. Leaving it
	       // queued meant keepAlive stayed true forever, resetSocket()
	       // kept re-arming EPOLLWRNORM because fifo.size() > 0, the
	       // very next write attempt hit this identical stall
	       // immediately, and shutdown() got called again - forever,
	       // at whatever rate the kernel could fire EPOLLWRNORM.
	       // Confirmed live: 2.4 million+ repeats of this exact log
	       // line for one socket in under 4 milliseconds, pegging a
	       // worker thread at 100% CPU indefinitely (a real game
	       // server, brought down by nothing more than an ordinary
	       // port-scanner connection). Clearing the undeliverable data
	       // here lets keepAlive correctly evaluate to just !shutDown
	       // (now false) on the very next cycle, so the connection
	       // actually tears down instead of spinning forever.
	       std::queue<std::string>().swap(fifo);
	       shutdown("stuck write retry loop - no progress after repeated attempts");
	    }
         }
      }
      outlock.unlock();
   }
   
   int Socket::write(std::string data) {
      outlock.lock();
      fifo.emplace(data);
      outlock.unlock();
      // Was unconditional - re-arming here (resetSocket() -> EPOLL_CTL_MOD)
      // while this socket's OWN eventReceived() call is still mid-flight
      // on another thread (inHandler true - see eventReceived()'s own
      // comment on exactly this race) makes the fd immediately eligible
      // for a THIRD thread to pick up a fresh event and potentially
      // delete this object (Thread::run() does exactly that on a false
      // return) while the first thread is still touching it - a genuine
      // use-after-free, matching a real heap-corruption crash confirmed
      // live via dmesg under concurrent load. eventReceived() already
      // re-arms itself once it's actually done (checking the current
      // fifo, which already includes whatever was just queued above);
      // skip doing it a second, early time from here while that's still
      // in progress - the data isn't lost, just picked up on that call's
      // own re-arm instead of this one's.
      if(!inHandler && lock.try_lock()) {
         resetSocket();
         lock.unlock();
      }
      return 1;
   }

   void Socket::output(std::stringstream &out)
   {
      out << "|" << descriptor << "|";
   }

   bool Socket::needsToWrite()
   {
      return fifo.size() > 0;
   }

   void Socket::shutdown(std::string text) {
      coreutils::Log(coreutils::LOG_DEBUG_2) << "Shutdown requested from " << this->text << " (" << descriptor << ") with reason: " << text << ".";
      shutDown = true;
      reset = false;
   }

    void Socket::enableSocket() {
      struct epoll_event event;
      event.data.ptr = this;
      event.events = EPOLLIN | EPOLLRDHUP | EPOLLONESHOT | EPOLLET;
      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_ADD, getDescriptor(), &event);
   }

   void Socket::disableSocket() {
      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_DEL, getDescriptor(), NULL);
   }

   void Socket::resetSocket() {
      struct epoll_event event;
      event.data.ptr = this;
      event.events = EPOLLIN | EPOLLRDHUP | EPOLLONESHOT | EPOLLET;
      if(fifo.size() > 0 && !waitingOnReadToWrite)
        event.events |= EPOLLWRNORM;
      // Was `if(!shutDown)` - refused to re-arm at all once shutting
      // down, even with data still queued to send (e.g. a graceful
      // QUIT/LOGOUT's own final reply - see eventReceived()'s own doc
      // comment on keepAlive for the full chain this broke). Re-arming
      // for just long enough to drain the fifo, when the peer is still
      // there to receive it (!readHangup - draining to an already-gone
      // peer is pointless), is what actually lets that reply go out
      // before this socket closes.
      if(!shutDown || (fifo.size() > 0 && !readHangup))
        epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_MOD, getDescriptor(), &event);
   }


}