Kolejka wiadomości (FMQ)

Infrastruktura wywołania procedury zdalnej (RPC) HIDL korzysta z mechanizmów bindera, co oznacza, że wywołania generują obciążenie, wymagają operacji jądra i mogą wywoływać działanie harmonogramu. W przypadku jednak, gdy dane muszą być przesyłane między procesami z mniejszym obciążeniem i bez udziału jądra, używany jest system Fast Message Queue (FMQ).

FMQ tworzy kolejki wiadomości z wybranymi właściwościami. Za pomocą wywołania RPC HIDL możesz wysłać obiekt MQDescriptorSync lub MQDescriptorUnsync, a proces odbierający będzie go używać do uzyskiwania dostępu do kolejki wiadomości.

Typy kolejek

Android obsługuje 2 typy kolejek (zwane też wersjami):

  • Kolejki niezsynchronizowane mogą zostać przepełnione i mogą mieć wielu odczytów. Każdy z czytelników musi odczytać dane na czas, w przeciwnym razie je straci.
  • Synchronizowane kolejki nie mogą przepełniać i mogą mieć tylko jednego czytelnika.

Oba typy kolejek nie mogą mieć wartości poniżej zera (odczyt z pustej kolejki kończy się niepowodzeniem) i mogą mieć tylko 1 nagrywarkę.

Niezsynchronizowane kolejki

Niezsynchronizowana kolejka ma tylko 1 program zapisu, ale może mieć dowolną liczbę czytelników. W kolejce jest 1 miejsce do zapisu, ale każdy czytnik śledzi swoje własne miejsce odczytu.

Zapisy do kolejki zawsze się udają (nie są sprawdzane pod kątem przepełnienia), o ile nie są większe niż skonfigurowana pojemność kolejki (zapisy większe niż pojemność kolejki od razu się nie udają). Każdy czytelnik może mieć inne miejsce odczytu, więc nie trzeba czekać, aż każdy z nich zapozna się z każdym fragmentem danych – dane spadają z kolejki zawsze wtedy, gdy nowe zapisy potrzebują wolnego miejsca.

Czytelnicy są odpowiedzialni za pobranie danych, zanim wypadną z końca kolejki. Odczyt, który próbuje odczytać więcej danych niż jest dostępnych, albo natychmiast się nie powiedzie (jeśli nie jest blokujący) albo czeka na dostępność wystarczającej ilości danych (jeśli jest blokujący). Odczyt, który próbuje odczytać więcej danych niż pojemność kolejki, zawsze kończy się błędem.

Jeśli czytnik nie nadąża za zapisem, tak że ilość danych zapisanych i jeszcze nie odczytanych przez tego czytnika jest większa niż pojemność kolejki, następne odczytanie nie zwróci danych. Zamiast tego zeruje pozycję odczytu czytnika na ostatnią pozycję zapisu, a następnie zwraca błąd. Jeśli dane dostępne do odczytu są sprawdzane po przepełnieniu, ale przed kolejnym odczytem, pokazują więcej danych do odczytu niż pojemność kolejki, co oznacza, że wystąpiło przepełnienie. (Jeśli kolejka przepełni się między sprawdzeniem dostępnych danych a próbą odczytu tych danych, jedynym sygnałem przepełnienia jest niepowodzenie odczytu).

Synchronizowane kolejki

Synchronizowana kolejka ma 1 element zapisujący i 1 element odczytujący z 1 pozycją zapisu i 1 pozycją odczytu. Nie można zapisać więcej danych, niż pozwala na to kolejka, ani odczytać więcej danych, niż obecnie zawiera kolejka. W zależności od tego, czy wywoływana jest funkcja blokująca bądź nieblokująca zapisu lub odczytu, próba przekroczenia dostępnego miejsca lub danych albo od razu zwraca błąd, albo blokada do czasu ukończenia żądanej operacji. Próby odczytu lub zapisu większej ilości danych niż pojemność kolejki zawsze kończą się niepowodzeniem.

