reduction-inter-io.F90

Go to the documentation of this file.

module reduction_inter_io_mod
  use datadefn_mod, only : default_precision, double_precision, string_length
  use configuration_parser_mod, only : io_configuration_type, io_configuration_inter_communication_description
  use collections_mod, only : map_type, hashmap_type, list_type, iterator_type, mapentry_type, c_get_generic, c_get_string, &
       c_add_string, c_put_generic, c_remove, c_is_empty, c_contains, c_free, c_generic_at, c_get_iterator, c_has_next, &
       c_next_mapentry, c_next_string
  use conversions_mod, only : conv_to_string
  use forthread_mod, only : forthread_mutex_init, forthread_mutex_lock, forthread_mutex_trylock, &
       forthread_mutex_unlock, forthread_mutex_destroy, forthread_rwlock_rdlock, forthread_rwlock_wrlock, &
       forthread_rwlock_unlock, forthread_rwlock_init, forthread_rwlock_destroy, forthread_rwlock_trywrlock
  use threadpool_mod, only : check_thread_status, threadpool_start_thread
  use logging_mod, only : log_error, log_log
  use inter_io_specifics_mod, only : handle_completion, register_inter_io_communication, find_inter_io_from_name, &
       package_inter_io_communication_message, unpackage_inter_io_communication_message
  use mpi, only : mpi_double_precision, mpi_int, mpi_any_source, mpi_request_null, mpi_status_ignore, mpi_character, mpi_byte
  use mpi_communication_mod, only : lock_mpi, unlock_mpi, wait_for_mpi_request
  implicit none

#ifndef TEST_MODE
  private
#endif

  
  integer, parameter :: mean=1, min=2, max=3, sum=4
  integer, parameter :: my_inter_io_tag=1, perform_clean_every=200
  character(len=*), parameter :: my_inter_io_name="reductioninterio"

  type reduction_progress_type
     real(DEFAULT_PRECISION), dimension(:), allocatable :: values
     character, dimension(:), allocatable :: send_buffer
     character(len=STRING_LENGTH) :: field_name
     integer :: contributed_moncs, contributed_io_servers, timestep, reduction_operator, async_handle, mutex, root
     procedure(handle_completion), pointer, nopass :: completion_procedure
  end type reduction_progress_type  

  integer, volatile :: reduction_progress_rwlock, inter_io_description_mutex, clean_progress_mutex, &
       reduction_count_mutex, previous_clean_reduction_count, reduction_count
  type(hashmap_type), volatile :: reduction_progresses
  logical, volatile :: initialised=.false.

  public init_reduction_inter_io, check_reduction_inter_io_for_completion, finalise_reduction_inter_io, &
       perform_inter_io_reduction, get_reduction_operator
