ტარანტულისთვის ჩვენივე ვადაგასული მოდულის დაწერა

რამდენიმე ხნის წინ სივრცეებში ტუპლების გაწმენდის პრობლემა დაგვხვდა ტარანტული. დასუფთავება უნდა დაწყებულიყო არა მაშინ, როცა ტარანტულს მეხსიერება უკვე ამოეწურა, არამედ წინასწარ და გარკვეული სიხშირით. ამ ამოცანისთვის tarantool-ს აქვს Lua-ზე დაწერილი მოდული, რომელსაც ე.წ ვადის გასვლა. ამ მოდულის ხანმოკლე გამოყენების შემდეგ მივხვდით, რომ ის ჩვენთვის არ იყო შესაფერისი: დიდი რაოდენობით მონაცემების მუდმივი გაწმენდის გამო, ლუა GC-ში ჩამოიხრჩო. ამიტომ, ჩვენ ვიფიქრეთ საკუთარი ვადაგასული მოდულის შემუშავებაზე, იმ იმედით, რომ პროგრამირების მშობლიურ ენაზე დაწერილი კოდი საუკეთესოდ მოაგვარებდა ჩვენს პრობლემებს.

ჩვენთვის კარგი მაგალითი იყო ტარანტულის მოდული ე.წ memcached. მასში გამოყენებული მიდგომა ეფუძნება იმ ფაქტს, რომ სივრცეში იქმნება ცალკე ველი, რომელიც მიუთითებს ტუპლის სიცოცხლეზე, სხვა სიტყვებით რომ ვთქვათ, ttl. ფონური მოდული სკანირებს სივრცეს, ადარებს TTL-ს მიმდინარე დროს და წყვეტს წაშალოს თუ არა ტუპლი. 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 ბიბლიოთეკა. ბიბლიოთეკიდან ექსპორტირებულია ორი ფუნქცია: დაწყება და მოკვლა. პირველი ნაბიჯი არის ამ ფუნქციების ხელმისაწვდომობა ლუასგან box.schema.func.create და box.schema.user.grant გამოყენებით. შემდეგ შექმენით სივრცე, რომლის tuples შეიცავს მხოლოდ სამ ველს: პირველი არის უნიკალური იდენტიფიკატორი, მეორე არის ელფოსტა და მესამე არის ტუპლის სიცოცხლის ხანგრძლივობა. ჩვენ ვაშენებთ ხის ინდექსს პირველი ველის თავზე და ვუწოდებთ პირველად. შემდეგ ჩვენ ვიღებთ კავშირის ობიექტს ჩვენს მშობლიურ ბიბლიოთეკასთან.

მოსამზადებელი სამუშაოების შემდეგ გაუშვით დაწყების ფუნქცია:

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

სკანირების დროს ეს მაგალითი ზუსტად ისევე იმუშავებს, როგორც ვადაგასული მოდული, რომელიც ლუა-ზეა დაწერილი. დაწყების ფუნქციის პირველი არგუმენტი არის ამოცანის უნიკალური სახელი. მეორე არის სივრცის იდენტიფიკატორი. მესამე არის უნიკალური ინდექსი, რომლითაც ტუპლები წაიშლება. მეოთხე არის მაჩვენებელი, რომლითაც გაივლიან ტოპებს. მეხუთე არის დუბლის ველის რიცხვი სიცოცხლის ხანგრძლივობასთან ერთად (ნუმერაცია იწყება 1-დან და არა 0-დან!). მეექვსე და მეშვიდე არის სკანირების პარამეტრები. 1024 არის ტოპების მაქსიმალური რაოდენობა, რომელთა ნახვა შესაძლებელია ერთ ტრანზაქციაში. 3600 — სრული სკანირების დრო წამებში.

გაითვალისწინეთ, რომ მაგალითი იყენებს იგივე ინდექსს ცოცვისა და წაშლისთვის. თუ ეს არის ხის ინდექსი, მაშინ გადაკვეთა ხორციელდება პატარა გასაღებიდან უფრო დიდზე. თუ არსებობს სხვა, მაგალითად, ჰეშის ინდექსი, მაშინ გადაკვეთა ხდება, როგორც წესი, შემთხვევითი თანმიმდევრობით. ყველა კოსმოსური ტიპი დასკანირებულია ერთ სკანირებაში.

მოდით ჩავსვათ რამდენიმე ტოპი სივრცეში 60 წამის სიცოცხლის ხანგრძლივობით:

