gras: work on stats messages and xml formatting

10 files changed, 120 insertions, 11 deletions

diff --git a/include/gras/top_block.hpp b/include/gras/top_block.hpp
index 2f0dc87..a96cfe8 100644
--- a/include/gras/top_block.hpp
+++ b/include/gras/top_block.hpp
@@ -77,6 +77,12 @@ struct GRAS_API TopBlock : HierBlock
      */
     virtual bool wait(const double timeout);
 
+    /*!
+     * Get block usage statistics in XML format.
+     * An external app will visualize the data.
+     */
+    std::string get_stats_xml(void);
+
     //! Deprecated
     void start(const size_t max_items);
 
diff --git a/lib/block_handlers.cpp b/lib/block_handlers.cpp
index f7a6e9e..af0fc20 100644
--- a/lib/block_handlers.cpp
+++ b/lib/block_handlers.cpp
@@ -129,3 +129,14 @@ void BlockActor::handle_self_kick(
     MESSAGE_TRACER();
     this->handle_task();
 }
+
+void BlockActor::handle_get_stats(
+    const GetStatsMessage &,
+    const Theron::Address from
+){
+    GetStatsMessage message;
+    message.block_id = this->block_ptr->to_string();
+    message.stats = this->stats;
+    message.stats_time = time_now();
+    this->Send(message, from); //ACK
+}
diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index 9acf525..f2621b1 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -225,6 +225,7 @@ void BlockActor::handle_task(void)
     this->stats.work_count++;
     this->stats.total_time_work += work_time;
     this->stats.total_time_work_other += task_time - work_time;
+    this->stats.time_last_work = work_stop;
 }
 
 void BlockActor::consume(const size_t i, const size_t items)
diff --git a/lib/gras_impl/block_actor.hpp b/lib/gras_impl/block_actor.hpp
index ac0bf46..2aee666 100644
--- a/lib/gras_impl/block_actor.hpp
+++ b/lib/gras_impl/block_actor.hpp
@@ -6,12 +6,12 @@
 #include <gras_impl/debug.hpp>
 #include <gras_impl/bitset.hpp>
 #include <gras/gras.hpp>
-#include <gras/chrono.hpp>
 #include <gras/block.hpp>
 #include <gras/top_block.hpp>
 #include <gras/thread_pool.hpp>
 #include <Apology/Worker.hpp>
 #include <gras_impl/token.hpp>
+#include <gras_impl/stats.hpp>
 #include <gras_impl/messages.hpp>
 #include <gras_impl/output_buffer_queues.hpp>
 #include <gras_impl/input_buffer_queues.hpp>
@@ -22,16 +22,6 @@
 namespace gras
 {
 
-struct BlockStats
-{
-    time_ticks_t start_time;
-    time_ticks_t stop_time;
-
-    size_t work_count;
-    time_ticks_t total_time_work;
-    time_ticks_t total_time_work_other;
-};
-
 struct BlockActor : Apology::Worker
 {
     BlockActor(void);
@@ -67,6 +57,7 @@ struct BlockActor : Apology::Worker
         this->RegisterHandler(this, &BlockActor::handle_output_update);
 
         this->RegisterHandler(this, &BlockActor::handle_self_kick);
+        this->RegisterHandler(this, &BlockActor::handle_get_stats);
     }
 
     //handlers
@@ -94,6 +85,7 @@ struct BlockActor : Apology::Worker
     void handle_output_update(const OutputUpdateMessage &, const Theron::Address);
 
     void handle_self_kick(const SelfKickMessage &, const Theron::Address);
+    void handle_get_stats(const GetStatsMessage &, const Theron::Address);
 
     //helpers
     void buffer_returner(const size_t index, SBuffer &buffer);
diff --git a/lib/gras_impl/messages.hpp b/lib/gras_impl/messages.hpp
index 57f308b..386cd0a 100644
--- a/lib/gras_impl/messages.hpp
+++ b/lib/gras_impl/messages.hpp
@@ -8,6 +8,7 @@
 #include <gras/tags.hpp>
 #include <gras/sbuffer.hpp>
 #include <gras_impl/token.hpp>
+#include <gras_impl/stats.hpp>
 
 namespace gras
 {
@@ -128,6 +129,13 @@ struct SelfKickMessage
     //empty
 };
 
+struct GetStatsMessage
+{
+    std::string block_id;
+    BlockStats stats;
+    time_ticks_t stats_time;
+};
+
 } //namespace gras
 
 #include <Theron/Register.h>
@@ -157,5 +165,6 @@ THERON_DECLARE_REGISTERED_MESSAGE(gras::OutputAllocMessage);
 THERON_DECLARE_REGISTERED_MESSAGE(gras::OutputUpdateMessage);
 
 THERON_DECLARE_REGISTERED_MESSAGE(gras::SelfKickMessage);
+THERON_DECLARE_REGISTERED_MESSAGE(gras::GetStatsMessage);
 
 #endif /*INCLUDED_LIBGRAS_IMPL_MESSAGES_HPP*/
