1 files changed, 162 insertions, 80 deletions

diff --git a/OpenSim/Region/ClientStack/LindenUDP/LLUDPClient.cs b/OpenSim/Region/ClientStack/LindenUDP/LLUDPClient.cs
index 871e8e8..4eee6b6 100644
--- a/OpenSim/Region/ClientStack/LindenUDP/LLUDPClient.cs
+++ b/OpenSim/Region/ClientStack/LindenUDP/LLUDPClient.cs
@@ -28,8 +28,10 @@
 using System;
 using System.Collections.Generic;
 using System.Net;
+using log4net;
 using OpenSim.Framework;
 using OpenMetaverse;
+using OpenMetaverse.Packets;
 
 namespace OpenSim.Region.ClientStack.LindenUDP
 {
@@ -59,9 +61,16 @@ namespace OpenSim.Region.ClientStack.LindenUDP
     /// </summary>
     public sealed class LLUDPClient
     {
+        // TODO: Make this a config setting
+        /// <summary>Percentage of the task throttle category that is allocated to avatar and prim
+        /// state updates</summary>
+        const float STATE_TASK_PERCENTAGE = 0.8f;
+
+        private static readonly ILog m_log = LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);
+
         /// <summary>The number of packet categories to throttle on. If a throttle category is added
         /// or removed, this number must also change</summary>
-        const int THROTTLE_CATEGORY_COUNT = 7;
+        const int THROTTLE_CATEGORY_COUNT = 8;
 
         /// <summary>Fired when updated networking stats are produced for this client</summary>
         public event PacketStats OnPacketStats;
@@ -80,10 +89,8 @@ namespace OpenSim.Region.ClientStack.LindenUDP
         /// <summary>Packets we have sent that need to be ACKed by the client</summary>
         public readonly UnackedPacketCollection NeedAcks = new UnackedPacketCollection();
         /// <summary>ACKs that are queued up, waiting to be sent to the client</summary>
-        public readonly LocklessQueue<uint> PendingAcks = new LocklessQueue<uint>();
+        public readonly OpenSim.Framework.LocklessQueue<uint> PendingAcks = new OpenSim.Framework.LocklessQueue<uint>();
 
-        /// <summary>Reference to the IClientAPI for this client</summary>
-        public LLClientView ClientAPI;
         /// <summary>Current packet sequence number</summary>
         public int CurrentSequence;
         /// <summary>Current ping sequence number</summary>
@@ -123,21 +130,21 @@ namespace OpenSim.Region.ClientStack.LindenUDP
         private int m_packetsSentReported;
 
         /// <summary>Throttle bucket for this agent's connection</summary>
-        private readonly TokenBucket throttle;
+        private readonly TokenBucket m_throttle;
         /// <summary>Throttle buckets for each packet category</summary>
-        private readonly TokenBucket[] throttleCategories;
+        private readonly TokenBucket[] m_throttleCategories;
         /// <summary>Throttle rate defaults and limits</summary>
-        private readonly ThrottleRates defaultThrottleRates;
+        private readonly ThrottleRates m_defaultThrottleRates;
         /// <summary>Outgoing queues for throttled packets</summary>
-        private readonly LocklessQueue<OutgoingPacket>[] packetOutboxes = new LocklessQueue<OutgoingPacket>[THROTTLE_CATEGORY_COUNT];
+        private readonly OpenSim.Framework.LocklessQueue<OutgoingPacket>[] m_packetOutboxes = new OpenSim.Framework.LocklessQueue<OutgoingPacket>[THROTTLE_CATEGORY_COUNT];
         /// <summary>A container that can hold one packet for each outbox, used to store
         /// dequeued packets that are being held for throttling</summary>
-        private readonly OutgoingPacket[] nextPackets = new OutgoingPacket[THROTTLE_CATEGORY_COUNT];
-        /// <summary>An optimization to store the length of dequeued packets being held
-        /// for throttling. This avoids expensive calls to Packet.Length</summary>
-        private readonly int[] nextPacketLengths = new int[THROTTLE_CATEGORY_COUNT];
+        private readonly OutgoingPacket[] m_nextPackets = new OutgoingPacket[THROTTLE_CATEGORY_COUNT];
+        /// <summary>Flags to prevent queue empty callbacks from stacking up on
+        /// top of each other</summary>
+        private readonly bool[] m_onQueueEmptyRunning = new bool[THROTTLE_CATEGORY_COUNT];
         /// <summary>A reference to the LLUDPServer that is managing this client</summary>
-        private readonly LLUDPServer udpServer;
+        private readonly LLUDPServer m_udpServer;
 
         /// <summary>
         /// Default constructor
@@ -151,24 +158,21 @@ namespace OpenSim.Region.ClientStack.LindenUDP
         /// <param name="remoteEndPoint">Remote endpoint for this connection</param>
         public LLUDPClient(LLUDPServer server, ThrottleRates rates, TokenBucket parentThrottle, uint circuitCode, UUID agentID, IPEndPoint remoteEndPoint)
         {
-            udpServer = server;
             AgentID = agentID;
             RemoteEndPoint = remoteEndPoint;
             CircuitCode = circuitCode;
-            defaultThrottleRates = rates;
+            m_udpServer = server;
+            m_defaultThrottleRates = rates;
+            m_throttle = new TokenBucket(parentThrottle, rates.TotalLimit, rates.Total);
+            m_throttleCategories = new TokenBucket[THROTTLE_CATEGORY_COUNT];
 
             for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
-                packetOutboxes[i] = new LocklessQueue<OutgoingPacket>();
-
-            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.Task] = new TokenBucket(throttle, rates.TaskLimit, rates.Task);
-            throttleCategories[(int)ThrottleOutPacketType.Texture] = new TokenBucket(throttle, rates.TextureLimit, rates.Texture);
-            throttleCategories[(int)ThrottleOutPacketType.Asset] = new TokenBucket(throttle, rates.AssetLimit, rates.Asset);
+            {
+                ThrottleOutPacketType type = (ThrottleOutPacketType)i;
+
+                m_packetOutboxes[i] = new OpenSim.Framework.LocklessQueue<OutgoingPacket>();
+                m_throttleCategories[i] = new TokenBucket(m_throttle, rates.GetLimit(type), rates.GetRate(type));
+            }
 
             // Set the granularity variable used for retransmission calculations to
             // the measured resolution of Environment.TickCount
@@ -176,6 +180,9 @@ namespace OpenSim.Region.ClientStack.LindenUDP
 
             // Default the retransmission timeout to three seconds
             RTO = 3000;
+
+            // Initialize this to a sane value to prevent early disconnects
+            TickLastPacketReceived = Environment.TickCount;
         }
 
         /// <summary>
