future_queue.hpp

Go to the documentation of this file.

// SPDX-FileCopyrightText: Deutsches Elektronen-Synchrotron DESY, MSK, ChimeraTK Project <chimeratk-support@desy.de>
// SPDX-License-Identifier: LGPL-3.0-or-later
#pragma once
 
#include "semaphore.hpp"
 
#include <atomic>
#include <cassert>
#include <future> // just for std::launch
#include <vector>
 
namespace cppext {
 
  /*********************************************************************************************************************/
 
  class MOVE_DATA {};
 
  class SWAP_DATA {};
 
  namespace detail {
    class TerminateInternalThread {};
  } // namespace detail
 
  /*********************************************************************************************************************/
 
  namespace detail {
    struct shared_state_base;
 
    template<typename T>
    struct shared_state;
 
    template<typename T, typename FEATURES, typename TOUT, typename CALLABLE>
    struct continuation_process_async;
 
    struct shared_state_ptr {
      shared_state_ptr();
 
      shared_state_ptr(const shared_state_ptr& other);
 
      shared_state_ptr& operator=(const shared_state_ptr& other);
 
      ~shared_state_ptr();
 
      template<typename T>
      void make_new(size_t length);
 
      shared_state_base* operator->();
      const shared_state_base* operator->() const;
 
      template<typename T>
      shared_state<T>* cast();
 
      operator bool() const;
 
      bool operator==(const shared_state_ptr& other) const;
 
      shared_state_base* get() const;
 
      void set(shared_state_base* ptr_);
 
     private:
      void free();
 
      shared_state_base* ptr;
    };
 
    template<typename T>
    shared_state_ptr make_shared_state(size_t length) {
      shared_state_ptr p;
      p.make_new<T>(length);
      return p;
    }
 
  } // namespace detail
 
  template<typename T, typename FEATURES>
  class future_queue;
 
  /*********************************************************************************************************************/
 
  class future_queue_base {
   public:
    size_t write_available() const;
 
    size_t read_available() const;
 
    bool push_exception(std::exception_ptr exception);
 
    bool push_overwrite_exception(std::exception_ptr exception);
 
    bool empty();
 
    void wait();
 
    size_t size() const;
 
    bool operator==(const future_queue_base& other) const;
    bool operator!=(const future_queue_base& other) const;
 
   protected:
    future_queue_base(const detail::shared_state_ptr& d_ptr_);
 
    future_queue_base();
 
    bool obtain_write_slot(size_t& index);
 
    void update_read_index_max();
 
    size_t setNotificationQueue(future_queue<size_t, MOVE_DATA>& notificationQueue, size_t indexToSend);
 
    cppext::detail::shared_state_base* get_notification_queue();
 
    void send_notification(cppext::detail::shared_state_base* notification_queue);
 
    void decrement_previous_data_counter();
 
    detail::shared_state_ptr d;
 
    template<typename T, typename FEATURES>
    friend class ::cppext::future_queue;
 
    template<typename ITERATOR_TYPE>
    friend future_queue<size_t, MOVE_DATA> when_any(ITERATOR_TYPE begin, ITERATOR_TYPE end);
 
    template<typename ITERATOR_TYPE>
    friend future_queue<void, MOVE_DATA> when_all(ITERATOR_TYPE begin, ITERATOR_TYPE end);
 
    friend struct detail::shared_state_base;
    friend struct detail::shared_state_ptr;
 
    template<typename T, typename FEATURES, typename TOUT, typename CALLABLE>
    friend struct detail::continuation_process_async;
  };
 
  /*********************************************************************************************************************/
 
  template<typename T, typename FEATURES = MOVE_DATA>
  class future_queue : public future_queue_base {
   public:
    future_queue(size_t length);
 
    future_queue();
 
    future_queue(const future_queue& other) = default;
 
    future_queue& operator=(const future_queue& other) = default;
 
    template<typename U = T,
        typename std::enable_if<std::is_same<T, U>::value && !std::is_same<U, void>::value, int>::type = 0>
    bool push(U&& t);
    template<typename U = T,
        typename std::enable_if<!std::is_same<U, void>::value && std::is_copy_constructible<T>::value, int>::type = 0>
    bool push(const U& t);
 
    bool push(void);
 
    template<typename U = T,
        typename std::enable_if<std::is_same<T, U>::value && !std::is_same<U, void>::value, int>::type = 0>
    bool push_overwrite(U&& t);
 
    template<typename U = T,
        typename std::enable_if<!std::is_same<U, void>::value && std::is_copy_constructible<T>::value, int>::type = 0>
    bool push_overwrite(const U& t);
 
    bool push_overwrite();
 
    template<typename U = T,
        typename std::enable_if<std::is_same<T, U>::value && !std::is_same<U, void>::value, int>::type = 0>
    bool pop(U& t);
 
    bool pop();
 
    template<typename U = T,
        typename std::enable_if<std::is_same<T, U>::value && !std::is_same<U, void>::value, int>::type = 0>
    void pop_wait(U& t);
 
    void pop_wait();
 
    template<typename U = T,
        typename std::enable_if<std::is_same<T, U>::value && !std::is_same<U, void>::value, int>::type = 0>
    U& front();
 
    template<typename U = T,
        typename std::enable_if<std::is_same<T, U>::value && std::is_same<U, void>::value, int>::type = 0>
    void front() const;
 
    template<typename T2, typename FEATURES2 = MOVE_DATA, typename CALLABLE>
    future_queue<T2, FEATURES2> then(CALLABLE callable, std::launch policy = std::launch::async);
 
    typedef T value_type;
  };
 
  /*********************************************************************************************************************/
 
