QLocalSocket/Win: implement duplex communication in blocking mode

[ChangeLog][QtNetwork][QLocalSocket] The waitFor*() functions on
Windows now support duplex operation, as they already did on Unix.

As a side effect, this restores the behavior that a single call to
waitForReadyRead() won't emit both readyRead() and disconnected(),
which also matches Unix behavior. The groundwork for that misbehavior
was laid by incorrect refactoring in d1a671b69 already, but at this
point it was harmless, as the pipe couldn't be newly closed after a
successful read. That changed with f265c87e0, which made the queuing
of signals async.

Change-Id: I1eb80e8f147bb58825143e0fe1e4300c59ae0fbb
Reviewed-by: Oswald Buddenhagen <oswald.buddenhagen@gmx.de>

2 files changed, 117 insertions, 23 deletions

diff --git a/src/network/socket/qlocalsocket_p.h b/src/network/socket/qlocalsocket_p.h
index 34dc37870c..1a429cdefa 100644
--- a/src/network/socket/qlocalsocket_p.h
+++ b/src/network/socket/qlocalsocket_p.h
@@ -133,6 +133,7 @@ public:
 #elif defined(Q_OS_WIN)
     ~QLocalSocketPrivate();
     void destroyPipeHandles();
+    qint64 pipeWriterBytesToWrite() const;
     void _q_canRead();
     void _q_bytesWritten(qint64 bytes);
     void _q_canWrite();
diff --git a/src/network/socket/qlocalsocket_win.cpp b/src/network/socket/qlocalsocket_win.cpp
index 66eed86501..37ca09e5d3 100644
--- a/src/network/socket/qlocalsocket_win.cpp
+++ b/src/network/socket/qlocalsocket_win.cpp
@@ -39,9 +39,61 @@
 
 #include "qlocalsocket_p.h"
 #include <qscopedvaluerollback.h>
+#include <qdeadlinetimer.h>
 
 QT_BEGIN_NAMESPACE
 
