[BUG] Steam Buffer doesn't block when buffer contain data but less than Trigger Level

[cperkulator]

Describe the bug
Blocking on a Stream Buffer only appears to work if the buffer is empty when xStreamBufferReceive is called.

Target

• Development board: Custom
• Instruction Set Architecture: Tricore
• IDE and version: [e.g. Freedom Studio 4.12.0.2019-08-2]
• Toolchain and version: [e.g. riscv64-unknown-elf-gcc-8.3.0-2019.08.0]

Host

• Host OS: [e.g. MacOS]
• Version: [e.g. Mojave 10.14.6]

To Reproduce

Trigger level = 2 (for simplicity)

      Reader Task:                                                                                   Writer Task/Interrupt:
      -------------                                                                                    -------------
       (Kickoff reading loop)
       xStreamBufferReceive (handle, buf_ptr, 4, portMAX_DELAY)
       blocks because no data in buffer      
                                                                                                               (Kickoff writing loop)
                                                                                                               write sample 1 into the StreamBuffer
                                                                                                               write sample 2 into the StreamBuffer
                                                                                                               trigger level reached, notify blocked reader
      unblocked 
      read 2 bytes since trigger level was 2
                                                                                                               interrupt to write sample
                                                                                                               write sample 1 into the StreamBuffer
      xStreamBufferReceive (handle, buf_ptr, 4, portMAX_DELAY)
     doesn't block because data is in the buffer
     reads 1 byte despite trigger level being 2!

Expected behavior

Trigger level = 2 (for simplicity)

      Reader Task:                                                                                   Writer Task/Interrupt:
      -------------                                                                                    -------------
       (Kickoff reading loop)
       xStreamBufferReceive (handle, buf_ptr, 4, portMAX_DELAY)
       blocks because no data in buffer      
                                                                                                               (Kickoff writing loop)
                                                                                                               write sample 1 into the StreamBuffer
                                                                                                               write sample 2 into the StreamBuffer
                                                                                                               trigger level reached, notify blocked reader
      unblocked 
      read 2 bytes since trigger level was 2
                                                                                                               interrupt to write sample
                                                                                                               write sample 1 into the StreamBuffer
      xStreamBufferReceive (handle, buf_ptr, 4, portMAX_DELAY)
     blocks because trigger level has not been reached yet
                                                                                                               write sample 2 into the StreamBuffer
                                                                                                               trigger level reached, notify blocked reader
      unblocked 
      read 2 bytes since trigger level was 2

Additional context
This code will not block when if the amount of data in the buffer is less than the trigger level.

This happens because xBytesToStoreMessageLength is 0 for a steam buffer. This if is skipped if the buffer contains any amount of data thus not putting the task to sleep even if the amount of data is less than the trigger level.

I will be submitting a PR shortly

[RichardBarry]

I've not had time to study in detail - but I think you are reporting the intended behaviour. From the docs: "A stream buffer's trigger level is the number of bytes that must be in the stream buffer before a task that is blocked on the stream buffer to wait for data is moved out of the blocked state." i.e. the trigger level is only used to unblock a task that is already blocked, not cause a task to enter the blocked state.

[cperkulator]

I've not had time to study in detail - but I think you are reporting the intended behaviour. From the docs: "A stream buffer's trigger level is the number of bytes that must be in the stream buffer before a task that is blocked on the stream buffer to wait for data is moved out of the blocked state." i.e. the trigger level is only used to unblock a task that is already blocked, not cause a task to enter the blocked state.

but it does cause a task to block, but only when the buffer is empty.

As mentioned in this issue it doesn't quite make sense to have stream buffers only block if the buffer is empty. With the current implementation, you would have to put the stream buffer receive in a loop until you get the amount of data you want.

[RichardBarry]

Again I'm not in a position to investigate deeply - so I may be missing what you are saying - but that is the intended behaviour. This page describes the stream buffer's receive block time as: "The maximum amount of time the task should remain in the Blocked state to wait for data to become available if the stream buffer is empty. Read through the sample scenarios described on that page to see if you think anything is incorrect - in which case we will naturally either fix the code or the documentation.

The intent of the trigger level is to prevent a task from rapidly transitioning between the blocked and running states if single bytes come in slowly. That won't happen if there are already bytes available - the reading task will return the bytes immediately without ever entering the blocked state. If there aren't any bytes available, and a block time is specified, then the task will block until either the block time expires or the trigger level is hit - only entering the Blocked state once in each case.

