/*
 * .============.
 * // M A K E / \
 * // C++ DEV / \
 * // E A S Y / \/ \
 * ++ ----------. \/\ .
 * \\ \ \ /\ /
 * \\ \ \ /
 * \\ \ \ /
 * -============'
 *
 * Copyright (c) 2018 Hevake and contributors, all rights reserved.
 *
 * This file is part of cpp-tbox (https://github.com/cpp-main/cpp-tbox)
 * Use of this source code is governed by MIT license that can be found
 * in the LICENSE file in the root of the source tree. All contributing
 * project authors may be found in the CONTRIBUTORS.md file in the root
 * of the source tree.
 */
#include "async_pipe.h"
#include <tbox/base/defines.h>

#include <cstring>
#include <cassert>

#include <vector>
#include <deque>
#include <chrono>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <iostream>

namespace tbox {
namespace util {

using namespace std;

class AsyncPipe::Impl {
 public:
 class Buffer {
 public:
 Buffer(size_t cap);
 ~Buffer();

 NONCOPYABLE(Buffer);
 IMMOVABLE(Buffer);

 public:
 size_t append(const void *data_ptr, size_t data_size);

 inline bool full() const { return capacity_ == size_; }
 inline bool empty() const { return size_ == 0; }
 inline void *data() const { return data_; }
 inline size_t size() const { return size_; }
 inline void reset() { size_ = 0; }

 private:
 size_t capacity_;
 size_t size_ = 0;
 uint8_t *data_ = nullptr;
 };

 public:
 Impl();
 ~Impl();

 bool initialize(const Config &cfg);
 void setCallback(const Callback &cb) { cb_ = cb; }
 void cleanup();
 void append(const void *data_ptr, size_t data_size);
 void appendLock();
 void appendUnlock();
 void appendLockless(const void *data_ptr, size_t data_size);

 protected:
 void threadFunc();

 private:
 Config cfg_;
 Callback cb_;

 Buffer* curr_buffer_ = nullptr; //!< 当前缓冲
 vector<Buffer*> free_buffers_; //!< 可用缓冲数组
 deque<Buffer*> full_buffers_; //!< 已满缓冲队列
 size_t buff_num_; //!< 缓冲个数

 bool inited_ = false; //!< 是否已经启动子线程
 bool stop_signal_ = false; //!< 停止信号
 thread backend_thread_;

