summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorJosh Blum2013-04-06 21:33:53 -0700
committerJosh Blum2013-04-06 21:33:53 -0700
commit303319d96477d1e2838dda8605568f7b8ee3d95d (patch)
treedd7234805f5a048ef7b0423b40dd9547215085ca
parent74a8855cccdd81c673d0ae3d94a74f94bbbe9bfd (diff)
downloadsandhi-303319d96477d1e2838dda8605568f7b8ee3d95d.tar.gz
sandhi-303319d96477d1e2838dda8605568f7b8ee3d95d.tar.bz2
sandhi-303319d96477d1e2838dda8605568f7b8ee3d95d.zip
gras: cleaner way for msgs to override item reserve
-rw-r--r--lib/gras_impl/block_actor.hpp1
-rw-r--r--lib/gras_impl/input_buffer_queues.hpp6
-rw-r--r--lib/input_handlers.cpp11
3 files changed, 7 insertions, 11 deletions
diff --git a/lib/gras_impl/block_actor.hpp b/lib/gras_impl/block_actor.hpp
index 8a2d5ae..250b4c6 100644
--- a/lib/gras_impl/block_actor.hpp
+++ b/lib/gras_impl/block_actor.hpp
@@ -135,6 +135,7 @@ struct BlockActor : Apology::Worker
const bool has_input_bufs = not this->input_queues.empty(i) and this->input_queues.ready(i);
const bool has_input_msgs = not this->input_msgs[i].empty();
this->inputs_available.set(i, has_input_bufs or has_input_msgs);
+ this->input_queues.update_has_msg(i, has_input_msgs);
}
GRAS_FORCE_INLINE bool is_input_done(const size_t i)
diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp
index 48533b9..7bdbddc 100644
--- a/lib/gras_impl/input_buffer_queues.hpp
+++ b/lib/gras_impl/input_buffer_queues.hpp
@@ -151,6 +151,12 @@ struct InputBufferQueues
return _enqueued_bytes[i]/_items_sizes[i];
}
+ GRAS_FORCE_INLINE void update_has_msg(const size_t i, const bool has)
+ {
+ if (has) _bitset.set(i);
+ else __update(i);
+ }
+
BitSet _bitset;
std::vector<size_t> _items_sizes;
std::vector<size_t> _enqueued_bytes;
diff --git a/lib/input_handlers.cpp b/lib/input_handlers.cpp
index 4702329..c82efb3 100644
--- a/lib/input_handlers.cpp
+++ b/lib/input_handlers.cpp
@@ -23,17 +23,6 @@ void BlockActor::handle_input_msg(const InputMsgMessage &message, const Theron::
MESSAGE_TRACER();
const size_t index = message.index;
- //got an input message? remove the item reserve
- //This is for user convenience so msg ports
- //dont need any special configuration to work.
- if (this->input_configs[index].reserve_items)
- {
- this->input_configs[index].reserve_items = 0;
- InputUpdateMessage message;
- message.index = index;
- this->handle_input_update(message, Theron::Address());
- }
-
//handle incoming async message, push into the msg storage
if (this->block_state == BLOCK_STATE_DONE) return;
this->input_msgs[index].push_back(message.msg);