@@ -183,8 +190,15 @@ namespace OpenSim.Region.ClientStack.LindenUDP
         /// </summary>
         public void Shutdown()
         {
-            // TODO: Do we need to invalidate the circuit?
             IsConnected = false;
+            NeedAcks.Clear();
+            for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
+            {
+                m_packetOutboxes[i].Clear();
+                m_nextPackets[i] = null;
+            }
+            OnPacketStats = null;
+            OnQueueEmpty = null;
         }
 
         /// <summary>
@@ -200,13 +214,13 @@ namespace OpenSim.Region.ClientStack.LindenUDP
             info.pendingAcks = new Dictionary<uint, uint>();
             info.needAck = new Dictionary<uint, byte[]>();
 
-            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.Task].DripRate;
-            info.assetThrottle = throttleCategories[(int)ThrottleOutPacketType.Asset].DripRate;
-            info.textureThrottle = throttleCategories[(int)ThrottleOutPacketType.Texture].DripRate;
+            info.resendThrottle = m_throttleCategories[(int)ThrottleOutPacketType.Resend].DripRate;
+            info.landThrottle = m_throttleCategories[(int)ThrottleOutPacketType.Land].DripRate;
+            info.windThrottle = m_throttleCategories[(int)ThrottleOutPacketType.Wind].DripRate;
+            info.cloudThrottle = m_throttleCategories[(int)ThrottleOutPacketType.Cloud].DripRate;
+            info.taskThrottle = m_throttleCategories[(int)ThrottleOutPacketType.State].DripRate + m_throttleCategories[(int)ThrottleOutPacketType.Task].DripRate;
+            info.assetThrottle = m_throttleCategories[(int)ThrottleOutPacketType.Asset].DripRate;
+            info.textureThrottle = m_throttleCategories[(int)ThrottleOutPacketType.Texture].DripRate;
             info.totalThrottle = info.resendThrottle + info.landThrottle + info.windThrottle + info.cloudThrottle +
                 info.taskThrottle + info.assetThrottle + info.textureThrottle;
 
@@ -267,6 +281,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP
                 adjData = throttleData;
             }
 
+            // 0.125f converts from bits to bytes
             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;
@@ -274,22 +289,40 @@ namespace OpenSim.Region.ClientStack.LindenUDP
             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);
+            // State is a subcategory of task that we allocate a percentage to
+            int state = (int)((float)task * STATE_TASK_PERCENTAGE);
+            task -= state;
+
+            int ceiling = Int32.MaxValue;
+            if (m_defaultThrottleRates.Total != 0)
+            {
+                ceiling = m_defaultThrottleRates.Total;
+                if (ceiling < Packet.MTU) ceiling = Packet.MTU;
+            }
 
