Move TCPSession lifetime to shared_ptr ownership, fix the whole double-delete/UAF class of crash
Every crash root-caused earlier tonight (Subscription::subscribers, TCPServer::sessions, the Socket::write()/eventReceived() re-arm race) was the same underlying shape: some thread calling a method - or Thread::run() calling delete - on a TCPSession* while another thread concurrently freed that same object, because nothing actually guaranteed the object stayed alive for the duration of a call. Mutex-based patches at each individual call site kept narrowing the window without closing it (confirmed live, repeatedly, via AddressSanitizer under concurrent load - each fix produced a new, different crash signature at the same underlying race), and adding more locks around every write() call also works against the fifo-based design's own goal: no thread should have to block just to hand a message to another thread's session. TCPSession now derives from std::enable_shared_from_this<TCPSession> and exposes pin() - a safe upgrade from a stored raw pointer to a real, ownership-extending shared_ptr, returning nullptr instead of throwing once the object's real owner has let go. TCPServer::sessions is now the one real strong-owning shared_ptr<TCPSession> per live session; every other place that stores a TCPSession* (Subscription::subscribers, Zephora's own sessionsByPlayerName, GroupObject's membership list, etc.) is just a non-owning observer and must pin() before touching the object. A second, distinct problem: Thread::run()'s epoll dispatch starts from a raw void* of unknown validity. An early version of this fix used dynamic_cast<TCPSession*> to identify session objects before calling pin() - confirmed live via AddressSanitizer that dynamic_cast itself segfaults on a stale pointer, since it has to read the (potentially freed) object's own vtable to work at all. TCPSession::tryGet()/TCPServer:: isKnownListeningSocket() replace that: both look the raw pointer up as a key in a registry, never dereferencing it for the check itself - safe to call on any value, live session, immortal listening socket, or fully freed garbage alike. Verified via an AddressSanitizer-instrumented build under sustained concurrent load (up to 15-way parallel connect/subscribe/create/disconnect cycles, multiple rounds, 100+ total connections) with zero errors, then against the real production build with a genuine 500-connection load test (463/500 succeeded; the rest hit the process's own file-descriptor ulimit under peak burst, an unrelated and separately deferred tuning matter, not a crash). Files touched: TCPSession.h (enable_shared_from_this, pin(), the tryGet() registry), TCPServer.h/.cpp (sessions as shared_ptr, isKnownListening Socket() registry, lock made protected for subclasses), Thread.cpp (the whole dispatch rewrite), Subscription.cpp/SubscriptionManager.cpp (every subscriber write() site now pin()s first), ConsoleServer.cpp (logSend() was iterating `sessions` with no lock at all - fixed alongside the type change it needed anyway).
This commit is contained in:
1 parent
8009564a7a
commit
d058322cc6
10 files changed
+406
-59
No files matched your search
+7
-2
@@ -12,8 +12,13 @@ namespace core {
|
||||
}
|
||||
|
||||
void ConsoleServer::logSend(std::string out) {
|
||||
for(auto *session : sessions)
|
||||
((ConsoleSession *)session)->writeLog(out);
|
||||
// Was unlocked entirely (and iterated raw pointers before
|
||||
// TCPServer::sessions became shared_ptr-owned - see its own doc
|
||||
// comment) despite connectionRequest()/removeFromSessionList()
|
||||
// both mutating this same vector under `lock` from other threads.
|
||||
std::lock_guard<std::mutex> guard(lock);
|
||||
for(auto &session : sessions)
|
||||
((ConsoleSession *)session.get())->writeLog(out);
|
||||
}
|
||||
|
||||
TCPSession * ConsoleServer::getSocketAccept(EPoll &ePoll) {
|
||||
|
||||
+56
-22
@@ -144,7 +144,7 @@ namespace core
|
||||
// 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);
|
||||
bool keepAlive = preventSelfDelete || !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
|
||||
@@ -162,24 +162,44 @@ namespace core
|
||||
// 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
|
||||
// EPOLLWRNORM handling), with no lock held around clearing it or
|
||||
// calling resetSocket() below - 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)
|
||||
// here). A first attempt narrowed this to just checking `inHandler`
|
||||
// in both places, without holding `lock` across the check-and-act
|
||||
// in either one - still a real TOCTOU window (inHandler read as
|
||||
// false here, then write() on another thread sees false too and
|
||||
// wins the race to resetSocket() first), confirmed live via
|
||||
// AddressSanitizer as a genuine double-delete of the same session
|
||||
// object (two threads both reaching ~TCPSession() on it) even
|
||||
// after that narrowing. Holding `lock` for the whole
|
||||
// clear-then-reset sequence - the exact same mutex write() already
|
||||
// takes before its own resetSocket() call - makes the two
|
||||
// mutually exclusive for real: whichever thread gets there first
|
||||
// fully finishes (clearing inHandler and re-arming, or observing
|
||||
// inHandler still true and skipping) before the other can act at
|
||||
// all, closing the window rather than just narrowing it.
|
||||
if(keepAlive) {
|
||||
lock.lock();
|
||||
inHandler = false;
|
||||
resetSocket();
|
||||
lock.unlock();
|
||||
}
|
||||
// else: this object is about to be delete()'d by Thread::run() the
|
||||
// moment this call returns, unconditionally, regardless of
|
||||
// anything write() does. Leaving inHandler true forever (never
|
||||
// clearing it here) means no other thread's write() can ever
|
||||
// resetSocket() this dying object in the meantime - it has no
|
||||
// future to reset it FOR, and letting write() re-arm it here, even
|
||||
// briefly, was exactly the remaining gap: a third thread could
|
||||
// pick up that re-armed fd and touch/delete an object Thread::run()
|
||||
// was already about to free, a genuine double-delete confirmed
|
||||
// live via AddressSanitizer even after write()/resetSocket() were
|
||||
// otherwise fully serialized against each other.
|
||||
return keepAlive;
|
||||
}
|
||||
|
||||
@@ -327,14 +347,28 @@ namespace core
|
||||
// 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();
|
||||
// live via dmesg/AddressSanitizer 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.
|
||||
//
|
||||
// inHandler must be re-checked AFTER acquiring `lock`, not just
|
||||
// before try_lock() - checking it first and then separately
|
||||
// acquiring the lock leaves a real TOCTOU gap (this thread reads
|
||||
// inHandler==false, eventReceived() flips it back and forth on
|
||||
// another thread, both end up calling resetSocket()); a first
|
||||
// attempt at this fix only checked before, and a genuine
|
||||
// double-delete of the same session object still reproduced live
|
||||
// under AddressSanitizer despite it. eventReceived() holds this
|
||||
// exact same `lock` across its own inHandler-clear-then-reset
|
||||
// sequence now (see its own comment), so re-checking inHandler
|
||||
// only after try_lock() succeeds makes the two truly mutually
|
||||
// exclusive.
|
||||
if(lock.try_lock()) {
|
||||
if(!inHandler)
|
||||
resetSocket();
|
||||
lock.unlock();
|
||||
}
|
||||
return 1;
|
||||
|
||||
@@ -124,6 +124,23 @@ namespace core {
|
||||
|
||||
protected:
|
||||
|
||||
// Set by Timer's own constructor (see its own doc comment) - a
|
||||
// Timer is always a Socket subobject privately/multiply inherited
|
||||
// alongside a session's own real network Socket (e.g. ZephoraSession
|
||||
// : TCPSession, Timer), never a standalone connection with its own
|
||||
// independent lifetime. eventReceived()'s ordinary keepAlive
|
||||
// contract - returning false means "delete me" (see Thread::run(),
|
||||
// which then does exactly that on this object's own Socket* base) -
|
||||
// is correct for a real connection but catastrophic here: deleting
|
||||
// through a pointer to just the Timer subobject deletes the entire
|
||||
// composite object at the WRONG size (Timer/Socket's own smaller
|
||||
// size, not the true derived object's), a new-delete-type-mismatch
|
||||
// confirmed live via AddressSanitizer under concurrent load. A
|
||||
// Timer's own fd is already correctly torn down as part of its
|
||||
// containing object's own destructor chain; it never needs (and
|
||||
// must never trigger) independent deletion through this path.
|
||||
bool preventSelfDelete = false;
|
||||
|
||||
EPoll &ePoll; // The EPoll control object.
|
||||
|
||||
void setBufferSize(int length);
|
||||
|
||||
+17
-3
@@ -25,8 +25,15 @@ namespace core {
|
||||
std::stringstream out;
|
||||
|
||||
out << "cancel:" << id << std::endl;
|
||||
// subscribers holds plain observer pointers, not ownership (see
|
||||
// TCPSession::pin()'s own doc comment) - a subscriber's session may
|
||||
// have already started tearing down on another thread by the time
|
||||
// this runs. pin() safely tells the two cases apart instead of
|
||||
// blindly dereferencing a pointer that might already be mid-
|
||||
// destruction or fully freed.
|
||||
for (auto subscriber : subscribers) {
|
||||
subscriber->write(out.str());
|
||||
if (auto safe = subscriber->pin())
|
||||
safe->write(out.str());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -65,8 +72,14 @@ namespace core {
|
||||
}
|
||||
|
||||
int Subscription::event(std::stringstream &out) {
|
||||
// See ~Subscription()'s comment above - same pin()-before-use
|
||||
// reasoning applies to every subscriber write here. This is the
|
||||
// broadcast path for every *SUBSCRIBERS-mode topic (chat, item
|
||||
// database updates, etc.) - the exact crash pin() closes here was
|
||||
// confirmed live under concurrent load this session.
|
||||
for (auto subscriber = subscribers.begin(); subscriber < subscribers.end(); ++subscriber)
|
||||
(*subscriber)->write(out.str());
|
||||
if (auto safe = (*subscriber)->pin())
|
||||
safe->write(out.str());
|
||||
return 1;
|
||||
}
|
||||
|
||||
@@ -85,7 +98,8 @@ namespace core {
|
||||
void Subscription::sendToAll(std::stringstream &data, TCPSession &sender) {
|
||||
for (auto session : subscribers)
|
||||
if (session != &sender)
|
||||
session->write(data.str());
|
||||
if (auto safe = session->pin())
|
||||
safe->write(data.str());
|
||||
data.str("");
|
||||
}
|
||||
|
||||
|
||||
+15
-6
@@ -58,10 +58,15 @@ namespace core
|
||||
{
|
||||
for (auto subscribed = subscription->subscribers.begin(); subscribed < subscription->subscribers.end(); subscribed++)
|
||||
{
|
||||
if ((*subscribed)->alias == session.alias)
|
||||
{
|
||||
subscription->subscribers.erase(subscribed++);
|
||||
}
|
||||
// pin() before reading .alias - a raw subscriber
|
||||
// pointer here could already be mid-destruction or
|
||||
// fully freed on another thread (see
|
||||
// TCPSession::pin()'s own doc comment).
|
||||
if (auto safe = (*subscribed)->pin())
|
||||
if (safe->alias == session.alias)
|
||||
{
|
||||
subscription->subscribers.erase(subscribed++);
|
||||
}
|
||||
subscription->subscriptionOwner(session.alias, "*NEXT");
|
||||
}
|
||||
}
|
||||
@@ -150,8 +155,12 @@ namespace core
|
||||
// fix for "find a session by the player name they typed"
|
||||
// isn't possible at this layer without a larger change; this
|
||||
// stops the crash without pretending to fix that gap.
|
||||
TCPSession *tempSession = session.server->getSessionByAlias(&invitee);
|
||||
if (tempSession == NULL)
|
||||
// getSessionByAlias() now returns an already-pin()'d
|
||||
// shared_ptr - see its own doc comment - so tempSession itself
|
||||
// guarantees the session stays alive for this write() call
|
||||
// below, no separate pin() needed.
|
||||
auto tempSession = session.server->getSessionByAlias(&invitee);
|
||||
if (!tempSession)
|
||||
{
|
||||
coreutils::Log(coreutils::LOG_DEBUG_1) << "invite: could not find session for " << invitee;
|
||||
return 0;
|
||||
|
||||
+51
-14
@@ -32,6 +32,14 @@ namespace core {
|
||||
|
||||
if (listen(getDescriptor(), 20) < 0)
|
||||
throw coreutils::Exception("Error on listen to socket");
|
||||
|
||||
// See isKnownListeningSocket()'s own doc comment - registers this
|
||||
// listening socket as safe for Thread::run() to dispatch to
|
||||
// directly, distinct from the shared_ptr-owned TCPSession registry.
|
||||
{
|
||||
std::lock_guard<std::mutex> guard(listeningSocketsMutex());
|
||||
listeningSockets().insert(this);
|
||||
}
|
||||
}
|
||||
|
||||
TCPServer::~TCPServer() {
|
||||
@@ -43,8 +51,20 @@ namespace core {
|
||||
coreutils::Log(coreutils::LOG_DEBUG_2) << "Connection request is being received on socket " << getDescriptor() << ".";
|
||||
lock.lock();
|
||||
TCPSession *session = accept();
|
||||
if (session)
|
||||
sessions.push_back(session);
|
||||
// The shared_ptr constructed here IS the session's real, strong
|
||||
// ownership from now on (see sessions' own doc comment). A second
|
||||
// reference to the SAME shared_ptr (never a second, independent
|
||||
// std::shared_ptr<TCPSession>(session) call on the same raw
|
||||
// pointer - that would create a second, competing control block
|
||||
// and a guaranteed double-free) also goes into the tryGet()
|
||||
// registry, which is what Thread::run()'s epoll dispatch actually
|
||||
// uses to safely reach this object (see TCPSession::tryGet()'s
|
||||
// own comment on why).
|
||||
if (session) {
|
||||
auto sessionPtr = std::shared_ptr<TCPSession>(session);
|
||||
sessions.push_back(sessionPtr);
|
||||
TCPSession::registerLive(session, sessionPtr);
|
||||
}
|
||||
// Must run ON THE SESSION, not `this` (the listening server) - the
|
||||
// unqualified registerSocket()/acceptSocket() calls this replaced
|
||||
// built the SSL* handshake state on the SERVER's own TLS-base
|
||||
@@ -97,15 +117,32 @@ namespace core {
|
||||
}
|
||||
|
||||
void TCPServer::removeFromSessionList(TCPSession *session) {
|
||||
std::vector<TCPSession *>::iterator cursor;
|
||||
lock.lock();
|
||||
for (cursor = sessions.begin(); cursor < sessions.end(); ++cursor)
|
||||
if (*cursor == session)
|
||||
for (auto cursor = sessions.begin(); cursor < sessions.end(); ++cursor)
|
||||
if (cursor->get() == session)
|
||||
{
|
||||
// Drops the one strong-owning reference. If nothing else is
|
||||
// currently pin()'d to this session, the object is actually
|
||||
// destroyed right here, synchronously - same timing as
|
||||
// before this change for the common (no-contention) case.
|
||||
// If some other thread IS mid-call via its own pin()'d
|
||||
// shared_ptr right now, destruction is safely deferred until
|
||||
// that reference is released too, instead of the crash
|
||||
// (double-delete / use-after-free, confirmed live via
|
||||
// AddressSanitizer under concurrent load) that raw-pointer
|
||||
// ownership produced here before.
|
||||
sessions.erase(cursor);
|
||||
break;
|
||||
}
|
||||
lock.unlock();
|
||||
// Removes this pointer from the tryGet() lookup registry too (see
|
||||
// TCPSession::tryGet()'s own comment). Safe to do after releasing
|
||||
// `lock` and regardless of whether the erase() above was actually
|
||||
// the last reference: tryGet() correctly reports "gone" the moment
|
||||
// a session's real refcount hits zero, via the weak_ptr's own
|
||||
// expiration - this call is just tidying the registry map itself,
|
||||
// not something later correctness depends on.
|
||||
TCPSession::unregisterLive(session);
|
||||
}
|
||||
|
||||
void TCPServer::sessionErrorHandler(std::string errorString, std::stringstream &out) {
|
||||
@@ -139,7 +176,7 @@ namespace core {
|
||||
// subscriber list this session's earlier fix addressed.
|
||||
std::lock_guard<std::mutex> guard(lock);
|
||||
int sequence = 0;
|
||||
for (auto *sessionx : sessions)
|
||||
for (auto &sessionx : sessions)
|
||||
{
|
||||
session.out << "|" << ++sequence;
|
||||
sessionx->output(session.out);
|
||||
@@ -150,35 +187,35 @@ namespace core {
|
||||
|
||||
void TCPServer::sendToAll(std::stringstream &data) {
|
||||
std::lock_guard<std::mutex> guard(lock);
|
||||
for (auto session : sessions)
|
||||
for (auto &session : sessions)
|
||||
session->write(data.str());
|
||||
data.str("");
|
||||
}
|
||||
|
||||
void TCPServer::sendToAll(std::stringstream &data, TCPSession &sender) {
|
||||
std::lock_guard<std::mutex> guard(lock);
|
||||
for (auto session : sessions)
|
||||
if (session != &sender)
|
||||
for (auto &session : sessions)
|
||||
if (session.get() != &sender)
|
||||
session->write(data.str());
|
||||
data.str("");
|
||||
}
|
||||
|
||||
void TCPServer::sendToAll(std::stringstream &data, TCPSession &sender, SessionFilter filter) {
|
||||
std::lock_guard<std::mutex> guard(lock);
|
||||
for (auto session : sessions)
|
||||
for (auto &session : sessions)
|
||||
if (filter.test(*session))
|
||||
if (session != &sender)
|
||||
if (session.get() != &sender)
|
||||
session->write(data.str());
|
||||
data.str("");
|
||||
}
|
||||
|
||||
TCPSession *TCPServer::getSessionByAlias(void *alias) {
|
||||
std::shared_ptr<TCPSession> TCPServer::getSessionByAlias(void *alias) {
|
||||
coreutils::Log(coreutils::LOG_DEBUG_1) << alias;
|
||||
std::lock_guard<std::mutex> guard(lock);
|
||||
for (auto session : sessions)
|
||||
for (auto &session : sessions)
|
||||
if (session->alias == alias)
|
||||
return session;
|
||||
return NULL;
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
}
|
||||
+60
-4
@@ -9,6 +9,8 @@
|
||||
# include "TCPSession.h"
|
||||
# include "TCPSocket.h"
|
||||
# include "TLSInfo.h"
|
||||
# include <mutex>
|
||||
# include <set>
|
||||
|
||||
namespace core {
|
||||
|
||||
@@ -60,8 +62,20 @@ namespace core {
|
||||
///
|
||||
/// The list of sessions that are currently open and being maintained by this object.
|
||||
///
|
||||
/// This is the one real, strong-owning reference per live session -
|
||||
/// see TCPSession::pin()'s own doc comment for the full reasoning.
|
||||
/// Every other place that keeps a TCPSession* around (Subscription::
|
||||
/// subscribers, a Zephora-specific sessionsByPlayerName map, etc.)
|
||||
/// is just a non-owning observer and must pin() before touching the
|
||||
/// object. A session stops existing, for real, the instant its
|
||||
/// entry here is erased and every other outstanding pin() (if any
|
||||
/// thread happened to be mid-call) has released its own reference -
|
||||
/// removeFromSessionList() is the only place that ever removes an
|
||||
/// entry, and Thread::run() calls it instead of ever calling
|
||||
/// `delete` on a TCPSession directly (see its own comment).
|
||||
///
|
||||
|
||||
std::vector<TCPSession *> sessions;
|
||||
std::vector<std::shared_ptr<TCPSession>> sessions;
|
||||
|
||||
///
|
||||
/// The commands object is a CommandList and is used to store Command objects to be
|
||||
@@ -119,10 +133,13 @@ namespace core {
|
||||
|
||||
///
|
||||
/// Use the getSessionByAlias to retrieve a session pointer by the value
|
||||
/// of the alias pointer.
|
||||
/// of the alias pointer. Returns an already-pin()'d shared_ptr (null
|
||||
/// if not found) rather than a raw pointer, so callers can't
|
||||
/// accidentally hold onto - or use - a reference that isn't actually
|
||||
/// keeping the session alive.
|
||||
///
|
||||
|
||||
TCPSession *getSessionByAlias(void *alias);
|
||||
std::shared_ptr<TCPSession> getSessionByAlias(void *alias);
|
||||
|
||||
protected:
|
||||
///
|
||||
@@ -159,9 +176,48 @@ namespace core {
|
||||
|
||||
int processCommand(coreutils::ZString &request, TCPSession &session) override;
|
||||
|
||||
protected:
|
||||
// Was private - ConsoleServer::logSend() (and any other subclass
|
||||
// that iterates `sessions`, itself public further up) needs to be
|
||||
// able to synchronize against the exact same mutex
|
||||
// connectionRequest()/removeFromSessionList() already use, rather
|
||||
// than reading a shared_ptr vector with no locking of its own.
|
||||
std::mutex lock;
|
||||
|
||||
public:
|
||||
///
|
||||
/// A listening TCPServer/ConsoleServer (as opposed to an accepted
|
||||
/// TCPSession) is constructed once, directly by main(), and lives
|
||||
/// for the whole process - never through the shared_ptr-based
|
||||
/// ownership TCPSession::registerLive()/pin() provide (see their
|
||||
/// own comments), since nothing ever calls `new` on one of these.
|
||||
/// Thread::run()'s epoll dispatch still needs a SAFE way to tell
|
||||
/// "this raw pointer is one of those known-immortal listening
|
||||
/// sockets, fine to call methods on directly" apart from "this is
|
||||
/// an unrecognized/stale pointer that must not be touched at all"
|
||||
/// (see TCPSession::tryGet()'s own comment on why dynamic_cast
|
||||
/// can't make that distinction safely). Registering `this` once,
|
||||
/// in the constructor, into this tiny set is that signal - it's
|
||||
/// checked by address only, never dereferenced for the check
|
||||
/// itself.
|
||||
///
|
||||
|
||||
static bool isKnownListeningSocket(void *ptr) {
|
||||
std::lock_guard<std::mutex> guard(listeningSocketsMutex());
|
||||
return listeningSockets().count(ptr) > 0;
|
||||
}
|
||||
|
||||
private:
|
||||
TCPSession *accept();
|
||||
std::mutex lock;
|
||||
|
||||
static std::mutex &listeningSocketsMutex() {
|
||||
static std::mutex m;
|
||||
return m;
|
||||
}
|
||||
static std::set<void *> &listeningSockets() {
|
||||
static std::set<void *> s;
|
||||
return s;
|
||||
}
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
+117
-1
@@ -3,6 +3,9 @@
|
||||
|
||||
# include "SessionFilter.h"
|
||||
# include "TCPSocket.h"
|
||||
# include <map>
|
||||
# include <memory>
|
||||
# include <mutex>
|
||||
# include <sstream>
|
||||
|
||||
namespace core {
|
||||
@@ -22,7 +25,120 @@ namespace core {
|
||||
///
|
||||
///
|
||||
|
||||
class TCPSession : public TCPSocket {
|
||||
class TCPSession : public TCPSocket, public std::enable_shared_from_this<TCPSession> {
|
||||
|
||||
public:
|
||||
|
||||
///
|
||||
/// pin() safely upgrades a raw TCPSession* (as stored in
|
||||
/// Subscription::subscribers, ZephoraServer's own
|
||||
/// sessionsByPlayerName, GroupManager's membership lists, etc.)
|
||||
/// into a real, ownership-extending shared_ptr before it's used.
|
||||
///
|
||||
/// The core problem this solves: every crash found and root-caused
|
||||
/// this session (SubscriptionManager.cpp/TCPServer.cpp/Socket.cpp's
|
||||
/// own fixes) was the same shape - one thread calling a method
|
||||
/// (write(), a destructor) on a TCPSession* while another thread
|
||||
/// concurrently deleted that same object, because nothing actually
|
||||
/// guaranteed the object stayed alive for the duration of the
|
||||
/// call. Mutex-based fixes at each individual call site kept
|
||||
/// narrowing the window without fully closing it, and a mutex
|
||||
/// around every write() call also isn't what the fifo-based
|
||||
/// design (see Socket::write()) was going for in the first place -
|
||||
/// no thread should have to block just to hand off a message.
|
||||
///
|
||||
/// TCPServer::sessions now holds the one real, strong-owning
|
||||
/// shared_ptr<TCPSession> per live session (see its own comment).
|
||||
/// Every OTHER place that stores a TCPSession* for later use is a
|
||||
/// plain observer, not an owner - it must call pin() and check for
|
||||
/// null immediately before touching the object, rather than
|
||||
/// dereferencing the raw pointer directly. pin() returns nullptr
|
||||
/// (instead of throwing, unlike raw shared_from_this()) once the
|
||||
/// object's last real owner has let go and it's on its way out -
|
||||
/// at that point there is nothing left to safely do with it, and
|
||||
/// the caller should just skip it as if it were never there,
|
||||
/// exactly like a stale/already-unsubscribed entry.
|
||||
///
|
||||
/// This adds no blocking anywhere: shared_ptr's own refcounting is
|
||||
/// a couple of atomic increments, not a mutex - the actual
|
||||
/// deletion still only ever happens once, whenever the last
|
||||
/// outstanding shared_ptr (held by TCPServer::sessions, or
|
||||
/// briefly by whichever thread(s) are mid-call via their own
|
||||
/// pin()'d reference) goes out of scope.
|
||||
///
|
||||
|
||||
std::shared_ptr<TCPSession> pin() {
|
||||
try { return shared_from_this(); }
|
||||
catch (const std::bad_weak_ptr &) { return nullptr; }
|
||||
}
|
||||
|
||||
///
|
||||
/// A second, distinct safety problem pin()/shared_from_this()
|
||||
/// alone cannot solve: Thread::run()'s epoll dispatch loop starts
|
||||
/// from a raw `void*` (epoll_event.data.ptr) whose TYPE isn't known
|
||||
/// yet - the original code cast it straight to `Socket*` and
|
||||
/// called virtual methods on it; an early version of this fix used
|
||||
/// dynamic_cast<TCPSession*> to identify session objects before
|
||||
/// calling pin(). Both of those REQUIRE reading the pointed-to
|
||||
/// object's own memory (its vtable, at minimum) to work at all -
|
||||
/// confirmed live via AddressSanitizer that dynamic_cast on a
|
||||
/// stale pointer (one epoll handed out for a TCPSession that
|
||||
/// something else had already fully destroyed and freed on another
|
||||
/// thread) segfaults outright, before pin()'s own safety logic
|
||||
/// ever gets a chance to run.
|
||||
///
|
||||
/// tryGet() sidesteps this entirely: it looks the raw pointer VALUE
|
||||
/// up as a key in a plain map, and only ever touches a weak_ptr's
|
||||
/// own control block (a separate heap allocation that outlives the
|
||||
/// TCPSession object itself for as long as any weak_ptr or
|
||||
/// shared_ptr referencing it exists) - never the TCPSession's own
|
||||
/// memory - to decide whether it's still valid. That makes it safe
|
||||
/// to call on ANY raw pointer value, including ones that turn out
|
||||
/// to be completely unrelated to this registry (never a
|
||||
/// TCPSession, e.g. a listening TCPServer/ConsoleServer - see
|
||||
/// TCPServer::isKnownListeningSocket() for how Thread::run()
|
||||
/// distinguishes that case) or ones pointing at freed memory.
|
||||
///
|
||||
/// registerLive()/unregisterLive() are called exactly once each,
|
||||
/// by TCPServer::connectionRequest() and removeFromSessionList()
|
||||
/// respectively - see their own comments for why those are the
|
||||
/// right, and only, call sites.
|
||||
///
|
||||
|
||||
static void registerLive(TCPSession *raw, std::shared_ptr<TCPSession> owner) {
|
||||
std::lock_guard<std::mutex> guard(registryMutex());
|
||||
registry()[raw] = owner;
|
||||
}
|
||||
|
||||
static void unregisterLive(TCPSession *raw) {
|
||||
std::lock_guard<std::mutex> guard(registryMutex());
|
||||
registry().erase(raw);
|
||||
}
|
||||
|
||||
static std::shared_ptr<TCPSession> tryGet(void *raw) {
|
||||
std::lock_guard<std::mutex> guard(registryMutex());
|
||||
auto it = registry().find((TCPSession *)raw);
|
||||
if (it == registry().end())
|
||||
return nullptr;
|
||||
return it->second.lock();
|
||||
}
|
||||
|
||||
private:
|
||||
// Function-local statics (constructed on first use, in whichever
|
||||
// translation unit/thread reaches them first, thread-safely per
|
||||
// C++11's own guarantee for static local initialization) instead
|
||||
// of ordinary static data members - avoids needing a .cpp
|
||||
// definition for these two and any static-initialization-order
|
||||
// concerns with other translation units, since TCPSession.h is
|
||||
// included widely.
|
||||
static std::mutex ®istryMutex() {
|
||||
static std::mutex m;
|
||||
return m;
|
||||
}
|
||||
static std::map<TCPSession *, std::weak_ptr<TCPSession>> ®istry() {
|
||||
static std::map<TCPSession *, std::weak_ptr<TCPSession>> r;
|
||||
return r;
|
||||
}
|
||||
|
||||
public:
|
||||
|
||||
|
||||
+60
-7
@@ -1,5 +1,6 @@
|
||||
#include "Thread.h"
|
||||
#include "EPoll.h"
|
||||
#include "TCPServer.h"
|
||||
#include <sys/epoll.h>
|
||||
|
||||
namespace core {
|
||||
@@ -100,14 +101,66 @@ namespace core {
|
||||
} else if(rc > 0) {
|
||||
for(int ix = 0; ix < rc; ++ix) {
|
||||
++count;
|
||||
if(((Socket *)events[ix].data.ptr)->eventReceived(events[ix], ++ePoll.eventId)) {
|
||||
// coreutils::Log(coreutils::LOG_DEBUG_4) << "return true";
|
||||
// ePoll.resetSocket((Socket *)events[ix].data.ptr);
|
||||
// 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 {
|
||||
((Socket *)events[ix].data.ptr)->shutDown = true;
|
||||
// coreutils::Log(coreutils::LOG_DEBUG_4) << "return false";
|
||||
delete (Socket *)events[ix].data.ptr;
|
||||
}
|
||||
// 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 << ")";
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,6 +7,12 @@ namespace core
|
||||
|
||||
Timer::Timer(EPoll &ePoll, double delay = 0.0f) : Socket(ePoll, "Timer")
|
||||
{
|
||||
// See preventSelfDelete's own doc comment on Socket - a Timer is
|
||||
// always a subobject of some larger composite (e.g. ZephoraSession
|
||||
// : TCPSession, Timer), never a standalone connection, so its own
|
||||
// eventReceived() must never be allowed to signal "delete me" to
|
||||
// Thread::run()'s generic dispatch.
|
||||
preventSelfDelete = true;
|
||||
setDescriptor(timerfd_create(CLOCK_REALTIME, 0));
|
||||
setTimer(delay);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user