read_res1_des_mod.f

Go to the documentation of this file.

      MODULE read_res1_des

      use cdist, only: bdist_io
      use compar, only: pe_io
      use compar, only: mype
      use des_allocate
      use desmpi
      use error_manager
      use mpi_comm_des, only: desmpi_gatherv
      use mpi_comm_des, only: desmpi_scatterv

      IMPLICIT NONE

      PRIVATE

      PUBLIC :: init_read_res_des
      PUBLIC :: finl_read_res_des

      PUBLIC :: read_par_pos
      PUBLIC :: read_par_col

      PUBLIC :: read_res_des
      PUBLIC :: read_res_parray
      PUBLIC :: read_res_carray

      INTERFACE read_res_des
         MODULE PROCEDURE read_res_des_0i
         MODULE PROCEDURE read_res_des_1i
         MODULE PROCEDURE read_res_des_0d
         MODULE PROCEDURE read_res_des_1d
         MODULE PROCEDURE read_res_des_0l
         MODULE PROCEDURE read_res_des_1l
      END INTERFACE

      INTERFACE read_res_parray
         MODULE PROCEDURE read_res_parray_1b
         MODULE PROCEDURE read_res_parray_1i
         MODULE PROCEDURE read_res_parray_1d
         MODULE PROCEDURE read_res_parray_1l
      END INTERFACE

      INTERFACE read_res_carray
         MODULE PROCEDURE read_res_carray_1i
         MODULE PROCEDURE read_res_carray_1d
         MODULE PROCEDURE read_res_carray_1l
      END INTERFACE


      INTEGER, PARAMETER :: rdes_unit = 901

      INTEGER :: pin_count
      INTEGER :: cin_count

! Send/Recv parameters for Particle arrays:
      INTEGER :: prootcnt, pproccnt
      INTEGER :: precv
      INTEGER, allocatable :: pscatter(:)
      INTEGER, allocatable :: pdispls(:)

! Variables used for reading restart file
      INTEGER, ALLOCATABLE :: prestartmap(:)
      INTEGER, ALLOCATABLE :: crestartmap(:)

! Send/Recv parameters for Particle arrays:
      INTEGER :: crootcnt, cproccnt
      INTEGER :: crecv
      INTEGER, allocatable :: cscatter(:)
      INTEGER, allocatable :: cdispls(:)

      INTEGER, ALLOCATABLE :: ipar_col(:,:)

      CONTAINS

!``````````````````````````````````````````````````````````````````````!
! Subroutine: INIT_READ_RES_DES                                        !
!                                                                      !
! Purpose: Construct the file name and open the DES RES file.          !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE init_read_res_des(BASE, lVERSION, lNEXT_REC)

      use discretelement, only: max_pip, pip
      use discretelement, only: ighost_cnt
      use discretelement, only: neigh_num

      use compar, only: numpes
      use machine, only: open_n1

      use mpi_utility, only: bcast
      use mpi_utility, only: global_all_sum

      implicit none

      CHARACTER(len=*), INTENT(IN)  :: BASE
      DOUBLE PRECISION, INTENT(OUT) :: lVERSION
      INTEGER, INTENT(OUT) :: lNEXT_REC

      CHARACTER(len=32) :: lFNAME

! Integer Error Flag
      INTEGER :: IER


      allocate(pscatter(0:numpes-1))
      allocate(pdispls(0:numpes-1))

      allocate(cscatter(0:numpes-1))
      allocate(cdispls(0:numpes-1))


      IF(bdist_io) THEN

         WRITE(lfname,'(A,I4.4,A)') base//'_DES_',mype,'.RES'
         OPEN(convert='BIG_ENDIAN',unit=rdes_unit, file=lfname,        &
            form='UNFORMATTED', status='UNKNOWN', access='DIRECT',     &
            recl=open_n1)

         READ(rdes_unit, rec=1) lversion
         READ(rdes_unit, rec=2) pin_count
         READ(rdes_unit, rec=3) ighost_cnt
         READ(rdes_unit, rec=4) cin_count

         IF(pip > max_pip) THEN
            write(*,*) "From des_read_restart:"
            write(*,*) "Error: The pip is grater than current max_pip"
            write(*,*) "pip=" ,pip,"; max_pip =", max_pip

         ENDIF

         pip = pin_count
         neigh_num = cin_count

         CALL particle_grow(neigh_num)

      ELSE

         IF(mype == pe_io) THEN
            WRITE(lfname,'(A,A)') base//'_DES.RES'
            OPEN(convert='BIG_ENDIAN',unit=rdes_unit, file=lfname,     &
               form='UNFORMATTED', status='UNKNOWN', access='DIRECT',  &
               recl=open_n1)

            READ(rdes_unit, rec=1) pin_count

            READ(rdes_unit, rec=1) lversion
            READ(rdes_unit, rec=2) pin_count
!           READ(RDES_UNIT, REC=3) -NOTHING-
            READ(rdes_unit, rec=4) cin_count

         ELSE
            pin_count = 10
         ENDIF

         ier = 0

! Allocate the particle restart map. This is used in determining were
! particle data is sent. Only process zero needs this array.
         allocate( prestartmap(pin_count), stat=ier)
         IF(ier/=0) THEN
            WRITE(err_msg, 1200) 'pRestartMap', trim(ival(pin_count))
            CALL flush_err_msg
         ENDIF

         CALL bcast(lversion, pe_io)

! Allocate the collision restart map array. All ranks allocatet this
! array so that mapping the collision data can be done in parallel.
         CALL bcast(cin_count, pe_io)
         allocate( crestartmap(cin_count), stat=ier)
         IF(ier/=0) THEN
            WRITE(err_msg, 1200) 'cRestartMap', trim(ival(cin_count))
            CALL flush_err_msg
         ENDIF

 1200 FORMAT('Error 1200: Unable to allocate sufficient memory to ',&
         'read in DES',/'restart file. size(',a,') = ',a)

         CALL global_all_sum(ier)
         IF(ier/=0) CALL mfix_exit(mype)

      ENDIF

      lnext_rec = 5

      RETURN
      END SUBROUTINE init_read_res_des


