Queue, semaphore, mutex trong FreeRTOS: chia sẻ dữ liệu an toàn giữa các task trên ESP32

Vì sao biến dùng chung bị sai trên ESP32 hai nhân, queue chuyển dữ liệu, semaphore để ngắt báo cho task, mutex bảo vệ Serial, thí nghiệm hai nhân cùng tăng một biến.

Chia chương trình thành nhiều task giúp mỗi việc giữ đúng nhịp, nhưng ngay khi các task cần trao đổi dữ liệu thì xuất hiện lỗi khó tìm: số đếm bị thiếu, dòng in ra bị chen nhau, giá trị đọc được nửa cũ nửa mới. FreeRTOS có ba công cụ để giải quyết: queue (hàng đợi) để chuyển dữ liệu, semaphore để báo hiệu, mutex để bảo vệ tài nguyên dùng chung. Bài này giải thích từng công cụ, dùng khi nào, và có một thí nghiệm cho thấy hai nhân của ESP32 làm mất dữ liệu thật khi không có khóa.

1. Vì sao biến dùng chung bị sai

Câu lệnh dem++ trông như một bước, nhưng CPU làm ba bước: đọc giá trị từ RAM vào thanh ghi, cộng 1, ghi trở lại. Trên ESP32 có hai nhân chạy thật sự cùng lúc, nên nhân kia có thể đọc đúng lúc nhân này đã đọc mà chưa ghi. Kết quả: hai lần cộng, biến chỉ tăng 1. Ngay cả trên một nhân, bộ lập lịch cũng có thể chuyển task giữa ba bước đó.

Từ khóa volatile chỉ bắt trình dịch đọc biến từ RAM mỗi lần, không làm phép tăng thành một bước. Muốn an toàn phải có cơ chế khóa.

2. Queue: chuyển dữ liệu giữa các task

Queue là hàng đợi có số ô cố định, vào trước ra trước. xQueueCreate(20, sizeof(MauDo)) tạo 20 ô, mỗi ô vừa một MauDo. Điểm quan trọng: queue chép nguyên dữ liệu vào ô của nó, nên bên gửi dùng lại biến của mình ngay được và bên nhận luôn có một bản riêng, không ai đọc dở dang của ai.

• xQueueSend(q, &m, thoiGianCho): thêm vào cuối hàng. Hàng đầy thì chờ tối đa thoiGianCho tick; chờ 0 thì trả lỗi ngay. Task đọc trong bài chọn không chờ, để nhịp lấy mẫu không bị kéo dài, và đếm số lần phải bỏ mẫu.
• xQueueReceive(q, &m, portMAX_DELAY): lấy ở đầu hàng. Hàng rỗng thì task ngủ (không tốn CPU) cho tới khi có phần tử mới.
• uxQueueSpacesAvailable(q) cho biết còn bao nhiêu ô trống: còn ít nghĩa là bên xử lý không theo kịp.

Lỗi hay gặp là gửi con trỏ tới biến cục bộ qua queue: khi bên nhận đọc, biến đó có thể đã bị ghi đè hoặc không còn tồn tại. Gửi cả struct nhỏ (vài chục byte) là cách an toàn nhất.

3. Semaphore và mutex

Semaphore nhị phân là một lá cờ: bên này dựng cờ (Give), bên kia chờ cờ và hạ xuống (Take). Cách dùng điển hình là ngắt báo cho task: hàm ngắt phải thật ngắn, chỉ dựng cờ bằng xSemaphoreGiveFromISR() rồi thoát; việc dài (chống dội, in Serial, gửi mạng) để task làm. Trong ngắt chỉ được gọi các hàm FreeRTOS có đuôi FromISR. portYIELD_FROM_ISR() cho task vừa được báo chạy ngay sau khi ngắt kết thúc nếu nó có ưu tiên cao hơn task đang chạy.

Semaphore đếm giữ một con số thay vì một cờ. Trong bài, setup() dùng nó để chờ đủ hai task của thí nghiệm báo xong.

Mutex là khóa cho tài nguyên dùng chung (biến, cổng Serial, bus I2C): task nào Take được thì làm việc, xong phải Give; task khác Take trong lúc đó sẽ phải chờ. Khác với semaphore nhị phân, mutex có người giữ: chỉ task đã lấy mới được trả, và FreeRTOS tạm nâng ưu tiên của task đang giữ khóa để task ưu tiên cao đang chờ không bị kẹt lâu (kế thừa ưu tiên). Không dùng mutex trong hàm ngắt.

4. Chương trình

