Changeset 17219 for trunk/FACT++

Timestamp:

10/15/13 15:05:54 (11 years ago)

Author:

lyard

Message:

draft zofits working with Queue and MemoryManager

File:

• trunk/FACT++/src/queue.h (modified) (8 diffs)

trunk/FACT++/src/queue.h

@@ -10 +10 @@
 {
     size_t fSize;                 // Only necessary for before C++11
+    bool   fSort;                 // Sort the list before processing
 
     std::list<T> fList;
 
-    std::mutex fMutex;        // Mutex needed for the conditional
+    std::mutex fMutex;             // Mutex needed for the conditional
     std::condition_variable fCond; // Conditional
 
@@ -22 +23 @@
         kStop,
         kAbort,
+        kTrigger
     };
 
     state_t fState;               // Stop signal for the thread
 
-    typedef std::function<void(const T &)> callback;
+    typedef std::function<bool(const T &)> callback;
     callback fCallback;       // Callback function called by the thread
 
@@ -34 +36 @@
     {
         std::unique_lock<std::mutex> lock(fMutex);
+
+        // No filling allowed by default (the queue is
+        // always processed until it is empty)
+        size_t allowed = 0;
 
         while (1)
         {
-            while (fList.empty() && fState==kRun)
+            while (fSize==allowed && fState==kRun)
                 fCond.wait(lock);
 
+            // Check if the State flag has been changed
             if (fState==kAbort)
                 break;
@@ -46 +53 @@
                 break;
 
-            const T &val = fList.front();
+            // If thread got just woken up, move back the state to kRun
+            if (fState == kTrigger)
+                fState = kRun;
+
+            // Could have been a fState==kTrigger case
+            if (fList.empty())
+                continue;
+
+            // During the unlocked state, fSize might change.
+            // The current size of the queue needs to be saved.
+            allowed = fSize;
+
+            // get the first entry from the (sorted) list
+            const auto it = fSort ? min_element(fList.begin(), fList.end()) : fList.begin();
 
             // Theoretically, we can loose a signal here, but this is
@@ -52 +72 @@
             lock.unlock();
 
-            if (fCallback)
-                fCallback(val);
+            // If the first event in the queue could not be processed,
+            // no further processing makes sense until a new event has
+            // been posted (or the number of events in the queue has
+            // changed)  [allowed>0], in the case processing was
+            // successfull [alloed==0], the next event will be processed
+            // immediately.
+            if (!fCallback || !fCallback(*it))
+                allowed = 0;
 
             lock.lock();
 
-            fList.pop_front();
+            // Whenever an event was successfully processed, allowed
+            // is larger than zero and thus the event will be popped
+            if (allowed==0)
+                continue;
+
+            if (fSort)
+                fList.erase(it);
+            else
+                fList.pop_front() ;
+
             fSize--;
+            allowed--;
+
         }
 
@@ -68 +105 @@
 
 public:
-    Queue(const callback &f) : fSize(0), fState(kIdle), fCallback(f)
-    {
-        start();
-    }
+    Queue(const callback &f, bool sort=false, bool startup=true) : fSize(0), fSort(sort), fState(kIdle), fCallback(f)
+    {
+        if (startup)
+            start();
+    }
+
+    Queue(const Queue<T>& q) : fSize(0), fSort(q.fSort), fState(kIdle), fCallback(q.fCallback)
+    {
+    }
+
+    Queue<T>& operator = (const Queue<T>& q)
+    {
+        fSize     = 0;
+        fSort     = q.fSort;
+        fState    = kIdle;
+        fCallback = q.fCallback;
+        return *this;
+    }
+
     ~Queue()
     {
@@ -133 +185 @@
     {
         const std::lock_guard<std::mutex> lock(fMutex);
+
         if (fState==kIdle)
             return false;
@@ -139 +192 @@
         fSize++;
 
+        fCond.notify_one();
+
+        return true;
+    }
+
+    bool notify()
+    {
+        const std::lock_guard<std::mutex> lock(fMutex);
+        if (fState!=kRun)
+            return false;
+
+        fState = kTrigger;
         fCond.notify_one();