1 files changed, 90 insertions, 25 deletions

diff --git a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs
index 621e0fd..e52ac37 100644
--- a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs
+++ b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs
@@ -92,7 +92,7 @@ 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 OpenSim.Framework.LocklessQueue<uint> PendingAcks = new OpenSim.Framework.LocklessQueue<uint>();
+        public readonly DoubleLocklessQueue<uint> PendingAcks = new DoubleLocklessQueue<uint>();
 
         /// <summary>Current packet sequence number</summary>
         public int CurrentSequence;
@@ -146,7 +146,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP
         /// <summary>Throttle buckets for each packet category</summary>
         private readonly TokenBucket[] m_throttleCategories;
         /// <summary>Outgoing queues for throttled packets</summary>
-        private readonly OpenSim.Framework.LocklessQueue<OutgoingPacket>[] m_packetOutboxes = new OpenSim.Framework.LocklessQueue<OutgoingPacket>[THROTTLE_CATEGORY_COUNT];
+        private readonly DoubleLocklessQueue<OutgoingPacket>[] m_packetOutboxes = new DoubleLocklessQueue<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[] m_nextPackets = new OutgoingPacket[THROTTLE_CATEGORY_COUNT];
@@ -158,6 +158,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP
 
         private int m_defaultRTO = 1000; // 1sec is the recommendation in the RFC
         private int m_maxRTO = 60000;
+        public bool m_deliverPackets = true;
 
         /// <summary>
         /// Default constructor
@@ -201,7 +202,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP
                 ThrottleOutPacketType type = (ThrottleOutPacketType)i;
 
                 // Initialize the packet outboxes, where packets sit while they are waiting for tokens
-                m_packetOutboxes[i] = new OpenSim.Framework.LocklessQueue<OutgoingPacket>();
+                m_packetOutboxes[i] = new DoubleLocklessQueue<OutgoingPacket>();
                 // Initialize the token buckets that control the throttling for each category
                 m_throttleCategories[i] = new TokenBucket(m_throttleCategory, rates.GetRate(type));
             }
@@ -429,11 +430,23 @@ namespace OpenSim.Region.ClientStack.LindenUDP
         /// </returns>
         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)
             {
-                OpenSim.Framework.LocklessQueue<OutgoingPacket> queue = m_packetOutboxes[category];
+                DoubleLocklessQueue<OutgoingPacket> 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 there is already a packet waiting in the queue
@@ -441,7 +454,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP
                 // queued packets
                 if (queue.Count > 0)
                 {
-                    queue.Enqueue(packet);
+                    queue.Enqueue(packet, highPriority);
                     return true;
                 }
                 
@@ -454,7 +467,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP
                 else
                 {
                     // Force queue specified or not enough tokens in the bucket, queue this packet
-                    queue.Enqueue(packet);
+                    queue.Enqueue(packet, highPriority);
                     return true;
                 }
             }
@@ -483,8 +496,10 @@ namespace OpenSim.Region.ClientStack.LindenUDP
         /// <returns>True if any packets were sent, otherwise false</returns>
         public bool DequeueOutgoing()
         {
-            OutgoingPacket packet;
-            OpenSim.Framework.LocklessQueue<OutgoingPacket> queue;
+            if (m_deliverPackets == false) return false;
+
+            OutgoingPacket packet = null;
+            DoubleLocklessQueue<OutgoingPacket> queue;
             TokenBucket bucket;
             bool packetSent = false;
             ThrottleOutPacketTypeFlags emptyCategories = 0;
@@ -515,32 +530,49 @@ namespace OpenSim.Region.ClientStack.LindenUDP
                     // No dequeued packet waiting to be sent, try to pull one off
                     // this queue
                     queue = m_packetOutboxes[i];
-                    if (queue.Dequeue(out packet))
+                    if (queue != null)
                     {
-                        // 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))
+                        bool success = false;
+                        try
                         {
-                            // Send the packet
-                            m_udpServer.SendPacketFinal(packet);
-                            packetSent = true;
+                            success = queue.Dequeue(out packet);
                         }
-                        else
+                        catch
                         {
-                            // Save the dequeued packet for the next iteration
-                            m_nextPackets[i] = packet;
+                            m_packetOutboxes[i] = new DoubleLocklessQueue<OutgoingPacket>();
                         }
-
-                        // 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)
+                        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;
+                            }
+                            else
+                            {
+                                // 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)
+                                emptyCategories |= CategoryToFlag(i);
+                        }
+                        else
+                        {
+                            // No packets in this queue. Fire the queue empty callback
+                            // if it has not been called recently
                             emptyCategories |= CategoryToFlag(i);
+                        }
                     }
                     else
                     {
-                        // No packets in this queue. Fire the queue empty callback
-                        // if it has not been called recently
+                        m_packetOutboxes[i] = new DoubleLocklessQueue<OutgoingPacket>();
                         emptyCategories |= CategoryToFlag(i);
                     }
                 }
@@ -649,6 +681,10 @@ namespace OpenSim.Region.ClientStack.LindenUDP
             if (m_nextOnQueueEmpty == 0)
                 m_nextOnQueueEmpty = 1;
         }
+        internal void ForceThrottleSetting(int throttle, int setting)
+        {
+            m_throttleCategories[throttle].RequestedDripRate = Math.Max(setting, LLUDPServer.MTU); ;
+        }
 
         /// <summary>
         /// Converts a <seealso cref="ThrottleOutPacketType"/> integer to a
@@ -693,4 +729,33 @@ namespace OpenSim.Region.ClientStack.LindenUDP
             }
         }
     }
+
+    public class DoubleLocklessQueue<T> : OpenSim.Framework.LocklessQueue<T>
+    {
+        OpenSim.Framework.LocklessQueue<T> highQueue = new OpenSim.Framework.LocklessQueue<T>();
+
+        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);
+        }
+    }
 }