  namespace detail {
 
    struct when_any_notification_info {
      cppext::detail::shared_state_base* notifyerQueue{nullptr};
 
      size_t notifyerQueue_previousData{0};
    };
 
    struct shared_state_base {
      shared_state_base(size_t length)
      : nBuffers(length + 1), semaphores(length + 1), exceptions(length + 1), writeIndex(0), readIndexMax(0),
        readIndex(0), hasFrontOwnership(false) {}
 
      virtual ~shared_state_base() {}
 
      inline void free();
 
      std::atomic<size_t> reference_count{0};
 
      std::atomic<size_t> when_any_index;
 
      size_t nBuffers;
 
      std::vector<semaphore> semaphores;
 
      std::vector<std::exception_ptr> exceptions;
 
      std::atomic<size_t> writeIndex;
 
      std::atomic<size_t> readIndexMax;
 
      std::atomic<size_t> readIndex;
 
      bool hasFrontOwnership;
 
      bool is_continuation_deferred{false};
 
      std::function<void(void)> continuation_process_deferred;
 
      std::function<void(void)> continuation_process_deferred_wait;
 
      bool is_continuation_async{false};
 
      std::thread continuation_process_async;
 
      std::atomic<bool> continuation_process_async_terminated{false};
 
      bool is_continuation_when_all{false};
 
      future_queue_base continuation_origin;
 
      std::atomic<when_any_notification_info> when_any_notification{when_any_notification_info()};
    };
 
    template<typename T>
    struct shared_state : shared_state_base {
      shared_state(size_t length) : shared_state_base(length), buffers(length + 1) {}
 
      std::vector<T> buffers;
    };
 
    template<>
    struct shared_state<void> : shared_state_base {
      shared_state(size_t length) : shared_state_base(length) {}
    };
 
    inline void shared_state_base::free() {
      // Reduce reference count but atomically keep the old reference counter. Note
      // that the std::memory_order_relaxed refers to the access to the pointer not
      // to the reference counter.
      size_t oldCount = this->reference_count--;
 
      // Determine whether we need to destroy the shared state depending on possible
      // internal references.
      bool executeDelete = false;
 
      // Standard case: no continuation. If the last user is just destroying its
      // reference we delete the shared state.
      if(oldCount == 1 && !this->is_continuation_async && !this->is_continuation_deferred &&
          !this->is_continuation_when_all) {
        executeDelete = true;
      }
      // Deferred continuations (incl. when_all) have two internal use counts due to
      // the two std::functions, so we need to remove those functions first.
      else if(oldCount == 3 && (this->is_continuation_deferred || this->is_continuation_when_all)) {
        this->continuation_process_deferred = {};
        this->continuation_process_deferred_wait = {};
        executeDelete = true;
      }
      // Async continuations have one internal use count inside their thread, so we
      // need to terminate the thread first.
      else if(oldCount == 2 && this->is_continuation_async) {
        if(this->continuation_process_async.joinable()) {
          // Signal termination to internal thread and wait until thread has been
          // terminated
          while(this->continuation_process_async_terminated == false) {
            // Push a detail::TerminateInternalThread exception into the queue which
            // the internal thread is potentially waiting on.
            try {
              throw detail::TerminateInternalThread();
            }
            catch(...) {
              // Special case: the origin queue is a continuation itself (deferred
              // or when_all) - we need to push the exception to the origin of the
              // origin to actually reach the internal thread, since
              // deferred/when_all continuations do not really use their own queue
              if(this->continuation_origin.d->is_continuation_deferred ||
                  this->continuation_origin.d->is_continuation_when_all) {
                this->continuation_origin.d->continuation_origin.push_exception(std::current_exception());
              }
              // Standard case: just push the exception to the origin queue of the
              // continuation
              else {
                this->continuation_origin.push_exception(std::current_exception());
              }
            } // end catch
          } // end while
 
          this->continuation_process_async.join();
        }
        executeDelete = true;
      }
 
      if(executeDelete) {
        // Now that all potential internal references have been cleared the
        // reference count must be 0
        assert(reference_count == 0);
 
        // the when_any_notification notifyerQueue may have it's reference count
        // manually incremented by setNotificationQueue, and must be freed if it exists.
        if(when_any_notification.load().notifyerQueue) {
          when_any_notification.load().notifyerQueue->free();
        }
        delete this;
      } // end if executeDelete
 
    } // end shared_state_base::free()
 
  } // namespace detail
 
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
 
  template<typename ITERATOR_TYPE>
  future_queue<size_t> when_any(ITERATOR_TYPE begin, ITERATOR_TYPE end) {
    // Add lengthes of all queues - this will be the length of the notification
    // queue
    size_t summedLength = 0;
    for(ITERATOR_TYPE it = begin; it != end; ++it) summedLength += it->size();
 
    // Create a notification queue, so we can hand it on to the queues
    future_queue<size_t> notifyerQueue(summedLength);
 
    // Distribute the pointer to the notification queue to all participating
    // queues
    size_t index = 0;
    for(ITERATOR_TYPE it = begin; it != end; ++it) {
      size_t nPreviousValues = it->setNotificationQueue(notifyerQueue, index);
      for(size_t i = 0; i < nPreviousValues; ++i) notifyerQueue.push(index);
      ++index;
    }
 
    return notifyerQueue;
  }
 
  /*********************************************************************************************************************/
 
