PriorityQueue.cs 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444
  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.Collections.Concurrent;
  31. using System.Threading;
  32. using System.Runtime.InteropServices;
  33. using OpenSim.Framework;
  34. namespace OpenSim.Region.Framework.Scenes
  35. {
  36. public class PriorityQueue
  37. {
  38. // private static readonly ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
  39. public delegate bool UpdatePriorityHandler(ref int priority, ISceneEntity entity);
  40. /// <summary>
  41. /// Total number of queues (priorities) available
  42. /// </summary>
  43. public const int NumberOfQueues = 13; // includes immediate queues, m_queueCounts need to be set acording
  44. /// <summary>
  45. /// Number of queuest (priorities) that are processed immediately
  46. /// </summary.
  47. public const int NumberOfImmediateQueues = 2;
  48. // first queues are immediate, so no counts
  49. private static readonly int[] m_queueCounts = {0, 0, 8, 8, 5, 4, 3, 2, 1, 1, 1, 1, 1 };
  50. // this is ava, ava, attach, <10m, 20,40,80,160m,320,640,1280, +
  51. private PriorityMinHeap[] m_heaps = new PriorityMinHeap[NumberOfQueues];
  52. private ConcurrentDictionary<uint, EntityUpdate> m_lookupTable;
  53. // internal state used to ensure the deqeues are spread across the priority
  54. // queues "fairly". queuecounts is the amount to pull from each queue in
  55. // each pass. weighted towards the higher priority queues
  56. private int m_nextQueue = 0;
  57. private int m_countFromQueue = 0;
  58. // next request is a counter of the number of updates queued, it provides
  59. // a total ordering on the updates coming through the queue and is more
  60. // lightweight (and more discriminating) than tick count
  61. private ulong m_nextRequest = 0;
  62. /// <summary>
  63. /// Lock for enqueue and dequeue operations on the priority queue
  64. /// </summary>
  65. private object m_syncRoot = new object();
  66. public object SyncRoot {
  67. get { return m_syncRoot; }
  68. }
  69. #region constructor
  70. public PriorityQueue(int capacity)
  71. {
  72. capacity /= 4;
  73. for (int i = 0; i < m_heaps.Length; ++i)
  74. m_heaps[i] = new PriorityMinHeap(capacity);
  75. m_lookupTable = new ConcurrentDictionary<uint, EntityUpdate>();
  76. m_nextQueue = NumberOfImmediateQueues;
  77. m_countFromQueue = m_queueCounts[m_nextQueue];
  78. }
  79. #endregion Constructor
  80. #region PublicMethods
  81. public void Close()
  82. {
  83. for (int i = 0; i < m_heaps.Length; ++i)
  84. {
  85. m_heaps[i].Clear();
  86. m_heaps[i] = null;
  87. }
  88. m_heaps = null;
  89. m_lookupTable.Clear();
  90. m_lookupTable = null;
  91. }
  92. /// <summary>
  93. /// Return the number of items in the queues
  94. /// </summary>
  95. public int Count
  96. {
  97. get
  98. {
  99. return m_lookupTable.Count;
  100. }
  101. }
  102. /// <summary>
  103. /// Enqueue an item into the specified priority queue
  104. /// </summary>
  105. public bool Enqueue(int pqueue, EntityUpdate value)
  106. {
  107. ulong entry;
  108. EntityUpdate existentup;
  109. uint localid = value.Entity.LocalId;
  110. if (m_lookupTable.TryGetValue(localid, out existentup))
  111. {
  112. int eqqueue = existentup.PriorityQueue;
  113. existentup.UpdateFromNew(value, pqueue);
  114. value.Free();
  115. if (pqueue != eqqueue)
  116. {
  117. m_heaps[eqqueue].RemoveAt(existentup.PriorityQueueIndex);
  118. m_heaps[pqueue].Add(existentup);
  119. }
  120. return true;
  121. }
  122. entry = m_nextRequest++;
  123. value.Update(pqueue, entry);
  124. m_heaps[pqueue].Add(value);
  125. m_lookupTable[localid] = value;
  126. return true;
  127. }
  128. public void Remove(List<uint> ids)
  129. {
  130. EntityUpdate lookup;
  131. foreach (uint localid in ids)
  132. {
  133. if (m_lookupTable.TryRemove(localid, out lookup))
  134. {
  135. m_heaps[lookup.PriorityQueue].RemoveAt(lookup.PriorityQueueIndex);
  136. lookup.Free();
  137. }
  138. }
  139. }
  140. /// <summary>
  141. /// Remove an item from one of the queues. Specifically, it removes the
  142. /// oldest item from the next queue in order to provide fair access to
  143. /// all of the queues
  144. /// </summary>
  145. public bool TryDequeue(out EntityUpdate value)
  146. {
  147. // If there is anything in immediate queues, return it first no
  148. // matter what else. Breaks fairness. But very useful.
  149. for (int iq = 0; iq < NumberOfImmediateQueues; iq++)
  150. {
  151. if (m_heaps[iq].Count > 0)
  152. {
  153. value = m_heaps[iq].RemoveNext();
  154. return m_lookupTable.TryRemove(value.Entity.LocalId, out value);
  155. }
  156. }
  157. // To get the fair queing, we cycle through each of the
  158. // queues when finding an element to dequeue.
  159. // We pull (NumberOfQueues - QueueIndex) items from each queue in order
  160. // to give lower numbered queues a higher priority and higher percentage
  161. // of the bandwidth.
  162. PriorityMinHeap curheap = m_heaps[m_nextQueue];
  163. // Check for more items to be pulled from the current queue
  164. if (m_countFromQueue > 0 && curheap.Count > 0)
  165. {
  166. --m_countFromQueue;
  167. value = curheap.RemoveNext();
  168. return m_lookupTable.TryRemove(value.Entity.LocalId, out value);
  169. }
  170. // Find the next non-immediate queue with updates in it
  171. for (int i = NumberOfImmediateQueues; i < NumberOfQueues; ++i)
  172. {
  173. m_nextQueue++;
  174. if(m_nextQueue >= NumberOfQueues)
  175. m_nextQueue = NumberOfImmediateQueues;
  176. curheap = m_heaps[m_nextQueue];
  177. if (curheap.Count == 0)
  178. continue;
  179. m_countFromQueue = m_queueCounts[m_nextQueue];
  180. --m_countFromQueue;
  181. value = curheap.RemoveNext();
  182. return m_lookupTable.TryRemove(value.Entity.LocalId, out value);
  183. }
  184. value = null;
  185. return false;
  186. }
  187. public bool TryOrderedDequeue(out EntityUpdate value)
  188. {
  189. for (int iq = 0; iq < NumberOfQueues; ++iq)
  190. {
  191. PriorityMinHeap curheap = m_heaps[iq];
  192. if (curheap.Count > 0)
  193. {
  194. value = curheap.RemoveNext();
  195. return m_lookupTable.TryRemove(value.Entity.LocalId, out value);
  196. }
  197. }
  198. value = null;
  199. return false;
  200. }
  201. /// <summary>
  202. /// Reapply the prioritization function to each of the updates currently
  203. /// stored in the priority queues.
  204. /// </summary
  205. public void Reprioritize(UpdatePriorityHandler handler)
  206. {
  207. int pqueue = 0;
  208. foreach (EntityUpdate currentEU in m_lookupTable.Values)
  209. {
  210. if (handler(ref pqueue, currentEU.Entity))
  211. {
  212. // unless the priority queue has changed, there is no need to modify
  213. // the entry
  214. if (pqueue != currentEU.PriorityQueue)
  215. {
  216. m_heaps[currentEU.PriorityQueue].RemoveAt(currentEU.PriorityQueueIndex);
  217. currentEU.PriorityQueue = pqueue;
  218. m_heaps[pqueue].Add(currentEU);
  219. }
  220. }
  221. else
  222. {
  223. break;
  224. }
  225. }
  226. }
  227. /// <summary>
  228. /// </summary>
  229. public override string ToString()
  230. {
  231. string s = "";
  232. for (int i = 0; i < NumberOfQueues; i++)
  233. s += String.Format("{0,7} ", m_heaps[i].Count);
  234. return s;
  235. }
  236. #endregion PublicMethods
  237. }
  238. public class PriorityMinHeap
  239. {
  240. public const int MIN_CAPACITY = 16;
  241. private EntityUpdate[] m_items;
  242. private int m_size;
  243. private int minCapacity;
  244. public PriorityMinHeap(int _capacity)
  245. {
  246. minCapacity = MIN_CAPACITY;
  247. m_items = new EntityUpdate[_capacity];
  248. m_size = 0;
  249. }
  250. public int Count { get { return m_size; } }
  251. private bool BubbleUp(int index)
  252. {
  253. EntityUpdate tmp;
  254. EntityUpdate item = m_items[index];
  255. ulong itemEntryOrder = item.EntryOrder;
  256. int current, parent;
  257. for (current = index, parent = (current - 1) / 2;
  258. (current > 0) && m_items[parent].EntryOrder > itemEntryOrder;
  259. current = parent, parent = (current - 1) / 2)
  260. {
  261. tmp = m_items[parent];
  262. tmp.PriorityQueueIndex = current;
  263. m_items[current] = tmp;
  264. }
  265. if (current != index)
  266. {
  267. item.PriorityQueueIndex = current;
  268. m_items[current] = item;
  269. return true;
  270. }
  271. return false;
  272. }
  273. private void BubbleDown(int index)
  274. {
  275. if(m_size < 2)
  276. return;
  277. EntityUpdate childItem;
  278. EntityUpdate childItemR;
  279. EntityUpdate item = m_items[index];
  280. ulong itemEntryOrder = item.EntryOrder;
  281. int current;
  282. int child;
  283. int childlimit = m_size - 1;
  284. for (current = index, child = (2 * current) + 1;
  285. current < m_size / 2;
  286. current = child, child = (2 * current) + 1)
  287. {
  288. childItem = m_items[child];
  289. if (child < childlimit)
  290. {
  291. childItemR = m_items[child + 1];
  292. if(childItem.EntryOrder > childItemR.EntryOrder)
  293. {
  294. childItem = childItemR;
  295. ++child;
  296. }
  297. }
  298. if (childItem.EntryOrder >= itemEntryOrder)
  299. break;
  300. childItem.PriorityQueueIndex = current;
  301. m_items[current] = childItem;
  302. }
  303. if (current != index)
  304. {
  305. item.PriorityQueueIndex = current;
  306. m_items[current] = item;
  307. }
  308. }
  309. public void Add(EntityUpdate value)
  310. {
  311. if (m_size == m_items.Length)
  312. {
  313. int newcapacity = (int)((m_items.Length * 200L) / 100L);
  314. if (newcapacity < (m_items.Length + MIN_CAPACITY))
  315. newcapacity = m_items.Length + MIN_CAPACITY;
  316. Array.Resize<EntityUpdate>(ref m_items, newcapacity);
  317. }
  318. value.PriorityQueueIndex = m_size;
  319. m_items[m_size] = value;
  320. BubbleUp(m_size);
  321. ++m_size;
  322. }
  323. public void Clear()
  324. {
  325. for (int index = 0; index < m_size; ++index)
  326. m_items[index].Free();
  327. m_size = 0;
  328. }
  329. public void RemoveAt(int index)
  330. {
  331. if (m_size == 0)
  332. throw new InvalidOperationException("Heap is empty");
  333. if (index >= m_size)
  334. throw new ArgumentOutOfRangeException("index");
  335. --m_size;
  336. if (m_size > 0)
  337. {
  338. if (index != m_size)
  339. {
  340. EntityUpdate tmp = m_items[m_size];
  341. tmp.PriorityQueueIndex = index;
  342. m_items[index] = tmp;
  343. m_items[m_size] = null;
  344. if (!BubbleUp(index))
  345. BubbleDown(index);
  346. }
  347. }
  348. else if (m_items.Length > 4 * minCapacity)
  349. m_items = new EntityUpdate[minCapacity];
  350. }
  351. public EntityUpdate RemoveNext()
  352. {
  353. if (m_size == 0)
  354. throw new InvalidOperationException("Heap is empty");
  355. EntityUpdate item = m_items[0];
  356. --m_size;
  357. if (m_size > 0)
  358. {
  359. EntityUpdate tmp = m_items[m_size];
  360. tmp.PriorityQueueIndex = 0;
  361. m_items[0] = tmp;
  362. m_items[m_size] = null;
  363. BubbleDown(0);
  364. }
  365. else if (m_items.Length > 4 * minCapacity)
  366. m_items = new EntityUpdate[minCapacity];
  367. return item;
  368. }
  369. public bool Remove(EntityUpdate value)
  370. {
  371. int index = value.PriorityQueueIndex;
  372. if (index != -1)
  373. {
  374. RemoveAt(index);
  375. return true;
  376. }
  377. return false;
  378. }
  379. }
  380. }