Further fix for gather.

sfilippone · Nov 3, 2023 · baf18ce · baf18ce
1 parent 5caee55
commit baf18ce
Show file tree

Hide file tree

Showing 7 changed files with 98 additions and 161 deletions.
diff --git a/base/comm/psb_cgather_a.f90 b/base/comm/psb_cgather_a.f90
@@ -60,7 +60,7 @@ subroutine  psb_cgatherm(globx, locx, desc_a, info, iroot)
   type(psb_ctxt_type) :: ctxt
   integer(psb_mpk_) ::  np, me, root, iiroot, icomm, myrank, rootrank
   integer(psb_ipk_) :: ierr(5), err_act, lda_locx, lda_globx, lock, globk,&
-       & maxk, k, jlx, ilx, i, j, loc_rows
+       & maxk, k, jlx, ilx, i, j
   integer(psb_lpk_) :: m, n, ilocx,  jlocx, idx, iglobx, jglobx
   character(len=20) :: name, ch_err
 
@@ -232,11 +232,11 @@ subroutine  psb_cgatherv(globx, locx, desc_a, info, iroot)
 
   ! locals
   type(psb_ctxt_type) :: ctxt
-  integer(psb_mpk_) :: np, me, root, iiroot, icomm, myrank, rootrank, loc_rows
+  integer(psb_mpk_) :: np, me, root, iiroot, icomm, myrank, rootrank
   integer(psb_ipk_) :: ierr(5), err_act, lda_locx, lda_globx, lock, globk,&
        & maxk, k, jlx, ilx, i, j
   integer(psb_lpk_) :: m, n, ilocx,  jlocx, idx, iglobx, jglobx
-  integer(psb_mpk_), allocatable :: szs(:) 
+
   character(len=20)        :: name, ch_err
 
   name='psb_cgatherv'
@@ -307,32 +307,23 @@ subroutine  psb_cgatherv(globx, locx, desc_a, info, iroot)
     goto 9999
   end if
 
+  globx(:)=czero
+
+  do i=1,desc_a%get_local_rows()
+    call psb_loc_to_glob(i,idx,desc_a,info)
+    globx(idx) = locx(i)
+  end do
+
   ! adjust overlapped elements
   do i=1, size(desc_a%ovrlap_elem,1)
     if (me /= desc_a%ovrlap_elem(i,3)) then 
       idx = desc_a%ovrlap_elem(i,1)
-      locx(idx) = czero
+      call psb_loc_to_glob(idx,desc_a,info)
+      globx(idx) = czero
     end if
   end do
-  loc_rows = desc_a%get_local_rows()
-  if ((me == root).or.(root == -1)) then
-    allocate(szs(np))
-  end if
-  call psb_gather(ctxt,loc_rows,szs,root=root)
-  if ((me == root).or.(root == -1)) then
-    if (sum(szs) /= m) then 
-      info=psb_err_internal_error_
-      call psb_errpush(info,name)
-      goto 9999
-    end if
-    call psb_realloc(m,globx,info)
-    if (info /= psb_success_) then 
-      info=psb_err_alloc_dealloc_
-      call psb_errpush(info,name)
-      goto 9999
-    end if
-  end if
-  call psb_gatherv(ctxt,locx(1:loc_rows),globx,szs,root=root)
+
+  call psb_sum(ctxt,globx(1:m),root=root)
 
   call psb_erractionrestore(err_act)
   return  

diff --git a/base/comm/psb_dgather_a.f90 b/base/comm/psb_dgather_a.f90
@@ -60,7 +60,7 @@ subroutine  psb_dgatherm(globx, locx, desc_a, info, iroot)
   type(psb_ctxt_type) :: ctxt
   integer(psb_mpk_) ::  np, me, root, iiroot, icomm, myrank, rootrank
   integer(psb_ipk_) :: ierr(5), err_act, lda_locx, lda_globx, lock, globk,&
-       & maxk, k, jlx, ilx, i, j, loc_rows
+       & maxk, k, jlx, ilx, i, j
   integer(psb_lpk_) :: m, n, ilocx,  jlocx, idx, iglobx, jglobx
   character(len=20) :: name, ch_err
 
