Follow up to r1895106: Use less expensive atomics for wakeup.

If pipe writers (wakeup) put a single byte until consume it's consumed by the
reader (drain), an atomic cas for the writers (still) and an atomic (re)set
for the reader is enough (not need to cas on the reader side).

This requires that the reader never blocks on read though (e.g. spurious return
from poll), so (re)make the read side on the pipe non-blocking (finally).

Since synchronous non-blocking read is not a thing for Windows' Readfile(), add
a ->socket flag this arch's to apr_file_t (like the existing ->pipe one) which
file_socket_pipe_create() will set to make apr_file_read/write() handle
non-blocking (nor overlapped) socket pipes with WSARecv/Send().



git-svn-id: https://svn.apache.org/repos/asf/apr/apr/trunk@1895175 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 72 insertions, 40 deletions

diff --git a/file_io/win32/pipe.c b/file_io/win32/pipe.c
index 8de7822af..0b54b3112 100644
--- a/file_io/win32/pipe.c
+++ b/file_io/win32/pipe.c
@@ -378,32 +378,28 @@ static apr_status_t create_socket_pipe(SOCKET *rd, SOCKET *wr)
             rv =  apr_get_netos_error();
             goto cleanup;
         }
-        /* Verify the connection by reading the send identification.
-         */
-        do {
-            if (nc++)
-                Sleep(1);
-            nrd = recv(*rd, (char *)iid, sizeof(iid), 0);
-            rv = nrd == SOCKET_ERROR ? apr_get_netos_error() : APR_SUCCESS;
-        } while (APR_STATUS_IS_EAGAIN(rv));
-
-        if (nrd == sizeof(iid)) {
-            if (memcmp(uid, iid, sizeof(uid)) == 0) {
-                /* Wow, we recived what we send.
-                 * Put read side of the pipe to the blocking
-                 * mode and return.
-                 */
-                bm = 0;
-                if (ioctlsocket(*rd, FIONBIO, &bm) == SOCKET_ERROR) {
-                    rv = apr_get_netos_error();
-                    goto cleanup;
-                }
-                break;
-            }
+        /* Verify the connection by reading/waiting for the identification */
+        bm = 0;
+        if (ioctlsocket(*rd, FIONBIO, &bm) == SOCKET_ERROR) {
+            rv = apr_get_netos_error();
+            goto cleanup;
         }
-        else if (nrd == SOCKET_ERROR) {
+        nrd = recv(*rd, (char *)iid, sizeof(iid), 0);
+        if (nrd == SOCKET_ERROR) {
+            rv = apr_get_netos_error();
             goto cleanup;
         }
+        if (nrd == (int)sizeof(uid) && memcmp(iid, uid, sizeof(uid)) == 0) {
+            /* Got the right identifier, put the poll()able read side of
+             * the pipe in nonblocking mode and return.
+             */
+            bm = 1;
+            if (ioctlsocket(*rd, FIONBIO, &bm) == SOCKET_ERROR) {
+                rv = apr_get_netos_error();
+                goto cleanup;
+            }
+            break;
+        }
         closesocket(*rd);
     }
     /* We don't need the listening socket any more */
@@ -412,6 +408,7 @@ static apr_status_t create_socket_pipe(SOCKET *rd, SOCKET *wr)
 
 cleanup:
     /* Don't leak resources */
+    closesocket(ls);
     if (*rd != INVALID_SOCKET)
         closesocket(*rd);
     if (*wr != INVALID_SOCKET)
@@ -419,7 +416,6 @@ cleanup:
 
     *rd = INVALID_SOCKET;
     *wr = INVALID_SOCKET;
-    closesocket(ls);
     return rv;
 }
 
