ceph/src/crimson/net/Socket.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #pragma once
   5
   6 #include <seastar/core/gate.hh>
   7 #include <seastar/core/reactor.hh>
   8 #include <seastar/core/sharded.hh>
   9 #include <seastar/net/packet.hh>
  10
  11 #include "include/buffer.h"
  12
  13 #include "crimson/common/log.h"
  14 #include "Errors.h"
  15 #include "Fwd.h"
  16
  17 #ifdef UNIT_TESTS_BUILT
  18 #include "Interceptor.h"
  19 #endif
  20
  21 namespace crimson::net {
  22
  23 class Socket;
  24 using SocketRef = std::unique_ptr<Socket>;
  25
  26 class Socket
  27 {
  28   struct construct_tag {};
  29
  30  public:
  31   // if acceptor side, peer is using a different port (ephemeral_port)
  32   // if connector side, I'm using a different port (ephemeral_port)
  33   enum class side_t {
  34     acceptor,
  35     connector
  36   };
  37
  38   Socket(seastar::connected_socket&& _socket, side_t _side, uint16_t e_port, construct_tag)
  39     : sid{seastar::this_shard_id()},
  40       socket(std::move(_socket)),
  41       in(socket.input()),
  42       // the default buffer size 8192 is too small that may impact our write
  43       // performance. see seastar::net::connected_socket::output()
  44       out(socket.output(65536)),
  45       side(_side),
  46       ephemeral_port(e_port) {}
  47
  48   ~Socket() {
  49 #ifndef NDEBUG
  50     assert(closed);
  51 #endif
  52   }
  53
  54   Socket(Socket&& o) = delete;
  55
  56   static seastar::future<SocketRef>
  57   connect(const entity_addr_t& peer_addr) {
  58     return seastar::connect(peer_addr.in4_addr()
  59     ).then([] (seastar::connected_socket socket) {
  60       return std::make_unique<Socket>(
  61         std::move(socket), side_t::connector, 0, construct_tag{});
  62     });
  63   }
  64
  65   /// read the requested number of bytes into a bufferlist
  66   seastar::future<bufferlist> read(size_t bytes);
  67   using tmp_buf = seastar::temporary_buffer<char>;
  68   using packet = seastar::net::packet;
  69   seastar::future<tmp_buf> read_exactly(size_t bytes);
  70
  71   seastar::future<> write(packet&& buf) {
  72 #ifdef UNIT_TESTS_BUILT
  73     return try_trap_pre(next_trap_write).then([buf = std::move(buf), this] () mutable {
  74 #endif
  75       return out.write(std::move(buf));
  76 #ifdef UNIT_TESTS_BUILT
  77     }).then([this] {
  78       return try_trap_post(next_trap_write);
  79     });
  80 #endif
  81   }
  82   seastar::future<> flush() {
  83     return out.flush();
  84   }
  85   seastar::future<> write_flush(packet&& buf) {
  86 #ifdef UNIT_TESTS_BUILT
  87     return try_trap_pre(next_trap_write).then([buf = std::move(buf), this] () mutable {
  88 #endif
  89       return out.write(std::move(buf)).then([this] { return out.flush(); });
  90 #ifdef UNIT_TESTS_BUILT
  91     }).then([this] {
  92       return try_trap_post(next_trap_write);
  93     });
  94 #endif
  95   }
  96
  97   // preemptively disable further reads or writes, can only be shutdown once.
  98   void shutdown();
  99
 100   /// Socket can only be closed once.
 101   seastar::future<> close();
 102
 103   // shutdown input_stream only, for tests
 104   void force_shutdown_in() {
 105     socket.shutdown_input();
 106   }
 107
 108   // shutdown output_stream only, for tests
 109   void force_shutdown_out() {
 110     socket.shutdown_output();
 111   }
 112
 113   side_t get_side() const {
 114     return side;
 115   }
 116
 117   uint16_t get_ephemeral_port() const {
 118     return ephemeral_port;
 119   }
 120
 121   // learn my ephemeral_port as connector.
 122   // unfortunately, there's no way to identify which port I'm using as
 123   // connector with current seastar interface.
 124   void learn_ephemeral_port_as_connector(uint16_t port) {
 125     assert(side == side_t::connector &&
 126            (ephemeral_port == 0 || ephemeral_port == port));
 127     ephemeral_port = port;
 128   }
 129
 130  private:
 131   const seastar::shard_id sid;
 132   seastar::connected_socket socket;
 133   seastar::input_stream<char> in;
 134   seastar::output_stream<char> out;
 135   side_t side;
 136   uint16_t ephemeral_port;
 137
 138 #ifndef NDEBUG
 139   bool closed = false;
 140 #endif
 141
 142   /// buffer state for read()
 143   struct {
 144     bufferlist buffer;
 145     size_t remaining;
 146   } r;
 147
 148 #ifdef UNIT_TESTS_BUILT
 149  public:
 150   void set_trap(bp_type_t type, bp_action_t action, socket_blocker* blocker_);
 151
 152  private:
 153   bp_action_t next_trap_read = bp_action_t::CONTINUE;
 154   bp_action_t next_trap_write = bp_action_t::CONTINUE;
 155   socket_blocker* blocker = nullptr;
 156   seastar::future<> try_trap_pre(bp_action_t& trap);
 157   seastar::future<> try_trap_post(bp_action_t& trap);
 158
 159 #endif
 160   friend class FixedCPUServerSocket;
 161 };
 162
 163 class FixedCPUServerSocket
 164     : public seastar::peering_sharded_service<FixedCPUServerSocket> {
 165   const seastar::shard_id cpu;
 166   entity_addr_t addr;
 167   std::optional<seastar::server_socket> listener;
 168   seastar::gate shutdown_gate;
 169
 170   using sharded_service_t = seastar::sharded<FixedCPUServerSocket>;
 171   std::unique_ptr<sharded_service_t> service;
 172
 173   struct construct_tag {};
 174
 175   static seastar::logger& logger() {
 176     return crimson::get_logger(ceph_subsys_ms);
 177   }
 178
 179   seastar::future<> reset() {
 180     return container().invoke_on_all([] (auto& ss) {
 181       assert(ss.shutdown_gate.is_closed());
 182       ss.shutdown_gate = seastar::gate();
 183       ss.addr = entity_addr_t();
 184       ss.listener.reset();
 185     });
 186   }
 187
 188 public:
 189   FixedCPUServerSocket(seastar::shard_id cpu, construct_tag) : cpu{cpu} {}
 190   ~FixedCPUServerSocket() {
 191     assert(!listener);
 192     // detect whether user have called destroy() properly
 193     ceph_assert(!service);
 194   }
 195
 196   FixedCPUServerSocket(FixedCPUServerSocket&&) = delete;
 197   FixedCPUServerSocket(const FixedCPUServerSocket&) = delete;
 198   FixedCPUServerSocket& operator=(const FixedCPUServerSocket&) = delete;
 199
 200   using listen_ertr = crimson::errorator<
 201     crimson::ct_error::address_in_use // The address is already bound
 202     >;
 203   listen_ertr::future<> listen(entity_addr_t addr);
 204
 205   // fn_accept should be a nothrow function of type
 206   // seastar::future<>(SocketRef, entity_addr_t)
 207   template <typename Func>
 208   seastar::future<> accept(Func&& fn_accept) {
 209     assert(seastar::this_shard_id() == cpu);
 210     logger().trace("FixedCPUServerSocket({})::accept()...", addr);
 211     return container().invoke_on_all(
 212         [fn_accept = std::move(fn_accept)] (auto& ss) mutable {
 213       assert(ss.listener);
 214       // gate accepting
 215       // FixedCPUServerSocket::shutdown() will drain the continuations in the gate
 216       // so ignore the returned future
 217       std::ignore = seastar::with_gate(ss.shutdown_gate,
 218           [&ss, fn_accept = std::move(fn_accept)] () mutable {
 219         return seastar::keep_doing([&ss, fn_accept = std::move(fn_accept)] () mutable {
 220           return ss.listener->accept().then(
 221               [&ss, fn_accept = std::move(fn_accept)]
 222               (seastar::accept_result accept_result) mutable {
 223             // assert seastar::listen_options::set_fixed_cpu() works
 224             assert(seastar::this_shard_id() == ss.cpu);
 225             auto [socket, paddr] = std::move(accept_result);
 226             entity_addr_t peer_addr;
 227             peer_addr.set_sockaddr(&paddr.as_posix_sockaddr());
 228             peer_addr.set_type(entity_addr_t::TYPE_ANY);
 229             SocketRef _socket = std::make_unique<Socket>(
 230                 std::move(socket), Socket::side_t::acceptor,
 231                 peer_addr.get_port(), Socket::construct_tag{});
 232             std::ignore = seastar::with_gate(ss.shutdown_gate,
 233                 [socket = std::move(_socket), peer_addr,
 234                  &ss, fn_accept = std::move(fn_accept)] () mutable {
 235               logger().trace("FixedCPUServerSocket({})::accept(): "
 236                              "accepted peer {}", ss.addr, peer_addr);
 237               return fn_accept(std::move(socket), peer_addr
 238               ).handle_exception([&ss, peer_addr] (auto eptr) {
 239                 logger().error("FixedCPUServerSocket({})::accept(): "
 240                                "fn_accept(s, {}) got unexpected exception {}",
 241                                ss.addr, peer_addr, eptr);
 242                 ceph_abort();
 243               });
 244             });
 245           });
 246         }).handle_exception_type([&ss] (const std::system_error& e) {
 247           if (e.code() == std::errc::connection_aborted ||
 248               e.code() == std::errc::invalid_argument) {
 249             logger().trace("FixedCPUServerSocket({})::accept(): stopped ({})",
 250                            ss.addr, e);
 251           } else {
 252             throw;
 253           }
 254         }).handle_exception([&ss] (auto eptr) {
 255           logger().error("FixedCPUServerSocket({})::accept(): "
 256                          "got unexpected exception {}", ss.addr, eptr);
 257           ceph_abort();
 258         });
 259       });
 260     });
 261   }
 262
 263   seastar::future<> shutdown();
 264   seastar::future<> destroy();
 265   static seastar::future<FixedCPUServerSocket*> create();
 266 };
 267
 268 } // namespace crimson::net