Reduction inter IO action which will perform reductions between IO servers. This is not as trivial as calling the MPI function as it is nondeterministic when messages will arrive and hence when one reduction on a process and a reduction on another should be called. More...

Data Types
type	reduction_progress_type

Functions/Subroutines
subroutine, public	init_reduction_inter_io (io_configuration)
	Initialises the reduction action. More...

subroutine	handle_recv_data_from_io_server (io_configuration, data_buffer, inter_io_index)
	Handles the receiving of data from some other IO server. This is issued call back style within a thread to handle that data. More...

logical function, public	check_reduction_inter_io_for_completion (io_configuration)
	Checks this action for completion, when all are completed then the IO server can shutdown as this is called once all MONC processes have deregistered. More...

subroutine, public	finalise_reduction_inter_io (io_configuration)
	Finalises the reduction action, waiting for all outstanding requests and then freeing data. More...

subroutine, public	perform_inter_io_reduction (io_configuration, field_values, field_size, reduction_field_name, reduction_op, root, timestep, completion_procedure)
	Actually handles the processing for this data wrt the vertical reduction. More...

subroutine	clean_progress (myrank)

logical function	check_and_clean_progress (myrank)
	Checks all the reduction progresses and will remove any that have completed. This is designed to be called from an IO server other than 0 (the master IO server) and it checks if the outstanding async send handle has completed. Checking on the master IO server or checking any progress that is not currently sending is fine and will not impact the correctness (but obviously the progress wont be freed) More...

subroutine	integrate_io_server_collective_values (reduction_op, reduction_progress, single_server_values, number_elements, collective_values_empty)
	Integrates the collective values from another IO server into the currently stored values. More...

subroutine	handle_local_moncs_completed_collective (io_configuration, reduction_progress)
	Handles the case where the local MONC processes have completed their collective operation for a specific reduction and, for this IO server, it either needs to send its value to the master IO server or, if it is the master, check for completion. More...

subroutine	handle_process_recv_from_other_io_server (io_configuration, inter_io_comm, myrank, data_buffer, number_io_servers)
	Handles the data received from another IO server, locates the correct reduction progress, appends the information and then checks for & deals with the situation where that reduction is completed. More...

subroutine	handle_collective_completed (io_configuration, reduction_progress)
	Handles the situation where collective communication for a specific reduction has completed across all IO servers. More...

type(reduction_progress_type) function, pointer	find_or_add_reduction_progress (timestep, reduction_operator, root, field_name, completion_procedure)
	Finds or adds a specific reduction progress based upon the timestep and reduction operator. If none can be found then a new progress is added in. With new progresses this procedure will initialise them. More...

type(reduction_progress_type) function, pointer	find_reduction_progress (timestep, reduction_operator, field_name, issue_read_lock)
	Locates a specific reduction progress based upon the timestep, operator and field name. More...

subroutine	remove_reduction_progress (reduction_progress)
	Removes a specific reduction progress. More...

character(len=string_length) function	generate_reduction_key (field_name, timestep, reduction_operator)
	Generates the lookup key that is used for the map storage of reduction progresses. More...

type(reduction_progress_type) function, pointer	retrieve_reduction_progress (mapentry)
	Helper function to retrieve the reduction progress from a mapentry. More...

integer function, public	get_reduction_operator (op_string)
	Given the map of action attributes this procedure will identify the reduction operator that has been selected by the configuration. More...

Variables
integer, parameter	mean =1

integer, parameter	min =2

integer, parameter	max =3

integer, parameter	sum =4

integer, parameter	my_inter_io_tag =1

integer, parameter	perform_clean_every =200

character(len= *), parameter	my_inter_io_name ="reductioninterio"

integer, volatile	reduction_progress_rwlock

integer, volatile	inter_io_description_mutex

integer, volatile	clean_progress_mutex

integer, volatile	reduction_count_mutex

integer, volatile	previous_clean_reduction_count

integer, volatile	reduction_count