  template<typename ITERATOR_TYPE>
  future_queue<void> when_all(ITERATOR_TYPE begin, ITERATOR_TYPE end) {
    // Create a notification queue in a shared pointer, so we can hand it on to
    // the queues
    future_queue<void> notifyerQueue(1);
 
    // copy the list of participating queues
    std::vector<future_queue_base> participants;
    for(auto it = begin; it != end; ++it) participants.push_back(*it);
 
    // obtain notification queue for any update to any queue
    auto anyNotify = when_any(begin, end);
 
    // define function to be executed (inside the notifyer queue) on non-blocking
    // functions like pop() or empty()
    notifyerQueue.d->continuation_process_deferred = std::function<void(void)>([notifyerQueue, participants]() mutable {
      bool empty = false;
      for(auto& q : participants) {
        if(q.empty()) {
          empty = true;
          break;
        }
      }
      if(!empty) notifyerQueue.push();
    });
 
    // define function to be executed (inside the notifyer queue) on blocking
    // functions like pop_wait() or wait()
    notifyerQueue.d->continuation_process_deferred_wait =
        std::function<void(void)>([notifyerQueue, participants, anyNotify]() mutable {
          while(true) {
            anyNotify.pop_wait();
            bool empty = false;
            for(auto& q : participants) {
              if(q.empty()) {
                empty = true;
                break;
              }
            }
            if(!empty) break;
          }
          notifyerQueue.push();
        });
 
    // set flag marking the notifyerQueue a when_all continuation and save the
    // notification queue of the when_any as the origin.
    notifyerQueue.d->is_continuation_when_all = true;
    notifyerQueue.d->continuation_origin = anyNotify;
 
    return notifyerQueue;
  }
 
  /*********************************************************************************************************************/
 
  namespace detail {
 
    template<typename T>
    void data_assign(T& a, T&& b, MOVE_DATA) {
      // in order not to depend on the move assignment operator, which might not
      // always be available, we perform an in-place destruction followed by an
      // in-place move construction.
      a.~T();
      new(&a) T(std::move(b));
    }
 
    template<typename T>
    void data_assign(T& a, T&& b, SWAP_DATA) {
      std::swap(a, b);
    }
 
  } // namespace detail
 
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
 
  namespace detail {
 
    inline shared_state_ptr::shared_state_ptr() : ptr(nullptr) {}
 
    inline shared_state_ptr::shared_state_ptr(const shared_state_ptr& other) {
      // Copy the pointer and increase the reference count
      set(other.get());
      if(get() != nullptr) get()->reference_count++;
    }
 
    inline shared_state_ptr& shared_state_ptr::operator=(const shared_state_ptr& other) {
      // Free previous target, copy the new pointer and increase its reference count
      free();
      set(other.get());
      if(get() != nullptr) get()->reference_count++;
      return *this;
    }
 
    inline shared_state_ptr::~shared_state_ptr() {
      free();
    }
 
    inline shared_state_base* shared_state_ptr::get() const {
      return ptr;
    }
 
    inline void shared_state_ptr::set(shared_state_base* ptr_) {
      ptr = ptr_;
    }
 
    inline void shared_state_ptr::free() {
      // Don't do anything if called on a nullptr (i.e. default constructed or
      // already destroyed)
      if(get() == nullptr) {
        return;
      }
 
      get()->free();
      set(nullptr);
    }
 
    template<typename T>
    void shared_state_ptr::make_new(size_t length) {
      free();
      ptr = new shared_state<T>(length);
      get()->reference_count = 1;
    }
 
    inline shared_state_base* shared_state_ptr::operator->() {
      assert(get() != nullptr);
      return get();
    }
 
    inline const shared_state_base* shared_state_ptr::operator->() const {
      assert(get() != nullptr);
      return get();
    }
 
    template<typename T>
    shared_state<T>* shared_state_ptr::cast() {
      assert(get() != nullptr);
      return static_cast<shared_state<T>*>(get());
    }
 
    inline shared_state_ptr::operator bool() const {
      return get() != nullptr;
    }
 
    inline bool shared_state_ptr::operator==(const shared_state_ptr& other) const {
      return get() == other.get();
    }
 
  } // namespace detail
 
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
 
  inline size_t future_queue_base::write_available() const {
    // Obtain indices in this particular order to ensure consistency. Result might
    // be too small (but not too big) if writing happens concurrently.
    size_t l_writeIndex = d->writeIndex;
    size_t l_readIndex = d->readIndex;
    if(l_writeIndex - l_readIndex < d->nBuffers - 1) {
      return d->nBuffers - (l_writeIndex - l_readIndex) - 1;
    }
    else {
      return 0;
    }
  }
 
  inline size_t future_queue_base::read_available() const {
    // Single consumer, so atomicity doesn't matter
    return d->readIndexMax - d->readIndex;
  }
 
  inline cppext::detail::shared_state_base* future_queue_base::get_notification_queue() {
    // if there is no notification queue, atomically increment counter while making sure now notification queue is
    // placed concurrently
    detail::when_any_notification_info info, info_n;
    do {
      info = d->when_any_notification.load(std::memory_order_acquire);
      if(info.notifyerQueue) break;
      info_n = info;
      ++info_n.notifyerQueue_previousData;
    } while(!d->when_any_notification.compare_exchange_weak(info, info_n));
    return info.notifyerQueue;
  }
 
  inline void future_queue_base::send_notification(cppext::detail::shared_state_base* notification_queue) {
    // if there is a notification queue, push to it
    if(notification_queue) {
      future_queue<size_t, MOVE_DATA> n;
      n.d.set(notification_queue);
      bool nret = n.push(d->when_any_index);
      n.d.set(nullptr); // prevent reference count from being decremented
      (void)nret;
      // This assert doesn't really hold. It might spuriously fail during destruction of certain combinations of
      // continuations and when_any/when_all.
      // assert(nret == true);
    }
  }
 
