More EventQueue tuning.

diff --git a/lib/base/eventqueue.cpp b/lib/base/eventqueue.cpp
index 86001e5071d0cd588c2231a417806408d6851ee8..00ae9b80c8611df93107dc635be4d808ee7eb827 100644 (file)
--- a/lib/base/eventqueue.cpp
+++ b/lib/base/eventqueue.cpp
@@ -37,8 +37,8 @@ EventQueue::EventQueue(void)
        for (int i = 0; i < 2; i++)
                SpawnWorker();
 
-       boost::thread reportThread(boost::bind(&EventQueue::ReportThreadProc, this));
-       reportThread.detach();
+       boost::thread managerThread(boost::bind(&EventQueue::ManagerThreadProc, this));
+       managerThread.detach();
 }
 
 EventQueue::~EventQueue(void)
@@ -97,8 +97,13 @@ void EventQueue::QueueThreadProc(int tid)
 
                        m_ThreadStates[tid] = ThreadBusy;
 
-                       m_Latency += Utility::GetTime() - event.Timestamp;
+                       double latency = Utility::GetTime() - event.Timestamp;
+
+                       m_Latency += latency;
                        m_LatencyCount++;
+
+                       if (latency > m_MaxLatency)
+                               m_MaxLatency = latency;
                }
 
 #ifdef _DEBUG
@@ -178,7 +183,7 @@ void EventQueue::Post(const EventQueueCallback& callback)
        m_CV.notify_one();
 }
 
-void EventQueue::ReportThreadProc(void)
+void EventQueue::ManagerThreadProc(void)
 {
        for (;;) {
                Utility::Sleep(5);
@@ -186,7 +191,7 @@ void EventQueue::ReportThreadProc(void)
                double now = Utility::GetTime();
 
                int pending, alive, busy;
-               double avg_latency;
+               double avg_latency, max_latency;
 
                {
                        boost::mutex::scoped_lock lock(m_Mutex);
@@ -211,7 +216,10 @@ void EventQueue::ReportThreadProc(void)
                        m_Latency = 0;
                        m_LatencyCount = 0;
 
-                       if (pending > alive - busy) {
+                       max_latency = m_MaxLatency;
+                       m_MaxLatency = 0;
+
+                       if (max_latency > 0.1) {
                                /* Spawn a few additional workers. */
                                for (int i = 0; i < 8; i++)
                                        SpawnWorker();
@@ -221,7 +229,9 @@ void EventQueue::ReportThreadProc(void)
                }
 
                std::ostringstream msgbuf;
-               msgbuf << "Pending tasks: " << pending << "; Busy threads: " << busy << "; Idle threads: " << alive - busy << "; Average latency: " << (long)(avg_latency * 1000) << "ms";
+               msgbuf << "Pending tasks: " << pending << "; Busy threads: " << busy
+                   << "; Idle threads: " << alive - busy << "; Average latency: " << (long)(avg_latency * 1000) << "ms"
+                   << "; Max latency: " << (long)(max_latency * 1000) << "ms";
                Log(LogInformation, "base", msgbuf.str());
        }
 }
@@ -233,7 +243,7 @@ void EventQueue::SpawnWorker(void)
 {
        for (int i = 0; i < sizeof(m_ThreadStates) / sizeof(m_ThreadStates[0]); i++) {
                if (m_ThreadStates[i] == ThreadDead) {
-                       Log(LogInformation, "debug", "Spawning worker thread.");
+                       Log(LogDebug, "debug", "Spawning worker thread.");
 
                        m_ThreadStates[i] = ThreadIdle;
                        boost::thread worker(boost::bind(&EventQueue::QueueThreadProc, this, i));
@@ -249,7 +259,7 @@ void EventQueue::SpawnWorker(void)
  */
 void EventQueue::KillWorker(void)
 {
-       Log(LogInformation, "base", "Killing worker thread.");
+       Log(LogDebug, "base", "Killing worker thread.");
 
        m_ThreadDeaths++;
 }

diff --git a/lib/base/eventqueue.h b/lib/base/eventqueue.h
index f173bdb7148a7719bd9454a05224328023b1f127..4ab7661c3cfd79368588daf76421c1538bc5384b 100644 (file)
--- a/lib/base/eventqueue.h
+++ b/lib/base/eventqueue.h
@@ -68,6 +68,8 @@ private:
        double m_Latency;
        int m_LatencyCount;
 
+       double m_MaxLatency;
+
        boost::mutex m_Mutex;
        boost::condition_variable m_CV;
 
@@ -75,7 +77,7 @@ private:
        std::deque<EventQueueWorkItem> m_Events;
 
        void QueueThreadProc(int tid);
-       void ReportThreadProc(void);
+       void ManagerThreadProc(void);
 
        void SpawnWorker(void);
        void KillWorker(void);