reduction_inter_io_mod Module Reference

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 175 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 (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 83 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 155 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 91 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 70 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 51 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 238 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 127 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.

  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.

  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.

  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.

  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.

  integer, volatile :: reduction_progress_rwlock, inter_io_description_mutex, clean_progress_mutex, &
       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.

  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.