תור הודעות מהיר (FMQ)

למי שזקוק לתמיכה ב-AIDL, כדאי להיכנס אל FMQ עם AIDL.

תשתית הקריאה להליך מרוחק (RPC) של HIDL משתמשת במנגנוני Binder, כלומר קריאות כוללות תקורה, דורשות פעולות ליבה (kernel) ועשויות להפעיל פעולת התזמון. עם זאת, במקרים שבהם צריך להעביר נתונים בין תהליכים עם פחות תקורה וללא מעורבות ליבה (kernel) – Fast Message Queue נעשה שימוש במערכת (FMQ).

FMQ יוצר תורי הודעות עם המאפיינים הרצויים. אובייקט MQDescriptorSync או MQDescriptorUnsync יכול להיות נשלחה בקריאת HIDL RPC ומשמשת את תהליך המקבל כדי לגשת אל תור ההודעות.

התכונה 'תורים להודעות מהירות' נתמכת רק ב-C++ ובמכשירים עם Android מגרסה 8.0 ואילך.

סוגי תורים של הודעות

ב-Android יש תמיכה בשני סוגי תורים (שנקראים טעמים):

  • תורים לא מסונכרנים יכולים לגלוש, והם יכולים לכלול הרבה קוראים; כל קורא חייב לקרוא את הנתונים בזמן או לאבד אותם.
  • תורים מסונכרנים לא ניתנים לגלישה, והם יכולים לכלול רק קורא אחד.

לא ניתן להשתמש בשני סוגי התורים (הקראה מתור ריק) נכשל) ויכול להיות לו רק כותב אחד.

לא מסונכרן

תור לא מסונכרן מכיל רק מחבר אחד, אבל יכול להכיל כל מספר של לקוראים. יש מיקום אחד לכתיבה לתור; עם זאת, כל קורא לעקוב אחרי המיקום הבלתי תלוי שלו לקריאה.

כתיבה לתור תמיד מצליחה (לא מסומנת אם יש חריגות) כל עוד הם לא גדולים מהקיבולת שהוגדרה בתור (כותבים גדולים מהערך הקיבולת של 'הבאים בתור' נכשלת באופן מיידי). מאחר שלכל קורא יכול להיות קריאה שונה במקום להמתין שכל קורא יקרא כל פיסת נתונים, יכול לצאת מהתור בכל פעם שכותבים חדשים צריכים מקום.

הקוראים אחראים לאחזור נתונים לפני שהם יוצאים בתור. קריאה שמנסה לקרוא יותר נתונים מאלה שזמינים הפעולה נכשלת באופן מיידי (אם היא לא מובילה לחסימה) או ממתינה שמספיק נתונים יהיו זמינים (אם ). קריאה שמנסה לקרוא יותר נתונים מהקיבולת תמיד של תור נכשל באופן מיידי.

אם קורא לא מצליח לעקוב אחר הכותב, כך שכמות הנתונים שקורא זה נכתב ועדיין לא נקרא על ידי הקורא, גדול יותר מקיבולת התור, הקריאה הבאה לא מחזירה נתונים במקום זאת, הוא מאפס את קריאת הנתונים כדי שווה למיקום הכתיבה האחרון, ואז מחזירה כשל. אם הנתונים הזמינים לקריאה נבדקים לאחר גלישה, אבל לפני הקריאה הבאה, מציג יותר נתונים זמינים לקריאה מאשר קיבולת התור, מה שמציין אירעה חריגה. (אם התור גולש בין בדיקת הנתונים הזמינים) ומנסה לקרוא את הנתונים האלה, האינדיקציה היחידה לגלישה היא הקריאה נכשלה.)

סביר להניח שקוראים של תור לא מסונכרן לא רוצים לאפס את מצביעי הקריאה והכתיבה של התור. לכן, כשיוצרים את התור קוראי מתאר צריכים להשתמש בארגומנט 'false' עבור 'resetPointers' הפרמטר.

מסונכרנת

