ext/kgio/connect.c

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