Konfigurowanie FMQ

Kolejka wiadomości wymaga kilku obiektów MessageQueue: jednego do zapisu i co najmniej jednego do odczytu. Nie ma wyraźnej konfiguracji, która określa, który obiekt ma być używany do zapisu lub odczytu. Użytkownik musi zadbać o to, aby żaden obiekt nie był używany do odczytu i zapisu, aby było maksymalnie 1 nagrywacz i aby w przypadku kolejek synchronizowanych było maksymalnie 1 czytnik.

Tworzenie pierwszego obiektu MessageQueue

Kolejka wiadomości jest tworzona i konfigurowana z pojedynczym wywołaniem:

#include <fmq/MessageQueue.h>
using android::hardware::kSynchronizedReadWrite;
using android::hardware::kUnsynchronizedWrite;
using android::hardware::MQDescriptorSync;
using android::hardware::MQDescriptorUnsync;
using android::hardware::MessageQueue;
....
// For a synchronized nonblocking FMQ
mFmqSynchronized =
  new (std::nothrow) MessageQueue<uint16_t, kSynchronizedReadWrite>
      (kNumElementsInQueue);
// For an unsynchronized FMQ that supports blocking
mFmqUnsynchronizedBlocking =
  new (std::nothrow) MessageQueue<uint16_t, kUnsynchronizedWrite>
      (kNumElementsInQueue, true /* enable blocking operations */);
  • Inicjator MessageQueue<T, flavor>(numElements) tworzy i inicjuje obiekt, który obsługuje funkcję kolejki wiadomości.
  • Konstruktor MessageQueue<T, flavor>(numElements, configureEventFlagWord) tworzy i inicjalizuje obiekt, który obsługuje kolejkę wiadomości z blokowaniem.
  • flavor może być równe kSynchronizedReadWrite w przypadku kolejki zsynchronizowanej lub kUnsynchronizedWrite w przypadku kolejki niezsynchronizowanej.
  • uint16_t (w tym przykładzie) może być dowolnym typem zdefiniowanym w HIDL, który nie obejmuje zatopionych buforów (nie ma typów string ani vec), uchwytów ani interfejsów.
  • kNumElementsInQueue wskazuje rozmiar kolejki pod względem liczby wpisów; określa rozmiar bufora pamięci współdzielonej przydzielonej do kolejki.

Utwórz drugi obiekt MessageQueue

Druga strona kolejki wiadomości jest tworzona za pomocą obiektu MQDescriptor uzyskanego od pierwszej strony. Obiekt MQDescriptor jest wysyłany za pomocą wywołania HIDL lub AIDL RPC do procesu, który obsługuje drugi koniec kolejki komunikatów. MQDescriptor zawiera informacje o kolejce, w tym:

  • Informacje do mapowania bufora i wskaźnika zapisu.
  • Informacje do mapowania wskaźnika odczytu (jeśli kolejka jest zsynchronizowana).
  • Informacje do mapowania słowa flagi zdarzenia (jeśli kolejka jest blokowana).
  • Typ obiektu (<T, flavor>), który obejmuje zdefiniowany przez HIDL typ elementów kolejki oraz typ kolejki (zsynchronizowany lub niezsynchronizowany).

Obiekt MQDescriptor możesz użyć do utworzenia obiektu MessageQueue:

MessageQueue<T, flavor>::MessageQueue(const MQDescriptor<T, flavor>& Desc, bool resetPointers)