תור מסונכרן כולל כותב אחד וקורא אחד עם כתיבה יחידה מקום אחד שבו יש קריאה אחת. אי אפשר לכתוב יותר נתונים מאשר יש בתור מקום לעוד נתונים, או לקרוא אותם, בהשוואה למה ששמור כרגע בתור. תלוי אם פונקציית הכתיבה או הקריאה החוסמת היא נקרא, מנסה לחרוג משטח האחסון הזמין או שהנתונים לא הוחזרו מיד או לחסום אותו עד שאפשר יהיה להשלים את הפעולה הרצויה. מתבצע ניסיון לקרוא או לכתוב יותר נתונים ממה שקיבולת התור תמיד נכשלת באופן מיידי.

הגדרת FMQ

תור הודעות מחייב מספר אובייקטים מסוג MessageQueue: אחד עד להיכתב אליהם, ולפעמים גם לקרוא מהם. אין תוכן בוטה תצורה של האובייקט שמשמש לכתיבה או לקריאה; היא תלויה כדי לוודא שאף אובייקט לא משמש גם לקריאה וגם לכתיבה, הוא יוצר אחד לכל היותר, ובתורים מסונכרנים יש לפחות כותב אחד בקורא.

יצירת אובייקט MessageQueue הראשון

תור הודעות נוצר ומוגדר באמצעות שיחה אחת:

#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 non-blocking 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 */);
  • המאתחל MessageQueue<T, flavor>(numElements) יוצר ומאתחל אובייקט שתומך בפונקציונליות של תור ההודעות.
  • המאתחל MessageQueue<T, flavor>(numElements, configureEventFlagWord) יוצר ומאתחל אובייקט שתומך בפונקציונליות של תור ההודעות באמצעות חסימה.
  • flavor יכול להיות kSynchronizedReadWrite עבור תור מסונכרן או kUnsynchronizedWrite עבור פריט לא מסונכרן לרשימת 'הבאים בתור'.
  • uint16_t (בדוגמה הזו) יכול להיות כל אחד סוג מוגדר HIDL לא כוללת מאגרי נתונים זמניים מקוננים (לא string או vec שונים), כינויים או ממשקים.
  • kNumElementsInQueue מציין את גודל התור במספר רשומות; הוא קובע את הגודל של מאגר האחסון המשותף שיוקצה בתור.

יצירת האובייקט MessageQueue השני

הצד השני של תור ההודעות נוצר באמצעות אובייקט MQDescriptor התקבל מהצד הראשון. אובייקט MQDescriptor נשלח באמצעות קריאה של HIDL או AIDL לתהליך RPC שבו נמצא הקצה השני של תור ההודעות. MQDescriptor מכיל מידע על התור, כולל:

  • מידע למיפוי של מאגר הנתונים הזמני ושל מצביע הכתיבה.
  • מידע למיפוי מצביע הקריאה (אם התור מסונכרן).
  • מידע למיפוי מילת הסימון של האירוע (אם התור חוסם).
  • סוג האובייקט (<T, flavor>), שכולל את המאפיין סוג מוגדר HIDL של רכיבים בתור והסגנון של התור (מסונכרנים או לא מסונכרנים).

אפשר להשתמש באובייקט MQDescriptor כדי ליצור אובייקט MessageQueue:

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

הפרמטר resetPointers מציין אם לאפס את הקריאה וכותבים את המיקומים עד 0 בזמן יצירת האובייקט MessageQueue. בתור לא מסונכרן, מיקום הקריאה (שהוא מקומי לכל אחד מהם) אובייקט MessageQueue בתורים לא מסונכרנים) תמיד מוגדר ל-0 במהלך היצירה. בדרך כלל, MQDescriptor מופעל במהלך יצירת האובייקט הראשון בתור ההודעות. כדי לקבל שליטה רבה יותר על זיכרון, אפשר להגדיר את MQDescriptor באופן ידני (הפרמטר MQDescriptor מוגדר כאן system/libhidl/base/include/hidl/MQDescriptor.h) ואז יוצרים כל אובייקט MessageQueue כפי שמתואר בקטע הזה.

חסימה של תורי הצגה ודגלי אירועים

