tarantool을 위한 자체 제한 만료 모듈 작성

얼마 전 우리는 공간의 튜플을 정리하는 문제에 직면했습니다. 타란툴. tarantool에 이미 메모리가 부족할 때가 아니라 사전에 특정 빈도로 청소를 시작해야 했습니다. 이 작업을 위해 tarantool에는 Lua로 작성된 모듈이 있습니다. 만료. 이 모듈을 짧은 시간 동안 사용한 후 우리는 이것이 우리에게 적합하지 않다는 것을 깨달았습니다. 대량의 데이터를 지속적으로 정리하기 때문에 Lua는 GC에 걸렸습니다. 따라서 우리는 네이티브 프로그래밍 언어로 작성된 코드가 가능한 최선의 방법으로 문제를 해결할 수 있기를 바라면서 자체적으로 제한이 있는 만료 모듈을 개발하는 것에 대해 생각했습니다.

우리에게 좋은 예는 다음과 같은 tarantool 모듈이었습니다. memcached. 여기에 사용된 접근 방식은 튜플의 수명, 즉 ttl을 나타내는 별도의 필드가 공간에 생성된다는 사실을 기반으로 합니다. 백그라운드의 모듈은 공간을 스캔하고 TTL을 현재 시간과 비교하여 튜플을 삭제할지 여부를 결정합니다. memcached 모듈 코드는 간단하고 우아하지만 너무 일반적입니다. 첫째, 크롤링 및 삭제되는 인덱스 유형을 고려하지 않습니다. 둘째, 각 패스에서 모든 튜플이 스캔되며 그 수가 상당히 클 수 있습니다. 그리고 만료된 모듈에서 첫 번째 문제가 해결되었다면(트리 인덱스가 별도의 클래스로 분리됨) 두 번째 문제는 여전히 관심을 받지 못했습니다. 이 세 가지 사항은 내 자신의 코드를 작성하는 데 유리한 선택을 미리 결정했습니다.

기술

tarantool에 대한 문서에는 매우 좋은 내용이 있습니다. 지도 시간 C에서 저장 프로시저를 작성하는 방법에 대해 설명합니다. 우선, 아래에 표시될 명령과 코드가 포함된 삽입을 이해하기 위해 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)

단순화를 위해 libcapped-expirationd.so 라이브러리가 있는 디렉터리에서 tarantool을 시작합니다. 라이브러리에서 두 가지 함수(start 및 kill)를 내보냅니다. 첫 번째 단계는 box.schema.func.create 및 box.schema.user.grant를 사용하여 Lua에서 이러한 기능을 사용할 수 있도록 만드는 것입니다. 그런 다음 튜플에 세 개의 필드만 포함되는 공간을 만듭니다. 첫 번째는 고유 식별자, 두 번째는 이메일, 세 번째는 튜플의 수명입니다. 첫 번째 필드 위에 트리 인덱스를 구축하고 이를 기본이라고 부릅니다. 다음으로 네이티브 라이브러리에 대한 연결 개체를 가져옵니다.

준비 작업이 끝나면 시작 기능을 실행합니다.

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 함수의 첫 번째 인수는 작업의 고유 이름입니다. 두 번째는 공간 식별자입니다. 세 번째는 튜플이 삭제되는 고유 인덱스입니다. 네 번째는 튜플을 탐색하는 데 사용되는 인덱스입니다. 다섯 번째는 수명이 있는 튜플 필드의 수입니다(번호는 1이 아닌 0부터 시작합니다!). 여섯번째와 일곱번째는 스캐닝 설정입니다. 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}

삽입이 성공했는지 확인해 보겠습니다.

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

60초 이상(첫 번째 튜플 삽입 시작부터 계산) 후에 선택을 반복하고 제한 만료 모듈이 이미 처리되었는지 확인합니다.

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초 이상(첫 번째 튜플 삽입 시작부터 계산) 후에 선택을 반복하고 제한 만료 모듈이 이미 처리되었는지 확인합니다.

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

공간에는 아직 30초 정도 더 유지될 튜플이 남아 있습니다. 또한 ID가 2이고 수명이 1576421257인 튜플에서 ID가 3이고 수명이 1576421287인 튜플로 이동할 때 검색이 중지되었습니다. 수명이 1576421287 이상인 튜플은 순서로 인해 검색되지 않았습니다. exp 인덱스 키. 이것이 바로 우리가 처음에 달성하고 싶었던 비용 절감입니다.

작업을 중지해 보겠습니다.

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

Реализация

프로젝트의 모든 기능을 설명하는 가장 좋은 방법은 원본 소스입니다. 암호! 출판의 일부로 우리는 가장 중요한 사항, 즉 공간 우회 알고리즘에만 중점을 둘 것입니다.

start 메소드에 전달하는 인수는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;
};

name 속성은 작업의 이름입니다. space_id 속성은 공간 식별자입니다. rm_index_id 속성은 튜플이 삭제될 고유 인덱스의 식별자입니다. it_index_id 속성은 튜플을 탐색할 인덱스의 식별자입니다. it_index_type 속성은 튜플이 순회되는 인덱스 유형입니다. Filed_no 속성은 수명이 있는 튜플 필드의 수입니다. scan_size 속성은 한 트랜잭션에서 스캔되는 최대 튜플 수입니다. scan_time 속성은 전체 스캔 시간(초)입니다.

구문 분석 인수는 고려하지 않습니다. 이것은 힘들지만 간단한 작업이며 도서관이 여러분을 도와줄 것입니다. msgpuck. 간단한 유형인 mp_bool, mp_double, mp_int, mp_uint 및 mp_array를 사용하지 않고 mp_map 유형의 복잡한 데이터 구조로 Lua에서 전달되는 인덱스에서만 문제가 발생할 수 있습니다. 그러나 전체 인덱스를 구문 분석할 필요는 없습니다. 고유성을 확인하고 유형을 계산하고 식별자를 추출하기만 하면 됩니다.

구문 분석에 사용되는 모든 함수의 프로토타입을 나열합니다.

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보다 크지 않은 각 튜플 블록은 단일 트랜잭션에서 스캔되고 수정됩니다. 성공하면 이 트랜잭션이 커밋되고, 오류가 발생하면 롤백됩니다. 만료_iterate 함수의 마지막 인수는 스캔이 시작되거나 계속되는 반복기에 대한 포인터입니다. 이 반복자는 오류가 발생하거나 공간이 부족하거나 프로세스를 미리 중지할 수 없을 때까지 내부적으로 증가됩니다. expired_expired 함수는 튜플의 수명을 확인하고,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;
}

함수 코드 만료_만료:

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

함수 코드 만료_해제 가능:

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

신청

다음에서 소스 코드를 볼 수 있습니다. 여기에!

출처 : habr.com