text: /*
   * 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 {
  
  /// \addtogroup future-util
  /// @{
  
  /// \cond internal
  
  template <typename T, bool IsFuture>
  struct reducer_with_get_traits;
  
  template <typename T>
  struct reducer_with_get_traits<T, false> {
      using result_type = decltype(std::declval<T>().get());
      using future_type = future<result_type>;
      static future_type maybe_call_get(future<> f, lw_shared_ptr<T> r) {
          return f.then([r = std::move(r)] () mutable {
              return make_ready_future<result_type>(std::move(*r).get());
          });
      }
  };
  
  template <typename T>
  struct reducer_with_get_traits<T, true> {
      using future_type = decltype(std::declval<T>().get());
      static future_type maybe_call_get(future<> f, lw_shared_ptr<T> r) {
          return f.then([r = std::move(r)] () mutable {
              return r->get();
          }).then_wrapped([r] (future_type f) {
              return f;
          });
      }
  };
  
  template <typename T, typename V = void>
  struct reducer_traits {
      using future_type = future<>;
      static future_type maybe_call_get(future<> f, lw_shared_ptr<T> r) {
          return f.then([r = std::move(r)] {});
      }
  };
  
  template <typename T>
  struct reducer_traits<T, decltype(std::declval<T>().get(), void())> : public reducer_with_get_traits<T, is_future<std::result_of_t<decltype(&T::get)(T)>>::value> {};
  
  /// \endcond
  
  /// Map a function over a range and reduce the result.
  ///
  /// \param begin an \c InputIterator designating the beginning of the range
  /// \param end an \c InputIterator designating the end of the range
  /// \param mapper is a callable which transforms values from the iterator range into a future<T>
  /// \param r is an object which can be called with T as parameter and yields a future<>
  ///     It may have a get() method which returns a value of type U which holds the result of reduction.
  /// \return Th reduced value wrapped in a future.
  ///     If the reducer has no get() method then this function returns future<>.
  
  // TODO: specialize for non-deferring reducer
  template <typename Iterator, typename Mapper, typename Reducer>
  inline
  auto
  map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Reducer&& r)
      -> typename reducer_traits<Reducer>::future_type
  {
      auto r_ptr = make_lw_shared(std::forward<Reducer>(r));
      future<> ret = make_ready_future<>();
      while (begin != end) {
          ret = futurize_invoke(mapper, *begin++).then_wrapped([ret = std::move(ret), r_ptr] (auto f) mutable {
              return ret.then_wrapped([f = std::move(f), r_ptr] (auto rf) mutable {
                  if (rf.failed()) {
                      f.ignore_ready_future();
                      return std::move(rf);
                  } else {
                      return futurize_apply(*r_ptr, std::move(f.get()));
                  }
              });
          });
      }
      return reducer_traits<Reducer>::maybe_call_get(std::move(ret), r_ptr);
  }
  
  /// Asynchronous map/reduce transformation.
  ///
  /// Given a range of objects, an asynchronous unary function
  /// operating on these objects, an initial value, and a
  /// binary function for reducing, map_reduce() will
  /// transform each object in the range, then invoke
  /// the the reducing function with the result.
  ///
  /// Example:
  ///
  /// Calculate the total size of several files:
  ///
  /// \code
  ///  map_reduce(files.begin(), files.end(),
  ///             std::mem_fn(file::size),
  ///             size_t(0),
  ///             std::plus<size_t>())
  /// \endcode
  ///
  /// Requirements:
  ///    - Iterator: an InputIterator.
  ///    - Mapper: unary function taking Iterator::value_type and producing a future<...>.
  ///    - Initial: any value type
  ///    - Reduce: a binary function taking two Initial values and returning an Initial
  ///
  /// Return type:
  ///    - future<Initial>
  ///
  /// \param begin beginning of object range to operate on
  /// \param end end of object range to operate on
  /// \param mapper map function to call on each object, returning a future
  /// \param initial initial input value to reduce function
  /// \param reduce binary function for merging two result values from \c mapper
  ///
  /// \return equivalent to \c reduce(reduce(initial, mapper(obj0)), mapper(obj1)) ...
  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 {
          Initial result;
          Reduce reduce;
      };
      auto s = make_lw_shared(state{std::move(initial), std::move(reduce)});
      future<> ret = make_ready_future<>();
      while (begin != end) {
          ret = futurize_invoke(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));
      });
  }
  
  /// Asynchronous map/reduce transformation (range version).
  ///
  /// Given a range of objects, an asynchronous unary function
  /// operating on these objects, an initial value, and a
  /// binary function for reducing, map_reduce() will
  /// transform each object in the range, then invoke
  /// the the reducing function with the result.
  ///
  /// Example:
  ///
  /// Calculate the total size of several files:
  ///
  /// \code
  ///  std::vector<file> files = ...;
  ///  map_reduce(files,
  ///             std::mem_fn(file::size),
  ///             size_t(0),
  ///             std::plus<size_t>())
  /// \endcode
  ///
  /// Requirements:
  ///    - Iterator: an InputIterator.
  ///    - Mapper: unary function taking Iterator::value_type and producing a future<...>.
  ///    - Initial: any value type
  ///    - Reduce: a binary function taking two Initial values and returning an Initial
  ///
  /// Return type:
  ///    - future<Initial>
  ///
  /// \param range object range to operate on
  /// \param mapper map function to call on each object, returning a future
  /// \param initial initial input value to reduce function
  /// \param reduce binary function for merging two result values from \c mapper
  ///
  /// \return equivalent to \c reduce(reduce(initial, mapper(obj0)), mapper(obj1)) ...
  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));
  }
  
  /// Implements @Reducer concept. Calculates the result by
  /// adding elements to the accumulator.
  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
  