Method LoadIntoMemoryFromSo

paddle/fluid/framework/data_feed.cc:607–655

template <typename T>
void InMemoryDataFeed<T>::LoadIntoMemoryFromSo() {
#if (defined _LINUX) && (defined PADDLE_WITH_HETERPS)
  VLOG(3) << "LoadIntoMemoryFromSo() begin, thread_id=" << thread_id_;
  int buf_len = 1024 * 1024 * 10;
  char* buf = reinterpret_cast<char*>(malloc(buf_len + 10));
  auto ps_gpu_ptr = PSGPUWrapper::GetInstance();

#ifdef PADDLE_WITH_PSLIB
  paddle::framework::CustomParser* parser =
      global_dlmanager_pool().Load(so_parser_name_, slot_conf_);
#endif

  std::string filename;
  while (this->PickOneFile(&filename)) {
    VLOG(3) << "PickOneFile, filename=" << filename
            << ", thread_id=" << thread_id_;
    platform::Timer timeline;
    timeline.Start();
    if (ps_gpu_ptr->UseAfsApi()) {
#ifdef PADDLE_WITH_PSLIB
      auto afs_reader = ps_gpu_ptr->OpenReader(filename);
      int read_len = 0;
      char* cursor = buf;
      int remain = 0;
      while ((read_len = afs_reader->read(cursor, buf_len - remain)) > 0) {
        std::vector<T> instances;
        read_len += remain;
        remain = ParseInstanceFromSo(read_len, buf, &instances, parser);
        input_channel_->Write(std::move(instances));
        instances = std::vector<T>();
        if (remain) {
          memmove(buf, buf + read_len - remain, remain);
        }
        cursor = buf + remain;
      }
#endif
    } else {
      VLOG(0) << "Should Call InitAfsApi First";
    }

    timeline.Pause();
    VLOG(3) << "LoadIntoMemoryFromSo() read all lines, file=" << filename
            << ", cost time=" << timeline.ElapsedSec()
            << " seconds, thread_id=" << thread_id_;
  }
  free(buf);
  VLOG(3) << "LoadIntoMemoryFromSo() end, thread_id=" << thread_id_;
#endif
}

// explicit instantiation
template class InMemoryDataFeed<Record>;