LLUDPClient.cs 35 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838
  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.Net;
  29. using System.Threading;
  30. using log4net;
  31. using OpenSim.Framework;
  32. using OpenMetaverse;
  33. using TokenBucket = OpenSim.Region.ClientStack.LindenUDP.TokenBucket;
  34. namespace OpenSim.Region.ClientStack.LindenUDP
  35. {
  36. #region Delegates
  37. /// <summary>
  38. /// Fired when updated networking stats are produced for this client
  39. /// </summary>
  40. /// <param name="inPackets">Number of incoming packets received since this
  41. /// event was last fired</param>
  42. /// <param name="outPackets">Number of outgoing packets sent since this
  43. /// event was last fired</param>
  44. /// <param name="unAckedBytes">Current total number of bytes in packets we
  45. /// are waiting on ACKs for</param>
  46. public delegate void PacketStats(int inPackets, int outPackets, int unAckedBytes);
  47. /// <summary>
  48. /// Fired when the queue for one or more packet categories is empty. This
  49. /// event can be hooked to put more data on the empty queues
  50. /// </summary>
  51. /// <param name="category">Categories of the packet queues that are empty</param>
  52. public delegate void QueueEmpty(ThrottleOutPacketTypeFlags categories);
  53. #endregion Delegates
  54. /// <summary>
  55. /// Tracks state for a client UDP connection and provides client-specific methods
  56. /// </summary>
  57. public sealed class LLUDPClient
  58. {
  59. private static readonly ILog m_log = LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);
  60. /// <summary>The number of packet categories to throttle on. If a throttle category is added
  61. /// or removed, this number must also change</summary>
  62. const int THROTTLE_CATEGORY_COUNT = 8;
  63. /// <summary>
  64. /// Controls whether information is logged about each outbound packet immediately before it is sent. For debug purposes.
  65. /// </summary>
  66. /// <remarks>Any level above 0 will turn on logging.</remarks>
  67. public int DebugDataOutLevel { get; set; }
  68. /// <summary>
  69. /// Controls whether information is logged about each outbound packet immediately before it is sent. For debug purposes.
  70. /// </summary>
  71. /// <remarks>Any level above 0 will turn on logging.</remarks>
  72. public int ThrottleDebugLevel { get; set; }
  73. /// <summary>Fired when updated networking stats are produced for this client</summary>
  74. public event PacketStats OnPacketStats;
  75. /// <summary>Fired when the queue for a packet category is empty. This event can be
  76. /// hooked to put more data on the empty queue</summary>
  77. public event QueueEmpty OnQueueEmpty;
  78. public event Func<ThrottleOutPacketTypeFlags, bool> HasUpdates;
  79. /// <summary>AgentID for this client</summary>
  80. public readonly UUID AgentID;
  81. /// <summary>The remote address of the connected client</summary>
  82. public readonly IPEndPoint RemoteEndPoint;
  83. /// <summary>Circuit code that this client is connected on</summary>
  84. public readonly uint CircuitCode;
  85. /// <summary>Sequence numbers of packets we've received (for duplicate checking)</summary>
  86. public IncomingPacketHistoryCollection PacketArchive = new(1024);
  87. /// <summary>Packets we have sent that need to be ACKed by the client</summary>
  88. public UnackedPacketCollection NeedAcks = new();
  89. /// <summary>ACKs that are queued up, waiting to be sent to the client</summary>
  90. public DoubleLocklessQueue<uint> PendingAcks = new();
  91. public int AckStalls;
  92. /// <summary>Current packet sequence number</summary>
  93. public int CurrentSequence;
  94. /// <summary>Current ping sequence number</summary>
  95. public byte CurrentPingSequence;
  96. /// <summary>True when this connection is alive, otherwise false</summary>
  97. public bool IsConnected = true;
  98. /// <summary>True when this connection is paused, otherwise false</summary>
  99. public bool IsPaused;
  100. /// <summary>Environment.TickCount when the last packet was received for this client</summary>
  101. public int TickLastPacketReceived;
  102. /// <summary>Smoothed round-trip time. A smoothed average of the round-trip time for sending a
  103. /// reliable packet to the client and receiving an ACK</summary>
  104. public float SRTT;
  105. /// <summary>Round-trip time variance. Measures the consistency of round-trip times</summary>
  106. public float RTTVAR;
  107. /// <summary>Retransmission timeout. Packets that have not been acknowledged in this number of
  108. /// milliseconds or longer will be resent</summary>
  109. /// <remarks>Calculated from <seealso cref="SRTT"/> and <seealso cref="RTTVAR"/> using the
  110. /// guidelines in RFC 2988</remarks>
  111. public int m_RTO;
  112. /// <summary>Number of bytes received since the last acknowledgement was sent out. This is used
  113. /// to loosely follow the TCP delayed ACK algorithm in RFC 1122 (4.2.3.2)</summary>
  114. public int BytesSinceLastACK;
  115. /// <summary>Number of packets received from this client</summary>
  116. public int PacketsReceived;
  117. /// <summary>Number of packets sent to this client</summary>
  118. public int PacketsSent;
  119. /// <summary>Number of packets resent to this client</summary>
  120. public int PacketsResent;
  121. /// <summary>Total byte count of unacked packets sent to this client</summary>
  122. public int UnackedBytes;
  123. private int m_packetsUnAckReported;
  124. /// <summary>Total number of received packets that we have reported to the OnPacketStats event(s)</summary>
  125. private int m_packetsReceivedReported;
  126. /// <summary>Total number of sent packets that we have reported to the OnPacketStats event(s)</summary>
  127. private int m_packetsSentReported;
  128. /// <summary>Holds the Environment.TickCount value of when the next OnQueueEmpty can be fired</summary>
  129. private double m_nextOnQueueEmpty = 0;
  130. /// <summary>Throttle bucket for this agent's connection</summary>
  131. private readonly AdaptiveTokenBucket m_throttleClient;
  132. public AdaptiveTokenBucket FlowThrottle
  133. {
  134. get { return m_throttleClient; }
  135. }
  136. /// <summary>Throttle buckets for each packet category</summary>
  137. private readonly TokenBucket[] m_throttleCategories;
  138. /// <summary>Outgoing queues for throttled packets</summary>
  139. private readonly DoubleLocklessQueue<OutgoingPacket>[] m_packetOutboxes = new DoubleLocklessQueue<OutgoingPacket>[THROTTLE_CATEGORY_COUNT];
  140. /// <summary>A container that can hold one packet for each outbox, used to store
  141. /// dequeued packets that are being held for throttling</summary>
  142. private readonly OutgoingPacket[] m_nextPackets = new OutgoingPacket[THROTTLE_CATEGORY_COUNT];
  143. /// <summary>A reference to the LLUDPServer that is managing this client</summary>
  144. private readonly LLUDPServer m_udpServer;
  145. private readonly int m_defaultRTO = 1000; // 1sec is the recommendation in the RFC
  146. private readonly int m_maxRTO = 3000;
  147. private readonly int m_minRTO = 250;
  148. private readonly float m_burstTime;
  149. private readonly float m_maxRate;
  150. public double m_lastStartpingTimeMS;
  151. public int m_pingMS;
  152. public int PingTimeMS
  153. {
  154. get
  155. {
  156. if (m_pingMS < 10)
  157. return 10;
  158. if(m_pingMS > 2000)
  159. return 2000;
  160. return m_pingMS;
  161. }
  162. }
  163. private readonly ClientInfo m_info = new();
  164. /// <summary>
  165. /// Default constructor
  166. /// </summary>
  167. /// <param name="server">Reference to the UDP server this client is connected to</param>
  168. /// <param name="rates">Default throttling rates and maximum throttle limits</param>
  169. /// <param name="parentThrottle">Parent HTB (hierarchical token bucket)
  170. /// that the child throttles will be governed by</param>
  171. /// <param name="circuitCode">Circuit code for this connection</param>
  172. /// <param name="agentID">AgentID for the connected agent</param>
  173. /// <param name="remoteEndPoint">Remote endpoint for this connection</param>
  174. /// <param name="defaultRTO">
  175. /// Default retransmission timeout for unacked packets. The RTO will never drop
  176. /// beyond this number.
  177. /// </param>
  178. /// <param name="maxRTO">
  179. /// The maximum retransmission timeout for unacked packets. The RTO will never exceed this number.
  180. /// </param>
  181. public LLUDPClient(
  182. LLUDPServer server, ThrottleRates rates, TokenBucket parentThrottle, uint circuitCode, UUID agentID,
  183. IPEndPoint remoteEndPoint, int defaultRTO, int maxRTO)
  184. {
  185. AgentID = agentID;
  186. RemoteEndPoint = remoteEndPoint;
  187. CircuitCode = circuitCode;
  188. m_udpServer = server;
  189. if (defaultRTO != 0)
  190. m_defaultRTO = defaultRTO;
  191. if (maxRTO != 0)
  192. m_maxRTO = maxRTO;
  193. m_burstTime = rates.BurstTime;
  194. m_maxRate = rates.ClientMaxRate;
  195. // Create a token bucket throttle for this client that has the scene token bucket as a parent
  196. m_throttleClient = new AdaptiveTokenBucket(parentThrottle, m_maxRate, m_maxRate * m_burstTime, rates.AdaptiveThrottlesEnabled);
  197. // Create an array of token buckets for this clients different throttle categories
  198. m_throttleCategories = new TokenBucket[THROTTLE_CATEGORY_COUNT];
  199. for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
  200. {
  201. ThrottleOutPacketType type = (ThrottleOutPacketType)i;
  202. // Initialize the packet outboxes, where packets sit while they are waiting for tokens
  203. m_packetOutboxes[i] = new DoubleLocklessQueue<OutgoingPacket>();
  204. // Initialize the token buckets that control the throttling for each category
  205. float rate = rates.GetRate(type);
  206. float burst = rate * m_burstTime;
  207. m_throttleCategories[i] = new TokenBucket(m_throttleClient, rate , burst);
  208. }
  209. m_RTO = m_defaultRTO;
  210. // Initialize this to a sane value to prevent early disconnects
  211. TickLastPacketReceived = Environment.TickCount & Int32.MaxValue;
  212. m_pingMS = 20; // so filter doesnt start at 0;
  213. }
  214. /// <summary>
  215. /// Shuts down this client connection
  216. /// </summary>
  217. public void Shutdown()
  218. {
  219. IsConnected = false;
  220. for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
  221. {
  222. m_packetOutboxes[i].Clear();
  223. m_nextPackets[i] = null;
  224. }
  225. // pull the throttle out of the scene throttle
  226. m_throttleClient.Parent.UnregisterRequest(m_throttleClient);
  227. PendingAcks.Clear();
  228. NeedAcks.Clear();
  229. }
  230. /// <summary>
  231. /// Gets information about this client connection
  232. /// </summary>
  233. /// <returns>Information about the client connection</returns>
  234. public ClientInfo GetClientInfo()
  235. {
  236. // TODO: This data structure is wrong in so many ways. Locking and copying the entire lists
  237. // of pending and needed ACKs for every client every time some method wants information about
  238. // this connection is a recipe for poor performance
  239. m_info.resendThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Resend].DripRate;
  240. m_info.landThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Land].DripRate;
  241. m_info.windThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Wind].DripRate;
  242. m_info.cloudThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Cloud].DripRate;
  243. m_info.taskThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Task].DripRate;
  244. m_info.assetThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Asset].DripRate;
  245. m_info.textureThrottle = (int)m_throttleCategories[(int)ThrottleOutPacketType.Texture].DripRate;
  246. m_info.totalThrottle = (int)m_throttleClient.DripRate;
  247. return m_info;
  248. }
  249. /// <summary>
  250. /// Modifies the UDP throttles
  251. /// </summary>
  252. /// <param name="info">New throttling values</param>
  253. public void SetClientInfo(ClientInfo info)
  254. {
  255. // TODO: Allowing throttles to be manually set from this function seems like a reasonable
  256. // idea. On the other hand, letting external code manipulate our ACK accounting is not
  257. // going to happen
  258. throw new NotImplementedException();
  259. }
  260. /// <summary>
  261. /// Get the total number of pakcets queued for this client.
  262. /// </summary>
  263. /// <returns></returns>
  264. public int GetTotalPacketsQueuedCount()
  265. {
  266. int total = 0;
  267. for (int i = 0; i <= (int)ThrottleOutPacketType.Asset; i++)
  268. total += m_packetOutboxes[i].Count;
  269. return total;
  270. }
  271. /// <summary>
  272. /// Get the number of packets queued for the given throttle type.
  273. /// </summary>
  274. /// <returns></returns>
  275. /// <param name="throttleType"></param>
  276. public int GetPacketsQueuedCount(ThrottleOutPacketType throttleType)
  277. {
  278. int icat = (int)throttleType;
  279. if (icat > 0 && icat < THROTTLE_CATEGORY_COUNT)
  280. return m_packetOutboxes[icat].Count;
  281. else
  282. return 0;
  283. }
  284. /// <summary>
  285. /// Return statistics information about client packet queues.
  286. /// </summary>
  287. /// <remarks>
  288. /// FIXME: This should really be done in a more sensible manner rather than sending back a formatted string.
  289. /// </remarks>
  290. /// <returns></returns>
  291. public string GetStats()
  292. {
  293. return string.Format(
  294. "{0,7} {1,7} {2,7} {3,9} {4,7} {5,7} {6,7} {7,7} {8,7} {9,8} {10,7} {11,7}",
  295. Util.EnvironmentTickCountSubtract(TickLastPacketReceived),
  296. PacketsReceived,
  297. PacketsSent,
  298. PacketsResent,
  299. UnackedBytes,
  300. m_packetOutboxes[(int)ThrottleOutPacketType.Resend].Count,
  301. m_packetOutboxes[(int)ThrottleOutPacketType.Land].Count,
  302. m_packetOutboxes[(int)ThrottleOutPacketType.Wind].Count,
  303. m_packetOutboxes[(int)ThrottleOutPacketType.Cloud].Count,
  304. m_packetOutboxes[(int)ThrottleOutPacketType.Task].Count,
  305. m_packetOutboxes[(int)ThrottleOutPacketType.Texture].Count,
  306. m_packetOutboxes[(int)ThrottleOutPacketType.Asset].Count);
  307. }
  308. public void SendPacketStats()
  309. {
  310. PacketStats callback = OnPacketStats;
  311. if (callback is not null)
  312. {
  313. int newPacketsReceived = PacketsReceived - m_packetsReceivedReported;
  314. int newPacketsSent = PacketsSent - m_packetsSentReported;
  315. int newPacketUnAck = UnackedBytes - m_packetsUnAckReported;
  316. callback(newPacketsReceived, newPacketsSent, UnackedBytes);
  317. m_packetsReceivedReported += newPacketsReceived;
  318. m_packetsSentReported += newPacketsSent;
  319. m_packetsUnAckReported += newPacketUnAck;
  320. }
  321. }
  322. public void SetThrottles(byte[] throttleData)
  323. {
  324. SetThrottles(throttleData, 1.0f);
  325. }
  326. public void SetThrottles(byte[] throttleData, float factor)
  327. {
  328. byte[] adjData;
  329. if (!BitConverter.IsLittleEndian)
  330. {
  331. byte[] newData = new byte[7 * 4];
  332. Buffer.BlockCopy(throttleData, 0, newData, 0, 7 * 4);
  333. for (int i = 0; i < 7; i++)
  334. Array.Reverse(newData, i * 4, 4);
  335. adjData = newData;
  336. }
  337. else
  338. {
  339. adjData = throttleData;
  340. }
  341. // 0.125f converts from bits to bytes
  342. float scale = 0.125f * factor;
  343. float resend = Utils.BytesToFloat(adjData) * scale;
  344. float land = Utils.BytesToFloat(adjData, 4) * scale;
  345. float wind = Utils.BytesToFloat(adjData, 8) * scale;
  346. float cloud = Utils.BytesToFloat(adjData, 12) * scale;
  347. float task = Utils.BytesToFloat(adjData, 16) * scale;
  348. float texture = Utils.BytesToFloat(adjData, 20) * scale;
  349. float asset = Utils.BytesToFloat(adjData, 24) * scale;
  350. resend = Math.Clamp(resend, 1000, m_maxRate);
  351. land = Math.Clamp(land, 1000, m_maxRate);
  352. wind = Math.Clamp(wind, 50, m_maxRate);
  353. cloud = Math.Clamp(cloud, 50, m_maxRate);
  354. task = Math.Clamp(task, 1000, m_maxRate);
  355. texture = Math.Clamp(texture, 1000, m_maxRate);
  356. asset = Math.Clamp(asset, 1000, m_maxRate);
  357. float total = resend + land + wind + cloud + task + texture + asset;
  358. if(total > m_maxRate)
  359. {
  360. scale = m_maxRate / total;
  361. resend *= scale;
  362. land *= scale;
  363. wind *= scale;
  364. cloud *= scale;
  365. task *= scale;
  366. texture *= scale;
  367. asset *= scale;
  368. float ntotal = total * scale;
  369. m_log.Debug($"[LLUDPCLIENT]: limiting {AgentID} bandwith from {total} to {ntotal}");
  370. total = ntotal;
  371. }
  372. if (ThrottleDebugLevel > 0)
  373. {
  374. m_log.DebugFormat(
  375. "[LLUDPCLIENT]: {0} is setting throttles in {1} to Resend={2}, Land={3}, Wind={4}, Cloud={5}, Task={6}, Texture={7}, Asset={8}, TOTAL = {9}",
  376. AgentID, m_udpServer.Scene.Name, resend, land, wind, cloud, task, texture, asset, total);
  377. }
  378. TokenBucket bucket;
  379. bucket = m_throttleCategories[(int)ThrottleOutPacketType.Resend];
  380. bucket.RequestedDripRate = resend;
  381. bucket.RequestedBurst = resend * m_burstTime;
  382. bucket = m_throttleCategories[(int)ThrottleOutPacketType.Land];
  383. bucket.RequestedDripRate = land;
  384. bucket.RequestedBurst = land * m_burstTime;
  385. bucket = m_throttleCategories[(int)ThrottleOutPacketType.Wind];
  386. bucket.RequestedDripRate = wind;
  387. bucket.RequestedBurst = wind * m_burstTime;
  388. bucket = m_throttleCategories[(int)ThrottleOutPacketType.Cloud];
  389. bucket.RequestedDripRate = cloud;
  390. bucket.RequestedBurst = cloud * m_burstTime;
  391. bucket = m_throttleCategories[(int)ThrottleOutPacketType.Asset];
  392. bucket.RequestedDripRate = asset;
  393. bucket.RequestedBurst = asset * m_burstTime;
  394. bucket = m_throttleCategories[(int)ThrottleOutPacketType.Task];
  395. bucket.RequestedDripRate = task;
  396. bucket.RequestedBurst = task * m_burstTime;
  397. bucket = m_throttleCategories[(int)ThrottleOutPacketType.Texture];
  398. bucket.RequestedDripRate = texture;
  399. bucket.RequestedBurst = texture * m_burstTime;
  400. }
  401. public byte[] GetThrottlesPacked(float multiplier)
  402. {
  403. byte[] data = new byte[7 * 4];
  404. // multiply by 8 to convert bytes back to bits
  405. multiplier *= 8;
  406. float rate;
  407. rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Resend].RequestedDripRate * multiplier;
  408. Utils.FloatToBytes(rate, data, 0);
  409. rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Land].RequestedDripRate * multiplier;
  410. Utils.FloatToBytes(rate, data, 4);
  411. rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Wind].RequestedDripRate * multiplier;
  412. Utils.FloatToBytes(rate, data, 8);
  413. rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Cloud].RequestedDripRate * multiplier;
  414. Utils.FloatToBytes(rate, data, 12);
  415. rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Task].RequestedDripRate * multiplier;
  416. Utils.FloatToBytes(rate, data, 16);
  417. rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Texture].RequestedDripRate * multiplier;
  418. Utils.FloatToBytes(rate, data, 20);
  419. rate = (float)m_throttleCategories[(int)ThrottleOutPacketType.Asset].RequestedDripRate * multiplier;
  420. Utils.FloatToBytes(rate, data, 24);
  421. return data;
  422. }
  423. public int GetCatBytesCanSend(ThrottleOutPacketType cat, int timeMS)
  424. {
  425. int icat = (int)cat;
  426. if (icat > 0 && icat < THROTTLE_CATEGORY_COUNT)
  427. {
  428. TokenBucket bucket = m_throttleCategories[icat];
  429. return bucket.GetCatBytesCanSend(timeMS);
  430. }
  431. else
  432. return 0;
  433. }
  434. /// <summary>
  435. /// Queue an outgoing packet if appropriate.
  436. /// </summary>
  437. /// <param name="packet"></param>
  438. /// <param name="forceQueue">Always queue the packet if at all possible.</param>
  439. /// <returns>
  440. /// true if the packet has been queued,
  441. /// false if the packet has not been queued and should be sent immediately.
  442. /// </returns>
  443. public bool EnqueueOutgoing(OutgoingPacket packet)
  444. {
  445. int category = (int)packet.Category;
  446. if (category >= 0 && category < m_packetOutboxes.Length)
  447. {
  448. DoubleLocklessQueue<OutgoingPacket> queue = m_packetOutboxes[category];
  449. queue.Enqueue(packet, false);
  450. return true;
  451. }
  452. else
  453. {
  454. // We don't have a token bucket for this category, so it will not be queued
  455. return false;
  456. }
  457. }
  458. /// <summary>
  459. /// Loops through all of the packet queues for this client and tries to send
  460. /// an outgoing packet from each, obeying the throttling bucket limits
  461. /// </summary>
  462. ///
  463. /// <remarks>
  464. /// Packet queues are inspected in ascending numerical order starting from 0. Therefore, queues with a lower
  465. /// ThrottleOutPacketType number will see their packet get sent first (e.g. if both Land and Wind queues have
  466. /// packets, then the packet at the front of the Land queue will be sent before the packet at the front of the
  467. /// wind queue).
  468. ///
  469. /// This function is only called from a synchronous loop in the
  470. /// UDPServer so we don't need to bother making this thread safe
  471. /// </remarks>
  472. ///
  473. /// <returns>True if any packets were sent, otherwise false</returns>
  474. public bool DequeueOutgoing()
  475. {
  476. //if (m_deliverPackets == false) return false;
  477. OutgoingPacket packet;
  478. DoubleLocklessQueue<OutgoingPacket> queue;
  479. bool packetSent = false;
  480. ThrottleOutPacketTypeFlags emptyCategories = 0;
  481. //string queueDebugOutput = String.Empty; // Serious debug business
  482. // do resends
  483. packet = m_nextPackets[0];
  484. if (packet is not null)
  485. {
  486. if (packet.Buffer is not null)
  487. {
  488. if (m_throttleCategories[0].RemoveTokens(packet.Buffer.DataLength))
  489. {
  490. // Send the packet
  491. m_udpServer.SendPacketFinal(packet);
  492. packetSent = true;
  493. m_nextPackets[0] = null;
  494. }
  495. }
  496. else
  497. m_nextPackets[0] = null;
  498. }
  499. else
  500. {
  501. queue = m_packetOutboxes[0];
  502. if (queue is not null)
  503. {
  504. if(queue.Dequeue(out packet))
  505. {
  506. // A packet was pulled off the queue. See if we have
  507. // enough tokens in the bucket to send it out
  508. if (packet.Buffer is not null)
  509. {
  510. if (m_throttleCategories[0].RemoveTokens(packet.Buffer.DataLength))
  511. {
  512. // Send the packet
  513. m_udpServer.SendPacketFinal(packet);
  514. packetSent = true;
  515. }
  516. else
  517. {
  518. // Save the dequeued packet for the next iteration
  519. m_nextPackets[0] = packet;
  520. }
  521. }
  522. }
  523. }
  524. else
  525. {
  526. m_packetOutboxes[0] = new DoubleLocklessQueue<OutgoingPacket>();
  527. }
  528. }
  529. if(NeedAcks.Count() > 50)
  530. {
  531. Interlocked.Increment(ref AckStalls);
  532. return true;
  533. }
  534. for (int i = 1; i < THROTTLE_CATEGORY_COUNT; i++)
  535. {
  536. //queueDebugOutput += m_packetOutboxes[i].Count + " "; // Serious debug business
  537. packet = m_nextPackets[i];
  538. if (packet is not null)
  539. {
  540. if(packet.Buffer is null)
  541. {
  542. if (m_packetOutboxes[i].Count < 5)
  543. emptyCategories |= CategoryToFlag(i);
  544. m_nextPackets[i] = null;
  545. continue;
  546. }
  547. if (m_throttleCategories[i].RemoveTokens(packet.Buffer.DataLength))
  548. {
  549. // Send the packet
  550. m_udpServer.SendPacketFinal(packet);
  551. m_nextPackets[i] = null;
  552. packetSent = true;
  553. if (m_packetOutboxes[i].Count < 5)
  554. emptyCategories |= CategoryToFlag(i);
  555. }
  556. }
  557. else
  558. {
  559. // No dequeued packet waiting to be sent, try to pull one off
  560. // this queue
  561. queue = m_packetOutboxes[i];
  562. if(queue.Dequeue(out packet))
  563. {
  564. if (packet.Buffer is null)
  565. {
  566. // packet canceled elsewhere (by a ack for example)
  567. if (queue.Count < 5)
  568. emptyCategories |= CategoryToFlag(i);
  569. continue;
  570. }
  571. if (m_throttleCategories[i].RemoveTokens(packet.Buffer.DataLength))
  572. {
  573. // Send the packet
  574. m_udpServer.SendPacketFinal(packet);
  575. packetSent = true;
  576. if (queue.Count < 5)
  577. emptyCategories |= CategoryToFlag(i);
  578. }
  579. else
  580. {
  581. // Save the dequeued packet for the next iteration
  582. m_nextPackets[i] = packet;
  583. }
  584. }
  585. else
  586. {
  587. // No packets in this queue. Fire the queue empty callback
  588. // if it has not been called recently
  589. emptyCategories |= CategoryToFlag(i);
  590. }
  591. }
  592. }
  593. if (emptyCategories != 0)
  594. BeginFireQueueEmpty(emptyCategories);
  595. //m_log.Info("[LLUDPCLIENT]: Queues: " + queueDebugOutput); // Serious debug business
  596. return packetSent;
  597. }
  598. /// <summary>
  599. /// Called when we get a ping update
  600. /// </summary>
  601. /// <param name="r"> ping time in ms
  602. /// acknowledgement</param>
  603. public void UpdateRoundTrip(int p)
  604. {
  605. p *= 5;
  606. if( p> m_maxRTO)
  607. p = m_maxRTO;
  608. else if(p < m_minRTO)
  609. p = m_minRTO;
  610. m_RTO = p;
  611. }
  612. const double MIN_CALLBACK_MS = 20.0;
  613. public bool QueueEmptyRunning;
  614. /// <summary>
  615. /// Does an early check to see if this queue empty callback is already
  616. /// running, then asynchronously firing the event
  617. /// </summary>
  618. /// <param name="categories">Throttle categories to fire the callback for</param>
  619. private void BeginFireQueueEmpty(ThrottleOutPacketTypeFlags categories)
  620. {
  621. if (!QueueEmptyRunning && HasUpdates(categories) && OnQueueEmpty is not null)
  622. {
  623. double start = Util.GetTimeStampMS();
  624. if (start < m_nextOnQueueEmpty)
  625. return;
  626. QueueEmptyRunning = true;
  627. m_nextOnQueueEmpty = start + MIN_CALLBACK_MS;
  628. // Asynchronously run the callback
  629. if (m_udpServer.OqrEngine.IsRunning)
  630. {
  631. LLUDPClient udpcli = this;
  632. ThrottleOutPacketTypeFlags cats = categories;
  633. void act(LLUDPClient arg1, ThrottleOutPacketTypeFlags arg2)
  634. {
  635. QueueEmpty callback = udpcli.OnQueueEmpty;
  636. if (callback != null)
  637. {
  638. try
  639. {
  640. callback(cats);
  641. }
  642. catch { }
  643. }
  644. udpcli.QueueEmptyRunning = false;
  645. udpcli = null;
  646. callback = null;
  647. }
  648. _ = m_udpServer.OqrEngine.QueueJob(AgentID.ToString(), () => act(udpcli, cats));
  649. }
  650. else
  651. Util.FireAndForget(FireQueueEmpty, categories, "LLUDPClient.BeginFireQueueEmpty");
  652. }
  653. }
  654. /// <summary>
  655. /// Fires the OnQueueEmpty callback and sets the minimum time that it
  656. /// can be called again
  657. /// </summary>
  658. /// <param name="o">Throttle categories to fire the callback for,
  659. /// stored as an object to match the WaitCallback delegate
  660. /// signature</param>
  661. public void FireQueueEmpty(object o)
  662. {
  663. QueueEmpty callback = OnQueueEmpty;
  664. if (callback is not null)
  665. {
  666. ThrottleOutPacketTypeFlags categories = (ThrottleOutPacketTypeFlags)o;
  667. try { callback(categories); }
  668. catch (Exception e) { m_log.Error("[LLUDPCLIENT]: OnQueueEmpty(" + categories + ") threw an exception: " + e.Message, e); }
  669. }
  670. QueueEmptyRunning = false;
  671. }
  672. internal void ForceThrottleSetting(int throttle, int setting)
  673. {
  674. if (throttle > 0 && throttle < THROTTLE_CATEGORY_COUNT)
  675. m_throttleCategories[throttle].RequestedDripRate = Math.Max(setting, LLUDPServer.MTU);
  676. }
  677. internal int GetThrottleSetting(int throttle)
  678. {
  679. if (throttle > 0 && throttle < THROTTLE_CATEGORY_COUNT)
  680. return (int)m_throttleCategories[throttle].RequestedDripRate;
  681. else
  682. return 0;
  683. }
  684. public void FreeUDPBuffer(UDPPacketBuffer buf)
  685. {
  686. m_udpServer.FreeUDPBuffer(buf);
  687. }
  688. /// <summary>
  689. /// Converts a <seealso cref="ThrottleOutPacketType"/> integer to a
  690. /// flag value
  691. /// </summary>
  692. /// <param name="i">Throttle category to convert</param>
  693. /// <returns>Flag representation of the throttle category</returns>
  694. private static ThrottleOutPacketTypeFlags CategoryToFlag(int i)
  695. {
  696. ThrottleOutPacketType category = (ThrottleOutPacketType)i;
  697. return category switch
  698. {
  699. ThrottleOutPacketType.Land => ThrottleOutPacketTypeFlags.Land,// Terrain data
  700. ThrottleOutPacketType.Wind => ThrottleOutPacketTypeFlags.Wind,// Wind data
  701. ThrottleOutPacketType.Cloud => ThrottleOutPacketTypeFlags.Cloud,// Cloud data
  702. ThrottleOutPacketType.Task => ThrottleOutPacketTypeFlags.Task,// Object updates and everything not on the other categories
  703. ThrottleOutPacketType.Texture => ThrottleOutPacketTypeFlags.Texture,// Textures data (also impacts http texture and mesh by default)
  704. ThrottleOutPacketType.Asset => ThrottleOutPacketTypeFlags.Asset,// Non-texture Assets data
  705. _ => 0,
  706. };
  707. }
  708. }
  709. public class DoubleLocklessQueue<T> : OpenSim.Framework.LocklessQueue<T>
  710. {
  711. readonly OpenSim.Framework.LocklessQueue<T> highQueue = new();
  712. public override int Count
  713. {
  714. get
  715. {
  716. return base.Count + highQueue.Count;
  717. }
  718. }
  719. public override bool Dequeue(out T item)
  720. {
  721. if (highQueue.Dequeue(out item))
  722. return true;
  723. return base.Dequeue(out item);
  724. }
  725. public void Enqueue(T item, bool highPriority)
  726. {
  727. if (highPriority)
  728. highQueue.Enqueue(item);
  729. else
  730. Enqueue(item);
  731. }
  732. }
  733. }