allreduction-inter-io.F90

Go to the documentation of this file.

module allreduction_inter_io_mod
  use datadefn_mod, only : default_precision, double_precision, string_length
  use configuration_parser_mod, only : io_configuration_type
  use collections_mod, only : hashmap_type, c_get_generic, c_put_generic, c_remove, c_is_empty
  use conversions_mod, only : conv_to_string
  use forthread_mod, only : forthread_rwlock_rdlock, forthread_rwlock_wrlock, forthread_rwlock_unlock, &
       forthread_rwlock_init, forthread_rwlock_destroy
  use threadpool_mod, only : check_thread_status
  use inter_io_specifics_mod, only : handle_completion
  use reduction_inter_io_mod, only : init_reduction_inter_io, check_reduction_inter_io_for_completion, &
       finalise_reduction_inter_io, perform_inter_io_reduction, get_reduction_operator
  use broadcast_inter_io_mod, only : init_broadcast_inter_io, perform_inter_io_broadcast, finalise_broadcast_inter_io
  implicit none

#ifndef TEST_MODE
  private
#endif

  type allreduce_type
     integer :: root
     procedure(handle_completion), pointer, nopass :: completion_procedure
  end type allreduce_type  

  logical, volatile :: initialised=.false.

  integer, volatile :: allreduce_rwlock
  type(hashmap_type), volatile :: allreduce_types

  public init_allreduction_inter_io, finalise_allreduction_inter_io, perform_inter_io_allreduction, &
       check_allreduction_inter_io_for_completion
contains

  subroutine init_allreduction_inter_io(io_configuration)
    type(io_configuration_type), intent(inout) :: io_configuration

    if (.not. initialised) then
      initialised=.true.
      call check_thread_status(forthread_rwlock_init(allreduce_rwlock, -1))
    end if
    call init_reduction_inter_io(io_configuration)
    call init_broadcast_inter_io(io_configuration)
  end subroutine init_allreduction_inter_io

  subroutine finalise_allreduction_inter_io(io_configuration)
    type(io_configuration_type), intent(inout) :: io_configuration

    if (initialised) then
      initialised=.false.
      call check_thread_status(forthread_rwlock_destroy(allreduce_rwlock))
    end if
    call finalise_reduction_inter_io(io_configuration)
    call finalise_broadcast_inter_io()
  end subroutine finalise_allreduction_inter_io

  logical function check_allreduction_inter_io_for_completion(io_configuration)
    type(io_configuration_type), intent(inout) :: io_configuration

    call check_thread_status(forthread_rwlock_rdlock(allreduce_rwlock))
    check_allreduction_inter_io_for_completion=c_is_empty(allreduce_types)
    call check_thread_status(forthread_rwlock_unlock(allreduce_rwlock))
  end function check_allreduction_inter_io_for_completion
  
  subroutine perform_inter_io_allreduction(io_configuration, field_values, field_size, field_name, reduction_op, root, &
       timestep, completion_procedure)
    type(io_configuration_type), intent(inout) :: io_configuration
    real(kind=DOUBLE_PRECISION), dimension(:) :: field_values
    integer, intent(in) :: field_size, reduction_op, root, timestep    
    character(len=*), intent(in) :: field_name
    procedure(handle_completion) :: completion_procedure

    if (io_configuration%my_io_rank .eq. root) then
      call add_allreduce_information_if_needed(field_name, timestep, root, completion_procedure)
    end if
    call perform_inter_io_reduction(io_configuration, field_values, field_size, field_name, reduction_op, &
       root, timestep, internal_reduction_completion_procedure)
    if (io_configuration%my_io_rank .ne. root) then
      ! None root processes issue a broadcast
      call perform_inter_io_broadcast(io_configuration, field_values, size(field_values), field_name, root, &
           timestep, completion_procedure)
    end if    
  end subroutine perform_inter_io_allreduction  

  subroutine internal_reduction_completion_procedure(io_configuration, values, field_name, timestep)
    type(io_configuration_type), intent(inout) :: io_configuration
    real(DEFAULT_PRECISION), dimension(:) :: values
    character(len=*) :: field_name
    integer :: timestep

    type(allreduce_type), pointer :: allreduce_information

    allreduce_information=>find_allreduce_information(field_name, timestep, .true.)    
    call perform_inter_io_broadcast(io_configuration, values, size(values), field_name, &
         allreduce_information%root, timestep, allreduce_information%completion_procedure)
    call remove_allreduce_information(field_name, timestep, .true.)
  end subroutine internal_reduction_completion_procedure

  subroutine add_allreduce_information_if_needed(field_name, timestep, root, completion_procedure)
    character(len=*), intent(in) :: field_name
    integer, intent(in) :: timestep, root
    procedure(handle_completion) :: completion_procedure

    type(allreduce_type), pointer :: allreduce_information
    class(*), pointer :: generic

    allreduce_information=>find_allreduce_information(field_name, timestep, .true.)
    if (.not. associated(allreduce_information)) then
      call check_thread_status(forthread_rwlock_wrlock(allreduce_rwlock))
      allreduce_information=>find_allreduce_information(field_name, timestep, .false.)
      if (.not. associated(allreduce_information)) then
        allocate(allreduce_information)
        allreduce_information%completion_procedure=>completion_procedure
        allreduce_information%root=root
        generic=>allreduce_information
        call c_put_generic(allreduce_types, trim(field_name)//"#"//conv_to_string(timestep), generic, .true.)
      end if
      call check_thread_status(forthread_rwlock_unlock(allreduce_rwlock))
    end if
  end subroutine add_allreduce_information_if_needed
  
  function find_allreduce_information(field_name, timestep, dolock)
    character(len=*), intent(in) :: field_name
    integer, intent(in) :: timestep
    logical, intent(in) :: dolock
    type(allreduce_type), pointer :: find_allreduce_information

    class(*), pointer :: generic

    if (dolock) call check_thread_status(forthread_rwlock_rdlock(allreduce_rwlock))
    generic=>c_get_generic(allreduce_types, trim(field_name)//"#"//conv_to_string(timestep))
    if (dolock) call check_thread_status(forthread_rwlock_unlock(allreduce_rwlock))
    
    if (associated(generic)) then
      select type(generic)
        type is (allreduce_type)
          find_allreduce_information=>generic
      end select      
    else
      find_allreduce_information=>null()
    end if
  end function find_allreduce_information

  subroutine remove_allreduce_information(field_name, timestep, dolock)
    character(len=*), intent(in) :: field_name
    integer, intent(in) :: timestep
    logical, intent(in) :: dolock

    if (dolock) call check_thread_status(forthread_rwlock_wrlock(allreduce_rwlock))
    call c_remove(allreduce_types, trim(field_name)//"#"//conv_to_string(timestep))
    if (dolock) call check_thread_status(forthread_rwlock_unlock(allreduce_rwlock))
  end subroutine remove_allreduce_information  
end module allreduction_inter_io_mod