Shkruajmë modulinn tonë të capped expirationd për tarantool

Disa kohë më parë, para nesh doli problemi i pastrimit të tupave në hapësira tarantool. Pastrimi duhej të niste jo kur tarantool po i mbaronte kujtesa, por paraprakisht dhe me një periodikë të caktuar. Për këtë detyrë, në tarantool ka një modul të shkruar në Lua, i quajtur expirationd. Pas një përdorimi të shkurtër të këtij moduli, ne kuptuam se ai nuk na përshtatej: në pastrimin e përhershëm të volumit të madh të të dhënave, Lua ngec në GC. Prandaj, ne menduam për zhvillimin e modulit tonë të capped expirationd, duke shpresuar se kodi i shkruar në një gjuhë programimi native do të zgjidhte detyrat tona në mënyrën më të mirë.

Një shembull i mirë për ne ishte moduli tarantool i quajtur memcached. Qasja e përdorur në të bazohet në atë që në hapësirë krijohet një fushë e veçantë, në të cilën specifikohet koha e jetës së tupës, për ta thënë ndryshe, ttl. Moduli në sfond skanon hapësirën, krahasohet ttl me kohën aktuale dhe merr vendimin nëse të fshijë tupin apo jo. Kodi i modulit memcached është i thjeshtë dhe elegant, por shumë i përgjithshëm. Së pari, ai nuk merr parasysh tipin e indekseve, sipas të cilëve bëhet kalimi dhe fshirja. Së dyti, në çdo kalim skanohen të gjitha tupat, numri i të cilave mund të jetë mjaft i madh. Dhe nëse në modulinn expirationd problemi i parë ishte zgjidhur (indeksi tree u ndan në një klasë të veçantë), problemi i dytë ende nuk iu kushtua asnjë vëmendje. Këto tre pika e përcaktuan zgjedhjen në favor të shkruarjes së kodit tonë.

Përshkrimi

Në dokumentacionin e tarantool ka një tutorial shumë të mirë për mënyrën se si të shkruani procedurat tuaja të ruajtura në gjuhën C. Së pari, sugjeroj të njiheni me të, për të kuptuar ato futje me komandat dhe kodin që do të takoni më poshtë. Po ashtu, vlen të kushtohet vëmendje për referencën në objektet që janë në dispozicion gjatë shkruajtjes së modulit tuaj të capped, sidomos në box, fiber, është emri i skedarit ( dhe txn.

Le të fillojmë nga larg dhe të shohim si duket moduli capped expirationd nga jashtë:

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)

Për lehtësi, ne e fillojmë tarantool në katalogun ku ndodhet biblioteka jonë libcapped-expirationd.so. Nga biblioteka eksportohen dy funksione: start dhe kill. Fillimisht, duhet t'i bëjmë këto funksione të aksesueshme nga Lua duke përdorur box.schema.func.create dhe box.schema.user.grant. Më pas, krijoni një hapësirë ku tuple-et do të përmbajnë vetëm tri fusha: e para është identifikatori unik, e dyta është adresa elektronike, dhe e treta është koha e jetës së tuple-it. Përsipër të fushës së parë, ndërtojmë një indeks tree dhe e quajmë të parën. Më pas, marrim objektin e lidhjes me bibliotekën tonë natyrore.

Pas përgatitjeve, ne e aktivizojmë funksionin start:

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

Ky shembull do të funksionojë duke skanuar në mënyrë të ngjashme me modulit expirationd, i shkruar në Lua. Argumenti i parë në funksionin start kalon emrin unik të detyrës. I dyti është identifikatori i hapësirës. I treti është indeksi unik, mbi të cilin do të bëhet fshirja e tuple-eve. I katërti është indeksi mbi të cilin do të bëhet kalimi i tuple-eve. I pesti është numri i fushës së tuple-it me kohën e jetës (numërimi shkon nga 1, jo nga 0!). I gjashti dhe i shtati janë cilësimet e skanimit. 1024 është numri maksimal i tuple-eve që shqyrtohen brenda një transaksioni. 3600 është koha totale e skanimit në sekonda.

Vini re se për kalimin dhe fshirjen në shembull përdoret të njëjtin indeks. Nëse është një indeks tree, kalimi bëhet nga çelësi më i vogël në atë më të madhin. Nëse është ndonjë tjetër, për shembull, një indeks hash, kalimi bëhet zakonisht në mënyrë të rastësishme. Në një skanim shqyrtohen të gjithë tuple-t e hapësirës.

Le të bëjmë një insertim në hapësirë me disa tuple me një kohë jete prej 60 sekondash:

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}

Le të kontrollojmë se insertimi ka kaluar me sukses:

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

Le të përsërisim select pas 60+ sekondash (duke llogaritur nga fillimi i insertimit të tuple-it të parë) dhe do të shohim se moduli capped expirationd është aktivizuar:

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

Ndërpresim detyrën:

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

Le të shqyrtojmë shembullin e dytë, kur për traversimin përdoret një indeks i veçantë:

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)

