ext/kgio/connect.c

   1 #include "kgio.h"
   2 #include "my_fileno.h"
   3 #include "sock_for_fd.h"
   4 #include "blocking_io_region.h"
   5
   6 static void close_fail(int fd, const char *msg)
   7 {
   8         int saved_errno = errno;
   9         (void)close(fd);
  10         errno = saved_errno;
  11         rb_sys_fail(msg);
  12 }
  13
  14 static int MY_SOCK_STREAM =
  15 #if defined(SOCK_NONBLOCK) && defined(SOCK_CLOEXEC)
  16 #  ifdef HAVE_RB_FD_FIX_CLOEXEC
  17   (SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC)
  18 #  else
  19   (SOCK_STREAM|SOCK_NONBLOCK)
  20 #  endif
  21 #else
  22   SOCK_STREAM
  23 #endif /* ! SOCK_NONBLOCK */
  24 ;
  25
  26 /* do not set close-on-exec by default on Ruby <2.0.0 */
  27 #ifndef HAVE_RB_FD_FIX_CLOEXEC
  28 #  define rb_fd_fix_cloexec(fd) for (;0;)
  29 #endif /* HAVE_RB_FD_FIX_CLOEXEC */
  30
  31 /* try to use SOCK_NONBLOCK and SOCK_CLOEXEC */
  32 static int my_socket(int domain)
  33 {
  34         int fd;
  35
  36 retry:
  37         fd = socket(domain, MY_SOCK_STREAM, 0);
  38
  39         if (fd == -1) {
  40                 switch (errno) {
  41                 case EMFILE:
  42                 case ENFILE:
  43 #ifdef ENOBUFS
  44                 case ENOBUFS:
  45 #endif /* ENOBUFS */
  46                         errno = 0;
  47                         rb_gc();
  48                         fd = socket(domain, MY_SOCK_STREAM, 0);
  49                         break;
  50                 case EINVAL:
  51                         if (MY_SOCK_STREAM != SOCK_STREAM) {
  52                                 MY_SOCK_STREAM = SOCK_STREAM;
  53                                 goto retry;
  54                         }
  55                 }
  56                 if (fd == -1)
  57                         rb_sys_fail("socket");
  58         }
  59
  60         if (MY_SOCK_STREAM == SOCK_STREAM) {
  61                 if (fcntl(fd, F_SETFL, O_RDWR | O_NONBLOCK) == -1)
  62                         close_fail(fd, "fcntl(F_SETFL, O_RDWR | O_NONBLOCK)");
  63                 rb_fd_fix_cloexec(fd);
  64         }
  65
  66         return fd;
  67 }
  68
  69 static VALUE
  70 my_connect(VALUE klass, int io_wait, int domain, void *addr, socklen_t addrlen)
  71 {
  72         int fd = my_socket(domain);
  73
  74         if (connect(fd, addr, addrlen) == -1) {
  75                 if (errno == EINPROGRESS) {
  76                         VALUE io = sock_for_fd(klass, fd);
  77
  78                         if (io_wait) {
  79                                 errno = EAGAIN;
  80                                 (void)kgio_call_wait_writable(io);
  81                         }
  82                         return io;
  83                 }
  84                 close_fail(fd, "connect");
  85         }
  86         return sock_for_fd(klass, fd);
  87 }
  88
  89 static void
  90 tcp_getaddr(struct addrinfo *hints, struct sockaddr_storage *addr,
  91              VALUE ip, VALUE port)
  92 {
  93         int rc;
  94         struct addrinfo *res;
  95         const char *ipname = StringValuePtr(ip);
  96         char ipport[6];
  97         unsigned uport;
  98
  99         if (TYPE(port) != T_FIXNUM)
 100                 rb_raise(rb_eTypeError, "port must be a non-negative integer");
 101         uport = FIX2UINT(port);
 102
 103         rc = snprintf(ipport, sizeof(ipport), "%u", uport);
 104         if (rc >= (int)sizeof(ipport) || rc <= 0)
 105                 rb_raise(rb_eArgError, "invalid TCP port: %u", uport);
 106         memset(hints, 0, sizeof(struct addrinfo));
 107         hints->ai_family = AF_UNSPEC;
 108         hints->ai_socktype = SOCK_STREAM;
 109         hints->ai_protocol = IPPROTO_TCP;
 110         /* disallow non-deterministic DNS lookups */
 111         hints->ai_flags = AI_NUMERICHOST;
 112
 113         rc = getaddrinfo(ipname, ipport, hints, &res);
 114         if (rc != 0)
 115                 rb_raise(rb_eArgError, "getaddrinfo(%s:%s): %s",
 116                          ipname, ipport, gai_strerror(rc));
 117
 118         /* copy needed data and free ASAP to avoid needing rb_ensure */
 119         hints->ai_family = res->ai_family;
 120         hints->ai_addrlen = res->ai_addrlen;
 121         memcpy(addr, res->ai_addr, res->ai_addrlen);
 122         freeaddrinfo(res);
 123 }
 124
 125 static VALUE tcp_connect(VALUE klass, VALUE ip, VALUE port, int io_wait)
 126 {
 127         struct addrinfo hints;
 128         struct sockaddr_storage addr;
 129
 130         tcp_getaddr(&hints, &addr, ip, port);
 131
 132         return my_connect(klass, io_wait, hints.ai_family,
 133                           &addr, hints.ai_addrlen);
 134 }
 135
 136 static struct sockaddr *sockaddr_from(socklen_t *addrlen, VALUE addr)
 137 {
 138         if (TYPE(addr) == T_STRING) {
 139                 *addrlen = (socklen_t)RSTRING_LEN(addr);
 140                 return (struct sockaddr *)(RSTRING_PTR(addr));
 141         }
 142         rb_raise(rb_eTypeError, "invalid address");
 143         return NULL;
 144 }
 145
 146 #if defined(MSG_FASTOPEN) && defined(HAVE_RB_THREAD_BLOCKING_REGION)
 147 #ifndef HAVE_RB_STR_SUBSEQ
 148 #define rb_str_subseq rb_str_substr
 149 #endif
 150 struct tfo_args {
 151         int fd;
 152         void *buf;
 153         size_t buflen;
 154         struct sockaddr *addr;
 155         socklen_t addrlen;
 156 };
 157
 158 static VALUE tfo_sendto(void *_a)
 159 {
 160         struct tfo_args *a = _a;
 161         ssize_t w;
 162
 163         w = sendto(a->fd, a->buf, a->buflen, MSG_FASTOPEN, a->addr, a->addrlen);
 164         return (VALUE)w;
 165 }
 166
 167 /*
 168  * call-seq:
 169  *
 170  *      s = Kgio::Socket.new(:INET, :STREAM)
 171  *      addr = Socket.pack_sockaddr_in(80, "example.com")
 172  *      s.fastopen("hello world", addr) -> nil
 173  *
 174  * Starts a TCP connection using TCP Fast Open.  This uses a blocking
 175  * sendto() syscall and is only available on Ruby 1.9 or later.
 176  * This raises exceptions (including Errno::EINPROGRESS/Errno::EAGAIN)
 177  * on errors.  Using this is only recommended for blocking sockets.
 178  *      s.setsockopt(:SOCKET, :SNDTIMEO, [1,0].pack("l_l_"))
 179  */
 180 static VALUE fastopen(VALUE sock, VALUE buf, VALUE addr)
 181 {
 182         struct tfo_args a;
 183         VALUE str = (TYPE(buf) == T_STRING) ? buf : rb_obj_as_string(buf);
 184         ssize_t w;
 185
 186         a.fd = my_fileno(sock);
 187         a.buf = RSTRING_PTR(str);
 188         a.buflen = (size_t)RSTRING_LEN(str);
 189         a.addr = sockaddr_from(&a.addrlen, addr);
 190
 191         /* n.b. rb_thread_blocking_region preserves errno */
 192         w = (ssize_t)rb_thread_io_blocking_region(tfo_sendto, &a, a.fd);
 193         if (w < 0)
 194                 rb_sys_fail("sendto");
 195         if ((size_t)w == a.buflen)
 196                 return Qnil;
 197
 198         return rb_str_subseq(str, w, a.buflen - w);
 199 }
 200 #endif /* MSG_FASTOPEN */
 201
 202 /*
 203  * call-seq:
 204  *
 205  *      Kgio::TCPSocket.new('127.0.0.1', 80) -> socket
 206  *
 207  * Creates a new Kgio::TCPSocket object and initiates a
 208  * non-blocking connection.
 209  *
 210  * This may block and call any method defined to +kgio_wait_writable+
 211  * for the class.
 212  *
 213  * Unlike the TCPSocket.new in Ruby, this does NOT perform DNS
 214  * lookups (which is subject to a different set of timeouts and
 215  * best handled elsewhere).
 216  */
 217 static VALUE kgio_tcp_connect(VALUE klass, VALUE ip, VALUE port)
 218 {
 219         return tcp_connect(klass, ip, port, 1);
 220 }
 221
 222 /*
 223  * call-seq:
 224  *
 225  *      Kgio::TCPSocket.start('127.0.0.1', 80) -> socket
 226  *
 227  * Creates a new Kgio::TCPSocket object and initiates a
 228  * non-blocking connection.  The caller should select/poll
 229  * on the socket for writability before attempting to write
 230  * or optimistically attempt a write and handle :wait_writable
 231  * or Errno::EAGAIN.
 232  *
 233  * Unlike the TCPSocket.new in Ruby, this does NOT perform DNS
 234  * lookups (which is subject to a different set of timeouts and
 235  * best handled elsewhere).
 236  */
 237 static VALUE kgio_tcp_start(VALUE klass, VALUE ip, VALUE port)
 238 {
 239         return tcp_connect(klass, ip, port, 0);
 240 }
 241
 242 static VALUE unix_connect(VALUE klass, VALUE path, int io_wait)
 243 {
 244         struct sockaddr_un addr = { 0 };
 245         long len;
 246
 247         StringValue(path);
 248         len = RSTRING_LEN(path);
 249         if ((long)sizeof(addr.sun_path) <= len)
 250                 rb_raise(rb_eArgError,
 251                          "too long unix socket path (max: %dbytes)",
 252                          (int)sizeof(addr.sun_path)-1);
 253
 254         memcpy(addr.sun_path, RSTRING_PTR(path), len);
 255         addr.sun_family = AF_UNIX;
 256
 257         return my_connect(klass, io_wait, PF_UNIX, &addr, sizeof(addr));
 258 }
 259
 260 /*
 261  * call-seq:
 262  *
 263  *      Kgio::UNIXSocket.new("/path/to/unix/socket") -> socket
 264  *
 265  * Creates a new Kgio::UNIXSocket object and initiates a
 266  * non-blocking connection.
 267  *
 268  * This may block and call any method defined to +kgio_wait_writable+
 269  * for the class.
 270  */
 271 static VALUE kgio_unix_connect(VALUE klass, VALUE path)
 272 {
 273         return unix_connect(klass, path, 1);
 274 }
 275
 276 /*
 277  * call-seq:
 278  *
 279  *      Kgio::UNIXSocket.start("/path/to/unix/socket") -> socket
 280  *
 281  * Creates a new Kgio::UNIXSocket object and initiates a
 282  * non-blocking connection.  The caller should select/poll
 283  * on the socket for writability before attempting to write
 284  * or optimistically attempt a write and handle :wait_writable
 285  * or Errno::EAGAIN.
 286  */
 287 static VALUE kgio_unix_start(VALUE klass, VALUE path)
 288 {
 289         return unix_connect(klass, path, 0);
 290 }
 291
 292 static VALUE stream_connect(VALUE klass, VALUE addr, int io_wait)
 293 {
 294         int domain;
 295         socklen_t addrlen;
 296         struct sockaddr *sockaddr = sockaddr_from(&addrlen, addr);
 297
 298         switch (((struct sockaddr_storage *)(sockaddr))->ss_family) {
 299         case AF_UNIX: domain = PF_UNIX; break;
 300         case AF_INET: domain = PF_INET; break;
 301         case AF_INET6: domain = PF_INET6; break;
 302         default:
 303                 rb_raise(rb_eArgError, "invalid address family");
 304         }
 305
 306         return my_connect(klass, io_wait, domain, sockaddr, addrlen);
 307 }
 308
 309 /* call-seq:
 310  *
 311  *      addr = Socket.pack_sockaddr_in(80, 'example.com')
 312  *      Kgio::Socket.connect(addr) -> socket
 313  *
 314  *      addr = Socket.pack_sockaddr_un("/path/to/unix/socket")
 315  *      Kgio::Socket.connect(addr) -> socket
 316  *
 317  * Creates a generic Kgio::Socket object and initiates a
 318  * non-blocking connection.
 319  *
 320  * This may block and call any method defined to +kgio_wait_writable+
 321  * for the class.
 322  */
 323 static VALUE kgio_connect(VALUE klass, VALUE addr)
 324 {
 325         return stream_connect(klass, addr, 1);
 326 }
 327
 328 /*
 329  * If passed one argument, this is identical to Kgio::Socket.connect.
 330  * If passed two or three arguments, it uses its superclass method:
 331  *
 332  *   Socket.new(domain, socktype [, protocol ])
 333  */
 334 static VALUE kgio_new(int argc, VALUE *argv, VALUE klass)
 335 {
 336         if (argc == 1)
 337                 /* backwards compat, the only way for kgio <= 2.7.4 */
 338                 return stream_connect(klass, argv[0], 1);
 339
 340         return rb_call_super(argc, argv);
 341 }
 342
 343 /* call-seq:
 344  *
 345  *      addr = Socket.pack_sockaddr_in(80, 'example.com')
 346  *      Kgio::Socket.start(addr) -> socket
 347  *
 348  *      addr = Socket.pack_sockaddr_un("/path/to/unix/socket")
 349  *      Kgio::Socket.start(addr) -> socket
 350  *
 351  * Creates a generic Kgio::Socket object and initiates a
 352  * non-blocking connection.  The caller should select/poll
 353  * on the socket for writability before attempting to write
 354  * or optimistically attempt a write and handle :wait_writable
 355  * or Errno::EAGAIN.
 356  */
 357 static VALUE kgio_start(VALUE klass, VALUE addr)
 358 {
 359         return stream_connect(klass, addr, 0);
 360 }
 361
 362 void init_kgio_connect(void)
 363 {
 364         VALUE mKgio = rb_define_module("Kgio");
 365         VALUE cSocket = rb_const_get(rb_cObject, rb_intern("Socket"));
 366         VALUE mSocketMethods = rb_const_get(mKgio, rb_intern("SocketMethods"));
 367         VALUE cKgio_Socket, cTCPSocket, cUNIXSocket;
 368
 369         /*
 370          * Document-class: Kgio::Socket
 371          *
 372          * A generic socket class with Kgio::SocketMethods included.
 373          * This is returned by all Kgio methods that accept(2) a connected
 374          * stream socket.
 375          */
 376         cKgio_Socket = rb_define_class_under(mKgio, "Socket", cSocket);
 377         rb_include_module(cKgio_Socket, mSocketMethods);
 378         rb_define_singleton_method(cKgio_Socket, "new", kgio_new, -1);
 379         rb_define_singleton_method(cKgio_Socket, "connect", kgio_connect, 1);
 380         rb_define_singleton_method(cKgio_Socket, "start", kgio_start, 1);
 381 #if defined(MSG_FASTOPEN) && defined(HAVE_RB_THREAD_BLOCKING_REGION)
 382         rb_define_method(cKgio_Socket, "fastopen", fastopen, 2);
 383 #endif
 384         /*
 385          * Document-class: Kgio::TCPSocket
 386          *
 387          * Kgio::TCPSocket should be used in place of the plain TCPSocket
 388          * when kgio_* methods are needed.
 389          */
 390         cTCPSocket = rb_const_get(rb_cObject, rb_intern("TCPSocket"));
 391         cTCPSocket = rb_define_class_under(mKgio, "TCPSocket", cTCPSocket);
 392         rb_include_module(cTCPSocket, mSocketMethods);
 393         rb_define_singleton_method(cTCPSocket, "new", kgio_tcp_connect, 2);
 394         rb_define_singleton_method(cTCPSocket, "start", kgio_tcp_start, 2);
 395
 396         /*
 397          * Document-class: Kgio::UNIXSocket
 398          *
 399          * Kgio::UNIXSocket should be used in place of the plain UNIXSocket
 400          * when kgio_* methods are needed.
 401          */
 402         cUNIXSocket = rb_const_get(rb_cObject, rb_intern("UNIXSocket"));
 403         cUNIXSocket = rb_define_class_under(mKgio, "UNIXSocket", cUNIXSocket);
 404         rb_include_module(cUNIXSocket, mSocketMethods);
 405         rb_define_singleton_method(cUNIXSocket, "new", kgio_unix_connect, 1);
 406         rb_define_singleton_method(cUNIXSocket, "start", kgio_unix_start, 1);
 407         init_sock_for_fd();
 408 }