Make RdmaEndpoint::_state atomic to fix data race on TCP fallback

[chenBright]

What problem does this PR solve?

Issue Number: resolve

Problem Summary:

RdmaEndpoint::_state is a plain (non-atomic) enum, but it is written
by the RDMA handshake bthread and read concurrently by the
event dispatcher bthread in OnNewDataFromTcp. This is a data race,
and on a weak memory model (ARM / POWER / RISC-V) it can let the two
threads concurrently mutate _socket->_read_buf and corrupt an IOBuf.

time   bthread A (OnNewDataFromTcp)                    bthread B (ProcessHandshakeAtServer)
────   ─────────────────────────                       ─────────────────────────────────────
 T0                                                    Program order:
                                                         _read_buf.append(magic, ...)
                                                         _state = FALLBACK_TCP
                                                         TryReadOnTcp()  (fetch_add acq_rel)

       ═════════════════════════════ Actual order after CPU reordering ══════════════════════

 T1                                                    ep->_state = FALLBACK_TCP
                                                         ← hoisted: this store takes effect FIRST
                                                         (_nevent is still 0,
                                                          _read_buf still has no magic in it)

 T2   (new bytes arrive on the fd,
        epoll edge-trigger fires)
       Socket::StartInputEvent
         _nevent.fetch_add(acq_rel): 0 → 1, == 0
         → spawn bthread A: OnEdge → OnNewDataFromTcp

 T3   read ep->_state (plain read, no acquire)
       => observes FALLBACK_TCP ✓ (T1 is visible)

 T4   else if (_state == FALLBACK_TCP) ✓
       → InputMessenger::OnNewMessages(m)
         └─ DoRead →
              _read_buf.append_from_file_descriptor(fd, ...)
         ⚠ A is now WRITING into _read_buf

 T5                                                    s->_read_buf.append(magic, MAGIC_STR_LEN)
                                                         ← reordered: this append happens NOW
                                                         ⚠⚠⚠ A and B are concurrently mutating
                                                              the SAME IOBuf object!

                                                       Consequences:
                                                         • IOBuf internal linked list /
                                                           _ref_array / _data pointer / refcount
                                                           are mutated concurrently
                                                         • memory corruption, use-after-free,
                                                           refcount drift, crash
                                                         • or — more insidious — the bytes
                                                           appear to be accepted but the
                                                           internal state of _read_buf is
                                                           already corrupted

What is changed and the side effects?

Changed:

Make _state a butil::atomic<State>.

• Terminal-state stores use memory_order_release and the corresponding loads use
  memory_order_acquire, so that everything published before a terminal state becomes
  visible to the reader:
  • the magic bytes appended back into _socket->_read_buf before FALLBACK_TCP;
  • the RDMA window/resource setup done before ESTABLISHED.
• All other (non-terminal) handshake transitions use memory_order_relaxed.

Side effects:

• Performance effects:

• Breaking backward compatibility:

---

Check List:

• Please make sure your changes are compilable.
• When providing us with a new feature, it is best to add related tests.
• Please follow Contributor Covenant Code of Conduct.

[Copilot]

Pull request overview

This PR addresses a concurrency bug in the RDMA/TCP handshake path by making RdmaEndpoint::_state atomic and introducing explicit acquire/release semantics so the TCP event thread can safely observe handshake terminal states (e.g., FALLBACK_TCP, ESTABLISHED) along with the data they’re intended to “publish” (such as the re-appended magic bytes on TCP fallback).

Changes:

• Convert RdmaEndpoint::_state from a plain State enum to butil::atomic<State>.
• Replace direct reads/writes of _state with load()/store() and apply acquire/release semantics on key transitions and reads.
• Adjust TCP read/dispatch logic (OnNewDataFromTcp, TryReadOnTcp, debug/state reporting) to use atomic loads.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.

File	Description
src/brpc/rdma/rdma_endpoint.h	Changes _state to an atomic and updates the TryReadOnTcp declaration accordingly.
src/brpc/rdma/rdma_endpoint.cpp	Updates all _state accesses to atomic operations and adds acquire/release ordering around handshake terminal states and TCP fallback/established paths.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.