!``````````````````````````````````````````````````````````````````````!
! Subroutine: CLOSE_RES_DES                                            !
!                                                                      !
! Purpose: Close the DES RES file.                                     !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE finl_read_res_des


      IF(bdist_io .OR. mype == pe_io) close(rdes_unit)

      IF(allocated(dprocbuf)) deallocate(dprocbuf)
      IF(allocated(drootbuf)) deallocate(drootbuf)
      IF(allocated(iprocbuf)) deallocate(iprocbuf)
      IF(allocated(irootbuf)) deallocate(irootbuf)

      IF(allocated(prestartmap)) deallocate(prestartmap)
      IF(allocated(crestartmap)) deallocate(crestartmap)

      IF(allocated(pscatter)) deallocate(pscatter)
      IF(allocated(pdispls)) deallocate(pdispls)

      IF(allocated(cscatter)) deallocate(cscatter)
      IF(allocated(cdispls)) deallocate(cdispls)




      RETURN
      END SUBROUTINE finl_read_res_des



!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_PAR_POS                                             !
!                                                                      !
! Purpose: Generates the mapping used by the scatter routines to send  !
! read data to the correct rank.                                       !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_par_pos(lNEXT_REC)

      use discretelement, only: pip
      use discretelement, only: des_pos_new
      use geometry, only: no_k
      use compar, only: numpes

      use mpi_utility, only: global_sum
      USE in_binary_512

      implicit none

      INTEGER, INTENT(INOUT) :: lNEXT_REC

      INTEGER :: lDIMN
      INTEGER :: LC1, lPROC
      INTEGER :: lScatterCNTS(0:numpes-1)
! The number of particles on each process.
      INTEGER :: PAR_CNT(0:numpes-1)

!-----------------------------------------------

      CALL init_err_msg("READ_PAR_POS")

      ldimn = merge(2,3,no_k)


! All process read positions for distributed IO restarts.
      IF(bdist_io) THEN
         DO lc1 = 1, ldimn
            CALL read_res_des(lnext_rec, des_pos_new(:,lc1))
         ENDDO
         RETURN
      ENDIF

      allocate( dpar_pos(pin_count, ldimn))

! Only the IO proccess reads positions.
      IF(mype == pe_io) THEN
         DO lc1=1, merge(2,3,no_k)
            CALL in_bin_512(rdes_unit, dpar_pos(:,lc1),                &
               pin_count, lnext_rec)
         ENDDO
      ENDIF

! Use the particle postions and the domain coverage of each process
! to determine which processor each particle belongs.
      CALL map_parray_to_proc(par_cnt)

! Send the particle position data to the individual ranks.
      CALL scatter_par_pos(par_cnt)

! Set up the read/scatter arrary information.
      pproccnt = pip
      prootcnt = pin_count

! Set the recv count for this process.
      precv = pip

! Construct an array for the Root process that states the number of
! (real) particles on each process.
      lscattercnts(:) = 0; lscattercnts(mype) = pip
      CALL global_sum(lscattercnts,pscatter)

! Calculate the displacements for each process in the global array.
      pdispls(0) = 0
      DO lproc = 1, numpes-1
         pdispls(lproc) = pdispls(lproc-1) + pscatter(lproc-1)
      ENDDO

      IF(allocated(dpar_pos)) deallocate(dpar_pos)

      CALL finl_err_msg

      RETURN
      END SUBROUTINE read_par_pos


!``````````````````````````````````````````````````````````````````````!
! Subroutine: MAP_pARRAY_TO_PROC                                       !
!                                                                      !
! Purpose: Use the particle positions to determine which processor     !
! they live on and count the number of particles on each process.      !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE map_parray_to_proc(lPAR_CNT)

      use discretelement, only: pip
      use discretelement, only: xe, yn, zt
      use geometry, only: imin1, imax1
      use geometry, only: jmin1, jmax1
      use geometry, only: kmin1, kmax1
      use geometry, only: no_k, do_k
      use compar, only: numpes
      use compar, only: istart1_all, iend1_all
      use compar, only: jstart1_all, jend1_all
      use compar, only: kstart1_all, kend1_all

      use mpi_utility, only: bcast
      use mpi_utility, only: global_all_sum

      implicit none

      INTEGER, INTENT(OUT) :: lPAR_CNT(0:numpes-1)

! Data dimensionality flag.
      INTEGER :: lDIMN
! Loop counters.
      INTEGER :: LC1, lPROC
! Error flag.
      INTEGER :: IER(0:numpes-1)
! The X/Y/Z bounds of the physical space "owned" by each process.
      DOUBLE PRECISION :: lxmin(0:numpes-1), lxmax(0:numpes-1)
      DOUBLE PRECISION :: lymin(0:numpes-1), lymax(0:numpes-1)
      DOUBLE PRECISION :: lzmin(0:numpes-1), lzmax(0:numpes-1)
!-----------------------------------------------

      CALL init_err_msg("MAP_pARRAY_TO_PROC")

! Initialize the error flag.
      ier = 0

      ldimn = merge(2, 3, no_k)

! set the domain range for each processor
      DO lproc= 0, numpes-1
         lxmin(lproc) = xe(istart1_all(lproc)-1)
         lxmax(lproc) = xe(iend1_all(lproc))
         lymin(lproc) = yn(jstart1_all(lproc)-1)
         lymax(lproc) = yn(jend1_all(lproc))
         lzmin(lproc) = zt(kstart1_all(lproc)-1)
         lzmax(lproc) = zt(kend1_all(lproc))

! modify the range for mass inlet and outlet, as particles injected
! can lie outside the domain and not ghost particles
         IF(istart1_all(lproc).eq.imin1) &
            lxmin(lproc) = xe(istart1_all(lproc)-2)
         IF(iend1_all(lproc).eq.imax1) &
            lxmax(lproc) = xe(iend1_all(lproc)+1)
         IF(jstart1_all(lproc).eq.jmin1) &
            lymin(lproc) = yn(jstart1_all(lproc)-2)
         IF(jend1_all(lproc).eq.jmax1)  &
            lymax(lproc) = yn(jend1_all(lproc)+1)
         IF(kstart1_all(lproc).eq.kmin1 .AND. do_k) &
            lzmin(lproc) = zt(kstart1_all(lproc)-2)
         IF(kend1_all(lproc).eq.kmax1 .AND. do_k) &
            lzmax(lproc) = zt(kend1_all(lproc)+1)
      ENDDO

