ValueStreamChannels.hpp

#pragma once

#include "ValueStream.hpp"

#include "BackendComponents.hpp"

#include <boost/intrusive/set.hpp>
#include <deque>
#include <queue>

namespace Showcase
{

  template <typename T> class ChannelGuard;
  template <typename T> class CallbackLock;

  template <typename T> class ValueStreamSplitter;


  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------



  template <typename T>
  class ValueStreamChannel : public ValueStream<T>
  {
  public:
    ValueStreamChannel(int channel, std::shared_ptr<ValueStreamSplitter<T>> source);
    virtual ~ValueStreamChannel();

    virtual folly::Future<T> next() override;

    std::shared_ptr<ValueStreamSplitter<T>> splitter();

  private:
    std::shared_ptr<ValueStreamSplitter<T>> m_source;
    const int m_channel;
  };
  template <typename T>
  using ValueStreamChannelPtr = std::shared_ptr<ValueStreamChannel<T>>;


  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------


  template <typename T>
  class CloneableValueStreamChannel : public ValueStream<CallbackLock<T>>
  {
  public:
    CloneableValueStreamChannel(int channel, std::shared_ptr<ValueStreamSplitter<T>> source);
    virtual ~CloneableValueStreamChannel();

    virtual folly::Future<CallbackLock<T>> next() override;
    folly::Future<T> nextUnwrapped();

    ChannelGuard<ValueStreamChannel<T>> clone();
    ChannelGuard<CloneableValueStreamChannel<T>> clone2(); // a bit crappy name....

    std::shared_ptr<ValueStreamSplitter<T>> splitter();

  private:
    std::shared_ptr<ValueStreamSplitter<T>> m_source;
    const int m_channel;
  };
  template <typename T>
  using CloneableValueStreamChannelPtr = std::shared_ptr<CloneableValueStreamChannel<T>>;


  template <typename T>
  std::shared_ptr<CloneableValueStreamChannel<T>>
  getCloneableChannel(ValueStreamPtr<T> stream)
  {
    if(auto ch = std::dynamic_pointer_cast<CloneableValueStreamChannel<T>>(stream)) {
      return ch;
    } else {
      return getSplitter(stream)->openCloneableChannel().channel();
    }
  }


  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------


  template <typename T>
  struct ChannelInfo : public boost::intrusive::set_base_hook<>
  {
    ChannelInfo(int ch, bool cloneable)
      : nextCachedValue(0), channel(ch), onTheWay(0), isCloneable(cloneable)
    {}

    bool operator<(const ChannelInfo& ci) const { return channel < ci.channel; }
    bool operator>(const ChannelInfo& ci) const { return channel > ci.channel; }
    bool operator==(const ChannelInfo& ci) const { return channel == ci.channel; }

    size_t nextCachedValue;
    int channel;
    std::queue<folly::Promise<CallbackLock<T>>> promises;


    std::mutex channelMutex;
    std::condition_variable condition;
    size_t onTheWay;
    const bool isCloneable;
  };


  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------


  template <typename T>
  class ValueStreamSplitter : public ValueStream<T>
  {
  public:
    using ReturnType = CallbackLock<T>;

    ValueStreamSplitter(std::shared_ptr<ValueStream<T>> stream);
    virtual ~ValueStreamSplitter();

    folly::Future<ReturnType> next(int channel);

    ChannelGuard<ValueStreamChannel<T>> pauseChannels();

    ChannelGuard<ValueStreamChannel<T>> openChannel();
    ChannelGuard<CloneableValueStreamChannel<T>> openCloneableChannel();

    ChannelGuard<ValueStreamChannel<T>> cloneChannel(int target);
    ChannelGuard<CloneableValueStreamChannel<T>> cloneChannel2(int target);

    void closeChannel(int channel);
    bool hasChannel(int channel) const;

    std::shared_ptr<ValueStream<T>> stream();

    size_t channelsOpen() const;

  private:
    virtual folly::Future<T> next() override;

    ChannelGuard<ValueStreamChannel<T>> newNonCloneableChannel(int cloneTarget, bool clone);
    ChannelGuard<CloneableValueStreamChannel<T>> newCloneableChannel(int cloneTarget, bool clone);

