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";
         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();
      }
      inHandler = false;
      // 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();
      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;
	       if((size_t)sent >= fifo.front().length())
		 fifo.pop();
	       else
		 fifo.front().erase(0, sent);
	    } 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("stuck write retry loop - no progress after repeated attempts");
	    }
         }
      }
      outlock.unlock();
   }
   
   int Socket::write(std::string data) {
      outlock.lock();
      fifo.emplace(data);
      outlock.unlock();
      if(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)
        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);
   }


}