Changeset 134 for trunk

Timestamp:

Oct 9, 2006, 12:14:18 AM (19 years ago)

Author:

dmik

Message:

Kernel: QProcess: Significantly improved the speed of reading data from the child process using readyReadStd[out|err]() signals (it's now approximately 380 times faster on my system for a 28 MB file).

Location:

trunk

Files:

• include/qprocess.h (modified) (1 diff)
• src/kernel/qprocess_pm.cpp (modified) (26 diffs)

trunk/include/qprocess.h

@@ -166 +166 @@
     friend class QProc;
 #endif
+#if defined(Q_OS_OS2)
+    friend class QProcessMonitor;
+#endif
 
 #if defined(Q_DISABLE_COPY) // Disabled copy constructor and operator=

trunk/src/kernel/qprocess_pm.cpp

@@ -45 +45 @@
 #include "qregexp.h"
 #include "qdir.h"
+#include "qthread.h"
+#include "qintdict.h"
+#include "qmutex.h"
 #include "private/qinternal_p.h"
 #include "qt_os2.h"
@@ -52 +55 @@
 //#define QT_QPROCESS_DEBUG
 
-const HFILE HF_STDIN = 0;
-const HFILE HF_STDOUT = 1;
-const HFILE HF_STDERR = 2;
-const HFILE HF_ANY = 0xFFFFFFFF; 
-
-const PID PID_NULL = ~1; 
-
-const ULONG PIPE_SIZE = 1024;
-
-// just fake constants, for convenience
-const int FD_PIPE_STDOUT = 1;
-const int FD_PIPE_STDERR = 2;
+#define HF_STDIN        HFILE( 0 )
+#define HF_STDOUT       HFILE( 1 )
+#define HF_STDERR       HFILE( 2 )
+#define HF_NULL         HFILE( ~0 ) 
+
+#define HP_NULL         HPIPE( ~0 ) 
+#define KEY_NULL        USHORT( ~0 )
+
+#define PID_NULL        PID( ~0 ) 
+
+enum
+{
+    PIPE_SIZE_STDIN = 65520, // max
+    PIPE_SIZE_STDOUT = 65520, // max
+    PIPE_SIZE_STDERR = 4096,
+
+    POLL_TIMER = 100,
+
+    // new pipe data notification
+    WM_U_PIPE_RDATA = WM_USER + 0,
+    WM_U_PIPE_CLOSE = WM_USER + 1,
+};
 
 #if defined(QT_QPROCESS_DEBUG)
 #include <stdarg.h>
+static HFILE StdErrHandle = HF_NULL;
+QtMsgHandler OldMsgHandler = NULL;
+static void StdErrMsgHandler( QtMsgType type, const char *msg )
+{
+    if ( OldMsgHandler == NULL ) {
+       size_t len = strlen( msg );
+       ULONG written = 0;
+       DosWrite( StdErrHandle, msg, len, &written );
+       const char *EOL = "\n\r";
+       DosWrite( StdErrHandle, EOL, 2, &written );
+    } else {
+        OldMsgHandler( type, msg );
+    }
+}
+#define InstallQtMsgHandler() \
+    do { \
+        DosDupHandle( HF_STDERR, &StdErrHandle ); \
+        qInstallMsgHandler( StdErrMsgHandler ); \
+    } while (0)
+#define UninstallQtMsgHandler() \
+    do { \
+        qInstallMsgHandler( OldMsgHandler ); \
+        DosClose( StdErrHandle ); \
+    } while (0)
+#else
+#define InstallQtMsgHandler() do {} while (0)
+#define UninstallQtMsgHandler() do {} while (0)
 #endif
 
@@ -77 +117 @@
 {
 public:
+    struct Pipe
+    {
+        Pipe () : pipe( HP_NULL ), key( KEY_NULL ), pending( 0 )
+                , closed( false ) {}
+        HPIPE pipe;
+        USHORT key;
+        QMembuf buf;
+        // note: when QProcess is watched by QProcessMonitor, the below fields
+        // must be accessed from under the QProcessMonitor lock
+        uint pending;
+        bool closed : 1;
+    };
+
     QProcessPrivate( QProcess *proc )
     {
         stdinBufRead = 0;
-        pipeStdin = HF_ANY;
-        pipeStdout = HF_ANY;
-        pipeStderr = HF_ANY;
+        pipeStdin = HP_NULL;
+        pid = PID_NULL;
         exitValuesCalculated = FALSE;
 
         lookup = new QTimer( proc );
-        qApp->connect( lookup, SIGNAL(timeout()),
-                proc, SLOT(timeout()) );
-
-        pid = PID_NULL;
+        qApp->connect( lookup, SIGNAL(timeout()), proc, SLOT(timeout()) );
     }
 
@@ -102 +151 @@
             delete stdinBuf.dequeue();
         }
+        stdinBufRead = 0;
         closeHandles();
-        stdinBufRead = 0;
-        pipeStdin = HF_ANY;
-        pipeStdout = HF_ANY;
-        pipeStderr = HF_ANY;
+        stdout.buf.clear();
+        stderr.buf.clear();
+        pid = PID_NULL;
         exitValuesCalculated = FALSE;
-
-        pid = PID_NULL;
-    }
-
+    }
+
+    Pipe *findPipe( USHORT key )
+    {
+        if ( stdout.key == key ) return &stdout;
+        if ( stderr.key == key ) return &stderr;
+        return NULL;
+    }
+    
+    void closePipe( Pipe *pipe )
+    {
+        if ( pipe->pipe != HP_NULL ) {
+            Q_ASSERT( pipe->key == KEY_NULL );
+            DosDisConnectNPipe( pipe->pipe );
+            DosClose( pipe->pipe );
+            pipe->pipe = HP_NULL;
+            pipe->pending = 0;
+            pipe->closed = FALSE;
+        }
+    }
+    
+    bool readPipe( Pipe *pipe );
+    
     void closeHandles()
     {
-        if( pipeStdin != HF_ANY ) {
+        if( pipeStdin != HP_NULL ) {
             DosDisConnectNPipe( pipeStdin );
             DosClose( pipeStdin );
-            pipeStdin = HF_ANY;
+            pipeStdin = HP_NULL;
         }
-        if( pipeStdout != HF_ANY ) {
-            // check if STDERR was redirected to STDOUT
-            if ( pipeStderr == pipeStdout )
-                pipeStderr = HF_ANY;
-            DosDisConnectNPipe( pipeStdout );
-            DosClose( pipeStdout );
-            pipeStdout = HF_ANY;
-        }
-        if( pipeStderr != HF_ANY ) {
-            DosDisConnectNPipe( pipeStderr );
-            DosClose( pipeStderr );
-            pipeStderr = HF_ANY;
-        }
-    }
-
-    QMembuf bufStdout;
-    QMembuf bufStderr;
+        closePipe( &stdout );
+        closePipe( &stderr );
+    }
 
     QPtrQueue <QByteArray> stdinBuf;
