text: /*
   * pixi_interface.cc
   *
   *  Created on: Feb 4, 2014
   *      Author: root
   */
  
  #include "pixi_interface.h"
  
  // Copyright (2013-2014) JDSU T&M Pte Ltd.  All rights reserved.
  
  #ifndef NULLPTR
  #  define NULLPTR nullptr
  #endif
  
  typedef int ResultCode;
  
  #include "nt.h"
  #include <log4cxx/logger.h>
  #include <stdexcept>
  #include "xtreme/iwf/config_manager.h"
  #include "xtreme/iwf/config_multiplexer.h"
  #include "xtreme/iwf/internal/telemetry_sink_simple.h"
  #include "xtreme/iwf/active_heartbeat_sink.h"
  #include "xtreme/iwf/manual_parser.h"
  #include "xtreme/iwf/timestamper.h"
  #include "xtreme/iwf/early_filter_sink.h"
  #include "xtreme/iwf/early_job_filter_sink.h"
  #include "xtreme/iwf/cp_up_classifier.h"
  #include "xtreme/common/debug.h"
  #include "xtreme/common/event_frame.h"
  #include "xtreme/common/event_timesync.h"
  #include "xtreme/common/event_block_end.h"
  #include "xtreme/common/frame_block_pool.h"
  
  #include "xtreme/common/notification_def.h"
  #include "xtreme/common/xtreme_frame_buffer.h"
  #include "xtreme/common/internal/TcpEngine/logger.h"
  
  #include "TcpEngineThread.h"
  
  namespace {
  
  void g_release_nt_buffers(void* object, PiDataLenBytes len, const std::vector<uint8_t*>& nt_buffers) {
    xtreme::TcpEngine::Pixi_interface* interface = static_cast<xtreme::TcpEngine::Pixi_interface*>(object);
    interface->release_nt_buffers(len, nt_buffers);
  }
  }   // anonymous namespace
  
  using namespace xtreme;
  using namespace xtreme::iwf;
  
  namespace xtreme{
  namespace TcpEngine {
  
  int64_t Pixi_interface::time_sync_period_ = 100 * 1000 * 1000;  // 100 milliseconds.
  Pixi_manager Pixi_manager::instance_(true, true);
  Pixi_manager::Pixi_manager(bool telemetry_enabled, bool zero_copy)
      : telemetry_enabled_(telemetry_enabled),
        zero_copy_(zero_copy) {
  }
  
  Pixi_manager::~Pixi_manager() {
    std::map<uint16_t, Pixi_interface*>::iterator iter;
    for (iter = cp_interfaces_.begin(); iter != cp_interfaces_.end(); ++iter) {
      delete iter->second;
    }
    for (iter = up_interfaces_.begin(); iter != up_interfaces_.end(); ++iter) {
      delete iter->second;
    }
    TCP_MSG_INFO("%s", "Pixi_manager singleton is destroyed");
  }
  
  Pixi_interface* Pixi_manager::grab_Pixi_interface(uint16_t stream_id, bool is_cp) {
    if (is_cp) {
      std::map<uint16_t, Pixi_interface*>::iterator iter = cp_interfaces_.find(stream_id);
      if (iter != cp_interfaces_.end()) {
        return iter->second;
      }
    } else {
      std::map<uint16_t, Pixi_interface*>::iterator iter = up_interfaces_.find(stream_id);
      if (iter != up_interfaces_.end()) {
        return iter->second;
      }
    }
  
    Pixi_interface* interface = new Pixi_interface(stream_id, is_cp);
    if (is_cp) {
      cp_interfaces_.insert(std::pair<uint16_t, Pixi_interface*>(stream_id, interface));
      TCP_MSG_INFO("pixi-nr-interface for control-plane is created for stream-id=%lu", (uint32_t)stream_id);
    } else {
      up_interfaces_.insert(std::pair<uint16_t, Pixi_interface*>(stream_id, interface));
      TCP_MSG_INFO("pixi-nr-interface for user-plane is created for stream-id=%lu", (uint32_t)stream_id);
    }
    return interface;
  }
  
  void Pixi_manager::release_Pixi_interface(uint16_t stream_id, Pixi_interface* interface) {
    std::map<uint16_t, Pixi_interface*>::iterator iter;
    iter = cp_interfaces_.find(stream_id);
    if (iter != cp_interfaces_.end()) {
      delete interface;
      cp_interfaces_.erase(iter);
      TCP_MSG_INFO("pixi-nr-interface for control-plane for stream-id=%lu  is destroyed", (uint32_t)stream_id);
      return;
    }
    iter = up_interfaces_.find(stream_id);
    if (iter != up_interfaces_.end()) {
      delete interface;
      up_interfaces_.erase(iter);
      TCP_MSG_INFO("pixi-nr-interface for user-plane for stream-id=%lu  is destroyed", (uint32_t)stream_id);
      return;
    }
  }
  
  //==========================================================================================
  // Pixi_interface
  //==========================================================================================
  size_t Pixi_interface::pooled_block_size_ = 163780;
  Pixi_interface::Pixi_interface(uint16_t stream_id, bool is_control_plane) {
    Pixi_manager::instance_.zero_copy_ = xtreme::iwf::get_option().zero_copy;
    stream_id_ = stream_id;
    is_control_plane_ = is_control_plane;
    first_frame_event_ = true;
  
    manual_parser_ = new ManualParserDataSink(NULL);
    tcp_thread_ = new xtreme::iwf::TcpEngineThread(stream_id_, get_option().measure_engine, get_option().time_wait, get_option().max_queue,
                                                   get_option().cpu_core);
    tcp_thread_->register_counter_type();
    if (xtreme::iwf::get_option().share_mode) {
      tcp_thread_->Startup();
      tcp_thread_sink_ = new xtreme::iwf::MyDataSink( 0, *tcp_thread_);
      manual_parser_->set_data_sink(tcp_thread_sink_);
    }
  
    new_pooled_block();
  }
  
  Pixi_interface::~Pixi_interface() {
    // current_block_ is an intrusive-ptr of a PiBlock.  When the ref-count drops to 0, the vector
    // of buffers in the PiBlock will be returned back to napatech via a call to
    // Pixi_interface::release_nt_buffers().
    // Without the next line, the ref-count will drop to 0 at the end of this destructor.  At that
    // time, calling release_nt_buffers() may cause a crash since this Pixi_interface instance
    // is already deleted.
    // The next line forces an early call to release_nt_buffers() of the dying instance.
    current_block_ = nullptr;
    if (tcp_thread_) {
      delete tcp_thread_;
      tcp_thread_ = NULL;
    }
    if (xtreme::iwf::get_option().share_mode) {
      delete tcp_thread_sink_;
      tcp_thread_sink_ = NULL;
    }
    delete manual_parser_;
  }
  
  void Pixi_interface::process_timeout() {
    if (last_timestamp_.IsZero()) {
      last_timestamp_ = Timestamp::SystemNow();
      --last_timestamp_.sec_;
    }
    Timestamp ts(last_timestamp_);
    ts += Duration(0, 10000000ULL);
  
    last_timestamp_ = ts;
  }
  
  int Pixi_interface::process_frame(uint8_t* payload, uint32_t orig_len, uint32_t captured_len, uint32_t packet_time_sec,
                                    uint32_t packet_time_nsec, uint16_t card_id, uint16_t port_id, uint8_t* nt_buffer) {
    // null payload && buffer signals timeout.
    if (nullptr == payload && nullptr == nt_buffer) {
      process_timeout();
      return xtreme::pixi_napatech::kPixiResultOk;
    }
  
    if (current_block_free_ < captured_len) {
      current_block_->length_bytes_ = pooled_block_size_ - current_block_free_;
      if (Pixi_manager::instance_.zero_copy_) {
        TCP_MSG_TRACE(
            "block-end for:%llu  nt-buffers with:%llu; bytes on %lu", current_block_->nt_buffers_.size() % current_block_->length_bytes_ % stream_id_);
        TCP_MSG_TRACE("block-end events outstanding:%llu ", BlockEndEvent::get_outstanding());
        TCP_MSG_TRACE("pi-blocks outstanding:%llu ", PiBlock::get_outstanding());
      } else {
        TCP_MSG_TRACE("flushing pooled events for stream=%d, current_block_free=%llu", stream_id_ % current_block_free_);
      }
      BlockEndEvent* block_end_event = new BlockEndEvent(last_timestamp_, current_block_);
      if (xtreme::iwf::get_option().share_mode) {
        manual_parser_->process(block_end_event);
      }else{
        tcp_thread_->process(block_end_event);
      }
      new_pooled_block();
    }
  
    FrameEvent* frame = NULL;
    if (Pixi_manager::instance_.zero_copy_) {
      frame = new FrameEvent(captured_len, orig_len, card_id, port_id, payload);
      current_block_->nt_buffers_.push_back(nt_buffer);
    } else {
      frame = new FrameEvent(captured_len, orig_len, card_id, port_id, current_block_free_ptr_);
      memcpy(current_block_free_ptr_, payload, captured_len);
    }
    current_block_free_ -= captured_len;
    current_block_free_ptr_ += captured_len;
  
    frame->set_timestamp(Timestamp(packet_time_sec, packet_time_nsec));
  
  /*  // check for timestamps
    TimeSyncEvent* time_sync = check_for_time_sync(frame->timestamp());
    if (time_sync) {
      manual_parser_->process(time_sync);
    }*/
    manual_parser_->process(frame);
    last_timestamp_ = frame->timestamp();
    if (!xtreme::iwf::get_option().share_mode) {
      tcp_thread_->process(frame);
    }
  
    if (Pixi_manager::instance_.zero_copy_) {
      return xtreme::pixi_napatech::kPixiResultInUse;
    } else {
      return xtreme::pixi_napatech::kPixiResultOk;
    }
  }
  
  void Pixi_interface::new_pooled_block() {
    TCP_MSG_TRACE("creating new pooled block for stream=%lu, block_size=%llu" , stream_id_ % pooled_block_size_);
    if (Pixi_manager::instance_.zero_copy_) {
      current_block_ = new PiBlock(stream_id_, this, g_release_nt_buffers);
    } else {
      current_block_ = FrameBlockPool::get(0, pooled_block_size_, nullptr);
    }
    current_block_free_ = pooled_block_size_;
    current_block_free_ptr_ = current_block_->p_data_;
  }
  
  TimeSyncEvent* Pixi_interface::check_for_time_sync(const Timestamp &ts_in) {
    if (first_frame_event_) {
      first_frame_event_ = false;
      Timestamp ts = get_time_sync_timestamp(ts_in);
      // ts.nsec_ = 0;
      return new TimeSyncEvent(ts);
    }
  
    Duration diff = ts_in - next_time_sync_time_;
    if (diff.nsecs_ >= 0) {
      Timestamp ts = get_time_sync_timestamp(ts_in);
      return new TimeSyncEvent(ts);
    }
    return nullptr;
  }
  
  Timestamp Pixi_interface::get_time_sync_timestamp(const Timestamp &ts_in) {
    Timestamp ts;
    ts.sec_ = ts_in.sec_;
    ts.nsec_ = 0;
    while (ts.nsec_ + time_sync_period_ <= ts_in.nsec_) {
      ts.nsec_ += time_sync_period_;
    }
    while (ts.nsec_ >= (1000 * 1000 * 1000)) {
      ++ts.sec_;
      ts.nsec_ -= (1000 * 1000 * 1000);
    }
  
    next_time_sync_time_ = ts;
    next_time_sync_time_.nsec_ += time_sync_period_;
    if (next_time_sync_time_.nsec_ >= (1000 * 1000 * 1000)) {
      ++next_time_sync_time_.sec_;
      next_time_sync_time_.nsec_ -= (1000 * 1000 * 1000);
    }
  
    return ts;
  }
  
  void Pixi_interface::release_nt_buffers(PiDataLenBytes len, const std::vector<uint8_t*>& nt_buffers) {
    TCP_MSG_TRACE("%llu nt-buffers occupying %lu bytes can now be released on %d ", nt_buffers.size() % len % stream_id_);
    nt_buffers_.push(new std::vector<uint8_t*>(nt_buffers));
  }
  
  std::vector<uint8_t*>* Pixi_interface::get_released_nt_buffers() {
    if (nt_buffers_.empty()) {
      return nullptr;
    }
    std::vector<uint8_t*>* batch = nullptr;
    if (nt_buffers_.try_pop(batch)) {
      TCP_MSG_TRACE("returning:%llu nt-buffers to napatech on %d", batch->size()% stream_id_);
      return batch;
    }
    return nullptr;
  }
  
  void Pixi_interface::send_timeout(uint16_t stream_id) {
    process_frame(NULL, 0, 0, 0, 0, 0, 0, NULL);
  }
  
  int Pixi_interface::send_frame(uint16_t stream_id, uint8_t* opaque_data) {
    NtNetBuf_t hNetBuf = reinterpret_cast<NtNetBuf_t>(opaque_data);
    uint8_t* payload = reinterpret_cast<uint8_t*>(NT_NET_GET_PKT_L2_PTR(hNetBuf));
    uint32_t packet_length = NT_NET_GET_PKT_WIRE_LENGTH(hNetBuf);
  #if 0
    uint32_t captured_length = NT_NET_GET_PKT_CAP_LENGTH(hNetBuf);
  #else
    uint32_t captured_length = packet_length;
  #endif
    uint32_t packet_time_secs = NT_NET_GET_PKT_TIMESTAMP(hNetBuf) / 100000000;
    uint32_t packet_time_nsecs = (NT_NET_GET_PKT_TIMESTAMP(hNetBuf) % 100000000) * 10;
    uint16_t card_id = 0;
    uint16_t port_id = NT_NET_GET_PKT_RXPORT(hNetBuf);
    return process_frame(payload, packet_length, captured_length, packet_time_secs, packet_time_nsecs, card_id, port_id, opaque_data);
  }
  
  void Pixi_interface::shut_down(){
    if (tcp_thread_) {
      tcp_thread_->stop();
    }
  }
  
  }
  }
  