created any_inputs_done logic for done logic

4 files changed, 15 insertions, 12 deletions

diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index f18b9d4..6696b62 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -248,14 +248,11 @@ void BlockActor::handle_task(void)
 
 GRAS_FORCE_INLINE void BlockActor::conclusion(void)
 {
-
+    //missing at least one upstream provider?
     //since nothing else is coming in, its safe to mark done
-    if (this->inputs_done.all()) //no upstream providers
+    if (this->any_inputs_done() or this->forecast_fail)
     {
-        if (not this->input_queues.all_ready() or this->forecast_fail)
-        {
-            this->mark_done();
-        }
+        this->mark_done();
     }
 
     //still have IO ready? kick off another task
diff --git a/lib/gras_impl/block_actor.hpp b/lib/gras_impl/block_actor.hpp
index 0d193b0..745483f 100644
--- a/lib/gras_impl/block_actor.hpp
+++ b/lib/gras_impl/block_actor.hpp
@@ -110,6 +110,11 @@ struct BlockActor : Apology::Worker
     void sort_tags(const size_t index);
     void trim_tags(const size_t index);
     void conclusion(void);
+    GRAS_FORCE_INLINE bool any_inputs_done(void)
+    {
+        if (this->inputs_done.none()) return false;
+        return ((~this->input_queues.ready_bitset()) & this->inputs_done).any();
+    }
 
     //per port properties
     std::vector<size_t> input_items_sizes;
diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp
index 439401c..3bd2ebd 100644
--- a/lib/gras_impl/input_buffer_queues.hpp
+++ b/lib/gras_impl/input_buffer_queues.hpp
@@ -53,6 +53,11 @@ struct InputBufferQueues
 
     void resize(const size_t size);
 
+    GRAS_FORCE_INLINE const BitSet &ready_bitset(void) const
+    {
+        return _bitset;
+    }
+
     GRAS_FORCE_INLINE void push(const size_t i, const SBuffer &buffer)
     {
         ASSERT(not _queues[i].full());
diff --git a/lib/input_handlers.cpp b/lib/input_handlers.cpp
index 8d778f1..574aee4 100644
--- a/lib/input_handlers.cpp
+++ b/lib/input_handlers.cpp
@@ -56,13 +56,9 @@ void BlockActor::handle_input_check(const InputCheckMessage &message, const Ther
 
     //an upstream block declared itself done, recheck the token
     this->inputs_done.set(index, this->input_tokens[index].unique());
-
-    if (this->inputs_done.all()) //no upstream providers
+    if (this->any_inputs_done()) //missing an upstream provider
     {
-        if (not this->input_queues.all_ready())
-        {
-            this->mark_done();
-        }
+        this->mark_done();
     }
 }