1 files changed, 103 insertions, 85 deletions

diff --git a/OpenSim/Framework/Servers/HttpServer/PollServiceRequestManager.cs b/OpenSim/Framework/Servers/HttpServer/PollServiceRequestManager.cs
index c6a3e65..a2f6a11 100644
--- a/OpenSim/Framework/Servers/HttpServer/PollServiceRequestManager.cs
+++ b/OpenSim/Framework/Servers/HttpServer/PollServiceRequestManager.cs
@@ -30,13 +30,10 @@ using System.Collections;
 using System.Threading;
 using System.Reflection;
 using log4net;
-using HttpServer;
-using OpenSim.Framework;
 using OpenSim.Framework.Monitoring;
 using Amib.Threading;
-using System.IO;
-using System.Text;
 using System.Collections.Generic;
+using System.Collections.Concurrent;
 
 namespace OpenSim.Framework.Servers.HttpServer
 {
@@ -46,9 +43,9 @@ 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 Queue<PollServiceHttpRequest> m_retryRequests = new Queue<PollServiceHttpRequest>();
+        private Dictionary<int, Queue<PollServiceHttpRequest>> m_bycontext;
+        private BlockingCollection<PollServiceHttpRequest> m_requests = new BlockingCollection<PollServiceHttpRequest>();
+        private static ConcurrentQueue<PollServiceHttpRequest> m_retryRequests = new ConcurrentQueue<PollServiceHttpRequest>();
 
         private uint m_WorkerThreadCount = 0;
         private Thread[] m_workerThreads;
@@ -65,8 +62,7 @@ namespace OpenSim.Framework.Servers.HttpServer
             m_WorkerThreadCount = pWorkerThreadCount;
             m_workerThreads = new Thread[m_WorkerThreadCount];
 
-            PollServiceHttpRequestComparer preqCp = new PollServiceHttpRequestComparer();
-            m_bycontext = new Dictionary<PollServiceHttpRequest, Queue<PollServiceHttpRequest>>(preqCp);
+            m_bycontext = new Dictionary<int, Queue<PollServiceHttpRequest>>(256);
 
             STPStartInfo startInfo = new STPStartInfo();
             startInfo.IdleTimeout = 30000;
@@ -105,32 +101,28 @@ namespace OpenSim.Framework.Servers.HttpServer
                 true,
                 null,
                 1000 * 60 * 10);