    int newChannelImpl(int cloneTarget, bool clone, bool isCloneable,
                       std::unique_lock<std::recursive_mutex>& splitterLock,
                       std::unique_lock<std::mutex>& channelLock);

    void clearUsedCache();
    int findFreeChannel() const;
    void initChannel(int channel, bool isCloneable);
    void launchNewQuery();
    void receivedValue(const T& val, std::unique_lock<std::recursive_mutex>&& channelLock);

    boost::intrusive::set<ChannelInfo<T>> m_channels;

    std::shared_ptr<ValueStream<T>> m_stream;
    size_t m_futuresWaited;
    // Used for keeping the values in the same order they are fed from source
    size_t m_valueCounter;
    size_t m_valuesHandled;
    std::condition_variable m_valuesInOrderCondition;
    std::mutex m_valuesHandledMutex;

    std::deque<T> m_cachedValues;

    mutable std::recursive_mutex m_channelsMutex;
  };

  template <typename T>
  std::shared_ptr<ValueStreamSplitter<T>> getSplitter(ValueStreamPtr<T> stream)
  {
    if(auto ptr = std::dynamic_pointer_cast<ValueStreamSplitter<T>>(stream))
      return ptr;
    else if(auto ptr = std::dynamic_pointer_cast<ValueStreamChannel<T>>(stream))
      return ptr->splitter();
    else if(auto ptr = std::dynamic_pointer_cast<CloneableValueStreamChannel<T>>(stream))
      return ptr->splitter();
    else
      return std::make_shared<ValueStreamSplitter<T>>(stream);
  }


  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------



  template <typename Channel>
  class ChannelGuard
  {
  public:
    /* Please give me static_if or some sane way to do sanity checks!
    static_assert(HasValueType<Channel>::value && (
                  std::is_same<Channel, ValueStreamChannel<typename Channel::ValueType>>::value
                    || (HasValueType<typename Channel::ValueType>::value &&
                        std::is_same<typename Channel::ValueType, CallbackLock<typename Channel::ValueType::ValueType>>::value &&
                        std::is_same<Channel, CloneableValueStreamChannel<typename Channel::ValueType::ValueType>>::value)
                    ),
                  "Template parameter is not ValueStreamChannel.");
    */

    ChannelGuard();
    ChannelGuard(std::shared_ptr<Channel> channel, std::unique_lock<std::recursive_mutex>&& splitterLock,
                 std::unique_lock<std::mutex>&& channelLock);

    ChannelGuard(ChannelGuard&& b) = default;
    ChannelGuard& operator=(ChannelGuard&& b);

    ChannelGuard(const ChannelGuard&) = delete;
    ChannelGuard& operator=(const ChannelGuard&) = delete;

    std::shared_ptr<Channel> channel() const;

  private:
    std::shared_ptr<Channel> m_channel;
    std::unique_lock<std::recursive_mutex> m_splitterLock;
    std::unique_lock<std::mutex> m_channelLock;
  };


  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------


  template <typename T>
  class CallbackLock
  {
  public:
    typedef T ValueType;

    CallbackLock(T&& val, std::mutex& channelMutex,
                 std::condition_variable& cv, size_t& onTheWay, bool useSync);

    CallbackLock(CallbackLock&& b);

    CallbackLock& operator=(CallbackLock&& b) = delete;
    CallbackLock(const CallbackLock&) = delete;
    CallbackLock& operator=(const CallbackLock&) = delete;

    ~CallbackLock();

    const T& value() const;

  private:
    std::mutex& m_channelMutex;
    std::condition_variable& m_condition;
    size_t& m_onTheWay;
    T m_value;
    bool m_useGuard;
  };


  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------

  template <typename Stream>
  ChannelGuard<Stream>::ChannelGuard()
    : m_channel(nullptr)
  {
  }

  template <typename Stream>
  ChannelGuard<Stream>::ChannelGuard(std::shared_ptr<Stream> channel,
                                     std::unique_lock<std::recursive_mutex>&& splitterLock,
                                     std::unique_lock<std::mutex>&& channelLock)
    : m_channel(channel),
      m_splitterLock(std::move(splitterLock)),
      m_channelLock(std::move(channelLock))
  {
  }

