1 files changed, 193 insertions, 22 deletions

diff --git a/OpenSim/Framework/Servers/HttpServer/PollServiceRequestManager.cs b/OpenSim/Framework/Servers/HttpServer/PollServiceRequestManager.cs
index 28bba70..4ffe6e5 100644
--- a/OpenSim/Framework/Servers/HttpServer/PollServiceRequestManager.cs
+++ b/OpenSim/Framework/Servers/HttpServer/PollServiceRequestManager.cs
@@ -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);
-                });
+                lock (m_retryRequests)
+                    m_retryRequests.Enqueue(req);
+            }
+        }
 
-                t.Change(100, Timeout.Infinite);
+        public void Enqueue(PollServiceHttpRequest req)
+        {
+            lock (m_bycontext)
+            {
+                Queue<PollServiceHttpRequest> ctxQeueue;
+                if (m_bycontext.TryGetValue(req, out ctxQeueue))
+                {
+                    ctxQeueue.Enqueue(req);
+                }
+                else
+                {
+                    ctxQeueue = new Queue<PollServiceHttpRequest>();
+                    m_bycontext[req] = ctxQeueue;
+                    EnqueueInt(req);
+                }
+            }
+        }
 
+        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 Enqueue(PollServiceHttpRequest req)
+
+        public void EnqueueInt(PollServiceHttpRequest req)
         {
             if (IsRunning)
             {
-                if (req.PollServiceArgs.Type == PollServiceEventArgs.EventType.LongPoll)
+                if (req.PollServiceArgs.Type != PollServiceEventArgs.EventType.LongPoll)
                 {
-                    lock (m_longPollRequests)
-                        m_longPollRequests.Add(req);
+                    m_requests.Enqueue(req);
                 }
                 else
-                    m_requests.Enqueue(req);
+                {
+                    lock (m_slowRequests)
+                        m_slowRequests.Enqueue(req);
+                }
             }
         }
 
-        private void CheckLongPollThreads()
+        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);
 
+            // 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();
+
+            try
+            {
+                foreach (PollServiceHttpRequest req in m_retryRequests)
+                {
+                    req.DoHTTPstop(m_server);
+                }
+            }
+            catch
+            {
+            }
+
             PollServiceHttpRequest wreq;
+            m_retryRequests.Clear();
 
-            lock (m_longPollRequests)
+            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
                 }
             }
         }
+
     }
-}
\ No newline at end of file
+}
+