  inline void future_queue_base::decrement_previous_data_counter() {
    detail::when_any_notification_info info, info_n;
    do {
      info = d->when_any_notification.load(std::memory_order_acquire);
      if(info.notifyerQueue) break; // no need to deal with this counter if notification queue present
      info_n = info;
      assert(info_n.notifyerQueue_previousData > 0);
      --info_n.notifyerQueue_previousData;
    } while(!d->when_any_notification.compare_exchange_weak(info, info_n));
  }
 
  inline bool future_queue_base::push_exception(std::exception_ptr exception) {
    // obtain index to write to
    size_t myIndex;
    if(!obtain_write_slot(myIndex)) return false;
 
    // assign the payload data (data buffer is ignored if exception is set)
    d->exceptions[myIndex % d->nBuffers] = exception;
 
    // obtain notification queue or increment previous data counter (for when_any)
    auto notification_queue = get_notification_queue();
 
    // signal receiving end
    assert(!d->semaphores[myIndex % d->nBuffers].is_ready());
    d->semaphores[myIndex % d->nBuffers].unlock();
    update_read_index_max();
 
    // deal with when_any notifications
    send_notification(notification_queue);
    return true;
  }
 
  inline bool future_queue_base::push_overwrite_exception(std::exception_ptr exception) {
    assert(d->nBuffers - 1 > 1);
    bool ret = true;
 
    // obtain index to write to, if necessary remove old data first
    size_t myIndex;
    if(!obtain_write_slot(myIndex)) {
      if(d->semaphores[(myIndex - 1) % d->nBuffers].is_ready_and_reset()) {
        size_t expectedIndex = myIndex;
        bool success = d->writeIndex.compare_exchange_strong(expectedIndex, myIndex - 1);
        if(!success) {
          // in case of a concurrent push_overwrite(), our data effectively just got overwritten by the other thread
          // even before writing it...
          d->semaphores[(myIndex - 1) % d->nBuffers].unlock();
          return false;
        }
        ret = false;
      }
      else {
        return false;
      }
      if(!obtain_write_slot(myIndex)) return false;
    }
 
    // assign the payload data (data buffer is ignored if exception is set)
    d->exceptions[myIndex % d->nBuffers] = exception;
 
    // obtain notification queue or increment previous data counter (for when_any) (unless data was overwritten)
    cppext::detail::shared_state_base* notification_queue;
    if(ret) {
      notification_queue = get_notification_queue();
    }
 
    // obtain notification queue or increment previous data counter (for when_any)
    assert(!d->semaphores[myIndex % d->nBuffers].is_ready());
    d->semaphores[myIndex % d->nBuffers].unlock();
    update_read_index_max();
 
    // deal with when_any notifications (unless data was overwritten)
    if(ret) {
      send_notification(notification_queue);
    }
    return ret;
  }
 
  inline bool future_queue_base::empty() {
    if(d->hasFrontOwnership) return false;
    if(d->is_continuation_deferred || d->is_continuation_when_all) d->continuation_process_deferred();
    if(d->semaphores[d->readIndex % d->nBuffers].is_ready_and_reset()) {
      d->hasFrontOwnership = true;
      return false;
    }
    return true;
  }
 
  inline void future_queue_base::wait() {
    if(d->hasFrontOwnership) return;
    if(d->is_continuation_deferred || d->is_continuation_when_all) d->continuation_process_deferred_wait();
    d->semaphores[d->readIndex % d->nBuffers].wait_and_reset();
    d->hasFrontOwnership = true;
  }
 
  inline size_t future_queue_base::size() const {
    if(!d->is_continuation_deferred) {
      return d->nBuffers - 1;
    }
    else {
      return d->continuation_origin.size();
    }
  }
 
  inline bool future_queue_base::operator==(const future_queue_base& other) const {
    return d == other.d;
  }
 
  inline bool future_queue_base::operator!=(const future_queue_base& other) const {
    return !(d == other.d);
  }
 
  inline future_queue_base::future_queue_base(const detail::shared_state_ptr& d_ptr_) : d(d_ptr_) {}
 
  inline future_queue_base::future_queue_base() : d() {}
 
  inline bool future_queue_base::obtain_write_slot(size_t& index) {
    index = d->writeIndex;
    while(true) {
      if(index >= d->readIndex + d->nBuffers - 1) return false; // queue is full
      bool success = d->writeIndex.compare_exchange_weak(index, index + 1);
      if(success) break;
    }
    return true;
  }
 
  inline void future_queue_base::update_read_index_max() {
    size_t l_readIndex = d->readIndex;
    size_t l_writeIndex = d->writeIndex;
    size_t l_readIndexMax = d->readIndexMax;
    if(l_writeIndex >= l_readIndex + d->nBuffers) l_writeIndex = l_readIndex + d->nBuffers - 1;
    size_t newReadIndexMax = l_readIndexMax;
    do {
      for(size_t index = l_readIndexMax; index <= l_writeIndex - 1; ++index) {
        if(!d->semaphores[index % d->nBuffers].is_ready()) break;
        newReadIndexMax = index + 1;
      }
      d->readIndexMax.compare_exchange_weak(l_readIndexMax, newReadIndexMax);
    } while(d->readIndexMax < newReadIndexMax);
  }
 
