+++ /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) 2014 UnitedStack <haomai@unitedstack.com>
- *
- * Author: Haomai Wang <haomaiwang@gmail.com>
- *
- * 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 CEPH_ASYNCMESSENGER_H
-#define CEPH_ASYNCMESSENGER_H
-
-#include "include/types.h"
-#include "include/xlist.h"
-
-#include <map>
-using namespace std;
-#include "include/unordered_map.h"
-#include "include/unordered_set.h"
-
-#include "common/Mutex.h"
-#include "common/Cond.h"
-#include "common/Thread.h"
-
-#include "include/Spinlock.h"
-
-#include "msg/SimplePolicyMessenger.h"
-#include "msg/DispatchQueue.h"
-#include "include/assert.h"
-#include "AsyncConnection.h"
-#include "Event.h"
-
-
-class AsyncMessenger;
-
-/**
- * If the Messenger binds to a specific address, the Processor runs
- * and listens for incoming connections.
- */
-class Processor {
- AsyncMessenger *msgr;
- NetHandler net;
- Worker *worker;
- ServerSocket listen_socket;
- EventCallbackRef listen_handler;
-
- class C_processor_accept;
-
- public:
- Processor(AsyncMessenger *r, Worker *w, CephContext *c);
- ~Processor() { delete listen_handler; };
-
- void stop();
- int bind(const entity_addr_t &bind_addr,
- const set<int>& avoid_ports,
- entity_addr_t* bound_addr);
- void start();
- void accept();
-};
-
-/*
- * AsyncMessenger is represented for maintaining a set of asynchronous connections,
- * it may own a bind address and the accepted connections will be managed by
- * AsyncMessenger.
- *
- */
-
-class AsyncMessenger : public SimplePolicyMessenger {
- // First we have the public Messenger interface implementation...
-public:
- /**
- * Initialize the AsyncMessenger!
- *
- * @param cct The CephContext to use
- * @param name The name to assign ourselves
- * _nonce A unique ID to use for this AsyncMessenger. It should not
- * be a value that will be repeated if the daemon restarts.
- */
- AsyncMessenger(CephContext *cct, entity_name_t name, const std::string &type,
- string mname, uint64_t _nonce);
-
- /**
- * Destroy the AsyncMessenger. Pretty simple since all the work is done
- * elsewhere.
- */
- ~AsyncMessenger() override;
-
- /** @defgroup Accessors
- * @{
- */
- void set_addr_unknowns(const entity_addr_t &addr) override;
- void set_addr(const entity_addr_t &addr) override;
-
- int get_dispatch_queue_len() override {
- return dispatch_queue.get_queue_len();
- }
-
- double get_dispatch_queue_max_age(utime_t now) override {
- return dispatch_queue.get_max_age(now);
- }
- /** @} Accessors */
-
- /**
- * @defgroup Configuration functions
- * @{
- */
- void set_cluster_protocol(int p) override {
- assert(!started && !did_bind);
- cluster_protocol = p;
- }
-
- int bind(const entity_addr_t& bind_addr) override;
- int rebind(const set<int>& avoid_ports) override;
- int client_bind(const entity_addr_t& bind_addr) override;
-
- /** @} Configuration functions */
-
- /**
- * @defgroup Startup/Shutdown
- * @{
- */
- int start() override;
- void wait() override;
- int shutdown() override;
-
- /** @} // Startup/Shutdown */
-
- /**
- * @defgroup Messaging
- * @{
- */
- int send_message(Message *m, const entity_inst_t& dest) override {
- Mutex::Locker l(lock);
-
- return _send_message(m, dest);
- }
-
- /** @} // Messaging */
-
- /**
- * @defgroup Connection Management
- * @{
- */
- ConnectionRef get_connection(const entity_inst_t& dest) override;
- ConnectionRef get_loopback_connection() override;
- void mark_down(const entity_addr_t& addr) override;
- void mark_down_all() override {
- shutdown_connections(true);
- }
- /** @} // Connection Management */
-
- /**
- * @defgroup Inner classes
- * @{
- */
-
- /**
- * @} // Inner classes
- */
-
-protected:
- /**
- * @defgroup Messenger Interfaces
- * @{
- */
- /**
- * Start up the DispatchQueue thread once we have somebody to dispatch to.
- */
- void ready() override;
- /** @} // Messenger Interfaces */
-
-private:
-
- /**
- * @defgroup Utility functions
- * @{
- */
-
- /**
- * Create a connection associated with the given entity (of the given type).
- * Initiate the connection. (This function returning does not guarantee
- * connection success.)
- *
- * @param addr The address of the entity to connect to.
- * @param type The peer type of the entity at the address.
- *
- * @return a pointer to the newly-created connection. Caller does not own a
- * reference; take one if you need it.
- */
- AsyncConnectionRef create_connect(const entity_addr_t& addr, int type);
-
- /**
- * Queue up a Message for delivery to the entity specified
- * by addr and dest_type.
- * submit_message() is responsible for creating
- * new AsyncConnection (and closing old ones) as necessary.
- *
- * @param m The Message to queue up. This function eats a reference.
- * @param con The existing Connection to use, or NULL if you don't know of one.
- * @param dest_addr The address to send the Message to.
- * @param dest_type The peer type of the address we're sending to
- * just drop silently under failure.
- */
- void submit_message(Message *m, AsyncConnectionRef con,
- const entity_addr_t& dest_addr, int dest_type);
-
- int _send_message(Message *m, const entity_inst_t& dest);
- void _finish_bind(const entity_addr_t& bind_addr,
- const entity_addr_t& listen_addr);
-
- private:
- static const uint64_t ReapDeadConnectionThreshold = 5;
-
- NetworkStack *stack;
- std::vector<Processor*> processors;
- friend class Processor;
- DispatchQueue dispatch_queue;
-
- // the worker run messenger's cron jobs
- Worker *local_worker;
-
- std::string ms_type;
-
- /// overall lock used for AsyncMessenger data structures
- Mutex lock;
- // AsyncMessenger stuff
- /// approximately unique ID set by the Constructor for use in entity_addr_t
- uint64_t nonce;
-
- /// true, specifying we haven't learned our addr; set false when we find it.
- // maybe this should be protected by the lock?
- bool need_addr;
-
- /**
- * set to bind address if bind was called before NetworkStack was ready to
- * bind
- */
- entity_addr_t pending_bind_addr;
-
- /**
- * false; set to true if a pending bind exists
- */
- bool pending_bind = false;
-
- /**
- * The following aren't lock-protected since you shouldn't be able to race
- * the only writers.
- */
-
- /**
- * false; set to true if the AsyncMessenger bound to a specific address;
- * and set false again by Accepter::stop().
- */
- bool did_bind;
- /// counter for the global seq our connection protocol uses
- __u32 global_seq;
- /// lock to protect the global_seq
- ceph_spinlock_t global_seq_lock;
-
- /**
- * hash map of addresses to Asyncconnection
- *
- * NOTE: a Asyncconnection* with state CLOSED may still be in the map but is considered
- * invalid and can be replaced by anyone holding the msgr lock
- */
- ceph::unordered_map<entity_addr_t, AsyncConnectionRef> conns;
-
- /**
- * list of connection are in teh process of accepting
- *
- * These are not yet in the conns map.
- */
- set<AsyncConnectionRef> accepting_conns;
-
- /**
- * list of connection are closed which need to be clean up
- *
- * Because AsyncMessenger and AsyncConnection follow a lock rule that
- * we can lock AsyncMesenger::lock firstly then lock AsyncConnection::lock
- * but can't reversed. This rule is aimed to avoid dead lock.
- * So if AsyncConnection want to unregister itself from AsyncMessenger,
- * we pick up this idea that just queue itself to this set and do lazy
- * deleted for AsyncConnection. "_lookup_conn" must ensure not return a
- * AsyncConnection in this set.
- */
- Mutex deleted_lock;
- set<AsyncConnectionRef> deleted_conns;
-
- EventCallbackRef reap_handler;
-
- /// internal cluster protocol version, if any, for talking to entities of the same type.
- int cluster_protocol;
-
- Cond stop_cond;
- bool stopped;
-
- AsyncConnectionRef _lookup_conn(const entity_addr_t& k) {
- assert(lock.is_locked());
- ceph::unordered_map<entity_addr_t, AsyncConnectionRef>::iterator p = conns.find(k);
- if (p == conns.end())
- return NULL;
-
- // lazy delete, see "deleted_conns"
- Mutex::Locker l(deleted_lock);
- if (deleted_conns.erase(p->second)) {
- p->second->get_perf_counter()->dec(l_msgr_active_connections);
- conns.erase(p);
- return NULL;
- }
-
- return p->second;
- }
-
- void _init_local_connection() {
- assert(lock.is_locked());
- local_connection->peer_addr = my_inst.addr;
- local_connection->peer_type = my_inst.name.type();
- local_connection->set_features(CEPH_FEATURES_ALL);
- ms_deliver_handle_fast_connect(local_connection.get());
- }
-
- void shutdown_connections(bool queue_reset);
-
-public:
-
- /// con used for sending messages to ourselves
- ConnectionRef local_connection;
-
- /**
- * @defgroup AsyncMessenger internals
- * @{
- */
- /**
- * This wraps _lookup_conn.
- */
- AsyncConnectionRef lookup_conn(const entity_addr_t& k) {
- Mutex::Locker l(lock);
- return _lookup_conn(k);
- }
-
- int accept_conn(AsyncConnectionRef conn) {
- Mutex::Locker l(lock);
- auto it = conns.find(conn->peer_addr);
- if (it != conns.end()) {
- AsyncConnectionRef existing = it->second;
-
- // lazy delete, see "deleted_conns"
- // If conn already in, we will return 0
- Mutex::Locker l(deleted_lock);
- if (deleted_conns.erase(existing)) {
- existing->get_perf_counter()->dec(l_msgr_active_connections);
- conns.erase(it);
- } else if (conn != existing) {
- return -1;
- }
- }
- conns[conn->peer_addr] = conn;
- conn->get_perf_counter()->inc(l_msgr_active_connections);
- accepting_conns.erase(conn);
- return 0;
- }
-
- void learned_addr(const entity_addr_t &peer_addr_for_me);
- void add_accept(Worker *w, ConnectedSocket cli_socket, entity_addr_t &addr);
- NetworkStack *get_stack() {
- return stack;
- }
-
- /**
- * This wraps ms_deliver_get_authorizer. We use it for AsyncConnection.
- */
- AuthAuthorizer *get_authorizer(int peer_type, bool force_new) {
- return ms_deliver_get_authorizer(peer_type, force_new);
- }
-
- /**
- * This wraps ms_deliver_verify_authorizer; we use it for AsyncConnection.
- */
- bool verify_authorizer(Connection *con, int peer_type, int protocol, bufferlist& auth, bufferlist& auth_reply,
- bool& isvalid, CryptoKey& session_key) {
- return ms_deliver_verify_authorizer(con, peer_type, protocol, auth,
- auth_reply, isvalid, session_key);
- }
- /**
- * Increment the global sequence for this AsyncMessenger and return it.
- * This is for the connect protocol, although it doesn't hurt if somebody
- * else calls it.
- *
- * @return a global sequence ID that nobody else has seen.
- */
- __u32 get_global_seq(__u32 old=0) {
- ceph_spin_lock(&global_seq_lock);
- if (old > global_seq)
- global_seq = old;
- __u32 ret = ++global_seq;
- ceph_spin_unlock(&global_seq_lock);
- return ret;
- }
- /**
- * Get the protocol version we support for the given peer type: either
- * a peer protocol (if it matches our own), the protocol version for the
- * peer (if we're connecting), or our protocol version (if we're accepting).
- */
- int get_proto_version(int peer_type, bool connect) const;
-
- /**
- * Fill in the address and peer type for the local connection, which
- * is used for delivering messages back to ourself.
- */
- void init_local_connection() {
- Mutex::Locker l(lock);
- _init_local_connection();
- }
-
- /**
- * Unregister connection from `conns`
- *
- * See "deleted_conns"
- */
- void unregister_conn(AsyncConnectionRef conn) {
- Mutex::Locker l(deleted_lock);
- deleted_conns.insert(conn);
-
- if (deleted_conns.size() >= ReapDeadConnectionThreshold) {
- local_worker->center.dispatch_event_external(reap_handler);
- }
- }
-
- /**
- * Reap dead connection from `deleted_conns`
- *
- * @return the number of dead connections
- *
- * See "deleted_conns"
- */
- int reap_dead();
-
- /**
- * @} // AsyncMessenger Internals
- */
-} ;
-
-#endif /* CEPH_ASYNCMESSENGER_H */
Code Review / stor4nfv.git / blobdiff