/*
* Copyright (c) Contributors, http://opensimulator.org/
* See CONTRIBUTORS.TXT for a full list of copyright holders.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of the OpenSimulator Project nor the
* names of its contributors may be used to endorse or promote products
* derived from this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE DEVELOPERS ``AS IS'' AND ANY
* EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
* DISCLAIMED. IN NO EVENT SHALL THE CONTRIBUTORS BE LIABLE FOR ANY
* DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
* LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
using System;
using System.Collections.Generic;
using System.Net;
using System.Threading;
using log4net;
using OpenSim.Framework;
using OpenSim.Framework.Monitoring;
using OpenMetaverse;
using OpenMetaverse.Packets;
using TokenBucket = OpenSim.Region.ClientStack.LindenUDP.TokenBucket;
namespace OpenSim.Region.ClientStack.LindenUDP
{
#region Delegates
///
/// Fired when updated networking stats are produced for this client
///
/// Number of incoming packets received since this
/// event was last fired
/// Number of outgoing packets sent since this
/// event was last fired
/// Current total number of bytes in packets we
/// are waiting on ACKs for
public delegate void PacketStats(int inPackets, int outPackets, int unAckedBytes);
///
/// Fired when the queue for one or more packet categories is empty. This
/// event can be hooked to put more data on the empty queues
///
/// Categories of the packet queues that are empty
public delegate void QueueEmpty(ThrottleOutPacketTypeFlags categories);
#endregion Delegates
///
/// Tracks state for a client UDP connection and provides client-specific methods
///
public sealed class LLUDPClient
{
// TODO: Make this a config setting
/// Percentage of the task throttle category that is allocated to avatar and prim
/// state updates
const float STATE_TASK_PERCENTAGE = 0.8f;
private static readonly ILog m_log = LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);
/// The number of packet categories to throttle on. If a throttle category is added
/// or removed, this number must also change
const int THROTTLE_CATEGORY_COUNT = 8;
///
/// Controls whether information is logged about each outbound packet immediately before it is sent. For debug purposes.
///
/// Any level above 0 will turn on logging.
public int DebugDataOutLevel { get; set; }
///
/// Controls whether information is logged about each outbound packet immediately before it is sent. For debug purposes.
///
/// Any level above 0 will turn on logging.
public int ThrottleDebugLevel
{
get
{
return m_throttleDebugLevel;
}
set
{
m_throttleDebugLevel = value;
m_throttleClient.DebugLevel = m_throttleDebugLevel;
foreach (TokenBucket tb in m_throttleCategories)
tb.DebugLevel = m_throttleDebugLevel;
}
}
private int m_throttleDebugLevel;
/// Fired when updated networking stats are produced for this client
public event PacketStats OnPacketStats;
/// Fired when the queue for a packet category is empty. This event can be
/// hooked to put more data on the empty queue
public event QueueEmpty OnQueueEmpty;
public event Func HasUpdates;
/// AgentID for this client
public readonly UUID AgentID;
/// The remote address of the connected client
public readonly IPEndPoint RemoteEndPoint;
/// Circuit code that this client is connected on
public readonly uint CircuitCode;
/// Sequence numbers of packets we've received (for duplicate checking)
public readonly IncomingPacketHistoryCollection PacketArchive = new IncomingPacketHistoryCollection(200);
///
/// If true then we take action in response to unacked reliably sent packets such as resending the packet.
///
public bool ProcessUnackedSends { get; set; }
/// Packets we have sent that need to be ACKed by the client
public readonly UnackedPacketCollection NeedAcks = new UnackedPacketCollection();
/// ACKs that are queued up, waiting to be sent to the client
public readonly DoubleLocklessQueue PendingAcks = new DoubleLocklessQueue();
/// Current packet sequence number
public int CurrentSequence;
/// Current ping sequence number
public byte CurrentPingSequence;
/// True when this connection is alive, otherwise false
public bool IsConnected = true;
/// True when this connection is paused, otherwise false
public bool IsPaused;
/// Environment.TickCount when the last packet was received for this client
public int TickLastPacketReceived;
/// Smoothed round-trip time. A smoothed average of the round-trip time for sending a
/// reliable packet to the client and receiving an ACK
public float SRTT;
/// Round-trip time variance. Measures the consistency of round-trip times
public float RTTVAR;
/// Retransmission timeout. Packets that have not been acknowledged in this number of
/// milliseconds or longer will be resent
/// Calculated from and using the
/// guidelines in RFC 2988
public int RTO;
/// Number of bytes received since the last acknowledgement was sent out. This is used
/// to loosely follow the TCP delayed ACK algorithm in RFC 1122 (4.2.3.2)
public int BytesSinceLastACK;
/// Number of packets received from this client
public int PacketsReceived;
/// Number of packets sent to this client
public int PacketsSent;
/// Number of packets resent to this client
public int PacketsResent;
/// Total byte count of unacked packets sent to this client
public int UnackedBytes;
/// Total number of received packets that we have reported to the OnPacketStats event(s)
private int m_packetsReceivedReported;
/// Total number of sent packets that we have reported to the OnPacketStats event(s)
private int m_packetsSentReported;
/// Holds the Environment.TickCount value of when the next OnQueueEmpty can be fired
private int m_nextOnQueueEmpty = 1;
/// Throttle bucket for this agent's connection
private readonly AdaptiveTokenBucket m_throttleClient;
public AdaptiveTokenBucket FlowThrottle
{
get { return m_throttleClient; }
}
/// Throttle buckets for each packet category
private readonly TokenBucket[] m_throttleCategories;
/// Outgoing queues for throttled packets
private readonly DoubleLocklessQueue[] m_packetOutboxes = new DoubleLocklessQueue[THROTTLE_CATEGORY_COUNT];
/// A container that can hold one packet for each outbox, used to store
/// dequeued packets that are being held for throttling
private readonly OutgoingPacket[] m_nextPackets = new OutgoingPacket[THROTTLE_CATEGORY_COUNT];
/// A reference to the LLUDPServer that is managing this client
private readonly LLUDPServer m_udpServer;
/// Caches packed throttle information
private byte[] m_packedThrottles;
private int m_defaultRTO = 1000; // 1sec is the recommendation in the RFC
private int m_maxRTO = 60000;
public bool m_deliverPackets = true;
private float m_burstTime;
public int m_lastStartpingTimeMS;
public int m_pingMS;
public int PingTimeMS
{
get
{
if (m_pingMS < 10)
return 10;
if(m_pingMS > 2000)
return 2000;
return m_pingMS;
}
}
///
/// This is the percentage of the udp texture queue to add to the task queue since
/// textures are now generally handled through http.
///
private double m_cannibalrate = 0.0;
private ClientInfo m_info = new ClientInfo();
///
/// Default constructor
///
/// Reference to the UDP server this client is connected to
/// Default throttling rates and maximum throttle limits
/// Parent HTB (hierarchical token bucket)
/// that the child throttles will be governed by
/// Circuit code for this connection
/// AgentID for the connected agent
/// Remote endpoint for this connection
///
/// Default retransmission timeout for unacked packets. The RTO will never drop
/// beyond this number.
///
///
/// The maximum retransmission timeout for unacked packets. The RTO will never exceed this number.
///
public LLUDPClient(
LLUDPServer server, ThrottleRates rates, TokenBucket parentThrottle, uint circuitCode, UUID agentID,
IPEndPoint remoteEndPoint, int defaultRTO, int maxRTO)
{
AgentID = agentID;
RemoteEndPoint = remoteEndPoint;
CircuitCode = circuitCode;
m_udpServer = server;
if (defaultRTO != 0)
m_defaultRTO = defaultRTO;
if (maxRTO != 0)
m_maxRTO = maxRTO;
<<<<<<< HEAD
ProcessUnackedSends = true;
// Create a token bucket throttle for this client that has the scene token bucket as a parent
m_throttleClient
= new AdaptiveTokenBucket(
string.Format("adaptive throttle for {0} in {1}", AgentID, server.Scene.Name),
parentThrottle, 0, rates.Total, rates.MinimumAdaptiveThrottleRate, rates.AdaptiveThrottlesEnabled);
=======
m_burstTime = rates.BrustTime;
float m_burst = rates.ClientMaxRate * m_burstTime;
// Create a token bucket throttle for this client that has the scene token bucket as a parent
m_throttleClient = new AdaptiveTokenBucket(parentThrottle, rates.ClientMaxRate, m_burst, rates.AdaptiveThrottlesEnabled);
>>>>>>> avn/ubitvar
// Create an array of token buckets for this clients different throttle categories
m_throttleCategories = new TokenBucket[THROTTLE_CATEGORY_COUNT];
m_cannibalrate = rates.CannibalizeTextureRate;
m_burst = rates.Total * rates.BrustTime;
for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
{
ThrottleOutPacketType type = (ThrottleOutPacketType)i;
// Initialize the packet outboxes, where packets sit while they are waiting for tokens
<<<<<<< HEAD
m_packetOutboxes[i] = new OpenSim.Framework.LocklessQueue();
// Initialize the token buckets that control the throttling for each category
m_throttleCategories[i]
= new TokenBucket(
string.Format("{0} throttle for {1} in {2}", type, AgentID, server.Scene.Name),
m_throttleClient, rates.GetRate(type), 0);
=======
m_packetOutboxes[i] = new DoubleLocklessQueue();
// Initialize the token buckets that control the throttling for each category
m_throttleCategories[i] = new TokenBucket(m_throttleClient, rates.GetRate(type), m_burst);
>>>>>>> avn/ubitvar
}
// Default the retransmission timeout to one second
RTO = m_defaultRTO;
// Initialize this to a sane value to prevent early disconnects
TickLastPacketReceived = Environment.TickCount & Int32.MaxValue;
m_pingMS = (int)(3.0 * server.TickCountResolution); // so filter doesnt start at 0;
}
///
/// Shuts down this client connection
///
public void Shutdown()
{
IsConnected = false;
for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
{
m_packetOutboxes[i].Clear();
m_nextPackets[i] = null;
}
// pull the throttle out of the scene throttle
m_throttleClient.Parent.UnregisterRequest(m_throttleClient);
OnPacketStats = null;
OnQueueEmpty = null;
}
///
/// Gets information about this client connection
///
/// Information about the client connection
public ClientInfo GetClientInfo()
{
// TODO: This data structure is wrong in so many ways. Locking and copying the entire lists
// of pending and needed ACKs for every client every time some method wants information about
// this connection is a recipe for poor performance
m_info.resendThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Resend].DripRate;
m_info.landThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Land].DripRate;
m_info.windThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Wind].DripRate;
m_info.cloudThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Cloud].DripRate;
m_info.taskThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Task].DripRate;
m_info.assetThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Asset].DripRate;
m_info.textureThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Texture].DripRate;
m_info.totalThrottle = (int)m_throttleClient.DripRate;
<<<<<<< HEAD
m_info.targetThrottle = (int)m_throttleClient.TargetDripRate;
m_info.maxThrottle = (int)m_throttleClient.MaxDripRate;
=======
>>>>>>> avn/ubitvar
return m_info;
}
///
/// Modifies the UDP throttles
///
/// New throttling values
public void SetClientInfo(ClientInfo info)
{
// TODO: Allowing throttles to be manually set from this function seems like a reasonable
// idea. On the other hand, letting external code manipulate our ACK accounting is not
// going to happen
throw new NotImplementedException();
}
///
/// Get the total number of pakcets queued for this client.
///
///
public int GetTotalPacketsQueuedCount()
{
int total = 0;
for (int i = 0; i <= (int)ThrottleOutPacketType.Asset; i++)
total += m_packetOutboxes[i].Count;
return total;
}
///
/// Get the number of packets queued for the given throttle type.
///
///
///
public int GetPacketsQueuedCount(ThrottleOutPacketType throttleType)
{
if ((int)throttleType > 0)
return m_packetOutboxes[(int)throttleType].Count;
else
return 0;
}
///
/// Return statistics information about client packet queues.
///
///
/// FIXME: This should really be done in a more sensible manner rather than sending back a formatted string.
///
///
public string GetStats()
{
return string.Format(
"{0,7} {1,7} {2,7} {3,9} {4,7} {5,7} {6,7} {7,7} {8,7} {9,8} {10,7} {11,7}",
Util.EnvironmentTickCountSubtract(TickLastPacketReceived),
PacketsReceived,
PacketsSent,
PacketsResent,
UnackedBytes,
m_packetOutboxes[(int)ThrottleOutPacketType.Resend].Count,
m_packetOutboxes[(int)ThrottleOutPacketType.Land].Count,
m_packetOutboxes[(int)ThrottleOutPacketType.Wind].Count,
m_packetOutboxes[(int)ThrottleOutPacketType.Cloud].Count,
m_packetOutboxes[(int)ThrottleOutPacketType.Task].Count,
m_packetOutboxes[(int)ThrottleOutPacketType.Texture].Count,
m_packetOutboxes[(int)ThrottleOutPacketType.Asset].Count);
}
public void SendPacketStats()
{
PacketStats callback = OnPacketStats;
if (callback != null)
{
int newPacketsReceived = PacketsReceived - m_packetsReceivedReported;
int newPacketsSent = PacketsSent - m_packetsSentReported;
callback(newPacketsReceived, newPacketsSent, UnackedBytes);
m_packetsReceivedReported += newPacketsReceived;
m_packetsSentReported += newPacketsSent;
}
}
public void SetThrottles(byte[] throttleData)
{
SetThrottles(throttleData, 1.0f);
}
public void SetThrottles(byte[] throttleData, float factor)
{
byte[] adjData;
int pos = 0;
if (!BitConverter.IsLittleEndian)
{
byte[] newData = new byte[7 * 4];
Buffer.BlockCopy(throttleData, 0, newData, 0, 7 * 4);
for (int i = 0; i < 7; i++)
Array.Reverse(newData, i * 4, 4);
adjData = newData;
}
else
{
adjData = throttleData;
}
// 0.125f converts from bits to bytes
float scale = 0.125f * factor;
int resend = (int)(BitConverter.ToSingle(adjData, pos) * scale); pos += 4;
int land = (int)(BitConverter.ToSingle(adjData, pos) * scale); pos += 4;
int wind = (int)(BitConverter.ToSingle(adjData, pos) * scale); pos += 4;
int cloud = (int)(BitConverter.ToSingle(adjData, pos) * scale); pos += 4;
int task = (int)(BitConverter.ToSingle(adjData, pos) * scale); pos += 4;
int texture = (int)(BitConverter.ToSingle(adjData, pos) * scale); pos += 4;
int asset = (int)(BitConverter.ToSingle(adjData, pos) * scale);
if (ThrottleDebugLevel > 0)
{
long total = resend + land + wind + cloud + task + texture + asset;
m_log.DebugFormat(
"[LLUDPCLIENT]: {0} is setting throttles in {1} to Resend={2}, Land={3}, Wind={4}, Cloud={5}, Task={6}, Texture={7}, Asset={8}, TOTAL = {9}",
AgentID, m_udpServer.Scene.Name, resend, land, wind, cloud, task, texture, asset, total);
}
// Make sure none of the throttles are set below our packet MTU,
// otherwise a throttle could become permanently clogged
/* not using floats
resend = Math.Max(resend, LLUDPServer.MTU);
land = Math.Max(land, LLUDPServer.MTU);
wind = Math.Max(wind, LLUDPServer.MTU);
cloud = Math.Max(cloud, LLUDPServer.MTU);
task = Math.Max(task, LLUDPServer.MTU);
texture = Math.Max(texture, LLUDPServer.MTU);
asset = Math.Max(asset, LLUDPServer.MTU);
*/
// Since most textures are now delivered through http, make it possible
// to cannibalize some of the bw from the texture throttle to use for
// the task queue (e.g. object updates)
task = task + (int)(m_cannibalrate * texture);
texture = (int)((1 - m_cannibalrate) * texture);
<<<<<<< HEAD
//int total = resend + land + wind + cloud + task + texture + asset;
if (ThrottleDebugLevel > 0)
{
long total = resend + land + wind + cloud + task + texture + asset;
m_log.DebugFormat(
"[LLUDPCLIENT]: {0} is setting throttles in {1} to Resend={2}, Land={3}, Wind={4}, Cloud={5}, Task={6}, Texture={7}, Asset={8}, TOTAL = {9}",
AgentID, m_udpServer.Scene.Name, resend, land, wind, cloud, task, texture, asset, total);
}
=======
int total = resend + land + wind + cloud + task + texture + asset;
float m_burst = total * m_burstTime;
//m_log.DebugFormat("[LLUDPCLIENT]: {0} is setting throttles. Resend={1}, Land={2}, Wind={3}, Cloud={4}, Task={5}, Texture={6}, Asset={7}, Total={8}",
// AgentID, resend, land, wind, cloud, task, texture, asset, total);
>>>>>>> avn/ubitvar
// Update the token buckets with new throttle values
if (m_throttleClient.AdaptiveEnabled)
{
long total = resend + land + wind + cloud + task + texture + asset;
m_throttleClient.TargetDripRate = total;
}
TokenBucket bucket;
bucket = m_throttleCategories[(int)ThrottleOutPacketType.Resend];
bucket.RequestedDripRate = resend;
bucket.RequestedBurst = m_burst;
bucket = m_throttleCategories[(int)ThrottleOutPacketType.Land];
bucket.RequestedDripRate = land;
bucket.RequestedBurst = m_burst;
bucket = m_throttleCategories[(int)ThrottleOutPacketType.Wind];
bucket.RequestedDripRate = wind;
bucket.RequestedBurst = m_burst;
bucket = m_throttleCategories[(int)ThrottleOutPacketType.Cloud];
bucket.RequestedDripRate = cloud;
bucket.RequestedBurst = m_burst;
bucket = m_throttleCategories[(int)ThrottleOutPacketType.Asset];
bucket.RequestedDripRate = asset;
bucket.RequestedBurst = m_burst;
bucket = m_throttleCategories[(int)ThrottleOutPacketType.Task];
bucket.RequestedDripRate = task;
bucket.RequestedBurst = m_burst;
bucket = m_throttleCategories[(int)ThrottleOutPacketType.Texture];
bucket.RequestedDripRate = texture;
bucket.RequestedBurst = m_burst;
// Reset the packed throttles cached data
m_packedThrottles = null;
}
public byte[] GetThrottlesPacked(float multiplier)
{
byte[] data = m_packedThrottles;
if (data == null)
{
float rate;
data = new byte[7 * 4];
int i = 0;
// multiply by 8 to convert bytes back to bits
multiplier *= 8;
rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Resend].RequestedDripRate * multiplier;
Buffer.BlockCopy(Utils.FloatToBytes(rate), 0, data, i, 4); i += 4;
rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Land].RequestedDripRate * multiplier;
Buffer.BlockCopy(Utils.FloatToBytes(rate), 0, data, i, 4); i += 4;
rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Wind].RequestedDripRate * multiplier;
Buffer.BlockCopy(Utils.FloatToBytes(rate), 0, data, i, 4); i += 4;
rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Cloud].RequestedDripRate * multiplier;
Buffer.BlockCopy(Utils.FloatToBytes(rate), 0, data, i, 4); i += 4;
rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Task].RequestedDripRate * multiplier;
Buffer.BlockCopy(Utils.FloatToBytes(rate), 0, data, i, 4); i += 4;
rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Texture].RequestedDripRate * multiplier;
Buffer.BlockCopy(Utils.FloatToBytes(rate), 0, data, i, 4); i += 4;
rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Asset].RequestedDripRate * multiplier;
Buffer.BlockCopy(Utils.FloatToBytes(rate), 0, data, i, 4); i += 4;
m_packedThrottles = data;
}
return data;
}
public int GetCatBytesInSendQueue(ThrottleOutPacketType cat)
{
;
int icat = (int)cat;
if (icat > 0 && icat < THROTTLE_CATEGORY_COUNT)
{
TokenBucket bucket = m_throttleCategories[icat];
return m_packetOutboxes[icat].Count;
}
else
return 0;
}
public int GetCatBytesCanSend(ThrottleOutPacketType cat, int timeMS)
{
int icat = (int)cat;
if (icat > 0 && icat < THROTTLE_CATEGORY_COUNT)
{
TokenBucket bucket = m_throttleCategories[icat];
return bucket.GetCatBytesCanSend(timeMS);
}
else
return 0;
}
///
/// Queue an outgoing packet if appropriate.
///
///
/// Always queue the packet if at all possible.
///
/// true if the packet has been queued,
/// false if the packet has not been queued and should be sent immediately.
///
public bool EnqueueOutgoing(OutgoingPacket packet, bool forceQueue)
{
return EnqueueOutgoing(packet, forceQueue, false);
}
public bool EnqueueOutgoing(OutgoingPacket packet, bool forceQueue, bool highPriority)
{
int category = (int)packet.Category;
if (category >= 0 && category < m_packetOutboxes.Length)
{
DoubleLocklessQueue queue = m_packetOutboxes[category];
if (m_deliverPackets == false)
{
queue.Enqueue(packet, highPriority);
return true;
}
TokenBucket bucket = m_throttleCategories[category];
// Don't send this packet if queue is not empty
if (queue.Count > 0 || m_nextPackets[category] != null)
{
queue.Enqueue(packet, highPriority);
return true;
}
if (!forceQueue && bucket.RemoveTokens(packet.Buffer.DataLength))
{
// enough tokens so it can be sent imediatly by caller
return false;
}
else
{
// Force queue specified or not enough tokens in the bucket, queue this packet
queue.Enqueue(packet, highPriority);
return true;
}
}
else
{
// We don't have a token bucket for this category, so it will not be queued
return false;
}
}
///
/// Loops through all of the packet queues for this client and tries to send
/// an outgoing packet from each, obeying the throttling bucket limits
///
///
///
/// Packet queues are inspected in ascending numerical order starting from 0. Therefore, queues with a lower
/// ThrottleOutPacketType number will see their packet get sent first (e.g. if both Land and Wind queues have
/// packets, then the packet at the front of the Land queue will be sent before the packet at the front of the
/// wind queue).
///
/// This function is only called from a synchronous loop in the
/// UDPServer so we don't need to bother making this thread safe
///
///
/// True if any packets were sent, otherwise false
public bool DequeueOutgoing()
{
// if (m_deliverPackets == false) return false;
OutgoingPacket packet = null;
DoubleLocklessQueue queue;
TokenBucket bucket;
bool packetSent = false;
ThrottleOutPacketTypeFlags emptyCategories = 0;
//string queueDebugOutput = String.Empty; // Serious debug business
for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
{
bucket = m_throttleCategories[i];
//queueDebugOutput += m_packetOutboxes[i].Count + " "; // Serious debug business
if (m_nextPackets[i] != null)
{
// This bucket was empty the last time we tried to send a packet,
// leaving a dequeued packet still waiting to be sent out. Try to
// send it again
OutgoingPacket nextPacket = m_nextPackets[i];
if (bucket.RemoveTokens(nextPacket.Buffer.DataLength))
{
// Send the packet
m_udpServer.SendPacketFinal(nextPacket);
m_nextPackets[i] = null;
packetSent = true;
if (m_packetOutboxes[i].Count < 5)
emptyCategories |= CategoryToFlag(i);
}
}
else
{
// No dequeued packet waiting to be sent, try to pull one off
// this queue
queue = m_packetOutboxes[i];
if (queue != null)
{
bool success = false;
try
{
success = queue.Dequeue(out packet);
}
catch
{
m_packetOutboxes[i] = new DoubleLocklessQueue();
}
if (success)
{
// A packet was pulled off the queue. See if we have
// enough tokens in the bucket to send it out
if (bucket.RemoveTokens(packet.Buffer.DataLength))
{
// Send the packet
m_udpServer.SendPacketFinal(packet);
packetSent = true;
if (queue.Count < 5)
emptyCategories |= CategoryToFlag(i);
}
else
{
// Save the dequeued packet for the next iteration
m_nextPackets[i] = packet;
}
}
else
{
// No packets in this queue. Fire the queue empty callback
// if it has not been called recently
emptyCategories |= CategoryToFlag(i);
}
}
else
{
m_packetOutboxes[i] = new DoubleLocklessQueue();
emptyCategories |= CategoryToFlag(i);
}
}
}
if (emptyCategories != 0)
BeginFireQueueEmpty(emptyCategories);
//m_log.Info("[LLUDPCLIENT]: Queues: " + queueDebugOutput); // Serious debug business
return packetSent;
}
///
/// Called when an ACK packet is received and a round-trip time for a
/// packet is calculated. This is used to calculate the smoothed
/// round-trip time, round trip time variance, and finally the
/// retransmission timeout
///
/// Round-trip time of a single packet and its
/// acknowledgement
public void UpdateRoundTrip(float r)
{
const float ALPHA = 0.125f;
const float BETA = 0.25f;
const float K = 4.0f;
if (RTTVAR == 0.0f)
{
// First RTT measurement
SRTT = r;
RTTVAR = r * 0.5f;
}
else
{
// Subsequence RTT measurement
RTTVAR = (1.0f - BETA) * RTTVAR + BETA * Math.Abs(SRTT - r);
SRTT = (1.0f - ALPHA) * SRTT + ALPHA * r;
}
int rto = (int)(SRTT + Math.Max(m_udpServer.TickCountResolution, K * RTTVAR));
// Clamp the retransmission timeout to manageable values
rto = Utils.Clamp(rto, m_defaultRTO, m_maxRTO);
RTO = rto;
//if (RTO != rto)
// m_log.Debug("[LLUDPCLIENT]: Setting RTO to " + RTO + "ms from " + rto + "ms with an RTTVAR of " +
//RTTVAR + " based on new RTT of " + r + "ms");
}
///
/// Exponential backoff of the retransmission timeout, per section 5.5
/// of RFC 2988
///
public void BackoffRTO()
{
// Reset SRTT and RTTVAR, we assume they are bogus since things
// didn't work out and we're backing off the timeout
SRTT = 0.0f;
RTTVAR = 0.0f;
// Double the retransmission timeout
RTO = Math.Min(RTO * 2, m_maxRTO);
}
const int MIN_CALLBACK_MS = 10;
///
/// Does an early check to see if this queue empty callback is already
/// running, then asynchronously firing the event
///
/// Throttle categories to fire the callback for
private void BeginFireQueueEmpty(ThrottleOutPacketTypeFlags categories)
{
if (!m_isQueueEmptyRunning)
{
int start = Environment.TickCount & Int32.MaxValue;
if (start < m_nextOnQueueEmpty)
return;
m_isQueueEmptyRunning = true;
m_nextOnQueueEmpty = start + MIN_CALLBACK_MS;
if (m_nextOnQueueEmpty == 0)
m_nextOnQueueEmpty = 1;
if (HasUpdates(categories))
{
if (!m_udpServer.OqrEngine.IsRunning)
{
// Asynchronously run the callback
Util.FireAndForget(FireQueueEmpty, categories, "LLUDPClient.BeginFireQueueEmpty");
}
else
{
m_udpServer.OqrEngine.QueueJob(AgentID.ToString(), () => FireQueueEmpty(categories));
}
}
else
{
m_isQueueEmptyRunning = false;
}
}
}
private bool m_isQueueEmptyRunning;
///
/// Fires the OnQueueEmpty callback and sets the minimum time that it
/// can be called again
///
/// Throttle categories to fire the callback for,
/// stored as an object to match the WaitCallback delegate
/// signature
public void FireQueueEmpty(object o)
{
<<<<<<< HEAD
// m_log.DebugFormat("[LLUDPCLIENT]: FireQueueEmpty for {0} in {1}", AgentID, m_udpServer.Scene.Name);
// int start = Environment.TickCount & Int32.MaxValue;
// const int MIN_CALLBACK_MS = 30;
// if (m_udpServer.IsRunningOutbound)
// {
ThrottleOutPacketTypeFlags categories = (ThrottleOutPacketTypeFlags)o;
QueueEmpty callback = OnQueueEmpty;
=======
ThrottleOutPacketTypeFlags categories = (ThrottleOutPacketTypeFlags)o;
QueueEmpty callback = OnQueueEmpty;
>>>>>>> avn/ubitvar
if (callback != null)
{
// if (m_udpServer.IsRunningOutbound)
// {
try { callback(categories); }
catch (Exception e) { m_log.Error("[LLUDPCLIENT]: OnQueueEmpty(" + categories + ") threw an exception: " + e.Message, e); }
// }
}
m_isQueueEmptyRunning = false;
}
internal void ForceThrottleSetting(int throttle, int setting)
{
if (throttle > 0 && throttle < THROTTLE_CATEGORY_COUNT)
m_throttleCategories[throttle].RequestedDripRate = Math.Max(setting, LLUDPServer.MTU);
}
internal int GetThrottleSetting(int throttle)
{
if (throttle > 0 && throttle < THROTTLE_CATEGORY_COUNT)
return (int)m_throttleCategories[throttle].RequestedDripRate;
else
return 0;
}
///
/// Converts a integer to a
/// flag value
///
/// Throttle category to convert
/// Flag representation of the throttle category
private static ThrottleOutPacketTypeFlags CategoryToFlag(int i)
{
ThrottleOutPacketType category = (ThrottleOutPacketType)i;
/*
* Land = 1,
/// Wind data
Wind = 2,
/// Cloud data
Cloud = 3,
/// Any packets that do not fit into the other throttles
Task = 4,
/// Texture assets
Texture = 5,
/// Non-texture assets
Asset = 6,
*/
switch (category)
{
case ThrottleOutPacketType.Land:
return ThrottleOutPacketTypeFlags.Land;
case ThrottleOutPacketType.Wind:
return ThrottleOutPacketTypeFlags.Wind;
case ThrottleOutPacketType.Cloud:
return ThrottleOutPacketTypeFlags.Cloud;
case ThrottleOutPacketType.Task:
return ThrottleOutPacketTypeFlags.Task;
case ThrottleOutPacketType.Texture:
return ThrottleOutPacketTypeFlags.Texture;
case ThrottleOutPacketType.Asset:
return ThrottleOutPacketTypeFlags.Asset;
default:
return 0;
}
}
}
public class DoubleLocklessQueue : OpenSim.Framework.LocklessQueue
{
OpenSim.Framework.LocklessQueue highQueue = new OpenSim.Framework.LocklessQueue();
public override int Count
{
get
{
return base.Count + highQueue.Count;
}
}
public override bool Dequeue(out T item)
{
if (highQueue.Dequeue(out item))
return true;
return base.Dequeue(out item);
}
public void Enqueue(T item, bool highPriority)
{
if (highPriority)
highQueue.Enqueue(item);
else
Enqueue(item);
}
}
}