#include "Socket.h" #include "EPoll.h" #include "Exception.h" #include "Log.h" #include "ZString.h" #include 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 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; 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().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(); 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 && !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); } }