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 using SocketConnectionRef = seastar::shared_ptr<SocketConnection>;
  29
  30 #ifdef UNIT_TESTS_BUILT
  31 class Interceptor;
  32 #endif
  33
  34 /**
  35  * ConnectionHandler
  36  *
  37  * The interface class to implement Connection, called by SocketConnection.
  38  */
  39 class ConnectionHandler {
  40 public:
  41   using clock_t = seastar::lowres_system_clock;
  42
  43   virtual ~ConnectionHandler() = default;
  44
  45   ConnectionHandler(const ConnectionHandler &) = delete;
  46   ConnectionHandler(ConnectionHandler &&) = delete;
  47   ConnectionHandler &operator=(const ConnectionHandler &) = delete;
  48   ConnectionHandler &operator=(ConnectionHandler &&) = delete;
  49
  50   virtual bool is_connected() const = 0;
  51
  52   virtual seastar::future<> send(MessageURef) = 0;
  53
  54   virtual seastar::future<> send_keepalive() = 0;
  55
  56   virtual clock_t::time_point get_last_keepalive() const = 0;
  57
  58   virtual clock_t::time_point get_last_keepalive_ack() const = 0;
  59
  60   virtual void set_last_keepalive_ack(clock_t::time_point) = 0;
  61
  62   virtual void mark_down() = 0;
  63
  64 protected:
  65   ConnectionHandler() = default;
  66 };
  67
  68 class SocketConnection : public Connection {
  69   const seastar::shard_id core;
  70
  71   SocketMessenger& messenger;
  72
  73   std::unique_ptr<ConnectionHandler> io_handler;
  74
  75   std::unique_ptr<ProtocolV2> protocol;
  76
  77   SocketRef socket;
  78
  79   entity_name_t peer_name = {0, entity_name_t::NEW};
  80
  81   entity_addr_t peer_addr;
  82
  83   // which of the peer_addrs we're connecting to (as client)
  84   // or should reconnect to (as peer)
  85   entity_addr_t target_addr;
  86
  87   uint64_t features = 0;
  88
  89   ceph::net::Policy<crimson::common::Throttle> policy;
  90
  91   uint64_t peer_global_id = 0;
  92
  93   std::unique_ptr<user_private_t> user_private;
  94
  95  // Connection interfaces, public to users
  96  public:
  97   SocketConnection(SocketMessenger& messenger,
  98                    ChainedDispatchers& dispatchers);
  99
 100   ~SocketConnection() override;
 101
 102   const entity_name_t &get_peer_name() const override {
 103     return peer_name;
 104   }
 105
 106   const entity_addr_t &get_peer_addr() const override {
 107     return peer_addr;
 108   }
 109
 110   const entity_addr_t &get_peer_socket_addr() const override {
 111     return target_addr;
 112   }
 113
 114   uint64_t get_features() const override {
 115     return features;
 116   }
 117
 118   bool is_connected() const override;
 119
 120   seastar::future<> send(MessageURef msg) override;
 121
 122   seastar::future<> send_keepalive() override;
 123
 124   clock_t::time_point get_last_keepalive() const override;
 125
 126   clock_t::time_point get_last_keepalive_ack() const override;
 127
 128   void set_last_keepalive_ack(clock_t::time_point when) override;
 129
 130   void mark_down() override;
 131
 132   bool has_user_private() const override {
 133     return user_private != nullptr;
 134   }
 135
 136   user_private_t &get_user_private() override {
 137     assert(has_user_private());
 138     return *user_private;
 139   }
 140
 141   void set_user_private(std::unique_ptr<user_private_t> new_user_private) override {
 142     assert(!has_user_private());
 143     user_private = std::move(new_user_private);
 144   }
 145
 146   void print(std::ostream& out) const override;
 147
 148  // public to SocketMessenger
 149  public:
 150   /// start a handshake from the client's perspective,
 151   /// only call when SocketConnection first construct
 152   void start_connect(const entity_addr_t& peer_addr,
 153                      const entity_name_t& peer_name);
 154
 155   /// start a handshake from the server's perspective,
 156   /// only call when SocketConnection first construct
 157   void start_accept(SocketRef&& socket,
 158                     const entity_addr_t& peer_addr);
 159
 160   seastar::future<> close_clean_yielded();
 161
 162   seastar::socket_address get_local_address() const;
 163
 164   SocketMessenger &get_messenger() const {
 165     return messenger;
 166   }
 167
 168   ConnectionRef get_local_shared_foreign_from_this();
 169
 170 private:
 171   seastar::shard_id shard_id() const;
 172
 173   void set_peer_type(entity_type_t peer_type) {
 174     // it is not allowed to assign an unknown value when the current
 175     // value is known
 176     assert(!(peer_type == 0 &&
 177              peer_name.type() != 0));
 178     // it is not allowed to assign a different known value when the
 179     // current value is also known.
 180     assert(!(peer_type != 0 &&
 181              peer_name.type() != 0 &&
 182              peer_type != peer_name.type()));
 183     peer_name._type = peer_type;
 184   }
 185
 186   void set_peer_id(int64_t peer_id) {
 187     // it is not allowed to assign an unknown value when the current
 188     // value is known
 189     assert(!(peer_id == entity_name_t::NEW &&
 190              peer_name.num() != entity_name_t::NEW));
 191     // it is not allowed to assign a different known value when the
 192     // current value is also known.
 193     assert(!(peer_id != entity_name_t::NEW &&
 194              peer_name.num() != entity_name_t::NEW &&
 195              peer_id != peer_name.num()));
 196     peer_name._num = peer_id;
 197   }
 198
 199   void set_peer_name(entity_name_t name) {
 200     set_peer_type(name.type());
 201     set_peer_id(name.num());
 202   }
 203
 204   void set_features(uint64_t f) {
 205     features = f;
 206   }
 207
 208 #ifdef UNIT_TESTS_BUILT
 209   bool is_closed_clean() const override;
 210
 211   bool is_closed() const override;
 212
 213   // peer wins if myaddr > peeraddr
 214   bool peer_wins() const override;
 215
 216   Interceptor *interceptor = nullptr;
 217 #else
 218   // peer wins if myaddr > peeraddr
 219   bool peer_wins() const;
 220 #endif
 221
 222   friend class IOHandler;
 223   friend class ProtocolV2;
 224   friend class FrameAssemblerV2;
 225 };
 226
 227 } // namespace crimson::net
 228
 229 #if FMT_VERSION >= 90000
 230 template <> struct fmt::formatter<crimson::net::SocketConnection> : fmt::ostream_formatter {};
 231 #endif