  inline size_t future_queue_base::setNotificationQueue(
      future_queue<size_t, MOVE_DATA>& notificationQueue, size_t indexToSend) {
    if(!d->is_continuation_deferred) {
      d->when_any_index = indexToSend;
 
      // create new info struct with notification queue
      detail::when_any_notification_info info;
      info.notifyerQueue = notificationQueue.d.get();
      info.notifyerQueue_previousData = 0;
 
      // atomically exchange info struct while making sure it has not been altered at the target in the mean time
      detail::when_any_notification_info info_o;
      do {
        info_o = d->when_any_notification;
      } while(!d->when_any_notification.compare_exchange_weak(info_o, info));
 
      // artificially increment the reference count of the notification queue, since we have to store a plain pointer
      // in the info struct rather than a shared_state_ptr (which is not trivially copyable).
      info.notifyerQueue->reference_count++;
 
      return info_o.notifyerQueue_previousData;
    }
    else {
      return d->continuation_origin.setNotificationQueue(notificationQueue, indexToSend);
    }
  }
 
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
  /*********************************************************************************************************************/
 
  template<typename T, typename FEATURES>
  future_queue<T, FEATURES>::future_queue(size_t length) : future_queue_base(detail::make_shared_state<T>(length)) {}
 
  template<typename T, typename FEATURES>
  future_queue<T, FEATURES>::future_queue() {}
 
  /*********************************************************************************************************************/
  template<typename T, typename FEATURES>
  template<typename U, typename std::enable_if<std::is_same<T, U>::value && !std::is_same<U, void>::value, int>::type>
  bool future_queue<T, FEATURES>::push(U&& t) {
    // obtain index to write to
    size_t myIndex;
    if(!obtain_write_slot(myIndex)) return false;
 
    // assign the payload data
    detail::data_assign(future_queue_base::d.cast<T>()->buffers[myIndex % d->nBuffers], std::move(t), FEATURES());
    d->exceptions[myIndex % d->nBuffers] = nullptr;
 
    // obtain notification queue or increment previous data counter (for when_any)
    auto notification_queue = get_notification_queue();
 
    // signal receiving end
    assert(!d->semaphores[myIndex % d->nBuffers].is_ready());
    d->semaphores[myIndex % d->nBuffers].unlock();
    update_read_index_max(); // basically only for read_available()
 
    // deal with when_any notifications
    send_notification(notification_queue);
    return true;
  }
 
  template<typename T, typename FEATURES>
  template<typename U,
      typename std::enable_if<!std::is_same<U, void>::value && std::is_copy_constructible<T>::value, int>::type>
  bool future_queue<T, FEATURES>::push(const U& t) {
    // Create copy and pass this copy as an Rvalue reference to the other
    // implementation
    return push(T(t));
  }
 
  template<typename T, typename FEATURES>
  bool future_queue<T, FEATURES>::push(void) {
    static_assert(
        std::is_same<T, void>::value, "future_queue<T,FEATURES>::push(void) may only be called for T = void.");
    // obtain index to write to
    size_t myIndex;
    if(!obtain_write_slot(myIndex)) return false;
 
    // assign the payload data
    d->exceptions[myIndex % d->nBuffers] = nullptr;
 
    // obtain notification queue or increment previous data counter (for when_any)
    auto notification_queue = get_notification_queue();
 
    // signal receiving end
    assert(!d->semaphores[myIndex % d->nBuffers].is_ready());
    d->semaphores[myIndex % d->nBuffers].unlock();
    update_read_index_max(); // basically only for read_available()
 
    // deal with when_any notifications
    send_notification(notification_queue);
    return true;
  }
 
  template<typename T, typename FEATURES>
  template<typename U, typename std::enable_if<std::is_same<T, U>::value && !std::is_same<U, void>::value, int>::type>
  bool future_queue<T, FEATURES>::push_overwrite(U&& t) {
    assert(d->nBuffers - 1 > 1);
    bool ret = true;
 
    // obtain index to write to, if necessary remove old data first
    size_t myIndex;
    if(!obtain_write_slot(myIndex)) {
      if(d->semaphores[(myIndex - 1) % d->nBuffers].is_ready_and_reset()) {
        size_t expectedIndex = myIndex;
        bool success = d->writeIndex.compare_exchange_strong(expectedIndex, myIndex - 1);
        if(!success) {
          // in case of a concurrent push_overwrite(), our data effectively just got overwritten by the other thread
          // even before writing it...
          d->semaphores[(myIndex - 1) % d->nBuffers].unlock();
          return false;
        }
        ret = false;
      }
      else {
        return false;
      }
      if(!obtain_write_slot(myIndex)) return false;
    }
 
    // assign the payload data
    detail::data_assign(future_queue_base::d.cast<T>()->buffers[myIndex % d->nBuffers], std::move(t), FEATURES());
    d->exceptions[myIndex % d->nBuffers] = nullptr;
 
    // obtain notification queue or increment previous data counter (for when_any) (unless data was overwritten)
    [[maybe_unused]] cppext::detail::shared_state_base* notification_queue;
    if(ret) {
      notification_queue = get_notification_queue();
    }
 
    // signal receiving end
    assert(!d->semaphores[myIndex % d->nBuffers].is_ready());
    d->semaphores[myIndex % d->nBuffers].unlock();
    update_read_index_max();
 
    // deal with when_any notifications (unless data was overwritten)
    if(ret) {
      send_notification(notification_queue);
    }
    return ret;
  }
 
  template<typename T, typename FEATURES>
  template<typename U,
      typename std::enable_if<!std::is_same<U, void>::value && std::is_copy_constructible<T>::value, int>::type>
  bool future_queue<T, FEATURES>::push_overwrite(const U& t) {
    // Create copy and pass this copy as an Rvalue reference to the other
    // implementation
    return push_overwrite(T(t));
  }
 
