scale_process.F90

Go to the documentation of this file.

!-------------------------------------------------------------------------------
module scale_process
  !-----------------------------------------------------------------------------
  !
  !++ used modules
  !
  use mpi
  use gtool_file, only: &
     filecloseall
  use scale_precision
  use scale_stdio
  !-----------------------------------------------------------------------------
  implicit none
  private
  !-----------------------------------------------------------------------------
  !
  !++ Public procedure
  !
  public :: prc_mpistart
  public :: prc_universal_setup
  public :: prc_global_setup
  public :: prc_local_setup
  public :: prc_mpistop
  public :: prc_mpifinish
  public :: prc_mpisplit
  public :: prc_mpisplit_letkf

  public :: prc_mpibarrier
  public :: prc_mpitime
  public :: prc_mpitimestat

  !-----------------------------------------------------------------------------
  !
  !++ Public parameters & variables
  !
  !-----------------------------------------------------------------------------
  !                          [ communicator system ]
  !    MPI_COMM_WORLD
  !          |
  ! PRC_UNIVERSAL_COMM_WORLD --split--> BULK_COMM_WORLD
  !                                     |
  !                            PRC_GLOBAL_COMM_WORLD --split--> PRC_LOCAL_COMM_WORLD
  !-----------------------------------------------------------------------------
  integer, public, parameter :: prc_masterrank      = 0
  integer, public, parameter :: prc_domain_nlim = 10000

  ! universal world
  integer, public :: prc_universal_comm_world = -1
  integer, public :: prc_universal_myrank     = -1
  integer, public :: prc_universal_nprocs     = -1
  logical, public :: prc_universal_ismaster   = .false. 

  integer, public :: prc_universal_jobid      = 0

  ! global world
  integer, public :: prc_global_comm_world    = -1
  integer, public :: prc_global_myrank        = -1
  integer, public :: prc_global_nprocs        = -1
  logical, public :: prc_global_ismaster      = .false. 

  integer, public :: prc_global_domainid      = 0

  ! local world
  integer, public :: prc_local_comm_world     = -1
  integer, public :: prc_nprocs               = 1
  integer, public :: prc_myrank               = 0
  logical, public :: prc_ismaster             = .false. 

  ! error handling
  logical, public :: prc_mpi_alive = .false.            
  integer, public :: prc_universal_handler
  integer, public :: prc_abort_comm_world
  integer, public :: prc_abort_handler

  !-----------------------------------------------------------------------------
  !
  !++ Private procedure
  !
  private :: prc_mpicoloring
  private :: prc_sort_ascd

  !-----------------------------------------------------------------------------
  !
  !++ Private parameters & variables
  !
  integer, private, parameter :: prc_abort_code = -1
!  integer, private, parameter :: PRC_ABORT_code_p = 2 !< mpi abort code in error handler from parent
!  integer, private, parameter :: PRC_ABORT_code_d = 3 !< mpi abort code in error handler from daughter

  !-----------------------------------------------------------------------------
contains
  !-----------------------------------------------------------------------------
  subroutine prc_mpistart( &
       comm )
    implicit none

    integer, intent(out) :: comm ! communicator

    integer :: ierr
    !---------------------------------------------------------------------------

    call mpi_init(ierr)

    prc_mpi_alive = .true.
    prc_universal_handler = mpi_errhandler_null
    call mpi_comm_create_errhandler( prc_mpi_errorhandler, prc_universal_handler, ierr )

    comm = mpi_comm_world

    return
  end subroutine prc_mpistart

  !-----------------------------------------------------------------------------
  subroutine prc_universal_setup( &
       comm,    &
       nprocs,  &
       ismaster )
    implicit none

    integer, intent(in)  :: comm     ! communicator
    integer, intent(out) :: nprocs   ! number of procs in this communicator
    logical, intent(out) :: ismaster ! master process in this communicator?

    integer :: ierr
    !---------------------------------------------------------------------------

    prc_universal_comm_world = comm

    call mpi_comm_size(prc_universal_comm_world,prc_universal_nprocs,ierr)
    call mpi_comm_rank(prc_universal_comm_world,prc_universal_myrank,ierr)

    if ( prc_universal_myrank == prc_masterrank ) then
       prc_universal_ismaster = .true.
    else
       prc_universal_ismaster = .false.
    endif

    nprocs   = prc_universal_nprocs
    ismaster = prc_universal_ismaster



    prc_abort_comm_world = prc_universal_comm_world

    call mpi_comm_set_errhandler(prc_abort_comm_world,prc_universal_handler,ierr)
    call mpi_comm_get_errhandler(prc_abort_comm_world,prc_abort_handler    ,ierr)

    return
  end subroutine prc_universal_setup

  !-----------------------------------------------------------------------------
  subroutine prc_global_setup( &
       abortall, &
       comm      )
    implicit none

    logical, intent(in)  :: abortall ! abort all jobs?
    integer, intent(in)  :: comm     ! communicator

    integer :: ierr
    !---------------------------------------------------------------------------

    prc_global_comm_world = comm

    call mpi_comm_size(prc_global_comm_world,prc_global_nprocs,ierr)
    call mpi_comm_rank(prc_global_comm_world,prc_global_myrank,ierr)

    if ( prc_global_myrank == prc_masterrank ) then
       prc_global_ismaster = .true.
    else
       prc_global_ismaster = .false.
    endif

