diff options
Diffstat (limited to 'lib/gras_impl')
| -rw-r--r-- | lib/gras_impl/block_actor.hpp | 1 | ||||
| -rw-r--r-- | lib/gras_impl/input_buffer_queues.hpp | 16 |
2 files changed, 16 insertions, 1 deletions
diff --git a/lib/gras_impl/block_actor.hpp b/lib/gras_impl/block_actor.hpp index 71fd96e..d32c105 100644 --- a/lib/gras_impl/block_actor.hpp +++ b/lib/gras_impl/block_actor.hpp @@ -110,6 +110,7 @@ struct BlockActor : Apology::Worker void buffer_returner(const size_t index, SBuffer &buffer); void mark_done(void); void handle_task(void); + void input_fail(const size_t index); void sort_tags(const size_t index); void trim_tags(const size_t index); GRAS_FORCE_INLINE bool any_inputs_done(void) diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp index 1b663c3..d17c4f0 100644 --- a/lib/gras_impl/input_buffer_queues.hpp +++ b/lib/gras_impl/input_buffer_queues.hpp @@ -58,6 +58,19 @@ struct InputBufferQueues void accumulate(const size_t i, const size_t item_size); + /*! + * Can we consider this queue's buffers to be accumulated? + * Either the first buffer holds all of the enqueued bytes + * or the first buffer is larger than we can accumulate. + */ + GRAS_FORCE_INLINE bool is_accumulated(const size_t i) const + { + ASSERT(not _queues[i].empty()); + return + (_queues[i].front().length == _enqueued_bytes[i]) or + (_queues[i].front().length >= MAX_AUX_BUFF_BYTES); + } + GRAS_FORCE_INLINE void push(const size_t i, const SBuffer &buffer) { ASSERT(not _queues[i].full()); @@ -182,7 +195,8 @@ GRAS_FORCE_INLINE void InputBufferQueues::accumulate(const size_t i, const size_ } _queues[i].push_front(accum_buff); - return; + + ASSERT(this->is_accumulated(i)); } GRAS_FORCE_INLINE void InputBufferQueues::consume(const size_t i, const size_t bytes_consumed) |