ext/kgio/connect.c

   1 #include "kgio.h"
   2 #include "sock_for_fd.h"
   3
   4 static void close_fail(int fd, const char *msg)
   5 {
   6         int saved_errno = errno;
   7         (void)close(fd);
   8         errno = saved_errno;
   9         rb_sys_fail(msg);
  10 }
  11
  12 #ifdef SOCK_NONBLOCK
  13 #  define MY_SOCK_STREAM (SOCK_STREAM|SOCK_NONBLOCK)
  14 #else
  15 #  define MY_SOCK_STREAM SOCK_STREAM
  16 #endif /* ! SOCK_NONBLOCK */
  17
  18 static VALUE
  19 my_connect(VALUE klass, int io_wait, int domain, void *addr, socklen_t addrlen)
  20 {
  21         int fd = socket(domain, MY_SOCK_STREAM, 0);
  22
  23         if (fd == -1) {
  24                 switch (errno) {
  25                 case EMFILE:
  26                 case ENFILE:
  27 #ifdef ENOBUFS
  28                 case ENOBUFS:
  29 #endif /* ENOBUFS */
  30                         errno = 0;
  31                         rb_gc();
  32                         fd = socket(domain, MY_SOCK_STREAM, 0);
  33                 }
  34                 if (fd == -1)
  35                         rb_sys_fail("socket");
  36         }
  37
  38 #ifndef SOCK_NONBLOCK
  39         if (fcntl(fd, F_SETFL, O_RDWR | O_NONBLOCK) == -1)
  40                 close_fail(fd, "fcntl(F_SETFL, O_RDWR | O_NONBLOCK)");
  41 #endif /* SOCK_NONBLOCK */
  42
  43         if (connect(fd, addr, addrlen) == -1) {
  44                 if (errno == EINPROGRESS) {
  45                         VALUE io = sock_for_fd(klass, fd);
  46
  47                         if (io_wait) {
  48                                 errno = EAGAIN;
  49                                 kgio_call_wait_writable(io, fd);
  50                         }
  51                         return io;
  52                 }
  53                 close_fail(fd, "connect");
  54         }
  55         return sock_for_fd(klass, fd);
  56 }
  57
  58 static VALUE tcp_connect(VALUE klass, VALUE ip, VALUE port, int io_wait)
  59 {
  60         struct sockaddr_in addr = { 0 };
  61
  62         addr.sin_family = AF_INET;
  63         addr.sin_port = htons((unsigned short)NUM2INT(port));
  64
  65         switch (inet_pton(AF_INET, StringValuePtr(ip), &addr.sin_addr)) {
  66         case 1:
  67                 return my_connect(klass, io_wait, PF_INET, &addr, sizeof(addr));
  68         case -1:
  69                 rb_sys_fail("inet_pton");
  70         }
  71         rb_raise(rb_eArgError, "invalid address: %s", StringValuePtr(ip));
  72
  73         return Qnil;
  74 }
  75
  76 /*
  77  * call-seq:
  78  *
  79  *      Kgio::TCPSocket.new('127.0.0.1', 80) -> socket
  80  *
  81  * Creates a new Kgio::TCPSocket object and initiates a
  82  * non-blocking connection.
  83  *
  84  * This may block and call any method assigned to Kgio.wait_writable.
  85  *
  86  * Unlike the TCPSocket.new in Ruby, this does NOT perform DNS
  87  * lookups (which is subject to a different set of timeouts and
  88  * best handled elsewhere).
  89  */
  90 static VALUE kgio_tcp_connect(VALUE klass, VALUE ip, VALUE port)
  91 {
  92         return tcp_connect(klass, ip, port, 1);
  93 }
  94
  95 /*
  96  * call-seq:
  97  *
  98  *      Kgio::TCPSocket.start('127.0.0.1', 80) -> socket
  99  *
 100  * Creates a new Kgio::TCPSocket object and initiates a
 101  * non-blocking connection.  The caller should select/poll
 102  * on the socket for writability before attempting to write
 103  * or optimistically attempt a write and handle :wait_writable
 104  * or Errno::EAGAIN.
 105  *
 106  * Unlike the TCPSocket.new in Ruby, this does NOT perform DNS
 107  * lookups (which is subject to a different set of timeouts and
 108  * best handled elsewhere).
 109  */
 110 static VALUE kgio_tcp_start(VALUE klass, VALUE ip, VALUE port)
 111 {
 112         return tcp_connect(klass, ip, port, 0);
 113 }
 114
 115 static VALUE unix_connect(VALUE klass, VALUE path, int io_wait)
 116 {
 117         struct sockaddr_un addr = { 0 };
 118         long len;
 119
 120         StringValue(path);
 121         len = RSTRING_LEN(path);
 122         if (sizeof(addr.sun_path) <= len)
 123                 rb_raise(rb_eArgError,
 124                          "too long unix socket path (max: %dbytes)",
 125                          (int)sizeof(addr.sun_path)-1);
 126
 127         memcpy(addr.sun_path, RSTRING_PTR(path), len);
 128         addr.sun_family = AF_UNIX;
 129
 130         return my_connect(klass, io_wait, PF_UNIX, &addr, sizeof(addr));
 131 }
 132
 133 /*
 134  * call-seq:
 135  *
 136  *      Kgio::UNIXSocket.new("/path/to/unix/socket") -> socket
 137  *
 138  * Creates a new Kgio::UNIXSocket object and initiates a
 139  * non-blocking connection.
 140  *
 141  * This may block and call any method assigned to Kgio.wait_writable.
 142  */
 143 static VALUE kgio_unix_connect(VALUE klass, VALUE path)
 144 {
 145         return unix_connect(klass, path, 1);
 146 }
 147
 148 /*
 149  * call-seq:
 150  *
 151  *      Kgio::UNIXSocket.start("/path/to/unix/socket") -> socket
 152  *
 153  * Creates a new Kgio::UNIXSocket object and initiates a
 154  * non-blocking connection.  The caller should select/poll
 155  * on the socket for writability before attempting to write
 156  * or optimistically attempt a write and handle :wait_writable
 157  * or Errno::EAGAIN.
 158  */
 159 static VALUE kgio_unix_start(VALUE klass, VALUE path)
 160 {
 161         return unix_connect(klass, path, 0);
 162 }
 163
 164 static VALUE stream_connect(VALUE klass, VALUE addr, int io_wait)
 165 {
 166         int domain;
 167         socklen_t addrlen;
 168         struct sockaddr *sockaddr;
 169
 170         if (TYPE(addr) == T_STRING) {
 171                 sockaddr = (struct sockaddr *)(RSTRING_PTR(addr));
 172                 addrlen = (socklen_t)RSTRING_LEN(addr);
 173         } else {
 174                 rb_raise(rb_eTypeError, "invalid address");
 175         }
 176         switch (((struct sockaddr_in *)(sockaddr))->sin_family) {
 177         case AF_UNIX: domain = PF_UNIX; break;
 178         case AF_INET: domain = PF_INET; break;
 179 #ifdef AF_INET6 /* IPv6 support incomplete */
 180         case AF_INET6: domain = PF_INET6; break;
 181 #endif /* AF_INET6 */
 182         default:
 183                 rb_raise(rb_eArgError, "invalid address family");
 184         }
 185
 186         return my_connect(klass, io_wait, domain, sockaddr, addrlen);
 187 }
 188
 189 /* call-seq:
 190  *
 191  *      addr = Socket.pack_sockaddr_in(80, 'example.com')
 192  *      Kgio::Socket.connect(addr) -> socket
 193  *
 194  *      addr = Socket.pack_sockaddr_un("/path/to/unix/socket")
 195  *      Kgio::Socket.connect(addr) -> socket
 196  *
 197  * Creates a generic Kgio::Socket object and initiates a
 198  * non-blocking connection.
 199  *
 200  * This may block and call any method assigned to Kgio.wait_writable.
 201  */
 202 static VALUE kgio_connect(VALUE klass, VALUE addr)
 203 {
 204         return stream_connect(klass, addr, 1);
 205 }
 206
 207 /* call-seq:
 208  *
 209  *      addr = Socket.pack_sockaddr_in(80, 'example.com')
 210  *      Kgio::Socket.start(addr) -> socket
 211  *
 212  *      addr = Socket.pack_sockaddr_un("/path/to/unix/socket")
 213  *      Kgio::Socket.start(addr) -> socket
 214  *
 215  * Creates a generic Kgio::Socket object and initiates a
 216  * non-blocking connection.  The caller should select/poll
 217  * on the socket for writability before attempting to write
 218  * or optimistically attempt a write and handle :wait_writable
 219  * or Errno::EAGAIN.
 220  */
 221 static VALUE kgio_start(VALUE klass, VALUE addr)
 222 {
 223         return stream_connect(klass, addr, 0);
 224 }
 225
 226 void init_kgio_connect(void)
 227 {
 228         VALUE mKgio = rb_define_module("Kgio");
 229         VALUE cSocket = rb_const_get(rb_cObject, rb_intern("Socket"));
 230         VALUE mSocketMethods = rb_const_get(mKgio, rb_intern("SocketMethods"));
 231         VALUE cKgio_Socket, cTCPSocket, cUNIXSocket;
 232
 233         /*
 234          * Document-class: Kgio::Socket
 235          *
 236          * A generic socket class with Kgio::SocketMethods included.
 237          * This is returned by all Kgio methods that accept(2) a connected
 238          * stream socket.
 239          */
 240         cKgio_Socket = rb_define_class_under(mKgio, "Socket", cSocket);
 241         rb_include_module(cKgio_Socket, mSocketMethods);
 242         rb_define_singleton_method(cKgio_Socket, "new", kgio_connect, 1);
 243         rb_define_singleton_method(cKgio_Socket, "start", kgio_start, 1);
 244
 245         cTCPSocket = rb_const_get(rb_cObject, rb_intern("TCPSocket"));
 246         cTCPSocket = rb_define_class_under(mKgio, "TCPSocket", cTCPSocket);
 247         rb_include_module(cTCPSocket, mSocketMethods);
 248         rb_define_singleton_method(cTCPSocket, "new", kgio_tcp_connect, 2);
 249         rb_define_singleton_method(cTCPSocket, "start", kgio_tcp_start, 2);
 250
 251         cUNIXSocket = rb_const_get(rb_cObject, rb_intern("UNIXSocket"));
 252         cUNIXSocket = rb_define_class_under(mKgio, "UNIXSocket", cUNIXSocket);
 253         rb_include_module(cUNIXSocket, mSocketMethods);
 254         rb_define_singleton_method(cUNIXSocket, "new", kgio_unix_connect, 1);
 255         rb_define_singleton_method(cUNIXSocket, "start", kgio_unix_start, 1);
 256         init_sock_for_fd();
 257 }