! build the send buffer in PE_IO proc
! first pass to get the count of particles
      ier = 0
      prestartmap(:) = -1
      lpar_cnt(:) = 0
      IF(mype == pe_io) THEN
         DO lc1 = 1, pin_count
            DO lproc=0, numpes-1
               IF(dpar_pos(lc1,1) >= lxmin(lproc) .AND. &
                  dpar_pos(lc1,1) <  lxmax(lproc) .AND. &
                  dpar_pos(lc1,2) >= lymin(lproc) .AND. &
                  dpar_pos(lc1,2) <  lymax(lproc)) THEN
                  IF(no_k)THEN
                     lpar_cnt(lproc) = lpar_cnt(lproc) + 1
                     prestartmap(lc1) = lproc
                     EXIT
                  ELSE
                     IF(dpar_pos(lc1,3) >= lzmin(lproc) .AND. &
                        dpar_pos(lc1,3) <  lzmax(lproc)) THEN
                        lpar_cnt(lproc) = lpar_cnt(lproc) + 1
                        prestartmap(lc1) = lproc
                        EXIT
                     ENDIF
                  ENDIF
               ENDIF
            ENDDO  ! Loop over processes
            IF (prestartmap(lc1) == -1) then
               ier(mype) = -1
               WRITE(err_msg,1000) trim(ival(lc1))
               CALL flush_err_msg(footer=.false.)
               IF(no_k) THEN
                  WRITE(err_msg,1001) dpar_pos(lc1,1:2)
                  CALL flush_err_msg(header=.false.)
               ELSE
                  WRITE(err_msg,1002) dpar_pos(lc1,1:3)
                  CALL flush_err_msg(header=.false.)
               ENDIF
            ENDIF
         ENDDO  ! Loop over particles
      ENDIF

 1000 FORMAT('Error 1000: Unable to locate particle inside domain:',/&
         3x,'Particle Number:',a)
 1001 FORMAT(3x,'X POS: ',g12.5,/3x,'Y POS: ',g12.5)
 1002 FORMAT(3x,'X POS: ',g12.5,/3x,'Y POS: ',g12.5,/3x,'Z POS: ',g12.5)

! Send out the error flag and exit if needed.
      CALL bcast(ier, pe_io)
      IF(ier(pe_io) /= 0) CALL mfix_exit(mype)

! PE_IO sends out the number of particles for each process.
      CALL bcast(lpar_cnt(0:numpes-1), pe_io)

! Each process stores the number of particles-on-its-process. The error
! flag is set if that number exceeds the maximum.
      pip = lpar_cnt(mype)
      CALL particle_grow(pip)

! Global collection of error flags to abort it the max was exceeded.
      CALL global_all_sum(ier)
      IF(sum(ier) /= 0) THEN
         WRITE(err_msg,1100)
         CALL flush_err_msg(footer=.false.)
         DO lc1=0, numpes-1
            IF(ier(lc1) /= 0) THEN
               WRITE(err_msg,"(3(2x,I10))")lc1,ier(lc1)-1,lpar_cnt(lc1)
               CALL flush_err_msg(header=.false., footer=.false.)
            ENDIF
         ENDDO
         WRITE(err_msg,"('Aborting.')")
         CALL flush_err_msg(header=.false.,abort=.true.)
      ENDIF

 1100 FORMAT('Error 1100: Maximum number of particles exceeded.',2/    &
         5x,'Process',5x,'Maximum',7x,'Count')


      CALL finl_err_msg

      RETURN
      END SUBROUTINE map_parray_to_proc



!``````````````````````````````````````````````````````````````````````!
! Subroutine: DES_RESTART_MAP                                          !
!                                                                      !
! Purpose: Generates the mapping used by the scatter routines to send  !
! read data to the correct rank.                                       !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE scatter_par_pos(lPAR_CNT)

      use compar, only: numpes

      use discretelement, only: des_pos_new
      use discretelement, only: pip
      use functions, only: set_normal
      use geometry, only: no_k

      implicit none

! Number of particles on each process.
      INTEGER, INTENT(INOUT) :: lPAR_CNT(0:numpes-1)
! Dimensionality flag.
      INTEGER :: lDIMN
! Loop counters.
      INTEGER :: LC1, lPROC, lBuf

      ldimn = merge(2,3,no_k)

! Set up the recv count and allocate the local process buffer.
      iscr_recvcnt = pip*ldimn
      allocate (dprocbuf(iscr_recvcnt))

! Allocate the buffer for the root.
      IF (mype == pe_io) THEN
         allocate (drootbuf(pin_count*ldimn))
      ELSE
         allocate (drootbuf(10))
      ENDIF

! The IO processor builds drootbuffer and iDISLS
      IF(mype == pe_io) THEN
! Determine the offsets for each process and the amount of data that
! is to be scattered to each.
         idispls(0) = 0
         iscattercnts(0) = lpar_cnt(0)*ldimn
         DO lproc = 1, numpes-1
            idispls(lproc) = idispls(lproc-1) + iscattercnts(lproc-1)
            iscattercnts(lproc) = lpar_cnt(lproc)*ldimn
         ENDDO
! Copy the position data into the root buffer, mapped to the owner
! process.
         lpar_cnt(:) = 0
         DO lc1 = 1,pin_count
            lproc = prestartmap(lc1)
            lbuf = idispls(lproc) + lpar_cnt(lproc)*ldimn+1
            drootbuf(lbuf:lbuf+ldimn-1) = dpar_pos(lc1,1:ldimn)
            lbuf = lbuf + ldimn
            lpar_cnt(lproc) = lpar_cnt(lproc) + 1
         ENDDO
      ENDIF
      CALL desmpi_scatterv(ptype=2)

! Unpack the particle data.
      DO lc1 = 1, pip
         lbuf = (lc1-1)*ldimn+1
         des_pos_new(lc1,1:ldimn) = dprocbuf(lbuf:lbuf+ldimn-1)
         lbuf = lbuf + ldimn
         CALL set_normal(lc1)
      ENDDO

      IF(allocated(drootbuf)) deallocate(drootbuf)
      IF(allocated(dprocbuf)) deallocate(dprocbuf)
      IF(allocated(dpar_pos)) deallocate(dpar_pos)

      RETURN
      END SUBROUTINE scatter_par_pos

!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_PAR_COL                                             !
!                                                                      !
! Purpose: Generates the mapping used by the scatter routines to send  !
! read data to the correct rank.                                       !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_par_col(lNEXT_REC)

      use discretelement, only: neighbors, neigh_num
      use compar, only: numpes

      use mpi_init_des, only: des_restart_ghost
      use mpi_utility, only: bcast
      use mpi_utility, only: global_sum
      use mpi_utility, only: global_all_sum
      use in_binary_512i

      implicit none

      INTEGER, INTENT(INOUT) :: lNEXT_REC

      INTEGER :: LC1, lPROC
      INTEGER :: lScatterCNTS(0:numpes-1)
! The number of particles on each process.
      INTEGER :: COL_CNT(0:numpes-1)

!-----------------------------------------------

      CALL init_err_msg("READ_PAR_COL")

! All process read positions for distributed IO restarts.
      IF(bdist_io) THEN
         CALL read_res_des(lnext_rec, neighbors(:))
      ENDIF

      CALL des_restart_ghost

      allocate(ipar_col(2, cin_count))
      ipar_col = 0

! Only the IO proccess reads positions.
      IF(mype == pe_io) THEN
         DO lc1=1, 2
            CALL in_bin_512i(rdes_unit, ipar_col(lc1,:),               &
               cin_count, lnext_rec)
         ENDDO
      ENDIF

! Broadcast collision data to all the other processes.
       CALL global_all_sum(ipar_col)

! Determine which process owns the neighbor datasets. This is done either
! through matching global ids or a search. The actual method depends
! on the ability to allocate a large enough array.
      CALL map_carray_to_proc(col_cnt)

! Send the particle position data to the individual ranks.
      CALL global_to_loc_col

! Set up the read/scatter arrary information.
      cproccnt = neigh_num
      crootcnt = cin_count

! Set the recv count for this process.
      crecv = neigh_num

! Construct an array for the Root process that states the number of
! (real) particles on each process.
      lscattercnts(:) = 0; lscattercnts(mype) = neigh_num
      CALL global_sum(lscattercnts,cscatter)

! Calculate the displacements for each process in the global array.
      cdispls(0) = 0
      DO lproc = 1, numpes-1
         cdispls(lproc) = cdispls(lproc-1) + cscatter(lproc-1)
      ENDDO

      CALL finl_err_msg

      RETURN
      END SUBROUTINE read_par_col


!``````````````````````````````````````````````````````````````````````!
! Subroutine: MAP_cARRAY_TO_PROC                                       !
!                                                                      !
! Purpose: Use the particle positions to determine which processor     !
! they live on and count the number of particles on each process.      !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE map_carray_to_proc(lCOL_CNT)

      use compar, only: numpes, mype
      use discretelement, only: iglobal_id
      use discretelement, only: pip
      use discretelement, only: neigh_num
      use functions, only: is_ghost, is_entering_ghost, is_exiting_ghost

      use mpi_utility, only: global_all_sum
      use mpi_utility, only: global_all_max

      implicit none

      INTEGER, INTENT(OUT) :: lCOL_CNT(0:numpes-1)

! Loop counters.
      INTEGER :: LC1, LC2
! Error flag.
      INTEGER :: IER
! Max global id.
      INTEGER :: MAX_ID, lSTAT

      INTEGER, ALLOCATABLE :: lGLOBAL_OWNER(:)

!-----------------------------------------------

      CALL init_err_msg("MAP_cARRAY_TO_PROC")

! Initialize the error flag.
      ier = 0

      max_id = maxval(iglobal_id(1:pip))
      CALL global_all_max(max_id)

      allocate(lglobal_owner(max_id), stat=lstat)
      CALL global_all_sum(lstat)

! All ranks successfully allocated the array. This permits a crude
! but much faster collision owner detection.
      IF(lstat == 0) THEN

         WRITE(err_msg,"('Matching DES neighbor data by global owner.')")
         CALL flush_err_msg(header=.false.,footer=.false.)

         lglobal_owner = 0
         DO lc1=1, pip
            IF(.NOT.is_ghost(lc1) .AND. .NOT.is_entering_ghost(lc1) &
               .AND. .NOT.is_exiting_ghost(lc1)) &
               lglobal_owner(iglobal_id(lc1)) = mype + 1
         ENDDO

! Loop over the neighbor list and match the read global ID to
! one of the global IDs.
         lcol_cnt = 0
         crestartmap = 0
         DO lc1=1, cin_count
            IF(lglobal_owner(ipar_col(1,lc1)) == mype + 1) THEN
               crestartmap(lc1) = mype + 1
               lcol_cnt(mype) = lcol_cnt(mype) + 1
            ENDIF
         ENDDO
! One or more ranks could not allocate the memory needed to do the
! quick and dirty match so do a search instead.
      ELSE

         WRITE(err_msg,"('Matching DES neighbor data by search.')")
         CALL flush_err_msg(header=.false.,footer=.false.)

! Loop over the neighbor list and match the read global ID to
! one of the global IDs.
         lcol_cnt = 0
         crestartmap = 0
         lc1_lp: DO lc1=1, cin_count
            DO lc2=1, pip!-iGHOST_CNT
               IF(ipar_col(1,lc1) == iglobal_id(lc2)) THEN
                  crestartmap(lc1) = mype + 1
                  lcol_cnt(mype) = lcol_cnt(mype) + 1
                  cycle lc1_lp
               ENDIF
            ENDDO
         ENDDO lc1_lp

      ENDIF

! Clean up the large array as it is no longer needed.
      IF(allocated(lglobal_owner)) deallocate(lglobal_owner)

! Calculate the number of matched collisions over all processes. Throw
! and error if it doesn't match the number of read collisions.
      CALL global_all_sum(lcol_cnt)
      IF(sum(lcol_cnt) /= cin_count) THEN
         WRITE(err_msg,1000) cin_count, sum(lcol_cnt)
         CALL flush_err_msg(abort=.true.)
      ENDIF

1000 FORMAT('Error 1000: Unable to establish the own of all read ',    &
         'collision data.',/3x,'Number of Collisions: ',i10,/3x,       &
         'Matched Collisions:   ',i10)

! Sync the collision restart map arcross all ranks.
      CALL global_all_sum(crestartmap)

! Error checking and cleanup.
      DO lc1 = 1, cin_count
! Verify that each collision is owned by a rank.
         IF (crestartmap(lc1) == 0) THEN
            ier = -1
            WRITE(err_msg,1100) trim(ival(lc1)), trim(ival(            &
               ipar_col(1,lc1))), trim(ival(ipar_col(2,lc1)))
            CALL flush_err_msg(abort=.true.)

 1100 FORMAT('Error 1100: Unable to locate process neighbor owner:',/  &
         3x,'Neighbor Number:',a,/3x,'Particles: ',a,' and ',a)

         ELSEIF(crestartmap(lc1) > numpes) THEN

            ier = -1
            WRITE(err_msg,1101) trim(ival(lc1)), trim(ival(            &
              ipar_col(1,lc1))), trim(ival(ipar_col(2,lc1)))
             CALL flush_err_msg(abort=.true.)

 1101 FORMAT('Error 1101: More than one process neighbor owner:',/     &
         3x,'Neighbor Number:',a,/3x,'Particles: ',a,' and ',a)

! Shift the rank ID to the correct value.
         ELSE
            crestartmap(lc1) = crestartmap(lc1) - 1
         ENDIF
      ENDDO

! Send out the error flag and exit if needed.
      CALL global_all_sum(ier, pe_io)
      IF(ier /= 0) CALL mfix_exit(mype)

! Each process stores the number of particles-on-its-process. The error
! flag is set if that number exceeds the maximum.
      neigh_num = lcol_cnt(mype)

      CALL neighbor_grow(neigh_num)

      CALL finl_err_msg

      RETURN
      END SUBROUTINE map_carray_to_proc


!``````````````````````````````````````````````````````````````````````!
! Subroutine: GLOBAL_TO_LOC_COL                                        !
!                                                                      !
! Purpose: Generates the mapping used by the scatter routines to send  !
! read data to the correct rank.                                       !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE global_to_loc_col

      use discretelement, only: iglobal_id
      use discretelement, only: pip

      use mpi_utility, only: global_all_sum
      use mpi_utility, only: global_all_max

      use funits, only: dmp_log

      use error_manager

      implicit none

! Loop counters.
      INTEGER :: LC1, LC2, LC3, IER
      INTEGER :: UNMATCHED
      INTEGER, ALLOCATABLE :: iLOCAL_ID(:)

! Max global id.
      INTEGER :: MAX_ID, lSTAT
! Debug flags.
      LOGICAL :: dFlag
      LOGICAL, parameter :: setDBG = .false.

      CALL init_err_msg("GLOBAL_TO_LOC_COL")

! Initialize the error flag.
      ier = 0

! Set the local debug flag.
      dflag = (dmp_log .AND. setdbg)

      max_id = maxval(iglobal_id(1:pip))
      CALL global_all_max(max_id)

      allocate(ilocal_id(max_id), stat=lstat)
      CALL global_all_sum(lstat)

! All ranks successfully allocated the array. This permits a crude
! but much faster collision owner detection.
      IF(lstat /= 0) THEN
         WRITE(err_msg,1000)
         CALL flush_err_msg(abort=.true.)
      ENDIF

 1000 FORMAT('Error 1000: Unable to allocate sufficient memory to ',&
         'generate the',/'map from global to local particle IDs.')

      ilocal_id = 0
      DO lc1=1, pip
         ilocal_id(iglobal_id(lc1)) = lc1
      ENDDO

! Store the particle data.
      lc3 = 1
      lc2 = 0
      unmatched = 0

! FIXME Fix Restart
!      LP1: DO LC1 = 1, cIN_COUNT
!          IF(cRestartMap(LC1) == myPE) THEN
!             LC2 = LC2 + 1
!             NEIGHBORS(LC2) = iLOCAL_ID(iPAR_COL(1,LC1))
!             NEIGHBORS(2,LC2) = iLOCAL_ID(iPAR_COL(2,LC1))
! ! Verify that the local indices are valid. If they do not match it is
! ! likely because one of the neighbor was removed via an outlet at the time
! ! the RES file was written but the ghost data wasn't updated.
!             IF(NEIGHBORS(1,LC2) == 0 .OR. NEIGHBORS(2,LC2) == 0) THEN
!                UNMATCHED = UNMATCHED + 1
!                IF(dFLAG) THEN
!                   WRITE(ERR_MSG,1100) iPAR_COL(1,LC1), NEIGHBORS(1,LC2),   &
!                      iPAR_COL(2,LC1), NEIGHBORS(2,LC2)
!                   CALL FLUSH_ERR_MSG(ABORT=.FALSE.)
!                ENDIF
!                DO WHILE(PEA(LC3,1))
!                   LC3 = LC3 + 1
!                ENDDO
!                NEIGHBORS(2,LC2) = LC3
!             ENDIF
!          ENDIF
!       ENDDO LP1

! 1100 FORMAT('Error 1100: Particle neighbor local indices are invalid.',/  &
!         5x,'Global-ID    Local-ID',/' 1:  ',2(3x,I9),/' 2:  ',2(3x,I9))

      CALL global_all_sum(unmatched)
      IF(unmatched /= 0) THEN
         WRITE(err_msg,1101) trim(ival(unmatched))
         CALL flush_err_msg
      ENDIF

 1101 FORMAT(' Warning: 1101: ',a,' particle neighbor datasets were ',&
         'not matched',/' during restart.')

      IF(allocated(ilocal_id)) deallocate(ilocal_id)

      CALL finl_err_msg

      RETURN
      END SUBROUTINE global_to_loc_col



!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_DES_0I                                          !
!                                                                      !
! Purpose: Write scalar integers to RES file.                          !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_des_0i(lNEXT_REC, INPUT_I)

      use mpi_utility, only: bcast

      IMPLICIT NONE

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      INTEGER, INTENT(OUT) :: INPUT_I

      IF(bdist_io) THEN
         READ(rdes_unit, rec=lnext_rec) input_i
      ELSE
         IF(mype == pe_io) READ(rdes_unit, rec=lnext_rec) input_i
         CALL bcast(input_i, pe_io)
      ENDIF

      lnext_rec = lnext_rec + 1

      RETURN
      END SUBROUTINE read_res_des_0i


