Copilot commented on code in PR #3428: URL: https://github.com/apache/brpc/pull/3428#discussion_r3950264271
########## test/brpc_urma_unittest.cpp: ########## @@ -0,0 +1,614 @@ +// 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 <cstring> +#include <limits> +#include <gtest/gtest.h> +#include <gflags/gflags.h> + +#if BRPC_WITH_URMA +#include "butil/atomicops.h" +#include "butil/sys_byteorder.h" +#include "urma_api.h" +#include "brpc/urma/urma_handshake.h" +#include "brpc/urma/urma_handshake.pb.h" +#include "brpc/urma/urma_helper.h" +#include "urma_types.h" + +using namespace brpc; + +namespace brpc { +namespace urma { + +DECLARE_int32(urma_client_handshake_version); +extern bool g_skip_urma_init; +extern butil::atomic<bool> g_urma_available; + +} // namespace urma +} // namespace brpc + +// --------------------------------------------------------------------------- +// v2 binary HelloMessage: serialize + deserialize round-trips. +// --------------------------------------------------------------------------- +TEST(UrmaHandshakeTest, v2_serialize_deserialize_roundtrip) { + urma::v2_wire::HelloMessage m; + m.msg_len = urma::v2_wire::HELLO_PACKET_LEN; + m.hello_ver = urma::v2_wire::HELLO_V2_VERSION; + m.impl_ver = urma::v2_wire::IMPL_V2_VERSION; + m.buffer_size = 8192; + m.recv_buffer_cnt = 127; + m.jetty_id = 0x12345678; + for (int i = 0; i < 16; ++i) { + m.eid[i] = static_cast<uint8_t>(i + 1); + } + m.uasid = 0xdeadbeef; + m.tp_type = 1; // URMA_CTP + for (int i = 0; i < 16; ++i) { + m.seg_eid[i] = static_cast<uint8_t>(16 - i); + } + m.seg_uasid = 0xcafebabe; + m.seg_va = 0x1122334455667788ULL; + m.seg_len = 1ULL << 20; + m.seg_token_id = 0x42424242; + + uint8_t buf[urma::v2_wire::HELLO_BODY_LEN]; + m.Serialize(buf); + + urma::v2_wire::HelloMessage m2; + m2.Deserialize(buf); + EXPECT_EQ(m.msg_len, m2.msg_len); + EXPECT_EQ(m.hello_ver, m2.hello_ver); + EXPECT_EQ(m.impl_ver, m2.impl_ver); + EXPECT_EQ(m.buffer_size, m2.buffer_size); + EXPECT_EQ(m.recv_buffer_cnt, m2.recv_buffer_cnt); + EXPECT_EQ(m.jetty_id, m2.jetty_id); + EXPECT_EQ(0, memcmp(m.eid, m2.eid, 16)); + EXPECT_EQ(m.uasid, m2.uasid); + EXPECT_EQ(m.tp_type, m2.tp_type); + EXPECT_EQ(0, memcmp(m.seg_eid, m2.seg_eid, 16)); + EXPECT_EQ(m.seg_uasid, m2.seg_uasid); + EXPECT_EQ(m.seg_va, m2.seg_va); + EXPECT_EQ(m.seg_len, m2.seg_len); + EXPECT_EQ(m.seg_token_id, m2.seg_token_id); +} + +// --------------------------------------------------------------------------- +// v2 packet on the wire: "URMA" magic + body. +// --------------------------------------------------------------------------- +TEST(UrmaHandshakeTest, v2_packet_magic_is_urma) { + EXPECT_EQ(4u, urma::v2_wire::MAGIC_STR_LEN); + char magic[4] = {'U', 'R', 'M', 'A'}; + EXPECT_EQ(0, memcmp(magic, "URMA", 4)); + EXPECT_EQ(4u + 82u, urma::v2_wire::HELLO_PACKET_LEN); +} + +// --------------------------------------------------------------------------- +// v3 protobuf UrmaHello: serialize + parse round-trips. +// --------------------------------------------------------------------------- +TEST(UrmaHandshakeTest, v3_protobuf_roundtrip) { + urma::UrmaHello msg; + msg.set_buffer_size(8192); + msg.set_recv_buffer_cnt(127); + msg.set_jetty_id(0x12345678); + uint8_t eid[16]; + for (int i = 0; i < 16; ++i) { + eid[i] = static_cast<uint8_t>(i + 1); + } + msg.set_eid(eid, 16); + msg.set_uasid(0xdeadbeef); + msg.set_tp_type(1); + uint8_t seg_eid[16]; + for (int i = 0; i < 16; ++i) { + seg_eid[i] = static_cast<uint8_t>(16 - i); + } + msg.set_seg_eid(seg_eid, 16); + msg.set_seg_uasid(0xcafebabe); + msg.set_seg_va(0x1122334455667788ULL); + msg.set_seg_len(1ULL << 20); + msg.set_seg_token_id(0x42424242); + + std::string body; + ASSERT_TRUE(msg.SerializeToString(&body)); + urma::UrmaHello msg2; + ASSERT_TRUE(msg2.ParseFromString(body)); + EXPECT_EQ(msg.buffer_size(), msg2.buffer_size()); + EXPECT_EQ(msg.recv_buffer_cnt(), msg2.recv_buffer_cnt()); + EXPECT_EQ(msg.jetty_id(), msg2.jetty_id()); + EXPECT_EQ(16, msg2.eid().size()); + EXPECT_EQ(0, memcmp(msg.eid().data(), msg2.eid().data(), 16)); + EXPECT_EQ(msg.uasid(), msg2.uasid()); + EXPECT_EQ(msg.tp_type(), msg2.tp_type()); + EXPECT_EQ(16, msg2.seg_eid().size()); + EXPECT_EQ(msg.seg_uasid(), msg2.seg_uasid()); + EXPECT_EQ(msg.seg_va(), msg2.seg_va()); + EXPECT_EQ(msg.seg_len(), msg2.seg_len()); + EXPECT_EQ(msg.seg_token_id(), msg2.seg_token_id()); +} + +// --------------------------------------------------------------------------- +// CreateServerHandshakeByMagic dispatches on the magic bytes. +// --------------------------------------------------------------------------- +TEST(UrmaHandshakeTest, server_handshake_factory_dispatches_on_magic) { + // We cannot fully exercise the server handshake without a real socket + + // endpoint, but we can verify the factory returns the right protocol + // version for each magic, and nullptr for an unknown magic. + uint8_t magic_v2[4] = {'U', 'R', 'M', 'A'}; + uint8_t magic_v3[4] = {'U', 'R', 'M', '3'}; + uint8_t magic_bad[4] = {'P', 'R', 'P', 'C'}; + + // v2 magic -> protocol version 2 + urma::UrmaHandshake* hs2 = + urma::CreateServerHandshakeByMagic(nullptr, magic_v2); + // Note: the factory dereferences the endpoint only inside SendLocalHello / + // ReceiveAndParseRemoteHello; passing nullptr is safe for the version query. + // (We delete immediately to avoid touching the endpoint.) + if (hs2) { + EXPECT_EQ(2, hs2->ProtocolVersion()); + delete hs2; + } + // v3 magic -> protocol version 3 + urma::UrmaHandshake* hs3 = + urma::CreateServerHandshakeByMagic(nullptr, magic_v3); + if (hs3) { + EXPECT_EQ(3, hs3->ProtocolVersion()); + delete hs3; + } + // unknown magic -> nullptr (caller falls back to TCP) + urma::UrmaHandshake* hsb = + urma::CreateServerHandshakeByMagic(nullptr, magic_bad); + EXPECT_EQ(nullptr, hsb); +} + +// --------------------------------------------------------------------------- +// CreateClientHandshake picks the version from the gflag. +// --------------------------------------------------------------------------- +TEST(UrmaHandshakeTest, client_handshake_factory_respects_flag) { + const int saved = urma::FLAGS_urma_client_handshake_version; + + urma::FLAGS_urma_client_handshake_version = 2; + urma::UrmaHandshake* hs2 = urma::CreateClientHandshake(nullptr); + if (hs2) { + EXPECT_EQ(2, hs2->ProtocolVersion()); + delete hs2; + } + + urma::FLAGS_urma_client_handshake_version = 3; + urma::UrmaHandshake* hs3 = urma::CreateClientHandshake(nullptr); + if (hs3) { + EXPECT_EQ(3, hs3->ProtocolVersion()); + delete hs3; + } + + urma::FLAGS_urma_client_handshake_version = saved; +} Review Comment: This test mutates a gflag (FLAGS_urma_client_handshake_version) and restores it manually at the end. Using GFLAGS_NAMESPACE::FlagSaver is the established pattern in this repo to guarantee restoration even on early test exit/failure, and avoids leaking flag state into subsequent tests. ########## src/brpc/urma_transport.cpp: ########## @@ -0,0 +1,248 @@ +// 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/urma_transport.h" + +#if BRPC_WITH_URMA + +#include <gflags/gflags.h> + +#include "butil/iobuf.h" +#include "butil/logging.h" +#include "bthread/bthread.h" +#include "bthread/types.h" + +#include "brpc/event_dispatcher.h" +#include "brpc/input_messenger.h" +#include "brpc/socket.h" +#include "brpc/tcp_transport.h" +#include "brpc/urma/urma_endpoint.h" +#include "brpc/urma/urma_helper.h" + +namespace brpc { + +// Defined in urma_helper.cpp. +DECLARE_bool(urma_use_polling); +DECLARE_bool(urma_disable_bthread); + +void UrmaTransport::Init(Socket* socket, const SocketOptions& options) { + CHECK(_urma_ep == nullptr); + if (options.socket_mode == SOCKET_MODE_URMA) { + _urma_ep = new (std::nothrow) urma::UrmaEndpoint(socket); + if (!_urma_ep) { + const int saved_errno = errno; + PLOG(ERROR) << "Fail to create UrmaEndpoint"; + socket->SetFailed(saved_errno, "Fail to create UrmaEndpoint: %s", + berror(saved_errno)); + } + _urma_state = URMA_UNKNOWN; + } else { + _urma_state = URMA_OFF; + socket->_socket_mode = SOCKET_MODE_TCP; + } + _socket = socket; + _default_connect = options.app_connect; + _on_edge_trigger = options.on_edge_triggered_events; + if (options.need_on_edge_trigger && _on_edge_trigger == nullptr) { + _on_edge_trigger = urma::UrmaEndpoint::OnNewDataFromTcp; + } + _tcp_transport = std::make_shared<TcpTransport>(); + _tcp_transport->Init(socket, options); +} + +void UrmaTransport::Release() { + if (_urma_ep) { + delete _urma_ep; + _urma_ep = nullptr; + } +} + +int UrmaTransport::Reset(int32_t /*expected_nref*/) { + if (_urma_ep) { + _urma_ep->Reset(); + } + _urma_state = URMA_UNKNOWN; + return 0; +} + +std::shared_ptr<AppConnect> UrmaTransport::Connect() { + if (_default_connect == nullptr) { + return std::make_shared<urma::UrmaConnect>(); + } + return _default_connect; +} + +int UrmaTransport::CutFromIOBuf(butil::IOBuf* buf) { + if (_urma_ep && + _urma_state.load(butil::memory_order_acquire) != URMA_OFF) { + butil::IOBuf* data_arr[1] = {buf}; + return _urma_ep->CutFromIOBufList(data_arr, 1); + } else { + return _tcp_transport->CutFromIOBuf(buf); + } +} Review Comment: CutFromIOBuf() routes data to UrmaEndpoint while _urma_state is URMA_UNKNOWN. In that state, UrmaEndpoint::CutFromIOBufList() can return ENOTCONN because resources/remote_jetty are not established yet. This should mirror RdmaTransport and only use the URMA data path after negotiation resolves to URMA_ON; otherwise fall back to TCP until the handshake completes. This issue also appears on line 99 of the same file. ########## src/brpc/input_messenger.cpp: ########## @@ -56,294 +54,291 @@ BRPC_VALIDATE_GFLAG(log_connection_close, PassValidate); DEFINE_bool(socket_keepalive, false, "Enable keepalive of sockets if this value is true"); -DEFINE_int32(socket_keepalive_idle_s, -1, - "Set idle time for socket keepalive in seconds if this value is positive"); +DEFINE_int32( + socket_keepalive_idle_s, -1, + "Set idle time for socket keepalive in seconds if this value is positive"); -DEFINE_int32(socket_keepalive_interval_s, -1, - "Set interval between keepalives in seconds if this value is positive"); +DEFINE_int32( + socket_keepalive_interval_s, -1, + "Set interval between keepalives in seconds if this value is positive"); DEFINE_int32(socket_keepalive_count, -1, "Set number of keepalives before death if this value is positive"); -DEFINE_int32(socket_tcp_user_timeout_ms, -1, - "If this value is positive, set number of milliseconds that transmitted " - "data may remain unacknowledged, or bufferred data may remain untransmitted " - "(due to zero window size) before TCP will forcibly close the corresponding " - "connection and return ETIMEDOUT to the application. Only linux supports " - "TCP_USER_TIMEOUT."); +DEFINE_int32( + socket_tcp_user_timeout_ms, -1, + "If this value is positive, set number of milliseconds that transmitted " + "data may remain unacknowledged, or bufferred data may remain " + "untransmitted " + "(due to zero window size) before TCP will forcibly close the " + "corresponding " + "connection and return ETIMEDOUT to the application. Only linux supports " + "TCP_USER_TIMEOUT."); DECLARE_bool(usercode_in_pthread); DECLARE_bool(usercode_in_coroutine); -void* ProcessInputMessage(void* void_arg) { - InputMessageBase* msg = static_cast<InputMessageBase*>(void_arg); - msg->_process(msg); - return nullptr; +void *ProcessInputMessage(void *void_arg) { + InputMessageBase *msg = static_cast<InputMessageBase *>(void_arg); + msg->_process(msg); + return nullptr; } struct RunLastMessage { - inline void operator()(InputMessageBase* last_msg) { - ProcessInputMessage(last_msg); - } + inline void operator()(InputMessageBase *last_msg) { + ProcessInputMessage(last_msg); + } }; InputMessageClosure::~InputMessageClosure() noexcept(false) { - if (_msg) { - ProcessInputMessage(_msg); - } + if (_msg) { + ProcessInputMessage(_msg); + } } -void InputMessageClosure::reset(InputMessageBase* m) { - if (_msg) { - ProcessInputMessage(_msg); - } - _msg = m; +void InputMessageClosure::reset(InputMessageBase *m) { + if (_msg) { + ProcessInputMessage(_msg); + } + _msg = m; } -void InputMessenger::OnNewMessages(Socket* m) { - // Notes: - // - If the socket has only one message, the message will be parsed and - // processed in this bthread. nova-pbrpc and http works in this way. - // - If the socket has several messages, all messages will be parsed ( - // meaning cutting from butil::IOBuf. serializing from protobuf is part of - // "process") in this bthread. All messages except the last one will be - // processed in separate bthreads. To minimize the overhead, scheduling - // is batched(notice the BTHREAD_NOSIGNAL and bthread_flush). - // - Verify will always be called in this bthread at most once and before - // any process. - InputMessengerProcessor& processor = m->fd_input_processor(); - int progress = Socket::PROGRESS_INIT; - - // Notice that all *return* no matter successful or not will run last - // message, even if the socket is about to be closed. This should be - // OK in most cases. - InputMessageClosure last_msg; - 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; - - // 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 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) { - if (errno == EINTR) { - continue; // just retry - } - const int saved_errno = errno; - PLOG(WARNING) << "Fail to read from " << *m; - m->SetFailed(saved_errno, "Fail to read from %s: %s", - m->description().c_str(), berror(saved_errno)); - return; - } else if (!m->MoreReadEvents(&progress)) { - return; - } else { // new events during processing - continue; - } - } - - if (processor.ProcessNewMessage(nr, read_eof, received_us, - base_realtime, last_msg) < 0) { - return; +void InputMessenger::OnNewMessages(Socket *m) { + // Notes: + // - If the socket has only one message, the message will be parsed and + // processed in this bthread. nova-pbrpc and http works in this way. + // - If the socket has several messages, all messages will be parsed ( + // meaning cutting from butil::IOBuf. serializing from protobuf is part of + // "process") in this bthread. All messages except the last one will be + // processed in separate bthreads. To minimize the overhead, scheduling + // is batched(notice the BTHREAD_NOSIGNAL and bthread_flush). + // - Verify will always be called in this bthread at most once and before + // any process. + InputMessengerProcessor &processor = m->fd_input_processor(); + int progress = Socket::PROGRESS_INIT; + + // Notice that all *return* no matter successful or not will run last + // message, even if the socket is about to be closed. This should be + // OK in most cases. + InputMessageClosure last_msg; + 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; + + // 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 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) { + if (errno == EINTR) { + continue; // just retry } + const int saved_errno = errno; + PLOG(WARNING) << "Fail to read from " << *m; + m->SetFailed(saved_errno, "Fail to read from %s: %s", + m->description().c_str(), berror(saved_errno)); + return; + } else if (!m->MoreReadEvents(&progress)) { + return; + } else { // new events during processing + continue; + } } - if (read_eof) { - m->SetEOF(); + if (processor.ProcessNewMessage(nr, read_eof, received_us, base_realtime, + last_msg) < 0) { + return; } + } + + if (read_eof) { + m->SetEOF(); + } } InputMessenger::InputMessenger(size_t capacity) - : _handlers(nullptr) - , _max_index(-1) - , _non_protocol(false) - , _capacity(capacity) { -} + : _handlers(nullptr), _max_index(-1), _non_protocol(false), + _capacity(capacity) {} InputMessenger::~InputMessenger() { - delete[] _handlers; - _handlers = nullptr; - _max_index.store(-1, butil::memory_order_relaxed); - _capacity = 0; + delete[] _handlers; + _handlers = nullptr; + _max_index.store(-1, butil::memory_order_relaxed); + _capacity = 0; } -int InputMessenger::AddHandler(const InputMessageHandler& handler) { - if (handler.parse == nullptr || handler.process == nullptr - || handler.name == nullptr) { - CHECK(false) << "Invalid argument"; - return -1; - } - BAIDU_SCOPED_LOCK(_add_handler_mutex); - if (nullptr == _handlers) { - _handlers = new InputMessageHandler[_capacity]; - memset(_handlers, 0, sizeof(*_handlers) * _capacity); - _non_protocol = false; - } - if (_non_protocol) { - CHECK(false) << "AddNonProtocolHandler was invoked"; - return -1; - } - ProtocolType type = FindProtocolOfHandler(handler); - if (type == PROTOCOL_UNKNOWN) { - CHECK(false) << "Adding a handler which doesn't belong to any protocol"; - return -1; - } - const int index = type; - if (index >= (int)_capacity) { - LOG(FATAL) << "Can't add more handlers than " << _capacity; - return -1; - } - if (_handlers[index].parse == nullptr) { - // The same protocol might be added more than twice - _handlers[index] = handler; - } else if (_handlers[index].parse != handler.parse - || _handlers[index].process != handler.process) { - CHECK(_handlers[index].parse == handler.parse); - CHECK(_handlers[index].process == handler.process); - return -1; - } - if (index > _max_index.load(butil::memory_order_relaxed)) { - _max_index.store(index, butil::memory_order_release); - } - return 0; -} - -int InputMessenger::AddNonProtocolHandler(const InputMessageHandler& handler) { - if (handler.parse == nullptr || handler.process == nullptr - || handler.name == nullptr) { - CHECK(false) << "Invalid argument"; - return -1; - } - BAIDU_SCOPED_LOCK(_add_handler_mutex); - if (nullptr == _handlers) { - _handlers = new InputMessageHandler[_capacity]; - memset(_handlers, 0, sizeof(*_handlers) * _capacity); - _non_protocol = true; - } - if (!_non_protocol) { - CHECK(false) << "AddHandler was invoked"; - return -1; - } - const int index = _max_index.load(butil::memory_order_relaxed) + 1; +int InputMessenger::AddHandler(const InputMessageHandler &handler) { + if (handler.parse == nullptr || handler.process == nullptr || + handler.name == nullptr) { + CHECK(false) << "Invalid argument"; + return -1; + } + BAIDU_SCOPED_LOCK(_add_handler_mutex); + if (nullptr == _handlers) { + _handlers = new InputMessageHandler[_capacity]; + memset(_handlers, 0, sizeof(*_handlers) * _capacity); + _non_protocol = false; + } + if (_non_protocol) { + CHECK(false) << "AddNonProtocolHandler was invoked"; + return -1; + } + ProtocolType type = FindProtocolOfHandler(handler); + if (type == PROTOCOL_UNKNOWN) { + CHECK(false) << "Adding a handler which doesn't belong to any protocol"; + return -1; + } + const int index = type; + if (index >= (int)_capacity) { + LOG(FATAL) << "Can't add more handlers than " << _capacity; + return -1; + } + if (_handlers[index].parse == nullptr) { + // The same protocol might be added more than twice _handlers[index] = handler; + } else if (_handlers[index].parse != handler.parse || + _handlers[index].process != handler.process) { + CHECK(_handlers[index].parse == handler.parse); + CHECK(_handlers[index].process == handler.process); + return -1; + } + if (index > _max_index.load(butil::memory_order_relaxed)) { _max_index.store(index, butil::memory_order_release); - return 0; + } + return 0; } -int InputMessenger::Create(const butil::EndPoint& remote_side, - time_t health_check_interval_s, - SocketId* id) { - SocketOptions options; - options.remote_side = remote_side; - options.user = this; - options.on_edge_triggered_events = OnNewMessages; - options.health_check_interval_s = health_check_interval_s; - if (FLAGS_socket_keepalive) { - options.keepalive_options = std::make_shared<SocketKeepaliveOptions>(); - options.keepalive_options->keepalive_idle_s - = FLAGS_socket_keepalive_idle_s; - options.keepalive_options->keepalive_interval_s - = FLAGS_socket_keepalive_interval_s; - options.keepalive_options->keepalive_count - = FLAGS_socket_keepalive_count; - } - options.tcp_user_timeout_ms = FLAGS_socket_tcp_user_timeout_ms; - return Socket::Create(options, id); +int InputMessenger::AddNonProtocolHandler(const InputMessageHandler &handler) { + if (handler.parse == nullptr || handler.process == nullptr || + handler.name == nullptr) { + CHECK(false) << "Invalid argument"; + return -1; + } + BAIDU_SCOPED_LOCK(_add_handler_mutex); + if (nullptr == _handlers) { + _handlers = new InputMessageHandler[_capacity]; + memset(_handlers, 0, sizeof(*_handlers) * _capacity); + _non_protocol = true; + } + if (!_non_protocol) { + CHECK(false) << "AddHandler was invoked"; + return -1; + } + const int index = _max_index.load(butil::memory_order_relaxed) + 1; + _handlers[index] = handler; + _max_index.store(index, butil::memory_order_release); + return 0; Review Comment: AddNonProtocolHandler() increments _max_index and writes into _handlers without checking that the new index is still within _capacity. This can write past the allocated handler array if enough non-protocol handlers are registered. -- 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]
