source: NEMO/branches/2021/dev_r14312_MPI_Interface/src/OCE/LBC/lbc_lnk_pt2pt_generic.h90 @ 14379

  
#if ! defined BLOCK_ISEND && ! defined BLOCK_FILL
   SUBROUTINE lbc_lnk_pt2pt_/**/PRECISION( cdname, ptab, cd_nat, psgn, kfld, kfillmode, pfillval, khls, lsend, lrecv, ld4only )
      CHARACTER(len=*)              , INTENT(in   ) ::   cdname      ! name of the calling subroutine
      TYPE(PTR_4d_/**/PRECISION),  DIMENSION(:), INTENT(inout) ::   ptab        ! pointer of arrays on which apply the b.c.
      CHARACTER(len=1), DIMENSION(:), INTENT(in   ) ::   cd_nat      ! nature of array grid-points
      REAL(PRECISION),  DIMENSION(:), INTENT(in   ) ::   psgn        ! sign used across the north fold boundary
      INTEGER                       , INTENT(in   ) ::   kfld        ! number of pt3d arrays
      INTEGER ,             OPTIONAL, INTENT(in   ) ::   kfillmode   ! filling method for halo over land (default = constant)
      REAL(PRECISION),      OPTIONAL, INTENT(in   ) ::   pfillval    ! background value (used at closed boundaries)
      INTEGER ,             OPTIONAL, INTENT(in   ) ::   khls        ! halo size, default = nn_hls
      LOGICAL, DIMENSION(4),OPTIONAL, INTENT(in   ) ::   lsend, lrecv  ! communication with other 4 proc
      LOGICAL,              OPTIONAL, INTENT(in   ) ::   ld4only     ! if .T., do only 4-neighbour comm (ignore corners)
      !
      INTEGER  ::     ji,   jj,  jk,  jl,  jf, jn     ! dummy loop indices
      INTEGER  ::    ipi,  ipj, ipk, ipl, ipf         ! dimension of the input array
      INTEGER  ::   ip0i, ip1i, im0i, im1i
      INTEGER  ::   ip0j, ip1j, im0j, im1j
      INTEGER  ::   ishti, ishtj, ishti2, ishtj2
      INTEGER  ::   ifill_nfd, icomm, ierr
      INTEGER  ::   ihls, idxs, idxr, iszS, iszR
      INTEGER, DIMENSION(4)  ::   iwewe, issnn
      INTEGER, DIMENSION(8)  ::   isizei, ishtSi, ishtRi, ishtPi
      INTEGER, DIMENSION(8)  ::   isizej, ishtSj, ishtRj, ishtPj
      INTEGER, DIMENSION(8)  ::   ifill, iszall, ishtS, ishtR
      INTEGER, DIMENSION(8)  ::   ireq             ! mpi_request id
      INTEGER, DIMENSION(8)  ::   iStag, iRtag     ! Send and Recv mpi_tag id
      REAL(PRECISION) ::   zland
      LOGICAL, DIMENSION(8)  ::   llsend, llrecv
      LOGICAL  ::   ll4only                                        ! default: 8 neighbourgs
      LOGICAL  ::   ll_IdoNFold
      !!----------------------------------------------------------------------
      !
      ! ----------------------------------------- !
      !     1. local variables initialization     !
      ! ----------------------------------------- !
      !
      ipi = SIZE(ptab(1)%pt4d,1)
      ipj = SIZE(ptab(1)%pt4d,2)
      ipk = SIZE(ptab(1)%pt4d,3)
      ipl = SIZE(ptab(1)%pt4d,4)
      ipf = kfld
      !
      IF( narea == 1 .AND. numcom == -1 ) CALL mpp_report( cdname, ipk, ipl, ipf, ld_lbc = .TRUE. )
      !
      idxs = 1   ! initalize index for send buffer
      idxr = 1   ! initalize index for recv buffer
      icomm = mpi_comm_oce        ! shorter name
      !
      ! take care of optional parameters
      !
      ihls = nn_hls   ! default definition
      IF( PRESENT( khls ) )   ihls = khls
      IF( ihls > n_hlsmax ) THEN
         WRITE(ctmp1,*) TRIM(cdname), ' is calling lbc_lnk with khls > n_hlsmax : ', khls, '>', n_hlsmax
         CALL ctl_stop( 'STOP', ctmp1 )
      ENDIF
      IF( ipi /= Ni_0+2*ihls ) THEN
         WRITE(ctmp1,*) TRIM(cdname), ' is calling lbc_lnk with an input array which does not match ihls along i: ', ipi, ihls, Ni_0
         CALL ctl_stop( 'STOP', ctmp1 )
      ENDIF
      IF( ipj /= Nj_0+2*ihls ) THEN
         WRITE(ctmp1,*) TRIM(cdname), ' is calling lbc_lnk with an input array which does not match ihls along j:', ipj, ihls , Nj_0
         CALL ctl_stop( 'STOP', ctmp1 )
      ENDIF
      !
      ll4only = .FALSE.    ! default definition
      IF( PRESENT(ld4only) )   ll4only = ld4only
      !
      zland = 0._wp                                     ! land filling value: zero by default
      IF( PRESENT( pfillval ) )   zland = pfillval      ! set land value
      !
      ! define llsend and llrecv: logicals which say if mpi-neibourgs for send or receive exist or not.
      IF     ( PRESENT(lsend) .AND. PRESENT(lrecv) ) THEN   ! localy defined neighbourgs 
         llsend(1:4) = lsend(1:4)   ;   llrecv(1:4) = lrecv(1:4)
      ELSE IF( PRESENT(lsend) .OR.  PRESENT(lrecv) ) THEN
         WRITE(ctmp1,*) TRIM(cdname), ' is calling lbc_lnk with only one of the two arguments lsend or lrecv'
         CALL ctl_stop( 'STOP', ctmp1 )
      ELSE                                              ! default neighbours
         llsend(:) = mpiSnei(ihls,:) >= 0
         IF( ll4only )   llsend(5:8) = .FALSE.          ! exclude corners
         llrecv(:) = mpiRnei(ihls,:) >= 0
         IF( ll4only )   llrecv(5:8) = .FALSE.          ! exclude corners
      ENDIF
      !
      ! define ifill: which method should be used to fill each parts (sides+corners) of the halos
      ! default definition
      DO jn = 1, 4
         IF(             llrecv(jn) ) THEN   ;   ifill(jn) = jpfillmpi    ! with an mpi communication
         ELSEIF(    l_SelfPerio(jn) ) THEN   ;   ifill(jn) = jpfillperio  ! with self-periodicity
         ELSEIF( PRESENT(kfillmode) ) THEN   ;   ifill(jn) = kfillmode    ! localy defined
         ELSE                                ;   ifill(jn) = jpfillcst    ! constant value (zland)
         ENDIF
      END DO
      DO jn = 5, 8
         IF(             llrecv(jn) ) THEN   ;   ifill(jn) = jpfillmpi    ! with an mpi communication
         ELSE                                ;   ifill(jn) = jpfillnothing! do nothing
         ENDIF
      END DO
         !
      ! north fold treatment
      ll_IdoNFold = l_IdoNFold .AND. ifill(jpno) /= jpfillnothing
      IF( ll_IdoNFold ) THEN
         ifill_nfd = ifill(jpno)             ! if we are here, this means llrecv(jpno) = .false. and l_SelfPerio(jpno) = .false.
         ifill( (/jpno/) ) = jpfillnothing   ! we do north fold -> do nothing for northern halo
      ENDIF
      
      ! We first define the localization and size of the parts of the array that will be sent (s), received (r)
      ! or used for periodocity (p). The localization is defined as "the bottom left corner - 1" in i and j directions.
      ! This is a shift that will be applied later in the do loops to pick-up the appropriate part of the array
      !
      ! all definitions bellow do not refer to N[ij][se]0 so we can use it with any local value of ihls
      !                   !                       ________________________
      ip0i =          0   !          im0j = inner |__|__|__________|__|__|
      ip1i =       ihls   !   im1j = inner - halo |__|__|__________|__|__|
      im1i = ipi-2*ihls   !                       |  |  |          |  |  |
      im0i = ipi - ihls   !                       |  |  |          |  |  |
      ip0j =          0   !                       |  |  |          |  |  |
      ip1j =       ihls   !                       |__|__|__________|__|__|
      im1j = ipj-2*ihls   !           ip1j = halo |__|__|__________|__|__|
      im0j = ipj - ihls   !              ip0j = 0 |__|__|__________|__|__|
      !                   !                    ip0i ip1i        im1i im0i
      !
      iwewe(:) = (/ jpwe,jpea,jpwe,jpea /)   ;   issnn(:) = (/ jpso,jpso,jpno,jpno /)
      !cd     sides:     west  east south north      ;   corners: so-we, so-ea, no-we, no-ea
      isizei(1:4) = (/ ihls, ihls,  ipi,  ipi /)   ;   isizei(5:8) = ihls              ! i- count
      isizej(1:4) = (/  ipj,  ipj, ihls, ihls /)   ;   isizej(5:8) = ihls              ! j- count
      ishtSi(1:4) = (/ ip1i, im1i, ip0i, ip0i /)   ;   ishtSi(5:8) = ishtSi( iwewe )   ! i- shift send data
      ishtSj(1:4) = (/ ip0j, ip0j, ip1j, im1j /)   ;   ishtSj(5:8) = ishtSj( issnn )   ! j- shift send data
      ishtRi(1:4) = (/ ip0i, im0i, ip0i, ip0i /)   ;   ishtRi(5:8) = ishtRi( iwewe )   ! i- shift received data location
      ishtRj(1:4) = (/ ip0j, ip0j, ip0j, im0j /)   ;   ishtRj(5:8) = ishtRj( issnn )   ! j- shift received data location
      ishtPi(1:4) = (/ im1i, ip1i, ip0i, ip0i /)   ;   ishtPi(5:8) = ishtPi( iwewe )   ! i- shift data used for periodicity
      ishtPj(1:4) = (/ ip0j, ip0j, im1j, ip1j /)   ;   ishtPj(5:8) = ishtPj( issnn )   ! j- shift data used for periodicity
      !
      ! -------------------------------- !
      !     2. Prepare MPI exchanges     !
      ! -------------------------------- !
      !
      iStag = (/ 1, 2, 3, 4, 5, 6, 7, 8 /)   ! any value but each one must be different
      ! define iRtag with the corresponding iStag, e.g. data received at west where sent at east.
      iRtag(jpwe) = iStag(jpea)   ;   iRtag(jpea) = iStag(jpwe)   ;   iRtag(jpso) = iStag(jpno)   ;   iRtag(jpno) = iStag(jpso)
      iRtag(jpsw) = iStag(jpne)   ;   iRtag(jpse) = iStag(jpnw)   ;   iRtag(jpnw) = iStag(jpse)   ;   iRtag(jpne) = iStag(jpsw)
      !
      iszall(:) = isizei(:) * isizej(:) * ipk * ipl * ipf
      ishtS(1) = 0
      DO jn = 2, 8
         ishtS(jn) = ishtS(jn-1) + iszall(jn-1) * COUNT( (/llsend(jn-1)/) )
      END DO
      ishtR(1) = 0
      DO jn = 2, 8
         ishtR(jn) = ishtR(jn-1) + iszall(jn-1) * COUNT( (/llrecv(jn-1)/) )
      END DO

      ! Allocate buffer arrays to be sent/received if needed
      iszS = SUM(iszall, mask = llsend)                             ! send buffer size
      IF( ALLOCATED(BUFFSND) ) THEN
         CALL mpi_waitall(8, nreq_p2p, MPI_STATUSES_IGNORE, ierr)   ! wait for Isend from the PREVIOUS call
         IF( SIZE(BUFFSND) < iszS )    DEALLOCATE(BUFFSND)          ! send buffer is too small
      ENDIF
      IF( .NOT. ALLOCATED(BUFFSND) )   ALLOCATE( BUFFSND(iszS) )
      iszR = SUM(iszall, mask = llrecv)                             ! recv buffer size
      IF( ALLOCATED(BUFFRCV) ) THEN
         IF( SIZE(BUFFRCV) < iszR )    DEALLOCATE(BUFFRCV)          ! recv buffer is too small
      ENDIF
      IF( .NOT. ALLOCATED(BUFFRCV) )   ALLOCATE( BUFFRCV(iszR) )
      !
      ! default definition when no communication is done. understood by mpi_waitall
      nreq_p2p(:) = MPI_REQUEST_NULL   ! WARNING: Must be done after the call to mpi_waitall just above
      !
      ! ----------------------------------------------- !
      !     3. Do east and west MPI_Isend if needed     !
      ! ----------------------------------------------- !
      !
      DO jn = 1, 2
