Start reworking of handling of dependency lists.

base/internals/psi_bld_glb_dep_list.F90

@@ -117,3 +117,99 @@ subroutine psi_i_bld_glb_dep_list(ictxt,loc_dl,length_dl,dep_list,dl_lda,info)
   return
 
 end subroutine psi_i_bld_glb_dep_list
+
+subroutine psi_i_bld_glb_csr_dep_list(ictxt,loc_dl,length_dl,c_dep_list,dl_ptr,info)
+  use psi_mod, psb_protect_name => psi_i_bld_glb_csr_dep_list
+#ifdef MPI_MOD
+  use mpi
+#endif
+  use psb_penv_mod
+  use psb_const_mod
+  use psb_error_mod
+  use psb_desc_mod
+  use psb_sort_mod
+  implicit none
+#ifdef MPI_H
+  include 'mpif.h'
+#endif
+  !     ....scalar parameters...
+  integer(psb_ipk_), intent(in)  :: ictxt
+  integer(psb_ipk_), intent(in)  :: loc_dl(:), length_dl(0:)
+  integer(psb_ipk_), allocatable, intent(out) :: c_dep_list(:), dl_ptr(:) 
+  integer(psb_ipk_), intent(out) :: info
+
+
+  !     .....local arrays....
+  integer(psb_ipk_) :: int_err(5)
+
+  !     .....local scalars...
+  integer(psb_ipk_) :: i, proc,j,err_act, length, myld
+  integer(psb_ipk_) :: err
+  integer(psb_ipk_) :: debug_level, debug_unit
+  integer(psb_mpk_) :: iictxt, icomm, me, np, minfo
+  logical, parameter :: dist_symm_list=.false., print_dl=.false.
+  character  name*20
+  name='psi_bld_glb_csr_dep_list'
+
+  call psb_erractionsave(err_act)
+  debug_unit  = psb_get_debug_unit()
+  debug_level = psb_get_debug_level()
+
+  iictxt = ictxt 
+  info = psb_success_
+
+  call psb_info(iictxt,me,np)
+
+  myld = length_dl(me)
+  length = sum(length_dl(0:np-1)) 
+  allocate(dl_ptr(0:np),stat=info)
+  dl_ptr(0) = 0
+  do i=1, np
+    dl_ptr(i) = dl_ptr(i-1) + length_dl(i-1)
+  end do
+
+  if (length /= dl_ptr(np)) then
+    write(0,*) me,trim(name),' Inconsistency: ',length,dl_ptr(np)
+  end if
+  if (debug_level >= psb_debug_inner_) &
+       & write(debug_unit,*) me,' ',trim(name),': Dep_list length ',length_dl(me)
+
+  allocate(c_dep_list(length),stat=info)
+  
+  if (info /= psb_success_) then 
+    call psb_errpush(psb_err_from_subroutine_,name,a_err='Allocate')
+    goto 9999      
+  end if
+  icomm = psb_get_mpi_comm(iictxt)
+  call mpi_allgather(loc_dl,myld,psb_mpi_ipk_,&
+       & c_dep_list,length_dl,dl_ptr,psb_mpi_ipk_,icomm,minfo)
+  
+  info = minfo  
+  if (info /= psb_success_) then 
+    info=psb_err_internal_error_
+    goto 9999
+  endif
+  dl_ptr = dl_ptr + 1 
+  if (print_dl) then
+    if (me == 0) then
+      write(0,*) ' Dep_list '
+      do i=0,np-1
+        write(0,*) 'Proc ',i,':',c_dep_list(dl_ptr(i):dl_ptr(i+1)-1) 
+      end do
+      flush(0)
+    end if
+    call psb_barrier(ictxt)
+  end if
+
+  call psb_erractionrestore(err_act)
+  return
+
+
+9999 continue
+
+  call psb_errpush(info,name,i_err=int_err)
+  call psb_error_handler(err_act)
+
+  return
+
+end subroutine psi_i_bld_glb_csr_dep_list

base/internals/psi_crea_index.f90

@@ -66,7 +66,7 @@ subroutine psi_i_crea_index(desc_a,index_in,index_out,nxch,nsnd,nrcv,info)
   !         ....local scalars...      
   integer(psb_ipk_) :: ictxt, me, np, mode, err_act, dl_lda, ldl
   !         ...parameters...
-  integer(psb_ipk_), allocatable :: dep_list(:,:), length_dl(:), loc_dl(:)
+  integer(psb_ipk_), allocatable :: dep_list(:,:), length_dl(:), loc_dl(:), c_dep_list(:), dl_ptr(:)
   integer(psb_ipk_) :: dlmax, dlavg
   integer(psb_ipk_),parameter    :: root=psb_root_,no_comm=-1
   integer(psb_ipk_) :: debug_level, debug_unit
