bounded_ws_deque.h

#ifndef PLS_INTERNAL_DATA_STRUCTURES_BOUNDED_WS_DEQUE_H_
#define PLS_INTERNAL_DATA_STRUCTURES_BOUNDED_WS_DEQUE_H_

#include <cstdio>
#include <array>
#include <atomic>

#include "pls/internal/base/system_details.h"
#include "pls/internal/data_structures/stamped_integer.h"
#include "pls/internal/data_structures/optional.h"

namespace pls {
namespace internal {
namespace data_structures {

/**
 * Classic, text book ws bounded deque based on arrays.
 * Stores a fixed amount of fixed size objects in an array,
 * allowing or local push/pop on the bottom and remote
 * pop on the top.
 *
 * The local operations are cheap as long as head and tail are
 * far enough apart, making it ideal to avoid cache problems.
 *
 * Depends on overaligned datatypes to be cache line friendly.
 * This does not concern C++14 and upwards, but hinders you to properly
 * allocate it on the heap in C++11 (see base::alignment::alignment_wrapper for a solution).
 */
// TODO: Relax memory orderings in here...
template<typename T>
class bounded_ws_deque {
 public:
  bounded_ws_deque(T *item_array, size_t size) : size_{size}, item_array_{item_array} {}

  void push_bottom(T item) {
    item_array_[local_bottom_] = item;
    local_bottom_++;
    bottom_.store(local_bottom_, std::memory_order_release);
  }

  bool is_empty() {
    return top_.load().value_ < bottom_.load();
  }

  optional<T> pop_top() {
    stamped_integer old_top = top_.load();
    unsigned int new_stamp = old_top.stamp_ + 1;
    unsigned int new_value = old_top.value_ + 1;

    if (bottom_.load() <= old_top.value_) {
      return optional<T>();
    }

    optional<T> result(item_array_[old_top.value_]);
    if (top_.compare_exchange_strong(old_top, {new_stamp, new_value})) {
      return result;
    }

    return optional<T>();
  }

  optional<T> pop_bottom() {
    if (local_bottom_ == 0) {
      return optional<T>();
    }

    local_bottom_--;
    bottom_.store(local_bottom_, std::memory_order_seq_cst);

    optional<T> result(item_array_[local_bottom_]);

    stamped_integer old_top = top_.load(std::memory_order_acquire);
    if (local_bottom_ > old_top.value_) {
      // Enough distance to just return the value
      return result;
    }
    if (local_bottom_ == old_top.value_) {
      local_bottom_ = 0;
      bottom_.store(local_bottom_);
      if (top_.compare_exchange_strong(old_top, {old_top.stamp_ + 1, 0})) {
        // We won the competition and the queue is empty
        return result;
      }
    }

    // The queue is empty and we lost the competition
    local_bottom_ = 0;
    bottom_.store(local_bottom_);
    top_.store({old_top.stamp_ + 1, 0});
    return optional<T>();
  }

 private:
  alignas(base::system_details::CACHE_LINE_SIZE) std::atomic<stamped_integer> top_{stamped_integer{0, 0}};
  alignas(base::system_details::CACHE_LINE_SIZE) std::atomic<unsigned int> bottom_{0};
  unsigned int local_bottom_{0};
  size_t size_;
  T *item_array_;
};

template<typename T, size_t SIZE>
class static_bounded_ws_deque {
 public:
  static_bounded_ws_deque() : items_{}, deque_{items_.data(), SIZE} {}

  bounded_ws_deque<T> &get_deque() { return deque_; }
 private:
  std::array<T, SIZE> items_;
  bounded_ws_deque<T> deque_;
};

}
}
}

#endif //PLS_INTERNAL_DATA_STRUCTURES_BOUNDED_WS_DEQUE_H_