Changeset 16779 for trunk/Cosy

Timestamp:

06/09/13 12:51:13 (11 years ago)

Author:

tbretz

Message:

A less CPU hungry version of the msg queue using C++11 techniques.

Location:

trunk/Cosy/base

Files:

• msgqueue.cc (modified) (7 diffs)
• msgqueue.h (modified) (3 diffs)

trunk/Cosy/base/msgqueue.cc

@@ -1 +1 @@
 #include "msgqueue.h"
+
+#include <iostream>
 
 using namespace std;
@@ -7 +9 @@
 // This creates the Message queue thread,
 //
-MsgQueue::MsgQueue() : MThread("MsqQueue"), fNextMsg(0), fNextPtr(0)
+MsgQueue::MsgQueue() : fState(kRun), fThread(std::bind(&MsgQueue::Thread, this))
 {
-    RunThread();
 }
 
@@ -19 +20 @@
 {
     CancelThread();
+}
 
-    fMuxMsg.Lock();
+void MsgQueue::CancelThread()
+{
+    const std::lock_guard<std::mutex> lock(fMutex);
+    if (fState!=kIdle)
+        return;
 
-    if (fNextPtr)
-    {
-        delete [] fNextPtr;
-        fNextPtr = 0;
-    }
-
-    fMuxMsg.UnLock();
+    fState = kAbort;
+    fCond.notify_one();
+    fThread.join();
 }
 
@@ -43 +45 @@
 // the old action can be finished correctly by the user.
 //
-Int_t MsgQueue::Proc(int msg, void *mp)
+int MsgQueue::Proc(int msg, void *mp)
 {
     return 0;
@@ -50 +52 @@
 int MsgQueue::Break() const
 {
-    return fNextMsg>0 || IsThreadCanceled();
+    return fSize>0 || fState!=kRun;
 }
 
@@ -60 +62 @@
 // in this thread. This makes sure, that the calling program is not stalled.
 //
-Int_t MsgQueue::Thread()
+void MsgQueue::Thread()
 {
+    std::unique_lock<std::mutex> lock(fMutex);
+
     while (1)
     {
-        Int_t msg = 0;
-        char *ptr = 0;
+        while (fList.empty() && fState==kRun)
+            fCond.wait(lock);
 
-        fMuxMsg.Lock();
+        if (fState==kAbort)
+            break;
 
-        if (fNextMsg)
-        {
-            msg = fNextMsg;
-            ptr = fNextPtr;
+        if (fState==kStop && fList.empty())
+            break;
 
-            fNextMsg = 0;
-            fNextPtr = 0;
-        }
+        const auto &val = fList.front();
+        fSize--;
 
-        fMuxMsg.UnLock();
+        lock.unlock();
+        Proc(val.first, const_cast<char*>(val.second.data()));
+        lock.lock();
 
-        if (ptr)
-        {
-            Proc(msg, ptr);
-            delete [] ptr;
-        }
+        fList.pop_front();
+    }
 
-        usleep(1);
-        TThread::CancelPoint();
-    }
-    return 0;
+    fList.clear();
+    fSize = 0;
+
+    fState = kIdle;
 }
 
@@ -102 +103 @@
 void *MsgQueue::PostMsg(int msg, void *mp, int size)
 {
-    fMuxMsg.Lock();
+    const std::lock_guard<std::mutex> lock(fMutex);
+    if (fState==kIdle)
+        return false;
 
-    fNextMsg = msg;
+    fSize++;
+    fList.emplace_back(msg, vector<char>(reinterpret_cast<char*>(mp),reinterpret_cast<char*>(mp)+size));
 
-    if (fNextPtr)
-        delete [] fNextPtr;
-    fNextPtr = new char[size];
-
-    memcpy(fNextPtr, mp, size);
-
-    fMuxMsg.UnLock();
+    fCond.notify_one();
 
     return 0;
+
 }

trunk/Cosy/base/msgqueue.h

@@ -2 +2 @@
 #define COSY_MsgQueue
 
-#ifndef ROOT_TMutex
-#include <TMutex.h>
-#endif
-
-#ifndef MARS_MThread
-#include "MThread.h"
+#ifndef __CINT__
+#include <list>
+#include <mutex>
+#include <thread>
+#include <vector>
+#include <condition_variable>
 #endif
 
@@ -13 +13 @@
 #define WM_QUIT 0xffff
 
-class MsgQueue : public MThread
+class MsgQueue
 {
 private:
-    int    fNextMsg;
-    char  *fNextPtr;
+    size_t fSize;                 // Only necessary for before C++11
 
-    TMutex fMuxMsg;
+#ifndef __CINT__
+    std::list<std::pair<int, std::vector<char>>> fList;
 
-    Int_t Thread();
+    std::mutex fMutex;        // Mutex needed for the conditional
+    std::condition_variable fCond; // Conditional
+#endif
+
+    enum state_t
+    {
+        kIdle,
+        kRun,
+        kStop,
+        kAbort,
+    };
+
+    state_t fState;               // Stop signal for the thread
+#ifndef __CINT__
+    std::thread fThread;      // Handle to the thread
+#endif
+
+    void Thread();
 
 public:
@@ -29 +46 @@
     int Break() const;
 
-    virtual Int_t Proc(int msg, void *mp1);
-    Int_t Proc(int msg) { return Proc(msg, 0); }
+    void CancelThread();
+
+    virtual int Proc(int msg, void *mp1);
+    int Proc(int msg) { return Proc(msg, 0); }
 
     void *PostMsg(int msg, void *mp1, int size);