| 553 | |
| 554 | template <typename T> |
| 555 | void InMemoryDataFeed<T>::LoadIntoMemory() { |
| 556 | #ifdef _LINUX |
| 557 | if (!so_parser_name_.empty()) { |
| 558 | LoadIntoMemoryFromSo(); |
| 559 | return; |
| 560 | } |
| 561 | VLOG(3) << "LoadIntoMemory() begin, thread_id=" << thread_id_; |
| 562 | std::string filename; |
| 563 | while (this->PickOneFile(&filename)) { |
| 564 | VLOG(3) << "PickOneFile, filename=" << filename |
| 565 | << ", thread_id=" << thread_id_; |
| 566 | #ifdef PADDLE_WITH_BOX_PS |
| 567 | if (BoxWrapper::GetInstance()->UseAfsApi()) { |
| 568 | this->fp_ = BoxWrapper::GetInstance()->afs_manager->GetFile( |
| 569 | filename, this->pipe_command_); |
| 570 | } else { |
| 571 | #endif |
| 572 | int err_no = 0; |
| 573 | this->fp_ = fs_open_read(filename, &err_no, this->pipe_command_, true); |
| 574 | #ifdef PADDLE_WITH_BOX_PS |
| 575 | } |
| 576 | #endif |
| 577 | PADDLE_ENFORCE_EQ(this->fp_ != nullptr, |
| 578 | true, |
| 579 | common::errors::InvalidArgument( |
| 580 | "This fp should not be null, please check!")); |
| 581 | __fsetlocking(&*(this->fp_), FSETLOCKING_BYCALLER); |
| 582 | paddle::framework::ChannelWriter<T> writer(input_channel_); |
| 583 | T instance; |
| 584 | platform::Timer timeline; |
| 585 | timeline.Start(); |
| 586 | while (ParseOneInstanceFromPipe(&instance)) { |
| 587 | writer << std::move(instance); |
| 588 | instance = T(); |
| 589 | } |
| 590 | STAT_ADD(STAT_total_feasign_num_in_mem, fea_num_); |
| 591 | { |
| 592 | std::lock_guard<std::mutex> flock(*mutex_for_fea_num_); |
| 593 | *total_fea_num_ += fea_num_; |
| 594 | fea_num_ = 0; |
| 595 | } |
| 596 | writer.Flush(); |
| 597 | timeline.Pause(); |
| 598 | VLOG(3) << "LoadIntoMemory() read all lines, file=" << filename |
| 599 | << ", cost time=" << timeline.ElapsedSec() |
| 600 | << " seconds, thread_id=" << thread_id_; |
| 601 | } |
| 602 | VLOG(3) << "LoadIntoMemory() end, thread_id=" << thread_id_; |
| 603 | #endif |
| 604 | } |
| 605 | |
| 606 | template <typename T> |
| 607 | void InMemoryDataFeed<T>::LoadIntoMemoryFromSo() { |
| 608 | #if (defined _LINUX) && (defined PADDLE_WITH_HETERPS) |
| 609 | VLOG(3) << "LoadIntoMemoryFromSo() begin, thread_id=" << thread_id_; |
| 610 | int buf_len = 1024 * 1024 * 10; |
| 611 | char* buf = reinterpret_cast<char*>(malloc(buf_len + 10)); |
| 612 | auto ps_gpu_ptr = PSGPUWrapper::GetInstance(); |
nothing calls this directly
no test coverage detected