diff --git a/src/client/cluster_client.cpp b/src/client/cluster_client.cpp index 4b94d7eb..5edf547a 100644 --- a/src/client/cluster_client.cpp +++ b/src/client/cluster_client.cpp @@ -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; } diff --git a/src/client/cluster_client.h b/src/client/cluster_client.h index 2865bb74..ac7f93c5 100644 --- a/src/client/cluster_client.h +++ b/src/client/cluster_client.h @@ -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 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 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); diff --git a/src/cmd/cli_merge.cpp b/src/cmd/cli_merge.cpp index f972be2c..a0f3e1d9 100644 --- a/src/cmd/cli_merge.cpp +++ b/src/cmd/cli_merge.cpp @@ -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) diff --git a/src/cmd/cli_rm_data.cpp b/src/cmd/cli_rm_data.cpp index 12f24b24..d76256cc 100644 --- a/src/cmd/cli_rm_data.cpp +++ b/src/cmd/cli_rm_data.cpp @@ -11,7 +11,6 @@ struct rm_pg_t { pg_num_t pg_num; - osd_num_t rm_osd_num; std::set objects; std::set::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&& objects, std::vector && 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); } }