box.space.tester:insert{0, 'user0@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{1, 'user1@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{2, 'user2@tarantool.io', math.floor(fiber.time()) + 60}

მოდით შევამოწმოთ, რომ ჩასმა წარმატებით დასრულდა:

tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
  - [1, 'user1@tarantool.io', 1576418976]
  - [2, 'user2@tarantool.io', 1576418976]
...

გავიმეოროთ არჩევა 60+ წამის შემდეგ (ითვლით პირველი ტოპის ჩასმის დასაწყისიდან) და ვნახოთ, რომ დახურული ვადაგასული მოდული უკვე დამუშავებულია:

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

შევაჩეროთ დავალება:

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

მოდით შევხედოთ მეორე მაგალითს, სადაც ცალკე ინდექსი გამოიყენება crawl-ისთვის:

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. ეს ინდექსი არ უნდა იყოს უნიკალური, განსხვავებით ინდექსისგან, რომელსაც პირველადი ეწოდება. გადაკვეთა განხორციელდება ექსპლუატაციის ინდექსით, ხოლო წაშლა პირველადი. ჩვენ გვახსოვს, რომ ადრე ორივე კეთდებოდა მხოლოდ პირველადი ინდექსის გამოყენებით.

მოსამზადებელი სამუშაოების შემდეგ ჩვენ ვატარებთ დაწყების ფუნქციას ახალი არგუმენტებით:

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, 'user0@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{1, 'user1@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{2, 'user2@tarantool.io', math.floor(fiber.time()) + 60}

30 წამის შემდეგ, ანალოგიით, ჩვენ დავამატებთ კიდევ რამდენიმე ტოპს:

box.space.tester:insert{3, 'user3@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{4, 'user4@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{5, 'user5@tarantool.io', math.floor(fiber.time()) + 60}

მოდით შევამოწმოთ, რომ ჩასმა წარმატებით დასრულდა:

tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576421257]
  - [1, 'user1@tarantool.io', 1576421257]
  - [2, 'user2@tarantool.io', 1576421257]
  - [3, 'user3@tarantool.io', 1576421287]
  - [4, 'user4@tarantool.io', 1576421287]
  - [5, 'user5@tarantool.io', 1576421287]
...

გავიმეოროთ არჩევა 60+ წამის შემდეგ (ითვლით პირველი ტოპის ჩასმის დასაწყისიდან) და ვნახოთ, რომ დახურული ვადაგასული მოდული უკვე დამუშავებულია:

tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
  - [4, 'user4@tarantool.io', 1576421287]
  - [5, 'user5@tarantool.io', 1576421287]
...

სივრცეში ჯერ კიდევ დარჩა რამდენიმე ტოპი, რომლებსაც კიდევ 30 წამი ექნებათ სიცოცხლე. უფრო მეტიც, სკანირება შეწყდა, როდესაც გადაინაცვლებდა ტოპიდან ID 2-ით და სიცოცხლის ხანგრძლივობა 1576421257 ტუპლზე ID 3-ით და 1576421287 სიცოცხლის ხანგრძლივობაზე. ექსპლუატაციის ინდექსის გასაღებები. ეს არის დანაზოგი, რომლის მიღწევაც თავიდანვე გვინდოდა.

შევაჩეროთ დავალება:

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 ატრიბუტი არის უნიკალური ინდექსის იდენტიფიკატორი, რომლითაც წყლები წაიშლება. it_index_id ატრიბუტი არის ინდექსის იდენტიფიკატორი, რომლითაც მოხდება ტოპების გავლა. it_index_type ატრიბუტი არის ინდექსის ტიპი, რომლითაც მოხდება ტოტების გავლა. filed_no ატრიბუტი არის tuple ველის ნომერი სიცოცხლის ხანგრძლივობასთან ერთად. scan_size ატრიბუტი არის ტოპების მაქსიმალური რაოდენობა, რომლებიც დასკანირებულია ერთ ტრანზაქციაში. 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);

ახლა გადავიდეთ ყველაზე მნიშვნელოვანზე - სივრცის გვერდის ავლით და ტუპლების წაშლის ლოგიკაზე. ტოპების თითოეული ბლოკი, რომელიც არ აღემატება scan_size-ს, დასკანირებულია და იცვლება ერთი ტრანზაქციის ფარგლებში. წარმატების შემთხვევაში, ეს ტრანზაქცია ჩადენილია; თუ შეცდომა მოხდა, ის უკან იხევა. expirationd_iterate ფუნქციის ბოლო არგუმენტი არის ინდიკატორი იტერატორისკენ, საიდანაც იწყება ან გრძელდება სკანირება. ეს გამეორება იზრდება შიგნით, სანამ შეცდომა არ მოხდება, სივრცე ამოიწურება ან პროცესის წინასწარ შეჩერება შეუძლებელია. ფუნქცია expirationd_expired ამოწმებს tuple-ის სიცოცხლეს, expirationd_delete წაშლის ტოპს, 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);
}

Expirationd_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