@@ -232,11 +232,11 @@ subroutine  psb_dgatherv(globx, locx, desc_a, info, iroot)
 
   ! locals
   type(psb_ctxt_type) :: ctxt
-  integer(psb_mpk_) :: np, me, root, iiroot, icomm, myrank, rootrank, loc_rows
+  integer(psb_mpk_) :: np, me, root, iiroot, icomm, myrank, rootrank
   integer(psb_ipk_) :: ierr(5), err_act, lda_locx, lda_globx, lock, globk,&
        & maxk, k, jlx, ilx, i, j
   integer(psb_lpk_) :: m, n, ilocx,  jlocx, idx, iglobx, jglobx
-  integer(psb_mpk_), allocatable :: szs(:) 
+
   character(len=20)        :: name, ch_err
 
   name='psb_dgatherv'
@@ -307,32 +307,23 @@ subroutine  psb_dgatherv(globx, locx, desc_a, info, iroot)
     goto 9999
   end if
 
+  globx(:)=dzero
+
+  do i=1,desc_a%get_local_rows()
+    call psb_loc_to_glob(i,idx,desc_a,info)
+    globx(idx) = locx(i)
+  end do
+
   ! adjust overlapped elements
   do i=1, size(desc_a%ovrlap_elem,1)
     if (me /= desc_a%ovrlap_elem(i,3)) then 
       idx = desc_a%ovrlap_elem(i,1)
-      locx(idx) = dzero
+      call psb_loc_to_glob(idx,desc_a,info)
+      globx(idx) = dzero
     end if
   end do
-  loc_rows = desc_a%get_local_rows()
-  if ((me == root).or.(root == -1)) then
-    allocate(szs(np))
-  end if
-  call psb_gather(ctxt,loc_rows,szs,root=root)
-  if ((me == root).or.(root == -1)) then
-    if (sum(szs) /= m) then 
-      info=psb_err_internal_error_
-      call psb_errpush(info,name)
-      goto 9999
-    end if
-    call psb_realloc(m,globx,info)
-    if (info /= psb_success_) then 
-      info=psb_err_alloc_dealloc_
-      call psb_errpush(info,name)
-      goto 9999
-    end if
-  end if
-  call psb_gatherv(ctxt,locx(1:loc_rows),globx,szs,root=root)
+
+  call psb_sum(ctxt,globx(1:m),root=root)
 
   call psb_erractionrestore(err_act)
   return  

diff --git a/base/comm/psb_egather_a.f90 b/base/comm/psb_egather_a.f90
@@ -60,7 +60,7 @@ subroutine  psb_egatherm(globx, locx, desc_a, info, iroot)
   type(psb_ctxt_type) :: ctxt
   integer(psb_mpk_) ::  np, me, root, iiroot, icomm, myrank, rootrank
   integer(psb_ipk_) :: ierr(5), err_act, lda_locx, lda_globx, lock, globk,&
-       & maxk, k, jlx, ilx, i, j, loc_rows
+       & maxk, k, jlx, ilx, i, j
   integer(psb_lpk_) :: m, n, ilocx,  jlocx, idx, iglobx, jglobx
   character(len=20) :: name, ch_err
 
@@ -232,11 +232,11 @@ subroutine  psb_egatherv(globx, locx, desc_a, info, iroot)
 
   ! locals
   type(psb_ctxt_type) :: ctxt
-  integer(psb_mpk_) :: np, me, root, iiroot, icomm, myrank, rootrank, loc_rows
+  integer(psb_mpk_) :: np, me, root, iiroot, icomm, myrank, rootrank
   integer(psb_ipk_) :: ierr(5), err_act, lda_locx, lda_globx, lock, globk,&
        & maxk, k, jlx, ilx, i, j
   integer(psb_lpk_) :: m, n, ilocx,  jlocx, idx, iglobx, jglobx
