mpicommunication.F90

Go to the documentation of this file.

module mpi_communication_mod
  use datadefn_mod, only : string_length
  use collections_mod, only : map_type, c_put_integer
  use configuration_parser_mod, only : io_configuration_data_definition_type, io_configuration_inter_communication_description
  use logging_mod, only : log_error, log_log, log_master_log
  use io_server_client_mod, only : array_field_type, map_field_type, integer_data_type, boolean_data_type, string_data_type, &
       float_data_type, double_data_type, command_tag, data_tag, inter_io_communication, get_data_description_from_name, &
       data_sizing_description_type, populate_mpi_type_extents, append_mpi_datatype, get_mpi_datatype_from_internal_representation
  use forthread_mod, only : forthread_mutex_lock, forthread_mutex_unlock, forthread_mutex_init, forthread_mutex_destroy
  use threadpool_mod, only : check_thread_status
  use mpi, only : mpi_comm_world, mpi_source, mpi_int, mpi_byte, mpi_status_size, mpi_request_null, &
       mpi_status_ignore, mpi_statuses_ignore, mpi_any_source, mpi_thread_multiple, mpi_thread_serialized
  use iso_c_binding
  implicit none

#ifndef TEST_MODE
  private
#endif

  integer, parameter :: ms_wait_between_tests=100

  
  interface 
     subroutine usleep(useconds) bind(C) 
       use iso_c_binding 
       implicit none 
       integer(c_int32_t), value :: useconds 
     end subroutine usleep
  end interface

  integer :: command_buffer,& !< Buffer used to receive the command data into when it arrives on that channel
       command_request_handle, & !< Request handle representing the asynchronous P2P command request
       mpi_threading_mode
  integer, volatile :: mpi_mutex
  logical :: manage_mpi_thread_safety

  public 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, initialise_mpi_communication, &
       lock_mpi, unlock_mpi, wait_for_mpi_request, waitall_for_mpi_requests, pause_for_mpi_interleaving
