text: /*
   * DFS_queues.cpp
   *
   *  Created on: Jul 10, 2013
   *      Author: root
   */
  
  #define LINUX
  
  #include "xtreme/iwf/dfs_queues.h"
  #include "xtreme/iwf/dfs_writer.h"
  #include "xtreme/iwf/internal/dfs_file_aio.h"
  #include "xtreme/iwf/internal/dfs_logger.h"
  #include <time.h>
  
  namespace xtreme {
  namespace iwf {
  
  const uint TIME_FMT = strlen("2012-12-31 12:59:59.123456789") + 1;
  #define NANO 1000000000L
  
  #define ROP_FRAME_INTERVAL_FOR_LOG 1000
  
  // buf needs to store 30 characters
  int timestamp2str(char *buf, uint len, Timestamp *timestamp) {
    int ret;
    struct tm t;
    struct timespec ts;
    ts.tv_sec = timestamp->sec_;
    ts.tv_nsec = timestamp->nsec_;
    tzset();
    if (localtime_r(&(ts.tv_sec), &t) == NULL)
      return 1;
  
    ret = strftime(buf, len, "%F %T", &t);
    if (ret == 0)
      return 2;
    len -= ret - 1;
  
    ret = snprintf(&buf[strlen(buf)], len, ".%09ld", ts.tv_nsec);
    if (ret >= (int) len)
      return 3;
  
    return 0;
  }
  
  bool DFS_queue::can_push_xtreme_block() {
    return ((block_queue_.size() > capacity_ + 100) ? false : true);
  }
  
  void DFS_queue::push(xtreme::FrameEvent* event) {
    tbb::spin_mutex::scoped_lock lock(spinlock_);
    if (event == NULL) {
      tbb::spin_mutex::scoped_lock lock_tm(latest_timestamp_lock_);
      stats_.drop_frame_interval_for_log_++;
      return;
    }
    DfsFramePtr dfsframe = DfsFramePtr(new DFS_frame(event));
    current_xtreme_blocks_.push_back(dfsframe);
  }
  
  void DFS_queue::flush() {
    tbb::spin_mutex::scoped_lock lock(spinlock_);
  
    DfsFrameBlocks_ptr frame_blocks = DfsFrameBlocks_ptr(new DfsFrameBlocks);
    // make copy to Frames
    frame_blocks->assign(current_xtreme_blocks_.begin(), current_xtreme_blocks_.end());
  /*  for (XtremeBlocks::iterator it = current_xtreme_blocks_.begin();
        it != current_xtreme_blocks_.end(); it++) {
      DfsFramePtr dfsframe = DfsFramePtr(new DFS_frame((*it)));
      frame_blocks->push_back(dfsframe);
    }*/
    block_queue_.push_back(frame_blocks);
    current_xtreme_blocks_.clear();
  }
  
  Timestamp DFS_queue::begin_time(uint32_t & delete_a_block) {
    tbb::spin_mutex::scoped_lock lock(spinlock_);
    if (block_queue_.empty()) {
      return Timestamp(0);
    }
  
    std::deque<DfsFrameBlocks_ptr>::iterator it = block_queue_.begin();
  
    DfsFrameBlocks_ptr xb;
    for (; it != block_queue_.end();) {
      //assume the block is not empty
      xb = (*it);
      if (xb == NULL) {
        it = block_queue_.erase(it);
        continue;
      }
      if (0 == xb->size()) {
        it = block_queue_.erase(it);
        delete xb;
        xb = NULL;
        delete_a_block += 1;
        if (it != block_queue_.end()) {
          xb = (*it);
        } else {
          break;
        }
        continue;
      }
      break;  //just need the first xtreme_block and fullfill the conditions.
    }
  
    if (xb == NULL) {
      return Timestamp(0);
    }
    return xb->front()->timestamp();
  }
  
  
  DfsFramePtr DFS_queue::copy_out_first(uint32_t & delete_a_block) {
    tbb::spin_mutex::scoped_lock lock(spinlock_);
    if (block_queue_.empty()) {
      return NULL;
    }
  
    std::deque<DfsFrameBlocks_ptr>::iterator it = block_queue_.begin();
  
    DfsFrameBlocks_ptr xb;
    for (; it != block_queue_.end();) {
      //assume the block is not empty
      xb = (*it);
      if (xb == NULL) {
        it = block_queue_.erase(it);
        continue;
      }
      if (0 == xb->size()) {
        it = block_queue_.erase(it);
        delete xb;
        xb = NULL;
        delete_a_block += 1;
        if (it != block_queue_.end()) {
          xb = (*it);
        } else {
          break;
        }
        continue;
      }
      break;  //just need the first xtreme_block and fullfill the conditions.
    }
  
    if (NULL == xb) {
      return NULL;
    }
  
    DfsFramePtr pt_frame = xb->front();
    xb->pop_front();
    DFS_MSG_TRACE("current xtremeBlock size = %llu", xb->size());
    if (pt_frame != nullptr) {
      tbb::spin_mutex::scoped_lock lock_tm(latest_timestamp_lock_);
      stats_.latest_frame_timestamp_ = pt_frame->timestamp();
    }
    return pt_frame;
  }
  
  tbb::spin_mutex & DFS_queue::latest_timestamp_lock() {
    return latest_timestamp_lock_;
  }
  
  DFS_Telemetry_stats & DFS_queue::telemetry_stats() {
    return stats_;
  }
  
  bool DFS_queue::empty() {
    if (block_queue_.empty()) {
      return true;
    }
    return false;
  }
  
  uint64_t DFS_queue::clear() {
    uint64_t nSize = 0;
    tbb::spin_mutex::scoped_lock lock(spinlock_);
    for (std::deque<DfsFrameBlocks_ptr>::iterator it = block_queue_.begin(); it != block_queue_.end(); ++it) {
      DfsFrameBlocks_ptr block = (*it);
      if (NULL != block) {
        nSize += block->size();
        /*
         * call clear will release dfs frame, and destroy
         */
        BOOST_FOREACH(xtreme::iwf::DfsFramePtr dfsframe, *block) {
          delete dfsframe;
          dfsframe = NULL;
        }
        block->clear();
      }
      delete block;
      block = NULL;
    }
    block_queue_.clear();
    return nSize;
  }
  
  DFS_option DFS_queues_merger::dfs_option_;
  
  DFS_queues_merger::DFS_queues_merger()
      : writer_(0),
        bytes_written_(0),
        need_stop_immediately_(false) {
  
  }
  
  DFS_queues_merger::~DFS_queues_merger() {
    stop();
  }
  
  void DFS_queues_merger::Run() {
    Timestamp tm_first, tm_target;
    bytes_written_ = 0;
  
    if (NULL == writer_) {
      DFS_MSG_INFO("dfs will create write File (type=0,recd;type=1,pcap) type=%d", dfs_option_.pcap_);
      // always create  as pcap
      writer_ = new PCAP_writer(dfs_option_.filename_, dfs_option_.splitTime_, dfs_option_.splitSize_,
                                dfs_option_.maxSizeMBytes_, dfs_option_.maxDiskPerc_, dfs_option_.maxDays_);
      // disable below, since 1.0 won;t support rec format.
      /* if (dfs_option_.pcap_) {
       *      writer_ = new PCAP_writer(dfs_option_.filename_, dfs_option_.splitTime_, dfs_option_.splitSize_,
       *           dfs_option_.maxSizeMBytes_, dfs_option_.maxDiskPerc_, dfs_option_.maxDays_);
       * } else {
       *      writer_ = new DFS_writer(dfs_option_.filename_, dfs_option_.splitTime_, dfs_option_.splitSize_,
       *           dfs_option_.maxSizeMBytes_, dfs_option_.maxDiskPerc_, dfs_option_.maxDays_);
       * }*/
      // writer_->next_file();
      std::stringstream ss;
      ss << "dfs has open for writing: " << dfs_option_.filename_ << ", split time:" << dfs_option_.splitTime_ << "(min)"
         << ", split size:" << dfs_option_.splitSize_ << "(Bytes)" << ", file max size:" << dfs_option_.maxSizeMBytes_
         << ", purge after exceeding disk percentage:" << dfs_option_.maxDiskPerc_ << ", max days keep :"
         << dfs_option_.maxDays_;
      DFS_MSG_INFO("%s", ss.str());
    }
    char timestr[TIME_FMT];
    uint64_t nPacketWritten = 0;
    uint32_t ndelete_block = 0;
  
    // store the index of sorted array_of_block_queue_ on asc.
    // this will help to eliminate the copy to frameevent.
    size_t arrEvent_index[array_of_block_queue_.size()];
    // initialize to orginal sequence.
    for(size_t i = 0; i < array_of_block_queue_.size(); i++){
      arrEvent_index[i] = i;
    }
    while (!need_stop_immediately_) {
      ndelete_block = 0;
      memset(timestr, 0, TIME_FMT - 1);
      if (array_of_block_queue_.size() == 0) {
        DFS_MSG_ERROR("block queue is empty, %s", "will return");
        break;
      } else if (array_of_block_queue_.size() == 1) {
        IDFS_queue* pQueue = array_of_block_queue_[0];
        DfsFramePtr pEvent = pQueue->copy_out_first(ndelete_block);
        if (pEvent != NULL) {
          bytes_written_ += writer_->write(0, pEvent);
          //log the total length
          DFS_MSG_TRACE("dfs has written %lld, bytes into pcap file; writtern packets = %llu",
                        bytes_written_ % ++nPacketWritten);
        }
      } else {
  #if 0
        // bubble sort is faster, but not fastest...
        /*
         * need to find the second earlier timestamp since batch writting need this second earlier as latch to ensure the time oerdered packets.
         * e.g
         * Q1: -> , 3rd, 2nd, 1st
         * Q2: -> , 2nd, 1st
         * Q3: -> , 4th, 3rd, 2nd, 1st
         *
         * flush to disk in order like, Q2:1st, Q1:1st, Q3:1st, under the assumption that
         * Q1:2nd is later than Q3:1st, and Q2:2nd is later than Q3:1st. if not, Q3:1st can't be flused into disk.
         * In another words, after bubble sorting.
         * Any of packet in each queue will be flushed into disk under the condition that it must be earlier than other 2nd earlier packets in other queues.
         *
         * so that the detail steps are
         * 1. get the array of earlest packet, and arrary of second earliest packets
         * 2. sort the earlest packet
         * 3. batch to flush sorted earlist packets, but need to compare it with array of second of earliest packets.
         * this way is complicated to the other algo and make performance lower
         * another algo is called "flush earliest packet once pick and remove from its queue"
         */
        size_t i, j, temp;
        i = j = temp = 0;
        for (i = 0; i < array_of_block_queue_.size(); i++) {
          for (j = 0; j < array_of_block_queue_.size() -1 - i; j++) {
            IDFS_queue* pQueue = array_of_block_queue_.at(arrEvent_index[j]);
            if (NULL == pQueue) {
              continue;
            }
            if (pQueue->empty()) {
              continue;
            }
            tm_first = pQueue->begin_time(ndelete_block);
            if (tm_first.IsZero()) {
              // now, we can;t remove the the queue whose begin time is 0, since nobody know when current queue will have frame coming in.
              // array_of_block_queue_.erase(array_of_block_queue_.begin() + st);
              continue;
            }
            IDFS_queue* pQueueTarget = array_of_block_queue_.at(arrEvent_index[j + 1]);
            if (pQueueTarget) {
              tm_target = pQueueTarget->begin_time(ndelete_block);
            }else{
              tm_target.Reset();
            }
            if (Dfs_Logger::dfs_logger_->isTraceEnabled()) {
              timestamp2str(timestr, TIME_FMT, &tm_target);
              std::string strFirstTime(timestr);
              timestamp2str(timestr, TIME_FMT, &tm_first);
              std::string strTargetTime(timestr);
              DFS_MSG_TRACE("i=%llu, j=%llu, tm_first=%s; tm_target=%s", i % j % strFirstTime % strTargetTime);
            }
            if (tm_first > tm_target) {
              temp = arrEvent_index[j];
              arrEvent_index[j] = arrEvent_index[j + 1];
              arrEvent_index[j + 1] = temp;
            }
          }
        }
        size_t num_of_emtpty_queue = 0;
        for(size_t i = 0; i < array_of_block_queue_.size(); i++){
          IDFS_queue* pQueueTarget = array_of_block_queue_.at(arrEvent_index[i]);
          if (pQueueTarget == NULL) {
            continue;
          }
          DfsFramePtr pEvent = pQueueTarget->copy_out_first(ndelete_block);
          if (pEvent != NULL) {
            bytes_written_ += writer_->write(0, pEvent);
            //log the total length
            DFS_MSG_TRACE("dfs has written %lld bytes into pcap file;  writter packets = %llu",
                          bytes_written_ % ++nPacketWritten);
            if (Dfs_Logger::dfs_logger_->isTraceEnabled()) {
              tm_target = pEvent->timestamp();
              timestamp2str(timestr, TIME_FMT, &tm_target);
              std::string strTime(timestr);
              DFS_MSG_TRACE("i= %llu; tm_target=%s",i % strTime);
            }
          }else{
            num_of_emtpty_queue++;
          }
  
          if (0 != ndelete_block) {
            writer_->age_blocks_written() += ndelete_block;
          }
          // reset to orginal sequence
          arrEvent_index[i] = i;
        }
        if (num_of_emtpty_queue == array_of_block_queue_.size()) {
          usleep(100000);
          continue;
        }
  #endif
  
        //old algo, "flush earliest packet once pick and remove from its queue"
        /*
         * note: we actually don;t need the 2nd eariler algo here, since it's different to bubble sort in above.
         * here is deque, remove the front once get the earliest packet
         */
        int nFound_earlest_frame_in_nth_queue = -1;
        int nToDeleteQueue = -1;
        for (size_t st = 0; st < array_of_block_queue_.size() - 1;) {
          IDFS_queue* pQueue = array_of_block_queue_[st];
          if (NULL == pQueue) {
            st++;
            continue;
          }
          if (pQueue->empty()) {
            // TODO..., need to remove from the XtremeBlocks_queue;
            // but, it come to have deques after while, then need to add it again?
            st++;
            continue;
          }
          tm_first = pQueue->begin_time(ndelete_block);
          if (Dfs_Logger::dfs_logger_->isTraceEnabled()) {
            timestamp2str(timestr, TIME_FMT, &tm_first);
            std::string strTime(timestr);
            DFS_MSG_TRACE("st=%llu ; tm_first=%s", st % (strTime));
          }
          if (tm_first.IsZero()) {
            // now, we can;t remove the the queue whose begin time is 0, since nobody know when current queue will have frame coming in.
            // array_of_block_queue_.erase(array_of_block_queue_.begin() + st);
            st++;
            continue;
          }
  
          for (size_t j = st + 1; j < array_of_block_queue_.size(); j++) {
            IDFS_queue* pQueueTarget = array_of_block_queue_[j];
            tm_target = pQueueTarget->begin_time(ndelete_block);
            if (Dfs_Logger::dfs_logger_->isTraceEnabled()) {
              timestamp2str(timestr, TIME_FMT, &tm_target);
              std::string strTime(timestr);
              DFS_MSG_TRACE("j= %llu; tm_target=%s", j % strTime);
            }
            if (tm_target.IsZero()) {            //0 time, means the current queue is empty. erase this and skip
              nToDeleteQueue = j;
              nFound_earlest_frame_in_nth_queue = st;
              continue;
            }
            if (tm_target < tm_first || tm_target == tm_first) {
              tm_first = tm_target;
              DFS_MSG_TRACE("j= %llu; tm_first = tm_target", j);
              nFound_earlest_frame_in_nth_queue = j;
            }
          }
          if (nFound_earlest_frame_in_nth_queue == -1) {
            // if the smallest is not in the other block, then need to stream current 1st block which is not emtpy.
            nFound_earlest_frame_in_nth_queue = st;
          }
          //only need to seach one around.
          break;
        }
  
        IDFS_queue* pQueueTarget = NULL;
        if (nFound_earlest_frame_in_nth_queue >= 0 && !array_of_block_queue_.empty()) {
          DFS_MSG_TRACE("nFound_earlest_frame_in_nth_queue = %d", nFound_earlest_frame_in_nth_queue);
          // no need to think second earlier packet.
          pQueueTarget = array_of_block_queue_.at(nFound_earlest_frame_in_nth_queue);
          if (pQueueTarget == NULL) {
            break;
          }
        }
  
        if (NULL != pQueueTarget) {
          tbb::spin_mutex::scoped_lock lock_tm(writer_lock_);
                  if (writer_ == NULL) {
            break;
          }
          DFS_MSG_TRACE("dfs will write packet from pipeline=%lu", pQueueTarget->pipeline_id());        
          DfsFramePtr pEvent = pQueueTarget->copy_out_first(ndelete_block);
          if (pEvent != NULL) {
            bytes_written_ += writer_->write(0, pEvent);
            //log the total length
            DFS_MSG_DEBUG("dfs has written %lld bytes into pcap file;  writter packets = %llu",
                          bytes_written_ % ++nPacketWritten);
          }
          if (0 != ndelete_block) {
            writer_->age_blocks_written() += ndelete_block;
          }
          delete pEvent;
          pEvent = NULL;
        } else {
          usleep(100000);
          continue;
        }
      }
      if (array_of_block_queue_.empty()) {
        //need to use semphone, better,
        usleep(100000);
      }
    }
  }
  
  void DFS_queues_merger::add_xtremeblock_queue(IDFS_queue* pqueue) {
    array_of_block_queue_.push_back(pqueue);
  }
  
  void DFS_queues_merger::clear_dfs_queue() {
    BOOST_FOREACH( IDFS_queue * queue, array_of_block_queue_ ) {
      if (NULL != queue) {
        uint64_t nSize = queue->clear();
        DFS_MSG_INFO("release unfinished block of packets, sizeof frame=%llu", nSize);
      }
    }
    /*
     * no need to to clear the queue, since queue are still needed at next DFS is enabled
     */
    //array_of_block_queue_.clear();
  }
  
  void DFS_queues_merger::stop() {
    need_stop_immediately_ = true;
    clear_dfs_queue();
    if (writer_) {
      tbb::spin_mutex::scoped_lock lock_tm(writer_lock_);
      writer_->close_all();
      delete writer_;
      writer_ = NULL;
      DFS_MSG_INFO("close writer %s:", "and delete");
    }
  
    Join();
  }
  
  
  size_t DFS_queues_merger::queue_length() {
    return array_of_block_queue_.size();
  }
  
  void DFS_frame::copy_from_frame(FrameEvent * frame_event) {
    try{
      if (frame_event->payload_len() > 0) {
        payload_ = static_cast<uint8_t *>(calloc(frame_event->payload_len(), sizeof(uint8_t)));
        //DFS_MSG_TRACE("dfs frame, %lu bytes are allocated from payload data=%p, to %p",
        //              frame_event->payload_len() % (uint32_t*)frame_event->payload() % (uint32_t*)payload_);
        memcpy(payload_, frame_event->payload(), frame_event->payload_len());
      }
      stored_length_ = frame_event->payload_len();
      original_length_ = frame_event->original_length();
      timestamp_ = frame_event->timestamp();
    }catch(std::bad_alloc &) {
      DFS_MSG_ERROR("exception caught in memory allocating in %s", "copy_from_frame()");
    }
  }
  
  }
  }
  