כברירת מחדל, תורים לא תומכים בחסימת קריאה/כתיבה. יש שני סוגים של חסימת שיחות קריאה/כתיבה:

  • טופס קצר, עם שלושה פרמטרים (מצביע הנתונים, מספר הפריטים, ). תומכת בחסימה של פעולות קריאה/כתיבה בודדות לרשימת 'הבאים בתור'. כשמשתמשים בטופס הזה, התור יטפל בדגל האירוע ובמסכות הביטים של האירוע באופן פנימי, והאובייקט בתור ההודעה הראשונה צריך לאתחל עם פרמטר שני של true. מוצרים לדוגמה: 
    // For an unsynchronized FMQ that supports blocking
mFmqUnsynchronizedBlocking =
  new (std::nothrow) MessageQueue<uint16_t, kUnsynchronizedWrite>
      (kNumElementsInQueue, true /* enable blocking operations */);
  • תבנית ארוכה, עם שישה פרמטרים (כולל סימון אירוע ומסכות ביטים). תמיכה בשימוש באובייקט EventFlag משותף בין כמה תורים ומאפשר לציין את מסכות הביטים של ההתראות שבהן יש להשתמש. במקרה הזה, הפרמטר לכל קריאה וכתיבה יש לספק דגל אירוע ומסכות ביטים.

בפורמט הארוך, אפשר לספק את EventFlag באופן מפורש בכל שיחה של readBlocking() ו-writeBlocking(). אחד מ- אפשר לאתחל את התורים באמצעות דגל אירוע פנימי, שאחר כך צריך להיות נשלף מ-MessageQueue האובייקטים של התור באמצעות getEventFlagWord() ומשמש ליצירת EventFlag של האובייקטים בכל תהליך לשימוש עם מכשירי FMQ אחרים. לחלופין, אפשר לאתחל EventFlag אובייקטים עם כל אובייקט משותף מתאים זיכרון.

באופן כללי, בכל תור צריך להשתמש רק באחד וחסימה של סרטונים ארוכים. זו לא שגיאה לערבב ביניהם, אבל כדאי להיזהר צריך תכנות כדי לקבל את התוצאה הרצויה.

סימון הזיכרון לקריאה בלבד

כברירת מחדל, לזיכרון המשותף יש הרשאות לקריאה ולכתיבה. עבור לא מסונכרן תורים (kUnsynchronizedWrite), ייתכן שהכותב ירצה להסיר את הרשאות הכתיבה לכולם של הקוראים לפני שהוא מחלק את האובייקטים MQDescriptorUnsync. כך אפשר לוודא שהשני תהליכים לא יכולים לכתוב לתור, ומומלץ להגן מפני באגים או התנהגות לא תקינה שהקוראים מעבדים. אם הכותב רוצה שהקוראים יוכלו לאפס את התור בכל פעם שהם משתמשים MQDescriptorUnsync כדי ליצור את הצד הנקרא של התור, ואז אי אפשר לסמן את הזיכרון לקריאה בלבד. זוהי התנהגות ברירת המחדל של ה-constructor של 'MessageQueue'. לכן, אם יש כבר למשתמשים הקיימים בתור הזה, צריך לשנות את הקוד שלהם כדי ליצור את התור עם resetPointer=false

  • כותב: קריאה אל ashmem_set_prot_region באמצעות מתאר קובץ של MQDescriptor והאזור מוגדר לקריאה בלבד (PROT_READ): 
    int res = ashmem_set_prot_region(mqDesc->handle->data[0], PROT_READ)
  • Reader: יצירת תור הודעות באמצעות resetPointer=false ברירת המחדל היא true): 
    mFmq = new (std::nothrow) MessageQueue(mqDesc, false);

שימוש ב'תור ההודעות'

ה-API הציבורי של האובייקט MessageQueue הוא:

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);

ניתן להשתמש ב-availableToWrite() וב-availableToRead() כדי לקבוע כמה נתונים אפשר להעביר בפעולה אחת. תוך שימוש תור לא מסונכרן:

  • הפונקציה availableToWrite() תמיד מחזירה את הקיבולת של התור.
  • לכל קורא יש עמדת קריאה משלו ומתבצעת חישוב משלו availableToRead()
  • מבחינת קורא איטי, התור יכול לגלוש; כתוצאה מכך, יכול להיות ש-availableToRead() יחזיר ערך שגדול מ- את גודל התור. הקריאה הראשונה אחרי גלישה נכשלה והתוצאה את מיקום הקריאה בשביל שאותו קורא מוגדר השווה למצב הכתיבה הנוכחי, אם החריגה דווחה באמצעות availableToRead()

