Cześć, ostatnio dowiedzieliśmy się w jaki sposób zarządzać dostępem do zasobu za pomocą semaforów. W dzisiejszym artykule pokażę jeszcze inny sposób na dostarczenie zawartości do zasobu przez kilka różnych wątków bez problemów z synchronizacją. Zapraszam do krótkiej lektury o kolejkach.

Koncepcja

Posłużymy się jedną z podstawowych struktur jaką jest kolejka. Kolejka w programowaniu działa dokładnie tak samo jak kolejka np. w sklepie. Dane, które pierwsze zostaną umieszczone w kolejce zostaną też jako pierwsze obsłużone. Kolejka jest przeciwieństwem stosu, na który wkładamy elementy, a następnie zdejmujemy je z niego w odwrotnej kolejności (ostatni element na stosie zostaje przetworzony jako pierwszy).

Aby taka kolejka mogła działac musimy trochę inaczej zaprojektować nasze taski. W wariancie z mutexami oba wątki (DigitalRead oraz AnalogRead) pisały bezpośrednio na port szeregowy. W przypadku kolejki rozwiążemy sprawę nieco inaczej. Dodamy kolejny wątek (SerialWrite), który będzie konsumentem treści z kolejki, natomiast dwa dotychczasowe wątki (DigitalRead oraz AnalogRead) będą tzw. producentami. Chodzi o to, żeby wątek konsumenta nieustannie monitorował, czy w kolejce są jakieś dane do skonsumowania, natoamist wątki producentów uzupełniały koleję nowymi danymi.

Na powyższej animacji przestawiono zasadę działania kolejki. Źródło: https://www.freertos.org/Embedded-RTOS-Queues.html

Dokumentacja

Dokładniej o kolejkach w systemie freeRTOS możemy poczytać tutaj. Dowiemy się z tej storny szczegółów działania takiej kolejki. Warto zauważyć, że nie będziemy w kolejce przesyłali całych stringów, a jedynie konkretne dane np. liczby odczytane z wejść cyfrowych i analogowych lub proste struktury.

W celu wykorzystania kolejek w projekcie musimy załączyć plik nagłówkowy queue.h

Aby utworzyć koleję do globalnej zmiennej typu QueueHandle_t przypisujemy rezultat funkcji xQueueCreate. Funkcja ta przyjmuje dwa parametry: rozmiar kolejki (w elementach), oraz rozmiar pojedynczego elementu (przyda się tutaj funkcja sizeof).

Dodawanie odbywa się poprzez funkcję o nazwie xQueueSend. Ta z kolei przyjmuje trzy parametry. Pierwszym z nich jest uchwyt do kolejki, czyli po prostu zmienna, w której zaalokowaliśmy naszą koleję wywołując funkcję xQueueCreate. Drugim parametrem jest wskaźnik do zmiennej z danymi, które mają zostać dodane do kolejki, natomiast trzeci parametr opisuje ilość ticków, na które wywołanie funkcji zablokuje wątek w oczekiwaniu na miejsce w kolejce (w przypadku, jeżeli kolejka jest pełna).

Zdejmowanie elementó odbywa się analogicznie. W tym przypadku użyjemy funkcji xQueueReceive. Jej peirwszy i ostatni parametr są takie same jak w przypadku xQueueSend. Środkowy parametr jest natomiast adresem w pamięci, do którego mają być zapisane dane zdjęte z kolejki.

Implementacja

Pierwszym krokeim jest załączenie plików nagłówkowych:

#include <Arduino_FreeRTOS.h>
#include <queue.h>

Teraz możemy zadeklarować zmienną dla kolejki oraz funkcje dla wątków:

QueueHandle_t serialQueue;

void TaskDigitalRead( void *pvParameters );
void TaskAnalogRead( void *pvParameters );
void TaskSerialWrite( void *pvParameters );

W funkcji setup uruchamiamy port szeregowy, inicjujemy kolejkę oraz uruchamiamy wątki:

void setup() {
  Serial.begin(9600);
  while (!Serial) {
    ; // wait for serial port to connect. Needed for native USB, on LEONARDO, MICRO, YUN, and other 32u4 based boards.
  }
  serialQueue = xQueueCreate(16, sizeof(int));
  if(serialQueue == NULL){
    Serial.println("Failed to create serialQueue");
    while(true);
  }

  xTaskCreate(
    TaskDigitalRead
    ,  (const portCHAR *)"DigitalRead"  // A name just for humans
    ,  128  // This stack size can be checked & adjusted by reading the Stack Highwater
    ,  NULL
    ,  2  // Priority, with 3 (configMAX_PRIORITIES - 1) being the highest, and 0 being the lowest.
    ,  NULL );

  xTaskCreate(
    TaskAnalogRead
    ,  (const portCHAR *) "AnalogRead"
    ,  128  // Stack size
    ,  NULL
    ,  1  // Priority
    ,  NULL );

    xTaskCreate(
    TaskSerialWrite
    ,  (const portCHAR *) "SerialWrite"
    ,  128  // Stack size
    ,  NULL
    ,  1  // Priority
    ,  NULL );
}

Funkcja loop jak zwykle w przypadku użycia freeRTOS’a pozostaje pusta.

void loop()
{
}

Przyjrzyjmy się teraz wątkom producentów:

void TaskDigitalRead( void *pvParameters __attribute__((unused)) )
{
  uint8_t pushButton = 2;
  pinMode(pushButton, INPUT);
  for (;;)
  {
    int buttonState = digitalRead(pushButton);
    xQueueSend(serialQueue, &buttonState, portMAX_DELAY);
    vTaskDelay(1);
  }
}

void TaskAnalogRead( void *pvParameters __attribute__((unused)) )
{
  for (;;)
  {
    uint8_t analogIn = A0;
    int sensorValue = analogRead(analogIn);
    xQueueSend(serialQueue, &sensorValue, portMAX_DELAY);
    vTaskDelay(1);
  }
}

Jak widać zapisują one dane do zmiennej typu int, a następnie jej adres przekazywany jest do funkcji xQueueSend. W ten sposób kolejka pobiera spod danego adresu odpowiednią ilość bajtów.

Wątek konsumenta działa analogicznie, tylko w drugą stornę:

void TaskSerialWrite( void *pvParameters __attribute__((unused)) )
{
  for (;;)
  {
    int currentItem;
    while(xQueueReceive(serialQueue, &currentItem, portMAX_DELAY) == pdTRUE) {
      Serial.println(currentItem);
    }
    vTaskDelay(1);
  }
}

Deklarujemy zmienną typu int o nazwie currentItem, a następnie dopóki w kolejce znajdują się jakieś dane do skonsumowania zapisujemy odpowiednią ilość bajtów z pamięci kolejki do owej zmiennej, po czym wypisujemy ją na port szeregowy.

Gotowe, możemy teraz skompilować program, wgrać go na Arduino i zoabczyć jak działa nasza implementacja kolejki :)