ceph/src/msg/async/net_handler.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2014 UnitedStack <haomai@unitedstack.com>
   7  *
   8  * Author: Haomai Wang <haomaiwang@gmail.com>
   9  *
  10  * This is free software; you can redistribute it and/or
  11  * modify it under the terms of the GNU Lesser General Public
  12  * License version 2.1, as published by the Free Software
  13  * Foundation.  See file COPYING.
  14  *
  15  */
  16
  17 #include <sys/types.h>
  18 #include <sys/socket.h>
  19 #include <netinet/in.h>
  20 #include <netinet/ip.h>
  21 #include <netinet/tcp.h>
  22 #include <arpa/inet.h>
  23
  24 #include "net_handler.h"
  25 #include "common/errno.h"
  26 #include "common/debug.h"
  27
  28 #define dout_subsys ceph_subsys_ms
  29 #undef dout_prefix
  30 #define dout_prefix *_dout << "NetHandler "
  31
  32 namespace ceph{
  33
  34 int NetHandler::create_socket(int domain, bool reuse_addr)
  35 {
  36   int s;
  37   int r = 0;
  38
  39   if ((s = ::socket(domain, SOCK_STREAM, 0)) == -1) {
  40     r = errno;
  41     lderr(cct) << __func__ << " couldn't create socket " << cpp_strerror(r) << dendl;
  42     return -r;
  43   }
  44
  45 #if !defined(__FreeBSD__)
  46   /* Make sure connection-intensive things like the benchmark
  47    * will be able to close/open sockets a zillion of times */
  48   if (reuse_addr) {
  49     int on = 1;
  50     if (::setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) == -1) {
  51       r = errno;
  52       lderr(cct) << __func__ << " setsockopt SO_REUSEADDR failed: "
  53                  << strerror(r) << dendl;
  54       close(s);
  55       return -r;
  56     }
  57   }
  58 #endif
  59
  60   return s;
  61 }
  62
  63 int NetHandler::set_nonblock(int sd)
  64 {
  65   int flags;
  66   int r = 0;
  67
  68   /* Set the socket nonblocking.
  69    * Note that fcntl(2) for F_GETFL and F_SETFL can't be
  70    * interrupted by a signal. */
  71   if ((flags = fcntl(sd, F_GETFL)) < 0 ) {
  72     r = errno;
  73     lderr(cct) << __func__ << " fcntl(F_GETFL) failed: " << cpp_strerror(r) << dendl;
  74     return -r;
  75   }
  76   if (fcntl(sd, F_SETFL, flags | O_NONBLOCK) < 0) {
  77     r = errno;
  78     lderr(cct) << __func__ << " fcntl(F_SETFL,O_NONBLOCK): " << cpp_strerror(r) << dendl;
  79     return -r;
  80   }
  81
  82   return 0;
  83 }
  84
  85 void NetHandler::set_close_on_exec(int sd)
  86 {
  87   int flags = fcntl(sd, F_GETFD, 0);
  88   if (flags < 0) {
  89     int r = errno;
  90     lderr(cct) << __func__ << " fcntl(F_GETFD): "
  91                << cpp_strerror(r) << dendl;
  92     return;
  93   }
  94   if (fcntl(sd, F_SETFD, flags | FD_CLOEXEC)) {
  95     int r = errno;
  96     lderr(cct) << __func__ << " fcntl(F_SETFD): "
  97                << cpp_strerror(r) << dendl;
  98   }
  99 }
 100
 101 int NetHandler::set_socket_options(int sd, bool nodelay, int size)
 102 {
 103   int r = 0;
 104   // disable Nagle algorithm?
 105   if (nodelay) {
 106     int flag = 1;
 107     r = ::setsockopt(sd, IPPROTO_TCP, TCP_NODELAY, (char*)&flag, sizeof(flag));
 108     if (r < 0) {
 109       r = errno;
 110       ldout(cct, 0) << "couldn't set TCP_NODELAY: " << cpp_strerror(r) << dendl;
 111     }
 112   }
 113   if (size) {
 114     r = ::setsockopt(sd, SOL_SOCKET, SO_RCVBUF, (void*)&size, sizeof(size));
 115     if (r < 0)  {
 116       r = errno;
 117       ldout(cct, 0) << "couldn't set SO_RCVBUF to " << size << ": " << cpp_strerror(r) << dendl;
 118     }
 119   }
 120
 121   // block ESIGPIPE
 122 #ifdef CEPH_USE_SO_NOSIGPIPE
 123   int val = 1;
 124   r = ::setsockopt(sd, SOL_SOCKET, SO_NOSIGPIPE, (void*)&val, sizeof(val));
 125   if (r) {
 126     r = errno;
 127     ldout(cct,0) << "couldn't set SO_NOSIGPIPE: " << cpp_strerror(r) << dendl;
 128   }
 129 #endif
 130   return -r;
 131 }
 132
 133 void NetHandler::set_priority(int sd, int prio, int domain)
 134 {
 135 #ifdef SO_PRIORITY
 136   if (prio < 0) {
 137     return;
 138   }
 139 #ifdef IPTOS_CLASS_CS6
 140   int iptos = IPTOS_CLASS_CS6;
 141   int r = -1;
 142   switch (domain) {
 143   case AF_INET:
 144     r = ::setsockopt(sd, IPPROTO_IP, IP_TOS, &iptos, sizeof(iptos));
 145     break;
 146   case AF_INET6:
 147     r = ::setsockopt(sd, IPPROTO_IPV6, IPV6_TCLASS, &iptos, sizeof(iptos));
 148     break;
 149   default:
 150     lderr(cct) << "couldn't set ToS of unknown family (" << domain << ")"
 151                << " to " << iptos << dendl;
 152     return;
 153   }
 154   if (r < 0) {
 155     r = errno;
 156     ldout(cct,0) << "couldn't set TOS to " << iptos
 157                  << ": " << cpp_strerror(r) << dendl;
 158   }
 159
 160 #endif  // IPTOS_CLASS_CS6
 161   // setsockopt(IPTOS_CLASS_CS6) sets the priority of the socket as 0.
 162   // See http://goo.gl/QWhvsD and http://goo.gl/laTbjT
 163   // We need to call setsockopt(SO_PRIORITY) after it.
 164   r = ::setsockopt(sd, SOL_SOCKET, SO_PRIORITY, &prio, sizeof(prio));
 165   if (r < 0) {
 166     r = errno;
 167     ldout(cct, 0) << __func__ << " couldn't set SO_PRIORITY to " << prio
 168                   << ": " << cpp_strerror(r) << dendl;
 169   }
 170 #else
 171   return;
 172 #endif  // SO_PRIORITY
 173 }
 174
 175 int NetHandler::generic_connect(const entity_addr_t& addr, const entity_addr_t &bind_addr, bool nonblock)
 176 {
 177   int ret;
 178   int s = create_socket(addr.get_family());
 179   if (s < 0)
 180     return s;
 181
 182   if (nonblock) {
 183     ret = set_nonblock(s);
 184     if (ret < 0) {
 185       close(s);
 186       return ret;
 187     }
 188   }
 189
 190   set_socket_options(s, cct->_conf->ms_tcp_nodelay, cct->_conf->ms_tcp_rcvbuf);
 191
 192   {
 193     entity_addr_t addr = bind_addr;
 194     if (cct->_conf->ms_bind_before_connect && (!addr.is_blank_ip())) {
 195       addr.set_port(0);
 196       ret = ::bind(s, addr.get_sockaddr(), addr.get_sockaddr_len());
 197       if (ret < 0) {
 198         ret = errno;
 199         ldout(cct, 2) << __func__ << " client bind error " << ", " << cpp_strerror(ret) << dendl;
 200         close(s);
 201         return -ret;
 202       }
 203     }
 204   }
 205
 206   ret = ::connect(s, addr.get_sockaddr(), addr.get_sockaddr_len());
 207   if (ret < 0) {
 208     ret = errno;
 209     if (errno == EINPROGRESS && nonblock)
 210       return s;
 211
 212     ldout(cct, 10) << __func__ << " connect: " << cpp_strerror(ret) << dendl;
 213     close(s);
 214     return -ret;
 215   }
 216
 217   return s;
 218 }
 219
 220 int NetHandler::reconnect(const entity_addr_t &addr, int sd)
 221 {
 222   int r = 0;
 223   int ret = ::connect(sd, addr.get_sockaddr(), addr.get_sockaddr_len());
 224
 225   if (ret < 0 && errno != EISCONN) {
 226     r = errno;
 227     ldout(cct, 10) << __func__ << " reconnect: " << strerror(r) << dendl;
 228     if (r == EINPROGRESS || r == EALREADY)
 229       return 1;
 230     return -r;
 231   }
 232
 233   return 0;
 234 }
 235
 236 int NetHandler::connect(const entity_addr_t &addr, const entity_addr_t& bind_addr)
 237 {
 238   return generic_connect(addr, bind_addr, false);
 239 }
 240
 241 int NetHandler::nonblock_connect(const entity_addr_t &addr, const entity_addr_t& bind_addr)
 242 {
 243   return generic_connect(addr, bind_addr, true);
 244 }
 245
 246
 247 }