diff --git a/OpenSim/Region/ClientStack/Linden/UDP/LLClientView.cs b/OpenSim/Region/ClientStack/Linden/UDP/LLClientView.cs index f7ab6a37c9..eedb5202e1 100755 --- a/OpenSim/Region/ClientStack/Linden/UDP/LLClientView.cs +++ b/OpenSim/Region/ClientStack/Linden/UDP/LLClientView.cs @@ -4773,7 +4773,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP } } - uint priority = m_prioritizer.GetUpdatePriority(this, entity); + int priority = m_prioritizer.GetUpdatePriority(this, entity); lock (m_entityUpdates.SyncRoot) m_entityUpdates.Enqueue(priority, EntityUpdatesPool.Get(entity, updateFlags)); @@ -4789,7 +4789,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP { // If the update exists in priority queue, it will be updated. // If it does not exist then it will be added with the current (rather than its original) priority - uint priority = m_prioritizer.GetUpdatePriority(this, update.Entity); + int priority = m_prioritizer.GetUpdatePriority(this, update.Entity); lock (m_entityUpdates.SyncRoot) m_entityUpdates.Enqueue(priority, update); @@ -5778,9 +5778,9 @@ namespace OpenSim.Region.ClientStack.LindenUDP CheckGroupsInViewBusy = false; } - private bool UpdatePriorityHandler(ref uint priority, ISceneEntity entity) + private bool UpdatePriorityHandler(ref int priority, ISceneEntity entity) { - if (entity == null) + if (!IsActive) return false; priority = m_prioritizer.GetUpdatePriority(this, entity); @@ -5968,16 +5968,14 @@ namespace OpenSim.Region.ClientStack.LindenUDP public void SendObjectPropertiesFamilyData(ISceneEntity entity, uint requestFlags) { - uint priority = 0; // time based ordering only lock (m_entityProps.SyncRoot) - m_entityProps.Enqueue(priority, EntityUpdatesPool.Get(entity, (PrimUpdateFlags)requestFlags, true, false)); + m_entityProps.Enqueue(0, EntityUpdatesPool.Get(entity, (PrimUpdateFlags)requestFlags, true, false)); } private void ResendPropertyUpdate(EntityUpdate update) { - uint priority = 0; lock (m_entityProps.SyncRoot) - m_entityProps.Enqueue(priority, update); + m_entityProps.Enqueue(0, update); } private void ResendPropertyUpdates(List updates, OutgoingPacket oPacket) @@ -6002,9 +6000,8 @@ namespace OpenSim.Region.ClientStack.LindenUDP public void SendObjectPropertiesReply(ISceneEntity entity) { - uint priority = 0; // time based ordering only lock (m_entityProps.SyncRoot) - m_entityProps.Enqueue(priority, EntityUpdatesPool.Get(entity,0,false,true)); + m_entityProps.Enqueue(0, EntityUpdatesPool.Get(entity,0,false,true)); } static private readonly byte[] ObjectPropertyUpdateHeader = new byte[] { diff --git a/OpenSim/Region/Framework/Scenes/EntityUpdates.cs b/OpenSim/Region/Framework/Scenes/EntityUpdates.cs index ae6e51039c..fe42e1af2e 100644 --- a/OpenSim/Region/Framework/Scenes/EntityUpdates.cs +++ b/OpenSim/Region/Framework/Scenes/EntityUpdates.cs @@ -50,25 +50,14 @@ namespace OpenSim.Region.Framework.Scenes { // for priority queue - private uint m_pqueue; + public int PriorityQueue; + public int PriorityQueueIndex; private ulong m_entryorder; private ISceneEntity m_entity; private PrimUpdateFlags m_flags; public ObjectPropertyUpdateFlags m_propsFlags; - public uint PriorityQueue - { - get - { - return m_pqueue; - } - set - { - m_pqueue = value; - } - } - public ulong EntryOrder { get @@ -111,7 +100,7 @@ namespace OpenSim.Region.Framework.Scenes set { m_flags = value; } } - public void Update(uint pqueue, ulong entry) + public void Update(int pqueue, ulong entry) { if ((m_flags & PrimUpdateFlags.CancelKill) != 0) { @@ -121,11 +110,11 @@ namespace OpenSim.Region.Framework.Scenes m_flags = PrimUpdateFlags.FullUpdatewithAnim; } - m_pqueue = pqueue; + PriorityQueue = pqueue; m_entryorder = entry; } - public void Update(EntityUpdate oldupdate, uint pqueue, ulong entry) + public void Update(EntityUpdate oldupdate, int pqueue, ulong entry) { // we are on the new one m_propsFlags |= oldupdate.PropsFlags; @@ -133,6 +122,7 @@ namespace OpenSim.Region.Framework.Scenes PrimUpdateFlags updateFlags = oldupdate.Flags; if ((m_flags & PrimUpdateFlags.UpdateProbe) != 0) updateFlags &= ~PrimUpdateFlags.UpdateProbe; + if ((m_flags & PrimUpdateFlags.CancelKill) != 0) { if ((m_flags & PrimUpdateFlags.UpdateProbe) != 0) @@ -143,13 +133,36 @@ namespace OpenSim.Region.Framework.Scenes else m_flags |= updateFlags; - m_pqueue = pqueue; + PriorityQueue = pqueue; m_entryorder = entry; } + public void UpdateFromNew(EntityUpdate newupdate, int pqueue) + { + // we are on the new one + m_propsFlags |= newupdate.PropsFlags; + PrimUpdateFlags newFlags = newupdate.Flags; + + if ((newFlags & PrimUpdateFlags.UpdateProbe) != 0) + m_flags &= ~PrimUpdateFlags.UpdateProbe; + + if ((newFlags & PrimUpdateFlags.CancelKill) != 0) + { + if ((newFlags & PrimUpdateFlags.UpdateProbe) != 0) + m_flags = PrimUpdateFlags.UpdateProbe; + else + newFlags = PrimUpdateFlags.FullUpdatewithAnim; + } + else + m_flags |= newFlags; + + PriorityQueue = pqueue; + } + public void Free() { m_entity = null; + PriorityQueueIndex = -1; EntityUpdatesPool.Free(this); } @@ -173,7 +186,7 @@ namespace OpenSim.Region.Framework.Scenes public override string ToString() { - return String.Format("[{0},{1},{2}]", m_pqueue, m_entryorder, m_entity.LocalId); + return String.Format("[{0},{1},{2}]", PriorityQueue, m_entryorder, m_entity.LocalId); } public int CompareTo(EntityUpdate other) diff --git a/OpenSim/Region/Framework/Scenes/Prioritizer.cs b/OpenSim/Region/Framework/Scenes/Prioritizer.cs index 9f2399beb6..df4f0ddcd1 100644 --- a/OpenSim/Region/Framework/Scenes/Prioritizer.cs +++ b/OpenSim/Region/Framework/Scenes/Prioritizer.cs @@ -55,7 +55,7 @@ namespace OpenSim.Region.Framework.Scenes /// /// Returns the priority queue into which the update should be placed. /// - public uint GetUpdatePriority(IClientAPI client, ISceneEntity entity) + public int GetUpdatePriority(IClientAPI client, ISceneEntity entity) { // If entity is null we have a serious problem if (entity == null) @@ -68,8 +68,7 @@ namespace OpenSim.Region.Framework.Scenes if (client.AgentId == entity.UUID) return 0; - uint priority; - + int priority; switch (m_scene.UpdatePrioritizationScheme) { case UpdatePrioritizationSchemes.SimpleAngularDistance: @@ -80,13 +79,14 @@ namespace OpenSim.Region.Framework.Scenes priority = GetPriorityByBestAvatarResponsiveness(client, entity); break; } - + if(priority >= PriorityQueue.NumberOfQueues - 1) + return PriorityQueue.NumberOfQueues - 1; return priority; } - private uint GetPriorityByBestAvatarResponsiveness(IClientAPI client, ISceneEntity entity) + private int GetPriorityByBestAvatarResponsiveness(IClientAPI client, ISceneEntity entity) { - uint pqueue = 2; // keep compiler happy + int pqueue = 2; // keep compiler happy ScenePresence presence = m_scene.GetScenePresence(client.AgentId); if (presence != null) @@ -122,7 +122,7 @@ namespace OpenSim.Region.Framework.Scenes return pqueue; } - private uint ComputeDistancePriority(IClientAPI client, ISceneEntity entity, bool useFrontBack) + private int ComputeDistancePriority(IClientAPI client, ISceneEntity entity, bool useFrontBack) { // Get this agent's position ScenePresence presence = m_scene.GetScenePresence(client.AgentId); @@ -164,7 +164,7 @@ namespace OpenSim.Region.Framework.Scenes // And convert the distance to a priority queue, this computation gives queues // at 10, 20, 40, 80, 160, 320, 640, and 1280m - uint pqueue = PriorityQueue.NumberOfImmediateQueues + 1; // reserve attachments queue + int pqueue = PriorityQueue.NumberOfImmediateQueues + 1; // reserve attachments queue if (distance > 10f) { float tmp = (float)Math.Log((double)distance) * 1.442695f - 3.321928f; @@ -172,7 +172,7 @@ namespace OpenSim.Region.Framework.Scenes // now // 1st constant is 1/(log(2)) (natural log) so we get log2(distance) // 2st constant makes it be log2(distance/10) - pqueue += (uint)tmp; + pqueue += (int)tmp; } // If this is a root agent, then determine front & back @@ -193,19 +193,18 @@ namespace OpenSim.Region.Framework.Scenes return pqueue; } - private uint GetPriorityByAngularDistance(IClientAPI client, ISceneEntity entity) + private int GetPriorityByAngularDistance(IClientAPI client, ISceneEntity entity) { ScenePresence presence = m_scene.GetScenePresence(client.AgentId); if (presence == null) return PriorityQueue.NumberOfQueues - 1; - uint pqueue = ComputeAngleDistancePriority(presence, entity); - return pqueue; + return ComputeAngleDistancePriority(presence, entity); } - private uint ComputeAngleDistancePriority(ScenePresence presence, ISceneEntity entity) + private int ComputeAngleDistancePriority(ScenePresence presence, ISceneEntity entity) { - uint pqueue = PriorityQueue.NumberOfImmediateQueues; + int pqueue = PriorityQueue.NumberOfImmediateQueues; float distance; Vector3 presencePos = presence.AbsolutePosition; @@ -216,7 +215,7 @@ namespace OpenSim.Region.Framework.Scenes if (distance > 400f) { float tmp = (float)Math.Log(distance) * 0.7213475f - 4.321928f; - pqueue += (uint)tmp; + pqueue += (int)tmp; } return pqueue; } @@ -238,7 +237,7 @@ namespace OpenSim.Region.Framework.Scenes if (distance > 400f) { float tmp = (float)Math.Log(distance) * 0.7213475f - 4.321928f; - pqueue += (uint)tmp; + pqueue += (int)tmp; } return pqueue; } @@ -265,7 +264,7 @@ namespace OpenSim.Region.Framework.Scenes // now // 1st constant is 1/(log(2)) (natural log) so we get log2(distance) // 2st constant makes it be log2(distance/10) - pqueue += (uint)tmp; + pqueue += (int)tmp; } return pqueue; diff --git a/OpenSim/Region/Framework/Scenes/PriorityQueue.cs b/OpenSim/Region/Framework/Scenes/PriorityQueue.cs index 614ce4b6b7..40d75ab1f6 100644 --- a/OpenSim/Region/Framework/Scenes/PriorityQueue.cs +++ b/OpenSim/Region/Framework/Scenes/PriorityQueue.cs @@ -26,7 +26,12 @@ */ using System; +using System.Collections; using System.Collections.Generic; +using System.Collections.Concurrent; +using System.Threading; +using System.Runtime.InteropServices; + using OpenSim.Framework; namespace OpenSim.Region.Framework.Scenes @@ -35,32 +40,30 @@ namespace OpenSim.Region.Framework.Scenes { // private static readonly ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType); - public delegate bool UpdatePriorityHandler(ref uint priority, ISceneEntity entity); + public delegate bool UpdatePriorityHandler(ref int priority, ISceneEntity entity); /// /// Total number of queues (priorities) available /// - public const uint NumberOfQueues = 13; // includes immediate queues, m_queueCounts need to be set acording + public const int NumberOfQueues = 13; // includes immediate queues, m_queueCounts need to be set acording /// /// Number of queuest (priorities) that are processed immediately /// [] m_heaps = new MinHeap[NumberOfQueues]; - private Dictionary m_lookupTable; + private PriorityMinHeap[] m_heaps = new PriorityMinHeap[NumberOfQueues]; + private ConcurrentDictionary m_lookupTable; // internal state used to ensure the deqeues are spread across the priority // queues "fairly". queuecounts is the amount to pull from each queue in // each pass. weighted towards the higher priority queues - private uint m_nextQueue = 0; - private uint m_countFromQueue = 0; - private int m_capacity; - private int m_added; + private int m_nextQueue = 0; + private int m_countFromQueue = 0; // next request is a counter of the number of updates queued, it provides // a total ordering on the updates coming through the queue and is more @@ -72,24 +75,19 @@ namespace OpenSim.Region.Framework.Scenes /// private object m_syncRoot = new object(); public object SyncRoot { - get { return this.m_syncRoot; } + get { return m_syncRoot; } } #region constructor - public PriorityQueue() : this(MinHeap.DEFAULT_CAPACITY) { } - public PriorityQueue(int capacity) { - m_capacity = 16; capacity /= 4; - for (int i = 0; i < m_heaps.Length; ++i) - m_heaps[i] = new MinHeap(capacity); + m_heaps[i] = new PriorityMinHeap(capacity); - m_lookupTable = new Dictionary(m_capacity); + m_lookupTable = new ConcurrentDictionary(); m_nextQueue = NumberOfImmediateQueues; m_countFromQueue = m_queueCounts[m_nextQueue]; - m_added = 0; } #endregion Constructor @@ -98,8 +96,7 @@ namespace OpenSim.Region.Framework.Scenes { for (int i = 0; i < m_heaps.Length; ++i) { - foreach(EntityUpdate eu in m_heaps[i]) - eu.Free(); + m_heaps[i].Clear(); m_heaps[i] = null; } @@ -115,74 +112,53 @@ namespace OpenSim.Region.Framework.Scenes { get { - int count = 0; - for (int i = 0; i < m_heaps.Length; ++i) - count += m_heaps[i].Count; - - return count; + return m_lookupTable.Count; } } /// /// Enqueue an item into the specified priority queue /// - public bool Enqueue(uint pqueue, EntityUpdate value) + public bool Enqueue(int pqueue, EntityUpdate value) { - LookupItem lookup; - IHandle lookupH; ulong entry; + EntityUpdate existentup; uint localid = value.Entity.LocalId; - if (m_lookupTable.TryGetValue(localid, out lookup)) + if (m_lookupTable.TryGetValue(localid, out existentup)) { - lookupH = lookup.Handle; - entry = lookup.Heap[lookupH].EntryOrder; - EntityUpdate up = lookup.Heap[lookupH]; - lookup.Heap.Remove(lookupH); + int eqqueue = existentup.PriorityQueue; - if((up.Flags & PrimUpdateFlags.CancelKill) != 0) - entry = m_nextRequest++; + existentup.UpdateFromNew(value, pqueue); + value.Free(); - pqueue = Util.Clamp(pqueue, 0, NumberOfQueues - 1); - value.Update(up, pqueue, entry); - up.Free(); - - lookup.Heap = m_heaps[pqueue]; - lookup.Heap.Add(value, ref lookup.Handle); - m_lookupTable[localid] = lookup; + if (pqueue != eqqueue) + { + m_heaps[eqqueue].RemoveAt(existentup.PriorityQueueIndex); + m_heaps[pqueue].Add(existentup); + } return true; } entry = m_nextRequest++; - ++m_added; - pqueue = Util.Clamp(pqueue, 0, NumberOfQueues - 1); value.Update(pqueue, entry); - lookup.Heap = m_heaps[pqueue]; - lookup.Heap.Add(value, ref lookup.Handle); - m_lookupTable[localid] = lookup; - + m_heaps[pqueue].Add(value); + m_lookupTable[localid] = value; return true; } public void Remove(List ids) { - LookupItem lookup; - + EntityUpdate lookup; foreach (uint localid in ids) { - if (m_lookupTable.TryGetValue(localid, out lookup)) + if (m_lookupTable.TryRemove(localid, out lookup)) { - lookup.Heap[lookup.Handle].Free(); - lookup.Heap.Remove(lookup.Handle); - m_lookupTable.Remove(localid); + m_heaps[lookup.PriorityQueue].RemoveAt(lookup.PriorityQueueIndex); + lookup.Free(); } } - if(m_lookupTable.Count == 0 && m_added > 8 * m_capacity) - { - m_lookupTable = new Dictionary(m_capacity); - m_added = 0; - } } /// @@ -194,14 +170,13 @@ namespace OpenSim.Region.Framework.Scenes { // If there is anything in immediate queues, return it first no // matter what else. Breaks fairness. But very useful. - + for (int iq = 0; iq < NumberOfImmediateQueues; iq++) { if (m_heaps[iq].Count > 0) { - value = m_heaps[iq].RemoveMin(); - m_lookupTable.Remove(value.Entity.LocalId); - return true; + value = m_heaps[iq].RemoveNext(); + return m_lookupTable.TryRemove(value.Entity.LocalId, out value); } } @@ -211,19 +186,18 @@ namespace OpenSim.Region.Framework.Scenes // to give lower numbered queues a higher priority and higher percentage // of the bandwidth. - MinHeap curheap = m_heaps[m_nextQueue]; + PriorityMinHeap curheap = m_heaps[m_nextQueue]; // Check for more items to be pulled from the current queue if (m_countFromQueue > 0 && curheap.Count > 0) { --m_countFromQueue; - value = curheap.RemoveMin(); - m_lookupTable.Remove(value.Entity.LocalId); - return true; + value = curheap.RemoveNext(); + return m_lookupTable.TryRemove(value.Entity.LocalId, out value); } // Find the next non-immediate queue with updates in it - for (uint i = NumberOfImmediateQueues; i < NumberOfQueues; ++i) + for (int i = NumberOfImmediateQueues; i < NumberOfQueues; ++i) { m_nextQueue++; if(m_nextQueue >= NumberOfQueues) @@ -236,17 +210,11 @@ namespace OpenSim.Region.Framework.Scenes m_countFromQueue = m_queueCounts[m_nextQueue]; --m_countFromQueue; - value = curheap.RemoveMin(); - m_lookupTable.Remove(value.Entity.LocalId); - return true; + value = curheap.RemoveNext(); + return m_lookupTable.TryRemove(value.Entity.LocalId, out value); } - value = default(EntityUpdate); - if(m_lookupTable.Count == 0 && m_added > 8 * m_capacity) - { - m_lookupTable = new Dictionary(m_capacity); - m_added = 0; - } + value = null; return false; } @@ -254,21 +222,15 @@ namespace OpenSim.Region.Framework.Scenes { for (int iq = 0; iq < NumberOfQueues; ++iq) { - MinHeap curheap = m_heaps[iq]; + PriorityMinHeap curheap = m_heaps[iq]; if (curheap.Count > 0) { - value = curheap.RemoveMin(); - m_lookupTable.Remove(value.Entity.LocalId); - return true; + value = curheap.RemoveNext(); + return m_lookupTable.TryRemove(value.Entity.LocalId, out value); } } - value = default(EntityUpdate); - if(m_lookupTable.Count == 0 && m_added > 8 * m_capacity) - { - m_lookupTable = new Dictionary(m_capacity); - m_added = 0; - } + value = null; return false; } @@ -278,35 +240,24 @@ namespace OpenSim.Region.Framework.Scenes /// (m_lookupTable.Values)) + int pqueue = 0; + foreach (EntityUpdate currentEU in m_lookupTable.Values) { - if (lookup.Heap.TryGetValue(lookup.Handle, out currentEU)) + if (handler(ref pqueue, currentEU.Entity)) { - if (handler(ref pqueue, currentEU.Entity)) + // unless the priority queue has changed, there is no need to modify + // the entry + if (pqueue != currentEU.PriorityQueue) { - // unless the priority queue has changed, there is no need to modify - // the entry - pqueue = Util.Clamp(pqueue, 0, NumberOfQueues - 1); - if (pqueue != currentEU.PriorityQueue) - { - currentEU.PriorityQueue = pqueue; - - lookup.Heap.Remove(lookup.Handle); - LookupItem litem = lookup; - litem.Heap = m_heaps[pqueue]; - litem.Heap.Add(currentEU, ref litem.Handle); - m_lookupTable[currentEU.Entity.LocalId] = litem; - } - } - else - { - // m_log.WarnFormat("[PQUEUE]: UpdatePriorityHandler returned false for {0}",item.Value.Entity.UUID); - lookup.Heap.Remove(lookup.Handle); - m_lookupTable.Remove(currentEU.Entity.LocalId); + m_heaps[currentEU.PriorityQueue].RemoveAt(currentEU.PriorityQueueIndex); + currentEU.PriorityQueue = pqueue; + m_heaps[pqueue].Add(currentEU); } } + else + { + break; + } } } @@ -321,14 +272,148 @@ namespace OpenSim.Region.Framework.Scenes } #endregion PublicMethods + } + public class PriorityMinHeap + { + public const int MIN_CAPACITY = 16; -#region LookupItem - private struct LookupItem + private EntityUpdate[] m_items; + private int m_size; + private int minCapacity; + + public PriorityMinHeap(int _capacity) { - internal MinHeap Heap; - internal IHandle Handle; + minCapacity = MIN_CAPACITY; + m_items = new EntityUpdate[_capacity]; + m_size = 0; + } + + public int Count { get { return m_size; } } + + [System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.AggressiveInlining)] + private void Set(EntityUpdate item, int index) + { + m_items[index] = item; + item.PriorityQueueIndex = index; + } + + private bool BubbleUp(int index) + { + EntityUpdate item = m_items[index]; + int current, parent; + + for (current = index, parent = (current - 1) / 2; + (current > 0) && m_items[parent].EntryOrder > item.EntryOrder; + current = parent, parent = (current - 1) / 2) + { + Set(m_items[parent], current); + } + + if (current != index) + { + Set(item, current); + return true; + } + return false; + } + + private void BubbleDown(int index) + { + if(m_size < 2) + return; + + EntityUpdate item = m_items[index]; + int current; + int child; + + for (current = index, child = (2 * current) + 1; + current < m_size / 2; + current = child, child = (2 * current) + 1) + { + if ((child < m_size - 1) && m_items[child].EntryOrder > m_items[child + 1].EntryOrder) + ++child; + if (m_items[child].EntryOrder >= item.EntryOrder) + break; + Set(m_items[child], current); + } + + if (current != index) + Set(item, current); + } + + public void Add(EntityUpdate value) + { + if (m_size == m_items.Length) + { + int newcapacity = (int)((m_items.Length * 200L) / 100L); + if (newcapacity < (m_items.Length + MIN_CAPACITY)) + newcapacity = m_items.Length + MIN_CAPACITY; + Array.Resize(ref m_items, newcapacity); + } + + Set(value, m_size); + BubbleUp(m_size); + ++m_size; + } + + public void Clear() + { + for (int index = 0; index < m_size; ++index) + m_items[index].Free(); + m_size = 0; + } + + public void RemoveAt(int index) + { + if (m_size == 0) + throw new InvalidOperationException("Heap is empty"); + if (index >= m_size) + throw new ArgumentOutOfRangeException("index"); + + --m_size; + if (m_size > 0) + { + if (index != m_size) + { + Set(m_items[m_size], index); + m_items[m_size] = null; + if (!BubbleUp(index)) + BubbleDown(index); + } + } + else if (m_items.Length > 4 * minCapacity) + m_items = new EntityUpdate[minCapacity]; + } + + public EntityUpdate RemoveNext() + { + if (m_size == 0) + throw new InvalidOperationException("Heap is empty"); + + EntityUpdate item = m_items[0]; + --m_size; + if (m_size > 0) + { + Set(m_items[m_size], 0); + m_items[m_size] = null; + BubbleDown(0); + } + else if (m_items.Length > 4 * minCapacity) + m_items = new EntityUpdate[minCapacity]; + + return item; + } + + public bool Remove(EntityUpdate value) + { + int index = value.PriorityQueueIndex; + if (index != -1) + { + RemoveAt(index); + return true; + } + return false; } -#endregion } }