 mutex curr_buffer_mutex_; //!< 锁 curr_buffer_ 的
 mutex full_buffers_mutex_; //!< 锁 full_buffers_ 的
 mutex free_buffers_mutex_; //!< 锁 free_buffers_ 的
 mutex buff_num_mutex_; //!< 锁 buff_num_ 的
 condition_variable full_buffers_cv_; //!< full_buffers_ 不为空条件变量
 condition_variable free_buffers_cv_; //!< free_buffers_ 不为空条件变量
};

AsyncPipe::Impl::Buffer::Buffer(size_t cap) :
 capacity_(cap)
{
 data_ = new uint8_t [cap];
}

AsyncPipe::Impl::Buffer::~Buffer()
{
 delete [] data_;
}

size_t AsyncPipe::Impl::Buffer::append(const void *data_ptr, size_t data_size)
{
 size_t wsize = data_size;
 if ((size_ + data_size) > capacity_)
 wsize = capacity_ - size_;

 ::memcpy((data_ + size_), data_ptr, wsize);
 size_ += wsize;

 return wsize;
}

AsyncPipe::AsyncPipe() :
 impl_(new Impl)
{ }

AsyncPipe::~AsyncPipe()
{
 delete impl_;
}

bool AsyncPipe::initialize(const Config &cfg)
{
 return impl_->initialize(cfg);
}

void AsyncPipe::setCallback(const Callback &cb)
{
 impl_->setCallback(cb);
}

void AsyncPipe::cleanup()
{
 impl_->cleanup();
}

void AsyncPipe::append(const void *data_ptr, size_t data_size)
{
 impl_->append(data_ptr, data_size);
}

void AsyncPipe::appendLock()
{
 impl_->appendLock();
}

void AsyncPipe::appendUnlock()
{
 impl_->appendUnlock();
}

void AsyncPipe::appendLockless(const void *data_ptr, size_t data_size)
{
 impl_->appendLockless(data_ptr, data_size);
}

AsyncPipe::Impl::Impl()
{ }

AsyncPipe::Impl::~Impl()
{
 cleanup();
}

bool AsyncPipe::Impl::initialize(const Config &cfg)
{
 if (cfg.buff_size == 0) {
 std::cerr << "Err: AsyncPipe::Config::buff_size == 0" << std::endl;
 return false;
 }

 if (cfg.buff_min_num == 0) {
 std::cerr << "Err: AsyncPipe::Config::buff_min_num == 0" << std::endl;
 return false;
 }

 if (cfg.buff_min_num > cfg.buff_max_num) {
 std::cerr << "Err: AsyncPipe::Config::buff_max_num < buff_min_num" << std::endl;
 return false;
 }

 if (cfg.interval == 0) {
 std::cerr << "Err: AsyncPipe::Config::interval == 0" << std::endl;
 return false;
 }

 cfg_ = cfg;

 free_buffers_.reserve(cfg.buff_min_num);
 for (size_t i = 0; i < cfg.buff_min_num; ++i)
 free_buffers_.push_back(new Buffer(cfg.buff_size));
 buff_num_ = cfg.buff_min_num;

 auto bt = thread(std::bind(&AsyncPipe::Impl::threadFunc, this));
 backend_thread_.swap(bt);
 inited_ = true;
 return true;
}

void AsyncPipe::Impl::cleanup()
{
 if (!inited_)
 return;

 stop_signal_ = true;
 full_buffers_cv_.notify_all();
 backend_thread_.join();
 stop_signal_ = false;

 assert(full_buffers_.empty());
 CHECK_DELETE_RESET_OBJ(curr_buffer_);
 for (auto item : free_buffers_)
 CHECK_DELETE_RESET_OBJ(item);
 free_buffers_.clear();

 cb_ = nullptr;
 inited_ = false;
}

void AsyncPipe::Impl::append(const void *data_ptr, size_t data_size)
{
 std::lock_guard<std::mutex> lg(curr_buffer_mutex_);
 appendLockless(data_ptr, data_size);
}

void AsyncPipe::Impl::appendLock()
{
 curr_buffer_mutex_.lock();
}

void AsyncPipe::Impl::appendUnlock()
{
 curr_buffer_mutex_.unlock();
}

void AsyncPipe::Impl::appendLockless(const void *data_ptr, size_t data_size)
{
 const uint8_t *ptr = static_cast<const uint8_t*>(data_ptr);
 size_t remain_size = data_size;

 while (remain_size > 0) {
 if (curr_buffer_ == nullptr) {
 //! 如果 curr_buffer_ 没有分配,则应该从 free_buffers_ 中取一个出来
 std::unique_lock<std::mutex> lk(free_buffers_mutex_);
 if (free_buffers_.empty()) { //! 如里 free_buffers_ 为空
 buff_num_mutex_.lock();
 //! 如果缓冲块数还没有达到最大限值,则可以继续申请
 if (buff_num_ < cfg_.buff_max_num) {
 ++buff_num_;
 buff_num_mutex_.unlock();
 free_buffers_.push_back(new Buffer(cfg_.buff_size));
 } else { //! 否则只能等待后端释放
 buff_num_mutex_.unlock();
 free_buffers_cv_.wait(lk, [this] { return !free_buffers_.empty(); });
 }
 }

 //! 将 free_buffers_ 中最后的一个弹出来,给到 curr_buffer_
 curr_buffer_ = free_buffers_.back();
 free_buffers_.pop_back();
 //! Q: 为什么从 free_buffers_ 尾部取,而不是向 full_buffers_ 那样从头部取呢?
 //! A: 因为 free_buffers_ 所存空闲缓冲,没有顺序要求。而 full_buffers_ 必须要有顺序性
 //! 既然不需要顺序性,那么 vector 的尾部进出是最高效的。
 }
 auto size = curr_buffer_->append(ptr, remain_size);
 if (curr_buffer_->full()) {
 //! 如果当前缓冲满了
 std::lock_guard<std::mutex> lg2(full_buffers_mutex_);
 full_buffers_.push_back(curr_buffer_); //! 将 curr_buffer_ 放到 full_buffers_ 中
 full_buffers_cv_.notify_all(); //! 通知后台线程开始干活
 curr_buffer_ = nullptr;
 }
 ptr += size;
 remain_size -= size;
 }
}

void AsyncPipe::Impl::threadFunc()
{
 for (;;) {
 bool is_wake_for_quit = false; //! 是否因需要停止而被唤醒
 bool is_wake_for_timeup = true; //! 是否因超时而被唤醒

 {
 //! 等待唤醒信号
 std::unique_lock<std::mutex> lk(full_buffers_mutex_);
 if (full_buffers_.empty()) {
 //! 等待三种情况: 1.超时,2.停止,3.full_buffers_不为空
 full_buffers_cv_.wait_for(lk, std::chrono::milliseconds(cfg_.interval),
 [this, &is_wake_for_timeup, &is_wake_for_quit] {
 if (stop_signal_)
 is_wake_for_quit = true;

 if (is_wake_for_quit || !full_buffers_.empty()) {
 is_wake_for_timeup = false;
 return true;
 }
 return false;
 }
 );
 } else {
 is_wake_for_timeup = false;
 }
 }

 //! 如果是超时或是收到停止信号,则先将 curr_buff_ 移到 full_buffers_
 if (is_wake_for_timeup || is_wake_for_quit) {
 if (curr_buffer_mutex_.try_lock()) {
 if (curr_buffer_ != nullptr) {
 //! Q: 这里为什么不锁 full_buffers_mutex_ ?
 //! A: 因为锁住了 curr_buffer_mutex_ 就不会有前端调用 appendLockless(),仅有后端的线程操作。
 //! 所以不锁 full_buffers_mutex_ 也是安全的
 full_buffers_.push_back(curr_buffer_);
 curr_buffer_ = nullptr;
 }
 curr_buffer_mutex_.unlock();
 }
 }

 //! 然后逐一处理 full_buffers_ 中的数据
 for (;;) {
 Buffer *buff = nullptr;
 {
 //! 尝试从 full_buffers_ 中取一个出来
 std::lock_guard<std::mutex> lg(full_buffers_mutex_);
 if (!full_buffers_.empty()) {
 buff = full_buffers_.front();
 full_buffers_.pop_front();
 } else {
 break;
 }
 }

 if (buff != nullptr) {
 //! 进行处理
 if (cb_)
 cb_(buff->data(), buff->size());
 buff->reset();

 buff_num_mutex_.lock();
 if (buff_num_ > cfg_.buff_min_num) {
 --buff_num_;
 buff_num_mutex_.unlock();
 delete buff;
 } else {
 buff_num_mutex_.unlock();
 //! 将处理后的缓冲放回 free_buffers_ 中
 std::lock_guard<std::mutex> lg(free_buffers_mutex_);
 free_buffers_.push_back(buff);
 free_buffers_cv_.notify_all();
 }
 }
 }

 if (is_wake_for_quit)
 break;
 }
}

}
}