-  integer(psb_mpk_), allocatable :: szs(:) 
+
   character(len=20)        :: name, ch_err
 
   name='psb_egatherv'
@@ -307,32 +307,23 @@ subroutine  psb_egatherv(globx, locx, desc_a, info, iroot)
     goto 9999
   end if
 
+  globx(:)=ezero
+
+  do i=1,desc_a%get_local_rows()
+    call psb_loc_to_glob(i,idx,desc_a,info)
+    globx(idx) = locx(i)
+  end do
+
   ! adjust overlapped elements
   do i=1, size(desc_a%ovrlap_elem,1)
     if (me /= desc_a%ovrlap_elem(i,3)) then 
       idx = desc_a%ovrlap_elem(i,1)
-      locx(idx) = ezero
+      call psb_loc_to_glob(idx,desc_a,info)
+      globx(idx) = ezero
     end if
   end do
-  loc_rows = desc_a%get_local_rows()
-  if ((me == root).or.(root == -1)) then
-    allocate(szs(np))
-  end if
-  call psb_gather(ctxt,loc_rows,szs,root=root)
-  if ((me == root).or.(root == -1)) then
-    if (sum(szs) /= m) then 
-      info=psb_err_internal_error_
-      call psb_errpush(info,name)
-      goto 9999
-    end if
-    call psb_realloc(m,globx,info)
-    if (info /= psb_success_) then 
-      info=psb_err_alloc_dealloc_
-      call psb_errpush(info,name)
-      goto 9999
-    end if
-  end if
-  call psb_gatherv(ctxt,locx(1:loc_rows),globx,szs,root=root)
+
+  call psb_sum(ctxt,globx(1:m),root=root)
 
   call psb_erractionrestore(err_act)
   return  

diff --git a/base/comm/psb_i2gather_a.f90 b/base/comm/psb_i2gather_a.f90
@@ -60,7 +60,7 @@ subroutine  psb_i2gatherm(globx, locx, desc_a, info, iroot)
   type(psb_ctxt_type) :: ctxt
   integer(psb_mpk_) ::  np, me, root, iiroot, icomm, myrank, rootrank
   integer(psb_ipk_) :: ierr(5), err_act, lda_locx, lda_globx, lock, globk,&
-       & maxk, k, jlx, ilx, i, j, loc_rows
+       & maxk, k, jlx, ilx, i, j
   integer(psb_lpk_) :: m, n, ilocx,  jlocx, idx, iglobx, jglobx
   character(len=20) :: name, ch_err
 
@@ -232,11 +232,11 @@ subroutine  psb_i2gatherv(globx, locx, desc_a, info, iroot)
 
   ! locals
   type(psb_ctxt_type) :: ctxt
-  integer(psb_mpk_) :: np, me, root, iiroot, icomm, myrank, rootrank, loc_rows
+  integer(psb_mpk_) :: np, me, root, iiroot, icomm, myrank, rootrank
   integer(psb_ipk_) :: ierr(5), err_act, lda_locx, lda_globx, lock, globk,&
        & maxk, k, jlx, ilx, i, j
   integer(psb_lpk_) :: m, n, ilocx,  jlocx, idx, iglobx, jglobx
-  integer(psb_mpk_), allocatable :: szs(:) 
+
   character(len=20)        :: name, ch_err
 
   name='psb_i2gatherv'
@@ -307,32 +307,23 @@ subroutine  psb_i2gatherv(globx, locx, desc_a, info, iroot)
     goto 9999
   end if
 
+  globx(:)=i2zero
+
+  do i=1,desc_a%get_local_rows()
+    call psb_loc_to_glob(i,idx,desc_a,info)
+    globx(idx) = locx(i)
+  end do
+
   ! adjust overlapped elements
   do i=1, size(desc_a%ovrlap_elem,1)
     if (me /= desc_a%ovrlap_elem(i,3)) then 
       idx = desc_a%ovrlap_elem(i,1)
-      locx(idx) = i2zero
+      call psb_loc_to_glob(idx,desc_a,info)
+      globx(idx) = i2zero
     end if
   end do