Parametr resetPointers wskazuje, czy podczas tworzenia obiektu MessageQueue pozycje odczytu i zapisu mają być resetowane do 0. W niezsynchronizowanej kolejce pozycja odczytu (która jest lokalna dla każdego obiektu MessageQueue w niezsynchronizowanych kolejkach) jest zawsze ustawiana na 0 podczas tworzenia. Zwykle MQDescriptor jest inicjowany podczas tworzenia pierwszego obiektu kolejki wiadomości. Aby uzyskać większą kontrolę nad współdzieloną pamięcią, możesz ręcznie skonfigurować MQDescriptor (definiowany w system/libhidl/base/include/hidl/MQDescriptor.h), a potem utworzyć każdy obiekt MessageQueue zgodnie z opisem w tej sekcji.MQDescriptor

Blokowanie kolejek i flag zdarzeń

Domyślnie kolejki nie obsługują blokowania odczytów ani zapisów. Istnieją 2 rodzaje blokowania wywołań odczytu i zapisu:

  • Wersja krótka z 3 parametrami (wskaźnikiem danych, liczbą elementów, limitem czasu) obsługuje blokowanie poszczególnych operacji odczytu i zapisu w jednej kolejce. Gdy używasz tego formularza, kolejka obsługuje flagę zdarzenia i maski bitowe wewnętrznie, a obiekt kolejki pierwszej wiadomości musi zostać zainicjowany za pomocą drugiego parametru true. Na przykład: 
    // For an unsynchronized FMQ that supports blocking
mFmqUnsynchronizedBlocking =
  new (std::nothrow) MessageQueue<uint16_t, kUnsynchronizedWrite>
      (kNumElementsInQueue, true /* enable blocking operations */);
  • Długa wersja z 6 parametrami (w tym flagą zdarzenia i maskami bitowymi), obsługuje używanie wspólnego obiektu EventFlag między wieloma kolejkami i umożliwia określenie masek bitowych powiadomień. W takim przypadku do każdego wywołania odczytu i zapisu musisz podać flagę zdarzenia i maski bitowe.

W przypadku długiej formy możesz podać wartość EventFlag w każdym wywołaniu funkcji readBlocking() i writeBlocking(). Możesz zainicjować jedną z kolejek przy użyciu wewnętrznej flagi zdarzenia, która musi zostać wyodrębniona z obiektów MessageQueue tej kolejki za pomocą polecenia getEventFlagWord() i użyta do utworzenia obiektów EventFlag w każdym procesie do użycia z innymi FMQ. Możesz też zainicjować obiekty EventFlag za pomocą dowolnej odpowiedniej pamięci współdzielonej.

Ogólnie w każdej kolejce powinna być stosowana tylko jedna opcja: nieblokujące, krótkie lub długie. Mieszanie ich nie jest błędem, ale aby uzyskać pożądany efekt, konieczne jest ostrożne programowanie.

Oznaczanie pamięci jako tylko do odczytu

Domyślnie pamięć współdzielona ma uprawnienia do odczytu i zapisu. W przypadku niezsynchronizowanych kolejek (kUnsynchronizedWrite) autor może chcieć usunąć uprawnienia do zapisu dla wszystkich czytelników, zanim przekaże obiekty MQDescriptorUnsync. Dzięki temu inne procesy nie będą mogły zapisywać w kolejce, co zalecamy w celu ochrony przed błędami i niewłaściwym działaniem procesów czytnika. Jeśli autor chce, aby czytelnicy mogli zresetować kolejkę, gdy użyją MQDescriptorUnsync do utworzenia strony odczytu kolejki, pamięć nie może być oznaczona jako dostępna tylko do odczytu. Jest to domyślne działanie konstruktora MessageQueue. Jeśli więc istnieją obecni użytkownicy tej kolejki, musisz zmienić ich kod, aby utworzyć kolejkę za pomocą funkcji resetPointer=false.

  • Writer: wywołaj funkcję ashmem_set_prot_region z deskryptorem pliku MQDescriptor i regionem ustawionym na tylko do odczytu (PROT_READ): 
    int res = ashmem_set_prot_region(mqDesc->handle->data[0], PROT_READ)
  • Czytnik: utwórz kolejkę wiadomości z resetPointer=false (domyślnie true): 
    mFmq = new (std::nothrow) MessageQueue(mqDesc, false);

