ceph/src/msg/xio/XioMessenger.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) 2004-2006 Sage Weil <sage@newdream.net>
   7  * Portions Copyright (C) 2013 CohortFS, LLC
   8  *
   9  * This is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Lesser General Public
  11  * License version 2.1, as published by the Free Software
  12  * Foundation.  See file COPYING.
  13  *
  14  */
  15
  16 #ifndef XIO_MESSENGER_H
  17 #define XIO_MESSENGER_H
  18
  19 #include "msg/SimplePolicyMessenger.h"
  20
  21 #include <atomic>
  22
  23 extern "C" {
  24 #include "libxio.h"
  25 }
  26
  27 #include "XioConnection.h"
  28 #include "XioPortal.h"
  29 #include "QueueStrategy.h"
  30 #include "common/Thread.h"
  31 #include "common/Mutex.h"
  32 #include "include/Spinlock.h"
  33
  34 class XioInit {
  35   /* safe to be called multiple times */
  36   void package_init(CephContext *cct);
  37
  38 protected:
  39   XioInit(CephContext *cct) {
  40     this->package_init(cct);
  41   }
  42 };
  43
  44 class XioMessenger : public SimplePolicyMessenger, XioInit
  45 {
  46 private:
  47   static std::atomic<uint64_t> nInstances = { 0 };
  48   std::atomic<uint64_t> nsessions = { 0 };
  49   std::atomic<bool> shutdown_called = { false };
  50   Spinlock conns_sp;
  51   XioConnection::ConnList conns_list;
  52   XioConnection::EntitySet conns_entity_map;
  53   XioPortals portals;
  54   DispatchStrategy* dispatch_strategy;
  55   XioLoopbackConnectionRef loop_con;
  56   uint32_t special_handling;
  57   Mutex sh_mtx;
  58   Cond sh_cond;
  59   bool need_addr;
  60   bool did_bind;
  61
  62   /// approximately unique ID set by the Constructor for use in entity_addr_t
  63   uint64_t nonce;
  64
  65   friend class XioConnection;
  66
  67 public:
  68   XioMessenger(CephContext *cct, entity_name_t name,
  69                string mname, uint64_t nonce,
  70                uint64_t cflags = 0,
  71                DispatchStrategy* ds = new QueueStrategy(1));
  72
  73   virtual ~XioMessenger();
  74
  75   XioPortal* get_portal() { return portals.get_next_portal(); }
  76
  77   virtual void set_myaddr(const entity_addr_t& a) {
  78     Messenger::set_myaddr(a);
  79     loop_con->set_peer_addr(a);
  80   }
  81
  82   int _send_message(Message *m, const entity_inst_t &dest);
  83   int _send_message(Message *m, Connection *con);
  84   int _send_message_impl(Message *m, XioConnection *xcon);
  85
  86   uint32_t get_special_handling() { return special_handling; }
  87   void set_special_handling(int n) { special_handling = n; }
  88   int pool_hint(uint32_t size);
  89   void try_insert(XioConnection *xcon);
  90
  91   /* xio hooks */
  92   int new_session(struct xio_session *session,
  93                   struct xio_new_session_req *req,
  94                   void *cb_user_context);
  95
  96   int session_event(struct xio_session *session,
  97                     struct xio_session_event_data *event_data,
  98                     void *cb_user_context);
  99
 100   /* Messenger interface */
 101   virtual void set_addr_unknowns(const entity_addr_t &addr) override
 102     { } /* XXX applicable? */
 103
 104   virtual int get_dispatch_queue_len()
 105     { return 0; } /* XXX bogus? */
 106
 107   virtual double get_dispatch_queue_max_age(utime_t now)
 108     { return 0; } /* XXX bogus? */
 109
 110   virtual void set_cluster_protocol(int p)
 111     { }
 112
 113   virtual int bind(const entity_addr_t& addr);
 114
 115   virtual int rebind(const set<int>& avoid_ports);
 116
 117   virtual int start();
 118
 119   virtual void wait();
 120
 121   virtual int shutdown();
 122
 123   virtual int send_message(Message *m, const entity_inst_t &dest) {
 124     return _send_message(m, dest);
 125   }
 126
 127   virtual int lazy_send_message(Message *m, const entity_inst_t& dest)
 128     { return EINVAL; }
 129
 130   virtual int lazy_send_message(Message *m, Connection *con)
 131     { return EINVAL; }
 132
 133   virtual ConnectionRef get_connection(const entity_inst_t& dest);
 134
 135   virtual ConnectionRef get_loopback_connection();
 136
 137   void unregister_xcon(XioConnection *xcon);
 138   virtual void mark_down(const entity_addr_t& a);
 139   virtual void mark_down(Connection *con);
 140   virtual void mark_down_all();
 141   virtual void mark_down_on_empty(Connection *con);
 142   virtual void mark_disposable(Connection *con);
 143
 144   void ds_dispatch(Message *m)
 145     { dispatch_strategy->ds_dispatch(m); }
 146
 147   /**
 148    * Tell the XioMessenger its full IP address.
 149    *
 150    * This is used by clients when connecting to other endpoints, and
 151    * probably shouldn't be called by anybody else.
 152    */
 153   void learned_addr(const entity_addr_t& peer_addr_for_me);
 154
 155 private:
 156   int get_nconns_per_portal(uint64_t cflags);
 157   int get_nportals(uint64_t cflags);
 158
 159 protected:
 160   virtual void ready()
 161     { }
 162 };
 163
 164 XioCommand* pool_alloc_xio_command(XioConnection *xcon);
 165
 166
 167 #endif /* XIO_MESSENGER_H */