text: ﻿/**
   * Tencent is pleased to support the open source community by making Tars available.
   *
   * Copyright (C) 2016THL A29 Limited, a Tencent company. All rights reserved.
   *
   * Licensed under the BSD 3-Clause License (the "License"); you may not use this file except 
   * in compliance with the License. You may obtain a copy of the License at
   *
   * https://opensource.org/licenses/BSD-3-Clause
   *
   * Unless required by applicable law or agreed to in writing, software distributed 
   * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR 
   * CONDITIONS OF ANY KIND, either express or implied. See the License for the 
   * specific language governing permissions and limitations under the License.
   */
  #ifndef __TC_THREAD_QUEUE_H_
  #define __TC_THREAD_QUEUE_H_
  
  #include <deque>
  #include <vector>
  #include <cassert>
  #include <mutex>
  #include <condition_variable>
  
  using namespace std;
  
  namespace horsedb
  {
  /////////////////////////////////////////////////
  /** 
   * @file tc_thread_queue.h
   * @brief 线程队列类.
   *  
   * @author jarodruan@upchina.com
   */
         
  /////////////////////////////////////////////////
  /**
   * @brief 线程安全队列
   */
  template<typename T, typename D = deque<T> >
  class TC_ThreadQueue
  {
  public:
      TC_ThreadQueue():_size(0){};
  
  public:
  
      typedef D queue_type;
  
  	/**
  	 * @brief 从头部获取数据, 没有数据抛异常
  	 *
  	 * @param t
  	 * @return bool: true, 获取了数据, false, 无数据
  	 */
  	T front();
  
      /**
       * @brief 从头部获取数据, 没有数据则等待.
       *
       * @param t 
  	 * @param millsecond(wait = true时才生效)  阻塞等待时间(ms)
  	 *                    0 表示不阻塞 
       * 					 -1 永久等待
       * @param wait, 是否wait
       * @return bool: true, 获取了数据, false, 无数据
       */
      bool pop_front(T& t, size_t millsecond = 0, bool wait = true);
  
  	/**
  	 * @brief 从头部获取数据.
  	 *
  	 * @return bool: true, 获取了数据, false, 无数据
  	 */
  	bool pop_front();
  
      /**
       * @brief 通知等待在队列上面的线程都醒过来
       */
      void notifyT();
  
      /**
  	 * @brief 放数据到队列后端. 
  	 *  
       * @param t
       */
      void push_back(const T& t, bool notify = true);
  
      /**
  	 * @brief  放数据到队列后端. 
  	 *  
       * @param vt
       */
      void push_back(const queue_type &qt, bool notify = true);
  
      /**
  	 * @brief  放数据到队列前端. 
  	 *  
       * @param t
       */
      void push_front(const T& t, bool notify = true);
  
      /**
  	 * @brief  放数据到队列前端. 
  	 *  
       * @param vt
       */
      void push_front(const queue_type &qt, bool notify = true);
  
      /**
  	 * @brief  交换数据
  	 *  
       * @param q
  	 * @param millsecond(wait = true时才生效)  阻塞等待时间(ms)
  	 *                   0 表示不阻塞 
       * 					 -1 如果为则永久等待
       * @param 是否等待有数据
       * @return 有数据返回true, 无数据返回false
       */
      bool swap(queue_type &q, size_t millsecond = 0, bool wait = true);
  
      /**
       * @brief  队列大小.
       *
       * @return size_t 队列大小
       */
      size_t size() const;
  
      /**
       * @brief  清空队列
       */
      void clear();
  
      /**
       * @brief  是否数据为空.
       *
       * @return bool 为空返回true，否则返回false
       */
      bool empty() const;
  
  protected:
  	TC_ThreadQueue(const TC_ThreadQueue&) = delete;
  	TC_ThreadQueue(TC_ThreadQueue&&) = delete;
  	TC_ThreadQueue& operator=(const TC_ThreadQueue&) = delete;
  	TC_ThreadQueue& operator=(TC_ThreadQueue&&) = delete;
  
  protected:
      /**
       * 队列
       */
      queue_type          _queue;
  
      /**
       * 队列长度
       */
      size_t              _size;
  
      //条件变量
  	std::condition_variable _cond;
  
  	//锁
      mutable std::mutex _mutex;
  };
  
  template<typename T, typename D> T TC_ThreadQueue<T, D>::front()
  {
  	std::unique_lock<std::mutex> lock(_mutex);
  
  	return  _queue.front();
  }
  
  template<typename T, typename D> bool TC_ThreadQueue<T, D>::pop_front(T& t, size_t millsecond, bool wait)
  {
      if(wait) {
  
          std::unique_lock<std::mutex> lock(_mutex);
  
          if (_queue.empty()) {
              if (millsecond == 0) {
                  return false;
              }
              if (millsecond == (size_t) -1) {
                  _cond.wait(lock);
              }
              else {
                  //超时了
                  if (_cond.wait_for(lock, std::chrono::milliseconds(millsecond)) == std::cv_status::timeout) {
                      return false;
                  }
              }
          }
  
          if (_queue.empty()) {
              return false;
          }
  
          t = _queue.front();
          _queue.pop_front();
          assert(_size > 0);
          --_size;
  
          return true;
      }
      else
      {
          std::lock_guard<std::mutex> lock (_mutex);
          if (_queue.empty())
          {
              return false;
          }
  
          t = _queue.front();
          _queue.pop_front();
          assert(_size > 0);
          --_size;
  
          return true;
      }
  }
  
  
  template<typename T, typename D> bool TC_ThreadQueue<T, D>::pop_front()
  {
  	std::unique_lock<std::mutex> lock(_mutex);
  	if (_queue.empty())
  	{
  		return false;
  	}
  
  	_queue.pop_front();
  	assert(_size > 0);
  	--_size;
  
  	return true;
  }
  
  template<typename T, typename D> void TC_ThreadQueue<T, D>::notifyT()
  {
  	std::unique_lock<std::mutex> lock(_mutex);
  	_cond.notify_all();
  }
  
  template<typename T, typename D> void TC_ThreadQueue<T, D>::push_back(const T& t, bool notify)
  {
      if(notify) {
          std::unique_lock<std::mutex> lock(_mutex);
  
          _queue.push_back(t);
          ++_size;
  
          _cond.notify_one();
      }
      else
      {
          std::lock_guard<std::mutex> lock (_mutex);
          _queue.push_back(t);
          ++_size;
      }
  }
  
  template<typename T, typename D> void TC_ThreadQueue<T, D>::push_back(const queue_type &qt, bool notify)
  {
      if(notify) {
          std::unique_lock<std::mutex> lock(_mutex);
  
          typename queue_type::const_iterator it = qt.begin();
          typename queue_type::const_iterator itEnd = qt.end();
          while (it != itEnd) {
              _queue.push_back(*it);
              ++it;
              ++_size;
          }
          _cond.notify_all();
      }
      else
      {
          std::lock_guard<std::mutex> lock (_mutex);
  
          typename queue_type::const_iterator it = qt.begin();
          typename queue_type::const_iterator itEnd = qt.end();
          while (it != itEnd) {
              _queue.push_back(*it);
              ++it;
              ++_size;
          }
      }
  }
  
  template<typename T, typename D> void TC_ThreadQueue<T, D>::push_front(const T& t, bool notify)
  {
      if(notify) {
          std::unique_lock<std::mutex> lock(_mutex);
  
          _cond.notify_one();
  
          _queue.push_front(t);
  
          ++_size;
      }
      else
      {
          std::lock_guard<std::mutex> lock (_mutex);
  
          _queue.push_front(t);
  
          ++_size;
      }
  }
  
  template<typename T, typename D> void TC_ThreadQueue<T, D>::push_front(const queue_type &qt, bool notify)
  {
      if(notify) {
          std::unique_lock<std::mutex> lock(_mutex);
  
          typename queue_type::const_iterator it = qt.begin();
          typename queue_type::const_iterator itEnd = qt.end();
          while (it != itEnd) {
              _queue.push_front(*it);
              ++it;
              ++_size;
  
          }
  
          _cond.notify_all();
      }
      else
      {
          std::lock_guard<std::mutex> lock (_mutex);
  
          typename queue_type::const_iterator it = qt.begin();
          typename queue_type::const_iterator itEnd = qt.end();
          while (it != itEnd) {
              _queue.push_front(*it);
              ++it;
              ++_size;
  
          }
      }
  }
  
  template<typename T, typename D> bool TC_ThreadQueue<T, D>::swap(queue_type &q, size_t millsecond, bool wait)
  {
      if(wait) {
          std::unique_lock<std::mutex> lock(_mutex);
  
          if (_queue.empty()) {
              if (millsecond == 0) {
                  return false;
              }
              if (millsecond == (size_t) -1) {
                  _cond.wait(lock);
              }
              else {
                  //超时了
                  if (_cond.wait_for(lock, std::chrono::milliseconds(millsecond)) == std::cv_status::timeout) {
                      return false;
                  }
              }
          }
  
          if (_queue.empty()) {
              return false;
          }
  
          q.swap(_queue);
          _size = _queue.size();
  
          return true;
      }
      else
      {
          std::lock_guard<std::mutex> lock (_mutex);
  
          if (_queue.empty()) {
              return false;
          }
  
          q.swap(_queue);
  
          _size = _queue.size();
  
          return true;
      }
  }
  
  template<typename T, typename D> size_t TC_ThreadQueue<T, D>::size() const
  {
  	std::lock_guard<std::mutex> lock(_mutex);
      return _size;
  }
  
  template<typename T, typename D> void TC_ThreadQueue<T, D>::clear()
  {
  	std::lock_guard<std::mutex> lock(_mutex);
      _queue.clear();
      _size = 0;
  }
  
  template<typename T, typename D> bool TC_ThreadQueue<T, D>::empty() const
  {
  	std::lock_guard<std::mutex> lock(_mutex);
      return _queue.empty();
  }
  
  }
  #endif
  
  