bug fixes for dealing with inactive job workers

6a0f0557 · xiaotong · 51e02fd6 · 6a0f0557 · 6a0f0557 · 6a0f0557
Commit 6a0f0557 authored Mar 31, 2021 by xiaotong
--- a/source/train/XLeader.cpp
+++ b/source/train/XLeader.cpp
@@ -306,28 +306,27 @@ void XLeader::AddCollectWorker(DATA_COLLECT_TYPE mode)
    cworkers.Add(worker);
 }
-/* 
+/* add model-update workers */
-add a model-update worker 
+void XLeader::AddUpdateWorker(int n)
->> model - the model
-*/
-void XLeader::AddUpdateWorker(XModel * model)
 {
-    XWorkerUpdate * worker = new XWorkerUpdate();
+    for (int i = 0; i < n; i++) {
-    uworkers.Add(worker);
+        XWorkerUpdate* worker = new XWorkerUpdate();
+        uworkers.Add(worker);
+    }
 }
-/* add a data-broadcasting worker */
+/*  add a data-broadcasting worker */
 void XLeader::AddBroadcastWorker()
 {
-    XWorkerBroadcast * worker = new XWorkerBroadcast();
+    XWorkerBroadcast* worker = new XWorkerBroadcast();
    bworkers.Add(worker);
 }
 /* 
-add a parameter worker (or a pipeline) 
+add parameter worker (or a pipeline) 
->> n - number of parameters
+>> n - number of workers
 */
-void XLeader::AddParamterWorker(int n)
+void XLeader::AddAuxiliaryWorker(int n)
 {
    for (int i = 0; i < n; i++) {
        XWorker * worker = new XWorker();
@@ -349,17 +348,7 @@ void XLeader::DestroyParamMap()
 /* generate the map of parameters */
 void XLeader::MakeParamMap()
 {
-    int modelCount = 0;
+    int modelCount = CountModels();
-    for (int i = 0; i < jworkers.count; i++) {
-        XWorker * worker = (XWorker*)jworkers[i];
-        if (worker->GetWorkerType() == XWORKER_TYPE_JOB) {
-            modelCount += worker->GetModelNum();
-            CheckNTErrors(worker->GetModelNum() == 1, "Wrong model number!");
-        }
-        else {
-            ShowNTErrors("TODO: support a new XWorker type!");
-        }
-    }
    if(modelCount != modelNum){
        DestroyParamMap();
@@ -390,4 +379,22 @@ void XLeader::MakeParamMap()
    modelNum = modelCount;
 }
+/* count all the models */
+int XLeader::CountModels()
+{
+    int modelCount = 0;
+    for (int i = 0; i < jworkers.count; i++) {
+        XWorker* worker = (XWorker*)jworkers[i];
+        if (worker->GetWorkerType() == XWORKER_TYPE_JOB) {
+            modelCount += worker->GetModelNum();
+            CheckNTErrors(worker->GetModelNum() == 1, "Wrong model number!");
+        }
+        else {
+            ShowNTErrors("TODO: support a new XWorker type!");
+        }
+    }
+    return modelCount;
+}
 } /* end of the nts (NiuTrans.Tensor) namespace */
--- a/source/train/XLeader.h
+++ b/source/train/XLeader.h
@@ -161,20 +161,23 @@ public:
    /* add a data-collecting worker */
    void AddCollectWorker(DATA_COLLECT_TYPE mode = DATA_COLLECT_P2P);
-    /* add a model-update worker */
+    /* add model-update workers */
-    void AddUpdateWorker(XModel * model);
+    void AddUpdateWorker(int n = 1);
    /* add a data-broadcasting worker */
    void AddBroadcastWorker();
-    /* add a parameter worker (or a pipeline) */
+    /* add auxiliary worker (or a pipeline) */
-    void AddParamterWorker(int n);
+    void AddAuxiliaryWorker(int n);
    /* destroy the parameter map (and gradient map) */
    void DestroyParamMap();
    /* generate the map of parameters */
    void MakeParamMap();
+    /* count all the models */
+    int CountModels();
 };
 }