[cperkulator]

So the stream buffer does function as you specify. I'm confused on why that is the desired behavior, however.

The current implementation would require the user to read all of the bytes out of the buffer if they expect their task to go back to sleep until the next trigger level is achieved.

Given the current implementation, I wonder what your suggestion would be for the situation in which we are trying to use the stream buffer for. We have a fast task that queues up a sample one at a time and a slow task that processes 10 samples at a time. We were hoping to use the stream buffer, so that our read task would always sleep until 10 samples are available. However, if there is less than 10 samples in the buffer, our task just returns whatever is left (as you mention). So now we have to call xStreamBufferReceive in a loop until we get the amount of data we intend to have. This results in extra copies and context switches that we were hoping to avoid. Is that the intended behavior or is there a better primitive we should use?

Also I would really appreciate you taking a look at my PR on how I would expect this to work.

And thanks for the conversation on this!

[aggarg]

There are two possible scenarios -

1. When you call xStreamBufferReceive, if there is any data in the buffer, that is returned even if it is less than the trigger level.
2. If there is no data in the buffer, the calling task enters the blocked state and remains blocked until xTriggerLevelBytes are received or the task's block time expires.

So, it appears you are observing the intended behavior. Regarding your question of processing 10 samples at time, can you create an object which contains 10 samples and then use a queue to send it to the other task once all 10 samples are collected. Something like the following -

typedef struct Samples
{
    uint32_t values[ 10 ];
    uint32_t num_values;
} Samples_t;

#define QUEUE_LENGTH 5
xQueue = xQueueCreate( QUEUE_LENGTH, sizeof( Samples_t ) );

void SampleCollectorTask( void * params )
{
    Samples_t samples;
    memset( &( samples ), 0, sizeof( Samples_t ) );

    for( ;; )
    {
        /* Collect a sample. */
        uint32_t sample = Collect_Sample();

        /* Store the collected sample in the local object. */
        samples.values[ num_values ] = sample;
        samples.num_values++;

        /* Once all the samples are collected, send it to the queue. */
        if( samples.num_values == 10 )
        {
            xQueueSend( xQueue, &( samples ), pdMS_TO_TICKS( 100 ) );

            /* Clear the object for next set of samples. */
            memset( &( samples ), 0, sizeof( Samples_t ) );
        }
    }
}

[cperkulator]

That's effectively what we have now. In our fast sampling task, we locally buffer up samples and when that buffer is full we stick that whole buffer in a message buffer. That message buffer has a trigger level of 1 and unblocks our slower task for processing the data.

The issue with that approach, is that with the amount of data we are collecting the memcpy from the local buffer to the message buffer takes too much time and it happens in our fast task.

We were hoping that the stream buffer would allow us to write values one at a time in the fast task and unblock our slow task when all of the data is ready. We could take the approach of using the stream buffer and unblocking multiple times until we receive all of our data but again we were hoping to only unblock once since the timing is fairly tight.

Our use case seems like a pretty standard scenario so maybe it would be worth adding my PR but with a config around it?

Thanks!

[RichardBarry]

Could you instead send a pointer to the buffer, rather than copy the data from the buffer into the stream buffer? That would require a bit of management code to know which buffers are in use and which not.

We were hoping that the stream buffer would allow us to write values one at a time in the fast task and unblock our slow task when all of the data is ready.

I understand your frustration that the stream buffer implementation isn't right for your application, but what is right for your application won't be universally right for all applications. Just like the current implementation, written for common use cases, isn't right for your application. That is true to a greater or lesser extent for most design decisions.

[cperkulator]

We'll investigate other ways to manage it. Thanks for the guidance on this!

Would a PR be welcome for adding a mechanism to do what I'm talking about, that is separate from stream buffers? i.e. a mix between a stream buffer and a message buffer such that you can stream stuff in but always read a certain block size out? I feel like that would be really hand for scenarios in which you only want to process data after a certain amount of it is ready as in our case.

[RichardBarry]

If this can be done neatly, then yes, definitely.

[aggarg]

Thank you for the great discussion. For now, I am closing this issue and the corresponding PR. Let me know if you need anything else.