Merge tls into master: bring in subscription ownership, keep all TLS/stability work
tls and master diverged from a July 2024 fork point and never reconciled - tls carries ~19 months of real socket/TLS stability fixes (uninitialized tlsInfo, TCPServer::connectionRequest() unconditionally initializing TLS, the WANT_READ write busy-spin, the console authentication fix from this session), while master independently picked up a Subscription/SubscriptionManager rework (MString-based ownership tracking: owner/ownership/selection/time fields and a working subscriptionOwner() method) that tls never had. Every file both branches touched was resolved by keeping tls's side entirely (Command.h, Socket.cpp, TCPServer.h, TCPSession.cpp/h, TCPSocket.h were pure reformatting on master's side, or an alternate/superseded fix for a bug tls already fixed a different way - e.g. master's TCPSession uuid/alias handling vs tls's existing uuid_unparse-based fix), with two deliberate exceptions: - TCPServer.cpp's constructor was missing the commands.add(subscriptions, "publish"/"subscribe"/"catalog"/...) wiring master had - without it, SubscriptionManager existed but was completely unreachable through the normal command dispatch. Restored on top of tls's TLS-aware constructor. - SubscriptionManager.cpp's "invite" command handler had a real incompatibility: master's version assigned into an `invitee` member typed as coreutils::MString via a stray `(coreutils::MString *)` cast that doesn't match the member's actual (non-pointer) type, and called session.server.getSessionByAlias() against a TCPServer reference - tls made TCPSession::server a pointer (part of removing the separate TCPSession2 class), so that call site no longer compiles as written. Took tls's cleaner local-variable version instead (functionally identical - looks up a session by alias and forwards an invite message) and removed the now-unused `invitee` member from SubscriptionManager.h. One additional compile fix needed after resolving conflicts: Subscription::subscriptionOwners is an MString (master's ownership design), but TCPSession::alias is void* (tls's identity mechanism, unchanged and depended on elsewhere) - MString has an implicit constructor from void* already used successfully elsewhere in this same file (the "publish" handler's Subscription constructor call), but operator== doesn't accept the same implicit conversion, so removeSessionSubscriptions()'s ownership check needed an explicit coreutils::MString(session.alias) construction to compile. Also drops output/main (a stray checked-in compiled binary, already removed on tls) and the doxygen docs/html output (already removed on tls, was carried on master since the fork). Verified: full ServerCore rebuild (./compile) succeeds with zero errors or warnings beyond the one compile fix documented above.
This commit is contained in:
Brad Arant committed
commit
3609ed603e
2883 files changed
+5296
-84468
No files matched your search
@@ -0,0 +1,325 @@
|
||||
#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);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
Reference in new issue
Block a user