-            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);
+            resend = Utils.Clamp(resend, Packet.MTU, ceiling);
+            land = Utils.Clamp(land, Packet.MTU, ceiling);
+            wind = Utils.Clamp(wind, Packet.MTU, ceiling);
+            cloud = Utils.Clamp(cloud, Packet.MTU, ceiling);
+            task = Utils.Clamp(task, Packet.MTU, ceiling);
+            texture = Utils.Clamp(texture, Packet.MTU, ceiling);
+            asset = Utils.Clamp(asset, Packet.MTU, ceiling);
+            state = Utils.Clamp(state, Packet.MTU, ceiling);
+
+            int total = resend + land + wind + cloud + task + texture + asset + state;
+            int taskTotal = task + state;
+
+            m_log.DebugFormat("[LLUDPCLIENT]: {0} is setting throttles. Resend={1}, Land={2}, Wind={3}, Cloud={4}, Task={5}, Texture={6}, Asset={7}, State={8}, Total={9}",
+                AgentID, resend, land, wind, cloud, task, texture, asset, state, total);
+
+            SetThrottle(ThrottleOutPacketType.Resend, resend, resend);
+            SetThrottle(ThrottleOutPacketType.Land, land, land);
+            SetThrottle(ThrottleOutPacketType.Wind, wind, wind);
+            SetThrottle(ThrottleOutPacketType.Cloud, cloud, cloud);
+            SetThrottle(ThrottleOutPacketType.Task, task, taskTotal);
+            SetThrottle(ThrottleOutPacketType.Texture, texture, texture);
+            SetThrottle(ThrottleOutPacketType.Asset, asset, asset);
+            SetThrottle(ThrottleOutPacketType.State, state, taskTotal);
         }
 
         public byte[] GetThrottlesPacked()
@@ -297,25 +330,26 @@ namespace OpenSim.Region.ClientStack.LindenUDP
             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), 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;
+            Buffer.BlockCopy(Utils.FloatToBytes((float)m_throttleCategories[(int)ThrottleOutPacketType.Resend].DripRate), 0, data, i, 4); i += 4;
+            Buffer.BlockCopy(Utils.FloatToBytes((float)m_throttleCategories[(int)ThrottleOutPacketType.Land].DripRate), 0, data, i, 4); i += 4;
+            Buffer.BlockCopy(Utils.FloatToBytes((float)m_throttleCategories[(int)ThrottleOutPacketType.Wind].DripRate), 0, data, i, 4); i += 4;
+            Buffer.BlockCopy(Utils.FloatToBytes((float)m_throttleCategories[(int)ThrottleOutPacketType.Cloud].DripRate), 0, data, i, 4); i += 4;
+            Buffer.BlockCopy(Utils.FloatToBytes((float)(m_throttleCategories[(int)ThrottleOutPacketType.Task].DripRate) +
+                                                        m_throttleCategories[(int)ThrottleOutPacketType.State].DripRate), 0, data, i, 4); i += 4;
+            Buffer.BlockCopy(Utils.FloatToBytes((float)m_throttleCategories[(int)ThrottleOutPacketType.Texture].DripRate), 0, data, i, 4); i += 4;
+            Buffer.BlockCopy(Utils.FloatToBytes((float)m_throttleCategories[(int)ThrottleOutPacketType.Asset].DripRate), 0, data, i, 4); i += 4;
 
             return data;
         }
 
-        public void SetThrottle(ThrottleOutPacketType category, int rate)
+        public void SetThrottle(ThrottleOutPacketType category, int rate, int maxBurst)
         {
             int i = (int)category;
-            if (i >= 0 && i < throttleCategories.Length)
+            if (i >= 0 && i < m_throttleCategories.Length)
             {
-                TokenBucket bucket = throttleCategories[(int)category];
-                bucket.MaxBurst = rate;
+                TokenBucket bucket = m_throttleCategories[(int)category];
                 bucket.DripRate = rate;
+                bucket.MaxBurst = maxBurst;
             }
         }
 
