source: NEMO/branches/2019/dev_r11613_ENHANCE-04_namelists_as_internalfiles/src/OCE/LBC/lib_mpp.F90 @ 11624

Last change on this file since 11624 was 11624, checked in by acc, 2 years ago

Branch 2019/dev_r11613_ENHANCE-04_namelists_as_internalfiles. Substantive changes required to replace all namelists with internal files. These are the key changes only; to compile and run tests all REWIND and CLOSE operations on the (no longer) units have to be removed. These changes affect many more files but can be scripted so are not included here in order to make a later merge easier. The scripts used to prepare code for testing are included on: wiki:2019WP/ENHANCE-04_AndrewC-reporting/Internal_Namelists. With these additional changes this code passes most SETTE tests but the AGRIF preprocessor does not currently accept the new allocatable character strings. To be investigated.

  • Property svn:keywords set to Id
File size: 58.6 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   !!   load_nml      : Read, condense and buffer namelist file into character array for use as an internal file
35   !!----------------------------------------------------------------------
36   !!----------------------------------------------------------------------
37   !!   mpp_start     : get local communicator its size and rank
38   !!   mpp_lnk       : interface (defined in lbclnk) for message passing of 2d or 3d arrays (mpp_lnk_2d, mpp_lnk_3d)
39   !!   mpp_lnk_icb   : interface for message passing of 2d arrays with extra halo for icebergs (mpp_lnk_2d_icb)
40   !!   mpprecv       :
41   !!   mppsend       :
42   !!   mppscatter    :
43   !!   mppgather     :
44   !!   mpp_min       : generic interface for mppmin_int , mppmin_a_int , mppmin_real, mppmin_a_real
45   !!   mpp_max       : generic interface for mppmax_int , mppmax_a_int , mppmax_real, mppmax_a_real
46   !!   mpp_sum       : generic interface for mppsum_int , mppsum_a_int , mppsum_real, mppsum_a_real
47   !!   mpp_minloc    :
48   !!   mpp_maxloc    :
49   !!   mppsync       :
50   !!   mppstop       :
51   !!   mpp_ini_north : initialisation of north fold
52   !!   mpp_lbc_north_icb : alternative to mpp_nfd for extra outer halo with icebergs
53   !!----------------------------------------------------------------------
54   USE dom_oce        ! ocean space and time domain
55   USE in_out_manager ! I/O manager
56
57   IMPLICIT NONE
58   PRIVATE
59   !
60   PUBLIC   ctl_stop, ctl_warn, ctl_opn, ctl_nam, load_nml
61   PUBLIC   mpp_start, mppstop, mppsync, mpp_comm_free
62   PUBLIC   mpp_ini_north
63   PUBLIC   mpp_min, mpp_max, mpp_sum, mpp_minloc, mpp_maxloc
64   PUBLIC   mpp_delay_max, mpp_delay_sum, mpp_delay_rcv
65   PUBLIC   mppscatter, mppgather
66   PUBLIC   mpp_ini_znl
67   PUBLIC   mppsend, mpprecv                          ! needed by TAM and ICB routines
68   PUBLIC   mpp_report
69   PUBLIC   tic_tac
70#if ! defined key_mpp_mpi
71   PUBLIC MPI_Wtime
72#endif
73   
74   !! * Interfaces
75   !! define generic interface for these routine as they are called sometimes
76   !! with scalar arguments instead of array arguments, which causes problems
77   !! for the compilation on AIX system as well as NEC and SGI. Ok on COMPACQ
78   INTERFACE mpp_min
79      MODULE PROCEDURE mppmin_a_int, mppmin_int, mppmin_a_real, mppmin_real
80   END INTERFACE
81   INTERFACE mpp_max
82      MODULE PROCEDURE mppmax_a_int, mppmax_int, mppmax_a_real, mppmax_real
83   END INTERFACE
84   INTERFACE mpp_sum
85      MODULE PROCEDURE mppsum_a_int, mppsum_int, mppsum_a_real, mppsum_real,   &
86         &             mppsum_realdd, mppsum_a_realdd
87   END INTERFACE
88   INTERFACE mpp_minloc
89      MODULE PROCEDURE mpp_minloc2d ,mpp_minloc3d
90   END INTERFACE
91   INTERFACE mpp_maxloc
92      MODULE PROCEDURE mpp_maxloc2d ,mpp_maxloc3d
93   END INTERFACE
94
95   !! ========================= !!
96   !!  MPI  variable definition !!
97   !! ========================= !!
98#if   defined key_mpp_mpi
99!$AGRIF_DO_NOT_TREAT
100   INCLUDE 'mpif.h'
101!$AGRIF_END_DO_NOT_TREAT
102   LOGICAL, PUBLIC, PARAMETER ::   lk_mpp = .TRUE.    !: mpp flag
103#else   
104   INTEGER, PUBLIC, PARAMETER ::   MPI_STATUS_SIZE = 1
105   INTEGER, PUBLIC, PARAMETER ::   MPI_DOUBLE_PRECISION = 8
106   LOGICAL, PUBLIC, PARAMETER ::   lk_mpp = .FALSE.    !: mpp flag
107#endif
108
109   INTEGER, PARAMETER         ::   nprocmax = 2**10   ! maximun dimension (required to be a power of 2)
110
111   INTEGER, PUBLIC ::   mppsize        ! number of process
112   INTEGER, PUBLIC ::   mpprank        ! process number  [ 0 - size-1 ]
113!$AGRIF_DO_NOT_TREAT
114   INTEGER, PUBLIC ::   mpi_comm_oce   ! opa local communicator
115!$AGRIF_END_DO_NOT_TREAT
116
117   INTEGER :: MPI_SUMDD
118
119   ! variables used for zonal integration
120   INTEGER, PUBLIC ::   ncomm_znl       !: communicator made by the processors on the same zonal average
121   LOGICAL, PUBLIC ::   l_znl_root      !: True on the 'left'most processor on the same row
122   INTEGER         ::   ngrp_znl        !  group ID for the znl processors
123   INTEGER         ::   ndim_rank_znl   !  number of processors on the same zonal average
124   INTEGER, DIMENSION(:), ALLOCATABLE, SAVE ::   nrank_znl  ! dimension ndim_rank_znl, number of the procs into the same znl domain
125
126   ! North fold condition in mpp_mpi with jpni > 1 (PUBLIC for TAM)
127   INTEGER, PUBLIC ::   ngrp_world        !: group ID for the world processors
128   INTEGER, PUBLIC ::   ngrp_opa          !: group ID for the opa processors
129   INTEGER, PUBLIC ::   ngrp_north        !: group ID for the northern processors (to be fold)
130   INTEGER, PUBLIC ::   ncomm_north       !: communicator made by the processors belonging to ngrp_north
131   INTEGER, PUBLIC ::   ndim_rank_north   !: number of 'sea' processor in the northern line (can be /= jpni !)
132   INTEGER, PUBLIC ::   njmppmax          !: value of njmpp for the processors of the northern line
133   INTEGER, PUBLIC ::   north_root        !: number (in the comm_opa) of proc 0 in the northern comm
134   INTEGER, PUBLIC, DIMENSION(:), ALLOCATABLE, SAVE ::   nrank_north   !: dimension ndim_rank_north
135
136   ! Communications summary report
137   CHARACTER(len=128), DIMENSION(:), ALLOCATABLE ::   crname_lbc                   !: names of lbc_lnk calling routines
138   CHARACTER(len=128), DIMENSION(:), ALLOCATABLE ::   crname_glb                   !: names of global comm calling routines
139   CHARACTER(len=128), DIMENSION(:), ALLOCATABLE ::   crname_dlg                   !: names of delayed global comm calling routines
140   INTEGER, PUBLIC                               ::   ncom_stp = 0                 !: copy of time step # istp
141   INTEGER, PUBLIC                               ::   ncom_fsbc = 1                !: copy of sbc time step # nn_fsbc
142   INTEGER, PUBLIC                               ::   ncom_dttrc = 1               !: copy of top time step # nn_dttrc
143   INTEGER, PUBLIC                               ::   ncom_freq                    !: frequency of comm diagnostic
144   INTEGER, PUBLIC , DIMENSION(:,:), ALLOCATABLE ::   ncomm_sequence               !: size of communicated arrays (halos)
145   INTEGER, PARAMETER, PUBLIC                    ::   ncom_rec_max = 5000          !: max number of communication record
146   INTEGER, PUBLIC                               ::   n_sequence_lbc = 0           !: # of communicated arraysvia lbc
147   INTEGER, PUBLIC                               ::   n_sequence_glb = 0           !: # of global communications
148   INTEGER, PUBLIC                               ::   n_sequence_dlg = 0           !: # of delayed global communications
149   INTEGER, PUBLIC                               ::   numcom = -1                  !: logical unit for communicaton report
150   LOGICAL, PUBLIC                               ::   l_full_nf_update = .TRUE.    !: logical for a full (2lines) update of bc at North fold report
151   INTEGER,                    PARAMETER, PUBLIC ::   nbdelay = 2       !: number of delayed operations
152   !: name (used as id) of allreduce-delayed operations
153   ! Warning: we must use the same character length in an array constructor (at least for gcc compiler)
154   CHARACTER(len=32), DIMENSION(nbdelay), PUBLIC ::   c_delaylist = (/ 'cflice', 'fwb   ' /)
155   !: component name where the allreduce-delayed operation is performed
156   CHARACTER(len=3),  DIMENSION(nbdelay), PUBLIC ::   c_delaycpnt = (/ 'ICE'   , 'OCE' /)
157   TYPE, PUBLIC ::   DELAYARR
158      REAL(   wp), POINTER, DIMENSION(:) ::  z1d => NULL()
159      COMPLEX(wp), POINTER, DIMENSION(:) ::  y1d => NULL()
160   END TYPE DELAYARR
161   TYPE( DELAYARR ), DIMENSION(nbdelay), PUBLIC, SAVE  ::   todelay         !: must have SAVE for default initialization of DELAYARR
162   INTEGER,          DIMENSION(nbdelay), PUBLIC        ::   ndelayid = -1   !: mpi request id of the delayed operations
163
164   ! timing summary report
165   REAL(wp), DIMENSION(2), PUBLIC ::  waiting_time = 0._wp
166   REAL(wp)              , PUBLIC ::  compute_time = 0._wp, elapsed_time = 0._wp
167   
168   REAL(wp), DIMENSION(:), ALLOCATABLE, SAVE ::   tampon   ! buffer in case of bsend
169
170   LOGICAL, PUBLIC ::   ln_nnogather                !: namelist control of northfold comms
171   LOGICAL, PUBLIC ::   l_north_nogather = .FALSE.  !: internal control of northfold comms
172   
173   !!----------------------------------------------------------------------
174   !! NEMO/OCE 4.0 , NEMO Consortium (2018)
175   !! $Id$
176   !! Software governed by the CeCILL license (see ./LICENSE)
177   !!----------------------------------------------------------------------
178CONTAINS
179
180   SUBROUTINE mpp_start( localComm )
181      !!----------------------------------------------------------------------
182      !!                  ***  routine mpp_start  ***
183      !!
184      !! ** Purpose :   get mpi_comm_oce, mpprank and mppsize
185      !!----------------------------------------------------------------------
186      INTEGER         , OPTIONAL   , INTENT(in   ) ::   localComm    !
187      !
188      INTEGER ::   ierr
189      LOGICAL ::   llmpi_init
190      !!----------------------------------------------------------------------
191#if defined key_mpp_mpi
192      !
193      CALL mpi_initialized ( llmpi_init, ierr )
194      IF( ierr /= MPI_SUCCESS ) CALL ctl_stop( 'STOP', ' lib_mpp: Error in routine mpi_initialized' )
195
196      IF( .NOT. llmpi_init ) THEN
197         IF( PRESENT(localComm) ) THEN
198            WRITE(ctmp1,*) ' lib_mpp: You cannot provide a local communicator '
199            WRITE(ctmp2,*) '          without calling MPI_Init before ! '
200            CALL ctl_stop( 'STOP', ctmp1, ctmp2 )
201         ENDIF
202         CALL mpi_init( ierr )
203         IF( ierr /= MPI_SUCCESS ) CALL ctl_stop( 'STOP', ' lib_mpp: Error in routine mpi_init' )
204      ENDIF
205       
206      IF( PRESENT(localComm) ) THEN
207         IF( Agrif_Root() ) THEN
208            mpi_comm_oce = localComm
209         ENDIF
210      ELSE
211         CALL mpi_comm_dup( mpi_comm_world, mpi_comm_oce, ierr)
212         IF( ierr /= MPI_SUCCESS ) CALL ctl_stop( 'STOP', ' lib_mpp: Error in routine mpi_comm_dup' )
213      ENDIF
214
215# if defined key_agrif
216      IF( Agrif_Root() ) THEN
217         CALL Agrif_MPI_Init(mpi_comm_oce)
218      ELSE
219         CALL Agrif_MPI_set_grid_comm(mpi_comm_oce)
220      ENDIF
221# endif
222
223      CALL mpi_comm_rank( mpi_comm_oce, mpprank, ierr )
224      CALL mpi_comm_size( mpi_comm_oce, mppsize, ierr )
225      !
226      CALL MPI_OP_CREATE(DDPDD_MPI, .TRUE., MPI_SUMDD, ierr)
227      !
228#else
229      IF( PRESENT( localComm ) ) mpi_comm_oce = localComm
230      mppsize = 1
231      mpprank = 0
232#endif
233   END SUBROUTINE mpp_start
234
235
236   SUBROUTINE mppsend( ktyp, pmess, kbytes, kdest, md_req )
237      !!----------------------------------------------------------------------
238      !!                  ***  routine mppsend  ***
239      !!
240      !! ** Purpose :   Send messag passing array
241      !!
242      !!----------------------------------------------------------------------
243      REAL(wp), INTENT(inout) ::   pmess(*)   ! array of real
244      INTEGER , INTENT(in   ) ::   kbytes     ! size of the array pmess
245      INTEGER , INTENT(in   ) ::   kdest      ! receive process number
246      INTEGER , INTENT(in   ) ::   ktyp       ! tag of the message
247      INTEGER , INTENT(in   ) ::   md_req     ! argument for isend
248      !!
249      INTEGER ::   iflag
250      !!----------------------------------------------------------------------
251      !
252#if defined key_mpp_mpi
253      CALL mpi_isend( pmess, kbytes, mpi_double_precision, kdest , ktyp, mpi_comm_oce, md_req, iflag )
254#endif
255      !
256   END SUBROUTINE mppsend
257
258
259   SUBROUTINE mpprecv( ktyp, pmess, kbytes, ksource )
260      !!----------------------------------------------------------------------
261      !!                  ***  routine mpprecv  ***
262      !!
263      !! ** Purpose :   Receive messag passing array
264      !!
265      !!----------------------------------------------------------------------
266      REAL(wp), INTENT(inout) ::   pmess(*)   ! array of real
267      INTEGER , INTENT(in   ) ::   kbytes     ! suze of the array pmess
268      INTEGER , INTENT(in   ) ::   ktyp       ! Tag of the recevied message
269      INTEGER, OPTIONAL, INTENT(in) :: ksource    ! source process number
270      !!
271      INTEGER :: istatus(mpi_status_size)
272      INTEGER :: iflag
273      INTEGER :: use_source
274      !!----------------------------------------------------------------------
275      !
276#if defined key_mpp_mpi
277      ! If a specific process number has been passed to the receive call,
278      ! use that one. Default is to use mpi_any_source
279      use_source = mpi_any_source
280      IF( PRESENT(ksource) )   use_source = ksource
281      !
282      CALL mpi_recv( pmess, kbytes, mpi_double_precision, use_source, ktyp, mpi_comm_oce, istatus, iflag )
283#endif
284      !
285   END SUBROUTINE mpprecv
286
287
288   SUBROUTINE mppgather( ptab, kp, pio )
289      !!----------------------------------------------------------------------
290      !!                   ***  routine mppgather  ***
291      !!
292      !! ** Purpose :   Transfert between a local subdomain array and a work
293      !!     array which is distributed following the vertical level.
294      !!
295      !!----------------------------------------------------------------------
296      REAL(wp), DIMENSION(jpi,jpj)      , INTENT(in   ) ::   ptab   ! subdomain input array
297      INTEGER                           , INTENT(in   ) ::   kp     ! record length
298      REAL(wp), DIMENSION(jpi,jpj,jpnij), INTENT(  out) ::   pio    ! subdomain input array
299      !!
300      INTEGER :: itaille, ierror   ! temporary integer
301      !!---------------------------------------------------------------------
302      !
303      itaille = jpi * jpj
304#if defined key_mpp_mpi
305      CALL mpi_gather( ptab, itaille, mpi_double_precision, pio, itaille     ,   &
306         &                            mpi_double_precision, kp , mpi_comm_oce, ierror )
307#else
308      pio(:,:,1) = ptab(:,:)
309#endif
310      !
311   END SUBROUTINE mppgather
312
313
314   SUBROUTINE mppscatter( pio, kp, ptab )
315      !!----------------------------------------------------------------------
316      !!                  ***  routine mppscatter  ***
317      !!
318      !! ** Purpose :   Transfert between awork array which is distributed
319      !!      following the vertical level and the local subdomain array.
320      !!
321      !!----------------------------------------------------------------------
322      REAL(wp), DIMENSION(jpi,jpj,jpnij)  ::   pio    ! output array
323      INTEGER                             ::   kp     ! Tag (not used with MPI
324      REAL(wp), DIMENSION(jpi,jpj)        ::   ptab   ! subdomain array input
325      !!
326      INTEGER :: itaille, ierror   ! temporary integer
327      !!---------------------------------------------------------------------
328      !
329      itaille = jpi * jpj
330      !
331#if defined key_mpp_mpi
332      CALL mpi_scatter( pio, itaille, mpi_double_precision, ptab, itaille     ,   &
333         &                            mpi_double_precision, kp  , mpi_comm_oce, ierror )
334#else
335      ptab(:,:) = pio(:,:,1)
336#endif
337      !
338   END SUBROUTINE mppscatter
339
340   
341   SUBROUTINE mpp_delay_sum( cdname, cdelay, y_in, pout, ldlast, kcom )
342     !!----------------------------------------------------------------------
343      !!                   ***  routine mpp_delay_sum  ***
344      !!
345      !! ** Purpose :   performed delayed mpp_sum, the result is received on next call
346      !!
347      !!----------------------------------------------------------------------
348      CHARACTER(len=*), INTENT(in   )               ::   cdname  ! name of the calling subroutine
349      CHARACTER(len=*), INTENT(in   )               ::   cdelay  ! name (used as id) of the delayed operation
350      COMPLEX(wp),      INTENT(in   ), DIMENSION(:) ::   y_in
351      REAL(wp),         INTENT(  out), DIMENSION(:) ::   pout
352      LOGICAL,          INTENT(in   )               ::   ldlast  ! true if this is the last time we call this routine
353      INTEGER,          INTENT(in   ), OPTIONAL     ::   kcom
354      !!
355      INTEGER ::   ji, isz
356      INTEGER ::   idvar
357      INTEGER ::   ierr, ilocalcomm
358      COMPLEX(wp), ALLOCATABLE, DIMENSION(:) ::   ytmp
359      !!----------------------------------------------------------------------
360#if defined key_mpp_mpi
361      ilocalcomm = mpi_comm_oce
362      IF( PRESENT(kcom) )   ilocalcomm = kcom
363
364      isz = SIZE(y_in)
365     
366      IF( narea == 1 .AND. numcom == -1 ) CALL mpp_report( cdname, ld_dlg = .TRUE. )
367
368      idvar = -1
369      DO ji = 1, nbdelay
370         IF( TRIM(cdelay) == TRIM(c_delaylist(ji)) ) idvar = ji
371      END DO
372      IF ( idvar == -1 )   CALL ctl_stop( 'STOP',' mpp_delay_sum : please add a new delayed exchange for '//TRIM(cdname) )
373
374      IF ( ndelayid(idvar) == 0 ) THEN         ! first call    with restart: %z1d defined in iom_delay_rst
375         !                                       --------------------------
376         IF ( SIZE(todelay(idvar)%z1d) /= isz ) THEN                  ! Check dimension coherence
377            IF(lwp) WRITE(numout,*) ' WARNING: the nb of delayed variables in restart file is not the model one'
378            DEALLOCATE(todelay(idvar)%z1d)
379            ndelayid(idvar) = -1                                      ! do as if we had no restart
380         ELSE
381            ALLOCATE(todelay(idvar)%y1d(isz))
382            todelay(idvar)%y1d(:) = CMPLX(todelay(idvar)%z1d(:), 0., wp)   ! create %y1d, complex variable needed by mpi_sumdd
383         END IF
384      ENDIF
385     
386      IF( ndelayid(idvar) == -1 ) THEN         ! first call without restart: define %y1d and %z1d from y_in with blocking allreduce
387         !                                       --------------------------
388         ALLOCATE(todelay(idvar)%z1d(isz), todelay(idvar)%y1d(isz))   ! allocate also %z1d as used for the restart
389         CALL mpi_allreduce( y_in(:), todelay(idvar)%y1d(:), isz, MPI_DOUBLE_COMPLEX, mpi_sumdd, ilocalcomm, ierr )   ! get %y1d
390         todelay(idvar)%z1d(:) = REAL(todelay(idvar)%y1d(:), wp)      ! define %z1d from %y1d
391      ENDIF
392
393      IF( ndelayid(idvar) > 0 )   CALL mpp_delay_rcv( idvar )         ! make sure %z1d is received
394
395      ! send back pout from todelay(idvar)%z1d defined at previous call
396      pout(:) = todelay(idvar)%z1d(:)
397
398      ! send y_in into todelay(idvar)%y1d with a non-blocking communication
399# if defined key_mpi2
400      IF( ln_timing ) CALL tic_tac( .TRUE., ld_global = .TRUE.)
401      CALL  mpi_allreduce( y_in(:), todelay(idvar)%y1d(:), isz, MPI_DOUBLE_COMPLEX, mpi_sumdd, ilocalcomm, ndelayid(idvar), ierr )
402      IF( ln_timing ) CALL tic_tac(.FALSE., ld_global = .TRUE.)
403# else
404      CALL mpi_iallreduce( y_in(:), todelay(idvar)%y1d(:), isz, MPI_DOUBLE_COMPLEX, mpi_sumdd, ilocalcomm, ndelayid(idvar), ierr )
405# endif
406#else
407      pout(:) = REAL(y_in(:), wp)
408#endif
409
410   END SUBROUTINE mpp_delay_sum
411
412   
413   SUBROUTINE mpp_delay_max( cdname, cdelay, p_in, pout, ldlast, kcom )
414      !!----------------------------------------------------------------------
415      !!                   ***  routine mpp_delay_max  ***
416      !!
417      !! ** Purpose :   performed delayed mpp_max, the result is received on next call
418      !!
419      !!----------------------------------------------------------------------
420      CHARACTER(len=*), INTENT(in   )                 ::   cdname  ! name of the calling subroutine
421      CHARACTER(len=*), INTENT(in   )                 ::   cdelay  ! name (used as id) of the delayed operation
422      REAL(wp),         INTENT(in   ), DIMENSION(:)   ::   p_in    !
423      REAL(wp),         INTENT(  out), DIMENSION(:)   ::   pout    !
424      LOGICAL,          INTENT(in   )                 ::   ldlast  ! true if this is the last time we call this routine
425      INTEGER,          INTENT(in   ), OPTIONAL       ::   kcom
426      !!
427      INTEGER ::   ji, isz
428      INTEGER ::   idvar
429      INTEGER ::   ierr, ilocalcomm
430      !!----------------------------------------------------------------------
431#if defined key_mpp_mpi
432      ilocalcomm = mpi_comm_oce
433      IF( PRESENT(kcom) )   ilocalcomm = kcom
434
435      isz = SIZE(p_in)
436
437      IF( narea == 1 .AND. numcom == -1 ) CALL mpp_report( cdname, ld_dlg = .TRUE. )
438
439      idvar = -1
440      DO ji = 1, nbdelay
441         IF( TRIM(cdelay) == TRIM(c_delaylist(ji)) ) idvar = ji
442      END DO
443      IF ( idvar == -1 )   CALL ctl_stop( 'STOP',' mpp_delay_max : please add a new delayed exchange for '//TRIM(cdname) )
444
445      IF ( ndelayid(idvar) == 0 ) THEN         ! first call    with restart: %z1d defined in iom_delay_rst
446         !                                       --------------------------
447         IF ( SIZE(todelay(idvar)%z1d) /= isz ) THEN                  ! Check dimension coherence
448            IF(lwp) WRITE(numout,*) ' WARNING: the nb of delayed variables in restart file is not the model one'
449            DEALLOCATE(todelay(idvar)%z1d)
450            ndelayid(idvar) = -1                                      ! do as if we had no restart
451         END IF
452      ENDIF
453
454      IF( ndelayid(idvar) == -1 ) THEN         ! first call without restart: define %z1d from p_in with a blocking allreduce
455         !                                       --------------------------
456         ALLOCATE(todelay(idvar)%z1d(isz))
457         CALL mpi_allreduce( p_in(:), todelay(idvar)%z1d(:), isz, MPI_DOUBLE_PRECISION, mpi_max, ilocalcomm, ierr )   ! get %z1d
458      ENDIF
459
460      IF( ndelayid(idvar) > 0 )   CALL mpp_delay_rcv( idvar )         ! make sure %z1d is received
461
462      ! send back pout from todelay(idvar)%z1d defined at previous call
463      pout(:) = todelay(idvar)%z1d(:)
464
465      ! send p_in into todelay(idvar)%z1d with a non-blocking communication
466# if defined key_mpi2
467      IF( ln_timing ) CALL tic_tac( .TRUE., ld_global = .TRUE.)
468      CALL  mpi_allreduce( p_in(:), todelay(idvar)%z1d(:), isz, MPI_DOUBLE_PRECISION, mpi_max, ilocalcomm, ndelayid(idvar), ierr )
469      IF( ln_timing ) CALL tic_tac(.FALSE., ld_global = .TRUE.)
470# else
471      CALL mpi_iallreduce( p_in(:), todelay(idvar)%z1d(:), isz, MPI_DOUBLE_PRECISION, mpi_max, ilocalcomm, ndelayid(idvar), ierr )
472# endif
473#else
474      pout(:) = p_in(:)
475#endif
476
477   END SUBROUTINE mpp_delay_max
478
479   
480   SUBROUTINE mpp_delay_rcv( kid )
481      !!----------------------------------------------------------------------
482      !!                   ***  routine mpp_delay_rcv  ***
483      !!
484      !! ** Purpose :  force barrier for delayed mpp (needed for restart)
485      !!
486      !!----------------------------------------------------------------------
487      INTEGER,INTENT(in   )      ::  kid 
488      INTEGER ::   ierr
489      !!----------------------------------------------------------------------
490#if defined key_mpp_mpi
491      IF( ndelayid(kid) /= -2 ) THEN 
492#if ! defined key_mpi2
493         IF( ln_timing ) CALL tic_tac( .TRUE., ld_global = .TRUE.)
494         CALL mpi_wait( ndelayid(kid), MPI_STATUS_IGNORE, ierr )                        ! make sure todelay(kid) is received
495         IF( ln_timing ) CALL tic_tac(.FALSE., ld_global = .TRUE.)
496#endif
497         IF( ASSOCIATED(todelay(kid)%y1d) )   todelay(kid)%z1d(:) = REAL(todelay(kid)%y1d(:), wp)  ! define %z1d from %y1d
498         ndelayid(kid) = -2   ! add flag to know that mpi_wait was already called on kid
499      ENDIF
500#endif
501   END SUBROUTINE mpp_delay_rcv
502
503   
504   !!----------------------------------------------------------------------
505   !!    ***  mppmax_a_int, mppmax_int, mppmax_a_real, mppmax_real  ***
506   !!   
507   !!----------------------------------------------------------------------
508   !!
509#  define OPERATION_MAX
510#  define INTEGER_TYPE
511#  define DIM_0d
512#     define ROUTINE_ALLREDUCE           mppmax_int
513#     include "mpp_allreduce_generic.h90"
514#     undef ROUTINE_ALLREDUCE
515#  undef DIM_0d
516#  define DIM_1d
517#     define ROUTINE_ALLREDUCE           mppmax_a_int
518#     include "mpp_allreduce_generic.h90"
519#     undef ROUTINE_ALLREDUCE
520#  undef DIM_1d
521#  undef INTEGER_TYPE
522!
523#  define REAL_TYPE
524#  define DIM_0d
525#     define ROUTINE_ALLREDUCE           mppmax_real
526#     include "mpp_allreduce_generic.h90"
527#     undef ROUTINE_ALLREDUCE
528#  undef DIM_0d
529#  define DIM_1d
530#     define ROUTINE_ALLREDUCE           mppmax_a_real
531#     include "mpp_allreduce_generic.h90"
532#     undef ROUTINE_ALLREDUCE
533#  undef DIM_1d
534#  undef REAL_TYPE
535#  undef OPERATION_MAX
536   !!----------------------------------------------------------------------
537   !!    ***  mppmin_a_int, mppmin_int, mppmin_a_real, mppmin_real  ***
538   !!   
539   !!----------------------------------------------------------------------
540   !!
541#  define OPERATION_MIN
542#  define INTEGER_TYPE
543#  define DIM_0d
544#     define ROUTINE_ALLREDUCE           mppmin_int
545#     include "mpp_allreduce_generic.h90"
546#     undef ROUTINE_ALLREDUCE
547#  undef DIM_0d
548#  define DIM_1d
549#     define ROUTINE_ALLREDUCE           mppmin_a_int
550#     include "mpp_allreduce_generic.h90"
551#     undef ROUTINE_ALLREDUCE
552#  undef DIM_1d
553#  undef INTEGER_TYPE
554!
555#  define REAL_TYPE
556#  define DIM_0d
557#     define ROUTINE_ALLREDUCE           mppmin_real
558#     include "mpp_allreduce_generic.h90"
559#     undef ROUTINE_ALLREDUCE
560#  undef DIM_0d
561#  define DIM_1d
562#     define ROUTINE_ALLREDUCE           mppmin_a_real
563#     include "mpp_allreduce_generic.h90"
564#     undef ROUTINE_ALLREDUCE
565#  undef DIM_1d
566#  undef REAL_TYPE
567#  undef OPERATION_MIN
568
569   !!----------------------------------------------------------------------
570   !!    ***  mppsum_a_int, mppsum_int, mppsum_a_real, mppsum_real  ***
571   !!   
572   !!   Global sum of 1D array or a variable (integer, real or complex)
573   !!----------------------------------------------------------------------
574   !!
575#  define OPERATION_SUM
576#  define INTEGER_TYPE
577#  define DIM_0d
578#     define ROUTINE_ALLREDUCE           mppsum_int
579#     include "mpp_allreduce_generic.h90"
580#     undef ROUTINE_ALLREDUCE
581#  undef DIM_0d
582#  define DIM_1d
583#     define ROUTINE_ALLREDUCE           mppsum_a_int
584#     include "mpp_allreduce_generic.h90"
585#     undef ROUTINE_ALLREDUCE
586#  undef DIM_1d
587#  undef INTEGER_TYPE
588!
589#  define REAL_TYPE
590#  define DIM_0d
591#     define ROUTINE_ALLREDUCE           mppsum_real
592#     include "mpp_allreduce_generic.h90"
593#     undef ROUTINE_ALLREDUCE
594#  undef DIM_0d
595#  define DIM_1d
596#     define ROUTINE_ALLREDUCE           mppsum_a_real
597#     include "mpp_allreduce_generic.h90"
598#     undef ROUTINE_ALLREDUCE
599#  undef DIM_1d
600#  undef REAL_TYPE
601#  undef OPERATION_SUM
602
603#  define OPERATION_SUM_DD
604#  define COMPLEX_TYPE
605#  define DIM_0d
606#     define ROUTINE_ALLREDUCE           mppsum_realdd
607#     include "mpp_allreduce_generic.h90"
608#     undef ROUTINE_ALLREDUCE
609#  undef DIM_0d
610#  define DIM_1d
611#     define ROUTINE_ALLREDUCE           mppsum_a_realdd
612#     include "mpp_allreduce_generic.h90"
613#     undef ROUTINE_ALLREDUCE
614#  undef DIM_1d
615#  undef COMPLEX_TYPE
616#  undef OPERATION_SUM_DD
617
618   !!----------------------------------------------------------------------
619   !!    ***  mpp_minloc2d, mpp_minloc3d, mpp_maxloc2d, mpp_maxloc3d
620   !!   
621   !!----------------------------------------------------------------------
622   !!
623#  define OPERATION_MINLOC
624#  define DIM_2d
625#     define ROUTINE_LOC           mpp_minloc2d
626#     include "mpp_loc_generic.h90"
627#     undef ROUTINE_LOC
628#  undef DIM_2d
629#  define DIM_3d
630#     define ROUTINE_LOC           mpp_minloc3d
631#     include "mpp_loc_generic.h90"
632#     undef ROUTINE_LOC
633#  undef DIM_3d
634#  undef OPERATION_MINLOC
635
636#  define OPERATION_MAXLOC
637#  define DIM_2d
638#     define ROUTINE_LOC           mpp_maxloc2d
639#     include "mpp_loc_generic.h90"
640#     undef ROUTINE_LOC
641#  undef DIM_2d
642#  define DIM_3d
643#     define ROUTINE_LOC           mpp_maxloc3d
644#     include "mpp_loc_generic.h90"
645#     undef ROUTINE_LOC
646#  undef DIM_3d
647#  undef OPERATION_MAXLOC
648
649   SUBROUTINE mppsync()
650      !!----------------------------------------------------------------------
651      !!                  ***  routine mppsync  ***
652      !!
653      !! ** Purpose :   Massively parallel processors, synchroneous
654      !!
655      !!-----------------------------------------------------------------------
656      INTEGER :: ierror
657      !!-----------------------------------------------------------------------
658      !
659#if defined key_mpp_mpi
660      CALL mpi_barrier( mpi_comm_oce, ierror )
661#endif
662      !
663   END SUBROUTINE mppsync
664
665
666   SUBROUTINE mppstop( ld_abort ) 
667      !!----------------------------------------------------------------------
668      !!                  ***  routine mppstop  ***
669      !!
670      !! ** purpose :   Stop massively parallel processors method
671      !!
672      !!----------------------------------------------------------------------
673      LOGICAL, OPTIONAL, INTENT(in) :: ld_abort    ! source process number
674      LOGICAL ::   ll_abort
675      INTEGER ::   info
676      !!----------------------------------------------------------------------
677      ll_abort = .FALSE.
678      IF( PRESENT(ld_abort) ) ll_abort = ld_abort
679      !
680#if defined key_mpp_mpi
681      IF(ll_abort) THEN
682         CALL mpi_abort( MPI_COMM_WORLD )
683      ELSE
684         CALL mppsync
685         CALL mpi_finalize( info )
686      ENDIF
687#endif
688      IF( ll_abort ) STOP 123
689      !
690   END SUBROUTINE mppstop
691
692
693   SUBROUTINE mpp_comm_free( kcom )
694      !!----------------------------------------------------------------------
695      INTEGER, INTENT(in) ::   kcom
696      !!
697      INTEGER :: ierr
698      !!----------------------------------------------------------------------
699      !
700#if defined key_mpp_mpi
701      CALL MPI_COMM_FREE(kcom, ierr)
702#endif
703      !
704   END SUBROUTINE mpp_comm_free
705
706
707   SUBROUTINE mpp_ini_znl( kumout )
708      !!----------------------------------------------------------------------
709      !!               ***  routine mpp_ini_znl  ***
710      !!
711      !! ** Purpose :   Initialize special communicator for computing zonal sum
712      !!
713      !! ** Method  : - Look for processors in the same row
714      !!              - Put their number in nrank_znl
715      !!              - Create group for the znl processors
716      !!              - Create a communicator for znl processors
717      !!              - Determine if processor should write znl files
718      !!
719      !! ** output
720      !!      ndim_rank_znl = number of processors on the same row
721      !!      ngrp_znl = group ID for the znl processors
722      !!      ncomm_znl = communicator for the ice procs.
723      !!      n_znl_root = number (in the world) of proc 0 in the ice comm.
724      !!
725      !!----------------------------------------------------------------------
726      INTEGER, INTENT(in) ::   kumout   ! ocean.output logical units
727      !
728      INTEGER :: jproc      ! dummy loop integer
729      INTEGER :: ierr, ii   ! local integer
730      INTEGER, ALLOCATABLE, DIMENSION(:) ::   kwork
731      !!----------------------------------------------------------------------
732#if defined key_mpp_mpi
733      !-$$     WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ngrp_world     : ', ngrp_world
734      !-$$     WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - mpi_comm_world : ', mpi_comm_world
735      !-$$     WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - mpi_comm_oce   : ', mpi_comm_oce
736      !
737      ALLOCATE( kwork(jpnij), STAT=ierr )
738      IF( ierr /= 0 ) CALL ctl_stop( 'STOP', 'mpp_ini_znl : failed to allocate 1D array of length jpnij')
739
740      IF( jpnj == 1 ) THEN
741         ngrp_znl  = ngrp_world
742         ncomm_znl = mpi_comm_oce
743      ELSE
744         !
745         CALL MPI_ALLGATHER ( njmpp, 1, mpi_integer, kwork, 1, mpi_integer, mpi_comm_oce, ierr )
746         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - kwork pour njmpp : ', kwork
747         !-$$        CALL flush(numout)
748         !
749         ! Count number of processors on the same row
750         ndim_rank_znl = 0
751         DO jproc=1,jpnij
752            IF ( kwork(jproc) == njmpp ) THEN
753               ndim_rank_znl = ndim_rank_znl + 1
754            ENDIF
755         END DO
756         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ndim_rank_znl : ', ndim_rank_znl
757         !-$$        CALL flush(numout)
758         ! Allocate the right size to nrank_znl
759         IF (ALLOCATED (nrank_znl)) DEALLOCATE(nrank_znl)
760         ALLOCATE(nrank_znl(ndim_rank_znl))
761         ii = 0
762         nrank_znl (:) = 0
763         DO jproc=1,jpnij
764            IF ( kwork(jproc) == njmpp) THEN
765               ii = ii + 1
766               nrank_znl(ii) = jproc -1
767            ENDIF
768         END DO
769         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - nrank_znl : ', nrank_znl
770         !-$$        CALL flush(numout)
771
772         ! Create the opa group
773         CALL MPI_COMM_GROUP(mpi_comm_oce,ngrp_opa,ierr)
774         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ngrp_opa : ', ngrp_opa
775         !-$$        CALL flush(numout)
776
777         ! Create the znl group from the opa group
778         CALL MPI_GROUP_INCL  ( ngrp_opa, ndim_rank_znl, nrank_znl, ngrp_znl, ierr )
779         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ngrp_znl ', ngrp_znl
780         !-$$        CALL flush(numout)
781
782         ! Create the znl communicator from the opa communicator, ie the pool of procs in the same row
783         CALL MPI_COMM_CREATE ( mpi_comm_oce, ngrp_znl, ncomm_znl, ierr )
784         !-$$        WRITE (numout,*) 'mpp_ini_znl ', nproc, ' - ncomm_znl ', ncomm_znl
785         !-$$        CALL flush(numout)
786         !
787      END IF
788
789      ! Determines if processor if the first (starting from i=1) on the row
790      IF ( jpni == 1 ) THEN
791         l_znl_root = .TRUE.
792      ELSE
793         l_znl_root = .FALSE.
794         kwork (1) = nimpp
795         CALL mpp_min ( 'lib_mpp', kwork(1), kcom = ncomm_znl)
796         IF ( nimpp == kwork(1)) l_znl_root = .TRUE.
797      END IF
798
799      DEALLOCATE(kwork)
800#endif
801
802   END SUBROUTINE mpp_ini_znl
803
804
805   SUBROUTINE mpp_ini_north
806      !!----------------------------------------------------------------------
807      !!               ***  routine mpp_ini_north  ***
808      !!
809      !! ** Purpose :   Initialize special communicator for north folding
810      !!      condition together with global variables needed in the mpp folding
811      !!
812      !! ** Method  : - Look for northern processors
813      !!              - Put their number in nrank_north
814      !!              - Create groups for the world processors and the north processors
815      !!              - Create a communicator for northern processors
816      !!
817      !! ** output
818      !!      njmppmax = njmpp for northern procs
819      !!      ndim_rank_north = number of processors in the northern line
820      !!      nrank_north (ndim_rank_north) = number  of the northern procs.
821      !!      ngrp_world = group ID for the world processors
822      !!      ngrp_north = group ID for the northern processors
823      !!      ncomm_north = communicator for the northern procs.
824      !!      north_root = number (in the world) of proc 0 in the northern comm.
825      !!
826      !!----------------------------------------------------------------------
827      INTEGER ::   ierr
828      INTEGER ::   jjproc
829      INTEGER ::   ii, ji
830      !!----------------------------------------------------------------------
831      !
832#if defined key_mpp_mpi
833      njmppmax = MAXVAL( njmppt )
834      !
835      ! Look for how many procs on the northern boundary
836      ndim_rank_north = 0
837      DO jjproc = 1, jpnij
838         IF( njmppt(jjproc) == njmppmax )   ndim_rank_north = ndim_rank_north + 1
839      END DO
840      !
841      ! Allocate the right size to nrank_north
842      IF (ALLOCATED (nrank_north)) DEALLOCATE(nrank_north)
843      ALLOCATE( nrank_north(ndim_rank_north) )
844
845      ! Fill the nrank_north array with proc. number of northern procs.
846      ! Note : the rank start at 0 in MPI
847      ii = 0
848      DO ji = 1, jpnij
849         IF ( njmppt(ji) == njmppmax   ) THEN
850            ii=ii+1
851            nrank_north(ii)=ji-1
852         END IF
853      END DO
854      !
855      ! create the world group
856      CALL MPI_COMM_GROUP( mpi_comm_oce, ngrp_world, ierr )
857      !
858      ! Create the North group from the world group
859      CALL MPI_GROUP_INCL( ngrp_world, ndim_rank_north, nrank_north, ngrp_north, ierr )
860      !
861      ! Create the North communicator , ie the pool of procs in the north group
862      CALL MPI_COMM_CREATE( mpi_comm_oce, ngrp_north, ncomm_north, ierr )
863      !
864#endif
865   END SUBROUTINE mpp_ini_north
866
867
868   SUBROUTINE DDPDD_MPI( ydda, yddb, ilen, itype )
869      !!---------------------------------------------------------------------
870      !!   Routine DDPDD_MPI: used by reduction operator MPI_SUMDD
871      !!
872      !!   Modification of original codes written by David H. Bailey
873      !!   This subroutine computes yddb(i) = ydda(i)+yddb(i)
874      !!---------------------------------------------------------------------
875      INTEGER                     , INTENT(in)    ::   ilen, itype
876      COMPLEX(wp), DIMENSION(ilen), INTENT(in)    ::   ydda
877      COMPLEX(wp), DIMENSION(ilen), INTENT(inout) ::   yddb
878      !
879      REAL(wp) :: zerr, zt1, zt2    ! local work variables
880      INTEGER  :: ji, ztmp           ! local scalar
881      !!---------------------------------------------------------------------
882      !
883      ztmp = itype   ! avoid compilation warning
884      !
885      DO ji=1,ilen
886      ! Compute ydda + yddb using Knuth's trick.
887         zt1  = real(ydda(ji)) + real(yddb(ji))
888         zerr = zt1 - real(ydda(ji))
889         zt2  = ((real(yddb(ji)) - zerr) + (real(ydda(ji)) - (zt1 - zerr))) &
890                + aimag(ydda(ji)) + aimag(yddb(ji))
891
892         ! The result is zt1 + zt2, after normalization.
893         yddb(ji) = cmplx ( zt1 + zt2, zt2 - ((zt1 + zt2) - zt1),wp )
894      END DO
895      !
896   END SUBROUTINE DDPDD_MPI
897
898
899   SUBROUTINE mpp_report( cdname, kpk, kpl, kpf, ld_lbc, ld_glb, ld_dlg )
900      !!----------------------------------------------------------------------
901      !!                  ***  routine mpp_report  ***
902      !!
903      !! ** Purpose :   report use of mpp routines per time-setp
904      !!
905      !!----------------------------------------------------------------------
906      CHARACTER(len=*),           INTENT(in   ) ::   cdname      ! name of the calling subroutine
907      INTEGER         , OPTIONAL, INTENT(in   ) ::   kpk, kpl, kpf
908      LOGICAL         , OPTIONAL, INTENT(in   ) ::   ld_lbc, ld_glb, ld_dlg
909      !!
910      CHARACTER(len=128)                        ::   ccountname  ! name of a subroutine to count communications
911      LOGICAL ::   ll_lbc, ll_glb, ll_dlg
912      INTEGER ::    ji,  jj,  jk,  jh, jf, jcount   ! dummy loop indices
913      !!----------------------------------------------------------------------
914#if defined key_mpp_mpi
915      !
916      ll_lbc = .FALSE.
917      IF( PRESENT(ld_lbc) ) ll_lbc = ld_lbc
918      ll_glb = .FALSE.
919      IF( PRESENT(ld_glb) ) ll_glb = ld_glb
920      ll_dlg = .FALSE.
921      IF( PRESENT(ld_dlg) ) ll_dlg = ld_dlg
922      !
923      ! find the smallest common frequency: default = frequency product, if multiple, choose the larger of the 2 frequency
924      IF( ncom_dttrc /= 1 )   CALL ctl_stop( 'STOP', 'mpp_report, ncom_dttrc /= 1 not coded...' ) 
925      ncom_freq = ncom_fsbc
926      !
927      IF ( ncom_stp == nit000+ncom_freq ) THEN   ! avoid to count extra communications in potential initializations at nit000
928         IF( ll_lbc ) THEN
929            IF( .NOT. ALLOCATED(ncomm_sequence) ) ALLOCATE( ncomm_sequence(ncom_rec_max,2) )
930            IF( .NOT. ALLOCATED(    crname_lbc) ) ALLOCATE(     crname_lbc(ncom_rec_max  ) )
931            n_sequence_lbc = n_sequence_lbc + 1
932            IF( n_sequence_lbc > ncom_rec_max ) CALL ctl_stop( 'STOP', 'lib_mpp, increase ncom_rec_max' )   ! deadlock
933            crname_lbc(n_sequence_lbc) = cdname     ! keep the name of the calling routine
934            ncomm_sequence(n_sequence_lbc,1) = kpk*kpl   ! size of 3rd and 4th dimensions
935            ncomm_sequence(n_sequence_lbc,2) = kpf       ! number of arrays to be treated (multi)
936         ENDIF
937         IF( ll_glb ) THEN
938            IF( .NOT. ALLOCATED(crname_glb) ) ALLOCATE( crname_glb(ncom_rec_max) )
939            n_sequence_glb = n_sequence_glb + 1
940            IF( n_sequence_glb > ncom_rec_max ) CALL ctl_stop( 'STOP', 'lib_mpp, increase ncom_rec_max' )   ! deadlock
941            crname_glb(n_sequence_glb) = cdname     ! keep the name of the calling routine
942         ENDIF
943         IF( ll_dlg ) THEN
944            IF( .NOT. ALLOCATED(crname_dlg) ) ALLOCATE( crname_dlg(ncom_rec_max) )
945            n_sequence_dlg = n_sequence_dlg + 1
946            IF( n_sequence_dlg > ncom_rec_max ) CALL ctl_stop( 'STOP', 'lib_mpp, increase ncom_rec_max' )   ! deadlock
947            crname_dlg(n_sequence_dlg) = cdname     ! keep the name of the calling routine
948         ENDIF
949      ELSE IF ( ncom_stp == nit000+2*ncom_freq ) THEN
950         CALL ctl_opn( numcom, 'communication_report.txt', 'REPLACE', 'FORMATTED', 'SEQUENTIAL', -1, numout, .FALSE., narea )
951         WRITE(numcom,*) ' '
952         WRITE(numcom,*) ' ------------------------------------------------------------'
953         WRITE(numcom,*) ' Communication pattern report (second oce+sbc+top time step):'
954         WRITE(numcom,*) ' ------------------------------------------------------------'
955         WRITE(numcom,*) ' '
956         WRITE(numcom,'(A,I4)') ' Exchanged halos : ', n_sequence_lbc
957         jj = 0; jk = 0; jf = 0; jh = 0
958         DO ji = 1, n_sequence_lbc
959            IF ( ncomm_sequence(ji,1) .GT. 1 ) jk = jk + 1
960            IF ( ncomm_sequence(ji,2) .GT. 1 ) jf = jf + 1
961            IF ( ncomm_sequence(ji,1) .GT. 1 .AND. ncomm_sequence(ji,2) .GT. 1 ) jj = jj + 1
962            jh = MAX (jh, ncomm_sequence(ji,1)*ncomm_sequence(ji,2))
963         END DO
964         WRITE(numcom,'(A,I3)') ' 3D Exchanged halos : ', jk
965         WRITE(numcom,'(A,I3)') ' Multi arrays exchanged halos : ', jf
966         WRITE(numcom,'(A,I3)') '   from which 3D : ', jj
967         WRITE(numcom,'(A,I10)') ' Array max size : ', jh*jpi*jpj
968         WRITE(numcom,*) ' '
969         WRITE(numcom,*) ' lbc_lnk called'
970         DO ji = 1, n_sequence_lbc - 1
971            IF ( crname_lbc(ji) /= 'already counted' ) THEN
972               ccountname = crname_lbc(ji)
973               crname_lbc(ji) = 'already counted'
974               jcount = 1
975               DO jj = ji + 1, n_sequence_lbc
976                  IF ( ccountname ==  crname_lbc(jj) ) THEN
977                     jcount = jcount + 1
978                     crname_lbc(jj) = 'already counted'
979                  END IF
980               END DO
981               WRITE(numcom,'(A, I4, A, A)') ' - ', jcount,' times by subroutine ', TRIM(ccountname)
982            END IF
983         END DO
984         IF ( crname_lbc(n_sequence_lbc) /= 'already counted' ) THEN
985            WRITE(numcom,'(A, I4, A, A)') ' - ', 1,' times by subroutine ', TRIM(crname_lbc(ncom_rec_max))
986         END IF
987         WRITE(numcom,*) ' '
988         IF ( n_sequence_glb > 0 ) THEN
989            WRITE(numcom,'(A,I4)') ' Global communications : ', n_sequence_glb
990            jj = 1
991            DO ji = 2, n_sequence_glb
992               IF( crname_glb(ji-1) /= crname_glb(ji) ) THEN
993                  WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_glb(ji-1))
994                  jj = 0
995               END IF
996               jj = jj + 1 
997            END DO
998            WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_glb(n_sequence_glb))
999            DEALLOCATE(crname_glb)
1000         ELSE
1001            WRITE(numcom,*) ' No MPI global communication '
1002         ENDIF
1003         WRITE(numcom,*) ' '
1004         IF ( n_sequence_dlg > 0 ) THEN
1005            WRITE(numcom,'(A,I4)') ' Delayed global communications : ', n_sequence_dlg
1006            jj = 1
1007            DO ji = 2, n_sequence_dlg
1008               IF( crname_dlg(ji-1) /= crname_dlg(ji) ) THEN
1009                  WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_dlg(ji-1))
1010                  jj = 0
1011               END IF
1012               jj = jj + 1 
1013            END DO
1014            WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_dlg(n_sequence_dlg))
1015            DEALLOCATE(crname_dlg)
1016         ELSE
1017            WRITE(numcom,*) ' No MPI delayed global communication '
1018         ENDIF
1019         WRITE(numcom,*) ' '
1020         WRITE(numcom,*) ' -----------------------------------------------'
1021         WRITE(numcom,*) ' '
1022         DEALLOCATE(ncomm_sequence)
1023         DEALLOCATE(crname_lbc)
1024      ENDIF
1025#endif
1026   END SUBROUTINE mpp_report
1027
1028   
1029   SUBROUTINE tic_tac (ld_tic, ld_global)
1030
1031    LOGICAL,           INTENT(IN) :: ld_tic
1032    LOGICAL, OPTIONAL, INTENT(IN) :: ld_global
1033    REAL(wp), DIMENSION(2), SAVE :: tic_wt
1034    REAL(wp),               SAVE :: tic_ct = 0._wp
1035    INTEGER :: ii
1036#if defined key_mpp_mpi
1037
1038    IF( ncom_stp <= nit000 ) RETURN
1039    IF( ncom_stp == nitend ) RETURN
1040    ii = 1
1041    IF( PRESENT( ld_global ) ) THEN
1042       IF( ld_global ) ii = 2
1043    END IF
1044   
1045    IF ( ld_tic ) THEN
1046       tic_wt(ii) = MPI_Wtime()                                                    ! start count tic->tac (waiting time)
1047       IF ( tic_ct > 0.0_wp ) compute_time = compute_time + MPI_Wtime() - tic_ct   ! cumulate count tac->tic
1048    ELSE
1049       waiting_time(ii) = waiting_time(ii) + MPI_Wtime() - tic_wt(ii)              ! cumulate count tic->tac
1050       tic_ct = MPI_Wtime()                                                        ! start count tac->tic (waiting time)
1051    ENDIF
1052#endif
1053   
1054   END SUBROUTINE tic_tac
1055
1056#if ! defined key_mpp_mpi
1057   SUBROUTINE mpi_wait(request, status, ierror)
1058      INTEGER                            , INTENT(in   ) ::   request
1059      INTEGER, DIMENSION(MPI_STATUS_SIZE), INTENT(  out) ::   status
1060      INTEGER                            , INTENT(  out) ::   ierror
1061   END SUBROUTINE mpi_wait
1062
1063   
1064   FUNCTION MPI_Wtime()
1065      REAL(wp) ::  MPI_Wtime
1066      MPI_Wtime = -1.
1067   END FUNCTION MPI_Wtime
1068#endif
1069
1070   !!----------------------------------------------------------------------
1071   !!   ctl_stop, ctl_warn, get_unit, ctl_opn, ctl_nam, load_nml   routines
1072   !!----------------------------------------------------------------------
1073
1074   SUBROUTINE ctl_stop( cd1, cd2, cd3, cd4, cd5 ,   &
1075      &                 cd6, cd7, cd8, cd9, cd10 )
1076      !!----------------------------------------------------------------------
1077      !!                  ***  ROUTINE  stop_opa  ***
1078      !!
1079      !! ** Purpose :   print in ocean.outpput file a error message and
1080      !!                increment the error number (nstop) by one.
1081      !!----------------------------------------------------------------------
1082      CHARACTER(len=*), INTENT(in   )           ::   cd1
1083      CHARACTER(len=*), INTENT(in   ), OPTIONAL ::        cd2, cd3, cd4, cd5
1084      CHARACTER(len=*), INTENT(in   ), OPTIONAL ::   cd6, cd7, cd8, cd9, cd10
1085      !!----------------------------------------------------------------------
1086      !
1087      nstop = nstop + 1
1088      !
1089      ! force to open ocean.output file if not already opened
1090      IF( numout == 6 ) CALL ctl_opn( numout, 'ocean.output', 'APPEND', 'FORMATTED', 'SEQUENTIAL', -1, 6, .FALSE. )
1091      !
1092                            WRITE(numout,*)
1093                            WRITE(numout,*) ' ===>>> : E R R O R'
1094                            WRITE(numout,*)
1095                            WRITE(numout,*) '         ==========='
1096                            WRITE(numout,*)
1097                            WRITE(numout,*) TRIM(cd1)
1098      IF( PRESENT(cd2 ) )   WRITE(numout,*) TRIM(cd2)
1099      IF( PRESENT(cd3 ) )   WRITE(numout,*) TRIM(cd3)
1100      IF( PRESENT(cd4 ) )   WRITE(numout,*) TRIM(cd4)
1101      IF( PRESENT(cd5 ) )   WRITE(numout,*) TRIM(cd5)
1102      IF( PRESENT(cd6 ) )   WRITE(numout,*) TRIM(cd6)
1103      IF( PRESENT(cd7 ) )   WRITE(numout,*) TRIM(cd7)
1104      IF( PRESENT(cd8 ) )   WRITE(numout,*) TRIM(cd8)
1105      IF( PRESENT(cd9 ) )   WRITE(numout,*) TRIM(cd9)
1106      IF( PRESENT(cd10) )   WRITE(numout,*) TRIM(cd10)
1107                            WRITE(numout,*)
1108      !
1109                               CALL FLUSH(numout    )
1110      IF( numstp     /= -1 )   CALL FLUSH(numstp    )
1111      IF( numrun     /= -1 )   CALL FLUSH(numrun    )
1112      IF( numevo_ice /= -1 )   CALL FLUSH(numevo_ice)
1113      !
1114      IF( cd1 == 'STOP' ) THEN
1115         WRITE(numout,*) 
1116         WRITE(numout,*)  'huge E-R-R-O-R : immediate stop'
1117         WRITE(numout,*) 
1118         CALL mppstop( ld_abort = .true. )
1119      ENDIF
1120      !
1121   END SUBROUTINE ctl_stop
1122
1123
1124   SUBROUTINE ctl_warn( cd1, cd2, cd3, cd4, cd5,   &
1125      &                 cd6, cd7, cd8, cd9, cd10 )
1126      !!----------------------------------------------------------------------
1127      !!                  ***  ROUTINE  stop_warn  ***
1128      !!
1129      !! ** Purpose :   print in ocean.outpput file a error message and
1130      !!                increment the warning number (nwarn) by one.
1131      !!----------------------------------------------------------------------
1132      CHARACTER(len=*), INTENT(in), OPTIONAL ::  cd1, cd2, cd3, cd4, cd5
1133      CHARACTER(len=*), INTENT(in), OPTIONAL ::  cd6, cd7, cd8, cd9, cd10
1134      !!----------------------------------------------------------------------
1135      !
1136      nwarn = nwarn + 1
1137      !
1138      IF(lwp) THEN
1139                               WRITE(numout,*)
1140                               WRITE(numout,*) ' ===>>> : W A R N I N G'
1141                               WRITE(numout,*)
1142                               WRITE(numout,*) '         ==============='
1143                               WRITE(numout,*)
1144         IF( PRESENT(cd1 ) )   WRITE(numout,*) TRIM(cd1)
1145         IF( PRESENT(cd2 ) )   WRITE(numout,*) TRIM(cd2)
1146         IF( PRESENT(cd3 ) )   WRITE(numout,*) TRIM(cd3)
1147         IF( PRESENT(cd4 ) )   WRITE(numout,*) TRIM(cd4)
1148         IF( PRESENT(cd5 ) )   WRITE(numout,*) TRIM(cd5)
1149         IF( PRESENT(cd6 ) )   WRITE(numout,*) TRIM(cd6)
1150         IF( PRESENT(cd7 ) )   WRITE(numout,*) TRIM(cd7)
1151         IF( PRESENT(cd8 ) )   WRITE(numout,*) TRIM(cd8)
1152         IF( PRESENT(cd9 ) )   WRITE(numout,*) TRIM(cd9)
1153         IF( PRESENT(cd10) )   WRITE(numout,*) TRIM(cd10)
1154                               WRITE(numout,*)
1155      ENDIF
1156      CALL FLUSH(numout)
1157      !
1158   END SUBROUTINE ctl_warn
1159
1160
1161   SUBROUTINE ctl_opn( knum, cdfile, cdstat, cdform, cdacce, klengh, kout, ldwp, karea )
1162      !!----------------------------------------------------------------------
1163      !!                  ***  ROUTINE ctl_opn  ***
1164      !!
1165      !! ** Purpose :   Open file and check if required file is available.
1166      !!
1167      !! ** Method  :   Fortan open
1168      !!----------------------------------------------------------------------
1169      INTEGER          , INTENT(  out) ::   knum      ! logical unit to open
1170      CHARACTER(len=*) , INTENT(in   ) ::   cdfile    ! file name to open
1171      CHARACTER(len=*) , INTENT(in   ) ::   cdstat    ! disposition specifier
1172      CHARACTER(len=*) , INTENT(in   ) ::   cdform    ! formatting specifier
1173      CHARACTER(len=*) , INTENT(in   ) ::   cdacce    ! access specifier
1174      INTEGER          , INTENT(in   ) ::   klengh    ! record length
1175      INTEGER          , INTENT(in   ) ::   kout      ! number of logical units for write
1176      LOGICAL          , INTENT(in   ) ::   ldwp      ! boolean term for print
1177      INTEGER, OPTIONAL, INTENT(in   ) ::   karea     ! proc number
1178      !
1179      CHARACTER(len=80) ::   clfile
1180      INTEGER           ::   iost
1181      !!----------------------------------------------------------------------
1182      !
1183      ! adapt filename
1184      ! ----------------
1185      clfile = TRIM(cdfile)
1186      IF( PRESENT( karea ) ) THEN
1187         IF( karea > 1 )   WRITE(clfile, "(a,'_',i4.4)") TRIM(clfile), karea-1
1188      ENDIF
1189#if defined key_agrif
1190      IF( .NOT. Agrif_Root() )   clfile = TRIM(Agrif_CFixed())//'_'//TRIM(clfile)
1191      knum=Agrif_Get_Unit()
1192#else
1193      knum=get_unit()
1194#endif
1195      IF( TRIM(cdfile) == '/dev/null' )   clfile = TRIM(cdfile)   ! force the use of /dev/null
1196      !
1197      IF(       cdacce(1:6) == 'DIRECT' )  THEN   ! cdacce has always more than 6 characters
1198         OPEN( UNIT=knum, FILE=clfile, FORM=cdform, ACCESS=cdacce, STATUS=cdstat, RECL=klengh         , ERR=100, IOSTAT=iost )
1199      ELSE IF( TRIM(cdstat) == 'APPEND' )  THEN   ! cdstat can have less than 6 characters
1200         OPEN( UNIT=knum, FILE=clfile, FORM=cdform, ACCESS=cdacce, STATUS='UNKNOWN', POSITION='APPEND', ERR=100, IOSTAT=iost )
1201      ELSE
1202         OPEN( UNIT=knum, FILE=clfile, FORM=cdform, ACCESS=cdacce, STATUS=cdstat                      , ERR=100, IOSTAT=iost )
1203      ENDIF
1204      IF( iost /= 0 .AND. TRIM(clfile) == '/dev/null' ) &   ! for windows
1205         &  OPEN(UNIT=knum,FILE='NUL', FORM=cdform, ACCESS=cdacce, STATUS=cdstat                      , ERR=100, IOSTAT=iost )   
1206      IF( iost == 0 ) THEN
1207         IF(ldwp) THEN
1208            WRITE(kout,*) '     file   : ', TRIM(clfile),' open ok'
1209            WRITE(kout,*) '     unit   = ', knum
1210            WRITE(kout,*) '     status = ', cdstat
1211            WRITE(kout,*) '     form   = ', cdform
1212            WRITE(kout,*) '     access = ', cdacce
1213            WRITE(kout,*)
1214         ENDIF
1215      ENDIF
1216100   CONTINUE
1217      IF( iost /= 0 ) THEN
1218         WRITE(ctmp1,*) ' ===>>>> : bad opening file: ', TRIM(clfile)
1219         WRITE(ctmp2,*) ' =======   ===  '
1220         WRITE(ctmp3,*) '           unit   = ', knum
1221         WRITE(ctmp4,*) '           status = ', cdstat
1222         WRITE(ctmp5,*) '           form   = ', cdform
1223         WRITE(ctmp6,*) '           access = ', cdacce
1224         WRITE(ctmp7,*) '           iostat = ', iost
1225         WRITE(ctmp8,*) '           we stop. verify the file '
1226         CALL ctl_stop( 'STOP', ctmp1, ctmp2, ctmp3, ctmp4, ctmp5, ctmp6, ctmp7, ctmp8 )
1227      ENDIF
1228      !
1229   END SUBROUTINE ctl_opn
1230
1231
1232   SUBROUTINE ctl_nam ( kios, cdnam )
1233      !!----------------------------------------------------------------------
1234      !!                  ***  ROUTINE ctl_nam  ***
1235      !!
1236      !! ** Purpose :   Informations when error while reading a namelist
1237      !!
1238      !! ** Method  :   Fortan open
1239      !!----------------------------------------------------------------------
1240      INTEGER                                , INTENT(inout) ::   kios    ! IO status after reading the namelist
1241      CHARACTER(len=*)                       , INTENT(in   ) ::   cdnam   ! group name of namelist for which error occurs
1242      !
1243      CHARACTER(len=5) ::   clios   ! string to convert iostat in character for print
1244      !!----------------------------------------------------------------------
1245      !
1246      WRITE (clios, '(I5.0)')   kios
1247      IF( kios < 0 ) THEN         
1248         CALL ctl_warn( 'end of record or file while reading namelist '   &
1249            &           // TRIM(cdnam) // ' iostat = ' // TRIM(clios) )
1250      ENDIF
1251      !
1252      IF( kios > 0 ) THEN
1253         CALL ctl_stop( 'misspelled variable in namelist '   &
1254            &           // TRIM(cdnam) // ' iostat = ' // TRIM(clios) )
1255      ENDIF
1256      kios = 0
1257      !
1258   END SUBROUTINE ctl_nam
1259
1260
1261   INTEGER FUNCTION get_unit()
1262      !!----------------------------------------------------------------------
1263      !!                  ***  FUNCTION  get_unit  ***
1264      !!
1265      !! ** Purpose :   return the index of an unused logical unit
1266      !!----------------------------------------------------------------------
1267      LOGICAL :: llopn
1268      !!----------------------------------------------------------------------
1269      !
1270      get_unit = 15   ! choose a unit that is big enough then it is not already used in NEMO
1271      llopn = .TRUE.
1272      DO WHILE( (get_unit < 998) .AND. llopn )
1273         get_unit = get_unit + 1
1274         INQUIRE( unit = get_unit, opened = llopn )
1275      END DO
1276      IF( (get_unit == 999) .AND. llopn ) THEN
1277         CALL ctl_stop( 'STOP', 'get_unit: All logical units until 999 are used...' )
1278      ENDIF
1279      !
1280   END FUNCTION get_unit
1281
1282   SUBROUTINE load_nml( cdnambuff , cdnamfile, kout, ldwp)
1283      CHARACTER(LEN=:)    , ALLOCATABLE, INTENT(INOUT) :: cdnambuff
1284      CHARACTER(LEN=*), INTENT(IN )                :: cdnamfile
1285      CHARACTER(LEN=256)                           :: chline
1286      INTEGER, INTENT(IN)                          :: kout
1287      LOGICAL, INTENT(IN)                          :: ldwp
1288      INTEGER                                      :: itot, iun, iltc, inl, ios
1289      !
1290      ! Check if the namelist buffer has already been allocated. Return if it has.
1291      !
1292      IF ( ALLOCATED( cdnambuff ) ) RETURN
1293      !
1294      ! Open namelist file
1295      !
1296      CALL ctl_opn( iun, cdnamfile, 'OLD', 'FORMATTED', 'SEQUENTIAL', -1, kout, ldwp )
1297      !
1298      ! First pass: count characters excluding comments and trimable white space
1299      !
1300      itot=0
1301  10  READ(iun,'(A256)',END=20,ERR=20) chline
1302      iltc = LEN_TRIM(chline)
1303      IF ( iltc.GT.0 ) THEN
1304       inl = INDEX(chline, '!') 
1305       IF( inl.eq.0 ) THEN
1306        itot = itot + iltc + 1                                ! +1 for the newline character
1307       ELSEIF( inl.GT.0 .AND. LEN_TRIM( chline(1:inl-1) ).GT.0 ) THEN
1308        itot = itot + inl                                  !  includes +1 for the newline character
1309       ENDIF
1310      ENDIF
1311      GOTO 10
1312  20  CONTINUE
1313      !
1314      ! Allocate text cdnambuff for condensed namelist
1315      !
1316      ALLOCATE( CHARACTER(LEN=itot) :: cdnambuff )
1317      WRITE(*,*) 'ALLOCATED ', itot
1318      !
1319      ! Second pass: read and transfer pruned characters into cdnambuff
1320      !
1321      REWIND(iun)
1322      itot=1
1323  30  READ(iun,'(A256)',END=40,ERR=40) chline
1324      iltc = LEN_TRIM(chline)
1325      IF ( iltc.GT.0 ) THEN
1326       inl = INDEX(chline, '!')
1327       IF( inl.eq.0 ) THEN
1328        inl = iltc
1329       ELSE
1330        inl = inl - 1
1331       ENDIF
1332       IF( inl.GT.0 .AND. LEN_TRIM( chline(1:inl) ).GT.0 ) THEN
1333          cdnambuff(itot:itot+inl-1) = chline(1:inl)
1334          WRITE( cdnambuff(itot+inl:itot+inl), '(a)' ) NEW_LINE('A')
1335          itot = itot + inl + 1
1336       ENDIF
1337      ENDIF
1338      GOTO 30
1339  40  CONTINUE
1340      WRITE(*,*) 'ASSIGNED ',itot - 1
1341      !
1342      ! Close namelist file
1343      !
1344      CLOSE(iun)
1345      !write(*,'(32A)') cdnambuff
1346  END SUBROUTINE load_nml
1347
1348
1349   !!----------------------------------------------------------------------
1350END MODULE lib_mpp
Note: See TracBrowser for help on using the repository browser.