QProcess/Win: use asynchronous I/O for reading stdout and stderr

This saves us from using the 100 ms polling timer to read process output.

Task-number: QTBUG-23012
Change-Id: I3f9398d7a4d30826d2d89ac04bd2fd031500ff57
Reviewed-by: Oswald Buddenhagen <oswald.buddenhagen@nokia.com>

src/corelib/io/qprocess.cpp

@@ -761,8 +761,6 @@ QProcessPrivate::QProcessPrivate()
     exitStatus = QProcess::NormalExit;
     startupSocketNotifier = 0;
     deathNotifier = 0;
-    notifier = 0;
-    pipeWriter = 0;
     childStartedPipe[0] = INVALID_Q_PIPE;
     childStartedPipe[1] = INVALID_Q_PIPE;
     deathPipe[0] = INVALID_Q_PIPE;
@@ -773,6 +771,9 @@ QProcessPrivate::QProcessPrivate()
     emittedReadyRead = false;
     emittedBytesWritten = false;
 #ifdef Q_OS_WIN
+    notifier = 0;
+    stdoutReader = 0;
+    stderrReader = 0;
     pipeWriter = 0;
     processFinishedNotifier = 0;
 #endif // Q_OS_WIN
@@ -843,13 +844,15 @@ void QProcessPrivate::cleanup()
         qDeleteInEventHandler(deathNotifier);
         deathNotifier = 0;
     }
+#ifdef Q_OS_WIN
     if (notifier) {
         qDeleteInEventHandler(notifier);
         notifier = 0;
     }
-    destroyPipe(stdoutChannel.pipe);
+#endif
-    destroyPipe(stderrChannel.pipe);
+    destroyChannel(&stdoutChannel);
-    destroyPipe(stdinChannel.pipe);
+    destroyChannel(&stderrChannel);
+    destroyChannel(&stdinChannel);
     destroyPipe(childStartedPipe);
     destroyPipe(deathPipe);
 #ifdef Q_OS_UNIX
@@ -873,7 +876,7 @@ bool QProcessPrivate::_q_canReadStandardOutput()
     if (available == 0) {
         if (stdoutChannel.notifier)
             stdoutChannel.notifier->setEnabled(false);
-        destroyPipe(stdoutChannel.pipe);
+        destroyChannel(&stdoutChannel);
 #if defined QPROCESS_DEBUG
         qDebug("QProcessPrivate::canReadStandardOutput(), 0 bytes available");
 #endif
@@ -928,7 +931,7 @@ bool QProcessPrivate::_q_canReadStandardError()
     if (available == 0) {
         if (stderrChannel.notifier)
             stderrChannel.notifier->setEnabled(false);
-        destroyPipe(stderrChannel.pipe);
+        destroyChannel(&stderrChannel);
         return false;
     }
 
@@ -981,7 +984,7 @@ bool QProcessPrivate::_q_canWrite()
     qint64 written = writeToStdin(writeBuffer.readPointer(),
                                       writeBuffer.nextDataBlockSize());
     if (written < 0) {
-        destroyPipe(stdinChannel.pipe);
+        destroyChannel(&stdinChannel);
         processError = QProcess::WriteError;
         q->setErrorString(QProcess::tr("Error writing to process"));
         emit q->error(processError);
@@ -1125,7 +1128,7 @@ void QProcessPrivate::closeWriteChannel()
     // instead.
     flushPipeWriter();
 #endif
-    destroyPipe(stdinChannel.pipe);
+    destroyChannel(&stdinChannel);
 }
 
 /*!

src/corelib/io/qprocess_p.h

@@ -74,6 +74,7 @@ typedef int Q_PIPE;
 QT_BEGIN_NAMESPACE
 
 class QSocketNotifier;
+class QWindowsPipeReader;
 class QWindowsPipeWriter;
 class QWinEventNotifier;
 class QTimer;
@@ -291,14 +292,19 @@ public:
     Q_PIPE childStartedPipe[2];
     Q_PIPE deathPipe[2];
     void destroyPipe(Q_PIPE pipe[2]);
+    void destroyChannel(Channel *channel);
 
     QSocketNotifier *startupSocketNotifier;
     QSocketNotifier *deathNotifier;
 
+#ifdef Q_OS_WIN
     // the wonderful windows notifier
     QTimer *notifier;
+    QWindowsPipeReader *stdoutReader;
+    QWindowsPipeReader *stderrReader;
     QWindowsPipeWriter *pipeWriter;
     QWinEventNotifier *processFinishedNotifier;
+#endif
 
     void startProcess();
 #if defined(Q_OS_UNIX) && !defined(Q_OS_SYMBIAN)

src/corelib/io/qprocess_unix.cpp

@@ -357,6 +357,11 @@ void QProcessPrivate::destroyPipe(int *pipe)
     }
 }
 
+void QProcessPrivate::destroyChannel(Channel *channel)
+{
+    destroyPipe(channel->pipe);
+}
+
 /*
     Create the pipes to a QProcessPrivate::Channel.
 

src/corelib/io/qprocess_win.cpp

@@ -41,10 +41,12 @@
 
 #include "qprocess.h"
 #include "qprocess_p.h"
+#include "qwindowspipereader_p.h"
 #include "qwindowspipewriter_p.h"
 
 #include <qdatetime.h>
 #include <qdir.h>
+#include <qelapsedtimer.h>
 #include <qfileinfo.h>
 #include <qregexp.h>
 #include <qtimer.h>
@@ -152,6 +154,26 @@ bool QProcessPrivate::createChannel(Channel &channel)
             qt_create_pipe(channel.pipe, isStdInChannel);
         else
             duplicateStdWriteChannel(channel.pipe, (&channel == &stdoutChannel) ? STD_OUTPUT_HANDLE : STD_ERROR_HANDLE);
+
+        QWindowsPipeReader *pipeReader = 0;
+        if (&channel == &stdoutChannel) {
+            if (!stdoutReader) {
+                stdoutReader = new QWindowsPipeReader(q);
+                q->connect(stdoutReader, SIGNAL(readyRead()), SLOT(_q_canReadStandardOutput()));
+            }
+            pipeReader = stdoutReader;
+        } else if (&channel == &stderrChannel) {
+            if (!stderrReader) {
+                stderrReader = new QWindowsPipeReader(q);
+                q->connect(stderrReader, SIGNAL(readyRead()), SLOT(_q_canReadStandardError()));
+            }
+            pipeReader = stderrReader;
+        }
+        if (pipeReader) {
+            pipeReader->setHandle(channel.pipe[0]);
+            pipeReader->startAsyncRead();
+        }
+
         return true;
     } else if (channel.type == Channel::Redirect) {
         // we're redirecting the channel to/from a file
@@ -263,11 +285,6 @@ bool QProcessPrivate::createChannel(Channel &channel)
 
 void QProcessPrivate::destroyPipe(Q_PIPE pipe[2])
 {
-    if (pipe[0] == stdinChannel.pipe[0] && pipe[1] == stdinChannel.pipe[1] && pipeWriter) {
-        delete pipeWriter;
-        pipeWriter = 0;
-    }
-
     if (pipe[0] != INVALID_Q_PIPE) {
         CloseHandle(pipe[0]);
         pipe[0] = INVALID_Q_PIPE;
@@ -278,6 +295,26 @@ void QProcessPrivate::destroyPipe(Q_PIPE pipe[2])
     }
 }
 
+void QProcessPrivate::destroyChannel(Channel *channel)
+{
+    if (channel == &stdinChannel) {
+        if (pipeWriter) {
+            delete pipeWriter;
+            pipeWriter = 0;
+        }
+    } else if (channel == &stdoutChannel) {
+        if (stdoutReader) {
+            stdoutReader->deleteLater();
+            stdoutReader = 0;
+        }
+    } else if (channel == &stderrChannel) {
+        if (stderrReader) {
+            stderrReader->deleteLater();
+            stderrReader = 0;
+        }
+    }
+    destroyPipe(channel->pipe);
+}
 
 static QString qt_create_commandline(const QString &program, const QStringList &arguments)
 {
@@ -502,8 +539,10 @@ qint64 QProcessPrivate::bytesAvailableFromStdout() const
     if (stdoutChannel.pipe[0] == INVALID_Q_PIPE)
         return 0;
 
-    DWORD bytesAvail = 0;
+    if (!stdoutReader)
-	PeekNamedPipe(stdoutChannel.pipe[0], 0, 0, 0, &bytesAvail, 0);
+        return 0;
+
+    DWORD bytesAvail = stdoutReader->bytesAvailable();
 #if defined QPROCESS_DEBUG
     qDebug("QProcessPrivate::bytesAvailableFromStdout() == %d", bytesAvail);
 #endif
@@ -515,8 +554,10 @@ qint64 QProcessPrivate::bytesAvailableFromStderr() const
     if (stderrChannel.pipe[0] == INVALID_Q_PIPE)
         return 0;
 
-    DWORD bytesAvail = 0;
+    if (!stderrReader)
-	PeekNamedPipe(stderrChannel.pipe[0], 0, 0, 0, &bytesAvail, 0);
+        return 0;
+
+    DWORD bytesAvail = stderrReader->bytesAvailable();
 #if defined QPROCESS_DEBUG
     qDebug("QProcessPrivate::bytesAvailableFromStderr() == %d", bytesAvail);
 #endif
@@ -525,22 +566,12 @@ qint64 QProcessPrivate::bytesAvailableFromStderr() const
 
 qint64 QProcessPrivate::readFromStdout(char *data, qint64 maxlen)
 {
-    DWORD read = qMin(maxlen, bytesAvailableFromStdout());
+    return stdoutReader ? stdoutReader->read(data, maxlen) : 0;
-    DWORD bytesRead = 0;
-
-    if (read > 0 && !ReadFile(stdoutChannel.pipe[0], data, read, &bytesRead, 0))
-        return -1;
-    return bytesRead;
 }
 
 qint64 QProcessPrivate::readFromStderr(char *data, qint64 maxlen)
 {
-    DWORD read = qMin(maxlen, bytesAvailableFromStderr());
+    return stderrReader ? stderrReader->read(data, maxlen) : 0;
-    DWORD bytesRead = 0;
-
-    if (read > 0 && !ReadFile(stderrChannel.pipe[0], data, read, &bytesRead, 0))
-        return -1;
-    return bytesRead;
 }
 
 
@@ -583,6 +614,30 @@ bool QProcessPrivate::waitForStarted(int)
     return false;
 }
 
+static bool drainOutputPipes(QProcessPrivate *d)
+{
+    if (!d->stdoutReader && !d->stderrReader)
+        return false;
+
+    bool readyReadEmitted = false;
+    forever {
+        bool readOperationActive = false;
+        if (d->stdoutReader) {
+            readyReadEmitted |= d->stdoutReader->waitForReadyRead(0);
+            readOperationActive = d->stdoutReader->isReadOperationActive();
+        }
+        if (d->stderrReader) {
+            readyReadEmitted |= d->stderrReader->waitForReadyRead(0);
+            readOperationActive |= d->stderrReader->isReadOperationActive();
+        }
+        if (!readOperationActive)
+            break;
+        Sleep(100);
+    }
+
+    return readyReadEmitted;
+}
+
 bool QProcessPrivate::waitForReadyRead(int msecs)
 {
     Q_Q(QProcess);
@@ -594,26 +649,18 @@ bool QProcessPrivate::waitForReadyRead(int msecs)
             return false;
         if (pipeWriter && pipeWriter->waitForWrite(0))
             timer.resetIncrements();
-        bool readyReadEmitted = false;
-        if (bytesAvailableFromStdout() != 0) {
-            readyReadEmitted = _q_canReadStandardOutput() ? true : readyReadEmitted;
-            timer.resetIncrements();
-        }
 
-        if (bytesAvailableFromStderr() != 0) {
+        if (processChannelMode != QProcess::ForwardedChannels
-            readyReadEmitted = _q_canReadStandardError() ? true : readyReadEmitted;
+                && ((stdoutReader && stdoutReader->waitForReadyRead(0))
-            timer.resetIncrements();
+                    || (stderrReader && stderrReader->waitForReadyRead(0))))
-        }
-
-        if (readyReadEmitted)
             return true;
 
         if (!pid)
             return false;
         if (WaitForSingleObject(pid->hProcess, 0) == WAIT_OBJECT_0) {
-            // find the return value if there is noew data to read
+            bool readyReadEmitted = drainOutputPipes(this);
             _q_processDied();
-            return false;
+            return readyReadEmitted;
         }
 
         Sleep(timer.nextSleepTime());
@@ -694,7 +741,6 @@ bool QProcessPrivate::waitForBytesWritten(int msecs)
     return false;
 }
 
-
 bool QProcessPrivate::waitForFinished(int msecs)
 {
     Q_Q(QProcess);
@@ -709,21 +755,18 @@ bool QProcessPrivate::waitForFinished(int msecs)
             return false;
         if (pipeWriter && pipeWriter->waitForWrite(0))
             timer.resetIncrements();
-
+        if (stdoutReader && stdoutReader->waitForReadyRead(0))
-        if (bytesAvailableFromStdout() != 0) {
-            _q_canReadStandardOutput();
             timer.resetIncrements();
-        }
+        if (stderrReader && stderrReader->waitForReadyRead(0))
-
-        if (bytesAvailableFromStderr() != 0) {
-            _q_canReadStandardError();
             timer.resetIncrements();
-        }
 
-        if (!pid)
+        if (!pid) {
+            drainOutputPipes(this);
             return true;
+        }
 
         if (WaitForSingleObject(pid->hProcess, timer.nextSleepTime()) == WAIT_OBJECT_0) {
+            drainOutputPipes(this);
             _q_processDied();
             return true;
         }
@@ -731,6 +774,7 @@ bool QProcessPrivate::waitForFinished(int msecs)
         if (timer.hasTimedOut())
             break;
     }
+
     processError = QProcess::Timedout;
     q->setErrorString(QProcess::tr("Process operation timed out"));
     return false;
@@ -790,12 +834,6 @@ void QProcessPrivate::_q_notified()
     if (!writeBuffer.isEmpty() && (!pipeWriter || pipeWriter->waitForWrite(0)))
         _q_canWrite();
 
-    if (bytesAvailableFromStdout())
-        _q_canReadStandardOutput();
-
-    if (bytesAvailableFromStderr())
-        _q_canReadStandardError();
-
     if (processState != QProcess::NotRunning)
         notifier->start(NOTIFYTIMEOUT);
 }

src/corelib/io/qprocess_wince.cpp

@@ -60,6 +60,11 @@ void QProcessPrivate::destroyPipe(Q_PIPE pipe[2])
     Q_UNUSED(pipe);
 }
 
+void QProcessPrivate::destroyChannel(Channel *channel)
+{
+    Q_UNUSED(channel);
+}
+
 static QString qt_create_commandline(const QString &program, const QStringList &arguments)
 {
     QString args;