rmqt_future.h

// Copyright 2020-2023 Bloomberg Finance L.P.
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
 
#ifndef INCLUDED_RMQT_FUTURE
#define INCLUDED_RMQT_FUTURE
 
#include <rmqt_result.h>
 
#include <ball_log.h>
#include <bdlf_bind.h>
#include <bslmt_condition.h>
#include <bslmt_lockguard.h>
#include <bslmt_mutex.h>
#include <bsls_assert.h>
#include <bsls_systemtime.h>
 
#include <bsl_exception.h>
#include <bsl_functional.h>
#include <bsl_list.h>
#include <bsl_memory.h>
#include <bsl_utility.h>
 
//@PURPOSE: An async-style Future/Promise object
//
//@CLASSES:
//  rmqt::Future: Represents an action which may or may not have completed yet.
//
//  rmqt::Future<T>::Impl: The thread-safe control block for Future<T>. These
//                         are shared between copies of `Future` objects.
//
// rmqt::FutureUtil: 'helper' functionality for working with Futures which is
//                    used in multiple places
 
namespace BloombergLP {
namespace rmqt {
 
 
template <typename T>
class Future;
 
template <typename T = void>
class Future {
  public:
    typedef bsl::function<void(const Result<T>& result)> Maker;
 
    typedef bsl::pair<typename Future<T>::Maker, Future<T> > Pair;
 
    static Pair make();
 
    static Pair make(const bsl::function<void()>& cancelFunction);
 
    ~Future();
 
    Result<T> tryResult();
 
    Result<T> blockResult();
 
    Result<T> timedWaitResult(const bsls::TimeInterval& absoluteTime);
 
    Result<T> waitResult(const bsls::TimeInterval& relativeTimeout);
 
    explicit Future(const Result<T>& result);
 
    Future(const Future& future);
 
    Future& operator=(const Future& future);
 
    template <typename newT>
    Future<newT>
    then(const bsl::function<Result<newT>(const Result<T>&)>& converter);
 
    template <typename newT>
    Future<newT>
    thenFuture(const bsl::function<Future<newT>(const Result<T>&)>& nextFuture);
 
  private:
    class Impl;
 
  private:
#ifdef __xlC__
    // IBM compilers don't feel the above friend declaration is good enough
    // to allow permission to a future of a different T's
    // impl, so this is necessary. But also doesn't sit
    // well with pretty much any other compiler
    template <typename newT>
    friend class typename Future<newT>::Impl;
#else
    // Required for Future<T>::Impl::addChain
    template <typename newT>
    friend class Future;
#endif
 
    Future();
 
    bsl::shared_ptr<Impl> d_impl;
}; // class Future
 
class FutureUtil {
 
  public:
    template <typename A, typename B>
    static Result<B> convertViaManagedPtr(const Result<A>& a);
 
    template <typename A, typename B>
    static bsl::function<Future<B>(const Result<A>&)> propagateError(
        const bsl::function<Future<B>(const bsl::shared_ptr<A>&)>& fn)
    {
        return bdlf::BindUtil::bind(
            &propagateErrorImplWithItem<A, B>, fn, bdlf::PlaceHolders::_1);
    }
 
    template <typename T>
    static bsl::function<Future<T>(const Result<void>&)>
    propagateError(const bsl::function<Future<T>()>& fn)
    {
 
        return bdlf::BindUtil::bind(
            &propagateErrorImpl<T>, fn, bdlf::PlaceHolders::_1);
    }
 
    template <typename T>
    static bsl::function<rmqt::Result<T>()>
    resultWrapper(const bsl::function<T()>& tProducer)
    {
        return bdlf::BindUtil::bind(&resultWrapperImpl<T>, tProducer);
    }
 
    template <typename T>
    static void processResult(const typename Future<T>::Maker& maker,
                              const bsl::function<Result<T>()>& resultProducer)
    {
        maker(resultProducer());
    }
 
    template <typename T>
    static bsl::function<Result<T>(const Result<T>&)>
    makerWrapper(const typename Future<T>::Maker& maker)
    {
        return bdlf::BindUtil::bind(
            &makerWrapperImpl<T>, maker, bdlf::PlaceHolders::_1);
    }
 
