Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions indra/llmessage/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -228,6 +228,7 @@ if (LL_TESTS)
LL_ADD_INTEGRATION_TEST(llsdmessagebuilder "" "${test_libs}")
LL_ADD_INTEGRATION_TEST(llsdmessagereader "" "${test_libs}")
LL_ADD_INTEGRATION_TEST(llmessageconfig "" "${test_libs}")
LL_ADD_INTEGRATION_TEST(llmessagetemplate "" "${test_libs}")
LL_ADD_INTEGRATION_TEST(llmessagetemplateparser "" "${test_libs}")
LL_ADD_INTEGRATION_TEST(llpartdata "" "${test_libs}")
LL_ADD_INTEGRATION_TEST(llservicebuilder "" "${test_libs}")
Expand Down
149 changes: 110 additions & 39 deletions indra/llmessage/llcircuit.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -421,6 +421,8 @@ S32 LLCircuitData::resendUnackedPackets(const F64Seconds now)

LLCircuit::LLCircuit(const F32Seconds circuit_heartbeat_interval, const F32Seconds circuit_timeout)
: mLastCircuit(NULL),
mCircuitsUnusedCount(0),
mDeferredDeletion(false),
mHeartbeatInterval(circuit_heartbeat_interval),
mHeartbeatTimeout(circuit_timeout)
{}
Expand All @@ -433,6 +435,13 @@ LLCircuit::~LLCircuit()
llcompose1(
DeletePointerFunctor<LLCircuitData>(),
llselect2nd<circuit_data_map::value_type>()));

// The receiver thread is long gone by now, nothing can be using these.
for (auto& dead : mGraveyard)
{
delete dead.first;
}
mGraveyard.clear();
}

LLCircuitData *LLCircuit::addCircuitData(const LLHost &host, TPACKETID in_id)
Expand All @@ -452,36 +461,89 @@ LLCircuitData *LLCircuit::addCircuitData(const LLHost &host, TPACKETID in_id)
void LLCircuit::removeCircuitData(const LLHost &host)
{
LL_INFOS() << "LLCircuit::removeCircuitData for " << host << LL_ENDL;
std::lock_guard<std::mutex> lock(mCircuitMutex);
mLastCircuit = NULL;
circuit_data_map::iterator it = mCircuitData.find(host);
if(it != mCircuitData.end())
LLCircuitData *cdp = NULL;
{
LLCircuitData *cdp = it->second;
mCircuitData.erase(it);
std::lock_guard<std::mutex> lock(mCircuitMutex);
circuit_data_map::iterator it = mCircuitData.find(host);
if(it != mCircuitData.end())
{
cdp = it->second;
mCircuitData.erase(it);

LLCircuit::ping_set_t::iterator psit = mPingSet.find(cdp);
if (psit != mPingSet.end())
{
mPingSet.erase(psit);
}
else
{
LL_WARNS() << "Couldn't find entry for next ping in ping set!" << LL_ENDL;
}

// Clean up from optimization maps
mUnackedCircuitMap.erase(host);
mSendAckMap.erase(host);

if (mDeferredDeletion)
{
// The UDP receiver thread may be decoding a packet on this
// circuit right now, and it uses plain pointers. The circuit
// is no longer reachable from any of the maps, so nobody can
// pick it up anymore: bury it and let cleanupGraveyard()
// delete it once the receiver thread is done with it.
mGraveyard.emplace_back(cdp, mCircuitsUnusedCount.load());
cdp = NULL;
}
}

// This also has to happen AFTER we nuke the circuit, because various
// callbacks for the circuit may result in messages being sent to
// this circuit, and the setting of mLastCircuit. We don't check
// if the host matches, but we don't really care because mLastCircuit
// is an optimization, and this happens VERY rarely.
mLastCircuit = NULL;
}

// Deleting a circuit fires reliable packet callbacks, so it has to happen
// with mCircuitMutex released.
delete cdp;

cleanupGraveyard();
}