השיטות read() ו-write() מוחזרות true אם כל הנתונים המבוקשים יכולים להיות (והועברו) אל/מאת בתור. השיטות האלה לא חוסמות, הם מצליחים (וחוזרים true), או החזרה נכשלה (false) באופן מיידי.

יש להמתין בשיטות readBlocking() ו-writeBlocking() עד להשלמת הפעולה המבוקשת, או עד שהזמן הקצוב לתפוגה יסתיים ( המשמעות של הערך 0 timeOutNanos היא אף פעם לא זמן קצוב לתפוגה.

פעולות חסימה מיושמות באמצעות מילה של סימון אירוע. כברירת מחדל, כל תור יוצר ומשתמש במילת סימון משלו כדי לתמוך בצורה הקצרה readBlocking() וגם writeBlocking() ייתכן כי מספר תורים לשיתוף מילה אחת, כך שתהליך קוראת לאחד מהתורים. מצביע למילת הדגל של האירוע בתור יכול להיות שמתקבלת באמצעות קריאה אל getEventFlagWord(), והמצביע הזה (או כל אל המיקום המתאים בזיכרון) כדי ליצור אובייקט EventFlag שצריך להעביר לפורמט הארוך readBlocking() ו-writeBlocking() עבור לרשימת 'הבאים בתור'. readNotification וגם writeNotification הפרמטרים מציינים באילו ביטים בדגל האירוע יש להשתמש כדי לאותת קריאות כותב בתור הזה. readNotification והקבוצה writeNotification הן מסכות ביטים של 32 ביט.

readBlocking() מחכה על writeNotification הביטים. אם הערך של הפרמטר הוא 0, הקריאה תמיד תיכשל. אם הערך של readNotification הוא 0, השיחה לא תיכשל, אבל קריאה מוצלחת לא תגדיר קטעי התראות. בתור מסונכרן, המשמעות היא שהקריאה התואמת ל-writeBlocking() אף פעם לא מתעוררת אלא אם הביט מוגדר במקום אחר. בתור לא מסונכרן, writeBlocking() לא מחכה (עדיין יש להשתמש בו כדי להגדיר את כתוב ביט של התראה), והוא מתאים לכך שפעולות קריאה לא יגדירו של התראות. באופן דומה, writeblocking() ייכשל אם הערך readNotification הוא 0, וכתיבה מוצלחת מגדירה את הערך שצוין writeNotification ביטים.

כדי להמתין בכמה תורים בבת אחת, צריך להשתמש באובייקט EventFlag wait() ממתינה לקבלת מסכת התראות. השיטה wait() מחזירה מילת סטטוס עם הביטים שגרמו הגדרת ההתעוררות מידע זה ישמש לאחר מכן כדי לאמת שהתור התואם מספיק מקום או נתונים לפעולת הכתיבה/קריאה הרצויה ללא חסימה של write()/read(). כדי לבצע פעולת פרסום התראה, שימוש בשיחה אחרתEventFlag אמצעי תשלום wake(). להגדרה של EventFlag הפשטה, system/libfmq/include/fmq/EventFlag.h.

אפס פעולות העתקה

read/write/readBlocking/writeBlocking() ממשקי API לוקחים מצביע אל מאגר נתונים זמני של קלט/פלט כארגומנט ומשתמשים memcpy() שיחות פנימיות להעתקת נתונים בין מאגר לצלצול FMQ. כדי לשפר את הביצועים, מערכת Android מגרסה 8.0 ואילך כוללת קבוצה של ממשקי API שמספקים גישה ישירה של מצביע למאגר הנתונים הזמני, וכך מבטלים את הצורך צריכים להשתמש ב-memcpy קריאות.

צריך להשתמש בממשקי ה-API הציבוריים הבאים לפעולות FMQ ללא עותק:

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);
  • השיטה beginWrite מספקת מצביעי בסיס לטבעת ה-FMQ מאגר נתונים זמני. אחרי שהנתונים נכתבים, שומרים אותם באמצעות commitWrite(). השיטות beginRead/commitRead פועלות באותו אופן.
  • ה-methods beginRead מתוך Write כוללות את הקלט מספר ההודעות לקריאה/כתיבה והחזרת ערך בוליאני שמציין אם קריאה/כתיבה. אם ניתן לבצע קריאה או כתיבה, memTx מבנה האובייקט מאוכלס בסמנים בסיסיים שאפשר להשתמש בהם לזיהוי ישיר. את הגישה לזיכרון המשותף של מאגר הנתונים הזמני של הצלצול.
  • המבנה MemRegion מכיל פרטים על בלוק זיכרון, כולל מצביע הבסיס (הכתובת הבסיסית של בלוק הזיכרון) והאורך מונחים של T (האורך של בלוק הזיכרון במונחים של הגדרת HIDL) הסוג של תור ההודעות).
  • המבנה MemTransaction מכיל שני ערכי MemRegion את, first ו-second בתור קריאה או כתיבה יכול להיות שמאגר הנתונים הזמני של הצלצול יחייב עיגול לתחילת התור. הזה צריכים להיות שני מצביעים בסיסיים כדי לקרוא ולכתוב נתונים ב-FMQ למאגר הנתונים הזמני.

