QThread Worker Passing Large Quantities of Data from Stream to UI

AndyB

Hi everyone,

I found several related posts here and elsewhere discussing the best way to use QThreads and I believe the best way for my application here will be to use a worker class. Before showing some details, here's a little context. I am developing an acquisition system which needs to process a stream of data coming over USB in an efficient way. The data is packaged by a microcontroller / embedded system in a certain format, and the data contains several serialized channels, each with header info + channel data. The goal of the C++ program is to handle this stream as a master-slave connection, and when the C++ program receives enough data, to update a QCustomPlot in realtime. The most critical part here is to ensure that any data that is placed on the COM port's FIFO is serviced as soon as possible, otherwise the embedded system will stall and we may miss the next ADC sampling window. To accomplish this, here's what I'm currently trying to do...

I have a class called "Arduino" which is handling the connection and recording parameters. The mainwindow will create an instance of this class and populate the data members according to the GUI and user input. Then, when I want to start a recording, I call Arduino::startStream():

bool Arduino::startStream(){
    if(streamThread != nullptr)
        return false;
    //txBuffer = new chanData[1000*numChannels];
    streamThread = new QThread;
    streamWorker = new Worker();
    streamWorker->moveToThread(streamThread);
    //Setting up emitter function links for inter-thread synchronization/communication
    connect(streamThread,&QThread::started,streamWorker,&Worker::startStream);
    connect(streamWorker,&Worker::shareDataBatch,this,&Arduino::passDataBatch);
    connect(streamWorker,&Worker::memMapInitialized,this,&Arduino::wrFHeader);
    connect(streamWorker,&Worker::doneStreaming,streamThread,&QThread::quit);
    connect(streamWorker,&Worker::doneStreaming,streamWorker,&Worker::deleteLater);
    connect(streamThread,&QThread::finished,this,&Arduino::streamFinished);
    connect(streamThread,&QThread::finished,streamThread,&QThread::deleteLater);

    streamWorker->updateMembers(isRecording, numChannels, batchSize, &hSerial, &status, &params);
    //streamWorker->isRecording = isRecording;

    //QMessageBox messageBox;
    updateSerialWrBuffers();//update ParamBuff
    if(isRecording){
        if(writeData(recParamBuff,sizeof(recParamBuff)/sizeof(char))){
            //To do: error reporting / handling
        }
        else{
            return false;
        }
    }
    else{
        if(writeData(runParamBuff,sizeof(runParamBuff)/sizeof(char))){
            //To do: error reporting / handling
        }
        else{
            return false;
        }
    }
    streamThread->start(QThread::HighestPriority);
    currentlyRunning = true;
    return true;
}

Noting that streamThread is a private QThread* that is initialized with value nullptr, and streamWorker is also a private data member in Arduino class of type Worker* initialized with nullptr. The Worker class is a custom worker class with the below cpp file

#include "worker.h"

Worker::Worker(QObject *parent)
    : QObject{parent}
{

}

void Worker::updateMembers(bool _isRecording, uint16_t _numChannels, uint16_t _batchSize, HANDLE* _hSerial, COMSTAT* _status, boost::iostreams::mapped_file_params *_params){
    numChannels = _numChannels;
    batchSize = _batchSize;
    hSerial = _hSerial;
    status = _status;
    params = _params;
    isRecording = _isRecording;
}

void Worker::packageBuffer(unsigned char* dataBuff, uint32_t numFrames){
    if(runningSumNumFrames+numFrames > 896){
        emit shareDataBatch(txBuffer, runningSumNumFrames);
        runningSumNumFrames = 0;
    }
    memcpy(txBuffer + (runningSumNumFrames*numChannels*pixelByteSize),dataBuff,numFrames*numChannels*pixelByteSize);
    runningSumNumFrames += numFrames;
}

void Worker::transmitBuff(){
    emit shareDataBatch(txBuffer, runningSumNumFrames);
    runningSumNumFrames = 0;
}