  /*********************************************************************************************************************/
  template<typename T, typename FEATURES>
  template<typename U, typename std::enable_if<std::is_same<T, U>::value && !std::is_same<U, void>::value, int>::type>
  bool future_queue<T, FEATURES>::pop(U& t) {
    if((d->is_continuation_deferred || d->is_continuation_when_all) && !d->hasFrontOwnership) {
      d->continuation_process_deferred();
    }
    if(d->hasFrontOwnership || d->semaphores[d->readIndex % d->nBuffers].is_ready_and_reset()) {
      std::exception_ptr e;
      if(d->exceptions[d->readIndex % d->nBuffers]) {
        e = d->exceptions[d->readIndex % d->nBuffers];
      }
      else {
        detail::data_assign(
            t, std::move(future_queue_base::d.cast<T>()->buffers[d->readIndex % d->nBuffers]), FEATURES());
      }
      assert(d->readIndex < d->writeIndex);
      d->readIndex++;
      d->hasFrontOwnership = false;
      decrement_previous_data_counter();
      if(e) std::rethrow_exception(e);
      return true;
    }
    else {
      return false;
    }
  }
 
  template<typename T, typename FEATURES>
  bool future_queue<T, FEATURES>::pop() {
    if((d->is_continuation_deferred || d->is_continuation_when_all) && !d->hasFrontOwnership) {
      d->continuation_process_deferred();
    }
    if(d->hasFrontOwnership || d->semaphores[d->readIndex % d->nBuffers].is_ready_and_reset()) {
      std::exception_ptr e;
      if(d->exceptions[d->readIndex % d->nBuffers]) {
        e = d->exceptions[d->readIndex % d->nBuffers];
      }
      assert(d->readIndex < d->writeIndex);
      d->readIndex++;
      d->hasFrontOwnership = false;
      decrement_previous_data_counter();
      if(e) std::rethrow_exception(e);
      return true;
    }
    else {
      return false;
    }
  }
 
  template<typename T, typename FEATURES>
  template<typename U, typename std::enable_if<std::is_same<T, U>::value && !std::is_same<U, void>::value, int>::type>
  void future_queue<T, FEATURES>::pop_wait(U& t) {
    if(!d->hasFrontOwnership) {
      if(d->is_continuation_deferred || d->is_continuation_when_all) d->continuation_process_deferred_wait();
      d->semaphores[d->readIndex % d->nBuffers].wait_and_reset();
    }
    else {
      d->hasFrontOwnership = false;
    }
    std::exception_ptr e;
    if(d->exceptions[d->readIndex % d->nBuffers]) {
      e = d->exceptions[d->readIndex % d->nBuffers];
    }
    else {
      detail::data_assign(
          t, std::move(future_queue_base::d.cast<U>()->buffers[d->readIndex % d->nBuffers]), FEATURES());
    }
    assert(d->readIndex < d->writeIndex);
    d->readIndex++;
    decrement_previous_data_counter();
    if(e) std::rethrow_exception(e);
  }
 
  template<typename T, typename FEATURES>
  void future_queue<T, FEATURES>::pop_wait() {
    if(!d->hasFrontOwnership) {
      if(d->is_continuation_deferred || d->is_continuation_when_all) d->continuation_process_deferred_wait();
      d->semaphores[d->readIndex % d->nBuffers].wait_and_reset();
    }
    else {
      d->hasFrontOwnership = false;
    }
    std::exception_ptr e;
    if(d->exceptions[d->readIndex % d->nBuffers]) {
      e = d->exceptions[d->readIndex % d->nBuffers];
    }
    assert(d->readIndex < d->writeIndex);
    d->readIndex++;
    decrement_previous_data_counter();
    if(e) std::rethrow_exception(e);
  }
 
  /*********************************************************************************************************************/
  template<typename T, typename FEATURES>
  template<typename U, typename std::enable_if<std::is_same<T, U>::value && !std::is_same<U, void>::value, int>::type>
  U& future_queue<T, FEATURES>::front() {
    assert(d->hasFrontOwnership);
    if(d->exceptions[d->readIndex % d->nBuffers]) std::rethrow_exception(d->exceptions[d->readIndex % d->nBuffers]);
    return future_queue_base::d.cast<T>()->buffers[d->readIndex % d->nBuffers];
  }
 
  template<typename T, typename FEATURES>
  template<typename U, typename std::enable_if<std::is_same<T, U>::value && std::is_same<U, void>::value, int>::type>
  void future_queue<T, FEATURES>::front() const {
    assert(d->hasFrontOwnership);
    if(d->exceptions[d->readIndex % d->nBuffers]) std::rethrow_exception(d->exceptions[d->readIndex % d->nBuffers]);
  }
 
  /*********************************************************************************************************************/
  namespace detail {
    // ----------------------------------------------------------------------------------------------------------------
    // ----------------------------------------------------------------------------------------------------------------
    // helper functions used inside future_queue::then()
 
    // ----------------------------------------------------------------------------------------------------------------
    // continuation_process_deferred: function to be executed in a deferred
    // continuation in non-blocking functions
 
    // continuation_process_deferred for non-void data types
    template<typename T, typename FEATURES, typename TOUT, typename CALLABLE>
    struct continuation_process_deferred {
      continuation_process_deferred(
          future_queue<T, FEATURES> q_input_, future_queue<TOUT> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        // written this way so the callable is able to swap with the internal buffer
        if(q_input.empty()) return;
        try {
          q_output.push(callable(q_input.front()));
        }
        catch(...) {
          q_output.push_exception(std::current_exception());
        }
        try {
          q_input.pop();
        }
        catch(...) {
          // exception already pushed to the output queue, so ignore here
        }
      }
      future_queue<T, FEATURES> q_input;
      future_queue<TOUT> q_output;
      CALLABLE callable;
    };
 
