| // 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 __PROCESS_DISPATCH_HPP__ |
| #define __PROCESS_DISPATCH_HPP__ |
| |
| #include <functional> |
| #include <memory> |
| #include <string> |
| |
| #include <process/process.hpp> |
| |
| #include <stout/lambda.hpp> |
| #include <stout/preprocessor.hpp> |
| #include <stout/result_of.hpp> |
| |
| namespace process { |
| |
| // The dispatch mechanism enables you to "schedule" a method to get |
| // invoked on a process. The result of that method invocation is |
| // accessible via the future that is returned by the dispatch method |
| // (note, however, that it might not be the _same_ future as the one |
| // returned from the method, if the method even returns a future, see |
| // below). Assuming some class 'Fibonacci' has a (visible) method |
| // named 'compute' that takes an integer, N (and returns the Nth |
| // fibonacci number) you might use dispatch like so: |
| // |
| // PID<Fibonacci> pid = spawn(new Fibonacci(), true); // Use the GC. |
| // Future<int> f = dispatch(pid, &Fibonacci::compute, 10); |
| // |
| // Because the pid argument is "typed" we can ensure that methods are |
| // only invoked on processes that are actually of that type. Providing |
| // this mechanism for varying numbers of function types and arguments |
| // requires support for variadic templates, slated to be released in |
| // C++11. Until then, we use the Boost preprocessor macros to |
| // accomplish the same thing (albeit less cleanly). See below for |
| // those definitions. |
| // |
| // Dispatching is done via a level of indirection. The dispatch |
| // routine itself creates a promise that is passed as an argument to a |
| // partially applied 'dispatcher' function (defined below). The |
| // dispatcher routines get passed to the actual process via an |
| // internal routine called, not surprisingly, 'dispatch', defined |
| // below: |
| |
| namespace internal { |
| |
| // The internal dispatch routine schedules a function to get invoked |
| // within the context of the process associated with the specified pid |
| // (first argument), unless that process is no longer valid. Note that |
| // this routine does not expect anything in particular about the |
| // specified function (second argument). The semantics are simple: the |
| // function gets applied/invoked with the process as its first |
| // argument. |
| void dispatch( |
| const UPID& pid, |
| std::unique_ptr<lambda::CallableOnce<void(ProcessBase*)>> f, |
| const Option<const std::type_info*>& functionType = None()); |
| |
| |
| // NOTE: This struct is used by the public `dispatch(const UPID& pid, F&& f)` |
| // function. See comments there for the reason why we need this. |
| template <typename R> |
| struct Dispatch; |
| |
| |
| // Partial specialization for callable objects returning `void` to be dispatched |
| // on a process. |
| // NOTE: This struct is used by the public `dispatch(const UPID& pid, F&& f)` |
| // function. See comments there for the reason why we need this. |
| template <> |
| struct Dispatch<void> |
| { |
| template <typename F> |
| void operator()(const UPID& pid, F&& f) |
| { |
| std::unique_ptr<lambda::CallableOnce<void(ProcessBase*)>> f_( |
| new lambda::CallableOnce<void(ProcessBase*)>( |
| lambda::partial( |
| [](typename std::decay<F>::type&& f, ProcessBase*) { |
| std::move(f)(); |
| }, |
| std::forward<F>(f), |
| lambda::_1))); |
| |
| internal::dispatch(pid, std::move(f_)); |
| } |
| }; |
| |
| |
| // Partial specialization for callable objects returning `Future<R>` to be |
| // dispatched on a process. |
| // NOTE: This struct is used by the public `dispatch(const UPID& pid, F&& f)` |
| // function. See comments there for the reason why we need this. |
| template <typename R> |
| struct Dispatch<Future<R>> |
| { |
| template <typename F> |
| Future<R> operator()(const UPID& pid, F&& f) |
| { |
| std::unique_ptr<Promise<R>> promise(new Promise<R>()); |
| Future<R> future = promise->future(); |
| |
| std::unique_ptr<lambda::CallableOnce<void(ProcessBase*)>> f_( |
| new lambda::CallableOnce<void(ProcessBase*)>( |
| lambda::partial( |
| [](std::unique_ptr<Promise<R>> promise, |
| typename std::decay<F>::type&& f, |
| ProcessBase*) { |
| promise->associate(std::move(f)()); |
| }, |
| std::move(promise), |
| std::forward<F>(f), |
| lambda::_1))); |
| |
| internal::dispatch(pid, std::move(f_)); |
| |
| return future; |
| } |
| }; |
| |
| |
| // Dispatches a callable object returning `R` on a process. |
| // NOTE: This struct is used by the public `dispatch(const UPID& pid, F&& f)` |
| // function. See comments there for the reason why we need this. |
| template <typename R> |
| struct Dispatch |
| { |
| template <typename F> |
| Future<R> operator()(const UPID& pid, F&& f) |
| { |
| std::unique_ptr<Promise<R>> promise(new Promise<R>()); |
| Future<R> future = promise->future(); |
| |
| std::unique_ptr<lambda::CallableOnce<void(ProcessBase*)>> f_( |
| new lambda::CallableOnce<void(ProcessBase*)>( |
| lambda::partial( |
| [](std::unique_ptr<Promise<R>> promise, |
| typename std::decay<F>::type&& f, |
| ProcessBase*) { |
| promise->set(std::move(f)()); |
| }, |
| std::move(promise), |
| std::forward<F>(f), |
| lambda::_1))); |
| |
| internal::dispatch(pid, std::move(f_)); |
| |
| return future; |
| } |
| }; |
| |
| } // namespace internal { |
| |
| |
| // Okay, now for the definition of the dispatch routines |
| // themselves. For each routine we provide the version in C++11 using |
| // variadic templates so the reader can see what the Boost |
| // preprocessor macros are effectively providing. Using C++11 closures |
| // would shorten these definitions even more. |
| // |
| // First, definitions of dispatch for methods returning void: |
| |
| template <typename T> |
| void dispatch(const PID<T>& pid, void (T::*method)()) |
| { |
| std::unique_ptr<lambda::CallableOnce<void(ProcessBase*)>> f( |
| new lambda::CallableOnce<void(ProcessBase*)>( |
| [=](ProcessBase* process) { |
| assert(process != nullptr); |
| T* t = dynamic_cast<T*>(process); |
| assert(t != nullptr); |
| (t->*method)(); |
| })); |
| |
| internal::dispatch(pid, std::move(f), &typeid(method)); |
| } |
| |
| template <typename T> |
| void dispatch(const Process<T>& process, void (T::*method)()) |
| { |
| dispatch(process.self(), method); |
| } |
| |
| template <typename T> |
| void dispatch(const Process<T>* process, void (T::*method)()) |
| { |
| dispatch(process->self(), method); |
| } |
| |
| // Due to a bug (http://gcc.gnu.org/bugzilla/show_bug.cgi?id=41933) |
| // with variadic templates and lambdas, we still need to do |
| // preprocessor expansions. |
| |
| // The following assumes base names for type and variable are `A` and `a`. |
| #define FORWARD(Z, N, DATA) std::forward<A ## N>(a ## N) |
| #define MOVE(Z, N, DATA) std::move(a ## N) |
| #define DECL(Z, N, DATA) typename std::decay<A ## N>::type&& a ## N |
| |
| #define TEMPLATE(Z, N, DATA) \ |
| template <typename T, \ |
| ENUM_PARAMS(N, typename P), \ |
| ENUM_PARAMS(N, typename A)> \ |
| void dispatch( \ |
| const PID<T>& pid, \ |
| void (T::*method)(ENUM_PARAMS(N, P)), \ |
| ENUM_BINARY_PARAMS(N, A, &&a)) \ |
| { \ |
| std::unique_ptr<lambda::CallableOnce<void(ProcessBase*)>> f( \ |
| new lambda::CallableOnce<void(ProcessBase*)>( \ |
| lambda::partial( \ |
| [method](ENUM(N, DECL, _), ProcessBase* process) { \ |
| assert(process != nullptr); \ |
| T* t = dynamic_cast<T*>(process); \ |
| assert(t != nullptr); \ |
| (t->*method)(ENUM(N, MOVE, _)); \ |
| }, \ |
| ENUM(N, FORWARD, _), \ |
| lambda::_1))); \ |
| \ |
| internal::dispatch(pid, std::move(f), &typeid(method)); \ |
| } \ |
| \ |
| template <typename T, \ |
| ENUM_PARAMS(N, typename P), \ |
| ENUM_PARAMS(N, typename A)> \ |
| void dispatch( \ |
| const Process<T>& process, \ |
| void (T::*method)(ENUM_PARAMS(N, P)), \ |
| ENUM_BINARY_PARAMS(N, A, &&a)) \ |
| { \ |
| dispatch(process.self(), method, ENUM(N, FORWARD, _)); \ |
| } \ |
| \ |
| template <typename T, \ |
| ENUM_PARAMS(N, typename P), \ |
| ENUM_PARAMS(N, typename A)> \ |
| void dispatch( \ |
| const Process<T>* process, \ |
| void (T::*method)(ENUM_PARAMS(N, P)), \ |
| ENUM_BINARY_PARAMS(N, A, &&a)) \ |
| { \ |
| dispatch(process->self(), method, ENUM(N, FORWARD, _)); \ |
| } |
| |
| REPEAT_FROM_TO(1, 13, TEMPLATE, _) // Args A0 -> A11. |
| #undef TEMPLATE |
| |
| |
| // Next, definitions of methods returning a future: |
| |
| template <typename R, typename T> |
| Future<R> dispatch(const PID<T>& pid, Future<R> (T::*method)()) |
| { |
| std::unique_ptr<Promise<R>> promise(new Promise<R>()); |
| Future<R> future = promise->future(); |
| |
| std::unique_ptr<lambda::CallableOnce<void(ProcessBase*)>> f( |
| new lambda::CallableOnce<void(ProcessBase*)>( |
| lambda::partial( |
| [=](std::unique_ptr<Promise<R>> promise, ProcessBase* process) { |
| assert(process != nullptr); |
| T* t = dynamic_cast<T*>(process); |
| assert(t != nullptr); |
| promise->associate((t->*method)()); |
| }, |
| std::move(promise), |
| lambda::_1))); |
| |
| internal::dispatch(pid, std::move(f), &typeid(method)); |
| |
| return future; |
| } |
| |
| template <typename R, typename T> |
| Future<R> dispatch(const Process<T>& process, Future<R> (T::*method)()) |
| { |
| return dispatch(process.self(), method); |
| } |
| |
| template <typename R, typename T> |
| Future<R> dispatch(const Process<T>* process, Future<R> (T::*method)()) |
| { |
| return dispatch(process->self(), method); |
| } |
| |
| #define TEMPLATE(Z, N, DATA) \ |
| template <typename R, \ |
| typename T, \ |
| ENUM_PARAMS(N, typename P), \ |
| ENUM_PARAMS(N, typename A)> \ |
| Future<R> dispatch( \ |
| const PID<T>& pid, \ |
| Future<R> (T::*method)(ENUM_PARAMS(N, P)), \ |
| ENUM_BINARY_PARAMS(N, A, &&a)) \ |
| { \ |
| std::unique_ptr<Promise<R>> promise(new Promise<R>()); \ |
| Future<R> future = promise->future(); \ |
| \ |
| std::unique_ptr<lambda::CallableOnce<void(ProcessBase*)>> f( \ |
| new lambda::CallableOnce<void(ProcessBase*)>( \ |
| lambda::partial( \ |
| [method](std::unique_ptr<Promise<R>> promise, \ |
| ENUM(N, DECL, _), \ |
| ProcessBase* process) { \ |
| assert(process != nullptr); \ |
| T* t = dynamic_cast<T*>(process); \ |
| assert(t != nullptr); \ |
| promise->associate( \ |
| (t->*method)(ENUM(N, MOVE, _))); \ |
| }, \ |
| std::move(promise), \ |
| ENUM(N, FORWARD, _), \ |
| lambda::_1))); \ |
| \ |
| internal::dispatch(pid, std::move(f), &typeid(method)); \ |
| \ |
| return future; \ |
| } \ |
| \ |
| template <typename R, \ |
| typename T, \ |
| ENUM_PARAMS(N, typename P), \ |
| ENUM_PARAMS(N, typename A)> \ |
| Future<R> dispatch( \ |
| const Process<T>& process, \ |
| Future<R> (T::*method)(ENUM_PARAMS(N, P)), \ |
| ENUM_BINARY_PARAMS(N, A, &&a)) \ |
| { \ |
| return dispatch(process.self(), method, ENUM(N, FORWARD, _)); \ |
| } \ |
| \ |
| template <typename R, \ |
| typename T, \ |
| ENUM_PARAMS(N, typename P), \ |
| ENUM_PARAMS(N, typename A)> \ |
| Future<R> dispatch( \ |
| const Process<T>* process, \ |
| Future<R> (T::*method)(ENUM_PARAMS(N, P)), \ |
| ENUM_BINARY_PARAMS(N, A, &&a)) \ |
| { \ |
| return dispatch(process->self(), method, ENUM(N, FORWARD, _)); \ |
| } |
| |
| REPEAT_FROM_TO(1, 13, TEMPLATE, _) // Args A0 -> A11. |
| #undef TEMPLATE |
| |
| |
| // Next, definitions of methods returning a value. |
| |
| template <typename R, typename T> |
| Future<R> dispatch(const PID<T>& pid, R (T::*method)()) |
| { |
| std::unique_ptr<Promise<R>> promise(new Promise<R>()); |
| Future<R> future = promise->future(); |
| |
| std::unique_ptr<lambda::CallableOnce<void(ProcessBase*)>> f( |
| new lambda::CallableOnce<void(ProcessBase*)>( |
| lambda::partial( |
| [=](std::unique_ptr<Promise<R>> promise, ProcessBase* process) { |
| assert(process != nullptr); |
| T* t = dynamic_cast<T*>(process); |
| assert(t != nullptr); |
| promise->set((t->*method)()); |
| }, |
| std::move(promise), |
| lambda::_1))); |
| |
| internal::dispatch(pid, std::move(f), &typeid(method)); |
| |
| return future; |
| } |
| |
| template <typename R, typename T> |
| Future<R> dispatch(const Process<T>& process, R (T::*method)()) |
| { |
| return dispatch(process.self(), method); |
| } |
| |
| template <typename R, typename T> |
| Future<R> dispatch(const Process<T>* process, R (T::*method)()) |
| { |
| return dispatch(process->self(), method); |
| } |
| |
| #define TEMPLATE(Z, N, DATA) \ |
| template <typename R, \ |
| typename T, \ |
| ENUM_PARAMS(N, typename P), \ |
| ENUM_PARAMS(N, typename A)> \ |
| Future<R> dispatch( \ |
| const PID<T>& pid, \ |
| R (T::*method)(ENUM_PARAMS(N, P)), \ |
| ENUM_BINARY_PARAMS(N, A, &&a)) \ |
| { \ |
| std::unique_ptr<Promise<R>> promise(new Promise<R>()); \ |
| Future<R> future = promise->future(); \ |
| \ |
| std::unique_ptr<lambda::CallableOnce<void(ProcessBase*)>> f( \ |
| new lambda::CallableOnce<void(ProcessBase*)>( \ |
| lambda::partial( \ |
| [method](std::unique_ptr<Promise<R>> promise, \ |
| ENUM(N, DECL, _), \ |
| ProcessBase* process) { \ |
| assert(process != nullptr); \ |
| T* t = dynamic_cast<T*>(process); \ |
| assert(t != nullptr); \ |
| promise->set((t->*method)(ENUM(N, MOVE, _))); \ |
| }, \ |
| std::move(promise), \ |
| ENUM(N, FORWARD, _), \ |
| lambda::_1))); \ |
| \ |
| internal::dispatch(pid, std::move(f), &typeid(method)); \ |
| \ |
| return future; \ |
| } \ |
| \ |
| template <typename R, \ |
| typename T, \ |
| ENUM_PARAMS(N, typename P), \ |
| ENUM_PARAMS(N, typename A)> \ |
| Future<R> dispatch( \ |
| const Process<T>& process, \ |
| R (T::*method)(ENUM_PARAMS(N, P)), \ |
| ENUM_BINARY_PARAMS(N, A, &&a)) \ |
| { \ |
| return dispatch(process.self(), method, ENUM(N, FORWARD, _)); \ |
| } \ |
| \ |
| template <typename R, \ |
| typename T, \ |
| ENUM_PARAMS(N, typename P), \ |
| ENUM_PARAMS(N, typename A)> \ |
| Future<R> dispatch( \ |
| const Process<T>* process, \ |
| R (T::*method)(ENUM_PARAMS(N, P)), \ |
| ENUM_BINARY_PARAMS(N, A, &&a)) \ |
| { \ |
| return dispatch(process->self(), method, ENUM(N, FORWARD, _)); \ |
| } |
| |
| REPEAT_FROM_TO(1, 13, TEMPLATE, _) // Args A0 -> A11. |
| #undef TEMPLATE |
| |
| #undef DECL |
| #undef MOVE |
| #undef FORWARD |
| |
| // We use partial specialization of |
| // - internal::Dispatch<void> vs |
| // - internal::Dispatch<Future<R>> vs |
| // - internal::Dispatch |
| // in order to determine whether R is void, Future or other types. |
| template <typename F, typename R = typename result_of<F()>::type> |
| auto dispatch(const UPID& pid, F&& f) |
| -> decltype(internal::Dispatch<R>()(pid, std::forward<F>(f))) |
| { |
| return internal::Dispatch<R>()(pid, std::forward<F>(f)); |
| } |
| |
| } // namespace process { |
| |
| #endif // __PROCESS_DISPATCH_HPP__ |
