starpu-max/mpi/examples/native_fortran/nf_mpi_redux.f90

! StarPU --- Runtime system for heterogeneous multicore architectures.
!
! Copyright (C) 2016-2021  Université de Bordeaux, CNRS (LaBRI UMR 5800), Inria
!
! StarPU is free software; you can redistribute it and/or modify
! it under the terms of the GNU Lesser General Public License as published by
! the Free Software Foundation; either version 2.1 of the License, or (at
! your option) any later version.
!
! StarPU is distributed in the hope that it will be useful, but
! WITHOUT ANY WARRANTY; without even the implied warranty of
! MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
!
! See the GNU Lesser General Public License in COPYING.LGPL for more details.
!
program nf_mpi_redux
  use iso_c_binding
  use fstarpu_mod
  use fstarpu_mpi_mod

  implicit none

  integer, target                         :: ret, np, i, j, trial
  type(c_ptr)                             :: work_cl, task_rw_cl,task_red_cl, task_ini_cl
  character(kind=c_char,len=*), parameter :: name=C_CHAR_"task"//C_NULL_CHAR
  character(kind=c_char,len=*), parameter :: namered=C_CHAR_"task_red"//C_NULL_CHAR
  character(kind=c_char,len=*), parameter :: nameini=C_CHAR_"task_ini"//C_NULL_CHAR
  real(kind(1.d0)), target                :: a,tmp
  real(kind(1.d0)), target, allocatable   :: b(:)
  integer(kind=8)                         :: tag, err
  type(c_ptr)                             :: ahdl
  type(c_ptr), target, allocatable        :: bhdl(:)
  type(c_ptr)                             :: task_mode, codelet_mode
  integer, target                         :: comm_world,comm_w_rank, comm_size
  integer(c_int), target                  :: w_node, nworkers, work_coef

  call fstarpu_fxt_autostart_profiling(0)
  ret = fstarpu_init(c_null_ptr)
  ret = fstarpu_mpi_init(1)

  comm_world = fstarpu_mpi_world_comm()
  comm_w_rank  = fstarpu_mpi_world_rank()
  comm_size  = fstarpu_mpi_world_size()
  if (comm_size.lt.2) then
    write(*,'(" ")')
    write(*,'("This application is meant to run with at least two nodes (found ",i4," ; i am ",i4,").")') comm_size, comm_w_rank
    stop 2
  end if
  allocate(b(comm_size-1), bhdl(comm_size-1))
  nworkers = fstarpu_worker_get_count()
  if (nworkers.lt.1) then
    write(*,'(" ")')
    write(*,'("This application is meant to run with at least one worker per node.")')
    stop 2
  end if

  ! allocate and reduction codelets
  task_red_cl = fstarpu_codelet_allocate()
  call fstarpu_codelet_set_name(task_red_cl, namered)
  call fstarpu_codelet_add_cpu_func(task_red_cl,C_FUNLOC(cl_cpu_task_red))
  call fstarpu_codelet_add_buffer(task_red_cl, FSTARPU_RW.ior.FSTARPU_COMMUTE)
  call fstarpu_codelet_add_buffer(task_red_cl, FSTARPU_R)

  task_ini_cl = fstarpu_codelet_allocate()
  call fstarpu_codelet_set_name(task_ini_cl, nameini)
  call fstarpu_codelet_add_cpu_func(task_ini_cl,C_FUNLOC(cl_cpu_task_ini))
  call fstarpu_codelet_add_buffer(task_ini_cl, FSTARPU_W)

  work_coef=2

  do trial=1,2

    if (trial.eq.2) then
          write(*,*) "Using STARPU_MPI_REDUX"
          codelet_mode = FSTARPU_RW.ior.FSTARPU_COMMUTE
          task_mode = FSTARPU_MPI_REDUX
    else if (trial.eq.1) then
          write(*,*) "Using STARPU_REDUX"
          codelet_mode = FSTARPU_REDUX
          task_mode = FSTARPU_REDUX
    end if
    ! allocate and fill codelet structs
    work_cl = fstarpu_codelet_allocate()
    call fstarpu_codelet_set_name(work_cl, name)
    call fstarpu_codelet_add_cpu_func(work_cl, C_FUNLOC(cl_cpu_task))
    call fstarpu_codelet_add_buffer(work_cl, codelet_mode)
    call fstarpu_codelet_add_buffer(work_cl, FSTARPU_R)
    err = fstarpu_mpi_barrier(comm_world)

    if(comm_w_rank.eq.0) then
      write(*,'(" ")')
      a = 1.0
      write(*,*) "init a = ", a
    else
      b(comm_w_rank) = 1.0 / (comm_w_rank + 1.0)
      write(*,*) "init b_",comm_w_rank,"=", b(comm_w_rank)
    end if

    err = fstarpu_mpi_barrier(comm_world)

    tag = 0
    if(comm_w_rank.eq.0) then
      call fstarpu_variable_data_register(ahdl, 0, c_loc(a),c_sizeof(a))
      do i=1,comm_size-1
          call fstarpu_variable_data_register(bhdl(i), -1, c_null_ptr,c_sizeof(b(i)))
      end do
    else
      call fstarpu_variable_data_register(ahdl, -1, c_null_ptr,c_sizeof(a))
      do i=1,comm_size-1
        if (i.eq.comm_w_rank) then
          call fstarpu_variable_data_register(bhdl(i), 0, c_loc(b(i)),c_sizeof(b(i)))
        else
          call fstarpu_variable_data_register(bhdl(i), -1, c_null_ptr,c_sizeof(b(i)))
        end if
      end do
    end if
    call fstarpu_mpi_data_register(ahdl,  tag,  0)
    do i=1,comm_size-1
       call fstarpu_mpi_data_register(bhdl(i), tag+i,i)
    end do

    tag = tag + comm_size

    call fstarpu_data_set_reduction_methods(ahdl,task_red_cl,task_ini_cl)

    err = fstarpu_mpi_barrier(comm_world)


    call fstarpu_fxt_start_profiling()
    do w_node=1,comm_size-1
      do i=1,work_coef*nworkers
        call fstarpu_mpi_task_insert( (/ c_loc(comm_world),   &
               work_cl,                                         &
               task_mode, ahdl,                            &
               FSTARPU_R, bhdl(w_node),                      &
               FSTARPU_EXECUTE_ON_NODE, c_loc(w_node),          &
               C_NULL_PTR /))
      end do
    end do
    call fstarpu_mpi_redux_data(comm_world, ahdl)
    err = fstarpu_mpi_wait_for_all(comm_world)

    if(comm_w_rank.eq.0) then
      tmp = 0
      do w_node=1,comm_size-1
        tmp = tmp + 1.0 / (w_node+1.0)
      end do
      write(*,*) 'computed result ---> ',a, "expected =",&
        1.0 + (comm_size-1.0)*(comm_size)/2.0 + work_coef*nworkers*((comm_size-1.0)*3.0 + tmp)
    end if
    err = fstarpu_mpi_barrier(comm_world)
    call fstarpu_data_unregister(ahdl)
    do w_node=1,comm_size-1
      call fstarpu_data_unregister(bhdl(w_node))
    end do
    call fstarpu_codelet_free(work_cl)

  end do

  call fstarpu_fxt_stop_profiling()
  call fstarpu_codelet_free(task_red_cl)
  call fstarpu_codelet_free(task_ini_cl)


  err = fstarpu_mpi_shutdown()
  call fstarpu_shutdown()
  deallocate(b, bhdl)
  stop 0

