ogb912/opensim
1
0
Fork 0
mirror of https://github.com/opensim/opensim.git synced 2026-08-06 17:32:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

bad merge?

Browse Source
This commit is contained in:
UbitUmarov
2015-09-01 14:54:35 +01:00
parent f811fdfb70 31a50a73cb
commit 371c9dd2af
406 changed files with 62562 additions and 8067 deletions
Download Patch File Download Diff File Expand all files Collapse all files

221
OpenSim/Framework/Servers/HttpServer/PollServiceRequestManager.cs
View File

@@ -65,15 +65,25 @@ namespace OpenSim.Framework.Servers.HttpServer
private readonly BaseHttpServer m_server;
private Dictionary<PollServiceHttpRequest, Queue<PollServiceHttpRequest>> m_bycontext;
private BlockingQueue<PollServiceHttpRequest> m_requests = new BlockingQueue<PollServiceHttpRequest>();
private static List<PollServiceHttpRequest> m_longPollRequests = new List<PollServiceHttpRequest>();
private static Queue<PollServiceHttpRequest> m_slowRequests = new Queue<PollServiceHttpRequest>();
private static Queue<PollServiceHttpRequest> m_retryRequests = new Queue<PollServiceHttpRequest>();
private uint m_WorkerThreadCount = 0;
private Thread[] m_workerThreads;
private Thread m_retrysThread;
<<<<<<< HEAD
private SmartThreadPool m_threadPool = new SmartThreadPool(20000, 12, 2);
// private int m_timeout = 1000; // increase timeout 250; now use the event one
=======
private bool m_running = true;
private int slowCount = 0;
private SmartThreadPool m_threadPool;
>>>>>>> avn/ubitvar
public PollServiceRequestManager(
BaseHttpServer pSrv, bool performResponsesAsync, uint pWorkerThreadCount, int pTimeout)
@@ -83,6 +93,7 @@ namespace OpenSim.Framework.Servers.HttpServer
m_WorkerThreadCount = pWorkerThreadCount;
m_workerThreads = new Thread[m_WorkerThreadCount];
<<<<<<< HEAD
StatsManager.RegisterStat(
new Stat(
"QueuedPollResponses",
@@ -108,10 +119,25 @@ namespace OpenSim.Framework.Servers.HttpServer
MeasuresOfInterest.AverageChangeOverTime,
stat => stat.Value = ResponsesProcessed,
StatVerbosity.Debug));
=======
PollServiceHttpRequestComparer preqCp = new PollServiceHttpRequestComparer();
m_bycontext = new Dictionary<PollServiceHttpRequest, Queue<PollServiceHttpRequest>>(preqCp);
STPStartInfo startInfo = new STPStartInfo();
startInfo.IdleTimeout = 30000;
startInfo.MaxWorkerThreads = 15;
startInfo.MinWorkerThreads = 1;
startInfo.ThreadPriority = ThreadPriority.Normal;
startInfo.StartSuspended = true;
startInfo.ThreadPoolName = "PoolService";
m_threadPool = new SmartThreadPool(startInfo);
>>>>>>> avn/ubitvar
}
public void Start()
{
<<<<<<< HEAD
IsRunning = true;
if (PerformResponsesAsync)
@@ -139,40 +165,100 @@ namespace OpenSim.Framework.Servers.HttpServer
null,
1000 * 60 * 10);
}
=======
m_threadPool.Start();
//startup worker threads
for (uint i = 0; i < m_WorkerThreadCount; i++)
{
m_workerThreads[i]
= Watchdog.StartThread(
PoolWorkerJob,
string.Format("PollServiceWorkerThread {0}:{1}", i, m_server.Port),
ThreadPriority.Normal,
false,
false,
null,
int.MaxValue);
}
m_retrysThread = Watchdog.StartThread(
this.CheckRetries,
string.Format("PollServiceWatcherThread:{0}", m_server.Port),
ThreadPriority.Normal,
false,
true,
null,
1000 * 60 * 10);
>>>>>>> avn/ubitvar
}
private void ReQueueEvent(PollServiceHttpRequest req)
{
if (IsRunning)
{
// delay the enqueueing for 100ms. There's no need to have the event
// actively on the queue
Timer t = new Timer(self => {
((Timer)self).Dispose();
m_requests.Enqueue(req);
});
t.Change(100, Timeout.Infinite);
lock (m_retryRequests)
m_retryRequests.Enqueue(req);
}
}
public void Enqueue(PollServiceHttpRequest req)
{
if (IsRunning)
lock (m_bycontext)
{
if (req.PollServiceArgs.Type == PollServiceEventArgs.EventType.LongPoll)
Queue<PollServiceHttpRequest> ctxQeueue;
if (m_bycontext.TryGetValue(req, out ctxQeueue))
{
lock (m_longPollRequests)
m_longPollRequests.Add(req);
ctxQeueue.Enqueue(req);
}
else
m_requests.Enqueue(req);
{
ctxQeueue = new Queue<PollServiceHttpRequest>();
m_bycontext[req] = ctxQeueue;
EnqueueInt(req);
}
}
}
private void CheckLongPollThreads()
public void byContextDequeue(PollServiceHttpRequest req)
{
Queue<PollServiceHttpRequest> ctxQeueue;
lock (m_bycontext)
{
if (m_bycontext.TryGetValue(req, out ctxQeueue))
{
if (ctxQeueue.Count > 0)
{
PollServiceHttpRequest newreq = ctxQeueue.Dequeue();
EnqueueInt(newreq);
}
else
{
m_bycontext.Remove(req);
}
}
}
}
public void EnqueueInt(PollServiceHttpRequest req)
{
if (IsRunning)
{
if (req.PollServiceArgs.Type != PollServiceEventArgs.EventType.LongPoll)
{
m_requests.Enqueue(req);
}
else
{
lock (m_slowRequests)
m_slowRequests.Enqueue(req);
}
}
}
private void CheckRetries()
{
<<<<<<< HEAD
// The only purpose of this thread is to check the EQs for events.
// If there are events, that thread will be placed in the "ready-to-serve" queue, m_requests.
// If there are no events, that thread will be back to its "waiting" queue, m_longPollRequests.
@@ -180,13 +266,15 @@ namespace OpenSim.Framework.Servers.HttpServer
// so if they aren't ready to be served by a worker thread (no events), they are placed
// directly back in the "ready-to-serve" queue by the worker thread.
while (IsRunning)
=======
while (m_running)
>>>>>>> avn/ubitvar
{
Thread.Sleep(500);
Thread.Sleep(100); // let the world move .. back to faster rate
Watchdog.UpdateThread();
// List<PollServiceHttpRequest> not_ready = new List<PollServiceHttpRequest>();
lock (m_longPollRequests)
lock (m_retryRequests)
{
<<<<<<< HEAD
if (m_longPollRequests.Count > 0 && IsRunning)
{
List<PollServiceHttpRequest> ready = m_longPollRequests.FindAll(req =>
@@ -199,28 +287,67 @@ namespace OpenSim.Framework.Servers.HttpServer
m_requests.Enqueue(req);
m_longPollRequests.Remove(req);
});
=======
while (m_retryRequests.Count > 0 && m_running)
m_requests.Enqueue(m_retryRequests.Dequeue());
}
slowCount++;
if (slowCount >= 10)
{
slowCount = 0;
>>>>>>> avn/ubitvar
lock (m_slowRequests)
{
while (m_slowRequests.Count > 0 && m_running)
m_requests.Enqueue(m_slowRequests.Dequeue());
}
}
}
}
public void Stop()
{
<<<<<<< HEAD
IsRunning = false;
// m_timeout = -10000; // cause all to expire
=======
m_running = false;
>>>>>>> avn/ubitvar
Thread.Sleep(1000); // let the world move
foreach (Thread t in m_workerThreads)
Watchdog.AbortThread(t.ManagedThreadId);
PollServiceHttpRequest wreq;
// any entry in m_bycontext should have a active request on the other queues
// so just delete contents to easy GC
foreach (Queue<PollServiceHttpRequest> qu in m_bycontext.Values)
qu.Clear();
m_bycontext.Clear();
lock (m_longPollRequests)
try
{
foreach (PollServiceHttpRequest req in m_retryRequests)
{
req.DoHTTPstop(m_server);
}
}
catch
{
}
PollServiceHttpRequest wreq;
m_retryRequests.Clear();
lock (m_slowRequests)
{
<<<<<<< HEAD
if (m_longPollRequests.Count > 0 && IsRunning)
m_longPollRequests.ForEach(req => m_requests.Enqueue(req));
=======
while (m_slowRequests.Count > 0)
m_requests.Enqueue(m_slowRequests.Dequeue());
>>>>>>> avn/ubitvar
}
while (m_requests.Count() > 0)
@@ -228,16 +355,19 @@ namespace OpenSim.Framework.Servers.HttpServer
try
{
wreq = m_requests.Dequeue(0);
<<<<<<< HEAD
ResponsesProcessed++;
wreq.DoHTTPGruntWork(
m_server, wreq.PollServiceArgs.NoEvents(wreq.RequestID, wreq.PollServiceArgs.Id));
=======
wreq.DoHTTPstop(m_server);
>>>>>>> avn/ubitvar
}
catch
{
}
}
m_longPollRequests.Clear();
m_requests.Clear();
}
@@ -247,6 +377,11 @@ namespace OpenSim.Framework.Servers.HttpServer
{
while (IsRunning)
{
<<<<<<< HEAD
=======
PollServiceHttpRequest req = m_requests.Dequeue(5000);
>>>>>>> avn/ubitvar
Watchdog.UpdateThread();
WaitPerformResponse();
}
@@ -265,6 +400,7 @@ namespace OpenSim.Framework.Servers.HttpServer
{
Hashtable responsedata = req.PollServiceArgs.GetEvents(req.RequestID, req.PollServiceArgs.Id);
<<<<<<< HEAD
if (responsedata == null)
return;
@@ -287,11 +423,15 @@ namespace OpenSim.Framework.Servers.HttpServer
else
{
m_threadPool.QueueWorkItem(x =>
=======
if (req.PollServiceArgs.Type == PollServiceEventArgs.EventType.LongPoll) // This is the event queue
>>>>>>> avn/ubitvar
{
try
{
ResponsesProcessed++;
req.DoHTTPGruntWork(m_server, responsedata);
byContextDequeue(req);
}
catch (ObjectDisposedException e) // Browser aborted before we could read body, server closed the stream
{
@@ -300,6 +440,7 @@ namespace OpenSim.Framework.Servers.HttpServer
}
catch (Exception e)
{
<<<<<<< HEAD
m_log.Error(e);
}
@@ -318,6 +459,34 @@ namespace OpenSim.Framework.Servers.HttpServer
else
{
ReQueueEvent(req);
=======
try
{
req.DoHTTPGruntWork(m_server, responsedata);
byContextDequeue(req);
}
catch (ObjectDisposedException) // Browser aborted before we could read body, server closed the stream
{
// Ignore it, no need to reply
}
return null;
}, null);
}
}
else
{
if ((Environment.TickCount - req.RequestTime) > req.PollServiceArgs.TimeOutms)
{
req.DoHTTPGruntWork(m_server,
req.PollServiceArgs.NoEvents(req.RequestID, req.PollServiceArgs.Id));
byContextDequeue(req);
}
else
{
ReQueueEvent(req);
}
>>>>>>> avn/ubitvar
}
}
}
@@ -327,5 +496,7 @@ namespace OpenSim.Framework.Servers.HttpServer
}
}
}
}
}
}