    // continuation_process_deferred for void input and non-void output data types
    template<typename FEATURES, typename TOUT, typename CALLABLE>
    struct continuation_process_deferred<void, FEATURES, TOUT, CALLABLE> {
      continuation_process_deferred(
          future_queue<void, FEATURES> q_input_, future_queue<TOUT> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        bool got_data = q_input.pop();
        if(got_data) q_output.push(callable());
      }
      future_queue<void, FEATURES> q_input;
      future_queue<TOUT> q_output;
      CALLABLE callable;
    };
 
    // continuation_process_deferred for non-void input and void output data types
    template<typename T, typename FEATURES, typename CALLABLE>
    struct continuation_process_deferred<T, FEATURES, void, CALLABLE> {
      continuation_process_deferred(
          future_queue<T, FEATURES> q_input_, future_queue<void> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        // written this way so the callable is able to swap with the internal buffer
        if(q_input.empty()) return;
        try {
          callable(q_input.front());
          q_output.push();
        }
        catch(...) {
          q_output.push_exception(std::current_exception());
        }
        try {
          q_input.pop();
        }
        catch(...) {
          // exception already pushed to the output queue, so ignore here
        }
      }
      future_queue<T, FEATURES> q_input;
      future_queue<void> q_output;
      CALLABLE callable;
    };
 
    // continuation_process_deferred for void input and void output data types
    template<typename FEATURES, typename CALLABLE>
    struct continuation_process_deferred<void, FEATURES, void, CALLABLE> {
      continuation_process_deferred(
          future_queue<void, FEATURES> q_input_, future_queue<void> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        try {
          bool got_data = q_input.pop();
          if(got_data) {
            callable();
            q_output.push();
          }
        }
        catch(...) {
          q_output.push_exception(std::current_exception());
        }
      }
      future_queue<void, FEATURES> q_input;
      future_queue<void> q_output;
      CALLABLE callable;
    };
 
    // factory for continuation_process_deferred
    template<typename T, typename FEATURES, typename TOUT, typename CALLABLE>
    continuation_process_deferred<T, FEATURES, TOUT, CALLABLE> make_continuation_process_deferred(
        future_queue<T, FEATURES> q_input, future_queue<TOUT> q_output, CALLABLE callable) {
      return {q_input, q_output, callable};
    }
 
    // ----------------------------------------------------------------------------------------------------------------
    // continuation_process_deferred_wait: function to be executed in a deferred
    // continuation in blocking functions
 
    // continuation_process_deferred_wait for non-void data types
    template<typename T, typename FEATURES, typename TOUT, typename CALLABLE>
    struct continuation_process_deferred_wait {
      continuation_process_deferred_wait(
          future_queue<T, FEATURES> q_input_, future_queue<TOUT> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        // written this way so the callable is able to swap with the internal buffer
        try {
          q_input.wait();
          q_output.push(callable(q_input.front()));
        }
        catch(...) {
          q_output.push_exception(std::current_exception());
        }
        try {
          q_input.pop();
        }
        catch(...) {
          // exception already pushed to the output queue, so ignore here
        }
      }
      future_queue<T, FEATURES> q_input;
      future_queue<TOUT> q_output;
      CALLABLE callable;
    };
 
    // continuation_process_deferred_wait for void input and non-void output data
    // types
    template<typename FEATURES, typename TOUT, typename CALLABLE>
    struct continuation_process_deferred_wait<void, FEATURES, TOUT, CALLABLE> {
      continuation_process_deferred_wait(
          future_queue<void, FEATURES> q_input_, future_queue<TOUT> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        try {
          q_input.pop_wait();
          q_output.push(callable());
        }
        catch(...) {
          q_output.push_exception(std::current_exception());
        }
      }
      future_queue<void, FEATURES> q_input;
      future_queue<TOUT> q_output;
      CALLABLE callable;
    };
 
    // continuation_process_deferred_wait for non-void input and void output data
    // types
    template<typename T, typename FEATURES, typename CALLABLE>
    struct continuation_process_deferred_wait<T, FEATURES, void, CALLABLE> {
      continuation_process_deferred_wait(
          future_queue<T, FEATURES> q_input_, future_queue<void> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        // written this way so the callable is able to swap with the internal buffer
        try {
          q_input.wait();
          callable(q_input.front());
          q_output.push();
        }
        catch(...) {
          q_output.push_exception(std::current_exception());
        }
        try {
          q_input.pop();
        }
        catch(...) {
          // exception already pushed to the output queue, so ignore here
        }
      }
      future_queue<T, FEATURES> q_input;
      future_queue<void> q_output;
      CALLABLE callable;
    };
 
    // continuation_process_deferred_wait for void input and void output data types
    template<typename FEATURES, typename CALLABLE>
    struct continuation_process_deferred_wait<void, FEATURES, void, CALLABLE> {
      continuation_process_deferred_wait(
          future_queue<void, FEATURES> q_input_, future_queue<void> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        try {
          q_input.pop_wait();
          callable();
          q_output.push();
        }
        catch(...) {
          q_output.push_exception(std::current_exception());
        }
      }
      future_queue<void, FEATURES> q_input;
      future_queue<void> q_output;
      CALLABLE callable;
    };
 
    // factory for continuation_process_deferred_wait
    template<typename T, typename FEATURES, typename TOUT, typename CALLABLE>
    continuation_process_deferred_wait<T, FEATURES, TOUT, CALLABLE> make_continuation_process_deferred_wait(
        future_queue<T, FEATURES> q_input, future_queue<TOUT> q_output, CALLABLE callable) {
      return {q_input, q_output, callable};
    }
 