!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_1I                                              !
!                                                                      !
! Purpose: Write scalar integers to RES file.                          !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_des_1i(lNEXT_REC, INPUT_I)

      use mpi_utility, only: bcast
      USE in_binary_512i

      IMPLICIT NONE

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      INTEGER, INTENT(OUT) :: INPUT_I(:)

      INTEGER :: lSIZE

      lsize = size(input_i)

      IF(bdist_io) THEN
         CALL in_bin_512i(rdes_unit, input_i, lsize, lnext_rec)
      ELSE
         IF(mype == pe_io) &
            CALL in_bin_512i(rdes_unit, input_i, lsize, lnext_rec)
         CALL bcast(input_i, pe_io)
      ENDIF


      RETURN
      END SUBROUTINE read_res_des_1i


!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_DES_0D                                          !
!                                                                      !
! Purpose: Write scalar double percision values to RES file.           !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_des_0d(lNEXT_REC, INPUT_D)

      use mpi_utility, only: bcast

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      DOUBLE PRECISION, INTENT(OUT) :: INPUT_D

      IF(bdist_io) THEN
         READ(rdes_unit, rec=lnext_rec) input_d
      ELSE
         IF(mype == pe_io) READ(rdes_unit, rec=lnext_rec) input_d
         CALL bcast(input_d, pe_io)
      ENDIF
      lnext_rec = lnext_rec + 1

      RETURN
      END SUBROUTINE read_res_des_0d


!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_DES_1D                                          !
!                                                                      !
! Purpose: Write scalar integers to RES file.                          !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_des_1d(lNEXT_REC, INPUT_D)

      use mpi_utility, only: bcast
      USE in_binary_512

      IMPLICIT NONE

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      DOUBLE PRECISION, INTENT(OUT) :: INPUT_D(:)

      INTEGER :: lSIZE

      lsize = size(input_d)

      IF(bdist_io) THEN
         CALL in_bin_512(rdes_unit, input_d, lsize, lnext_rec)
      ELSE
         IF(mype == pe_io) &
            CALL in_bin_512(rdes_unit, input_d, lsize, lnext_rec)
         CALL bcast(input_d, pe_io)
      ENDIF


      RETURN
      END SUBROUTINE read_res_des_1d


!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_DES_0L                                          !
!                                                                      !
! Purpose: Write scalar logical values to RES file.                    !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_des_0l(lNEXT_REC, OUTPUT_L)

      use mpi_utility, only: bcast

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      LOGICAL, INTENT(OUT) :: OUTPUT_L

      INTEGER :: OUTPUT_I

      output_l = .true.

      IF(bdist_io)THEN
         READ(rdes_unit, rec=lnext_rec) output_i
      ELSE
         IF(mype == pe_io) READ(rdes_unit, rec=lnext_rec) output_i
         CALL bcast(output_i, pe_io)
      ENDIF

      IF(output_i == 1) output_l = .true.
      lnext_rec = lnext_rec + 1

      RETURN
      END SUBROUTINE read_res_des_0l


!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_DES_1L                                          !
!                                                                      !
! Purpose: Write scalar integers to RES file.                          !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_des_1l(lNEXT_REC, INPUT_L)

      use mpi_utility, only: bcast
      USE in_binary_512i

      IMPLICIT NONE

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      LOGICAL, INTENT(OUT) :: INPUT_L(:)

      INTEGER, ALLOCATABLE :: INPUT_I(:)

      INTEGER :: lSIZE, LC1

      lsize = size(input_i)
      ALLOCATE( input_i(lsize))

      IF(bdist_io) THEN
         CALL in_bin_512i(rdes_unit, input_i, lsize, lnext_rec)
      ELSE
         IF(mype == pe_io) &
            CALL in_bin_512i(rdes_unit, input_i, lsize, lnext_rec)
         CALL bcast(input_i, pe_io)
      ENDIF

      DO lc1=1, lsize
         IF(input_i(lc1) == 1) THEN
            input_l(lc1) = .true.
         ELSE
            input_l(lc1) = .false.
         ENDIF
      ENDDO

      IF(allocated(input_i)) deallocate(input_i)

      RETURN
      END SUBROUTINE read_res_des_1l

!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_DES_1B                                          !
!                                                                      !
! Purpose: Write scalar bytes to RES file.                             !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_parray_1b(lNEXT_REC, OUTPUT_B)

      use discretelement, only: pip

      use desmpi, only: irootbuf
      use desmpi, only: iprocbuf

      use compar, only: numpes
      USE in_binary_512i

      IMPLICIT NONE

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      INTEGER(KIND=1), INTENT(OUT) :: OUTPUT_B(:)

! Loop counters
      INTEGER :: LC1

      INTEGER :: lPROC

      INTEGER, ALLOCATABLE :: OUTPUT_I(:)
      INTEGER, ALLOCATABLE :: lBUF_I(:)
      INTEGER, ALLOCATABLE :: lCOUNT(:)

      allocate(iprocbuf(pproccnt))
      allocate(irootbuf(prootcnt))


      idispls = pdispls
      iscr_recvcnt = precv
      iscattercnts = pscatter

      allocate(output_i(size(output_b)))
      output_i(:) = output_b(:)

      IF(bdist_io) THEN
         CALL in_bin_512i(rdes_unit, output_i, pin_count, lnext_rec)
         output_b(:) = output_i(:)
      ELSE

         IF(mype == pe_io) THEN
            allocate(lbuf_i(pin_count))
            allocate(lcount(0:numpes-1))

            CALL in_bin_512i(rdes_unit, lbuf_i, pin_count, lnext_rec)

            lcount = 0
            DO lc1=1, pin_count
               lproc = prestartmap(lc1)
               lcount(lproc) = lcount(lproc) + 1
               irootbuf(idispls(lproc) + lcount(lproc)) = lbuf_i(lc1)
            ENDDO

            deallocate(lbuf_i)
            deallocate(lcount)
         ENDIF
         CALL desmpi_scatterv(ptype=1)
         DO lc1=1, pip
            output_b(lc1) = iprocbuf(lc1)
         ENDDO

      ENDIF

      deallocate(iprocbuf)
      deallocate(irootbuf)
      deallocate(output_i)

      RETURN
      END SUBROUTINE read_res_parray_1b

