Copilot commented on code in PR #3645: URL: https://github.com/apache/brpc/pull/3645#discussion_r4366402333
########## src/brpc/handshake/handshake_io.cpp: ########## @@ -0,0 +1,162 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "brpc/handshake/handshake_io.h" + +#include <cstdint> +#include <errno.h> +#include <unistd.h> + +#include "bthread/butex.h" +#include "butil/time.h" +#include "brpc/errno.pb.h" +#include "brpc/socket.h" + +namespace brpc { +namespace handshake { + +size_t IOBufHandshakeInput::Size() const { + return _source != NULL ? _source->size() : 0; +} + +bool IOBufHandshakeInput::CopyTo(void* data, size_t len) const { + return _source != NULL && _source->copy_to(data, len) == len; +} + +bool IOBufHandshakeInput::Consume(size_t len) { + return _source != NULL && _source->pop_front(len) == len; +} + +static const int WAIT_TIMEOUT_MS = 50; + +SocketHandshakeIO::SocketHandshakeIO(Socket* socket) + : _socket(socket) + , _read_butex(bthread::butex_create_checked<butil::atomic<int> >()) { +} Review Comment: `butex_create_checked` returns pooled storage without initializing `Butex::value` (`src/bthread/butex.cpp`). In C++14 builds, the first `ReadExactLoop` load or `NotifyReadable` fetch-add therefore accesses an indeterminate atomic value, which is undefined behavior. Initialize `_read_butex` to zero here before either operation can run, as `UrmaEndpoint` already does. ########## src/brpc/urma_transport.cpp: ########## @@ -190,6 +182,62 @@ void UrmaTransport::Debug(std::ostream& os) { } } +int UrmaTransport::PrepareUpgradeResources(bool server_side) { + if (!_urma_ep || _urma_ep->AllocateResources() < 0) { + return -1; + } + // A bonding provider must import the peer before receives are posted. + if ((!server_side || !urma::IsUrmaBondingDevice()) && + _urma_ep->PostRecv(_urma_ep->_rq_size, + urma::FLAGS_urma_recv_zerocopy) < 0) { + return -1; + } + return 0; +} + +int UrmaTransport::NegotiateUpgradeResources( + const urma::ParsedHello& remote, bool server_side) { + CHECK(_urma_ep != nullptr); + _urma_ep->ApplyRemoteHello(remote); + if (_urma_ep->ImportPeer(remote) < 0) { + return -1; + } + if (server_side && urma::IsUrmaBondingDevice() && + _urma_ep->PostRecv(_urma_ep->_rq_size, + urma::FLAGS_urma_recv_zerocopy) < 0) { + return -1; + } + return 0; +} + +void UrmaTransport::ActivateUpgrade() { + CHECK(_urma_ep != nullptr); + _urma_state.store(URMA_ON, butil::memory_order_release); + _urma_ep->_state.store(urma::UrmaEndpoint::ESTABLISHED, + butil::memory_order_release); +} + +int UrmaTransport::StartUpgradeEvents() { + return _urma_ep ? _urma_ep->StartCqEvents() : -1; +} + +void UrmaTransport::DeactivateUpgrade() { + _urma_state.store(URMA_OFF, butil::memory_order_release); + if (_urma_ep) { + _urma_ep->_state.store(urma::UrmaEndpoint::FALLBACK_TCP, + butil::memory_order_release); + _urma_ep->DeallocateResources(); Review Comment: When a disabled server Hello or failed peer import triggers TCP fallback, `DeallocateResources()` destroys the provider resources but leaves `_rbuf` owning the posted receive buffers. These unused buffers remain pinned for the connection's TCP lifetime and can exhaust the shared URMA pool across many fallback connections. Call `Reset()` before restoring `FALLBACK_TCP` so the buffers and window bookkeeping are cleared too. ########## src/brpc/handshake/handshake_io.cpp: ########## @@ -0,0 +1,162 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "brpc/handshake/handshake_io.h" + +#include <cstdint> +#include <errno.h> +#include <unistd.h> + +#include "bthread/butex.h" +#include "butil/time.h" +#include "brpc/errno.pb.h" +#include "brpc/socket.h" + +namespace brpc { +namespace handshake { + +size_t IOBufHandshakeInput::Size() const { + return _source != NULL ? _source->size() : 0; +} + +bool IOBufHandshakeInput::CopyTo(void* data, size_t len) const { + return _source != NULL && _source->copy_to(data, len) == len; +} + +bool IOBufHandshakeInput::Consume(size_t len) { + return _source != NULL && _source->pop_front(len) == len; +} + +static const int WAIT_TIMEOUT_MS = 50; + +SocketHandshakeIO::SocketHandshakeIO(Socket* socket) + : _socket(socket) + , _read_butex(bthread::butex_create_checked<butil::atomic<int> >()) { +} + +SocketHandshakeIO::~SocketHandshakeIO() { + bthread::butex_destroy(_read_butex); +} + +void SocketHandshakeIO::Reset(Socket* socket) { + _socket = socket; +} + +void SocketHandshakeIO::NotifyReadable() { + _read_butex->fetch_add(1, butil::memory_order_release); + bthread::butex_wake(_read_butex); +} + +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) { + const int expected = read_butex->load(butil::memory_order_acquire); + const timespec duetime = butil::milliseconds_from_now(WAIT_TIMEOUT_MS); + const ssize_t nr = read_once(received, len - received); + if (nr < 0) { + if (errno == EINTR) { + continue; + } + 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; + } + } else if (nr == 0) { + errno = EEOF; + return -1; + } else { + received += nr; + } + } + return 0; +} + +int SocketHandshakeIO::ReadExact(void* data, size_t len) { + CHECK(data != NULL); + CHECK(_socket != NULL); + const int fd = _socket->fd(); + 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); + }); +} + +template <typename WriteOnce, typename WaitWritable> +static int WriteAllLoop(size_t len, WriteOnce write_once, + WaitWritable wait_writable) { + size_t written = 0; + while (written < len) { + const timespec duetime = butil::milliseconds_from_now(WAIT_TIMEOUT_MS); + const ssize_t nw = write_once(written, len - written); + if (nw > 0) { + written += nw; + continue; + } + if (nw == 0) { + errno = EPIPE; + return -1; + } + if (errno == EINTR) { + continue; + } + if (errno != EAGAIN) { + return -1; + } + if (wait_writable(&duetime) < 0 && + errno != ETIMEDOUT) { + return -1; + } + } + return 0; +} + +int SocketHandshakeIO::WriteAll(const void* data, size_t len) { + CHECK(data != NULL); + CHECK(_socket != NULL); + const int fd = _socket->fd(); + return WriteAllLoop(len, + [data, fd](size_t offset, size_t remaining) { + return write(fd, static_cast<const uint8_t*>(data) + offset, + remaining); + }, + [this](const timespec* duetime) { + return _socket->WaitEpollOut( + _socket->fd(), true, duetime); + }); Review Comment: If a handshake write returns EAGAIN and the Socket is then failed while the peer stops reading, this loop never terminates: SetFailed only wakes the waiter, and subsequent 50 ms timeouts are retried without checking failure. The handshake's Socket reference keeps the fd open, so this can also prevent recycling and health-check recovery. Check the Socket's failed state before waiting again, as the read loop already does. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
