diff options
| -rw-r--r-- | lib/block_task.cpp | 20 | ||||
| -rw-r--r-- | lib/gras_impl/block_actor.hpp | 2 | ||||
| -rw-r--r-- | lib/input_handlers.cpp | 2 | ||||
| -rw-r--r-- | lib/topology_handler.cpp | 2 |
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); |