contains

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

    if (.not. initialised) then
      initialised=.true.
      previous_clean_reduction_count=0
      reduction_count=0
      call check_thread_status(forthread_rwlock_init(reduction_progress_rwlock, -1))
      call check_thread_status(forthread_mutex_init(inter_io_description_mutex, -1))
      call check_thread_status(forthread_mutex_init(clean_progress_mutex, -1))
      call check_thread_status(forthread_mutex_init(reduction_count_mutex, -1))
      call register_inter_io_communication(io_configuration, my_inter_io_tag, handle_recv_data_from_io_server, my_inter_io_name)
    end if
  end subroutine init_reduction_inter_io  

  subroutine handle_recv_data_from_io_server(io_configuration, data_buffer, inter_io_index)
    type(io_configuration_type), intent(inout) :: io_configuration
    character, dimension(:), intent(inout) :: data_buffer
    integer, intent(in) :: inter_io_index    

    call handle_process_recv_from_other_io_server(io_configuration, io_configuration%inter_io_communications(inter_io_index), &
         io_configuration%my_io_rank, data_buffer, io_configuration%number_of_io_servers)
  end subroutine handle_recv_data_from_io_server

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

    check_reduction_inter_io_for_completion=check_and_clean_progress(io_configuration%my_io_rank)
  end function check_reduction_inter_io_for_completion  

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

    type(reduction_progress_type) :: progress
    type(iterator_type) :: iterator

    if (initialised) then
      call check_thread_status(forthread_rwlock_rdlock(reduction_progress_rwlock))
      if (.not. c_is_empty(reduction_progresses)) then
        iterator=c_get_iterator(reduction_progresses)
        do while (c_has_next(iterator))
          progress=retrieve_reduction_progress(c_next_mapentry(iterator))
          if (progress%async_handle /= mpi_request_null) then
            call wait_for_mpi_request(progress%async_handle)
          end if
        end do
      end if
      call check_thread_status(forthread_rwlock_unlock(reduction_progress_rwlock))
      call check_thread_status(forthread_rwlock_destroy(reduction_progress_rwlock))
      call check_thread_status(forthread_mutex_destroy(inter_io_description_mutex))
      call check_thread_status(forthread_mutex_destroy(clean_progress_mutex))
      call check_thread_status(forthread_mutex_destroy(reduction_count_mutex))
      initialised=.false.
    end if
  end subroutine finalise_reduction_inter_io

  subroutine perform_inter_io_reduction(io_configuration, field_values, field_size, reduction_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) :: reduction_field_name
    procedure(handle_completion) :: completion_procedure

    type(reduction_progress_type), pointer :: reduction_progress
    logical :: collective_values_new
        
    call clean_progress(io_configuration%my_io_rank)
    reduction_progress=>find_or_add_reduction_progress(timestep, reduction_op, root, reduction_field_name, completion_procedure)

    call check_thread_status(forthread_mutex_lock(reduction_progress%mutex))
    reduction_progress%contributed_moncs=reduction_progress%contributed_moncs+1

    collective_values_new=.not. allocated(reduction_progress%values)
    if (collective_values_new) allocate(reduction_progress%values(field_size))

    call integrate_io_server_collective_values(reduction_op, reduction_progress, field_values, field_size, collective_values_new)
    if (reduction_progress%contributed_moncs == io_configuration%number_of_moncs) then
      reduction_progress%contributed_io_servers=reduction_progress%contributed_io_servers+1      
      call handle_local_moncs_completed_collective(io_configuration, reduction_progress)
    else
      call check_thread_status(forthread_mutex_unlock(reduction_progress%mutex))
    end if    
  end subroutine perform_inter_io_reduction

  subroutine clean_progress(myrank)
    integer, intent(in) :: myrank

    logical :: cc_dummy

    call check_thread_status(forthread_mutex_lock(reduction_count_mutex))  
    reduction_count=reduction_count+1
    if (previous_clean_reduction_count + perform_clean_every .lt. reduction_count) then      
      previous_clean_reduction_count=reduction_count
      call check_thread_status(forthread_mutex_unlock(reduction_count_mutex))        
      cc_dummy=check_and_clean_progress(myrank)
    else
      call check_thread_status(forthread_mutex_unlock(reduction_count_mutex))      
    end if
  end subroutine clean_progress

  logical function check_and_clean_progress(myrank)
    integer, intent(in) :: myrank

    integer :: i, entries, completed, ierr, num_to_remove, have_lock
    type(list_type) :: entries_to_remove
    type(iterator_type) :: iterator
    type(mapentry_type) :: mapentry
    type(reduction_progress_type), pointer :: specific_reduction_progress
    character(len=STRING_LENGTH) :: entry_key
    class(*), pointer :: generic
    logical :: destroy_lock

    check_and_clean_progress=.true.
    have_lock=forthread_mutex_trylock(clean_progress_mutex)
    if (have_lock == 0) then
      call check_thread_status(forthread_rwlock_rdlock(reduction_progress_rwlock))
      iterator=c_get_iterator(reduction_progresses)
      do while (c_has_next(iterator))
        mapentry=c_next_mapentry(iterator)
        destroy_lock=.false.
        specific_reduction_progress=>retrieve_reduction_progress(mapentry)
        if (myrank /= specific_reduction_progress%root) then
          call check_thread_status(forthread_mutex_lock(specific_reduction_progress%mutex))
          if (specific_reduction_progress%async_handle /= mpi_request_null) then
            call wait_for_mpi_request(specific_reduction_progress%async_handle)
            if (completed == 1) then
              if (allocated(specific_reduction_progress%send_buffer)) deallocate(specific_reduction_progress%send_buffer)
              destroy_lock=.true.
              call c_add_string(entries_to_remove, mapentry%key)
            else
              check_and_clean_progress=.false.
            end if
          end if
          call check_thread_status(forthread_mutex_unlock(specific_reduction_progress%mutex))
          if (destroy_lock) call check_thread_status(forthread_mutex_destroy(specific_reduction_progress%mutex))
        end if
      end do
      call check_thread_status(forthread_rwlock_unlock(reduction_progress_rwlock))

      if (.not. c_is_empty(entries_to_remove)) then
        call check_thread_status(forthread_rwlock_wrlock(reduction_progress_rwlock))
        iterator=c_get_iterator(entries_to_remove)
        do while (c_has_next(iterator))
          entry_key=c_next_string(iterator)
          generic=>c_get_generic(reduction_progresses, entry_key)
          call c_remove(reduction_progresses, entry_key)
          deallocate(generic)
        end do
        call check_thread_status(forthread_rwlock_unlock(reduction_progress_rwlock))
      end if
      call c_free(entries_to_remove)
      call check_thread_status(forthread_mutex_unlock(clean_progress_mutex))
    end if
  end function check_and_clean_progress

  subroutine integrate_io_server_collective_values(reduction_op, reduction_progress, single_server_values, &
       number_elements, collective_values_empty)
    integer, intent(in) :: reduction_op, number_elements
    logical, intent(in) :: collective_values_empty
    type(reduction_progress_type), intent(inout) :: reduction_progress
    real(kind=DOUBLE_PRECISION), dimension(:), intent(in) :: single_server_values

    integer :: k

    if (collective_values_empty) then
      reduction_progress%values=single_server_values
    else
      if (reduction_op == mean .or. reduction_op == sum) then
        reduction_progress%values=reduction_progress%values+single_server_values       
      else if (reduction_op == min .or. reduction_op == max) then
        do k=1, number_elements
          if (reduction_op == min) then
            if (single_server_values(k) .lt. reduction_progress%values(k)) &
                 reduction_progress%values(k)=single_server_values(k)
          else if (reduction_op == max) then
            if (single_server_values(k) .gt. reduction_progress%values(k)) &
                 reduction_progress%values(k)=single_server_values(k)
          end if
        end do
      end if
    end if
  end subroutine integrate_io_server_collective_values

  subroutine handle_local_moncs_completed_collective(io_configuration, reduction_progress)
    type(io_configuration_type), intent(inout) :: io_configuration
    type(reduction_progress_type), intent(inout) :: reduction_progress

    integer :: ierr, inter_io_comm_index
    
    if (io_configuration%my_io_rank == reduction_progress%root .and. &
         reduction_progress%contributed_io_servers == io_configuration%number_of_io_servers) then
      call handle_collective_completed(io_configuration, reduction_progress)
    else
      if (io_configuration%my_io_rank /= reduction_progress%root) then
        inter_io_comm_index=find_inter_io_from_name(io_configuration, my_inter_io_name)

        reduction_progress%send_buffer=package_inter_io_communication_message(reduction_progress%field_name, &
             reduction_progress%timestep, reduction_progress%values, reduction_progress%reduction_operator)
        call lock_mpi()
        call mpi_isend(reduction_progress%send_buffer, size(reduction_progress%send_buffer), &
             mpi_byte, reduction_progress%root, &
             io_configuration%inter_io_communications(inter_io_comm_index)%message_tag, &
             io_configuration%io_communicator, reduction_progress%async_handle, ierr)
        call unlock_mpi()
        ! Deallocate the current value as this is finished with and has been packed into the send buffer
        if (allocated(reduction_progress%values)) deallocate(reduction_progress%values)
      end if
      call check_thread_status(forthread_mutex_unlock(reduction_progress%mutex))
    end if
  end subroutine handle_local_moncs_completed_collective

  subroutine handle_process_recv_from_other_io_server(io_configuration, inter_io_comm, myrank, data_buffer, number_io_servers)
    type(io_configuration_type), intent(inout) :: io_configuration
    type(io_configuration_inter_communication_description), intent(inout) :: inter_io_comm
    character, dimension(:), intent(inout) :: data_buffer
    integer, intent(in) :: number_io_servers, myrank

    type(reduction_progress_type), pointer :: reduction_progress
    character(len=STRING_LENGTH) :: field_name
    integer :: timestep, reduction_op
    real(kind=DEFAULT_PRECISION), dimension(:), allocatable :: field_values
    logical :: collective_values_new

    call unpackage_inter_io_communication_message(data_buffer, field_name, timestep, field_values, reduction_op)
    reduction_progress=>find_or_add_reduction_progress(timestep, reduction_op, myrank, field_name)

    call check_thread_status(forthread_mutex_lock(reduction_progress%mutex))
    collective_values_new=.not. allocated(reduction_progress%values)
    if (collective_values_new) allocate(reduction_progress%values(size(field_values)))

    reduction_progress%contributed_io_servers=reduction_progress%contributed_io_servers+1    
    call integrate_io_server_collective_values(reduction_op, reduction_progress, &
         field_values, size(field_values), collective_values_new)
    if (reduction_progress%contributed_io_servers == number_io_servers) then
      call handle_collective_completed(io_configuration, reduction_progress)
      deallocate(field_values)
      return
    end if
    deallocate(field_values)
    call check_thread_status(forthread_mutex_unlock(reduction_progress%mutex))
  end subroutine handle_process_recv_from_other_io_server

  subroutine handle_collective_completed(io_configuration, reduction_progress)
    type(io_configuration_type), intent(inout) :: io_configuration
    type(reduction_progress_type), intent(inout) :: reduction_progress
    
    if (reduction_progress%reduction_operator == mean) then
      reduction_progress%values=reduction_progress%values/io_configuration%number_of_global_moncs
    end if    
    call reduction_progress%completion_procedure(io_configuration, reduction_progress%values, &
         reduction_progress%field_name, reduction_progress%timestep)
    call check_thread_status(forthread_mutex_unlock(reduction_progress%mutex))
    call check_thread_status(forthread_mutex_destroy(reduction_progress%mutex))
    if (allocated(reduction_progress%values)) deallocate(reduction_progress%values)
    call remove_reduction_progress(reduction_progress)
  end subroutine handle_collective_completed

  function find_or_add_reduction_progress(timestep, reduction_operator, root, field_name, completion_procedure)
    integer, intent(in) :: timestep, reduction_operator, root
    type(reduction_progress_type), pointer :: find_or_add_reduction_progress
    character(len=*), intent(in) :: field_name
    procedure(handle_completion), optional :: completion_procedure

    class(*), pointer :: generic
    type(reduction_progress_type), pointer :: new_progress

    find_or_add_reduction_progress=>find_reduction_progress(timestep, reduction_operator, field_name)
    if (.not. associated(find_or_add_reduction_progress)) then
      call check_thread_status(forthread_rwlock_wrlock(reduction_progress_rwlock))
      find_or_add_reduction_progress=>find_reduction_progress(timestep, reduction_operator, field_name, issue_read_lock=.false.)
      if (.not. associated(find_or_add_reduction_progress)) then
        allocate(new_progress)
        call check_thread_status(forthread_mutex_init(new_progress%mutex, -1))
        new_progress%timestep=timestep
        new_progress%reduction_operator=reduction_operator
        new_progress%contributed_moncs=0
        new_progress%contributed_io_servers=0
        new_progress%root=root
        new_progress%async_handle=mpi_request_null
        new_progress%field_name=field_name
        if (present(completion_procedure)) then
          new_progress%completion_procedure=>completion_procedure
        else
          new_progress%completion_procedure=>null()
        end if        
        generic=>new_progress
        call c_put_generic(reduction_progresses, generate_reduction_key(field_name, timestep, reduction_operator), &
             generic, .false.)
        find_or_add_reduction_progress=>new_progress
      end if
      call check_thread_status(forthread_rwlock_unlock(reduction_progress_rwlock))
    end if      
    if (.not. associated(find_or_add_reduction_progress%completion_procedure) .and. present(completion_procedure)) then
      find_or_add_reduction_progress%completion_procedure=>completion_procedure
    end if
  end function find_or_add_reduction_progress

  function find_reduction_progress(timestep, reduction_operator, field_name, issue_read_lock)
    integer, intent(in) :: timestep, reduction_operator    
    logical, intent(in), optional :: issue_read_lock
    type(reduction_progress_type), pointer :: find_reduction_progress
    character(len=*), intent(in) :: field_name

    class(*), pointer :: generic
    logical :: do_read_lock

    if (present(issue_read_lock)) then
      do_read_lock=issue_read_lock      
    else
      do_read_lock=.true.
    end if

    if (do_read_lock) call check_thread_status(forthread_rwlock_rdlock(reduction_progress_rwlock))
    generic=>c_get_generic(reduction_progresses, generate_reduction_key(field_name, timestep, reduction_operator))     
    if (do_read_lock) call check_thread_status(forthread_rwlock_unlock(reduction_progress_rwlock))
    if (associated(generic)) then
      select type(generic)
      type is (reduction_progress_type)      
        find_reduction_progress=>generic
      end select
    else
      find_reduction_progress=>null()
    end if
  end function find_reduction_progress

  subroutine remove_reduction_progress(reduction_progress)
    type(reduction_progress_type), intent(in) :: reduction_progress
        
    class(*), pointer :: generic
    character(len=STRING_LENGTH) :: specific_key

    specific_key=generate_reduction_key(reduction_progress%field_name, reduction_progress%timestep,&
         reduction_progress%reduction_operator)
    call check_thread_status(forthread_rwlock_wrlock(reduction_progress_rwlock))
    generic=>c_get_generic(reduction_progresses, specific_key)
    call c_remove(reduction_progresses, specific_key)
    call check_thread_status(forthread_rwlock_unlock(reduction_progress_rwlock))
    if (associated(generic)) deallocate(generic)
  end subroutine remove_reduction_progress

  character(len=STRING_LENGTH) function generate_reduction_key(field_name, timestep, reduction_operator)
    character(len=*), intent(in) :: field_name
    integer, intent(in) :: timestep, reduction_operator

    generate_reduction_key=trim(field_name)//"#"//trim(conv_to_string(timestep))//"#"// trim(conv_to_string(reduction_operator))
  end function generate_reduction_key

  function retrieve_reduction_progress(mapentry)
    type(mapentry_type), intent(in) :: mapentry
    type(reduction_progress_type), pointer :: retrieve_reduction_progress

    class(*), pointer :: generic

    generic=>c_get_generic(mapentry)

    if (associated(generic)) then
      select type(generic)
      type is (reduction_progress_type)      
        retrieve_reduction_progress=>generic
      end select
    else
      retrieve_reduction_progress=>null()
    end if 
  end function retrieve_reduction_progress

  integer function get_reduction_operator(op_string)
    character(len=*), intent(in) :: op_string
    
    if (op_string .eq. "mean") then
      get_reduction_operator=mean
    else if (op_string .eq. "min") then
      get_reduction_operator=min
    else if (op_string .eq. "max") then
      get_reduction_operator=max
    else if (op_string .eq. "sum") then
      get_reduction_operator=sum
    else
      call log_log(log_error, "Reduction operator '"//trim(op_string)//"' not recognised")
    end if
  end function get_reduction_operator  
end module reduction_inter_io_mod