diff --git a/src/brpc/rdma/block_pool.cpp b/src/brpc/rdma/block_pool.cpp index d8dbb8abda..3a332a918e 100644 --- a/src/brpc/rdma/block_pool.cpp +++ b/src/brpc/rdma/block_pool.cpp @@ -44,7 +44,7 @@ DEFINE_bool(rdma_memory_pool_user_specified_memory, false, DEFINE_string(rdma_recv_block_type, "default", "Default size type for recv WR: " "default(8KB - 32B)/large(64KB - 32B)/huge(2MB - 32B)"); -static RegisterCallback g_cb = NULL; +static RegisterCallback g_cb = nullptr; // Number of bytes in 1MB static const size_t BYTES_IN_MB = 1048576; @@ -82,13 +82,13 @@ static const int32_t RDMA_MEMORY_POOL_MAX_BUCKETS = 16; static size_t g_buckets = 1; static bool g_dump_enable = false; -static butil::Mutex* g_dump_mutex = NULL; +static butil::Mutex* g_dump_mutex = nullptr; // Only for default block size -static __thread IdleNode* tls_idle_list = NULL; +static __thread IdleNode* tls_idle_list = nullptr; static __thread size_t tls_idle_num = 0; static __thread bool tls_inited = false; -static butil::Mutex* g_tls_info_mutex = NULL; +static butil::Mutex* g_tls_info_mutex = nullptr; static size_t g_tls_info_cnt = 0; static size_t* g_tls_info[1024]; @@ -102,14 +102,14 @@ struct GlobalInfo { std::vector expansion_list[BLOCK_SIZE_COUNT]; std::vector expansion_size[BLOCK_SIZE_COUNT]; }; -static GlobalInfo* g_info = NULL; +static GlobalInfo* g_info = nullptr; static inline Region* GetRegion(const void* buf) { if (!buf) { errno = EINVAL; - return NULL; + return nullptr; } - Region* r = NULL; + Region* r = nullptr; uintptr_t addr = (uintptr_t)buf; for (int i = 0; i < FLAGS_rdma_memory_pool_max_regions; ++i) { if (g_regions[i].start == 0) { @@ -140,13 +140,13 @@ static void* ExtendBlockPoolImpl(void* region_base, size_t region_size, int bloc if (g_region_num == FLAGS_rdma_memory_pool_max_regions) { LOG_EVERY_SECOND(ERROR) << "Memory pool reaches max regions"; errno = ENOMEM; - return NULL; + return nullptr; } uint32_t id = g_cb(region_base, region_size); if (id == 0) { errno = EINVAL; - return NULL; + return nullptr; } IdleNode* node[g_buckets]; @@ -158,7 +158,7 @@ static void* ExtendBlockPoolImpl(void* region_base, size_t region_size, int bloc butil::return_object(node[j]); } errno = ENOMEM; - return NULL; + return nullptr; } } @@ -187,14 +187,14 @@ static void* ExtendBlockPoolImpl(void* region_base, size_t region_size, int bloc static void* ExtendBlockPool(size_t region_size, int block_type) { if (region_size < 1 || block_type < 0) { errno = EINVAL; - return NULL; + return nullptr; } if (FLAGS_rdma_memory_pool_user_specified_memory) { LOG_EVERY_SECOND(ERROR) << "Fail to extend new region, " "rdma_memory_pool_user_specified_memory is " "true, ExtendBlockPool is disabled"; - return NULL; + return nullptr; } // Regularize region size @@ -203,10 +203,10 @@ static void* ExtendBlockPool(size_t region_size, int block_type) { LOG(INFO) << "Start extend rdma memory " << region_size / BYTES_IN_MB << "MB"; - void* region_base = NULL; + void* region_base = nullptr; if (posix_memalign(®ion_base, 4096, region_size) != 0) { PLOG_EVERY_SECOND(ERROR) << "Memory not enough"; - return NULL; + return nullptr; } return ExtendBlockPoolImpl(region_base, region_size, block_type); @@ -219,12 +219,12 @@ void* ExtendBlockPoolByUser(void* region_base, size_t region_size, int block_typ if (!FLAGS_rdma_memory_pool_user_specified_memory) { LOG_EVERY_SECOND(ERROR) << "User extend memory is disabled"; - return NULL; + return nullptr; } if (reinterpret_cast(region_base) % 4096 != 0) { LOG_EVERY_SECOND(ERROR) << "region_base must be 4096 aligned"; errno = EINVAL; - return NULL; + return nullptr; } region_size = @@ -284,17 +284,14 @@ bool InitBlockPool(RegisterCallback cb) { return false; } g_buckets = FLAGS_rdma_memory_pool_buckets; - g_info = new (std::nothrow) GlobalInfo; - if (!g_info) { - return false; - } + g_info = new GlobalInfo; for (int i = 0; i < BLOCK_SIZE_COUNT; ++i) { - g_info->idle_list[i].resize(g_buckets, NULL); + g_info->idle_list[i].resize(g_buckets, nullptr); if (g_info->idle_list[i].size() != g_buckets) { return false; } - g_info->lock[i].resize(g_buckets, NULL); + g_info->lock[i].resize(g_buckets, nullptr); if (g_info->lock[i].size() != g_buckets) { return false; } @@ -304,12 +301,9 @@ bool InitBlockPool(RegisterCallback cb) { } g_info->region_num[i] = 0; for (size_t j = 0; j < g_buckets; ++j) { - g_info->lock[i][j] = new (std::nothrow) butil::Mutex; - if (!g_info->lock[i][j]) { - return false; - } + g_info->lock[i][j] = new butil::Mutex; } - g_info->expansion_list[i].resize(g_buckets, NULL); + g_info->expansion_list[i].resize(g_buckets, nullptr); if (g_info->expansion_list[i].size() != g_buckets) { return false; } @@ -327,18 +321,18 @@ bool InitBlockPool(RegisterCallback cb) { } if (ExtendBlockPool(FLAGS_rdma_memory_pool_initial_size_mb, - GetRdmaBlockType()) != NULL) { + GetRdmaBlockType()) != nullptr) { return true; } return false; } static void MoveExpansionList2EmptyIdleList(int block_type, size_t index) { - CHECK(NULL == g_info->idle_list[block_type][index]); + CHECK(nullptr == g_info->idle_list[block_type][index]); g_info->idle_list[block_type][index] = g_info->expansion_list[block_type][index]; g_info->idle_size[block_type][index] += g_info->expansion_size[block_type][index]; - g_info->expansion_list[block_type][index] = NULL; + g_info->expansion_list[block_type][index] = nullptr; g_info->expansion_size[block_type][index] = 0; } @@ -354,8 +348,8 @@ static void* AllocBlockFrom(int block_type) { } }; - void* ptr = NULL; - if (0 == block_type && NULL != tls_idle_list) { + void* ptr = nullptr; + if (0 == block_type && nullptr != tls_idle_list) { CHECK(tls_idle_num > 0); IdleNode* n = tls_idle_list; tls_idle_list = n->next; @@ -368,14 +362,14 @@ static void* AllocBlockFrom(int block_type) { size_t index = butil::fast_rand() % g_buckets; BAIDU_SCOPED_LOCK(*g_info->lock[block_type][index]); IdleNode* node = g_info->idle_list[block_type][index]; - if (NULL == node) { + if (nullptr == node) { BAIDU_SCOPED_LOCK(g_info->extend_lock); node = g_info->idle_list[block_type][index]; - if (NULL == node && NULL != g_info->expansion_list[block_type][index]) { + if (nullptr == node && nullptr != g_info->expansion_list[block_type][index]) { MoveExpansionList2EmptyIdleList(block_type, index); node = g_info->idle_list[block_type][index]; } - if (NULL == node) { + if (nullptr == node) { // There is no block left, extend a new region. if (!ExtendBlockPool(FLAGS_rdma_memory_pool_increase_size_mb, block_type)) { LOG_EVERY_SECOND(ERROR) << "Fail to extend new region. " @@ -384,13 +378,13 @@ static void* AllocBlockFrom(int block_type) { << "rdma_memory_pool_initial_size_mb, " << "rdma_memory_pool_increase_size_mb, " << "rdma_memory_pool_max_regions."; - return NULL; + return nullptr; } MoveExpansionList2EmptyIdleList(block_type, index); node = g_info->idle_list[block_type][index]; } } - CHECK(NULL != node); + CHECK(nullptr != node); ptr = node->start; if (node->len > g_block_size[block_type]) { @@ -406,7 +400,7 @@ static void* AllocBlockFrom(int block_type) { if (block_type == 0) { node = g_info->idle_list[0][index]; tls_idle_list = node; - IdleNode* last_node = NULL; + IdleNode* last_node = nullptr; while (node) { if (tls_idle_num > (uint32_t)FLAGS_rdma_memory_pool_tls_cache_num / 2 || node->len > g_block_size[0]) { @@ -417,12 +411,12 @@ static void* AllocBlockFrom(int block_type) { node = node->next; } if (tls_idle_num == 0) { - tls_idle_list = NULL; + tls_idle_list = nullptr; } else { g_info->idle_list[0][index] = node; } if (last_node) { - last_node->next = NULL; + last_node->next = nullptr; } } @@ -432,14 +426,14 @@ static void* AllocBlockFrom(int block_type) { void* AllocBlock(size_t size) { if (size == 0 || size > g_block_size[BLOCK_SIZE_COUNT - 1]) { errno = EINVAL; - return NULL; + return nullptr; } for (int i = 0; i < BLOCK_SIZE_COUNT; ++i) { if (size <= g_block_size[i]) { return AllocBlockFrom(i);; } } - return NULL; + return nullptr; } void RecycleAll() { @@ -515,7 +509,7 @@ int DeallocBlock(void* buf) { // Recycle half the cached blocks in tls for default block size int num = FLAGS_rdma_memory_pool_tls_cache_num / 2; IdleNode* new_head = tls_idle_list; - IdleNode* recycle_tail = NULL; + IdleNode* recycle_tail = nullptr; for (int i = 0; i < num; ++i) { recycle_tail = new_head; len += recycle_tail->len; @@ -614,18 +608,18 @@ void DestroyBlockPool() { butil::return_object(node); node = tmp; } - g_info->idle_list[i][j] = NULL; + g_info->idle_list[i][j] = nullptr; // Release the per-bucket mutexes allocated in InitBlockPool. delete g_info->lock[i][j]; - g_info->lock[i][j] = NULL; + g_info->lock[i][j] = nullptr; } } delete g_info; - g_info = NULL; + g_info = nullptr; delete g_dump_mutex; - g_dump_mutex = NULL; + g_dump_mutex = nullptr; delete g_tls_info_mutex; - g_tls_info_mutex = NULL; + g_tls_info_mutex = nullptr; for (int i = 0; i < g_region_num; ++i) { if (g_regions[i].start == 0) { break; @@ -634,7 +628,7 @@ void DestroyBlockPool() { g_regions[i].start = 0; } g_region_num = 0; - g_cb = NULL; + g_cb = nullptr; } // Just for UT diff --git a/src/brpc/rdma/block_pool.h b/src/brpc/rdma/block_pool.h index c9589fb035..b0b8ffc1e4 100644 --- a/src/brpc/rdma/block_pool.h +++ b/src/brpc/rdma/block_pool.h @@ -72,7 +72,7 @@ typedef uint32_t (*RegisterCallback)(void*, size_t); // The argument is a callback called when the pool is enlarged with a new // region. It should be the memory registration in brpc. However, // in block_pool, we just abstract it into a function to get region id. -// Return the first region's address, NULL if failed and errno is set. +// Return the first region's address, nullptr if failed and errno is set. bool InitBlockPool(RegisterCallback cb); // In scenarios where users need to manually specify memory regions (e.g., using @@ -83,10 +83,10 @@ bool InitBlockPool(RegisterCallback cb); void* ExtendBlockPoolByUser(void* region_base, size_t region_size, int block_type); // Allocate a buf with length at least @a size (require: size>0) -// Return the address allocated, NULL if failed and errno is set. +// Return the address allocated, nullptr if failed and errno is set. void* AllocBlock(size_t size); -// Deallocate the buf (require: buf!=NULL) +// Deallocate the buf (require: buf!=nullptr) // Return 0 if success, -1 if failed and errno is set. // If the given buf is not in any region, the errno is ERANGE. int DeallocBlock(void* buf); diff --git a/src/brpc/rdma/rdma_endpoint.cpp b/src/brpc/rdma/rdma_endpoint.cpp index e2ce2e0c1b..fd70fb5c2b 100644 --- a/src/brpc/rdma/rdma_endpoint.cpp +++ b/src/brpc/rdma/rdma_endpoint.cpp @@ -88,23 +88,23 @@ extern const uint16_t MIN_QP_SIZE = 16; static const uint16_t MAX_QP_SIZE = 4096; extern const uint16_t MIN_BLOCK_SIZE = 1024; -static butil::Mutex* g_rdma_resource_mutex = NULL; -static RdmaResource* g_rdma_resource_list = NULL; +static butil::Mutex* g_rdma_resource_mutex = nullptr; +static RdmaResource* g_rdma_resource_list = nullptr; RdmaResource::~RdmaResource() { - if (NULL != qp) { + if (nullptr != qp) { IbvDestroyQp(qp); } - if (NULL != polling_cq) { + if (nullptr != polling_cq) { IbvDestroyCq(polling_cq); } - if (NULL != send_cq) { + if (nullptr != send_cq) { IbvDestroyCq(send_cq); } - if (NULL != recv_cq) { + if (nullptr != recv_cq) { IbvDestroyCq(recv_cq); } - if (NULL != comp_channel) { + if (nullptr != comp_channel) { IbvDestroyCompChannel(comp_channel); } } @@ -113,7 +113,7 @@ RdmaEndpoint::RdmaEndpoint(Socket* s) : _socket(s) , _state(UNINIT) , _handshake_version(0) - , _resource(NULL) + , _resource(nullptr) , _send_cq_events(0) , _recv_cq_events(0) , _cq_sid(INVALID_SOCKET_ID) @@ -160,7 +160,7 @@ void RdmaEndpoint::Reset() { _state.store(UNINIT, butil::memory_order_relaxed); _handshake_version = 0; _outgoing_ece.reset(); - _resource = NULL; + _resource = nullptr; _send_cq_events = 0; _recv_cq_events = 0; _cq_sid = INVALID_SOCKET_ID; @@ -187,7 +187,7 @@ void RdmaConnect::StartConnect(const Socket* socket, void (*done)(int err, void* data), void* data) { auto* rdma_transport = static_cast(socket->_transport.get()); - CHECK(rdma_transport->_rdma_ep != NULL); + CHECK(rdma_transport->_rdma_ep != nullptr); SocketUniquePtr s; if (Socket::Address(socket->id(), &s) != 0) { return; @@ -223,7 +223,7 @@ void RdmaConnect::Run() { void RdmaEndpoint::OnNewDataFromTcp(Socket* m) { auto* rdma_transport = static_cast(m->_transport.get()); RdmaEndpoint* ep = rdma_transport->GetRdmaEp(); - CHECK(ep != NULL); + CHECK(ep != nullptr); int progress = Socket::PROGRESS_INIT; while (true) { @@ -313,7 +313,7 @@ static int ReadFromFdLoop(butil::atomic* read_butex, } int RdmaEndpoint::ReadFromFd(void* data, size_t len) { - CHECK(data != NULL); + CHECK(data != nullptr); const int fd = _socket->fd(); return ReadFromFdLoop(_read_butex, len, [data, fd](size_t offset, size_t remaining) { @@ -322,7 +322,7 @@ int RdmaEndpoint::ReadFromFd(void* data, size_t len) { } int RdmaEndpoint::ReadFromFd(butil::IOPortal* data, size_t len) { - CHECK(data != NULL); + CHECK(data != nullptr); const int fd = _socket->fd(); return ReadFromFdLoop(_read_butex, len, [data, fd](size_t /*offset*/, size_t remaining) { @@ -366,7 +366,7 @@ static int WriteToFdLoop(size_t len, WriteOnce&& write_once, WaitWritable&& wait } int RdmaEndpoint::WriteToFd(void* data, size_t len) { - CHECK(data != NULL); + CHECK(data != nullptr); Socket* s = _socket; const int fd = s->fd(); return WriteToFdLoop(len, @@ -379,7 +379,7 @@ int RdmaEndpoint::WriteToFd(void* data, size_t len) { } int RdmaEndpoint::WriteToFd(butil::IOBuf* data) { - CHECK(data != NULL); + CHECK(data != nullptr); Socket* s = _socket; const int fd = s->fd(); return WriteToFdLoop(data->size(), @@ -428,7 +428,7 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg) { << "Start handshake on " << s->description(); std::unique_ptr handshake = CreateClientHandshake(ep); - CHECK(handshake != NULL); + CHECK(handshake != nullptr); ep->_handshake_version = handshake->ProtocolVersion(); // First initialize CQ and QP resources. @@ -439,7 +439,7 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg) { errno = 0; rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF; ep->_state.store(FALLBACK_TCP, butil::memory_order_release); - return NULL; + return nullptr; } // Send hello message to server @@ -451,7 +451,7 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg) { s->SetFailed(saved_errno, "Fail to complete rdma handshake from %s: %s", s->description().c_str(), berror(saved_errno)); ep->_state.store(FAILED, butil::memory_order_relaxed); - return NULL; + return nullptr; } // Receive and parse remote hello. @@ -465,7 +465,7 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg) { s->SetFailed(saved_errno, "Fail to complete rdma handshake from %s: %s", s->description().c_str(), berror(saved_errno)); ep->_state.store(FAILED, butil::memory_order_relaxed); - return NULL; + return nullptr; } if (r != RemoteHelloResult::NEGOTIATED) { @@ -496,7 +496,7 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg) { s->SetFailed(saved_errno, "Fail to complete rdma handshake from %s: %s", s->description().c_str(), berror(saved_errno)); ep->_state.store(FAILED, butil::memory_order_relaxed); - return NULL; + return nullptr; } if (rdma_transport->_rdma_state == RdmaTransport::RDMA_ON) { @@ -512,7 +512,7 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg) { errno = 0; - return NULL; + return nullptr; } // Server-side handshake entry: the state machine. @@ -533,9 +533,9 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg) { ParseResult RdmaEndpoint::ExecuteServerHandshake(butil::IOBuf* source, Socket* s) { RdmaTransport* rdma_transport = static_cast(s->_transport.get()); RdmaEndpoint* ep = rdma_transport->_rdma_ep; - CHECK(ep != NULL); + CHECK(ep != nullptr); - if (s->parsing_context() == NULL) { + if (s->parsing_context() == nullptr) { // Phase 1: read the client hello, negotiate, reply server hello. if (source->size() < HELLO_MAGIC_LEN) { return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); @@ -547,7 +547,7 @@ ParseResult RdmaEndpoint::ExecuteServerHandshake(butil::IOBuf* source, Socket* s // magic is NOT consumed; ReceiveAndParseRemoteHello() reads it again // from `source`). std::unique_ptr hs = CreateServerHandshakeByMagic(ep, source, magic); - if (hs == NULL) { + if (hs == nullptr) { return MakeParseError(PARSE_ERROR_TRY_OTHERS); } ep->_handshake_version = hs->ProtocolVersion(); @@ -611,7 +611,7 @@ ParseResult RdmaEndpoint::ExecuteServerHandshake(butil::IOBuf* source, Socket* s LOG(WARNING) << "Too many bytes in handshake ACK, drop connection: " << s->description(); ep->_state.store(FAILED, butil::memory_order_relaxed); - s->reset_parsing_context(NULL); + s->reset_parsing_context(nullptr); return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG); } @@ -624,7 +624,7 @@ ParseResult RdmaEndpoint::ExecuteServerHandshake(butil::IOBuf* source, Socket* s << "Server handshake ends (use tcp) on " << s->description(); rdma_transport->_rdma_state = RdmaTransport::RDMA_OFF; ep->_state.store(FALLBACK_TCP, butil::memory_order_release); - s->reset_parsing_context(NULL); + s->reset_parsing_context(nullptr); return MakeParseError(PARSE_ERROR_TRY_OTHERS); } @@ -632,7 +632,7 @@ ParseResult RdmaEndpoint::ExecuteServerHandshake(butil::IOBuf* source, Socket* s LOG(WARNING) << "Client wants RDMA in ACK but server fell back: " << s->description(); ep->_state.store(FAILED, butil::memory_order_relaxed); - s->reset_parsing_context(NULL); + s->reset_parsing_context(nullptr); return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG); } @@ -641,7 +641,7 @@ ParseResult RdmaEndpoint::ExecuteServerHandshake(butil::IOBuf* source, Socket* s << ") on " << s->description(); rdma_transport->_rdma_state = RdmaTransport::RDMA_ON; ep->_state.store(ESTABLISHED, butil::memory_order_relaxed); - s->reset_parsing_context(NULL); + s->reset_parsing_context(nullptr); return MakeParseError(PARSE_ERROR_TRY_OTHERS); } @@ -718,7 +718,7 @@ ssize_t RdmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) { return -1; } - CHECK(from != NULL); + CHECK(from != nullptr); CHECK(ndata > 0); size_t total_len = 0; @@ -817,7 +817,7 @@ ssize_t RdmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) { _sq_unsignaled = 0; } - ibv_send_wr* bad = NULL; + ibv_send_wr* bad = nullptr; int err = ibv_post_send(_resource->qp, &wr, &bad); if (err != 0) { // We use other way to guarantee the Send Queue is not full. @@ -867,7 +867,7 @@ int RdmaEndpoint::SendImm(uint32_t imm) { wr.send_flags |= IBV_SEND_SOLICITED | IBV_SEND_SIGNALED; wr.wr_id = 0; - ibv_send_wr* bad = NULL; + ibv_send_wr* bad = nullptr; int err = ibv_post_send(_resource->qp, &wr, &bad); if (err != 0) { std::ostringstream oss; @@ -970,7 +970,7 @@ int RdmaEndpoint::DoPostRecv(void* block, size_t block_size) { wr.num_sge = 1; wr.sg_list = &sge; - ibv_recv_wr* bad = NULL; + ibv_recv_wr* bad = nullptr; int err = ibv_post_recv(_resource->qp, &wr, &bad); if (err != 0) { LOG(WARNING) << "Fail to ibv_post_recv: " << berror(err); @@ -1025,52 +1025,52 @@ static RdmaResource* AllocateQpCq(uint16_t sq_size, uint16_t rq_size) { std::unique_ptr resource(new RdmaResource); if (!FLAGS_rdma_use_polling) { resource->comp_channel = IbvCreateCompChannel(GetRdmaContext()); - if (NULL == resource->comp_channel) { + if (nullptr == resource->comp_channel) { PLOG(WARNING) << "Fail to create comp channel for CQ"; - return NULL; + return nullptr; } if (butil::make_close_on_exec(resource->comp_channel->fd) < 0) { PLOG(WARNING) << "Fail to set comp channel close-on-exec"; - return NULL; + return nullptr; } if (butil::make_non_blocking(resource->comp_channel->fd) < 0) { PLOG(WARNING) << "Fail to set comp channel nonblocking"; - return NULL; + return nullptr; } resource->send_cq = IbvCreateCq(GetRdmaContext(), FLAGS_rdma_prepared_qp_size, - NULL, resource->comp_channel, GetRdmaCompVector()); - if (NULL == resource->send_cq) { + nullptr, resource->comp_channel, GetRdmaCompVector()); + if (nullptr == resource->send_cq) { PLOG(WARNING) << "Fail to create send CQ"; - return NULL; + return nullptr; } resource->recv_cq = IbvCreateCq(GetRdmaContext(), FLAGS_rdma_prepared_qp_size, - NULL, resource->comp_channel, GetRdmaCompVector()); - if (NULL == resource->recv_cq) { + nullptr, resource->comp_channel, GetRdmaCompVector()); + if (nullptr == resource->recv_cq) { PLOG(WARNING) << "Fail to create recv CQ"; - return NULL; + return nullptr; } resource->qp = AllocateQp(resource->send_cq, resource->recv_cq, sq_size, rq_size); - if (NULL == resource->qp) { + if (nullptr == resource->qp) { PLOG(WARNING) << "Fail to create QP"; - return NULL; + return nullptr; } } else { resource->polling_cq = - IbvCreateCq(GetRdmaContext(), 2 * FLAGS_rdma_prepared_qp_size, NULL, NULL, 0); - if (NULL == resource->polling_cq) { + IbvCreateCq(GetRdmaContext(), 2 * FLAGS_rdma_prepared_qp_size, nullptr, nullptr, 0); + if (nullptr == resource->polling_cq) { PLOG(WARNING) << "Fail to create polling CQ"; - return NULL; + return nullptr; } resource->qp = AllocateQp(resource->polling_cq, resource->polling_cq, sq_size, rq_size); - if (NULL == resource->qp) { + if (nullptr == resource->qp) { PLOG(WARNING) << "Fail to create QP"; - return NULL; + return nullptr; } } @@ -1101,7 +1101,7 @@ int RdmaEndpoint::DoAllocateResources() { return 0; } - CHECK(_resource == NULL); + CHECK(_resource == nullptr); if (_sq_size <= FLAGS_rdma_prepared_qp_size && _rq_size <= FLAGS_rdma_prepared_qp_size) { @@ -1114,7 +1114,7 @@ int RdmaEndpoint::DoAllocateResources() { if (!_resource) { _resource = AllocateQpCq(_sq_size, _rq_size); } else { - _resource->next = NULL; + _resource->next = nullptr; } if (!_resource) { return -1; @@ -1156,7 +1156,7 @@ int RdmaEndpoint::DoAllocateResources() { if (_rbuf.size() != _rq_size) { return -1; } - _rbuf_data.resize(_rq_size, NULL); + _rbuf_data.resize(_rq_size, nullptr); if (_rbuf_data.size() != _rq_size) { return -1; } @@ -1196,7 +1196,7 @@ int RdmaEndpoint::BringUpQp(const ParsedHello& remote, bool is_server) { // Client: `remote->ece' is the server's reduced ECE; // just set it here. bool use_ece = true; - if (IbvSetEce != NULL && remote.ece.has_value()) { + if (IbvSetEce != nullptr && remote.ece.has_value()) { ibv_ece ece = *remote.ece; int err = IbvSetEce(_resource->qp, &ece); if (err != 0) { @@ -1262,7 +1262,7 @@ int RdmaEndpoint::BringUpQp(const ParsedHello& remote, bool is_server) { // On the server side, now that the QP reached RTS, query the reduced/negotiated // ECE (the subset of enhancements supported by both peers) so it can be returned // to the client in the server hello. - if (is_server && use_ece && IbvQueryEce != NULL && remote.ece.has_value()) { + if (is_server && use_ece && IbvQueryEce != nullptr && remote.ece.has_value()) { ibv_ece ece; int qerr = IbvQueryEce(_resource->qp, &ece); if (qerr == 0) { @@ -1277,7 +1277,7 @@ int RdmaEndpoint::BringUpQp(const ParsedHello& remote, bool is_server) { } static void DeallocateCq(ibv_cq* cq) { - if (NULL == cq) { + if (nullptr == cq) { return; } @@ -1286,7 +1286,7 @@ static void DeallocateCq(ibv_cq* cq) { } static int DrainCq(ibv_cq* cq) { - if (NULL == cq) { + if (nullptr == cq) { return 0; } @@ -1318,27 +1318,27 @@ void RdmaEndpoint::DeallocateResources() { } } - if (NULL != _resource->send_cq) { + if (nullptr != _resource->send_cq) { IbvAckCqEvents(_resource->send_cq, _send_cq_events); } - if (NULL != _resource->recv_cq) { + if (nullptr != _resource->recv_cq) { IbvAckCqEvents(_resource->recv_cq, _recv_cq_events); } bool remove_consumer = true; _reclaim: if (!move_to_rdma_resource_list) { - if (NULL != _resource->qp) { + if (nullptr != _resource->qp) { int err = IbvDestroyQp(_resource->qp); LOG_IF(WARNING, 0 != err) << "Fail to destroy QP: " << berror(err); - _resource->qp = NULL; + _resource->qp = nullptr; } DeallocateCq(_resource->polling_cq); DeallocateCq(_resource->send_cq); DeallocateCq(_resource->recv_cq); - if (NULL != _resource->comp_channel) { + if (nullptr != _resource->comp_channel) { // Destroy send_comp_channel will destroy this fd, // so that we should remove it from epoll fd first int fd = _resource->comp_channel->fd; @@ -1349,12 +1349,12 @@ void RdmaEndpoint::DeallocateResources() { } - _resource->polling_cq = NULL; - _resource->send_cq = NULL; - _resource->recv_cq = NULL; - _resource->comp_channel = NULL; + _resource->polling_cq = nullptr; + _resource->send_cq = nullptr; + _resource->recv_cq = nullptr; + _resource->comp_channel = nullptr; delete _resource; - _resource = NULL; + _resource = nullptr; } if (INVALID_SOCKET_ID != _cq_sid) { @@ -1363,7 +1363,7 @@ void RdmaEndpoint::DeallocateResources() { if (remove_consumer) { s->_io_event.RemoveConsumer(s->_fd); } - s->_user = NULL; // Do not release user (this RdmaEndpoint). + s->_user = nullptr; // Do not release user (this RdmaEndpoint). s->_fd = -1; // Already remove fd from epoll fd. s->SetFailed(); } @@ -1393,7 +1393,7 @@ void RdmaEndpoint::DeallocateResources() { _resource->next = g_rdma_resource_list; g_rdma_resource_list = _resource; } - _resource = NULL; + _resource = nullptr; } // Detach everything from this endpoint so that the function is @@ -1407,8 +1407,8 @@ void RdmaEndpoint::DeallocateResources() { static const int MAX_CQ_EVENTS = 128; int RdmaEndpoint::GetAndAckEvents(SocketUniquePtr& s) { - void* context = NULL; - ibv_cq* cq = NULL; + void* context = nullptr; + ibv_cq* cq = nullptr; while (true) { if (IbvGetCqEvent(_resource->comp_channel, &cq, &context) != 0) { if (errno != EAGAIN) { @@ -1773,7 +1773,7 @@ void RdmaEndpoint::PollingModeRelease(bthread_tag_t tag) { auto& running = group.running; running.store(false, std::memory_order_relaxed); for (int i = 0; i < FLAGS_rdma_poller_num; ++i) { - bthread_join(pollers[i].tid, NULL); + bthread_join(pollers[i].tid, nullptr); } } diff --git a/src/brpc/rdma/rdma_endpoint.h b/src/brpc/rdma/rdma_endpoint.h index 03bec81408..388e31d78e 100644 --- a/src/brpc/rdma/rdma_endpoint.h +++ b/src/brpc/rdma/rdma_endpoint.h @@ -74,19 +74,19 @@ class RdmaConnect : public AppConnect { private: void Run(); - void (*_done)(int, void*){NULL}; - void* _data{NULL}; + void (*_done)(int, void*){nullptr}; + void* _data{nullptr}; }; struct RdmaResource { - RdmaResource* next{NULL}; - ibv_qp* qp{NULL}; + RdmaResource* next{nullptr}; + ibv_qp* qp{nullptr}; // For polling mode. - ibv_cq* polling_cq{NULL}; + ibv_cq* polling_cq{nullptr}; // For event mode. - ibv_cq* send_cq{NULL}; - ibv_cq* recv_cq{NULL}; - ibv_comp_channel* comp_channel{NULL}; + ibv_cq* send_cq{nullptr}; + ibv_cq* recv_cq{nullptr}; + ibv_comp_channel* comp_channel{nullptr}; RdmaResource() = default; ~RdmaResource(); DISALLOW_COPY_AND_ASSIGN(RdmaResource); diff --git a/src/brpc/rdma/rdma_handshake.cpp b/src/brpc/rdma/rdma_handshake.cpp index 180c2b3f0b..17c6715562 100644 --- a/src/brpc/rdma/rdma_handshake.cpp +++ b/src/brpc/rdma/rdma_handshake.cpp @@ -388,7 +388,7 @@ int RdmaHandshakeClientV3::SendLocalHello() { // Query local ECE capabilities so they can be advertised in the client // hello. v3-only. Best-effort: any failure or missing API just means we // won't advertise ECE (the peer then degrades to no-ECE establishment). - if (FLAGS_rdma_ece && IbvQueryEce != NULL && + if (FLAGS_rdma_ece && IbvQueryEce != nullptr && _ep->_resource && _ep->_resource->qp) { ibv_ece ece; if (IbvQueryEce(_ep->_resource->qp, &ece) == 0) { @@ -505,7 +505,7 @@ std::unique_ptr CreateServerHandshakeByMagic( return std::unique_ptr( new RdmaHandshakeServerV3(ep, source)); } - return NULL; + return nullptr; } } // namespace rdma diff --git a/src/brpc/rdma/rdma_handshake.h b/src/brpc/rdma/rdma_handshake.h index 6238d424f0..2d10220ab9 100644 --- a/src/brpc/rdma/rdma_handshake.h +++ b/src/brpc/rdma/rdma_handshake.h @@ -171,7 +171,7 @@ class RdmaHandshakeServerV3 : public ServerRdmaHandshake { std::unique_ptr CreateClientHandshake(RdmaEndpoint* ep); // Pick the server-side handshake based on the 4B magic already read. -// Returns NULL if `magic` is not a recognized RDMA magic +// Returns nullptr if `magic` is not a recognized RDMA magic // (the caller should then fallback to TCP). // "RDMA" -> RdmaHandshakeServerV2 // "RDM3" -> RdmaHandshakeServerV3 diff --git a/src/brpc/rdma/rdma_handshake_server.cpp b/src/brpc/rdma/rdma_handshake_server.cpp index 4072a11c90..6dfb0c91f5 100644 --- a/src/brpc/rdma/rdma_handshake_server.cpp +++ b/src/brpc/rdma/rdma_handshake_server.cpp @@ -154,7 +154,7 @@ static int SendUnnegotiableHello(Socket* socket, int version) { // Fallback handshake for connections that are NOT in RDMA mode. static ParseResult FallbackServerHandshake(butil::IOBuf* source, Socket* socket) { - if (socket->parsing_context() == NULL) { + if (socket->parsing_context() == nullptr) { if (source->size() < HELLO_MAGIC_LEN) { return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); } @@ -194,7 +194,7 @@ static ParseResult FallbackServerHandshake(butil::IOBuf* source, Socket* socket) CHECK_EQ(source->pop_front(HELLO_ACK_LEN), HELLO_ACK_LEN); // Handshake done (downgraded to TCP); drop the context and let // InputMessenger parse the following real RPC. - socket->reset_parsing_context(NULL); + socket->reset_parsing_context(nullptr); return MakeParseError(PARSE_ERROR_TRY_OTHERS); } diff --git a/src/brpc/rdma/rdma_helper.cpp b/src/brpc/rdma/rdma_helper.cpp index b0e13ad72c..38bd58cac4 100644 --- a/src/brpc/rdma/rdma_helper.cpp +++ b/src/brpc/rdma/rdma_helper.cpp @@ -43,37 +43,37 @@ extern void (*blockmem_deallocate)(void*); namespace brpc { namespace rdma { -void* g_handle_ibverbs = NULL; +void* g_handle_ibverbs = nullptr; bool g_skip_rdma_init = false; -ibv_device** (*IbvGetDeviceList)(int*) = NULL; -void (*IbvFreeDeviceList)(ibv_device**) = NULL; -ibv_context* (*IbvOpenDevice)(ibv_device*) = NULL; -int (*IbvCloseDevice)(ibv_context*) = NULL; -const char* (*IbvGetDeviceName)(ibv_device*) = NULL; -int (*IbvForkInit)(void) = NULL; -int (*IbvQueryDevice)(ibv_context*, ibv_device_attr*) = NULL; -int (*IbvQueryPort)(ibv_context*, uint8_t, ibv_port_attr*) = NULL; -int (*IbvQueryGid)(ibv_context*, uint8_t, int, ibv_gid*) = NULL; -ibv_pd* (*IbvAllocPd)(ibv_context*) = NULL; -int (*IbvDeallocPd)(ibv_pd*) = NULL; -ibv_cq* (*IbvCreateCq)(ibv_context*, int, void*, ibv_comp_channel*, int) = NULL; -int (*IbvDestroyCq)(ibv_cq*) = NULL; -ibv_qp* (*IbvCreateQp)(ibv_pd*, ibv_qp_init_attr*) = NULL; -int (*IbvModifyQp)(ibv_qp*, ibv_qp_attr*, ibv_qp_attr_mask) = NULL; -int (*IbvQueryQp)(ibv_qp*, ibv_qp_attr*, ibv_qp_attr_mask, ibv_qp_init_attr*) = NULL; -int (*IbvDestroyQp)(ibv_qp*) = NULL; -ibv_comp_channel* (*IbvCreateCompChannel)(ibv_context*) = NULL; -int (*IbvDestroyCompChannel)(ibv_comp_channel*) = NULL; -ibv_mr* (*IbvRegMr)(ibv_pd*, void*, size_t, int) = NULL; -int (*IbvDeregMr)(ibv_mr*) = NULL; -int (*IbvGetCqEvent)(ibv_comp_channel*, ibv_cq**, void**) = NULL; -void (*IbvAckCqEvents)(ibv_cq*, unsigned int) = NULL; -int (*IbvGetAsyncEvent)(ibv_context*, ibv_async_event*) = NULL; -void (*IbvAckAsyncEvent)(ibv_async_event*) = NULL; -const char* (*IbvEventTypeStr)(ibv_event_type) = NULL; -int (*IbvQueryEce)(ibv_qp*, ibv_ece*) = NULL; -int (*IbvSetEce)(ibv_qp*, ibv_ece*) = NULL; +ibv_device** (*IbvGetDeviceList)(int*) = nullptr; +void (*IbvFreeDeviceList)(ibv_device**) = nullptr; +ibv_context* (*IbvOpenDevice)(ibv_device*) = nullptr; +int (*IbvCloseDevice)(ibv_context*) = nullptr; +const char* (*IbvGetDeviceName)(ibv_device*) = nullptr; +int (*IbvForkInit)(void) = nullptr; +int (*IbvQueryDevice)(ibv_context*, ibv_device_attr*) = nullptr; +int (*IbvQueryPort)(ibv_context*, uint8_t, ibv_port_attr*) = nullptr; +int (*IbvQueryGid)(ibv_context*, uint8_t, int, ibv_gid*) = nullptr; +ibv_pd* (*IbvAllocPd)(ibv_context*) = nullptr; +int (*IbvDeallocPd)(ibv_pd*) = nullptr; +ibv_cq* (*IbvCreateCq)(ibv_context*, int, void*, ibv_comp_channel*, int) = nullptr; +int (*IbvDestroyCq)(ibv_cq*) = nullptr; +ibv_qp* (*IbvCreateQp)(ibv_pd*, ibv_qp_init_attr*) = nullptr; +int (*IbvModifyQp)(ibv_qp*, ibv_qp_attr*, ibv_qp_attr_mask) = nullptr; +int (*IbvQueryQp)(ibv_qp*, ibv_qp_attr*, ibv_qp_attr_mask, ibv_qp_init_attr*) = nullptr; +int (*IbvDestroyQp)(ibv_qp*) = nullptr; +ibv_comp_channel* (*IbvCreateCompChannel)(ibv_context*) = nullptr; +int (*IbvDestroyCompChannel)(ibv_comp_channel*) = nullptr; +ibv_mr* (*IbvRegMr)(ibv_pd*, void*, size_t, int) = nullptr; +int (*IbvDeregMr)(ibv_mr*) = nullptr; +int (*IbvGetCqEvent)(ibv_comp_channel*, ibv_cq**, void**) = nullptr; +void (*IbvAckCqEvents)(ibv_cq*, unsigned int) = nullptr; +int (*IbvGetAsyncEvent)(ibv_context*, ibv_async_event*) = nullptr; +void (*IbvAckAsyncEvent)(ibv_async_event*) = nullptr; +const char* (*IbvEventTypeStr)(ibv_event_type) = nullptr; +int (*IbvQueryEce)(ibv_qp*, ibv_ece*) = nullptr; +int (*IbvSetEce)(ibv_qp*, ibv_ece*) = nullptr; // NOTE: // ibv_post_send, ibv_post_recv, ibv_poll_cq, ibv_req_notify_cq are all inline function @@ -97,18 +97,18 @@ DEFINE_int32(rdma_port, 1, "The port number to use. For RoCE, it is always 1."); DEFINE_int32(rdma_gid_index, -1, "The GID index to use. -1 means using the last one."); // static const size_t SYSFS_SIZE = 4096; -static ibv_device** g_devices = NULL; -static ibv_context* g_context = NULL; +static ibv_device** g_devices = nullptr; +static ibv_context* g_context = nullptr; static SocketId g_async_socket; -static ibv_pd* g_pd = NULL; -static std::vector* g_mrs = NULL; // mr registered by brpc +static ibv_pd* g_pd = nullptr; +static std::vector* g_mrs = nullptr; // mr registered by brpc static butil::FlatMap* g_user_mrs; // mr registered by user -static butil::Mutex* g_user_mrs_lock = NULL; +static butil::Mutex* g_user_mrs_lock = nullptr; // Store the original IOBuf memalloc and memdealloc functions -static void* (*g_mem_alloc)(size_t) = NULL; -static void (*g_mem_dealloc)(void*) = NULL; +static void* (*g_mem_alloc)(size_t) = nullptr; +static void (*g_mem_dealloc)(void*) = nullptr; namespace { struct IbvDeviceDeleter { @@ -141,32 +141,32 @@ static void GlobalRelease() { } g_user_mrs->clear(); delete g_user_mrs; - g_user_mrs = NULL; + g_user_mrs = nullptr; } delete g_user_mrs_lock; - g_user_mrs_lock = NULL; + g_user_mrs_lock = nullptr; if (g_mrs) { for (size_t i = 0; i < g_mrs->size(); ++i) { IbvDeregMr((*g_mrs)[i]); } delete g_mrs; - g_mrs = NULL; + g_mrs = nullptr; } if (g_pd) { IbvDeallocPd(g_pd); - g_pd = NULL; + g_pd = nullptr; } if (g_context) { IbvCloseDevice(g_context); - g_context = NULL; + g_context = nullptr; } if (g_devices) { IbvFreeDeviceList(g_devices); - g_devices = NULL; + g_devices = nullptr; } } @@ -194,7 +194,7 @@ uint32_t RdmaRegisterMemory(void* buf, size_t size) { static void* BlockAllocate(size_t len) { if (len == 0) { errno = EINVAL; - return NULL; + return nullptr; } void* ptr = AllocBlock(len); if (!ptr) { @@ -349,11 +349,11 @@ static void OnRdmaAsyncEvent(Socket* m) { // Soft-load an OPTIONAL symbol: if the symbol is missing (e.g. the // installed libibverbs predates rdma-core v35 which introduced the ECE -// APIs), leave the function pointer NULL and continue instead of failing +// APIs), leave the function pointer nullptr and continue instead of failing // the whole RDMA initialization. Callers MUST null-check before use. #define LoadSymbolOptional(handle, func, symbol) \ *(void**)(&func) = dlsym(handle, symbol); \ - LOG_IF(WARNING, func == NULL) \ + LOG_IF(WARNING, func == nullptr) \ << "Optional symbol not found (feature disabled): " << symbol; static int ReadRdmaDynamicLib() { @@ -530,28 +530,13 @@ static void GlobalRdmaInitializeOrDieImpl() { ExitWithError(); } - g_user_mrs_lock = new (std::nothrow) butil::Mutex; - if (!g_user_mrs_lock) { - PLOG(WARNING) << "Fail to construct g_user_mrs_lock"; - ExitWithError(); - } - - g_user_mrs = new (std::nothrow) butil::FlatMap(); - if (!g_user_mrs) { - PLOG(WARNING) << "Fail to construct g_user_mrs"; - ExitWithError(); - } - + g_user_mrs_lock = new butil::Mutex; + g_user_mrs = new butil::FlatMap(); if (g_user_mrs->init(65536) < 0) { PLOG(WARNING) << "Fail to initialize g_user_mrs"; ExitWithError(); } - - g_mrs = new (std::nothrow) std::vector; - if (!g_mrs) { - PLOG(ERROR) << "Fail to allocate a RDMA MR list"; - ExitWithError(); - } + g_mrs = new std::vector; ibv_device_attr attr; if (IbvQueryDevice(g_context, &attr) != 0) { @@ -637,7 +622,7 @@ uint32_t RegisterMemoryForRdma(void* buf, size_t len) { } void DeregisterMemoryForRdma(void* buf) { - ibv_mr* mr = NULL; + ibv_mr* mr = nullptr; { BAIDU_SCOPED_LOCK(*g_user_mrs_lock); ibv_mr** mr_ptr = g_user_mrs->seek(buf);