!    if ( .NOT. abortall ) then
!       PRC_ABORT_COMM_WORLD = PRC_GLOBAL_COMM_WORLD
!
!       call MPI_COMM_SET_ERRHANDLER(PRC_ABORT_COMM_WORLD,PRC_UNIVERSAL_handler,ierr)
!       call MPI_COMM_GET_ERRHANDLER(PRC_ABORT_COMM_WORLD,PRC_ABORT_handler    ,ierr)
!    endif

    return
  end subroutine prc_global_setup

  !-----------------------------------------------------------------------------
  subroutine prc_local_setup( &
       comm,    &
       myrank,  &
       ismaster )
    implicit none

    integer, intent(in)  :: comm     ! communicator
    integer, intent(out) :: myrank   ! myrank         in this communicator
    logical, intent(out) :: ismaster ! master process in this communicator?

    integer :: ierr
    !---------------------------------------------------------------------------

    prc_local_comm_world = comm

    call mpi_comm_rank(prc_local_comm_world,prc_myrank,ierr)
    call mpi_comm_size(prc_local_comm_world,prc_nprocs,ierr)

    if ( prc_myrank == prc_masterrank ) then
       prc_ismaster = .true.
    else
       prc_ismaster = .false.
    endif

    myrank   = prc_myrank
    ismaster = prc_ismaster

    return
  end subroutine prc_local_setup

  !-----------------------------------------------------------------------------
  subroutine prc_mpistop
    implicit none

    integer :: ierr
    !---------------------------------------------------------------------------

    if ( prc_mpi_alive ) then
       ! tentative approach; input "PRC_UNIVERSAL_COMM_WORLD".
       call mpi_comm_call_errhandler(prc_universal_comm_world,prc_abort_code,ierr)
    endif

    stop
  end subroutine prc_mpistop

  !-----------------------------------------------------------------------------
  subroutine prc_mpifinish
    implicit none

    integer :: ierr
    !---------------------------------------------------------------------------

    if (prc_universal_handler .NE. mpi_errhandler_null) then
        call mpi_errhandler_free(prc_universal_handler, ierr)
    endif
    if (prc_abort_handler .NE. mpi_errhandler_null) then
        call mpi_errhandler_free(prc_abort_handler, ierr)
    endif

    ! Stop MPI
    if ( prc_mpi_alive ) then
       if ( io_l ) then
          write(io_fid_log,*)
          write(io_fid_log,*) '++++++ Stop MPI'
          write(io_fid_log,*)
       endif

       ! free splitted communicator
       if ( prc_local_comm_world /= prc_global_comm_world ) then
          call mpi_comm_free(prc_local_comm_world,ierr)
       endif

       call mpi_barrier(prc_universal_comm_world,ierr)

       call mpi_finalize(ierr)
       if( io_l ) write(io_fid_log,*) '*** MPI is peacefully finalized'
    endif

    ! Close logfile, configfile
    if ( io_l ) then
       if( io_fid_log /= io_fid_stdout ) close(io_fid_log)
    endif
    close(io_fid_conf)

    ! Stop program
    stop
  end subroutine prc_mpifinish

  !-----------------------------------------------------------------------------
  subroutine prc_mpisplit( &
      ORG_COMM,         & ! [in ]
      NUM_DOMAIN,       & ! [in ]
      PRC_DOMAINS,      & ! [in ]
      CONF_FILES,       & ! [in ]
      LOG_SPLIT,        & ! [in ]
      bulk_split,       & ! [in ]
      color_reorder,    & ! [in ]
      INTRA_COMM,       & ! [out]
      inter_parent,     & ! [out]
      inter_child,      & ! [out]
      fname_local       ) ! [out]
    implicit none

    integer, intent(in)  :: ORG_COMM
    integer, intent(in)  :: NUM_DOMAIN
    integer, intent(in)  :: PRC_DOMAINS(:)
    character(len=H_LONG), intent(in) :: CONF_FILES(:)
    logical, intent(in)  :: LOG_SPLIT
    logical, intent(in)  :: bulk_split
    logical, intent(in)  :: color_reorder

    integer, intent(out) :: intra_comm
    integer, intent(out) :: inter_parent
    integer, intent(out) :: inter_child
    character(len=H_LONG), intent(out) :: fname_local

    integer :: PARENT_COL(prc_domain_nlim)        ! parent color number
    integer :: CHILD_COL(prc_domain_nlim)         ! child  color number
    integer :: PRC_ROOT(0:prc_domain_nlim)        ! root process in the color
    integer, allocatable :: COLOR_LIST(:)   ! member list in each color
    integer, allocatable :: KEY_LIST(:)     ! local process number in each color

    integer :: total_nmax
    integer :: ORG_myrank  ! my rank number in the original communicator
    integer :: ORG_nmax    ! total rank number in the original communicator

    logical :: do_create_p(prc_domain_nlim)
    logical :: do_create_c(prc_domain_nlim)
    logical :: reordering

    character(len=H_LONG) :: COL_FILE(0:prc_domain_nlim)
    character(len=4)      :: col_num

    integer :: i
    integer :: itag, ierr
    !---------------------------------------------------------------------------

    intra_comm       = org_comm
    inter_parent     = mpi_comm_null
    inter_child      = mpi_comm_null
    fname_local      = conf_files(1)

    if ( num_domain > 1 ) then ! multi domain run
       call mpi_comm_rank(org_comm,org_myrank,ierr)
       call mpi_comm_size(org_comm,org_nmax,  ierr)
       allocate ( color_list(0:org_nmax-1) )
       allocate ( key_list(0:org_nmax-1) )

       total_nmax = 0
       do i = 1, num_domain
          total_nmax = total_nmax + prc_domains(i)
       enddo
       if ( total_nmax /= org_nmax ) then
          if ( prc_universal_ismaster ) write (*,*) ""
          if ( prc_universal_ismaster ) write (*,*) "ERROR: MPI PROCESS NUMBER is INCONSISTENT"
          if ( prc_universal_ismaster ) write (*,*) "REQUESTED NPROCS = ", total_nmax, "  LAUNCHED NPROCS = ", org_nmax
          call prc_mpistop
       endif

       reordering = color_reorder
       if ( bulk_split ) then
          reordering = .false.
       endif
       call prc_mpicoloring( org_comm,     & ! [in ]
                             num_domain,   & ! [in ]
                             prc_domains,  & ! [in ]
                             conf_files,   & ! [in ]
                             reordering,   & ! [in ]
                             log_split,    & ! [in ]
                             color_list,   & ! [out]
                             prc_root,     & ! [out]
                             key_list,     & ! [out]
                             parent_col,   & ! [out]
                             child_col,    & ! [out]
                             col_file      ) ! [out]


       ! split comm_world
       call mpi_comm_split(org_comm,               &
                           color_list(org_myrank), &
                           key_list(org_myrank),   &
                           intra_comm, ierr)
       if ( bulk_split ) then
          write(col_num,'(I4.4)') color_list(org_myrank)
          fname_local = col_num
          prc_universal_jobid  = color_list(org_myrank)
       else
          fname_local = col_file(color_list(org_myrank))
       endif

       ! set parent-child relationship
       do_create_p(:) = .false.
       do_create_c(:) = .false.
       if ( .NOT. bulk_split ) then
          do i = 1, num_domain-1
             if ( prc_universal_ismaster ) write ( *, '(1X,A,I4)' ) "relationship: ", i
             if ( prc_universal_ismaster ) write ( *, '(1X,A,I4,A,I4)' ) &
                               "--- parent color = ", parent_col(i), "  child color = ", child_col(i)
             if ( color_list(org_myrank) == parent_col(i) ) then
                do_create_p(i) = .true.
             elseif ( color_list(org_myrank) == child_col(i) ) then
                do_create_c(i) = .true.
             endif
          enddo
       endif

       ! create inter-commnunicator
       inter_parent = mpi_comm_null
       inter_child  = mpi_comm_null
       if ( .NOT. bulk_split ) then
          do i = 1, num_domain-1
             itag = i*100
             if ( do_create_p(i) ) then ! as a parent
                call mpi_intercomm_create( intra_comm, prc_masterrank,           &
                                           org_comm,   prc_root(child_col(i)), &
                                           itag, inter_child,  ierr)
             elseif ( do_create_c(i) ) then ! as a child
                call mpi_intercomm_create( intra_comm, prc_masterrank,           &
                                           org_comm,   prc_root(parent_col(i)), &
                                           itag, inter_parent, ierr)
             endif
             call mpi_barrier(org_comm, ierr)
          enddo
       endif

       deallocate ( color_list, key_list )

    elseif ( num_domain == 1 ) then ! single domain run
       if ( prc_universal_ismaster ) write (*,*) "*** a single comunicator"
    else
       if ( prc_universal_ismaster ) write (*,*) "ERROR: REQUESTED DOMAIN NUMBER IS NOT ACCEPTABLE"
       call prc_mpistop
    endif

    return
  end subroutine prc_mpisplit

  !-----------------------------------------------------------------------------
  subroutine prc_mpisplit_letkf( &
      ORG_COMM,         & ! [in ]
      mem_np,           & ! [in ]
      nitmax,           & ! [in ]
      nprocs,           & ! [in ]
      proc2mem,         & ! [in ]
      INTRA_COMM        ) ! [out]
    implicit none

    integer, intent(in)  :: ORG_COMM
    integer, intent(in)  :: mem_np
    integer, intent(in)  :: nitmax
    integer, intent(in)  :: nprocs
    integer, intent(in)  :: proc2mem(2,nitmax,nprocs)
    integer, intent(out) :: intra_comm

    integer :: ORG_myrank  ! my rank number in the original communicator
    integer :: color, key
    integer :: ierr
    !---------------------------------------------------------------------------

    call mpi_comm_rank( org_comm, org_myrank, ierr)

    if ( proc2mem(1,1,org_myrank+1) >= 1 ) then
       color = proc2mem(1,1,org_myrank+1) - 1
       key = proc2mem(2,1,org_myrank+1)
    else
       color = mpi_undefined
       key = mpi_undefined
    end if

    call mpi_comm_split( org_comm,               &
                         color, &
                         key,   &
                         intra_comm, ierr )

    return
  end subroutine prc_mpisplit_letkf

  !-----------------------------------------------------------------------------
  subroutine prc_mpicoloring( &
      ORG_COMM,        & ! [in ]
      NUM_DOMAIN,      & ! [in ]
      PRC_DOMAINS,     & ! [in ]
      CONF_FILES,      & ! [in ]
      color_reorder,   & ! [in ]
      LOG_SPLIT,       & ! [in ]
      COLOR_LIST,      & ! [out]
      PRC_ROOT,        & ! [out]
      KEY_LIST,        & ! [out]
      PARENT_COL,      & ! [out]
      CHILD_COL,       & ! [out]
      COL_FILE         ) ! [out]
    implicit none

    integer, intent(in)  :: ORG_COMM
    integer, intent(in)  :: NUM_DOMAIN
    integer, intent(in)  :: PRC_DOMAINS(:)
    character(len=H_LONG), intent(in) :: CONF_FILES(:)
    logical, intent(in)  :: color_reorder
    logical, intent(in)  :: LOG_SPLIT
    integer, intent(out) :: COLOR_LIST(:)             ! member list in each color
    integer, intent(out) :: PRC_ROOT(0:prc_domain_nlim)     ! root process in each color
    integer, intent(out) :: KEY_LIST(:)               ! local process number in each color
    integer, intent(out) :: PARENT_COL(:)             ! parent color number
    integer, intent(out) :: CHILD_COL(:)              ! child  color number
    character(len=H_LONG), intent(out) :: COL_FILE(0:prc_domain_nlim) ! conf file in each color

    integer :: touch(prc_domain_nlim)
    integer :: PRC_ORDER(prc_domain_nlim)                   ! reordered number of process
    integer :: ORDER2DOM(prc_domain_nlim)                   ! get domain number by order number
    integer :: DOM2ORDER(prc_domain_nlim)                   ! get order number by domain number
    integer :: DOM2COL(prc_domain_nlim)                     ! get color number by domain number
    integer :: COL2DOM(0:prc_domain_nlim)                   ! get domain number by color number
    integer :: RO_PRC_DOMAINS(prc_domain_nlim)              ! reordered values
    integer :: RO_DOM2COL(prc_domain_nlim)                  ! reordered values
    integer :: RO_PARENT_COL(prc_domain_nlim)               ! reordered values
    integer :: RO_CHILD_COL(prc_domain_nlim)                ! reordered values
    character(len=H_LONG) :: RO_CONF_FILES(prc_domain_nlim) ! reordered values

    integer :: ORG_nmax   ! parent domain number
    integer :: id_parent  ! parent domain number
    integer :: id_child   ! child domain number
    integer :: dnum, nprc, order, key
    integer :: i, j
    integer :: ierr
    !---------------------------------------------------------------------------

    call mpi_comm_size(org_comm,org_nmax, ierr)

    if ( color_reorder ) then
       !--- make color order
       !    domain num is counted from 1
       !    color num  is counted from 0
       touch(:) = -1
       prc_order(:) = prc_domains(:)
       call prc_sort_ascd( prc_order(1:num_domain), 1, num_domain )

       do i = 1, num_domain
       do j = 1, num_domain
          if ( prc_domains(i) == prc_order(j) .AND. touch(j) < 0 ) then
             dom2col(i         ) = j - 1  ! domain_num --> color_num
             col2dom(dom2col(i)) = i      ! color_num  --> domain_num
             touch(j) = 1
             exit
          endif
       enddo
       enddo

       parent_col(:) = -1
       child_col(:)  = -1
       do i = 1, num_domain
          id_parent = i - 1
          id_child  = i + 1

          if ( 1 <= id_parent .AND. id_parent <= num_domain ) then
             parent_col(i) = dom2col(id_parent)
          endif
          if ( 1 <= id_child  .AND. id_child  <= num_domain ) then
             child_col(i) = dom2col(id_child)
          endif

          if ( prc_universal_ismaster .AND. log_split ) then
             write( *, '(1X,A,I2,1X,A,I2,2(2X,A,I2))' )  &
                  "DOMAIN: ", i, "MY_COL: ", dom2col(i), &
                  "PARENT: COL= ", parent_col(i), "CHILD: COL= ", child_col(i)
          endif
       enddo

       !--- reorder following color order
       do i = 1, num_domain
          dnum = col2dom(i-1)
          order2dom(i)      = dnum
          dom2order(dnum)   = i
          ro_prc_domains(i) = prc_domains(dnum)
          ro_dom2col(dnum)  = dom2col(dnum)
          ro_conf_files(i)  = conf_files(dnum)
          ro_parent_col(i)  = parent_col(dnum)
          ro_child_col(i)   = child_col(dnum)
       enddo

       !--- set relationship by ordering of relationship number
       parent_col(:)   = -1
       child_col(:)    = -1
       do i = 1, num_domain-1
          parent_col(i) = ro_parent_col( dom2order(i+1) ) ! from child to parent
          child_col(i)  = ro_child_col( dom2order(i)   ) ! from parent to child
       enddo

       do i = 1, num_domain
          if ( prc_universal_ismaster ) write ( *, * ) ""
          if ( prc_universal_ismaster ) write ( *, '(1X,A,I2,A,I5)' ) "ORDER (",i,") -> DOMAIN: ", order2dom(i)
          if ( prc_universal_ismaster ) write ( *, '(1X,A,I1,A,I5)' ) "NUM PRC_DOMAINS(",i,")  = ", ro_prc_domains(i)
          if ( prc_universal_ismaster ) write ( *, '(1X,A,I1,A,I3)' ) "MY COLOR(",i,") = ", ro_dom2col(order2dom(i))
          if ( prc_universal_ismaster ) write ( *, '(1X,A,I1,A,I3)' ) "PARENT COLOR(",i,") = ", ro_parent_col(i)
          if ( prc_universal_ismaster ) write ( *, '(1X,A,I1,A,I3)' ) "CHILD COLOR(",i,") = ", ro_child_col(i)
          if ( prc_universal_ismaster ) write ( *, '(1X,A,I1,A,A)'  ) "CONF_FILES(",i,")    = ", trim(ro_conf_files(i))
       enddo
       if ( prc_universal_ismaster ) write ( *, * ) ""

       do i=1, num_domain
          col_file(i-1) = ro_conf_files(i) ! final copy
       enddo

    else !--- without reordering of colors
       order2dom(:)      = -1
       ro_dom2col(:)     = -1
       ro_prc_domains(:) = -1
       ro_prc_domains(:) = -1
       ro_parent_col(:)  = -1
       ro_child_col(:)   = -1

       do i = 1, num_domain
          order2dom(i)      = i
          ro_dom2col(i)     = i-1
          ro_prc_domains(i) = prc_domains(i)
          ro_conf_files(i)  = conf_files(i)
       enddo

       do i = 1, num_domain
          id_parent = i - 1
          id_child  = i + 1

          if ( 1 <= id_parent .AND. id_parent <= num_domain ) then
             ro_parent_col(i) = ro_dom2col(id_parent)
          endif
          if ( 1 <= id_child  .AND. id_child  <= num_domain ) then
             ro_child_col(i) = ro_dom2col(id_child)
          endif
       enddo

       ! make relationship
       do i = 1, num_domain-1
          parent_col(i) = ro_parent_col(i+1) ! from child to parent
          child_col(i)  = ro_child_col(i  ) ! from parent to child
       enddo

    endif

    ! make a process table
    order = 1
    key   = 0
    nprc  = ro_prc_domains(order)
    prc_root(:) = -999

    do i = 0, org_nmax-1
       color_list(i+1) = ro_dom2col(order2dom(order))
       key_list(i+1)   = key
       if ( key == 0 ) then
          prc_root(color_list(i+1)) = i
          col_file(color_list(i+1)) = ro_conf_files(order)
       endif
       if ( log_split .AND. prc_universal_ismaster ) then
          write ( *, '(1X,4(A,I5))' ) "PE:", i, "   COLOR:", color_list(i+1), &
                "   KEY:", key_list(i+1), "   PRC_ROOT:", prc_root(color_list(i+1))
       endif
       key = key + 1
       if ( key >= nprc ) then
          order = order + 1
          key   = 0
          nprc  = ro_prc_domains(order)
       endif
    enddo

    return
  end subroutine prc_mpicoloring
  !-----------------------------------------------------------------------------

  !-----------------------------------------------------------------------------
  recursive subroutine prc_sort_ascd(a, top, bottom)
    implicit none
    integer, intent(inout) :: a(:)
    integer, intent(in)    :: top, bottom
    integer :: i, j, cnt, trg
    !---------------------------------------------------------------------------
    cnt = a( (top+bottom) / 2 )
    i = top; j = bottom
    do
       do while ( a(i) > cnt ) !ascending evaluation
          i = i + 1
       enddo
       do while ( cnt > a(j) ) !ascending evaluation
          j = j - 1
       enddo
       if ( i >= j ) exit
       trg = a(i);  a(i) = a(j);  a(j) = trg
       i = i + 1
       j = j - 1
    enddo
    if ( top < i-1    ) call prc_sort_ascd( a, top, i-1    )
    if ( j+1 < bottom ) call prc_sort_ascd( a, j+1, bottom )
    return
  end subroutine prc_sort_ascd
  !-----------------------------------------------------------------------------

  !-----------------------------------------------------------------------------
  subroutine prc_mpibarrier
    implicit none

    integer  :: ierr
    !---------------------------------------------------------------------------

    if ( prc_mpi_alive ) then
       call mpi_barrier(prc_local_comm_world,ierr)
    endif

  end subroutine prc_mpibarrier

  !-----------------------------------------------------------------------------
  function prc_mpitime() result(time)
    implicit none

    real(DP) :: time
    !---------------------------------------------------------------------------

    if ( prc_mpi_alive ) then
       time = real(MPI_WTIME(), kind=dp)
    else
       call cpu_time(time)
    endif

  end function prc_mpitime

  !-----------------------------------------------------------------------------
  subroutine prc_mpitimestat( &
      avgvar, &
      maxvar, &
      minvar, &
      maxidx, &
      minidx, &
      var     )
    implicit none

    real(DP), intent(out) :: avgvar(:)
    real(DP), intent(out) :: maxvar(:)
    real(DP), intent(out) :: minvar(:)
    integer,  intent(out) :: maxidx(:)
    integer,  intent(out) :: minidx(:)
    real(DP), intent(in)  :: var(:)

    real(DP), allocatable :: statval(:,:)
    integer               :: vsize

    real(DP) :: totalvar
    integer  :: ierr
    integer  :: v, p
    !---------------------------------------------------------------------------

    vsize = size(var(:))

    allocate( statval(vsize,0:prc_nprocs-1) )
    statval(:,:) = 0.0_dp

    do v = 1, vsize
       statval(v,prc_myrank) = var(v)
    enddo

    ! MPI broadcast
    do p = 0, prc_nprocs-1
       call mpi_bcast( statval(1,p),         &
                       vsize,                &
                       mpi_double_precision, &
                       p,                    &
                       prc_local_comm_world,     &
                       ierr                  )
    enddo

    do v = 1, vsize
       totalvar = 0.0_dp
       do p = 0, prc_nprocs-1
          totalvar = totalvar + statval(v,p)
       enddo
       avgvar(v) = totalvar / prc_nprocs

       maxvar(v)   = maxval(statval(v,:))
       minvar(v)   = minval(statval(v,:))
       maxidx(v:v) = maxloc(statval(v,:))
       minidx(v:v) = minloc(statval(v,:))
    enddo

    deallocate( statval )

    return
  end subroutine prc_mpitimestat

  !-----------------------------------------------------------------------------
  subroutine prc_mpi_errorhandler( &
      comm,     &
      errcode   )
    implicit none

    ! attributes are needed to be the same with COMM_ERRHANDLER function
    integer :: comm
    integer :: errcode

    character(len=MPI_MAX_ERROR_STRING) :: msg
    integer :: len
    integer :: ierr
    !---------------------------------------------------------------------------

    ! Print Error Messages
    if ( prc_mpi_alive ) then
          ! flush 1kbyte
       if ( io_l ) then
          write(io_fid_log,'(32A32)') '                                '
          write(io_fid_log,*) '++++++ Abort MPI'
          write(io_fid_log,*) ''
       end if

       if ( io_l ) then
          write(*,*) '++++++ BULK   ID: ',prc_universal_jobid
          write(*,*) '++++++ DOMAIN ID: ',prc_global_domainid
          write(*,*) '++++++ MASTER LOCATION: ',prc_universal_myrank,'/',prc_universal_nprocs
          write(*,*) '++++++ GLOBAL LOCATION: ',prc_global_myrank,'/',prc_global_nprocs
          write(*,*) '++++++ LOCAL  LOCATION: ',prc_myrank,'/',prc_nprocs
          write(*,*) ''
       end if

       if ( errcode == prc_abort_code ) then ! called from PRC_MPIstop
       elseif ( errcode <= mpi_err_lastcode ) then
          call mpi_error_string(errcode, msg, len, ierr)
          if ( io_l ) write(io_fid_log,*) '++++++ ', errcode, trim(msg)
          write(*,*) '++++++ ', errcode, trim(msg)
       else
          if ( io_l ) write(io_fid_log,*) '++++++ Unexpected error code', errcode
          write(*,*) '++++++ Unexpected error code', errcode
       endif

       if ( comm /= prc_abort_comm_world ) then
          if ( io_l ) write(io_fid_log,*) '++++++ Unexpected communicator'
          write(*,*) '++++++ Unexpected communicator'
       endif
       if ( io_l ) write(io_fid_log,*) ''
       write(*,*) ''
    endif

    call filecloseall

    ! Close logfile, configfile
    if ( io_l ) then
       if ( io_fid_log /= io_fid_stdout ) close(io_fid_log)
    endif
    close(io_fid_conf)

    ! Abort MPI
    if ( prc_mpi_alive ) then
       call sleep(5)
       call mpi_abort(prc_abort_comm_world, prc_abort_code, ierr)
    endif

    stop
  end subroutine prc_mpi_errorhandler

end module scale_process
!-------------------------------------------------------------------------------