Improved memory allocation in extract_dep_list

base/internals/psi_crea_index.f90

@@ -85,17 +85,6 @@ subroutine psi_i_crea_index(desc_a,index_in,index_out,nxch,nsnd,nrcv,info)
     goto 9999
   endif
 
-  ! allocate dependency list
-  ! This should be computed more efficiently to save space when
-  ! the number of processors becomes very high
-!!$  dl_lda=np+1
-!!$
-!!$  allocate(dep_list(max(1,dl_lda),0:np),length_dl(0:np),stat=info)
-!!$  if (info /= psb_success_) then 
-!!$    call psb_errpush(psb_err_from_subroutine_,name,a_err='Allocate')
-!!$    goto 9999      
-!!$  end if
-
   ! ...extract dependence list (ordered list of identifer process
   !    which every process must communcate with...
   if (debug_level >= psb_debug_inner_) &
@@ -117,7 +106,7 @@ subroutine psi_i_crea_index(desc_a,index_in,index_out,nxch,nsnd,nrcv,info)
   if (debug_level >= psb_debug_inner_) &
        & write(debug_unit,*) me,' ',trim(name),': root sorting dep list'
 
-  call psi_dl_check(dep_list,max(1,dl_lda),np,length_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)

base/internals/psi_extrct_dl.F90

@@ -145,7 +145,7 @@ subroutine psi_i_extract_dep_list(ictxt,is_bld,is_upd,desc_str,dep_list,&
   integer(psb_ipk_) :: i,pointer_dep_list,proc,j,err_act
   integer(psb_ipk_) :: err
   integer(psb_ipk_) :: debug_level, debug_unit
-  integer(psb_mpk_) :: iictxt, icomm, me, np, dl_mpi, minfo
+  integer(psb_mpk_) :: iictxt, icomm, me, np,  minfo
   character  name*20
   name='psi_extrct_dl'
 
@@ -156,13 +156,11 @@ subroutine psi_i_extract_dep_list(ictxt,is_bld,is_upd,desc_str,dep_list,&
   info = psb_success_
 
   call psb_info(iictxt,me,np)
-  dl_lda=np+1
+  allocate(itmp(np+1),length_dl(0:np),stat=info)
-  allocate(dep_list(dl_lda,0:np),length_dl(0:np),stat=info)
   if (info /= psb_success_) then 
-    call psb_errpush(psb_err_from_subroutine_,name,a_err='Allocate')
+    info=psb_err_alloc_dealloc_
-    goto 9999      
+    goto 9999
   end if
-
   do i=0,np 
     length_dl(i) = 0
   enddo
@@ -188,35 +186,35 @@ subroutine psi_i_extract_dep_list(ictxt,is_bld,is_upd,desc_str,dep_list,&
           info = 9999
           int_err(1) = i
           int_err(2) = desc_str(i)
-          goto 998
+          goto 9999
         endif
         !            if((me == 1).and.(proc == 3))write(psb_err_unit,*)'found 3'
         if (mode == 1) then
           !              ...search if already exist proc 
-          !                 in dep_list(*,me)...  
+          !                 in itmp(*)...  
           j=1
           do while ((j < pointer_dep_list).and.&
-               & (dep_list(j,me) /= proc))
+               & (itmp(j) /= proc))
             j=j+1
           enddo
 
           if (j == pointer_dep_list) then
             !                 ...if not found.....
-            dep_list(pointer_dep_list,me)=proc
+            itmp(pointer_dep_list)=proc
             pointer_dep_list=pointer_dep_list+1
           endif
         else if (mode == 0) then
           if (pointer_dep_list > dl_lda) then
             info = psb_err_alloc_dealloc_
-            goto 998
+            goto 9999
           endif
-          dep_list(pointer_dep_list,me)=proc
+          itmp(pointer_dep_list)=proc
           pointer_dep_list=pointer_dep_list+1
         endif
       endif
       i=i+desc_str(i+1)+2
     enddo
-    
+
   else if (is_upd) then
 
     do while (desc_str(i) /= -1)
@@ -234,24 +232,24 @@ subroutine psi_i_extract_dep_list(ictxt,is_bld,is_upd,desc_str,dep_list,&
           !              ...search if already exist proc....                 
           j=1
           do while ((j < pointer_dep_list).and.&
-               & (dep_list(j,me) /= proc))
+               & (itmp(j) /= proc))
             j=j+1
           enddo
           if (j == pointer_dep_list) then
             !                 ...if not found.....
             if (pointer_dep_list > dl_lda) then
               info = psb_err_alloc_dealloc_
-              goto 998
+              goto 9999
             endif
-            dep_list(pointer_dep_list,me)=proc
+            itmp(pointer_dep_list)=proc
             pointer_dep_list=pointer_dep_list+1
           endif
         else if (mode == 0) then
           if (pointer_dep_list > dl_lda) then
             info = psb_err_alloc_dealloc_
-            goto 998
+            goto 9999
           endif
-          dep_list(pointer_dep_list,me)=proc
+          itmp(pointer_dep_list)=proc
           pointer_dep_list=pointer_dep_list+1
         endif
       endif
@@ -263,26 +261,19 @@ subroutine psi_i_extract_dep_list(ictxt,is_bld,is_upd,desc_str,dep_list,&
   endif
 
   length_dl(me)=pointer_dep_list-1
+  dl_lda = max(length_dl(me),1)
+  call psb_max(iictxt, dl_lda)
 
-  !     ... check for errors...
+  allocate(dep_list(dl_lda,0:np),stat=info)
-998 continue 
+  if (info /= psb_success_) then 
-  if (debug_level >= psb_debug_inner_)&
+    call psb_errpush(psb_err_from_subroutine_,name,a_err='Allocate')
-       & write(debug_unit,*) me,' ',trim(name),': info ',info
+    goto 9999      
-  err = info
+  end if
-
-  if (err /= 0) goto 9999
 
   call psb_sum(iictxt,length_dl(0:np))
   call psb_get_mpicomm(iictxt,icomm )
-  allocate(itmp(dl_lda),stat=info)
+  call mpi_allgather(itmp,dl_lda,psb_mpi_ipk_,&
-  if (info /= psb_success_) then 
+       & dep_list,dl_lda,psb_mpi_ipk_,icomm,minfo)
-    info=psb_err_alloc_dealloc_
-    goto 9999
-  endif
-  itmp(1:dl_lda) = dep_list(1:dl_lda,me)
-  dl_mpi = dl_lda
-  call mpi_allgather(itmp,dl_mpi,psb_mpi_ipk_,&
-       & dep_list,dl_mpi,psb_mpi_ipk_,icomm,minfo)
   info = minfo
   if (info == 0) deallocate(itmp,stat=info)
   if (info /= psb_success_) then