PacketQueue.cs 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532
  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 OpenSim 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.Generic;
  29. using System.Threading;
  30. using System.Timers;
  31. using libsecondlife;
  32. using libsecondlife.Packets;
  33. using OpenSim.Framework;
  34. using OpenSim.Framework.Statistics;
  35. using OpenSim.Framework.Statistics.Interfaces;
  36. using Timer=System.Timers.Timer;
  37. namespace OpenSim.Region.ClientStack
  38. {
  39. public class PacketQueue : IPullStatsProvider
  40. {
  41. //private static readonly log4net.ILog m_log = log4net.LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);
  42. private bool m_enabled = true;
  43. private BlockingQueue<QueItem> SendQueue;
  44. private Queue<QueItem> IncomingPacketQueue;
  45. private Queue<QueItem> OutgoingPacketQueue;
  46. private Queue<QueItem> ResendOutgoingPacketQueue;
  47. private Queue<QueItem> LandOutgoingPacketQueue;
  48. private Queue<QueItem> WindOutgoingPacketQueue;
  49. private Queue<QueItem> CloudOutgoingPacketQueue;
  50. private Queue<QueItem> TaskOutgoingPacketQueue;
  51. private Queue<QueItem> TextureOutgoingPacketQueue;
  52. private Queue<QueItem> AssetOutgoingPacketQueue;
  53. private Dictionary<uint, uint> PendingAcks = new Dictionary<uint, uint>();
  54. private Dictionary<uint, Packet> NeedAck = new Dictionary<uint, Packet>();
  55. // All throttle times and number of bytes are calculated by dividing by this value
  56. // This value also determines how many times per throttletimems the timer will run
  57. // If throttleimems is 1000 ms, then the timer will fire every 1000/7 milliseconds
  58. private int throttleTimeDivisor = 7;
  59. private int throttletimems = 1000;
  60. private PacketThrottle ResendThrottle;
  61. private PacketThrottle LandThrottle;
  62. private PacketThrottle WindThrottle;
  63. private PacketThrottle CloudThrottle;
  64. private PacketThrottle TaskThrottle;
  65. private PacketThrottle AssetThrottle;
  66. private PacketThrottle TextureThrottle;
  67. private PacketThrottle TotalThrottle;
  68. // private long LastThrottle;
  69. // private long ThrottleInterval;
  70. private Timer throttleTimer;
  71. private LLUUID m_agentId;
  72. public PacketQueue(LLUUID agentId)
  73. {
  74. // While working on this, the BlockingQueue had me fooled for a bit.
  75. // The Blocking queue causes the thread to stop until there's something
  76. // in it to process. it's an on-purpose threadlock though because
  77. // without it, the clientloop will suck up all sim resources.
  78. SendQueue = new BlockingQueue<QueItem>();
  79. IncomingPacketQueue = new Queue<QueItem>();
  80. OutgoingPacketQueue = new Queue<QueItem>();
  81. ResendOutgoingPacketQueue = new Queue<QueItem>();
  82. LandOutgoingPacketQueue = new Queue<QueItem>();
  83. WindOutgoingPacketQueue = new Queue<QueItem>();
  84. CloudOutgoingPacketQueue = new Queue<QueItem>();
  85. TaskOutgoingPacketQueue = new Queue<QueItem>();
  86. TextureOutgoingPacketQueue = new Queue<QueItem>();
  87. AssetOutgoingPacketQueue = new Queue<QueItem>();
  88. // Set up the throttle classes (min, max, current) in bytes
  89. ResendThrottle = new PacketThrottle(5000, 100000, 50000);
  90. LandThrottle = new PacketThrottle(1000, 100000, 100000);
  91. WindThrottle = new PacketThrottle(1000, 100000, 10000);
  92. CloudThrottle = new PacketThrottle(1000, 100000, 50000);
  93. TaskThrottle = new PacketThrottle(1000, 800000, 100000);
  94. AssetThrottle = new PacketThrottle(1000, 800000, 80000);
  95. TextureThrottle = new PacketThrottle(1000, 800000, 100000);
  96. // Total Throttle trumps all
  97. // Number of bytes allowed to go out per second. (256kbps per client)
  98. TotalThrottle = new PacketThrottle(0, 162144, 1536000);
  99. throttleTimer = new Timer((int) (throttletimems/throttleTimeDivisor));
  100. throttleTimer.Elapsed += new ElapsedEventHandler(ThrottleTimerElapsed);
  101. throttleTimer.Start();
  102. // TIMERS needed for this
  103. // LastThrottle = DateTime.Now.Ticks;
  104. // ThrottleInterval = (long)(throttletimems/throttleTimeDivisor);
  105. m_agentId = agentId;
  106. if (StatsManager.SimExtraStats != null)
  107. {
  108. StatsManager.SimExtraStats.RegisterPacketQueueStatsProvider(m_agentId, this);
  109. }
  110. }
  111. /* STANDARD QUEUE MANIPULATION INTERFACES */
  112. public void Enqueue(QueItem item)
  113. {
  114. if (!m_enabled)
  115. {
  116. return;
  117. }
  118. // We could micro lock, but that will tend to actually
  119. // probably be worse than just synchronizing on SendQueue
  120. if (item == null)
  121. {
  122. SendQueue.Enqueue(item);
  123. return;
  124. }
  125. lock (this) {
  126. switch (item.throttleType)
  127. {
  128. case ThrottleOutPacketType.Resend:
  129. ThrottleCheck(ref ResendThrottle, ref ResendOutgoingPacketQueue, item);
  130. break;
  131. case ThrottleOutPacketType.Texture:
  132. ThrottleCheck(ref TextureThrottle, ref TextureOutgoingPacketQueue, item);
  133. break;
  134. case ThrottleOutPacketType.Task:
  135. ThrottleCheck(ref TaskThrottle, ref TaskOutgoingPacketQueue, item);
  136. break;
  137. case ThrottleOutPacketType.Land:
  138. ThrottleCheck(ref LandThrottle, ref LandOutgoingPacketQueue, item);
  139. break;
  140. case ThrottleOutPacketType.Asset:
  141. ThrottleCheck(ref AssetThrottle, ref AssetOutgoingPacketQueue, item);
  142. break;
  143. case ThrottleOutPacketType.Cloud:
  144. ThrottleCheck(ref CloudThrottle, ref CloudOutgoingPacketQueue, item);
  145. break;
  146. case ThrottleOutPacketType.Wind:
  147. ThrottleCheck(ref WindThrottle, ref WindOutgoingPacketQueue, item);
  148. break;
  149. default:
  150. // Acknowledgements and other such stuff should go directly to the blocking Queue
  151. // Throttling them may and likely 'will' be problematic
  152. SendQueue.Enqueue(item);
  153. break;
  154. }
  155. }
  156. }
  157. public QueItem Dequeue()
  158. {
  159. return SendQueue.Dequeue();
  160. }
  161. public void Flush()
  162. {
  163. lock (this)
  164. {
  165. while (PacketsWaiting())
  166. {
  167. //Now comes the fun part.. we dump all our elements into m_packetQueue that we've saved up.
  168. if (ResendOutgoingPacketQueue.Count > 0)
  169. {
  170. SendQueue.Enqueue(ResendOutgoingPacketQueue.Dequeue());
  171. }
  172. if (LandOutgoingPacketQueue.Count > 0)
  173. {
  174. SendQueue.Enqueue(LandOutgoingPacketQueue.Dequeue());
  175. }
  176. if (WindOutgoingPacketQueue.Count > 0)
  177. {
  178. SendQueue.Enqueue(WindOutgoingPacketQueue.Dequeue());
  179. }
  180. if (CloudOutgoingPacketQueue.Count > 0)
  181. {
  182. SendQueue.Enqueue(CloudOutgoingPacketQueue.Dequeue());
  183. }
  184. if (TaskOutgoingPacketQueue.Count > 0)
  185. {
  186. SendQueue.Enqueue(TaskOutgoingPacketQueue.Dequeue());
  187. }
  188. if (TextureOutgoingPacketQueue.Count > 0)
  189. {
  190. SendQueue.Enqueue(TextureOutgoingPacketQueue.Dequeue());
  191. }
  192. if (AssetOutgoingPacketQueue.Count > 0)
  193. {
  194. SendQueue.Enqueue(AssetOutgoingPacketQueue.Dequeue());
  195. }
  196. }
  197. // m_log.Info("[THROTTLE]: Processed " + throttleLoops + " packets");
  198. }
  199. }
  200. public void Close()
  201. {
  202. Flush();
  203. m_enabled = false;
  204. throttleTimer.Stop();
  205. if (StatsManager.SimExtraStats != null)
  206. {
  207. StatsManager.SimExtraStats.DeregisterPacketQueueStatsProvider(m_agentId);
  208. }
  209. }
  210. private void ResetCounters()
  211. {
  212. ResendThrottle.Reset();
  213. LandThrottle.Reset();
  214. WindThrottle.Reset();
  215. CloudThrottle.Reset();
  216. TaskThrottle.Reset();
  217. AssetThrottle.Reset();
  218. TextureThrottle.Reset();
  219. TotalThrottle.Reset();
  220. }
  221. private bool PacketsWaiting()
  222. {
  223. return (ResendOutgoingPacketQueue.Count > 0 ||
  224. LandOutgoingPacketQueue.Count > 0 ||
  225. WindOutgoingPacketQueue.Count > 0 ||
  226. CloudOutgoingPacketQueue.Count > 0 ||
  227. TaskOutgoingPacketQueue.Count > 0 ||
  228. AssetOutgoingPacketQueue.Count > 0 ||
  229. TextureOutgoingPacketQueue.Count > 0);
  230. }
  231. public void ProcessThrottle()
  232. {
  233. // I was considering this.. Will an event fire if the thread it's on is blocked?
  234. // Then I figured out.. it doesn't really matter.. because this thread won't be blocked for long
  235. // The General overhead of the UDP protocol gets sent to the queue un-throttled by this
  236. // so This'll pick up about around the right time.
  237. int MaxThrottleLoops = 4550; // 50*7 packets can be dequeued at once.
  238. int throttleLoops = 0;
  239. // We're going to dequeue all of the saved up packets until
  240. // we've hit the throttle limit or there's no more packets to send
  241. lock (this)
  242. {
  243. ResetCounters();
  244. // m_log.Info("[THROTTLE]: Entering Throttle");
  245. while (TotalThrottle.UnderLimit() && PacketsWaiting() &&
  246. (throttleLoops <= MaxThrottleLoops))
  247. {
  248. throttleLoops++;
  249. //Now comes the fun part.. we dump all our elements into m_packetQueue that we've saved up.
  250. if (ResendThrottle.UnderLimit() && ResendOutgoingPacketQueue.Count > 0)
  251. {
  252. QueItem qpack = ResendOutgoingPacketQueue.Dequeue();
  253. SendQueue.Enqueue(qpack);
  254. TotalThrottle.Add(qpack.Packet.ToBytes().Length);
  255. ResendThrottle.Add(qpack.Packet.ToBytes().Length);
  256. }
  257. if (LandThrottle.UnderLimit() && LandOutgoingPacketQueue.Count > 0)
  258. {
  259. QueItem qpack = LandOutgoingPacketQueue.Dequeue();
  260. SendQueue.Enqueue(qpack);
  261. TotalThrottle.Add(qpack.Packet.ToBytes().Length);
  262. LandThrottle.Add(qpack.Packet.ToBytes().Length);
  263. }
  264. if (WindThrottle.UnderLimit() && WindOutgoingPacketQueue.Count > 0)
  265. {
  266. QueItem qpack = WindOutgoingPacketQueue.Dequeue();
  267. SendQueue.Enqueue(qpack);
  268. TotalThrottle.Add(qpack.Packet.ToBytes().Length);
  269. WindThrottle.Add(qpack.Packet.ToBytes().Length);
  270. }
  271. if (CloudThrottle.UnderLimit() && CloudOutgoingPacketQueue.Count > 0)
  272. {
  273. QueItem qpack = CloudOutgoingPacketQueue.Dequeue();
  274. SendQueue.Enqueue(qpack);
  275. TotalThrottle.Add(qpack.Packet.ToBytes().Length);
  276. CloudThrottle.Add(qpack.Packet.ToBytes().Length);
  277. }
  278. if (TaskThrottle.UnderLimit() && TaskOutgoingPacketQueue.Count > 0)
  279. {
  280. QueItem qpack = TaskOutgoingPacketQueue.Dequeue();
  281. SendQueue.Enqueue(qpack);
  282. TotalThrottle.Add(qpack.Packet.ToBytes().Length);
  283. TaskThrottle.Add(qpack.Packet.ToBytes().Length);
  284. }
  285. if (TextureThrottle.UnderLimit() && TextureOutgoingPacketQueue.Count > 0)
  286. {
  287. QueItem qpack = TextureOutgoingPacketQueue.Dequeue();
  288. SendQueue.Enqueue(qpack);
  289. TotalThrottle.Add(qpack.Packet.ToBytes().Length);
  290. TextureThrottle.Add(qpack.Packet.ToBytes().Length);
  291. }
  292. if (AssetThrottle.UnderLimit() && AssetOutgoingPacketQueue.Count > 0)
  293. {
  294. QueItem qpack = AssetOutgoingPacketQueue.Dequeue();
  295. SendQueue.Enqueue(qpack);
  296. TotalThrottle.Add(qpack.Packet.ToBytes().Length);
  297. AssetThrottle.Add(qpack.Packet.ToBytes().Length);
  298. }
  299. }
  300. // m_log.Info("[THROTTLE]: Processed " + throttleLoops + " packets");
  301. }
  302. }
  303. private void ThrottleTimerElapsed(object sender, ElapsedEventArgs e)
  304. {
  305. // just to change the signature, and that ProcessThrottle
  306. // will be used elsewhere possibly
  307. ProcessThrottle();
  308. }
  309. private void ThrottleCheck(ref PacketThrottle throttle, ref Queue<QueItem> q, QueItem item)
  310. {
  311. // The idea.. is if the packet throttle queues are empty
  312. // and the client is under throttle for the type. Queue
  313. // it up directly. This basically short cuts having to
  314. // wait for the timer to fire to put things into the
  315. // output queue
  316. if ((q.Count == 0) && (throttle.UnderLimit()))
  317. {
  318. Monitor.Enter(this);
  319. throttle.Add(item.Packet.ToBytes().Length);
  320. TotalThrottle.Add(item.Packet.ToBytes().Length);
  321. SendQueue.Enqueue(item);
  322. Monitor.Pulse(this);
  323. Monitor.Exit(this);
  324. }
  325. else
  326. {
  327. q.Enqueue(item);
  328. }
  329. }
  330. private int ScaleThrottle(int value, int curmax, int newmax)
  331. {
  332. return (int) (((float) value/(float) curmax)*newmax);
  333. }
  334. public byte[] GetThrottlesPacked(float multiplier)
  335. {
  336. int singlefloat = 4;
  337. float tResend = ResendThrottle.Throttle*multiplier;
  338. float tLand = LandThrottle.Throttle*multiplier;
  339. float tWind = WindThrottle.Throttle*multiplier;
  340. float tCloud = CloudThrottle.Throttle*multiplier;
  341. float tTask = TaskThrottle.Throttle*multiplier;
  342. float tTexture = TextureThrottle.Throttle*multiplier;
  343. float tAsset = AssetThrottle.Throttle*multiplier;
  344. byte[] throttles = new byte[singlefloat*7];
  345. int i = 0;
  346. Buffer.BlockCopy(BitConverter.GetBytes(tResend), 0, throttles, singlefloat*i, singlefloat);
  347. i++;
  348. Buffer.BlockCopy(BitConverter.GetBytes(tLand), 0, throttles, singlefloat*i, singlefloat);
  349. i++;
  350. Buffer.BlockCopy(BitConverter.GetBytes(tWind), 0, throttles, singlefloat*i, singlefloat);
  351. i++;
  352. Buffer.BlockCopy(BitConverter.GetBytes(tCloud), 0, throttles, singlefloat*i, singlefloat);
  353. i++;
  354. Buffer.BlockCopy(BitConverter.GetBytes(tTask), 0, throttles, singlefloat*i, singlefloat);
  355. i++;
  356. Buffer.BlockCopy(BitConverter.GetBytes(tTexture), 0, throttles, singlefloat*i, singlefloat);
  357. i++;
  358. Buffer.BlockCopy(BitConverter.GetBytes(tAsset), 0, throttles, singlefloat*i, singlefloat);
  359. return throttles;
  360. }
  361. public void SetThrottleFromClient(byte[] throttle)
  362. {
  363. int tResend = -1;
  364. int tLand = -1;
  365. int tWind = -1;
  366. int tCloud = -1;
  367. int tTask = -1;
  368. int tTexture = -1;
  369. int tAsset = -1;
  370. int tall = -1;
  371. int singlefloat = 4;
  372. //Agent Throttle Block contains 7 single floatingpoint values.
  373. int j = 0;
  374. // Some Systems may be big endian...
  375. // it might be smart to do this check more often...
  376. if (!BitConverter.IsLittleEndian)
  377. for (int i = 0; i < 7; i++)
  378. Array.Reverse(throttle, j + i*singlefloat, singlefloat);
  379. // values gotten from libsecondlife.org/wiki/Throttle. Thanks MW_
  380. // bytes
  381. // Convert to integer, since.. the full fp space isn't used.
  382. tResend = (int) BitConverter.ToSingle(throttle, j);
  383. j += singlefloat;
  384. tLand = (int) BitConverter.ToSingle(throttle, j);
  385. j += singlefloat;
  386. tWind = (int) BitConverter.ToSingle(throttle, j);
  387. j += singlefloat;
  388. tCloud = (int) BitConverter.ToSingle(throttle, j);
  389. j += singlefloat;
  390. tTask = (int) BitConverter.ToSingle(throttle, j);
  391. j += singlefloat;
  392. tTexture = (int) BitConverter.ToSingle(throttle, j);
  393. j += singlefloat;
  394. tAsset = (int) BitConverter.ToSingle(throttle, j);
  395. tall = tResend + tLand + tWind + tCloud + tTask + tTexture + tAsset;
  396. /*
  397. m_log.Info("[CLIENT]: Client AgentThrottle - Got throttle:resendbytes=" + tResend +
  398. " landbytes=" + tLand +
  399. " windbytes=" + tWind +
  400. " cloudbytes=" + tCloud +
  401. " taskbytes=" + tTask +
  402. " texturebytes=" + tTexture +
  403. " Assetbytes=" + tAsset +
  404. " Allbytes=" + tall);
  405. */
  406. // Total Sanity
  407. // Make sure that the client sent sane total values.
  408. // If the client didn't send acceptable values....
  409. // Scale the clients values down until they are acceptable.
  410. if (tall <= TotalThrottle.Max)
  411. {
  412. ResendThrottle.Throttle = tResend;
  413. LandThrottle.Throttle = tLand;
  414. WindThrottle.Throttle = tWind;
  415. CloudThrottle.Throttle = tCloud;
  416. TaskThrottle.Throttle = tTask;
  417. TextureThrottle.Throttle = tTexture;
  418. AssetThrottle.Throttle = tAsset;
  419. TotalThrottle.Throttle = tall;
  420. }
  421. else if (tall < 1)
  422. {
  423. // client is stupid, penalize him by minning everything
  424. ResendThrottle.Throttle = ResendThrottle.Min;
  425. LandThrottle.Throttle = LandThrottle.Min;
  426. WindThrottle.Throttle = WindThrottle.Min;
  427. CloudThrottle.Throttle = CloudThrottle.Min;
  428. TaskThrottle.Throttle = TaskThrottle.Min;
  429. TextureThrottle.Throttle = TextureThrottle.Min;
  430. AssetThrottle.Throttle = AssetThrottle.Min;
  431. TotalThrottle.Throttle = TotalThrottle.Min;
  432. }
  433. else
  434. {
  435. // we're over so figure out percentages and use those
  436. ResendThrottle.Throttle = tResend;
  437. LandThrottle.Throttle = ScaleThrottle(tLand, tall, TotalThrottle.Max);
  438. WindThrottle.Throttle = ScaleThrottle(tWind, tall, TotalThrottle.Max);
  439. CloudThrottle.Throttle = ScaleThrottle(tCloud, tall, TotalThrottle.Max);
  440. TaskThrottle.Throttle = ScaleThrottle(tTask, tall, TotalThrottle.Max);
  441. TextureThrottle.Throttle = ScaleThrottle(tTexture, tall, TotalThrottle.Max);
  442. AssetThrottle.Throttle = ScaleThrottle(tAsset, tall, TotalThrottle.Max);
  443. TotalThrottle.Throttle = TotalThrottle.Max;
  444. }
  445. // effectively wiggling the slider causes things reset
  446. ResetCounters();
  447. }
  448. // See IPullStatsProvider
  449. public string GetStats()
  450. {
  451. return string.Format("{0,7} {1,7} {2,7} {3,7} {4,7} {5,7} {6,7} {7,7} {8,7} {9,7}",
  452. SendQueue.Count(),
  453. IncomingPacketQueue.Count,
  454. OutgoingPacketQueue.Count,
  455. ResendOutgoingPacketQueue.Count,
  456. LandOutgoingPacketQueue.Count,
  457. WindOutgoingPacketQueue.Count,
  458. CloudOutgoingPacketQueue.Count,
  459. TaskOutgoingPacketQueue.Count,
  460. TextureOutgoingPacketQueue.Count,
  461. AssetOutgoingPacketQueue.Count);
  462. }
  463. public QueItem[] GetQueueArray()
  464. {
  465. return SendQueue.GetQueueArray();
  466. }
  467. }
  468. }