Compare commits

...

1 Commits

Author SHA1 Message Date
Vitaliy Filippov 0df51e8b21 Postpone handling incoming messages ? 2023-02-26 02:53:24 +03:00
3 changed files with 21 additions and 7 deletions

View File

@ -134,6 +134,7 @@ protected:
msgr_rdma_context_t *rdma_context = NULL;
uint64_t rdma_max_sge = 0, rdma_max_send = 0, rdma_max_recv = 0;
uint64_t rdma_max_msg = 0;
std::vector<rdma_hb_t> rdma_handle_buffers;
#endif
std::vector<int> read_ready_clients;

View File

@ -492,13 +492,9 @@ void osd_messenger_t::handle_rdma_events()
if (!is_send)
{
rc->cur_recv--;
if (!handle_read_buffer(cl, rc->recv_buffers[rc->next_recv_buf], wc[i].byte_len))
{
// handle_read_buffer may stop the client
continue;
}
try_recv_rdma_wr(cl, rc->recv_buffers[rc->next_recv_buf]);
rc->next_recv_buf = (rc->next_recv_buf+1) % rc->recv_buffers.size();
rdma_handle_buffers.push_back((rdma_hb_t){ .peer_fd = client_id, .buf = rc->recv_buffers[0], .len = wc[i].byte_len });
rc->recv_buffers.erase(rc->recv_buffers.begin(), rc->recv_buffers.begin()+1);
try_recv_rdma(cl);
}
else
{
@ -547,6 +543,16 @@ void osd_messenger_t::handle_rdma_events()
}
}
} while (event_count > 0);
for (auto & hb: rdma_handle_buffers)
{
auto cl_it = clients.find(hb.peer_fd);
if (cl_it != clients.end())
{
handle_read_buffer(cl_it->second, hb.buf, hb.len);
}
free(hb.buf);
}
rdma_handle_buffers.clear();
for (auto cb: set_immediate)
{
cb();

View File

@ -57,3 +57,10 @@ struct msgr_rdma_connection_t
static msgr_rdma_connection_t *create(msgr_rdma_context_t *ctx, uint32_t max_send, uint32_t max_recv, uint32_t max_sge, uint32_t max_msg);
int connect(msgr_rdma_address_t *dest);
};
struct rdma_hb_t
{
int peer_fd;
void *buf;
uint64_t len;
};