Question with queue and tasks

Kernel
1

This is my first attempt at ESP IDF and messing with FREERTOS.
I am using an ESP32 DEVKIT, PlatformIO, and ESP IDF.
I am using ESP IDF build 4.4.0
I created 2 tasks, and one queue to share. The issue is that if the xQueueSend has a delay that is longer that the xQueueReceive, the queue loses data. Am i doing something wrong?
My understanding of the tasks are that they timeshare since the priority is set the same.
Even though the receive task cannot complete in its time slot (10ms), i thought that through context switching, the receive task would resume where it left off. In this case even seems as when it restore and reads from the queue it is missing one of the items in the queue, as if the queue got reinitialized

extern "C"
{
#include "freertos/FreeRTOS.h"
#include "freertos/queue.h"
#include "freertos/task.h"
#include "esp_log.h"
#include "string.h"
}
#include <string>

static const char *TAG = "main";

struct message
{
    std::string rx_buffer;
};

struct commander_parameters
{
    QueueHandle_t xUdpQueue;
};

struct udp_server_parameters
{
    QueueHandle_t xUdpQueue;
};

void udp_server_task(void *pvParameters);
void commander_task(void *pvParameters);

extern "C" void app_main()
{
    BaseType_t xReturned;

    QueueHandle_t xUdpQueue = xQueueCreate(20, sizeof(message));

    udp_server_parameters udp_parm = {
        .xUdpQueue = xUdpQueue,
    };

    commander_parameters commander_parm = {
        .xUdpQueue = xUdpQueue,
    };

    xReturned = xTaskCreate(commander_task, "commander_task", 1024 * 16, &commander_parm, 11, NULL);
    if (xReturned != pdPASS)
    {
        ESP_LOGE(TAG, "Failed to create Commander Task");
    }

    xReturned = xTaskCreate(udp_server_task, "udp_task", 1024 * 16, &udp_parm, 11, NULL);
    if (xReturned != pdPASS)
    {
        ESP_LOGE(TAG, "Failed to create UDP Server Task");
    }

    vTaskStartScheduler();
}

void udp_server_task(void *pvParameters)
{
    // ESP_LOGI(TAG, "UDP Server Task starting up...");
    QueueHandle_t xUdpQueue;
    char rx_buffer[128];
    message udp_message;
    int ret;

    udp_server_parameters *pTaskParameters = (udp_server_parameters *)pvParameters;
    if (NULL == pTaskParameters ||
        NULL == pTaskParameters->xUdpQueue)
    {
        // ESP_LOGE(TAG, "UDP Task parameters were missing, exiting.");
        vTaskDelete(NULL); // Delete self.
    }
    // ESP_LOGE(TAG, "UDP Task parameters were OK");

    xUdpQueue = pTaskParameters->xUdpQueue;

    // ESP_LOGE(TAG, "UDP OK");

    while (true)
    {
        for (int x = 1; x < 10; x++)
        {
            sprintf(rx_buffer, "%i", x);
            udp_message.rx_buffer = rx_buffer;
            printf("Write queue:%s\n", udp_message.rx_buffer.c_str());
            ret = xQueueSend(xUdpQueue, &udp_message, pdMS_TO_TICKS(0));
            vTaskDelay(pdMS_TO_TICKS(500));
        }
    }
}

void commander_task(void *pvParameters)
{
    // ESP_LOGD(TAG, "Commander Task starting...");
    message commander_message;
    QueueHandle_t xUdpQueue;

    commander_parameters *pTaskParameters = (commander_parameters *)pvParameters;
    if (NULL == pTaskParameters ||
        NULL == pTaskParameters->xUdpQueue)
    {
        ESP_LOGE(TAG, "Task parameters were missing, exiting.");
        vTaskDelete(NULL); // Delete self.
    }

    // ESP_LOGE(TAG, "Task parameters were OK.");

    xUdpQueue = pTaskParameters->xUdpQueue;

    // ESP_LOGE(TAG, "Got parameters");

    while (true)
    {
        if (xQueueReceive(xUdpQueue, &commander_message, portMAX_DELAY))
        {
            printf("Recv Buffer:%s\n", commander_message.rx_buffer.c_str());
            vTaskDelay(pdMS_TO_TICKS(1000));
        }
    }
}

OUTPUT:

