ClauseBuffer.hpp

#pragma once
 
#include "containers/ClauseExchange.hpp"
#include "utils/Logger.hpp"
#include "utils/Parameters.hpp"
#include <atomic>
#include <boost/lockfree/policies.hpp>
#include <boost/lockfree/queue.hpp>
#include <vector>
class ClauseBuffer
{
  private:
    boost::lockfree::queue<ClauseExchange*, boost::lockfree::fixed_sized<false>> queue;
    std::atomic<size_t> m_size; 
 
  public:
    ClauseBuffer() = delete;
 
    explicit ClauseBuffer(size_t size)
        : queue(size)
        , m_size(0)
    {
    }
 
    ClauseBuffer(const ClauseBuffer&) = delete;
 
    ClauseBuffer(const ClauseBuffer&&) = delete;
 
    ClauseBuffer& operator=(const ClauseBuffer&) = delete;
 
    ClauseBuffer(ClauseBuffer&&) = default;
 
    ClauseBuffer& operator=(ClauseBuffer&&) = default;
 
    ~ClauseBuffer() { clear(); }
 
    bool addClause(ClauseExchangePtr clause)
    {
        ClauseExchange* raw = clause->toRawPtr();
        if (queue.push(raw)) {
            m_size.fetch_add(1, std::memory_order_release);
            return true;
        } else {
            // Reference count is decremented, since we do not store the returned ClauseExchangePtr, thus it goes out of scope
            ClauseExchange::fromRawPtr(raw);
            return false;
        }
    }
 
    size_t addClauses(const std::vector<ClauseExchangePtr>& clauses)
    {
        size_t old_size = m_size.load(std::memory_order_relaxed);
 
        for (const auto& clause : clauses) {
            addClause(clause);
        }
 
        return m_size.load(std::memory_order_acquire) - old_size;
    }
 
    bool tryAddClauseBounded(ClauseExchangePtr clause)
    {
        ClauseExchange* raw = clause->toRawPtr();
        if (queue.bounded_push(raw)) {
            m_size.fetch_add(1, std::memory_order_release);
            return true;
        } else {
            ClauseExchange::fromRawPtr(raw);
            return false;
        }
    }
 
    size_t tryAddClausesBounded(const std::vector<ClauseExchangePtr>& clauses)
    {
        size_t old_size = m_size.load(std::memory_order_relaxed);
        for (const auto& clause : clauses) {
            if (!tryAddClauseBounded(clause)) {
                // If a push fails, we've reached the capacity, so we break
                break;
            }
        }
        return m_size.load(std::memory_order_acquire) - old_size;
    }
 
    bool getClause(ClauseExchangePtr& clause)
    {
        LOGDEBUG3("Size before pop %ld", this->size());
        ClauseExchange* raw;
        if (queue.pop(raw)) {
            clause = ClauseExchange::fromRawPtr(raw);
            m_size.fetch_sub(1, std::memory_order_release);
            return true;
        }
        return false;
    }
 
    void getClauses(std::vector<ClauseExchangePtr>& clauses)
    {
        ClauseExchange* raw;
        while (queue.pop(raw)) {
            clauses.push_back(ClauseExchange::fromRawPtr(raw));
            m_size.fetch_sub(1, std::memory_order_release);
        }
    }
 
    size_t size() const { return m_size.load(std::memory_order_acquire); }
 
    void clear()
    {
        ClauseExchange* raw;
        while (queue.pop(raw)) {
            ClauseExchange::fromRawPtr(raw);
        }
        m_size.store(0, std::memory_order_release);
    }
 
    bool empty() const { return m_size.load(std::memory_order_acquire) == 0; }
};