#define BLOCK_ISEND
#  include "lbc_lnk_pt2pt_generic.h90"         
#undef BLOCK_ISEND
      END DO
      !
      ! ----------------------------------- !
      !     4. Fill east and west halos     !
      ! ----------------------------------- !
      !
      DO jn = 1, 2
#define BLOCK_FILL
#  include "lbc_lnk_pt2pt_generic.h90"         
#undef BLOCK_FILL
      END DO
      !
      ! ------------------------------------------------- !
      !     5. Do north and south MPI_Isend if needed     !
      ! ------------------------------------------------- !
      !
      DO jn = 3, 4
#define BLOCK_ISEND
#  include "lbc_lnk_pt2pt_generic.h90"         
#undef BLOCK_ISEND
      END DO
      !
      ! ------------------------------- !
      !     6. north fold treatment     !
      ! ------------------------------- !
      !
      ! Must be done after receiving data from East/West neighbourgs (as it is coded in mpp_nfd, to be changed one day...)
      ! Do it after MPI_iSend to south/north neighbourgs so they won't wait (too much) to receive their data
      ! Do if before MPI_Recv from south/north neighbourgs so we have more time to receive data
      !
      IF( ll_IdoNFold ) THEN
         IF( jpni == 1 )  THEN   ;   CALL lbc_nfd( ptab, cd_nat, psgn                  , ihls, ipf )   ! self NFold
         ELSE                    ;   CALL mpp_nfd( ptab, cd_nat, psgn, ifill_nfd, zland, ihls, ipf )   ! mpi  NFold
         ENDIF
      ENDIF
      !
      ! ------------------------------------- !
      !     7. Fill south and north halos     !
      ! ------------------------------------- !
      !
      DO jn = 3, 4
