| 317 | |
| 318 | template <typename TensorDataType> |
| 319 | void buffered_data_coordinator<TensorDataType>::fetch_data_asynchronous( |
| 320 | execution_mode mode) |
| 321 | { |
| 322 | data_buffer<IODataType>& current_buffer = get_active_buffer(mode); |
| 323 | data_buffer<IODataType>& next_buffer = get_next_buffer(mode); |
| 324 | auto next_buffer_idx = this->get_next_buffer_idx(mode); |
| 325 | auto next_buffer_id = next_buffer_idx % m_data_buffers.size(); |
| 326 | |
| 327 | // Wait for the background thread to complete fetching the data |
| 328 | if (current_buffer.is_background_fetching_in_progress()) { |
| 329 | current_buffer.get_data_fetch_future().get(); |
| 330 | current_buffer.set_background_fetching_in_progress(false); |
| 331 | } |
| 332 | |
| 333 | dataset& ds = get_dataset(mode); |
| 334 | //************************************************************************ |
| 335 | // Get the next mini-batchs size from the data reader |
| 336 | El::Int next_mini_batch_size = ds.get_next_mini_batch_size(); |
| 337 | |
| 338 | // If there is no valid data and there is not already a background |
| 339 | // thread to fetch the data, queue up the background thread |
| 340 | if (next_mini_batch_size > 0 && next_buffer.num_samples_ready() == 0 && |
| 341 | !next_buffer.is_background_fetching_in_progress()) { |
| 342 | // Store the size of the current mini-batch so that others can obtain it |
| 343 | // without worrying about where the data reader is currently at. |
| 344 | m_current_mini_batch_size[next_buffer_id][mode] = next_mini_batch_size; |
| 345 | El::Int relative_base_position = ds.get_next_position(); |
| 346 | |
| 347 | // Start data store exchange if necessary (this should be moved |
| 348 | // earlier as a future optimization) |
| 349 | get_data_reader(mode)->start_data_store_mini_batch_exchange( |
| 350 | // Use the relative position of the mini-batch (adjusted for rank) |
| 351 | relative_base_position - ds.get_base_offset(), |
| 352 | next_mini_batch_size, |
| 353 | ds.at_new_epoch()); |
| 354 | // Finish data store exchange before accessing samples |
| 355 | get_data_reader(mode)->finish_data_store_mini_batch_exchange(); |
| 356 | |
| 357 | // Set the size for the I/O buffers |
| 358 | fp_setup_data(next_buffer, next_mini_batch_size); |
| 359 | |
| 360 | std::future<void> background_fetch_done = get_io_thread_pool().submit_job( |
| 361 | std::bind(&buffered_data_coordinator::fetch_data_in_background, |
| 362 | this, |
| 363 | next_buffer_idx, |
| 364 | std::ref(next_buffer), |
| 365 | next_mini_batch_size, |
| 366 | relative_base_position, |
| 367 | mode)); |
| 368 | next_buffer.set_data_fetch_future(std::move(background_fetch_done)); |
| 369 | next_buffer.set_background_fetching_in_progress(true); |
| 370 | } |
| 371 | } |
| 372 | |
| 373 | template <typename TensorDataType> |
| 374 | bool buffered_data_coordinator<TensorDataType>::ready_for_next_fetch( |
no test coverage detected