LLCircuit::ping_set_t::iterator psit = mPingSet.find(cdp);
if (psit != mPingSet.end())
void LLCircuit::cleanupGraveyard()
{
// Deleted with mCircuitMutex released, see removeCircuitData().
std::vector<LLCircuitData*> doomed;
{
std::lock_guard<std::mutex> lock(mCircuitMutex);
if (mGraveyard.empty())
{
mPingSet.erase(psit);
return;
}
else
// The receiver thread bumps mCircuitsUnusedCount between packets, so
// a different value means it has been between packets since the entry
// was buried and can't be holding a pointer to it.
bool thread_gone = !mDeferredDeletion;
U32 unused_count = mCircuitsUnusedCount.load();
for (auto it = mGraveyard.begin(); it != mGraveyard.end(); )
{
LL_WARNS() << "Couldn't find entry for next ping in ping set!" << LL_ENDL;
if (thread_gone || unused_count != it->second)
{
doomed.push_back(it->first);
it = mGraveyard.erase(it);
}
else
{
++it;
}
}

// Clean up from optimization maps
mUnackedCircuitMap.erase(host);
mSendAckMap.erase(host);
delete cdp;
}

// This also has to happen AFTER we nuke the circuit, because various
// callbacks for the circuit may result in messages being sent to
// this circuit, and the setting of mLastCircuit. We don't check
// if the host matches, but we don't really care because mLastCircuit
// is an optimization, and this happens VERY rarely.
mLastCircuit = NULL;
for (LLCircuitData* dead : doomed)
{
delete dead;
}
}

void LLCircuitData::setAlive(bool b_alive)
Expand Down Expand Up @@ -871,6 +933,9 @@ void LLCircuit::updateWatchDogTimers(LLMessageSystem *msgsys)
}
}
}

// Delete circuits that were buried while the receiver thread was busy.
cleanupGraveyard();
}


