LifoStorageCuda.hpp

/*
 * ------------------------------------------------------------------------------------------------------------
 * SPDX-License-Identifier: LGPL-2.1-only
 *
 * Copyright (c) 2018-2020 Lawrence Livermore National Security LLC
 * Copyright (c) 2018-2020 The Board of Trustees of the Leland Stanford Junior University
 * Copyright (c) 2018-2020 TotalEnergies
 * Copyright (c) 2019-     GEOSX Contributors
 * All rights reserved
 *
 * See top level LICENSE, COPYRIGHT, CONTRIBUTORS, NOTICE, and ACKNOWLEDGEMENTS files for details.
 * ------------------------------------------------------------------------------------------------------------
 */
#ifndef LIFOSTORAGECUDA_HPP
#define LIFOSTORAGECUDA_HPP
 
#include <deque>
#include <future>
#include <mutex>
#include <condition_variable>
#include <camp/camp.hpp>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <algorithm>
 
#include "common/FixedSizeDeque.hpp"
#include "common/GEOS_RAJA_Interface.hpp"
#include "common/TimingMacros.hpp"
#include "common/FixedSizeDequeWithMutexes.hpp"
#include "common/MultiMutexesLock.hpp"
#include "common/LifoStorageCommon.hpp"
 
namespace geos
{
template< typename T, typename INDEX_TYPE >
class LifoStorageCuda : public LifoStorageCommon< T, INDEX_TYPE >
{
  typedef LifoStorageCommon< T, INDEX_TYPE > baseLifo;
public:
 
 
  LifoStorageCuda( std::string name, size_t elemCnt, int numberOfBuffersToStoreOnDevice, int numberOfBuffersToStoreOnHost, int maxNumberOfBuffers ):
    LifoStorageCommon< T, INDEX_TYPE >( name, elemCnt, numberOfBuffersToStoreOnHost, maxNumberOfBuffers ),
    m_deviceDeque( numberOfBuffersToStoreOnDevice, elemCnt, LvArray::MemorySpace::cuda ),
    m_pushToDeviceEvents( maxNumberOfBuffers ),
    m_popFromDeviceEvents( maxNumberOfBuffers )
  {}
 
  void pushAsync( arrayView1d< T > array ) override final
  {
    //To be sure 2 pushes are not mixed
    pushWait();
    int id = baseLifo::m_bufferCount++;
    GEOS_ERROR_IF( baseLifo::m_hostDeque.capacity() == 0 && m_deviceDeque.capacity() < baseLifo::m_maxNumberOfBuffers,
                   "Cannot save on a Lifo without host storage (please set lifoSize, lifoOnDevice and lifoOnHost in xml file)" );
 
    m_pushToDeviceEvents[id] = m_deviceDeque.emplaceFront( array );
 
    if( baseLifo::m_maxNumberOfBuffers - id > (int)m_deviceDeque.capacity() )
    {
      LIFO_MARK_SCOPE( geosx::lifoStorage::pushAddTasks );
      // This buffer will go to host memory, and maybe on disk
      std::packaged_task< void() > task( std::bind( &LifoStorageCuda< T, INDEX_TYPE >::deviceToHost, this, baseLifo::m_bufferToHostCount++ ) );
      {
        std::unique_lock< std::mutex > lock( baseLifo::m_task_queue_mutex[0] );
        baseLifo::m_task_queue[0].emplace_back( std::move( task ) );
      }
      baseLifo::m_task_queue_not_empty_cond[0].notify_all();
    }
  }
 
  void pushWait() override final
  {
    if( baseLifo::m_bufferCount > 0 )
    {
      m_pushToDeviceEvents[baseLifo::m_bufferCount-1].wait();
    }
  }
 
