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

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 140 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:

◆ 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 224 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 442 of file ioserver.F90.

     integer, intent(in) :: source
 
     character, dimension(:), allocatable :: buffer
     character(len=STRING_LENGTH) :: q_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) + (string_length * num_q_fields)
     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))
         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
       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 192 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 296 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 246 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 210 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 517 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 338 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 407 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 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 270 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 486 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 384 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 180 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 50 of file ioserver.F90.

◆ contine_poll_interio_messages

logical, volatile io_server_mod::contine_poll_interio_messages

private

Definition at line 50 of file ioserver.F90.

50 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 48 of file ioserver.F90.

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

49 initialised_present_data

◆ initialised_present_data

logical, volatile io_server_mod::initialised_present_data

private

Definition at line 48 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 47 of file ioserver.F90.

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

◆ monc_registration_lock

integer, volatile io_server_mod::monc_registration_lock

private

Definition at line 54 of file ioserver.F90.

54 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 44 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 44 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 44 of file ioserver.F90.

◆ registree_definition_descriptions

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

private

Definition at line 52 of file ioserver.F90.

52 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 51 of file ioserver.F90.

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

Functions/Subroutines

Variables

Detailed Description

Function/Subroutine Documentation

◆ await_command()

◆ free_individual_registered_monc_aspects()

◆ get_monc_information_data()

◆ handle_command_message()

◆ handle_data_message()

◆ handle_deregistration_command()

◆ handle_inter_io_communication_command()

◆ handle_monc_dimension_information()

◆ handle_monc_registration()

◆ init_data_definition()

◆ io_server_run()

◆ pull_back_data_message_and_handle()

◆ register_present_field_names_to_federators()

◆ send_configuration_to_registree()

◆ termination_callback()

Variable Documentation

◆ already_registered_finishing_call

◆ contine_poll_interio_messages

◆ contine_poll_messages

◆ initialised_present_data

◆ io_configuration

◆ monc_registration_lock

◆ mpi_type_data_sizing_description

◆ mpi_type_definition_description

◆ mpi_type_field_description

◆ registree_definition_descriptions

◆ registree_field_descriptions