Używanie MessageQueue

Publiczny interfejs API obiektu MessageQueue to:

size_t availableToWrite() // Space available (number of elements).
size_t availableToRead() // Number of elements available.
size_t getQuantumSize() // Size of type T in bytes.
size_t getQuantumCount() // Number of items of type T that fit in the FMQ.
bool isValid() // Whether the FMQ is configured correctly.
const MQDescriptor<T, flavor>* getDesc() // Return info to send to other process.

bool write(const T* data)  // Write one T to FMQ; true if successful.
bool write(const T* data, size_t count) // Write count T's; no partial writes.

bool read(T* data); // read one T from FMQ; true if successful.
bool read(T* data, size_t count); // Read count T's; no partial reads.

bool writeBlocking(const T* data, size_t count, int64_t timeOutNanos = 0);
bool readBlocking(T* data, size_t count, int64_t timeOutNanos = 0);

// Allows multiple queues to share a single event flag word
std::atomic<uint32_t>* getEventFlagWord();

bool writeBlocking(const T* data, size_t count, uint32_t readNotification,
uint32_t writeNotification, int64_t timeOutNanos = 0,
android::hardware::EventFlag* evFlag = nullptr); // Blocking write operation for count Ts.

bool readBlocking(T* data, size_t count, uint32_t readNotification,
uint32_t writeNotification, int64_t timeOutNanos = 0,
android::hardware::EventFlag* evFlag = nullptr) // Blocking read operation for count Ts;

// APIs to allow zero copy read/write operations
bool beginWrite(size_t nMessages, MemTransaction* memTx) const;
bool commitWrite(size_t nMessages);
bool beginRead(size_t nMessages, MemTransaction* memTx) const;
bool commitRead(size_t nMessages);

Możesz użyć funkcji availableToWrite() i availableToRead(), aby określić, ile danych można przenieść w ramach jednej operacji. W niezsynchronizowanej kolejce:

  • Funkcja availableToWrite() zawsze zwraca pojemność kolejki.
  • Każdy czytelnik ma swoją pozycję czytania i sam wykonuje obliczenia dotyczące availableToRead().
  • Z punktu widzenia wolnego czytelnika kolejka może się przepełnić, co może spowodować, że funkcja availableToRead() zwróci wartość większą niż rozmiar kolejki. Pierwszy odczyt po przekroczeniu limitu kończy się niepowodzeniem i spowoduje to ustawienie pozycji odczytu dla tego czytnika na taką samą jak bieżący wskaźnik zapisu, niezależnie od tego, czy przepełnienie zostało zgłoszone przez availableToRead().

Metody read() i write() zwracają true, jeśli wszystkie żądane dane mogły zostać przesłane do kolejki i z niej. Te metody nie blokują; albo działają prawidłowo (i zwracają wartość true), albo od razu zwracają błąd (false).

Metody readBlocking() i writeBlocking() czekają, aż żądana operacja zostanie wykonana lub do momentu przekroczenia limitu czasu (wartość 0 w parametrze timeOutNanos oznacza, że nie ma limitu czasu).

Operacje blokowania są implementowane za pomocą słowa flagi zdarzenia. Domyślnie każda kolejka tworzy i używa własnego słowa flagi, aby obsługiwać krótkie formy readBlocking() i writeBlocking(). Wiele kolejek może używać tego samego słowa, dzięki czemu proces może czekać na zapis lub odczyt w dowolnej z tych kolejek. Wywołując funkcję getEventFlagWord(), można uzyskać wskaźnik do słowa flagi zdarzenia kolejki i użyć tego wskaźnika (lub dowolnego wskaźnika do odpowiedniej lokalizacji współdzielonej pamięci) do utworzenia obiektu EventFlag, który będzie przekazywany do długiej formy readBlocking() oraz writeBlocking() dla innej kolejki. Parametry readNotification i writeNotification określają, które bity w flagach zdarzenia mają być używane do sygnalizowania odczytów i zapisów w tej kolejce. readNotification i writeNotification to 32-bitowe maski bitowe.