contains

  subroutine initialise_mpi_communication(provided_threading)
    integer, intent(in) :: provided_threading

    mpi_threading_mode=provided_threading
    if (mpi_threading_mode .ne. mpi_thread_multiple .and. mpi_threading_mode .ne. mpi_thread_serialized) then
      call log_master_log(log_error, "You must run MONC in MPI thread serialized or thread multiple mode for the IO server")
    end if    
    manage_mpi_thread_safety=provided_threading == mpi_thread_serialized
    call check_thread_status(forthread_mutex_init(mpi_mutex, -1))
  end subroutine initialise_mpi_communication

  subroutine lock_mpi()
    if (manage_mpi_thread_safety) call check_thread_status(forthread_mutex_lock(mpi_mutex))
  end subroutine lock_mpi

  subroutine unlock_mpi()
    if (manage_mpi_thread_safety) call check_thread_status(forthread_mutex_unlock(mpi_mutex))
  end subroutine unlock_mpi

  subroutine pause_for_mpi_interleaving()
    call usleep(int(ms_wait_between_tests, c_int32_t))
  end subroutine pause_for_mpi_interleaving  

  subroutine wait_for_mpi_request(request, status)
    integer, intent(inout) :: request
    integer, intent(inout), optional :: status(mpi_status_size)

    integer :: ierr, flag

    if (manage_mpi_thread_safety) then
      flag=0
      do while (flag .ne. 1)
        call lock_mpi()
        if (present(status)) then
          call mpi_test(request, flag, status, ierr)
        else
          call mpi_test(request, flag, mpi_status_ignore, ierr)
        end if
        call unlock_mpi()      
        if (flag .ne. 1) call pause_for_mpi_interleaving()
      end do
    else
      if (present(status)) then
        call mpi_wait(request, status, ierr)
      else
        call mpi_wait(request, mpi_status_ignore, ierr)
      end if
    end if
  end subroutine wait_for_mpi_request  

  subroutine waitall_for_mpi_requests(requests, count)
    integer, dimension(:), intent(inout) :: requests
    integer, intent(in) :: count

    integer :: ierr, flag

    if (manage_mpi_thread_safety) then
      flag=0
      do while (flag .ne. 1)
        call lock_mpi()
        call mpi_testall(count, requests, flag, mpi_statuses_ignore, ierr)
        call unlock_mpi()      
        if (flag .ne. 1) call pause_for_mpi_interleaving()
      end do
    else
      call mpi_waitall(count, requests, mpi_statuses_ignore, ierr)
    end if
  end subroutine waitall_for_mpi_requests 

  integer function get_number_io_servers(io_comm)
    integer, intent(in) :: io_comm

    integer :: number, ierr

    call mpi_comm_size(io_comm, number, ierr)
    get_number_io_servers=number
  end function get_number_io_servers

  integer function get_my_io_rank(io_comm)
    integer, intent(in) :: io_comm

    integer :: number, ierr

    call mpi_comm_rank(io_comm, number, ierr)
    get_my_io_rank=number
  end function get_my_io_rank  

  subroutine register_command_receive()
    integer :: ierr

    call lock_mpi()
    call mpi_irecv(command_buffer, 1, mpi_int, mpi_any_source, command_tag, &
         mpi_comm_world, command_request_handle, ierr)
    call unlock_mpi()
  end subroutine register_command_receive

  integer function data_receive(mpi_datatype, num_elements, source, dump_data, data_dump_id, description_data)
    integer, intent(in) :: mpi_datatype, num_elements, source
    integer, intent(in), optional :: data_dump_id
    character, dimension(:), allocatable, intent(inout), optional :: dump_data
    type(data_sizing_description_type), dimension(:), intent(inout), optional :: description_data
    integer :: ierr, request, status(mpi_status_size), recv_count, tag_to_use

    if (present(dump_data)) then
      tag_to_use=data_tag
      if (present(data_dump_id)) tag_to_use=tag_to_use+data_dump_id
      call lock_mpi()
      call mpi_irecv(dump_data, num_elements, mpi_datatype, source, tag_to_use, mpi_comm_world, request, ierr)
      call unlock_mpi()
      call wait_for_mpi_request(request, status)
      call lock_mpi()
      call mpi_get_count(status, mpi_datatype, recv_count, ierr)      
      call unlock_mpi()
      data_receive=recv_count
    else if (present(description_data)) then
      call lock_mpi()
      call mpi_irecv(description_data, num_elements, mpi_datatype, source, data_tag, mpi_comm_world, request, ierr)
      call unlock_mpi()
      call wait_for_mpi_request(request, status)
      call lock_mpi()
      call mpi_get_count(status, mpi_datatype, recv_count, ierr)      
      call unlock_mpi()
      data_receive=recv_count
    end if    
  end function data_receive

  subroutine cancel_requests()
    call cancel_request(command_request_handle)
  end subroutine cancel_requests  

  subroutine cancel_request(req)
    integer, intent(in) :: req

    integer :: ierr

    if (req .ne. mpi_request_null) then
      call lock_mpi()
      call mpi_cancel(req, ierr)
      call unlock_mpi()
    end if
  end subroutine cancel_request  

  logical function test_for_command(command, source)
    integer, intent(out) :: command, source

    integer :: ierr, status(mpi_status_size), complete

    call lock_mpi()
    call mpi_test(command_request_handle, complete, status, ierr)
    call unlock_mpi()

    if (complete .eq. 1) then
      command = command_buffer
      source = status(mpi_source)
      call register_command_receive()
      test_for_command=.true.
    else
      test_for_command=.false.
    end if    
  end function test_for_command

  logical function test_for_inter_io(inter_io_communications, number_of_inter_io, io_communicator, command, source, data_buffer)
    integer, intent(in) :: number_of_inter_io, io_communicator
    integer, intent(out) :: command, source
    type(io_configuration_inter_communication_description), dimension(:), intent(inout) :: inter_io_communications
    character, dimension(:), allocatable, intent(inout) :: data_buffer

    integer :: i, ierr, status(mpi_status_size), message_size
    logical :: message_pending

    call lock_mpi()
    do i=1, number_of_inter_io
      call mpi_iprobe(mpi_any_source, inter_io_communications(i)%message_tag, io_communicator, message_pending, status, ierr)
      if (message_pending) then
        call mpi_get_count(status, mpi_byte, message_size, ierr)
        allocate(data_buffer(message_size))
        call mpi_recv(data_buffer, message_size, mpi_byte, mpi_any_source, inter_io_communications(i)%message_tag, &
             io_communicator, mpi_status_ignore, ierr)
        call unlock_mpi()
        command=inter_io_communication
        source=i
        test_for_inter_io=.true.
        return
      end if
    end do
    call unlock_mpi()
    test_for_inter_io=.false.
  end function test_for_inter_io
  
  subroutine free_mpi_type(the_type)
    integer, intent(in) :: the_type

    integer :: ierr

    call mpi_type_free(the_type, ierr)
  end subroutine free_mpi_type  

  integer function build_mpi_datatype(data_definition, data_size_info, data_size, field_start_locations, &
       field_end_locations, field_dimensions)
    type(io_configuration_data_definition_type), intent(in) :: data_definition
    type(data_sizing_description_type), dimension(:), intent(in) :: data_size_info
    integer, intent(out) :: data_size
    type(map_type), intent(out) :: field_start_locations, field_end_locations
    type(map_type), intent(out), optional :: field_dimensions

    integer :: type_extents(5), type_counts, i, j, field_start, data_type, field_array_sizes, &
         temp_size, prev_data_type, old_types(20), offsets(20), block_counts(20), new_type, current_location, ierr, field_ignores
    logical :: field_found
    type(data_sizing_description_type) :: field_size_info

    type_extents=populate_mpi_type_extents()

    field_start=1
    data_type=0
    type_counts=0
    field_array_sizes=0
    field_ignores=0
    current_location=1
    do i=1,data_definition%number_of_data_fields
      if (data_type == 0) then        
        prev_data_type=data_type        
        data_type=data_definition%fields(i)%data_type
      else
        if (data_type .ne. data_definition%fields(i)%data_type) then
          ! For efficient type packing, combine multiple fields with the same type - therefore when the type changes work the previous one pack
          call append_mpi_datatype(field_start, i-1-field_ignores, field_array_sizes, data_type, &
               type_extents, prev_data_type, type_counts+1, old_types, offsets, block_counts)          
          field_start=i
          field_array_sizes=0
          field_ignores=0
          prev_data_type=data_type                   
          data_type=data_definition%fields(i)%data_type
          type_counts=type_counts+1
        end if
      end if
      call c_put_integer(field_start_locations, data_definition%fields(i)%name, current_location)
      if (data_definition%fields(i)%field_type .eq. array_field_type .or. &
           data_definition%fields(i)%field_type .eq. map_field_type) then
        ! Grab the field info based upon the field name
        field_found=get_data_description_from_name(data_size_info, data_definition%fields(i)%name, field_size_info)
        if (.not. field_found .or. field_size_info%dimensions == 0) then
          ! If no field info, or the dimension is 0 then this MONC process is not sending that field - check it is optional
          if (.not. data_definition%fields(i)%optional) then
            call log_log(log_error, "Non optional field `"//trim(data_definition%fields(i)%name)//&
                 "' omitted from MONC IO server registration")
          end if
          field_ignores=field_ignores+1
        else
          ! If the field is specified then use the size data to assemble the field size and append to current size
          temp_size=1
          do j=1, field_size_info%dimensions
            temp_size=temp_size*field_size_info%dim_sizes(j)
          end do          
          if (data_definition%fields(i)%field_type .eq. map_field_type) then
            field_array_sizes=(field_array_sizes+temp_size*string_length*2)-1
            current_location=current_location+temp_size*string_length*2
          else
            field_array_sizes=(field_array_sizes+temp_size)-1
            current_location=current_location+temp_size*type_extents(data_type)
          end if
        end if
      else
        if (data_definition%fields(i)%optional) then
          if (get_data_description_from_name(data_size_info, data_definition%fields(i)%name)) then
            if (data_type==string_data_type) then
              field_array_sizes=(field_array_sizes+string_length)-1
              current_location=current_location+type_extents(data_type)*string_length
            else
              current_location=current_location+type_extents(data_type)
            end if
          else
            field_ignores=field_ignores+1
          end if
        else
          if (data_type==string_data_type) then
            field_array_sizes=(field_array_sizes+string_length)-1
            current_location=current_location+type_extents(data_type)*string_length
          else
            current_location=current_location+type_extents(data_type)
          end if
        end if
      end if
      call c_put_integer(field_end_locations, data_definition%fields(i)%name, current_location-1)
      if (present(field_dimensions)) then
        call c_put_integer(field_dimensions, data_definition%fields(i)%name, field_size_info%dimensions)
      end if
    end do
    if (field_start .le. i-1) then
      ! If there are outstanding fields to process then we do this here
      call append_mpi_datatype(field_start, i-1, field_array_sizes, data_type, &
               type_extents, prev_data_type, type_counts+1, old_types, offsets, block_counts)
      type_counts=type_counts+1
    end if
    call lock_mpi()
    call mpi_type_struct(type_counts, block_counts, offsets, old_types, new_type, ierr) 
    call mpi_type_commit(new_type, ierr)
    call unlock_mpi()
    call mpi_type_size(new_type, data_size, ierr)
    build_mpi_datatype=new_type
  end function build_mpi_datatype
end module mpi_communication_mod