#define BLOCK_FILL
#  include "lbc_lnk_pt2pt_generic.h90"         
#undef BLOCK_FILL
      END DO
      !
      ! ----------------------------------------------- !
      !     8. Specific problem in corner treatment     !
      !              ( very rate case... )              !
      ! ----------------------------------------------- !
      !
      DO jn = 5, 8
#define BLOCK_ISEND
#  include "lbc_lnk_pt2pt_generic.h90"         
#undef BLOCK_ISEND
      END DO
      DO jn = 5, 8
#define BLOCK_FILL
#  include "lbc_lnk_pt2pt_generic.h90"         
#undef BLOCK_FILL
      END DO
      !
      ! -------------------------------------------- !
      !     9. deallocate local temporary arrays     !
      !        if they areg larger than jpi*jpj      !  <- arbitrary max size...
      ! -------------------------------------------- !
      !
      IF( iszR > jpi*jpj )   DEALLOCATE(BUFFRCV)                    ! blocking receive -> can directly deallocate
      IF( iszS > jpi*jpj ) THEN
         CALL mpi_waitall(8, nreq_p2p, MPI_STATUSES_IGNORE, ierr)   ! must wait before deallocate send buffer
         DEALLOCATE(BUFFSND)
      ENDIF
      !
   END SUBROUTINE lbc_lnk_pt2pt_/**/PRECISION
