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

Since March 2022 along with NEMO 4.2 release, the code development moved to a self-hosted GitLab.
This present forge is now archived and remained online for history.

lib_mpp.F90 in NEMO/branches/2021/dev_r14312_MPI_Interface/src/OCE/LBC – NEMO

Context Navigation

source: NEMO/branches/2021/dev_r14312_MPI_Interface/src/OCE/LBC/lib_mpp.F90 @ 14349

Last change on this file since 14349 was 14349, checked in by smasson, 4 years ago
dev_r14312_MPI_Interface: further simplifications of lbclk and lbcnfd, #2598
Property svn:keywords set to `Id`
File size: 72.4 KB

Line
1	MODULE 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	!! mpp_bcast_nml : broadcast/receive namelist character buffer from reading process to all others
54	!!----------------------------------------------------------------------
55	USE dom_oce ! ocean space and time domain
56	USE in_out_manager ! I/O manager
57	#if ! defined key_mpi_off
58	USE MPI
59	#endif
60
61	IMPLICIT NONE
62	PRIVATE
63	!
64	PUBLIC ctl_stop, ctl_warn, ctl_opn, ctl_nam, load_nml
65	PUBLIC mpp_start, mppstop, mppsync, mpp_comm_free
66	PUBLIC mpp_ini_north
67	PUBLIC mpp_min, mpp_max, mpp_sum, mpp_minloc, mpp_maxloc
68	PUBLIC mpp_delay_max, mpp_delay_sum, mpp_delay_rcv
69	PUBLIC mppscatter, mppgather
70	PUBLIC mpp_ini_znl
71	PUBLIC mpp_ini_nc
72	PUBLIC mppsend, mpprecv ! needed by TAM and ICB routines
73	PUBLIC mppsend_sp, mpprecv_sp ! needed by TAM and ICB routines
74	PUBLIC mppsend_dp, mpprecv_dp ! needed by TAM and ICB routines
75	PUBLIC mpp_report
76	PUBLIC mpp_bcast_nml
77	PUBLIC tic_tac
78	#if defined key_mpp_off
79	PUBLIC MPI_wait
80	PUBLIC MPI_Wtime
81	#endif
82
83	!! * Interfaces
84	!! define generic interface for these routine as they are called sometimes
85	!! with scalar arguments instead of array arguments, which causes problems
86	!! for the compilation on AIX system as well as NEC and SGI. Ok on COMPACQ
87	INTERFACE mpp_min
88	MODULE PROCEDURE mppmin_a_int, mppmin_int
89	MODULE PROCEDURE mppmin_a_real_sp, mppmin_real_sp
90	MODULE PROCEDURE mppmin_a_real_dp, mppmin_real_dp
91	END INTERFACE
92	INTERFACE mpp_max
93	MODULE PROCEDURE mppmax_a_int, mppmax_int
94	MODULE PROCEDURE mppmax_a_real_sp, mppmax_real_sp
95	MODULE PROCEDURE mppmax_a_real_dp, mppmax_real_dp
96	END INTERFACE
97	INTERFACE mpp_sum
98	MODULE PROCEDURE mppsum_a_int, mppsum_int
99	MODULE PROCEDURE mppsum_realdd, mppsum_a_realdd
100	MODULE PROCEDURE mppsum_a_real_sp, mppsum_real_sp
101	MODULE PROCEDURE mppsum_a_real_dp, mppsum_real_dp
102	END INTERFACE
103	INTERFACE mpp_minloc
104	MODULE PROCEDURE mpp_minloc2d_sp ,mpp_minloc3d_sp
105	MODULE PROCEDURE mpp_minloc2d_dp ,mpp_minloc3d_dp
106	END INTERFACE
107	INTERFACE mpp_maxloc
108	MODULE PROCEDURE mpp_maxloc2d_sp ,mpp_maxloc3d_sp
109	MODULE PROCEDURE mpp_maxloc2d_dp ,mpp_maxloc3d_dp
110	END INTERFACE
111
112	TYPE, PUBLIC :: PTR_4D_sp !: array of 4D pointers (used in lbclnk and lbcnfd)
113	REAL(sp), DIMENSION (:,:,:,:), POINTER :: pt4d
114	END TYPE PTR_4D_sp
115
116	TYPE, PUBLIC :: PTR_4D_dp !: array of 4D pointers (used in lbclnk and lbcnfd)
117	REAL(dp), DIMENSION (:,:,:,:), POINTER :: pt4d
118	END TYPE PTR_4D_dp
119
120	!! ========================= !!
121	!! MPI variable definition !!
122	!! ========================= !!
123	#if ! defined key_mpi_off
124	LOGICAL, PUBLIC, PARAMETER :: lk_mpp = .TRUE. !: mpp flag
125	#else
126	INTEGER, PUBLIC, PARAMETER :: MPI_STATUS_SIZE = 1
127	INTEGER, PUBLIC, PARAMETER :: MPI_REAL = 4
128	INTEGER, PUBLIC, PARAMETER :: MPI_DOUBLE_PRECISION = 8
129	LOGICAL, PUBLIC, PARAMETER :: lk_mpp = .FALSE. !: mpp flag
130	#endif
131
132	INTEGER, PUBLIC :: mppsize ! number of process
133	INTEGER, PUBLIC :: mpprank ! process number [ 0 - size-1 ]
134	!$AGRIF_DO_NOT_TREAT
135	INTEGER, PUBLIC :: mpi_comm_oce ! opa local communicator
136	!$AGRIF_END_DO_NOT_TREAT
137
138	INTEGER :: MPI_SUMDD
139
140	! Neighbourgs informations
141	INTEGER, DIMENSION(8), PUBLIC :: mpinei !: 8-neighbourg MPI indexes (starting at 0, -1 if no neighbourg)
142	INTEGER, PARAMETER, PUBLIC :: jpwe = 1 !: WEst
143	INTEGER, PARAMETER, PUBLIC :: jpea = 2 !: EAst
144	INTEGER, PARAMETER, PUBLIC :: jpso = 3 !: SOuth
145	INTEGER, PARAMETER, PUBLIC :: jpno = 4 !: NOrth
146	INTEGER, PARAMETER, PUBLIC :: jpsw = 5 !: South-West
147	INTEGER, PARAMETER, PUBLIC :: jpse = 6 !: South-East
148	INTEGER, PARAMETER, PUBLIC :: jpnw = 7 !: North-West
149	INTEGER, PARAMETER, PUBLIC :: jpne = 8 !: North-East
150
151	LOGICAL, DIMENSION(8), PUBLIC :: l_SelfPerio ! should we explicitely take care of I/J periodicity
152	LOGICAL, PUBLIC :: l_IdoNFold
153
154	! variables used for zonal integration
155	INTEGER, PUBLIC :: ncomm_znl !: communicator made by the processors on the same zonal average
156	LOGICAL, PUBLIC :: l_znl_root !: True on the 'left'most processor on the same row
157	INTEGER :: ngrp_znl !: group ID for the znl processors
158	INTEGER :: ndim_rank_znl !: number of processors on the same zonal average
159	INTEGER, DIMENSION(:), ALLOCATABLE, SAVE :: nrank_znl ! dimension ndim_rank_znl, number of the procs into the same znl domain
160
161	! variables used for MPI3 neighbourhood collectives
162	INTEGER, PUBLIC :: mpi_nc_com4 ! MPI3 neighbourhood collectives communicator
163	INTEGER, PUBLIC :: mpi_nc_com8 ! MPI3 neighbourhood collectives communicator (with diagionals)
164
165	! North fold condition in mpp_mpi with jpni > 1 (PUBLIC for TAM)
166	INTEGER, PUBLIC :: ngrp_world !: group ID for the world processors
167	INTEGER, PUBLIC :: ngrp_opa !: group ID for the opa processors
168	INTEGER, PUBLIC :: ngrp_north !: group ID for the northern processors (to be fold)
169	INTEGER, PUBLIC :: ncomm_north !: communicator made by the processors belonging to ngrp_north
170	INTEGER, PUBLIC :: ndim_rank_north !: number of 'sea' processor in the northern line (can be /= jpni !)
171	INTEGER, PUBLIC :: njmppmax !: value of njmpp for the processors of the northern line
172	INTEGER, PUBLIC :: north_root !: number (in the comm_opa) of proc 0 in the northern comm
173	INTEGER, PUBLIC, DIMENSION(:), ALLOCATABLE, SAVE :: nrank_north !: dimension ndim_rank_north
174
175	! Communications summary report
176	CHARACTER(len=lca), DIMENSION(:), ALLOCATABLE :: crname_lbc !: names of lbc_lnk calling routines
177	CHARACTER(len=lca), DIMENSION(:), ALLOCATABLE :: crname_glb !: names of global comm calling routines
178	CHARACTER(len=lca), DIMENSION(:), ALLOCATABLE :: crname_dlg !: names of delayed global comm calling routines
179	INTEGER, PUBLIC :: ncom_stp = 0 !: copy of time step # istp
180	INTEGER, PUBLIC :: ncom_fsbc = 1 !: copy of sbc time step # nn_fsbc
181	INTEGER, PUBLIC :: ncom_freq !: frequency of comm diagnostic
182	INTEGER, PUBLIC , DIMENSION(:,:), ALLOCATABLE :: ncomm_sequence !: size of communicated arrays (halos)
183	INTEGER, PARAMETER, PUBLIC :: ncom_rec_max = 5000 !: max number of communication record
184	INTEGER, PUBLIC :: n_sequence_lbc = 0 !: # of communicated arraysvia lbc
185	INTEGER, PUBLIC :: n_sequence_glb = 0 !: # of global communications
186	INTEGER, PUBLIC :: n_sequence_dlg = 0 !: # of delayed global communications
187	INTEGER, PUBLIC :: numcom = -1 !: logical unit for communicaton report
188	LOGICAL, PUBLIC :: l_full_nf_update = .TRUE. !: logical for a full (2lines) update of bc at North fold report
189	INTEGER, PARAMETER, PUBLIC :: nbdelay = 2 !: number of delayed operations
190	!: name (used as id) of allreduce-delayed operations
191	! Warning: we must use the same character length in an array constructor (at least for gcc compiler)
192	CHARACTER(len=32), DIMENSION(nbdelay), PUBLIC :: c_delaylist = (/ 'cflice', 'fwb ' /)
193	!: component name where the allreduce-delayed operation is performed
194	CHARACTER(len=3), DIMENSION(nbdelay), PUBLIC :: c_delaycpnt = (/ 'ICE' , 'OCE' /)
195	TYPE, PUBLIC :: DELAYARR
196	REAL( wp), POINTER, DIMENSION(:) :: z1d => NULL()
197	COMPLEX(dp), POINTER, DIMENSION(:) :: y1d => NULL()
198	END TYPE DELAYARR
199	TYPE( DELAYARR ), DIMENSION(nbdelay), PUBLIC, SAVE :: todelay !: must have SAVE for default initialization of DELAYARR
200	INTEGER, DIMENSION(nbdelay), PUBLIC :: ndelayid = -1 !: mpi request id of the delayed operations
201
202	! timing summary report
203	REAL(dp), DIMENSION(2), PUBLIC :: waiting_time = 0._dp
204	REAL(dp) , PUBLIC :: compute_time = 0._dp, elapsed_time = 0._dp
205
206	REAL(wp), DIMENSION(:), ALLOCATABLE, SAVE :: tampon ! buffer in case of bsend
207
208	LOGICAL, PUBLIC :: ln_nnogather !: namelist control of northfold comms
209	INTEGER, PUBLIC :: nn_comm !: namelist control of comms
210
211	INTEGER, PUBLIC, PARAMETER :: jpfillnothing = 1
212	INTEGER, PUBLIC, PARAMETER :: jpfillcst = 2
213	INTEGER, PUBLIC, PARAMETER :: jpfillcopy = 3
214	INTEGER, PUBLIC, PARAMETER :: jpfillperio = 4
215	INTEGER, PUBLIC, PARAMETER :: jpfillmpi = 5
216
217	!! * Substitutions
218	# include "do_loop_substitute.h90"
219	!!----------------------------------------------------------------------
220	!! NEMO/OCE 4.0 , NEMO Consortium (2018)
221	!! $Id$
222	!! Software governed by the CeCILL license (see ./LICENSE)
223	!!----------------------------------------------------------------------
224	CONTAINS
225
226	SUBROUTINE mpp_start( localComm )
227	!!----------------------------------------------------------------------
228	!! * routine mpp_start *
229	!!
230	!! ** Purpose : get mpi_comm_oce, mpprank and mppsize
231	!!----------------------------------------------------------------------
232	INTEGER , OPTIONAL , INTENT(in ) :: localComm !
233	!
234	INTEGER :: ierr
235	LOGICAL :: llmpi_init
236	!!----------------------------------------------------------------------
237	#if ! defined key_mpi_off
238	!
239	CALL mpi_initialized ( llmpi_init, ierr )
240	IF( ierr /= MPI_SUCCESS ) CALL ctl_stop( 'STOP', ' lib_mpp: Error in routine mpi_initialized' )
241
242	IF( .NOT. llmpi_init ) THEN
243	IF( PRESENT(localComm) ) THEN
244	WRITE(ctmp1,*) ' lib_mpp: You cannot provide a local communicator '
245	WRITE(ctmp2,*) ' without calling MPI_Init before ! '
246	CALL ctl_stop( 'STOP', ctmp1, ctmp2 )
247	ENDIF
248	CALL mpi_init( ierr )
249	IF( ierr /= MPI_SUCCESS ) CALL ctl_stop( 'STOP', ' lib_mpp: Error in routine mpi_init' )
250	ENDIF
251
252	IF( PRESENT(localComm) ) THEN
253	IF( Agrif_Root() ) THEN
254	mpi_comm_oce = localComm
255	ENDIF
256	ELSE
257	CALL mpi_comm_dup( mpi_comm_world, mpi_comm_oce, ierr)
258	IF( ierr /= MPI_SUCCESS ) CALL ctl_stop( 'STOP', ' lib_mpp: Error in routine mpi_comm_dup' )
259	ENDIF
260
261	# if defined key_agrif
262	IF( Agrif_Root() ) THEN
263	CALL Agrif_MPI_Init(mpi_comm_oce)
264	ELSE
265	CALL Agrif_MPI_set_grid_comm(mpi_comm_oce)
266	ENDIF
267	# endif
268
269	CALL mpi_comm_rank( mpi_comm_oce, mpprank, ierr )
270	CALL mpi_comm_size( mpi_comm_oce, mppsize, ierr )
271	!
272	CALL MPI_OP_CREATE(DDPDD_MPI, .TRUE., MPI_SUMDD, ierr)
273	!
274	#else
275	IF( PRESENT( localComm ) ) mpi_comm_oce = localComm
276	mppsize = 1
277	mpprank = 0
278	#endif
279	END SUBROUTINE mpp_start
280
281
282	SUBROUTINE mppsend( ktyp, pmess, kbytes, kdest, md_req )
283	!!----------------------------------------------------------------------
284	!! * routine mppsend *
285	!!
286	!! ** Purpose : Send messag passing array
287	!!
288	!!----------------------------------------------------------------------
289	REAL(wp), INTENT(inout) :: pmess(*) ! array of real
290	INTEGER , INTENT(in ) :: kbytes ! size of the array pmess
291	INTEGER , INTENT(in ) :: kdest ! receive process number
292	INTEGER , INTENT(in ) :: ktyp ! tag of the message
293	INTEGER , INTENT(inout) :: md_req ! argument for isend
294	!!
295	INTEGER :: iflag
296	INTEGER :: mpi_working_type
297	!!----------------------------------------------------------------------
298	!
299	#if ! defined key_mpi_off
300	IF (wp == dp) THEN
301	mpi_working_type = mpi_double_precision
302	ELSE
303	mpi_working_type = mpi_real
304	END IF
305	CALL mpi_isend( pmess, kbytes, mpi_working_type, kdest , ktyp, mpi_comm_oce, md_req, iflag )
306	#endif
307	!
308	END SUBROUTINE mppsend
309
310
311	SUBROUTINE mppsend_dp( ktyp, pmess, kbytes, kdest, md_req )
312	!!----------------------------------------------------------------------
313	!! * routine mppsend *
314	!!
315	!! ** Purpose : Send messag passing array
316	!!
317	!!----------------------------------------------------------------------
318	REAL(dp), INTENT(inout) :: pmess(*) ! array of real
319	INTEGER , INTENT(in ) :: kbytes ! size of the array pmess
320	INTEGER , INTENT(in ) :: kdest ! receive process number
321	INTEGER , INTENT(in ) :: ktyp ! tag of the message
322	INTEGER , INTENT(inout) :: md_req ! argument for isend
323	!!
324	INTEGER :: iflag
325	!!----------------------------------------------------------------------
326	!
327	#if ! defined key_mpi_off
328	CALL mpi_isend( pmess, kbytes, mpi_double_precision, kdest , ktyp, mpi_comm_oce, md_req, iflag )
329	#endif
330	!
331	END SUBROUTINE mppsend_dp
332
333
334	SUBROUTINE mppsend_sp( ktyp, pmess, kbytes, kdest, md_req )
335	!!----------------------------------------------------------------------
336	!! * routine mppsend *
337	!!
338	!! ** Purpose : Send messag passing array
339	!!
340	!!----------------------------------------------------------------------
341	REAL(sp), INTENT(inout) :: pmess(*) ! array of real
342	INTEGER , INTENT(in ) :: kbytes ! size of the array pmess
343	INTEGER , INTENT(in ) :: kdest ! receive process number
344	INTEGER , INTENT(in ) :: ktyp ! tag of the message
345	INTEGER , INTENT(inout) :: md_req ! argument for isend
346	!!
347	INTEGER :: iflag
348	!!----------------------------------------------------------------------
349	!
350	#if ! defined key_mpi_off
351	CALL mpi_isend( pmess, kbytes, mpi_real, kdest , ktyp, mpi_comm_oce, md_req, iflag )
352	#endif
353	!
354	END SUBROUTINE mppsend_sp
355
356
357	SUBROUTINE mpprecv( ktyp, pmess, kbytes, ksource )
358	!!----------------------------------------------------------------------
359	!! * routine mpprecv *
360	!!
361	!! ** Purpose : Receive messag passing array
362	!!
363	!!----------------------------------------------------------------------
364	REAL(wp), INTENT(inout) :: pmess(*) ! array of real
365	INTEGER , INTENT(in ) :: kbytes ! suze of the array pmess
366	INTEGER , INTENT(in ) :: ktyp ! Tag of the recevied message
367	INTEGER, OPTIONAL, INTENT(in) :: ksource ! source process number
368	!!
369	INTEGER :: istatus(mpi_status_size)
370	INTEGER :: iflag
371	INTEGER :: use_source
372	INTEGER :: mpi_working_type
373	!!----------------------------------------------------------------------
374	!
375	#if ! defined key_mpi_off
376	! If a specific process number has been passed to the receive call,
377	! use that one. Default is to use mpi_any_source
378	use_source = mpi_any_source
379	IF( PRESENT(ksource) ) use_source = ksource
380	!
381	IF (wp == dp) THEN
382	mpi_working_type = mpi_double_precision
383	ELSE
384	mpi_working_type = mpi_real
385	END IF
386	CALL mpi_recv( pmess, kbytes, mpi_working_type, use_source, ktyp, mpi_comm_oce, istatus, iflag )
387	#endif
388	!
389	END SUBROUTINE mpprecv
390
391	SUBROUTINE mpprecv_dp( ktyp, pmess, kbytes, ksource )
392	!!----------------------------------------------------------------------
393	!! * routine mpprecv *
394	!!
395	!! ** Purpose : Receive messag passing array
396	!!
397	!!----------------------------------------------------------------------
398	REAL(dp), INTENT(inout) :: pmess(*) ! array of real
399	INTEGER , INTENT(in ) :: kbytes ! suze of the array pmess
400	INTEGER , INTENT(in ) :: ktyp ! Tag of the recevied message
401	INTEGER, OPTIONAL, INTENT(in) :: ksource ! source process number
402	!!
403	INTEGER :: istatus(mpi_status_size)
404	INTEGER :: iflag
405	INTEGER :: use_source
406	!!----------------------------------------------------------------------
407	!
408	#if ! defined key_mpi_off
409	! If a specific process number has been passed to the receive call,
410	! use that one. Default is to use mpi_any_source
411	use_source = mpi_any_source
412	IF( PRESENT(ksource) ) use_source = ksource
413	!
414	CALL mpi_recv( pmess, kbytes, mpi_double_precision, use_source, ktyp, mpi_comm_oce, istatus, iflag )
415	#endif
416	!
417	END SUBROUTINE mpprecv_dp
418
419
420	SUBROUTINE mpprecv_sp( ktyp, pmess, kbytes, ksource )
421	!!----------------------------------------------------------------------
422	!! * routine mpprecv *
423	!!
424	!! ** Purpose : Receive messag passing array
425	!!
426	!!----------------------------------------------------------------------
427	REAL(sp), INTENT(inout) :: pmess(*) ! array of real
428	INTEGER , INTENT(in ) :: kbytes ! suze of the array pmess
429	INTEGER , INTENT(in ) :: ktyp ! Tag of the recevied message
430	INTEGER, OPTIONAL, INTENT(in) :: ksource ! source process number
431	!!
432	INTEGER :: istatus(mpi_status_size)
433	INTEGER :: iflag
434	INTEGER :: use_source
435	!!----------------------------------------------------------------------
436	!
437	#if ! defined key_mpi_off
438	! If a specific process number has been passed to the receive call,
439	! use that one. Default is to use mpi_any_source
440	use_source = mpi_any_source
441	IF( PRESENT(ksource) ) use_source = ksource
442	!
443	CALL mpi_recv( pmess, kbytes, mpi_real, use_source, ktyp, mpi_comm_oce, istatus, iflag )
444	#endif
445	!
446	END SUBROUTINE mpprecv_sp
447
448
449	SUBROUTINE mppgather( ptab, kp, pio )
450	!!----------------------------------------------------------------------
451	!! * routine mppgather *
452	!!
453	!! ** Purpose : Transfert between a local subdomain array and a work
454	!! array which is distributed following the vertical level.
455	!!
456	!!----------------------------------------------------------------------
457	REAL(wp), DIMENSION(jpi,jpj) , INTENT(in ) :: ptab ! subdomain input array
458	INTEGER , INTENT(in ) :: kp ! record length
459	REAL(wp), DIMENSION(jpi,jpj,jpnij), INTENT( out) :: pio ! subdomain input array
460	!!
461	INTEGER :: itaille, ierror ! temporary integer
462	!!---------------------------------------------------------------------
463	!
464	itaille = jpi * jpj
465	#if ! defined key_mpi_off
466	CALL mpi_gather( ptab, itaille, mpi_double_precision, pio, itaille , &
467	& mpi_double_precision, kp , mpi_comm_oce, ierror )
468	#else
469	pio(:,:,1) = ptab(:,:)
470	#endif
471	!
472	END SUBROUTINE mppgather
473
474
475	SUBROUTINE mppscatter( pio, kp, ptab )
476	!!----------------------------------------------------------------------
477	!! * routine mppscatter *
478	!!
479	!! ** Purpose : Transfert between awork array which is distributed
480	!! following the vertical level and the local subdomain array.
481	!!
482	!!----------------------------------------------------------------------
483	REAL(wp), DIMENSION(jpi,jpj,jpnij) :: pio ! output array
484	INTEGER :: kp ! Tag (not used with MPI
485	REAL(wp), DIMENSION(jpi,jpj) :: ptab ! subdomain array input
486	!!
487	INTEGER :: itaille, ierror ! temporary integer
488	!!---------------------------------------------------------------------
489	!
490	itaille = jpi * jpj
491	!
492	#if ! defined key_mpi_off
493	CALL mpi_scatter( pio, itaille, mpi_double_precision, ptab, itaille , &
494	& mpi_double_precision, kp , mpi_comm_oce, ierror )
495	#else
496	ptab(:,:) = pio(:,:,1)
497	#endif
498	!
499	END SUBROUTINE mppscatter
500
501
502	SUBROUTINE mpp_delay_sum( cdname, cdelay, y_in, pout, ldlast, kcom )
503	!!----------------------------------------------------------------------
504	!! * routine mpp_delay_sum *
505	!!
506	!! ** Purpose : performed delayed mpp_sum, the result is received on next call
507	!!
508	!!----------------------------------------------------------------------
509	CHARACTER(len=*), INTENT(in ) :: cdname ! name of the calling subroutine
510	CHARACTER(len=*), INTENT(in ) :: cdelay ! name (used as id) of the delayed operation
511	COMPLEX(dp), INTENT(in ), DIMENSION(:) :: y_in
512	REAL(wp), INTENT( out), DIMENSION(:) :: pout
513	LOGICAL, INTENT(in ) :: ldlast ! true if this is the last time we call this routine
514	INTEGER, INTENT(in ), OPTIONAL :: kcom
515	!!
516	INTEGER :: ji, isz
517	INTEGER :: idvar
518	INTEGER :: ierr, ilocalcomm
519	COMPLEX(dp), ALLOCATABLE, DIMENSION(:) :: ytmp
520	!!----------------------------------------------------------------------
521	#if ! defined key_mpi_off
522	ilocalcomm = mpi_comm_oce
523	IF( PRESENT(kcom) ) ilocalcomm = kcom
524
525	isz = SIZE(y_in)
526
527	IF( narea == 1 .AND. numcom == -1 ) CALL mpp_report( cdname, ld_dlg = .TRUE. )
528
529	idvar = -1
530	DO ji = 1, nbdelay
531	IF( TRIM(cdelay) == TRIM(c_delaylist(ji)) ) idvar = ji
532	END DO
533	IF ( idvar == -1 ) CALL ctl_stop( 'STOP',' mpp_delay_sum : please add a new delayed exchange for '//TRIM(cdname) )
534
535	IF ( ndelayid(idvar) == 0 ) THEN ! first call with restart: %z1d defined in iom_delay_rst
536	! --------------------------
537	IF ( SIZE(todelay(idvar)%z1d) /= isz ) THEN ! Check dimension coherence
538	IF(lwp) WRITE(numout,*) ' WARNING: the nb of delayed variables in restart file is not the model one'
539	DEALLOCATE(todelay(idvar)%z1d)
540	ndelayid(idvar) = -1 ! do as if we had no restart
541	ELSE
542	ALLOCATE(todelay(idvar)%y1d(isz))
543	todelay(idvar)%y1d(:) = CMPLX(todelay(idvar)%z1d(:), 0., wp) ! create %y1d, complex variable needed by mpi_sumdd
544	ndelayid(idvar) = MPI_REQUEST_NULL ! initialised request to a valid value
545	END IF
546	ENDIF
547
548	IF( ndelayid(idvar) == -1 ) THEN ! first call without restart: define %y1d and %z1d from y_in with blocking allreduce
549	! --------------------------
550	ALLOCATE(todelay(idvar)%z1d(isz), todelay(idvar)%y1d(isz)) ! allocate also %z1d as used for the restart
551	CALL mpi_allreduce( y_in(:), todelay(idvar)%y1d(:), isz, MPI_DOUBLE_COMPLEX, mpi_sumdd, ilocalcomm, ierr ) ! get %y1d
552	ndelayid(idvar) = MPI_REQUEST_NULL
553	ENDIF
554
555	CALL mpp_delay_rcv( idvar ) ! make sure %z1d is received
556
557	! send back pout from todelay(idvar)%z1d defined at previous call
558	pout(:) = todelay(idvar)%z1d(:)
559
560	! send y_in into todelay(idvar)%y1d with a non-blocking communication
561	# if defined key_mpi2
562	IF( ln_timing ) CALL tic_tac( .TRUE., ld_global = .TRUE.)
563	CALL mpi_allreduce( y_in(:), todelay(idvar)%y1d(:), isz, MPI_DOUBLE_COMPLEX, mpi_sumdd, ilocalcomm, ierr )
564	ndelayid(idvar) = MPI_REQUEST_NULL
565	IF( ln_timing ) CALL tic_tac(.FALSE., ld_global = .TRUE.)
566	# else
567	CALL mpi_iallreduce( y_in(:), todelay(idvar)%y1d(:), isz, MPI_DOUBLE_COMPLEX, mpi_sumdd, ilocalcomm, ndelayid(idvar), ierr )
568	# endif
569	#else
570	pout(:) = REAL(y_in(:), wp)
571	#endif
572
573	END SUBROUTINE mpp_delay_sum
574
575
576	SUBROUTINE mpp_delay_max( cdname, cdelay, p_in, pout, ldlast, kcom )
577	!!----------------------------------------------------------------------
578	!! * routine mpp_delay_max *
579	!!
580	!! ** Purpose : performed delayed mpp_max, the result is received on next call
581	!!
582	!!----------------------------------------------------------------------
583	CHARACTER(len=*), INTENT(in ) :: cdname ! name of the calling subroutine
584	CHARACTER(len=*), INTENT(in ) :: cdelay ! name (used as id) of the delayed operation
585	REAL(wp), INTENT(in ), DIMENSION(:) :: p_in !
586	REAL(wp), INTENT( out), DIMENSION(:) :: pout !
587	LOGICAL, INTENT(in ) :: ldlast ! true if this is the last time we call this routine
588	INTEGER, INTENT(in ), OPTIONAL :: kcom
589	!!
590	INTEGER :: ji, isz
591	INTEGER :: idvar
592	INTEGER :: ierr, ilocalcomm
593	INTEGER :: MPI_TYPE
594	!!----------------------------------------------------------------------
595
596	#if ! defined key_mpi_off
597	if( wp == dp ) then
598	MPI_TYPE = MPI_DOUBLE_PRECISION
599	else if ( wp == sp ) then
600	MPI_TYPE = MPI_REAL
601	else
602	CALL ctl_stop( "Error defining type, wp is neither dp nor sp" )
603
604	end if
605
606	ilocalcomm = mpi_comm_oce
607	IF( PRESENT(kcom) ) ilocalcomm = kcom
608
609	isz = SIZE(p_in)
610
611	IF( narea == 1 .AND. numcom == -1 ) CALL mpp_report( cdname, ld_dlg = .TRUE. )
612
613	idvar = -1
614	DO ji = 1, nbdelay
615	IF( TRIM(cdelay) == TRIM(c_delaylist(ji)) ) idvar = ji
616	END DO
617	IF ( idvar == -1 ) CALL ctl_stop( 'STOP',' mpp_delay_max : please add a new delayed exchange for '//TRIM(cdname) )
618
619	IF ( ndelayid(idvar) == 0 ) THEN ! first call with restart: %z1d defined in iom_delay_rst
620	! --------------------------
621	IF ( SIZE(todelay(idvar)%z1d) /= isz ) THEN ! Check dimension coherence
622	IF(lwp) WRITE(numout,*) ' WARNING: the nb of delayed variables in restart file is not the model one'
623	DEALLOCATE(todelay(idvar)%z1d)
624	ndelayid(idvar) = -1 ! do as if we had no restart
625	ELSE
626	ndelayid(idvar) = MPI_REQUEST_NULL
627	END IF
628	ENDIF
629
630	IF( ndelayid(idvar) == -1 ) THEN ! first call without restart: define %z1d from p_in with a blocking allreduce
631	! --------------------------
632	ALLOCATE(todelay(idvar)%z1d(isz))
633	CALL mpi_allreduce( p_in(:), todelay(idvar)%z1d(:), isz, MPI_DOUBLE_PRECISION, mpi_max, ilocalcomm, ierr ) ! get %z1d
634	ndelayid(idvar) = MPI_REQUEST_NULL
635	ENDIF
636
637	CALL mpp_delay_rcv( idvar ) ! make sure %z1d is received
638
639	! send back pout from todelay(idvar)%z1d defined at previous call
640	pout(:) = todelay(idvar)%z1d(:)
641
642	! send p_in into todelay(idvar)%z1d with a non-blocking communication
643	! (PM) Should we get rid of MPI2 option ? MPI3 was release in 2013. Who is still using MPI2 ?
644	# if defined key_mpi2
645	IF( ln_timing ) CALL tic_tac( .TRUE., ld_global = .TRUE.)
646	CALL mpi_allreduce( p_in(:), todelay(idvar)%z1d(:), isz, MPI_TYPE, mpi_max, ilocalcomm, ierr )
647	IF( ln_timing ) CALL tic_tac(.FALSE., ld_global = .TRUE.)
648	# else
649	CALL mpi_iallreduce( p_in(:), todelay(idvar)%z1d(:), isz, MPI_TYPE, mpi_max, ilocalcomm, ndelayid(idvar), ierr )
650	# endif
651	#else
652	pout(:) = p_in(:)
653	#endif
654
655	END SUBROUTINE mpp_delay_max
656
657
658	SUBROUTINE mpp_delay_rcv( kid )
659	!!----------------------------------------------------------------------
660	!! * routine mpp_delay_rcv *
661	!!
662	!! ** Purpose : force barrier for delayed mpp (needed for restart)
663	!!
664	!!----------------------------------------------------------------------
665	INTEGER,INTENT(in ) :: kid
666	INTEGER :: ierr
667	!!----------------------------------------------------------------------
668	#if ! defined key_mpi_off
669	IF( ln_timing ) CALL tic_tac( .TRUE., ld_global = .TRUE.)
670	! test on ndelayid(kid) useless as mpi_wait return immediatly if the request handle is MPI_REQUEST_NULL
671	CALL mpi_wait( ndelayid(kid), MPI_STATUS_IGNORE, ierr ) ! after this ndelayid(kid) = MPI_REQUEST_NULL
672	IF( ln_timing ) CALL tic_tac( .FALSE., ld_global = .TRUE.)
673	IF( ASSOCIATED(todelay(kid)%y1d) ) todelay(kid)%z1d(:) = REAL(todelay(kid)%y1d(:), wp) ! define %z1d from %y1d
674	#endif
675	END SUBROUTINE mpp_delay_rcv
676
677	SUBROUTINE mpp_bcast_nml( cdnambuff , kleng )
678	CHARACTER(LEN=:) , ALLOCATABLE, INTENT(INOUT) :: cdnambuff
679	INTEGER , INTENT(INOUT) :: kleng
680	!!----------------------------------------------------------------------
681	!! * routine mpp_bcast_nml *
682	!!
683	!! ** Purpose : broadcast namelist character buffer
684	!!
685	!!----------------------------------------------------------------------
686	!!
687	INTEGER :: iflag
688	!!----------------------------------------------------------------------
689	!
690	#if ! defined key_mpi_off
691	call MPI_BCAST(kleng, 1, MPI_INT, 0, mpi_comm_oce, iflag)
692	call MPI_BARRIER(mpi_comm_oce, iflag)
693	!$AGRIF_DO_NOT_TREAT
694	IF ( .NOT. ALLOCATED(cdnambuff) ) ALLOCATE( CHARACTER(LEN=kleng) :: cdnambuff )
695	!$AGRIF_END_DO_NOT_TREAT
696	call MPI_BCAST(cdnambuff, kleng, MPI_CHARACTER, 0, mpi_comm_oce, iflag)
697	call MPI_BARRIER(mpi_comm_oce, iflag)
698	#endif
699	!
700	END SUBROUTINE mpp_bcast_nml
701
702
703	!!----------------------------------------------------------------------
704	!! * mppmax_a_int, mppmax_int, mppmax_a_real, mppmax_real *
705	!!
706	!!----------------------------------------------------------------------
707	!!
708	# define OPERATION_MAX
709	# define INTEGER_TYPE
710	# define DIM_0d
711	# define ROUTINE_ALLREDUCE mppmax_int
712	# include "mpp_allreduce_generic.h90"
713	# undef ROUTINE_ALLREDUCE
714	# undef DIM_0d
715	# define DIM_1d
716	# define ROUTINE_ALLREDUCE mppmax_a_int
717	# include "mpp_allreduce_generic.h90"
718	# undef ROUTINE_ALLREDUCE
719	# undef DIM_1d
720	# undef INTEGER_TYPE
721	!
722	!!
723	!! ---- SINGLE PRECISION VERSIONS
724	!!
725	# define SINGLE_PRECISION
726	# define REAL_TYPE
727	# define DIM_0d
728	# define ROUTINE_ALLREDUCE mppmax_real_sp
729	# include "mpp_allreduce_generic.h90"
730	# undef ROUTINE_ALLREDUCE
731	# undef DIM_0d
732	# define DIM_1d
733	# define ROUTINE_ALLREDUCE mppmax_a_real_sp
734	# include "mpp_allreduce_generic.h90"
735	# undef ROUTINE_ALLREDUCE
736	# undef DIM_1d
737	# undef SINGLE_PRECISION
738	!!
739	!!
740	!! ---- DOUBLE PRECISION VERSIONS
741	!!
742	!
743	# define DIM_0d
744	# define ROUTINE_ALLREDUCE mppmax_real_dp
745	# include "mpp_allreduce_generic.h90"
746	# undef ROUTINE_ALLREDUCE
747	# undef DIM_0d
748	# define DIM_1d
749	# define ROUTINE_ALLREDUCE mppmax_a_real_dp
750	# include "mpp_allreduce_generic.h90"
751	# undef ROUTINE_ALLREDUCE
752	# undef DIM_1d
753	# undef REAL_TYPE
754	# undef OPERATION_MAX
755	!!----------------------------------------------------------------------
756	!! * mppmin_a_int, mppmin_int, mppmin_a_real, mppmin_real *
757	!!
758	!!----------------------------------------------------------------------
759	!!
760	# define OPERATION_MIN
761	# define INTEGER_TYPE
762	# define DIM_0d
763	# define ROUTINE_ALLREDUCE mppmin_int
764	# include "mpp_allreduce_generic.h90"
765	# undef ROUTINE_ALLREDUCE
766	# undef DIM_0d
767	# define DIM_1d
768	# define ROUTINE_ALLREDUCE mppmin_a_int
769	# include "mpp_allreduce_generic.h90"
770	# undef ROUTINE_ALLREDUCE
771	# undef DIM_1d
772	# undef INTEGER_TYPE
773	!
774	!!
775	!! ---- SINGLE PRECISION VERSIONS
776	!!
777	# define SINGLE_PRECISION
778	# define REAL_TYPE
779	# define DIM_0d
780	# define ROUTINE_ALLREDUCE mppmin_real_sp
781	# include "mpp_allreduce_generic.h90"
782	# undef ROUTINE_ALLREDUCE
783	# undef DIM_0d
784	# define DIM_1d
785	# define ROUTINE_ALLREDUCE mppmin_a_real_sp
786	# include "mpp_allreduce_generic.h90"
787	# undef ROUTINE_ALLREDUCE
788	# undef DIM_1d
789	# undef SINGLE_PRECISION
790	!!
791	!! ---- DOUBLE PRECISION VERSIONS
792	!!
793
794	# define DIM_0d
795	# define ROUTINE_ALLREDUCE mppmin_real_dp
796	# include "mpp_allreduce_generic.h90"
797	# undef ROUTINE_ALLREDUCE
798	# undef DIM_0d
799	# define DIM_1d
800	# define ROUTINE_ALLREDUCE mppmin_a_real_dp
801	# include "mpp_allreduce_generic.h90"
802	# undef ROUTINE_ALLREDUCE
803	# undef DIM_1d
804	# undef REAL_TYPE
805	# undef OPERATION_MIN
806
807	!!----------------------------------------------------------------------
808	!! * mppsum_a_int, mppsum_int, mppsum_a_real, mppsum_real *
809	!!
810	!! Global sum of 1D array or a variable (integer, real or complex)
811	!!----------------------------------------------------------------------
812	!!
813	# define OPERATION_SUM
814	# define INTEGER_TYPE
815	# define DIM_0d
816	# define ROUTINE_ALLREDUCE mppsum_int
817	# include "mpp_allreduce_generic.h90"
818	# undef ROUTINE_ALLREDUCE
819	# undef DIM_0d
820	# define DIM_1d
821	# define ROUTINE_ALLREDUCE mppsum_a_int
822	# include "mpp_allreduce_generic.h90"
823	# undef ROUTINE_ALLREDUCE
824	# undef DIM_1d
825	# undef INTEGER_TYPE
826
827	!!
828	!! ---- SINGLE PRECISION VERSIONS
829	!!
830	# define OPERATION_SUM
831	# define SINGLE_PRECISION
832	# define REAL_TYPE
833	# define DIM_0d
834	# define ROUTINE_ALLREDUCE mppsum_real_sp
835	# include "mpp_allreduce_generic.h90"
836	# undef ROUTINE_ALLREDUCE
837	# undef DIM_0d
838	# define DIM_1d
839	# define ROUTINE_ALLREDUCE mppsum_a_real_sp
840	# include "mpp_allreduce_generic.h90"
841	# undef ROUTINE_ALLREDUCE
842	# undef DIM_1d
843	# undef REAL_TYPE
844	# undef OPERATION_SUM
845
846	# undef SINGLE_PRECISION
847
848	!!
849	!! ---- DOUBLE PRECISION VERSIONS
850	!!
851	# define OPERATION_SUM
852	# define REAL_TYPE
853	# define DIM_0d
854	# define ROUTINE_ALLREDUCE mppsum_real_dp
855	# include "mpp_allreduce_generic.h90"
856	# undef ROUTINE_ALLREDUCE
857	# undef DIM_0d
858	# define DIM_1d
859	# define ROUTINE_ALLREDUCE mppsum_a_real_dp
860	# include "mpp_allreduce_generic.h90"
861	# undef ROUTINE_ALLREDUCE
862	# undef DIM_1d
863	# undef REAL_TYPE
864	# undef OPERATION_SUM
865
866	# define OPERATION_SUM_DD
867	# define COMPLEX_TYPE
868	# define DIM_0d
869	# define ROUTINE_ALLREDUCE mppsum_realdd
870	# include "mpp_allreduce_generic.h90"
871	# undef ROUTINE_ALLREDUCE
872	# undef DIM_0d
873	# define DIM_1d
874	# define ROUTINE_ALLREDUCE mppsum_a_realdd
875	# include "mpp_allreduce_generic.h90"
876	# undef ROUTINE_ALLREDUCE
877	# undef DIM_1d
878	# undef COMPLEX_TYPE
879	# undef OPERATION_SUM_DD
880
881	!!----------------------------------------------------------------------
882	!! *** mpp_minloc2d, mpp_minloc3d, mpp_maxloc2d, mpp_maxloc3d
883	!!
884	!!----------------------------------------------------------------------
885	!!
886	!!
887	!! ---- SINGLE PRECISION VERSIONS
888	!!
889	# define SINGLE_PRECISION
890	# define OPERATION_MINLOC
891	# define DIM_2d
892	# define ROUTINE_LOC mpp_minloc2d_sp
893	# include "mpp_loc_generic.h90"
894	# undef ROUTINE_LOC
895	# undef DIM_2d
896	# define DIM_3d
897	# define ROUTINE_LOC mpp_minloc3d_sp
898	# include "mpp_loc_generic.h90"
899	# undef ROUTINE_LOC
900	# undef DIM_3d
901	# undef OPERATION_MINLOC
902
903	# define OPERATION_MAXLOC
904	# define DIM_2d
905	# define ROUTINE_LOC mpp_maxloc2d_sp
906	# include "mpp_loc_generic.h90"
907	# undef ROUTINE_LOC
908	# undef DIM_2d
909	# define DIM_3d
910	# define ROUTINE_LOC mpp_maxloc3d_sp
911	# include "mpp_loc_generic.h90"
912	# undef ROUTINE_LOC
913	# undef DIM_3d
914	# undef OPERATION_MAXLOC
915	# undef SINGLE_PRECISION
916	!!
917	!! ---- DOUBLE PRECISION VERSIONS
918	!!
919	# define OPERATION_MINLOC
920	# define DIM_2d
921	# define ROUTINE_LOC mpp_minloc2d_dp
922	# include "mpp_loc_generic.h90"
923	# undef ROUTINE_LOC
924	# undef DIM_2d
925	# define DIM_3d
926	# define ROUTINE_LOC mpp_minloc3d_dp
927	# include "mpp_loc_generic.h90"
928	# undef ROUTINE_LOC
929	# undef DIM_3d
930	# undef OPERATION_MINLOC
931
932	# define OPERATION_MAXLOC
933	# define DIM_2d
934	# define ROUTINE_LOC mpp_maxloc2d_dp
935	# include "mpp_loc_generic.h90"
936	# undef ROUTINE_LOC
937	# undef DIM_2d
938	# define DIM_3d
939	# define ROUTINE_LOC mpp_maxloc3d_dp
940	# include "mpp_loc_generic.h90"
941	# undef ROUTINE_LOC
942	# undef DIM_3d
943	# undef OPERATION_MAXLOC
944
945
946	SUBROUTINE mppsync()
947	!!----------------------------------------------------------------------
948	!! * routine mppsync *
949	!!
950	!! ** Purpose : Massively parallel processors, synchroneous
951	!!
952	!!-----------------------------------------------------------------------
953	INTEGER :: ierror
954	!!-----------------------------------------------------------------------
955	!
956	#if ! defined key_mpi_off
957	CALL mpi_barrier( mpi_comm_oce, ierror )
958	#endif
959	!
960	END SUBROUTINE mppsync
961
962
963	SUBROUTINE mppstop( ld_abort )
964	!!----------------------------------------------------------------------
965	!! * routine mppstop *
966	!!
967	!! ** purpose : Stop massively parallel processors method
968	!!
969	!!----------------------------------------------------------------------
970	LOGICAL, OPTIONAL, INTENT(in) :: ld_abort ! source process number
971	LOGICAL :: ll_abort
972	INTEGER :: info, ierr
973	!!----------------------------------------------------------------------
974	ll_abort = .FALSE.
975	IF( PRESENT(ld_abort) ) ll_abort = ld_abort
976	!
977	#if ! defined key_mpi_off
978	IF(ll_abort) THEN
979	CALL mpi_abort( MPI_COMM_WORLD, 123, info )
980	ELSE
981	CALL mppsync
982	CALL mpi_finalize( info )
983	ENDIF
984	#endif
985	IF( ll_abort ) STOP 123
986	!
987	END SUBROUTINE mppstop
988
989
990	SUBROUTINE mpp_comm_free( kcom )
991	!!----------------------------------------------------------------------
992	INTEGER, INTENT(inout) :: kcom
993	!!
994	INTEGER :: ierr
995	!!----------------------------------------------------------------------
996	!
997	#if ! defined key_mpi_off
998	CALL MPI_COMM_FREE(kcom, ierr)
999	#endif
1000	!
1001	END SUBROUTINE mpp_comm_free
1002
1003
1004	SUBROUTINE mpp_ini_znl( kumout )
1005	!!----------------------------------------------------------------------
1006	!! * routine mpp_ini_znl *
1007	!!
1008	!! ** Purpose : Initialize special communicator for computing zonal sum
1009	!!
1010	!! ** Method : - Look for processors in the same row
1011	!! - Put their number in nrank_znl
1012	!! - Create group for the znl processors
1013	!! - Create a communicator for znl processors
1014	!! - Determine if processor should write znl files
1015	!!
1016	!! ** output
1017	!! ndim_rank_znl = number of processors on the same row
1018	!! ngrp_znl = group ID for the znl processors
1019	!! ncomm_znl = communicator for the ice procs.
1020	!! n_znl_root = number (in the world) of proc 0 in the ice comm.
1021	!!
1022	!!----------------------------------------------------------------------
1023	INTEGER, INTENT(in) :: kumout ! ocean.output logical units
1024	!
1025	INTEGER :: jproc ! dummy loop integer
1026	INTEGER :: ierr, ii ! local integer
1027	INTEGER, ALLOCATABLE, DIMENSION(:) :: kwork
1028	!!----------------------------------------------------------------------
1029	#if ! defined key_mpi_off
1030	!-$$ WRITE (numout,*) 'mpp_ini_znl ', mpprank, ' - ngrp_world : ', ngrp_world
1031	!-$$ WRITE (numout,*) 'mpp_ini_znl ', mpprank, ' - mpi_comm_world : ', mpi_comm_world
1032	!-$$ WRITE (numout,*) 'mpp_ini_znl ', mpprank, ' - mpi_comm_oce : ', mpi_comm_oce
1033	!
1034	ALLOCATE( kwork(jpnij), STAT=ierr )
1035	IF( ierr /= 0 ) CALL ctl_stop( 'STOP', 'mpp_ini_znl : failed to allocate 1D array of length jpnij')
1036
1037	IF( jpnj == 1 ) THEN
1038	ngrp_znl = ngrp_world
1039	ncomm_znl = mpi_comm_oce
1040	ELSE
1041	!
1042	CALL MPI_ALLGATHER ( njmpp, 1, mpi_integer, kwork, 1, mpi_integer, mpi_comm_oce, ierr )
1043	!-$$ WRITE (numout,*) 'mpp_ini_znl ', mpprank, ' - kwork pour njmpp : ', kwork
1044	!-$$ CALL flush(numout)
1045	!
1046	! Count number of processors on the same row
1047	ndim_rank_znl = 0
1048	DO jproc=1,jpnij
1049	IF ( kwork(jproc) == njmpp ) THEN
1050	ndim_rank_znl = ndim_rank_znl + 1
1051	ENDIF
1052	END DO
1053	!-$$ WRITE (numout,*) 'mpp_ini_znl ', mpprank, ' - ndim_rank_znl : ', ndim_rank_znl
1054	!-$$ CALL flush(numout)
1055	! Allocate the right size to nrank_znl
1056	IF (ALLOCATED (nrank_znl)) DEALLOCATE(nrank_znl)
1057	ALLOCATE(nrank_znl(ndim_rank_znl))
1058	ii = 0
1059	nrank_znl (:) = 0
1060	DO jproc=1,jpnij
1061	IF ( kwork(jproc) == njmpp) THEN
1062	ii = ii + 1
1063	nrank_znl(ii) = jproc -1
1064	ENDIF
1065	END DO
1066	!-$$ WRITE (numout,*) 'mpp_ini_znl ', mpprank, ' - nrank_znl : ', nrank_znl
1067	!-$$ CALL flush(numout)
1068
1069	! Create the opa group
1070	CALL MPI_COMM_GROUP(mpi_comm_oce,ngrp_opa,ierr)
1071	!-$$ WRITE (numout,*) 'mpp_ini_znl ', mpprank, ' - ngrp_opa : ', ngrp_opa
1072	!-$$ CALL flush(numout)
1073
1074	! Create the znl group from the opa group
1075	CALL MPI_GROUP_INCL ( ngrp_opa, ndim_rank_znl, nrank_znl, ngrp_znl, ierr )
1076	!-$$ WRITE (numout,*) 'mpp_ini_znl ', mpprank, ' - ngrp_znl ', ngrp_znl
1077	!-$$ CALL flush(numout)
1078
1079	! Create the znl communicator from the opa communicator, ie the pool of procs in the same row
1080	CALL MPI_COMM_CREATE ( mpi_comm_oce, ngrp_znl, ncomm_znl, ierr )
1081	!-$$ WRITE (numout,*) 'mpp_ini_znl ', mpprank, ' - ncomm_znl ', ncomm_znl
1082	!-$$ CALL flush(numout)
1083	!
1084	END IF
1085
1086	! Determines if processor if the first (starting from i=1) on the row
1087	IF ( jpni == 1 ) THEN
1088	l_znl_root = .TRUE.
1089	ELSE
1090	l_znl_root = .FALSE.
1091	kwork (1) = nimpp
1092	CALL mpp_min ( 'lib_mpp', kwork(1), kcom = ncomm_znl)
1093	IF ( nimpp == kwork(1)) l_znl_root = .TRUE.
1094	END IF
1095
1096	DEALLOCATE(kwork)
1097	#endif
1098
1099	END SUBROUTINE mpp_ini_znl
1100
1101
1102	SUBROUTINE mpp_ini_nc
1103	!!----------------------------------------------------------------------
1104	!! * routine mpp_ini_nc *
1105	!!
1106	!! ** Purpose : Initialize special communicators for MPI3 neighbourhood
1107	!! collectives
1108	!!
1109	!! ** Method : - Create graph communicators starting from the processes
1110	!! distribution along i and j directions
1111	!
1112	!! ** output
1113	!! mpi_nc_com4 = MPI3 neighbourhood collectives communicator
1114	!! mpi_nc_com8 = MPI3 neighbourhood collectives communicator (with diagonals)
1115	!!----------------------------------------------------------------------
1116	INTEGER, DIMENSION(:), ALLOCATABLE :: inei4, inei8
1117	INTEGER :: icnt4, icnt8
1118	INTEGER :: ierr
1119	LOGICAL, PARAMETER :: ireord = .FALSE.
1120	!!----------------------------------------------------------------------
1121	#if ! defined key_mpi_off && ! defined key_mpi2
1122
1123	icnt4 = COUNT( mpinei(1:4) >= 0 )
1124	icnt8 = COUNT( mpinei(1:8) >= 0 )
1125
1126	ALLOCATE( inei4(icnt4), inei8(icnt8) ) ! ok if icnt4 or icnt8 = 0
1127
1128	inei4 = PACK( mpinei(1:4), mask = mpinei(1:4) >= 0 )
1129	inei8 = PACK( mpinei(1:8), mask = mpinei(1:8) >= 0 )
1130
1131	CALL MPI_Dist_graph_create_adjacent(mpi_comm_oce, icnt4, inei4, MPI_UNWEIGHTED, &
1132	& icnt4, inei4, MPI_UNWEIGHTED, MPI_INFO_NULL, ireord, mpi_nc_com4, ierr)
1133	CALL MPI_Dist_graph_create_adjacent(mpi_comm_oce, icnt8, inei8, MPI_UNWEIGHTED, &
1134	& icnt8, inei8, MPI_UNWEIGHTED, MPI_INFO_NULL, ireord, mpi_nc_com8, ierr)
1135
1136	DEALLOCATE (inei4, inei8)
1137	#endif
1138	END SUBROUTINE mpp_ini_nc
1139
1140
1141	SUBROUTINE mpp_ini_north
1142	!!----------------------------------------------------------------------
1143	!! * routine mpp_ini_north *
1144	!!
1145	!! ** Purpose : Initialize special communicator for north folding
1146	!! condition together with global variables needed in the mpp folding
1147	!!
1148	!! ** Method : - Look for northern processors
1149	!! - Put their number in nrank_north
1150	!! - Create groups for the world processors and the north processors
1151	!! - Create a communicator for northern processors
1152	!!
1153	!! ** output
1154	!! ndim_rank_north = number of processors in the northern line
1155	!! nrank_north (ndim_rank_north) = number of the northern procs.
1156	!! ngrp_world = group ID for the world processors
1157	!! ngrp_north = group ID for the northern processors
1158	!! ncomm_north = communicator for the northern procs.
1159	!! north_root = number (in the world) of proc 0 in the northern comm.
1160	!!
1161	!!----------------------------------------------------------------------
1162	INTEGER :: ierr
1163	INTEGER :: jjproc
1164	INTEGER :: ii, ji
1165	!!----------------------------------------------------------------------
1166	!
1167	#if ! defined key_mpi_off
1168	!
1169	! Look for how many procs on the northern boundary
1170	ndim_rank_north = 0
1171	DO jjproc = 1, jpni
1172	IF( nfproc(jjproc) /= -1 ) ndim_rank_north = ndim_rank_north + 1
1173	END DO
1174	!
1175	! Allocate the right size to nrank_north
1176	IF (ALLOCATED (nrank_north)) DEALLOCATE(nrank_north)
1177	ALLOCATE( nrank_north(ndim_rank_north) )
1178
1179	! Fill the nrank_north array with proc. number of northern procs.
1180	! Note : the rank start at 0 in MPI
1181	ii = 0
1182	DO ji = 1, jpni
1183	IF ( nfproc(ji) /= -1 ) THEN
1184	ii=ii+1
1185	nrank_north(ii)=nfproc(ji)
1186	END IF
1187	END DO
1188	!
1189	! create the world group
1190	CALL MPI_COMM_GROUP( mpi_comm_oce, ngrp_world, ierr )
1191	!
1192	! Create the North group from the world group
1193	CALL MPI_GROUP_INCL( ngrp_world, ndim_rank_north, nrank_north, ngrp_north, ierr )
1194	!
1195	! Create the North communicator , ie the pool of procs in the north group
1196	CALL MPI_COMM_CREATE( mpi_comm_oce, ngrp_north, ncomm_north, ierr )
1197	!
1198	#endif
1199	END SUBROUTINE mpp_ini_north
1200
1201
1202	SUBROUTINE DDPDD_MPI( ydda, yddb, ilen, itype )
1203	!!---------------------------------------------------------------------
1204	!! Routine DDPDD_MPI: used by reduction operator MPI_SUMDD
1205	!!
1206	!! Modification of original codes written by David H. Bailey
1207	!! This subroutine computes yddb(i) = ydda(i)+yddb(i)
1208	!!---------------------------------------------------------------------
1209	INTEGER , INTENT(in) :: ilen, itype
1210	COMPLEX(dp), DIMENSION(ilen), INTENT(in) :: ydda
1211	COMPLEX(dp), DIMENSION(ilen), INTENT(inout) :: yddb
1212	!
1213	REAL(dp) :: zerr, zt1, zt2 ! local work variables
1214	INTEGER :: ji, ztmp ! local scalar
1215	!!---------------------------------------------------------------------
1216	!
1217	ztmp = itype ! avoid compilation warning
1218	!
1219	DO ji=1,ilen
1220	! Compute ydda + yddb using Knuth's trick.
1221	zt1 = real(ydda(ji)) + real(yddb(ji))
1222	zerr = zt1 - real(ydda(ji))
1223	zt2 = ((real(yddb(ji)) - zerr) + (real(ydda(ji)) - (zt1 - zerr))) &
1224	+ aimag(ydda(ji)) + aimag(yddb(ji))
1225
1226	! The result is zt1 + zt2, after normalization.
1227	yddb(ji) = cmplx ( zt1 + zt2, zt2 - ((zt1 + zt2) - zt1),wp )
1228	END DO
1229	!
1230	END SUBROUTINE DDPDD_MPI
1231
1232
1233	SUBROUTINE mpp_report( cdname, kpk, kpl, kpf, ld_lbc, ld_glb, ld_dlg )
1234	!!----------------------------------------------------------------------
1235	!! * routine mpp_report *
1236	!!
1237	!! ** Purpose : report use of mpp routines per time-setp
1238	!!
1239	!!----------------------------------------------------------------------
1240	CHARACTER(len=*), INTENT(in ) :: cdname ! name of the calling subroutine
1241	INTEGER , OPTIONAL, INTENT(in ) :: kpk, kpl, kpf
1242	LOGICAL , OPTIONAL, INTENT(in ) :: ld_lbc, ld_glb, ld_dlg
1243	!!
1244	CHARACTER(len=128) :: ccountname ! name of a subroutine to count communications
1245	LOGICAL :: ll_lbc, ll_glb, ll_dlg
1246	INTEGER :: ji, jj, jk, jh, jf, jcount ! dummy loop indices
1247	!!----------------------------------------------------------------------
1248	#if ! defined key_mpi_off
1249	!
1250	ll_lbc = .FALSE.
1251	IF( PRESENT(ld_lbc) ) ll_lbc = ld_lbc
1252	ll_glb = .FALSE.
1253	IF( PRESENT(ld_glb) ) ll_glb = ld_glb
1254	ll_dlg = .FALSE.
1255	IF( PRESENT(ld_dlg) ) ll_dlg = ld_dlg
1256	!
1257	! find the smallest common frequency: default = frequency product, if multiple, choose the larger of the 2 frequency
1258	ncom_freq = ncom_fsbc
1259	!
1260	IF ( ncom_stp == nit000+ncom_freq ) THEN ! avoid to count extra communications in potential initializations at nit000
1261	IF( ll_lbc ) THEN
1262	IF( .NOT. ALLOCATED(ncomm_sequence) ) ALLOCATE( ncomm_sequence(ncom_rec_max,2) )
1263	IF( .NOT. ALLOCATED( crname_lbc) ) ALLOCATE( crname_lbc(ncom_rec_max ) )
1264	n_sequence_lbc = n_sequence_lbc + 1
1265	IF( n_sequence_lbc > ncom_rec_max ) CALL ctl_stop( 'STOP', 'lib_mpp, increase ncom_rec_max' ) ! deadlock
1266	crname_lbc(n_sequence_lbc) = cdname ! keep the name of the calling routine
1267	ncomm_sequence(n_sequence_lbc,1) = kpk*kpl ! size of 3rd and 4th dimensions
1268	ncomm_sequence(n_sequence_lbc,2) = kpf ! number of arrays to be treated (multi)
1269	ENDIF
1270	IF( ll_glb ) THEN
1271	IF( .NOT. ALLOCATED(crname_glb) ) ALLOCATE( crname_glb(ncom_rec_max) )
1272	n_sequence_glb = n_sequence_glb + 1
1273	IF( n_sequence_glb > ncom_rec_max ) CALL ctl_stop( 'STOP', 'lib_mpp, increase ncom_rec_max' ) ! deadlock
1274	crname_glb(n_sequence_glb) = cdname ! keep the name of the calling routine
1275	ENDIF
1276	IF( ll_dlg ) THEN
1277	IF( .NOT. ALLOCATED(crname_dlg) ) ALLOCATE( crname_dlg(ncom_rec_max) )
1278	n_sequence_dlg = n_sequence_dlg + 1
1279	IF( n_sequence_dlg > ncom_rec_max ) CALL ctl_stop( 'STOP', 'lib_mpp, increase ncom_rec_max' ) ! deadlock
1280	crname_dlg(n_sequence_dlg) = cdname ! keep the name of the calling routine
1281	ENDIF
1282	ELSE IF ( ncom_stp == nit000+2*ncom_freq ) THEN
1283	CALL ctl_opn( numcom, 'communication_report.txt', 'REPLACE', 'FORMATTED', 'SEQUENTIAL', -1, numout, .FALSE., narea )
1284	WRITE(numcom,*) ' '
1285	WRITE(numcom,*) ' ------------------------------------------------------------'
1286	WRITE(numcom,*) ' Communication pattern report (second oce+sbc+top time step):'
1287	WRITE(numcom,*) ' ------------------------------------------------------------'
1288	WRITE(numcom,*) ' '
1289	WRITE(numcom,'(A,I4)') ' Exchanged halos : ', n_sequence_lbc
1290	jj = 0; jk = 0; jf = 0; jh = 0
1291	DO ji = 1, n_sequence_lbc
1292	IF ( ncomm_sequence(ji,1) .GT. 1 ) jk = jk + 1
1293	IF ( ncomm_sequence(ji,2) .GT. 1 ) jf = jf + 1
1294	IF ( ncomm_sequence(ji,1) .GT. 1 .AND. ncomm_sequence(ji,2) .GT. 1 ) jj = jj + 1
1295	jh = MAX (jh, ncomm_sequence(ji,1)*ncomm_sequence(ji,2))
1296	END DO
1297	WRITE(numcom,'(A,I3)') ' 3D Exchanged halos : ', jk
1298	WRITE(numcom,'(A,I3)') ' Multi arrays exchanged halos : ', jf
1299	WRITE(numcom,'(A,I3)') ' from which 3D : ', jj
1300	WRITE(numcom,'(A,I10)') ' Array max size : ', jhjpijpj
1301	WRITE(numcom,*) ' '
1302	WRITE(numcom,*) ' lbc_lnk called'
1303	DO ji = 1, n_sequence_lbc - 1
1304	IF ( crname_lbc(ji) /= 'already counted' ) THEN
1305	ccountname = crname_lbc(ji)
1306	crname_lbc(ji) = 'already counted'
1307	jcount = 1
1308	DO jj = ji + 1, n_sequence_lbc
1309	IF ( ccountname == crname_lbc(jj) ) THEN
1310	jcount = jcount + 1
1311	crname_lbc(jj) = 'already counted'
1312	END IF
1313	END DO
1314	WRITE(numcom,'(A, I4, A, A)') ' - ', jcount,' times by subroutine ', TRIM(ccountname)
1315	END IF
1316	END DO
1317	IF ( crname_lbc(n_sequence_lbc) /= 'already counted' ) THEN
1318	WRITE(numcom,'(A, I4, A, A)') ' - ', 1,' times by subroutine ', TRIM(crname_lbc(ncom_rec_max))
1319	END IF
1320	WRITE(numcom,*) ' '
1321	IF ( n_sequence_glb > 0 ) THEN
1322	WRITE(numcom,'(A,I4)') ' Global communications : ', n_sequence_glb
1323	jj = 1
1324	DO ji = 2, n_sequence_glb
1325	IF( crname_glb(ji-1) /= crname_glb(ji) ) THEN
1326	WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_glb(ji-1))
1327	jj = 0
1328	END IF
1329	jj = jj + 1
1330	END DO
1331	WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_glb(n_sequence_glb))
1332	DEALLOCATE(crname_glb)
1333	ELSE
1334	WRITE(numcom,*) ' No MPI global communication '
1335	ENDIF
1336	WRITE(numcom,*) ' '
1337	IF ( n_sequence_dlg > 0 ) THEN
1338	WRITE(numcom,'(A,I4)') ' Delayed global communications : ', n_sequence_dlg
1339	jj = 1
1340	DO ji = 2, n_sequence_dlg
1341	IF( crname_dlg(ji-1) /= crname_dlg(ji) ) THEN
1342	WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_dlg(ji-1))
1343	jj = 0
1344	END IF
1345	jj = jj + 1
1346	END DO
1347	WRITE(numcom,'(A, I4, A, A)') ' - ', jj,' times by subroutine ', TRIM(crname_dlg(n_sequence_dlg))
1348	DEALLOCATE(crname_dlg)
1349	ELSE
1350	WRITE(numcom,*) ' No MPI delayed global communication '
1351	ENDIF
1352	WRITE(numcom,*) ' '
1353	WRITE(numcom,*) ' -----------------------------------------------'
1354	WRITE(numcom,*) ' '
1355	DEALLOCATE(ncomm_sequence)
1356	DEALLOCATE(crname_lbc)
1357	ENDIF
1358	#endif
1359	END SUBROUTINE mpp_report
1360
1361
1362	SUBROUTINE tic_tac (ld_tic, ld_global)
1363
1364	LOGICAL, INTENT(IN) :: ld_tic
1365	LOGICAL, OPTIONAL, INTENT(IN) :: ld_global
1366	REAL(dp), DIMENSION(2), SAVE :: tic_wt
1367	REAL(dp), SAVE :: tic_ct = 0._dp
1368	INTEGER :: ii
1369	#if ! defined key_mpi_off
1370
1371	IF( ncom_stp <= nit000 ) RETURN
1372	IF( ncom_stp == nitend ) RETURN
1373	ii = 1
1374	IF( PRESENT( ld_global ) ) THEN
1375	IF( ld_global ) ii = 2
1376	END IF
1377
1378	IF ( ld_tic ) THEN
1379	tic_wt(ii) = MPI_Wtime() ! start count tic->tac (waiting time)
1380	IF ( tic_ct > 0.0_dp ) compute_time = compute_time + MPI_Wtime() - tic_ct ! cumulate count tac->tic
1381	ELSE
1382	waiting_time(ii) = waiting_time(ii) + MPI_Wtime() - tic_wt(ii) ! cumulate count tic->tac
1383	tic_ct = MPI_Wtime() ! start count tac->tic (waiting time)
1384	ENDIF
1385	#endif
1386
1387	END SUBROUTINE tic_tac
1388
1389	#if defined key_mpi_off
1390	SUBROUTINE mpi_wait(request, status, ierror)
1391	INTEGER , INTENT(in ) :: request
1392	INTEGER, DIMENSION(MPI_STATUS_SIZE), INTENT( out) :: status
1393	INTEGER , INTENT( out) :: ierror
1394	END SUBROUTINE mpi_wait
1395
1396
1397	FUNCTION MPI_Wtime()
1398	REAL(wp) :: MPI_Wtime
1399	MPI_Wtime = -1.
1400	END FUNCTION MPI_Wtime
1401	#endif
1402
1403	!!----------------------------------------------------------------------
1404	!! ctl_stop, ctl_warn, get_unit, ctl_opn, ctl_nam, load_nml routines
1405	!!----------------------------------------------------------------------
1406
1407	SUBROUTINE ctl_stop( cd1, cd2, cd3, cd4, cd5 , &
1408	& cd6, cd7, cd8, cd9, cd10 )
1409	!!----------------------------------------------------------------------
1410	!! * ROUTINE stop_opa *
1411	!!
1412	!! ** Purpose : print in ocean.outpput file a error message and
1413	!! increment the error number (nstop) by one.
1414	!!----------------------------------------------------------------------
1415	CHARACTER(len=*), INTENT(in ) :: cd1
1416	CHARACTER(len=*), INTENT(in ), OPTIONAL :: cd2, cd3, cd4, cd5
1417	CHARACTER(len=*), INTENT(in ), OPTIONAL :: cd6, cd7, cd8, cd9, cd10
1418	!
1419	CHARACTER(LEN=8) :: clfmt ! writing format
1420	INTEGER :: inum
1421	!!----------------------------------------------------------------------
1422	!
1423	nstop = nstop + 1
1424	!
1425	IF( cd1 == 'STOP' .AND. narea /= 1 ) THEN ! Immediate stop: add an arror message in 'ocean.output' file
1426	CALL ctl_opn( inum, 'ocean.output', 'APPEND', 'FORMATTED', 'SEQUENTIAL', -1, 6, .FALSE. )
1427	WRITE(inum,*)
1428	WRITE(inum,) ' ==>>> Look for "E R R O R" messages in all existing ocean.output* files'
1429	CLOSE(inum)
1430	ENDIF
1431	IF( numout == 6 ) THEN ! force to open ocean.output file if not already opened
1432	CALL ctl_opn( numout, 'ocean.output', 'REPLACE', 'FORMATTED', 'SEQUENTIAL', -1, -1, .FALSE., narea )
1433	ENDIF
1434	!
1435	WRITE(numout,*)
1436	WRITE(numout,*) ' ===>>> : E R R O R'
1437	WRITE(numout,*)
1438	WRITE(numout,*) ' ==========='
1439	WRITE(numout,*)
1440	WRITE(numout,*) TRIM(cd1)
1441	IF( PRESENT(cd2 ) ) WRITE(numout,*) TRIM(cd2)
1442	IF( PRESENT(cd3 ) ) WRITE(numout,*) TRIM(cd3)
1443	IF( PRESENT(cd4 ) ) WRITE(numout,*) TRIM(cd4)
1444	IF( PRESENT(cd5 ) ) WRITE(numout,*) TRIM(cd5)
1445	IF( PRESENT(cd6 ) ) WRITE(numout,*) TRIM(cd6)
1446	IF( PRESENT(cd7 ) ) WRITE(numout,*) TRIM(cd7)
1447	IF( PRESENT(cd8 ) ) WRITE(numout,*) TRIM(cd8)
1448	IF( PRESENT(cd9 ) ) WRITE(numout,*) TRIM(cd9)
1449	IF( PRESENT(cd10) ) WRITE(numout,*) TRIM(cd10)
1450	WRITE(numout,*)
1451	!
1452	CALL FLUSH(numout )
1453	IF( numstp /= -1 ) CALL FLUSH(numstp )
1454	IF( numrun /= -1 ) CALL FLUSH(numrun )
1455	IF( numevo_ice /= -1 ) CALL FLUSH(numevo_ice)
1456	!
1457	IF( cd1 == 'STOP' ) THEN
1458	WRITE(numout,*)
1459	WRITE(numout,*) 'huge E-R-R-O-R : immediate stop'
1460	WRITE(numout,*)
1461	CALL FLUSH(numout)
1462	CALL SLEEP(60) ! make sure that all output and abort files are written by all cores. 60s should be enough...
1463	CALL mppstop( ld_abort = .true. )
1464	ENDIF
1465	!
1466	END SUBROUTINE ctl_stop
1467
1468
1469	SUBROUTINE ctl_warn( cd1, cd2, cd3, cd4, cd5, &
1470	& cd6, cd7, cd8, cd9, cd10 )
1471	!!----------------------------------------------------------------------
1472	!! * ROUTINE stop_warn *
1473	!!
1474	!! ** Purpose : print in ocean.outpput file a error message and
1475	!! increment the warning number (nwarn) by one.
1476	!!----------------------------------------------------------------------
1477	CHARACTER(len=*), INTENT(in), OPTIONAL :: cd1, cd2, cd3, cd4, cd5
1478	CHARACTER(len=*), INTENT(in), OPTIONAL :: cd6, cd7, cd8, cd9, cd10
1479	!!----------------------------------------------------------------------
1480	!
1481	nwarn = nwarn + 1
1482	!
1483	IF(lwp) THEN
1484	WRITE(numout,*)
1485	WRITE(numout,*) ' ===>>> : W A R N I N G'
1486	WRITE(numout,*)
1487	WRITE(numout,*) ' ==============='
1488	WRITE(numout,*)
1489	IF( PRESENT(cd1 ) ) WRITE(numout,*) TRIM(cd1)
1490	IF( PRESENT(cd2 ) ) WRITE(numout,*) TRIM(cd2)
1491	IF( PRESENT(cd3 ) ) WRITE(numout,*) TRIM(cd3)
1492	IF( PRESENT(cd4 ) ) WRITE(numout,*) TRIM(cd4)
1493	IF( PRESENT(cd5 ) ) WRITE(numout,*) TRIM(cd5)
1494	IF( PRESENT(cd6 ) ) WRITE(numout,*) TRIM(cd6)
1495	IF( PRESENT(cd7 ) ) WRITE(numout,*) TRIM(cd7)
1496	IF( PRESENT(cd8 ) ) WRITE(numout,*) TRIM(cd8)
1497	IF( PRESENT(cd9 ) ) WRITE(numout,*) TRIM(cd9)
1498	IF( PRESENT(cd10) ) WRITE(numout,*) TRIM(cd10)
1499	WRITE(numout,*)
1500	ENDIF
1501	CALL FLUSH(numout)
1502	!
1503	END SUBROUTINE ctl_warn
1504
1505
1506	SUBROUTINE ctl_opn( knum, cdfile, cdstat, cdform, cdacce, klengh, kout, ldwp, karea )
1507	!!----------------------------------------------------------------------
1508	!! * ROUTINE ctl_opn *
1509	!!
1510	!! ** Purpose : Open file and check if required file is available.
1511	!!
1512	!! ** Method : Fortan open
1513	!!----------------------------------------------------------------------
1514	INTEGER , INTENT( out) :: knum ! logical unit to open
1515	CHARACTER(len=*) , INTENT(in ) :: cdfile ! file name to open
1516	CHARACTER(len=*) , INTENT(in ) :: cdstat ! disposition specifier
1517	CHARACTER(len=*) , INTENT(in ) :: cdform ! formatting specifier
1518	CHARACTER(len=*) , INTENT(in ) :: cdacce ! access specifier
1519	INTEGER , INTENT(in ) :: klengh ! record length
1520	INTEGER , INTENT(in ) :: kout ! number of logical units for write
1521	LOGICAL , INTENT(in ) :: ldwp ! boolean term for print
1522	INTEGER, OPTIONAL, INTENT(in ) :: karea ! proc number
1523	!
1524	CHARACTER(len=80) :: clfile
1525	CHARACTER(LEN=10) :: clfmt ! writing format
1526	INTEGER :: iost
1527	INTEGER :: idg ! number of digits
1528	!!----------------------------------------------------------------------
1529	!
1530	! adapt filename
1531	! ----------------
1532	clfile = TRIM(cdfile)
1533	IF( PRESENT( karea ) ) THEN
1534	IF( karea > 1 ) THEN
1535	! Warning: jpnij is maybe not already defined when calling ctl_opn -> use mppsize instead of jpnij
1536	idg = MAX( INT(LOG10(REAL(MAX(1,mppsize-1),wp))) + 1, 4 ) ! how many digits to we need to write? min=4, max=9
1537	WRITE(clfmt, "('(a,a,i', i1, '.', i1, ')')") idg, idg ! '(a,a,ix.x)'
1538	WRITE(clfile, clfmt) TRIM(clfile), '_', karea-1
1539	ENDIF
1540	ENDIF
1541	#if defined key_agrif
1542	IF( .NOT. Agrif_Root() ) clfile = TRIM(Agrif_CFixed())//'_'//TRIM(clfile)
1543	knum=Agrif_Get_Unit()
1544	#else
1545	knum=get_unit()
1546	#endif
1547	IF( TRIM(cdfile) == '/dev/null' ) clfile = TRIM(cdfile) ! force the use of /dev/null
1548	!
1549	IF( cdacce(1:6) == 'DIRECT' ) THEN ! cdacce has always more than 6 characters
1550	OPEN( UNIT=knum, FILE=clfile, FORM=cdform, ACCESS=cdacce, STATUS=cdstat, RECL=klengh , ERR=100, IOSTAT=iost )
1551	ELSE IF( TRIM(cdstat) == 'APPEND' ) THEN ! cdstat can have less than 6 characters
1552	OPEN( UNIT=knum, FILE=clfile, FORM=cdform, ACCESS=cdacce, STATUS='UNKNOWN', POSITION='APPEND', ERR=100, IOSTAT=iost )
1553	ELSE
1554	OPEN( UNIT=knum, FILE=clfile, FORM=cdform, ACCESS=cdacce, STATUS=cdstat , ERR=100, IOSTAT=iost )
1555	ENDIF
1556	IF( iost /= 0 .AND. TRIM(clfile) == '/dev/null' ) & ! for windows
1557	& OPEN(UNIT=knum,FILE='NUL', FORM=cdform, ACCESS=cdacce, STATUS=cdstat , ERR=100, IOSTAT=iost )
1558	IF( iost == 0 ) THEN
1559	IF(ldwp .AND. kout > 0) THEN
1560	WRITE(kout,*) ' file : ', TRIM(clfile),' open ok'
1561	WRITE(kout,*) ' unit = ', knum
1562	WRITE(kout,*) ' status = ', cdstat
1563	WRITE(kout,*) ' form = ', cdform
1564	WRITE(kout,*) ' access = ', cdacce
1565	WRITE(kout,*)
1566	ENDIF
1567	ENDIF
1568	100 CONTINUE
1569	IF( iost /= 0 ) THEN
1570	WRITE(ctmp1,*) ' ===>>>> : bad opening file: ', TRIM(clfile)
1571	WRITE(ctmp2,*) ' ======= === '
1572	WRITE(ctmp3,*) ' unit = ', knum
1573	WRITE(ctmp4,*) ' status = ', cdstat
1574	WRITE(ctmp5,*) ' form = ', cdform
1575	WRITE(ctmp6,*) ' access = ', cdacce
1576	WRITE(ctmp7,*) ' iostat = ', iost
1577	WRITE(ctmp8,*) ' we stop. verify the file '
1578	CALL ctl_stop( 'STOP', ctmp1, ctmp2, ctmp3, ctmp4, ctmp5, ctmp6, ctmp7, ctmp8 )
1579	ENDIF
1580	!
1581	END SUBROUTINE ctl_opn
1582
1583
1584	SUBROUTINE ctl_nam ( kios, cdnam )
1585	!!----------------------------------------------------------------------
1586	!! * ROUTINE ctl_nam *
1587	!!
1588	!! ** Purpose : Informations when error while reading a namelist
1589	!!
1590	!! ** Method : Fortan open
1591	!!----------------------------------------------------------------------
1592	INTEGER , INTENT(inout) :: kios ! IO status after reading the namelist
1593	CHARACTER(len=*) , INTENT(in ) :: cdnam ! group name of namelist for which error occurs
1594	!
1595	CHARACTER(len=5) :: clios ! string to convert iostat in character for print
1596	!!----------------------------------------------------------------------
1597	!
1598	WRITE (clios, '(I5.0)') kios
1599	IF( kios < 0 ) THEN
1600	CALL ctl_warn( 'end of record or file while reading namelist ' &
1601	& // TRIM(cdnam) // ' iostat = ' // TRIM(clios) )
1602	ENDIF
1603	!
1604	IF( kios > 0 ) THEN
1605	CALL ctl_stop( 'misspelled variable in namelist ' &
1606	& // TRIM(cdnam) // ' iostat = ' // TRIM(clios) )
1607	ENDIF
1608	kios = 0
1609	!
1610	END SUBROUTINE ctl_nam
1611
1612
1613	INTEGER FUNCTION get_unit()
1614	!!----------------------------------------------------------------------
1615	!! * FUNCTION get_unit *
1616	!!
1617	!! ** Purpose : return the index of an unused logical unit
1618	!!----------------------------------------------------------------------
1619	LOGICAL :: llopn
1620	!!----------------------------------------------------------------------
1621	!
1622	get_unit = 15 ! choose a unit that is big enough then it is not already used in NEMO
1623	llopn = .TRUE.
1624	DO WHILE( (get_unit < 998) .AND. llopn )
1625	get_unit = get_unit + 1
1626	INQUIRE( unit = get_unit, opened = llopn )
1627	END DO
1628	IF( (get_unit == 999) .AND. llopn ) THEN
1629	CALL ctl_stop( 'STOP', 'get_unit: All logical units until 999 are used...' )
1630	ENDIF
1631	!
1632	END FUNCTION get_unit
1633
1634	SUBROUTINE load_nml( cdnambuff , cdnamfile, kout, ldwp)
1635	CHARACTER(LEN=:) , ALLOCATABLE, INTENT(INOUT) :: cdnambuff
1636	CHARACTER(LEN=*), INTENT(IN ) :: cdnamfile
1637	CHARACTER(LEN=256) :: chline
1638	CHARACTER(LEN=1) :: csp
1639	INTEGER, INTENT(IN) :: kout
1640	LOGICAL, INTENT(IN) :: ldwp !: .true. only for the root broadcaster
1641	INTEGER :: itot, iun, iltc, inl, ios, itotsav
1642	!
1643	!csp = NEW_LINE('A')
1644	! a new line character is the best seperator but some systems (e.g.Cray)
1645	! seem to terminate namelist reads from internal files early if they
1646	! encounter new-lines. Use a single space for safety.
1647	csp = ' '
1648	!
1649	! Check if the namelist buffer has already been allocated. Return if it has.
1650	!
1651	IF ( ALLOCATED( cdnambuff ) ) RETURN
1652	IF( ldwp ) THEN
1653	!
1654	! Open namelist file
1655	!
1656	CALL ctl_opn( iun, cdnamfile, 'OLD', 'FORMATTED', 'SEQUENTIAL', -1, kout, ldwp )
1657	!
1658	! First pass: count characters excluding comments and trimable white space
1659	!
1660	itot=0
1661	10 READ(iun,'(A256)',END=20,ERR=20) chline
1662	iltc = LEN_TRIM(chline)
1663	IF ( iltc.GT.0 ) THEN
1664	inl = INDEX(chline, '!')
1665	IF( inl.eq.0 ) THEN
1666	itot = itot + iltc + 1 ! +1 for the newline character
1667	ELSEIF( inl.GT.0 .AND. LEN_TRIM( chline(1:inl-1) ).GT.0 ) THEN
1668	itot = itot + inl ! includes +1 for the newline character
1669	ENDIF
1670	ENDIF
1671	GOTO 10
1672	20 CONTINUE
1673	!
1674	! Allocate text cdnambuff for condensed namelist
1675	!
1676	!$AGRIF_DO_NOT_TREAT
1677	ALLOCATE( CHARACTER(LEN=itot) :: cdnambuff )
1678	!$AGRIF_END_DO_NOT_TREAT
1679	itotsav = itot
1680	!
1681	! Second pass: read and transfer pruned characters into cdnambuff
1682	!
1683	REWIND(iun)
1684	itot=1
1685	30 READ(iun,'(A256)',END=40,ERR=40) chline
1686	iltc = LEN_TRIM(chline)
1687	IF ( iltc.GT.0 ) THEN
1688	inl = INDEX(chline, '!')
1689	IF( inl.eq.0 ) THEN
1690	inl = iltc
1691	ELSE
1692	inl = inl - 1
1693	ENDIF
1694	IF( inl.GT.0 .AND. LEN_TRIM( chline(1:inl) ).GT.0 ) THEN
1695	cdnambuff(itot:itot+inl-1) = chline(1:inl)
1696	WRITE( cdnambuff(itot+inl:itot+inl), '(a)' ) csp
1697	itot = itot + inl + 1
1698	ENDIF
1699	ENDIF
1700	GOTO 30
1701	40 CONTINUE
1702	itot = itot - 1
1703	IF( itotsav .NE. itot ) WRITE(,) 'WARNING in load_nml. Allocated ',itotsav,' for read buffer; but used ',itot
1704	!
1705	! Close namelist file
1706	!
1707	CLOSE(iun)
1708	!write(*,'(32A)') cdnambuff
1709	ENDIF
1710	#if ! defined key_mpi_off
1711	CALL mpp_bcast_nml( cdnambuff, itot )
1712	#endif
1713	END SUBROUTINE load_nml
1714
1715
1716	!!----------------------------------------------------------------------
1717	END MODULE lib_mpp

Note: See TracBrowser for help on using the repository browser.

Download in other formats: