Skip to content

N
NiuTrans.Tensor
Commit 59ccf22d by xiaotong
Options

reduce the waiting time in waiting and clean the useless code

parent 74baf792
正在显示 包含 5 行增加 和 175 行删除
source/train/XWorkerBroadcast.h
......@@ -35,7 +35,7 @@
namespace nts { // namespace nts(NiuTrans.Tensor)
#define SLEEP_TIME_IN_BROADCASTING 20
#define SLEEP_TIME_IN_BROADCASTING 5
/*
data broadcasting method
......
source/train/XWorkerCollect.cpp
......@@ -200,146 +200,6 @@ void XWorkerCollect::UpdateDataAll(XList * memberActive, XList * memberAll, XMod
XSleep(sleepTime);
}
/* reset the flags */
//for (int j = 0; j < tp.count; j++)
// server->flags[j] = PARAM_STATE_COLLECTED;
delete[] finishedCount;
}
/*
collect data
>> sourceList - the list of data tensors we collect data from
>> target - the target tensor we place the result, that is
target += \sum_i source_i
*/
void XWorkerCollect::CollectData(XList * sourceList, XModel * target, long sleepTime)
{
TensorList & tp = target->params;
int finished = 0;
for (int j = 0; j < tp.count; j++)
target->flags[j] = PARAM_STATE_NOT_READY;
/* check */
for (int i = 0; i < sourceList->count; i++) {
TensorList & sp = ((XModel*)sourceList->GetItem(i))->params;
CheckNTErrors(sp.count == tp.count, "Incompatiable models!");
}
/* counts how many member models are collect for each parameters */
int * finishedCount = new int[tp.count];
memset(finishedCount, 0, sizeof(int) * tp.count);
//fprintf(stderr, "collect data in 0\n");
/* This is a simple implementation of the wait-and-collect process. But
there is a risk that some models are not available, that is, the
loop would never stop. A solution might be that we force the loop
to break after waiting for a short time. */
while (1) {
if (collectMode == DATA_COLLECT_P2P) {
for (int j = 0; j < tp.count; j++) {
/* tp[j]->isGradFinished is true only if the model finishes theA computation
(in another process) */
if (target->flags[j] != PARAM_STATE_NOT_READY || !tp[j]->isGradFinished)
continue;
/* check if all the models (or part of them) are ready */
for (int i = 0; i < sourceList->count; i++) {
XModel * source = (XModel*)sourceList->GetItem(i);
TensorList & sp = source->params;
/* sp[j]->isGradFinished is true only if the model finishes the computation
(in another process) */
if (source->flags[j] == PARAM_STATE_NOT_READY && sp[j]->isGradFinished) {
/* data transmit */
CollectP2P(sp[j]->grad, tp[j]->grad);
/* reset the flag */
source->flags[j] = PARAM_STATE_COLLECTED;
finished++;
finishedCount[j]++;
if (finishedCount[j] == sourceList->count) {
target->flags[j] = PARAM_STATE_COLLECTED;
}
}
}
}
}
else if (collectMode == DATA_COLLECT_REDUCESUM) {
for (int j = 0; j < tp.count; j++) {
bool ready = true;
/* tp[j]->isGradFinished is true only if the model finishes the computation
(in another process) */
if (target->flags[j] != PARAM_STATE_NOT_READY || !tp[j]->isGradFinished)
continue;
/* check if all the models (or part of them) are ready */
for (int i = 0; i < sourceList->count; i++) {
XModel * source = (XModel*)sourceList->GetItem(i);
TensorList & sp = source->params;
/* sp[j]->isGradFinished is true only if the model finishes the computation
(in another process) */
if (source->flags[j] == PARAM_STATE_COLLECTED ||
source->flags[j] == PARAM_STATE_UPDATED ||
!sp[j]->isGradFinished)
{
ready = false;
break;
}
else if(source->flags[j] == PARAM_STATE_NOT_READY){
source->flags[j] = PARAM_STATE_READY;
}
}
if (ready) {
XList tensorList(sourceList->count);
for (int i = 0; i < sourceList->count; i++) {
XModel * source = (XModel*)sourceList->GetItem(i);
TensorList & sp = source->params;
tensorList.Add(sp.GetItem(j)->grad);
}
/* data transmit */
CollectReduceSum(&tensorList, tp.GetItem(j)->grad);
/* reset the flags */
for (int i = 0; i < sourceList->count; i++) {
XModel * source = (XModel*)sourceList->GetItem(i);
source->flags[j] = PARAM_STATE_COLLECTED;
}
target->flags[j] = PARAM_STATE_COLLECTED;
finished += sourceList->count;
}
}
}
else {
ShowNTErrors("Unsupported data collection mode!");
}
/* the collection finishes if all data tensors are processed */
if (finished == tp.count * sourceList->count)
break;
XSleep(sleepTime);
}
//fprintf(stderr, "collect data in 1\n");
/* reset the flags */
//for (int j = 0; j < tp.count; j++)
// target->flags[j] = PARAM_STATE_COLLECTED;
delete[] finishedCount;
}
......@@ -373,29 +233,6 @@ void XWorkerCollect::UpdateAll(XList * args)
SLEEP_TIME_IN_COLLECTING);
}
/* wrapper of CollectData */
void XWorkerCollect::Collect(XList * args)
{
//fprintf(stderr, "collect data 0\n");
XWorkerCollect * collecter = (XWorkerCollect*)args->GetItem(0);
int sourceNum = args->GetItemInt(1);
/* the source models */
XList source;
for (int i = 0; i < sourceNum; i++) {
XModel * model = (XModel*)args->GetItem(2 + i);
source.Add(model);
}
/* the target model */
XModel * target = (XModel*)args->GetItem(2 + sourceNum);
collecter->CollectData(&source, target, SLEEP_TIME_IN_COLLECTING);
//fprintf(stderr, "collect data 1\n");
}
/*
P2P data collection
target += source
......
source/train/XWorkerCollect.h
......@@ -37,8 +37,8 @@
namespace nts { // namespace nts(NiuTrans.Tensor)
#define SLEEP_TIME_IN_COLLECTING 20
#define SLEEP_TIME_IN_COLLECTING_OTHER 40
#define SLEEP_TIME_IN_COLLECTING 5
#define SLEEP_TIME_IN_COLLECTING_OTHER 5
/*
data collection method
......@@ -66,7 +66,7 @@ public:
void SetCollectMode(DATA_COLLECT_TYPE myMode);
/* collect the gradient data, update the parameters, and broadcast the
new parameters to all models. NOTE that this method just collect graident
new parameters to all models. NOTE that this method just collects graidents
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. */
......@@ -74,17 +74,10 @@ public:
XOptimizer * optimizer, XWorkerUpdate * updater, XWorkerBroadcast * broadcaster,
long sleepTime);
/* collect the gradient data (i.e., a reducer) */
void CollectData(XList * sourceList, XModel * target, long sleepTime);
/* wrapper of UpdateDataAll */
static
void UpdateAll(XList * args);
/* wrapper of CollectData */
static
void Collect(XList * args);
/* P2P data collection */
void CollectP2P(XTensor * source, XTensor * target);
......
source/train/XWorkerUpdate.h
......@@ -34,7 +34,7 @@
namespace nts { // namespace nts(NiuTrans.Tensor)
#define SLEEP_TIME_IN_MODEL_UPDATE 20
#define SLEEP_TIME_IN_MODEL_UPDATE 5
/* The class defines the model-update worker */
class XWorkerUpdate : public XWorker
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论