Commit 063b630a authored by Joseph Noir's avatar Joseph Noir

Add make function that accepts a list of policies

parent 07655b4e
...@@ -10,6 +10,7 @@ ...@@ -10,6 +10,7 @@
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/default_multiplexer.hpp"
using namespace caf; using namespace caf;
using namespace caf::io::network;
namespace { namespace {
...@@ -28,46 +29,31 @@ public: ...@@ -28,46 +29,31 @@ public:
}; };
void caf_main(actor_system& system, const config&) { void caf_main(actor_system& system, const config&) {
io::network::default_multiplexer mpx{&system}; default_multiplexer mpx{&system};
// std::thread t([&]() { // Setup thread to run the multiplexer.
// std::cout << "starting multiplexer" << std::endl;
// mpx.run();
// });
auto backend_supervisor = mpx.make_supervisor();
std::thread t; std::thread t;
// The only backend that returns a `nullptr` by default is the std::atomic<bool> init_done{false};
// `test_multiplexer` which does not have its own thread but uses the main std::mutex mtx;
// thread instead. Other backends can set `middleman_detach_multiplexer` to std::condition_variable cv;
// false to suppress creation of the supervisor. t = std::thread{[&] {
if (backend_supervisor != nullptr) { system.thread_started();
std::atomic<bool> init_done{false}; std::cout << "starting multiplexer" << std::endl;
std::mutex mtx; {
std::condition_variable cv; std::unique_lock<std::mutex> guard{mtx};
t = std::thread{[&] { mpx.thread_id(std::this_thread::get_id());
system.thread_started(); init_done = true;
std::cout << "starting multiplexer" << std::endl; cv.notify_one();
{ }
std::unique_lock<std::mutex> guard{mtx}; mpx.run();
mpx.thread_id(std::this_thread::get_id()); system.thread_terminates();
init_done = true; }};
cv.notify_one(); std::unique_lock<std::mutex> guard{mtx};
} while (init_done == false)
mpx.run(); cv.wait(guard);
system.thread_terminates(); // Create an event handling actor to run in the multiplexer.
}};
std::unique_lock<std::mutex> guard{mtx};
while (init_done == false)
cv.wait(guard);
}
actor_config cfg{&mpx}; actor_config cfg{&mpx};
// io::network::newb n{cfg, mpx, -1}; auto n = make_newb<detail::protocol_policy,
// n.eq_impl(make_message_id(message_priority::normal), nullptr, nullptr, 1); detail::generic_policy>(system, cfg, mpx, -1);
// auto n = make_actor<io::network::newb>(system.next_actor_id(), system.node(),+
// &system, cfg, mpx, -1);
auto res = system.spawn_impl<io::network::newb, hidden>(cfg, mpx, -1);
// auto m = caf::io::make_middleman_actor(system, res);
auto n = actor_cast<actor>(res);
anon_send(n, 1); anon_send(n, 1);
t.join(); t.join();
} }
......
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#pragma once #pragma once
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
...@@ -32,39 +33,36 @@ namespace caf { ...@@ -32,39 +33,36 @@ namespace caf {
namespace io { namespace io {
namespace network { namespace network {
template <class T, class... Xs>
class newb; class newb;
} // namespace network } // namespace network
} // namespace io } // namespace io
template <> template <class T, class... Xs>
class behavior_type_of<io::network::newb> { class behavior_type_of<io::network::newb<T, Xs...>> {
public: public:
using type = behavior; using type = behavior;
}; };
namespace io {
namespace network {
/** namespace detail {
* TODO:
* - [ ] create a class `newb` that is an event handler and actor
* - [ ] get it running in the multiplexer
* - [ ] create a base policy class
* - [ ] is there a difference between a protocol policy and a guarantees?
* - [ ] build `make_policy` which creates a `newb` with multiple policies
* - [ ] get a call hierarchy in both directions
* - [ ] what should policy their constrcutors do?
* - [ ]
*/
/*
class policy {
struct generic_policy {
generic_policy(int i)
: idx{i} {
// nop
}
int idx;
}; };
class protocol { struct protocol_policy : public generic_policy {
protocol_policy(int i)
: generic_policy{i} {
// nop
}
// define a base protocol // define a base protocol
// - create a socket
// - sending // - sending
// * enqueue // * enqueue
// * // *
...@@ -72,30 +70,54 @@ class protocol { ...@@ -72,30 +70,54 @@ class protocol {
// - fork // - fork
}; };
class mutation { struct mutating_policy : public generic_policy {
mutating_policy(int i)
: generic_policy{i} {
// nop
}
// adjust a protocol // adjust a protocol
}; };
*/
//template <class Protocol, class... Policies> template <class T>
class newb : public extend<scheduled_actor, newb>:: struct is_network_policy_type : std::is_base_of<generic_policy, T> {};
} // namespace detail
namespace io {
namespace network {
/**
* TODO:
* - [x] create a class `newb` that is an event handler and actor
* - [x] get it running in the multiplexer
* - [ ] create a base policy class
* - [x] build `make_policy` which creates a `newb` with multiple policies
* - [ ] is there a difference between a protocol policy and a guarantee?
* - [ ] get a call hierarchy in both directions
* - [ ] what should policy their constrcutors do?
* - [ ] ...
*/
template <class Protocol, class... Policies>
class newb : public extend<scheduled_actor,
newb<Protocol, Policies...>>::template
with<mixin::sender, mixin::requester, with<mixin::sender, mixin::requester,
mixin::behavior_changer>, mixin::behavior_changer>,
public dynamically_typed_actor_base, public dynamically_typed_actor_base,
public event_handler { public event_handler {
public: public:
using super = extend<scheduled_actor, newb>:: using super = typename extend<scheduled_actor, newb<Protocol, Policies...>>::
with<mixin::sender, mixin::requester, mixin::behavior_changer>; template with<mixin::sender, mixin::requester, mixin::behavior_changer>;
using signatures = none_t; using signatures = none_t;
// using base_protocol = Protocol;
// -- constructors and destructors ------------------------------------------- // -- constructors and destructors -------------------------------------------
newb(actor_config& cfg, default_multiplexer& dm, native_socket sockfd) newb(actor_config& cfg, default_multiplexer& dm, native_socket sockfd)
: super(cfg), : super(cfg),
event_handler(dm, sockfd) { event_handler(dm, sockfd),
protocol_{0},
policies_{Policies{1}...} {
// nop // nop
} }
...@@ -121,10 +143,10 @@ public: ...@@ -121,10 +143,10 @@ public:
CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide)); CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide));
// add implicit reference count held by middleman/multiplexer // add implicit reference count held by middleman/multiplexer
if (!hide) if (!hide)
register_at_system(); super::register_at_system();
if (lazy && mailbox().try_block()) if (lazy && super::mailbox().try_block())
return; return;
intrusive_ptr_add_ref(ctrl()); intrusive_ptr_add_ref(super::ctrl());
eu->exec_later(this); eu->exec_later(this);
} }
...@@ -137,7 +159,7 @@ public: ...@@ -137,7 +159,7 @@ public:
if (bhvr) { if (bhvr) {
// make_behavior() did return a behavior instead of using become() // make_behavior() did return a behavior instead of using become()
CAF_LOG_DEBUG("make_behavior() did return a valid behavior"); CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
become(std::move(bhvr)); this->become(std::move(bhvr));
} }
} }
...@@ -151,7 +173,8 @@ public: ...@@ -151,7 +173,8 @@ public:
// -- overridden modifiers of resumable -------------------------------------- // -- overridden modifiers of resumable --------------------------------------
resume_result resume(execution_unit* ctx, size_t mt) override { multiplexer::runnable::resume_result resume(execution_unit* ctx,
size_t mt) override {
CAF_ASSERT(ctx != nullptr); CAF_ASSERT(ctx != nullptr);
CAF_ASSERT(ctx == &backend()); CAF_ASSERT(ctx == &backend());
return scheduled_actor::resume(ctx, mt); return scheduled_actor::resume(ctx, mt);
...@@ -164,14 +187,15 @@ public: ...@@ -164,14 +187,15 @@ public:
} }
void removed_from_loop(operation op) override { void removed_from_loop(operation op) override {
std::cout << "removing myself from the loop!" << std::endl; std::cout << "removing myself from the loop for "
<< to_string(op) << std::endl;
} }
// -- members ---------------------------------------------------------------- // -- members ----------------------------------------------------------------
/// Returns the `multiplexer` running this broker. /// Returns the `multiplexer` running this broker.
network::multiplexer& backend() { network::multiplexer& backend() {
return system().middleman().backend(); return super::system().middleman().backend();
} }
behavior make_behavior() { behavior make_behavior() {
...@@ -185,7 +209,7 @@ public: ...@@ -185,7 +209,7 @@ public:
void init_newb() { void init_newb() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
setf(is_initialized_flag); super::setf(super::is_initialized_flag);
} }
/// @cond PRIVATE /// @cond PRIVATE
...@@ -201,9 +225,23 @@ public: ...@@ -201,9 +225,23 @@ public:
/// @endcond /// @endcond
private: private:
// std::tuple<Policies...> policies_; Protocol protocol_;
std::tuple<Policies...> policies_;
}; };
template <class Protocol, class... Policies>
actor make_newb(actor_system& sys, actor_config& cfg, default_multiplexer& mpx,
native_socket sockfd) {
using policy_types = detail::type_list<Protocol, Policies...>;
static_assert(detail::tl_forall<policy_types, detail::is_network_policy_type>::value,
"Only network policies allowed as template parameters");
static_assert(std::is_base_of<detail::protocol_policy, Protocol>::value,
"First template argument must be a protocol policy");
using newb_t = newb<Protocol, Policies...>;
auto res = sys.spawn_impl<newb_t, hidden>(cfg, mpx, sockfd);
return actor_cast<actor>(res);
}
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment