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 82ff8239 Add IPC_V2 data path for UBRing IPC (#3573)
82ff8239 is described below
commit 82ff823902379eb68b0875ab179995ea6a7627c8
Author: Chuang Zhang <[email protected]>
AuthorDate: Tue Oct 13 19:43:17 2026 +0800
Add IPC_V2 data path for UBRing IPC (#3573)
* Add IPC_V2 data path for UBRing IPC
* Move IPC_V2 ring implementation out of header
---------
Co-authored-by: BGQ99 <[email protected]>
---
src/brpc/ubshm/shm/shm_mgr.cpp | 4 +
src/brpc/ubshm/shm/shm_mgr.h | 4 +-
src/brpc/ubshm/ub_endpoint.cpp | 63 ++-
src/brpc/ubshm/ub_endpoint.h | 13 +-
src/brpc/ubshm/ub_ring.cpp | 376 ++++++++++++-
src/brpc/ubshm/ub_ring.h | 38 +-
.../ubshm/{shm/shm_mgr.h => ubr_data_format.h} | 38 +-
src/brpc/ubshm/ubr_msg_v2.cpp | 276 ++++++++++
src/brpc/ubshm/ubr_msg_v2.h | 176 +++++++
src/brpc/ubshm/ubr_trx.h | 5 +-
test/brpc_ubring_unittest.cpp | 583 +++++++++++++++++++++
11 files changed, 1496 insertions(+), 80 deletions(-)
diff --git a/src/brpc/ubshm/shm/shm_mgr.cpp b/src/brpc/ubshm/shm/shm_mgr.cpp
index ea01113e..d4096bc3 100644
--- a/src/brpc/ubshm/shm/shm_mgr.cpp
+++ b/src/brpc/ubshm/shm/shm_mgr.cpp
@@ -87,6 +87,10 @@ void SetShmType(SHM_TYPE type) {
g_shm_type = type;
}
+SHM_TYPE GetShmType() {
+ return g_shm_type;
+}
+
RETURN_CODE ShmLocalMalloc(SHM *shm) {
if (UNLIKELY(!CheckInputShmParam(shm))) {
LOG(ERROR) << "Input param shm is invalid.";
diff --git a/src/brpc/ubshm/shm/shm_mgr.h b/src/brpc/ubshm/shm/shm_mgr.h
index 597f5e4b..b1e55f03 100644
--- a/src/brpc/ubshm/shm/shm_mgr.h
+++ b/src/brpc/ubshm/shm/shm_mgr.h
@@ -26,6 +26,8 @@ namespace brpc {
namespace ubring {
void SetShmType(SHM_TYPE type);
+SHM_TYPE GetShmType();
+
RETURN_CODE ShmMgrInit(void);
void ShmMgrFini(void);
@@ -48,4 +50,4 @@ RETURN_CODE ShmFree(SHM *shm);
}
}
-#endif //BRPC_SHM_MGR_H
\ No newline at end of file
+#endif //BRPC_SHM_MGR_H
diff --git a/src/brpc/ubshm/ub_endpoint.cpp b/src/brpc/ubshm/ub_endpoint.cpp
index 44685eb3..ec7e0842 100644
--- a/src/brpc/ubshm/ub_endpoint.cpp
+++ b/src/brpc/ubshm/ub_endpoint.cpp
@@ -33,6 +33,7 @@
#include "brpc/ubshm/ub_helper.h"
#include "brpc/ubshm/ub_endpoint.h"
#include "brpc/ubshm/shm/shm_def.h"
+#include "brpc/ubshm/shm/shm_mgr.h"
#include "brpc/ubshm/common/common.h"
#include "brpc/ubshm_transport.h"
#include "brpc/ubshm/ubr_trx.h"
@@ -86,6 +87,27 @@ void HelloFormatExtension::Deserialize(const void* data) {
format_id = butil::NetToHost16(net_format_id);
}
+UbrDataFormat PreferredDataFormatForShmType(SHM_TYPE shm_type) {
+ switch (shm_type) {
+ case SHM_TYPE_IPC:
+ return UBR_DATA_FORMAT_IPC_V2;
+ case SHM_TYPE_UBS:
+ return UBR_DATA_FORMAT_LEGACY_64;
+ default:
+ return UBR_DATA_FORMAT_NONE;
+ }
+}
+
+UbrDataFormat SelectDataFormat(UbrDataFormat local_format,
+ uint16_t remote_format_id) {
+ if ((local_format != UBR_DATA_FORMAT_LEGACY_64 &&
+ local_format != UBR_DATA_FORMAT_IPC_V2) ||
+ remote_format_id != static_cast<uint16_t>(local_format)) {
+ return UBR_DATA_FORMAT_NONE;
+ }
+ return local_format;
+}
+
void HelloMessage::Serialize(void* data) const {
char* current_pos = static_cast<char*>(data);
const uint16_t net_msg_len = butil::HostToNet16(msg_len);
@@ -426,6 +448,8 @@ void* UBShmEndpoint::ProcessHandshakeAtClient(void* arg) {
return nullptr;
}
+ const UbrDataFormat local_format =
+ PreferredDataFormatForShmType(GetShmType());
UbrDataFormat selected_format = UBR_DATA_FORMAT_NONE;
if (!HelloNegotiationValid(remote_msg)) {
LOG(WARNING) << "Fail to negotiate with server, fallback to tcp:"
@@ -433,7 +457,8 @@ void* UBShmEndpoint::ProcessHandshakeAtClient(void* arg) {
ub_transport->_ub_state = UBShmTransport::UB_OFF;
} else {
HelloFormatExtension local_extension = {
- HelloFormatExtension::WIRE_SIZE, UBR_DATA_FORMAT_LEGACY_64};
+ HelloFormatExtension::WIRE_SIZE,
+ static_cast<uint16_t>(local_format)};
local_extension.Serialize(data);
ep->_state = C_FORMAT_SEND;
if (ep->WriteToFd(data, HelloFormatExtension::WIRE_SIZE) < 0) {
@@ -460,16 +485,19 @@ void* UBShmEndpoint::ProcessHandshakeAtClient(void* arg) {
}
HelloFormatExtension remote_extension;
remote_extension.Deserialize(data);
- if (remote_extension.extension_len != HelloFormatExtension::WIRE_SIZE
||
- remote_extension.format_id == UBR_DATA_FORMAT_NONE ||
- remote_extension.format_id != local_extension.format_id) {
+ if (remote_extension.extension_len == HelloFormatExtension::WIRE_SIZE)
{
+ selected_format =
+ SelectDataFormat(local_format, remote_extension.format_id);
+ }
+ if (selected_format == UBR_DATA_FORMAT_NONE) {
LOG(WARNING) << "Fail to negotiate data format with server, "
<< "fallback to tcp:" << s->description();
ub_transport->_ub_state = UBShmTransport::UB_OFF;
} else {
- selected_format = UBR_DATA_FORMAT_LEGACY_64;
ep->_state = C_MAP_REMOTE_SHM;
- if (ep->_ub_ring->UbrMapRemoteShm(&local_trx_shm, shm_name) < 0) {
+ if (ep->_ub_ring->UbrMapRemoteShm(
+ &local_trx_shm, shm_name, remote_msg.len,
+ selected_format) < 0) {
LOG(WARNING) << "Fail to map the remote shm, fallback to tcp:"
<< s->description();
ub_transport->_ub_state = UBShmTransport::UB_OFF;
@@ -563,6 +591,8 @@ void* UBShmEndpoint::ProcessHandshakeAtServer(void* arg) {
ep->_state = FAILED;
return nullptr;
}
+ const UbrDataFormat local_format =
+ PreferredDataFormatForShmType(GetShmType());
if (!HelloNegotiationValid(remote_msg)) {
LOG(WARNING) << "Fail to negotiate with client, fallback to tcp:"
<< s->description();
@@ -588,7 +618,8 @@ void* UBShmEndpoint::ProcessHandshakeAtServer(void* arg) {
LOG(WARNING) << "Copy client shared memory name failed, ret=" <<
result;
ub_transport->_ub_state = UBShmTransport::UB_OFF;
}
- if (result >= 0 && ep->AllocateServerResources(&remote_trx_shm,
&local_trx_shm) < 0) {
+ if (result >= 0 && ep->AllocateServerResources(
+ &remote_trx_shm, &local_trx_shm, local_format) < 0) {
LOG(WARNING) << "Fail to allocate ub resources, fallback to tcp:"
<< s->description();
ub_transport->_ub_state = UBShmTransport::UB_OFF;
@@ -636,10 +667,13 @@ void* UBShmEndpoint::ProcessHandshakeAtServer(void* arg) {
remote_extension.Deserialize(data);
HelloFormatExtension local_extension = {
HelloFormatExtension::WIRE_SIZE, UBR_DATA_FORMAT_NONE};
- if (remote_extension.extension_len == HelloFormatExtension::WIRE_SIZE
&&
- remote_extension.format_id == UBR_DATA_FORMAT_LEGACY_64) {
- local_extension.format_id = UBR_DATA_FORMAT_LEGACY_64;
- selected_format = UBR_DATA_FORMAT_LEGACY_64;
+ if (remote_extension.extension_len == HelloFormatExtension::WIRE_SIZE)
{
+ selected_format =
+ SelectDataFormat(local_format, remote_extension.format_id);
+ }
+ if (selected_format != UBR_DATA_FORMAT_NONE) {
+ local_extension.format_id =
+ static_cast<uint16_t>(selected_format);
} else {
ub_transport->_ub_state = UBShmTransport::UB_OFF;
}
@@ -783,7 +817,9 @@ int UBShmEndpoint::AllocateClientResources(ubring::SHM*
local_trx_shm, const cha
return 0;
}
-int UBShmEndpoint::AllocateServerResources(ubring::SHM* remote_trx_shm,
ubring::SHM* local_trx_shm) {
+int UBShmEndpoint::AllocateServerResources(ubring::SHM* remote_trx_shm,
+ ubring::SHM* local_trx_shm,
+ UbrDataFormat format) {
if (BAIDU_UNLIKELY(g_skip_ub_init)) {
// For UT
return 0;
@@ -800,7 +836,8 @@ int UBShmEndpoint::AllocateServerResources(ubring::SHM*
remote_trx_shm, ubring::
PLOG(WARNING) << "Fail to create socket for UBRing poller";
return -1;
}
- int ret = _ub_ring->UbrAllocateServerShm(remote_trx_shm, local_trx_shm);
+ int ret = _ub_ring->UbrAllocateServerShm(
+ remote_trx_shm, local_trx_shm, format);
if (ret != 0) {
return ret;
}
diff --git a/src/brpc/ubshm/ub_endpoint.h b/src/brpc/ubshm/ub_endpoint.h
index 2d7b1f55..526966c9 100644
--- a/src/brpc/ubshm/ub_endpoint.h
+++ b/src/brpc/ubshm/ub_endpoint.h
@@ -30,6 +30,7 @@
#include "butil/macros.h"
#include "butil/containers/mpsc_queue.h"
#include "brpc/socket.h"
+#include "brpc/ubshm/ubr_data_format.h"
#include "brpc/ubshm/ub_helper.h"
#include "brpc/ubshm/ub_ring.h"
#include "brpc/ubshm/shm/shm_def.h"
@@ -43,11 +44,6 @@ DECLARE_int32(ub_poller_num);
DECLARE_bool(ub_edisp_unsched);
DECLARE_bool(ub_disable_bthread);
-enum UbrDataFormat {
- UBR_DATA_FORMAT_NONE = 0,
- UBR_DATA_FORMAT_LEGACY_64 = 1,
-};
-
struct HelloFormatExtension {
// The V3 format extension is a fixed-size frame. A different wire size
// requires negotiation through a new hello version.
@@ -60,6 +56,10 @@ struct HelloFormatExtension {
void Deserialize(const void* data);
};
+UbrDataFormat PreferredDataFormatForShmType(SHM_TYPE shm_type);
+UbrDataFormat SelectDataFormat(UbrDataFormat local_format,
+ uint16_t remote_format_id);
+
struct HelloMessage {
void Serialize(void* data) const;
void Deserialize(void* data);
@@ -176,7 +176,8 @@ private:
// Return 0 if success, -1 if failed and errno set
int AllocateClientResources(SHM* local_trx_shm, const char* shm_name);
- int AllocateServerResources(SHM* remote_trx_shm, SHM* local_trx_shm);
+ int AllocateServerResources(SHM* remote_trx_shm, SHM* local_trx_shm,
+ UbrDataFormat format);
// Release resources
void DeallocateResources();
diff --git a/src/brpc/ubshm/ub_ring.cpp b/src/brpc/ubshm/ub_ring.cpp
index ea342673..595b96d0 100644
--- a/src/brpc/ubshm/ub_ring.cpp
+++ b/src/brpc/ubshm/ub_ring.cpp
@@ -19,11 +19,13 @@
#include <gflags/gflags.h>
#include <unistd.h>
#include <ctime>
+#include <limits>
#include "bthread/bthread.h"
#include "butil/logging.h"
#include "brpc/ubshm/ub_ring.h"
#include "brpc/ubshm/ub_ring_manager.h"
#include "brpc/ubshm/shm/shm_ipc.h"
+#include "brpc/ubshm/ubr_msg_v2.h"
namespace brpc {
namespace ubring {
@@ -336,6 +338,10 @@ void *UBRing::UbrAsynClearCallback(void *args)
int UBRing::UbrTrxSend(const void *buf, uint32_t buf_len)
{
+ if (_trx == nullptr || _trx->data_format != UBR_DATA_FORMAT_LEGACY_64) {
+ errno = EPROTONOSUPPORT;
+ return UBRING_ERR;
+ }
if (UNLIKELY(CheckTrxSendPreCheck(_trx) != UBRING_OK)) {
return UBRING_ERR;
}
@@ -378,6 +384,10 @@ int UBRing::UbrTrxSend(const void *buf, uint32_t buf_len)
int UBRing::UbrTrxRecv(void *buf, uint32_t buf_len)
{
+ if (_trx == nullptr || _trx->data_format != UBR_DATA_FORMAT_LEGACY_64) {
+ errno = EPROTONOSUPPORT;
+ return UBRING_ERR;
+ }
RETURN_CODE rc = UBRING_OK;
if (UNLIKELY((rc = CheckTrxRecvParam(_trx, buf, buf_len)) != UBRING_OK)) {
return (rc == UBR_NOT_CONNECTED) ? 0 : rc;
@@ -393,6 +403,10 @@ int UBRing::UbrTrxRecv(void *buf, uint32_t buf_len)
int UBRing::UbrTrxRecvBlockMode(uint8_t *dest, uint32_t buf_len)
{
+ if (_trx == nullptr || _trx->data_format != UBR_DATA_FORMAT_LEGACY_64) {
+ errno = EPROTONOSUPPORT;
+ return UBRING_ERR;
+ }
RETURN_CODE rc = UBRING_OK;
if (UNLIKELY((rc = CheckTrxRecvParam(_trx, dest, buf_len)) != UBRING_OK)) {
return (rc == UBR_NOT_CONNECTED) ? 0 : rc;
@@ -449,11 +463,102 @@ int UBRing::UbrTrxRecvBlockMode(uint8_t *dest, uint32_t
buf_len)
return (int)total_copied;
}
+ssize_t UBRing::UbrIpcV2Writev(const struct iovec* iov, int iovcnt)
+{
+ size_t total = 0;
+ if (!ValidateIpcV2Iov(iov, iovcnt, total)) {
+ errno = EINVAL;
+ return -1;
+ }
+ if (total > static_cast<size_t>(std::numeric_limits<ssize_t>::max())) {
+ errno = EMSGSIZE;
+ return -1;
+ }
+ if (_trx->ubr_tx.remote_data_q.addr == nullptr ||
+ _trx->ubr_tx.local_data_status_q.addr == nullptr ||
+ _trx->ubr_tx.remote_rx_event_q.addr == nullptr) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ UbrDataStatusQMsg* status = reinterpret_cast<UbrDataStatusQMsg*>(
+ _trx->ubr_tx.local_data_status_q.addr);
+ IpcV2TxView tx(
+ reinterpret_cast<IpcV2Slot*>(_trx->ubr_tx.remote_data_q.addr),
+ &status->tail, _trx->ubr_tx.capacity, &_trx->ubr_tx.write_pos);
+ size_t written = 0;
+ const IpcV2RingResult result = tx.TryWritev(iov, iovcnt, written);
+ if (result == IPC_V2_RING_RETRY) {
+ return UBRING_RETRY;
+ }
+ if (result == IPC_V2_RING_TOO_LARGE) {
+ errno = EMSGSIZE;
+ return -1;
+ }
+ if (result != IPC_V2_RING_OK) {
+ errno = EBADMSG;
+ return -1;
+ }
+ if (written != 0) {
+ ++_trx->ubr_tx.out_io_id;
+ reinterpret_cast<UbrEventQMsg*>(
+ _trx->ubr_tx.remote_rx_event_q.addr)->io_id =
+ _trx->ubr_tx.out_io_id;
+ }
+ return static_cast<ssize_t>(written);
+}
+
+ssize_t UBRing::UbrIpcV2Readv(const struct iovec* iov, int iovcnt)
+{
+ size_t total = 0;
+ if (!ValidateIpcV2Iov(iov, iovcnt, total)) {
+ errno = EINVAL;
+ return -1;
+ }
+ if (_trx->ubr_rx.local_data_q.addr == nullptr ||
+ _trx->ubr_rx.remote_data_status_q.addr == nullptr) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ UbrDataStatusQMsg* status = reinterpret_cast<UbrDataStatusQMsg*>(
+ _trx->ubr_rx.remote_data_status_q.addr);
+ IpcV2RxView rx(
+ reinterpret_cast<IpcV2Slot*>(_trx->ubr_rx.local_data_q.addr),
+ &status->tail, _trx->ubr_rx.capacity, &_trx->ubr_rx.read_pos,
+ &_trx->ubr_rx.ipc_v2_read_offset);
+ const bool update_ep_eof_pos =
+ _trx->ubr_rx.read_pos == _trx->ubr_rx.ep_eof_pos;
+ size_t read = 0;
+ bool batch_end = false;
+ const IpcV2RingResult result =
+ rx.TryReadv(iov, iovcnt, read, batch_end);
+ if (result == IPC_V2_RING_RETRY) {
+ errno = EAGAIN;
+ return -1;
+ }
+ if (result != IPC_V2_RING_OK) {
+ errno = EBADMSG;
+ return -1;
+ }
+ if (update_ep_eof_pos) {
+ _trx->ubr_rx.ep_eof_pos = _trx->ubr_rx.read_pos;
+ }
+ return static_cast<ssize_t>(read);
+}
+
ssize_t UBRing::UbrTrxWritev(const struct iovec *iov, int iovcnt)
{
if (UNLIKELY(CheckTrxSendPreCheck(_trx) != UBRING_OK)) {
return UBRING_ERR;
}
+ if (_trx->data_format == UBR_DATA_FORMAT_IPC_V2) {
+ return UbrIpcV2Writev(iov, iovcnt);
+ }
+ if (_trx->data_format != UBR_DATA_FORMAT_LEGACY_64) {
+ errno = EPROTONOSUPPORT;
+ return UBRING_ERR;
+ }
size_t buf_len = 0;
for (int i = 0; i < iovcnt; i++) {
@@ -511,6 +616,13 @@ ssize_t UBRing::UbrTrxReadv(const struct iovec *iov, int
iovcnt)
if (UNLIKELY((rc = CheckTrxRecvParam(_trx, iov, (uint32_t)iovcnt)) !=
UBRING_OK)) {
return (rc == UBR_NOT_CONNECTED) ? 0 : rc;
}
+ if (_trx->data_format == UBR_DATA_FORMAT_IPC_V2) {
+ return UbrIpcV2Readv(iov, iovcnt);
+ }
+ if (_trx->data_format != UBR_DATA_FORMAT_LEGACY_64) {
+ errno = EPROTONOSUPPORT;
+ return UBRING_ERR;
+ }
UbrMsgFormat *data_msg = (UbrMsgFormat *)_trx->ubr_rx.local_data_q.addr;
uint32_t read_pos_end = _trx->ubr_rx.read_pos;
uint8_t flag = data_msg[read_pos_end].header[UBR_MSG_FLAG_INDEX];
@@ -529,6 +641,10 @@ ssize_t UBRing::UbrTrxReadv(const struct iovec *iov, int
iovcnt)
ssize_t UBRing::UbrTrxReadvBlockMode(const struct iovec *iov, int iovcnt)
{
+ if (_trx == nullptr || _trx->data_format != UBR_DATA_FORMAT_LEGACY_64) {
+ errno = EPROTONOSUPPORT;
+ return UBRING_ERR;
+ }
RETURN_CODE rc = UBRING_OK;
if (UNLIKELY((rc = CheckTrxRecvParam(_trx, iov, (uint32_t)iovcnt)) !=
UBRING_OK)) {
return (rc == UBR_NOT_CONNECTED) ? 0 : rc;
@@ -561,6 +677,12 @@ RETURN_CODE UBRing::IsUbrTrxReadable(uint32_t ep_event)
if (UNLIKELY(_trx->ubr_tx.trx_state != UBR_STATE_CONNECTED)) {
return UBRING_ERR;
}
+ if (_trx->data_format == UBR_DATA_FORMAT_IPC_V2) {
+ return IsIpcV2Readable(ep_event);
+ }
+ if (_trx->data_format != UBR_DATA_FORMAT_LEGACY_64) {
+ return UBRING_ERR;
+ }
uint64_t io_id = ((UbrEventQMsg
*)_trx->ubr_rx.local_rx_event_q.addr)->io_id;
if ((ep_event & EPOLLET) && io_id == _trx->ubr_rx.in_io_id) {
@@ -606,6 +728,12 @@ RETURN_CODE UBRing::IsUbrTrxWriteable(uint32_t ep_event)
LOG(ERROR) << "The trx is not connected state.";
return UBRING_ERR;
}
+ if (_trx->data_format == UBR_DATA_FORMAT_IPC_V2) {
+ return IsIpcV2Writeable(ep_event);
+ }
+ if (_trx->data_format != UBR_DATA_FORMAT_LEGACY_64) {
+ return UBRING_ERR;
+ }
UbrDataStatusQMsg *data_status_msg = (UbrDataStatusQMsg
*)_trx->ubr_tx.local_data_status_q.addr;
uint32_t cap = _trx->ubr_tx.capacity;
@@ -625,6 +753,82 @@ RETURN_CODE UBRing::IsUbrTrxWriteable(uint32_t ep_event)
return UBRING_OK;
}
+RETURN_CODE UBRing::IsIpcV2Readable(uint32_t ep_event)
+{
+ if (_trx->ubr_rx.local_rx_event_q.addr == nullptr ||
+ _trx->ubr_rx.local_data_q.addr == nullptr ||
+ _trx->ubr_rx.capacity < 2 ||
+ _trx->ubr_rx.read_pos >= _trx->ubr_rx.capacity ||
+ _trx->ubr_rx.ep_eof_pos >= _trx->ubr_rx.capacity) {
+ return UBRING_ERR;
+ }
+ const uint64_t io_id = reinterpret_cast<UbrEventQMsg*>(
+ _trx->ubr_rx.local_rx_event_q.addr)->io_id;
+ if ((ep_event & EPOLLET) && io_id == _trx->ubr_rx.in_io_id) {
+ return MPA_MUXER_NOT_READY;
+ }
+ const uint32_t pos = (ep_event & EPOLLET)
+ ? _trx->ubr_rx.ep_eof_pos : _trx->ubr_rx.read_pos;
+ IpcV2Slot* slots = reinterpret_cast<IpcV2Slot*>(
+ _trx->ubr_rx.local_data_q.addr);
+ IpcV2Slot& slot = slots[pos];
+ const uint32_t state =
+ __atomic_load_n(&slot.header.state, __ATOMIC_ACQUIRE);
+ if (state == IPC_V2_SLOT_EMPTY &&
+ (pos != _trx->ubr_rx.read_pos ||
+ _trx->ubr_rx.ipc_v2_read_offset == 0)) {
+ return MPA_MUXER_NOT_READY;
+ }
+ // Leave metadata validation to UbrIpcV2Readv so PollIn can observe
+ // EBADMSG and fail the connection instead of repeatedly polling it.
+ if (ep_event & EPOLLET) {
+ _trx->ubr_rx.in_io_id = io_id;
+ }
+ return UBRING_OK;
+}
+
+RETURN_CODE UBRing::IsIpcV2Writeable(uint32_t ep_event)
+{
+ if (_trx->ubr_tx.remote_data_q.addr == nullptr ||
+ _trx->ubr_tx.local_data_status_q.addr == nullptr ||
+ _trx->ubr_tx.capacity < 2 ||
+ _trx->ubr_tx.write_pos >= _trx->ubr_tx.capacity) {
+ return UBRING_ERR;
+ }
+ IpcV2Slot* slots = reinterpret_cast<IpcV2Slot*>(
+ _trx->ubr_tx.remote_data_q.addr);
+ const uint32_t current_state = __atomic_load_n(
+ &slots[_trx->ubr_tx.write_pos].header.state, __ATOMIC_ACQUIRE);
+ if (current_state != IPC_V2_SLOT_EMPTY &&
+ current_state != IPC_V2_SLOT_READY) {
+ return UBRING_ERR;
+ }
+ UbrDataStatusQMsg* status = reinterpret_cast<UbrDataStatusQMsg*>(
+ _trx->ubr_tx.local_data_status_q.addr);
+ const uint32_t tail = __atomic_load_n(&status->tail, __ATOMIC_ACQUIRE);
+ if (tail >= _trx->ubr_tx.capacity) {
+ return UBRING_ERR;
+ }
+ const uint32_t capacity = _trx->ubr_tx.capacity;
+ const uint64_t available = _trx->ubr_tx.write_pos > tail
+ ? static_cast<uint64_t>(tail) + capacity - _trx->ubr_tx.write_pos
+ : tail - _trx->ubr_tx.write_pos;
+ const uint32_t remain_slots = static_cast<uint32_t>(available);
+ if (remain_slots == 0) {
+ _trx->ubr_tx.ep_last_cap = 0;
+ return MPA_MUXER_NOT_READY;
+ }
+ if (current_state != IPC_V2_SLOT_EMPTY) {
+ return UBRING_ERR;
+ }
+ if ((ep_event & EPOLLET) && _trx->ubr_tx.ep_last_cap >= remain_slots) {
+ _trx->ubr_tx.ep_last_cap = remain_slots;
+ return MPA_MUXER_NOT_READY;
+ }
+ _trx->ubr_tx.ep_last_cap = remain_slots;
+ return UBRING_OK;
+}
+
RETURN_CODE UBRing::UbrSetTimeout(UbrTaskStep task_type, int timeout)
{
if (task_type >= UBR_TASK_STEP_NUM || timeout < 0) {
@@ -770,19 +974,14 @@ RETURN_CODE UBRing::UbrTrxMapRemoteShm(SHM *remote_shm)
return UBRING_OK;
}
-RETURN_CODE UBRing::UbrServerTrxInit(SHM *local_shm, SHM *remote_shm)
+RETURN_CODE UBRing::InitializeServerLegacyFormat()
{
- RETURN_CODE rc = UbrTrxMapShm(local_shm, remote_shm);
- if (UNLIKELY(rc != UBRING_OK)) {
- LOG(ERROR) <<"Trx map shared memory failed.";
- return rc;
- }
-
uint32_t local_data_msg_cap = (uint32_t)(_trx->ubr_rx.local_data_q.len /
UBR_MSG_LEN);
uint32_t remote_data_msg_cap = (uint32_t)(_trx->ubr_tx.remote_data_q.len /
UBR_MSG_LEN);
_trx->ubr_rx.capacity = local_data_msg_cap;
_trx->ubr_tx.capacity = remote_data_msg_cap;
- rc = UBRingManager::GetUbrDealMsgMaxCnt(_trx->ubr_rx.capacity,
&_trx->ubr_rx.deal_msg_max_cnt);
+ RETURN_CODE rc = UBRingManager::GetUbrDealMsgMaxCnt(
+ _trx->ubr_rx.capacity, &_trx->ubr_rx.deal_msg_max_cnt);
if (UNLIKELY(rc != UBRING_OK)) {
LOG(ERROR) << "Get ubring deal msg max cnt.";
return rc;
@@ -792,6 +991,28 @@ RETURN_CODE UBRing::UbrServerTrxInit(SHM *local_shm, SHM
*remote_shm)
((UbrDataStatusQMsg *)(_trx->ubr_tx.local_data_status_q.addr))->tail =
remote_data_msg_cap - 1;
((UbrDataStatusQMsg *)(_trx->ubr_rx.remote_data_status_q.addr))->tail =
local_data_msg_cap - 1;
+ return UBRING_OK;
+}
+
+RETURN_CODE UBRing::UbrServerTrxInit(SHM* local_shm, SHM* remote_shm,
+ UbrDataFormat format)
+{
+ RETURN_CODE rc = UbrTrxMapShm(local_shm, remote_shm);
+ if (UNLIKELY(rc != UBRING_OK)) {
+ LOG(ERROR) << "Trx map shared memory failed.";
+ return rc;
+ }
+
+ if (format == UBR_DATA_FORMAT_LEGACY_64) {
+ rc = InitializeServerLegacyFormat();
+ } else if (format == UBR_DATA_FORMAT_IPC_V2) {
+ rc = UbrPrepareIpcV2Format() == 0 ? UBRING_OK : UBRING_ERR;
+ } else {
+ rc = UBRING_ERR;
+ }
+ if (UNLIKELY(rc != UBRING_OK)) {
+ return rc;
+ }
if (UNLIKELY(UbrAddTimer() != UBRING_OK)) {
LOG(ERROR) << "Ubr add timer failed, local_name=" << local_shm->name;
@@ -807,10 +1028,58 @@ RETURN_CODE UBRing::UbrServerTrxInit(SHM *local_shm, SHM
*remote_shm)
((UbrEventQMsg *)_trx->ubr_rx.local_rx_event_q.addr)->flag =
UBR_STATE_CONNECTED;
_trx->ubr_tx.trx_state = UBR_STATE_CONNECTED;
_trx->ubr_rx.trx_state = UBR_STATE_CONNECTED;
+ _trx->data_format = format;
return UBRING_OK;
}
-int UBRing::UbrAllocateServerShm(SHM* remote_trx_shm, SHM* local_trx_shm) {
+int UBRing::UbrPrepareIpcV2Format()
+{
+ if (_trx == nullptr || _trx->data_format != UBR_DATA_FORMAT_NONE) {
+ LOG(ERROR) << "Prepare IPC_V2 format failed, invalid transaction
state.";
+ return -1;
+ }
+ if (_trx->ubr_rx.local_data_q.addr == nullptr ||
+ _trx->ubr_tx.remote_data_q.addr == nullptr ||
+ _trx->ubr_tx.local_data_status_q.addr == nullptr ||
+ _trx->ubr_rx.remote_data_status_q.addr == nullptr) {
+ LOG(ERROR) << "Prepare IPC_V2 format failed, queue is not mapped.";
+ return -1;
+ }
+
+ UbrDataStatusQMsg* local_status = reinterpret_cast<UbrDataStatusQMsg*>(
+ _trx->ubr_tx.local_data_status_q.addr);
+ UbrDataStatusQMsg* remote_status = reinterpret_cast<UbrDataStatusQMsg*>(
+ _trx->ubr_rx.remote_data_status_q.addr);
+ IpcV2QueueView tx_queue;
+ IpcV2QueueView rx_queue;
+ if (!MapIpcV2Queue(_trx->ubr_tx.remote_data_q.addr,
+ _trx->ubr_tx.remote_data_q.len,
+ &local_status->tail, tx_queue) ||
+ !MapIpcV2Queue(_trx->ubr_rx.local_data_q.addr,
+ _trx->ubr_rx.local_data_q.len,
+ &remote_status->tail, rx_queue)) {
+ LOG(ERROR) << "Prepare IPC_V2 format failed, invalid queue layout.";
+ return -1;
+ }
+ if (!IpcV2RxView::InitializeShared(rx_queue.slots, rx_queue.tail,
+ rx_queue.capacity)) {
+ LOG(ERROR) << "Prepare IPC_V2 receive queue failed.";
+ return -1;
+ }
+
+ _trx->ubr_tx.capacity = tx_queue.capacity;
+ _trx->ubr_tx.write_pos = 0;
+ _trx->ubr_tx.ep_last_cap = 0;
+ _trx->ubr_rx.capacity = rx_queue.capacity;
+ _trx->ubr_rx.read_pos = 0;
+ _trx->ubr_rx.ep_eof_pos = 0;
+ _trx->ubr_rx.ipc_v2_read_offset = 0;
+ _trx->data_format = UBR_DATA_FORMAT_IPC_V2;
+ return 0;
+}
+
+int UBRing::UbrAllocateServerShm(SHM* remote_trx_shm, SHM* local_trx_shm,
+ UbrDataFormat format) {
UbrSetSleepTask(UBR_TASK_ACCEPT_MAP_FRONT);
if (UNLIKELY((ShmRemoteMalloc(remote_trx_shm)) != UBRING_OK)) {
LOG(ERROR) << "Trx apply remote shared memory failed.";
@@ -831,7 +1100,8 @@ int UBRing::UbrAllocateServerShm(SHM* remote_trx_shm, SHM*
local_trx_shm) {
return -1;
}
_trx->type = TCP_TRX;
- if (UNLIKELY((UbrServerTrxInit(local_trx_shm, remote_trx_shm)) !=
UBRING_OK)) {
+ if (UNLIKELY((UbrServerTrxInit(
+ local_trx_shm, remote_trx_shm, format)) != UBRING_OK)) {
LOG(ERROR) << "Server trx init failed.";
UbrTrxFreeShm(_trx);
UBRingManager::ReleaseUbrTrxFromMgr(_trx);
@@ -849,7 +1119,8 @@ int UBRing::UbrAllocateLocalShm(SHM *local_trx_shm, const
char *shm_name)
}
_trx->type = TCP_TRX;
- if (UNLIKELY((ApplyAndMapLocalShm(local_trx_shm, shm_name)) != UBRING_OK))
{
+ RETURN_CODE rc = ApplyAndMapLocalShm(local_trx_shm, shm_name);
+ if (UNLIKELY(rc != UBRING_OK)) {
LOG(ERROR) << "Trx apply or map local shared memory failed,
local_name=" << shm_name;
_trx = nullptr;
return -1;
@@ -857,29 +1128,71 @@ int UBRing::UbrAllocateLocalShm(SHM *local_trx_shm,
const char *shm_name)
return 0;
}
-int UBRing::UbrMapRemoteShm(SHM *local_trx_shm, const char *local_name)
+int UBRing::UbrMapRemoteShm(SHM *local_trx_shm, const char *local_name,
+ uint64_t remote_shm_len, UbrDataFormat format)
{
- RETURN_CODE rc = UbrMapRemoteShmAddTimer(local_trx_shm, local_name);
+ RETURN_CODE rc = UbrMapRemoteShmAddTimer(
+ local_trx_shm, local_name, remote_shm_len, format);
if (UNLIKELY(rc != UBRING_OK)) {
LOG(ERROR) << "Connect Trx failed, local shm name=" <<
local_trx_shm->name;
return -1;
}
+ return 0;
+}
+
+void UBRing::InitializeClientLegacyFormat()
+{
+ // The local and remote Legacy initialization has already set the RX and
+ // TX capacities. Only prewrite the queues here.
PrewriteUbrRx(&_trx->ubr_rx);
PrewriteUbrTx(&_trx->ubr_tx);
+}
+
+RETURN_CODE UBRing::InitializeClientLocalLegacyFormat()
+{
+ _trx->ubr_rx.capacity =
+ (uint32_t)(_trx->ubr_rx.local_data_q.len / UBR_MSG_LEN);
+ RETURN_CODE rc = UBRingManager::GetUbrDealMsgMaxCnt(
+ _trx->ubr_rx.capacity, &_trx->ubr_rx.deal_msg_max_cnt);
+ if (UNLIKELY(rc != UBRING_OK)) {
+ LOG(ERROR) << "Get ubring deal msg max cnt.";
+ return rc;
+ }
+ return UBRING_OK;
+}
+
+void UBRing::InitializeClientRemoteLegacyFormat()
+{
+ _trx->ubr_tx.capacity =
+ (uint32_t)(_trx->ubr_tx.remote_data_q.len / UBR_MSG_LEN);
+}
+
+void UBRing::ActivateClientFormat()
+{
+ // Only publish the connected states here. UbrMapRemoteShmAddTimer has
+ // already started the timers; preserve that legacy ordering.
((UbrEventQMsg *)_trx->ubr_rx.remote_tx_event_q.addr)->flag =
UBR_STATE_CONNECTED;
((UbrEventQMsg *)_trx->ubr_rx.local_rx_event_q.addr)->flag =
UBR_STATE_CONNECTED;
_trx->ubr_tx.trx_state = UBR_STATE_CONNECTED;
_trx->ubr_rx.trx_state = UBR_STATE_CONNECTED;
- return 0;
}
-RETURN_CODE UBRing::UbrMapRemoteShmAddTimer(SHM *local_trx_shm, const char
*local_name)
+RETURN_CODE UBRing::UbrMapRemoteShmAddTimer(SHM *local_trx_shm, const char
*local_name,
+ uint64_t remote_shm_len,
+ UbrDataFormat format)
{
uint64_t start_time = GetCurNanoSeconds();
- size_t remote_server_len = UBR_MSG_LEN * (((UbrDataStatusQMsg
*)(_trx->ubr_tx.local_data_status_q.addr))->tail + 1) +
- UBR_MSG_LEN * ((DATAQ_ADDR_OFFSET / UBR_MSG_LEN)
+ 1);
- SHM remote_trx_shm = {nullptr, remote_server_len, 0, {0},
local_trx_shm->fd};
+ // Use the base Hello length, not a tail initialized for a particular
format.
+ // Bound the conversion and the current Legacy capacity before mapping.
+ // ShmRemoteMalloc additionally checks the minimum size and allocation
unit.
+ if (remote_shm_len > std::numeric_limits<size_t>::max() ||
+ remote_shm_len / UBR_MSG_LEN > std::numeric_limits<uint32_t>::max()) {
+ LOG(ERROR) << "Remote shared memory length is too large: " <<
remote_shm_len;
+ return SHM_ERR_INPUT_INVALID;
+ }
+ SHM remote_trx_shm = {nullptr, static_cast<size_t>(remote_shm_len),
+ 0, {0}, local_trx_shm->fd};
int result = snprintf(remote_trx_shm.name,
SHM_MAX_NAME_BUFF_LEN,
"%s_%s_%s",
@@ -896,6 +1209,22 @@ RETURN_CODE UBRing::UbrMapRemoteShmAddTimer(SHM
*local_trx_shm, const char *loca
LOG(ERROR) << "Connect Trx map shared memory failed, remote shm=" <<
remote_trx_shm.name;
return rc;
}
+ if (format == UBR_DATA_FORMAT_LEGACY_64) {
+ InitializeClientRemoteLegacyFormat();
+ rc = InitializeClientLocalLegacyFormat();
+ if (LIKELY(rc == UBRING_OK)) {
+ InitializeClientLegacyFormat();
+ }
+ } else if (format == UBR_DATA_FORMAT_IPC_V2) {
+ rc = UbrPrepareIpcV2Format() == 0 ? UBRING_OK : UBRING_ERR;
+ } else {
+ rc = UBRING_ERR;
+ }
+ if (UNLIKELY(rc != UBRING_OK)) {
+ LOG(ERROR) << "Initialize client data format failed, local_name="
+ << local_name << ", format=" << format;
+ return rc;
+ }
if (UNLIKELY(UbrAddTimer() != UBRING_OK)) {
LOG(ERROR) << "Ubr add timer failed, local_name=" << local_name;
@@ -914,6 +1243,8 @@ RETURN_CODE UBRing::UbrMapRemoteShmAddTimer(SHM
*local_trx_shm, const char *loca
return UBRING_ERR_TIMEOUT;
}
+ ActivateClientFormat();
+ _trx->data_format = format;
return UBRING_OK;
}
@@ -952,14 +1283,6 @@ RETURN_CODE UBRing::ApplyAndMapLocalShm(SHM
*local_trx_shm, const char *local_na
}
((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) {
- LOG(ERROR) << "Get ubring deal msg max cnt, local shm name=" <<
local_trx_shm->name;
- ShmLocalFree(local_trx_shm);
- UBRingManager::ReleaseUbrTrxFromMgr(_trx);
- return rc;
- }
return UBRING_OK;
}
@@ -976,7 +1299,6 @@ RETURN_CODE UBRing::ApplyAndMapRemoteShm(SHM
*remote_trx_shm)
ShmRemoteFree(remote_trx_shm);
return rc;
}
- _trx->ubr_tx.capacity = (uint32_t)(_trx->ubr_tx.remote_data_q.len /
UBR_MSG_LEN);
return UBRING_OK;
}
diff --git a/src/brpc/ubshm/ub_ring.h b/src/brpc/ubshm/ub_ring.h
index 0a02dd2f..bd667d1f 100644
--- a/src/brpc/ubshm/ub_ring.h
+++ b/src/brpc/ubshm/ub_ring.h
@@ -107,16 +107,36 @@ public:
return UBRING_OK;
}
- int UbrAllocateServerShm(SHM* remote_trx_shm, SHM* local_trx_shm);
+ int UbrAllocateServerShm(SHM* remote_trx_shm, SHM* local_trx_shm,
+ UbrDataFormat format);
- int UbrMapRemoteShm(SHM *local_trx_shm, const char *local_name);
+ // Prepare the IPC_V2 queue layout and connection-local cursors. This does
+ // not start timers, publish CONNECTED or enable the data path.
+ int UbrPrepareIpcV2Format();
+
+ UbrDataFormat data_format() const {
+ return _trx == nullptr ? UBR_DATA_FORMAT_NONE : _trx->data_format;
+ }
+
+#ifdef UNIT_TEST
+ void SetTrxForTest(UbrTrx* trx) { _trx = trx; }
+#endif
+
+ int UbrMapRemoteShm(SHM *local_trx_shm, const char *local_name,
+ uint64_t remote_shm_len, UbrDataFormat format);
int UbrAllocateLocalShm(SHM *local_trx_shm, const char *shm_name);
- RETURN_CODE UbrMapRemoteShmAddTimer(SHM *local_trx_shm, const char
*local_name);
+ RETURN_CODE UbrMapRemoteShmAddTimer(SHM *local_trx_shm, const char
*local_name,
+ uint64_t remote_shm_len,
+ UbrDataFormat format);
static inline RETURN_CODE CheckTrxSendPreCheck(UbrTrx *trx)
{
+ if (UNLIKELY(trx == nullptr)) {
+ LOG(ERROR) << "Trx send failed, trx is null.";
+ return UBRING_ERR;
+ }
if (UNLIKELY(trx->ubr_tx.trx_state != UBR_STATE_CONNECTED)) {
LOG(ERROR) << "Trx send failed, trx is not connected state.";
return UBRING_ERR;
@@ -202,6 +222,10 @@ public:
}
private:
+ ssize_t UbrIpcV2Writev(const struct iovec* iov, int iovcnt);
+ ssize_t UbrIpcV2Readv(const struct iovec* iov, int iovcnt);
+ RETURN_CODE IsIpcV2Readable(uint32_t ep_event);
+ RETURN_CODE IsIpcV2Writeable(uint32_t ep_event);
RETURN_CODE UbrTrxMapLocalShm(SHM *local_shm);
RETURN_CODE UbrTrxMapRemoteShm(SHM *remote_shm);
RETURN_CODE ApplyAndMapLocalShm(SHM *local_trx_shm, const char
*local_name);
@@ -211,7 +235,13 @@ private:
ssize_t StartReadv(UbrTrx *trx, const struct iovec *iov, int iovcnt,
size_t remain_buf_len);
void PreWriteAddr(uint8_t *addr, size_t len);
RETURN_CODE WritevHasEnoughSpace(size_t buf_len);
- RETURN_CODE UbrServerTrxInit(SHM *local_shm, SHM *remote_shm);
+ RETURN_CODE UbrServerTrxInit(SHM* local_shm, SHM* remote_shm,
+ UbrDataFormat format);
+ RETURN_CODE InitializeServerLegacyFormat();
+ RETURN_CODE InitializeClientLocalLegacyFormat();
+ void InitializeClientRemoteLegacyFormat();
+ void InitializeClientLegacyFormat();
+ void ActivateClientFormat();
static RETURN_CODE UbrClearResourceCheck(UbrTrx *trx, uint64_t start_time,
UbrCloseType close_type);
static RETURN_CODE ClearTrxResource(UbrTrx *trx, uint64_t start_time,
UbrCloseType close_type, int op=0);
diff --git a/src/brpc/ubshm/shm/shm_mgr.h b/src/brpc/ubshm/ubr_data_format.h
similarity index 60%
copy from src/brpc/ubshm/shm/shm_mgr.h
copy to src/brpc/ubshm/ubr_data_format.h
index 597f5e4b..dc834f03 100644
--- a/src/brpc/ubshm/shm/shm_mgr.h
+++ b/src/brpc/ubshm/ubr_data_format.h
@@ -15,37 +15,19 @@
// specific language governing permissions and limitations
// under the License.
-#ifndef BRPC_SHM_MGR_H
-#define BRPC_SHM_MGR_H
-
-#include <stdint.h>
-#include "brpc/ubshm/common/common.h"
-#include "brpc/ubshm/shm/shm_def.h"
+#ifndef BRPC_UBSHM_UBR_DATA_FORMAT_H
+#define BRPC_UBSHM_UBR_DATA_FORMAT_H
namespace brpc {
namespace ubring {
-void SetShmType(SHM_TYPE type);
-
-RETURN_CODE ShmMgrInit(void);
-
-void ShmMgrFini(void);
-
-RETURN_CODE ShmLocalMalloc(SHM *shm);
-
-RETURN_CODE ShmLocalCalloc(SHM *shm);
-
-RETURN_CODE ShmLocalFree(SHM *shm);
-
-RETURN_CODE ShmRemoteMalloc(SHM *shm);
-
-RETURN_CODE ShmRemoteFree(SHM *shm);
-
-RETURN_CODE ShmLocalMmap(SHM *shm, int prot);
-RETURN_CODE ShmMunmap(SHM *shm);
+enum UbrDataFormat {
+ UBR_DATA_FORMAT_NONE = 0,
+ UBR_DATA_FORMAT_LEGACY_64 = 1,
+ UBR_DATA_FORMAT_IPC_V2 = 2,
+};
-RETURN_CODE ShmFree(SHM *shm);
-}
-}
+} // namespace ubring
+} // namespace brpc
-#endif //BRPC_SHM_MGR_H
\ No newline at end of file
+#endif // BRPC_UBSHM_UBR_DATA_FORMAT_H
diff --git a/src/brpc/ubshm/ubr_msg_v2.cpp b/src/brpc/ubshm/ubr_msg_v2.cpp
new file mode 100644
index 00000000..4821f0ee
--- /dev/null
+++ b/src/brpc/ubshm/ubr_msg_v2.cpp
@@ -0,0 +1,276 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "brpc/ubshm/ubr_msg_v2.h"
+
+#include <limits>
+#include <string.h>
+
+namespace brpc {
+namespace ubring {
+
+bool ValidateIpcV2Iov(const struct iovec* iov, int iovcnt, size_t& total) {
+ total = 0;
+ if (iovcnt < 0 || (iovcnt > 0 && iov == nullptr)) {
+ return false;
+ }
+ for (int i = 0; i < iovcnt; ++i) {
+ if (iov[i].iov_len != 0 && iov[i].iov_base == nullptr) {
+ return false;
+ }
+ if (iov[i].iov_len > std::numeric_limits<size_t>::max() - total) {
+ return false;
+ }
+ total += iov[i].iov_len;
+ }
+ return true;
+}
+
+void IpcV2Slot::Initialize() {
+ header.payload_len = 0;
+ header.flags = 0;
+ header.reserved = 0;
+ __atomic_store_n(&header.state, IPC_V2_SLOT_EMPTY, __ATOMIC_RELEASE);
+}
+
+IpcV2SlotResult IpcV2Slot::TryConsume(void* destination, size_t size,
+ uint32_t& offset, size_t& copied,
+ bool& batch_end) {
+ copied = 0;
+ batch_end = false;
+ if (size == 0) {
+ return IPC_V2_SLOT_OK;
+ }
+ if (destination == nullptr) {
+ return IPC_V2_SLOT_INVALID;
+ }
+ const uint32_t state =
+ __atomic_load_n(&header.state, __ATOMIC_ACQUIRE);
+ if (state == IPC_V2_SLOT_EMPTY) {
+ return offset == 0 ? IPC_V2_SLOT_NOT_READY : IPC_V2_SLOT_INVALID;
+ }
+ if (state != IPC_V2_SLOT_READY) {
+ return IPC_V2_SLOT_INVALID;
+ }
+ const uint32_t length = header.payload_len;
+ const uint32_t flags = header.flags;
+ if (length == 0 || length > PAYLOAD_SIZE || offset >= length ||
+ (flags & ~static_cast<uint32_t>(IPC_V2_SLOT_EOF)) != 0 ||
+ header.reserved != 0) {
+ return IPC_V2_SLOT_INVALID;
+ }
+ const size_t remaining = length - offset;
+ copied = size < remaining ? size : remaining;
+ memcpy(destination, payload + offset, copied);
+ offset += static_cast<uint32_t>(copied);
+ if (offset == length) {
+ batch_end = (flags & IPC_V2_SLOT_EOF) != 0;
+ offset = 0;
+ // Do not access this slot after returning ownership to Producer.
+ __atomic_store_n(&header.state, IPC_V2_SLOT_EMPTY, __ATOMIC_RELEASE);
+ }
+ return IPC_V2_SLOT_OK;
+}
+
+bool IpcV2Slot::CalculateCapacity(size_t data_bytes, uint32_t& capacity) {
+ capacity = 0;
+ const size_t slots = data_bytes / SLOT_SIZE;
+ if (slots < 2 || slots > std::numeric_limits<uint32_t>::max()) {
+ return false;
+ }
+ capacity = static_cast<uint32_t>(slots);
+ return true;
+}
+
+IpcV2RingResult IpcV2TxView::TryWritev(const struct iovec* iov, int iovcnt,
+ size_t& written) {
+ written = 0;
+ size_t total = 0;
+ if (!ValidateIpcV2Iov(iov, iovcnt, total)) {
+ return IPC_V2_RING_INVALID;
+ }
+ if (total == 0) {
+ return IPC_V2_RING_OK;
+ }
+ if (_slots == nullptr || _tail == nullptr || _write_pos == nullptr ||
+ _capacity < 2 || *_write_pos >= _capacity) {
+ return IPC_V2_RING_INVALID;
+ }
+ const uint32_t current_state = __atomic_load_n(
+ &_slots[*_write_pos].header.state, __ATOMIC_ACQUIRE);
+ if (current_state != IPC_V2_SLOT_EMPTY &&
+ current_state != IPC_V2_SLOT_READY) {
+ return IPC_V2_RING_INVALID;
+ }
+ const size_t required = total / IpcV2Slot::PAYLOAD_SIZE +
+ (total % IpcV2Slot::PAYLOAD_SIZE != 0);
+ if (required >= _capacity) {
+ return IPC_V2_RING_TOO_LARGE;
+ }
+ const uint32_t tail = __atomic_load_n(_tail, __ATOMIC_ACQUIRE);
+ if (tail >= _capacity) {
+ return IPC_V2_RING_INVALID;
+ }
+ const uint64_t available = *_write_pos > tail
+ ? static_cast<uint64_t>(tail) + _capacity - *_write_pos
+ : tail - *_write_pos;
+ if (available < required) {
+ return IPC_V2_RING_RETRY;
+ }
+ for (size_t i = 0; i < required; ++i) {
+ const uint32_t pos = static_cast<uint32_t>(
+ (static_cast<uint64_t>(*_write_pos) + i) % _capacity);
+ if (__atomic_load_n(&_slots[pos].header.state, __ATOMIC_ACQUIRE) !=
+ IPC_V2_SLOT_EMPTY) {
+ return IPC_V2_RING_INVALID;
+ }
+ }
+
+ int iov_index = 0;
+ size_t iov_offset = 0;
+ size_t remaining = total;
+ while (remaining != 0) {
+ IpcV2Slot& slot = _slots[*_write_pos];
+ const size_t slot_length = remaining < IpcV2Slot::PAYLOAD_SIZE
+ ? remaining : IpcV2Slot::PAYLOAD_SIZE;
+ size_t slot_offset = 0;
+ while (slot_offset < slot_length) {
+ while (iov_index < iovcnt && iov[iov_index].iov_len == 0) {
+ ++iov_index;
+ }
+ const size_t available = iov[iov_index].iov_len - iov_offset;
+ const size_t copy_length = available < slot_length - slot_offset
+ ? available : slot_length - slot_offset;
+ memcpy(slot.payload + slot_offset,
+ static_cast<const uint8_t*>(iov[iov_index].iov_base) +
+ iov_offset,
+ copy_length);
+ slot_offset += copy_length;
+ iov_offset += copy_length;
+ if (iov_offset == iov[iov_index].iov_len) {
+ ++iov_index;
+ iov_offset = 0;
+ }
+ }
+ slot.header.payload_len = static_cast<uint32_t>(slot_length);
+ slot.header.flags = remaining == slot_length
+ ? static_cast<uint32_t>(IPC_V2_SLOT_EOF) : 0u;
+ slot.header.reserved = 0;
+ __atomic_store_n(&slot.header.state, IPC_V2_SLOT_READY,
+ __ATOMIC_RELEASE);
+ *_write_pos = (*_write_pos + 1) % _capacity;
+ remaining -= slot_length;
+ }
+ written = total;
+ return IPC_V2_RING_OK;
+}
+
+bool IpcV2RxView::InitializeShared(IpcV2Slot* slots, uint32_t* tail,
+ uint32_t capacity) {
+ if (slots == nullptr || tail == nullptr || capacity < 2) {
+ return false;
+ }
+ for (uint32_t i = 0; i < capacity; ++i) {
+ slots[i].Initialize();
+ }
+ __atomic_store_n(tail, capacity - 1, __ATOMIC_RELEASE);
+ return true;
+}
+
+IpcV2RingResult IpcV2RxView::TryReadv(const struct iovec* iov, int iovcnt,
+ size_t& read, bool& batch_end) {
+ read = 0;
+ batch_end = false;
+ size_t destination_size = 0;
+ if (!ValidateIpcV2Iov(iov, iovcnt, destination_size)) {
+ return IPC_V2_RING_INVALID;
+ }
+ if (destination_size == 0) {
+ return IPC_V2_RING_OK;
+ }
+ if (_slots == nullptr || _tail == nullptr || _read_pos == nullptr ||
+ _read_offset == nullptr || _capacity < 2 ||
+ *_read_pos >= _capacity) {
+ return IPC_V2_RING_INVALID;
+ }
+
+ int iov_index = 0;
+ size_t iov_offset = 0;
+ while (read < destination_size) {
+ while (iov_index < iovcnt && iov[iov_index].iov_len == 0) {
+ ++iov_index;
+ }
+ size_t copied = 0;
+ bool slot_batch_end = false;
+ const IpcV2SlotResult result = _slots[*_read_pos].TryConsume(
+ static_cast<uint8_t*>(iov[iov_index].iov_base) + iov_offset,
+ iov[iov_index].iov_len - iov_offset, *_read_offset, copied,
+ slot_batch_end);
+ if (result == IPC_V2_SLOT_NOT_READY) {
+ return read == 0 ? IPC_V2_RING_RETRY : IPC_V2_RING_OK;
+ }
+ if (result != IPC_V2_SLOT_OK || copied == 0) {
+ return IPC_V2_RING_INVALID;
+ }
+ read += copied;
+ iov_offset += copied;
+ if (iov_offset == iov[iov_index].iov_len) {
+ ++iov_index;
+ iov_offset = 0;
+ }
+ if (*_read_offset == 0) {
+ // TryConsume releases EMPTY first; tail follows that release.
+ __atomic_store_n(_tail, *_read_pos, __ATOMIC_RELEASE);
+ *_read_pos = (*_read_pos + 1) % _capacity;
+ if (slot_batch_end) {
+ batch_end = true;
+ return IPC_V2_RING_OK;
+ }
+ }
+ }
+ return IPC_V2_RING_OK;
+}
+
+bool MapIpcV2Queue(uint8_t* data_addr, size_t data_len, uint32_t* tail,
+ IpcV2QueueView& queue) {
+ queue = IpcV2QueueView();
+ if (data_addr == nullptr || tail == nullptr ||
+ reinterpret_cast<uintptr_t>(data_addr) % alignof(IpcV2Slot) != 0 ||
+ reinterpret_cast<uintptr_t>(tail) % alignof(uint32_t) != 0) {
+ return false;
+ }
+ uint32_t capacity = 0;
+ if (!IpcV2Slot::CalculateCapacity(data_len, capacity)) {
+ return false;
+ }
+ queue.slots = reinterpret_cast<IpcV2Slot*>(data_addr);
+ queue.tail = tail;
+ queue.capacity = capacity;
+ return true;
+}
+
+bool InitializeIpcV2ReceiveQueue(uint8_t* data_addr, size_t data_len,
+ uint32_t* tail, IpcV2QueueView& queue) {
+ if (!MapIpcV2Queue(data_addr, data_len, tail, queue)) {
+ return false;
+ }
+ return IpcV2RxView::InitializeShared(queue.slots, queue.tail,
+ queue.capacity);
+}
+
+} // namespace ubring
+} // namespace brpc
diff --git a/src/brpc/ubshm/ubr_msg_v2.h b/src/brpc/ubshm/ubr_msg_v2.h
new file mode 100644
index 00000000..a08b4642
--- /dev/null
+++ b/src/brpc/ubshm/ubr_msg_v2.h
@@ -0,0 +1,176 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#ifndef BRPC_UBSHM_UBR_MSG_V2_H
+#define BRPC_UBSHM_UBR_MSG_V2_H
+
+#include <stddef.h>
+#include <stdint.h>
+#include <sys/uio.h>
+#include <type_traits>
+
+namespace brpc {
+namespace ubring {
+
+enum IpcV2SlotState : uint32_t {
+ IPC_V2_SLOT_UNINITIALIZED = 0,
+ IPC_V2_SLOT_EMPTY = 1,
+ IPC_V2_SLOT_READY = 2,
+};
+
+enum IpcV2SlotFlags : uint32_t {
+ // End of one lower-level write batch, not an RPC or connection boundary.
+ IPC_V2_SLOT_EOF = 1u << 0,
+};
+
+enum IpcV2SlotResult {
+ IPC_V2_SLOT_OK,
+ IPC_V2_SLOT_NOT_READY,
+ IPC_V2_SLOT_INVALID,
+};
+
+enum IpcV2RingResult {
+ IPC_V2_RING_OK,
+ IPC_V2_RING_RETRY,
+ IPC_V2_RING_INVALID,
+ IPC_V2_RING_TOO_LARGE,
+};
+
+struct IpcV2SlotHeader {
+ // Access state atomically. The producer publishes READY with release and
+ // the consumer observes it with acquire; recycling EMPTY is symmetric.
+ uint32_t state;
+ uint32_t payload_len;
+ uint32_t flags;
+ uint32_t reserved;
+};
+
+struct alignas(64) IpcV2Slot {
+ enum : size_t {
+ SLOT_SIZE = 4096,
+ PAYLOAD_OFFSET = 16,
+ PAYLOAD_SIZE = SLOT_SIZE - PAYLOAD_OFFSET,
+ };
+
+ IpcV2SlotHeader header;
+ uint8_t payload[PAYLOAD_SIZE];
+
+ // Initialization happens before the peer is allowed to access this ring.
+ void Initialize();
+
+ IpcV2SlotResult TryConsume(void* destination, size_t size,
+ uint32_t& offset, size_t& copied,
+ bool& batch_end);
+
+ // `data_bytes` excludes the separately allocated control area. Trailing
+ // bytes that do not fit one slot are intentionally unused.
+ static bool CalculateCapacity(size_t data_bytes, uint32_t& capacity);
+};
+
+static_assert(std::is_standard_layout<IpcV2SlotHeader>::value,
+ "IPC_V2 header must have a predictable layout");
+static_assert(sizeof(IpcV2SlotHeader) == IpcV2Slot::PAYLOAD_OFFSET,
+ "IPC_V2 header must occupy 16 bytes");
+static_assert(offsetof(IpcV2SlotHeader, state) == 0,
+ "Unexpected IPC_V2 state offset");
+static_assert(offsetof(IpcV2SlotHeader, payload_len) == 4,
+ "Unexpected IPC_V2 payload length offset");
+static_assert(offsetof(IpcV2SlotHeader, flags) == 8,
+ "Unexpected IPC_V2 flags offset");
+static_assert(offsetof(IpcV2SlotHeader, reserved) == 12,
+ "Unexpected IPC_V2 reserved offset");
+static_assert(std::is_standard_layout<IpcV2Slot>::value,
+ "IPC_V2 slot must have a predictable layout");
+static_assert(offsetof(IpcV2Slot, header) == 0,
+ "IPC_V2 header must be first");
+static_assert(offsetof(IpcV2Slot, payload) == IpcV2Slot::PAYLOAD_OFFSET,
+ "Unexpected IPC_V2 payload offset");
+static_assert(sizeof(IpcV2Slot) == IpcV2Slot::SLOT_SIZE,
+ "IPC_V2 slot must occupy 4096 bytes");
+static_assert(alignof(IpcV2Slot) == 64,
+ "IPC_V2 slot must be 64-byte aligned");
+static_assert(__atomic_always_lock_free(sizeof(uint32_t), nullptr),
+ "IPC_V2 slot state requires lock-free 32-bit atomics");
+
+bool ValidateIpcV2Iov(const struct iovec* iov, int iovcnt, size_t& total);
+
+// These views implement only one SPSC direction. UBRing owns notification,
+// activation, teardown and shared-memory lifetime management.
+class IpcV2TxView {
+public:
+ IpcV2TxView(IpcV2Slot* slots, uint32_t* tail, uint32_t capacity,
+ uint32_t* write_pos)
+ : _slots(slots), _tail(tail), _capacity(capacity),
+ _write_pos(write_pos) {}
+
+ IpcV2RingResult TryWritev(const struct iovec* iov, int iovcnt,
+ size_t& written);
+
+private:
+ IpcV2TxView(const IpcV2TxView&) = delete;
+ IpcV2TxView& operator=(const IpcV2TxView&) = delete;
+
+ IpcV2Slot* _slots;
+ uint32_t* _tail;
+ uint32_t _capacity;
+ uint32_t* _write_pos;
+};
+
+class IpcV2RxView {
+public:
+ IpcV2RxView(IpcV2Slot* slots, uint32_t* tail, uint32_t capacity,
+ uint32_t* read_pos, uint32_t* read_offset)
+ : _slots(slots), _tail(tail), _capacity(capacity),
+ _read_pos(read_pos), _read_offset(read_offset) {}
+
+ static bool InitializeShared(IpcV2Slot* slots, uint32_t* tail,
+ uint32_t capacity);
+
+ IpcV2RingResult TryReadv(const struct iovec* iov, int iovcnt,
+ size_t& read, bool& batch_end);
+
+private:
+ IpcV2RxView(const IpcV2RxView&) = delete;
+ IpcV2RxView& operator=(const IpcV2RxView&) = delete;
+
+ IpcV2Slot* _slots;
+ uint32_t* _tail;
+ uint32_t _capacity;
+ uint32_t* _read_pos;
+ uint32_t* _read_offset;
+};
+
+struct IpcV2QueueView {
+ IpcV2QueueView() : slots(nullptr), tail(nullptr), capacity(0) {}
+
+ IpcV2Slot* slots;
+ uint32_t* tail;
+ uint32_t capacity;
+};
+
+// Adapt an existing UBRing DataQ and its existing status tail to IPC_V2.
+// No shared memory is modified unless initialization succeeds.
+bool MapIpcV2Queue(uint8_t* data_addr, size_t data_len,
+ uint32_t* tail, IpcV2QueueView& queue);
+
+bool InitializeIpcV2ReceiveQueue(uint8_t* data_addr, size_t data_len,
+ uint32_t* tail, IpcV2QueueView& queue);
+
+} // namespace ubring
+} // namespace brpc
+
+#endif // BRPC_UBSHM_UBR_MSG_V2_H
diff --git a/src/brpc/ubshm/ubr_trx.h b/src/brpc/ubshm/ubr_trx.h
index b4e89645..4111b8ee 100644
--- a/src/brpc/ubshm/ubr_trx.h
+++ b/src/brpc/ubshm/ubr_trx.h
@@ -23,6 +23,7 @@
#include "brpc/ubshm/shm/shm_def.h"
#include "brpc/ubshm/common/common.h"
#include "brpc/ubshm/common/thread_lock.h"
+#include "brpc/ubshm/ubr_data_format.h"
#include "brpc/ubshm/ubr_msg.h"
/*
+----------------------------------------------------------------------------+
@@ -124,6 +125,7 @@ typedef struct TagUbrRx {
uint32_t deal_msg_num;
uint32_t deal_msg_max_cnt;
uint32_t ep_eof_pos;
+ uint32_t ipc_v2_read_offset;
volatile EventQState trx_state;
} UbrRx;
@@ -133,6 +135,7 @@ typedef struct TagUbrTrx {
uint64_t ubr_id;
uint32_t trx_mgr_index;
UbrTrxType type;
+ UbrDataFormat data_format;
SHM local_shm;
SHM remote_shm;
int timer_fd;
@@ -159,4 +162,4 @@ typedef enum {
}
}
-#endif //BRPC_UBR_TRX_H
\ No newline at end of file
+#endif //BRPC_UBR_TRX_H
diff --git a/test/brpc_ubring_unittest.cpp b/test/brpc_ubring_unittest.cpp
index 295d3f12..c9b9c91d 100644
--- a/test/brpc_ubring_unittest.cpp
+++ b/test/brpc_ubring_unittest.cpp
@@ -15,9 +15,14 @@
// limitations under the License.
#include <gtest/gtest.h>
+#include <algorithm>
+#include <atomic>
+#include <chrono>
#include <cstring>
+#include <limits>
#include <gflags/gflags.h>
#include <string>
+#include <thread>
#include "butil/macros.h"
#include "butil/sys_byteorder.h"
#include "brpc/socket.h"
@@ -30,6 +35,7 @@
#include "brpc/ubshm/ub_ring_manager.h"
#include "brpc/ubshm/ub_ring.h"
#include "brpc/ubshm/ubr_msg.h"
+#include "brpc/ubshm/ubr_msg_v2.h"
namespace brpc {
namespace ubring {
@@ -44,6 +50,542 @@ extern bool g_skip_ub_init;
} // namespace brpc
namespace {
+TEST(IpcV2LayoutTest, fixed_layout) {
+ using brpc::ubring::IpcV2Slot;
+
+ EXPECT_EQ(4096u, sizeof(IpcV2Slot));
+ EXPECT_EQ(64u, alignof(IpcV2Slot));
+ EXPECT_EQ(16u, offsetof(IpcV2Slot, payload));
+ EXPECT_EQ(4080u, sizeof(((IpcV2Slot*)nullptr)->payload));
+ EXPECT_EQ(0, brpc::ubring::UBR_DATA_FORMAT_NONE);
+ EXPECT_EQ(1, brpc::ubring::UBR_DATA_FORMAT_LEGACY_64);
+ EXPECT_EQ(2, brpc::ubring::UBR_DATA_FORMAT_IPC_V2);
+}
+
+TEST(IpcV2RingTest, fixed_writev_readv_boundaries_and_wrap) {
+ using namespace brpc::ubring;
+ IpcV2Slot slots[4] = {};
+ uint32_t tail = 0;
+ uint32_t write_pos = 0;
+ uint32_t read_pos = 0;
+ uint32_t read_offset = 0;
+ IpcV2TxView tx(slots, &tail, 4, &write_pos);
+ IpcV2RxView rx(slots, &tail, 4, &read_pos, &read_offset);
+ EXPECT_EQ(IPC_V2_SLOT_UNINITIALIZED,
+ __atomic_load_n(&slots[0].header.state, __ATOMIC_ACQUIRE));
+ ASSERT_TRUE(IpcV2RxView::InitializeShared(slots, &tail, 4));
+ EXPECT_EQ(3u, __atomic_load_n(&tail, __ATOMIC_ACQUIRE));
+
+ std::string source(IpcV2Slot::PAYLOAD_SIZE + 19, '\0');
+ for (size_t i = 0; i < source.size(); ++i) {
+ source[i] = static_cast<char>((i * 29 + 7) & 0xff);
+ }
+ struct iovec write_iov[4] = {
+ {nullptr, 0},
+ {&source[0], 7},
+ {&source[7], IpcV2Slot::PAYLOAD_SIZE - 3},
+ {&source[IpcV2Slot::PAYLOAD_SIZE + 4], 15},
+ };
+ size_t written = 99;
+ ASSERT_EQ(IPC_V2_RING_OK, tx.TryWritev(write_iov, 4, written));
+ EXPECT_EQ(source.size(), written);
+ EXPECT_EQ(2u, write_pos);
+ EXPECT_EQ(IpcV2Slot::PAYLOAD_SIZE, slots[0].header.payload_len);
+ EXPECT_EQ(0u, slots[0].header.flags);
+ EXPECT_EQ(19u, slots[1].header.payload_len);
+ EXPECT_EQ(static_cast<uint32_t>(IPC_V2_SLOT_EOF), slots[1].header.flags);
+
+ std::string output(source.size(), '?');
+ struct iovec first_read = {&output[0], 31};
+ size_t read = 99;
+ bool batch_end = true;
+ ASSERT_EQ(IPC_V2_RING_OK, rx.TryReadv(&first_read, 1, read, batch_end));
+ EXPECT_EQ(31u, read);
+ EXPECT_FALSE(batch_end);
+ EXPECT_EQ(0u, read_pos);
+ EXPECT_EQ(31u, read_offset);
+ EXPECT_EQ(3u, __atomic_load_n(&tail, __ATOMIC_ACQUIRE));
+ struct iovec remaining_iov[4] = {
+ {nullptr, 0},
+ {&output[31], 5},
+ {&output[36], IpcV2Slot::PAYLOAD_SIZE - 20},
+ {&output[IpcV2Slot::PAYLOAD_SIZE + 16], 3},
+ };
+ ASSERT_EQ(IPC_V2_RING_OK, rx.TryReadv(remaining_iov, 4, read, batch_end));
+ EXPECT_EQ(source.size() - 31, read);
+ EXPECT_TRUE(batch_end);
+ EXPECT_EQ(source, output);
+ EXPECT_EQ(2u, read_pos);
+ EXPECT_EQ(0u, read_offset);
+ EXPECT_EQ(1u, __atomic_load_n(&tail, __ATOMIC_ACQUIRE));
+
+ std::string oversized(IpcV2Slot::PAYLOAD_SIZE * 4, 'x');
+ struct iovec oversized_iov = {&oversized[0], oversized.size()};
+ EXPECT_EQ(IPC_V2_RING_TOO_LARGE,
+ tx.TryWritev(&oversized_iov, 1, written));
+ EXPECT_EQ(0u, written);
+
+ char values[4] = {'a', 'b', 'c', 'd'};
+ for (int i = 0; i < 3; ++i) {
+ struct iovec item = {&values[i], 1};
+ ASSERT_EQ(IPC_V2_RING_OK, tx.TryWritev(&item, 1, written));
+ }
+ struct iovec fourth = {&values[3], 1};
+ EXPECT_EQ(IPC_V2_RING_RETRY, tx.TryWritev(&fourth, 1, written));
+ EXPECT_EQ(0u, written);
+ char result[4] = {};
+ struct iovec one = {&result[0], 1};
+ ASSERT_EQ(IPC_V2_RING_OK, rx.TryReadv(&one, 1, read, batch_end));
+ ASSERT_TRUE(batch_end);
+ ASSERT_EQ(IPC_V2_RING_OK, tx.TryWritev(&fourth, 1, written));
+ for (int i = 1; i < 4; ++i) {
+ struct iovec destination = {&result[i], 1};
+ ASSERT_EQ(IPC_V2_RING_OK,
+ rx.TryReadv(&destination, 1, read, batch_end));
+ ASSERT_TRUE(batch_end);
+ }
+ EXPECT_EQ(0, memcmp(values, result, sizeof(values)));
+}
+
+TEST(IpcV2RingTest, rejects_invalid_iovec_without_publication) {
+ using namespace brpc::ubring;
+ IpcV2Slot slots[2] = {};
+ uint32_t tail = 0;
+ uint32_t write_pos = 0;
+ uint32_t read_pos = 0;
+ uint32_t read_offset = 0;
+ IpcV2TxView tx(slots, &tail, 2, &write_pos);
+ IpcV2RxView rx(slots, &tail, 2, &read_pos, &read_offset);
+ size_t bytes = 99;
+ bool batch_end = true;
+ char value = 'x';
+ struct iovec one_byte = {&value, 1};
+ EXPECT_EQ(IPC_V2_RING_INVALID, tx.TryWritev(&one_byte, 1, bytes));
+ EXPECT_EQ(IPC_V2_RING_INVALID,
+ rx.TryReadv(&one_byte, 1, bytes, batch_end));
+ ASSERT_TRUE(IpcV2RxView::InitializeShared(slots, &tail, 2));
+ EXPECT_EQ(IPC_V2_RING_INVALID, tx.TryWritev(nullptr, 1, bytes));
+ EXPECT_EQ(IPC_V2_RING_INVALID, tx.TryWritev(nullptr, -1, bytes));
+ EXPECT_EQ(IPC_V2_RING_INVALID,
+ rx.TryReadv(nullptr, 1, bytes, batch_end));
+ struct iovec null_data = {nullptr, 1};
+ EXPECT_EQ(IPC_V2_RING_INVALID, tx.TryWritev(&null_data, 1, bytes));
+ EXPECT_EQ(IPC_V2_RING_OK, tx.TryWritev(nullptr, 0, bytes));
+ EXPECT_EQ(0u, bytes);
+ EXPECT_EQ(IPC_V2_RING_OK, rx.TryReadv(nullptr, 0, bytes, batch_end));
+ EXPECT_EQ(0u, bytes);
+ EXPECT_FALSE(batch_end);
+ EXPECT_EQ(IPC_V2_SLOT_EMPTY,
+ __atomic_load_n(&slots[0].header.state, __ATOMIC_ACQUIRE));
+}
+
+TEST(IpcV2RingTest, concurrent_writev_readv) {
+ using namespace brpc::ubring;
+ IpcV2Slot slots[8] = {};
+ uint32_t tail = 0;
+ uint32_t write_pos = 0;
+ uint32_t read_pos = 0;
+ uint32_t read_offset = 0;
+ ASSERT_TRUE(IpcV2RxView::InitializeShared(slots, &tail, 8));
+ IpcV2TxView tx(slots, &tail, 8, &write_pos);
+ IpcV2RxView rx(slots, &tail, 8, &read_pos, &read_offset);
+ const uint32_t rounds = 200;
+ const auto deadline = std::chrono::steady_clock::now() +
+ std::chrono::seconds(10);
+ std::atomic<bool> stop(false);
+ std::string producer_error;
+ std::string consumer_error;
+ const auto message_size = [](uint32_t sequence) {
+ return 1 + (sequence * 53) % (IpcV2Slot::PAYLOAD_SIZE * 2);
+ };
+ const auto fill = [](uint32_t sequence, char* data, size_t length) {
+ for (size_t i = 0; i < length; ++i) {
+ data[i] = static_cast<char>((sequence * 11 + i * 37) & 0xff);
+ }
+ };
+ std::thread producer([&] {
+ std::string input(IpcV2Slot::PAYLOAD_SIZE * 2, '\0');
+ for (uint32_t sequence = 0; sequence < rounds; ++sequence) {
+ const size_t length = message_size(sequence);
+ fill(sequence, &input[0], length);
+ const size_t first = length < 3 ? length : 3;
+ struct iovec parts[3] = {
+ {&input[0], first},
+ {nullptr, 0},
+ {&input[first], length - first},
+ };
+ for (;;) {
+ size_t written = 0;
+ const IpcV2RingResult result = tx.TryWritev(parts, 3, written);
+ if (result == IPC_V2_RING_OK && written == length) {
+ break;
+ }
+ if (result != IPC_V2_RING_RETRY ||
+ stop.load(std::memory_order_relaxed) ||
+ std::chrono::steady_clock::now() >= deadline) {
+ producer_error = "valid writev failed or timed out";
+ stop.store(true, std::memory_order_relaxed);
+ return;
+ }
+ std::this_thread::yield();
+ }
+ }
+ });
+
+ std::string output(IpcV2Slot::PAYLOAD_SIZE * 2, '\0');
+ std::string expected(IpcV2Slot::PAYLOAD_SIZE * 2, '\0');
+ for (uint32_t sequence = 0; sequence < rounds; ++sequence) {
+ const size_t length = message_size(sequence);
+ fill(sequence, &expected[0], length);
+ size_t received = 0;
+ bool ended = false;
+ while (!ended && !stop.load(std::memory_order_relaxed)) {
+ const size_t chunk = std::min(size_t(113), length - received);
+ struct iovec destination = {&output[received], chunk};
+ size_t read = 0;
+ const IpcV2RingResult result =
+ rx.TryReadv(&destination, 1, read, ended);
+ if (result == IPC_V2_RING_RETRY) {
+ if (std::chrono::steady_clock::now() >= deadline) {
+ consumer_error = "consumer timed out";
+ break;
+ }
+ std::this_thread::yield();
+ continue;
+ }
+ if (result != IPC_V2_RING_OK || read == 0 || read > chunk) {
+ consumer_error = "valid readv failed";
+ break;
+ }
+ received += read;
+ }
+ if (!consumer_error.empty() || stop.load(std::memory_order_relaxed)) {
+ break;
+ }
+ if (received != length ||
+ memcmp(output.data(), expected.data(), length) != 0) {
+ consumer_error = "batch length, order or content mismatch";
+ break;
+ }
+ }
+ stop.store(true, std::memory_order_relaxed);
+ producer.join();
+ EXPECT_TRUE(producer_error.empty()) << producer_error;
+ EXPECT_TRUE(consumer_error.empty()) << consumer_error;
+}
+
+TEST(IpcV2QueueMappingTest, reuses_existing_tail_and_preserves_neighbors) {
+ using namespace brpc::ubring;
+ struct alignas(64) QueueStorage {
+ IpcV2Slot slots[2];
+ uint8_t trailing[31];
+ } storage;
+ memset(&storage, 0xa5, sizeof(storage));
+ UbrDataStatusQMsg status = {};
+ status.tail = 99;
+ status.timeout = 12345;
+ status.heart_beat = 7;
+
+ IpcV2QueueView queue;
+ ASSERT_TRUE(InitializeIpcV2ReceiveQueue(
+ reinterpret_cast<uint8_t*>(storage.slots), sizeof(storage),
+ &status.tail, queue));
+ EXPECT_EQ(storage.slots, queue.slots);
+ EXPECT_EQ(&status.tail, queue.tail);
+ EXPECT_EQ(2u, queue.capacity);
+ EXPECT_EQ(1u, __atomic_load_n(&status.tail, __ATOMIC_ACQUIRE));
+ EXPECT_EQ(12345u, status.timeout);
+ EXPECT_EQ(7u, status.heart_beat);
+ EXPECT_EQ(IPC_V2_SLOT_EMPTY,
+ __atomic_load_n(&storage.slots[0].header.state,
+ __ATOMIC_ACQUIRE));
+ EXPECT_EQ(IPC_V2_SLOT_EMPTY,
+ __atomic_load_n(&storage.slots[1].header.state,
+ __ATOMIC_ACQUIRE));
+ const uint8_t* trailing = reinterpret_cast<const uint8_t*>(&storage) +
+ sizeof(storage.slots);
+ for (size_t i = 0; i < sizeof(storage) - sizeof(storage.slots); ++i) {
+ EXPECT_EQ(0xa5, trailing[i]);
+ }
+}
+
+TEST(IpcV2QueueMappingTest, rejects_invalid_layout_without_modification) {
+ using namespace brpc::ubring;
+ IpcV2Slot slots[2] = {};
+ uint32_t tail = 77;
+ IpcV2QueueView queue;
+ EXPECT_FALSE(MapIpcV2Queue(nullptr, sizeof(slots), &tail, queue));
+ EXPECT_FALSE(MapIpcV2Queue(
+ reinterpret_cast<uint8_t*>(slots), sizeof(IpcV2Slot), &tail, queue));
+ EXPECT_FALSE(MapIpcV2Queue(
+ reinterpret_cast<uint8_t*>(slots) + 1, sizeof(slots) - 1,
+ &tail, queue));
+ alignas(uint32_t) uint8_t tail_storage[sizeof(uint32_t) + 1] = {};
+ EXPECT_FALSE(MapIpcV2Queue(
+ reinterpret_cast<uint8_t*>(slots), sizeof(slots),
+ reinterpret_cast<uint32_t*>(tail_storage + 1), queue));
+ EXPECT_EQ(nullptr, queue.slots);
+ EXPECT_EQ(nullptr, queue.tail);
+ EXPECT_EQ(0u, queue.capacity);
+ EXPECT_EQ(77u, tail);
+}
+
+TEST(IpcV2FormatPreparationTest, prepares_layout_without_activation) {
+ using namespace brpc::ubring;
+ IpcV2Slot local_slots[2] = {};
+ IpcV2Slot remote_slots[3] = {};
+ UbrDataStatusQMsg local_status = {};
+ UbrDataStatusQMsg remote_status = {};
+ local_status.tail = 91;
+ local_status.timeout = 111;
+ local_status.heart_beat = 3;
+ remote_status.tail = 92;
+ remote_status.timeout = 222;
+ remote_status.heart_beat = 4;
+ UbrTrx trx = {};
+ trx.ubr_tx.remote_data_q.addr =
+ reinterpret_cast<uint8_t*>(remote_slots);
+ trx.ubr_tx.remote_data_q.len = sizeof(remote_slots);
+ trx.ubr_tx.local_data_status_q.addr =
+ reinterpret_cast<uint8_t*>(&local_status);
+ trx.ubr_rx.local_data_q.addr = reinterpret_cast<uint8_t*>(local_slots);
+ trx.ubr_rx.local_data_q.len = sizeof(local_slots);
+ trx.ubr_rx.remote_data_status_q.addr =
+ reinterpret_cast<uint8_t*>(&remote_status);
+
+ UBRing ring;
+ ring.SetTrxForTest(&trx);
+ ASSERT_EQ(0, ring.UbrPrepareIpcV2Format());
+ EXPECT_EQ(UBR_DATA_FORMAT_IPC_V2, ring.data_format());
+ EXPECT_EQ(3u, trx.ubr_tx.capacity);
+ EXPECT_EQ(0u, trx.ubr_tx.write_pos);
+ EXPECT_EQ(2u, trx.ubr_rx.capacity);
+ EXPECT_EQ(0u, trx.ubr_rx.read_pos);
+ EXPECT_EQ(0u, trx.ubr_rx.ipc_v2_read_offset);
+ EXPECT_EQ(UBR_STATE_NONE, trx.ubr_tx.trx_state);
+ EXPECT_EQ(UBR_STATE_NONE, trx.ubr_rx.trx_state);
+ EXPECT_EQ(0, trx.timer_fd);
+ EXPECT_EQ(0, trx.hb_timer_fd);
+ EXPECT_EQ(91u, local_status.tail);
+ EXPECT_EQ(1u, __atomic_load_n(&remote_status.tail, __ATOMIC_ACQUIRE));
+ EXPECT_EQ(111u, local_status.timeout);
+ EXPECT_EQ(3u, local_status.heart_beat);
+ EXPECT_EQ(222u, remote_status.timeout);
+ EXPECT_EQ(4u, remote_status.heart_beat);
+ EXPECT_EQ(IPC_V2_SLOT_EMPTY,
+ __atomic_load_n(&local_slots[0].header.state,
+ __ATOMIC_ACQUIRE));
+ EXPECT_EQ(IPC_V2_SLOT_UNINITIALIZED,
+ __atomic_load_n(&remote_slots[0].header.state,
+ __ATOMIC_ACQUIRE));
+ EXPECT_EQ(-1, ring.UbrPrepareIpcV2Format());
+}
+
+TEST(IpcV2FormatPreparationTest, failure_keeps_connection_unselected) {
+ using namespace brpc::ubring;
+ alignas(64) uint8_t local_storage[sizeof(IpcV2Slot) * 2 + 1] = {};
+ IpcV2Slot remote_slots[2] = {};
+ UbrDataStatusQMsg local_status = {};
+ UbrDataStatusQMsg remote_status = {};
+ local_status.tail = 71;
+ remote_status.tail = 72;
+ UbrTrx trx = {};
+ trx.ubr_tx.remote_data_q.addr =
+ reinterpret_cast<uint8_t*>(remote_slots);
+ trx.ubr_tx.remote_data_q.len = sizeof(remote_slots);
+ trx.ubr_tx.local_data_status_q.addr =
+ reinterpret_cast<uint8_t*>(&local_status);
+ trx.ubr_rx.local_data_q.addr = local_storage + 1;
+ trx.ubr_rx.local_data_q.len = sizeof(local_storage) - 1;
+ trx.ubr_rx.remote_data_status_q.addr =
+ reinterpret_cast<uint8_t*>(&remote_status);
+
+ UBRing ring;
+ ring.SetTrxForTest(&trx);
+ EXPECT_EQ(-1, ring.UbrPrepareIpcV2Format());
+ EXPECT_EQ(UBR_DATA_FORMAT_NONE, ring.data_format());
+ EXPECT_EQ(0u, trx.ubr_tx.capacity);
+ EXPECT_EQ(0u, trx.ubr_rx.capacity);
+ EXPECT_EQ(71u, local_status.tail);
+ EXPECT_EQ(72u, remote_status.tail);
+}
+
+struct IpcV2UbrTestContext {
+ IpcV2UbrTestContext() {
+ memset(local_slots, 0, sizeof(local_slots));
+ memset(remote_slots, 0, sizeof(remote_slots));
+ memset(&local_status, 0, sizeof(local_status));
+ memset(&remote_status, 0, sizeof(remote_status));
+ memset(&local_rx_event, 0, sizeof(local_rx_event));
+ memset(&local_tx_event, 0, sizeof(local_tx_event));
+ memset(&remote_rx_event, 0, sizeof(remote_rx_event));
+ trx.local_shm.addr = reinterpret_cast<uint8_t*>(local_slots);
+ trx.ubr_tx.remote_data_q.addr =
+ reinterpret_cast<uint8_t*>(remote_slots);
+ trx.ubr_tx.remote_data_q.len = sizeof(remote_slots);
+ trx.ubr_tx.local_data_status_q.addr =
+ reinterpret_cast<uint8_t*>(&local_status);
+ trx.ubr_tx.local_tx_event_q.addr =
+ reinterpret_cast<uint8_t*>(&local_tx_event);
+ trx.ubr_tx.remote_rx_event_q.addr =
+ reinterpret_cast<uint8_t*>(&remote_rx_event);
+ trx.ubr_rx.local_data_q.addr =
+ reinterpret_cast<uint8_t*>(local_slots);
+ trx.ubr_rx.local_data_q.len = sizeof(local_slots);
+ trx.ubr_rx.remote_data_status_q.addr =
+ reinterpret_cast<uint8_t*>(&remote_status);
+ trx.ubr_rx.local_rx_event_q.addr =
+ reinterpret_cast<uint8_t*>(&local_rx_event);
+ ring.SetTrxForTest(&trx);
+ }
+
+ bool PrepareAndConnect() {
+ if (ring.UbrPrepareIpcV2Format() != 0 ||
+ !brpc::ubring::IpcV2RxView::InitializeShared(
+ remote_slots, &local_status.tail, CAPACITY)) {
+ return false;
+ }
+ trx.ubr_tx.trx_state = brpc::ubring::UBR_STATE_CONNECTED;
+ trx.ubr_rx.trx_state = brpc::ubring::UBR_STATE_CONNECTED;
+ return true;
+ }
+
+ enum { CAPACITY = 4 };
+ brpc::ubring::IpcV2Slot local_slots[CAPACITY];
+ brpc::ubring::IpcV2Slot remote_slots[CAPACITY];
+ brpc::ubring::UbrDataStatusQMsg local_status;
+ brpc::ubring::UbrDataStatusQMsg remote_status;
+ brpc::ubring::UbrEventQMsg local_rx_event;
+ brpc::ubring::UbrEventQMsg local_tx_event;
+ brpc::ubring::UbrEventQMsg remote_rx_event;
+ brpc::ubring::UbrTrx trx{};
+ brpc::ubring::UBRing ring;
+};
+
+TEST(IpcV2UbrDispatchTest, writev_readv_and_notifications) {
+ using namespace brpc::ubring;
+ IpcV2UbrTestContext context;
+ ASSERT_TRUE(context.PrepareAndConnect());
+ EXPECT_EQ(UBRING_OK, context.ring.IsUbrTrxWriteable(0));
+
+ std::string source(IpcV2Slot::PAYLOAD_SIZE + 19, '\0');
+ for (size_t i = 0; i < source.size(); ++i) {
+ source[i] = static_cast<char>((i * 13 + 5) & 0xff);
+ }
+ struct iovec source_iov[2] = {
+ {&source[0], 17},
+ {&source[17], source.size() - 17},
+ };
+ ASSERT_EQ(static_cast<ssize_t>(source.size()),
+ context.ring.UbrTrxWritev(source_iov, 2));
+ EXPECT_EQ(1u, context.trx.ubr_tx.out_io_id);
+ EXPECT_EQ(1u, context.remote_rx_event.io_id);
+
+ uint32_t peer_read_pos = 0;
+ uint32_t peer_read_offset = 0;
+ IpcV2RxView peer_rx(
+ context.remote_slots, &context.local_status.tail,
+ IpcV2UbrTestContext::CAPACITY, &peer_read_pos, &peer_read_offset);
+ std::string peer_output(source.size(), '?');
+ struct iovec peer_destination = {&peer_output[0], peer_output.size()};
+ size_t peer_read = 0;
+ bool peer_batch_end = false;
+ ASSERT_EQ(IPC_V2_RING_OK,
+ peer_rx.TryReadv(&peer_destination, 1, peer_read,
+ peer_batch_end));
+ EXPECT_TRUE(peer_batch_end);
+ EXPECT_EQ(source, peer_output);
+
+ char empty_value = 0;
+ struct iovec empty_destination = {&empty_value, 1};
+ errno = 0;
+ EXPECT_EQ(-1, context.ring.UbrTrxReadv(&empty_destination, 1));
+ EXPECT_EQ(EAGAIN, errno);
+ EXPECT_EQ(MPA_MUXER_NOT_READY, context.ring.IsUbrTrxReadable(0));
+
+ uint32_t peer_write_pos = 0;
+ IpcV2TxView peer_tx(
+ context.local_slots, &context.remote_status.tail,
+ IpcV2UbrTestContext::CAPACITY, &peer_write_pos);
+ size_t peer_written = 0;
+ ASSERT_EQ(IPC_V2_RING_OK,
+ peer_tx.TryWritev(source_iov, 2, peer_written));
+ context.local_rx_event.io_id = 1;
+ EXPECT_EQ(UBRING_OK, context.ring.IsUbrTrxReadable(0));
+
+ std::string output(source.size(), '?');
+ struct iovec first = {&output[0], 31};
+ ASSERT_EQ(31, context.ring.UbrTrxReadv(&first, 1));
+ EXPECT_EQ(0u, context.trx.ubr_rx.read_pos);
+ EXPECT_EQ(31u, context.trx.ubr_rx.ipc_v2_read_offset);
+ struct iovec remaining = {&output[31], output.size() - 31};
+ ASSERT_EQ(static_cast<ssize_t>(output.size() - 31),
+ context.ring.UbrTrxReadv(&remaining, 1));
+ EXPECT_EQ(2u, context.trx.ubr_rx.read_pos);
+ EXPECT_EQ(0u, context.trx.ubr_rx.ipc_v2_read_offset);
+ EXPECT_EQ(source, output);
+}
+
+TEST(IpcV2UbrDispatchTest, backpressure_and_error_mapping) {
+ using namespace brpc::ubring;
+ IpcV2UbrTestContext context;
+ ASSERT_TRUE(context.PrepareAndConnect());
+ char values[4] = {'a', 'b', 'c', 'd'};
+ for (int i = 0; i < 3; ++i) {
+ struct iovec source = {&values[i], 1};
+ ASSERT_EQ(1, context.ring.UbrTrxWritev(&source, 1));
+ }
+ const uint64_t notified_io_id = context.remote_rx_event.io_id;
+ struct iovec fourth = {&values[3], 1};
+ EXPECT_EQ(UBRING_RETRY, context.ring.UbrTrxWritev(&fourth, 1));
+ EXPECT_EQ(notified_io_id, context.remote_rx_event.io_id);
+ EXPECT_EQ(MPA_MUXER_NOT_READY,
+ context.ring.IsUbrTrxWriteable(0));
+
+ std::string oversized(
+ IpcV2Slot::PAYLOAD_SIZE * IpcV2UbrTestContext::CAPACITY, 'x');
+ struct iovec too_large = {&oversized[0], oversized.size()};
+ errno = 0;
+ EXPECT_EQ(-1, context.ring.UbrTrxWritev(&too_large, 1));
+ EXPECT_EQ(EMSGSIZE, errno);
+ EXPECT_EQ(notified_io_id, context.remote_rx_event.io_id);
+
+ context.local_slots[0].header.payload_len =
+ IpcV2Slot::PAYLOAD_SIZE + 1;
+ context.local_slots[0].header.flags = IPC_V2_SLOT_EOF;
+ context.local_slots[0].header.reserved = 0;
+ __atomic_store_n(&context.local_slots[0].header.state,
+ IPC_V2_SLOT_READY, __ATOMIC_RELEASE);
+ char destination = 0;
+ struct iovec invalid = {&destination, 1};
+ EXPECT_EQ(UBRING_OK, context.ring.IsUbrTrxReadable(0));
+ errno = 0;
+ EXPECT_EQ(-1, context.ring.UbrTrxReadv(&invalid, 1));
+ EXPECT_EQ(EBADMSG, errno);
+
+ context.trx.data_format = UBR_DATA_FORMAT_NONE;
+ errno = 0;
+ EXPECT_EQ(UBRING_ERR, context.ring.UbrTrxWritev(&fourth, 1));
+ EXPECT_EQ(EPROTONOSUPPORT, errno);
+}
+
+TEST(IpcV2UbrDispatchTest, rejects_uninitialized_remote_ring) {
+ using namespace brpc::ubring;
+ IpcV2UbrTestContext context;
+ ASSERT_EQ(0, context.ring.UbrPrepareIpcV2Format());
+ context.trx.ubr_tx.trx_state = UBR_STATE_CONNECTED;
+ context.trx.ubr_rx.trx_state = UBR_STATE_CONNECTED;
+
+ EXPECT_EQ(UBRING_ERR, context.ring.IsUbrTrxWriteable(0));
+ char value = 'x';
+ struct iovec source = {&value, 1};
+ errno = 0;
+ EXPECT_EQ(-1, context.ring.UbrTrxWritev(&source, 1));
+ EXPECT_EQ(EBADMSG, errno);
+ EXPECT_EQ(0u, context.trx.ubr_tx.out_io_id);
+ EXPECT_EQ(0u, context.remote_rx_event.io_id);
+}
+
struct __attribute__((packed)) HelloMessageLayout {
uint16_t msg_len;
uint16_t hello_ver;
@@ -76,6 +618,12 @@ TEST(HelloFormatExtensionTest,
serialize_deserialize_roundtrip) {
decoded.Deserialize(buffer);
EXPECT_EQ(extension.extension_len, decoded.extension_len);
EXPECT_EQ(extension.format_id, decoded.format_id);
+
+ extension.format_id = brpc::ubring::UBR_DATA_FORMAT_IPC_V2;
+ extension.Serialize(buffer);
+ decoded.Deserialize(buffer);
+ EXPECT_EQ(extension.extension_len, decoded.extension_len);
+ EXPECT_EQ(extension.format_id, decoded.format_id);
}
TEST(HelloFormatExtensionTest, serialize_uses_network_byte_order) {
@@ -108,6 +656,41 @@ TEST(HelloFormatExtensionTest, deserialize_unknown_format)
{
EXPECT_EQ(0x1234, extension.format_id);
}
+TEST(HelloFormatExtensionTest, selects_format_for_local_backend) {
+ using namespace brpc::ubring;
+ EXPECT_EQ(UBR_DATA_FORMAT_IPC_V2,
+ PreferredDataFormatForShmType(SHM_TYPE_IPC));
+ EXPECT_EQ(UBR_DATA_FORMAT_LEGACY_64,
+ PreferredDataFormatForShmType(SHM_TYPE_UBS));
+ EXPECT_EQ(UBR_DATA_FORMAT_NONE,
+ PreferredDataFormatForShmType(SHM_TYPE_UB));
+ EXPECT_EQ(UBR_DATA_FORMAT_NONE,
+ PreferredDataFormatForShmType(SHM_TYPE_UNSUPPORT));
+}
+
+TEST(HelloFormatExtensionTest, selects_only_matching_supported_format) {
+ using namespace brpc::ubring;
+ EXPECT_EQ(UBR_DATA_FORMAT_IPC_V2,
+ SelectDataFormat(UBR_DATA_FORMAT_IPC_V2,
+ UBR_DATA_FORMAT_IPC_V2));
+ EXPECT_EQ(UBR_DATA_FORMAT_LEGACY_64,
+ SelectDataFormat(UBR_DATA_FORMAT_LEGACY_64,
+ UBR_DATA_FORMAT_LEGACY_64));
+ EXPECT_EQ(UBR_DATA_FORMAT_NONE,
+ SelectDataFormat(UBR_DATA_FORMAT_IPC_V2,
+ UBR_DATA_FORMAT_LEGACY_64));
+ EXPECT_EQ(UBR_DATA_FORMAT_NONE,
+ SelectDataFormat(UBR_DATA_FORMAT_LEGACY_64,
+ UBR_DATA_FORMAT_IPC_V2));
+ EXPECT_EQ(UBR_DATA_FORMAT_NONE,
+ SelectDataFormat(UBR_DATA_FORMAT_IPC_V2, 0x1234));
+ EXPECT_EQ(UBR_DATA_FORMAT_NONE,
+ SelectDataFormat(UBR_DATA_FORMAT_NONE,
+ UBR_DATA_FORMAT_NONE));
+ EXPECT_EQ(UBR_DATA_FORMAT_NONE,
+ SelectDataFormat(static_cast<UbrDataFormat>(0x1234), 0x1234));
+}
+
TEST_F(HelloMessageTest, serialize_deserialize_roundtrip) {
msg.msg_len = 64;
msg.hello_ver = 2;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