Këtu është po e njëjta gjë si në shembullin e parë, përveç disa detajeve. Përdorim indeksin tree mbi fushën e tretë dhe e quajmë atë exp. Ky indeks nuk ka nevojë të jetë unik, ndryshe nga indeksi i quajtur primary. Traversimi do të kryhet me indeksi exp, ndërsa fshirja me primary. E mbajmë mend se më parë, të dyja bëheshin vetëm duke përdorur indekset primary.

Pas punëve përgatitore, aktivizojmë funksionin start me argumentet e reja:

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

Sërish do të bëjmë një insertim në hapësirë me disa tuple me jetëgjatësi 60 sekonda:

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}

Pas 30 sekondash, për ngjashmëri do të shtojmë edhe disa tuple:

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}

Le të kontrollojmë se insertimi ka kaluar me sukses:

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]
...

Le të përsërisim select pas 60+ sekondash (duke llogaritur nga fillimi i insertimit të tuple-it të parë) dhe do të shohim se moduli capped expirationd është aktivizuar:

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

Në hapësirë kanë mbetur tuple që kanë për të jetuar edhe rreth 30 sekonda. Më shumë, skanimi u ndal te kalimi nga tuple me identifikues 2 dhe jetëgjatësi 1576421257 te tuple me identifikues 3 dhe jetëgjatësi 1576421287. Tuple me jetëgjatësi 1576421287 dhe më shumë nuk u panë për shkak të renditjes së çelësave të indeksit exp. Kjo është ajo ekonomi që ne deshëm të arrijmë në fillim.

Ndërpresim detyrën:

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

Implementimi

Më së miri në të gjitha veçoritë e projektit gjithmonë do të flasë burimi i tij kod! Në kuadër të publikimit ne do të ndalemi vetëm tek momentet më të rëndësishme, sidomos, te algoritmet e traversimit të hapësirës.

Argumentet që ne i kalojmë metodës start ruhen në një strukturë të quajtur expirationd_task:

struktura expirationd_task
{
  char emri[256];
  uint32_t id_hapsire;
  uint32_t id_indeksi_rm;
  uint32_t id_indeksi_it;
  uint32_t lloji_indeksit_it;
  uint32_t numri_fushës;
  uint32_t madhësia_skane;
  uint32_t koha_skane;
};

Atributi emri është emri i detyrës. Atributi id_hapsire është identifikuesi i hapësirës. Atributi id_indeksi_rm është identifikuesi i indeksit unik, sipas të cilit do të kryhet fshirja e tuple-ve. Atributi id_indeksi_it është identifikuesi i indeksit, sipas të cilit do të kryhet kalimi nëpër tuple. Atributi lloji_indeksit_it është tipi i indeksit, sipas të cilit do të kryhet kalimi nëpër tuple. Atributi numri_fushës është numri i fushës së tuple-it me kohën e jetës. Atributi madhësia_skane është numri maksimal i tuple-ve që shqyrtohen brenda një transaksioni. Atributi koha_skane është koha e skanimit të plotë në sekonda.

Nuk do të merremi me analizimin e argumenteve. Kjo është një punë e hollësishme, por jo e komplikuar, në të cilën do t'ju ndihmojë biblioteka msgpuck. Vështirësi mund të lindin vetëm me indeksat, të cilët kalohen nga Lua në formën e një strukture të ndërlikuar të dhënash me tipin mp_map, dhe jo me ndihmën e tipave të thjeshtë mp_bool, mp_double, mp_int, mp_uint dhe mp_array. Por nuk është e nevojshme të analizoni të gjithë indeksin. Mjafton vetëm të kontrolloni unikalitetin e tij, të llogaritni tipin dhe të nxirrni identifikuesin.

Le të rendisim prototipet e të gjitha funksioneve që përdoren për analizimin:

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

Tani tani kalojmë te e rëndësishmja — logjika e skanimit të hapësirës dhe fshirjes së tuple-ve. Çdo bllok tuple-vet me një madhësi jo më të madhe se scan_size shqyrtohet dhe modifikohet nën një transaksion. Në rast suksesi, kjo transaksion përfundon, ndërsa në rast gabimi — rrethohet. Argumenti i fundit në funksionin expirationd_iterate kalon një tregues në iteratorin nga i cili fillon ose vazhdon skanimi. Ky iterator inkrementohet brenda derisa të ndodhi një gabim, hapësira të mbarojë, ose të shfaqet mundësia për të ndaluar procesin paraprakisht. Funksioni expirationd_expired kontrollon kohën e jetesës së tuple-ve, expirationd_delete — fshin tuple-n, expirationd_breakable — kontrollon nëse duhet të vazhdojmë.

Kodi i funksionit 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 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;
}

Kodi i funksionit 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;
}

Kodi i funksionit 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);
}

Kodi i funksionit 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;
}

Aplikacioni

Mund të konsultoheni me kodin burimor në këtu!

Burimi: habr.com