/* * 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); } } }