-
-
         }
 
         private void ReQueueEvent(PollServiceHttpRequest req)
         {
             if (m_running)
-            {
-                lock (m_retryRequests)
-                    m_retryRequests.Enqueue(req);
-            }
+                m_retryRequests.Enqueue(req);
         }
 
         public void Enqueue(PollServiceHttpRequest req)
         {
+            Queue<PollServiceHttpRequest> ctxQeueue;
+            int rhash = req.contextHash;
             lock (m_bycontext)
             {
-                Queue<PollServiceHttpRequest> ctxQeueue;
-                if (m_bycontext.TryGetValue(req, out ctxQeueue))
+                if (m_bycontext.TryGetValue(rhash, out ctxQeueue))
                 {
                     ctxQeueue.Enqueue(req);
                 }
                 else
                 {
                     ctxQeueue = new Queue<PollServiceHttpRequest>();
-                    m_bycontext[req] = ctxQeueue;
+                    m_bycontext[rhash] = ctxQeueue;
                     EnqueueInt(req);
                 }
             }
@@ -139,9 +131,10 @@ namespace OpenSim.Framework.Servers.HttpServer
         public void byContextDequeue(PollServiceHttpRequest req)
         {
             Queue<PollServiceHttpRequest> ctxQeueue;
+            int rhash = req.contextHash;
             lock (m_bycontext)
             {
-                if (m_bycontext.TryGetValue(req, out ctxQeueue))
+                if (m_bycontext.TryGetValue(rhash, out ctxQeueue))
                 {
                     if (ctxQeueue.Count > 0)
                     {
@@ -150,30 +143,41 @@ namespace OpenSim.Framework.Servers.HttpServer
                     }
                     else
                     {
-                        m_bycontext.Remove(req);
+                        m_bycontext.Remove(rhash);
                     }
                 }
             }
         }
 
+        public void DropByContext(PollServiceHttpRequest req)
+        {
+            Queue<PollServiceHttpRequest> ctxQeueue;
+            int rhash = req.contextHash;
+            lock (m_bycontext)
+            {
+                if (m_bycontext.TryGetValue(rhash, out ctxQeueue))
+                {
+                    ctxQeueue.Clear();
+                    m_bycontext.Remove(rhash);
+                }
+            }
+        }
+
         public void EnqueueInt(PollServiceHttpRequest req)
         {
             if (m_running)
-                m_requests.Enqueue(req);
+                m_requests.Add(req);
         }
 
         private void CheckRetries()
         {
+            PollServiceHttpRequest preq;
             while (m_running)
-
             {
-                Thread.Sleep(100); // let the world move  .. back to faster rate
+                Thread.Sleep(100);
                 Watchdog.UpdateThread();
-                lock (m_retryRequests)
-                {
-                    while (m_retryRequests.Count > 0 && m_running)
-                        m_requests.Enqueue(m_retryRequests.Dequeue());
-                }
+                while (m_running && m_retryRequests.TryDequeue(out preq))
+                    m_requests.Add(preq);
             }
         }
 
@@ -194,103 +198,117 @@ namespace OpenSim.Framework.Servers.HttpServer
                 qu.Clear();
             m_bycontext.Clear();
 
+            PollServiceHttpRequest req;
             try
             {
-                foreach (PollServiceHttpRequest req in m_retryRequests)
-                {
+                while(m_retryRequests.TryDequeue(out req))
                     req.DoHTTPstop(m_server);
-                }
             }
             catch
             {
             }
 
-            PollServiceHttpRequest wreq;
-
-            m_retryRequests.Clear();
-
-            while (m_requests.Count() > 0)
+            try
+            {
+                while(m_requests.TryTake(out req, 0))
+                    req.DoHTTPstop(m_server);
+            }
+            catch
             {
-                try
-                {
-                    wreq = m_requests.Dequeue(0);
-                    wreq.DoHTTPstop(m_server);
-                }
-                catch
-                {
-                }
             }
 
-            m_requests.Clear();
+            m_requests.Dispose();
+
         }
 
         // work threads
 
         private void PoolWorkerJob()
         {
+            PollServiceHttpRequest req;
             while (m_running)
             {
-                PollServiceHttpRequest req = m_requests.Dequeue(4500);
-                Watchdog.UpdateThread();
-                if (req != null)
+                req = null;
+                if(!m_requests.TryTake(out req, 4500) || req == null)
                 {
-                    try
+                    Watchdog.UpdateThread();
+                    continue;
+                }
+
+                Watchdog.UpdateThread();               
+
+                try
+                {
+                    if(!req.HttpContext.CanSend())
                     {
-                        if (req.PollServiceArgs.HasEvents(req.RequestID, req.PollServiceArgs.Id))
+                        req.PollServiceArgs.Drop(req.RequestID, req.PollServiceArgs.Id);
+                        byContextDequeue(req);
+                        continue;
+                    }
+
+                    if(req.HttpContext.IsSending())
+                    {
+                        if ((Environment.TickCount - req.RequestTime) > req.PollServiceArgs.TimeOutms)
                         {
-                            Hashtable responsedata = req.PollServiceArgs.GetEvents(req.RequestID, req.PollServiceArgs.Id);
+                            req.PollServiceArgs.Drop(req.RequestID, req.PollServiceArgs.Id);
+                            byContextDequeue(req);
+                        }
+                        else
+                          ReQueueEvent(req);
+                        continue;
+                    }
 
+                    if (req.PollServiceArgs.HasEvents(req.RequestID, req.PollServiceArgs.Id))
+                    {
+                        PollServiceHttpRequest nreq = req;
+                        m_threadPool.QueueWorkItem(x =>
+                        {
+                            try
+                            {
+                                Hashtable responsedata = nreq.PollServiceArgs.GetEvents(nreq.RequestID, nreq.PollServiceArgs.Id);
+                                nreq.DoHTTPGruntWork(m_server, responsedata);
+                            }
+                            catch (ObjectDisposedException) { }
+                            finally
+                            {
+                                byContextDequeue(nreq);
+                                nreq = null;
+                            }
+                            return null;
+                        }, null);
+                    }
+                    else
+                    {
+                        if ((Environment.TickCount - req.RequestTime) > req.PollServiceArgs.TimeOutms)
+                        {
+                            PollServiceHttpRequest nreq = req;
                             m_threadPool.QueueWorkItem(x =>
                             {
                                 try
                                 {
-                                    req.DoHTTPGruntWork(m_server, responsedata);
-                                }
-                                catch (ObjectDisposedException)
-                                {
+                                    nreq.DoHTTPGruntWork(m_server,
+                                            nreq.PollServiceArgs.NoEvents(nreq.RequestID, nreq.PollServiceArgs.Id));
                                 }
+                                catch (ObjectDisposedException) {}
                                 finally
                                 {
-                                    byContextDequeue(req);
+                                    byContextDequeue(nreq);
+                                    nreq = null;
                                 }
                                 return null;
                             }, null);
                         }
                         else
                         {
-                            if ((Environment.TickCount - req.RequestTime) > req.PollServiceArgs.TimeOutms)
-                            {
-                                m_threadPool.QueueWorkItem(x =>
-                                {
-                                    try
-                                    {
-                                        req.DoHTTPGruntWork(m_server,
-                                            req.PollServiceArgs.NoEvents(req.RequestID, req.PollServiceArgs.Id));
-                                    }
-                                    catch (ObjectDisposedException)
-                                    {
-                                    // Ignore it, no need to reply
-                                    }
-                                    finally
-                                    {
-                                        byContextDequeue(req);
-                                    }
-                                    return null;
-                                }, null);
-                            }
-                            else
-                            {
-                                ReQueueEvent(req);
-                            }
+                            ReQueueEvent(req);
                         }
                     }
-                    catch (Exception e)
-                    {
-                        m_log.ErrorFormat("Exception in poll service thread: " + e.ToString());
-                    }
+                }
+                catch (Exception e)
+                {
+                    m_log.ErrorFormat("Exception in poll service thread: " + e.ToString());
                 }
             }
         }
-
     }
 }