Fixed connect(2) errors processing on old Linuxes.

While connect(2) states that non-blocking connect should use EPOLLOUT:

  EINPROGRESS
    The socket is non-blocking and the connection cannot be completed
    immediately.  It is possible to select(2) or poll(2) for completion by
    selecting the socket for writing.  After select(2) indicates writability,
    use getsockopt(2) to read the SO_ERROR option at level SOL_SOCKET to
    determine whether connect() completed successfully (SO_ERROR is zero)
    or unsuccessfully (SO_ERROR is one of the usual error codes listed here,
    explaining the reason for the failure).

On connect error, Linux 2.6.32 (CentOS 6) may return EPOLLRDHUP, EPOLLERR,
EPOLLHUP, EPOLLIN, but not EPOLLOUT.

4 files changed, 61 insertions, 30 deletions

diff --git a/src/nxt_conn_connect.c b/src/nxt_conn_connect.c
index 12b6c80c..d045853f 100644
--- a/src/nxt_conn_connect.c
+++ b/src/nxt_conn_connect.c
@@ -7,6 +7,9 @@
 #include <nxt_main.h>
 
 
+static nxt_err_t nxt_conn_connect_test_error(nxt_task_t *task, nxt_conn_t *c);
+
+
 void
 nxt_conn_sys_socket(nxt_task_t *task, void *obj, void *data)
 {
@@ -49,7 +52,7 @@ nxt_conn_io_connect(nxt_task_t *task, void *obj, void *data)
 
     case NXT_AGAIN:
         c->socket.write_handler = nxt_conn_connect_test;
-        c->socket.error_handler = state->error_handler;
+        c->socket.error_handler = nxt_conn_connect_error;
 
         engine = task->thread->engine;
 
@@ -118,8 +121,7 @@ nxt_conn_socket(nxt_task_t *task, nxt_conn_t *c)
 void
 nxt_conn_connect_test(nxt_task_t *task, void *obj, void *data)
 {
-    int         ret, err;
-    socklen_t   len;
+    nxt_err_t   err;
     nxt_conn_t  *c;
 
     c = obj;
@@ -132,48 +134,35 @@ nxt_conn_connect_test(nxt_task_t *task, void *obj, void *data)
         nxt_timer_disable(task->thread->engine, &c->write_timer);
     }
 
-    err = 0;
-    len = sizeof(int);
-
-    /*
-     * Linux and BSDs return 0 and store a pending error in the err argument;
-     * Solaris returns -1 and sets the errno.
-     */
-
-    ret = getsockopt(c->socket.fd, SOL_SOCKET, SO_ERROR, (void *) &err, &len);
-
-    if (nxt_slow_path(ret == -1)) {
-        err = nxt_errno;
-    }
+    err = nxt_conn_connect_test_error(task, c);
 
     if (err == 0) {
         nxt_work_queue_add(c->write_work_queue, c->write_state->ready_handler,
                            task, c, data);
-        return;
+    } else {
+        nxt_conn_connect_error(task, c, data);
     }
-
-    c->socket.error = err;
-
-    nxt_log(task, nxt_socket_error_level(err), "connect(%d, %*s) failed %E",
-            c->socket.fd, (size_t) c->remote->length,
-            nxt_sockaddr_start(c->remote), err);
-
-    nxt_conn_connect_error(task, c, data);
 }
 
 
 void
 nxt_conn_connect_error(nxt_task_t *task, void *obj, void *data)
 {
+    nxt_err_t               err;
     nxt_conn_t              *c;
     nxt_work_handler_t      handler;
     const nxt_conn_state_t  *state;
 
     c = obj;
+    err = c->socket.error;
+
+    if (err == 0) {
+        err = nxt_conn_connect_test_error(task, c);
+    }
 
     state = c->write_state;
 
-    switch (c->socket.error) {
+    switch (err) {
 
     case NXT_ECONNREFUSED:
 #if (NXT_LINUX)
@@ -193,3 +182,22 @@ nxt_conn_connect_error(nxt_task_t *task, void *obj, void *data)
 
     nxt_work_queue_add(c->write_work_queue, handler, task, c, data);
 }
+
+
+static nxt_err_t
+nxt_conn_connect_test_error(nxt_task_t *task, nxt_conn_t *c)
+{
+    nxt_err_t  err;
+
+    err = nxt_socket_error(c->socket.fd);
+
+    if (err != 0) {
+        c->socket.error = err;
+
+        nxt_log(task, nxt_socket_error_level(err), "connect(%d, %*s) failed %E",
+                c->socket.fd, (size_t) c->remote->length,
+                nxt_sockaddr_start(c->remote), err);
+    }
+
+    return err;
+}
diff --git a/src/nxt_epoll_engine.c b/src/nxt_epoll_engine.c
index 9cdaab9b..a944834e 100644
--- a/src/nxt_epoll_engine.c
+++ b/src/nxt_epoll_engine.c
@@ -944,12 +944,12 @@ nxt_epoll_poll(nxt_event_engine_t *engine, nxt_msec_t timeout)
                 nxt_work_queue_add(ev->read_work_queue, ev->read_handler,
                                    ev->task, ev, ev->data);
 
+                error = 0;
+
             } else if (engine->u.epoll.mode == 0) {
                 /* Level-triggered mode. */
                 nxt_epoll_disable_read(engine, ev);
             }
-
-            error = 0;
         }
 
         if ((events & EPOLLOUT) != 0) {
@@ -964,12 +964,12 @@ nxt_epoll_poll(nxt_event_engine_t *engine, nxt_msec_t timeout)
                 nxt_work_queue_add(ev->write_work_queue, ev->write_handler,
                                    ev->task, ev, ev->data);
 
+                error = 0;
+
             } else if (engine->u.epoll.mode == 0) {
                 /* Level-triggered mode. */
                 nxt_epoll_disable_write(engine, ev);
             }