    template <typename T>
    static bsl::function<Future<T>(const Result<Future<T> >&)> unravelFuture()
    {
        bsl::function<Future<T>(const bsl::shared_ptr<Future<T> >&)> convert =
            bdlf::BindUtil::bind(&unravelImpl<T>, bdlf::PlaceHolders::_1);
        return propagateError<Future<T> >(convert);
    }
 
    template <typename T>
    static Future<T> flatten(Future<Future<T> > futurefuture)
    {
        return futurefuture.template thenFuture<T>(
            FutureUtil::unravelFuture<T>());
    }
 
  private:
    template <typename T>
    static Future<T> propagateErrorImpl(const bsl::function<Future<T>()>& t,
                                        const Result<void>& r)
    {
        if (!r) {
            typename Future<T>::Pair fail = Future<T>::make();
            fail.first(Result<T>(r.error(), r.returnCode()));
            return fail.second;
        }
        return t();
    }
 
    template <typename A, typename B>
    static Future<B> propagateErrorImplWithItem(
        const bsl::function<Future<B>(const bsl::shared_ptr<A>&)>& b,
        const Result<A>& a)
    {
        if (!a) {
            typename Future<B>::Pair fail = Future<B>::make();
            fail.first(Result<B>(a.error(), a.returnCode()));
            return fail.second;
        }
        return b(a.value());
    }
 
    template <typename T>
    static Result<T> makerWrapperImpl(const typename Future<T>::Maker& maker,
                                      const Result<T>& result)
    {
        maker(result);
        return result;
    }
 
    template <typename T>
    static Future<T> unravelImpl(const bsl::shared_ptr<Future<T> >& t)
    {
        return *t;
    }
 
