diff --git a/include/foo/periodic_msg_source.h b/include/foo/periodic_msg_source.h index 02f1738..7393fad 100644 --- a/include/foo/periodic_msg_source.h +++ b/include/foo/periodic_msg_source.h @@ -30,6 +30,16 @@ public: typedef boost::shared_ptr sptr; static sptr make(pmt::pmt_t msg, float interval, int num_msg = -1, bool debug = false); + virtual void set_nmsg(int nmsg) = 0; + virtual int get_nmsg() = 0; + + virtual void set_delay(float delay) = 0; + virtual float get_delay() = 0; + + virtual void start_tx() = 0; + virtual void stop_tx() = 0; + virtual bool is_running() = 0; + virtual uint32_t n_sent() = 0; }; } // namespace foo diff --git a/lib/periodic_msg_source_impl.cc b/lib/periodic_msg_source_impl.cc index 07b1786..07e5dfd 100644 --- a/lib/periodic_msg_source_impl.cc +++ b/lib/periodic_msg_source_impl.cc @@ -30,7 +30,8 @@ periodic_msg_source_impl::periodic_msg_source_impl(pmt::pmt_t msg, gr::io_signature::make(0, 0, 0), gr::io_signature::make(0, 0, 0)), d_msg(msg), - d_num_msg(num_msg), + d_nmsg_total(num_msg), + d_nmsg_left(num_msg), d_interval(interval), d_debug(debug), d_finished(false) { @@ -40,19 +41,27 @@ periodic_msg_source_impl::periodic_msg_source_impl(pmt::pmt_t msg, message_port_register_in(pmt::mp("eof in")); set_msg_handler(pmt::mp("eof in"), boost::bind(&periodic_msg_source_impl::eof_in, this, _1)); - d_thread = boost::shared_ptr(new boost::thread(boost::bind(&periodic_msg_source_impl::run, this, this))); + d_thread = new boost::thread(boost::bind(&periodic_msg_source_impl::run, this, this)); } periodic_msg_source_impl::~periodic_msg_source_impl() { + gr::thread::scoped_lock(d_mutex); + d_finished = true; d_thread->interrupt(); d_thread->join(); + delete d_thread; } void periodic_msg_source_impl::eof_in (pmt::pmt_t msg) { + + std::cout << "EOF IN CALLED" << std::endl; + if(pmt::is_eof_object(msg)) { detail().get()->set_done(true); + dout << "stopping msg source" << std::endl; + message_port_pub(pmt::mp("out"), pmt::PMT_EOF); return; } else { dout << "non EOF message at eof port!" << std::endl; @@ -61,31 +70,100 @@ periodic_msg_source_impl::eof_in (pmt::pmt_t msg) { void periodic_msg_source_impl::run(periodic_msg_source_impl *instance) { + + try { + // flow graph startup delay boost::this_thread::sleep(boost::posix_time::milliseconds(500)); - while(!d_finished && d_num_msg) { + std::cout << "starting thread" << std::endl; - dout << "number of messages left: " << d_num_msg << std::endl; + while(1) { + std::cout << "in while loop" << std::endl; + float delay; + { + std::cout << "in scope" << std::endl; + gr::thread::scoped_lock(d_mutex); + std::cout << "got lock" << std::endl; + if(d_finished || !d_nmsg_left) { + d_finished = true; + break; + } - message_port_pub( pmt::mp("out"), d_msg ); + dout << "number of messages left: " << d_nmsg_left << std::endl; - if(d_num_msg > 0) { - d_num_msg--; + message_port_pub( pmt::mp("out"), d_msg ); + + if(d_nmsg_left > 0) { + d_nmsg_left--; + } + + delay = d_interval; } - - if(!d_num_msg) { - break; - } - - boost::this_thread::sleep(boost::posix_time::milliseconds(d_interval)); + boost::this_thread::sleep(boost::posix_time::milliseconds(delay)); } - // delay eof a bit - boost::this_thread::sleep(boost::posix_time::milliseconds(500)); + } catch(boost::thread_interrupted) { + std::cout << "interrupted start" << std::endl; + gr::thread::scoped_lock(d_mutex); + d_finished = true; + std::cout << "interrupted end" << std::endl; + } +} - dout << "stopping msg source" << std::endl; - message_port_pub(pmt::mp("out"), pmt::PMT_EOF); +void +periodic_msg_source_impl::set_nmsg(int nmsg) { + gr::thread::scoped_lock(d_mutex); + d_nmsg_total = nmsg; +} + +int +periodic_msg_source_impl::get_nmsg() { + gr::thread::scoped_lock(d_mutex); + return d_nmsg_total; +} + +void +periodic_msg_source_impl::set_delay(float delay) { + gr::thread::scoped_lock(d_mutex); + d_interval = delay; +} + +float +periodic_msg_source_impl::get_delay() { + gr::thread::scoped_lock(d_mutex); + return d_interval; +} + + +void +periodic_msg_source_impl::start_tx() { + gr::thread::scoped_lock(d_mutex); + + if(is_running()) return; + + d_nmsg_left = d_nmsg_total; + d_finished = false; + d_thread->join(); + delete d_thread; + + d_thread = new boost::thread(boost::bind(&periodic_msg_source_impl::run, this, this)); +} + +void +periodic_msg_source_impl::stop_tx() { + d_thread->interrupt(); +} + +bool +periodic_msg_source_impl::is_running() { + return !d_finished; +} + +uint32_t +periodic_msg_source_impl::n_sent() { + gr::thread::scoped_lock(d_mutex); + return d_nmsg_total - d_nmsg_left; } periodic_msg_source::sptr diff --git a/lib/periodic_msg_source_impl.h b/lib/periodic_msg_source_impl.h index 7f3a97c..3f4e21f 100644 --- a/lib/periodic_msg_source_impl.h +++ b/lib/periodic_msg_source_impl.h @@ -27,18 +27,31 @@ namespace foo { void run(periodic_msg_source_impl *instance); void eof_in(pmt::pmt_t msg); - int d_num_msg; + int d_nmsg_total; + int d_nmsg_left; bool d_debug; bool d_finished; float d_interval; pmt::pmt_t d_msg; - boost::shared_ptr d_thread; + boost::thread *d_thread; + gr::thread::mutex d_mutex; public: periodic_msg_source_impl(pmt::pmt_t msg, float interval, int num_msg, bool debug); virtual ~periodic_msg_source_impl(); + void set_nmsg(int nmsg); + int get_nmsg(); + + void set_delay(float delay); + float get_delay(); + + void start_tx(); + void stop_tx(); + bool is_running(); + uint32_t n_sent(); + }; } // namespace foo