Chỉ cần ESP32: nút BOOT (GPIO0) có sẵn trên bo. Nối thêm biến trở (hai đầu 3V3, GND, chân giữa GPIO34) nếu muốn số đo thay đổi theo tay xoay.

• Thí nghiệm lúc khởi động: hai task ở hai nhân cùng tăng một biến 100 000 lần, một lần không khóa, một lần có mutex.
• Task đọc (ưu tiên 2): mỗi 100 ms gửi thời điểm và điện áp GPIO34 vào queue.
• Task xử lý (ưu tiên 1): nhận từ queue, mỗi 10 mẫu in trung bình, nhỏ nhất, lớn nhất, số ô trống của queue.
• Task nút (ưu tiên 3): chờ cờ từ ngắt nút BOOT, chống dội rồi in số lần bấm.
• Hai task cùng in Serial qua hàm inDong() có mutex, nên các dòng không bao giờ bị chen giữa.

// PYWORLD.VN - Queue, semaphore và mutex của FreeRTOS trên ESP32: thí nghiệm hai nhân cùng tăng một biến, queue chuyển số đo, ngắt nút báo cho task
// Chỉ cần ESP32: nút BOOT (GPIO0) có sẵn trên bo. Muốn số đo thay đổi thì nối biến trở: hai đầu 3V3, GND, chân giữa GPIO34
// Không cần thư viện ngoài

#include <stdarg.h>

const int CHAN_NUT = 0;                  // nút BOOT trên bo, bấm thì xuống mức 0
const int CHAN_DO = 34;

struct MauDo {                           // một phần tử của queue: queue chép nguyên giá trị, không giữ con trỏ
  uint32_t thoiGianMs;
  int mv;
};

QueueHandle_t hangDoi;                   // task đọc → task xử lý
SemaphoreHandle_t coNutBam;              // semaphore nhị phân: ngắt nút → task nút
SemaphoreHandle_t khoaSerial;            // mutex: mỗi lúc chỉ một task in ra Serial
volatile uint32_t soLanDay = 0;          // số lần queue đầy, mẫu bị bỏ

// In một dòng có giữ mutex, để dòng của hai task không chen vào nhau.
void inDong(const char *dinhDang, ...) {
  char dong[160];
  va_list ds;
  va_start(ds, dinhDang);
  vsnprintf(dong, sizeof(dong), dinhDang, ds);
  va_end(ds);
  xSemaphoreTake(khoaSerial, portMAX_DELAY);
  Serial.println(dong);
  xSemaphoreGive(khoaSerial);
}

// ---- Thí nghiệm: hai task ở hai nhân cùng tăng một biến ----
const uint32_t SO_LAN = 100000;
volatile uint32_t demKhongKhoa = 0, demCoKhoa = 0;
SemaphoreHandle_t khoaDem, daXong;

void tangDem(void *thamSo) {
  for (uint32_t i = 0; i < SO_LAN; i++) demKhongKhoa = demKhongKhoa + 1;   // đọc, cộng, ghi: nhân kia có thể chen vào giữa
  for (uint32_t i = 0; i < SO_LAN; i++) {
    xSemaphoreTake(khoaDem, portMAX_DELAY);              // chỉ một task được ở trong đoạn này
    demCoKhoa = demCoKhoa + 1;
    xSemaphoreGive(khoaDem);
  }
  xSemaphoreGive(daXong);                                // báo đã xong (semaphore đếm)
  vTaskDelete(NULL);
}

void thiNghiemHaiNhan() {
  khoaDem = xSemaphoreCreateMutex();
  daXong = xSemaphoreCreateCounting(2, 0);
  uint32_t batDau = millis();
  xTaskCreatePinnedToCore(tangDem, "tang 0", 2048, NULL, 1, NULL, 0);
  xTaskCreatePinnedToCore(tangDem, "tang 1", 2048, NULL, 1, NULL, 1);
  xSemaphoreTake(daXong, portMAX_DELAY);                 // chờ đủ hai task
  xSemaphoreTake(daXong, portMAX_DELAY);
  Serial.printf("Khong khoa: %lu / %lu (mat %lu lan cong)\n", (unsigned long)demKhongKhoa, (unsigned long)(2 * SO_LAN),
                (unsigned long)(2 * SO_LAN - demKhongKhoa));
  Serial.printf("Co mutex:   %lu / %lu, mat %lu ms\n", (unsigned long)demCoKhoa, (unsigned long)(2 * SO_LAN),
                (unsigned long)(millis() - batDau));
}

