psblas3-caf:

base/comm/internals/psi_dswapdata.F90 base/internals/psi_crea_index.f90 base/internals/psi_desc_impl.f90 base/modules/desc/psb_desc_mod.F90 base/modules/psi_d_mod.f90 base/modules/psi_i_mod.f90 base/modules/psi_penv_mod.F90 test/pargen/ppde3d.f90 test/pargen/runs/ppde.inp First version of swapdata with CAF.
9 years ago · 003243d78d
parent 9b482355fa
commit 003243d78d
9 changed files with 488 additions and 31 deletions
--- a/base/comm/internals/psi_dswapdata.F90
+++ b/base/comm/internals/psi_dswapdata.F90
@ -1045,6 +1045,7 @@ subroutine psi_dswapdata_vect(flag,beta,y,desc_a,work,info,data)
  ! locals
  integer(psb_ipk_) :: ictxt, np, me, icomm, idxs, idxr, totxch, data_, err_act
  class(psb_i_base_vect_type), pointer :: d_vidx
+  class(psb_xch_idx_type), pointer     :: d_xchg
  character(len=20)  :: name

  info=psb_success_
@ -1073,13 +1074,18 @@ subroutine psi_dswapdata_vect(flag,beta,y,desc_a,work,info,data)
    data_ = psb_comm_halo_
  end if

-  call desc_a%get_list(data_,d_vidx,totxch,idxr,idxs,info) 
+  call desc_a%get_list(data_,d_vidx,totxch,idxr,idxs,info)
+  if (info == 0) call desc_a%get_list(data_,d_xchg,info)
  if (info /= psb_success_) then 
    call psb_errpush(psb_err_internal_error_,name,a_err='psb_cd_get_list')
    goto 9999
  end if

-  call psi_swapdata(ictxt,icomm,flag,beta,y,d_vidx,totxch,idxs,idxr,work,info)
+  if (.false.) then 
+    call psi_swapdata(ictxt,icomm,flag,beta,y,d_vidx,totxch,idxs,idxr,work,info)
+  else
+    call psi_swapdata(ictxt,icomm,flag,beta,y,d_xchg,info)
+  end if
  if (info /= psb_success_) goto 9999

  call psb_erractionrestore(err_act)
@ -1090,6 +1096,189 @@ subroutine psi_dswapdata_vect(flag,beta,y,desc_a,work,info,data)
    return
 end subroutine psi_dswapdata_vect

