Seeking Suggestions for Multi-Threaded Application Design

[swamyonline]

Hi,
I seek your suggestions for my multi-threaded application design. The key characteristics of the application are as follow:

1. Data reception on TCP socket from 10-15 sources (channels).
2. Data processing for each channel - just extract data from raw bytes and assemble them into meaningful structures.
3. Displaying data on Main Window with different sub-windows in different formats( text & graphical).
4. Sending processed data to other systems on TCP/ UDP sockets.
5. The display to be updated every 1 second.

My proposal for the above:
1. Attach QDataStream for each channel (socket) for easy access and manipulation of data.
2. Worker thread for each data channel for processing.
3. "After processing, storing the processed data into saperate QDataStream objects, again for easy access of data by GUI thread" or "Use of signals/ slots in threads for data communication with Main Window".

My questions are:
1. Which threading mechanism should I adapt for better utilization of multi-core CPU (4th generation i7 with 8GB of RAM)?
2. Which mechanism to adapt for storing data received on TCP sockets, for easy access and processing by other threads?
3. Is MVC mechanism suitable (in this case) for data-control as it is being accessed and utilized by other windows. Sub-Windows also include tabular displays.


I hope I have conveyed my requirements.

Please suggest your points, they matter a lot in my design.

Thanks in advance.

swamy.

[wysota]

There is no need to use threading for anything in the posted list. Unless you expect to be satitating the link and transfering hundreds of megabits per second you can handle all the data in the GUI thread without complicating your code with needless serializing. If after reading the data you wish to process it using threads then that's just of course fine and QtConcurrent::run() is one of the approaches you can take.

[anda_skoa]

3. "After processing, storing the processed data into saperate QDataStream objects, again for easy access of data by GUI thread" or "Use of signals/ slots in threads for data communication with Main Window".

Definitely the latter. Serializing and deserializing the data again just for communication within the process would be a waste of processing time.

I am with wysota on the multithreading thing: start with a single threaded version. If the networking or processing becomes too much for the main thread, move that to a thread.

Cheers,
_

[swamyonline]

Hi,
Sorry for late reply as there was no access of Internet for me. Thanks "wysota" and "anda_skoa" for your suggestions.
And any suggestions for handling socket's data: for storing (after reading from socket) and making the same data available for processing threads? I have attached QDataStream variable to QTcpSocket variable in 'DataReceiver' class and accessing the same variable in 'DataProcessorThread' class.

It is real time data reception, processing and displaying. You can understand my worry for efficiency. .

Regards.
swamy.

[anda_skoa]

When you write processing threads, do you mean you have more than one thread process the same data from one socket?

Cheers,
_

[swamyonline]

NO, only one thread. I am reading data from socket and processing the 'read data' in a 'processing thread'. Each socket is associated with a processing thread.

// Sample Code Snippet - Not Complete

// Data Receiving Class

class DataClient: public QObject
{
public:
    DataClient();
    QTcpSocket m_tcpSocket;
    QDataStream m_dstreamData;        // To be associated with socket for data access  
    void slotReadData();
};
DataClient::DataClient()
{
    // Setup Network Session and Establish Socket Connection
    .... 
    m_dstreamData.setDevice(m_tcpSocket);
    m_dstreamData.setVersion(QDataStream::Qt_4_0);
   // Setup Handlers for Reading Data from Socket
   connect(m_tcpSocket, SIGNAL(readyRead()), this, SLOT(slotReadData()));
   ....
}
void DataClient::slotReadData()
{
     qint64 nBytesAvailable = m_tcpSocket->bytesAvailable();
     if (nBytesAvailable < (int)sizeof(quint16))  // Sufficient Data NOT Arrived
     {
         return;
     }
     // Any other Data Checking Statements
     ...
}

class DataClient: public QObject
{
public:
    DataClient();
    QTcpSocket m_tcpSocket;
    QDataStream m_dstreamData;        // To be associated with socket for data access  
    void slotReadData();
};
DataClient::DataClient()
{
    // Setup Network Session and Establish Socket Connection
    .... 
    m_dstreamData.setDevice(m_tcpSocket);
    m_dstreamData.setVersion(QDataStream::Qt_4_0);
   // Setup Handlers for Reading Data from Socket
   connect(m_tcpSocket, SIGNAL(readyRead()), this, SLOT(slotReadData()));
   ....
}
void DataClient::slotReadData()
{
     qint64 nBytesAvailable = m_tcpSocket->bytesAvailable();
     if (nBytesAvailable < (int)sizeof(quint16))  // Sufficient Data NOT Arrived
     {
         return;
     }
     // Any other Data Checking Statements
     ...
}

