+++ /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>
- * 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_CONNECTION_H
-#define XIO_CONNECTION_H
-
-#include <atomic>
-
-#include <boost/intrusive/avl_set.hpp>
-#include <boost/intrusive/list.hpp>
-
-extern "C" {
-#include "libxio.h"
-}
-
-#include "XioInSeq.h"
-#include "XioSubmit.h"
-#include "msg/Connection.h"
-#include "msg/Messenger.h"
-#include "auth/AuthSessionHandler.h"
-
-#define XIO_ALL_FEATURES (CEPH_FEATURES_ALL)
-
-
-#define XIO_NOP_TAG_MARKDOWN 0x0001
-
-namespace bi = boost::intrusive;
-
-class XioPortal;
-class XioMessenger;
-class XioSend;
-
-class XioConnection : public Connection
-{
-public:
- enum type { ACTIVE, PASSIVE };
-
- enum class session_states : unsigned {
- INIT = 0,
- START,
- UP,
- FLOW_CONTROLLED,
- DISCONNECTED,
- DELETED,
- BARRIER
- };
-
- enum class session_startup_states : unsigned {
- IDLE = 0,
- CONNECTING,
- ACCEPTING,
- READY,
- FAIL
- };
-
-private:
- XioConnection::type xio_conn_type;
- XioPortal *portal;
- std::atomic<bool> connected = { false };
- entity_inst_t peer;
- struct xio_session *session;
- struct xio_connection *conn;
- pthread_spinlock_t sp;
- std::atomic<int64_t> send = { 0 };
- std::atomic<int64_t> recv = { 0 };
- uint32_t n_reqs; // Accelio-initiated reqs in progress (!counting partials)
- uint32_t magic;
- uint32_t special_handling;
- uint64_t scount;
- uint32_t send_ctr;
- int q_high_mark;
- int q_low_mark;
-
- struct lifecycle {
- // different from Pipe states?
- enum lf_state {
- INIT,
- LOCAL_DISCON,
- REMOTE_DISCON,
- RECONNECTING,
- UP,
- DEAD } state;
-
- /* XXX */
- uint32_t reconnects;
- uint32_t connect_seq, peer_global_seq;
- uint64_t in_seq, out_seq_acked; // atomic<uint64_t>, got receipt
- std::atomic<int64_t> out_seq = { 0 };
-
- lifecycle() : state(lifecycle::INIT), reconnects(0), connect_seq(0),
- peer_global_seq(0), in_seq(0), out_seq_acked(0)
- {}
-
- void set_in_seq(uint64_t seq) {
- in_seq = seq;
- }
-
- uint64_t next_out_seq() {
- return ++out_seq;
- }
-
- } state;
-
- /* batching */
- XioInSeq in_seq;
-
- class CState
- {
- public:
- static const int FLAG_NONE = 0x0000;
- static const int FLAG_BAD_AUTH = 0x0001;
- static const int FLAG_MAPPED = 0x0002;
- static const int FLAG_RESET = 0x0004;
-
- static const int OP_FLAG_NONE = 0x0000;
- static const int OP_FLAG_LOCKED = 0x0001;
- static const int OP_FLAG_LRU = 0x0002;
-
- uint64_t features;
- Messenger::Policy policy;
-
- CryptoKey session_key;
- ceph::shared_ptr<AuthSessionHandler> session_security;
- AuthAuthorizer *authorizer;
- XioConnection *xcon;
- uint32_t protocol_version;
-
- std::atomic<session_states> session_state = { 0 };
- std::atomic<session_startup_state> startup_state = { 0 };
-
- uint32_t reconnects;
- uint32_t connect_seq, global_seq, peer_global_seq;
- uint64_t in_seq, out_seq_acked; // atomic<uint64_t>, got receipt
- std::atomic<uint64_t> out_seq = { 0 };
-
- uint32_t flags;
-
- explicit CState(XioConnection* _xcon)
- : features(0),
- authorizer(NULL),
- xcon(_xcon),
- protocol_version(0),
- session_state(INIT),
- startup_state(IDLE),
- reconnects(0),
- connect_seq(0),
- global_seq(0),
- peer_global_seq(0),
- in_seq(0),
- out_seq_acked(0),
- flags(FLAG_NONE) {}
-
- uint64_t get_session_state() {
- return session_state;
- }
-
- uint64_t get_startup_state() {
- return startup_state;
- }
-
- void set_in_seq(uint64_t seq) {
- in_seq = seq;
- }
-
- uint64_t next_out_seq() {
- return ++out_seq;
- };
-
- // state machine
- int init_state();
- int next_state(Message* m);
-#if 0 // future (session startup)
- int msg_connect(MConnect *m);
- int msg_connect_reply(MConnectReply *m);
- int msg_connect_reply(MConnectAuthReply *m);
- int msg_connect_auth(MConnectAuth *m);
- int msg_connect_auth_reply(MConnectAuthReply *m);
-#endif
- int state_up_ready(uint32_t flags);
- int state_flow_controlled(uint32_t flags);
- int state_discon();
- int state_fail(Message* m, uint32_t flags);
-
- } cstate; /* CState */
-
- // message submission queue
- struct SendQ {
- bool keepalive;
- bool ack;
- utime_t ack_time;
- Message::Queue mqueue; // deferred
- XioSubmit::Queue requeue;
-
- SendQ():keepalive(false), ack(false){}
- } outgoing;
-
- // conns_entity_map comparison functor
- struct EntityComp
- {
- // for internal ordering
- bool operator()(const XioConnection &lhs, const XioConnection &rhs) const
- { return lhs.get_peer() < rhs.get_peer(); }
-
- // for external search by entity_inst_t(peer)
- bool operator()(const entity_inst_t &peer, const XioConnection &c) const
- { return peer < c.get_peer(); }
-
- bool operator()(const XioConnection &c, const entity_inst_t &peer) const
- { return c.get_peer() < peer; }
- };
-
- bi::list_member_hook<> conns_hook;
- bi::avl_set_member_hook<> conns_entity_map_hook;
-
- typedef bi::list< XioConnection,
- bi::member_hook<XioConnection, bi::list_member_hook<>,
- &XioConnection::conns_hook > > ConnList;
-
- typedef bi::member_hook<XioConnection, bi::avl_set_member_hook<>,
- &XioConnection::conns_entity_map_hook> EntityHook;
-
- typedef bi::avl_set< XioConnection, EntityHook,
- bi::compare<EntityComp> > EntitySet;
-
- friend class XioPortal;
- friend class XioMessenger;
- friend class XioDispatchHook;
- friend class XioMarkDownHook;
- friend class XioSend;
-
- int on_disconnect_event() {
- connected = false;
- pthread_spin_lock(&sp);
- discard_out_queues(CState::OP_FLAG_LOCKED);
- pthread_spin_unlock(&sp);
- return 0;
- }
-
- int on_teardown_event() {
- pthread_spin_lock(&sp);
- if (conn)
- xio_connection_destroy(conn);
- conn = NULL;
- pthread_spin_unlock(&sp);
- this->put();
- return 0;
- }
-
- int xio_qdepth_high_mark() {
- return q_high_mark;
- }
-
- int xio_qdepth_low_mark() {
- return q_low_mark;
- }
-
-public:
- XioConnection(XioMessenger *m, XioConnection::type _type,
- const entity_inst_t& peer);
-
- ~XioConnection() {
- if (conn)
- xio_connection_destroy(conn);
- }
- ostream& conn_prefix(std::ostream *_dout);
-
- bool is_connected() override { return connected; }
-
- int send_message(Message *m) override;
- void send_keepalive() override {send_keepalive_or_ack();}
- void send_keepalive_or_ack(bool ack = false, const utime_t *tp = nullptr);
- void mark_down() override;
- int _mark_down(uint32_t flags);
- void mark_disposable() override;
- int _mark_disposable(uint32_t flags);
-
- const entity_inst_t& get_peer() const { return peer; }
-
- XioConnection* get() {
-#if 0
- cout << "XioConnection::get " << this << " " << nref.load() << std::endl;
-#endif
- RefCountedObject::get();
- return this;
- }
-
- void put() {
- RefCountedObject::put();
-#if 0
- cout << "XioConnection::put " << this << " " << nref.load() << std::endl;
-#endif
- }
-
- void disconnect() {
- if (is_connected()) {
- connected = false;
- xio_disconnect(conn); // normal teardown will clean up conn
- }
- }
-
- uint32_t get_magic() { return magic; }
- void set_magic(int _magic) { magic = _magic; }
- uint32_t get_special_handling() { return special_handling; }
- void set_special_handling(int n) { special_handling = n; }
- uint64_t get_scount() { return scount; }
-
- int passive_setup(); /* XXX */
-
- int handle_data_msg(struct xio_session *session, struct xio_msg *msg,
- int more_in_batch, void *cb_user_context);
- int on_msg(struct xio_session *session, struct xio_msg *msg,
- int more_in_batch, void *cb_user_context);
- int on_ow_msg_send_complete(struct xio_session *session, struct xio_msg *msg,
- void *conn_user_context);
- int on_msg_error(struct xio_session *session, enum xio_status error,
- struct xio_msg *msg, void *conn_user_context);
- void msg_send_fail(XioSend *xsend, int code);
- void msg_release_fail(struct xio_msg *msg, int code);
-private:
- void send_keepalive_or_ack_internal(bool ack = false, const utime_t *tp = nullptr);
- int flush_out_queues(uint32_t flags);
- int discard_out_queues(uint32_t flags);
- int adjust_clru(uint32_t flags);
-};
-
-typedef boost::intrusive_ptr<XioConnection> XioConnectionRef;
-
-class XioLoopbackConnection : public Connection
-{
-private:
- std::atomic<uint64_t> seq = { 0 };
-public:
- explicit XioLoopbackConnection(Messenger *m) : Connection(m->cct, m)
- {
- const entity_inst_t& m_inst = m->get_myinst();
- peer_addr = m_inst.addr;
- peer_type = m_inst.name.type();
- set_features(XIO_ALL_FEATURES); /* XXXX set to ours */
- }
-
- XioLoopbackConnection* get() {
- return static_cast<XioLoopbackConnection*>(RefCountedObject::get());
- }
-
- bool is_connected() override { return true; }
-
- int send_message(Message *m) override;
- void send_keepalive() override;
- void mark_down() override {}
- void mark_disposable() override {}
-
- uint64_t get_seq() {
- return seq;
- }
-
- uint64_t next_seq() {
- return ++seq;
- }
-};
-
-typedef boost::intrusive_ptr<XioLoopbackConnection> XioLoopbackConnectionRef;
-
-#endif /* XIO_CONNECTION_H */
Code Review / stor4nfv.git / blobdiff