QThread and QQueue

[qball2k5]

Hey!

I have an application that has three thread.

Thread One: GUI
Thread Two: Messeger, comminicated with a device using an Aardvark (I2C bus)
Thread Three: Decoder, decodes the data from the Messeger, sends it to UI

The problem is that the Decoder seems to have a tendency to stop working, but I don't ever tell it to stop. I can see that the Messeger continuing to send data to te Decoder, but the decoder doesn't process it. Here's how the decoder thread looks.

Decoder::Decoder(QObject *parent) 
: QThread(parent)
{
    stopped = false;
    abort = false;
}
 
Decoder::~Decoder()
{
    mutex.lock();
    waitCondition.wakeOne();
    mutex.unlock();
 
     wait();
}
 
void Decoder::run()
{
	do
	{
	    mutex.lock();
	    if (decodeQueue.isEmpty())
		waitCondition.wait(&mutex);            
	    mutex.unlock();
 
	    if(mutex.tryLock())
	    {
		QStringList packet = decodeQueue.dequeue();
		mutex.unlock();
 
		Decode(packet);
	    }
 
	    msleep(2);                    
 
	}while(!stopped);
 
	if(stopped)
	    quit();
	else
	    exec();		
}
 
void Decoder::stop()
{
    stopped = true;
}
 
void Decoder::DecodePacket( const QStringList &packet )
{
    mutex.lock();
    decodeQueue.enqueue(packet);
    waitCondition.wakeOne();
    mutex.unlock();
}
 
void Decoder::Decode( QStringList packet )
{ 
     //Decodes packet and emits data to the GUi
}

Decoder::Decoder(QObject *parent) 
: QThread(parent)
{
    stopped = false;
    abort = false;
}

Decoder::~Decoder()
{
    mutex.lock();
    waitCondition.wakeOne();
    mutex.unlock();

     wait();
}

void Decoder::run()
{
	do
	{
	    mutex.lock();
	    if (decodeQueue.isEmpty())
		waitCondition.wait(&mutex);            
	    mutex.unlock();
	    
	    if(mutex.tryLock())
	    {
		QStringList packet = decodeQueue.dequeue();
		mutex.unlock();
		    
		Decode(packet);
	    }
          	
	    msleep(2);                    
	    
	}while(!stopped);
	
	if(stopped)
	    quit();
	else
	    exec();		
}

void Decoder::stop()
{
    stopped = true;
}

void Decoder::DecodePacket( const QStringList &packet )
{
    mutex.lock();
    decodeQueue.enqueue(packet);
    waitCondition.wakeOne();
    mutex.unlock();
}

void Decoder::Decode( QStringList packet )
{ 
     //Decodes packet and emits data to the GUi
}

To copy to clipboard, switch view to plain text mode 

Any ideas?

Thanks.

Rob

[jacek]

void Decoder::run()
{
        ...
	if(stopped)
	    quit();
	else
	    exec();		
}

void Decoder::run()
{
        ...
	if(stopped)
	    quit();
	else
	    exec();		
}

To copy to clipboard, switch view to plain text mode 

This part is a bit weird.

Decodes packet and emits data to the GUi

Are you sure that you use queued connection for this? Decoder object lives in a thread that created it, while Decoder::run() lives in a new thread --- this will fool automatic connections.

I'm not sure why do you unlock mutex just to try to lock it again.

Decoder::~Decoder()
{
    stopped = true;
 
    mutex.lock();
    waitCondition.wakeOne();
    mutex.unlock();
 
    wait();
}
 
Decoder::run() 
{
   while( ! stopped ) {
      mutex.lock();
 
      while( decodeQueue.isEmpty() && ! stopped ) {
         waitCondition.wait( &mutex );
      }
 
      if( ! stopped ) {
         QStringList packet = decodeQueue.dequeue();
      }
 
      mutex.unlock();
 
      if( ! stopped ) {
         Decode( packet );
      }
   }
}

Decoder::~Decoder()
{
    stopped = true;

    mutex.lock();
    waitCondition.wakeOne();
    mutex.unlock();
 
    wait();
}

Decoder::run() 
{
   while( ! stopped ) {
      mutex.lock();
   
      while( decodeQueue.isEmpty() && ! stopped ) {
         waitCondition.wait( &mutex );
      }

      if( ! stopped ) {
         QStringList packet = decodeQueue.dequeue();
      }

      mutex.unlock();

      if( ! stopped ) {
         Decode( packet );
      }
   }
}

To copy to clipboard, switch view to plain text mode 

[qball2k5]

I'm definately using a queued connection, I set that up for all my signals.

This is the slot that recieves the QStringList from the messager

void Decoder::DecodePacket( const QStringList &packet )
{
mutex.lock();    
decodeQueue.enqueue(packet);
waitCondition.wakeOne();
mutex.unlock();
}

void Decoder::DecodePacket( const QStringList &packet )
{
mutex.lock();    
decodeQueue.enqueue(packet);
waitCondition.wakeOne();
mutex.unlock();
}

To copy to clipboard, switch view to plain text mode 

In your example, won't the mutex be locked so this will never be able to add a new packet into the queue? Or am I missing something?

As for the wierd part, I don't know why I put that there, thought I needed it. It was in one of the examples.

[qball2k5]

You know what....it works much better. I have another thread question though.

My application right now is using like 85% of the CPU. I know it's because my messager thread. The messager thread basically has a queue of outgoing messages and if a message needs to be sent, it sends it. Then it checks to see if it needs to read data from the Aardvark (I2C). If it does, it reads it then passes it to the Decoder.

Here's how it looks. It works, just not very efficieintly. Any ideas?

Messager::~Messager()
{
    //Destructor
    mutex.lock();
    waitCondition.wakeOne();
    mutex.unlock();
 
     wait();
}
 
void Messager::run()
{
 
    do	    
    {
	if(!packetQueue.isEmpty())
	{	    	    
	    if(mutex.tryLock())
	    {		
		Packet sendPacket = packetQueue.dequeue();
		UINT8 address = addressQueue.dequeue();
		UPDATE type = typeQueue.dequeue();
 
		mutex.unlock();
 
		if(SendMessage(sendPacket, address))
		    emit PacketSent(true, type);
		else
		    emit PacketSent(false, type);	   
	    }
	}	
 
	msleep(pollRate);	
 
	if(ReadDevice1)
	    ReadMessage(D1_ADDRESS);
 
 
	if(ReadDevice2)
	    ReadMessage(D2_ADDRESS);
 
 
	if(ReadDevice3)
	    ReadMessage(D3_ADDRESS);
 
 
	if(ReadDevice4)
	    ReadMessage(D4_ADDRESS);
 
 
	if (ReadDevice5)	
	    ReadMessage(D5_ADDRESS);	
 
	if(ReadDevice6)
	    ReadMessage(D6_ADDRESS);
 
 
	if(abort)
	    stopped = true;	    
 
    }while(!stopped);   
 
    if(stopped)
	quit();
    else
	exec();	
 
}
 
void Messager::SendPacket(Packet myPacket, UINT8 address, UPDATE Type)
{
    QMutexLocker locker(&mutex);
 
    packetQueue.enqueue(myPacket);
    addressQueue.enqueue(address);
    typeQueue.enqueue(Type);       
}
 
bool Messager::Config(int port, int bitrate)
{
    //Configure I2C device
}
 
bool Messager::SendMessage(Packet outgoingPacket, UINT8 addr)
{    
    //Write packet to I2C bus
}
 
bool Messager::ReadMessage(UINT8 address)
{
   //Read data from I2C bus, emit PacketRecieved signal to Decoder
}
 
void Messager::EnableDevice1Status(bool Enabled)
{
    mutex.lock();
    ReadDevice1 = Enabled;
    mutex.unlock();
}

Messager::~Messager()
{
    //Destructor
    mutex.lock();
    waitCondition.wakeOne();
    mutex.unlock();

     wait();
}

void Messager::run()
{
   
    do	    
    {
	if(!packetQueue.isEmpty())
	{	    	    
	    if(mutex.tryLock())
	    {		
		Packet sendPacket = packetQueue.dequeue();
		UINT8 address = addressQueue.dequeue();
		UPDATE type = typeQueue.dequeue();
		
		mutex.unlock();
		
		if(SendMessage(sendPacket, address))
		    emit PacketSent(true, type);
		else
		    emit PacketSent(false, type);	   
	    }
	}	

	msleep(pollRate);	
	
	if(ReadDevice1)
	    ReadMessage(D1_ADDRESS);

	
	if(ReadDevice2)
	    ReadMessage(D2_ADDRESS);

	
	if(ReadDevice3)
	    ReadMessage(D3_ADDRESS);

	
	if(ReadDevice4)
	    ReadMessage(D4_ADDRESS);

	
	if (ReadDevice5)	
	    ReadMessage(D5_ADDRESS);	
	
	if(ReadDevice6)
	    ReadMessage(D6_ADDRESS);

	
	if(abort)
	    stopped = true;	    
	
    }while(!stopped);   
    
    if(stopped)
	quit();
    else
	exec();	

}

void Messager::SendPacket(Packet myPacket, UINT8 address, UPDATE Type)
{
    QMutexLocker locker(&mutex);
    
    packetQueue.enqueue(myPacket);
    addressQueue.enqueue(address);
    typeQueue.enqueue(Type);       
}

bool Messager::Config(int port, int bitrate)
{
    //Configure I2C device
}

bool Messager::SendMessage(Packet outgoingPacket, UINT8 addr)
{    
    //Write packet to I2C bus
}

bool Messager::ReadMessage(UINT8 address)
{
   //Read data from I2C bus, emit PacketRecieved signal to Decoder
}

