diff options
Diffstat (limited to 'lib')
| -rw-r--r-- | lib/block_handlers.cpp | 16 | ||||
| -rw-r--r-- | lib/gras_impl/input_buffer_queues.hpp | 5 | ||||
| -rw-r--r-- | lib/gras_impl/stats.hpp | 6 | ||||
| -rw-r--r-- | lib/top_block_query.cpp | 41 |
4 files changed, 64 insertions, 4 deletions
diff --git a/lib/block_handlers.cpp b/lib/block_handlers.cpp index 9865331..1f744f4 100644 --- a/lib/block_handlers.cpp +++ b/lib/block_handlers.cpp @@ -137,11 +137,25 @@ void BlockActor::handle_get_stats( ){ MESSAGE_TRACER(); + //instantaneous states we update here, + //and not interleaved with the rest of the code + const size_t num_inputs = this->get_num_inputs(); + this->stats.items_enqueued.resize(num_inputs); + this->stats.tags_enqueued.resize(num_inputs); + this->stats.msgs_enqueued.resize(num_inputs); + for (size_t i = 0; i < num_inputs; i++) + { + this->stats.items_enqueued[i] = this->input_queues.get_items_enqueued(i); + this->stats.tags_enqueued[i] = this->input_tags[i].size(); + this->stats.msgs_enqueued[i] = this->input_msgs[i].size(); + } + + //create the message reply object GetStatsMessage message; message.block_id = this->block_ptr->to_string(); message.stats = this->stats; message.stats_time = time_now(); - this->Send(message, from); //ACK + this->Send(message, from); //ACK this->highPrioAck(); } diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp index 51ff3e6..48533b9 100644 --- a/lib/gras_impl/input_buffer_queues.hpp +++ b/lib/gras_impl/input_buffer_queues.hpp @@ -146,6 +146,11 @@ struct InputBufferQueues _bitset.set(i, _enqueued_bytes[i] >= _reserve_bytes[i]); } + GRAS_FORCE_INLINE size_t get_items_enqueued(const size_t i) + { + return _enqueued_bytes[i]/_items_sizes[i]; + } + BitSet _bitset; std::vector<size_t> _items_sizes; std::vector<size_t> _enqueued_bytes; diff --git a/lib/gras_impl/stats.hpp b/lib/gras_impl/stats.hpp index 3f78b8d..7edab29 100644 --- a/lib/gras_impl/stats.hpp +++ b/lib/gras_impl/stats.hpp @@ -29,6 +29,7 @@ struct BlockStats time_ticks_t start_time; time_ticks_t stop_time; + //overall tracking of ports std::vector<item_index_t> items_consumed; std::vector<item_index_t> tags_consumed; std::vector<item_index_t> msgs_consumed; @@ -36,6 +37,11 @@ struct BlockStats std::vector<item_index_t> tags_produced; std::vector<item_index_t> msgs_produced; + //instantaneous port status + std::vector<size_t> items_enqueued; + std::vector<size_t> msgs_enqueued; + std::vector<size_t> tags_enqueued; + item_index_t work_count; time_ticks_t time_last_work; time_ticks_t total_time_prep; diff --git a/lib/top_block_query.cpp b/lib/top_block_query.cpp index a8091d2..f00efd2 100644 --- a/lib/top_block_query.cpp +++ b/lib/top_block_query.cpp @@ -5,7 +5,9 @@ #include <boost/foreach.hpp> #include <boost/property_tree/ptree.hpp> #include <boost/property_tree/json_parser.hpp> +#include <boost/property_tree/xml_parser.hpp> #include <boost/regex.hpp> +#include <algorithm> #include <sstream> using namespace gras; @@ -36,13 +38,28 @@ static std::string my_write_json(const boost::property_tree::ptree &pt) return rv; } -std::string TopBlock::query(const std::string &) +static std::string query_blocks(ElementImpl *self, const boost::property_tree::ptree &) +{ + boost::property_tree::ptree root; + boost::property_tree::ptree e; + BOOST_FOREACH(Apology::Worker *worker, self->executor->get_workers()) + { + boost::property_tree::ptree t; + t.put_value(dynamic_cast<BlockActor *>(worker)->block_ptr->to_string()); + e.push_back(std::make_pair("", t)); + } + root.push_back(std::make_pair("blocks", e)); + return my_write_json(root); +} + +static std::string query_stats(ElementImpl *self, const boost::property_tree::ptree &) { //get stats with custom receiver and set high prio GetStatsReceiver receiver; size_t outstandingCount(0); - BOOST_FOREACH(Apology::Worker *worker, (*this)->executor->get_workers()) + BOOST_FOREACH(Apology::Worker *worker, self->executor->get_workers()) { + //send a message to the block's actor to query stats dynamic_cast<BlockActor *>(worker)->highPrioPreNotify(); worker->Push(GetStatsMessage(), receiver.GetAddress()); outstandingCount++; @@ -51,7 +68,6 @@ std::string TopBlock::query(const std::string &) //create root level node boost::property_tree::ptree root; - root.put("id", this->to_string()); root.put("now", time_now()); root.put("tps", time_tps()); @@ -81,6 +97,9 @@ std::string TopBlock::query(const std::string &) } \ block.push_back(std::make_pair(#l, e)); \ } + my_block_ptree_append(items_enqueued); + my_block_ptree_append(tags_enqueued); + my_block_ptree_append(msgs_enqueued); my_block_ptree_append(items_consumed); my_block_ptree_append(tags_consumed); my_block_ptree_append(msgs_consumed); @@ -93,3 +112,19 @@ std::string TopBlock::query(const std::string &) return my_write_json(root); } + +std::string TopBlock::query(const std::string &args) +{ + //why the fuck does no OS ever patch boost when there is a bug + //https://svn.boost.org/trac/boost/ticket/6785 + //serialize the path args into xml -- but I just wanted json + std::stringstream query_args_ss(args); + boost::property_tree::ptree query_args_pt; + boost::property_tree::xml_parser::read_xml(query_args_ss, query_args_pt); + + //dispatch based on path arg + std::string path = query_args_pt.get<std::string>("args.path"); + if (path == "/blocks.json") return query_blocks(this->get(), query_args_pt); + if (path == "/stats.json") return query_stats(this->get(), query_args_pt); + return ""; +} |