Hello, I am trying to create an OOT block similar to Delay except that if it does not receive any data for 10ms, it will reset the delay (set d_delta to the same # of samples as the initial delay) so that the next portion of received data will have a delay before it, as seen
here. The inputted data is a cc1110 formatted packet that is converted from PDU to tagged stream, GFSK modulated, then sent though the custom Delay block and transmitted and received by a USRP B210.
The initial peak in the 'with custom Delay block' picture is what confuses me as I don't know why it appears there. The peak appears when the custom Delay block first receives samples after d_delta is reset. My initial thought is that it is the USRP s initializing or something but that's just a wild guess.
The code for my custom Delay block can be found below. Main points of interest are the 'void queue_delay_impl::run()' function which determines if it has been 10ms since the last sample has been received and resets d_delta if so. And the else statement in the general_work function which provides delays by using memset() to insert 0's at the beginning of the data stream when d_delta is reset.
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include <gnuradio/io_signature.h>
#include "queue_delay_impl.h"
#include "ampl_key_impl.h"
#include <vector>
#include <iostream>
#include <string.h>
#include <cstring>
#include <chrono>
#include <sys/time.h>
#include <ctime>
namespace gr {
namespace pduadd {
queue_delay::sptr queue_delay::make(size_t itemsize, int pre_tx)
{
return gnuradio::get_initial_sptr (new queue_delay_impl(itemsize, pre_tx));
}
/*
* The private constructor
*/
queue_delay_impl::queue_delay_impl(size_t itemsize, int pre_tx)
: gr::block("queue_delay",
gr::io_signature::make(1, 1, itemsize),
gr::io_signature::make(1, 1, itemsize)),
d_pre_tx(pre_tx),
d_itemsize(itemsize)
{
//runs when flowgraph starts
if (pre_tx < 0) {
throw std::runtime_error("delay: Cannot initialize block with a Pre Tx < 0.");
}
set_dly(pre_tx);
d_delta = 0;
}
/*
* Our virtual destructor.
*/
queue_delay_impl::~queue_delay_impl()
{}
bool queue_delay_impl::start()
{
d_finished = false;
d_thread = gr::thread::thread(std::bind(&queue_delay_impl::run, this));
return block::start();
}
bool queue_delay_impl::stop()
{
// Shut down the thread fdsafdsafdsafdsafdsa
d_finished = true;
d_thread.interrupt();
d_thread.join();
return block::stop();
}
void queue_delay_impl::run()
{
while (!d_finished) {
boost::this_thread::sleep(
boost::posix_time::milliseconds(static_cast<long>(1)));
if (d_finished) {
return;
}
if(timer > millis){
millis = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
if(timer <= millis && state){
//std::cout << "state change" << std::endl;
state = false;
}
else if(timer <= millis){
d_delta = d_pre_tx;
std::cout << "queue ending millis: " << millis << std::endl;
}
}
}
}
void queue_delay_impl::forecast(int noutput_items, gr_vector_int& ninput_items_required)
{
// make sure all inputs have noutput_items available
unsigned ninputs = ninput_items_required.size();
for (unsigned i = 0; i < ninputs; i++)
ninput_items_required[i] = noutput_items;
}
void queue_delay_impl::set_dly(int d)
{
// only set a new delta if there is a change in the delay; this
// protects from quickly-repeated calls to this function that
// would end with d_delta=0.
//runs when flowgraph starts
if (d != dly()) {
gr::thread::scoped_lock l(d_setlock);
int old = dly();
set_history(d + 1); //changes dly() from 0 to d (pre_tx)
declare_sample_delay(history() - 1);
d_delta += dly() - old;
}
}
int
queue_delay_impl::general_work (int noutput_items,
gr_vector_int &ninput_items,
gr_vector_const_void_star &input_items,
gr_vector_void_star &output_items)
{
gr::thread::scoped_lock l(d_setlock);
assert(input_items.size() == output_items.size());
const char* iptr;
char* optr;
int cons, ret;
//AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
millis = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
timer = millis + 10;
if(start_pre){
start_pre = false;
stop_pre = true;
}
std::cout << "queue work millis: " << millis << std::endl;
//AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
// No change in delay; just memcpy ins to outs
if (d_delta == 0) {//runs a few times at start to pad
//std::cout << "input " << input_items.size() << std::endl;
for (size_t i = 0; i < input_items.size(); i++) {
iptr = (const char*)input_items[i];
optr = (char*)output_items[i];
std::memcpy(optr, iptr, noutput_items * d_itemsize);
auto mi = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
std::cout << "delt0 millis: " << mi << std::endl;
}
cons = noutput_items;
ret = noutput_items;
}
// Skip over d_delta items on the input
else if (d_delta < 0) {
//std::cout << "2" << std::endl;
int n_to_copy, n_adj;
int delta = -d_delta;
n_to_copy = std::max(0, noutput_items - delta);
n_adj = std::min(delta, noutput_items);
for (size_t i = 0; i < input_items.size(); i++) {
iptr = (const char*)input_items[i];
optr = (char*)output_items[i];
std::memcpy(optr, iptr + delta * d_itemsize, n_to_copy * d_itemsize);
}
cons = noutput_items;
ret = n_to_copy;
delta -= n_adj;
d_delta = -delta;
}
// produce but not consume (inserts zeros)
else { // d_delta > 0
//std::cout << "3" << std::endl;
int n_from_input, n_padding;
n_from_input = std::max(0, noutput_items - d_delta);
n_padding = std::min(d_delta, noutput_items);
//std::cout << "pad " << n_padding << std::endl;
for (size_t i = 0; i < input_items.size(); i++) {
iptr = (const char*)input_items[i];
optr = (char*)output_items[i];
std::memset(optr, 0, n_padding * d_itemsize);
std::memcpy(optr, iptr, n_from_input * d_itemsize);
auto mill = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
std::cout << "delta>0 millis: " << mill << std::endl;
//std::cout << "i " << i << std::endl;
}
cons = n_from_input;
ret = noutput_items;
d_delta -= n_padding;
}
consume_each(cons);
return ret;
}
} /* namespace pduadd */
} /* namespace gr */
