ringbuffer.hpp

/*
 * Copyright 2023 The IceFlow Authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * 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.
 *
 * SPDX-License-Identifier: Apache-2.0
 */
 
#ifndef ICEFLOW_CORE_RINGBUFFER_HPP
#define ICEFLOW_CORE_RINGBUFFER_HPP
 
#include <condition_variable>
 
#include "boost/circular_buffer.hpp"
 
namespace iceflow {
 
template <typename T>
 
class RingBuffer {
public:
  //   TODO: make it configurable - pass this parameter from the Config
  //   file/Constructor
  RingBuffer() : m_queue(10000) {}
 
  RingBuffer &operator=(const RingBuffer &) = delete;
 
  RingBuffer(RingBuffer const &other) {
    std::lock_guard<std::mutex> lock(other.m_mutex);
    m_queue = other.m_queue;
  }
 
  int size() const { // queue size
    std::lock_guard<std::mutex> lock(m_mutex);
    int x = static_cast<int>(m_queue.size());
    return x;
  }
 
  void push(const T &value) {
    {
      std::unique_lock<std::mutex> lock(m_mutex);
      m_queue.push_back(value);
      lock.unlock();
    }
    m_queueCondition.notify_all();
  }
  void pushData(T &value, int threshold) {
    {
      std::unique_lock<std::mutex> lock(m_mutex);
      while (m_queue.size() >= threshold) {
        m_queueCondition.wait(lock);
      }
      m_queue.push_back(value);
      lock.unlock();
    }
    m_queueCondition.notify_all();
  }
 
  bool empty() const {
    std::lock_guard<std::mutex> lock(m_mutex);
    return m_queue.empty();
  }
 
  bool tryAndPop(T &value) {
    std::lock_guard<std::mutex> lock(m_mutex);
    if (m_queue.empty())
      return false;
    value = m_queue.front();
    m_queue.pop();
    return true;
  }
 
  std::shared_ptr<T> tryAndPop() {
    std::lock_guard<std::mutex> lock(m_mutex);
    if (m_queue.empty())
      return std::shared_ptr<T>();
    std::shared_ptr<T> value = std::make_shared<T>(m_queue.front());
    m_queue.pop();
    return value;
  }
 
  void waitAndPop(T &value) {
    std::unique_lock<std::mutex> lock(m_mutex);
    m_queueCondition.wait(lock, [this] { return !m_queue.empty(); });
    value = m_queue.front();
    m_queue.pop();
  }
 
  std::shared_ptr<T> waitAndPop() {
    std::unique_lock<std::mutex> lock(m_mutex);
    m_queueCondition.wait(lock, [this] { return !m_queue.empty(); });
    std::shared_ptr<T> value = std::make_shared<T>(m_queue.front());
    m_queue.pop();
    return value;
  }
 
  T waitAndPopValue() {
    std::unique_lock<std::mutex> lock(m_mutex);
    m_queueCondition.wait(lock, [this] { return !m_queue.empty(); });
    T value = m_queue.front();
    m_queue.pop_front();
    lock.unlock();
    m_queueCondition.notify_all();
    return value;
  }
 
  bool waitForAndPop(T &value, std::chrono::milliseconds waitDuration) {
    std::unique_lock<std::mutex> lock(m_mutex);
    if (m_queueCondition.wait_for(lock, waitDuration,
                                  [this] { return !m_queue.empty(); })) {
      // Got something
      value = m_queue.front();
      m_queue.pop();
      return true;
    }
    // Timed out
    return false;
  }
 
  std::shared_ptr<T> waitForAndPop(std::chrono::milliseconds waitDuration) {
    std::unique_lock<std::mutex> lock(m_mutex);
    if (m_queueCondition.wait_for(lock, waitDuration,
                                  [this] { return !m_queue.empty(); })) {
      // Got something
      std::shared_ptr<T> value = std::make_shared<T>(m_queue.front());
      m_queue.pop();
      return value;
    }
    // Timed out
    return std::shared_ptr<T>();
  }
 
private:
  mutable std::mutex m_mutex;
  boost::circular_buffer<T> m_queue;
 
  std::condition_variable m_queueCondition;
};
 
} // namespace iceflow
 
#endif // ICEFLOW_CORE_RINGBUFFER_HPP