  void popAsync( arrayView1d< T > array ) override final
  {
    LIFO_MARK_FUNCTION;
    int id = --baseLifo::m_bufferCount;
 
    if( baseLifo::m_bufferToHostCount > 0 )
    {
      LIFO_MARK_SCOPE( geosx::LifoStorageCuda::popAddTasks );
      // Trigger pull one buffer from host, and maybe from disk
      std::packaged_task< void() > task( std::bind( &LifoStorageCuda< T, INDEX_TYPE >::hostToDevice, this, --baseLifo::m_bufferToHostCount, id ) );
      {
        std::unique_lock< std::mutex > lock( baseLifo::m_task_queue_mutex[0] );
        baseLifo::m_task_queue[0].emplace_back( std::move( task ) );
      }
      baseLifo::m_task_queue_not_empty_cond[0].notify_all();
    }
    m_popFromDeviceEvents[id] = m_deviceDeque.popFront( array );
  }
 
  void popWait() override final
  {
    if( baseLifo::m_bufferCount < baseLifo::m_maxNumberOfBuffers )
    {
      int bufferCount = baseLifo::m_bufferCount;
      auto cuda_event = m_popFromDeviceEvents[bufferCount].try_get< camp::resources::CudaEvent >();
      if( cuda_event ) cudaEventSynchronize( cuda_event->getCudaEvent_t() );
    }
  }
 
  static int computeNumberOfBufferOnDevice( int percent, size_t bufferSize, int maxNumberOfBuffers )
  {
    GEOS_ERROR_IF( percent > 100, "Error, percentage of memory should be smaller than 100, check lifoOnDevice (should be greater than -100)" );
    size_t free, total;
    GEOS_ERROR_IF( cudaSuccess != cudaMemGetInfo( &free, &total ), "Error getting CUDA device available memory" );
    double freeGB = ( ( double ) free ) / ( 1024.0 * 1024.0 * 1024.0 );
    LIFO_LOG_RANK( " LIFO : available memory on device " << freeGB << " GB" );
    return std::min( ( int )( 0.01 * percent * free / bufferSize ), maxNumberOfBuffers );
  }
 
private:
 
  void deviceToHost( int id )
  {
    LIFO_MARK_FUNCTION;
    // The copy to host will only start when the data is copied on device buffer
    baseLifo::m_hostDeque.getStream().wait_for( const_cast< camp::resources::Event * >( &m_pushToDeviceEvents[id] ) );
    baseLifo::m_hostDeque.emplaceFrontFromBack( m_deviceDeque );
 
    if( baseLifo::m_maxNumberOfBuffers - id > (int)(m_deviceDeque.capacity() + baseLifo::m_hostDeque.capacity()) )
    {
      // This buffer will go to host then maybe to disk
      std::packaged_task< void() > task( std::bind( &LifoStorageCuda< T, INDEX_TYPE >::hostToDisk, this, baseLifo::m_bufferToDiskCount++ ) );
      {
        std::unique_lock< std::mutex > lock( baseLifo::m_task_queue_mutex[1] );
        baseLifo::m_task_queue[1].emplace_back( std::move( task ) );
      }
      baseLifo::m_task_queue_not_empty_cond[1].notify_all();
    }
  }
 
  void hostToDevice( int id, int id_pop )
  {
    LIFO_MARK_FUNCTION;
    // enqueue diskToHost on worker #2 if needed
    if( baseLifo::m_bufferToDiskCount > 0 )
    {
      // This buffer will go to host then to disk
      std::packaged_task< void() > task( std::bind( &LifoStorageCuda< T, INDEX_TYPE >::diskToHost, this, --baseLifo::m_bufferToDiskCount ) );
      {
        std::unique_lock< std::mutex > lock( baseLifo::m_task_queue_mutex[1] );
        baseLifo::m_task_queue[1].emplace_back( std::move( task ) );
      }
      baseLifo::m_task_queue_not_empty_cond[1].notify_all();
    }
 
    m_deviceDeque.emplaceBackFromFront( baseLifo::m_hostDeque );
  }
 
  FixedSizeDequeWithMutexes< T, INDEX_TYPE > m_deviceDeque;
  // Events associated to ith  copies to device buffer
  std::vector< camp::resources::Event > m_pushToDeviceEvents;
  // Events associated to ith  copies from device buffer
  std::vector< camp::resources::Event > m_popFromDeviceEvents;
};
}
#endif // LIFOSTORAGE_HPP