handle_local_worker_results_to_batch Subroutine

public subroutine handle_local_worker_results_to_batch(node_comm, world_comm, worker_map, batch_count, batch_ids, batch_results, results_received)

Drain pending local worker results and append them to the outbound batch.

Arguments

Type IntentOptional Attributes Name
type(comm_t), intent(in) :: node_comm
type(comm_t), intent(in) :: world_comm
integer(kind=int64), intent(inout) :: worker_map(:)
integer(kind=int32), intent(inout) :: batch_count
integer(kind=int64), intent(inout) :: batch_ids(:)
type(calculation_result_t), intent(inout) :: batch_results(:)
integer(kind=int64), intent(inout), optional :: results_received

Calls

proc~~handle_local_worker_results_to_batch~~CallsGraph proc~handle_local_worker_results_to_batch handle_local_worker_results_to_batch abort_comm abort_comm proc~handle_local_worker_results_to_batch->abort_comm error error proc~handle_local_worker_results_to_batch->error iprobe iprobe proc~handle_local_worker_results_to_batch->iprobe proc~append_result_to_batch append_result_to_batch proc~handle_local_worker_results_to_batch->proc~append_result_to_batch proc~error_get_message error_t%error_get_message proc~handle_local_worker_results_to_batch->proc~error_get_message proc~flush_group_results flush_group_results proc~handle_local_worker_results_to_batch->proc~flush_group_results proc~result_destroy calculation_result_t%result_destroy proc~handle_local_worker_results_to_batch->proc~result_destroy proc~result_irecv result_irecv proc~handle_local_worker_results_to_batch->proc~result_irecv to_char to_char proc~handle_local_worker_results_to_batch->to_char proc~flush_group_results->proc~result_destroy isend isend proc~flush_group_results->isend proc~result_isend result_isend proc~flush_group_results->proc~result_isend proc~result_reset calculation_result_t%result_reset proc~result_destroy->proc~result_reset irecv irecv proc~result_irecv->irecv recv recv proc~result_irecv->recv proc~result_isend->isend send send proc~result_isend->send proc~energy_reset energy_t%energy_reset proc~result_reset->proc~energy_reset proc~error_clear error_t%error_clear proc~result_reset->proc~error_clear proc~mp2_reset mp2_energy_t%mp2_reset proc~energy_reset->proc~mp2_reset

Called by

proc~~handle_local_worker_results_to_batch~~CalledByGraph proc~handle_local_worker_results_to_batch handle_local_worker_results_to_batch proc~gmbe_group_global_coordinator gmbe_group_global_coordinator proc~gmbe_group_global_coordinator->proc~handle_local_worker_results_to_batch proc~group_global_coordinator_impl group_global_coordinator_impl proc~group_global_coordinator_impl->proc~handle_local_worker_results_to_batch proc~gmbe_run_distributed gmbe_context_t%gmbe_run_distributed proc~gmbe_run_distributed->proc~gmbe_group_global_coordinator interface~node_coordinator node_coordinator proc~gmbe_run_distributed->interface~node_coordinator proc~node_coordinator_impl node_coordinator_impl proc~node_coordinator_impl->proc~group_global_coordinator_impl proc~node_coordinator node_coordinator proc~node_coordinator->proc~node_coordinator_impl interface~node_coordinator->proc~node_coordinator proc~mbe_run_distributed mbe_context_t%mbe_run_distributed proc~mbe_run_distributed->interface~node_coordinator

Variables

Type Visibility Attributes Name Initial
logical, private :: has_pending
integer(kind=int64), private :: item_idx
type(MPI_Status), private :: local_status
type(request_t), private :: req
type(calculation_result_t), private :: worker_result
integer, private :: worker_source

Source Code

   subroutine handle_local_worker_results_to_batch(node_comm, world_comm, worker_map, batch_count, batch_ids, batch_results, &
                                                   results_received)
      !! Drain pending local worker results and append them to the outbound batch.
      type(comm_t), intent(in) :: node_comm
      type(comm_t), intent(in) :: world_comm
      integer(int64), intent(inout) :: worker_map(:)
      integer(int32), intent(inout) :: batch_count
      integer(int64), intent(inout) :: batch_ids(:)
      type(calculation_result_t), intent(inout) :: batch_results(:)
      integer(int64), intent(inout), optional :: results_received

      type(MPI_Status) :: local_status
      logical :: has_pending
      integer :: worker_source
      type(request_t) :: req
      type(calculation_result_t) :: worker_result
      integer(int64) :: item_idx

      if (node_comm%size() <= 1) return

      do
         call iprobe(node_comm, MPI_ANY_SOURCE, TAG_WORKER_SCALAR_RESULT, has_pending, local_status)
         if (.not. has_pending) exit

         worker_source = local_status%MPI_SOURCE

         if (worker_map(worker_source) == 0) then
            call logger%error("Received result from worker "//to_char(worker_source)// &
                              " but no item was assigned!")
            call abort_comm(world_comm, 1)
         end if

         call result_irecv(worker_result, node_comm, worker_source, TAG_WORKER_SCALAR_RESULT, req)
         call wait(req)

         if (worker_result%has_error) then
            call logger%error("Item "//to_char(worker_map(worker_source))// &
                              " calculation failed: "// &
                              worker_result%error%get_message())
            call abort_comm(world_comm, 1)
         end if

         item_idx = worker_map(worker_source)
         worker_map(worker_source) = 0

         if (batch_count >= size(batch_ids)) then
            call flush_group_results(world_comm, batch_count, batch_ids, batch_results)
         end if

         call append_result_to_batch(item_idx, worker_result, batch_count, batch_ids, batch_results)
         if (present(results_received)) results_received = results_received + 1_int64
         if (batch_count >= size(batch_ids)) then
            call flush_group_results(world_comm, batch_count, batch_ids, batch_results)
         end if
         call worker_result%destroy()
      end do
   end subroutine handle_local_worker_results_to_batch