improve front() / pop_front()

lightful · lightful · commit c17e7d51decf · 2017-04-30T19:25:59.000+02:00
diff --git a/include/sys++/ActorThread.hpp b/include/sys++/ActorThread.hpp
@@ -260,9 +260,11 @@ template <typename Runnable> class ActorThread
 
                 ActorQueue() : head(reinterpret_cast<Item*>(new Aligned<Item>)), // dummy transient placeholder
                                tail(head.load(std::memory_order_relaxed)),
-                               count(0)
+                               count(0),
+                               lastFront(nullptr),
+                               prevFront(tail.load(std::memory_order_relaxed))
                 {
-                    head.load(std::memory_order_relaxed)->next.store(nullptr, std::memory_order_relaxed);
+                    prevFront->next.store(nullptr, std::memory_order_relaxed);
                 }
 
                 void clear() { while (front()) pop_front(); }
@@ -273,27 +275,26 @@ template <typename Runnable> class ActorThread
                     ::operator delete(head.load(std::memory_order_relaxed)); // also suitable for a dummy instance
                 }
 
-                template <typename Linkable> std::size_t emplace_back(Linkable* item) // e.g. any type derived from 'Linked'
+                template <typename Linkable> std::size_t push_back(Linkable* item) // e.g. any type derived from 'Linked'
                 {
                     item->next.store(nullptr, std::memory_order_relaxed);
                     Item* back = head.exchange(item, std::memory_order_acq_rel);
                     back->next.store(item, std::memory_order_release);
                     return count.fetch_add(1, std::memory_order_release); // amount of queued items minus 1
                 }
 
-                inline Item* front() // returns nullptr if empty
+                inline Item* front() // returns nullptr if empty (must be always invoked just before pop_front())
                 {
-                    return tail.load(std::memory_order_relaxed)->next.load(std::memory_order_acquire);
+                    return lastFront = prevFront->next.load(std::memory_order_acquire);
                 }
 
-                inline void pop_front() // also destroys and ultimately deletes the item
+                void pop_front() // also destroys and ultimately deletes the item
                 {
                     count.fetch_sub(1, std::memory_order_release);
-                    Item* top = tail.load(std::memory_order_relaxed);
-                    Item* item = top->next.load(std::memory_order_acquire);
-                    tail.store(item, std::memory_order_release);
-                    item->~Item(); // (atomic destructor is trivial) memory deletion is actually deferred one step behind
-                    ::operator delete(top); // delete the *previous* item memory no longer needed
+                    tail.store(lastFront, std::memory_order_release);
+                    lastFront->~Item(); // (atomic destructor is trivial) memory deletion is actually deferred one step behind
+                    ::operator delete(prevFront); // delete the *previous* item memory no longer needed
+                    prevFront = lastFront;
                 }
 
                 inline std::size_t size() const { return count.load(std::memory_order_acquire); }
@@ -305,6 +306,8 @@ template <typename Runnable> class ActorThread
                 std::atomic<Item*> head; // the stored objects hold the linked list pointers (single memory allocation)
                 std::atomic<Item*> tail;
                 std::atomic<std::size_t> count;
+                Item* lastFront;
+                Item* prevFront;
         };
 
         struct ActorParcel : public ActorQueue<ActorParcel>::Linked