Expand Down Expand Up @@ -1107,16 +1172,16 @@ bool LLCircuitData::checkCircuitTimeout()
// correctly place the packet in the correct list to be acked later.
bool LLCircuitData::collectRAck(TPACKETID packet_num)
{
// Global lock order: LLCircuit::mCircuitMutex before mDataMutex.
std::lock_guard<std::mutex> circuit_lock(gMessageSystem->mCircuitInfo.mCircuitMutex);
std::lock_guard<std::mutex> data_lock(mDataMutex);
if (mAcks.empty())
{
std::lock_guard<std::mutex> circuit_lock(gMessageSystem->mCircuitInfo.mCircuitMutex);
// First extra ack, we need to add ourselves to the list of circuits that need to send acks
gMessageSystem->mCircuitInfo.mSendAckMap[mHost] = this;
}

mAcks.push_back(packet_num);
std::lock_guard<std::mutex> circuit_lock(gMessageSystem->mCircuitInfo.mCircuitMutex);
if (mAckCreationTime == 0)
{
mAckCreationTime = getAgeInSeconds();
Expand All @@ -1129,38 +1194,44 @@ bool LLCircuitData::collectRAck(TPACKETID packet_num)
void LLCircuit::sendAcks(F32 collect_time)
{
collect_time = llclamp(collect_time, 0.f, LL_COLLECT_ACK_TIME_MAX);
std::vector<LLCircuitData*> circuits_to_flush; // Collect while holding mCircuitMutex
// Collect while holding mCircuitMutex, send with no lock held.
std::vector<std::pair<LLCircuitData*, std::vector<TPACKETID> > > circuits_to_flush;
{
std::lock_guard<std::mutex> circuit_lock(mCircuitMutex);
circuit_data_map::iterator it = mSendAckMap.begin();
while (it != mSendAckMap.end())
{
circuit_data_map::iterator cur_it = it++;
LLCircuitData* cd = (*cur_it).second;
std::lock_guard<std::mutex> data_lock(cd->mDataMutex);
S32 count = (S32)cd->mAcks.size();
F32 age = cd->getAgeInSeconds() - cd->mAckCreationTime;
if (age > collect_time || count == 0)
std::vector<TPACKETID> acks;
{
if (count > 0)
std::lock_guard<std::mutex> data_lock(cd->mDataMutex);
S32 count = (S32)cd->mAcks.size();
F32 age = cd->getAgeInSeconds() - cd->mAckCreationTime;
if (age <= collect_time && count != 0)
{
circuits_to_flush.push_back(cd);
continue;
}
// Take the acks while both locks are held, so that a circuit
// removed from mSendAckMap never keeps pending acks behind:
// any ack collected after this point sees an empty mAcks and
// re-registers the circuit.
acks.swap(cd->mAcks);
cd->mAckCreationTime = 0.f;
// remove data map
mSendAckMap.erase(cur_it);
}
if (!acks.empty())
{
circuits_to_flush.emplace_back(cd, std::move(acks));
}
}
}

for (LLCircuitData* cd : circuits_to_flush)
for (const auto& flush : circuits_to_flush)
{
std::vector<TPACKETID> acks;
{
std::lock_guard<std::mutex> data_lock(cd->mDataMutex);
acks.swap(cd->mAcks);
cd->mAckCreationTime = 0.f;
}

LLCircuitData* cd = flush.first;
const std::vector<TPACKETID>& acks = flush.second;
S32 count = (S32)acks.size();

// send the packet acks
Expand Down
42 changes: 41 additions & 1 deletion indra/llmessage/llcircuit.h
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,9 @@
#ifndef LL_LLCIRCUIT_H
#define LL_LLCIRCUIT_H

#include <atomic>
#include <map>
#include <mutex>
#include <vector>

#include "llerror.h"
Expand Down Expand Up @@ -73,7 +75,16 @@ class LLSD;
// Classes
//


// LOCK ORDER
//
// Two mutexes guard circuit state: LLCircuit::mCircuitMutex (the circuit maps)
// and LLCircuitData::mDataMutex (per-circuit data). Whenever both are needed,
// they must be acquired in this order:
//
// LLCircuit::mCircuitMutex -> LLCircuitData::mDataMutex
//
// Never acquire mCircuitMutex while holding a circuit's mDataMutex, and never
// run callbacks or send messages while holding either of them.
class LLCircuitData
{
public:
Expand Down Expand Up @@ -188,6 +199,8 @@ class LLCircuitData
// Call this method when a reliable message comes in - this will
// correctly place the packet in the correct list to be acked
// later. RAack = requested ack
// Takes LLCircuit::mCircuitMutex and then mDataMutex, so callers must not
// hold either of them (see the lock order note above).
bool collectRAck(TPACKETID packet_num);


Expand Down Expand Up @@ -305,13 +318,30 @@ class LLCircuit
~LLCircuit();

// ACCESSORS
// The returned circuit stays valid for as long as the caller keeps the
// main thread from running removeCircuitData(); on the UDP receiver
// thread it stays valid until the next markCircuitsUnused().
LLCircuitData* findCircuit(const LLHost& host) const;
bool isCircuitAlive(const LLHost& host) const;

// MANIPULATORS
LLCircuitData *addCircuitData(const LLHost &host, TPACKETID in_id);
// Main thread only. While the UDP receiver thread is running, the circuit
// is parked in the graveyard instead of being deleted right away, since
// that thread may still be decoding a packet on it.
void removeCircuitData(const LLHost &host);

// Called from the main thread, deletes the circuits in the graveyard that
// the UDP receiver thread is guaranteed to be done with.
void cleanupGraveyard();
// Called from the UDP receiver thread at a point where it does not hold
// any LLCircuitData pointer, so that circuits removed before this point
// can be deleted by the main thread.
void markCircuitsUnused() { mCircuitsUnusedCount.fetch_add(1); }
// Enables the graveyard. Called from the main thread when the UDP
// receiver thread is started and after it has been joined.
void setDeferredDeletion(bool defer) { mDeferredDeletion = defer; }

void updateWatchDogTimers(LLMessageSystem *msgsys);
void resendUnackedPackets(S32& unacked_list_length, S32& unacked_list_size);

Expand Down Expand Up @@ -346,6 +376,16 @@ class LLCircuit
// set in otherwise const methods, so it is declared mutable.
mutable LLCircuitData* mLastCircuit;

// Circuits that were removed from the maps while the UDP receiver thread
// might still have been using them, with the value mCircuitsUnusedCount
// had at the time of removal. Guarded by mCircuitMutex.
std::vector<std::pair<LLCircuitData*, U32> > mGraveyard;
// Bumped by the receiver thread whenever it is between packets and hence
// holds no circuit pointer. A graveyard entry may be deleted as soon as
// this has moved past the value stored with it.
std::atomic<U32> mCircuitsUnusedCount;
std::atomic<bool> mDeferredDeletion;

private:
const F32Seconds mHeartbeatInterval;
const F32Seconds mHeartbeatTimeout;
Expand Down
Loading
Loading