created common function is_work_allowed and input check

at least one port needs a tag or an item,
this helps the reserve items == 0 case

4 files changed, 20 insertions, 6 deletions

diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index 0af7cb1..8b48b61 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -96,6 +96,16 @@ void BlockActor::output_fail(const size_t i)
     this->flush_output(i, true);
 }
 
+GRAS_FORCE_INLINE bool BlockActor::is_work_allowed(void)
+{
+    return (
+        this->block_state == BLOCK_STATE_LIVE and
+        this->input_queues.all_ready() and
+        this->inputs_available.any() and
+        this->output_queues.all_ready()
+    );
+}
+
 void BlockActor::handle_task(void)
 {
     #ifdef WORK_DEBUG
@@ -107,11 +117,7 @@ void BlockActor::handle_task(void)
     //-- Handle task may get called for incoming buffers,
     //-- however, not all ports may have available buffers.
     //------------------------------------------------------------------
-    if (not(
-        this->block_state == BLOCK_STATE_LIVE and
-        this->input_queues.all_ready() and
-        this->output_queues.all_ready()
-    )) return;
+    if (not this->is_work_allowed()) return;
 
     //------------------------------------------------------------------
     //-- Asynchronous notification through atomic variable
@@ -211,11 +217,13 @@ void BlockActor::handle_task(void)
     //since nothing else is coming in, its safe to mark done
     for (size_t i = 0; i < num_inputs; i++)
     {
+        const bool nothing = this->input_queues.empty(i) and this->input_tags[i].empty();
+        this->inputs_available.set(i, not nothing);
         if (this->is_input_done(i)) this->mark_done();
     }
 
     //still have IO ready? kick off another task
-    if (this->input_queues.all_ready() and this->output_queues.all_ready())
+    if (this->is_work_allowed())
     {
         this->Push(SelfKickMessage(), Theron::Address());
     }
diff --git a/lib/gras_impl/block_actor.hpp b/lib/gras_impl/block_actor.hpp
index e44fbe6..d204ec4 100644
--- a/lib/gras_impl/block_actor.hpp
+++ b/lib/gras_impl/block_actor.hpp
@@ -94,6 +94,7 @@ struct BlockActor : Apology::Worker
     void consume(const size_t index, const size_t items);
     void produce_buffer(const size_t index, const SBuffer &buffer);
     void flush_output(const size_t index, const bool force_pop = false);
+    bool is_work_allowed(void);
 
     GRAS_FORCE_INLINE bool is_input_done(const size_t i)
     {
@@ -126,6 +127,7 @@ struct BlockActor : Apology::Worker
     //buffer queues and ready conditions
     InputBufferQueues input_queues;
     OutputBufferQueues<SBuffer> output_queues;
+    BitSet inputs_available;
 
     //tag tracking
     std::vector<bool> input_tags_changed;
diff --git a/lib/input_handlers.cpp b/lib/input_handlers.cpp
index 0069f5b..6b9fd2d 100644
--- a/lib/input_handlers.cpp
+++ b/lib/input_handlers.cpp
@@ -13,6 +13,7 @@ void BlockActor::handle_input_tag(const InputTagMessage &message, const Theron::
     //handle incoming stream tag, push into the tag storage
     this->input_tags[index].push_back(message.tag);
     this->input_tags_changed[index] = true;
+    this->inputs_available.set(index);
     this->handle_task();
 }
 
@@ -24,6 +25,7 @@ void BlockActor::handle_input_buffer(const InputBufferMessage &message, const Th
     //handle incoming stream buffer, push into the queue
     if (this->block_state == BLOCK_STATE_DONE) return;
     this->input_queues.push(index, message.buffer);
+    this->inputs_available.set(index);
     this->handle_task();
 }
 
diff --git a/lib/topology_handler.cpp b/lib/topology_handler.cpp
index 0b8f160..5cd2f79 100644
--- a/lib/topology_handler.cpp
+++ b/lib/topology_handler.cpp
@@ -57,6 +57,8 @@ void BlockActor::handle_topology(
     this->output_items.resize(num_outputs);
     this->input_queues.resize(num_inputs);
     this->output_queues.resize(num_outputs);
+    this->inputs_available.resize(num_inputs);
+    if (num_inputs == 0) this->inputs_available.resize(1, true); //so its always "available"
 
     this->input_tokens.resize(num_inputs);
     this->output_tokens.resize(num_outputs);