type(hashmap_type), volatile	reduction_progresses

logical, volatile	initialised =.false.

Detailed Description

Reduction inter IO action which will perform reductions between IO servers. This is not as trivial as calling the MPI function as it is nondeterministic when messages will arrive and hence when one reduction on a process and a reduction on another should be called.

Function/Subroutine Documentation

◆ check_and_clean_progress()

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

private

Checks all the reduction progresses and will remove any that have completed. This is designed to be called from an IO server other than 0 (the master IO server) and it checks if the outstanding async send handle has completed. Checking on the master IO server or checking any progress that is not currently sending is fine and will not impact the correctness (but obviously the progress wont be freed)

Definition at line 174 of file reduction-inter-io.F90.

     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 (specific_reduction_progress%async_handle == mpi_request_null) 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

Here is the call graph for this function:

Here is the caller graph for this function:

◆ check_reduction_inter_io_for_completion()

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

Checks this action for completion, when all are completed then the IO server can shutdown as this is called once all MONC processes have deregistered.

Parameters

io_configuration Configuration state of the IO server

Returns: Whether the action has completed

Definition at line 82 of file reduction-inter-io.F90.

     type(io_configuration_type), intent(inout) :: io_configuration
  
     check_reduction_inter_io_for_completion=check_and_clean_progress(io_configuration%my_io_rank)

Here is the call graph for this function:

◆ clean_progress()

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

private

Definition at line 154 of file reduction-inter-io.F90.

     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

Here is the call graph for this function:

Here is the caller graph for this function:

◆ finalise_reduction_inter_io()

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

Finalises the reduction action, waiting for all outstanding requests and then freeing data.

Parameters

io_configuration Configuration state of the IO server

Definition at line 90 of file reduction-inter-io.F90.

     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

Here is the call graph for this function:

Here is the caller graph for this function:

◆ find_or_add_reduction_progress()

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

private

Finds or adds a specific reduction progress based upon the timestep and reduction operator. If none can be found then a new progress is added in. With new progresses this procedure will initialise them.

Parameters

timestep	The timestep to match
reduction_operator	The reduction operator to match
field_name	The name of the field that the reduction type represents
num_vectors	The number of reduction vectors (items to reduce) to be stored

Returns: A reduction progress data object

Definition at line 360 of file reduction-inter-io.F90.

     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

Here is the call graph for this function:

Here is the caller graph for this function:

◆ find_reduction_progress()

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

private

Locates a specific reduction progress based upon the timestep, operator and field name.

Parameters

timestep	The timestep to search for
reduction_operator	The reduction operator to search for
field_name	The field name which must match
reduction_progress_location	Optional location which is set to be the index of the matching progress item

Returns: Pointer to the reduction progress or null if none is found

Definition at line 406 of file reduction-inter-io.F90.

     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

Here is the call graph for this function:

Here is the caller graph for this function:

◆ generate_reduction_key()

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

private

Generates the lookup key that is used for the map storage of reduction progresses.

Parameters

field_name	The field name
timestep	The timestep
reduction_operator	The reduction operator

Definition at line 455 of file reduction-inter-io.F90.

     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))

Here is the caller graph for this function:

◆ get_reduction_operator()

integer function, public reduction_inter_io_mod::get_reduction_operator ( character(len=*), intent(in) op_string )

Given the map of action attributes this procedure will identify the reduction operator that has been selected by the configuration.

Parameters

action_attributes Action attributes from the IO server configuration

Returns: The reduction operator

Definition at line 487 of file reduction-inter-io.F90.

     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

Here is the call graph for this function:

◆ handle_collective_completed()

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

private

Handles the situation where collective communication for a specific reduction has completed across all IO servers.

Parameters

reduction_progress	The reduction progress data type
number_io_servers	The total number of IO servers

Definition at line 338 of file reduction-inter-io.F90.

     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)

Here is the call graph for this function:

Here is the caller graph for this function:

◆ handle_local_moncs_completed_collective()

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