+subroutine psi_dswap_xchg_vect(iictxt,iicomm,flag,beta,y,xchg,info)
+  use psi_mod, psb_protect_name => psi_dswap_xchg_vect
+  use psb_error_mod
+  use psb_realloc_mod
+  use psb_desc_mod
+  use psb_penv_mod
+  use psb_d_base_vect_mod
+  use iso_fortran_env
+  implicit none   
+  integer(psb_ipk_), intent(in)          :: iictxt,iicomm,flag
+  integer(psb_ipk_), intent(out)         :: info
+  class(psb_d_base_vect_type)            :: y
+  real(psb_dpk_)                         :: beta
+  class(psb_xch_idx_type), intent(inout) :: xchg
+
+  ! locals
+  integer(psb_mpik_) :: ictxt, icomm, np, me,&
+       & proc_to_comm, p2ptag, iret
+  integer(psb_ipk_) :: nesd, nerv,&
+       & err_act, i, idx_pt, totsnd_, totrcv_,p1,p2,isz,rp1,rp2,&
+       & snd_pt, rcv_pt, pnti, n, ip, img, nxch, myself
+  integer :: count
+  real(psb_dpk_), allocatable, save :: buffer(:)[:], sndbuf(:)
+  type(event_type), allocatable, save :: ufg(:)[:]
+  type(event_type), allocatable, save :: clear[:]
+  integer, save :: last_clear_count = 0
+  logical :: swap_mpi, swap_sync, swap_send, swap_recv,&
+       & albf,do_send,do_recv
+  integer(psb_ipk_) :: ierr(5)
+  character(len=20)  :: name
+
+  info=psb_success_
+  name='psi_xchg_vect'
+  call psb_erractionsave(err_act)
+  ictxt = iictxt
+  icomm = iicomm
+
+  call psb_info(ictxt,me,np) 
+  if (np == -1) then
+    info=psb_err_context_error_
+    call psb_errpush(info,name)
+    goto 9999
+  endif
+
+  if (np /= num_images()) then
+    write(*,*) 'Something is wrong MPI vs CAF ', np, num_images()
+    info = psb_err_internal_error_
+    call psb_errpush(info,name,a_err='Num_images  /= np')
+    goto 9999
+  end if
+
+  n=1
+
+  swap_mpi  = iand(flag,psb_swap_mpi_)  /= 0
+  swap_sync = iand(flag,psb_swap_sync_) /= 0
+  swap_send = iand(flag,psb_swap_send_) /= 0
+  swap_recv = iand(flag,psb_swap_recv_) /= 0
+  do_send = swap_mpi .or. swap_sync .or. swap_send
+  do_recv = swap_mpi .or. swap_sync .or. swap_recv
+
+  if (.not.(do_send.and.do_recv)) then
+    info = psb_err_internal_error_
+    call psb_errpush(info,name,a_err='Unimplemented case in xchg_vect')
+    goto 9999
+  end if
+
+  if (.not.allocated(ufg)) then
+    !write(*,*) 'Allocating events',np
+    allocate(ufg(np)[*],stat=info)
+    if (info == 0) allocate(clear[*],stat=info)
+    if (info /= 0) then
+
+      info = psb_err_internal_error_
+      call psb_errpush(info,name,a_err='Coarray events allocation')
+      goto 9999
+    end if
+  else
+    if (last_clear_count>0) &
+         & event wait(clear,until_count=last_clear_count)
+  end if
+
+  if (psb_size(buffer) < xchg%max_buffer_size) then
+    !
+    ! By construction, max_buffer_size was computed with a collective. 
+    !
+    if (allocated(buffer)) deallocate(buffer)
+    !write(*,*) 'Allocating buffer',xchg%max_buffer_size
+    allocate(buffer(xchg%max_buffer_size)[*],stat=info)
+    if (info == 0) allocate(sndbuf(xchg%max_buffer_size),stat=info)
+    if (info /= 0) then
+      info = psb_err_internal_error_
+      call psb_errpush(info,name,a_err='Coarray buffer allocation')
+      goto 9999
+    end if
+  end if
+
+  if (.true.) then 
+    !sync all
+    nxch = size(xchg%prcs_xch)
+    myself = this_image()
+    do ip = 1, nxch
+      img = xchg%prcs_xch(ip) + 1
+      p1  = xchg%loc_snd_bnd(ip)
+      p2  = xchg%loc_snd_bnd(ip+1)-1
+      rp1 = xchg%rmt_rcv_bnd(ip,1)
+      rp2 = xchg%rmt_rcv_bnd(ip,2)
+      isz = p2-p1+1
+      !write(0,*) myself,'Posting for ',img,' boundaries: ',p1,p2
+      call y%gth(isz,xchg%loc_snd_idx(p1:p2),buffer(p1:p2))
+      event post(ufg(myself)[img])
+    end do
+
+    do ip = 1, nxch
+      img = xchg%prcs_xch(ip) + 1
+      event wait(ufg(img))
+      img = xchg%prcs_xch(ip) + 1
+      p1  = xchg%loc_rcv_bnd(ip)
+      p2  = xchg%loc_rcv_bnd(ip+1)-1
+      isz = p2-p1+1
+      rp1 = xchg%rmt_snd_bnd(ip,1)
+      rp2 = xchg%rmt_snd_bnd(ip,2)
+      !write(0,*) myself,'Getting from ',img,'Remote boundaries: ',rp1,rp2
+      call y%sct(isz,xchg%loc_rcv_idx(p1:p2),buffer(rp1:rp2)[img],beta)
+      event post(clear[img])
+    end do
+    last_clear_count = nxch
+
+  else
+
+    nxch = size(xchg%prcs_xch)
+    myself = this_image()
+    do ip = 1, nxch
+      img = xchg%prcs_xch(ip) + 1
+      p1  = xchg%loc_snd_bnd(ip)
+      p2  = xchg%loc_snd_bnd(ip+1)-1
+      rp1 = xchg%rmt_rcv_bnd(ip,1)
+      rp2 = xchg%rmt_rcv_bnd(ip,2)
+      isz = p2-p1+1
+      !write(0,*) myself,'Posting for ',img,' boundaries: ',rp1,rp2
+      if (.false.) then 
+        call y%gth(isz,xchg%loc_snd_idx(p1:p2),buffer(rp1:rp2)[img])
+      else
+        call y%gth(isz,xchg%loc_snd_idx(p1:p2),sndbuf(p1:p2))
+        buffer(rp1:rp2)[img] = sndbuf(p1:p2)
+      end if
+    end do
+    !
+    ! Doing event post later should provide more opportunities for
+    ! overlap
+    ! 
+    do ip= 1, nxch
+      img = xchg%prcs_xch(ip) + 1
+      event post(ufg(myself)[img])
+    end do
+
+    do ip = 1, nxch
+      img = xchg%prcs_xch(ip) + 1
+      event wait(ufg(img))
+      img = xchg%prcs_xch(ip) + 1
+      p1  = xchg%loc_rcv_bnd(ip)
+      p2  = xchg%loc_rcv_bnd(ip+1)-1
+      isz = p2-p1+1
+      rp1 = xchg%rmt_snd_bnd(ip,1)
+      rp2 = xchg%rmt_snd_bnd(ip,2)
+      !write(0,*) myself,'Getting from ',img,' boundaries: ',p1,p2
+      call y%sct(isz,xchg%loc_rcv_idx(p1:p2),buffer(p1:p2),beta)
+      event post(clear[img])
+    end do
+
+    last_clear_count = nxch
+
+  end if
+
+  call psb_erractionrestore(err_act)
+  return
+
+9999 call psb_error_handler(ictxt,err_act)
+
+  return
+
+end subroutine psi_dswap_xchg_vect
+
+

 !
 !
