| 2225 | } |
| 2226 | |
| 2227 | void SlotRecordInMemoryDataFeed::LoadIntoMemoryByFile() { |
| 2228 | #if (defined _LINUX) && (defined PADDLE_WITH_HETERPS) |
| 2229 | paddle::framework::CustomParser* parser = |
| 2230 | global_dlmanager_pool().Load(so_parser_name_, all_slots_info_); |
| 2231 | PADDLE_ENFORCE_EQ(parser != nullptr, |
| 2232 | true, |
| 2233 | common::errors::InvalidArgument( |
| 2234 | "Parser should not be null, please check!")); |
| 2235 | // get slotrecord object |
| 2236 | auto pull_record_func = [this](std::vector<SlotRecord>& record_vec, |
| 2237 | int max_fetch_num, |
| 2238 | int offset) { |
| 2239 | if (offset > 0) { |
| 2240 | input_channel_->WriteMove(offset, &record_vec[0]); |
| 2241 | if (max_fetch_num > 0) { |
| 2242 | SlotRecordPool().get(&record_vec[0], offset); |
| 2243 | } else { // free all |
| 2244 | max_fetch_num = static_cast<int>(record_vec.size()); |
| 2245 | if (max_fetch_num > offset) { |
| 2246 | SlotRecordPool().put(&record_vec[offset], (max_fetch_num - offset)); |
| 2247 | } |
| 2248 | } |
| 2249 | } else if (max_fetch_num > 0) { |
| 2250 | SlotRecordPool().get(&record_vec, max_fetch_num); |
| 2251 | } else { |
| 2252 | SlotRecordPool().put(&record_vec); |
| 2253 | } |
| 2254 | }; |
| 2255 | |
| 2256 | std::string filename; |
| 2257 | while (this->PickOneFile(&filename)) { |
| 2258 | VLOG(3) << "PickOneFile, filename=" << filename |
| 2259 | << ", thread_id=" << thread_id_; |
| 2260 | platform::Timer timeline; |
| 2261 | timeline.Start(); |
| 2262 | |
| 2263 | int lines = 0; |
| 2264 | bool is_ok = true; |
| 2265 | auto ps_gpu_ptr = PSGPUWrapper::GetInstance(); |
| 2266 | do { |
| 2267 | if (ps_gpu_ptr->UseAfsApi()) { |
| 2268 | #ifdef PADDLE_WITH_PSLIB |
| 2269 | auto afs_reader = ps_gpu_ptr->OpenReader(filename); |
| 2270 | is_ok = parser->ParseFileInstance( |
| 2271 | [this, afs_reader](char* buf, int len) { |
| 2272 | return afs_reader->read(buf, len); |
| 2273 | }, |
| 2274 | pull_record_func, |
| 2275 | lines); |
| 2276 | #elif defined(PADDLE_WITH_HETERPS) && defined(PADDLE_WITH_PSCORE) |
| 2277 | auto afs_reader = ps_gpu_ptr->OpenReader(filename); |
| 2278 | is_ok = parser->ParseFileInstance( |
| 2279 | [this, ps_gpu_ptr, afs_reader](char* buf, int len) { |
| 2280 | return ps_gpu_ptr->AfsRead(afs_reader, buf, len); |
| 2281 | }, |
| 2282 | pull_record_func, |
| 2283 | lines); |
| 2284 | ps_gpu_ptr->CloseReader(afs_reader); |
nothing calls this directly
no test coverage detected