#endif

#if defined BLOCK_ISEND
   IF( llsend(jn) ) THEN
      ishti = ishtSi(jn)
      ishtj = ishtSj(jn)
      DO jf = 1, ipf  ;  DO jl = 1, ipl  ;  DO jk = 1, ipk  ;  DO jj = 1,isizej(jn)  ;  DO ji = 1,isizei(jn)
         BUFFSND(idxs) = ptab(jf)%pt4d(ishti+ji,ishtj+jj,jk,jl)
         idxs = idxs + 1
      END DO   ;   END DO   ;   END DO   ;   END DO   ;   END DO
#if ! defined key_mpi_off
      IF( ln_timing ) CALL tic_tac(.TRUE.)
      ! non-blocking send of the west/east side using local buffer
      CALL MPI_ISEND( BUFFSND(ishtS(jn)+1), iszall(jn), MPI_TYPE, mpiSnei(ihls,jn), iStag(jn), icomm, nreq_p2p(jn), ierr )
      IF( ln_timing ) CALL tic_tac(.FALSE.)
#endif
   ENDIF
#endif

#if defined BLOCK_FILL
   ishti = ishtRi(jn)
   ishtj = ishtRj(jn)
   SELECT CASE ( ifill(jn) )
   CASE ( jpfillnothing )               ! no filling 
   CASE ( jpfillmpi   )                 ! fill with data received by MPI
