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

#include "CPL_Core_PipeSender.h"

namespace CPL {
namespace Core {

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


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


// _____________________________________________________________________________
/// @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)を返します。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
U64
  PipeSender::WriteSharedBuffer
    ( pio<PipeWriteParam> in_write_param_table
    , _in<size_t        > in_num_write_param
    , _in<TimePoint     > in_time_point_end
    )
{
  auto size_written =
    GetActivePipe()->WriteSharedBuffer
      ( in_write_param_table
      , in_num_write_param
      , in_time_point_end
      );

  if( size_written > 0 )
  {
    //ChangePipeState( FlagPipeState::Update, td_time_msec(500) );

    m_idx_buffer = !m_idx_buffer;
  }
  else
  if( size_written == -1 )
  {
    //ChangePipeState( FlagPipeState::SendError, td_time_msec(500) );
  }
  // タイムアウト場合は正常離脱の場合もあるので、この関数ではステートを変更してはならない。
  // 呼び出し側で判断すること。

  return size_written;
}


//// _____________________________________________________________________________
///// @brief 送信処理の基本実装（ポインタ仕様）
//// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//ResultTransSend
//  PipeSender::zzf_Send
//    ( pin <void     > in_ptr_buf
//    , _in <size_t   > in_size_send
//    , _in <TimePoint> in_time_point_end
//    , pout<U32      > out_ptr_send_ID
//    )
//{
//  try
//  {
//    PipeSendCacheQueueElement tmp_cache_queue_element;
//    auto* ptr_cache_queue_element = &tmp_cache_queue_element;
//    std::unique_lock<MutexBase> lock_queue( *m_up_mutex_cache_queue, std::defer_lock );
//    auto size_cache = GetSharedBuffer()->GetSize();
//
//    // １度の送信サイズが最大キャッシュサイズを超える場合、同期転送とする。
//    if( in_size_send > m_size_cache_max )
//    {
//      return
//        zzf_SendSynchro
//          ( in_ptr_buf
//          , in_size_send
//          , in_time_point_end
//          , out_ptr_send_ID
//          );
//    }
//    while( m_size_cache_now > m_size_cache_max )
//    {
//      if( IsTimeout( in_time_point_end ) )
//      {
//        return ResultTransSend::Timeout_Myself;
//      }
//      CPL_SleepNextTimeSlice();
//    }
//    // キューが２つ以上ある場合は、キュー終端のバッファに追記する。
//    // キューが１つ以下しか残っていない場合、新しいバッファを作成しキューに追加する。
//    // ただし、最大キャッシュサイズを超える場合、キャッシュが送信されるまで待機。
//    if( m_cache_queue.size() > 1  &&
//        m_cache_queue.back().sp_buffer->size() + in_size_send < size_cache )
//    {
//      lock_queue.lock(); // 既存キューが保持するバッファの内容を書き換える場合、この時点でロックする。
//      ptr_cache_queue_element = &m_cache_queue.back();
//    }
//    else
//    {
//      ptr_cache_queue_element->sp_buffer = CPL_MAKE_SHARED( ByteBuffer );
//      ptr_cache_queue_element->sp_buffer->reserve
//        ( PipeCachePacketHeader::ComputeSize(in_size_send)
//        );
//      ptr_cache_queue_element->is_packing = true; // このデータはパッケージ情報が含まれる。
//    }
//
//    auto sp_buffer = ptr_cache_queue_element->sp_buffer; // コード簡易化用
//    // setup packet header
//    PipeCachePacketHeader packet_header;
//    packet_header.size_packet_data = in_size_send;
//    // copy packet header
//    sp_buffer->insert
//      ( sp_buffer->end()
//      , reinterpret_cast<U8_cptr>(&packet_header)
//      , reinterpret_cast<U8_cptr>(&packet_header) + sizeof(packet_header)
//      );
//    // copy packet body
//    sp_buffer->insert
//      ( sp_buffer->end()
//      , reinterpret_cast<U8_cptr>(in_ptr_buf)
//      , reinterpret_cast<U8_cptr>(in_ptr_buf) + in_size_send
//      );
//
//    m_size_cache_now += sizeof(packet_header) + in_size_send;
//
//    // push queue
//    if( !lock_queue.owns_lock() )
//    {
//      lock_queue.lock(); // 新規バッファに操作していた場合ロックはされていないので、この時点でロックする。
//      m_cache_queue.push( *ptr_cache_queue_element );
//    }
//
//    if( Config::Communicator::ShowMsg_Send )
//      CPL_DBG_OUTPUT
//        ( SwS("CorePLi: send: From UserBuffer >> SendCache : size(%I64d)\n")
//        , (U64)in_size_send
//        );
//
//    return ResultTransSend::Success;
//  }
//  catch(...)
//  {
//    if( Config::Communicator::ShowMsg_Send )
//      CPL_DBG_OUTPUT( SwS("CorePLi: send: Error.\n") );
//
//    return ResultTransSend::Error;
//  }
//}

//// _____________________________________________________________________________
///// @brief 送信処理の基本実装（スマートポインタ仕様）
//// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//ResultTransSend
//  PipeSender::zzf_Send
//    ( _in <SpByteBuffer> in_sp_buffer
//    , _in <TimePoint   > in_time_point_end
//    , _in <bool        > in_flag_copy
//    , pout<U32         > out_ptr_send_ID
//    )
//{
//  try
//  {
//    auto sp_buffer = in_sp_buffer;
//    if( in_flag_copy )
//    {
//      sp_buffer = CPL_MAKE_SHARED( ByteBuffer, in_sp_buffer->size() );
//      memcpy
//        ( &(*sp_buffer)[0]
//        , &(*in_sp_buffer)[0]
//        , in_sp_buffer->size()
//        );
//    }
//
//    PipeSendCacheQueueElement tmp_cache_queue_element =
//    { sp_buffer
//    , in_flag_copy
//    };
//    std::unique_lock<MutexBase> lock_queue( *m_up_mutex_cache_queue );
//    m_cache_queue.push( tmp_cache_queue_element );
//    m_size_cache_now += tmp_cache_queue_element.sp_buffer->size();
//
//    if( Config::Communicator::ShowMsg_Send )
//      CPL_DBG_OUTPUT
//        ( SwS("CorePLi: send: From UserBuffer >> SendCache : size(%I64d)\n")
//        , (U64)in_sp_buffer->size()
//        );
//
//    return ResultTransSend::Success;
//  }
//  catch(...)
//  {
//    if( Config::Communicator::ShowMsg_Send )
//      CPL_DBG_OUTPUT( SwS("CorePLi: send: Error.\n") );
//    return ResultTransSend::Error;
//  }
//}

// _____________________________________________________________________________
/// @brief 同期付送信処理の基本実装
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
eResultTransSend
  PipeSender::zzf_SendSynchro
    ( pin <void     > in_ptr_buf
    , _in <U64      > in_size_send
    , _in <TimePoint> in_time_point_end
    , pout<U32      > out_ptr_send_ID
    )
{
  try
  {
    U64  size_count_rest = in_size_send;
    auto addr_src = static_cast<U8_cptr>(in_ptr_buf);
    bool is_header = false;
    size_t flag_wait = 0;

    while( !IsTimeout( in_time_point_end ) )
    {
      flag_wait = 0;
      CPL_EXCTION_( m_mutex_thread )
      {
        // 先にキャッシュの内容が転送されるまで待つ。
        if( m_cache_mng.IsEmpty() )
        {
          U64 size_written = 0;
          is_header = size_count_rest == in_size_send;

          PipeWriteParam
            pwp( is_header ? size_count_rest : 0
               , addr_src
               , size_count_rest
               );

          size_written =
            WriteSharedBuffer( &pwp, 1, td_time_sec( 1 ) );

          if( size_written == 0 ) // timeout
          {
            continue;
          }
          if( size_written == -1 )
          {
            // エラーの場合
          }
          size_count_rest -= (size_written - is_header * sizeof(U64));
          if( Config::Communicator::ShowMsg_Send_UBtoSB )
            CPL_OUTPUT
              ( SwS("CorePLi: [time:%.6lf][TID:%Iu]: send: UB >> SB%d : [write:%I64u][rest:%I64u][request:%I64u]\n")
              , std::chrono::duration_cast<td_time_usec>(TimeCounter::StaticPoint()).count() / 1000000.0
              , GetThreadID()
              , !m_idx_buffer
              , (U64)size_written
              , (U64)(in_size_send - size_count_rest)
              , (U64)in_size_send
              );
          if( size_count_rest == 0 )
          {
            return eResultTransSend::Success;
          }
          addr_src += size_written;
        }
        else
        {
          flag_wait = 1;
        }
      }
      CPL_Sleep( td_time_msec(flag_wait) );
    }
    return eResultTransSend::Timeout_Myself;
  }
  catch(...)
  {
    return eResultTransSend::Error;
  }
}

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


// _____________________________________________________________________________
/// @brief 送信キャッシュへ書き込み。
/// 
/// @copydetails Sender::Send(pin<void>,_in<size_t>,_in<TimePoint>,pout<U32>)
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
eResultTransSend
  PipeSender::Send
    ( pin <void     > in_ptr_buf
    , _in <size_t   > in_size_send
    , _in <TimePoint> in_time_point_end
    , pout<U32      > out_send_ID
    )
{
  if( in_size_send == 0 )
  {
    return eResultTransSend::NoneData;
  }

  if( m_cache_mng.GetCacheSizeMax() == 0 )
  {
    return
      zzf_SendSynchro
        ( in_ptr_buf
        , in_size_send
        , in_time_point_end
        , out_send_ID
        );
  }
  size_t size_written = 0;
  SmartSleeper ss(2);
  while( !IsTimeout( in_time_point_end ) )
  {
    size_written = 0;
    //size_t res0 = 0, res1 = 0;
    //CPL_EXCTION_( td_time_sec(3), m_mutex_thread )
    if( auto locking = LockMutex( td_time_sec(3), m_mutex_thread ) )
    {
      size_t size_add = sizeof(PipePacketHeader) + in_size_send;
      if( m_cache_mng.GetCacheSizeMax() < size_add ) // cache max over
      {
        size_written = -1;
      }
      else
      if( m_cache_mng.GetFreeSize() >= size_add ) // cache space not over
      {
        PipePacketHeader pph = { sizeof(PipePacketHeader) + in_size_send };
        size_written += m_cache_mng.Store( &pph, sizeof(pph) );
        size_written += m_cache_mng.Store( in_ptr_buf, in_size_send );
      }
    }
    else // timeout
    {
      continue;
    }

    if( size_written == 0 ) // cache space over
    {
      ss.Sleep();
      continue;
    }
    if( size_written == -1 ) // cache max over
    {
      return
        zzf_SendSynchro
          ( in_ptr_buf
          , in_size_send
          , in_time_point_end
          , out_send_ID
          );
    }
    if( Config::Communicator::ShowMsg_Send_UBtoSC )
      CPL_OUTPUT
        ( SwS("CorePLi: [time:%.6lf][TID:%Iu]: send: UB >> SC : [req_size:%I64u][cache:%I64u][cache_max:%I64u]\n")
        , std::chrono::duration_cast<td_time_usec>(TimeCounter::StaticPoint()).count() / 1000000.0
        , GetThreadID()
        , (U64)in_size_send // 今キャッシュへの書き込こんだサイズ。
        , (U64)m_cache_mng.GetCacheSizeNow() // 現在のキャッシュ済みサイズ。
        , (U64)m_cache_mng.GetCacheSizeMax() // 最大キャッシュ可能サイズ。
        );
    // ここが処理される場合、正常に読込完了したということ。
    return eResultTransSend::Success;
  }

  return eResultTransSend::Timeout_Myself;
}

// _____________________________________________________________________________
/// @brief 送信キャッシュへ書き込み。
/// 
/// @copydetails Sender::Send(cref<ByteBuffer>,_in<TimePoint>,pout<U32>)
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
eResultTransSend
  PipeSender::Send
    ( cref<ByteBuffer> in_buffer
    , _in <TimePoint > in_time_point_end
    , pout<U32       > out_send_ID
    )
{
  return
    Send( &in_buffer[0]
        , in_buffer.size()
        , in_time_point_end
        , out_send_ID
        );
}

// _____________________________________________________________________________
/// @brief 送信キャッシュへ書き込み。
/// 
/// @copydetails Sender::Send(rvref<ByteBuffer>,_in<TimePoint>,pout<U32>)
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
eResultTransSend
  PipeSender::Send
    ( rvref<ByteBuffer> in_buffer
    , _in  <TimePoint > in_time_point_end
    , pout <U32       > out_send_ID
    )
{
#ifdef _MSC_VER
  #pragma message( "CorePLi: この関数はまだ未対応です。 - " CPL_FUNCNAME )
#endif
  return eResultTransSend::Error;
}

//// _____________________________________________________________________________
///// @brief 送信キャッシュへ書き込み。
///// 
///// @copydetails Sender::Send(_in<SpByteBuffer>,_in<TimePoint>,_in<bool>,pout<U32>)
//// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//ResultTransSend
//  PipeSender::Send
//    ( _in <SpByteBuffer> in_sp_buffer
//    , _in <TimePoint   > in_time_point_end
//    , _in <bool        > in_flag_copy
//    , pout<U32         > out_send_ID
//    )
//{
//  return zzf_Send( in_sp_buffer, in_time_point_end, in_flag_copy, out_send_ID );
//}

//// _____________________________________________________________________________
///// @brief 送信キャッシュへ書き込み。
///// 
///// @copydetails Sender::Send(cptr<SpByteBuffer>,_in<int>,_in<TimePoint>,_in<bool>,pout<U32>)
//// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//ResultTransSend
//  PipeSender::Send
//    ( cptr<SpByteBuffer> in_buffer_table
//    , _in <int         > in_buf_table_num
//    , _in <TimePoint   > in_time_point_end
//    , _in <bool        > in_flag_copy
//    , pout<U32         > out_send_ID
//    )
//{
//  try
//  {
//    for( int i = 0; i < in_buf_table_num; ++i )
//    {
//      auto res =
//        zzf_Send( in_buffer_table[i], in_time_point_end, in_flag_copy, out_send_ID );
//      if( res != ResultTransSend::Success )
//      {
//        throw res;
//      }
//    }
//    return ResultTransSend::Success;
//  }
//  catch( ResultTransSend res )
//  {
//    return res;
//  }
//  catch(...)
//  {
//    return ResultTransSend::Error;
//  }
//}


// _____________________________________________________________________________
/// @brief 最大キャッシュサイズの設定。
/// 
/// @param[in] in_size_max_cache
///   : 最大キャッシュサイズを指定。\n
///   : ただし、この値を元に調整された値が設定される。\n
///   : 0を指定すると、同期送信となる。
/// 
/// @return 実際に設定されたキャッシュサイズを返す。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
size_t PipeSender::SetSendCacheMaxSize( _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
/////   : タイムアウトする時点を指定\n
/////   : 実質タイムアウトさせない場合、TimePoint::Max()を指定する。
///// 
///// @return ResultPipe参照。
//// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//ResultPipe
//  PipeSender::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::Host_to_Guest
//          , in_time_point_end
//          );
//        break;
//
//      case FlagPipeType::Guest:
//        ChangePipeMode
//          ( FlagPipeMode::Guest_to_Host
//          , 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 PipeSender::operator()()
{
  try
  {
    SetThrowOption( false, false );
    //auto sp_mutex = GetSharedBuffer(m_pipe_param)->GetMutex();

    while( !IsCancelThread()  &&  IsEmpty() )
    {
      CPL_SleepNextTimeSlice();
    }
    if( IsCancelThread() )
    {
      return -1;
    }

    td_time_sec def_timeout{ 3 };
    SmartSleeper ss(3);
    //U64 size_packet = 0; // パケットサイズ（デバッグ用）
    // １度の書き込みで使用できる最大バッファサイズ。
    size_t const size_buffer =
      static_cast<size_t>
        (GetActivePipe()->GetSharedBuffer()->GetSize() - sizeof(PipeBufferHeader));
    td_time_usec wait_value( size_buffer / 8 );

    while( !IsCancelThread() )
    {
      if( m_cache_mng.IsEmpty() )
      {
        // キャッシュへの書き込みの関係で、ある程度スリープを入れた方が速い。
        CPL_Sleep( wait_value );
        continue;
      }
      // 書き込めるようになるまで待機。
      auto res_wait = GetActivePipe()->WaitUntilWritable( def_timeout );
      if( (U32)res_wait & (U32)eResultPipe::MaskTimeout )
      {
        continue;
      }
      else
      if( (U32)res_wait == (U32)eResultPipe::MaskError )
      {
        ChangePipeState( ePipeState::SendError, def_timeout );
        continue;
      }

      auto lock = Core::LockMutex( def_timeout, m_mutex_thread );
      if( !lock ) // タイムアウトチェック。
      {
        continue;
      }

      double d_ = std::chrono::duration_cast<td_time_usec>(TimeCounter::StaticPoint()).count() / 1000000.0;
      // 共有バッファへ書き込み。
      auto size_written =
        GetActivePipe()->WriteSharedBuffer( m_cache_mng, def_timeout );

      if( size_written > 0 )
      {
        if( Config::Communicator::ShowMsg_Send_SCtoSB )
          CPL_OUTPUT
            ( SwS("CorePLi: [time:%.6lf][TID:%Iu]: send: SC >> SB%d : [write:%I64u]\n")
            , d_
            , GetThreadID()
            , m_idx_buffer
            , (U64)size_written     // 書き込まれたサイズ
            );

        m_idx_buffer = !m_idx_buffer;
      }
    } // while loop
  }
  catch(...)
  {
    return -1;
  }

  return 0;
}


//namespace zzz {
//  struct BufferParamHelper
//  {
//    PipeWriteParam* table;
//    size_t          num_table;
//    PipeWriteParam* active;
//    size_t          read_pos = 0;
//    U64             size_total = 0;
//    BufferParamHelper( PipeWriteParam* in_ptr_PWT, size_t n )
//      : table    ( in_ptr_PWT )
//      , num_table( n )
//      , active   ( in_ptr_PWT )
//    {
//      for( int i = 0; i < n; ++i )
//      {
//        size_total += table[i].size_write;
//      }
//    }
//
//    bool IsEmpty() const
//    {
//      return active == nullptr;
//    }
//    char_cptr GetReadPointer() const
//    {
//      return active ? (active->addr_src + read_pos) : nullptr;
//    }
//    size_t    GetActiveReadSize() const
//    {
//      return active ? (active->size_write - read_pos) : 0;
//    }
//
//    // シーク。終端に達すると-1を返す。
//    U64 Seek( _in<U64> in_offs )
//    {
//      if( active == nullptr ) return -1;
//      auto offs = in_offs;
//      auto* end = table + num_table;
//      while( active < end )
//      {
//        if( active->size_write <= read_pos + offs )
//        {
//          offs -= (active->size_write - read_pos);
//          read_pos = 0;
//          ++active;
//        }
//        else
//        {
//          read_pos += offs;
//          return in_offs;
//        }
//      }
//      active = nullptr;
//      return -1;
//    }
//  };
//}


//// _____________________________________________________________________________
///// @brief 実際に共有バッファに書き込む処理
//// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//ResultTransSend
//  PipeSender::zzf_WriteSharedBuffer
//    ( pin <void     > in_ptr_buf
//    , _in <size_t   > in_size_write
//    , _in <TimePoint> in_time_point_end
//    , pout<U32      > out_ptr_send_ID
//    , _in <bool     > in_flag_packing
//    )
//{
//  if( GetSharedBuffer() == nullptr )
//  {
//    return ResultTransSend::Error;
//  }
//
//  auto locking = LockMutex( GetThreadMutex() );
//
//  // 
//  using td_write_param = std::tuple<U64,void_cptr,U64>;
//  static std::vector<td_write_param> tmp_write_param_table;
//
//  try
//  {
//    // 書き込み可能待ち。
//    auto res = WaitUntilWritable( in_time_point_end );
//    if( res == ResultPipe::Timeout_Myself )
//    {
//      return ResultTransSend::Timeout_Myself;
//    }
//
//    // 例外設定を一時的に設定。
//    // 自身の関数には、エラーおよびタイムアウトの場合例外を投げてもらうようにする。
//    TempOptionThrow tmp_op_change( this, true, true );
//
//    // バッファテーブル参照用のヘルパーオブジェクト。
//    zzz::BufferParamHelper wph( in_write_param_table, in_num_write_param );
//
//    auto& mem_name =
//      GetSharedBuffer(in_pipe_param)->GetSharedBufferName(); // デバッグ用
//    size_t counter_size_write = 0; // 書き込みサイズのカウンター
//
//    // 状態初期化。
//    //ChangePipeState( in_pipe_param, FlagPipeState::Wait, in_time_point_end );
//
//    ////auto state_partner = GetPipeState_forPartner( in_pipe_param );
//
//    ////if( state_partner == FlagPipeState::Readed )
//    ////{
//    ////  WaitStateNotEqual( in_pipe_param, FlagPipeState::Readed, in_time_point_end );
//    ////  state_partner = GetPipeState_forPartner( in_pipe_param );
//    ////}
//    ////if( state_partner != FlagPipeState::Wait  &&
//    ////    state_partner != FlagPipeState::Timeout  &&
//    ////    state_partner != FlagPipeState::OverSize )
//    ////{
//    ////  return ResultTransSend::Error;
//    ////}
//
//    ////// モードがインタラクティブの場合、一度リクエストを発行し、相手の準備待ちが必要となる。
//    ////if( GetPipeMode( in_pipe_param ) == FlagPipeMode::Interactive )
//    ////{
//    ////  ChangePipeState( in_pipe_param, FlagPipeState::Request, in_time_point_end );
//    ////  WaitStateEqual( in_pipe_param, FlagPipeState::Ready, in_time_point_end );
//    ////}
//
//    // 送信ループ。
//    while( true )
//    {
//      // 送信先が待機状態(or TIMEOUT)になるまで待機。
//      //WaitStateEqual( FlagPipeState::Wait, in_time_point_end );
//
//      ////ChangePipeState( in_pipe_param, FlagPipeState::Updating, in_time_point_end );
//
//      auto   buf_size  = GetSharedBuffer(in_pipe_param)->GetSize();
//      auto   pos_write = PipeHeader::GetHeaderSize();
//      size_t size_send_once = 0;
//      // 書込み用パラメタテーブルの構築。
//      tmp_write_param_table.clear();
//      do {
//        // 一回のデータ部書き込みサイズを計算。
//        size_t
//          size_write_once =
//            Min<size_t>
//              ( wph.GetActiveReadSize()//size_total - counter_size_write
//              , static_cast<size_t>(buf_size) - pos_write
//              );
//        if( size_write_once == 0 )
//        {
//          break;
//        }
//
//        ////// バッファに書き込み。
//        ////GetSharedBuffer(in_pipe_param)->Write
//        ////  ( pos_write
//        ////  , wph.GetReadPointer()
//        ////  , size_write_once
//        ////  );
//        tmp_write_param_table.push_back
//          ( td_write_param(pos_write,wph.GetReadPointer(), size_write_once )
//          );
//
//        size_send_once += size_write_once;
//        wph.Seek( size_write_once );
//        if( wph.IsEmpty() )
//        {
//          break;
//        }
//        pos_write += size_write_once;
//      } while( true );
//
//      // バッファに書き込み。
//      GetSharedBuffer(in_pipe_param)->Write
//        ( &tmp_write_param_table[0]
//        , tmp_write_param_table.size()
//        );
//
//      counter_size_write += size_send_once;
//
//      // トータルサイズと書き込みサイズを更新。
//      zzf_SetWriteSize( in_pipe_param, wph.size_total, (U32)size_send_once );
//
//      // 書き込み完了フラグをセット
//      zzf_SetAccessFlag( in_pipe_param, StatePipeAccess::Updated );
//      ////ChangePipeState( in_pipe_param, FlagPipeState::Update, in_time_point_end );
//
//      ////// 送信先が読み込み完了になるまで待機。
//      ////WaitStateEqual
//      ////  ( in_pipe_param
//      ////  , FlagPipeState::Readed
//      ////  , Min<td_time_point>(in_time_point_end, TimePointOffset(td_time_msec(500)) )
//      ////  );
//
//      //// 送信元フラグを待機状態に。
//      ////ChangePipeState( in_pipe_param, FlagPipeState::Wait, in_time_point_end );
//
//      // 終了判定。
//      if( wph.size_total <= counter_size_write )
//      {
//        break;
//      }
//
//      ////// 送信先が待機状態になるまで待機。
//      ////WaitStateEqual
//      ////  ( in_pipe_param
//      ////  , FlagPipeState::Wait
//      ////  , Min<td_time_point>(in_time_point_end, TimePointOffset(td_time_msec(500)) )
//      ////  );
//
//      //CPL_SleepNextTimeSlice();
//    }
//
//    return ResultTransSend::Success;
//  }
//  catch( ResultPipe res_trans )
//  {
//    switch( res_trans )
//    {
//    case ResultPipe::Timeout_Myself:
//      return ResultTransSend::Timeout_Myself;
//    case ResultPipe::Timeout_Partner:
//      return ResultTransSend::Timeout_Partner;
//    default:
//      return ResultTransSend::Error;
//    }
//  }
//  catch(...)
//  {
//    return ResultTransSend::Error;
//  }
//}
//{
//  ResultTransSend res;
//  if( in_flag_packing )
//  {
//    PipeCachePacketHeader pkt_header =
//    { in_size_write
//    };
//    PipeSendRegulator::BufferParam param[] =
//    { PipeSendRegulator::BufferParam( reinterpret_cast<char_cptr>(&pkt_header), sizeof(pkt_header) )
//    , PipeSendRegulator::BufferParam( reinterpret_cast<char_cptr>(in_ptr_buf), in_size_write )
//    };
//    res =
//      PipeSendRegulator::Send
//        ( m_pipe_param
//        , param
//        , 2
//        , in_time_point_end
//        , nullptr
//        );
//
//  }
//  else
//  {
//    res =
//      PipeSendRegulator::Send
//        ( m_pipe_param
//        , in_ptr_buf
//        , in_size_write
//        , in_time_point_end
//        , nullptr
//        );
//  }
//
//  if( Config::Communicator::ShowMsg_Send )
//  {
//    if( res == ResultTransSend::Success )
//    {
//      CPL_DBG_OUTPUT
//        ( SwS("CorePLi: send: From SendCache to SharedBuffer stored. : size(%I64d)\n")
//        , (U64)(in_size_write + in_flag_packing * sizeof(PipeCachePacketHeader))
//        );
//    }
//  }
//  return res;
//}


} // namespace Core
} // namespace CPL