כדי לקבל את הכתובת הבסיסית ואת האורך ממבנה 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

כדי לקבל הפניות ל-MemRegion הראשון והשני בתוך אובייקט MemTransaction:

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

דוגמה לכתיבה ב-FMQ באמצעות ממשקי API של אפס העתקה:

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
}

גם שיטות העזרה הבאות הן חלק מ-MemTransaction:

  • T* getSlot(size_t idx);
    החזרת מצביע לחריץ idx בתוך MemRegions שייכים לMemTransaction הזה לאובייקט. אם האובייקט MemTransaction מייצג את הזיכרון אזורים לקריאה/כתיבה של N פריטים מסוג T, ואז הטווח החוקי של idx הוא בין 0 ל-N-1.
  • bool copyTo(const T* data, size_t startIdx, size_t nMessages = 1);
    כתיבת nMessages פריטים מסוג T באזורי הזיכרון שתואר על ידי האובייקט, החל מהאינדקס startIdx. השיטה הזו משתמש ב-memcpy() ולא מיועד לשימוש כעותק אפס פעולה. אם האובייקט MemTransaction מייצג זיכרון קריאה/כתיבה של N פריטים מסוג T, אז הטווח החוקי של idx הוא בין 0 ל-N-1.
  • bool copyFrom(T* data, size_t startIdx, size_t nMessages = 1);
    שיטת עזרה לקריאת nMessages פריטים מסוג T אזורי זיכרון שמתוארים על ידי האובייקט החל מ-startIdx. הזה השיטה משתמשת ב-memcpy() ולא מיועדת לעותק ללא עותק פעולה.

שליחת התור ב-HIDL

בצד היצירה:

  1. יוצרים אובייקט של תור הודעות כמו שמתואר למעלה.
  2. מוודאים שהאובייקט תקין באמצעות isValid().
  3. אם אתם ממתינים בתורים מרובים על ידי העברה של EventFlag בצורה הארוכה readBlocking()/writeBlocking(), אפשר לחלץ את מצביע סימון אירוע (באמצעות getEventFlagWord()) מ- אובייקט MessageQueue שאותחל כדי ליצור את הדגל, וגם משתמשים בדגל הזה כדי ליצור את אובייקט EventFlag הדרוש.
  4. משתמשים בשיטה MessageQueue getDesc() כדי לקבל שמתאר את האובייקט.
  5. בקובץ .hal, צריך לציין ל-method פרמטר מסוג fmq_sync או fmq_unsync כאשר T הוא סוג מתאים בהגדרת HIDL. משתמשים בפונקציה הזו כדי לשלוח את האובייקט שהוחזר על ידי getDesc() לתהליך המקבל.

בצד המקבל:

  1. כדי ליצור אובייקט MessageQueue, צריך להשתמש באובייקט המתאר. להיות חשוב להשתמש באותו סגנון ובאותו סוג נתונים בתור, או שהתבנית לא תצליח להדר.
  2. אם חילצת דגל של אירוע, צריך לחלץ את הדגל מהמערך המתאים אובייקט MessageQueue בתהליך המקבל.
  3. כדי להעביר נתונים צריך להשתמש באובייקט MessageQueue.