From 4f03a5c6264ba3cd888454360f4c6819fa38023d Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Thu, 6 Jun 2013 11:55:53 -0700 Subject: gras: changes for actor in worker Apology commit --- lib/block_produce.cpp | 21 ++++++++------------- 1 file changed, 8 insertions(+), 13 deletions(-) (limited to 'lib/block_produce.cpp') diff --git a/lib/block_produce.cpp b/lib/block_produce.cpp index 31a91f7..5b94dc2 100644 --- a/lib/block_produce.cpp +++ b/lib/block_produce.cpp @@ -14,7 +14,7 @@ void Block::produce(const size_t which_output, const size_t num_items) void Block::produce(const size_t num_items) { - const size_t num_outputs = (*this)->block->get_num_outputs(); + const size_t num_outputs = (*this)->worker->get_num_outputs(); for (size_t o = 0; o < num_outputs; o++) { (*this)->block->produce(o, num_items); @@ -42,7 +42,13 @@ void Block::pop_output_buffer(const size_t which_output, const size_t num_bytes) void Block::post_output_buffer(const size_t which_output, const SBuffer &buffer) { - (*this)->block->produce_buffer(which_output, buffer); + (*this)->block_data->output_queues.consume(which_output); + ASSERT((buffer.length % (*this)->block_data->output_configs[which_output].item_size) == 0); + const size_t items = buffer.length/(*this)->block_data->output_configs[which_output].item_size; + (*this)->block_data->stats.items_produced[which_output] += items; + InputBufferMessage buff_msg; + buff_msg.buffer = buffer; + (*this)->worker->post_downstream(which_output, buff_msg); } GRAS_FORCE_INLINE void BlockActor::produce(const size_t i, const size_t items) @@ -57,14 +63,3 @@ GRAS_FORCE_INLINE void BlockActor::produce(const size_t i, const size_t items) buff.length += bytes; data->produce_outputs[i] = true; } - -GRAS_FORCE_INLINE void BlockActor::produce_buffer(const size_t i, const SBuffer &buffer) -{ - data->output_queues.consume(i); - ASSERT((buffer.length % data->output_configs[i].item_size) == 0); - const size_t items = buffer.length/data->output_configs[i].item_size; - data->stats.items_produced[i] += items; - InputBufferMessage buff_msg; - buff_msg.buffer = buffer; - this->post_downstream(i, buff_msg); -} -- cgit