Compare commits
9 Commits
33950c1ec8
...
3bcf276d4d
Author | SHA1 | Date |
---|---|---|
Vitaliy Filippov | 3bcf276d4d | |
Vitaliy Filippov | 38db53f5ee | |
Vitaliy Filippov | cd543a90bc | |
Vitaliy Filippov | f600cc07b0 | |
Vitaliy Filippov | 6a8e530e6b | |
Vitaliy Filippov | 5cadb170b9 | |
Vitaliy Filippov | e72d4ed1d4 | |
Vitaliy Filippov | ff479a102d | |
Vitaliy Filippov | 27d0d5b06a |
|
@ -205,9 +205,8 @@ This parameter usually doesn't require to be changed.
|
||||||
- Default: 131072
|
- Default: 131072
|
||||||
|
|
||||||
Block size for this pool. The value from /vitastor/config/global is used when
|
Block size for this pool. The value from /vitastor/config/global is used when
|
||||||
unspecified. If your cluster has OSDs with different block sizes then pool must
|
unspecified. Only OSDs with matching block_size are used for each pool. If you
|
||||||
be restricted by [osd_tags](#osd_tags) to only include OSDs with matching block
|
want to further restrict OSDs for the pool, use [osd_tags](#osd_tags).
|
||||||
size.
|
|
||||||
|
|
||||||
Read more about this parameter in [Cluster-Wide Disk Layout Parameters](layout-cluster.en.md#block_size).
|
Read more about this parameter in [Cluster-Wide Disk Layout Parameters](layout-cluster.en.md#block_size).
|
||||||
|
|
||||||
|
@ -216,10 +215,9 @@ Read more about this parameter in [Cluster-Wide Disk Layout Parameters](layout-c
|
||||||
- Type: integer
|
- Type: integer
|
||||||
- Default: 4096
|
- Default: 4096
|
||||||
|
|
||||||
"Sector" size of virtual disks in this pool. The value from
|
"Sector" size of virtual disks in this pool. The value from /vitastor/config/global
|
||||||
/vitastor/config/global is used when unspecified. Similar to block_size, the
|
is used when unspecified. Similarly to block_size, only OSDs with matching
|
||||||
pool must be restricted by [osd_tags](#osd_tags) to only include OSDs with
|
bitmap_granularity are used for each pool.
|
||||||
matching bitmap_granularity.
|
|
||||||
|
|
||||||
Read more about this parameter in [Cluster-Wide Disk Layout Parameters](layout-cluster.en.md#bitmap_granularity).
|
Read more about this parameter in [Cluster-Wide Disk Layout Parameters](layout-cluster.en.md#bitmap_granularity).
|
||||||
|
|
||||||
|
@ -229,10 +227,11 @@ Read more about this parameter in [Cluster-Wide Disk Layout Parameters](layout-c
|
||||||
- Default: none
|
- Default: none
|
||||||
|
|
||||||
Immediate commit setting for this pool. The value from /vitastor/config/global
|
Immediate commit setting for this pool. The value from /vitastor/config/global
|
||||||
is used when unspecified. Similar to block_size, the pool must be restricted by
|
is used when unspecified. Similarly to block_size, only OSDs with compatible
|
||||||
[osd_tags](#osd_tags) to only include OSDs with compatible immediate_commit.
|
bitmap_granularity are used for each pool. "Compatible" means that a pool with
|
||||||
Compatible means that a pool with non-immediate commit will work with OSDs with
|
non-immediate commit will use OSDs with immediate commit enabled, but not vice
|
||||||
immediate commit enabled, but not vice versa.
|
versa. I.e., pools with "none" use all OSDs, pools with "small" only use OSDs
|
||||||
|
with "all" or "small", and pools with "all" only use OSDs with "all".
|
||||||
|
|
||||||
Read more about this parameter in [Cluster-Wide Disk Layout Parameters](layout-cluster.en.md#immediate_commit).
|
Read more about this parameter in [Cluster-Wide Disk Layout Parameters](layout-cluster.en.md#immediate_commit).
|
||||||
|
|
||||||
|
|
|
@ -208,8 +208,9 @@ PG в Vitastor эферемерны, то есть вы можете менят
|
||||||
|
|
||||||
Размер блока для данного пула. Если не задан, используется значение из
|
Размер блока для данного пула. Если не задан, используется значение из
|
||||||
/vitastor/config/global. Если в вашем кластере есть OSD с разными размерами
|
/vitastor/config/global. Если в вашем кластере есть OSD с разными размерами
|
||||||
блока, пул должен быть ограничен только OSD, блок которых равен блоку пула,
|
блока, пул будет использовать только OSD с размером блока, равным размеру блока
|
||||||
с помощью [osd_tags](#osd_tags).
|
пула. Если вы хотите сильнее ограничить набор используемых для пула OSD -
|
||||||
|
используйте [osd_tags](#osd_tags).
|
||||||
|
|
||||||
О самом параметре читайте в разделе [Дисковые параметры уровня кластера](layout-cluster.ru.md#block_size).
|
О самом параметре читайте в разделе [Дисковые параметры уровня кластера](layout-cluster.ru.md#block_size).
|
||||||
|
|
||||||
|
@ -219,9 +220,8 @@ PG в Vitastor эферемерны, то есть вы можете менят
|
||||||
- По умолчанию: 4096
|
- По умолчанию: 4096
|
||||||
|
|
||||||
Размер "сектора" виртуальных дисков в данном пуле. Если не задан, используется
|
Размер "сектора" виртуальных дисков в данном пуле. Если не задан, используется
|
||||||
значение из /vitastor/config/global. Аналогично block_size, пул должен быть
|
значение из /vitastor/config/global. Аналогично block_size, каждый пул будет
|
||||||
ограничен OSD со значением bitmap_granularity, равным значению пула, с помощью
|
использовать только OSD с совпадающей с пулом настройкой bitmap_granularity.
|
||||||
[osd_tags](#osd_tags).
|
|
||||||
|
|
||||||
О самом параметре читайте в разделе [Дисковые параметры уровня кластера](layout-cluster.ru.md#bitmap_granularity).
|
О самом параметре читайте в разделе [Дисковые параметры уровня кластера](layout-cluster.ru.md#bitmap_granularity).
|
||||||
|
|
||||||
|
@ -231,11 +231,13 @@ PG в Vitastor эферемерны, то есть вы можете менят
|
||||||
- По умолчанию: none
|
- По умолчанию: none
|
||||||
|
|
||||||
Настройка мгновенного коммита для данного пула. Если не задана, используется
|
Настройка мгновенного коммита для данного пула. Если не задана, используется
|
||||||
значение из /vitastor/config/global. Аналогично block_size, пул должен быть
|
значение из /vitastor/config/global. Аналогично block_size, каждый пул будет
|
||||||
ограничен OSD со значением bitmap_granularity, совместимым со значением пула, с
|
использовать только OSD с *совместимыми* настройками immediate_commit.
|
||||||
помощью [osd_tags](#osd_tags). Совместимость означает, что пул с отключенным
|
"Совместимыми" означает, что пул с отключенным мгновенным коммитом будет
|
||||||
мгновенным коммитом может работать на OSD с включённым мгновенным коммитом, но
|
использовать OSD с включённым мгновенным коммитом, но не наоборот. То есть,
|
||||||
не наоборот.
|
пул со значением "none" будет использовать все OSD, пул со "small" будет
|
||||||
|
использовать OSD с "all" или "small", а пул с "all" будет использовать только
|
||||||
|
OSD с "all".
|
||||||
|
|
||||||
О самом параметре читайте в разделе [Дисковые параметры уровня кластера](layout-cluster.ru.md#immediate_commit).
|
О самом параметре читайте в разделе [Дисковые параметры уровня кластера](layout-cluster.ru.md#immediate_commit).
|
||||||
|
|
||||||
|
|
41
mon/mon.js
41
mon/mon.js
|
@ -78,9 +78,15 @@ const etcd_tree = {
|
||||||
disk_alignment: 4096,
|
disk_alignment: 4096,
|
||||||
bitmap_granularity: 4096,
|
bitmap_granularity: 4096,
|
||||||
immediate_commit: false, // 'all' or 'small'
|
immediate_commit: false, // 'all' or 'small'
|
||||||
|
// client - configurable online
|
||||||
|
client_max_dirty_bytes: 33554432,
|
||||||
|
client_max_dirty_ops: 1024,
|
||||||
|
client_enable_writeback: false,
|
||||||
|
client_max_buffered_bytes: 33554432,
|
||||||
|
client_max_buffered_ops: 1024,
|
||||||
|
client_max_writeback_iodepth: 256,
|
||||||
// client and osd - configurable online
|
// client and osd - configurable online
|
||||||
log_level: 0,
|
log_level: 0,
|
||||||
client_dirty_limit: 33554432,
|
|
||||||
peer_connect_interval: 5, // seconds. min: 1
|
peer_connect_interval: 5, // seconds. min: 1
|
||||||
peer_connect_timeout: 5, // seconds. min: 1
|
peer_connect_timeout: 5, // seconds. min: 1
|
||||||
osd_idle_timeout: 5, // seconds. min: 1
|
osd_idle_timeout: 5, // seconds. min: 1
|
||||||
|
@ -1156,6 +1162,33 @@ class Mon
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
filter_osds_by_block_layout(flat_tree, block_size, bitmap_granularity, immediate_commit)
|
||||||
|
{
|
||||||
|
for (const host in flat_tree)
|
||||||
|
{
|
||||||
|
let found = 0;
|
||||||
|
for (const osd in flat_tree[host])
|
||||||
|
{
|
||||||
|
const osd_stat = this.state.osd.stats[osd];
|
||||||
|
if (osd_stat && (osd_stat.bs_block_size && osd_stat.bs_block_size != block_size ||
|
||||||
|
osd_stat.bitmap_granularity && osd_stat.bitmap_granularity != bitmap_granularity ||
|
||||||
|
osd_stat.immediate_commit == 'small' && immediate_commit == 'all' ||
|
||||||
|
osd_stat.immediate_commit == 'none' && immediate_commit != 'none'))
|
||||||
|
{
|
||||||
|
delete flat_tree[host][osd];
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
found++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (!found)
|
||||||
|
{
|
||||||
|
delete flat_tree[host];
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
get_affinity_osds(pool_cfg, up_osds, osd_tree)
|
get_affinity_osds(pool_cfg, up_osds, osd_tree)
|
||||||
{
|
{
|
||||||
let aff_osds = up_osds;
|
let aff_osds = up_osds;
|
||||||
|
@ -1216,6 +1249,12 @@ class Mon
|
||||||
pool_tree = pool_tree ? pool_tree.children : [];
|
pool_tree = pool_tree ? pool_tree.children : [];
|
||||||
pool_tree = LPOptimizer.flatten_tree(pool_tree, levels, pool_cfg.failure_domain, 'osd');
|
pool_tree = LPOptimizer.flatten_tree(pool_tree, levels, pool_cfg.failure_domain, 'osd');
|
||||||
this.filter_osds_by_tags(osd_tree, pool_tree, pool_cfg.osd_tags);
|
this.filter_osds_by_tags(osd_tree, pool_tree, pool_cfg.osd_tags);
|
||||||
|
this.filter_osds_by_block_layout(
|
||||||
|
pool_tree,
|
||||||
|
pool_cfg.block_size || this.config.block_size || 131072,
|
||||||
|
pool_cfg.bitmap_granularity || this.config.bitmap_granularity || 4096,
|
||||||
|
pool_cfg.immediate_commit || this.config.immediate_commit || 'none'
|
||||||
|
);
|
||||||
// These are for the purpose of building history.osd_sets
|
// These are for the purpose of building history.osd_sets
|
||||||
const real_prev_pgs = [];
|
const real_prev_pgs = [];
|
||||||
let pg_history = [];
|
let pg_history = [];
|
||||||
|
|
|
@ -137,6 +137,7 @@ endif (${WITH_FIO})
|
||||||
add_library(vitastor_client SHARED
|
add_library(vitastor_client SHARED
|
||||||
cluster_client.cpp
|
cluster_client.cpp
|
||||||
cluster_client_list.cpp
|
cluster_client_list.cpp
|
||||||
|
cluster_client_wb.cpp
|
||||||
vitastor_c.cpp
|
vitastor_c.cpp
|
||||||
cli_common.cpp
|
cli_common.cpp
|
||||||
cli_alloc_osd.cpp
|
cli_alloc_osd.cpp
|
||||||
|
@ -300,7 +301,7 @@ target_link_libraries(test_crc32
|
||||||
add_executable(test_cluster_client
|
add_executable(test_cluster_client
|
||||||
EXCLUDE_FROM_ALL
|
EXCLUDE_FROM_ALL
|
||||||
test_cluster_client.cpp
|
test_cluster_client.cpp
|
||||||
pg_states.cpp osd_ops.cpp cluster_client.cpp cluster_client_list.cpp msgr_op.cpp mock/messenger.cpp msgr_stop.cpp
|
pg_states.cpp osd_ops.cpp cluster_client.cpp cluster_client_list.cpp cluster_client_wb.cpp msgr_op.cpp mock/messenger.cpp msgr_stop.cpp
|
||||||
etcd_state_client.cpp timerfd_manager.cpp str_util.cpp ../json11/json11.cpp
|
etcd_state_client.cpp timerfd_manager.cpp str_util.cpp ../json11/json11.cpp
|
||||||
)
|
)
|
||||||
target_compile_definitions(test_cluster_client PUBLIC -D__MOCK__)
|
target_compile_definitions(test_cluster_client PUBLIC -D__MOCK__)
|
||||||
|
|
|
@ -384,6 +384,10 @@ void blockstore_impl_t::enqueue_op(blockstore_op_t *op)
|
||||||
ringloop->set_immediate([op]() { std::function<void (blockstore_op_t*)>(op->callback)(op); });
|
ringloop->set_immediate([op]() { std::function<void (blockstore_op_t*)>(op->callback)(op); });
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
if (op->opcode == BS_OP_SYNC)
|
||||||
|
{
|
||||||
|
unsynced_queued_ops = 0;
|
||||||
|
}
|
||||||
init_op(op);
|
init_op(op);
|
||||||
submit_queue.push_back(op);
|
submit_queue.push_back(op);
|
||||||
ringloop->wakeup();
|
ringloop->wakeup();
|
||||||
|
@ -393,6 +397,7 @@ void blockstore_impl_t::init_op(blockstore_op_t *op)
|
||||||
{
|
{
|
||||||
// Call constructor without allocating memory. We'll call destructor before returning op back
|
// Call constructor without allocating memory. We'll call destructor before returning op back
|
||||||
new ((void*)op->private_data) blockstore_op_private_t;
|
new ((void*)op->private_data) blockstore_op_private_t;
|
||||||
|
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
|
||||||
PRIV(op)->wait_for = 0;
|
PRIV(op)->wait_for = 0;
|
||||||
PRIV(op)->op_state = 0;
|
PRIV(op)->op_state = 0;
|
||||||
PRIV(op)->pending_ops = 0;
|
PRIV(op)->pending_ops = 0;
|
||||||
|
|
|
@ -210,7 +210,7 @@ struct blockstore_op_private_t
|
||||||
std::vector<copy_buffer_t> read_vec;
|
std::vector<copy_buffer_t> read_vec;
|
||||||
|
|
||||||
// Sync, write
|
// Sync, write
|
||||||
int min_flushed_journal_sector, max_flushed_journal_sector;
|
uint64_t min_flushed_journal_sector, max_flushed_journal_sector;
|
||||||
|
|
||||||
// Write
|
// Write
|
||||||
struct iovec iov_zerofill[3];
|
struct iovec iov_zerofill[3];
|
||||||
|
@ -220,7 +220,6 @@ struct blockstore_op_private_t
|
||||||
|
|
||||||
// Sync
|
// Sync
|
||||||
std::vector<obj_ver_id> sync_big_writes, sync_small_writes;
|
std::vector<obj_ver_id> sync_big_writes, sync_small_writes;
|
||||||
int sync_small_checked, sync_big_checked;
|
|
||||||
};
|
};
|
||||||
|
|
||||||
typedef uint32_t pool_id_t;
|
typedef uint32_t pool_id_t;
|
||||||
|
@ -263,6 +262,8 @@ class blockstore_impl_t
|
||||||
int throttle_target_parallelism = 1;
|
int throttle_target_parallelism = 1;
|
||||||
// Minimum difference in microseconds between target and real execution times to throttle the response
|
// Minimum difference in microseconds between target and real execution times to throttle the response
|
||||||
int throttle_threshold_us = 50;
|
int throttle_threshold_us = 50;
|
||||||
|
// Maximum writes between automatically added fsync operations
|
||||||
|
uint64_t autosync_writes = 128;
|
||||||
/******* END OF OPTIONS *******/
|
/******* END OF OPTIONS *******/
|
||||||
|
|
||||||
struct ring_consumer_t ring_consumer;
|
struct ring_consumer_t ring_consumer;
|
||||||
|
@ -274,6 +275,7 @@ class blockstore_impl_t
|
||||||
std::vector<blockstore_op_t*> submit_queue;
|
std::vector<blockstore_op_t*> submit_queue;
|
||||||
std::vector<obj_ver_id> unsynced_big_writes, unsynced_small_writes;
|
std::vector<obj_ver_id> unsynced_big_writes, unsynced_small_writes;
|
||||||
int unsynced_big_write_count = 0;
|
int unsynced_big_write_count = 0;
|
||||||
|
int unsynced_queued_ops = 0;
|
||||||
allocator *data_alloc = NULL;
|
allocator *data_alloc = NULL;
|
||||||
uint8_t *zero_object;
|
uint8_t *zero_object;
|
||||||
|
|
||||||
|
|
|
@ -198,6 +198,7 @@ void blockstore_impl_t::prepare_journal_sector_write(int cur_sector, blockstore_
|
||||||
priv->pending_ops++;
|
priv->pending_ops++;
|
||||||
if (!priv->min_flushed_journal_sector)
|
if (!priv->min_flushed_journal_sector)
|
||||||
priv->min_flushed_journal_sector = 1+cur_sector;
|
priv->min_flushed_journal_sector = 1+cur_sector;
|
||||||
|
assert(priv->min_flushed_journal_sector <= journal.sector_count);
|
||||||
priv->max_flushed_journal_sector = 1+cur_sector;
|
priv->max_flushed_journal_sector = 1+cur_sector;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -19,6 +19,10 @@ void blockstore_impl_t::parse_config(blockstore_config_t & config, bool init)
|
||||||
throttle_target_mbs = strtoull(config["throttle_target_mbs"].c_str(), NULL, 10);
|
throttle_target_mbs = strtoull(config["throttle_target_mbs"].c_str(), NULL, 10);
|
||||||
throttle_target_parallelism = strtoull(config["throttle_target_parallelism"].c_str(), NULL, 10);
|
throttle_target_parallelism = strtoull(config["throttle_target_parallelism"].c_str(), NULL, 10);
|
||||||
throttle_threshold_us = strtoull(config["throttle_threshold_us"].c_str(), NULL, 10);
|
throttle_threshold_us = strtoull(config["throttle_threshold_us"].c_str(), NULL, 10);
|
||||||
|
if (config.find("autosync_writes") != config.end())
|
||||||
|
{
|
||||||
|
autosync_writes = strtoull(config["autosync_writes"].c_str(), NULL, 10);
|
||||||
|
}
|
||||||
if (!max_flusher_count)
|
if (!max_flusher_count)
|
||||||
{
|
{
|
||||||
max_flusher_count = 256;
|
max_flusher_count = 256;
|
||||||
|
|
|
@ -27,8 +27,6 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
|
||||||
unsynced_big_write_count -= unsynced_big_writes.size();
|
unsynced_big_write_count -= unsynced_big_writes.size();
|
||||||
PRIV(op)->sync_big_writes.swap(unsynced_big_writes);
|
PRIV(op)->sync_big_writes.swap(unsynced_big_writes);
|
||||||
PRIV(op)->sync_small_writes.swap(unsynced_small_writes);
|
PRIV(op)->sync_small_writes.swap(unsynced_small_writes);
|
||||||
PRIV(op)->sync_small_checked = 0;
|
|
||||||
PRIV(op)->sync_big_checked = 0;
|
|
||||||
unsynced_big_writes.clear();
|
unsynced_big_writes.clear();
|
||||||
unsynced_small_writes.clear();
|
unsynced_small_writes.clear();
|
||||||
if (PRIV(op)->sync_big_writes.size() > 0)
|
if (PRIV(op)->sync_big_writes.size() > 0)
|
||||||
|
|
|
@ -127,8 +127,9 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if (wait_big && !is_del && !deleted && op->len < dsk.data_block_size &&
|
bool imm = (op->len < dsk.data_block_size ? (immediate_commit != IMMEDIATE_NONE) : (immediate_commit == IMMEDIATE_ALL));
|
||||||
immediate_commit != IMMEDIATE_ALL)
|
if (wait_big && !is_del && !deleted && op->len < dsk.data_block_size && !imm ||
|
||||||
|
!imm && unsynced_queued_ops >= autosync_writes)
|
||||||
{
|
{
|
||||||
// Issue an additional sync so that the previous big write can reach the journal
|
// Issue an additional sync so that the previous big write can reach the journal
|
||||||
blockstore_op_t *sync_op = new blockstore_op_t;
|
blockstore_op_t *sync_op = new blockstore_op_t;
|
||||||
|
@ -139,6 +140,8 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
|
||||||
};
|
};
|
||||||
enqueue_op(sync_op);
|
enqueue_op(sync_op);
|
||||||
}
|
}
|
||||||
|
else if (!imm)
|
||||||
|
unsynced_queued_ops++;
|
||||||
#ifdef BLOCKSTORE_DEBUG
|
#ifdef BLOCKSTORE_DEBUG
|
||||||
if (is_del)
|
if (is_del)
|
||||||
printf("Delete %lx:%lx v%lu\n", op->oid.inode, op->oid.stripe, op->version);
|
printf("Delete %lx:%lx v%lu\n", op->oid.inode, op->oid.stripe, op->version);
|
||||||
|
@ -378,7 +381,6 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
|
||||||
sqe, dsk.data_fd, PRIV(op)->iov_zerofill, vcnt, dsk.data_offset + (loc << dsk.block_order) + op->offset - stripe_offset
|
sqe, dsk.data_fd, PRIV(op)->iov_zerofill, vcnt, dsk.data_offset + (loc << dsk.block_order) + op->offset - stripe_offset
|
||||||
);
|
);
|
||||||
PRIV(op)->pending_ops = 1;
|
PRIV(op)->pending_ops = 1;
|
||||||
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
|
|
||||||
if (immediate_commit != IMMEDIATE_ALL)
|
if (immediate_commit != IMMEDIATE_ALL)
|
||||||
{
|
{
|
||||||
// Increase the counter, but don't save into unsynced_writes yet (can't sync until the write is finished)
|
// Increase the counter, but don't save into unsynced_writes yet (can't sync until the write is finished)
|
||||||
|
@ -415,17 +417,11 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
|
||||||
write_iodepth++;
|
write_iodepth++;
|
||||||
// Got SQEs. Prepare previous journal sector write if required
|
// Got SQEs. Prepare previous journal sector write if required
|
||||||
auto cb = [this, op](ring_data_t *data) { handle_write_event(data, op); };
|
auto cb = [this, op](ring_data_t *data) { handle_write_event(data, op); };
|
||||||
if (immediate_commit == IMMEDIATE_NONE)
|
if (immediate_commit == IMMEDIATE_NONE &&
|
||||||
{
|
!journal.entry_fits(sizeof(journal_entry_small_write) + dyn_size))
|
||||||
if (!journal.entry_fits(sizeof(journal_entry_small_write) + dyn_size))
|
|
||||||
{
|
{
|
||||||
prepare_journal_sector_write(journal.cur_sector, op);
|
prepare_journal_sector_write(journal.cur_sector, op);
|
||||||
}
|
}
|
||||||
else
|
|
||||||
{
|
|
||||||
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// Then pre-fill journal entry
|
// Then pre-fill journal entry
|
||||||
journal_entry_small_write *je = (journal_entry_small_write*)prefill_single_journal_entry(
|
journal_entry_small_write *je = (journal_entry_small_write*)prefill_single_journal_entry(
|
||||||
journal, op->opcode == BS_OP_WRITE_STABLE ? JE_SMALL_WRITE_INSTANT : JE_SMALL_WRITE,
|
journal, op->opcode == BS_OP_WRITE_STABLE ? JE_SMALL_WRITE_INSTANT : JE_SMALL_WRITE,
|
||||||
|
@ -750,18 +746,12 @@ int blockstore_impl_t::dequeue_del(blockstore_op_t *op)
|
||||||
}
|
}
|
||||||
write_iodepth++;
|
write_iodepth++;
|
||||||
// Prepare journal sector write
|
// Prepare journal sector write
|
||||||
if (immediate_commit == IMMEDIATE_NONE)
|
if (immediate_commit == IMMEDIATE_NONE &&
|
||||||
{
|
(dsk.journal_block_size - journal.in_sector_pos) < sizeof(journal_entry_del) &&
|
||||||
if ((dsk.journal_block_size - journal.in_sector_pos) < sizeof(journal_entry_del) &&
|
|
||||||
journal.sector_info[journal.cur_sector].dirty)
|
journal.sector_info[journal.cur_sector].dirty)
|
||||||
{
|
{
|
||||||
prepare_journal_sector_write(journal.cur_sector, op);
|
prepare_journal_sector_write(journal.cur_sector, op);
|
||||||
}
|
}
|
||||||
else
|
|
||||||
{
|
|
||||||
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// Pre-fill journal entry
|
// Pre-fill journal entry
|
||||||
journal_entry_del *je = (journal_entry_del*)prefill_single_journal_entry(
|
journal_entry_del *je = (journal_entry_del*)prefill_single_journal_entry(
|
||||||
journal, JE_DELETE, sizeof(struct journal_entry_del)
|
journal, JE_DELETE, sizeof(struct journal_entry_del)
|
||||||
|
|
|
@ -349,6 +349,7 @@ static int run(cli_tool_t *p, json11::Json::object cfg)
|
||||||
p->ringloop->wait();
|
p->ringloop->wait();
|
||||||
}
|
}
|
||||||
// Destroy the client
|
// Destroy the client
|
||||||
|
p->cli->flush();
|
||||||
delete p->cli;
|
delete p->cli;
|
||||||
delete p->epmgr;
|
delete p->epmgr;
|
||||||
delete p->ringloop;
|
delete p->ringloop;
|
||||||
|
|
|
@ -53,6 +53,7 @@ struct snap_merger_t
|
||||||
std::map<inode_t, std::vector<uint64_t>> layer_lists;
|
std::map<inode_t, std::vector<uint64_t>> layer_lists;
|
||||||
std::map<inode_t, uint64_t> layer_block_size;
|
std::map<inode_t, uint64_t> layer_block_size;
|
||||||
std::map<inode_t, uint64_t> layer_list_pos;
|
std::map<inode_t, uint64_t> layer_list_pos;
|
||||||
|
std::vector<snap_rw_op_t*> continue_rwo, continue_rwo2;
|
||||||
int in_flight = 0;
|
int in_flight = 0;
|
||||||
uint64_t last_fsync_offset = 0;
|
uint64_t last_fsync_offset = 0;
|
||||||
uint64_t last_written_offset = 0;
|
uint64_t last_written_offset = 0;
|
||||||
|
@ -304,6 +305,12 @@ struct snap_merger_t
|
||||||
oit = merge_offsets.begin();
|
oit = merge_offsets.begin();
|
||||||
resume_5:
|
resume_5:
|
||||||
// Now read, overwrite and optionally delete offsets one by one
|
// Now read, overwrite and optionally delete offsets one by one
|
||||||
|
continue_rwo2.swap(continue_rwo);
|
||||||
|
for (auto rwo: continue_rwo2)
|
||||||
|
{
|
||||||
|
next_write(rwo);
|
||||||
|
}
|
||||||
|
continue_rwo2.clear();
|
||||||
while (in_flight < parent->iodepth*parent->parallel_osds &&
|
while (in_flight < parent->iodepth*parent->parallel_osds &&
|
||||||
oit != merge_offsets.end() && !rwo_error.size())
|
oit != merge_offsets.end() && !rwo_error.size())
|
||||||
{
|
{
|
||||||
|
@ -464,7 +471,8 @@ struct snap_merger_t
|
||||||
rwo->error_offset = op->offset;
|
rwo->error_offset = op->offset;
|
||||||
rwo->error_read = true;
|
rwo->error_read = true;
|
||||||
}
|
}
|
||||||
next_write(rwo);
|
continue_rwo.push_back(rwo);
|
||||||
|
parent->ringloop->wakeup();
|
||||||
};
|
};
|
||||||
parent->cli->execute(op);
|
parent->cli->execute(op);
|
||||||
}
|
}
|
||||||
|
@ -544,11 +552,9 @@ struct snap_merger_t
|
||||||
}
|
}
|
||||||
// Increment CAS version
|
// Increment CAS version
|
||||||
rwo->op.version = subop->version;
|
rwo->op.version = subop->version;
|
||||||
if (use_cas)
|
|
||||||
next_write(rwo);
|
|
||||||
else
|
|
||||||
autofree_op(rwo);
|
|
||||||
delete subop;
|
delete subop;
|
||||||
|
continue_rwo.push_back(rwo);
|
||||||
|
parent->ringloop->wakeup();
|
||||||
};
|
};
|
||||||
parent->cli->execute(subop);
|
parent->cli->execute(subop);
|
||||||
}
|
}
|
||||||
|
|
|
@ -3,21 +3,13 @@
|
||||||
|
|
||||||
#include <stdexcept>
|
#include <stdexcept>
|
||||||
#include <assert.h>
|
#include <assert.h>
|
||||||
#include "cluster_client.h"
|
#include "cluster_client_impl.h"
|
||||||
|
#include "http_client.h" // json_is_true
|
||||||
#define SCRAP_BUFFER_SIZE 4*1024*1024
|
|
||||||
#define PART_SENT 1
|
|
||||||
#define PART_DONE 2
|
|
||||||
#define PART_ERROR 4
|
|
||||||
#define PART_RETRY 8
|
|
||||||
#define CACHE_DIRTY 1
|
|
||||||
#define CACHE_FLUSHING 2
|
|
||||||
#define CACHE_REPEATING 3
|
|
||||||
#define OP_FLUSH_BUFFER 0x02
|
|
||||||
#define OP_IMMEDIATE_COMMIT 0x04
|
|
||||||
|
|
||||||
cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json & config)
|
cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json & config)
|
||||||
{
|
{
|
||||||
|
wb = new writeback_cache_t();
|
||||||
|
|
||||||
cli_config = config.object_items();
|
cli_config = config.object_items();
|
||||||
file_config = osd_messenger_t::read_config(config);
|
file_config = osd_messenger_t::read_config(config);
|
||||||
config = osd_messenger_t::merge_configs(cli_config, file_config, etcd_global_config, {});
|
config = osd_messenger_t::merge_configs(cli_config, file_config, etcd_global_config, {});
|
||||||
|
@ -37,31 +29,15 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
|
||||||
continue_lists();
|
continue_lists();
|
||||||
continue_raw_ops(peer_osd);
|
continue_raw_ops(peer_osd);
|
||||||
}
|
}
|
||||||
else if (dirty_buffers.size())
|
else
|
||||||
{
|
{
|
||||||
// peer_osd just dropped connection
|
// peer_osd just dropped connection
|
||||||
// determine WHICH dirty_buffers are now obsolete and repeat them
|
// determine WHICH dirty_buffers are now obsolete and repeat them
|
||||||
for (auto wr_it = dirty_buffers.begin(), flush_it = wr_it, last_it = wr_it; ; )
|
if (wb->repeat_ops_for(this, peer_osd) > 0)
|
||||||
{
|
{
|
||||||
bool end = wr_it == dirty_buffers.end();
|
|
||||||
bool flush_this = !end && wr_it->second.state != CACHE_REPEATING &&
|
|
||||||
affects_osd(wr_it->first.inode, wr_it->first.stripe, wr_it->second.len, peer_osd);
|
|
||||||
if (flush_it != wr_it && (end || !flush_this ||
|
|
||||||
wr_it->first.inode != flush_it->first.inode ||
|
|
||||||
wr_it->first.stripe != last_it->first.stripe+last_it->second.len))
|
|
||||||
{
|
|
||||||
flush_buffers(flush_it, wr_it);
|
|
||||||
flush_it = wr_it;
|
|
||||||
}
|
|
||||||
if (end)
|
|
||||||
break;
|
|
||||||
last_it = wr_it;
|
|
||||||
wr_it++;
|
|
||||||
if (!flush_this)
|
|
||||||
flush_it = wr_it;
|
|
||||||
}
|
|
||||||
continue_ops();
|
continue_ops();
|
||||||
}
|
}
|
||||||
|
}
|
||||||
};
|
};
|
||||||
msgr.exec_op = [this](osd_op_t *op)
|
msgr.exec_op = [this](osd_op_t *op)
|
||||||
{
|
{
|
||||||
|
@ -88,16 +64,14 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
|
||||||
|
|
||||||
cluster_client_t::~cluster_client_t()
|
cluster_client_t::~cluster_client_t()
|
||||||
{
|
{
|
||||||
for (auto bp: dirty_buffers)
|
msgr.repeer_pgs = [this](osd_num_t){};
|
||||||
{
|
|
||||||
free(bp.second.buf);
|
|
||||||
}
|
|
||||||
dirty_buffers.clear();
|
|
||||||
if (ringloop)
|
if (ringloop)
|
||||||
{
|
{
|
||||||
ringloop->unregister_consumer(&consumer);
|
ringloop->unregister_consumer(&consumer);
|
||||||
}
|
}
|
||||||
free(scrap_buffer);
|
free(scrap_buffer);
|
||||||
|
delete wb;
|
||||||
|
wb = NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
cluster_op_t::~cluster_op_t()
|
cluster_op_t::~cluster_op_t()
|
||||||
|
@ -146,6 +120,19 @@ void cluster_client_t::init_msgr()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void cluster_client_t::unshift_op(cluster_op_t *op)
|
||||||
|
{
|
||||||
|
op->next = op_queue_head;
|
||||||
|
if (op_queue_head)
|
||||||
|
{
|
||||||
|
op_queue_head->prev = op;
|
||||||
|
op_queue_head = op;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
op_queue_tail = op_queue_head = op;
|
||||||
|
inc_wait(op->opcode, op->flags, op->next, 1);
|
||||||
|
}
|
||||||
|
|
||||||
void cluster_client_t::calc_wait(cluster_op_t *op)
|
void cluster_client_t::calc_wait(cluster_op_t *op)
|
||||||
{
|
{
|
||||||
op->prev_wait = 0;
|
op->prev_wait = 0;
|
||||||
|
@ -166,7 +153,7 @@ void cluster_client_t::calc_wait(cluster_op_t *op)
|
||||||
{
|
{
|
||||||
for (auto prev = op->prev; prev; prev = prev->prev)
|
for (auto prev = op->prev; prev; prev = prev->prev)
|
||||||
{
|
{
|
||||||
if (prev->opcode == OSD_OP_SYNC || prev->opcode == OSD_OP_WRITE && !(prev->flags & OP_IMMEDIATE_COMMIT))
|
if (prev->opcode == OSD_OP_SYNC || prev->opcode == OSD_OP_WRITE && (!(prev->flags & OP_IMMEDIATE_COMMIT) || enable_writeback))
|
||||||
{
|
{
|
||||||
op->prev_wait++;
|
op->prev_wait++;
|
||||||
}
|
}
|
||||||
|
@ -176,20 +163,6 @@ void cluster_client_t::calc_wait(cluster_op_t *op)
|
||||||
}
|
}
|
||||||
else /* if (op->opcode == OSD_OP_READ || op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP) */
|
else /* if (op->opcode == OSD_OP_READ || op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP) */
|
||||||
{
|
{
|
||||||
for (auto prev = op_queue_head; prev && prev != op; prev = prev->next)
|
|
||||||
{
|
|
||||||
if (prev->opcode == OSD_OP_WRITE && (prev->flags & OP_FLUSH_BUFFER))
|
|
||||||
{
|
|
||||||
op->prev_wait++;
|
|
||||||
}
|
|
||||||
else if (prev->opcode == OSD_OP_WRITE || prev->opcode == OSD_OP_READ ||
|
|
||||||
prev->opcode == OSD_OP_READ_BITMAP || prev->opcode == OSD_OP_READ_CHAIN_BITMAP)
|
|
||||||
{
|
|
||||||
// Flushes are always in the beginning (we're scanning from the beginning of the queue)
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (!op->prev_wait)
|
|
||||||
continue_rw(op);
|
continue_rw(op);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -201,10 +174,8 @@ void cluster_client_t::inc_wait(uint64_t opcode, uint64_t flags, cluster_op_t *n
|
||||||
while (next)
|
while (next)
|
||||||
{
|
{
|
||||||
auto n2 = next->next;
|
auto n2 = next->next;
|
||||||
if (next->opcode == OSD_OP_SYNC && !(flags & OP_IMMEDIATE_COMMIT) ||
|
if (next->opcode == OSD_OP_SYNC && (!(flags & OP_IMMEDIATE_COMMIT) || enable_writeback) ||
|
||||||
next->opcode == OSD_OP_WRITE && (flags & OP_FLUSH_BUFFER) && !(next->flags & OP_FLUSH_BUFFER) ||
|
next->opcode == OSD_OP_WRITE && (flags & OP_FLUSH_BUFFER) && !(next->flags & OP_FLUSH_BUFFER))
|
||||||
(next->opcode == OSD_OP_READ || next->opcode == OSD_OP_READ_BITMAP ||
|
|
||||||
next->opcode == OSD_OP_READ_CHAIN_BITMAP) && (flags & OP_FLUSH_BUFFER))
|
|
||||||
{
|
{
|
||||||
next->prev_wait += inc;
|
next->prev_wait += inc;
|
||||||
assert(next->prev_wait >= 0);
|
assert(next->prev_wait >= 0);
|
||||||
|
@ -255,13 +226,37 @@ void cluster_client_t::erase_op(cluster_op_t *op)
|
||||||
op_queue_tail = op->prev;
|
op_queue_tail = op->prev;
|
||||||
op->next = op->prev = NULL;
|
op->next = op->prev = NULL;
|
||||||
if (flags & OP_FLUSH_BUFFER)
|
if (flags & OP_FLUSH_BUFFER)
|
||||||
|
{
|
||||||
|
// Completed flushes change writeback buffer states,
|
||||||
|
// so the callback should be run before inc_wait()
|
||||||
|
// 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
|
||||||
std::function<void(cluster_op_t*)>(op->callback)(op);
|
std::function<void(cluster_op_t*)>(op->callback)(op);
|
||||||
if (!(flags & OP_IMMEDIATE_COMMIT))
|
}
|
||||||
|
if (!(flags & OP_IMMEDIATE_COMMIT) || enable_writeback)
|
||||||
|
{
|
||||||
inc_wait(opcode, flags, next, -1);
|
inc_wait(opcode, flags, next, -1);
|
||||||
|
}
|
||||||
|
if (!(flags & OP_FLUSH_BUFFER))
|
||||||
|
{
|
||||||
// Call callback at the end to avoid inconsistencies in prev_wait
|
// Call callback at the end to avoid inconsistencies in prev_wait
|
||||||
// if the callback adds more operations itself
|
// if the callback adds more operations itself
|
||||||
if (!(flags & OP_FLUSH_BUFFER))
|
|
||||||
std::function<void(cluster_op_t*)>(op->callback)(op);
|
std::function<void(cluster_op_t*)>(op->callback)(op);
|
||||||
|
}
|
||||||
|
if (flags & OP_FLUSH_BUFFER)
|
||||||
|
{
|
||||||
|
int i = 0;
|
||||||
|
while (i < wb->writeback_overflow.size() && wb->writebacks_active < client_max_writeback_iodepth)
|
||||||
|
{
|
||||||
|
execute_internal(wb->writeback_overflow[i]);
|
||||||
|
i++;
|
||||||
|
}
|
||||||
|
if (i > 0)
|
||||||
|
{
|
||||||
|
wb->writeback_overflow.erase(wb->writeback_overflow.begin(), wb->writeback_overflow.begin()+i);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
void cluster_client_t::continue_ops(bool up_retry)
|
void cluster_client_t::continue_ops(bool up_retry)
|
||||||
|
@ -305,6 +300,7 @@ void cluster_client_t::on_load_config_hook(json11::Json::object & etcd_global_co
|
||||||
{
|
{
|
||||||
this->etcd_global_config = etcd_global_config;
|
this->etcd_global_config = etcd_global_config;
|
||||||
config = osd_messenger_t::merge_configs(cli_config, file_config, etcd_global_config, {});
|
config = osd_messenger_t::merge_configs(cli_config, file_config, etcd_global_config, {});
|
||||||
|
// client_max_dirty_bytes/client_dirty_limit
|
||||||
if (config.find("client_max_dirty_bytes") != config.end())
|
if (config.find("client_max_dirty_bytes") != config.end())
|
||||||
{
|
{
|
||||||
client_max_dirty_bytes = config["client_max_dirty_bytes"].uint64_value();
|
client_max_dirty_bytes = config["client_max_dirty_bytes"].uint64_value();
|
||||||
|
@ -320,11 +316,34 @@ void cluster_client_t::on_load_config_hook(json11::Json::object & etcd_global_co
|
||||||
{
|
{
|
||||||
client_max_dirty_bytes = DEFAULT_CLIENT_MAX_DIRTY_BYTES;
|
client_max_dirty_bytes = DEFAULT_CLIENT_MAX_DIRTY_BYTES;
|
||||||
}
|
}
|
||||||
|
// client_max_dirty_ops
|
||||||
client_max_dirty_ops = config["client_max_dirty_ops"].uint64_value();
|
client_max_dirty_ops = config["client_max_dirty_ops"].uint64_value();
|
||||||
if (!client_max_dirty_ops)
|
if (!client_max_dirty_ops)
|
||||||
{
|
{
|
||||||
client_max_dirty_ops = DEFAULT_CLIENT_MAX_DIRTY_OPS;
|
client_max_dirty_ops = DEFAULT_CLIENT_MAX_DIRTY_OPS;
|
||||||
}
|
}
|
||||||
|
// client_enable_writeback
|
||||||
|
enable_writeback = json_is_true(config["client_enable_writeback"]) &&
|
||||||
|
json_is_true(config["client_writeback_allowed"]);
|
||||||
|
// client_max_buffered_bytes
|
||||||
|
client_max_buffered_bytes = config["client_max_buffered_bytes"].uint64_value();
|
||||||
|
if (!client_max_buffered_bytes)
|
||||||
|
{
|
||||||
|
client_max_buffered_bytes = DEFAULT_CLIENT_MAX_BUFFERED_BYTES;
|
||||||
|
}
|
||||||
|
// client_max_buffered_ops
|
||||||
|
client_max_buffered_ops = config["client_max_buffered_ops"].uint64_value();
|
||||||
|
if (!client_max_buffered_ops)
|
||||||
|
{
|
||||||
|
client_max_buffered_ops = DEFAULT_CLIENT_MAX_BUFFERED_OPS;
|
||||||
|
}
|
||||||
|
// client_max_writeback_iodepth
|
||||||
|
client_max_writeback_iodepth = config["client_max_writeback_iodepth"].uint64_value();
|
||||||
|
if (!client_max_writeback_iodepth)
|
||||||
|
{
|
||||||
|
client_max_writeback_iodepth = DEFAULT_CLIENT_MAX_WRITEBACK_IODEPTH;
|
||||||
|
}
|
||||||
|
// up_wait_retry_interval
|
||||||
up_wait_retry_interval = config["up_wait_retry_interval"].uint64_value();
|
up_wait_retry_interval = config["up_wait_retry_interval"].uint64_value();
|
||||||
if (!up_wait_retry_interval)
|
if (!up_wait_retry_interval)
|
||||||
{
|
{
|
||||||
|
@ -384,6 +403,8 @@ void cluster_client_t::on_change_hook(std::map<std::string, etcd_kv_t> & changes
|
||||||
|
|
||||||
bool cluster_client_t::get_immediate_commit(uint64_t inode)
|
bool cluster_client_t::get_immediate_commit(uint64_t inode)
|
||||||
{
|
{
|
||||||
|
if (enable_writeback)
|
||||||
|
return false;
|
||||||
pool_id_t pool_id = INODE_POOL(inode);
|
pool_id_t pool_id = INODE_POOL(inode);
|
||||||
if (!pool_id)
|
if (!pool_id)
|
||||||
return true;
|
return true;
|
||||||
|
@ -418,6 +439,41 @@ void cluster_client_t::on_ready(std::function<void(void)> fn)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
bool cluster_client_t::flush()
|
||||||
|
{
|
||||||
|
if (!ringloop)
|
||||||
|
{
|
||||||
|
if (wb->writeback_queue.size())
|
||||||
|
{
|
||||||
|
wb->start_writebacks(this, 0);
|
||||||
|
cluster_op_t *sync = new cluster_op_t;
|
||||||
|
sync->opcode = OSD_OP_SYNC;
|
||||||
|
sync->callback = [this](cluster_op_t *sync)
|
||||||
|
{
|
||||||
|
delete sync;
|
||||||
|
};
|
||||||
|
execute(sync);
|
||||||
|
}
|
||||||
|
return op_queue_head == NULL;
|
||||||
|
}
|
||||||
|
bool sync_done = false;
|
||||||
|
cluster_op_t *sync = new cluster_op_t;
|
||||||
|
sync->opcode = OSD_OP_SYNC;
|
||||||
|
sync->callback = [this, &sync_done](cluster_op_t *sync)
|
||||||
|
{
|
||||||
|
delete sync;
|
||||||
|
sync_done = true;
|
||||||
|
};
|
||||||
|
execute(sync);
|
||||||
|
while (!sync_done)
|
||||||
|
{
|
||||||
|
ringloop->loop();
|
||||||
|
if (!sync_done)
|
||||||
|
ringloop->wait();
|
||||||
|
}
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* How writes are synced when immediate_commit is false
|
* How writes are synced when immediate_commit is false
|
||||||
*
|
*
|
||||||
|
@ -438,6 +494,9 @@ void cluster_client_t::on_ready(std::function<void(void)> fn)
|
||||||
* 3) if yes, send all SYNCs. otherwise, leave current SYNC as is.
|
* 3) if yes, send all SYNCs. otherwise, leave current SYNC as is.
|
||||||
* 4) if any of them fail due to disconnected peers, repeat SYNC after repeating all writes
|
* 4) if any of them fail due to disconnected peers, repeat SYNC after repeating all writes
|
||||||
* 5) if any of them fail due to other errors, fail the SYNC operation
|
* 5) if any of them fail due to other errors, fail the SYNC operation
|
||||||
|
*
|
||||||
|
* If writeback caching is turned on and writeback limit is not exhausted:
|
||||||
|
* data is just copied and the write is confirmed to the client.
|
||||||
*/
|
*/
|
||||||
void cluster_client_t::execute(cluster_op_t *op)
|
void cluster_client_t::execute(cluster_op_t *op)
|
||||||
{
|
{
|
||||||
|
@ -453,36 +512,73 @@ void cluster_client_t::execute(cluster_op_t *op)
|
||||||
offline_ops.push_back(op);
|
offline_ops.push_back(op);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
op->flags = op->flags & OSD_OP_IGNORE_READONLY; // the only allowed flag
|
||||||
|
execute_internal(op);
|
||||||
|
}
|
||||||
|
|
||||||
|
void cluster_client_t::execute_internal(cluster_op_t *op)
|
||||||
|
{
|
||||||
op->cur_inode = op->inode;
|
op->cur_inode = op->inode;
|
||||||
op->retval = 0;
|
op->retval = 0;
|
||||||
op->flags = op->flags & OSD_OP_IGNORE_READONLY; // the only allowed flag
|
|
||||||
// check alignment, readonly flag and so on
|
// check alignment, readonly flag and so on
|
||||||
if (!check_rw(op))
|
if (!check_rw(op))
|
||||||
{
|
{
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
if (op->opcode == OSD_OP_WRITE && enable_writeback && !(op->flags & OP_FLUSH_BUFFER) &&
|
||||||
|
!op->version /* FIXME no CAS writeback */)
|
||||||
|
{
|
||||||
|
if (wb->writebacks_active >= client_max_writeback_iodepth)
|
||||||
|
{
|
||||||
|
// Writeback queue is full, postpone the operation
|
||||||
|
wb->writeback_overflow.push_back(op);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
// Just copy and acknowledge the operation
|
||||||
|
wb->copy_write(op, CACHE_DIRTY);
|
||||||
|
while (wb->writeback_bytes + op->len > client_max_buffered_bytes || wb->writeback_queue_size > client_max_buffered_ops)
|
||||||
|
{
|
||||||
|
// Initiate some writeback (asynchronously)
|
||||||
|
wb->start_writebacks(this, 1);
|
||||||
|
}
|
||||||
|
op->retval = op->len;
|
||||||
|
std::function<void(cluster_op_t*)>(op->callback)(op);
|
||||||
|
return;
|
||||||
|
}
|
||||||
if (op->opcode == OSD_OP_WRITE && !(op->flags & OP_IMMEDIATE_COMMIT))
|
if (op->opcode == OSD_OP_WRITE && !(op->flags & OP_IMMEDIATE_COMMIT))
|
||||||
{
|
{
|
||||||
if (!(op->flags & OP_FLUSH_BUFFER))
|
if (!(op->flags & OP_FLUSH_BUFFER))
|
||||||
{
|
{
|
||||||
copy_write(op, dirty_buffers);
|
wb->copy_write(op, CACHE_WRITTEN);
|
||||||
}
|
}
|
||||||
if (dirty_bytes >= client_max_dirty_bytes || dirty_ops >= client_max_dirty_ops)
|
if (dirty_bytes >= client_max_dirty_bytes || dirty_ops >= client_max_dirty_ops)
|
||||||
{
|
{
|
||||||
// Push an extra SYNC operation to flush previous writes
|
// Push an extra SYNC operation to flush previous writes
|
||||||
cluster_op_t *sync_op = new cluster_op_t;
|
cluster_op_t *sync_op = new cluster_op_t;
|
||||||
sync_op->opcode = OSD_OP_SYNC;
|
sync_op->opcode = OSD_OP_SYNC;
|
||||||
|
sync_op->flags = OP_FLUSH_BUFFER;
|
||||||
sync_op->callback = [](cluster_op_t* sync_op)
|
sync_op->callback = [](cluster_op_t* sync_op)
|
||||||
{
|
{
|
||||||
delete sync_op;
|
delete sync_op;
|
||||||
};
|
};
|
||||||
execute(sync_op);
|
execute_internal(sync_op);
|
||||||
}
|
}
|
||||||
dirty_bytes += op->len;
|
dirty_bytes += op->len;
|
||||||
dirty_ops++;
|
dirty_ops++;
|
||||||
}
|
}
|
||||||
else if (op->opcode == OSD_OP_SYNC)
|
else if (op->opcode == OSD_OP_SYNC)
|
||||||
{
|
{
|
||||||
|
// Flush the whole write-back queue first
|
||||||
|
if (!(op->flags & OP_FLUSH_BUFFER) && wb->writeback_overflow.size() > 0)
|
||||||
|
{
|
||||||
|
// Writeback queue is full, postpone the operation
|
||||||
|
wb->writeback_overflow.push_back(op);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if (wb->writeback_queue.size())
|
||||||
|
{
|
||||||
|
wb->start_writebacks(this, 0);
|
||||||
|
}
|
||||||
dirty_bytes = 0;
|
dirty_bytes = 0;
|
||||||
dirty_ops = 0;
|
dirty_ops = 0;
|
||||||
}
|
}
|
||||||
|
@ -494,7 +590,7 @@ void cluster_client_t::execute(cluster_op_t *op)
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
op_queue_tail = op_queue_head = op;
|
op_queue_tail = op_queue_head = op;
|
||||||
if (!(op->flags & OP_IMMEDIATE_COMMIT))
|
if (!(op->flags & OP_IMMEDIATE_COMMIT) || enable_writeback)
|
||||||
calc_wait(op);
|
calc_wait(op);
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
|
@ -568,136 +664,6 @@ void cluster_client_t::execute_raw(osd_num_t osd_num, osd_op_t *op)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
static std::map<object_id, cluster_buffer_t>::iterator find_dirty(uint64_t inode, uint64_t offset, std::map<object_id, cluster_buffer_t> & dirty_buffers)
|
|
||||||
{
|
|
||||||
auto dirty_it = dirty_buffers.lower_bound((object_id){
|
|
||||||
.inode = inode,
|
|
||||||
.stripe = offset,
|
|
||||||
});
|
|
||||||
while (dirty_it != dirty_buffers.begin())
|
|
||||||
{
|
|
||||||
dirty_it--;
|
|
||||||
if (dirty_it->first.inode != inode ||
|
|
||||||
(dirty_it->first.stripe + dirty_it->second.len) <= offset)
|
|
||||||
{
|
|
||||||
dirty_it++;
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return dirty_it;
|
|
||||||
}
|
|
||||||
|
|
||||||
void cluster_client_t::copy_write(cluster_op_t *op, std::map<object_id, cluster_buffer_t> & dirty_buffers)
|
|
||||||
{
|
|
||||||
// Save operation for replay when one of PGs goes out of sync
|
|
||||||
// (primary OSD drops our connection in this case)
|
|
||||||
auto dirty_it = find_dirty(op->inode, op->offset, dirty_buffers);
|
|
||||||
uint64_t pos = op->offset, len = op->len, iov_idx = 0, iov_pos = 0;
|
|
||||||
while (len > 0)
|
|
||||||
{
|
|
||||||
uint64_t new_len = 0;
|
|
||||||
if (dirty_it == dirty_buffers.end() || dirty_it->first.inode != op->inode)
|
|
||||||
{
|
|
||||||
new_len = len;
|
|
||||||
}
|
|
||||||
else if (dirty_it->first.stripe > pos)
|
|
||||||
{
|
|
||||||
new_len = dirty_it->first.stripe - pos;
|
|
||||||
if (new_len > len)
|
|
||||||
{
|
|
||||||
new_len = len;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (new_len > 0)
|
|
||||||
{
|
|
||||||
dirty_it = dirty_buffers.emplace_hint(dirty_it, (object_id){
|
|
||||||
.inode = op->inode,
|
|
||||||
.stripe = pos,
|
|
||||||
}, (cluster_buffer_t){
|
|
||||||
.buf = malloc_or_die(new_len),
|
|
||||||
.len = new_len,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
// FIXME: Split big buffers into smaller ones on overwrites. But this will require refcounting
|
|
||||||
dirty_it->second.state = CACHE_DIRTY;
|
|
||||||
uint64_t cur_len = (dirty_it->first.stripe + dirty_it->second.len - pos);
|
|
||||||
if (cur_len > len)
|
|
||||||
{
|
|
||||||
cur_len = len;
|
|
||||||
}
|
|
||||||
while (cur_len > 0 && iov_idx < op->iov.count)
|
|
||||||
{
|
|
||||||
unsigned iov_len = (op->iov.buf[iov_idx].iov_len - iov_pos);
|
|
||||||
if (iov_len <= cur_len)
|
|
||||||
{
|
|
||||||
memcpy((uint8_t*)dirty_it->second.buf + pos - dirty_it->first.stripe,
|
|
||||||
(uint8_t*)op->iov.buf[iov_idx].iov_base + iov_pos, iov_len);
|
|
||||||
pos += iov_len;
|
|
||||||
len -= iov_len;
|
|
||||||
cur_len -= iov_len;
|
|
||||||
iov_pos = 0;
|
|
||||||
iov_idx++;
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
memcpy((uint8_t*)dirty_it->second.buf + pos - dirty_it->first.stripe,
|
|
||||||
(uint8_t*)op->iov.buf[iov_idx].iov_base + iov_pos, cur_len);
|
|
||||||
pos += cur_len;
|
|
||||||
len -= cur_len;
|
|
||||||
iov_pos += cur_len;
|
|
||||||
cur_len = 0;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
dirty_it++;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
void cluster_client_t::flush_buffers(std::map<object_id, cluster_buffer_t>::iterator from_it,
|
|
||||||
std::map<object_id, cluster_buffer_t>::iterator to_it)
|
|
||||||
{
|
|
||||||
auto prev_it = std::prev(to_it);
|
|
||||||
cluster_op_t *op = new cluster_op_t;
|
|
||||||
op->flags = OSD_OP_IGNORE_READONLY|OP_FLUSH_BUFFER;
|
|
||||||
op->opcode = OSD_OP_WRITE;
|
|
||||||
op->cur_inode = op->inode = from_it->first.inode;
|
|
||||||
op->offset = from_it->first.stripe;
|
|
||||||
op->len = prev_it->first.stripe + prev_it->second.len - from_it->first.stripe;
|
|
||||||
uint32_t calc_len = 0;
|
|
||||||
uint64_t flush_id = ++last_flush_id;
|
|
||||||
for (auto it = from_it; it != to_it; it++)
|
|
||||||
{
|
|
||||||
it->second.state = CACHE_REPEATING;
|
|
||||||
it->second.flush_id = flush_id;
|
|
||||||
op->iov.push_back(it->second.buf, it->second.len);
|
|
||||||
calc_len += it->second.len;
|
|
||||||
}
|
|
||||||
assert(calc_len == op->len);
|
|
||||||
op->callback = [this, flush_id](cluster_op_t* op)
|
|
||||||
{
|
|
||||||
for (auto dirty_it = find_dirty(op->inode, op->offset, dirty_buffers);
|
|
||||||
dirty_it != dirty_buffers.end() && dirty_it->first.inode == op->inode &&
|
|
||||||
dirty_it->first.stripe < op->offset+op->len; dirty_it++)
|
|
||||||
{
|
|
||||||
if (dirty_it->second.flush_id == flush_id && dirty_it->second.state == CACHE_REPEATING)
|
|
||||||
{
|
|
||||||
dirty_it->second.flush_id = 0;
|
|
||||||
dirty_it->second.state = CACHE_DIRTY;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
delete op;
|
|
||||||
};
|
|
||||||
op->next = op_queue_head;
|
|
||||||
if (op_queue_head)
|
|
||||||
{
|
|
||||||
op_queue_head->prev = op;
|
|
||||||
op_queue_head = op;
|
|
||||||
}
|
|
||||||
else
|
|
||||||
op_queue_tail = op_queue_head = op;
|
|
||||||
inc_wait(op->opcode, op->flags, op->next, 1);
|
|
||||||
continue_rw(op);
|
|
||||||
}
|
|
||||||
|
|
||||||
int cluster_client_t::continue_rw(cluster_op_t *op)
|
int cluster_client_t::continue_rw(cluster_op_t *op)
|
||||||
{
|
{
|
||||||
if (op->state == 0)
|
if (op->state == 0)
|
||||||
|
@ -801,7 +767,8 @@ resume_2:
|
||||||
erase_op(op);
|
erase_op(op);
|
||||||
return 1;
|
return 1;
|
||||||
}
|
}
|
||||||
else if (op->retval != 0 && op->retval != -EPIPE && op->retval != -EIO && op->retval != -ENOSPC)
|
else if (op->retval != 0 && !(op->flags & OP_FLUSH_BUFFER) &&
|
||||||
|
op->retval != -EPIPE && op->retval != -EIO && op->retval != -ENOSPC)
|
||||||
{
|
{
|
||||||
// Fatal error (neither -EPIPE, -EIO nor -ENOSPC)
|
// Fatal error (neither -EPIPE, -EIO nor -ENOSPC)
|
||||||
// FIXME: Add a parameter to allow to not wait for EIOs (incomplete or corrupted objects) to heal
|
// FIXME: Add a parameter to allow to not wait for EIOs (incomplete or corrupted objects) to heal
|
||||||
|
@ -873,50 +840,6 @@ static void add_iov(int size, bool skip, cluster_op_t *op, int &iov_idx, size_t
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
static void copy_to_op(cluster_op_t *op, uint64_t offset, uint8_t *buf, uint64_t len, uint32_t bitmap_granularity)
|
|
||||||
{
|
|
||||||
if (op->opcode == OSD_OP_READ)
|
|
||||||
{
|
|
||||||
// Not OSD_OP_READ_BITMAP or OSD_OP_READ_CHAIN_BITMAP
|
|
||||||
int iov_idx = 0;
|
|
||||||
uint64_t cur_offset = op->offset;
|
|
||||||
while (iov_idx < op->iov.count && cur_offset+op->iov.buf[iov_idx].iov_len <= offset)
|
|
||||||
{
|
|
||||||
cur_offset += op->iov.buf[iov_idx].iov_len;
|
|
||||||
iov_idx++;
|
|
||||||
}
|
|
||||||
while (iov_idx < op->iov.count && cur_offset < offset+len)
|
|
||||||
{
|
|
||||||
auto & v = op->iov.buf[iov_idx];
|
|
||||||
auto begin = (cur_offset < offset ? offset : cur_offset);
|
|
||||||
auto end = (cur_offset+v.iov_len > offset+len ? offset+len : cur_offset+v.iov_len);
|
|
||||||
memcpy(
|
|
||||||
v.iov_base + begin - cur_offset,
|
|
||||||
buf + (cur_offset <= offset ? 0 : cur_offset-offset),
|
|
||||||
end - begin
|
|
||||||
);
|
|
||||||
cur_offset += v.iov_len;
|
|
||||||
iov_idx++;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// Set bitmap bits
|
|
||||||
int start_bit = (offset-op->offset)/bitmap_granularity;
|
|
||||||
int end_bit = (offset-op->offset+len)/bitmap_granularity;
|
|
||||||
for (int bit = start_bit; bit < end_bit;)
|
|
||||||
{
|
|
||||||
if (!(bit%8) && bit <= end_bit-8)
|
|
||||||
{
|
|
||||||
((uint8_t*)op->bitmap_buf)[bit/8] = 0xFF;
|
|
||||||
bit += 8;
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
((uint8_t*)op->bitmap_buf)[bit/8] |= (1 << (bit%8));
|
|
||||||
bit++;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
void cluster_client_t::slice_rw(cluster_op_t *op)
|
void cluster_client_t::slice_rw(cluster_op_t *op)
|
||||||
{
|
{
|
||||||
// Slice the request into individual object stripe requests
|
// Slice the request into individual object stripe requests
|
||||||
|
@ -945,52 +868,11 @@ void cluster_client_t::slice_rw(cluster_op_t *op)
|
||||||
int iov_idx = 0;
|
int iov_idx = 0;
|
||||||
size_t iov_pos = 0;
|
size_t iov_pos = 0;
|
||||||
int i = 0;
|
int i = 0;
|
||||||
bool dirty_copied = false;
|
|
||||||
if (dirty_buffers.size() && (op->opcode == OSD_OP_READ ||
|
|
||||||
op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP))
|
|
||||||
{
|
|
||||||
// We also have to return reads from CACHE_REPEATING buffers - they are not
|
// We also have to return reads from CACHE_REPEATING buffers - they are not
|
||||||
// guaranteed to be present on target OSDs at the moment of repeating
|
// guaranteed to be present on target OSDs at the moment of repeating
|
||||||
// And we're also free to return data from other cached buffers just
|
// And we're also free to return data from other cached buffers just
|
||||||
// because it's faster
|
// because it's faster
|
||||||
auto dirty_it = find_dirty(op->cur_inode, op->offset, dirty_buffers);
|
bool dirty_copied = wb->read_from_cache(op, pool_cfg.bitmap_granularity);
|
||||||
while (dirty_it != dirty_buffers.end() && dirty_it->first.inode == op->cur_inode &&
|
|
||||||
dirty_it->first.stripe < op->offset+op->len)
|
|
||||||
{
|
|
||||||
uint64_t begin = dirty_it->first.stripe, end = dirty_it->first.stripe + dirty_it->second.len;
|
|
||||||
if (begin < op->offset)
|
|
||||||
begin = op->offset;
|
|
||||||
if (end > op->offset+op->len)
|
|
||||||
end = op->offset+op->len;
|
|
||||||
bool skip_prev = true;
|
|
||||||
uint64_t cur = begin, prev = begin;
|
|
||||||
while (cur < end)
|
|
||||||
{
|
|
||||||
unsigned bmp_loc = (cur - op->offset)/pool_cfg.bitmap_granularity;
|
|
||||||
bool skip = (((*((uint8_t*)op->bitmap_buf + bmp_loc/8)) >> (bmp_loc%8)) & 0x1);
|
|
||||||
if (skip_prev != skip)
|
|
||||||
{
|
|
||||||
if (cur > prev && !skip)
|
|
||||||
{
|
|
||||||
// Copy data
|
|
||||||
dirty_copied = true;
|
|
||||||
copy_to_op(op, prev, (uint8_t*)dirty_it->second.buf + prev - dirty_it->first.stripe, cur-prev, pool_cfg.bitmap_granularity);
|
|
||||||
}
|
|
||||||
skip_prev = skip;
|
|
||||||
prev = cur;
|
|
||||||
}
|
|
||||||
cur += pool_cfg.bitmap_granularity;
|
|
||||||
}
|
|
||||||
assert(cur > prev);
|
|
||||||
if (!skip_prev)
|
|
||||||
{
|
|
||||||
// Copy data
|
|
||||||
dirty_copied = true;
|
|
||||||
copy_to_op(op, prev, (uint8_t*)dirty_it->second.buf + prev - dirty_it->first.stripe, cur-prev, pool_cfg.bitmap_granularity);
|
|
||||||
}
|
|
||||||
dirty_it++;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for (uint64_t stripe = first_stripe; stripe <= last_stripe; stripe += pg_block_size)
|
for (uint64_t stripe = first_stripe; stripe <= last_stripe; stripe += pg_block_size)
|
||||||
{
|
{
|
||||||
pg_num_t pg_num = (stripe/pool_cfg.pg_stripe_size) % pool_cfg.real_pg_count + 1; // like map_to_pg()
|
pg_num_t pg_num = (stripe/pool_cfg.pg_stripe_size) % pool_cfg.real_pg_count + 1; // like map_to_pg()
|
||||||
|
@ -1162,13 +1044,7 @@ int cluster_client_t::continue_sync(cluster_op_t *op)
|
||||||
do_it++;
|
do_it++;
|
||||||
}
|
}
|
||||||
// Post sync to affected OSDs
|
// Post sync to affected OSDs
|
||||||
for (auto & prev_op: dirty_buffers)
|
wb->fsync_start();
|
||||||
{
|
|
||||||
if (prev_op.second.state == CACHE_DIRTY)
|
|
||||||
{
|
|
||||||
prev_op.second.state = CACHE_FLUSHING;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
op->parts.resize(dirty_osds.size());
|
op->parts.resize(dirty_osds.size());
|
||||||
op->retval = 0;
|
op->retval = 0;
|
||||||
{
|
{
|
||||||
|
@ -1193,13 +1069,7 @@ resume_1:
|
||||||
}
|
}
|
||||||
if (op->retval != 0)
|
if (op->retval != 0)
|
||||||
{
|
{
|
||||||
for (auto uw_it = dirty_buffers.begin(); uw_it != dirty_buffers.end(); uw_it++)
|
wb->fsync_error();
|
||||||
{
|
|
||||||
if (uw_it->second.state == CACHE_FLUSHING)
|
|
||||||
{
|
|
||||||
uw_it->second.state = CACHE_DIRTY;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (op->retval == -EPIPE || op->retval == -EIO || op->retval == -ENOSPC)
|
if (op->retval == -EPIPE || op->retval == -EIO || op->retval == -ENOSPC)
|
||||||
{
|
{
|
||||||
// Retry later
|
// Retry later
|
||||||
|
@ -1213,16 +1083,7 @@ resume_1:
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
for (auto uw_it = dirty_buffers.begin(); uw_it != dirty_buffers.end(); )
|
wb->fsync_ok();
|
||||||
{
|
|
||||||
if (uw_it->second.state == CACHE_FLUSHING)
|
|
||||||
{
|
|
||||||
free(uw_it->second.buf);
|
|
||||||
dirty_buffers.erase(uw_it++);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
uw_it++;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
erase_op(op);
|
erase_op(op);
|
||||||
return 1;
|
return 1;
|
||||||
|
|
|
@ -8,6 +8,9 @@
|
||||||
|
|
||||||
#define DEFAULT_CLIENT_MAX_DIRTY_BYTES 32*1024*1024
|
#define DEFAULT_CLIENT_MAX_DIRTY_BYTES 32*1024*1024
|
||||||
#define DEFAULT_CLIENT_MAX_DIRTY_OPS 1024
|
#define DEFAULT_CLIENT_MAX_DIRTY_OPS 1024
|
||||||
|
#define DEFAULT_CLIENT_MAX_BUFFERED_BYTES 32*1024*1024
|
||||||
|
#define DEFAULT_CLIENT_MAX_BUFFERED_OPS 1024
|
||||||
|
#define DEFAULT_CLIENT_MAX_WRITEBACK_IODEPTH 256
|
||||||
#define INODE_LIST_DONE 1
|
#define INODE_LIST_DONE 1
|
||||||
#define INODE_LIST_HAS_UNSTABLE 2
|
#define INODE_LIST_HAS_UNSTABLE 2
|
||||||
#define OSD_OP_READ_BITMAP OSD_OP_SEC_READ_BMP
|
#define OSD_OP_READ_BITMAP OSD_OP_SEC_READ_BMP
|
||||||
|
@ -64,18 +67,12 @@ protected:
|
||||||
cluster_op_t *prev = NULL, *next = NULL;
|
cluster_op_t *prev = NULL, *next = NULL;
|
||||||
int prev_wait = 0;
|
int prev_wait = 0;
|
||||||
friend class cluster_client_t;
|
friend class cluster_client_t;
|
||||||
};
|
friend class writeback_cache_t;
|
||||||
|
|
||||||
struct cluster_buffer_t
|
|
||||||
{
|
|
||||||
void *buf;
|
|
||||||
uint64_t len;
|
|
||||||
int state;
|
|
||||||
uint64_t flush_id;
|
|
||||||
};
|
};
|
||||||
|
|
||||||
struct inode_list_t;
|
struct inode_list_t;
|
||||||
struct inode_list_osd_t;
|
struct inode_list_osd_t;
|
||||||
|
class writeback_cache_t;
|
||||||
|
|
||||||
// FIXME: Split into public and private interfaces
|
// FIXME: Split into public and private interfaces
|
||||||
class cluster_client_t
|
class cluster_client_t
|
||||||
|
@ -84,17 +81,23 @@ class cluster_client_t
|
||||||
ring_loop_t *ringloop;
|
ring_loop_t *ringloop;
|
||||||
|
|
||||||
std::map<pool_id_t, uint64_t> pg_counts;
|
std::map<pool_id_t, uint64_t> pg_counts;
|
||||||
// FIXME: Implement inmemory_commit mode. Note that it requires to return overlapping reads from memory.
|
// client_max_dirty_* is actually "max unsynced", for the case when immediate_commit is off
|
||||||
uint64_t client_max_dirty_bytes = 0;
|
uint64_t client_max_dirty_bytes = 0;
|
||||||
uint64_t client_max_dirty_ops = 0;
|
uint64_t client_max_dirty_ops = 0;
|
||||||
|
// writeback improves (1) small consecutive writes and (2) Q1 writes without fsync
|
||||||
|
bool enable_writeback = false;
|
||||||
|
// client_max_buffered_* is the real "dirty limit" - maximum amount of writes buffered in memory
|
||||||
|
uint64_t client_max_buffered_bytes = 0;
|
||||||
|
uint64_t client_max_buffered_ops = 0;
|
||||||
|
uint64_t client_max_writeback_iodepth = 0;
|
||||||
|
|
||||||
int log_level;
|
int log_level;
|
||||||
int up_wait_retry_interval = 500; // ms
|
int up_wait_retry_interval = 500; // ms
|
||||||
|
|
||||||
int retry_timeout_id = 0;
|
int retry_timeout_id = 0;
|
||||||
std::vector<cluster_op_t*> offline_ops;
|
std::vector<cluster_op_t*> offline_ops;
|
||||||
cluster_op_t *op_queue_head = NULL, *op_queue_tail = NULL;
|
cluster_op_t *op_queue_head = NULL, *op_queue_tail = NULL;
|
||||||
std::map<object_id, cluster_buffer_t> dirty_buffers;
|
writeback_cache_t *wb = NULL;
|
||||||
uint64_t last_flush_id = 0;
|
|
||||||
std::set<osd_num_t> dirty_osds;
|
std::set<osd_num_t> dirty_osds;
|
||||||
uint64_t dirty_bytes = 0, dirty_ops = 0;
|
uint64_t dirty_bytes = 0, dirty_ops = 0;
|
||||||
|
|
||||||
|
@ -124,10 +127,10 @@ public:
|
||||||
void execute_raw(osd_num_t osd_num, osd_op_t *op);
|
void execute_raw(osd_num_t osd_num, osd_op_t *op);
|
||||||
bool is_ready();
|
bool is_ready();
|
||||||
void on_ready(std::function<void(void)> fn);
|
void on_ready(std::function<void(void)> fn);
|
||||||
|
bool flush();
|
||||||
|
|
||||||
bool get_immediate_commit(uint64_t inode);
|
bool get_immediate_commit(uint64_t inode);
|
||||||
|
|
||||||
static void copy_write(cluster_op_t *op, std::map<object_id, cluster_buffer_t> & dirty_buffers);
|
|
||||||
void continue_ops(bool up_retry = false);
|
void continue_ops(bool up_retry = false);
|
||||||
inode_list_t *list_inode_start(inode_t inode,
|
inode_list_t *list_inode_start(inode_t inode,
|
||||||
std::function<void(inode_list_t* lst, std::set<object_id>&& objects, pg_num_t pg_num, osd_num_t primary_osd, int status)> callback);
|
std::function<void(inode_list_t* lst, std::set<object_id>&& objects, pg_num_t pg_num, osd_num_t primary_osd, int status)> callback);
|
||||||
|
@ -140,12 +143,12 @@ public:
|
||||||
|
|
||||||
protected:
|
protected:
|
||||||
bool affects_osd(uint64_t inode, uint64_t offset, uint64_t len, osd_num_t osd);
|
bool affects_osd(uint64_t inode, uint64_t offset, uint64_t len, osd_num_t osd);
|
||||||
void flush_buffers(std::map<object_id, cluster_buffer_t>::iterator from_it,
|
|
||||||
std::map<object_id, cluster_buffer_t>::iterator to_it);
|
|
||||||
void on_load_config_hook(json11::Json::object & config);
|
void on_load_config_hook(json11::Json::object & config);
|
||||||
void on_load_pgs_hook(bool success);
|
void on_load_pgs_hook(bool success);
|
||||||
void on_change_hook(std::map<std::string, etcd_kv_t> & changes);
|
void on_change_hook(std::map<std::string, etcd_kv_t> & changes);
|
||||||
void on_change_osd_state_hook(uint64_t peer_osd);
|
void on_change_osd_state_hook(uint64_t peer_osd);
|
||||||
|
void execute_internal(cluster_op_t *op);
|
||||||
|
void unshift_op(cluster_op_t *op);
|
||||||
int continue_rw(cluster_op_t *op);
|
int continue_rw(cluster_op_t *op);
|
||||||
bool check_rw(cluster_op_t *op);
|
bool check_rw(cluster_op_t *op);
|
||||||
void slice_rw(cluster_op_t *op);
|
void slice_rw(cluster_op_t *op);
|
||||||
|
@ -161,4 +164,6 @@ protected:
|
||||||
void continue_listing(inode_list_t *lst);
|
void continue_listing(inode_list_t *lst);
|
||||||
void send_list(inode_list_osd_t *cur_list);
|
void send_list(inode_list_osd_t *cur_list);
|
||||||
void continue_raw_ops(osd_num_t peer_osd);
|
void continue_raw_ops(osd_num_t peer_osd);
|
||||||
|
|
||||||
|
friend class writeback_cache_t;
|
||||||
};
|
};
|
||||||
|
|
|
@ -0,0 +1,57 @@
|
||||||
|
// Copyright (c) Vitaliy Filippov, 2019+
|
||||||
|
// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
|
||||||
|
|
||||||
|
#pragma once
|
||||||
|
|
||||||
|
#include "cluster_client.h"
|
||||||
|
|
||||||
|
#define SCRAP_BUFFER_SIZE 4*1024*1024
|
||||||
|
#define PART_SENT 1
|
||||||
|
#define PART_DONE 2
|
||||||
|
#define PART_ERROR 4
|
||||||
|
#define PART_RETRY 8
|
||||||
|
#define CACHE_DIRTY 1
|
||||||
|
#define CACHE_WRITTEN 2
|
||||||
|
#define CACHE_FLUSHING 3
|
||||||
|
#define CACHE_REPEATING 4
|
||||||
|
#define OP_FLUSH_BUFFER 0x02
|
||||||
|
#define OP_IMMEDIATE_COMMIT 0x04
|
||||||
|
|
||||||
|
struct cluster_buffer_t
|
||||||
|
{
|
||||||
|
uint8_t *buf;
|
||||||
|
uint64_t len;
|
||||||
|
int state;
|
||||||
|
uint64_t flush_id;
|
||||||
|
uint64_t *refcnt;
|
||||||
|
};
|
||||||
|
|
||||||
|
typedef std::map<object_id, cluster_buffer_t>::iterator dirty_buf_it_t;
|
||||||
|
|
||||||
|
class writeback_cache_t
|
||||||
|
{
|
||||||
|
public:
|
||||||
|
uint64_t writeback_bytes = 0;
|
||||||
|
int writeback_queue_size = 0;
|
||||||
|
int writebacks_active = 0;
|
||||||
|
uint64_t last_flush_id = 0;
|
||||||
|
|
||||||
|
std::map<object_id, cluster_buffer_t> dirty_buffers;
|
||||||
|
std::vector<cluster_op_t*> writeback_overflow;
|
||||||
|
std::vector<object_id> writeback_queue;
|
||||||
|
std::multimap<uint64_t, uint64_t*> flushed_buffers; // flush_id => refcnt
|
||||||
|
|
||||||
|
~writeback_cache_t();
|
||||||
|
dirty_buf_it_t find_dirty(uint64_t inode, uint64_t offset);
|
||||||
|
bool is_left_merged(dirty_buf_it_t dirty_it);
|
||||||
|
bool is_right_merged(dirty_buf_it_t dirty_it);
|
||||||
|
bool is_merged(const dirty_buf_it_t & dirty_it);
|
||||||
|
void copy_write(cluster_op_t *op, int state);
|
||||||
|
int repeat_ops_for(cluster_client_t *cli, osd_num_t peer_osd);
|
||||||
|
void start_writebacks(cluster_client_t *cli, int count);
|
||||||
|
bool read_from_cache(cluster_op_t *op, uint32_t bitmap_granularity);
|
||||||
|
void flush_buffers(cluster_client_t *cli, dirty_buf_it_t from_it, dirty_buf_it_t to_it);
|
||||||
|
void fsync_start();
|
||||||
|
void fsync_error();
|
||||||
|
void fsync_ok();
|
||||||
|
};
|
|
@ -0,0 +1,498 @@
|
||||||
|
// Copyright (c) Vitaliy Filippov, 2019+
|
||||||
|
// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
|
||||||
|
|
||||||
|
#include <cassert>
|
||||||
|
|
||||||
|
#include "cluster_client_impl.h"
|
||||||
|
|
||||||
|
writeback_cache_t::~writeback_cache_t()
|
||||||
|
{
|
||||||
|
for (auto & bp: dirty_buffers)
|
||||||
|
{
|
||||||
|
if (!--(*bp.second.refcnt))
|
||||||
|
{
|
||||||
|
free(bp.second.refcnt); // refcnt is allocated with the buffer
|
||||||
|
}
|
||||||
|
}
|
||||||
|
dirty_buffers.clear();
|
||||||
|
}
|
||||||
|
|
||||||
|
dirty_buf_it_t writeback_cache_t::find_dirty(uint64_t inode, uint64_t offset)
|
||||||
|
{
|
||||||
|
auto dirty_it = dirty_buffers.lower_bound((object_id){
|
||||||
|
.inode = inode,
|
||||||
|
.stripe = offset,
|
||||||
|
});
|
||||||
|
while (dirty_it != dirty_buffers.begin())
|
||||||
|
{
|
||||||
|
dirty_it--;
|
||||||
|
if (dirty_it->first.inode != inode ||
|
||||||
|
(dirty_it->first.stripe + dirty_it->second.len) <= offset)
|
||||||
|
{
|
||||||
|
dirty_it++;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return dirty_it;
|
||||||
|
}
|
||||||
|
|
||||||
|
bool writeback_cache_t::is_left_merged(dirty_buf_it_t dirty_it)
|
||||||
|
{
|
||||||
|
if (dirty_it != dirty_buffers.begin())
|
||||||
|
{
|
||||||
|
auto prev_it = dirty_it;
|
||||||
|
prev_it--;
|
||||||
|
if (prev_it->first.inode == dirty_it->first.inode &&
|
||||||
|
prev_it->first.stripe+prev_it->second.len == dirty_it->first.stripe &&
|
||||||
|
prev_it->second.state == CACHE_DIRTY)
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
bool writeback_cache_t::is_right_merged(dirty_buf_it_t dirty_it)
|
||||||
|
{
|
||||||
|
auto next_it = dirty_it;
|
||||||
|
next_it++;
|
||||||
|
if (next_it != dirty_buffers.end() &&
|
||||||
|
next_it->first.inode == dirty_it->first.inode &&
|
||||||
|
next_it->first.stripe == dirty_it->first.stripe+dirty_it->second.len &&
|
||||||
|
next_it->second.state == CACHE_DIRTY)
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
bool writeback_cache_t::is_merged(const dirty_buf_it_t & dirty_it)
|
||||||
|
{
|
||||||
|
return is_left_merged(dirty_it) || is_right_merged(dirty_it);
|
||||||
|
}
|
||||||
|
|
||||||
|
void writeback_cache_t::copy_write(cluster_op_t *op, int state)
|
||||||
|
{
|
||||||
|
// Save operation for replay when one of PGs goes out of sync
|
||||||
|
// (primary OSD drops our connection in this case)
|
||||||
|
// ...or just save it for writeback if write buffering is enabled
|
||||||
|
if (op->len == 0)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
auto dirty_it = find_dirty(op->inode, op->offset);
|
||||||
|
auto new_end = op->offset + op->len;
|
||||||
|
while (dirty_it != dirty_buffers.end() &&
|
||||||
|
dirty_it->first.inode == op->inode &&
|
||||||
|
dirty_it->first.stripe < op->offset+op->len)
|
||||||
|
{
|
||||||
|
assert(dirty_it->first.stripe + dirty_it->second.len > op->offset);
|
||||||
|
// Remove overlapping part(s) of buffers
|
||||||
|
auto old_end = dirty_it->first.stripe + dirty_it->second.len;
|
||||||
|
if (dirty_it->first.stripe < op->offset)
|
||||||
|
{
|
||||||
|
if (old_end > new_end)
|
||||||
|
{
|
||||||
|
// Split into end and start
|
||||||
|
dirty_it->second.len = op->offset - dirty_it->first.stripe;
|
||||||
|
dirty_it = dirty_buffers.emplace_hint(dirty_it, (object_id){
|
||||||
|
.inode = op->inode,
|
||||||
|
.stripe = new_end,
|
||||||
|
}, (cluster_buffer_t){
|
||||||
|
.buf = dirty_it->second.buf + new_end - dirty_it->first.stripe,
|
||||||
|
.len = old_end - new_end,
|
||||||
|
.state = dirty_it->second.state,
|
||||||
|
.flush_id = dirty_it->second.flush_id,
|
||||||
|
.refcnt = dirty_it->second.refcnt,
|
||||||
|
});
|
||||||
|
(*dirty_it->second.refcnt)++;
|
||||||
|
if (dirty_it->second.state == CACHE_DIRTY)
|
||||||
|
{
|
||||||
|
writeback_bytes -= op->len;
|
||||||
|
writeback_queue_size++;
|
||||||
|
}
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
// Only leave the beginning
|
||||||
|
if (dirty_it->second.state == CACHE_DIRTY)
|
||||||
|
{
|
||||||
|
writeback_bytes -= old_end - op->offset;
|
||||||
|
if (is_left_merged(dirty_it) && !is_right_merged(dirty_it))
|
||||||
|
{
|
||||||
|
writeback_queue_size++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
dirty_it->second.len = op->offset - dirty_it->first.stripe;
|
||||||
|
dirty_it++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
else if (old_end > new_end)
|
||||||
|
{
|
||||||
|
// Only leave the end
|
||||||
|
if (dirty_it->second.state == CACHE_DIRTY)
|
||||||
|
{
|
||||||
|
writeback_bytes -= new_end - dirty_it->first.stripe;
|
||||||
|
if (!is_left_merged(dirty_it) && is_right_merged(dirty_it))
|
||||||
|
{
|
||||||
|
writeback_queue_size++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
auto new_dirty_it = dirty_buffers.emplace_hint(dirty_it, (object_id){
|
||||||
|
.inode = op->inode,
|
||||||
|
.stripe = new_end,
|
||||||
|
}, (cluster_buffer_t){
|
||||||
|
.buf = dirty_it->second.buf + new_end - dirty_it->first.stripe,
|
||||||
|
.len = old_end - new_end,
|
||||||
|
.state = dirty_it->second.state,
|
||||||
|
.flush_id = dirty_it->second.flush_id,
|
||||||
|
.refcnt = dirty_it->second.refcnt,
|
||||||
|
});
|
||||||
|
dirty_buffers.erase(dirty_it);
|
||||||
|
dirty_it = new_dirty_it;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
// Remove the whole buffer
|
||||||
|
if (dirty_it->second.state == CACHE_DIRTY && !is_merged(dirty_it))
|
||||||
|
{
|
||||||
|
writeback_bytes -= dirty_it->second.len;
|
||||||
|
assert(writeback_queue_size > 0);
|
||||||
|
writeback_queue_size--;
|
||||||
|
}
|
||||||
|
if (!--(*dirty_it->second.refcnt))
|
||||||
|
{
|
||||||
|
free(dirty_it->second.refcnt);
|
||||||
|
}
|
||||||
|
dirty_buffers.erase(dirty_it++);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Overlapping buffers are removed, just insert the new one
|
||||||
|
uint64_t *refcnt = (uint64_t*)malloc_or_die(sizeof(uint64_t) + op->len);
|
||||||
|
uint8_t *buf = (uint8_t*)refcnt + sizeof(uint64_t);
|
||||||
|
*refcnt = 1;
|
||||||
|
dirty_it = dirty_buffers.emplace_hint(dirty_it, (object_id){
|
||||||
|
.inode = op->inode,
|
||||||
|
.stripe = op->offset,
|
||||||
|
}, (cluster_buffer_t){
|
||||||
|
.buf = buf,
|
||||||
|
.len = op->len,
|
||||||
|
.state = state,
|
||||||
|
.refcnt = refcnt,
|
||||||
|
});
|
||||||
|
if (state == CACHE_DIRTY)
|
||||||
|
{
|
||||||
|
writeback_bytes += op->len;
|
||||||
|
// Track consecutive write-back operations
|
||||||
|
if (!is_merged(dirty_it))
|
||||||
|
{
|
||||||
|
// <writeback_queue> is OK to contain more than actual number of consecutive
|
||||||
|
// requests as long as it doesn't miss anything. But <writeback_queue_size>
|
||||||
|
// is always calculated correctly.
|
||||||
|
writeback_queue_size++;
|
||||||
|
writeback_queue.push_back((object_id){
|
||||||
|
.inode = op->inode,
|
||||||
|
.stripe = op->offset,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
uint64_t pos = 0, len = op->len, iov_idx = 0;
|
||||||
|
while (len > 0 && iov_idx < op->iov.count)
|
||||||
|
{
|
||||||
|
auto & iov = op->iov.buf[iov_idx];
|
||||||
|
memcpy(buf + pos, iov.iov_base, iov.iov_len);
|
||||||
|
pos += iov.iov_len;
|
||||||
|
iov_idx++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
int writeback_cache_t::repeat_ops_for(cluster_client_t *cli, osd_num_t peer_osd)
|
||||||
|
{
|
||||||
|
int repeated = 0;
|
||||||
|
if (dirty_buffers.size())
|
||||||
|
{
|
||||||
|
// peer_osd just dropped connection
|
||||||
|
// determine WHICH dirty_buffers are now obsolete and repeat them
|
||||||
|
for (auto wr_it = dirty_buffers.begin(), flush_it = wr_it, last_it = wr_it; ; )
|
||||||
|
{
|
||||||
|
bool end = wr_it == dirty_buffers.end();
|
||||||
|
bool flush_this = !end && wr_it->second.state != CACHE_REPEATING &&
|
||||||
|
cli->affects_osd(wr_it->first.inode, wr_it->first.stripe, wr_it->second.len, peer_osd);
|
||||||
|
if (flush_it != wr_it && (end || !flush_this ||
|
||||||
|
wr_it->first.inode != flush_it->first.inode ||
|
||||||
|
wr_it->first.stripe != last_it->first.stripe+last_it->second.len))
|
||||||
|
{
|
||||||
|
repeated++;
|
||||||
|
flush_buffers(cli, flush_it, wr_it);
|
||||||
|
flush_it = wr_it;
|
||||||
|
}
|
||||||
|
if (end)
|
||||||
|
break;
|
||||||
|
last_it = wr_it;
|
||||||
|
wr_it++;
|
||||||
|
if (!flush_this)
|
||||||
|
flush_it = wr_it;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return repeated;
|
||||||
|
}
|
||||||
|
|
||||||
|
void writeback_cache_t::flush_buffers(cluster_client_t *cli, dirty_buf_it_t from_it, dirty_buf_it_t to_it)
|
||||||
|
{
|
||||||
|
auto prev_it = to_it;
|
||||||
|
prev_it--;
|
||||||
|
bool is_writeback = from_it->second.state == CACHE_DIRTY;
|
||||||
|
cluster_op_t *op = new cluster_op_t;
|
||||||
|
op->flags = OSD_OP_IGNORE_READONLY|OP_FLUSH_BUFFER;
|
||||||
|
op->opcode = OSD_OP_WRITE;
|
||||||
|
op->cur_inode = op->inode = from_it->first.inode;
|
||||||
|
op->offset = from_it->first.stripe;
|
||||||
|
op->len = prev_it->first.stripe + prev_it->second.len - from_it->first.stripe;
|
||||||
|
uint32_t calc_len = 0;
|
||||||
|
uint64_t flush_id = ++last_flush_id;
|
||||||
|
for (auto it = from_it; it != to_it; it++)
|
||||||
|
{
|
||||||
|
it->second.state = CACHE_REPEATING;
|
||||||
|
it->second.flush_id = flush_id;
|
||||||
|
(*it->second.refcnt)++;
|
||||||
|
flushed_buffers.emplace(flush_id, it->second.refcnt);
|
||||||
|
op->iov.push_back(it->second.buf, it->second.len);
|
||||||
|
calc_len += it->second.len;
|
||||||
|
}
|
||||||
|
assert(calc_len == op->len);
|
||||||
|
writebacks_active++;
|
||||||
|
op->callback = [this, cli, flush_id](cluster_op_t* op)
|
||||||
|
{
|
||||||
|
// Buffer flushes should be always retried, regardless of the error,
|
||||||
|
// so they should never result in an error here
|
||||||
|
assert(op->retval == op->len);
|
||||||
|
for (auto fl_it = flushed_buffers.find(flush_id);
|
||||||
|
fl_it != flushed_buffers.end() && fl_it->first == flush_id; )
|
||||||
|
{
|
||||||
|
if (!--(*fl_it->second)) // refcnt
|
||||||
|
{
|
||||||
|
free(fl_it->second);
|
||||||
|
}
|
||||||
|
flushed_buffers.erase(fl_it++);
|
||||||
|
}
|
||||||
|
for (auto dirty_it = find_dirty(op->inode, op->offset);
|
||||||
|
dirty_it != dirty_buffers.end() && dirty_it->first.inode == op->inode &&
|
||||||
|
dirty_it->first.stripe < op->offset+op->len; dirty_it++)
|
||||||
|
{
|
||||||
|
if (dirty_it->second.flush_id == flush_id && dirty_it->second.state == CACHE_REPEATING)
|
||||||
|
{
|
||||||
|
dirty_it->second.flush_id = 0;
|
||||||
|
dirty_it->second.state = CACHE_WRITTEN;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
delete op;
|
||||||
|
writebacks_active--;
|
||||||
|
// We can't call execute_internal because it affects an invalid copy of the list here
|
||||||
|
// (erase_op remembers `next` after writeback callback)
|
||||||
|
};
|
||||||
|
if (is_writeback)
|
||||||
|
{
|
||||||
|
cli->execute_internal(op);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
// Insert repeated flushes into the beginning
|
||||||
|
cli->unshift_op(op);
|
||||||
|
cli->continue_rw(op);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
void writeback_cache_t::start_writebacks(cluster_client_t *cli, int count)
|
||||||
|
{
|
||||||
|
if (!writeback_queue.size())
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
std::vector<object_id> queue_copy;
|
||||||
|
queue_copy.swap(writeback_queue);
|
||||||
|
int started = 0, i = 0;
|
||||||
|
for (i = 0; i < queue_copy.size() && (!count || started < count); i++)
|
||||||
|
{
|
||||||
|
object_id & req = queue_copy[i];
|
||||||
|
auto dirty_it = find_dirty(req.inode, req.stripe);
|
||||||
|
if (dirty_it == dirty_buffers.end() ||
|
||||||
|
dirty_it->first.inode != req.inode ||
|
||||||
|
dirty_it->second.state != CACHE_DIRTY)
|
||||||
|
{
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
auto from_it = dirty_it;
|
||||||
|
uint64_t off = dirty_it->first.stripe;
|
||||||
|
while (from_it != dirty_buffers.begin())
|
||||||
|
{
|
||||||
|
from_it--;
|
||||||
|
if (from_it->second.state != CACHE_DIRTY ||
|
||||||
|
from_it->first.inode != req.inode ||
|
||||||
|
from_it->first.stripe+from_it->second.len != off)
|
||||||
|
{
|
||||||
|
from_it++;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
off = from_it->first.stripe;
|
||||||
|
}
|
||||||
|
off = dirty_it->first.stripe + dirty_it->second.len;
|
||||||
|
auto to_it = dirty_it;
|
||||||
|
to_it++;
|
||||||
|
while (to_it != dirty_buffers.end())
|
||||||
|
{
|
||||||
|
if (to_it->second.state != CACHE_DIRTY ||
|
||||||
|
to_it->first.inode != req.inode ||
|
||||||
|
to_it->first.stripe != off)
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
off = to_it->first.stripe + to_it->second.len;
|
||||||
|
to_it++;
|
||||||
|
}
|
||||||
|
started++;
|
||||||
|
assert(writeback_queue_size > 0);
|
||||||
|
writeback_queue_size--;
|
||||||
|
writeback_bytes -= off - from_it->first.stripe;
|
||||||
|
flush_buffers(cli, from_it, to_it);
|
||||||
|
}
|
||||||
|
queue_copy.erase(queue_copy.begin(), queue_copy.begin()+i);
|
||||||
|
if (writeback_queue.size())
|
||||||
|
{
|
||||||
|
queue_copy.insert(queue_copy.end(), writeback_queue.begin(), writeback_queue.end());
|
||||||
|
}
|
||||||
|
queue_copy.swap(writeback_queue);
|
||||||
|
}
|
||||||
|
|
||||||
|
static void copy_to_op(cluster_op_t *op, uint64_t offset, uint8_t *buf, uint64_t len, uint32_t bitmap_granularity)
|
||||||
|
{
|
||||||
|
if (op->opcode == OSD_OP_READ)
|
||||||
|
{
|
||||||
|
// Not OSD_OP_READ_BITMAP or OSD_OP_READ_CHAIN_BITMAP
|
||||||
|
int iov_idx = 0;
|
||||||
|
uint64_t cur_offset = op->offset;
|
||||||
|
while (iov_idx < op->iov.count && cur_offset+op->iov.buf[iov_idx].iov_len <= offset)
|
||||||
|
{
|
||||||
|
cur_offset += op->iov.buf[iov_idx].iov_len;
|
||||||
|
iov_idx++;
|
||||||
|
}
|
||||||
|
while (iov_idx < op->iov.count && cur_offset < offset+len)
|
||||||
|
{
|
||||||
|
auto & v = op->iov.buf[iov_idx];
|
||||||
|
auto begin = (cur_offset < offset ? offset : cur_offset);
|
||||||
|
auto end = (cur_offset+v.iov_len > offset+len ? offset+len : cur_offset+v.iov_len);
|
||||||
|
memcpy(
|
||||||
|
v.iov_base + begin - cur_offset,
|
||||||
|
buf + (cur_offset <= offset ? 0 : cur_offset-offset),
|
||||||
|
end - begin
|
||||||
|
);
|
||||||
|
cur_offset += v.iov_len;
|
||||||
|
iov_idx++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Set bitmap bits
|
||||||
|
int start_bit = (offset-op->offset)/bitmap_granularity;
|
||||||
|
int end_bit = (offset-op->offset+len)/bitmap_granularity;
|
||||||
|
for (int bit = start_bit; bit < end_bit;)
|
||||||
|
{
|
||||||
|
if (!(bit%8) && bit <= end_bit-8)
|
||||||
|
{
|
||||||
|
((uint8_t*)op->bitmap_buf)[bit/8] = 0xFF;
|
||||||
|
bit += 8;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
((uint8_t*)op->bitmap_buf)[bit/8] |= (1 << (bit%8));
|
||||||
|
bit++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
bool writeback_cache_t::read_from_cache(cluster_op_t *op, uint32_t bitmap_granularity)
|
||||||
|
{
|
||||||
|
bool dirty_copied = false;
|
||||||
|
if (dirty_buffers.size() && (op->opcode == OSD_OP_READ ||
|
||||||
|
op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP))
|
||||||
|
{
|
||||||
|
// We also have to return reads from CACHE_REPEATING buffers - they are not
|
||||||
|
// guaranteed to be present on target OSDs at the moment of repeating
|
||||||
|
// And we're also free to return data from other cached buffers just
|
||||||
|
// because it's faster
|
||||||
|
auto dirty_it = find_dirty(op->cur_inode, op->offset);
|
||||||
|
while (dirty_it != dirty_buffers.end() && dirty_it->first.inode == op->cur_inode &&
|
||||||
|
dirty_it->first.stripe < op->offset+op->len)
|
||||||
|
{
|
||||||
|
uint64_t begin = dirty_it->first.stripe, end = dirty_it->first.stripe + dirty_it->second.len;
|
||||||
|
if (begin < op->offset)
|
||||||
|
begin = op->offset;
|
||||||
|
if (end > op->offset+op->len)
|
||||||
|
end = op->offset+op->len;
|
||||||
|
bool skip_prev = true;
|
||||||
|
uint64_t cur = begin, prev = begin;
|
||||||
|
while (cur < end)
|
||||||
|
{
|
||||||
|
unsigned bmp_loc = (cur - op->offset)/bitmap_granularity;
|
||||||
|
bool skip = (((*((uint8_t*)op->bitmap_buf + bmp_loc/8)) >> (bmp_loc%8)) & 0x1);
|
||||||
|
if (skip_prev != skip)
|
||||||
|
{
|
||||||
|
if (cur > prev && !skip)
|
||||||
|
{
|
||||||
|
// Copy data
|
||||||
|
dirty_copied = true;
|
||||||
|
copy_to_op(op, prev, dirty_it->second.buf + prev - dirty_it->first.stripe, cur-prev, bitmap_granularity);
|
||||||
|
}
|
||||||
|
skip_prev = skip;
|
||||||
|
prev = cur;
|
||||||
|
}
|
||||||
|
cur += bitmap_granularity;
|
||||||
|
}
|
||||||
|
assert(cur > prev);
|
||||||
|
if (!skip_prev)
|
||||||
|
{
|
||||||
|
// Copy data
|
||||||
|
dirty_copied = true;
|
||||||
|
copy_to_op(op, prev, dirty_it->second.buf + prev - dirty_it->first.stripe, cur-prev, bitmap_granularity);
|
||||||
|
}
|
||||||
|
dirty_it++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return dirty_copied;
|
||||||
|
}
|
||||||
|
|
||||||
|
void writeback_cache_t::fsync_start()
|
||||||
|
{
|
||||||
|
for (auto & prev_op: dirty_buffers)
|
||||||
|
{
|
||||||
|
if (prev_op.second.state == CACHE_WRITTEN)
|
||||||
|
{
|
||||||
|
prev_op.second.state = CACHE_FLUSHING;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
void writeback_cache_t::fsync_error()
|
||||||
|
{
|
||||||
|
for (auto & prev_op: dirty_buffers)
|
||||||
|
{
|
||||||
|
if (prev_op.second.state == CACHE_FLUSHING)
|
||||||
|
{
|
||||||
|
prev_op.second.state = CACHE_WRITTEN;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
void writeback_cache_t::fsync_ok()
|
||||||
|
{
|
||||||
|
for (auto uw_it = dirty_buffers.begin(); uw_it != dirty_buffers.end(); )
|
||||||
|
{
|
||||||
|
if (uw_it->second.state == CACHE_FLUSHING)
|
||||||
|
{
|
||||||
|
if (!--(*uw_it->second.refcnt))
|
||||||
|
free(uw_it->second.refcnt);
|
||||||
|
dirty_buffers.erase(uw_it++);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
uw_it++;
|
||||||
|
}
|
||||||
|
}
|
|
@ -24,6 +24,7 @@
|
||||||
#include <netinet/tcp.h>
|
#include <netinet/tcp.h>
|
||||||
|
|
||||||
#include <vector>
|
#include <vector>
|
||||||
|
#include <string>
|
||||||
|
|
||||||
#include "vitastor_c.h"
|
#include "vitastor_c.h"
|
||||||
#include "fio_headers.h"
|
#include "fio_headers.h"
|
||||||
|
@ -203,6 +204,15 @@ static void watch_callback(void *opaque, long watch)
|
||||||
bsd->watch = (void*)watch;
|
bsd->watch = (void*)watch;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static void opt_push(std::vector<char *> & options, const char *opt, const char *value)
|
||||||
|
{
|
||||||
|
if (value)
|
||||||
|
{
|
||||||
|
options.push_back(strdup(opt));
|
||||||
|
options.push_back(strdup(value));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
static int sec_setup(struct thread_data *td)
|
static int sec_setup(struct thread_data *td)
|
||||||
{
|
{
|
||||||
sec_options *o = (sec_options*)td->eo;
|
sec_options *o = (sec_options*)td->eo;
|
||||||
|
@ -254,8 +264,27 @@ static int sec_setup(struct thread_data *td)
|
||||||
{
|
{
|
||||||
o->inode = 0;
|
o->inode = 0;
|
||||||
}
|
}
|
||||||
bsd->cli = vitastor_c_create_uring(o->config_path, o->etcd_host, o->etcd_prefix,
|
std::vector<char *> options;
|
||||||
o->use_rdma, o->rdma_device, o->rdma_port_num, o->rdma_gid_index, o->rdma_mtu, o->cluster_log);
|
opt_push(options, "config_path", o->config_path);
|
||||||
|
opt_push(options, "etcd_address", o->etcd_host);
|
||||||
|
opt_push(options, "etcd_prefix", o->etcd_prefix);
|
||||||
|
if (o->use_rdma != -1)
|
||||||
|
opt_push(options, "use_rdma", std::to_string(o->use_rdma).c_str());
|
||||||
|
opt_push(options, "rdma_device", o->rdma_device);
|
||||||
|
if (o->rdma_port_num)
|
||||||
|
opt_push(options, "rdma_port_num", std::to_string(o->rdma_port_num).c_str());
|
||||||
|
if (o->rdma_gid_index)
|
||||||
|
opt_push(options, "rdma_gid_index", std::to_string(o->rdma_gid_index).c_str());
|
||||||
|
if (o->rdma_mtu)
|
||||||
|
opt_push(options, "rdma_mtu", std::to_string(o->rdma_mtu).c_str());
|
||||||
|
if (o->cluster_log)
|
||||||
|
opt_push(options, "log_level", std::to_string(o->cluster_log).c_str());
|
||||||
|
// allow writeback caching if -direct is not set
|
||||||
|
opt_push(options, "client_writeback_allowed", td->o.odirect ? "0" : "1");
|
||||||
|
bsd->cli = vitastor_c_create_uring_json((const char**)options.data(), options.size());
|
||||||
|
for (auto opt: options)
|
||||||
|
free(opt);
|
||||||
|
options.clear();
|
||||||
if (o->image)
|
if (o->image)
|
||||||
{
|
{
|
||||||
bsd->watch = NULL;
|
bsd->watch = NULL;
|
||||||
|
|
|
@ -55,3 +55,10 @@ json11::Json::object osd_messenger_t::merge_configs(const json11::Json::object &
|
||||||
{
|
{
|
||||||
return cli_config;
|
return cli_config;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
bool json_is_true(const json11::Json & val)
|
||||||
|
{
|
||||||
|
if (val.is_string())
|
||||||
|
return val == "true" || val == "yes" || val == "1";
|
||||||
|
return val.bool_value();
|
||||||
|
}
|
||||||
|
|
|
@ -22,4 +22,10 @@ public:
|
||||||
void submit()
|
void submit()
|
||||||
{
|
{
|
||||||
}
|
}
|
||||||
|
void wait()
|
||||||
|
{
|
||||||
|
}
|
||||||
|
void loop()
|
||||||
|
{
|
||||||
|
}
|
||||||
};
|
};
|
||||||
|
|
|
@ -341,6 +341,7 @@ public:
|
||||||
ringloop->loop();
|
ringloop->loop();
|
||||||
ringloop->wait();
|
ringloop->wait();
|
||||||
}
|
}
|
||||||
|
cli->flush();
|
||||||
delete cli;
|
delete cli;
|
||||||
delete epmgr;
|
delete epmgr;
|
||||||
delete ringloop;
|
delete ringloop;
|
||||||
|
|
|
@ -34,7 +34,10 @@ nfs_proxy_t::~nfs_proxy_t()
|
||||||
if (cmd)
|
if (cmd)
|
||||||
delete cmd;
|
delete cmd;
|
||||||
if (cli)
|
if (cli)
|
||||||
|
{
|
||||||
|
cli->flush();
|
||||||
delete cli;
|
delete cli;
|
||||||
|
}
|
||||||
if (epmgr)
|
if (epmgr)
|
||||||
delete epmgr;
|
delete epmgr;
|
||||||
if (ringloop)
|
if (ringloop)
|
||||||
|
@ -261,16 +264,8 @@ void nfs_proxy_t::run(json11::Json cfg)
|
||||||
ringloop->loop();
|
ringloop->loop();
|
||||||
ringloop->wait();
|
ringloop->wait();
|
||||||
}
|
}
|
||||||
/*// Sync at the end
|
|
||||||
cluster_op_t *close_sync = new cluster_op_t;
|
|
||||||
close_sync->opcode = OSD_OP_SYNC;
|
|
||||||
close_sync->callback = [&stop](cluster_op_t *op)
|
|
||||||
{
|
|
||||||
stop = true;
|
|
||||||
delete op;
|
|
||||||
};
|
|
||||||
cli->execute(close_sync);*/
|
|
||||||
// Destroy the client
|
// Destroy the client
|
||||||
|
cli->flush();
|
||||||
delete cli;
|
delete cli;
|
||||||
delete epmgr;
|
delete epmgr;
|
||||||
delete ringloop;
|
delete ringloop;
|
||||||
|
|
|
@ -186,10 +186,12 @@ json11::Json osd_t::get_statistics()
|
||||||
if (bs)
|
if (bs)
|
||||||
{
|
{
|
||||||
st["blockstore_ready"] = bs->is_started();
|
st["blockstore_ready"] = bs->is_started();
|
||||||
st["data_block_size"] = (uint64_t)bs->get_block_size();
|
|
||||||
st["size"] = bs->get_block_count() * bs->get_block_size();
|
st["size"] = bs->get_block_count() * bs->get_block_size();
|
||||||
st["free"] = bs->get_free_block_count() * bs->get_block_size();
|
st["free"] = bs->get_free_block_count() * bs->get_block_size();
|
||||||
}
|
}
|
||||||
|
st["data_block_size"] = (uint64_t)bs_block_size;
|
||||||
|
st["bitmap_granularity"] = (uint64_t)bs_bitmap_granularity;
|
||||||
|
st["immediate_commit"] = immediate_commit == IMMEDIATE_ALL ? "all" : (immediate_commit == IMMEDIATE_SMALL ? "small" : "none");
|
||||||
st["host"] = self_state["host"];
|
st["host"] = self_state["host"];
|
||||||
json11::Json::object op_stats, subop_stats;
|
json11::Json::object op_stats, subop_stats;
|
||||||
for (int i = OSD_OP_MIN; i <= OSD_OP_MAX; i++)
|
for (int i = OSD_OP_MIN; i <= OSD_OP_MAX; i++)
|
||||||
|
|
|
@ -388,6 +388,43 @@ static void vitastor_aio_set_fd_handler(void *vcli, int fd, int unused1, IOHandl
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
typedef struct str_array
|
||||||
|
{
|
||||||
|
const char **items;
|
||||||
|
int len, alloc;
|
||||||
|
} str_array;
|
||||||
|
|
||||||
|
static void strarray_push(str_array *a, const char *str)
|
||||||
|
{
|
||||||
|
if (a->len >= a->alloc)
|
||||||
|
{
|
||||||
|
a->alloc = !a->alloc ? 4 : 2*a->alloc;
|
||||||
|
a->items = (const char**)realloc(a->items, a->alloc*sizeof(char*));
|
||||||
|
if (!a->items)
|
||||||
|
{
|
||||||
|
fprintf(stderr, "bad alloc\n");
|
||||||
|
abort();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
a->items[a->len++] = str;
|
||||||
|
}
|
||||||
|
|
||||||
|
static void strarray_push_kv(str_array *a, const char *key, const char *value)
|
||||||
|
{
|
||||||
|
if (key && value)
|
||||||
|
{
|
||||||
|
strarray_push(a, key);
|
||||||
|
strarray_push(a, value);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
static void strarray_free(str_array *a)
|
||||||
|
{
|
||||||
|
free(a->items);
|
||||||
|
a->items = NULL;
|
||||||
|
a->len = a->alloc = 0;
|
||||||
|
}
|
||||||
|
|
||||||
static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, Error **errp)
|
static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, Error **errp)
|
||||||
{
|
{
|
||||||
VitastorRPC task;
|
VitastorRPC task;
|
||||||
|
@ -406,6 +443,36 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
|
||||||
client->rdma_gid_index = qdict_get_try_int(options, "rdma-gid-index", 0);
|
client->rdma_gid_index = qdict_get_try_int(options, "rdma-gid-index", 0);
|
||||||
client->rdma_mtu = qdict_get_try_int(options, "rdma-mtu", 0);
|
client->rdma_mtu = qdict_get_try_int(options, "rdma-mtu", 0);
|
||||||
client->ctx = bdrv_get_aio_context(bs);
|
client->ctx = bdrv_get_aio_context(bs);
|
||||||
|
str_array opt = {};
|
||||||
|
strarray_push_kv(&opt, "config_path", qdict_get_try_str(options, "config-path"));
|
||||||
|
strarray_push_kv(&opt, "etcd_address", qdict_get_try_str(options, "etcd-host"));
|
||||||
|
strarray_push_kv(&opt, "etcd_prefix", qdict_get_try_str(options, "etcd-prefix"));
|
||||||
|
strarray_push_kv(&opt, "use_rdma", qdict_get_try_str(options, "use-rdma"));
|
||||||
|
strarray_push_kv(&opt, "rdma_device", qdict_get_try_str(options, "rdma-device"));
|
||||||
|
strarray_push_kv(&opt, "rdma_port_num", qdict_get_try_str(options, "rdma-port-num"));
|
||||||
|
strarray_push_kv(&opt, "rdma_gid_index", qdict_get_try_str(options, "rdma-gid-index"));
|
||||||
|
strarray_push_kv(&opt, "rdma_mtu", qdict_get_try_str(options, "rdma-mtu"));
|
||||||
|
strarray_push_kv(&opt, "client_writeback_allowed", (flags & BDRV_O_NOCACHE) ? "0" : "1");
|
||||||
|
client->proxy = vitastor_c_create_uring_json(opt.items, opt.len);
|
||||||
|
strarray_free(&opt);
|
||||||
|
if (client->proxy)
|
||||||
|
{
|
||||||
|
client->uring_eventfd = vitastor_c_uring_register_eventfd(client->proxy);
|
||||||
|
if (client->uring_eventfd < 0)
|
||||||
|
{
|
||||||
|
fprintf(stderr, "vitastor: failed to create io_uring eventfd: %s\n", strerror(errno));
|
||||||
|
error_setg(errp, "failed to create io_uring eventfd");
|
||||||
|
vitastor_close(bs);
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
universal_aio_set_fd_handler(client->ctx, client->uring_eventfd, vitastor_uring_handler, NULL, client);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
// Writeback cache is unusable without io_uring because the client can't correctly flush on exit
|
||||||
|
fprintf(stderr, "vitastor: failed to create io_uring: %s - I/O will be slower%s\n",
|
||||||
|
strerror(errno), (flags & BDRV_O_NOCACHE ? "" : " and writeback cache will be disabled"));
|
||||||
|
client->uring_eventfd = -1;
|
||||||
#if defined VITASTOR_C_API_VERSION && VITASTOR_C_API_VERSION >= 2
|
#if defined VITASTOR_C_API_VERSION && VITASTOR_C_API_VERSION >= 2
|
||||||
client->proxy = vitastor_c_create_qemu_uring(
|
client->proxy = vitastor_c_create_qemu_uring(
|
||||||
vitastor_aio_set_fd_handler, client, client->config_path, client->etcd_host, client->etcd_prefix,
|
vitastor_aio_set_fd_handler, client, client->config_path, client->etcd_host, client->etcd_prefix,
|
||||||
|
@ -415,12 +482,10 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
|
||||||
{
|
{
|
||||||
fprintf(stderr, "vitastor: failed to create io_uring: %s - I/O will be slower\n", strerror(errno));
|
fprintf(stderr, "vitastor: failed to create io_uring: %s - I/O will be slower\n", strerror(errno));
|
||||||
client->uring_eventfd = -1;
|
client->uring_eventfd = -1;
|
||||||
#endif
|
|
||||||
client->proxy = vitastor_c_create_qemu(
|
client->proxy = vitastor_c_create_qemu(
|
||||||
vitastor_aio_set_fd_handler, client, client->config_path, client->etcd_host, client->etcd_prefix,
|
vitastor_aio_set_fd_handler, client, client->config_path, client->etcd_host, client->etcd_prefix,
|
||||||
client->use_rdma, client->rdma_device, client->rdma_port_num, client->rdma_gid_index, client->rdma_mtu, 0
|
client->use_rdma, client->rdma_device, client->rdma_port_num, client->rdma_gid_index, client->rdma_mtu, 0
|
||||||
);
|
);
|
||||||
#if defined VITASTOR_C_API_VERSION && VITASTOR_C_API_VERSION >= 2
|
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
|
@ -434,7 +499,13 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
|
||||||
}
|
}
|
||||||
universal_aio_set_fd_handler(client->ctx, client->uring_eventfd, vitastor_uring_handler, NULL, client);
|
universal_aio_set_fd_handler(client->ctx, client->uring_eventfd, vitastor_uring_handler, NULL, client);
|
||||||
}
|
}
|
||||||
|
#else
|
||||||
|
client->proxy = vitastor_c_create_qemu(
|
||||||
|
vitastor_aio_set_fd_handler, client, client->config_path, client->etcd_host, client->etcd_prefix,
|
||||||
|
client->use_rdma, client->rdma_device, client->rdma_port_num, client->rdma_gid_index, client->rdma_mtu, 0
|
||||||
|
);
|
||||||
#endif
|
#endif
|
||||||
|
}
|
||||||
image = client->image = g_strdup(qdict_get_try_str(options, "image"));
|
image = client->image = g_strdup(qdict_get_try_str(options, "image"));
|
||||||
client->readonly = (flags & BDRV_O_RDWR) ? 1 : 0;
|
client->readonly = (flags & BDRV_O_RDWR) ? 1 : 0;
|
||||||
// Get image metadata (size and readonly flag) or just wait until the client is ready
|
// Get image metadata (size and readonly flag) or just wait until the client is ready
|
||||||
|
|
|
@ -4,7 +4,7 @@
|
||||||
#include <stdio.h>
|
#include <stdio.h>
|
||||||
#include <stdlib.h>
|
#include <stdlib.h>
|
||||||
#include <assert.h>
|
#include <assert.h>
|
||||||
#include "cluster_client.h"
|
#include "cluster_client_impl.h"
|
||||||
|
|
||||||
void configure_single_pg_pool(cluster_client_t *cli)
|
void configure_single_pg_pool(cluster_client_t *cli)
|
||||||
{
|
{
|
||||||
|
@ -47,11 +47,11 @@ void configure_single_pg_pool(cluster_client_t *cli)
|
||||||
cli->st_cli.on_change_hook(changes);
|
cli->st_cli.on_change_hook(changes);
|
||||||
}
|
}
|
||||||
|
|
||||||
int *test_write(cluster_client_t *cli, uint64_t offset, uint64_t len, uint8_t c, std::function<void()> cb = NULL)
|
int *test_write(cluster_client_t *cli, uint64_t offset, uint64_t len, uint8_t c, std::function<void()> cb = NULL, bool instant = false)
|
||||||
{
|
{
|
||||||
printf("Post write %lx+%lx\n", offset, len);
|
printf("Post write %lx+%lx\n", offset, len);
|
||||||
int *r = new int;
|
int *r = new int;
|
||||||
*r = -1;
|
*r = instant ? -2 : -1;
|
||||||
cluster_op_t *op = new cluster_op_t();
|
cluster_op_t *op = new cluster_op_t();
|
||||||
op->opcode = OSD_OP_WRITE;
|
op->opcode = OSD_OP_WRITE;
|
||||||
op->inode = 0x1000000000001;
|
op->inode = 0x1000000000001;
|
||||||
|
@ -72,6 +72,13 @@ int *test_write(cluster_client_t *cli, uint64_t offset, uint64_t len, uint8_t c,
|
||||||
cb();
|
cb();
|
||||||
};
|
};
|
||||||
cli->execute(op);
|
cli->execute(op);
|
||||||
|
if (instant)
|
||||||
|
{
|
||||||
|
long res = *r;
|
||||||
|
assert(*r >= 0);
|
||||||
|
delete r;
|
||||||
|
return (int*)res;
|
||||||
|
}
|
||||||
return r;
|
return r;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -160,6 +167,13 @@ osd_op_t *find_op(cluster_client_t *cli, osd_num_t osd_num, uint64_t opcode, uin
|
||||||
}
|
}
|
||||||
op_it++;
|
op_it++;
|
||||||
}
|
}
|
||||||
|
op_it = cli->msgr.clients[peer_fd]->sent_ops.begin();
|
||||||
|
while (op_it != cli->msgr.clients[peer_fd]->sent_ops.end())
|
||||||
|
{
|
||||||
|
printf("Found opcode %lu offset %lx size %x\n", op_it->second->req.hdr.opcode, op_it->second->req.rw.offset, op_it->second->req.rw.len);
|
||||||
|
op_it++;
|
||||||
|
}
|
||||||
|
printf("Not found opcode %lu offset %lx size %lx\n", opcode, offset, len);
|
||||||
return NULL;
|
return NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -341,7 +355,7 @@ void test1()
|
||||||
|
|
||||||
void test2()
|
void test2()
|
||||||
{
|
{
|
||||||
std::map<object_id, cluster_buffer_t> unsynced_writes;
|
writeback_cache_t *wb = new writeback_cache_t();
|
||||||
cluster_op_t *op = new cluster_op_t();
|
cluster_op_t *op = new cluster_op_t();
|
||||||
op->opcode = OSD_OP_WRITE;
|
op->opcode = OSD_OP_WRITE;
|
||||||
op->inode = 1;
|
op->inode = 1;
|
||||||
|
@ -350,19 +364,19 @@ void test2()
|
||||||
op->iov.push_back(malloc_or_die(4096*1024), 4096);
|
op->iov.push_back(malloc_or_die(4096*1024), 4096);
|
||||||
// 0-4k = 0x55
|
// 0-4k = 0x55
|
||||||
memset(op->iov.buf[0].iov_base, 0x55, op->iov.buf[0].iov_len);
|
memset(op->iov.buf[0].iov_base, 0x55, op->iov.buf[0].iov_len);
|
||||||
cluster_client_t::copy_write(op, unsynced_writes);
|
wb->copy_write(op, CACHE_WRITTEN);
|
||||||
// 8k-12k = 0x66
|
// 8k-12k = 0x66
|
||||||
op->offset = 8192;
|
op->offset = 8192;
|
||||||
memset(op->iov.buf[0].iov_base, 0x66, op->iov.buf[0].iov_len);
|
memset(op->iov.buf[0].iov_base, 0x66, op->iov.buf[0].iov_len);
|
||||||
cluster_client_t::copy_write(op, unsynced_writes);
|
wb->copy_write(op, CACHE_WRITTEN);
|
||||||
// 4k-1M+4k = 0x77
|
// 4k-1M+4k = 0x77
|
||||||
op->len = op->iov.buf[0].iov_len = 1048576;
|
op->len = op->iov.buf[0].iov_len = 1048576;
|
||||||
op->offset = 4096;
|
op->offset = 4096;
|
||||||
memset(op->iov.buf[0].iov_base, 0x77, op->iov.buf[0].iov_len);
|
memset(op->iov.buf[0].iov_base, 0x77, op->iov.buf[0].iov_len);
|
||||||
cluster_client_t::copy_write(op, unsynced_writes);
|
wb->copy_write(op, CACHE_WRITTEN);
|
||||||
// check it
|
// check it
|
||||||
assert(unsynced_writes.size() == 4);
|
assert(wb->dirty_buffers.size() == 2);
|
||||||
auto uit = unsynced_writes.begin();
|
auto uit = wb->dirty_buffers.begin();
|
||||||
int i;
|
int i;
|
||||||
assert(uit->first.inode == 1);
|
assert(uit->first.inode == 1);
|
||||||
assert(uit->first.stripe == 0);
|
assert(uit->first.stripe == 0);
|
||||||
|
@ -372,35 +386,106 @@ void test2()
|
||||||
uit++;
|
uit++;
|
||||||
assert(uit->first.inode == 1);
|
assert(uit->first.inode == 1);
|
||||||
assert(uit->first.stripe == 4096);
|
assert(uit->first.stripe == 4096);
|
||||||
assert(uit->second.len == 4096);
|
assert(uit->second.len == 1048576);
|
||||||
for (i = 0; i < uit->second.len && ((uint8_t*)uit->second.buf)[i] == 0x77; i++) {}
|
|
||||||
assert(i == uit->second.len);
|
|
||||||
uit++;
|
|
||||||
assert(uit->first.inode == 1);
|
|
||||||
assert(uit->first.stripe == 8192);
|
|
||||||
assert(uit->second.len == 4096);
|
|
||||||
for (i = 0; i < uit->second.len && ((uint8_t*)uit->second.buf)[i] == 0x77; i++) {}
|
|
||||||
assert(i == uit->second.len);
|
|
||||||
uit++;
|
|
||||||
assert(uit->first.inode == 1);
|
|
||||||
assert(uit->first.stripe == 12*1024);
|
|
||||||
assert(uit->second.len == 1016*1024);
|
|
||||||
for (i = 0; i < uit->second.len && ((uint8_t*)uit->second.buf)[i] == 0x77; i++) {}
|
for (i = 0; i < uit->second.len && ((uint8_t*)uit->second.buf)[i] == 0x77; i++) {}
|
||||||
assert(i == uit->second.len);
|
assert(i == uit->second.len);
|
||||||
uit++;
|
uit++;
|
||||||
// free memory
|
// free memory
|
||||||
free(op->iov.buf[0].iov_base);
|
free(op->iov.buf[0].iov_base);
|
||||||
delete op;
|
delete op;
|
||||||
for (auto p: unsynced_writes)
|
delete wb;
|
||||||
{
|
|
||||||
free(p.second.buf);
|
|
||||||
}
|
|
||||||
printf("[ok] copy_write test\n");
|
printf("[ok] copy_write test\n");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void test_writeback()
|
||||||
|
{
|
||||||
|
json11::Json config = json11::Json::object {
|
||||||
|
{ "client_enable_writeback", true },
|
||||||
|
{ "client_writeback_allowed", true },
|
||||||
|
{ "client_max_buffered_bytes", 1024*1024 },
|
||||||
|
{ "client_max_buffered_ops", 2 },
|
||||||
|
{ "client_max_writeback_iodepth", 2 },
|
||||||
|
{ "client_max_dirty_bytes", 1024*1024 },
|
||||||
|
{ "client_max_dirty_ops", 2 },
|
||||||
|
};
|
||||||
|
timerfd_manager_t *tfd = new timerfd_manager_t([](int fd, bool wr, std::function<void(int, int)> callback){});
|
||||||
|
cluster_client_t *cli = new cluster_client_t(NULL, tfd, config);
|
||||||
|
|
||||||
|
configure_single_pg_pool(cli);
|
||||||
|
pretend_connected(cli, 1);
|
||||||
|
|
||||||
|
// Check that 3 consecutive writes are merged by writeback
|
||||||
|
assert((long)test_write(cli, 0, 4096, 0x55, NULL, true) == 1);
|
||||||
|
check_op_count(cli, 1, 0);
|
||||||
|
assert((long)test_write(cli, 4096, 4096, 0x55, NULL, true) == 1);
|
||||||
|
check_op_count(cli, 1, 0);
|
||||||
|
assert((long)test_write(cli, 8192, 4096, 0x55, NULL, true) == 1);
|
||||||
|
check_op_count(cli, 1, 0);
|
||||||
|
|
||||||
|
assert((long)test_write(cli, 1024*1024, 4096, 0x66, NULL, true) == 1);
|
||||||
|
check_op_count(cli, 1, 0);
|
||||||
|
|
||||||
|
// 3rd and 4th writes should trigger 1 writeback each
|
||||||
|
assert((long)test_write(cli, 2*1024*1024, 4096, 0x66, NULL, true) == 1);
|
||||||
|
check_op_count(cli, 1, 1);
|
||||||
|
assert((long)test_write(cli, 3*1024*1024, 4096, 0x66, NULL, true) == 1);
|
||||||
|
check_op_count(cli, 1, 2);
|
||||||
|
|
||||||
|
// 5th write should be postponed until at least 1 writeback is completed
|
||||||
|
int *r1 = test_write(cli, 4*1024*1024, 4096, 0x67, NULL);
|
||||||
|
check_op_count(cli, 1, 2);
|
||||||
|
can_complete(r1);
|
||||||
|
pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 3*4096), 0);
|
||||||
|
check_completed(r1);
|
||||||
|
// autosync because max_dirty_ops=2, flush waits for sync
|
||||||
|
check_op_count(cli, 1, 1);
|
||||||
|
pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 1024*1024, 4096), 0);
|
||||||
|
check_op_count(cli, 1, 1);
|
||||||
|
pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
|
||||||
|
check_op_count(cli, 1, 1);
|
||||||
|
pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 2*1024*1024, 4096), 0);
|
||||||
|
check_op_count(cli, 1, 0);
|
||||||
|
|
||||||
|
int *r2 = test_sync(cli);
|
||||||
|
check_op_count(cli, 1, 1);
|
||||||
|
pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 3*1024*1024, 4096), 0);
|
||||||
|
check_op_count(cli, 1, 1);
|
||||||
|
// autosync because max_dirty_ops=2, flush waits for sync
|
||||||
|
pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
|
||||||
|
check_op_count(cli, 1, 1);
|
||||||
|
pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 4*1024*1024, 4096), 0);
|
||||||
|
check_op_count(cli, 1, 1);
|
||||||
|
can_complete(r2);
|
||||||
|
pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
|
||||||
|
check_completed(r2);
|
||||||
|
|
||||||
|
// Check cutting of the beginning and end
|
||||||
|
assert((long)test_write(cli, 0, 32768, 0x55, NULL, true) == 1);
|
||||||
|
check_op_count(cli, 1, 0);
|
||||||
|
assert((long)test_write(cli, 32768, 32768, 0x56, NULL, true) == 1);
|
||||||
|
check_op_count(cli, 1, 0);
|
||||||
|
assert((long)test_write(cli, 16384, 32768, 0x57, NULL, true) == 1);
|
||||||
|
check_op_count(cli, 1, 0);
|
||||||
|
assert((long)test_write(cli, 16384+4096, 32768-4096, 0x58, NULL, true) == 1);
|
||||||
|
check_op_count(cli, 1, 0);
|
||||||
|
r2 = test_sync(cli);
|
||||||
|
check_op_count(cli, 1, 1);
|
||||||
|
pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 65536), 0);
|
||||||
|
check_op_count(cli, 1, 1);
|
||||||
|
can_complete(r2);
|
||||||
|
pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
|
||||||
|
check_completed(r2);
|
||||||
|
|
||||||
|
// Free client
|
||||||
|
delete cli;
|
||||||
|
delete tfd;
|
||||||
|
printf("[ok] writeback test\n");
|
||||||
|
}
|
||||||
|
|
||||||
int main(int narg, char *args[])
|
int main(int narg, char *args[])
|
||||||
{
|
{
|
||||||
test1();
|
test1();
|
||||||
test2();
|
test2();
|
||||||
|
test_writeback();
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|
|
@ -129,6 +129,7 @@ again:
|
||||||
if (exp.it_value.tv_sec < 0 || exp.it_value.tv_sec == 0 && exp.it_value.tv_nsec <= 0)
|
if (exp.it_value.tv_sec < 0 || exp.it_value.tv_sec == 0 && exp.it_value.tv_nsec <= 0)
|
||||||
{
|
{
|
||||||
// It already happened
|
// It already happened
|
||||||
|
// FIXME: Postpone to setImmediate/BH to avoid reenterability problems
|
||||||
trigger_nearest();
|
trigger_nearest();
|
||||||
goto again;
|
goto again;
|
||||||
}
|
}
|
||||||
|
|
|
@ -19,10 +19,10 @@ fi
|
||||||
|
|
||||||
if [ "$IMMEDIATE_COMMIT" != "" ]; then
|
if [ "$IMMEDIATE_COMMIT" != "" ]; then
|
||||||
NO_SAME="--journal_no_same_sector_overwrites true --journal_sector_buffer_count 1024 --disable_data_fsync 1 --immediate_commit all --log_level 10"
|
NO_SAME="--journal_no_same_sector_overwrites true --journal_sector_buffer_count 1024 --disable_data_fsync 1 --immediate_commit all --log_level 10"
|
||||||
$ETCDCTL put /vitastor/config/global '{"recovery_queue_depth":1,"osd_out_time":1,"immediate_commit":"all"}'
|
$ETCDCTL put /vitastor/config/global '{"recovery_queue_depth":1,"osd_out_time":1,"immediate_commit":"all","client_enable_writeback":true}'
|
||||||
else
|
else
|
||||||
NO_SAME="--journal_sector_buffer_count 1024 --log_level 10"
|
NO_SAME="--journal_sector_buffer_count 1024 --log_level 10"
|
||||||
$ETCDCTL put /vitastor/config/global '{"recovery_queue_depth":1,"osd_out_time":1}'
|
$ETCDCTL put /vitastor/config/global '{"recovery_queue_depth":1,"osd_out_time":1,"client_enable_writeback":true}'
|
||||||
fi
|
fi
|
||||||
|
|
||||||
start_osd_on()
|
start_osd_on()
|
||||||
|
|
Loading…
Reference in New Issue