diff --git a/lib/gras_impl/stats.hpp b/lib/gras_impl/stats.hpp
new file mode 100644
index 0000000..138b231
--- /dev/null
+++ b/lib/gras_impl/stats.hpp
@@ -0,0 +1,24 @@
+// Copyright (C) by Josh Blum. See LICENSE.txt for licensing information.
+
+#ifndef INCLUDED_LIBGRAS_IMPL_STATS_HPP
+#define INCLUDED_LIBGRAS_IMPL_STATS_HPP
+
+#include <gras/chrono.hpp>
+
+namespace gras
+{
+
+struct BlockStats
+{
+    time_ticks_t start_time;
+    time_ticks_t stop_time;
+
+    size_t work_count;
+    time_ticks_t time_last_work;
+    time_ticks_t total_time_work;
+    time_ticks_t total_time_work_other;
+};
+
+} //namespace gras
+
+#endif /*INCLUDED_LIBGRAS_IMPL_STATS_HPP*/
diff --git a/lib/register_messages.cpp b/lib/register_messages.cpp
index caa8ed8..f68bc14 100644
--- a/lib/register_messages.cpp
+++ b/lib/register_messages.cpp
@@ -26,3 +26,4 @@ THERON_DEFINE_REGISTERED_MESSAGE(gras::OutputAllocMessage);
 THERON_DEFINE_REGISTERED_MESSAGE(gras::OutputUpdateMessage);
 
 THERON_DEFINE_REGISTERED_MESSAGE(gras::SelfKickMessage);
+THERON_DEFINE_REGISTERED_MESSAGE(gras::GetStatsMessage);
diff --git a/lib/top_block.cpp b/lib/top_block.cpp
index 4d2f956..0898e50 100644
--- a/lib/top_block.cpp
+++ b/lib/top_block.cpp
@@ -3,6 +3,8 @@
 #include "element_impl.hpp"
 #include <gras/top_block.hpp>
 #include <boost/thread/thread.hpp> //sleep
+#include <boost/foreach.hpp>
+#include <boost/format.hpp>
 
 using namespace gras;
 
@@ -137,6 +139,40 @@ bool TopBlock::wait(const double timeout)
     return (*this)->token.unique();
 }
 
+///////////////////////// Stats gathering interface ////////////////////////
+
+struct GetStatsReceiver : Theron::Receiver
+{
+    GetStatsReceiver(void)
+    {
+        this->RegisterHandler(this, &GetStatsReceiver::handle_get_stats);
+    }
+
+    void handle_get_stats(const GetStatsMessage &message, const Theron::Address)
+    {
+        this->messages.push_back(message);
+    }
+
+    std::vector<GetStatsMessage> messages;
+};
+
+std::string TopBlock::get_stats_xml(void)
+{
+    GetStatsReceiver receiver;
+    (*this)->executor->post_all(GetStatsMessage(), receiver);
+    std::string xml;
+    BOOST_FOREACH(const GetStatsMessage &message, receiver.messages)
+    {
+        std::string block_xml;
+        block_xml += str(boost::format("    <id>%s</id>\n") % message.block_id);
+        block_xml += str(boost::format("    <tps>%ull</tps>\n") % time_tps());
+        block_xml += str(boost::format("    <start_time>%llu</start_time>\n") % message.stats.start_time);
+        block_xml += str(boost::format("    <stop_time>%llu</stop_time>\n") % message.stats.stop_time);
+        xml += str(boost::format("<block>\n%s</block>\n") % block_xml);
+    }
+    return str(boost::format("<gras_stats>\n%s</gras_stats>") % xml);
+}
+
 ///////////////////////// Deprecated interfaces ////////////////////////
 
 void TopBlock::start(const size_t max_items)
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 1915cfa..ebab7a6 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -12,3 +12,4 @@ GR_ADD_TEST(hier_block_test ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/hie
 GR_ADD_TEST(thread_pool_test ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/thread_pool_test.py)
 GR_ADD_TEST(sbuffer_test ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/sbuffer_test.py)
 GR_ADD_TEST(io_sig_test ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/io_sig_test.py)
+GR_ADD_TEST(stats_test ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/stats_test.py)
diff --git a/tests/stats_test.py b/tests/stats_test.py
new file mode 100644
index 0000000..9d937e9
--- /dev/null
+++ b/tests/stats_test.py
@@ -0,0 +1,28 @@
+# Copyright (C) by Josh Blum. See LICENSE.txt for licensing information.
+
+import unittest
+import gras
+import numpy
+from demo_blocks import *
+
+class StatsTest(unittest.TestCase):
+
+    def setUp(self):
+        self.tb = gras.TopBlock()
+
+    def tearDown(self):
+        self.tb = None
+
+    def test_simple(self):
+        vec_source = VectorSource(numpy.uint32, [0, 9, 8, 7, 6])
+        vec_sink = VectorSink(numpy.uint32)
+
+        self.tb.connect(vec_source, vec_sink)
+        self.tb.run()
+
+        self.assertEqual(vec_sink.get_vector(), (0, 9, 8, 7, 6))
+
+        print self.tb.get_stats_xml()
+
+if __name__ == '__main__':
+    unittest.main()