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

Rework msgr send/receive to allow encryption support

Browse Source
This commit is contained in:
Vitaliy Filippov
2026-04-03 21:22:42 +03:00
parent 9cf30f286e
commit 73e9de5143
8 changed files with 598 additions and 407 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/msgr_send.cpp
+176 -122
View File
@@ -5,6 +5,12 @@
#include <limits.h>
#include <sys/epoll.h>
#ifdef WITH_OPENSSL
#include <openssl/conf.h>
#include <openssl/evp.h>
#include <openssl/err.h>
#endif
#include "messenger.h"
void osd_messenger_t::outbox_push(osd_op_t *cur_op)
@@ -21,6 +27,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
{
clock_gettime(CLOCK_REALTIME, &cur_op->tv_begin);
cur_op->req.hdr.id = ++cl->send_op_id;
cl->sent_ops[cur_op->req.hdr.id] = cur_op;
}
else
{
@@ -37,77 +44,9 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
}
// Can't be not found because client IDs are unique
assert(found);
}
auto & to_send_list = cl->write_msg.msg_iovlen ? cl->next_send_list : cl->send_list;
auto & to_outbox = cl->write_msg.msg_iovlen ? cl->next_outbox : cl->outbox;
if (cur_op->op_type == OSD_OP_IN)
{
measure_exec(cur_op);
to_send_list.push_back((iovec){ .iov_base = cur_op->reply.buf, .iov_len = OSD_PACKET_SIZE });
}
else
{
to_send_list.push_back((iovec){ .iov_base = cur_op->req.buf, .iov_len = OSD_PACKET_SIZE });
cl->sent_ops[cur_op->req.hdr.id] = cur_op;
}
to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = MSGR_SENDP_HDR });
// Bitmap
if (cur_op->op_type == OSD_OP_IN &&
cur_op->req.hdr.opcode == OSD_OP_SEC_READ &&
cur_op->reply.sec_rw.attr_len > 0)
{
to_send_list.push_back((iovec){
.iov_base = cur_op->bitmap,
.iov_len = cur_op->reply.sec_rw.attr_len,
});
to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
}
else if (cur_op->op_type == OSD_OP_OUT &&
(cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE || cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) &&
cur_op->req.sec_rw.attr_len > 0)
{
to_send_list.push_back((iovec){
.iov_base = cur_op->bitmap,
.iov_len = cur_op->req.sec_rw.attr_len,
});
to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
}
// Operation data
if ((cur_op->op_type == OSD_OP_IN
? (cur_op->req.hdr.opcode == OSD_OP_READ ||
cur_op->req.hdr.opcode == OSD_OP_SEC_READ ||
cur_op->req.hdr.opcode == OSD_OP_SEC_LIST ||
cur_op->req.hdr.opcode == OSD_OP_SHOW_CONFIG ||
cur_op->req.hdr.opcode == OSD_OP_DESCRIBE)
: (cur_op->req.hdr.opcode == OSD_OP_WRITE ||
cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE ||
cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE ||
cur_op->req.hdr.opcode == OSD_OP_SEC_STABILIZE ||
cur_op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK ||
cur_op->req.hdr.opcode == OSD_OP_SHOW_CONFIG)) && cur_op->iov.count > 0)
{
for (int i = 0; i < cur_op->iov.count; i++)
{
if (cur_op->iov.buf[i].iov_len > 0)
{
assert(cur_op->iov.buf[i].iov_base);
to_send_list.push_back(cur_op->iov.buf[i]);
to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
}
}
}
if (cur_op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
{
if (cur_op->op_type == OSD_OP_IN && cur_op->reply.hdr.retval > 0)
to_send_list.push_back((iovec){ .iov_base = cur_op->buf, .iov_len = (size_t)cur_op->reply.hdr.retval });
else if (cur_op->op_type == OSD_OP_OUT && cur_op->req.sec_read_bmp.len > 0)
to_send_list.push_back((iovec){ .iov_base = cur_op->buf, .iov_len = (size_t)cur_op->req.sec_read_bmp.len });
to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
}
if (cur_op->op_type == OSD_OP_IN)
{
to_outbox[to_outbox.size()-1].flags |= MSGR_SENDP_FREE;
}
cl->write_ops.push_back(cur_op);
#ifdef WITH_RDMA
if (cl->peer_state == PEER_RDMA)
{
@@ -118,7 +57,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
if (!ringloop)
{
// FIXME: It's worse because it doesn't allow batching
while (cl->outbox.size())
while (cl->write_ops.size())
{
try_send(cl);
}
@@ -185,11 +124,25 @@ void osd_messenger_t::measure_exec(osd_op_t *cur_op)
bool osd_messenger_t::try_send(osd_client_t *cl)
{
if (!cl->send_list.size() || cl->write_msg.msg_iovlen > 0 || cl->peer_state == PEER_STOPPED || cl->peer_fd < 0)
if (!cl->write_op && !cl->write_ops.size() || cl->write_msg.msg_iovlen > 0 || cl->peer_state == PEER_STOPPED || cl->peer_fd < 0)
{
return true;
}
assert(cl->peer_state != PEER_RDMA);
while ((cl->write_op || cl->write_ops.size()) && cl->send_list.size() < IOV_MAX)
{
if (!cl->write_op)
{
cl->write_op = cl->write_ops.front();
cl->write_ops.pop_front();
}
osd_op_t *op = cl->write_op;
op_get_write_buffers(cl, cl->send_list);
if (!cl->write_op && op->op_type == OSD_OP_IN)
{
cl->send_free_ops.push_back(op);
}
}
if (ringloop && !use_sync_send_recv)
{
auto iothread = iothreads.size() ? iothreads[cl->peer_fd % iothreads.size()] : NULL;
@@ -202,20 +155,24 @@ bool osd_messenger_t::try_send(osd_client_t *cl)
data_local = {};
}
if (!sqe)
{
return false;
}
cl->send_list_size = 0;
for (auto & iov: cl->send_list)
{
cl->send_list_size += iov.iov_len;
}
cl->write_msg.msg_iov = cl->send_list.data();
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, data->prev, data->more, cl); };
bool use_zc = has_sendmsg_zc && min_zerocopy_send_size >= 0;
if (use_zc && min_zerocopy_send_size > 0)
if (use_zc && min_zerocopy_send_size > 0 &&
cl->send_list_size/cl->write_msg.msg_iovlen < min_zerocopy_send_size)
{
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;
use_zc = false;
}
if (use_zc)
{
@@ -266,6 +223,7 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
if (!prev)
{
cl->write_msg.msg_iovlen = 0;
cl->send_list.clear();
}
if (!more)
{
@@ -298,57 +256,26 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
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())
if (cl->send_list_size > result)
{
iovec & iov = cl->send_list[done];
if (iov.iov_len <= result)
{
if (cl->outbox[done].flags & MSGR_SENDP_FREE)
{
// Reply fully sent
if (more)
cl->zc_free_list.push_back(cl->outbox[done].op);
else
delete cl->outbox[done].op;
}
result -= iov.iov_len;
done++;
}
fprintf(stderr, "Client %ju socket write error: expected to send "
"%zu bytes with MSG_WAITALL but sent %u. Disconnecting client\n", cl->client_id, cl->send_list_size, result);
stop_client(cl->peer_fd);
return;
}
for (auto op: cl->send_free_ops)
{
if (more)
cl->zc_free_list.push_back(op);
else
{
iov.iov_len -= result;
iov.iov_base = (uint8_t*)iov.iov_base + result;
break;
}
delete op;
}
if (more)
{
int expected = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
if (done != expected)
{
fprintf(stderr, "Client %ju socket write error: expected to send "
"%d iovecs with MSG_WAITALL but sent %d. Disconnecting client\n", cl->client_id, expected, done);
stop_client(cl->client_id);
return;
}
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);
cl->outbox.erase(cl->outbox.begin(), cl->outbox.begin()+done);
}
if (cl->next_send_list.size())
{
cl->send_list.insert(cl->send_list.end(), cl->next_send_list.begin(), cl->next_send_list.end());
cl->outbox.insert(cl->outbox.end(), cl->next_outbox.begin(), cl->next_outbox.end());
cl->next_send_list.clear();
cl->next_outbox.clear();
}
cl->write_state = cl->outbox.size() > 0 ? CL_WRITE_READY : 0;
cl->send_free_ops.clear();
cl->write_state = cl->write_op || cl->write_ops.size() ? CL_WRITE_READY : 0;
#ifdef WITH_RDMA
if (cl->rdma_conn && !cl->outbox.size() && cl->peer_state == PEER_RDMA_CONNECTING)
if (cl->rdma_conn && !cl->write_op && !cl->write_ops.size() && cl->peer_state == PEER_RDMA_CONNECTING)
{
// FIXME: Ignore pings during RDMA state transition
if (log_level > 0)
@@ -366,3 +293,130 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
write_ready_clients.push_back(cl->client_id);
}
}
static inline bool op_write_headers(osd_op_t *op, std::function<bool(uint8_t*, size_t)> op_write_buf)
{
// Header
if (!op_write_buf((op->op_type == OSD_OP_IN ? op->reply.buf : op->req.buf), OSD_PACKET_SIZE))
return false;
// Bitmap
if (op->op_type == OSD_OP_IN &&
op->req.hdr.opcode == OSD_OP_SEC_READ &&
op->reply.sec_rw.attr_len > 0)
{
if (!op_write_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len))
return false;
}
else if (op->op_type == OSD_OP_OUT &&
(op->req.hdr.opcode == OSD_OP_SEC_WRITE || op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) &&
op->req.sec_rw.attr_len > 0)
{
if (!op_write_buf((uint8_t*)op->bitmap, op->req.sec_rw.attr_len))
return false;
}
if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
{
if (op->op_type == OSD_OP_IN && op->reply.hdr.retval > 0)
{
if (!op_write_buf((uint8_t*)op->buf, (size_t)op->reply.hdr.retval))
return false;
}
else if (op->op_type == OSD_OP_OUT && op->req.sec_read_bmp.len > 0)
{
if (!op_write_buf((uint8_t*)op->buf, (size_t)op->req.sec_read_bmp.len))
return false;
}
}
return true;
}
static inline bool op_has_data(osd_op_t *op)
{
return (op->op_type == OSD_OP_IN
? (op->req.hdr.opcode == OSD_OP_READ ||
op->req.hdr.opcode == OSD_OP_SEC_READ ||
op->req.hdr.opcode == OSD_OP_SEC_LIST ||
op->req.hdr.opcode == OSD_OP_SHOW_CONFIG ||
op->req.hdr.opcode == OSD_OP_DESCRIBE)
: (op->req.hdr.opcode == OSD_OP_WRITE ||
op->req.hdr.opcode == OSD_OP_SEC_WRITE ||
op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE ||
op->req.hdr.opcode == OSD_OP_SEC_STABILIZE ||
op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK ||
op->req.hdr.opcode == OSD_OP_SHOW_CONFIG)) && op->iov.count > 0;
}
size_t osd_messenger_t::op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_len)
{
size_t done = 0;
size_t from = cl->write_op_pos;
auto op_write_buf = [&](uint8_t *src, size_t src_len)
{
if (from < src_len)
{
size_t n = src_len-from;
if (n > dst_len-done)
n = dst_len-done;
memcpy(dst+done, src+from, n);
done += n;
cl->write_op_pos += n;
from += n;
if (from < src_len)
return false;
from = 0;
}
else
from -= src_len;
return true;
};
if (!op_write_headers(cl->write_op, op_write_buf))
{
return done;
}
// Operation data
if (op_has_data(cl->write_op))
{
for (int i = 0; i < cl->write_op->iov.count; i++)
{
if (!op_write_buf((uint8_t*)cl->write_op->iov.buf[i].iov_base, cl->write_op->iov.buf[i].iov_len))
return done;
}
}
cl->write_op = NULL;
cl->write_op_pos = 0;
return done;
}
void osd_messenger_t::op_get_write_buffers(osd_client_t *cl, std::vector<iovec> & lst)
{
size_t from = cl->write_op_pos;
auto op_write_buf = [&](uint8_t *src, size_t src_len)
{
if (lst.size() >= IOV_MAX)
return false;
if (from < src_len)
{
lst.push_back((iovec){ .iov_base = src+from, .iov_len = src_len-from });
cl->write_op_pos += src_len-from;
from = 0;
}
else
from -= src_len;
return true;
};
if (!op_write_headers(cl->write_op, op_write_buf))
{
return;
}
// Operation data
if (op_has_data(cl->write_op))
{
for (int i = 0; i < cl->write_op->iov.count; i++)
{
if (!op_write_buf((uint8_t*)cl->write_op->iov.buf[i].iov_base, cl->write_op->iov.buf[i].iov_len))
return;
}
}
cl->write_op = NULL;
cl->write_op_pos = 0;
}