source: NEMO/branches/UKMO/NEMO_4.0_mirror/src/OCE/LBC/lib_mpp.F90 @ 10888

Last change on this file since 10888 was 10888, checked in by davestorkey, 20 months ago

branches/UKMO/NEMO_4.0_mirror : clear SVN keywords

File size: 90.4 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   !!            3.5  !  2012  (S.Mocavero, I. Epicoco) Add mpp_lnk_bdy_3d/2d routines to optimize the BDY comm.
22   !!            3.5  !  2013  (C. Ethe, G. Madec)  message passing arrays as local variables
23   !!            3.5  !  2013  (S.Mocavero, I.Epicoco - CMCC) north fold optimizations
24   !!            3.6  !  2015  (O. Tintó and M. Castrillo - BSC) Added '_multiple' case for 2D lbc and max
25   !!            4.0  !  2017  (G. Madec) automatique allocation of array argument (use any 3rd dimension)
26   !!             -   !  2017  (G. Madec) create generic.h90 files to generate all lbc and north fold routines
27   !!----------------------------------------------------------------------
28
29   !!----------------------------------------------------------------------
30   !!   ctl_stop      : update momentum and tracer Kz from a tke scheme
31   !!   ctl_warn      : initialization, namelist read, and parameters control
32   !!   ctl_opn       : Open file and check if required file is available.
33   !!   ctl_nam       : Prints informations when an error occurs while reading a namelist
34   !!   get_unit      : give the index of an unused logical unit
35   !!----------------------------------------------------------------------
36#if   defined key_mpp_mpi
37   !!----------------------------------------------------------------------
38   !!   'key_mpp_mpi'             MPI massively parallel processing library
39   !!----------------------------------------------------------------------
40   !!   lib_mpp_alloc : allocate mpp arrays
41   !!   mynode        : indentify the processor unit
42   !!   mpp_lnk       : interface (defined in lbclnk) for message passing of 2d or 3d arrays (mpp_lnk_2d, mpp_lnk_3d)
43   !!   mpp_lnk_icb   : interface for message passing of 2d arrays with extra halo for icebergs (mpp_lnk_2d_icb)
44   !!   mpprecv       :
45   !!   mppsend       :
46   !!   mppscatter    :
47   !!   mppgather     :
48   !!   mpp_min       : generic interface for mppmin_int , mppmin_a_int , mppmin_real, mppmin_a_real
49   !!   mpp_max       : generic interface for mppmax_int , mppmax_a_int , mppmax_real, mppmax_a_real
50   !!   mpp_sum       : generic interface for mppsum_int , mppsum_a_int , mppsum_real, mppsum_a_real
51   !!   mpp_minloc    :
52   !!   mpp_maxloc    :
53   !!   mppsync       :
54   !!   mppstop       :
55   !!   mpp_ini_north : initialisation of north fold
56   !!   mpp_lbc_north_icb : alternative to mpp_nfd for extra outer halo with icebergs
57   !!----------------------------------------------------------------------
58   USE dom_oce        ! ocean space and time domain
59   USE lbcnfd         ! north fold treatment
60   USE in_out_manager ! I/O manager
61
62   IMPLICIT NONE
63   PRIVATE
64
65   INTERFACE mpp_nfd
66      MODULE PROCEDURE   mpp_nfd_2d    , mpp_nfd_3d    , mpp_nfd_4d
67      MODULE PROCEDURE   mpp_nfd_2d_ptr, mpp_nfd_3d_ptr, mpp_nfd_4d_ptr
68   END INTERFACE
69
70   ! Interface associated to the mpp_lnk_... routines is defined in lbclnk
71   PUBLIC   mpp_lnk_2d    , mpp_lnk_3d    , mpp_lnk_4d
72   PUBLIC   mpp_lnk_2d_ptr, mpp_lnk_3d_ptr, mpp_lnk_4d_ptr
73   !
74!!gm  this should be useless
75   PUBLIC   mpp_nfd_2d    , mpp_nfd_3d    , mpp_nfd_4d
76   PUBLIC   mpp_nfd_2d_ptr, mpp_nfd_3d_ptr, mpp_nfd_4d_ptr
77!!gm end
78   !
79   PUBLIC   ctl_stop, ctl_warn, get_unit, ctl_opn, ctl_nam
80   PUBLIC   mynode, mppstop, mppsync, mpp_comm_free
81   PUBLIC   mpp_ini_north
82   PUBLIC   mpp_lnk_2d_icb
83   PUBLIC   mpp_lbc_north_icb
84   PUBLIC   mpp_min, mpp_max, mpp_sum, mpp_minloc, mpp_maxloc
85   PUBLIC   mpp_delay_max, mpp_delay_sum, mpp_delay_rcv
86   PUBLIC   mppscatter, mppgather
87   PUBLIC   mpp_ini_znl
88   PUBLIC   mppsend, mpprecv                          ! needed by TAM and ICB routines
89   PUBLIC   mpp_lnk_bdy_2d, mpp_lnk_bdy_3d, mpp_lnk_bdy_4d
90   
91   !! * Interfaces
92   !! define generic interface for these routine as they are called sometimes
93   !! with scalar arguments instead of array arguments, which causes problems
94   !! for the compilation on AIX system as well as NEC and SGI. Ok on COMPACQ
95   INTERFACE mpp_min
96      MODULE PROCEDURE mppmin_a_int, mppmin_int, mppmin_a_real, mppmin_real
97   END INTERFACE
98   INTERFACE mpp_max
99      MODULE PROCEDURE mppmax_a_int, mppmax_int, mppmax_a_real, mppmax_real
100   END INTERFACE
101   INTERFACE mpp_sum
102      MODULE PROCEDURE mppsum_a_int, mppsum_int, mppsum_a_real, mppsum_real,   &
103         &             mppsum_realdd, mppsum_a_realdd
104   END INTERFACE
105   INTERFACE mpp_minloc
106      MODULE PROCEDURE mpp_minloc2d ,mpp_minloc3d
107   END INTERFACE
108   INTERFACE mpp_maxloc
109      MODULE PROCEDURE mpp_maxloc2d ,mpp_maxloc3d
110   END INTERFACE
111
112   !! ========================= !!
113   !!  MPI  variable definition !!
114   !! ========================= !!
115!$AGRIF_DO_NOT_TREAT
116   INCLUDE 'mpif.h'
117!$AGRIF_END_DO_NOT_TREAT
118
119   LOGICAL, PUBLIC, PARAMETER ::   lk_mpp = .TRUE.    !: mpp flag
120
121   INTEGER, PARAMETER         ::   nprocmax = 2**10   ! maximun dimension (required to be a power of 2)
122
123   INTEGER, PUBLIC ::   mppsize        ! number of process
124   INTEGER, PUBLIC ::   mpprank        ! process number  [ 0 - size-1 ]
125!$AGRIF_DO_NOT_TREAT
126   INTEGER, PUBLIC ::   mpi_comm_oce   ! opa local communicator
127!$AGRIF_END_DO_NOT_TREAT
128
129   INTEGER :: MPI_SUMDD
130
131   ! variables used for zonal integration
132   INTEGER, PUBLIC ::   ncomm_znl       !: communicator made by the processors on the same zonal average
133   LOGICAL, PUBLIC ::   l_znl_root      !: True on the 'left'most processor on the same row
134   INTEGER         ::   ngrp_znl        !  group ID for the znl processors
135   INTEGER         ::   ndim_rank_znl   !  number of processors on the same zonal average
136   INTEGER, DIMENSION(:), ALLOCATABLE, SAVE ::   nrank_znl  ! dimension ndim_rank_znl, number of the procs into the same znl domain
137
138   ! North fold condition in mpp_mpi with jpni > 1 (PUBLIC for TAM)
139   INTEGER, PUBLIC ::   ngrp_world        !: group ID for the world processors
140   INTEGER, PUBLIC ::   ngrp_opa          !: group ID for the opa processors
141   INTEGER, PUBLIC ::   ngrp_north        !: group ID for the northern processors (to be fold)
142   INTEGER, PUBLIC ::   ncomm_north       !: communicator made by the processors belonging to ngrp_north
143   INTEGER, PUBLIC ::   ndim_rank_north   !: number of 'sea' processor in the northern line (can be /= jpni !)
144   INTEGER, PUBLIC ::   njmppmax          !: value of njmpp for the processors of the northern line
145   INTEGER, PUBLIC ::   north_root        !: number (in the comm_opa) of proc 0 in the northern comm
146   INTEGER, PUBLIC, DIMENSION(:), ALLOCATABLE, SAVE ::   nrank_north   !: dimension ndim_rank_north
147
148   ! Type of send : standard, buffered, immediate
149   CHARACTER(len=1), PUBLIC ::   cn_mpi_send        !: type od mpi send/recieve (S=standard, B=bsend, I=isend)
150   LOGICAL         , PUBLIC ::   l_isend = .FALSE.  !: isend use indicator (T if cn_mpi_send='I')
151   INTEGER         , PUBLIC ::   nn_buffer          !: size of the buffer in case of mpi_bsend
152
153   ! Communications summary report
154   CHARACTER(len=128), DIMENSION(:), ALLOCATABLE ::   crname_lbc                   !: names of lbc_lnk calling routines
155   CHARACTER(len=128), DIMENSION(:), ALLOCATABLE ::   crname_glb                   !: names of global comm calling routines
156   CHARACTER(len=128), DIMENSION(:), ALLOCATABLE ::   crname_dlg                   !: names of delayed global comm calling routines
157   INTEGER, PUBLIC                               ::   ncom_stp = 0                 !: copy of time step # istp
158   INTEGER, PUBLIC                               ::   ncom_fsbc = 1                !: copy of sbc time step # nn_fsbc
159   INTEGER, PUBLIC                               ::   ncom_dttrc = 1               !: copy of top time step # nn_dttrc
160   INTEGER, PUBLIC                               ::   ncom_freq                    !: frequency of comm diagnostic
161   INTEGER, PUBLIC , DIMENSION(:,:), ALLOCATABLE ::   ncomm_sequence               !: size of communicated arrays (halos)
162   INTEGER, PARAMETER, PUBLIC                    ::   ncom_rec_max = 5000          !: max number of communication record
163   INTEGER, PUBLIC                               ::   n_sequence_lbc = 0           !: # of communicated arraysvia lbc
164   INTEGER, PUBLIC                               ::   n_sequence_glb = 0           !: # of global communications
165   INTEGER, PUBLIC                               ::   n_sequence_dlg = 0           !: # of delayed global communications
166   INTEGER, PUBLIC                               ::   numcom = -1                  !: logical unit for communicaton report
167   LOGICAL, PUBLIC                               ::   l_full_nf_update = .TRUE.    !: logical for a full (2lines) update of bc at North fold report
168   INTEGER,                    PARAMETER, PUBLIC ::   nbdelay = 2       !: number of delayed operations
169   !: name (used as id) of allreduce-delayed operations
170   ! Warning: we must use the same character length in an array constructor (at least for gcc compiler)
171   CHARACTER(len=32), DIMENSION(nbdelay), PUBLIC ::   c_delaylist = (/ 'cflice', 'fwb   ' /)
172   !: component name where the allreduce-delayed operation is performed
173   CHARACTER(len=3),  DIMENSION(nbdelay), PUBLIC ::   c_delaycpnt = (/ 'ICE'   , 'OCE' /)
174   TYPE, PUBLIC ::   DELAYARR
175      REAL(   wp), POINTER, DIMENSION(:) ::  z1d => NULL()
176      COMPLEX(wp), POINTER, DIMENSION(:) ::  y1d => NULL()
177   END TYPE DELAYARR
178   TYPE( DELAYARR ), DIMENSION(nbdelay), PUBLIC, SAVE  ::   todelay         !: must have SAVE for default initialization of DELAYARR
179   INTEGER,          DIMENSION(nbdelay), PUBLIC        ::   ndelayid = -1   !: mpi request id of the delayed operations
180
181   ! timing summary report
182   REAL(wp), DIMENSION(2), PUBLIC ::  waiting_time = 0._wp
183   REAL(wp)              , PUBLIC ::  compute_time = 0._wp, elapsed_time = 0._wp
184   
185   REAL(wp), DIMENSION(:), ALLOCATABLE, SAVE ::   tampon   ! buffer in case of bsend
186
187   LOGICAL, PUBLIC ::   ln_nnogather                !: namelist control of northfold comms
188   LOGICAL, PUBLIC ::   l_north_nogather = .FALSE.  !: internal control of northfold comms
189
190   !!----------------------------------------------------------------------
191   !! NEMO/OCE 4.0 , NEMO Consortium (2018)
192   !! $Id$
193   !! Software governed by the CeCILL license (see ./LICENSE)
194   !!----------------------------------------------------------------------
195CONTAINS
196
197   FUNCTION mynode( ldtxt, ldname, kumnam_ref, kumnam_cfg, kumond, kstop, localComm )
198      !!----------------------------------------------------------------------
199      !!                  ***  routine mynode  ***
200      !!
201      !! ** Purpose :   Find processor unit
202      !!----------------------------------------------------------------------
203      CHARACTER(len=*),DIMENSION(:), INTENT(  out) ::   ldtxt        !
204      CHARACTER(len=*)             , INTENT(in   ) ::   ldname       !
205      INTEGER                      , INTENT(in   ) ::   kumnam_ref   ! logical unit for reference namelist
206      INTEGER                      , INTENT(in   ) ::   kumnam_cfg   ! logical unit for configuration namelist
207      INTEGER                      , INTENT(inout) ::   kumond       ! logical unit for namelist output
208      INTEGER                      , INTENT(inout) ::   kstop        ! stop indicator
209      INTEGER         , OPTIONAL   , INTENT(in   ) ::   localComm    !
210      !
211      INTEGER ::   mynode, ierr, code, ji, ii, ios
212      LOGICAL ::   mpi_was_called
213      !
214      NAMELIST/nammpp/ cn_mpi_send, nn_buffer, jpni, jpnj, ln_nnogather
215      !!----------------------------------------------------------------------
216      !
217      ii = 1
218      WRITE(ldtxt(ii),*)                                                                  ;   ii = ii + 1
219      WRITE(ldtxt(ii),*) 'mynode : mpi initialisation'                                    ;   ii = ii + 1
220      WRITE(ldtxt(ii),*) '~~~~~~ '                                                        ;   ii = ii + 1
221      !
222      REWIND( kumnam_ref )              ! Namelist nammpp in reference namelist: mpi variables
223      READ  ( kumnam_ref, nammpp, IOSTAT = ios, ERR = 901)
224901   IF( ios /= 0 )   CALL ctl_nam ( ios , 'nammpp in reference namelist', lwp )
225      !
226      REWIND( kumnam_cfg )              ! Namelist nammpp in configuration namelist: mpi variables
227      READ  ( kumnam_cfg, nammpp, IOSTAT = ios, ERR = 902 )
228902   IF( ios >  0 )   CALL ctl_nam ( ios , 'nammpp in configuration namelist', lwp )
229      !
230      !                              ! control print
231      WRITE(ldtxt(ii),*) '   Namelist nammpp'                                             ;   ii = ii + 1
232      WRITE(ldtxt(ii),*) '      mpi send type          cn_mpi_send = ', cn_mpi_send       ;   ii = ii + 1
233      WRITE(ldtxt(ii),*) '      size exported buffer   nn_buffer   = ', nn_buffer,' bytes';   ii = ii + 1
234      !
235      IF( jpni < 1 .OR. jpnj < 1  ) THEN
236         WRITE(ldtxt(ii),*) '      jpni and jpnj will be calculated automatically' ;   ii = ii + 1
237      ELSE
238         WRITE(ldtxt(ii),*) '      processor grid extent in i         jpni = ',jpni       ;   ii = ii + 1
239         WRITE(ldtxt(ii),*) '      processor grid extent in j         jpnj = ',jpnj       ;   ii = ii + 1
240      ENDIF
241
242      WRITE(ldtxt(ii),*) '      avoid use of mpi_allgather at the north fold  ln_nnogather = ', ln_nnogather  ; ii = ii + 1
243
244      CALL mpi_initialized ( mpi_was_called, code )
245      IF( code /= MPI_SUCCESS ) THEN
246         DO ji = 1, SIZE(ldtxt)
247            IF( TRIM(ldtxt(ji)) /= '' )   WRITE(*,*) ldtxt(ji)      ! control print of mynode
248         END DO
249         WRITE(*, cform_err)
250         WRITE(*, *) 'lib_mpp: Error in routine mpi_initialized'
251         CALL mpi_abort( mpi_comm_world, code, ierr )
252      ENDIF
253
254      IF( mpi_was_called ) THEN
255         !
256         SELECT CASE ( cn_mpi_send )
257         CASE ( 'S' )                ! Standard mpi send (blocking)
258            WRITE(ldtxt(ii),*) '           Standard blocking mpi send (send)'             ;   ii = ii + 1
259         CASE ( 'B' )                ! Buffer mpi send (blocking)
260            WRITE(ldtxt(ii),*) '           Buffer blocking mpi send (bsend)'              ;   ii = ii + 1
261            IF( Agrif_Root() )   CALL mpi_init_oce( ldtxt, ii, ierr )
262         CASE ( 'I' )                ! Immediate mpi send (non-blocking send)
263            WRITE(ldtxt(ii),*) '           Immediate non-blocking send (isend)'           ;   ii = ii + 1
264            l_isend = .TRUE.
265         CASE DEFAULT
266            WRITE(ldtxt(ii),cform_err)                                                    ;   ii = ii + 1
267            WRITE(ldtxt(ii),*) '           bad value for cn_mpi_send = ', cn_mpi_send     ;   ii = ii + 1
268            kstop = kstop + 1
269         END SELECT
270         !
271      ELSEIF ( PRESENT(localComm) .AND. .NOT. mpi_was_called ) THEN
272         WRITE(ldtxt(ii),cform_err)                                                    ;   ii = ii + 1
273         WRITE(ldtxt(ii),*) ' lib_mpp: You cannot provide a local communicator '          ;   ii = ii + 1
274         WRITE(ldtxt(ii),*) '          without calling MPI_Init before ! '                ;   ii = ii + 1
275         kstop = kstop + 1
276      ELSE
277         SELECT CASE ( cn_mpi_send )
278         CASE ( 'S' )                ! Standard mpi send (blocking)
279            WRITE(ldtxt(ii),*) '           Standard blocking mpi send (send)'             ;   ii = ii + 1
280            CALL mpi_init( ierr )
281         CASE ( 'B' )                ! Buffer mpi send (blocking)
282            WRITE(ldtxt(ii),*) '           Buffer blocking mpi send (bsend)'              ;   ii = ii + 1
283            IF( Agrif_Root() )   CALL mpi_init_oce( ldtxt, ii, ierr )
284         CASE ( 'I' )                ! Immediate mpi send (non-blocking send)
285            WRITE(ldtxt(ii),*) '           Immediate non-blocking send (isend)'           ;   ii = ii + 1
286            l_isend = .TRUE.
287            CALL mpi_init( ierr )
288         CASE DEFAULT
289            WRITE(ldtxt(ii),cform_err)                                                    ;   ii = ii + 1
290            WRITE(ldtxt(ii),*) '           bad value for cn_mpi_send = ', cn_mpi_send     ;   ii = ii + 1
291            kstop = kstop + 1
292         END SELECT
293         !
294      ENDIF
295
296      IF( PRESENT(localComm) ) THEN
297         IF( Agrif_Root() ) THEN
298            mpi_comm_oce = localComm
299         ENDIF
300      ELSE
301         CALL mpi_comm_dup( mpi_comm_world, mpi_comm_oce, code)
302         IF( code /= MPI_SUCCESS ) THEN
303            DO ji = 1, SIZE(ldtxt)
304               IF( TRIM(ldtxt(ji)) /= '' )   WRITE(*,*) ldtxt(ji)      ! control print of mynode
305            END DO
306            WRITE(*, cform_err)
307            WRITE(*, *) ' lib_mpp: Error in routine mpi_comm_dup'
308            CALL mpi_abort( mpi_comm_world, code, ierr )
309         ENDIF
310      ENDIF
311
312#if defined key_agrif
313      IF( Agrif_Root() ) THEN
314         CALL Agrif_MPI_Init(mpi_comm_oce)
315      ELSE
316         CALL Agrif_MPI_set_grid_comm(mpi_comm_oce)
317      ENDIF
318#endif
319
320      CALL mpi_comm_rank( mpi_comm_oce, mpprank, ierr )
321      CALL mpi_comm_size( mpi_comm_oce, mppsize, ierr )
322      mynode = mpprank
323
324      IF( mynode == 0 ) THEN
325         CALL ctl_opn( kumond, TRIM(ldname), 'UNKNOWN', 'FORMATTED', 'SEQUENTIAL', -1, 6, .FALSE. , 1 )
326         WRITE(kumond, nammpp)     
327      ENDIF
328      !
329      CALL MPI_OP_CREATE(DDPDD_MPI, .TRUE., MPI_SUMDD, ierr)
330      !
331   END FUNCTION mynode
332
333   !!----------------------------------------------------------------------
334   !!                   ***  routine mpp_lnk_(2,3,4)d  ***
335   !!
336   !!   * Argument : dummy argument use in mpp_lnk_... routines
337   !!                ptab   :   array or pointer of arrays on which the boundary condition is applied
338   !!                cd_nat :   nature of array grid-points
339   !!                psgn   :   sign used across the north fold boundary
340   !!                kfld   :   optional, number of pt3d arrays
341   !!                cd_mpp :   optional, fill the overlap area only
342   !!                pval   :   optional, background value (used at closed boundaries)
343   !!----------------------------------------------------------------------
344   !
345   !                       !==  2D array and array of 2D pointer  ==!
346   !
347#  define DIM_2d
348#     define ROUTINE_LNK           mpp_lnk_2d
349#     include "mpp_lnk_generic.h90"
350#     undef ROUTINE_LNK
351#     define MULTI
352#     define ROUTINE_LNK           mpp_lnk_2d_ptr
353#     include "mpp_lnk_generic.h90"
354#     undef ROUTINE_LNK
355#     undef MULTI
356#  undef DIM_2d
357   !
358   !                       !==  3D array and array of 3D pointer  ==!
359   !
360#  define DIM_3d
361#     define ROUTINE_LNK           mpp_lnk_3d
362#     include "mpp_lnk_generic.h90"
363#     undef ROUTINE_LNK
364#     define MULTI
365#     define ROUTINE_LNK           mpp_lnk_3d_ptr
366#     include "mpp_lnk_generic.h90"
367#     undef ROUTINE_LNK
368#     undef MULTI
369#  undef DIM_3d
370   !
371   !                       !==  4D array and array of 4D pointer  ==!
372   !
373#  define DIM_4d
374#     define ROUTINE_LNK           mpp_lnk_4d
375#     include "mpp_lnk_generic.h90"
376#     undef ROUTINE_LNK
377#     define MULTI
378#     define ROUTINE_LNK           mpp_lnk_4d_ptr
379#     include "mpp_lnk_generic.h90"
380#     undef ROUTINE_LNK
381#     undef MULTI
382#  undef DIM_4d
383
384   !!----------------------------------------------------------------------
385   !!                   ***  routine mpp_nfd_(2,3,4)d  ***
386   !!
387   !!   * Argument : dummy argument use in mpp_nfd_... routines
388   !!                ptab   :   array or pointer of arrays on which the boundary condition is applied
389   !!                cd_nat :   nature of array grid-points
390   !!                psgn   :   sign used across the north fold boundary
391   !!                kfld   :   optional, number of pt3d arrays
392   !!                cd_mpp :   optional, fill the overlap area only
393   !!                pval   :   optional, background value (used at closed boundaries)
394   !!----------------------------------------------------------------------
395   !
396   !                       !==  2D array and array of 2D pointer  ==!
397   !
398#  define DIM_2d
399#     define ROUTINE_NFD           mpp_nfd_2d
400#     include "mpp_nfd_generic.h90"
401#     undef ROUTINE_NFD
402#     define MULTI
403#     define ROUTINE_NFD           mpp_nfd_2d_ptr
404#     include "mpp_nfd_generic.h90"
405#     undef ROUTINE_NFD
406#     undef MULTI
407#  undef DIM_2d
408   !
409   !                       !==  3D array and array of 3D pointer  ==!
410   !
411#  define DIM_3d
412#     define ROUTINE_NFD           mpp_nfd_3d
413#     include "mpp_nfd_generic.h90"
414#     undef ROUTINE_NFD
415#     define MULTI
416#     define ROUTINE_NFD           mpp_nfd_3d_ptr
417#     include "mpp_nfd_generic.h90"
418#     undef ROUTINE_NFD
419#     undef MULTI
420#  undef DIM_3d
421   !
422   !                       !==  4D array and array of 4D pointer  ==!
423   !
424#  define DIM_4d
425#     define ROUTINE_NFD           mpp_nfd_4d
426#     include "mpp_nfd_generic.h90"
427#     undef ROUTINE_NFD
428#     define MULTI
429#     define ROUTINE_NFD           mpp_nfd_4d_ptr
430#     include "mpp_nfd_generic.h90"
431#     undef ROUTINE_NFD
432#     undef MULTI
433#  undef DIM_4d
434
435
436   !!----------------------------------------------------------------------
437   !!                   ***  routine mpp_lnk_bdy_(2,3,4)d  ***
438   !!
439   !!   * Argument : dummy argument use in mpp_lnk_... routines
440   !!                ptab   :   array or pointer of arrays on which the boundary condition is applied
441   !!                cd_nat :   nature of array grid-points
442   !!                psgn   :   sign used across the north fold boundary
443   !!                kb_bdy :   BDY boundary set
444   !!                kfld   :   optional, number of pt3d arrays
445   !!----------------------------------------------------------------------
446   !
447   !                       !==  2D array and array of 2D pointer  ==!
448   !
449#  define DIM_2d
450#     define ROUTINE_BDY           mpp_lnk_bdy_2d
451#     include "mpp_bdy_generic.h90"
452#     undef ROUTINE_BDY
453#  undef DIM_2d
454   !
455   !                       !==  3D array and array of 3D pointer  ==!
456   !
457#  define DIM_3d
458#     define ROUTINE_BDY           mpp_lnk_bdy_3d
459#     include "mpp_bdy_generic.h90"
460#     undef ROUTINE_BDY
461#  undef DIM_3d
462   !
463   !                       !==  4D array and array of 4D pointer  ==!
464   !
465#  define DIM_4d
466#     define ROUTINE_BDY           mpp_lnk_bdy_4d
467#     include "mpp_bdy_generic.h90"
468#     undef ROUTINE_BDY
469#  undef DIM_4d
470
471   !!----------------------------------------------------------------------
472   !!
473   !!   load_array  &   mpp_lnk_2d_9    à generaliser a 3D et 4D
474   
475   
476   !!    mpp_lnk_sum_2d et 3D   ====>>>>>>   à virer du code !!!!
477   
478   
479   !!----------------------------------------------------------------------
480
481
482
483   SUBROUTINE mppsend( ktyp, pmess, kbytes, kdest, md_req )
484      !!----------------------------------------------------------------------
485      !!                  ***  routine mppsend  ***
486      !!
487      !! ** Purpose :   Send messag passing array
488      !!
489      !!----------------------------------------------------------------------
490      REAL(wp), INTENT(inout) ::   pmess(*)   ! array of real
491      INTEGER , INTENT(in   ) ::   kbytes     ! size of the array pmess
492      INTEGER , INTENT(in   ) ::   kdest      ! receive process number
493      INTEGER , INTENT(in   ) ::   ktyp       ! tag of the message
494      INTEGER , INTENT(in   ) ::   md_req     ! argument for isend
495      !!
496      INTEGER ::   iflag
497      !!----------------------------------------------------------------------
498      !
499      SELECT CASE ( cn_mpi_send )
500      CASE ( 'S' )                ! Standard mpi send (blocking)
501         CALL mpi_send ( pmess, kbytes, mpi_double_precision, kdest , ktyp, mpi_comm_oce        , iflag )
502      CASE ( 'B' )                ! Buffer mpi send (blocking)
503         CALL mpi_bsend( pmess, kbytes, mpi_double_precision, kdest , ktyp, mpi_comm_oce        , iflag )
504      CASE ( 'I' )                ! Immediate mpi send (non-blocking send)
505         ! be carefull, one more argument here : the mpi request identifier..
506         CALL mpi_isend( pmess, kbytes, mpi_double_precision, kdest , ktyp, mpi_comm_oce, md_req, iflag )
507      END SELECT
508      !
509   END SUBROUTINE mppsend
510
511
512   SUBROUTINE mpprecv( ktyp, pmess, kbytes, ksource )
513      !!----------------------------------------------------------------------
514      !!                  ***  routine mpprecv  ***
515      !!
516      !! ** Purpose :   Receive messag passing array
517      !!
518      !!----------------------------------------------------------------------
519      REAL(wp), INTENT(inout) ::   pmess(*)   ! array of real
520      INTEGER , INTENT(in   ) ::   kbytes     ! suze of the array pmess
521      INTEGER , INTENT(in   ) ::   ktyp       ! Tag of the recevied message
522      INTEGER, OPTIONAL, INTENT(in) :: ksource    ! source process number
523      !!
524      INTEGER :: istatus(mpi_status_size)
525      INTEGER :: iflag
526      INTEGER :: use_source
527      !!----------------------------------------------------------------------
528      !
529      ! If a specific process number has been passed to the receive call,
530      ! use that one. Default is to use mpi_any_source
531      use_source = mpi_any_source
532      IF( PRESENT(ksource) )   use_source = ksource
533      !
534      CALL mpi_recv( pmess, kbytes, mpi_double_precision, use_source, ktyp, mpi_comm_oce, istatus, iflag )
535      !
536   END SUBROUTINE mpprecv
537
538
539   SUBROUTINE mppgather( ptab, kp, pio )
540      !!----------------------------------------------------------------------
541      !!                   ***  routine mppgather  ***
542      !!
543      !! ** Purpose :   Transfert between a local subdomain array and a work
544      !!     array which is distributed following the vertical level.
545      !!
546      !!----------------------------------------------------------------------
547      REAL(wp), DIMENSION(jpi,jpj)      , INTENT(in   ) ::   ptab   ! subdomain input array
548      INTEGER                           , INTENT(in   ) ::   kp     ! record length
549      REAL(wp), DIMENSION(jpi,jpj,jpnij), INTENT(  out) ::   pio    ! subdomain input array
550      !!
551      INTEGER :: itaille, ierror   ! temporary integer
552      !!---------------------------------------------------------------------
553      !
554      itaille = jpi * jpj
555      CALL mpi_gather( ptab, itaille, mpi_double_precision, pio, itaille     ,   &
556         &                            mpi_double_precision, kp , mpi_comm_oce, ierror )
557      !
558   END SUBROUTINE mppgather
559
560
561   SUBROUTINE mppscatter( pio, kp, ptab )
562      !!----------------------------------------------------------------------
563      !!                  ***  routine mppscatter  ***
564      !!
565      !! ** Purpose :   Transfert between awork array which is distributed
566      !!      following the vertical level and the local subdomain array.
567      !!
568      !!----------------------------------------------------------------------
569      REAL(wp), DIMENSION(jpi,jpj,jpnij)  ::   pio    ! output array
570      INTEGER                             ::   kp     ! Tag (not used with MPI
571      REAL(wp), DIMENSION(jpi,jpj)        ::   ptab   ! subdomain array input
572      !!
573      INTEGER :: itaille, ierror   ! temporary integer
574      !!---------------------------------------------------------------------
575      !
576      itaille = jpi * jpj
577      !
578      CALL mpi_scatter( pio, itaille, mpi_double_precision, ptab, itaille     ,   &
579         &                            mpi_double_precision, kp  , mpi_comm_oce, ierror )
580      !
581   END SUBROUTINE mppscatter
582
583   
584   SUBROUTINE mpp_delay_sum( cdname, cdelay, y_in, pout, ldlast, kcom )
585     !!----------------------------------------------------------------------
586      !!                   ***  routine mpp_delay_sum  ***
587      !!
588      !! ** Purpose :   performed delayed mpp_sum, the result is received on next call
589      !!
590      !!----------------------------------------------------------------------
591      CHARACTER(len=*), INTENT(in   )               ::   cdname  ! name of the calling subroutine
592      CHARACTER(len=*), INTENT(in   )               ::   cdelay  ! name (used as id) of the delayed operation
593      COMPLEX(wp),      INTENT(in   ), DIMENSION(:) ::   y_in
594      REAL(wp),         INTENT(  out), DIMENSION(:) ::   pout
595      LOGICAL,          INTENT(in   )               ::   ldlast  ! true if this is the last time we call this routine
596      INTEGER,          INTENT(in   ), OPTIONAL     ::   kcom
597      !!
598      INTEGER ::   ji, isz
599      INTEGER ::   idvar
600      INTEGER ::   ierr, ilocalcomm
601      COMPLEX(wp), ALLOCATABLE, DIMENSION(:) ::   ytmp
602      !!----------------------------------------------------------------------
603      ilocalcomm = mpi_comm_oce
604      IF( PRESENT(kcom) )   ilocalcomm = kcom
605
606      isz = SIZE(y_in)
607     
608      IF( narea == 1 .AND. numcom == -1 ) CALL mpp_report( cdname, ld_dlg = .TRUE. )
609
610      idvar = -1
611      DO ji = 1, nbdelay
612         IF( TRIM(cdelay) == TRIM(c_delaylist(ji)) ) idvar = ji
613      END DO
614      IF ( idvar == -1 )   CALL ctl_stop( 'STOP',' mpp_delay_sum : please add a new delayed exchange for '//TRIM(cdname) )
615
616      IF ( ndelayid(idvar) == 0 ) THEN         ! first call    with restart: %z1d defined in iom_delay_rst
617         !                                       --------------------------
618         IF ( SIZE(todelay(idvar)%z1d) /= isz ) THEN                  ! Check dimension coherence
619            IF(lwp) WRITE(numout,*) ' WARNING: the nb of delayed variables in restart file is not the model one'
620            DEALLOCATE(todelay(idvar)%z1d)
621            ndelayid(idvar) = -1                                      ! do as if we had no restart
622         ELSE
623            ALLOCATE(todelay(idvar)%y1d(isz))
624            todelay(idvar)%y1d(:) = CMPLX(todelay(idvar)%z1d(:), 0., wp)   ! create %y1d, complex variable needed by mpi_sumdd
625         END IF
626      ENDIF
627     
628      IF( ndelayid(idvar) == -1 ) THEN         ! first call without restart: define %y1d and %z1d from y_in with blocking allreduce
629         !                                       --------------------------
630         ALLOCATE(todelay(idvar)%z1d(isz), todelay(idvar)%y1d(isz))   ! allocate also %z1d as used for the restart
631         CALL mpi_allreduce( y_in(:), todelay(idvar)%y1d(:), isz, MPI_DOUBLE_COMPLEX, mpi_sumdd, ilocalcomm, ierr )   ! get %y1d
632         todelay(idvar)%z1d(:) = REAL(todelay(idvar)%y1d(:), wp)      ! define %z1d from %y1d
633      ENDIF
634
635      IF( ndelayid(idvar) > 0 )   CALL mpp_delay_rcv( idvar )         ! make sure %z1d is received
636
637      ! send back pout from todelay(idvar)%z1d defined at previous call
638      pout(:) = todelay(idvar)%z1d(:)
639
640      ! send y_in into todelay(idvar)%y1d with a non-blocking communication
641#if defined key_mpi2
642      IF( ln_timing ) CALL tic_tac( .TRUE., ld_global = .TRUE.)
643      CALL  mpi_allreduce( y_in(:), todelay(idvar)%y1d(:), isz, MPI_DOUBLE_COMPLEX, mpi_sumdd, ilocalcomm, ndelayid(idvar), ierr )
644      IF( ln_timing ) CALL tic_tac(.FALSE., ld_global = .TRUE.)
645#else
646      CALL mpi_iallreduce( y_in(:), todelay(idvar)%y1d(:), isz, MPI_DOUBLE_COMPLEX, mpi_sumdd, ilocalcomm, ndelayid(idvar), ierr )
647#endif
648
649   END SUBROUTINE mpp_delay_sum
650
651   
652   SUBROUTINE mpp_delay_max( cdname, cdelay, p_in, pout, ldlast, kcom )
653      !!----------------------------------------------------------------------
654      !!                   ***  routine mpp_delay_max  ***
655      !!
656      !! ** Purpose :   performed delayed mpp_max, the result is received on next call
657      !!
658      !!----------------------------------------------------------------------
659      CHARACTER(len=*), INTENT(in   )                 ::   cdname  ! name of the calling subroutine
660      CHARACTER(len=*), INTENT(in   )                 ::   cdelay  ! name (used as id) of the delayed operation
661      REAL(wp),         INTENT(in   ), DIMENSION(:)   ::   p_in    !
662      REAL(wp),         INTENT(  out), DIMENSION(:)   ::   pout    !
663      LOGICAL,          INTENT(in   )                 ::   ldlast  ! true if this is the last time we call this routine
664      INTEGER,          INTENT(in   ), OPTIONAL       ::   kcom
665      !!
666      INTEGER ::   ji, isz
667      INTEGER ::   idvar
668      INTEGER ::   ierr, ilocalcomm
669      !!----------------------------------------------------------------------
670      ilocalcomm = mpi_comm_oce
671      IF( PRESENT(kcom) )   ilocalcomm = kcom
672
673      isz = SIZE(p_in)
674
675      IF( narea == 1 .AND. numcom == -1 ) CALL mpp_report( cdname, ld_dlg = .TRUE. )
676
677      idvar = -1
678      DO ji = 1, nbdelay
679         IF( TRIM(cdelay) == TRIM(c_delaylist(ji)) ) idvar = ji
680      END DO
681      IF ( idvar == -1 )   CALL ctl_stop( 'STOP',' mpp_delay_max : please add a new delayed exchange for '//TRIM(cdname) )
682
683      IF ( ndelayid(idvar) == 0 ) THEN         ! first call    with restart: %z1d defined in iom_delay_rst
684         !                                       --------------------------
685         IF ( SIZE(todelay(idvar)%z1d) /= isz ) THEN                  ! Check dimension coherence
686            IF(lwp) WRITE(numout,*) ' WARNING: the nb of delayed variables in restart file is not the model one'
687            DEALLOCATE(todelay(idvar)%z1d)
688            ndelayid(idvar) = -1                                      ! do as if we had no restart
689         END IF
690      ENDIF
691
692      IF( ndelayid(idvar) == -1 ) THEN         ! first call without restart: define %z1d from p_in with a blocking allreduce
693         !                                       --------------------------
694         ALLOCATE(todelay(idvar)%z1d(isz))
695         CALL mpi_allreduce( p_in(:), todelay(idvar)%z1d(:), isz, MPI_DOUBLE_PRECISION, mpi_max, ilocalcomm, ierr )   ! get %z1d
696      ENDIF
697
698      IF( ndelayid(idvar) > 0 )   CALL mpp_delay_rcv( idvar )         ! make sure %z1d is received
699
700      ! send back pout from todelay(idvar)%z1d defined at previous call
701      pout(:) = todelay(idvar)%z1d(:)
702
703      ! send p_in into todelay(idvar)%z1d with a non-blocking communication
704#if defined key_mpi2
705      IF( ln_timing ) CALL tic_tac( .TRUE., ld_global = .TRUE.)
706      CALL  mpi_allreduce( p_in(:), todelay(idvar)%z1d(:), isz, MPI_DOUBLE_PRECISION, mpi_max, ilocalcomm, ndelayid(idvar), ierr )
707      IF( ln_timing ) CALL tic_tac(.FALSE., ld_global = .TRUE.)
708#else
709      CALL mpi_iallreduce( p_in(:), todelay(idvar)%z1d(:), isz, MPI_DOUBLE_PRECISION, mpi_max, ilocalcomm, ndelayid(idvar), ierr )
710#endif
711
712   END SUBROUTINE mpp_delay_max
713
714   
715   SUBROUTINE mpp_delay_rcv( kid )
716      !!----------------------------------------------------------------------
717      !!                   ***  routine mpp_delay_rcv  ***
718      !!
719      !! ** Purpose :  force barrier for delayed mpp (needed for restart)
720      !!
721      !!----------------------------------------------------------------------
722      INTEGER,INTENT(in   )      ::  kid 
723      INTEGER ::   ierr
724      !!----------------------------------------------------------------------
725      IF( ndelayid(kid) /= -2 ) THEN 
726#if ! defined key_mpi2
727         IF( ln_timing ) CALL tic_tac( .TRUE., ld_global = .TRUE.)
728         CALL mpi_wait( ndelayid(kid), MPI_STATUS_IGNORE, ierr )                        ! make sure todelay(kid) is received
729         IF( ln_timing ) CALL tic_tac(.FALSE., ld_global = .TRUE.)
730#endif
731         IF( ASSOCIATED(todelay(kid)%y1d) )   todelay(kid)%z1d(:) = REAL(todelay(kid)%y1d(:), wp)  ! define %z1d from %y1d
732         ndelayid(kid) = -2   ! add flag to know that mpi_wait was already called on kid
733      ENDIF
734   END SUBROUTINE mpp_delay_rcv
735
736   
737   !!----------------------------------------------------------------------
738   !!    ***  mppmax_a_int, mppmax_int, mppmax_a_real, mppmax_real  ***
739   !!   
740   !!----------------------------------------------------------------------
741   !!
742#  define OPERATION_MAX
743#  define INTEGER_TYPE
744#  define DIM_0d
745#     define ROUTINE_ALLREDUCE           mppmax_int
746#     include "mpp_allreduce_generic.h90"
747#     undef ROUTINE_ALLREDUCE
748#  undef DIM_0d
749#  define DIM_1d
750#     define ROUTINE_ALLREDUCE           mppmax_a_int
751#     include "mpp_allreduce_generic.h90"
752#     undef ROUTINE_ALLREDUCE
753#  undef DIM_1d
754#  undef INTEGER_TYPE
755!
756#  define REAL_TYPE
757#  define DIM_0d
758#     define ROUTINE_ALLREDUCE           mppmax_real
759#     include "mpp_allreduce_generic.h90"
760#     undef ROUTINE_ALLREDUCE
761#  undef DIM_0d
762#  define DIM_1d
763#     define ROUTINE_ALLREDUCE           mppmax_a_real
764#     include "mpp_allreduce_generic.h90"
765#     undef ROUTINE_ALLREDUCE
766#  undef DIM_1d
767#  undef REAL_TYPE
768#  undef OPERATION_MAX
769   !!----------------------------------------------------------------------
770   !!    ***  mppmin_a_int, mppmin_int, mppmin_a_real, mppmin_real  ***
771   !!   
772   !!----------------------------------------------------------------------
773   !!
774#  define OPERATION_MIN
775#  define INTEGER_TYPE
776#  define DIM_0d
777#     define ROUTINE_ALLREDUCE           mppmin_int
778#     include "mpp_allreduce_generic.h90"
779#     undef ROUTINE_ALLREDUCE
780#  undef DIM_0d
781#  define DIM_1d
782#     define ROUTINE_ALLREDUCE           mppmin_a_int
783#     include "mpp_allreduce_generic.h90"
784#     undef ROUTINE_ALLREDUCE
785#  undef DIM_1d
786#  undef INTEGER_TYPE
787!
788#  define REAL_TYPE
789#  define DIM_0d
790#     define ROUTINE_ALLREDUCE           mppmin_real
791#     include "mpp_allreduce_generic.h90"
792#     undef ROUTINE_ALLREDUCE
793#  undef DIM_0d
794#  define DIM_1d
795#     define ROUTINE_ALLREDUCE           mppmin_a_real
796#     include "mpp_allreduce_generic.h90"
797#     undef ROUTINE_ALLREDUCE
798#  undef DIM_1d
799#  undef REAL_TYPE
800#  undef OPERATION_MIN
801
802   !!----------------------------------------------------------------------
803   !!    ***  mppsum_a_int, mppsum_int, mppsum_a_real, mppsum_real  ***
804   !!   
805   !!   Global sum of 1D array or a variable (integer, real or complex)
806   !!----------------------------------------------------------------------
807   !!
808#  define OPERATION_SUM
809#  define INTEGER_TYPE
810#  define DIM_0d
811#     define ROUTINE_ALLREDUCE           mppsum_int
812#     include "mpp_allreduce_generic.h90"
813#     undef ROUTINE_ALLREDUCE
814#  undef DIM_0d
815#  define DIM_1d
816#     define ROUTINE_ALLREDUCE           mppsum_a_int
817#     include "mpp_allreduce_generic.h90"
818#     undef ROUTINE_ALLREDUCE
819#  undef DIM_1d
820#  undef INTEGER_TYPE
821!
822#  define REAL_TYPE
823#  define DIM_0d
824#     define ROUTINE_ALLREDUCE           mppsum_real
825#     include "mpp_allreduce_generic.h90"
826#     undef ROUTINE_ALLREDUCE
827#  undef DIM_0d
828#  define DIM_1d
829#     define ROUTINE_ALLREDUCE           mppsum_a_real
830#     include "mpp_allreduce_generic.h90"
831#     undef ROUTINE_ALLREDUCE
832#  undef DIM_1d
833#  undef REAL_TYPE
834#  undef OPERATION_SUM
835
836#  define OPERATION_SUM_DD
837#  define COMPLEX_TYPE
838#  define DIM_0d
839#     define ROUTINE_ALLREDUCE           mppsum_realdd
840#     include "mpp_allreduce_generic.h90"
841#     undef ROUTINE_ALLREDUCE
842#  undef DIM_0d
843#  define DIM_1d
844#     define ROUTINE_ALLREDUCE           mppsum_a_realdd
845#     include "mpp_allreduce_generic.h90"
846#     undef ROUTINE_ALLREDUCE
847#  undef DIM_1d
848#  undef COMPLEX_TYPE
849#  undef OPERATION_SUM_DD
850
851   !!----------------------------------------------------------------------
852   !!    ***  mpp_minloc2d, mpp_minloc3d, mpp_maxloc2d, mpp_maxloc3d
853   !!   
854   !!----------------------------------------------------------------------
855   !!
856#  define OPERATION_MINLOC
857#  define DIM_2d
858#     define ROUTINE_LOC           mpp_minloc2d
859#     include "mpp_loc_generic.h90"
860#     undef ROUTINE_LOC
861#  undef DIM_2d
862#  define DIM_3d
863#     define ROUTINE_LOC           mpp_minloc3d
864#     include "mpp_loc_generic.h90"
865#     undef ROUTINE_LOC
866#  undef DIM_3d
867#  undef OPERATION_MINLOC
868
869#  define OPERATION_MAXLOC
870#  define DIM_2d
871#     define ROUTINE_LOC           mpp_maxloc2d
872#     include "mpp_loc_generic.h90"
873#     undef ROUTINE_LOC
874#  undef DIM_2d
875#  define DIM_3d
876#     define ROUTINE_LOC           mpp_maxloc3d
877#     include "mpp_loc_generic.h90"
878#     undef ROUTINE_LOC
879#  undef DIM_3d
880#  undef OPERATION_MAXLOC
881
882   SUBROUTINE mppsync()
883      !!----------------------------------------------------------------------
884      !!                  ***  routine mppsync  ***
885      !!
886      !! ** Purpose :   Massively parallel processors, synchroneous
887      !!
888      !!-----------------------------------------------------------------------
889      INTEGER :: ierror
890      !!-----------------------------------------------------------------------
891      !
892      CALL mpi_barrier( mpi_comm_oce, ierror )
893      !
894   END SUBROUTINE mppsync
895
896
897   SUBROUTINE mppstop( ldfinal, ld_force_abort ) 
898      !!----------------------------------------------------------------------
899      !!                  ***  routine mppstop  ***
900      !!
901      !! ** purpose :   Stop massively parallel processors method
902      !!
903      !!----------------------------------------------------------------------
904      LOGICAL, OPTIONAL, INTENT(in) :: ldfinal    ! source process number
905      LOGICAL, OPTIONAL, INTENT(in) :: ld_force_abort    ! source process number
906      LOGICAL ::   llfinal, ll_force_abort
907      INTEGER ::   info
908      !!----------------------------------------------------------------------
909      llfinal = .FALSE.
910      IF( PRESENT(ldfinal) ) llfinal = ldfinal
911      ll_force_abort = .FALSE.
912      IF( PRESENT(ld_force_abort) ) ll_force_abort = ld_force_abort
913      !
914      IF(ll_force_abort) THEN
915         CALL mpi_abort( MPI_COMM_WORLD )
916      ELSE
917         CALL mppsync
918         CALL mpi_finalize( info )
919      ENDIF
920      IF( .NOT. llfinal ) STOP 123
921      !
922   END SUBROUTINE mppstop
923
924
925   SUBROUTINE mpp_comm_free( kcom )
926      !!----------------------------------------------------------------------
927      INTEGER, INTENT(in) ::   kcom
928      !!
929      INTEGER :: ierr
930      !!----------------------------------------------------------------------
931      !
932      CALL MPI_COMM_FREE(kcom, ierr)
933      !
934   END SUBROUTINE mpp_comm_free
935
936
937   SUBROUTINE mpp_ini_znl( kumout )
938      !!----------------------------------------------------------------------
939      !!               ***  routine mpp_ini_znl  ***
940      !!
941      !! ** Purpose :   Initialize special communicator for computing zonal sum
942      !!
943      !! ** Method  : - Look for processors in the same row
944      !!              - Put their number in nrank_znl
945      !!              - Create group for the znl processors
946      !!              - Create a communicator for znl processors
947      !!              - Determine if processor should write znl files
948      !!
949      !! ** output
950      !!      ndim_rank_znl = number of processors on the same row
951      !!      ngrp_znl = group ID for the znl processors
952      !!      ncomm_znl = communicator for the ice procs.
953      !!      n_znl_root = number (in the world) of proc 0 in the ice comm.
954      !!
955      !!----------------------------------------------------------------------
956      INTEGER, INTENT(in) ::   kumout   ! ocean.output logical units
957      !
958      INTEGER :: jproc      ! dummy loop integer
959      INTEGER :: ierr, ii   ! local integer
960      INTEGER, ALLOCATABLE, DIMENSION(:) ::   kwork
961      !!----------------------------------------------------------------------
962      !-$$     WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ngrp_world     : ', ngrp_world
963      !-$$     WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - mpi_comm_world : ', mpi_comm_world
964      !-$$     WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - mpi_comm_oce   : ', mpi_comm_oce
965      !
966      ALLOCATE( kwork(jpnij), STAT=ierr )
967      IF( ierr /= 0 ) THEN
968         WRITE(kumout, cform_err)
969         WRITE(kumout,*) 'mpp_ini_znl : failed to allocate 1D array of length jpnij'
970         CALL mppstop
971      ENDIF
972
973      IF( jpnj == 1 ) THEN
974         ngrp_znl  = ngrp_world
975         ncomm_znl = mpi_comm_oce
976      ELSE
977         !
978         CALL MPI_ALLGATHER ( njmpp, 1, mpi_integer, kwork, 1, mpi_integer, mpi_comm_oce, ierr )
979         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - kwork pour njmpp : ', kwork
980         !-$$        CALL flush(numout)
981         !
982         ! Count number of processors on the same row
983         ndim_rank_znl = 0
984         DO jproc=1,jpnij
985            IF ( kwork(jproc) == njmpp ) THEN
986               ndim_rank_znl = ndim_rank_znl + 1
987            ENDIF
988         END DO
989         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ndim_rank_znl : ', ndim_rank_znl
990         !-$$        CALL flush(numout)
991         ! Allocate the right size to nrank_znl
992         IF (ALLOCATED (nrank_znl)) DEALLOCATE(nrank_znl)
993         ALLOCATE(nrank_znl(ndim_rank_znl))
994         ii = 0
995         nrank_znl (:) = 0
996         DO jproc=1,jpnij
997            IF ( kwork(jproc) == njmpp) THEN
998               ii = ii + 1
999               nrank_znl(ii) = jproc -1
1000            ENDIF
1001         END DO
1002         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - nrank_znl : ', nrank_znl
1003         !-$$        CALL flush(numout)
1004
1005         ! Create the opa group
1006         CALL MPI_COMM_GROUP(mpi_comm_oce,ngrp_opa,ierr)
1007         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ngrp_opa : ', ngrp_opa
1008         !-$$        CALL flush(numout)
1009
1010         ! Create the znl group from the opa group
1011         CALL MPI_GROUP_INCL  ( ngrp_opa, ndim_rank_znl, nrank_znl, ngrp_znl, ierr )
1012         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ngrp_znl ', ngrp_znl
1013         !-$$        CALL flush(numout)
1014
1015         ! Create the znl communicator from the opa communicator, ie the pool of procs in the same row
1016         CALL MPI_COMM_CREATE ( mpi_comm_oce, ngrp_znl, ncomm_znl, ierr )
1017         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ncomm_znl ', ncomm_znl
1018         !-$$        CALL flush(numout)
1019         !
1020      END IF
1021
1022      ! Determines if processor if the first (starting from i=1) on the row
1023      IF ( jpni == 1 ) THEN
1024         l_znl_root = .TRUE.
1025      ELSE
1026         l_znl_root = .FALSE.
1027         kwork (1) = nimpp
1028         CALL mpp_min ( 'lib_mpp', kwork(1), kcom = ncomm_znl)
1029         IF ( nimpp == kwork(1)) l_znl_root = .TRUE.
1030      END IF
1031
1032      DEALLOCATE(kwork)
1033
1034   END SUBROUTINE mpp_ini_znl
1035
1036
1037   SUBROUTINE mpp_ini_north
1038      !!----------------------------------------------------------------------
1039      !!               ***  routine mpp_ini_north  ***
1040      !!
1041      !! ** Purpose :   Initialize special communicator for north folding
1042      !!      condition together with global variables needed in the mpp folding
1043      !!
1044      !! ** Method  : - Look for northern processors
1045      !!              - Put their number in nrank_north
1046      !!              - Create groups for the world processors and the north processors
1047      !!              - Create a communicator for northern processors
1048      !!
1049      !! ** output
1050      !!      njmppmax = njmpp for northern procs
1051      !!      ndim_rank_north = number of processors in the northern line
1052      !!      nrank_north (ndim_rank_north) = number  of the northern procs.
1053      !!      ngrp_world = group ID for the world processors
1054      !!      ngrp_north = group ID for the northern processors
1055      !!      ncomm_north = communicator for the northern procs.
1056      !!      north_root = number (in the world) of proc 0 in the northern comm.
1057      !!
1058      !!----------------------------------------------------------------------
1059      INTEGER ::   ierr
1060      INTEGER ::   jjproc
1061      INTEGER ::   ii, ji
1062      !!----------------------------------------------------------------------
1063      !
1064      njmppmax = MAXVAL( njmppt )
1065      !
1066      ! Look for how many procs on the northern boundary
1067      ndim_rank_north = 0
1068      DO jjproc = 1, jpnij
1069         IF( njmppt(jjproc) == njmppmax )   ndim_rank_north = ndim_rank_north + 1
1070      END DO
1071      !
1072      ! Allocate the right size to nrank_north
1073      IF (ALLOCATED (nrank_north)) DEALLOCATE(nrank_north)
1074      ALLOCATE( nrank_north(ndim_rank_north) )
1075
1076      ! Fill the nrank_north array with proc. number of northern procs.
1077      ! Note : the rank start at 0 in MPI
1078      ii = 0
1079      DO ji = 1, jpnij
1080         IF ( njmppt(ji) == njmppmax   ) THEN
1081            ii=ii+1
1082            nrank_north(ii)=ji-1
1083         END IF
1084      END DO
1085      !
1086      ! create the world group
1087      CALL MPI_COMM_GROUP( mpi_comm_oce, ngrp_world, ierr )
1088      !
1089      ! Create the North group from the world group
1090      CALL MPI_GROUP_INCL( ngrp_world, ndim_rank_north, nrank_north, ngrp_north, ierr )
1091      !
1092      ! Create the North communicator , ie the pool of procs in the north group
1093      CALL MPI_COMM_CREATE( mpi_comm_oce, ngrp_north, ncomm_north, ierr )
1094      !
1095   END SUBROUTINE mpp_ini_north
1096
1097
1098   SUBROUTINE mpi_init_oce( ldtxt, ksft, code )
1099      !!---------------------------------------------------------------------
1100      !!                   ***  routine mpp_init.opa  ***
1101      !!
1102      !! ** Purpose :: export and attach a MPI buffer for bsend
1103      !!
1104      !! ** Method  :: define buffer size in namelist, if 0 no buffer attachment
1105      !!            but classical mpi_init
1106      !!
1107      !! History :: 01/11 :: IDRIS initial version for IBM only
1108      !!            08/04 :: R. Benshila, generalisation
1109      !!---------------------------------------------------------------------
1110      CHARACTER(len=*),DIMENSION(:), INTENT(  out) ::   ldtxt
1111      INTEGER                      , INTENT(inout) ::   ksft
1112      INTEGER                      , INTENT(  out) ::   code
1113      INTEGER                                      ::   ierr, ji
1114      LOGICAL                                      ::   mpi_was_called
1115      !!---------------------------------------------------------------------
1116      !
1117      CALL mpi_initialized( mpi_was_called, code )      ! MPI initialization
1118      IF ( code /= MPI_SUCCESS ) THEN
1119         DO ji = 1, SIZE(ldtxt)
1120            IF( TRIM(ldtxt(ji)) /= '' )   WRITE(*,*) ldtxt(ji)      ! control print of mynode
1121         END DO
1122         WRITE(*, cform_err)
1123         WRITE(*, *) ' lib_mpp: Error in routine mpi_initialized'
1124         CALL mpi_abort( mpi_comm_world, code, ierr )
1125      ENDIF
1126      !
1127      IF( .NOT. mpi_was_called ) THEN
1128         CALL mpi_init( code )
1129         CALL mpi_comm_dup( mpi_comm_world, mpi_comm_oce, code )
1130         IF ( code /= MPI_SUCCESS ) THEN
1131            DO ji = 1, SIZE(ldtxt)
1132               IF( TRIM(ldtxt(ji)) /= '' )   WRITE(*,*) ldtxt(ji)      ! control print of mynode
1133            END DO
1134            WRITE(*, cform_err)
1135            WRITE(*, *) ' lib_mpp: Error in routine mpi_comm_dup'
1136            CALL mpi_abort( mpi_comm_world, code, ierr )
1137         ENDIF
1138      ENDIF
1139      !
1140      IF( nn_buffer > 0 ) THEN
1141         WRITE(ldtxt(ksft),*) 'mpi_bsend, buffer allocation of  : ', nn_buffer   ;   ksft = ksft + 1
1142         ! Buffer allocation and attachment
1143         ALLOCATE( tampon(nn_buffer), stat = ierr )
1144         IF( ierr /= 0 ) THEN
1145            DO ji = 1, SIZE(ldtxt)
1146               IF( TRIM(ldtxt(ji)) /= '' )   WRITE(*,*) ldtxt(ji)      ! control print of mynode
1147            END DO
1148            WRITE(*, cform_err)
1149            WRITE(*, *) ' lib_mpp: Error in ALLOCATE', ierr
1150            CALL mpi_abort( mpi_comm_world, code, ierr )
1151         END IF
1152         CALL mpi_buffer_attach( tampon, nn_buffer, code )
1153      ENDIF
1154      !
1155   END SUBROUTINE mpi_init_oce
1156
1157
1158   SUBROUTINE DDPDD_MPI( ydda, yddb, ilen, itype )
1159      !!---------------------------------------------------------------------
1160      !!   Routine DDPDD_MPI: used by reduction operator MPI_SUMDD
1161      !!
1162      !!   Modification of original codes written by David H. Bailey
1163      !!   This subroutine computes yddb(i) = ydda(i)+yddb(i)
1164      !!---------------------------------------------------------------------
1165      INTEGER                     , INTENT(in)    ::   ilen, itype
1166      COMPLEX(wp), DIMENSION(ilen), INTENT(in)    ::   ydda
1167      COMPLEX(wp), DIMENSION(ilen), INTENT(inout) ::   yddb
1168      !
1169      REAL(wp) :: zerr, zt1, zt2    ! local work variables
1170      INTEGER  :: ji, ztmp           ! local scalar
1171      !!---------------------------------------------------------------------
1172      !
1173      ztmp = itype   ! avoid compilation warning
1174      !
1175      DO ji=1,ilen
1176      ! Compute ydda + yddb using Knuth's trick.
1177         zt1  = real(ydda(ji)) + real(yddb(ji))
1178         zerr = zt1 - real(ydda(ji))
1179         zt2  = ((real(yddb(ji)) - zerr) + (real(ydda(ji)) - (zt1 - zerr))) &
1180                + aimag(ydda(ji)) + aimag(yddb(ji))
1181
1182         ! The result is zt1 + zt2, after normalization.
1183         yddb(ji) = cmplx ( zt1 + zt2, zt2 - ((zt1 + zt2) - zt1),wp )
1184      END DO
1185      !
1186   END SUBROUTINE DDPDD_MPI
1187
1188
1189   SUBROUTINE mpp_lbc_north_icb( pt2d, cd_type, psgn, kextj)
1190      !!---------------------------------------------------------------------
1191      !!                   ***  routine mpp_lbc_north_icb  ***
1192      !!
1193      !! ** Purpose :   Ensure proper north fold horizontal bondary condition
1194      !!              in mpp configuration in case of jpn1 > 1 and for 2d
1195      !!              array with outer extra halo
1196      !!
1197      !! ** Method  :   North fold condition and mpp with more than one proc
1198      !!              in i-direction require a specific treatment. We gather
1199      !!              the 4+kextj northern lines of the global domain on 1
1200      !!              processor and apply lbc north-fold on this sub array.
1201      !!              Then we scatter the north fold array back to the processors.
1202      !!              This routine accounts for an extra halo with icebergs
1203      !!              and assumes ghost rows and columns have been suppressed.
1204      !!
1205      !!----------------------------------------------------------------------
1206      REAL(wp), DIMENSION(:,:), INTENT(inout) ::   pt2d     ! 2D array with extra halo
1207      CHARACTER(len=1)        , INTENT(in   ) ::   cd_type  ! nature of pt3d grid-points
1208      !                                                     !   = T ,  U , V , F or W -points
1209      REAL(wp)                , INTENT(in   ) ::   psgn     ! = -1. the sign change across the
1210      !!                                                    ! north fold, =  1. otherwise
1211      INTEGER                 , INTENT(in   ) ::   kextj    ! Extra halo width at north fold
1212      !
1213      INTEGER ::   ji, jj, jr
1214      INTEGER ::   ierr, itaille, ildi, ilei, iilb
1215      INTEGER ::   ipj, ij, iproc
1216      !
1217      REAL(wp), DIMENSION(:,:)  , ALLOCATABLE  ::  ztab_e, znorthloc_e
1218      REAL(wp), DIMENSION(:,:,:), ALLOCATABLE  ::  znorthgloio_e
1219      !!----------------------------------------------------------------------
1220      !
1221      ipj=4
1222      ALLOCATE(        ztab_e(jpiglo, 1-kextj:ipj+kextj)       ,       &
1223     &            znorthloc_e(jpimax, 1-kextj:ipj+kextj)       ,       &
1224     &          znorthgloio_e(jpimax, 1-kextj:ipj+kextj,jpni)    )
1225      !
1226      ztab_e(:,:)      = 0._wp
1227      znorthloc_e(:,:) = 0._wp
1228      !
1229      ij = 1 - kextj
1230      ! put the last ipj+2*kextj lines of pt2d into znorthloc_e
1231      DO jj = jpj - ipj + 1 - kextj , jpj + kextj
1232         znorthloc_e(1:jpi,ij)=pt2d(1:jpi,jj)
1233         ij = ij + 1
1234      END DO
1235      !
1236      itaille = jpimax * ( ipj + 2*kextj )
1237      !
1238      IF( ln_timing ) CALL tic_tac(.TRUE.)
1239      CALL MPI_ALLGATHER( znorthloc_e(1,1-kextj)    , itaille, MPI_DOUBLE_PRECISION,    &
1240         &                znorthgloio_e(1,1-kextj,1), itaille, MPI_DOUBLE_PRECISION,    &
1241         &                ncomm_north, ierr )
1242      !
1243      IF( ln_timing ) CALL tic_tac(.FALSE.)
1244      !
1245      DO jr = 1, ndim_rank_north            ! recover the global north array
1246         iproc = nrank_north(jr) + 1
1247         ildi = nldit (iproc)
1248         ilei = nleit (iproc)
1249         iilb = nimppt(iproc)
1250         DO jj = 1-kextj, ipj+kextj
1251            DO ji = ildi, ilei
1252               ztab_e(ji+iilb-1,jj) = znorthgloio_e(ji,jj,jr)
1253            END DO
1254         END DO
1255      END DO
1256
1257      ! 2. North-Fold boundary conditions
1258      ! ----------------------------------
1259      CALL lbc_nfd( ztab_e(:,1-kextj:ipj+kextj), cd_type, psgn, kextj )
1260
1261      ij = 1 - kextj
1262      !! Scatter back to pt2d
1263      DO jj = jpj - ipj + 1 - kextj , jpj + kextj
1264         DO ji= 1, jpi
1265            pt2d(ji,jj) = ztab_e(ji+nimpp-1,ij)
1266         END DO
1267         ij  = ij +1
1268      END DO
1269      !
1270      DEALLOCATE( ztab_e, znorthloc_e, znorthgloio_e )
1271      !
1272   END SUBROUTINE mpp_lbc_north_icb
1273
1274
1275   SUBROUTINE mpp_lnk_2d_icb( cdname, pt2d, cd_type, psgn, kexti, kextj )
1276      !!----------------------------------------------------------------------
1277      !!                  ***  routine mpp_lnk_2d_icb  ***
1278      !!
1279      !! ** Purpose :   Message passing management for 2d array (with extra halo for icebergs)
1280      !!                This routine receives a (1-kexti:jpi+kexti,1-kexti:jpj+kextj)
1281      !!                array (usually (0:jpi+1, 0:jpj+1)) from lbc_lnk_icb calls.
1282      !!
1283      !! ** Method  :   Use mppsend and mpprecv function for passing mask
1284      !!      between processors following neighboring subdomains.
1285      !!            domain parameters
1286      !!                    jpi    : first dimension of the local subdomain
1287      !!                    jpj    : second dimension of the local subdomain
1288      !!                    kexti  : number of columns for extra outer halo
1289      !!                    kextj  : number of rows for extra outer halo
1290      !!                    nbondi : mark for "east-west local boundary"
1291      !!                    nbondj : mark for "north-south local boundary"
1292      !!                    noea   : number for local neighboring processors
1293      !!                    nowe   : number for local neighboring processors
1294      !!                    noso   : number for local neighboring processors
1295      !!                    nono   : number for local neighboring processors
1296      !!----------------------------------------------------------------------
1297      CHARACTER(len=*)                                        , INTENT(in   ) ::   cdname      ! name of the calling subroutine
1298      REAL(wp), DIMENSION(1-kexti:jpi+kexti,1-kextj:jpj+kextj), INTENT(inout) ::   pt2d     ! 2D array with extra halo
1299      CHARACTER(len=1)                                        , INTENT(in   ) ::   cd_type  ! nature of ptab array grid-points
1300      REAL(wp)                                                , INTENT(in   ) ::   psgn     ! sign used across the north fold
1301      INTEGER                                                 , INTENT(in   ) ::   kexti    ! extra i-halo width
1302      INTEGER                                                 , INTENT(in   ) ::   kextj    ! extra j-halo width
1303      !
1304      INTEGER  ::   jl   ! dummy loop indices
1305      INTEGER  ::   imigr, iihom, ijhom        ! local integers
1306      INTEGER  ::   ipreci, iprecj             !   -       -
1307      INTEGER  ::   ml_req1, ml_req2, ml_err   ! for key_mpi_isend
1308      INTEGER, DIMENSION(MPI_STATUS_SIZE) ::   ml_stat   ! for key_mpi_isend
1309      !!
1310      REAL(wp), DIMENSION(1-kexti:jpi+kexti,nn_hls+kextj,2) ::   r2dns, r2dsn
1311      REAL(wp), DIMENSION(1-kextj:jpj+kextj,nn_hls+kexti,2) ::   r2dwe, r2dew
1312      !!----------------------------------------------------------------------
1313
1314      ipreci = nn_hls + kexti      ! take into account outer extra 2D overlap area
1315      iprecj = nn_hls + kextj
1316
1317      IF( narea == 1 .AND. numcom == -1 ) CALL mpp_report( cdname, 1, 1, 1, ld_lbc = .TRUE. )
1318
1319      ! 1. standard boundary treatment
1320      ! ------------------------------
1321      ! Order matters Here !!!!
1322      !
1323      !                                      ! East-West boundaries
1324      !                                           !* Cyclic east-west
1325      IF( l_Iperio ) THEN
1326         pt2d(1-kexti:     1   ,:) = pt2d(jpim1-kexti: jpim1 ,:)       ! east
1327         pt2d(  jpi  :jpi+kexti,:) = pt2d(     2     :2+kexti,:)       ! west
1328         !
1329      ELSE                                        !* closed
1330         IF( .NOT. cd_type == 'F' )   pt2d(  1-kexti   :nn_hls   ,:) = 0._wp    ! east except at F-point
1331                                      pt2d(jpi-nn_hls+1:jpi+kexti,:) = 0._wp    ! west
1332      ENDIF
1333      !                                      ! North-South boundaries
1334      IF( l_Jperio ) THEN                         !* cyclic (only with no mpp j-split)
1335         pt2d(:,1-kextj:     1   ) = pt2d(:,jpjm1-kextj:  jpjm1)       ! north
1336         pt2d(:,  jpj  :jpj+kextj) = pt2d(:,     2     :2+kextj)       ! south
1337      ELSE                                        !* closed
1338         IF( .NOT. cd_type == 'F' )   pt2d(:,  1-kextj   :nn_hls   ) = 0._wp    ! north except at F-point
1339                                      pt2d(:,jpj-nn_hls+1:jpj+kextj) = 0._wp    ! south
1340      ENDIF
1341      !
1342
1343      ! north fold treatment
1344      ! -----------------------
1345      IF( npolj /= 0 ) THEN
1346         !
1347         SELECT CASE ( jpni )
1348                   CASE ( 1 )     ;   CALL lbc_nfd          ( pt2d(1:jpi,1:jpj+kextj), cd_type, psgn, kextj )
1349                   CASE DEFAULT   ;   CALL mpp_lbc_north_icb( pt2d(1:jpi,1:jpj+kextj), cd_type, psgn, kextj )
1350         END SELECT
1351         !
1352      ENDIF
1353
1354      ! 2. East and west directions exchange
1355      ! ------------------------------------
1356      ! we play with the neigbours AND the row number because of the periodicity
1357      !
1358      SELECT CASE ( nbondi )      ! Read Dirichlet lateral conditions
1359      CASE ( -1, 0, 1 )                ! all exept 2 (i.e. close case)
1360         iihom = jpi-nreci-kexti
1361         DO jl = 1, ipreci
1362            r2dew(:,jl,1) = pt2d(nn_hls+jl,:)
1363            r2dwe(:,jl,1) = pt2d(iihom +jl,:)
1364         END DO
1365      END SELECT
1366      !
1367      !                           ! Migrations
1368      imigr = ipreci * ( jpj + 2*kextj )
1369      !
1370      IF( ln_timing ) CALL tic_tac(.TRUE.)
1371      !
1372      SELECT CASE ( nbondi )
1373      CASE ( -1 )
1374         CALL mppsend( 2, r2dwe(1-kextj,1,1), imigr, noea, ml_req1 )
1375         CALL mpprecv( 1, r2dew(1-kextj,1,2), imigr, noea )
1376         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1377      CASE ( 0 )
1378         CALL mppsend( 1, r2dew(1-kextj,1,1), imigr, nowe, ml_req1 )
1379         CALL mppsend( 2, r2dwe(1-kextj,1,1), imigr, noea, ml_req2 )
1380         CALL mpprecv( 1, r2dew(1-kextj,1,2), imigr, noea )
1381         CALL mpprecv( 2, r2dwe(1-kextj,1,2), imigr, nowe )
1382         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1383         IF(l_isend) CALL mpi_wait(ml_req2,ml_stat,ml_err)
1384      CASE ( 1 )
1385         CALL mppsend( 1, r2dew(1-kextj,1,1), imigr, nowe, ml_req1 )
1386         CALL mpprecv( 2, r2dwe(1-kextj,1,2), imigr, nowe )
1387         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1388      END SELECT
1389      !
1390      IF( ln_timing ) CALL tic_tac(.FALSE.)
1391      !
1392      !                           ! Write Dirichlet lateral conditions
1393      iihom = jpi - nn_hls
1394      !
1395      SELECT CASE ( nbondi )
1396      CASE ( -1 )
1397         DO jl = 1, ipreci
1398            pt2d(iihom+jl,:) = r2dew(:,jl,2)
1399         END DO
1400      CASE ( 0 )
1401         DO jl = 1, ipreci
1402            pt2d(jl-kexti,:) = r2dwe(:,jl,2)
1403            pt2d(iihom+jl,:) = r2dew(:,jl,2)
1404         END DO
1405      CASE ( 1 )
1406         DO jl = 1, ipreci
1407            pt2d(jl-kexti,:) = r2dwe(:,jl,2)
1408         END DO
1409      END SELECT
1410
1411
1412      ! 3. North and south directions
1413      ! -----------------------------
1414      ! always closed : we play only with the neigbours
1415      !
1416      IF( nbondj /= 2 ) THEN      ! Read Dirichlet lateral conditions
1417         ijhom = jpj-nrecj-kextj
1418         DO jl = 1, iprecj
1419            r2dsn(:,jl,1) = pt2d(:,ijhom +jl)
1420            r2dns(:,jl,1) = pt2d(:,nn_hls+jl)
1421         END DO
1422      ENDIF
1423      !
1424      !                           ! Migrations
1425      imigr = iprecj * ( jpi + 2*kexti )
1426      !
1427      IF( ln_timing ) CALL tic_tac(.TRUE.)
1428      !
1429      SELECT CASE ( nbondj )
1430      CASE ( -1 )
1431         CALL mppsend( 4, r2dsn(1-kexti,1,1), imigr, nono, ml_req1 )
1432         CALL mpprecv( 3, r2dns(1-kexti,1,2), imigr, nono )
1433         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1434      CASE ( 0 )
1435         CALL mppsend( 3, r2dns(1-kexti,1,1), imigr, noso, ml_req1 )
1436         CALL mppsend( 4, r2dsn(1-kexti,1,1), imigr, nono, ml_req2 )
1437         CALL mpprecv( 3, r2dns(1-kexti,1,2), imigr, nono )
1438         CALL mpprecv( 4, r2dsn(1-kexti,1,2), imigr, noso )
1439         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1440         IF(l_isend) CALL mpi_wait(ml_req2,ml_stat,ml_err)
1441      CASE ( 1 )
1442         CALL mppsend( 3, r2dns(1-kexti,1,1), imigr, noso, ml_req1 )
1443         CALL mpprecv( 4, r2dsn(1-kexti,1,2), imigr, noso )
1444         IF(l_isend) CALL mpi_wait(ml_req1,ml_stat,ml_err)
1445      END SELECT
1446      !
1447      IF( ln_timing ) CALL tic_tac(.FALSE.)
1448      !
1449      !                           ! Write Dirichlet lateral conditions
1450      ijhom = jpj - nn_hls
1451      !
1452      SELECT CASE ( nbondj )
1453      CASE ( -1 )
1454         DO jl = 1, iprecj
1455            pt2d(:,ijhom+jl) = r2dns(:,jl,2)
1456         END DO
1457      CASE ( 0 )
1458         DO jl = 1, iprecj
1459            pt2d(:,jl-kextj) = r2dsn(:,jl,2)
1460            pt2d(:,ijhom+jl) = r2dns(:,jl,2)
1461         END DO
1462      CASE ( 1 )
1463         DO jl = 1, iprecj
1464            pt2d(:,jl-kextj) = r2dsn(:,jl,2)
1465         END DO
1466      END SELECT
1467      !
1468   END SUBROUTINE mpp_lnk_2d_icb
1469
1470
1471   SUBROUTINE mpp_report( cdname, kpk, kpl, kpf, ld_lbc, ld_glb, ld_dlg )
1472      !!----------------------------------------------------------------------
1473      !!                  ***  routine mpp_report  ***
1474      !!
1475      !! ** Purpose :   report use of mpp routines per time-setp
1476      !!
1477      !!----------------------------------------------------------------------
1478      CHARACTER(len=*),           INTENT(in   ) ::   cdname      ! name of the calling subroutine
1479      INTEGER         , OPTIONAL, INTENT(in   ) ::   kpk, kpl, kpf
1480      LOGICAL         , OPTIONAL, INTENT(in   ) ::   ld_lbc, ld_glb, ld_dlg
1481      !!
1482      LOGICAL ::   ll_lbc, ll_glb, ll_dlg
1483      INTEGER ::    ji,  jj,  jk,  jh, jf   ! dummy loop indices
1484      !!----------------------------------------------------------------------
1485      !
1486      ll_lbc = .FALSE.
1487      IF( PRESENT(ld_lbc) ) ll_lbc = ld_lbc
1488      ll_glb = .FALSE.
1489      IF( PRESENT(ld_glb) ) ll_glb = ld_glb
1490      ll_dlg = .FALSE.
1491      IF( PRESENT(ld_dlg) ) ll_dlg = ld_dlg
1492      !
1493      ! find the smallest common frequency: default = frequency product, if multiple, choose the larger of the 2 frequency
1494      IF( ncom_dttrc /= 1 )   CALL ctl_stop( 'STOP', 'mpp_report, ncom_dttrc /= 1 not coded...' ) 
1495      ncom_freq = ncom_fsbc
1496      !
1497      IF ( ncom_stp == nit000+ncom_freq ) THEN   ! avoid to count extra communications in potential initializations at nit000
1498         IF( ll_lbc ) THEN
1499            IF( .NOT. ALLOCATED(ncomm_sequence) ) ALLOCATE( ncomm_sequence(ncom_rec_max,2) )
1500            IF( .NOT. ALLOCATED(    crname_lbc) ) ALLOCATE(     crname_lbc(ncom_rec_max  ) )
1501            n_sequence_lbc = n_sequence_lbc + 1
1502            IF( n_sequence_lbc > ncom_rec_max ) CALL ctl_stop( 'STOP', 'lib_mpp, increase ncom_rec_max' )   ! deadlock
1503            crname_lbc(n_sequence_lbc) = cdname     ! keep the name of the calling routine
1504            ncomm_sequence(n_sequence_lbc,1) = kpk*kpl   ! size of 3rd and 4th dimensions
1505            ncomm_sequence(n_sequence_lbc,2) = kpf       ! number of arrays to be treated (multi)
1506         ENDIF
1507         IF( ll_glb ) THEN
1508            IF( .NOT. ALLOCATED(crname_glb) ) ALLOCATE( crname_glb(ncom_rec_max) )
1509            n_sequence_glb = n_sequence_glb + 1
1510            IF( n_sequence_glb > ncom_rec_max ) CALL ctl_stop( 'STOP', 'lib_mpp, increase ncom_rec_max' )   ! deadlock
1511            crname_glb(n_sequence_glb) = cdname     ! keep the name of the calling routine
1512         ENDIF
1513         IF( ll_dlg ) THEN
1514            IF( .NOT. ALLOCATED(crname_dlg) ) ALLOCATE( crname_dlg(ncom_rec_max) )
1515            n_sequence_dlg = n_sequence_dlg + 1
1516            IF( n_sequence_dlg > ncom_rec_max ) CALL ctl_stop( 'STOP', 'lib_mpp, increase ncom_rec_max' )   ! deadlock
1517            crname_dlg(n_sequence_dlg) = cdname     ! keep the name of the calling routine
1518         ENDIF
1519      ELSE IF ( ncom_stp == nit000+2*ncom_freq ) THEN
1520         CALL ctl_opn( numcom, 'communication_report.txt', 'REPLACE', 'FORMATTED', 'SEQUENTIAL', -1, numout, .FALSE., narea )
1521         WRITE(numcom,*) ' '
1522         WRITE(numcom,*) ' ------------------------------------------------------------'
1523         WRITE(numcom,*) ' Communication pattern report (second oce+sbc+top time step):'
1524         WRITE(numcom,*) ' ------------------------------------------------------------'
1525         WRITE(numcom,*) ' '
1526         WRITE(numcom,'(A,I4)') ' Exchanged halos : ', n_sequence_lbc
1527         jj = 0; jk = 0; jf = 0; jh = 0
1528         DO ji = 1, n_sequence_lbc
1529            IF ( ncomm_sequence(ji,1) .GT. 1 ) jk = jk + 1
1530            IF ( ncomm_sequence(ji,2) .GT. 1 ) jf = jf + 1
1531            IF ( ncomm_sequence(ji,1) .GT. 1 .AND. ncomm_sequence(ji,2) .GT. 1 ) jj = jj + 1
1532            jh = MAX (jh, ncomm_sequence(ji,1)*ncomm_sequence(ji,2))
1533         END DO
1534         WRITE(numcom,'(A,I3)') ' 3D Exchanged halos : ', jk
1535         WRITE(numcom,'(A,I3)') ' Multi arrays exchanged halos : ', jf
1536         WRITE(numcom,'(A,I3)') '   from which 3D : ', jj
1537         WRITE(numcom,'(A,I10)') ' Array max size : ', jh*jpi*jpj
1538         WRITE(numcom,*) ' '
1539         WRITE(numcom,*) ' lbc_lnk called'
1540         jj = 1
1541         DO ji = 2, n_sequence_lbc
1542            IF( crname_lbc(ji-1) /= crname_lbc(ji) ) THEN
1543               WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_lbc(ji-1))
1544               jj = 0
1545            END IF
1546            jj = jj + 1 
1547         END DO
1548         WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_lbc(n_sequence_lbc))
1549         WRITE(numcom,*) ' '
1550         IF ( n_sequence_glb > 0 ) THEN
1551            WRITE(numcom,'(A,I4)') ' Global communications : ', n_sequence_glb
1552            jj = 1
1553            DO ji = 2, n_sequence_glb
1554               IF( crname_glb(ji-1) /= crname_glb(ji) ) THEN
1555                  WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_glb(ji-1))
1556                  jj = 0
1557               END IF
1558               jj = jj + 1 
1559            END DO
1560            WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_glb(n_sequence_glb))
1561            DEALLOCATE(crname_glb)
1562         ELSE
1563            WRITE(numcom,*) ' No MPI global communication '
1564         ENDIF
1565         WRITE(numcom,*) ' '
1566         IF ( n_sequence_dlg > 0 ) THEN
1567            WRITE(numcom,'(A,I4)') ' Delayed global communications : ', n_sequence_dlg
1568            jj = 1
1569            DO ji = 2, n_sequence_dlg
1570               IF( crname_dlg(ji-1) /= crname_dlg(ji) ) THEN
1571                  WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_dlg(ji-1))
1572                  jj = 0
1573               END IF
1574               jj = jj + 1 
1575            END DO
1576            WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_dlg(n_sequence_dlg))
1577            DEALLOCATE(crname_dlg)
1578         ELSE
1579            WRITE(numcom,*) ' No MPI delayed global communication '
1580         ENDIF
1581         WRITE(numcom,*) ' '
1582         WRITE(numcom,*) ' -----------------------------------------------'
1583         WRITE(numcom,*) ' '
1584         DEALLOCATE(ncomm_sequence)
1585         DEALLOCATE(crname_lbc)
1586      ENDIF
1587   END SUBROUTINE mpp_report
1588
1589   
1590   SUBROUTINE tic_tac (ld_tic, ld_global)
1591
1592    LOGICAL,           INTENT(IN) :: ld_tic
1593    LOGICAL, OPTIONAL, INTENT(IN) :: ld_global
1594    REAL(wp), DIMENSION(2), SAVE :: tic_wt
1595    REAL(wp),               SAVE :: tic_ct = 0._wp
1596    INTEGER :: ii
1597
1598    IF( ncom_stp <= nit000 ) RETURN
1599    IF( ncom_stp == nitend ) RETURN
1600    ii = 1
1601    IF( PRESENT( ld_global ) ) THEN
1602       IF( ld_global ) ii = 2
1603    END IF
1604   
1605    IF ( ld_tic ) THEN
1606       tic_wt(ii) = MPI_Wtime()                                                    ! start count tic->tac (waiting time)
1607       IF ( tic_ct > 0.0_wp ) compute_time = compute_time + MPI_Wtime() - tic_ct   ! cumulate count tac->tic
1608    ELSE
1609       waiting_time(ii) = waiting_time(ii) + MPI_Wtime() - tic_wt(ii)              ! cumulate count tic->tac
1610       tic_ct = MPI_Wtime()                                                        ! start count tac->tic (waiting time)
1611    ENDIF
1612   
1613   END SUBROUTINE tic_tac
1614
1615   
1616#else
1617   !!----------------------------------------------------------------------
1618   !!   Default case:            Dummy module        share memory computing
1619   !!----------------------------------------------------------------------
1620   USE in_out_manager
1621
1622   INTERFACE mpp_sum
1623      MODULE PROCEDURE mppsum_int, mppsum_a_int, mppsum_real, mppsum_a_real, mppsum_realdd, mppsum_a_realdd
1624   END INTERFACE
1625   INTERFACE mpp_max
1626      MODULE PROCEDURE mppmax_a_int, mppmax_int, mppmax_a_real, mppmax_real
1627   END INTERFACE
1628   INTERFACE mpp_min
1629      MODULE PROCEDURE mppmin_a_int, mppmin_int, mppmin_a_real, mppmin_real
1630   END INTERFACE
1631   INTERFACE mpp_minloc
1632      MODULE PROCEDURE mpp_minloc2d ,mpp_minloc3d
1633   END INTERFACE
1634   INTERFACE mpp_maxloc
1635      MODULE PROCEDURE mpp_maxloc2d ,mpp_maxloc3d
1636   END INTERFACE
1637
1638   LOGICAL, PUBLIC, PARAMETER ::   lk_mpp = .FALSE.      !: mpp flag
1639   LOGICAL, PUBLIC            ::   ln_nnogather          !: namelist control of northfold comms (needed here in case "key_mpp_mpi" is not used)
1640   INTEGER, PUBLIC            ::   mpi_comm_oce          ! opa local communicator
1641
1642   INTEGER, PARAMETER, PUBLIC               ::   nbdelay = 0   ! make sure we don't enter loops: DO ji = 1, nbdelay
1643   CHARACTER(len=32), DIMENSION(1), PUBLIC  ::   c_delaylist = 'empty'
1644   CHARACTER(len=32), DIMENSION(1), PUBLIC  ::   c_delaycpnt = 'empty'
1645   LOGICAL, PUBLIC                          ::   l_full_nf_update = .TRUE.
1646   TYPE ::   DELAYARR
1647      REAL(   wp), POINTER, DIMENSION(:) ::  z1d => NULL()
1648      COMPLEX(wp), POINTER, DIMENSION(:) ::  y1d => NULL()
1649   END TYPE DELAYARR
1650   TYPE( DELAYARR ), DIMENSION(1), PUBLIC  ::   todelay             
1651   INTEGER,  PUBLIC, DIMENSION(1)           ::   ndelayid = -1
1652   !!----------------------------------------------------------------------
1653CONTAINS
1654
1655   INTEGER FUNCTION lib_mpp_alloc(kumout)          ! Dummy function
1656      INTEGER, INTENT(in) ::   kumout
1657      lib_mpp_alloc = 0
1658   END FUNCTION lib_mpp_alloc
1659
1660   FUNCTION mynode( ldtxt, ldname, kumnam_ref, knumnam_cfg,  kumond , kstop, localComm ) RESULT (function_value)
1661      INTEGER, OPTIONAL            , INTENT(in   ) ::   localComm
1662      CHARACTER(len=*),DIMENSION(:) ::   ldtxt
1663      CHARACTER(len=*) ::   ldname
1664      INTEGER ::   kumnam_ref, knumnam_cfg , kumond , kstop
1665      IF( PRESENT( localComm ) ) mpi_comm_oce = localComm
1666      function_value = 0
1667      IF( .FALSE. )   ldtxt(:) = 'never done'
1668      CALL ctl_opn( kumond, TRIM(ldname), 'UNKNOWN', 'FORMATTED', 'SEQUENTIAL', -1, 6, .FALSE. , 1 )
1669   END FUNCTION mynode
1670
1671   SUBROUTINE mppsync                       ! Dummy routine
1672   END SUBROUTINE mppsync
1673
1674   !!----------------------------------------------------------------------
1675   !!    ***  mppmax_a_int, mppmax_int, mppmax_a_real, mppmax_real  ***
1676   !!   
1677   !!----------------------------------------------------------------------
1678   !!
1679#  define OPERATION_MAX
1680#  define INTEGER_TYPE
1681#  define DIM_0d
1682#     define ROUTINE_ALLREDUCE           mppmax_int
1683#     include "mpp_allreduce_generic.h90"
1684#     undef ROUTINE_ALLREDUCE
1685#  undef DIM_0d
1686#  define DIM_1d
1687#     define ROUTINE_ALLREDUCE           mppmax_a_int
1688#     include "mpp_allreduce_generic.h90"
1689#     undef ROUTINE_ALLREDUCE
1690#  undef DIM_1d
1691#  undef INTEGER_TYPE
1692!
1693#  define REAL_TYPE
1694#  define DIM_0d
1695#     define ROUTINE_ALLREDUCE           mppmax_real
1696#     include "mpp_allreduce_generic.h90"
1697#     undef ROUTINE_ALLREDUCE
1698#  undef DIM_0d
1699#  define DIM_1d
1700#     define ROUTINE_ALLREDUCE           mppmax_a_real
1701#     include "mpp_allreduce_generic.h90"
1702#     undef ROUTINE_ALLREDUCE
1703#  undef DIM_1d
1704#  undef REAL_TYPE
1705#  undef OPERATION_MAX
1706   !!----------------------------------------------------------------------
1707   !!    ***  mppmin_a_int, mppmin_int, mppmin_a_real, mppmin_real  ***
1708   !!   
1709   !!----------------------------------------------------------------------
1710   !!
1711#  define OPERATION_MIN
1712#  define INTEGER_TYPE
1713#  define DIM_0d
1714#     define ROUTINE_ALLREDUCE           mppmin_int
1715#     include "mpp_allreduce_generic.h90"
1716#     undef ROUTINE_ALLREDUCE
1717#  undef DIM_0d
1718#  define DIM_1d
1719#     define ROUTINE_ALLREDUCE           mppmin_a_int
1720#     include "mpp_allreduce_generic.h90"
1721#     undef ROUTINE_ALLREDUCE
1722#  undef DIM_1d
1723#  undef INTEGER_TYPE
1724!
1725#  define REAL_TYPE
1726#  define DIM_0d
1727#     define ROUTINE_ALLREDUCE           mppmin_real
1728#     include "mpp_allreduce_generic.h90"
1729#     undef ROUTINE_ALLREDUCE
1730#  undef DIM_0d
1731#  define DIM_1d
1732#     define ROUTINE_ALLREDUCE           mppmin_a_real
1733#     include "mpp_allreduce_generic.h90"
1734#     undef ROUTINE_ALLREDUCE
1735#  undef DIM_1d
1736#  undef REAL_TYPE
1737#  undef OPERATION_MIN
1738
1739   !!----------------------------------------------------------------------
1740   !!    ***  mppsum_a_int, mppsum_int, mppsum_a_real, mppsum_real  ***
1741   !!   
1742   !!   Global sum of 1D array or a variable (integer, real or complex)
1743   !!----------------------------------------------------------------------
1744   !!
1745#  define OPERATION_SUM
1746#  define INTEGER_TYPE
1747#  define DIM_0d
1748#     define ROUTINE_ALLREDUCE           mppsum_int
1749#     include "mpp_allreduce_generic.h90"
1750#     undef ROUTINE_ALLREDUCE
1751#  undef DIM_0d
1752#  define DIM_1d
1753#     define ROUTINE_ALLREDUCE           mppsum_a_int
1754#     include "mpp_allreduce_generic.h90"
1755#     undef ROUTINE_ALLREDUCE
1756#  undef DIM_1d
1757#  undef INTEGER_TYPE
1758!
1759#  define REAL_TYPE
1760#  define DIM_0d
1761#     define ROUTINE_ALLREDUCE           mppsum_real
1762#     include "mpp_allreduce_generic.h90"
1763#     undef ROUTINE_ALLREDUCE
1764#  undef DIM_0d
1765#  define DIM_1d
1766#     define ROUTINE_ALLREDUCE           mppsum_a_real
1767#     include "mpp_allreduce_generic.h90"
1768#     undef ROUTINE_ALLREDUCE
1769#  undef DIM_1d
1770#  undef REAL_TYPE
1771#  undef OPERATION_SUM
1772
1773#  define OPERATION_SUM_DD
1774#  define COMPLEX_TYPE
1775#  define DIM_0d
1776#     define ROUTINE_ALLREDUCE           mppsum_realdd
1777#     include "mpp_allreduce_generic.h90"
1778#     undef ROUTINE_ALLREDUCE
1779#  undef DIM_0d
1780#  define DIM_1d
1781#     define ROUTINE_ALLREDUCE           mppsum_a_realdd
1782#     include "mpp_allreduce_generic.h90"
1783#     undef ROUTINE_ALLREDUCE
1784#  undef DIM_1d
1785#  undef COMPLEX_TYPE
1786#  undef OPERATION_SUM_DD
1787
1788   !!----------------------------------------------------------------------
1789   !!    ***  mpp_minloc2d, mpp_minloc3d, mpp_maxloc2d, mpp_maxloc3d
1790   !!   
1791   !!----------------------------------------------------------------------
1792   !!
1793#  define OPERATION_MINLOC
1794#  define DIM_2d
1795#     define ROUTINE_LOC           mpp_minloc2d
1796#     include "mpp_loc_generic.h90"
1797#     undef ROUTINE_LOC
1798#  undef DIM_2d
1799#  define DIM_3d
1800#     define ROUTINE_LOC           mpp_minloc3d
1801#     include "mpp_loc_generic.h90"
1802#     undef ROUTINE_LOC
1803#  undef DIM_3d
1804#  undef OPERATION_MINLOC
1805
1806#  define OPERATION_MAXLOC
1807#  define DIM_2d
1808#     define ROUTINE_LOC           mpp_maxloc2d
1809#     include "mpp_loc_generic.h90"
1810#     undef ROUTINE_LOC
1811#  undef DIM_2d
1812#  define DIM_3d
1813#     define ROUTINE_LOC           mpp_maxloc3d
1814#     include "mpp_loc_generic.h90"
1815#     undef ROUTINE_LOC
1816#  undef DIM_3d
1817#  undef OPERATION_MAXLOC
1818
1819   SUBROUTINE mpp_delay_sum( cdname, cdelay, y_in, pout, ldlast, kcom )
1820      CHARACTER(len=*), INTENT(in   )               ::   cdname  ! name of the calling subroutine
1821      CHARACTER(len=*), INTENT(in   )               ::   cdelay  ! name (used as id) of the delayed operation
1822      COMPLEX(wp),      INTENT(in   ), DIMENSION(:) ::   y_in
1823      REAL(wp),         INTENT(  out), DIMENSION(:) ::   pout
1824      LOGICAL,          INTENT(in   )               ::   ldlast  ! true if this is the last time we call this routine
1825      INTEGER,          INTENT(in   ), OPTIONAL     ::   kcom
1826      !
1827      pout(:) = REAL(y_in(:), wp)
1828   END SUBROUTINE mpp_delay_sum
1829
1830   SUBROUTINE mpp_delay_max( cdname, cdelay, p_in, pout, ldlast, kcom )
1831      CHARACTER(len=*), INTENT(in   )               ::   cdname  ! name of the calling subroutine
1832      CHARACTER(len=*), INTENT(in   )               ::   cdelay  ! name (used as id) of the delayed operation
1833      REAL(wp),         INTENT(in   ), DIMENSION(:) ::   p_in
1834      REAL(wp),         INTENT(  out), DIMENSION(:) ::   pout
1835      LOGICAL,          INTENT(in   )               ::   ldlast  ! true if this is the last time we call this routine
1836      INTEGER,          INTENT(in   ), OPTIONAL     ::   kcom
1837      !
1838      pout(:) = p_in(:)
1839   END SUBROUTINE mpp_delay_max
1840
1841   SUBROUTINE mpp_delay_rcv( kid )
1842      INTEGER,INTENT(in   )      ::  kid 
1843      WRITE(*,*) 'mpp_delay_rcv: You should not have seen this print! error?', kid
1844   END SUBROUTINE mpp_delay_rcv
1845   
1846   SUBROUTINE mppstop( ldfinal, ld_force_abort )
1847      LOGICAL, OPTIONAL, INTENT(in) :: ldfinal    ! source process number
1848      LOGICAL, OPTIONAL, INTENT(in) :: ld_force_abort    ! source process number
1849      STOP      ! non MPP case, just stop the run
1850   END SUBROUTINE mppstop
1851
1852   SUBROUTINE mpp_ini_znl( knum )
1853      INTEGER :: knum
1854      WRITE(*,*) 'mpp_ini_znl: You should not have seen this print! error?', knum
1855   END SUBROUTINE mpp_ini_znl
1856
1857   SUBROUTINE mpp_comm_free( kcom )
1858      INTEGER :: kcom
1859      WRITE(*,*) 'mpp_comm_free: You should not have seen this print! error?', kcom
1860   END SUBROUTINE mpp_comm_free
1861   
1862#endif
1863
1864   !!----------------------------------------------------------------------
1865   !!   All cases:         ctl_stop, ctl_warn, get_unit, ctl_opn, ctl_nam   routines
1866   !!----------------------------------------------------------------------
1867
1868   SUBROUTINE ctl_stop( cd1, cd2, cd3, cd4, cd5 ,   &
1869      &                 cd6, cd7, cd8, cd9, cd10 )
1870      !!----------------------------------------------------------------------
1871      !!                  ***  ROUTINE  stop_opa  ***
1872      !!
1873      !! ** Purpose :   print in ocean.outpput file a error message and
1874      !!                increment the error number (nstop) by one.
1875      !!----------------------------------------------------------------------
1876      CHARACTER(len=*), INTENT(in), OPTIONAL ::  cd1, cd2, cd3, cd4, cd5
1877      CHARACTER(len=*), INTENT(in), OPTIONAL ::  cd6, cd7, cd8, cd9, cd10
1878      !!----------------------------------------------------------------------
1879      !
1880      nstop = nstop + 1
1881
1882      ! force to open ocean.output file
1883      IF( numout == 6 ) CALL ctl_opn( numout, 'ocean.output', 'APPEND', 'FORMATTED', 'SEQUENTIAL', -1, 6, .FALSE. )
1884       
1885      WRITE(numout,cform_err)
1886      IF( PRESENT(cd1 ) )   WRITE(numout,*) TRIM(cd1)
1887      IF( PRESENT(cd2 ) )   WRITE(numout,*) TRIM(cd2)
1888      IF( PRESENT(cd3 ) )   WRITE(numout,*) TRIM(cd3)
1889      IF( PRESENT(cd4 ) )   WRITE(numout,*) TRIM(cd4)
1890      IF( PRESENT(cd5 ) )   WRITE(numout,*) TRIM(cd5)
1891      IF( PRESENT(cd6 ) )   WRITE(numout,*) TRIM(cd6)
1892      IF( PRESENT(cd7 ) )   WRITE(numout,*) TRIM(cd7)
1893      IF( PRESENT(cd8 ) )   WRITE(numout,*) TRIM(cd8)
1894      IF( PRESENT(cd9 ) )   WRITE(numout,*) TRIM(cd9)
1895      IF( PRESENT(cd10) )   WRITE(numout,*) TRIM(cd10)
1896
1897                               CALL FLUSH(numout    )
1898      IF( numstp     /= -1 )   CALL FLUSH(numstp    )
1899      IF( numrun     /= -1 )   CALL FLUSH(numrun    )
1900      IF( numevo_ice /= -1 )   CALL FLUSH(numevo_ice)
1901      !
1902      IF( cd1 == 'STOP' ) THEN
1903         WRITE(numout,*)  'huge E-R-R-O-R : immediate stop'
1904         CALL mppstop(ld_force_abort = .true.)
1905      ENDIF
1906      !
1907   END SUBROUTINE ctl_stop
1908
1909
1910   SUBROUTINE ctl_warn( cd1, cd2, cd3, cd4, cd5,   &
1911      &                 cd6, cd7, cd8, cd9, cd10 )
1912      !!----------------------------------------------------------------------
1913      !!                  ***  ROUTINE  stop_warn  ***
1914      !!
1915      !! ** Purpose :   print in ocean.outpput file a error message and
1916      !!                increment the warning number (nwarn) by one.
1917      !!----------------------------------------------------------------------
1918      CHARACTER(len=*), INTENT(in), OPTIONAL ::  cd1, cd2, cd3, cd4, cd5
1919      CHARACTER(len=*), INTENT(in), OPTIONAL ::  cd6, cd7, cd8, cd9, cd10
1920      !!----------------------------------------------------------------------
1921      !
1922      nwarn = nwarn + 1
1923      IF(lwp) THEN
1924         WRITE(numout,cform_war)
1925         IF( PRESENT(cd1 ) ) WRITE(numout,*) TRIM(cd1)
1926         IF( PRESENT(cd2 ) ) WRITE(numout,*) TRIM(cd2)
1927         IF( PRESENT(cd3 ) ) WRITE(numout,*) TRIM(cd3)
1928         IF( PRESENT(cd4 ) ) WRITE(numout,*) TRIM(cd4)
1929         IF( PRESENT(cd5 ) ) WRITE(numout,*) TRIM(cd5)
1930         IF( PRESENT(cd6 ) ) WRITE(numout,*) TRIM(cd6)
1931         IF( PRESENT(cd7 ) ) WRITE(numout,*) TRIM(cd7)
1932         IF( PRESENT(cd8 ) ) WRITE(numout,*) TRIM(cd8)
1933         IF( PRESENT(cd9 ) ) WRITE(numout,*) TRIM(cd9)
1934         IF( PRESENT(cd10) ) WRITE(numout,*) TRIM(cd10)
1935      ENDIF
1936      CALL FLUSH(numout)
1937      !
1938   END SUBROUTINE ctl_warn
1939
1940
1941   SUBROUTINE ctl_opn( knum, cdfile, cdstat, cdform, cdacce, klengh, kout, ldwp, karea )
1942      !!----------------------------------------------------------------------
1943      !!                  ***  ROUTINE ctl_opn  ***
1944      !!
1945      !! ** Purpose :   Open file and check if required file is available.
1946      !!
1947      !! ** Method  :   Fortan open
1948      !!----------------------------------------------------------------------
1949      INTEGER          , INTENT(  out) ::   knum      ! logical unit to open
1950      CHARACTER(len=*) , INTENT(in   ) ::   cdfile    ! file name to open
1951      CHARACTER(len=*) , INTENT(in   ) ::   cdstat    ! disposition specifier
1952      CHARACTER(len=*) , INTENT(in   ) ::   cdform    ! formatting specifier
1953      CHARACTER(len=*) , INTENT(in   ) ::   cdacce    ! access specifier
1954      INTEGER          , INTENT(in   ) ::   klengh    ! record length
1955      INTEGER          , INTENT(in   ) ::   kout      ! number of logical units for write
1956      LOGICAL          , INTENT(in   ) ::   ldwp      ! boolean term for print
1957      INTEGER, OPTIONAL, INTENT(in   ) ::   karea     ! proc number
1958      !
1959      CHARACTER(len=80) ::   clfile
1960      INTEGER           ::   iost
1961      !!----------------------------------------------------------------------
1962      !
1963      ! adapt filename
1964      ! ----------------
1965      clfile = TRIM(cdfile)
1966      IF( PRESENT( karea ) ) THEN
1967         IF( karea > 1 )   WRITE(clfile, "(a,'_',i4.4)") TRIM(clfile), karea-1
1968      ENDIF
1969#if defined key_agrif
1970      IF( .NOT. Agrif_Root() )   clfile = TRIM(Agrif_CFixed())//'_'//TRIM(clfile)
1971      knum=Agrif_Get_Unit()
1972#else
1973      knum=get_unit()
1974#endif
1975      IF( TRIM(cdfile) == '/dev/null' )   clfile = TRIM(cdfile)   ! force the use of /dev/null
1976      !
1977      iost=0
1978      IF( cdacce(1:6) == 'DIRECT' )  THEN         ! cdacce has always more than 6 characters
1979         OPEN( UNIT=knum, FILE=clfile, FORM=cdform, ACCESS=cdacce, STATUS=cdstat, RECL=klengh         , ERR=100, IOSTAT=iost )
1980      ELSE IF( TRIM(cdstat) == 'APPEND' )  THEN   ! cdstat can have less than 6 characters
1981         OPEN( UNIT=knum, FILE=clfile, FORM=cdform, ACCESS=cdacce, STATUS='UNKNOWN', POSITION='APPEND', ERR=100, IOSTAT=iost )
1982      ELSE
1983         OPEN( UNIT=knum, FILE=clfile, FORM=cdform, ACCESS=cdacce, STATUS=cdstat                      , ERR=100, IOSTAT=iost )
1984      ENDIF
1985      IF( iost /= 0 .AND. TRIM(clfile) == '/dev/null' ) &   ! for windows
1986         &  OPEN(UNIT=knum,FILE='NUL', FORM=cdform, ACCESS=cdacce, STATUS=cdstat                      , ERR=100, IOSTAT=iost )   
1987      IF( iost == 0 ) THEN
1988         IF(ldwp) THEN
1989            WRITE(kout,*) '     file   : ', TRIM(clfile),' open ok'
1990            WRITE(kout,*) '     unit   = ', knum
1991            WRITE(kout,*) '     status = ', cdstat
1992            WRITE(kout,*) '     form   = ', cdform
1993            WRITE(kout,*) '     access = ', cdacce
1994            WRITE(kout,*)
1995         ENDIF
1996      ENDIF
1997100   CONTINUE
1998      IF( iost /= 0 ) THEN
1999         IF(ldwp) THEN
2000            WRITE(kout,*)
2001            WRITE(kout,*) ' ===>>>> : bad opening file: ', TRIM(clfile)
2002            WRITE(kout,*) ' =======   ===  '
2003            WRITE(kout,*) '           unit   = ', knum
2004            WRITE(kout,*) '           status = ', cdstat
2005            WRITE(kout,*) '           form   = ', cdform
2006            WRITE(kout,*) '           access = ', cdacce
2007            WRITE(kout,*) '           iostat = ', iost
2008            WRITE(kout,*) '           we stop. verify the file '
2009            WRITE(kout,*)
2010         ELSE  !!! Force writing to make sure we get the information - at least once - in this violent STOP!!
2011            WRITE(*,*)
2012            WRITE(*,*) ' ===>>>> : bad opening file: ', TRIM(clfile)
2013            WRITE(*,*) ' =======   ===  '
2014            WRITE(*,*) '           unit   = ', knum
2015            WRITE(*,*) '           status = ', cdstat
2016            WRITE(*,*) '           form   = ', cdform
2017            WRITE(*,*) '           access = ', cdacce
2018            WRITE(*,*) '           iostat = ', iost
2019            WRITE(*,*) '           we stop. verify the file '
2020            WRITE(*,*)
2021         ENDIF
2022         CALL FLUSH( kout ) 
2023         STOP 'ctl_opn bad opening'
2024      ENDIF
2025      !
2026   END SUBROUTINE ctl_opn
2027
2028
2029   SUBROUTINE ctl_nam ( kios, cdnam, ldwp )
2030      !!----------------------------------------------------------------------
2031      !!                  ***  ROUTINE ctl_nam  ***
2032      !!
2033      !! ** Purpose :   Informations when error while reading a namelist
2034      !!
2035      !! ** Method  :   Fortan open
2036      !!----------------------------------------------------------------------
2037      INTEGER         , INTENT(inout) ::   kios    ! IO status after reading the namelist
2038      CHARACTER(len=*), INTENT(in   ) ::   cdnam   ! group name of namelist for which error occurs
2039      CHARACTER(len=5)                ::   clios   ! string to convert iostat in character for print
2040      LOGICAL         , INTENT(in   ) ::   ldwp    ! boolean term for print
2041      !!----------------------------------------------------------------------
2042      !
2043      WRITE (clios, '(I5.0)')   kios
2044      IF( kios < 0 ) THEN         
2045         CALL ctl_warn( 'end of record or file while reading namelist '   &
2046            &           // TRIM(cdnam) // ' iostat = ' // TRIM(clios) )
2047      ENDIF
2048      !
2049      IF( kios > 0 ) THEN
2050         CALL ctl_stop( 'misspelled variable in namelist '   &
2051            &           // TRIM(cdnam) // ' iostat = ' // TRIM(clios) )
2052      ENDIF
2053      kios = 0
2054      RETURN
2055      !
2056   END SUBROUTINE ctl_nam
2057
2058
2059   INTEGER FUNCTION get_unit()
2060      !!----------------------------------------------------------------------
2061      !!                  ***  FUNCTION  get_unit  ***
2062      !!
2063      !! ** Purpose :   return the index of an unused logical unit
2064      !!----------------------------------------------------------------------
2065      LOGICAL :: llopn
2066      !!----------------------------------------------------------------------
2067      !
2068      get_unit = 15   ! choose a unit that is big enough then it is not already used in NEMO
2069      llopn = .TRUE.
2070      DO WHILE( (get_unit < 998) .AND. llopn )
2071         get_unit = get_unit + 1
2072         INQUIRE( unit = get_unit, opened = llopn )
2073      END DO
2074      IF( (get_unit == 999) .AND. llopn ) THEN
2075         CALL ctl_stop( 'get_unit: All logical units until 999 are used...' )
2076         get_unit = -1
2077      ENDIF
2078      !
2079   END FUNCTION get_unit
2080
2081   !!----------------------------------------------------------------------
2082END MODULE lib_mpp
Note: See TracBrowser for help on using the repository browser.