source: trunk/NEMOGCM/NEMO/OPA_SRC/LBC/lib_mpp.F90 @ 3435

Last change on this file since 3435 was 3435, checked in by rfurner, 8 years ago

ticket 982, removed unnecessary ifdef statments from the lib_mpp to fix bug preventing compiling of closea.F90.

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