-
-            error = 0;
         }
 
         if (!error) {
diff --git a/src/nxt_socket.c b/src/nxt_socket.c
index 95a298d8..a89663b1 100644
--- a/src/nxt_socket.c
+++ b/src/nxt_socket.c
@@ -300,6 +300,28 @@ nxt_socket_shutdown(nxt_task_t *task, nxt_socket_t s, nxt_uint_t how)
 }
 
 
+nxt_err_t
+nxt_socket_error(nxt_socket_t s)
+{
+    int        ret, err;
+    socklen_t  len;
+
+    err = 0;
+    len = sizeof(int);
+    /*  
+     * Linux and BSDs return 0 and store a pending error in the err argument; 
+     * Solaris returns -1 and sets the errno.
+     */  
+    ret = getsockopt(s, SOL_SOCKET, SO_ERROR, (void *) &err, &len);
+
+    if (nxt_slow_path(ret == -1)) {
+        err = nxt_errno;
+    }   
+
+    return err;
+}
+
+
 nxt_uint_t
 nxt_socket_error_level(nxt_err_t err)
 {
diff --git a/src/nxt_socket.h b/src/nxt_socket.h
index 3f00648d..6a450f83 100644
--- a/src/nxt_socket.h
+++ b/src/nxt_socket.h
@@ -106,6 +106,7 @@ NXT_EXPORT nxt_int_t nxt_socket_connect(nxt_task_t *task, nxt_socket_t s,
     nxt_sockaddr_t *sa);
 NXT_EXPORT void nxt_socket_shutdown(nxt_task_t *task, nxt_socket_t s,
     nxt_uint_t how);
+nxt_err_t nxt_socket_error(nxt_socket_t s);
 nxt_uint_t nxt_socket_error_level(nxt_err_t err);
 
 NXT_EXPORT nxt_int_t nxt_socketpair_create(nxt_task_t *task,