shared_future.hh

/*
 * This file is open source software, licensed to you under the terms
 * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
 * distributed with this work for additional information regarding copyright
 * ownership.  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.
 */
 
/*
 * Copyright (C) 2015 ScyllaDB
 */
 
#pragma once
 
#ifndef SEASTAR_MODULE
#include <seastar/core/future.hh>
#include <seastar/core/abortable_fifo.hh>
#include <seastar/core/abort_on_expiry.hh>
#include <seastar/core/timed_out_error.hh>
#include <seastar/util/modules.hh>
#include <exception>
#include <optional>
#endif
 
namespace seastar {
 
 
template<typename Clock>
struct with_clock {};
 
 
template <typename... T>
struct future_option_traits;
 
template <typename Clock, typename T>
struct future_option_traits<with_clock<Clock>, T> {
    using clock_type = Clock;
 
    using future_type = future<T>;
    using promise_type = promise<T>;
};
 
template <typename Clock>
struct future_option_traits<with_clock<Clock>> {
    using clock_type = Clock;
 
    using future_type = future<>;
    using promise_type = promise<>;
};
 
template <typename T>
struct future_option_traits<T> : public future_option_traits<with_clock<lowres_clock>, T> {
};
 
template <>
struct future_option_traits<> : future_option_traits<void> {
};
 
 
template<typename... T>
class shared_future {
    template <typename... U> friend class shared_promise;
    using options = future_option_traits<T...>;
public:
    using clock = typename options::clock_type;
    using time_point = typename clock::time_point;
    using future_type = typename future_option_traits<T...>::future_type;
    using promise_type = typename future_option_traits<T...>::promise_type;
    using value_tuple_type = typename future_type::tuple_type;
private:
    class shared_state final : public enable_lw_shared_from_this<shared_state>, public task {
        future_type _original_future;
        // Ensures that shared_state is alive until run_and_dispose() runs (if the task was scheduled)
        lw_shared_ptr<shared_state> _keepaliver;
        struct entry {
            promise_type pr;
            std::optional<abort_on_expiry<clock>> timer;
        };
 
        struct entry_expiry {
            void operator()(entry& e) noexcept {
                if (e.timer) {
                    e.pr.set_exception(std::make_exception_ptr(timed_out_error()));
                } else {
                    e.pr.set_exception(std::make_exception_ptr(abort_requested_exception()));
                }
            };
        };
 
        internal::abortable_fifo<entry, entry_expiry> _peers;
 
    public:
        ~shared_state() {
            // Don't warn if the shared future is exceptional. Any
            // warnings will be reported by the futures returned by
            // get_future.
            if (_original_future.failed()) {
                _original_future.ignore_ready_future();
            }
        }
        explicit shared_state(future_type f) noexcept
                : task(default_scheduling_group()) // SG is set later, when the task is scheduled
                , _original_future(std::move(f)) {
        }
        void run_and_dispose() noexcept override {
            auto& state = _original_future._state;
            if (_original_future.failed()) {
                while (_peers) {
                    _peers.front().pr.set_exception(state.get_exception());
                    _peers.pop_front();
                }
            } else {
                while (_peers) {
                    auto& p = _peers.front().pr;
                    try {
                        p.set_value(state.get_value());
                    } catch (...) {
                        p.set_exception(std::current_exception());
                    }
                    _peers.pop_front();
                }
            }
            _keepaliver.release();
        }
 
        task* waiting_task() noexcept override {
            return nullptr;
        }
 
        future_type get_future(time_point timeout = time_point::max()) noexcept {
            // Note that some functions called below may throw,
            // like pushing to _peers or copying _original_future's ready value.
            // We'd rather terminate than propagate these errors similar to
            // .then()'s failure to allocate a continuation as the caller cannot
            // distinguish between an error returned by the original future to
            // failing to perform `get_future` itself.
            memory::scoped_critical_alloc_section _;
            if (!_original_future.available()) {
                entry& e = _peers.emplace_back();
 
                auto f = e.pr.get_future();
                if (timeout != time_point::max()) {
                    e.timer.emplace(timeout);
                    abort_source& as = e.timer->abort_source();
                   _peers.make_back_abortable(as);
                }
                if (!_keepaliver) {
                    this->set_scheduling_group(current_scheduling_group());
                    _original_future.set_task(*this);
                    _keepaliver = this->shared_from_this();
                }
                return f;
            } else if (_original_future.failed()) {
                return future_type(exception_future_marker(), std::exception_ptr(_original_future._state.get_exception()));
            } else {
                return future_type(ready_future_marker(), _original_future._state.get_value());
            }
        }
 