Write queue:4
Write queue:5
Recv Buffer:5
Write queue:6
Write queue:7
Recv Buffer:7
Write queue:8
Write queue:9
Recv Buffer:9
Write queue:1
Write queue:2
Recv Buffer:2
Write queue:3
Write queue:4
Recv Buffer:4
Write queue:5
Write queue:6
Recv Buffer:6
Write queue:7
Write queue:8
Recv Buffer:8
Write queue:9
Write queue:1
Recv Buffer:1

2

Since your send has a delay of 0, if the queue is full it will discard the data, and that is what you are seeing.

If you check the return value from the send, you will see it marking the failures. Normally, the xQueueSend will have a delay specified in it, to make the sender wait for space in the queue. Ultimately, the sender can only get as far ahead of the receive as you have room in the queue, then the sender needs to wait, or you need to discard some data.

1 Like
3

Thank you for the reply. However to me that does not appear the case. I understand that eventually if the queue isnt processed as fast as it fills until full, i indeed would lose items.
However, this is happening at first run. I attached a better example showing this behavior directly after startup. Notice at the beginning that the receive was able to keep up.

(541) spi_flash: trying chip: generice[0m
e[0;32mI (545) spi_flash: detected chip: generice[0m
e[0;32mI (549) spi_flash: flash io: dioe[0m
D (553) cpu_start: calling init function: 0x400d7368e[0m
D (558) cpu_start: calling init function: 0x400d55d0e[0m
D (563) cpu_start: calling init function: 0x400d51b0e[0m
D (568) cpu_start: calling init function: 0x400d1740e[0m
D (573) cpu_start: calling init function: 0x400d0d80e[0m
D (579) intr_alloc: Connected src 17 to int 3 (cpu 0)e[0m
D (584) intr_alloc: Connected src 24 to int 9 (cpu 0)e[0m
e[0;32mI (589) cpu_start: Starting scheduler on PRO CPU.e[0m
D (0) intr_alloc: Connected src 25 to int 2 (cpu 1)e[0m
e[0;32mI (0) cpu_start: Starting scheduler on APP CPU.e[0m
D (614) heap_init: New heap initialised at 0x3ffe0440e[0m
D (614) heap_init: New heap initialised at 0x3ffe4350e[0m
D (624) intr_alloc: Connected src 16 to int 12 (cpu 0)e[0m
mWrite queue:1
Recv Buffer:1
Write queue:2
Recv Buffer:2
Write queue:3
Write queue:4
Recv Buffer:4
Write queue:5
Write queue:6
Recv Buffer:6
Write queue:7
Write queue:8
Recv Buffer:8
Write queue:9
Write queue:1
Recv Buffer:1
Write queue:2
Write queue:3
Recv Buffer:3
Write queue:4
Write queue:5
Recv Buffer:5
Write queue:6
Write queue:7
Recv Buffer:7
Write queue:8
Write queue:9
Recv Buffer:9
Write queue:1
Write queue:2
Recv Buffer:2
Write queue:3
Write queue:4
Recv Buffer:4
Write queue:5
Write queue:6

4

Looking closer, your “message” isn’t a valid object for a queue. Queues use memcpy to copy the object to the queue, which isn’t valid for a std::string object. You can only use object that are “Plain Old Data” with Queues.

This is the limitation of FreeRTOS being a C library, not a C++ one built on Templates.

5

Richard, thank you for replying. I changed the struct to an int and i have the same issue.

6

your code queues data twice as fast as unqueuing, so you will necessarily lose data. A queue is a shock absorber, meaning that if on the average, you consume data at least as fast as you process it, your queue helps you absorbing the peaks. If you always consume slower than producing, a queue will not help.

7

Thanks for responding, but if you look at the new log i posted, you can see this behavior is happening before the queue fills. The queue length is set at 20.

8

In case your MCU is a Cortex-M3/4/7 the main stack is reset and reused as ISR stack when starting the scheduler as documented.
That means task parameters allocated in (app_)main on stack might get corrupted.
You should make the task params static or global or allocated them on heap.

9

Richard of course was correct, i shouldn’t have doubted.
Changing the struct from a c++ “string” to a char array has indeed fixed the issue.
I tried with a char* first, with the same results as the “String”.

struct message
{
int counter;
char data[128];
};

10

char* has the issue that you copy the pointer, but not the characters its points to, so each Queue send needs to use a different buffer to store the string.

1 Like