To copy to clipboard, switch view to plain text mode 

// Data Processing Thread Class

class DataClient;    // Forward Declaration
class DataProcessor: public QThread
{
public:
     DataProcessor();
     DataClient *m_pDataClient;
     bool m_bStopped;
     void run();
};
DataProcessor::DataProcessor()
{
     // Initialization 
     ....
    m_pDataClient = new DataClient;
    m_bStopped = false;
    .....
}
void DataProcessor: run()
{
     while(!m_bStopped)
     {
          // Access DataClient's Socket for Data and Process
          m_pDataClient->m_dstreamData >> dataItem1;
          m_pDataClient->m_dstreamData >> dataItem2;
          .....
          .....
          m_pDataClient->m_dstreamData >> dataItemn;
 
          // Process the Data
          .....
          // Publish Processed Data to MainWindow and other Objects by Emitting Signals
     }
}

class DataClient;    // Forward Declaration
class DataProcessor: public QThread
{
public:
     DataProcessor();
     DataClient *m_pDataClient;
     bool m_bStopped;
     void run();
};
DataProcessor::DataProcessor()
{
     // Initialization 
     ....
    m_pDataClient = new DataClient;
    m_bStopped = false;
    .....
}
void DataProcessor: run()
{
     while(!m_bStopped)
     {
          // Access DataClient's Socket for Data and Process
          m_pDataClient->m_dstreamData >> dataItem1;
          m_pDataClient->m_dstreamData >> dataItem2;
          .....
          .....
          m_pDataClient->m_dstreamData >> dataItemn;

          // Process the Data
          .....
          // Publish Processed Data to MainWindow and other Objects by Emitting Signals
     }
}

To copy to clipboard, switch view to plain text mode 

So each socket (around 10 sockets for 10 data channels) is associated with a processing thread.

Does the above scheme work fine?

swamy.

[anda_skoa]

You'll have to decide if you want to do blocking I/O or asynchronous I/O.
Currently you are mixing both.

I would suggest asynchronous I/O, i.e. the code that you have in DataClient. To make that work properly don't reimplement the thread's run() method but let the base implementation run the thread's event loop.

Something more like this

class DataClient : public QObject
{
    Q_OBJECT
 
signals:
    void dataReady();
 
private slots:
    void slotReadData()
    {
         qint64 nBytesAvailable = m_tcpSocket->bytesAvailable();
         if (nBytesAvailable < (int)sizeof(quint16))  // Sufficient Data NOT Arrived
         {
             return;
         }
         // Any other Data Checking Statements
         ...
 
        emit dataReady();
    }
};
 
class DataProcessor : public QObject
{
    Q_OBJECT
 
public:
    DataProcessor()
    {
        // make sure m_pDataClient has the data processor as its parent, otherwise they end up in different threads.
 
        //....
        connect(m_pDataClient, SIGNAL(dataReady()), this, SLOT(slotDataReady()));
    }
 
private slots:
    void slotDataReady(); // processing method
};

class DataClient : public QObject
{
    Q_OBJECT

signals:
    void dataReady();

private slots:
    void slotReadData()
    {
         qint64 nBytesAvailable = m_tcpSocket->bytesAvailable();
         if (nBytesAvailable < (int)sizeof(quint16))  // Sufficient Data NOT Arrived
         {
             return;
         }
         // Any other Data Checking Statements
         ...

        emit dataReady();
    }
};

class DataProcessor : public QObject
{
    Q_OBJECT

public:
    DataProcessor()
    {
        // make sure m_pDataClient has the data processor as its parent, otherwise they end up in different threads.

        //....
        connect(m_pDataClient, SIGNAL(dataReady()), this, SLOT(slotDataReady()));
    }

private slots:
    void slotDataReady(); // processing method
};

To copy to clipboard, switch view to plain text mode 

DataProcessor *processor = new DataProcessor;
QThread *thread = new QThread;
processor->moveToThread(thread);

DataProcessor *processor = new DataProcessor;
QThread *thread = new QThread;
processor->moveToThread(thread);

To copy to clipboard, switch view to plain text mode 

If you want the thread to stop, call thread->quit() or through a signal/slot connection (quit is a slot).

One advantage of this approach is that you can test this without the additional threads, i.e. let the data processor do its work in the main thread, or have multiple processors on the same thread, etc.


Cheers,
_

[swamyonline]

Back after a while. The idea of using a 'blocking data processor' is to handle data frames of different sizes (over and under sized) read from socket. I am working on your suggestions. Thanks again.

swamy.