  template <typename Stream>
  ChannelGuard<Stream>& ChannelGuard<Stream>::operator=(ChannelGuard<Stream> &&b)
  {
    m_channel = std::move(b.m_channel);
    m_splitterLock = std::move(b.m_splitterLock);
    m_channelLock = std::move(b.m_channelLock);
    return *this;
  }

  template <typename Stream>
  std::shared_ptr<Stream> ChannelGuard<Stream>::channel() const
  {
    return m_channel;
  }


  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------


  template <typename T>
  CallbackLock<T>::CallbackLock(T&& val, std::mutex& channelMutex,
                                std::condition_variable& cv, size_t& onTheWay,
                                bool useSync)
    : m_channelMutex(channelMutex),
      m_condition(cv),
      m_onTheWay(onTheWay),
      m_value(std::move(val)),
      m_useGuard(useSync)
  {
  }

  template <typename T>
  CallbackLock<T>::CallbackLock(CallbackLock &&b)
    : m_channelMutex(b.m_channelMutex),
      m_condition(b.m_condition),
      m_onTheWay(b.m_onTheWay),
      m_value(std::move(b.m_value)),
      m_useGuard(b.m_useGuard)
  {
    b.m_useGuard = false;
  }

  template <typename T>
  CallbackLock<T>::~CallbackLock()
  {
    if(m_useGuard) {
      std::unique_lock<std::mutex> lck(m_channelMutex);
      assert(m_onTheWay > 0);
      --m_onTheWay;
      bool signal = m_onTheWay == 0;
      lck.unlock();
      if(signal)
        m_condition.notify_one();
    }
  }

  template <typename T>
  const T& CallbackLock<T>::value() const
  {
    return m_value;
  }


  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------


  template <typename T>
  ValueStreamChannel<T>::ValueStreamChannel(int channel, std::shared_ptr<ValueStreamSplitter<T>> source)
    : m_source(source),
      m_channel(channel)
  {
    assert(m_source);
  }

  template <typename T>
  ValueStreamChannel<T>::~ValueStreamChannel()
  {
    m_source->closeChannel(m_channel);
  }

  template <typename T>
  folly::Future<T> ValueStreamChannel<T>::next()
  {
    auto unwrap = [] (const CallbackLock<T>& cbl) -> T {
      return cbl.value();
    };
    return m_source->next(m_channel).thenValue(unwrap);
  }

  template <typename T>
  std::shared_ptr<ValueStreamSplitter<T>> ValueStreamChannel<T>::splitter()
  {
    return m_source;
  }


  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------


  template <typename T>
  CloneableValueStreamChannel<T>::
  CloneableValueStreamChannel(int channel, std::shared_ptr<ValueStreamSplitter<T>> source)
    : m_source(source),
      m_channel(channel)
  {
    assert(m_source);
  }

  template <typename T>
  CloneableValueStreamChannel<T>::~CloneableValueStreamChannel()
  {
    m_source->closeChannel(m_channel);
  }

  template <typename T>
  folly::Future<CallbackLock<T>> CloneableValueStreamChannel<T>::next()
  {
    return m_source->next(m_channel);
  }

  template <typename T>
  std::shared_ptr<ValueStreamSplitter<T>>
  CloneableValueStreamChannel<T>::splitter()
  {
    return m_source;
  }

  template <typename T>
  folly::Future<T> CloneableValueStreamChannel<T>::nextUnwrapped()
  {
    auto unwrap = [] (const CallbackLock<T>& cbl) -> T {
      return cbl.value();
    };
    return m_source->next(m_channel).thenValue(unwrap);
  }

  template <typename T>
  ChannelGuard<ValueStreamChannel<T>> CloneableValueStreamChannel<T>::clone()
  {
    return m_source->cloneChannel(m_channel);
  }

  template <typename T>
  ChannelGuard<CloneableValueStreamChannel<T>> CloneableValueStreamChannel<T>::clone2()
  {
    return m_source->cloneChannel2(m_channel);
  }

  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------


  template <typename T>
  struct ChannelInfoCmp
  {
    bool operator()(int channel, const ChannelInfo<T>& ci) const {
      return channel < ci.channel;
    }

    bool operator()(const ChannelInfo<T>& ci, int channel) const {
      return ci.channel < channel;
    }
  };

  template <typename T>
  struct ChannelInfoDisposer
  {
    void operator()(ChannelInfo<T>* ci)
    {
      if(ci->isCloneable) {
        std::unique_lock<std::mutex> g(ci->channelMutex);
        if(ci->onTheWay > 0) {
          ci->condition.wait(g, [ci] { return ci->onTheWay == 0; });
          assert(ci->onTheWay == 0);
        }
      }
      delete ci;
    }
  };


  // ------------------------------------------------------------------
  // ------------------------------------------------------------------
  // ------------------------------------------------------------------


  template <typename T>
  ValueStreamSplitter<T>::ValueStreamSplitter(ValueStreamPtr<T> stream)
    : m_stream(stream),
      m_futuresWaited(0),
      m_valueCounter(0),
      m_valuesHandled(0)
  {
  }

  template <typename T>
  ValueStreamSplitter<T>::~ValueStreamSplitter()
  {
    std::lock_guard<std::recursive_mutex> g(m_channelsMutex);
    m_stream.reset();
    m_channels.clear_and_dispose(ChannelInfoDisposer<T>());
  }

  template <typename T>
  folly::Future<CallbackLock<T>>
  ValueStreamSplitter<T>::next(int channel)
  {
    folly::Future<ReturnType> future = folly::makeFuture<ReturnType>(
           folly::exception_wrapper(std::runtime_error("empty")));

    bool newQuery = false;
    std::unique_lock<std::recursive_mutex> channelsLock(m_channelsMutex);
    {
      auto cit = m_channels.find(channel, ChannelInfoCmp<T>());
      assert(cit != m_channels.end());
      ChannelInfo<T>& ci = *cit;

      std::unique_lock<std::mutex> lck(ci.channelMutex);

      if(ci.nextCachedValue < m_cachedValues.size()) {
        T val = m_cachedValues[ci.nextCachedValue++];

        channelsLock.unlock();

        assert(ci.promises.empty());

        ++ci.onTheWay;

        CallbackLock<T> cbl(std::move(val), ci.channelMutex,
                            ci.condition, ci.onTheWay, ci.isCloneable);
        return folly::makeFuture(std::move(cbl));
      } else {
        assert(ci.promises.size() <= m_futuresWaited);
        if(m_futuresWaited == ci.promises.size()) {
          newQuery = true;
          ++m_futuresWaited;
        }

        folly::Promise<ReturnType> promise;
        future = promise.getFuture();
        ci.promises.emplace(std::move(promise));
      }
    }

    if(newQuery) {
      launchNewQuery();
    }

    return future;
  }

  template <typename T>
  ChannelGuard<ValueStreamChannel<T>> ValueStreamSplitter<T>::pauseChannels()
  {
    std::unique_lock<std::recursive_mutex> lock(m_channelsMutex);
    std::unique_lock<std::mutex> dummy;
    return ChannelGuard<ValueStreamChannel<T>>(nullptr, std::move(lock), std::move(dummy));
  }

  template <typename T>
  ChannelGuard<ValueStreamChannel<T>> ValueStreamSplitter<T>::openChannel()
  {
    return newNonCloneableChannel(0, false);
  }

  template <typename T>
  ChannelGuard<CloneableValueStreamChannel<T>>
  ValueStreamSplitter<T>::openCloneableChannel()
  {
    return newCloneableChannel(0, false);
  }

  template <typename T>
  ChannelGuard<ValueStreamChannel<T>> ValueStreamSplitter<T>::cloneChannel(int target)
  {
    return newNonCloneableChannel(target, false);
  }

  template <typename T>
  ChannelGuard<CloneableValueStreamChannel<T>>
  ValueStreamSplitter<T>::cloneChannel2(int target)
  {
    return newCloneableChannel(target, true);
  }

  template <typename T>
  void ValueStreamSplitter<T>::closeChannel(int channel)
  {
    std::lock_guard<std::recursive_mutex> g(m_channelsMutex);
    auto it = m_channels.find(channel, ChannelInfoCmp<T>());
    assert(it != m_channels.end());
    m_channels.erase_and_dispose(it, ChannelInfoDisposer<T>());
  }

  template <typename T>
  bool ValueStreamSplitter<T>::hasChannel(int channel) const
  {
    std::lock_guard<std::recursive_mutex> g(m_channelsMutex);
    return m_channels.find(channel, ChannelInfoCmp<T>()) == m_channels.end();
  }

