diff options
Diffstat (limited to '')
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 | |||
28 | using System; | ||
29 | using System.Collections.Concurrent; | ||
30 | using System.Reflection; | ||
31 | using System.Threading; | ||
32 | using log4net; | ||
33 | using OpenSim.Framework; | ||
34 | using OpenSim.Framework.Monitoring; | ||
35 | using OpenSim.Region.Framework.Scenes; | ||
36 | |||
37 | namespace 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 | ||