diff options
author | Justin Clark-Casey (justincc) | 2015-01-12 20:56:37 +0000 |
---|---|---|
committer | Justin Clark-Casey (justincc) | 2015-01-12 20:56:37 +0000 |
commit | 8e1e8a0920a9e94305619e9afb8e053b4daefb89 (patch) | |
tree | cc6cb595807393c465f520cd1a1fc7a8aab33b2d /OpenSim | |
parent | SimulatorFeatures: the viewer also takes GridName in OpenSim extras. Added th... (diff) | |
download | opensim-SC-8e1e8a0920a9e94305619e9afb8e053b4daefb89.zip opensim-SC-8e1e8a0920a9e94305619e9afb8e053b4daefb89.tar.gz opensim-SC-8e1e8a0920a9e94305619e9afb8e053b4daefb89.tar.bz2 opensim-SC-8e1e8a0920a9e94305619e9afb8e053b4daefb89.tar.xz |
Make the performance controlling job processing threads introduced in conference code use a generic JobEngine class rather than 4 slightly different copy/pasted versions.
Diffstat (limited to '')
10 files changed, 195 insertions, 1307 deletions
diff --git a/OpenSim/Framework/Monitoring/JobEngine.cs b/OpenSim/Framework/Monitoring/JobEngine.cs deleted file mode 100644 index 5925867..0000000 --- a/OpenSim/Framework/Monitoring/JobEngine.cs +++ /dev/null | |||
@@ -1,320 +0,0 @@ | |||
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 | |||
35 | namespace OpenSim.Framework.Monitoring | ||
36 | { | ||
37 | public class Job | ||
38 | { | ||
39 | public string Name; | ||
40 | public WaitCallback Callback; | ||
41 | public object O; | ||
42 | |||
43 | public Job(string name, WaitCallback callback, object o) | ||
44 | { | ||
45 | Name = name; | ||
46 | Callback = callback; | ||
47 | O = o; | ||
48 | } | ||
49 | } | ||
50 | |||
51 | public class JobEngine | ||
52 | { | ||
53 | private static readonly ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType); | ||
54 | |||
55 | public int LogLevel { get; set; } | ||
56 | |||
57 | public bool IsRunning { get; private set; } | ||
58 | |||
59 | /// <summary> | ||
60 | /// The timeout in milliseconds to wait for at least one event to be written when the recorder is stopping. | ||
61 | /// </summary> | ||
62 | public int RequestProcessTimeoutOnStop { get; set; } | ||
63 | |||
64 | /// <summary> | ||
65 | /// Controls whether we need to warn in the log about exceeding the max queue size. | ||
66 | /// </summary> | ||
67 | /// <remarks> | ||
68 | /// This is flipped to false once queue max has been exceeded and back to true when it falls below max, in | ||
69 | /// order to avoid spamming the log with lots of warnings. | ||
70 | /// </remarks> | ||
71 | private bool m_warnOverMaxQueue = true; | ||
72 | |||
73 | private BlockingCollection<Job> m_requestQueue; | ||
74 | |||
75 | private CancellationTokenSource m_cancelSource = new CancellationTokenSource(); | ||
76 | |||
77 | private Stat m_requestsWaitingStat; | ||
78 | |||
79 | private Job m_currentJob; | ||
80 | |||
81 | /// <summary> | ||
82 | /// Used to signal that we are ready to complete stop. | ||
83 | /// </summary> | ||
84 | private ManualResetEvent m_finishedProcessingAfterStop = new ManualResetEvent(false); | ||
85 | |||
86 | public JobEngine() | ||
87 | { | ||
88 | RequestProcessTimeoutOnStop = 5000; | ||
89 | |||
90 | MainConsole.Instance.Commands.AddCommand( | ||
91 | "Debug", | ||
92 | false, | ||
93 | "debug jobengine", | ||
94 | "debug jobengine <start|stop|status|log>", | ||
95 | "Start, stop, get status or set logging level of the job engine.", | ||
96 | "If stopped then all outstanding jobs are processed immediately.", | ||
97 | HandleControlCommand); | ||
98 | } | ||
99 | |||
100 | public void Start() | ||
101 | { | ||
102 | lock (this) | ||
103 | { | ||
104 | if (IsRunning) | ||
105 | return; | ||
106 | |||
107 | IsRunning = true; | ||
108 | |||
109 | m_finishedProcessingAfterStop.Reset(); | ||
110 | |||
111 | m_requestQueue = new BlockingCollection<Job>(new ConcurrentQueue<Job>(), 5000); | ||
112 | |||
113 | m_requestsWaitingStat = | ||
114 | new Stat( | ||
115 | "JobsWaiting", | ||
116 | "Number of jobs waiting for processing.", | ||
117 | "", | ||
118 | "", | ||
119 | "server", | ||
120 | "jobengine", | ||
121 | StatType.Pull, | ||
122 | MeasuresOfInterest.None, | ||
123 | stat => stat.Value = m_requestQueue.Count, | ||
124 | StatVerbosity.Debug); | ||
125 | |||
126 | StatsManager.RegisterStat(m_requestsWaitingStat); | ||
127 | |||
128 | WorkManager.StartThread( | ||
129 | ProcessRequests, | ||
130 | "JobEngineThread", | ||
131 | ThreadPriority.Normal, | ||
132 | false, | ||
133 | true, | ||
134 | null, | ||
135 | int.MaxValue); | ||
136 | } | ||
137 | } | ||
138 | |||
139 | public void Stop() | ||
140 | { | ||
141 | lock (this) | ||
142 | { | ||
143 | try | ||
144 | { | ||
145 | if (!IsRunning) | ||
146 | return; | ||
147 | |||
148 | IsRunning = false; | ||
149 | |||
150 | int requestsLeft = m_requestQueue.Count; | ||
151 | |||
152 | if (requestsLeft <= 0) | ||
153 | { | ||
154 | m_cancelSource.Cancel(); | ||
155 | } | ||
156 | else | ||
157 | { | ||
158 | m_log.InfoFormat("[JOB ENGINE]: Waiting to write {0} events after stop.", requestsLeft); | ||
159 | |||
160 | while (requestsLeft > 0) | ||
161 | { | ||
162 | if (!m_finishedProcessingAfterStop.WaitOne(RequestProcessTimeoutOnStop)) | ||
163 | { | ||
164 | // After timeout no events have been written | ||
165 | if (requestsLeft == m_requestQueue.Count) | ||
166 | { | ||
167 | m_log.WarnFormat( | ||
168 | "[JOB ENGINE]: No requests processed after {0} ms wait. Discarding remaining {1} requests", | ||
169 | RequestProcessTimeoutOnStop, requestsLeft); | ||
170 | |||
171 | break; | ||
172 | } | ||
173 | } | ||
174 | |||
175 | requestsLeft = m_requestQueue.Count; | ||
176 | } | ||
177 | } | ||
178 | } | ||
179 | finally | ||
180 | { | ||
181 | m_cancelSource.Dispose(); | ||
182 | StatsManager.DeregisterStat(m_requestsWaitingStat); | ||
183 | m_requestsWaitingStat = null; | ||
184 | m_requestQueue = null; | ||
185 | } | ||
186 | } | ||
187 | } | ||
188 | |||
189 | public bool QueueRequest(string name, WaitCallback req, object o) | ||
190 | { | ||
191 | if (LogLevel >= 1) | ||
192 | m_log.DebugFormat("[JOB ENGINE]: Queued job {0}", name); | ||
193 | |||
194 | if (m_requestQueue.Count < m_requestQueue.BoundedCapacity) | ||
195 | { | ||
196 | // m_log.DebugFormat( | ||
197 | // "[OUTGOING QUEUE REFILL ENGINE]: Adding request for categories {0} for {1} in {2}", | ||
198 | // categories, client.AgentID, m_udpServer.Scene.Name); | ||
199 | |||
200 | m_requestQueue.Add(new Job(name, req, o)); | ||
201 | |||
202 | if (!m_warnOverMaxQueue) | ||
203 | m_warnOverMaxQueue = true; | ||
204 | |||
205 | return true; | ||
206 | } | ||
207 | else | ||
208 | { | ||
209 | if (m_warnOverMaxQueue) | ||
210 | { | ||
211 | // m_log.WarnFormat( | ||
212 | // "[JOB ENGINE]: Request queue at maximum capacity, not recording request from {0} in {1}", | ||
213 | // client.AgentID, m_udpServer.Scene.Name); | ||
214 | |||
215 | m_log.WarnFormat("[JOB ENGINE]: Request queue at maximum capacity, not recording job"); | ||
216 | |||
217 | m_warnOverMaxQueue = false; | ||
218 | } | ||
219 | |||
220 | return false; | ||
221 | } | ||
222 | } | ||
223 | |||
224 | private void ProcessRequests() | ||
225 | { | ||
226 | try | ||
227 | { | ||
228 | while (IsRunning || m_requestQueue.Count > 0) | ||
229 | { | ||
230 | m_currentJob = m_requestQueue.Take(m_cancelSource.Token); | ||
231 | |||
232 | // QueueEmpty callback = req.Client.OnQueueEmpty; | ||
233 | // | ||
234 | // if (callback != null) | ||
235 | // { | ||
236 | // try | ||
237 | // { | ||
238 | // callback(req.Categories); | ||
239 | // } | ||
240 | // catch (Exception e) | ||
241 | // { | ||
242 | // m_log.Error("[OUTGOING QUEUE REFILL ENGINE]: ProcessRequests(" + req.Categories + ") threw an exception: " + e.Message, e); | ||
243 | // } | ||
244 | // } | ||
245 | |||
246 | if (LogLevel >= 1) | ||
247 | m_log.DebugFormat("[JOB ENGINE]: Processing job {0}", m_currentJob.Name); | ||
248 | |||
249 | try | ||
250 | { | ||
251 | m_currentJob.Callback.Invoke(m_currentJob.O); | ||
252 | } | ||
253 | catch (Exception e) | ||
254 | { | ||
255 | m_log.Error( | ||
256 | string.Format( | ||
257 | "[JOB ENGINE]: Job {0} failed, continuing. Exception ", m_currentJob.Name), e); | ||
258 | } | ||
259 | |||
260 | if (LogLevel >= 1) | ||
261 | m_log.DebugFormat("[JOB ENGINE]: Processed job {0}", m_currentJob.Name); | ||
262 | |||
263 | m_currentJob = null; | ||
264 | } | ||
265 | } | ||
266 | catch (OperationCanceledException) | ||
267 | { | ||
268 | } | ||
269 | |||
270 | m_finishedProcessingAfterStop.Set(); | ||
271 | } | ||
272 | |||
273 | private void HandleControlCommand(string module, string[] args) | ||
274 | { | ||
275 | // if (SceneManager.Instance.CurrentScene != null && SceneManager.Instance.CurrentScene != m_udpServer.Scene) | ||
276 | // return; | ||
277 | |||
278 | if (args.Length < 3) | ||
279 | { | ||
280 | MainConsole.Instance.Output("Usage: debug jobengine <stop|start|status|log>"); | ||
281 | return; | ||
282 | } | ||
283 | |||
284 | string subCommand = args[2]; | ||
285 | |||
286 | if (subCommand == "stop") | ||
287 | { | ||
288 | Stop(); | ||
289 | MainConsole.Instance.OutputFormat("Stopped job engine."); | ||
290 | } | ||
291 | else if (subCommand == "start") | ||
292 | { | ||
293 | Start(); | ||
294 | MainConsole.Instance.OutputFormat("Started job engine."); | ||
295 | } | ||
296 | else if (subCommand == "status") | ||
297 | { | ||
298 | MainConsole.Instance.OutputFormat("Job engine running: {0}", IsRunning); | ||
299 | MainConsole.Instance.OutputFormat("Current job {0}", m_currentJob != null ? m_currentJob.Name : "none"); | ||
300 | MainConsole.Instance.OutputFormat( | ||
301 | "Jobs waiting: {0}", IsRunning ? m_requestQueue.Count.ToString() : "n/a"); | ||
302 | MainConsole.Instance.OutputFormat("Log Level: {0}", LogLevel); | ||
303 | } | ||
304 | else if (subCommand == "log") | ||
305 | { | ||
306 | // int logLevel; | ||
307 | int logLevel = int.Parse(args[3]); | ||
308 | // if (ConsoleUtil.TryParseConsoleInt(MainConsole.Instance, args[4], out logLevel)) | ||
309 | // { | ||
310 | LogLevel = logLevel; | ||
311 | MainConsole.Instance.OutputFormat("Set debug log level to {0}", LogLevel); | ||
312 | // } | ||
313 | } | ||
314 | else | ||
315 | { | ||
316 | MainConsole.Instance.OutputFormat("Unrecognized job engine subcommand {0}", subCommand); | ||
317 | } | ||
318 | } | ||
319 | } | ||
320 | } | ||
diff --git a/OpenSim/Framework/Monitoring/WorkManager.cs b/OpenSim/Framework/Monitoring/WorkManager.cs index 9d0eefc..134661b 100644 --- a/OpenSim/Framework/Monitoring/WorkManager.cs +++ b/OpenSim/Framework/Monitoring/WorkManager.cs | |||
@@ -57,7 +57,29 @@ namespace OpenSim.Framework.Monitoring | |||
57 | 57 | ||
58 | static WorkManager() | 58 | static WorkManager() |
59 | { | 59 | { |
60 | JobEngine = new JobEngine(); | 60 | JobEngine = new JobEngine("Non-blocking non-critical job engine", "JOB ENGINE"); |
61 | |||
62 | StatsManager.RegisterStat( | ||
63 | new Stat( | ||
64 | "JobsWaiting", | ||
65 | "Number of jobs waiting for processing.", | ||
66 | "", | ||
67 | "", | ||
68 | "server", | ||
69 | "jobengine", | ||
70 | StatType.Pull, | ||
71 | MeasuresOfInterest.None, | ||
72 | stat => stat.Value = JobEngine.JobsWaiting, | ||
73 | StatVerbosity.Debug)); | ||
74 | |||
75 | MainConsole.Instance.Commands.AddCommand( | ||
76 | "Debug", | ||
77 | false, | ||
78 | "debug jobengine", | ||
79 | "debug jobengine <start|stop|status|log>", | ||
80 | "Start, stop, get status or set logging level of the job engine.", | ||
81 | "If stopped then all outstanding jobs are processed immediately.", | ||
82 | HandleControlCommand); | ||
61 | } | 83 | } |
62 | 84 | ||
63 | /// <summary> | 85 | /// <summary> |
@@ -200,7 +222,7 @@ namespace OpenSim.Framework.Monitoring | |||
200 | } | 222 | } |
201 | 223 | ||
202 | if (JobEngine.IsRunning) | 224 | if (JobEngine.IsRunning) |
203 | JobEngine.QueueRequest(name, callback, obj); | 225 | JobEngine.QueueJob(name, () => callback(obj)); |
204 | else if (canRunInThisThread) | 226 | else if (canRunInThisThread) |
205 | callback(obj); | 227 | callback(obj); |
206 | else if (mustNotTimeout) | 228 | else if (mustNotTimeout) |
@@ -208,5 +230,55 @@ namespace OpenSim.Framework.Monitoring | |||
208 | else | 230 | else |
209 | Util.FireAndForget(callback, obj, name); | 231 | Util.FireAndForget(callback, obj, name); |
210 | } | 232 | } |
233 | |||
234 | private static void HandleControlCommand(string module, string[] args) | ||
235 | { | ||
236 | // if (SceneManager.Instance.CurrentScene != null && SceneManager.Instance.CurrentScene != m_udpServer.Scene) | ||
237 | // return; | ||
238 | |||
239 | if (args.Length < 3) | ||
240 | { | ||
241 | MainConsole.Instance.Output("Usage: debug jobengine <stop|start|status|log>"); | ||
242 | return; | ||
243 | } | ||
244 | |||
245 | string subCommand = args[2]; | ||
246 | |||
247 | if (subCommand == "stop") | ||
248 | { | ||
249 | JobEngine.Stop(); | ||
250 | MainConsole.Instance.OutputFormat("Stopped job engine."); | ||
251 | } | ||
252 | else if (subCommand == "start") | ||
253 | { | ||
254 | JobEngine.Start(); | ||
255 | MainConsole.Instance.OutputFormat("Started job engine."); | ||
256 | } | ||
257 | else if (subCommand == "status") | ||
258 | { | ||
259 | MainConsole.Instance.OutputFormat("Job engine running: {0}", JobEngine.IsRunning); | ||
260 | |||
261 | JobEngine.Job job = JobEngine.CurrentJob; | ||
262 | MainConsole.Instance.OutputFormat("Current job {0}", job != null ? job.Name : "none"); | ||
263 | |||
264 | MainConsole.Instance.OutputFormat( | ||
265 | "Jobs waiting: {0}", JobEngine.IsRunning ? JobEngine.JobsWaiting.ToString() : "n/a"); | ||
266 | MainConsole.Instance.OutputFormat("Log Level: {0}", JobEngine.LogLevel); | ||
267 | } | ||
268 | else if (subCommand == "log") | ||
269 | { | ||
270 | // int logLevel; | ||
271 | int logLevel = int.Parse(args[3]); | ||
272 | // if (ConsoleUtil.TryParseConsoleInt(MainConsole.Instance, args[4], out logLevel)) | ||
273 | // { | ||
274 | JobEngine.LogLevel = logLevel; | ||
275 | MainConsole.Instance.OutputFormat("Set debug log level to {0}", JobEngine.LogLevel); | ||
276 | // } | ||
277 | } | ||
278 | else | ||
279 | { | ||
280 | MainConsole.Instance.OutputFormat("Unrecognized job engine subcommand {0}", subCommand); | ||
281 | } | ||
282 | } | ||
211 | } | 283 | } |
212 | } \ No newline at end of file | 284 | } \ No newline at end of file |
diff --git a/OpenSim/Region/ClientStack/Linden/UDP/IncomingPacketAsyncHandlingEngine.cs b/OpenSim/Region/ClientStack/Linden/UDP/IncomingPacketAsyncHandlingEngine.cs deleted file mode 100644 index 6f40b24..0000000 --- a/OpenSim/Region/ClientStack/Linden/UDP/IncomingPacketAsyncHandlingEngine.cs +++ /dev/null | |||
@@ -1,328 +0,0 @@ | |||
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 class Job | ||
40 | { | ||
41 | public string Name; | ||
42 | public WaitCallback Callback; | ||
43 | public object O; | ||
44 | |||
45 | public Job(string name, WaitCallback callback, object o) | ||
46 | { | ||
47 | Name = name; | ||
48 | Callback = callback; | ||
49 | O = o; | ||
50 | } | ||
51 | } | ||
52 | |||
53 | // TODO: These kinds of classes MUST be generalized with JobEngine, etc. | ||
54 | public class IncomingPacketAsyncHandlingEngine | ||
55 | { | ||
56 | private static readonly ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType); | ||
57 | |||
58 | public int LogLevel { get; set; } | ||
59 | |||
60 | public bool IsRunning { get; private set; } | ||
61 | |||
62 | /// <summary> | ||
63 | /// The timeout in milliseconds to wait for at least one event to be written when the recorder is stopping. | ||
64 | /// </summary> | ||
65 | public int RequestProcessTimeoutOnStop { get; set; } | ||
66 | |||
67 | /// <summary> | ||
68 | /// Controls whether we need to warn in the log about exceeding the max queue size. | ||
69 | /// </summary> | ||
70 | /// <remarks> | ||
71 | /// This is flipped to false once queue max has been exceeded and back to true when it falls below max, in | ||
72 | /// order to avoid spamming the log with lots of warnings. | ||
73 | /// </remarks> | ||
74 | private bool m_warnOverMaxQueue = true; | ||
75 | |||
76 | private BlockingCollection<Job> m_requestQueue; | ||
77 | |||
78 | private CancellationTokenSource m_cancelSource = new CancellationTokenSource(); | ||
79 | |||
80 | private LLUDPServer m_udpServer; | ||
81 | |||
82 | private Stat m_requestsWaitingStat; | ||
83 | |||
84 | private Job m_currentJob; | ||
85 | |||
86 | /// <summary> | ||
87 | /// Used to signal that we are ready to complete stop. | ||
88 | /// </summary> | ||
89 | private ManualResetEvent m_finishedProcessingAfterStop = new ManualResetEvent(false); | ||
90 | |||
91 | public IncomingPacketAsyncHandlingEngine(LLUDPServer server) | ||
92 | { | ||
93 | //LogLevel = 1; | ||
94 | m_udpServer = server; | ||
95 | RequestProcessTimeoutOnStop = 5000; | ||
96 | |||
97 | // MainConsole.Instance.Commands.AddCommand( | ||
98 | // "Debug", | ||
99 | // false, | ||
100 | // "debug jobengine", | ||
101 | // "debug jobengine <start|stop|status>", | ||
102 | // "Start, stop or get status of the job engine.", | ||
103 | // "If stopped then all jobs are processed immediately.", | ||
104 | // HandleControlCommand); | ||
105 | } | ||
106 | |||
107 | public void Start() | ||
108 | { | ||
109 | lock (this) | ||
110 | { | ||
111 | if (IsRunning) | ||
112 | return; | ||
113 | |||
114 | IsRunning = true; | ||
115 | |||
116 | m_finishedProcessingAfterStop.Reset(); | ||
117 | |||
118 | m_requestQueue = new BlockingCollection<Job>(new ConcurrentQueue<Job>(), 5000); | ||
119 | |||
120 | m_requestsWaitingStat = | ||
121 | new Stat( | ||
122 | "IncomingPacketAsyncRequestsWaiting", | ||
123 | "Number of incoming packets waiting for async processing in engine.", | ||
124 | "", | ||
125 | "", | ||
126 | "clientstack", | ||
127 | m_udpServer.Scene.Name, | ||
128 | StatType.Pull, | ||
129 | MeasuresOfInterest.None, | ||
130 | stat => stat.Value = m_requestQueue.Count, | ||
131 | StatVerbosity.Debug); | ||
132 | |||
133 | StatsManager.RegisterStat(m_requestsWaitingStat); | ||
134 | |||
135 | WorkManager.StartThread( | ||
136 | ProcessRequests, | ||
137 | string.Format("Incoming Packet Async Handling Engine Thread ({0})", m_udpServer.Scene.Name), | ||
138 | ThreadPriority.Normal, | ||
139 | false, | ||
140 | true, | ||
141 | null, | ||
142 | int.MaxValue); | ||
143 | } | ||
144 | } | ||
145 | |||
146 | public void Stop() | ||
147 | { | ||
148 | lock (this) | ||
149 | { | ||
150 | try | ||
151 | { | ||
152 | if (!IsRunning) | ||
153 | return; | ||
154 | |||
155 | IsRunning = false; | ||
156 | |||
157 | int requestsLeft = m_requestQueue.Count; | ||
158 | |||
159 | if (requestsLeft <= 0) | ||
160 | { | ||
161 | m_cancelSource.Cancel(); | ||
162 | } | ||
163 | else | ||
164 | { | ||
165 | m_log.InfoFormat("[INCOMING PACKET ASYNC HANDLING ENGINE]: Waiting to write {0} events after stop.", requestsLeft); | ||
166 | |||
167 | while (requestsLeft > 0) | ||
168 | { | ||
169 | if (!m_finishedProcessingAfterStop.WaitOne(RequestProcessTimeoutOnStop)) | ||
170 | { | ||
171 | // After timeout no events have been written | ||
172 | if (requestsLeft == m_requestQueue.Count) | ||
173 | { | ||
174 | m_log.WarnFormat( | ||
175 | "[INCOMING PACKET ASYNC HANDLING ENGINE]: No requests processed after {0} ms wait. Discarding remaining {1} requests", | ||
176 | RequestProcessTimeoutOnStop, requestsLeft); | ||
177 | |||
178 | break; | ||
179 | } | ||
180 | } | ||
181 | |||
182 | requestsLeft = m_requestQueue.Count; | ||
183 | } | ||
184 | } | ||
185 | } | ||
186 | finally | ||
187 | { | ||
188 | m_cancelSource.Dispose(); | ||
189 | StatsManager.DeregisterStat(m_requestsWaitingStat); | ||
190 | m_requestsWaitingStat = null; | ||
191 | m_requestQueue = null; | ||
192 | } | ||
193 | } | ||
194 | } | ||
195 | |||
196 | public bool QueueRequest(string name, WaitCallback req, object o) | ||
197 | { | ||
198 | if (LogLevel >= 1) | ||
199 | m_log.DebugFormat("[INCOMING PACKET ASYNC HANDLING ENGINE]: Queued job {0}", name); | ||
200 | |||
201 | if (m_requestQueue.Count < m_requestQueue.BoundedCapacity) | ||
202 | { | ||
203 | // m_log.DebugFormat( | ||
204 | // "[OUTGOING QUEUE REFILL ENGINE]: Adding request for categories {0} for {1} in {2}", | ||
205 | // categories, client.AgentID, m_udpServer.Scene.Name); | ||
206 | |||
207 | m_requestQueue.Add(new Job(name, req, o)); | ||
208 | |||
209 | if (!m_warnOverMaxQueue) | ||
210 | m_warnOverMaxQueue = true; | ||
211 | |||
212 | return true; | ||
213 | } | ||
214 | else | ||
215 | { | ||
216 | if (m_warnOverMaxQueue) | ||
217 | { | ||
218 | // m_log.WarnFormat( | ||
219 | // "[JOB ENGINE]: Request queue at maximum capacity, not recording request from {0} in {1}", | ||
220 | // client.AgentID, m_udpServer.Scene.Name); | ||
221 | |||
222 | m_log.WarnFormat("[INCOMING PACKET ASYNC HANDLING ENGINE]: Request queue at maximum capacity, not recording job"); | ||
223 | |||
224 | m_warnOverMaxQueue = false; | ||
225 | } | ||
226 | |||
227 | return false; | ||
228 | } | ||
229 | } | ||
230 | |||
231 | private void ProcessRequests() | ||
232 | { | ||
233 | try | ||
234 | { | ||
235 | while (IsRunning || m_requestQueue.Count > 0) | ||
236 | { | ||
237 | m_currentJob = m_requestQueue.Take(m_cancelSource.Token); | ||
238 | |||
239 | // QueueEmpty callback = req.Client.OnQueueEmpty; | ||
240 | // | ||
241 | // if (callback != null) | ||
242 | // { | ||
243 | // try | ||
244 | // { | ||
245 | // callback(req.Categories); | ||
246 | // } | ||
247 | // catch (Exception e) | ||
248 | // { | ||
249 | // m_log.Error("[OUTGOING QUEUE REFILL ENGINE]: ProcessRequests(" + req.Categories + ") threw an exception: " + e.Message, e); | ||
250 | // } | ||
251 | // } | ||
252 | |||
253 | if (LogLevel >= 1) | ||
254 | m_log.DebugFormat("[INCOMING PACKET ASYNC HANDLING ENGINE]: Processing job {0}", m_currentJob.Name); | ||
255 | |||
256 | try | ||
257 | { | ||
258 | m_currentJob.Callback.Invoke(m_currentJob.O); | ||
259 | } | ||
260 | catch (Exception e) | ||
261 | { | ||
262 | m_log.Error( | ||
263 | string.Format( | ||
264 | "[INCOMING PACKET ASYNC HANDLING ENGINE]: Job {0} failed, continuing. Exception ", m_currentJob.Name), e); | ||
265 | } | ||
266 | |||
267 | if (LogLevel >= 1) | ||
268 | m_log.DebugFormat("[INCOMING PACKET ASYNC HANDLING ENGINE]: Processed job {0}", m_currentJob.Name); | ||
269 | |||
270 | m_currentJob = null; | ||
271 | } | ||
272 | } | ||
273 | catch (OperationCanceledException) | ||
274 | { | ||
275 | } | ||
276 | |||
277 | m_finishedProcessingAfterStop.Set(); | ||
278 | } | ||
279 | |||
280 | // private void HandleControlCommand(string module, string[] args) | ||
281 | // { | ||
282 | // // if (SceneManager.Instance.CurrentScene != null && SceneManager.Instance.CurrentScene != m_udpServer.Scene) | ||
283 | // // return; | ||
284 | // | ||
285 | // if (args.Length < 3) | ||
286 | // { | ||
287 | // MainConsole.Instance.Output("Usage: debug jobengine <stop|start|status|loglevel>"); | ||
288 | // return; | ||
289 | // } | ||
290 | // | ||
291 | // string subCommand = args[2]; | ||
292 | // | ||
293 | // if (subCommand == "stop") | ||
294 | // { | ||
295 | // Stop(); | ||
296 | // MainConsole.Instance.OutputFormat("Stopped job engine."); | ||
297 | // } | ||
298 | // else if (subCommand == "start") | ||
299 | // { | ||
300 | // Start(); | ||
301 | // MainConsole.Instance.OutputFormat("Started job engine."); | ||
302 | // } | ||
303 | // else if (subCommand == "status") | ||
304 | // { | ||
305 | // MainConsole.Instance.OutputFormat("Job engine running: {0}", IsRunning); | ||
306 | // MainConsole.Instance.OutputFormat("Current job {0}", m_currentJob != null ? m_currentJob.Name : "none"); | ||
307 | // MainConsole.Instance.OutputFormat( | ||
308 | // "Jobs waiting: {0}", IsRunning ? m_requestQueue.Count.ToString() : "n/a"); | ||
309 | // MainConsole.Instance.OutputFormat("Log Level: {0}", LogLevel); | ||
310 | // } | ||
311 | // | ||
312 | // else if (subCommand == "loglevel") | ||
313 | // { | ||
314 | // // int logLevel; | ||
315 | // int logLevel = int.Parse(args[3]); | ||
316 | // // if (ConsoleUtil.TryParseConsoleInt(MainConsole.Instance, args[4], out logLevel)) | ||
317 | // // { | ||
318 | // LogLevel = logLevel; | ||
319 | // MainConsole.Instance.OutputFormat("Set log level to {0}", LogLevel); | ||
320 | // // } | ||
321 | // } | ||
322 | // else | ||
323 | // { | ||
324 | // MainConsole.Instance.OutputFormat("Unrecognized job engine subcommand {0}", subCommand); | ||
325 | // } | ||
326 | // } | ||
327 | } | ||
328 | } | ||
diff --git a/OpenSim/Region/ClientStack/Linden/UDP/LLClientView.cs b/OpenSim/Region/ClientStack/Linden/UDP/LLClientView.cs index 5da0ca1..bb4f8a7 100644 --- a/OpenSim/Region/ClientStack/Linden/UDP/LLClientView.cs +++ b/OpenSim/Region/ClientStack/Linden/UDP/LLClientView.cs | |||
@@ -724,10 +724,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP | |||
724 | object obj = new AsyncPacketProcess(this, pprocessor.method, packet); | 724 | object obj = new AsyncPacketProcess(this, pprocessor.method, packet); |
725 | 725 | ||
726 | if (pprocessor.InEngine) | 726 | if (pprocessor.InEngine) |
727 | m_udpServer.IpahEngine.QueueRequest( | 727 | m_udpServer.IpahEngine.QueueJob(packet.Type.ToString(), () => ProcessSpecificPacketAsync(obj)); |
728 | packet.Type.ToString(), | ||
729 | ProcessSpecificPacketAsync, | ||
730 | obj); | ||
731 | else | 728 | else |
732 | Util.FireAndForget(ProcessSpecificPacketAsync, obj, packet.Type.ToString()); | 729 | Util.FireAndForget(ProcessSpecificPacketAsync, obj, packet.Type.ToString()); |
733 | 730 | ||
diff --git a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs index de91856..ce6e3ee 100644 --- a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs +++ b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPClient.cs | |||
@@ -736,7 +736,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP | |||
736 | } | 736 | } |
737 | else | 737 | else |
738 | { | 738 | { |
739 | m_udpServer.OqrEngine.QueueRequest(this, categories); | 739 | m_udpServer.OqrEngine.QueueJob(AgentID.ToString(), () => FireQueueEmpty(categories)); |
740 | } | 740 | } |
741 | } | 741 | } |
742 | else | 742 | else |
diff --git a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServer.cs b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServer.cs index 2f97516..7bd16e6 100644 --- a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServer.cs +++ b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServer.cs | |||
@@ -367,14 +367,17 @@ namespace OpenSim.Region.ClientStack.LindenUDP | |||
367 | /// Queue some low priority but potentially high volume async requests so that they don't overwhelm available | 367 | /// Queue some low priority but potentially high volume async requests so that they don't overwhelm available |
368 | /// threadpool threads. | 368 | /// threadpool threads. |
369 | /// </summary> | 369 | /// </summary> |
370 | public IncomingPacketAsyncHandlingEngine IpahEngine { get; private set; } | 370 | public JobEngine IpahEngine { get; private set; } |
371 | 371 | ||
372 | /// <summary> | 372 | /// <summary> |
373 | /// Experimental facility to run queue empty processing within a controlled number of threads rather than | 373 | /// Run queue empty processing within a single persistent thread. |
374 | /// requiring massive numbers of short-lived threads from the threadpool when there are a high number of | ||
375 | /// connections. | ||
376 | /// </summary> | 374 | /// </summary> |
377 | public OutgoingQueueRefillEngine OqrEngine { get; private set; } | 375 | /// <remarks> |
376 | /// This is the alternative to having every | ||
377 | /// connection schedule its own job in the threadpool which causes performance problems when there are many | ||
378 | /// connections. | ||
379 | /// </remarks> | ||
380 | public JobEngine OqrEngine { get; private set; } | ||
378 | 381 | ||
379 | public LLUDPServer( | 382 | public LLUDPServer( |
380 | IPAddress listenIP, ref uint port, int proxyPortOffsetParm, bool allow_alternate_port, | 383 | IPAddress listenIP, ref uint port, int proxyPortOffsetParm, bool allow_alternate_port, |
@@ -459,9 +462,6 @@ namespace OpenSim.Region.ClientStack.LindenUDP | |||
459 | 462 | ||
460 | if (usePools) | 463 | if (usePools) |
461 | EnablePools(); | 464 | EnablePools(); |
462 | |||
463 | IpahEngine = new IncomingPacketAsyncHandlingEngine(this); | ||
464 | OqrEngine = new OutgoingQueueRefillEngine(this); | ||
465 | } | 465 | } |
466 | 466 | ||
467 | public void Start() | 467 | public void Start() |
@@ -633,6 +633,16 @@ namespace OpenSim.Region.ClientStack.LindenUDP | |||
633 | 633 | ||
634 | Scene = (Scene)scene; | 634 | Scene = (Scene)scene; |
635 | m_location = new Location(Scene.RegionInfo.RegionHandle); | 635 | m_location = new Location(Scene.RegionInfo.RegionHandle); |
636 | |||
637 | IpahEngine | ||
638 | = new JobEngine( | ||
639 | string.Format("Incoming Packet Async Handling Engine ({0})", Scene.Name), | ||
640 | "INCOMING PACKET ASYNC HANDLING ENGINE"); | ||
641 | |||
642 | OqrEngine | ||
643 | = new JobEngine( | ||
644 | string.Format("Outgoing Queue Refill Engine ({0})", Scene.Name), | ||
645 | "OUTGOING QUEUE REFILL ENGINE"); | ||
636 | 646 | ||
637 | StatsManager.RegisterStat( | 647 | StatsManager.RegisterStat( |
638 | new Stat( | 648 | new Stat( |
@@ -713,6 +723,32 @@ namespace OpenSim.Region.ClientStack.LindenUDP | |||
713 | MeasuresOfInterest.AverageChangeOverTime, | 723 | MeasuresOfInterest.AverageChangeOverTime, |
714 | stat => stat.Value = GetTotalQueuedOutgoingPackets(), | 724 | stat => stat.Value = GetTotalQueuedOutgoingPackets(), |
715 | StatVerbosity.Info)); | 725 | StatVerbosity.Info)); |
726 | |||
727 | StatsManager.RegisterStat( | ||
728 | new Stat( | ||
729 | "IncomingPacketAsyncRequestsWaiting", | ||
730 | "Number of incoming packets waiting for async processing in engine.", | ||
731 | "", | ||
732 | "", | ||
733 | "clientstack", | ||
734 | Scene.Name, | ||
735 | StatType.Pull, | ||
736 | MeasuresOfInterest.None, | ||
737 | stat => stat.Value = IpahEngine.JobsWaiting, | ||
738 | StatVerbosity.Debug)); | ||
739 | |||
740 | StatsManager.RegisterStat( | ||
741 | new Stat( | ||
742 | "OQRERequestsWaiting", | ||
743 | "Number of outgong queue refill requests waiting for processing.", | ||
744 | "", | ||
745 | "", | ||
746 | "clientstack", | ||
747 | Scene.Name, | ||
748 | StatType.Pull, | ||
749 | MeasuresOfInterest.None, | ||
750 | stat => stat.Value = OqrEngine.JobsWaiting, | ||
751 | StatVerbosity.Debug)); | ||
716 | 752 | ||
717 | // We delay enabling pool stats to AddScene() instead of Initialize() so that we can distinguish pool stats by | 753 | // We delay enabling pool stats to AddScene() instead of Initialize() so that we can distinguish pool stats by |
718 | // scene name | 754 | // scene name |
diff --git a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServerCommands.cs b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServerCommands.cs index e0398d5..17a394d 100644 --- a/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServerCommands.cs +++ b/OpenSim/Region/ClientStack/Linden/UDP/LLUDPServerCommands.cs | |||
@@ -186,6 +186,15 @@ namespace OpenSim.Region.ClientStack.LindenUDP | |||
186 | "debug lludp toggle agentupdate", | 186 | "debug lludp toggle agentupdate", |
187 | "Toggle whether agentupdate packets are processed or simply discarded.", | 187 | "Toggle whether agentupdate packets are processed or simply discarded.", |
188 | HandleAgentUpdateCommand); | 188 | HandleAgentUpdateCommand); |
189 | |||
190 | MainConsole.Instance.Commands.AddCommand( | ||
191 | "Debug", | ||
192 | false, | ||
193 | "debug lludp oqre", | ||
194 | "debug lludp oqre <start|stop|status>", | ||
195 | "Start, stop or get status of OutgoingQueueRefillEngine.", | ||
196 | "If stopped then refill requests are processed directly via the threadpool.", | ||
197 | HandleOqreCommand); | ||
189 | } | 198 | } |
190 | 199 | ||
191 | private void HandleShowServerThrottlesCommand(string module, string[] args) | 200 | private void HandleShowServerThrottlesCommand(string module, string[] args) |
@@ -758,5 +767,42 @@ namespace OpenSim.Region.ClientStack.LindenUDP | |||
758 | MainConsole.Instance.OutputFormat( | 767 | MainConsole.Instance.OutputFormat( |
759 | "Packet debug level for new clients is {0}", m_udpServer.DefaultClientPacketDebugLevel); | 768 | "Packet debug level for new clients is {0}", m_udpServer.DefaultClientPacketDebugLevel); |
760 | } | 769 | } |
770 | |||
771 | private void HandleOqreCommand(string module, string[] args) | ||
772 | { | ||
773 | if (SceneManager.Instance.CurrentScene != null && SceneManager.Instance.CurrentScene != m_udpServer.Scene) | ||
774 | return; | ||
775 | |||
776 | if (args.Length != 4) | ||
777 | { | ||
778 | MainConsole.Instance.Output("Usage: debug lludp oqre <stop|start|status>"); | ||
779 | return; | ||
780 | } | ||
781 | |||
782 | string subCommand = args[3]; | ||
783 | |||
784 | if (subCommand == "stop") | ||
785 | { | ||
786 | m_udpServer.OqrEngine.Stop(); | ||
787 | MainConsole.Instance.OutputFormat("Stopped OQRE for {0}", m_udpServer.Scene.Name); | ||
788 | } | ||
789 | else if (subCommand == "start") | ||
790 | { | ||
791 | m_udpServer.OqrEngine.Start(); | ||
792 | MainConsole.Instance.OutputFormat("Started OQRE for {0}", m_udpServer.Scene.Name); | ||
793 | } | ||
794 | else if (subCommand == "status") | ||
795 | { | ||
796 | MainConsole.Instance.OutputFormat("OQRE in {0}", m_udpServer.Scene.Name); | ||
797 | MainConsole.Instance.OutputFormat("Running: {0}", m_udpServer.OqrEngine.IsRunning); | ||
798 | MainConsole.Instance.OutputFormat( | ||
799 | "Requests waiting: {0}", | ||
800 | m_udpServer.OqrEngine.IsRunning ? m_udpServer.OqrEngine.JobsWaiting.ToString() : "n/a"); | ||
801 | } | ||
802 | else | ||
803 | { | ||
804 | MainConsole.Instance.OutputFormat("Unrecognized OQRE subcommand {0}", subCommand); | ||
805 | } | ||
806 | } | ||
761 | } | 807 | } |
762 | } \ No newline at end of file | 808 | } \ No newline at end of file |
diff --git a/OpenSim/Region/ClientStack/Linden/UDP/OutgoingQueueRefillEngine.cs b/OpenSim/Region/ClientStack/Linden/UDP/OutgoingQueueRefillEngine.cs deleted file mode 100644 index 1e915c3..0000000 --- a/OpenSim/Region/ClientStack/Linden/UDP/OutgoingQueueRefillEngine.cs +++ /dev/null | |||
@@ -1,288 +0,0 @@ | |||
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 | private Stat m_oqreRequestsWaitingStat; | ||
78 | |||
79 | /// <summary> | ||
80 | /// Used to signal that we are ready to complete stop. | ||
81 | /// </summary> | ||
82 | private ManualResetEvent m_finishedProcessingAfterStop = new ManualResetEvent(false); | ||
83 | |||
84 | public OutgoingQueueRefillEngine(LLUDPServer server) | ||
85 | { | ||
86 | RequestProcessTimeoutOnStop = 5000; | ||
87 | m_udpServer = server; | ||
88 | |||
89 | MainConsole.Instance.Commands.AddCommand( | ||
90 | "Debug", | ||
91 | false, | ||
92 | "debug lludp oqre", | ||
93 | "debug lludp oqre <start|stop|status>", | ||
94 | "Start, stop or get status of OutgoingQueueRefillEngine.", | ||
95 | "If stopped then refill requests are processed directly via the threadpool.", | ||
96 | HandleOqreCommand); | ||
97 | } | ||
98 | |||
99 | public void Start() | ||
100 | { | ||
101 | lock (this) | ||
102 | { | ||
103 | if (IsRunning) | ||
104 | return; | ||
105 | |||
106 | IsRunning = true; | ||
107 | |||
108 | m_finishedProcessingAfterStop.Reset(); | ||
109 | |||
110 | m_requestQueue = new BlockingCollection<RefillRequest>(new ConcurrentQueue<RefillRequest>(), 5000); | ||
111 | |||
112 | m_oqreRequestsWaitingStat = | ||
113 | new Stat( | ||
114 | "OQRERequestsWaiting", | ||
115 | "Number of outgong queue refill requests waiting for processing.", | ||
116 | "", | ||
117 | "", | ||
118 | "clientstack", | ||
119 | m_udpServer.Scene.Name, | ||
120 | StatType.Pull, | ||
121 | MeasuresOfInterest.None, | ||
122 | stat => stat.Value = m_requestQueue.Count, | ||
123 | StatVerbosity.Debug); | ||
124 | |||
125 | StatsManager.RegisterStat(m_oqreRequestsWaitingStat); | ||
126 | |||
127 | WorkManager.StartThread( | ||
128 | ProcessRequests, | ||
129 | String.Format("OutgoingQueueRefillEngineThread ({0})", m_udpServer.Scene.Name), | ||
130 | ThreadPriority.Normal, | ||
131 | false, | ||
132 | true, | ||
133 | null, | ||
134 | int.MaxValue); | ||
135 | } | ||
136 | } | ||
137 | |||
138 | public void Stop() | ||
139 | { | ||
140 | lock (this) | ||
141 | { | ||
142 | try | ||
143 | { | ||
144 | if (!IsRunning) | ||
145 | return; | ||
146 | |||
147 | IsRunning = false; | ||
148 | |||
149 | int requestsLeft = m_requestQueue.Count; | ||
150 | |||
151 | if (requestsLeft <= 0) | ||
152 | { | ||
153 | m_cancelSource.Cancel(); | ||
154 | } | ||
155 | else | ||
156 | { | ||
157 | m_log.InfoFormat("[OUTGOING QUEUE REFILL ENGINE]: Waiting to write {0} events after stop.", requestsLeft); | ||
158 | |||
159 | while (requestsLeft > 0) | ||
160 | { | ||
161 | if (!m_finishedProcessingAfterStop.WaitOne(RequestProcessTimeoutOnStop)) | ||
162 | { | ||
163 | // After timeout no events have been written | ||
164 | if (requestsLeft == m_requestQueue.Count) | ||
165 | { | ||
166 | m_log.WarnFormat( | ||
167 | "[OUTGOING QUEUE REFILL ENGINE]: No requests processed after {0} ms wait. Discarding remaining {1} requests", | ||
168 | RequestProcessTimeoutOnStop, requestsLeft); | ||
169 | |||
170 | break; | ||
171 | } | ||
172 | } | ||
173 | |||
174 | requestsLeft = m_requestQueue.Count; | ||
175 | } | ||
176 | } | ||
177 | } | ||
178 | finally | ||
179 | { | ||
180 | m_cancelSource.Dispose(); | ||
181 | StatsManager.DeregisterStat(m_oqreRequestsWaitingStat); | ||
182 | m_oqreRequestsWaitingStat = null; | ||
183 | m_requestQueue = null; | ||
184 | } | ||
185 | } | ||
186 | } | ||
187 | |||
188 | public bool QueueRequest(LLUDPClient client, ThrottleOutPacketTypeFlags categories) | ||
189 | { | ||
190 | if (m_requestQueue.Count < m_requestQueue.BoundedCapacity) | ||
191 | { | ||
192 | // m_log.DebugFormat( | ||
193 | // "[OUTGOING QUEUE REFILL ENGINE]: Adding request for categories {0} for {1} in {2}", | ||
194 | // categories, client.AgentID, m_udpServer.Scene.Name); | ||
195 | |||
196 | m_requestQueue.Add(new RefillRequest(client, categories)); | ||
197 | |||
198 | if (!m_warnOverMaxQueue) | ||
199 | m_warnOverMaxQueue = true; | ||
200 | |||
201 | return true; | ||
202 | } | ||
203 | else | ||
204 | { | ||
205 | if (m_warnOverMaxQueue) | ||
206 | { | ||
207 | m_log.WarnFormat( | ||
208 | "[OUTGOING QUEUE REFILL ENGINE]: Request queue at maximum capacity, not recording request from {0} in {1}", | ||
209 | client.AgentID, m_udpServer.Scene.Name); | ||
210 | |||
211 | m_warnOverMaxQueue = false; | ||
212 | } | ||
213 | |||
214 | return false; | ||
215 | } | ||
216 | } | ||
217 | |||
218 | private void ProcessRequests() | ||
219 | { | ||
220 | Thread.CurrentThread.Priority = ThreadPriority.Highest; | ||
221 | |||
222 | try | ||
223 | { | ||
224 | while (IsRunning || m_requestQueue.Count > 0) | ||
225 | { | ||
226 | RefillRequest req = m_requestQueue.Take(m_cancelSource.Token); | ||
227 | |||
228 | // QueueEmpty callback = req.Client.OnQueueEmpty; | ||
229 | // | ||
230 | // if (callback != null) | ||
231 | // { | ||
232 | // try | ||
233 | // { | ||
234 | // callback(req.Categories); | ||
235 | // } | ||
236 | // catch (Exception e) | ||
237 | // { | ||
238 | // m_log.Error("[OUTGOING QUEUE REFILL ENGINE]: ProcessRequests(" + req.Categories + ") threw an exception: " + e.Message, e); | ||
239 | // } | ||
240 | // } | ||
241 | |||
242 | req.Client.FireQueueEmpty(req.Categories); | ||
243 | } | ||
244 | } | ||
245 | catch (OperationCanceledException) | ||
246 | { | ||
247 | } | ||
248 | |||
249 | m_finishedProcessingAfterStop.Set(); | ||
250 | } | ||
251 | |||
252 | private void HandleOqreCommand(string module, string[] args) | ||
253 | { | ||
254 | if (SceneManager.Instance.CurrentScene != null && SceneManager.Instance.CurrentScene != m_udpServer.Scene) | ||
255 | return; | ||
256 | |||
257 | if (args.Length != 4) | ||
258 | { | ||
259 | MainConsole.Instance.Output("Usage: debug lludp oqre <stop|start|status>"); | ||
260 | return; | ||
261 | } | ||
262 | |||
263 | string subCommand = args[3]; | ||
264 | |||
265 | if (subCommand == "stop") | ||
266 | { | ||
267 | Stop(); | ||
268 | MainConsole.Instance.OutputFormat("Stopped OQRE for {0}", m_udpServer.Scene.Name); | ||
269 | } | ||
270 | else if (subCommand == "start") | ||
271 | { | ||
272 | Start(); | ||
273 | MainConsole.Instance.OutputFormat("Started OQRE for {0}", m_udpServer.Scene.Name); | ||
274 | } | ||
275 | else if (subCommand == "status") | ||
276 | { | ||
277 | MainConsole.Instance.OutputFormat("OQRE in {0}", m_udpServer.Scene.Name); | ||
278 | MainConsole.Instance.OutputFormat("Running: {0}", IsRunning); | ||
279 | MainConsole.Instance.OutputFormat( | ||
280 | "Requests waiting: {0}", IsRunning ? m_requestQueue.Count.ToString() : "n/a"); | ||
281 | } | ||
282 | else | ||
283 | { | ||
284 | MainConsole.Instance.OutputFormat("Unrecognized OQRE subcommand {0}", subCommand); | ||
285 | } | ||
286 | } | ||
287 | } | ||
288 | } \ No newline at end of file | ||
diff --git a/OpenSim/Region/CoreModules/Framework/EntityTransfer/HGEntityTransferModule.cs b/OpenSim/Region/CoreModules/Framework/EntityTransfer/HGEntityTransferModule.cs index fceda80..fa23590 100644 --- a/OpenSim/Region/CoreModules/Framework/EntityTransfer/HGEntityTransferModule.cs +++ b/OpenSim/Region/CoreModules/Framework/EntityTransfer/HGEntityTransferModule.cs | |||
@@ -31,6 +31,7 @@ using System.Reflection; | |||
31 | 31 | ||
32 | using OpenSim.Framework; | 32 | using OpenSim.Framework; |
33 | using OpenSim.Framework.Client; | 33 | using OpenSim.Framework.Client; |
34 | using OpenSim.Framework.Monitoring; | ||
34 | using OpenSim.Region.Framework.Interfaces; | 35 | using OpenSim.Region.Framework.Interfaces; |
35 | using OpenSim.Region.Framework.Scenes; | 36 | using OpenSim.Region.Framework.Scenes; |
36 | using OpenSim.Services.Connectors.Hypergrid; | 37 | using OpenSim.Services.Connectors.Hypergrid; |
@@ -113,7 +114,7 @@ namespace OpenSim.Region.CoreModules.Framework.EntityTransfer | |||
113 | /// <summary> | 114 | /// <summary> |
114 | /// Used for processing analysis of incoming attachments in a controlled fashion. | 115 | /// Used for processing analysis of incoming attachments in a controlled fashion. |
115 | /// </summary> | 116 | /// </summary> |
116 | private HGIncomingSceneObjectEngine m_incomingSceneObjectEngine; | 117 | private JobEngine m_incomingSceneObjectEngine; |
117 | 118 | ||
118 | #region ISharedRegionModule | 119 | #region ISharedRegionModule |
119 | 120 | ||
@@ -160,7 +161,24 @@ namespace OpenSim.Region.CoreModules.Framework.EntityTransfer | |||
160 | scene.RegisterModuleInterface<IUserAgentVerificationModule>(this); | 161 | scene.RegisterModuleInterface<IUserAgentVerificationModule>(this); |
161 | //scene.EventManager.OnIncomingSceneObject += OnIncomingSceneObject; | 162 | //scene.EventManager.OnIncomingSceneObject += OnIncomingSceneObject; |
162 | 163 | ||
163 | m_incomingSceneObjectEngine = new HGIncomingSceneObjectEngine(scene.Name); | 164 | m_incomingSceneObjectEngine |
165 | = new JobEngine( | ||
166 | string.Format("HG Incoming Scene Object Engine ({0})", scene.Name), | ||
167 | "HG INCOMING SCENE OBJECT ENGINE"); | ||
168 | |||
169 | StatsManager.RegisterStat( | ||
170 | new Stat( | ||
171 | "HGIncomingAttachmentsWaiting", | ||
172 | "Number of incoming attachments waiting for processing.", | ||
173 | "", | ||
174 | "", | ||
175 | "entitytransfer", | ||
176 | Name, | ||
177 | StatType.Pull, | ||
178 | MeasuresOfInterest.None, | ||
179 | stat => stat.Value = m_incomingSceneObjectEngine.JobsWaiting, | ||
180 | StatVerbosity.Debug)); | ||
181 | |||
164 | m_incomingSceneObjectEngine.Start(); | 182 | m_incomingSceneObjectEngine.Start(); |
165 | } | 183 | } |
166 | } | 184 | } |
@@ -548,11 +566,11 @@ namespace OpenSim.Region.CoreModules.Framework.EntityTransfer | |||
548 | 566 | ||
549 | private void RemoveIncomingSceneObjectJobs(string commonIdToRemove) | 567 | private void RemoveIncomingSceneObjectJobs(string commonIdToRemove) |
550 | { | 568 | { |
551 | List<Job> jobsToReinsert = new List<Job>(); | 569 | List<JobEngine.Job> jobsToReinsert = new List<JobEngine.Job>(); |
552 | int jobsRemoved = 0; | 570 | int jobsRemoved = 0; |
553 | 571 | ||
554 | Job job; | 572 | JobEngine.Job job; |
555 | while ((job = m_incomingSceneObjectEngine.RemoveNextRequest()) != null) | 573 | while ((job = m_incomingSceneObjectEngine.RemoveNextJob()) != null) |
556 | { | 574 | { |
557 | if (job.CommonId != commonIdToRemove) | 575 | if (job.CommonId != commonIdToRemove) |
558 | jobsToReinsert.Add(job); | 576 | jobsToReinsert.Add(job); |
@@ -566,8 +584,8 @@ namespace OpenSim.Region.CoreModules.Framework.EntityTransfer | |||
566 | 584 | ||
567 | if (jobsToReinsert.Count > 0) | 585 | if (jobsToReinsert.Count > 0) |
568 | { | 586 | { |
569 | foreach (Job jobToReinsert in jobsToReinsert) | 587 | foreach (JobEngine.Job jobToReinsert in jobsToReinsert) |
570 | m_incomingSceneObjectEngine.QueueRequest(jobToReinsert); | 588 | m_incomingSceneObjectEngine.QueueJob(jobToReinsert); |
571 | } | 589 | } |
572 | } | 590 | } |
573 | 591 | ||
@@ -594,10 +612,9 @@ namespace OpenSim.Region.CoreModules.Framework.EntityTransfer | |||
594 | { | 612 | { |
595 | if (aCircuit.ServiceURLs != null && aCircuit.ServiceURLs.ContainsKey("AssetServerURI")) | 613 | if (aCircuit.ServiceURLs != null && aCircuit.ServiceURLs.ContainsKey("AssetServerURI")) |
596 | { | 614 | { |
597 | m_incomingSceneObjectEngine.QueueRequest( | 615 | m_incomingSceneObjectEngine.QueueJob( |
598 | string.Format("HG UUID Gather for attachment {0} for {1}", so.Name, aCircuit.Name), | 616 | string.Format("HG UUID Gather for attachment {0} for {1}", so.Name, aCircuit.Name), |
599 | so.OwnerID.ToString(), | 617 | () => |
600 | o => | ||
601 | { | 618 | { |
602 | string url = aCircuit.ServiceURLs["AssetServerURI"].ToString(); | 619 | string url = aCircuit.ServiceURLs["AssetServerURI"].ToString(); |
603 | // m_log.DebugFormat( | 620 | // m_log.DebugFormat( |
@@ -663,8 +680,8 @@ namespace OpenSim.Region.CoreModules.Framework.EntityTransfer | |||
663 | // m_log.DebugFormat( | 680 | // m_log.DebugFormat( |
664 | // "[HG ENTITY TRANSFER MODULE]: Completed incoming attachment {0} for HG user {1} with asset server {2}", | 681 | // "[HG ENTITY TRANSFER MODULE]: Completed incoming attachment {0} for HG user {1} with asset server {2}", |
665 | // so.Name, so.OwnerID, url); | 682 | // so.Name, so.OwnerID, url); |
666 | }, | 683 | }, |
667 | null); | 684 | so.OwnerID.ToString()); |
668 | } | 685 | } |
669 | } | 686 | } |
670 | } | 687 | } |
diff --git a/OpenSim/Region/CoreModules/Framework/EntityTransfer/HGIncomingSceneObjectEngine.cs b/OpenSim/Region/CoreModules/Framework/EntityTransfer/HGIncomingSceneObjectEngine.cs deleted file mode 100644 index f62e7f4..0000000 --- a/OpenSim/Region/CoreModules/Framework/EntityTransfer/HGIncomingSceneObjectEngine.cs +++ /dev/null | |||
@@ -1,344 +0,0 @@ | |||
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.CoreModules.Framework.EntityTransfer | ||
38 | { | ||
39 | public class Job | ||
40 | { | ||
41 | public string Name { get; private set; } | ||
42 | public string CommonId { get; private set; } | ||
43 | public WaitCallback Callback { get; private set; } | ||
44 | public object O { get; private set; } | ||
45 | |||
46 | public Job(string name, string commonId, WaitCallback callback, object o) | ||
47 | { | ||
48 | Name = name; | ||
49 | CommonId = commonId; | ||
50 | Callback = callback; | ||
51 | O = o; | ||
52 | } | ||
53 | } | ||
54 | |||
55 | // TODO: These kinds of classes MUST be generalized with JobEngine, etc. | ||
56 | public class HGIncomingSceneObjectEngine | ||
57 | { | ||
58 | private static readonly ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType); | ||
59 | |||
60 | public int LogLevel { get; set; } | ||
61 | |||
62 | public bool IsRunning { get; private set; } | ||
63 | |||
64 | public string Name { get; set; } | ||
65 | |||
66 | /// <summary> | ||
67 | /// The timeout in milliseconds to wait for at least one event to be written when the recorder is stopping. | ||
68 | /// </summary> | ||
69 | public int RequestProcessTimeoutOnStop { get; set; } | ||
70 | |||
71 | /// <summary> | ||
72 | /// Controls whether we need to warn in the log about exceeding the max queue size. | ||
73 | /// </summary> | ||
74 | /// <remarks> | ||
75 | /// This is flipped to false once queue max has been exceeded and back to true when it falls below max, in | ||
76 | /// order to avoid spamming the log with lots of warnings. | ||
77 | /// </remarks> | ||
78 | private bool m_warnOverMaxQueue = true; | ||
79 | |||
80 | private BlockingCollection<Job> m_requestQueue; | ||
81 | |||
82 | private CancellationTokenSource m_cancelSource = new CancellationTokenSource(); | ||
83 | |||
84 | private Stat m_requestsWaitingStat; | ||
85 | |||
86 | private Job m_currentJob; | ||
87 | |||
88 | /// <summary> | ||
89 | /// Used to signal that we are ready to complete stop. | ||
90 | /// </summary> | ||
91 | private ManualResetEvent m_finishedProcessingAfterStop = new ManualResetEvent(false); | ||
92 | |||
93 | public HGIncomingSceneObjectEngine(string name) | ||
94 | { | ||
95 | // LogLevel = 1; | ||
96 | Name = name; | ||
97 | RequestProcessTimeoutOnStop = 5000; | ||
98 | |||
99 | // MainConsole.Instance.Commands.AddCommand( | ||
100 | // "Debug", | ||
101 | // false, | ||
102 | // "debug jobengine", | ||
103 | // "debug jobengine <start|stop|status>", | ||
104 | // "Start, stop or get status of the job engine.", | ||
105 | // "If stopped then all jobs are processed immediately.", | ||
106 | // HandleControlCommand); | ||
107 | } | ||
108 | |||
109 | public void Start() | ||
110 | { | ||
111 | lock (this) | ||
112 | { | ||
113 | if (IsRunning) | ||
114 | return; | ||
115 | |||
116 | IsRunning = true; | ||
117 | |||
118 | m_finishedProcessingAfterStop.Reset(); | ||
119 | |||
120 | m_requestQueue = new BlockingCollection<Job>(new ConcurrentQueue<Job>(), 5000); | ||
121 | |||
122 | m_requestsWaitingStat = | ||
123 | new Stat( | ||
124 | "HGIncomingAttachmentsWaiting", | ||
125 | "Number of incoming attachments waiting for processing.", | ||
126 | "", | ||
127 | "", | ||
128 | "entitytransfer", | ||
129 | Name, | ||
130 | StatType.Pull, | ||
131 | MeasuresOfInterest.None, | ||
132 | stat => stat.Value = m_requestQueue.Count, | ||
133 | StatVerbosity.Debug); | ||
134 | |||
135 | StatsManager.RegisterStat(m_requestsWaitingStat); | ||
136 | |||
137 | WorkManager.StartThread( | ||
138 | ProcessRequests, | ||
139 | string.Format("HG Incoming Scene Object Engine Thread ({0})", Name), | ||
140 | ThreadPriority.Normal, | ||
141 | false, | ||
142 | true, | ||
143 | null, | ||
144 | int.MaxValue); | ||
145 | } | ||
146 | } | ||
147 | |||
148 | public void Stop() | ||
149 | { | ||
150 | lock (this) | ||
151 | { | ||
152 | try | ||
153 | { | ||
154 | if (!IsRunning) | ||
155 | return; | ||
156 | |||
157 | IsRunning = false; | ||
158 | |||
159 | int requestsLeft = m_requestQueue.Count; | ||
160 | |||
161 | if (requestsLeft <= 0) | ||
162 | { | ||
163 | m_cancelSource.Cancel(); | ||
164 | } | ||
165 | else | ||
166 | { | ||
167 | m_log.InfoFormat("[HG INCOMING SCENE OBJECT ENGINE]: Waiting to write {0} events after stop.", requestsLeft); | ||
168 | |||
169 | while (requestsLeft > 0) | ||
170 | { | ||
171 | if (!m_finishedProcessingAfterStop.WaitOne(RequestProcessTimeoutOnStop)) | ||
172 | { | ||
173 | // After timeout no events have been written | ||
174 | if (requestsLeft == m_requestQueue.Count) | ||
175 | { | ||
176 | m_log.WarnFormat( | ||
177 | "[HG INCOMING SCENE OBJECT ENGINE]: No requests processed after {0} ms wait. Discarding remaining {1} requests", | ||
178 | RequestProcessTimeoutOnStop, requestsLeft); | ||
179 | |||
180 | break; | ||
181 | } | ||
182 | } | ||
183 | |||
184 | requestsLeft = m_requestQueue.Count; | ||
185 | } | ||
186 | } | ||
187 | } | ||
188 | finally | ||
189 | { | ||
190 | m_cancelSource.Dispose(); | ||
191 | StatsManager.DeregisterStat(m_requestsWaitingStat); | ||
192 | m_requestsWaitingStat = null; | ||
193 | m_requestQueue = null; | ||
194 | } | ||
195 | } | ||
196 | } | ||
197 | |||
198 | public Job RemoveNextRequest() | ||
199 | { | ||
200 | Job nextRequest; | ||
201 | m_requestQueue.TryTake(out nextRequest); | ||
202 | |||
203 | return nextRequest; | ||
204 | } | ||
205 | |||
206 | public bool QueueRequest(string name, string commonId, WaitCallback req, object o) | ||
207 | { | ||
208 | return QueueRequest(new Job(name, commonId, req, o)); | ||
209 | } | ||
210 | |||
211 | public bool QueueRequest(Job job) | ||
212 | { | ||
213 | if (LogLevel >= 1) | ||
214 | m_log.DebugFormat( | ||
215 | "[HG INCOMING SCENE OBJECT ENGINE]: Queued job {0}, common ID {1}", job.Name, job.CommonId); | ||
216 | |||
217 | if (m_requestQueue.Count < m_requestQueue.BoundedCapacity) | ||
218 | { | ||
219 | // m_log.DebugFormat( | ||
220 | // "[OUTGOING QUEUE REFILL ENGINE]: Adding request for categories {0} for {1} in {2}", | ||
221 | // categories, client.AgentID, m_udpServer.Scene.Name); | ||
222 | |||
223 | m_requestQueue.Add(job); | ||
224 | |||
225 | if (!m_warnOverMaxQueue) | ||
226 | m_warnOverMaxQueue = true; | ||
227 | |||
228 | return true; | ||
229 | } | ||
230 | else | ||
231 | { | ||
232 | if (m_warnOverMaxQueue) | ||
233 | { | ||
234 | // m_log.WarnFormat( | ||
235 | // "[JOB ENGINE]: Request queue at maximum capacity, not recording request from {0} in {1}", | ||
236 | // client.AgentID, m_udpServer.Scene.Name); | ||
237 | |||
238 | m_log.WarnFormat("[HG INCOMING SCENE OBJECT ENGINE]: Request queue at maximum capacity, not recording job"); | ||
239 | |||
240 | m_warnOverMaxQueue = false; | ||
241 | } | ||
242 | |||
243 | return false; | ||
244 | } | ||
245 | } | ||
246 | |||
247 | private void ProcessRequests() | ||
248 | { | ||
249 | try | ||
250 | { | ||
251 | while (IsRunning || m_requestQueue.Count > 0) | ||
252 | { | ||
253 | m_currentJob = m_requestQueue.Take(m_cancelSource.Token); | ||
254 | |||
255 | // QueueEmpty callback = req.Client.OnQueueEmpty; | ||
256 | // | ||
257 | // if (callback != null) | ||
258 | // { | ||
259 | // try | ||
260 | // { | ||
261 | // callback(req.Categories); | ||
262 | // } | ||
263 | // catch (Exception e) | ||
264 | // { | ||
265 | // m_log.Error("[OUTGOING QUEUE REFILL ENGINE]: ProcessRequests(" + req.Categories + ") threw an exception: " + e.Message, e); | ||
266 | // } | ||
267 | // } | ||
268 | |||
269 | if (LogLevel >= 1) | ||
270 | m_log.DebugFormat("[HG INCOMING SCENE OBJECT ENGINE]: Processing job {0}", m_currentJob.Name); | ||
271 | |||
272 | try | ||
273 | { | ||
274 | m_currentJob.Callback.Invoke(m_currentJob.O); | ||
275 | } | ||
276 | catch (Exception e) | ||
277 | { | ||
278 | m_log.Error( | ||
279 | string.Format( | ||
280 | "[HG INCOMING SCENE OBJECT ENGINE]: Job {0} failed, continuing. Exception ", m_currentJob.Name), e); | ||
281 | } | ||
282 | |||
283 | if (LogLevel >= 1) | ||
284 | m_log.DebugFormat("[HG INCOMING SCENE OBJECT ENGINE]: Processed job {0}", m_currentJob.Name); | ||
285 | |||
286 | m_currentJob = null; | ||
287 | } | ||
288 | } | ||
289 | catch (OperationCanceledException) | ||
290 | { | ||
291 | } | ||
292 | |||
293 | m_finishedProcessingAfterStop.Set(); | ||
294 | } | ||
295 | |||
296 | // private void HandleControlCommand(string module, string[] args) | ||
297 | // { | ||
298 | // // if (SceneManager.Instance.CurrentScene != null && SceneManager.Instance.CurrentScene != m_udpServer.Scene) | ||
299 | // // return; | ||
300 | // | ||
301 | // if (args.Length < 3) | ||
302 | // { | ||
303 | // MainConsole.Instance.Output("Usage: debug jobengine <stop|start|status|loglevel>"); | ||
304 | // return; | ||
305 | // } | ||
306 | // | ||
307 | // string subCommand = args[2]; | ||
308 | // | ||
309 | // if (subCommand == "stop") | ||
310 | // { | ||
311 | // Stop(); | ||
312 | // MainConsole.Instance.OutputFormat("Stopped job engine."); | ||
313 | // } | ||
314 | // else if (subCommand == "start") | ||
315 | // { | ||
316 | // Start(); | ||
317 | // MainConsole.Instance.OutputFormat("Started job engine."); | ||
318 | // } | ||
319 | // else if (subCommand == "status") | ||
320 | // { | ||
321 | // MainConsole.Instance.OutputFormat("Job engine running: {0}", IsRunning); | ||
322 | // MainConsole.Instance.OutputFormat("Current job {0}", m_currentJob != null ? m_currentJob.Name : "none"); | ||
323 | // MainConsole.Instance.OutputFormat( | ||
324 | // "Jobs waiting: {0}", IsRunning ? m_requestQueue.Count.ToString() : "n/a"); | ||
325 | // MainConsole.Instance.OutputFormat("Log Level: {0}", LogLevel); | ||
326 | // } | ||
327 | // | ||
328 | // else if (subCommand == "loglevel") | ||
329 | // { | ||
330 | // // int logLevel; | ||
331 | // int logLevel = int.Parse(args[3]); | ||
332 | // // if (ConsoleUtil.TryParseConsoleInt(MainConsole.Instance, args[4], out logLevel)) | ||
333 | // // { | ||
334 | // LogLevel = logLevel; | ||
335 | // MainConsole.Instance.OutputFormat("Set log level to {0}", LogLevel); | ||
336 | // // } | ||
337 | // } | ||
338 | // else | ||
339 | // { | ||
340 | // MainConsole.Instance.OutputFormat("Unrecognized job engine subcommand {0}", subCommand); | ||
341 | // } | ||
342 | // } | ||
343 | } | ||
344 | } | ||