    // ----------------------------------------------------------------------------------------------------------------
    // continuation_process_async: function to be executed in the internal thread of
    // a async continuation
 
    // continuation_process_async for non-void data types
    template<typename T, typename FEATURES, typename TOUT, typename CALLABLE>
    struct continuation_process_async {
      continuation_process_async(future_queue<T, FEATURES> q_input_, future_queue<TOUT> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        while(true) {
          // written this way so the callable is able to swap with the internal
          // buffer
          q_input.wait();
          T* v;
          try {
            v = &(q_input.front());
            // TODO how to handle full output queues?
            q_output.push(callable(*v));
          }
          catch(detail::TerminateInternalThread&) {
            q_output.d->continuation_process_async_terminated = true;
            return;
          }
          catch(...) {
            // TODO how to handle full output queues?
            q_output.push_exception(std::current_exception());
          }
          try {
            q_input.pop();
          }
          catch(...) {
            // exception already pushed to the output queue, so ignore here
          }
        }
      }
      future_queue<T, FEATURES> q_input;
      future_queue<TOUT> q_output;
      CALLABLE callable;
    };
 
    // continuation_process_async for void input and non-void output data types
    template<typename FEATURES, typename TOUT, typename CALLABLE>
    struct continuation_process_async<void, FEATURES, TOUT, CALLABLE> {
      continuation_process_async(
          future_queue<void, FEATURES> q_input_, future_queue<TOUT> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        while(true) {
          try {
            q_input.pop_wait();
            // TODO how to handle full output queues?
            q_output.push(callable());
          }
          catch(detail::TerminateInternalThread&) {
            q_output.d->continuation_process_async_terminated = true;
            return;
          }
          catch(...) {
            // TODO how to handle full output queues?
            q_output.push_exception(std::current_exception());
          }
        }
      }
      future_queue<void, FEATURES> q_input;
      future_queue<TOUT> q_output;
      CALLABLE callable;
    };
 
    // continuation_process_async for non-void input and void output data types
    template<typename T, typename FEATURES, typename CALLABLE>
    struct continuation_process_async<T, FEATURES, void, CALLABLE> {
      continuation_process_async(future_queue<T, FEATURES> q_input_, future_queue<void> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        while(true) {
          // written this way so the callable is able to swap with the internal
          // buffer
          q_input.wait();
          T* v;
          try {
            v = &(q_input.front());
            callable(*v);
            // TODO how to handle full output queues?
            q_output.push();
          }
          catch(detail::TerminateInternalThread&) {
            q_output.d->continuation_process_async_terminated = true;
            return;
          }
          catch(...) {
            // TODO how to handle full output queues?
            q_output.push_exception(std::current_exception());
          }
          try {
            q_input.pop();
          }
          catch(...) {
            // exception already pushed to the output queue, so ignore here
          }
        }
      }
      future_queue<T, FEATURES> q_input;
      future_queue<void> q_output;
      CALLABLE callable;
    };
 
    // continuation_process_async for void input and void output data types
    template<typename FEATURES, typename CALLABLE>
    struct continuation_process_async<void, FEATURES, void, CALLABLE> {
      continuation_process_async(
          future_queue<void, FEATURES> q_input_, future_queue<void> q_output_, CALLABLE callable_)
      : q_input(q_input_), q_output(q_output_), callable(callable_) {}
      void operator()() {
        while(true) {
          try {
            q_input.pop_wait();
            callable();
            // TODO how to handle full output queues?
            q_output.push();
          }
          catch(detail::TerminateInternalThread&) {
            q_output.d->continuation_process_async_terminated = true;
            return;
          }
          catch(...) {
            // TODO how to handle full output queues?
            q_output.push_exception(std::current_exception());
          }
        }
      }
      future_queue<void, FEATURES> q_input;
      future_queue<void> q_output;
      CALLABLE callable;
    };
 
    // factory for continuation_process_async
    template<typename T, typename FEATURES, typename TOUT, typename CALLABLE>
    continuation_process_async<T, FEATURES, TOUT, CALLABLE> make_continuation_process_async(
        future_queue<T, FEATURES> q_input, future_queue<TOUT> q_output, CALLABLE callable) {
      return {q_input, q_output, callable};
    }
  } // namespace detail
 
  // ----------------------------------------------------------------------------------------------------------------
  // ----------------------------------------------------------------------------------------------------------------
  // actual implementation of future_queue::then()
 
  template<typename T, typename FEATURES>
  template<typename T2, typename FEATURES2, typename CALLABLE>
  future_queue<T2, FEATURES2> future_queue<T, FEATURES>::then(CALLABLE callable, std::launch policy) {
    future_queue<T, FEATURES> q_input(*this);
    if(policy == std::launch::deferred) {
      future_queue<T2, FEATURES2> q_output(1);
      q_output.d->continuation_process_deferred =
          detail::make_continuation_process_deferred(q_input, q_output, callable);
      q_output.d->continuation_process_deferred_wait =
          detail::make_continuation_process_deferred_wait(q_input, q_output, callable);
      q_output.d->continuation_origin = *this;
      q_output.d->is_continuation_deferred = true;
      return q_output;
    }
    else {
      future_queue<T2, FEATURES2> q_output(size());
      q_output.d->continuation_process_async =
          std::thread(detail::make_continuation_process_async(q_input, q_output, callable));
      q_output.d->continuation_origin = *this;
      q_output.d->is_continuation_async = true;
      return q_output;
    }
  }
 
} // namespace cppext