  template <typename T>
  std::shared_ptr<ValueStream<T>> ValueStreamSplitter<T>::stream()
  {
    assert(m_channels.empty());
    return m_stream;
  }

  template <typename T>
  size_t ValueStreamSplitter<T>::channelsOpen() const
  {
    std::lock_guard<std::recursive_mutex> g(m_channelsMutex);
    return m_channels.size();
  }

  template <typename T>
  folly::Future<T> ValueStreamSplitter<T>::next()
  {
    return next(0).thenValue([] (const ReturnType& p) -> T { return p.value(); });
  }

  template <typename T>
  ChannelGuard<ValueStreamChannel<T>>
  ValueStreamSplitter<T>::newNonCloneableChannel(int cloneTarget, bool clone)
  {
    std::unique_lock<std::recursive_mutex> splitterLock;
    std::unique_lock<std::mutex> channelLock;
    int channel = newChannelImpl(cloneTarget, clone, false, splitterLock, channelLock);

    auto me = std::static_pointer_cast<ValueStreamSplitter<T>>(this->shared_from_this());
    auto stream = std::make_shared<ValueStreamChannel<T>>(channel, me);

    return ChannelGuard<ValueStreamChannel<T>>(stream, std::move(splitterLock), std::move(channelLock));
  }

  template <typename T>
  ChannelGuard<CloneableValueStreamChannel<T>>
  ValueStreamSplitter<T>::newCloneableChannel(int cloneTarget, bool clone)
  {
    std::unique_lock<std::recursive_mutex> splitterLock;
    std::unique_lock<std::mutex> channelLock;
    int channel = newChannelImpl(cloneTarget, clone, true, splitterLock, channelLock);

    auto me = std::static_pointer_cast<ValueStreamSplitter<T>>(this->shared_from_this());
    auto stream = std::make_shared<CloneableValueStreamChannel<T>>(channel, me);

    return ChannelGuard<CloneableValueStreamChannel<T>>(stream, std::move(splitterLock), std::move(channelLock));
  }

  template <typename T>
  int ValueStreamSplitter<T>::
  newChannelImpl(int target, bool clone, bool isCloneable,
                 std::unique_lock<std::recursive_mutex>& splitterLock,
                 std::unique_lock<std::mutex>& channelLock)
  {
    splitterLock = std::unique_lock<std::recursive_mutex>(m_channelsMutex);

    int channel = findFreeChannel();

    initChannel(channel, isCloneable);

    auto itt = m_channels.find(channel, ChannelInfoCmp<T>());
    assert(itt != m_channels.end());
    ChannelInfo<T>& newChannel = *itt;

    if(clone) {
      assert(m_channels.find(target, ChannelInfoCmp<T>()) != m_channels.end());
      auto it = m_channels.find(target, ChannelInfoCmp<T>());
      ChannelInfo<T>& ci = *it;

      assert(ci.isCloneable);
      channelLock = std::unique_lock<std::mutex>(ci.channelMutex);

      size_t& onTheWay = ci.onTheWay;
      ci.condition.wait(channelLock, [&onTheWay] { return onTheWay == 0; });

      newChannel.nextCachedValue = ci.nextCachedValue;

      assert(ci.promises.size() <= m_futuresWaited);
    } else {
      newChannel.nextCachedValue = m_cachedValues.size();
    }
    return channel;
  }

  template <typename T>
  void ValueStreamSplitter<T>::clearUsedCache()
  {
    for(auto it = m_channels.begin(); it != m_channels.end(); ++it) {
      it->channelMutex.lock();
    }

    size_t minNextValue = m_cachedValues.size();
    for(auto it = m_channels.begin(); it != m_channels.end(); ++it) {
      assert(it->nextCachedValue <= m_cachedValues.size());
      minNextValue = std::min(it->nextCachedValue, minNextValue);
    }

    auto updateIndices = [&] {
      for(auto it = m_channels.begin(); it != m_channels.end(); ++it) {
        assert(it->nextCachedValue >= minNextValue);
        it->nextCachedValue -= minNextValue;
      }
    };


    if(minNextValue > 0 && minNextValue == m_cachedValues.size()) {
      m_cachedValues.clear();
      updateIndices();
    } else if(minNextValue > 0) {
      assert(minNextValue < m_cachedValues.size());
      auto eraseIt = m_cachedValues.begin();
      std::advance(eraseIt, minNextValue);
      m_cachedValues.erase(m_cachedValues.begin(), eraseIt);
      updateIndices();
    }

    for(auto it = m_channels.begin(); it != m_channels.end(); ++it) {
      it->channelMutex.unlock();
    }
  }

