Merge r1895106, r1895111, r1895175, r1895181, r1895465 from trunk:

Fix drain wakeup pipe issue when multiple threads call apr_pollset_wakeup/apr_pollcb_wakeup for the same pollset filling up drain pipe. Use atomics so that wakeup call is noop if some other thread allready done this


Follow up to r1895106: now we want blocking reads on unix too so revert r1894914.


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

If pipe writers (wakeup) put a single byte until it's consumed by the
reader (drain), it's enough to use an atomic cas for the writers (still) and
an atomic (re)set for the reader (no more cas here).

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

Since synchronous non-blocking read is not a thing for Windows' Readfile(), add
a ->socket flag to this arch's 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().


Use enum instead multiple booleans


Fix remaining change when apr_filetype_e was added


Submitted by: mturk, ylavic, ylavic, mturk, mturk


git-svn-id: https://svn.apache.org/repos/asf/apr/apr/branches/1.7.x@1896510 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 78 insertions, 49 deletions

diff --git a/file_io/win32/pipe.c b/file_io/win32/pipe.c
index dd8ef1352..7dcbfb00e 100644
--- a/file_io/win32/pipe.c
+++ b/file_io/win32/pipe.c
@@ -43,7 +43,7 @@ APR_DECLARE(apr_status_t) apr_file_pipe_timeout_set(apr_file_t *thepipe,
         thepipe->timeout = timeout;
         return APR_SUCCESS;
     }
-    if (!thepipe->pipe) {
+    if (thepipe->ftype != APR_FILETYPE_PIPE) {
         return APR_ENOTIMPL;
     }
     if (timeout && !(thepipe->pOverlapped)) {
@@ -108,7 +108,7 @@ APR_DECLARE(apr_status_t) apr_file_pipe_create_pools(apr_file_t **in,
     (*in) = (apr_file_t *)apr_pcalloc(pool_in, sizeof(apr_file_t));
     (*in)->pool = pool_in;
     (*in)->fname = NULL;
-    (*in)->pipe = 1;
+    (*in)->ftype = APR_FILETYPE_PIPE;
     (*in)->timeout = -1;
     (*in)->ungetchar = -1;
     (*in)->eof_hit = 0;
@@ -123,7 +123,7 @@ APR_DECLARE(apr_status_t) apr_file_pipe_create_pools(apr_file_t **in,
     (*out) = (apr_file_t *)apr_pcalloc(pool_out, sizeof(apr_file_t));
     (*out)->pool = pool_out;
     (*out)->fname = NULL;
-    (*out)->pipe = 1;
+    (*out)->ftype = APR_FILETYPE_PIPE;
     (*out)->timeout = -1;
     (*out)->ungetchar = -1;
     (*out)->eof_hit = 0;
@@ -250,7 +250,7 @@ APR_DECLARE(apr_status_t) apr_os_pipe_put_ex(apr_file_t **file,
 {
     (*file) = apr_pcalloc(pool, sizeof(apr_file_t));
     (*file)->pool = pool;
-    (*file)->pipe = 1;
+    (*file)->ftype = APR_FILETYPE_PIPE;
     (*file)->timeout = -1;
     (*file)->ungetchar = -1;
     (*file)->filehand = *thefile;
@@ -364,32 +364,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 */
@@ -398,6 +394,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)
@@ -405,7 +402,6 @@ cleanup:
 
     *rd = INVALID_SOCKET;
     *wr = INVALID_SOCKET;
-    closesocket(ls);
     return rv;
 }
 
@@ -434,21 +430,21 @@ apr_status_t apr_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)->ftype = APR_FILETYPE_SOCKET;
+    (*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)->ftype = APR_FILETYPE_SOCKET;
     (*out)->timeout = -1;
     (*out)->ungetchar = -1;
     (*out)->eof_hit = 0;
@@ -456,7 +452,7 @@ apr_status_t apr_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,
@@ -470,7 +466,7 @@ apr_status_t apr_file_socket_pipe_create(apr_file_t **in,
 apr_status_t apr_file_socket_pipe_close(apr_file_t *file)
 {
     apr_status_t stat;
-    if (!file->pipe)
+    if (file->ftype != APR_FILETYPE_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 701bec75b..fa4a23fbe 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.
@@ -40,7 +42,7 @@ static apr_status_t read_with_timeout(apr_file_t *file, void *buf, apr_size_t le
         /* Peek at the pipe. If there is no data available, return APR_EAGAIN.
          * If data is available, go ahead and read it.
          */
-        if (file->pipe) {
+        if (file->ftype == APR_FILETYPE_PIPE) {
             DWORD bytes;
             if (!PeekNamedPipe(file->filehand, NULL, 0, NULL, &bytes, NULL)) {
                 rv = apr_get_os_error();
@@ -68,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 && file->ftype == APR_FILETYPE_FILE) {
         file->pOverlapped->Offset     = (DWORD)file->filePtr;
         file->pOverlapped->OffsetHigh = (DWORD)(file->filePtr >> 32);
     }
 
-    if (ReadFile(file->filehand, buf, len, 
-                 &bytesread, file->pOverlapped)) {
+    if (file->ftype == APR_FILETYPE_SOCKET) {
+        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 {
@@ -133,7 +150,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 && file->ftype == APR_FILETYPE_FILE) {
         file->filePtr += bytesread;
     }
     *nbytes = bytesread;
@@ -246,7 +263,7 @@ APR_DECLARE(apr_status_t) apr_file_read(apr_file_t *thefile, void *buf, apr_size
 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). 
@@ -298,9 +315,29 @@ 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->ftype == APR_FILETYPE_SOCKET) {
+        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 {
+        if (thefile->ftype != APR_FILETYPE_FILE) {
+            rv = WriteFile(thefile->filehand, buf, (DWORD)*nbytes, &bwrote,
+                           thefile->pOverlapped);
+        }
+        else {
             apr_off_t offset = 0;
             apr_status_t rc;
             if (thefile->append) {
@@ -332,10 +369,6 @@ APR_DECLARE(apr_status_t) apr_file_write(apr_file_t *thefile, const void *buf, a
                 apr_thread_mutex_unlock(thefile->mutex);
             }
         }
-        else {
-            rv = WriteFile(thefile->filehand, buf, (DWORD)*nbytes, &bwrote,
-                           thefile->pOverlapped);
-        }
         if (rv) {
             *nbytes = bwrote;
             rv = APR_SUCCESS;
@@ -382,7 +415,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 && thefile->ftype == APR_FILETYPE_FILE) {
             thefile->filePtr += *nbytes;
         }
     }