Compare commits
33 Commits
d1f92d3cad
...
7ef14d586f
Author | SHA1 | Date |
---|---|---|
Vitaliy Filippov | 7ef14d586f | |
Vitaliy Filippov | 87a2dab9fc | |
Vitaliy Filippov | 178b2b5f29 | |
Vitaliy Filippov | 40591feed9 | |
Vitaliy Filippov | fc613bab16 | |
Vitaliy Filippov | 5cbdff4951 | |
Vitaliy Filippov | 37ae8ea273 | |
Vitaliy Filippov | c9b2da267e | |
Vitaliy Filippov | abe3de9e8b | |
Vitaliy Filippov | f5fee7b5f1 | |
Vitaliy Filippov | 47d5cd2345 | |
Vitaliy Filippov | 25e1944342 | |
Vitaliy Filippov | 0e3373289f | |
Vitaliy Filippov | a523f334d8 | |
Vitaliy Filippov | cdc24938be | |
Vitaliy Filippov | a9275989f6 | |
Vitaliy Filippov | e509dfbe93 | |
Vitaliy Filippov | 3e051a77cb | |
Vitaliy Filippov | 69fe41c5d1 | |
Vitaliy Filippov | f5229d6b29 | |
Vitaliy Filippov | c1ea3d007f | |
Vitaliy Filippov | 4072f292d4 | |
Vitaliy Filippov | a4d84b27fc | |
Vitaliy Filippov | b327b6413e | |
Vitaliy Filippov | 93f334054f | |
Vitaliy Filippov | f1c5d2abaa | |
Vitaliy Filippov | e56371a49c | |
Vitaliy Filippov | da1cb65e0c | |
Vitaliy Filippov | 2f184faa08 | |
Vitaliy Filippov | d6584ee1b0 | |
Vitaliy Filippov | 59b4fb0ad7 | |
Vitaliy Filippov | 6997cc632a | |
Vitaliy Filippov | d456b28d6f |
|
@ -32,7 +32,7 @@ void blockstore_init_meta::handle_event(ring_data_t *data, int buf_num)
|
|||
if (data->res < 0)
|
||||
{
|
||||
throw std::runtime_error(
|
||||
std::string("read metadata failed at offset ") + std::to_string(buf_num >= 0 ? bufs[buf_num].offset : last_read_offset) +
|
||||
std::string("read metadata failed at offset ") + std::to_string(bufs[buf_num].offset) +
|
||||
std::string(": ") + strerror(-data->res)
|
||||
);
|
||||
}
|
||||
|
@ -63,7 +63,6 @@ int blockstore_init_meta::loop()
|
|||
throw std::runtime_error("Failed to allocate metadata read buffer");
|
||||
// Read superblock
|
||||
GET_SQE();
|
||||
last_read_offset = 0;
|
||||
data->iov = { metadata_buffer, (size_t)bs->dsk.meta_block_size };
|
||||
data->callback = [this](ring_data_t *data) { handle_event(data, -1); };
|
||||
my_uring_prep_readv(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset);
|
||||
|
@ -101,7 +100,6 @@ resume_1:
|
|||
{
|
||||
printf("Initializing metadata area\n");
|
||||
GET_SQE();
|
||||
last_read_offset = 0;
|
||||
data->iov = (struct iovec){ metadata_buffer, (size_t)bs->dsk.meta_block_size };
|
||||
data->callback = [this](ring_data_t *data) { handle_event(data, -1); };
|
||||
my_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset);
|
||||
|
@ -261,11 +259,9 @@ resume_2:
|
|||
next_offset = entries_to_zero[i]/entries_per_block;
|
||||
for (j = i; j < entries_to_zero.size() && entries_to_zero[j]/entries_per_block == next_offset; j++) {}
|
||||
GET_SQE();
|
||||
last_read_offset = (1+next_offset)*bs->dsk.meta_block_size;
|
||||
data->iov = { metadata_buffer, (size_t)bs->dsk.meta_block_size };
|
||||
data->callback = [this](ring_data_t *data) { handle_event(data, -1); };
|
||||
my_uring_prep_readv(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset + (1+next_offset)*bs->dsk.meta_block_size);
|
||||
bs->ringloop->submit();
|
||||
submitted++;
|
||||
resume_5:
|
||||
if (submitted > 0)
|
||||
|
@ -282,7 +278,6 @@ resume_5:
|
|||
data->iov = { metadata_buffer, (size_t)bs->dsk.meta_block_size };
|
||||
data->callback = [this](ring_data_t *data) { handle_event(data, -1); };
|
||||
my_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset + (1+next_offset)*bs->dsk.meta_block_size);
|
||||
bs->ringloop->submit();
|
||||
submitted++;
|
||||
resume_6:
|
||||
if (submitted > 0)
|
||||
|
@ -304,7 +299,6 @@ resume_6:
|
|||
{
|
||||
GET_SQE();
|
||||
my_uring_prep_fsync(sqe, bs->dsk.meta_fd, IORING_FSYNC_DATASYNC);
|
||||
last_read_offset = 0;
|
||||
data->iov = { 0 };
|
||||
data->callback = [this](ring_data_t *data) { handle_event(data, -1); };
|
||||
submitted++;
|
||||
|
|
|
@ -23,7 +23,6 @@ class blockstore_init_meta
|
|||
struct ring_data_t *data;
|
||||
uint64_t md_offset = 0;
|
||||
uint64_t next_offset = 0;
|
||||
uint64_t last_read_offset = 0;
|
||||
uint64_t entries_loaded = 0;
|
||||
unsigned entries_per_block = 0;
|
||||
int i = 0, j = 0;
|
||||
|
|
|
@ -64,6 +64,10 @@ std::string validate_pool_config(json11::Json::object & new_cfg, json11::Json ol
|
|||
(new_cfg["parity_chunks"].uint64_value() > 1 ? 1 : 0);
|
||||
}
|
||||
}
|
||||
if (new_cfg["scheme"] != "ec")
|
||||
{
|
||||
new_cfg.erase("parity_chunks");
|
||||
}
|
||||
|
||||
// Check integer values and unknown keys
|
||||
for (auto kv_it = new_cfg.begin(); kv_it != new_cfg.end(); )
|
||||
|
@ -114,12 +118,6 @@ std::string validate_pool_config(json11::Json::object & new_cfg, json11::Json ol
|
|||
}
|
||||
}
|
||||
|
||||
// Check after merging
|
||||
if (new_cfg["scheme"] != "ec")
|
||||
{
|
||||
new_cfg.erase("parity_chunks");
|
||||
}
|
||||
|
||||
// Prevent autovivification of object keys. Now we don't modify the config, we just check it
|
||||
json11::Json cfg = new_cfg;
|
||||
|
||||
|
|
|
@ -238,8 +238,7 @@ void cluster_client_t::erase_op(cluster_op_t *op)
|
|||
// which may continue following SYNCs, but these SYNCs
|
||||
// should know about the changed buffer state
|
||||
// This is ugly but this is the way we do it
|
||||
auto cb = std::move(op->callback);
|
||||
cb(op);
|
||||
std::function<void(cluster_op_t*)>(op->callback)(op);
|
||||
}
|
||||
if (!(flags & OP_IMMEDIATE_COMMIT) || enable_writeback)
|
||||
{
|
||||
|
@ -249,8 +248,7 @@ void cluster_client_t::erase_op(cluster_op_t *op)
|
|||
{
|
||||
// Call callback at the end to avoid inconsistencies in prev_wait
|
||||
// if the callback adds more operations itself
|
||||
auto cb = std::move(op->callback);
|
||||
cb(op);
|
||||
std::function<void(cluster_op_t*)>(op->callback)(op);
|
||||
}
|
||||
if (flags & OP_FLUSH_BUFFER)
|
||||
{
|
||||
|
@ -550,8 +548,7 @@ void cluster_client_t::execute(cluster_op_t *op)
|
|||
op->opcode != OSD_OP_READ_BITMAP && op->opcode != OSD_OP_READ_CHAIN_BITMAP && op->opcode != OSD_OP_WRITE)
|
||||
{
|
||||
op->retval = -EINVAL;
|
||||
auto cb = std::move(op->callback);
|
||||
cb(op);
|
||||
std::function<void(cluster_op_t*)>(op->callback)(op);
|
||||
return;
|
||||
}
|
||||
if (!pgs_loaded)
|
||||
|
@ -589,8 +586,7 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
|
|||
wb->start_writebacks(this, 1);
|
||||
}
|
||||
op->retval = op->len;
|
||||
auto cb = std::move(op->callback);
|
||||
cb(op);
|
||||
std::function<void(cluster_op_t*)>(op->callback)(op);
|
||||
return;
|
||||
}
|
||||
if (op->opcode == OSD_OP_WRITE && !(op->flags & OP_IMMEDIATE_COMMIT))
|
||||
|
@ -659,8 +655,7 @@ bool cluster_client_t::check_rw(cluster_op_t *op)
|
|||
if (!pool_id)
|
||||
{
|
||||
op->retval = -EINVAL;
|
||||
auto cb = std::move(op->callback);
|
||||
cb(op);
|
||||
std::function<void(cluster_op_t*)>(op->callback)(op);
|
||||
return false;
|
||||
}
|
||||
auto pool_it = st_cli.pool_config.find(pool_id);
|
||||
|
@ -668,8 +663,7 @@ bool cluster_client_t::check_rw(cluster_op_t *op)
|
|||
{
|
||||
// Pools are loaded, but this one is unknown
|
||||
op->retval = -EINVAL;
|
||||
auto cb = std::move(op->callback);
|
||||
cb(op);
|
||||
std::function<void(cluster_op_t*)>(op->callback)(op);
|
||||
return false;
|
||||
}
|
||||
// Check alignment
|
||||
|
@ -677,8 +671,7 @@ bool cluster_client_t::check_rw(cluster_op_t *op)
|
|||
op->offset % pool_it->second.bitmap_granularity || op->len % pool_it->second.bitmap_granularity)
|
||||
{
|
||||
op->retval = -EINVAL;
|
||||
auto cb = std::move(op->callback);
|
||||
cb(op);
|
||||
std::function<void(cluster_op_t*)>(op->callback)(op);
|
||||
return false;
|
||||
}
|
||||
if (pool_it->second.immediate_commit == IMMEDIATE_ALL)
|
||||
|
@ -691,8 +684,7 @@ bool cluster_client_t::check_rw(cluster_op_t *op)
|
|||
if (ino_it != st_cli.inode_config.end() && ino_it->second.readonly)
|
||||
{
|
||||
op->retval = -EROFS;
|
||||
auto cb = std::move(op->callback);
|
||||
cb(op);
|
||||
std::function<void(cluster_op_t*)>(op->callback)(op);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
|
|
@ -118,8 +118,7 @@ void kv_cli_t::run(const json11::Json::object & cfg)
|
|||
finished = true;
|
||||
}
|
||||
});
|
||||
interactive = isatty(0);
|
||||
if (interactive)
|
||||
interactive = true;
|
||||
printf("> ");
|
||||
}
|
||||
catch (std::exception & e)
|
||||
|
@ -237,7 +236,7 @@ void kv_cli_t::handle_cmd(const std::string & cmd, std::function<void()> cb)
|
|||
if (res < 0)
|
||||
fprintf(stderr, "Error opening index: %s (code %d)\n", strerror(-res), res);
|
||||
else
|
||||
fprintf(interactive ? stdout : stderr, "Index opened. Current size: %lu bytes\n", db->get_size());
|
||||
printf("Index opened. Current size: %lu bytes\n", db->get_size());
|
||||
cb();
|
||||
});
|
||||
}
|
||||
|
@ -273,15 +272,15 @@ void kv_cli_t::handle_cmd(const std::string & cmd, std::function<void()> cb)
|
|||
}
|
||||
else if (opname == "get" || opname == "set" || opname == "del")
|
||||
{
|
||||
std::string key = scan_escaped(cmd, pos);
|
||||
if (opname == "get" || opname == "del")
|
||||
{
|
||||
if (key == "")
|
||||
if (pos == std::string::npos)
|
||||
{
|
||||
fprintf(stderr, "Usage: %s <key>\n", opname.c_str());
|
||||
cb();
|
||||
return;
|
||||
}
|
||||
auto key = trim(cmd.substr(pos+1));
|
||||
if (opname == "get")
|
||||
{
|
||||
db->get(key, [this, cb](int res, const std::string & value)
|
||||
|
@ -303,33 +302,34 @@ void kv_cli_t::handle_cmd(const std::string & cmd, std::function<void()> cb)
|
|||
if (res < 0)
|
||||
fprintf(stderr, "Error: %s (code %d)\n", strerror(-res), res);
|
||||
else
|
||||
fprintf(interactive ? stdout : stderr, "OK\n");
|
||||
printf("OK\n");
|
||||
cb();
|
||||
});
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
if (key == "" || pos >= cmd.size())
|
||||
auto pos2 = cmd.find_first_of(" \t", pos+1);
|
||||
if (pos2 == std::string::npos)
|
||||
{
|
||||
fprintf(stderr, "Usage: set <key> <value>\n");
|
||||
cb();
|
||||
return;
|
||||
}
|
||||
auto value = trim(cmd.substr(pos));
|
||||
auto key = trim(cmd.substr(pos+1, pos2-pos-1));
|
||||
auto value = trim(cmd.substr(pos2+1));
|
||||
db->set(key, value, [this, cb](int res)
|
||||
{
|
||||
if (res < 0)
|
||||
fprintf(stderr, "Error: %s (code %d)\n", strerror(-res), res);
|
||||
else
|
||||
fprintf(interactive ? stdout : stderr, "OK\n");
|
||||
printf("OK\n");
|
||||
cb();
|
||||
});
|
||||
}
|
||||
}
|
||||
else if (opname == "list" || opname == "dump")
|
||||
else if (opname == "list")
|
||||
{
|
||||
bool dump = opname == "dump";
|
||||
std::string start, end;
|
||||
if (pos != std::string::npos)
|
||||
{
|
||||
|
@ -358,9 +358,6 @@ void kv_cli_t::handle_cmd(const std::string & cmd, std::function<void()> cb)
|
|||
}
|
||||
else
|
||||
{
|
||||
if (dump)
|
||||
printf("set %s %s\n", auto_addslashes(key).c_str(), value.c_str());
|
||||
else
|
||||
printf("%s = %s\n", key.c_str(), value.c_str());
|
||||
db->list_next(handle, NULL);
|
||||
}
|
||||
|
@ -370,7 +367,7 @@ void kv_cli_t::handle_cmd(const std::string & cmd, std::function<void()> cb)
|
|||
{
|
||||
db->close([=]()
|
||||
{
|
||||
fprintf(interactive ? stdout : stderr, "Index closed\n");
|
||||
printf("Index closed\n");
|
||||
cb();
|
||||
});
|
||||
}
|
||||
|
@ -385,8 +382,7 @@ void kv_cli_t::handle_cmd(const std::string & cmd, std::function<void()> cb)
|
|||
stderr, "Unknown operation: %s. Supported operations:\n"
|
||||
"open <pool_id> <inode_id> [block_size]\n"
|
||||
"config <property> <value>\n"
|
||||
"get <key>\nset <key> <value>\ndel <key>\n"
|
||||
"list [<start> [end]]\ndump [<start> [end]]\n"
|
||||
"get <key>\nset <key> <value>\ndel <key>\nlist [<start> [end]]\n"
|
||||
"close\nquit\n", opname.c_str()
|
||||
);
|
||||
cb();
|
||||
|
|
|
@ -701,13 +701,15 @@ void kv_db_t::find_size(uint64_t min, uint64_t max, int phase, std::function<voi
|
|||
op->inode = inode_id;
|
||||
op->offset = (phase == 1 ? min : (min+max)/2) * ino_block_size;
|
||||
op->len = kv_block_size;
|
||||
if (op->len)
|
||||
{
|
||||
op->iov.push_back(malloc_or_die(op->len), op->len);
|
||||
}
|
||||
op->callback = [=](cluster_op_t *op)
|
||||
{
|
||||
if (op->retval != op->len)
|
||||
{
|
||||
// error
|
||||
free(op->iov.buf[0].iov_base);
|
||||
cb(op->retval >= 0 ? -EIO : op->retval, 0);
|
||||
return;
|
||||
}
|
||||
|
@ -894,9 +896,7 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p
|
|||
if (op->retval != op->len)
|
||||
{
|
||||
// error
|
||||
free(op->iov.buf[0].iov_base);
|
||||
cb(op->retval >= 0 ? -EIO : op->retval, BLK_NOCHANGE);
|
||||
delete op;
|
||||
return;
|
||||
}
|
||||
invalidate(db, op->offset, op->version);
|
||||
|
@ -909,8 +909,6 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p
|
|||
if (blk->updating > 0 && recheck_policy == KV_RECHECK_WAIT)
|
||||
{
|
||||
// Wait until block update stops
|
||||
free(op->iov.buf[0].iov_base);
|
||||
delete op;
|
||||
db->continue_update.emplace(blk->offset, [=, blk_offset = blk->offset]()
|
||||
{
|
||||
get_block(db, offset, cur_level, recheck_policy, cb);
|
||||
|
@ -1341,8 +1339,6 @@ static void write_new_block(kv_db_t *db, kv_block_t *blk, std::function<void(int
|
|||
del_block_level(db, blk);
|
||||
db->block_cache.erase(blk->offset);
|
||||
cb(op->retval >= 0 ? -EIO : op->retval, NULL);
|
||||
free(op->iov.buf[0].iov_base);
|
||||
delete op;
|
||||
return;
|
||||
}
|
||||
invalidate(db, op->offset, op->version);
|
||||
|
@ -1357,8 +1353,6 @@ static void write_new_block(kv_db_t *db, kv_block_t *blk, std::function<void(int
|
|||
// Block is already occupied => place again
|
||||
place_again(db, blk, cb);
|
||||
}
|
||||
free(op->iov.buf[0].iov_base);
|
||||
delete op;
|
||||
};
|
||||
db->cli->execute(op);
|
||||
}
|
||||
|
|
|
@ -348,65 +348,3 @@ std::vector<std::string> explode(const std::string & sep, const std::string & va
|
|||
}
|
||||
return res;
|
||||
}
|
||||
|
||||
// extract possibly double-quoted part of string with escape characters
|
||||
std::string scan_escaped(const std::string & cmd, size_t & pos)
|
||||
{
|
||||
std::string key;
|
||||
auto pos2 = cmd.find_first_not_of(" \t\r\n", pos);
|
||||
if (pos2 == std::string::npos)
|
||||
{
|
||||
pos = cmd.size();
|
||||
return "";
|
||||
}
|
||||
pos = pos2;
|
||||
if (cmd[pos] != '"')
|
||||
{
|
||||
pos2 = cmd.find_first_of(" \t\r\n", pos);
|
||||
pos2 = pos2 == std::string::npos ? cmd.size() : pos2;
|
||||
key = cmd.substr(pos, pos2-pos);
|
||||
pos2 = cmd.find_first_not_of(" \t\r\n", pos2);
|
||||
pos = pos2 == std::string::npos ? cmd.size() : pos2;
|
||||
return key;
|
||||
}
|
||||
pos++;
|
||||
while (pos < cmd.size())
|
||||
{
|
||||
auto pos2 = cmd.find_first_of("\\\"", pos);
|
||||
pos2 = pos2 == std::string::npos ? cmd.size() : pos2;
|
||||
if (pos2 > pos)
|
||||
key += cmd.substr(pos, pos2-pos);
|
||||
pos = pos2;
|
||||
if (pos >= cmd.size())
|
||||
break;
|
||||
if (cmd[pos] == '"')
|
||||
{
|
||||
pos++;
|
||||
break;
|
||||
}
|
||||
if (cmd[pos] == '\\')
|
||||
{
|
||||
if (pos < cmd.size()-1)
|
||||
key += cmd[++pos];
|
||||
pos++;
|
||||
}
|
||||
}
|
||||
return key;
|
||||
}
|
||||
|
||||
std::string auto_addslashes(const std::string & str)
|
||||
{
|
||||
auto pos = str.find_first_of("\\\"");
|
||||
if (pos == std::string::npos)
|
||||
return str;
|
||||
std::string res = "\""+str.substr(0, pos)+"\\"+str[pos];
|
||||
while (pos < str.size()-1)
|
||||
{
|
||||
auto pos2 = str.find_first_of("\\\"", pos+1);
|
||||
if (pos2 == std::string::npos)
|
||||
return res + str.substr(pos+1) + "\"";
|
||||
res += str.substr(pos, pos2-pos)+"\\"+str[pos2];
|
||||
pos = pos2;
|
||||
}
|
||||
return res+"\"";
|
||||
}
|
||||
|
|
|
@ -22,5 +22,3 @@ std::string str_repeat(const std::string & str, int times);
|
|||
size_t utf8_length(const std::string & s);
|
||||
size_t utf8_length(const char *s);
|
||||
std::vector<std::string> explode(const std::string & sep, const std::string & value, bool trim);
|
||||
std::string scan_escaped(const std::string & cmd, size_t & pos);
|
||||
std::string auto_addslashes(const std::string & str);
|
||||
|
|
Loading…
Reference in New Issue