diff options
Diffstat (limited to 'lib')
| -rw-r--r-- | lib/block.cpp | 6 | ||||
| -rw-r--r-- | lib/block_handlers.cpp | 2 | ||||
| -rw-r--r-- | lib/block_task.cpp | 9 | ||||
| -rw-r--r-- | lib/gras_impl/block_actor.hpp | 6 | ||||
| -rw-r--r-- | lib/gras_impl/input_buffer_queues.hpp | 5 | ||||
| -rw-r--r-- | lib/input_handlers.cpp | 8 |
6 files changed, 23 insertions, 13 deletions
diff --git a/lib/block.cpp b/lib/block.cpp index e4a0d0a..15e5807 100644 --- a/lib/block.cpp +++ b/lib/block.cpp @@ -54,6 +54,7 @@ Block::Block(const std::string &name): this->set_fixed_rate(true); this->set_relative_rate(1.0); this->set_tag_propagation_policy(TPP_ALL_TO_ALL); + this->set_interruptible_work(false); } template <typename V, typename T> @@ -239,3 +240,8 @@ void Block::set_buffer_affinity(const Affinity &affinity) { (*this)->block->buffer_affinity = affinity; } + +void Block::set_interruptible_work(const bool enb) +{ + (*this)->block->interruptible_work = enb; +} diff --git a/lib/block_handlers.cpp b/lib/block_handlers.cpp index 59feecf..2e71ac1 100644 --- a/lib/block_handlers.cpp +++ b/lib/block_handlers.cpp @@ -130,7 +130,7 @@ void BlockActor::handle_top_thread_group( //spawn a new thread if this block is a source this->thread_group = message; this->interruptible_thread.reset(); //erase old one - if (this->get_num_inputs() == 0) //its a source + if (this->interruptible_work) { this->interruptible_thread = boost::make_shared<InterruptibleThread>( this->thread_group, boost::bind(&BlockActor::task_work, this) diff --git a/lib/block_task.cpp b/lib/block_task.cpp index 960b41e..a64b3d7 100644 --- a/lib/block_task.cpp +++ b/lib/block_task.cpp @@ -244,14 +244,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 14bfacc..e3e8463 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; @@ -157,6 +162,7 @@ struct BlockActor : Apology::Worker Block::tag_propagation_policy_t tag_prop_policy; //interruptible thread stuff + bool interruptible_work; SharedThreadGroup thread_group; boost::shared_ptr<InterruptibleThread> interruptible_thread; diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp index 5b610d1..d9d864c 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(); } } |