roll back

source/train/XLeader.cpp

@@ -125,78 +125,6 @@ void XLeader::InitForRun()
     }
 }
 
-/* mark the workers as LOCKED */
-void XLeader::LockWorkers()
-{
-    int paramNum = serverModel.paramNum;
-
-    for (int i = 0; i < jworkers.count; i++) {
-        XWorkerJob * worker = (XWorkerJob*)jworkers[i];
-        worker->ResetJobCount();
-        worker->LockCount();
-    }
-
-    for (int i = 0; i < cworkers.count; i++) {
-        XWorkerCollect * worker = (XWorkerCollect*)cworkers[i];
-        worker->ResetJobCount();
-        worker->LockCount();
-    }
-
-    for (int i = 0; i < uworkers.count; i++) {
-        XWorkerUpdate * worker = (XWorkerUpdate*)uworkers[i];
-        worker->ResetJobCount(paramNum);
-        worker->LockCount();
-    }
-
-    for (int i = 0; i < bworkers.count; i++) {
-        XWorkerBroadcast * worker = (XWorkerBroadcast*)bworkers[i];
-        worker->ResetJobCount(paramNum);
-        worker->LockCount();
-    }
-}
-
-/*
-wait for unlocked workers (i.e., all workers finish their jobs).
-It is used as a response to LockWorkers(). It tries to lock the mutex
-in each worker. Because the mutex has been locked in LockWorkers(),
-WaitForUnlockedWorkers() must wait until the worker finishes its job and
-unlock the mutex. In this way, WaitForUnlockedWorkers() would return only 
-if all workers finish their jobs.
->> activeJobWorkers - flags that indicat whether a job worker is active
-*/
-void XLeader::WaitForUnlockedWorkers(const int * activeJobWorkers)
-{
-    int activeNum = 0;
-    for (int i = 0; i < jworkers.count; i++) {
-        XWorkerJob * worker = (XWorkerJob*)jworkers[i];
-        if (activeJobWorkers[i] > 0) {
-            worker->LockCount();
-            CheckNTErrors(worker->IsFinished(), 
-                          "Something is wrong with the worker!");
-            worker->UnlockCount();
-            activeNum++;
-        }
-    }
-
-    XList workers;
-    workers.AddList(&cworkers);
-    workers.AddList(&uworkers);
-    workers.AddList(&bworkers);
-
-    for (int i = 0; i < workers.count; i++) {
-        XWorker * worker = (XWorker*)workers[i];
-        if (activeNum > 0) {
-            worker->LockCount();
-            CheckNTErrors(worker->IsFinished(),
-                          "Something is wrong with the worker!");
-            worker->UnlockCount();
-        }
-        else {
-            worker->UnlockCount();
-        }
-    }
-}
-
 /* get loss */
 float XLeader::GetLoss()
 {
@@ -350,7 +278,7 @@ bool XLeader::Run(XConfig * config, DataDistributeBase * dataDistributor,
     
     InitForRun();
 
-    LockWorkers();
+    //LockWorkers();
 
     for (int i = 0; i < jworkers.count; i++)
         active[i] = 0;
@@ -380,9 +308,6 @@ bool XLeader::Run(XConfig * config, DataDistributeBase * dataDistributor,
             /* job in queue 1: make a record of the run */
             worker->AddJobRecord();
 
-            /* job in quene 1: finish the job */
-            worker->AddJobCountFinished();
-
             active[i] = 1;
             activeJobCount++;
         }
@@ -415,8 +340,7 @@ bool XLeader::Run(XConfig * config, DataDistributeBase * dataDistributor,
         collecter->AddJobUpdateAll(&members, &membersAll, &serverModel, 
                                    optimizer, updater, broadcaster);
         collecter->AddJobCollectOther(&memberRecords, &serverRecord);
-        collecter->AddJobCountFinished();
-
+        
         /* jobs in queue 2: collect the (gradient) data and other stuff. This
            is a reduce process. */
         //collecter->AddJobCollect(&members, &serverModel);
@@ -430,11 +354,9 @@ bool XLeader::Run(XConfig * config, DataDistributeBase * dataDistributor,
            not involved in this run. */
         //broadcaster->AddJobBroadcast(&serverModel, &membersAll);
 
-        //WaitForFinishing();
+        WaitForFinishing();
     }
 
-    WaitForUnlockedWorkers(active);
-
     for (int i = 0; i < jworkers.count; i++) {
         XWorkerJob * worker = (XWorkerJob*)jworkers[i];
         worker->Clear();

source/train/XWorker.cpp

@@ -38,16 +38,12 @@ XWorker::XWorker()
     id = -1;
     state = XWORKER_UNSTARTED;
     isInstantRun = false;
-    jobCountFinished = 0;
-    jobCountExpected = -1;
-    MUTEX_INIT(countMutex);
 }
 
 /* de-constructor */
 XWorker::~XWorker()
 {
     Stop();
-    MUTEX_DELE(countMutex);
 }
 
 /* set device id */
@@ -113,73 +109,5 @@ bool XWorker::IsEmpty()
 {
     return queue.IsEmpty();
 }
-    
-/*
-reset the job counts
->> myJobCountExpected - expected count
->> myJobCountFinished - current count
-*/
-void XWorker::ResetJobCount(int myJobCountExpected, int myJobCountFinished)
-{
-    jobCountExpected = myJobCountExpected;
-    jobCountFinished = myJobCountFinished;
-}
-
-/* lock the count mutex */
-void XWorker::LockCount()
-{
-    MUTEX_LOCK(countMutex);
-}
-
-/* unlock the count mutex */
-void XWorker::UnlockCount()
-{
-    MUTEX_UNLOCK(countMutex);
-}
-
-/* check if the worker finishes its job */
-bool XWorker::IsFinished()
-{
-    if (jobCountExpected <= 0)
-        return true;
-
-    if (jobCountExpected == jobCountFinished)
-        return true;
-
-    return false;
-}
-    
-/* count a finished job */
-void XWorker::CountFinishedJob()
-{
-    jobCountFinished++;
-    
-    if(jobCountExpected < 0 || jobCountFinished >= jobCountExpected){
-        MUTEX_UNLOCK(countMutex);
-    }
-    else{
-        ShowNTErrors("More jobs than expected!");
-    }
-}
-
-/* wrapper of CountFinishedJob() */
-void XWorker::CountFinished(XList * args)
-{
-    XWorker * worker = (XWorker*)args->GetItem(0);
-    
-    worker->CountFinishedJob();
-}
-    
-/* add a new job of counting finished jobs */
-void XWorker::AddJobCountFinished()
-{
-    XList args;
-    args.Add(this);
-    
-    if (isInstantRun)
-        XWorker::CountFinished(&args);
-    else
-        queue.EnqueueJob((void*)(char*)XWorker::CountFinished, &args);
-}
 
 } /* end of the nts (NiuTrans.Tensor) namespace */

source/train/XWorker.h

@@ -64,15 +64,6 @@ protected:
     /* fire the flag of instant run */
     bool isInstantRun;
     
-    /* count how many jobs have been finished */
-    int jobCountFinished;
-    
-    /* the expected number of jobs */
-    int jobCountExpected;
-    
-    /* mutex for accessing the job counts */
-    MUTEX_HANDLE countMutex;
-
 public:
     /* constructor */
     XWorker();
@@ -109,30 +100,6 @@ public:
 
     /* whether the job queue is empty? */
     bool IsEmpty();
-    
-    /* reset the job counts */
-    void ResetJobCount(int myJobCountExpected = -1, int myJobCountFinished = 0);
-    
-    /* lock the count mutex */
-    void LockCount();
-
-    /* unlock the count mutex */
-    void UnlockCount();
-
-    /* check if the worker finishes its job */
-    bool IsFinished();
-    
-    /* count a finished job */
-    void CountFinishedJob();
-    
-    /* wrapper of CountFinishedJob() */
-    static
-    void CountFinished(XList * args);
-    
-    /* add a new job of counting finished jobs */
-    void AddJobCountFinished();
-    
-    
 };
 
 }

