Some time ago, we faced the challenge of cleaning tuples in spaces . The cleanup needed to be initiated not when Tarantool was already running out of memory, but in advance and at specific intervals. For this task, Tarantool has a module written in Lua called . After a short period of using this module, we realized it wasn't suitable for us: during constant cleanups of large volumes of data, Lua was stalling in GC. Therefore, we considered developing our own capped expirationd module, hoping that code written in a native programming language would better address our needs.
A good example for us was the Tarantool module called . The approach used in it is based on having a separate field in the space that specifies the lifetime of the tuple, in other words, ttl. The module scans the space in the background, compares ttl with the current time, and decides whether to delete the tuple or not. The code of the memcached module is simple and elegant, but too general. Firstly, it does not consider the type of index used for traversal and deletion. Secondly, all tuples are scanned on each pass, which can be quite large in number. While in the expirationd module the first problem was addressed (tree index was allocated to a separate class), the second still received no attention. These three points dictated the choice in favor of writing our own code.
Description
The documentation for Tarantool has a very good on how to write your own stored procedures in C. I first suggest getting acquainted with it to understand the inserts with commands and code that will be encountered below. It is also worth paying attention to the to the objects available when writing your own capped module, namely to , , and .
Let's start from the basics and look at what the capped expirationd module looks like on the outside:
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)To simplify, we run tarantool in the directory where our library libcapped-expirationd.so is located. Two functions are exported from the library: start and kill. First, we need to make these functions available from Lua using box.schema.func.create and box.schema.user.grant. Then, we create a space whose tuples will contain only three fields: the first is a unique identifier, the second is an email address, and the third is the lifetime of the tuple. We build a tree index on the first field and call it primary. Next, we obtain the connection object to our native library.
After the preparatory work, we run the start function:
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})This example will work exactly the same way as the expirationd module written in Lua. The first argument to the start function is a unique task name. The second is the space identifier. The third is a unique index by which tuples will be deleted. The fourth is the index by which tuples will be scanned. The fifth is the index of the tuple field with the lifetime (numbering starts from 1, not 0!). The sixth and seventh are the scanning settings. 1024 is the maximum number of tuples viewed in a single transaction. 3600 is the total scanning time in seconds.
Note that the same index is used for scanning and deleting in the example. If it is a tree index, the scan proceeds from the smaller key to the larger one. If it is some other type, such as a hash index, scanning generally occurs in a random order. All tuples of the space are viewed in one scan.
Let's insert several tuples into the space with a lifetime of 60 seconds:
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}Let's check that the insertion was successful:
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
- [1, 'user1@tarantool.io', 1576418976]
- [2, 'user2@tarantool.io', 1576418976]
...We will repeat the select after 60+ seconds (counting from the start of the insertion of the first tuple) and see that the capped expirationd module has already executed:
tarantool> box.space.tester.index.primary:select()
---
- []
...We will stop the task:
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})Let's consider a second example where a separate index is used for iteration:
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)Here everything is the same as in the first example, with one small exception. We build a tree index on top of the third field and name it exp. This index does not need to be unique, unlike the index named primary. The iteration will take place using the exp index, while deletion will occur using the primary. We recall that previously, both actions were only done using the primary index.
After the preparatory work, we launch the start function with new arguments:
capped_connection:call('libcapped-expirationd.start', {'indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.exp, 3, 1024, 3600})Again, we will insert several tuples into the space with a lifespan of 60 seconds:
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}After 30 seconds, we will similarly add a few more tuples:
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}Let's check that the insertion was successful:
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]
...We will repeat the select after 60+ seconds (counting from the start of the insertion of the first tuple) and see that the capped expirationd module has already executed:
tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
- [4, 'user4@tarantool.io', 1576421287]
- [5, 'user5@tarantool.io', 1576421287]
...In the space, there are tuples that still have about 30 seconds to live. Moreover, the scan stopped when transitioning from the tuple with ID 2 and a lifespan of 1576421257 to the tuple with ID 3 and a lifespan of 1576421287. Tuples with a lifespan of 1576421287 or more were not viewed due to the ordering of the exp index keys. This is the efficiency we wanted to achieve in the first place.
We will stop the task:
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Implementation
The original source will always highlight all the special features of the project best. ! In this publication, we will only focus on the most important aspects, namely the algorithms for iterating through the space.
The arguments we pass to the start method are saved in a structure called 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;
};The name attribute is the task name. The space_id attribute is the space identifier. The rm_index_id attribute is the identifier of the unique index used for deleting tuples. The it_index_id attribute is the identifier of the index used for traversing tuples. The it_index_type attribute represents the type of index used for traversing tuples. The filed_no attribute is the number of the tuple field containing the lifetime. The scan_size attribute indicates the maximum number of tuples scanned in a single transaction. The scan_time attribute represents the total scanning time in seconds.
We will not consider argument parsing. It is a meticulous but straightforward job that you can accomplish with the library . Difficulties may arise only with the indices that are transferred from Lua in the form of a complex data structure of type mp_map, rather than through simple types like mp_bool, mp_double, mp_int, mp_uint, and mp_array. However, it is not necessary to parse the entire index. It is sufficient to verify its uniqueness, determine the type, and extract the identifier.
Let’s enumerate the prototypes of all functions used for parsing:
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);Now, let's move on to the most important part — the logic of traversing the space and deleting tuples. Each block of tuples, not larger than scan_size, is processed and altered within a single transaction. In case of success, this transaction is committed; in case of an error, it is rolled back. The last argument in the expirationd_iterate function is a pointer to the iterator from which the scanning starts or continues. This iterator is incremented until an error occurs, the space is exhausted, or the process can be stopped early. The expirationd_expired function checks the lifetime of the tuple, expirationd_delete — deletes the tuple, expirationd_breakable — checks if we need to move further.
Code for the expirationd_iterate function:
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;
}Code for the expirationd_expired function:
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;
}Code for the expirationd_delete function:
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);
}Code for the expirationd_breakable function:
static bool
expirationd_breakable(struct expirationd_task *task)
{
return task->it_index_id != task->rm_index_id && task->it_index_type == ITER_GT;
}Application
You can view the source code at !
Source: habr.com
