--- /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) 2016 XSKY <haomai@xsky.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.
+ *
+ */
+
+#include <sys/socket.h>
+#include <netinet/tcp.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <errno.h>
+
+#include <algorithm>
+
+#include "PosixStack.h"
+
+#include "include/buffer.h"
+#include "include/str_list.h"
+#include "common/errno.h"
+#include "common/strtol.h"
+#include "common/dout.h"
+#include "common/simple_spin.h"
+#include "msg/Messenger.h"
+#include "include/sock_compat.h"
+
+#define dout_subsys ceph_subsys_ms
+#undef dout_prefix
+#define dout_prefix *_dout << "PosixStack "
+
+class PosixConnectedSocketImpl final : public ConnectedSocketImpl {
+ NetHandler &handler;
+ int _fd;
+ entity_addr_t sa;
+ bool connected;
+
+ public:
+ explicit PosixConnectedSocketImpl(NetHandler &h, const entity_addr_t &sa, int f, bool connected)
+ : handler(h), _fd(f), sa(sa), connected(connected) {}
+
+ int is_connected() override {
+ if (connected)
+ return 1;
+
+ int r = handler.reconnect(sa, _fd);
+ if (r == 0) {
+ connected = true;
+ return 1;
+ } else if (r < 0) {
+ return r;
+ } else {
+ return 0;
+ }
+ }
+
+ ssize_t zero_copy_read(bufferptr&) override {
+ return -EOPNOTSUPP;
+ }
+
+ ssize_t read(char *buf, size_t len) override {
+ ssize_t r = ::read(_fd, buf, len);
+ if (r < 0)
+ r = -errno;
+ return r;
+ }
+
+ // return the sent length
+ // < 0 means error occured
+ static ssize_t do_sendmsg(int fd, struct msghdr &msg, unsigned len, bool more)
+ {
+ size_t sent = 0;
+ while (1) {
+ MSGR_SIGPIPE_STOPPER;
+ ssize_t r;
+ r = ::sendmsg(fd, &msg, MSG_NOSIGNAL | (more ? MSG_MORE : 0));
+ if (r < 0) {
+ if (errno == EINTR) {
+ continue;
+ } else if (errno == EAGAIN) {
+ break;
+ }
+ return -errno;
+ }
+
+ sent += r;
+ if (len == sent) break;
+
+ while (r > 0) {
+ if (msg.msg_iov[0].iov_len <= (size_t)r) {
+ // drain this whole item
+ r -= msg.msg_iov[0].iov_len;
+ msg.msg_iov++;
+ msg.msg_iovlen--;
+ } else {
+ msg.msg_iov[0].iov_base = (char *)msg.msg_iov[0].iov_base + r;
+ msg.msg_iov[0].iov_len -= r;
+ break;
+ }
+ }
+ }
+ return (ssize_t)sent;
+ }
+
+ ssize_t send(bufferlist &bl, bool more) override {
+ size_t sent_bytes = 0;
+ std::list<bufferptr>::const_iterator pb = bl.buffers().begin();
+ uint64_t left_pbrs = bl.buffers().size();
+ while (left_pbrs) {
+ struct msghdr msg;
+ struct iovec msgvec[IOV_MAX];
+ uint64_t size = MIN(left_pbrs, IOV_MAX);
+ left_pbrs -= size;
+ memset(&msg, 0, sizeof(msg));
+ msg.msg_iovlen = 0;
+ msg.msg_iov = msgvec;
+ unsigned msglen = 0;
+ while (size > 0) {
+ msgvec[msg.msg_iovlen].iov_base = (void*)(pb->c_str());
+ msgvec[msg.msg_iovlen].iov_len = pb->length();
+ msg.msg_iovlen++;
+ msglen += pb->length();
+ ++pb;
+ size--;
+ }
+
+ ssize_t r = do_sendmsg(_fd, msg, msglen, left_pbrs || more);
+ if (r < 0)
+ return r;
+
+ // "r" is the remaining length
+ sent_bytes += r;
+ if (static_cast<unsigned>(r) < msglen)
+ break;
+ // only "r" == 0 continue
+ }
+
+ if (sent_bytes) {
+ bufferlist swapped;
+ if (sent_bytes < bl.length()) {
+ bl.splice(sent_bytes, bl.length()-sent_bytes, &swapped);
+ bl.swap(swapped);
+ } else {
+ bl.clear();
+ }
+ }
+
+ return static_cast<ssize_t>(sent_bytes);
+ }
+ void shutdown() override {
+ ::shutdown(_fd, SHUT_RDWR);
+ }
+ void close() override {
+ ::close(_fd);
+ }
+ int fd() const override {
+ return _fd;
+ }
+ friend class PosixServerSocketImpl;
+ friend class PosixNetworkStack;
+};
+
+class PosixServerSocketImpl : public ServerSocketImpl {
+ NetHandler &handler;
+ int _fd;
+
+ public:
+ explicit PosixServerSocketImpl(NetHandler &h, int f): handler(h), _fd(f) {}
+ int accept(ConnectedSocket *sock, const SocketOptions &opts, entity_addr_t *out, Worker *w) override;
+ void abort_accept() override {
+ ::close(_fd);
+ }
+ int fd() const override {
+ return _fd;
+ }
+};
+
+int PosixServerSocketImpl::accept(ConnectedSocket *sock, const SocketOptions &opt, entity_addr_t *out, Worker *w) {
+ assert(sock);
+ sockaddr_storage ss;
+ socklen_t slen = sizeof(ss);
+ int sd = ::accept(_fd, (sockaddr*)&ss, &slen);
+ if (sd < 0) {
+ return -errno;
+ }
+
+ handler.set_close_on_exec(sd);
+ int r = handler.set_nonblock(sd);
+ if (r < 0) {
+ ::close(sd);
+ return -errno;
+ }
+
+ r = handler.set_socket_options(sd, opt.nodelay, opt.rcbuf_size);
+ if (r < 0) {
+ ::close(sd);
+ return -errno;
+ }
+
+ assert(NULL != out); //out should not be NULL in accept connection
+
+ out->set_sockaddr((sockaddr*)&ss);
+ handler.set_priority(sd, opt.priority, out->get_family());
+
+ std::unique_ptr<PosixConnectedSocketImpl> csi(new PosixConnectedSocketImpl(handler, *out, sd, true));
+ *sock = ConnectedSocket(std::move(csi));
+ return 0;
+}
+
+void PosixWorker::initialize()
+{
+}
+
+int PosixWorker::listen(entity_addr_t &sa, const SocketOptions &opt,
+ ServerSocket *sock)
+{
+ int listen_sd = net.create_socket(sa.get_family(), true);
+ if (listen_sd < 0) {
+ return -errno;
+ }
+
+ int r = net.set_nonblock(listen_sd);
+ if (r < 0) {
+ ::close(listen_sd);
+ return -errno;
+ }
+
+ net.set_close_on_exec(listen_sd);
+ r = net.set_socket_options(listen_sd, opt.nodelay, opt.rcbuf_size);
+ if (r < 0) {
+ ::close(listen_sd);
+ return -errno;
+ }
+
+ r = ::bind(listen_sd, sa.get_sockaddr(), sa.get_sockaddr_len());
+ if (r < 0) {
+ r = -errno;
+ ldout(cct, 10) << __func__ << " unable to bind to " << sa.get_sockaddr()
+ << ": " << cpp_strerror(r) << dendl;
+ ::close(listen_sd);
+ return r;
+ }
+
+ r = ::listen(listen_sd, cct->_conf->ms_tcp_listen_backlog);
+ if (r < 0) {
+ r = -errno;
+ lderr(cct) << __func__ << " unable to listen on " << sa << ": " << cpp_strerror(r) << dendl;
+ ::close(listen_sd);
+ return r;
+ }
+
+ *sock = ServerSocket(
+ std::unique_ptr<PosixServerSocketImpl>(
+ new PosixServerSocketImpl(net, listen_sd)));
+ return 0;
+}
+
+int PosixWorker::connect(const entity_addr_t &addr, const SocketOptions &opts, ConnectedSocket *socket) {
+ int sd;
+
+ if (opts.nonblock) {
+ sd = net.nonblock_connect(addr, opts.connect_bind_addr);
+ } else {
+ sd = net.connect(addr, opts.connect_bind_addr);
+ }
+
+ if (sd < 0) {
+ return -errno;
+ }
+
+ net.set_priority(sd, opts.priority, addr.get_family());
+ *socket = ConnectedSocket(
+ std::unique_ptr<PosixConnectedSocketImpl>(new PosixConnectedSocketImpl(net, addr, sd, !opts.nonblock)));
+ return 0;
+}
+
+PosixNetworkStack::PosixNetworkStack(CephContext *c, const string &t)
+ : NetworkStack(c, t)
+{
+ vector<string> corestrs;
+ get_str_vec(cct->_conf->ms_async_affinity_cores, corestrs);
+ for (auto & corestr : corestrs) {
+ string err;
+ int coreid = strict_strtol(corestr.c_str(), 10, &err);
+ if (err == "")
+ coreids.push_back(coreid);
+ else
+ lderr(cct) << __func__ << " failed to parse " << corestr << " in " << cct->_conf->ms_async_affinity_cores << dendl;
+ }
+}
Code Review / stor4nfv.git / blobdiff