--- /dev/null
+// -*- 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>
+ *
+ * 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.
+ *
+ */
+
+#include "DirectMessenger.h"
+#include "msg/DispatchStrategy.h"
+
+
+class DirectConnection : public Connection {
+ /// sent messages are dispatched here
+ DispatchStrategy *const dispatchers;
+
+ /// the connection that will be attached to outgoing messages, so that replies
+ /// can be dispatched back to the sender. the pointer is atomic for
+ /// thread-safety between mark_down() and send_message(). no reference is held
+ /// on this Connection to avoid cyclical refs. we don't need a reference
+ /// because its owning DirectMessenger will mark both connections down (and
+ /// clear this pointer) before dropping its own reference
+ std::atomic<Connection*> reply_connection{nullptr};
+
+ public:
+ DirectConnection(CephContext *cct, DirectMessenger *m,
+ DispatchStrategy *dispatchers)
+ : Connection(cct, m),
+ dispatchers(dispatchers)
+ {}
+
+ /// sets the Connection that will receive replies to outgoing messages
+ void set_direct_reply_connection(ConnectionRef conn);
+
+ /// return true if a peer connection exists
+ bool is_connected() override;
+
+ /// pass the given message directly to our dispatchers
+ int send_message(Message *m) override;
+
+ /// release our pointer to the peer connection. later calls to is_connected()
+ /// will return false, and send_message() will fail with -ENOTCONN
+ void mark_down() override;
+
+ /// noop - keepalive messages are not needed within a process
+ void send_keepalive() override {}
+
+ /// noop - reconnect/recovery semantics are not needed within a process
+ void mark_disposable() override {}
+};
+
+void DirectConnection::set_direct_reply_connection(ConnectionRef conn)
+{
+ reply_connection.store(conn.get());
+}
+
+bool DirectConnection::is_connected()
+{
+ // true between calls to set_direct_reply_connection() and mark_down()
+ return reply_connection.load() != nullptr;
+}
+
+int DirectConnection::send_message(Message *m)
+{
+ // read reply_connection atomically and take a reference
+ ConnectionRef conn = reply_connection.load();
+ if (!conn) {
+ m->put();
+ return -ENOTCONN;
+ }
+ // attach reply_connection to the Message, so that calls to
+ // m->get_connection()->send_message() can be dispatched back to the sender
+ m->set_connection(conn);
+
+ dispatchers->ds_dispatch(m);
+ return 0;
+}
+
+void DirectConnection::mark_down()
+{
+ Connection *conn = reply_connection.load();
+ if (!conn) {
+ return; // already marked down
+ }
+ if (!reply_connection.compare_exchange_weak(conn, nullptr)) {
+ return; // lost the race to mark down
+ }
+ // called only once to avoid loops
+ conn->mark_down();
+}
+
+
+static ConnectionRef create_loopback(DirectMessenger *m,
+ entity_name_t name,
+ DispatchStrategy *dispatchers)
+{
+ auto loopback = boost::intrusive_ptr<DirectConnection>(
+ new DirectConnection(m->cct, m, dispatchers));
+ // loopback replies go to itself
+ loopback->set_direct_reply_connection(loopback);
+ loopback->set_peer_type(name.type());
+ loopback->set_features(CEPH_FEATURES_ALL);
+ return loopback;
+}
+
+DirectMessenger::DirectMessenger(CephContext *cct, entity_name_t name,
+ string mname, uint64_t nonce,
+ DispatchStrategy *dispatchers)
+ : SimplePolicyMessenger(cct, name, mname, nonce),
+ dispatchers(dispatchers),
+ loopback_connection(create_loopback(this, name, dispatchers))
+{
+ dispatchers->set_messenger(this);
+}
+
+DirectMessenger::~DirectMessenger()
+{
+}
+
+int DirectMessenger::set_direct_peer(DirectMessenger *peer)
+{
+ if (get_myinst() == peer->get_myinst()) {
+ return -EADDRINUSE; // must have a different entity instance
+ }
+ peer_inst = peer->get_myinst();
+
+ // allocate a Connection that dispatches to the peer messenger
+ auto direct_connection = boost::intrusive_ptr<DirectConnection>(
+ new DirectConnection(cct, peer, peer->dispatchers.get()));
+
+ direct_connection->set_peer_addr(peer_inst.addr);
+ direct_connection->set_peer_type(peer_inst.name.type());
+ direct_connection->set_features(CEPH_FEATURES_ALL);
+
+ // if set_direct_peer() was already called on the peer messenger, we can
+ // finish by attaching their connections. if not, the later call to
+ // peer->set_direct_peer() will attach their connection to ours
+ auto connection = peer->get_connection(get_myinst());
+ if (connection) {
+ auto p = static_cast<DirectConnection*>(connection.get());
+
+ p->set_direct_reply_connection(direct_connection);
+ direct_connection->set_direct_reply_connection(p);
+ }
+
+ peer_connection = std::move(direct_connection);
+ return 0;
+}
+
+int DirectMessenger::bind(const entity_addr_t &bind_addr)
+{
+ if (peer_connection) {
+ return -EINVAL; // can't change address after sharing it with the peer
+ }
+ set_myaddr(bind_addr);
+ loopback_connection->set_peer_addr(bind_addr);
+ return 0;
+}
+
+int DirectMessenger::client_bind(const entity_addr_t &bind_addr)
+{
+ // same as bind
+ return bind(bind_addr);
+}
+
+int DirectMessenger::start()
+{
+ if (!peer_connection) {
+ return -EINVAL; // did not connect to a peer
+ }
+ if (started) {
+ return -EINVAL; // already started
+ }
+
+ dispatchers->start();
+ return SimplePolicyMessenger::start();
+}
+
+int DirectMessenger::shutdown()
+{
+ if (!started) {
+ return -EINVAL; // not started
+ }
+
+ mark_down_all();
+ peer_connection.reset();
+ loopback_connection.reset();
+
+ dispatchers->shutdown();
+ SimplePolicyMessenger::shutdown();
+ sem.Put(); // signal wait()
+ return 0;
+}
+
+void DirectMessenger::wait()
+{
+ sem.Get(); // wait on signal from shutdown()
+ dispatchers->wait();
+}
+
+ConnectionRef DirectMessenger::get_connection(const entity_inst_t& dst)
+{
+ if (dst == peer_inst) {
+ return peer_connection;
+ }
+ if (dst == get_myinst()) {
+ return loopback_connection;
+ }
+ return nullptr;
+}
+
+ConnectionRef DirectMessenger::get_loopback_connection()
+{
+ return loopback_connection;
+}
+
+int DirectMessenger::send_message(Message *m, const entity_inst_t& dst)
+{
+ auto conn = get_connection(dst);
+ if (!conn) {
+ m->put();
+ return -ENOTCONN;
+ }
+ return conn->send_message(m);
+}
+
+void DirectMessenger::mark_down(const entity_addr_t& addr)
+{
+ ConnectionRef conn;
+ if (addr == peer_inst.addr) {
+ conn = peer_connection;
+ } else if (addr == get_myaddr()) {
+ conn = loopback_connection;
+ }
+ if (conn) {
+ conn->mark_down();
+ }
+}
+
+void DirectMessenger::mark_down_all()
+{
+ if (peer_connection) {
+ peer_connection->mark_down();
+ }
+ loopback_connection->mark_down();
+}
Code Review / stor4nfv.git / blobdiff