Implementation of FastICA in multithreading approach

Question

I am to implement a parallel version of FastICA, so I implemented the serial version of FastICA.

Now this is the architectural diagram of how it is going to parallelize. I just coded a basic thread model of this. But I need to know what I did was correct or wrong regarding inter threads synchronization.

#include <condition_variable>
#include <mutex>
#include <vector>
#include <iostream>
#include <thread>


int numberOfThreads=10;
std::vector<int> checkedEvent(numberOfThreads);
std::vector<double> data(numberOfThreads);
std::mutex lock;
int IsPrepare=0;

std::condition_variable data_prepare;

class MainProcess
{
private:
    int rank;
    double input;
public:
    MainProcess(int rank_, double dataPr){
        rank=rank_;
        input=dataPr;
    }
    MainProcess(){

    }
    void fastICA_processing()
    {
        while(true){
            std::unique_lock<std::mutex> lk(lock);
            data_prepare.wait(lk,[this]{return checkedEvent[rank]!=1;});
            data[rank]=input;
            checkedEvent[rank]=1;
            IsPrepare=IsPrepare+1;
            std::cerr <<IsPrepare<<" thread is processing"<<std::endl;
            lk.unlock();
        }

    }

    void controller()
    {
        while(true)
        {

            while(IsPrepare==numberOfThreads)
            {
                std::lock_guard<std::mutex> lk(lock);
                std::cerr<<"Now controller is checking fastICA is converged or not "<<std::endl;
                std::fill(checkedEvent.begin(), checkedEvent.end(), 0);
                IsPrepare=0;
                data_prepare.notify_all();
            }
        }

    }
    };

int main()
{
    std::vector<std::thread> thredArray;
    MainProcess p3(0,0);
    std::thread process_thread(&MainProcess::controller,&p3);
    for(int i=0;i<numberOfThreads;i++){
        MainProcess* p1= new MainProcess(i,i);
        thredArray.push_back(std::thread (&MainProcess::fastICA_processing,p1));
    }
    for(auto &t:thredArray)
    {
        t.join();
    }
    process_thread.join();

    return 0;
}

Note: Here "data" stands for input data. First fastICA_processing threads do some mathematical calculations and then the controller thread gets all that data and processes it. This will happen again and again. I didn't add threads detaching part here.

Answer 1

The code is a mess, but I can tell you that it will spin forever right here:

std::thread process_thread(&MainProcess::controller,&p3);

This starts controller running immediately.

void controller()
{
    while(true)
    {
        while(IsPrepare==numberOfThreads)
        {
            std::lock_guard<std::mutex> lk(lock);
            std::cerr<<"Now controller is checking fastICA is converged or not "<<std::endl;
            std::fill(checkedEvent.begin(), checkedEvent.end(), 0);
            IsPrepare=0;
            data_prepare.notify_all();
        }
    }
}

IsPrepare is 0, and numberOfThreads is 10, and there's no synchronization to worry about here (no locks, no atomics), so we can definitely say that 0 == 10 is invariably false. Thus controller reduces to

void controller()
{
    while(true)
    {
        while(false) {}
    }
}

Okay, so that thread won't be doing anything anytime soon. And with that code out of the way, the rest is pretty much irrelevant; it looks to me as if it will end up blocking forever, since there's nobody left to notify the condition variable.

Your problem might be suited to a TBB flowgraph; consider using TBB instead of hand-rolling your own stuff.