!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_DES_1I                                          !
!                                                                      !
! Purpose: Write scalar integers to RES file.                          !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_parray_1i(lNEXT_REC, OUTPUT_I)

      use discretelement, only: pip

      use desmpi, only: irootbuf
      use desmpi, only: iprocbuf

      use compar, only: numpes
      USE in_binary_512i

      IMPLICIT NONE

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      INTEGER, INTENT(OUT) :: OUTPUT_I(:)

! Loop counters
      INTEGER :: LC1

      INTEGER :: lPROC

      INTEGER, ALLOCATABLE :: lBUF_I(:)
      INTEGER, ALLOCATABLE :: lCOUNT(:)


      allocate(iprocbuf(pproccnt))
      allocate(irootbuf(prootcnt))

      idispls = pdispls
      iscr_recvcnt = precv
      iscattercnts = pscatter

      IF(bdist_io) THEN
         CALL in_bin_512i(rdes_unit, output_i, pin_count, lnext_rec)
      ELSE

         IF(mype == pe_io) THEN
            allocate(lbuf_i(pin_count))
            allocate(lcount(0:numpes-1))

            CALL in_bin_512i(rdes_unit, lbuf_i, pin_count, lnext_rec)

            lcount = 0
            DO lc1=1, pin_count
               lproc = prestartmap(lc1)
               lcount(lproc) = lcount(lproc) + 1
               irootbuf(idispls(lproc) + lcount(lproc)) = lbuf_i(lc1)
            ENDDO

            deallocate(lbuf_i)
            deallocate(lcount)
         ENDIF
         CALL desmpi_scatterv(ptype=1)
         DO lc1=1, pip
            output_i(lc1) = iprocbuf(lc1)
         ENDDO

      ENDIF

      deallocate(iprocbuf)
      deallocate(irootbuf)

      RETURN
      END SUBROUTINE read_res_parray_1i



!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_pARRAY_1D                                       !
!                                                                      !
! Purpose: Write scalar integers to RES file.                          !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_parray_1d(lNEXT_REC, OUTPUT_D)

      use discretelement, only: pip
      use desmpi, only: drootbuf
      use desmpi, only: dprocbuf
      use compar, only: numpes
      USE in_binary_512

      IMPLICIT NONE

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      DOUBLE PRECISION, INTENT(OUT) :: OUTPUT_D(:)

! Loop counters
      INTEGER :: LC1

      INTEGER :: lPROC

      DOUBLE PRECISION, ALLOCATABLE :: lBUF_D(:)
      INTEGER, ALLOCATABLE :: lCOUNT(:)


      allocate(dprocbuf(pproccnt))
      allocate(drootbuf(prootcnt))

      idispls = pdispls
      iscr_recvcnt = precv
      iscattercnts = pscatter

      IF(bdist_io) THEN
         CALL in_bin_512(rdes_unit, output_d, pin_count, lnext_rec)
      ELSE
         IF(mype == pe_io) THEN
            allocate(lbuf_d(pin_count))
            allocate(lcount(0:numpes-1))

            CALL in_bin_512(rdes_unit, lbuf_d, pin_count, lnext_rec)

            lcount = 0
            DO lc1=1, pin_count
               lproc = prestartmap(lc1)
               lcount(lproc) = lcount(lproc) + 1
               drootbuf(idispls(lproc) + lcount(lproc)) = lbuf_d(lc1)
            ENDDO

            deallocate(lbuf_d)
            deallocate(lcount)
         ENDIF
         CALL desmpi_scatterv(ptype=2)
         DO lc1=1, pip
            output_d(lc1) = dprocbuf(lc1)
         ENDDO
      ENDIF

      deallocate(dprocbuf)
      deallocate(drootbuf)

      RETURN
      END SUBROUTINE read_res_parray_1d


!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_pARRAY_1L                                       !
!                                                                      !
! Purpose: Write scalar integers to RES file.                          !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_parray_1l(lNEXT_REC, OUTPUT_L)

      use discretelement, only: pip
      use desmpi, only: irootbuf
      use desmpi, only: iprocbuf
      use compar, only: numpes
      USE in_binary_512i

      IMPLICIT NONE

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      LOGICAL, INTENT(OUT) :: OUTPUT_L(:)

! Loop counters
      INTEGER :: LC1

      INTEGER :: lPROC

      INTEGER, ALLOCATABLE :: lBUF_I(:)
      INTEGER, ALLOCATABLE :: lCOUNT(:)

      allocate(iprocbuf(pproccnt))
      allocate(irootbuf(prootcnt))

      idispls = pdispls
      iscr_recvcnt = precv
      iscattercnts = pscatter

      IF(bdist_io) THEN
         allocate(lbuf_i(pin_count))
         CALL in_bin_512i(rdes_unit, lbuf_i, pin_count, lnext_rec)
         DO lc1=1,pin_count
            IF(lbuf_i(lc1) == 1) THEN
               output_l(lc1) = .true.
            ELSE
               output_l(lc1) = .false.
            ENDIF
         ENDDO
         deallocate(lbuf_i)
      ELSE
         IF(mype == pe_io) THEN
            allocate(lbuf_i(pin_count))
            allocate(lcount(0:numpes-1))

            CALL in_bin_512i(rdes_unit, lbuf_i, pin_count, lnext_rec)

            lcount = 0
            DO lc1=1, pin_count
               lproc = prestartmap(lc1)
               lcount(lproc) = lcount(lproc) + 1
               irootbuf(idispls(lproc) + lcount(lproc)) = lbuf_i(lc1)
            ENDDO

            deallocate(lbuf_i)
            deallocate(lcount)
         ENDIF
         CALL desmpi_scatterv(ptype=1)
         DO lc1=1, pip
            IF(iprocbuf(lc1) == 1) THEN
               output_l(lc1) = .true.
            ELSE
               output_l(lc1) = .false.
            ENDIF
         ENDDO
      ENDIF

      deallocate(iprocbuf)
      deallocate(irootbuf)

      RETURN
      END SUBROUTINE read_res_parray_1l


