כתיבת מודול פג תפוגה משלנו עבור tarantool

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

דוגמה טובה עבורנו הייתה מודול tarantool שנקרא נשלח למחשב. הגישה הנהוגה בו מתבססת על כך שנוצר שדה נפרד בחלל, המעיד על משך חייו של הטפול, במילים אחרות, ttl. המודול ברקע סורק את החלל, משווה את ה-TTL לשעה הנוכחית ומחליט אם למחוק את ה-tuple או לא. קוד המודול ה-memcached הוא פשוט ואלגנטי, אבל גנרי מדי. ראשית, זה לא לוקח בחשבון את סוג האינדקס שנסרק ונמחק. שנית, בכל מעבר נסרקים כל הטפולים, שמספרם יכול להיות די גדול. ואם במודול שפג תוקפו נפתרה הבעיה הראשונה (אינדקס העץ הופרד למחלקה נפרדת), אז השני עדיין לא קיבל תשומת לב. שלוש הנקודות הללו קבעו מראש את הבחירה לטובת כתיבת קוד משלי.

תיאור

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

בואו נתחיל מרחוק ונראה איך נראה מודול עם תפוגה מוגבלת מבחוץ:

fiber = require('fiber')
net_box = require('net.box')
box.cfg{listen = 3300}
box.schema.func.create('libcapped-expirationd.start', {language = 'C'})
box.schema.user.grant('guest', 'execute', 'function', 'libcapped-expirationd.start')
box.schema.func.create('libcapped-expirationd.kill', {language = 'C'})
box.schema.user.grant('guest', 'execute', 'function', 'libcapped-expirationd.kill')
box.schema.space.create('tester')
box.space.tester:create_index('primary', {unique = true, parts = {1, 'unsigned'}})
capped_connection = net_box:new(3300)

למען הפשטות, אנו משיקים את tarantool בספרייה שבה נמצאת ספריית libcapped-expirationd.so שלנו. שתי פונקציות מיוצאות מהספרייה: התחל והרוג. הצעד הראשון הוא להפוך את הפונקציות הללו לזמינות מ- Lua באמצעות box.schema.func.create ו-box.schema.user.grant. לאחר מכן צור מרחב שהטופלים שלו יכילו רק שלושה שדות: הראשון הוא מזהה ייחודי, השני הוא דואר אלקטרוני, והשלישי הוא משך החיים של ה-tuple. אנו בונים אינדקס עץ על גבי השדה הראשון וקוראים לו ראשוני. לאחר מכן נקבל את אובייקט החיבור לספרייה המקורית שלנו.

לאחר עבודת ההכנה, הפעל את פונקציית ההתחלה:

capped_connection:call('libcapped-expirationd.start', {'non-indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.primary, 3, 1024, 3600})

דוגמה זו תעבוד במהלך הסריקה בדיוק כמו המודול שפג תפוגה, שנכתב ב-Lua. הארגומנט הראשון לפונקציית start הוא השם הייחודי של המשימה. השני הוא מזהה החלל. השלישי הוא אינדקס ייחודי שבאמצעותו יימחקו tuples. הרביעי הוא המדד לפיו יעברו הטפולים. החמישי הוא המספר של שדה הטפול עם משך החיים (המספור מתחיל מ-1, לא מ-0!). השישי והשביעי הם הגדרות סריקה. 1024 הוא המספר המרבי של tuples שניתן לראות בעסקה אחת. 3600 - זמן סריקה מלא בשניות.

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

הבה נכניס לחלל מספר tuples עם אורך חיים של 60 שניות:

box.space.tester:insert{0, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{1, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{2, '[email protected]', math.floor(fiber.time()) + 60}

בואו נבדוק שההכנסה הצליחה:

tarantool> box.space.tester.index.primary:select()
---
- - [0, '[email protected]', 1576418976]
  - [1, '[email protected]', 1576418976]
  - [2, '[email protected]', 1576418976]
...

בואו נחזור על הבחירה לאחר 60+ שניות (לספור מתחילת ההכנסה של ה-tuple הראשון) ונראה שמודול ה-capped expirationed כבר עובד:

tarantool> box.space.tester.index.primary:select()
---
  - []
...

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

capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})

הבה נסתכל על דוגמה שנייה שבה נעשה שימוש באינדקס נפרד עבור הסריקה:

fiber = require('fiber')
net_box = require('net.box')
box.cfg{listen = 3300}
box.schema.func.create('libcapped-expirationd.start', {language = 'C'})
box.schema.user.grant('guest', 'execute', 'function', 'libcapped-expirationd.start')
box.schema.func.create('libcapped-expirationd.kill', {language = 'C'})
box.schema.user.grant('guest', 'execute', 'function', 'libcapped-expirationd.kill')
box.schema.space.create('tester')
box.space.tester:create_index('primary', {unique = true, parts = {1, 'unsigned'}})
box.space.tester:create_index('exp', {unique = false, parts = {3, 'unsigned'}})
capped_connection = net_box:new(3300)

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

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

capped_connection:call('libcapped-expirationd.start', {'indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.exp, 3, 1024, 3600})

בוא נכניס שוב כמה טפולים לחלל עם אורך חיים של 60 שניות:

box.space.tester:insert{0, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{1, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{2, '[email protected]', math.floor(fiber.time()) + 60}

לאחר 30 שניות, באנלוגיה, נוסיף עוד כמה טפולים:

box.space.tester:insert{3, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{4, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{5, '[email protected]', math.floor(fiber.time()) + 60}

בואו נבדוק שההכנסה הצליחה:

tarantool> box.space.tester.index.primary:select()
---
- - [0, '[email protected]', 1576421257]
  - [1, '[email protected]', 1576421257]
  - [2, '[email protected]', 1576421257]
  - [3, '[email protected]', 1576421287]
  - [4, '[email protected]', 1576421287]
  - [5, '[email protected]', 1576421287]
...

בואו נחזור על הבחירה לאחר 60+ שניות (לספור מתחילת ההכנסה של ה-tuple הראשון) ונראה שמודול ה-capped expirationed כבר עובד:

tarantool> box.space.tester.index.primary:select()
---
- - [3, '[email protected]', 1576421287]
  - [4, '[email protected]', 1576421287]
  - [5, '[email protected]', 1576421287]
...

עדיין נותרו כמה טיפוסים בחלל שיהיו להם עוד כ-30 שניות לחיות. יתרה מכך, הסריקה הופסקה בעת מעבר מטופל עם מזהה 2 ואורך חיים של 1576421257 לטופל עם מזהה 3 ואורך חיים של 1576421287. Tuples עם אורך חיים של 1576421287 ומעלה לא נסרקו עקב הזמנה של מפתחות ה-exp index. זה החיסכון שרצינו להשיג כבר בהתחלה.

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

capped_connection:call('libcapped-expirationd.kill', {'indexed'})

Реализация

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

הארגומנטים שאנו מעבירים לשיטת ההתחלה מאוחסנים במבנה שנקרא expirationd_task:

struct expirationd_task
{
  char name[256];
  uint32_t space_id;
  uint32_t rm_index_id;
  uint32_t it_index_id;
  uint32_t it_index_type; 
  uint32_t field_no;
  uint32_t scan_size;
  uint32_t scan_time;
};

תכונת השם היא שם המשימה. התכונה space_id היא מזהה החלל. התכונה rm_index_id היא המזהה של האינדקס הייחודי שבאמצעותו tuples יימחקו. התכונה it_index_id היא המזהה של האינדקס שבאמצעותו יעברו tuples. התכונה it_index_type היא סוג האינדקס שבאמצעותו יעברו tuples. התכונה filed_no היא המספר של שדה ה-tuple עם משך החיים. התכונה scan_size היא המספר המרבי של tuples שנסרקים בעסקה אחת. התכונה scan_time היא זמן הסריקה המלא בשניות.

לא נשקול ניתוח טיעונים. זוהי עבודה קפדנית אך פשוטה, שהספרייה תעזור לך בה msgpuck. קשיים יכולים להתעורר רק עם אינדקסים המועברים מ-Lua כמבנה נתונים מורכב עם סוג mp_map, ולא באמצעות הסוגים הפשוטים mp_bool, mp_double, mp_int, mp_uint ו-mp_array. אבל אין צורך לנתח את כל האינדקס. אתה רק צריך לבדוק את הייחודיות שלו, לחשב את הסוג ולחלץ את המזהה.

אנו מפרטים את אבות הטיפוס של כל הפונקציות המשמשות לניתוח:

bool expirationd_parse_name(struct expirationd_task *task, const char **pos);
bool expirationd_parse_space_id(struct expirationd_task *task, const char **pos);
bool expirationd_parse_rm_index_id(struct expirationd_task *task, const char **pos);
bool expirationd_parse_rm_index_unique(struct expirationd_task *task, const char **pos);
bool expirationd_parse_rm_index(struct expirationd_task *task, const char **pos);
bool expirationd_parse_it_index_id(struct expirationd_task *task, const char **pos);
bool expirationd_parse_it_index_type(struct expirationd_task *task, const char **pos);
bool expirationd_parse_it_index(struct expirationd_task *task, const char **pos);
bool expirationd_parse_field_no(struct expirationd_task *task, const char **pos);
bool expirationd_parse_scan_size(struct expirationd_task *task, const char **pos);
bool expirationd_parse_scan_time(struct expirationd_task *task, const char **pos);

עכשיו נעבור לדבר החשוב ביותר - ההיגיון של עקיפת מקום ומחיקת tuples. כל בלוק של tuples לא גדול מ-scan_size נסרק ומשתנה תחת עסקה אחת. אם היא מצליחה, עסקה זו מתבצעת; אם מתרחשת שגיאה, היא מבוטלת. הארגומנט האחרון לפונקציה expirationd_iterate הוא מצביע לאיטרטור שממנו מתחילה או נמשכת הסריקה. איטרטור זה מוגדל באופן פנימי עד שמתרחשת שגיאה, השטח אוזל או שלא ניתן לעצור את התהליך מראש. הפונקציה expirationd_expired בודקת את משך החיים של tuple, expirationd_delete מוחקת tuple, expirationd_breakable בודקת אם אנחנו צריכים להמשיך הלאה.

קוד פונקציה Expirationd_iterate:

static bool
expirationd_iterate(struct expirationd_task *task, box_iterator_t **iterp)
{
  box_iterator_t *iter = *iterp;
  box_txn_begin();
  for (uint32_t i = 0; i < task->scan_size; ++i) {
    box_tuple_t *tuple = NULL;
    if (box_iterator_next(iter, &tuple) < 0) {
      box_iterator_free(iter);
      *iterp = NULL;
      box_txn_rollback();
      return false;
    }
    if (!tuple) {
      box_iterator_free(iter);
      *iterp = NULL;
      box_txn_commit();
      return true;
    }
    if (expirationd_expired(task, tuple))
      expirationd_delete(task, tuple);
    else if (expirationd_breakable(task))
      break;
  }
  box_txn_commit();
  return true;
}

קוד פונקציה expirationd_expired:

static bool
expirationd_expired(struct expirationd_task *task, box_tuple_t *tuple)
{
  const char *buf = box_tuple_field(tuple, task->field_no - 1);
  if (!buf || mp_typeof(*buf) != MP_UINT)
    return false;
  uint64_t val = mp_decode_uint(&buf);
  if (val > fiber_time64() / 1000000)
    return false;
  return true;
}

Expirationd_delete קוד פונקציה:

static void
expirationd_delete(struct expirationd_task *task, box_tuple_t *tuple)
{
  uint32_t len;
  const char *str = box_tuple_extract_key(tuple, task->space_id, task->rm_index_id, &len);
  box_delete(task->space_id, task->rm_index_id, str, str + len, NULL);
}

קוד פונקציה expiration_breakable:

static bool
expirationd_breakable(struct expirationd_task *task)
{
  return task->it_index_id != task->rm_index_id && task->it_index_type == ITER_GT;
}

יישום

אתה יכול לצפות בקוד המקור בכתובת כאן!

מקור: www.habr.com