/*
* 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 OpenSim.Framework;
using OpenMetaverse;
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 a packet category is empty. This event can be
/// hooked to put more data on the empty queue
///
/// Category of the packet queue that is empty
public delegate void QueueEmpty(ThrottleOutPacketType category);
#endregion Delegates
///
/// Tracks state for a client UDP connection and provides client-specific methods
///
public sealed class LLUDPClient
{
// FIXME: 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;
/// 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;
/// 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;
/// 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);
/// 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 OpenSim.Framework.LocklessQueue PendingAcks = new OpenSim.Framework.LocklessQueue();
/// 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 = true;
/// Environment.TickCount when the last packet was received for this client
public int TickLastPacketReceived;
/// Timer granularity. This is set to the measured resolution of Environment.TickCount
public readonly float G;
/// 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;
/// 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;
/// Throttle bucket for this agent's connection
private readonly TokenBucket throttle;
/// Throttle buckets for each packet category
private readonly TokenBucket[] throttleCategories;
/// Throttle rate defaults and limits
private readonly ThrottleRates defaultThrottleRates;
/// Outgoing queues for throttled packets
private readonly OpenSim.Framework.LocklessQueue[] packetOutboxes = new OpenSim.Framework.LocklessQueue[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[] nextPackets = new OutgoingPacket[THROTTLE_CATEGORY_COUNT];
/// An optimization to store the length of dequeued packets being held
/// for throttling. This avoids expensive calls to Packet.Length
private readonly int[] nextPacketLengths = new int[THROTTLE_CATEGORY_COUNT];
/// A reference to the LLUDPServer that is managing this client
private readonly LLUDPServer udpServer;
///
/// 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
public LLUDPClient(LLUDPServer server, ThrottleRates rates, TokenBucket parentThrottle, uint circuitCode, UUID agentID, IPEndPoint remoteEndPoint)
{
udpServer = server;
AgentID = agentID;
RemoteEndPoint = remoteEndPoint;
CircuitCode = circuitCode;
defaultThrottleRates = rates;
for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
packetOutboxes[i] = new OpenSim.Framework.LocklessQueue();
throttle = new TokenBucket(parentThrottle, 0, 0);
throttleCategories = new TokenBucket[THROTTLE_CATEGORY_COUNT];
throttleCategories[(int)ThrottleOutPacketType.Resend] = new TokenBucket(throttle, rates.ResendLimit, rates.Resend);
throttleCategories[(int)ThrottleOutPacketType.Land] = new TokenBucket(throttle, rates.LandLimit, rates.Land);
throttleCategories[(int)ThrottleOutPacketType.Wind] = new TokenBucket(throttle, rates.WindLimit, rates.Wind);
throttleCategories[(int)ThrottleOutPacketType.Cloud] = new TokenBucket(throttle, rates.CloudLimit, rates.Cloud);
throttleCategories[(int)ThrottleOutPacketType.Texture] = new TokenBucket(throttle, rates.TextureLimit, rates.Texture);
throttleCategories[(int)ThrottleOutPacketType.Asset] = new TokenBucket(throttle, rates.AssetLimit, rates.Asset);
// State and Transaction are actually sub-categories of the LLUDP generic "Task" category
TokenBucket stateBucket = new TokenBucket(throttle, (int)((float)rates.TaskLimit * STATE_TASK_PERCENTAGE), (int)((float)rates.Task * STATE_TASK_PERCENTAGE));
throttleCategories[(int)ThrottleOutPacketType.State] = stateBucket;
throttleCategories[(int)ThrottleOutPacketType.Task] = new TokenBucket(throttle, rates.TaskLimit - stateBucket.MaxBurst, rates.Task - stateBucket.DripRate);
// Set the granularity variable used for retransmission calculations to
// the measured resolution of Environment.TickCount
G = server.TickCountResolution;
// Default the retransmission timeout to three seconds
RTO = 3000;
// Initialize this to a sane value to prevent early disconnects
TickLastPacketReceived = Environment.TickCount;
}
///
/// Shuts down this client connection
///
public void Shutdown()
{
IsConnected = false;
NeedAcks.Clear();
for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
{
packetOutboxes[i].Clear();
nextPackets[i] = null;
}
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
ClientInfo info = new ClientInfo();
info.pendingAcks = new Dictionary();
info.needAck = new Dictionary();
info.resendThrottle = throttleCategories[(int)ThrottleOutPacketType.Resend].DripRate;
info.landThrottle = throttleCategories[(int)ThrottleOutPacketType.Land].DripRate;
info.windThrottle = throttleCategories[(int)ThrottleOutPacketType.Wind].DripRate;
info.cloudThrottle = throttleCategories[(int)ThrottleOutPacketType.Cloud].DripRate;
info.taskThrottle = throttleCategories[(int)ThrottleOutPacketType.State].DripRate + throttleCategories[(int)ThrottleOutPacketType.Task].DripRate;
info.assetThrottle = throttleCategories[(int)ThrottleOutPacketType.Asset].DripRate;
info.textureThrottle = throttleCategories[(int)ThrottleOutPacketType.Texture].DripRate;
info.totalThrottle = info.resendThrottle + info.landThrottle + info.windThrottle + info.cloudThrottle +
info.taskThrottle + info.assetThrottle + info.textureThrottle;
return 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();
}
public string GetStats()
{
// TODO: ???
return string.Format("{0,7} {1,7} {2,7} {3,7} {4,7} {5,7} {6,7} {7,7} {8,7} {9,7}",
0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
}
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)
{
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;
}
int resend = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int land = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int wind = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int cloud = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int task = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int texture = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int asset = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f);
resend = (resend <= defaultThrottleRates.ResendLimit) ? resend : defaultThrottleRates.ResendLimit;
land = (land <= defaultThrottleRates.LandLimit) ? land : defaultThrottleRates.LandLimit;
wind = (wind <= defaultThrottleRates.WindLimit) ? wind : defaultThrottleRates.WindLimit;
cloud = (cloud <= defaultThrottleRates.CloudLimit) ? cloud : defaultThrottleRates.CloudLimit;
task = (task <= defaultThrottleRates.TaskLimit) ? task : defaultThrottleRates.TaskLimit;
texture = (texture <= defaultThrottleRates.TextureLimit) ? texture : defaultThrottleRates.TextureLimit;
asset = (asset <= defaultThrottleRates.AssetLimit) ? asset : defaultThrottleRates.AssetLimit;
SetThrottle(ThrottleOutPacketType.Resend, resend);
SetThrottle(ThrottleOutPacketType.Land, land);
SetThrottle(ThrottleOutPacketType.Wind, wind);
SetThrottle(ThrottleOutPacketType.Cloud, cloud);
SetThrottle(ThrottleOutPacketType.Task, task);
SetThrottle(ThrottleOutPacketType.Texture, texture);
SetThrottle(ThrottleOutPacketType.Asset, asset);
}
public byte[] GetThrottlesPacked()
{
byte[] data = new byte[7 * 4];
int i = 0;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Resend].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Land].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Wind].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Cloud].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)(throttleCategories[(int)ThrottleOutPacketType.Task].DripRate) +
throttleCategories[(int)ThrottleOutPacketType.State].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Texture].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Asset].DripRate), 0, data, i, 4); i += 4;
return data;
}
public void SetThrottle(ThrottleOutPacketType category, int rate)
{
if (category == ThrottleOutPacketType.Task)
{
TokenBucket stateBucket = throttleCategories[(int)ThrottleOutPacketType.State];
TokenBucket taskBucket = throttleCategories[(int)ThrottleOutPacketType.Task];
stateBucket.MaxBurst = (int)((float)rate * STATE_TASK_PERCENTAGE);
stateBucket.DripRate = (int)((float)rate * STATE_TASK_PERCENTAGE);
taskBucket.MaxBurst = rate - stateBucket.MaxBurst;
taskBucket.DripRate = rate - stateBucket.DripRate;
}
else
{
int i = (int)category;
if (i >= 0 && i < throttleCategories.Length)
{
TokenBucket bucket = throttleCategories[(int)category];
bucket.MaxBurst = rate;
bucket.DripRate = rate;
}
}
}
public bool EnqueueOutgoing(OutgoingPacket packet)
{
int category = (int)packet.Category;
if (category >= 0 && category < packetOutboxes.Length)
{
OpenSim.Framework.LocklessQueue queue = packetOutboxes[category];
TokenBucket bucket = throttleCategories[category];
if (throttleCategories[category].RemoveTokens(packet.Buffer.DataLength))
{
// Enough tokens were removed from the bucket, the packet will not be queued
return false;
}
else
{
// Not enough tokens in the bucket, queue this packet
queue.Enqueue(packet);
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
/// any outgoing packets, obeying the throttling bucket limits
///
/// 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()
{
OutgoingPacket packet;
OpenSim.Framework.LocklessQueue queue;
TokenBucket bucket;
bool packetSent = false;
for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
{
bucket = throttleCategories[i];
if (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
if (bucket.RemoveTokens(nextPacketLengths[i]))
{
// Send the packet
udpServer.SendPacketFinal(nextPackets[i]);
nextPackets[i] = null;
packetSent = true;
}
}
else
{
// No dequeued packet waiting to be sent, try to pull one off
// this queue
queue = packetOutboxes[i];
if (queue.Dequeue(out packet))
{
// 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
udpServer.SendPacketFinal(packet);
packetSent = true;
}
else
{
// Save the dequeued packet and the length calculation for
// the next iteration
nextPackets[i] = packet;
nextPacketLengths[i] = packet.Buffer.DataLength;
}
// If the queue is empty after this dequeue, fire the queue
// empty callback now so it has a chance to fill before we
// get back here
if (queue.Count == 0)
FireQueueEmpty(i);
}
else
{
// No packets in this queue. Fire the queue empty callback
// if it has not been called recently
FireQueueEmpty(i);
}
}
}
return packetSent;
}
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;
}
// Always round retransmission timeout up to two seconds
RTO = Math.Max(2000, (int)(SRTT + Math.Max(G, K * RTTVAR)));
//m_log.Debug("[LLUDPCLIENT]: Setting agent " + this.Agent.FullName + "'s RTO to " + RTO + "ms with an RTTVAR of " +
// RTTVAR + " based on new RTT of " + r + "ms");
}
private void FireQueueEmpty(int queueIndex)
{
QueueEmpty callback = OnQueueEmpty;
if (callback != null)
Util.FireAndForget(delegate(object o) { callback((ThrottleOutPacketType)(int)o); }, queueIndex);
}
}
}