+namespace {
+struct QSocketPoller
+{
+    QSocketPoller(const QLocalSocketPrivate &socket);
+
+    bool poll(const QDeadlineTimer &deadline);
+
+    enum { maxHandles = 2 };
+    HANDLE handles[maxHandles];
+    DWORD handleCount = 0;
+    bool waitForClose = false;
+};
+
+QSocketPoller::QSocketPoller(const QLocalSocketPrivate &socket)
+{
+    if (socket.pipeWriter)
+        handles[handleCount++] = socket.pipeWriter->syncEvent();
+    if (socket.pipeReader->isReadOperationActive())
+        handles[handleCount++] = socket.pipeReader->syncEvent();
+    else
+        waitForClose = true;
+}
+
+/*!
+    Waits until new data is available for reading or write operation
+    completes. Returns \c true, if we need to check pipe workers;
+    otherwise it returns \c false (if an error occurred or the operation
+    timed out).
+
+    \note If the read operation is inactive, it succeeds after
+    a short wait, allowing the caller to check the state of the socket.
+ */
+bool QSocketPoller::poll(const QDeadlineTimer &deadline)
+{
+    const qint64 sleepTime = 10;
+    QDeadlineTimer timer(waitForClose ? qMin(deadline.remainingTime(), sleepTime)
+                                      : deadline.remainingTime());
+    DWORD waitRet;
+
+    do {
+        waitRet = WaitForMultipleObjectsEx(handleCount, handles, FALSE,
+                                           timer.remainingTime(), TRUE);
+    } while (waitRet == WAIT_IO_COMPLETION);
+
+    if (waitRet == WAIT_TIMEOUT)
+        return !deadline.hasExpired();
+
+    return waitRet - WAIT_OBJECT_0 < handleCount;
+}
+} // anonymous namespace
+
 void QLocalSocketPrivate::init()
 {
     Q_Q(QLocalSocket);
@@ -288,7 +340,7 @@ qint64 QLocalSocket::bytesAvailable() const
 qint64 QLocalSocket::bytesToWrite() const
 {
     Q_D(const QLocalSocket);
-    return d->writeBuffer.size() + (d->pipeWriter ? d->pipeWriter->bytesToWrite() : 0);
+    return d->writeBuffer.size() + d->pipeWriterBytesToWrite();
 }
 
 bool QLocalSocket::canReadLine() const
@@ -370,6 +422,11 @@ bool QLocalSocket::setSocketDescriptor(qintptr socketDescriptor,
     return true;
 }
 
+qint64 QLocalSocketPrivate::pipeWriterBytesToWrite() const
+{
+    return pipeWriter ? pipeWriter->bytesToWrite() : qint64(0);
+}
+
 void QLocalSocketPrivate::_q_bytesWritten(qint64 bytes)
 {
     Q_Q(QLocalSocket);
@@ -384,7 +441,7 @@ void QLocalSocketPrivate::_q_canWrite()
 {
     Q_Q(QLocalSocket);
     if (writeBuffer.isEmpty()) {
-        if (state == QLocalSocket::ClosingState)
+        if (state == QLocalSocket::ClosingState && pipeWriterBytesToWrite() == 0)
             q->close();
     } else {
         Q_ASSERT(pipeWriter);
@@ -428,11 +485,27 @@ bool QLocalSocket::waitForDisconnected(int msecs)
         qWarning("QLocalSocket::waitForDisconnected isn't supported for write only pipes.");
         return false;
     }
-    if (d->pipeReader->waitForPipeClosed(msecs)) {
-        d->_q_pipeClosed();
-        return true;
+
+    QDeadlineTimer deadline(msecs);
+    while (!d->pipeReader->isPipeClosed()) {
+        d->_q_canWrite();
+
+        QSocketPoller poller(*d);
+        if (!poller.poll(deadline))
+            return false;
+
+        if (d->pipeWriter)
+            d->pipeWriter->checkForWrite();
+
+        // When the read buffer is full, the read sequence is not running,
+        // so we need to peek the pipe to detect disconnection.
+        if (poller.waitForClose)
+            d->pipeReader->checkPipeState();
+
+        d->pipeReader->checkForReadyRead();
     }
-    return false;
+    d->_q_pipeClosed();
+    return true;
 }
 
 bool QLocalSocket::isValid() const
@@ -448,32 +521,52 @@ bool QLocalSocket::waitForReadyRead(int msecs)
     if (d->state != QLocalSocket::ConnectedState)
         return false;
 
-    // We already know that the pipe is gone, but did not enter the event loop yet.
-    if (d->pipeReader->isPipeClosed()) {
-        d->_q_pipeClosed();
-        return false;
-    }
+    QDeadlineTimer deadline(msecs);
+    while (!d->pipeReader->isPipeClosed()) {
+        d->_q_canWrite();
 
-    bool result = d->pipeReader->waitForReadyRead(msecs);
+        QSocketPoller poller(*d);
+        if (poller.waitForClose || !poller.poll(deadline))
+            return false;
 
-    // We just noticed that the pipe is gone.
-    if (d->pipeReader->isPipeClosed())
-        d->_q_pipeClosed();
+        if (d->pipeWriter)
+            d->pipeWriter->checkForWrite();
 
-    return result;
+        if (d->pipeReader->checkForReadyRead())
+            return true;
+    }
+    d->_q_pipeClosed();
+    return false;
 }
 
 bool QLocalSocket::waitForBytesWritten(int msecs)
 {
-    Q_D(const QLocalSocket);
-    if (!d->pipeWriter)
+    Q_D(QLocalSocket);
+
+    if (d->state == QLocalSocket::UnconnectedState)
         return false;
 
-    // Wait for the pipe writer to acknowledge that it has
-    // written. This will succeed if either the pipe writer has
-    // already written the data, or if it manages to write data
-    // within the given timeout.
-    return d->pipeWriter->waitForWrite(msecs);
+    QDeadlineTimer deadline(msecs);
+    while (!d->pipeReader->isPipeClosed()) {
+        if (bytesToWrite() == 0)
+            return false;
+        d->_q_canWrite();
+
+        QSocketPoller poller(*d);
+        if (!poller.poll(deadline))
+            return false;
+
+        Q_ASSERT(d->pipeWriter);
+        if (d->pipeWriter->checkForWrite())
+            return true;
+
+        if (poller.waitForClose)
+            d->pipeReader->checkPipeState();
+
+        d->pipeReader->checkForReadyRead();
+    }
+    d->_q_pipeClosed();
+    return false;
 }
 
 QT_END_NAMESPACE