#pragma once
/// ****************************************************************************
/// @file   
/// @author Dr.Neeton
/// @date   2015/03/18 - 2015/11/13
/// 
/// @brief  転送パイプ用バッファレギュレータ
/// 
/// ****************************************************************************

#include "CPL_Core_PipeBuffer.h"

namespace CPL {
namespace Core {

// _____________________________________________________________________________
/// @brief コンストラクタ
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
PipeBuffer::PipeBuffer
( _in<td_SpSharedBuffer> in_sp_buffer
, _in<td_SpSharedFlag  > in_sp_flag_write
, _in<td_SpSharedFlag  > in_sp_flag_read
) : m_sp_shared_buffer( in_sp_buffer )
  , m_sp_flag_write( in_sp_flag_write )
  , m_sp_flag_read ( in_sp_flag_read )
{
  if( !in_sp_buffer->IsEmpty() )
  {
    Create( in_sp_buffer->GetSize() );
  }
}


//// _____________________________________________________________________________
///// @brief コンストラクタ
///// 
///// @param[in] in_memory_name
/////   : 転送メモリ用の共有メモリオブジェクトへのスマートポインタを指定します。
//// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//PipeBuffer::PipeRegulator( _in<td_SpSharedBuffer> in_sp_shared_buffer )
//  : m_sp_shared_buffer( in_sp_shared_buffer )
//  , m_pipe_type( FlagPipeType::Invalid ) // 初期はinvalid。Create/Openで自動で設定する。
//  , m_pipe_state( FlagPipeState::Empty )
//{
//}

//// _____________________________________________________________________________
///// @brief デストラクタ
//// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//PipeBuffer::~PipeRegulator()
//{
//  try
//  {
//    ChangePipeState( FlagPipeState::Empty, td_time_msec(100) );
//  }
//  catch(...)
//  {}
//}


// _____________________________________________________________________________
/// @brief 名前の取得。
/// 
/// @return 転送パイプ名を取得。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
cref<td_tstring> PipeBuffer::GetPipeName() const
{
  static td_tstring str_none(SwS(""));
  return
    m_sp_shared_buffer
      ? m_sp_shared_buffer->GetSharedBufferName()
      : str_none;
}


// _____________________________________________________________________________
/// @brief 転送パイプのヘッダ情報を取得。
/// 
/// @param[in] out_mempipe_header
///   : データをコピーするためのPipeHeader変数を指定する。
/// 
/// @retval  1 : 成功。
/// @retval  0 : 失敗。
/// @return -1 : 共有メモリが無効である。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
int
  PipeBuffer::GetPipeHeader
    ( rout<PipeBufferHeader> out_mempipe_header
    ) const
{
  auto t0 = TimeCounter::StaticPoint();
  if( m_sp_shared_buffer == nullptr )
  {
    CPL_OUTPUT( SwS("Invalid shared buffer.") );
    return -1;
  }

  auto res =
    m_sp_shared_buffer->Read
      ( 0
      , &out_mempipe_header
      , sizeof(PipeBufferHeader)
      );

  //auto t1 = TimeCounter::StaticPoint();
  //auto t10 = std::chrono::duration_cast<td_time_usec>(t1 - t0).count();

  return (res != 0);
}


//// _____________________________________________________________________________
///// @brief バッファのステートフラグを取得。
///// 
///// @retval FlagPipeAction参照。
//// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//StatePipeAccess
//  PipeBuffer::GetPipeAccess() const
//{
//  PipeBufferHeader header;
//  if( GetPipeHeader( header ) == 1 )
//  {
//    return header.state_access;
//  }
//  else
//  {
//    return StatePipeAccess::Invalid;
//  }
//}



// _____________________________________________________________________________
/// @brief 作成（or オープン）
/// 
/// @param[in] in_size_pipe
///   : パイプとして使用する共有メモリのサイズを指定します。\n
///   : 既に作成されていた場合、何も処理しないので、この値は無視される。
/// 
/// @retval Error_Invalid_SharedBuffer
///   : 共有メモリオブジェクトが設定されていない。
/// @retval Success_Create_HostMode
///   : ホストモードで作成。\n
///   : 最初に転送用の共有バッファを作成したオブジェクトがホストになる。
/// @retval Success_Create_GuestMode
///   : ゲストモードで作成。\n
///   : 既に同名の共有バッファが存在する場合、その共有バッファが開かれる。\n
///   : この場合、in_size_pipeは無視される。
/// @retval Error_Faild_Create
///   : 共有バッファの作成に失敗。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
eResultPipe PipeBuffer::Create( _in<U64> in_size_pipe )
{
  // 共有バッファオブジェクトのチェック。
  // これはバッファそのもののチェックではなく、共有バッファオブジェクトが設定さ
  // れているかのチェック。
  //CPL_THROWorRETURN_IF
  //  ( m_sp_shared_buffer == nullptr
  //  , IsThrow_Error()
  //  , eResultPipe::Error_Invalid_SharedBuffer
  //  , "Shared buffer is not set.\n"
  //  );
  if( m_sp_shared_buffer == nullptr )
  {
    //CPL_THROWorRETURN
    //  ( IsThrow_Error()
    //  , eResultPipe::Error_Invalid_SharedBuffer
    //  , "Shared buffer is not set.\n"
    //  )
    if(IsThrow_Error()){
      //CPL_THROW_X(eResultPipe::Error_Invalid_SharedBuffer, "Shared buffer is not set.\n" )
      CPL_THROW_X(0,"abc");
      //{ CPL_DBG_OUTPUT_A("Shared buffer is not set.\n") throw eResultPipe::Error_Invalid_SharedBuffer; }
    }
    else {
      CPL_DBG_OUTPUT_A("Shared buffer is not set.\n") return eResultPipe::Error_Invalid_SharedBuffer; }
  }

  auto locking = LockMutex( m_mutex_thread_pipe_line );

  // 割り当て単位は64KB単位とする。
  auto pipe_size = AlignHigh<U64>( in_size_pipe, 64*Ki );

  // 既に共有メモリが作成されていないか、開いてみる。
  auto res_open_buffer =
    m_sp_shared_buffer->Open
      ( Core::eFlagSharedAccess::FileMap_ReadWrite
      );
  if( res_open_buffer )
  {
    return eResultPipe::Success_Create_GuestMode;
  }

  // オープンできなければ作成する。
  bool res_create =
    m_sp_shared_buffer->Create
      ( pipe_size
      , Core::eFlagSharedProtect::Page_ReadWrite
      );
  CPL_THROWorRETURN_IF
    ( !res_create
    , IsThrow_Error()
    , eResultPipe::Error_Faild_Create
    , "Error : Failed to create TransPipe.\n"
    )

  // 初期化
  PipeBufferHeader pbh( pipe_size - sizeof(PipeBufferHeader), 0 );
  m_sp_shared_buffer->Write( 0, &pbh, sizeof(pbh) );

  return eResultPipe::Success_Create_HostMode;
}

// _____________________________________________________________________________
/// @brief クローズ
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
int PipeBuffer::Close()
{
  CPL_SAFE_PTR(m_sp_shared_buffer)->Close();
  return 0;
}


// _____________________________________________________________________________
/// @brief ハンドル無効チェック
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
bool PipeBuffer::IsEmpty() const
{
  return m_sp_shared_buffer
    ? m_sp_shared_buffer->IsEmpty()
    : false;
}


// _____________________________________________________________________________
/// @brief 書き込み可能まで待機。
/// 
/// この関数は用途を考え、例外を投げない仕様とする。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
eResultPipe
  PipeBuffer::WaitUntilWritable
    ( _in<TimePoint> in_time_point_end
    ) const
{
  if( m_sp_flag_write->WaitOn( in_time_point_end ) )
  {
    return eResultPipe::Successful;
  }
  return eResultPipe::Timeout_Myself;
}


// _____________________________________________________________________________
/// @brief 読み込み可能まで待機。
/// 
/// この関数は用途を考え、例外を投げない仕様とする。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
eResultPipe
  PipeBuffer::WaitUntilReadable
    ( _in<TimePoint> in_time_point_end
    ) const
{
  if( m_sp_flag_read->WaitOn( in_time_point_end ) )
  {
    return eResultPipe::Successful;
  }
  return eResultPipe::Timeout_Myself;
}


// _____________________________________________________________________________
/// @brief 共有バッファへ書込み
/// 
/// この関数は同期、書き込み位置・サイズの調整等、低レベルな処理を行う。\n
/// 
/// @param[in] in_write_param_table
///   : 書き込みデータへのポインタを指定します。
/// @param[in] in_num_write_param
///   : 書き込むパラメタ数を指定します。
/// @param[in] in_time_point_end
///   : タイムアウトとするタイムポイント(TimePoint)を指定します。\n
///   : 現時点からオフセットで指定する場合はstd::chrono::durationで指定する(CPL::td_time_msec等でもよし)。\n
///   : 実質タイムアウトさせない場合、TimePoint::Max()を指定する。\n
///   : ※ 転送パイプではソケットと違い、0を指定しても非同期にはならないので注意。
/// 
/// @return
/// 実際に書き込まれたサイズを返す。\n
/// タイムアウトの場合は０、何らかのエラーの場合-1(全ビットON)を返す。
/// 
/// @note
/// この関数では排他制御は行わない。必要なら呼び出し側で行うこと。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
U64
  PipeBuffer::WriteSharedBuffer
    ( pio<PipeWriteParam> in_write_param_table
    , _in<size_t        > in_num_write_param
    , _in<TimePoint     > in_time_point_end
    )
{
  try
  {
    if( GetSharedBuffer() == nullptr )
    {
      return -1;
    }

    // 
    using td_write_param = std::tuple<U64,void_cptr,U64>;
    static thread_local std::vector<td_write_param> tmp_write_param_table;
    tmp_write_param_table.reserve( in_num_write_param*2 + 1 );
    //constexpr static const StatePipeAccess accessed(StatePipeAccess::Updated);

    //// 書き込み可能待ち。
    //auto res = WaitUntilWritable( in_time_point_end );
    //if( res == ResultPipe::Timeout_Myself )
    //{
    //  return 0;
    //}

    //auto lock_ = LockMutex( in_time_point_end, m_mutex_thread_pipe_line );
    //if( !lock_ )
    //{
    //  return 0;
    //}

    // 書込み用パラメタテーブルの構築（個別に書き込むのは非効率のため、書き込みパラメタテーブルを使う）
    tmp_write_param_table.clear();
    decltype(PipeBufferHeader::size_write) size_write = 0;
    U64 const size_buffer_all = GetSharedBuffer()->GetSize();
    U64 pos_begin = sizeof(PipeBufferHeader);
    U64 pos_end   = size_buffer_all;
    U64 pos       = pos_begin;
    // data body
    for( size_t i = 0; i < in_num_write_param; ++i )
    {
      auto& param = in_write_param_table[i];
      // packet header
      if( param.size_packet )
      {
        tmp_write_param_table.push_back
          ( td_write_param
              ( pos
              , &param.size_packet
              , sizeof(param.size_packet)
              )
          );
        pos        += sizeof(PipePacketHeader);
        size_write += sizeof(PipePacketHeader);
      }
      auto size_adjust = Min( param.size_write, pos_end - pos );
      // packet body
      tmp_write_param_table.push_back
        ( td_write_param
            ( pos
            , param.addr_src
            , size_adjust
            )
        );
      pos += size_adjust;
      size_write +=
        static_cast<U32>(Min<U64>( 0xffffffff, size_adjust )); // １度の書き込みサイズは最大32bit.
    }
    // write size
    tmp_write_param_table.push_back
      ( td_write_param
          ( offsetof(PipeBufferHeader,size_write)
          , &size_write
          , sizeof(size_write)
          )
      );
    //size_write += sizeof(size_write);
    //// "Updated"
    //tmp_write_param_table.push_back
    //  ( td_write_param
    //      ( offsetof(PipeBufferHeader,state_access)
    //      , &accessed
    //      , sizeof(accessed)
    //      )
    //  );

    //CPL_EXCTION_(m_mutex_thread_pipe_line)
    //{
      // バッファに書き込み。
      auto write_info =
        GetSharedBuffer()->Write
          ( &tmp_write_param_table[0]
          , tmp_write_param_table.size()
          );

      zzf_setFlagWritten();

      // 書き込まれたサイズは、ヘッダの書き込みサイズ分も含まれているので、その分マイナスする。
      return write_info.second - sizeof(size_write);
    //}
    //return -1;
  }
  catch( eResultPipe res_trans )
  {
    switch( res_trans )
    {
    case eResultPipe::Timeout_Myself:
    case eResultPipe::Timeout_Partner:
      return 0;
    default:
      return -1;
    }
  }
  catch(...)
  {
    return -1;
  }
}

// _____________________________________________________________________________
/// @brief 共有バッファへ書込み
/// 
/// この関数は同期、書き込み位置・サイズの調整等、低レベルな処理を行う。\n
/// 
/// @param[in,out] io_cache_mng : キャッシュマネージャへの参照。
/// @param[in] in_time_point_end
///   : タイムアウトとするタイムポイント(TimePoint)を指定します。\n
///   : 現時点からオフセットで指定する場合はstd::chrono::durationで指定する(CPL::td_time_msec等でもよし)。\n
///   : 実質タイムアウトさせない場合、TimePoint::Max()を指定する。\n
///   : ※ 転送パイプではソケットと違い、0を指定しても非同期にはならないので注意。
/// 
/// @return
/// 実際に書き込まれたサイズを返す。\n
/// タイムアウトの場合は０、何らかのエラーの場合-1(全ビットON)を返す。
/// 
/// @note
/// この関数では排他制御は行わない。必要なら呼び出し側で行うこと。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
U64
  PipeBuffer::WriteSharedBuffer
    ( rio<PipeCacheManager> io_cache_mng
    , _in<TimePoint       > in_time_point_end
    )
{
  try
  {
    if( GetSharedBuffer() == nullptr )
    {
      return -1;
    }

    // 
    using td_write_param = std::tuple<U64,void_cptr,U64>;
    static thread_local std::vector<td_write_param> tmp_write_param_table;
    static thread_local std::vector<U64> tmp_packet_size_table;
    tmp_write_param_table.reserve( io_cache_mng.GetCacheBlockNum() * 2 + 1 );
    tmp_packet_size_table.reserve( io_cache_mng.GetCacheBlockNum() );

    // 書込み用パラメタテーブルの構築（個別に書き込むのは非効率のため、書き込みパラメタテーブルを使う）
    tmp_write_param_table.clear();
    tmp_packet_size_table.clear();
    decltype(PipeBufferHeader::size_write) size_write = 0;
    U64 const size_buffer_all = GetSharedBuffer()->GetSize();
    U64 pos_begin = sizeof(PipeBufferHeader);
    U64 pos_end   = size_buffer_all;
    U64 pos       = pos_begin;
    // data body
    do
    {
      auto info_load = io_cache_mng.GetLoadInfo();
      if( info_load.addr == nullptr )
      {
        break;
      }

      auto size_adjust =
        Min( static_cast<SDEF>(info_load.size)
           , static_cast<SDEF>(pos_end - pos)
           );
      io_cache_mng.Seek( size_adjust, false );

      // packet body
      tmp_write_param_table.push_back
        ( td_write_param
            ( pos
            , info_load.addr
            , size_adjust
            )
        );
      pos += size_adjust;
      size_write +=
        static_cast<U32>(Min<U64>( 0xffffffff, size_adjust )); // １度の書き込みサイズは最大32bit.
    } while( pos < pos_end );
    // write size
    tmp_write_param_table.push_back
      ( td_write_param
          ( offsetof(PipeBufferHeader,size_write)
          , &size_write
          , sizeof(size_write)
          )
      );

    // バッファに書き込み。
    auto write_info =
      GetSharedBuffer()->Write
        ( &tmp_write_param_table[0]
        , tmp_write_param_table.size()
        );

    // 書き込み済みフラグ（読み込みOKフラグ）設定。
    zzf_setFlagWritten();

    // 書き込まれたサイズは、ヘッダの書き込みサイズ分も含まれているので、その分マイナスする。
    auto size_written = write_info.second - sizeof(size_write);

    if( size_write - size_written > 0 )
    {
      // 書き込みパラメタ生成時時に補正してるので、ここは実行されないはずだが…
      io_cache_mng.Seek( static_cast<SDEF>(size_write - size_written), false );
    }
    io_cache_mng.Flush();
    return size_written;
  }
  catch( eResultPipe res_trans )
  {
    switch( res_trans )
    {
    case eResultPipe::Timeout_Myself:
    case eResultPipe::Timeout_Partner:
      return 0;
    default:
      return -1;
    }
  }
  catch(...)
  {
    return -1;
  }
}



// _____________________________________________________________________________
/// @brief 共有バッファからの読み込み。
/// 
/// バッファが更新済みになるまで待機し、更新されたらデータを読み込む基本関数。\n
/// 更新待機にタイムアウトも指定可能。\n
/// \n
/// ただし、相手の状態等はチェックしない低レベル関数のため、より高度な処理は上位
/// モジュールが行うこととする。
/// 
/// @param[out] out_ptr_buf
///   : データ取得用のバッファへのポインタ。\n
///   : この領域は、最低でも設定されている共有バッファ以上のサイズが確保されていなければならない。
/// @param[in] in_time_point_end
///   : タイムアウトとするタイムポイント(TimePoint)を指定します。\n
///   : 現時点からオフセットで指定する場合はstd::chrono::durationで指定する(CPL::td_time_msec等でもよし)。\n
///   : 実質タイムアウトさせない場合、TimePoint::Max()を指定する。
/// 
/// @return
/// 実際に読み込まれたサイズを返す（バッファヘッダは含まない）\n
/// タイムアウトの場合は0、エラーの場合は-1(全ビットON)を返す。
///
/// @note
/// この関数では排他制御は行わない。必要なら呼び出し側で行うこと。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
U64
  PipeBuffer::ReadSharedBuffer
    ( pout<void     > out_ptr_buf
    , _in <TimePoint> in_time_point_end
    )
{
  //constexpr static const StatePipeAccess accessed(StatePipeAccess::Readed);
  //constexpr static const FlagPipeState  state_readed(FlagPipeState::Readed);

  try
  {
    auto sp_buffer = GetSharedBuffer();
    auto const& mem_name = sp_buffer->GetSharedBufferName(); // デバッグ用

    CPL_THROW_X_IF
      ( sp_buffer == nullptr
      , eResultPipe::Error_Invalid_SharedBuffer
      , "%s : Memory pipe for receiving has timed out.\n"
      , ChangeCodeStrings<std::string>( mem_name ).c_str()
      )

    //auto state_partner = GetPipeState_forPartner();

    //// 相手が無効の場合。
    //CPL_THROW_X_IF
    //  ( state_partner == FlagPipeState::Empty
    //  , ResultPipe::Error_Empty_Partner
    //  , "Error : empty partner.\n"
    //  )

    //// アクセス権チェック（待機）
    //auto res_access = WaitUntilReadable( in_time_point_end );
    //if( res_access == ResultPipe::Timeout_Myself )
    //{
    //  return 0;
    //}

    //auto locking = LockMutex( m_mutex_thread_pipe_line );

    // 
    PipeBufferHeader header_info;
    GetPipeHeader( header_info );

    // データ部読み込み。
    auto read_size =
      sp_buffer->Read
        ( sizeof(PipeBufferHeader)
        , out_ptr_buf
        , header_info.size_write
        );

    if( read_size == 0 )
    {
      return -1;
    }

    zzf_setFlagReaded();
    return read_size;
    
    //CPL_EXCTION_(m_mutex_thread_pipe_line)
    //{
    //  // 読み込み完了通知の書き込み。
    //  // access flag to Readed
    //  sp_buffer->Write
    //    ( offsetof(PipeBufferHeader,state_access)
    //    , &accessed
    //    , sizeof(accessed)
    //    );

    //  return read_size;
    //}
    //return -1;
  }
  catch( eResultPipe res_trans )
  {
    switch( res_trans )
    {
    case eResultPipe::Timeout_Myself:
    case eResultPipe::Timeout_Partner:
      return 0;
    default:
      return -1;
    }
  }
  catch( std::bad_alloc const& )
  {
    return -1;
  }
  catch(...)
  {
    return -1;
  }
}




} // namespace Core
} // namespace CEG

