fix: address latest transport handshake review feedback

src/brpc/adapter_transport.cpp

@@ -17,7 +17,9 @@
 
 #include "brpc/adapter_transport.h"
 
+#include <algorithm>
 #include <cstdint>
+#include <cstring>
 #include <errno.h>
 #include <unistd.h>
 
@@ -40,6 +42,12 @@ namespace brpc {
 
 namespace {
 
+bool MatchesMagicPrefix(const char *prefix, size_t prefix_len,
+                        const char *magic, size_t magic_len) {
+    const size_t compare_len = std::min(prefix_len, magic_len);
+    return memcmp(prefix, magic, compare_len) == 0;
+}
+
 class AdapterConnect : public AppConnect {
 public:
     explicit AdapterConnect(const std::shared_ptr<AppConnect>& app_connect)
@@ -161,13 +169,26 @@ ParseResult AdapterTransport::ProcessUpgradeReadable(butil::IOBuf* source) {
                 _socket->parsing_context());
         CHECK(context->adapter() != NULL);
         result = context->adapter()->ExecuteServerHandshake(source, _socket);
-    } else {
-        const char* first = static_cast<const char*>(source->fetch1());
-        handshake::HandshakeAdapter* adapter =
-            first != NULL && *first == 'U'
-            ? handshake::GetUBShmServerHandshakeAdapter()
-            : handshake::GetRdmaServerHandshakeAdapter();
-        result = adapter->ExecuteServerHandshake(source, _socket);
+    } else if (!source->empty()) {
+        static const size_t MAX_MAGIC_LEN = 4;
+        char prefix[MAX_MAGIC_LEN] = {};
+        const size_t prefix_len = std::min(source->size(), MAX_MAGIC_LEN);
+        source->copy_to(prefix, prefix_len);
+
+        const bool matches_ub =
+            MatchesMagicPrefix(prefix, prefix_len, "UB", 2);
+        const bool matches_rdma =
+            MatchesMagicPrefix(prefix, prefix_len, "RDMA", 4) ||
+            MatchesMagicPrefix(prefix, prefix_len, "RDM3", 4);
+        if (!matches_ub && !matches_rdma) {
+            result = ParseResult(PARSE_ERROR_TRY_OTHERS);
+        } else {
+            handshake::HandshakeAdapter* adapter =
+                matches_ub
+                ? handshake::GetUBShmServerHandshakeAdapter()
+                : handshake::GetRdmaServerHandshakeAdapter();
+            result = adapter->ExecuteServerHandshake(source, _socket);
+        }
     }
     const int phase = _handshake.phase();
     if (!connection_completed() &&
@@ -560,6 +581,9 @@ void AdapterTransport::CheckUnexpectedTcpData() {
                                _socket->description().c_str());
             return;
         }
+        if (errno == EINTR) {
+            continue;
+        }
         if (errno != EAGAIN) {
             const int saved_errno = errno;
             _socket->SetFailed(saved_errno, "Fail to read from %s: %s",

src/brpc/handshake/handshake_io.cpp

@@ -63,6 +63,7 @@ void SocketHandshakeIO::NotifyReadable() {
 
 template <typename ReadOnce>
 static int ReadExactLoop(butil::atomic<int>* read_butex,
+                         SocketId socket_id,
                          size_t len, ReadOnce read_once) {
     size_t received = 0;
     while (received < len) {
@@ -76,6 +77,11 @@ static int ReadExactLoop(butil::atomic<int>* read_butex,
             if (errno != EAGAIN) {
                 return -1;
             }
+            SocketUniquePtr alive;
+            if (Socket::Address(socket_id, &alive) != 0) {
+                errno = EFAILEDSOCKET;
+                return -1;
+            }
             if (bthread::butex_wait(read_butex, expected, &duetime) < 0 &&
                 errno != EWOULDBLOCK && errno != ETIMEDOUT) {
                 return -1;
@@ -94,7 +100,7 @@ int SocketHandshakeIO::ReadExact(void* data, size_t len) {
     CHECK(data != NULL);
     CHECK(_socket != NULL);
     const int fd = _socket->fd();
-    return ReadExactLoop(_read_butex, len,
+    return ReadExactLoop(_read_butex, _socket->id(), len,
         [data, fd](size_t offset, size_t remaining) {
             return read(fd, static_cast<uint8_t*>(data) + offset, remaining);
         });
@@ -116,7 +122,7 @@ static int WriteAllLoop(size_t len, WriteOnce write_once,
             return -1;
         }
         if (errno == EINTR) {
-                       continue;
+            continue;
         }
         if (errno != EAGAIN) {
             return -1;