!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_DES_1I                                          !
!                                                                      !
! Purpose: Write scalar integers to RES file.                          !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_carray_1i(lNEXT_REC, OUTPUT_I)

      use desmpi, only: irootbuf
      use desmpi, only: iprocbuf
      use compar, only: numpes
      use discretelement, only: neigh_num
      USE in_binary_512i

      IMPLICIT NONE

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      INTEGER, INTENT(OUT) :: OUTPUT_I(:)

! Loop counters
      INTEGER :: LC1

      INTEGER :: lPROC

      INTEGER, ALLOCATABLE :: lBUF_I(:)
      INTEGER, ALLOCATABLE :: lCOUNT(:)


      allocate(iprocbuf(cproccnt))
      allocate(irootbuf(crootcnt))

      idispls = cdispls
      iscr_recvcnt = crecv
      iscattercnts = cscatter

      IF(bdist_io) THEN
         CALL in_bin_512i(rdes_unit, output_i, cin_count, lnext_rec)
      ELSE
         IF(mype == pe_io) THEN
            allocate(lbuf_i(cin_count))
            allocate(lcount(0:numpes-1))

            CALL in_bin_512i(rdes_unit, lbuf_i, cin_count, lnext_rec)

            lcount = 0
            DO lc1=1, cin_count
               lproc = crestartmap(lc1)
               lcount(lproc) = lcount(lproc) + 1
               irootbuf(idispls(lproc) + lcount(lproc)) = lbuf_i(lc1)
            ENDDO

            deallocate(lbuf_i)
            deallocate(lcount)
         ENDIF
         CALL desmpi_scatterv(ptype=1)
         DO lc1=1, neigh_num
            output_i(lc1) = iprocbuf(lc1)
         ENDDO
      ENDIF

      deallocate(iprocbuf)
      deallocate(irootbuf)

      RETURN
      END SUBROUTINE read_res_carray_1i


!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_cARRAY_1D                                       !
!                                                                      !
! Purpose: Write scalar integers to RES file.                          !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_carray_1d(lNEXT_REC, OUTPUT_D)

      use compar, only: numpes
      use discretelement, only: neigh_num
      use desmpi, only: drootbuf
      use desmpi, only: dprocbuf
      USE in_binary_512

      IMPLICIT NONE

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      DOUBLE PRECISION, INTENT(OUT) :: OUTPUT_D(:)

! Loop counters
      INTEGER :: LC1

      INTEGER :: lPROC

      DOUBLE PRECISION, ALLOCATABLE :: lBUF_D(:)
      INTEGER, ALLOCATABLE :: lCOUNT(:)


      allocate(dprocbuf(cproccnt))
      allocate(drootbuf(crootcnt))

      idispls = cdispls
      iscr_recvcnt = crecv
      iscattercnts = cscatter


      IF(bdist_io) THEN
         CALL in_bin_512(rdes_unit, output_d, cin_count, lnext_rec)
      ELSE
         IF(mype == pe_io) THEN
            allocate(lbuf_d(cin_count))
            allocate(lcount(0:numpes-1))

            CALL in_bin_512(rdes_unit, lbuf_d, cin_count, lnext_rec)

            lcount = 0
            DO lc1=1, cin_count
               lproc = crestartmap(lc1)
               lcount(lproc) = lcount(lproc) + 1
               drootbuf(idispls(lproc) + lcount(lproc)) = lbuf_d(lc1)
            ENDDO

            deallocate(lbuf_d)
            deallocate(lcount)
         ENDIF
         CALL desmpi_scatterv(ptype=2)
         DO lc1=1, neigh_num
            output_d(lc1) = dprocbuf(lc1)
         ENDDO
      ENDIF

      deallocate(dprocbuf)
      deallocate(drootbuf)

      RETURN
      END SUBROUTINE read_res_carray_1d


!``````````````````````````````````````````````````````````````````````!
! Subroutine: READ_RES_pARRAY_1L                                       !
!                                                                      !
! Purpose: Write scalar integers to RES file.                          !
!``````````````````````````````````````````````````````````````````````!
      SUBROUTINE read_res_carray_1l(lNEXT_REC, OUTPUT_L)

      use compar, only: numpes
      use discretelement, only: neigh_num
      use desmpi, only: irootbuf
      use desmpi, only: iprocbuf
      USE in_binary_512i

      IMPLICIT NONE

      INTEGER, INTENT(INOUT) :: lNEXT_REC
      LOGICAL, INTENT(OUT) :: OUTPUT_L(:)

! Loop counters
      INTEGER :: LC1

      INTEGER :: lPROC

      INTEGER, ALLOCATABLE :: lBUF_I(:)
      INTEGER, ALLOCATABLE :: lCOUNT(:)

      allocate(iprocbuf(cproccnt))
      allocate(irootbuf(crootcnt))

      idispls = cdispls
      iscr_recvcnt = crecv
      iscattercnts = cscatter

      IF(bdist_io) THEN
         allocate(lbuf_i(cin_count))
         CALL in_bin_512i(rdes_unit, lbuf_i, cin_count, lnext_rec)
         DO lc1=1,cin_count
            IF(lbuf_i(lc1) == 1) THEN
               output_l(lc1) = .true.
            ELSE
               output_l(lc1) = .false.
            ENDIF
         ENDDO
         deallocate(lbuf_i)
      ELSE
         IF(mype == pe_io) THEN
            allocate(lbuf_i(cin_count))
            allocate(lcount(0:numpes-1))

            CALL in_bin_512i(rdes_unit, lbuf_i, cin_count, lnext_rec)

            lcount = 0
            DO lc1=1, cin_count
               lproc = crestartmap(lc1)
               lcount(lproc) = lcount(lproc) + 1
               irootbuf(idispls(lproc) + lcount(lproc)) = lbuf_i(lc1)
            ENDDO

            deallocate(lbuf_i)
            deallocate(lcount)
         ENDIF
         CALL desmpi_scatterv(ptype=1)
         DO lc1=1, neigh_num
            IF(iprocbuf(lc1) == 1) THEN
               output_l(lc1) = .true.
            ELSE
               output_l(lc1) = .false.
            ENDIF
         ENDDO
      ENDIF

      deallocate(iprocbuf)
      deallocate(irootbuf)

      RETURN
      END SUBROUTINE read_res_carray_1l

      END MODULE read_res1_des