tag is just an offset and object, created stream tag and pkt message

4 files changed, 41 insertions, 15 deletions

diff --git a/lib/block.cpp b/lib/block.cpp
index 878bd16..4f7c7c9 100644
--- a/lib/block.cpp
+++ b/lib/block.cpp
@@ -139,18 +139,18 @@ TagIter Block::get_input_tags(const size_t which_input)
     return TagIter(input_tags.begin(), input_tags.end());
 }
 
-void Block::erase_input_tags(const size_t which_input)
+void Block::post_output_msg(const size_t which_output, const PMCC &msg)
 {
-    (*this)->block->input_tags[which_input].clear();
+    this->post_output_tag(which_output, Tag(0, msg));
 }
 
-Tag Block::pop_input_msg(const size_t which_input)
+PMCC Block::pop_input_msg(const size_t which_input)
 {
     std::vector<Tag> &input_tags = (*this)->block->input_tags[which_input];
-    if (input_tags.empty()) return Tag();
-    Tag t = input_tags.front();
+    if (input_tags.empty()) return PMCC();
+    PMCC p = input_tags.front().object;
     input_tags.erase(input_tags.begin());
-    return t;
+    return p;
 }
 
 void Block::propagate_tags(const size_t i, const TagIter &iter)
diff --git a/lib/gras_impl/debug.hpp b/lib/gras_impl/debug.hpp
index c8c7dfe..a90eb6c 100644
--- a/lib/gras_impl/debug.hpp
+++ b/lib/gras_impl/debug.hpp
@@ -25,7 +25,7 @@ extern void *operator new(std::size_t n) throw (std::bad_alloc);
 //-- define to enable these debugs:
 //----------------------------------------------------------------------
 //#define WORK_DEBUG
-#define ASSERTING
+//#define ASSERTING
 //#define MESSAGE_TRACING
 //#define ITEM_CONSPROD
 
diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp
index e8b994b..af0cf37 100644
--- a/lib/gras_impl/input_buffer_queues.hpp
+++ b/lib/gras_impl/input_buffer_queues.hpp
@@ -53,12 +53,7 @@ struct InputBufferQueues
         if (_queues[i].empty()) return get_null_buff();
 
         //there are enough enqueued bytes, but not in the front buffer
-        const bool must_accumulate = _queues[i].front().length < _reserve_bytes[i];
-
-        //should we accumulate? a guess at heuristic improvement
-        const bool should_accumulate = _queues[i].front().length <= 128;
-
-        if (must_accumulate or should_accumulate) this->accumulate(i);
+        if (_queues[i].front().length < _reserve_bytes[i]) this->accumulate(i);
 
         ASSERT(_queues[i].front().length >= _reserve_bytes[i]);
 
diff --git a/lib/tags.cpp b/lib/tags.cpp
index 490c5c5..0b510f6 100644
--- a/lib/tags.cpp
+++ b/lib/tags.cpp
@@ -4,8 +4,8 @@
 
 using namespace gras;
 
-Tag::Tag(const item_index_t &offset, const PMCC &key, const PMCC &value, const PMCC &srcid):
-    offset(offset), key(key), value(value), srcid(srcid)
+Tag::Tag(const item_index_t &offset, const PMCC &object):
+    offset(offset), object(object)
 {
     //NOP
 }
@@ -14,3 +14,34 @@ bool gras::operator<(const Tag &lhs, const Tag &rhs)
 {
     return lhs.offset < rhs.offset;
 }
+
+StreamTag::StreamTag(const PMCC &key, const PMCC &val, const PMCC &src):
+    key(key), val(val), src(src)
+{
+    //NOP
+}
+
+bool gras::operator==(const StreamTag &lhs, const StreamTag &rhs)
+{
+    return
+        lhs.key.eq(rhs.key) and
+        lhs.val.eq(rhs.val) and
+        lhs.src.eq(rhs.src);
+}
+
+PacketMsg::PacketMsg(const PMCC &info, const SBuffer &buff):
+    info(info), buff(buff)
+{
+    //NOP
+}
+
+PacketMsg::PacketMsg(const SBuffer &buff):
+    buff(buff)
+{
+    //NOP
+}
+
+bool gras::operator==(const PacketMsg &lhs, const PacketMsg &rhs)
+{
+    return lhs.info.eq(rhs.info) and lhs.buff == rhs.buff;
+}