-  loc_rows = desc_a%get_local_rows()
-  if ((me == root).or.(root == -1)) then
-    allocate(szs(np))
-  end if
-  call psb_gather(ctxt,loc_rows,szs,root=root)
-  if ((me == root).or.(root == -1)) then
-    if (sum(szs) /= m) then 
-      info=psb_err_internal_error_
-      call psb_errpush(info,name)
-      goto 9999
-    end if
-    call psb_realloc(m,globx,info)
-    if (info /= psb_success_) then 
-      info=psb_err_alloc_dealloc_
-      call psb_errpush(info,name)
-      goto 9999
-    end if
-  end if
-  call psb_gatherv(ctxt,locx(1:loc_rows),globx,szs,root=root)
+
+  call psb_sum(ctxt,globx(1:m),root=root)
 
   call psb_erractionrestore(err_act)
   return  

diff --git a/base/comm/psb_mgather_a.f90 b/base/comm/psb_mgather_a.f90
@@ -60,7 +60,7 @@ subroutine  psb_mgatherm(globx, locx, desc_a, info, iroot)
   type(psb_ctxt_type) :: ctxt
   integer(psb_mpk_) ::  np, me, root, iiroot, icomm, myrank, rootrank
   integer(psb_ipk_) :: ierr(5), err_act, lda_locx, lda_globx, lock, globk,&
-       & maxk, k, jlx, ilx, i, j, loc_rows
+       & maxk, k, jlx, ilx, i, j
   integer(psb_lpk_) :: m, n, ilocx,  jlocx, idx, iglobx, jglobx
   character(len=20) :: name, ch_err
 
@@ -232,11 +232,11 @@ subroutine  psb_mgatherv(globx, locx, desc_a, info, iroot)
 
   ! locals
   type(psb_ctxt_type) :: ctxt
-  integer(psb_mpk_) :: np, me, root, iiroot, icomm, myrank, rootrank, loc_rows
+  integer(psb_mpk_) :: np, me, root, iiroot, icomm, myrank, rootrank
   integer(psb_ipk_) :: ierr(5), err_act, lda_locx, lda_globx, lock, globk,&
        & maxk, k, jlx, ilx, i, j
   integer(psb_lpk_) :: m, n, ilocx,  jlocx, idx, iglobx, jglobx
-  integer(psb_mpk_), allocatable :: szs(:) 
+
   character(len=20)        :: name, ch_err
 
   name='psb_mgatherv'
@@ -307,32 +307,23 @@ subroutine  psb_mgatherv(globx, locx, desc_a, info, iroot)
     goto 9999
   end if
 
+  globx(:)=mzero
+
+  do i=1,desc_a%get_local_rows()
+    call psb_loc_to_glob(i,idx,desc_a,info)
+    globx(idx) = locx(i)
+  end do
+
   ! adjust overlapped elements
   do i=1, size(desc_a%ovrlap_elem,1)
     if (me /= desc_a%ovrlap_elem(i,3)) then 
       idx = desc_a%ovrlap_elem(i,1)
-      locx(idx) = mzero
+      call psb_loc_to_glob(idx,desc_a,info)
+      globx(idx) = mzero
     end if
   end do
-  loc_rows = desc_a%get_local_rows()
-  if ((me == root).or.(root == -1)) then
-    allocate(szs(np))
-  end if
-  call psb_gather(ctxt,loc_rows,szs,root=root)
-  if ((me == root).or.(root == -1)) then
-    if (sum(szs) /= m) then 
-      info=psb_err_internal_error_
-      call psb_errpush(info,name)
-      goto 9999
-    end if
-    call psb_realloc(m,globx,info)
-    if (info /= psb_success_) then 
-      info=psb_err_alloc_dealloc_
-      call psb_errpush(info,name)
-      goto 9999
-    end if
-  end if
-  call psb_gatherv(ctxt,locx(1:loc_rows),globx,szs,root=root)
+
+  call psb_sum(ctxt,globx(1:m),root=root)
 
   call psb_erractionrestore(err_act)
   return