Skip to content

Commit b67dbca

Browse files
JacksonYao287JacksonYao287
committed
Homeobject side graceful shutdwon
this PR add counters to record the pending request and start shutdown homestore when no pending request
1 parent 1f5683e commit b67dbca

File tree

9 files changed

+178
-32
lines changed

9 files changed

+178
-32
lines changed

‎conanfile.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@
99

1010
class HomeObjectConan(ConanFile):
1111
name = "homeobject"
12-
version = "2.2.12"
12+
version = "2.2.13"
1313

1414
homepage = "https://github.com/eBay/HomeObject"
1515
description = "Blob Store built on HomeReplication"

‎src/include/homeobject/blob_manager.hpp‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,8 @@
1010
namespace homeobject {
1111

1212
ENUM(BlobErrorCode, uint16_t, UNKNOWN = 1, TIMEOUT, INVALID_ARG, UNSUPPORTED_OP, NOT_LEADER, REPLICATION_ERROR,
13-
UNKNOWN_SHARD, UNKNOWN_BLOB, CHECKSUM_MISMATCH, READ_FAILED, INDEX_ERROR, SEALED_SHARD, RETRY_REQUEST);
13+
UNKNOWN_SHARD, UNKNOWN_BLOB, UNKNOWN_PG, CHECKSUM_MISMATCH, READ_FAILED, INDEX_ERROR, SEALED_SHARD, RETRY_REQUEST,
14+
SHUTTING_DOWN);
1415
struct BlobError {
1516
BlobErrorCode code;
1617
// set when we are not the current leader of the PG.
@@ -27,7 +28,8 @@ struct BlobError {
2728
struct Blob {
2829
Blob() = default;
2930
Blob(sisl::io_blob_safe b, std::string const& u, uint64_t o) : body(std::move(b)), user_key(u), object_off(o) {}
30-
Blob(sisl::io_blob_safe b, std::string const& u, uint64_t o, peer_id_t l) : body(std::move(b)), user_key(u), object_off(o), current_leader(l) {}
31+
Blob(sisl::io_blob_safe b, std::string const& u, uint64_t o, peer_id_t l) :
32+
body(std::move(b)), user_key(u), object_off(o), current_leader(l) {}
3133

3234
Blob clone() const;
3335

‎src/include/homeobject/pg_manager.hpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
namespace homeobject {
1313

1414
ENUM(PGError, uint16_t, UNKNOWN = 1, INVALID_ARG, TIMEOUT, UNKNOWN_PG, NOT_LEADER, UNKNOWN_PEER, UNSUPPORTED_OP,
15-
CRC_MISMATCH, NO_SPACE_LEFT, DRIVE_WRITE_ERROR, RETRY_REQUEST);
15+
CRC_MISMATCH, NO_SPACE_LEFT, DRIVE_WRITE_ERROR, RETRY_REQUEST, SHUTTING_DOWN);
1616

1717
struct PGMember {
1818
// Max length is based on homestore::replica_member_info::max_name_len - 1. Last byte is null terminated.

‎src/include/homeobject/shard_manager.hpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
namespace homeobject {
1111

1212
ENUM(ShardError, uint16_t, UNKNOWN = 1, TIMEOUT, INVALID_ARG, NOT_LEADER, UNSUPPORTED_OP, UNKNOWN_PG, UNKNOWN_SHARD,
13-
PG_NOT_READY, CRC_MISMATCH, NO_SPACE_LEFT, RETRY_REQUEST);
13+
PG_NOT_READY, CRC_MISMATCH, NO_SPACE_LEFT, RETRY_REQUEST, SHUTTING_DOWN);
1414

1515
struct ShardInfo {
1616
enum class State : uint8_t {
@@ -22,7 +22,7 @@ struct ShardInfo {
2222
shard_id_t id;
2323
pg_id_t placement_group;
2424
State state;
25-
uint64_t lsn; // created_lsn
25+
uint64_t lsn; // created_lsn
2626
uint64_t created_time;
2727
uint64_t last_modified_time;
2828
uint64_t available_capacity_bytes;

‎src/lib/homestore_backend/hs_blob_manager.cpp‎

Lines changed: 38 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,12 @@ struct put_blob_req_ctx : public repl_result_ctx< BlobManager::Result< HSHomeObj
8282
};
8383

8484
BlobManager::AsyncResult< blob_id_t > HSHomeObject::_put_blob(ShardInfo const& shard, Blob&& blob) {
85+
if (is_shutting_down()) {
86+
LOGI("service is being shut down");
87+
return folly::makeUnexpected(BlobErrorCode::SHUTTING_DOWN);
88+
}
89+
incr_pending_request_num();
90+
8591
auto& pg_id = shard.placement_group;
8692
shared< homestore::ReplDev > repl_dev;
8793
blob_id_t new_blob_id;
@@ -101,11 +107,13 @@ BlobManager::AsyncResult< blob_id_t > HSHomeObject::_put_blob(ShardInfo const& s
101107

102108
if (!repl_dev->is_leader()) {
103109
LOGW("failed to put blob for pg [{}], shard [{}], not leader", pg_id, shard.id);
110+
decr_pending_request_num();
104111
return folly::makeUnexpected(BlobError(BlobErrorCode::NOT_LEADER, repl_dev->get_leader_id()));
105112
}
106113

107114
if (!repl_dev->is_ready_for_traffic()) {
108115
LOGW("failed to put blob for pg [{}], shard [{}], not ready for traffic", pg_id, shard.id);
116+
decr_pending_request_num();
109117
return folly::makeUnexpected(BlobError(BlobErrorCode::RETRY_REQUEST));
110118
}
111119

@@ -176,10 +184,12 @@ BlobManager::AsyncResult< blob_id_t > HSHomeObject::_put_blob(ShardInfo const& s
176184
if (result.hasError()) {
177185
auto err = result.error();
178186
if (err.getCode() == BlobErrorCode::NOT_LEADER) { err.current_leader = repl_dev->get_leader_id(); }
187+
decr_pending_request_num();
179188
return folly::makeUnexpected(err);
180189
}
181190
auto blob_info = result.value();
182191
BLOGT(blob_info.shard_id, blob_info.blob_id, "Put blob success blkid=[{}]", blob_info.pbas.to_string());
192+
decr_pending_request_num();
183193
return blob_info.blob_id;
184194
});
185195
}
@@ -258,6 +268,12 @@ void HSHomeObject::on_blob_put_commit(int64_t lsn, sisl::blob const& header, sis
258268

259269
BlobManager::AsyncResult< Blob > HSHomeObject::_get_blob(ShardInfo const& shard, blob_id_t blob_id, uint64_t req_offset,
260270
uint64_t req_len) const {
271+
if (is_shutting_down()) {
272+
LOGI("service is being shut down");
273+
return folly::makeUnexpected(BlobErrorCode::SHUTTING_DOWN);
274+
}
275+
incr_pending_request_num();
276+
261277
auto& pg_id = shard.placement_group;
262278
auto hs_pg = get_hs_pg(pg_id);
263279
RELEASE_ASSERT(hs_pg, "PG not found");
@@ -269,12 +285,14 @@ BlobManager::AsyncResult< Blob > HSHomeObject::_get_blob(ShardInfo const& shard,
269285

270286
if (!repl_dev->is_ready_for_traffic()) {
271287
LOGW("failed to get blob for pg [{}], shard [{}], not ready for traffic", pg_id, shard.id);
288+
decr_pending_request_num();
272289
return folly::makeUnexpected(BlobError(BlobErrorCode::RETRY_REQUEST));
273290
}
274291

275292
auto r = get_blob_from_index_table(index_table, shard.id, blob_id);
276293
if (!r) {
277294
BLOGE(shard.id, blob_id, "Blob not found in index during get blob");
295+
decr_pending_request_num();
278296
return folly::makeUnexpected(r.error());
279297
}
280298

@@ -298,17 +316,20 @@ BlobManager::AsyncResult< Blob > HSHomeObject::_get_blob_data(const shared< home
298316
read_buf = std::move(read_buf)](auto&& result) mutable -> BlobManager::AsyncResult< Blob > {
299317
if (result) {
300318
BLOGE(shard_id, blob_id, "Failed to get blob: err={}", blob_id, shard_id, result.value());
319+
decr_pending_request_num();
301320
return folly::makeUnexpected(BlobError(BlobErrorCode::READ_FAILED));
302321
}
303322

304323
BlobHeader const* header = r_cast< BlobHeader const* >(read_buf.cbytes());
305324
if (!header->valid()) {
306325
BLOGE(shard_id, blob_id, "Invalid header found: [header={}]", header->to_string());
326+
decr_pending_request_num();
307327
return folly::makeUnexpected(BlobError(BlobErrorCode::READ_FAILED));
308328
}
309329

310330
if (header->shard_id != shard_id) {
311331
BLOGE(shard_id, blob_id, "Invalid shard_id in header: [header={}]", header->to_string());
332+
decr_pending_request_num();
312333
return folly::makeUnexpected(BlobError(BlobErrorCode::READ_FAILED));
313334
}
314335

@@ -325,12 +346,14 @@ BlobManager::AsyncResult< Blob > HSHomeObject::_get_blob_data(const shared< home
325346
if (std::memcmp(computed_hash, header->hash, BlobHeader::blob_max_hash_len) != 0) {
326347
BLOGE(shard_id, blob_id, "Hash mismatch header [{}] [computed={:np}]", header->to_string(),
327348
spdlog::to_hex(computed_hash, computed_hash + BlobHeader::blob_max_hash_len));
349+
decr_pending_request_num();
328350
return folly::makeUnexpected(BlobError(BlobErrorCode::CHECKSUM_MISMATCH));
329351
}
330352

331353
if (req_offset + req_len > header->blob_size) {
332354
BLOGE(shard_id, blob_id, "Invalid offset length requested in get blob offset={} len={} size={}",
333355
req_offset, req_len, header->blob_size);
356+
decr_pending_request_num();
334357
return folly::makeUnexpected(BlobError(BlobErrorCode::INVALID_ARG));
335358
}
336359

@@ -341,6 +364,7 @@ BlobManager::AsyncResult< Blob > HSHomeObject::_get_blob_data(const shared< home
341364
std::memcpy(body.bytes(), blob_bytes + req_offset, res_len);
342365

343366
BLOGT(blob_id, shard_id, "Blob get success: blkid={}", blkid.to_string());
367+
decr_pending_request_num();
344368
return Blob(std::move(body), std::move(user_key), header->object_offset, repl_dev->get_leader_id());
345369
});
346370
}
@@ -362,6 +386,7 @@ HSHomeObject::blob_put_get_blk_alloc_hints(sisl::blob const& header, cintrusive<
362386
auto hs_pg = get_hs_pg(msg_header->pg_id);
363387
if (hs_pg == nullptr) {
364388
LOGW("Received a blob_put on an unknown pg:{}, underlying engine will retry this later", msg_header->pg_id);
389+
if (ctx) { ctx->promise_.setValue(folly::makeUnexpected(BlobError(BlobErrorCode::UNKNOWN_PG))); }
365390
return folly::makeUnexpected(homestore::ReplServiceError::RESULT_NOT_EXIST_YET);
366391
}
367392

@@ -370,6 +395,7 @@ HSHomeObject::blob_put_get_blk_alloc_hints(sisl::blob const& header, cintrusive<
370395
if (shard_iter == _shard_map.end()) {
371396
LOGW("Received a blob_put on an unknown shard:{}, underlying engine will retry this later",
372397
msg_header->shard_id);
398+
if (ctx) { ctx->promise_.setValue(folly::makeUnexpected(BlobError(BlobErrorCode::UNKNOWN_SHARD))); }
373399
return folly::makeUnexpected(homestore::ReplServiceError::RESULT_NOT_EXIST_YET);
374400
}
375401

@@ -392,6 +418,12 @@ HSHomeObject::blob_put_get_blk_alloc_hints(sisl::blob const& header, cintrusive<
392418
}
393419

394420
BlobManager::NullAsyncResult HSHomeObject::_del_blob(ShardInfo const& shard, blob_id_t blob_id) {
421+
if (is_shutting_down()) {
422+
LOGI("service is being shut down");
423+
return folly::makeUnexpected(BlobErrorCode::SHUTTING_DOWN);
424+
}
425+
incr_pending_request_num();
426+
395427
BLOGT(shard.id, blob_id, "deleting blob");
396428
auto& pg_id = shard.placement_group;
397429
auto hs_pg = get_hs_pg(pg_id);
@@ -402,11 +434,13 @@ BlobManager::NullAsyncResult HSHomeObject::_del_blob(ShardInfo const& shard, blo
402434

403435
if (!repl_dev->is_leader()) {
404436
LOGW("failed to del blob for pg [{}], shard [{}], blob_id [{}], not leader", pg_id, shard.id, blob_id);
437+
decr_pending_request_num();
405438
return folly::makeUnexpected(BlobError(BlobErrorCode::NOT_LEADER, repl_dev->get_leader_id()));
406439
}
407440

408441
if (!repl_dev->is_ready_for_traffic()) {
409442
LOGW("failed to del blob for pg [{}], shard [{}], not ready for traffic", pg_id, shard.id);
443+
decr_pending_request_num();
410444
return folly::makeUnexpected(BlobError(BlobErrorCode::RETRY_REQUEST));
411445
}
412446

@@ -424,14 +458,16 @@ BlobManager::NullAsyncResult HSHomeObject::_del_blob(ShardInfo const& shard, blo
424458
std::memcpy(req->key_buf().bytes(), &blob_id, sizeof(blob_id_t));
425459

426460
repl_dev->async_alloc_write(req->cheader_buf(), req->ckey_buf(), sisl::sg_list{}, req);
427-
return req->result().deferValue([repl_dev](const auto& result) -> folly::Expected< folly::Unit, BlobError > {
461+
return req->result().deferValue([this, repl_dev](const auto& result) -> folly::Expected< folly::Unit, BlobError > {
428462
if (result.hasError()) {
429463
auto err = result.error();
430464
if (err.getCode() == BlobErrorCode::NOT_LEADER) { err.current_leader = repl_dev->get_leader_id(); }
465+
decr_pending_request_num();
431466
return folly::makeUnexpected(err);
432467
}
433468
auto blob_info = result.value();
434469
BLOGT(blob_info.shard_id, blob_info.blob_id, "Delete blob successful");
470+
decr_pending_request_num();
435471
return folly::Unit();
436472
});
437473
}
@@ -470,7 +506,7 @@ void HSHomeObject::on_blob_del_commit(int64_t lsn, sisl::blob const& header, sis
470506
if (ctx) ctx->promise_.setValue(folly::makeUnexpected(r.error()));
471507
return;
472508
} else {
473-
if (ctx) { ctx->promise_.setValue(BlobManager::Result< BlobInfo >(blob_info)); }
509+
if (ctx) ctx->promise_.setValue(BlobManager::Result< BlobInfo >(blob_info));
474510
return;
475511
}
476512
}

‎src/lib/homestore_backend/hs_homeobject.cpp‎

Lines changed: 15 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -60,9 +60,7 @@ class HSReplApplication : public homestore::ReplApplication {
6060
return it->second;
6161
}
6262

63-
void on_repl_devs_init_completed() override {
64-
_home_object->on_replica_restart();
65-
}
63+
void on_repl_devs_init_completed() override { _home_object->on_replica_restart(); }
6664

6765
std::pair< std::string, uint16_t > lookup_peer(homestore::replica_id_t uuid) const override {
6866
std::string endpoint;
@@ -320,9 +318,22 @@ HSHomeObject::~HSHomeObject() {
320318
}
321319
trigger_timed_events();
322320
#endif
321+
322+
start_shutting_down();
323+
// Wait for all pending requests to complete
324+
while (true) {
325+
auto pending_reqs = get_pending_request_num();
326+
if (0 == pending_reqs) break;
327+
LOGI("waiting for {} pending requests to complete", pending_reqs);
328+
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
329+
}
330+
LOGI("start shutting down HomeStore");
331+
323332
homestore::HomeStore::instance()->shutdown();
324333
homestore::HomeStore::reset_instance();
325334
iomanager.stop();
335+
336+
LOGI("complete shutting down HomeStore");
326337
}
327338

328339
HomeObjectStats HSHomeObject::_get_stats() const {
@@ -340,7 +351,7 @@ HomeObjectStats HSHomeObject::_get_stats() const {
340351

341352
stats.num_open_shards = num_open_shards;
342353
stats.avail_open_shards = chunk_selector()->total_chunks() - num_open_shards;
343-
stats.num_disks = chunk_selector()->total_disks();
354+
stats.num_disks = chunk_selector()->total_disks();
344355
return stats;
345356
}
346357

‎src/lib/homestore_backend/hs_homeobject.hpp‎

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -393,7 +393,8 @@ class HSHomeObject : public HomeObjectImpl {
393393
bool update_cursor(objId id);
394394
objId expected_next_obj_id();
395395
bool generate_shard_blob_list();
396-
BlobManager::AsyncResult< sisl::io_blob_safe > load_blob_data(const BlobInfo& blob_info, ResyncBlobState& state);
396+
BlobManager::AsyncResult< sisl::io_blob_safe > load_blob_data(const BlobInfo& blob_info,
397+
ResyncBlobState& state);
397398
bool create_pg_snapshot_data(sisl::io_blob_safe& meta_blob);
398399
bool create_shard_snapshot_data(sisl::io_blob_safe& meta_blob);
399400
bool create_blobs_snapshot_data(sisl::io_blob_safe& data_blob);
@@ -409,7 +410,7 @@ class HSHomeObject : public HomeObjectImpl {
409410

410411
objId cur_obj_id_{0, 0};
411412
int64_t cur_shard_idx_{-1};
412-
std::vector<BlobInfo> cur_blob_list_{0};
413+
std::vector< BlobInfo > cur_blob_list_{0};
413414
uint64_t cur_start_blob_idx_{0};
414415
uint64_t cur_batch_blob_count_{0};
415416
flatbuffers::FlatBufferBuilder builder_;
@@ -677,7 +678,7 @@ class HSHomeObject : public HomeObjectImpl {
677678
const homestore::MultiBlkId& pbas, cintrusive< homestore::repl_req_ctx >& hs_ctx);
678679
void on_blob_del_commit(int64_t lsn, sisl::blob const& header, sisl::blob const& key,
679680
cintrusive< homestore::repl_req_ctx >& hs_ctx);
680-
bool local_add_blob_info(pg_id_t pg_id, BlobInfo const &blob_info);
681+
bool local_add_blob_info(pg_id_t pg_id, BlobInfo const& blob_info);
681682
homestore::ReplResult< homestore::blk_alloc_hints >
682683
blob_put_get_blk_alloc_hints(sisl::blob const& header, cintrusive< homestore::repl_req_ctx >& ctx);
683684
void compute_blob_payload_hash(BlobHeader::HashAlgorithm algorithm, const uint8_t* blob_bytes, size_t blob_size,
@@ -746,13 +747,28 @@ class HSHomeObject : public HomeObjectImpl {
746747
*/
747748
void destroy_pg_index_table(pg_id_t pg_id);
748749

749-
/**
750+
/**
750751
* @brief Destroy the superblock for the PG identified by pg_id.
751752
* Ensures all operations are persisted by triggering a cp flush before destruction.
752753
*
753754
* @param pg_id The ID of the PG to be destroyed.
754755
*/
756+
755757
void destroy_pg_superblk(pg_id_t pg_id);
758+
759+
// graceful shutdown related
760+
private:
761+
std::atomic_bool shutting_down{false};
762+
mutable std::atomic_uint64_t pending_request_num{0};
763+
764+
bool is_shutting_down() const { return shutting_down.load(); }
765+
void start_shutting_down() { shutting_down = true; }
766+
767+
uint64_t get_pending_request_num() const { return pending_request_num.load(); }
768+
769+
// only leader will call incr and decr pending request num
770+
void incr_pending_request_num() const { pending_request_num++; }
771+
void decr_pending_request_num() const { pending_request_num--; }
756772
};
757773

758774
class BlobIndexServiceCallbacks : public homestore::IndexServiceCallbacks {

0 commit comments

Comments
 (0)