How to loop multithreaded job

[stefan]

Hi all!

My base classes are:

class Manager: public QThread
class Worker: public QThread

class Manager: public QThread
class Worker: public QThread

To copy to clipboard, switch view to plain text mode 

Manager starts 4 (independed) workers and "listens" for their signals (and counts finished workers). Manager restarts workers several times.
It also emits final results to GUI thread.
This is all working flawless.

Now, i want to loop whole job (manager's job) in Tester thread.

void Tester::run(void)
{
    for (int iParam1 = 0; iParam1 < parameter1.size(); iParam1++)
    {
        for (int iParam2 = 0; iParam2 < parameter2.size(); iParam2++)
        {
            double tmpValueSum = 0.0;
            for (int iRun = 0; iRun < numberOfRuns; iRun++)
            {
                manager->parameter1 = parameter1[iParam1];
                manager->parameter2 = parameter2[iParam2];
 
                manager->run();
                manager->wait(); // This is a problem
            }
            results[iParam1][iParam2] = tmpValueSum / double(numberOfRuns);
        }
    }
}

void Tester::run(void)
{
    for (int iParam1 = 0; iParam1 < parameter1.size(); iParam1++)
    {
        for (int iParam2 = 0; iParam2 < parameter2.size(); iParam2++)
        {
            double tmpValueSum = 0.0;
            for (int iRun = 0; iRun < numberOfRuns; iRun++)
            {
                manager->parameter1 = parameter1[iParam1];
                manager->parameter2 = parameter2[iParam2];

                manager->run();
                manager->wait(); // This is a problem
            }
            results[iParam1][iParam2] = tmpValueSum / double(numberOfRuns);
        }
    }
}

To copy to clipboard, switch view to plain text mode 

I miss something in the way I understood wait function. I tried exit(), quit() and exec() in Manager...
What modifications should be implemented to make this possible?

In base case, i only have manager->run(); and there is no need for waiting... Manager emits some output and final results to GUI thread.

Thanks!

EDIT: I found this example
.
As I see it, the only important difference is that my thread raises other threads and have few connections...

[wysota]

Calling run() does not start a new thread.

[anda_skoa]

As wysota said, QThread::run() is the method that you would like to have executed by the new thread. If you call it directly, obviously your current thread will execute it instead, in your case your test program.

What you actually want to do is start the manager so its run() method is executed by a new thread.

manager->start();

manager->start();

To copy to clipboard, switch view to plain text mode 

Cheers,
_

[wysota]

...which leads to suspecting the OP never had 4 threads running there at all.

[stefan]

What you actually want to do is start the manager so its run() method is executed by a new thread.

manager->start();

manager->start();

To copy to clipboard, switch view to plain text mode 

Cheers,
_

ough... I know that. It's a typo... this could be the cause of all problems. I'll try it later.
Sorry, i really didn't see it...

[stefan]

Hi, it's me again

I have a problem running my code on windows. It works on Linux!
The probelm must be with threads...

I suspect on this part of code:

QVector<Pso*> psoInstances;
for(int iThread=0; iThread<numberOfThreads; iThread++)
{
    Pso* pso = initPsoInstance();
    psoInstances.push_back(pso);
 
    // Dimension vs Swarm size
    int maxfuncalls = 100000;
    pso->dimensions = parameter1[iParam1];
    QVector<double> lb(pso->dimensions), ub(pso->dimensions);
    lb.fill(-50.0);
    ub.fill(50.0);
    pso->lowerBound = lb;
    pso->upperBound = ub;                                        
    pso->psoParameters.swarmSize= parameter2[iParam2];
    pso->psoParameters.maxIterations = int(double(maxfuncalls)/double(pso->psoParameters.swarmSize));
 
    pso->clear();
    pso->start();
}
 
for(int iThread=0; iThread<numberOfThreads; iThread++)
{
    psoInstances[iThread]->wait();
 
    tmpValueSum += psoInstances[iThread]->iteration * psoInstances[iThread]->psoParameters.swarmSize;
 
    delete psoInstances[iThread]->evaluator;
    delete psoInstances[iThread];
}
psoInstances.clear();

QVector<Pso*> psoInstances;
for(int iThread=0; iThread<numberOfThreads; iThread++)
{
    Pso* pso = initPsoInstance();
    psoInstances.push_back(pso);

    // Dimension vs Swarm size
    int maxfuncalls = 100000;
    pso->dimensions = parameter1[iParam1];
    QVector<double> lb(pso->dimensions), ub(pso->dimensions);
    lb.fill(-50.0);
    ub.fill(50.0);
    pso->lowerBound = lb;
    pso->upperBound = ub;                                        
    pso->psoParameters.swarmSize= parameter2[iParam2];
    pso->psoParameters.maxIterations = int(double(maxfuncalls)/double(pso->psoParameters.swarmSize));

    pso->clear();
    pso->start();
}

for(int iThread=0; iThread<numberOfThreads; iThread++)
{
    psoInstances[iThread]->wait();

    tmpValueSum += psoInstances[iThread]->iteration * psoInstances[iThread]->psoParameters.swarmSize;

    delete psoInstances[iThread]->evaluator;
    delete psoInstances[iThread];
}
psoInstances.clear();

To copy to clipboard, switch view to plain text mode 

I'm running this in Visual Studio 2010.
During the execution the following message is repeatedly outputted:

The thread 'Swarm' (0xc74) has exited with code 0 (0x0).

The thread 'Swarm' (0xc74) has exited with code 0 (0x0).

To copy to clipboard, switch view to plain text mode 

When debugging the code it get stuck on wait() function in second loop.

Is there any thread-related difference in Qt code behavior between Linux and Windows?

Just to mention

class Pso: public QThread

class Pso: public QThread

To copy to clipboard, switch view to plain text mode 

Any suggestion is more then welcome!

Thanx

---

Added after 22 minutes:

I found problem!

There is no problem with wait function. In Debug the code i too slow to do anything. It works in Release.
Is this normal for debugging multithread calculations?
It's completely uselss to do anything in Debug... I would wait days for result.
Also, in Release there is no message

The thread 'Swarm' (0xc74) has exited with code 0 (0x0).

The thread 'Swarm' (0xc74) has exited with code 0 (0x0).

To copy to clipboard, switch view to plain text mode 

[wysota]

I thing the problem is the thread exits before you call wait on it. In general you have a somewhat wrong approach. You should monitor each thread's finished() signal and when you receive it from all the threads, you should clean up your environment. There is no need for this "for" loop you have there. The fact that it works for you in Linux is probably either incidental or caused by different semantics of the wait() call. Either way this is not a good approach.

[stefan]

I thing the problem is the thread exits before you call wait on it. In general you have a somewhat wrong approach. You should monitor each thread's finished() signal and when you receive it from all the threads, you should clean up your environment. There is no need for this "for" loop you have there. The fact that it works for you in Linux is probably either incidental or caused by different semantics of the wait() call. Either way this is not a good approach.

Thanks wysota!

The problem is that this code is in two external for loops. If I do this by implementing onPsoFinished slot (and connect it to Pso::finished() signal) then I cant have external for loops. I must do function which increments all 2 counters and starts new set of Pso threads when previous is finished. It looks messy.
Why cant I use wait function in this way? What is its purpose if not waiting for thread to exit?

[wysota]

The problem is that this code is in two external for loops. If I do this by implementing onPsoFinished slot (and connect it to Pso::finished() signal) then I cant have external for loops. I must do function which increments all 2 counters and starts new set of Pso threads when previous is finished. It looks messy.

I don't know what you mean. I don't see any reason to have new set of threads each time.

What is its purpose if not waiting for thread to exit?

You are waiting for thread 1 to exit, then thread 2, then thread 3, then... What if thread 2 exits before thread 1? Then you're calling wait() on a thread that has already finished and end up with the message you're getting.

[stefan]

I don't know what you mean. I don't see any reason to have new set of threads each time.

Even if i dont create new threads i must "restart" them...

Maybe this "pseudo code" would clear things...

for p1=1:10
    for p2=1:20
        for run=1:25
            Pso.set(p1,p2)
            run 4 x Pso (in 4 threads)(each Pso results are random based!)
        calculate average of all runs (4x25=100)
        results[p1][p2] = average

for p1=1:10
    for p2=1:20
        for run=1:25
            Pso.set(p1,p2)
            run 4 x Pso (in 4 threads)(each Pso results are random based!)
        calculate average of all runs (4x25=100)
        results[p1][p2] = average

To copy to clipboard, switch view to plain text mode 

The best way would be to make threading on external loop (one thread for each p1) but Pso execution time is greatly depended on p1 and p2.
That is why I made threading inside of "runs" loop.

I know how to do it with finished() signal - but code dont look so clean

You are waiting for thread 1 to exit, then thread 2, then thread 3, then... What if thread 2 exits before thread 1? Then you're calling wait() on a thread that has already finished and end up with the message you're getting.

I see... That make sense!
Waiting for allready finished thread is a bad thing to do?
Qt Doc:

This function will return true if the thread has finished. It also returns true if the thread has not been started yet.

[wysota]

Waiting for allready finished thread is a bad thing to do?

No idea, apparently it doesn't work for you.

I would approach the problem from a completely different end. I'd probably have a single object containing all information about the whole job and make it manage everything.

[anda_skoa]

One way to wait for a set of workers to complete their work is to use a QSemaphore.

Each worker thread acquires one unit of the semaphore when it enters its run() method and releases one unit before it exits.
The main thread releases one unit whenever it starts a thread (thus allowing the thread to acquire it).
When the main thread wants to wait for all to be finished it acquires as many units as it created threads. if all threads have exited already, this will return immediately. if any thread is still running, the acquire(numThreads) will block until the last worker has release the unit it was holding.

Cheers,
_

[wysota]

Actually with a semaphore you can omit the step of acquiring units by the threads -- simpy initialize the semaphore with the value of 0. The advantage is that there is no race condition if some worker threads are started with a delay larger than it takes the main thread to try and acquire the semaphore.