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/trunk/src/OCE/LBC – NEMO

Context Navigation

source: NEMO/trunk/src/OCE/LBC/lib_mpp.F90 @ 14275

Last change on this file since 14275 was 14275, checked in by smasson, 3 years ago
trunk: suppress nproc ( = mpprank = narea-1)
Property svn:keywords set to `Id`
File size: 73.9 KB

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

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

Download in other formats: