map_reduce.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) 2020 ScyllaDB.
 */
 
#pragma once
 
#include <iterator>
 
#include <seastar/core/future.hh>
#include <seastar/core/shared_ptr.hh>
 
namespace seastar {
 
 
 
template <typename T, typename Ptr, bool IsFuture>
struct reducer_with_get_traits;
 
template <typename T, typename Ptr>
struct reducer_with_get_traits<T, Ptr, false> {
    using result_type = decltype(std::declval<T>().get());
    using future_type = future<result_type>;
    static future_type maybe_call_get(future<> f, Ptr r) {
        return f.then([r = std::move(r)] () mutable {
            return make_ready_future<result_type>(std::move(r->reducer).get());
        });
    }
};
 
template <typename T, typename Ptr>
struct reducer_with_get_traits<T, Ptr, true> {
    using future_type = decltype(std::declval<T>().get());
    static future_type maybe_call_get(future<> f, Ptr r) {
        return f.then([r = std::move(r)] () mutable {
            return r->reducer.get();
        }).then_wrapped([r] (future_type f) {
            return f;
        });
    }
};
 
template <typename T, typename Ptr = lw_shared_ptr<T>, typename V = void>
struct reducer_traits {
    using future_type = future<>;
    static future_type maybe_call_get(future<> f, Ptr r) {
        return f.then([r = std::move(r)] {});
    }
};
 
template <typename T, typename Ptr>
struct reducer_traits<T, Ptr, decltype(std::declval<T>().get(), void())> : public reducer_with_get_traits<T, Ptr, is_future<std::invoke_result_t<decltype(&T::get),T>>::value> {};
 
 
 
// TODO: specialize for non-deferring reducer
template <typename Iterator, typename Mapper, typename Reducer>
SEASTAR_CONCEPT( requires requires (Iterator i, Mapper mapper, Reducer reduce) {
     *i++;
     { i != i } -> std::convertible_to<bool>;
     mapper(*i);
     reduce(futurize_invoke(mapper, *i).get0());
} )
inline
auto
map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Reducer&& r)
    -> typename reducer_traits<Reducer>::future_type
{
    struct state {
        Mapper mapper;
        Reducer reducer;
    };
    auto s = make_lw_shared(state{std::forward<Mapper>(mapper), std::forward<Reducer>(r)});
    future<> ret = make_ready_future<>();
    while (begin != end) {
        ret = futurize_invoke(s->mapper, *begin++).then_wrapped([ret = std::move(ret), s] (auto f) mutable {
            return ret.then_wrapped([f = std::move(f), s] (auto rf) mutable {
                if (rf.failed()) {
                    f.ignore_ready_future();
                    return rf;
                } else {
                    return futurize_invoke(s->reducer, std::move(f.get0()));
                }
            });
        });
    }
    return reducer_traits<Reducer, lw_shared_ptr<state>>::maybe_call_get(std::move(ret), s);
}
 
template <typename Iterator, typename Mapper, typename Initial, typename Reduce>
SEASTAR_CONCEPT( requires requires (Iterator i, Mapper mapper, Initial initial, Reduce reduce) {
     *i++;
     { i != i} -> std::convertible_to<bool>;
     mapper(*i);
     requires is_future<decltype(mapper(*i))>::value;
     { reduce(std::move(initial), mapper(*i).get0()) } -> std::convertible_to<Initial>;
} )
inline
future<Initial>
map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Initial initial, Reduce reduce) {
    struct state {
        Mapper mapper;
        Initial result;
        Reduce reduce;
    };
    auto s = make_lw_shared(state{std::forward<Mapper>(mapper), std::move(initial), std::move(reduce)});
    future<> ret = make_ready_future<>();
    while (begin != end) {
        ret = futurize_invoke(s->mapper, *begin++).then_wrapped([s = s.get(), ret = std::move(ret)] (auto f) mutable {
            try {
                s->result = s->reduce(std::move(s->result), std::move(f.get0()));
                return std::move(ret);
            } catch (...) {
                return std::move(ret).then_wrapped([ex = std::current_exception()] (auto f) {
                    f.ignore_ready_future();
                    return make_exception_future<>(ex);
                });
            }
        });
    }
    return ret.then([s] {
        return make_ready_future<Initial>(std::move(s->result));
    });
}
 
template <typename Range, typename Mapper, typename Initial, typename Reduce>
SEASTAR_CONCEPT( requires requires (Range range, Mapper mapper, Initial initial, Reduce reduce) {
     std::begin(range);
     std::end(range);
     mapper(*std::begin(range));
     requires is_future<std::remove_reference_t<decltype(mapper(*std::begin(range)))>>::value;
     { reduce(std::move(initial), mapper(*std::begin(range)).get0()) } -> std::convertible_to<Initial>;
} )
inline
future<Initial>
map_reduce(Range&& range, Mapper&& mapper, Initial initial, Reduce reduce) {
    return map_reduce(std::begin(range), std::end(range), std::forward<Mapper>(mapper),
            std::move(initial), std::move(reduce));
}
 
template <typename Result, typename Addend = Result>
class adder {
private:
    Result _result;
public:
    future<> operator()(const Addend& value) {
        _result += value;
        return make_ready_future<>();
    }
    Result get() && {
        return std::move(_result);
    }
};
 
 
} // namespace seastar