+    uint stdinBufRead;
 
     HPIPE pipeStdin;
-    HPIPE pipeStdout;
-    HPIPE pipeStderr;
+    
+    Pipe stdout;
+    Pipe stderr;
 
     PID pid;
@@ -147 +203 @@
     QTimer *lookup;
 
-    uint stdinBufRead;
-
-    bool exitValuesCalculated;
+    bool exitValuesCalculated : 1;
 };
 
+class QProcessMonitor : public QPMObjectWindow
+{
+public:
+    class Thread : public QThread
+    {
+    public:
+        Thread( QProcessMonitor *m ) : mon( m ) {}
+        void run() { mon->monitor(); }
+        QProcessMonitor *mon;
+    };
+
+    QProcessMonitor();
+    ~QProcessMonitor();
+
+    QMutex &mlock() { return lock; }
+    
+    bool isOk() const { return pipeSem != 0 && thread != NULL; }
+
+    bool addProcess( QProcess *proc );
+    void removeProcess( QProcessPrivate *d,
+                        QProcessPrivate::Pipe *pipe = NULL );
+
+    void monitor();
+    MRESULT message( ULONG msg, MPARAM mp1, MPARAM mp2 );
+    
+private:
+
+    struct PipeStates
+    {
+        PipeStates() {
+            size = 4;
+            arr = new PIPESEMSTATE[ size ]; 
+        }
+        ~PipeStates() {
+            delete[] arr;
+        }
+        void ensure( size_t sz ) {
+            // best size for sz pipes is sz * 2 (to be able to store both
+            // NPSS_RDATA & NPSS_CLOSE for every pipe) + one for NPSS_EOI
+            size_t newSize = sz * 2 + 1; 
+            newSize = ((newSize + 5 - 1) / 5) * 5;
+            if ( newSize != size ) {
+                delete[] arr;
+                size = newSize;
+                arr = new PIPESEMSTATE[ size ];
+            }
+        }
+        size_t dataSize() { return size * sizeof(PIPESEMSTATE); }
+        PIPESEMSTATE *data() { return arr; }
+        PIPESEMSTATE &operator[]( size_t i ) {
+            Q_ASSERT( i < size );
+            return arr[i];
+        }
+    private:
+        PIPESEMSTATE *arr;
+        size_t size;
+    };
+
+    bool addPipe( QProcess *proc, QProcessPrivate::Pipe *pipe );
+    void removePipe( QProcessPrivate::Pipe *pipe );
+
+    HEV pipeSem;
+    Thread *thread;
+    bool threadRunning;
+    QIntDict<QProcess> pipeKeys;
+    PipeStates pipeStates;
+    QMutex lock;
+};
+
+QProcessMonitor::QProcessMonitor()
+    : pipeSem( 0 ), thread( NULL ), threadRunning( FALSE )
+{
+    APIRET rc = DosCreateEventSem( NULL, &pipeSem, DC_SEM_SHARED, 0 );
+    if ( rc != NO_ERROR ) {
+#if defined(QT_CHECK_STATE) || defined(QT_QPROCESS_DEBUG)
+        qSystemWarning( "Failed to create a semaphore!", rc );
+#endif
+        return;
+    }
+    
+    thread = new Thread( this );
+    Q_ASSERT( thread != NULL );
+}
+
+QProcessMonitor::~QProcessMonitor()
+{
+    if ( thread ) {
+        lock.lock();
+        if ( threadRunning ) {
+            threadRunning = FALSE;
+            DosPostEventSem( pipeSem );
+        }
+        lock.unlock();
+        thread->wait();
+        delete thread;
+    }
+    
+    if ( pipeSem )
+        DosCloseEventSem( pipeSem );
+}
+
+bool QProcessMonitor::addPipe( QProcess *proc, QProcessPrivate::Pipe *pipe )
+{
+    // Note: we use HPIPE as pipe keys in DosSetNPipeSem. However, HPIPE
+    // is 32-bit (ulong), while usKey in PIPESEMSTATE is 16-bit (USHORT)
+    // We unreasonably assume that the system will never assign HPIPE >= 0xFFFF,
+    // and just cast HPIPE to USHORT. There is an assertion checking this
+    // condition, so this method should simply return FALSE once our assumption
+    // breaks.
+
+    Q_ASSERT( pipe->pipe < HPIPE( KEY_NULL ) );
+    if ( pipe->pipe >= HPIPE( KEY_NULL ) )
+        return FALSE;
+
+    Q_ASSERT( pipe->pipe != HP_NULL && pipe->key == KEY_NULL );
+    
+    pipe->key = USHORT( pipe->pipe );
+    APIRET rc = DosSetNPipeSem( pipe->pipe, (HSEM) pipeSem, pipe->key );
+    if ( rc != NO_ERROR ) {
+#if defined(QT_CHECK_STATE) || defined(QT_QPROCESS_DEBUG)
+        qSystemWarning( "Failed to set the pipe semaphore!", rc );
+#endif        
+        return FALSE;
+    }
+
+    pipeKeys.insert( pipe->key, proc );
+    return TRUE;
+}
+
+void QProcessMonitor::removePipe( QProcessPrivate::Pipe *pipe )
+{
+    Q_ASSERT( pipe->pipe != HP_NULL && pipe->key != KEY_NULL );
+
+    /// @todo (r=dmik) is this really necessary to detach pipeSem?
+    DosSetNPipeSem( pipe->pipe, 0, 0 );
+    bool ok = pipeKeys.remove( pipe->key );
+    pipe->key = KEY_NULL;
+    Q_ASSERT( ok );
+    Q_UNUSED( ok );
+}
+
+bool QProcessMonitor::addProcess( QProcess *proc )
+{
+#if defined(QT_QPROCESS_DEBUG)
+    qDebug( "QProcessMonitor::addProcess(): proc=%p d=%p", proc, proc->d );
+#endif
+
+    QProcessPrivate *d = proc->d; 
+    
+    // check if we need to monitor this process before entering the lock
+    if ( d->stdout.pipe == HP_NULL && d->stderr.pipe == HP_NULL )
+        return TRUE;
+    
+    uint newPipes = 0;
+
+    QMutexLocker locker( &lock ); 
+    
+    if ( d->stdout.pipe != HP_NULL ) {
+        if ( !addPipe( proc, &d->stdout ) )
+            return FALSE;
+        ++ newPipes;
+    }
+
+    if ( d->stderr.pipe != HP_NULL ) {
+        if ( !addPipe( proc, &d->stderr ) ) {
+            removePipe( &d->stderr );
+            return FALSE;
+        }
+        ++ newPipes;
+    }
+    
+    Q_ASSERT( newPipes > 0 );
+
+    pipeStates.ensure( pipeKeys.count() );
+    
+    // start the monitor thread if necessary
+    if ( pipeKeys.count() == newPipes ) {
+        threadRunning = TRUE;
+        thread->start();
+    }
+    
+#if defined(QT_QPROCESS_DEBUG)
+    qDebug( "QProcessMonitor::addProcess(): added %d pipes", newPipes );
+#endif        
+    return TRUE;
+}
+
+void QProcessMonitor::removeProcess( QProcessPrivate *d,
+                                     QProcessPrivate::Pipe *pipe /* = NULL */)
+{
+#if defined(QT_QPROCESS_DEBUG)
+    qDebug( "QProcessMonitor::removeProcess(): d=%p pipe=%p key=%04hX",
+            d, pipe, pipe ? pipe->key : KEY_NULL );
+#endif
+    
+    // check if we monitor this process before entering the lock
+    if ( d->stdout.pipe == HP_NULL && d->stderr.pipe == HP_NULL )
+        return;
+
+    lock.lock(); 
+
+    if ( pipeKeys.count() == 0 ) {
+        // Nothing to do. This is an outdated general removeProcess (d, NULL)
+        // call from the QProcess destructor or from isRunning(). Just return.
+        lock.unlock();
+        return;
+    }
+    
+    if ( pipe ) {
+        removePipe( pipe );
+    } else {
+        if ( d->stdout.pipe != HP_NULL && d->stdout.key != KEY_NULL )
+            removePipe( &d->stdout );
+        if ( d->stderr.pipe != HP_NULL && d->stderr.key != KEY_NULL )
+            removePipe( &d->stderr );
+    }
+
+    pipeStates.ensure( pipeKeys.count() );
+    
+    // stop the monitor thread if no more necessary
+    if ( pipeKeys.count() == 0 ) {
+        Q_ASSERT( threadRunning );
+        if ( threadRunning ) {
+            threadRunning = FALSE;
+            DosPostEventSem( pipeSem );
+        }
+        lock.unlock();
+        thread->wait();
+    } else {
+        lock.unlock();
+    }
+}
+
+/** Monitor thread function */
+void QProcessMonitor::monitor()
+{
+#if defined(QT_QPROCESS_DEBUG)
+    qDebug( "QProcessMonitor::monitor() ENTER" );
+#endif
+
+    lock.lock();
+    while( 1 ) {
+        lock.unlock();
+        APIRET rc = DosWaitEventSem( pipeSem, SEM_INDEFINITE_WAIT );
+        lock.lock();
+        if ( rc != NO_ERROR ) {
+#if defined(QT_CHECK_STATE) || defined(QT_QPROCESS_DEBUG)
+            qSystemWarning( "Failed to wait on event semaphore!", rc );
+#endif            
+            break;
+        }
+
+        ULONG posts = 0;
+        DosResetEventSem( pipeSem, &posts );
+#if defined(QT_QPROCESS_DEBUG)        
+        qDebug( "QProcessMonitor::monitor(): got semaphore (posts=%ld)", posts );
+#endif
+
+        if ( !threadRunning )
+            break;
+
+        rc = DosQueryNPipeSemState( (HSEM) pipeSem, pipeStates.data(),
+                                    pipeStates.dataSize() );
+        if ( rc != NO_ERROR ) {
+#if defined(QT_CHECK_STATE) || defined(QT_QPROCESS_DEBUG)
+            qSystemWarning( "Failed to query pipe semaphore state!", rc );
+#endif
+            continue;
+        }
+
+        // In the returned information array, CLOSE and READ records for the
+        // same pipe key may be mixed. We need CLOSE messages to be posted after
+        // READ messages, so we do two passes.
+
+        int pass = 0;
+        for ( int i = 0; pass < 2; ++ i ) {
+            if ( pipeStates[i].fStatus == NPSS_EOI ) {
+                ++ pass;
+                i = -1;
+                continue;
+            }
+            if ( pass == 0 && pipeStates[i].fStatus != NPSS_RDATA )
+                continue;
+            if ( pass == 1 && pipeStates[i].fStatus != NPSS_CLOSE )
+                continue;
+
+#if defined(QT_QPROCESS_DEBUG)
+            qDebug( " %d/%d: fStatus=%u fFlag=%02X usKey=%04hX usAvail=%hu",
+                    pass, i, (uint) pipeStates[i].fStatus,
+                    (uint) pipeStates[i].fFlag, pipeStates[i].usKey,
+                    pipeStates[i].usAvail );
+#endif
+            QProcess *proc = pipeKeys.find( pipeStates[i].usKey );
+            Q_ASSERT( proc );
+            if ( !proc )
+                continue;
+#if defined(QT_QPROCESS_DEBUG)
+            qDebug( "  proc=%p (%s/%s) d=%p",
+                       proc, proc->name(), proc->className(), proc->d );
+#endif
+            QProcessPrivate::Pipe *pipe = proc->d->findPipe( pipeStates[i].usKey );
+            Q_ASSERT( pipe );
+            if ( !pipe )
+                continue;
+
+            if ( pipeStates[i].fStatus == NPSS_RDATA ) {
+                bool wasPending = pipe->pending > 0;    
+                pipe->pending = pipeStates[i].usAvail;
+                // inform the GUI thread that there is new data
+                if ( !wasPending )
+                    WinPostMsg( hwnd(), WM_U_PIPE_RDATA, MPFROMP( proc ),
+                                MPFROMSHORT( pipe->key ) );
+            } else if ( pipeStates[i].fStatus == NPSS_CLOSE ) {
+                // there may be repeated CLOSE records for the same pipe
+                // in subsequent DosQueryNPipeSemState() calls
+                if ( pipe->closed == FALSE ) {
+                    pipe->closed = TRUE;
+                    // inform the GUI thread that the client's pipe end
+                    // was closed
+                    WinPostMsg( hwnd(), WM_U_PIPE_CLOSE, MPFROMP( proc ),
+                                MPFROMSHORT( pipe->key ) );
+                }
+            }
+        }
+    }
+    lock.unlock();
+
+#if defined(QT_QPROCESS_DEBUG)
+    qDebug( "QProcessMonitor::monitor() LEAVE" );
+#endif            
+}
+
+MRESULT QProcessMonitor::message( ULONG msg, MPARAM mp1, MPARAM mp2 )
+{
+    switch ( msg ) {
+        case WM_U_PIPE_RDATA: {
+            QProcess *proc = (QProcess *) PVOIDFROMMP( mp1 );
+            USHORT key = SHORT1FROMMP( mp2 );
+#if defined(QT_QPROCESS_DEBUG)
+            qDebug( "QProcessMonitor::WM_U_PIPE_RDATA: proc=%p (%s/%s) d=%p "
+                    "key=%04hX",
+                    proc, proc->name(), proc->className(), proc->d, key );
+#endif
+            // check parameter validity (we can safely do it outside the lock
+            // because the GUI thread is the only that can modify pipeKeys)
+            if ( proc == pipeKeys.find( key ) ) {
+                QProcessPrivate *d = proc->d;
+                if ( d->stdout.key == key ) {
+                    emit proc->readyReadStdout();
+                } else {
+                    Q_ASSERT( d->stderr.key == key );
+                    emit proc->readyReadStderr();
+                }
+            }
+            break;
+        }
+        case WM_U_PIPE_CLOSE: {
+            QProcess *proc = (QProcess *) PVOIDFROMMP( mp1 );
+            USHORT key = SHORT1FROMMP( mp2 );
+#if defined(QT_QPROCESS_DEBUG)
+            qDebug( "QProcessMonitor::WM_U_PIPE_CLOSE: proc=%p (%s/%s) d=%p "
+                    "key=%04hX",
+                    proc, proc->name(), proc->className(), proc->d, key );
+#endif
+            // check parameter validity (we can safely do it outside the lock
+            // because the GUI thread is the only that can modify pipeKeys)
+            if ( proc == pipeKeys.find( key ) ) {
+                QProcessPrivate *d = proc->d;
+                QProcessPrivate::Pipe *pipe = d->findPipe( key );
+                Q_ASSERT( pipe );
+                if ( pipe ) {
+                    Q_ASSERT( pipe->closed );
+                    // remove the single pipe from watching
+                    removeProcess( d, pipe ); 
+                    // close the pipe since no more necessary
+                    // (pipe is not watched anymore, no need to lock access)
+                    if ( pipe->pending == 0 )
+                        d->closePipe( pipe );
+                }
+            }
+            break;
+        }
+        default:
+            break;
+    }
+    
+    return FALSE;
+}
+
+static QProcessMonitor *processMonitor = NULL;
+
+void QProcessMonitor_cleanup()
+{
+#if defined(QT_QPROCESS_DEBUG)
+    qDebug( "QProcessMonitor_cleanup()" );
+#endif
+    delete processMonitor;
+    processMonitor = NULL;
+}
+
+/** Returns true if some data was successfully read */
+bool QProcessPrivate::readPipe( Pipe *pipe )
+{
+#if defined(QT_QPROCESS_DEBUG)
+    qDebug( "QProcessPrivate::readPipe(): this=%p key=%04hX", this, pipe->key );
+#endif
+
+    if ( pipe->pipe == HP_NULL )
+        return false;
+
+    // Pipe::pending and Pipe::closed need the lock
+    QMutex *lock = NULL;
+    
+    // Notes:
+    // 1. If procesMonitor is NULL, it which means we're somewhere inside
+    //    the QApplication destruction procedure.
+    // 2. If pipe->key us KEY_NULL, it means GUI thread has already processed
+    //    the WM_U_PIPE_CLOSE message and we're free to access fields w/o a
+    //    lock. pipe->key is assigned only on the GUI thread, so it's safe
+    //    to check it here w/o the lock.
+
+    if ( processMonitor && pipe->key != KEY_NULL )
+        lock = &processMonitor->mlock();
+    
+    QByteArray *ba = NULL;
+    bool close = FALSE;
+    
+    {
+        QMutexLocker locker( lock );
+        
+        if ( pipe->pending == 0 )
+            return false;
+        
+        ba = new QByteArray( pipe->pending );
+            
+        ULONG read = 0;
+        APIRET rc = DosRead( pipe->pipe, ba->data(), pipe->pending, &read );
+        if ( rc != NO_ERROR ) {
+            delete ba;
+#if defined(QT_CHECK_STATE) || defined(QT_QPROCESS_DEBUG)
+            qSystemWarning( "Failed to read from the pipe!", rc ); 
+#endif
+            return false;
+        }
+    
+        Q_ASSERT( read == pipe->pending );
+        if ( read < pipe->pending )
+            ba->resize( read );
+        
+        pipe->pending -= read;
+        
+        close = pipe->pending == 0 && pipe->closed;
+    }
+    
+    if ( close ) {
+        // the client's end of pipe has been closed, so close our end as well
+        if ( processMonitor && pipe->key != KEY_NULL ) {
+            // WM_U_PIPE_CLOSE has been posted but not yet processed
+            processMonitor->removeProcess( this, pipe );
+        }
+        closePipe( pipe );
+    }
+        
+    pipe->buf.append( ba );
+    return true;
+}
 
 /***********************************************************************
@@ -164 +684 @@
     exitStat = 0;
     exitNormal = FALSE;
+    
+#if defined(QT_QPROCESS_DEBUG)
+    qDebug( "~QProcess::init(): d=%p", d );
+#endif
 }
 
 void QProcess::reset()
 {
+    // remove monitoring if the monitor is there (note that it's done before
+    // resetting QProcessPrivate which has fields protected by the
+    // QProcessMonitor lock) 
+    if ( processMonitor )
+        processMonitor->removeProcess( d );
+    
     d->reset();
     exitStat = 0;
     exitNormal = FALSE;
-    d->bufStdout.clear();
-    d->bufStderr.clear();
 }
 
 QMembuf* QProcess::membufStdout()
 {
-    if( d->pipeStdout != 0 )
-        socketRead( FD_PIPE_STDOUT );
-    return &d->bufStdout;
+    if( d->stdout.pipe != 0 )
+        d->readPipe( &d->stdout );
+    return &d->stdout.buf;
 }
 
 QMembuf* QProcess::membufStderr()
 {
-    if( d->pipeStderr != 0 )
-        socketRead( FD_PIPE_STDERR );
-    return &d->bufStderr;
+    if( d->stderr.pipe != 0 )
+        d->readPipe( &d->stderr );
+    return &d->stderr.buf;
 }
 
 QProcess::~QProcess()
 {
+#if defined(QT_QPROCESS_DEBUG)
+    qDebug( "~QProcess::QProcess(): d=%p", d );
+#endif
+
+    // remove monitoring if the monitor is there (note that it's done before
+    // resetting QProcessPrivate which has fields protected by the
+    // QProcessMonitor lock) 
+    if ( processMonitor )
+        processMonitor->removeProcess( d );
+
     delete d;
 }
 
-#if defined(QT_QPROCESS_DEBUG)
-static void DosPrintf( HFILE fd, const char *fmt, ... )
-{
-   static char buf[ 1024 ];
-    
-   va_list args;
-   va_start( args, fmt );
-   vsnprintf( buf, sizeof(buf), fmt, args );
-   va_end ( args );
-   
-   ULONG written = 0;
-   DosWrite( fd, buf, strlen( buf ), &written );
-}
-#endif
-
 bool QProcess::start( QStringList *env )
 {
@@ -214 +737 @@
     qDebug( "QProcess::start()" );
 #endif
+
     reset();
 
@@ -444 +968 @@
     // create STDIN/OUT/ERR redirection pipes
     
-    HFILE tmpStdin = HF_ANY, tmpStdout = HF_ANY, tmpStderr = HF_ANY;
+    if ( comms & (Stdout | Stderr) ) {
+        // lazily create the process monitor
+        if ( !processMonitor ) {
+            processMonitor = new QProcessMonitor();
+            Q_ASSERT( processMonitor );
+            if ( !processMonitor )
+                return FALSE;
+            if ( !processMonitor->isOk() ) {
+                QProcessMonitor_cleanup();
+                return FALSE;
+            }
+            qAddPostRoutine( QProcessMonitor_cleanup ); 
+        }
+    }
+    
+    HFILE tmpStdin = HF_NULL, tmpStdout = HF_NULL, tmpStderr = HF_NULL;
     HFILE realStdin = HF_STDIN, realStdout = HF_STDOUT, realStderr = HF_STDERR;
 
@@ -451 +990 @@
     DosGetInfoBlocks( NULL, &ppib );
 
+    // use the custom Qt message handler to print errors/wranings
+    // to the console when the real STDERR handle is redirected
+    InstallQtMsgHandler();
+
     if ( comms & Stdin ) {
-        HFILE clientStdin = HF_ANY;
+        HFILE clientStdin = HF_NULL;
         do {
             // create a Stdin pipe
@@ -459 +1002 @@
             rc = DosCreateNPipe( pathBuf, &d->pipeStdin,
                                  NP_ACCESS_OUTBOUND, NP_NOWAIT | NP_TYPE_BYTE | 1,
-                                 PIPE_SIZE, PIPE_SIZE, 0 );
+                                 PIPE_SIZE_STDIN, 0, 0 );
             if ( rc != NO_ERROR )
                 break;
@@ -478 +1021 @@
         } while( 0 );
         // close the client end anyway (no more necessary)
-        if ( clientStdin != HF_ANY )
+        if ( clientStdin != HF_NULL )
             DosClose ( clientStdin );
         // close all opened handles on error
         if ( rc  != NO_ERROR ) {
 #if defined(QT_QPROCESS_DEBUG)
-            DosPrintf( tmpStderr != HF_ANY ? tmpStderr : realStderr, 
-                       "Failed to create a STDIN redirection (%s): SYS%04ld\n\r",
-                       pathBuf, rc ); 
-#endif
-            if ( tmpStdin != HF_ANY )
+            qDebug( "Failed to create a STDIN redirection (%s): SYS%04ld",
+                    pathBuf, rc ); 
+#endif
+            if ( tmpStdin != HF_NULL )
                 DosClose ( tmpStdin );
             d->closeHandles();
+            UninstallQtMsgHandler();
             return FALSE;
         }
     }
     if ( comms & Stdout ) {
-        HFILE clientStdout = HF_ANY;
+        HFILE clientStdout = HF_NULL;
         do {
             // create a Stdout pipe
             sprintf( pathBuf, "\\pipe\\Qt\\%08lX\\QProcess\\%p\\Stdout",
                      ppib->pib_ulpid, this );
-            rc = DosCreateNPipe( pathBuf, &d->pipeStdout,
+            rc = DosCreateNPipe( pathBuf, &d->stdout.pipe,
                                  NP_ACCESS_INBOUND, NP_NOWAIT | NP_TYPE_BYTE | 1,
-                                 PIPE_SIZE, PIPE_SIZE, 0 );
+                                 0, PIPE_SIZE_STDOUT, 0 );
             if ( rc != NO_ERROR )
                 break;
-            DosConnectNPipe( d->pipeStdout );
+            DosConnectNPipe( d->stdout.pipe );
             // open a client end of the Stdout pipe
             ULONG action = 0;
@@ -520 +1063 @@
         } while( 0 );
         // close the client end anyway (no more necessary)
-        if ( clientStdout != HF_ANY )
+        if ( clientStdout != HF_NULL )
             DosClose ( clientStdout );
         // close all opened handles on error
         if ( rc != NO_ERROR ) {
 #if defined(QT_QPROCESS_DEBUG)
-            DosPrintf( tmpStderr != HF_ANY ? tmpStderr : realStderr, 
-                       "Failed to create a STDOUT redirection (%s): SYS%04ld\n\r", 
-                       pathBuf, rc ); 
-#endif
-            if ( tmpStdin != HF_ANY ) {
+            qDebug( "Failed to create a STDOUT redirection (%s): SYS%04ld", 
+                    pathBuf, rc ); 
+#endif
+            if ( tmpStdin != HF_NULL ) {
                 DosDupHandle( tmpStdin, &realStdin );
                 DosClose( tmpStdin );
             }
-            if ( tmpStdout != HF_ANY )
+            if ( tmpStdout != HF_NULL )
                 DosClose ( tmpStdout );
             d->closeHandles();
+            UninstallQtMsgHandler();
             return FALSE;
         }
     }
     if ( (comms & Stderr) && !(comms & DupStderr) ) {
-        HFILE clientStderr = HF_ANY;
+        HFILE clientStderr = HF_NULL;
         do {
             // create a Stderr pipe
             sprintf( pathBuf, "\\pipe\\Qt\\%08lX\\QProcess\\%p\\Stderr",
                      ppib->pib_ulpid, this );
-            rc = DosCreateNPipe( pathBuf, &d->pipeStderr,
+            rc = DosCreateNPipe( pathBuf, &d->stderr.pipe,
                                  NP_ACCESS_INBOUND, NP_NOWAIT | NP_TYPE_BYTE | 1,
-                                 PIPE_SIZE, PIPE_SIZE, 0 );
+                                 0, PIPE_SIZE_STDERR, 0 );
             if ( rc != NO_ERROR )
                 break;
-            DosConnectNPipe( d->pipeStderr );
+            DosConnectNPipe( d->stderr.pipe );
             // open a client end of the Stderr pipe
             ULONG action = 0;
@@ -566 +1109 @@
         } while( 0 );
         // close the client end anyway (no more necessary)
-        if ( clientStderr != HF_ANY )
+        if ( clientStderr != HF_NULL )
             DosClose ( clientStderr );
         // close all opened handles on error
         if ( rc != NO_ERROR ) {
 #if defined(QT_QPROCESS_DEBUG)
-            DosPrintf( tmpStderr != HF_ANY ? tmpStderr : realStderr, 
-                       "Failed to create a STDERR redirection (%s): SYS%04ld\n\r", 
-                       pathBuf, rc ); 
-#endif
-            if ( tmpStdin != HF_ANY ) {
+            qDebug( "Failed to create a STDERR redirection (%s): SYS%04ld", 
+                    pathBuf, rc ); 
+#endif
+            if ( tmpStdin != HF_NULL ) {
                 DosDupHandle( tmpStdin, &realStdin );
                 DosClose( tmpStdin );
             }
-            if ( tmpStdout != HF_ANY ) {
+            if ( tmpStdout != HF_NULL ) {
                 DosDupHandle( tmpStdout, &realStdout );
                 DosClose( tmpStdout );
             }
-            if ( tmpStderr != HF_ANY )
+            if ( tmpStderr != HF_NULL )
                 DosClose ( tmpStderr );
             d->closeHandles();
+            UninstallQtMsgHandler();
             return FALSE;
         }
     }
     if ( comms & DupStderr ) {
-        d->pipeStderr = d->pipeStdout;
+        // leave d->stderr.pipe equal to HP_NULL
         do {
             // save the real STDERR handle
@@ -602 +1145 @@
         if ( rc != NO_ERROR ) {
 #if defined(QT_QPROCESS_DEBUG)
-            DosPrintf( tmpStderr != HF_ANY ? tmpStderr : realStderr, 
-                       "Failed to redirect STDERR to STDOUT: SYS%04ld\n\r", 
-                       rc ); 
-#endif
-            if ( tmpStdin != HF_ANY ) {
+            qDebug( "Failed to redirect STDERR to STDOUT: SYS%04ld", 
+                    rc ); 
+#endif
+            if ( tmpStdin != HF_NULL ) {
                 DosDupHandle( tmpStdin, &realStdin );
                 DosClose( tmpStdin );
             }
-            if ( tmpStdout != HF_ANY ) {
+            if ( tmpStdout != HF_NULL ) {
                 DosDupHandle( tmpStdout, &realStdout );
                 DosClose( tmpStdout );
             }
-            if ( tmpStderr != HF_ANY )
+            if ( tmpStderr != HF_NULL )
                 DosClose ( tmpStderr );
             d->closeHandles();
+            UninstallQtMsgHandler();
             return FALSE;
         }
     }
 
+    // add this process to the monitor
+    if ( !processMonitor->addProcess( this ) ) {
+        if ( tmpStdin != HF_NULL ) {
+            DosDupHandle( tmpStdin, &realStdin );
+            DosClose( tmpStdin );
+        }
+        if ( tmpStdout != HF_NULL ) {
+            DosDupHandle( tmpStdout, &realStdout );
+            DosClose( tmpStdout );
+        }
+        if ( tmpStderr != HF_NULL ) {
+            DosDupHandle( tmpStderr, &realStderr );
+            DosClose ( tmpStderr );
+        }
+        d->closeHandles();
+        UninstallQtMsgHandler();
+        return FALSE;
+    }
+    
     // set the working directory
     QString curDir = QDir::currentDirPath();
     QDir::setCurrent( workingDir.path() );
 #if defined(QT_QPROCESS_DEBUG)
-    DosPrintf( tmpStderr != HF_ANY ? tmpStderr : realStderr,
-               "QProcess::start(): curDir='%s', workingDir='%s'\n\r",
-               curDir.local8Bit().data(),
-               QDir::currentDirPath().local8Bit().data() ); 
+    qDebug( "QProcess::start(): curDir='%s', workingDir='%s'",
+            curDir.local8Bit().data(),
+            QDir::currentDirPath().local8Bit().data() ); 
 #endif
     
@@ -642 +1203 @@
     
     // cancel STDIN/OUT/ERR redirections
-    if ( tmpStdin != HF_ANY ) {
+    if ( tmpStdin != HF_NULL ) {
         DosDupHandle( tmpStdin, &realStdin );
         DosClose( tmpStdin );
     }
-    if ( tmpStdout != HF_ANY ) {
+    if ( tmpStdout != HF_NULL ) {
         DosDupHandle( tmpStdout, &realStdout );
         DosClose( tmpStdout );
     }
-    if ( tmpStderr != HF_ANY ) {
+    if ( tmpStderr != HF_NULL ) {
         DosDupHandle( tmpStderr, &realStderr );
         DosClose( tmpStderr );
     }
+    
+    UninstallQtMsgHandler();
     
     if ( rc != NO_ERROR ) {
@@ -664 +1227 @@
         qSystemWarning( "Failed to start a new process!", rc );
 #endif
+        processMonitor->removeProcess( d );
         d->closeHandles();
         return FALSE;
@@ -671 +1235 @@
     d->pid = resc.codeTerminate;
     
-    if ( ioRedirection || notifyOnExit ) {
-        d->lookup->start( 100 );
+    // timer is not necessary for ioRedirection (we use the monitor thread)
+    if ( /* ioRedirection || */ notifyOnExit ) {
+        d->lookup->start( POLL_TIMER );
     }
 