@@ -154,7 +154,7 @@ subroutine psi_i_crea_index(desc_a,index_in,index_out,nxch,nsnd,nrcv,info)
 !!$         & '  avg:',dlavg, choose_sorting(dlmax,dlavg,np)
 
     if (choose_sorting(dlmax,dlavg,np)) then 
-
+      if (.true.) then 
         call psi_bld_glb_dep_list(ictxt,&
              & loc_dl,length_dl,dep_list,dl_lda,info)
 
@@ -172,7 +172,10 @@ subroutine psi_i_crea_index(desc_a,index_in,index_out,nxch,nsnd,nrcv,info)
         if (do_timings) call psb_toc(idx_phase1)
         if (do_timings) call psb_tic(idx_phase2)
 
-      call psi_dl_check(dep_list,dl_lda,np,length_dl)
+        !
+        ! The dependency list has been symmetrized inside xtract_loc_dl
+        !       
+!!$       call psi_dl_check(dep_list,dl_lda,np,length_dl)
         
         ! ....now i can sort dependency lists.
         call psi_sort_dl(dep_list,length_dl,np,info)
@@ -183,6 +186,24 @@ subroutine psi_i_crea_index(desc_a,index_in,index_out,nxch,nsnd,nrcv,info)
         if (do_timings) call psb_toc(idx_phase2)
         ldl = length_dl(me)
         loc_dl = dep_list(1:ldl,me)
+      else
+        if (do_timings) call psb_toc(idx_phase1)
+        if (do_timings) call psb_tic(idx_phase2)
+        call psi_bld_glb_dep_list(ictxt,&
+             & loc_dl,length_dl,c_dep_list,dl_ptr,info)
+
+!!$        call psi_dl_check(dep_list,dl_lda,np,length_dl)
+!!$
+!!$        ! ....now i can sort dependency lists.
+        call psi_sort_dl(dl_ptr,c_dep_list,length_dl,np,info)
+!!$        if(info /= psb_success_) then
+!!$          call psb_errpush(psb_err_from_subroutine_,name,a_err='psi_sort_dl')
+!!$          goto 9999
+!!$        end if
+        if (do_timings) call psb_toc(idx_phase2)
+        
+        
+      end if
     else
       ! Do nothing
       ldl    = length_dl(me)

base/internals/psi_sort_dl.f90

@@ -89,5 +89,19 @@ subroutine psi_i_sort_dl(dep_list,l_dep_list,np,info)
 
 end subroutine psi_i_sort_dl
 
+subroutine psi_i_csr_sort_dl(dl_ptr,c_dep_list,l_dep_list,np,info)
+  use psi_mod, psb_protect_name => psi_i_csr_sort_dl
+  use psb_const_mod
+  use psb_error_mod
+  implicit none
+  
+  integer(psb_ipk_), intent(in)    :: c_dep_list(:), dl_ptr(0:)
+  integer(psb_ipk_), intent(inout) :: l_dep_list(0:)
+  integer(psb_ipk_), intent(in)    :: np
+  integer(psb_ipk_), intent(out)   :: info
+  ! Local variables
+
+  
+end subroutine psi_i_csr_sort_dl
 
       

base/modules/psi_i_mod.F90

@@ -111,6 +111,14 @@ module psi_i_mod
       integer(psb_ipk_) :: np
       integer(psb_ipk_) :: info
     end subroutine psi_i_sort_dl
+    subroutine psi_i_csr_sort_dl(dl_ptr,c_dep_list,l_dep_list,np,info)
+      import
+      implicit none 
+      integer(psb_ipk_), intent(in)    :: c_dep_list(:), dl_ptr(0:)
+      integer(psb_ipk_), intent(inout) :: l_dep_list(0:)
+      integer(psb_ipk_), intent(in)    :: np
+      integer(psb_ipk_), intent(out)   :: info
+    end subroutine psi_i_csr_sort_dl
   end interface
 
   interface psi_extract_dep_list
@@ -136,7 +144,14 @@ module psi_i_mod
       integer(psb_ipk_), allocatable, intent(out) :: dep_list(:,:)
       integer(psb_ipk_), intent(out) :: info
     end subroutine psi_i_bld_glb_dep_list
-  end interface
+    subroutine psi_i_bld_glb_csr_dep_list(ictxt,loc_dl,length_dl,c_dep_list,dl_ptr,info)
+      import 
+      integer(psb_ipk_), intent(in)  :: ictxt
+      integer(psb_ipk_), intent(in)  :: loc_dl(:), length_dl(0:)
+      integer(psb_ipk_), allocatable, intent(out) :: c_dep_list(:), dl_ptr(:) 
+      integer(psb_ipk_), intent(out) :: info
+    end subroutine psi_i_bld_glb_csr_dep_list
+  end interface psi_bld_glb_dep_list
 
   interface  psi_extract_loc_dl
     subroutine psi_i_xtr_loc_dl(ictxt,is_bld,is_upd,desc_str,loc_dl,length_dl,info)