diff options
author | Ansariel <ansariel.hiller@phoenixviewer.com> | 2024-05-22 21:25:21 +0200 |
---|---|---|
committer | Andrey Lihatskiy <alihatskiy@productengine.com> | 2024-05-22 22:40:26 +0300 |
commit | e2e37cced861b98de8c1a7c9c0d3a50d2d90e433 (patch) | |
tree | 1bb897489ce524986f6196201c10ac0d8861aa5f /indra/llmessage/llcircuit.cpp | |
parent | 069ea06848f766466f1a281144c82a0f2bd79f3a (diff) |
Fix line endlings
Diffstat (limited to 'indra/llmessage/llcircuit.cpp')
-rw-r--r-- | indra/llmessage/llcircuit.cpp | 2838 |
1 files changed, 1419 insertions, 1419 deletions
diff --git a/indra/llmessage/llcircuit.cpp b/indra/llmessage/llcircuit.cpp index bd9b3553fe..fa206d9282 100644 --- a/indra/llmessage/llcircuit.cpp +++ b/indra/llmessage/llcircuit.cpp @@ -1,1419 +1,1419 @@ -/**
- * @file llcircuit.cpp
- * @brief Class to track UDP endpoints for the message system.
- *
- * $LicenseInfo:firstyear=2002&license=viewerlgpl$
- * Second Life Viewer Source Code
- * Copyright (C) 2010, Linden Research, Inc.
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation;
- * version 2.1 of the License only.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library; if not, write to the Free Software
- * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
- *
- * Linden Research, Inc., 945 Battery Street, San Francisco, CA 94111 USA
- * $/LicenseInfo$
- */
-
-#include "linden_common.h"
-
-#if LL_WINDOWS
-
-#include <process.h>
-
-#else
-
-#if LL_LINUX
-#include <dlfcn.h> // RTLD_LAZY
-#endif
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <netinet/in.h>
-
-#endif
-
-
-#if !defined(USE_CIRCUIT_LIST)
-#include <algorithm>
-#endif
-#include <sstream>
-#include <iterator>
-#include <stack>
-
-#include "llcircuit.h"
-
-#include "message.h"
-#include "llrand.h"
-#include "llstl.h"
-#include "lltransfermanager.h"
-#include "llmodularmath.h"
-
-const S32 PING_START_BLOCK = 3; // How many pings behind we have to be to consider ourself blocked.
-const S32 PING_RELEASE_BLOCK = 2; // How many pings behind we have to be to consider ourself unblocked.
-
-const F32Seconds TARGET_PERIOD_LENGTH(5.f);
-const F32Seconds LL_DUPLICATE_SUPPRESSION_TIMEOUT(60.f); //this can be long, as time-based cleanup is
- // only done when wrapping packetids, now...
-
-LLCircuitData::LLCircuitData(const LLHost &host, TPACKETID in_id,
- const F32Seconds circuit_heartbeat_interval, const F32Seconds circuit_timeout)
-: mHost (host),
- mWrapID(0),
- mPacketsOutID(0),
- mPacketsInID(in_id),
- mHighestPacketID(in_id),
- mTimeoutCallback(NULL),
- mTimeoutUserData(NULL),
- mTrusted(false),
- mbAllowTimeout(true),
- mbAlive(true),
- mBlocked(false),
- mPingTime(0.0),
- mLastPingSendTime(0.0),
- mLastPingReceivedTime(0.0),
- mNextPingSendTime(0.0),
- mPingsInTransit(0),
- mLastPingID(0),
- mPingDelay(INITIAL_PING_VALUE_MSEC),
- mPingDelayAveraged(INITIAL_PING_VALUE_MSEC),
- mUnackedPacketCount(0),
- mUnackedPacketBytes(0),
- mLastPacketInTime(0.0),
- mLocalEndPointID(),
- mPacketsOut(0),
- mPacketsIn(0),
- mPacketsLost(0),
- mBytesIn(0),
- mBytesOut(0),
- mLastPeriodLength(-1.f),
- mBytesInLastPeriod(0),
- mBytesOutLastPeriod(0),
- mBytesInThisPeriod(0),
- mBytesOutThisPeriod(0),
- mPeakBPSIn(0.f),
- mPeakBPSOut(0.f),
- mPeriodTime(0.0),
- mExistenceTimer(),
- mAckCreationTime(0.f),
- mCurrentResendCount(0),
- mLastPacketGap(0),
- mHeartbeatInterval(circuit_heartbeat_interval),
- mHeartbeatTimeout(circuit_timeout)
-{
- // Need to guarantee that this time is up to date, we may be creating a circuit even though we haven't been
- // running a message system loop.
- F64Seconds mt_sec = LLMessageSystem::getMessageTimeSeconds(true);
- F32 distribution_offset = ll_frand();
-
- mPingTime = mt_sec;
- mLastPingSendTime = mt_sec + mHeartbeatInterval * distribution_offset;
- mLastPingReceivedTime = mt_sec;
- mNextPingSendTime = mLastPingSendTime + 0.95*mHeartbeatInterval + F32Seconds(ll_frand(0.1f*mHeartbeatInterval.value()));
- mPeriodTime = mt_sec;
-
- mLocalEndPointID.generate();
-}
-
-
-LLCircuitData::~LLCircuitData()
-{
- LLReliablePacket *packetp = NULL;
-
- // Clean up all pending transfers.
- gTransferManager.cleanupConnection(mHost);
-
- // remove all pending reliable messages on this circuit
- std::vector<TPACKETID> doomed;
- reliable_iter iter;
- reliable_iter end = mUnackedPackets.end();
- for(iter = mUnackedPackets.begin(); iter != end; ++iter)
- {
- packetp = iter->second;
- gMessageSystem->mFailedResendPackets++;
- if(gMessageSystem->mVerboseLog)
- {
- doomed.push_back(packetp->mPacketID);
- }
- if (packetp->mCallback)
- {
- packetp->mCallback(packetp->mCallbackData,LL_ERR_CIRCUIT_GONE);
- }
-
- // Update stats
- mUnackedPacketCount--;
- mUnackedPacketBytes -= packetp->mBufferLength;
-
- delete packetp;
- }
-
- // remove all pending final retry reliable messages on this circuit
- end = mFinalRetryPackets.end();
- for(iter = mFinalRetryPackets.begin(); iter != end; ++iter)
- {
- packetp = iter->second;
- gMessageSystem->mFailedResendPackets++;
- if(gMessageSystem->mVerboseLog)
- {
- doomed.push_back(packetp->mPacketID);
- }
- if (packetp->mCallback)
- {
- packetp->mCallback(packetp->mCallbackData,LL_ERR_CIRCUIT_GONE);
- }
-
- // Update stats
- mUnackedPacketCount--;
- mUnackedPacketBytes -= packetp->mBufferLength;
-
- delete packetp;
- }
-
- // log aborted reliable packets for this circuit.
- if(gMessageSystem->mVerboseLog && !doomed.empty())
- {
- std::ostringstream str;
- std::ostream_iterator<TPACKETID> append(str, " ");
- str << "MSG: -> " << mHost << "\tABORTING RELIABLE:\t";
- std::copy(doomed.begin(), doomed.end(), append);
- LL_INFOS() << str.str() << LL_ENDL;
- }
-}
-
-
-void LLCircuitData::ackReliablePacket(TPACKETID packet_num)
-{
- reliable_iter iter;
- LLReliablePacket *packetp;
-
- iter = mUnackedPackets.find(packet_num);
- if (iter != mUnackedPackets.end())
- {
- packetp = iter->second;
-
- if(gMessageSystem->mVerboseLog)
- {
- std::ostringstream str;
- str << "MSG: <- " << packetp->mHost << "\tRELIABLE ACKED:\t"
- << packetp->mPacketID;
- LL_INFOS() << str.str() << LL_ENDL;
- }
- if (packetp->mCallback)
- {
- if (packetp->mTimeout < F32Seconds(0.f)) // negative timeout will always return timeout even for successful ack, for debugging
- {
- packetp->mCallback(packetp->mCallbackData,LL_ERR_TCP_TIMEOUT);
- }
- else
- {
- packetp->mCallback(packetp->mCallbackData,LL_ERR_NOERR);
- }
- }
-
- // Update stats
- mUnackedPacketCount--;
- mUnackedPacketBytes -= packetp->mBufferLength;
-
- // Cleanup
- delete packetp;
- mUnackedPackets.erase(iter);
- return;
- }
-
- iter = mFinalRetryPackets.find(packet_num);
- if (iter != mFinalRetryPackets.end())
- {
- packetp = iter->second;
- // LL_INFOS() << "Packet " << packet_num << " removed from the pending list" << LL_ENDL;
- if(gMessageSystem->mVerboseLog)
- {
- std::ostringstream str;
- str << "MSG: <- " << packetp->mHost << "\tRELIABLE ACKED:\t"
- << packetp->mPacketID;
- LL_INFOS() << str.str() << LL_ENDL;
- }
- if (packetp->mCallback)
- {
- if (packetp->mTimeout < F32Seconds(0.f)) // negative timeout will always return timeout even for successful ack, for debugging
- {
- packetp->mCallback(packetp->mCallbackData,LL_ERR_TCP_TIMEOUT);
- }
- else
- {
- packetp->mCallback(packetp->mCallbackData,LL_ERR_NOERR);
- }
- }
-
- // Update stats
- mUnackedPacketCount--;
- mUnackedPacketBytes -= packetp->mBufferLength;
-
- // Cleanup
- delete packetp;
- mFinalRetryPackets.erase(iter);
- }
- else
- {
- // Couldn't find this packet on either of the unacked lists.
- // maybe it's a duplicate ack?
- }
-}
-
-
-
-S32 LLCircuitData::resendUnackedPackets(const F64Seconds now)
-{
- LLReliablePacket *packetp;
-
-
- //
- // Theoretically we should search through the list for the packet with the oldest
- // packet ID, as otherwise when we WRAP we will resend reliable packets out of order.
- // Since resends are ALREADY out of order, and wrapping is highly rare (16+million packets),
- // I'm not going to worry about this for now - djs
- //
-
- reliable_iter iter;
- bool have_resend_overflow = false;
- for (iter = mUnackedPackets.begin(); iter != mUnackedPackets.end();)
- {
- packetp = iter->second;
-
- // Only check overflow if we haven't had one yet.
- if (!have_resend_overflow)
- {
- have_resend_overflow = mThrottles.checkOverflow(TC_RESEND, 0);
- }
-
- if (have_resend_overflow)
- {
- // We've exceeded our bandwidth for resends.
- // Time to stop trying to send them.
-
- // If we have too many unacked packets, we need to start dropping expired ones.
- if (mUnackedPacketBytes > 512000)
- {
- if (now > packetp->mExpirationTime)
- {
- // This circuit has overflowed. Do not retry. Do not pass go.
- packetp->mRetries = 0;
- // Remove it from this list and add it to the final list.
- mUnackedPackets.erase(iter++);
- mFinalRetryPackets[packetp->mPacketID] = packetp;
- }
- else
- {
- ++iter;
- }
- // Move on to the next unacked packet.
- continue;
- }
-
- if (mUnackedPacketBytes > 256000 && !(getPacketsOut() % 1024))
- {
- // Warn if we've got a lot of resends waiting.
- LL_WARNS() << mHost << " has " << mUnackedPacketBytes
- << " bytes of reliable messages waiting" << LL_ENDL;
- }
- // Stop resending. There are less than 512000 unacked packets.
- break;
- }
-
- if (now > packetp->mExpirationTime)
- {
- packetp->mRetries--;
-
- // retry
- mCurrentResendCount++;
-
- gMessageSystem->mResentPackets++;
-
- if(gMessageSystem->mVerboseLog)
- {
- std::ostringstream str;
- str << "MSG: -> " << packetp->mHost
- << "\tRESENDING RELIABLE:\t" << packetp->mPacketID;
- LL_INFOS() << str.str() << LL_ENDL;
- }
-
- packetp->mBuffer[0] |= LL_RESENT_FLAG; // tag packet id as being a resend
-
- gMessageSystem->mPacketRing.sendPacket(packetp->mSocket,
- (char *)packetp->mBuffer, packetp->mBufferLength,
- packetp->mHost);
-
- mThrottles.throttleOverflow(TC_RESEND, packetp->mBufferLength * 8.f);
-
- // The new method, retry time based on ping
- if (packetp->mPingBasedRetry)
- {
- packetp->mExpirationTime = now + llmax(LL_MINIMUM_RELIABLE_TIMEOUT_SECONDS, F32Seconds(LL_RELIABLE_TIMEOUT_FACTOR * getPingDelayAveraged()));
- }
- else
- {
- // custom, constant retry time
- packetp->mExpirationTime = now + packetp->mTimeout;
- }
-
- if (!packetp->mRetries)
- {
- // Last resend, remove it from this list and add it to the final list.
- mUnackedPackets.erase(iter++);
- mFinalRetryPackets[packetp->mPacketID] = packetp;
- }
- else
- {
- // Don't remove it yet, it still gets to try to resend at least once.
- ++iter;
- }
- }
- else
- {
- // Don't need to do anything with this packet, keep iterating.
- ++iter;
- }
- }
-
-
- for (iter = mFinalRetryPackets.begin(); iter != mFinalRetryPackets.end();)
- {
- packetp = iter->second;
- if (now > packetp->mExpirationTime)
- {
- // fail (too many retries)
- //LL_INFOS() << "Packet " << packetp->mPacketID << " removed from the pending list: exceeded retry limit" << LL_ENDL;
- //if (packetp->mMessageName)
- //{
- // LL_INFOS() << "Packet name " << packetp->mMessageName << LL_ENDL;
- //}
- gMessageSystem->mFailedResendPackets++;
-
- if(gMessageSystem->mVerboseLog)
- {
- std::ostringstream str;
- str << "MSG: -> " << packetp->mHost << "\tABORTING RELIABLE:\t"
- << packetp->mPacketID;
- LL_INFOS() << str.str() << LL_ENDL;
- }
-
- if (packetp->mCallback)
- {
- packetp->mCallback(packetp->mCallbackData,LL_ERR_TCP_TIMEOUT);
- }
-
- // Update stats
- mUnackedPacketCount--;
- mUnackedPacketBytes -= packetp->mBufferLength;
-
- mFinalRetryPackets.erase(iter++);
- delete packetp;
- }
- else
- {
- ++iter;
- }
- }
-
- return mUnackedPacketCount;
-}
-
-
-LLCircuit::LLCircuit(const F32Seconds circuit_heartbeat_interval, const F32Seconds circuit_timeout)
-: mLastCircuit(NULL),
- mHeartbeatInterval(circuit_heartbeat_interval),
- mHeartbeatTimeout(circuit_timeout)
-{}
-
-LLCircuit::~LLCircuit()
-{
- // delete pointers in the map.
- std::for_each(mCircuitData.begin(),
- mCircuitData.end(),
- llcompose1(
- DeletePointerFunctor<LLCircuitData>(),
- llselect2nd<circuit_data_map::value_type>()));
-}
-
-LLCircuitData *LLCircuit::addCircuitData(const LLHost &host, TPACKETID in_id)
-{
- // This should really validate if one already exists
- LL_INFOS() << "LLCircuit::addCircuitData for " << host << LL_ENDL;
- LLCircuitData *tempp = new LLCircuitData(host, in_id, mHeartbeatInterval, mHeartbeatTimeout);
- mCircuitData.insert(circuit_data_map::value_type(host, tempp));
- mPingSet.insert(tempp);
-
- mLastCircuit = tempp;
- return tempp;
-}
-
-void LLCircuit::removeCircuitData(const LLHost &host)
-{
- LL_INFOS() << "LLCircuit::removeCircuitData for " << host << LL_ENDL;
- mLastCircuit = NULL;
- circuit_data_map::iterator it = mCircuitData.find(host);
- if(it != mCircuitData.end())
- {
- LLCircuitData *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);
- 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;
-}
-
-void LLCircuitData::setAlive(bool b_alive)
-{
- if (mbAlive != b_alive)
- {
- mPacketsOutID = 0;
- mPacketsInID = 0;
- mbAlive = b_alive;
- }
- if (b_alive)
- {
- mLastPingReceivedTime = LLMessageSystem::getMessageTimeSeconds();
- mPingsInTransit = 0;
- mBlocked = false;
- }
-}
-
-
-void LLCircuitData::setAllowTimeout(bool allow)
-{
- mbAllowTimeout = allow;
-
- if (allow)
- {
- // resuming circuit
- // make sure it's alive
- setAlive(true);
- }
-}
-
-
-// Reset per-period counters if necessary.
-void LLCircuitData::checkPeriodTime()
-{
- F64Seconds mt_sec = LLMessageSystem::getMessageTimeSeconds();
- F64Seconds period_length = mt_sec - mPeriodTime;
- if ( period_length > TARGET_PERIOD_LENGTH)
- {
- F32 bps_in = F32Bits(mBytesInThisPeriod).value() / period_length.value();
- if (bps_in > mPeakBPSIn)
- {
- mPeakBPSIn = bps_in;
- }
-
- F32 bps_out = F32Bits(mBytesOutThisPeriod).value() / period_length.value();
- if (bps_out > mPeakBPSOut)
- {
- mPeakBPSOut = bps_out;
- }
-
- mBytesInLastPeriod = mBytesInThisPeriod;
- mBytesOutLastPeriod = mBytesOutThisPeriod;
- mBytesInThisPeriod = S32Bytes(0);
- mBytesOutThisPeriod = S32Bytes(0);
- mLastPeriodLength = F32Seconds::convert(period_length);
-
- mPeriodTime = mt_sec;
- }
-}
-
-
-void LLCircuitData::addBytesIn(S32Bytes bytes)
-{
- mBytesIn += bytes;
- mBytesInThisPeriod += bytes;
-}
-
-
-void LLCircuitData::addBytesOut(S32Bytes bytes)
-{
- mBytesOut += bytes;
- mBytesOutThisPeriod += bytes;
-}
-
-
-void LLCircuitData::addReliablePacket(S32 mSocket, U8 *buf_ptr, S32 buf_len, LLReliablePacketParams *params)
-{
- LLReliablePacket *packet_info;
-
- packet_info = new LLReliablePacket(mSocket, buf_ptr, buf_len, params);
-
- mUnackedPacketCount++;
- mUnackedPacketBytes += packet_info->mBufferLength;
-
- if (params && params->mRetries)
- {
- mUnackedPackets[packet_info->mPacketID] = packet_info;
- }
- else
- {
- mFinalRetryPackets[packet_info->mPacketID] = packet_info;
- }
-}
-
-
-void LLCircuit::resendUnackedPackets(S32& unacked_list_length, S32& unacked_list_size)
-{
- F64Seconds now = LLMessageSystem::getMessageTimeSeconds();
- unacked_list_length = 0;
- unacked_list_size = 0;
-
- LLCircuitData* circ;
- circuit_data_map::iterator end = mUnackedCircuitMap.end();
- for(circuit_data_map::iterator it = mUnackedCircuitMap.begin(); it != end; ++it)
- {
- circ = (*it).second;
- unacked_list_length += circ->resendUnackedPackets(now);
- unacked_list_size += circ->getUnackedPacketBytes();
- }
-}
-
-
-bool LLCircuitData::isDuplicateResend(TPACKETID packetnum)
-{
- return (mRecentlyReceivedReliablePackets.find(packetnum) != mRecentlyReceivedReliablePackets.end());
-}
-
-
-void LLCircuit::dumpResends()
-{
- circuit_data_map::iterator end = mCircuitData.end();
- for(circuit_data_map::iterator it = mCircuitData.begin(); it != end; ++it)
- {
- (*it).second->dumpResendCountAndReset();
- }
-}
-
-LLCircuitData* LLCircuit::findCircuit(const LLHost& host) const
-{
- // An optimization on finding the previously found circuit.
- if (mLastCircuit && (mLastCircuit->mHost == host))
- {
- return mLastCircuit;
- }
-
- circuit_data_map::const_iterator it = mCircuitData.find(host);
- if(it == mCircuitData.end())
- {
- return NULL;
- }
- mLastCircuit = it->second;
- return mLastCircuit;
-}
-
-
-bool LLCircuit::isCircuitAlive(const LLHost& host) const
-{
- LLCircuitData *cdp = findCircuit(host);
- if(cdp)
- {
- return cdp->mbAlive;
- }
-
- return false;
-}
-
-void LLCircuitData::setTimeoutCallback(void (*callback_func)(const LLHost &host, void *user_data), void *user_data)
-{
- mTimeoutCallback = callback_func;
- mTimeoutUserData = user_data;
-}
-
-void LLCircuitData::checkPacketInID(TPACKETID id, bool receive_resent)
-{
- // Done as floats so we don't have to worry about running out of room
- // with U32 getting poked into an S32.
- F32 delta = (F32)mHighestPacketID - (F32)id;
- if (delta > (0.5f*LL_MAX_OUT_PACKET_ID))
- {
- // We've almost definitely wrapped, reset the mLastPacketID to be low again.
- mHighestPacketID = id;
- }
- else if (delta < (-0.5f*LL_MAX_OUT_PACKET_ID))
- {
- // This is almost definitely an old packet coming in after a wrap, ignore it.
- }
- else
- {
- mHighestPacketID = llmax(mHighestPacketID, id);
- }
-
- // Save packet arrival time
- mLastPacketInTime = LLMessageSystem::getMessageTimeSeconds();
-
- // Have we received anything on this circuit yet?
- if (0 == mPacketsIn)
- {
- // Must be first packet from unclosed circuit.
- mPacketsIn++;
- setPacketInID((id + 1) % LL_MAX_OUT_PACKET_ID);
-
- mLastPacketGap = 0;
- return;
- }
-
- mPacketsIn++;
-
-
- // now, check to see if we've got a gap
- U32 gap = 0;
- if (mPacketsInID == id)
- {
- // nope! bump and wrap the counter, then return
- mPacketsInID++;
- mPacketsInID = (mPacketsInID) % LL_MAX_OUT_PACKET_ID;
- }
- else if (id < mWrapID)
- {
- // id < mWrapID will happen if the first few packets are out of order. . .
- // at that point we haven't marked anything "potentially lost" and
- // the out-of-order packet will cause a full wrap marking all the IDs "potentially lost"
-
- // do nothing
- }
- else
- {
- // we have a gap! if that id is in the map, remove it from the map, leave mCurrentCircuit->mPacketsInID
- // alone
- // otherwise, walk from mCurrentCircuit->mPacketsInID to id with wrapping, adding the values to the map
- // and setting mPacketsInID to id + 1 % LL_MAX_OUT_PACKET_ID
-
- // babbage: all operands in expression are unsigned, so modular
- // arithmetic will always find correct gap, regardless of wrap arounds.
- const U8 width = 24;
- gap = LLModularMath::subtract<width>(mPacketsInID, id);
-
- if (mPotentialLostPackets.find(id) != mPotentialLostPackets.end())
- {
- if(gMessageSystem->mVerboseLog)
- {
- std::ostringstream str;
- str << "MSG: <- " << mHost << "\tRECOVERING LOST:\t" << id;
- LL_INFOS() << str.str() << LL_ENDL;
- }
- // LL_INFOS() << "removing potential lost: " << id << LL_ENDL;
- mPotentialLostPackets.erase(id);
- }
- else if (!receive_resent) // don't freak out over out-of-order reliable resends
- {
- U64Microseconds time = LLMessageSystem::getMessageTimeUsecs();
- TPACKETID index = mPacketsInID;
- S32 gap_count = 0;
- if ((index < id) && ((id - index) < 16))
- {
- while (index != id)
- {
- if(gMessageSystem->mVerboseLog)
- {
- std::ostringstream str;
- str << "MSG: <- " << mHost << "\tPACKET GAP:\t"
- << index;
- LL_INFOS() << str.str() << LL_ENDL;
- }
-
-// LL_INFOS() << "adding potential lost: " << index << LL_ENDL;
- mPotentialLostPackets[index] = time;
- index++;
- index = index % LL_MAX_OUT_PACKET_ID;
- gap_count++;
- }
- }
- else
- {
- LL_INFOS() << "packet_out_of_order - got packet " << id << " expecting " << index << " from " << mHost << LL_ENDL;
- if(gMessageSystem->mVerboseLog)
- {
- std::ostringstream str;
- str << "MSG: <- " << mHost << "\tPACKET GAP:\t"
- << id << " expected " << index;
- LL_INFOS() << str.str() << LL_ENDL;
- }
- }
-
- mPacketsInID = id + 1;
- mPacketsInID = (mPacketsInID) % LL_MAX_OUT_PACKET_ID;
-
- if (gap_count > 128)
- {
- LL_WARNS() << "Packet loss gap filler running amok!" << LL_ENDL;
- }
- else if (gap_count > 16)
- {
- LL_WARNS() << "Sustaining large amounts of packet loss!" << LL_ENDL;
- }
-
- }
- }
- mLastPacketGap = gap;
-}
-
-
-void LLCircuit::updateWatchDogTimers(LLMessageSystem *msgsys)
-{
- F64Seconds cur_time = LLMessageSystem::getMessageTimeSeconds();
- S32 count = mPingSet.size();
- S32 cur = 0;
-
- // Only process each circuit once at most, stop processing if no circuits
- while((cur < count) && !mPingSet.empty())
- {
- cur++;
-
- LLCircuit::ping_set_t::iterator psit = mPingSet.begin();
- LLCircuitData *cdp = *psit;
-
- if (!cdp->mbAlive)
- {
- // We suspect that this case should never happen, given how
- // the alive status is set.
- // Skip over dead circuits, just add the ping interval and push it to the back
- // Always remember to remove it from the set before changing the sorting
- // key (mNextPingSendTime)
- mPingSet.erase(psit);
- cdp->mNextPingSendTime = cur_time + mHeartbeatInterval;
- mPingSet.insert(cdp);
- continue;
- }
- else
- {
- // Check to see if this needs a ping
- if (cur_time < cdp->mNextPingSendTime)
- {
- // This circuit doesn't need a ping, break out because
- // we have a sorted list, thus no more circuits need pings
- break;
- }
-
- // Update watchdog timers
- if (cdp->updateWatchDogTimers(msgsys))
- {
- // Randomize our pings a bit by doing some up to 5% early or late
- F64Seconds dt = 0.95f*mHeartbeatInterval + F32Seconds(ll_frand(0.1f*mHeartbeatInterval.value()));
-
- // Remove it, and reinsert it with the new next ping time.
- // Always remove before changing the sorting key.
- mPingSet.erase(psit);
- cdp->mNextPingSendTime = cur_time + dt;
- mPingSet.insert(cdp);
-
- // Update our throttles
- cdp->mThrottles.dynamicAdjust();
-
- // Update some stats, this is not terribly important
- cdp->checkPeriodTime();
- }
- else
- {
- // This mPingSet.erase isn't necessary, because removing the circuit will
- // remove the ping set.
- //mPingSet.erase(psit);
- removeCircuitData(cdp->mHost);
- }
- }
- }
-}
-
-
-bool LLCircuitData::updateWatchDogTimers(LLMessageSystem *msgsys)
-{
- F64Seconds cur_time = LLMessageSystem::getMessageTimeSeconds();
- mLastPingSendTime = cur_time;
-
- if (!checkCircuitTimeout())
- {
- // Pass this back to the calling LLCircuit, this circuit needs to be cleaned up.
- return false;
- }
-
- // WARNING!
- // Duplicate suppression can FAIL if packets are delivered out of
- // order, although it's EXTREMELY unlikely. It would require
- // that the ping get delivered out of order enough that the ACK
- // for the packet that it was out of order with was received BEFORE
- // the ping was sent.
-
- // Find the current oldest reliable packetID
- // This is to handle the case if we actually manage to wrap our
- // packet IDs - the oldest will actually have a higher packet ID
- // than the current.
- bool wrapped = false;
- reliable_iter iter;
- iter = mUnackedPackets.upper_bound(getPacketOutID());
- if (iter == mUnackedPackets.end())
- {
- // Nothing AFTER this one, so we want the lowest packet ID
- // then.
- iter = mUnackedPackets.begin();
- wrapped = true;
- }
-
- TPACKETID packet_id = 0;
-
- // Check against the "final" packets
- bool wrapped_final = false;
- reliable_iter iter_final;
- iter_final = mFinalRetryPackets.upper_bound(getPacketOutID());
- if (iter_final == mFinalRetryPackets.end())
- {
- iter_final = mFinalRetryPackets.begin();
- wrapped_final = true;
- }
-
- //LL_INFOS() << mHost << " - unacked count " << mUnackedPackets.size() << LL_ENDL;
- //LL_INFOS() << mHost << " - final count " << mFinalRetryPackets.size() << LL_ENDL;
- if (wrapped != wrapped_final)
- {
- // One of the "unacked" or "final" lists hasn't wrapped. Whichever one
- // hasn't has the oldest packet.
- if (!wrapped)
- {
- // Hasn't wrapped, so the one on the
- // unacked packet list is older
- packet_id = iter->first;
- //LL_INFOS() << mHost << ": nowrapped unacked" << LL_ENDL;
- }
- else
- {
- packet_id = iter_final->first;
- //LL_INFOS() << mHost << ": nowrapped final" << LL_ENDL;
- }
- }
- else
- {
- // They both wrapped, we can just use the minimum of the two.
- if ((iter == mUnackedPackets.end()) && (iter_final == mFinalRetryPackets.end()))
- {
- // Wow! No unacked packets at all!
- // Send the ID of the last packet we sent out.
- // This will flush all of the destination's
- // unacked packets, theoretically.
- //LL_INFOS() << mHost << ": No unacked!" << LL_ENDL;
- packet_id = getPacketOutID();
- }
- else
- {
- bool had_unacked = false;
- if (iter != mUnackedPackets.end())
- {
- // Unacked list has the lowest so far
- packet_id = iter->first;
- had_unacked = true;
- //LL_INFOS() << mHost << ": Unacked" << LL_ENDL;
- }
-
- if (iter_final != mFinalRetryPackets.end())
- {
- // Use the lowest of the unacked list and the final list
- if (had_unacked)
- {
- // Both had a packet, use the lowest.
- packet_id = llmin(packet_id, iter_final->first);
- //LL_INFOS() << mHost << ": Min of unacked/final" << LL_ENDL;
- }
- else
- {
- // Only the final had a packet, use it.
- packet_id = iter_final->first;
- //LL_INFOS() << mHost << ": Final!" << LL_ENDL;
- }
- }
- }
- }
-
- // Send off the another ping.
- pingTimerStart();
- msgsys->newMessageFast(_PREHASH_StartPingCheck);
- msgsys->nextBlock(_PREHASH_PingID);
- msgsys->addU8Fast(_PREHASH_PingID, nextPingID());
- msgsys->addU32Fast(_PREHASH_OldestUnacked, packet_id);
- msgsys->sendMessage(mHost);
-
- // Also do lost packet accounting.
- // Check to see if anything on our lost list is old enough to
- // be considered lost
-
- LLCircuitData::packet_time_map::iterator it;
- U64Microseconds timeout = llmin(LL_MAX_LOST_TIMEOUT, F32Seconds(getPingDelayAveraged()) * LL_LOST_TIMEOUT_FACTOR);
-
- U64Microseconds mt_usec = LLMessageSystem::getMessageTimeUsecs();
- for (it = mPotentialLostPackets.begin(); it != mPotentialLostPackets.end(); )
- {
- U64Microseconds delta_t_usec = mt_usec - (*it).second;
- if (delta_t_usec > timeout)
- {
- // let's call this one a loss!
- mPacketsLost++;
- gMessageSystem->mDroppedPackets++;
- if(gMessageSystem->mVerboseLog)
- {
- std::ostringstream str;
- str << "MSG: <- " << mHost << "\tLOST PACKET:\t"
- << (*it).first;
- LL_INFOS() << str.str() << LL_ENDL;
- }
- mPotentialLostPackets.erase(it++);
- }
- else
- {
- ++it;
- }
- }
-
- return true;
-}
-
-
-void LLCircuitData::clearDuplicateList(TPACKETID oldest_id)
-{
- // purge old data from the duplicate suppression queue
-
- // we want to KEEP all x where oldest_id <= x <= last incoming packet, and delete everything else.
-
- //LL_INFOS() << mHost << ": clearing before oldest " << oldest_id << LL_ENDL;
- //LL_INFOS() << "Recent list before: " << mRecentlyReceivedReliablePackets.size() << LL_ENDL;
- if (oldest_id < mHighestPacketID)
- {
- // Clean up everything with a packet ID less than oldest_id.
- packet_time_map::iterator pit_start;
- packet_time_map::iterator pit_end;
- pit_start = mRecentlyReceivedReliablePackets.begin();
- pit_end = mRecentlyReceivedReliablePackets.lower_bound(oldest_id);
- mRecentlyReceivedReliablePackets.erase(pit_start, pit_end);
- }
-
- // Do timeout checks on everything with an ID > mHighestPacketID.
- // This should be empty except for wrapping IDs. Thus, this should be
- // highly rare.
- U64Microseconds mt_usec = LLMessageSystem::getMessageTimeUsecs();
-
- packet_time_map::iterator pit;
- for(pit = mRecentlyReceivedReliablePackets.upper_bound(mHighestPacketID);
- pit != mRecentlyReceivedReliablePackets.end(); )
- {
- // Validate that the packet ID seems far enough away
- if ((pit->first - mHighestPacketID) < 100)
- {
- LL_WARNS() << "Probably incorrectly timing out non-wrapped packets!" << LL_ENDL;
- }
- U64Microseconds delta_t_usec = mt_usec - (*pit).second;
- F64Seconds delta_t_sec = delta_t_usec;
- if (delta_t_sec > LL_DUPLICATE_SUPPRESSION_TIMEOUT)
- {
- // enough time has elapsed we're not likely to get a duplicate on this one
- LL_INFOS() << "Clearing " << pit->first << " from recent list" << LL_ENDL;
- mRecentlyReceivedReliablePackets.erase(pit++);
- }
- else
- {
- ++pit;
- }
- }
- //LL_INFOS() << "Recent list after: " << mRecentlyReceivedReliablePackets.size() << LL_ENDL;
-}
-
-bool LLCircuitData::checkCircuitTimeout()
-{
- F64Seconds time_since_last_ping = LLMessageSystem::getMessageTimeSeconds() - mLastPingReceivedTime;
-
- // Nota Bene: This needs to be turned off if you are debugging multiple simulators
- if (time_since_last_ping > mHeartbeatTimeout)
- {
- LL_WARNS() << "LLCircuitData::checkCircuitTimeout for " << mHost << " last ping " << time_since_last_ping << " seconds ago." <<LL_ENDL;
- setAlive(false);
- if (mTimeoutCallback)
- {
- LL_WARNS() << "LLCircuitData::checkCircuitTimeout for " << mHost << " calling callback." << LL_ENDL;
- mTimeoutCallback(mHost, mTimeoutUserData);
- }
- if (!isAlive())
- {
- // The callback didn't try and resurrect the circuit. We should kill it.
- LL_WARNS() << "LLCircuitData::checkCircuitTimeout for " << mHost << " still dead, dropping." << LL_ENDL;
- return false;
- }
- }
-
- return true;
-}
-
-// Call this method when a reliable message comes in - this will
-// correctly place the packet in the correct list to be acked later.
-bool LLCircuitData::collectRAck(TPACKETID packet_num)
-{
- if (mAcks.empty())
- {
- // 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);
- if (mAckCreationTime == 0)
- {
- mAckCreationTime = getAgeInSeconds();
- }
- return true;
-}
-
-// this method is called during the message system processAcks() to
-// send out any acks that did not get sent already.
-void LLCircuit::sendAcks(F32 collect_time)
-{
- collect_time = llclamp(collect_time, 0.f, LL_COLLECT_ACK_TIME_MAX);
- LLCircuitData* cd;
- circuit_data_map::iterator it = mSendAckMap.begin();
- while (it != mSendAckMap.end())
- {
- circuit_data_map::iterator cur_it = it++;
- cd = (*cur_it).second;
- S32 count = (S32)cd->mAcks.size();
- F32 age = cd->getAgeInSeconds() - cd->mAckCreationTime;
- if (age > collect_time || count == 0)
- {
- if (count>0)
- {
- // send the packet acks
- S32 acks_this_packet = 0;
- for(S32 i = 0; i < count; ++i)
- {
- if(acks_this_packet == 0)
- {
- gMessageSystem->newMessageFast(_PREHASH_PacketAck);
- }
- gMessageSystem->nextBlockFast(_PREHASH_Packets);
- gMessageSystem->addU32Fast(_PREHASH_ID, cd->mAcks[i]);
- ++acks_this_packet;
- if(acks_this_packet > 250)
- {
- gMessageSystem->sendMessage(cd->mHost);
- acks_this_packet = 0;
- }
- }
- if(acks_this_packet > 0)
- {
- gMessageSystem->sendMessage(cd->mHost);
- }
-
- if(gMessageSystem->mVerboseLog)
- {
- std::ostringstream str;
- str << "MSG: -> " << cd->mHost << "\tPACKET ACKS:\t";
- std::ostream_iterator<TPACKETID> append(str, " ");
- std::copy(cd->mAcks.begin(), cd->mAcks.end(), append);
- LL_INFOS() << str.str() << LL_ENDL;
- }
-
- // empty out the acks list
- cd->mAcks.clear();
- cd->mAckCreationTime = 0.f;
- }
- // remove data map
- mSendAckMap.erase(cur_it);
- }
- }
-}
-
-
-std::ostream& operator<<(std::ostream& s, LLCircuitData& circuit)
-{
- F32 age = circuit.mExistenceTimer.getElapsedTimeF32();
-
- using namespace std;
- s << "Circuit " << circuit.mHost << " "
- << circuit.mRemoteID << " "
- << (circuit.mbAlive ? "Alive" : "Not Alive") << " "
- << (circuit.mbAllowTimeout ? "Timeout Allowed" : "Timeout Not Allowed")
- << endl;
-
- s << " Packets Lost: " << circuit.mPacketsLost
- << " Measured Ping: " << circuit.mPingDelay
- << " Averaged Ping: " << circuit.mPingDelayAveraged
- << endl;
-
- s << "Global In/Out " << S32(age) << " sec"
- << " KBytes: " << circuit.mBytesIn.valueInUnits<LLUnits::Kilobytes>() << "/" << circuit.mBytesOut.valueInUnits<LLUnits::Kilobytes>()
- << " Kbps: "
- << S32(circuit.mBytesIn.valueInUnits<LLUnits::Kilobits>() / circuit.mExistenceTimer.getElapsedTimeF32().value())
- << "/"
- << S32(circuit.mBytesOut.valueInUnits<LLUnits::Kilobits>() / circuit.mExistenceTimer.getElapsedTimeF32().value())
- << " Packets: " << circuit.mPacketsIn << "/" << circuit.mPacketsOut
- << endl;
-
- s << "Recent In/Out " << circuit.mLastPeriodLength
- << " KBytes: "
- << circuit.mBytesInLastPeriod.valueInUnits<LLUnits::Kilobytes>()
- << "/"
- << circuit.mBytesOutLastPeriod.valueInUnits<LLUnits::Kilobytes>()
- << " Kbps: "
- << (S32)(circuit.mBytesInLastPeriod.valueInUnits<LLUnits::Kilobits>() / circuit.mLastPeriodLength.value())
- << "/"
- << (S32)(circuit.mBytesOutLastPeriod.valueInUnits<LLUnits::Kilobits>() / circuit.mLastPeriodLength.value())
- << " Peak kbps: "
- << S32(circuit.mPeakBPSIn / 1024.f)
- << "/"
- << S32(circuit.mPeakBPSOut / 1024.f)
- << endl;
-
- return s;
-}
-
-void LLCircuitData::getInfo(LLSD& info) const
-{
- info["Host"] = mHost.getIPandPort();
- info["Alive"] = mbAlive;
- info["Age"] = mExistenceTimer.getElapsedTimeF32();
-}
-
-void LLCircuitData::dumpResendCountAndReset()
-{
- if (mCurrentResendCount)
- {
- LL_INFOS() << "Circuit: " << mHost << " resent " << mCurrentResendCount << " packets" << LL_ENDL;
- mCurrentResendCount = 0;
- }
-}
-
-std::ostream& operator<<(std::ostream& s, LLCircuit &circuit)
-{
- s << "Circuit Info:" << std::endl;
- LLCircuit::circuit_data_map::iterator end = circuit.mCircuitData.end();
- LLCircuit::circuit_data_map::iterator it;
- for(it = circuit.mCircuitData.begin(); it != end; ++it)
- {
- s << *((*it).second) << std::endl;
- }
- return s;
-}
-
-void LLCircuit::getInfo(LLSD& info) const
-{
- LLCircuit::circuit_data_map::const_iterator end = mCircuitData.end();
- LLCircuit::circuit_data_map::const_iterator it;
- LLSD circuit_info;
- for(it = mCircuitData.begin(); it != end; ++it)
- {
- (*it).second->getInfo(circuit_info);
- info["Circuits"].append(circuit_info);
- }
-}
-
-void LLCircuit::getCircuitRange(
- const LLHost& key,
- LLCircuit::circuit_data_map::iterator& first,
- LLCircuit::circuit_data_map::iterator& end)
-{
- end = mCircuitData.end();
- first = mCircuitData.upper_bound(key);
-}
-
-TPACKETID LLCircuitData::nextPacketOutID()
-{
- mPacketsOut++;
-
- TPACKETID id;
-
- id = (mPacketsOutID + 1) % LL_MAX_OUT_PACKET_ID;
-
- if (id < mPacketsOutID)
- {
- // we just wrapped on a circuit, reset the wrap ID to zero
- mWrapID = 0;
- }
- mPacketsOutID = id;
- return id;
-}
-
-
-void LLCircuitData::setPacketInID(TPACKETID id)
-{
- id = id % LL_MAX_OUT_PACKET_ID;
- mPacketsInID = id;
- mRecentlyReceivedReliablePackets.clear();
-
- mWrapID = id;
-}
-
-
-void LLCircuitData::pingTimerStop(const U8 ping_id)
-{
- F64Seconds mt_secs = LLMessageSystem::getMessageTimeSeconds();
-
- // Nota Bene: no averaging of ping times until we get a feel for how this works
- F64Seconds time = mt_secs - mPingTime;
- if (time == F32Seconds(0.0))
- {
- // Ack, we got our ping response on the same frame! Sigh, let's get a real time otherwise
- // all of our ping calculations will be skewed.
- mt_secs = LLMessageSystem::getMessageTimeSeconds(true);
- }
- mLastPingReceivedTime = mt_secs;
-
- // If ping is longer than 1 second, we'll get sequence deltas in the ping.
- // Approximate by assuming each ping counts for 1 second (slightly low, probably)
- S32 delta_ping = (S32)mLastPingID - (S32) ping_id;
- if (delta_ping < 0)
- {
- delta_ping += 256;
- }
-
- U32Milliseconds msec = delta_ping*mHeartbeatInterval + time;
- setPingDelay(msec);
-
- mPingsInTransit = delta_ping;
- if (mBlocked && (mPingsInTransit <= PING_RELEASE_BLOCK))
- {
- mBlocked = false;
- }
-}
-
-
-void LLCircuitData::pingTimerStart()
-{
- mPingTime = LLMessageSystem::getMessageTimeSeconds();
- mPingsInTransit++;
-
- if (!mBlocked && (mPingsInTransit > PING_START_BLOCK))
- {
- mBlocked = true;
- }
-}
-
-
-U32 LLCircuitData::getPacketsIn() const
-{
- return mPacketsIn;
-}
-
-
-S32Bytes LLCircuitData::getBytesIn() const
-{
- return mBytesIn;
-}
-
-
-S32Bytes LLCircuitData::getBytesOut() const
-{
- return mBytesOut;
-}
-
-
-U32 LLCircuitData::getPacketsOut() const
-{
- return mPacketsOut;
-}
-
-
-TPACKETID LLCircuitData::getPacketOutID() const
-{
- return mPacketsOutID;
-}
-
-
-U32 LLCircuitData::getPacketsLost() const
-{
- return mPacketsLost;
-}
-
-
-bool LLCircuitData::isAlive() const
-{
- return mbAlive;
-}
-
-
-bool LLCircuitData::isBlocked() const
-{
- return mBlocked;
-}
-
-
-bool LLCircuitData::getAllowTimeout() const
-{
- return mbAllowTimeout;
-}
-
-
-U32Milliseconds LLCircuitData::getPingDelay() const
-{
- return mPingDelay;
-}
-
-
-F32Milliseconds LLCircuitData::getPingInTransitTime()
-{
- // This may be inaccurate in the case of a circuit that was "dead" and then revived,
- // but only until the first round trip ping is sent - djs
- F32Milliseconds time_since_ping_was_sent(0);
-
- if (mPingsInTransit)
- {
- time_since_ping_was_sent = F32Milliseconds::convert(((mPingsInTransit*mHeartbeatInterval - F32Seconds(1))
- + (LLMessageSystem::getMessageTimeSeconds() - mPingTime)));
- }
-
- return time_since_ping_was_sent;
-}
-
-
-void LLCircuitData::setPingDelay(U32Milliseconds ping)
-{
- mPingDelay = ping;
- mPingDelayAveraged = llmax((F32Milliseconds)ping, getPingDelayAveraged());
- mPingDelayAveraged = ((1.f - LL_AVERAGED_PING_ALPHA) * mPingDelayAveraged)
- + (LL_AVERAGED_PING_ALPHA * (F32Milliseconds) ping);
- mPingDelayAveraged = llclamp(mPingDelayAveraged,
- LL_AVERAGED_PING_MIN,
- LL_AVERAGED_PING_MAX);
-}
-
-
-F32Milliseconds LLCircuitData::getPingDelayAveraged()
-{
- return llmin(llmax(getPingInTransitTime(), mPingDelayAveraged), LL_AVERAGED_PING_MAX);
-}
-
-
-bool LLCircuitData::getTrusted() const
-{
- return mTrusted;
-}
-
-
-void LLCircuitData::setTrusted(bool t)
-{
- mTrusted = t;
-}
-
-F32 LLCircuitData::getAgeInSeconds() const
-{
- return mExistenceTimer.getElapsedTimeF32();
-}
+/** + * @file llcircuit.cpp + * @brief Class to track UDP endpoints for the message system. + * + * $LicenseInfo:firstyear=2002&license=viewerlgpl$ + * Second Life Viewer Source Code + * Copyright (C) 2010, Linden Research, Inc. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; + * version 2.1 of the License only. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + * + * Linden Research, Inc., 945 Battery Street, San Francisco, CA 94111 USA + * $/LicenseInfo$ + */ + +#include "linden_common.h" + +#if LL_WINDOWS + +#include <process.h> + +#else + +#if LL_LINUX +#include <dlfcn.h> // RTLD_LAZY +#endif +#include <sys/types.h> +#include <sys/socket.h> +#include <netinet/in.h> + +#endif + + +#if !defined(USE_CIRCUIT_LIST) +#include <algorithm> +#endif +#include <sstream> +#include <iterator> +#include <stack> + +#include "llcircuit.h" + +#include "message.h" +#include "llrand.h" +#include "llstl.h" +#include "lltransfermanager.h" +#include "llmodularmath.h" + +const S32 PING_START_BLOCK = 3; // How many pings behind we have to be to consider ourself blocked. +const S32 PING_RELEASE_BLOCK = 2; // How many pings behind we have to be to consider ourself unblocked. + +const F32Seconds TARGET_PERIOD_LENGTH(5.f); +const F32Seconds LL_DUPLICATE_SUPPRESSION_TIMEOUT(60.f); //this can be long, as time-based cleanup is + // only done when wrapping packetids, now... + +LLCircuitData::LLCircuitData(const LLHost &host, TPACKETID in_id, + const F32Seconds circuit_heartbeat_interval, const F32Seconds circuit_timeout) +: mHost (host), + mWrapID(0), + mPacketsOutID(0), + mPacketsInID(in_id), + mHighestPacketID(in_id), + mTimeoutCallback(NULL), + mTimeoutUserData(NULL), + mTrusted(false), + mbAllowTimeout(true), + mbAlive(true), + mBlocked(false), + mPingTime(0.0), + mLastPingSendTime(0.0), + mLastPingReceivedTime(0.0), + mNextPingSendTime(0.0), + mPingsInTransit(0), + mLastPingID(0), + mPingDelay(INITIAL_PING_VALUE_MSEC), + mPingDelayAveraged(INITIAL_PING_VALUE_MSEC), + mUnackedPacketCount(0), + mUnackedPacketBytes(0), + mLastPacketInTime(0.0), + mLocalEndPointID(), + mPacketsOut(0), + mPacketsIn(0), + mPacketsLost(0), + mBytesIn(0), + mBytesOut(0), + mLastPeriodLength(-1.f), + mBytesInLastPeriod(0), + mBytesOutLastPeriod(0), + mBytesInThisPeriod(0), + mBytesOutThisPeriod(0), + mPeakBPSIn(0.f), + mPeakBPSOut(0.f), + mPeriodTime(0.0), + mExistenceTimer(), + mAckCreationTime(0.f), + mCurrentResendCount(0), + mLastPacketGap(0), + mHeartbeatInterval(circuit_heartbeat_interval), + mHeartbeatTimeout(circuit_timeout) +{ + // Need to guarantee that this time is up to date, we may be creating a circuit even though we haven't been + // running a message system loop. + F64Seconds mt_sec = LLMessageSystem::getMessageTimeSeconds(true); + F32 distribution_offset = ll_frand(); + + mPingTime = mt_sec; + mLastPingSendTime = mt_sec + mHeartbeatInterval * distribution_offset; + mLastPingReceivedTime = mt_sec; + mNextPingSendTime = mLastPingSendTime + 0.95*mHeartbeatInterval + F32Seconds(ll_frand(0.1f*mHeartbeatInterval.value())); + mPeriodTime = mt_sec; + + mLocalEndPointID.generate(); +} + + +LLCircuitData::~LLCircuitData() +{ + LLReliablePacket *packetp = NULL; + + // Clean up all pending transfers. + gTransferManager.cleanupConnection(mHost); + + // remove all pending reliable messages on this circuit + std::vector<TPACKETID> doomed; + reliable_iter iter; + reliable_iter end = mUnackedPackets.end(); + for(iter = mUnackedPackets.begin(); iter != end; ++iter) + { + packetp = iter->second; + gMessageSystem->mFailedResendPackets++; + if(gMessageSystem->mVerboseLog) + { + doomed.push_back(packetp->mPacketID); + } + if (packetp->mCallback) + { + packetp->mCallback(packetp->mCallbackData,LL_ERR_CIRCUIT_GONE); + } + + // Update stats + mUnackedPacketCount--; + mUnackedPacketBytes -= packetp->mBufferLength; + + delete packetp; + } + + // remove all pending final retry reliable messages on this circuit + end = mFinalRetryPackets.end(); + for(iter = mFinalRetryPackets.begin(); iter != end; ++iter) + { + packetp = iter->second; + gMessageSystem->mFailedResendPackets++; + if(gMessageSystem->mVerboseLog) + { + doomed.push_back(packetp->mPacketID); + } + if (packetp->mCallback) + { + packetp->mCallback(packetp->mCallbackData,LL_ERR_CIRCUIT_GONE); + } + + // Update stats + mUnackedPacketCount--; + mUnackedPacketBytes -= packetp->mBufferLength; + + delete packetp; + } + + // log aborted reliable packets for this circuit. + if(gMessageSystem->mVerboseLog && !doomed.empty()) + { + std::ostringstream str; + std::ostream_iterator<TPACKETID> append(str, " "); + str << "MSG: -> " << mHost << "\tABORTING RELIABLE:\t"; + std::copy(doomed.begin(), doomed.end(), append); + LL_INFOS() << str.str() << LL_ENDL; + } +} + + +void LLCircuitData::ackReliablePacket(TPACKETID packet_num) +{ + reliable_iter iter; + LLReliablePacket *packetp; + + iter = mUnackedPackets.find(packet_num); + if (iter != mUnackedPackets.end()) + { + packetp = iter->second; + + if(gMessageSystem->mVerboseLog) + { + std::ostringstream str; + str << "MSG: <- " << packetp->mHost << "\tRELIABLE ACKED:\t" + << packetp->mPacketID; + LL_INFOS() << str.str() << LL_ENDL; + } + if (packetp->mCallback) + { + if (packetp->mTimeout < F32Seconds(0.f)) // negative timeout will always return timeout even for successful ack, for debugging + { + packetp->mCallback(packetp->mCallbackData,LL_ERR_TCP_TIMEOUT); + } + else + { + packetp->mCallback(packetp->mCallbackData,LL_ERR_NOERR); + } + } + + // Update stats + mUnackedPacketCount--; + mUnackedPacketBytes -= packetp->mBufferLength; + + // Cleanup + delete packetp; + mUnackedPackets.erase(iter); + return; + } + + iter = mFinalRetryPackets.find(packet_num); + if (iter != mFinalRetryPackets.end()) + { + packetp = iter->second; + // LL_INFOS() << "Packet " << packet_num << " removed from the pending list" << LL_ENDL; + if(gMessageSystem->mVerboseLog) + { + std::ostringstream str; + str << "MSG: <- " << packetp->mHost << "\tRELIABLE ACKED:\t" + << packetp->mPacketID; + LL_INFOS() << str.str() << LL_ENDL; + } + if (packetp->mCallback) + { + if (packetp->mTimeout < F32Seconds(0.f)) // negative timeout will always return timeout even for successful ack, for debugging + { + packetp->mCallback(packetp->mCallbackData,LL_ERR_TCP_TIMEOUT); + } + else + { + packetp->mCallback(packetp->mCallbackData,LL_ERR_NOERR); + } + } + + // Update stats + mUnackedPacketCount--; + mUnackedPacketBytes -= packetp->mBufferLength; + + // Cleanup + delete packetp; + mFinalRetryPackets.erase(iter); + } + else + { + // Couldn't find this packet on either of the unacked lists. + // maybe it's a duplicate ack? + } +} + + + +S32 LLCircuitData::resendUnackedPackets(const F64Seconds now) +{ + LLReliablePacket *packetp; + + + // + // Theoretically we should search through the list for the packet with the oldest + // packet ID, as otherwise when we WRAP we will resend reliable packets out of order. + // Since resends are ALREADY out of order, and wrapping is highly rare (16+million packets), + // I'm not going to worry about this for now - djs + // + + reliable_iter iter; + bool have_resend_overflow = false; + for (iter = mUnackedPackets.begin(); iter != mUnackedPackets.end();) + { + packetp = iter->second; + + // Only check overflow if we haven't had one yet. + if (!have_resend_overflow) + { + have_resend_overflow = mThrottles.checkOverflow(TC_RESEND, 0); + } + + if (have_resend_overflow) + { + // We've exceeded our bandwidth for resends. + // Time to stop trying to send them. + + // If we have too many unacked packets, we need to start dropping expired ones. + if (mUnackedPacketBytes > 512000) + { + if (now > packetp->mExpirationTime) + { + // This circuit has overflowed. Do not retry. Do not pass go. + packetp->mRetries = 0; + // Remove it from this list and add it to the final list. + mUnackedPackets.erase(iter++); + mFinalRetryPackets[packetp->mPacketID] = packetp; + } + else + { + ++iter; + } + // Move on to the next unacked packet. + continue; + } + + if (mUnackedPacketBytes > 256000 && !(getPacketsOut() % 1024)) + { + // Warn if we've got a lot of resends waiting. + LL_WARNS() << mHost << " has " << mUnackedPacketBytes + << " bytes of reliable messages waiting" << LL_ENDL; + } + // Stop resending. There are less than 512000 unacked packets. + break; + } + + if (now > packetp->mExpirationTime) + { + packetp->mRetries--; + + // retry + mCurrentResendCount++; + + gMessageSystem->mResentPackets++; + + if(gMessageSystem->mVerboseLog) + { + std::ostringstream str; + str << "MSG: -> " << packetp->mHost + << "\tRESENDING RELIABLE:\t" << packetp->mPacketID; + LL_INFOS() << str.str() << LL_ENDL; + } + + packetp->mBuffer[0] |= LL_RESENT_FLAG; // tag packet id as being a resend + + gMessageSystem->mPacketRing.sendPacket(packetp->mSocket, + (char *)packetp->mBuffer, packetp->mBufferLength, + packetp->mHost); + + mThrottles.throttleOverflow(TC_RESEND, packetp->mBufferLength * 8.f); + + // The new method, retry time based on ping + if (packetp->mPingBasedRetry) + { + packetp->mExpirationTime = now + llmax(LL_MINIMUM_RELIABLE_TIMEOUT_SECONDS, F32Seconds(LL_RELIABLE_TIMEOUT_FACTOR * getPingDelayAveraged())); + } + else + { + // custom, constant retry time + packetp->mExpirationTime = now + packetp->mTimeout; + } + + if (!packetp->mRetries) + { + // Last resend, remove it from this list and add it to the final list. + mUnackedPackets.erase(iter++); + mFinalRetryPackets[packetp->mPacketID] = packetp; + } + else + { + // Don't remove it yet, it still gets to try to resend at least once. + ++iter; + } + } + else + { + // Don't need to do anything with this packet, keep iterating. + ++iter; + } + } + + + for (iter = mFinalRetryPackets.begin(); iter != mFinalRetryPackets.end();) + { + packetp = iter->second; + if (now > packetp->mExpirationTime) + { + // fail (too many retries) + //LL_INFOS() << "Packet " << packetp->mPacketID << " removed from the pending list: exceeded retry limit" << LL_ENDL; + //if (packetp->mMessageName) + //{ + // LL_INFOS() << "Packet name " << packetp->mMessageName << LL_ENDL; + //} + gMessageSystem->mFailedResendPackets++; + + if(gMessageSystem->mVerboseLog) + { + std::ostringstream str; + str << "MSG: -> " << packetp->mHost << "\tABORTING RELIABLE:\t" + << packetp->mPacketID; + LL_INFOS() << str.str() << LL_ENDL; + } + + if (packetp->mCallback) + { + packetp->mCallback(packetp->mCallbackData,LL_ERR_TCP_TIMEOUT); + } + + // Update stats + mUnackedPacketCount--; + mUnackedPacketBytes -= packetp->mBufferLength; + + mFinalRetryPackets.erase(iter++); + delete packetp; + } + else + { + ++iter; + } + } + + return mUnackedPacketCount; +} + + +LLCircuit::LLCircuit(const F32Seconds circuit_heartbeat_interval, const F32Seconds circuit_timeout) +: mLastCircuit(NULL), + mHeartbeatInterval(circuit_heartbeat_interval), + mHeartbeatTimeout(circuit_timeout) +{} + +LLCircuit::~LLCircuit() +{ + // delete pointers in the map. + std::for_each(mCircuitData.begin(), + mCircuitData.end(), + llcompose1( + DeletePointerFunctor<LLCircuitData>(), + llselect2nd<circuit_data_map::value_type>())); +} + +LLCircuitData *LLCircuit::addCircuitData(const LLHost &host, TPACKETID in_id) +{ + // This should really validate if one already exists + LL_INFOS() << "LLCircuit::addCircuitData for " << host << LL_ENDL; + LLCircuitData *tempp = new LLCircuitData(host, in_id, mHeartbeatInterval, mHeartbeatTimeout); + mCircuitData.insert(circuit_data_map::value_type(host, tempp)); + mPingSet.insert(tempp); + + mLastCircuit = tempp; + return tempp; +} + +void LLCircuit::removeCircuitData(const LLHost &host) +{ + LL_INFOS() << "LLCircuit::removeCircuitData for " << host << LL_ENDL; + mLastCircuit = NULL; + circuit_data_map::iterator it = mCircuitData.find(host); + if(it != mCircuitData.end()) + { + LLCircuitData *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); + 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; +} + +void LLCircuitData::setAlive(bool b_alive) +{ + if (mbAlive != b_alive) + { + mPacketsOutID = 0; + mPacketsInID = 0; + mbAlive = b_alive; + } + if (b_alive) + { + mLastPingReceivedTime = LLMessageSystem::getMessageTimeSeconds(); + mPingsInTransit = 0; + mBlocked = false; + } +} + + +void LLCircuitData::setAllowTimeout(bool allow) +{ + mbAllowTimeout = allow; + + if (allow) + { + // resuming circuit + // make sure it's alive + setAlive(true); + } +} + + +// Reset per-period counters if necessary. +void LLCircuitData::checkPeriodTime() +{ + F64Seconds mt_sec = LLMessageSystem::getMessageTimeSeconds(); + F64Seconds period_length = mt_sec - mPeriodTime; + if ( period_length > TARGET_PERIOD_LENGTH) + { + F32 bps_in = F32Bits(mBytesInThisPeriod).value() / period_length.value(); + if (bps_in > mPeakBPSIn) + { + mPeakBPSIn = bps_in; + } + + F32 bps_out = F32Bits(mBytesOutThisPeriod).value() / period_length.value(); + if (bps_out > mPeakBPSOut) + { + mPeakBPSOut = bps_out; + } + + mBytesInLastPeriod = mBytesInThisPeriod; + mBytesOutLastPeriod = mBytesOutThisPeriod; + mBytesInThisPeriod = S32Bytes(0); + mBytesOutThisPeriod = S32Bytes(0); + mLastPeriodLength = F32Seconds::convert(period_length); + + mPeriodTime = mt_sec; + } +} + + +void LLCircuitData::addBytesIn(S32Bytes bytes) +{ + mBytesIn += bytes; + mBytesInThisPeriod += bytes; +} + + +void LLCircuitData::addBytesOut(S32Bytes bytes) +{ + mBytesOut += bytes; + mBytesOutThisPeriod += bytes; +} + + +void LLCircuitData::addReliablePacket(S32 mSocket, U8 *buf_ptr, S32 buf_len, LLReliablePacketParams *params) +{ + LLReliablePacket *packet_info; + + packet_info = new LLReliablePacket(mSocket, buf_ptr, buf_len, params); + + mUnackedPacketCount++; + mUnackedPacketBytes += packet_info->mBufferLength; + + if (params && params->mRetries) + { + mUnackedPackets[packet_info->mPacketID] = packet_info; + } + else + { + mFinalRetryPackets[packet_info->mPacketID] = packet_info; + } +} + + +void LLCircuit::resendUnackedPackets(S32& unacked_list_length, S32& unacked_list_size) +{ + F64Seconds now = LLMessageSystem::getMessageTimeSeconds(); + unacked_list_length = 0; + unacked_list_size = 0; + + LLCircuitData* circ; + circuit_data_map::iterator end = mUnackedCircuitMap.end(); + for(circuit_data_map::iterator it = mUnackedCircuitMap.begin(); it != end; ++it) + { + circ = (*it).second; + unacked_list_length += circ->resendUnackedPackets(now); + unacked_list_size += circ->getUnackedPacketBytes(); + } +} + + +bool LLCircuitData::isDuplicateResend(TPACKETID packetnum) +{ + return (mRecentlyReceivedReliablePackets.find(packetnum) != mRecentlyReceivedReliablePackets.end()); +} + + +void LLCircuit::dumpResends() +{ + circuit_data_map::iterator end = mCircuitData.end(); + for(circuit_data_map::iterator it = mCircuitData.begin(); it != end; ++it) + { + (*it).second->dumpResendCountAndReset(); + } +} + +LLCircuitData* LLCircuit::findCircuit(const LLHost& host) const +{ + // An optimization on finding the previously found circuit. + if (mLastCircuit && (mLastCircuit->mHost == host)) + { + return mLastCircuit; + } + + circuit_data_map::const_iterator it = mCircuitData.find(host); + if(it == mCircuitData.end()) + { + return NULL; + } + mLastCircuit = it->second; + return mLastCircuit; +} + + +bool LLCircuit::isCircuitAlive(const LLHost& host) const +{ + LLCircuitData *cdp = findCircuit(host); + if(cdp) + { + return cdp->mbAlive; + } + + return false; +} + +void LLCircuitData::setTimeoutCallback(void (*callback_func)(const LLHost &host, void *user_data), void *user_data) +{ + mTimeoutCallback = callback_func; + mTimeoutUserData = user_data; +} + +void LLCircuitData::checkPacketInID(TPACKETID id, bool receive_resent) +{ + // Done as floats so we don't have to worry about running out of room + // with U32 getting poked into an S32. + F32 delta = (F32)mHighestPacketID - (F32)id; + if (delta > (0.5f*LL_MAX_OUT_PACKET_ID)) + { + // We've almost definitely wrapped, reset the mLastPacketID to be low again. + mHighestPacketID = id; + } + else if (delta < (-0.5f*LL_MAX_OUT_PACKET_ID)) + { + // This is almost definitely an old packet coming in after a wrap, ignore it. + } + else + { + mHighestPacketID = llmax(mHighestPacketID, id); + } + + // Save packet arrival time + mLastPacketInTime = LLMessageSystem::getMessageTimeSeconds(); + + // Have we received anything on this circuit yet? + if (0 == mPacketsIn) + { + // Must be first packet from unclosed circuit. + mPacketsIn++; + setPacketInID((id + 1) % LL_MAX_OUT_PACKET_ID); + + mLastPacketGap = 0; + return; + } + + mPacketsIn++; + + + // now, check to see if we've got a gap + U32 gap = 0; + if (mPacketsInID == id) + { + // nope! bump and wrap the counter, then return + mPacketsInID++; + mPacketsInID = (mPacketsInID) % LL_MAX_OUT_PACKET_ID; + } + else if (id < mWrapID) + { + // id < mWrapID will happen if the first few packets are out of order. . . + // at that point we haven't marked anything "potentially lost" and + // the out-of-order packet will cause a full wrap marking all the IDs "potentially lost" + + // do nothing + } + else + { + // we have a gap! if that id is in the map, remove it from the map, leave mCurrentCircuit->mPacketsInID + // alone + // otherwise, walk from mCurrentCircuit->mPacketsInID to id with wrapping, adding the values to the map + // and setting mPacketsInID to id + 1 % LL_MAX_OUT_PACKET_ID + + // babbage: all operands in expression are unsigned, so modular + // arithmetic will always find correct gap, regardless of wrap arounds. + const U8 width = 24; + gap = LLModularMath::subtract<width>(mPacketsInID, id); + + if (mPotentialLostPackets.find(id) != mPotentialLostPackets.end()) + { + if(gMessageSystem->mVerboseLog) + { + std::ostringstream str; + str << "MSG: <- " << mHost << "\tRECOVERING LOST:\t" << id; + LL_INFOS() << str.str() << LL_ENDL; + } + // LL_INFOS() << "removing potential lost: " << id << LL_ENDL; + mPotentialLostPackets.erase(id); + } + else if (!receive_resent) // don't freak out over out-of-order reliable resends + { + U64Microseconds time = LLMessageSystem::getMessageTimeUsecs(); + TPACKETID index = mPacketsInID; + S32 gap_count = 0; + if ((index < id) && ((id - index) < 16)) + { + while (index != id) + { + if(gMessageSystem->mVerboseLog) + { + std::ostringstream str; + str << "MSG: <- " << mHost << "\tPACKET GAP:\t" + << index; + LL_INFOS() << str.str() << LL_ENDL; + } + +// LL_INFOS() << "adding potential lost: " << index << LL_ENDL; + mPotentialLostPackets[index] = time; + index++; + index = index % LL_MAX_OUT_PACKET_ID; + gap_count++; + } + } + else + { + LL_INFOS() << "packet_out_of_order - got packet " << id << " expecting " << index << " from " << mHost << LL_ENDL; + if(gMessageSystem->mVerboseLog) + { + std::ostringstream str; + str << "MSG: <- " << mHost << "\tPACKET GAP:\t" + << id << " expected " << index; + LL_INFOS() << str.str() << LL_ENDL; + } + } + + mPacketsInID = id + 1; + mPacketsInID = (mPacketsInID) % LL_MAX_OUT_PACKET_ID; + + if (gap_count > 128) + { + LL_WARNS() << "Packet loss gap filler running amok!" << LL_ENDL; + } + else if (gap_count > 16) + { + LL_WARNS() << "Sustaining large amounts of packet loss!" << LL_ENDL; + } + + } + } + mLastPacketGap = gap; +} + + +void LLCircuit::updateWatchDogTimers(LLMessageSystem *msgsys) +{ + F64Seconds cur_time = LLMessageSystem::getMessageTimeSeconds(); + S32 count = mPingSet.size(); + S32 cur = 0; + + // Only process each circuit once at most, stop processing if no circuits + while((cur < count) && !mPingSet.empty()) + { + cur++; + + LLCircuit::ping_set_t::iterator psit = mPingSet.begin(); + LLCircuitData *cdp = *psit; + + if (!cdp->mbAlive) + { + // We suspect that this case should never happen, given how + // the alive status is set. + // Skip over dead circuits, just add the ping interval and push it to the back + // Always remember to remove it from the set before changing the sorting + // key (mNextPingSendTime) + mPingSet.erase(psit); + cdp->mNextPingSendTime = cur_time + mHeartbeatInterval; + mPingSet.insert(cdp); + continue; + } + else + { + // Check to see if this needs a ping + if (cur_time < cdp->mNextPingSendTime) + { + // This circuit doesn't need a ping, break out because + // we have a sorted list, thus no more circuits need pings + break; + } + + // Update watchdog timers + if (cdp->updateWatchDogTimers(msgsys)) + { + // Randomize our pings a bit by doing some up to 5% early or late + F64Seconds dt = 0.95f*mHeartbeatInterval + F32Seconds(ll_frand(0.1f*mHeartbeatInterval.value())); + + // Remove it, and reinsert it with the new next ping time. + // Always remove before changing the sorting key. + mPingSet.erase(psit); + cdp->mNextPingSendTime = cur_time + dt; + mPingSet.insert(cdp); + + // Update our throttles + cdp->mThrottles.dynamicAdjust(); + + // Update some stats, this is not terribly important + cdp->checkPeriodTime(); + } + else + { + // This mPingSet.erase isn't necessary, because removing the circuit will + // remove the ping set. + //mPingSet.erase(psit); + removeCircuitData(cdp->mHost); + } + } + } +} + + +bool LLCircuitData::updateWatchDogTimers(LLMessageSystem *msgsys) +{ + F64Seconds cur_time = LLMessageSystem::getMessageTimeSeconds(); + mLastPingSendTime = cur_time; + + if (!checkCircuitTimeout()) + { + // Pass this back to the calling LLCircuit, this circuit needs to be cleaned up. + return false; + } + + // WARNING! + // Duplicate suppression can FAIL if packets are delivered out of + // order, although it's EXTREMELY unlikely. It would require + // that the ping get delivered out of order enough that the ACK + // for the packet that it was out of order with was received BEFORE + // the ping was sent. + + // Find the current oldest reliable packetID + // This is to handle the case if we actually manage to wrap our + // packet IDs - the oldest will actually have a higher packet ID + // than the current. + bool wrapped = false; + reliable_iter iter; + iter = mUnackedPackets.upper_bound(getPacketOutID()); + if (iter == mUnackedPackets.end()) + { + // Nothing AFTER this one, so we want the lowest packet ID + // then. + iter = mUnackedPackets.begin(); + wrapped = true; + } + + TPACKETID packet_id = 0; + + // Check against the "final" packets + bool wrapped_final = false; + reliable_iter iter_final; + iter_final = mFinalRetryPackets.upper_bound(getPacketOutID()); + if (iter_final == mFinalRetryPackets.end()) + { + iter_final = mFinalRetryPackets.begin(); + wrapped_final = true; + } + + //LL_INFOS() << mHost << " - unacked count " << mUnackedPackets.size() << LL_ENDL; + //LL_INFOS() << mHost << " - final count " << mFinalRetryPackets.size() << LL_ENDL; + if (wrapped != wrapped_final) + { + // One of the "unacked" or "final" lists hasn't wrapped. Whichever one + // hasn't has the oldest packet. + if (!wrapped) + { + // Hasn't wrapped, so the one on the + // unacked packet list is older + packet_id = iter->first; + //LL_INFOS() << mHost << ": nowrapped unacked" << LL_ENDL; + } + else + { + packet_id = iter_final->first; + //LL_INFOS() << mHost << ": nowrapped final" << LL_ENDL; + } + } + else + { + // They both wrapped, we can just use the minimum of the two. + if ((iter == mUnackedPackets.end()) && (iter_final == mFinalRetryPackets.end())) + { + // Wow! No unacked packets at all! + // Send the ID of the last packet we sent out. + // This will flush all of the destination's + // unacked packets, theoretically. + //LL_INFOS() << mHost << ": No unacked!" << LL_ENDL; + packet_id = getPacketOutID(); + } + else + { + bool had_unacked = false; + if (iter != mUnackedPackets.end()) + { + // Unacked list has the lowest so far + packet_id = iter->first; + had_unacked = true; + //LL_INFOS() << mHost << ": Unacked" << LL_ENDL; + } + + if (iter_final != mFinalRetryPackets.end()) + { + // Use the lowest of the unacked list and the final list + if (had_unacked) + { + // Both had a packet, use the lowest. + packet_id = llmin(packet_id, iter_final->first); + //LL_INFOS() << mHost << ": Min of unacked/final" << LL_ENDL; + } + else + { + // Only the final had a packet, use it. + packet_id = iter_final->first; + //LL_INFOS() << mHost << ": Final!" << LL_ENDL; + } + } + } + } + + // Send off the another ping. + pingTimerStart(); + msgsys->newMessageFast(_PREHASH_StartPingCheck); + msgsys->nextBlock(_PREHASH_PingID); + msgsys->addU8Fast(_PREHASH_PingID, nextPingID()); + msgsys->addU32Fast(_PREHASH_OldestUnacked, packet_id); + msgsys->sendMessage(mHost); + + // Also do lost packet accounting. + // Check to see if anything on our lost list is old enough to + // be considered lost + + LLCircuitData::packet_time_map::iterator it; + U64Microseconds timeout = llmin(LL_MAX_LOST_TIMEOUT, F32Seconds(getPingDelayAveraged()) * LL_LOST_TIMEOUT_FACTOR); + + U64Microseconds mt_usec = LLMessageSystem::getMessageTimeUsecs(); + for (it = mPotentialLostPackets.begin(); it != mPotentialLostPackets.end(); ) + { + U64Microseconds delta_t_usec = mt_usec - (*it).second; + if (delta_t_usec > timeout) + { + // let's call this one a loss! + mPacketsLost++; + gMessageSystem->mDroppedPackets++; + if(gMessageSystem->mVerboseLog) + { + std::ostringstream str; + str << "MSG: <- " << mHost << "\tLOST PACKET:\t" + << (*it).first; + LL_INFOS() << str.str() << LL_ENDL; + } + mPotentialLostPackets.erase(it++); + } + else + { + ++it; + } + } + + return true; +} + + +void LLCircuitData::clearDuplicateList(TPACKETID oldest_id) +{ + // purge old data from the duplicate suppression queue + + // we want to KEEP all x where oldest_id <= x <= last incoming packet, and delete everything else. + + //LL_INFOS() << mHost << ": clearing before oldest " << oldest_id << LL_ENDL; + //LL_INFOS() << "Recent list before: " << mRecentlyReceivedReliablePackets.size() << LL_ENDL; + if (oldest_id < mHighestPacketID) + { + // Clean up everything with a packet ID less than oldest_id. + packet_time_map::iterator pit_start; + packet_time_map::iterator pit_end; + pit_start = mRecentlyReceivedReliablePackets.begin(); + pit_end = mRecentlyReceivedReliablePackets.lower_bound(oldest_id); + mRecentlyReceivedReliablePackets.erase(pit_start, pit_end); + } + + // Do timeout checks on everything with an ID > mHighestPacketID. + // This should be empty except for wrapping IDs. Thus, this should be + // highly rare. + U64Microseconds mt_usec = LLMessageSystem::getMessageTimeUsecs(); + + packet_time_map::iterator pit; + for(pit = mRecentlyReceivedReliablePackets.upper_bound(mHighestPacketID); + pit != mRecentlyReceivedReliablePackets.end(); ) + { + // Validate that the packet ID seems far enough away + if ((pit->first - mHighestPacketID) < 100) + { + LL_WARNS() << "Probably incorrectly timing out non-wrapped packets!" << LL_ENDL; + } + U64Microseconds delta_t_usec = mt_usec - (*pit).second; + F64Seconds delta_t_sec = delta_t_usec; + if (delta_t_sec > LL_DUPLICATE_SUPPRESSION_TIMEOUT) + { + // enough time has elapsed we're not likely to get a duplicate on this one + LL_INFOS() << "Clearing " << pit->first << " from recent list" << LL_ENDL; + mRecentlyReceivedReliablePackets.erase(pit++); + } + else + { + ++pit; + } + } + //LL_INFOS() << "Recent list after: " << mRecentlyReceivedReliablePackets.size() << LL_ENDL; +} + +bool LLCircuitData::checkCircuitTimeout() +{ + F64Seconds time_since_last_ping = LLMessageSystem::getMessageTimeSeconds() - mLastPingReceivedTime; + + // Nota Bene: This needs to be turned off if you are debugging multiple simulators + if (time_since_last_ping > mHeartbeatTimeout) + { + LL_WARNS() << "LLCircuitData::checkCircuitTimeout for " << mHost << " last ping " << time_since_last_ping << " seconds ago." <<LL_ENDL; + setAlive(false); + if (mTimeoutCallback) + { + LL_WARNS() << "LLCircuitData::checkCircuitTimeout for " << mHost << " calling callback." << LL_ENDL; + mTimeoutCallback(mHost, mTimeoutUserData); + } + if (!isAlive()) + { + // The callback didn't try and resurrect the circuit. We should kill it. + LL_WARNS() << "LLCircuitData::checkCircuitTimeout for " << mHost << " still dead, dropping." << LL_ENDL; + return false; + } + } + + return true; +} + +// Call this method when a reliable message comes in - this will +// correctly place the packet in the correct list to be acked later. +bool LLCircuitData::collectRAck(TPACKETID packet_num) +{ + if (mAcks.empty()) + { + // 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); + if (mAckCreationTime == 0) + { + mAckCreationTime = getAgeInSeconds(); + } + return true; +} + +// this method is called during the message system processAcks() to +// send out any acks that did not get sent already. +void LLCircuit::sendAcks(F32 collect_time) +{ + collect_time = llclamp(collect_time, 0.f, LL_COLLECT_ACK_TIME_MAX); + LLCircuitData* cd; + circuit_data_map::iterator it = mSendAckMap.begin(); + while (it != mSendAckMap.end()) + { + circuit_data_map::iterator cur_it = it++; + cd = (*cur_it).second; + S32 count = (S32)cd->mAcks.size(); + F32 age = cd->getAgeInSeconds() - cd->mAckCreationTime; + if (age > collect_time || count == 0) + { + if (count>0) + { + // send the packet acks + S32 acks_this_packet = 0; + for(S32 i = 0; i < count; ++i) + { + if(acks_this_packet == 0) + { + gMessageSystem->newMessageFast(_PREHASH_PacketAck); + } + gMessageSystem->nextBlockFast(_PREHASH_Packets); + gMessageSystem->addU32Fast(_PREHASH_ID, cd->mAcks[i]); + ++acks_this_packet; + if(acks_this_packet > 250) + { + gMessageSystem->sendMessage(cd->mHost); + acks_this_packet = 0; + } + } + if(acks_this_packet > 0) + { + gMessageSystem->sendMessage(cd->mHost); + } + + if(gMessageSystem->mVerboseLog) + { + std::ostringstream str; + str << "MSG: -> " << cd->mHost << "\tPACKET ACKS:\t"; + std::ostream_iterator<TPACKETID> append(str, " "); + std::copy(cd->mAcks.begin(), cd->mAcks.end(), append); + LL_INFOS() << str.str() << LL_ENDL; + } + + // empty out the acks list + cd->mAcks.clear(); + cd->mAckCreationTime = 0.f; + } + // remove data map + mSendAckMap.erase(cur_it); + } + } +} + + +std::ostream& operator<<(std::ostream& s, LLCircuitData& circuit) +{ + F32 age = circuit.mExistenceTimer.getElapsedTimeF32(); + + using namespace std; + s << "Circuit " << circuit.mHost << " " + << circuit.mRemoteID << " " + << (circuit.mbAlive ? "Alive" : "Not Alive") << " " + << (circuit.mbAllowTimeout ? "Timeout Allowed" : "Timeout Not Allowed") + << endl; + + s << " Packets Lost: " << circuit.mPacketsLost + << " Measured Ping: " << circuit.mPingDelay + << " Averaged Ping: " << circuit.mPingDelayAveraged + << endl; + + s << "Global In/Out " << S32(age) << " sec" + << " KBytes: " << circuit.mBytesIn.valueInUnits<LLUnits::Kilobytes>() << "/" << circuit.mBytesOut.valueInUnits<LLUnits::Kilobytes>() + << " Kbps: " + << S32(circuit.mBytesIn.valueInUnits<LLUnits::Kilobits>() / circuit.mExistenceTimer.getElapsedTimeF32().value()) + << "/" + << S32(circuit.mBytesOut.valueInUnits<LLUnits::Kilobits>() / circuit.mExistenceTimer.getElapsedTimeF32().value()) + << " Packets: " << circuit.mPacketsIn << "/" << circuit.mPacketsOut + << endl; + + s << "Recent In/Out " << circuit.mLastPeriodLength + << " KBytes: " + << circuit.mBytesInLastPeriod.valueInUnits<LLUnits::Kilobytes>() + << "/" + << circuit.mBytesOutLastPeriod.valueInUnits<LLUnits::Kilobytes>() + << " Kbps: " + << (S32)(circuit.mBytesInLastPeriod.valueInUnits<LLUnits::Kilobits>() / circuit.mLastPeriodLength.value()) + << "/" + << (S32)(circuit.mBytesOutLastPeriod.valueInUnits<LLUnits::Kilobits>() / circuit.mLastPeriodLength.value()) + << " Peak kbps: " + << S32(circuit.mPeakBPSIn / 1024.f) + << "/" + << S32(circuit.mPeakBPSOut / 1024.f) + << endl; + + return s; +} + +void LLCircuitData::getInfo(LLSD& info) const +{ + info["Host"] = mHost.getIPandPort(); + info["Alive"] = mbAlive; + info["Age"] = mExistenceTimer.getElapsedTimeF32(); +} + +void LLCircuitData::dumpResendCountAndReset() +{ + if (mCurrentResendCount) + { + LL_INFOS() << "Circuit: " << mHost << " resent " << mCurrentResendCount << " packets" << LL_ENDL; + mCurrentResendCount = 0; + } +} + +std::ostream& operator<<(std::ostream& s, LLCircuit &circuit) +{ + s << "Circuit Info:" << std::endl; + LLCircuit::circuit_data_map::iterator end = circuit.mCircuitData.end(); + LLCircuit::circuit_data_map::iterator it; + for(it = circuit.mCircuitData.begin(); it != end; ++it) + { + s << *((*it).second) << std::endl; + } + return s; +} + +void LLCircuit::getInfo(LLSD& info) const +{ + LLCircuit::circuit_data_map::const_iterator end = mCircuitData.end(); + LLCircuit::circuit_data_map::const_iterator it; + LLSD circuit_info; + for(it = mCircuitData.begin(); it != end; ++it) + { + (*it).second->getInfo(circuit_info); + info["Circuits"].append(circuit_info); + } +} + +void LLCircuit::getCircuitRange( + const LLHost& key, + LLCircuit::circuit_data_map::iterator& first, + LLCircuit::circuit_data_map::iterator& end) +{ + end = mCircuitData.end(); + first = mCircuitData.upper_bound(key); +} + +TPACKETID LLCircuitData::nextPacketOutID() +{ + mPacketsOut++; + + TPACKETID id; + + id = (mPacketsOutID + 1) % LL_MAX_OUT_PACKET_ID; + + if (id < mPacketsOutID) + { + // we just wrapped on a circuit, reset the wrap ID to zero + mWrapID = 0; + } + mPacketsOutID = id; + return id; +} + + +void LLCircuitData::setPacketInID(TPACKETID id) +{ + id = id % LL_MAX_OUT_PACKET_ID; + mPacketsInID = id; + mRecentlyReceivedReliablePackets.clear(); + + mWrapID = id; +} + + +void LLCircuitData::pingTimerStop(const U8 ping_id) +{ + F64Seconds mt_secs = LLMessageSystem::getMessageTimeSeconds(); + + // Nota Bene: no averaging of ping times until we get a feel for how this works + F64Seconds time = mt_secs - mPingTime; + if (time == F32Seconds(0.0)) + { + // Ack, we got our ping response on the same frame! Sigh, let's get a real time otherwise + // all of our ping calculations will be skewed. + mt_secs = LLMessageSystem::getMessageTimeSeconds(true); + } + mLastPingReceivedTime = mt_secs; + + // If ping is longer than 1 second, we'll get sequence deltas in the ping. + // Approximate by assuming each ping counts for 1 second (slightly low, probably) + S32 delta_ping = (S32)mLastPingID - (S32) ping_id; + if (delta_ping < 0) + { + delta_ping += 256; + } + + U32Milliseconds msec = delta_ping*mHeartbeatInterval + time; + setPingDelay(msec); + + mPingsInTransit = delta_ping; + if (mBlocked && (mPingsInTransit <= PING_RELEASE_BLOCK)) + { + mBlocked = false; + } +} + + +void LLCircuitData::pingTimerStart() +{ + mPingTime = LLMessageSystem::getMessageTimeSeconds(); + mPingsInTransit++; + + if (!mBlocked && (mPingsInTransit > PING_START_BLOCK)) + { + mBlocked = true; + } +} + + +U32 LLCircuitData::getPacketsIn() const +{ + return mPacketsIn; +} + + +S32Bytes LLCircuitData::getBytesIn() const +{ + return mBytesIn; +} + + +S32Bytes LLCircuitData::getBytesOut() const +{ + return mBytesOut; +} + + +U32 LLCircuitData::getPacketsOut() const +{ + return mPacketsOut; +} + + +TPACKETID LLCircuitData::getPacketOutID() const +{ + return mPacketsOutID; +} + + +U32 LLCircuitData::getPacketsLost() const +{ + return mPacketsLost; +} + + +bool LLCircuitData::isAlive() const +{ + return mbAlive; +} + + +bool LLCircuitData::isBlocked() const +{ + return mBlocked; +} + + +bool LLCircuitData::getAllowTimeout() const +{ + return mbAllowTimeout; +} + + +U32Milliseconds LLCircuitData::getPingDelay() const +{ + return mPingDelay; +} + + +F32Milliseconds LLCircuitData::getPingInTransitTime() +{ + // This may be inaccurate in the case of a circuit that was "dead" and then revived, + // but only until the first round trip ping is sent - djs + F32Milliseconds time_since_ping_was_sent(0); + + if (mPingsInTransit) + { + time_since_ping_was_sent = F32Milliseconds::convert(((mPingsInTransit*mHeartbeatInterval - F32Seconds(1)) + + (LLMessageSystem::getMessageTimeSeconds() - mPingTime))); + } + + return time_since_ping_was_sent; +} + + +void LLCircuitData::setPingDelay(U32Milliseconds ping) +{ + mPingDelay = ping; + mPingDelayAveraged = llmax((F32Milliseconds)ping, getPingDelayAveraged()); + mPingDelayAveraged = ((1.f - LL_AVERAGED_PING_ALPHA) * mPingDelayAveraged) + + (LL_AVERAGED_PING_ALPHA * (F32Milliseconds) ping); + mPingDelayAveraged = llclamp(mPingDelayAveraged, + LL_AVERAGED_PING_MIN, + LL_AVERAGED_PING_MAX); +} + + +F32Milliseconds LLCircuitData::getPingDelayAveraged() +{ + return llmin(llmax(getPingInTransitTime(), mPingDelayAveraged), LL_AVERAGED_PING_MAX); +} + + +bool LLCircuitData::getTrusted() const +{ + return mTrusted; +} + + +void LLCircuitData::setTrusted(bool t) +{ + mTrusted = t; +} + +F32 LLCircuitData::getAgeInSeconds() const +{ + return mExistenceTimer.getElapsedTimeF32(); +} |