RunCommandImpl.cc

// -*- tab-width: 2; indent-tabs-mode: nil; coding: utf-8-with-signature -*-
//-----------------------------------------------------------------------------
// Copyright 2000-2026 CEA (www.cea.fr) IFPEN (www.ifpenergiesnouvelles.com)
// See the top-level COPYRIGHT file for details.
// SPDX-License-Identifier: Apache-2.0
//-----------------------------------------------------------------------------
/*---------------------------------------------------------------------------*/
/* RunCommandImpl.cc                                           (C) 2000-2026 */
/*                                                                           */
/* Implementation of command management on accelerator.                      */
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#include "arccore/common/accelerator/internal/RunCommandImpl.h"
#include "arccore/common/accelerator/internal/AcceleratorCoreGlobalInternal.h"
 
#include "arccore/base/FatalErrorException.h"
#include "arccore/base/ForLoopTraceInfo.h"
#include "arccore/base/PlatformUtils.h"
#include "arccore/base/Convert.h"
#include "arccore/base/ConcurrencyBase.h"
 
#include "arccore/common/accelerator/Runner.h"
#include "arccore/common/accelerator/internal/IRunQueueEventImpl.h"
#include "arccore/common/accelerator/internal/IRunQueueStream.h"
#include "arccore/common/accelerator/internal/IRunnerRuntime.h"
#include "arccore/common/accelerator/internal/RunQueueImpl.h"
#include "arccore/common/accelerator/internal/ReduceMemoryImpl.h"
#include "arccore/common/accelerator/internal/RunnerImpl.h"
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
namespace Arcane::Accelerator::Impl
{
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
RunCommandImpl::
RunCommandImpl(RunQueueImpl* queue)
: m_queue(queue)
, m_execution_policy(queue->runner()->executionPolicy())
, m_use_accelerator(isAcceleratorPolicy(m_execution_policy))
{
  _init();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
RunCommandImpl::
~RunCommandImpl()
{
  _freePools();
  delete m_start_event;
  delete m_stop_event;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void RunCommandImpl::
_freePools()
{
  while (!m_reduce_memory_pool.empty()) {
    delete m_reduce_memory_pool.top();
    m_reduce_memory_pool.pop();
  }
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
IRunQueueEventImpl* RunCommandImpl::
_createEvent()
{
  if (m_use_sequential_timer_event)
    return getSequentialRunQueueRuntime()->createEventImplWithTimer();
  return runner()->_createEventWithTimer();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void RunCommandImpl::
_init()
{
  // Only uses accelerator timers if profiling is enabled.
  // This is done to avoid calling accelerator events since we
  // do not yet know their influence on performance. If it is
  // negligible, we can enable it by default.
 
  // TODO: we should possibly have a sequential instance and
  // one associated with runner() to handle the case where ProfilingRegistry::hasProfiling()
  // changes during execution.
  if (m_use_accelerator && !ProfilingRegistry::hasProfiling())
    m_use_sequential_timer_event = true;
 
  m_start_event = _createEvent();
  m_stop_event = _createEvent();
 
  if (auto v = Convert::Type<Int32>::tryParseFromEnvironment("ARCANE_ACCELERATOR_ALLOW_REUSE_COMMAND", true))
    m_is_allow_reuse_command = (v.value() != 0);
 
  if (auto v = Convert::Type<Int32>::tryParseFromEnvironment("ARCANE_ACCELERATOR_NB_GRID_STRIDE", true)) {
    Int32 n = v.value();
    if (n <= 0)
      n = 1;
    m_default_nb_stride = n;
  }
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
RunCommandImpl* RunCommandImpl::
create(RunQueueImpl* r)
{
  RunCommandImpl* c = r->_internalCreateOrGetRunCommandImpl();
  c->_reset();
  return c;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*!
 * \brief Notification of the start of command execution.
 */
void RunCommandImpl::
notifyBeginLaunchKernel()
{
  if (m_has_been_launched) {
    if (!m_is_allow_reuse_command)
      ARCCORE_FATAL("Command has already been launched. You can not re-use the same command.\n"
                    "  You can temporarily allow it if you set environment variable\n"
                    "    ARCANE_ACCELERATOR_ALLOW_REUSE_COMMAND to 1\n");
  }
  IRunQueueStream* stream = internalStream();
  stream->notifyBeginLaunchKernel(*this);
  // TODO: use the correct stream in sequential mode
  m_has_been_launched = true;
  if (m_use_profiling) {
    m_start_event->recordQueue(stream);
    m_begin_time = platform::getRealTimeNS();
    m_loop_one_exec_stat_ptr = &m_loop_one_exec_stat;
    m_loop_one_exec_stat.setBeginTime(m_begin_time);
  }
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*!
 * \brief Notification of the end of command launch.
 *
 * The command continues to execute in the background.
 */
void RunCommandImpl::
notifyEndLaunchKernel()
{
  IRunQueueStream* stream = internalStream();
  // TODO: use the correct stream in sequential mode
  if (m_use_profiling)
    m_stop_event->recordQueue(stream);
  stream->notifyEndLaunchKernel(*this);
  m_queue->_addRunningCommand(this);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*!
 * \brief Notification of a SYCL kernel launch.
 *
 * \a sycl_event_ptr is of type sycl::event* and contains
 * the event associated with the command returned during
 * calls to sycl::queue::submit().
 */
void RunCommandImpl::
notifyLaunchKernelSyclEvent(void* sycl_event_ptr)
{
  IRunQueueStream* stream = internalStream();
  stream->_setSyclLastCommandEvent(sycl_event_ptr);
  // We must register the queue associated with the event again
  // because when notifyBeginLaunchKernel() was called,
  // the event associated with this queue was not yet available.
  m_start_event->recordQueue(stream);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*!
 * \brief Notification of kernel execution end.
 *
 * After this call, we are sure that the command has finished executing and we
 * can recycle it. In asynchronous mode, this method is called during the
 * synchronization of a queue.
 */
void RunCommandImpl::
notifyEndExecuteKernel()
{
  // Do nothing if the command has not been launched.
  if (!m_has_been_launched)
    return;
 
  Int64 diff_time_ns = 0;
  if (m_use_profiling) {
    diff_time_ns = m_stop_event->elapsedTime(m_start_event);
    runner()->addTime((double)diff_time_ns / 1.0e9);
  }
 
  ForLoopOneExecStat* exec_info = m_loop_one_exec_stat_ptr;
  if (exec_info) {
    exec_info->setEndTime(m_begin_time + diff_time_ns);
    //std::cout << "END_EXEC exec_info=" << m_loop_run_info.traceInfo().traceInfo() << "\n";
    ForLoopTraceInfo flti(traceInfo(), kernelName());
    ProfilingRegistry::_threadLocalForLoopInstance()->merge(*exec_info, flti);
  }
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void RunCommandImpl::
_reset()
{
  m_kernel_name = String();
  m_trace_info = TraceInfo();
  m_nb_thread_per_block = 0;
  m_use_profiling = ProfilingRegistry::hasProfiling();
  m_parallel_loop_options = ConcurrencyBase::defaultParallelLoopOptions();
  m_begin_time = 0;
  m_loop_one_exec_stat.reset();
  m_loop_one_exec_stat_ptr = nullptr;
  m_has_been_launched = false;
  m_has_living_run_command = false;
  m_may_be_put_in_pool = false;
  m_shared_memory_size = 0;
  m_nb_stride = m_default_nb_stride;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
IReduceMemoryImpl* RunCommandImpl::
getOrCreateReduceMemoryImpl()
{
  ReduceMemoryImpl* p = _getOrCreateReduceMemoryImpl();
  if (p) {
    m_active_reduce_memory_list.insert(p);
  }
  return p;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void RunCommandImpl::
releaseReduceMemoryImpl(ReduceMemoryImpl* p)
{
  auto x = m_active_reduce_memory_list.find(p);
  if (x == m_active_reduce_memory_list.end())
    ARCCORE_FATAL("ReduceMemoryImpl in not in active list");
  m_active_reduce_memory_list.erase(x);
  m_reduce_memory_pool.push(p);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
IRunQueueStream* RunCommandImpl::
internalStream() const
{
  return m_queue->_internalStream();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
RunnerImpl* RunCommandImpl::
runner() const
{
  return m_queue->runner();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
ReduceMemoryImpl* RunCommandImpl::
_getOrCreateReduceMemoryImpl()
{
  // No need to allocate specific memory if we are not
  // on an accelerator
  if (!m_use_accelerator)
    return nullptr;
 
  auto& pool = m_reduce_memory_pool;
 
  if (!pool.empty()) {
    ReduceMemoryImpl* p = pool.top();
    pool.pop();
    return p;
  }
  return new ReduceMemoryImpl(this);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*!
 * \brief Method called when the associated RunCommand instance is destroyed.
 */
void RunCommandImpl::
_notifyDestroyRunCommand()
{
  // If the command has not been launched, it must be put back into the
  // queue's command pool (otherwise there will be a memory leak)
  if (!m_has_been_launched || m_may_be_put_in_pool)
    m_queue->_putInCommandPool(this);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
} // namespace Arcane::Accelerator::Impl
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/