concurrent.hpp

Go to the documentation of this file.

#pragma once
 
#include <atomic>
#include <atomic_queue/include/atomic_queue/atomic_queue.h>
#include <cstdint>
#include <cstdlib>
#include <cstring>
#include <deque>
#include <mutex>
#include <oneapi/tbb/concurrent_map.h>
#include <phmap/parallel_hashmap/phmap.h>
#include <stddef.h>
 
namespace jaffarCommon
{
 
namespace concurrent
{
 
template <class T>
class DrainBuffer
{
public:
  DrainBuffer() = default;
  ~DrainBuffer()
  {
    if (_buffer != nullptr) free(_buffer);
  }
 
  __JAFFAR_COMMON_INLINE__ void reserve(const size_t capacity)
  {
    _capacity = capacity;
    _buffer   = (T*)malloc(capacity * sizeof(T));
  }
 
  __JAFFAR_COMMON_INLINE__ void clear()
  {
    _count = 0;
    _claim.store(0, std::memory_order_relaxed);
  }
 
  __JAFFAR_COMMON_INLINE__ void push_back_no_lock(T element) { _buffer[_count++] = element; }
 
  __JAFFAR_COMMON_INLINE__ size_t pop_front_get_batch(T* elements, const size_t maxCount)
  {
    uint64_t observed = _claim.load(std::memory_order_acquire);
    uint64_t desired;
    uint32_t front;
    size_t   take;
    do {
      front               = (uint32_t)(observed & 0xFFFFFFFFULL);
      const uint32_t back = (uint32_t)(observed >> 32);
      if ((size_t)front + (size_t)back >= _count) return 0; // empty
      const size_t available = _count - (size_t)front - (size_t)back;
      take                   = maxCount < available ? maxCount : available;
      desired                = (observed & 0xFFFFFFFF00000000ULL) | (uint64_t)(front + (uint32_t)take);
    } while (_claim.compare_exchange_weak(observed, desired, std::memory_order_acq_rel, std::memory_order_acquire) == false);
 
    memcpy(elements, &_buffer[front], take * sizeof(T));
    return take;
  }
 
  __JAFFAR_COMMON_INLINE__ bool pop_front_get(T& element) { return pop_front_get_batch(&element, 1) == 1; }
 
  __JAFFAR_COMMON_INLINE__ bool pop_back_get(T& element)
  {
    uint64_t observed = _claim.load(std::memory_order_acquire);
    uint64_t desired;
    uint32_t back;
    do {
      const uint32_t front = (uint32_t)(observed & 0xFFFFFFFFULL);
      back                 = (uint32_t)(observed >> 32);
      if ((size_t)front + (size_t)back >= _count) return false; // empty
      desired = (observed & 0x00000000FFFFFFFFULL) | ((uint64_t)(back + 1) << 32);
    } while (_claim.compare_exchange_weak(observed, desired, std::memory_order_acq_rel, std::memory_order_acquire) == false);
 
    element = _buffer[_count - 1 - (size_t)back];
    return true;
  }
 
  __JAFFAR_COMMON_INLINE__ size_t wasSize() const
  {
    const uint64_t observed = _claim.load(std::memory_order_relaxed);
    const size_t   claimed  = (size_t)(observed & 0xFFFFFFFFULL) + (size_t)(observed >> 32);
    return claimed >= _count ? 0 : _count - claimed;
  }
 
private:
  T* _buffer = nullptr;
 
  size_t _capacity = 0;
 
  size_t _count = 0;
 
  std::atomic<uint64_t> _claim{0};
};
 
template <class T>
using atomicQueue_t = atomic_queue::AtomicQueueB<T>;
 
template <class V>
using HashSet_t = phmap::parallel_flat_hash_set<V, phmap::priv::hash_default_hash<V>, phmap::priv::hash_default_eq<V>, std::allocator<V>, 8, std::mutex>;
 
template <class K, class V>
using HashMap_t = phmap::parallel_flat_hash_map<K, V, phmap::priv::hash_default_hash<K>, phmap::priv::hash_default_eq<K>, std::allocator<std::pair<const K, V>>, 8, std::mutex>;
 
template <class K, class V, class C = std::greater<K>>
using concurrentMultimap_t = oneapi::tbb::concurrent_multimap<K, V, C>;
 
template <class T>
class Deque
{
public:
  Deque()  = default;
  ~Deque() = default;
 
  __JAFFAR_COMMON_INLINE__ auto& getInternalStorage() { return _internalDeque; }
 
  __JAFFAR_COMMON_INLINE__ void push_back_no_lock(T element)
  {
    _internalDeque.push_back(element);
    _size.fetch_add(1, std::memory_order_relaxed);
  }
 
  __JAFFAR_COMMON_INLINE__ void push_back(T element)
  {
    _mutex.lock();
    _internalDeque.push_back(element);
    _size.fetch_add(1, std::memory_order_relaxed);
    _mutex.unlock();
  }
 
  __JAFFAR_COMMON_INLINE__ void push_front_no_lock(T element)
  {
    _internalDeque.push_front(element);
    _size.fetch_add(1, std::memory_order_relaxed);
  }
 
  __JAFFAR_COMMON_INLINE__ void push_front(T element)
  {
    _mutex.lock();
    _internalDeque.push_front(element);
    _size.fetch_add(1, std::memory_order_relaxed);
    _mutex.unlock();
  }
 
  __JAFFAR_COMMON_INLINE__ T front() const { return _internalDeque.front(); }
 
  __JAFFAR_COMMON_INLINE__ T back() const { return _internalDeque.back(); }
 
  __JAFFAR_COMMON_INLINE__ void pop_front()
  {
    _mutex.lock();
    _internalDeque.pop_front();
    _size.fetch_sub(1, std::memory_order_relaxed);
    _mutex.unlock();
  }
 
  __JAFFAR_COMMON_INLINE__ void pop_back()
  {
    _mutex.lock();
    _internalDeque.pop_back();
    _size.fetch_sub(1, std::memory_order_relaxed);
    _mutex.unlock();
  }
 
  __JAFFAR_COMMON_INLINE__ bool pop_back_get(T& element)
  {
    _mutex.lock();
 
    if (_internalDeque.empty())
    {
      _mutex.unlock();
      return false;
    }
 
    element = _internalDeque.back();
    _internalDeque.pop_back();
    _size.fetch_sub(1, std::memory_order_relaxed);
 
    _mutex.unlock();
    return true;
  }
 
  __JAFFAR_COMMON_INLINE__ bool pop_front_get(T& element)
  {
    _mutex.lock();
 
    if (_internalDeque.empty())
    {
      _mutex.unlock();
      return false;
    }
 
    element = _internalDeque.front();
    _internalDeque.pop_front();
    _size.fetch_sub(1, std::memory_order_relaxed);
 
    _mutex.unlock();
    return true;
  }
 
  __JAFFAR_COMMON_INLINE__ size_t pop_front_get_batch(T* elements, const size_t maxCount)
  {
    _mutex.lock();
 
    size_t count = 0;
    while (count < maxCount && _internalDeque.empty() == false)
    {
      elements[count++] = _internalDeque.front();
      _internalDeque.pop_front();
    }
    if (count > 0) _size.fetch_sub(count, std::memory_order_relaxed);
 
    _mutex.unlock();
    return count;
  }
 
  __JAFFAR_COMMON_INLINE__ size_t wasSize() const { return _size.load(std::memory_order_relaxed); }
 
private:
  std::mutex _mutex;
 
  std::atomic<size_t> _size{0};
 
  std::deque<T> _internalDeque;
};
 
} // namespace concurrent
 
} // namespace jaffarCommon