+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-
-#include "ProtocolV1.h"
-
-#include <seastar/core/shared_future.hh>
-#include <seastar/core/sleep.hh>
-#include <seastar/net/packet.hh>
-
-#include "include/msgr.h"
-#include "include/random.h"
-#include "auth/Auth.h"
-#include "auth/AuthSessionHandler.h"
-
-#include "crimson/auth/AuthClient.h"
-#include "crimson/auth/AuthServer.h"
-#include "crimson/common/log.h"
-#include "chained_dispatchers.h"
-#include "Errors.h"
-#include "Socket.h"
-#include "SocketConnection.h"
-#include "SocketMessenger.h"
-
-WRITE_RAW_ENCODER(ceph_msg_connect);
-WRITE_RAW_ENCODER(ceph_msg_connect_reply);
-
-using crimson::common::local_conf;
-
-std::ostream& operator<<(std::ostream& out, const ceph_msg_connect& c)
-{
- return out << "connect{features=" << std::hex << c.features << std::dec
- << " host_type=" << c.host_type
- << " global_seq=" << c.global_seq
- << " connect_seq=" << c.connect_seq
- << " protocol_version=" << c.protocol_version
- << " authorizer_protocol=" << c.authorizer_protocol
- << " authorizer_len=" << c.authorizer_len
- << " flags=" << std::hex << static_cast<uint16_t>(c.flags) << std::dec << '}';
-}
-
-std::ostream& operator<<(std::ostream& out, const ceph_msg_connect_reply& r)
-{
- return out << "connect_reply{tag=" << static_cast<uint16_t>(r.tag)
- << " features=" << std::hex << r.features << std::dec
- << " global_seq=" << r.global_seq
- << " connect_seq=" << r.connect_seq
- << " protocol_version=" << r.protocol_version
- << " authorizer_len=" << r.authorizer_len
- << " flags=" << std::hex << static_cast<uint16_t>(r.flags) << std::dec << '}';
-}
-
-namespace {
-
-seastar::logger& logger() {
- return crimson::get_logger(ceph_subsys_ms);
-}
-
-template <typename T>
-seastar::net::packet make_static_packet(const T& value) {
- return { reinterpret_cast<const char*>(&value), sizeof(value) };
-}
-
-// store the banner in a non-const string for buffer::create_static()
-char banner[] = CEPH_BANNER;
-constexpr size_t banner_size = sizeof(CEPH_BANNER)-1;
-
-constexpr size_t client_header_size = banner_size + sizeof(ceph_entity_addr);
-constexpr size_t server_header_size = banner_size + 2 * sizeof(ceph_entity_addr);
-
-// check that the buffer starts with a valid banner without requiring it to
-// be contiguous in memory
-void validate_banner(bufferlist::const_iterator& p)
-{
- auto b = std::cbegin(banner);
- auto end = b + banner_size;
- while (b != end) {
- const char *buf{nullptr};
- auto remaining = std::distance(b, end);
- auto len = p.get_ptr_and_advance(remaining, &buf);
- if (!std::equal(buf, buf + len, b)) {
- throw std::system_error(
- make_error_code(crimson::net::error::bad_connect_banner));
- }
- b += len;
- }
-}
-
-// return a static bufferptr to the given object
-template <typename T>
-bufferptr create_static(T& obj)
-{
- return buffer::create_static(sizeof(obj), reinterpret_cast<char*>(&obj));
-}
-
-uint32_t get_proto_version(entity_type_t peer_type, bool connect)
-{
- constexpr entity_type_t my_type = CEPH_ENTITY_TYPE_OSD;
- // see also OSD.h, unlike other connection of simple/async messenger,
- // crimson msgr is only used by osd
- constexpr uint32_t CEPH_OSD_PROTOCOL = 10;
- if (peer_type == my_type) {
- // internal
- return CEPH_OSD_PROTOCOL;
- } else {
- // public
- switch (connect ? peer_type : my_type) {
- case CEPH_ENTITY_TYPE_OSD: return CEPH_OSDC_PROTOCOL;
- case CEPH_ENTITY_TYPE_MDS: return CEPH_MDSC_PROTOCOL;
- case CEPH_ENTITY_TYPE_MON: return CEPH_MONC_PROTOCOL;
- default: return 0;
- }
- }
-}
-
-void discard_up_to(std::deque<MessageRef>* queue,
- crimson::net::seq_num_t seq)
-{
- while (!queue->empty() &&
- queue->front()->get_seq() < seq) {
- queue->pop_front();
- }
-}
-
-} // namespace anonymous
-
-namespace crimson::net {
-
-ProtocolV1::ProtocolV1(ChainedDispatchers& dispatchers,
- SocketConnection& conn,
- SocketMessenger& messenger)
- : Protocol(proto_t::v1, dispatchers, conn), messenger{messenger} {}
-
-ProtocolV1::~ProtocolV1() {}
-
-bool ProtocolV1::is_connected() const
-{
- return state == state_t::open;
-}
-
-// connecting state
-
-void ProtocolV1::reset_session()
-{
- conn.out_q = {};
- conn.sent = {};
- conn.in_seq = 0;
- h.connect_seq = 0;
- if (HAVE_FEATURE(conn.features, MSG_AUTH)) {
- // Set out_seq to a random value, so CRC won't be predictable.
- // Constant to limit starting sequence number to 2^31. Nothing special
- // about it, just a big number.
- constexpr uint64_t SEQ_MASK = 0x7fffffff;
- conn.out_seq = ceph::util::generate_random_number<uint64_t>(0, SEQ_MASK);
- } else {
- // previously, seq #'s always started at 0.
- conn.out_seq = 0;
- }
-}
-
-seastar::future<stop_t>
-ProtocolV1::handle_connect_reply(msgr_tag_t tag)
-{
- if (h.auth_payload.length() && !conn.peer_is_mon()) {
- if (tag == CEPH_MSGR_TAG_CHALLENGE_AUTHORIZER) { // more
- h.auth_more = messenger.get_auth_client()->handle_auth_reply_more(
- conn.shared_from_this(), auth_meta, h.auth_payload);
- return seastar::make_ready_future<stop_t>(stop_t::no);
- } else {
- int ret = messenger.get_auth_client()->handle_auth_done(
- conn.shared_from_this(), auth_meta, 0, 0, h.auth_payload);
- if (ret < 0) {
- // fault
- logger().warn("{} AuthClient::handle_auth_done() return {}", conn, ret);
- throw std::system_error(make_error_code(error::negotiation_failure));
- }
- }
- }
-
- switch (tag) {
- case CEPH_MSGR_TAG_FEATURES:
- logger().error("{} connect protocol feature mispatch", __func__);
- throw std::system_error(make_error_code(error::negotiation_failure));
- case CEPH_MSGR_TAG_BADPROTOVER:
- logger().error("{} connect protocol version mispatch", __func__);
- throw std::system_error(make_error_code(error::negotiation_failure));
- case CEPH_MSGR_TAG_BADAUTHORIZER:
- logger().error("{} got bad authorizer", __func__);
- throw std::system_error(make_error_code(error::negotiation_failure));
- case CEPH_MSGR_TAG_RESETSESSION:
- reset_session();
- return seastar::make_ready_future<stop_t>(stop_t::no);
- case CEPH_MSGR_TAG_RETRY_GLOBAL:
- return messenger.get_global_seq(h.reply.global_seq).then([this] (auto gs) {
- h.global_seq = gs;
- return seastar::make_ready_future<stop_t>(stop_t::no);
- });
- case CEPH_MSGR_TAG_RETRY_SESSION:
- ceph_assert(h.reply.connect_seq > h.connect_seq);
- h.connect_seq = h.reply.connect_seq;
- return seastar::make_ready_future<stop_t>(stop_t::no);
- case CEPH_MSGR_TAG_WAIT:
- // TODO: state wait
- throw std::system_error(make_error_code(error::negotiation_failure));
- case CEPH_MSGR_TAG_SEQ:
- case CEPH_MSGR_TAG_READY:
- if (auto missing = (conn.policy.features_required & ~(uint64_t)h.reply.features);
- missing) {
- logger().error("{} missing required features", __func__);
- throw std::system_error(make_error_code(error::negotiation_failure));
- }
- return seastar::futurize_invoke([this, tag] {
- if (tag == CEPH_MSGR_TAG_SEQ) {
- return socket->read_exactly(sizeof(seq_num_t))
- .then([this] (auto buf) {
- auto acked_seq = reinterpret_cast<const seq_num_t*>(buf.get());
- discard_up_to(&conn.out_q, *acked_seq);
- return socket->write_flush(make_static_packet(conn.in_seq));
- });
- }
- // tag CEPH_MSGR_TAG_READY
- return seastar::now();
- }).then([this] {
- // hooray!
- h.peer_global_seq = h.reply.global_seq;
- conn.policy.lossy = h.reply.flags & CEPH_MSG_CONNECT_LOSSY;
- h.connect_seq++;
- h.backoff = 0ms;
- conn.set_features(h.reply.features & h.connect.features);
- if (auth_meta->authorizer) {
- session_security.reset(
- get_auth_session_handler(nullptr,
- auth_meta->authorizer->protocol,
- auth_meta->session_key,
- conn.features));
- } else {
- session_security.reset();
- }
- return seastar::make_ready_future<stop_t>(stop_t::yes);
- });
- break;
- default:
- // unknown tag
- logger().error("{} got unknown tag", __func__, int(tag));
- throw std::system_error(make_error_code(error::negotiation_failure));
- }
-}
-
-ceph::bufferlist ProtocolV1::get_auth_payload()
-{
- // only non-mons connectings to mons use MAuth messages
- if (conn.peer_is_mon() &&
- messenger.get_mytype() != CEPH_ENTITY_TYPE_MON) {
- return {};
- } else {
- if (h.auth_more.length()) {
- logger().info("using augmented (challenge) auth payload");
- return std::move(h.auth_more);
- } else {
- auto [auth_method, preferred_modes, auth_bl] =
- messenger.get_auth_client()->get_auth_request(
- conn.shared_from_this(), auth_meta);
- auth_meta->auth_method = auth_method;
- return auth_bl;
- }
- }
-}
-
-seastar::future<stop_t>
-ProtocolV1::repeat_connect()
-{
- // encode ceph_msg_connect
- memset(&h.connect, 0, sizeof(h.connect));
- h.connect.features = conn.policy.features_supported;
- h.connect.host_type = messenger.get_myname().type();
- h.connect.global_seq = h.global_seq;
- h.connect.connect_seq = h.connect_seq;
- h.connect.protocol_version = get_proto_version(conn.get_peer_type(), true);
- // this is fyi, actually, server decides!
- h.connect.flags = conn.policy.lossy ? CEPH_MSG_CONNECT_LOSSY : 0;
-
- ceph_assert(messenger.get_auth_client());
-
- bufferlist bl;
- bufferlist auth_bl = get_auth_payload();
- if (auth_bl.length()) {
- h.connect.authorizer_protocol = auth_meta->auth_method;
- h.connect.authorizer_len = auth_bl.length();
- bl.append(create_static(h.connect));
- bl.claim_append(auth_bl);
- } else {
- h.connect.authorizer_protocol = 0;
- h.connect.authorizer_len = 0;
- bl.append(create_static(h.connect));
- };
- return socket->write_flush(std::move(bl))
- .then([this] {
- // read the reply
- return socket->read(sizeof(h.reply));
- }).then([this] (bufferlist bl) {
- auto p = bl.cbegin();
- ::decode(h.reply, p);
- ceph_assert(p.end());
- return socket->read(h.reply.authorizer_len);
- }).then([this] (bufferlist bl) {
- h.auth_payload = std::move(bl);
- return handle_connect_reply(h.reply.tag);
- });
-}
-
-void ProtocolV1::start_connect(const entity_addr_t& _peer_addr,
- const entity_name_t& _peer_name)
-{
- ceph_assert(state == state_t::none);
- logger().trace("{} trigger connecting, was {}", conn, static_cast<int>(state));
- state = state_t::connecting;
- set_write_state(write_state_t::delay);
-
- ceph_assert(!socket);
- ceph_assert(!gate.is_closed());
- conn.peer_addr = _peer_addr;
- conn.target_addr = _peer_addr;
- conn.set_peer_name(_peer_name);
- conn.policy = messenger.get_policy(_peer_name.type());
- messenger.register_conn(
- seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()));
- gate.dispatch_in_background("start_connect", *this, [this] {
- return Socket::connect(conn.peer_addr)
- .then([this](SocketRef sock) {
- socket = std::move(sock);
- if (state != state_t::connecting) {
- assert(state == state_t::closing);
- return socket->close().then([] {
- throw std::system_error(make_error_code(error::protocol_aborted));
- });
- }
- return seastar::now();
- }).then([this] {
- return messenger.get_global_seq();
- }).then([this] (auto gs) {
- h.global_seq = gs;
- // read server's handshake header
- return socket->read(server_header_size);
- }).then([this] (bufferlist headerbl) {
- auto p = headerbl.cbegin();
- validate_banner(p);
- entity_addr_t saddr, caddr;
- ::decode(saddr, p);
- ::decode(caddr, p);
- ceph_assert(p.end());
- if (saddr != conn.peer_addr) {
- logger().error("{} my peer_addr {} doesn't match what peer advertized {}",
- conn, conn.peer_addr, saddr);
- throw std::system_error(
- make_error_code(crimson::net::error::bad_peer_address));
- }
- if (state != state_t::connecting) {
- assert(state == state_t::closing);
- throw std::system_error(make_error_code(error::protocol_aborted));
- }
- socket->learn_ephemeral_port_as_connector(caddr.get_port());
- if (unlikely(caddr.is_msgr2())) {
- logger().warn("{} peer sent a v2 address for me: {}",
- conn, caddr);
- throw std::system_error(
- make_error_code(crimson::net::error::bad_peer_address));
- }
- caddr.set_type(entity_addr_t::TYPE_LEGACY);
- return messenger.learned_addr(caddr, conn);
- }).then([this] {
- // encode/send client's handshake header
- bufferlist bl;
- bl.append(buffer::create_static(banner_size, banner));
- ::encode(messenger.get_myaddr(), bl, 0);
- return socket->write_flush(std::move(bl));
- }).then([=] {
- return seastar::repeat([this] {
- return repeat_connect();
- });
- }).then([this] {
- if (state != state_t::connecting) {
- assert(state == state_t::closing);
- throw std::system_error(make_error_code(error::protocol_aborted));
- }
- execute_open(open_t::connected);
- }).handle_exception([this] (std::exception_ptr eptr) {
- // TODO: handle fault in the connecting state
- logger().warn("{} connecting fault: {}", conn, eptr);
- close(true);
- });
- });
-}
-
-// accepting state
-
-seastar::future<stop_t> ProtocolV1::send_connect_reply(
- msgr_tag_t tag, bufferlist&& authorizer_reply)
-{
- h.reply.tag = tag;
- h.reply.features = static_cast<uint64_t>((h.connect.features &
- conn.policy.features_supported) |
- conn.policy.features_required);
- h.reply.authorizer_len = authorizer_reply.length();
- return socket->write(make_static_packet(h.reply))
- .then([this, reply=std::move(authorizer_reply)]() mutable {
- return socket->write_flush(std::move(reply));
- }).then([] {
- return stop_t::no;
- });
-}
-
-seastar::future<stop_t> ProtocolV1::send_connect_reply_ready(
- msgr_tag_t tag, bufferlist&& authorizer_reply)
-{
- return messenger.get_global_seq(
- ).then([this, tag, auth_len = authorizer_reply.length()] (auto gs) {
- h.global_seq = gs;
- h.reply.tag = tag;
- h.reply.features = conn.policy.features_supported;
- h.reply.global_seq = h.global_seq;
- h.reply.connect_seq = h.connect_seq;
- h.reply.flags = 0;
- if (conn.policy.lossy) {
- h.reply.flags = h.reply.flags | CEPH_MSG_CONNECT_LOSSY;
- }
- h.reply.authorizer_len = auth_len;
-
- session_security.reset(
- get_auth_session_handler(nullptr,
- auth_meta->auth_method,
- auth_meta->session_key,
- conn.features));
-
- return socket->write(make_static_packet(h.reply));
- }).then([this, reply=std::move(authorizer_reply)]() mutable {
- if (reply.length()) {
- return socket->write(std::move(reply));
- } else {
- return seastar::now();
- }
- }).then([this] {
- if (h.reply.tag == CEPH_MSGR_TAG_SEQ) {
- return socket->write_flush(make_static_packet(conn.in_seq))
- .then([this] {
- return socket->read_exactly(sizeof(seq_num_t));
- }).then([this] (auto buf) {
- auto acked_seq = reinterpret_cast<const seq_num_t*>(buf.get());
- discard_up_to(&conn.out_q, *acked_seq);
- });
- } else {
- return socket->flush();
- }
- }).then([] {
- return stop_t::yes;
- });
-}
-
-seastar::future<stop_t> ProtocolV1::replace_existing(
- SocketConnectionRef existing,
- bufferlist&& authorizer_reply,
- bool is_reset_from_peer)
-{
- msgr_tag_t reply_tag;
- if (HAVE_FEATURE(h.connect.features, RECONNECT_SEQ) &&
- !is_reset_from_peer) {
- reply_tag = CEPH_MSGR_TAG_SEQ;
- } else {
- reply_tag = CEPH_MSGR_TAG_READY;
- }
- if (!existing->is_lossy()) {
- // XXX: we decided not to support lossless connection in v1. as the
- // client's default policy is
- // Messenger::Policy::lossy_client(CEPH_FEATURE_OSDREPLYMUX) which is
- // lossy. And by the time
- // will all be performed using v2 protocol.
- ceph_abort("lossless policy not supported for v1");
- }
- existing->protocol->close(true);
- return send_connect_reply_ready(reply_tag, std::move(authorizer_reply));
-}
-
-seastar::future<stop_t> ProtocolV1::handle_connect_with_existing(
- SocketConnectionRef existing, bufferlist&& authorizer_reply)
-{
- ProtocolV1 *exproto = dynamic_cast<ProtocolV1*>(existing->protocol.get());
-
- if (h.connect.global_seq < exproto->peer_global_seq()) {
- h.reply.global_seq = exproto->peer_global_seq();
- return send_connect_reply(CEPH_MSGR_TAG_RETRY_GLOBAL);
- } else if (existing->is_lossy()) {
- return replace_existing(existing, std::move(authorizer_reply));
- } else if (h.connect.connect_seq == 0 && exproto->connect_seq() > 0) {
- return replace_existing(existing, std::move(authorizer_reply), true);
- } else if (h.connect.connect_seq < exproto->connect_seq()) {
- // old attempt, or we sent READY but they didn't get it.
- h.reply.connect_seq = exproto->connect_seq() + 1;
- return send_connect_reply(CEPH_MSGR_TAG_RETRY_SESSION);
- } else if (h.connect.connect_seq == exproto->connect_seq()) {
- // if the existing connection successfully opened, and/or
- // subsequently went to standby, then the peer should bump
- // their connect_seq and retry: this is not a connection race
- // we need to resolve here.
- if (exproto->get_state() == state_t::open ||
- exproto->get_state() == state_t::standby) {
- if (conn.policy.resetcheck && exproto->connect_seq() == 0) {
- return replace_existing(existing, std::move(authorizer_reply));
- } else {
- h.reply.connect_seq = exproto->connect_seq() + 1;
- return send_connect_reply(CEPH_MSGR_TAG_RETRY_SESSION);
- }
- } else if (existing->peer_wins()) {
- return replace_existing(existing, std::move(authorizer_reply));
- } else {
- return send_connect_reply(CEPH_MSGR_TAG_WAIT);
- }
- } else if (conn.policy.resetcheck &&
- exproto->connect_seq() == 0) {
- return send_connect_reply(CEPH_MSGR_TAG_RESETSESSION);
- } else {
- return replace_existing(existing, std::move(authorizer_reply));
- }
-}
-
-bool ProtocolV1::require_auth_feature() const
-{
- if (h.connect.authorizer_protocol != CEPH_AUTH_CEPHX) {
- return false;
- }
- if (local_conf()->cephx_require_signatures) {
- return true;
- }
- if (h.connect.host_type == CEPH_ENTITY_TYPE_OSD ||
- h.connect.host_type == CEPH_ENTITY_TYPE_MDS ||
- h.connect.host_type == CEPH_ENTITY_TYPE_MGR) {
- return local_conf()->cephx_cluster_require_signatures;
- } else {
- return local_conf()->cephx_service_require_signatures;
- }
-}
-
-bool ProtocolV1::require_cephx_v2_feature() const
-{
- if (h.connect.authorizer_protocol != CEPH_AUTH_CEPHX) {
- return false;
- }
- if (local_conf()->cephx_require_version >= 2) {
- return true;
- }
- if (h.connect.host_type == CEPH_ENTITY_TYPE_OSD ||
- h.connect.host_type == CEPH_ENTITY_TYPE_MDS ||
- h.connect.host_type == CEPH_ENTITY_TYPE_MGR) {
- return local_conf()->cephx_cluster_require_version >= 2;
- } else {
- return local_conf()->cephx_service_require_version >= 2;
- }
-}
-
-seastar::future<stop_t> ProtocolV1::repeat_handle_connect()
-{
- return socket->read(sizeof(h.connect))
- .then([this](bufferlist bl) {
- auto p = bl.cbegin();
- ::decode(h.connect, p);
- if (conn.get_peer_type() != 0 &&
- conn.get_peer_type() != h.connect.host_type) {
- logger().error("{} repeat_handle_connect(): my peer type does not match"
- " what peer advertises {} != {}",
- conn, conn.get_peer_type(), h.connect.host_type);
- throw std::system_error(make_error_code(error::protocol_aborted));
- }
- conn.set_peer_type(h.connect.host_type);
- conn.policy = messenger.get_policy(h.connect.host_type);
- if (!conn.policy.lossy && !conn.policy.server && conn.target_addr.get_port() <= 0) {
- logger().error("{} we don't know how to reconnect to peer {}",
- conn, conn.target_addr);
- throw std::system_error(
- make_error_code(crimson::net::error::bad_peer_address));
- }
- return socket->read(h.connect.authorizer_len);
- }).then([this] (bufferlist authorizer) {
- memset(&h.reply, 0, sizeof(h.reply));
- // TODO: set reply.protocol_version
- if (h.connect.protocol_version != get_proto_version(h.connect.host_type, false)) {
- return send_connect_reply(
- CEPH_MSGR_TAG_BADPROTOVER, bufferlist{});
- }
- if (require_auth_feature()) {
- conn.policy.features_required |= CEPH_FEATURE_MSG_AUTH;
- }
- if (require_cephx_v2_feature()) {
- conn.policy.features_required |= CEPH_FEATUREMASK_CEPHX_V2;
- }
- if (auto feat_missing = conn.policy.features_required & ~(uint64_t)h.connect.features;
- feat_missing != 0) {
- return send_connect_reply(
- CEPH_MSGR_TAG_FEATURES, bufferlist{});
- }
-
- bufferlist authorizer_reply;
- auth_meta->auth_method = h.connect.authorizer_protocol;
- if (!HAVE_FEATURE((uint64_t)h.connect.features, CEPHX_V2)) {
- // peer doesn't support it and we won't get here if we require it
- auth_meta->skip_authorizer_challenge = true;
- }
- auto more = static_cast<bool>(auth_meta->authorizer_challenge);
- ceph_assert(messenger.get_auth_server());
- int r = messenger.get_auth_server()->handle_auth_request(
- conn.shared_from_this(), auth_meta, more, auth_meta->auth_method, authorizer,
- &authorizer_reply);
-
- if (r < 0) {
- session_security.reset();
- return send_connect_reply(
- CEPH_MSGR_TAG_BADAUTHORIZER, std::move(authorizer_reply));
- } else if (r == 0) {
- ceph_assert(authorizer_reply.length());
- return send_connect_reply(
- CEPH_MSGR_TAG_CHALLENGE_AUTHORIZER, std::move(authorizer_reply));
- }
-
- // r > 0
- if (auto existing = messenger.lookup_conn(conn.peer_addr); existing) {
- if (existing->protocol->proto_type != proto_t::v1) {
- logger().warn("{} existing {} proto version is {} not 1, close existing",
- conn, *existing,
- static_cast<int>(existing->protocol->proto_type));
- // NOTE: this is following async messenger logic, but we may miss the reset event.
- existing->mark_down();
- } else {
- return handle_connect_with_existing(existing, std::move(authorizer_reply));
- }
- }
- if (h.connect.connect_seq > 0) {
- return send_connect_reply(CEPH_MSGR_TAG_RESETSESSION,
- std::move(authorizer_reply));
- }
- h.connect_seq = h.connect.connect_seq + 1;
- h.peer_global_seq = h.connect.global_seq;
- conn.set_features((uint64_t)conn.policy.features_supported & (uint64_t)h.connect.features);
- // TODO: cct
- return send_connect_reply_ready(CEPH_MSGR_TAG_READY, std::move(authorizer_reply));
- });
-}
-
-void ProtocolV1::start_accept(SocketRef&& sock,
- const entity_addr_t& _peer_addr)
-{
- ceph_assert(state == state_t::none);
- logger().trace("{} trigger accepting, was {}",
- conn, static_cast<int>(state));
- state = state_t::accepting;
- set_write_state(write_state_t::delay);
-
- ceph_assert(!socket);
- // until we know better
- conn.target_addr = _peer_addr;
- socket = std::move(sock);
- messenger.accept_conn(
- seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()));
- gate.dispatch_in_background("start_accept", *this, [this] {
- // stop learning my_addr before sending it out, so it won't change
- return messenger.learned_addr(messenger.get_myaddr(), conn).then([this] {
- // encode/send server's handshake header
- bufferlist bl;
- bl.append(buffer::create_static(banner_size, banner));
- ::encode(messenger.get_myaddr(), bl, 0);
- ::encode(conn.target_addr, bl, 0);
- return socket->write_flush(std::move(bl));
- }).then([this] {
- // read client's handshake header and connect request
- return socket->read(client_header_size);
- }).then([this] (bufferlist bl) {
- auto p = bl.cbegin();
- validate_banner(p);
- entity_addr_t addr;
- ::decode(addr, p);
- ceph_assert(p.end());
- if ((addr.is_legacy() || addr.is_any()) &&
- addr.is_same_host(conn.target_addr)) {
- // good
- } else {
- logger().error("{} peer advertized an invalid peer_addr: {},"
- " which should be v1 and the same host with {}.",
- conn, addr, conn.peer_addr);
- throw std::system_error(
- make_error_code(crimson::net::error::bad_peer_address));
- }
- conn.peer_addr = addr;
- conn.target_addr = conn.peer_addr;
- return seastar::repeat([this] {
- return repeat_handle_connect();
- });
- }).then([this] {
- if (state != state_t::accepting) {
- assert(state == state_t::closing);
- throw std::system_error(make_error_code(error::protocol_aborted));
- }
- messenger.register_conn(
- seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()));
- messenger.unaccept_conn(
- seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()));
- execute_open(open_t::accepted);
- }).handle_exception([this] (std::exception_ptr eptr) {
- // TODO: handle fault in the accepting state
- logger().warn("{} accepting fault: {}", conn, eptr);
- close(false);
- });
- });
-}
-
-// open state
-
-ceph::bufferlist ProtocolV1::do_sweep_messages(
- const std::deque<MessageRef>& msgs,
- size_t num_msgs,
- bool require_keepalive,
- std::optional<utime_t> _keepalive_ack,
- bool require_ack)
-{
- static const size_t RESERVE_MSG_SIZE = sizeof(CEPH_MSGR_TAG_MSG) +
- sizeof(ceph_msg_header) +
- sizeof(ceph_msg_footer);
- static const size_t RESERVE_MSG_SIZE_OLD = sizeof(CEPH_MSGR_TAG_MSG) +
- sizeof(ceph_msg_header) +
- sizeof(ceph_msg_footer_old);
-
- ceph::bufferlist bl;
- if (likely(num_msgs)) {
- if (HAVE_FEATURE(conn.features, MSG_AUTH)) {
- bl.reserve(num_msgs * RESERVE_MSG_SIZE);
- } else {
- bl.reserve(num_msgs * RESERVE_MSG_SIZE_OLD);
- }
- }
-
- if (unlikely(require_keepalive)) {
- k.req.stamp = ceph::coarse_real_clock::to_ceph_timespec(
- ceph::coarse_real_clock::now());
- logger().trace("{} write keepalive2 {}", conn, k.req.stamp.tv_sec);
- bl.append(create_static(k.req));
- }
-
- if (unlikely(_keepalive_ack.has_value())) {
- logger().trace("{} write keepalive2 ack {}", conn, *_keepalive_ack);
- k.ack.stamp = ceph_timespec(*_keepalive_ack);
- bl.append(create_static(k.ack));
- }
-
- if (require_ack) {
- // XXX: we decided not to support lossless connection in v1. as the
- // client's default policy is
- // Messenger::Policy::lossy_client(CEPH_FEATURE_OSDREPLYMUX) which is
- // lossy. And by the time of crimson-osd's GA, the in-cluster communication
- // will all be performed using v2 protocol.
- ceph_abort("lossless policy not supported for v1");
- }
-
- std::for_each(msgs.begin(), msgs.begin()+num_msgs, [this, &bl](const MessageRef& msg) {
- ceph_assert(!msg->get_seq() && "message already has seq");
- msg->set_seq(++conn.out_seq);
- auto& header = msg->get_header();
- header.src = messenger.get_myname();
- msg->encode(conn.features, messenger.get_crc_flags());
- if (session_security) {
- session_security->sign_message(msg.get());
- }
- logger().debug("{} --> #{} === {} ({})",
- conn, msg->get_seq(), *msg, msg->get_type());
- bl.append(CEPH_MSGR_TAG_MSG);
- bl.append((const char*)&header, sizeof(header));
- bl.append(msg->get_payload());
- bl.append(msg->get_middle());
- bl.append(msg->get_data());
- auto& footer = msg->get_footer();
- if (HAVE_FEATURE(conn.features, MSG_AUTH)) {
- bl.append((const char*)&footer, sizeof(footer));
- } else {
- ceph_msg_footer_old old_footer;
- if (messenger.get_crc_flags() & MSG_CRC_HEADER) {
- old_footer.front_crc = footer.front_crc;
- old_footer.middle_crc = footer.middle_crc;
- } else {
- old_footer.front_crc = old_footer.middle_crc = 0;
- }
- if (messenger.get_crc_flags() & MSG_CRC_DATA) {
- old_footer.data_crc = footer.data_crc;
- } else {
- old_footer.data_crc = 0;
- }
- old_footer.flags = footer.flags;
- bl.append((const char*)&old_footer, sizeof(old_footer));
- }
- });
-
- return bl;
-}
-
-seastar::future<> ProtocolV1::handle_keepalive2_ack()
-{
- return socket->read_exactly(sizeof(ceph_timespec))
- .then([this] (auto buf) {
- auto t = reinterpret_cast<const ceph_timespec*>(buf.get());
- k.ack_stamp = *t;
- logger().trace("{} got keepalive2 ack {}", conn, t->tv_sec);
- });
-}
-
-seastar::future<> ProtocolV1::handle_keepalive2()
-{
- return socket->read_exactly(sizeof(ceph_timespec))
- .then([this] (auto buf) {
- utime_t ack{*reinterpret_cast<const ceph_timespec*>(buf.get())};
- notify_keepalive_ack(ack);
- });
-}
-
-seastar::future<> ProtocolV1::handle_ack()
-{
- return socket->read_exactly(sizeof(ceph_le64))
- .then([this] (auto buf) {
- auto seq = reinterpret_cast<const ceph_le64*>(buf.get());
- discard_up_to(&conn.sent, *seq);
- });
-}
-
-seastar::future<> ProtocolV1::maybe_throttle()
-{
- if (!conn.policy.throttler_bytes) {
- return seastar::now();
- }
- const auto to_read = (m.header.front_len +
- m.header.middle_len +
- m.header.data_len);
- return conn.policy.throttler_bytes->get(to_read);
-}
-
-seastar::future<> ProtocolV1::read_message()
-{
- return socket->read(sizeof(m.header))
- .then([this] (bufferlist bl) {
- // throttle the traffic, maybe
- auto p = bl.cbegin();
- ::decode(m.header, p);
- return maybe_throttle();
- }).then([this] {
- // read front
- return socket->read(m.header.front_len);
- }).then([this] (bufferlist bl) {
- m.front = std::move(bl);
- // read middle
- return socket->read(m.header.middle_len);
- }).then([this] (bufferlist bl) {
- m.middle = std::move(bl);
- // read data
- return socket->read(m.header.data_len);
- }).then([this] (bufferlist bl) {
- m.data = std::move(bl);
- // read footer
- return socket->read(sizeof(m.footer));
- }).then([this] (bufferlist bl) {
- auto p = bl.cbegin();
- ::decode(m.footer, p);
- auto conn_ref = seastar::static_pointer_cast<SocketConnection>(
- conn.shared_from_this());
- auto msg = ::decode_message(nullptr, 0, m.header, m.footer,
- m.front, m.middle, m.data, conn_ref);
- if (unlikely(!msg)) {
- logger().warn("{} decode message failed", conn);
- throw std::system_error{make_error_code(error::corrupted_message)};
- }
- constexpr bool add_ref = false; // Message starts with 1 ref
- // TODO: change MessageRef with foreign_ptr
- auto msg_ref = MessageRef{msg, add_ref};
-
- if (session_security) {
- if (unlikely(session_security->check_message_signature(msg))) {
- logger().warn("{} message signature check failed", conn);
- throw std::system_error{make_error_code(error::corrupted_message)};
- }
- }
- // TODO: set time stamps
- msg->set_byte_throttler(conn.policy.throttler_bytes);
-
- if (unlikely(!conn.update_rx_seq(msg->get_seq()))) {
- // skip this message
- return seastar::now();
- }
-
- logger().debug("{} <== #{} === {} ({})",
- conn, msg_ref->get_seq(), *msg_ref, msg_ref->get_type());
- // throttle the reading process by the returned future
- return dispatchers.ms_dispatch(conn_ref, std::move(msg_ref));
- });
-}
-
-seastar::future<> ProtocolV1::handle_tags()
-{
- return seastar::keep_doing([this] {
- // read the next tag
- return socket->read_exactly(1)
- .then([this] (auto buf) {
- switch (buf[0]) {
- case CEPH_MSGR_TAG_MSG:
- return read_message();
- case CEPH_MSGR_TAG_ACK:
- return handle_ack();
- case CEPH_MSGR_TAG_KEEPALIVE:
- return seastar::now();
- case CEPH_MSGR_TAG_KEEPALIVE2:
- return handle_keepalive2();
- case CEPH_MSGR_TAG_KEEPALIVE2_ACK:
- return handle_keepalive2_ack();
- case CEPH_MSGR_TAG_CLOSE:
- logger().info("{} got tag close", conn);
- throw std::system_error(make_error_code(error::protocol_aborted));
- default:
- logger().error("{} got unknown msgr tag {}",
- conn, static_cast<int>(buf[0]));
- throw std::system_error(make_error_code(error::read_eof));
- }
- });
- });
-}
-
-void ProtocolV1::execute_open(open_t type)
-{
- logger().trace("{} trigger open, was {}", conn, static_cast<int>(state));
- state = state_t::open;
- set_write_state(write_state_t::open);
-
- if (type == open_t::connected) {
- dispatchers.ms_handle_connect(
- seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()));
- } else { // type == open_t::accepted
- dispatchers.ms_handle_accept(
- seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()));
- }
-
- gate.dispatch_in_background("execute_open", *this, [this] {
- // start background processing of tags
- return handle_tags()
- .handle_exception_type([this] (const std::system_error& e) {
- logger().warn("{} open fault: {}", conn, e);
- if (e.code() == error::protocol_aborted ||
- e.code() == std::errc::connection_reset ||
- e.code() == error::read_eof) {
- close(true);
- return seastar::now();
- } else {
- throw e;
- }
- }).handle_exception([this] (std::exception_ptr eptr) {
- // TODO: handle fault in the open state
- logger().warn("{} open fault: {}", conn, eptr);
- close(true);
- });
- });
-}
-
-// closing state
-
-void ProtocolV1::trigger_close()
-{
- logger().trace("{} trigger closing, was {}",
- conn, static_cast<int>(state));
- messenger.closing_conn(
- seastar::static_pointer_cast<SocketConnection>(
- conn.shared_from_this()));
-
- if (state == state_t::accepting) {
- messenger.unaccept_conn(seastar::static_pointer_cast<SocketConnection>(
- conn.shared_from_this()));
- } else if (state >= state_t::connecting && state < state_t::closing) {
- messenger.unregister_conn(seastar::static_pointer_cast<SocketConnection>(
- conn.shared_from_this()));
- } else {
- // cannot happen
- ceph_assert(false);
- }
-
- if (!socket) {
- ceph_assert(state == state_t::connecting);
- }
-
- state = state_t::closing;
-}
-
-void ProtocolV1::on_closed()
-{
- messenger.closed_conn(
- seastar::static_pointer_cast<SocketConnection>(
- conn.shared_from_this()));
-}
-
-seastar::future<> ProtocolV1::fault()
-{
- if (conn.policy.lossy) {
- messenger.unregister_conn(seastar::static_pointer_cast<SocketConnection>(
- conn.shared_from_this()));
- }
- // XXX: we decided not to support lossless connection in v1. as the
- // client's default policy is
- // Messenger::Policy::lossy_client(CEPH_FEATURE_OSDREPLYMUX) which is
- // lossy. And by the time of crimson-osd's GA, the in-cluster communication
- // will all be performed using v2 protocol.
- ceph_abort("lossless policy not supported for v1");
- return seastar::now();
-}
-
-void ProtocolV1::print(std::ostream& out) const
-{
- out << conn;
-}
-
-} // namespace crimson::net
projects / ceph.git / commitdiff