http upgrade and connections use non blocking sockets

CMakeLists.txt

@@ -13,6 +13,7 @@ set (CMAKE_CXX_EXTENSIONS OFF)
 set( IXWEBSOCKET_SOURCES
     ixwebsocket/IXEventFd.cpp
     ixwebsocket/IXSocket.cpp
+    ixwebsocket/IXSocketConnect.cpp
     ixwebsocket/IXWebSocket.cpp
     ixwebsocket/IXWebSocketTransport.cpp
     ixwebsocket/IXWebSocketPerMessageDeflate.cpp
@@ -22,6 +23,7 @@ set( IXWEBSOCKET_SOURCES
 set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXEventFd.h
     ixwebsocket/IXSocket.h
+    ixwebsocket/IXSocketConnect.h
     ixwebsocket/IXWebSocket.h
     ixwebsocket/IXWebSocketTransport.h
     ixwebsocket/IXWebSocketSendInfo.h

ixwebsocket/IXSocket.cpp

@@ -5,6 +5,7 @@
  */
 
 #include "IXSocket.h"
+#include "IXSocketConnect.h"
 
 #ifdef _WIN32
 # include <basetsd.h>
@@ -34,12 +35,6 @@
 #include <algorithm>
 #include <iostream>
 
-// Android needs extra headers for TCP_NODELAY and IPPROTO_TCP
-#ifdef ANDROID
-# include <linux/in.h>
-# include <linux/tcp.h>
-#endif
-
 namespace ix 
 {
     Socket::Socket() : 
@@ -53,97 +48,6 @@ namespace ix
         close();
     }
 
-    bool Socket::connectToAddress(const struct addrinfo *address, 
-                                  int& sockfd,
-                                  std::string& errMsg)
-    {
-        sockfd = -1;
-
-        int fd = socket(address->ai_family,
-                        address->ai_socktype,
-                        address->ai_protocol);
-        if (fd < 0)
-        {
-            errMsg = "Cannot create a socket";
-            return false;
-        }
-
-        int maxRetries = 3;
-        for (int i = 0; i < maxRetries; ++i)
-        {
-            if (::connect(fd, address->ai_addr, address->ai_addrlen) != -1)
-            {
-                sockfd = fd;
-                return true;
-            }
-
-            // EINTR means we've been interrupted, in which case we try again.
-            if (errno != EINTR) break;
-        }
-
-        closeSocket(fd);
-        sockfd = -1;
-        errMsg = strerror(errno);
-        return false;
-    }
-
-    int Socket::hostname_connect(const std::string& hostname,
-                                 int port,
-                                 std::string& errMsg)
-    {
-        struct addrinfo hints;
-        memset(&hints, 0, sizeof(hints));
-        hints.ai_flags = AI_ADDRCONFIG | AI_NUMERICSERV;
-        hints.ai_family = AF_UNSPEC;
-        hints.ai_socktype = SOCK_STREAM;
-
-        std::string sport = std::to_string(port);
-
-        struct addrinfo *res = nullptr;
-        int getaddrinfo_result = getaddrinfo(hostname.c_str(), sport.c_str(), 
-                                             &hints, &res);
-        if (getaddrinfo_result)
-        {
-            errMsg = gai_strerror(getaddrinfo_result);
-            return -1;
-        }
-
-        int sockfd = -1;
-
-        // iterate through the records to find a working peer
-        struct addrinfo *address;
-        bool success = false;
-        for (address = res; address != nullptr; address = address->ai_next)
-        {
-            success = connectToAddress(address, sockfd, errMsg);
-            if (success)
-            {
-                break;
-            }
-        }
-        freeaddrinfo(res);
-        return sockfd;
-    }
-
-    void Socket::configure()
-    {
-        int flag = 1;
-        setsockopt(_sockfd, IPPROTO_TCP, TCP_NODELAY, (char*) &flag, sizeof(flag)); // Disable Nagle's algorithm
-
-#ifdef _WIN32
-        unsigned long nonblocking = 1;
-        ioctlsocket(_sockfd, FIONBIO, &nonblocking);
-#else
-        fcntl(_sockfd, F_SETFL, O_NONBLOCK); // make socket non blocking
-#endif
-
-#ifdef SO_NOSIGPIPE
-        int value = 1;
-        setsockopt(_sockfd, SOL_SOCKET, SO_NOSIGPIPE, 
-                   (void *)&value, sizeof(value));
-#endif
-    }
-
     void Socket::poll(const OnPollCallback& onPollCallback)
     {
         if (_sockfd == -1)
@@ -181,7 +85,7 @@ namespace ix
 
         if (!_eventfd.clear()) return false;
 
-        _sockfd = Socket::hostname_connect(host, port, errMsg);
+        _sockfd = SocketConnect::connect(host, port, errMsg);
         return _sockfd != -1;
     }
 

ixwebsocket/IXSocket.h

@@ -24,9 +24,6 @@ namespace ix
         Socket();
         virtual ~Socket();
 
-        int hostname_connect(const std::string& hostname,
-                             int port,
-                             std::string& errMsg);
         void configure();
 
         virtual void poll(const OnPollCallback& onPollCallback);
@@ -54,9 +51,6 @@ namespace ix
         EventFd _eventfd;
 
     private:
-        bool connectToAddress(const struct addrinfo *address, 
-                              int& sockfd,
-                              std::string& errMsg);
     };
 
 }

ixwebsocket/IXSocketAppleSSL.cpp

@@ -6,6 +6,7 @@
  *  Adapted from Satori SDK Apple SSL code.
  */
 #include "IXSocketAppleSSL.h"
+#include "IXSocketConnect.h"
 
 #include <fcntl.h>
 #include <netdb.h>
@@ -162,7 +163,7 @@ namespace ix
         {
             std::lock_guard<std::mutex> lock(_mutex);
 
-            _sockfd = Socket::hostname_connect(host, port, errMsg);
+            _sockfd = SocketConnect::connect(host, port, errMsg);
             if (_sockfd == -1) return false;
 
             _sslContext = SSLCreateContext(kCFAllocatorDefault, kSSLClientSide, kSSLStreamType);

ixwebsocket/IXSocketConnect.cpp

@@ -0,0 +1,186 @@
+/*
+ *  IXSocketConnect.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXSocketConnect.h"
+
+#ifdef _WIN32
+# include <basetsd.h>
+# include <WinSock2.h>
+# include <ws2def.h>
+# include <WS2tcpip.h>
+# include <io.h>
+#else
+# include <unistd.h>
+# include <errno.h>
+# include <netdb.h>
+# include <netinet/tcp.h>
+# include <sys/socket.h>
+# include <sys/time.h>
+# include <sys/select.h>
+# include <sys/stat.h>
+#endif
+
+#include <fcntl.h>
+#include <sys/types.h>
+
+// Android needs extra headers for TCP_NODELAY and IPPROTO_TCP
+#ifdef ANDROID
+# include <linux/in.h>
+# include <linux/tcp.h>
+#endif
+
+
+namespace
+{
+    void closeSocket(int fd)
+    {
+#ifdef _WIN32
+        closesocket(fd);
+#else
+        ::close(fd);
+#endif
+    }
+}
+
+namespace ix 
+{
+    bool SocketConnect::connectToAddress(const struct addrinfo *address, 
+                                         int& sockfd,
+                                         std::string& errMsg)
+    {
+        sockfd = -1;
+
+        int fd = socket(address->ai_family,
+                        address->ai_socktype,
+                        address->ai_protocol);
+        if (fd < 0)
+        {
+            errMsg = "Cannot create a socket";
+            return false;
+        }
+
+        // Set the socket to non blocking mode, so that slow responses cannot
+        // block us for too long while we are trying to shut-down.
+        SocketConnect::configure(fd);
+
+        if (::connect(fd, address->ai_addr, address->ai_addrlen) == -1
+            && errno != EINPROGRESS)
+        {
+            closeSocket(fd);
+            sockfd = -1;
+            errMsg = strerror(errno);
+            return false;
+        }
+
+        // 10 seconds timeout, each time we wait for 50ms with select -> 200 attempts
+        int maxRetries = 200;
+        for (int i = 0; i < maxRetries; ++i)
+        {
+            fd_set wfds;
+            FD_ZERO(&wfds);
+            FD_SET(fd, &wfds);
+
+            // 50ms timeout
+            struct timeval timeout;
+            timeout.tv_sec = 0;
+            timeout.tv_usec = 1000 * 50;
+
+            select(fd + 1, nullptr, &wfds, nullptr, &timeout);
+
+            // Nothing was written to the socket, wait again.
+            if (!FD_ISSET(fd, &wfds)) continue;
+
+            // Something was written to the socket
+            int optval = -1;
+            socklen_t optlen;
+
+            // getsockopt() puts the errno value for connect into optval so 0
+            // means no-error.
+            if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &optval, &optlen) == -1 ||
+                optval != 0)
+            {
+                closeSocket(fd);
+                sockfd = -1;
+                errMsg = strerror(optval);
+                return false;
+            }
+            else
+            {
+                // Success !
+                sockfd = fd;
+                return true;
+            }
+        }
+
+        closeSocket(fd);
+        sockfd = -1;
+        errMsg = strerror(errno);
+        return false;
+    }
+
+    int SocketConnect::connect(const std::string& hostname,
+                               int port,
+                               std::string& errMsg)
+    {
+        //
+        // First do DNS resolution
+        //
+        struct addrinfo hints;
+        memset(&hints, 0, sizeof(hints));
+        hints.ai_flags = AI_ADDRCONFIG | AI_NUMERICSERV;
+        hints.ai_family = AF_UNSPEC;
+        hints.ai_socktype = SOCK_STREAM;
+
+        std::string sport = std::to_string(port);
+
+        struct addrinfo *res = nullptr;
+        int getaddrinfo_result = getaddrinfo(hostname.c_str(), sport.c_str(), 
+                                             &hints, &res);
+        if (getaddrinfo_result)
+        {
+            errMsg = gai_strerror(getaddrinfo_result);
+            return -1;
+        }
+
+        int sockfd = -1;
+
+        // iterate through the records to find a working peer
+        struct addrinfo *address;
+        bool success = false;
+        for (address = res; address != nullptr; address = address->ai_next)
+        {
+            //
+            // Second try to connect to the remote host
+            //
+            success = connectToAddress(address, sockfd, errMsg);
+            if (success)
+            {
+                break;
+            }
+        }
+        freeaddrinfo(res);
+        return sockfd;
+    }
+
+    void SocketConnect::configure(int sockfd)
+    {
+        int flag = 1;
+        setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, (char*) &flag, sizeof(flag)); // Disable Nagle's algorithm
+
+#ifdef _WIN32
+        unsigned long nonblocking = 1;
+        ioctlsocket(_sockfd, FIONBIO, &nonblocking);
+#else
+        fcntl(sockfd, F_SETFL, O_NONBLOCK); // make socket non blocking
+#endif
+
+#ifdef SO_NOSIGPIPE
+        int value = 1;
+        setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE, 
+                   (void *)&value, sizeof(value));
+#endif
+    }
+}

ixwebsocket/IXSocketConnect.h

@@ -0,0 +1,28 @@
+/*
+ *  IXSocketConnect.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include <string>
+
+struct addrinfo;
+
+namespace ix 
+{
+    class SocketConnect {
+    public:
+        static int connect(const std::string& hostname,
+                           int port,
+                           std::string& errMsg);
+
+        static bool connectToAddress(const struct addrinfo *address, 
+                                     int& sockfd,
+                                     std::string& errMsg);
+
+        static void configure(int sockfd);
+    };
+}
+

ixwebsocket/IXSocketOpenSSL.cpp

@@ -7,6 +7,7 @@
  */
 
 #include "IXSocketOpenSSL.h"
+#include "IXSocketConnect.h"
 #include <cassert>
 #include <iostream>
 
@@ -285,7 +286,7 @@ namespace ix
                 return false;
             }
 
