CUDAStream.cpp

Go to the documentation of this file.

/***********************************************************************************\
* (c) Copyright 2023-2026 CERN for the benefit of the LHCb and ATLAS collaborations *
*                                                                                   *
* This software is distributed under the terms of the Apache version 2 licence,     *
* copied verbatim in the file "LICENSE".                                            *
*                                                                                   *
* In applying this licence, CERN does not waive the privileges and immunities       *
* granted to it by virtue of its status as an Intergovernmental Organization        *
* or submit itself to any jurisdiction.                                             *
\***********************************************************************************/
 
// Gaudi
#include <Gaudi/AsynchronousAlgorithm.h>
#include <Gaudi/CUDA/CUDAStream.h>
 
// CUDA
#ifndef __CUDACC__
#  include <cuda_runtime.h>
#endif
 
// Others
#include <boost/fiber/cuda/waitfor.hpp>
#include <boost/fiber/mutex.hpp>
#include <boost/fiber/recursive_mutex.hpp>
 
// Standard Library
#include <cstdio>
#include <deque>
#include <format>
#include <mutex>
#include <string>
 
namespace Gaudi::CUDA {
  namespace {
    class StreamList {
      using Mutex_t  = std::recursive_mutex;
      using Stream_t = cudaStream_t;
 
    private:
      std::deque<Stream_t> queue;
      Mutex_t              queue_mtx;
 
    public:
      void push( const Stream_t& s ) {
        std::unique_lock lck( queue_mtx );
        queue.push_back( s );
      }

      bool pop( Stream_t& s ) {
        std::unique_lock lck( queue_mtx );
        if ( queue.empty() ) { return false; }
        s = queue.front();
        queue.pop_front();
        return true;
      }
 
      ~StreamList() {
        Stream_t s;
        while ( pop( s ) ) {
          cudaStreamDestroy( s );
          s = nullptr;
        }
      }
    };
    StreamList  available_streams{};
    std::string err_fmt( cudaError_t err, std::string file, int line ) {
      const char* errname = cudaGetErrorName( err );
      const char* errstr  = cudaGetErrorString( err );
      std::string errmsg =
          std::format( "Encountered CUDA error {} [{}]: {} on {}:{}", errname, int( err ), errstr, file, line );
      return errmsg;
    }
  } // namespace
 
  Stream::Stream( const Gaudi::Algorithm* parent, cudaStream_t stream )
      : m_stream( stream )
      , m_owning( stream == nullptr )
      , m_parent( parent )
      , m_async_parent( dynamic_cast<const Gaudi::AsynchronousAlgorithm*>( parent ) )
      , m_dependents( 0 ) {}
 
  Stream::Stream( const Gaudi::Algorithm* parent ) : Stream( parent, nullptr ) {
    if ( !available_streams.pop( m_stream ) ) {
      cudaError_t err = cudaStreamCreate( &m_stream );
      if ( err != cudaSuccess ) {
        cudaGetLastError();
        throw GaudiException( err_fmt( err, __FILE__, __LINE__ ), "CUDAStreamException", StatusCode::FAILURE );
      }
      err = cudaStreamSynchronize( m_stream );
      if ( err != cudaSuccess ) {
        cudaGetLastError();
        throw GaudiException( err_fmt( err, __FILE__, __LINE__ ), "CUDAStreamException", StatusCode::FAILURE );
      }
    }
  }
 
  Stream::~Stream() {
    if ( m_dependents != 0 ) {
      m_parent->error() << std::format( "Stream destroyed before all its dependents ({} remaining)", m_dependents )
                        << endmsg;
    }
    if ( await().isFailure() ) { m_parent->error() << "Error in Stream destructor" << endmsg; }
    if ( m_owning ) { available_streams.push( m_stream ); }
  }
 
  Stream::operator cudaStream_t() { return m_stream; }
 
  const Gaudi::Algorithm*             Stream::parent() { return m_parent; }
  const Gaudi::AsynchronousAlgorithm* Stream::asyncParent() { return m_async_parent; }
 
  StatusCode Stream::await() {
    auto temp_error = cudaSuccess;
    if ( m_async_parent != nullptr ) {
      auto sc = m_async_parent->decorateSuspension( [this, &temp_error]() {
        auto res   = boost::fibers::cuda::waitfor_all( m_stream );
        temp_error = std::get<1>( res );
        return StatusCode::SUCCESS;
      } );
      if ( !sc.isSuccess() ) {
        m_parent->error() << "Error suspending on stream " << endmsg;
        return sc;
      }
    } else {
      temp_error = cudaStreamSynchronize( m_stream );
    }
    if ( ( temp_error ) != cudaSuccess ) {
      cudaGetLastError();
      std::string errmsg = err_fmt( temp_error, __FILE__, __LINE__ );
      m_parent->error() << errmsg << endmsg;
      return StatusCode::FAILURE;
    }
    return StatusCode::SUCCESS;
  }
 
  void Stream::registerDependency() { ++m_dependents; }
  void Stream::removeDependency() { --m_dependents; }
} // namespace Gaudi::CUDA