@@ -725 +1290 @@
     if ( rc == ERROR_CHILD_NOT_COMPLETE )
         return TRUE;
-    
-    // there might be data to read
+
     QProcess *that = (QProcess *) this;
-    that->socketRead( FD_PIPE_STDOUT ); // try stdout
-    that->socketRead( FD_PIPE_STDERR ); // try stderr
+
+    // There might be data to read, but WM_U_PIPE_RDATA messages won't be
+    // converted to signals after removeProcess() is called below. Therefore,
+    // emit signals ourselved if necessary.
+
+    if ( d->readPipe( &d->stdout ) )
+        emit that->readyReadStdout();
+    if ( d->readPipe( &d->stderr ) )
+        emit that->readyReadStderr();
     
     // compute the exit values
@@ -737 +1308 @@
         d->exitValuesCalculated = TRUE;
     }
+
+    processMonitor->removeProcess( d );
     
     d->pid = PID_NULL;
@@ -745 +1318 @@
 bool QProcess::canReadLineStdout() const
 {
-    if( d->pipeStdout == HF_ANY )
-        return d->bufStdout.size() != 0;
+    if( d->stdout.pipe == HP_NULL )
+        return d->stdout.buf.size() != 0;
 
     QProcess *that = (QProcess *) this;
@@ -754 +1327 @@
 bool QProcess::canReadLineStderr() const
 {
-    if( d->pipeStderr == HF_ANY )
-        return d->bufStderr.size() != 0;
+    if( d->stderr.pipe == HP_NULL )
+        return d->stderr.buf.size() != 0;
 
     QProcess *that = (QProcess *) this;
@@ -769 +1342 @@
 void QProcess::closeStdin( )
 {
-    if ( d->pipeStdin != HF_ANY ) {
+    if ( d->pipeStdin != HP_NULL ) {
         DosDisConnectNPipe( d->pipeStdin );
         DosClose( d->pipeStdin );
-        d->pipeStdin = HF_ANY;
-    }
-}
-
-void QProcess::socketRead( int fd )
-{
-    Q_ASSERT( fd == FD_PIPE_STDOUT || fd == FD_PIPE_STDERR );
-    
-    HPIPE pipe = HF_ANY;
-    switch( fd ) {
-        case FD_PIPE_STDOUT: pipe = d->pipeStdout; break; 
-        case FD_PIPE_STDERR: pipe = d->pipeStderr; break;
-    }
-    
-    if ( pipe == HF_ANY )
-        return;
-    
-    // get the number of bytes that are waiting to be read
-    ULONG read = 0;
-    AVAILDATA avail = { 0 };
-    ULONG status = 0;
-    APIRET rc = DosPeekNPipe( pipe, NULL, 0, &read, &avail, &status );
-    if ( rc != NO_ERROR ) {
-#if defined(QT_QPROCESS_DEBUG)
-        qSystemWarning( "Failed to peek from the pipe!", rc ); 
-#endif
-        return;
-    }
-    
-    if ( avail.cbpipe > 0 ) {
-        QMembuf *buffer;
-        if ( fd == FD_PIPE_STDOUT )
-            buffer = &d->bufStdout;
-        else
-            buffer = &d->bufStderr; 
-        
-        QByteArray *ba = new QByteArray( avail.cbpipe );
-        
-        rc = DosRead( pipe, ba->data(), avail.cbpipe, &read );
-        if ( rc != NO_ERROR ) {
-            delete ba;
-#if defined(QT_QPROCESS_DEBUG)
-            qSystemWarning( "Failed to read from the pipe!", rc ); 
-#endif
-            return;
-        }
-        
-        if ( read < avail.cbpipe )
-            ba->resize( avail.cbpipe );
-
-        buffer->append( ba );
-        if ( fd == FD_PIPE_STDOUT )
-            emit readyReadStdout();
-        else
-            emit readyReadStderr(); 
-    }
+        d->pipeStdin = HP_NULL;
+    }
+}
+
+/* dummy, since MOC doesn't understand preprocessor defines */
+void QProcess::socketRead( int fd ) 
+{
 }
 
 void QProcess::socketWrite( int )
 {
-    if ( d->pipeStdin == HF_ANY )
+    if ( d->pipeStdin == HP_NULL )
         return;
     
@@ -839 +1363 @@
         APIRET rc = DosWrite( d->pipeStdin,
                               d->stdinBuf.head()->data() + d->stdinBufRead,
-                              QMIN( PIPE_SIZE, d->stdinBuf.head()->size() - d->stdinBufRead ),
+                              QMIN( PIPE_SIZE_STDIN,
+                                    d->stdinBuf.head()->size() - d->stdinBufRead ),
                               &written );
         if ( rc != NO_ERROR ) {
-#if defined(QT_QPROCESS_DEBUG)
+#if defined(QT_CHECK_STATE) || defined(QT_QPROCESS_DEBUG)
             qSystemWarning( "Failed to write to the pipe!", rc ); 
 #endif
-            d->lookup->start( 100 );
+            d->lookup->start( POLL_TIMER );
             return;
         }
@@ -878 +1403 @@
         socketWrite( 0 );
 
-    if ( ioRedirection ) {
-        socketRead( FD_PIPE_STDOUT ); // try stdout
-        socketRead( FD_PIPE_STDERR ); // try stderr
-    }
-
     if ( isRunning() ) {
         // enable timer again, if needed
-        if ( !d->stdinBuf.isEmpty() || ioRedirection || notifyOnExit )
-            d->lookup->start( 100 );
+        // timer is not necessary for ioRedirection (we use the monitor thread)
+        if ( !d->stdinBuf.isEmpty() || /* ioRedirection || */ notifyOnExit )
+            d->lookup->start( POLL_TIMER );
     } else if ( notifyOnExit ) {
         emit processExited();
@@ -899 +1420 @@
 {
     ioRedirection = value;
+    // timer is not necessary for ioRedirection (we use the monitor thread)
+#if 0
     if ( !ioRedirection && !notifyOnExit )
         d->lookup->stop();
     if ( ioRedirection ) {
         if ( isRunning() )
-            d->lookup->start( 100 );
-    }
+            d->lookup->start( POLL_TIMER );
+    }
+#endif
 }
 
@@ -914 +1438 @@
 {
     notifyOnExit = value;
-    if ( !ioRedirection && !notifyOnExit )
+    // timer is not necessary for ioRedirection (we use the monitor thread)
+    if ( /* !ioRedirection && */ !notifyOnExit )
         d->lookup->stop();
     if ( notifyOnExit ) {
         if ( isRunning() )
-            d->lookup->start( 100 );
+            d->lookup->start( POLL_TIMER );
     }
 }