-            _sockfd = Socket::hostname_connect(host, port, errMsg);
+            _sockfd = SocketConnect::connect(host, port, errMsg);
             if (_sockfd == -1) return false;
 
             _ssl_context = openSSLCreateContext(errMsg);

ixwebsocket/IXWebSocketTransport.cpp

@@ -231,14 +231,13 @@ namespace ix
         if (_socket->send(const_cast<char*>(request.c_str()), requestSize) != requestSize)
         {
             return WebSocketInitResult(false, 0, std::string("Failed sending GET request to ") + _url);
-
         }
 
         char line[256];
         int i;
         for (i = 0; i < 2 || (i < 255 && line[i-2] != '\r' && line[i-1] != '\n'); ++i)
         {
-            if (_socket->recv(line+i, 1) == 0)
+            if (!readByte(line+i))
             {
                 return WebSocketInitResult(false, 0, std::string("Failed reading HTTP status line from ") + _url);
             } 
@@ -282,7 +281,7 @@ namespace ix
                  i < 2 || (i < 255 && line[i-2] != '\r' && line[i-1] != '\n');
                  ++i)
             {
-                if (_socket->recv(line+i, 1) == 0)
+                if (!readByte(line+i))
                 {
                     return WebSocketInitResult(false, status, std::string("Failed reading response header from ") + _url);
                 }
@@ -344,7 +343,6 @@ namespace ix
             }
         }
 
-        _socket->configure();
         setReadyState(OPEN);
 
         return WebSocketInitResult(true, status, "", headers);
@@ -807,4 +805,33 @@ namespace ix
         _socket->close();
     }
 
+    // Used by init
+    bool WebSocketTransport::readByte(void* buffer)
+    {
+        while (true) 
+        {
+            if (_readyState == CLOSING) return false;
+
+            int ret;
+            ret = _socket->recv(buffer, 1);
+
+            // We read one byte, as needed, all good.
+            if (ret == 1)
+            {
+                return true;
+            }
+            // There is possibly something to be read, try again
+            else if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK || 
+                                 _socket->getErrno() == EAGAIN))
+            {
+                continue;
+            }
+            // There was an error during the read, abort
+            else 
+            {
+                return false;
+            }
+        }
+    }
+
 } // namespace ix

ixwebsocket/IXWebSocketTransport.h

@@ -157,5 +157,7 @@ namespace ix
         unsigned getRandomUnsigned();
         void unmaskReceiveBuffer(const wsheader_type& ws);
         std::string genRandomString(const int len);
+
+        bool readByte(void* buffer);
     };
 }