diff --git a/src/client/cluster_client.h b/src/client/cluster_client.h index 23a8b05c..52dafb32 100644 --- a/src/client/cluster_client.h +++ b/src/client/cluster_client.h @@ -155,9 +155,6 @@ public: void list_inode(inode_t inode, uint64_t min_offset, uint64_t max_offset, int max_parallel_pgs, std::function&& objects)> pg_callback); - //inline uint32_t get_bs_bitmap_granularity() { return st_cli.global_bitmap_granularity; } - //inline uint64_t get_bs_block_size() { return st_cli.global_block_size; } - #ifndef __MOCK__ protected: #endif diff --git a/src/client/messenger.h b/src/client/messenger.h index c7c30bea..f4f7d319 100644 --- a/src/client/messenger.h +++ b/src/client/messenger.h @@ -179,7 +179,7 @@ protected: public: timerfd_manager_t *tfd = NULL; - ring_loop_t *ringloop = NULL; + ring_loop_i *ringloop = NULL; bool has_sendmsg_zc = false; // osd_num_t is only for logging and asserts uint64_t next_client_id = 1; diff --git a/src/osd/osd.cpp b/src/osd/osd.cpp index b1384b63..6e5a0cf5 100644 --- a/src/osd/osd.cpp +++ b/src/osd/osd.cpp @@ -15,22 +15,19 @@ #include "str_util.h" #include "json_util.h" -osd_t::osd_t(const json11::Json & config, ring_loop_t *ringloop) +osd_t::osd_t(const json11::Json & config, ring_loop_i *ringloop, timerfd_manager_t *tfd) { zero_buffer_size = 1<<20; zero_buffer = malloc_or_die(zero_buffer_size); memset(zero_buffer, 0, zero_buffer_size); this->ringloop = ringloop; + this->tfd = tfd; this->cli_config = config.object_items(); this->file_config = msgr.read_config(this->cli_config); parse_config(true); - epmgr = new epoll_manager_t(ringloop); - // FIXME: Use timerfd_interval based directly on io_uring - this->tfd = epmgr->tfd; - if (json_is_true(this->config["osd_memlock"])) { // Lock all OSD memory if requested @@ -99,7 +96,6 @@ osd_t::~osd_t() } ringloop->unregister_consumer(&consumer); ringloop->unregister_consumer(&init_consumer); - delete epmgr; if (bs) delete bs; #ifdef WITH_RDMACM @@ -398,7 +394,7 @@ void osd_t::bind_socket() { int listen_fd = create_and_bind_socket(bind_address, listening_port ? listening_port : bind_port, listen_backlog, &listening_port); fcntl(listen_fd, F_SETFL, fcntl(listen_fd, F_GETFL, 0) | O_NONBLOCK); - epmgr->set_fd_handler(listen_fd, false, [this](int fd, int events) + tfd->set_fd_handler(listen_fd, false, [this](int fd, int events) { msgr.accept_connections(fd); }); diff --git a/src/osd/osd.h b/src/osd/osd.h index 2c5430ee..d5fd458b 100644 --- a/src/osd/osd.h +++ b/src/osd/osd.h @@ -19,7 +19,6 @@ #include "blockstore.h" #include "ringloop.h" #include "timerfd_manager.h" -#include "epoll_manager.h" #include "osd_peering_pg.h" #include "messenger.h" #include "etcd_state_client.h" @@ -210,9 +209,8 @@ class osd_t void *zero_buffer = NULL; uint64_t zero_buffer_size = 0; uint32_t bs_block_size, bs_bitmap_granularity, clean_entry_bitmap_size; - ring_loop_t *ringloop = NULL; + ring_loop_i *ringloop = NULL; timerfd_manager_t *tfd = NULL; - epoll_manager_t *epmgr = NULL; int listening_port = 0; std::vector bind_addresses; @@ -392,7 +390,7 @@ class osd_t } public: - osd_t(const json11::Json & config, ring_loop_t *ringloop); + osd_t(const json11::Json & config, ring_loop_i *ringloop, timerfd_manager_t *tfd); ~osd_t(); void force_stop(int exitcode); bool shutdown(); diff --git a/src/osd/osd_main.cpp b/src/osd/osd_main.cpp index 79ea4a6d..826190ee 100644 --- a/src/osd/osd_main.cpp +++ b/src/osd/osd_main.cpp @@ -1,6 +1,7 @@ // Copyright (c) Vitaliy Filippov, 2019+ // License: VNPL-1.1 (see README.md for details) +#include "epoll_manager.h" #include "osd.h" #include @@ -63,13 +64,15 @@ int main(int narg, char *args[]) signal(SIGINT, handle_sigint); signal(SIGTERM, handle_sigint); ring_loop_t *ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE); - osd = new osd_t(config, ringloop); + epoll_manager_t *epmgr = new epoll_manager_t(ringloop); + osd = new osd_t(config, ringloop, epmgr->tfd); while (1) { ringloop->loop(); ringloop->wait(); } delete osd; + delete epmgr; delete ringloop; return 0; } diff --git a/src/test/CMakeLists.txt b/src/test/CMakeLists.txt index 6d9765f6..57057c50 100644 --- a/src/test/CMakeLists.txt +++ b/src/test/CMakeLists.txt @@ -4,10 +4,9 @@ project(vitastor) ### Test stubs -# stub_osd, stub_bench, osd_test +# stub_osd, stub_bench add_executable(stub_osd stub_osd.cpp ../util/rw_blocking.cpp ../util/addr_util.cpp) add_executable(stub_bench stub_bench.cpp ../util/rw_blocking.cpp ../util/addr_util.cpp) -add_executable(osd_test osd_test.cpp ../util/rw_blocking.cpp ../util/addr_util.cpp) # bindiff add_executable(bindiff diff --git a/src/test/mock/ringloop.h b/src/test/mock/ringloop.h deleted file mode 100644 index 6ed6ed0b..00000000 --- a/src/test/mock/ringloop.h +++ /dev/null @@ -1,31 +0,0 @@ -// Copyright (c) Vitaliy Filippov, 2019+ -// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details) - -#pragma once - -#include - -struct ring_consumer_t -{ - std::function loop; -}; - -class ring_loop_t -{ -public: - void register_consumer(ring_consumer_t *consumer) - { - } - void unregister_consumer(ring_consumer_t *consumer) - { - } - void submit() - { - } - void wait() - { - } - void loop() - { - } -}; diff --git a/src/test/osd_test.cpp b/src/test/osd_test.cpp deleted file mode 100644 index a4b9c0a4..00000000 --- a/src/test/osd_test.cpp +++ /dev/null @@ -1,389 +0,0 @@ -// Copyright (c) Vitaliy Filippov, 2019+ -// License: VNPL-1.1 (see README.md for details) - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include - -#include "malloc_or_die.h" -#include "addr_util.h" -#include "osd_ops.h" -#include "rw_blocking.h" -#include "test_pattern.h" - -int connect_osd(const char *osd_address, int osd_port); - -uint64_t test_read(int connect_fd, uint64_t inode, uint64_t stripe, uint64_t version, uint64_t offset, uint64_t len); - -uint64_t test_write(int connect_fd, uint64_t inode, uint64_t stripe, uint64_t version, uint64_t pattern); - -void* test_primary_read(int connect_fd, uint64_t inode, uint64_t offset, uint64_t len); - -void test_primary_write(int connect_fd, uint64_t inode, uint64_t offset, uint64_t len, uint64_t pattern); - -void test_primary_sync(int connect_fd); - -void test_sync_stab_all(int connect_fd); - -void test_list_stab(int connect_fd); - -int main0(int narg, char *args[]) -{ - int connect_fd; - // Prepare data for cluster read - connect_fd = connect_osd("127.0.0.1", 11203); - test_write(connect_fd, 2, 0, 1, PATTERN0); - close(connect_fd); - connect_fd = connect_osd("127.0.0.1", 11204); - test_write(connect_fd, 2, 1, 1, PATTERN1); - close(connect_fd); - connect_fd = connect_osd("127.0.0.1", 11205); - test_write(connect_fd, 2, 2, 1, PATTERN0^PATTERN1); - close(connect_fd); - return 0; -} - -int main1(int narg, char *args[]) -{ - int connect_fd; - void *data; - // Cluster read - connect_fd = connect_osd("127.0.0.1", 11203); - data = test_primary_read(connect_fd, 2, 0, 128*1024); - if (data) - { - check_pattern(data, 128*1024, PATTERN0); - printf("inode=2 0-128K OK\n"); - free(data); - } - data = test_primary_read(connect_fd, 2, 0, 256*1024); - if (data) - { - check_pattern(data, 128*1024, PATTERN0); - check_pattern(data+128*1024, 128*1024, PATTERN1); - printf("inode=2 0-256K OK\n"); - free(data); - } - close(connect_fd); - return 0; -} - -int main2(int narg, char *args[]) -{ - int connect_fd; - // Cluster write (sync not implemented yet) - connect_fd = connect_osd("127.0.0.1", 11203); - test_primary_write(connect_fd, 2, 0, 128*1024, PATTERN0); - test_primary_write(connect_fd, 2, 128*1024, 128*1024, PATTERN1); - test_sync_stab_all(connect_fd); - close(connect_fd); - connect_fd = connect_osd("127.0.0.1", 11204); - if (connect_fd >= 0) - { - test_sync_stab_all(connect_fd); - close(connect_fd); - } - connect_fd = connect_osd("127.0.0.1", 11205); - if (connect_fd >= 0) - { - test_sync_stab_all(connect_fd); - close(connect_fd); - } - return 0; -} - -int main3(int narg, char *args[]) -{ - int connect_fd; - connect_fd = connect_osd("127.0.0.1", 11203); - test_list_stab(connect_fd); - close(connect_fd); - return 0; -} - -int main4(int narg, char *args[]) -{ - int connect_fd; - // Cluster write (sync not implemented yet) - connect_fd = connect_osd("127.0.0.1", 11203); - test_primary_write(connect_fd, 2, 0, 128*1024, PATTERN0); - test_primary_write(connect_fd, 2, 128*1024, 128*1024, PATTERN1); - test_primary_sync(connect_fd); - close(connect_fd); - return 0; -} - -int main(int narg, char *args[]) -{ - int connect_fd; - connect_fd = connect_osd("192.168.7.2", 43051); - test_read(connect_fd, 1, 1039663104, UINT64_MAX, 0, 128*1024); - close(connect_fd); - return 0; -} - -int connect_osd(const char *osd_address, int osd_port) -{ - struct sockaddr_storage addr; - if (!string_to_addr(osd_address, 0, osd_port, &addr)) - { - fprintf(stderr, "server address: %s is not valid\n", osd_address); - return -1; - } - - int connect_fd = socket(addr.ss_family, SOCK_STREAM, 0); - if (connect_fd < 0) - { - perror("socket"); - return -1; - } - if (connect(connect_fd, (sockaddr*)&addr, sizeof(addr)) < 0) - { - perror("connect"); - close(connect_fd); - return -1; - } - int one = 1; - setsockopt(connect_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one)); - - return connect_fd; -} - -bool check_reply(int r, osd_any_op_t & op, osd_any_reply_t & reply, int expected) -{ - if (r != OSD_PACKET_SIZE) - { - printf("read failed\n"); - return false; - } - if (reply.hdr.magic != SECONDARY_OSD_REPLY_MAGIC || - reply.hdr.id != op.hdr.id || reply.hdr.opcode != op.hdr.opcode) - { - printf("bad reply: magic, id or opcode does not match request\n"); - return false; - } - if (expected >= 0 && reply.hdr.retval != expected) - { - printf("operation failed, retval=%jd\n", reply.hdr.retval); - return false; - } - return true; -} - -uint64_t test_read(int connect_fd, uint64_t inode, uint64_t stripe, uint64_t version, uint64_t offset, uint64_t len) -{ - osd_any_op_t op; - osd_any_reply_t reply; - op.hdr.magic = SECONDARY_OSD_OP_MAGIC; - op.hdr.id = 1; - op.hdr.opcode = OSD_OP_SEC_READ; - op.sec_rw.oid = { - .inode = inode, - .stripe = stripe, - }; - op.sec_rw.version = version; - op.sec_rw.offset = offset; - op.sec_rw.len = len; - void *data = memalign_or_die(MEM_ALIGNMENT, op.sec_rw.len); - write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); - int r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); - if (!check_reply(r, op, reply, op.sec_rw.len)) - { - free(data); - return 0; - } - r = read_blocking(connect_fd, data, len); - if (r != len) - { - free(data); - perror("read data"); - return 0; - } - free(data); - printf("Read %jx:%jx v%ju = v%ju\n", inode, stripe, version, reply.sec_rw.version); - op.hdr.opcode = OSD_OP_SEC_LIST; - op.sec_list.list_pg = 1; - op.sec_list.pg_count = 1; - op.sec_list.pg_stripe_size = 4*1024*1024; - write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); - r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); - if (reply.hdr.retval < 0 || !check_reply(r, op, reply, reply.hdr.retval)) - { - return 0; - } - data = memalign_or_die(MEM_ALIGNMENT, sizeof(obj_ver_id)*reply.hdr.retval); - r = read_blocking(connect_fd, data, sizeof(obj_ver_id)*reply.hdr.retval); - if (r != sizeof(obj_ver_id)*reply.hdr.retval) - { - free(data); - perror("read data"); - return 0; - } - obj_ver_id *ov = (obj_ver_id*)data; - for (int i = 0; i < reply.hdr.retval; i++) - { - if (ov[i].oid.inode == inode && (ov[i].oid.stripe & ~(4096-1)) == (stripe & ~(4096-1))) - { - printf("list: %jx:%jx v%ju stable=%d\n", ov[i].oid.inode, ov[i].oid.stripe, ov[i].version, i < reply.sec_list.stable_count ? 1 : 0); - } - } - return 0; -} - -uint64_t test_write(int connect_fd, uint64_t inode, uint64_t stripe, uint64_t version, uint64_t pattern) -{ - osd_any_op_t op; - osd_any_reply_t reply; - op.hdr.magic = SECONDARY_OSD_OP_MAGIC; - op.hdr.id = 1; - op.hdr.opcode = OSD_OP_SEC_WRITE; - op.sec_rw.oid = { - .inode = inode, - .stripe = stripe, - }; - op.sec_rw.version = version; - op.sec_rw.offset = 0; - op.sec_rw.len = 128*1024; - void *data = memalign_or_die(MEM_ALIGNMENT, op.sec_rw.len); - for (int i = 0; i < (op.sec_rw.len)/sizeof(uint64_t); i++) - ((uint64_t*)data)[i] = pattern; - write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); - write_blocking(connect_fd, data, op.sec_rw.len); - int r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); - if (!check_reply(r, op, reply, op.sec_rw.len)) - { - free(data); - return 0; - } - version = reply.sec_rw.version; - op.hdr.opcode = OSD_OP_TEST_SYNC_STAB_ALL; - op.hdr.id = 2; - write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); - r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); - if (!check_reply(r, op, reply, 0)) - { - free(data); - return 0; - } - free(data); - return version; -} - -void* test_primary_read(int connect_fd, uint64_t inode, uint64_t offset, uint64_t len) -{ - osd_any_op_t op; - osd_any_reply_t reply; - op.hdr.magic = SECONDARY_OSD_OP_MAGIC; - op.hdr.id = 1; - op.hdr.opcode = OSD_OP_READ; - op.rw.inode = inode; - op.rw.offset = offset; - op.rw.len = len; - void *data = memalign_or_die(MEM_ALIGNMENT, len); - write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); - int r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); - if (!check_reply(r, op, reply, len)) - { - free(data); - return NULL; - } - r = read_blocking(connect_fd, data, len); - if (r != len) - { - free(data); - perror("read data"); - return NULL; - } - return data; -} - -void test_primary_write(int connect_fd, uint64_t inode, uint64_t offset, uint64_t len, uint64_t pattern) -{ - osd_any_op_t op; - osd_any_reply_t reply; - op.hdr.magic = SECONDARY_OSD_OP_MAGIC; - op.hdr.id = 1; - op.hdr.opcode = OSD_OP_WRITE; - op.rw.inode = inode; - op.rw.offset = offset; - op.rw.len = len; - void *data = memalign_or_die(MEM_ALIGNMENT, len); - set_pattern(data, len, pattern); - write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); - write_blocking(connect_fd, data, len); - free(data); - int r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); - assert(check_reply(r, op, reply, len)); -} - -void test_primary_sync(int connect_fd) -{ - osd_any_op_t op; - osd_any_reply_t reply; - op.hdr.magic = SECONDARY_OSD_OP_MAGIC; - op.hdr.id = 1; - op.hdr.opcode = OSD_OP_SYNC; - write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); - int r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); - assert(check_reply(r, op, reply, 0)); -} - -void test_sync_stab_all(int connect_fd) -{ - osd_any_op_t op; - osd_any_reply_t reply; - op.hdr.magic = SECONDARY_OSD_OP_MAGIC; - op.hdr.id = 1; - op.hdr.opcode = OSD_OP_TEST_SYNC_STAB_ALL; - write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); - int r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); - assert(check_reply(r, op, reply, 0)); -} - -void test_list_stab(int connect_fd) -{ - osd_any_op_t op; - osd_any_reply_t reply; - op.hdr.magic = SECONDARY_OSD_OP_MAGIC; - op.hdr.id = 1; - op.hdr.opcode = OSD_OP_SEC_LIST; - op.sec_list.pg_count = 0; - assert(write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE) == OSD_PACKET_SIZE); - int r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); - assert(check_reply(r, op, reply, -1)); - int total_count = reply.hdr.retval; - int stable_count = reply.sec_list.stable_count; - obj_ver_id *data = (obj_ver_id*)malloc_or_die(total_count * sizeof(obj_ver_id)); - assert(data); - assert(read_blocking(connect_fd, data, total_count * sizeof(obj_ver_id)) == (total_count * sizeof(obj_ver_id))); - int last_start = stable_count; - for (int i = stable_count; i <= total_count; i++) - { - // Stabilize in portions of 32 entries - if (i - last_start >= 32 || i == total_count) - { - op.hdr.opcode = OSD_OP_SEC_STABILIZE; - op.sec_stab.len = sizeof(obj_ver_id) * (i - last_start); - assert(write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE) == OSD_PACKET_SIZE); - assert(write_blocking(connect_fd, data + last_start, op.sec_stab.len) == op.sec_stab.len); - r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); - assert(check_reply(r, op, reply, 0)); - last_start = i; - } - } - obj_ver_id *data2 = (obj_ver_id*)malloc_or_die(sizeof(obj_ver_id) * 32); - assert(data2); - free(data2); - free(data); -} diff --git a/src/util/timerfd_manager.cpp b/src/util/timerfd_manager.cpp index e64ebc88..0a991eda 100644 --- a/src/util/timerfd_manager.cpp +++ b/src/util/timerfd_manager.cpp @@ -16,6 +16,7 @@ timerfd_manager_t::timerfd_manager_t(std::functionset_fd_handler = set_fd_handler; wait_state = 0; + // FIXME: Use timerfd_interval based directly on io_uring if (set_fd_handler) { timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK);