io_server_mod Module Reference

The main IO server functionality which handles waiting for commands and data both of which are delt with. The lower level details of the communication, configuration parsing etc are all held elsewhere. The server can be thought of similar to a bus, with command and data channels. The command gives context to what is on the data channel and not all commands require data (such as deregistration of MONC process) More...

Functions/Subroutines
subroutine, public	io_server_run (options_database, io_communicator_arg, provided_threading, total_global_processes, continuation_run, io_configuration_file)
	Called to start the IO server and once this subroutine returns then it indicates that the IO server has finished. The runtime is spent in here awaiting commands and then dealing with them. Termination occurs when all MONC processes have deregistered, note that to trigger this then at least one MONC process must first register. More...
subroutine	check_for_condi_conflict (raw_contents, options_database)
	Handle potential conditional diagnostics conflict Provides a more helpful error in the case where conditional diagnostics are requested as output, but their components are not enabled. We check this by searching the io_xml_configuration. More...
logical function	await_command (command, source, data_buffer)
	Awaits a command or shutdown from MONC processes and other IO servers. More...
subroutine	termination_callback (io_configuration, values, field_name, timestep)
	This is the termination callback which is called once all MONCs have deregistered, no sends are active by inter IO communications and all threads are idle. This shuts down the inter IO listening and kickstarts finalisation and closure. More...
subroutine	handle_command_message (command, source, data_buffer)
	Called to handle a specific command that has been recieved. More...
subroutine	handle_inter_io_communication_command (arguments, data_buffer)
	Handles inter IO server communications. More...
subroutine	free_individual_registered_monc_aspects ()
	Frees up the memory associated with individual registered MONCs. This is done at the end for all MONCs as we can't deallocate dynamically in a threaded environment without excessive ordering and locking in case some data processing is queued or in progress. More...
subroutine	handle_deregistration_command (arguments, data_buffer)
	Deregisteres a specific MONC source process. More...
subroutine	pull_back_data_message_and_handle (source, data_set)
	Retrieves the message from MONC off the data channel and throws this to a thread in the thread pool to actually process We do it this way to enforce ordering between the command (including the data set ID) and the raw data itself. More...
subroutine	handle_data_message (arguments, data_buffer)
	Handles the command for data download from a specific process. This will allocate the receive buffer and then call to get the data. Once it has been received then the data is run against handling rules. More...
subroutine	handle_monc_registration (arguments, data_buffer)
	Handles registration from some MONC process. The source process sends some data description to this IO server which basically tells the IO server the size of the array datas (which might be different on different processes in the case of uneven decomposition.) Based upon this a communication (MPI) data type is constructed and the data size in bytes determined. More...
integer function, dimension(2)	send_configuration_to_registree (source)
	Sends the data and field descriptions to the MONC process that just registered with the IO server. More...
subroutine	init_data_definition (source, monc_defn)
	Initialise the sizing of data definitions from a MONC process. The IO server determines, from configuration, the structure of each data definition but the size of the arrays depends upon the MONC process (due to uneven distribution of data etc...) This receives the sizing message and then builds the MPI datatype for each data definition that the IO server will receive from that specific MONC process. The field sizings are for all fields in every data definition, and these are applied to each data definition which will simply ignore non matching fields. More...
subroutine	get_monc_information_data (source)
	Retrieves MONC information data, this is sent by MONC (and received) regardless, but only actioned if the data has not already been set. More...
subroutine	register_present_field_names_to_federators (data_description, recv_count)
	Registers with the writer federator the set of fields (prognostic and diagnostic) that are available, this is based on the array/optional fields present from MONC and the non-optional scalars. This is quite an expensive operation, so only done once. More...
subroutine	handle_monc_dimension_information (data_description, monc_defn)
	Handles the provided local MONC dimension and data layout information. More...

Variables
integer	mpi_type_data_sizing_description
	The MPI type for field sizing (i.e. array size etc send when MONCs register) More...
integer	mpi_type_definition_description
	The MPI data type for data descriptions sent to MONCs. More...
integer	mpi_type_field_description
	The MPI data type for field descriptions sent to MONCs. More...
type(io_configuration_type), save, volatile	io_configuration
	Internal representation of the IO configuration. More...
logical, volatile	contine_poll_messages
	Whether to continue waiting command messages from any MONC processes. More...
logical, volatile	initialised_present_data
logical, volatile	contine_poll_interio_messages
logical, volatile	already_registered_finishing_call
type(field_description_type), dimension(:), allocatable	registree_field_descriptions
type(definition_description_type), dimension(:), allocatable	registree_definition_descriptions
integer, volatile	monc_registration_lock

Detailed Description

The main IO server functionality which handles waiting for commands and data both of which are delt with. The lower level details of the communication, configuration parsing etc are all held elsewhere. The server can be thought of similar to a bus, with command and data channels. The command gives context to what is on the data channel and not all commands require data (such as deregistration of MONC process)

Function/Subroutine Documentation

await_command()

logical function io_server_mod::await_command ( integer, intent(out)  command, integer, intent(out)  source, character, dimension(:), allocatable  data_buffer  )

private

Awaits a command or shutdown from MONC processes and other IO servers.

Parameters

command	The command received is output
source	The source process received is output

Returns

Whether to continue polling for commands (and whether to process the current output)

Definition at line 166 of file ioserver.F90.

    integer, intent(out) :: command, source
    character, dimension(:), allocatable :: data_buffer
 
    logical :: completed, inter_io_complete
 
    completed=.false.
    await_command=.false.
    do while(.not. completed)
      if (.not. contine_poll_messages .and. .not. contine_poll_interio_messages) return
      if (contine_poll_messages) then
        if (test_for_command(command, source)) then
          await_command=.true.
          return
        end if
      end if
      if (contine_poll_interio_messages .and. allocated(io_configuration%inter_io_communications)) then       
        inter_io_complete=test_for_inter_io(io_configuration%inter_io_communications, &
             io_configuration%number_inter_io_communications, io_configuration%io_communicator, command, source, data_buffer) 
        if (inter_io_complete) then
          await_command=.true.
          return
        end if
      end if
      if (.not. contine_poll_messages .and. .not. already_registered_finishing_call) then
        if (check_diagnostic_federator_for_completion(io_configuration) .and. threadpool_is_idle()) then
          already_registered_finishing_call=.true.          
          call perform_global_callback(io_configuration, "termination", 1, termination_callback)          
        end if
      end if  
      if (.not. completed) call pause_for_mpi_interleaving()
    end do    

Here is the call graph for this function:

Here is the caller graph for this function:

check_for_condi_conflict()

subroutine io_server_mod::check_for_condi_conflict ( character, dimension(:), intent(in)  raw_contents, type(hashmap_type), intent(inout)  options_database  )

private

Handle potential conditional diagnostics conflict Provides a more helpful error in the case where conditional diagnostics are requested as output, but their components are not enabled. We check this by searching the io_xml_configuration.

Parameters

raw_contents,intended	to be the io_xml_configuration character array
options_database

Definition at line 145 of file ioserver.F90.

    character, dimension(:), intent(in) :: raw_contents
    type(hashmap_type), intent(inout) :: options_database
    character(len=size(raw_contents)) :: string_to_process
    integer :: i
 
    if (.not. options_get_logical(options_database, "conditional_diagnostics_column_enabled")) then
      do i=1, size(raw_contents)
        string_to_process(i:i)=raw_contents(i)
      end do
      if (index(string_to_process,"CondDiags_") .ne. 0) then
        call log_log(log_error, &
            "Conditional diagnostics are DISABLED but requested via xml.  Enable or remove request to resolve.")
      end if
    end if

Here is the call graph for this function:

Here is the caller graph for this function:

free_individual_registered_monc_aspects()

subroutine io_server_mod::free_individual_registered_monc_aspects

private

Frees up the memory associated with individual registered MONCs. This is done at the end for all MONCs as we can't deallocate dynamically in a threaded environment without excessive ordering and locking in case some data processing is queued or in progress.

Definition at line 250 of file ioserver.F90.

    integer :: i, specific_monc_data_type
    type(iterator_type) :: types_iterator
 
    do i=1, size(io_configuration%registered_moncs)
      types_iterator=c_get_iterator(io_configuration%registered_moncs(i)%registered_monc_types)
      do while (c_has_next(types_iterator))
        specific_monc_data_type=c_get_integer(c_next_mapentry(types_iterator))
        call free_mpi_type(specific_monc_data_type)
      end do      
      if (allocated(io_configuration%registered_moncs(i)%field_start_locations)) &
           deallocate(io_configuration%registered_moncs(i)%field_start_locations)
      if (allocated(io_configuration%registered_moncs(i)%field_end_locations)) &
           deallocate(io_configuration%registered_moncs(i)%field_end_locations)
      if (allocated(io_configuration%registered_moncs(i)%definition_names)) &
           deallocate(io_configuration%registered_moncs(i)%definition_names)
      if (allocated(io_configuration%registered_moncs(i)%dimensions)) deallocate(io_configuration%registered_moncs(i)%dimensions)
    end do

Here is the call graph for this function:

Here is the caller graph for this function:

get_monc_information_data()

subroutine io_server_mod::get_monc_information_data ( integer, intent(in)  source )

private

Retrieves MONC information data, this is sent by MONC (and received) regardless, but only actioned if the data has not already been set.

Parameters

source

MONC source process

Definition at line 468 of file ioserver.F90.

    integer, intent(in) :: source
 
    character, dimension(:), allocatable :: buffer
    character(len=STRING_LENGTH) :: q_field_name, cd_field_name
    integer :: buffer_size, z_size, num_q_fields, n, current_point, recv_count
    type(data_sizing_description_type) :: field_description
    real(kind=default_precision) :: dreal
    logical :: field_found
 
 
    z_size=c_get_integer(io_configuration%dimension_sizing, "z")
    num_q_fields=c_get_integer(io_configuration%dimension_sizing, "qfields")
 
    buffer_size=(kind(dreal)*z_size)*2 + (string_length * num_q_fields) &
                 + 2*ncond*string_length + 2*ndiag*string_length 
    allocate(buffer(buffer_size))
    recv_count=data_receive(mpi_byte, buffer_size, source, buffer)
    if (.not. io_configuration%general_info_set) then
      call check_thread_status(forthread_mutex_lock(io_configuration%general_info_mutex))
      if (.not. io_configuration%general_info_set) then
        io_configuration%general_info_set=.true.
        allocate(io_configuration%zn_field(z_size))
        allocate(io_configuration%z_field(z_size))
        io_configuration%zn_field=transfer(buffer(1:kind(dreal)*z_size), io_configuration%zn_field)
        current_point=(kind(dreal)*z_size)
        if (num_q_fields .gt. 0) then
          do n=1, num_q_fields
            q_field_name=transfer(buffer(current_point+1:current_point+string_length), q_field_name)
            current_point=current_point+string_length
            call replace_character(q_field_name, " ", "_")
            call c_add_string(io_configuration%q_field_names, q_field_name)
          end do
        end if
        io_configuration%z_field=transfer(buffer(current_point+1:current_point+kind(dreal)*z_size), &
                                          io_configuration%z_field)
        current_point=current_point+(kind(dreal)*z_size)
 
        do n=1,ncond
          cond_request(n)=transfer(buffer(current_point+1:current_point+string_length), cd_field_name)
          current_point=current_point+string_length
          cond_long(n)=transfer(buffer(current_point+1:current_point+string_length), cd_field_name)
          current_point=current_point+string_length
        end do
        do n=1,ndiag
          diag_request(n)=transfer(buffer(current_point+1:current_point+string_length), cd_field_name)
          current_point=current_point+string_length
          diag_long(n)=transfer(buffer(current_point+1:current_point+string_length), cd_field_name)
          current_point=current_point+string_length
        end do
 
      end if
      call provide_q_field_names_to_writer_federator(io_configuration%q_field_names)
      call check_thread_status(forthread_mutex_unlock(io_configuration%general_info_mutex))
    end if
    deallocate(buffer)

Here is the call graph for this function:

Here is the caller graph for this function:

handle_command_message()

subroutine io_server_mod::handle_command_message ( integer, intent(in)  command, integer, intent(in)  source, character, dimension(:), intent(inout), allocatable  data_buffer  )

private

Called to handle a specific command that has been recieved.

Parameters

command	The command which has been received from some process
source	The PID of the source (MONC) process

Definition at line 218 of file ioserver.F90.

    integer, intent(in) :: command, source
    character, dimension(:), allocatable, intent(inout) :: data_buffer
 
    if (command == register_command) then
      call threadpool_start_thread(handle_monc_registration, (/ source /))
    else if (command == deregister_command) then
      call threadpool_start_thread(handle_deregistration_command, (/ source /))
    else if (command == inter_io_communication) then
      call threadpool_start_thread(handle_inter_io_communication_command, (/ source /), data_buffer=data_buffer)      
      deallocate(data_buffer)
    else if (command .ge. data_command_start) then      
      call pull_back_data_message_and_handle(source, command-data_command_start)
    end if    

Here is the call graph for this function:

Here is the caller graph for this function:

handle_data_message()

subroutine io_server_mod::handle_data_message ( integer, dimension(:), intent(in)  arguments, character, dimension(:), intent(inout), optional, allocatable  data_buffer  )

private

Handles the command for data download from a specific process. This will allocate the receive buffer and then call to get the data. Once it has been received then the data is run against handling rules.

Parameters

arguments,element	1 is the source & element 2 is the data_set
data_buffer	The actual data from MONC read from the data channel

Definition at line 322 of file ioserver.F90.

    integer, dimension(:), intent(in) :: arguments
    character, dimension(:), allocatable, intent(inout), optional :: data_buffer
 
    integer :: monc_location, data_set, source, matched_datadefn_index
 
    source=arguments(1)
    data_set=arguments(2)
 
    call check_thread_status(forthread_rwlock_rdlock(monc_registration_lock))
    monc_location=get_monc_location(io_configuration, source)
 
    call check_thread_status(forthread_mutex_lock(io_configuration%registered_moncs(monc_location)%active_mutex))
    io_configuration%registered_moncs(monc_location)%active_threads=&
         io_configuration%registered_moncs(monc_location)%active_threads+1
    call check_thread_status(forthread_mutex_unlock(io_configuration%registered_moncs(monc_location)%active_mutex))
    
    matched_datadefn_index=retrieve_data_definition(io_configuration, &
         io_configuration%registered_moncs(monc_location)%definition_names(data_set))
 
    if (matched_datadefn_index .gt. 0) then
      call inform_writer_federator_time_point(io_configuration, source, data_set, data_buffer)
      call pass_fields_to_diagnostics_federator(io_configuration, source, data_set, data_buffer)
      call provide_monc_data_to_writer_federator(io_configuration, source, data_set, data_buffer)
      call check_writer_for_trigger(io_configuration, source, data_set, data_buffer)
    else
      call log_log(log_warn, "IO server can not find matching data definition with name "&
           //io_configuration%registered_moncs(monc_location)%definition_names(data_set))
    end if    
 
    call check_thread_status(forthread_mutex_lock(io_configuration%registered_moncs(monc_location)%active_mutex))
    io_configuration%registered_moncs(monc_location)%active_threads=&
         io_configuration%registered_moncs(monc_location)%active_threads-1
    call check_thread_status(forthread_cond_signal(io_configuration%registered_moncs(monc_location)%deactivate_condition_variable))
    call check_thread_status(forthread_mutex_unlock(io_configuration%registered_moncs(monc_location)%active_mutex))
    call check_thread_status(forthread_rwlock_unlock(monc_registration_lock))

Here is the call graph for this function:

Here is the caller graph for this function:

handle_deregistration_command()

subroutine io_server_mod::handle_deregistration_command ( integer, dimension(:), intent(in)  arguments, character, dimension(:), intent(inout), optional, allocatable  data_buffer  )

private

Deregisteres a specific MONC source process.

Parameters

source

The MONC process PID that we are deregistering

Definition at line 272 of file ioserver.F90.

    integer, dimension(:), intent(in) :: arguments
    character, dimension(:), allocatable, intent(inout), optional :: data_buffer
 
    integer :: monc_location, source
 
    source=arguments(1)
    monc_location=get_monc_location(io_configuration, source)
    call check_thread_status(forthread_mutex_lock(io_configuration%registered_moncs(monc_location)%active_mutex))
    do while (io_configuration%registered_moncs(monc_location)%active_threads .gt. 0)
      call check_thread_status(forthread_cond_wait(io_configuration%registered_moncs(monc_location)%deactivate_condition_variable,&
             io_configuration%registered_moncs(monc_location)%active_mutex))
    end do
    call check_thread_status(forthread_mutex_unlock(io_configuration%registered_moncs(monc_location)%active_mutex))
    call check_thread_status(forthread_rwlock_wrlock(monc_registration_lock))
    io_configuration%active_moncs=io_configuration%active_moncs-1
    if (io_configuration%active_moncs==0) contine_poll_messages=.false.
    call check_thread_status(forthread_rwlock_unlock(monc_registration_lock))

Here is the call graph for this function:

Here is the caller graph for this function:

handle_inter_io_communication_command()

subroutine io_server_mod::handle_inter_io_communication_command ( integer, dimension(:), intent(in)  arguments, character, dimension(:), intent(inout), optional, allocatable  data_buffer  )

private

Handles inter IO server communications.

Parameters

arguments

The thread based arguments, this is the index of the inter IO server description

Definition at line 236 of file ioserver.F90.

    integer, dimension(:), intent(in) :: arguments
    character, dimension(:), allocatable, intent(inout), optional :: data_buffer
 
    integer :: source
 
    source=arguments(1)
 
    call io_configuration%inter_io_communications(source)%handling_procedure(io_configuration, data_buffer, source)

Here is the caller graph for this function:

handle_monc_dimension_information()

subroutine io_server_mod::handle_monc_dimension_information ( type(data_sizing_description_type), dimension(:)  data_description, type(io_configuration_registered_monc_type), intent(inout)  monc_defn  )

private

Handles the provided local MONC dimension and data layout information.

Parameters

data_description	The data descriptions sent over from MONC
monc_defn	The corresponding MONC definition data structure

Definition at line 562 of file ioserver.F90.

    type(io_configuration_registered_monc_type), intent(inout) :: monc_defn
    type(data_sizing_description_type), dimension(:) :: data_description
 
    type(data_sizing_description_type) :: field_description
    integer :: i
    logical :: field_found
 
    field_found=get_data_description_from_name(data_description, local_sizes_key, field_description)
    if (.not. field_found) call log_log(log_error, "Malformed MONC registration, no local size information")
    do i=1,3
      monc_defn%local_dim_sizes(i)=field_description%dim_sizes(i)
    end do
    field_found=get_data_description_from_name(data_description, local_start_points_key, field_description)
    if (.not. field_found) call log_log(log_error, "Malformed MONC registration, no local start point information")
    do i=1,3
      monc_defn%local_dim_starts(i)=field_description%dim_sizes(i)
    end do
    field_found=get_data_description_from_name(data_description, local_end_points_key, field_description)
    if (.not. field_found) call log_log(log_error, "Malformed MONC registration, no local end point information")
    do i=1,3
      monc_defn%local_dim_ends(i)=field_description%dim_sizes(i)
    end do

Here is the call graph for this function:

Here is the caller graph for this function:

handle_monc_registration()

subroutine io_server_mod::handle_monc_registration ( integer, dimension(:), intent(in)  arguments, character, dimension(:), intent(inout), optional, allocatable  data_buffer  )

private

Handles registration from some MONC process. The source process sends some data description to this IO server which basically tells the IO server the size of the array datas (which might be different on different processes in the case of uneven decomposition.) Based upon this a communication (MPI) data type is constructed and the data size in bytes determined.

Parameters

source

The PID of the MONC process that is registering itself

Definition at line 364 of file ioserver.F90.

    integer, dimension(:), intent(in) :: arguments
    character, dimension(:), allocatable, intent(inout), optional :: data_buffer
 
    integer :: configuration_send_request(2), ierr, number_data_definitions, this_monc_index, source
 
    source=arguments(1)
    configuration_send_request=send_configuration_to_registree(source)
    number_data_definitions=io_configuration%number_of_data_definitions
 
    call check_thread_status(forthread_rwlock_wrlock(monc_registration_lock))
 
    io_configuration%number_of_moncs=io_configuration%number_of_moncs+1
    this_monc_index=io_configuration%number_of_moncs
    if (io_configuration%number_of_moncs .gt. size(io_configuration%registered_moncs)) then
      call log_log(log_error, "You have a high ratio of computational cores to IO servers, the limit is currently 100")
      ! The extension of the MONC registration array is broken as the pointers involved in the map does not get copied across
      ! we could manually do this, but that is for another day! If you need to extend these limits either increase the constants
      ! or fix the extension, I don't think it will be too hard to fix the extension bit (copy the maps manually)
      call extend_registered_moncs_array(io_configuration)      
    end if
 
    io_configuration%active_moncs=io_configuration%active_moncs+1
    call check_thread_status(forthread_rwlock_unlock(monc_registration_lock))
 
    call c_put_integer(io_configuration%monc_to_index, conv_to_string(source), this_monc_index)
 
    call check_thread_status(forthread_mutex_init(io_configuration%registered_moncs(this_monc_index)%active_mutex, -1))
    call check_thread_status(forthread_cond_init(&
         io_configuration%registered_moncs(this_monc_index)%deactivate_condition_variable, -1))
    io_configuration%registered_moncs(this_monc_index)%active_threads=0
    io_configuration%registered_moncs(this_monc_index)%source_id=source
 
    allocate(io_configuration%registered_moncs(this_monc_index)%field_start_locations(number_data_definitions), &
         io_configuration%registered_moncs(this_monc_index)%field_end_locations(number_data_definitions), &
         io_configuration%registered_moncs(this_monc_index)%definition_names(number_data_definitions), &
         io_configuration%registered_moncs(this_monc_index)%dimensions(number_data_definitions))
 
    ! Wait for configuration to have been sent to registree
    call waitall_for_mpi_requests(configuration_send_request, 2)
    call init_data_definition(source, io_configuration%registered_moncs(this_monc_index))

Here is the call graph for this function:

Here is the caller graph for this function:

init_data_definition()

subroutine io_server_mod::init_data_definition ( integer, intent(in)  source, type(io_configuration_registered_monc_type), intent(inout)  monc_defn  )

private

Initialise the sizing of data definitions from a MONC process. The IO server determines, from configuration, the structure of each data definition but the size of the arrays depends upon the MONC process (due to uneven distribution of data etc...) This receives the sizing message and then builds the MPI datatype for each data definition that the IO server will receive from that specific MONC process. The field sizings are for all fields in every data definition, and these are applied to each data definition which will simply ignore non matching fields.

Parameters

source	The source MONC PID
monc_defn	The corresponding MONC definition data structure

Definition at line 433 of file ioserver.F90.

    integer, intent(in) :: source
    type(io_configuration_registered_monc_type), intent(inout) :: monc_defn
 
    type(data_sizing_description_type) :: data_description(io_configuration%number_of_distinct_data_fields+4)
    integer :: created_mpi_type, data_size, recv_count, i
    type(data_sizing_description_type) :: field_description
    logical :: field_found
    
    recv_count=data_receive(mpi_type_data_sizing_description, io_configuration%number_of_distinct_data_fields+4, &
         source, description_data=data_description)
 
    call handle_monc_dimension_information(data_description, monc_defn)
     
    do i=1, io_configuration%number_of_data_definitions
      created_mpi_type=build_mpi_datatype(io_configuration%data_definitions(i), data_description, data_size, &
           monc_defn%field_start_locations(i), monc_defn%field_end_locations(i), monc_defn%dimensions(i))            
 
      call c_put_integer(monc_defn%registered_monc_types, conv_to_string(i), created_mpi_type)
      call c_put_integer(monc_defn%registered_monc_buffer_sizes, conv_to_string(i), data_size)
 
      monc_defn%definition_names(i)=io_configuration%data_definitions(i)%name
    end do
    if (.not. initialised_present_data) then
      initialised_present_data=.true.
      field_found=get_data_description_from_name(data_description, number_q_indicies_key, field_description)
      call c_put_integer(io_configuration%dimension_sizing, "active_q_indicies", field_description%dim_sizes(1))
      call register_present_field_names_to_federators(data_description, recv_count)
    end if
    call get_monc_information_data(source)

Here is the call graph for this function:

Here is the caller graph for this function:

io_server_run()

subroutine, public io_server_mod::io_server_run	(	type(hashmap_type), intent(inout)	options_database,
		integer, intent(in)	io_communicator_arg,
		integer, intent(in)	provided_threading,
		integer, intent(in)	total_global_processes,
		logical, intent(in)	continuation_run,
		character(len=long_string_length), intent(in)	io_configuration_file
	)

Called to start the IO server and once this subroutine returns then it indicates that the IO server has finished. The runtime is spent in here awaiting commands and then dealing with them. Termination occurs when all MONC processes have deregistered, note that to trigger this then at least one MONC process must first register.

Parameters

io_communicator_arg	The IO communicator containing just the IO servers
io_xml_configuration	Textual XML configuration that is used to set up the IO server

Definition at line 66 of file ioserver.F90.

    type(hashmap_type), intent(inout) :: options_database
    integer, intent(in) :: io_communicator_arg, provided_threading, total_global_processes
    logical, intent(in) :: continuation_run
    character(len=LONG_STRING_LENGTH), intent(in) :: io_configuration_file
 
    integer :: command, source, my_rank, ierr
    character, dimension(:), allocatable :: data_buffer, io_xml_configuration
    type(hashmap_type) :: diagnostic_generation_frequency
 
 
    if (continuation_run) then
      ! Handle case where we need to allocate this due to no IO server config
      call read_io_server_configuration(options_get_string(options_database, "checkpoint"), &
           io_xml_configuration, io_communicator_arg)
    end if
 
    if (.not. allocated(io_xml_configuration)) then
      io_xml_configuration=get_io_xml(io_configuration_file)
      if (continuation_run) then
        call mpi_comm_rank(io_communicator_arg, my_rank, ierr)
        if (my_rank == 0) then
          call log_log(log_warn, "No IO server configuration in checkpoint file - starting from XML provided file instead")
        end if
      end if      
    end if    
 
    call check_for_condi_conflict(io_xml_configuration, options_database)
    call configuration_parse(options_database, io_xml_configuration, io_configuration)
    deallocate(io_xml_configuration)
    call threadpool_init(io_configuration)
    call initialise_mpi_communication(provided_threading)
    call check_thread_status(forthread_rwlock_init(monc_registration_lock, -1))
    call check_thread_status(forthread_mutex_init(io_configuration%general_info_mutex, -1))
    initialised_present_data=.false.
    contine_poll_messages=.true.
    contine_poll_interio_messages=.true.
    already_registered_finishing_call=.false.
    io_configuration%io_communicator=io_communicator_arg
    io_configuration%number_of_io_servers=get_number_io_servers(io_communicator_arg)
    io_configuration%number_of_global_moncs=total_global_processes-io_configuration%number_of_io_servers
    io_configuration%my_io_rank=get_my_io_rank(io_communicator_arg)
    call initialise_logging(io_configuration%my_io_rank)
    registree_definition_descriptions=build_definition_description_type_from_configuration(io_configuration)
    registree_field_descriptions=build_field_description_type_from_configuration(io_configuration)
    diagnostic_generation_frequency=initialise_diagnostic_federator(io_configuration)
    call initialise_writer_federator(io_configuration, diagnostic_generation_frequency, continuation_run)
    call c_free(diagnostic_generation_frequency)
    call initialise_writer_field_manager(io_configuration, continuation_run)
 
    mpi_type_data_sizing_description=build_mpi_type_data_sizing_description()
    mpi_type_definition_description=build_mpi_type_definition_description()
    mpi_type_field_description=build_mpi_type_field_description()
 
    call register_command_receive()
 
    do while (await_command(command, source, data_buffer))      
      call handle_command_message(command, source, data_buffer)
    end do
    call threadpool_deactivate()
    call finalise_writer_field_manager()
    call finalise_writer_federator()
    call finalise_diagnostic_federator(io_configuration)
    call check_thread_status(forthread_rwlock_destroy(monc_registration_lock))
    call free_individual_registered_monc_aspects()
    call cancel_requests()
    call free_mpi_type(mpi_type_data_sizing_description)
    call free_mpi_type(mpi_type_definition_description)
    call free_mpi_type(mpi_type_field_description)    
    call threadpool_finalise()

Here is the call graph for this function:

Here is the caller graph for this function:

pull_back_data_message_and_handle()

subroutine io_server_mod::pull_back_data_message_and_handle ( integer, intent(in)  source, integer, intent(in)  data_set  )

private

Retrieves the message from MONC off the data channel and throws this to a thread in the thread pool to actually process We do it this way to enforce ordering between the command (including the data set ID) and the raw data itself.

Parameters

source	Source PID of the MONC process
data_set	ID of the data set being communicated

Definition at line 296 of file ioserver.F90.

    integer, intent(in) :: source, data_set
 
    integer :: specific_monc_data_type, specific_monc_buffer_size, recv_count, monc_location
    character, dimension(:), allocatable :: data_buffer
 
    call check_thread_status(forthread_rwlock_rdlock(monc_registration_lock))
    monc_location=get_monc_location(io_configuration, source)
 
    specific_monc_data_type=c_get_integer(io_configuration%registered_moncs(monc_location)%registered_monc_types, &
         conv_to_string(data_set))
    specific_monc_buffer_size=c_get_integer(io_configuration%registered_moncs(monc_location)%registered_monc_buffer_sizes, &
         conv_to_string(data_set))
 
    allocate(data_buffer(specific_monc_buffer_size))
    recv_count=data_receive(specific_monc_data_type, 1, source, dump_data=data_buffer, data_dump_id=data_set)
 
    call check_thread_status(forthread_rwlock_unlock(monc_registration_lock))
    call threadpool_start_thread(handle_data_message, (/ source,  data_set /), data_buffer=data_buffer)
    deallocate(data_buffer)

Here is the call graph for this function:

Here is the caller graph for this function:

register_present_field_names_to_federators()

subroutine io_server_mod::register_present_field_names_to_federators ( type(data_sizing_description_type), dimension(:), intent(in)  data_description, integer, intent(in)  recv_count  )

private

Registers with the writer federator the set of fields (prognostic and diagnostic) that are available, this is based on the array/optional fields present from MONC and the non-optional scalars. This is quite an expensive operation, so only done once.

Parameters

data_description	Array of data descriptions from MONC
recv_count	Number of data descriptions

Definition at line 531 of file ioserver.F90.

    type(data_sizing_description_type), dimension(:), intent(in) :: data_description
    integer, intent(in) :: recv_count
 
    type(hashset_type) :: present_field_names
    type(hashmap_type) :: diagnostics_field_names_and_roots
    integer :: i, j
 
    do i=1, recv_count
      call c_add_string(present_field_names, data_description(i)%field_name)
    end do
    do i=1, io_configuration%number_of_data_definitions
      do j=1, io_configuration%data_definitions(i)%number_of_data_fields
        if (io_configuration%data_definitions(i)%fields(j)%field_type == scalar_field_type .and. .not. &
             io_configuration%data_definitions(i)%fields(j)%optional) then
          call c_add_string(present_field_names, io_configuration%data_definitions(i)%fields(j)%name)
        end if        
      end do      
    end do
    call c_add_string(present_field_names, "time")
    call c_add_string(present_field_names, "timestep")
    call inform_writer_federator_fields_present(io_configuration, present_field_names)
    diagnostics_field_names_and_roots=determine_diagnostics_fields_available(present_field_names)
    call inform_writer_federator_fields_present(io_configuration, diag_field_names_and_roots=diagnostics_field_names_and_roots)
    call c_free(present_field_names)
    call c_free(diagnostics_field_names_and_roots)

Here is the call graph for this function:

Here is the caller graph for this function:

send_configuration_to_registree()

integer function, dimension(2) io_server_mod::send_configuration_to_registree ( integer, intent(in)  source )

private

Sends the data and field descriptions to the MONC process that just registered with the IO server.

Parameters

source

The MPI rank (MPI_COMM_WORLD) of the registree

Returns

The nonblocking send request handles which can be waited for completion later (overlap compute and communication)

Definition at line 410 of file ioserver.F90.

    integer, intent(in) :: source
    integer :: send_configuration_to_registree(2)
    
    integer :: ierr, srequest(2)
 
    call lock_mpi()
    call mpi_isend(registree_definition_descriptions, size(registree_definition_descriptions), mpi_type_definition_description, &
         source, data_tag, mpi_comm_world, srequest(1), ierr)
    call mpi_isend(registree_field_descriptions, size(registree_field_descriptions), mpi_type_field_description, &
         source, data_tag, mpi_comm_world, srequest(2), ierr)
    call unlock_mpi()
 
    send_configuration_to_registree=srequest    

Here is the call graph for this function:

Here is the caller graph for this function:

termination_callback()

subroutine io_server_mod::termination_callback ( type(io_configuration_type), intent(inout)  io_configuration, real(default_precision), dimension(:)  values, character(len=string_length)  field_name, integer  timestep  )

private

This is the termination callback which is called once all MONCs have deregistered, no sends are active by inter IO communications and all threads are idle. This shuts down the inter IO listening and kickstarts finalisation and closure.

Parameters

io_configuration	The IO server configuration
values	Values (ignored)
field_name	Field name identifier
timestep	Timestep identifier

Definition at line 206 of file ioserver.F90.

    type(io_configuration_type), intent(inout) :: io_configuration
    real(DEFAULT_PRECISION), dimension(:) :: values
    character(len=STRING_LENGTH) :: field_name
    integer :: timestep
 
    contine_poll_interio_messages=.false.

Here is the caller graph for this function:

Variable Documentation

already_registered_finishing_call

logical, volatile io_server_mod::already_registered_finishing_call

private

Definition at line 52 of file ioserver.F90.

contine_poll_interio_messages

logical, volatile io_server_mod::contine_poll_interio_messages

private

Definition at line 52 of file ioserver.F90.

  logical, volatile :: contine_poll_interio_messages, already_registered_finishing_call

contine_poll_messages

logical, volatile io_server_mod::contine_poll_messages

private

Whether to continue waiting command messages from any MONC processes.

Definition at line 50 of file ioserver.F90.

  logical, volatile :: contine_poll_messages, & !< Whether to continue waiting command messages from any MONC processes
       initialised_present_data

initialised_present_data

logical, volatile io_server_mod::initialised_present_data

private

Definition at line 50 of file ioserver.F90.

io_configuration

type(io_configuration_type), save, volatile io_server_mod::io_configuration

private

Internal representation of the IO configuration.

Definition at line 49 of file ioserver.F90.

  type(io_configuration_type), volatile, save :: io_configuration

monc_registration_lock

integer, volatile io_server_mod::monc_registration_lock

private

Definition at line 56 of file ioserver.F90.

  integer, volatile :: monc_registration_lock

mpi_type_data_sizing_description

integer io_server_mod::mpi_type_data_sizing_description

private

The MPI type for field sizing (i.e. array size etc send when MONCs register)

Definition at line 46 of file ioserver.F90.

  integer :: mpi_type_data_sizing_description, & !< The MPI type for field sizing (i.e. array size etc send when MONCs register)
       mpi_type_definition_description, & !< The MPI data type for data descriptions sent to MONCs
       mpi_type_field_description

mpi_type_definition_description

integer io_server_mod::mpi_type_definition_description

private

The MPI data type for data descriptions sent to MONCs.

Definition at line 46 of file ioserver.F90.

mpi_type_field_description

integer io_server_mod::mpi_type_field_description

private

The MPI data type for field descriptions sent to MONCs.

Definition at line 46 of file ioserver.F90.

registree_definition_descriptions

type(definition_description_type), dimension(:), allocatable io_server_mod::registree_definition_descriptions

private

Definition at line 54 of file ioserver.F90.

  type(definition_description_type), dimension(:), allocatable :: registree_definition_descriptions

registree_field_descriptions

type(field_description_type), dimension(:), allocatable io_server_mod::registree_field_descriptions

private

Definition at line 53 of file ioserver.F90.

  type(field_description_type), dimension(:), allocatable :: registree_field_descriptions