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/net/Connection.h"
  21 #include "crimson/net/Socket.h"
  22 #include "crimson/thread/Throttle.h"
  23
  24 namespace crimson::net {
  25
  26 class Dispatcher;
  27 class Protocol;
  28 class SocketMessenger;
  29 class SocketConnection;
  30 using SocketConnectionRef = seastar::shared_ptr<SocketConnection>;
  31
  32 class SocketConnection : public Connection {
  33   SocketMessenger& messenger;
  34   std::unique_ptr<Protocol> protocol;
  35
  36   // if acceptor side, ephemeral_port is different from peer_addr.get_port();
  37   // if connector side, ephemeral_port is different from my_addr.get_port().
  38   enum class side_t {
  39     none,
  40     acceptor,
  41     connector
  42   };
  43   side_t side = side_t::none;
  44   uint16_t ephemeral_port = 0;
  45   void set_ephemeral_port(uint16_t port, side_t _side) {
  46     ephemeral_port = port;
  47     side = _side;
  48   }
  49
  50   ceph::net::Policy<crimson::thread::Throttle> policy;
  51
  52   /// the seq num of the last transmitted message
  53   seq_num_t out_seq = 0;
  54   /// the seq num of the last received message
  55   seq_num_t in_seq = 0;
  56   /// update the seq num of last received message
  57   /// @returns true if the @c seq is valid, and @c in_seq is updated,
  58   ///          false otherwise.
  59   bool update_rx_seq(seq_num_t seq);
  60
  61   // messages to be resent after connection gets reset
  62   std::deque<MessageRef> out_q;
  63   std::deque<MessageRef> pending_q;
  64   // messages sent, but not yet acked by peer
  65   std::deque<MessageRef> sent;
  66
  67   seastar::shard_id shard_id() const;
  68
  69  public:
  70   SocketConnection(SocketMessenger& messenger,
  71                    Dispatcher& dispatcher,
  72                    bool is_msgr2);
  73   ~SocketConnection() override;
  74
  75   Messenger* get_messenger() const override;
  76
  77   bool is_connected() const override;
  78
  79 #ifdef UNIT_TESTS_BUILT
  80   bool is_closed() const override;
  81
  82   bool peer_wins() const override;
  83 #else
  84   bool peer_wins() const;
  85 #endif
  86
  87   seastar::future<> send(MessageRef msg) override;
  88
  89   seastar::future<> keepalive() override;
  90
  91   seastar::future<> close() override;
  92
  93   void print(ostream& out) const override;
  94
  95   /// start a handshake from the client's perspective,
  96   /// only call when SocketConnection first construct
  97   void start_connect(const entity_addr_t& peer_addr,
  98                      const entity_type_t& peer_type);
  99   /// start a handshake from the server's perspective,
 100   /// only call when SocketConnection first construct
 101   void start_accept(SocketRef&& socket,
 102                     const entity_addr_t& peer_addr);
 103
 104   bool is_server_side() const {
 105     return policy.server;
 106   }
 107
 108   bool is_lossy() const {
 109     return policy.lossy;
 110   }
 111
 112   friend class Protocol;
 113   friend class ProtocolV1;
 114   friend class ProtocolV2;
 115 };
 116
 117 } // namespace crimson::net