updates of worker pipeline

959545df · xiaotong · 7c1aeb5c · 959545df · 959545df · 959545df
Commit 959545df authored Mar 19, 2021 by xiaotong
--- a/source/train/XLeader.cpp
+++ b/source/train/XLeader.cpp
@@ -394,15 +394,15 @@ bool XLeader::Run(XConfig * config, DataDistributeBase * dataDistributor,
            isDataOK = false;
        else {
            /* job in queue 1: refresh the model */
-            worker->AddJobRefresh(worker->GetJobQueue(), jmodel);
+            worker->AddJobRefresh(jmodel);
            /* job in queue 1: run the model */
-            worker->AddJobNeuralNet(worker->GetJobQueue(), jmodel,
+            worker->AddJobNeuralNet(jmodel,
                                    worker->GetInput(), worker->GetOutput(), 
                                    worker->GetGold(), worker->GetLoss());
            /* job in queue 1: make a record of the run */
-            worker->AddJobRecord(worker->GetJobQueue(), &serverRecord);
+            worker->AddJobRecord(&serverRecord);
            /* job in queue 1: mark finished */
            worker->AddJobEnqueueFinished();
@@ -424,6 +424,9 @@ bool XLeader::Run(XConfig * config, DataDistributeBase * dataDistributor,
        /* 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());
@@ -431,13 +434,18 @@ bool XLeader::Run(XConfig * config, DataDistributeBase * dataDistributor,
                members.Add(worker->GetModel());
        }
+        for (int i = 0; i < pworkers.count; i++) {
+            XWorker * worker = (XWorker*)pworkers[i];
+            jobQueues.Add(worker->GetJobQueue());
+        }
        /* jobs in queue 2: collect the (gradient) data and other stuff. This
           is a reduce process. The collector will add a job in queue 3
           to update the model. The updater will  add a job job in queue 4 to 
           broadcast the lastest parameters to workers. NOTE that we would update 
           a worker to the laster model parameters, even if it is not involved 
           in this run. */
-        collecter->AddJobUpdateAll(collecter->GetJobQueue(),
+        collecter->AddJobUpdateAll(&jobQueues,
                                   &members, &membersAll, &serverModel, 
                                   optimizer, updater, broadcaster);
        collecter->AddJobEnqueueFinished();
@@ -455,57 +463,4 @@ bool XLeader::Run(XConfig * config, DataDistributeBase * dataDistributor,
    return isDataOK;
 }
-/* wait until all workers finish their job */
-void XLeader::WaitForFinishing(int sleepTime)
-{
-    while (1) {
-        bool finished = true;
-        if (finished) {
-            for (int i = 0; i < jworkers.count; i++) {
-                XWorkerJob* worker = (XWorkerJob*)jworkers[i];
-                if (worker->GetJobNum() > 0) {
-                    finished = false;
-                    break;
-                }
-            }
-        }
-        if (finished) {
-            for (int i = 0; i < cworkers.count; i++) {
-                XWorkerJob* worker = (XWorkerJob*)cworkers[i];
-                if (worker->GetJobNum() > 0) {
-                    finished = false;
-                    break;
-                }
-            }
-        }
-        if (finished) {
-            for (int i = 0; i < uworkers.count; i++) {
-                XWorkerJob* worker = (XWorkerJob*)uworkers[i];
-                if (worker->GetJobNum() > 0) {
-                    finished = false;
-                    break;
-                }
-            }
-        }
-        if (finished) {
-            for (int i = 0; i < bworkers.count; i++) {
-                XWorkerJob* worker = (XWorkerJob*)bworkers[i];
-                if (worker->GetJobNum() > 0) {
-                    finished = false;
-                    break;
-                }
-            }
-        }
-        if (finished)
-            break;
-        XSleep(sleepTime);
-    }
-}
 } /* end of the nts (NiuTrans.Tensor) namespace */
--- a/source/train/XLeader.h
+++ b/source/train/XLeader.h
@@ -161,9 +161,6 @@ public:
    /* run the model (for one time) */
    bool Run(XConfig * config, DataDistributeBase * dataDistributor, 
             XModel * model, XOptimizer * optimizer);
-    /* wait until all workers finish their job */
-    void WaitForFinishing(int sleepTime = SLEEP_TIME_IN_WAITING_FOR_JOBS);
 };
 }

--- a/source/train/XWorkerBroadcast.cpp
+++ b/source/train/XWorkerBroadcast.cpp
@@ -51,11 +51,14 @@ void XWorkerBroadcast::SetBroadcastMode(DATA_BROADCAST_TYPE myMode)
 /* 
 broadcast data for a parameter 
+>> jobQueue - the queue where we push jobs
 >> source - the data (as a model) that we want to broadcast
 >> targetList - the target places that we recieve the data
 >> pid - the parameter index
 */
-void XWorkerBroadcast::BroadcastDataSingle(XModel * source, XList * targetList, int pid)
+void XWorkerBroadcast::BroadcastDataSingle(XQueue * jobQueue, 
+                                           XModel * source, XList * targetList, 
+                                           int pid)
 {
    CheckNTErrors(source->params[pid].flag == PARAM_STATE_UPDATED,
                  "The parameter is not ready for broadcasting");
@@ -72,89 +75,29 @@ void XWorkerBroadcast::BroadcastDataSingle(XModel * source, XList * targetList, 
 }
 /* 
-broadcast data for a model
->> source - the data that we want to broadcast
->> targetList - the target places that we recieve the data
->> sleepTime - the waiting time in broadcasting
-*/
-void XWorkerBroadcast::BroadcastData(XModel * source, XList * targetList, int sleepTime)
-{
-    int finished = 0;
-    int * finishedFlag = new int[source->paramNum];
-    memset(finishedFlag, 0, sizeof(int) * source->paramNum);
-    /* check */
-    for (int i = 0; i < targetList->count; i++) {
-        XModel * target = (XModel*)targetList->GetItem(i);
-        CheckNTErrors(source->paramNum == target->paramNum, "Incompatiable models!");
-    }
-    /* the major body of broadcasting */
-    while (1) {
-        for (int i = 0; i < source->paramNum; i++) {
-            if (source->params[i].flag == PARAM_STATE_UPDATED && finishedFlag[i] == 0) {
-                /* broadcasting */
-                BroadcastDataSingle(source, targetList, i);
-                /* counting */
-                finished += targetList->count;
-                finishedFlag[i] = 1;
-            }
-        }
-        if (finished == source->paramNum * targetList->count)
-            break;
-        XSleep(sleepTime);
-    }
-    delete[] finishedFlag;
-}
-/* 
 wrapper of BroadcastDataSingle 
 >> args - the list of arguments
 */
 void XWorkerBroadcast::BroadcastSingle(XList * args)
 {
-    XWorkerBroadcast * broadcaster = (XWorkerBroadcast*)args->GetItem(0);
+    int paramCount = 0;
-    XModel * source = (XModel*)args->GetItem(1);
+    XWorkerBroadcast * broadcaster = (XWorkerBroadcast*)args->GetItem(paramCount++);
+    XQueue * jobQueue = (XQueue*)args->GetItem(paramCount++);
+    XModel * source = (XModel*)args->GetItem(paramCount++);
    /* target models */
-    int targetNum = args->GetItemInt(2);
+    int targetNum = args->GetItemInt(paramCount++);
    XList target;
    for (int i = 0; i < targetNum; i++) {
-        XModel * model = (XModel*)args->GetItem(3 + i);
+        XModel * model = (XModel*)args->GetItem(paramCount++);
        target.Add(model);
    }
    /* parameter index */
-    int p = args->GetInt(3 + targetNum);
+    int p = args->GetInt(paramCount++);
-    broadcaster->BroadcastDataSingle(source, &target, p);
-}
-/* 
-wrapper of BroadcastData 
->> args - the list of arguments
-*/
-void XWorkerBroadcast::Broadcast(XList * args)
-{
-    //fprintf(stderr, "broadcast 0\n");
-    XWorkerBroadcast * broadcaster = (XWorkerBroadcast*)args->GetItem(0);
-    XModel * source = (XModel*)args->GetItem(1);
-    /* target models */
-    int targetNum = args->GetItemInt(2);
-    XList target;
-    for (int i = 0; i < targetNum; i++) {
-        XModel * model = (XModel*)args->GetItem(3 + i);
-        target.Add(model);
-    }
-    broadcaster->BroadcastData(source, &target, SLEEP_TIME_IN_BROADCASTING);
+    broadcaster->BroadcastDataSingle(jobQueue, source, &target, p);
-    //fprintf(stderr, "broadcast 1\n");
 }
 /* 
@@ -174,11 +117,12 @@ void XWorkerBroadcast::BroadcastP2P(XTensor * source, XTensor * target)
 /* 
 add a new job of broadcasting data (for a parameter)
+>> jobQueue - the queue that we push jobs here
 >> source - the data that we want to broadcast
 >> targetList - the target places that we recieve the data
 >> pid - the parameter index
 */
-bool XWorkerBroadcast::AddJobBroadcastSingle(XModel * source, XList * targetList, int pid)
+bool XWorkerBroadcast::AddJobBroadcastSingle(XQueue * jobQueue, XModel * source, XList * targetList, int pid)
 {
    CheckNTErrors(source != NULL, "no input source tensor!");
    CheckNTErrors(targetList != NULL, "no input target tensor list!");
@@ -186,6 +130,7 @@ bool XWorkerBroadcast::AddJobBroadcastSingle(XModel * source, XList * targetList
    XList args;
    args.Add(this);
+    args.Add(jobQueue);
    args.Add(source);
    args.AddInt(targetList->count);
    args.AddList(targetList);
@@ -199,28 +144,4 @@ bool XWorkerBroadcast::AddJobBroadcastSingle(XModel * source, XList * targetList
    return true;
 }
-/* 
-add a new job of broadcasting data (for a model)
->> source - the data that we want to broadcast
->> targetList - the target places that we recieve the data
-*/
-bool XWorkerBroadcast::AddJobBroadcast(XModel * source, XList * targetList)
-{
-    CheckNTErrors(source != NULL, "no input source tensor!");
-    CheckNTErrors(targetList != NULL, "no input target tensor list!");
-    XList args;
-    args.Add(this);
-    args.Add(source);
-    args.AddInt(targetList->count);
-    args.AddList(targetList);
-    if (isInstantRun)
-        XWorkerBroadcast::Broadcast(&args);
-    else
-        queue.EnqueueJob((void*)(char*)XWorkerBroadcast::Broadcast, &args);
-    return true;
-}
 }
--- a/source/train/XWorkerBroadcast.h
+++ b/source/train/XWorkerBroadcast.h
@@ -61,27 +61,17 @@ public:
    void SetBroadcastMode(DATA_BROADCAST_TYPE myMode);
    /* broadcast data for a parameter */
-    void BroadcastDataSingle(XModel * source, XList * targetList, int pid);
+    void BroadcastDataSingle(XQueue * jobQueue, XModel * source, XList * targetList, int pid);
-    /* broadcast data for a model */
-    void BroadcastData(XModel * source, XList * targetList, int sleepTime);
    /* wrapper of BroadcastDataSingle */
    static
    void BroadcastSingle(XList * args);
-    /* wrapper of BroadcastData */
-    static
-    void Broadcast(XList * args);
    /* P2P data broadcasting */
    void BroadcastP2P(XTensor * source, XTensor * target);
    /* add a new job of broadcasting data (for a parameter) */
-    bool AddJobBroadcastSingle(XModel * source, XList * targetList, int pid);
+    bool AddJobBroadcastSingle(XQueue * jobQueue, XModel * source, XList * targetList, int pid);
-    /* add a new job of broadcasting data (for a model) */
-    bool AddJobBroadcast(XModel * source, XList * targetList);
 };
 }

--- a/source/train/XWorkerCollect.cpp
+++ b/source/train/XWorkerCollect.cpp
@@ -54,6 +54,7 @@ new parameters to all models. NOTE that this method just collect graident
 from member models. Then it calls an XWorkerUpdate to update the parameters.
 The XWorkerUpdate also calls an XWorkerBroadcast to broadcast the new parameter
 to member models back. 
+>> jobQueues - queues that we process the jobs
 >> memberActive - member models that are active, i.e., have generated gradients
 >> memberAll -  all member models
 >> server - the server model
@@ -63,7 +64,8 @@ to member models back.
                 models
 >> sleepTime - waiting time in collecting
 */
-void XWorkerCollect::UpdateDataAll(XList * memberActive, XList * memberAll, XModel * server,
+void XWorkerCollect::UpdateDataAll(XList * jobQueues, XList * memberActive, XList * memberAll, 
+                                   XModel * server,
                                   XOptimizer * optimizer, XWorkerUpdate * updater, 
                                   XWorkerBroadcast * broadcaster, int sleepTime)
 {
@@ -83,6 +85,8 @@ void XWorkerCollect::UpdateDataAll(XList * memberActive, XList * memberAll, XMod
        CheckNTErrors(source->paramNum == server->paramNum, "Incompatiable models!");
    }
+    CheckNTErrors(jobQueues->count == server->paramNum, "Incompatiable model!");
    /* counts how many member models are collect for each parameters */
    int * finishedCount = new int[server->paramNum];
    memset(finishedCount, 0, sizeof(int) * server->paramNum);
@@ -125,7 +129,7 @@ void XWorkerCollect::UpdateDataAll(XList * memberActive, XList * memberAll, XMod
                        if (finishedCount[j] == memberActive->count) {
                            paramServer.flag = PARAM_STATE_COLLECTED;
                            if (updater != NULL) {
-                                updater->AddJobUpdate(updater->GetJobQueue(), 
+                                updater->AddJobUpdate((XQueue*)jobQueues->GetItem(j), 
                                                      server, memberAll, j, 
                                                      optimizer, broadcaster);
                                updater->AddJobEnqueueFinished();
@@ -155,29 +159,39 @@ void XWorkerCollect::UpdateDataAll(XList * memberActive, XList * memberAll, XMod
 /* wrapper of UpdateDataAll */
 void XWorkerCollect::UpdateAll(XList * args)
 {
-    XWorkerCollect * collecter = (XWorkerCollect*)args->GetItem(0);
+    int paramCount = 0;
-    int activeNum = args->GetInt(1);
+    XWorkerCollect * collecter = (XWorkerCollect*)args->GetItem(paramCount++);
+    int queueNum = args->GetInt(paramCount++);
+    XList jobQueues;
+    for (int i = 0; i < queueNum; i++) {
+        XQueue * queue = (XQueue*)args->GetItem(paramCount++);
+        jobQueues.Add(queue);
+    }
+    int activeNum = args->GetInt(paramCount++);
    XList memberActive;
    for (int i = 0; i < activeNum; i++) {
-        XModel * member = (XModel*)args->GetItem(2 + i);
+        XModel * member = (XModel*)args->GetItem(paramCount++);
        memberActive.Add(member);
    }
-    int allNum = args->GetInt(2 + activeNum);
+    int allNum = args->GetInt(paramCount++);
    XList memberAll;
    for (int i = 0; i < allNum; i++) {
-        XModel * member = (XModel*)args->GetItem(2 + activeNum + 1 + i);
+        XModel * member = (XModel*)args->GetItem(paramCount++);
        memberAll.Add(member);
    }
-    XModel * server = (XModel*)args->GetItem(2 + activeNum + 1 + allNum);
+    XModel * server = (XModel*)args->GetItem(paramCount++);
-    XOptimizer * optimizer = (XOptimizer*)args->GetItem(2 + activeNum + 1 + allNum + 1);
+    XOptimizer * optimizer = (XOptimizer*)args->GetItem(paramCount++);
-    XWorkerUpdate * updater = (XWorkerUpdate*)args->GetItem(2 + activeNum + 1 + allNum + 2);
+    XWorkerUpdate * updater = (XWorkerUpdate*)args->GetItem(paramCount++);
-    XWorkerBroadcast * broadcaster = (XWorkerBroadcast*)args->GetItem(2 + activeNum + 1 + allNum + 3);
+    XWorkerBroadcast * broadcaster = (XWorkerBroadcast*)args->GetItem(paramCount++);
-    collecter->UpdateDataAll(&memberActive, &memberAll, server, 
+    collecter->UpdateDataAll(&jobQueues, &memberActive, &memberAll, server, 
                             optimizer, updater, broadcaster, 
                             SLEEP_TIME_IN_COLLECTING);
 }
@@ -243,7 +257,7 @@ void XWorkerCollect::CollectAllReduce(XList * all)
 /* 
 add a new job of collecting data, update the parameter and 
 broadcast the new parameter
->> myQueue - the queue where we push the job
+>> jobQueues - the queues that we would use in following jobs
 >> memberActive - member models that are active, i.e., have generated gradients
 >> memberAll -  all member models
 >> server - the server model
@@ -253,8 +267,11 @@ broadcast the new parameter
                 models
 << return - successful or not
 */
-bool XWorkerCollect::AddJobUpdateAll(XQueue * myQueue, XList * memberActive, XList * memberAll, XModel * server,
+bool XWorkerCollect::AddJobUpdateAll(XList * jobQueues,
-                                     XOptimizer * optimizer, XWorkerUpdate * updater, XWorkerBroadcast * broadcaster)
+                                     XList * memberActive, XList * memberAll, 
+                                     XModel * server,
+                                     XOptimizer * optimizer, 
+                                     XWorkerUpdate * updater, XWorkerBroadcast * broadcaster)
 {
    CheckNTErrors(memberActive != NULL, "No input (active) member list!");
    CheckNTErrors(memberAll != NULL, "No input (all) member list!");
@@ -265,6 +282,8 @@ bool XWorkerCollect::AddJobUpdateAll(XQueue * myQueue, XList * memberActive, XLi
    XList args;
    args.Add(this);
+    args.AddInt(jobQueues->count);
+    args.AddList(jobQueues);
    args.AddInt(memberActive->count);
    args.AddList(memberActive);
    args.AddInt(memberAll->count);
@@ -274,12 +293,10 @@ bool XWorkerCollect::AddJobUpdateAll(XQueue * myQueue, XList * memberActive, XLi
    args.Add(updater);
    args.Add(broadcaster);
-    XQueue &q = myQueue != NULL ? *myQueue : queue;
    if (isInstantRun)
        XWorkerCollect::UpdateAll(&args);
    else
-        q.EnqueueJob((void*)(char*)XWorkerCollect::UpdateAll, &args);
+        queue.EnqueueJob((void*)(char*)XWorkerCollect::UpdateAll, &args);
    return true;
 }

--- a/source/train/XWorkerCollect.h
+++ b/source/train/XWorkerCollect.h
@@ -70,7 +70,7 @@ public:
       from member models. Then it calls an XWorkerUpdate to update the parameters.
       The XWorkerUpdate also calls an XWorkerBroadcast to broadcast the new parameter
       to member models back. */
-    void UpdateDataAll(XList * memberActive, XList * memberAll, XModel * server,
+    void UpdateDataAll(XList * jobQueues, XList * memberActive, XList * memberAll, XModel * server,
                       XOptimizer * optimizer, XWorkerUpdate * updater, XWorkerBroadcast * broadcaster, 
                       int sleepTime);
@@ -88,7 +88,7 @@ public:
    void CollectAllReduce(XList * all);
    /* add a new job of collecting data, update the parameter and broadcast the new parameter */
-    bool AddJobUpdateAll(XQueue * myQueue, XList * memberActive, XList * memberAll, XModel * server,
+    bool AddJobUpdateAll(XList * jobQueues, XList * memberActive, XList * memberAll, XModel * server,
                         XOptimizer * optimizer, XWorkerUpdate * updater, XWorkerBroadcast * broadcaster);
 };

--- a/source/train/XWorkerJob.cpp
+++ b/source/train/XWorkerJob.cpp
@@ -175,11 +175,10 @@ int XWorkerJob::GetPredictNum()
 /* 
 add a new job of model refreshment 
->> myQueue - the queue where we push the job
 >> myModel - the model
 << return - succeeded or not
 */
-bool XWorkerJob::AddJobRefresh(XQueue * myQueue, XModel * myModel)
+bool XWorkerJob::AddJobRefresh(XModel * myModel)
 {
    //fprintf(stderr, "refresh 0\n");
@@ -188,12 +187,10 @@ bool XWorkerJob::AddJobRefresh(XQueue * myQueue, XModel * myModel)
    XList args(1);
    args.Add(myModel);
-    XQueue &q = myQueue != NULL ? *myQueue : queue;
    if(isInstantRun)
        XModel::Refresh(&args);
    else
-        q.EnqueueJob((void*)(char*)XModel::Refresh, &args);
+        queue.EnqueueJob((void*)(char*)XModel::Refresh, &args);
    //fprintf(stderr, "refresh 1\n");
@@ -202,7 +199,6 @@ bool XWorkerJob::AddJobRefresh(XQueue * myQueue, XModel * myModel)
 /* 
 add a new job of neural network forward and backward computation (with the input) 
->> myQueue - the queue where we push the job
 >> myModel - the model
 >> inputs - inputs of the neural network
 >> outputs - outputs of the neural network
@@ -210,8 +206,9 @@ add a new job of neural network forward and backward computation (with the input
 >> losses - losses of the outputs respect to the gold standards
 << return - succeeded or not
 */
-bool XWorkerJob::AddJobNeuralNet(XQueue * myQueue, XModel * myModel,
+bool XWorkerJob::AddJobNeuralNet(XModel * myModel,
-                                 XList * inputs, XList * outputs, XList * golds, XList * losses)
+                                 XList * inputs, XList * outputs, 
+                                 XList * golds, XList * losses)
 {
    CheckNTErrors(myModel != NULL, "no input neural network!");
    CheckNTErrors(inputs != NULL, "no inputs of the model!");
@@ -224,12 +221,10 @@ bool XWorkerJob::AddJobNeuralNet(XQueue * myQueue, XModel * myModel,
    args.Add(golds);
    args.Add(losses);
-    XQueue &q = myQueue != NULL ? *myQueue : queue;
    if(isInstantRun)
        XModel::Run(&args);
    else
-        q.EnqueueJob((void*)(char*)XModel::Run, &args);
+        queue.EnqueueJob((void*)(char*)XModel::Run, &args);
    SetState(XWORKER_STARTED);
@@ -260,21 +255,18 @@ void XWorkerJob::RecordMeStatic(XList* args)
 /* 
 add a new job of recording the running of the nerual network 
->> myQueue - the queue where we push the job
 >> serverRecord - the model record on the server side
 */
-bool XWorkerJob::AddJobRecord(XQueue * myQueue, XNNRecord * serverRecord)
+bool XWorkerJob::AddJobRecord(XNNRecord * serverRecord)
 {
    XList args;
    args.Add(this);
    args.Add(serverRecord);
-    XQueue &q = myQueue != NULL ? *myQueue : queue;
    if (isInstantRun)
        XWorkerJob::RecordMeStatic(&args);
    else
-        q.EnqueueJob((void*)(char*)XWorkerJob::RecordMeStatic, &args);
+        queue.EnqueueJob((void*)(char*)XWorkerJob::RecordMeStatic, &args);
    return true;
 }

--- a/source/train/XWorkerJob.h
+++ b/source/train/XWorkerJob.h
@@ -107,13 +107,13 @@ public:
    int GetPredictNum();
    /* add a new job of model refreshment */
-    bool AddJobRefresh(XQueue * myQueue, XModel * myModel);
+    bool AddJobRefresh(XModel * myModel);
    /* add a new job of neural network forward and backward computation (with the input) */
-    bool AddJobNeuralNet(XQueue * myQueue, XModel * myModel, XList * inputs, XList * outputs, XList * golds, XList * losses);
+    bool AddJobNeuralNet(XModel * myModel, XList * inputs, XList * outputs, XList * golds, XList * losses);
    /* add a new job of recording the running of the nerual network */
-    bool AddJobRecord(XQueue * myQueue, XNNRecord * serverRecord);
+    bool AddJobRecord(XNNRecord * serverRecord);
 private:
    /* wrapper of RecordMe */

--- a/source/train/XWorkerUpdate.cpp
+++ b/source/train/XWorkerUpdate.cpp
@@ -54,13 +54,14 @@ XOptimizer * XWorkerUpdate::GetOptimizer()
 /* 
 update a parameter of a model 
+>> jobQueue - the queue that place the jobs called here
 >> model - the model that we want to update (on the server side)
 >> members - models that would share the updated parameters
 >> pid - the parameter index
 >> optimizer - the optimizer
 >> broadcaster - the worker that would broadcast the new parameter to members
 */
-void XWorkerUpdate::UpdateParameter(XModel * server, XList * members, int pid,
+void XWorkerUpdate::UpdateParameter(XQueue * jobQueue, XModel * server, XList * members, int pid,
                                    XOptimizer * optimizer, XWorkerBroadcast * broadcaster)
 {
@@ -78,7 +79,7 @@ void XWorkerUpdate::UpdateParameter(XModel * server, XList * members, int pid,
    server->params[pid].flag = PARAM_STATE_UPDATED;
    /* broadcast the new parameter to other models (in anotehr worker/thread) */
-    broadcaster->AddJobBroadcastSingle(server, members, pid);
+    broadcaster->AddJobBroadcastSingle(jobQueue, server, members, pid);
    broadcaster->AddJobEnqueueFinished();
 }
@@ -88,36 +89,40 @@ wrapper of UpdateParameter
 */
 void XWorkerUpdate::Update(XList * args)
 {
+    int paramCount = 0;
    CheckNTErrors(args != NULL && args->count >= 6, "Illegal argument list!");
-    XWorkerUpdate * updater = (XWorkerUpdate*)args->GetItem(0);
+    XWorkerUpdate * updater = (XWorkerUpdate*)args->GetItem(paramCount++);
-    XModel * server = (XModel*)args->GetItem(1);
+    XQueue * jobQueue = (XQueue*)args->GetItem(paramCount++);
-    int memNum = args->GetInt(2);
+    XModel * server = (XModel*)args->GetItem(paramCount++);
+    int memNum = args->GetInt(paramCount++);
    XList members;
    for (int i = 0; i < memNum; i++) {
-        XModel * member = (XModel*)args->GetItem(3 + i);
+        XModel * member = (XModel*)args->GetItem(paramCount++);
        members.Add(member);
    }
-    int pid = args->GetInt(3 + memNum);
+    int pid = args->GetInt(paramCount++);
-    XOptimizer * optimizer = (XOptimizer*)args->GetItem(3 + memNum + 1);
+    XOptimizer * optimizer = (XOptimizer*)args->GetItem(paramCount++);
-    XWorkerBroadcast * broadcaster = (XWorkerBroadcast*)args->GetItem(3 + memNum + 2);
+    XWorkerBroadcast * broadcaster = (XWorkerBroadcast*)args->GetItem(paramCount++);
    if(updater != NULL)
-        updater->UpdateParameter(server, &members, pid, optimizer, broadcaster);
+        updater->UpdateParameter(jobQueue, server, &members, pid, optimizer, broadcaster);
 }
 /* 
 add a new job of model update (for a parameter) 
->> myQueue - the queue where we push the job
+>> jobQueue - the queue for sub-jobs executed in the job
 >> model - the model that we want to update (on the server side)
 >> members - models that would share the updated parameters
 >> pid - the parameter index
 >> optimizer - the optimizer
 >> broadcaster - the worker that would broadcast the new parameter to members
 */
-bool XWorkerUpdate::AddJobUpdate(XQueue * myQueue, XModel * model, XList * members, int pid,
+bool XWorkerUpdate::AddJobUpdate(XQueue * jobQueue,
+                                 XModel * model, XList * members, int pid,
                                 XOptimizer * optimizer, XWorkerBroadcast * broadcaster)
 {
    CheckNTErrors(model != NULL, "No input model!");
@@ -128,6 +133,7 @@ bool XWorkerUpdate::AddJobUpdate(XQueue * myQueue, XModel * model, XList * membe
    XList args;
    args.Add(this);
+    args.Add(jobQueue);
    args.Add(model);
    args.AddInt(members->count);
    args.AddList(members);
@@ -135,12 +141,10 @@ bool XWorkerUpdate::AddJobUpdate(XQueue * myQueue, XModel * model, XList * membe
    args.Add(optimizer);
    args.Add(broadcaster);
-    XQueue &q = myQueue != NULL ? *myQueue : queue;
    if (isInstantRun)
        XWorkerUpdate::Update(&args);
    else
-        q.EnqueueJob((void*)(char*)XWorkerUpdate::Update, &args);
+        queue.EnqueueJob((void*)(char*)XWorkerUpdate::Update, &args);
    return true;
 }

--- a/source/train/XWorkerUpdate.h
+++ b/source/train/XWorkerUpdate.h
@@ -57,7 +57,7 @@ public:
    XOptimizer * GetOptimizer();
    /* update the parameter */
-    void UpdateParameter(XModel * server, XList * members, int pid,
+    void UpdateParameter(XQueue * jobQueue, XModel * server, XList * members, int pid,
                         XOptimizer * optimizer, XWorkerBroadcast * broadcaster);
@@ -67,7 +67,7 @@ public:
    /* add a new job of model update (for a parameter) */
-    bool AddJobUpdate(XQueue * myQueue, XModel * model, XList * members, int pid,
+    bool AddJobUpdate(XQueue * jobQueue, XModel * model, XList * members, int pid,
                      XOptimizer * optimizer, XWorkerBroadcast * broadcaster);
 };