dbcsr_create_brd Subroutine

   SUBROUTINE dbcsr_create_brd(dbcsr_mat, brd_mat, equal_dist, floor_dist, numnodes)
      !! Converts a DBCSR matrix to a block row distributed matrix.

      TYPE(dbcsr_type), INTENT(IN)                       :: dbcsr_mat
         !! DBCSR matrix to be converted
      TYPE(dbcsr_type), INTENT(OUT)                      :: brd_mat
         !! converted matrix
      LOGICAL, INTENT(IN)                                :: equal_dist, floor_dist
         !! see documentation of csr_create_from_dbcsr
         !! see documentation of csr_create_from_dbcsr
      INTEGER, INTENT(IN)                                :: numnodes
         !! number of nodes to use for block row distribution

      CHARACTER(LEN=*), PARAMETER :: routineN = 'dbcsr_create_brd'

      CHARACTER                                          :: matrix_type
      CHARACTER(LEN=default_string_length)               :: matrix_name
      INTEGER :: cs, data_type, end_ind, handle, i, k, l, m, mynode, nblkcols_total, &
                 nblkrows_total, nfullrows_local, nfullrows_total, node_size, numnodes_total, row_index, &
                 split_row, start_ind
      INTEGER, ALLOCATABLE, DIMENSION(:)                 :: rdist_tmp, row_blk_size_new_tmp
      INTEGER, ALLOCATABLE, DIMENSION(:, :)              :: pgrid
      INTEGER, DIMENSION(:), POINTER, CONTIGUOUS         :: cdist, col_blk_size, rdist, &
                                                            row_blk_size, row_blk_size_new
      REAL(KIND=real_8)                                  :: chunk_size
      TYPE(dbcsr_distribution_obj)                       :: dist_current, dist_new
      TYPE(dbcsr_mp_obj)                                 :: mp_obj_current, mp_obj_new
      TYPE(mp_comm_type)                                 :: mp_group

      CALL timeset(routineN, handle)

      NULLIFY (row_blk_size, rdist, row_blk_size_new)
      CALL dbcsr_get_info(dbcsr_mat, &
                          nblkrows_total=nblkrows_total, &
                          nblkcols_total=nblkcols_total, &
                          nfullrows_total=nfullrows_total, &
                          row_blk_size=row_blk_size, &
                          col_blk_size=col_blk_size, &
                          matrix_type=matrix_type, &
                          data_type=data_type)

      matrix_name = dbcsr_name(dbcsr_mat)

      ALLOCATE (cdist(nblkcols_total))
      cdist = 0

      dist_current = dbcsr_distribution(dbcsr_mat)
      mp_obj_current = dbcsr_distribution_mp(dist_current)
      mp_group = dbcsr_mp_group(mp_obj_current)
      mynode = dbcsr_mp_mynode(mp_obj_current)
      numnodes_total = dbcsr_mp_numnodes(mp_obj_current)

      ALLOCATE (pgrid(numnodes_total, 1))

      IF (equal_dist) THEN ! Equally distribute rows over processors -> cut blocks

         ! Calculate the number of rows a processor can hold
         IF (floor_dist) THEN
            nfullrows_local = FLOOR(REAL(nfullrows_total, KIND=dp)/numnodes)
         ELSE
            nfullrows_local = CEILING(REAL(nfullrows_total, KIND=dp)/numnodes)
         END IF

         ! allocate maximum amount of memory possibly needed
         ALLOCATE (rdist_tmp(nblkrows_total + numnodes - 1)) ! row distribution
         ALLOCATE (row_blk_size_new_tmp(nblkrows_total + numnodes - 1)) ! new sizes of block rows

         k = 0 ! counter for block rows
         m = 0 ! node counter
         node_size = nfullrows_local ! space available on current node in number of rows
         IF (node_size .GT. 0) THEN
            DO l = 1, nblkrows_total
               split_row = row_blk_size(l) ! size of current block row (number of rows)
               DO WHILE (split_row .GE. node_size) ! cut block row and send it to two nodes
                  k = k + 1
                  m = m + 1
                  row_blk_size_new_tmp(k) = node_size ! size of first part of block row
                  rdist_tmp(k) = m - 1 ! send first part to node m
                  split_row = split_row - node_size ! size of remaining part of block rows
                  node_size = nfullrows_local ! space available on next node
                  IF (floor_dist .AND. (m .EQ. numnodes - 1)) THEN ! send all remaining rows to last node
                     node_size = nfullrows_total - (numnodes - 1)*node_size
                  END IF
               END DO
               IF (split_row .GT. 0) THEN ! enough space left on next node for remaining rows
                  k = k + 1
                  row_blk_size_new_tmp(k) = split_row ! size of remaining part of block row
                  rdist_tmp(k) = m ! send to next node
                  node_size = node_size - split_row ! remaining space on next node
               END IF
            END DO
         ELSE ! send everything to last node if node_size = 0
            rdist_tmp(1:nblkrows_total) = numnodes - 1
            row_blk_size_new_tmp(1:nblkrows_total) = row_blk_size ! row blocks unchanged
            k = nblkrows_total
         END IF

         ! Copy data to correctly allocated variables
         ALLOCATE (row_blk_size_new(k))
         row_blk_size_new = row_blk_size_new_tmp(1:k)
         ALLOCATE (rdist(k))
         rdist = rdist_tmp(1:k)

      ELSE ! Leave block rows intact (do not cut)
         ALLOCATE (rdist(nblkrows_total))
         rdist = 0
         IF (numnodes .GT. nblkrows_total) THEN
            rdist = (/(i, i=0, nblkrows_total - 1)/)
         ELSE
            chunk_size = REAL(nblkrows_total, KIND=dp)/numnodes
            row_index = 0
            start_ind = 1
            DO i = 0, numnodes - 1
               cs = NINT(i*chunk_size) - NINT((i - 1)*chunk_size)
               end_ind = MIN(start_ind - 1 + cs, nblkrows_total)
               rdist(start_ind:end_ind) = row_index
               start_ind = end_ind + 1
               row_index = row_index + 1
            END DO
         END IF
         row_blk_size_new => row_blk_size
      END IF

      pgrid(:, :) = RESHAPE((/(i, i=0, numnodes_total - 1)/), (/numnodes_total, 1/))
      CALL dbcsr_mp_new(mp_obj_new, mp_group, pgrid, mynode, numnodes=numnodes_total)
      CALL dbcsr_distribution_new(dist_new, mp_obj_new, rdist, cdist, reuse_arrays=.TRUE.)

      CALL dbcsr_create(brd_mat, TRIM(matrix_name)//" row-block distributed", &
                        dist_new, matrix_type, row_blk_size_new, col_blk_size, data_type=data_type)
      CALL dbcsr_complete_redistribute(dbcsr_mat, brd_mat)

      DEALLOCATE (pgrid)

      IF (equal_dist) DEALLOCATE (row_blk_size_new)

      CALL dbcsr_distribution_release(dist_new)
      CALL dbcsr_mp_release(mp_obj_new)

      CALL timestop(handle)

   END SUBROUTINE dbcsr_create_brd