Vấn đề khi gửi dữ liệu vào Queue: Dữ liệu đã nhận nhưng sao chưa thực thi?

Giới thiệu

Ở bài viết trước, mình đã nêu cho các bạn biết khái niệm Queue là gì, cách sử dụng nó ra sao và những lưu ý khi sử dụng nó.

Nhắc lại một chút cho các bạn chưa đọc bài viết đó thì trong các hệ thống nhúng thời gian thực, FreeRTOS là một lựa chọn phổ biến nhờ tính linh hoạt và nhẹ. Một trong những cơ chế giao tiếp liên tiến trình (IPC) được sử dụng nhiều nhất là hàng đợi (queue).

Có một vấn đề mà mình cũng đã nhắc tới trong bài trước đó là: khi hệ thống phải xử lý lượng lớn dữ liệu hoặc dữ liệu đến liên tục từ các thiết bị ngoại vi (như ADC, UART, SPI, hoặc cảm biến), người dùng thường gặp phải tình huống queue bị đầy hoặc dữ liệu bị xử lý chậm, gây mất dữ liệu hoặc phản hồi trễ.

Bài viết này sẽ phân tích nguyên nhân gây ra hiện tượng trên và đưa ra các giải pháp cụ thể nhằm cải thiện hiệu suất xử lý dữ liệu qua queue trong FreeRTOS.

1. Tổng quan về Queue trong FreeRTOS

Trong FreeRTOS, queue là một hàng đợi FIFO (First In First Out) dùng để truyền dữ liệu giữa các task hoặc giữa ISR (Interrupt Service Routine) và task.

Cách hoạt động:

• ISR hoặc Task gửi dữ liệu vào queue bằng các hàm như xQueueSendFromISR() hoặc xQueueSend().
• Một task khác sẽ nhận dữ liệu từ queue bằng xQueueReceive().

Vấn đề:

• Nếu task nhận dữ liệu xử lý quá chậm, queue sẽ bị đầy.
• ISR gửi dữ liệu sẽ không gửi được tiếp, hoặc mất dữ liệu nếu dùng timeout ngắn.

2. Tình huống ví dụ

Xét hệ thống đơn giản:

• Một cảm biến gửi dữ liệu ADC mỗi 1ms vào ISR.
• ISR gọi xQueueSendFromISR() để gửi dữ liệu vào hàng đợi.
• Một task DataProcessingTask nhận và xử lý dữ liệu từ queue (ví dụ: lọc tín hiệu, ghi log, gửi MQTT,...).

Hiện tượng:

• Sau vài giây chạy ổn định, hệ thống bắt đầu mất dữ liệu hoặc trễ phản hồi do queue bị đầy.

3. Nguyên nhân gây chậm trễ xử lý dữ liệu

4. Giải pháp

4.1 Tối ưu tốc độ xử lý của task

• Phân tách xử lý nặng: Không nên thực hiện xử lý nặng (như tính toán số học phức tạp, giao tiếp chậm) trực tiếp trong task nhận queue.
• Dùng buffer trung gian: Đọc nhanh từ queue, đẩy vào một buffer nội bộ, sau đó xử lý buffer khi rảnh.

while (1) { if (xQueueReceive(queue, &data, portMAX_DELAY)) { buffer[write_index++] = data; // xử lý sau }
}

4.2 Tăng độ ưu tiên của task xử lý queue

• Đảm bảo task xử lý queue có độ ưu tiên cao hơn ISR hoặc task sinh dữ liệu nếu cần xử lý nhanh chóng.

xTaskCreate(DataProcessingTask, "Proc", 256, NULL, tskIDLE_PRIORITY + 3, NULL);

4.3 Tăng kích thước queue

• Nếu dung lượng RAM cho phép, tăng kích thước queue sẽ giúp “đệm” được nhiều dữ liệu hơn khi task chưa xử lý kịp.

xQueueCreate(64, sizeof(sensor_data_t));

4.4 Xử lý batch nhiều phần tử mỗi lần

• Đọc hết các phần tử trong queue (nếu có) thay vì đọc từng phần tử một.

while (uxQueueMessagesWaiting(queue) > 0) { xQueueReceive(queue, &data, 0); // xử lý data
}

4.5 ISR yield đúng cách

• Trong ISR, nếu có task đang chờ queue, cần gọi portYIELD_FROM_ISR() để chuyển ngay sang task xử lý.

BaseType_t xHigherPriorityTaskWoken = pdFALSE;
xQueueSendFromISR(queue, &data, &xHigherPriorityTaskWoken);
portYIELD_FROM_ISR(xHigherPriorityTaskWoken);

4.6 Giới hạn tốc độ ISR hoặc task gửi dữ liệu

• Nếu dữ liệu đến quá nhanh, cần "giảm tốc độ" gửi (debounce, timer pacing, downsampling,...).

// Ví dụ: gửi dữ liệu 1 lần mỗi 10 chu kỳ
static int count = 0;
if (++count >= 10) { count = 0; xQueueSendFromISR(queue, &data, &xHigherPriorityTaskWoken);
}

4.7 Sử dụng Stream Buffer hoặc Message Buffer thay vì Queue (khi phù hợp)

• StreamBuffer: Dùng cho chuỗi byte (ví dụ UART), không cần cấu trúc cố định.