//Read a batch of data at a time. transmitBuff is called by timer, so we only
//will have issues if the timer interval is shorter than the transmitBuff() time.
//To Do: Consider mutex or some other form of locking to prevent memory corruption etc..
void Worker::readLoop(){
    const uint16_t frameByteSize = numChannels*pixelByteSize;
    //const uint16_t buffSize = batchSize*frameByteSize;
    uint32_t bytesToRead, numFrames = 1;
    //unsigned char *txBuff = &frame[0];
    DWORD bytesRead;
    //DWORD dwCommModemStatus;
    DWORD errors;

    while(currentlyStreaming){
        if(*hSerial != INVALID_HANDLE_VALUE){
            ClearCommError(*hSerial, &errors, status);//update status
            bytesToRead = status->cbInQue;
            if(bytesToRead > frameByteSize){
                bytesToRead -= (bytesToRead % frameByteSize);//ensures we read full frames without residuals
                numFrames = bytesToRead/frameByteSize;
                if(numFrames > 128){
                    numFrames = 128;
                    bytesToRead = numFrames*frameByteSize;
                }
                /*if(frames != nullptr)
                    delete[] frames;
                frames = new unsigned char[bytesToRead];*/
                //frames.resize(numChannels*maxBatchNumber*pixelByteSize);

                ReadFile(*hSerial, frames, bytesToRead, &bytesRead, 0);

                //Empty out the serial FIFO
                //while(bytesToRead > frameByteSize){

                //bytesToRead -= bytesRead;
                if(isRecording){
                    //Check if we didn't read a full buffer of data. To Do: emit error signal here
                    //if(bytesRead < frameByteSize){
                        //Error
                    //}
                    //copy over the newly captured buffer to memory map then increment pointer
                    memcpy(frames,recFMPtr,bytesToRead);
                    recFMPtr += bytesToRead;//To Do: check if we are within memory map pre-allocated space
                }
                emit addToBuffer(frames,numFrames);
                    //QCoreApplication::processEvents();
                //}
            }
        }
        //QCoreApplication::processEvents();
    }

    //if(!currentlyStreaming){
    //If we enter this conditional, that means we want to stop the stream
    //Note: before setting this bool false, the arduino class needs to also
    //write a 'T' character to the teensy to ensure that data stops streaming
    //So we are assuming that this is the case here, and what needs to be done now
    //is to read all remaining data posted to the serial port, turn the timer off, etc..
    Sleep(100);
    if(*hSerial != INVALID_HANDLE_VALUE){
        //check how much data is left to read from serial port
        ClearCommError(*hSerial, &errors, status);//update status
        bytesToRead = status->cbInQue;
        if(bytesToRead > 0){
            if(isRecording){
                ReadFile(*hSerial, recFMPtr, bytesToRead, &bytesRead, 0);
                //copy over the newly captured buffer to memory map then increment pointer
                recFMPtr += bytesRead;//To Do: check if we are within memory map pre-allocated space
            }
            else{
                unsigned char *throwaway[bytesToRead];
                ReadFile(*hSerial, throwaway, bytesToRead, &bytesRead, 0);
            }
        }
    }
    endStream();
    //}
}

//Greatest Common Divisor helper function
uint32_t Worker::gcd(uint32_t a, uint32_t b) {
   if (b == 0)
   return a;
   return gcd(b, a % b);
}


