/* * 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); /// 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; 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); // 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 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); } // 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; 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) { int icat = (int)throttleType; if (icat > 0 && icat < THROTTLE_CATEGORY_COUNT) return m_packetOutboxes[icat].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); // Make sure none of the throttles are set below our packet MTU, // otherwise a throttle could become permanently clogged /* now 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); int total = resend + land + wind + cloud + task + texture + asset; float m_burst = total * m_burstTime; if (ThrottleDebugLevel > 0) { 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); } 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 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) { ThrottleOutPacketTypeFlags categories = (ThrottleOutPacketTypeFlags)o; QueueEmpty callback = OnQueueEmpty; 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); } } }