/// ****************************************************************************
/// @file   
/// @author Dr.Neeton
/// @date   2015/02/22 - 2015/11/11
/// 
/// @brief  受信用転送パイプ
/// 
/// ****************************************************************************

#include "CPL_Core_PipeReceiver.h"

namespace CPL {
namespace Core {

// _____________________________________________________________________________
/// @brief デフォルトコンストラクタ
/// 
/// これは派生クラスからインタラクティブモードでの動作の場合に呼び出すこととする。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
PipeReceiver::PipeReceiver()
  : PipeBase( NotCalled_() ) // これは呼ばれないはず。
{
  // モード設定は派生クラスで。
}


// _____________________________________________________________________________
/// @brief デストラクタ
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
PipeReceiver::~PipeReceiver()
{
  this->CancelThread();
}


// _____________________________________________________________________________
/// @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)を返す。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
U64
  PipeReceiver::ReadSharedBuffer
    ( pout<void     > out_ptr_buf
    , _in <TimePoint> in_time_point_end
    )
{
  auto res = GetActivePipe()->WaitUntilReadable( in_time_point_end );
  if( (U32)res & (U32)eResultPipe::MaskTimeout )
  {
    return 0;
  }
  if( (U32)res & (U32)eResultPipe::MaskError )
  {
    return -1;
  }

  auto size_readed =
    m_sp_pipe_buffer[ m_idx_buffer ]->ReadSharedBuffer
      ( out_ptr_buf
      , in_time_point_end
      );

  if( size_readed > 0 )
  {
    m_idx_buffer = !m_idx_buffer;
  }

  return size_readed;
}


//// _____________________________________________________________________________
///// @brief 
//// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//void PipeReceiver::zzf_SetupReceiverType()
//{
//  ChangeMode( false, TimePoint::Max() );
//
//  // スレッド開始。
//  this->Start();
//}

// _____________________________________________________________________________
/// @brief 受信キャッシュからデータの読み込み。
/// 
/// @param[out] out_buf
///   : データを格納するための、ByteBufferオブジェクトへの参照。
/// @param[in] in_time_point_end
///   : タイムアウトとするstd::chrono::time_point<>(td_time_point or TimePoint)を指定。\n
///   : 現時点からオフセットで指定する場合はstd::chrono::durationで指定する(CPL::td_time_msec等でもよし)。\n
///   : 実質タイムアウトさせない場合、TimePoint::Max()を指定する。\n
///   : また、極端に短い時間を指定するとタイムアウト扱いになる場合があるので注意。
/// 
/// @return eResultTransRecv参照。
/// 
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
eResultTransRecv
  PipeReceiver::Recv
    ( rout<ByteBuffer> out_buf
    , _in <TimePoint > in_time_point_end
    )
{
  SmartSleeper ss(3);
  // データが受信される、またはタイムアウトまで待機。
  while( m_cache_mng.IsEmpty() )
  {
    if( IsCancelThread() || IsTimeout( in_time_point_end ) )
    {
      return eResultTransRecv::NoneData;
    }
    ss.Sleep();
  }

  auto lock = Core::LockMutex( m_mutex_thread );

  //static U64 size_packet = 0; // パケットサイズを保持（デバッグ用）
  auto size_buf = out_buf.size();
  //
  if( m_size_recv_rest == 0 )
  {
    static_assert( sizeof(m_size_recv_rest)==sizeof(U64), "m_size_recv_rest is not 64bit.\n" );
    U64 packet_size;
    auto res_size =
      m_cache_mng.Load( &packet_size, sizeof(packet_size), 0, false );
    if( res_size < sizeof(packet_size) )
    {
      m_cache_mng.Seek( -(S64)res_size, false );
      return eResultTransRecv::NoneData;
    }
    m_size_recv_rest = packet_size;
#ifdef CPL_DEBUG
    m_size_packet = m_size_recv_rest;
    //CPL_OUTPUT
    //  ( SwS("CorePLi: [time:%.6lf][TID:%Iu]: recv: [packet size:%I64u]\n")
    //  , std::chrono::duration_cast<td_time_usec>(TimeCounter::StaticPoint()).count() / 1000000.0
    //  , GetThreadID()
    //  , (U64)m_size_packet );

#endif
    m_size_recv_rest -= sizeof(PipePacketHeader);
  }
  auto size_readed =
    m_cache_mng.Load( out_buf, 0, static_cast<size_t>(m_size_recv_rest), true );
  m_size_recv_rest -= size_readed;
  if( size_readed > 100 )
  {
    __debugbreak();
  }

  size_buf = out_buf.size();

  if( Config::Communicator::ShowMsg_Recv_RCtoUB )
    CPL_OUTPUT
      ( SwS("CorePLi: [time:%.6lf][TID:%Iu]: recv: RC >> UB : [read:%I64u][rest:%I64u][cache_now_size:%I64u][cache_block_num:%d]\n")//[packet_body:%I64u]
      , std::chrono::duration_cast<td_time_usec>(TimeCounter::StaticPoint()).count() / 1000000.0
      , GetThreadID()
      , (U64)size_readed
      , (U64)m_size_recv_rest
      //, (U64)(m_size_packet - sizeof(PipePacketHeader))
      , (U64)m_cache_mng.GetCacheSizeNow()
      , (int)m_cache_mng.GetCacheBlockNum()
      );
  
  if( m_size_recv_rest == 0 )
  {
    return eResultTransRecv::Received;
  }
  else
  {
    return eResultTransRecv::Receiving;
  }
}


// _____________________________________________________________________________
/// @brief 最大キャッシュサイズの設定。
/// 
/// @param[in] in_size_max_cache
///   : 最大キャッシュサイズを指定。\n
///   : ただし、この値を元に調整された値が設定される。\n
/// 
/// @return 実際に設定されたキャッシュサイズを返す。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
size_t PipeReceiver::SetRecvCacheMaxSize( _in<size_t> in_size_max_cache )
{
  auto const size_buf =
    static_cast<size_t>(GetActivePipe()->GetSharedBuffer()->GetSize());
  auto size_cache_max =
    AlignHigh( Max( in_size_max_cache, size_buf )
             , static_cast<size_t>(64_Ki)
             );
  m_cache_mng.SetMaxCacheSize( size_cache_max );
  return size_cache_max;
}


//// _____________________________________________________________________________
///// @brief モード変更
///// 
///// @param[in] in_flag_interactive
/////   : true  - 共有バッファを送受信共有で使用する場合trueを指定。
/////   : false - 共有バッファを受信専用バッファ
///// @param[in] in_time_point_end
/////   : タイムアウトとするタイムポイント(TimePoint)を指定。\n
/////   : 現時点からオフセットで指定する場合はstd::chrono::durationで指定する(CPL::td_time_msec等でもよし)。\n
/////   : 実質タイムアウトさせない場合、TimePoint::Max()を指定する。
///// 
///// @return ResultPipe参照。
//// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//ResultPipe
//  PipeReceiver::ChangeMode
//    ( _in<bool     > in_flag_interactive
//    , _in<TimePoint> in_time_point_end
//    )
//{
//  try
//  {
//    if( in_flag_interactive )
//    {
//      ChangePipeMode
//        ( FlagPipeMode::Interactive
//        , in_time_point_end
//        );
//    }
//    else
//    {
//      switch( GetPipeType() )
//      {
//      case FlagPipeType::Host:
//        ChangePipeMode
//          ( FlagPipeMode::Guest_to_Host
//          , in_time_point_end
//          );
//        break;
//
//      case FlagPipeType::Guest:
//        ChangePipeMode
//          ( FlagPipeMode::Host_to_Guest
//          , in_time_point_end
//          );
//        break;
//
//      default:
//        throw ResultPipe::Error_Invalid_Type;
//      }
//    }
//    return ResultPipe::Success_ModeChange;
//  }
//  catch( ResultPipe res )
//  {
//    if( static_cast<U32>(res) & 0x0F00 )
//    {
//      CPL_THROWorRETURN
//        ( IsThrow_Timeout()
//        , res
//        , "Timeout : %s", CPL_FUNCNAME
//        )
//    }
//    else
//    {
//      CPL_THROWorRETURN
//        ( IsThrow_Error()
//        , res
//        , "Error : %s", CPL_FUNCNAME
//        )
//    }
//  }
//  catch(...)
//  {
//    return ResultPipe::Error_Faild_ModeChange;
//  }
//}


// _____________________________________________________________________________
/// @brief レシーバー用制御スレッド
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
size_t PipeReceiver::operator()()
{
  const td_time_msec time_slice_default(1000);
  try
  {
    SetThrowOption( false, false );
    
    while( !IsCancelThread()  &&  IsEmpty() )
    {
      CPL_SleepNextTimeSlice();
    }
    if( IsCancelThread() )
    {
      return -1;
    }

    auto size_buffer = GetActivePipe()->GetSharedBuffer()->GetSize();
    auto size_header = sizeof(PipeBufferHeader);
    auto size_max_read = (size_buffer - size_header);
    U64 size_pkt_read_rest = 0;
    U64 size_pkt = 0; // パケットサイズ（デバッグ用）

    while( !IsCancelThread() )
    {
      ByteBuffer tmp_buf_recv( static_cast<size_t>(size_max_read) );
      auto size_readed =
        ReadSharedBuffer
          ( &tmp_buf_recv[0]
          , time_slice_default
          );

      if( size_readed == 0 ) // 読み込みサイズ=0はタイムアウト、つまりデータなし。
      {
        continue;
      }
      else
      if( size_readed == -1 ) // エラー
      {
      }
      else // 正常読み込み。
      {
        // 
        U8_cptr addr_read = &tmp_buf_recv[0];
        if( size_pkt_read_rest == 0 ) // パケット先頭
        {
          auto ptr_header = reinterpret_cast<PipePacketHeader const*>(addr_read);
          size_pkt =
          size_pkt_read_rest = ptr_header->size;
        }
        size_t size_cache_write;
        SmartSleeper ss(3);
        // shared buffer -> cache
        while( !IsCancelThread() )
        {
          //CPL_EXCTION_( td_time_sec(1), m_mutex_thread )
          //{
            size_cache_write =
              m_cache_mng.Store
                ( addr_read
                , static_cast<size_t>(size_readed)
                );
          //}
          //else
          //{
          //  CPL_BREAK(true);
          //}

          // キャッシュに空きがない
          if( size_cache_write == 0 )
          {
            ss.Sleep();
            continue;
          }
          else
          // １回の書き込み要求がキャッシュサイズを超えている。
          if( size_cache_write == -1 )
          {
            // 仕様上ありえないはずなので、抜ける。
            break;
          }
          else // 正常にキャッシュへ書き込み完了。
          {
            size_pkt_read_rest -= size_cache_write;

            if( Config::Communicator::ShowMsg_Recv_SBtoRC )
              CPL_OUTPUT
                ( SwS("CorePLi: [time:%.6lf][TID:%Iu]: recv: SB%d >> RC : [cache_write_size:%I64u][cache_now_size:%I64u][cache_block_num:%d]\n")
                , std::chrono::duration_cast<td_time_usec>(TimeCounter::StaticPoint()).count() / 1000000.0
                , GetThreadID()
                , !m_idx_buffer
                , (U64)size_cache_write
                , (U64)m_cache_mng.GetCacheSizeNow()
                , (int)m_cache_mng.GetCacheBlockNum()
                );
            break;
          }
        }
      }
    }
  }
  catch(...)
  {
    return -1;
  }

  return 0;
}


} // namespace Core
} // namespace CPL
