PollServiceRequestManager.cs 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331
  1. /*
  2. * Copyright (c) Contributors, http://opensimulator.org/
  3. * See CONTRIBUTORS.TXT for a full list of copyright holders.
  4. *
  5. * Redistribution and use in source and binary forms, with or without
  6. * modification, are permitted provided that the following conditions are met:
  7. * * Redistributions of source code must retain the above copyright
  8. * notice, this list of conditions and the following disclaimer.
  9. * * Redistributions in binary form must reproduce the above copyright
  10. * notice, this list of conditions and the following disclaimer in the
  11. * documentation and/or other materials provided with the distribution.
  12. * * Neither the name of the OpenSimulator Project nor the
  13. * names of its contributors may be used to endorse or promote products
  14. * derived from this software without specific prior written permission.
  15. *
  16. * THIS SOFTWARE IS PROVIDED BY THE DEVELOPERS ``AS IS'' AND ANY
  17. * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  18. * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  19. * DISCLAIMED. IN NO EVENT SHALL THE CONTRIBUTORS BE LIABLE FOR ANY
  20. * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  21. * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  22. * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  23. * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  24. * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  25. * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  26. */
  27. using System;
  28. using System.Collections;
  29. using System.Threading;
  30. using System.Reflection;
  31. using log4net;
  32. using HttpServer;
  33. using OpenSim.Framework;
  34. using OpenSim.Framework.Monitoring;
  35. using Amib.Threading;
  36. using System.IO;
  37. using System.Text;
  38. using System.Collections.Generic;
  39. namespace OpenSim.Framework.Servers.HttpServer
  40. {
  41. public class PollServiceRequestManager
  42. {
  43. private static readonly ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
  44. /// <summary>
  45. /// Is the poll service request manager running?
  46. /// </summary>
  47. /// <remarks>
  48. /// Can be running either synchronously or asynchronously
  49. /// </remarks>
  50. public bool IsRunning { get; private set; }
  51. /// <summary>
  52. /// Is the poll service performing responses asynchronously (with its own threads) or synchronously (via
  53. /// external calls)?
  54. /// </summary>
  55. public bool PerformResponsesAsync { get; private set; }
  56. /// <summary>
  57. /// Number of responses actually processed and sent to viewer (or aborted due to error).
  58. /// </summary>
  59. public int ResponsesProcessed { get; private set; }
  60. private readonly BaseHttpServer m_server;
  61. private BlockingQueue<PollServiceHttpRequest> m_requests = new BlockingQueue<PollServiceHttpRequest>();
  62. private static List<PollServiceHttpRequest> m_longPollRequests = new List<PollServiceHttpRequest>();
  63. private uint m_WorkerThreadCount = 0;
  64. private Thread[] m_workerThreads;
  65. private SmartThreadPool m_threadPool = new SmartThreadPool(20000, 12, 2);
  66. // private int m_timeout = 1000; // increase timeout 250; now use the event one
  67. public PollServiceRequestManager(
  68. BaseHttpServer pSrv, bool performResponsesAsync, uint pWorkerThreadCount, int pTimeout)
  69. {
  70. m_server = pSrv;
  71. PerformResponsesAsync = performResponsesAsync;
  72. m_WorkerThreadCount = pWorkerThreadCount;
  73. m_workerThreads = new Thread[m_WorkerThreadCount];
  74. StatsManager.RegisterStat(
  75. new Stat(
  76. "QueuedPollResponses",
  77. "Number of poll responses queued for processing.",
  78. "",
  79. "",
  80. "httpserver",
  81. m_server.Port.ToString(),
  82. StatType.Pull,
  83. MeasuresOfInterest.AverageChangeOverTime,
  84. stat => stat.Value = m_requests.Count(),
  85. StatVerbosity.Debug));
  86. StatsManager.RegisterStat(
  87. new Stat(
  88. "ProcessedPollResponses",
  89. "Number of poll responses processed.",
  90. "",
  91. "",
  92. "httpserver",
  93. m_server.Port.ToString(),
  94. StatType.Pull,
  95. MeasuresOfInterest.AverageChangeOverTime,
  96. stat => stat.Value = ResponsesProcessed,
  97. StatVerbosity.Debug));
  98. }
  99. public void Start()
  100. {
  101. IsRunning = true;
  102. if (PerformResponsesAsync)
  103. {
  104. //startup worker threads
  105. for (uint i = 0; i < m_WorkerThreadCount; i++)
  106. {
  107. m_workerThreads[i]
  108. = Watchdog.StartThread(
  109. PoolWorkerJob,
  110. string.Format("PollServiceWorkerThread{0}:{1}", i, m_server.Port),
  111. ThreadPriority.Normal,
  112. false,
  113. false,
  114. null,
  115. int.MaxValue);
  116. }
  117. Watchdog.StartThread(
  118. this.CheckLongPollThreads,
  119. string.Format("LongPollServiceWatcherThread:{0}", m_server.Port),
  120. ThreadPriority.Normal,
  121. false,
  122. true,
  123. null,
  124. 1000 * 60 * 10);
  125. }
  126. }
  127. private void ReQueueEvent(PollServiceHttpRequest req)
  128. {
  129. if (IsRunning)
  130. {
  131. // delay the enqueueing for 100ms. There's no need to have the event
  132. // actively on the queue
  133. Timer t = new Timer(self => {
  134. ((Timer)self).Dispose();
  135. m_requests.Enqueue(req);
  136. });
  137. t.Change(100, Timeout.Infinite);
  138. }
  139. }
  140. public void Enqueue(PollServiceHttpRequest req)
  141. {
  142. if (IsRunning)
  143. {
  144. if (req.PollServiceArgs.Type == PollServiceEventArgs.EventType.LongPoll)
  145. {
  146. lock (m_longPollRequests)
  147. m_longPollRequests.Add(req);
  148. }
  149. else
  150. m_requests.Enqueue(req);
  151. }
  152. }
  153. private void CheckLongPollThreads()
  154. {
  155. // The only purpose of this thread is to check the EQs for events.
  156. // If there are events, that thread will be placed in the "ready-to-serve" queue, m_requests.
  157. // If there are no events, that thread will be back to its "waiting" queue, m_longPollRequests.
  158. // All other types of tasks (Inventory handlers, http-in, etc) don't have the long-poll nature,
  159. // so if they aren't ready to be served by a worker thread (no events), they are placed
  160. // directly back in the "ready-to-serve" queue by the worker thread.
  161. while (IsRunning)
  162. {
  163. Thread.Sleep(500);
  164. Watchdog.UpdateThread();
  165. // List<PollServiceHttpRequest> not_ready = new List<PollServiceHttpRequest>();
  166. lock (m_longPollRequests)
  167. {
  168. if (m_longPollRequests.Count > 0 && IsRunning)
  169. {
  170. List<PollServiceHttpRequest> ready = m_longPollRequests.FindAll(req =>
  171. (req.PollServiceArgs.HasEvents(req.RequestID, req.PollServiceArgs.Id) || // there are events in this EQ
  172. (Environment.TickCount - req.RequestTime) > req.PollServiceArgs.TimeOutms) // no events, but timeout
  173. );
  174. ready.ForEach(req =>
  175. {
  176. m_requests.Enqueue(req);
  177. m_longPollRequests.Remove(req);
  178. });
  179. }
  180. }
  181. }
  182. }
  183. public void Stop()
  184. {
  185. IsRunning = false;
  186. // m_timeout = -10000; // cause all to expire
  187. Thread.Sleep(1000); // let the world move
  188. foreach (Thread t in m_workerThreads)
  189. Watchdog.AbortThread(t.ManagedThreadId);
  190. PollServiceHttpRequest wreq;
  191. lock (m_longPollRequests)
  192. {
  193. if (m_longPollRequests.Count > 0 && IsRunning)
  194. m_longPollRequests.ForEach(req => m_requests.Enqueue(req));
  195. }
  196. while (m_requests.Count() > 0)
  197. {
  198. try
  199. {
  200. wreq = m_requests.Dequeue(0);
  201. ResponsesProcessed++;
  202. wreq.DoHTTPGruntWork(
  203. m_server, wreq.PollServiceArgs.NoEvents(wreq.RequestID, wreq.PollServiceArgs.Id));
  204. }
  205. catch
  206. {
  207. }
  208. }
  209. m_longPollRequests.Clear();
  210. m_requests.Clear();
  211. }
  212. // work threads
  213. private void PoolWorkerJob()
  214. {
  215. while (IsRunning)
  216. {
  217. Watchdog.UpdateThread();
  218. WaitPerformResponse();
  219. }
  220. }
  221. public void WaitPerformResponse()
  222. {
  223. PollServiceHttpRequest req = m_requests.Dequeue(5000);
  224. // m_log.DebugFormat("[YYY]: Dequeued {0}", (req == null ? "null" : req.PollServiceArgs.Type.ToString()));
  225. if (req != null)
  226. {
  227. try
  228. {
  229. if (req.PollServiceArgs.HasEvents(req.RequestID, req.PollServiceArgs.Id))
  230. {
  231. Hashtable responsedata = req.PollServiceArgs.GetEvents(req.RequestID, req.PollServiceArgs.Id);
  232. if (responsedata == null)
  233. return;
  234. // This is the event queue.
  235. // Even if we're not running we can still perform responses by explicit request.
  236. if (req.PollServiceArgs.Type == PollServiceEventArgs.EventType.LongPoll
  237. || !PerformResponsesAsync)
  238. {
  239. try
  240. {
  241. ResponsesProcessed++;
  242. req.DoHTTPGruntWork(m_server, responsedata);
  243. }
  244. catch (ObjectDisposedException e) // Browser aborted before we could read body, server closed the stream
  245. {
  246. // Ignore it, no need to reply
  247. m_log.Error(e);
  248. }
  249. }
  250. else
  251. {
  252. m_threadPool.QueueWorkItem(x =>
  253. {
  254. try
  255. {
  256. ResponsesProcessed++;
  257. req.DoHTTPGruntWork(m_server, responsedata);
  258. }
  259. catch (ObjectDisposedException e) // Browser aborted before we could read body, server closed the stream
  260. {
  261. // Ignore it, no need to reply
  262. m_log.Error(e);
  263. }
  264. catch (Exception e)
  265. {
  266. m_log.Error(e);
  267. }
  268. return null;
  269. }, null);
  270. }
  271. }
  272. else
  273. {
  274. if ((Environment.TickCount - req.RequestTime) > req.PollServiceArgs.TimeOutms)
  275. {
  276. ResponsesProcessed++;
  277. req.DoHTTPGruntWork(
  278. m_server, req.PollServiceArgs.NoEvents(req.RequestID, req.PollServiceArgs.Id));
  279. }
  280. else
  281. {
  282. ReQueueEvent(req);
  283. }
  284. }
  285. }
  286. catch (Exception e)
  287. {
  288. m_log.ErrorFormat("Exception in poll service thread: " + e.ToString());
  289. }
  290. }
  291. }
  292. }
  293. }