diff --git a/base/comm/psb_dhalo.f90 b/base/comm/psb_dhalo.f90 index ac9551c6..9cd13b22 100644 --- a/base/comm/psb_dhalo.f90 +++ b/base/comm/psb_dhalo.f90 @@ -49,6 +49,7 @@ ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psb_dhalom(x,desc_a,info,alpha,jx,ik,work,tran,mode,data) @@ -274,6 +275,7 @@ end subroutine psb_dhalom ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psb_dhalov(x,desc_a,info,alpha,work,tran,mode,data) diff --git a/base/comm/psb_ihalo.f90 b/base/comm/psb_ihalo.f90 index ce95f48f..c6fba0bb 100644 --- a/base/comm/psb_ihalo.f90 +++ b/base/comm/psb_ihalo.f90 @@ -50,6 +50,7 @@ ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psb_ihalom(x,desc_a,info,alpha,jx,ik,work,tran,mode,data) @@ -278,6 +279,7 @@ end subroutine psb_ihalom ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psb_ihalov(x,desc_a,info,alpha,work,tran,mode,data) diff --git a/base/comm/psb_zhalo.f90 b/base/comm/psb_zhalo.f90 index 4e13e0f4..c39a8fb3 100644 --- a/base/comm/psb_zhalo.f90 +++ b/base/comm/psb_zhalo.f90 @@ -49,6 +49,7 @@ ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psb_zhalom(x,desc_a,info,alpha,jx,ik,work,tran,mode,data) @@ -275,6 +276,7 @@ end subroutine psb_zhalom ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psb_zhalov(x,desc_a,info,alpha,work,tran,mode,data) diff --git a/base/internals/psi_crea_ovr_elem.f90 b/base/internals/psi_crea_ovr_elem.f90 index 0befdd68..97e4c4b0 100644 --- a/base/internals/psi_crea_ovr_elem.f90 +++ b/base/internals/psi_crea_ovr_elem.f90 @@ -37,9 +37,9 @@ ! See also description in base/modules/psb_desc_type.f90 ! ! Arguments: -! ovr_elem(:) - integer, allocatable Array containing the output list -! desc_a - type(psb_desc_type). The communication descriptor. -! info - integer. return code. +! ovr_elem(:,:) - integer, allocatable Array containing the output list +! desc_a - type(psb_desc_type). The communication descriptor. +! info - integer. return code. ! subroutine psi_crea_ovr_elem(me,desc_overlap,ovr_elem,info) diff --git a/base/internals/psi_dswapdata.F90 b/base/internals/psi_dswapdata.F90 index a7cde9d6..0470e5de 100644 --- a/base/internals/psi_dswapdata.F90 +++ b/base/internals/psi_dswapdata.F90 @@ -77,6 +77,7 @@ ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! ! @@ -169,6 +170,12 @@ subroutine psi_dswapdatam(flag,n,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default call psb_errpush(4010,name,a_err='wrong Data selector') goto 9999 @@ -551,7 +558,7 @@ end subroutine psi_dswapdatam ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index -! +! psb_comm_mov_ use ovr_mst_idx ! subroutine psi_dswapdatav(flag,beta,y,desc_a,work,info,data) @@ -646,6 +653,12 @@ subroutine psi_dswapdatav(flag,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default call psb_errpush(4010,name,a_err='wrong Data selector') goto 9999 diff --git a/base/internals/psi_dswaptran.F90 b/base/internals/psi_dswaptran.F90 index 8cf3bb91..3e5a3ee6 100644 --- a/base/internals/psi_dswaptran.F90 +++ b/base/internals/psi_dswaptran.F90 @@ -81,6 +81,7 @@ ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psi_dswaptranm(flag,n,beta,y,desc_a,work,info,data) @@ -172,6 +173,12 @@ subroutine psi_dswaptranm(flag,n,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default call psb_errpush(4010,name,a_err='wrong Data selector') goto 9999 @@ -554,6 +561,7 @@ end subroutine psi_dswaptranm ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psi_dswaptranv(flag,beta,y,desc_a,work,info,data) @@ -649,6 +657,12 @@ subroutine psi_dswaptranv(flag,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default write(0,*) 'Really wrong?? ',data_, psb_comm_halo_, psb_comm_ovr_, psb_comm_ext_ call psb_errpush(4010,name,a_err='wrong Data selector') diff --git a/base/internals/psi_iswapdata.F90 b/base/internals/psi_iswapdata.F90 index fe07e892..25be48d6 100644 --- a/base/internals/psi_iswapdata.F90 +++ b/base/internals/psi_iswapdata.F90 @@ -77,6 +77,7 @@ ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psi_iswapdatam(flag,n,beta,y,desc_a,work,info,data) @@ -168,6 +169,12 @@ subroutine psi_iswapdatam(flag,n,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default call psb_errpush(4010,name,a_err='wrong Data selector') goto 9999 @@ -550,6 +557,7 @@ end subroutine psi_iswapdatam ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psi_iswapdatav(flag,beta,y,desc_a,work,info,data) @@ -645,6 +653,12 @@ subroutine psi_iswapdatav(flag,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default call psb_errpush(4010,name,a_err='wrong Data selector') goto 9999 diff --git a/base/internals/psi_iswaptran.F90 b/base/internals/psi_iswaptran.F90 index 214de31b..e690f0d4 100644 --- a/base/internals/psi_iswaptran.F90 +++ b/base/internals/psi_iswaptran.F90 @@ -81,6 +81,7 @@ ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psi_iswaptranm(flag,n,beta,y,desc_a,work,info,data) @@ -171,6 +172,12 @@ subroutine psi_iswaptranm(flag,n,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default call psb_errpush(4010,name,a_err='wrong Data selector') goto 9999 @@ -553,6 +560,7 @@ end subroutine psi_iswaptranm ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psi_iswaptranv(flag,beta,y,desc_a,work,info,data) @@ -646,6 +654,12 @@ subroutine psi_iswaptranv(flag,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default call psb_errpush(4010,name,a_err='wrong Data selector') goto 9999 diff --git a/base/internals/psi_zswapdata.F90 b/base/internals/psi_zswapdata.F90 index cea78147..4248fdf3 100644 --- a/base/internals/psi_zswapdata.F90 +++ b/base/internals/psi_zswapdata.F90 @@ -77,6 +77,7 @@ ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psi_zswapdatam(flag,n,beta,y,desc_a,work,info,data) @@ -168,6 +169,12 @@ subroutine psi_zswapdatam(flag,n,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default call psb_errpush(4010,name,a_err='wrong Data selector') goto 9999 @@ -550,6 +557,7 @@ end subroutine psi_zswapdatam ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psi_zswapdatav(flag,beta,y,desc_a,work,info,data) @@ -645,6 +653,12 @@ subroutine psi_zswapdatav(flag,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default call psb_errpush(4010,name,a_err='wrong Data selector') goto 9999 diff --git a/base/internals/psi_zswaptran.F90 b/base/internals/psi_zswaptran.F90 index 49539996..d427f845 100644 --- a/base/internals/psi_zswaptran.F90 +++ b/base/internals/psi_zswaptran.F90 @@ -81,6 +81,7 @@ ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psi_zswaptranm(flag,n,beta,y,desc_a,work,info,data) @@ -172,6 +173,12 @@ subroutine psi_zswaptranm(flag,n,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default call psb_errpush(4010,name,a_err='wrong Data selector') goto 9999 @@ -554,6 +561,7 @@ end subroutine psi_zswaptranm ! psb_comm_halo_ use halo_index ! psb_comm_ext_ use ext_index ! psb_comm_ovrl_ use ovrl_index +! psb_comm_mov_ use ovr_mst_idx ! ! subroutine psi_zswaptranv(flag,beta,y,desc_a,work,info,data) @@ -648,6 +656,12 @@ subroutine psi_zswaptranv(flag,beta,y,desc_a,work,info,data) totxch = desc_a%matrix_data(psb_text_xch_) idxr = desc_a%matrix_data(psb_text_rcv_) idxs = desc_a%matrix_data(psb_text_snd_) + + case(psb_comm_mov_) + d_idx => desc_a%ovr_mst_idx + totxch = desc_a%matrix_data(psb_tmov_xch_) + idxr = desc_a%matrix_data(psb_tmov_rcv_) + idxs = desc_a%matrix_data(psb_tmov_snd_) case default call psb_errpush(4010,name,a_err='wrong Data selector') goto 9999 diff --git a/base/modules/psb_desc_type.f90 b/base/modules/psb_desc_type.f90 index 7471dd05..0405af86 100644 --- a/base/modules/psb_desc_type.f90 +++ b/base/modules/psb_desc_type.f90 @@ -46,13 +46,15 @@ module psb_descriptor_type ! For overlap update. integer, parameter :: psb_none_=0, psb_sum_=1 integer, parameter :: psb_avg_=2, psb_square_root_=3 - integer, parameter :: psb_zero_=999 + integer, parameter :: psb_setzero_=4 + ! The following are bit fields. integer, parameter :: psb_swap_send_=1, psb_swap_recv_=2 integer, parameter :: psb_swap_sync_=4, psb_swap_mpi_=8 integer, parameter :: psb_no_comm_=-1 - integer, parameter :: psb_comm_halo_=1, psb_comm_ovr_=2, psb_comm_ext_=3 + integer, parameter :: psb_comm_halo_=1, psb_comm_ovr_=2 + integer, parameter :: psb_comm_ext_=3, psb_comm_mov_=4 integer, parameter :: psb_ovt_xhal_ = 123, psb_ovt_asov_=psb_ovt_xhal_+1 ! @@ -71,7 +73,10 @@ module psb_descriptor_type integer, parameter :: psb_text_xch_=17 integer, parameter :: psb_text_snd_=18 integer, parameter :: psb_text_rcv_=19 - integer, parameter :: psb_mdata_size_=20 + integer, parameter :: psb_tmov_xch_=20 + integer, parameter :: psb_tmov_snd_=21 + integer, parameter :: psb_tmov_rcv_=22 + integer, parameter :: psb_mdata_size_=24 integer, parameter :: psb_desc_asb_=3099 integer, parameter :: psb_desc_bld_=psb_desc_asb_+1 integer, parameter :: psb_desc_repl_=3199 @@ -111,6 +116,7 @@ module psb_descriptor_type !| integer, allocatable :: bnd_elem(:) !| integer, allocatable :: ovrlap_index(:) !| integer, allocatable :: ovrlap_elem(:,:) + !| integer, allocatable :: ovr_mst_idx(:) !| integer, allocatable :: loc_to_glob(:) !| integer, allocatable :: glob_to_loc (:) !| integer, allocatable :: hashv(:), glb_lc(:,:), ptree(:) @@ -253,7 +259,14 @@ module psb_descriptor_type ! it only contains the end-of-list marker -1). ! ! 10. ovrlap_elem contains a list of overlap indices together with their degree - ! of overlap, i.e. the number of processes "owning" them. + ! of overlap, i.e. the number of processes "owning" the, and the "master" + ! process whose value has to be considered authoritative when the need arises. + ! + ! 11. ovr_mst_idx is a list defining a retrieve of a copy of the values for + ! overlap entries from their respecitve "master" processes by means of + ! an halo exchange call. This is used for those cases where there is + ! an overlap in the base data distribution. + ! ! It is complex, but it does the following: ! 1. Allows a purely local matrix/stencil buildup phase, requiring only ! one synch point at the end (CDASB) @@ -272,6 +285,7 @@ module psb_descriptor_type integer, allocatable :: bnd_elem(:) integer, allocatable :: ovrlap_index(:) integer, allocatable :: ovrlap_elem(:,:) + integer, allocatable :: ovr_mst_idx(:) integer, allocatable :: loc_to_glob(:) integer, allocatable :: glob_to_loc (:) integer, allocatable :: hashv(:), glb_lc(:,:), ptree(:) @@ -305,6 +319,7 @@ contains if (allocated(desc%bnd_elem)) val = val + 4*size(desc%bnd_elem) if (allocated(desc%ovrlap_index)) val = val + 4*size(desc%ovrlap_index) if (allocated(desc%ovrlap_elem)) val = val + 4*size(desc%ovrlap_elem) + if (allocated(desc%ovr_mst_idx)) val = val + 4*size(desc%ovr_mst_idx) if (allocated(desc%loc_to_glob)) val = val + 4*size(desc%loc_to_glob) if (allocated(desc%glob_to_loc)) val = val + 4*size(desc%glob_to_loc) if (allocated(desc%hashv)) val = val + 4*size(desc%hashv) diff --git a/base/modules/psi_mod.f90 b/base/modules/psi_mod.f90 index eb00a9d0..9ac4a5c9 100644 --- a/base/modules/psi_mod.f90 +++ b/base/modules/psi_mod.f90 @@ -377,7 +377,7 @@ contains integer :: np,me integer :: ictxt, err_act,nxch,nsnd,nrcv,j,k ! ...local array... - integer, allocatable :: idx_out(:) + integer, allocatable :: idx_out(:), tmp_mst_idx(:) ! ...parameters integer :: debug_level, debug_unit @@ -403,7 +403,7 @@ contains ! first the halo index if (debug_level>0) write(debug_unit,*) me,'Calling crea_index on halo' call psi_crea_index(cdesc,halo_in, idx_out,.false.,nxch,nsnd,nrcv,info) - if(info /= 0) then + if (info /= 0) then call psb_errpush(4010,name,a_err='psi_crea_index') goto 9999 end if @@ -419,7 +419,7 @@ contains ! then ext index if (debug_level>0) write(debug_unit,*) me,'Calling crea_index on ext' call psi_crea_index(cdesc,ext_in, idx_out,.false.,nxch,nsnd,nrcv,info) - if(info /= 0) then + if (info /= 0) then call psb_errpush(4010,name,a_err='psi_crea_index') goto 9999 end if @@ -433,12 +433,12 @@ contains ! then the overlap index call psi_crea_index(cdesc,ovrlap_in, idx_out,.true.,nxch,nsnd,nrcv,info) - if(info /= 0) then + if (info /= 0) then call psb_errpush(4010,name,a_err='psi_crea_index') goto 9999 end if call psb_transfer(idx_out,cdesc%ovrlap_index,info) - if(info /= 0) then + if (info /= 0) then call psb_errpush(4010,name,a_err='psb_transfer') goto 9999 end if @@ -447,20 +447,39 @@ contains cdesc%matrix_data(psb_tovr_snd_) = nsnd cdesc%matrix_data(psb_tovr_rcv_) = nrcv - if (debug_level>0) write(debug_unit,*) me,'Calling crea_ovr_elem' ! next ovrlap_elem + if (debug_level>0) write(debug_unit,*) me,'Calling crea_ovr_elem' call psi_crea_ovr_elem(me,cdesc%ovrlap_index,cdesc%ovrlap_elem,info) if (debug_level>0) write(debug_unit,*) me,'Done crea_ovr_elem' - if(info /= 0) then + if (info /= 0) then call psb_errpush(4010,name,a_err='psi_crea_ovr_elem') goto 9999 end if + ! Extract ovr_mst_idx from ovrlap_elem + if (debug_level>0) write(debug_unit,*) me,'Calling bld_ovr_mst' + call psi_bld_ovr_mst(me,cdesc%ovrlap_elem,tmp_mst_idx,info) + if (info == 0) call psi_crea_index(cdesc,& + & tmp_mst_idx,idx_out,.false.,nxch,nsnd,nrcv,info) + if (debug_level>0) write(debug_unit,*) me,'Done crea_indx' + if (info /= 0) then + call psb_errpush(4010,name,a_err='psi_bld_ovr_mst') + goto 9999 + end if + call psb_transfer(idx_out,cdesc%ovr_mst_idx,info) + if (info /= 0) then + call psb_errpush(4010,name,a_err='psb_transfer') + goto 9999 + end if + + cdesc%matrix_data(psb_tmov_xch_) = nxch + cdesc%matrix_data(psb_tmov_snd_) = nsnd + cdesc%matrix_data(psb_tmov_rcv_) = nrcv ! finally bnd_elem call psi_crea_bnd_elem(idx_out,cdesc,info) if (info == 0) call psb_transfer(idx_out,cdesc%bnd_elem,info) - if(info /= 0) then + if (info /= 0) then call psb_errpush(4010,name,a_err='psi_crea_bnd_elem') goto 9999 end if @@ -618,6 +637,12 @@ contains ndm = desc_a%ovrlap_elem(i,2) x(idx) = x(idx)/real(ndm) end do + case(psb_setzero_) + do i=1,size(desc_a%ovrlap_elem,1) + idx = desc_a%ovrlap_elem(i,1) + if (me /= desc_a%ovrlap_elem(i,3))& + & x(idx) = dzero + end do case(psb_sum_) ! do nothing @@ -684,6 +709,12 @@ contains ndm = desc_a%ovrlap_elem(i,2) x(idx,:) = x(idx,:)/real(ndm) end do + case(psb_setzero_) + do i=1,size(desc_a%ovrlap_elem,1) + idx = desc_a%ovrlap_elem(i,1) + if (me /= desc_a%ovrlap_elem(i,3))& + & x(idx,:) = dzero + end do case(psb_sum_) ! do nothing @@ -749,6 +780,12 @@ contains ndm = desc_a%ovrlap_elem(i,2) x(idx) = x(idx)/real(ndm) end do + case(psb_setzero_) + do i=1,size(desc_a%ovrlap_elem,1) + idx = desc_a%ovrlap_elem(i,1) + if (me /= desc_a%ovrlap_elem(i,3))& + & x(idx) = zzero + end do case(psb_sum_) ! do nothing @@ -815,6 +852,12 @@ contains ndm = desc_a%ovrlap_elem(i,2) x(idx,:) = x(idx,:)/real(ndm) end do + case(psb_setzero_) + do i=1,size(desc_a%ovrlap_elem,1) + idx = desc_a%ovrlap_elem(i,1) + if (me /= desc_a%ovrlap_elem(i,3))& + & x(idx,:) = zzero + end do case(psb_sum_) ! do nothing @@ -881,6 +924,12 @@ contains ndm = desc_a%ovrlap_elem(i,2) x(idx) = x(idx)/real(ndm) end do + case(psb_setzero_) + do i=1,size(desc_a%ovrlap_elem,1) + idx = desc_a%ovrlap_elem(i,1) + if (me /= desc_a%ovrlap_elem(i,3))& + & x(idx) = izero + end do case(psb_sum_) ! do nothing @@ -948,6 +997,12 @@ contains ndm = desc_a%ovrlap_elem(i,2) x(idx,:) = x(idx,:)/real(ndm) end do + case(psb_setzero_) + do i=1,size(desc_a%ovrlap_elem,1) + idx = desc_a%ovrlap_elem(i,1) + if (me /= desc_a%ovrlap_elem(i,3))& + & x(idx,:) = izero + end do case(psb_sum_) ! do nothing @@ -1935,6 +1990,59 @@ contains end do end if end subroutine psi_zsctv + + subroutine psi_bld_ovr_mst(me,ovrlap_elem,mst_idx,info) + use psb_const_mod + use psb_error_mod + use psb_penv_mod + use psb_descriptor_type + use psb_realloc_mod + implicit none + + ! ....scalars parameters.... + integer, intent(in) :: me, ovrlap_elem(:,:) + integer, allocatable, intent(out) :: mst_idx(:) + integer, intent(out) :: info + + integer :: i, j, proc, nov,isz, ip, err_act, idx + character(len=20) :: name + + name='psi_bld_ovr_mst' + call psb_get_erraction(err_act) + + nov = size(ovrlap_elem,1) + isz = 3*nov+1 + call psb_realloc(isz,mst_idx,info) + if (info /= 0) then + call psb_errpush(4001,name,a_err='reallocate') + goto 9999 + end if + mst_idx = -1 + j = 1 + do i=1, nov + proc = ovrlap_elem(i,3) + if (me /= proc) then + idx = ovrlap_elem(i,1) + mst_idx(j+0) = proc + mst_idx(j+1) = 1 + mst_idx(j+2) = idx + j = j + 3 + end if + end do + mst_idx(j) = -1 + + call psb_erractionrestore(err_act) + return + +9999 continue + call psb_erractionrestore(err_act) + + if (err_act == psb_act_abort_) then + call psb_error() + return + end if + return + end subroutine psi_bld_ovr_mst end module psi_mod diff --git a/base/tools/psb_cdcpy.f90 b/base/tools/psb_cdcpy.f90 index b2a65964..5fff9a6a 100644 --- a/base/tools/psb_cdcpy.f90 +++ b/base/tools/psb_cdcpy.f90 @@ -84,6 +84,7 @@ subroutine psb_cdcpy(desc_in, desc_out, info) if (info == 0) call psb_safe_cpy(desc_in%ovrlap_index,desc_out%ovrlap_index,info) if (info == 0) call psb_safe_cpy(desc_in%bnd_elem,desc_out%bnd_elem,info) if (info == 0) call psb_safe_cpy(desc_in%ovrlap_elem,desc_out%ovrlap_elem,info) + if (info == 0) call psb_safe_cpy(desc_in%ovr_mst_idx,desc_out%ovr_mst_idx,info) if (info == 0) call psb_safe_cpy(desc_in%loc_to_glob,desc_out%loc_to_glob,info) if (info == 0) call psb_safe_cpy(desc_in%glob_to_loc,desc_out%glob_to_loc,info) if (info == 0) call psb_safe_cpy(desc_in%lprm,desc_out%lprm,info) diff --git a/base/tools/psb_cdfree.f90 b/base/tools/psb_cdfree.f90 index 8a247d1a..0403ccd4 100644 --- a/base/tools/psb_cdfree.f90 +++ b/base/tools/psb_cdfree.f90 @@ -155,6 +155,14 @@ subroutine psb_cdfree(desc_a,info) end if !deallocate ovrlap_index field + deallocate(desc_a%ovr_mst_idx,stat=info) + if (info /= 0) then + info=2055 + call psb_errpush(info,name) + goto 9999 + end if + + deallocate(desc_a%lprm,stat=info) if (info /= 0) then info=2057 diff --git a/base/tools/psb_cdtransfer.f90 b/base/tools/psb_cdtransfer.f90 index 27e78581..179928aa 100644 --- a/base/tools/psb_cdtransfer.f90 +++ b/base/tools/psb_cdtransfer.f90 @@ -88,6 +88,8 @@ subroutine psb_cdtransfer(desc_in, desc_out, info) & call psb_transfer( desc_in%ovrlap_elem , desc_out%ovrlap_elem , info) if (info == 0) & & call psb_transfer( desc_in%ovrlap_index, desc_out%ovrlap_index , info) + if (info == 0) & + & call psb_transfer( desc_in%ovr_mst_idx , desc_out%ovr_mst_idx , info) if (info == 0) & & call psb_transfer( desc_in%ext_index , desc_out%ext_index , info) if (info == 0) & diff --git a/prec/psb_dbjac_aply.f90 b/prec/psb_dbjac_aply.f90 index 43d4dbee..54a4dbdb 100644 --- a/prec/psb_dbjac_aply.f90 +++ b/prec/psb_dbjac_aply.f90 @@ -122,7 +122,7 @@ subroutine psb_dbjac_aply(alpha,prec,x,beta,y,desc_data,trans,work,info) if(info /=0) goto 9999 end select - + case default info = 4001 @@ -130,6 +130,8 @@ subroutine psb_dbjac_aply(alpha,prec,x,beta,y,desc_data,trans,work,info) goto 9999 end select + call psb_halo(y,desc_data,info,data=psb_comm_mov_) + if (n_col <= size(work)) then if ((4*n_col+n_col) <= size(work)) then else diff --git a/prec/psb_dprc_aply.f90 b/prec/psb_dprc_aply.f90 index 1e6f2c1a..f845a85b 100644 --- a/prec/psb_dprc_aply.f90 +++ b/prec/psb_dprc_aply.f90 @@ -74,9 +74,6 @@ subroutine psb_dprc_aply(prec,x,y,desc_data,info,trans, work) call psb_gprec_aply(done,prec,x,dzero,y,desc_data,trans_,work_,info) - ! If the original distribution has an overlap we should fix that. - call psb_ovrl(y,desc_data,info,update=psb_avg_) - if (present(work)) then else deallocate(work_) diff --git a/prec/psb_zbjac_aply.f90 b/prec/psb_zbjac_aply.f90 index 65d4eda0..136852e6 100644 --- a/prec/psb_zbjac_aply.f90 +++ b/prec/psb_zbjac_aply.f90 @@ -138,6 +138,8 @@ subroutine psb_zbjac_aply(alpha,prec,x,beta,y,desc_data,trans,work,info) goto 9999 end select + call psb_halo(y,desc_data,info,data=psb_comm_mov_) + if (n_col <= size(work)) then if ((4*n_col+n_col) <= size(work)) then else diff --git a/prec/psb_zprc_aply.f90 b/prec/psb_zprc_aply.f90 index 4c862149..d190a050 100644 --- a/prec/psb_zprc_aply.f90 +++ b/prec/psb_zprc_aply.f90 @@ -76,7 +76,6 @@ subroutine psb_zprc_aply(prec,x,y,desc_data,info,trans, work) call psb_gprec_aply(zone,prec,x,zzero,y,desc_data,trans_,work_,info) ! If the original distribution has an overlap we should fix that. - call psb_ovrl(y,desc_data,info,update=psb_avg_) if (present(work)) then else