QThread, signalling and readyRead()

[kachofool]

Hi all,

I'm trying to develop a multi-threaded application that controls serial i/o. I'm having some odd behaviour that I think is due to incorrectly using QThread (between run and exec maybe).

* I have two threads, a subclass of QThread, that send requests to devices connected through the serial port (one thread / port).

* Both threads send a request, and then block using QWaitCondition until the device sends a response, which is indicated with a readyRead() signal (I'm using QextSerial in EventDriven mode).

* The readyRead() signal is connected to an object that controls the two threads. The object has public slots that call QWaitCondition.wakeOne() when the readyRead() signal is received for each thread.

Everything works as it should when there is only one thread running. Once a second thread is run, the other thread will block. But the threads aren't really ever interdependent in that sense, and they should never wait on each other. So it just seems like there's no concurrency here and everything is linear. I've attached relevant parts of my code and would appreciate any insight.

class DeviceThread : public QThread
{
	Q_OBJECT
 
public:
AxisCouple *_axis;
...
protected:
void DeviceThread::run()
{
	switch(_device)
	{
		case SENSOR:
		{
			// connect to sensor serial port
			QextSerialPort* devPort = _axis->_sensor.GetSerialPort();
			QObject::connect(devPort, SIGNAL(readyRead()),
					_axis, SLOT(SensorReady())	);
 
			dataRow dRow;
			while (_keepRunning)	// START SENSOR LOOP
			{
				// request sensor data
				_axis->_sensor.RequestPos();
 
				// wait until serial port signals rx data
				_axis->_lockSensor.lock();
				_axis->_sensorReady.wait(&(_axis->_lockSensor));
 
				// atomic on (ignore signals)
				QObject::disconnect(devPort, SIGNAL(readyRead()),
						_axis, SLOT(SensorReady())	);
 
				// read encoder and get timestamp
				//dRow.mstime_ =  _axis->getElapsedTime();
				dRow.mstime_ = 0;
				dRow.data_ = _axis->_sensor.RecvPos();
				std::cerr << "sensor position: " << dRow.data_ << std::endl;
 
				_axis->_lockSensor.unlock();
 
					_axis->_lockSensorBuffer.lock();
					_axis->_sensorBuff.push_back(dRow);
					_axis->_lockSensorBuffer.unlock();
 
				// atomic off (allow signals)
				QObject::connect(devPort, SIGNAL(readyRead()),
						_axis, SLOT(SensorReady())	);
			}
		}
			break;
 
		case MOVER:
		{
			// connect to mover serial port
			QextSerialPort* mvPort = _axis->_mover.GetSerialPort();
			QObject::connect(mvPort, SIGNAL(readyRead()),
					_axis, SLOT(MoverReady())	);
 
			int sSize(0), minDataPoints(0);
			double nextPos(0), prevPos(0);
			std::vector<dataRow> sn;
 
 
			while (_keepRunning)	// START MOVER LOOP
			{	
				// copy sensor buffer and clear before releasing
				_axis->_lockSensorBuffer.lock();
				sSize = _axis->_sensorBuff.size();
				if (sSize > minDataPoints)
				{
					sn.assign(	_axis->_sensorBuff.begin(),
								_axis->_sensorBuff.end());
 
					_axis->_sensorBuff.clear();	
				}
				_axis->_lockSensorBuffer.unlock();
 
				if (sSize > minDataPoints)
				{
					prevPos = nextPos;
					nextPos = (sn[0]).data_;
				}
 
				// update mover position data
				_axis->_lockMoverData.lock();
					_axis->_moverPrevPos = prevPos;
					_axis->_moverNextPos = nextPos;
				_axis->_lockMoverData.unlock();
 
				// send mover new position data
				_axis->_mover.PositionMotor(nextPos, prevPos);
				std::cerr << "sent something!" << std::endl;
 
				// wait until serial port signals rx data
				_axis->_lockMover.lock();
				_axis->_moverReady.wait(&(_axis->_lockMover));
 
				// atomic on (ignore signals) after (1) event recvd
				QObject::disconnect(mvPort, SIGNAL(readyRead()),
						_axis, SLOT(MoverReady())	);
 
				try
				{	_axis->_mover.VerifyMotor(nextPos);	}
 
				catch(ErrorMVPRead&)
				{	_axis->_mover.StopMotor();	}
 
				_axis->_lockMover.unlock();
 
				// atomic off (allow signals)
				QObject::connect(mvPort, SIGNAL(readyRead()),
						_axis, SLOT(MoverReady())	);
 
			}						
		}							// END MOVER LOOP
			break;
 
...
...
 
class AxisCouple : public QObject
{
	Q_OBJECT
...
 
AxisCouple::AxisCouple(std::string type, QObject* parent) 
	: QObject(parent)
{
	// setup threads
	_sensorThread = new DeviceThread(this, SENSOR);
	_sensor.InitializeSensor();
	_sensorThread->RunDevice();
 
	_moverThread = new DeviceThread(this, MOVER);
	_mover.InitializeMover();
	_moverThread->RunDevice();
 
}
 
// public slots
void AxisCouple::SensorReady()
{
	// announce  data available
	_sensorReady.wakeOne();
}
 
void AxisCouple::MoverReady()
{
	// announce data available
	_moverReady.wakeOne();
}

class DeviceThread : public QThread
{
	Q_OBJECT

public:
AxisCouple *_axis;
...
protected:
void DeviceThread::run()
{
	switch(_device)
	{
		case SENSOR:
		{
			// connect to sensor serial port
			QextSerialPort* devPort = _axis->_sensor.GetSerialPort();
			QObject::connect(devPort, SIGNAL(readyRead()),
					_axis, SLOT(SensorReady())	);

			dataRow dRow;
			while (_keepRunning)	// START SENSOR LOOP
			{
				// request sensor data
				_axis->_sensor.RequestPos();
				
				// wait until serial port signals rx data
				_axis->_lockSensor.lock();
				_axis->_sensorReady.wait(&(_axis->_lockSensor));
				
				// atomic on (ignore signals)
				QObject::disconnect(devPort, SIGNAL(readyRead()),
						_axis, SLOT(SensorReady())	);
				
				// read encoder and get timestamp
				//dRow.mstime_ =  _axis->getElapsedTime();
				dRow.mstime_ = 0;
				dRow.data_ = _axis->_sensor.RecvPos();
				std::cerr << "sensor position: " << dRow.data_ << std::endl;
					
				_axis->_lockSensor.unlock();

					_axis->_lockSensorBuffer.lock();
					_axis->_sensorBuff.push_back(dRow);
					_axis->_lockSensorBuffer.unlock();

				// atomic off (allow signals)
				QObject::connect(devPort, SIGNAL(readyRead()),
						_axis, SLOT(SensorReady())	);
			}
		}
			break;

		case MOVER:
		{
			// connect to mover serial port
			QextSerialPort* mvPort = _axis->_mover.GetSerialPort();
			QObject::connect(mvPort, SIGNAL(readyRead()),
					_axis, SLOT(MoverReady())	);
			
			int sSize(0), minDataPoints(0);
			double nextPos(0), prevPos(0);
			std::vector<dataRow> sn;
			
			
			while (_keepRunning)	// START MOVER LOOP
			{	
				// copy sensor buffer and clear before releasing
				_axis->_lockSensorBuffer.lock();
				sSize = _axis->_sensorBuff.size();
				if (sSize > minDataPoints)
				{
					sn.assign(	_axis->_sensorBuff.begin(),
								_axis->_sensorBuff.end());
								
					_axis->_sensorBuff.clear();	
				}
				_axis->_lockSensorBuffer.unlock();
				
				if (sSize > minDataPoints)
				{
					prevPos = nextPos;
					nextPos = (sn[0]).data_;
				}
				
				// update mover position data
				_axis->_lockMoverData.lock();
					_axis->_moverPrevPos = prevPos;
					_axis->_moverNextPos = nextPos;
				_axis->_lockMoverData.unlock();
				
				// send mover new position data
				_axis->_mover.PositionMotor(nextPos, prevPos);
				std::cerr << "sent something!" << std::endl;
				
				// wait until serial port signals rx data
				_axis->_lockMover.lock();
				_axis->_moverReady.wait(&(_axis->_lockMover));
				
				// atomic on (ignore signals) after (1) event recvd
				QObject::disconnect(mvPort, SIGNAL(readyRead()),
						_axis, SLOT(MoverReady())	);
						
				try
				{	_axis->_mover.VerifyMotor(nextPos);	}
				
				catch(ErrorMVPRead&)
				{	_axis->_mover.StopMotor();	}
										
				_axis->_lockMover.unlock();
				
				// atomic off (allow signals)
				QObject::connect(mvPort, SIGNAL(readyRead()),
						_axis, SLOT(MoverReady())	);
	
			}						
		}							// END MOVER LOOP
			break;

...
...

class AxisCouple : public QObject
{
	Q_OBJECT
...

AxisCouple::AxisCouple(std::string type, QObject* parent) 
	: QObject(parent)
{
	// setup threads
	_sensorThread = new DeviceThread(this, SENSOR);
	_sensor.InitializeSensor();
	_sensorThread->RunDevice();
	
	_moverThread = new DeviceThread(this, MOVER);
	_mover.InitializeMover();
	_moverThread->RunDevice();
	
}

// public slots
void AxisCouple::SensorReady()
{
	// announce  data available
	_sensorReady.wakeOne();
}

void AxisCouple::MoverReady()
{
	// announce data available
	_moverReady.wakeOne();
}

To copy to clipboard, switch view to plain text mode 

I've tried to add specific comments to my code to highlight what I'm talking about.

Regards,

-KF

[wysota]

Where do you create the _axis object? Also it looks like you are accessing shared data without proper synchronization.

[kachofool]

Hi wysota,

I've initialized an AxisCouple object in my main.cpp file.
It's the default file with no changes except the init:

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);
 
    AxisCouple newRCNSAxis("Rotation");
 
    return a.exec();
}

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);
    
    AxisCouple newRCNSAxis("Rotation");

    return a.exec();
}

To copy to clipboard, switch view to plain text mode 

I use a (this) pointer when launching the DeviceThreads as in the above post to link the two.

With regards to improper synchronization, I'm unable to locate where shared data isn't safely being accessed; The only shared variable between threads is 'sensorBuffer' and I'm using a mutex to protect this variable.

[wysota]

The whole _axis object is shared, isn't it? The QExtSerialPort variable is also shared and it can't be protected with a mutex as it's a QObject so you don't have total control over it.
Its slots will be executed in the main thread, by the way - I don't know if this is something that you intended. You can read its data only from the main thread.

Also get rid of the connect and disconnect statements. I'm not entirely sure what you meant them to do but I'm pretty sure it's not a good design.

[kachofool]

Hi wysota,

Thanks for your help. The axis object is shared, but I'm mostly using this object to call member functions that are used by my system. The two device threads shouldn't be manipulating common shared memory except for the shared variable I'm protecting with the mutex (I could totally be wrong since I'm new to multithreading, so feel free to school me ).

There are two seperate QextSerialPorts instances, one for each device. The two serial port pointers I set up in the thread class point to different memory locations.

My reasoning for disconnecting and reconnecting the signals and slots is to ignore signals while the program is doing something specific. Since I'm using event driven I/O over the serial port, I can get multiple events from the device for something that actually represents one proper response I need to process. My event handler should only be called once for every 'proper response'...to do this I wait until I get a response from the device after requesting something, and then 'ignore' subsequent events by disconnecting the associated signals and slots until I'm done addressing the response. I'm used to thinking in terms of system level interrupts where interrupts are ignored during specific blocks of code.

It's fine if the slots for the axis couple object are executed in the main thread as long as it properly wakes up the blocked device threads, which is where I'm having the problem.

[kachofool]

After going through my code some more, I think part of the problem is that I get multiple signals (ie. the signals my threads are waiting on) from QextSerial's readyRead() event so rapidly, that they preempt my attempt to disconnect the signals. I'm not sure how to implement the functionality to ignore these events during certain blocks of code.

I was thinking I could try to run exec() loops for the threads themselves (I'm not sure how to do this -- do I just overload exec() like I overloaded run() previously?) and add slots to my thread objects that tell the thread to ignore signals from QextSerialPort. I still foresee the same problem though -- as in, I'll receive a signal, and before I finish running my slot, I get another signal (ad inifinitum).

As a very last ditch attempt I'm thinking I can modify QextSerialPort's code so it only sends 1 readReady signal and then stops until my class sends QextSerialPort some sort of acknowledge.

Does anyone have any other ideas? Or maybe I'm going about this whole thing wrong?

[wysota]

I think you are going for the whole thing wrong First you are assuming that by the time you receive readyRead() all the data you need will be available, which is incorrect, only part of it may be there. And to overcome this issue you should read the data from the port when it comes in and only when there is enough to process it, signal the wait condition. This way you'll get rid of two problems at once. And no need to disconnect signals.

Next thing - by groupping everything into one object, your program has become everything but a good designed object oriented application. Instead of aggregating, separate the data used by different tasks. Make them two classes containing only the data they need. There is a question whether you need the threads at all - from what I see, you don't, you can do everything in one thread. Once you have clear design, everything will become easier.

[kachofool]

Part of the reason I feel totally lost is that I'm migrating from pthreads, where I don't need to create a separate class for threads, and where I can run things concurrently in a way that seems coherent. If I have different devices that need to talk to each other and to my application, I would think that threads would be involved.

I don't understand how I can emit a signal only when there are enough bytes available to indicate a proper response from my device.

I don't understand how I could run the GUI, and several devices that all interact which each other with a single thread. I couldn't even run the GUI by itself doing mundane operations like opening up a dialog box without multiple threads. So could you please point me in the right direction?

[wysota]

Part of the reason I feel totally lost is that I'm migrating from pthreads, where I don't need to create a separate class for threads,

Probably because pthreads has a C interface

and where I can run things concurrently in a way that seems coherent.

I don't think this has anything to do with your current situation.

If I have different devices that need to talk to each other and to my application, I would think that threads would be involved.

Only if i/o operations are blocking. Otherwise you wouldn't need threads.

I don't understand how I can emit a signal only when there are enough bytes available to indicate a proper response from my device.

I didn't say you should. I said you should react on readyRead() in a different manner than you are doing now. Buffer the data until there is enough to perform your task. Then you can emit a signal or do whatever you like.

For instance assuming a "datagram" would be a complete line of text, you could do this in your slot connected to readyRead():

void Class::onReadyRead() {
  while(device->canReadLine()) {
    QByteArray data = device->readLine();
    doSomethingWithThe(data);
  }
}

void Class::onReadyRead() {
  while(device->canReadLine()) {
    QByteArray data = device->readLine();
    doSomethingWithThe(data);
  }
}

To copy to clipboard, switch view to plain text mode 

So could you please point me in the right direction?

Signals and slots are your friends.

[kachofool]

My 'class::OnReadyRead' never runs to completion -- that's the first problem.

The second issue is I don't understand how to set up the structure of my program using Qt. I have i/o going on with several devices. The i/o uses software 'handshaking', in that I talk to my device, wait for a response, verify the response (take corrective action if necessary), rinse and repeat. The thread communicating to the device should block while waiting for a response (so in this sense there is blocking involved).

So I'm really trying to figure out what method to set this sort of system up in. What I did initially was set up threads to talk to my devices, and block them using wait conditions after they had sent something to the device. I then tried to use signals (to and from the main application thread) to tell these threads to unblock, but this didn't work.

When you say 'signals and slots are your friend' could you please emphasize in the context of what I'm trying to accomplish? I really appreciate the help so far.

-KF