ceph/src/crimson/net/SocketConnection.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2017 Red Hat, Inc
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #pragma once
  16
  17 #include <seastar/core/sharded.hh>
  18
  19 #include "msg/Policy.h"
  20 #include "crimson/common/throttle.h"
  21 #include "crimson/net/Connection.h"
  22 #include "crimson/net/Socket.h"
  23
  24 namespace crimson::net {
  25
  26 class ProtocolV2;
  27 class SocketMessenger;
  28 class SocketConnection;
  29 using SocketConnectionRef = seastar::shared_ptr<SocketConnection>;
  30
  31 #ifdef UNIT_TESTS_BUILT
  32 class Interceptor;
  33 #endif
  34
  35 /**
  36  * ConnectionHandler
  37  *
  38  * The interface class to implement Connection, called by SocketConnection.
  39  *
  40  * The operations must be done in get_shard_id().
  41  */
  42 class ConnectionHandler {
  43 public:
  44   using clock_t = seastar::lowres_system_clock;
  45
  46   virtual ~ConnectionHandler() = default;
  47
  48   ConnectionHandler(const ConnectionHandler &) = delete;
  49   ConnectionHandler(ConnectionHandler &&) = delete;
  50   ConnectionHandler &operator=(const ConnectionHandler &) = delete;
  51   ConnectionHandler &operator=(ConnectionHandler &&) = delete;
  52
  53   virtual seastar::shard_id get_shard_id() const = 0;
  54
  55   virtual bool is_connected() const = 0;
  56
  57   virtual seastar::future<> send(MessageFRef) = 0;
  58
  59   virtual seastar::future<> send_keepalive() = 0;
  60
  61   virtual clock_t::time_point get_last_keepalive() const = 0;
  62
  63   virtual clock_t::time_point get_last_keepalive_ack() const = 0;
  64
  65   virtual void set_last_keepalive_ack(clock_t::time_point) = 0;
  66
  67   virtual void mark_down() = 0;
  68
  69 protected:
  70   ConnectionHandler() = default;
  71 };
  72
  73 class SocketConnection : public Connection {
  74  /*
  75   * Connection interfaces, public to users
  76   * Working in ConnectionHandler::get_shard_id()
  77   */
  78  public:
  79   SocketConnection(SocketMessenger& messenger,
  80                    ChainedDispatchers& dispatchers);
  81
  82   ~SocketConnection() override;
  83
  84   const seastar::shard_id get_shard_id() const override {
  85     return io_handler->get_shard_id();
  86   }
  87
  88   const entity_name_t &get_peer_name() const override {
  89     return peer_name;
  90   }
  91
  92   const entity_addr_t &get_peer_addr() const override {
  93     return peer_addr;
  94   }
  95
  96   const entity_addr_t &get_peer_socket_addr() const override {
  97     return target_addr;
  98   }
  99
 100   uint64_t get_features() const override {
 101     return features;
 102   }
 103
 104   bool is_connected() const override;
 105
 106   seastar::future<> send(MessageURef msg) override;
 107
 108   seastar::future<> send_keepalive() override;
 109
 110   clock_t::time_point get_last_keepalive() const override;
 111
 112   clock_t::time_point get_last_keepalive_ack() const override;
 113
 114   void set_last_keepalive_ack(clock_t::time_point when) override;
 115
 116   void mark_down() override;
 117
 118   bool has_user_private() const override {
 119     return user_private != nullptr;
 120   }
 121
 122   user_private_t &get_user_private() override {
 123     assert(has_user_private());
 124     return *user_private;
 125   }
 126
 127   void set_user_private(std::unique_ptr<user_private_t> new_user_private) override {
 128     assert(!has_user_private());
 129     user_private = std::move(new_user_private);
 130   }
 131
 132   void print(std::ostream& out) const override;
 133
 134  /*
 135   * Public to SocketMessenger
 136   * Working in SocketMessenger::get_shard_id();
 137   */
 138  public:
 139   /// start a handshake from the client's perspective,
 140   /// only call when SocketConnection first construct
 141   void start_connect(const entity_addr_t& peer_addr,
 142                      const entity_name_t& peer_name);
 143
 144   /// start a handshake from the server's perspective,
 145   /// only call when SocketConnection first construct
 146   void start_accept(SocketFRef&& socket,
 147                     const entity_addr_t& peer_addr);
 148
 149   seastar::future<> close_clean_yielded();
 150
 151   seastar::socket_address get_local_address() const;
 152
 153   seastar::shard_id get_messenger_shard_id() const;
 154
 155   SocketMessenger &get_messenger() const;
 156
 157   ConnectionRef get_local_shared_foreign_from_this();
 158
 159 private:
 160   void set_peer_type(entity_type_t peer_type);
 161
 162   void set_peer_id(int64_t peer_id);
 163
 164   void set_peer_name(entity_name_t name) {
 165     set_peer_type(name.type());
 166     set_peer_id(name.num());
 167   }
 168
 169   void set_features(uint64_t f);
 170
 171   void set_socket(Socket *s);
 172
 173 #ifdef UNIT_TESTS_BUILT
 174   bool is_protocol_ready() const override;
 175
 176   bool is_protocol_standby() const override;
 177
 178   bool is_protocol_closed_clean() const override;
 179
 180   bool is_protocol_closed() const override;
 181
 182   // peer wins if myaddr > peeraddr
 183   bool peer_wins() const override;
 184
 185   Interceptor *interceptor = nullptr;
 186 #else
 187   // peer wins if myaddr > peeraddr
 188   bool peer_wins() const;
 189 #endif
 190
 191 private:
 192   const seastar::shard_id msgr_sid;
 193
 194   /*
 195    * Core owner is messenger core, may allow to access from the I/O core.
 196    */
 197   SocketMessenger& messenger;
 198
 199   std::unique_ptr<ProtocolV2> protocol;
 200
 201   Socket *socket = nullptr;
 202
 203   entity_name_t peer_name = {0, entity_name_t::NEW};
 204
 205   entity_addr_t peer_addr;
 206
 207   // which of the peer_addrs we're connecting to (as client)
 208   // or should reconnect to (as peer)
 209   entity_addr_t target_addr;
 210
 211   uint64_t features = 0;
 212
 213   ceph::net::Policy<crimson::common::Throttle> policy;
 214
 215   uint64_t peer_global_id = 0;
 216
 217   /*
 218    * Core owner is I/O core (mutable).
 219    */
 220   std::unique_ptr<ConnectionHandler> io_handler;
 221
 222   /*
 223    * Core owner is up to the connection user.
 224    */
 225   std::unique_ptr<user_private_t> user_private;
 226
 227   friend class IOHandler;
 228   friend class ProtocolV2;
 229   friend class FrameAssemblerV2;
 230 };
 231
 232 } // namespace crimson::net
 233
 234 #if FMT_VERSION >= 90000
 235 template <> struct fmt::formatter<crimson::net::SocketConnection> : fmt::ostream_formatter {};
 236 #endif