aboutsummaryrefslogtreecommitdiffstatshomepage
diff options
context:
space:
mode:
authorJustin Clark-Casey (justincc)2014-08-19 00:11:04 +0100
committerJustin Clark-Casey (justincc)2014-08-19 00:17:12 +0100
commit84cea46c10e3f44c4d869d439a01ca7f80b56ece (patch)
tree3dbbbebb44fba4c2ac45046ed698752decf79d42
parentMake LLUDPServer.Scene publicly gettable/privately settable instead of protec... (diff)
downloadopensim-SC_OLD-84cea46c10e3f44c4d869d439a01ca7f80b56ece.zip
opensim-SC_OLD-84cea46c10e3f44c4d869d439a01ca7f80b56ece.tar.gz
opensim-SC_OLD-84cea46c10e3f44c4d869d439a01ca7f80b56ece.tar.bz2
opensim-SC_OLD-84cea46c10e3f44c4d869d439a01ca7f80b56ece.tar.xz
Add experimental OutgoingQueueRefillEngine to handle queue refill processing on a controlled number of threads rather than the threadpool.
Disabled by default. Currently can only be enabled with console "debug lludp oqre start" command, though this can be started and stopped whilst simulator is running. When a connection requires packet queue refill processing (used to populate queues with entity updates, entity prop updates and image queue updates), this is done via Threadpool requests. However, with a very high number of connections (e.g. 100 root + 300 child) a very large number of simultaneous requests may be causing performance issues. This commit adds an experimental engine for processing these requests from a queue with a persistent thread instead. Unlike inbound processing, there are no network requests in this processing that might hold the thread up for a long time. Early implementation - currently only one thread which may (or may not) get overloaded with requests. Added for testing purposes.
-rw-r--r--OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs13
-rw-r--r--OpenSim/Region/ClientStack/Linden/UDP/LLUDPServer.cs13
-rw-r--r--OpenSim/Region/ClientStack/Linden/UDP/OutgoingQueueRefillEngine.cs267
3 files changed, 288 insertions, 5 deletions
diff --git a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs
index 2f363f4..c34bafa 100644
--- a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs
+++ b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs
@@ -650,8 +650,15 @@ namespace OpenSim.Region.ClientStack.LindenUDP
650 650
651 if (HasUpdates(m_categories)) 651 if (HasUpdates(m_categories))
652 { 652 {
653 // Asynchronously run the callback 653 if (!m_udpServer.OqrEngine.IsRunning)
654 Util.FireAndForget(FireQueueEmpty, categories); 654 {
655 // Asynchronously run the callback
656 Util.FireAndForget(FireQueueEmpty, categories);
657 }
658 else
659 {
660 m_udpServer.OqrEngine.QueueRequest(this, categories);
661 }
655 } 662 }
656 else 663 else
657 { 664 {
@@ -670,7 +677,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP
670 /// <param name="o">Throttle categories to fire the callback for, 677 /// <param name="o">Throttle categories to fire the callback for,
671 /// stored as an object to match the WaitCallback delegate 678 /// stored as an object to match the WaitCallback delegate
672 /// signature</param> 679 /// signature</param>
673 private void FireQueueEmpty(object o) 680 public void FireQueueEmpty(object o)
674 { 681 {
675// m_log.DebugFormat("[LLUDPCLIENT]: FireQueueEmpty for {0} in {1}", AgentID, m_udpServer.Scene.Name); 682// m_log.DebugFormat("[LLUDPCLIENT]: FireQueueEmpty for {0} in {1}", AgentID, m_udpServer.Scene.Name);
676 683
diff --git a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServer.cs b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServer.cs
index 465c86b..3c9bb1b 100644
--- a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServer.cs
+++ b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServer.cs
@@ -355,6 +355,13 @@ namespace OpenSim.Region.ClientStack.LindenUDP
355 /// </summary> 355 /// </summary>
356 private IClientAPI m_currentIncomingClient; 356 private IClientAPI m_currentIncomingClient;
357 357
358 /// <summary>
359 /// Experimental facility to run queue empty processing within a controlled number of threads rather than
360 /// requiring massive numbers of short-lived threads from the threadpool when there are a high number of
361 /// connections.
362 /// </summary>
363 public OutgoingQueueRefillEngine OqrEngine { get; private set; }
364
358 public LLUDPServer( 365 public LLUDPServer(
359 IPAddress listenIP, ref uint port, int proxyPortOffsetParm, bool allow_alternate_port, 366 IPAddress listenIP, ref uint port, int proxyPortOffsetParm, bool allow_alternate_port,
360 IConfigSource configSource, AgentCircuitManager circuitManager) 367 IConfigSource configSource, AgentCircuitManager circuitManager)
@@ -432,6 +439,8 @@ namespace OpenSim.Region.ClientStack.LindenUDP
432 439
433 if (usePools) 440 if (usePools)
434 EnablePools(); 441 EnablePools();
442
443 OqrEngine = new OutgoingQueueRefillEngine(this);
435 } 444 }
436 445
437 public void Start() 446 public void Start()
@@ -469,7 +478,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP
469 478
470 Watchdog.StartThread( 479 Watchdog.StartThread(
471 OutgoingPacketHandler, 480 OutgoingPacketHandler,
472 string.Format("Outgoing Packets ({0})", Scene.RegionInfo.RegionName), 481 string.Format("Outgoing Packets ({0})", Scene.Name),
473 ThreadPriority.Normal, 482 ThreadPriority.Normal,
474 false, 483 false,
475 true, 484 true,
@@ -479,7 +488,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP
479 488
480 public void Stop() 489 public void Stop()
481 { 490 {
482 m_log.Info("[LLUDPSERVER]: Shutting down the LLUDP server for " + Scene.RegionInfo.RegionName); 491 m_log.Info("[LLUDPSERVER]: Shutting down the LLUDP server for " + Scene.Name);
483 base.StopOutbound(); 492 base.StopOutbound();
484 base.StopInbound(); 493 base.StopInbound();
485 } 494 }
diff --git a/OpenSim/Region/ClientStack/Linden/UDP/OutgoingQueueRefillEngine.cs b/OpenSim/Region/ClientStack/Linden/UDP/OutgoingQueueRefillEngine.cs
new file mode 100644
index 0000000..8777402
--- /dev/null
+++ b/OpenSim/Region/ClientStack/Linden/UDP/OutgoingQueueRefillEngine.cs
@@ -0,0 +1,267 @@
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
28using System;
29using System.Collections.Concurrent;
30using System.Reflection;
31using System.Threading;
32using log4net;
33using OpenSim.Framework;
34using OpenSim.Framework.Monitoring;
35using OpenSim.Region.Framework.Scenes;
36
37namespace OpenSim.Region.ClientStack.LindenUDP
38{
39 public struct RefillRequest
40 {
41 public LLUDPClient Client;
42 public ThrottleOutPacketTypeFlags Categories;
43
44 public RefillRequest(LLUDPClient client, ThrottleOutPacketTypeFlags categories)
45 {
46 Client = client;
47 Categories = categories;
48 }
49 }
50
51 public class OutgoingQueueRefillEngine
52 {
53 private static readonly ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
54
55 public bool IsRunning { get; private set; }
56
57 /// <summary>
58 /// The timeout in milliseconds to wait for at least one event to be written when the recorder is stopping.
59 /// </summary>
60 public int RequestProcessTimeoutOnStop { get; set; }
61
62 /// <summary>
63 /// Controls whether we need to warn in the log about exceeding the max queue size.
64 /// </summary>
65 /// <remarks>
66 /// This is flipped to false once queue max has been exceeded and back to true when it falls below max, in
67 /// order to avoid spamming the log with lots of warnings.
68 /// </remarks>
69 private bool m_warnOverMaxQueue = true;
70
71 private BlockingCollection<RefillRequest> m_requestQueue;
72
73 private CancellationTokenSource m_cancelSource = new CancellationTokenSource();
74
75 private LLUDPServer m_udpServer;
76
77 /// <summary>
78 /// Used to signal that we are ready to complete stop.
79 /// </summary>
80 private ManualResetEvent m_finishedProcessingAfterStop = new ManualResetEvent(false);
81
82 public OutgoingQueueRefillEngine(LLUDPServer server)
83 {
84 RequestProcessTimeoutOnStop = 5000;
85 m_udpServer = server;
86
87 MainConsole.Instance.Commands.AddCommand(
88 "Debug",
89 false,
90 "debug lludp oqre",
91 "debug lludp oqre <start|stop|status>",
92 "Start, stop or get status of OutgoingQueueRefillEngine.",
93 "Experimental.",
94 HandleOqreCommand);
95 }
96
97 public void Start()
98 {
99 lock (this)
100 {
101 if (IsRunning)
102 return;
103
104 IsRunning = true;
105
106 m_finishedProcessingAfterStop.Reset();
107
108 m_requestQueue = new BlockingCollection<RefillRequest>(new ConcurrentQueue<RefillRequest>(), 5000);
109
110 Watchdog.StartThread(
111 ProcessRequests,
112 String.Format("OutgoingQueueRefillEngineThread ({0})", m_udpServer.Scene.Name),
113 ThreadPriority.Normal,
114 false,
115 true,
116 null,
117 int.MaxValue);
118 }
119 }
120
121 public void Stop()
122 {
123 lock (this)
124 {
125 try
126 {
127 if (!IsRunning)
128 return;
129
130 IsRunning = false;
131
132 int requestsLeft = m_requestQueue.Count;
133
134 if (requestsLeft <= 0)
135 {
136 m_cancelSource.Cancel();
137 }
138 else
139 {
140 m_log.InfoFormat("[OUTGOING QUEUE REFILL ENGINE]: Waiting to write {0} events after stop.", requestsLeft);
141
142 while (requestsLeft > 0)
143 {
144 if (!m_finishedProcessingAfterStop.WaitOne(RequestProcessTimeoutOnStop))
145 {
146 // After timeout no events have been written
147 if (requestsLeft == m_requestQueue.Count)
148 {
149 m_log.WarnFormat(
150 "[OUTGOING QUEUE REFILL ENGINE]: No requests processed after {0} ms wait. Discarding remaining {1} requests",
151 RequestProcessTimeoutOnStop, requestsLeft);
152
153 break;
154 }
155 }
156
157 requestsLeft = m_requestQueue.Count;
158 }
159 }
160 }
161 finally
162 {
163 m_cancelSource.Dispose();
164 m_requestQueue = null;
165 }
166 }
167 }
168
169 public bool QueueRequest(LLUDPClient client, ThrottleOutPacketTypeFlags categories)
170 {
171 if (m_requestQueue.Count < m_requestQueue.BoundedCapacity)
172 {
173// m_log.DebugFormat(
174// "[OUTGOING QUEUE REFILL ENGINE]: Adding request for categories {0} for {1} in {2}",
175// categories, client.AgentID, m_udpServer.Scene.Name);
176
177 m_requestQueue.Add(new RefillRequest(client, categories));
178
179 if (!m_warnOverMaxQueue)
180 m_warnOverMaxQueue = true;
181
182 return true;
183 }
184 else
185 {
186 if (m_warnOverMaxQueue)
187 {
188 m_log.WarnFormat(
189 "[OUTGOING QUEUE REFILL ENGINE]: Request queue at maximum capacity, not recording request from {0} in {1}",
190 client.AgentID, m_udpServer.Scene.Name);
191
192 m_warnOverMaxQueue = false;
193 }
194
195 return false;
196 }
197 }
198
199 private void ProcessRequests()
200 {
201 try
202 {
203 while (IsRunning || m_requestQueue.Count > 0)
204 {
205 RefillRequest req = m_requestQueue.Take(m_cancelSource.Token);
206
207 // QueueEmpty callback = req.Client.OnQueueEmpty;
208 //
209 // if (callback != null)
210 // {
211 // try
212 // {
213 // callback(req.Categories);
214 // }
215 // catch (Exception e)
216 // {
217 // m_log.Error("[OUTGOING QUEUE REFILL ENGINE]: ProcessRequests(" + req.Categories + ") threw an exception: " + e.Message, e);
218 // }
219 // }
220
221 req.Client.FireQueueEmpty(req.Categories);
222 }
223 }
224 catch (OperationCanceledException)
225 {
226 }
227
228 m_finishedProcessingAfterStop.Set();
229 }
230
231 private void HandleOqreCommand(string module, string[] args)
232 {
233 if (SceneManager.Instance.CurrentScene != null && SceneManager.Instance.CurrentScene != m_udpServer.Scene)
234 return;
235
236 if (args.Length != 4)
237 {
238 MainConsole.Instance.Output("Usage: debug lludp oqre <stop|start|status>");
239 return;
240 }
241
242 string subCommand = args[3];
243
244 if (subCommand == "stop")
245 {
246 Stop();
247 MainConsole.Instance.OutputFormat("Stopped OQRE for {0}", m_udpServer.Scene.Name);
248 }
249 else if (subCommand == "start")
250 {
251 Start();
252 MainConsole.Instance.OutputFormat("Started OQRE for {0}", m_udpServer.Scene.Name);
253 }
254 else if (subCommand == "status")
255 {
256 MainConsole.Instance.OutputFormat("OQRE in {0}", m_udpServer.Scene.Name);
257 MainConsole.Instance.OutputFormat("Running: {0}", IsRunning);
258 MainConsole.Instance.OutputFormat(
259 "Requests waiting: {0}", IsRunning ? m_requestQueue.Count.ToString() : "n/a");
260 }
261 else
262 {
263 MainConsole.Instance.OutputFormat("Unrecognized OQRE subcommand {0}", subCommand);
264 }
265 }
266 }
267} \ No newline at end of file