[ceph.git] / ceph / src / msg / xio / XioMessenger.h

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
 * Portions Copyright (C) 2013 CohortFS, LLC
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software
 * Foundation.  See file COPYING.
 *
 */

#ifndef XIO_MESSENGER_H
#define XIO_MESSENGER_H

#include "msg/SimplePolicyMessenger.h"
extern "C" {
#include "libxio.h"
}
#include "XioConnection.h"
#include "XioPortal.h"
#include "QueueStrategy.h"
#include "include/atomic.h"
#include "common/Thread.h"
#include "common/Mutex.h"
#include "include/Spinlock.h"

class XioInit {
  /* safe to be called multiple times */
  void package_init(CephContext *cct);

protected:
  XioInit(CephContext *cct) {
    this->package_init(cct);
  }
};

class XioMessenger : public SimplePolicyMessenger, XioInit
{
private:
  static atomic_t nInstances;
  atomic_t nsessions;
  atomic_t shutdown_called;
  Spinlock conns_sp;
  XioConnection::ConnList conns_list;
  XioConnection::EntitySet conns_entity_map;
  XioPortals portals;
  DispatchStrategy* dispatch_strategy;
  XioLoopbackConnectionRef loop_con;
  uint32_t special_handling;
  Mutex sh_mtx;
  Cond sh_cond;
  bool need_addr;
  bool did_bind;

  /// approximately unique ID set by the Constructor for use in entity_addr_t
  uint64_t nonce;

  friend class XioConnection;

public:
  XioMessenger(CephContext *cct, entity_name_t name,
	       string mname, uint64_t nonce,
	       uint64_t cflags = 0,
	       DispatchStrategy* ds = new QueueStrategy(1));

  virtual ~XioMessenger();

  XioPortal* get_portal() { return portals.get_next_portal(); }

  virtual void set_myaddr(const entity_addr_t& a) {
    Messenger::set_myaddr(a);
    loop_con->set_peer_addr(a);
  }

  int _send_message(Message *m, const entity_inst_t &dest);
  int _send_message(Message *m, Connection *con);
  int _send_message_impl(Message *m, XioConnection *xcon);

  uint32_t get_special_handling() { return special_handling; }
  void set_special_handling(int n) { special_handling = n; }
  int pool_hint(uint32_t size);
  void try_insert(XioConnection *xcon);

  /* xio hooks */
  int new_session(struct xio_session *session,
		  struct xio_new_session_req *req,
		  void *cb_user_context);

  int session_event(struct xio_session *session,
		    struct xio_session_event_data *event_data,
		    void *cb_user_context);

  /* Messenger interface */
  virtual void set_addr_unknowns(const entity_addr_t &addr) override
    { } /* XXX applicable? */

  virtual int get_dispatch_queue_len()
    { return 0; } /* XXX bogus? */

  virtual double get_dispatch_queue_max_age(utime_t now)
    { return 0; } /* XXX bogus? */

  virtual void set_cluster_protocol(int p)
    { }

  virtual int bind(const entity_addr_t& addr);

  virtual int rebind(const set<int>& avoid_ports);

  virtual int start();

  virtual void wait();

  virtual int shutdown();

  virtual int send_message(Message *m, const entity_inst_t &dest) {
    return _send_message(m, dest);
  }

  virtual int lazy_send_message(Message *m, const entity_inst_t& dest)
    { return EINVAL; }

  virtual int lazy_send_message(Message *m, Connection *con)
    { return EINVAL; }

  virtual ConnectionRef get_connection(const entity_inst_t& dest);

  virtual ConnectionRef get_loopback_connection();

  void unregister_xcon(XioConnection *xcon);
  virtual void mark_down(const entity_addr_t& a);
  virtual void mark_down(Connection *con);
  virtual void mark_down_all();
  virtual void mark_down_on_empty(Connection *con);
  virtual void mark_disposable(Connection *con);

  void ds_dispatch(Message *m)
    { dispatch_strategy->ds_dispatch(m); }

  /**
   * Tell the XioMessenger its full IP address.
   *
   * This is used by clients when connecting to other endpoints, and
   * probably shouldn't be called by anybody else.
   */
  void learned_addr(const entity_addr_t& peer_addr_for_me);

private:
  int get_nconns_per_portal(uint64_t cflags);
  int get_nportals(uint64_t cflags);

protected:
  virtual void ready()
    { }
};

XioCommand* pool_alloc_xio_command(XioConnection *xcon);


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