  template <typename T>
  int ValueStreamSplitter<T>::findFreeChannel() const
  {

    auto it = m_channels.begin();
    if(it == m_channels.end()) {
      return 0;
    } else if (it->channel > 0) {
      return it->channel - 1;
    } else {
      auto itt = m_channels.rbegin();
      assert(itt != m_channels.rend());
      return itt->channel + 1;
    }
  }

  template <typename T>
  void ValueStreamSplitter<T>::initChannel(int channel, bool isCloneable)
  {
    typename boost::intrusive::set<ChannelInfo<T>>::insert_commit_data insert_data;
    auto ins_pair = m_channels.insert_check(channel, ChannelInfoCmp<T>(), insert_data);
    (void)ins_pair;
    assert(ins_pair.second);
    m_channels.insert_commit(*new ChannelInfo<T>(channel, isCloneable), insert_data);
  }

  template <typename T>
  void ValueStreamSplitter<T>::launchNewQuery()
  {
    std::weak_ptr<ValueStreamSplitter<T>> weak =
        std::static_pointer_cast<ValueStreamSplitter<T>>(this->shared_from_this());

    folly::Future<T> fut = folly::makeFuture<T>(
      folly::exception_wrapper(std::runtime_error("Empty future")));

    size_t count;
    {
      std::lock_guard<std::recursive_mutex> g(m_channelsMutex);

      count = m_valueCounter;
      ++m_valueCounter;
      fut = m_stream->next();
    }

    std::move(fut).thenValue([weak, count](const T& val) {
      auto me = weak.lock();
      if(!me)
        return;

      {
        std::unique_lock<std::mutex> valueMutex(me->m_valuesHandledMutex);
        me->m_valuesInOrderCondition.wait(valueMutex, [count, me] {
          return count == me->m_valuesHandled;
        });

        std::unique_lock<std::recursive_mutex> channelLock(me->m_channelsMutex);
        valueMutex.unlock();
        me->receivedValue(val, std::move(channelLock));
      }
      me->m_valuesInOrderCondition.notify_all();
    });
  }

  template <typename T>
  void ValueStreamSplitter<T>::receivedValue(const T& val, std::unique_lock<std::recursive_mutex>&& channelLock)
  {
    clearUsedCache();

    assert(m_futuresWaited > 0);
    --m_futuresWaited;

    m_cachedValues.push_back(val);

    std::set<int> channels;
    for(auto it = m_channels.begin(); it != m_channels.end(); ++it) {
      channels.insert(it->channel);
    }


    for(auto it = channels.begin(); it != channels.end(); ++it) {

      auto cit = m_channels.find(*it, ChannelInfoCmp<T>());
      if(cit == m_channels.end())
        continue;

      ChannelInfo<T>& ci = *cit;

      std::unique_lock<std::mutex> lck(ci.channelMutex);

      assert(ci.promises.size() <= m_futuresWaited + 1);

      if(!ci.promises.empty()) {
        size_t next = ci.nextCachedValue++;
        assert(ci.nextCachedValue == m_cachedValues.size() );

        ++ci.onTheWay;

        auto promise = std::move(ci.promises.front());
        ci.promises.pop();

        T value = m_cachedValues[next];
        auto& mutex = ci.channelMutex;
        auto& cond = ci.condition;
        auto& onTheWay = ci.onTheWay;
        bool cloneable = ci.isCloneable;

        lck.unlock(); 

        auto cblock = CallbackLock<T>(std::move(value), mutex, cond, onTheWay, cloneable);

        // promise may end up making calls to next via callback chain
        promise.setValue(std::move(cblock));
      }
    }
    std::lock_guard<std::mutex> g(m_valuesHandledMutex);
    ++m_valuesHandled;
    channelLock.unlock();
  }

}