Skip to content

Commit c715949

Browse files
yanglimingcnyanglimingcn
committed
rdma support polling mode
1 parent df0fbdc commit c715949

5 files changed

Lines changed: 212 additions & 39 deletions

File tree

‎docs/cn/rdma.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,8 @@ RDMA要求数据收发所使用的内存空间必须被注册（memory register
4747

4848
RDMA是硬件相关的通信技术，有很多独特的概念，比如device、port、GID、LID、MaxSge等。这些参数在初始化时会从对应的网卡中读取出来，并且做出默认的选择（参见src/brpc/rdma/rdma_helper.cpp）。有时默认的选择并非用户的期望，则可以通过flag参数方式指定。
4949

50+
RDMA支持事件驱动和轮询两种模式，默认是事件驱动模式，通过设置rdma_use_polling可以开启轮询模式。轮询模式下还可以设置轮询器数目（rdma_poller_num），以及是否主动放弃CPU（rdma_poller_yield）。轮询模式下还可以设置一个回调函数，在每次轮询时调用，可以配合io_uring/spdk等使用。在配合使用spdk等驱动的时候，因为spdk只能使用轮询模式，并且只能在单线程使用，所以这时候需要设置（rdma_enable_edisp_schedule）不能被调度。
51+
5052
# 参数
5153

5254
可配置参数说明：
@@ -68,3 +70,7 @@ RDMA是硬件相关的通信技术，有很多独特的概念，比如device、p
6870
* rdma_memory_pool_max_regions: 最大的内存池块数，默认16
6971
* rdma_memory_pool_buckets: 内存池中为避免竞争采用的bucket数目，默认为4
7072
* rdma_memory_pool_tls_cache_num: 内存池中thread local的缓存block数目，默认为128
73+
* rdma_use_polling: 是否使用RDMA的轮询模式，默认false
74+
* rdma_poller_num: 轮询模式下的poller数目，默认1
75+
* rdma_poller_yield: 轮询模式下的poller是否主动放弃CPU，默认是false
76+
* rdma_enable_edisp_schedule: 让事件驱动器可以被调度，默认是true

‎src/brpc/rdma/rdma_endpoint.cpp‎

Lines changed: 164 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,11 @@ DEFINE_int32(rdma_prepared_qp_size, 128, "SQ and RQ size for prepared QP.");
5858
DEFINE_int32(rdma_prepared_qp_cnt, 1024, "Initial count of prepared QP.");
5959
DEFINE_bool(rdma_trace_verbose, false, "Print log message verbosely");
6060
BRPC_VALIDATE_GFLAG(rdma_trace_verbose, brpc::PassValidate);
61+
DEFINE_bool(rdma_use_polling, false, "Use polling mode for RDMA.");
62+
DEFINE_int32(rdma_poller_num, 1, "Poller number in RDMA polling mode.");
63+
DEFINE_bool(rdma_poller_yield, false, "Yield thread in RDMA polling mode.");
64+
DEFINE_bool(rdma_enable_edisp_schedule, true,
65+
"Enable event dispatcher schedule in RDMA");
6166

6267
static const size_t IOBUF_BLOCK_HEADER_LEN = 32; // implementation-dependent
6368

@@ -699,7 +704,6 @@ void* RdmaEndpoint::ProcessHandshakeAtServer(void* arg) {
699704
LOG_IF(INFO, FLAGS_rdma_trace_verbose)
700705
<< "Handshake ends (use tcp) on " << s->description();
701706
}
702-
703707
ep->TryReadOnTcp();
704708

705709
return NULL;
@@ -1041,26 +1045,36 @@ static RdmaResource* AllocateQpCq(uint16_t sq_size, uint16_t rq_size) {
10411045
return NULL;
10421046
}
10431047

