1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 * Ceph - scalable distributed file system
6 * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
7 * Portions Copyright (C) 2013 CohortFS, LLC
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.
16 #ifndef XIO_MESSENGER_H
17 #define XIO_MESSENGER_H
19 #include "msg/SimplePolicyMessenger.h"
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"
35 /* safe to be called multiple times */
36 void package_init(CephContext *cct);
39 XioInit(CephContext *cct) {
40 this->package_init(cct);
44 class XioMessenger : public SimplePolicyMessenger, XioInit
47 static std::atomic<uint64_t> nInstances = { 0 };
48 std::atomic<uint64_t> nsessions = { 0 };
49 std::atomic<bool> shutdown_called = { false };
51 XioConnection::ConnList conns_list;
52 XioConnection::EntitySet conns_entity_map;
54 DispatchStrategy* dispatch_strategy;
55 XioLoopbackConnectionRef loop_con;
56 uint32_t special_handling;
62 /// approximately unique ID set by the Constructor for use in entity_addr_t
65 friend class XioConnection;
68 XioMessenger(CephContext *cct, entity_name_t name,
69 string mname, uint64_t nonce,
71 DispatchStrategy* ds = new QueueStrategy(1));
73 virtual ~XioMessenger();
75 XioPortal* get_portal() { return portals.get_next_portal(); }
77 virtual void set_myaddr(const entity_addr_t& a) {
78 Messenger::set_myaddr(a);
79 loop_con->set_peer_addr(a);
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);
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);
92 int new_session(struct xio_session *session,
93 struct xio_new_session_req *req,
94 void *cb_user_context);
96 int session_event(struct xio_session *session,
97 struct xio_session_event_data *event_data,
98 void *cb_user_context);
100 /* Messenger interface */
101 virtual void set_addr_unknowns(const entity_addr_t &addr) override
102 { } /* XXX applicable? */
103 virtual void set_addr(const entity_addr_t &addr) override
104 { } /* XXX applicable? */
106 virtual int get_dispatch_queue_len()
107 { return 0; } /* XXX bogus? */
109 virtual double get_dispatch_queue_max_age(utime_t now)
110 { return 0; } /* XXX bogus? */
112 virtual void set_cluster_protocol(int p)
115 virtual int bind(const entity_addr_t& addr);
117 virtual int rebind(const set<int>& avoid_ports);
123 virtual int shutdown();
125 virtual int send_message(Message *m, const entity_inst_t &dest) {
126 return _send_message(m, dest);
129 virtual int lazy_send_message(Message *m, const entity_inst_t& dest)
132 virtual int lazy_send_message(Message *m, Connection *con)
135 virtual ConnectionRef get_connection(const entity_inst_t& dest);
137 virtual ConnectionRef get_loopback_connection();
139 void unregister_xcon(XioConnection *xcon);
140 virtual void mark_down(const entity_addr_t& a);
141 virtual void mark_down(Connection *con);
142 virtual void mark_down_all();
143 virtual void mark_down_on_empty(Connection *con);
144 virtual void mark_disposable(Connection *con);
146 void ds_dispatch(Message *m)
147 { dispatch_strategy->ds_dispatch(m); }
150 * Tell the XioMessenger its full IP address.
152 * This is used by clients when connecting to other endpoints, and
153 * probably shouldn't be called by anybody else.
155 void learned_addr(const entity_addr_t& peer_addr_for_me);
158 int get_nconns_per_portal(uint64_t cflags);
159 int get_nportals(uint64_t cflags);
166 XioCommand* pool_alloc_xio_command(XioConnection *xcon);
169 #endif /* XIO_MESSENGER_H */