diff --git a/src/blockstore/blockstore_mock.cpp b/src/blockstore/blockstore_mock.cpp index ad5a05c0..ab34a53c 100644 --- a/src/blockstore/blockstore_mock.cpp +++ b/src/blockstore/blockstore_mock.cpp @@ -69,17 +69,17 @@ std::string blockstore_mock_t::get_op_diag(blockstore_op_t *op) uint32_t blockstore_mock_t::get_block_size() { - return 0; + return block_size; } uint64_t blockstore_mock_t::get_block_count() { - return 0; + return block_count; } uint64_t blockstore_mock_t::get_free_block_count() { - return 0; + return block_count; } uint64_t blockstore_mock_t::get_journal_size() @@ -89,7 +89,7 @@ uint64_t blockstore_mock_t::get_journal_size() uint32_t blockstore_mock_t::get_bitmap_granularity() { - return 0; + return bitmap_granularity; } uint64_t blockstore_mock_t::get_live_entries() diff --git a/src/blockstore/blockstore_mock.h b/src/blockstore/blockstore_mock.h index 861066c5..e2bac1d4 100644 --- a/src/blockstore/blockstore_mock.h +++ b/src/blockstore/blockstore_mock.h @@ -8,6 +8,9 @@ class blockstore_mock_t: public blockstore_i { public: + uint32_t block_size = 128*1024; + uint32_t bitmap_granularity = 4096; + uint64_t block_count = 100*1024*8; std::map inode_space; blockstore_mock_t(const blockstore_config_t & config); diff --git a/src/client/etcd_state_client_mock.cpp b/src/client/etcd_state_client_mock.cpp index 1d25aced..b1ffde2e 100644 --- a/src/client/etcd_state_client_mock.cpp +++ b/src/client/etcd_state_client_mock.cpp @@ -5,6 +5,13 @@ #include "etcd_state_client_mock.h" #include "str_util.h" +etcd_state_client_mock_t::etcd_state_client_mock_t() +{ + timespec tv; + clock_gettime(CLOCK_REALTIME, &tv); + srand48(tv.tv_sec*1000000000 + tv.tv_nsec); +} + void etcd_state_client_mock_t::etcd_add_watch(json11::Json watch) { } @@ -83,6 +90,7 @@ void etcd_state_client_mock_t::etcd_call(std::string api, json11::Json payload, else assert(0); } + std::map changes; bool has_mod = false; for (auto& op: payload[ok ? "success" : "failure"].array_items()) { @@ -131,6 +139,14 @@ void etcd_state_client_mock_t::etcd_call(std::string api, json11::Json payload, .mod_revision = mod_revision, .lease_id = lease_id, }; + std::string err; + json11::Json json_value = json11::Json::parse(value, err); + if (err != "") + { + fprintf(stderr, "Invalid JSON in etcd key %s during test: %s\n", key.c_str(), value.c_str()); + exit(1); + } + changes[key] = { .key = key, .value = json_value, .mod_revision = mod_revision }; responses.push_back(json11::Json::object { { "response_put", json11::Json::object{ { "header", json11::Json::object{ { "revision", mod_revision } } } } }, }); @@ -143,7 +159,9 @@ void etcd_state_client_mock_t::etcd_call(std::string api, json11::Json payload, uint64_t n_del = 0; for (auto it = data.lower_bound(key); it != data.end() && (range_end == "" || it->first < range_end); ) { - printf("\\- del: %s\n", it->first.c_str()); + auto & key = it->first; + printf("\\- del: %s\n", key.c_str()); + changes[key] = { .key = key, .mod_revision = mod_revision }; n_del++; data.erase(it++); } @@ -152,7 +170,25 @@ void etcd_state_client_mock_t::etcd_call(std::string api, json11::Json payload, }); } } - callback("", json11::Json::object{ { "header", json11::Json::object{ { "revision", mod_revision } } }, { "succeeded", ok }, { "responses", responses } }); + callback("", json11::Json::object{ + { "header", json11::Json::object{ { "revision", mod_revision } } }, + { "succeeded", ok }, + { "responses", responses } + }); + // Push changes to watcher + if (changes.size()) + { + for (auto & kv: changes) + parse_state(kv.second); + if (on_change_hook != NULL) + on_change_hook(changes); + } + } + else if (api == "/lease/grant") + { + uint64_t lease_id = (((uint64_t)lrand48()) << 32) | lrand48(); + leases[lease_id] = payload["TTL"].uint64_value(); + callback("", json11::Json::object{ { "ID", std::to_string(lease_id) } }); } else callback("Unsupported", json11::Json()); diff --git a/src/client/etcd_state_client_mock.h b/src/client/etcd_state_client_mock.h index e37b3f7f..38b18854 100644 --- a/src/client/etcd_state_client_mock.h +++ b/src/client/etcd_state_client_mock.h @@ -28,7 +28,9 @@ struct etcd_state_client_mock_t: public etcd_state_client_t bool paused = false; std::vector queue; public: + std::map leases; std::map data; + etcd_state_client_mock_t(); void set(const std::string& key, json11::Json data, uint64_t mod_revision = 0, uint64_t lease_id = 0); void pause(); void resume(); diff --git a/src/osd/CMakeLists.txt b/src/osd/CMakeLists.txt index 4302da93..5e6fe8f2 100644 --- a/src/osd/CMakeLists.txt +++ b/src/osd/CMakeLists.txt @@ -39,7 +39,10 @@ target_link_libraries(osd_test vitastor_common Jerasure ${ISAL_LIBRARIES} + ${LIBURING_LIBRARIES} ) +add_test(NAME osd_test COMMAND osd_test) +add_dependencies(build_tests osd_test) # osd_rmw_test add_executable(osd_rmw_test EXCLUDE_FROM_ALL osd_rmw_test.cpp ../util/allocator.cpp) diff --git a/src/osd/osd.cpp b/src/osd/osd.cpp index 1be9f422..cb5f256f 100644 --- a/src/osd/osd.cpp +++ b/src/osd/osd.cpp @@ -30,7 +30,10 @@ osd_t::osd_t(const json11::Json & config, ring_loop_i *ringloop, timerfd_manager this->cli_config = config.object_items(); this->file_config = msgr.read_config(this->cli_config); parse_config(true); +} +void osd_t::start() +{ if (json_is_true(this->config["osd_memlock"])) { // Lock all OSD memory if requested @@ -130,17 +133,22 @@ void osd_t::init_blockstore(std::function on_init) autosync_writes = max_autosync; if (on_init) { - init_consumer.loop = [this, on_init]() + if (bs->is_started()) + on_init(); + else { - // Wait for blockstore initialisation before actually starting OSD logic - // to prevent peering timeouts during restart with filled databases - if (bs->is_started()) + init_consumer.loop = [this, on_init]() { - ringloop->set_immediate([this, on_init] { init_consumer.loop = NULL; on_init(); }); - ringloop->unregister_consumer(&init_consumer); - } - }; - ringloop->register_consumer(&init_consumer); + // Wait for blockstore initialisation before actually starting OSD logic + // to prevent peering timeouts during restart with filled databases + if (bs->is_started()) + { + ringloop->set_immediate([this, on_init] { init_consumer.loop = NULL; on_init(); }); + ringloop->unregister_consumer(&init_consumer); + } + }; + ringloop->register_consumer(&init_consumer); + } } } else if (on_init) diff --git a/src/osd/osd.h b/src/osd/osd.h index 57c60556..2c17576b 100644 --- a/src/osd/osd.h +++ b/src/osd/osd.h @@ -97,8 +97,12 @@ struct osd_pg_lock_t uint64_t state = 0; }; +struct osd_test_fixture_t; + class osd_t { + friend struct osd_test_fixture_t; + // config json11::Json::object cli_config, file_config, etcd_global_config, etcd_osd_config, config; @@ -394,6 +398,11 @@ public: osd_t(const json11::Json & config, ring_loop_i *ringloop, timerfd_manager_t *tfd, std::unique_ptr st_cli_ptr, std::function bs_factory); ~osd_t(); + // Wire the OSD into the world: install timers, hook the messenger up to + // ringloop, register as a ringloop consumer, and kick off the etcd + // config-load chain. Separated from the constructor so tests can build + // an inert osd_t, prime mocks, then drive startup explicitly. + void start(); void force_stop(int exitcode); bool shutdown(); }; diff --git a/src/osd/osd_main.cpp b/src/osd/osd_main.cpp index 691d00c5..9491c1ca 100644 --- a/src/osd/osd_main.cpp +++ b/src/osd/osd_main.cpp @@ -74,6 +74,7 @@ int main(int narg, char *args[]) return blockstore_i::create(cfg, ringloop, tfd); } ); + osd->start(); while (1) { ringloop->loop(); diff --git a/src/test/osd_test.cpp b/src/test/osd_test.cpp index 1d6b64f5..048a9a14 100644 --- a/src/test/osd_test.cpp +++ b/src/test/osd_test.cpp @@ -3,27 +3,147 @@ #include #include +#include #include -#include "osd.h" -#include "etcd_state_client_mock.h" -#include "ringloop_mock.h" -#include "blockstore_mock.h" -void test1() +#include "osd.h" +#include "osd_primary.h" +#include "osd_test_fixture.h" + +// Verify that an OSD configured with an etcd address issues a range read for +// //config/global and //config/pools right after construction. +// +// The etcd mock is paused before constructing the OSD, so the txn stays in +// the mock's queue instead of executing its callback synchronously inside +// the constructor — that gives us a chance to inspect what the OSD asked +// for without having to drive the rest of the startup sequence. +void test_load_global_config() { - json11::Json config; - timerfd_manager_t *tfd = new timerfd_manager_t([](int fd, bool wr, std::function callback){}); - etcd_state_client_mock_t *st_cli = new etcd_state_client_mock_t(); - ring_loop_mock_t *ringloop = new ring_loop_mock_t(RINGLOOP_DEFAULT_SIZE, [&](io_uring_sqe *sqe) {}); - st_cli->pause(); - osd_t *osd = new osd_t(config, ringloop, tfd, std::unique_ptr(st_cli), [](blockstore_config_t & cfg) - { - return new blockstore_mock_t({}); + osd_test_fixture_t f; + f.st_cli->pause(); + f.start(json11::Json::object { + { "osd_num", 1 }, + { "etcd_address", "127.0.0.1:2379" }, + { "etcd_prefix", "/vitastor" }, + { "run_primary", false }, }); + + assert(f.st_cli->queue.size() == 1); + assert(f.st_cli->queue[0].api == "/kv/txn"); + auto & ops = f.st_cli->queue[0].payload["success"].array_items(); + assert(ops.size() == 2); + assert(base64_decode(ops[0]["request_range"]["key"].string_value()) == "/vitastor/config/global"); + assert(base64_decode(ops[1]["request_range"]["key"].string_value()) == "/vitastor/config/pools"); + + printf("test_load_global_config passed\n"); +} + +// Build a self-issued OSD_OP_WRITE op for the given inode/offset/length, +// filled with the given byte. client_id=0 (SELF_CLIENT) so finish_op +// delivers the reply through the callback we set rather than over the wire. +static osd_op_t *make_write_op(inode_t inode, uint64_t offset, uint64_t len, uint8_t fill) +{ + auto *op = new osd_op_t(); + op->op_type = OSD_OP_IN; + op->client_id = 0; + op->req.rw.header.magic = SECONDARY_OSD_OP_MAGIC; + op->req.rw.header.id = 1; + op->req.rw.header.opcode = OSD_OP_WRITE; + op->req.rw.inode = inode; + op->req.rw.offset = offset; + op->req.rw.len = len; + op->buf = malloc(len); + memset(op->buf, fill, len); + return op; +} + +// Drive a single 4 KiB replicated write through the primary OSD state machine +// and verify it produces (1) a local blockstore write and (2) a peer subop, +// then completes the client op once both finish. +// +// Layout: pool 1, replicated x2, primary = OSD 1 (us), secondary = OSD 2. +void test_replicated_write() +{ + osd_test_fixture_t f; + f.configure_replicated_pool(/*pool_id*/ 1, /*pg_size*/ 2, /*pg_minsize*/ 1, /*pg_count*/ 1, + { { 1, 2 } }); + f.start(json11::Json::object { + { "osd_num", 1 }, + { "etcd_address", "127.0.0.1:2379" }, + { "immediate_commit", "all" }, + { "block_size", 131072 }, + { "bitmap_granularity", 4096 }, + }); + // Peer 2 wasn't online during initial peering -> PG is INCOMPLETE. + // Connecting it re-triggers peering; completing the LIST subops with + // "no objects" lets the PG transition to PG_ACTIVE. + f.connect_peer(2); + f.complete_peering_empty(); + assert(f.pg(1, 1).state & PG_ACTIVE); + + auto *write_op = make_write_op(INODE_WITH_POOL(1, 1), 0, 4096, 0xab); + int final_retval = -1; + write_op->callback = [&final_retval](osd_op_t *op) { + final_retval = op->reply.hdr.retval; + }; + f.exec(write_op); + + // Stage 1: the primary always pumps one subop through SUBMIT_RMW_READ + // even for a "fresh" full-block replicated write — for our setup that + // subop is a zero-length local read that just resolves the object's + // current version. + assert(f.bs->queued.size() == 1); + auto *zero_read = f.bs->take(); + assert(zero_read->opcode == BS_OP_READ); + assert(zero_read->len == 0); + zero_read->version = 0; // object doesn't exist yet -> current version 0 + zero_read->retval = 0; + zero_read->callback(zero_read); + + // Stage 2: with fact_ver=0 and target_ver=1 the primary issues the + // actual writes — one local (to bs) and one remote (to OSD 2). + auto *peer = f.peer(2); + assert(f.bs->queued.size() == 1); + assert(peer->sent_ops.size() == 1); + + auto *local_write = f.bs->take(); + assert(local_write->opcode == BS_OP_WRITE_STABLE); + assert(local_write->len == 4096); + assert(local_write->oid.inode == INODE_WITH_POOL(1, 1)); + assert(local_write->oid.stripe == 0); + assert(local_write->version == 1); + + auto sent_it = peer->sent_ops.begin(); + osd_op_t *remote_write = sent_it->second; + peer->sent_ops.erase(sent_it); + assert(remote_write->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE); + assert(remote_write->osd_num == 2); + assert(remote_write->req.sec_rw.oid.inode == INODE_WITH_POOL(1, 1)); + assert(remote_write->req.sec_rw.len == 4096); + assert(remote_write->req.sec_rw.version == 1); + + // Local write completes first; the primary is still waiting for the peer. + local_write->retval = local_write->len; + local_write->callback(local_write); + assert(final_retval == -1); + + // Peer reply triggers handle_primary_subop, which sees both subops done + // and lets continue_primary_write reach finish_op. + remote_write->reply.hdr.retval = remote_write->req.sec_rw.len; + remote_write->reply.sec_rw.version = 1; + remote_write->callback(remote_write); + + assert(final_retval == 4096); + assert(f.pg(1, 1).inflight == 0); + assert(f.pg(1, 1).write_queue.empty()); + + delete write_op; + printf("test_replicated_write passed\n"); } int main(int narg, char *args[]) { - test1(); + test_load_global_config(); + test_replicated_write(); return 0; } diff --git a/src/test/osd_test_fixture.h b/src/test/osd_test_fixture.h new file mode 100644 index 00000000..d773501e --- /dev/null +++ b/src/test/osd_test_fixture.h @@ -0,0 +1,313 @@ +// Copyright (c) Vitaliy Filippov, 2019+ +// License: VNPL-1.1 (see README.md for details) + +#pragma once + +#include +#include +#include +#include + +#include "osd.h" +#include "blockstore_mock.h" +#include "etcd_state_client_mock.h" +#include "ringloop_mock.h" +#include "str_util.h" + +// blockstore_mock that captures enqueued ops so tests can drive them by hand. +class capturing_bs_t: public blockstore_mock_t +{ +public: + using blockstore_mock_t::blockstore_mock_t; + std::vector queued; + + void enqueue_op(blockstore_op_t *op) override + { + queued.push_back(op); + } + + blockstore_op_t *take() + { + assert(!queued.empty()); + auto *op = queued.front(); + queued.erase(queued.begin()); + return op; + } + + // Pop the first queued op with the given opcode. Aborts if none. + blockstore_op_t *take(int opcode) + { + for (auto it = queued.begin(); it != queued.end(); ++it) + { + if ((*it)->opcode == opcode) + { + auto *op = *it; + queued.erase(it); + return op; + } + } + fprintf(stderr, "capturing_bs_t::take: no queued op with opcode %d\n", opcode); + abort(); + } +}; + +// Test harness for primary OSD operations. +// +// Owns the mocks (ringloop, timerfd_manager, etcd, blockstore) and exposes +// helpers that drive the OSD through phases its production lifecycle would +// otherwise reach via async etcd/network events: configuring pools through +// the etcd mock, pretending peer OSDs (dis)connected, completing peering +// LIST subops. +// +// A friend of osd_t so it can poke at PG state and message queues; tests +// themselves should only talk to the fixture's public interface (and to +// the osd_t for things like exec_op). +struct osd_test_fixture_t +{ + timerfd_manager_t *tfd = nullptr; + ring_loop_mock_t *ringloop = nullptr; + etcd_state_client_mock_t *st_cli = nullptr; // owned by osd_t after start() + osd_t *osd = nullptr; + capturing_bs_t *bs = nullptr; + + osd_test_fixture_t() + { + tfd = new timerfd_manager_t([](int, bool, std::function) {}); + ringloop = new ring_loop_mock_t(RINGLOOP_DEFAULT_SIZE, [](io_uring_sqe *) {}); + st_cli = new etcd_state_client_mock_t(); + } + + ~osd_test_fixture_t() + { + if (osd) + { + // Drop PGs before tearing msgr down: stop_client would otherwise + // call repeer_pgs() on PGs that reference peers we're about to + // delete during ~osd_messenger_t(). + osd->pgs.clear(); + delete osd; + } + delete ringloop; + delete tfd; + } + + // Populate the etcd mock with a single replicated pool's pool/PG config + // BEFORE start(). osd_t's constructor-time config load will then pick it + // up synchronously and run apply_pg_config() on construction. + // + // pgs[i] is the osd_set for PG i+1; primary = pgs[i][0]. + void configure_replicated_pool(pool_id_t pool_id, int pg_size, int pg_minsize, + int pg_count, const std::vector> & pgs) + { + assert((int)pgs.size() == pg_count); + auto pool_id_s = std::to_string(pool_id); + st_cli->set("/vitastor/config/pools", json11::Json::object { + { pool_id_s, json11::Json::object { + { "name", "pool_"+pool_id_s }, + { "scheme", "replicated" }, + { "pg_size", pg_size }, + { "pg_minsize", pg_minsize }, + { "pg_count", pg_count }, + { "failure_domain", "osd" }, + { "immediate_commit", "none" }, + } }, + }); + json11::Json::object items_pool; + for (int i = 0; i < pg_count; i++) + { + json11::Json::array osd_set; + for (auto n: pgs[i]) + osd_set.push_back((double)n); + items_pool[std::to_string(i+1)] = json11::Json::object { + { "osd_set", osd_set }, + { "primary", (double)pgs[i][0] }, + }; + } + st_cli->set("/vitastor/pg/config", json11::Json::object { + { "items", json11::Json::object{ { pool_id_s, items_pool } } }, + }); + } + + // Construct the osd_t AND call osd->start(). With the etcd mock unpaused, + // the whole config-load chain (config/global, config/pools, lease, + // /osd/state, pg/config, apply_pg_config, start_pg_peering) runs + // synchronously inside start(). Peers that aren't yet connect_peer()'d + // will leave their PGs in PG_INCOMPLETE; call connect_peer() + + // complete_peering_empty() afterwards to drive them to PG_ACTIVE. + // + // If a test needs a fully inert osd_t (no timers, no etcd traffic) — for + // instance to verify constructor-time behavior — use construct() instead + // and call osd->start() yourself. + void construct(json11::Json::object osd_config) + { + assert(!osd); + osd = new osd_t(osd_config, ringloop, tfd, + std::unique_ptr(st_cli), + [this](blockstore_config_t & cfg) -> blockstore_i* { + return (bs = new capturing_bs_t(cfg)); + }); + } + + void start(json11::Json::object osd_config) + { + construct(osd_config); + osd->start(); + } + + // Pretend a peer OSD has connected (no real TCP). Registers it in the + // messenger and triggers re-peering of PGs that include it. + void connect_peer(osd_num_t osd_num) + { + auto *msgr = &osd->msgr; + auto *cl = new osd_client_t(); + cl->client_id = msgr->next_client_id++; + cl->osd_num = osd_num; + cl->peer_fd = -1; + cl->peer_state = PEER_CONNECTED; + msgr->osd_peers[osd_num] = cl; + msgr->clients[cl->client_id] = cl; + msgr->wanted_peers.erase(osd_num); + msgr->repeer_pgs(osd_num); + } + + void disconnect_peer(osd_num_t osd_num) + { + osd->msgr.stop_client(osd->msgr.osd_peers.at(osd_num)->client_id); + } + + // Complete every outstanding peering LIST op (local BS_OP_LIST + peer + // OSD_OP_SEC_LIST) with "no objects" so PGs transition to PG_ACTIVE + // without us having to invent object lists. + void complete_peering_empty() + { + for (auto it = bs->queued.begin(); it != bs->queued.end(); ) + { + auto *op = *it; + if (op->opcode == BS_OP_LIST) + { + it = bs->queued.erase(it); + op->retval = 0; + op->version = 0; + op->buf = NULL; + op->callback(op); + } + else + ++it; + } + for (auto & p: osd->msgr.osd_peers) + { + auto *cl = p.second; + std::vector list_ops; + for (auto & kv: cl->sent_ops) + if (kv.second->req.hdr.opcode == OSD_OP_SEC_LIST) + list_ops.push_back(kv.second); + for (auto *op: list_ops) + { + cl->sent_ops.erase(op->req.hdr.id); + op->reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC; + op->reply.hdr.id = op->req.hdr.id; + op->reply.hdr.opcode = op->req.hdr.opcode; + op->reply.hdr.retval = 0; + op->reply.sec_list.stable_count = 0; + op->buf = NULL; + op->callback(op); + } + } + // Drive handle_peers so PGs see lists_done and finalize calc_object_states. + ringloop->wakeup(); + ringloop->loop(); + } + + // Inline list response builder. `objects` are pairs (oid, version); + // the first `stable_count` entries are reported as stable. The buffer + // is malloc'd and ownership transfers to the peering machinery, which + // frees it after calc_object_states. + static obj_ver_id *build_list_buf(const std::vector & objects) + { + if (objects.empty()) + return nullptr; + auto *buf = (obj_ver_id*)malloc(objects.size() * sizeof(obj_ver_id)); + for (size_t i = 0; i < objects.size(); i++) + buf[i] = objects[i]; + return buf; + } + + // Pop the first BS_OP_LIST from bs->queued and reply with the given + // object list. The peering callback consumes (and later frees) op->buf. + void reply_local_list(const std::vector & objects, uint64_t stable_count) + { + auto *op = bs->take(BS_OP_LIST); + op->buf = (uint8_t*)build_list_buf(objects); + op->retval = (int)objects.size(); + op->version = stable_count; + op->callback(op); + } + + // Pop the OSD_OP_SEC_LIST sent to `osd_num` and reply with the given + // object list. Same ownership rules as reply_local_list. + void reply_peer_list(osd_num_t osd_num, + const std::vector & objects, uint64_t stable_count) + { + auto *cl = osd->msgr.osd_peers.at(osd_num); + osd_op_t *op = nullptr; + for (auto & kv: cl->sent_ops) + { + if (kv.second->req.hdr.opcode == OSD_OP_SEC_LIST) + { + op = kv.second; + break; + } + } + assert(op); + cl->sent_ops.erase(op->req.hdr.id); + op->reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC; + op->reply.hdr.id = op->req.hdr.id; + op->reply.hdr.opcode = op->req.hdr.opcode; + op->reply.hdr.retval = (int64_t)objects.size(); + op->reply.sec_list.stable_count = stable_count; + op->buf = build_list_buf(objects); + op->callback(op); + } + + // Pop the first sent op of given opcode from peer's outbox. Aborts if + // none — caller should know which subops to expect. + osd_op_t *peer_take(osd_num_t osd_num, uint64_t opcode) + { + auto *cl = osd->msgr.osd_peers.at(osd_num); + for (auto & kv: cl->sent_ops) + { + if (kv.second->req.hdr.opcode == opcode) + { + auto *op = kv.second; + cl->sent_ops.erase(op->req.hdr.id); + return op; + } + } + fprintf(stderr, "peer_take: OSD %ju has no sent op with opcode %ju\n", osd_num, opcode); + abort(); + } + + // Drive ringloop to flush handle_peers and any pending lambdas. + void pump() + { + ringloop->wakeup(); + ringloop->loop(); + } + + // Submit an op to the OSD's primary state machine. Wrapper because + // osd_t::exec_op is private — tests aren't friends, the fixture is. + void exec(osd_op_t *op) + { + osd->exec_op(op); + } + + osd_client_t *peer(osd_num_t osd_num) + { + return osd->msgr.osd_peers.at(osd_num); + } + + pg_t &pg(pool_id_t pool, pg_num_t num) + { + return osd->pgs.at({ .pool_id = pool, .pg_num = num }); + } +}; diff --git a/src/test/ringloop_mock.h b/src/test/ringloop_mock.h index 148f2d00..d8d846bc 100644 --- a/src/test/ringloop_mock.h +++ b/src/test/ringloop_mock.h @@ -4,6 +4,7 @@ #pragma once #include "ringloop.h" +#include class ring_loop_mock_t: public ring_loop_i { diff --git a/src/test/test_cluster_client.cpp b/src/test/test_cluster_client.cpp index 77ef832f..af0f4739 100644 --- a/src/test/test_cluster_client.cpp +++ b/src/test/test_cluster_client.cpp @@ -114,24 +114,34 @@ void check_completed(int *r) delete r; } -void pretend_connected(cluster_client_t *cli, osd_num_t osd_num) +void pretend_connected(osd_messenger_t *msgr, osd_num_t osd_num) { printf("OSD %ju connected\n", osd_num); auto cl = new osd_client_t(); - cl->client_id = cli->msgr.next_client_id++; + cl->client_id = msgr->next_client_id++; cl->osd_num = osd_num; cl->peer_fd = -1; cl->peer_state = PEER_CONNECTED; - cli->msgr.osd_peers[osd_num] = cl; - cli->msgr.clients[cl->client_id] = cl; - cli->msgr.wanted_peers.erase(osd_num); - cli->msgr.repeer_pgs(osd_num); + msgr->osd_peers[osd_num] = cl; + msgr->clients[cl->client_id] = cl; + msgr->wanted_peers.erase(osd_num); + msgr->repeer_pgs(osd_num); +} + +void pretend_disconnected(osd_messenger_t *msgr, osd_num_t osd_num) +{ + printf("OSD %ju disconnected\n", osd_num); + msgr->stop_client(msgr->osd_peers.at(osd_num)->client_id); +} + +void pretend_connected(cluster_client_t *cli, osd_num_t osd_num) +{ + pretend_connected(&cli->msgr, osd_num); } void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num) { - printf("OSD %ju disconnected\n", osd_num); - cli->msgr.stop_client(cli->msgr.osd_peers.at(osd_num)->client_id); + pretend_disconnected(&cli->msgr, osd_num); } void check_disconnected(cluster_client_t *cli, osd_num_t osd_num)