thebestvibedev/tromcho.net
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement FS defragmentation

Browse Source
This commit is contained in:
Vitaliy Filippov
2024-07-12 16:11:35 +03:00
parent 1771d2ef36
commit 990c3ba7eb
16 changed files with 893 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/usage/cli.en.md
+1
View File
@@ -176,6 +176,7 @@ Remove inode data without changing metadata.
--wait-list Retrieve full objects listings before starting to remove objects. --wait-list Retrieve full objects listings before starting to remove objects.
Requires more memory, but allows to show correct removal progress. Requires more memory, but allows to show correct removal progress.
--min-offset Purge only data starting with specified offset. --min-offset Purge only data starting with specified offset.
--max-offset Purge only data before specified offset.
``` ```
## merge-data ## merge-data
docs/usage/cli.ru.md
+1
View File
@@ -184,6 +184,7 @@ vitastor-cli snap-create [-p|--pool <id|name>] <image>@<snapshot>
--wait-list Сначала запросить полный листинг объектов, а потом начать удалять. --wait-list Сначала запросить полный листинг объектов, а потом начать удалять.
Требует больше памяти, но позволяет правильно печатать прогресс удаления. Требует больше памяти, но позволяет правильно печатать прогресс удаления.
--min-offset Удалять только данные, начиная с заданного смещения. --min-offset Удалять только данные, начиная с заданного смещения.
--max-offset Удалять только данные до (исключительно) заданного смещения.
``` ```
## merge-data ## merge-data
src/client/msgr_receive.cpp
+2
View File
@@ -179,6 +179,8 @@ bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, void *curbuf, int rem
bool osd_messenger_t::handle_finished_read(osd_client_t *cl) bool osd_messenger_t::handle_finished_read(osd_client_t *cl)
{ {
cl->ping_time_remaining = osd_ping_timeout;
cl->idle_time_remaining = osd_idle_timeout;
cl->recv_list.reset(); cl->recv_list.reset();
if (cl->read_state == CL_READ_HDR) if (cl->read_state == CL_READ_HDR)
{ {
src/cmd/cli.cpp
+1
View File
@@ -70,6 +70,7 @@ static const char* help_text =
" --wait-list Retrieve full objects listings before starting to remove objects.\n" " --wait-list Retrieve full objects listings before starting to remove objects.\n"
" Requires more memory, but allows to show correct removal progress.\n" " Requires more memory, but allows to show correct removal progress.\n"
" --min-offset Purge only data starting with specified offset.\n" " --min-offset Purge only data starting with specified offset.\n"
" --max-offset Purge only data before specified offset.\n"
"\n" "\n"
"vitastor-cli merge-data <from> <to> [--target <target>]\n" "vitastor-cli merge-data <from> <to> [--target <target>]\n"
" Merge layer data without changing metadata. Merge <from>..<to> to <target>.\n" " Merge layer data without changing metadata. Merge <from>..<to> to <target>.\n"
src/cmd/cli_rm_data.cpp
+5 -3
View File
@@ -25,6 +25,7 @@ struct rm_inode_t
uint64_t inode = 0; uint64_t inode = 0;
pool_id_t pool_id = 0; pool_id_t pool_id = 0;
uint64_t min_offset = 0; uint64_t min_offset = 0;
uint64_t max_offset = 0;
bool down_ok = false; bool down_ok = false;
cli_tool_t *parent = NULL; cli_tool_t *parent = NULL;
@@ -52,7 +53,7 @@ struct rm_inode_t
.obj_done = 0, .obj_done = 0,
.synced = parent->cli->get_immediate_commit(inode), .synced = parent->cli->get_immediate_commit(inode),
}); });
if (min_offset == 0) if (min_offset == 0 && max_offset == 0)
{ {
total_count += objects.size(); total_count += objects.size();
} }
@@ -60,7 +61,7 @@ struct rm_inode_t
{ {
for (object_id oid: objects) for (object_id oid: objects)
{ {
if (oid.stripe >= min_offset) if (oid.stripe >= min_offset && (!max_offset || oid.stripe < max_offset))
{ {
total_count++; total_count++;
} }
@@ -116,7 +117,7 @@ struct rm_inode_t
} }
while (cur_list->in_flight < parent->iodepth && cur_list->obj_pos != cur_list->objects.end()) while (cur_list->in_flight < parent->iodepth && cur_list->obj_pos != cur_list->objects.end())
{ {
if (cur_list->obj_pos->stripe >= min_offset) if (cur_list->obj_pos->stripe >= min_offset && (!max_offset || cur_list->obj_pos->stripe < max_offset))
{ {
osd_op_t *op = new osd_op_t(); osd_op_t *op = new osd_op_t();
op->op_type = OSD_OP_OUT; op->op_type = OSD_OP_OUT;
@@ -287,6 +288,7 @@ std::function<bool(cli_result_t &)> cli_tool_t::start_rm_data(json11::Json cfg)
remover->down_ok = cfg["down_ok"].bool_value(); remover->down_ok = cfg["down_ok"].bool_value();
remover->pool_id = INODE_POOL(remover->inode); remover->pool_id = INODE_POOL(remover->inode);
remover->min_offset = cfg["min_offset"].uint64_value(); remover->min_offset = cfg["min_offset"].uint64_value();
remover->max_offset = cfg["max_offset"].uint64_value();
return [remover](cli_result_t & result) return [remover](cli_result_t & result)
{ {
remover->loop(); remover->loop();
src/nfs/CMakeLists.txt
+1
View File
@@ -8,6 +8,7 @@ add_executable(vitastor-nfs
nfs_block.cpp nfs_block.cpp
nfs_kv.cpp nfs_kv.cpp
nfs_kv_create.cpp nfs_kv_create.cpp
nfs_kv_defrag.cpp
nfs_kv_getattr.cpp nfs_kv_getattr.cpp
nfs_kv_link.cpp nfs_kv_link.cpp
nfs_kv_lookup.cpp nfs_kv_lookup.cpp
src/nfs/nfs_kv.cpp
+95 -15
View File
@@ -116,17 +116,37 @@ std::string kv_direntry_filename(const std::string & key)
return key; return key;
} }
std::string kv_inode_key(uint64_t ino) std::string kv_inode_prefix_key(uint64_t ino, const char *prefix)
{ {
char key[32] = { 0 }; int max = 32+strlen(prefix);
snprintf(key, sizeof(key), "i%x", INODE_POOL(ino)); char key[max] = { 0 };
int n = strnlen(key, sizeof(key)-1); snprintf(key, max, "%s%x", prefix, INODE_POOL(ino));
snprintf(key+n+1, sizeof(key)-n-1, "%jx", INODE_NO_POOL(ino)); int n = strnlen(key, max-1);
int m = strnlen(key+n+1, sizeof(key)-n-2); snprintf(key+n+1, max-n-1, "%jx", INODE_NO_POOL(ino));
int m = strnlen(key+n+1, max-n-2);
key[n] = 'G'+m; key[n] = 'G'+m;
return std::string(key); return std::string(key);
} }
std::string kv_inode_key(uint64_t ino)
{
return kv_inode_prefix_key(ino, "i");
}
uint64_t kv_key_inode(const std::string & key, int prefix_len)
{
if (key.size() < prefix_len)
return 0;
uint32_t pool_id = 0;
char len_plus_g = 0;
uint64_t inode_id = 0;
char null_byte = 0;
int scanned = sscanf(key.c_str()+prefix_len, "%x%c%jx%c", &pool_id, &len_plus_g, &inode_id, &null_byte);
if (scanned != 3 || !inode_id || INODE_POOL(inode_id) != 0)
return 0;
return INODE_WITH_POOL(pool_id, inode_id);
}
std::string kv_fh(uint64_t ino) std::string kv_fh(uint64_t ino)
{ {
char key[32] = { 0 }; char key[32] = { 0 };
@@ -248,8 +268,36 @@ void kv_fs_state_t::init(nfs_proxy_t *proxy, json11::Json cfg)
if (!id_alloc_batch_size) if (!id_alloc_batch_size)
id_alloc_batch_size = 200; id_alloc_batch_size = 200;
touch_interval = cfg["touch_interval"].uint64_value(); touch_interval = cfg["touch_interval"].uint64_value();
if (touch_interval < 100) // ms if (!touch_interval)
touch_interval = 1000; // ms
else if (touch_interval < 100)
touch_interval = 100; touch_interval = 100;
volume_stats_interval_mul = cfg["volume_stats_interval"].uint64_value() / touch_interval;
if (!volume_stats_interval_mul)
volume_stats_interval_mul = 1;
volume_touch_interval_mul = cfg["volume_touch_interval"].uint64_value() / touch_interval;
if (!volume_touch_interval_mul)
volume_touch_interval_mul = 30;
volume_untouched_sec = cfg["volume_untouched"].uint64_value();
if (!volume_untouched_sec)
volume_untouched_sec = 86400;
if (volume_untouched_sec < 60)
volume_untouched_sec = 60;
defrag_percent = cfg["defrag_percent"].is_null() ? 50 : cfg["defrag_percent"].uint64_value();
if (defrag_percent < 0)
defrag_percent = 0;
if (defrag_percent > 100)
defrag_percent = 100;
defrag_block_count = cfg["defrag_block_count"].is_null() ? 16 : cfg["defrag_block_count"].uint64_value();
if (defrag_block_count < 1)
defrag_block_count = 1;
if (defrag_block_count > 1048576)
defrag_block_count = 1048576;
defrag_iodepth = cfg["defrag_iodepth"].is_null() ? 16 : cfg["defrag_iodepth"].uint64_value();
if (defrag_iodepth < 1)
defrag_iodepth = 1;
if (defrag_iodepth > 1048576)
defrag_iodepth = 1048576;
pool_block_size = pool_cfg.pg_stripe_size; pool_block_size = pool_cfg.pg_stripe_size;
pool_alignment = pool_cfg.bitmap_granularity; pool_alignment = pool_cfg.bitmap_granularity;
// Open DB and wait // Open DB and wait
@@ -261,6 +309,7 @@ void kv_fs_state_t::init(nfs_proxy_t *proxy, json11::Json cfg)
{ {
kv_cfg[kv.first] = kv.second.as_string(); kv_cfg[kv.first] = kv.second.as_string();
} }
// Open K/V DB
proxy->db->open(fs_kv_inode, kv_cfg, [&](int res) proxy->db->open(fs_kv_inode, kv_cfg, [&](int res)
{ {
open_done = true; open_done = true;
@@ -279,6 +328,7 @@ void kv_fs_state_t::init(nfs_proxy_t *proxy, json11::Json cfg)
strerror(-open_res), open_res); strerror(-open_res), open_res);
exit(1); exit(1);
} }
// Proceed
fs_inode_count = ((uint64_t)1 << (64-POOL_ID_BITS)) - 1; fs_inode_count = ((uint64_t)1 << (64-POOL_ID_BITS)) - 1;
shared_inode_threshold = pool_block_size; shared_inode_threshold = pool_block_size;
if (!cfg["shared_inode_threshold"].is_null()) if (!cfg["shared_inode_threshold"].is_null())
@@ -299,31 +349,36 @@ kv_fs_state_t::~kv_fs_state_t()
} }
} }
static void touch_inode(nfs_proxy_t *proxy, inode_t ino, bool allow_cache) void kv_fs_state_t::update_inode(inode_t ino, bool allow_cache, std::function<void(json11::Json::object &)> change, std::function<void(int)> cb)
{ {
kv_read_inode(proxy, ino, [proxy, ino](int res, const std::string & value, json11::Json attrs) // FIXME: Use "update" query
kv_read_inode(proxy, ino, [=](int res, const std::string & value, json11::Json attrs)
{ {
if (!res) if (!res)
{ {
auto ientry = attrs.object_items(); auto ientry = attrs.object_items();
ientry["mtime"] = ientry["ctime"] = nfstime_now_str(); change(ientry);
ientry.erase("verf");
// FIXME: Use "update" query
bool *found = new bool; bool *found = new bool;
*found = true; *found = true;
proxy->db->set(kv_inode_key(ino), json11::Json(ientry).dump(), [proxy, ino, found](int res) proxy->db->set(kv_inode_key(ino), json11::Json(ientry).dump(), [=](int res)
{ {
if (!*found) if (!*found)
res = -ENOENT; res = -ENOENT;
delete found; delete found;
if (res == -EAGAIN) if (res == -EAGAIN)
touch_inode(proxy, ino, false); update_inode(ino, false, change, cb);
else if (cb)
cb(res);
}, [value, found](int res, const std::string & old_value) }, [value, found](int res, const std::string & old_value)
{ {
*found = res == 0; *found = res == 0;
return res == 0 && old_value == value; return res == 0 && old_value == value;
}); });
} }
else if (cb)
{
cb(res);
}
}, allow_cache); }, allow_cache);
} }
@@ -332,6 +387,31 @@ void kv_fs_state_t::touch_inodes()
std::set<inode_t> q = std::move(touch_queue); std::set<inode_t> q = std::move(touch_queue);
for (auto ino: q) for (auto ino: q)
{ {
touch_inode(proxy, ino, true); update_inode(ino, true, [](json11::Json::object & ientry)
{
ientry["mtime"] = ientry["ctime"] = nfstime_now_str();
ientry.erase("verf");
}, NULL);
}
if (++volume_stats_ctr >= volume_stats_interval_mul)
{
volume_stats_ctr = 0;
auto shr = std::move(volume_removed);
for (auto & sp: shr)
{
update_inode(sp.first, true, [removed = sp.second](json11::Json::object & ientry)
{
ientry["removed"] = ientry["removed"].uint64_value() + removed;
}, NULL);
}
}
if (!((volume_touch_ctr++) % volume_touch_interval_mul) && cur_shared_inode)
{
volume_touch_ctr = 1;
update_inode(cur_shared_inode, true, [size = cur_shared_offset](json11::Json::object & ientry)
{
ientry["opentime"] = nfstime_now_str();
ientry["size"] = size;
}, NULL);
} }
} }
src/nfs/nfs_kv.h
+19 -1
View File
@@ -60,6 +60,13 @@ struct kv_fs_state_t
uint64_t pool_alignment = 0; uint64_t pool_alignment = 0;
uint64_t shared_inode_threshold = 0; uint64_t shared_inode_threshold = 0;
uint64_t touch_interval = 1000; uint64_t touch_interval = 1000;
uint64_t volume_stats_interval_mul = 1;
uint64_t volume_touch_interval_mul = 30;
uint64_t volume_untouched_sec = 86400;
uint64_t defrag_percent = 50;
uint64_t defrag_block_count = 16;
uint64_t defrag_iodepth = 16;
bool dry_run = false;
std::map<list_cookie_t, list_cookie_val_t> list_cookies; std::map<list_cookie_t, list_cookie_val_t> list_cookies;
std::map<pool_id_t, kv_idgen_t> idgen; std::map<pool_id_t, kv_idgen_t> idgen;
@@ -67,12 +74,19 @@ struct kv_fs_state_t
uint64_t cur_shared_inode = 0, cur_shared_offset = 0; uint64_t cur_shared_inode = 0, cur_shared_offset = 0;
std::map<inode_t, kv_inode_extend_t> extends; std::map<inode_t, kv_inode_extend_t> extends;
std::set<inode_t> touch_queue; std::set<inode_t> touch_queue;
std::map<inode_t, uint64_t> volume_removed;
uint64_t volume_stats_ctr = 0;
uint64_t volume_touch_ctr = 0;
std::vector<uint8_t> zero_block; std::vector<uint8_t> zero_block;
std::vector<uint8_t> scrap_block; std::vector<uint8_t> scrap_block;
void init(nfs_proxy_t *proxy, json11::Json cfg); void init(nfs_proxy_t *proxy, json11::Json cfg);
void touch_inodes(); void touch_inodes();
void update_inode(inode_t ino, bool allow_cache, std::function<void(json11::Json::object &)> change, std::function<void(int)> cb);
void upgrade_db(std::function<void(int)> cb);
void defrag_all(json11::Json cfg, std::function<void(int)> cb);
void defrag_volume(inode_t ino, bool no_rm, bool dry_run, std::function<void(int, uint64_t, uint64_t, uint64_t)> cb);
~kv_fs_state_t(); ~kv_fs_state_t();
}; };
@@ -105,16 +119,20 @@ int kv_map_type(const std::string & type);
fattr3 get_kv_attributes(nfs_client_t *self, uint64_t ino, json11::Json attrs); fattr3 get_kv_attributes(nfs_client_t *self, uint64_t ino, json11::Json attrs);
std::string kv_direntry_key(uint64_t dir_ino, const std::string & filename); std::string kv_direntry_key(uint64_t dir_ino, const std::string & filename);
std::string kv_direntry_filename(const std::string & key); std::string kv_direntry_filename(const std::string & key);
std::string kv_inode_prefix_key(uint64_t ino, const char *prefix);
std::string kv_inode_key(uint64_t ino); std::string kv_inode_key(uint64_t ino);
uint64_t kv_key_inode(const std::string & key, int prefix_len = 1);
std::string kv_fh(uint64_t ino); std::string kv_fh(uint64_t ino);
uint64_t kv_fh_inode(const std::string & fh); uint64_t kv_fh_inode(const std::string & fh);
bool kv_fh_valid(const std::string & fh); bool kv_fh_valid(const std::string & fh);
void allocate_new_id(nfs_client_t *self, pool_id_t pool_id, std::function<void(int res, uint64_t new_id)> cb); void allocate_new_id(nfs_proxy_t *proxy, pool_id_t pool_id, std::function<void(int res, uint64_t new_id)> cb);
void kv_read_inode(nfs_proxy_t *proxy, uint64_t ino, void kv_read_inode(nfs_proxy_t *proxy, uint64_t ino,
std::function<void(int res, const std::string & value, json11::Json ientry)> cb, std::function<void(int res, const std::string & value, json11::Json ientry)> cb,
bool allow_cache = false); bool allow_cache = false);
uint64_t align_shared_size(nfs_client_t *self, uint64_t size); uint64_t align_shared_size(nfs_client_t *self, uint64_t size);
void nfs_do_rmw(nfs_rmw_t *rmw); void nfs_do_rmw(nfs_rmw_t *rmw);
void nfs_move_inode_from(nfs_proxy_t *proxy, uint64_t ino, uint64_t shared_ino,
uint64_t shared_offset, std::function<void(int res, bool moved)> cb);
int kv_nfs3_getattr_proc(void *opaque, rpc_op_t *rop); int kv_nfs3_getattr_proc(void *opaque, rpc_op_t *rop);
int kv_nfs3_setattr_proc(void *opaque, rpc_op_t *rop); int kv_nfs3_setattr_proc(void *opaque, rpc_op_t *rop);
src/nfs/nfs_kv_create.cpp
+15 -11
View File
@@ -9,9 +9,9 @@
#include "nfs_proxy.h" #include "nfs_proxy.h"
#include "nfs_kv.h" #include "nfs_kv.h"
void allocate_new_id(nfs_client_t *self, pool_id_t pool_id, std::function<void(int res, uint64_t new_id)> cb) void allocate_new_id(nfs_proxy_t *proxy, pool_id_t pool_id, std::function<void(int res, uint64_t new_id)> cb)
{ {
auto & idgen = self->parent->kvfs->idgen[pool_id]; auto & idgen = proxy->kvfs->idgen[pool_id];
if (idgen.unallocated_ids.size()) if (idgen.unallocated_ids.size())
{ {
auto new_id = idgen.unallocated_ids.back(); auto new_id = idgen.unallocated_ids.back();
@@ -31,9 +31,9 @@ void allocate_new_id(nfs_client_t *self, pool_id_t pool_id, std::function<void(i
cb(-ENOSPC, 0); cb(-ENOSPC, 0);
return; return;
} }
self->parent->db->get((pool_id ? "id"+std::to_string(pool_id) : "id"), [=](int res, const std::string & prev_str) proxy->db->get((pool_id ? "id"+std::to_string(pool_id) : "id"), [=](int res, const std::string & prev_str)
{ {
auto & idgen = self->parent->kvfs->idgen[pool_id]; auto & idgen = proxy->kvfs->idgen[pool_id];
if (res < 0 && res != -ENOENT) if (res < 0 && res != -ENOENT)
{ {
cb(res, 0); cb(res, 0);
@@ -49,17 +49,21 @@ void allocate_new_id(nfs_client_t *self, pool_id_t pool_id, std::function<void(i
{ {
prev_val = idgen.min_id; prev_val = idgen.min_id;
} }
uint64_t new_val = prev_val + self->parent->kvfs->id_alloc_batch_size; uint64_t new_val = prev_val + proxy->kvfs->id_alloc_batch_size;
if (new_val >= self->parent->kvfs->fs_inode_count) if (new_val >= proxy->kvfs->fs_inode_count)
{ {
new_val = self->parent->kvfs->fs_inode_count; new_val = proxy->kvfs->fs_inode_count;
} }
self->parent->db->set((pool_id ? "id"+std::to_string(pool_id) : "id"), std::to_string(new_val), [=](int res) if (!pool_id && res == -ENOENT)
{
proxy->db->set("version", "1", [](int){});
}
proxy->db->set((pool_id ? "id"+std::to_string(pool_id) : "id"), std::to_string(new_val), [=](int res)
{ {
if (res == -EAGAIN) if (res == -EAGAIN)
{ {
// CAS failure - retry // CAS failure - retry
allocate_new_id(self, pool_id, cb); allocate_new_id(proxy, pool_id, cb);
} }
else if (res < 0) else if (res < 0)
{ {
@@ -67,7 +71,7 @@ void allocate_new_id(nfs_client_t *self, pool_id_t pool_id, std::function<void(i
} }
else else
{ {
auto & idgen = self->parent->kvfs->idgen[pool_id]; auto & idgen = proxy->kvfs->idgen[pool_id];
idgen.next_id = prev_val+2; idgen.next_id = prev_val+2;
idgen.allocated_id = new_val; idgen.allocated_id = new_val;
cb(0, INODE_WITH_POOL(pool_id, prev_val+1)); cb(0, INODE_WITH_POOL(pool_id, prev_val+1));
@@ -125,7 +129,7 @@ resume_1:
st->pool_id = kv_map_type(st->attrs["type"].string_value()) == NF3REG st->pool_id = kv_map_type(st->attrs["type"].string_value()) == NF3REG
? st->self->parent->default_pool_id ? st->self->parent->default_pool_id
: 0; : 0;
allocate_new_id(st->self, st->pool_id, [st](int res, uint64_t new_id) allocate_new_id(st->self->parent, st->pool_id, [st](int res, uint64_t new_id)
{ {
st->res = res; st->res = res;
st->new_id = new_id; st->new_id = new_id;
src/nfs/nfs_kv_defrag.cpp
+528
View File
@@ -0,0 +1,528 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
//
// NFS proxy over VitastorKV database - defragmentation
#include <sys/time.h>
#include "nfs_proxy.h"
#include "nfs_common.h"
#include "nfs_kv.h"
#include "str_util.h"
#include "cli.h"
struct kv_fs_defrag_t
{
nfs_proxy_t *proxy = NULL;
inode_t shared_ino = 0;
bool dry_run = false;
bool no_rm = false;
bool progress = true;
uint64_t bitmap_granularity = 0;
uint64_t buf_size = 0;
uint8_t *block_buf = NULL;
timespec prev_progress = {};
int errcode = 0;
bool reading = false;
bool empty = false;
uint64_t last_offset = 0;
uint64_t real_size = 0;
uint64_t buf_pos = 0;
uint64_t iodepth = 0;
uint64_t num_moved = 0, num_unused = 0;
uint64_t bytes_moved = 0, bytes_unused = 0;
uint64_t max_ctime = 0;
bool handling = false;
std::function<void(int, uint64_t, uint64_t, uint64_t)> cb;
void read();
void handle_read();
void finish(int retval);
};
void kv_fs_defrag_t::finish(int retval)
{
auto cb = std::move(this->cb);
delete block_buf;
block_buf = NULL;
cb(retval, real_size, bytes_unused, max_ctime);
delete this;
}
void kv_fs_defrag_t::read()
{
if (errcode)
{
finish(errcode);
return;
}
auto op = new cluster_op_t;
op->opcode = OSD_OP_READ;
op->inode = shared_ino;
op->offset = last_offset;
op->len = buf_size;
op->iov.push_back(block_buf, buf_size);
reading = true;
op->callback = [this](cluster_op_t *op)
{
reading = false;
if (op->retval != op->len)
{
fprintf(stderr, "Error reading 0x%jx bytes from volume 0x%jx at 0x%jx: %s (code %d)\n",
op->len, shared_ino, op->offset, strerror(-op->retval), op->retval);
finish(op->retval >= 0 ? -EIO : op->retval);
}
else
{
// Check that any data was actually read or it's the last iteration
uint64_t bitmap_size = (op->len / bitmap_granularity + 7) / 8;
uint64_t bitmap_pos = 0;
empty = true;
for (; bitmap_pos < bitmap_size; bitmap_pos += 8)
{
if (*((uint64_t*)(op->bitmap_buf + bitmap_pos)))
empty = false;
}
for (; bitmap_pos < bitmap_size; bitmap_pos++)
{
if (*((uint8_t*)(op->bitmap_buf + bitmap_pos)))
empty = false;
}
buf_pos = 0;
handle_read();
}
delete op;
};
proxy->cli->execute(op);
}
void kv_fs_defrag_t::handle_read()
{
if (handling)
{
return;
}
handling = true;
while (!empty && !errcode && buf_pos < buf_size && iodepth < proxy->kvfs->defrag_iodepth)
{
// Next object header may be at any position after any number of zeroes
// Commonly it's either in the beginning of a 4 KB sector or in the end of it
if ((*(uint64_t*)(block_buf+buf_pos)) == SHARED_FILE_MAGIC_V1)
{
iodepth++;
shared_file_header_t *hdr = (shared_file_header_t*)(block_buf+buf_pos);
uint64_t shared_offset = last_offset + buf_pos;
buf_pos += hdr->alloc;
real_size = shared_offset + hdr->alloc;
auto move_cb = [this, ino = hdr->inode, alloc = hdr->alloc, shared_offset](int res, bool was_moved)
{
if (res < 0 && res != -ENOENT)
{
fprintf(stderr, "Error checking/moving inode 0x%jx from volume 0x%jx offset 0x%jx: %s (code %d)\n",
ino, shared_ino, shared_offset, strerror(-res), res);
errcode = res;
}
else
{
if (was_moved)
{
bytes_moved += alloc;
num_moved++;
}
else
{
bytes_unused += alloc;
num_unused++;
}
if (proxy->trace)
{
fprintf(
stderr, was_moved
? "Moved inode 0x%jx (%ju bytes) from volume 0x%jx offset 0x%ju\n"
: "Inode 0x%jx (%ju bytes) data in volume 0x%jx at offset 0x%ju is unused",
ino, alloc, shared_ino, shared_offset
);
}
else if (progress)
{
timespec now;
clock_gettime(CLOCK_REALTIME, &now);
if (now.tv_sec >= prev_progress.tv_sec+2)
{
prev_progress = now;
fprintf(stderr, "Processed %s, %s %s, unused %s\n", format_size(real_size).c_str(),
dry_run ? "in use" : "moved", format_size(bytes_moved).c_str(), format_size(bytes_unused).c_str());
}
}
}
iodepth--;
handle_read();
};
if (dry_run)
{
kv_read_inode(proxy, hdr->inode, [=](int res, const std::string & value, json11::Json attrs)
{
uint64_t ctime = (uint64_t)attrs["ctime"].number_value();
if (max_ctime < ctime)
max_ctime = ctime;
move_cb(res, !res && attrs["shared_ino"] == shared_ino && attrs["shared_offset"] == shared_offset);
});
}
else
{
nfs_move_inode_from(proxy, hdr->inode, shared_ino, shared_offset, move_cb);
}
}
else
{
buf_pos += 8;
}
}
handling = false;
if (errcode)
{
if (iodepth)
{
// Wait for completion
return;
}
finish(errcode);
}
else if (empty)
{
if (iodepth)
{
// Wait for completion
return;
}
// Finish - now we can purge shared inode
fprintf(
stderr, dry_run
? "Estimated volume 0x%jx - in use %s (%ju files), unused %s (%ju files), last inode change time %s\n"
: "Defragmented volume 0x%jx - moved %s (%ju files), unused %s (%ju files), last inode change time %s. Purging volume data\n",
shared_ino, format_size(bytes_moved).c_str(), num_moved, format_size(bytes_unused).c_str(), num_unused, format_datetime(max_ctime).c_str()
);
if (dry_run || no_rm)
{
finish(0);
}
else
{
proxy->cmd->loop_and_wait(proxy->cmd->start_rm_data(json11::Json::object {
{ "inode", INODE_NO_POOL(shared_ino) },
{ "pool", (uint64_t)INODE_POOL(shared_ino) },
{ "progress", (uint64_t)proxy->trace }
}), [this](const cli_result_t & r)
{
if (r.err)
{
fprintf(stderr, "Failed to remove volume 0x%jx data: %s (code %d)\n",
shared_ino, r.text.c_str(), r.err);
finish(r.err);
}
else
{
proxy->db->del(kv_inode_key(shared_ino), [=](int res)
{
if (res < 0)
{
fprintf(stderr, "Failed to remove volume key %s: %s (code %d)\n",
kv_inode_key(shared_ino).c_str(), strerror(-res), res);
finish(res);
}
else
{
proxy->db->del(kv_inode_prefix_key(shared_ino, "shared"), [=](int res)
{
if (res < 0)
{
fprintf(stderr, "Failed to remove volume key %s: %s (code %d)\n",
kv_inode_prefix_key(shared_ino, "shared").c_str(), strerror(-res), res);
}
finish(res);
});
}
});
}
});
}
}
else if (!reading && buf_pos >= buf_size)
{
last_offset += buf_pos;
read();
}
}
// Linear read all object headers, check which of them are still alive, move them away
void kv_fs_state_t::defrag_volume(inode_t ino, bool no_rm, bool dry_run, std::function<void(int, uint64_t, uint64_t, uint64_t)> cb)
{
auto pool_it = proxy->cli->st_cli.pool_config.find(INODE_POOL(ino));
if (pool_it == proxy->cli->st_cli.pool_config.end())
{
fprintf(stderr, "Volume 0x%jx references a non-existing pool with ID %u, skipping\n", ino, INODE_POOL(ino));
cb(0, 0, 0, 0);
return;
}
auto st = new kv_fs_defrag_t;
st->proxy = proxy;
st->shared_ino = ino;
st->dry_run = dry_run;
st->no_rm = no_rm;
st->buf_size = pool_it->second.pg_stripe_size * defrag_block_count;
st->block_buf = (uint8_t*)malloc_or_die(st->buf_size);
st->bitmap_granularity = pool_it->second.bitmap_granularity;
st->cb = cb;
clock_gettime(CLOCK_REALTIME, &st->prev_progress);
st->read();
}
struct kv_fs_defrag_all_t
{
std::function<void(int)> cb;
nfs_proxy_t *proxy = NULL;
bool dry_run = false;
bool no_rm = false;
bool recalc_stats = false;
bool include_empty = false;
timespec now = {};
void *list_shared = NULL;
uint64_t ino = 0;
json11::Json ientry;
bool recalc = false;
uint64_t real_size = 0;
uint64_t removed_size = 0;
uint64_t opentime = 0;
int res = 0;
void run(int);
};
void kv_fs_defrag_all_t::run(int st)
{
if (st == 1)
goto resume_1;
else if (st == 2)
goto resume_2;
else if (st == 3)
goto resume_3;
else if (st == 4)
goto resume_4;
else if (st == 5)
goto resume_5;
else if (st == 6)
goto resume_6;
clock_gettime(CLOCK_REALTIME, &now);
list_shared = proxy->db->list_start("shared");
proxy->db->list_next(list_shared, [this](int res, const std::string & key, const std::string & value)
{
if (res == -ENOENT || key.substr(0, 6) != "shared")
this->res = -ENOENT;
else
{
this->res = res;
this->ino = kv_key_inode(key, 6);
}
run(1);
});
return;
while (true)
{
resume_1:
if (res < 0)
{
if (res == -ENOENT)
res = 0;
break;
}
kv_read_inode(proxy, ino, [this](int res, const std::string & value, json11::Json attrs)
{
this->res = res;
this->ientry = attrs;
run(2);
});
return;
resume_2:
if (res == -ENOENT)
{
// This shared inode is already removed
proxy->db->del(kv_inode_prefix_key(ino, "shared"), [this](int res)
{
run(3);
});
return;
resume_3:
proxy->db->list_next(list_shared, NULL);
return;
}
real_size = ientry["size"].uint64_value();
removed_size = ientry["removed"].uint64_value();
opentime = (uint64_t)ientry["opentime"].number_value();
recalc = false;
if (!real_size && !opentime || recalc_stats)
{
// Statistics are missing - recalculate statistics
recalc = true;
fprintf(stderr, "Shared volume 0x%jx misses size and removal statistics, recalculating\n", ino);
proxy->kvfs->defrag_volume(ino, true, true, [this](int res, uint64_t sz, uint64_t rm, uint64_t tm)
{
this->res = res;
this->real_size = sz;
this->removed_size = rm;
this->opentime = tm;
run(4);
});
return;
resume_4:
if (res < 0)
{
break;
}
proxy->kvfs->update_inode(ino, true, [this](json11::Json::object & ientry)
{
ientry["size"] = real_size;
ientry["removed"] = removed_size;
ientry["opentime"] = opentime;
}, [this](int res)
{
this->res = res;
run(5);
});
return;
resume_5:
if (res < 0)
{
fprintf(stderr, "Warning: Failed to update shared volume 0x%jx metadata: %s (code %d)\n", ino, strerror(-res), res);
}
}
if ((opentime && opentime < now.tv_sec - proxy->kvfs->volume_untouched_sec || !opentime && include_empty) &&
(real_size && removed_size || include_empty) &&
removed_size >= (real_size * proxy->kvfs->defrag_percent / 100))
{
// This volume needs defrag
fprintf(
stderr, "Shared volume 0x%jx requires defragmentation: last "
"open-for-append time %s, size %s, removed %s\n",
ino, format_datetime(opentime).c_str(), format_size(real_size).c_str(), format_size(removed_size).c_str()
);
if (!recalc || !dry_run)
{
proxy->kvfs->defrag_volume(ino, no_rm, dry_run, [this](int res, uint64_t, uint64_t, uint64_t)
{
this->res = res;
run(6);
});
return;
resume_6:
if (res < 0)
{
break;
}
}
}
else
{
fprintf(
stderr, "Shared volume 0x%jx does not require defragmentation: last "
"open-for-append time %s, size %s, removed %s\n",
ino, format_datetime(opentime).c_str(), format_size(real_size).c_str(), format_size(removed_size).c_str()
);
}
proxy->db->list_next(list_shared, NULL);
return;
}
proxy->db->list_close(list_shared);
auto cb = std::move(this->cb);
cb(res);
delete this;
}
void kv_fs_state_t::defrag_all(json11::Json cfg, std::function<void(int)> cb)
{
auto st = new kv_fs_defrag_all_t;
st->cb = cb;
st->proxy = proxy;
st->dry_run = cfg["dry_run"].bool_value();
st->no_rm = cfg["no_rm"].bool_value();
st->recalc_stats = cfg["recalc_stats"].bool_value();
st->include_empty = cfg["include_empty"].bool_value();
st->run(0);
}
void kv_fs_state_t::upgrade_db(std::function<void(int)> cb)
{
// In the future, FS metadata format upgrades should be added here
// Currently we only do one thing: we create missing shared inode list keys ("sharedXXX")
proxy->db->get("version", [=](int res, const std::string & ver_value)
{
if (res < 0 && res != -ENOENT)
{
cb(res);
return;
}
json11::Json ver;
if (res == 0)
{
std::string err;
ver = json11::Json::parse(ver_value, err);
if (err != "")
{
fprintf(stderr, "Invalid JSON in `version` key, value: %s, error: %s\n", ver_value.c_str(), err.c_str());
cb(-EINVAL);
return;
}
}
if (ver.uint64_value() > 1 || ver.is_object())
{
cb(0);
return;
}
// Create missing shared inode index keys
auto list_inodes = proxy->db->list_start("i");
proxy->db->list_next(list_inodes, [=](int res, const std::string & key, const std::string & value)
{
if (res == -ENOENT || key.substr(0, 1) != "i" || key == "id")
{
proxy->db->list_close(list_inodes);
proxy->db->set("version", "1", [=](int res)
{
cb(0);
}, [=](int res, const std::string & value)
{
return res == -ENOENT || ver_value == value;
});
return;
}
uint64_t inode_id = kv_key_inode(key, 1);
if (!inode_id)
{
fprintf(stderr, "Invalid inode key %s, skipping\n", key.c_str());
}
else
{
std::string err;
auto ientry = json11::Json::parse(value, err);
if (err != "")
{
fprintf(stderr, "Invalid JSON in key %s (inode %ju), skipping\n", key.c_str(), inode_id);
}
else if (ientry["type"] == "shared")
{
proxy->db->set(kv_inode_prefix_key(inode_id, "shared"), "{}", [=](int res)
{
if (res < 0)
{
fprintf(stderr, "Error writing key %s: %s (code %d)\n",
kv_inode_prefix_key(inode_id, "shared").c_str(), strerror(-res), res);
}
proxy->db->list_next(list_inodes, NULL);
});
return;
}
}
proxy->db->list_next(list_inodes, NULL);
});
});
}
src/nfs/nfs_kv_remove.cpp
+25 -18
View File
@@ -228,28 +228,35 @@ resume_6:
return; return;
} }
// (6) If regular file and inode is deleted: delete data // (6) If regular file and inode is deleted: delete data
if ((!st->type || st->type == NF3REG) && st->ientry["nlink"].uint64_value() <= 1 && if ((!st->type || st->type == NF3REG) && st->ientry["nlink"].uint64_value() <= 1)
!st->ientry["shared_ino"].uint64_value())
{ {
// Remove data if (!st->ientry["shared_ino"].uint64_value())
st->self->parent->cmd->loop_and_wait(st->self->parent->cmd->start_rm_data(json11::Json::object {
{ "inode", INODE_NO_POOL(st->ino) },
{ "pool", (uint64_t)INODE_POOL(st->ino) },
}), [st](const cli_result_t & r)
{ {
if (r.err) // Remove data
st->self->parent->cmd->loop_and_wait(st->self->parent->cmd->start_rm_data(json11::Json::object {
{ "inode", INODE_NO_POOL(st->ino) },
{ "pool", (uint64_t)INODE_POOL(st->ino) },
}), [st](const cli_result_t & r)
{ {
fprintf(stderr, "Failed to remove inode %jx data: %s (code %d)\n", if (r.err)
st->ino, r.text.c_str(), r.err); {
} fprintf(stderr, "Failed to remove inode %jx data: %s (code %d)\n",
st->res = r.err; st->ino, r.text.c_str(), r.err);
nfs_kv_continue_delete(st, 7); }
}); st->res = r.err;
return; nfs_kv_continue_delete(st, 7);
});
return;
resume_7: resume_7:
auto cb = std::move(st->cb); auto cb = std::move(st->cb);
cb(st->res); cb(st->res);
return; return;
}
else
{
// Record removed part of the shared inode as obsolete in statistics
st->self->parent->kvfs->volume_removed[st->ientry["shared_ino"].uint64_value()] += st->ientry["shared_alloc"].uint64_value();
}
} }
if (!st->res) if (!st->res)
{ {
src/nfs/nfs_kv_write.cpp
+110 -56
View File
@@ -8,17 +8,18 @@
#include "nfs_proxy.h" #include "nfs_proxy.h"
#include "nfs_kv.h" #include "nfs_kv.h"
// FIXME: Implement shared inode defragmentator
// FIXME: Implement fsck for vitastor-fs and for vitastor-kv // FIXME: Implement fsck for vitastor-fs and for vitastor-kv
struct nfs_kv_write_state struct nfs_kv_write_state
{ {
nfs_client_t *self = NULL; nfs_proxy_t *proxy = NULL;
rpc_op_t *rop = NULL; rpc_op_t *rop = NULL;
uint64_t ino = 0; uint64_t ino = 0;
uint64_t offset = 0, size = 0; uint64_t offset = 0, size = 0;
bool stable = false; bool stable = false;
uint8_t *buf = NULL; uint8_t *buf = NULL;
uint64_t force_move_from_inode = 0; // force move from this shared inode ID
uint64_t force_move_from_offset = 0;
std::function<void(int res)> cb; std::function<void(int res)> cb;
// state // state
bool allow_cache = true; bool allow_cache = true;
@@ -46,14 +47,14 @@ struct nfs_kv_write_state
} }
}; };
#define align_down(size) ((size) & ~(st->self->parent->kvfs->pool_alignment-1)) #define align_down(size) ((size) & ~(st->proxy->kvfs->pool_alignment-1))
#define align_up(size) (((size) + st->self->parent->kvfs->pool_alignment-1) & ~(st->self->parent->kvfs->pool_alignment-1)) #define align_up(size) (((size) + st->proxy->kvfs->pool_alignment-1) & ~(st->proxy->kvfs->pool_alignment-1))
static void nfs_kv_continue_write(nfs_kv_write_state *st, int state); static void nfs_kv_continue_write(nfs_kv_write_state *st, int state);
static void allocate_shared_space(nfs_kv_write_state *st) static void allocate_shared_space(nfs_kv_write_state *st)
{ {
auto kvfs = st->self->parent->kvfs; auto kvfs = st->proxy->kvfs;
st->shared_inode = kvfs->cur_shared_inode; st->shared_inode = kvfs->cur_shared_inode;
if (st->new_size < 3*kvfs->pool_alignment - sizeof(shared_file_header_t)) if (st->new_size < 3*kvfs->pool_alignment - sizeof(shared_file_header_t))
{ {
@@ -67,13 +68,13 @@ static void allocate_shared_space(nfs_kv_write_state *st)
st->shared_offset = align_up(kvfs->cur_shared_offset + sizeof(shared_file_header_t)) - sizeof(shared_file_header_t); st->shared_offset = align_up(kvfs->cur_shared_offset + sizeof(shared_file_header_t)) - sizeof(shared_file_header_t);
st->shared_alloc = sizeof(shared_file_header_t) + align_up(st->new_size); st->shared_alloc = sizeof(shared_file_header_t) + align_up(st->new_size);
} }
st->self->parent->kvfs->cur_shared_offset = st->shared_offset + st->shared_alloc; st->proxy->kvfs->cur_shared_offset = st->shared_offset + st->shared_alloc;
} }
static void finish_allocate_shared(nfs_client_t *self, int res) static void finish_allocate_shared(nfs_proxy_t *proxy, int res)
{ {
std::vector<shared_alloc_queue_t> waiting; std::vector<shared_alloc_queue_t> waiting;
waiting.swap(self->parent->kvfs->allocating_shared); waiting.swap(proxy->kvfs->allocating_shared);
for (auto & w: waiting) for (auto & w: waiting)
{ {
auto st = w.st; auto st = w.st;
@@ -88,31 +89,44 @@ static void finish_allocate_shared(nfs_client_t *self, int res)
static void allocate_shared_inode(nfs_kv_write_state *st, int state) static void allocate_shared_inode(nfs_kv_write_state *st, int state)
{ {
if (st->self->parent->kvfs->cur_shared_inode == 0) if (st->proxy->kvfs->cur_shared_inode == 0)
{ {
st->self->parent->kvfs->allocating_shared.push_back({ st, state }); st->proxy->kvfs->allocating_shared.push_back({ st, state });
if (st->self->parent->kvfs->allocating_shared.size() > 1) if (st->proxy->kvfs->allocating_shared.size() > 1)
{ {
return; return;
} }
allocate_new_id(st->self, st->self->parent->default_pool_id, [st](int res, uint64_t new_id) allocate_new_id(st->proxy, st->proxy->default_pool_id, [st](int res, uint64_t new_id)
{ {
if (res < 0) if (res < 0)
{ {
finish_allocate_shared(st->self, res); finish_allocate_shared(st->proxy, res);
return; return;
} }
st->self->parent->kvfs->cur_shared_inode = new_id; st->proxy->kvfs->cur_shared_inode = new_id;
st->self->parent->kvfs->cur_shared_offset = 0; st->proxy->kvfs->volume_touch_ctr = 0;
st->self->parent->db->set( st->proxy->kvfs->cur_shared_offset = 0;
st->proxy->db->set(
kv_inode_key(new_id), json11::Json(json11::Json::object{ { "type", "shared" } }).dump(), kv_inode_key(new_id), json11::Json(json11::Json::object{ { "type", "shared" } }).dump(),
[st](int res) [st](int res)
{ {
if (res < 0) if (res < 0)
{ {
st->self->parent->kvfs->cur_shared_inode = 0; st->proxy->kvfs->cur_shared_inode = 0;
finish_allocate_shared(st->proxy, res);
}
else
{
st->proxy->db->set(
kv_inode_prefix_key(st->proxy->kvfs->cur_shared_inode, "shared"),
"{}", [st](int res)
{
if (res < 0)
st->proxy->kvfs->cur_shared_inode = 0;
finish_allocate_shared(st->proxy, res);
}
);
} }
finish_allocate_shared(st->self, res);
}, },
[](int res, const std::string & old_value) [](int res, const std::string & old_value)
{ {
@@ -151,7 +165,7 @@ static void nfs_do_write(uint64_t ino, uint64_t offset, uint64_t size, std::func
nfs_kv_continue_write(st, state); nfs_kv_continue_write(st, state);
} }
}; };
st->self->parent->cli->execute(op); st->proxy->cli->execute(op);
} }
static void nfs_do_unshare_write(nfs_kv_write_state *st, int state) static void nfs_do_unshare_write(nfs_kv_write_state *st, int state)
@@ -162,7 +176,7 @@ static void nfs_do_unshare_write(nfs_kv_write_state *st, int state)
{ {
op->iov.push_back(st->aligned_buf, size); op->iov.push_back(st->aligned_buf, size);
if (aligned_size > size) if (aligned_size > size)
op->iov.push_back(st->self->parent->kvfs->zero_block.data(), aligned_size-size); op->iov.push_back(st->proxy->kvfs->zero_block.data(), aligned_size-size);
}, st, state); }, st, state);
} }
@@ -269,7 +283,7 @@ static void nfs_do_shared_read(nfs_kv_write_state *st, int state)
auto pre = shared_offset-align_down(shared_offset); auto pre = shared_offset-align_down(shared_offset);
if (pre > 0) if (pre > 0)
{ {
op->iov.push_back(st->self->parent->kvfs->scrap_block.data(), pre); op->iov.push_back(st->proxy->kvfs->scrap_block.data(), pre);
} }
op->iov.push_back(&st->shdr, sizeof(shared_file_header_t)); op->iov.push_back(&st->shdr, sizeof(shared_file_header_t));
op->iov.push_back(st->aligned_buf, data_size); op->iov.push_back(st->aligned_buf, data_size);
@@ -277,7 +291,7 @@ static void nfs_do_shared_read(nfs_kv_write_state *st, int state)
post = align_up(post) - post; post = align_up(post) - post;
if (post > 0) if (post > 0)
{ {
op->iov.push_back(st->self->parent->kvfs->scrap_block.data(), post); op->iov.push_back(st->proxy->kvfs->scrap_block.data(), post);
} }
op->len = pre+sizeof(shared_file_header_t)+data_size+post; op->len = pre+sizeof(shared_file_header_t)+data_size+post;
op->callback = [st, state](cluster_op_t *op) op->callback = [st, state](cluster_op_t *op)
@@ -302,7 +316,7 @@ static void nfs_do_shared_read(nfs_kv_write_state *st, int state)
nfs_kv_continue_write(st, state); nfs_kv_continue_write(st, state);
} }
}; };
st->self->parent->cli->execute(op); st->proxy->cli->execute(op);
} }
static void nfs_do_fsync(nfs_kv_write_state *st, int state) static void nfs_do_fsync(nfs_kv_write_state *st, int state)
@@ -315,10 +329,10 @@ static void nfs_do_fsync(nfs_kv_write_state *st, int state)
delete op; delete op;
nfs_kv_continue_write(st, state); nfs_kv_continue_write(st, state);
}; };
st->self->parent->cli->execute(op); st->proxy->cli->execute(op);
} }
static bool nfs_do_shared_readmodify(nfs_kv_write_state *st, int base_state, int state, bool unshare) static bool nfs_do_shared_readmodify(nfs_kv_write_state *st, int base_state, int state)
{ {
assert(state <= base_state); assert(state <= base_state);
if (state < base_state) goto resume_0; if (state < base_state) goto resume_0;
@@ -382,7 +396,7 @@ static void nfs_do_shared_write(nfs_kv_write_state *st, int state)
if (unaligned_is_free && aligned_offset < write_offset) if (unaligned_is_free && aligned_offset < write_offset)
{ {
// zero padding // zero padding
op->iov.push_back(st->self->parent->kvfs->zero_block.data(), write_offset-aligned_offset); op->iov.push_back(st->proxy->kvfs->zero_block.data(), write_offset-aligned_offset);
} }
// header // header
op->iov.push_back(&st->shdr, sizeof(shared_file_header_t)); op->iov.push_back(&st->shdr, sizeof(shared_file_header_t));
@@ -394,10 +408,13 @@ static void nfs_do_shared_write(nfs_kv_write_state *st, int state)
op->iov.push_back(st->aligned_buf, st->offset); op->iov.push_back(st->aligned_buf, st->offset);
} }
else else
add_zero(op, st->offset, st->self->parent->kvfs->zero_block); add_zero(op, st->offset, st->proxy->kvfs->zero_block);
} }
// new data // new data
op->iov.push_back(st->buf, st->size); if (st->size > 0)
{
op->iov.push_back(st->buf, st->size);
}
if (st->offset+st->size < st->new_size) if (st->offset+st->size < st->new_size)
{ {
if (has_old) if (has_old)
@@ -406,19 +423,19 @@ static void nfs_do_shared_write(nfs_kv_write_state *st, int state)
op->iov.push_back(st->aligned_buf+st->offset+st->size, st->new_size-(st->offset+st->size)); op->iov.push_back(st->aligned_buf+st->offset+st->size, st->new_size-(st->offset+st->size));
} }
else else
add_zero(op, st->offset, st->self->parent->kvfs->zero_block); add_zero(op, st->offset, st->proxy->kvfs->zero_block);
} }
if (unaligned_is_free && (aligned_size+aligned_offset) > (write_size+write_offset)) if (unaligned_is_free && (aligned_size+aligned_offset) > (write_size+write_offset))
{ {
// zero padding // zero padding
op->iov.push_back(st->self->parent->kvfs->zero_block.data(), aligned_size+aligned_offset - (write_size+write_offset)); op->iov.push_back(st->proxy->kvfs->zero_block.data(), aligned_size+aligned_offset - (write_size+write_offset));
} }
}, st, state); }, st, state);
} }
static void nfs_do_align_write(nfs_kv_write_state *st, uint64_t ino, uint64_t offset, uint64_t shared_alloc, int state) static void nfs_do_align_write(nfs_kv_write_state *st, uint64_t ino, uint64_t offset, uint64_t shared_alloc, int state)
{ {
auto alignment = st->self->parent->kvfs->pool_alignment; auto alignment = st->proxy->kvfs->pool_alignment;
uint64_t end = (offset+st->size); uint64_t end = (offset+st->size);
uint8_t *good_buf = st->buf; uint8_t *good_buf = st->buf;
uint64_t good_offset = offset; uint64_t good_offset = offset;
@@ -467,7 +484,7 @@ static void nfs_do_align_write(nfs_kv_write_state *st, uint64_t ino, uint64_t of
good_size = 0; good_size = 0;
s = s > st->size ? st->size : s; s = s > st->size ? st->size : s;
st->rmw[0] = (nfs_rmw_t){ st->rmw[0] = (nfs_rmw_t){
.parent = st->self->parent, .parent = st->proxy,
.ino = ino, .ino = ino,
.offset = offset, .offset = offset,
.buf = st->buf, .buf = st->buf,
@@ -496,7 +513,7 @@ static void nfs_do_align_write(nfs_kv_write_state *st, uint64_t ino, uint64_t of
else else
good_size = 0; good_size = 0;
st->rmw[1] = (nfs_rmw_t){ st->rmw[1] = (nfs_rmw_t){
.parent = st->self->parent, .parent = st->proxy,
.ino = ino, .ino = ino,
.offset = end - s, .offset = end - s,
.buf = st->buf + st->size - s, .buf = st->buf + st->size - s,
@@ -521,7 +538,7 @@ static void nfs_do_align_write(nfs_kv_write_state *st, uint64_t ino, uint64_t of
op->iov.push_back(&st->shdr, sizeof(shared_file_header_t)); op->iov.push_back(&st->shdr, sizeof(shared_file_header_t));
op->iov.push_back(good_buf, good_size); op->iov.push_back(good_buf, good_size);
if (end_pad) if (end_pad)
op->iov.push_back(st->self->parent->kvfs->zero_block.data(), end_pad); op->iov.push_back(st->proxy->kvfs->zero_block.data(), end_pad);
}, st, state); }, st, state);
} }
st->waiting--; st->waiting--;
@@ -590,7 +607,7 @@ static void nfs_kv_extend_inode(nfs_kv_write_state *st, int state, int base_stat
st->ext->cur_extend = st->ext->next_extend; st->ext->cur_extend = st->ext->next_extend;
st->ext->next_extend = 0; st->ext->next_extend = 0;
st->res2 = -EAGAIN; st->res2 = -EAGAIN;
st->self->parent->db->set(kv_inode_key(st->ino), new_normal_ientry(st), [st, base_state](int res) st->proxy->db->set(kv_inode_key(st->ino), new_normal_ientry(st), [st, base_state](int res)
{ {
st->res = res; st->res = res;
nfs_kv_continue_write(st, base_state+1); nfs_kv_continue_write(st, base_state+1);
@@ -738,13 +755,13 @@ static void nfs_kv_continue_write(nfs_kv_write_state *st, int state)
abort(); abort();
} }
resume_0: resume_0:
if (!st->size) if (!st->size && !st->force_move_from_inode)
{ {
auto cb = std::move(st->cb); auto cb = std::move(st->cb);
cb(0); cb(0);
return; return;
} }
kv_read_inode(st->self->parent, st->ino, [st](int res, const std::string & value, json11::Json attrs) kv_read_inode(st->proxy, st->ino, [st](int res, const std::string & value, json11::Json attrs)
{ {
st->res = res; st->res = res;
st->ientry_text = value; st->ientry_text = value;
@@ -759,22 +776,30 @@ resume_1:
cb(st->res == 0 ? -EINVAL : st->res); cb(st->res == 0 ? -EINVAL : st->res);
return; return;
} }
st->was_immediate = st->self->parent->cli->get_immediate_commit(st->ino); st->was_immediate = st->proxy->cli->get_immediate_commit(st->ino);
st->new_size = st->ientry["size"].uint64_value(); st->new_size = st->ientry["size"].uint64_value();
if (st->new_size < st->offset + st->size) if (st->new_size < st->offset + st->size)
{ {
st->new_size = st->offset + st->size; st->new_size = st->offset + st->size;
} }
if (st->offset + st->size + sizeof(shared_file_header_t) < st->self->parent->kvfs->shared_inode_threshold) if (st->offset + st->size + sizeof(shared_file_header_t) < st->proxy->kvfs->shared_inode_threshold)
{ {
if (st->ientry["size"].uint64_value() == 0 && if (// Zero size, should be allocated to handle write
st->ientry["shared_ino"].uint64_value() == 0 || st->ientry["size"].uint64_value() == 0 &&
st->ientry["shared_ino"].uint64_value() == 0 &&
st->offset+st->size > 0 ||
// Empty with non-zero size, fits shared inode threshold
st->ientry["empty"].bool_value() && st->ientry["empty"].bool_value() &&
(st->ientry["size"].uint64_value() + sizeof(shared_file_header_t)) < st->self->parent->kvfs->shared_inode_threshold || (st->ientry["size"].uint64_value() + sizeof(shared_file_header_t)) < st->proxy->kvfs->shared_inode_threshold ||
// Shared, does not fit currently allocated shared inode space
st->ientry["shared_ino"].uint64_value() != 0 && st->ientry["shared_ino"].uint64_value() != 0 &&
st->ientry["shared_alloc"].uint64_value() < sizeof(shared_file_header_t)+st->offset+st->size) st->ientry["shared_alloc"].uint64_value() < sizeof(shared_file_header_t)+st->offset+st->size ||
// Shared, requested to be moved away forcibly by defrag
st->force_move_from_inode != 0 &&
st->ientry["shared_ino"].uint64_value() == st->force_move_from_inode &&
st->ientry["shared_offset"].uint64_value() == st->force_move_from_offset)
{ {
// Either empty, or shared and requires moving into a larger place (redirect-write) // Inode requires moving into a larger place (redirect-write)
allocate_shared_inode(st, 2); allocate_shared_inode(st, 2);
return; return;
resume_2: resume_2:
@@ -785,7 +810,7 @@ resume_2:
return; return;
} }
resume_3: resume_3:
if (!nfs_do_shared_readmodify(st, 3, state, false)) if (!nfs_do_shared_readmodify(st, 3, state))
{ {
return; return;
} }
@@ -803,7 +828,7 @@ resume_4:
cb(st->res); cb(st->res);
return; return;
} }
st->self->parent->db->set(kv_inode_key(st->ino), new_moved_ientry(st), [st](int res) st->proxy->db->set(kv_inode_key(st->ino), new_moved_ientry(st), [st](int res)
{ {
st->res = res; st->res = res;
nfs_kv_continue_write(st, 5); nfs_kv_continue_write(st, 5);
@@ -831,7 +856,7 @@ resume_5:
cb(0); cb(0);
return; return;
} }
else if (st->ientry["shared_ino"].uint64_value() != 0) else if (st->ientry["shared_ino"].uint64_value() != 0 && st->size > 0)
{ {
// Non-empty, shared, can be updated in-place // Non-empty, shared, can be updated in-place
nfs_do_align_write(st, st->ientry["shared_ino"].uint64_value(), nfs_do_align_write(st, st->ientry["shared_ino"].uint64_value(),
@@ -846,7 +871,7 @@ resume_7:
} }
resume_8: resume_8:
// We always have to change inode entry on shared writes // We always have to change inode entry on shared writes
st->self->parent->db->set(kv_inode_key(st->ino), new_shared_ientry(st), [st](int res) st->proxy->db->set(kv_inode_key(st->ino), new_shared_ientry(st), [st](int res)
{ {
st->res = res; st->res = res;
nfs_kv_continue_write(st, 9); nfs_kv_continue_write(st, 9);
@@ -866,6 +891,12 @@ resume_9:
} }
// Fall through for non-shared // Fall through for non-shared
} }
if (!st->size)
{
auto cb = std::move(st->cb);
cb(0);
return;
}
// Unshare? // Unshare?
if (st->ientry["shared_ino"].uint64_value() != 0) if (st->ientry["shared_ino"].uint64_value() != 0)
{ {
@@ -889,7 +920,7 @@ resume_11:
return; return;
} }
} }
st->self->parent->db->set(kv_inode_key(st->ino), new_unshared_ientry(st), [st](int res) st->proxy->db->set(kv_inode_key(st->ino), new_unshared_ientry(st), [st](int res)
{ {
st->res = res; st->res = res;
nfs_kv_continue_write(st, 12); nfs_kv_continue_write(st, 12);
@@ -910,6 +941,8 @@ resume_12:
cb(st->res); cb(st->res);
return; return;
} }
// Record removed part of the shared inode as obsolete in statistics
st->proxy->kvfs->volume_removed[st->ientry["shared_ino"].uint64_value()] += st->ientry["shared_alloc"].uint64_value();
st->ientry_text = new_unshared_ientry(st); st->ientry_text = new_unshared_ientry(st);
} }
// Non-shared write // Non-shared write
@@ -932,7 +965,7 @@ resume_14:
st->ientry["size"].uint64_value() < st->new_size || st->ientry["size"].uint64_value() < st->new_size ||
st->ientry["shared_ino"].uint64_value() != 0) st->ientry["shared_ino"].uint64_value() != 0)
{ {
st->ext = &st->self->parent->kvfs->extends[st->ino]; st->ext = &st->proxy->kvfs->extends[st->ino];
st->ext->refcnt++; st->ext->refcnt++;
resume_15: resume_15:
if (st->ext->next_extend < st->new_size) if (st->ext->next_extend < st->new_size)
@@ -957,12 +990,12 @@ resume_16:
assert(st->ext->refcnt >= 0); assert(st->ext->refcnt >= 0);
if (st->ext->refcnt == 0) if (st->ext->refcnt == 0)
{ {
st->self->parent->kvfs->extends.erase(st->ino); st->proxy->kvfs->extends.erase(st->ino);
} }
} }
else else
{ {
st->self->parent->kvfs->touch_queue.insert(st->ino); st->proxy->kvfs->touch_queue.insert(st->ino);
} }
if (st->res == -EAGAIN) if (st->res == -EAGAIN)
{ {
@@ -976,15 +1009,16 @@ resume_16:
int kv_nfs3_write_proc(void *opaque, rpc_op_t *rop) int kv_nfs3_write_proc(void *opaque, rpc_op_t *rop)
{ {
nfs_kv_write_state *st = new nfs_kv_write_state; nfs_kv_write_state *st = new nfs_kv_write_state;
st->self = (nfs_client_t*)opaque; nfs_client_t *self = (nfs_client_t*)opaque;
st->proxy = ((nfs_client_t*)opaque)->parent;
st->rop = rop; st->rop = rop;
WRITE3args *args = (WRITE3args*)rop->request; WRITE3args *args = (WRITE3args*)rop->request;
WRITE3res *reply = (WRITE3res*)rop->reply; WRITE3res *reply = (WRITE3res*)rop->reply;
st->ino = kv_fh_inode(args->file); st->ino = kv_fh_inode(args->file);
st->offset = args->offset; st->offset = args->offset;
st->size = (args->count > args->data.size ? args->data.size : args->count); st->size = (args->count > args->data.size ? args->data.size : args->count);
if (st->self->parent->trace) if (st->proxy->trace)
fprintf(stderr, "[%d] WRITE %ju %ju+%ju\n", st->self->nfs_fd, st->ino, st->offset, st->size); fprintf(stderr, "[%d] WRITE %ju %ju+%ju\n", self->nfs_fd, st->ino, st->offset, st->size);
if (!st->ino || st->size > MAX_REQUEST_SIZE) if (!st->ino || st->size > MAX_REQUEST_SIZE)
{ {
*reply = (WRITE3res){ .status = NFS3ERR_INVAL }; *reply = (WRITE3res){ .status = NFS3ERR_INVAL };
@@ -1002,7 +1036,7 @@ int kv_nfs3_write_proc(void *opaque, rpc_op_t *rop)
{ {
reply->resok.count = (unsigned)st->size; reply->resok.count = (unsigned)st->size;
reply->resok.committed = st->stable || st->was_immediate ? FILE_SYNC : UNSTABLE; reply->resok.committed = st->stable || st->was_immediate ? FILE_SYNC : UNSTABLE;
*(uint64_t*)reply->resok.verf = st->self->parent->server_id; *(uint64_t*)reply->resok.verf = st->proxy->server_id;
} }
rpc_queue_reply(st->rop); rpc_queue_reply(st->rop);
delete st; delete st;
@@ -1010,3 +1044,23 @@ int kv_nfs3_write_proc(void *opaque, rpc_op_t *rop)
nfs_kv_continue_write(st, 0); nfs_kv_continue_write(st, 0);
return 1; return 1;
} }
void nfs_move_inode_from(nfs_proxy_t *proxy, uint64_t ino, uint64_t shared_ino, uint64_t shared_offset, std::function<void(int res, bool moved)> cb)
{
nfs_kv_write_state *st = new nfs_kv_write_state;
st->proxy = proxy;
st->ino = ino;
st->offset = 0;
st->size = 0;
st->force_move_from_inode = shared_ino;
st->force_move_from_offset = shared_offset;
st->buf = NULL;
st->stable = true;
st->allow_cache = false;
st->cb = [cb, st](int res)
{
cb(res, st->shared_inode != 0);
delete st;
};
nfs_kv_continue_write(st, 0);
}
src/nfs/nfs_proxy.cpp
+77 -29
View File
@@ -68,6 +68,32 @@ static const char* help_text =
" --port <PORT> use port <PORT> for NFS services (default is 2049)\n" " --port <PORT> use port <PORT> for NFS services (default is 2049)\n"
" --portmap 0 do not listen on port 111 (portmap/rpcbind, requires root)\n" " --portmap 0 do not listen on port 111 (portmap/rpcbind, requires root)\n"
"\n" "\n"
"vitastor-nfs --fs <NAME> upgrade\n"
" Upgrade FS metadata. Can be run online, but server should be restarted\n"
" after upgrade.\n"
"\n"
"vitastor-nfs --fs <NAME> defrag [OPTIONS] [--dry-run]\n"
" Defragment volumes used for small file storage having more than\n"
" <defrag_percent> %% of data removed. Can be run online. Options:\n"
" --volume_untouched 86400\n"
" Defragment volumes last appended to at least this number of seconds ago\n"
" --defrag_percent 50\n"
" Defragment volumes with at least this %% of removed data\n"
" --defrag_block_count 16\n"
" Read this number of pool blocks at once during defrag\n"
" --defrag_iodepth 16\n"
" Move up to this number of files in parallel during defrag\n"
" --trace\n"
" Print verbose defragmentation status\n"
" --dry-run\n"
" Skip modifications, only print status\n"
" --recalc-stats\n"
" Recalculate all volume statistics\n"
" --include-empty\n"
" Include old and empty volumes; make sure to restart NFS servers before using it\n"
" --no-rm\n"
" Move, but do not delete data\n"
"\n"
"OPTIONS:\n" "OPTIONS:\n"
" --fs <NAME> use VitastorFS with metadata in image <NAME>\n" " --fs <NAME> use VitastorFS with metadata in image <NAME>\n"
" --block use pseudo-FS presenting images as files\n" " --block use pseudo-FS presenting images as files\n"
@@ -113,7 +139,9 @@ json11::Json::object nfs_proxy_t::parse_args(int narg, const char *args[])
else if (args[i][0] == '-' && args[i][1] == '-') else if (args[i][0] == '-' && args[i][1] == '-')
{ {
const char *opt = args[i]+2; const char *opt = args[i]+2;
cfg[opt] = !strcmp(opt, "json") || !strcmp(opt, "block") || i == narg-1 ? "1" : args[++i]; cfg[str_replace(opt, "-", "_")] = !strcmp(opt, "json") || !strcmp(opt, "block") ||
!strcmp(opt, "dry-run") || !strcmp(opt, "recalc-stats") ||
!strcmp(opt, "include-empty") || !strcmp(opt, "no-rm") || i == narg-1 ? "1" : args[++i];
} }
else else
{ {
@@ -132,6 +160,10 @@ json11::Json::object nfs_proxy_t::parse_args(int narg, const char *args[])
else if (cmd.size() >= 1 && cmd[0] == "start") else if (cmd.size() >= 1 && cmd[0] == "start")
{ {
} }
else if (cmd.size() >= 1 && (cmd[0] == "upgrade" || cmd[0] == "defrag") && cfg["fs"].string_value() != "")
{
cfg["cmd"] = cmd[0];
}
else else
{ {
printf("%s", help_text); printf("%s", help_text);
@@ -211,6 +243,50 @@ void nfs_proxy_t::run(json11::Json cfg)
kvfs = new kv_fs_state_t(); kvfs = new kv_fs_state_t();
kvfs->init(this, cfg); kvfs->init(this, cfg);
} }
if (cfg["cmd"].is_null())
{
run_server(cfg);
}
else if (cfg["cmd"] == "defrag")
{
kvfs->defrag_all(cfg, [this](int res) { finished = true; });
}
else if (cfg["cmd"] == "upgrade")
{
kvfs->upgrade_db([this](int res) { finished = true; });
}
while (!finished)
{
ringloop->loop();
ringloop->wait();
}
// Destroy the client
cli->flush();
if (kvfs)
{
delete kvfs;
kvfs = NULL;
}
if (blockfs)
{
delete blockfs;
blockfs = NULL;
}
if (db)
{
delete db;
db = NULL;
}
delete cli;
delete epmgr;
delete ringloop;
cli = NULL;
epmgr = NULL;
ringloop = NULL;
}
void nfs_proxy_t::run_server(json11::Json cfg)
{
// Self-register portmap and NFS // Self-register portmap and NFS
pmap.reg_ports.insert((portmap_id_t){ pmap.reg_ports.insert((portmap_id_t){
.prog = PMAP_PROGRAM, .prog = PMAP_PROGRAM,
@@ -285,34 +361,6 @@ void nfs_proxy_t::run(json11::Json cfg)
{ {
write_pid(); write_pid();
} }
while (!finished)
{
ringloop->loop();
ringloop->wait();
}
// Destroy the client
cli->flush();
if (kvfs)
{
delete kvfs;
kvfs = NULL;
}
if (blockfs)
{
delete blockfs;
blockfs = NULL;
}
if (db)
{
delete db;
db = NULL;
}
delete cli;
delete epmgr;
delete ringloop;
cli = NULL;
epmgr = NULL;
ringloop = NULL;
} }
void nfs_proxy_t::watch_stats() void nfs_proxy_t::watch_stats()
src/nfs/nfs_proxy.h
+1
View File
@@ -66,6 +66,7 @@ public:
static json11::Json::object parse_args(int narg, const char *args[]); static json11::Json::object parse_args(int narg, const char *args[]);
void run(json11::Json cfg); void run(json11::Json cfg);
void run_server(json11::Json cfg);
void watch_stats(); void watch_stats();
void parse_stats(etcd_kv_t & kv); void parse_stats(etcd_kv_t & kv);
void check_default_pool(); void check_default_pool();
src/util/str_util.cpp
+11
View File
@@ -4,6 +4,7 @@
#include <assert.h> #include <assert.h>
#include <string.h> #include <string.h>
#include <unistd.h> #include <unistd.h>
#include <time.h>
#include <fcntl.h> #include <fcntl.h>
#include "str_util.h" #include "str_util.h"
@@ -474,3 +475,13 @@ std::string realpath_str(std::string path, bool nofail)
free(p); free(p);
return rp; return rp;
} }
std::string format_datetime(uint64_t unixtime)
{
char buf[128];
time_t ut = (time_t)unixtime;
tm lt;
localtime_r(&ut, &lt);
int len = strftime(buf, 128, "%Y-%m-%d %H:%M:%S", &lt);
return std::string(buf, len);
}
src/util/str_util.h
+1
View File
@@ -30,3 +30,4 @@ std::string scan_escaped(const std::string & cmd, size_t & pos, bool allow_unquo
std::string auto_addslashes(const std::string & str, const char *toescape = "\\\""); std::string auto_addslashes(const std::string & str, const char *toescape = "\\\"");
std::string addslashes(const std::string & str, const char *toescape = "\\\""); std::string addslashes(const std::string & str, const char *toescape = "\\\"");
std::string realpath_str(std::string path, bool nofail = true); std::string realpath_str(std::string path, bool nofail = true);
std::string format_datetime(uint64_t unixtime);