// ---- Queue: task đọc gửi mẫu, task xử lý nhận ----
void taskDoc(void *thamSo) {
  TickType_t lanTruoc = xTaskGetTickCount();
  for (;;) {
    MauDo m = {millis(), (int)analogReadMilliVolts(CHAN_DO)};
    if (xQueueSend(hangDoi, &m, 0) != pdTRUE) soLanDay = soLanDay + 1;   // không chờ: queue đầy thì bỏ mẫu và đếm lại
    vTaskDelayUntil(&lanTruoc, pdMS_TO_TICKS(100));
  }
}

void taskXuLy(void *thamSo) {
  MauDo m;
  int dem = 0, nho = 0, lon = 0;
  long tong = 0;
  for (;;) {
    xQueueReceive(hangDoi, &m, portMAX_DELAY);           // ngủ cho tới khi có mẫu, không tốn CPU
    if (dem == 0) nho = lon = m.mv;
    nho = min(nho, m.mv);
    lon = max(lon, m.mv);
    tong += m.mv;
    if (++dem == 10) {                                   // 10 mẫu = 1 giây
      inDong("t=%5.1fs | trung binh %4ld mV, nho nhat %4d, lon nhat %4d | queue con %u cho, day %lu lan", m.thoiGianMs / 1000.0f,
             tong / 10, nho, lon, (unsigned)uxQueueSpacesAvailable(hangDoi), (unsigned long)soLanDay);
      dem = 0;
      tong = 0;
    }
  }
}

// ---- Semaphore nhị phân: ngắt chỉ "báo", việc xử lý để task làm ----
void IRAM_ATTR ngatNut() {
  BaseType_t canDoiTask = pdFALSE;
  xSemaphoreGiveFromISR(coNutBam, &canDoiTask);          // trong ngắt chỉ dùng hàm có đuôi FromISR
  portYIELD_FROM_ISR(canDoiTask);                        // task nút có ưu tiên cao hơn thì chạy ngay khi ngắt xong
}

void taskNut(void *thamSo) {
  int soLanBam = 0;
  for (;;) {
    xSemaphoreTake(coNutBam, portMAX_DELAY);
    vTaskDelay(pdMS_TO_TICKS(50));                       // chờ hết dội phím
    while (xSemaphoreTake(coNutBam, 0) == pdTRUE) {}     // bỏ các lần "báo" do dội
    if (digitalRead(CHAN_NUT) == LOW) inDong(">> Nut BOOT duoc bam, lan %d", ++soLanBam);
  }
}

void setup() {
  Serial.begin(115200);
  delay(500);
  thiNghiemHaiNhan();

  hangDoi = xQueueCreate(20, sizeof(MauDo));             // 20 phần tử, mỗi phần tử 8 byte
  coNutBam = xSemaphoreCreateBinary();
  khoaSerial = xSemaphoreCreateMutex();
  pinMode(CHAN_NUT, INPUT_PULLUP);
  attachInterrupt(CHAN_NUT, ngatNut, FALLING);
  xTaskCreatePinnedToCore(taskDoc, "doc", 2048, NULL, 2, NULL, 1);
  xTaskCreatePinnedToCore(taskXuLy, "xu ly", 4096, NULL, 1, NULL, 1);
  xTaskCreatePinnedToCore(taskNut, "nut", 4096, NULL, 3, NULL, 1);
}

void loop() {
  vTaskDelay(portMAX_DELAY);
}

5. Kết quả mong đợi

Kết quả không khóa mỗi lần chạy một khác và luôn thiếu: đó chính là các lần cộng bị mất. Có mutex thì luôn đủ 200 000, đổi lại chậm hơn nhiều vì mỗi lần cộng phải lấy và trả khóa, nhiều lần phải chờ nhân kia. Bài học: khóa đúng chỗ cần khóa và giữ khóa thật ngắn; dữ liệu chảy một chiều từ task này sang task kia thì dùng queue, không cần biến dùng chung.

Với biến đếm đơn giản dùng trong cả ngắt và task, ESP32 còn có khóa quay portMUX_TYPE với portENTER_CRITICAL / portEXIT_CRITICAL: nhanh hơn mutex nhưng tắt ngắt trong lúc giữ khóa, nên chỉ dùng cho vài lệnh.

6. Chọn công cụ nào

7. Xử lý lỗi thường gặp

8. Xem thêm

• Ngắt GPIO trên ESP32: các quy tắc viết hàm ngắt.
• I2C là gì: bus dùng chung cần mutex khi nhiều task cùng đọc cảm biến.
• Bài FreeRTOS trên ESP32 (task, độ ưu tiên, hai nhân) ở đầu loạt ESP32 nâng cao.