AsyncMessenger::AsyncMessenger(CephContext *cct, entity_name_t name,
		string mname, uint64_t _nonce)
	: SimplePolicyMessenger(cct, name,mname, _nonce),
	......
{
	ceph_spin_init(&global_seq_lock);
	StackSingleton *single;
	cct->lookup_or_create_singleton_object<StackSingleton>(single, "AsyncMessenger::NetworkStack"); // 一个进程对应一个唯一的stack，以前这里是WorkerPool
	stack = single->stack.get();
	stack->start(); // 启动线程
	......
}

struct StackSingleton {
	std::shared_ptr<NetworkStack> stack;
	StackSingleton(CephContext *c) {
		stack = NetworkStack::create(c, c->_conf->ms_async_transport_type); // 根据类型创建不同的stack
	}
	~StackSingleton() {
		stack->stop();
	}
};

NetworkStack::NetworkStack(CephContext *c, const string &t): type(t), started(false), cct(c)
{
	......
	for (unsigned i = 0; i < num_workers; ++i) {
		Worker *w = create_worker(cct, type, i); // 创建worker
		w->center.init(InitEventNumber, i); // 初始化事件处理器
		workers.push_back(w);
	}
	......
}

void NetworkStack::start()
{
	......
	for (unsigned i = 0; i < num_workers; ++i) {
		if (workers[i]->is_init())
			continue;
		std::function<void ()> thread = add_thread(i); // 获取线程入口的匿名函数
		spawn_worker(i, std::move(thread)); // 启动线程
	}
	started = true;
	......
}

// PosixNetworkStack创建线程的实现
virtual void spawn_worker(unsigned i, std::function<void ()> &&func) override {
	threads.resize(i+1);
	threads[i] = std::thread(func); // 使用c++11的线程库创建线程
}

// 线程入口函数
std::function<void ()> NetworkStack::add_thread(unsigned i)
{
	Worker *w = workers[i];
	return [this, w]() {
		const uint64_t EventMaxWaitUs = 30000000;
		w->center.set_owner();
		w->initialize();
		w->init_done();
		while (!w->done) {
			int r = w->center.process_events(EventMaxWaitUs); // 处理事件
		}
	}
	w->reset();
	w->destroy();
};

class Worker {
	unsigned id; // worker的ID
	EventCenter center; // 处理事件

	......
	virtual int listen(entity_addr_t &addr,
			const SocketOptions &opts, ServerSocket *) = 0;
	virtual int connect(const entity_addr_t &addr,
			const SocketOptions &opts, ConnectedSocket *socket) = 0;
	......
};

class ConnectedSocket {
	std::unique_ptr<ConnectedSocketImpl> _csi; // 依赖于具体协议栈的实现
	ssize_t read(char* buf, size_t len) { // 读数据
		return _csi->read(buf, len);
	}
	ssize_t send(bufferlist &bl, bool more) { // 写数据
		return _csi->send(bl, more);
	}
};

class ServerSocket {
	std::unique_ptr<ServerSocketImpl> _ssi; // 依赖于具体协议栈的实现

	int accept(ConnectedSocket *sock, const SocketOptions &opt, entity_addr_t *out, Worker *w) { // 接受连接请求，成功后第一个参数为已经成功连接的套接字
		return _ssi->accept(sock, opt, out, w);
	}
};

int PosixWorker::listen(entity_addr_t &sa, const SocketOptions &opt,
		ServerSocket *sock)
{
	int listen_sd = net.create_socket(sa.get_family(), true); // 创建套接字
	r = ::bind(listen_sd, sa.get_sockaddr(), sa.get_sockaddr_len()); // 绑定地址
	r = ::listen(listen_sd, 128); // 开始监听
	*sock = ServerSocket(
			std::unique_ptr<PosixServerSocketImpl>(
				new PosixServerSocketImpl(net, listen_sd))); // 创建serversocket，返回给用户，可以用此socket进行accept请求
	return 0;
}

int PosixWorker::connect(const entity_addr_t &addr, const SocketOptions &opts, ConnectedSocket *socket)
{
	int sd;
	sd = net.connect(addr); // 发起连接
	// 创建connectedsocket，返回给用户，可以用此socket进行send/read
	*socket = ConnectedSocket(std::unique_ptr<PosixConnectedSocketImpl>(new PosixConnectedSocketImpl(net, addr, sd, !opts.nonblock)));
	return 0;
}

int PosixServerSocketImpl::accept(ConnectedSocket *sock, const SocketOptions &opt, entity_addr_t *out, Worker *w)
{
	int sd = ::accept(_fd, (sockaddr*)&ss, &slen); // 接受连接请求

	std::unique_ptr<PosixConnectedSocketImpl> csi(new PosixConnectedSocketImpl(handler, *out, sd, true)); // 创建连接成功的socket
	*sock = ConnectedSocket(std::move(csi)); // 返回给用户使用
	return 0;
}

int Processor::bind(const entity_addr_t &bind_addr,
		const set<int>& avoid_ports,
		entity_addr_t* bound_addr)
{
	......
	// 向worker的事件中心提交一个外部事件，当这个事件执行的时候，就是执行第二个参数的匿名函数对象，也就是调用worker的listen
	worker->center.submit_to(worker->center.get_id(), [this, &listen_addr, &opts, &r]() {
			r = worker->listen(listen_addr, opts, &listen_socket); // 事件被执行后，listen_socket就被更新，可以用它来进行accept
	}, false);
	......
}

void Processor::start()
{
	// 向worker的事件中心提交一个外部事件，当这个事件执行的时候，就是执行第二个参数的匿名函数对象，也就是将fd加入事件中心进行管理
	if (listen_socket) {
		worker->center.submit_to(worker->center.get_id(), [this]() {
			worker->center.create_file_event(listen_socket.fd(), EVENT_READABLE, listen_handler); }, false);
	}
}

// 上面加入的可读事件，回调是listen_handler，即执行如下函数
void Processor::accept()
{
	.......
	while (true) {
		int r = listen_socket.accept(&cli_socket, opts, &addr, w); // 执行监听socket的accept，成功后第一个参数的值为connectedsocket类型，可以用来收发数据
	}
	......
}

ssize_t AsyncConnection::_process_connection()
{
	......
	case STATE_CONNECTING:
		{
			......
			r = worker->connect(get_peer_addr(), opts, &cs); // 调用worker的connect
			center->create_file_event(cs.fd(), EVENT_READABLE, read_handler); // 将连接成功的connectedsocket加入事件中心进行管理
			state = STATE_CONNECTING_RE;
			break;
		}
}