aboutsummaryrefslogtreecommitdiffstats
path: root/src/kdpipeiodevice.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/kdpipeiodevice.cpp')
-rw-r--r--src/kdpipeiodevice.cpp951
1 files changed, 0 insertions, 951 deletions
diff --git a/src/kdpipeiodevice.cpp b/src/kdpipeiodevice.cpp
deleted file mode 100644
index 5661790a..00000000
--- a/src/kdpipeiodevice.cpp
+++ /dev/null
@@ -1,951 +0,0 @@
-/*
- Copyright (C) 2007 Klar�lvdalens 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 "kdpipeiodevice.h"
-
-#include <QtCore>
-
-#include <cassert>
-#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
-
-using namespace _gpgme_;
-
-#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;
-
-// comment to get trace output:
-//#define qDebug if(1){}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::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()
-{
- QPointer<Private> thisPointer( this );
- qDebug( "KDPipeIODevice::Private::emitReadyRead %p", this );
-
- emit q->readyRead();
-
- if ( !thisPointer )
- return;
-
- bool mustNotify = false;
-
- if ( reader ) {
- qDebug( "KDPipeIODevice::Private::emitReadyRead %p: locking reader (CONSUMER THREAD)", this );
- synchronized( reader ) {
- qDebug( "KDPipeIODevice::Private::emitReadyRead %p: locked reader (CONSUMER THREAD)", this );
- reader->readyReadSentCondition.wakeAll();
- mustNotify = !reader->bufferEmpty() && reader->isReading;
- qDebug( "KDPipeIODevice::emitReadyRead %p: bufferEmpty: %d reader in ReadFile: %d", this, reader->bufferEmpty(), reader->isReading );
- }
- }
- if ( mustNotify )
- QTimer::singleShot( 100, this, SLOT( emitReadyRead() ) );
- 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;
-
- return msecs >= 0 ? r->bufferNotEmptyCondition.wait( &r->mutex, msecs ) : r->bufferNotEmptyCondition.wait( &r->mutex );
-}
-
-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 );
-
- std::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;
-
- std::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: Reader::run: read %ld bytes", this, static_cast<long>(numRead) );
- qDebug( "%p: Reader::run(fd=%d): %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: %d", this, 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"
generated by cgit v1.2.3 (git 2.25.1) at 2025-04-29 14:40:52 +0000