| 279 | } |
| 280 | |
| 281 | void PSGPUWorker::TrainFiles() { |
| 282 | VLOG(0) << "Begin to train files"; |
| 283 | platform::SetNumThreads(1); |
| 284 | platform::Timer timeline; |
| 285 | timeline.Start(); |
| 286 | |
| 287 | int total_ins_num = 0; |
| 288 | #if defined(PADDLE_WITH_NCCL) || defined(PADDLE_WITH_RCCL) |
| 289 | platform::SetDeviceId(thread_id_); |
| 290 | #elif defined(PADDLE_WITH_XPU_BKCL) |
| 291 | platform::SetXPUDeviceId(thread_id_); |
| 292 | #endif |
| 293 | |
| 294 | // how to accumulate fetched values here |
| 295 | device_reader_->Start(); |
| 296 | int cur_batch; |
| 297 | int batch_cnt = 0; |
| 298 | |
| 299 | // async infershape |
| 300 | pack_is_end_.store(false); |
| 301 | if (scope_num_ != 1) { |
| 302 | for (size_t i = 0; i < thread_scope_vec_.size(); i++) { |
| 303 | TaskData task; |
| 304 | task.scope = thread_scope_vec_[i]; |
| 305 | free_task_queue_.Push(task); |
| 306 | } |
| 307 | thread_count_.store(task_threads_num_); |
| 308 | task_threads_.reserve(task_threads_num_); |
| 309 | for (int i = 0; i < task_threads_num_; i++) { |
| 310 | task_threads_.emplace_back(std::thread([this]() -> void { |
| 311 | while (true) { |
| 312 | auto pack = device_reader_->get_pack(nullptr); |
| 313 | if (pack == nullptr) { |
| 314 | int thread_num = thread_count_.fetch_sub(1); |
| 315 | if (thread_num == 1) { |
| 316 | pack_is_end_.store(true); |
| 317 | } |
| 318 | return; |
| 319 | } |
| 320 | auto task = free_task_queue_.Pop(); |
| 321 | task.pack = pack; |
| 322 | task.ins_num = pack->ins_num(); |
| 323 | device_reader_->PackToScope(task.pack, task.scope); |
| 324 | for (size_t i = 0; i < ops_.size(); i++) { |
| 325 | auto& op = ops_[i]; |
| 326 | bool need_skip = false; |
| 327 | for (auto t = 0u; t < skip_ops_.size(); ++t) { |
| 328 | if (op->Type().find(skip_ops_[t]) != std::string::npos) { |
| 329 | need_skip = true; |
| 330 | break; |
| 331 | } |
| 332 | } |
| 333 | if (!need_skip) { |
| 334 | paddle::framework::RuntimeContext ctx( |
| 335 | op->Inputs(), op->Outputs(), *task.scope); |
| 336 | op->RuntimeInferShape(*task.scope, place_, ctx); |
| 337 | } |
| 338 | } |
no test coverage detected