Method fetch

src/data_readers/data_reader.cpp:86–163

}

int lbann::generic_data_reader::fetch(std::vector<conduit::Node>& samples,
                                      El::Matrix<El::Int>& indices_fetched,
                                      El::Int current_position_in_data_set,
                                      El::Int sample_stride,
                                      size_t mb_size,
                                      const execution_mode mode)
{
  // Check to make sure that a valid map was passed
  if (samples.empty()) {
    LBANN_ERROR("fetch function called with no valid buffers");
  }

  // BVE FIXME
  if (!(current_position_in_data_set < get_num_data()) /*position_valid()*/) {
    if (current_position_in_data_set >=
        get_num_data() /*position_is_overrun()*/) {
      return 0;
    }
    else {
      LBANN_ERROR(
        std::string{} + "generic data reader load error: !position_valid" +
        " -- current pos = " + std::to_string(current_position_in_data_set) +
        " and there are " + std::to_string(m_shuffled_indices.size()) +
        " indices");
    }
  }

  /// Allow each thread to perform any preprocessing necessary on the
  /// data source prior to fetching data
  for (int t = 0; t < static_cast<int>(m_io_thread_pool->get_num_threads());
       t++) {
    preprocess_data_source(t);
  }

  // Fetch data is executed by the thread pool so it has to dispatch
  // work to other threads in the thread pool and do some work locally
  for (int t = 0; t < static_cast<int>(m_io_thread_pool->get_num_threads());
       t++) {
    // Queue up work into other threads and then finish off the
    // mini-batch in the active thread
    if (t == m_io_thread_pool->get_local_thread_id()) {
      continue;
    }
    else {
      m_io_thread_pool->submit_job_to_work_group(
        std::bind(&generic_data_reader::fetch_data_block_conduit,
                  this,
                  std::ref(samples),
                  current_position_in_data_set,
                  t,
                  m_io_thread_pool->get_num_threads(),
                  sample_stride,
                  mb_size,
                  std::ref(indices_fetched),
                  mode));
    }
  }
  fetch_data_block_conduit(samples,