1044-
res->comp_channel = IbvCreateCompChannel(GetRdmaContext());
1045-
if (!res->comp_channel) {
1046-
PLOG(WARNING) << "Fail to create comp channel for CQ";
1047-
delete res;
1048-
return NULL;
1049-
}
1048+
if (!FLAGS_rdma_use_polling) {
1049+
res->comp_channel = IbvCreateCompChannel(GetRdmaContext());
1050+
if (!res->comp_channel) {
1051+
PLOG(WARNING) << "Fail to create comp channel for CQ";
1052+
delete res;
1053+
return NULL;
1054+
}
10501055

1051-
butil::make_close_on_exec(res->comp_channel->fd);
1052-
if (butil::make_non_blocking(res->comp_channel->fd) < 0) {
1053-
PLOG(WARNING) << "Fail to set comp channel nonblocking";
1054-
delete res;
1055-
return NULL;
1056-
}
1056+
butil::make_close_on_exec(res->comp_channel->fd);
1057+
if (butil::make_non_blocking(res->comp_channel->fd) < 0) {
1058+
PLOG(WARNING) << "Fail to set comp channel nonblocking";
1059+
delete res;
1060+
return NULL;
1061+
}
10571062

1058-
res->cq = IbvCreateCq(GetRdmaContext(), 2 * FLAGS_rdma_prepared_qp_size,
1059-
NULL, res->comp_channel, GetRdmaCompVector());
1060-
if (!res->cq) {
1061-
PLOG(WARNING) << "Fail to create CQ";
1062-
delete res;
1063-
return NULL;
1063+
res->cq = IbvCreateCq(GetRdmaContext(), 2 * FLAGS_rdma_prepared_qp_size,
1064+
NULL, res->comp_channel, GetRdmaCompVector());
1065+
if (!res->cq) {
1066+
PLOG(WARNING) << "Fail to create CQ";
1067+
delete res;
1068+
return NULL;
1069+
}
1070+
} else {
1071+
res->cq = IbvCreateCq(GetRdmaContext(), 2 * FLAGS_rdma_prepared_qp_size,
1072+
NULL, NULL, 0);
1073+
if (!res->cq) {
1074+
PLOG(WARNING) << "Fail to create CQ";
1075+
delete res;
1076+
return NULL;
1077+
}
10641078
}
10651079

10661080
ibv_qp_init_attr attr;
@@ -1117,19 +1131,30 @@ int RdmaEndpoint::AllocateResources() {
11171131
return -1;
11181132
}
11191133

1120-
SocketOptions options;
1121-
options.user = this;
1122-
options.keytable_pool = _socket->_keytable_pool;
1123-
options.fd = _resource->comp_channel->fd;
1124-
options.on_edge_triggered_events = PollCq;
1125-
if (Socket::Create(options, &_cq_sid) < 0) {
1126-
PLOG(WARNING) << "Fail to create socket for cq";
1127-
return -1;
1128-
}
1134+
if (!FLAGS_rdma_use_polling) {
1135+
SocketOptions options;
1136+
options.user = this;
1137+
options.keytable_pool = _socket->_keytable_pool;
1138+
options.fd = _resource->comp_channel->fd;
1139+
options.on_edge_triggered_events = PollCq;
1140+
if (Socket::Create(options, &_cq_sid) < 0) {
1141+
PLOG(WARNING) << "Fail to create socket for cq";
1142+
return -1;
1143+
}
11291144

1130-
if (ibv_req_notify_cq(_resource->cq, 1) < 0) {
1131-
PLOG(WARNING) << "Fail to arm CQ comp channel";
1132-
return -1;
1145+
if (ibv_req_notify_cq(_resource->cq, 1) < 0) {
1146+
PLOG(WARNING) << "Fail to arm CQ comp channel";
1147+
return -1;
1148+
}
1149+
} else {
1150+
SocketOptions options;
1151+
options.user = this;
1152+
options.keytable_pool = _socket->_keytable_pool;
1153+
if (Socket::Create(options, &_cq_sid) < 0) {
1154+
PLOG(WARNING) << "Fail to create socket for cq";
1155+
return -1;
1156+
}
1157+
PollerAddCqSocket();
11331158
}
11341159

11351160
_sbuf.resize(_sq_size - RESERVED_WR_NUM);
@@ -1227,6 +1252,9 @@ void RdmaEndpoint::DeallocateResources() {
12271252
if (!_resource) {
12281253
return;
12291254
}
1255+
if (FLAGS_rdma_use_polling) {
1256+
PollerRemoveCqSocket();
1257+
}
12301258
bool move_to_rdma_resource_list = false;
12311259
if (_sq_size <= FLAGS_rdma_prepared_qp_size &&
12321260
_rq_size <= FLAGS_rdma_prepared_qp_size &&
@@ -1237,7 +1265,10 @@ void RdmaEndpoint::DeallocateResources() {
12371265
move_to_rdma_resource_list = true;
12381266
}
12391267
}
1240-
int fd = _resource->comp_channel->fd;
1268+
int fd = -1;
1269+
if (_resource->comp_channel) {
1270+
fd = _resource->comp_channel->fd;
1271+
}
12411272
if (!move_to_rdma_resource_list) {
12421273
if (_resource->qp) {
12431274
if (IbvDestroyQp(_resource->qp) < 0) {
@@ -1327,12 +1358,14 @@ void RdmaEndpoint::PollCq(Socket* m) {
13271358
}
13281359
CHECK(ep == s->_rdma_ep);
13291360

1330-
if (ep->GetAndAckEvents() < 0) {
1331-
const int saved_errno = errno;
1332-
PLOG(ERROR) << "Fail to get cq event: " << s->description();
1333-
s->SetFailed(saved_errno, "Fail to get cq event from %s: %s",
1334-
s->description().c_str(), berror(saved_errno));
1335-
return;
1361+
if (!FLAGS_rdma_use_polling) {
1362+
if (ep->GetAndAckEvents() < 0) {
1363+
const int saved_errno = errno;
1364+
PLOG(ERROR) << "Fail to get cq event: " << s->description();
1365+
s->SetFailed(saved_errno, "Fail to get cq event from %s: %s",
1366+
s->description().c_str(), berror(saved_errno));
1367+
return;
1368+
}
13361369
}
13371370

13381371
int progress = Socket::PROGRESS_INIT;
@@ -1349,6 +1382,9 @@ void RdmaEndpoint::PollCq(Socket* m) {
13491382
return;
13501383
}
13511384
if (cnt == 0) {
1385+
if (FLAGS_rdma_use_polling) {
1386+
return;
1387+
}
13521388
if (!notified) {
13531389
// Since RDMA only provides one shot event, we have to call the
13541390
// notify function every time. Because there is a possibility
@@ -1370,7 +1406,7 @@ void RdmaEndpoint::PollCq(Socket* m) {
13701406
}
13711407
if (ep->GetAndAckEvents() < 0) {
13721408
s->SetFailed(errno, "Fail to ack CQ event on %s",
1373-
s->description().c_str());
1409+
s->description().c_str());
13741410
return;
13751411
}
13761412
notified = false;
@@ -1474,6 +1510,13 @@ int RdmaEndpoint::GlobalInitialize() {
14741510
g_rdma_resource_list = res;
14751511
}
14761512

1513+
if (FLAGS_rdma_use_polling) {
1514+
auto rc = PollingModeInitialize();
1515+
if (rc != 0) {
1516+
return -1;
1517+
}
1518+
}
1519+
14771520
return 0;
14781521
}
14791522

@@ -1486,6 +1529,88 @@ void RdmaEndpoint::GlobalRelease() {
14861529
delete res;
14871530
}
14881531
}
1532+
1533+
if (FLAGS_rdma_use_polling) {
1534+
PollingModeRelease();
1535+
}
1536+
}
1537+
1538+
std::vector<RdmaEndpoint::Poller> RdmaEndpoint::_pollers;
1539+
std::atomic<bool> RdmaEndpoint::_running(false);
1540+
1541+
void RdmaEndpoint::SetCallbackFn(std::function<void()> cb) {
1542+
for (int i = 0; i < FLAGS_rdma_poller_num; ++i) {
1543+
auto& poller = _pollers[i];
1544+
std::unique_lock<bthread::Mutex> lk(poller.callback_mutex);
1545+
poller.callback = cb;
1546+
}
1547+
}
1548+
1549+
int RdmaEndpoint::PollingModeInitialize() {
1550+
auto fn = [](void* args) -> void* {
1551+
auto poller = static_cast<Poller*>(args);
1552+
while (_running.load(butil::memory_order_relaxed)) {
1553+
std::list<Socket*> sockets;
1554+
{
1555+
std::unique_lock<bthread::Mutex> lk(poller->sockets_mutex);
1556+
sockets = poller->sockets; // copy all sockets is not good
1557+
}
1558+
for (auto m : sockets) {
1559+
PollCq(m);
1560+
}
1561+
{
1562+
std::unique_lock<bthread::Mutex> lk(poller->callback_mutex);
1563+
if (poller->callback) {
1564+
poller->callback();
1565+
}
1566+
}
1567+
if (FLAGS_rdma_poller_yield) {
1568+
bthread_yield();
1569+
}
1570+
}
1571+
return nullptr;
1572+
};
1573+
_pollers = std::vector<Poller>(FLAGS_rdma_poller_num);
1574+
_running.store(true, butil::memory_order_relaxed);
1575+
for (int i = 0; i < FLAGS_rdma_poller_num; ++i) {
1576+
auto rc = bthread_start_background(
1577+
&_pollers[i].tid, &BTHREAD_ATTR_NORMAL, fn, &_pollers[i]);
1578+
if (rc != 0) {
1579+
LOG(ERROR) << "Fail to start rdma polling bthread";
1580+
return -1;
1581+
}
1582+
}
1583+
return 0;
1584+
}
1585+
1586+
void RdmaEndpoint::PollingModeRelease() {
1587+
_running.store(false, butil::memory_order_relaxed);
1588+
for (int i = 0; i < FLAGS_rdma_poller_num; ++i) {
1589+
bthread_join(_pollers[i].tid, nullptr);
1590+
}
1591+
}
1592+
1593+
// Add socket to poller
1594+
void RdmaEndpoint::PollerAddCqSocket() {
1595+
SocketUniquePtr s;
1596+
if (Socket::Address(_cq_sid, &s) != 0) {
1597+
LOG(ERROR) << "Fail to address socket id: " << _cq_sid;
1598+
return;
1599+
}
1600+
auto index = butil::fmix32(_cq_sid) % FLAGS_rdma_poller_num;
1601+
auto& poller = _pollers[index];
1602+
std::unique_lock<bthread::Mutex> lk(poller.sockets_mutex);
1603+
_cq_iter = poller.sockets.insert(poller.sockets.end(), s.get());
1604+
}
1605+
1606+
// Remove socket from poller
1607+
void RdmaEndpoint::PollerRemoveCqSocket() {
1608+
auto index = butil::fmix32(_cq_sid) % FLAGS_rdma_poller_num;
1609+
auto& poller = _pollers[index];
1610+
std::unique_lock<bthread::Mutex> lk(poller.sockets_mutex);
1611+
if (_cq_iter != poller.sockets.end()) {
1612+
poller.sockets.erase(_cq_iter);
1613+
}
14891614
}
14901615

14911616
} // namespace rdma

‎src/brpc/rdma/rdma_endpoint.h‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@
2323
#include <iostream>
2424
#include <string>
2525
#include <vector>
26+
#include <list>
27+
#include <functional>
2628
#include <infiniband/verbs.h>
2729
#include "butil/atomicops.h"
2830
#include "butil/iobuf.h"
@@ -34,6 +36,8 @@ namespace brpc {
3436
class Socket;
3537
namespace rdma {
3638

39+
DECLARE_bool(rdma_enable_edisp_schedule);
40+
3741
class RdmaConnect : public AppConnect {
3842
public:
3943
void StartConnect(const Socket* socket,
@@ -90,6 +94,9 @@ friend class brpc::Socket;
9094
// Callback when there is new epollin event on TCP fd
9195
static void OnNewDataFromTcp(Socket* m);
9296

97+
// Set callback function
98+
static void SetCallbackFn(std::function<void()> cb);
99+
93100
private:
94101
enum State {
95102
UNINIT = 0x0,
@@ -191,6 +198,17 @@ friend class brpc::Socket;
191198
// Try to read data on TCP fd in _socket
192199
inline void TryReadOnTcp();
193200

201+
// Initialize polling mode
202+
static int PollingModeInitialize();
203+
204+
static void PollingModeRelease();
205+
206+
// Add cq socket to poller
207+
void PollerAddCqSocket();
208+
209+
// Remove cq socket to poller
210+
void PollerRemoveCqSocket();
211+
194212
// Not owner
195213
Socket* _socket;
196214

@@ -206,6 +224,9 @@ friend class brpc::Socket;
206224
// the SocketId which wrap the comp channel of CQ
207225
SocketId _cq_sid;
208226

227+
// the Socket pointer's iterator in list
228+
std::list<Socket*>::iterator _cq_iter;
229+
209230
// Capacity of local Send Queue and local Recv Queue
210231
uint16_t _sq_size;
211232
uint16_t _rq_size;
@@ -245,6 +266,18 @@ friend class brpc::Socket;
245266
butil::atomic<int> *_read_butex;
246267

247268
DISALLOW_COPY_AND_ASSIGN(RdmaEndpoint);
269+
270+
struct Poller {
271+
bthread_t tid;
272+
// TODO(yangliming): lockless queue is better
273+
std::list<Socket*> sockets;
274+
bthread::Mutex sockets_mutex;
275+
// Callback used for io_uring/spdk async engine
276+
std::function<void()> callback;
277+
bthread::Mutex callback_mutex;
278+
};
279+
static std::vector<Poller> _pollers;
280+
static std::atomic<bool> _running;
248281
};
249282

250283
} // namespace rdma

‎src/brpc/socket.cpp‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2273,6 +2273,14 @@ int Socket::OnInputEvent(void* user_data, uint32_t events,
22732273
attr.tag = bthread_self_tag();
22742274
if (FLAGS_usercode_in_coroutine) {
22752275
ProcessEvent(p);
2276+
#if BRPC_WITH_RDMA
2277+
} else if (!rdma::FLAGS_rdma_enable_edisp_schedule) {
2278+
auto rc = bthread_start_background(&tid, &attr, ProcessEvent, p);
2279+
if (rc != 0) {
2280+
LOG(FATAL) << "Fail to start ProcessEvent";
2281+
ProcessEvent(p);
2282+
}
2283+
#endif
22762284
} else if (bthread_start_urgent(&tid, &attr, ProcessEvent, p) != 0) {
22772285
LOG(FATAL) << "Fail to start ProcessEvent";
22782286
ProcessEvent(p);

‎src/brpc/span.cpp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818

1919
#include <netinet/in.h>
20+
#include <functional>
2021
#include <gflags/gflags.h>
2122
#include <leveldb/db.h>
2223
#include <leveldb/comparator.h>

0 commit comments

Comments
 (0)