New URL for NEMO forge!   http://forge.nemo-ocean.eu

Since March 2022 along with NEMO 4.2 release, the code development moved to a self-hosted GitLab.
This present forge is now archived and remained online for history.
lib_mpp.F90 in branches/2012/dev_NOC_2012_rev3555/NEMOGCM/NEMO/OPA_SRC/LBC – NEMO

source: branches/2012/dev_NOC_2012_rev3555/NEMOGCM/NEMO/OPA_SRC/LBC/lib_mpp.F90 @ 4146

Last change on this file since 4146 was 3632, checked in by acc, 12 years ago

Branch dev_NOC_2012_r3555. #1006. Step 9: Merge in trunk changes between revision 3385 and 3452

  • Property svn:keywords set to Id
File size: 129.7 KB
RevLine 
[3]1MODULE lib_mpp
[13]2   !!======================================================================
3   !!                       ***  MODULE  lib_mpp  ***
[1344]4   !! Ocean numerics:  massively parallel processing library
[13]5   !!=====================================================================
[1344]6   !! History :  OPA  !  1994  (M. Guyon, J. Escobar, M. Imbard)  Original code
7   !!            7.0  !  1997  (A.M. Treguier)  SHMEM additions
8   !!            8.0  !  1998  (M. Imbard, J. Escobar, L. Colombet ) SHMEM and MPI
9   !!                 !  1998  (J.M. Molines) Open boundary conditions
10   !!   NEMO     1.0  !  2003  (J.-M. Molines, G. Madec)  F90, free form
11   !!                 !  2003  (J.M. Molines) add mpp_ini_north(_3d,_2d)
12   !!             -   !  2004  (R. Bourdalle Badie)  isend option in mpi
13   !!                 !  2004  (J.M. Molines) minloc, maxloc
14   !!             -   !  2005  (G. Madec, S. Masson)  npolj=5,6 F-point & ice cases
15   !!             -   !  2005  (R. Redler) Replacement of MPI_COMM_WORLD except for MPI_Abort
16   !!             -   !  2005  (R. Benshila, G. Madec)  add extra halo case
17   !!             -   !  2008  (R. Benshila) add mpp_ini_ice
18   !!            3.2  !  2009  (R. Benshila) SHMEM suppression, north fold in lbc_nfd
[1345]19   !!            3.2  !  2009  (O. Marti)    add mpp_ini_znl
[2715]20   !!            4.0  !  2011  (G. Madec)  move ctl_ routines from in_out_manager
[13]21   !!----------------------------------------------------------------------
[2715]22
23   !!----------------------------------------------------------------------
24   !!   ctl_stop   : update momentum and tracer Kz from a tke scheme
25   !!   ctl_warn   : initialization, namelist read, and parameters control
26   !!   ctl_opn    : Open file and check if required file is available.
27   !!   get_unit    : give the index of an unused logical unit
28   !!----------------------------------------------------------------------
[1344]29#if   defined key_mpp_mpi 
[13]30   !!----------------------------------------------------------------------
[1344]31   !!   'key_mpp_mpi'             MPI massively parallel processing library
32   !!----------------------------------------------------------------------
[2715]33   !!   lib_mpp_alloc : allocate mpp arrays
34   !!   mynode        : indentify the processor unit
35   !!   mpp_lnk       : interface (defined in lbclnk) for message passing of 2d or 3d arrays (mpp_lnk_2d, mpp_lnk_3d)
[473]36   !!   mpp_lnk_3d_gather :  Message passing manadgement for two 3D arrays
[2715]37   !!   mpp_lnk_e     : interface (defined in lbclnk) for message passing of 2d array with extra halo (mpp_lnk_2d_e)
38   !!   mpprecv         :
39   !!   mppsend       :   SUBROUTINE mpp_ini_znl
40   !!   mppscatter    :
41   !!   mppgather     :
42   !!   mpp_min       : generic interface for mppmin_int , mppmin_a_int , mppmin_real, mppmin_a_real
43   !!   mpp_max       : generic interface for mppmax_int , mppmax_a_int , mppmax_real, mppmax_a_real
44   !!   mpp_sum       : generic interface for mppsum_int , mppsum_a_int , mppsum_real, mppsum_a_real
45   !!   mpp_minloc    :
46   !!   mpp_maxloc    :
47   !!   mppsync       :
48   !!   mppstop       :
49   !!   mppobc        : variant of mpp_lnk for open boundary condition
[1344]50   !!   mpp_ini_north : initialisation of north fold
51   !!   mpp_lbc_north : north fold processors gathering
52   !!   mpp_lbc_north_e : variant of mpp_lbc_north for extra outer halo
[13]53   !!----------------------------------------------------------------------
[2715]54   USE dom_oce        ! ocean space and time domain
55   USE lbcnfd         ! north fold treatment
56   USE in_out_manager ! I/O manager
[3]57
[13]58   IMPLICIT NONE
[415]59   PRIVATE
[1344]60   
[2715]61   PUBLIC   ctl_stop, ctl_warn, get_unit, ctl_opn
[1344]62   PUBLIC   mynode, mppstop, mppsync, mpp_comm_free
63   PUBLIC   mpp_ini_north, mpp_lbc_north, mpp_lbc_north_e
64   PUBLIC   mpp_min, mpp_max, mpp_sum, mpp_minloc, mpp_maxloc
65   PUBLIC   mpp_lnk_3d, mpp_lnk_3d_gather, mpp_lnk_2d, mpp_lnk_2d_e
[3294]66   PUBLIC   mppscatter, mppgather
[1528]67   PUBLIC   mppobc, mpp_ini_ice, mpp_ini_znl
[2715]68   PUBLIC   mppsize
[3609]69   PUBLIC   mppsend, mpprecv                          ! needed by ICB routines
[2715]70   PUBLIC   lib_mpp_alloc   ! Called in nemogcm.F90
[415]71
[13]72   !! * Interfaces
73   !! define generic interface for these routine as they are called sometimes
[1344]74   !! with scalar arguments instead of array arguments, which causes problems
75   !! for the compilation on AIX system as well as NEC and SGI. Ok on COMPACQ
[13]76   INTERFACE mpp_min
77      MODULE PROCEDURE mppmin_a_int, mppmin_int, mppmin_a_real, mppmin_real
78   END INTERFACE
79   INTERFACE mpp_max
[681]80      MODULE PROCEDURE mppmax_a_int, mppmax_int, mppmax_a_real, mppmax_real
[13]81   END INTERFACE
82   INTERFACE mpp_sum
[1976]83      MODULE PROCEDURE mppsum_a_int, mppsum_int, mppsum_a_real, mppsum_real, &
84                       mppsum_realdd, mppsum_a_realdd
[13]85   END INTERFACE
86   INTERFACE mpp_lbc_north
87      MODULE PROCEDURE mpp_lbc_north_3d, mpp_lbc_north_2d 
88   END INTERFACE
[1344]89   INTERFACE mpp_minloc
90      MODULE PROCEDURE mpp_minloc2d ,mpp_minloc3d
91   END INTERFACE
92   INTERFACE mpp_maxloc
93      MODULE PROCEDURE mpp_maxloc2d ,mpp_maxloc3d
94   END INTERFACE
[1976]95   
[51]96   !! ========================= !!
97   !!  MPI  variable definition !!
98   !! ========================= !!
[1629]99!$AGRIF_DO_NOT_TREAT
[2004]100   INCLUDE 'mpif.h'
[1629]101!$AGRIF_END_DO_NOT_TREAT
[1344]102   
103   LOGICAL, PUBLIC, PARAMETER ::   lk_mpp = .TRUE.    !: mpp flag
[3]104
[1344]105   INTEGER, PARAMETER         ::   nprocmax = 2**10   ! maximun dimension (required to be a power of 2)
106   
107   INTEGER ::   mppsize        ! number of process
108   INTEGER ::   mpprank        ! process number  [ 0 - size-1 ]
[2363]109!$AGRIF_DO_NOT_TREAT
[2249]110   INTEGER, PUBLIC ::   mpi_comm_opa   ! opa local communicator
[2363]111!$AGRIF_END_DO_NOT_TREAT
[3]112
[2480]113   INTEGER :: MPI_SUMDD
[1976]114
[869]115   ! variables used in case of sea-ice
[3625]116   INTEGER, PUBLIC ::   ncomm_ice       !: communicator made by the processors with sea-ice (public so that it can be freed in limthd)
117   INTEGER ::   ngrp_iworld     !  group ID for the world processors (for rheology)
[1345]118   INTEGER ::   ngrp_ice        !  group ID for the ice processors (for rheology)
119   INTEGER ::   ndim_rank_ice   !  number of 'ice' processors
120   INTEGER ::   n_ice_root      !  number (in the comm_ice) of proc 0 in the ice comm
[2715]121   INTEGER, DIMENSION(:), ALLOCATABLE, SAVE ::   nrank_ice     ! dimension ndim_rank_ice
[1345]122
123   ! variables used for zonal integration
124   INTEGER, PUBLIC ::   ncomm_znl       !: communicator made by the processors on the same zonal average
125   LOGICAL, PUBLIC ::   l_znl_root      ! True on the 'left'most processor on the same row
126   INTEGER ::   ngrp_znl        ! group ID for the znl processors
127   INTEGER ::   ndim_rank_znl   ! number of processors on the same zonal average
[2715]128   INTEGER, DIMENSION(:), ALLOCATABLE, SAVE ::   nrank_znl  ! dimension ndim_rank_znl, number of the procs into the same znl domain
[1344]129   
130   ! North fold condition in mpp_mpi with jpni > 1
131   INTEGER ::   ngrp_world        ! group ID for the world processors
[1345]132   INTEGER ::   ngrp_opa          ! group ID for the opa processors
[1344]133   INTEGER ::   ngrp_north        ! group ID for the northern processors (to be fold)
134   INTEGER ::   ncomm_north       ! communicator made by the processors belonging to ngrp_north
135   INTEGER ::   ndim_rank_north   ! number of 'sea' processor in the northern line (can be /= jpni !)
136   INTEGER ::   njmppmax          ! value of njmpp for the processors of the northern line
137   INTEGER ::   north_root        ! number (in the comm_opa) of proc 0 in the northern comm
[2715]138   INTEGER, DIMENSION(:), ALLOCATABLE, SAVE ::   nrank_north   ! dimension ndim_rank_north
[3]139
[1344]140   ! Type of send : standard, buffered, immediate
[3609]141   CHARACTER(len=1)         ::   cn_mpi_send = 'S'    ! type od mpi send/recieve (S=standard, B=bsend, I=isend)
142   LOGICAL         , PUBLIC ::   l_isend = .FALSE.   ! isend use indicator (T if cn_mpi_send='I')
143   INTEGER                  ::   nn_buffer = 0       ! size of the buffer in case of mpi_bsend
[1344]144     
[2715]145   REAL(wp), DIMENSION(:), ALLOCATABLE, SAVE :: tampon  ! buffer in case of bsend
[3]146
[1344]147   ! message passing arrays
[2715]148   REAL(wp), DIMENSION(:,:,:,:,:), ALLOCATABLE, SAVE ::   t4ns, t4sn   ! 2 x 3d for north-south & south-north
149   REAL(wp), DIMENSION(:,:,:,:,:), ALLOCATABLE, SAVE ::   t4ew, t4we   ! 2 x 3d for east-west & west-east
150   REAL(wp), DIMENSION(:,:,:,:,:), ALLOCATABLE, SAVE ::   t4p1, t4p2   ! 2 x 3d for north fold
151   REAL(wp), DIMENSION(:,:,:,:)  , ALLOCATABLE, SAVE ::   t3ns, t3sn   ! 3d for north-south & south-north
152   REAL(wp), DIMENSION(:,:,:,:)  , ALLOCATABLE, SAVE ::   t3ew, t3we   ! 3d for east-west & west-east
153   REAL(wp), DIMENSION(:,:,:,:)  , ALLOCATABLE, SAVE ::   t3p1, t3p2   ! 3d for north fold
154   REAL(wp), DIMENSION(:,:,:)    , ALLOCATABLE, SAVE ::   t2ns, t2sn   ! 2d for north-south & south-north
155   REAL(wp), DIMENSION(:,:,:)    , ALLOCATABLE, SAVE ::   t2ew, t2we   ! 2d for east-west & west-east
156   REAL(wp), DIMENSION(:,:,:)    , ALLOCATABLE, SAVE ::   t2p1, t2p2   ! 2d for north fold
157
158   ! Arrays used in mpp_lbc_north_3d()
159   REAL(wp), DIMENSION(:,:,:)  , ALLOCATABLE, SAVE   ::   ztab, znorthloc
160   REAL(wp), DIMENSION(:,:,:,:), ALLOCATABLE, SAVE   ::   znorthgloio
[3294]161   REAL(wp), DIMENSION(:,:,:)  , ALLOCATABLE, SAVE   ::   zfoldwk      ! Workspace for message transfers avoiding mpi_allgather
[2715]162
163   ! Arrays used in mpp_lbc_north_2d()
164   REAL(wp), DIMENSION(:,:)  , ALLOCATABLE, SAVE    ::   ztab_2d, znorthloc_2d
165   REAL(wp), DIMENSION(:,:,:), ALLOCATABLE, SAVE    ::   znorthgloio_2d
[3294]166   REAL(wp), DIMENSION(:,:)  , ALLOCATABLE, SAVE    ::   zfoldwk_2d    ! Workspace for message transfers avoiding mpi_allgather
[2715]167
168   ! Arrays used in mpp_lbc_north_e()
169   REAL(wp), DIMENSION(:,:)  , ALLOCATABLE, SAVE    ::   ztab_e, znorthloc_e
170   REAL(wp), DIMENSION(:,:,:), ALLOCATABLE, SAVE    ::   znorthgloio_e
171
[3294]172   ! North fold arrays used to minimise the use of allgather operations. Set in nemo_northcomms (nemogcm) so need to be public
173   INTEGER, PUBLIC,  PARAMETER :: jpmaxngh = 8                 ! Assumed maximum number of active neighbours
174   INTEGER, PUBLIC,  PARAMETER :: jptyps   = 5                 ! Number of different neighbour lists to be used for northfold exchanges
175   INTEGER, PUBLIC,  DIMENSION (jpmaxngh,jptyps)    ::   isendto
176   INTEGER, PUBLIC,  DIMENSION (jptyps)             ::   nsndto
177   LOGICAL, PUBLIC                                  ::   ln_nnogather     = .FALSE.  ! namelist control of northfold comms
178   LOGICAL, PUBLIC                                  ::   l_north_nogather = .FALSE.  ! internal control of northfold comms
179   INTEGER, PUBLIC                                  ::   ityp
[51]180   !!----------------------------------------------------------------------
[2287]181   !! NEMO/OPA 3.3 , NEMO Consortium (2010)
[888]182   !! $Id$
[2715]183   !! Software governed by the CeCILL licence     (NEMOGCM/NEMO_CeCILL.txt)
[1344]184   !!----------------------------------------------------------------------
[3]185CONTAINS
186
[2715]187   INTEGER FUNCTION lib_mpp_alloc( kumout )
[51]188      !!----------------------------------------------------------------------
[2715]189      !!              ***  routine lib_mpp_alloc  ***
190      !!----------------------------------------------------------------------
191      INTEGER, INTENT(in) ::   kumout   ! ocean.output logical unit
192      !!----------------------------------------------------------------------
193      !
194      ALLOCATE( t4ns(jpi,jprecj,jpk,2,2) , t4sn(jpi,jprecj,jpk,2,2) ,                                            &
195         &      t4ew(jpj,jpreci,jpk,2,2) , t4we(jpj,jpreci,jpk,2,2) ,                                            &
196         &      t4p1(jpi,jprecj,jpk,2,2) , t4p2(jpi,jprecj,jpk,2,2) ,                                            &
197         &      t3ns(jpi,jprecj,jpk,2)   , t3sn(jpi,jprecj,jpk,2)   ,                                            &
198         &      t3ew(jpj,jpreci,jpk,2)   , t3we(jpj,jpreci,jpk,2)   ,                                            &
199         &      t3p1(jpi,jprecj,jpk,2)   , t3p2(jpi,jprecj,jpk,2)   ,                                            &
200         &      t2ns(jpi,jprecj    ,2)   , t2sn(jpi,jprecj    ,2)   ,                                            &
201         &      t2ew(jpj,jpreci    ,2)   , t2we(jpj,jpreci    ,2)   ,                                            &
202         &      t2p1(jpi,jprecj    ,2)   , t2p2(jpi,jprecj    ,2)   ,                                            &
203         !
204         &      ztab(jpiglo,4,jpk) , znorthloc(jpi,4,jpk) , znorthgloio(jpi,4,jpk,jpni) ,                        &
[3294]205         &      zfoldwk(jpi,4,jpk) ,                                                                             &
[2715]206         !
207         &      ztab_2d(jpiglo,4)  , znorthloc_2d(jpi,4)  , znorthgloio_2d(jpi,4,jpni)  ,                        &
[3294]208         &      zfoldwk_2d(jpi,4)  ,                                                                             &
[2715]209         !
210         &      ztab_e(jpiglo,4+2*jpr2dj) , znorthloc_e(jpi,4+2*jpr2dj) , znorthgloio_e(jpi,4+2*jpr2dj,jpni) ,   &
211         !
212         &      STAT=lib_mpp_alloc )
213         !
214      IF( lib_mpp_alloc /= 0 ) THEN
215         WRITE(kumout,cform_war)
216         WRITE(kumout,*) 'lib_mpp_alloc : failed to allocate arrays'
217      ENDIF
218      !
219   END FUNCTION lib_mpp_alloc
220
221
222   FUNCTION mynode( ldtxt, kumnam, kstop, localComm )
223      !!----------------------------------------------------------------------
[51]224      !!                  ***  routine mynode  ***
225      !!                   
226      !! ** Purpose :   Find processor unit
227      !!----------------------------------------------------------------------
[1579]228      CHARACTER(len=*),DIMENSION(:), INTENT(  out) ::   ldtxt 
[2715]229      INTEGER                      , INTENT(in   ) ::   kumnam       ! namelist logical unit
230      INTEGER                      , INTENT(inout) ::   kstop        ! stop indicator
[1579]231      INTEGER, OPTIONAL            , INTENT(in   ) ::   localComm
[2715]232      !
[2481]233      INTEGER ::   mynode, ierr, code, ji, ii
[532]234      LOGICAL ::   mpi_was_called
[2715]235      !
[3294]236      NAMELIST/nammpp/ cn_mpi_send, nn_buffer, jpni, jpnj, jpnij, ln_nnogather
[51]237      !!----------------------------------------------------------------------
[1344]238      !
[2481]239      ii = 1
240      WRITE(ldtxt(ii),*)                                                                          ;   ii = ii + 1
241      WRITE(ldtxt(ii),*) 'mynode : mpi initialisation'                                            ;   ii = ii + 1
242      WRITE(ldtxt(ii),*) '~~~~~~ '                                                                ;   ii = ii + 1
[1344]243      !
[2715]244      jpni = -1; jpnj = -1; jpnij = -1
245      REWIND( kumnam )               ! Namelist namrun : parameters of the run
246      READ  ( kumnam, nammpp )
[1344]247      !                              ! control print
[2481]248      WRITE(ldtxt(ii),*) '   Namelist nammpp'                                                     ;   ii = ii + 1
249      WRITE(ldtxt(ii),*) '      mpi send type                      cn_mpi_send = ', cn_mpi_send   ;   ii = ii + 1
250      WRITE(ldtxt(ii),*) '      size in bytes of exported buffer   nn_buffer   = ', nn_buffer     ;   ii = ii + 1
[300]251
[2731]252#if defined key_agrif
253      IF( .NOT. Agrif_Root() ) THEN
254         jpni  = Agrif_Parent(jpni ) 
255         jpnj  = Agrif_Parent(jpnj )
256         jpnij = Agrif_Parent(jpnij)
257      ENDIF
258#endif
259
[2715]260      IF(jpnij < 1)THEN
261         ! If jpnij is not specified in namelist then we calculate it - this
262         ! means there will be no land cutting out.
263         jpnij = jpni * jpnj
264      END IF
265
266      IF( (jpni < 1) .OR. (jpnj < 1) )THEN
267         WRITE(ldtxt(ii),*) '      jpni, jpnj and jpnij will be calculated automatically'; ii = ii + 1
268      ELSE
269         WRITE(ldtxt(ii),*) '      processor grid extent in i         jpni = ',jpni; ii = ii + 1
270         WRITE(ldtxt(ii),*) '      processor grid extent in j         jpnj = ',jpnj; ii = ii + 1
271         WRITE(ldtxt(ii),*) '      number of local domains           jpnij = ',jpnij; ii = ii +1
272      END IF
273
[3294]274      WRITE(ldtxt(ii),*) '      avoid use of mpi_allgather at the north fold  ln_nnogather = ', ln_nnogather  ; ii = ii + 1
275
[2480]276      CALL mpi_initialized ( mpi_was_called, code )
277      IF( code /= MPI_SUCCESS ) THEN
[2481]278         DO ji = 1, SIZE(ldtxt) 
279            IF( TRIM(ldtxt(ji)) /= '' )   WRITE(*,*) ldtxt(ji)      ! control print of mynode
280         END DO         
[2480]281         WRITE(*, cform_err)
282         WRITE(*, *) 'lib_mpp: Error in routine mpi_initialized'
283         CALL mpi_abort( mpi_comm_world, code, ierr )
284      ENDIF
[415]285
[2480]286      IF( mpi_was_called ) THEN
287         !
288         SELECT CASE ( cn_mpi_send )
289         CASE ( 'S' )                ! Standard mpi send (blocking)
[2481]290            WRITE(ldtxt(ii),*) '           Standard blocking mpi send (send)'                     ;   ii = ii + 1
[2480]291         CASE ( 'B' )                ! Buffer mpi send (blocking)
[2481]292            WRITE(ldtxt(ii),*) '           Buffer blocking mpi send (bsend)'                      ;   ii = ii + 1
293            IF( Agrif_Root() )   CALL mpi_init_opa( ldtxt, ii, ierr ) 
[2480]294         CASE ( 'I' )                ! Immediate mpi send (non-blocking send)
[2481]295            WRITE(ldtxt(ii),*) '           Immediate non-blocking send (isend)'                   ;   ii = ii + 1
[2480]296            l_isend = .TRUE.
297         CASE DEFAULT
[2481]298            WRITE(ldtxt(ii),cform_err)                                                            ;   ii = ii + 1
299            WRITE(ldtxt(ii),*) '           bad value for cn_mpi_send = ', cn_mpi_send             ;   ii = ii + 1
[2715]300            kstop = kstop + 1
[2480]301         END SELECT
302      ELSE IF ( PRESENT(localComm) .and. .not. mpi_was_called ) THEN
[2481]303         WRITE(ldtxt(ii),*) ' lib_mpp: You cannot provide a local communicator '                  ;   ii = ii + 1
304         WRITE(ldtxt(ii),*) '          without calling MPI_Init before ! '                        ;   ii = ii + 1
[2715]305         kstop = kstop + 1
[532]306      ELSE
[1601]307         SELECT CASE ( cn_mpi_send )
[524]308         CASE ( 'S' )                ! Standard mpi send (blocking)
[2481]309            WRITE(ldtxt(ii),*) '           Standard blocking mpi send (send)'                     ;   ii = ii + 1
[2480]310            CALL mpi_init( ierr )
[524]311         CASE ( 'B' )                ! Buffer mpi send (blocking)
[2481]312            WRITE(ldtxt(ii),*) '           Buffer blocking mpi send (bsend)'                      ;   ii = ii + 1
313            IF( Agrif_Root() )   CALL mpi_init_opa( ldtxt, ii, ierr )
[524]314         CASE ( 'I' )                ! Immediate mpi send (non-blocking send)
[2481]315            WRITE(ldtxt(ii),*) '           Immediate non-blocking send (isend)'                   ;   ii = ii + 1
[524]316            l_isend = .TRUE.
[2480]317            CALL mpi_init( ierr )
[524]318         CASE DEFAULT
[2481]319            WRITE(ldtxt(ii),cform_err)                                                            ;   ii = ii + 1
320            WRITE(ldtxt(ii),*) '           bad value for cn_mpi_send = ', cn_mpi_send             ;   ii = ii + 1
[2715]321            kstop = kstop + 1
[524]322         END SELECT
[2480]323         !
[415]324      ENDIF
[570]325
[2480]326      IF( PRESENT(localComm) ) THEN
327         IF( Agrif_Root() ) THEN
328            mpi_comm_opa = localComm
329         ENDIF
330      ELSE
331         CALL mpi_comm_dup( mpi_comm_world, mpi_comm_opa, code)
332         IF( code /= MPI_SUCCESS ) THEN
[2481]333            DO ji = 1, SIZE(ldtxt) 
334               IF( TRIM(ldtxt(ji)) /= '' )   WRITE(*,*) ldtxt(ji)      ! control print of mynode
335            END DO
[2480]336            WRITE(*, cform_err)
337            WRITE(*, *) ' lib_mpp: Error in routine mpi_comm_dup'
338            CALL mpi_abort( mpi_comm_world, code, ierr )
339         ENDIF
340      ENDIF
341
[1344]342      CALL mpi_comm_rank( mpi_comm_opa, mpprank, ierr )
343      CALL mpi_comm_size( mpi_comm_opa, mppsize, ierr )
[629]344      mynode = mpprank
[2480]345      !
[1976]346      CALL MPI_OP_CREATE(DDPDD_MPI, .TRUE., MPI_SUMDD, ierr)
347      !
[51]348   END FUNCTION mynode
[3]349
350
[888]351   SUBROUTINE mpp_lnk_3d( ptab, cd_type, psgn, cd_mpp, pval )
[51]352      !!----------------------------------------------------------------------
353      !!                  ***  routine mpp_lnk_3d  ***
354      !!
355      !! ** Purpose :   Message passing manadgement
356      !!
357      !! ** Method  :   Use mppsend and mpprecv function for passing mask
358      !!      between processors following neighboring subdomains.
359      !!            domain parameters
360      !!                    nlci   : first dimension of the local subdomain
361      !!                    nlcj   : second dimension of the local subdomain
362      !!                    nbondi : mark for "east-west local boundary"
363      !!                    nbondj : mark for "north-south local boundary"
364      !!                    noea   : number for local neighboring processors
365      !!                    nowe   : number for local neighboring processors
366      !!                    noso   : number for local neighboring processors
367      !!                    nono   : number for local neighboring processors
368      !!
369      !! ** Action  :   ptab with update value at its periphery
370      !!
371      !!----------------------------------------------------------------------
[1344]372      REAL(wp), DIMENSION(jpi,jpj,jpk), INTENT(inout) ::   ptab     ! 3D array on which the boundary condition is applied
373      CHARACTER(len=1)                , INTENT(in   ) ::   cd_type  ! define the nature of ptab array grid-points
374      !                                                             ! = T , U , V , F , W points
375      REAL(wp)                        , INTENT(in   ) ::   psgn     ! =-1 the sign change across the north fold boundary
376      !                                                             ! =  1. , the sign is kept
377      CHARACTER(len=3), OPTIONAL      , INTENT(in   ) ::   cd_mpp   ! fill the overlap area only
378      REAL(wp)        , OPTIONAL      , INTENT(in   ) ::   pval     ! background value (used at closed boundaries)
379      !!
[1718]380      INTEGER  ::   ji, jj, jk, jl             ! dummy loop indices
[1344]381      INTEGER  ::   imigr, iihom, ijhom        ! temporary integers
382      INTEGER  ::   ml_req1, ml_req2, ml_err   ! for key_mpi_isend
[888]383      REAL(wp) ::   zland
[1344]384      INTEGER, DIMENSION(MPI_STATUS_SIZE) ::   ml_stat   ! for key_mpi_isend
[51]385      !!----------------------------------------------------------------------
[3]386
[1344]387      IF( PRESENT( pval ) ) THEN   ;   zland = pval      ! set land value
388      ELSE                         ;   zland = 0.e0      ! zero by default
389      ENDIF
390
[51]391      ! 1. standard boundary treatment
392      ! ------------------------------
[1718]393      IF( PRESENT( cd_mpp ) ) THEN      ! only fill added line/raw with existing values
[1344]394         !
[1718]395         ! WARNING ptab is defined only between nld and nle
396         DO jk = 1, jpk
397            DO jj = nlcj+1, jpj                 ! added line(s)   (inner only)
398               ptab(nldi  :nlei  , jj          ,jk) = ptab(nldi:nlei,     nlej,jk)   
399               ptab(1     :nldi-1, jj          ,jk) = ptab(nldi     ,     nlej,jk)
400               ptab(nlei+1:nlci  , jj          ,jk) = ptab(     nlei,     nlej,jk)
401            END DO
402            DO ji = nlci+1, jpi                 ! added column(s) (full)
403               ptab(ji           ,nldj  :nlej  ,jk) = ptab(     nlei,nldj:nlej,jk)
404               ptab(ji           ,1     :nldj-1,jk) = ptab(     nlei,nldj     ,jk)
405               ptab(ji           ,nlej+1:jpj   ,jk) = ptab(     nlei,     nlej,jk)
406            END DO
[610]407         END DO
[1344]408         !
409      ELSE                              ! standard close or cyclic treatment
410         !
411         !                                   ! East-West boundaries
412         !                                        !* Cyclic east-west
413         IF( nbondi == 2 .AND. (nperio == 1 .OR. nperio == 4 .OR. nperio == 6) ) THEN
[473]414            ptab( 1 ,:,:) = ptab(jpim1,:,:)
415            ptab(jpi,:,:) = ptab(  2  ,:,:)
[1344]416         ELSE                                     !* closed
417            IF( .NOT. cd_type == 'F' )   ptab(     1       :jpreci,:,:) = zland    ! south except F-point
418                                         ptab(nlci-jpreci+1:jpi   ,:,:) = zland    ! north
[473]419         ENDIF
[1344]420         !                                   ! North-South boundaries (always closed)
421         IF( .NOT. cd_type == 'F' )   ptab(:,     1       :jprecj,:) = zland       ! south except F-point
422                                      ptab(:,nlcj-jprecj+1:jpj   ,:) = zland       ! north
423         !
[51]424      ENDIF
[3]425
[51]426      ! 2. East and west directions exchange
427      ! ------------------------------------
[1344]428      ! we play with the neigbours AND the row number because of the periodicity
429      !
430      SELECT CASE ( nbondi )      ! Read Dirichlet lateral conditions
431      CASE ( -1, 0, 1 )                ! all exept 2 (i.e. close case)
[51]432         iihom = nlci-nreci
433         DO jl = 1, jpreci
434            t3ew(:,jl,:,1) = ptab(jpreci+jl,:,:)
435            t3we(:,jl,:,1) = ptab(iihom +jl,:,:)
436         END DO
[1345]437      END SELECT 
[1344]438      !
439      !                           ! Migrations
[51]440      imigr = jpreci * jpj * jpk
[1344]441      !
[51]442      SELECT CASE ( nbondi ) 
443      CASE ( -1 )
[181]444         CALL mppsend( 2, t3we(1,1,1,1), imigr, noea, ml_req1 )
[3294]445         CALL mpprecv( 1, t3ew(1,1,1,2), imigr, noea )
[300]446         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
[51]447      CASE ( 0 )
[181]448         CALL mppsend( 1, t3ew(1,1,1,1), imigr, nowe, ml_req1 )
449         CALL mppsend( 2, t3we(1,1,1,1), imigr, noea, ml_req2 )
[3294]450         CALL mpprecv( 1, t3ew(1,1,1,2), imigr, noea )
451         CALL mpprecv( 2, t3we(1,1,1,2), imigr, nowe )
[300]452         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
453         IF(l_isend) CALL mpi_wait(ml_req2, ml_stat, ml_err)
[51]454      CASE ( 1 )
[181]455         CALL mppsend( 1, t3ew(1,1,1,1), imigr, nowe, ml_req1 )
[3294]456         CALL mpprecv( 2, t3we(1,1,1,2), imigr, nowe )
[300]457         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
[51]458      END SELECT
[1344]459      !
460      !                           ! Write Dirichlet lateral conditions
[51]461      iihom = nlci-jpreci
[1344]462      !
[51]463      SELECT CASE ( nbondi )
464      CASE ( -1 )
465         DO jl = 1, jpreci
466            ptab(iihom+jl,:,:) = t3ew(:,jl,:,2)
467         END DO
468      CASE ( 0 ) 
469         DO jl = 1, jpreci
470            ptab(jl      ,:,:) = t3we(:,jl,:,2)
471            ptab(iihom+jl,:,:) = t3ew(:,jl,:,2)
472         END DO
473      CASE ( 1 )
474         DO jl = 1, jpreci
475            ptab(jl      ,:,:) = t3we(:,jl,:,2)
476         END DO
477      END SELECT
[3]478
479
[51]480      ! 3. North and south directions
481      ! -----------------------------
[1344]482      ! always closed : we play only with the neigbours
483      !
484      IF( nbondj /= 2 ) THEN      ! Read Dirichlet lateral conditions
[51]485         ijhom = nlcj-nrecj
486         DO jl = 1, jprecj
487            t3sn(:,jl,:,1) = ptab(:,ijhom +jl,:)
488            t3ns(:,jl,:,1) = ptab(:,jprecj+jl,:)
489         END DO
490      ENDIF
[1344]491      !
492      !                           ! Migrations
[51]493      imigr = jprecj * jpi * jpk
[1344]494      !
[51]495      SELECT CASE ( nbondj )     
496      CASE ( -1 )
[181]497         CALL mppsend( 4, t3sn(1,1,1,1), imigr, nono, ml_req1 )
[3294]498         CALL mpprecv( 3, t3ns(1,1,1,2), imigr, nono )
[300]499         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
[51]500      CASE ( 0 )
[181]501         CALL mppsend( 3, t3ns(1,1,1,1), imigr, noso, ml_req1 )
502         CALL mppsend( 4, t3sn(1,1,1,1), imigr, nono, ml_req2 )
[3294]503         CALL mpprecv( 3, t3ns(1,1,1,2), imigr, nono )
504         CALL mpprecv( 4, t3sn(1,1,1,2), imigr, noso )
[300]505         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
506         IF(l_isend) CALL mpi_wait(ml_req2, ml_stat, ml_err)
[51]507      CASE ( 1 ) 
[181]508         CALL mppsend( 3, t3ns(1,1,1,1), imigr, noso, ml_req1 )
[3294]509         CALL mpprecv( 4, t3sn(1,1,1,2), imigr, noso )
[300]510         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
[51]511      END SELECT
[1344]512      !
513      !                           ! Write Dirichlet lateral conditions
[51]514      ijhom = nlcj-jprecj
[1344]515      !
[51]516      SELECT CASE ( nbondj )
517      CASE ( -1 )
518         DO jl = 1, jprecj
519            ptab(:,ijhom+jl,:) = t3ns(:,jl,:,2)
520         END DO
521      CASE ( 0 ) 
522         DO jl = 1, jprecj
523            ptab(:,jl      ,:) = t3sn(:,jl,:,2)
524            ptab(:,ijhom+jl,:) = t3ns(:,jl,:,2)
525         END DO
526      CASE ( 1 )
527         DO jl = 1, jprecj
528            ptab(:,jl,:) = t3sn(:,jl,:,2)
529         END DO
530      END SELECT
[3]531
532
[51]533      ! 4. north fold treatment
534      ! -----------------------
[1344]535      !
536      IF( npolj /= 0 .AND. .NOT. PRESENT(cd_mpp) ) THEN
537         !
538         SELECT CASE ( jpni )
539         CASE ( 1 )     ;   CALL lbc_nfd      ( ptab, cd_type, psgn )   ! only 1 northern proc, no mpp
540         CASE DEFAULT   ;   CALL mpp_lbc_north( ptab, cd_type, psgn )   ! for all northern procs.
541         END SELECT
542         !
[473]543      ENDIF
[1344]544      !
[51]545   END SUBROUTINE mpp_lnk_3d
[3]546
547
[888]548   SUBROUTINE mpp_lnk_2d( pt2d, cd_type, psgn, cd_mpp, pval )
[51]549      !!----------------------------------------------------------------------
550      !!                  ***  routine mpp_lnk_2d  ***
551      !!                 
552      !! ** Purpose :   Message passing manadgement for 2d array
553      !!
554      !! ** Method  :   Use mppsend and mpprecv function for passing mask
555      !!      between processors following neighboring subdomains.
556      !!            domain parameters
557      !!                    nlci   : first dimension of the local subdomain
558      !!                    nlcj   : second dimension of the local subdomain
559      !!                    nbondi : mark for "east-west local boundary"
560      !!                    nbondj : mark for "north-south local boundary"
561      !!                    noea   : number for local neighboring processors
562      !!                    nowe   : number for local neighboring processors
563      !!                    noso   : number for local neighboring processors
564      !!                    nono   : number for local neighboring processors
565      !!
566      !!----------------------------------------------------------------------
[1344]567      REAL(wp), DIMENSION(jpi,jpj), INTENT(inout) ::   pt2d     ! 2D array on which the boundary condition is applied
568      CHARACTER(len=1)            , INTENT(in   ) ::   cd_type  ! define the nature of ptab array grid-points
569      !                                                         ! = T , U , V , F , W and I points
570      REAL(wp)                    , INTENT(in   ) ::   psgn     ! =-1 the sign change across the north fold boundary
571      !                                                         ! =  1. , the sign is kept
572      CHARACTER(len=3), OPTIONAL  , INTENT(in   ) ::   cd_mpp   ! fill the overlap area only
573      REAL(wp)        , OPTIONAL  , INTENT(in   ) ::   pval     ! background value (used at closed boundaries)
574      !!
575      INTEGER  ::   ji, jj, jl   ! dummy loop indices
576      INTEGER  ::   imigr, iihom, ijhom        ! temporary integers
577      INTEGER  ::   ml_req1, ml_req2, ml_err   ! for key_mpi_isend
[888]578      REAL(wp) ::   zland
[1344]579      INTEGER, DIMENSION(MPI_STATUS_SIZE) ::   ml_stat   ! for key_mpi_isend
[51]580      !!----------------------------------------------------------------------
[3]581
[1344]582      IF( PRESENT( pval ) ) THEN   ;   zland = pval      ! set land value
583      ELSE                         ;   zland = 0.e0      ! zero by default
[888]584      ENDIF
585
[51]586      ! 1. standard boundary treatment
587      ! ------------------------------
[1344]588      !
[1718]589      IF( PRESENT( cd_mpp ) ) THEN      ! only fill added line/raw with existing values
[1344]590         !
[1718]591         ! WARNING pt2d is defined only between nld and nle
592         DO jj = nlcj+1, jpj                 ! added line(s)   (inner only)
593            pt2d(nldi  :nlei  , jj          ) = pt2d(nldi:nlei,     nlej)   
594            pt2d(1     :nldi-1, jj          ) = pt2d(nldi     ,     nlej)
595            pt2d(nlei+1:nlci  , jj          ) = pt2d(     nlei,     nlej)
[610]596         END DO
[1718]597         DO ji = nlci+1, jpi                 ! added column(s) (full)
598            pt2d(ji           ,nldj  :nlej  ) = pt2d(     nlei,nldj:nlej)
599            pt2d(ji           ,1     :nldj-1) = pt2d(     nlei,nldj     )
600            pt2d(ji           ,nlej+1:jpj   ) = pt2d(     nlei,     nlej)
[1344]601         END DO
602         !
603      ELSE                              ! standard close or cyclic treatment
604         !
605         !                                   ! East-West boundaries
606         IF( nbondi == 2 .AND.   &                ! Cyclic east-west
[473]607            &    (nperio == 1 .OR. nperio == 4 .OR. nperio == 6) ) THEN
[1344]608            pt2d( 1 ,:) = pt2d(jpim1,:)                                    ! west
609            pt2d(jpi,:) = pt2d(  2  ,:)                                    ! east
610         ELSE                                     ! closed
611            IF( .NOT. cd_type == 'F' )   pt2d(     1       :jpreci,:) = zland    ! south except F-point
612                                         pt2d(nlci-jpreci+1:jpi   ,:) = zland    ! north
[473]613         ENDIF
[1344]614         !                                   ! North-South boundaries (always closed)
615            IF( .NOT. cd_type == 'F' )   pt2d(:,     1       :jprecj) = zland    !south except F-point
616                                         pt2d(:,nlcj-jprecj+1:jpj   ) = zland    ! north
617         !
[51]618      ENDIF
[3]619
[1344]620      ! 2. East and west directions exchange
621      ! ------------------------------------
622      ! we play with the neigbours AND the row number because of the periodicity
623      !
624      SELECT CASE ( nbondi )      ! Read Dirichlet lateral conditions
625      CASE ( -1, 0, 1 )                ! all exept 2 (i.e. close case)
[51]626         iihom = nlci-nreci
627         DO jl = 1, jpreci
628            t2ew(:,jl,1) = pt2d(jpreci+jl,:)
629            t2we(:,jl,1) = pt2d(iihom +jl,:)
630         END DO
631      END SELECT
[1344]632      !
633      !                           ! Migrations
[51]634      imigr = jpreci * jpj
[1344]635      !
[51]636      SELECT CASE ( nbondi )
637      CASE ( -1 )
[181]638         CALL mppsend( 2, t2we(1,1,1), imigr, noea, ml_req1 )
[3294]639         CALL mpprecv( 1, t2ew(1,1,2), imigr, noea )
[300]640         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
[51]641      CASE ( 0 )
[181]642         CALL mppsend( 1, t2ew(1,1,1), imigr, nowe, ml_req1 )
643         CALL mppsend( 2, t2we(1,1,1), imigr, noea, ml_req2 )
[3294]644         CALL mpprecv( 1, t2ew(1,1,2), imigr, noea )
645         CALL mpprecv( 2, t2we(1,1,2), imigr, nowe )
[300]646         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
647         IF(l_isend) CALL mpi_wait(ml_req2,ml_stat,ml_err)
[51]648      CASE ( 1 )
[181]649         CALL mppsend( 1, t2ew(1,1,1), imigr, nowe, ml_req1 )
[3294]650         CALL mpprecv( 2, t2we(1,1,2), imigr, nowe )
[300]651         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
[51]652      END SELECT
[1344]653      !
654      !                           ! Write Dirichlet lateral conditions
[51]655      iihom = nlci - jpreci
[1344]656      !
[51]657      SELECT CASE ( nbondi )
658      CASE ( -1 )
659         DO jl = 1, jpreci
660            pt2d(iihom+jl,:) = t2ew(:,jl,2)
661         END DO
662      CASE ( 0 )
663         DO jl = 1, jpreci
664            pt2d(jl      ,:) = t2we(:,jl,2)
665            pt2d(iihom+jl,:) = t2ew(:,jl,2)
666         END DO
667      CASE ( 1 )
668         DO jl = 1, jpreci
669            pt2d(jl      ,:) = t2we(:,jl,2)
670         END DO
671      END SELECT
[3]672
673
[51]674      ! 3. North and south directions
675      ! -----------------------------
[1344]676      ! always closed : we play only with the neigbours
677      !
678      IF( nbondj /= 2 ) THEN      ! Read Dirichlet lateral conditions
[51]679         ijhom = nlcj-nrecj
680         DO jl = 1, jprecj
681            t2sn(:,jl,1) = pt2d(:,ijhom +jl)
682            t2ns(:,jl,1) = pt2d(:,jprecj+jl)
683         END DO
684      ENDIF
[1344]685      !
686      !                           ! Migrations
[51]687      imigr = jprecj * jpi
[1344]688      !
[51]689      SELECT CASE ( nbondj )
690      CASE ( -1 )
[181]691         CALL mppsend( 4, t2sn(1,1,1), imigr, nono, ml_req1 )
[3294]692         CALL mpprecv( 3, t2ns(1,1,2), imigr, nono )
[300]693         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
[51]694      CASE ( 0 )
[181]695         CALL mppsend( 3, t2ns(1,1,1), imigr, noso, ml_req1 )
696         CALL mppsend( 4, t2sn(1,1,1), imigr, nono, ml_req2 )
[3294]697         CALL mpprecv( 3, t2ns(1,1,2), imigr, nono )
698         CALL mpprecv( 4, t2sn(1,1,2), imigr, noso )
[300]699         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
700         IF(l_isend) CALL mpi_wait(ml_req2,ml_stat,ml_err)
[51]701      CASE ( 1 )
[181]702         CALL mppsend( 3, t2ns(1,1,1), imigr, noso, ml_req1 )
[3294]703         CALL mpprecv( 4, t2sn(1,1,2), imigr, noso )
[300]704         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
[51]705      END SELECT
[1344]706      !
707      !                           ! Write Dirichlet lateral conditions
[51]708      ijhom = nlcj - jprecj
[1344]709      !
[51]710      SELECT CASE ( nbondj )
711      CASE ( -1 )
712         DO jl = 1, jprecj
713            pt2d(:,ijhom+jl) = t2ns(:,jl,2)
714         END DO
715      CASE ( 0 )
716         DO jl = 1, jprecj
717            pt2d(:,jl      ) = t2sn(:,jl,2)
718            pt2d(:,ijhom+jl) = t2ns(:,jl,2)
719         END DO
720      CASE ( 1 ) 
721         DO jl = 1, jprecj
722            pt2d(:,jl      ) = t2sn(:,jl,2)
723         END DO
[1344]724      END SELECT
[3]725
[1344]726
[51]727      ! 4. north fold treatment
728      ! -----------------------
[1344]729      !
730      IF( npolj /= 0 .AND. .NOT. PRESENT(cd_mpp) ) THEN
731         !
732         SELECT CASE ( jpni )
733         CASE ( 1 )     ;   CALL lbc_nfd      ( pt2d, cd_type, psgn )   ! only 1 northern proc, no mpp
734         CASE DEFAULT   ;   CALL mpp_lbc_north( pt2d, cd_type, psgn )   ! for all northern procs.
735         END SELECT
736         !
[473]737      ENDIF
[1344]738      !
[51]739   END SUBROUTINE mpp_lnk_2d
[3]740
741
[473]742   SUBROUTINE mpp_lnk_3d_gather( ptab1, cd_type1, ptab2, cd_type2, psgn )
743      !!----------------------------------------------------------------------
744      !!                  ***  routine mpp_lnk_3d_gather  ***
745      !!
746      !! ** Purpose :   Message passing manadgement for two 3D arrays
747      !!
748      !! ** Method  :   Use mppsend and mpprecv function for passing mask
749      !!      between processors following neighboring subdomains.
750      !!            domain parameters
751      !!                    nlci   : first dimension of the local subdomain
752      !!                    nlcj   : second dimension of the local subdomain
753      !!                    nbondi : mark for "east-west local boundary"
754      !!                    nbondj : mark for "north-south local boundary"
755      !!                    noea   : number for local neighboring processors
756      !!                    nowe   : number for local neighboring processors
757      !!                    noso   : number for local neighboring processors
758      !!                    nono   : number for local neighboring processors
759      !!
760      !! ** Action  :   ptab1 and ptab2  with update value at its periphery
761      !!
762      !!----------------------------------------------------------------------
[1344]763      REAL(wp), DIMENSION(jpi,jpj,jpk), INTENT(inout) ::   ptab1     ! first and second 3D array on which
764      REAL(wp), DIMENSION(jpi,jpj,jpk), INTENT(inout) ::   ptab2     ! the boundary condition is applied
765      CHARACTER(len=1)                , INTENT(in   ) ::   cd_type1  ! nature of ptab1 and ptab2 arrays
766      CHARACTER(len=1)                , INTENT(in   ) ::   cd_type2  ! i.e. grid-points = T , U , V , F or W points
767      REAL(wp)                        , INTENT(in   ) ::   psgn      ! =-1 the sign change across the north fold boundary
768      !!                                                             ! =  1. , the sign is kept
769      INTEGER  ::   jl   ! dummy loop indices
770      INTEGER  ::   imigr, iihom, ijhom        ! temporary integers
771      INTEGER  ::   ml_req1, ml_req2, ml_err   ! for key_mpi_isend
772      INTEGER, DIMENSION(MPI_STATUS_SIZE) ::   ml_stat   ! for key_mpi_isend
[473]773      !!----------------------------------------------------------------------
774
775      ! 1. standard boundary treatment
776      ! ------------------------------
[1344]777      !                                      ! East-West boundaries
778      !                                           !* Cyclic east-west
779      IF( nbondi == 2 .AND. (nperio == 1 .OR. nperio == 4 .OR. nperio == 6) ) THEN
[473]780         ptab1( 1 ,:,:) = ptab1(jpim1,:,:)
781         ptab1(jpi,:,:) = ptab1(  2  ,:,:)
782         ptab2( 1 ,:,:) = ptab2(jpim1,:,:)
783         ptab2(jpi,:,:) = ptab2(  2  ,:,:)
[1344]784      ELSE                                        !* closed
785         IF( .NOT. cd_type1 == 'F' )   ptab1(     1       :jpreci,:,:) = 0.e0    ! south except at F-point
786         IF( .NOT. cd_type2 == 'F' )   ptab2(     1       :jpreci,:,:) = 0.e0
787                                       ptab1(nlci-jpreci+1:jpi   ,:,:) = 0.e0    ! north
788                                       ptab2(nlci-jpreci+1:jpi   ,:,:) = 0.e0
[473]789      ENDIF
790
[1344]791     
792      !                                      ! North-South boundaries
793      IF( .NOT. cd_type1 == 'F' )   ptab1(:,     1       :jprecj,:) = 0.e0    ! south except at F-point
794      IF( .NOT. cd_type2 == 'F' )   ptab2(:,     1       :jprecj,:) = 0.e0
795                                    ptab1(:,nlcj-jprecj+1:jpj   ,:) = 0.e0    ! north
796                                    ptab2(:,nlcj-jprecj+1:jpj   ,:) = 0.e0
[473]797
798
799      ! 2. East and west directions exchange
800      ! ------------------------------------
[1344]801      ! we play with the neigbours AND the row number because of the periodicity
802      !
803      SELECT CASE ( nbondi )      ! Read Dirichlet lateral conditions
804      CASE ( -1, 0, 1 )                ! all exept 2 (i.e. close case)
[473]805         iihom = nlci-nreci
806         DO jl = 1, jpreci
807            t4ew(:,jl,:,1,1) = ptab1(jpreci+jl,:,:)
808            t4we(:,jl,:,1,1) = ptab1(iihom +jl,:,:)
809            t4ew(:,jl,:,2,1) = ptab2(jpreci+jl,:,:)
810            t4we(:,jl,:,2,1) = ptab2(iihom +jl,:,:)
811         END DO
812      END SELECT
[1344]813      !
814      !                           ! Migrations
[473]815      imigr = jpreci * jpj * jpk *2
[1344]816      !
[473]817      SELECT CASE ( nbondi ) 
818      CASE ( -1 )
819         CALL mppsend( 2, t4we(1,1,1,1,1), imigr, noea, ml_req1 )
[3294]820         CALL mpprecv( 1, t4ew(1,1,1,1,2), imigr, noea )
[473]821         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
822      CASE ( 0 )
823         CALL mppsend( 1, t4ew(1,1,1,1,1), imigr, nowe, ml_req1 )
824         CALL mppsend( 2, t4we(1,1,1,1,1), imigr, noea, ml_req2 )
[3294]825         CALL mpprecv( 1, t4ew(1,1,1,1,2), imigr, noea )
826         CALL mpprecv( 2, t4we(1,1,1,1,2), imigr, nowe )
[473]827         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
828         IF(l_isend) CALL mpi_wait(ml_req2, ml_stat, ml_err)
829      CASE ( 1 )
830         CALL mppsend( 1, t4ew(1,1,1,1,1), imigr, nowe, ml_req1 )
[3294]831         CALL mpprecv( 2, t4we(1,1,1,1,2), imigr, nowe )
[473]832         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
833      END SELECT
[1344]834      !
835      !                           ! Write Dirichlet lateral conditions
836      iihom = nlci - jpreci
837      !
[473]838      SELECT CASE ( nbondi )
839      CASE ( -1 )
840         DO jl = 1, jpreci
841            ptab1(iihom+jl,:,:) = t4ew(:,jl,:,1,2)
842            ptab2(iihom+jl,:,:) = t4ew(:,jl,:,2,2)
843         END DO
844      CASE ( 0 ) 
845         DO jl = 1, jpreci
846            ptab1(jl      ,:,:) = t4we(:,jl,:,1,2)
847            ptab1(iihom+jl,:,:) = t4ew(:,jl,:,1,2)
848            ptab2(jl      ,:,:) = t4we(:,jl,:,2,2)
849            ptab2(iihom+jl,:,:) = t4ew(:,jl,:,2,2)
850         END DO
851      CASE ( 1 )
852         DO jl = 1, jpreci
853            ptab1(jl      ,:,:) = t4we(:,jl,:,1,2)
854            ptab2(jl      ,:,:) = t4we(:,jl,:,2,2)
855         END DO
856      END SELECT
857
858
859      ! 3. North and south directions
860      ! -----------------------------
[1344]861      ! always closed : we play only with the neigbours
862      !
863      IF( nbondj /= 2 ) THEN      ! Read Dirichlet lateral conditions
864         ijhom = nlcj - nrecj
[473]865         DO jl = 1, jprecj
866            t4sn(:,jl,:,1,1) = ptab1(:,ijhom +jl,:)
867            t4ns(:,jl,:,1,1) = ptab1(:,jprecj+jl,:)
868            t4sn(:,jl,:,2,1) = ptab2(:,ijhom +jl,:)
869            t4ns(:,jl,:,2,1) = ptab2(:,jprecj+jl,:)
870         END DO
871      ENDIF
[1344]872      !
873      !                           ! Migrations
[473]874      imigr = jprecj * jpi * jpk * 2
[1344]875      !
[473]876      SELECT CASE ( nbondj )     
877      CASE ( -1 )
878         CALL mppsend( 4, t4sn(1,1,1,1,1), imigr, nono, ml_req1 )
[3294]879         CALL mpprecv( 3, t4ns(1,1,1,1,2), imigr, nono )
[473]880         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
881      CASE ( 0 )
882         CALL mppsend( 3, t4ns(1,1,1,1,1), imigr, noso, ml_req1 )
883         CALL mppsend( 4, t4sn(1,1,1,1,1), imigr, nono, ml_req2 )
[3294]884         CALL mpprecv( 3, t4ns(1,1,1,1,2), imigr, nono )
885         CALL mpprecv( 4, t4sn(1,1,1,1,2), imigr, noso )
[473]886         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
887         IF(l_isend) CALL mpi_wait(ml_req2, ml_stat, ml_err)
888      CASE ( 1 ) 
889         CALL mppsend( 3, t4ns(1,1,1,1,1), imigr, noso, ml_req1 )
[3294]890         CALL mpprecv( 4, t4sn(1,1,1,1,2), imigr, noso )
[473]891         IF(l_isend) CALL mpi_wait(ml_req1, ml_stat, ml_err)
892      END SELECT
[1344]893      !
894      !                           ! Write Dirichlet lateral conditions
895      ijhom = nlcj - jprecj
896      !
[473]897      SELECT CASE ( nbondj )
898      CASE ( -1 )
899         DO jl = 1, jprecj
900            ptab1(:,ijhom+jl,:) = t4ns(:,jl,:,1,2)
901            ptab2(:,ijhom+jl,:) = t4ns(:,jl,:,2,2)
902         END DO
903      CASE ( 0 ) 
904         DO jl = 1, jprecj
905            ptab1(:,jl      ,:) = t4sn(:,jl,:,1,2)
906            ptab1(:,ijhom+jl,:) = t4ns(:,jl,:,1,2)
907            ptab2(:,jl      ,:) = t4sn(:,jl,:,2,2)
908            ptab2(:,ijhom+jl,:) = t4ns(:,jl,:,2,2)
909         END DO
910      CASE ( 1 )
911         DO jl = 1, jprecj
912            ptab1(:,jl,:) = t4sn(:,jl,:,1,2)
913            ptab2(:,jl,:) = t4sn(:,jl,:,2,2)
914         END DO
915      END SELECT
916
917
918      ! 4. north fold treatment
919      ! -----------------------
[1344]920      IF( npolj /= 0 ) THEN
921         !
922         SELECT CASE ( jpni )
923         CASE ( 1 )                                           
924            CALL lbc_nfd      ( ptab1, cd_type1, psgn )   ! only for northern procs.
925            CALL lbc_nfd      ( ptab2, cd_type2, psgn )
926         CASE DEFAULT
927            CALL mpp_lbc_north( ptab1, cd_type1, psgn )   ! for all northern procs.
928            CALL mpp_lbc_north (ptab2, cd_type2, psgn)
929         END SELECT 
930         !
931      ENDIF
932      !
[473]933   END SUBROUTINE mpp_lnk_3d_gather
934
935
[3609]936   SUBROUTINE mpp_lnk_2d_e( pt2d, cd_type, psgn, jpri, jprj )
[311]937      !!----------------------------------------------------------------------
938      !!                  ***  routine mpp_lnk_2d_e  ***
939      !!                 
940      !! ** Purpose :   Message passing manadgement for 2d array (with halo)
941      !!
942      !! ** Method  :   Use mppsend and mpprecv function for passing mask
943      !!      between processors following neighboring subdomains.
944      !!            domain parameters
945      !!                    nlci   : first dimension of the local subdomain
946      !!                    nlcj   : second dimension of the local subdomain
[3609]947      !!                    jpri   : number of rows for extra outer halo
948      !!                    jprj   : number of columns for extra outer halo
[311]949      !!                    nbondi : mark for "east-west local boundary"
950      !!                    nbondj : mark for "north-south local boundary"
951      !!                    noea   : number for local neighboring processors
952      !!                    nowe   : number for local neighboring processors
953      !!                    noso   : number for local neighboring processors
954      !!                    nono   : number for local neighboring processors
955      !!
956      !!----------------------------------------------------------------------
[3609]957      INTEGER                                             , INTENT(in   ) ::   jpri
958      INTEGER                                             , INTENT(in   ) ::   jprj
959      REAL(wp), DIMENSION(1-jpri:jpi+jpri,1-jprj:jpj+jprj), INTENT(inout) ::   pt2d     ! 2D array with extra halo
960      CHARACTER(len=1)                                    , INTENT(in   ) ::   cd_type  ! nature of ptab array grid-points
961      !                                                                                 ! = T , U , V , F , W and I points
962      REAL(wp)                                            , INTENT(in   ) ::   psgn     ! =-1 the sign change across the
963      !!                                                                                ! north boundary, =  1. otherwise
[1344]964      INTEGER  ::   jl   ! dummy loop indices
965      INTEGER  ::   imigr, iihom, ijhom        ! temporary integers
966      INTEGER  ::   ipreci, iprecj             ! temporary integers
967      INTEGER  ::   ml_req1, ml_req2, ml_err   ! for key_mpi_isend
968      INTEGER, DIMENSION(MPI_STATUS_SIZE) ::   ml_stat   ! for key_mpi_isend
[3609]969      !!
970      REAL(wp), DIMENSION(1-jpri:jpi+jpri,jprecj+jprj,2) :: r2dns
971      REAL(wp), DIMENSION(1-jpri:jpi+jpri,jprecj+jprj,2) :: r2dsn
972      REAL(wp), DIMENSION(1-jprj:jpj+jprj,jpreci+jpri,2) :: r2dwe
973      REAL(wp), DIMENSION(1-jprj:jpj+jprj,jpreci+jpri,2) :: r2dew
[1344]974      !!----------------------------------------------------------------------
[311]975
[3609]976      ipreci = jpreci + jpri      ! take into account outer extra 2D overlap area
977      iprecj = jprecj + jprj
[311]978
979
980      ! 1. standard boundary treatment
981      ! ------------------------------
[1344]982      ! Order matters Here !!!!
983      !
984      !                                      !* North-South boundaries (always colsed)
[3609]985      IF( .NOT. cd_type == 'F' )   pt2d(:,  1-jprj   :  jprecj  ) = 0.e0    ! south except at F-point
986                                   pt2d(:,nlcj-jprecj+1:jpj+jprj) = 0.e0    ! north
[1344]987                               
988      !                                      ! East-West boundaries
989      !                                           !* Cyclic east-west
990      IF( nbondi == 2 .AND. (nperio == 1 .OR. nperio == 4 .OR. nperio == 6) ) THEN
[3609]991         pt2d(1-jpri:     1    ,:) = pt2d(jpim1-jpri:  jpim1 ,:)       ! east
992         pt2d(   jpi  :jpi+jpri,:) = pt2d(     2      :2+jpri,:)       ! west
[1344]993         !
994      ELSE                                        !* closed
[3609]995         IF( .NOT. cd_type == 'F' )   pt2d(  1-jpri   :jpreci    ,:) = 0.e0    ! south except at F-point
996                                      pt2d(nlci-jpreci+1:jpi+jpri,:) = 0.e0    ! north
[1344]997      ENDIF
998      !
[311]999
[1344]1000      ! north fold treatment
1001      ! -----------------------
1002      IF( npolj /= 0 ) THEN
1003         !
1004         SELECT CASE ( jpni )
[3609]1005         CASE ( 1 )     ;   CALL lbc_nfd        ( pt2d(1:jpi,1:jpj+jprj), cd_type, psgn, pr2dj=jprj )
[1344]1006         CASE DEFAULT   ;   CALL mpp_lbc_north_e( pt2d                    , cd_type, psgn               )
1007         END SELECT 
1008         !
[311]1009      ENDIF
1010
[1344]1011      ! 2. East and west directions exchange
1012      ! ------------------------------------
1013      ! we play with the neigbours AND the row number because of the periodicity
1014      !
1015      SELECT CASE ( nbondi )      ! Read Dirichlet lateral conditions
1016      CASE ( -1, 0, 1 )                ! all exept 2 (i.e. close case)
[3609]1017         iihom = nlci-nreci-jpri
[311]1018         DO jl = 1, ipreci
[3609]1019            r2dew(:,jl,1) = pt2d(jpreci+jl,:)
1020            r2dwe(:,jl,1) = pt2d(iihom +jl,:)
[311]1021         END DO
1022      END SELECT
[1344]1023      !
1024      !                           ! Migrations
[3609]1025      imigr = ipreci * ( jpj + 2*jprj)
[1344]1026      !
[311]1027      SELECT CASE ( nbondi )
1028      CASE ( -1 )
[3609]1029         CALL mppsend( 2, r2dwe(1-jprj,1,1), imigr, noea, ml_req1 )
1030         CALL mpprecv( 1, r2dew(1-jprj,1,2), imigr, noea )
[311]1031         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1032      CASE ( 0 )
[3609]1033         CALL mppsend( 1, r2dew(1-jprj,1,1), imigr, nowe, ml_req1 )
1034         CALL mppsend( 2, r2dwe(1-jprj,1,1), imigr, noea, ml_req2 )
1035         CALL mpprecv( 1, r2dew(1-jprj,1,2), imigr, noea )
1036         CALL mpprecv( 2, r2dwe(1-jprj,1,2), imigr, nowe )
[311]1037         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1038         IF(l_isend) CALL mpi_wait(ml_req2,ml_stat,ml_err)
1039      CASE ( 1 )
[3609]1040         CALL mppsend( 1, r2dew(1-jprj,1,1), imigr, nowe, ml_req1 )
1041         CALL mpprecv( 2, r2dwe(1-jprj,1,2), imigr, nowe )
[311]1042         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1043      END SELECT
[1344]1044      !
1045      !                           ! Write Dirichlet lateral conditions
[311]1046      iihom = nlci - jpreci
[1344]1047      !
[311]1048      SELECT CASE ( nbondi )
1049      CASE ( -1 )
1050         DO jl = 1, ipreci
[3609]1051            pt2d(iihom+jl,:) = r2dew(:,jl,2)
[311]1052         END DO
1053      CASE ( 0 )
1054         DO jl = 1, ipreci
[3609]1055            pt2d(jl-jpri,:) = r2dwe(:,jl,2)
1056            pt2d( iihom+jl,:) = r2dew(:,jl,2)
[311]1057         END DO
1058      CASE ( 1 )
1059         DO jl = 1, ipreci
[3609]1060            pt2d(jl-jpri,:) = r2dwe(:,jl,2)
[311]1061         END DO
1062      END SELECT
1063
1064
1065      ! 3. North and south directions
1066      ! -----------------------------
[1344]1067      ! always closed : we play only with the neigbours
1068      !
1069      IF( nbondj /= 2 ) THEN      ! Read Dirichlet lateral conditions
[3609]1070         ijhom = nlcj-nrecj-jprj
[311]1071         DO jl = 1, iprecj
[3609]1072            r2dsn(:,jl,1) = pt2d(:,ijhom +jl)
1073            r2dns(:,jl,1) = pt2d(:,jprecj+jl)
[311]1074         END DO
1075      ENDIF
[1344]1076      !
1077      !                           ! Migrations
[3609]1078      imigr = iprecj * ( jpi + 2*jpri )
[1344]1079      !
[311]1080      SELECT CASE ( nbondj )
1081      CASE ( -1 )
[3609]1082         CALL mppsend( 4, r2dsn(1-jpri,1,1), imigr, nono, ml_req1 )
1083         CALL mpprecv( 3, r2dns(1-jpri,1,2), imigr, nono )
[311]1084         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1085      CASE ( 0 )
[3609]1086         CALL mppsend( 3, r2dns(1-jpri,1,1), imigr, noso, ml_req1 )
1087         CALL mppsend( 4, r2dsn(1-jpri,1,1), imigr, nono, ml_req2 )
1088         CALL mpprecv( 3, r2dns(1-jpri,1,2), imigr, nono )
1089         CALL mpprecv( 4, r2dsn(1-jpri,1,2), imigr, noso )
[311]1090         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1091         IF(l_isend) CALL mpi_wait(ml_req2,ml_stat,ml_err)
1092      CASE ( 1 )
[3609]1093         CALL mppsend( 3, r2dns(1-jpri,1,1), imigr, noso, ml_req1 )
1094         CALL mpprecv( 4, r2dsn(1-jpri,1,2), imigr, noso )
[311]1095         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1096      END SELECT
[1344]1097      !
1098      !                           ! Write Dirichlet lateral conditions
[311]1099      ijhom = nlcj - jprecj 
[1344]1100      !
[311]1101      SELECT CASE ( nbondj )
1102      CASE ( -1 )
1103         DO jl = 1, iprecj
[3609]1104            pt2d(:,ijhom+jl) = r2dns(:,jl,2)
[311]1105         END DO
1106      CASE ( 0 )
1107         DO jl = 1, iprecj
[3609]1108            pt2d(:,jl-jprj) = r2dsn(:,jl,2)
1109            pt2d(:,ijhom+jl ) = r2dns(:,jl,2)
[311]1110         END DO
1111      CASE ( 1 ) 
1112         DO jl = 1, iprecj
[3609]1113            pt2d(:,jl-jprj) = r2dsn(:,jl,2)
[311]1114         END DO
[1344]1115      END SELECT
[311]1116
1117   END SUBROUTINE mpp_lnk_2d_e
1118
1119
[1344]1120   SUBROUTINE mppsend( ktyp, pmess, kbytes, kdest, md_req )
[51]1121      !!----------------------------------------------------------------------
1122      !!                  ***  routine mppsend  ***
1123      !!                   
1124      !! ** Purpose :   Send messag passing array
1125      !!
1126      !!----------------------------------------------------------------------
[1344]1127      REAL(wp), INTENT(inout) ::   pmess(*)   ! array of real
1128      INTEGER , INTENT(in   ) ::   kbytes     ! size of the array pmess
1129      INTEGER , INTENT(in   ) ::   kdest      ! receive process number
1130      INTEGER , INTENT(in   ) ::   ktyp       ! tag of the message
1131      INTEGER , INTENT(in   ) ::   md_req     ! argument for isend
1132      !!
1133      INTEGER ::   iflag
[51]1134      !!----------------------------------------------------------------------
[1344]1135      !
[1601]1136      SELECT CASE ( cn_mpi_send )
[300]1137      CASE ( 'S' )                ! Standard mpi send (blocking)
[1344]1138         CALL mpi_send ( pmess, kbytes, mpi_double_precision, kdest , ktyp, mpi_comm_opa        , iflag )
[300]1139      CASE ( 'B' )                ! Buffer mpi send (blocking)
[1344]1140         CALL mpi_bsend( pmess, kbytes, mpi_double_precision, kdest , ktyp, mpi_comm_opa        , iflag )
[300]1141      CASE ( 'I' )                ! Immediate mpi send (non-blocking send)
[1344]1142         ! be carefull, one more argument here : the mpi request identifier..
1143         CALL mpi_isend( pmess, kbytes, mpi_double_precision, kdest , ktyp, mpi_comm_opa, md_req, iflag )
[300]1144      END SELECT
[1344]1145      !
[51]1146   END SUBROUTINE mppsend
[3]1147
1148
[3294]1149   SUBROUTINE mpprecv( ktyp, pmess, kbytes, ksource )
[51]1150      !!----------------------------------------------------------------------
1151      !!                  ***  routine mpprecv  ***
1152      !!
1153      !! ** Purpose :   Receive messag passing array
1154      !!
1155      !!----------------------------------------------------------------------
[1344]1156      REAL(wp), INTENT(inout) ::   pmess(*)   ! array of real
1157      INTEGER , INTENT(in   ) ::   kbytes     ! suze of the array pmess
1158      INTEGER , INTENT(in   ) ::   ktyp       ! Tag of the recevied message
[3294]1159      INTEGER, OPTIONAL, INTENT(in) :: ksource    ! source process number
[1344]1160      !!
[51]1161      INTEGER :: istatus(mpi_status_size)
1162      INTEGER :: iflag
[3294]1163      INTEGER :: use_source
[1344]1164      !!----------------------------------------------------------------------
1165      !
[3294]1166
1167      ! If a specific process number has been passed to the receive call,
1168      ! use that one. Default is to use mpi_any_source
1169      use_source=mpi_any_source
1170      if(present(ksource)) then
1171         use_source=ksource
1172      end if
1173
1174      CALL mpi_recv( pmess, kbytes, mpi_double_precision, use_source, ktyp, mpi_comm_opa, istatus, iflag )
[1344]1175      !
[51]1176   END SUBROUTINE mpprecv
[3]1177
1178
[51]1179   SUBROUTINE mppgather( ptab, kp, pio )
1180      !!----------------------------------------------------------------------
1181      !!                   ***  routine mppgather  ***
1182      !!                   
1183      !! ** Purpose :   Transfert between a local subdomain array and a work
1184      !!     array which is distributed following the vertical level.
1185      !!
[1344]1186      !!----------------------------------------------------------------------
1187      REAL(wp), DIMENSION(jpi,jpj),       INTENT(in   ) ::   ptab   ! subdomain input array
1188      INTEGER ,                           INTENT(in   ) ::   kp     ! record length
1189      REAL(wp), DIMENSION(jpi,jpj,jpnij), INTENT(  out) ::   pio    ! subdomain input array
[51]1190      !!
[1344]1191      INTEGER :: itaille, ierror   ! temporary integer
[51]1192      !!---------------------------------------------------------------------
[1344]1193      !
1194      itaille = jpi * jpj
1195      CALL mpi_gather( ptab, itaille, mpi_double_precision, pio, itaille     ,   &
[532]1196         &                            mpi_double_precision, kp , mpi_comm_opa, ierror ) 
[1344]1197      !
[51]1198   END SUBROUTINE mppgather
[3]1199
1200
[51]1201   SUBROUTINE mppscatter( pio, kp, ptab )
1202      !!----------------------------------------------------------------------
1203      !!                  ***  routine mppscatter  ***
1204      !!
1205      !! ** Purpose :   Transfert between awork array which is distributed
1206      !!      following the vertical level and the local subdomain array.
1207      !!
1208      !!----------------------------------------------------------------------
1209      REAL(wp), DIMENSION(jpi,jpj,jpnij)  ::  pio        ! output array
1210      INTEGER                             ::   kp        ! Tag (not used with MPI
1211      REAL(wp), DIMENSION(jpi,jpj)        ::  ptab       ! subdomain array input
[1344]1212      !!
1213      INTEGER :: itaille, ierror   ! temporary integer
[51]1214      !!---------------------------------------------------------------------
[1344]1215      !
[51]1216      itaille=jpi*jpj
[1344]1217      !
1218      CALL mpi_scatter( pio, itaille, mpi_double_precision, ptab, itaille     ,   &
1219         &                            mpi_double_precision, kp  , mpi_comm_opa, ierror )
1220      !
[51]1221   END SUBROUTINE mppscatter
[3]1222
1223
[869]1224   SUBROUTINE mppmax_a_int( ktab, kdim, kcom )
[681]1225      !!----------------------------------------------------------------------
1226      !!                  ***  routine mppmax_a_int  ***
1227      !!
1228      !! ** Purpose :   Find maximum value in an integer layout array
1229      !!
1230      !!----------------------------------------------------------------------
[1344]1231      INTEGER , INTENT(in   )                  ::   kdim   ! size of array
1232      INTEGER , INTENT(inout), DIMENSION(kdim) ::   ktab   ! input array
1233      INTEGER , INTENT(in   ), OPTIONAL        ::   kcom   !
1234      !!
1235      INTEGER :: ierror, localcomm   ! temporary integer
[681]1236      INTEGER, DIMENSION(kdim) ::   iwork
[1344]1237      !!----------------------------------------------------------------------
1238      !
[869]1239      localcomm = mpi_comm_opa
[1344]1240      IF( PRESENT(kcom) )   localcomm = kcom
1241      !
1242      CALL mpi_allreduce( ktab, iwork, kdim, mpi_integer, mpi_max, localcomm, ierror )
1243      !
[681]1244      ktab(:) = iwork(:)
[1344]1245      !
[681]1246   END SUBROUTINE mppmax_a_int
1247
1248
[869]1249   SUBROUTINE mppmax_int( ktab, kcom )
[681]1250      !!----------------------------------------------------------------------
1251      !!                  ***  routine mppmax_int  ***
1252      !!
[1344]1253      !! ** Purpose :   Find maximum value in an integer layout array
[681]1254      !!
1255      !!----------------------------------------------------------------------
[1344]1256      INTEGER, INTENT(inout)           ::   ktab      ! ???
1257      INTEGER, INTENT(in   ), OPTIONAL ::   kcom      ! ???
1258      !!
1259      INTEGER ::   ierror, iwork, localcomm   ! temporary integer
1260      !!----------------------------------------------------------------------
1261      !
[869]1262      localcomm = mpi_comm_opa 
[1344]1263      IF( PRESENT(kcom) )   localcomm = kcom
1264      !
1265      CALL mpi_allreduce( ktab, iwork, 1, mpi_integer, mpi_max, localcomm, ierror)
1266      !
[681]1267      ktab = iwork
[1344]1268      !
[681]1269   END SUBROUTINE mppmax_int
1270
1271
[869]1272   SUBROUTINE mppmin_a_int( ktab, kdim, kcom )
[51]1273      !!----------------------------------------------------------------------
1274      !!                  ***  routine mppmin_a_int  ***
1275      !!
1276      !! ** Purpose :   Find minimum value in an integer layout array
1277      !!
1278      !!----------------------------------------------------------------------
1279      INTEGER , INTENT( in  )                  ::   kdim        ! size of array
1280      INTEGER , INTENT(inout), DIMENSION(kdim) ::   ktab        ! input array
[888]1281      INTEGER , INTENT( in  ), OPTIONAL        ::   kcom        ! input array
[1344]1282      !!
1283      INTEGER ::   ierror, localcomm   ! temporary integer
[51]1284      INTEGER, DIMENSION(kdim) ::   iwork
[1344]1285      !!----------------------------------------------------------------------
1286      !
[869]1287      localcomm = mpi_comm_opa
[1344]1288      IF( PRESENT(kcom) )   localcomm = kcom
1289      !
1290      CALL mpi_allreduce( ktab, iwork, kdim, mpi_integer, mpi_min, localcomm, ierror )
1291      !
[51]1292      ktab(:) = iwork(:)
[1344]1293      !
[51]1294   END SUBROUTINE mppmin_a_int
[3]1295
[13]1296
[1345]1297   SUBROUTINE mppmin_int( ktab, kcom )
[51]1298      !!----------------------------------------------------------------------
1299      !!                  ***  routine mppmin_int  ***
1300      !!
[1344]1301      !! ** Purpose :   Find minimum value in an integer layout array
[51]1302      !!
1303      !!----------------------------------------------------------------------
1304      INTEGER, INTENT(inout) ::   ktab      ! ???
[1345]1305      INTEGER , INTENT( in  ), OPTIONAL        ::   kcom        ! input array
[1344]1306      !!
[1345]1307      INTEGER ::  ierror, iwork, localcomm
[1344]1308      !!----------------------------------------------------------------------
1309      !
[1345]1310      localcomm = mpi_comm_opa
1311      IF( PRESENT(kcom) )   localcomm = kcom
[1344]1312      !
[1345]1313     CALL mpi_allreduce( ktab, iwork, 1, mpi_integer, mpi_min, localcomm, ierror )
1314      !
[51]1315      ktab = iwork
[1344]1316      !
[51]1317   END SUBROUTINE mppmin_int
[3]1318
[13]1319
[51]1320   SUBROUTINE mppsum_a_int( ktab, kdim )
1321      !!----------------------------------------------------------------------
1322      !!                  ***  routine mppsum_a_int  ***
1323      !!                   
[1344]1324      !! ** Purpose :   Global integer sum, 1D array case
[51]1325      !!
1326      !!----------------------------------------------------------------------
[1344]1327      INTEGER, INTENT(in   )                   ::   kdim      ! ???
[51]1328      INTEGER, INTENT(inout), DIMENSION (kdim) ::   ktab      ! ???
[1344]1329      !!
[51]1330      INTEGER :: ierror
1331      INTEGER, DIMENSION (kdim) ::  iwork
[1344]1332      !!----------------------------------------------------------------------
1333      !
1334      CALL mpi_allreduce( ktab, iwork, kdim, mpi_integer, mpi_sum, mpi_comm_opa, ierror )
1335      !
[51]1336      ktab(:) = iwork(:)
[1344]1337      !
[51]1338   END SUBROUTINE mppsum_a_int
[3]1339
[13]1340
[1344]1341   SUBROUTINE mppsum_int( ktab )
1342      !!----------------------------------------------------------------------
1343      !!                 ***  routine mppsum_int  ***
1344      !!                 
1345      !! ** Purpose :   Global integer sum
1346      !!
1347      !!----------------------------------------------------------------------
1348      INTEGER, INTENT(inout) ::   ktab
1349      !!
1350      INTEGER :: ierror, iwork
1351      !!----------------------------------------------------------------------
1352      !
1353      CALL mpi_allreduce( ktab, iwork, 1, mpi_integer, mpi_sum, mpi_comm_opa, ierror )
1354      !
1355      ktab = iwork
1356      !
1357   END SUBROUTINE mppsum_int
[3]1358
[13]1359
[1344]1360   SUBROUTINE mppmax_a_real( ptab, kdim, kcom )
1361      !!----------------------------------------------------------------------
1362      !!                 ***  routine mppmax_a_real  ***
1363      !!                 
1364      !! ** Purpose :   Maximum
1365      !!
1366      !!----------------------------------------------------------------------
1367      INTEGER , INTENT(in   )                  ::   kdim
1368      REAL(wp), INTENT(inout), DIMENSION(kdim) ::   ptab
1369      INTEGER , INTENT(in   ), OPTIONAL        ::   kcom
1370      !!
1371      INTEGER :: ierror, localcomm
1372      REAL(wp), DIMENSION(kdim) ::  zwork
1373      !!----------------------------------------------------------------------
1374      !
1375      localcomm = mpi_comm_opa
1376      IF( PRESENT(kcom) ) localcomm = kcom
1377      !
1378      CALL mpi_allreduce( ptab, zwork, kdim, mpi_double_precision, mpi_max, localcomm, ierror )
1379      ptab(:) = zwork(:)
1380      !
1381   END SUBROUTINE mppmax_a_real
[3]1382
1383
[1344]1384   SUBROUTINE mppmax_real( ptab, kcom )
1385      !!----------------------------------------------------------------------
1386      !!                  ***  routine mppmax_real  ***
1387      !!                   
1388      !! ** Purpose :   Maximum
1389      !!
1390      !!----------------------------------------------------------------------
1391      REAL(wp), INTENT(inout)           ::   ptab   ! ???
1392      INTEGER , INTENT(in   ), OPTIONAL ::   kcom   ! ???
1393      !!
1394      INTEGER  ::   ierror, localcomm
1395      REAL(wp) ::   zwork
1396      !!----------------------------------------------------------------------
1397      !
1398      localcomm = mpi_comm_opa 
1399      IF( PRESENT(kcom) )   localcomm = kcom
1400      !
1401      CALL mpi_allreduce( ptab, zwork, 1, mpi_double_precision, mpi_max, localcomm, ierror )
1402      ptab = zwork
1403      !
1404   END SUBROUTINE mppmax_real
[13]1405
[3]1406
[1344]1407   SUBROUTINE mppmin_a_real( ptab, kdim, kcom )
1408      !!----------------------------------------------------------------------
1409      !!                 ***  routine mppmin_a_real  ***
1410      !!                 
1411      !! ** Purpose :   Minimum of REAL, array case
1412      !!
1413      !!-----------------------------------------------------------------------
1414      INTEGER , INTENT(in   )                  ::   kdim
1415      REAL(wp), INTENT(inout), DIMENSION(kdim) ::   ptab
1416      INTEGER , INTENT(in   ), OPTIONAL        ::   kcom
1417      !!
1418      INTEGER :: ierror, localcomm
1419      REAL(wp), DIMENSION(kdim) ::   zwork
1420      !!-----------------------------------------------------------------------
1421      !
1422      localcomm = mpi_comm_opa 
1423      IF( PRESENT(kcom) ) localcomm = kcom
1424      !
1425      CALL mpi_allreduce( ptab, zwork, kdim, mpi_double_precision, mpi_min, localcomm, ierror )
1426      ptab(:) = zwork(:)
1427      !
1428   END SUBROUTINE mppmin_a_real
[3]1429
1430
[1344]1431   SUBROUTINE mppmin_real( ptab, kcom )
1432      !!----------------------------------------------------------------------
1433      !!                  ***  routine mppmin_real  ***
1434      !!
1435      !! ** Purpose :   minimum of REAL, scalar case
1436      !!
1437      !!-----------------------------------------------------------------------
1438      REAL(wp), INTENT(inout)           ::   ptab        !
1439      INTEGER , INTENT(in   ), OPTIONAL :: kcom
1440      !!
1441      INTEGER  ::   ierror
1442      REAL(wp) ::   zwork
1443      INTEGER :: localcomm
1444      !!-----------------------------------------------------------------------
1445      !
1446      localcomm = mpi_comm_opa 
1447      IF( PRESENT(kcom) )   localcomm = kcom
1448      !
1449      CALL mpi_allreduce( ptab, zwork, 1, mpi_double_precision, mpi_min, localcomm, ierror )
1450      ptab = zwork
1451      !
1452   END SUBROUTINE mppmin_real
[13]1453
[3]1454
[1344]1455   SUBROUTINE mppsum_a_real( ptab, kdim, kcom )
1456      !!----------------------------------------------------------------------
1457      !!                  ***  routine mppsum_a_real  ***
1458      !!
1459      !! ** Purpose :   global sum, REAL ARRAY argument case
1460      !!
1461      !!-----------------------------------------------------------------------
1462      INTEGER , INTENT( in )                     ::   kdim      ! size of ptab
1463      REAL(wp), DIMENSION(kdim), INTENT( inout ) ::   ptab      ! input array
1464      INTEGER , INTENT( in ), OPTIONAL           :: kcom
1465      !!
1466      INTEGER                   ::   ierror    ! temporary integer
1467      INTEGER                   ::   localcomm 
1468      REAL(wp), DIMENSION(kdim) ::   zwork     ! temporary workspace
1469      !!-----------------------------------------------------------------------
1470      !
1471      localcomm = mpi_comm_opa 
1472      IF( PRESENT(kcom) )   localcomm = kcom
1473      !
1474      CALL mpi_allreduce( ptab, zwork, kdim, mpi_double_precision, mpi_sum, localcomm, ierror )
1475      ptab(:) = zwork(:)
1476      !
1477   END SUBROUTINE mppsum_a_real
[869]1478
[3]1479
[1344]1480   SUBROUTINE mppsum_real( ptab, kcom )
1481      !!----------------------------------------------------------------------
1482      !!                  ***  routine mppsum_real  ***
1483      !!             
1484      !! ** Purpose :   global sum, SCALAR argument case
1485      !!
1486      !!-----------------------------------------------------------------------
1487      REAL(wp), INTENT(inout)           ::   ptab   ! input scalar
1488      INTEGER , INTENT(in   ), OPTIONAL ::   kcom
1489      !!
1490      INTEGER  ::   ierror, localcomm 
1491      REAL(wp) ::   zwork
1492      !!-----------------------------------------------------------------------
1493      !
1494      localcomm = mpi_comm_opa 
1495      IF( PRESENT(kcom) ) localcomm = kcom
1496      !
1497      CALL mpi_allreduce( ptab, zwork, 1, mpi_double_precision, mpi_sum, localcomm, ierror )
1498      ptab = zwork
1499      !
1500   END SUBROUTINE mppsum_real
[3]1501
[1976]1502   SUBROUTINE mppsum_realdd( ytab, kcom )
1503      !!----------------------------------------------------------------------
1504      !!                  ***  routine mppsum_realdd ***
1505      !!
1506      !! ** Purpose :   global sum in Massively Parallel Processing
1507      !!                SCALAR argument case for double-double precision
1508      !!
1509      !!-----------------------------------------------------------------------
1510      COMPLEX(wp), INTENT(inout)         :: ytab    ! input scalar
1511      INTEGER , INTENT( in  ), OPTIONAL :: kcom
[3]1512
[1976]1513      !! * Local variables   (MPI version)
1514      INTEGER  ::    ierror
1515      INTEGER  ::   localcomm
1516      COMPLEX(wp) :: zwork
1517
1518      localcomm = mpi_comm_opa
1519      IF( PRESENT(kcom) ) localcomm = kcom
1520
1521      ! reduce local sums into global sum
1522      CALL MPI_ALLREDUCE (ytab, zwork, 1, MPI_DOUBLE_COMPLEX, &
1523                       MPI_SUMDD,localcomm,ierror)
1524      ytab = zwork
1525
1526   END SUBROUTINE mppsum_realdd
1527 
1528 
1529   SUBROUTINE mppsum_a_realdd( ytab, kdim, kcom )
1530      !!----------------------------------------------------------------------
1531      !!                  ***  routine mppsum_a_realdd  ***
1532      !!
1533      !! ** Purpose :   global sum in Massively Parallel Processing
1534      !!                COMPLEX ARRAY case for double-double precision
1535      !!
1536      !!-----------------------------------------------------------------------
1537      INTEGER , INTENT( in )                        ::   kdim      ! size of ytab
1538      COMPLEX(wp), DIMENSION(kdim), INTENT( inout ) ::   ytab      ! input array
1539      INTEGER , INTENT( in  ), OPTIONAL :: kcom
1540
1541      !! * Local variables   (MPI version)
1542      INTEGER                      :: ierror    ! temporary integer
1543      INTEGER                      ::   localcomm
1544      COMPLEX(wp), DIMENSION(kdim) :: zwork     ! temporary workspace
1545
1546      localcomm = mpi_comm_opa
1547      IF( PRESENT(kcom) ) localcomm = kcom
1548
1549      CALL MPI_ALLREDUCE (ytab, zwork, kdim, MPI_DOUBLE_COMPLEX, &
1550                       MPI_SUMDD,localcomm,ierror)
1551      ytab(:) = zwork(:)
1552
1553   END SUBROUTINE mppsum_a_realdd
1554   
[1344]1555   SUBROUTINE mpp_minloc2d( ptab, pmask, pmin, ki,kj )
1556      !!------------------------------------------------------------------------
1557      !!             ***  routine mpp_minloc  ***
1558      !!
1559      !! ** Purpose :   Compute the global minimum of an array ptab
1560      !!              and also give its global position
1561      !!
1562      !! ** Method  :   Use MPI_ALLREDUCE with MPI_MINLOC
1563      !!
1564      !!--------------------------------------------------------------------------
1565      REAL(wp), DIMENSION (jpi,jpj), INTENT(in   ) ::   ptab    ! Local 2D array
1566      REAL(wp), DIMENSION (jpi,jpj), INTENT(in   ) ::   pmask   ! Local mask
1567      REAL(wp)                     , INTENT(  out) ::   pmin    ! Global minimum of ptab
1568      INTEGER                      , INTENT(  out) ::   ki, kj   ! index of minimum in global frame
1569      !!
1570      INTEGER , DIMENSION(2)   ::   ilocs
1571      INTEGER :: ierror
1572      REAL(wp) ::   zmin   ! local minimum
1573      REAL(wp), DIMENSION(2,1) ::   zain, zaout
1574      !!-----------------------------------------------------------------------
1575      !
1576      zmin  = MINVAL( ptab(:,:) , mask= pmask == 1.e0 )
1577      ilocs = MINLOC( ptab(:,:) , mask= pmask == 1.e0 )
1578      !
1579      ki = ilocs(1) + nimpp - 1
1580      kj = ilocs(2) + njmpp - 1
1581      !
1582      zain(1,:)=zmin
1583      zain(2,:)=ki+10000.*kj
1584      !
1585      CALL MPI_ALLREDUCE( zain,zaout, 1, MPI_2DOUBLE_PRECISION,MPI_MINLOC,MPI_COMM_OPA,ierror)
1586      !
1587      pmin = zaout(1,1)
1588      kj = INT(zaout(2,1)/10000.)
1589      ki = INT(zaout(2,1) - 10000.*kj )
1590      !
1591   END SUBROUTINE mpp_minloc2d
[13]1592
[3]1593
[1344]1594   SUBROUTINE mpp_minloc3d( ptab, pmask, pmin, ki, kj ,kk)
1595      !!------------------------------------------------------------------------
1596      !!             ***  routine mpp_minloc  ***
1597      !!
1598      !! ** Purpose :   Compute the global minimum of an array ptab
1599      !!              and also give its global position
1600      !!
1601      !! ** Method  :   Use MPI_ALLREDUCE with MPI_MINLOC
1602      !!
1603      !!--------------------------------------------------------------------------
1604      REAL(wp), DIMENSION (jpi,jpj,jpk), INTENT(in   ) ::   ptab         ! Local 2D array
1605      REAL(wp), DIMENSION (jpi,jpj,jpk), INTENT(in   ) ::   pmask        ! Local mask
1606      REAL(wp)                         , INTENT(  out) ::   pmin         ! Global minimum of ptab
1607      INTEGER                          , INTENT(  out) ::   ki, kj, kk   ! index of minimum in global frame
1608      !!
1609      INTEGER  ::   ierror
1610      REAL(wp) ::   zmin     ! local minimum
1611      INTEGER , DIMENSION(3)   ::   ilocs
1612      REAL(wp), DIMENSION(2,1) ::   zain, zaout
1613      !!-----------------------------------------------------------------------
1614      !
1615      zmin  = MINVAL( ptab(:,:,:) , mask= pmask == 1.e0 )
1616      ilocs = MINLOC( ptab(:,:,:) , mask= pmask == 1.e0 )
1617      !
1618      ki = ilocs(1) + nimpp - 1
1619      kj = ilocs(2) + njmpp - 1
1620      kk = ilocs(3)
1621      !
1622      zain(1,:)=zmin
1623      zain(2,:)=ki+10000.*kj+100000000.*kk
1624      !
1625      CALL MPI_ALLREDUCE( zain,zaout, 1, MPI_2DOUBLE_PRECISION,MPI_MINLOC,MPI_COMM_OPA,ierror)
1626      !
1627      pmin = zaout(1,1)
1628      kk   = INT( zaout(2,1) / 100000000. )
1629      kj   = INT( zaout(2,1) - kk * 100000000. ) / 10000
1630      ki   = INT( zaout(2,1) - kk * 100000000. -kj * 10000. )
1631      !
1632   END SUBROUTINE mpp_minloc3d
[13]1633
[3]1634
[1344]1635   SUBROUTINE mpp_maxloc2d( ptab, pmask, pmax, ki, kj )
1636      !!------------------------------------------------------------------------
1637      !!             ***  routine mpp_maxloc  ***
1638      !!
1639      !! ** Purpose :   Compute the global maximum of an array ptab
1640      !!              and also give its global position
1641      !!
1642      !! ** Method  :   Use MPI_ALLREDUCE with MPI_MINLOC
1643      !!
1644      !!--------------------------------------------------------------------------
1645      REAL(wp), DIMENSION (jpi,jpj), INTENT(in   ) ::   ptab     ! Local 2D array
1646      REAL(wp), DIMENSION (jpi,jpj), INTENT(in   ) ::   pmask    ! Local mask
1647      REAL(wp)                     , INTENT(  out) ::   pmax     ! Global maximum of ptab
1648      INTEGER                      , INTENT(  out) ::   ki, kj   ! index of maximum in global frame
1649      !! 
1650      INTEGER  :: ierror
1651      INTEGER, DIMENSION (2)   ::   ilocs
1652      REAL(wp) :: zmax   ! local maximum
1653      REAL(wp), DIMENSION(2,1) ::   zain, zaout
1654      !!-----------------------------------------------------------------------
1655      !
1656      zmax  = MAXVAL( ptab(:,:) , mask= pmask == 1.e0 )
1657      ilocs = MAXLOC( ptab(:,:) , mask= pmask == 1.e0 )
1658      !
1659      ki = ilocs(1) + nimpp - 1
1660      kj = ilocs(2) + njmpp - 1
1661      !
1662      zain(1,:) = zmax
1663      zain(2,:) = ki + 10000. * kj
1664      !
1665      CALL MPI_ALLREDUCE( zain,zaout, 1, MPI_2DOUBLE_PRECISION,MPI_MAXLOC,MPI_COMM_OPA,ierror)
1666      !
1667      pmax = zaout(1,1)
1668      kj   = INT( zaout(2,1) / 10000.     )
1669      ki   = INT( zaout(2,1) - 10000.* kj )
1670      !
1671   END SUBROUTINE mpp_maxloc2d
[3]1672
[13]1673
[1344]1674   SUBROUTINE mpp_maxloc3d( ptab, pmask, pmax, ki, kj, kk )
1675      !!------------------------------------------------------------------------
1676      !!             ***  routine mpp_maxloc  ***
1677      !!
1678      !! ** Purpose :  Compute the global maximum of an array ptab
1679      !!              and also give its global position
1680      !!
1681      !! ** Method : Use MPI_ALLREDUCE with MPI_MINLOC
1682      !!
1683      !!--------------------------------------------------------------------------
1684      REAL(wp), DIMENSION (jpi,jpj,jpk), INTENT(in   ) ::   ptab         ! Local 2D array
1685      REAL(wp), DIMENSION (jpi,jpj,jpk), INTENT(in   ) ::   pmask        ! Local mask
1686      REAL(wp)                         , INTENT(  out) ::   pmax         ! Global maximum of ptab
1687      INTEGER                          , INTENT(  out) ::   ki, kj, kk   ! index of maximum in global frame
1688      !!   
1689      REAL(wp) :: zmax   ! local maximum
1690      REAL(wp), DIMENSION(2,1) ::   zain, zaout
1691      INTEGER , DIMENSION(3)   ::   ilocs
1692      INTEGER :: ierror
1693      !!-----------------------------------------------------------------------
1694      !
1695      zmax  = MAXVAL( ptab(:,:,:) , mask= pmask == 1.e0 )
1696      ilocs = MAXLOC( ptab(:,:,:) , mask= pmask == 1.e0 )
1697      !
1698      ki = ilocs(1) + nimpp - 1
1699      kj = ilocs(2) + njmpp - 1
1700      kk = ilocs(3)
1701      !
1702      zain(1,:)=zmax
1703      zain(2,:)=ki+10000.*kj+100000000.*kk
1704      !
1705      CALL MPI_ALLREDUCE( zain,zaout, 1, MPI_2DOUBLE_PRECISION,MPI_MAXLOC,MPI_COMM_OPA,ierror)
1706      !
1707      pmax = zaout(1,1)
1708      kk   = INT( zaout(2,1) / 100000000. )
1709      kj   = INT( zaout(2,1) - kk * 100000000. ) / 10000
1710      ki   = INT( zaout(2,1) - kk * 100000000. -kj * 10000. )
1711      !
1712   END SUBROUTINE mpp_maxloc3d
[3]1713
[869]1714
[1344]1715   SUBROUTINE mppsync()
1716      !!----------------------------------------------------------------------
1717      !!                  ***  routine mppsync  ***
1718      !!                   
1719      !! ** Purpose :   Massively parallel processors, synchroneous
1720      !!
1721      !!-----------------------------------------------------------------------
1722      INTEGER :: ierror
1723      !!-----------------------------------------------------------------------
1724      !
1725      CALL mpi_barrier( mpi_comm_opa, ierror )
1726      !
1727   END SUBROUTINE mppsync
[3]1728
1729
[1344]1730   SUBROUTINE mppstop
1731      !!----------------------------------------------------------------------
1732      !!                  ***  routine mppstop  ***
1733      !!                   
[3294]1734      !! ** purpose :   Stop massively parallel processors method
[1344]1735      !!
1736      !!----------------------------------------------------------------------
1737      INTEGER ::   info
1738      !!----------------------------------------------------------------------
1739      !
1740      CALL mppsync
1741      CALL mpi_finalize( info )
1742      !
1743   END SUBROUTINE mppstop
[3]1744
1745
[2715]1746   SUBROUTINE mppobc( ptab, kd1, kd2, kl, kk, ktype, kij , kumout)
[1344]1747      !!----------------------------------------------------------------------
1748      !!                  ***  routine mppobc  ***
1749      !!
1750      !! ** Purpose :   Message passing manadgement for open boundary
1751      !!     conditions array
1752      !!
1753      !! ** Method  :   Use mppsend and mpprecv function for passing mask
1754      !!       between processors following neighboring subdomains.
1755      !!       domain parameters
1756      !!                    nlci   : first dimension of the local subdomain
1757      !!                    nlcj   : second dimension of the local subdomain
1758      !!                    nbondi : mark for "east-west local boundary"
1759      !!                    nbondj : mark for "north-south local boundary"
1760      !!                    noea   : number for local neighboring processors
1761      !!                    nowe   : number for local neighboring processors
1762      !!                    noso   : number for local neighboring processors
1763      !!                    nono   : number for local neighboring processors
1764      !!
1765      !!----------------------------------------------------------------------
[3294]1766      USE wrk_nemo        ! Memory allocation
[2715]1767      !
[1344]1768      INTEGER , INTENT(in   )                     ::   kd1, kd2   ! starting and ending indices
1769      INTEGER , INTENT(in   )                     ::   kl         ! index of open boundary
1770      INTEGER , INTENT(in   )                     ::   kk         ! vertical dimension
1771      INTEGER , INTENT(in   )                     ::   ktype      ! define north/south or east/west cdt
1772      !                                                           !  = 1  north/south  ;  = 2  east/west
1773      INTEGER , INTENT(in   )                     ::   kij        ! horizontal dimension
[2715]1774      INTEGER , INTENT(in   )                     ::   kumout     ! ocean.output logical unit
[1344]1775      REAL(wp), INTENT(inout), DIMENSION(kij,kk)  ::   ptab       ! variable array
[2715]1776      !
1777      INTEGER ::   ji, jj, jk, jl        ! dummy loop indices
1778      INTEGER ::   iipt0, iipt1, ilpt1   ! local integers
1779      INTEGER ::   ijpt0, ijpt1          !   -       -
1780      INTEGER ::   imigr, iihom, ijhom   !   -       -
[1344]1781      INTEGER ::   ml_req1, ml_req2, ml_err    ! for key_mpi_isend
1782      INTEGER ::   ml_stat(MPI_STATUS_SIZE)    ! for key_mpi_isend
[3294]1783      REAL(wp), POINTER, DIMENSION(:,:) ::   ztab   ! temporary workspace
[1344]1784      !!----------------------------------------------------------------------
[3]1785
[3294]1786      CALL wrk_alloc( jpi,jpj, ztab )
[2715]1787
[1344]1788      ! boundary condition initialization
1789      ! ---------------------------------
1790      ztab(:,:) = 0.e0
1791      !
1792      IF( ktype==1 ) THEN                                  ! north/south boundaries
1793         iipt0 = MAX( 1, MIN(kd1 - nimpp+1, nlci     ) )
1794         iipt1 = MAX( 0, MIN(kd2 - nimpp+1, nlci - 1 ) )
1795         ilpt1 = MAX( 1, MIN(kd2 - nimpp+1, nlci     ) )
1796         ijpt0 = MAX( 1, MIN(kl  - njmpp+1, nlcj     ) )
1797         ijpt1 = MAX( 0, MIN(kl  - njmpp+1, nlcj - 1 ) )
1798      ELSEIF( ktype==2 ) THEN                              ! east/west boundaries
1799         iipt0 = MAX( 1, MIN(kl  - nimpp+1, nlci     ) )
1800         iipt1 = MAX( 0, MIN(kl  - nimpp+1, nlci - 1 ) )
1801         ijpt0 = MAX( 1, MIN(kd1 - njmpp+1, nlcj     ) )
1802         ijpt1 = MAX( 0, MIN(kd2 - njmpp+1, nlcj - 1 ) )
1803         ilpt1 = MAX( 1, MIN(kd2 - njmpp+1, nlcj     ) )
1804      ELSE
[2715]1805         WRITE(kumout, cform_err)
1806         WRITE(kumout,*) 'mppobc : bad ktype'
1807         CALL mppstop
[1344]1808      ENDIF
1809     
1810      ! Communication level by level
1811      ! ----------------------------
1812!!gm Remark : this is very time consumming!!!
1813      !                                         ! ------------------------ !
1814      DO jk = 1, kk                             !   Loop over the levels   !
1815         !                                      ! ------------------------ !
1816         !
1817         IF( ktype == 1 ) THEN                               ! north/south boundaries
1818            DO jj = ijpt0, ijpt1
1819               DO ji = iipt0, iipt1
1820                  ztab(ji,jj) = ptab(ji,jk)
1821               END DO
1822            END DO
1823         ELSEIF( ktype == 2 ) THEN                           ! east/west boundaries
1824            DO jj = ijpt0, ijpt1
1825               DO ji = iipt0, iipt1
1826                  ztab(ji,jj) = ptab(jj,jk)
1827               END DO
1828            END DO
1829         ENDIF
[13]1830
[3]1831
[1344]1832         ! 1. East and west directions
1833         ! ---------------------------
1834         !
1835         IF( nbondi /= 2 ) THEN         ! Read Dirichlet lateral conditions
1836            iihom = nlci-nreci
1837            DO jl = 1, jpreci
1838               t2ew(:,jl,1) = ztab(jpreci+jl,:)
1839               t2we(:,jl,1) = ztab(iihom +jl,:)
1840            END DO
1841         ENDIF
1842         !
1843         !                              ! Migrations
1844         imigr=jpreci*jpj
1845         !
1846         IF( nbondi == -1 ) THEN
1847            CALL mppsend( 2, t2we(1,1,1), imigr, noea, ml_req1 )
[3294]1848            CALL mpprecv( 1, t2ew(1,1,2), imigr, noea )
[1344]1849            IF(l_isend)   CALL mpi_wait( ml_req1, ml_stat, ml_err )
1850         ELSEIF( nbondi == 0 ) THEN
1851            CALL mppsend( 1, t2ew(1,1,1), imigr, nowe, ml_req1 )
1852            CALL mppsend( 2, t2we(1,1,1), imigr, noea, ml_req2 )
[3294]1853            CALL mpprecv( 1, t2ew(1,1,2), imigr, noea )
1854            CALL mpprecv( 2, t2we(1,1,2), imigr, nowe )
[1344]1855            IF(l_isend)   CALL mpi_wait( ml_req1, ml_stat, ml_err )
1856            IF(l_isend)   CALL mpi_wait( ml_req2, ml_stat, ml_err )
1857         ELSEIF( nbondi == 1 ) THEN
1858            CALL mppsend( 1, t2ew(1,1,1), imigr, nowe, ml_req1 )
[3294]1859            CALL mpprecv( 2, t2we(1,1,2), imigr, nowe )
[1344]1860            IF(l_isend) CALL mpi_wait( ml_req1, ml_stat, ml_err )
1861         ENDIF
1862         !
1863         !                              ! Write Dirichlet lateral conditions
1864         iihom = nlci-jpreci
1865         !
1866         IF( nbondi == 0 .OR. nbondi == 1 ) THEN
1867            DO jl = 1, jpreci
1868               ztab(jl,:) = t2we(:,jl,2)
1869            END DO
1870         ENDIF
1871         IF( nbondi == -1 .OR. nbondi == 0 ) THEN
1872            DO jl = 1, jpreci
1873               ztab(iihom+jl,:) = t2ew(:,jl,2)
1874            END DO
1875         ENDIF
[3]1876
1877
[1344]1878         ! 2. North and south directions
1879         ! -----------------------------
1880         !
1881         IF( nbondj /= 2 ) THEN         ! Read Dirichlet lateral conditions
1882            ijhom = nlcj-nrecj
1883            DO jl = 1, jprecj
1884               t2sn(:,jl,1) = ztab(:,ijhom +jl)
1885               t2ns(:,jl,1) = ztab(:,jprecj+jl)
1886            END DO
1887         ENDIF
1888         !
1889         !                              ! Migrations
1890         imigr = jprecj * jpi
1891         !
1892         IF( nbondj == -1 ) THEN
1893            CALL mppsend( 4, t2sn(1,1,1), imigr, nono, ml_req1 )
[3294]1894            CALL mpprecv( 3, t2ns(1,1,2), imigr, nono )
[1344]1895            IF(l_isend) CALL mpi_wait( ml_req1, ml_stat, ml_err )
1896         ELSEIF( nbondj == 0 ) THEN
1897            CALL mppsend( 3, t2ns(1,1,1), imigr, noso, ml_req1 )
1898            CALL mppsend( 4, t2sn(1,1,1), imigr, nono, ml_req2 )
[3294]1899            CALL mpprecv( 3, t2ns(1,1,2), imigr, nono )
1900            CALL mpprecv( 4, t2sn(1,1,2), imigr, noso )
[1344]1901            IF( l_isend )   CALL mpi_wait( ml_req1, ml_stat, ml_err )
1902            IF( l_isend )   CALL mpi_wait( ml_req2, ml_stat, ml_err )
1903         ELSEIF( nbondj == 1 ) THEN
1904            CALL mppsend( 3, t2ns(1,1,1), imigr, noso, ml_req1 )
[3294]1905            CALL mpprecv( 4, t2sn(1,1,2), imigr, noso)
[1344]1906            IF( l_isend )   CALL mpi_wait( ml_req1, ml_stat, ml_err )
1907         ENDIF
1908         !
1909         !                              ! Write Dirichlet lateral conditions
1910         ijhom = nlcj - jprecj
1911         IF( nbondj == 0 .OR. nbondj == 1 ) THEN
1912            DO jl = 1, jprecj
1913               ztab(:,jl) = t2sn(:,jl,2)
1914            END DO
1915         ENDIF
1916         IF( nbondj == 0 .OR. nbondj == -1 ) THEN
1917            DO jl = 1, jprecj
1918               ztab(:,ijhom+jl) = t2ns(:,jl,2)
1919            END DO
1920         ENDIF
1921         IF( ktype==1 .AND. kd1 <= jpi+nimpp-1 .AND. nimpp <= kd2 ) THEN
1922            DO jj = ijpt0, ijpt1            ! north/south boundaries
1923               DO ji = iipt0,ilpt1
1924                  ptab(ji,jk) = ztab(ji,jj) 
1925               END DO
1926            END DO
1927         ELSEIF( ktype==2 .AND. kd1 <= jpj+njmpp-1 .AND. njmpp <= kd2 ) THEN
1928            DO jj = ijpt0, ilpt1            ! east/west boundaries
1929               DO ji = iipt0,iipt1
1930                  ptab(jj,jk) = ztab(ji,jj) 
1931               END DO
1932            END DO
1933         ENDIF
1934         !
1935      END DO
1936      !
[3294]1937      CALL wrk_dealloc( jpi,jpj, ztab )
[2715]1938      !
[1344]1939   END SUBROUTINE mppobc
1940   
[13]1941
[1344]1942   SUBROUTINE mpp_comm_free( kcom )
1943      !!----------------------------------------------------------------------
1944      !!----------------------------------------------------------------------
1945      INTEGER, INTENT(in) ::   kcom
1946      !!
1947      INTEGER :: ierr
1948      !!----------------------------------------------------------------------
1949      !
1950      CALL MPI_COMM_FREE(kcom, ierr)
1951      !
1952   END SUBROUTINE mpp_comm_free
[3]1953
[869]1954
[2715]1955   SUBROUTINE mpp_ini_ice( pindic, kumout )
[1344]1956      !!----------------------------------------------------------------------
1957      !!               ***  routine mpp_ini_ice  ***
1958      !!
1959      !! ** Purpose :   Initialize special communicator for ice areas
1960      !!      condition together with global variables needed in the ddmpp folding
1961      !!
1962      !! ** Method  : - Look for ice processors in ice routines
1963      !!              - Put their number in nrank_ice
1964      !!              - Create groups for the world processors and the ice processors
1965      !!              - Create a communicator for ice processors
1966      !!
1967      !! ** output
1968      !!      njmppmax = njmpp for northern procs
1969      !!      ndim_rank_ice = number of processors with ice
1970      !!      nrank_ice (ndim_rank_ice) = ice processors
[3625]1971      !!      ngrp_iworld = group ID for the world processors
[1344]1972      !!      ngrp_ice = group ID for the ice processors
1973      !!      ncomm_ice = communicator for the ice procs.
1974      !!      n_ice_root = number (in the world) of proc 0 in the ice comm.
1975      !!
1976      !!----------------------------------------------------------------------
[2715]1977      INTEGER, INTENT(in) ::   pindic
1978      INTEGER, INTENT(in) ::   kumout   ! ocean.output logical unit
[1344]1979      !!
1980      INTEGER :: jjproc
[2715]1981      INTEGER :: ii, ierr
1982      INTEGER, ALLOCATABLE, DIMENSION(:) ::   kice
1983      INTEGER, ALLOCATABLE, DIMENSION(:) ::   zwork
[1344]1984      !!----------------------------------------------------------------------
1985      !
[2715]1986      ! Since this is just an init routine and these arrays are of length jpnij
1987      ! then don't use wrk_nemo module - just allocate and deallocate.
1988      ALLOCATE( kice(jpnij), zwork(jpnij), STAT=ierr )
1989      IF( ierr /= 0 ) THEN
1990         WRITE(kumout, cform_err)
1991         WRITE(kumout,*) 'mpp_ini_ice : failed to allocate 2, 1D arrays (jpnij in length)'
1992         CALL mppstop
1993      ENDIF
1994
[1344]1995      ! Look for how many procs with sea-ice
1996      !
1997      kice = 0
1998      DO jjproc = 1, jpnij
1999         IF( jjproc == narea .AND. pindic .GT. 0 )   kice(jjproc) = 1   
2000      END DO
2001      !
2002      zwork = 0
2003      CALL MPI_ALLREDUCE( kice, zwork, jpnij, mpi_integer, mpi_sum, mpi_comm_opa, ierr )
2004      ndim_rank_ice = SUM( zwork )         
[3]2005
[1344]2006      ! Allocate the right size to nrank_north
[1441]2007      IF( ALLOCATED ( nrank_ice ) )   DEALLOCATE( nrank_ice )
[1344]2008      ALLOCATE( nrank_ice(ndim_rank_ice) )
2009      !
2010      ii = 0     
2011      nrank_ice = 0
2012      DO jjproc = 1, jpnij
2013         IF( zwork(jjproc) == 1) THEN
2014            ii = ii + 1
2015            nrank_ice(ii) = jjproc -1 
2016         ENDIF
2017      END DO
[1208]2018
[1344]2019      ! Create the world group
[3625]2020      CALL MPI_COMM_GROUP( mpi_comm_opa, ngrp_iworld, ierr )
[869]2021
[1344]2022      ! Create the ice group from the world group
[3625]2023      CALL MPI_GROUP_INCL( ngrp_iworld, ndim_rank_ice, nrank_ice, ngrp_ice, ierr )
[869]2024
[1344]2025      ! Create the ice communicator , ie the pool of procs with sea-ice
2026      CALL MPI_COMM_CREATE( mpi_comm_opa, ngrp_ice, ncomm_ice, ierr )
[869]2027
[1344]2028      ! Find proc number in the world of proc 0 in the north
2029      ! The following line seems to be useless, we just comment & keep it as reminder
[3625]2030      ! CALL MPI_GROUP_TRANSLATE_RANKS(ngrp_ice,1,0,ngrp_iworld,n_ice_root,ierr)
[1344]2031      !
[3625]2032      CALL MPI_GROUP_FREE(ngrp_ice, ierr)
2033      CALL MPI_GROUP_FREE(ngrp_iworld, ierr)
2034
[2715]2035      DEALLOCATE(kice, zwork)
2036      !
[1344]2037   END SUBROUTINE mpp_ini_ice
[869]2038
2039
[2715]2040   SUBROUTINE mpp_ini_znl( kumout )
[1345]2041      !!----------------------------------------------------------------------
2042      !!               ***  routine mpp_ini_znl  ***
2043      !!
2044      !! ** Purpose :   Initialize special communicator for computing zonal sum
2045      !!
2046      !! ** Method  : - Look for processors in the same row
2047      !!              - Put their number in nrank_znl
2048      !!              - Create group for the znl processors
2049      !!              - Create a communicator for znl processors
2050      !!              - Determine if processor should write znl files
2051      !!
2052      !! ** output
2053      !!      ndim_rank_znl = number of processors on the same row
2054      !!      ngrp_znl = group ID for the znl processors
2055      !!      ncomm_znl = communicator for the ice procs.
2056      !!      n_znl_root = number (in the world) of proc 0 in the ice comm.
2057      !!
2058      !!----------------------------------------------------------------------
[2715]2059      INTEGER, INTENT(in) ::   kumout   ! ocean.output logical units
[1345]2060      !
[2715]2061      INTEGER :: jproc      ! dummy loop integer
2062      INTEGER :: ierr, ii   ! local integer
2063      INTEGER, ALLOCATABLE, DIMENSION(:) ::   kwork
2064      !!----------------------------------------------------------------------
[1345]2065      !-$$     WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ngrp_world     : ', ngrp_world
2066      !-$$     WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - mpi_comm_world : ', mpi_comm_world
2067      !-$$     WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - mpi_comm_opa   : ', mpi_comm_opa
2068      !
[2715]2069      ALLOCATE( kwork(jpnij), STAT=ierr )
2070      IF( ierr /= 0 ) THEN
2071         WRITE(kumout, cform_err)
2072         WRITE(kumout,*) 'mpp_ini_znl : failed to allocate 1D array of length jpnij'
2073         CALL mppstop
2074      ENDIF
2075
2076      IF( jpnj == 1 ) THEN
[1345]2077         ngrp_znl  = ngrp_world
2078         ncomm_znl = mpi_comm_opa
2079      ELSE
2080         !
2081         CALL MPI_ALLGATHER ( njmpp, 1, mpi_integer, kwork, 1, mpi_integer, mpi_comm_opa, ierr )
2082         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - kwork pour njmpp : ', kwork
2083         !-$$        CALL flush(numout)
2084         !
2085         ! Count number of processors on the same row
2086         ndim_rank_znl = 0
2087         DO jproc=1,jpnij
2088            IF ( kwork(jproc) == njmpp ) THEN
2089               ndim_rank_znl = ndim_rank_znl + 1
2090            ENDIF
2091         END DO
2092         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ndim_rank_znl : ', ndim_rank_znl
2093         !-$$        CALL flush(numout)
2094         ! Allocate the right size to nrank_znl
[1441]2095         IF (ALLOCATED (nrank_znl)) DEALLOCATE(nrank_znl)
[1345]2096         ALLOCATE(nrank_znl(ndim_rank_znl))
2097         ii = 0     
2098         nrank_znl (:) = 0
2099         DO jproc=1,jpnij
2100            IF ( kwork(jproc) == njmpp) THEN
2101               ii = ii + 1
2102               nrank_znl(ii) = jproc -1 
2103            ENDIF
2104         END DO
2105         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - nrank_znl : ', nrank_znl
2106         !-$$        CALL flush(numout)
2107
2108         ! Create the opa group
2109         CALL MPI_COMM_GROUP(mpi_comm_opa,ngrp_opa,ierr)
2110         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ngrp_opa : ', ngrp_opa
2111         !-$$        CALL flush(numout)
2112
2113         ! Create the znl group from the opa group
2114         CALL MPI_GROUP_INCL  ( ngrp_opa, ndim_rank_znl, nrank_znl, ngrp_znl, ierr )
2115         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ngrp_znl ', ngrp_znl
2116         !-$$        CALL flush(numout)
2117
2118         ! Create the znl communicator from the opa communicator, ie the pool of procs in the same row
2119         CALL MPI_COMM_CREATE ( mpi_comm_opa, ngrp_znl, ncomm_znl, ierr )
2120         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ncomm_znl ', ncomm_znl
2121         !-$$        CALL flush(numout)
2122         !
2123      END IF
2124
2125      ! Determines if processor if the first (starting from i=1) on the row
2126      IF ( jpni == 1 ) THEN
2127         l_znl_root = .TRUE.
2128      ELSE
2129         l_znl_root = .FALSE.
2130         kwork (1) = nimpp
2131         CALL mpp_min ( kwork(1), kcom = ncomm_znl)
2132         IF ( nimpp == kwork(1)) l_znl_root = .TRUE.
2133      END IF
2134
[2715]2135      DEALLOCATE(kwork)
2136
[1345]2137   END SUBROUTINE mpp_ini_znl
2138
2139
[1344]2140   SUBROUTINE mpp_ini_north
2141      !!----------------------------------------------------------------------
2142      !!               ***  routine mpp_ini_north  ***
2143      !!
2144      !! ** Purpose :   Initialize special communicator for north folding
2145      !!      condition together with global variables needed in the mpp folding
2146      !!
2147      !! ** Method  : - Look for northern processors
2148      !!              - Put their number in nrank_north
2149      !!              - Create groups for the world processors and the north processors
2150      !!              - Create a communicator for northern processors
2151      !!
2152      !! ** output
2153      !!      njmppmax = njmpp for northern procs
2154      !!      ndim_rank_north = number of processors in the northern line
2155      !!      nrank_north (ndim_rank_north) = number  of the northern procs.
2156      !!      ngrp_world = group ID for the world processors
2157      !!      ngrp_north = group ID for the northern processors
2158      !!      ncomm_north = communicator for the northern procs.
2159      !!      north_root = number (in the world) of proc 0 in the northern comm.
2160      !!
2161      !!----------------------------------------------------------------------
2162      INTEGER ::   ierr
2163      INTEGER ::   jjproc
2164      INTEGER ::   ii, ji
2165      !!----------------------------------------------------------------------
2166      !
2167      njmppmax = MAXVAL( njmppt )
2168      !
2169      ! Look for how many procs on the northern boundary
2170      ndim_rank_north = 0
2171      DO jjproc = 1, jpnij
2172         IF( njmppt(jjproc) == njmppmax )   ndim_rank_north = ndim_rank_north + 1
2173      END DO
2174      !
2175      ! Allocate the right size to nrank_north
[1441]2176      IF (ALLOCATED (nrank_north)) DEALLOCATE(nrank_north)
[1344]2177      ALLOCATE( nrank_north(ndim_rank_north) )
[869]2178
[1344]2179      ! Fill the nrank_north array with proc. number of northern procs.
2180      ! Note : the rank start at 0 in MPI
2181      ii = 0
2182      DO ji = 1, jpnij
2183         IF ( njmppt(ji) == njmppmax   ) THEN
2184            ii=ii+1
2185            nrank_north(ii)=ji-1
2186         END IF
2187      END DO
2188      !
2189      ! create the world group
2190      CALL MPI_COMM_GROUP( mpi_comm_opa, ngrp_world, ierr )
2191      !
2192      ! Create the North group from the world group
2193      CALL MPI_GROUP_INCL( ngrp_world, ndim_rank_north, nrank_north, ngrp_north, ierr )
2194      !
2195      ! Create the North communicator , ie the pool of procs in the north group
2196      CALL MPI_COMM_CREATE( mpi_comm_opa, ngrp_north, ncomm_north, ierr )
2197      !
2198   END SUBROUTINE mpp_ini_north
[869]2199
2200
[1344]2201   SUBROUTINE mpp_lbc_north_3d( pt3d, cd_type, psgn )
[51]2202      !!---------------------------------------------------------------------
2203      !!                   ***  routine mpp_lbc_north_3d  ***
2204      !!
[1344]2205      !! ** Purpose :   Ensure proper north fold horizontal bondary condition
2206      !!              in mpp configuration in case of jpn1 > 1
[51]2207      !!
[1344]2208      !! ** Method  :   North fold condition and mpp with more than one proc
2209      !!              in i-direction require a specific treatment. We gather
2210      !!              the 4 northern lines of the global domain on 1 processor
2211      !!              and apply lbc north-fold on this sub array. Then we
2212      !!              scatter the north fold array back to the processors.
[51]2213      !!
2214      !!----------------------------------------------------------------------
[1344]2215      REAL(wp), DIMENSION(jpi,jpj,jpk), INTENT(inout) ::   pt3d      ! 3D array on which the b.c. is applied
2216      CHARACTER(len=1)                , INTENT(in   ) ::   cd_type   ! nature of pt3d grid-points
2217      !                                                              !   = T ,  U , V , F or W  gridpoints
2218      REAL(wp)                        , INTENT(in   ) ::   psgn      ! = -1. the sign change across the north fold
2219      !!                                                             ! =  1. , the sign is kept
2220      INTEGER ::   ji, jj, jr
2221      INTEGER ::   ierr, itaille, ildi, ilei, iilb
2222      INTEGER ::   ijpj, ijpjm1, ij, iproc
[3294]2223      INTEGER, DIMENSION (jpmaxngh)          ::   ml_req_nf          ! for mpi_isend when avoiding mpi_allgather
2224      INTEGER                                ::   ml_err             ! for mpi_isend when avoiding mpi_allgather
2225      INTEGER, DIMENSION(MPI_STATUS_SIZE)    ::   ml_stat            ! for mpi_isend when avoiding mpi_allgather
[51]2226      !!----------------------------------------------------------------------
[1344]2227      !   
2228      ijpj   = 4
[3294]2229      ityp = -1
[1344]2230      ijpjm1 = 3
[2480]2231      ztab(:,:,:) = 0.e0
[1344]2232      !
2233      DO jj = nlcj - ijpj +1, nlcj          ! put in znorthloc the last 4 jlines of pt3d
2234         ij = jj - nlcj + ijpj
2235         znorthloc(:,ij,:) = pt3d(:,jj,:)
2236      END DO
2237      !
2238      !                                     ! Build in procs of ncomm_north the znorthgloio
2239      itaille = jpi * jpk * ijpj
[3294]2240      IF ( l_north_nogather ) THEN
2241         !
2242         ! Avoid the use of mpi_allgather by exchanging only with the processes already identified
2243         ! (in nemo_northcomms) as being  involved in this process' northern boundary exchange
2244         !
2245         DO jj = nlcj-ijpj+1, nlcj          ! First put local values into the global array
2246            ij = jj - nlcj + ijpj
2247            DO ji = 1, nlci
2248               ztab(ji+nimpp-1,ij,:) = pt3d(ji,jj,:)
[1344]2249            END DO
2250         END DO
[3294]2251
2252         !
2253         ! Set the exchange type in order to access the correct list of active neighbours
2254         !
2255         SELECT CASE ( cd_type )
2256            CASE ( 'T' , 'W' )
2257               ityp = 1
2258            CASE ( 'U' )
2259               ityp = 2
2260            CASE ( 'V' )
2261               ityp = 3
2262            CASE ( 'F' )
2263               ityp = 4
2264            CASE ( 'I' )
2265               ityp = 5
2266            CASE DEFAULT
2267               ityp = -1                    ! Set a default value for unsupported types which
2268                                            ! will cause a fallback to the mpi_allgather method
2269         END SELECT
2270         IF ( ityp .gt. 0 ) THEN
2271
2272            DO jr = 1,nsndto(ityp)
2273               CALL mppsend(5, znorthloc, itaille, isendto(jr,ityp), ml_req_nf(jr) )
2274            END DO
2275            DO jr = 1,nsndto(ityp)
2276               CALL mpprecv(5, zfoldwk, itaille, isendto(jr,ityp))
2277               iproc = isendto(jr,ityp) + 1
2278               ildi = nldit (iproc)
2279               ilei = nleit (iproc)
2280               iilb = nimppt(iproc)
2281               DO jj = 1, ijpj
2282                  DO ji = ildi, ilei
2283                     ztab(ji+iilb-1,jj,:) = zfoldwk(ji,jj,:)
2284                  END DO
2285               END DO
2286            END DO
2287            IF (l_isend) THEN
2288               DO jr = 1,nsndto(ityp)
2289                  CALL mpi_wait(ml_req_nf(jr), ml_stat, ml_err)
2290               END DO
2291            ENDIF
2292
2293         ENDIF
2294
2295      ENDIF
2296
2297      IF ( ityp .lt. 0 ) THEN
2298         CALL MPI_ALLGATHER( znorthloc  , itaille, MPI_DOUBLE_PRECISION,                &
2299            &                znorthgloio, itaille, MPI_DOUBLE_PRECISION, ncomm_north, ierr )
2300         !
2301         DO jr = 1, ndim_rank_north         ! recover the global north array
2302            iproc = nrank_north(jr) + 1
2303            ildi  = nldit (iproc)
2304            ilei  = nleit (iproc)
2305            iilb  = nimppt(iproc)
2306            DO jj = 1, ijpj
2307               DO ji = ildi, ilei
2308                  ztab(ji+iilb-1,jj,:) = znorthgloio(ji,jj,:,jr)
2309               END DO
2310            END DO
2311         END DO
2312      ENDIF
[1344]2313      !
[3294]2314      ! The ztab array has been either:
2315      !  a. Fully populated by the mpi_allgather operation or
2316      !  b. Had the active points for this domain and northern neighbours populated
2317      !     by peer to peer exchanges
2318      ! Either way the array may be folded by lbc_nfd and the result for the span of
2319      ! this domain will be identical.
2320      !
[1344]2321      CALL lbc_nfd( ztab, cd_type, psgn )   ! North fold boundary condition
2322      !
2323      DO jj = nlcj-ijpj+1, nlcj             ! Scatter back to pt3d
2324         ij = jj - nlcj + ijpj
2325         DO ji= 1, nlci
2326            pt3d(ji,jj,:) = ztab(ji+nimpp-1,ij,:)
2327         END DO
2328      END DO
2329      !
2330   END SUBROUTINE mpp_lbc_north_3d
[3]2331
2332
[1344]2333   SUBROUTINE mpp_lbc_north_2d( pt2d, cd_type, psgn)
2334      !!---------------------------------------------------------------------
2335      !!                   ***  routine mpp_lbc_north_2d  ***
2336      !!
2337      !! ** Purpose :   Ensure proper north fold horizontal bondary condition
2338      !!              in mpp configuration in case of jpn1 > 1 (for 2d array )
2339      !!
2340      !! ** Method  :   North fold condition and mpp with more than one proc
2341      !!              in i-direction require a specific treatment. We gather
2342      !!              the 4 northern lines of the global domain on 1 processor
2343      !!              and apply lbc north-fold on this sub array. Then we
2344      !!              scatter the north fold array back to the processors.
2345      !!
2346      !!----------------------------------------------------------------------
2347      REAL(wp), DIMENSION(jpi,jpj), INTENT(inout) ::   pt2d      ! 3D array on which the b.c. is applied
2348      CHARACTER(len=1)            , INTENT(in   ) ::   cd_type   ! nature of pt3d grid-points
2349      !                                                          !   = T ,  U , V , F or W  gridpoints
2350      REAL(wp)                    , INTENT(in   ) ::   psgn      ! = -1. the sign change across the north fold
2351      !!                                                             ! =  1. , the sign is kept
2352      INTEGER ::   ji, jj, jr
2353      INTEGER ::   ierr, itaille, ildi, ilei, iilb
2354      INTEGER ::   ijpj, ijpjm1, ij, iproc
[3294]2355      INTEGER, DIMENSION (jpmaxngh)      ::   ml_req_nf          ! for mpi_isend when avoiding mpi_allgather
2356      INTEGER                            ::   ml_err             ! for mpi_isend when avoiding mpi_allgather
2357      INTEGER, DIMENSION(MPI_STATUS_SIZE)::   ml_stat            ! for mpi_isend when avoiding mpi_allgather
[1344]2358      !!----------------------------------------------------------------------
2359      !
2360      ijpj   = 4
[3294]2361      ityp = -1
[1344]2362      ijpjm1 = 3
[2715]2363      ztab_2d(:,:) = 0.e0
[1344]2364      !
[2715]2365      DO jj = nlcj-ijpj+1, nlcj             ! put in znorthloc_2d the last 4 jlines of pt2d
[1344]2366         ij = jj - nlcj + ijpj
[2715]2367         znorthloc_2d(:,ij) = pt2d(:,jj)
[1344]2368      END DO
[3]2369
[2715]2370      !                                     ! Build in procs of ncomm_north the znorthgloio_2d
[1344]2371      itaille = jpi * ijpj
[3294]2372      IF ( l_north_nogather ) THEN
2373         !
2374         ! Avoid the use of mpi_allgather by exchanging only with the processes already identified
2375         ! (in nemo_northcomms) as being  involved in this process' northern boundary exchange
2376         !
2377         DO jj = nlcj-ijpj+1, nlcj          ! First put local values into the global array
2378            ij = jj - nlcj + ijpj
2379            DO ji = 1, nlci
2380               ztab_2d(ji+nimpp-1,ij) = pt2d(ji,jj)
[13]2381            END DO
2382         END DO
[3294]2383
2384         !
2385         ! Set the exchange type in order to access the correct list of active neighbours
2386         !
2387         SELECT CASE ( cd_type )
2388            CASE ( 'T' , 'W' )
2389               ityp = 1
2390            CASE ( 'U' )
2391               ityp = 2
2392            CASE ( 'V' )
2393               ityp = 3
2394            CASE ( 'F' )
2395               ityp = 4
2396            CASE ( 'I' )
2397               ityp = 5
2398            CASE DEFAULT
2399               ityp = -1                    ! Set a default value for unsupported types which
2400                                            ! will cause a fallback to the mpi_allgather method
2401         END SELECT
2402
2403         IF ( ityp .gt. 0 ) THEN
2404
2405            DO jr = 1,nsndto(ityp)
2406               CALL mppsend(5, znorthloc_2d, itaille, isendto(jr,ityp), ml_req_nf(jr) )
2407            END DO
2408            DO jr = 1,nsndto(ityp)
2409               CALL mpprecv(5, zfoldwk_2d, itaille, isendto(jr,ityp))
2410               iproc = isendto(jr,ityp) + 1
2411               ildi = nldit (iproc)
2412               ilei = nleit (iproc)
2413               iilb = nimppt(iproc)
2414               DO jj = 1, ijpj
2415                  DO ji = ildi, ilei
2416                     ztab_2d(ji+iilb-1,jj) = zfoldwk_2d(ji,jj)
2417                  END DO
2418               END DO
2419            END DO
2420            IF (l_isend) THEN
2421               DO jr = 1,nsndto(ityp)
2422                  CALL mpi_wait(ml_req_nf(jr), ml_stat, ml_err)
2423               END DO
2424            ENDIF
2425
2426         ENDIF
2427
2428      ENDIF
2429
2430      IF ( ityp .lt. 0 ) THEN
2431         CALL MPI_ALLGATHER( znorthloc_2d  , itaille, MPI_DOUBLE_PRECISION,        &
2432            &                znorthgloio_2d, itaille, MPI_DOUBLE_PRECISION, ncomm_north, ierr )
2433         !
2434         DO jr = 1, ndim_rank_north            ! recover the global north array
2435            iproc = nrank_north(jr) + 1
2436            ildi = nldit (iproc)
2437            ilei = nleit (iproc)
2438            iilb = nimppt(iproc)
2439            DO jj = 1, ijpj
2440               DO ji = ildi, ilei
2441                  ztab_2d(ji+iilb-1,jj) = znorthgloio_2d(ji,jj,jr)
2442               END DO
2443            END DO
2444         END DO
2445      ENDIF
[1344]2446      !
[3294]2447      ! The ztab array has been either:
2448      !  a. Fully populated by the mpi_allgather operation or
2449      !  b. Had the active points for this domain and northern neighbours populated
2450      !     by peer to peer exchanges
2451      ! Either way the array may be folded by lbc_nfd and the result for the span of
2452      ! this domain will be identical.
2453      !
[2715]2454      CALL lbc_nfd( ztab_2d, cd_type, psgn )   ! North fold boundary condition
[1344]2455      !
2456      !
2457      DO jj = nlcj-ijpj+1, nlcj             ! Scatter back to pt2d
[13]2458         ij = jj - nlcj + ijpj
[1344]2459         DO ji = 1, nlci
[2715]2460            pt2d(ji,jj) = ztab_2d(ji+nimpp-1,ij)
[1344]2461         END DO
[13]2462      END DO
[1344]2463      !
[13]2464   END SUBROUTINE mpp_lbc_north_2d
[3]2465
2466
[1344]2467   SUBROUTINE mpp_lbc_north_e( pt2d, cd_type, psgn)
2468      !!---------------------------------------------------------------------
2469      !!                   ***  routine mpp_lbc_north_2d  ***
2470      !!
2471      !! ** Purpose :   Ensure proper north fold horizontal bondary condition
2472      !!              in mpp configuration in case of jpn1 > 1 and for 2d
2473      !!              array with outer extra halo
2474      !!
2475      !! ** Method  :   North fold condition and mpp with more than one proc
2476      !!              in i-direction require a specific treatment. We gather
2477      !!              the 4+2*jpr2dj northern lines of the global domain on 1
2478      !!              processor and apply lbc north-fold on this sub array.
2479      !!              Then we scatter the north fold array back to the processors.
2480      !!
2481      !!----------------------------------------------------------------------
2482      REAL(wp), DIMENSION(1-jpr2di:jpi+jpr2di,1-jpr2dj:jpj+jpr2dj), INTENT(inout) ::   pt2d     ! 2D array with extra halo
2483      CHARACTER(len=1)                                            , INTENT(in   ) ::   cd_type  ! nature of pt3d grid-points
2484      !                                                                                         !   = T ,  U , V , F or W -points
2485      REAL(wp)                                                    , INTENT(in   ) ::   psgn     ! = -1. the sign change across the 
2486      !!                                                                                        ! north fold, =  1. otherwise
2487      INTEGER ::   ji, jj, jr
2488      INTEGER ::   ierr, itaille, ildi, ilei, iilb
2489      INTEGER ::   ijpj, ij, iproc
2490      !!----------------------------------------------------------------------
2491      !
2492      ijpj=4
[2715]2493      ztab_e(:,:) = 0.e0
[311]2494
[1344]2495      ij=0
[2715]2496      ! put in znorthloc_e the last 4 jlines of pt2d
[1344]2497      DO jj = nlcj - ijpj + 1 - jpr2dj, nlcj +jpr2dj
2498         ij = ij + 1
2499         DO ji = 1, jpi
[2715]2500            znorthloc_e(ji,ij)=pt2d(ji,jj)
[1344]2501         END DO
2502      END DO
2503      !
2504      itaille = jpi * ( ijpj + 2 * jpr2dj )
[2715]2505      CALL MPI_ALLGATHER( znorthloc_e(1,1)  , itaille, MPI_DOUBLE_PRECISION,    &
2506         &                znorthgloio_e(1,1,1), itaille, MPI_DOUBLE_PRECISION, ncomm_north, ierr )
[1344]2507      !
2508      DO jr = 1, ndim_rank_north            ! recover the global north array
2509         iproc = nrank_north(jr) + 1
2510         ildi = nldit (iproc)
2511         ilei = nleit (iproc)
2512         iilb = nimppt(iproc)
2513         DO jj = 1, ijpj+2*jpr2dj
2514            DO ji = ildi, ilei
[2715]2515               ztab_e(ji+iilb-1,jj) = znorthgloio_e(ji,jj,jr)
[311]2516            END DO
[1344]2517         END DO
2518      END DO
[311]2519
2520
[1344]2521      ! 2. North-Fold boundary conditions
2522      ! ----------------------------------
[2715]2523      CALL lbc_nfd( ztab_e(:,:), cd_type, psgn, pr2dj = jpr2dj )
[311]2524
[1344]2525      ij = jpr2dj
2526      !! Scatter back to pt2d
2527      DO jj = nlcj - ijpj + 1 , nlcj +jpr2dj
2528      ij  = ij +1 
2529         DO ji= 1, nlci
[2715]2530            pt2d(ji,jj) = ztab_e(ji+nimpp-1,ij)
[311]2531         END DO
2532      END DO
[1344]2533      !
[311]2534   END SUBROUTINE mpp_lbc_north_e
2535
[389]2536
[2481]2537   SUBROUTINE mpi_init_opa( ldtxt, ksft, code )
[1344]2538      !!---------------------------------------------------------------------
2539      !!                   ***  routine mpp_init.opa  ***
2540      !!
2541      !! ** Purpose :: export and attach a MPI buffer for bsend
2542      !!
2543      !! ** Method  :: define buffer size in namelist, if 0 no buffer attachment
2544      !!            but classical mpi_init
2545      !!
2546      !! History :: 01/11 :: IDRIS initial version for IBM only 
2547      !!            08/04 :: R. Benshila, generalisation
2548      !!---------------------------------------------------------------------
[2481]2549      CHARACTER(len=*),DIMENSION(:), INTENT(  out) ::   ldtxt 
2550      INTEGER                      , INTENT(inout) ::   ksft
2551      INTEGER                      , INTENT(  out) ::   code
2552      INTEGER                                      ::   ierr, ji
2553      LOGICAL                                      ::   mpi_was_called
[1344]2554      !!---------------------------------------------------------------------
2555      !
2556      CALL mpi_initialized( mpi_was_called, code )      ! MPI initialization
[532]2557      IF ( code /= MPI_SUCCESS ) THEN
[2481]2558         DO ji = 1, SIZE(ldtxt) 
2559            IF( TRIM(ldtxt(ji)) /= '' )   WRITE(*,*) ldtxt(ji)      ! control print of mynode
2560         END DO         
2561         WRITE(*, cform_err)
2562         WRITE(*, *) ' lib_mpp: Error in routine mpi_initialized'
[1344]2563         CALL mpi_abort( mpi_comm_world, code, ierr )
[532]2564      ENDIF
[1344]2565      !
2566      IF( .NOT. mpi_was_called ) THEN
2567         CALL mpi_init( code )
2568         CALL mpi_comm_dup( mpi_comm_world, mpi_comm_opa, code )
[532]2569         IF ( code /= MPI_SUCCESS ) THEN
[2481]2570            DO ji = 1, SIZE(ldtxt) 
2571               IF( TRIM(ldtxt(ji)) /= '' )   WRITE(*,*) ldtxt(ji)      ! control print of mynode
2572            END DO
2573            WRITE(*, cform_err)
2574            WRITE(*, *) ' lib_mpp: Error in routine mpi_comm_dup'
[532]2575            CALL mpi_abort( mpi_comm_world, code, ierr )
2576         ENDIF
2577      ENDIF
[1344]2578      !
[897]2579      IF( nn_buffer > 0 ) THEN
[2481]2580         WRITE(ldtxt(ksft),*) 'mpi_bsend, buffer allocation of  : ', nn_buffer   ;   ksft = ksft + 1
[897]2581         ! Buffer allocation and attachment
[2481]2582         ALLOCATE( tampon(nn_buffer), stat = ierr )
[2715]2583         IF( ierr /= 0 ) THEN
[2481]2584            DO ji = 1, SIZE(ldtxt) 
2585               IF( TRIM(ldtxt(ji)) /= '' )   WRITE(*,*) ldtxt(ji)      ! control print of mynode
2586            END DO
2587            WRITE(*, cform_err)
2588            WRITE(*, *) ' lib_mpp: Error in ALLOCATE', ierr
2589            CALL mpi_abort( mpi_comm_world, code, ierr )
2590         END IF
2591         CALL mpi_buffer_attach( tampon, nn_buffer, code )
[897]2592      ENDIF
[1344]2593      !
[13]2594   END SUBROUTINE mpi_init_opa
[3]2595
[1976]2596   SUBROUTINE DDPDD_MPI (ydda, yddb, ilen, itype)
2597      !!---------------------------------------------------------------------
2598      !!   Routine DDPDD_MPI: used by reduction operator MPI_SUMDD
2599      !!
2600      !!   Modification of original codes written by David H. Bailey
2601      !!   This subroutine computes yddb(i) = ydda(i)+yddb(i)
2602      !!---------------------------------------------------------------------
2603      INTEGER, INTENT(in)                         :: ilen, itype
2604      COMPLEX(wp), DIMENSION(ilen), INTENT(in)     :: ydda
2605      COMPLEX(wp), DIMENSION(ilen), INTENT(inout)  :: yddb
2606      !
2607      REAL(wp) :: zerr, zt1, zt2    ! local work variables
2608      INTEGER :: ji, ztmp           ! local scalar
2609
2610      ztmp = itype   ! avoid compilation warning
2611
2612      DO ji=1,ilen
2613      ! Compute ydda + yddb using Knuth's trick.
2614         zt1  = real(ydda(ji)) + real(yddb(ji))
2615         zerr = zt1 - real(ydda(ji))
2616         zt2  = ((real(yddb(ji)) - zerr) + (real(ydda(ji)) - (zt1 - zerr))) &
2617                + aimag(ydda(ji)) + aimag(yddb(ji))
2618
2619         ! The result is zt1 + zt2, after normalization.
2620         yddb(ji) = cmplx ( zt1 + zt2, zt2 - ((zt1 + zt2) - zt1),wp )
2621      END DO
2622
2623   END SUBROUTINE DDPDD_MPI
2624
[13]2625#else
2626   !!----------------------------------------------------------------------
2627   !!   Default case:            Dummy module        share memory computing
2628   !!----------------------------------------------------------------------
[2715]2629   USE in_out_manager
[1976]2630
[13]2631   INTERFACE mpp_sum
[3294]2632      MODULE PROCEDURE mpp_sum_a2s, mpp_sum_as, mpp_sum_ai, mpp_sum_s, mpp_sum_i, mppsum_realdd, mppsum_a_realdd
[13]2633   END INTERFACE
2634   INTERFACE mpp_max
[681]2635      MODULE PROCEDURE mppmax_a_int, mppmax_int, mppmax_a_real, mppmax_real
[13]2636   END INTERFACE
2637   INTERFACE mpp_min
2638      MODULE PROCEDURE mppmin_a_int, mppmin_int, mppmin_a_real, mppmin_real
2639   END INTERFACE
2640   INTERFACE mppobc
2641      MODULE PROCEDURE mppobc_1d, mppobc_2d, mppobc_3d, mppobc_4d
2642   END INTERFACE
[1344]2643   INTERFACE mpp_minloc
2644      MODULE PROCEDURE mpp_minloc2d ,mpp_minloc3d
2645   END INTERFACE
2646   INTERFACE mpp_maxloc
2647      MODULE PROCEDURE mpp_maxloc2d ,mpp_maxloc3d
2648   END INTERFACE
[3]2649
[13]2650   LOGICAL, PUBLIC, PARAMETER ::   lk_mpp = .FALSE.      !: mpp flag
[3294]2651   LOGICAL, PUBLIC            ::   ln_nnogather  = .FALSE.  !: namelist control of northfold comms (needed here in case "key_mpp_mpi" is not used)
[869]2652   INTEGER :: ncomm_ice
[2715]2653   !!----------------------------------------------------------------------
[13]2654CONTAINS
[3]2655
[2715]2656   INTEGER FUNCTION lib_mpp_alloc(kumout)          ! Dummy function
2657      INTEGER, INTENT(in) ::   kumout
2658      lib_mpp_alloc = 0
2659   END FUNCTION lib_mpp_alloc
2660
2661   FUNCTION mynode( ldtxt, kumnam, kstop, localComm ) RESULT (function_value)
[1579]2662      INTEGER, OPTIONAL            , INTENT(in   ) ::   localComm
[2715]2663      CHARACTER(len=*),DIMENSION(:) ::   ldtxt 
2664      INTEGER ::   kumnam, kstop
[1559]2665      IF( PRESENT( localComm ) .OR. .NOT.PRESENT( localComm ) )   function_value = 0
[1579]2666      IF( .FALSE. )   ldtxt(:) = 'never done'
[13]2667   END FUNCTION mynode
[3]2668
[13]2669   SUBROUTINE mppsync                       ! Dummy routine
2670   END SUBROUTINE mppsync
[3]2671
[869]2672   SUBROUTINE mpp_sum_as( parr, kdim, kcom )      ! Dummy routine
[13]2673      REAL   , DIMENSION(:) :: parr
2674      INTEGER               :: kdim
[869]2675      INTEGER, OPTIONAL     :: kcom 
2676      WRITE(*,*) 'mpp_sum_as: You should not have seen this print! error?', kdim, parr(1), kcom
[13]2677   END SUBROUTINE mpp_sum_as
[3]2678
[869]2679   SUBROUTINE mpp_sum_a2s( parr, kdim, kcom )      ! Dummy routine
[13]2680      REAL   , DIMENSION(:,:) :: parr
2681      INTEGER               :: kdim
[869]2682      INTEGER, OPTIONAL     :: kcom 
2683      WRITE(*,*) 'mpp_sum_a2s: You should not have seen this print! error?', kdim, parr(1,1), kcom
[13]2684   END SUBROUTINE mpp_sum_a2s
[3]2685
[869]2686   SUBROUTINE mpp_sum_ai( karr, kdim, kcom )      ! Dummy routine
[13]2687      INTEGER, DIMENSION(:) :: karr
2688      INTEGER               :: kdim
[869]2689      INTEGER, OPTIONAL     :: kcom 
2690      WRITE(*,*) 'mpp_sum_ai: You should not have seen this print! error?', kdim, karr(1), kcom
[13]2691   END SUBROUTINE mpp_sum_ai
[3]2692
[869]2693   SUBROUTINE mpp_sum_s( psca, kcom )            ! Dummy routine
[13]2694      REAL                  :: psca
[869]2695      INTEGER, OPTIONAL     :: kcom 
2696      WRITE(*,*) 'mpp_sum_s: You should not have seen this print! error?', psca, kcom
[13]2697   END SUBROUTINE mpp_sum_s
[2480]2698
[869]2699   SUBROUTINE mpp_sum_i( kint, kcom )            ! Dummy routine
[13]2700      integer               :: kint
[2480]2701      INTEGER, OPTIONAL     :: kcom 
[869]2702      WRITE(*,*) 'mpp_sum_i: You should not have seen this print! error?', kint, kcom
[13]2703   END SUBROUTINE mpp_sum_i
2704
[3294]2705   SUBROUTINE mppsum_realdd( ytab, kcom )
2706      COMPLEX(wp), INTENT(inout)         :: ytab    ! input scalar
2707      INTEGER , INTENT( in  ), OPTIONAL :: kcom
2708      WRITE(*,*) 'mppsum_realdd: You should not have seen this print! error?', ytab
2709   END SUBROUTINE mppsum_realdd
2710 
2711   SUBROUTINE mppsum_a_realdd( ytab, kdim, kcom )
2712      INTEGER , INTENT( in )                        ::   kdim      ! size of ytab
2713      COMPLEX(wp), DIMENSION(kdim), INTENT( inout ) ::   ytab      ! input array
2714      INTEGER , INTENT( in  ), OPTIONAL :: kcom
2715      WRITE(*,*) 'mppsum_a_realdd: You should not have seen this print! error?', kdim, ytab(1), kcom
2716   END SUBROUTINE mppsum_a_realdd
2717
[869]2718   SUBROUTINE mppmax_a_real( parr, kdim, kcom )
[13]2719      REAL   , DIMENSION(:) :: parr
2720      INTEGER               :: kdim
[869]2721      INTEGER, OPTIONAL     :: kcom 
2722      WRITE(*,*) 'mppmax_a_real: You should not have seen this print! error?', kdim, parr(1), kcom
[13]2723   END SUBROUTINE mppmax_a_real
2724
[869]2725   SUBROUTINE mppmax_real( psca, kcom )
[13]2726      REAL                  :: psca
[869]2727      INTEGER, OPTIONAL     :: kcom 
2728      WRITE(*,*) 'mppmax_real: You should not have seen this print! error?', psca, kcom
[13]2729   END SUBROUTINE mppmax_real
2730
[869]2731   SUBROUTINE mppmin_a_real( parr, kdim, kcom )
[13]2732      REAL   , DIMENSION(:) :: parr
2733      INTEGER               :: kdim
[869]2734      INTEGER, OPTIONAL     :: kcom 
2735      WRITE(*,*) 'mppmin_a_real: You should not have seen this print! error?', kdim, parr(1), kcom
[13]2736   END SUBROUTINE mppmin_a_real
2737
[869]2738   SUBROUTINE mppmin_real( psca, kcom )
[13]2739      REAL                  :: psca
[869]2740      INTEGER, OPTIONAL     :: kcom 
2741      WRITE(*,*) 'mppmin_real: You should not have seen this print! error?', psca, kcom
[13]2742   END SUBROUTINE mppmin_real
2743
[869]2744   SUBROUTINE mppmax_a_int( karr, kdim ,kcom)
[681]2745      INTEGER, DIMENSION(:) :: karr
2746      INTEGER               :: kdim
[869]2747      INTEGER, OPTIONAL     :: kcom 
[888]2748      WRITE(*,*) 'mppmax_a_int: You should not have seen this print! error?', kdim, karr(1), kcom
[681]2749   END SUBROUTINE mppmax_a_int
2750
[869]2751   SUBROUTINE mppmax_int( kint, kcom)
[681]2752      INTEGER               :: kint
[869]2753      INTEGER, OPTIONAL     :: kcom 
2754      WRITE(*,*) 'mppmax_int: You should not have seen this print! error?', kint, kcom
[681]2755   END SUBROUTINE mppmax_int
2756
[869]2757   SUBROUTINE mppmin_a_int( karr, kdim, kcom )
[13]2758      INTEGER, DIMENSION(:) :: karr
2759      INTEGER               :: kdim
[869]2760      INTEGER, OPTIONAL     :: kcom 
2761      WRITE(*,*) 'mppmin_a_int: You should not have seen this print! error?', kdim, karr(1), kcom
[13]2762   END SUBROUTINE mppmin_a_int
2763
[869]2764   SUBROUTINE mppmin_int( kint, kcom )
[13]2765      INTEGER               :: kint
[869]2766      INTEGER, OPTIONAL     :: kcom 
2767      WRITE(*,*) 'mppmin_int: You should not have seen this print! error?', kint, kcom
[13]2768   END SUBROUTINE mppmin_int
2769
[2715]2770   SUBROUTINE mppobc_1d( parr, kd1, kd2, kl, kk, ktype, kij, knum )
2771      INTEGER  ::   kd1, kd2, kl , kk, ktype, kij, knum
[1344]2772      REAL, DIMENSION(:) ::   parr           ! variable array
[2715]2773      WRITE(*,*) 'mppobc: You should not have seen this print! error?', parr(1), kd1, kd2, kl, kk, ktype, kij, knum
[13]2774   END SUBROUTINE mppobc_1d
2775
[2715]2776   SUBROUTINE mppobc_2d( parr, kd1, kd2, kl, kk, ktype, kij, knum )
2777      INTEGER  ::   kd1, kd2, kl , kk, ktype, kij, knum
[1344]2778      REAL, DIMENSION(:,:) ::   parr           ! variable array
[2715]2779      WRITE(*,*) 'mppobc: You should not have seen this print! error?', parr(1,1), kd1, kd2, kl, kk, ktype, kij, knum
[13]2780   END SUBROUTINE mppobc_2d
2781
[2715]2782   SUBROUTINE mppobc_3d( parr, kd1, kd2, kl, kk, ktype, kij, knum )
2783      INTEGER  ::   kd1, kd2, kl , kk, ktype, kij, knum
[1344]2784      REAL, DIMENSION(:,:,:) ::   parr           ! variable array
[2715]2785      WRITE(*,*) 'mppobc: You should not have seen this print! error?', parr(1,1,1), kd1, kd2, kl, kk, ktype, kij, knum
[13]2786   END SUBROUTINE mppobc_3d
2787
[2715]2788   SUBROUTINE mppobc_4d( parr, kd1, kd2, kl, kk, ktype, kij, knum )
2789      INTEGER  ::   kd1, kd2, kl , kk, ktype, kij, knum
[1344]2790      REAL, DIMENSION(:,:,:,:) ::   parr           ! variable array
[2715]2791      WRITE(*,*) 'mppobc: You should not have seen this print! error?', parr(1,1,1,1), kd1, kd2, kl, kk, ktype, kij, knum
[13]2792   END SUBROUTINE mppobc_4d
2793
[1344]2794   SUBROUTINE mpp_minloc2d( ptab, pmask, pmin, ki, kj )
[181]2795      REAL                   :: pmin
2796      REAL , DIMENSION (:,:) :: ptab, pmask
2797      INTEGER :: ki, kj
[1528]2798      WRITE(*,*) 'mpp_minloc2d: You should not have seen this print! error?', pmin, ki, kj, ptab(1,1), pmask(1,1)
[181]2799   END SUBROUTINE mpp_minloc2d
2800
[1344]2801   SUBROUTINE mpp_minloc3d( ptab, pmask, pmin, ki, kj, kk )
[181]2802      REAL                     :: pmin
2803      REAL , DIMENSION (:,:,:) :: ptab, pmask
2804      INTEGER :: ki, kj, kk
[1528]2805      WRITE(*,*) 'mpp_minloc3d: You should not have seen this print! error?', pmin, ki, kj, kk, ptab(1,1,1), pmask(1,1,1)
[181]2806   END SUBROUTINE mpp_minloc3d
2807
[1344]2808   SUBROUTINE mpp_maxloc2d( ptab, pmask, pmax, ki, kj )
[181]2809      REAL                   :: pmax
2810      REAL , DIMENSION (:,:) :: ptab, pmask
2811      INTEGER :: ki, kj
[1528]2812      WRITE(*,*) 'mpp_maxloc2d: You should not have seen this print! error?', pmax, ki, kj, ptab(1,1), pmask(1,1)
[181]2813   END SUBROUTINE mpp_maxloc2d
2814
[1344]2815   SUBROUTINE mpp_maxloc3d( ptab, pmask, pmax, ki, kj, kk )
[181]2816      REAL                     :: pmax
2817      REAL , DIMENSION (:,:,:) :: ptab, pmask
2818      INTEGER :: ki, kj, kk
[1528]2819      WRITE(*,*) 'mpp_maxloc3d: You should not have seen this print! error?', pmax, ki, kj, kk, ptab(1,1,1), pmask(1,1,1)
[181]2820   END SUBROUTINE mpp_maxloc3d
2821
[51]2822   SUBROUTINE mppstop
[3294]2823      WRITE(*,*) 'mppstop: You should not have seen this print if running in mpp mode! error?...'
2824      WRITE(*,*) 'mppstop: ..otherwise this is a stop condition raised by ctl_stop in single processor mode'
2825      STOP
[51]2826   END SUBROUTINE mppstop
2827
[2715]2828   SUBROUTINE mpp_ini_ice( kcom, knum )
2829      INTEGER :: kcom, knum
2830      WRITE(*,*) 'mpp_ini_ice: You should not have seen this print! error?', kcom, knum
[888]2831   END SUBROUTINE mpp_ini_ice
[869]2832
[2715]2833   SUBROUTINE mpp_ini_znl( knum )
2834      INTEGER :: knum
2835      WRITE(*,*) 'mpp_ini_znl: You should not have seen this print! error?', knum
[1345]2836   END SUBROUTINE mpp_ini_znl
2837
[1344]2838   SUBROUTINE mpp_comm_free( kcom )
[869]2839      INTEGER :: kcom
[1344]2840      WRITE(*,*) 'mpp_comm_free: You should not have seen this print! error?', kcom
[869]2841   END SUBROUTINE mpp_comm_free
[3]2842#endif
[2715]2843
[13]2844   !!----------------------------------------------------------------------
[2715]2845   !!   All cases:         ctl_stop, ctl_warn, get_unit, ctl_opn   routines
2846   !!----------------------------------------------------------------------
2847
2848   SUBROUTINE ctl_stop( cd1, cd2, cd3, cd4, cd5 ,   &
2849      &                 cd6, cd7, cd8, cd9, cd10 )
2850      !!----------------------------------------------------------------------
2851      !!                  ***  ROUTINE  stop_opa  ***
2852      !!
2853      !! ** Purpose :   print in ocean.outpput file a error message and
2854      !!                increment the error number (nstop) by one.
2855      !!----------------------------------------------------------------------
2856      CHARACTER(len=*), INTENT(in), OPTIONAL ::  cd1, cd2, cd3, cd4, cd5
2857      CHARACTER(len=*), INTENT(in), OPTIONAL ::  cd6, cd7, cd8, cd9, cd10
2858      !!----------------------------------------------------------------------
2859      !
2860      nstop = nstop + 1 
2861      IF(lwp) THEN
2862         WRITE(numout,cform_err)
2863         IF( PRESENT(cd1 ) )   WRITE(numout,*) cd1
2864         IF( PRESENT(cd2 ) )   WRITE(numout,*) cd2
2865         IF( PRESENT(cd3 ) )   WRITE(numout,*) cd3
2866         IF( PRESENT(cd4 ) )   WRITE(numout,*) cd4
2867         IF( PRESENT(cd5 ) )   WRITE(numout,*) cd5
2868         IF( PRESENT(cd6 ) )   WRITE(numout,*) cd6
2869         IF( PRESENT(cd7 ) )   WRITE(numout,*) cd7
2870         IF( PRESENT(cd8 ) )   WRITE(numout,*) cd8
2871         IF( PRESENT(cd9 ) )   WRITE(numout,*) cd9
2872         IF( PRESENT(cd10) )   WRITE(numout,*) cd10
2873      ENDIF
2874                               CALL FLUSH(numout    )
2875      IF( numstp     /= -1 )   CALL FLUSH(numstp    )
2876      IF( numsol     /= -1 )   CALL FLUSH(numsol    )
2877      IF( numevo_ice /= -1 )   CALL FLUSH(numevo_ice)
2878      !
2879      IF( cd1 == 'STOP' ) THEN
2880         IF(lwp) WRITE(numout,*)  'huge E-R-R-O-R : immediate stop'
2881         CALL mppstop()
2882      ENDIF
2883      !
2884   END SUBROUTINE ctl_stop
2885
2886
2887   SUBROUTINE ctl_warn( cd1, cd2, cd3, cd4, cd5,   &
2888      &                 cd6, cd7, cd8, cd9, cd10 )
2889      !!----------------------------------------------------------------------
2890      !!                  ***  ROUTINE  stop_warn  ***
2891      !!
2892      !! ** Purpose :   print in ocean.outpput file a error message and
2893      !!                increment the warning number (nwarn) by one.
2894      !!----------------------------------------------------------------------
2895      CHARACTER(len=*), INTENT(in), OPTIONAL ::  cd1, cd2, cd3, cd4, cd5
2896      CHARACTER(len=*), INTENT(in), OPTIONAL ::  cd6, cd7, cd8, cd9, cd10
2897      !!----------------------------------------------------------------------
2898      !
2899      nwarn = nwarn + 1 
2900      IF(lwp) THEN
2901         WRITE(numout,cform_war)
2902         IF( PRESENT(cd1 ) ) WRITE(numout,*) cd1
2903         IF( PRESENT(cd2 ) ) WRITE(numout,*) cd2
2904         IF( PRESENT(cd3 ) ) WRITE(numout,*) cd3
2905         IF( PRESENT(cd4 ) ) WRITE(numout,*) cd4
2906         IF( PRESENT(cd5 ) ) WRITE(numout,*) cd5
2907         IF( PRESENT(cd6 ) ) WRITE(numout,*) cd6
2908         IF( PRESENT(cd7 ) ) WRITE(numout,*) cd7
2909         IF( PRESENT(cd8 ) ) WRITE(numout,*) cd8
2910         IF( PRESENT(cd9 ) ) WRITE(numout,*) cd9
2911         IF( PRESENT(cd10) ) WRITE(numout,*) cd10
2912      ENDIF
2913      CALL FLUSH(numout)
2914      !
2915   END SUBROUTINE ctl_warn
2916
2917
2918   SUBROUTINE ctl_opn( knum, cdfile, cdstat, cdform, cdacce, klengh, kout, ldwp, karea )
2919      !!----------------------------------------------------------------------
2920      !!                  ***  ROUTINE ctl_opn  ***
2921      !!
2922      !! ** Purpose :   Open file and check if required file is available.
2923      !!
2924      !! ** Method  :   Fortan open
2925      !!----------------------------------------------------------------------
2926      INTEGER          , INTENT(  out) ::   knum      ! logical unit to open
2927      CHARACTER(len=*) , INTENT(in   ) ::   cdfile    ! file name to open
2928      CHARACTER(len=*) , INTENT(in   ) ::   cdstat    ! disposition specifier
2929      CHARACTER(len=*) , INTENT(in   ) ::   cdform    ! formatting specifier
2930      CHARACTER(len=*) , INTENT(in   ) ::   cdacce    ! access specifier
2931      INTEGER          , INTENT(in   ) ::   klengh    ! record length
2932      INTEGER          , INTENT(in   ) ::   kout      ! number of logical units for write
2933      LOGICAL          , INTENT(in   ) ::   ldwp      ! boolean term for print
2934      INTEGER, OPTIONAL, INTENT(in   ) ::   karea     ! proc number
2935      !!
2936      CHARACTER(len=80) ::   clfile
2937      INTEGER           ::   iost
2938      !!----------------------------------------------------------------------
2939
2940      ! adapt filename
2941      ! ----------------
2942      clfile = TRIM(cdfile)
2943      IF( PRESENT( karea ) ) THEN
2944         IF( karea > 1 )   WRITE(clfile, "(a,'_',i4.4)") TRIM(clfile), karea-1
2945      ENDIF
2946#if defined key_agrif
2947      IF( .NOT. Agrif_Root() )   clfile = TRIM(Agrif_CFixed())//'_'//TRIM(clfile)
2948      knum=Agrif_Get_Unit()
2949#else
2950      knum=get_unit()
2951#endif
2952
2953      iost=0
2954      IF( cdacce(1:6) == 'DIRECT' )  THEN
2955         OPEN( UNIT=knum, FILE=clfile, FORM=cdform, ACCESS=cdacce, STATUS=cdstat, RECL=klengh, ERR=100, IOSTAT=iost )
2956      ELSE
2957         OPEN( UNIT=knum, FILE=clfile, FORM=cdform, ACCESS=cdacce, STATUS=cdstat             , ERR=100, IOSTAT=iost )
2958      ENDIF
2959      IF( iost == 0 ) THEN
2960         IF(ldwp) THEN
2961            WRITE(kout,*) '     file   : ', clfile,' open ok'
2962            WRITE(kout,*) '     unit   = ', knum
2963            WRITE(kout,*) '     status = ', cdstat
2964            WRITE(kout,*) '     form   = ', cdform
2965            WRITE(kout,*) '     access = ', cdacce
2966            WRITE(kout,*)
2967         ENDIF
2968      ENDIF
2969100   CONTINUE
2970      IF( iost /= 0 ) THEN
2971         IF(ldwp) THEN
2972            WRITE(kout,*)
2973            WRITE(kout,*) ' ===>>>> : bad opening file: ', clfile
2974            WRITE(kout,*) ' =======   ===  '
2975            WRITE(kout,*) '           unit   = ', knum
2976            WRITE(kout,*) '           status = ', cdstat
2977            WRITE(kout,*) '           form   = ', cdform
2978            WRITE(kout,*) '           access = ', cdacce
2979            WRITE(kout,*) '           iostat = ', iost
2980            WRITE(kout,*) '           we stop. verify the file '
2981            WRITE(kout,*)
2982         ENDIF
2983         STOP 'ctl_opn bad opening'
2984      ENDIF
2985     
2986   END SUBROUTINE ctl_opn
2987
2988
2989   INTEGER FUNCTION get_unit()
2990      !!----------------------------------------------------------------------
2991      !!                  ***  FUNCTION  get_unit  ***
2992      !!
2993      !! ** Purpose :   return the index of an unused logical unit
2994      !!----------------------------------------------------------------------
2995      LOGICAL :: llopn 
2996      !!----------------------------------------------------------------------
2997      !
2998      get_unit = 15   ! choose a unit that is big enough then it is not already used in NEMO
2999      llopn = .TRUE.
3000      DO WHILE( (get_unit < 998) .AND. llopn )
3001         get_unit = get_unit + 1
3002         INQUIRE( unit = get_unit, opened = llopn )
3003      END DO
3004      IF( (get_unit == 999) .AND. llopn ) THEN
3005         CALL ctl_stop( 'get_unit: All logical units until 999 are used...' )
3006         get_unit = -1
3007      ENDIF
3008      !
3009   END FUNCTION get_unit
3010
3011   !!----------------------------------------------------------------------
[3]3012END MODULE lib_mpp
Note: See TracBrowser for help on using the repository browser.