@@ -427,7 +430,7 @@ template <typename Runnable> class ActorThread
         {
             auto& mbox = HighPri? mboxHighPri : mboxNormPri;
             if (!dispatching) return; // don't store anything in a frozen queue
-            bool isIdle = !mbox.emplace_back(new Parcelable(std::forward<Any>(msg)));
+            bool isIdle = mbox.push_back(new Parcelable(std::forward<Any>(msg))) == 0;
             if (HighPri) mboxPaused = false;
             if (!isIdle) return;
             std::lock_guard<std::mutex> lock(mtx);
@@ -466,9 +469,8 @@ template <typename Runnable> class ActorThread
             {
                 bool hasHigh = !mboxHighPri.empty();
                 bool hasNorm = !mboxNormPri.empty();
-                bool isPaused = mboxPaused;
 
-                if (!isPaused && (hasHigh || hasNorm)) // consume the messages queue
+                if (!mboxPaused && (hasHigh || hasNorm)) // consume the messages queue
                 {
                     auto& mbox = hasHigh? mboxHighPri : mboxNormPri;
                     try
@@ -477,7 +479,7 @@ template <typename Runnable> class ActorThread
                         {
                             msg->deliverTo(runnable);
                             mbox.pop_front();
-                            if ((++burst % 16) == 0)
+                            if ((++burst % 64) == 0)
                             {
                                 if (externalDispatcher) // do not monopolize the CPU on this dispatcher
                                 {

Original file line number	Diff line number	Diff line change
`@@ -260,9 +260,11 @@ template <typename Runnable> class ActorThread`
`260`	`260`
`261`	`261`	`ActorQueue() : head(reinterpret_cast<Item*>(new Aligned<Item>)), // dummy transient placeholder`
`262`	`262`	`tail(head.load(std::memory_order_relaxed)),`
`263`		`- count(0)`
	`263`	`+ count(0),`
	`264`	`+ lastFront(nullptr),`
	`265`	`+ prevFront(tail.load(std::memory_order_relaxed))`
`264`	`266`	`{`
`265`		`- head.load(std::memory_order_relaxed)->next.store(nullptr, std::memory_order_relaxed);`
	`267`	`+ prevFront->next.store(nullptr, std::memory_order_relaxed);`
`266`	`268`	`}`
`267`	`269`
`268`	`270`	`void clear() { while (front()) pop_front(); }`
`@@ -273,27 +275,26 @@ template <typename Runnable> class ActorThread`
`273`	`275`	`::operator delete(head.load(std::memory_order_relaxed)); // also suitable for a dummy instance`
`274`	`276`	`}`
`275`	`277`
`276`		`- template <typename Linkable> std::size_t emplace_back(Linkable* item) // e.g. any type derived from 'Linked'`
	`278`	`+ template <typename Linkable> std::size_t push_back(Linkable* item) // e.g. any type derived from 'Linked'`
`277`	`279`	`{`
`278`	`280`	`item->next.store(nullptr, std::memory_order_relaxed);`
`279`	`281`	`Item* back = head.exchange(item, std::memory_order_acq_rel);`
`280`	`282`	`back->next.store(item, std::memory_order_release);`
`281`	`283`	`return count.fetch_add(1, std::memory_order_release); // amount of queued items minus 1`
`282`	`284`	`}`
`283`	`285`
`284`		`- inline Item* front() // returns nullptr if empty`
	`286`	`+ inline Item* front() // returns nullptr if empty (must be always invoked just before pop_front())`
`285`	`287`	`{`
`286`		`- return tail.load(std::memory_order_relaxed)->next.load(std::memory_order_acquire);`
	`288`	`+ return lastFront = prevFront->next.load(std::memory_order_acquire);`
`287`	`289`	`}`
`288`	`290`
`289`		`- inline void pop_front() // also destroys and ultimately deletes the item`
	`291`	`+ void pop_front() // also destroys and ultimately deletes the item`
`290`	`292`	`{`
`291`	`293`	`count.fetch_sub(1, std::memory_order_release);`
`292`		`- Item* top = tail.load(std::memory_order_relaxed);`
`293`		`- Item* item = top->next.load(std::memory_order_acquire);`
`294`		`- tail.store(item, std::memory_order_release);`
`295`		`- item->~Item(); // (atomic destructor is trivial) memory deletion is actually deferred one step behind`
`296`		`- ::operator delete(top); // delete the previous item memory no longer needed`
	`294`	`+ tail.store(lastFront, std::memory_order_release);`
	`295`	`+ lastFront->~Item(); // (atomic destructor is trivial) memory deletion is actually deferred one step behind`
	`296`	`+ ::operator delete(prevFront); // delete the previous item memory no longer needed`
	`297`	`+ prevFront = lastFront;`
`297`	`298`	`}`
`298`	`299`
`299`	`300`	`inline std::size_t size() const { return count.load(std::memory_order_acquire); }`
`@@ -305,6 +306,8 @@ template <typename Runnable> class ActorThread`
`305`	`306`	`std::atomic<Item*> head; // the stored objects hold the linked list pointers (single memory allocation)`
`306`	`307`	`std::atomic<Item*> tail;`
`307`	`308`	`std::atomic<std::size_t> count;`
	`309`	`+ Item* lastFront;`
	`310`	`+ Item* prevFront;`
`308`	`311`	`};`
`309`	`312`
`310`	`313`	`struct ActorParcel : public ActorQueue<ActorParcel>::Linked`
`@@ -427,7 +430,7 @@ template <typename Runnable> class ActorThread`
`427`	`430`	`{`
`428`	`431`	`auto& mbox = HighPri? mboxHighPri : mboxNormPri;`
`429`	`432`	`if (!dispatching) return; // don't store anything in a frozen queue`
`430`		`- bool isIdle = !mbox.emplace_back(new Parcelable(std::forward<Any>(msg)));`
	`433`	`+ bool isIdle = mbox.push_back(new Parcelable(std::forward<Any>(msg))) == 0;`
`431`	`434`	`if (HighPri) mboxPaused = false;`
`432`	`435`	`if (!isIdle) return;`
`433`	`436`	`std::lock_guard<std::mutex> lock(mtx);`
`@@ -466,9 +469,8 @@ template <typename Runnable> class ActorThread`
`466`	`469`	`{`
`467`	`470`	`bool hasHigh = !mboxHighPri.empty();`
`468`	`471`	`bool hasNorm = !mboxNormPri.empty();`
`469`		`- bool isPaused = mboxPaused;`
`470`	`472`
`471`		`- if (!isPaused && (hasHigh \|\| hasNorm)) // consume the messages queue`
	`473`	`+ if (!mboxPaused && (hasHigh \|\| hasNorm)) // consume the messages queue`
`472`	`474`	`{`
`473`	`475`	`auto& mbox = hasHigh? mboxHighPri : mboxNormPri;`
`474`	`476`	`try`
`@@ -477,7 +479,7 @@ template <typename Runnable> class ActorThread`
`477`	`479`	`{`
`478`	`480`	`msg->deliverTo(runnable);`
`479`	`481`	`mbox.pop_front();`
`480`		`- if ((++burst % 16) == 0)`
	`482`	`+ if ((++burst % 64) == 0)`
`481`	`483`	`{`
`482`	`484`	`if (externalDispatcher) // do not monopolize the CPU on this dispatcher`
`483`	`485`	`{`