#if ! defined key_mpi_off
      IF( ln_timing ) CALL tic_tac(.TRUE.)
      !                                 ! blocking receive of the west/east halo in local temporary arrays
      CALL MPI_RECV( BUFFRCV(ishtR(jn)+1), iszall(jn), MPI_TYPE, mpiRnei(ihls,jn), iRtag(jn), icomm, MPI_STATUS_IGNORE, ierr )
      IF( ln_timing ) CALL tic_tac(.FALSE.)
#endif
      DO jf = 1, ipf  ;  DO jl = 1, ipl  ;  DO jk = 1, ipk  ;  DO jj = 1,isizej(jn)  ;  DO ji = 1,isizei(jn)
         ptab(jf)%pt4d(ishti+ji,ishtj+jj,jk,jl) = BUFFRCV(idxr)
         idxr = idxr + 1
      END DO   ;   END DO   ;   END DO   ;   END DO   ;   END DO
   CASE ( jpfillperio )                 ! use periodicity
      ishti2 = ishtPi(jn)
      ishtj2 = ishtPj(jn)
      DO jf = 1, ipf  ;  DO jl = 1, ipl  ;  DO jk = 1, ipk  ;  DO jj = 1,isizej(jn)  ;  DO ji = 1,isizei(jn)
         ptab(jf)%pt4d(ishti+ji,ishtj+jj,jk,jl) = ptab(jf)%pt4d(ishti2+ji,ishtj2+jj,jk,jl)
      END DO   ;   END DO   ;   END DO   ;   END DO   ;   END DO
   CASE ( jpfillcopy  )                 ! filling with inner domain values
      ishti2 = ishtSi(jn)
      ishtj2 = ishtSj(jn)
      DO jf = 1, ipf  ;  DO jl = 1, ipl  ;  DO jk = 1, ipk  ;  DO jj = 1,isizej(jn)  ;  DO ji = 1,isizei(jn)
         ptab(jf)%pt4d(ishti+ji,ishtj+jj,jk,jl) = ptab(jf)%pt4d(ishti2+ji,ishtj2+jj,jk,jl)
      END DO   ;   END DO   ;   END DO   ;   END DO   ;   END DO
   CASE ( jpfillcst   )                 ! filling with constant value
      DO jf = 1, ipf  ;  DO jl = 1, ipl  ;  DO jk = 1, ipk  ;  DO jj = 1,isizej(jn)  ;  DO ji = 1,isizei(jn)
         ptab(jf)%pt4d(ishti+ji,ishtj+jj,jk,jl) = zland
      END DO   ;   END DO   ;   END DO   ;   END DO   ;   END DO
   END SELECT
#endif