void Messager::EnableDevice1Status(bool Enabled)
{
    mutex.lock();
    ReadDevice1 = Enabled;
    mutex.unlock();
}

To copy to clipboard, switch view to plain text mode 

It needs to read only when told to and send any message that gets put into the queue. I'm still new to threads. I'd really appreciate any ideas.

If this is confusing, I can attempt to elaborate better.

Thanks,

Rob

[jacek]

My application right now is using like 85% of the CPU. I know it's because my messager thread. The messager thread basically has a queue of outgoing messages and if a message needs to be sent, it sends it. Then it checks to see if it needs to read data from the Aardvark (I2C). If it does, it reads it then passes it to the Decoder.

Sounds like it's a job for QSocketNotifier.

To avoid problems, IMO, you should split the implementation into two classes: Messenger and MessengerThread.

MessengerThread should look like this:

void MessengerThread::run()
{
   Messenger m;
   // setup connections or whatever
   exec();
}

void MessengerThread::run()
{
   Messenger m;
   // setup connections or whatever
   exec();
}

To copy to clipboard, switch view to plain text mode 

while Messenger (derived from QObject) should use QSocketNotifiers to monitor the devices.

Of course you can do this using a single class, but you must watch out for connection types.

[qball2k5]

So....

I'm starting the two threads from the GUI thread, called DeckSimulator.

I call:

messager.start(QThread::TimeCriticalPriority);	
decoder.start(QThread::LowPriority);

messager.start(QThread::TimeCriticalPriority);	
decoder.start(QThread::LowPriority);

To copy to clipboard, switch view to plain text mode 

in the constructor of DeckSimulator

That's starts them on there own thread. Right? Or do I need to start the thread when I call SendPacket(from Messager) and DecodePacket(from Decoder). Like in the Mandalbrot example.

[jacek]

That's starts them on there own thread. Right?

Yes, but only the run() method will be in a new thread. messager and decoder objects will live in the GUI thread, as you instantiate them before the new thread is created.

Or do I need to start the thread when I call SendPacket(from Messager) and DecodePacket(from Decoder). Like in the Mandalbrot example.

No, you don't have to.

[qball2k5]

Sounds like it's a job for QSocketNotifier.

It's really just one device, the Aardvark device, but different address to read using that device.

It still looks like it locks up after a while and doesn't work. Could this be a stack issue?

It just stops decoding, even though I can see the reads being done. I see the packet signal getting emitted and the DecodePacket slot getting called by that emitted signal, via breakpoints...i put breakpoints in the run function and it's not doing anything. the Run function isn't going, but the stopped variable is false, so it should continue to run.

Any ideas?

[jacek]

It's really just one device, the Aardvark device, but different address to read using that device.

It doesn't matter how many devices you have, QSocketNotifier should notify you when there is something to read, but of course you need a file descriptor for this.

It just stops decoding, even though I can see the reads being done.

How about these?

Decoder::~Decoder()
{
    waitCondition.wakeOne();
    wait();
}
 
void Decoder::DecodePacket( const QStringList &packet )
{
   mutex.lock();    
   decodeQueue.enqueue(packet);
   mutex.unlock();
 
   waitCondition.wakeOne();
}

Decoder::~Decoder()
{
    waitCondition.wakeOne();
    wait();
}

void Decoder::DecodePacket( const QStringList &packet )
{
   mutex.lock();    
   decodeQueue.enqueue(packet);
   mutex.unlock();

   waitCondition.wakeOne();
}

To copy to clipboard, switch view to plain text mode 

[qball2k5]

It only seems to happen when I minimize it and the bring it back?

Do I need to do something special for this?

[jacek]

It only seems to happen when I minimize it and the bring it back?

Are you sure that signal/slot connections between decoder and reader threads are direct? If not, all data will go through GUI thread's event loop.

Anyway, how much data you receive from that device? Maybe you don't need 3 threads for this?

[jacek]

I'm definately using a queued connection, I set that up for all my signals.

This is the slot that recieves the QStringList from the messager

When you use queued connections, signals behave like events that are delivered to event loop that runs in a thread that owns the receiver.

There are two problems:

• your thread doesn't have a running event loop (and it doesn't have to), so you can't use queued connections to send signals to objects that live inside it. You have to use direct connections, but with proper locking mechanism to avoid problems.
• Decoder was created in the GUI thread, so all events (including those that realize queued connections) will go through the event loop in the GUI thread. If you block the GUI, you will also block the Decoder.

In other words, try using a direct connection when you connect to Decoder::DecodePacket() (but Decoder should be still connected to the GUI using queued connection).

In your example, won't the mutex be locked so this will never be able to add a new packet into the queue?

No, it shouldn't. QWaitCondition::wait() unlocks the mutex, waits and locks it again.

As for the wierd part, I don't know why I put that there, thought I needed it. It was in one of the examples.

You exit the while loop only when stopped == true, so the else clause is never executed.