void Worker::startStream(){
    //Allocate resources using new here, that way the resources are on the new thread

    //Creating new Timer that will periodically call transmitBuff() from
    //teensy4_1 each time the timer times out. When we want to stop reading data, we
    //need to end the timer thread.
    //frames = new QVector<QVector<chanData>>;
    //frames.resize(numChannels*maxBatchNumber*pixelByteSize);
    frames = new unsigned char[128*numChannels*pixelByteSize];
    txBuffer = new unsigned char[1024*numChannels*pixelByteSize];
    runningSumNumFrames = 0;
    streamTimer = new QTimer;
    streamTimer->setInterval(15);//Interval set to 15ms
    currentlyStreaming = true;
    //streamTimer->moveToThread(timerThread);
    //connect(timerThread, SIGNAL(started()),streamTimer, SLOT(start()));
    //connect(timerThread, SIGNAL(finished()),streamTimer, SLOT(deleteLater()));
    connect(streamTimer, &QTimer::timeout,this, &Worker::transmitBuff);
    connect(this, &Worker::addToBuffer, this, &Worker::packageBuffer);

    if(isRecording){
        mf = new boost::iostreams::mapped_file;
        mf->open(*params);
        recordFileMemory = (unsigned char*) mf->data();
        //first write file header by emitting the memory map address
        emit memMapInitialized(recordFileMemory);
        //Then update our local pointer which traverses our memory map
        recFMPtr = recordFileMemory + numTotalBytesInHeader;
    }
    else{

    }
    streamTimer->start();
    readLoop();
}

void Worker::endStream(){
    if(isRecording){
        mf->close();
        delete mf;
    }
    streamTimer->stop();
    delete streamTimer;
    delete[] frames;
    delete[] txBuffer;
    emit doneStreaming();
}

So what happens here is that the mainwindow will call startStream() on the Arduino class, and this will then create an instance of a worker and a qthread, and place the worker on this new thread. The worker then will create a timer and then enter an event loop which will rapidly pull data off the serial port as quickly as it can. Then each time the timer tics, it will emit the buffer of data it has accumulated up to the arduino class. This arduino class is setup to just echo this to another emit signal which the mainwindow class will see and process in its own slot:

void MainWindow::updateDataBlock(unsigned char *dataStream, uint32_t numFrames){
    int16_t tmpADCVal, tmpDACVal;
    uint32_t tmpFrameTimeVals;
    //Note that we read data blocks with length larger than 12 bytes, so we'll have multiple data points here to parse
    //Each channel has first 8 bytes are frame time and number, followed by 2 bytes for ADC and 2 bytes for DAC
    //QVector<chanData> frames;
    //frames.resize(teensy4_1->numChannels*numFrames*pixelByteSize);
    int fSize = teensy4_1->numChannels*numFrames*pixelByteSize;
    char frames[fSize];
    memcpy(frames,dataStream,fSize);

//Code to parse data and update plots...

Here are my questions:

Does this architecture make sense? Fundamentally, the idea here is to create a new instance of a worker class and qthread each time we want to start a new recording session. My thought was that this would allow for the GUI to not hang while we are recording, and by having the timer and the USB reading eventloop, I thought I could decouple the serial port servicing from the buffer emitting / plot updater. Is my architecture accomplishing this in a sensible way? It seems maybe not, because I am experiencing lag and other bugs.
When emitting large quantities of data, I believe you can just emit a pointer to the buffer (in this case, a character array). I have seen some forums that discuss the potential for a deep copy of the data in certain circumstances - obviously I don't want to have to perform more copies than necessary. Can someone take a look at this code and let me know whether or not this is a concern here?
Is the act of using an intermediate "echo" slot/signal between the mainwindow class and the worker class silly? I couldn't think of a simpler way to get the data from the worker class all the way up to the GUI for plotting other than redesigning the architecture to have the arduino class itself be a worker class with its own thread. I had done this before but this architecture has its own drawbacks as well.

Any info would be greatly appreciated! Please let me know if anything needs clarification.

Thanks everyone,

Andy

AndyB

Got it! Thanks for your help @Pl45m4

For those who are interested in this post, I ended up doing something slightly different here. I added a QQueue<unsigned char> data member in my Arduino class, and now, in the signal for the worker thread, I lock a shared mutex with the arduino object's thread. Then, in the slot which handles the stream of data coming in from the worker thread, I enqueue this data and then unlock the mutex. This way, I ensure that the data never gets jumbled up, and it also ensures proper memory access without risk.

Pl45m4

@AndyB said in QThread Worker Passing Large Quantities of Data from Stream to UI:

Is the act of using an intermediate "echo" slot/signal between the mainwindow class and the worker class silly? I couldn't think of a simpler way to get the data from the worker class all the way up to the GUI for plotting other than redesigning the architecture to have the arduino class itself be a worker class with its own thread. I had done this before but this architecture has its own drawbacks as well.

Hi, just a comment on [3.]:

connect(streamWorker, &Worker::shareDataBatch, this, &Arduino::passDataBatch);

I presume this is the "echo" connection you are speaking of, right?!
If Arduino::passDataBatch does nothing else than emitting another signal for MainWindow to get the data, you can implement a signal-forwarding to save (at least) one function and some LoC :)
Not much, but might be some improvement.
OR you find a way to connect the Worker directly to MainWindow, but I think this is not what you want as it might requires some re-design.