        future_type get_future(abort_source& as) noexcept {
            // Note that some functions called below may throw,
            // like pushing to _peers or copying _original_future's ready value.
            // We'd rather terminate than propagate these errors similar to
            // .then()'s failure to allocate a continuation as the caller cannot
            // distinguish between an error returned by the original future to
            // failing to perform `get_future` itself.
            memory::scoped_critical_alloc_section _;
            if (!_original_future.available()) {
                entry& e = _peers.emplace_back();
 
                auto f = e.pr.get_future();
                _peers.make_back_abortable(as);
                if (!_keepaliver) {
                    this->set_scheduling_group(current_scheduling_group());
                    _original_future.set_task(*this);
                    _keepaliver = this->shared_from_this();
                }
                return f;
            } else if (_original_future.failed()) {
                return future_type(exception_future_marker(), std::exception_ptr(_original_future._state.get_exception()));
            } else {
                return future_type(ready_future_marker(), _original_future._state.get_value());
            }
        }
 
        bool available() const noexcept {
            return _original_future.available();
        }
 
        bool failed() const noexcept {
            return _original_future.failed();
        }
 
        // Used only in tests (see shared_future_tester in futures_test.cc)
        bool has_scheduled_task() const noexcept {
            return _keepaliver != nullptr;
        }
    };
    lw_shared_ptr<shared_state> _state;
public:
    shared_future(future_type f)
        : _state(make_lw_shared<shared_state>(std::move(f))) { }
 
    shared_future() = default; // noexcept, based on the respective lw_shared_ptr constructor
    shared_future(const shared_future&) = default; // noexcept, based on the respective lw_shared_ptr constructor
    shared_future& operator=(const shared_future&) = default; // noexcept, based on respective constructor
    shared_future(shared_future&&) = default; // noexcept, based on the respective lw_shared_ptr constructor
    shared_future& operator=(shared_future&&) = default; // noexcept, based on the respective constructor
 
    future_type get_future(time_point timeout = time_point::max()) const noexcept {
        return _state->get_future(timeout);
    }
 
    future_type get_future(abort_source& as) const noexcept {
        return _state->get_future(as);
    }
 
    bool available() const noexcept {
        return _state->available();
    }
 
    bool failed() const noexcept {
        return _state->failed();
    }
 
    operator future_type() const noexcept {
        return get_future();
    }
 
    bool valid() const noexcept {
        return bool(_state);
    }
 
    friend class shared_future_tester;
};
 
SEASTAR_MODULE_EXPORT
template <typename... T>
class shared_promise {
public:
    using shared_future_type = shared_future<T...>;
    using future_type = typename shared_future_type::future_type;
    using promise_type = typename shared_future_type::promise_type;
    using clock = typename shared_future_type::clock;
    using time_point = typename shared_future_type::time_point;
    using value_tuple_type = typename shared_future_type::value_tuple_type;
private:
    promise_type _promise;
    shared_future_type _shared_future;
    static constexpr bool copy_noexcept = future_type::copy_noexcept;
public:
    shared_promise(const shared_promise&) = delete;
    shared_promise(shared_promise&&) = default; // noexcept, based on the respective promise and shared_future constructors
    shared_promise& operator=(shared_promise&&) = default; // noexcept, based on the respective promise and shared_future constructors
    shared_promise() : _promise(), _shared_future(_promise.get_future()) {
    }
 
    future_type get_shared_future(time_point timeout = time_point::max()) const noexcept {
        return _shared_future.get_future(timeout);
    }
 
    future_type get_shared_future(abort_source& as) const noexcept {
        return _shared_future.get_future(as);
    }
 
    void set_value(const value_tuple_type& result) noexcept(copy_noexcept) {
        _promise.set_value(result);
    }
 
    void set_value(value_tuple_type&& result) noexcept {
        _promise.set_value(std::move(result));
    }
 
    template <typename... A>
    void set_value(A&&... a) noexcept {
        _promise.set_value(std::forward<A>(a)...);
    }
 
    void set_exception(std::exception_ptr ex) noexcept {
        _promise.set_exception(std::move(ex));
    }
 
    template<typename Exception>
    void set_exception(Exception&& e) noexcept {
        set_exception(make_exception_ptr(std::forward<Exception>(e)));
    }
 
    bool available() const noexcept {
        return _shared_future.available();
    }
 
    bool failed() const noexcept {
        return _shared_future.failed();
    }
};
 
 
}