private

Handles the case where the local MONC processes have completed their collective operation for a specific reduction and, for this IO server, it either needs to send its value to the master IO server or, if it is the master, check for completion.

Parameters

io_configuration	Configuration state of the IO server
reduction_progress	The specific reduction progress data item that represents this reduction
z_size	Size in Z
reduction_progress_location	Location in the reduction progresses list that this single progress item resides at

Definition at line 272 of file reduction-inter-io.F90.

     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

Here is the call graph for this function:

Here is the caller graph for this function:

◆ handle_process_recv_from_other_io_server()

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

private

Handles the data received from another IO server, locates the correct reduction progress, appends the information and then checks for & deals with the situation where that reduction is completed.

Parameters

number_io_servers	The total number of IO servers
z_size	Number of levels in the vertical

Definition at line 304 of file reduction-inter-io.F90.

     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))

Here is the call graph for this function:

Here is the caller graph for this function:

◆ handle_recv_data_from_io_server()

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

private

Handles the receiving of data from some other IO server. This is issued call back style within a thread to handle that data.

Parameters

io_configuration	Configuration state of the IO server
inter_io_index	Index of the inter IO communication description

Definition at line 69 of file reduction-inter-io.F90.

     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)

Here is the call graph for this function:

Here is the caller graph for this function:

◆ init_reduction_inter_io()

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

Initialises the reduction action.

Parameters

io_configuration The IO server configuration

Definition at line 50 of file reduction-inter-io.F90.

     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

Here is the call graph for this function:

Here is the caller graph for this function:

◆ integrate_io_server_collective_values()

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

private

Integrates the collective values from another IO server into the currently stored values.

Parameters

reduction_op	The reduction operator to perform
reduction_progress	The progress data type which is updated
single_server_values	The values from the IO server which need to be integrated
dim_one_size	Size in first dimension
target_index	The index where we are putting the values into the current value array of reduction progress
collective_values_empty	Whether the collective values is empty

Definition at line 237 of file reduction-inter-io.F90.

     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

Here is the caller graph for this function:

◆ perform_inter_io_reduction()

subroutine, public reduction_inter_io_mod::perform_inter_io_reduction	(	type(io_configuration_type), intent(inout)	io_configuration,
		real(kind=double_precision), dimension(:)	field_values,
		integer, intent(in)	field_size,
		character(len=*), intent(in)	reduction_field_name,
		integer, intent(in)	reduction_op,
		integer, intent(in)	root,
		integer, intent(in)	timestep,
		procedure(handle_completion)	completion_procedure
	)

Actually handles the processing for this data wrt the vertical reduction.

Parameters

io_configuration	Configuration of the IO server
field_values	The values to communicate
field_size	Number of elements to communicate
reduction_field_name	Field name that the reduction will be performed over
reduction_op	The reduction operator to use
root	The root IO server process
timestep	The timestep this is issued at
completion_procedure	Callback completion procedure

Definition at line 125 of file reduction-inter-io.F90.

     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    

Here is the call graph for this function:

Here is the caller graph for this function:

◆ remove_reduction_progress()

subroutine reduction_inter_io_mod::remove_reduction_progress ( type(reduction_progress_type), intent(in) reduction_progress )

private

Removes a specific reduction progress.

Parameters

reduction_progress The reduction progress to remove from the list

Definition at line 436 of file reduction-inter-io.F90.

     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)

Here is the call graph for this function:

Here is the caller graph for this function:

◆ retrieve_reduction_progress()

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

private

Helper function to retrieve the reduction progress from a mapentry.

Parameters

mapentry The map entry to retrieve from

Returns: The progress data object in the map entry or null if none is found

Definition at line 465 of file reduction-inter-io.F90.

     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 

Here is the caller graph for this function:

Variable Documentation

◆ clean_progress_mutex

integer, volatile reduction_inter_io_mod::clean_progress_mutex

private

Definition at line 39 of file reduction-inter-io.F90.

