pelican/examples/example-2/server/StreamChunker.cpp
/* Function that is called whenever there is data ready to be processed.
 * This function should block until all data for the chunk has been read from
 * the device.
 */
Function that is called whenever there is data ready to be processed.
This function should block until all data for the chunk has been read from
the device.
Function that is called whenever there is data ready to be processed . This function should block until all data for the chunk has been read from the device .
void StreamChunker::next(QIODevice* device)
{
    QTcpSocket* socket = static_cast<QTcpSocket*>(device);
    quint64 totalBytesRead = 0;
    quint64 bytesRead = 0;
    quint32 packetSize = 0;
    quint32 packetCounter = 0;
    quint32 packetTime = 0;
    qint32 numPackets = 0;
    qint32 reportInterval = 0;
    size_t headerSize = 3*sizeof(quint32) + 2*sizeof(qint32);
    while (isActive() && totalBytesRead != headerSize)
    {
        while (socket->bytesAvailable() < (qint64)(headerSize)) {
            socket->waitForReadyRead(-1);
        }
        bytesRead = socket->read((char*)&packetSize, (qint64)sizeof(quint32));
        totalBytesRead+=bytesRead;
        bytesRead = socket->read((char*)&packetCounter, (qint64)sizeof(quint32));
        totalBytesRead+=bytesRead;
        bytesRead = socket->read((char*)&packetTime, (qint64)sizeof(quint32));
        totalBytesRead+=bytesRead;
        bytesRead = socket->read((char*)&numPackets, (qint64)sizeof(qint32));
        totalBytesRead+=bytesRead;
        bytesRead = socket->read((char*)&reportInterval, (qint64)sizeof(qint32));
        totalBytesRead+=bytesRead;
    }
     Writable chunks are obtained using the getDataStorage() method
     using the the following priority:
     1) A Pre-allocated chunk in the buffer that has already been served
        and has been marked for reuse.
     2) Allocating a new chunk in the buffer if space allows.
     3) Overwriting the oldest chunk in the buffer that matches the space
        requirements.
     If none of these conditions can be met, an invalid chunk is returned.
     If there are no usable chunks, i.e. the buffer is full we will
     be overwriting chunks.
#if 1
    if (numUsableChunks(packetSize) == 0) {
        ++totalOverwriteCounter_;
        ++intervalOverwriteCounter_;
    }
#endif
     Ask the data manager for a writable data chunk and get its data pointer.
    pelican::WritableData chunk = getDataStorage(packetSize);
     Check chunk->isValid() before proceeding!
    if (!chunk.isValid()) {
        QString error = "StreamChunker::next(): Unable to get a valid chunk. ";
        if (packetSize > maxChunkSize()) {
            error += QString(
                    "\n== The requested packet size is greater than the maximum"
                    "\n== chunk size allowed by the buffer.");
        }
        throw error;
    }
     Extract the data pointer from the chunk
    char* chunkPtr = (char*)chunk.ptr();
     Write the header into the chunk.
    quint32* header1 = reinterpret_cast<quint32*>(chunkPtr);
    header1[0] = packetSize;
    header1[1] = packetCounter;
    header1[2] = packetTime;
    qint32* header2 = reinterpret_cast<qint32*>(chunkPtr+3*sizeof(quint32));
    header2[0] = numPackets;
    header2[1] = reportInterval;
     NOTE Setting the read buffer size may have an impact on performance.
#if 0
    socket->setReadBufferSize(2*1024*1024);
