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

Remove unused osd_test.cpp and mock/ringloop.h, move epmgr out of osd_t

Browse Source
This commit is contained in:
Vitaliy Filippov
2026-05-18 01:17:56 +03:00
parent c9dfd0f67d
commit a675993c74
9 changed files with 12 additions and 438 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/cluster_client.h
-3
View File
@@ -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<void(
int status, int pgs_left, pg_num_t pg_num, std::set<object_id>&& 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
src/client/messenger.h
+1 -1
View File
@@ -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;
src/osd/osd.cpp
+3 -7
View File
@@ -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);
});
src/osd/osd.h
+2 -4
View File
@@ -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<std::string> 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();
src/osd/osd_main.cpp
+4 -1
View File
@@ -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 <sys/prctl.h>
@@ -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;
}
src/test/CMakeLists.txt
+1 -2
View File
@@ -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
src/test/mock/ringloop.h
-31
View File
@@ -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 <functional>
struct ring_consumer_t
{
std::function<void(void)> 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()
{
}
};
src/test/osd_test.cpp
-389
View File
@@ -1,389 +0,0 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <stdlib.h>
#include <malloc.h>
#include <stdexcept>
#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);
}
src/util/timerfd_manager.cpp
+1
View File
@@ -16,6 +16,7 @@ timerfd_manager_t::timerfd_manager_t(std::function<void(int, bool, std::function
{
this->set_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);