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

Add wait_up_timeout support to cluster_client and use it in vitastor-cli rm-data & merge

Browse Source
This commit is contained in:
Vitaliy Filippov
2025-01-01 17:57:58 +03:00
parent 5c5119aba4
commit 40b8a8b0da
4 changed files with 91 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/cluster_client.cpp
+68 -20
View File
@@ -6,6 +6,10 @@
#include "cluster_client_impl.h"
#include "json_util.h"
#define TRY_SEND_OFFLINE 0
#define TRY_SEND_CONNECTING 1
#define TRY_SEND_OK 2
cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json config)
{
wb = new writeback_cache_t();
@@ -146,12 +150,12 @@ void cluster_client_t::unshift_op(cluster_op_t *op)
void cluster_client_t::calc_wait(cluster_op_t *op)
{
op->prev_wait = 0;
if (op->opcode == OSD_OP_WRITE)
if (op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE)
{
for (auto prev = op->prev; prev; prev = prev->prev)
{
if (prev->opcode == OSD_OP_SYNC ||
prev->opcode == OSD_OP_WRITE && !(op->flags & OP_FLUSH_BUFFER) && (prev->flags & OP_FLUSH_BUFFER))
(prev->opcode == OSD_OP_WRITE || prev->opcode == OSD_OP_DELETE) && !(op->flags & OP_FLUSH_BUFFER) && (prev->flags & OP_FLUSH_BUFFER))
{
op->prev_wait++;
}
@@ -163,7 +167,8 @@ void cluster_client_t::calc_wait(cluster_op_t *op)
{
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) || enable_writeback))
if (prev->opcode == OSD_OP_SYNC || (prev->opcode == OSD_OP_WRITE || prev->opcode == OSD_OP_DELETE) &&
(!(prev->flags & OP_IMMEDIATE_COMMIT) || enable_writeback))
{
op->prev_wait++;
}
@@ -179,7 +184,7 @@ void cluster_client_t::calc_wait(cluster_op_t *op)
void cluster_client_t::inc_wait(uint64_t opcode, uint64_t flags, cluster_op_t *next, int inc)
{
if (opcode != OSD_OP_WRITE && opcode != OSD_OP_SYNC)
if (opcode != OSD_OP_WRITE && opcode != OSD_OP_DELETE && opcode != OSD_OP_SYNC)
{
return;
}
@@ -188,10 +193,10 @@ void cluster_client_t::inc_wait(uint64_t opcode, uint64_t flags, cluster_op_t *n
while (next)
{
auto n2 = next->next;
if (opcode == OSD_OP_WRITE
if ((opcode == OSD_OP_WRITE || opcode == OSD_OP_DELETE)
? (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_SYNC || next->opcode == OSD_OP_WRITE))
(next->opcode == OSD_OP_WRITE || next->opcode == OSD_OP_DELETE) && (flags & OP_FLUSH_BUFFER) && !(next->flags & OP_FLUSH_BUFFER))
: (next->opcode == OSD_OP_SYNC || next->opcode == OSD_OP_WRITE || next->opcode == OSD_OP_DELETE))
{
next->prev_wait += inc;
assert(next->prev_wait >= 0);
@@ -456,7 +461,7 @@ void cluster_client_t::on_change_pool_config_hook()
// And now they have to be resliced!
for (auto op = op_queue_head; op; op = op->next)
{
if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_READ ||
if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE || op->opcode == OSD_OP_READ ||
op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP) &&
INODE_POOL(op->cur_inode) == pool_item.first)
{
@@ -583,7 +588,8 @@ bool cluster_client_t::flush()
void cluster_client_t::execute(cluster_op_t *op)
{
if (op->opcode != OSD_OP_SYNC && op->opcode != OSD_OP_READ &&
op->opcode != OSD_OP_READ_BITMAP && op->opcode != OSD_OP_READ_CHAIN_BITMAP && op->opcode != OSD_OP_WRITE)
op->opcode != OSD_OP_READ_BITMAP && op->opcode != OSD_OP_READ_CHAIN_BITMAP &&
op->opcode != OSD_OP_WRITE && op->opcode != OSD_OP_DELETE)
{
op->retval = -EINVAL;
auto cb = std::move(op->callback);
@@ -595,7 +601,7 @@ void cluster_client_t::execute(cluster_op_t *op)
offline_ops.push_back(op);
return;
}
op->flags = op->flags & OSD_OP_IGNORE_READONLY; // the only allowed flag
op->flags = op->flags & (OSD_OP_IGNORE_READONLY | OSD_OP_WAIT_UP_TIMEOUT); // allowed client flags
execute_internal(op);
}
@@ -639,6 +645,7 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
}
if (op->opcode == OSD_OP_WRITE && !(op->flags & OP_IMMEDIATE_COMMIT))
{
// FIXME: Deletes should also be remembered and then repeated when OSD disconnects
if (!(op->flags & OP_FLUSH_BUFFER) && !op->version /* no CAS write-repeat */)
{
uint64_t flush_id = ++wb->last_flush_id;
@@ -801,9 +808,48 @@ resume_1:
{
if (!(op->parts[i].flags & PART_SENT))
{
if (!try_send(op, i))
int is_ok = try_send(op, i);
if (is_ok != TRY_SEND_OK)
{
// We'll need to retry again
if (op->flags & OSD_OP_WAIT_UP_TIMEOUT)
{
if (is_ok != TRY_SEND_OFFLINE)
{
// Reset "wait_up" timer
op->wait_up_until = {};
}
else if (!op->wait_up_until.tv_sec && !client_wait_up_timeout)
{
// Don't wait for the PG to come up at all and fail
op->parts[i].flags |= PART_ERROR;
if (!op->retval)
op->retval = -ETIMEDOUT;
break;
}
else if (!op->wait_up_until.tv_sec)
{
// Set "wait_up" timer
clock_gettime(CLOCK_REALTIME, &op->wait_up_until);
op->wait_up_until.tv_sec += client_wait_up_timeout;
}
else
{
// Check if the timeout expired
timespec tv;
clock_gettime(CLOCK_REALTIME, &tv);
if (tv.tv_sec > op->wait_up_until.tv_sec ||
tv.tv_sec == op->wait_up_until.tv_sec &&
tv.tv_nsec > op->wait_up_until.tv_nsec)
{
// Fail
op->parts[i].flags |= PART_ERROR;
if (!op->retval)
op->retval = -ETIMEDOUT;
break;
}
}
}
if (op->parts[i].flags & PART_RETRY)
{
op->retry_after = client_retry_interval;
@@ -1077,7 +1123,7 @@ bool cluster_client_t::affects_osd(uint64_t inode, uint64_t offset, uint64_t len
return false;
}
bool cluster_client_t::try_send(cluster_op_t *op, int i)
int cluster_client_t::try_send(cluster_op_t *op, int i)
{
if (!msgr_initialized)
{
@@ -1133,14 +1179,15 @@ bool cluster_client_t::try_send(cluster_op_t *op, int i)
};
part->op.iov = part->iov;
msgr.outbox_push(&part->op);
return true;
return TRY_SEND_OK;
}
else if (msgr.wanted_peers.find(primary_osd) == msgr.wanted_peers.end())
{
msgr.connect_peer(primary_osd, st_cli.peer_states[primary_osd]);
return TRY_SEND_CONNECTING;
}
}
return false;
return TRY_SEND_OFFLINE;
}
int cluster_client_t::continue_sync(cluster_op_t *op)
@@ -1212,13 +1259,12 @@ resume_1:
void cluster_client_t::send_sync(cluster_op_t *op, cluster_op_part_t *part)
{
auto peer_it = msgr.osd_peer_fds.find(part->osd_num);
assert(peer_it != msgr.osd_peer_fds.end());
auto peer_fd = msgr.osd_peer_fds.at(part->osd_num);
part->flags |= PART_SENT;
op->inflight_count++;
part->op = (osd_op_t){
.op_type = OSD_OP_OUT,
.peer_fd = peer_it->second,
.peer_fd = peer_fd,
.req = {
.hdr = {
.magic = SECONDARY_OSD_OP_MAGIC,
@@ -1252,9 +1298,11 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
{
// Operation failed, retry
part->flags |= PART_ERROR;
if (!op->retval || op->retval == -EPIPE || part->op.reply.hdr.retval == -EIO)
if (!op->retval || op->retval == -EPIPE ||
part->op.reply.hdr.retval == -ENOSPC && op->retval == -ETIMEDOUT ||
part->op.reply.hdr.retval == -EIO)
{
// Error priority: EIO > ENOSPC > EPIPE
// Error priority: EIO > ENOSPC > ETIMEDOUT > EPIPE
op->retval = part->op.reply.hdr.retval;
}
int stop_fd = -1;
@@ -1317,7 +1365,7 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
op->version = op->parts.size() == 1 ? part->op.reply.rw.version : 0;
}
}
else if (op->opcode == OSD_OP_WRITE)
else if (op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE)
{
op->version = op->parts.size() == 1 ? part->op.reply.rw.version : 0;
}
src/client/cluster_client.h
+5 -2
View File
@@ -15,6 +15,7 @@
#define OSD_OP_READ_CHAIN_BITMAP 0x102
#define OSD_OP_IGNORE_READONLY 0x08
#define OSD_OP_WAIT_UP_TIMEOUT 0x10
struct cluster_op_t;
@@ -39,7 +40,8 @@ struct cluster_op_t
// for reads and writes within a single object (stripe),
// reads can return current version and writes can use "CAS" semantics
uint64_t version = 0;
// now only OSD_OP_IGNORE_READONLY is supported
// flags: OSD_OP_IGNORE_READONLY - ignore inode readonly flag
// OSD_OP_WAIT_UP_TIMEOUT - do not retry the operation infinitely if PG is inactive, only for for <wait_up_timeout>
uint64_t flags = 0;
// negative retval is an error number
// write and read return len on success
@@ -57,6 +59,7 @@ protected:
bool needs_reslice = false;
int retry_after = 0;
int inflight_count = 0, done_count = 0;
timespec wait_up_until = {};
std::vector<cluster_op_part_t> parts;
void *part_bitmaps = NULL;
unsigned bitmap_buf_size = 0;
@@ -159,7 +162,7 @@ protected:
bool check_rw(cluster_op_t *op);
void slice_rw(cluster_op_t *op);
void reset_retry_timer(int new_duration);
bool try_send(cluster_op_t *op, int i);
int try_send(cluster_op_t *op, int i);
int continue_sync(cluster_op_t *op);
void send_sync(cluster_op_t *op, cluster_op_part_t *part);
void handle_op_part(cluster_op_part_t *part);
src/cmd/cli_merge.cpp
+1 -1
View File
@@ -611,7 +611,7 @@ struct snap_merger_t
subop->inode = inode_num;
subop->offset = offset;
subop->len = 0;
subop->flags = OSD_OP_IGNORE_READONLY;
subop->flags = OSD_OP_IGNORE_READONLY | OSD_OP_WAIT_UP_TIMEOUT;
subop->callback = [](cluster_op_t *subop)
{
if (subop->retval != 0)
src/cmd/cli_rm_data.cpp
+17 -63
View File
@@ -11,7 +11,6 @@
struct rm_pg_t
{
pg_num_t pg_num;
osd_num_t rm_osd_num;
std::set<object_id> objects;
std::set<object_id>::iterator obj_pos;
uint64_t obj_count = 0, obj_done = 0;
@@ -54,19 +53,6 @@ struct rm_inode_t
parent->cli->list_inode(inode, min_offset, max_offset, parent->parallel_osds, [this](
int errcode, int pgs_left, pg_num_t pg_num, std::set<object_id>&& objects, std::vector<osd_num_t> && inactive_osds)
{
osd_num_t rm_osd_num = 0;
auto pool_it = parent->cli->st_cli.pool_config.find(pool_id);
if (pool_it != parent->cli->st_cli.pool_config.end())
{
auto pg_it = pool_it->second.pg_config.find(pg_num);
if (pg_it != pool_it->second.pg_config.end())
rm_osd_num = pg_it->second.primary;
}
if (!rm_osd_num)
{
fprintf(stderr, "PG %u/%u is down, skipping\n", pool_id, pg_num);
errcode = -EPIPE;
}
if (errcode)
{
inactive_pgs.insert(pg_num);
@@ -79,7 +65,6 @@ struct rm_inode_t
}
rm_pg_t *rm = new rm_pg_t((rm_pg_t){
.pg_num = pg_num,
.rm_osd_num = rm_osd_num,
.objects = std::move(objects),
.obj_done = 0,
.synced = !objects.size() || parent->cli->get_immediate_commit(inode),
@@ -109,43 +94,23 @@ struct rm_inode_t
void send_ops(rm_pg_t *cur_list)
{
parent->cli->init_msgr();
if (parent->cli->msgr.osd_peer_fds.find(cur_list->rm_osd_num) ==
parent->cli->msgr.osd_peer_fds.end())
{
// Initiate connection
parent->cli->msgr.connect_peer(cur_list->rm_osd_num, parent->cli->st_cli.peer_states[cur_list->rm_osd_num]);
return;
}
while (cur_list->in_flight < parent->iodepth && cur_list->obj_pos != cur_list->objects.end())
{
if (cur_list->obj_pos->stripe >= min_offset && (!max_offset || cur_list->obj_pos->stripe < max_offset))
{
osd_op_t *op = new osd_op_t();
op->op_type = OSD_OP_OUT;
// Already checked that it exists above, but anyway
op->peer_fd = parent->cli->msgr.osd_peer_fds.at(cur_list->rm_osd_num);
op->req = (osd_any_op_t){
.rw = {
.header = {
.magic = SECONDARY_OSD_OP_MAGIC,
.id = parent->cli->next_op_id(),
.opcode = OSD_OP_DELETE,
},
.inode = cur_list->obj_pos->inode,
.offset = cur_list->obj_pos->stripe,
.len = 0,
},
};
op->callback = [this, cur_list](osd_op_t *op)
cluster_op_t *op = new cluster_op_t;
op->opcode = OSD_OP_DELETE;
op->inode = cur_list->obj_pos->inode;
op->offset = cur_list->obj_pos->stripe;
op->len = 0;
op->flags = OSD_OP_IGNORE_READONLY | OSD_OP_WAIT_UP_TIMEOUT;
op->callback = [this, cur_list](cluster_op_t *op)
{
cur_list->in_flight--;
if (op->reply.hdr.retval < 0)
if (op->retval < 0)
{
// FIXME: Retry -EPIPE
fprintf(stderr, "Failed to remove object %jx:%jx from PG %u (OSD %ju) (retval=%jd)\n",
op->req.rw.inode, op->req.rw.offset,
cur_list->pg_num, cur_list->rm_osd_num, op->reply.hdr.retval);
fprintf(stderr, "Failed to remove object %jx:%jx from PG %u (retval=%d)\n",
op->inode, op->offset, cur_list->pg_num, op->retval);
error_count++;
}
delete op;
@@ -155,7 +120,7 @@ struct rm_inode_t
};
cur_list->in_flight++;
cur_list->obj_pos++;
parent->cli->msgr.outbox_push(op);
parent->cli->execute(op);
}
else
{
@@ -165,33 +130,22 @@ struct rm_inode_t
if (cur_list->in_flight == 0 && cur_list->obj_pos == cur_list->objects.end() &&
!cur_list->synced)
{
osd_op_t *op = new osd_op_t();
op->op_type = OSD_OP_OUT;
op->peer_fd = parent->cli->msgr.osd_peer_fds.at(cur_list->rm_osd_num);
op->req = (osd_any_op_t){
.sync = {
.header = {
.magic = SECONDARY_OSD_OP_MAGIC,
.id = parent->cli->next_op_id(),
.opcode = OSD_OP_SYNC,
},
},
};
op->callback = [this, cur_list](osd_op_t *op)
cluster_op_t *op = new cluster_op_t;
op->opcode = OSD_OP_SYNC;
op->callback = [this, cur_list](cluster_op_t *op)
{
cur_list->in_flight--;
cur_list->synced = true;
if (op->reply.hdr.retval < 0)
if (op->retval < 0)
{
fprintf(stderr, "Failed to sync OSD %ju (retval=%jd)\n",
cur_list->rm_osd_num, op->reply.hdr.retval);
fprintf(stderr, "Failed to sync after deletion (retval=%d)\n", op->retval);
error_count++;
}
delete op;
continue_delete();
};
cur_list->in_flight++;
parent->cli->msgr.outbox_push(op);
parent->cli->execute(op);
}
}