chenBright commented on code in PR #3535:
URL: https://github.com/apache/brpc/pull/3535#discussion_r4463216090
##########
test/brpc_ubring_unittest.cpp:
##########
@@ -934,6 +1887,367 @@ TEST(UBRingRecvChunkHeaderTest,
crafted_chunk_is_rejected) {
chunk.header[UBR_MSG_LEN_INDEX], chunk.header[UBR_MSG_CUR_INDEX]));
}
+TEST_F(UBShmHandshakeTest,
negative_ack_after_server_allocation_serves_tcp_without_delay) {
+ ASSERT_NO_FATAL_FAILURE(PrepareUntilAck("negative_ack_", 1));
+
+ ASSERT_NO_FATAL_FAILURE(SendAndCheckUBFallbackRPC(peer, true, 301));
+ EXPECT_EQ(brpc::handshake::FALLBACK_TCP, adapter->handshake_phase());
+ auto* ep = brpc::UBShmTransport::Get(server_socket.get())->GetUBShmEp();
+ EXPECT_EQ(nullptr, ep->_ub_ring);
+ EXPECT_EQ(brpc::INVALID_SOCKET_ID, ep->_poller_sid);
+ ASSERT_NO_FATAL_FAILURE(SendAndCheckUBFallbackRPC(peer, false, 302));
+ client_transport->DeactivateUpgrade();
+}
+
+#if defined(OS_LINUX)
+TEST_F(UBShmEndpointTest,
cancel_waits_for_callback_before_releasing_descriptor) {
+ using namespace brpc::ubring;
+ struct Context {
+ bthread::CountdownEvent entered;
+ bthread::CountdownEvent release;
+ } context;
+ itimerspec spec{};
+ spec.it_value.tv_nsec = 1;
+ const int fd = TimerStart(&spec, [](void* arg) -> void* {
+ auto* ctx = static_cast<Context*>(arg);
+ ctx->entered.signal();
+ ctx->release.wait();
+ return nullptr;
+ }, &context);
+ ASSERT_GE(fd, 0);
+ bool released = false;
+ BRPC_SCOPE_EXIT {
+ if (!released) { context.release.signal(); }
+ DeleteTimerSafe(fd);
+ };
+ ASSERT_EQ(0,
context.entered.timed_wait(butil::milliseconds_from_now(1000)));
+ bthread::CountdownEvent cancel_started;
+ bthread::CountdownEvent canceled;
+ std::thread cancel([&] {
+ cancel_started.signal();
+ DeleteTimerSafe(fd);
+ canceled.signal();
+ });
+ cancel_started.wait();
+ const int wait_result =
canceled.timed_wait(butil::milliseconds_from_now(100));
+ spec.it_value.tv_nsec = 0;
+ spec.it_value.tv_sec = 10;
+ const int replacement = TimerStart(&spec, [](void*) -> void* {
+ return nullptr;
+ }, nullptr);
+ context.release.signal();
+ released = true;
+ cancel.join();
+ EXPECT_EQ(ETIMEDOUT, wait_result);
+ EXPECT_NE(fd, replacement);
+ EXPECT_GE(fcntl(replacement, F_GETFD), 0);
+ DeleteTimerSafe(replacement);
+}
+
+TEST_F(UBShmEndpointTest, callback_can_cancel_its_own_periodic_timer) {
+ using namespace brpc::ubring;
+ struct Context {
+ bthread::CountdownEvent ready;
+ bthread::CountdownEvent done;
+ int fd;
+ } context;
+ itimerspec spec{};
+ spec.it_value.tv_nsec = 1;
+ spec.it_interval.tv_sec = 10;
+ context.fd = TimerStart(&spec, [](void* arg) -> void* {
+ auto* ctx = static_cast<Context*>(arg);
+ ctx->ready.wait();
+ DeleteTimerSafe(ctx->fd);
+ ctx->done.signal();
+ return nullptr;
+ }, &context);
+ ASSERT_GE(context.fd, 0);
+ context.ready.signal();
+ EXPECT_EQ(0, context.done.timed_wait(butil::milliseconds_from_now(1000)));
+ DeleteTimerSafe(context.fd);
+ EXPECT_EQ(-1, fcntl(context.fd, F_GETFD));
+}
+#endif
+
+
+namespace {
+struct UbsCleanupState {
+ uint8_t local = 0;
+ uint8_t remote = 0;
+ std::atomic<int> unmap_error{UBSM_ERR_NET};
+ std::atomic<int> deallocate_error{0};
+ std::atomic<int> unmaps{0};
+ std::atomic<int> remote_unmaps{0};
+ std::atomic<int> deallocates{0};
+ std::atomic<int> finalizes{0};
+ std::atomic<bool> block_unmap{false};
+ std::atomic<bool> in_unmap{false};
+ std::atomic<bool> finalized_during_unmap{false};
+ bthread::CountdownEvent entered;
+ bthread::CountdownEvent resume;
+};
+UbsCleanupState* ubs_cleanup = nullptr;
+}
+
+class UbsCleanupTest : public UBShmEndpointTest {
+protected:
+ void SetUp() override {
+ using namespace brpc::ubring;
+ UBShmEndpointTest::SetUp();
+ saved_type = GetShmType();
+ saved_list = g_shm_list;
+ saved_fd = g_shm_timer_fd;
+ saved_unmap = ubsmem_shmem_unmap;
+ saved_deallocate = ubsmem_shmem_deallocate;
+ saved_finalize = ubsmem_finalize;
+ saved_delay = FLAGS_ub_flying_io_timeout_s;
+ list = static_cast<ShmList*>(calloc(1, sizeof(ShmList)));
+ ASSERT_NE(nullptr, list);
+ ASSERT_EQ(0, pthread_mutex_init(&list->shm_lock, nullptr));
+ g_shm_list = list;
+ g_shm_timer_fd = 0;
+ FLAGS_ub_flying_io_timeout_s = 1;
+ ubs_cleanup = &state;
+ SetShmType(SHM_TYPE_UBS);
+ ubsmem_shmem_unmap = [](void* address, size_t) {
+ auto& state = *ubs_cleanup;
+ ++state.unmaps;
+ if (address == &state.remote) {
+ ++state.remote_unmaps;
+ }
+ if (state.block_unmap.load()) {
+ state.in_unmap.store(true);
+ state.entered.signal();
+ state.resume.wait();
+ state.in_unmap.store(false);
+ }
+ return state.unmap_error.load();
+ };
+ ubsmem_shmem_deallocate = [](const char*) {
+ ++ubs_cleanup->deallocates;
+ return ubs_cleanup->deallocate_error.load();
+ };
+ ubsmem_finalize = []() {
+ ++ubs_cleanup->finalizes;
+ ubs_cleanup->finalized_during_unmap.store(
+ ubs_cleanup->in_unmap.load());
+ return 0;
+ };
+ }
+ void TearDown() override {
+ using namespace brpc::ubring;
+ if (g_shm_list != nullptr) {
+ DestroyShmTimer(g_shm_list);
+ }
+ g_shm_list = saved_list;
+ g_shm_timer_fd = saved_fd;
+ FLAGS_ub_flying_io_timeout_s = saved_delay;
+ ubsmem_shmem_unmap = saved_unmap;
+ ubsmem_shmem_deallocate = saved_deallocate;
+ ubsmem_finalize = saved_finalize;
+ SetShmType(saved_type);
+ ubs_cleanup = nullptr;
+ UBShmEndpointTest::TearDown();
+ }
+ brpc::ubring::SHM Mapping(bool remote = false) {
+ brpc::ubring::SHM shm{};
+ shm.addr = remote ? &state.remote : &state.local;
+ shm.len = 4 * 1024 * 1024;
+ strcpy(shm.name, remote ? "ubs_remote" : "ubs_local");
+ return shm;
+ }
+ UbsCleanupState state;
+ brpc::ubring::ShmList* list = nullptr;
+ brpc::ubring::ShmList* saved_list = nullptr;
+ int saved_fd = 0;
+ int saved_delay = 0;
+ brpc::ubring::SHM_TYPE saved_type;
+ decltype(brpc::ubring::ubsmem_shmem_unmap) saved_unmap;
+ decltype(brpc::ubring::ubsmem_shmem_deallocate) saved_deallocate;
+ decltype(brpc::ubring::ubsmem_finalize) saved_finalize;
+};
+
+TEST_F(UbsCleanupTest,
abort_hands_off_resources_before_reusing_transaction_slot) {
+ using namespace brpc::ubring;
+ UBRing ring;
+ ASSERT_EQ(UBRING_OK, UBRingManager::AcquireUbrTrxFromMgr(&ring._trx));
+ UbrTrx* old_trx = ring._trx;
+ BRPC_SCOPE_EXIT {
+ if (ring._trx != nullptr) {
+ ring.ReleaseTrx();
+ }
+ };
+ old_trx->local_shm = Mapping();
+ UbrEventQMsg event{};
+ event.flag = UBR_STATE_CONNECTED;
+ old_trx->ubr_tx.local_tx_event_q.addr = reinterpret_cast<uint8_t*>(&event);
+ ASSERT_EQ(UBRING_OK, ring.UbrTrxAbort());
+ EXPECT_EQ(nullptr, ring._trx);
+ EXPECT_EQ(0u, UBRingManager::g_ubr_mgr.trx_num);
+ ASSERT_EQ(1u, list->size);
+
+ UBRing replacement;
+ ASSERT_EQ(UBRING_OK,
UBRingManager::AcquireUbrTrxFromMgr(&replacement._trx));
+ BRPC_SCOPE_EXIT { replacement.ReleaseTrx(); };
+ EXPECT_EQ(old_trx, replacement._trx);
+ state.unmap_error.store(0);
+ UbsShmCallback(list);
+ EXPECT_EQ(0u, list->size);
+ EXPECT_EQ(1, state.deallocates.load());
+ EXPECT_EQ(1u, UBRingManager::g_ubr_mgr.trx_num);
+ EXPECT_EQ(1, ATOMIC_LOAD(replacement._trx->ref_count));
+}
+
+TEST_F(UbsCleanupTest, passive_cleanup_retries_both_mappings_independently) {
+ using namespace brpc::ubring;
+ UBRing ring;
+ ASSERT_EQ(UBRING_OK, UBRingManager::AcquireUbrTrxFromMgr(&ring._trx));
+ ring._trx->local_shm = Mapping();
+ ring._trx->remote_shm = Mapping(true);
+ ring._trx->passive_close = true;
+ ring._trx->close_state = UBR_CLOSE_SECOND;
+ ring._trx->ref_count = 2;
+ UBRing::UbrAsynClearCallback(ring._trx);
+ ASSERT_EQ(UBRING_OK, ring.UbrTrxClose());
+ EXPECT_EQ(0u, UBRingManager::g_ubr_mgr.trx_num);
+ EXPECT_EQ(1, state.remote_unmaps.load());
+ ASSERT_EQ(2u, list->size);
+ // A completed remote task is removed even while the local task still
fails.
+ state.unmap_error.store(0);
+ state.deallocate_error.store(UBSM_ERR_NET);
+ ubsmem_shmem_deallocate = [](const char* name) {
+ ++ubs_cleanup->deallocates;
+ return strcmp(name, "ubs_local") == 0
+ ? ubs_cleanup->deallocate_error.load() : 0;
+ };
+ UbsShmCallback(list);
+ EXPECT_EQ(1u, list->size);
+ state.deallocate_error.store(0);
+ UbsShmCallback(list);
+ EXPECT_EQ(0u, list->size);
+ EXPECT_EQ(4, state.unmaps.load());
+ EXPECT_EQ(3, state.deallocates.load());
+}
+
+TEST_F(UbsCleanupTest,
retry_resumes_at_deallocate_and_accepts_already_deleted_names) {
+ using namespace brpc::ubring;
+ state.unmap_error.store(0);
+ state.deallocate_error.store(UBSM_ERR_IN_USING);
+ SHM shm = Mapping();
+ ASSERT_EQ(UBRING_OK, UbsShmLocalFree(&shm));
+ EXPECT_EQ(nullptr, shm.addr);
+ UbsShmCallback(list);
+ EXPECT_EQ(1u, list->size);
+ state.deallocate_error.store(UBSM_ERR_NOT_FOUND);
+ UbsShmCallback(list);
+ EXPECT_EQ(0u, list->size);
+ EXPECT_EQ(1, state.unmaps.load());
+ EXPECT_EQ(3, state.deallocates.load());
+}
+
+TEST_F(UbsCleanupTest,
remote_retry_keeps_distinct_mappings_and_never_deallocates) {
+ using namespace brpc::ubring;
+ SHM first = Mapping();
+ SHM second = Mapping(true);
+ strcpy(second.name, first.name);
+ ASSERT_EQ(UBRING_OK, UbsShmRemoteFree(&first));
+ ASSERT_EQ(UBRING_OK, UbsShmRemoteFree(&second));
+ ASSERT_EQ(2u, list->size);
+ state.unmap_error.store(0);
+ UbsShmCallback(list);
+ EXPECT_EQ(0u, list->size);
+ EXPECT_EQ(4, state.unmaps.load());
+ EXPECT_EQ(0, state.deallocates.load());
+}
+
+TEST_F(UbsCleanupTest, duplicate_handoff_preserves_completed_operations) {
+ using namespace brpc::ubring;
+ SHM remote = Mapping();
+ SHM local = remote;
+ ASSERT_EQ(UBRING_OK, UbsShmRemoteFree(&remote));
+ state.unmap_error.store(0);
+ ASSERT_EQ(UBRING_OK, UbsShmLocalFree(&local));
+ EXPECT_EQ(1u, list->size);
+ EXPECT_EQ(1, state.unmaps.load());
+ UbsShmCallback(list);
+ EXPECT_EQ(0u, list->size);
+ EXPECT_EQ(2, state.unmaps.load());
+ EXPECT_EQ(1, state.deallocates.load());
+}
+
+TEST_F(UbsCleanupTest, rejected_handoff_keeps_a_cleanup_reference_and_retries)
{
+ using namespace brpc::ubring;
+ UBRing ring;
+ ASSERT_EQ(UBRING_OK, UBRingManager::AcquireUbrTrxFromMgr(&ring._trx));
+ ring._trx->local_shm = Mapping();
+ UbrEventQMsg event{};
+ event.flag = UBR_STATE_CONNECTED;
+ ring._trx->ubr_tx.local_tx_event_q.addr =
reinterpret_cast<uint8_t*>(&event);
+ g_shm_list = nullptr;
+ const RETURN_CODE rc = ring.UbrTrxAbort();
+ EXPECT_TRUE(WaitForUBCondition([&] { return state.unmaps.load() >= 2; }));
+ g_shm_list = list;
+ state.unmap_error.store(0);
+ EXPECT_EQ(UBRING_OK, rc);
+ EXPECT_EQ(nullptr, ring._trx);
+ EXPECT_EQ(1u, UBRingManager::g_ubr_mgr.trx_num);
Review Comment:
`unmaps >= 2` only shows that the retry entered the stub function: the
counter is incremented on line 2014, but the return value is not read until
line 2024. The test thread may execute `unmap_error.store(0)` between these two
steps, causing that retry to succeed immediately and release the last
transaction reference. The `EXPECT_EQ(1u, trx_num)` here would then fail
spuriously because the actual value is 0. Please use a barrier to pause the
retry callback, first assert that the reference is still retained, then restore
the success return value and release the callback, and finally synchronize
before checking that the transaction count reaches zero.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
