| 2413 | } |
| 2414 | |
| 2415 | void SlotRecordInMemoryDataFeed::LoadIntoMemoryByCommand() { |
| 2416 | #ifdef _LINUX |
| 2417 | std::string filename; |
| 2418 | BufferedLineFileReader line_reader; |
| 2419 | line_reader.set_sample_rate(sample_rate_); |
| 2420 | |
| 2421 | while (this->PickOneFile(&filename)) { |
| 2422 | VLOG(3) << "PickOneFile, filename=" << filename |
| 2423 | << ", thread_id=" << thread_id_; |
| 2424 | int lines = 0; |
| 2425 | std::vector<SlotRecord> record_vec; |
| 2426 | platform::Timer timeline; |
| 2427 | timeline.Start(); |
| 2428 | SlotRecordPool().get(&record_vec, OBJPOOL_BLOCK_SIZE); |
| 2429 | int offset = 0; |
| 2430 | |
| 2431 | do { |
| 2432 | int err_no = 0; |
| 2433 | this->fp_ = fs_open_read(filename, &err_no, this->pipe_command_, true); |
| 2434 | PADDLE_ENFORCE_EQ(this->fp_ != nullptr, |
| 2435 | true, |
| 2436 | common::errors::InvalidArgument( |
| 2437 | "This fp should not be null, please check!")); |
| 2438 | __fsetlocking(&*(this->fp_), FSETLOCKING_BYCALLER); |
| 2439 | |
| 2440 | lines = line_reader.read_file( |
| 2441 | this->fp_.get(), |
| 2442 | [this, &record_vec, &offset, &filename](const std::string& line) { |
| 2443 | if (ParseOneInstance(line, &record_vec[offset])) { |
| 2444 | ++offset; |
| 2445 | } else { |
| 2446 | LOG(WARNING) << "read file:[" << filename |
| 2447 | << "] item error, line:[" << line << "]"; |
| 2448 | return false; |
| 2449 | } |
| 2450 | if (offset >= OBJPOOL_BLOCK_SIZE) { |
| 2451 | input_channel_->Write(std::move(record_vec)); |
| 2452 | record_vec.clear(); |
| 2453 | SlotRecordPool().get(&record_vec, OBJPOOL_BLOCK_SIZE); |
| 2454 | offset = 0; |
| 2455 | } |
| 2456 | return true; |
| 2457 | }, |
| 2458 | lines); |
| 2459 | } while (line_reader.is_error()); |
| 2460 | if (offset > 0) { |
| 2461 | input_channel_->WriteMove(offset, &record_vec[0]); |
| 2462 | if (offset < OBJPOOL_BLOCK_SIZE) { |
| 2463 | SlotRecordPool().put(&record_vec[offset], |
| 2464 | (OBJPOOL_BLOCK_SIZE - offset)); |
| 2465 | } |
| 2466 | } else { |
| 2467 | SlotRecordPool().put(&record_vec); |
| 2468 | } |
| 2469 | record_vec.clear(); |
| 2470 | record_vec.shrink_to_fit(); |
| 2471 | timeline.Pause(); |
| 2472 | VLOG(3) << "LoadIntoMemory() read all lines, file=" << filename |
nothing calls this directly
no test coverage detected