@@ -448,21 +444,21 @@ apr_status_t file_socket_pipe_create(apr_file_t **in,
     (*in) = (apr_file_t *)apr_pcalloc(p, sizeof(apr_file_t));
     (*in)->pool = p;
     (*in)->fname = NULL;
-    (*in)->pipe = 1;
-    (*in)->timeout = -1;
+    (*in)->socket = 1;
+    (*in)->timeout = 0; /* read end of the pipe is non-blocking */
     (*in)->ungetchar = -1;
     (*in)->eof_hit = 0;
     (*in)->filePtr = 0;
     (*in)->bufpos = 0;
     (*in)->dataRead = 0;
     (*in)->direction = 0;
-    (*in)->pOverlapped = (OVERLAPPED*)apr_pcalloc(p, sizeof(OVERLAPPED));
+    (*in)->pOverlapped = NULL;
     (*in)->filehand = (HANDLE)rd;
 
     (*out) = (apr_file_t *)apr_pcalloc(p, sizeof(apr_file_t));
     (*out)->pool = p;
     (*out)->fname = NULL;
-    (*out)->pipe = 1;
+    (*out)->socket = 1;
     (*out)->timeout = -1;
     (*out)->ungetchar = -1;
     (*out)->eof_hit = 0;
@@ -470,7 +466,7 @@ apr_status_t file_socket_pipe_create(apr_file_t **in,
     (*out)->bufpos = 0;
     (*out)->dataRead = 0;
     (*out)->direction = 0;
-    (*out)->pOverlapped = (OVERLAPPED*)apr_pcalloc(p, sizeof(OVERLAPPED));
+    (*out)->pOverlapped = NULL;
     (*out)->filehand = (HANDLE)wr;
 
     apr_pool_cleanup_register(p, (void *)(*in), socket_pipe_cleanup,
@@ -484,7 +480,7 @@ apr_status_t file_socket_pipe_create(apr_file_t **in,
 apr_status_t file_socket_pipe_close(apr_file_t *file)
 {
     apr_status_t stat;
-    if (!file->pipe)
+    if (!file->socket)
         return apr_file_close(file);
     if ((stat = socket_pipe_cleanup(file)) == APR_SUCCESS) {
         apr_pool_cleanup_kill(file->pool, file, socket_pipe_cleanup);
diff --git a/file_io/win32/readwrite.c b/file_io/win32/readwrite.c
index 23d2358e7..7254d3e33 100644
--- a/file_io/win32/readwrite.c
+++ b/file_io/win32/readwrite.c
@@ -20,10 +20,12 @@
 #include "apr_strings.h"
 #include "apr_lib.h"
 #include "apr_errno.h"
-#include <malloc.h>
+#include "apr_arch_networkio.h"
 #include "apr_arch_atime.h"
 #include "apr_arch_misc.h"
 
+#include <malloc.h>
+
 /*
  * read_with_timeout() 
  * Uses async i/o to emulate unix non-blocking i/o with timeouts.
@@ -31,6 +33,7 @@
 static apr_status_t read_with_timeout(apr_file_t *file, void *buf, apr_size_t len_in, apr_size_t *nbytes)
 {
     apr_status_t rv;
+    int pipe_or_socket = (file->pipe || file->socket);
     DWORD len = (DWORD)len_in;
     DWORD bytesread = 0;
 
@@ -67,13 +70,28 @@ static apr_status_t read_with_timeout(apr_file_t *file, void *buf, apr_size_t le
         }
     }
 
-    if (file->pOverlapped && !file->pipe) {
+    if (file->pOverlapped && !pipe_or_socket) {
         file->pOverlapped->Offset     = (DWORD)file->filePtr;
         file->pOverlapped->OffsetHigh = (DWORD)(file->filePtr >> 32);
     }
 
-    if (ReadFile(file->filehand, buf, len, 
-                 &bytesread, file->pOverlapped)) {
+    if (file->socket && !file->pOverlapped) {
+        WSABUF wsaData;
+        DWORD flags = 0;
+
+        wsaData.buf = (char*) buf;
+        wsaData.len = (u_long)len;
+        if (WSARecv((SOCKET)file->filehand, &wsaData, 1, &bytesread,
+                    &flags, NULL, NULL) == SOCKET_ERROR) {
+            rv = apr_get_netos_error();
+            bytesread = 0;
+        }
+        else {
+            rv = APR_SUCCESS;
+        }
+    }
+    else if (ReadFile(file->filehand, buf, len, 
+                      &bytesread, file->pOverlapped)) {
         rv = APR_SUCCESS;
     }
     else {
@@ -139,7 +157,7 @@ static apr_status_t read_with_timeout(apr_file_t *file, void *buf, apr_size_t le
     if (rv == APR_SUCCESS && bytesread == 0)
         rv = APR_EOF;
     
-    if (rv == APR_SUCCESS && file->pOverlapped && !file->pipe) {
+    if (rv == APR_SUCCESS && file->pOverlapped && !pipe_or_socket) {
         file->filePtr += bytesread;
     }
     *nbytes = bytesread;
@@ -385,7 +403,7 @@ static apr_status_t write_buffered(apr_file_t *thefile, const char *buf,
 APR_DECLARE(apr_status_t) apr_file_write(apr_file_t *thefile, const void *buf, apr_size_t *nbytes)
 {
     apr_status_t rv;
-    DWORD bwrote;
+    DWORD bwrote = 0;
 
     /* If the file is open for xthread support, allocate and
      * initialize the overlapped and io completion event (hEvent). 
@@ -409,9 +427,27 @@ APR_DECLARE(apr_status_t) apr_file_write(apr_file_t *thefile, const void *buf, a
         if (thefile->flags & APR_FOPEN_XTHREAD) {
             apr_thread_mutex_unlock(thefile->mutex);
         }
-        return rv;
-    } else {
-        if (thefile->pipe) {
+    }
+    else if (thefile->socket && !thefile->pOverlapped) {
+        WSABUF wsaData;
+        DWORD flags = 0;
+
+        wsaData.buf = (char*) buf;
+        wsaData.len = (u_long)*nbytes;
+        if (WSASend((SOCKET)file->filehand, &wsaData, 1, &bwrote,
+                    flags, NULL, NULL) == SOCKET_ERROR) {
+            rv = apr_get_netos_error();
+            bwrote = 0;
+        }
+        else {
+            rv = APR_SUCCESS;
+        }
+        *nbytes = bwrote;
+    }
+    else {
+        int pipe_or_socket = (thefile->pipe || thefile->socket);
+
+        if (pipe_or_socket) {
             rv = WriteFile(thefile->filehand, buf, (DWORD)*nbytes, &bwrote,
                            thefile->pOverlapped);
         }
@@ -545,7 +581,7 @@ APR_DECLARE(apr_status_t) apr_file_write(apr_file_t *thefile, const void *buf, a
                 }
             }
         }
-        if (rv == APR_SUCCESS && thefile->pOverlapped && !thefile->pipe) {
+        if (rv == APR_SUCCESS && thefile->pOverlapped && !pipe_or_socket) {
             thefile->filePtr += *nbytes;
         }
     }