Signal-Forwarding:

In pseudocode:

class Worker (knows nothing by design)
- signal: void shareDataBatch( /*yourdata*/ )
class Arduino (knows about Worker, not about MainWindow)
- signal: void updateGUI( /*yourdata*/ )
  - connect(streamWorker, &Worker::shareDataBatch, this, &Arduino::updateGUI); // forwarding here
class MainWindow (knows Arduino, but cannot see the Worker)
- function/slot: void updateDataBlock( /*yourdata*/ )
  - connect(arduino, &Arduino::updateGUI, this, &MainWindow::updateDataBlock);

One more thing:

streamTimer = new QTimer;

If you make streamTimer a child of Worker ("this"), what you can do with all QObject classes in there, since in startStream() the Worker has been moved and changed its thread-affinity already...

delete streamTimer;

...you don't need to worry about the deletion.

Christian Ehrlicher

@AndyB said in QThread Worker Passing Large Quantities of Data from Stream to UI:

emit shareDataBatch(txBuffer, runningSumNumFrames);

You emit a signal with a raw pointer to some memory and access this in another thread? This will not work out. Use a proper container like e.g. QVector and make sure it does not detach (which is easy by making them const on usage).

AndyB

@Pl45m4

Thanks for your reply! Yes, when I say "echo", I do in fact mean the signal/slot for the arduino class relating the shareDataBatch and passDataBatch, as you quoted. Can you clarify what you mean by implementing a "signal-forwarding" scheme? Is that not what this signal/slot mechanism is doing already? Or is there another means to do this that I am unaware of? From your pseudocode, that's basically exactly what I'm doing I think.

I thought about trying to connect the worker directly to MainWindow (that's what I was doing previously, basically, but it caused other issues).

Thanks for pointing out the timer things too! That was helpful :)

@Christian-Ehrlicher

Hi Christian,

Thanks for your reply! Can you clarify how using a QVector would improve things, and what is wrong with passing a pointer to memory like this? I think what you're alluding to us that I should be passing by reference rather than using pointers which can detach, is that right?

Thanks again guys!

Pl45m4

@AndyB said in QThread Worker Passing Large Quantities of Data from Stream to UI:

Can you clarify what you mean by implementing a "signal-forwarding" scheme?

I mean what I wrote here :)

@Pl45m4 said in QThread Worker Passing Large Quantities of Data from Stream to UI:

If Arduino::passDataBatch does nothing else than emitting another signal for MainWindow to get the data, you can implement a signal-forwarding to save (at least) one function and some LoC :)
Not much, but might be some improvement.

@AndyB said in QThread Worker Passing Large Quantities of Data from Stream to UI:

Is that not what this signal/slot mechanism is doing already? Or is there another means to do this that I am unaware of? From your pseudocode, that's basically exactly what I'm doing I think.

If you forward the signal directly, you can remove that one function and one connection from your Arduino class.

Check my pseudocode "template" again.
(I've highlighted the important part now)

AndyB

Got it! Thanks for your help @Pl45m4

For those who are interested in this post, I ended up doing something slightly different here. I added a QQueue<unsigned char> data member in my Arduino class, and now, in the signal for the worker thread, I lock a shared mutex with the arduino object's thread. Then, in the slot which handles the stream of data coming in from the worker thread, I enqueue this data and then unlock the mutex. This way, I ensure that the data never gets jumbled up, and it also ensures proper memory access without risk.