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

Implement io_uring zero-copy send support

Browse Source
This commit is contained in:
Vitaliy Filippov
2025-05-01 18:47:10 +03:00
parent 96b5a72630
commit 9556eeae45
8 changed files with 215 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/messenger.cpp
+8
View File
@@ -167,6 +167,10 @@ void osd_messenger_t::init()
}
}
#endif
if (ringloop)
{
has_sendmsg_zc = ringloop->has_sendmsg_zc();
}
if (ringloop && iothread_count > 0)
{
for (int i = 0; i < iothread_count; i++)
@@ -329,6 +333,9 @@ void osd_messenger_t::parse_config(const json11::Json & config)
this->receive_buffer_size = 65536;
this->use_sync_send_recv = config["use_sync_send_recv"].bool_value() ||
config["use_sync_send_recv"].uint64_value();
this->min_zerocopy_send_size = config["min_zerocopy_send_size"].is_null()
? DEFAULT_MIN_ZEROCOPY_SEND_SIZE
: (int)config["min_zerocopy_send_size"].int64_value();
this->peer_connect_interval = config["peer_connect_interval"].uint64_value();
if (!this->peer_connect_interval)
this->peer_connect_interval = 5;
@@ -897,6 +904,7 @@ static const char* local_only_params[] = {
"tcp_header_buffer_size",
"use_rdma",
"use_sync_send_recv",
"min_zerocopy_send_size",
};
static const char **local_only_end = local_only_params + (sizeof(local_only_params)/sizeof(local_only_params[0]));
src/client/messenger.h
+8 -3
View File
@@ -32,6 +32,8 @@
#define VITASTOR_CONFIG_PATH "/etc/vitastor/vitastor.conf"
#define DEFAULT_MIN_ZEROCOPY_SEND_SIZE 32*1024
#define MSGR_SENDP_HDR 1
#define MSGR_SENDP_FREE 2
@@ -88,6 +90,7 @@ struct osd_client_t
int write_state = 0;
std::vector<iovec> send_list, next_send_list;
std::vector<msgr_sendp_t> outbox, next_outbox;
std::vector<osd_op_t*> zc_free_list;
~osd_client_t();
};
@@ -176,6 +179,7 @@ protected:
int osd_ping_timeout = 0;
int log_level = 0;
bool use_sync_send_recv = false;
int min_zerocopy_send_size = DEFAULT_MIN_ZEROCOPY_SEND_SIZE;
int iothread_count = 0;
#ifdef WITH_RDMA
@@ -202,8 +206,9 @@ protected:
std::vector<osd_op_t*> set_immediate_ops;
public:
timerfd_manager_t *tfd;
ring_loop_t *ringloop;
timerfd_manager_t *tfd = NULL;
ring_loop_t *ringloop = NULL;
bool has_sendmsg_zc = false;
// osd_num_t is only for logging and asserts
osd_num_t osd_num;
uint64_t next_subop_id = 1;
@@ -262,7 +267,7 @@ protected:
void cancel_op(osd_op_t *op);
bool try_send(osd_client_t *cl);
void handle_send(int result, osd_client_t *cl);
void handle_send(int result, bool prev, bool more, osd_client_t *cl);
bool handle_read(int result, osd_client_t *cl);
bool handle_read_buffer(osd_client_t *cl, void *curbuf, int remain);
src/client/msgr_send.cpp
+47 -7
View File
@@ -203,8 +203,24 @@ bool osd_messenger_t::try_send(osd_client_t *cl)
cl->write_msg.msg_iovlen = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
cl->refs++;
ring_data_t* data = ((ring_data_t*)sqe->user_data);
data->callback = [this, cl](ring_data_t *data) { handle_send(data->res, cl); };
my_uring_prep_sendmsg(sqe, peer_fd, &cl->write_msg, 0);
data->callback = [this, cl](ring_data_t *data) { handle_send(data->res, data->prev, data->more, cl); };
bool use_zc = has_sendmsg_zc && min_zerocopy_send_size >= 0;
if (use_zc && min_zerocopy_send_size > 0)
{
size_t avg_size = 0;
for (size_t i = 0; i < cl->write_msg.msg_iovlen; i++)
avg_size += cl->write_msg.msg_iov[i].iov_len;
if (avg_size/cl->write_msg.msg_iovlen < min_zerocopy_send_size)
use_zc = false;
}
if (use_zc)
{
my_uring_prep_sendmsg_zc(sqe, peer_fd, &cl->write_msg, MSG_WAITALL);
}
else
{
my_uring_prep_sendmsg(sqe, peer_fd, &cl->write_msg, MSG_WAITALL);
}
if (iothread)
{
iothread->add_sqe(sqe_local);
@@ -220,7 +236,7 @@ bool osd_messenger_t::try_send(osd_client_t *cl)
{
result = -errno;
}
handle_send(result, cl);
handle_send(result, false, false, cl);
}
return true;
}
@@ -240,10 +256,16 @@ void osd_messenger_t::send_replies()
write_ready_clients.clear();
}
void osd_messenger_t::handle_send(int result, osd_client_t *cl)
void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t *cl)
{
cl->write_msg.msg_iovlen = 0;
cl->refs--;
if (!prev)
{
cl->write_msg.msg_iovlen = 0;
}
if (!more)
{
cl->refs--;
}
if (cl->peer_state == PEER_STOPPED)
{
if (cl->refs <= 0)
@@ -261,6 +283,16 @@ void osd_messenger_t::handle_send(int result, osd_client_t *cl)
}
if (result >= 0)
{
if (prev)
{
// Second notification - only free a batch of postponed ops
int i = 0;
for (; i < cl->zc_free_list.size() && cl->zc_free_list[i]; i++)
delete cl->zc_free_list[i];
if (i > 0)
cl->zc_free_list.erase(cl->zc_free_list.begin(), cl->zc_free_list.begin()+i+1);
return;
}
int done = 0;
while (result > 0 && done < cl->send_list.size())
{
@@ -270,7 +302,10 @@ void osd_messenger_t::handle_send(int result, osd_client_t *cl)
if (cl->outbox[done].flags & MSGR_SENDP_FREE)
{
// Reply fully sent
delete cl->outbox[done].op;
if (more)
cl->zc_free_list.push_back(cl->outbox[done].op);
else
delete cl->outbox[done].op;
}
result -= iov.iov_len;
done++;
@@ -282,6 +317,11 @@ void osd_messenger_t::handle_send(int result, osd_client_t *cl)
break;
}
}
if (more)
{
assert(done == cl->send_list.size());
cl->zc_free_list.push_back(NULL); // end marker
}
if (done > 0)
{
cl->send_list.erase(cl->send_list.begin(), cl->send_list.begin()+done);
src/util/ringloop.cpp
+28 -1
View File
@@ -10,6 +10,10 @@
#include "ringloop.h"
#ifndef IORING_CQE_F_MORE
#define IORING_CQE_F_MORE (1U << 1)
#endif
ring_loop_t::ring_loop_t(int qd, bool multithreaded)
{
mt = multithreaded;
@@ -30,6 +34,16 @@ ring_loop_t::ring_loop_t(int qd, bool multithreaded)
free_ring_data[i] = i;
}
in_loop = false;
auto probe = io_uring_get_probe();
if (probe)
{
support_zc = io_uring_opcode_supported(probe, IORING_OP_SENDMSG_ZC);
#ifdef IORING_SETUP_R_DISABLED /* liburing 2.0 check */
io_uring_free_probe(probe);
#else
free(probe);
#endif
}
}
ring_loop_t::~ring_loop_t()
@@ -108,7 +122,17 @@ void ring_loop_t::loop()
if (mt)
mu.lock();
struct ring_data_t *d = (struct ring_data_t*)cqe->user_data;
if (d->callback)
if (cqe->flags & IORING_CQE_F_MORE)
{
// There will be a second notification
d->res = cqe->res;
d->more = true;
if (d->callback)
d->callback(d);
d->prev = true;
d->more = false;
}
else if (d->callback)
{
// First free ring_data item, then call the callback
// so it has at least 1 free slot for the next event
@@ -116,7 +140,10 @@ void ring_loop_t::loop()
struct ring_data_t dl;
dl.iov = d->iov;
dl.res = cqe->res;
dl.more = false;
dl.prev = d->prev;
dl.callback.swap(d->callback);
d->prev = d->more = false;
free_ring_data[free_ring_data_ptr++] = d - ring_datas;
if (mt)
mu.unlock();
src/util/ringloop.h
+17
View File
@@ -18,6 +18,10 @@
#define RINGLOOP_DEFAULT_SIZE 1024
#ifndef IORING_RECV_MULTISHOT /* liburing-2.3 check */
#define IORING_OP_SENDMSG_ZC 48
#endif
static inline void my_uring_prep_rw(int op, struct io_uring_sqe *sqe, int fd, const void *addr, unsigned len, off_t offset)
{
// Prepare a read/write operation without clearing user_data
@@ -62,6 +66,12 @@ static inline void my_uring_prep_sendmsg(struct io_uring_sqe *sqe, int fd, const
sqe->msg_flags = flags;
}
static inline void my_uring_prep_sendmsg_zc(struct io_uring_sqe *sqe, int fd, const struct msghdr *msg, unsigned flags)
{
my_uring_prep_rw(IORING_OP_SENDMSG_ZC, sqe, fd, msg, 1, 0);
sqe->msg_flags = flags;
}
static inline void my_uring_prep_poll_add(struct io_uring_sqe *sqe, int fd, short poll_mask)
{
my_uring_prep_rw(IORING_OP_POLL_ADD, sqe, fd, NULL, 0, 0);
@@ -112,6 +122,8 @@ struct ring_data_t
{
struct iovec iov; // for single-entry read/write operations
int res;
bool prev: 1;
bool more: 1;
std::function<void(ring_data_t*)> callback;
};
@@ -133,6 +145,7 @@ class ring_loop_t
bool loop_again;
struct io_uring ring;
int ring_eventfd = -1;
bool support_zc = false;
public:
ring_loop_t(int qd, bool multithreaded = false);
~ring_loop_t();
@@ -163,6 +176,10 @@ public:
{
return loop_again;
}
inline bool has_sendmsg_zc()
{
return support_zc;
}
void loop();
void wakeup();