readBlocking() czeka na bity writeNotification. Jeśli ten parametr ma wartość 0, wywołanie zawsze kończy się niepowodzeniem. Jeśli wartość readNotification wynosi 0, wywołanie nie kończy się niepowodzeniem, ale pomyślne odczyty nie dodają żadnych informacji o powiadomieniu. W kolejce synchronicznej oznacza to, że odpowiednia funkcja writeBlocking() nigdy się nie aktywuje, chyba że bit zostanie ustawiony gdzie indziej. W niesynchronizowanej kolejce writeBlocking() nie czeka (nadal powinien być używany do ustawienia bitu powiadomienia o zapisie), a w przypadku odczytu nie powinien ustawiać żadnych bitów powiadomienia. Podobnie writeblocking() zakończy się niepowodzeniem, jeśli readNotification ma wartość 0, a pomyślne zapisanie ustawia określone bity writeNotification.

Aby czekać na wiele kolejek jednocześnie, użyj metody EventFlag obiektu wait(), aby czekać na bitową maskę powiadomień. Metoda wait() zwraca słowo określające stan z bitami, które spowodowały wybudzenie. Następnie te informacje są używane do sprawdzania, czy odpowiednia kolejka ma wystarczająco dużo miejsca lub danych na potrzeby operacji zapisu i odczytu, a także do wykonywania nieblokujących operacji write() i read(). Aby uzyskać powiadomienie po operacji, użyj kolejnego wywołania metody EventFlag obiektu wake(). Definicję pojęcia EventFlag abstrahowania znajdziesz w artykule system/libfmq/include/fmq/EventFlag.h.

Operacje bez kopii

Metody read, write, readBlocking i writeBlocking() przyjmują jako argument wskaźnik do bufora wejścia-wyjścia i korzystają wewnętrznie z wywołań memcpy(), aby kopiować dane między tym samym buforem a buforem pierścieniowym FMQ. Aby zwiększyć wydajność, Android 8.0 i nowsze wersje zawierają zestaw interfejsów API, które zapewniają bezpośredni dostęp do wskaźnika do bufora pierścieniowego, eliminując konieczność korzystania z wywołań memcpy.

Do operacji FMQ bez kopiowania używaj tych publicznych interfejsów API:

bool beginWrite(size_t nMessages, MemTransaction* memTx) const;
bool commitWrite(size_t nMessages);

bool beginRead(size_t nMessages, MemTransaction* memTx) const;
bool commitRead(size_t nMessages);
  • Metoda beginWrite udostępnia podstawowe wskaźniki do pętli buforowej FMQ. Po zapisaniu danych zatwierdź je za pomocą commitWrite(). Metody beginRead i commitRead działają w ten sam sposób.
  • Metody beginRead i Write przyjmują jako dane wejściowe liczbę wiadomości do odczytu i zapisu oraz zwracają wartość logiczną wskazującą, czy odczyt lub zapis jest możliwy. Jeśli odczyt lub zapis jest możliwy, struktura memTx jest wypełniana wskaźnikami podstawowymi, których można używać do bezpośredniego dostępu do współdzielonej pamięci za pomocą wskaźnika w buforze pierścieniowym.
  • Struktura MemRegion zawiera szczegółowe informacje o bloku pamięci, w tym wskaźnik podstawowy (adres podstawowy bloku pamięci) i długość zgodnie z elementami T (długość bloku pamięci w kontekście zdefiniowanego przez HIDL typu kolejki komunikatów).
  • Struktura MemTransaction zawiera 2 struktury MemRegion, first i second, które w ramach operacji odczytu lub zapisu do pierścieniowego bufora mogą wymagać przewinięcia do początku kolejki. Oznacza to, że do odczytu i zapisu danych do pierścieniowego bufora FMQ potrzebne są 2 wskaźniki podstawowe.

Aby uzyskać adres bazowy i długość z struktury MemRegion:

T* getAddress(); // gets the base address
size_t getLength(); // gets the length of the memory region in terms of T
size_t getLengthInBytes(); // gets the length of the memory region in bytes

Aby uzyskać odwołania do pierwszego i drugiego elementu MemRegion w obiekcie MemTransaction:

const MemRegion& getFirstRegion(); // get a reference to the first MemRegion
const MemRegion& getSecondRegion(); // get a reference to the second MemRegion

Przykład zapisu do FMQ przy użyciu interfejsów zero copy:

MessageQueueSync::MemTransaction tx;
if (mQueue->beginRead(dataLen, &tx)) {
    auto first = tx.getFirstRegion();
    auto second = tx.getSecondRegion();

    foo(first.getAddress(), first.getLength()); // method that performs the data write
    foo(second.getAddress(), second.getLength()); // method that performs the data write

    if(commitWrite(dataLen) == false) {
       // report error
    }
} else {
   // report error
}

Te pomocnicze metody są też dostępne w obiekcie MemTransaction:

  • T* getSlot(size_t idx); zwraca wskaźnik do boksu idx w MemRegions, które są częścią tego obiektu MemTransaction. Jeśli obiekt MemTransaction reprezentuje regiony pamięci do odczytu i zapisu N elementów typu T, prawidłowy zakres idx mieści się w zakresie od 0 do N-1.
  • bool copyTo(const T* data, size_t startIdx, size_t nMessages = 1); zapisuje nMessages elementy typu T w regionach pamięci opisanych przez obiekt, zaczynając od indeksu startIdx. Ta metoda używa memcpy() i nie jest przeznaczona do operacji bez kopii. Jeśli obiekt MemTransaction reprezentuje pamięć do odczytu i zapisu N elementów typu T, prawidłowy zakres idx mieści się w zakresie od 0 do N–1.
  • bool copyFrom(T* data, size_t startIdx, size_t nMessages = 1); to metoda pomocnicza do odczytywania elementów nMessages typu T z regionów pamięci opisanych przez obiekt, począwszy od startIdx. Ta metoda używa memcpy() i nie jest przeznaczona do operacji bez kopii.

Wysyłanie kolejki przez HIDL

Po stronie tworzenia:

  1. Utwórz obiekt kolejki wiadomości w sposób opisany powyżej.
  2. Sprawdź, czy obiekt jest prawidłowy, za pomocą funkcji isValid().
  3. Jeśli oczekujesz w kilku kolejkach, przekazując EventFlag do długiej formy readBlocking() lub writeBlocking(), możesz wyodrębnić wskaźnik flagi zdarzenia (za pomocą getEventFlagWord()) z obiektu MessageQueue, który został zainicjowany w celu utworzenia flagi, i użyć tej flagi do utworzenia niezbędnego obiektu EventFlag.
  4. Aby uzyskać obiekt opisu, użyj metody MessageQueue getDesc().
  5. W pliku HAL nadaj metodzie parametr typu fmq_sync lub fmq_unsync, gdzie T to odpowiedni typ zdefiniowany w HIDL. Użyj tej opcji, aby wysłać obiekt zwrócony przez getDesc() do procesu odbierania.

Po stronie odbiorcy:

  1. Użyj obiektu opisu, aby utworzyć obiekt MessageQueue. Użyj tego samego typu kolejki i typu danych, w przeciwnym razie nie uda się skompilować szablonu.
  2. Jeśli wyodrębniasz flagę zdarzenia, wyodrębnij ją z odpowiedniego obiektu MessageQueue w procesie odbierania.
  3. Aby przenieść dane, użyj obiektu MessageQueue.