    template <typename T>
    static Result<T> resultWrapperImpl(const bsl::function<T()>& tProducer);
};
 
template <typename T>
bsl::pair<typename Future<T>::Maker, Future<T> > Future<T>::make()
{
    Future<T> f;
    f.d_impl = bsl::make_shared<Impl>(bsl::function<void()>());
 
    return bsl::make_pair(f.d_impl->generateMaker(), f);
}
 
template <typename T>
bsl::pair<typename Future<T>::Maker, Future<T> >
Future<T>::make(const bsl::function<void()>& cancelFunction)
{
    Future<T> f;
    f.d_impl = bsl::make_shared<Impl>(cancelFunction);
 
    return bsl::make_pair(f.d_impl->generateMaker(), f);
}
 
template <typename T>
Future<T>::~Future()
{
}
 
template <typename T>
Result<T> Future<T>::tryResult()
{
    return waitResult(bsls::TimeInterval(0));
}
 
template <typename T>
Result<T> Future<T>::blockResult()
{
    d_impl->blockUntilMade();
    return d_impl->result();
}
 
template <typename T>
Result<T> Future<T>::waitResult(const bsls::TimeInterval& relativeTimeout)
{
    bsls::TimeInterval timeoutTime = bsls::SystemTime::nowRealtimeClock();
    timeoutTime += relativeTimeout;
    return timedWaitResult(timeoutTime);
}
 
template <typename T>
Result<T> Future<T>::timedWaitResult(const bsls::TimeInterval& absoluteTime)
{
    if (d_impl->timedWaitUntilMade(absoluteTime)) {
        return d_impl->result();
    }
    return Result<T>("TIMED OUT", TIMEOUT);
}
 
template <typename T>
Future<T>::Future(const rmqt::Result<T>& result)
: d_impl()
{
    Pair p = make();
    d_impl = p.second.d_impl;
    p.first(result);
}
 
template <typename T>
Future<T>::Future()
: d_impl()
{
}
 
template <typename T>
Future<T>::Future(const Future<T>& future)
: d_impl(future.d_impl)
{
}
 
template <typename T>
Future<T>& Future<T>::operator=(const Future<T>& future)
{
    this->d_impl = future.d_impl;
    return *this;
}
 
template <typename T>
template <typename newT>
Future<newT>
Future<T>::then(const bsl::function<Result<newT>(const Result<T>&)>& converter)
{
    return d_impl->addChain(converter);
}
 
template <typename T>
template <typename newT>
Future<newT> Future<T>::thenFuture(
    const bsl::function<Future<newT>(const Result<T>&)>& nextFutureMaker)
{
    return d_impl->addChain(nextFutureMaker);
}
 
template <typename T>
Result<T> FutureUtil::resultWrapperImpl(const bsl::function<T()>& tProducer)
{
    return rmqt::Result<T>(bsl::make_shared<T>(tProducer()));
}
 
template <>
Result<void>
FutureUtil::resultWrapperImpl<void>(const bsl::function<void()>& tProducer);
 
template <typename A, typename B>
Result<B> FutureUtil::convertViaManagedPtr(const Result<A>& a)
{
    if (a) {
        bslma::ManagedPtr<A> am(a.value().managedPtr());
        return Result<B>(bsl::shared_ptr<B>(new B(am)));
    }
    return Result<B>(a.error(), a.returnCode());
}
 
template <typename T>
class Future<T>::Impl
: public bsl::enable_shared_from_this<typename Future<T>::Impl> {
  private:
    static void made(const bsl::weak_ptr<typename Future<T>::Impl>& weakSelf,
                     const Result<T>& item)
    {
        bsl::shared_ptr<Future<T>::Impl> self = weakSelf.lock();
 
        if (!self) {
            BALL_LOG_SET_CATEGORY("RMQT.FUTURE.IMPL");
            BALL_LOG_DEBUG << "Resolved " << bsl::string(!item ? "un" : "")
                           << "successful future<" << typeid(T).name()
                           << "> after cancel";
            // This is a safe race condition - e.g. if a user stops caring about
            // a future but the future resolves anyway in the background, we
            // just drop the result.
            return;
        }
 
        bslmt::LockGuard<bslmt::Mutex> guard(&self->d_mutex);
        self->d_result = item;
        self->d_done   = true;
        self->notifyChain(&guard);
        self->d_condition.broadcast(); // there could be more than one waiter
    }
 
    template <typename newT>
    void static converter(
        const typename Future<newT>::Maker& maker,
        const bsl::function<Result<newT>(const Result<T>&)>& newTConverter,
        const Result<T>& result)
    {
        maker(newTConverter(result));
    }
 
    template <typename newT>
    static void futureConverter(
        const bsl::function<Future<newT>(const Result<T>&)>& cGenerator,
        const typename Future<newT>::Maker& bMaker,
        const bsl::weak_ptr<typename Future<newT>::Impl>& wpBImpl,
        const Result<T>& aResult)
    {
        bsl::shared_ptr<typename Future<newT>::Impl> bImpl = wpBImpl.lock();
        if (bImpl) {
            Future<newT> cFut = cGenerator(aResult);
 
            Future<newT> dFut =
                cFut.then(FutureUtil::makerWrapper<newT>(bMaker));
 
            bImpl->updateCancel(dFut.d_impl->getLifetimeExtension());
        }
    }
 
    static void
    extendChainParentLifetime(const bsl::shared_ptr<typename Future<T>::Impl>&)
    {
        // This function exists only as a bind target so that a Future
        // created in .then() can hold onto the parent Future to keep it
        // alive as long as itself.
 
        // E.g.
 
        // rmqt::Future<bsl::string> asyncGetMultiplyAsString(int a, int b)
        // {
        //     rmqt::Future<int> x = asyncMultiply(a, b);
        //     rmqt::Future<bsl::string> y =
        //     x.then<bsl::string>(&convertIntToString);
        //
        //     return y;
        // }
 
        // In the above example, `x` will go out of scope at the end of the
        // function, but it's successor `y` is still alive.
 
        // Since the type of `y` does not know about `x`, but it needs to
        // extend it's lifetime (call the `x` destructor when `y`
        // destructs), we need to hold a type-erased form of `x`
 
        // Note: if `y` then goes out of scope, the whole future chain
        // should be destructed
    }
 
  public:
    Impl(const bsl::function<void()>& cancelFunc)
    : d_mutex()
    , d_condition()
    , d_result("Null")
    , d_chain()
    , d_done(false)
    , d_cancelFunc(cancelFunc)
    {
    }
 
    ~Impl()
    {
        BALL_LOG_SET_CATEGORY("RMQT.FUTURE.IMPL");
        try {
            bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);
            if (!d_done && d_cancelFunc) {
                d_cancelFunc();
            }
        }
        catch (bsl::exception& e) {
            BALL_LOG_ERROR << "Caught exception in future<" << typeid(T).name()
                           << "> dtor: " << e.what();
        }
        catch (...) {
            BALL_LOG_ERROR << "Caught unknown exception in future<"
                           << typeid(T).name() << "> dtor";
        }
    }
 
