diff options
Diffstat (limited to '')
-rw-r--r-- | src/seastar/include/seastar/core/alien.hh | 184 |
1 files changed, 184 insertions, 0 deletions
diff --git a/src/seastar/include/seastar/core/alien.hh b/src/seastar/include/seastar/core/alien.hh new file mode 100644 index 000000000..43510d4ea --- /dev/null +++ b/src/seastar/include/seastar/core/alien.hh @@ -0,0 +1,184 @@ +// -*- mode:C++; tab-width:4; c-basic-offset:4; indent-tabs-mode:nil -*- +/* + * 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) 2018 Red Hat + */ + +#pragma once + +#include <atomic> +#include <deque> +#include <future> +#include <memory> + +#include <boost/lockfree/queue.hpp> + +#include <seastar/core/future.hh> +#include <seastar/core/cacheline.hh> +#include <seastar/core/sstring.hh> +#include <seastar/core/metrics_registration.hh> + +/// \file + +namespace seastar { + +class reactor; + +/// \brief Integration with non-seastar applications. +namespace alien { + +class message_queue { + static constexpr size_t batch_size = 128; + static constexpr size_t prefetch_cnt = 2; + struct work_item; + struct lf_queue_remote { + reactor* remote; + }; + using lf_queue_base = boost::lockfree::queue<work_item*>; + // use inheritence to control placement order + struct lf_queue : lf_queue_remote, lf_queue_base { + lf_queue(reactor* remote) + : lf_queue_remote{remote}, lf_queue_base{batch_size} {} + void maybe_wakeup(); + } _pending; + struct alignas(seastar::cache_line_size) { + std::atomic<size_t> value{0}; + } _sent; + // keep this between two structures with statistics + // this makes sure that they have at least one cache line + // between them, so hw prefetcher will not accidentally prefetch + // cache line used by another cpu. + metrics::metric_groups _metrics; + struct alignas(seastar::cache_line_size) { + size_t _received = 0; + size_t _last_rcv_batch = 0; + }; + struct work_item { + virtual ~work_item() = default; + virtual void process() = 0; + }; + template <typename Func> + struct async_work_item : work_item { + Func _func; + async_work_item(Func&& func) : _func(std::move(func)) {} + void process() override { + _func(); + } + }; + template<typename Func> + size_t process_queue(lf_queue& q, Func process); + void submit_item(std::unique_ptr<work_item> wi); +public: + message_queue(reactor *to); + void start(); + void stop(); + template <typename Func> + void submit(Func&& func) { + auto wi = std::make_unique<async_work_item<Func>>(std::forward<Func>(func)); + submit_item(std::move(wi)); + } + size_t process_incoming(); + bool pure_poll_rx() const; +}; + +class smp { + struct qs_deleter { + unsigned count; + qs_deleter(unsigned n = 0) : count(n) {} + qs_deleter(const qs_deleter& d) : count(d.count) {} + void operator()(message_queue* qs) const; + }; + using qs = std::unique_ptr<message_queue[], qs_deleter>; +public: + static qs create_qs(const std::vector<reactor*>& reactors); + static qs _qs; + static bool poll_queues(); + static bool pure_poll_queues(); +}; + +/// Runs a function on a remote shard from an alien thread where engine() is not available. +/// +/// \param shard designates the shard to run the function on +/// \param func a callable to run on shard \c t. If \c func is a temporary object, +/// its lifetime will be extended by moving it. If \c func is a reference, +/// the caller must guarantee that it will survive the call. +/// \note the func must not throw and should return \c void. as we cannot identify the +/// alien thread, hence we are not able to post the fulfilled promise to the +/// message queue managed by the shard executing the alien thread which is +/// interested to the return value. Please use \c submit_to() instead, if +/// \c func throws. +template <typename Func> +void run_on(unsigned shard, Func func) { + smp::_qs[shard].submit(std::move(func)); +} + +namespace internal { +template<typename Func> +using return_value_t = typename futurize<std::invoke_result_t<Func>>::value_type; + +template<typename Func, + bool = std::is_empty_v<return_value_t<Func>>> +struct return_type_of { + using type = void; + static void set(std::promise<void>& p, return_value_t<Func>&&) { + p.set_value(); + } +}; +template<typename Func> +struct return_type_of<Func, false> { + using return_tuple_t = typename futurize<std::invoke_result_t<Func>>::tuple_type; + using type = std::tuple_element_t<0, return_tuple_t>; + static void set(std::promise<type>& p, return_value_t<Func>&& t) { +#if SEASTAR_API_LEVEL < 5 + p.set_value(std::get<0>(std::move(t))); +#else + p.set_value(std::move(t)); +#endif + } +}; +template <typename Func> using return_type_t = typename return_type_of<Func>::type; +} + +/// Runs a function on a remote shard from an alien thread where engine() is not available. +/// +/// \param shard designates the shard to run the function on +/// \param func a callable to run on \c shard. If \c func is a temporary object, +/// its lifetime will be extended by moving it. If \c func is a reference, +/// the caller must guarantee that it will survive the call. +/// \return whatever \c func returns, as a \c std::future<> +/// \note the caller must keep the returned future alive until \c func returns +template<typename Func, typename T = internal::return_type_t<Func>> +std::future<T> submit_to(unsigned shard, Func func) { + std::promise<T> pr; + auto fut = pr.get_future(); + run_on(shard, [pr = std::move(pr), func = std::move(func)] () mutable { + // std::future returned via std::promise above. + (void)func().then_wrapped([pr = std::move(pr)] (auto&& result) mutable { + try { + internal::return_type_of<Func>::set(pr, result.get()); + } catch (...) { + pr.set_exception(std::current_exception()); + } + }); + }); + return fut; +} + +} +} |