diff options
Diffstat (limited to 'gnuradio-core/src/lib/runtime')
-rw-r--r-- | gnuradio-core/src/lib/runtime/gr_block.cc | 102 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/gr_block.h | 95 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/gr_block.i | 14 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/gr_block_detail.cc | 120 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/gr_block_detail.h | 49 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/gr_block_executor.cc | 9 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/gr_buffer.cc | 12 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/gr_buffer.h | 9 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/gr_tags.h | 5 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc | 7 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/gr_types.h | 1 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/qa_gr_top_block.cc | 21 | ||||
-rw-r--r-- | gnuradio-core/src/lib/runtime/qa_gr_top_block.h | 7 |
13 files changed, 448 insertions, 3 deletions
diff --git a/gnuradio-core/src/lib/runtime/gr_block.cc b/gnuradio-core/src/lib/runtime/gr_block.cc index 43aebf0bf..fd82ab580 100644 --- a/gnuradio-core/src/lib/runtime/gr_block.cc +++ b/gnuradio-core/src/lib/runtime/gr_block.cc @@ -187,6 +187,13 @@ gr_block::add_item_tag(unsigned int which_output, } void +gr_block::remove_item_tag(unsigned int which_input, + const gr_tag_t &tag) +{ + d_detail->remove_item_tag(which_input, tag); +} + +void gr_block::get_tags_in_range(std::vector<gr_tag_t> &v, unsigned int which_output, uint64_t start, uint64_t end) @@ -244,6 +251,101 @@ gr_block::is_set_max_noutput_items() return d_max_noutput_items_set; } +void +gr_block::set_processor_affinity(const std::vector<unsigned int> &mask) +{ + d_affinity = mask; + if(d_detail) { + d_detail->set_processor_affinity(d_affinity); + } +} + +void +gr_block::unset_processor_affinity() +{ + d_affinity.clear(); + if(d_detail) { + d_detail->unset_processor_affinity(); + } +} + +float +gr_block::pc_noutput_items() +{ + if(d_detail) { + return d_detail->pc_noutput_items(); + } + else { + return 0; + } +} + +float +gr_block::pc_nproduced() +{ + if(d_detail) { + return d_detail->pc_nproduced(); + } + else { + return 0; + } +} + +float +gr_block::pc_input_buffers_full(int which) +{ + if(d_detail) { + return d_detail->pc_input_buffers_full(static_cast<size_t>(which)); + } + else { + return 0; + } +} + +std::vector<float> +gr_block::pc_input_buffers_full() +{ + if(d_detail) { + return d_detail->pc_input_buffers_full(); + } + else { + return std::vector<float>(1,0); + } +} + +float +gr_block::pc_output_buffers_full(int which) +{ + if(d_detail) { + return d_detail->pc_output_buffers_full(static_cast<size_t>(which)); + } + else { + return 0; + } +} + +std::vector<float> +gr_block::pc_output_buffers_full() +{ + if(d_detail) { + return d_detail->pc_output_buffers_full(); + } + else { + return std::vector<float>(1,0); + } +} + +float +gr_block::pc_work_time() +{ + if(d_detail) { + return d_detail->pc_work_time(); + } + else { + return 0; + } +} + std::ostream& operator << (std::ostream& os, const gr_block *m) { diff --git a/gnuradio-core/src/lib/runtime/gr_block.h b/gnuradio-core/src/lib/runtime/gr_block.h index 57e3fda90..c9d2d8f53 100644 --- a/gnuradio-core/src/lib/runtime/gr_block.h +++ b/gnuradio-core/src/lib/runtime/gr_block.h @@ -358,6 +358,64 @@ class GR_CORE_API gr_block : public gr_basic_block { d_min_output_buffer[port] = min_output_buffer; } + // --------------- Performance counter functions ------------- + + /*! + * \brief Gets average noutput_items performance counter. + */ + float pc_noutput_items(); + + /*! + * \brief Gets average num items produced performance counter. + */ + float pc_nproduced(); + + /*! + * \brief Gets average average fullness of \p which input buffer. + */ + float pc_input_buffers_full(int which); + + /*! + * \brief Gets average fullness of all input buffers. + */ + std::vector<float> pc_input_buffers_full(); + + /*! + * \brief Gets average fullness of \p which input buffer. + */ + float pc_output_buffers_full(int which); + + /*! + * \brief Gets average fullness of all output buffers. + */ + std::vector<float> pc_output_buffers_full(); + + /*! + * \brief Gets average clock cycles spent in work. + */ + float pc_work_time(); + + + // ---------------------------------------------------------------------------- + // Functions to handle thread affinity + + /*! + * \brief Set the thread's affinity to processor core \p n. + * + * \param mask a vector of unsigned ints of the core numbers available to this block. + */ + void set_processor_affinity(const std::vector<unsigned int> &mask); + + /*! + * \brief Remove processor affinity to a specific core. + */ + void unset_processor_affinity(); + + /*! + * \brief Get the current processor affinity. + */ + std::vector<unsigned int> processor_affinity() { return d_affinity; } + // ---------------------------------------------------------------------------- private: @@ -373,6 +431,7 @@ class GR_CORE_API gr_block : public gr_basic_block { bool d_max_noutput_items_set; // if d_max_noutput_items is valid int d_max_noutput_items; // value of max_noutput_items for this block tag_propagation_policy_t d_tag_propagation_policy; // policy for moving tags downstream + std::vector<unsigned int> d_affinity; // thread affinity proc. mask protected: gr_block (void){} //allows pure virtual interface sub-classes @@ -416,6 +475,42 @@ class GR_CORE_API gr_block : public gr_basic_block { void add_item_tag(unsigned int which_output, const gr_tag_t &tag); /*! + * \brief Removes a tag from the given input buffer. + * + * \param which_input an integer of which input stream to remove the tag from + * \param abs_offset a uint64 number of the absolute item number + * assicated with the tag. Can get from nitems_written. + * \param key the tag key as a PMT symbol + * \param value any PMT holding any value for the given key + * \param srcid optional source ID specifier; defaults to PMT_F + * + * If no such tag is found, does nothing. + */ + inline void remove_item_tag(unsigned int which_input, + uint64_t abs_offset, + const pmt::pmt_t &key, + const pmt::pmt_t &value, + const pmt::pmt_t &srcid=pmt::PMT_F) + { + gr_tag_t tag; + tag.offset = abs_offset; + tag.key = key; + tag.value = value; + tag.srcid = srcid; + this->remove_item_tag(which_input, tag); + } + + /*! + * \brief Removes a tag from the given input buffer. + * + * If no such tag is found, does nothing. + * + * \param which_input an integer of which input stream to remove the tag from + * \param tag the tag object to remove + */ + void remove_item_tag(unsigned int which_input, const gr_tag_t &tag); + + /*! * \brief Given a [start,end), returns a vector of all tags in the range. * * Range of counts is from start to end-1. diff --git a/gnuradio-core/src/lib/runtime/gr_block.i b/gnuradio-core/src/lib/runtime/gr_block.i index db6c1d04a..c016f2c28 100644 --- a/gnuradio-core/src/lib/runtime/gr_block.i +++ b/gnuradio-core/src/lib/runtime/gr_block.i @@ -66,6 +66,20 @@ class gr_block : public gr_basic_block { void set_min_output_buffer(long min_output_buffer); void set_min_output_buffer(int port, long min_output_buffer); + // Methods to access performance counters + float pc_noutput_items(); + float pc_nproduced(); + float pc_input_buffers_full(int which); + std::vector<float> pc_input_buffers_full(); + float pc_output_buffers_full(int which); + std::vector<float> pc_output_buffers_full(); + float pc_work_time(); + + // Methods to manage processor affinity. + void set_processor_affinity(const gr_vector_uint &mask); + void unset_processor_affinity(); + gr_vector_uint processor_affinity(); + // internal use gr_block_detail_sptr detail () const { return d_detail; } void set_detail (gr_block_detail_sptr detail) { d_detail = detail; } diff --git a/gnuradio-core/src/lib/runtime/gr_block_detail.cc b/gnuradio-core/src/lib/runtime/gr_block_detail.cc index 337c9518e..ff20e0e85 100644 --- a/gnuradio-core/src/lib/runtime/gr_block_detail.cc +++ b/gnuradio-core/src/lib/runtime/gr_block_detail.cc @@ -26,6 +26,7 @@ #include <gr_block_detail.h> #include <gr_buffer.h> +#include <iostream> using namespace pmt; @@ -41,7 +42,11 @@ gr_block_detail::gr_block_detail (unsigned int ninputs, unsigned int noutputs) : d_produce_or(0), d_ninputs (ninputs), d_noutputs (noutputs), d_input (ninputs), d_output (noutputs), - d_done (false) + d_done (false), + d_avg_noutput_items(0), d_avg_nproduced(0), + d_avg_input_buffers_full(ninputs, 0), + d_avg_output_buffers_full(noutputs, 0), + d_avg_work_time(0) { s_ncurrently_allocated++; } @@ -156,6 +161,18 @@ gr_block_detail::add_item_tag(unsigned int which_output, const gr_tag_t &tag) } void +gr_block_detail::remove_item_tag(unsigned int which_input, const gr_tag_t &tag) +{ + if(!pmt_is_symbol(tag.key)) { + throw pmt_wrong_type("gr_block_detail::add_item_tag key", tag.key); + } + else { + // Add tag to gr_buffer's deque tags + d_input[which_input]->buffer()->remove_item_tag(tag); + } +} + +void gr_block_detail::get_tags_in_range(std::vector<gr_tag_t> &v, unsigned int which_input, uint64_t abs_start, @@ -189,3 +206,104 @@ gr_block_detail::get_tags_in_range(std::vector<gr_tag_t> &v, } } } + +void +gr_block_detail::set_processor_affinity(const std::vector<unsigned int> &mask) +{ + if(threaded) { + try { + gruel::thread_bind_to_processor(thread, mask); + } + catch (std::runtime_error e) { + std::cerr << "set_processor_affinity: invalid mask." << std::endl;; + } + } +} + +void +gr_block_detail::unset_processor_affinity() +{ + if(threaded) { + gruel::thread_unbind(thread); + } +} + +void +gr_block_detail::start_perf_counters() +{ + d_start_of_work = gruel::high_res_timer_now(); +} + +void +gr_block_detail::stop_perf_counters(int noutput_items, int nproduced) +{ + float alpha = 0.05; + float beta = 1.0-alpha; + + d_end_of_work = gruel::high_res_timer_now(); + gruel::high_res_timer_type diff = d_end_of_work - d_start_of_work; + d_avg_work_time = beta*d_avg_work_time + alpha*diff; + + d_avg_nproduced = beta*d_avg_nproduced + alpha*nproduced; + d_avg_noutput_items = beta*d_avg_noutput_items + alpha*noutput_items; + + for(size_t i=0; i < d_input.size(); i++) { + float pfull = static_cast<float>(d_input[i]->items_available()) / + static_cast<float>(d_input[i]->max_possible_items_available()); + d_avg_input_buffers_full[i] = beta*d_avg_input_buffers_full[i] + alpha*pfull; + } + + for(size_t i=0; i < d_output.size(); i++) { + float pfull = 1.0f - static_cast<float>(d_output[i]->space_available()) / + static_cast<float>(d_output[i]->bufsize()); + d_avg_output_buffers_full[i] = beta*d_avg_output_buffers_full[i] + alpha*pfull; + } +} + +float +gr_block_detail::pc_noutput_items() +{ + return d_avg_noutput_items; +} + +float +gr_block_detail::pc_nproduced() +{ + return d_avg_nproduced; +} + +float +gr_block_detail::pc_input_buffers_full(size_t which) +{ + if(which < d_avg_input_buffers_full.size()) + return d_avg_input_buffers_full[which]; + else + return 0; +} + +std::vector<float> +gr_block_detail::pc_input_buffers_full() +{ + return d_avg_input_buffers_full; +} + +float +gr_block_detail::pc_output_buffers_full(size_t which) +{ + if(which < d_avg_output_buffers_full.size()) + return d_avg_output_buffers_full[which]; + else + return 0; +} + +std::vector<float> +gr_block_detail::pc_output_buffers_full() +{ + return d_avg_output_buffers_full; +} + +float +gr_block_detail::pc_work_time() +{ + return d_avg_work_time; +} diff --git a/gnuradio-core/src/lib/runtime/gr_block_detail.h b/gnuradio-core/src/lib/runtime/gr_block_detail.h index 16d9f4d42..a8ed8da90 100644 --- a/gnuradio-core/src/lib/runtime/gr_block_detail.h +++ b/gnuradio-core/src/lib/runtime/gr_block_detail.h @@ -27,6 +27,7 @@ #include <gr_runtime_types.h> #include <gr_tpb_detail.h> #include <gr_tags.h> +#include <gruel/high_res_timer.h> #include <stdexcept> /*! @@ -95,8 +96,7 @@ class GR_CORE_API gr_block_detail { /*! * \brief Adds a new tag to the given output stream. * - * This takes the input parameters and builds a PMT tuple - * from it. It then calls gr_buffer::add_item_tag(pmt::pmt_t t), + * Calls gr_buffer::add_item_tag(), * which appends the tag onto its deque. * * \param which_output an integer of which output stream to attach the tag @@ -105,6 +105,16 @@ class GR_CORE_API gr_block_detail { void add_item_tag(unsigned int which_output, const gr_tag_t &tag); /*! + * \brief Removes a tag from the given input stream. + * + * Calls gr_buffer::remove_item_tag(), which removes the tag from its deque. + * + * \param which_input an integer of which input stream to remove the tag from + * \param tag the tag object to add + */ + void remove_item_tag(unsigned int which_input, const gr_tag_t &tag); + + /*! * \brief Given a [start,end), returns a vector of all tags in the range. * * Pass-through function to gr_buffer_reader to get a vector of tags @@ -146,6 +156,33 @@ class GR_CORE_API gr_block_detail { uint64_t abs_end, const pmt::pmt_t &key); + /*! + * \brief Set core affinity of block to the cores in the vector mask. + * + * \param mask a vector of unsigned ints of the core numbers available to this block. + */ + void set_processor_affinity(const std::vector<unsigned int> &mask); + + /*! + * \brief Unset core affinity. + */ + void unset_processor_affinity(); + + bool threaded; // set if thread is currently running. + gruel::gr_thread_t thread; // portable thread handle + + void start_perf_counters(); + void stop_perf_counters(int noutput_items, int nproduced); + + // Calls to get performance counter items + float pc_noutput_items(); + float pc_nproduced(); + float pc_input_buffers_full(size_t which); + std::vector<float> pc_input_buffers_full(); + float pc_output_buffers_full(size_t which); + std::vector<float> pc_output_buffers_full(); + float pc_work_time(); + gr_tpb_detail d_tpb; // used by thread-per-block scheduler int d_produce_or; @@ -158,6 +195,14 @@ class GR_CORE_API gr_block_detail { std::vector<gr_buffer_sptr> d_output; bool d_done; + // Performance counters + float d_avg_noutput_items; + float d_avg_nproduced; + std::vector<float> d_avg_input_buffers_full; + std::vector<float> d_avg_output_buffers_full; + gruel::high_res_timer_type d_start_of_work, d_end_of_work; + float d_avg_work_time; + gr_block_detail (unsigned int ninputs, unsigned int noutputs); friend struct gr_tpb_detail; diff --git a/gnuradio-core/src/lib/runtime/gr_block_executor.cc b/gnuradio-core/src/lib/runtime/gr_block_executor.cc index 375b58f56..ee0ab9e37 100644 --- a/gnuradio-core/src/lib/runtime/gr_block_executor.cc +++ b/gnuradio-core/src/lib/runtime/gr_block_executor.cc @@ -420,9 +420,18 @@ gr_block_executor::run_one_iteration() for (int i = 0; i < d->ninputs(); i++) d_start_nitems_read[i] = d->nitems_read(i); +#ifdef GR_PERFORMANCE_COUNTERS + d->start_perf_counters(); +#endif /* GR_PERFORMANCE_COUNTERS */ + // Do the actual work of the block int n = m->general_work (noutput_items, d_ninput_items, d_input_items, d_output_items); + +#ifdef GR_PERFORMANCE_COUNTERS + d->stop_perf_counters(noutput_items, n); +#endif /* GR_PERFORMANCE_COUNTERS */ + LOG(*d_log << " general_work: noutput_items = " << noutput_items << " result = " << n << std::endl); diff --git a/gnuradio-core/src/lib/runtime/gr_buffer.cc b/gnuradio-core/src/lib/runtime/gr_buffer.cc index b923ca57a..369959d65 100644 --- a/gnuradio-core/src/lib/runtime/gr_buffer.cc +++ b/gnuradio-core/src/lib/runtime/gr_buffer.cc @@ -234,6 +234,18 @@ gr_buffer::add_item_tag(const gr_tag_t &tag) } void +gr_buffer::remove_item_tag(const gr_tag_t &tag) +{ + gruel::scoped_lock guard(*mutex()); + for (std::deque<gr_tag_t>::iterator it = d_item_tags.begin(); it != d_item_tags.end(); ++it) { + if (*it == tag) { + d_item_tags.erase(it); + break; + } + } +} + +void gr_buffer::prune_tags(uint64_t max_time) { /* NOTE: this function _should_ lock the mutex before editing diff --git a/gnuradio-core/src/lib/runtime/gr_buffer.h b/gnuradio-core/src/lib/runtime/gr_buffer.h index 67d48fb2d..28ea97726 100644 --- a/gnuradio-core/src/lib/runtime/gr_buffer.h +++ b/gnuradio-core/src/lib/runtime/gr_buffer.h @@ -103,6 +103,15 @@ class GR_CORE_API gr_buffer { void add_item_tag(const gr_tag_t &tag); /*! + * \brief Removes an existing tag from the buffer. + * + * If no such tag is found, does nothing. + * + * \param tag the tag that needs to be removed + */ + void remove_item_tag(const gr_tag_t &tag); + + /*! * \brief Removes all tags before \p max_time from buffer * * \param max_time the time (item number) to trim up until. diff --git a/gnuradio-core/src/lib/runtime/gr_tags.h b/gnuradio-core/src/lib/runtime/gr_tags.h index 8bffcd0fe..a9ca90235 100644 --- a/gnuradio-core/src/lib/runtime/gr_tags.h +++ b/gnuradio-core/src/lib/runtime/gr_tags.h @@ -45,6 +45,11 @@ struct GR_CORE_API gr_tag_t{ ){ return x.offset < y.offset; } + + inline bool operator == (const gr_tag_t &t) const + { + return (t.key == key) && (t.value == value) && (t.srcid == srcid) && (t.offset == offset); + } }; #endif /*INCLUDED_GR_TAGS_H*/ diff --git a/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc b/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc index 9f17a48a8..cea374fac 100644 --- a/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc +++ b/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc @@ -38,6 +38,13 @@ gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block, int max_noutput_item gr_block_executor::state s; pmt_t msg; + d->threaded = true; + d->thread = gruel::get_current_thread_id(); + + // Set thread affinity if it was set before fg was started. + if(block->processor_affinity().size() > 0) { + gruel::thread_bind_to_processor(d->thread, block->processor_affinity()); + } while (1){ boost::this_thread::interruption_point(); diff --git a/gnuradio-core/src/lib/runtime/gr_types.h b/gnuradio-core/src/lib/runtime/gr_types.h index ad6cee768..db13e456a 100644 --- a/gnuradio-core/src/lib/runtime/gr_types.h +++ b/gnuradio-core/src/lib/runtime/gr_types.h @@ -31,6 +31,7 @@ #include <gr_complex.h> typedef std::vector<int> gr_vector_int; +typedef std::vector<unsigned int> gr_vector_uint; typedef std::vector<float> gr_vector_float; typedef std::vector<double> gr_vector_double; typedef std::vector<void *> gr_vector_void_star; diff --git a/gnuradio-core/src/lib/runtime/qa_gr_top_block.cc b/gnuradio-core/src/lib/runtime/qa_gr_top_block.cc index a0b4755a8..1d3dafadf 100644 --- a/gnuradio-core/src/lib/runtime/qa_gr_top_block.cc +++ b/gnuradio-core/src/lib/runtime/qa_gr_top_block.cc @@ -262,3 +262,24 @@ void qa_gr_top_block::t10_reconfig_max_output_buffer() // Wait for flowgraph to end on its own tb->wait(); } + +void qa_gr_top_block::t11_set_block_affinity() +{ + gr_top_block_sptr tb = gr_make_top_block("top"); + gr_block_sptr src (gr_make_null_source(sizeof(float))); + gr_block_sptr snk (gr_make_null_sink(sizeof(float))); + + std::vector<unsigned int> set(1, 0), ret; + src->set_processor_affinity(set); + + tb->connect(src, 0, snk, 0); + tb->start(); + tb->stop(); + tb->wait(); + + ret = src->processor_affinity(); + + // We only set the core affinity to 0 because we always know at + // least one thread core exists to use. + CPPUNIT_ASSERT_EQUAL(set[0], ret[0]); +} diff --git a/gnuradio-core/src/lib/runtime/qa_gr_top_block.h b/gnuradio-core/src/lib/runtime/qa_gr_top_block.h index bb891abca..634eeab1f 100644 --- a/gnuradio-core/src/lib/runtime/qa_gr_top_block.h +++ b/gnuradio-core/src/lib/runtime/qa_gr_top_block.h @@ -38,6 +38,11 @@ class qa_gr_top_block : public CppUnit::TestCase CPPUNIT_TEST(t4_reconfigure); // triggers 'join never returns' bug CPPUNIT_TEST(t5_max_noutputs); CPPUNIT_TEST(t6_reconfig_max_noutputs); + CPPUNIT_TEST(t7_max_noutputs_per_block); + CPPUNIT_TEST(t8_reconfig_max_noutputs_per_block); + CPPUNIT_TEST(t9_max_output_buffer); + CPPUNIT_TEST(t10_reconfig_max_output_buffer); + CPPUNIT_TEST(t11_set_block_affinity); CPPUNIT_TEST_SUITE_END(); @@ -54,6 +59,8 @@ private: void t8_reconfig_max_noutputs_per_block(); void t9_max_output_buffer(); void t10_reconfig_max_output_buffer(); + void t11_set_block_affinity(); + }; #endif /* INCLUDED_QA_GR_TOP_BLOCK_H */ |