|
|
|
|
@ -140,11 +140,19 @@ Status CifarOp::operator()() {
|
|
|
|
|
io_block_queues_[i]->Add(std::make_unique<IOBlock>(std::vector<int64_t>(), IOBlock::kDeIoBlockNone)));
|
|
|
|
|
}
|
|
|
|
|
return Status::OK();
|
|
|
|
|
} else { // not the last repeat. Acquire lock, sleeps master thread, wait for the wake-up from reset
|
|
|
|
|
} else { // not the last repeat.
|
|
|
|
|
RETURN_IF_NOT_OK(
|
|
|
|
|
io_block_queues_[(buf_cnt_++) % num_workers_]->Add(std::make_unique<IOBlock>(IOBlock::kDeIoBlockFlagEoe)));
|
|
|
|
|
RETURN_IF_NOT_OK(wp_.Wait()); // Master thread goes to sleep after it has made all the IOBlocks
|
|
|
|
|
wp_.Clear();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (epoch_sync_flag_) {
|
|
|
|
|
// If epoch_sync_flag_ is set, then master thread sleeps until all the worker threads have finished their job for
|
|
|
|
|
// the current epoch.
|
|
|
|
|
RETURN_IF_NOT_OK(WaitForWorkers());
|
|
|
|
|
}
|
|
|
|
|
// If not the last repeat, self-reset and go to loop again.
|
|
|
|
|
if (!IsLastIteration()) {
|
|
|
|
|
RETURN_IF_NOT_OK(Reset());
|
|
|
|
|
RETURN_IF_NOT_OK(sampler_->GetNextSample(&sampler_buffer));
|
|
|
|
|
}
|
|
|
|
|
UpdateRepeatAndEpochCounter();
|
|
|
|
|
@ -156,7 +164,7 @@ Status CifarOp::LaunchThreadsAndInitOp() {
|
|
|
|
|
RETURN_STATUS_UNEXPECTED("Pipeline init failed, Execution tree not set.");
|
|
|
|
|
}
|
|
|
|
|
RETURN_IF_NOT_OK(io_block_queues_.Register(tree_->AllTasks()));
|
|
|
|
|
RETURN_IF_NOT_OK(wp_.Register(tree_->AllTasks()));
|
|
|
|
|
RETURN_IF_NOT_OK(wait_for_workers_post_.Register(tree_->AllTasks()));
|
|
|
|
|
RETURN_IF_NOT_OK(
|
|
|
|
|
tree_->AllTasks()->CreateAsyncTask("Get cifar data block", std::bind(&CifarOp::ReadCifarBlockDataAsync, this)));
|
|
|
|
|
RETURN_IF_NOT_OK(tree_->LaunchWorkers(num_workers_, std::bind(&CifarOp::WorkerEntry, this, std::placeholders::_1)));
|
|
|
|
|
@ -175,7 +183,13 @@ Status CifarOp::WorkerEntry(int32_t worker_id) {
|
|
|
|
|
std::unique_ptr<IOBlock> io_block;
|
|
|
|
|
RETURN_IF_NOT_OK(io_block_queues_[worker_id]->PopFront(&io_block));
|
|
|
|
|
while (io_block != nullptr) {
|
|
|
|
|
if (io_block->eoe() == true) {
|
|
|
|
|
if (io_block->wait() == true) {
|
|
|
|
|
// Sync io_block is a signal that master thread wants us to pause and sync with other workers.
|
|
|
|
|
// The last guy who comes to this sync point should reset the counter and wake up the master thread.
|
|
|
|
|
if (++num_workers_paused_ == num_workers_) {
|
|
|
|
|
wait_for_workers_post_.Set();
|
|
|
|
|
}
|
|
|
|
|
} else if (io_block->eoe() == true) {
|
|
|
|
|
RETURN_IF_NOT_OK(out_connector_->Add(worker_id, std::make_unique<DataBuffer>(0, DataBuffer::kDeBFlagEOE)));
|
|
|
|
|
buffer_id = worker_id;
|
|
|
|
|
} else if (io_block->eof() == true) {
|
|
|
|
|
@ -243,9 +257,9 @@ void CifarOp::Print(std::ostream &out, bool show_all) const {
|
|
|
|
|
|
|
|
|
|
// Reset Sampler and wakeup Master thread (functor)
|
|
|
|
|
Status CifarOp::Reset() {
|
|
|
|
|
MS_LOG(DEBUG) << Name() << " performing a self-reset.";
|
|
|
|
|
RETURN_IF_NOT_OK(sampler_->ResetSampler());
|
|
|
|
|
row_cnt_ = 0;
|
|
|
|
|
wp_.Set(); // wake up master thread after reset is done
|
|
|
|
|
return Status::OK();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
Lixia Chen
5 years ago