source/train/XWorkerBroadcast.cpp

@@ -191,17 +191,10 @@ bool XWorkerBroadcast::AddJobBroadcastSingle(XModel * source, XList * targetList
     args.AddList(targetList);
     args.AddInt(pid);
 
-    XList argsNULL;
-    argsNULL.Add(this);
-
-    if (isInstantRun) {
+    if (isInstantRun)
         XWorkerBroadcast::BroadcastSingle(&args);
-        XWorkerBroadcast::CountFinished(&argsNULL);
-    }
-    else {
+    else
         queue.EnqueueJob((void*)(char*)XWorkerBroadcast::BroadcastSingle, &args);
-        queue.EnqueueJob((void*)(char*)XWorkerBroadcast::CountFinished, &argsNULL);
-    }
 
     return true;
 }

source/train/XWorkerUpdate.cpp

@@ -189,17 +189,10 @@ bool XWorkerUpdate::AddJobUpdateSingle(XModel * model, XList * members, int pid,
     args.Add(optimizer);
     args.Add(broadcaster);
 
-    XList argsNULL;
-    argsNULL.Add(this);
-
-    if (isInstantRun) {
+    if (isInstantRun)
         XWorkerUpdate::UpdateSingle(&args);
-        XWorkerUpdate::CountFinished(&argsNULL);
-    }
-    else {
+    else
         queue.EnqueueJob((void*)(char*)XWorkerUpdate::UpdateSingle, &args);
-        queue.EnqueueJob((void*)(char*)XWorkerUpdate::CountFinished, &argsNULL);
-    }
 
     return true;
 }