ceph/src/crimson/net/Messenger.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 "Fwd.h"
  18 #include "crimson/common/throttle.h"
  19 #include "msg/Message.h"
  20 #include "msg/Policy.h"
  21
  22 class AuthAuthorizer;
  23
  24 namespace crimson::auth {
  25 class AuthClient;
  26 class AuthServer;
  27 }
  28
  29 namespace crimson::net {
  30
  31 #ifdef UNIT_TESTS_BUILT
  32 class Interceptor;
  33 #endif
  34
  35 using Throttle = crimson::common::Throttle;
  36 using SocketPolicy = ceph::net::Policy<Throttle>;
  37
  38 class Messenger {
  39   uint32_t crc_flags = 0;
  40   crimson::auth::AuthClient* auth_client = nullptr;
  41   crimson::auth::AuthServer* auth_server = nullptr;
  42   bool require_authorizer = true;
  43
  44 protected:
  45   entity_name_t my_name;
  46   entity_addrvec_t my_addrs;
  47
  48 public:
  49   Messenger(const entity_name_t& name)
  50     : my_name(name)
  51   {}
  52   virtual ~Messenger() {}
  53
  54 #ifdef UNIT_TESTS_BUILT
  55   Interceptor *interceptor = nullptr;
  56 #endif
  57
  58   entity_type_t get_mytype() const { return my_name.type(); }
  59   const entity_name_t& get_myname() const { return my_name; }
  60   const entity_addrvec_t& get_myaddrs() const { return my_addrs; }
  61   entity_addr_t get_myaddr() const { return my_addrs.front(); }
  62   virtual seastar::future<> set_myaddrs(const entity_addrvec_t& addrs) {
  63     my_addrs = addrs;
  64     return seastar::now();
  65   }
  66   virtual bool set_addr_unknowns(const entity_addrvec_t &addrs) = 0;
  67
  68   using bind_ertr = crimson::errorator<
  69     crimson::ct_error::address_in_use, // The address (range) is already bound
  70     crimson::ct_error::address_not_available
  71     >;
  72   /// bind to the given address
  73   virtual bind_ertr::future<> bind(const entity_addrvec_t& addr) = 0;
  74
  75   /// start the messenger
  76   virtual seastar::future<> start(const dispatchers_t&) = 0;
  77
  78   /// either return an existing connection to the peer,
  79   /// or a new pending connection
  80   virtual ConnectionRef
  81   connect(const entity_addr_t& peer_addr,
  82           const entity_name_t& peer_name) = 0;
  83
  84   ConnectionRef
  85   connect(const entity_addr_t& peer_addr,
  86           const entity_type_t& peer_type) {
  87     return connect(peer_addr, entity_name_t(peer_type, -1));
  88   }
  89
  90   // wait for messenger shutdown
  91   virtual seastar::future<> wait() = 0;
  92
  93   // stop dispatching events and messages
  94   virtual void stop() = 0;
  95
  96   virtual bool is_started() const = 0;
  97
  98   // free internal resources before destruction, must be called after stopped,
  99   // and must be called if is bound.
 100   virtual seastar::future<> shutdown() = 0;
 101
 102   uint32_t get_crc_flags() const {
 103     return crc_flags;
 104   }
 105   void set_crc_data() {
 106     crc_flags |= MSG_CRC_DATA;
 107   }
 108   void set_crc_header() {
 109     crc_flags |= MSG_CRC_HEADER;
 110   }
 111
 112   crimson::auth::AuthClient* get_auth_client() const { return auth_client; }
 113   void set_auth_client(crimson::auth::AuthClient *ac) {
 114     auth_client = ac;
 115   }
 116   crimson::auth::AuthServer* get_auth_server() const { return auth_server; }
 117   void set_auth_server(crimson::auth::AuthServer *as) {
 118     auth_server = as;
 119   }
 120
 121   virtual void print(std::ostream& out) const = 0;
 122
 123   virtual SocketPolicy get_policy(entity_type_t peer_type) const = 0;
 124
 125   virtual SocketPolicy get_default_policy() const = 0;
 126
 127   virtual void set_default_policy(const SocketPolicy& p) = 0;
 128
 129   virtual void set_policy(entity_type_t peer_type, const SocketPolicy& p) = 0;
 130
 131   virtual void set_policy_throttler(entity_type_t peer_type, Throttle* throttle) = 0;
 132
 133   // allow unauthenticated connections.  This is needed for compatibility with
 134   // pre-nautilus OSDs, which do not authenticate the heartbeat sessions.
 135   bool get_require_authorizer() const {
 136     return require_authorizer;
 137   }
 138   void set_require_authorizer(bool r) {
 139     require_authorizer = r;
 140   }
 141   static MessengerRef
 142   create(const entity_name_t& name,
 143          const std::string& lname,
 144          const uint64_t nonce);
 145 };
 146
 147 inline std::ostream& operator<<(std::ostream& out, const Messenger& msgr) {
 148   out << "[";
 149   msgr.print(out);
 150   out << "]";
 151   return out;
 152 }
 153
 154 } // namespace crimson::net