ioserver.F90

Go to the documentation of this file.

module io_server_mod
  use datadefn_mod, only : default_precision, string_length, long_string_length
  use configuration_parser_mod, only : data_size_stride, io_configuration_type, io_configuration_data_definition_type, &
       io_configuration_registered_monc_type, configuration_parse, extend_registered_moncs_array, retrieve_data_definition, &
       build_definition_description_type_from_configuration, build_field_description_type_from_configuration, get_monc_location, &
       get_io_xml
  use mpi_communication_mod, only : build_mpi_datatype, data_receive, test_for_command, register_command_receive, &
       cancel_requests, free_mpi_type, get_number_io_servers, get_my_io_rank, test_for_inter_io, lock_mpi, unlock_mpi, &
       waitall_for_mpi_requests, initialise_mpi_communication, pause_for_mpi_interleaving
  use diagnostic_federator_mod, only : initialise_diagnostic_federator, finalise_diagnostic_federator, &
       check_diagnostic_federator_for_completion, pass_fields_to_diagnostics_federator, determine_diagnostics_fields_available
  use writer_federator_mod, only : initialise_writer_federator, finalise_writer_federator, check_writer_for_trigger, &
       inform_writer_federator_fields_present, inform_writer_federator_time_point, provide_q_field_names_to_writer_federator
  use writer_field_manager_mod, only : initialise_writer_field_manager, finalise_writer_field_manager, &
       provide_monc_data_to_writer_federator
  use collections_mod, only : hashset_type, hashmap_type, map_type, iterator_type, c_get_integer, c_put_integer, c_is_empty, &
       c_remove, c_add_string, c_integer_at, c_free, c_get_iterator, c_has_next, c_next_mapentry
  use conversions_mod, only : conv_to_string
  use string_utils_mod, only : replace_character
  use optionsdatabase_mod, only : options_get_string
  use io_server_client_mod, only : register_command, deregister_command, inter_io_communication, data_command_start, data_tag, &
       local_sizes_key, local_start_points_key, local_end_points_key, number_q_indicies_key, scalar_field_type, &
       data_sizing_description_type, definition_description_type, field_description_type, build_mpi_type_data_sizing_description,&
       get_data_description_from_name, build_mpi_type_field_description, build_mpi_type_definition_description
  use forthread_mod, only : forthread_rwlock_rdlock, forthread_rwlock_wrlock, forthread_rwlock_tryrdlock, &
       forthread_rwlock_unlock, forthread_rwlock_init, forthread_rwlock_destroy, forthread_mutex_init, forthread_mutex_lock, &
       forthread_mutex_unlock, forthread_cond_wait, forthread_cond_signal, forthread_cond_init
  use threadpool_mod, only : threadpool_init, threadpool_finalise, threadpool_start_thread, check_thread_status, &
       threadpool_deactivate, threadpool_is_idle
  use global_callback_inter_io_mod, only : perform_global_callback
  use logging_mod, only : log_error, log_warn, log_log, initialise_logging
  use mpi, only : mpi_comm_world, mpi_statuses_ignore, mpi_byte
  use io_server_state_reader_mod, only : read_io_server_configuration
  implicit none

#ifndef TEST_MODE
  private
#endif  

  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
  type(io_configuration_type), volatile, save :: io_configuration
  logical, volatile :: contine_poll_messages, & !< Whether to continue waiting command messages from any MONC processes
       initialised_present_data
  logical, volatile :: contine_poll_interio_messages, 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

  public io_server_run
contains

  subroutine io_server_run(options_database, io_communicator_arg, &
       provided_threading, total_global_processes, continuation_run, io_configuration_file)
    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()
  end subroutine io_server_run

  logical function await_command(command, source, data_buffer)
    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    
  end function await_command

  subroutine termination_callback(io_configuration, values, field_name, timestep)
    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.
  end subroutine termination_callback  

  subroutine handle_command_message(command, source, data_buffer)
    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    
  end subroutine handle_command_message

  subroutine handle_inter_io_communication_command(arguments, data_buffer)
    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)
  end subroutine handle_inter_io_communication_command

  subroutine free_individual_registered_monc_aspects()
    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
  end subroutine free_individual_registered_monc_aspects  

  subroutine handle_deregistration_command(arguments, data_buffer)
    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))
  end subroutine handle_deregistration_command

  subroutine pull_back_data_message_and_handle(source, data_set)
    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)
  end subroutine pull_back_data_message_and_handle  

  subroutine handle_data_message(arguments, data_buffer)
    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))
  end subroutine handle_data_message

  subroutine handle_monc_registration(arguments, data_buffer)
    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))
  end subroutine handle_monc_registration

  function send_configuration_to_registree(source)
    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    
  end function send_configuration_to_registree  

  subroutine init_data_definition(source, monc_defn)
    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)
  end subroutine init_data_definition

  subroutine get_monc_information_data(source)
    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)
  end subroutine get_monc_information_data  

  subroutine register_present_field_names_to_federators(data_description, recv_count)
    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)
  end subroutine register_present_field_names_to_federators  

  subroutine handle_monc_dimension_information(data_description, monc_defn)
    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
  end subroutine handle_monc_dimension_information
end module io_server_mod