    void blockUntilMade()
    {
        bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);
        while (!d_done) {
            d_condition.wait(&d_mutex);
        }
    }
 
    // returns true if there was no timeout
    bool timedWaitUntilMade(const bsls::TimeInterval& absoluteTime)
    {
        bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);
        while (!d_done) {
            if (bslmt::Condition::e_TIMED_OUT ==
                d_condition.timedWait(&d_mutex, absoluteTime)) {
                return false;
            }
        }
        return true;
    }
 
    typename Future<T>::Maker generateMaker()
    {
        return bdlf::BindUtil::bind(&Future<T>::Impl::made,
                                    Future<T>::Impl::weak_from_this(),
                                    bdlf::PlaceHolders::_1);
    }
 
    const Result<T>& result(bslmt::LockGuard<bslmt::Mutex>* haveGuard = 0)
    {
        if (!haveGuard) {
            bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);
            return resultNoLock();
        }
        BSLS_ASSERT(haveGuard->ptr() == &d_mutex);
        return resultNoLock();
    }
 
    template <typename newT>
    Future<newT>
    addChain(const bsl::function<Result<newT>(const Result<T>&)>& newTConverter)
    {
        typename rmqt::Future<newT>::Pair futurePair(
            rmqt::Future<newT>::make(getLifetimeExtension()));
 
        bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);
        if (d_done) {
            converter(futurePair.first, newTConverter, result(&guard));
        }
        else {
            d_chain.push_back(
                bdlf::BindUtil::bind(&Future<T>::Impl::converter<newT>,
                                     futurePair.first,
                                     newTConverter,
                                     bdlf::PlaceHolders::_1));
        }
 
        return futurePair.second;
    }
 
    template <typename newT>
    Future<newT>
    addChain(const bsl::function<Future<newT>(const Result<T>&)>& futureMaker)
    {
        typename rmqt::Future<newT>::Pair futurePair(
            rmqt::Future<newT>::make(getLifetimeExtension()));
 
        bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);
        if (d_done) {
            futureConverter<newT>(
                futureMaker,
                futurePair.first,
                bsl::weak_ptr<typename rmqt::Future<newT>::Impl>(
                    futurePair.second.d_impl),
                result(&guard));
        }
        else {
            d_chain.push_back(bdlf::BindUtil::bind(
                &Future<T>::Impl::futureConverter<newT>,
                futureMaker,
                futurePair.first,
                bsl::weak_ptr<typename rmqt::Future<newT>::Impl>(
                    futurePair.second.d_impl),
                bdlf::PlaceHolders::_1));
        }
        return futurePair.second;
    }
 
    bsl::function<void()> getLifetimeExtension()
    {
        bsl::function<void(const bsl::shared_ptr<typename Future<T>::Impl>&)>
            func;
        func = &Future<T>::Impl::extendChainParentLifetime;
        return bdlf::BindUtil::bind(func, Future<T>::Impl::shared_from_this());
    }
 
    void updateCancel(const bsl::function<void()>& newCanc)
    {
        bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);
        d_cancelFunc = newCanc;
    }
 
  private:
    const Result<T>& resultNoLock()
    {
        BSLS_ASSERT(d_done);
        return d_result;
    }
 
    void notifyChain(bslmt::LockGuard<bslmt::Mutex>* haveGuard)
    {
        BSLS_ASSERT(haveGuard);
        typename bsl::list<bsl::function<void(const Result<T>&)> >::iterator it;
        for (it = d_chain.begin(); it != d_chain.end(); ++it) {
            (*it)(result(haveGuard));
        }
        d_chain.clear();
    }
 
  private:
    bslmt::Mutex d_mutex;
    bslmt::Condition d_condition;
    Result<T> d_result;
    bsl::list<typename rmqt::Future<T>::Maker> d_chain;
    bool d_done;
 
    // Called if the future is not completed when Future<T>::Impl destructs
    // This function is used to extend the lifetime of chained futures.
    // See extendChainParentLifetime
    bsl::function<void()> d_cancelFunc;
};
 
} // namespace rmqt
} // namespace BloombergLP
 
#endif