--- a/base/internals/psi_crea_index.f90
+++ b/base/internals/psi_crea_index.f90
@ -124,7 +124,7 @@ subroutine psi_crea_index(desc_a,index_in,index_out,glob_idx,nxch,nsnd,nrcv,info

  call psi_dl_check(dep_list,max(1,dl_lda),np,length_dl)

-  ! ....now i can sort dependency lists.
+  ! ....now I can sort dependency lists.
  call psi_sort_dl(dep_list,length_dl,np,info)
  if(info /= psb_success_) then
    call psb_errpush(psb_err_from_subroutine_,name,a_err='psi_sort_dl')
--- a/base/internals/psi_desc_impl.f90
+++ b/base/internals/psi_desc_impl.f90
@ -60,7 +60,99 @@ subroutine psi_renum_index(iperm,idx,info)

 end subroutine psi_renum_index

-subroutine psi_cnv_dsc(halo_in,ovrlap_in,ext_in,cdesc, info, mold)
+subroutine psi_cnv_v2xch(ictxt, vidx_in, xch_idx,info)
+  use psi_mod, psi_protect_name =>  psi_cnv_v2xch
+  use psb_realloc_mod
+  implicit none
+  integer(psb_ipk_), intent(in)         :: ictxt, vidx_in(:)
+  type(psb_xch_idx_type), intent(inout) :: xch_idx
+  integer(psb_ipk_), intent(out)        :: info
+
+  !     ....local scalars....      
+  integer(psb_ipk_) :: np, me
+  integer(psb_ipk_) :: err_act
+  integer(psb_ipk_) :: nxch, nsnd, nrcv, nesd,nerv, ip, j, k, ixch
+  !     ...parameters
+  integer(psb_ipk_) :: debug_level, debug_unit
+  logical, parameter :: debug=.false.
+  character(len=20)  :: name
+
+  name='psi_cnv_v2xch'
+  call psb_get_erraction(err_act)
+  debug_level = psb_get_debug_level()
+  debug_unit  = psb_get_debug_unit()
+
+  info = psb_success_
+  
+  call psb_info(ictxt,me,np)
+  if (np == -1) then
+    info = psb_err_context_error_
+    call psb_errpush(info,name)
+    goto 9999
+  endif
+
+
+  call psb_get_xch_idx(vidx_in, nxch, nsnd, nrcv)
+  xch_idx%max_buffer_size = max(nsnd,nrcv)
+  call psb_amx(ictxt,xch_idx%max_buffer_size)
+  if (info == 0) call psb_realloc(nxch,xch_idx%prcs_xch,info)
+  if (info == 0) call psb_realloc(nxch,2,xch_idx%rmt_snd_bnd,info)
+  if (info == 0) call psb_realloc(nxch,2,xch_idx%rmt_rcv_bnd,info)
+  if (info == 0) call psb_realloc(nxch+1,xch_idx%loc_snd_bnd,info)
+  if (info == 0) call psb_realloc(nxch+1,xch_idx%loc_rcv_bnd,info)
+  if (info == 0) call psb_realloc(nsnd,xch_idx%loc_snd_idx,info)
+  if (info == 0) call psb_realloc(nrcv,xch_idx%loc_rcv_idx,info)
+  if (info /= psb_success_) then 
+    info = psb_err_from_subroutine_
+    call psb_errpush(psb_err_from_subroutine_,name,a_err='Allocate')
+    goto 9999      
+  end if
+
+  
+  
+  ip     = 1
+  ixch   = 1
+  xch_idx%loc_snd_bnd(1) = 1
+  xch_idx%loc_rcv_bnd(1) = 1 
+  do 
+    if (ip > size(vidx_in)) then 
+      write(psb_err_unit,*) trim(name),': Warning: out of size of input vector '
+      exit
+    end if
+    if (vidx_in(ip) == -1) exit
+    xch_idx%prcs_xch(ixch) = vidx_in(ip)
+    nerv   = vidx_in(ip+psb_n_elem_recv_)
+!!$    write(*,*) 'Check on receive option ',ip,nerv,xch_idx%loc_rcv_bnd(ixch)
+    xch_idx%loc_rcv_idx(xch_idx%loc_rcv_bnd(ixch):xch_idx%loc_rcv_bnd(ixch)+nerv-1) = &
+         & vidx_in(ip+psb_n_elem_recv_+1:ip+psb_n_elem_recv_+nerv)
+    nesd   = vidx_in(ip+nerv+psb_n_elem_send_)
+!!$    write(*,*) 'Check on send option ',ip,nesd,xch_idx%loc_snd_bnd(ixch)
+    xch_idx%loc_snd_idx(xch_idx%loc_snd_bnd(ixch):xch_idx%loc_snd_bnd(ixch)+nesd-1) = &
+         & vidx_in(ip+nerv+psb_n_elem_send_+1:ip+nerv+psb_n_elem_send_+nesd)
+    xch_idx%loc_rcv_bnd(ixch+1) =  xch_idx%loc_rcv_bnd(ixch) + nerv
+    xch_idx%loc_snd_bnd(ixch+1) =  xch_idx%loc_snd_bnd(ixch) + nesd
+    call psb_snd(ictxt,xch_idx%loc_rcv_bnd(ixch:ixch+1),xch_idx%prcs_xch(ixch))
+    call psb_snd(ictxt,xch_idx%loc_snd_bnd(ixch:ixch+1),xch_idx%prcs_xch(ixch))
+    call psb_rcv(ictxt,xch_idx%rmt_rcv_bnd(ixch,1:2),xch_idx%prcs_xch(ixch))
+    call psb_rcv(ictxt,xch_idx%rmt_snd_bnd(ixch,1:2),xch_idx%prcs_xch(ixch))
+    ip     = ip+nerv+nesd+3
+    ixch   = ixch + 1 
+  end do
+  xch_idx%rmt_rcv_bnd(:,2) = xch_idx%rmt_rcv_bnd(:,2) - 1
+  xch_idx%rmt_snd_bnd(:,2) = xch_idx%rmt_snd_bnd(:,2) - 1
+  
+  
+  call psb_erractionrestore(err_act)
+  return
+
+9999 call psb_error_handler(ictxt,err_act)
+
+  return
+end subroutine psi_cnv_v2xch
+  
+  
+
+subroutine psi_cnv_dsc(halo_in,ovrlap_in,ext_in, cdesc, info, mold)

  use psi_mod, psi_protect_name =>  psi_cnv_dsc
  use psb_realloc_mod
@ -98,6 +190,7 @@ subroutine psi_cnv_dsc(halo_in,ovrlap_in,ext_in,cdesc, info, mold)
    goto 9999
  endif

+  cdesc%max_buffer_size=0

  ! first the halo index
  if (debug_level>0) write(debug_unit,*) me,'Calling crea_index on halo',&
@ -108,7 +201,8 @@ subroutine psi_cnv_dsc(halo_in,ovrlap_in,ext_in,cdesc, info, mold)
    goto 9999
  end if
  call psb_move_alloc(idx_out,cdesc%halo_index,info)
-
+  cdesc%max_buffer_size = max(cdesc%max_buffer_size, nsnd, nrcv)
+  call  psi_cnv_v2xch(ictxt, cdesc%halo_index, cdesc%halo_xch,info)
  if (debug_level>0) write(debug_unit,*) me,'Done crea_index on halo'
  if (debug_level>0) write(debug_unit,*) me,'Calling crea_index on ext'

@ -121,7 +215,9 @@ subroutine psi_cnv_dsc(halo_in,ovrlap_in,ext_in,cdesc, info, mold)
    goto 9999
  end if
  call psb_move_alloc(idx_out,cdesc%ext_index,info)
-
+  cdesc%max_buffer_size = max(cdesc%max_buffer_size, nsnd, nrcv)
+  call  psi_cnv_v2xch(ictxt, cdesc%ext_index, cdesc%ext_xch,info)
+  
  if (debug_level>0) write(debug_unit,*) me,'Done crea_index on ext'
  if (debug_level>0) write(debug_unit,*) me,'Calling crea_index on ovrlap'

@ -132,6 +228,8 @@ subroutine psi_cnv_dsc(halo_in,ovrlap_in,ext_in,cdesc, info, mold)
    goto 9999
  end if
  call psb_move_alloc(idx_out,cdesc%ovrlap_index,info)
+  cdesc%max_buffer_size = max(cdesc%max_buffer_size, nsnd, nrcv)
+  call  psi_cnv_v2xch(ictxt, cdesc%ovrlap_index, cdesc%ovrlap_xch,info)
  if (info /= psb_success_) then
    call psb_errpush(psb_err_from_subroutine_,name,a_err='psb_move_alloc')
    goto 9999
@ -157,6 +255,8 @@ subroutine psi_cnv_dsc(halo_in,ovrlap_in,ext_in,cdesc, info, mold)
    goto 9999
  end if
  call psb_move_alloc(idx_out,cdesc%ovr_mst_idx,info)
+  cdesc%max_buffer_size = max(cdesc%max_buffer_size, nsnd, nrcv)
+  call  psi_cnv_v2xch(ictxt, cdesc%ovr_mst_idx, cdesc%ovr_mst_xch,info)
  if (info /= psb_success_) then
    call psb_errpush(psb_err_from_subroutine_,name,a_err='psb_move_alloc')
    goto 9999
@ -171,7 +271,7 @@ subroutine psi_cnv_dsc(halo_in,ovrlap_in,ext_in,cdesc, info, mold)
  call cdesc%v_ovrlap_index%bld(cdesc%ovrlap_index,mold=mold)
  call cdesc%v_ovr_mst_idx%bld(cdesc%ovr_mst_idx,mold=mold)

-
+  call psb_amx(ictxt,cdesc%max_buffer_size)
  if (info /= psb_success_) then
    call psb_errpush(psb_err_from_subroutine_,name,a_err='psi_crea_bnd_elem')
    goto 9999
--- a/base/modules/desc/psb_desc_mod.F90
+++ b/base/modules/desc/psb_desc_mod.F90
@ -197,7 +197,19 @@ module psb_desc_mod
  !
  !
  !
-
+  type psb_xch_idx_type
+    integer(psb_ipk_), allocatable :: prcs_xch(:)
+    integer(psb_ipk_), allocatable :: rmt_snd_bnd(:,:)
+    integer(psb_ipk_), allocatable :: rmt_rcv_bnd(:,:)
+    integer(psb_ipk_), allocatable :: loc_rcv_bnd(:)
+    integer(psb_ipk_), allocatable :: loc_snd_bnd(:)
+    integer(psb_ipk_), allocatable :: loc_rcv_idx(:)
+    integer(psb_ipk_), allocatable :: loc_snd_idx(:)
+    integer(psb_ipk_)  :: max_buffer_size=0
+  contains
+    procedure, pass(xchg) :: sizeof => psb_xch_idx_sizeof
+    procedure, pass(xchg) :: print  => psb_xch_idx_print
+  end type psb_xch_idx_type

  type psb_desc_type
    class(psb_indx_map), allocatable :: indxmap
@ -212,18 +224,22 @@ module psb_desc_mod
    type(psb_i_vect_type)            :: v_ovrlap_index
    type(psb_i_vect_type)            :: v_ovr_mst_idx 

+    type(psb_xch_idx_type)           :: halo_xch
+    type(psb_xch_idx_type)           :: ext_xch
+    type(psb_xch_idx_type)           :: ovrlap_xch
+    type(psb_xch_idx_type)           :: ovr_mst_xch
+
+    
    integer(psb_ipk_), allocatable   :: ovrlap_elem(:,:)
    integer(psb_ipk_), allocatable   :: bnd_elem(:)
    integer(psb_ipk_), allocatable   :: lprm(:)
-    !type(psb_desc_type), pointer     :: base_desc => null()
+    type(psb_desc_type), pointer     :: base_desc => null()
    integer(psb_ipk_), allocatable   :: idx_space(:)

    !
    ! Test a coarray implementation
    !
-    !type(event_type), allocatable  :: up_for_grabs(:)[:]
-    real(psb_dpk_), allocatable    :: d_send_buf(:)[:]
-    integer(psb_ipk_), allocatable :: grab_idxes(:,:)
+    integer(psb_ipk_)              :: max_buffer_size
  contains
    procedure, pass(desc) :: is_ok           => psb_is_ok_desc
    procedure, pass(desc) :: is_valid        => psb_is_valid_desc
@ -241,9 +257,10 @@ module psb_desc_mod
    procedure, pass(desc) :: get_global_rows => psb_cd_get_global_rows
    procedure, pass(desc) :: get_global_cols => psb_cd_get_global_cols
    procedure, pass(desc) :: get_global_indices => psb_cd_get_global_indices
+    procedure, pass(desc) :: x_get_list      => psb_cd_x_get_list
    procedure, pass(desc) :: a_get_list      => psb_cd_get_list
    procedure, pass(desc) :: v_get_list      => psb_cd_v_get_list
-    generic, public       :: get_list => a_get_list, v_get_list
+    generic, public       :: get_list => a_get_list, v_get_list, x_get_list
    procedure, pass(desc) :: sizeof          => psb_cd_sizeof
    procedure, pass(desc) :: clone           => psb_cd_clone
    procedure, pass(desc) :: cnv             => psb_cd_cnv
@ -319,13 +336,73 @@ contains
    val = val + psb_sizeof_int*psb_size(desc%lprm)
    val = val + psb_sizeof_int*psb_size(desc%idx_space)
    if (allocated(desc%indxmap))  val = val + desc%indxmap%sizeof()
+
    val = val + desc%v_halo_index%sizeof()
    val = val + desc%v_ext_index%sizeof()
    val = val + desc%v_ovrlap_index%sizeof()
    val = val + desc%v_ovr_mst_idx%sizeof()

+    val = val + desc%halo_xch%sizeof()  
+    val = val + desc%ext_xch%sizeof()  
+    val = val + desc%ovrlap_xch%sizeof()
+    val = val + desc%ovr_mst_xch%sizeof()  
+    
  end function psb_cd_sizeof

+  function psb_xch_idx_sizeof(xchg)  result(val)
+    implicit none
+    !....Parameters...
+
+    class(psb_xch_idx_type), intent(in) :: xchg
+    integer(psb_long_int_k_) :: val
+
+    val = 0 
+    val = val + psb_sizeof_int*psb_size(xchg%prcs_xch)
+    val = val + psb_sizeof_int*psb_size(xchg%rmt_snd_bnd)
+    val = val + psb_sizeof_int*psb_size(xchg%rmt_rcv_bnd)
+    val = val + psb_sizeof_int*psb_size(xchg%loc_rcv_bnd)
+    val = val + psb_sizeof_int*psb_size(xchg%loc_snd_bnd)
+    val = val + psb_sizeof_int*psb_size(xchg%loc_rcv_idx)
+    val = val + psb_sizeof_int*psb_size(xchg%loc_snd_idx)
+
+  end function psb_xch_idx_sizeof
+
+  subroutine psb_xch_idx_print(iout,xchg)
+    implicit none
+    !....Parameters...
+
+    class(psb_xch_idx_type), intent(in) :: xchg
+    integer(psb_ipk_), intent(in) :: iout
+    integer(psb_ipk_) :: nxch, ip
+
+    write(iout,*) ' xch_idx printout'
+    write(iout,*) ' max buffer', xchg%max_buffer_size
+    nxch = psb_size(xchg%prcs_xch)
+    write(iout,*) ' number of exchanges ',nxch
+    if (nxch > 0) then
+      if (allocated(xchg%rmt_snd_bnd)) then
+        write(iout,*) ' remote sends '
+        do ip=1,nxch
+          write(iout,*) xchg%prcs_xch(ip),xchg%rmt_snd_bnd(ip,1:2)
+        end do
+      end if
+      if (allocated(xchg%rmt_rcv_bnd)) then
+        write(iout,*) ' remote recvs '
+        do ip=1,nxch
+          write(iout,*) xchg%prcs_xch(ip),xchg%rmt_rcv_bnd(ip,1:2)
+        end do
+      end if
+      if (allocated( xchg%loc_snd_bnd).and.allocated(xchg%loc_snd_idx)) then
+        write(iout,*) ' local sends '
+        do ip=1,nxch
+          write(iout,*) xchg%prcs_xch(ip),xchg%loc_snd_bnd(ip:ip+1)
+          write(iout,*) xchg%loc_snd_idx(xchg%loc_snd_bnd(ip):xchg%loc_snd_bnd(ip+1)-1)
+        end do
+      end if
+    end if
+    
+  end subroutine psb_xch_idx_print
+    


  subroutine psb_cd_set_large_threshold(ith)
@ -366,7 +443,7 @@ contains
    type(psb_desc_type), intent(inout) :: desc
    ! We have nothing left to do here.
    ! Perhaps we should delete this subroutine? 
-    !nullify(desc%base_desc)
+    nullify(desc%base_desc)

  end subroutine psb_nullify_desc

@ -375,7 +452,7 @@ contains
    class(psb_desc_type), intent(inout) :: desc
    ! We have nothing left to do here.
    ! Perhaps we should delete this subroutine? 
-    !nullify(desc%base_desc)
+    nullify(desc%base_desc)

  end subroutine nullify_desc

@ -692,11 +769,11 @@ contains
    case(psb_comm_ext_) 
      ipnt   => desc%ext_index
      if (debug_level >= psb_debug_ext_) then
-!!$        if (.not.associated(desc%base_desc)) then
-!!$          write(debug_unit,*) trim(name),&
-!!$               & ': Warning: trying to get ext_index on a descriptor ',&
-!!$               & 'which does not have a base_desc!'
-!!$        end if
+        if (.not.associated(desc%base_desc)) then
+          write(debug_unit,*) trim(name),&
+               & ': Warning: trying to get ext_index on a descriptor ',&
+               & 'which does not have a base_desc!'
+        end if
        if (.not.psb_is_ovl_desc(desc)) then
          write(debug_unit,*) trim(name),&
               & ': Warning: trying to get ext_index on a descriptor ',&
@ -762,11 +839,11 @@ contains
      if (.not.allocated(desc%v_ext_index%v)) &
           & info = psb_err_inconsistent_index_lists_
      if (debug_level >= psb_debug_ext_) then
-!!$        if (.not.associated(desc%base_desc)) then
-!!$          write(debug_unit,*) trim(name),&
-!!$               & ': Warning: trying to get ext_index on a descriptor ',&
-!!$               & 'which does not have a base_desc!'
-!!$        end if
+        if (.not.associated(desc%base_desc)) then
+          write(debug_unit,*) trim(name),&
+               & ': Warning: trying to get ext_index on a descriptor ',&
+               & 'which does not have a base_desc!'
+        end if
        if (.not.psb_is_ovl_desc(desc)) then
          write(debug_unit,*) trim(name),&
               & ': Warning: trying to get ext_index on a descriptor ',&
@ -798,6 +875,71 @@ contains

  end subroutine psb_cd_v_get_list

+  subroutine psb_cd_x_get_list(data,desc,ipnt,info)
+    use psb_const_mod
+    use psb_error_mod
+    use psb_penv_mod
+    implicit none
+    integer(psb_ipk_), intent(in)     :: data
+    class(psb_xch_idx_type), pointer  :: ipnt
+    class(psb_desc_type), target      :: desc
+    integer(psb_ipk_), intent(out)    :: info
+
+    !locals
+    integer(psb_ipk_) :: np,me,ictxt,err_act, debug_level,debug_unit
+    logical, parameter  :: debug=.false.,debugprt=.false.
+    character(len=20), parameter  :: name='psb_cd_v_get_list'
+
+    info = psb_success_
+    call psb_erractionsave(err_act)
+    debug_unit  = psb_get_debug_unit()
+    debug_level = psb_get_debug_level()
+
+    ictxt = psb_cd_get_context(desc)
+    
+    call psb_info(ictxt, me, np)
+
+    select case(data) 
+    case(psb_comm_halo_) 
+      ipnt   => desc%halo_xch
+    case(psb_comm_ovr_) 
+      ipnt   => desc%ovrlap_xch
+    case(psb_comm_ext_) 
+      ipnt   => desc%ext_xch
+      if (debug_level >= psb_debug_ext_) then
+        if (.not.associated(desc%base_desc)) then
+          write(debug_unit,*) trim(name),&
+               & ': Warning: trying to get ext_index on a descriptor ',&
+               & 'which does not have a base_desc!'
+        end if
+        if (.not.psb_is_ovl_desc(desc)) then
+          write(debug_unit,*) trim(name),&
+               & ': Warning: trying to get ext_index on a descriptor ',&
+               & 'which is not overlap-extended!'
+        end if
+      end if
+    case(psb_comm_mov_) 
+      ipnt   => desc%ovr_mst_xch
+      
+    case default
+      info=psb_err_from_subroutine_
+    end select
+    if (info /= psb_success_) then
+      call psb_errpush(info,name,a_err='wrong Data selector')
+      goto 9999
+    end if
+    
+
+
+    call psb_erractionrestore(err_act)
+    return
+
+9999 call psb_error_handler(err_act)
+
+    return
+
+  end subroutine psb_cd_x_get_list
+
  !
  ! Subroutine: psb_cdfree
  !   Frees a descriptor data structure.
@ -1023,7 +1165,7 @@ contains
        goto 9999
      endif

-!!$      desc_out%base_desc  => desc%base_desc
+      desc_out%base_desc  => desc%base_desc
      if (info == psb_success_)&
           & call psb_safe_ab_cpy(desc%halo_index,desc_out%halo_index,info)
      if (info == psb_success_)&
--- a/base/modules/psi_d_mod.f90
+++ b/base/modules/psi_d_mod.f90
@ -30,7 +30,7 @@
 !!$ 
 !!$  
 module psi_d_mod
-  use psb_desc_mod, only : psb_desc_type, psb_ipk_, psb_dpk_, psb_i_base_vect_type
+  use psb_desc_mod, only : psb_desc_type, psb_ipk_, psb_dpk_, psb_i_base_vect_type, psb_xch_idx_type
  use psb_d_base_vect_mod, only : psb_d_base_vect_type 
  use psb_d_base_multivect_mod, only : psb_d_base_multivect_type 

@ -92,6 +92,14 @@ module psi_d_mod
      real(psb_dpk_),target :: work(:)
      integer(psb_ipk_), intent(in)      :: idx(:),totxch,totsnd,totrcv
    end subroutine psi_dswapidxv
+    subroutine psi_dswap_xchg_vect(iictxt,iicomm,flag,beta,y,xchg,info)
+      import 
+      integer(psb_ipk_), intent(in)          :: iictxt,iicomm,flag
+      integer(psb_ipk_), intent(out)         :: info
+      class(psb_d_base_vect_type)            :: y
+      real(psb_dpk_)                         :: beta
+      class(psb_xch_idx_type), intent(inout) :: xchg
+    end subroutine psi_dswap_xchg_vect
    subroutine psi_dswap_vidx_vect(iictxt,iicomm,flag,beta,y,idx,&
         & totxch,totsnd,totrcv,work,info)
      import 
--- a/base/modules/psi_i_mod.f90
+++ b/base/modules/psi_i_mod.f90
@ -30,7 +30,7 @@
 !!$ 
 !!$  
 module psi_i_mod
-  use psb_desc_mod, only : psb_desc_type, psb_ipk_, psb_mpik_
+  use psb_desc_mod, only : psb_desc_type, psb_ipk_, psb_mpik_, psb_xch_idx_type
  use psb_i_base_vect_mod, only : psb_i_base_vect_type 
  use psb_i_base_multivect_mod, only : psb_i_base_multivect_type 

@ -52,6 +52,16 @@ module psi_i_mod
    end subroutine psi_crea_bnd_elem
  end interface

+  interface
+    subroutine psi_cnv_v2xch(ictxt, vidx_in, xch_idx,info)
+      import
+      integer(psb_ipk_), intent(in)         :: ictxt, vidx_in(:)
+      type(psb_xch_idx_type), intent(inout) :: xch_idx
+      integer(psb_ipk_), intent(out)        :: info
+    end subroutine psi_cnv_v2xch
+  end interface
+  
+
  interface
    subroutine psi_crea_index(desc_a,index_in,index_out,glob_idx,nxch,nsnd,nrcv,info)
      import 
--- a/base/modules/psi_penv_mod.F90
+++ b/base/modules/psi_penv_mod.F90
@ -410,7 +410,7 @@ contains
    if (present(close)) then 
      close_ = close
    else
-      close_ = .true.
+      close_ = .false.
    end if
 ! !$    if (close_) call psb_rsb_exit(info)
 ! !$    if (info.ne.psb_rsb_const_success) then 
--- a/test/pargen/ppde3d.f90
+++ b/test/pargen/ppde3d.f90
@ -175,6 +175,7 @@ program ppde3d
    write(*,*) 'Welcome to PSBLAS version: ',psb_version_string_
    write(*,*) 'This is the ',trim(name),' sample program'
  end if
+  !write(*,*) 'Check on image info:',iam,this_image()
  !
  !  get parameters
  !
@ -197,6 +198,13 @@ program ppde3d
  end if
  if (iam == psb_root_) write(psb_out_unit,'("Overall matrix creation time : ",es12.5)')t2
  if (iam == psb_root_) write(psb_out_unit,'(" ")')
+  if (iam == psb_root_) write(psb_err_unit,'("Check on new descriptor entries")')
+!!$  do i = 0, np -1
+!!$    if (iam == i) call desc_a%halo_xch%print(psb_err_unit)
+!!$    if (iam == i) flush(psb_err_unit)
+!!$    call psb_barrier(ictxt)
+!!$  end do
+  
  !
  !  prepare the preconditioner.
  !  
--- a/test/pargen/runs/ppde.inp
+++ b/test/pargen/runs/ppde.inp
@ -2,9 +2,9 @@
 BICGSTAB               Iterative method BICGSTAB CGS  BICG BICGSTABL RGMRES
 BJAC                   Preconditioner NONE  DIAG  BJAC 
 CSR                    Storage format for matrix A:  CSR COO JAD 
-040                    Domain size (acutal system is this**3)
+100                    Domain size (acutal system is this**3)
 2                      Stopping criterion
-1000                   MAXIT
+0404                   MAXIT
 -1                     ITRACE
 002                    IRST    restart for RGMRES  and BiCGSTABL