+++ /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_MSG_ASYNCCONNECTION_H
-#define CEPH_MSG_ASYNCCONNECTION_H
-
-#include <atomic>
-#include <pthread.h>
-#include <climits>
-#include <list>
-#include <mutex>
-#include <map>
-using namespace std;
-
-#include "auth/AuthSessionHandler.h"
-#include "common/ceph_time.h"
-#include "common/perf_counters.h"
-#include "include/buffer.h"
-#include "msg/Connection.h"
-#include "msg/Messenger.h"
-
-#include "Event.h"
-#include "Stack.h"
-
-class AsyncMessenger;
-class Worker;
-
-static const int ASYNC_IOV_MAX = (IOV_MAX >= 1024 ? IOV_MAX / 4 : IOV_MAX);
-
-/*
- * AsyncConnection maintains a logic session between two endpoints. In other
- * word, a pair of addresses can find the only AsyncConnection. AsyncConnection
- * will handle with network fault or read/write transactions. If one file
- * descriptor broken, AsyncConnection will maintain the message queue and
- * sequence, try to reconnect peer endpoint.
- */
-class AsyncConnection : public Connection {
-
- ssize_t read_bulk(char *buf, unsigned len);
- ssize_t do_sendmsg(struct msghdr &msg, unsigned len, bool more);
- ssize_t try_send(bufferlist &bl, bool more=false) {
- std::lock_guard<std::mutex> l(write_lock);
- outcoming_bl.claim_append(bl);
- return _try_send(more);
- }
- ssize_t _try_send(bool more=false);
- ssize_t _send(Message *m);
- void prepare_send_message(uint64_t features, Message *m, bufferlist &bl);
- ssize_t read_until(unsigned needed, char *p);
- ssize_t _process_connection();
- void _connect();
- void _stop();
- int handle_connect_reply(ceph_msg_connect &connect, ceph_msg_connect_reply &r);
- ssize_t handle_connect_msg(ceph_msg_connect &m, bufferlist &aubl, bufferlist &bl);
- void was_session_reset();
- void fault();
- void discard_out_queue();
- void discard_requeued_up_to(uint64_t seq);
- void requeue_sent();
- int randomize_out_seq();
- void handle_ack(uint64_t seq);
- void _append_keepalive_or_ack(bool ack=false, utime_t *t=NULL);
- ssize_t write_message(Message *m, bufferlist& bl, bool more);
- void inject_delay();
- ssize_t _reply_accept(char tag, ceph_msg_connect &connect, ceph_msg_connect_reply &reply,
- bufferlist &authorizer_reply) {
- bufferlist reply_bl;
- reply.tag = tag;
- reply.features = ((uint64_t)connect.features & policy.features_supported) | policy.features_required;
- reply.authorizer_len = authorizer_reply.length();
- reply_bl.append((char*)&reply, sizeof(reply));
- if (reply.authorizer_len) {
- reply_bl.append(authorizer_reply.c_str(), authorizer_reply.length());
- }
- ssize_t r = try_send(reply_bl);
- if (r < 0) {
- inject_delay();
- return -1;
- }
-
- state = STATE_ACCEPTING_WAIT_CONNECT_MSG;
- return 0;
- }
- bool is_queued() const {
- return !out_q.empty() || outcoming_bl.length();
- }
- void shutdown_socket() {
- for (auto &&t : register_time_events)
- center->delete_time_event(t);
- register_time_events.clear();
- if (last_tick_id) {
- center->delete_time_event(last_tick_id);
- last_tick_id = 0;
- }
- if (cs) {
- center->delete_file_event(cs.fd(), EVENT_READABLE|EVENT_WRITABLE);
- cs.shutdown();
- cs.close();
- }
- }
- Message *_get_next_outgoing(bufferlist *bl) {
- Message *m = 0;
- while (!m && !out_q.empty()) {
- map<int, list<pair<bufferlist, Message*> > >::reverse_iterator it = out_q.rbegin();
- if (!it->second.empty()) {
- list<pair<bufferlist, Message*> >::iterator p = it->second.begin();
- m = p->second;
- if (bl)
- bl->swap(p->first);
- it->second.erase(p);
- }
- if (it->second.empty())
- out_q.erase(it->first);
- }
- return m;
- }
- bool _has_next_outgoing() const {
- return !out_q.empty();
- }
- void reset_recv_state();
-
- /**
- * The DelayedDelivery is for injecting delays into Message delivery off
- * the socket. It is only enabled if delays are requested, and if they
- * are then it pulls Messages off the DelayQueue and puts them into the
- * AsyncMessenger event queue.
- */
- class DelayedDelivery : public EventCallback {
- std::set<uint64_t> register_time_events; // need to delete it if stop
- std::deque<std::pair<utime_t, Message*> > delay_queue;
- std::mutex delay_lock;
- AsyncMessenger *msgr;
- EventCenter *center;
- DispatchQueue *dispatch_queue;
- uint64_t conn_id;
- std::atomic_bool stop_dispatch;
-
- public:
- explicit DelayedDelivery(AsyncMessenger *omsgr, EventCenter *c,
- DispatchQueue *q, uint64_t cid)
- : msgr(omsgr), center(c), dispatch_queue(q), conn_id(cid),
- stop_dispatch(false) { }
- ~DelayedDelivery() override {
- assert(register_time_events.empty());
- assert(delay_queue.empty());
- }
- void set_center(EventCenter *c) { center = c; }
- void do_request(int id) override;
- void queue(double delay_period, utime_t release, Message *m) {
- std::lock_guard<std::mutex> l(delay_lock);
- delay_queue.push_back(std::make_pair(release, m));
- register_time_events.insert(center->create_time_event(delay_period*1000000, this));
- }
- void discard() {
- stop_dispatch = true;
- center->submit_to(center->get_id(), [this] () mutable {
- std::lock_guard<std::mutex> l(delay_lock);
- while (!delay_queue.empty()) {
- Message *m = delay_queue.front().second;
- dispatch_queue->dispatch_throttle_release(m->get_dispatch_throttle_size());
- m->put();
- delay_queue.pop_front();
- }
- for (auto i : register_time_events)
- center->delete_time_event(i);
- register_time_events.clear();
- stop_dispatch = false;
- }, true);
- }
- bool ready() const { return !stop_dispatch && delay_queue.empty() && register_time_events.empty(); }
- void flush();
- } *delay_state;
-
- public:
- AsyncConnection(CephContext *cct, AsyncMessenger *m, DispatchQueue *q, Worker *w);
- ~AsyncConnection() override;
- void maybe_start_delay_thread();
-
- ostream& _conn_prefix(std::ostream *_dout);
-
- bool is_connected() override {
- return can_write.load() == WriteStatus::CANWRITE;
- }
-
- // Only call when AsyncConnection first construct
- void connect(const entity_addr_t& addr, int type) {
- set_peer_type(type);
- set_peer_addr(addr);
- policy = msgr->get_policy(type);
- _connect();
- }
- // Only call when AsyncConnection first construct
- void accept(ConnectedSocket socket, entity_addr_t &addr);
- int send_message(Message *m) override;
-
- void send_keepalive() override;
- void mark_down() override;
- void mark_disposable() override {
- std::lock_guard<std::mutex> l(lock);
- policy.lossy = true;
- }
-
- private:
- enum {
- STATE_NONE,
- STATE_OPEN,
- STATE_OPEN_KEEPALIVE2,
- STATE_OPEN_KEEPALIVE2_ACK,
- STATE_OPEN_TAG_ACK,
- STATE_OPEN_MESSAGE_HEADER,
- STATE_OPEN_MESSAGE_THROTTLE_MESSAGE,
- STATE_OPEN_MESSAGE_THROTTLE_BYTES,
- STATE_OPEN_MESSAGE_THROTTLE_DISPATCH_QUEUE,
- STATE_OPEN_MESSAGE_READ_FRONT,
- STATE_OPEN_MESSAGE_READ_MIDDLE,
- STATE_OPEN_MESSAGE_READ_DATA_PREPARE,
- STATE_OPEN_MESSAGE_READ_DATA,
- STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH,
- STATE_OPEN_TAG_CLOSE,
- STATE_WAIT_SEND,
- STATE_CONNECTING,
- STATE_CONNECTING_RE,
- STATE_CONNECTING_WAIT_BANNER_AND_IDENTIFY,
- STATE_CONNECTING_SEND_CONNECT_MSG,
- STATE_CONNECTING_WAIT_CONNECT_REPLY,
- STATE_CONNECTING_WAIT_CONNECT_REPLY_AUTH,
- STATE_CONNECTING_WAIT_ACK_SEQ,
- STATE_CONNECTING_READY,
- STATE_ACCEPTING,
- STATE_ACCEPTING_WAIT_BANNER_ADDR,
- STATE_ACCEPTING_WAIT_CONNECT_MSG,
- STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH,
- STATE_ACCEPTING_WAIT_SEQ,
- STATE_ACCEPTING_READY,
- STATE_STANDBY,
- STATE_CLOSED,
- STATE_WAIT, // just wait for racing connection
- };
-
- static const int TCP_PREFETCH_MIN_SIZE;
- static const char *get_state_name(int state) {
- const char* const statenames[] = {"STATE_NONE",
- "STATE_OPEN",
- "STATE_OPEN_KEEPALIVE2",
- "STATE_OPEN_KEEPALIVE2_ACK",
- "STATE_OPEN_TAG_ACK",
- "STATE_OPEN_MESSAGE_HEADER",
- "STATE_OPEN_MESSAGE_THROTTLE_MESSAGE",
- "STATE_OPEN_MESSAGE_THROTTLE_BYTES",
- "STATE_OPEN_MESSAGE_THROTTLE_DISPATCH_QUEUE",
- "STATE_OPEN_MESSAGE_READ_FRONT",
- "STATE_OPEN_MESSAGE_READ_MIDDLE",
- "STATE_OPEN_MESSAGE_READ_DATA_PREPARE",
- "STATE_OPEN_MESSAGE_READ_DATA",
- "STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH",
- "STATE_OPEN_TAG_CLOSE",
- "STATE_WAIT_SEND",
- "STATE_CONNECTING",
- "STATE_CONNECTING_RE",
- "STATE_CONNECTING_WAIT_BANNER_AND_IDENTIFY",
- "STATE_CONNECTING_SEND_CONNECT_MSG",
- "STATE_CONNECTING_WAIT_CONNECT_REPLY",
- "STATE_CONNECTING_WAIT_CONNECT_REPLY_AUTH",
- "STATE_CONNECTING_WAIT_ACK_SEQ",
- "STATE_CONNECTING_READY",
- "STATE_ACCEPTING",
- "STATE_ACCEPTING_WAIT_BANNER_ADDR",
- "STATE_ACCEPTING_WAIT_CONNECT_MSG",
- "STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH",
- "STATE_ACCEPTING_WAIT_SEQ",
- "STATE_ACCEPTING_READY",
- "STATE_STANDBY",
- "STATE_CLOSED",
- "STATE_WAIT"};
- return statenames[state];
- }
-
- AsyncMessenger *async_msgr;
- uint64_t conn_id;
- PerfCounters *logger;
- int global_seq;
- __u32 connect_seq, peer_global_seq;
- std::atomic<uint64_t> out_seq{0};
- std::atomic<uint64_t> ack_left{0}, in_seq{0};
- int state;
- int state_after_send;
- ConnectedSocket cs;
- int port;
- Messenger::Policy policy;
-
- DispatchQueue *dispatch_queue;
-
- // lockfree, only used in own thread
- bufferlist outcoming_bl;
- bool open_write = false;
-
- std::mutex write_lock;
- enum class WriteStatus {
- NOWRITE,
- REPLACING,
- CANWRITE,
- CLOSED
- };
- std::atomic<WriteStatus> can_write;
- list<Message*> sent; // the first bufferlist need to inject seq
- map<int, list<pair<bufferlist, Message*> > > out_q; // priority queue for outbound msgs
- bool keepalive;
-
- std::mutex lock;
- utime_t backoff; // backoff time
- EventCallbackRef read_handler;
- EventCallbackRef write_handler;
- EventCallbackRef wakeup_handler;
- EventCallbackRef tick_handler;
- struct iovec msgvec[ASYNC_IOV_MAX];
- char *recv_buf;
- uint32_t recv_max_prefetch;
- uint32_t recv_start;
- uint32_t recv_end;
- set<uint64_t> register_time_events; // need to delete it if stop
- ceph::coarse_mono_clock::time_point last_active;
- uint64_t last_tick_id = 0;
- const uint64_t inactive_timeout_us;
-
- // Tis section are temp variables used by state transition
-
- // Open state
- utime_t recv_stamp;
- utime_t throttle_stamp;
- unsigned msg_left;
- uint64_t cur_msg_size;
- ceph_msg_header current_header;
- bufferlist data_buf;
- bufferlist::iterator data_blp;
- bufferlist front, middle, data;
- ceph_msg_connect connect_msg;
- // Connecting state
- bool got_bad_auth;
- AuthAuthorizer *authorizer;
- bufferlist authorizer_buf;
- ceph_msg_connect_reply connect_reply;
- // Accepting state
- entity_addr_t socket_addr;
- CryptoKey session_key;
- bool replacing; // when replacing process happened, we will reply connect
- // side with RETRY tag and accept side will clear replaced
- // connection. So when connect side reissue connect_msg,
- // there won't exists conflicting connection so we use
- // "replacing" to skip RESETSESSION to avoid detect wrong
- // presentation
- bool is_reset_from_peer;
- bool once_ready;
-
- // used only for local state, it will be overwrite when state transition
- char *state_buffer;
- // used only by "read_until"
- uint64_t state_offset;
- Worker *worker;
- EventCenter *center;
- ceph::shared_ptr<AuthSessionHandler> session_security;
-
- public:
- // used by eventcallback
- void handle_write();
- void process();
- void wakeup_from(uint64_t id);
- void tick(uint64_t id);
- void local_deliver();
- void stop(bool queue_reset) {
- lock.lock();
- bool need_queue_reset = (state != STATE_CLOSED) && queue_reset;
- _stop();
- lock.unlock();
- if (need_queue_reset)
- dispatch_queue->queue_reset(this);
- }
- void cleanup() {
- shutdown_socket();
- delete read_handler;
- delete write_handler;
- delete wakeup_handler;
- delete tick_handler;
- if (delay_state) {
- delete delay_state;
- delay_state = NULL;
- }
- }
- PerfCounters *get_perf_counter() {
- return logger;
- }
-}; /* AsyncConnection */
-
-typedef boost::intrusive_ptr<AsyncConnection> AsyncConnectionRef;
-
-#endif
Code Review / stor4nfv.git / blobdiff