@@ -323,12 +357,12 @@ namespace OpenSim.Region.ClientStack.LindenUDP
         {
             int category = (int)packet.Category;
 
-            if (category >= 0 && category < packetOutboxes.Length)
+            if (category >= 0 && category < m_packetOutboxes.Length)
             {
-                LocklessQueue<OutgoingPacket> queue = packetOutboxes[category];
-                TokenBucket bucket = throttleCategories[category];
+                OpenSim.Framework.LocklessQueue<OutgoingPacket> queue = m_packetOutboxes[category];
+                TokenBucket bucket = m_throttleCategories[category];
 
-                if (throttleCategories[category].RemoveTokens(packet.Buffer.DataLength))
+                if (m_throttleCategories[category].RemoveTokens(packet.Buffer.DataLength))
                 {
                     // Enough tokens were removed from the bucket, the packet will not be queued
                     return false;
@@ -357,24 +391,25 @@ namespace OpenSim.Region.ClientStack.LindenUDP
         public bool DequeueOutgoing()
         {
             OutgoingPacket packet;
-            LocklessQueue<OutgoingPacket> queue;
+            OpenSim.Framework.LocklessQueue<OutgoingPacket> queue;
             TokenBucket bucket;
             bool packetSent = false;
 
             for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
             {
-                bucket = throttleCategories[i];
+                bucket = m_throttleCategories[i];
 
-                if (nextPackets[i] != null)
+                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
-                    if (bucket.RemoveTokens(nextPacketLengths[i]))
+                    OutgoingPacket nextPacket = m_nextPackets[i];
+                    if (bucket.RemoveTokens(nextPacket.Buffer.DataLength))
                     {
                         // Send the packet
-                        udpServer.SendPacketFinal(nextPackets[i]);
-                        nextPackets[i] = null;
+                        m_udpServer.SendPacketFinal(nextPacket);
+                        m_nextPackets[i] = null;
                         packetSent = true;
                     }
                 }
@@ -382,7 +417,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP
                 {
                     // No dequeued packet waiting to be sent, try to pull one off
                     // this queue
-                    queue = packetOutboxes[i];
+                    queue = m_packetOutboxes[i];
                     if (queue.Dequeue(out packet))
                     {
                         // A packet was pulled off the queue. See if we have
@@ -390,23 +425,26 @@ namespace OpenSim.Region.ClientStack.LindenUDP
                         if (bucket.RemoveTokens(packet.Buffer.DataLength))
                         {
                             // Send the packet
-                            udpServer.SendPacketFinal(packet);
+                            m_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;
+                            // Save the dequeued packet for the next iteration
+                            m_nextPackets[i] = packet;
                         }
+
+                        // 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)
+                            BeginFireQueueEmpty(i);
                     }
                     else
                     {
                         // No packets in this queue. Fire the queue empty callback
-                        QueueEmpty callback = OnQueueEmpty;
-                        if (callback != null)
-                            callback((ThrottleOutPacketType)i);
+                        // if it has not been called recently
+                        BeginFireQueueEmpty(i);
                     }
                 }
             }
@@ -414,6 +452,14 @@ namespace OpenSim.Region.ClientStack.LindenUDP
             return packetSent;
         }
 
+        /// <summary>
+        /// 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
+        /// </summary>
+        /// <param name="r">Round-trip time of a single packet and its
+        /// acknowledgement</param>
         public void UpdateRoundTrip(float r)
         {
             const float ALPHA = 0.125f;
@@ -435,8 +481,44 @@ namespace OpenSim.Region.ClientStack.LindenUDP
 
             // Always round retransmission timeout up to two seconds
             RTO = Math.Max(2000, (int)(SRTT + Math.Max(G, K * RTTVAR)));
-            //Logger.Debug("Setting agent " + this.Agent.FullName + "'s RTO to " + RTO + "ms with an RTTVAR of " +
+            //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");
         }
+
+        /// <summary>
+        /// Does an early check to see if this queue empty callback is already
+        /// running, then asynchronously firing the event
+        /// </summary>
+        /// <param name="throttleIndex">Throttle category to fire the callback
+        /// for</param>
+        private void BeginFireQueueEmpty(int throttleIndex)
+        {
+            if (!m_onQueueEmptyRunning[throttleIndex])
+                Util.FireAndForget(FireQueueEmpty, throttleIndex);
+        }
+
+        /// <summary>
+        /// Checks to see if this queue empty callback is already running,
+        /// then firing the event
+        /// </summary>
+        /// <param name="o">Throttle category to fire the callback for, stored
+        /// as an object to match the WaitCallback delegate signature</param>
+        private void FireQueueEmpty(object o)
+        {
+            int i = (int)o;
+            ThrottleOutPacketType type = (ThrottleOutPacketType)i;
+            QueueEmpty callback = OnQueueEmpty;
+
+            if (callback != null)
+            {
+                if (!m_onQueueEmptyRunning[i])
+                {
+                    m_onQueueEmptyRunning[i] = true;
+                    try { callback(type); }
+                    catch (Exception e) { m_log.Error("[LLUDPCLIENT]: OnQueueEmpty(" + type + ") threw an exception: " + e.Message, e); }
+                    m_onQueueEmptyRunning[i] = false;
+                }
+            }
+        }
     }
 }