You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@brpc.apache.org by ww...@apache.org on 2026/09/02 05:54:03 UTC
(brpc) branch master updated: Fix RDMA server core dump on concurrent access to Socket::_read_buf (#3505)
This is an automated email from the ASF dual-hosted git repository.
wwbmmm 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 2ffdeca8 Fix RDMA server core dump on concurrent access to Socket::_read_buf (#3505)
2ffdeca8 is described below
commit 2ffdeca8b7885761479e26acb2ca15e5b36563b2
Author: Bright Chen <ch...@foxmail.com>
AuthorDate: Wed Sep 2 13:53:58 2026 +0800
Fix RDMA server core dump on concurrent access to Socket::_read_buf (#3505)
---
src/brpc/input_message_base.h | 1 +
src/brpc/input_messenger.cpp | 233 +----
src/brpc/input_messenger.h | 14 +-
src/brpc/input_messenger_processor.cpp | 293 +++++++
src/brpc/input_messenger_processor.h | 123 +++
src/brpc/rdma/rdma_endpoint.cpp | 290 +++++--
src/brpc/rdma/rdma_endpoint.h | 32 +-
src/brpc/rdma/rdma_handshake_server.cpp | 9 +-
src/brpc/rdma_transport.cpp | 7 +-
src/brpc/socket.cpp | 32 +-
src/brpc/socket.h | 35 +-
src/brpc/ubshm/ub_endpoint.cpp | 26 +-
test/brpc_rdma_unittest.cpp | 1411 ++++++++++++++++++-------------
13 files changed, 1577 insertions(+), 929 deletions(-)
diff --git a/src/brpc/input_message_base.h b/src/brpc/input_message_base.h
index b117eb99..0d441ee0 100644
--- a/src/brpc/input_message_base.h
+++ b/src/brpc/input_message_base.h
@@ -53,6 +53,7 @@ protected:
private:
friend class InputMessenger;
+friend class InputMessengerProcessor;
friend void* ProcessInputMessage(void*);
friend class Stream;
friend class Transport;
diff --git a/src/brpc/input_messenger.cpp b/src/brpc/input_messenger.cpp
index 26aac063..50dc1821 100644
--- a/src/brpc/input_messenger.cpp
+++ b/src/brpc/input_messenger.cpp
@@ -74,98 +74,6 @@ DEFINE_int32(socket_tcp_user_timeout_ms, -1,
DECLARE_bool(usercode_in_pthread);
DECLARE_bool(usercode_in_coroutine);
-DECLARE_uint64(max_body_size);
-
-const size_t MSG_SIZE_WINDOW = 10; // Take last so many message into stat.
-const size_t MIN_ONCE_READ = 4096;
-const size_t MAX_ONCE_READ = 524288;
-
-ParseResult InputMessenger::CutInputMessage(
- Socket* m, size_t* index, bool read_eof) {
- const int preferred = m->preferred_index();
- const int max_index = (int)_max_index.load(butil::memory_order_acquire);
- // Try preferred handler first. The preferred_index is set on last
- // selection or by client.
- if (preferred >= 0 && preferred <= max_index
- && _handlers[preferred].parse != nullptr) {
- int cur_index = preferred;
- do {
- ParseResult result =
- _handlers[cur_index].parse(&m->_read_buf, m, read_eof, _handlers[cur_index].arg);
- if (result.is_ok() ||
- result.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
- m->set_preferred_index(cur_index);
- *index = cur_index;
- return result;
- } else if (result.error() != PARSE_ERROR_TRY_OTHERS) {
- // Critical error, return directly.
- LOG_IF(ERROR, result.error() == PARSE_ERROR_TOO_BIG_DATA)
- << "A message from " << m->remote_side()
- << "(protocol=" << _handlers[cur_index].name
- << ") is bigger than " << FLAGS_max_body_size
- << " bytes, the connection will be closed."
- " Set max_body_size to allow bigger messages";
- return result;
- }
-
- if (m->CreatedByConnect()) {
- if((ProtocolType)cur_index == PROTOCOL_BAIDU_STD && cur_index == preferred) {
- // baidu_std may fall to streaming_rpc.
- cur_index = (int)PROTOCOL_STREAMING_RPC;
- continue;
- } else if((ProtocolType)cur_index == PROTOCOL_STREAMING_RPC && cur_index == preferred) {
- // streaming_rpc may fall to baidu_std.
- cur_index = (int)PROTOCOL_BAIDU_STD;
- continue;
- } else {
- // The protocol is fixed at client-side, no need to try others.
- LOG(ERROR) << "Fail to parse response from " << m->remote_side()
- << " by " << _handlers[preferred].name
- << " at client-side";
- return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
- }
- } else {
- // Try other protocols.
- break;
- }
- } while (true);
- // Clear context before trying next protocol which probably has
- // an incompatible context with the current one.
- if (m->parsing_context()) {
- m->reset_parsing_context(nullptr);
- }
- m->set_preferred_index(-1);
- }
- for (int i = 0; i <= max_index; ++i) {
- if (i == preferred || _handlers[i].parse == nullptr) {
- // Don't try preferred handler(already tried) or invalid handler
- continue;
- }
- ParseResult result = _handlers[i].parse(&m->_read_buf, m, read_eof, _handlers[i].arg);
- if (result.is_ok() ||
- result.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
- m->set_preferred_index(i);
- *index = i;
- return result;
- } else if (result.error() != PARSE_ERROR_TRY_OTHERS) {
- // Critical error, return directly.
- LOG_IF(ERROR, result.error() == PARSE_ERROR_TOO_BIG_DATA)
- << "A message from " << m->remote_side()
- << "(protocol=" << _handlers[i].name
- << ") is bigger than " << FLAGS_max_body_size
- << " bytes, the connection will be closed."
- " Set max_body_size to allow bigger messages";
- return result;
- }
- // Clear context before trying next protocol which definitely has
- // an incompatible context with the current one.
- if (m->parsing_context()) {
- m->reset_parsing_context(nullptr);
- }
- // Try other protocols.
- }
- return MakeParseError(PARSE_ERROR_TRY_OTHERS);
-}
void* ProcessInputMessage(void* void_arg) {
InputMessageBase* msg = static_cast<InputMessageBase*>(void_arg);
@@ -192,124 +100,6 @@ void InputMessageClosure::reset(InputMessageBase* m) {
_msg = m;
}
-int InputMessenger::ProcessNewMessage(
- Socket* m, ssize_t bytes, bool read_eof,
- const uint64_t received_us, const uint64_t base_realtime,
- InputMessageClosure& last_msg) {
- m->AddInputBytes(bytes);
-
- // Avoid this socket to be closed due to idle_timeout_s
- m->_last_readtime_us.store(received_us, butil::memory_order_relaxed);
-
- size_t last_size = m->_read_buf.length();
- int num_bthread_created = 0;
- while (1) {
- size_t index = 8888;
- ParseResult pr = CutInputMessage(m, &index, read_eof);
- if (!pr.is_ok()) {
- if (pr.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
- // incomplete message, re-read.
- // However, some buffer may have been consumed
- // under protocols like HTTP. Record this size
- m->_last_msg_size += (last_size - m->_read_buf.length());
- break;
- } else if (pr.error() == PARSE_ERROR_TRY_OTHERS) {
- LOG(WARNING)
- << "Close " << *m << " due to unknown message: "
- << butil::ToPrintable(m->_read_buf);
- m->SetFailed(EINVAL, "Close %s due to unknown message",
- m->description().c_str());
- return -1;
- } else {
- LOG(WARNING) << "Close " << *m << ": " << pr.error_str();
- m->SetFailed(EINVAL, "Close %s: %s",
- m->description().c_str(), pr.error_str());
- return -1;
- }
- }
-
- m->AddInputMessages(1);
- // Calculate average size of messages
- const size_t cur_size = m->_read_buf.length();
- if (cur_size == 0) {
- // _read_buf is consumed, it's good timing to return blocks
- // cached internally back to TLS, otherwise the memory is not
- // reused until next message arrives which is quite uncertain
- // in situations that most connections are idle.
- m->_read_buf.return_cached_blocks();
- }
- m->_last_msg_size += (last_size - cur_size);
- last_size = cur_size;
- const size_t old_avg = m->_avg_msg_size;
- if (old_avg != 0) {
- m->_avg_msg_size = (old_avg * (MSG_SIZE_WINDOW - 1) + m->_last_msg_size)
- / MSG_SIZE_WINDOW;
- } else {
- m->_avg_msg_size = m->_last_msg_size;
- }
- m->_last_msg_size = 0;
-
- if (pr.message() == nullptr) { // the Process() step can be skipped.
- continue;
- }
- pr.message()->_received_us = received_us;
- pr.message()->_base_real_us = base_realtime;
-
- // This unique_ptr prevents msg to be lost before transfering
- // ownership to last_msg
- DestroyingPtr<InputMessageBase> msg(pr.message());
- m->_transport->QueueMessage(last_msg, &num_bthread_created, false);
- if (_handlers[index].process == nullptr) {
- LOG(ERROR) << "process of index=" << index << " is NULL";
- continue;
- }
- m->ReAddress(&msg->_socket);
- m->PostponeEOF();
- msg->_process = _handlers[index].process;
- msg->_arg = _handlers[index].arg;
-
- if (_handlers[index].verify != nullptr) {
- int auth_error = 0;
- if (0 == m->FightAuthentication(&auth_error)) {
- // Get the right to authenticate
- if (_handlers[index].verify(msg.get())) {
- m->SetAuthentication(0);
- } else {
- m->SetAuthentication(ERPCAUTH);
- LOG(WARNING) << "Fail to authenticate " << *m;
- m->SetFailed(ERPCAUTH, "Fail to authenticate %s",
- m->description().c_str());
- return -1;
- }
- } else {
- LOG_IF(FATAL, auth_error != 0) <<
- "Impossible! Socket should have been "
- "destroyed when authentication failed";
- }
- }
- if (!m->is_read_progressive()) {
- // Transfer ownership to last_msg
- last_msg.reset(msg.release());
- } else {
- last_msg.reset(msg.release());
- m->_transport->QueueMessage(last_msg, &num_bthread_created, false);
- bthread_flush();
- num_bthread_created = 0;
- }
- }
- // In RDMA polling mode, all messages must be executed in a new bthread and
- // not in the bthread where the polling bthread is located, because the
- // method for processing messages may call synchronization primitives,
- // causing the polling bthread to be scheduled out.
- if (m->_socket_mode == SOCKET_MODE_RDMA || m->_socket_mode == SOCKET_MODE_UBRING) {
- m->_transport->QueueMessage(last_msg, &num_bthread_created, true);
- }
- if (num_bthread_created) {
- bthread_flush();
- }
- return 0;
-}
-
void InputMessenger::OnNewMessages(Socket* m) {
// Notes:
// - If the socket has only one message, the message will be parsed and
@@ -321,7 +111,7 @@ void InputMessenger::OnNewMessages(Socket* m) {
// is batched(notice the BTHREAD_NOSIGNAL and bthread_flush).
// - Verify will always be called in this bthread at most once and before
// any process.
- InputMessenger* messenger = static_cast<InputMessenger*>(m->user());
+ InputMessengerProcessor& processor = m->fd_input_processor();
int progress = Socket::PROGRESS_INIT;
// Notice that all *return* no matter successful or not will run last
@@ -333,21 +123,14 @@ void InputMessenger::OnNewMessages(Socket* m) {
const int64_t received_us = butil::cpuwide_time_us();
const int64_t base_realtime = butil::gettimeofday_us() - received_us;
- // Calculate bytes to be read.
- size_t once_read = m->_avg_msg_size * 16;
- if (once_read < MIN_ONCE_READ) {
- once_read = MIN_ONCE_READ;
- } else if (once_read > MAX_ONCE_READ) {
- once_read = MAX_ONCE_READ;
- }
-
// Read.
- const ssize_t nr = m->DoRead(once_read);
+ const ssize_t nr = m->DoRead(&processor.read_buf(),
+ processor.OnceReadSize());
if (nr <= 0) {
if (0 == nr) {
// Set `read_eof' flag and proceed to feed EOF into `Protocol'
- // (implied by m->_read_buf.empty), which may produce a new
- // `InputMessageBase' under some protocols such as HTTP
+ // (implied by an empty processor.read_buf()), which may produce
+ // a new `InputMessageBase' under some protocols such as HTTP
LOG_IF(WARNING, FLAGS_log_connection_close) << *m << " was closed by remote side";
read_eof = true;
} else if (errno != EAGAIN) {
@@ -366,10 +149,10 @@ void InputMessenger::OnNewMessages(Socket* m) {
}
}
- if (messenger->ProcessNewMessage(m, nr, read_eof, received_us,
- base_realtime, last_msg) < 0) {
+ if (processor.ProcessNewMessage(nr, read_eof, received_us,
+ base_realtime, last_msg) < 0) {
return;
- }
+ }
}
if (read_eof) {
diff --git a/src/brpc/input_messenger.h b/src/brpc/input_messenger.h
index d056263e..8a4b53d8 100644
--- a/src/brpc/input_messenger.h
+++ b/src/brpc/input_messenger.h
@@ -23,6 +23,7 @@
#include "brpc/socket.h" // SocketId, SocketUser
#include "brpc/parse_result.h" // ParseResult
#include "brpc/input_message_base.h" // InputMessageBase
+#include "brpc/input_messenger_processor.h" // InputMessengerProcessor
namespace brpc {
@@ -94,11 +95,11 @@ private:
// Process messages from connections.
// `Message' corresponds to a client's request or a server's response.
class InputMessenger : public SocketUser {
-friend class Socket;
friend class TcpTransport;
friend class RdmaTransport;
friend class rdma::RdmaEndpoint;
friend class ubring::UBShmEndpoint;
+friend class InputMessengerProcessor;
public:
explicit InputMessenger(size_t capacity = 128);
~InputMessenger();
@@ -137,17 +138,6 @@ protected:
private:
- // Find a valid scissor from `handlers' to cut off `header' and `payload'
- // from m->read_buf, save index of the scissor into `index'.
- ParseResult CutInputMessage(Socket* m, size_t* index, bool read_eof);
-
- // Process a new message just received in OnNewMessages
- // Return value >= 0 means success
- int ProcessNewMessage(
- Socket* m, ssize_t bytes, bool read_eof,
- const uint64_t received_us, const uint64_t base_realtime,
- InputMessageClosure& last_msg);
-
// User-supplied scissors and handlers.
// the index of handler is exactly the same as the protocol
InputMessageHandler* _handlers;
diff --git a/src/brpc/input_messenger_processor.cpp b/src/brpc/input_messenger_processor.cpp
new file mode 100644
index 00000000..f23a423c
--- /dev/null
+++ b/src/brpc/input_messenger_processor.cpp
@@ -0,0 +1,293 @@
+// 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 "butil/logging.h"
+#include "butil/binary_printer.h"
+#include "bthread/unstable.h"
+#include "brpc/options.pb.h"
+#include "brpc/transport.h"
+#include "brpc/input_messenger_processor.h"
+#include "brpc/input_messenger.h"
+
+namespace brpc {
+
+DECLARE_uint64(max_body_size);
+
+const size_t MSG_SIZE_WINDOW = 10; // Take last so many message into stat.
+const size_t MIN_ONCE_READ = 4096;
+const size_t MAX_ONCE_READ = 524288;
+
+static const char* StreamTypeName(InputMessengerProcessor::StreamType type) {
+ switch (type) {
+ case InputMessengerProcessor::STREAM_NONE: return "none";
+ case InputMessengerProcessor::STREAM_TCP_FD: return "tcp_fd";
+ case InputMessengerProcessor::STREAM_RDMA_QP: return "rdma_qp";
+ }
+ return "unknown";
+}
+
+InputMessengerProcessor::ParsingStreamGuard::ParsingStreamGuard(Socket* socket, StreamType type)
+ : _socket(socket) {
+ CHECK(socket != nullptr)
+ << "Parsing through a processor that was never Init()ed, socket is NULL";
+ CHECK_NE(STREAM_NONE, type)
+ << "Parsing through a processor that was never Init()ed, " << *socket;
+ CHECK_EQ(STREAM_NONE, socket->parsing_stream_type())
+ << "Two input streams of " << *socket << " are parsing at the same time: "
+ << StreamTypeName(socket->parsing_stream_type()) << " and "
+ << StreamTypeName(type);
+ socket->set_parsing_stream_type(type);
+}
+
+InputMessengerProcessor::ParsingStreamGuard::~ParsingStreamGuard() {
+ _socket->set_parsing_stream_type(STREAM_NONE);
+}
+
+ParseResult InputMessengerProcessor::CutInputMessage(InputMessenger* messenger,
+ size_t* index, bool read_eof) {
+ ParsingStreamGuard parsing_stream_guard(_socket, _stream_type);
+ InputMessageHandler* handlers = messenger->_handlers;
+ int preferred = _socket->preferred_index();
+ int max_index = (int)messenger->_max_index.load(butil::memory_order_acquire);
+ // Try preferred handler first. The preferred_index is set on last
+ // selection or by client.
+ if (preferred >= 0 && preferred <= max_index
+ && handlers[preferred].parse != nullptr) {
+ int cur_index = preferred;
+ do {
+ ParseResult result =
+ handlers[cur_index].parse(&_read_buf, _socket, read_eof,
+ handlers[cur_index].arg);
+ if (result.is_ok() ||
+ result.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
+ _socket->set_preferred_index(cur_index);
+ *index = cur_index;
+ return result;
+ } else if (result.error() != PARSE_ERROR_TRY_OTHERS) {
+ // Critical error, return directly.
+ LOG_IF(ERROR, result.error() == PARSE_ERROR_TOO_BIG_DATA)
+ << "A message from " << _socket->remote_side()
+ << "(protocol=" << handlers[cur_index].name
+ << ") is bigger than " << FLAGS_max_body_size
+ << " bytes, the connection will be closed."
+ " Set max_body_size to allow bigger messages";
+ return result;
+ }
+
+ if (_socket->CreatedByConnect()) {
+ if((ProtocolType)cur_index == PROTOCOL_BAIDU_STD && cur_index == preferred) {
+ // baidu_std may fall to streaming_rpc.
+ cur_index = (int)PROTOCOL_STREAMING_RPC;
+ continue;
+ } else if((ProtocolType)cur_index == PROTOCOL_STREAMING_RPC &&
+ cur_index == preferred) {
+ // streaming_rpc may fall to baidu_std.
+ cur_index = (int)PROTOCOL_BAIDU_STD;
+ continue;
+ } else {
+ // The protocol is fixed at client-side, no need to try others.
+ LOG(ERROR) << "Fail to parse response from " << _socket->remote_side()
+ << " by " << handlers[preferred].name
+ << " at client-side";
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
+ }
+ } else {
+ // Try other protocols.
+ //
+ // A handler may lean on this: returning PARSE_ERROR_NOT_ENOUGH_DATA
+ // above keeps `preferred_index' pinned on it, TRY_OTHERS here gives
+ // it up. RdmaEndpoint::ExecuteServerHandshake() pins itself that way
+ // to keep the last handshake read from being taken for a protocol
+ // detection. See the tail of its phase 2 before changing what
+ // happens to `preferred_index' here.
+ break;
+ }
+ } while (true);
+ // Clear context before trying next protocol which probably has
+ // an incompatible context with the current one.
+ if (_socket->parsing_context()) {
+ _socket->reset_parsing_context(nullptr);
+ }
+ _socket->set_preferred_index(-1);
+ }
+ for (int i = 0; i <= max_index; ++i) {
+ if (i == preferred || handlers[i].parse == nullptr) {
+ // Don't try preferred handler(already tried) or invalid handler
+ continue;
+ }
+ ParseResult result = handlers[i].parse(&_read_buf, _socket, read_eof, handlers[i].arg);
+ if (result.is_ok() ||
+ result.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
+ _socket->set_preferred_index(i);
+ *index = i;
+ return result;
+ } else if (result.error() != PARSE_ERROR_TRY_OTHERS) {
+ // Critical error, return directly.
+ LOG_IF(ERROR, result.error() == PARSE_ERROR_TOO_BIG_DATA)
+ << "A message from " << _socket->remote_side()
+ << "(protocol=" << handlers[i].name
+ << ") is bigger than " << FLAGS_max_body_size
+ << " bytes, the connection will be closed."
+ " Set max_body_size to allow bigger messages";
+ return result;
+ }
+ // Clear context before trying next protocol which definitely has
+ // an incompatible context with the current one.
+ if (_socket->parsing_context()) {
+ _socket->reset_parsing_context(nullptr);
+ }
+ // Try other protocols.
+ }
+ return MakeParseError(PARSE_ERROR_TRY_OTHERS);
+}
+
+size_t InputMessengerProcessor::OnceReadSize() const {
+ size_t once_read = _avg_msg_size * 16;
+ if (once_read < MIN_ONCE_READ) {
+ once_read = MIN_ONCE_READ;
+ } else if (once_read > MAX_ONCE_READ) {
+ once_read = MAX_ONCE_READ;
+ }
+ return once_read;
+}
+
+void InputMessengerProcessor::Reset() {
+ _read_buf.clear();
+ _last_msg_size = 0;
+ _avg_msg_size = 0;
+}
+
+int InputMessengerProcessor::ProcessNewMessage(ssize_t bytes, bool read_eof,
+ uint64_t received_us,
+ uint64_t base_realtime,
+ InputMessageClosure& last_msg) {
+ auto messenger = static_cast<InputMessenger*>(_socket->user());
+ const InputMessageHandler* handlers = messenger->_handlers;
+ _socket->AddInputBytes(bytes);
+
+ // Avoid this socket to be closed due to idle_timeout_s
+ _socket->_last_readtime_us.store(received_us, butil::memory_order_relaxed);
+
+ size_t last_size = _read_buf.length();
+ int num_bthread_created = 0;
+ while (true) {
+ size_t index = 8888;
+ ParseResult pr = CutInputMessage(messenger, &index, read_eof);
+ if (!pr.is_ok()) {
+ if (pr.error() == PARSE_ERROR_NOT_ENOUGH_DATA) {
+ // incomplete message, re-read.
+ // However, some buffer may have been consumed
+ // under protocols like HTTP. Record this size
+ _last_msg_size += (last_size - _read_buf.length());
+ break;
+ } else if (pr.error() == PARSE_ERROR_TRY_OTHERS) {
+ LOG(WARNING) << "Close " << *_socket << " due to unknown message: "
+ << butil::ToPrintable(_read_buf);
+ _socket->SetFailed(EINVAL, "Close %s due to unknown message",
+ _socket->description().c_str());
+ return -1;
+ } else {
+ LOG(WARNING) << "Close " << *_socket << ": " << pr.error_str();
+ _socket->SetFailed(EINVAL, "Close %s: %s",
+ _socket->description().c_str(), pr.error_str());
+ return -1;
+ }
+ }
+
+ _socket->AddInputMessages(1);
+ // Calculate average size of messages
+ const size_t cur_size = _read_buf.length();
+ if (cur_size == 0) {
+ // _read_buf is consumed, it's good timing to return blocks
+ // cached internally back to TLS, otherwise the memory is not
+ // reused until next message arrives which is quite uncertain
+ // in situations that most connections are idle.
+ _read_buf.return_cached_blocks();
+ }
+ _last_msg_size += (last_size - cur_size);
+ last_size = cur_size;
+ const size_t old_avg = _avg_msg_size;
+ if (old_avg != 0) {
+ _avg_msg_size = (old_avg * (MSG_SIZE_WINDOW - 1) + _last_msg_size) / MSG_SIZE_WINDOW;
+ } else {
+ _avg_msg_size = _last_msg_size;
+ }
+ _last_msg_size = 0;
+
+ if (pr.message() == nullptr) { // the Process() step can be skipped.
+ continue;
+ }
+ pr.message()->_received_us = received_us;
+ pr.message()->_base_real_us = base_realtime;
+
+ // This unique_ptr prevents msg to be lost before transfering
+ // ownership to last_msg
+ DestroyingPtr<InputMessageBase> msg(pr.message());
+ _socket->_transport->QueueMessage(last_msg, &num_bthread_created, false);
+ if (handlers[index].process == nullptr) {
+ LOG(ERROR) << "process of index=" << index << " is NULL";
+ continue;
+ }
+ _socket->ReAddress(&msg->_socket);
+ _socket->PostponeEOF();
+ msg->_process = handlers[index].process;
+ msg->_arg = handlers[index].arg;
+
+ if (handlers[index].verify != nullptr) {
+ int auth_error = 0;
+ if (0 == _socket->FightAuthentication(&auth_error)) {
+ // Get the right to authenticate
+ if (handlers[index].verify(msg.get())) {
+ _socket->SetAuthentication(0);
+ } else {
+ _socket->SetAuthentication(ERPCAUTH);
+ LOG(WARNING) << "Fail to authenticate " << *_socket;
+ _socket->SetFailed(ERPCAUTH, "Fail to authenticate %s",
+ _socket->description().c_str());
+ return -1;
+ }
+ } else {
+ LOG_IF(FATAL, auth_error != 0) <<
+ "Impossible! Socket should have been "
+ "destroyed when authentication failed";
+ }
+ }
+ if (!_socket->is_read_progressive()) {
+ // Transfer ownership to last_msg
+ last_msg.reset(msg.release());
+ } else {
+ last_msg.reset(msg.release());
+ _socket->_transport->QueueMessage(last_msg, &num_bthread_created, false);
+ bthread_flush();
+ num_bthread_created = 0;
+ }
+ }
+ // In RDMA polling mode, all messages must be executed in a new bthread and
+ // not in the bthread where the polling bthread is located, because the
+ // method for processing messages may call synchronization primitives,
+ // causing the polling bthread to be scheduled out.
+ if (_socket->_socket_mode == SOCKET_MODE_RDMA ||
+ _socket->_socket_mode == SOCKET_MODE_UBRING) {
+ _socket->_transport->QueueMessage(last_msg, &num_bthread_created, true);
+ }
+ if (num_bthread_created) {
+ bthread_flush();
+ }
+ return 0;
+}
+
+} // namespace brpc
diff --git a/src/brpc/input_messenger_processor.h b/src/brpc/input_messenger_processor.h
new file mode 100644
index 00000000..7b94087d
--- /dev/null
+++ b/src/brpc/input_messenger_processor.h
@@ -0,0 +1,123 @@
+// 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_INPUT_MESSENGER_PROCESSOR_H_
+#define BRPC_INPUT_MESSENGER_PROCESSOR_H_
+
+#include "butil/iobuf.h" // butil::IOPortal
+#include "butil/logging.h" // DCHECK
+#include "butil/macros.h" // DISALLOW_COPY_AND_ASSIGN
+#include "brpc/parse_result.h" // ParseResult
+
+namespace brpc {
+
+class Socket;
+class InputMessenger;
+class InputMessageClosure;
+
+// The state of one input stream: the data read but not cut off yet, and the
+// message-size statistics used to size the next read.
+class InputMessengerProcessor {
+public:
+ // Which stream of a Socket a processor drains.
+ enum StreamType {
+ STREAM_NONE,
+ STREAM_TCP_FD,
+ STREAM_RDMA_QP,
+ };
+
+ InputMessengerProcessor()
+ : _socket(nullptr), _stream_type(STREAM_NONE)
+ , _last_msg_size(0), _avg_msg_size(0) {}
+
+ DISALLOW_COPY_AND_ASSIGN(InputMessengerProcessor);
+
+ void Init(Socket* socket, StreamType type) {
+ DCHECK(socket != nullptr);
+ DCHECK_NE(STREAM_NONE, type) << "STREAM_NONE is not a stream";
+ _socket = socket;
+ _stream_type = type;
+ }
+
+ // Cut off and process all complete messages in read_buf(), which just
+ // grew by `bytes` bytes.
+ // Returns 0 on success, -1 otherwise.
+ int ProcessNewMessage(ssize_t bytes, bool read_eof,
+ uint64_t received_us,
+ uint64_t base_realtime,
+ InputMessageClosure& last_msg);
+
+ // Data read from the stream but not cut off yet. Only the bthread
+ // draining this particular stream may touch it.
+ butil::IOPortal& read_buf() { return _read_buf; }
+ const butil::IOPortal& read_buf() const { return _read_buf; }
+
+ // How many bytes to ask for on the next read, derived from the sizes of
+ // the messages seen recently.
+ size_t OnceReadSize() const;
+
+ uint32_t avg_msg_size() const { return _avg_msg_size; }
+
+ // Drop buffered data and reset the statistics.
+ void Reset();
+
+ // Reset the message-size statistics only, keeping buffered data.
+ void ResetMsgSizeStats() { _last_msg_size = 0; _avg_msg_size = 0; }
+
+private:
+
+ // Publishes on the Socket which stream the parse callbacks are cutting
+ // from and clears it when they return, which is what makes
+ // Socket::parsing_stream_type() mean "the stream being parsed right now".
+ //
+ // Entering also DCHECKs that no other stream of that Socket is parsing --
+ // the one-stream-at-a-time rule above, checked instead of assumed -- and
+ // that the processor was Init()ed.
+ //
+ // Nested, with Socket::set_parsing_stream_type() private to the enclosing
+ // class, so nothing else can publish and nothing can forget to unpublish.
+ class ParsingStreamGuard {
+ public:
+ ParsingStreamGuard(Socket* socket, StreamType type);
+ DISALLOW_COPY_AND_ASSIGN(ParsingStreamGuard);
+ ~ParsingStreamGuard();
+ private:
+ Socket* _socket;
+ };
+
+ // Find a valid scissor among messenger's handlers to cut off one message
+ // from `_read_buf`, save the index of the scissor into `index`.
+ ParseResult CutInputMessage(InputMessenger* messenger, size_t* index, bool read_eof);
+
+ // The Socket this stream belongs to. Not owned.
+ Socket* _socket;
+
+ // Which of that Socket's streams this is, never STREAM_NONE once Init()ed.
+ StreamType _stream_type;
+
+ butil::IOPortal _read_buf;
+
+ // Size of current incomplete message, set to 0 on complete.
+ uint32_t _last_msg_size;
+ // Average message size of last #MSG_SIZE_WINDOW messages (roughly)
+ uint32_t _avg_msg_size;
+};
+
+} // namespace brpc
+
+#endif // BRPC_INPUT_MESSENGER_PROCESSOR_H_
diff --git a/src/brpc/rdma/rdma_endpoint.cpp b/src/brpc/rdma/rdma_endpoint.cpp
index fd70fb5c..12145a31 100644
--- a/src/brpc/rdma/rdma_endpoint.cpp
+++ b/src/brpc/rdma/rdma_endpoint.cpp
@@ -147,6 +147,7 @@ RdmaEndpoint::RdmaEndpoint(Socket* s)
_rq_size = MAX_QP_SIZE;
}
_read_butex = bthread::butex_create_checked<butil::atomic<int> >();
+ _input_processor.Init(s, InputMessengerProcessor::STREAM_RDMA_QP);
}
RdmaEndpoint::~RdmaEndpoint() {
@@ -167,6 +168,7 @@ void RdmaEndpoint::Reset() {
_sbuf.clear();
_rbuf.clear();
_rbuf_data.clear();
+ _input_processor.Reset();
_remote_recv_block_size = 0;
_accumulated_ack = 0;
_unsolicited = 0;
@@ -220,8 +222,16 @@ void RdmaConnect::Run() {
_done(errno, _data);
}
-void RdmaEndpoint::OnNewDataFromTcp(Socket* m) {
- auto* rdma_transport = static_cast<RdmaTransport*>(m->_transport.get());
+void RdmaEndpoint::OnNewDataFromTcp(Socket* s) {
+ if (s->CreatedByConnect()) {
+ OnNewDataFromTcpAtClient(s);
+ } else {
+ OnNewDataFromTcpAtServer(s);
+ }
+}
+
+void RdmaEndpoint::OnNewDataFromTcpAtClient(Socket* s) {
+ auto* rdma_transport = static_cast<RdmaTransport*>(s->_transport.get());
RdmaEndpoint* ep = rdma_transport->GetRdmaEp();
CHECK(ep != nullptr);
@@ -237,36 +247,88 @@ void RdmaEndpoint::OnNewDataFromTcp(Socket* m) {
ep->_read_butex->fetch_add(1, butil::memory_order_release);
bthread::butex_wake(ep->_read_butex);
} else if (state == FALLBACK_TCP){ // handshake finishes
- InputMessenger::OnNewMessages(m);
+ InputMessenger::OnNewMessages(s);
return;
} else if (state == ESTABLISHED) {
- uint8_t tmp;
- ssize_t nr = read(ep->_socket->fd(), &tmp, 1);
- if (nr == 0) {
- ep->_socket->SetEOF();
- return;
- }
- if (nr > 0) {
- LOG(WARNING) << "Read unexpected data from " << ep->_socket;
- ep->_socket->SetFailed(EPROTO, "Read unexpected data from %s",
- ep->_socket->description().c_str());
+ if (!ep->HandleTcpEventAfterEstablished()) {
return;
}
+ }
+ if (!s->MoreReadEvents(&progress)) {
+ break;
+ }
+ }
+}
- if (errno != EAGAIN) {
+void RdmaEndpoint::OnNewDataFromTcpAtServer(Socket* s) {
+ auto* rdma_transport = static_cast<RdmaTransport*>(s->_transport.get());
+ RdmaEndpoint* ep = rdma_transport->GetRdmaEp();
+ CHECK(ep != nullptr);
+
+ int progress = Socket::PROGRESS_INIT;
+ while (true) {
+ if (s->Failed()) {
+ return;
+ }
+
+ // Pair with the release stores of ESTABLISHED / FALLBACK_TCP.
+ if (ep->_state.load(butil::memory_order_acquire) != ESTABLISHED) {
+ InputMessenger::OnNewMessages(s);
+ // That call may have just finished the handshake and turned RDMA
+ // on. Start consuming CQ events here rather than inside the parse
+ // callback: by now OnNewMessages is done with the Socket's
+ // `parsing_context` / `preferred_index`, so the QP stream can take
+ // them over without ever overlapping with the fd stream. This is
+ // the ordering StartCqEvents() asks for.
+ if (!s->Failed() &&
+ ep->_state.load(butil::memory_order_acquire) == ESTABLISHED &&
+ ep->StartCqEvents() < 0) {
const int saved_errno = errno;
- PLOG(WARNING) << "Fail to read from " << ep->_socket;
- ep->_socket->SetFailed(saved_errno, "Fail to read from %s: %s",
- ep->_socket->description().c_str(),
- berror(saved_errno));
+ PLOG(WARNING) << "Fail to start cq events on " << *s;
+ ep->_state.store(FAILED, butil::memory_order_relaxed);
+ s->SetFailed(saved_errno, "Fail to start cq events on %s: %s",
+ s->description().c_str(), berror(saved_errno));
}
+ return;
}
- if (!m->MoreReadEvents(&progress)) {
+ // RDMA carries the RPCs now, so the fd is watched for EOF only and must
+ // not be parsed: `preferred_index' / `parsing_context' live on the Socket
+ // and the QP stream is driving them (https://github.com/apache/brpc/issues/3479).
+ if (!ep->HandleTcpEventAfterEstablished()) {
+ return;
+ }
+ if (!s->MoreReadEvents(&progress)) {
break;
}
}
}
+bool RdmaEndpoint::HandleTcpEventAfterEstablished() {
+ uint8_t tmp;
+ ssize_t nr = read(_socket->fd(), &tmp, 1);
+ if (nr == 0) {
+ _socket->SetEOF();
+ return false;
+ }
+ if (nr > 0) {
+ LOG(WARNING) << "Read unexpected data from " << *_socket;
+ _socket->SetFailed(EPROTO, "Read unexpected data from %s",
+ _socket->description().c_str());
+ return false;
+ }
+
+ if (errno != EAGAIN) {
+ const int saved_errno = errno;
+ PLOG(WARNING) << "Fail to read from " << *_socket;
+ _socket->SetFailed(saved_errno, "Fail to read from %s: %s",
+ _socket->description().c_str(),
+ berror(saved_errno));
+ // The socket is dead now, so do not come back for another read of it.
+ return false;
+ }
+ return true;
+}
+
static const int WAIT_TIMEOUT_MS = 50;
// Drive an EAGAIN-aware read loop to completion (exactly `len` bytes).
@@ -500,7 +562,16 @@ void* RdmaEndpoint::ProcessHandshakeAtClient(void* arg) {
}
if (rdma_transport->_rdma_state == RdmaTransport::RDMA_ON) {
- ep->_state.store(ESTABLISHED, butil::memory_order_relaxed);
+ ep->_state.store(ESTABLISHED, butil::memory_order_release);
+ // The handshake is over, so the QP stream may start parsing now.
+ if (ep->StartCqEvents() < 0) {
+ const int saved_errno = errno;
+ PLOG(WARNING) << "Fail to start cq events on " << s->description();
+ s->SetFailed(saved_errno, "Fail to complete rdma handshake from %s: %s",
+ s->description().c_str(), berror(saved_errno));
+ ep->_state.store(FAILED, butil::memory_order_relaxed);
+ return nullptr;
+ }
LOG_IF(INFO, FLAGS_rdma_trace_verbose)
<< "Client handshake ends (use rdma v" << ep->_handshake_version
<< ") on " << s->description();
@@ -535,6 +606,36 @@ ParseResult RdmaEndpoint::ExecuteServerHandshake(butil::IOBuf* source, Socket* s
RdmaEndpoint* ep = rdma_transport->_rdma_ep;
CHECK(ep != nullptr);
+ const State state = ep->_state.load(butil::memory_order_acquire);
+ if (state >= ESTABLISHED) {
+ // The handshake is over (ESTABLISHED / FALLBACK_TCP / FAILED). Data
+ // arriving now belongs to a real protocol, yet CutInputMessage() still
+ // reaches us.
+ if (state == ESTABLISHED &&
+ s->parsing_stream_type() == InputMessengerProcessor::STREAM_TCP_FD) {
+ // RDMA is on, so the fd is not an RPC channel any more and whatever
+ // shows up on it is a protocol error. Reached even though
+ // OnNewDataFromTcpAtServer() stops handing the fd to OnNewMessages()
+ // once RDMA is on, because the handshake completes inside OnNewMessages():
+ // that round keeps reading the fd until it goes quiet.
+ if (source->empty()) {
+ // Nothing to reject yet. Asking for more data keeps the pin, so
+ // the rest of this round comes back here rather than reaching a
+ // real protocol, and lets OnNewMessages() report EOF as usual.
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
+ }
+ LOG(WARNING) << "Unexpected " << source->size() << " bytes on the tcp "
+ "fd of an RDMA connection, drop connection: "
+ << s->description();
+ ep->_state.store(FAILED, butil::memory_order_relaxed);
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
+ }
+ // Anything else, for the real protocol to parse: the stream carried by
+ // the QP, or an fd that stayed a normal RPC stream because the handshake
+ // fell back or failed.
+ return MakeParseError(PARSE_ERROR_TRY_OTHERS);
+ }
+
if (s->parsing_context() == nullptr) {
// Phase 1: read the client hello, negotiate, reply server hello.
if (source->size() < HELLO_MAGIC_LEN) {
@@ -607,13 +708,6 @@ ParseResult RdmaEndpoint::ExecuteServerHandshake(butil::IOBuf* source, Socket* s
if (source->size() < HELLO_ACK_LEN) {
return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
}
- if (source->size() > HELLO_ACK_LEN) {
- LOG(WARNING) << "Too many bytes in handshake ACK, drop connection: "
- << s->description();
- ep->_state.store(FAILED, butil::memory_order_relaxed);
- s->reset_parsing_context(nullptr);
- return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
- }
uint32_t flags_be = 0;
CHECK_EQ(source->cutn(&flags_be, HELLO_ACK_LEN), HELLO_ACK_LEN);
@@ -636,13 +730,38 @@ ParseResult RdmaEndpoint::ExecuteServerHandshake(butil::IOBuf* source, Socket* s
return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
}
+ if (!source->empty()) {
+ // RDMA is on, so the TCP fd is no longer an RPC channel. Anything
+ // trailing the ACK on it can only be a protocol error. This catches what
+ // arrived in the same read as the ACK.
+ LOG(WARNING) << "Unexpected " << source->size() << " bytes after the "
+ "handshake ACK of an RDMA connection, drop connection: "
+ << s->description();
+ ep->_state.store(FAILED, butil::memory_order_relaxed);
+ s->reset_parsing_context(nullptr);
+ return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
+ }
+
LOG_IF(INFO, FLAGS_rdma_trace_verbose)
<< "Server handshake ends (use rdma v" << ep->_handshake_version
<< ") on " << s->description();
rdma_transport->_rdma_state = RdmaTransport::RDMA_ON;
- ep->_state.store(ESTABLISHED, butil::memory_order_relaxed);
+ ep->_state.store(ESTABLISHED, butil::memory_order_release);
s->reset_parsing_context(nullptr);
- return MakeParseError(PARSE_ERROR_TRY_OTHERS);
+
+ // Two things are deliberately not done here.
+ //
+ // The CQ events are not started: this runs inside CutInputMessage, which
+ // keeps touching `preferred_index` / `parsing_context` after we return,
+ // and PollCq would race it for those. OnNewDataFromTcpAtServer() starts
+ // them once that is over.
+ //
+ // TRY_OTHERS is not returned: it would hand `preferred_index` to the real
+ // protocol, and the remaining reads of this OnNewMessages() round would
+ // parse the fd as an RPC stream although RDMA has just taken over. Asking
+ // for more data keeps this handler pinned, so those reads come back to the
+ // guard at the top of this function and are rejected there.
+ return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
}
bool RdmaEndpoint::IsWritable() const {
@@ -921,11 +1040,12 @@ ssize_t RdmaEndpoint::HandleCompletion(ibv_wc& wc) {
zerocopy = false;
}
CHECK_NE(_state.load(butil::memory_order_relaxed), FALLBACK_TCP);
+ butil::IOPortal& read_buf = _input_processor.read_buf();
if (zerocopy) {
- _rbuf[_rq_received].cutn(&_socket->_read_buf, wc.byte_len);
+ _rbuf[_rq_received].cutn(&read_buf, wc.byte_len);
} else {
// Copy data when the receive data is really small
- _socket->_read_buf.append(_rbuf_data[_rq_received], wc.byte_len);
+ read_buf.append(_rbuf_data[_rq_received], wc.byte_len);
}
}
if (0 != (wc.wc_flags & IBV_WC_WITH_IMM) && wc.imm_data > 0) {
@@ -1111,12 +1231,12 @@ int RdmaEndpoint::DoAllocateResources() {
g_rdma_resource_list = g_rdma_resource_list->next;
}
}
- if (!_resource) {
+ if (_resource == nullptr) {
_resource = AllocateQpCq(_sq_size, _rq_size);
} else {
_resource->next = nullptr;
}
- if (!_resource) {
+ if (_resource == nullptr) {
return -1;
}
@@ -1127,25 +1247,6 @@ int RdmaEndpoint::DoAllocateResources() {
if (0 != ReqNotifyCq(false, false)) {
return -1;
}
-
- SocketOptions options;
- options.user = this;
- options.keytable_pool = _socket->_keytable_pool;
- options.fd = _resource->comp_channel->fd;
- options.on_edge_triggered_events = PollCq;
- if (Socket::Create(options, &_cq_sid) < 0) {
- PLOG(WARNING) << "Fail to create socket for cq";
- return -1;
- }
- } else {
- 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";
- return -1;
- }
- PollerAddCqSid();
}
_sbuf.resize(_sq_size - RESERVED_WR_NUM);
@@ -1164,6 +1265,47 @@ int RdmaEndpoint::DoAllocateResources() {
return 0;
}
+int RdmaEndpoint::StartCqEvents() {
+ if (InputMessengerProcessor::STREAM_NONE != _socket->parsing_stream_type()) {
+ LOG(WARNING) << "StartCqEvents() called while " << *_socket << " is parsing";
+ errno = ERDMA;
+ return -1;
+ }
+
+ if (_cq_sid != INVALID_SOCKET_ID) {
+ // Already started.
+ return 0;
+ }
+ if (_resource == nullptr) {
+ if (BAIDU_UNLIKELY(g_skip_rdma_init)) {
+ // For UT: AllocateResources() succeeds without allocating anything.
+ return 0;
+ }
+
+ LOG(WARNING) << "No RDMA resource to start CQ events on, " << *_socket;
+ errno = ERDMA;
+ return -1;
+ }
+
+ SocketOptions options;
+ options.user = this;
+ options.keytable_pool = _socket->_keytable_pool;
+ if (!FLAGS_rdma_use_polling) {
+ options.fd = _resource->comp_channel->fd;
+ options.on_edge_triggered_events = PollCq;
+ }
+ if (Socket::Create(options, &_cq_sid) < 0) {
+ PLOG(WARNING) << "Fail to create socket for cq";
+ return -1;
+ }
+
+ if (FLAGS_rdma_use_polling) {
+ PollerAddCqSid();
+ }
+
+ return 0;
+}
+
int RdmaEndpoint::BringUpQp(const ParsedHello& remote, bool is_server) {
if (BAIDU_UNLIKELY(g_skip_rdma_init)) {
// For UT
@@ -1301,7 +1443,7 @@ static int DrainCq(ibv_cq* cq) {
}
void RdmaEndpoint::DeallocateResources() {
- if (!_resource) {
+ if (_resource == nullptr) {
return;
}
if (FLAGS_rdma_use_polling) {
@@ -1328,7 +1470,7 @@ void RdmaEndpoint::DeallocateResources() {
bool remove_consumer = true;
_reclaim:
if (!move_to_rdma_resource_list) {
- if (nullptr != _resource->qp) {
+ if (_resource->qp != nullptr) {
int err = IbvDestroyQp(_resource->qp);
LOG_IF(WARNING, 0 != err) << "Fail to destroy QP: " << berror(err);
_resource->qp = nullptr;
@@ -1338,15 +1480,16 @@ _reclaim:
DeallocateCq(_resource->send_cq);
DeallocateCq(_resource->recv_cq);
- if (nullptr != _resource->comp_channel) {
- // Destroy send_comp_channel will destroy this fd,
- // so that we should remove it from epoll fd first
- int fd = _resource->comp_channel->fd;
- GetGlobalEventDispatcher(fd, _socket->_io_event.bthread_tag()).RemoveConsumer(fd);
- remove_consumer = false;
+ if (_resource->comp_channel != nullptr) {
+ if (_cq_sid != INVALID_SOCKET_ID) {
+ // Destroy send_comp_channel will destroy this fd,
+ // so that we should remove it from epoll fd first
+ int fd = _resource->comp_channel->fd;
+ GetGlobalEventDispatcher(fd, _socket->_io_event.bthread_tag()).RemoveConsumer(fd);
+ remove_consumer = false;
+ }
int err = IbvDestroyCompChannel(_resource->comp_channel);
LOG_IF(WARNING, 0 != err) << "Fail to destroy CQ channel: " << berror(err);
-
}
_resource->polling_cq = nullptr;
@@ -1478,6 +1621,7 @@ void RdmaEndpoint::PollCq(Socket* m) {
}
auto* rdma_transport = static_cast<RdmaTransport*>(s->_transport.get());
CHECK(ep == rdma_transport->_rdma_ep);
+ CHECK_GE(ep->_state.load(butil::memory_order_acquire), ESTABLISHED);
bool send = false;
ibv_cq* cq = ep->_resource->recv_cq;
@@ -1595,9 +1739,8 @@ void RdmaEndpoint::PollCq(Socket* m) {
// Otherwise it may call too many bthread_flush to affect performance.
const int64_t received_us = butil::cpuwide_time_us();
const int64_t base_realtime = butil::gettimeofday_us() - received_us;
- InputMessenger* messenger = static_cast<InputMessenger*>(s->user());
- if (messenger->ProcessNewMessage(
- s.get(), bytes, false, received_us, base_realtime, last_msg) < 0) {
+ if (ep->_input_processor.ProcessNewMessage(bytes, false, received_us,
+ base_realtime, last_msg) < 0) {
return;
}
}
@@ -1638,7 +1781,8 @@ void RdmaEndpoint::DebugInfo(std::ostream& os, butil::StringPiece connector) con
<< connector << "rdma_unacked_rq_wr=" << _new_rq_wrs.load(butil::memory_order_relaxed)
<< connector << "rdma_received_ack=" << _accumulated_ack
<< connector << "rdma_unsolicited_sent=" << _unsolicited
- << connector << "rdma_unsignaled_sq_wr=" << _sq_unsignaled;
+ << connector << "rdma_unsignaled_sq_wr=" << _sq_unsignaled
+ << connector << "rdma_read_buf=" << _input_processor.read_buf().size();
}
int RdmaEndpoint::GlobalInitialize() {
@@ -1778,23 +1922,27 @@ void RdmaEndpoint::PollingModeRelease(bthread_tag_t tag) {
}
void RdmaEndpoint::PollerAddCqSid() {
+ if (_cq_sid == INVALID_SOCKET_ID) {
+ return;
+ }
+
auto index = butil::fmix32(_cq_sid) % FLAGS_rdma_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, CqSidOp::ADD});
- }
+ poller.op_queue.Enqueue(CqSidOp{_cq_sid, CqSidOp::ADD});
}
void RdmaEndpoint::PollerRemoveCqSid() {
+ if (INVALID_SOCKET_ID == _cq_sid) {
+ return;
+ }
+
auto index = butil::fmix32(_cq_sid) % FLAGS_rdma_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, CqSidOp::REMOVE});
- }
+ poller.op_queue.Enqueue(CqSidOp{_cq_sid, CqSidOp::REMOVE});
}
} // namespace rdma
diff --git a/src/brpc/rdma/rdma_endpoint.h b/src/brpc/rdma/rdma_endpoint.h
index 388e31d7..6d6ac391 100644
--- a/src/brpc/rdma/rdma_endpoint.h
+++ b/src/brpc/rdma/rdma_endpoint.h
@@ -129,7 +129,6 @@ public:
butil::StringPiece connector = "\n") const;
// Callback when there is new epollin event on TCP fd.
- // Only used by client-side RDMA sockets.
static void OnNewDataFromTcp(Socket* m);
// Real handshake for RDMA-mode sockets.
@@ -164,6 +163,11 @@ private:
// Process handshake at the client
static void* ProcessHandshakeAtClient(void* arg);
+ static void OnNewDataFromTcpAtClient(Socket* m);
+ static void OnNewDataFromTcpAtServer(Socket* m);
+
+ bool HandleTcpEventAfterEstablished();
+
// Allocate resources. On failure the endpoint is left with no RDMA
// resource attached, so that the handshake can safely fall back to TCP.
// Return 0 if success, -1 if failed and errno set
@@ -177,6 +181,27 @@ private:
// Release resources
void DeallocateResources();
+ // Create the Socket wrapping the CQ (and register it with the poller in
+ // polling mode), which is what makes CQ events reachable and thus starts
+ // PollCq.
+ //
+ // Must not be called before the handshake has reached ESTABLISHED, nor
+ // from within the fd stream's parsing path: PollCq() parses the input
+ // stream carried by the QP, and the Socket's `parsing_context` and
+ // `preferred_index` belong to the fd stream until the handshake is over
+ // and CutInputMessage has returned. It keeps writing both after the
+ // handshake handler hands the stream back. Those two are per-Socket,
+ // so letting PollCq in early makes two streams parse through one context.
+ // The server therefore calls this from OnNewDataFromTcpAtServer(), after
+ // OnNewMessages() returns, not from ExecuteServerHandshake().
+ //
+ // No CQE is lost by deferring: BringUpQp() fills the RQ before the QP
+ // reaches RTS, both CQs are armed by DoAllocateResources(), and adding an
+ // already readable fd to an edge-triggered epoll reports it immediately.
+ //
+ // Return 0 if success, -1 if failed and errno set
+ int StartCqEvents();
+
// Send Imm data to the remote side
// Arguments:
// imm: imm data in the WR
@@ -268,7 +293,7 @@ private:
Socket* _socket;
// State of Handshake. FALLBACK_TCP publishes RdmaTransport::_rdma_state
- // with release ordering and is consumed by OnNewDataFromTcp with acquire
+ // with release ordering and is consumed by OnNewDataFromTcpAtClient with acquire
// ordering. Other state accesses do not publish data and use relaxed
// ordering.
butil::atomic<State> _state;
@@ -301,6 +326,9 @@ private:
uint16_t _sq_size;
uint16_t _rq_size;
+ // The input stream carried by the QP.
+ InputMessengerProcessor _input_processor;
+
// Act as sendbuf and recvbuf, but requires no memcpy
std::vector<butil::IOBuf> _sbuf;
std::vector<butil::IOBuf> _rbuf;
diff --git a/src/brpc/rdma/rdma_handshake_server.cpp b/src/brpc/rdma/rdma_handshake_server.cpp
index 6dfb0c91..1b0ca422 100644
--- a/src/brpc/rdma/rdma_handshake_server.cpp
+++ b/src/brpc/rdma/rdma_handshake_server.cpp
@@ -153,6 +153,11 @@ static int SendUnnegotiableHello(Socket* socket, int version) {
}
// Fallback handshake for connections that are NOT in RDMA mode.
+//
+// Unlike the RDMA-mode path, which turns handshake bytes away once its endpoint
+// has left the handshake (the state >= ESTABLISHED guard in
+// RdmaEndpoint::ExecuteServerHandshake), this one keeps no record of having
+// run. See the tail of phase 2.
static ParseResult FallbackServerHandshake(butil::IOBuf* source, Socket* socket) {
if (socket->parsing_context() == nullptr) {
if (source->size() < HELLO_MAGIC_LEN) {
@@ -192,8 +197,8 @@ static ParseResult FallbackServerHandshake(butil::IOBuf* source, Socket* socket)
return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
}
CHECK_EQ(source->pop_front(HELLO_ACK_LEN), HELLO_ACK_LEN);
- // Handshake done (downgraded to TCP); drop the context and let
- // InputMessenger parse the following real RPC.
+ // Handshake done.
+ // Drop the context and let InputMessenger parse the following real RPC.
socket->reset_parsing_context(nullptr);
return MakeParseError(PARSE_ERROR_TRY_OTHERS);
}
diff --git a/src/brpc/rdma_transport.cpp b/src/brpc/rdma_transport.cpp
index ee5151c3..a2037aa7 100644
--- a/src/brpc/rdma_transport.cpp
+++ b/src/brpc/rdma_transport.cpp
@@ -43,12 +43,7 @@ void RdmaTransport::Init(Socket *socket, const SocketOptions &options) {
_default_connect = options.app_connect;
_on_edge_trigger = options.on_edge_triggered_events;
if (options.need_on_edge_trigger && _on_edge_trigger == nullptr) {
- // Server-side RDMA sockets drive the handshake through the standard
- // InputMessenger path (ParseRdmaHandshake), so they use OnNewMessages
- // just like TCP sockets. Only client-side sockets, whose handshake
- // (ProcessHandshakeAtClient) is an active blocking bthread relying on
- // _read_butex woken by OnNewDataFromTcp, still need OnNewDataFromTcp.
- if (options.user == static_cast<SocketUser*>(get_client_side_messenger())) {
+ if (_rdma_ep != nullptr) {
_on_edge_trigger = rdma::RdmaEndpoint::OnNewDataFromTcp;
} else {
_on_edge_trigger = InputMessenger::OnNewMessages;
diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp
index becb0944..33901874 100644
--- a/src/brpc/socket.cpp
+++ b/src/brpc/socket.cpp
@@ -465,10 +465,9 @@ Socket::Socket(Forbidden f)
, _conn(nullptr)
, _preferred_index(-1)
, _hc_count(0)
- , _last_msg_size(0)
- , _avg_msg_size(0)
, _last_readtime_us(0)
, _parsing_context(nullptr)
+ , _parsing_stream_type(InputMessengerProcessor::STREAM_NONE)
, _correlation_id(0)
, _health_check_interval_s(-1)
, _is_hc_related_ref_held(false)
@@ -499,6 +498,7 @@ Socket::Socket(Forbidden f)
CreateVarsOnce();
pthread_mutex_init(&_id_wait_list_mutex, nullptr);
_epollout_butex = bthread::butex_create_checked<butil::atomic<int> >();
+ _fd_input_processor.Init(this, InputMessengerProcessor::STREAM_TCP_FD);
}
Socket::~Socket() {
@@ -572,8 +572,7 @@ void Socket::ReleaseAllFailedWriteRequests(Socket::WriteRequest* req) {
int Socket::ResetFileDescriptor(int fd) {
// Reset message sizes when fd is changed.
- _last_msg_size = 0;
- _avg_msg_size = 0;
+ _fd_input_processor.ResetMsgSizeStats();
// MUST store `_fd' before adding itself into epoll device to avoid
// race conditions with the callback function inside epoll
static butil::atomic<uint64_t> BAIDU_CACHELINE_ALIGNMENT fd_version(0);
@@ -751,7 +750,9 @@ int Socket::OnCreated(const SocketOptions& options) {
_app_connect = _transport->Connect();
_preferred_index = -1;
_hc_count = 0;
- CHECK(_read_buf.empty());
+ CHECK(_fd_input_processor.read_buf().empty());
+ _parsing_stream_type.store(InputMessengerProcessor::STREAM_NONE,
+ butil::memory_order_relaxed);
const int64_t cpuwide_now = butil::cpuwide_time_us();
_last_readtime_us.store(cpuwide_now, butil::memory_order_relaxed);
reset_parsing_context(options.initial_parsing_context);
@@ -861,7 +862,7 @@ void Socket::BeforeRecycled() {
}
_transport->Release();
reset_parsing_context(nullptr);
- _read_buf.clear();
+ _fd_input_processor.read_buf().clear();
_auth_flag_error.store(0, butil::memory_order_relaxed);
bthread_id_error(_auth_id, 0);
@@ -1023,9 +1024,11 @@ int Socket::WaitAndReset(int32_t expected_nref) {
// parsing_context is very likely to be associated with the fd,
// removing it is a safer choice and required by http2.
reset_parsing_context(nullptr);
- // Must clear _read_buf otehrwise even if the connections is recovered,
- // the kept old data is likely to make parsing fail.
- _read_buf.clear();
+ _parsing_stream_type.store(InputMessengerProcessor::STREAM_NONE,
+ butil::memory_order_relaxed);
+ // Must clear the read buffer otehrwise even if the connections is
+ // recovered, the kept old data is likely to make parsing fail.
+ _fd_input_processor.read_buf().clear();
_ninprocess.store(1, butil::memory_order_relaxed);
_auth_flag_error.store(0, butil::memory_order_relaxed);
bthread_id_error(_auth_id, 0);
@@ -2080,7 +2083,7 @@ int Socket::SSLHandshake(int fd, bool server_mode) {
}
}
-ssize_t Socket::DoRead(size_t size_hint) {
+ssize_t Socket::DoRead(butil::IOPortal* read_buf, size_t size_hint) {
if (ssl_state() == SSL_UNKNOWN) {
int error_code = 0;
_ssl_state = DetectSSLState(fd(), &error_code);
@@ -2114,7 +2117,7 @@ ssize_t Socket::DoRead(size_t size_hint) {
errno = ESSL;
return -1;
}
- return _read_buf.append_from_file_descriptor(fd(), size_hint);
+ return read_buf->append_from_file_descriptor(fd(), size_hint);
}
CHECK_EQ(SSL_CONNECTED, ssl_state());
@@ -2122,7 +2125,7 @@ ssize_t Socket::DoRead(size_t size_hint) {
ssize_t nr = 0;
{
BAIDU_SCOPED_LOCK(_ssl_session_mutex);
- nr = _read_buf.append_from_SSL_channel(_ssl_session, &ssl_error, size_hint);
+ nr = read_buf->append_from_SSL_channel(_ssl_session, &ssl_error, size_hint);
}
switch (ssl_error) {
case SSL_ERROR_NONE: // `nr' > 0
@@ -2382,11 +2385,12 @@ void Socket::DebugSocket(std::ostream& os, SocketId id) {
os << " (" << messenger->NameOfProtocol(preferred_index) << ')';
}
const int64_t cpuwide_now = butil::cpuwide_time_us();
+ InputMessengerProcessor& input_processor = ptr->fd_input_processor();
os << "\nhc_count=" << ptr->_hc_count
- << "\navg_input_msg_size=" << ptr->_avg_msg_size
+ << "\navg_input_msg_size=" << input_processor.avg_msg_size()
// NOTE: We're assuming that butil::IOBuf.size() is thread-safe, it is now
// however it's not guaranteed.
- << "\nread_buf=" << ptr->_read_buf.size()
+ << "\nread_buf=" << input_processor.read_buf().size()
<< "\nlast_read_to_now=" << cpuwide_now - ptr->_last_readtime_us << "us"
<< "\nlast_write_to_now=" << cpuwide_now - ptr->_last_writetime_us << "us"
<< "\novercrowded=" << ptr->_overcrowded;
diff --git a/src/brpc/socket.h b/src/brpc/socket.h
index bbde3ecd..6f6f52fb 100644
--- a/src/brpc/socket.h
+++ b/src/brpc/socket.h
@@ -43,6 +43,7 @@
#include "brpc/versioned_ref_with_id.h"
#include "brpc/health_check_option.h"
#include "brpc/socket_mode.h"
+#include "brpc/input_messenger_processor.h" // InputMessengerProcessor
namespace brpc {
namespace policy {
@@ -317,6 +318,7 @@ struct SocketOptions {
class BAIDU_CACHELINE_ALIGNMENT/*note*/ Socket : public VersionedRefWithId<Socket> {
friend class EventDispatcher;
friend class InputMessenger;
+friend class InputMessengerProcessor;
friend class Acceptor;
friend class ConnectionsService;
friend class SocketUser;
@@ -722,11 +724,24 @@ private:
// Returns 0 on success, -1 otherwise
int SSLHandshake(int fd, bool server_mode);
+ // The input stream carried by `_fd`.
+ InputMessengerProcessor& fd_input_processor() { return _fd_input_processor; }
+ void set_parsing_stream_type(InputMessengerProcessor::StreamType type) {
+ _parsing_stream_type.store(type, butil::memory_order_relaxed);
+ }
+
+ // Which of this Socket's input streams the parse callbacks running right
+ // now were called for, STREAM_NONE while none are. Only meaningful with a
+ // parse callback on the stack.
+ InputMessengerProcessor::StreamType parsing_stream_type() const {
+ return _parsing_stream_type.load(butil::memory_order_relaxed);
+ }
+
// Based upon whether the underlying channel is using SSL (if
// SSLState is SSL_UNKNOWN, try to detect at first), read data
- // using the corresponding method into `_read_buf'. Returns read
+ // using the corresponding method into `read_buf`. Returns read
// bytes on success, 0 on EOF, -1 otherwise and errno is set
- ssize_t DoRead(size_t size_hint);
+ ssize_t DoRead(butil::IOPortal* read_buf, size_t size_hint);
// Based upon whether the underlying channel is using SSL, write
// `req' using the corresponding method. Returns written bytes on
@@ -890,7 +905,7 @@ private:
IOEvent<Socket> _io_event;
- // last chosen index of the protocol as a heuristic value to avoid
+ // Last chosen index of the protocol as a heuristic value to avoid
// iterating all protocol handlers each time.
int _preferred_index;
@@ -898,13 +913,10 @@ private:
// socket is revived. Only set in HealthCheckTask::OnTriggeringTask()
int _hc_count;
- // Size of current incomplete message, set to 0 on complete.
- uint32_t _last_msg_size;
- // Average message size of last #MSG_SIZE_WINDOW messages (roughly)
- uint32_t _avg_msg_size;
-
- // Storing data read from `_fd' but cut-off yet.
- butil::IOPortal _read_buf;
+ // The input stream carried by `_fd`, holding the data read from it but not cut off
+ // yet. Only the bthread draining the fd (InputMessenger:: OnNewMessages) may touch
+ // it, see InputMessengerProcessor.
+ InputMessengerProcessor _fd_input_processor;
// Set with cpuwide_time_us() at last read operation
butil::atomic<int64_t> _last_readtime_us;
@@ -912,6 +924,9 @@ private:
// Saved context for parsing, reset before trying other protocols.
butil::atomic<Destroyable*> _parsing_context;
+ // The stream whose data the handlers are currently cutting.
+ butil::atomic<InputMessengerProcessor::StreamType> _parsing_stream_type;
+
// Saving the correlation_id of RPC on protocols that cannot put
// correlation_id on-wire and do not send multiple requests on one
// connection simultaneously.
diff --git a/src/brpc/ubshm/ub_endpoint.cpp b/src/brpc/ubshm/ub_endpoint.cpp
index 31539fda..19bca4f9 100644
--- a/src/brpc/ubshm/ub_endpoint.cpp
+++ b/src/brpc/ubshm/ub_endpoint.cpp
@@ -52,8 +52,6 @@ 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 UBRing polling mode.");
-static const size_t MIN_ONCE_READ = 4096;
-static const size_t MAX_ONCE_READ = 524288;
static const size_t IOBUF_IOV_MAX = 256;
static const char* MAGIC_STR = "UB";
@@ -480,7 +478,7 @@ void* UBShmEndpoint::ProcessHandshakeAtServer(void* arg) {
LOG_IF(INFO, FLAGS_ub_trace_verbose) << "It seems that the "
<< "client does not use RDMA, fallback to TCP:"
<< s->description();
- s->_read_buf.append(data, MAGIC_STR_LEN);
+ s->fd_input_processor().read_buf().append(data, MAGIC_STR_LEN);
ep->_state = FALLBACK_TCP;
ub_transport->_ub_state = UBShmTransport::UB_OFF;
ep->TryReadOnTcp();
@@ -746,24 +744,19 @@ void UBShmEndpoint::PollIn(UBShmEndpoint* ep, uint32_t ep_event) {
return;
}
+ InputMessengerProcessor& processor = s->fd_input_processor();
bool read_eof = false;
while (!read_eof) {
const int64_t received_us = butil::cpuwide_time_us();
const int64_t base_realtime = butil::gettimeofday_us() - received_us;
- size_t once_read = s->_avg_msg_size * 16;
- if (once_read < MIN_ONCE_READ) {
- once_read = MIN_ONCE_READ;
- } else if (once_read > MAX_ONCE_READ) {
- once_read = MAX_ONCE_READ;
- }
-
- const ssize_t nr = s->_read_buf.append_from_reader(ep->_ub_ring, once_read);
+ const ssize_t nr = processor.read_buf().append_from_reader(
+ ep->_ub_ring, processor.OnceReadSize());
if (nr <= 0) {
if (0 == nr) {
// Set `read_eof' flag and proceed to feed EOF into `Protocol'
- // (implied by m->_read_buf.empty), which may produce a new
- // `InputMessageBase' under some protocols such as HTTP
+ // (implied by an empty processor.read_buf()), which may produce
+ // a new `InputMessageBase' under some protocols such as HTTP
LOG_IF(WARNING, FLAGS_log_connection_close) << *s << " was closed by remote side";
read_eof = true;
} else if (errno != EAGAIN) {
@@ -780,11 +773,10 @@ void UBShmEndpoint::PollIn(UBShmEndpoint* ep, uint32_t ep_event) {
}
}
- InputMessenger* messenger = static_cast<InputMessenger*>(s->user());
- if (messenger->ProcessNewMessage(s.get(), nr, read_eof, received_us,
- base_realtime, last_msg) < 0) {
+ if (processor.ProcessNewMessage(nr, read_eof, received_us,
+ base_realtime, last_msg) < 0) {
return;
- }
+ }
}
if (read_eof) {
diff --git a/test/brpc_rdma_unittest.cpp b/test/brpc_rdma_unittest.cpp
index 9c52acb7..d21095e1 100644
--- a/test/brpc_rdma_unittest.cpp
+++ b/test/brpc_rdma_unittest.cpp
@@ -21,11 +21,16 @@
#include <gtest/gtest.h>
#include <gflags/gflags.h>
#if BRPC_WITH_RDMA
+#include <errno.h>
+#include <unistd.h>
+#include <functional>
+#include <vector>
#include <google/protobuf/descriptor.h>
#include "butil/endpoint.h"
#include "butil/fd_guard.h"
#include "butil/iobuf.h"
#include "butil/sys_byteorder.h"
+#include "butil/time.h"
#include "butil/files/temp_file.h"
#include "brpc/acceptor.h"
#include "brpc/channel.h"
@@ -79,6 +84,90 @@ extern bool g_fail_resource_alloc_for_test;
static std::string g_ip = "127.0.0.1";
static butil::EndPoint g_ep;
+// Number of Echo requests the server has actually served. The churn tests below
+// cut the connection while requests are in flight, and from the client side
+// "the server never saw this request" is indistinguishable from "the server
+// answered it and the reply died with the connection" -- both just look like a
+// failed RPC. Only this counter tells the two apart, and it is the server having
+// real work in flight that makes the race those tests hunt reachable at all.
+static butil::atomic<int> g_echo_served(0);
+
+// The server side runs in its own threads, so the only thing a test can rely on
+// is that an expected transition happens *eventually*. Polling with a generous
+// upper bound keeps the common case as fast as the machine allows and does not
+// turn into a flake when CI is loaded, which a fixed sleep does.
+static const int64_t WAIT_TIMEOUT_US = 5000000;
+static const int64_t WAIT_INTERVAL_US = 1000;
+
+// Returns true if `pred` turned true before the timeout expired.
+static bool WaitUntil(const std::function<bool()>& pred,
+ int64_t timeout_us = WAIT_TIMEOUT_US) {
+ const int64_t deadline = butil::gettimeofday_us() + timeout_us;
+ while (!pred()) {
+ if (butil::gettimeofday_us() >= deadline) {
+ return false;
+ }
+ usleep((useconds_t)WAIT_INTERVAL_US);
+ }
+ return true;
+}
+
+// write(2) may transfer less than asked for, so a test that cares about the
+// whole buffer reaching the peer has to loop.
+static bool WriteAll(int fd, const void* buf, size_t len) {
+ const uint8_t* p = (const uint8_t*)buf;
+ for (size_t done = 0; done < len; ) {
+ ssize_t n = write(fd, p + done, len - done);
+ if (n < 0) {
+ if (errno == EINTR) {
+ continue;
+ }
+ return false;
+ }
+ done += n;
+ }
+ return true;
+}
+
+// Same for read(2). Returns false on error, on timeout (see ConnectToServer)
+// and on EOF before the whole buffer arrived.
+static bool ReadAll(int fd, void* buf, size_t len) {
+ uint8_t* p = (uint8_t*)buf;
+ for (size_t done = 0; done < len; ) {
+ ssize_t n = read(fd, p + done, len - done);
+ if (n < 0) {
+ if (errno == EINTR) {
+ continue;
+ }
+ return false;
+ }
+ if (n == 0) {
+ return false;
+ }
+ done += n;
+ }
+ return true;
+}
+
+// Connect a raw socket to the test server. Note that sin_addr is taken from
+// `g_ep`: a zeroed one means 0.0.0.0, which only happens to reach the local
+// server on some systems. The receive timeout keeps a missing server reply from
+// hanging the test forever.
+static void ConnectToServer(butil::fd_guard* sockfd) {
+ sockaddr_in addr;
+ bzero((char*)&addr, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(g_ep.port);
+ addr.sin_addr = g_ep.ip;
+ sockfd->reset(socket(AF_INET, SOCK_STREAM, 0));
+ ASSERT_TRUE(*sockfd >= 0);
+ timeval tv;
+ tv.tv_sec = WAIT_TIMEOUT_US / 1000000;
+ tv.tv_usec = WAIT_TIMEOUT_US % 1000000;
+ ASSERT_EQ(0, setsockopt(*sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)));
+ ASSERT_EQ(0, connect(*sockfd, (sockaddr*)&addr, sizeof(addr)));
+}
+
class MyEchoService : public ::test::EchoService {
void Echo(google::protobuf::RpcController* cntl_base,
const ::test::EchoRequest* req,
@@ -86,6 +175,7 @@ class MyEchoService : public ::test::EchoService {
google::protobuf::Closure* done) {
Controller* cntl = static_cast<Controller*>(cntl_base);
ClosureGuard done_guard(done);
+ g_echo_served.fetch_add(1, butil::memory_order_relaxed);
if (req->server_fail()) {
cntl->SetFailed(req->server_fail(), "Server fail1");
cntl->SetFailed(req->server_fail(), "Server fail2");
@@ -157,6 +247,19 @@ protected:
return nullptr;
}
+ // Accepting the connection happens in the server threads, so poll for it
+ // rather than sleeping. Returns nullptr if it never showed up.
+ Socket* WaitForServerSocket() {
+ Socket* s = nullptr;
+ WaitUntil([this, &s] { return (s = GetSocketFromServer(0)) != nullptr; });
+ return s;
+ }
+
+ // Ditto for the connection going away.
+ bool WaitForServerSocketGone() {
+ return WaitUntil([this] { return GetSocketFromServer(0) == nullptr; });
+ }
+
butil::TempFile _server_list;
std::string _naming_url;
@@ -164,6 +267,55 @@ protected:
MyEchoService _svc;
};
+// Shorthand for the RDMA transport behind a Socket, which every endpoint state
+// check below has to go through.
+static RdmaTransport* RdmaTransportOf(Socket* s) {
+ return static_cast<RdmaTransport*>(s->_transport.get());
+}
+static RdmaTransport* RdmaTransportOf(const SocketUniquePtr& s) {
+ return RdmaTransportOf(s.get());
+}
+
+// Polls until the endpoint reaches `expected` and returns the last state seen,
+// so that ASSERT_RDMA_STATE() reports what the endpoint actually settled on.
+static rdma::RdmaEndpoint::State WaitForRdmaState(
+ RdmaTransport* transport, rdma::RdmaEndpoint::State expected) {
+ rdma::RdmaEndpoint::State state = transport->_rdma_ep->_state;
+ WaitUntil([transport, expected, &state] {
+ state = transport->_rdma_ep->_state;
+ return state == expected;
+ });
+ return state;
+}
+
+// Waits for `transport` to reach `expected`, failing the test if it does not.
+#define ASSERT_RDMA_STATE(expected, transport) \
+ ASSERT_EQ(expected, WaitForRdmaState(transport, expected))
+
+// Polls until the fd stream of `s` holds exactly `size` bytes. Tests asserting
+// that a state did NOT change need this: waiting for the state itself would
+// return before the peer had read anything at all.
+static bool WaitForFdReadBuf(Socket* s, size_t size) {
+ return WaitUntil([s, size] {
+ return s->fd_input_processor().read_buf().size() == size;
+ });
+}
+
+// Build a well-formed v2 client hello: "RDMA" followed by the 36B body.
+static void MakeV2ClientHello(uint8_t (&data)[rdma::HELLO_V2_MSG_LEN_MIN]) {
+ rdma::v2_wire::HelloMessage msg{};
+ msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
+ msg.hello_ver = rdma::HELLO_V2_VERSION;
+ msg.impl_ver = rdma::IMPL_V2_VERSION;
+ msg.sq_size = 16;
+ msg.rq_size = 16;
+ msg.block_size = 8192;
+ msg.qp_num = 0;
+ msg.gid = rdma::GetRdmaGid();
+ memcpy(data, "RDMA", 4);
+ msg.Serialize(data + 4);
+}
+
// Parameterized fixture used by upper-layer RPC tests that have no
// dependency on the handshake wire format. The parameter is the
// client-side handshake protocol version (FLAGS_rdma_client_handshake_version),
@@ -190,20 +342,13 @@ private:
TEST_F(RdmaTest, client_close_before_hello_send) {
StartServer();
- sockaddr_in addr;
- bzero((char*)&addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_port = htons(PORT);
-
- butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd >= 0);
- ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- Socket* s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- close(sockfd);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd));
+ Socket* s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
+ sockfd.reset(-1);
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -211,26 +356,23 @@ TEST_F(RdmaTest, client_close_before_hello_send) {
TEST_F(RdmaTest, client_hello_msg_invalid_magic_str) {
StartServer();
- sockaddr_in addr;
- bzero((char*)&addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_port = htons(PORT);
-
- butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd >= 0);
- ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- Socket* s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd));
+ Socket* s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
memcpy(data, "PRPC", 4); // send as normal baidu_std protocol
- ASSERT_EQ(4, write(sockfd, data, 4));
- usleep(100000); // wait for server to handle the msg
+ ASSERT_TRUE(WriteAll(sockfd, data, 4));
+ // Wait for the bytes to show up in the fd stream (baidu_std wants 12B of
+ // header, so they stay buffered). Waiting on the state instead would prove
+ // nothing: it is already UNINIT before the server has read anything.
+ ASSERT_TRUE(WaitForFdReadBuf(s, 4));
// A non-RDMA magic makes ParseRdmaHandshake return TRY_OTHERS and hand the
// bytes to other protocols; it does not touch the endpoint state, so it
// stays UNINIT (the old blocking handshake used to set FALLBACK_TCP here).
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
StopServer();
}
@@ -238,62 +380,51 @@ TEST_F(RdmaTest, client_hello_msg_invalid_magic_str) {
TEST_F(RdmaTest, client_close_during_hello_send) {
StartServer();
- sockaddr_in addr;
- bzero((char*)&addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_port = htons(PORT);
Socket* s = nullptr;
uint8_t data[8];
- butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd1 >= 0);
- ASSERT_EQ(0, connect(sockfd1, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ butil::fd_guard sockfd1;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd1));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
memcpy(data, "RD", 2);
- ASSERT_EQ(2, write(sockfd1, data, 2)); // break in magic str
- usleep(100000); // wait for server to handle the msg
+ ASSERT_TRUE(WriteAll(sockfd1, data, 2)); // break in magic str
// Fewer than 4 magic bytes: ParseRdmaHandshake can't tell yet, returns
// NOT_ENOUGH_DATA and leaves the endpoint UNINIT (the old blocking
- // handshake used to set S_HELLO_WAIT before reading the magic).
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- close(sockfd1);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
-
- butil::fd_guard sockfd2(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd2 >= 0);
- ASSERT_EQ(0, connect(sockfd2, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ // handshake used to set S_HELLO_WAIT before reading the magic). Wait for
+ // the bytes to be buffered, the state alone would prove nothing.
+ ASSERT_TRUE(WaitForFdReadBuf(s, 2));
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
+ sockfd1.reset(-1);
+ ASSERT_TRUE(WaitForServerSocketGone());
+
+ butil::fd_guard sockfd2;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd2));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
memcpy(data, "RDMA", 4);
- ASSERT_EQ(4, write(sockfd2, data, 4)); // break after magic str
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- close(sockfd2);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
-
- butil::fd_guard sockfd3(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd3 >= 0);
- ASSERT_EQ(0, connect(sockfd3, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd2, data, 4)); // break after magic str
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
+ sockfd2.reset(-1);
+ ASSERT_TRUE(WaitForServerSocketGone());
+
+ butil::fd_guard sockfd3;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd3));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
// Send the 4B magic plus a valid msg_len (=40) but no body, so the server
// recognizes an RDMA v2 hello and waits for the remaining bytes. (A zero
// msg_len would now be rejected up-front as a protocol error.)
memcpy(data, "RDMA", 4);
uint16_t v2_len = butil::HostToNet16(rdma::HELLO_V2_MSG_LEN_MIN);
memcpy(data + 4, &v2_len, sizeof(v2_len));
- ASSERT_EQ(6, write(sockfd3, data, 6)); // magic + msg_len, body missing
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- close(sockfd3);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WriteAll(sockfd3, data, 6)); // magic + msg_len, body missing
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
+ sockfd3.reset(-1);
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -301,44 +432,34 @@ TEST_F(RdmaTest, client_close_during_hello_send) {
TEST_F(RdmaTest, client_hello_msg_invalid_len) {
StartServer();
- sockaddr_in addr;
- bzero((char*)&addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_port = htons(PORT);
Socket* s = nullptr;
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
- butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd1 >= 0);
- ASSERT_EQ(0, connect(sockfd1, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ butil::fd_guard sockfd1;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd1));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
memcpy(data, "RDMA", 4);
- ASSERT_EQ(4, write(sockfd1, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd1, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
memset(data + 4, 0, 36);
- ASSERT_EQ(36, write(sockfd1, data + 4, 36)); // Write invalid length.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
-
- butil::fd_guard sockfd2(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd2 >= 0);
- ASSERT_EQ(0, connect(sockfd2, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd1, data + 4, 36)); // Write invalid length.
+ ASSERT_TRUE(WaitForServerSocketGone());
+
+ butil::fd_guard sockfd2;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd2));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
memcpy(data, "RDMA", 4);
- ASSERT_EQ(4, write(sockfd2, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd2, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
uint16_t len = butil::HostToNet16(35);
memcpy(data + 4, &len, sizeof(len));
memset(data + 6, 0, 34);
- ASSERT_EQ(36, write(sockfd2, data + 4, 36)); // write invalid length
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WriteAll(sockfd2, data + 4, 36)); // write invalid length
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -346,25 +467,19 @@ TEST_F(RdmaTest, client_hello_msg_invalid_len) {
TEST_F(RdmaTest, client_hello_msg_invalid_version) {
StartServer();
- sockaddr_in addr;
- bzero((char*)&addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_port = htons(PORT);
Socket* s = nullptr;
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
uint16_t len = butil::HostToNet16(rdma::HELLO_V2_MSG_LEN_MIN);
uint16_t ver = butil::HostToNet16(1);
- butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd1 >= 0);
- ASSERT_EQ(0, connect(sockfd1, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ butil::fd_guard sockfd1;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd1));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
memcpy(data, "RDMA", 4);
- ASSERT_EQ(4, write(sockfd1, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd1, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
memcpy(data + 4, &len, 2);
memset(data + 6, 0, 34);
memcpy(data + 6, &ver, 2); // hello_ver == 1, impl_ver == 0
@@ -375,43 +490,35 @@ TEST_F(RdmaTest, client_hello_msg_invalid_version) {
// hello_ver). Now that Step 1 enforces a HELLO_V2_MSG_LEN_MAX upper bound,
// such an oversized msg_len would be rejected before reaching the
// version check, breaking the intent of this UT.
- ASSERT_EQ(36, write(sockfd1, data + 4, 36));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(RdmaTransport::RDMA_OFF, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_state);
+ ASSERT_TRUE(WriteAll(sockfd1, data + 4, 36));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, RdmaTransportOf(s)->_rdma_state);
uint32_t flags = 0;
- ASSERT_EQ(sizeof(flags), write(sockfd1, &flags, sizeof(flags)));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd1, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
sockfd1.reset(-1);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
-
- butil::fd_guard sockfd2(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd2 >= 0);
- ASSERT_EQ(0, connect(sockfd2, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_TRUE(WaitForServerSocketGone());
+
+ butil::fd_guard sockfd2;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd2));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
memcpy(data, "RDMA", 4);
- ASSERT_EQ(4, write(sockfd2, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd2, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
memcpy(data + 4, &len, 2);
memset(data + 6, 0, 32);
memcpy(data + 8, &ver, 2); // hello_ver == 0, impl_ver == 1
- // See comment above on `write(sockfd1, data + 4, 36)` for why we
+ // See comment above on `WriteAll(sockfd1, data + 4, 36)` for why we
// write from data + 4 instead of data.
- ASSERT_EQ(36, write(sockfd2, data + 4, 36));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(RdmaTransport::RDMA_OFF, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_state);
- ASSERT_EQ(sizeof(flags), write(sockfd2, &flags, sizeof(flags)));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd2, data + 4, 36));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, RdmaTransportOf(s)->_rdma_state);
+ ASSERT_TRUE(WriteAll(sockfd2, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
sockfd2.reset(-1);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -419,10 +526,6 @@ TEST_F(RdmaTest, client_hello_msg_invalid_version) {
TEST_F(RdmaTest, client_hello_msg_invalid_sq_rq_block_size) {
StartServer();
- sockaddr_in addr;
- bzero((char*)&addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_port = htons(PORT);
Socket* s = nullptr;
uint32_t flags = butil::HostToNet32(0);
rdma::v2_wire::HelloMessage msg{};
@@ -436,75 +539,60 @@ TEST_F(RdmaTest, client_hello_msg_invalid_sq_rq_block_size) {
msg.block_size = 8192;
memcpy(data, "RDMA", 4);
msg.Serialize(data + 4);
- butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd1 >= 0);
- ASSERT_EQ(0, connect(sockfd1, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(4, write(sockfd1, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(36, write(sockfd1, data + 4, 36));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(RdmaTransport::RDMA_OFF, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_state);
- ASSERT_EQ(sizeof(flags), write(sockfd1, &flags, sizeof(flags)));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ butil::fd_guard sockfd1;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd1));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd1, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
+ ASSERT_TRUE(WriteAll(sockfd1, data + 4, 36));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, RdmaTransportOf(s)->_rdma_state);
+ ASSERT_TRUE(WriteAll(sockfd1, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
sockfd1.reset(-1);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
msg.sq_size = 16;
msg.rq_size = 10;
msg.block_size = 8192;
memcpy(data, "RDMA", 4);
msg.Serialize(data + 4);
- butil::fd_guard sockfd2(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd2 >= 0);
- ASSERT_EQ(0, connect(sockfd2, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(4, write(sockfd2, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(36, write(sockfd2, data + 4, 36));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(RdmaTransport::RDMA_OFF, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_state);
- ASSERT_EQ(sizeof(flags), write(sockfd2, &flags, sizeof(flags)));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ butil::fd_guard sockfd2;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd2));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd2, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
+ ASSERT_TRUE(WriteAll(sockfd2, data + 4, 36));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, RdmaTransportOf(s)->_rdma_state);
+ ASSERT_TRUE(WriteAll(sockfd2, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
sockfd2.reset(-1);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
msg.sq_size = 16;
msg.rq_size = 16;
msg.block_size = 1000;
memcpy(data, "RDMA", 4);
msg.Serialize(data + 4);
- butil::fd_guard sockfd3(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd3 >= 0);
- ASSERT_EQ(0, connect(sockfd3, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(4, write(sockfd3, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(36, write(sockfd3, data + 4, 36));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(RdmaTransport::RDMA_OFF, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_state);
- ASSERT_EQ(sizeof(flags), write(sockfd3, &flags, sizeof(flags)));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ butil::fd_guard sockfd3;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd3));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd3, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
+ ASSERT_TRUE(WriteAll(sockfd3, data + 4, 36));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, RdmaTransportOf(s)->_rdma_state);
+ ASSERT_TRUE(WriteAll(sockfd3, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
sockfd3.reset(-1);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -512,36 +600,19 @@ TEST_F(RdmaTest, client_hello_msg_invalid_sq_rq_block_size) {
TEST_F(RdmaTest, client_close_after_qp_build) {
StartServer();
- sockaddr_in addr;
- bzero((char*)&addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_port = htons(PORT);
Socket* s = nullptr;
- rdma::v2_wire::HelloMessage msg{};
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
- msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
- msg.hello_ver = rdma::HELLO_V2_VERSION;
- msg.impl_ver = rdma::IMPL_V2_VERSION;
- msg.sq_size = 16;
- msg.rq_size = 16;
- msg.block_size = 8192;
- msg.qp_num = 0;
- msg.gid = rdma::GetRdmaGid();
- memcpy(data, "RDMA", 4);
- msg.Serialize(data + 4);
+ MakeV2ClientHello(data);
- butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd1 >= 0);
- ASSERT_EQ(0, connect(sockfd1, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(40, write(sockfd1, data, 40));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- close(sockfd1);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ butil::fd_guard sockfd1;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd1));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd1, data, sizeof(data)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
+ sockfd1.reset(-1);
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -549,43 +620,24 @@ TEST_F(RdmaTest, client_close_after_qp_build) {
TEST_F(RdmaTest, client_close_during_ack_send) {
StartServer();
- sockaddr_in addr;
- bzero((char*)&addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_port = htons(PORT);
Socket* s = nullptr;
- rdma::v2_wire::HelloMessage msg{};
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
- msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
- msg.hello_ver = rdma::HELLO_V2_VERSION;
- msg.impl_ver = rdma::IMPL_V2_VERSION;
- msg.sq_size = 16;
- msg.rq_size = 16;
- msg.block_size = 8192;
- msg.qp_num = 0;
- msg.gid = rdma::GetRdmaGid();
- memcpy(data, "RDMA", 4);
- msg.Serialize(data + 4);
+ MakeV2ClientHello(data);
- butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd1 >= 0);
- ASSERT_EQ(0, connect(sockfd1, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(4, write(sockfd1, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(36, write(sockfd1, data + 4, 36));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ butil::fd_guard sockfd1;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd1));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd1, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
+ ASSERT_TRUE(WriteAll(sockfd1, data + 4, 36));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
uint32_t flags = butil::HostToNet32(1);
- ASSERT_EQ(sizeof(flags), write(sockfd1, &flags, sizeof(flags)));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- close(sockfd1);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WriteAll(sockfd1, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::ESTABLISHED, RdmaTransportOf(s));
+ sockfd1.reset(-1);
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -593,64 +645,40 @@ TEST_F(RdmaTest, client_close_during_ack_send) {
TEST_F(RdmaTest, client_close_after_ack_send) {
StartServer();
- sockaddr_in addr;
- bzero((char*)&addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_port = htons(PORT);
Socket* s = nullptr;
- rdma::v2_wire::HelloMessage msg{};
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
- msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
- msg.hello_ver = rdma::HELLO_V2_VERSION;
- msg.impl_ver = rdma::IMPL_V2_VERSION;
- msg.sq_size = 16;
- msg.rq_size = 16;
- msg.block_size = 8192;
- msg.qp_num = 0;
- msg.gid = rdma::GetRdmaGid();
- memcpy(data, "RDMA", 4);
- msg.Serialize(data + 4);
+ MakeV2ClientHello(data);
- butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd1 >= 0);
- ASSERT_EQ(0, connect(sockfd1, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(4, write(sockfd1, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(36, write(sockfd1, data + 4, 36));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ butil::fd_guard sockfd1;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd1));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd1, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
+ ASSERT_TRUE(WriteAll(sockfd1, data + 4, 36));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
uint32_t flags = butil::HostToNet32(0);
- ASSERT_EQ(sizeof(flags), write(sockfd1, &flags, sizeof(flags)));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(RdmaTransport::RDMA_OFF, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_state);
- close(sockfd1);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
-
- butil::fd_guard sockfd2(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd2 >= 0);
- ASSERT_EQ(0, connect(sockfd2, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(4, write(sockfd2, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(36, write(sockfd2, data + 4, 36));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd1, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, RdmaTransportOf(s)->_rdma_state);
+ sockfd1.reset(-1);
+ ASSERT_TRUE(WaitForServerSocketGone());
+
+ butil::fd_guard sockfd2;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd2));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd2, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
+ ASSERT_TRUE(WriteAll(sockfd2, data + 4, 36));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
flags = butil::HostToNet32(1);
- ASSERT_EQ(sizeof(flags), write(sockfd2, &flags, sizeof(flags)));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- close(sockfd2);
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WriteAll(sockfd2, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::ESTABLISHED, RdmaTransportOf(s));
+ sockfd2.reset(-1);
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -658,63 +686,240 @@ TEST_F(RdmaTest, client_close_after_ack_send) {
TEST_F(RdmaTest, client_send_data_on_tcp_after_ack_send) {
StartServer();
- sockaddr_in addr;
- bzero((char*)&addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_port = htons(PORT);
Socket* s = nullptr;
- rdma::v2_wire::HelloMessage msg{};
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
- msg.msg_len = rdma::HELLO_V2_MSG_LEN_MIN;
- msg.hello_ver = rdma::HELLO_V2_VERSION;
- msg.impl_ver = rdma::IMPL_V2_VERSION;
- msg.sq_size = 16;
- msg.rq_size = 16;
- msg.block_size = 8192;
- msg.qp_num = 0;
- msg.gid = rdma::GetRdmaGid();
- memcpy(data, "RDMA", 4);
- msg.Serialize(data + 4);
+ MakeV2ClientHello(data);
- butil::fd_guard sockfd1(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd1 >= 0);
- ASSERT_EQ(0, connect(sockfd1, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(4, write(sockfd1, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(36, write(sockfd1, data + 4, 36));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ butil::fd_guard sockfd1;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd1));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd1, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
+ ASSERT_TRUE(WriteAll(sockfd1, data + 4, 36));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
uint32_t flags = butil::HostToNet32(0);
- ASSERT_EQ(sizeof(flags), write(sockfd1, &flags, sizeof(flags)));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(sizeof(flags), write(sockfd1, &flags, sizeof(flags)));
- usleep(100000);
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
-
- butil::fd_guard sockfd2(socket(AF_INET, SOCK_STREAM, 0));
- ASSERT_TRUE(sockfd2 >= 0);
- ASSERT_EQ(0, connect(sockfd2, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(4, write(sockfd2, data, 4)); // Write magic string.
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(36, write(sockfd2, data + 4, 36));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd1, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
+ // 4 more bytes on a fd that fell back to TCP are not a protocol baidu_std
+ // knows, so the connection is dropped.
+ ASSERT_TRUE(WriteAll(sockfd1, &flags, sizeof(flags)));
+ ASSERT_TRUE(WaitForServerSocketGone());
+
+ butil::fd_guard sockfd2;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd2));
+ s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
+ ASSERT_TRUE(WriteAll(sockfd2, data, 4)); // Write magic string.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_HELLO_WAIT, RdmaTransportOf(s));
+ ASSERT_TRUE(WriteAll(sockfd2, data + 4, 36));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
flags = butil::HostToNet32(1);
- ASSERT_EQ(sizeof(flags), write(sockfd2, &flags, sizeof(flags)));
- usleep(100000); // wait for server to handle the msg
- ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(sizeof(flags), write(sockfd2, &flags, sizeof(flags)));
- usleep(100000);
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WriteAll(sockfd2, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::ESTABLISHED, RdmaTransportOf(s));
+ // Once RDMA is on the fd carries no RPC data at all, so this is an error.
+ ASSERT_TRUE(WriteAll(sockfd2, &flags, sizeof(flags)));
+ ASSERT_TRUE(WaitForServerSocketGone());
+
+ StopServer();
+}
+
+// Connect, push a well-formed v2 hello and read back the server's reply, which
+// leaves the server in S_ACK_WAIT waiting for the 4B ACK.
+static void HandshakeUntilAckWait(butil::fd_guard* sockfd) {
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(sockfd));
+
+ uint8_t hello[rdma::HELLO_V2_MSG_LEN_MIN];
+ MakeV2ClientHello(hello);
+ ASSERT_TRUE(WriteAll(*sockfd, hello, sizeof(hello)));
+ // The server answers only once it has consumed our hello, so reading the
+ // whole reply is a synchronization point: no sleeping needed here.
+ uint8_t reply[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_TRUE(ReadAll(*sockfd, reply, sizeof(reply)));
+}
+
+// A client is free to pipeline its first request right behind the handshake
+// ACK. Only the 4B ACK belongs to the handshake. Whatever follows it must be
+// handed over to the real protocol instead of dropping the connection.
+TEST_F(RdmaTest, server_accepts_data_pipelined_behind_fallback_ack) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = RdmaTransportOf(s);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, transport);
+
+ // An ACK asking for TCP, plus the first 4 bytes of a baidu_std request. One
+ // write, so that both end up in the same read on the server.
+ uint8_t ack_and_data[rdma::HELLO_ACK_LEN + 4];
+ const uint32_t flags = butil::HostToNet32(0);
+ memcpy(ack_and_data, &flags, rdma::HELLO_ACK_LEN);
+ memcpy(ack_and_data + rdma::HELLO_ACK_LEN, "PRPC", 4);
+ ASSERT_TRUE(WriteAll(sockfd, ack_and_data, sizeof(ack_and_data)));
+
+ // The handshake took the ACK only and left "PRPC" to baidu_std, which is
+ // now waiting for the rest of its 12B header. So the connection lives on
+ // with those 4 bytes still buffered. Note that baidu_std gets them a moment
+ // after the handshake gave up the stream, hence the wait.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, transport);
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, transport->_rdma_state);
+ ASSERT_TRUE(GetSocketFromServer(0) != nullptr);
+ ASSERT_TRUE(WaitForFdReadBuf(s, 4));
+
+ sockfd.reset(-1);
+ ASSERT_TRUE(WaitForServerSocketGone());
+
+ StopServer();
+}
+
+// Once RDMA is on, the TCP fd is no longer an RPC channel, so bytes trailing
+// the ACK can only be a protocol error.
+TEST_F(RdmaTest, server_rejects_data_pipelined_behind_rdma_ack) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = RdmaTransportOf(s);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, transport);
+
+ uint8_t ack_and_data[rdma::HELLO_ACK_LEN + 4];
+ const uint32_t flags = butil::HostToNet32(rdma::HELLO_ACK_RDMA_OK);
+ memcpy(ack_and_data, &flags, rdma::HELLO_ACK_LEN);
+ memcpy(ack_and_data + rdma::HELLO_ACK_LEN, "PRPC", 4);
+ ASSERT_TRUE(WriteAll(sockfd, ack_and_data, sizeof(ack_and_data)));
+
+ // Note that `transport->_rdma_ep` is gone by now: dropping the connection
+ // recycles the Socket, and RdmaTransport::Release() deletes the endpoint.
+ ASSERT_TRUE(WaitForServerSocketGone());
+
+ StopServer();
+}
+
+// Once RDMA is on, the server must stop parsing its TCP fd altogether.
+TEST_F(RdmaTest, server_stops_parsing_tcp_fd_once_rdma_is_on) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = RdmaTransportOf(s);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, transport);
+
+ // A bare ACK asking for RDMA. Nothing trails it, so the handshake ends in
+ // ESTABLISHED instead of being rejected (see the test above).
+ const uint32_t flags = butil::HostToNet32(rdma::HELLO_ACK_RDMA_OK);
+ ASSERT_TRUE(WriteAll(sockfd, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::ESTABLISHED, transport);
+ ASSERT_EQ(RdmaTransport::RDMA_ON, transport->_rdma_state);
+ ASSERT_TRUE(GetSocketFromServer(0) != nullptr);
+
+ ASSERT_TRUE(WriteAll(sockfd, "PRPC", 4));
+ ASSERT_TRUE(WaitForServerSocketGone());
+
+ StopServer();
+}
+
+// The same bytes on the stream carried by the QP are a real RPC, and the handler
+// must decline so that CutInputMessage() moves on to the protocol handlers.
+TEST_F(RdmaTest, server_parses_qp_stream_after_rdma_is_on) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = RdmaTransportOf(s);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, transport);
+
+ const uint32_t flags = butil::HostToNet32(rdma::HELLO_ACK_RDMA_OK);
+ ASSERT_TRUE(WriteAll(sockfd, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::ESTABLISHED, transport);
+
+ InputMessengerProcessor& qp_stream = transport->_rdma_ep->_input_processor;
+ ASSERT_TRUE(qp_stream.read_buf().empty());
+ qp_stream.read_buf().append("PRPC");
+ InputMessageClosure last_msg;
+ // The real caller stamps messages with cpuwide_time_us() and derives
+ // base_realtime from it, so feed ProcessNewMessage() the same time domain:
+ // received_us also ends up in Socket::_last_readtime_us.
+ const uint64_t received_us = butil::cpuwide_time_us();
+ const uint64_t base_realtime = butil::gettimeofday_us() - received_us;
+ ASSERT_EQ(0, qp_stream.ProcessNewMessage(
+ 4, false, received_us, base_realtime, last_msg));
+ // baidu_std claimed the stream and is waiting for the rest of its header.
+ ASSERT_EQ((int)PROTOCOL_BAIDU_STD, s->preferred_index());
+ ASSERT_EQ(4u, qp_stream.read_buf().size());
+ ASSERT_TRUE(s->fd_input_processor().read_buf().empty());
+ ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED, transport->_rdma_ep->_state);
+ ASSERT_FALSE(s->Failed());
+
+ StopServer();
+}
+
+// After the handshake is over, CutInputMessage() still offers the data to every
+// registered handler, this one included. It must decline instead of reading the
+// data as a fresh client hello.
+TEST_F(RdmaTest, server_declines_handshake_bytes_after_fallback) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(HandshakeUntilAckWait(&sockfd));
+ Socket* s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = RdmaTransportOf(s);
+
+ const uint32_t flags = butil::HostToNet32(0);
+ ASSERT_TRUE(WriteAll(sockfd, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, transport);
+
+ // Replay a valid hello. baidu_std rejects it and no other protocol claims
+ // it, so the connection is dropped. What must NOT happen is a second
+ // handshake: that would answer with another server hello.
+ uint8_t hello[rdma::HELLO_V2_MSG_LEN_MIN];
+ MakeV2ClientHello(hello);
+ ASSERT_TRUE(WriteAll(sockfd, hello, sizeof(hello)));
+
+ // Note that `transport->_rdma_ep` is gone by now: dropping the connection
+ // recycles the Socket, and RdmaTransport::Release() deletes the endpoint.
+ ASSERT_TRUE(WaitForServerSocketGone());
+ uint8_t reply[rdma::HELLO_V2_MSG_LEN_MIN];
+ ASSERT_LE(recv(sockfd, reply, sizeof(reply), MSG_DONTWAIT), 0);
+
+ StopServer();
+}
+
+TEST_F(RdmaTest, fd_and_qp_input_streams_are_separate) {
+ StartServer();
+
+ butil::fd_guard sockfd;
+ ASSERT_NO_FATAL_FAILURE(ConnectToServer(&sockfd));
+ Socket* s = WaitForServerSocket();
+ ASSERT_TRUE(s != nullptr);
+ auto* transport = RdmaTransportOf(s);
+
+ // The fd stream belongs to the Socket, the QP stream to the endpoint.
+ InputMessengerProcessor& fd_stream = s->fd_input_processor();
+ InputMessengerProcessor& qp_stream = transport->_rdma_ep->_input_processor;
+ ASSERT_NE(&fd_stream, &qp_stream);
+ ASSERT_NE(&fd_stream.read_buf(), &qp_stream.read_buf());
+
+ // Two magic bytes are too few to dispatch on, so they stay buffered. In the
+ // fd stream, and only there.
+ ASSERT_TRUE(WriteAll(sockfd, "RD", 2));
+ ASSERT_TRUE(WaitForFdReadBuf(s, 2));
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, transport->_rdma_ep->_state);
+ ASSERT_EQ(2u, fd_stream.read_buf().size());
+ ASSERT_TRUE(qp_stream.read_buf().empty());
+
+ sockfd.reset(-1);
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -738,10 +943,9 @@ TEST_F(RdmaTest, server_miss_before_hello_send) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
@@ -769,18 +973,16 @@ TEST_F(RdmaTest, server_close_before_hello_send) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data, rdma::HELLO_V2_MSG_LEN_MIN));
close(acc_fd);
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FAILED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FAILED, RdmaTransportOf(s));
bthread_id_join(cntl.call_id());
ASSERT_EQ(EEOF, cntl.ErrorCode());
@@ -805,17 +1007,18 @@ TEST_F(RdmaTest, server_miss_during_magic_str) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data, rdma::HELLO_V2_MSG_LEN_MIN));
- ASSERT_EQ(2, write(acc_fd, "RD", 2));
- usleep(100000);
+ ASSERT_TRUE(WriteAll(acc_fd, "RD", 2));
+ // Half a magic is not enough to decide anything, so the client stays stuck
+ // in the handshake read and the RPC runs into its timeout. Joining below
+ // waits for exactly that, no sleeping needed.
bthread_id_join(cntl.call_id());
ASSERT_EQ(ERPCTIMEDOUT, cntl.ErrorCode());
@@ -840,20 +1043,19 @@ TEST_F(RdmaTest, server_close_during_magic_str) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data, rdma::HELLO_V2_MSG_LEN_MIN));
- ASSERT_EQ(2, write(acc_fd, "RD", 2));
- usleep(100000);
- close(acc_fd);
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FAILED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ // Half a magic and then EOF. TCP keeps the order, so the client always sees
+ // the two bytes first and then the close, which is what this test is about.
+ ASSERT_TRUE(WriteAll(acc_fd, "RD", 2));
+ acc_fd.reset(-1);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FAILED, RdmaTransportOf(s));
bthread_id_join(cntl.call_id());
ASSERT_EQ(EEOF, cntl.ErrorCode());
@@ -878,18 +1080,16 @@ TEST_F(RdmaTest, server_hello_invalid_magic_str) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
uint8_t data[rdma::HELLO_V2_MSG_LEN_MIN];
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, read(acc_fd, data, rdma::HELLO_V2_MSG_LEN_MIN));
ASSERT_EQ(4, write(acc_fd, "ABCD", 4));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FAILED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FAILED, RdmaTransportOf(s));
bthread_id_join(cntl.call_id());
ASSERT_EQ(EPROTO, cntl.ErrorCode());
@@ -914,10 +1114,9 @@ TEST_F(RdmaTest, server_miss_during_hello_msg) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
@@ -949,10 +1148,9 @@ TEST_F(RdmaTest, server_close_during_hello_msg) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
@@ -961,8 +1159,7 @@ TEST_F(RdmaTest, server_close_during_hello_msg) {
ASSERT_EQ(4, write(acc_fd, "RDMA", 4));
ASSERT_EQ(2, write(acc_fd, "00", 2));
close(acc_fd);
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FAILED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FAILED, RdmaTransportOf(s));
bthread_id_join(cntl.call_id());
ASSERT_EQ(EEOF, cntl.ErrorCode());
@@ -987,10 +1184,9 @@ TEST_F(RdmaTest, server_hello_invalid_msg_len) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
@@ -1001,8 +1197,7 @@ TEST_F(RdmaTest, server_hello_invalid_msg_len) {
memcpy(data + 4, &len, 2);
memset(data + 6, 0, 32);
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data, rdma::HELLO_V2_MSG_LEN_MIN));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FAILED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FAILED, RdmaTransportOf(s));
bthread_id_join(cntl.call_id());
ASSERT_EQ(EPROTO, cntl.ErrorCode());
@@ -1027,10 +1222,9 @@ TEST_F(RdmaTest, server_hello_invalid_version) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
@@ -1041,8 +1235,7 @@ TEST_F(RdmaTest, server_hello_invalid_version) {
memcpy(data + 4, &len, 2);
memset(data + 6, 0, 32);
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data, rdma::HELLO_V2_MSG_LEN_MIN));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
ASSERT_EQ(4, read(acc_fd, data, 4));
uint32_t* tmp = (uint32_t*)data;
ASSERT_EQ(0, butil::NetToHost32(*tmp));
@@ -1070,10 +1263,9 @@ TEST_F(RdmaTest, server_hello_invalid_sq_rq_size) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
@@ -1093,8 +1285,7 @@ TEST_F(RdmaTest, server_hello_invalid_sq_rq_size) {
msg.Serialize(data + 4);
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data, rdma::HELLO_V2_MSG_LEN_MIN));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
ASSERT_EQ(4, read(acc_fd, data, 4));
uint32_t* tmp = (uint32_t*)data;
ASSERT_EQ(0, butil::NetToHost32(*tmp));
@@ -1122,10 +1313,9 @@ TEST_F(RdmaTest, server_miss_after_ack) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
@@ -1145,8 +1335,7 @@ TEST_F(RdmaTest, server_miss_after_ack) {
msg.Serialize(data + 4);
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data, rdma::HELLO_V2_MSG_LEN_MIN));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::ESTABLISHED, RdmaTransportOf(s));
ASSERT_EQ(4, read(acc_fd, data, 4));
uint32_t* tmp = (uint32_t*)data;
ASSERT_EQ(1, butil::NetToHost32(*tmp));
@@ -1174,10 +1363,9 @@ TEST_F(RdmaTest, server_close_after_ack) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
@@ -1197,8 +1385,7 @@ TEST_F(RdmaTest, server_close_after_ack) {
msg.Serialize(data + 4);
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data, rdma::HELLO_V2_MSG_LEN_MIN));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::ESTABLISHED, RdmaTransportOf(s));
ASSERT_EQ(4, read(acc_fd, data, 4));
uint32_t* tmp = (uint32_t*)data;
ASSERT_EQ(1, butil::NetToHost32(*tmp));
@@ -1227,10 +1414,9 @@ TEST_F(RdmaTest, server_send_data_on_tcp_after_ack) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::C_HELLO_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::C_HELLO_WAIT, RdmaTransportOf(s));
butil::fd_guard acc_fd(accept(sockfd, nullptr, nullptr));
ASSERT_TRUE(acc_fd >= 0);
@@ -1250,8 +1436,7 @@ TEST_F(RdmaTest, server_send_data_on_tcp_after_ack) {
msg.Serialize(data + 4);
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data, rdma::HELLO_V2_MSG_LEN_MIN));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::ESTABLISHED, RdmaTransportOf(s));
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(acc_fd, data, rdma::HELLO_V2_MSG_LEN_MIN));
bthread_id_join(cntl.call_id());
@@ -1278,7 +1463,6 @@ TEST_F(RdmaTest, v2_client_hello_bytes_baseline) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
@@ -1320,9 +1504,8 @@ TEST_F(RdmaTest, v2_server_hello_bytes_baseline) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000);
- Socket* s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ Socket* s = WaitForServerSocket();
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
// Send a well-formed v2 hello so the server enters S_ACK_WAIT.
rdma::v2_wire::HelloMessage msg{};
@@ -1339,8 +1522,7 @@ TEST_F(RdmaTest, v2_server_hello_bytes_baseline) {
memcpy(data, "RDMA", 4);
msg.Serialize(data + 4);
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(sockfd, data, rdma::HELLO_V2_MSG_LEN_MIN));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
// Read server's reply hello and assert its byte-level layout.
uint8_t reply[rdma::HELLO_V2_MSG_LEN_MIN];
@@ -1364,12 +1546,10 @@ TEST_F(RdmaTest, v2_server_hello_bytes_baseline) {
// cleanly without requiring real RDMA hardware.
uint32_t flags = butil::HostToNet32(0);
ASSERT_EQ(sizeof(flags), write(sockfd, &flags, sizeof(flags)));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
sockfd.reset(-1);
- usleep(100000);
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -1384,10 +1564,9 @@ TEST_F(RdmaTest, v2_server_drains_tail_then_reads_ack) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000);
- Socket* s = GetSocketFromServer(0);
+ Socket* s = WaitForServerSocket();
ASSERT_TRUE(s != nullptr);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
// Build a v2 hello with msg_len = 48 (40 base + 8B zero tail).
rdma::v2_wire::HelloMessage msg{};
@@ -1404,19 +1583,19 @@ TEST_F(RdmaTest, v2_server_drains_tail_then_reads_ack) {
memcpy(buf, "RDMA", 4);
msg.Serialize(buf + 4);
memset(buf + 40, 0x00, 8); // 8B zero tail
- ASSERT_EQ(48, write(sockfd, buf, 48));
- usleep(100000);
+ ASSERT_TRUE(WriteAll(sockfd, buf, 48));
+ // The tail is drained as part of the hello, so the server ends up waiting
+ // for the ACK. Wait for that before sending it, otherwise the ACK could
+ // ride along in the same read and this would no longer test the drain.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
// Send the real ACK (flags=1 = ACK_MSG_RDMA_OK).
uint32_t flags = butil::HostToNet32(1);
ASSERT_EQ(sizeof(flags), write(sockfd, &flags, sizeof(flags)));
- usleep(100000);
-
- ASSERT_EQ(rdma::RdmaEndpoint::ESTABLISHED, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::ESTABLISHED, RdmaTransportOf(s));
sockfd.reset(-1);
- usleep(100000);
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -1431,10 +1610,9 @@ TEST_F(RdmaTest, v2_server_rejects_oversized_msg_len) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000);
- Socket* s = GetSocketFromServer(0);
+ Socket* s = WaitForServerSocket();
ASSERT_TRUE(s != nullptr);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
// Build a v2 hello with msg_len = 4097 (HELLO_V2_MSG_LEN_MAX + 1).
// We only send the 40B base; the server must reject before reading
@@ -1453,13 +1631,9 @@ TEST_F(RdmaTest, v2_server_rejects_oversized_msg_len) {
memcpy(buf, "RDMA", 4);
msg.Serialize(buf + 4);
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN, write(sockfd, buf, rdma::HELLO_V2_MSG_LEN_MIN));
- usleep(100000);
-
-
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
sockfd.reset(-1);
- usleep(100000);
StopServer();
}
@@ -1583,16 +1757,14 @@ TEST_F(RdmaTest, v3_server_hello_bytes_baseline) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000);
- Socket* s = GetSocketFromServer(0);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ Socket* s = WaitForServerSocket();
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
// Send a valid v3 hello.
std::string packet = MakeV3Packet(MakeValidV3Hello());
ASSERT_EQ((ssize_t)packet.size(),
write(sockfd, packet.data(), packet.size()));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
// Read server's reply hello: 4B magic + 4B pb_size + body.
uint8_t reply_magic[4];
@@ -1621,12 +1793,10 @@ TEST_F(RdmaTest, v3_server_hello_bytes_baseline) {
uint32_t flags = butil::HostToNet32(0);
ASSERT_EQ((ssize_t)sizeof(flags),
write(sockfd, &flags, sizeof(flags)));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
sockfd.reset(-1);
- usleep(100000);
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -1641,16 +1811,13 @@ TEST_F(RdmaTest, v3_server_rejects_zero_pb_size) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000);
- Socket* s = GetSocketFromServer(0);
+ Socket* s = WaitForServerSocket();
ASSERT_TRUE(s != nullptr);
// "RDM3" + pb_size = 0 (4B big-endian zero).
uint8_t buf[8] = {'R', 'D', 'M', '3', 0, 0, 0, 0};
ASSERT_EQ(8, write(sockfd, buf, 8));
- usleep(100000);
-
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
sockfd.reset(-1);
StopServer();
@@ -1666,8 +1833,7 @@ TEST_F(RdmaTest, v3_server_rejects_oversized_pb_size) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000);
- Socket* s = GetSocketFromServer(0);
+ Socket* s = WaitForServerSocket();
ASSERT_TRUE(s != nullptr);
uint8_t buf[8];
@@ -1677,9 +1843,7 @@ TEST_F(RdmaTest, v3_server_rejects_oversized_pb_size) {
butil::HostToNet32(static_cast<uint32_t>(rdma::HELLO_V3_MAX_PB_SIZE + 1));
memcpy(buf + 4, &pb_size_be, 4);
ASSERT_EQ(8, write(sockfd, buf, 8));
- usleep(100000);
-
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
sockfd.reset(-1);
StopServer();
@@ -1695,8 +1859,7 @@ TEST_F(RdmaTest, v3_server_rejects_invalid_pb_bytes) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000);
- Socket* s = GetSocketFromServer(0);
+ Socket* s = WaitForServerSocket();
ASSERT_TRUE(s != nullptr);
// "RDM3" + pb_size = 8 + 8 bytes of 0xff (invalid protobuf body).
@@ -1706,9 +1869,7 @@ TEST_F(RdmaTest, v3_server_rejects_invalid_pb_bytes) {
memcpy(buf + 4, &pb_size_be, 4);
memset(buf + 8, 0xff, 8);
ASSERT_EQ(16, write(sockfd, buf, 16));
- usleep(100000);
-
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
sockfd.reset(-1);
StopServer();
@@ -1724,21 +1885,18 @@ TEST_F(RdmaTest, v3_server_invalid_sq_size_falls_back) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000);
- Socket* s = GetSocketFromServer(0);
+ Socket* s = WaitForServerSocket();
ASSERT_TRUE(s != nullptr);
rdma::RdmaHello msg = MakeValidV3Hello();
msg.set_sq_size(0); // invalid: < MIN_QP_SIZE (16)
std::string packet = MakeV3Packet(msg);
- ASSERT_EQ((ssize_t)packet.size(),
- write(sockfd, packet.data(), packet.size()));
- usleep(100000);
+ ASSERT_TRUE(WriteAll(sockfd, packet.data(), packet.size()));
// Server validated the hello as invalid -> _rdma_state = RDMA_OFF,
// but still proceeds to S_ACK_WAIT (sends its own reply hello).
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(RdmaTransport::RDMA_OFF, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, RdmaTransportOf(s)->_rdma_state);
// Drain server's reply hello (content not asserted here; covered
// by v3_server_hello_bytes_baseline).
@@ -1755,12 +1913,10 @@ TEST_F(RdmaTest, v3_server_invalid_sq_size_falls_back) {
uint32_t flags = butil::HostToNet32(0);
ASSERT_EQ((ssize_t)sizeof(flags),
write(sockfd, &flags, sizeof(flags)));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
sockfd.reset(-1);
- usleep(100000);
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -1818,18 +1974,14 @@ TEST_F(RdmaTest, v3_server_accepts_client_hello_with_ece) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000);
- Socket* s = GetSocketFromServer(0);
+ Socket* s = WaitForServerSocket();
ASSERT_TRUE(s != nullptr);
rdma::RdmaHello msg = MakeValidV3HelloWithEce(0x02c9, 0x1, 0x0);
std::string packet = MakeV3Packet(msg);
ASSERT_EQ((ssize_t)packet.size(),
write(sockfd, packet.data(), packet.size()));
- usleep(100000);
-
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT,
- static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
rdma::RdmaHello reply;
ReadServerV3Reply(sockfd, &reply);
@@ -1837,13 +1989,10 @@ TEST_F(RdmaTest, v3_server_accepts_client_hello_with_ece) {
// ACK flags=0 -> clean FALLBACK_TCP so the test ends without hardware.
uint32_t flags = butil::HostToNet32(0);
ASSERT_EQ((ssize_t)sizeof(flags), write(sockfd, &flags, sizeof(flags)));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP,
- static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
sockfd.reset(-1);
- usleep(100000);
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -1860,26 +2009,24 @@ TEST_F(RdmaTest, v3_server_reply_has_no_ece_when_disabled) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000);
- Socket* s = GetSocketFromServer(0);
+ Socket* s = WaitForServerSocket();
ASSERT_TRUE(s != nullptr);
rdma::RdmaHello msg = MakeValidV3HelloWithEce(0x02c9, 0x1, 0x0);
std::string packet = MakeV3Packet(msg);
- ASSERT_EQ((ssize_t)packet.size(),
- write(sockfd, packet.data(), packet.size()));
- usleep(100000);
+ ASSERT_TRUE(WriteAll(sockfd, packet.data(), packet.size()));
+ // Reading the reply in full doubles as the synchronization point.
rdma::RdmaHello reply;
ReadServerV3Reply(sockfd, &reply);
EXPECT_FALSE(reply.has_ece());
uint32_t flags = butil::HostToNet32(0);
- ASSERT_EQ((ssize_t)sizeof(flags), write(sockfd, &flags, sizeof(flags)));
- usleep(100000);
+ ASSERT_TRUE(WriteAll(sockfd, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
sockfd.reset(-1);
- usleep(100000);
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -1897,26 +2044,24 @@ TEST_F(RdmaTest, v3_server_reply_has_no_ece_without_hw_negotiation) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000);
- Socket* s = GetSocketFromServer(0);
+ Socket* s = WaitForServerSocket();
ASSERT_TRUE(s != nullptr);
rdma::RdmaHello msg = MakeValidV3HelloWithEce(0x02c9, 0x1, 0x0);
std::string packet = MakeV3Packet(msg);
- ASSERT_EQ((ssize_t)packet.size(),
- write(sockfd, packet.data(), packet.size()));
- usleep(100000);
+ ASSERT_TRUE(WriteAll(sockfd, packet.data(), packet.size()));
+ // Reading the reply in full doubles as the synchronization point.
rdma::RdmaHello reply;
ReadServerV3Reply(sockfd, &reply);
EXPECT_FALSE(reply.has_ece());
uint32_t flags = butil::HostToNet32(0);
- ASSERT_EQ((ssize_t)sizeof(flags), write(sockfd, &flags, sizeof(flags)));
- usleep(100000);
+ ASSERT_TRUE(WriteAll(sockfd, &flags, sizeof(flags)));
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
sockfd.reset(-1);
- usleep(100000);
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -1952,14 +2097,11 @@ TEST_F(RdmaTest, client_alloc_resource_fail_fallback_tcp) {
req.set_sleep_us(200000);
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP,
- static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(RdmaTransport::RDMA_OFF,
- static_cast<RdmaTransport*>(s->_transport.get())->_rdma_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, RdmaTransportOf(s)->_rdma_state);
// The socket must not be failed, otherwise it can no longer carry TCP.
ASSERT_FALSE(s->Failed());
@@ -1981,11 +2123,9 @@ TEST_F(RdmaTest, server_alloc_resource_fail_fallback_tcp) {
butil::fd_guard sockfd(socket(AF_INET, SOCK_STREAM, 0));
ASSERT_TRUE(sockfd >= 0);
ASSERT_EQ(0, connect(sockfd, (sockaddr*)&addr, sizeof(sockaddr)));
- usleep(100000); // wait for server to handle the msg
- Socket* s = GetSocketFromServer(0);
+ Socket* s = WaitForServerSocket();
ASSERT_TRUE(s != nullptr);
- ASSERT_EQ(rdma::RdmaEndpoint::UNINIT,
- static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_EQ(rdma::RdmaEndpoint::UNINIT, RdmaTransportOf(s)->_rdma_ep->_state);
// Send a well-formed v2 hello: the negotiation succeeds
// but the resource allocation does not.
@@ -2004,24 +2144,18 @@ TEST_F(RdmaTest, server_alloc_resource_fail_fallback_tcp) {
msg.Serialize(data + 4);
ASSERT_EQ(rdma::HELLO_V2_MSG_LEN_MIN,
write(sockfd, data, rdma::HELLO_V2_MSG_LEN_MIN));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::S_ACK_WAIT,
- static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
- ASSERT_EQ(RdmaTransport::RDMA_OFF,
- static_cast<RdmaTransport*>(s->_transport.get())->_rdma_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::S_ACK_WAIT, RdmaTransportOf(s));
+ ASSERT_EQ(RdmaTransport::RDMA_OFF, RdmaTransportOf(s)->_rdma_state);
ASSERT_FALSE(s->Failed());
// Ack without RDMA so that the server finishes the handshake in TCP mode.
uint32_t flags = butil::HostToNet32(0);
ASSERT_EQ(sizeof(flags), write(sockfd, &flags, sizeof(flags)));
- usleep(100000);
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP,
- static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
ASSERT_FALSE(s->Failed());
sockfd.reset(-1);
- usleep(100000);
- ASSERT_EQ(nullptr, GetSocketFromServer(0));
+ ASSERT_TRUE(WaitForServerSocketGone());
StopServer();
}
@@ -2045,10 +2179,9 @@ TEST_F(RdmaTest, try_global_disable_rdma) {
req.set_sleep_us(200000);
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
- ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, static_cast<RdmaTransport*>(s->_transport.get())->_rdma_ep->_state);
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::FALLBACK_TCP, RdmaTransportOf(s));
bthread_id_join(cntl.call_id());
ASSERT_EQ(0, cntl.ErrorCode());
@@ -2145,6 +2278,73 @@ TEST_F(RdmaTest, channel_option_invalid) {
ASSERT_EQ(-1, channel.Init(g_ep, &chan_options));
}
+// Rounds, per-round RPC count and attachment sizes shared by the end-to-end
+// tests below. One RPC per test leaves everything that only shows up on the
+// second message untouched -- buffer reuse, an EOF read racing another writer
+// of the same input stream, resource recycling.
+static const int E2E_ROUND_NUM = 3;
+static const int E2E_RPC_NUM = 32;
+static const size_t E2E_ATTACH_SIZE[] = { 0, 4096, 128 * 1024 };
+
+static void ShutdownClientConnection(Controller& cntl) {
+ SocketUniquePtr s;
+ if (Socket::Address(cntl._single_server_id, &s) == 0) {
+ ::shutdown(s->fd(), SHUT_WR);
+ }
+}
+
+// Returns the number of RPCs that succeeded. A test that severs the connection
+// cannot predict which ones make it, but the ones that do must still be right,
+// so failures are tolerated here and the caller decides how many it demands.
+static int SendEchoRpcs(Channel& channel, int rpc_num, size_t attach_size,
+ const std::function<void(Controller&)>& disturb = nullptr,
+ int disturb_at = 0) {
+ std::vector<Controller> cntl(rpc_num);
+ std::vector<test::EchoRequest> req(rpc_num);
+ std::vector<test::EchoResponse> res(rpc_num);
+ std::vector<butil::IOBuf> attach(rpc_num);
+ for (int i = 0; i < rpc_num; ++i) {
+ req[i].set_message("hello");
+ req[i].set_code(i + 1);
+ if (attach_size > 0) {
+ EXPECT_EQ(0, attach[i].resize(
+ attach_size, static_cast<char>('a' + i % 26)));
+ cntl[i].request_attachment().append(attach[i]);
+ }
+ ::test::EchoService::Stub(&channel).Echo(&cntl[i], &req[i], &res[i], DoNothing());
+ if (disturb && i == disturb_at) {
+ disturb(cntl[i]);
+ }
+ }
+ int succeeded = 0;
+ for (int i = 0; i < rpc_num; ++i) {
+ bthread_id_join(cntl[i].call_id());
+ if (cntl[i].Failed()) {
+ continue;
+ }
+ ++succeeded;
+ EXPECT_EQ("MyEchoService", res[i].message()) << "rpc[" << i << "]";
+ EXPECT_EQ(1, res[i].code_list_size()) << "rpc[" << i << "]";
+ if (res[i].code_list_size() == 1) {
+ EXPECT_EQ(i + 1, res[i].code_list(0)) << "rpc[" << i << "]";
+ }
+ EXPECT_EQ(attach_size, cntl[i].response_attachment().size()) << "rpc[" << i << "]";
+ EXPECT_TRUE(attach[i].equals(cntl[i].response_attachment())) << "rpc[" << i << "]";
+ }
+ return succeeded;
+}
+
+static void SendEchoRpcsInRounds(Channel& channel) {
+ for (int round = 0; round < E2E_ROUND_NUM; ++round) {
+ for (size_t i = 0; i < arraysize(E2E_ATTACH_SIZE); ++i) {
+ ASSERT_EQ(E2E_RPC_NUM,
+ SendEchoRpcs(channel, E2E_RPC_NUM, E2E_ATTACH_SIZE[i]))
+ << "round=" << round
+ << " attach_size=" << E2E_ATTACH_SIZE[i];
+ }
+ }
+}
+
TEST_P(RdmaRpcTest, rdma_client_to_rdma_server) {
if (!FLAGS_rdma_test_enable) {
return;
@@ -2156,18 +2356,10 @@ TEST_P(RdmaRpcTest, rdma_client_to_rdma_server) {
ChannelOptions chan_options;
chan_options.socket_mode = SOCKET_MODE_RDMA;
chan_options.connect_timeout_ms = 500;
- chan_options.timeout_ms = 500;
+ chan_options.timeout_ms = 5000;
chan_options.max_retry = 0;
ASSERT_EQ(0, channel.Init(g_ep, &chan_options));
- Controller cntl;
- test::EchoRequest req;
- test::EchoResponse res;
- req.set_message(__FUNCTION__);
- google::protobuf::Closure* done = DoNothing();
- ::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- // usleep(100000);
- bthread_id_join(cntl.call_id());
- ASSERT_EQ(0, cntl.ErrorCode());
+ ASSERT_NO_FATAL_FAILURE(SendEchoRpcsInRounds(channel));
StopServer();
}
@@ -2178,18 +2370,10 @@ TEST_P(RdmaRpcTest, tcp_client_to_tcp_server) {
Channel channel;
ChannelOptions chan_options;
chan_options.connect_timeout_ms = 500;
- chan_options.timeout_ms = 500;
+ chan_options.timeout_ms = 5000;
chan_options.max_retry = 0;
ASSERT_EQ(0, channel.Init(g_ep, &chan_options));
- Controller cntl;
- test::EchoRequest req;
- test::EchoResponse res;
- req.set_message(__FUNCTION__);
- google::protobuf::Closure* done = DoNothing();
- ::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
- bthread_id_join(cntl.call_id());
- ASSERT_EQ(0, cntl.ErrorCode());
+ ASSERT_NO_FATAL_FAILURE(SendEchoRpcsInRounds(channel));
StopServer();
}
@@ -2200,18 +2384,10 @@ TEST_P(RdmaRpcTest, tcp_client_to_rdma_server) {
Channel channel;
ChannelOptions chan_options;
chan_options.connect_timeout_ms = 500;
- chan_options.timeout_ms = 500;
+ chan_options.timeout_ms = 5000;
chan_options.max_retry = 0;
ASSERT_EQ(0, channel.Init(g_ep, &chan_options));
- Controller cntl;
- test::EchoRequest req;
- test::EchoResponse res;
- req.set_message(__FUNCTION__);
- google::protobuf::Closure* done = DoNothing();
- ::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
- bthread_id_join(cntl.call_id());
- ASSERT_EQ(0, cntl.ErrorCode());
+ ASSERT_NO_FATAL_FAILURE(SendEchoRpcsInRounds(channel));
StopServer();
}
@@ -2223,18 +2399,73 @@ TEST_P(RdmaRpcTest, rdma_client_to_tcp_server) {
ChannelOptions chan_options;
chan_options.socket_mode = SOCKET_MODE_RDMA;
chan_options.connect_timeout_ms = 500;
- chan_options.timeout_ms = 500;
+ chan_options.timeout_ms = 5000;
chan_options.max_retry = 0;
ASSERT_EQ(0, channel.Init(g_ep, &chan_options));
- Controller cntl;
- test::EchoRequest req;
- test::EchoResponse res;
- req.set_message(__FUNCTION__);
- google::protobuf::Closure* done = DoNothing();
- ::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000);
- bthread_id_join(cntl.call_id());
- ASSERT_FALSE(cntl.Failed());
+ ASSERT_NO_FATAL_FAILURE(SendEchoRpcsInRounds(channel));
+
+ StopServer();
+}
+
+TEST_P(RdmaRpcTest, tcp_client_to_rdma_server_short_connection) {
+ StartServer();
+
+ Channel channel;
+ ChannelOptions chan_options;
+ chan_options.connect_timeout_ms = 1000;
+ chan_options.timeout_ms = 10000;
+ chan_options.max_retry = 0;
+ chan_options.connection_type = "short";
+ ASSERT_EQ(0, channel.Init(g_ep, &chan_options));
+ for (int round = 0; round < 8; ++round) {
+ ASSERT_EQ(E2E_RPC_NUM, SendEchoRpcs(channel, E2E_RPC_NUM, 4096))
+ << "round=" << round;
+ }
+
+ StopServer();
+}
+
+// Rounds of connection churn: a race needs attempts, not one well-timed shot.
+static const int CHURN_ROUND_NUM = 16;
+static const int CHURN_RPC_NUM = 64;
+static const size_t CHURN_ATTACH_SIZE = 32 * 1024;
+
+TEST_P(RdmaRpcTest, rdma_server_survives_connection_churn) {
+ StartServer();
+
+ ChannelOptions chan_options;
+ chan_options.connect_timeout_ms = 1000;
+ chan_options.timeout_ms = 3000;
+ chan_options.max_retry = 0;
+ const int served_before = g_echo_served.load(butil::memory_order_relaxed);
+ int succeeded = 0;
+ for (int round = 0; round < CHURN_ROUND_NUM; ++round) {
+ // A fresh Channel per round so the Socket is dropped from the socket
+ // map when the Channel dies, instead of the next round inheriting it.
+ Channel channel;
+ ASSERT_EQ(0, channel.Init(g_ep, &chan_options));
+ // Warm up before cutting. Without this the cut of round 0 lands on a
+ // connection that has not finished connecting yet, where fd() is still
+ // -1 and shutdown() is a silent no-op.
+ ASSERT_EQ(1, SendEchoRpcs(channel, 1, 0)) << "round=" << round;
+ succeeded += SendEchoRpcs(channel, CHURN_RPC_NUM, CHURN_ATTACH_SIZE,
+ ShutdownClientConnection,
+ round * CHURN_RPC_NUM / CHURN_ROUND_NUM);
+ ASSERT_FALSE(HasFailure()) << "round=" << round;
+ }
+
+ const int served = g_echo_served.load(butil::memory_order_relaxed) -
+ served_before - CHURN_ROUND_NUM;
+ LOG(INFO) << "server served " << served << " of "
+ << CHURN_ROUND_NUM * CHURN_RPC_NUM << " requests during the churn, "
+ << succeeded << " replies made it back";
+ ASSERT_GT(served, 0);
+
+ // The churn must leave the server able to serve a fresh connection, and
+ // serve it correctly.
+ Channel channel;
+ ASSERT_EQ(0, channel.Init(g_ep, &chan_options));
+ ASSERT_EQ(CHURN_RPC_NUM, SendEchoRpcs(channel, CHURN_RPC_NUM, CHURN_ATTACH_SIZE));
StopServer();
}
@@ -2602,6 +2833,46 @@ TEST_P(RdmaRpcTest, client_close_during_rpc) {
StopServer();
}
+TEST_P(RdmaRpcTest, rdma_client_close_during_rpc_repeatedly) {
+ if (!FLAGS_rdma_test_enable) {
+ return;
+ }
+
+ StartServer();
+
+ ChannelOptions chan_options;
+ chan_options.socket_mode = SOCKET_MODE_RDMA;
+ chan_options.connect_timeout_ms = 1000;
+ chan_options.timeout_ms = 3000;
+ chan_options.max_retry = 0;
+ const int served_before = g_echo_served.load(butil::memory_order_relaxed);
+ int succeeded = 0;
+ for (int round = 0; round < CHURN_ROUND_NUM; ++round) {
+ Channel channel;
+ ASSERT_EQ(0, channel.Init(g_ep, &chan_options));
+ // Warm up so the cut lands on an established RDMA connection.
+ ASSERT_EQ(1, SendEchoRpcs(channel, 1, 0)) << "round=" << round;
+ succeeded += SendEchoRpcs(channel, CHURN_RPC_NUM, CHURN_ATTACH_SIZE,
+ ShutdownClientConnection,
+ round * CHURN_RPC_NUM / CHURN_ROUND_NUM);
+ ASSERT_FALSE(HasFailure()) << "round=" << round;
+ }
+
+
+ const int served = g_echo_served.load(butil::memory_order_relaxed) -
+ served_before - CHURN_ROUND_NUM;
+ LOG(INFO) << "server served " << served << " of "
+ << CHURN_ROUND_NUM * CHURN_RPC_NUM << " requests during the churn, "
+ << succeeded << " replies made it back";
+ ASSERT_GT(served, 0);
+
+ Channel channel;
+ ASSERT_EQ(0, channel.Init(g_ep, &chan_options));
+ ASSERT_EQ(CHURN_RPC_NUM, SendEchoRpcs(channel, CHURN_RPC_NUM, CHURN_ATTACH_SIZE));
+
+ StopServer();
+}
+
TEST_P(RdmaRpcTest, verbs_error_handling) {
if (!FLAGS_rdma_test_enable) {
return;
@@ -2625,10 +2896,10 @@ TEST_P(RdmaRpcTest, verbs_error_handling) {
google::protobuf::Closure* done = DoNothing();
::test::EchoService::Stub(&channel).Echo(&cntl, &req, &res, done);
- usleep(100000); // wait for rdma handshake complete
-
SocketUniquePtr s;
ASSERT_EQ(0, Socket::Address(cntl._single_server_id, &s));
+ // The QP below only exists once the handshake is over.
+ ASSERT_RDMA_STATE(rdma::RdmaEndpoint::ESTABLISHED, RdmaTransportOf(s));
ibv_send_wr wr;
memset(&wr, 0, sizeof(wr));
ibv_sge sge;
@@ -2639,7 +2910,7 @@ TEST_P(RdmaRpcTest, verbs_error_handling) {
wr.sg_list = &sge;
wr.num_sge = 1;
ibv_send_wr* bad = nullptr;
- auto rdma_transport = static_cast<RdmaTransport*>(s->_transport.get());
+ auto rdma_transport = RdmaTransportOf(s);
ibv_post_send(rdma_transport->_rdma_ep->_resource->qp, &wr, &bad);
bthread_id_join(cntl.call_id());
ASSERT_EQ(ERDMA, cntl.ErrorCode());
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org