No OneTemporary
Actions

Size

29 KB

Subscribers

None

View Options

This document is not UTF8. It was detected as Shift JIS and converted to UTF8 for display.

	diff --git a/utils/kdpipeiodevice.cpp b/utils/kdpipeiodevice.cpp
	index 13b84a062..755445a79 100644
	--- a/utils/kdpipeiodevice.cpp
	+++ b/utils/kdpipeiodevice.cpp
	@@ -1,954 +1,958 @@
	/*
	Copyright (C) 2007 Klar舁vdalens Datakonsult AB

	KDPipeIODevice is free software; you can redistribute it and/or
	modify it under the terms of the GNU Library General Public
	License as published by the Free Software Foundation; either
	version 2 of the License, or (at your option) any later version.

	KDPipeIODevice is distributed in the hope that it will be useful,
	but WITHOUT ANY WARRANTY; without even the implied warranty of
	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	GNU Library General Public License for more details.

	You should have received a copy of the GNU Library General Public License
	along with KDPipeIODevice; see the file COPYING.LIB. If not, write to the
	Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
	Boston, MA 02110-1301, USA.
	*/

	#include <config-kleopatra.h>

	#include "kdpipeiodevice.h"

	#include <QtCore>

	#include <cassert>
	#include <cstring>
	#include <memory>
	#include <algorithm>

	#ifdef Q_OS_WIN32
	# ifndef NOMINMAX
	# define NOMINMAX
	# endif
	# include <windows.h>
	# include <io.h>
	#else
	# include <unistd.h>
	# include <errno.h>
	#endif

	#ifndef KDAB_CHECK_THIS
	# define KDAB_CHECK_CTOR (void)1
	# define KDAB_CHECK_DTOR KDAB_CHECK_CTOR
	# define KDAB_CHECK_THIS KDAB_CHECK_CTOR
	#endif

	#define LOCKED( d ) const QMutexLocker locker( &d->mutex )
	#define synchronized( d ) if ( int i = 0 ) {} else for ( const QMutexLocker locker( &d->mutex ) ; !i ; ++i )

	const unsigned int BUFFER_SIZE = 4096;
	const bool ALLOW_QIODEVICE_BUFFERING = true;

	namespace {
	KDPipeIODevice::DebugLevel s_debugLevel = KDPipeIODevice::NoDebug;
	}

	#define qDebug if( s_debugLevel == KDPipeIODevice::NoDebug ){}else qDebug

	namespace {

	class Reader : public QThread {
	Q_OBJECT
	public:
	Reader( int fd, Qt::HANDLE handle );
	~Reader();

	qint64 readData( char * data, qint64 maxSize );

	unsigned int bytesInBuffer() const {
	return ( wptr + sizeof buffer - rptr ) % sizeof buffer ;
	}

	bool bufferFull() const {
	return bytesInBuffer() == sizeof buffer - 1;
	}

	bool bufferEmpty() const {
	return bytesInBuffer() == 0;
	}

	bool bufferContains( char ch ) {
	const unsigned int bib = bytesInBuffer();
	for ( unsigned int i = rptr ; i < rptr + bib ; ++i )
	if ( buffer[i%sizeof buffer] == ch )
	return true;
	return false;
	}

	void notifyReadyRead();

	Q_SIGNALS:
	void readyRead();

	protected:
	/* reimp */ void run();

	private:
	int fd;
	Qt::HANDLE handle;
	public:
	QMutex mutex;
	QWaitCondition waitForCancelCondition;
	QWaitCondition bufferNotFullCondition;
	QWaitCondition bufferNotEmptyCondition;
	QWaitCondition hasStarted;
	QWaitCondition readyReadSentCondition;
	QWaitCondition blockedConsumerIsDoneCondition;
	bool cancel;
	bool eof;
	bool error;
	bool eofShortCut;
	int errorCode;
	+ bool isReading;
	bool consumerBlocksOnUs;

	private:
	unsigned int rptr, wptr;
	char buffer[BUFFER_SIZE+1]; // need to keep one byte free to detect empty state
	};


	Reader::Reader( int fd_, Qt::HANDLE handle_ ) : QThread(),
	fd( fd_ ),
	handle( handle_ ),
	mutex(),
	bufferNotFullCondition(),
	bufferNotEmptyCondition(),
	hasStarted(),
	cancel( false ),
	eof( false ),
	error( false ),
	eofShortCut( false ),
	errorCode( 0 ),
	+ isReading( false ),
	consumerBlocksOnUs( false ),
	rptr( 0 ),
	wptr( 0 )
	{

	}

	Reader::~Reader() {}


	class Writer : public QThread {
	Q_OBJECT
	public:
	Writer( int fd, Qt::HANDLE handle );
	~Writer();

	qint64 writeData( const char * data, qint64 size );

	unsigned int bytesInBuffer() const { return numBytesInBuffer; }

	bool bufferFull() const {
	return numBytesInBuffer == sizeof buffer;
	}

	bool bufferEmpty() const {
	return numBytesInBuffer == 0;
	}

	Q_SIGNALS:
	void bytesWritten( qint64 );

	protected:
	/* reimp */ void run();

	private:
	int fd;
	Qt::HANDLE handle;
	public:
	QMutex mutex;
	QWaitCondition bufferEmptyCondition;
	QWaitCondition bufferNotEmptyCondition;
	QWaitCondition hasStarted;
	bool cancel;
	bool error;
	int errorCode;
	private:
	unsigned int numBytesInBuffer;
	char buffer[BUFFER_SIZE];
	};
	}

	Writer::Writer( int fd_, Qt::HANDLE handle_ ) : QThread(),
	fd( fd_ ),
	handle( handle_ ),
	mutex(),
	bufferEmptyCondition(),
	bufferNotEmptyCondition(),
	hasStarted(),
	cancel( false ),
	error( false ),
	errorCode( 0 ),
	numBytesInBuffer( 0 )
	{

	}

	Writer::~Writer() {}


	class KDPipeIODevice::Private : public QObject {
	Q_OBJECT
	friend class ::KDPipeIODevice;
	KDPipeIODevice * const q;
	public:

	explicit Private( KDPipeIODevice * qq );
	~Private();

	bool doOpen( int, Qt::HANDLE, OpenMode );
	bool startReaderThread();
	bool startWriterThread();
	void stopThreads();

	public Q_SLOTS:
	void emitReadyRead();

	private:
	int fd;
	Qt::HANDLE handle;
	Reader * reader;
	Writer * writer;
	bool triedToStartReader;
	bool triedToStartWriter;
	};

	KDPipeIODevice::DebugLevel KDPipeIODevice::debugLevel()
	{
	return s_debugLevel;
	}

	void KDPipeIODevice::setDebugLevel( KDPipeIODevice::DebugLevel level )
	{
	s_debugLevel = level;
	}

	KDPipeIODevice::Private::Private( KDPipeIODevice * qq ) : QObject( qq ), q( qq ),
	fd( -1 ),
	handle( 0 ),
	reader( 0 ),
	writer( 0 ),
	triedToStartReader( false ),
	triedToStartWriter( false )
	{

	}

	KDPipeIODevice::Private::~Private() {
	qDebug( "KDPipeIODevice::~Private(): Destroying %p", q );
	}

	KDPipeIODevice::KDPipeIODevice( QObject * p )
	: QIODevice( p ), d( new Private( this ) )
	{
	KDAB_CHECK_CTOR;
	}

	KDPipeIODevice::KDPipeIODevice( int fd, OpenMode mode, QObject * p )
	: QIODevice( p ), d( new Private( this ) )
	{
	KDAB_CHECK_CTOR;
	open( fd, mode );
	}

	KDPipeIODevice::KDPipeIODevice( Qt::HANDLE handle, OpenMode mode, QObject * p )
	: QIODevice( p ), d( new Private( this ) )
	{
	KDAB_CHECK_CTOR;
	open( handle, mode );
	}

	KDPipeIODevice::~KDPipeIODevice() { KDAB_CHECK_DTOR;
	if ( isOpen() )
	close();
	delete d; d = 0;
	}


	bool KDPipeIODevice::open( int fd, OpenMode mode ) { KDAB_CHECK_THIS;
	#ifdef Q_OS_WIN32
	return d->doOpen( fd, (HANDLE)_get_osfhandle( fd ), mode );
	#else
	return d->doOpen( fd, 0, mode );
	#endif
	}

	bool KDPipeIODevice::open( Qt::HANDLE h, OpenMode mode ) { KDAB_CHECK_THIS;
	#ifdef Q_OS_WIN32
	return d->doOpen( -1, h, mode );
	#else
	Q_UNUSED( h );
	Q_UNUSED( mode );
	assert( !"KDPipeIODevice::open( Qt::HANDLE, OpenMode ) should never be called except on Windows." );
	#endif
	}

	bool KDPipeIODevice::Private::startReaderThread()
	{
	if ( triedToStartReader )
	return true;
	triedToStartReader = true;
	if ( reader && !reader->isRunning() && !reader->isFinished() ) {
	qDebug("KDPipeIODevice::Private::startReaderThread(): locking reader (CONSUMER THREAD)" );
	LOCKED( reader );
	qDebug("KDPipeIODevice::Private::startReaderThread(): locked reader (CONSUMER THREAD)" );
	reader->start( QThread::HighestPriority );
	qDebug("KDPipeIODevice::Private::startReaderThread(): waiting for hasStarted (CONSUMER THREAD)" );
	const bool hasStarted = reader->hasStarted.wait( &reader->mutex, 1000 );
	qDebug("KDPipeIODevice::Private::startReaderThread(): returned from hasStarted (CONSUMER THREAD)" );

	return hasStarted;
	}
	return true;
	}

	bool KDPipeIODevice::Private::startWriterThread()
	{
	if ( triedToStartWriter )
	return true;
	triedToStartWriter = true;
	if ( writer && !writer->isRunning() && !writer->isFinished() ) {
	LOCKED( writer );

	writer->start( QThread::HighestPriority );
	if ( !writer->hasStarted.wait( &writer->mutex, 1000 ) )
	return false;
	}
	return true;
	}

	void KDPipeIODevice::Private::emitReadyRead()
	{
	- static int s_counter = 0;
	- const int counter = s_counter++;
	QPointer<Private> thisPointer( this );
	- qDebug( "KDPipeIODevice::Private::emitReadyRead %p, %d", this, counter );
	+ qDebug( "KDPipeIODevice::Private::emitReadyRead %p", this );

	emit q->readyRead();

	if ( !thisPointer )
	return;
	-
	+ bool mustNotify = false;
	if ( reader ) {
	- qDebug( "KDPipeIODevice::Private::emitReadyRead %p, %d: locking reader (CONSUMER THREAD)", this, counter );
	+ qDebug( "KDPipeIODevice::Private::emitReadyRead %p: locking reader (CONSUMER THREAD)", this );
	synchronized( reader ) {
	- qDebug( "KDPipeIODevice::Private::emitReadyRead %p, %d: locked reader (CONSUMER THREAD)", this, counter );
	+ qDebug( "KDPipeIODevice::Private::emitReadyRead %p: locked reader (CONSUMER THREAD)", this );
	reader->readyReadSentCondition.wakeAll();
	+ mustNotify = !reader->bufferEmpty() && reader->isReading;
	+ qDebug( "KDPipeIODevice::Private::emitReadyRead %p: buffer empty: %d reader in ReadFile: %d", this, reader->bufferEmpty(), reader->isReading );
	}
	}
	- qDebug( "KDPipeIODevice::Private::emitReadyRead %p leaving %d", this, counter );
	+ qDebug( "KDPipeIODevice::Private::emitReadyRead %p leaving", this );

	}

	bool KDPipeIODevice::Private::doOpen( int fd_, Qt::HANDLE handle_, OpenMode mode_ ) {

	if ( q->isOpen() )
	return false;

	#ifdef Q_OS_WIN32
	if ( !handle_ )
	return false;
	#else
	if ( fd_ < 0 )
	return false;
	#endif

	if ( !(mode_ & ReadWrite) )
	return false; // need to have at least read -or- write


	std::auto_ptr<Reader> reader_;
	std::auto_ptr<Writer> writer_;

	if ( mode_ & ReadOnly ) {
	reader_.reset( new Reader( fd_, handle_ ) );
	qDebug( "KDPipeIODevice::doOpen (%p): created reader (%p) for fd %d", this, reader_.get(), fd_ );
	connect( reader_.get(), SIGNAL(readyRead()), this, SLOT(emitReadyRead()),
	Qt::QueuedConnection );
	}
	if ( mode_ & WriteOnly ) {
	writer_.reset( new Writer( fd_, handle_ ) );
	qDebug( "KDPipeIODevice::doOpen (%p): created writer (%p) for fd %d", this, writer_.get(), fd_ );
	connect( writer_.get(), SIGNAL(bytesWritten(qint64)), q, SIGNAL(bytesWritten(qint64)),
	Qt::QueuedConnection );
	}

	// commit to *this:
	fd = fd_;
	handle = handle_;
	reader = reader_.release();
	writer = writer_.release();

	q->setOpenMode( mode_\|Unbuffered );
	return true;
	}

	int KDPipeIODevice::descriptor() const { KDAB_CHECK_THIS;
	return d->fd;
	}


	Qt::HANDLE KDPipeIODevice::handle() const { KDAB_CHECK_THIS;
	return d->handle;
	}

	qint64 KDPipeIODevice::bytesAvailable() const { KDAB_CHECK_THIS;
	const qint64 base = QIODevice::bytesAvailable();
	if ( !d->triedToStartReader ) {
	d->startReaderThread();
	return base;
	}
	if ( d->reader )
	synchronized( d->reader ) {
	const qint64 inBuffer = d->reader->bytesInBuffer();
	return base + inBuffer;
	}
	return base;
	}

	qint64 KDPipeIODevice::bytesToWrite() const { KDAB_CHECK_THIS;
	d->startWriterThread();
	const qint64 base = QIODevice::bytesToWrite();
	if ( d->writer )
	synchronized( d->writer ) return base + d->writer->bytesInBuffer();
	return base;
	}

	bool KDPipeIODevice::canReadLine() const { KDAB_CHECK_THIS;
	d->startReaderThread();
	if ( QIODevice::canReadLine() )
	return true;
	if ( d->reader )
	synchronized( d->reader ) return d->reader->bufferContains( '\n' );
	return true;
	}

	bool KDPipeIODevice::isSequential() const {
	return true;
	}

	bool KDPipeIODevice::atEnd() const { KDAB_CHECK_THIS;
	d->startReaderThread();
	if ( !QIODevice::atEnd() ) {
	qDebug( "%p: KDPipeIODevice::atEnd returns false since QIODevice::atEnd does (with bytesAvailable=%ld)", this, static_cast<long>(bytesAvailable()) );
	return false;
	}
	if ( !isOpen() )
	return true;
	if ( d->reader->eofShortCut )
	return true;
	LOCKED( d->reader );
	const bool eof = ( d->reader->error \|\| d->reader->eof ) && d->reader->bufferEmpty();
	if ( !eof ) {
	if ( !d->reader->error && !d->reader->eof )
	qDebug( "%p: KDPipeIODevice::atEnd returns false since !reader->error && !reader->eof", this );
	if ( !d->reader->bufferEmpty() )
	qDebug( "%p: KDPipeIODevice::atEnd returns false since !reader->bufferEmpty()", this );
	}
	return eof;
	}

	bool KDPipeIODevice::waitForBytesWritten( int msecs ) { KDAB_CHECK_THIS;
	d->startWriterThread();
	Writer * const w = d->writer;
	if ( !w )
	return true;
	LOCKED( w );
	qDebug( "KDPipeIODevice::waitForBytesWritten (%p,w=%p): entered locked area", this, w
	);
	return w->bufferEmpty() \|\| w->error \|\| w->bufferEmptyCondition.wait( &w->mutex, msecs ) ;
	}

	bool KDPipeIODevice::waitForReadyRead( int msecs ) { KDAB_CHECK_THIS;
	qDebug( "KDPipeIODEvice::waitForReadyRead()(%p)", this);
	d->startReaderThread();
	if ( ALLOW_QIODEVICE_BUFFERING ) {
	if ( bytesAvailable() > 0 )
	return true;
	}
	Reader * const r = d->reader;
	if ( !r \|\| r->eofShortCut )
	return true;
	LOCKED( r );
	if ( r->bytesInBuffer() != 0 \|\| r->eof \|\| r->error )
	return true;
	assert( false );
	return r->bufferNotEmptyCondition.wait( &r->mutex, msecs ) ;
	}

	template <typename T>
	class TemporaryValue {
	public:
	TemporaryValue( T& var_, const T& tv ) : var( var_ ), oldValue( var_ ) { var = tv; }
	~TemporaryValue() { var = oldValue; }
	private:
	T& var;
	const T oldValue;
	};


	bool KDPipeIODevice::readWouldBlock() const
	{
	d->startReaderThread();
	LOCKED( d->reader );
	return d->reader->bufferEmpty() && !d->reader->eof && !d->reader->error;
	}

	bool KDPipeIODevice::writeWouldBlock() const
	{
	d->startWriterThread();
	LOCKED( d->writer );
	return !d->writer->bufferEmpty() && !d->writer->error;
	}


	qint64 KDPipeIODevice::readData( char * data, qint64 maxSize ) { KDAB_CHECK_THIS;
	qDebug( "%p: KDPipeIODevice::readData: data=%p, maxSize=%lld", this, data, maxSize );
	d->startReaderThread();
	Reader * const r = d->reader;

	assert( r );


	//assert( r->isRunning() ); // wrong (might be eof, error)
	assert( data \|\| maxSize == 0 );
	assert( maxSize >= 0 );

	if ( r->eofShortCut ) {
	qDebug( "%p: KDPipeIODevice::readData: hit eofShortCut, returning 0", this );
	return 0;
	}

	if ( maxSize < 0 )
	maxSize = 0;

	if ( ALLOW_QIODEVICE_BUFFERING ) {
	if ( bytesAvailable() > 0 )
	maxSize = std::min( maxSize, bytesAvailable() ); // don't block
	}
	qDebug( "%p: KDPipeIODevice::readData: try to lock reader (CONSUMER THREAD)", this );
	LOCKED( r );
	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?
	qDebug( "%p: KDPipeIODevice::readData: waiting for bufferNotEmptyCondition (CONSUMER THREAD)", this );
	const TemporaryValue<bool> tmp( d->reader->consumerBlocksOnUs, true );
	r->bufferNotEmptyCondition.wait( &r->mutex );
	r->blockedConsumerIsDoneCondition.wakeAll();
	qDebug( "%p: KDPipeIODevice::readData: woke up from bufferNotEmptyCondition (CONSUMER THREAD)", this );
	}

	if ( r->bufferEmpty() ) {
	qDebug( "%p: KDPipeIODevice::readData: got empty buffer, signal eof", this );
	// woken with an empty buffer must mean either EOF or error:
	assert( r->eof \|\| r->error );
	r->eofShortCut = true;
	return r->eof ? 0 : -1 ;
	}

	qDebug( "%p: KDPipeIODevice::readData: got bufferNotEmptyCondition, trying to read %lld bytes", this, maxSize );
	const qint64 bytesRead = r->readData( data, maxSize );
	qDebug( "%p: KDPipeIODevice::readData: read %lld bytes", this, bytesRead );
	qDebug( "%p (fd=%d): KDPipeIODevice::readData: %s", this, d->fd, data );

	return bytesRead;
	}

	qint64 Reader::readData( char * data, qint64 maxSize ) {
	qint64 numRead = rptr < wptr ? wptr - rptr : sizeof buffer - rptr ;
	if ( numRead > maxSize )
	numRead = maxSize;

	qDebug( "%p: KDPipeIODevice::readData: data=%p, maxSize=%lld; rptr=%u, wptr=%u (bytesInBuffer=%u); -> numRead=%lld", this,
	data, maxSize, rptr, wptr, bytesInBuffer(), numRead );

	memcpy( data, buffer + rptr, numRead );

	rptr = ( rptr + numRead ) % sizeof buffer ;

	if ( !bufferFull() ) {
	qDebug( "%p: KDPipeIODevice::readData: signal bufferNotFullCondition", this );
	bufferNotFullCondition.wakeAll();
	}

	return numRead;
	}

	qint64 KDPipeIODevice::writeData( const char * data, qint64 size ) { KDAB_CHECK_THIS;
	d->startWriterThread();
	Writer * const w = d->writer;

	assert( w );
	assert( w->error \|\| w->isRunning() );
	assert( data \|\| size == 0 );
	assert( size >= 0 );

	LOCKED( w );

	while ( !w->error && !w->bufferEmpty() ) {
	qDebug( "%p: KDPipeIODevice::writeData: wait for empty buffer", this );
	w->bufferEmptyCondition.wait( &w->mutex );
	qDebug( "%p: KDPipeIODevice::writeData: empty buffer signaled", this );

	}
	if ( w->error )
	return -1;

	assert( w->bufferEmpty() );

	return w->writeData( data, size );
	}

	qint64 Writer::writeData( const char * data, qint64 size ) {
	assert( bufferEmpty() );

	if ( size > static_cast<qint64>( sizeof buffer ) )
	size = sizeof buffer;

	memcpy( buffer, data, size );

	numBytesInBuffer = size;

	if ( !bufferEmpty() ) {
	bufferNotEmptyCondition.wakeAll();
	}
	return size;
	}

	void KDPipeIODevice::Private::stopThreads()
	{
	if ( triedToStartWriter )
	{
	if ( writer && q->bytesToWrite() > 0 )
	q->waitForBytesWritten( -1 );

	assert( q->bytesToWrite() == 0 );
	}
	if ( Reader * & r = reader ) {
	disconnect( r, SIGNAL( readyRead() ), this, SLOT( emitReadyRead() ) );
	synchronized( r ) {
	// tell thread to cancel:
	r->cancel = true;
	// and wake it, so it can terminate:
	r->waitForCancelCondition.wakeAll();
	r->bufferNotFullCondition.wakeAll();
	r->readyReadSentCondition.wakeAll();
	}
	}
	if ( Writer * & w = writer ) {
	synchronized( w ) {
	// tell thread to cancel:
	w->cancel = true;
	// and wake it, so it can terminate:
	w->bufferNotEmptyCondition.wakeAll();
	}
	}
	}

	void KDPipeIODevice::close() { KDAB_CHECK_THIS;
	qDebug( "KDPipeIODevice::close(%p)", this );
	if ( !isOpen() )
	return;

	// tell clients we're about to close:
	emit aboutToClose();
	d->stopThreads();

	#define waitAndDelete( t ) if ( t ) { t->wait(); QThread* const t2 = t; t = 0; delete t2; }
	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();
	}
	waitAndDelete( d->reader );
	#undef waitAndDelete
	#ifdef Q_OS_WIN32
	if ( d->fd != -1 )
	_close( d->fd );
	else
	CloseHandle( d->handle );
	#else
	::close( d->fd );
	#endif

	setOpenMode( NotOpen );
	d->fd = -1;
	d->handle = 0;
	}

	void Reader::run() {

	LOCKED( this );

	// too bad QThread doesn't have that itself; a signal isn't enough
	hasStarted.wakeAll();

	qDebug( "%p: Reader::run: started", this );

	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();
	if ( !cancel && wasEmpty )
	waitForCancelCondition.wait( &mutex );
	} else if ( !cancel && !bufferFull() && !bufferEmpty() ) {
	qDebug( "%p: Reader::run: buffer no longer empty, waking everyone", this );
	notifyReadyRead();
	}

	while ( !cancel && !error && bufferFull() ) {
	notifyReadyRead();
	if ( !cancel && bufferFull() ) {
	qDebug( "%p: Reader::run: buffer is full, going to sleep", this );
	bufferNotFullCondition.wait( &mutex );
	}
	}

	if ( cancel ) {
	qDebug( "%p: Reader::run: detected cancel", this );
	goto leave;
	}

	if ( !eof && !error ) {
	if ( rptr == wptr ) // optimize for larger chunks in case the buffer is empty
	rptr = wptr = 0;

	unsigned int numBytes = ( rptr + sizeof buffer - wptr - 1 ) % sizeof buffer;
	if ( numBytes > sizeof buffer - wptr )
	numBytes = sizeof buffer - wptr;

	qDebug( "%p: Reader::run: rptr=%d, wptr=%d -> numBytes=%d", this, rptr, wptr, numBytes );

	assert( numBytes > 0 );

	qDebug( "%p: Reader::run: trying to read %d bytes", this, numBytes );
	#ifdef Q_OS_WIN32
	+ isReading = true;
	mutex.unlock();
	DWORD numRead;
	const bool ok = ReadFile( handle, buffer + wptr, numBytes, &numRead, 0 );
	mutex.lock();
	+ isReading = false;
	if ( ok ) {
	if ( numRead == 0 ) {
	qDebug( "%p: Reader::run: got eof (numRead==0)", this );
	eof = true;
	}
	} else { // !ok
	errorCode = static_cast<int>( GetLastError() );
	if ( errorCode == ERROR_BROKEN_PIPE ) {
	assert( numRead == 0 );
	qDebug( "%p: Reader::run: got eof (broken pipe)", this );
	eof = true;
	} else {
	assert( numRead == 0 );
	qDebug( "%p: Reader::run: got error: %s (%d)", this, strerror( errorCode ), errorCode );
	error = true;
	}
	}
	#else
	qint64 numRead;
	mutex.unlock();
	do {
	numRead = ::read( fd, buffer + wptr, numBytes );
	} while ( numRead == -1 && errno == EINTR );
	mutex.lock();

	if ( numRead < 0 ) {
	errorCode = errno;
	error = true;
	qDebug( "%p: Reader::run: got error: %d", this, errorCode );
	} else if ( numRead == 0 ) {
	qDebug( "%p: Reader::run: eof detected", this );
	eof = true;
	}
	#endif
	qDebug( "%p (fd=%d): Reader::run: read %ld bytes", this, fd, static_cast<long>(numRead) );
	qDebug( "%p (fd=%d): Reader::run: %s", this, fd, buffer );

	if ( numRead > 0 ) {
	qDebug( "%p: Reader::run: buffer before: rptr=%4d, wptr=%4d", this, rptr, wptr );
	wptr = ( wptr + numRead ) % sizeof buffer;
	qDebug( "%p: Reader::run: buffer after: rptr=%4d, wptr=%4d", this, rptr, wptr );
	}
	}
	}
	leave:
	qDebug( "%p: Reader::run: terminated", this );
	}

	void Reader::notifyReadyRead()
	{
	qDebug( "notifyReadyRead: %d bytes available", bytesInBuffer() );
	assert( !cancel );

	if ( consumerBlocksOnUs ) {
	bufferNotEmptyCondition.wakeAll();
	blockedConsumerIsDoneCondition.wait( &mutex );
	return;
	}
	qDebug( "notifyReadyRead: emit signal" );
	emit readyRead();
	readyReadSentCondition.wait( &mutex );
	qDebug( "notifyReadyRead: returning from waiting, leave" );
	}

	void Writer::run() {

	LOCKED( this );

	// too bad QThread doesn't have that itself; a signal isn't enough
	hasStarted.wakeAll();

	qDebug( "%p: Writer::run: started", this );

	while ( true ) {

	while ( !cancel && bufferEmpty() ) {
	qDebug( "%p: Writer::run: buffer is empty, wake bufferEmptyCond listeners", this );
	bufferEmptyCondition.wakeAll();
	emit bytesWritten( 0 );
	qDebug( "%p: Writer::run: buffer is empty, going to sleep", this );
	bufferNotEmptyCondition.wait( &mutex );
	qDebug( "%p: Writer::run: woke up", this );
	}

	if ( cancel ) {
	qDebug( "%p: Writer::run: detected cancel", this );
	goto leave;
	}

	assert( numBytesInBuffer > 0 );

	qDebug( "%p: Writer::run: Trying to write %u bytes", this, numBytesInBuffer );
	qint64 totalWritten = 0;
	do {
	mutex.unlock();
	#ifdef Q_OS_WIN32
	DWORD numWritten;
	qDebug( "%p (fd=%d): Writer::run: buffer before WriteFile (numBytes=%lld): %s:", this, fd, numBytesInBuffer, buffer );
	qDebug( "%p (fd=%d): Writer::run: Going into WriteFile", this, fd );
	if ( !WriteFile( handle, buffer + totalWritten, numBytesInBuffer - totalWritten, &numWritten, 0 ) ) {
	mutex.lock();
	errorCode = static_cast<int>( GetLastError() );
	qDebug( "%p: Writer::run: got error code: %d", this, errorCode );
	error = true;
	goto leave;
	}
	#else
	qint64 numWritten;
	do {
	numWritten = ::write( fd, buffer + totalWritten, numBytesInBuffer - totalWritten );
	} while ( numWritten == -1 && errno == EINTR );

	if ( numWritten < 0 ) {
	mutex.lock();
	errorCode = errno;
	qDebug( "%p: Writer::run: got error code: %s (%d)", this, strerror( errorCode ), errorCode );
	error = true;
	goto leave;
	}
	#endif
	qDebug( "%p (fd=%d): Writer::run: buffer after WriteFile (numBytes=%u): %s:", this, fd, numBytesInBuffer, buffer );
	totalWritten += numWritten;
	mutex.lock();
	} while ( totalWritten < numBytesInBuffer );

	qDebug( "%p: Writer::run: wrote %lld bytes", this, totalWritten );

	numBytesInBuffer = 0;

	qDebug( "%p: Writer::run: buffer is empty, wake bufferEmptyCond listeners", this );
	bufferEmptyCondition.wakeAll();
	emit bytesWritten( totalWritten );
	}
	leave:
	qDebug( "%p: Writer::run: terminating", this );
	numBytesInBuffer = 0;
	qDebug( "%p: Writer::run: buffer is empty, wake bufferEmptyCond listeners", this );
	bufferEmptyCondition.wakeAll();
	emit bytesWritten( 0 );
	}

	// static
	std::pair<KDPipeIODevice,KDPipeIODevice> KDPipeIODevice::makePairOfConnectedPipes() {
	KDPipeIODevice * read = 0;
	KDPipeIODevice * write = 0;
	#ifdef Q_OS_WIN32
	HANDLE rh;
	HANDLE wh;
	SECURITY_ATTRIBUTES sa;
	memset( &sa, 0, sizeof(sa) );
	sa.nLength = sizeof(sa);
	sa.bInheritHandle = TRUE;
	if ( CreatePipe( &rh, &wh, &sa, BUFFER_SIZE ) ) {
	read = new KDPipeIODevice;
	read->open( rh, ReadOnly );
	write = new KDPipeIODevice;
	write->open( wh, WriteOnly );
	}
	#else
	int fds[2];
	if ( pipe( fds ) == 0 ) {
	read = new KDPipeIODevice;
	read->open( fds[0], ReadOnly );
	write = new KDPipeIODevice;
	write->open( fds[1], WriteOnly );
	}
	#endif
	return std::make_pair( read, write );
	}

	#ifdef KDAB_DEFINE_CHECKS
	KDAB_DEFINE_CHECKS( KDPipeIODevice ) {
	if ( !isOpen() ) {
	assert( openMode() == NotOpen );
	assert( !d->reader );
	assert( !d->writer );
	#ifdef Q_OS_WIN32
	assert( !d->handle );
	#else
	assert( d->fd < 0 );
	#endif
	} else {
	assert( openMode() != NotOpen );
	assert( openMode() & ReadWrite );
	if ( openMode() & ReadOnly ) {
	assert( d->reader );
	synchronized( d->reader )
	assert( d->reader->eof \|\| d->reader->error \|\| d->reader->isRunning() );
	}
	if ( openMode() & WriteOnly ) {
	assert( d->writer );
	synchronized( d->writer )
	assert( d->writer->error \|\| d->writer->isRunning() );
	}
	#ifdef Q_OS_WIN32
	assert( d->handle );
	#else
	assert( d->fd >= 0 );
	#endif
	}
	}
	#endif // KDAB_DEFINE_CHECKS

	#include "moc_kdpipeiodevice.cpp"
	#include "kdpipeiodevice.moc"

File Metadata

Mime Type: text/x-diff
Expires: Mon, May 12, 6:18 PM (13 h, 26 m)
Storage Engine: local-disk
Storage Format: Raw Data
Storage Handle: 16/ca/33b367554e3c6dc03e4fe0d701a3

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions