source/train/XLeaderPS.cpp
bool XLeaderPS::Run(XConfig* config, DataDistributeBase* dataDistributor, XOptimizer* optimizer)
{
    CheckNTErrors(jworkers.count > 0, "No jworkers!");
    CheckNTErrors(cworkers.count > 0, "No cworkers!");
    CheckNTErrors(uworkers.count > 0, "No uworkers!");
    CheckNTErrors(bworkers.count > 0, "No bworkers!");
    CheckNTErrors(aworkers.count > 0, "No pworkers!");

    bool isToUpdate = (optimizer != NULL);
    int activeJobCount = 0;
    int* active = new int[jworkers.count];

    InitForRun();

    /* run models on job workers */
    activeJobCount = RunModel(config, dataDistributor, active);

    /* update the model on the server side */
    if (activeJobCount > 0 && isToUpdate)
        RunUpdate(config, optimizer, active);

    WaitForFinishing(active, isToUpdate);

    for (int i = 0; i < jworkers.count; i++) {
        XWorkerJob* worker = (XWorkerJob*)jworkers[i];
        worker->Clear();
    }

    delete[] active;

    return activeJobCount > 0;
}
/*
run the model (for one time). Basically this is a map-reduce process.
>> config - the configuration
>> dataDistributor - data distributor
>> optimizer - the optimization method
<< return - if we can fetch the new data
*/
run the model (for one time). Basically this is a map-reduce process.
run the model ( for one time ) . Basically this is a map - reduce process .
bool XLeaderPS::Run(XConfig* config, DataDistributeBase* dataDistributor, XOptimizer* optimizer)
{
    CheckNTErrors(jworkers.count > 0, "No jworkers!");
    CheckNTErrors(cworkers.count > 0, "No cworkers!");
    CheckNTErrors(uworkers.count > 0, "No uworkers!");
    CheckNTErrors(bworkers.count > 0, "No bworkers!");
    CheckNTErrors(aworkers.count > 0, "No pworkers!");
    bool isToUpdate = (optimizer != NULL);
    int activeJobCount = 0;
    int* active = new int[jworkers.count];
    InitForRun();
    activeJobCount = RunModel(config, dataDistributor, active);
    if (activeJobCount > 0 && isToUpdate)
        RunUpdate(config, optimizer, active);
    WaitForFinishing(active, isToUpdate);
    for (int i = 0; i < jworkers.count; i++) {
        XWorkerJob* worker = (XWorkerJob*)jworkers[i];
        worker->Clear();
    }
    delete[] active;
    return activeJobCount > 0;
}
bool XLeaderPS :: Run ( XConfig * config , DataDistributeBase * dataDistributor , XOptimizer * optimizer ) { CheckNTErrors ( jworkers . count > 0 , " " ) ; CheckNTErrors ( cworkers . count > 0 , " " ) ; CheckNTErrors ( uworkers . count > 0 , " " ) ; CheckNTErrors ( bworkers . count > 0 , " " ) ; CheckNTErrors ( aworkers . count > 0 , " " ) ; bool isToUpdate = ( optimizer != NULL ) ; int activeJobCount = 0 ; int * active = new int [ jworkers . count ] ; InitForRun ( ) ; activeJobCount = RunModel ( config , dataDistributor , active ) ; if ( activeJobCount > 0 && isToUpdate ) RunUpdate ( config , optimizer , active ) ; WaitForFinishing ( active , isToUpdate ) ; for ( int i = 0 ; i < jworkers . count ; i ++ ) { XWorkerJob * worker = ( XWorkerJob * ) jworkers [ i ] ; worker -> Clear ( ) ; } delete [ ] active ; return activeJobCount > 0 ; }
run the model (for one time).
run the model ( for one time ) .
/* run models on job workers */ /* update the model on the server side */