This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 16e3c311 Clean up UBRing code and make configuration names (#3471)
16e3c311 is described below
commit 16e3c31158625d31a0c985dac730ccbcd4b4cd49
Author: Chuang Zhang <[email protected]>
AuthorDate: Mon Aug 24 15:54:34 2026 +0800
Clean up UBRing code and make configuration names (#3471)
---
src/brpc/ubshm/shm/shm_ubs.cpp | 11 +++----
src/brpc/ubshm/shm/shm_ubs.h | 4 +--
src/brpc/ubshm/ub_endpoint.cpp | 67 +++++++++++++++++++++---------------------
src/brpc/ubshm/ub_endpoint.h | 31 ++++++++++---------
src/brpc/ubshm/ub_ring.cpp | 42 +++++++++++++++-----------
src/brpc/ubshm/ub_ring.h | 4 +--
src/brpc/ubshm_transport.cpp | 11 +++----
test/brpc_ubring_unittest.cpp | 46 +++++++++++++++++++++++++++++
8 files changed, 135 insertions(+), 81 deletions(-)
diff --git a/src/brpc/ubshm/shm/shm_ubs.cpp b/src/brpc/ubshm/shm/shm_ubs.cpp
index 2d06b0a1..ccbd3259 100644
--- a/src/brpc/ubshm/shm/shm_ubs.cpp
+++ b/src/brpc/ubshm/shm/shm_ubs.cpp
@@ -43,8 +43,9 @@ namespace ubring {
DEFINE_uint32(node_location, 1, "Location of the ub machine.");
DEFINE_bool(shm_wr_delay_comp, true, "Indicates whether to enable the write
relay."
"0: relay; 1: non-relay.");
-DEFINE_int32(ub_flying_io_timeout, 5, "Waiting time for stopping data"
- "sending and receiving when the link is disconnected.");
+DEFINE_int32(ub_flying_io_timeout_s, 5,
+ "Time in seconds to wait for stopping data sending and receiving "
+ "when the link is disconnected.");
char g_region_name[MAX_REGION_NAME_DESC_LENGTH] = {0};
int g_shm_timer_fd = 0;
ShmList *g_shm_list = nullptr;
@@ -450,9 +451,9 @@ void *UbsShmCallback(void* args)
RETURN_CODE UbsShmAddTimer(ShmList *shm_list)
{
- uint32_t timer_interval = FLAGS_ub_flying_io_timeout;
+ const uint32_t timer_interval_s = FLAGS_ub_flying_io_timeout_s;
itimerspec time_spec = {
- .it_interval = {.tv_sec = timer_interval, .tv_nsec = 0},
+ .it_interval = {.tv_sec = timer_interval_s, .tv_nsec = 0},
.it_value = {.tv_sec = 0, .tv_nsec = 1}
};
int timer_fd = TimerStart(&time_spec, UbsShmCallback, (void*)shm_list);
@@ -562,4 +563,4 @@ RETURN_CODE AddShmToList(ShmList *shm_list, SHM *shm)
return UBRING_OK;
}
}
-}
\ No newline at end of file
+}
diff --git a/src/brpc/ubshm/shm/shm_ubs.h b/src/brpc/ubshm/shm/shm_ubs.h
index 3c79dedd..ee936c0a 100644
--- a/src/brpc/ubshm/shm/shm_ubs.h
+++ b/src/brpc/ubshm/shm/shm_ubs.h
@@ -19,7 +19,7 @@
#define BRPC_SHM_UBS_H
namespace brpc {
namespace ubring {
-DECLARE_int32(ub_flying_io_timeout);
+DECLARE_int32(ub_flying_io_timeout_s);
typedef enum TagUbsLogLevel {
UBSM_LOG_DEBUG_LEVEL = 0,
@@ -48,4 +48,4 @@ RETURN_CODE AddShmToList(ShmList *shm_list, SHM *shm);
RETURN_CODE IsExistInShmList(ShmList *shm_list, const SHM *shm);
}
}
-#endif //BRPC_SHM_UBS_H
\ No newline at end of file
+#endif //BRPC_SHM_UBS_H
diff --git a/src/brpc/ubshm/ub_endpoint.cpp b/src/brpc/ubshm/ub_endpoint.cpp
index 45794fdc..31539fda 100644
--- a/src/brpc/ubshm/ub_endpoint.cpp
+++ b/src/brpc/ubshm/ub_endpoint.cpp
@@ -48,9 +48,9 @@ DEFINE_int32(data_queue_size, 4, "data queue size for UB");
DEFINE_bool(ub_trace_verbose, false, "Print log message verbosely");
BRPC_VALIDATE_GFLAG(ub_trace_verbose, brpc::PassValidate);
DEFINE_int32(ub_poller_num, 1, "Poller number in ub polling mode.");
-DEFINE_bool(ub_poller_yield, false, "Yield thread in RDMA polling mode.");
+DEFINE_bool(ub_poller_yield, false, "Yield thread in UBRing polling mode.");
DEFINE_bool(ub_edisp_unsched, false, "Disable event dispatcher schedule");
-DEFINE_bool(ub_disable_bthread, false, "Disable bthread in RDMA");
+DEFINE_bool(ub_disable_bthread, false, "Disable bthread in UBRing polling
mode.");
static const size_t MIN_ONCE_READ = 4096;
static const size_t MAX_ONCE_READ = 524288;
@@ -126,7 +126,7 @@ UBShmEndpoint::UBShmEndpoint(Socket* s)
, _socket_id(s ? s->id() : INVALID_SOCKET_ID)
, _state(UNINIT)
, _ub_ring(nullptr)
- , _cq_sid(INVALID_SOCKET_ID)
+ , _poller_sid(INVALID_SOCKET_ID)
{
_read_butex = bthread::butex_create_checked<butil::atomic<int>>();
}
@@ -141,7 +141,7 @@ void UBShmEndpoint::Reset() {
delete _ub_ring;
_ub_ring = nullptr;
- _cq_sid = INVALID_SOCKET_ID;
+ _poller_sid = INVALID_SOCKET_ID;
_state = UNINIT;
}
@@ -678,15 +678,15 @@ int UBShmEndpoint::AllocateClientResources(ubring::SHM*
local_trx_shm, const cha
SocketOptions options;
options.user = this;
options.keytable_pool = _socket->_keytable_pool;
- if (Socket::Create(options, &_cq_sid) < 0) {
- PLOG(WARNING) << "Fail to create socket for cq";
+ if (Socket::Create(options, &_poller_sid) < 0) {
+ PLOG(WARNING) << "Fail to create socket for UBRing poller";
return -1;
}
int ret = _ub_ring->UbrAllocateLocalShm(local_trx_shm, shm_name);
if (ret != 0) {
return ret;
}
- PollerRegisterEvent(CqSidOp::ADD, EPOLLIN);
+ PollerRegisterEvent(PollerSidOp::ADD, EPOLLIN);
return 0;
}
@@ -703,16 +703,15 @@ int UBShmEndpoint::AllocateServerResources(ubring::SHM*
remote_trx_shm, ubring::
SocketOptions options;
options.user = this;
options.keytable_pool = _socket->_keytable_pool;
- if (Socket::Create(options, &_cq_sid) < 0) {
- PLOG(WARNING) << "Fail to create socket for cq";
+ if (Socket::Create(options, &_poller_sid) < 0) {
+ PLOG(WARNING) << "Fail to create socket for UBRing poller";
return -1;
}
int ret = _ub_ring->UbrAllocateServerShm(remote_trx_shm, local_trx_shm);
if (ret != 0) {
return ret;
}
- // TODO mwj should polling start after the connection is established?
- PollerRegisterEvent(CqSidOp::ADD, EPOLLIN);
+ PollerRegisterEvent(PollerSidOp::ADD, EPOLLIN);
return ret;
}
@@ -720,11 +719,11 @@ void UBShmEndpoint::DeallocateResources() {
if (!_ub_ring) {
return;
}
- PollerRegisterEvent(CqSidOp::REMOVE);
+ PollerRegisterEvent(PollerSidOp::REMOVE);
_ub_ring->UbrTrxClose();
- if (INVALID_SOCKET_ID != _cq_sid) {
+ if (INVALID_SOCKET_ID != _poller_sid) {
SocketUniquePtr s;
- if (Socket::Address(_cq_sid, &s) == 0) {
+ if (Socket::Address(_poller_sid, &s) == 0) {
s->_user = nullptr;
s->_fd = -1;
s->SetFailed();
@@ -840,27 +839,26 @@ int UBShmEndpoint::PollingModeInitialize(bthread_tag_t
tag,
std::unique_ptr<FnArgs> args(static_cast<FnArgs*>(p));
auto poller = args->poller;
auto running = args->running;
- std::unordered_set<CqSidOp, CqSidOpHash, CqSidOpEqual> cq_sids;
- CqSidOp op;
+ std::unordered_set<PollerSidOp, PollerSidOpHash, PollerSidOpEqual>
poller_sids;
+ PollerSidOp op;
if (poller->init_fn) {
poller->init_fn();
}
while (running->load(std::memory_order_relaxed)) {
while (poller->op_queue.Dequeue(op)) {
- if (op.type == CqSidOp::ADD) {
- cq_sids.emplace(op);
- } else if (op.type == CqSidOp::REMOVE) {
- cq_sids.erase(op);
-
- } else if (op.type == CqSidOp::MOD) {
- cq_sids.erase(op);
- cq_sids.emplace(op);
+ if (op.type == PollerSidOp::ADD) {
+ poller_sids.emplace(op);
+ } else if (op.type == PollerSidOp::REMOVE) {
+ poller_sids.erase(op);
+ } else if (op.type == PollerSidOp::MOD) {
+ poller_sids.erase(op);
+ poller_sids.emplace(op);
}
}
- for (auto cq : cq_sids) {
+ for (const auto& poller_sid : poller_sids) {
SocketUniquePtr s;
- if (Socket::Address(cq.sid, &s) < 0) {
+ if (Socket::Address(poller_sid.sid, &s) < 0) {
continue;
}
UBShmEndpoint* ep = static_cast<UBShmEndpoint*>(s->user());
@@ -868,12 +866,12 @@ int UBShmEndpoint::PollingModeInitialize(bthread_tag_t
tag,
continue;
}
- if (cq.event & EPOLLIN) {
- PollIn(ep, cq.event);
+ if (poller_sid.events & EPOLLIN) {
+ PollIn(ep, poller_sid.events);
}
- if (cq.event & EPOLLOUT) {
- PollOut(ep, cq.event);
+ if (poller_sid.events & EPOLLOUT) {
+ PollOut(ep, poller_sid.events);
}
}
if (poller->callback) {
@@ -918,13 +916,14 @@ void UBShmEndpoint::PollingModeRelease(bthread_tag_t tag)
{
}
}
-void UBShmEndpoint::PollerRegisterEvent(CqSidOp::OpType op, uint32_t events) {
- auto index = butil::fmix32(_cq_sid) % FLAGS_ub_poller_num;
+void UBShmEndpoint::PollerRegisterEvent(PollerSidOp::OpType op,
+ uint32_t events) {
+ auto index = butil::fmix32(_poller_sid) % FLAGS_ub_poller_num;
auto& group = _poller_groups[bthread_self_tag()];
auto& pollers = group.pollers;
auto& poller = pollers[index];
- if (INVALID_SOCKET_ID != _cq_sid) {
- poller.op_queue.Enqueue(CqSidOp{_cq_sid, events, op});
+ if (INVALID_SOCKET_ID != _poller_sid) {
+ poller.op_queue.Enqueue(PollerSidOp{_poller_sid, events, op});
}
}
diff --git a/src/brpc/ubshm/ub_endpoint.h b/src/brpc/ubshm/ub_endpoint.h
index 03c51345..a29a0927 100644
--- a/src/brpc/ubshm/ub_endpoint.h
+++ b/src/brpc/ubshm/ub_endpoint.h
@@ -98,19 +98,19 @@ public:
void PollerRegisterEpollOut(bool pollin) {
uint32_t events = EPOLLOUT | EPOLLET;
if (pollin) {
- PollerRegisterEvent(CqSidOp::MOD, events | EPOLLIN);
+ PollerRegisterEvent(PollerSidOp::MOD, events | EPOLLIN);
return;
}
- PollerRegisterEvent(CqSidOp::ADD, events);
+ PollerRegisterEvent(PollerSidOp::ADD, events);
}
void PollerUnRegisterEpollOut(bool pollin) {
uint32_t events = EPOLLIN | EPOLLET;
if (pollin) {
- PollerRegisterEvent(CqSidOp::MOD, events);
+ PollerRegisterEvent(PollerSidOp::MOD, events);
return;
}
- PollerRegisterEvent(CqSidOp::REMOVE);
+ PollerRegisterEvent(PollerSidOp::REMOVE);
}
// Callback when there is new epollin event on TCP fd
@@ -171,7 +171,7 @@ private:
// return -1 if encounter other errno
int WriteToFd(void* data, size_t len);
- // Poll CQ and get the work completion
+ // Poll inbound and outbound UBRing events.
static void PollIn(UBShmEndpoint* ep, uint32_t ep_event);
static void PollOut(UBShmEndpoint* ep, uint32_t ep_event);
@@ -188,32 +188,33 @@ private:
// ub resource
ubring::UBRing* _ub_ring{nullptr};
- SocketId _cq_sid;
+ // Synthetic SocketId registered with the UBRing poller.
+ SocketId _poller_sid;
// butex for inform read events on TCP fd during handshake
butil::atomic<int> *_read_butex;
DISALLOW_COPY_AND_ASSIGN(UBShmEndpoint);
- struct CqSidOp {
+ struct PollerSidOp {
enum OpType {
ADD,
REMOVE,
MOD
};
SocketId sid;
- uint32_t event;
+ uint32_t events;
OpType type;
};
- struct CqSidOpHash {
- std::size_t operator()(const CqSidOp& op) const {
+ struct PollerSidOpHash {
+ std::size_t operator()(const PollerSidOp& op) const {
return op.sid;
}
};
- struct CqSidOpEqual {
- bool operator()(const CqSidOp& lhs, const CqSidOp& rhs) const {
+ struct PollerSidOpEqual {
+ bool operator()(const PollerSidOp& lhs, const PollerSidOp& rhs) const {
return lhs.sid == rhs.sid;
}
};
@@ -221,7 +222,8 @@ private:
// Poller instance
struct BAIDU_CACHELINE_ALIGNMENT Poller {
bthread_t tid{INVALID_BTHREAD};
- butil::MPSCQueue<CqSidOp, butil::ObjectPoolAllocator<CqSidOp>>
op_queue;
+ butil::MPSCQueue<
+ PollerSidOp, butil::ObjectPoolAllocator<PollerSidOp>> op_queue;
// Callback used for io_uring/spdk etc
std::function<void()> callback;
// Init and Destroy function
@@ -236,7 +238,8 @@ private:
};
static std::vector<PollerGroup> _poller_groups;
- void PollerRegisterEvent(CqSidOp::OpType op, uint32_t events = EPOLLET);
+ void PollerRegisterEvent(PollerSidOp::OpType op,
+ uint32_t events = EPOLLET);
};
} // namespace ubring
diff --git a/src/brpc/ubshm/ub_ring.cpp b/src/brpc/ubshm/ub_ring.cpp
index 72df0154..cdc346ed 100644
--- a/src/brpc/ubshm/ub_ring.cpp
+++ b/src/brpc/ubshm/ub_ring.cpp
@@ -16,7 +16,6 @@
// under the License.
#include <errno.h>
-#include <iostream>
#include <gflags/gflags.h>
#include <unistd.h>
#include <ctime>
@@ -29,12 +28,16 @@
namespace brpc {
namespace ubring {
uint32_t g_sleep_time[UBR_TASK_STEP_NUM] = {0};
-#define TIME_COVERSION 1000
-DEFINE_int32(ub_disconnect_timeout, 5, "Ubshm disconnection timeout.");
-DEFINE_int32(ub_connect_timeout, 1, "Ubshm connection timeout.");
-DEFINE_int32(ub_hb_timer_interval, 5, "Heartbeat timer interval.");
-DEFINE_int32(ub_hb_retry_cnt, 10, "Heartbeat retry times.");
-DEFINE_int32(ub_event_queue_timer_interval, 100, "Interval of the
disconnection timer.");
+DEFINE_int32(ub_disconnect_timeout_s, 5,
+ "UBRing disconnection timeout in seconds.");
+DEFINE_int32(ub_connect_timeout_s, 1,
+ "UBRing connection timeout in seconds.");
+DEFINE_int32(ub_hb_timer_interval_s, 5,
+ "UBRing heartbeat timer interval in seconds.");
+DEFINE_int32(ub_hb_retry_cnt, 10,
+ "UBRing heartbeat retry count.");
+DEFINE_int32(ub_event_queue_timer_interval_us, 100,
+ "UBRing disconnection check interval in microseconds.");
UBRing::UBRing()
{}
@@ -69,7 +72,7 @@ RETURN_CODE UBRing::UbrTrxClose() {
((UbrEventQMsg *)_trx->ubr_rx.remote_tx_event_q.addr)->flag =
UBR_STATE_CLOSING;
}
- uint32_t disconnect_timeout = FLAGS_ub_disconnect_timeout;
+ const uint32_t disconnect_timeout_s = FLAGS_ub_disconnect_timeout_s;
uint64_t start_time = GetCurNanoSeconds();
if (_trx->ubr_tx.local_tx_event_q.addr != nullptr && ((UbrEventQMsg
*)_trx->ubr_tx.local_tx_event_q.addr)->flag == UBR_STATE_CONNECTED) {
@@ -82,7 +85,7 @@ RETURN_CODE UBRing::UbrTrxClose() {
}
while (_trx->ubr_rx.local_rx_event_q.addr != nullptr && ((UbrEventQMsg
*)_trx->ubr_rx.local_rx_event_q.addr)->flag != UBR_STATE_CLOSED) {
UbrSetSleepTask(UBR_TASK_CLOSE);
- if (HasTimedOut(start_time, disconnect_timeout) != UBRING_OK) {
+ if (HasTimedOut(start_time, disconnect_timeout_s) != UBRING_OK) {
LOG(WARNING) << "Local shm " << _trx->local_shm.name
<< " wait for the peer to close timed out, force cleanup.";
_trx->ubr_rx.trx_state = UBR_STATE_CLOSED;
@@ -127,9 +130,10 @@ RETURN_CODE UBRing::UbrAddCloseTimer() {
return UBRING_ERR;
}
- uint32_t event_q_timer_interval = FLAGS_ub_event_queue_timer_interval *
TIME_COVERSION;
+ const uint32_t event_q_timer_interval_ns =
+ FLAGS_ub_event_queue_timer_interval_us * USEC_TO_NSEC;
itimerspec time_spec = {
- .it_interval = {.tv_sec = 0, .tv_nsec = event_q_timer_interval},
+ .it_interval = {.tv_sec = 0, .tv_nsec = event_q_timer_interval_ns},
.it_value = {.tv_sec = 0, .tv_nsec = 1}
};
int timer_fd = TimerStart(&time_spec, UbrTrxCloseCallback, (void*)_trx);
@@ -202,7 +206,7 @@ RETURN_CODE UBRing::UbrAddHBTimer() {
}
itimerspec time_spec = {
- .it_interval = {.tv_sec = FLAGS_ub_hb_timer_interval, .tv_nsec =
0},
+ .it_interval = {.tv_sec = FLAGS_ub_hb_timer_interval_s, .tv_nsec =
0},
.it_value = {.tv_sec = 0, .tv_nsec = 1}
};
int timer_fd = TimerStart(&time_spec, UbrTrxHBCallback, (void*)_trx);
@@ -235,7 +239,8 @@ RETURN_CODE UBRing::UbrPassiveClearTrx(UbrTrx *trx, int fd,
PASSIVE_DISC_TYPE ty
DeleteTimerSafe((uint32_t)trx->hb_timer_fd);
type_name = "Ub event callback";
}
- bthread_usleep(FLAGS_ub_flying_io_timeout * 1000000LL); // yield-friendly
sleep
+ constexpr int64_t kMicrosecondsPerSecond = 1000000LL;
+ bthread_usleep(FLAGS_ub_flying_io_timeout_s * kMicrosecondsPerSecond);
int rc = ShmLocalFree(&trx->remote_shm);
if (rc != UBRING_OK) {
@@ -299,7 +304,7 @@ RETURN_CODE UBRing::UbrAddAsynClearTimer(UbrTrx *trx) {
itimerspec time_spec = {
.it_interval = {.tv_sec = 0, .tv_nsec = 0},
- .it_value = {.tv_sec = FLAGS_ub_flying_io_timeout, .tv_nsec = 0}
+ .it_value = {.tv_sec = FLAGS_ub_flying_io_timeout_s, .tv_nsec = 0}
};
int timer_fd = TimerStart(&time_spec, UbrAsynClearCallback, (void*)trx);
@@ -787,8 +792,10 @@ RETURN_CODE UBRing::UbrServerTrxInit(SHM *local_shm, SHM
*remote_shm)
return UBRING_ERR;
}
- ((UbrDataStatusQMsg *)(_trx->ubr_tx.local_data_status_q.addr))->timeout =
FLAGS_ub_connect_timeout;
- ((UbrDataStatusQMsg *)(_trx->ubr_rx.remote_data_status_q.addr))->timeout =
FLAGS_ub_connect_timeout;
+ ((UbrDataStatusQMsg *)(_trx->ubr_tx.local_data_status_q.addr))->timeout =
+ FLAGS_ub_connect_timeout_s;
+ ((UbrDataStatusQMsg *)(_trx->ubr_rx.remote_data_status_q.addr))->timeout =
+ FLAGS_ub_connect_timeout_s;
((UbrEventQMsg *)_trx->ubr_tx.remote_rx_event_q.addr)->flag =
UBR_STATE_CONNECTED;
((UbrEventQMsg *)_trx->ubr_rx.local_rx_event_q.addr)->flag =
UBR_STATE_CONNECTED;
@@ -937,7 +944,8 @@ RETURN_CODE UBRing::ApplyAndMapLocalShm(SHM *local_trx_shm,
const char *local_na
UBRingManager::ReleaseUbrTrxFromMgr(_trx);
return rc;
}
- ((UbrDataStatusQMsg *)_trx->ubr_tx.local_data_status_q.addr)->timeout =
FLAGS_ub_connect_timeout;
+ ((UbrDataStatusQMsg *)_trx->ubr_tx.local_data_status_q.addr)->timeout =
+ FLAGS_ub_connect_timeout_s;
_trx->ubr_rx.capacity = (uint32_t)(_trx->ubr_rx.local_data_q.len /
UBR_MSG_LEN);
rc = UBRingManager::GetUbrDealMsgMaxCnt(_trx->ubr_rx.capacity,
&_trx->ubr_rx.deal_msg_max_cnt);
if (rc != UBRING_OK) {
diff --git a/src/brpc/ubshm/ub_ring.h b/src/brpc/ubshm/ub_ring.h
index f1a5cf14..412c7598 100644
--- a/src/brpc/ubshm/ub_ring.h
+++ b/src/brpc/ubshm/ub_ring.h
@@ -28,7 +28,7 @@
namespace brpc {
namespace ubring {
-DECLARE_int32(ub_flying_io_timeout);
+DECLARE_int32(ub_flying_io_timeout_s);
extern uint32_t g_sleep_time[UBR_TASK_STEP_NUM];
class UBRing : public butil::IReader {
@@ -205,4 +205,4 @@ private:
}
}
-#endif //BRPC_UB_RING_H
\ No newline at end of file
+#endif //BRPC_UB_RING_H
diff --git a/src/brpc/ubshm_transport.cpp b/src/brpc/ubshm_transport.cpp
index df4eb36b..45f6a61f 100644
--- a/src/brpc/ubshm_transport.cpp
+++ b/src/brpc/ubshm_transport.cpp
@@ -88,17 +88,15 @@ ssize_t UBShmTransport::CutFromIOBufList(butil::IOBuf
**buf, size_t ndata) {
int UBShmTransport::WaitEpollOut(butil::atomic<int> *_epollout_butex,
bool pollin, const timespec duetime) {
- // LOG(INFO) << "mwj pollin4=" << pollin << " duetime=" <<
butil::timespec_to_microseconds(duetime);
if (_ub_state == UB_ON) {
- // LOG(INFO) << "mwj pollin1=" << pollin;
const int expected_val =
_epollout_butex->load(butil::memory_order_acquire);
CHECK(_ub_ep != nullptr);
if (!_ub_ep->IsWritable()) {
g_vars->nwaitepollout << 1;
_ub_ep->PollerRegisterEpollOut(pollin);
- auto mwj_ret = bthread::butex_wait(_epollout_butex, expected_val,
&duetime);
- // LOG(INFO) << "mwj pollin2=" << pollin << " mwj_ret=" << mwj_ret;
- if (mwj_ret < 0) {
+ const int wait_rc = bthread::butex_wait(
+ _epollout_butex, expected_val, &duetime);
+ if (wait_rc < 0) {
if (errno != EAGAIN && errno != ETIMEDOUT) {
const int saved_errno = errno;
PLOG(WARNING) << "Fail to wait ub window of " << _socket;
@@ -120,7 +118,6 @@ int UBShmTransport::WaitEpollOut(butil::atomic<int>
*_epollout_butex,
} else {
return _tcp_transport->WaitEpollOut(_epollout_butex, pollin, duetime);
}
- // LOG(INFO) << "mwj return 0";
return 0;
}
@@ -232,4 +229,4 @@ bool UBShmTransport::OptionsAvailableOverUB(const
ServerOptions* opt) {
return true;
}
} // namespace brpc
-#endif
\ No newline at end of file
+#endif
diff --git a/test/brpc_ubring_unittest.cpp b/test/brpc_ubring_unittest.cpp
index 14b2ef1b..53bb0a9b 100644
--- a/test/brpc_ubring_unittest.cpp
+++ b/test/brpc_ubring_unittest.cpp
@@ -16,12 +16,14 @@
#include <gtest/gtest.h>
#include <cstring>
+#include <gflags/gflags.h>
#include <string>
#include "butil/macros.h"
#include "butil/sys_byteorder.h"
#include "brpc/socket.h"
#if BRPC_WITH_UBRING
+#include "brpc/ubshm/common/common.h"
#include "brpc/ubshm/ub_endpoint.h"
#include "brpc/ubshm/shm/shm_def.h"
#include "brpc/ubshm/shm/shm_mgr.h"
@@ -29,6 +31,12 @@
namespace brpc {
namespace ubring {
+DECLARE_int32(ub_disconnect_timeout_s);
+DECLARE_int32(ub_connect_timeout_s);
+DECLARE_int32(ub_hb_timer_interval_s);
+DECLARE_int32(ub_event_queue_timer_interval_us);
+DECLARE_int32(ub_flying_io_timeout_s);
+
extern bool g_skip_ub_init;
} // namespace ubring
} // namespace brpc
@@ -135,6 +143,44 @@ TEST_F(HelloMessageTest, toString_contains_fields) {
EXPECT_NE(std::string::npos, s.find("UBRING_test"));
}
+TEST(UBRingConfigurationTest, time_flags_include_units_and_expected_defaults) {
+ struct TimeFlagExpectation {
+ const char* name;
+ const char* suffix;
+ const char* unit;
+ const char* default_value;
+ };
+ const TimeFlagExpectation expected_flags[] = {
+ {"ub_disconnect_timeout_s", "_s", "seconds", "5"},
+ {"ub_connect_timeout_s", "_s", "seconds", "1"},
+ {"ub_hb_timer_interval_s", "_s", "seconds", "5"},
+ {"ub_event_queue_timer_interval_us", "_us", "microseconds", "100"},
+ {"ub_flying_io_timeout_s", "_s", "seconds", "5"},
+ };
+
+ for (const auto& expected : expected_flags) {
+ GFLAGS_NAMESPACE::CommandLineFlagInfo info;
+ ASSERT_TRUE(GFLAGS_NAMESPACE::GetCommandLineFlagInfo(
+ expected.name, &info)) << expected.name;
+ const std::string flag_name(expected.name);
+ const std::string suffix(expected.suffix);
+ ASSERT_GE(flag_name.size(), suffix.size());
+ EXPECT_EQ(flag_name.size() - suffix.size(), flag_name.rfind(suffix));
+ EXPECT_NE(std::string::npos, info.description.find(expected.unit));
+ EXPECT_EQ(std::string(expected.default_value), info.default_value);
+ }
+
+ EXPECT_EQ(5, brpc::ubring::FLAGS_ub_disconnect_timeout_s);
+ EXPECT_EQ(1, brpc::ubring::FLAGS_ub_connect_timeout_s);
+ EXPECT_EQ(5, brpc::ubring::FLAGS_ub_hb_timer_interval_s);
+ EXPECT_EQ(100, brpc::ubring::FLAGS_ub_event_queue_timer_interval_us);
+ EXPECT_EQ(5, brpc::ubring::FLAGS_ub_flying_io_timeout_s);
+ EXPECT_EQ(100U * USEC_TO_NSEC,
+ static_cast<uint32_t>(
+ brpc::ubring::FLAGS_ub_event_queue_timer_interval_us) *
+ USEC_TO_NSEC);
+}
+
namespace brpc {
namespace ubring {
class UBShmEndpointTest : public ::testing::Test {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
