Qt::HANDLE handle;
public:
QMutex mutex;
+ QWaitCondition waitForCancelCondition;
QWaitCondition bufferNotFullCondition;
QWaitCondition bufferNotEmptyCondition;
QWaitCondition hasStarted;
error( false ),
eofShortCut( false ),
errorCode( 0 ),
- rptr( 0 ), wptr( 0 ),
- consumerBlocksOnUs( false )
+ consumerBlocksOnUs( false ),
+ rptr( 0 ), wptr( 0 )
{
}
bool startReaderThread();
bool startWriterThread();
void stopThreads();
- bool triedToStartReader;
- bool triedToStartWriter;
public Q_SLOTS:
void emitReadyRead();
Qt::HANDLE handle;
Reader * reader;
Writer * writer;
+ bool triedToStartReader;
+ bool triedToStartWriter;
};
KDPipeIODevice::Private::Private( KDPipeIODevice * qq )
if ( bytesAvailable() > 0 )
maxSize = std::min( maxSize, bytesAvailable() ); // don't block
}
- qDebug( "%p: KDPipeIODevice::readData: try to lock reader (CONSUMER THREAD)" );
+ qDebug( "%p: KDPipeIODevice::readData: try to lock reader (CONSUMER THREAD)", this );
LOCKED( r );
- qDebug( "%p: KDPipeIODevice::readData: locked reader (CONSUMER THREAD)" );
+ qDebug( "%p: KDPipeIODevice::readData: locked reader (CONSUMER THREAD)", this );
r->readyReadSentCondition.wakeAll();
if ( /* maxSize > 0 && */ r->bufferEmpty() && !r->error && !r->eof ) { // ### block on maxSize == 0?
// tell thread to cancel:
r->cancel = true;
// and wake it, so it can terminate:
+ r->waitForCancelCondition.wakeAll();
r->bufferNotFullCondition.wakeAll();
r->readyReadSentCondition.wakeAll();
}
qDebug( "KPipeIODevice::close(%p): wait and closing writer %p", this, d->writer );
waitAndDelete( d->writer );
qDebug( "KPipeIODevice::close(%p): wait and closing reader %p", this, d->reader );
- {
+ if ( d->reader ) {
LOCKED( d->reader );
d->reader->readyReadSentCondition.wakeAll();
}
while ( true ) {
if ( !cancel && ( eof || error ) ) {
+ //notify the client until the buffer is empty and then once
+ //again so he receives eof/error. After that, wait for him
+ //to cancel
+ const bool wasEmpty = bufferEmpty();
qDebug( "%p: Reader::run: received eof(%d) or error(%d), waking everyone", this, eof, error );
notifyReadyRead();
- cancel = true;
+ if ( !cancel && wasEmpty )
+ waitForCancelCondition.wait( &mutex );
} else if ( !cancel && !bufferFull() && !bufferEmpty() ) {
qDebug( "%p: Reader::run: buffer no longer empty, waking everyone", this );
notifyReadyRead();
}
- while ( !error && !cancel && bufferFull() ) {
+ while ( !cancel && !error && bufferFull() ) {
notifyReadyRead();
if ( bufferFull() ) {
qDebug( "%p: Reader::run: buffer is full, going to sleep", this );
goto leave;
}
#endif
- qDebug( "%p (fd=%d): Writer::run: buffer after WriteFile (numBytes=%lld): %s:", this, fd, numBytesInBuffer,
-buffer );
+ qDebug( "%p (fd=%d): Writer::run: buffer after WriteFile (numBytes=%u): %s:", this, fd, numBytesInBuffer, buffer );
totalWritten += numWritten;
mutex.lock();
} while ( totalWritten < numBytesInBuffer );