# Can a tagged stream block be a source block?

**URL:** <https://www.ruby-forum.com/t/can-a-tagged-stream-block-be-a-source-block/233329>\
**Category:** GNU Radio\
**Created:** [December 4, 2014, 2:18pm UTC](https://www.ruby-forum.com/t/can-a-tagged-stream-block-be-a-source-block/233329 "2014-12-04T14:18:56Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![addis\_a](https://www.ruby-forum.com/letter_avatar_proxy/v4/letter/a/c4cdca/32.png) [@addis\_a](https://www.ruby-forum.com/u/addis_a)\
**Post date:** [December 4, 2014, 2:18pm UTC](https://www.ruby-forum.com/t/can-a-tagged-stream-block-be-a-source-block/233329/1 "2014-12-04T14:18:56Z")

</div>

Hi

I am implementing a complete protocol in a module where most of the MAC  
layer will be placed in a tagged stream block. I intend to start out  
implementing the transmit chain and then the receive chain.  
The problem I have run into is that until I get to the receive chain,  
the  
MAC layer block will need to function as a source, but is that at all  
possible based on a tagged stream block?

At the moment the block crashes (silently) in line 102 of the  
gr::tagged\_stream\_block::general\_work.  
The line in question contains: if(d\_n\_input\_items\_reqd[0] == 0)

I believe the cause of the crash is that _d\_n\_input\_items\_reqd_ is not  
set  
or not set to a valid value by the _parse\_length\_tags_ method, since it  
is  
responsible for extracting the length tags on all input ports, but this  
block, being configured as a source block, does not have any input  
ports.

Would it be possible to overcome this by overriding the  
\*parse\_length\_tags \*method  
and setting _d\_n\_input\_items\_reqd_ to something valid?  
And if yes, what would be a valid value?

My implementation of _calculate\_output\_stream\_length_ just returns a  
constant integer.

Best regards  
David Marmoy

For convenience here is the relevant code from gr::tagged\_stream\_block:

void  
tagged\_stream\_block::parse\_length\_tags(const  
std::vector\<std::vector\<tag\_t\> \> &tags,  
gr\_vector\_int  
&n\_input\_items\_reqd)  
{  
for(unsigned i = 0; i \< tags.size(); i++) {  
for(unsigned k = 0; k \< tags[i].size(); k++) {  
if(tags[i][k].key == d\_length\_tag\_key) {  
n\_input\_items\_reqd[i] = pmt::to\_long(tags[i][k].value);  
remove\_item\_tag(i, tags[i][k]);  
}  
}  
}  
}

int  
tagged\_stream\_block::general\_work(int noutput\_items,  
gr\_vector\_int &ninput\_items,  
gr\_vector\_const\_void\_star  
&input\_items,  
gr\_vector\_void\_star &output\_items)  
{  
if(d\_length\_tag\_key\_str.empty()) {  
return work(noutput\_items, ninput\_items, input\_items,  
output\_items);  
}

```
if(d_n_input_items_reqd[0] == 0) { // Otherwise, it's already set 

```

from  
a previous call  
std::vector\<std::vector\<tag\_t\> \> tags(input\_items.size(),  
std::vector\<tag\_t\>());  
for(unsigned i = 0; i \< input\_items.size(); i++) {  
get\_tags\_in\_range(tags[i], i, nitems\_read(i), nitems\_read(i)+1);  
}  
d\_n\_input\_items\_reqd.assign(input\_items.size(), -1);  
parse\_length\_tags(tags, d\_n\_input\_items\_reqd);  
}  
for(unsigned i = 0; i \< input\_items.size(); i++) {  
if(d\_n\_input\_items\_reqd[i] == -1) {  
GR\_LOG\_FATAL(d\_logger, boost::format(“Missing a required length  
tag  
on port %1% at item #%2%”) % i % nitems\_read(i));  
throw std::runtime\_error(“Missing length tag.”);  
}  
if(d\_n\_input\_items\_reqd[i] \> ninput\_items[i]) {  
return 0;  
}  
}

```
int min_output_size =

```

calculate\_output\_stream\_length(d\_n\_input\_items\_reqd);  
if(noutput\_items \< min\_output\_size) {  
set\_min\_noutput\_items(min\_output\_size);  
return 0;  
}  
set\_min\_noutput\_items(1);

```
// WORK CALLED HERE //
int n_produced = work(noutput_items, d_n_input_items_reqd, 

```

input\_items,  
output\_items);  
//////////////////////

```
if(n_produced == WORK_DONE) {
  return n_produced;
}
for(int i = 0; i < (int) d_n_input_items_reqd.size(); i++) {
  consume(i, d_n_input_items_reqd[i]);
}
if (n_produced > 0) {
  update_length_tags(n_produced, output_items.size());
}

d_n_input_items_reqd.assign(input_items.size(), 0);

return n_produced;

```

}

---

<div class="post-metadata">

**Author:** ![David\_Marmoy](https://www.ruby-forum.com/letter_avatar_proxy/v4/letter/d/58956e/32.png) [@David\_Marmoy](https://www.ruby-forum.com/u/David_Marmoy)\
**Post date:** [December 4, 2014, 4:10pm UTC](https://www.ruby-forum.com/t/can-a-tagged-stream-block-be-a-source-block/233329/2 "2014-12-04T16:10:03Z")

</div>

On 12/04/2014 02:17 PM, David Marmoy wrote:

> I am implementing a complete protocol in a module where most of the MAC  
> layer will be placed in a tagged stream block. I intend to start out  
> implementing the transmit chain and then the receive chain.  
> The problem I have run into is that until I get to the receive chain,  
> the MAC layer block will need to function as a source, but is that at  
> all possible based on a tagged stream block?

Short answer: Erh, maybe. But not without changing the runtime.

Long answer: This a known shortcoming of tagged stream blocks. For 3.8,  
we will most likely be changing the structure of those blocks, and this  
will become possible.

Right now, we could update the code in tagged\_stream\_block to make this  
happen, but it would involve more testing and branches in the work  
function. It would not be a big problem to fix, though, you’re just the  
first to actually request this.

Actually, it’s probably something I should fix before people start  
writing tags by hand. David, until I have that ready you should change  
that offending line to:

102 if(!d\_n\_input\_items\_reqd.empty() && d\_n\_input\_items\_reqd[0] ==  
0) { // Otherwise, it’s already set from a previous call

parse\_length\_tags() will then never be called.  
calculate\_output\_stream\_length() however should return a valid length.

Cheers,  
M

---

<div class="post-metadata">

**Author:** ![David\_Marmoy](https://www.ruby-forum.com/letter_avatar_proxy/v4/letter/d/58956e/32.png) [@David\_Marmoy](https://www.ruby-forum.com/u/David_Marmoy)\
**Post date:** [December 4, 2014, 7:41pm UTC](https://www.ruby-forum.com/t/can-a-tagged-stream-block-be-a-source-block/233329/3 "2014-12-04T19:41:58Z")

</div>

On 12/04/2014 04:08 PM, Martin B. wrote:

> Actually, it’s probably something I should fix before people start  
> writing tags by hand. David, until I have that ready you should change  
> that offending line to:
> 
> 102 if(!d\_n\_input\_items\_reqd.empty() && d\_n\_input\_items\_reqd[0] ==  
> 0) { // Otherwise, it’s already set from a previous call
> 
> parse\_length\_tags() will then never be called.  
> calculate\_output\_stream\_length() however should return a valid length.

This was actually simpler than expected:  
[https://github.com/mbr0wn/gnuradio/tree/tsb/allow\_source](https://github.com/mbr0wn/gnuradio/tree/tsb/allow_source)

I also put an example in there (see the 2nd commit).

Cheers,  
M

---

<div class="post-metadata">

**Author:** ![David\_Marmoy](https://www.ruby-forum.com/letter_avatar_proxy/v4/letter/d/58956e/32.png) [@David\_Marmoy](https://www.ruby-forum.com/u/David_Marmoy)\
**Post date:** [December 4, 2014, 10:25pm UTC](https://www.ruby-forum.com/t/can-a-tagged-stream-block-be-a-source-block/233329/4 "2014-12-04T22:25:17Z")

</div>

> This was actually simpler than expected:  
> [https://github.com/mbr0wn/gnuradio/tree/tsb/allow\_source](https://github.com/mbr0wn/gnuradio/tree/tsb/allow_source)
> 
> I also put an example in there (see the 2nd commit).

Just to understand the implications of this change to pdu\_to\_tagged  
stream, max input PDU size is now effectively constrained by the max  
buffer size that can be handled by the scheduler and passed to work().  
Correct? That may not be a limitation for many people, and I suppose it  
would have surfaced before if someone tried to send a larger PDU into a  
pdu\_to\_tagged\_stream block connected to a downstream tsb-derived block.

I guess the upside is that now you only need one call to work() per PDU  
and you don’t need an internal buffer and a stateful work() to push a  
PDU through.

Sean

---

<div class="post-metadata">

**Author:** ![David\_Marmoy](https://www.ruby-forum.com/letter_avatar_proxy/v4/letter/d/58956e/32.png) [@David\_Marmoy](https://www.ruby-forum.com/u/David_Marmoy)\
**Post date:** [December 5, 2014, 11:16am UTC](https://www.ruby-forum.com/t/can-a-tagged-stream-block-be-a-source-block/233329/5 "2014-12-05T11:16:54Z")

</div>

On 12/04/2014 10:23 PM, Nowlan, Sean wrote:

> larger PDU into a pdu\_to\_tagged\_stream block connected to a  
> downstream tsb-derived block.

Exactly. It kind of adds a new constraint, but not really, because  
downstream blocks already bring this constraint with them.

> I guess the upside is that now you only need one call to work() per  
> PDU and you don’t need an internal buffer and a stateful work() to  
> push a PDU through.

Not quite, unfortunately: With or without usage of TSBs, we need to spin  
inside work. I gave it a rough benchmark (using htop), and both versions  
of pdu\_to\_tagged\_stream use up a lot of cycles. This would probably be  
better if we blocked on popping the message Q, but that’s something we  
don’t want in work functions.

The statefulness is still gone, though. If we’re spinning on work  
because the Q was empty, it returns pretty much immediately. Otherwise,  
the work() function is much tidier and uses less copying to intermediate  
buffers.

We could probably sack all the class attributes if we popped the message  
Q inside work() instead of (ab-)using calculate\_output\_stream\_length(),  
but the point of that method is to provide the parent block with the  
correct output length, for which we need to read the PDU.

More importantly, we remove TSB tag handling from that block. If we go  
to tagging the last item, we want no blocks to do the TSB tag handling  
themselves.

M