1
0

PriorityQueue.cs 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323
  1. /*
  2. * Copyright (c) Contributors, http://opensimulator.org/
  3. * See CONTRIBUTORS.TXT for a full list of copyright holders.
  4. *
  5. * Redistribution and use in source and binary forms, with or without
  6. * modification, are permitted provided that the following conditions are met:
  7. * * Redistributions of source code must retain the above copyright
  8. * notice, this list of conditions and the following disclaimer.
  9. * * Redistributions in binary form must reproduce the above copyright
  10. * notice, this list of conditions and the following disclaimer in the
  11. * documentation and/or other materials provided with the distribution.
  12. * * Neither the name of the OpenSimulator Project nor the
  13. * names of its contributors may be used to endorse or promote products
  14. * derived from this software without specific prior written permission.
  15. *
  16. * THIS SOFTWARE IS PROVIDED BY THE DEVELOPERS ``AS IS'' AND ANY
  17. * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  18. * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  19. * DISCLAIMED. IN NO EVENT SHALL THE CONTRIBUTORS BE LIABLE FOR ANY
  20. * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  21. * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  22. * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  23. * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  24. * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  25. * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  26. */
  27. using System;
  28. using System.Collections;
  29. using System.Collections.Generic;
  30. using System.Reflection;
  31. using OpenSim.Framework;
  32. using OpenSim.Framework.Client;
  33. using log4net;
  34. namespace OpenSim.Framework
  35. {
  36. public class PriorityQueue
  37. {
  38. // private static readonly ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
  39. public delegate bool UpdatePriorityHandler(ref uint priority, ISceneEntity entity);
  40. /// <summary>
  41. /// Total number of queues (priorities) available
  42. /// </summary>
  43. public const uint NumberOfQueues = 12;
  44. /// <summary>
  45. /// Number of queuest (priorities) that are processed immediately
  46. /// </summary.
  47. public const uint NumberOfImmediateQueues = 2;
  48. private MinHeap<MinHeapItem>[] m_heaps = new MinHeap<MinHeapItem>[NumberOfQueues];
  49. private Dictionary<uint, LookupItem> m_lookupTable;
  50. // internal state used to ensure the deqeues are spread across the priority
  51. // queues "fairly". queuecounts is the amount to pull from each queue in
  52. // each pass. weighted towards the higher priority queues
  53. private uint m_nextQueue = 0;
  54. private uint m_countFromQueue = 0;
  55. private uint[] m_queueCounts = { 8, 4, 4, 2, 2, 2, 2, 1, 1, 1, 1, 1 };
  56. // next request is a counter of the number of updates queued, it provides
  57. // a total ordering on the updates coming through the queue and is more
  58. // lightweight (and more discriminating) than tick count
  59. private UInt64 m_nextRequest = 0;
  60. /// <summary>
  61. /// Lock for enqueue and dequeue operations on the priority queue
  62. /// </summary>
  63. private object m_syncRoot = new object();
  64. public object SyncRoot {
  65. get { return this.m_syncRoot; }
  66. }
  67. #region constructor
  68. public PriorityQueue() : this(MinHeap<MinHeapItem>.DEFAULT_CAPACITY) { }
  69. public PriorityQueue(int capacity)
  70. {
  71. m_lookupTable = new Dictionary<uint, LookupItem>(capacity);
  72. for (int i = 0; i < m_heaps.Length; ++i)
  73. m_heaps[i] = new MinHeap<MinHeapItem>(capacity);
  74. m_nextQueue = NumberOfImmediateQueues;
  75. m_countFromQueue = m_queueCounts[m_nextQueue];
  76. }
  77. #endregion Constructor
  78. #region PublicMethods
  79. /// <summary>
  80. /// Return the number of items in the queues
  81. /// </summary>
  82. public int Count
  83. {
  84. get
  85. {
  86. int count = 0;
  87. for (int i = 0; i < m_heaps.Length; ++i)
  88. count += m_heaps[i].Count;
  89. return count;
  90. }
  91. }
  92. /// <summary>
  93. /// Enqueue an item into the specified priority queue
  94. /// </summary>
  95. public bool Enqueue(uint pqueue, IEntityUpdate value)
  96. {
  97. LookupItem lookup;
  98. uint localid = value.Entity.LocalId;
  99. UInt64 entry = m_nextRequest++;
  100. if (m_lookupTable.TryGetValue(localid, out lookup))
  101. {
  102. entry = lookup.Heap[lookup.Handle].EntryOrder;
  103. value.Update(lookup.Heap[lookup.Handle].Value);
  104. lookup.Heap.Remove(lookup.Handle);
  105. }
  106. pqueue = Util.Clamp<uint>(pqueue, 0, NumberOfQueues - 1);
  107. lookup.Heap = m_heaps[pqueue];
  108. lookup.Heap.Add(new MinHeapItem(pqueue, entry, value), ref lookup.Handle);
  109. m_lookupTable[localid] = lookup;
  110. return true;
  111. }
  112. /// <summary>
  113. /// Remove an item from one of the queues. Specifically, it removes the
  114. /// oldest item from the next queue in order to provide fair access to
  115. /// all of the queues
  116. /// </summary>
  117. public bool TryDequeue(out IEntityUpdate value, out Int32 timeinqueue)
  118. {
  119. // If there is anything in priority queue 0, return it first no
  120. // matter what else. Breaks fairness. But very useful.
  121. for (int iq = 0; iq < NumberOfImmediateQueues; iq++)
  122. {
  123. if (m_heaps[iq].Count > 0)
  124. {
  125. MinHeapItem item = m_heaps[iq].RemoveMin();
  126. m_lookupTable.Remove(item.Value.Entity.LocalId);
  127. timeinqueue = Util.EnvironmentTickCountSubtract(item.EntryTime);
  128. value = item.Value;
  129. return true;
  130. }
  131. }
  132. // To get the fair queing, we cycle through each of the
  133. // queues when finding an element to dequeue.
  134. // We pull (NumberOfQueues - QueueIndex) items from each queue in order
  135. // to give lower numbered queues a higher priority and higher percentage
  136. // of the bandwidth.
  137. // Check for more items to be pulled from the current queue
  138. if (m_heaps[m_nextQueue].Count > 0 && m_countFromQueue > 0)
  139. {
  140. m_countFromQueue--;
  141. MinHeapItem item = m_heaps[m_nextQueue].RemoveMin();
  142. m_lookupTable.Remove(item.Value.Entity.LocalId);
  143. timeinqueue = Util.EnvironmentTickCountSubtract(item.EntryTime);
  144. value = item.Value;
  145. return true;
  146. }
  147. // Find the next non-immediate queue with updates in it
  148. for (int i = 0; i < NumberOfQueues; ++i)
  149. {
  150. m_nextQueue = (uint)((m_nextQueue + 1) % NumberOfQueues);
  151. m_countFromQueue = m_queueCounts[m_nextQueue];
  152. // if this is one of the immediate queues, just skip it
  153. if (m_nextQueue < NumberOfImmediateQueues)
  154. continue;
  155. if (m_heaps[m_nextQueue].Count > 0)
  156. {
  157. m_countFromQueue--;
  158. MinHeapItem item = m_heaps[m_nextQueue].RemoveMin();
  159. m_lookupTable.Remove(item.Value.Entity.LocalId);
  160. timeinqueue = Util.EnvironmentTickCountSubtract(item.EntryTime);
  161. value = item.Value;
  162. return true;
  163. }
  164. }
  165. timeinqueue = 0;
  166. value = default(IEntityUpdate);
  167. return false;
  168. }
  169. /// <summary>
  170. /// Reapply the prioritization function to each of the updates currently
  171. /// stored in the priority queues.
  172. /// </summary
  173. public void Reprioritize(UpdatePriorityHandler handler)
  174. {
  175. MinHeapItem item;
  176. foreach (LookupItem lookup in new List<LookupItem>(this.m_lookupTable.Values))
  177. {
  178. if (lookup.Heap.TryGetValue(lookup.Handle, out item))
  179. {
  180. uint pqueue = item.PriorityQueue;
  181. uint localid = item.Value.Entity.LocalId;
  182. if (handler(ref pqueue, item.Value.Entity))
  183. {
  184. // unless the priority queue has changed, there is no need to modify
  185. // the entry
  186. pqueue = Util.Clamp<uint>(pqueue, 0, NumberOfQueues - 1);
  187. if (pqueue != item.PriorityQueue)
  188. {
  189. lookup.Heap.Remove(lookup.Handle);
  190. LookupItem litem = lookup;
  191. litem.Heap = m_heaps[pqueue];
  192. litem.Heap.Add(new MinHeapItem(pqueue, item), ref litem.Handle);
  193. m_lookupTable[localid] = litem;
  194. }
  195. }
  196. else
  197. {
  198. // m_log.WarnFormat("[PQUEUE]: UpdatePriorityHandler returned false for {0}",item.Value.Entity.UUID);
  199. lookup.Heap.Remove(lookup.Handle);
  200. this.m_lookupTable.Remove(localid);
  201. }
  202. }
  203. }
  204. }
  205. /// <summary>
  206. /// </summary>
  207. public override string ToString()
  208. {
  209. string s = "";
  210. for (int i = 0; i < NumberOfQueues; i++)
  211. s += String.Format("{0,7} ",m_heaps[i].Count);
  212. return s;
  213. }
  214. #endregion PublicMethods
  215. #region MinHeapItem
  216. private struct MinHeapItem : IComparable<MinHeapItem>
  217. {
  218. private IEntityUpdate value;
  219. internal IEntityUpdate Value {
  220. get {
  221. return this.value;
  222. }
  223. }
  224. private uint pqueue;
  225. internal uint PriorityQueue {
  226. get {
  227. return this.pqueue;
  228. }
  229. }
  230. private Int32 entrytime;
  231. internal Int32 EntryTime {
  232. get {
  233. return this.entrytime;
  234. }
  235. }
  236. private UInt64 entryorder;
  237. internal UInt64 EntryOrder
  238. {
  239. get {
  240. return this.entryorder;
  241. }
  242. }
  243. internal MinHeapItem(uint pqueue, MinHeapItem other)
  244. {
  245. this.entrytime = other.entrytime;
  246. this.entryorder = other.entryorder;
  247. this.value = other.value;
  248. this.pqueue = pqueue;
  249. }
  250. internal MinHeapItem(uint pqueue, UInt64 entryorder, IEntityUpdate value)
  251. {
  252. this.entrytime = Util.EnvironmentTickCount();
  253. this.entryorder = entryorder;
  254. this.value = value;
  255. this.pqueue = pqueue;
  256. }
  257. public override string ToString()
  258. {
  259. return String.Format("[{0},{1},{2}]",pqueue,entryorder,value.Entity.LocalId);
  260. }
  261. public int CompareTo(MinHeapItem other)
  262. {
  263. // I'm assuming that the root part of an SOG is added to the update queue
  264. // before the component parts
  265. return Comparer<UInt64>.Default.Compare(this.EntryOrder, other.EntryOrder);
  266. }
  267. }
  268. #endregion
  269. #region LookupItem
  270. private struct LookupItem
  271. {
  272. internal MinHeap<MinHeapItem> Heap;
  273. internal IHandle Handle;
  274. }
  275. #endregion
  276. }
  277. }