contains

  recursive subroutine cl_cpu_task (buffers, cl_args) bind(C)
    use iso_c_binding       ! C interfacing module
    use fstarpu_mod         ! StarPU interfacing module
    implicit none

    type(c_ptr), value, intent(in) :: buffers, cl_args ! cl_args is unused
    integer(c_int) :: ret, worker_id
    integer        :: comm_rank
    integer, target :: i
    real(kind(1.d0)), pointer :: a, b
    real(kind(1.d0))          :: old_a

    worker_id = fstarpu_worker_get_id()
    comm_rank  = fstarpu_mpi_world_rank()

    call c_f_pointer(fstarpu_variable_get_ptr(buffers, 0), a)
    call c_f_pointer(fstarpu_variable_get_ptr(buffers, 1), b)
    call nf_sleep(1.d0)
    old_a = a
    a = old_a + 3.0 + b
    write(*,*) "task   (c_w_rank:",comm_rank," worker_id:",worker_id,") from ",old_a,"to",a

    return
  end subroutine cl_cpu_task

  recursive subroutine cl_cpu_task_red (buffers, cl_args) bind(C)
    use iso_c_binding       ! C interfacing module
    use fstarpu_mod         ! StarPU interfacing module
    implicit none

    type(c_ptr), value, intent(in) :: buffers, cl_args ! cl_args is unused
    integer(c_int) :: ret, worker_id
    integer, target                         :: comm_rank
    real(kind(1.d0)), pointer :: as, ad
    real(kind(1.d0))           :: old_ad
    worker_id = fstarpu_worker_get_id()
    comm_rank  = fstarpu_mpi_world_rank()
    call c_f_pointer(fstarpu_variable_get_ptr(buffers, 0), ad)
    call c_f_pointer(fstarpu_variable_get_ptr(buffers, 1), as)
    old_ad = ad
    ad = ad + as
    call nf_sleep(1.d0)
    write(*,*) "red_cl (c_w_rank:",comm_rank,"worker_id:",worker_id,")",as, old_ad, ' ---> ',ad

    return
  end subroutine cl_cpu_task_red

  recursive subroutine cl_cpu_task_ini (buffers, cl_args) bind(C)
    use iso_c_binding       ! C interfacing module
    use fstarpu_mod         ! StarPU interfacing module
    implicit none

    type(c_ptr), value, intent(in) :: buffers, cl_args
        ! cl_args is unused
    integer(c_int) :: ret, worker_id
    integer, target                         :: comm_rank
    real(kind(1.d0)), pointer :: a
    worker_id = fstarpu_worker_get_id()
    comm_rank  = fstarpu_mpi_world_rank()
    call c_f_pointer(fstarpu_variable_get_ptr(buffers, 0), a)
    call nf_sleep(0.5d0)
    ! As this codelet is run by each worker in the REDUX mode case
    ! this initialization makes salient the number of copies spawned
    write(*,*) "ini_cl (c_w_rank:",comm_rank,"worker_id:",worker_id,") set to", comm_rank, "(was",a,")"
    a = comm_rank
    return
  end subroutine cl_cpu_task_ini

  subroutine nf_sleep(t)
    implicit none
    integer :: t_start, t_end, t_rate
    real(kind(1.d0))     :: ta, t
    call system_clock(t_start)
    do
       call system_clock(t_end, t_rate)
       ta = real(t_end-t_start)/real(t_rate)
       if(ta.gt.t) return
    end do
  end subroutine nf_sleep

end program