--- a/source/train/XLeaderAllReduce.cpp
+++ b/source/train/XLeaderAllReduce.cpp
@@ -57,8 +57,8 @@ void XLeaderAllReduce::MakeAll(XConfig * config, XModel * model, const int * dev
    Init();
    AddJobWorker(model, jobWorkerNum, devIDs);
    AddCollectWorker();
-    for(int i = 0; i < jobWorkerNum; i++)
+    AddUpdateWorker();
-        AddUpdateWorker(model);
+    AddAuxiliaryWorker(CountModels());
    XLeader::MakeAll(config, model);
 }
@@ -168,8 +168,8 @@ int XLeaderAllReduce::RunModel(XConfig* config, DataDistributeBase* dataDistribu
            /* job in queue 1: run the model */
            worker->AddJobNeuralNet(jmodel,
-                worker->GetInput(), worker->GetOutput(),
+                                    worker->GetInput(), worker->GetOutput(),
-                worker->GetGold(), worker->GetLoss());
+                                    worker->GetGold(), worker->GetLoss());
            /* job in queue 1: make a record of the run */
            worker->AddJobRecord(&serverRecord);
@@ -193,37 +193,23 @@ update the model in a standard server-worker manner
 */
 void XLeaderAllReduce::RunUpdate(XConfig* config, XOptimizer* optimizer, const int* active)
 {
-    /* workers */
    XWorkerCollect* collecter = (XWorkerCollect*)cworkers.GetItem(0);
-    XWorkerUpdate* updater = (XWorkerUpdate*)uworkers.GetItem(0);
-    XWorkerBroadcast* broadcaster = (XWorkerBroadcast*)bworkers.GetItem(0);
+    CheckNTErrors(uworkers.count >= modelNum, "No enough updaters!");
    /* parameter map */
    MakeParamMap();
    /* all member models */
    XList membersAll(jworkers.count);
-    /* job queues */
-    XList jobQueues;
    for (int i = 0; i < jworkers.count; i++) {
        XWorkerJob* worker = (XWorkerJob*)jworkers[i];
        membersAll.Add(worker->GetModel());
    }
-    for (int i = 0; i < aworkers.count; i++) {
+    /* we reduce gradient across all job workers and update the parameter
-        XWorker* worker = (XWorker*)aworkers[i];
+       on each job worker. */
-        jobQueues.Add(worker->GetJobQueue());
-    }
-    CheckNTErrors(jobQueues.count == serverModel.paramNum, "Incompatiable model!");
-    /* jobs in queue 2 (say jobQueue): collect the (gradient) data.
-    This is a reduce process. Then we add a job to to update the model. followed
-    by a job to broadcast the lastest parameters to workers. NOTE that we
-    would update a worker to the latest model parameters, even if it is not
-    involved in this run. */
    int finished = 0;
@@ -270,25 +256,27 @@ void XLeaderAllReduce::RunUpdate(XConfig* config, XOptimizer* optimizer, const i
            for (int n = 0, i = 0; n < jworkers.count; n++) {
                XWorkerJob* worker = (XWorkerJob*)jworkers[n];
                for (int m = 0; m < worker->GetModelNum(); m++, i++) {
+                    /* skip the inactive model */
+                    if (modelFlag[i] == 0)
+                        continue;
                    XTensorKeeper& paramWorker = paramMap[j][i];
                    /* isGradFinished is true only if the model finishes the computation
-                    (in another thread) */
+                       (in another thread) */
                    if (paramWorker.flag == PARAM_STATE_NOT_READY && paramWorker.tensor->isGradFinished) {
                        /* get the gradient */
-                        paramWorker.grad = paramWorker.tensor->grad;
+                        //paramWorker.grad = paramWorker.tensor->grad;                  
-                        /* the job queue of updating parameter j */
-                        XQueue* jobQueue = (XQueue*)jobQueues.GetItem(j);
                        /* data transmit */
-                        collecter->AddJobCollectDataP2P(jobQueue, paramWorker.grad, paramServer.grad);
+                        //collecter->AddJobCollectDataP2P(NULL, paramWorker.grad, paramServer.grad);
-                        collecter->AddJobEnqueueFinished(jobQueue);
+                        //collecter->AddJobEnqueueFinished();
                        /* We keep the worker parameter in a list. It would be used when we broadcast
-                        the updated paramter to the workers, that is, this is a list of worker
+                           the updated paramter to the workers, that is, this is a list of worker
-                        parameters. */
+                           parameters. */
                        paramList[j].Add(&paramWorker);
                        /* reset the flag */
@@ -297,19 +285,22 @@ void XLeaderAllReduce::RunUpdate(XConfig* config, XOptimizer* optimizer, const i
                        finishedCount[j]++;
                        /* we call model update (in another thread) and then
-                        broadcast the new parameters to member models
+                           broadcast the new parameters to member models
-                        (in another thread) */
+                           (in another thread) */
                        if (finishedCount[j] == activeModelCount) {
                            paramServer.flag = PARAM_STATE_COLLECTED;
-                            if (updater != NULL) {
-                                /* update the parameters */
-                                updater->AddJobUpdate(jobQueue, &paramServer, optimizer);
-                                updater->AddJobEnqueueFinished(jobQueue);
-                                /* broadcast the new parameter to other models */
+                            /* call the all-reduce method to collect the gradient and share
-                                broadcaster->AddJobBroadcast(jobQueue, &paramServer, &paramList[j]);
+                               the gradient sum across models */
-                                broadcaster->AddJobEnqueueFinished(jobQueue);
+                            /* update on every model. NOTE THAT we do not worry about the
+                               inconsistence issue of updated parameters across models because
+                               the all-reduce method can garantee that the model shared the same
+                               copy of the gradient. */
+                            for (int k = 0; k < modelNum; k++) {
+                                XWorkerUpdate* updater = (XWorkerUpdate*)uworkers[k];
+                                updater->AddJobUpdate(NULL, &paramServer, optimizer);
+                                updater->AddJobEnqueueFinished();
                            }
                        }
                        else if (finishedCount[j] > activeModelCount) {

--- a/source/train/XLeaderPS.cpp
+++ b/source/train/XLeaderPS.cpp
@@ -55,9 +55,9 @@ void XLeaderPS::MakeAll(XConfig * config, XModel * model, const int * devIDs, co
    Init();
    AddJobWorker(model, jobWorkerNum, devIDs);
    AddCollectWorker();
-    AddUpdateWorker(model);
+    AddUpdateWorker();
    AddBroadcastWorker();
-    AddParamterWorker(model->paramNum);
+    AddAuxiliaryWorker(model->paramNum);
    XLeader::MakeAll(config, model);
 }
@@ -253,6 +253,10 @@ void XLeaderPS::RunUpdate(XConfig* config, XOptimizer* optimizer, const int* act
        }
    }
+    if (activeModelCount != jworkers.count) {
+        int nnn = 0;
+    }
    XList* paramList = new XList[serverModel.paramNum];
    CheckNTErrors(modelCount == modelNum, "Wrong model number!");
@@ -276,6 +280,11 @@ void XLeaderPS::RunUpdate(XConfig* config, XOptimizer* optimizer, const int* act
            for (int n = 0, i = 0; n < jworkers.count; n++) {
                XWorkerJob* worker = (XWorkerJob*)jworkers[n];
                for (int m = 0; m < worker->GetModelNum(); m++, i++) {
+                    /* skip the inactive model */
+                    if (modelFlag[i] == 0)
+                        continue;
                    XTensorKeeper& paramWorker = paramMap[j][i];
                    /* isGradFinished is true only if the model finishes the computation