#endif
     NOTE The minimum read size has an impact on performance and has not
     been optimised.
    quint64 minReadSize = 1024;2*1024*1024;  bytes
     Read data block directly into the chunk.
    quint64 dataRemaining = packetSize - headerSize;
    while (isActive() && dataRemaining > 0)
    {
        if (dataRemaining < minReadSize) minReadSize = dataRemaining;
        while (socket->bytesAvailable() < (qint64)minReadSize) {
            socket->waitForReadyRead(-1);
        }
        bytesRead = socket->read(chunkPtr+totalBytesRead, dataRemaining);
        totalBytesRead+=bytesRead;
        dataRemaining-=bytesRead;
    }
    chunkCounter_++;
     Report performance
    if (chunkCounter_%reportInterval == 0)
    {
        const double B2MiB = 1.0/(1024.0*1024.0);
        int elapsed = timer_.elapsed();
        size_t dataReceived = packetSize * reportInterval;
        size_t maxBufferSize_ = maxBufferSize();
        size_t allocatedSize_ = allocatedSize();
        size_t usedSize_ = usedSize();
        size_t usableSize_ = usableSize(packetSize);
         Buffer % full (how much of the buffer is in use as a %)
        double pBufferFull = ((maxBufferSize_-usableSize_)/(double)maxBufferSize_)*100.0;
        char prefix[10];
        int l = (reportCounter_ < 10) ? 2 : ((reportCounter_ < 100) ? 1 : 0);
        sprintf(prefix, "S[%s%llu] ", string(l, '-').c_str(), reportCounter_);
        printf("%s\n", string(80,'*').c_str());
        printf("%sTotal chunk count    = %llu\n", prefix, chunkCounter_);
        printf("%sChunk size           = %-7.3f MiB [%i B]\n", prefix,
                packetSize*B2MiB, packetSize);
        printf("%sTotal data received  = %.1f MiB\n", prefix,
                (quint64)packetSize*chunkCounter_*B2MiB);
        printf("%s\n", prefix);
        printf("%sBuffer state:\n", prefix);
        printf("%s* Percent full       = %.1f%%\n", prefix, pBufferFull);
        printf("%s* Total size         = %-7.3f MiB [%lu B]\n", prefix,
                maxBufferSize_*B2MiB, maxBufferSize_);
        printf("%s* Allocated size     = %-7.3f MiB [%lu B]\n", prefix,
                allocatedSize_*B2MiB, allocatedSize_);
        printf("%s* In use (active)    = %-7.3f MiB [%lu B]\n", prefix,
                usedSize_*B2MiB, usedSize_);
        printf("%s* Usable space       = %-7.3f MiB [%lu B]\n", prefix,
                usableSize_*B2MiB, usableSize_);
        printf("%s* Total chunks       = %i\n", prefix, numChunks());
        printf("%s* Active chunks      = %i\n", prefix, numActiveChunks());
        printf("%s* Expired chunks     = %i\n", prefix, numExpiredChunks());
        printf("%s* Usable chunks      = %i\n", prefix, numUsableChunks(packetSize));
        printf("%s* Overwritten chunks = %llu\n", prefix, totalOverwriteCounter_);
        printf("%s* Overwritten chunks = %i, %.1f%% (in report interval)\n", prefix,
                intervalOverwriteCounter_,
                (double)intervalOverwriteCounter_/reportInterval*100.0);
        printf("%s\n", prefix);
        printf("%sReport interval:\n", prefix);
        printf("%s* Chunks received    = %i\n", prefix, reportInterval);
        printf("%s* Data received      = %-7.3f MiB [%lu B] \n", prefix,
                dataReceived*B2MiB, dataReceived);
        printf("%s* Time taken         = %.3f s\n", prefix, elapsed*1.0e-3);
        printf("%s* Data rate          = ", prefix);
        if (elapsed > 0)
            printf("%-7.3f MiB/s\n", dataReceived*B2MiB/(elapsed*1.0e-3));
        else
            printf("---     MiB/s\n");
        printf("%s\n\n", string(80,'*').c_str());
        fflush(stdout);
        reportCounter_++;
        intervalOverwriteCounter_ = 0;
        timer_.restart();
    }
}
void StreamChunker :: next ( QIODevice * device ) { QTcpSocket * socket = static_cast < QTcpSocket * > ( device ) ; quint64 totalBytesRead = 0 ; quint64 bytesRead = 0 ; quint32 packetSize = 0 ; quint32 packetCounter = 0 ; quint32 packetTime = 0 ; qint32 numPackets = 0 ; qint32 reportInterval = 0 ; size_t headerSize = 3 * sizeof ( quint32 ) + 2 * sizeof ( qint32 ) ; while ( isActive ( ) && totalBytesRead != headerSize ) { while ( socket -> bytesAvailable ( ) < ( qint64 ) ( headerSize ) ) { socket -> waitForReadyRead ( -1 ) ; } bytesRead = socket -> read ( ( char * ) & packetSize , ( qint64 ) sizeof ( quint32 ) ) ; totalBytesRead += bytesRead ; bytesRead = socket -> read ( ( char * ) & packetCounter , ( qint64 ) sizeof ( quint32 ) ) ; totalBytesRead += bytesRead ; bytesRead = socket -> read ( ( char * ) & packetTime , ( qint64 ) sizeof ( quint32 ) ) ; totalBytesRead += bytesRead ; bytesRead = socket -> read ( ( char * ) & numPackets , ( qint64 ) sizeof ( qint32 ) ) ; totalBytesRead += bytesRead ; bytesRead = socket -> read ( ( char * ) & reportInterval , ( qint64 ) sizeof ( qint32 ) ) ; totalBytesRead += bytesRead ; } #if 1 if ( numUsableChunks ( packetSize ) == 0 ) { ++ totalOverwriteCounter_ ; ++ intervalOverwriteCounter_ ; } #endif pelican :: WritableData chunk = getDataStorage ( packetSize ) ; if ( ! chunk . isValid ( ) ) { QString error = " " ; if ( packetSize > maxChunkSize ( ) ) { error += QString ( " \n " " \n " ) ; } throw error ; } char * chunkPtr = ( char * ) chunk . ptr ( ) ; quint32 * header1 = reinterpret_cast < quint32 * > ( chunkPtr ) ; header1 [ 0 ] = packetSize ; header1 [ 1 ] = packetCounter ; header1 [ 2 ] = packetTime ; qint32 * header2 = reinterpret_cast < qint32 * > ( chunkPtr + 3 * sizeof ( quint32 ) ) ; header2 [ 0 ] = numPackets ; header2 [ 1 ] = reportInterval ; #if 0 socket -> setReadBufferSize ( 2 * 1024 * 1024 ) ; #endif quint64 minReadSize = 1024 ; quint64 dataRemaining = packetSize - headerSize ; while ( isActive ( ) && dataRemaining > 0 ) { if ( dataRemaining < minReadSize ) minReadSize = dataRemaining ; while ( socket -> bytesAvailable ( ) < ( qint64 ) minReadSize ) { socket -> waitForReadyRead ( -1 ) ; } bytesRead = socket -> read ( chunkPtr + totalBytesRead , dataRemaining ) ; totalBytesRead += bytesRead ; dataRemaining -= bytesRead ; } chunkCounter_ ++ ; if ( chunkCounter_ % reportInterval == 0 ) { const double B2MiB = 1.0 / ( 1024.0 * 1024.0 ) ; int elapsed = timer_ . elapsed ( ) ; size_t dataReceived = packetSize * reportInterval ; size_t maxBufferSize_ = maxBufferSize ( ) ; size_t allocatedSize_ = allocatedSize ( ) ; size_t usedSize_ = usedSize ( ) ; size_t usableSize_ = usableSize ( packetSize ) ; double pBufferFull = ( ( maxBufferSize_ - usableSize_ ) / ( double ) maxBufferSize_ ) * 100.0 ; char prefix [ 10 ] ; int l = ( reportCounter_ < 10 ) ? 2 : ( ( reportCounter_ < 100 ) ? 1 : 0 ) ; sprintf ( prefix , " " , string ( l , ' ' ) . c_str ( ) , reportCounter_ ) ; printf ( " \n " , string ( 80 , ' ' ) . c_str ( ) ) ; printf ( " \n " , prefix , chunkCounter_ ) ; printf ( " \n " , prefix , packetSize * B2MiB , packetSize ) ; printf ( " \n " , prefix , ( quint64 ) packetSize * chunkCounter_ * B2MiB ) ; printf ( " \n " , prefix ) ; printf ( " \n " , prefix ) ; printf ( " \n " , prefix , pBufferFull ) ; printf ( " \n " , prefix , maxBufferSize_ * B2MiB , maxBufferSize_ ) ; printf ( " \n " , prefix , allocatedSize_ * B2MiB , allocatedSize_ ) ; printf ( " \n " , prefix , usedSize_ * B2MiB , usedSize_ ) ; printf ( " \n " , prefix , usableSize_ * B2MiB , usableSize_ ) ; printf ( " \n " , prefix , numChunks ( ) ) ; printf ( " \n " , prefix , numActiveChunks ( ) ) ; printf ( " \n " , prefix , numExpiredChunks ( ) ) ; printf ( " \n " , prefix , numUsableChunks ( packetSize ) ) ; printf ( " \n " , prefix , totalOverwriteCounter_ ) ; printf ( " \n " , prefix , intervalOverwriteCounter_ , ( double ) intervalOverwriteCounter_ / reportInterval * 100.0 ) ; printf ( " \n " , prefix ) ; printf ( " \n " , prefix ) ; printf ( " \n " , prefix , reportInterval ) ; printf ( " \n " , prefix , dataReceived * B2MiB , dataReceived ) ; printf ( " \n " , prefix , elapsed * 1.0e-3 ) ; printf ( " " , prefix ) ; if ( elapsed > 0 ) printf ( " \n " , dataReceived * B2MiB / ( elapsed * 1.0e-3 ) ) ; else printf ( " \n " ) ; printf ( " \n \n " , string ( 80 , ' ' ) . c_str ( ) ) ; fflush ( stdout ) ; reportCounter_ ++ ; intervalOverwriteCounter_ = 0 ; timer_ . restart ( ) ; } }
Function that is called whenever there is data ready to be processed.
Function that is called whenever there is data ready to be processed .
// Note: // The next() function is slowed down initially by allocating the // data chunks. // It is probably best therefore, when running the benchmark, to make sure // all chunks are allocated before the timing starts. // Header values. // Read header values. // // Writable chunks are obtained using the getDataStorage() method // using the the following priority: // // 1) A Pre-allocated chunk in the buffer that has already been served //    and has been marked for reuse. // 2) Allocating a new chunk in the buffer if space allows. // 3) Overwriting the oldest chunk in the buffer that matches the space //    requirements. // // If none of these conditions can be met, an invalid chunk is returned. // // If there are no usable chunks, i.e. the buffer is full we will // be overwriting chunks. // Ask the data manager for a writable data chunk and get its data pointer. // Check chunk->isValid() before proceeding! // Extract the data pointer from the chunk // Write the header into the chunk. // NOTE Setting the read buffer size may have an impact on performance. // NOTE The minimum read size has an impact on performance and has not // been optimised. //2*1024*1024; // bytes // Read data block directly into the chunk. // Report performance // Buffer % full (how much of the buffer is in use as a %)