◆ initialised

logical, volatile reduction_inter_io_mod::initialised =.false.

private

Definition at line 42 of file reduction-inter-io.F90.

42 logical, volatile :: initialised=.false.

◆ inter_io_description_mutex

integer, volatile reduction_inter_io_mod::inter_io_description_mutex

private

Definition at line 39 of file reduction-inter-io.F90.

◆ max

integer, parameter reduction_inter_io_mod::max =3

private

Definition at line 27 of file reduction-inter-io.F90.

◆ mean

integer, parameter reduction_inter_io_mod::mean =1

private

Definition at line 27 of file reduction-inter-io.F90.

27 integer, parameter :: MEAN=1, min=2, max=3, sum=4

◆ min

integer, parameter reduction_inter_io_mod::min =2

private

Definition at line 27 of file reduction-inter-io.F90.

◆ my_inter_io_name

character(len=*), parameter reduction_inter_io_mod::my_inter_io_name ="reductioninterio"

private

Definition at line 29 of file reduction-inter-io.F90.

29 character(len=*), parameter :: MY_INTER_IO_NAME="reductioninterio"

◆ my_inter_io_tag

integer, parameter reduction_inter_io_mod::my_inter_io_tag =1

private

Definition at line 28 of file reduction-inter-io.F90.

28 integer, parameter :: MY_INTER_IO_TAG=1, perform_clean_every=200

◆ perform_clean_every

integer, parameter reduction_inter_io_mod::perform_clean_every =200

private

Definition at line 28 of file reduction-inter-io.F90.

◆ previous_clean_reduction_count

integer, volatile reduction_inter_io_mod::previous_clean_reduction_count

private

Definition at line 39 of file reduction-inter-io.F90.

◆ reduction_count

integer, volatile reduction_inter_io_mod::reduction_count

private

Definition at line 39 of file reduction-inter-io.F90.

◆ reduction_count_mutex

integer, volatile reduction_inter_io_mod::reduction_count_mutex

private

Definition at line 39 of file reduction-inter-io.F90.

◆ reduction_progress_rwlock

integer, volatile reduction_inter_io_mod::reduction_progress_rwlock

private

Definition at line 39 of file reduction-inter-io.F90.

39 integer, volatile :: reduction_progress_rwlock, inter_io_description_mutex, clean_progress_mutex, &

40 reduction_count_mutex, previous_clean_reduction_count, reduction_count

◆ reduction_progresses

type(hashmap_type), volatile reduction_inter_io_mod::reduction_progresses

private

Definition at line 41 of file reduction-inter-io.F90.

41 type(hashmap_type), volatile :: reduction_progresses

◆ sum

integer, parameter reduction_inter_io_mod::sum =4

private

Definition at line 27 of file reduction-inter-io.F90.

Data Types

Functions/Subroutines

Variables

Detailed Description

Function/Subroutine Documentation

◆ check_and_clean_progress()

◆ check_reduction_inter_io_for_completion()

◆ clean_progress()

◆ finalise_reduction_inter_io()

◆ find_or_add_reduction_progress()

◆ find_reduction_progress()

◆ generate_reduction_key()

◆ get_reduction_operator()

◆ handle_collective_completed()

◆ handle_local_moncs_completed_collective()

◆ handle_process_recv_from_other_io_server()

◆ handle_recv_data_from_io_server()

◆ init_reduction_inter_io()

◆ integrate_io_server_collective_values()

◆ perform_inter_io_reduction()

◆ remove_reduction_progress()

◆ retrieve_reduction_progress()

Variable Documentation

◆ clean_progress_mutex

◆ initialised

◆ inter_io_description_mutex

◆ max

◆ mean

◆ min

◆ my_inter_io_name

◆ my_inter_io_tag

◆ perform_clean_every

◆ previous_clean_reduction_count

◆ reduction_count

◆ reduction_count_mutex

◆ reduction_progress_rwlock

◆ reduction_progresses

◆ sum