Commit 51840609 authored by Dominik Charousset's avatar Dominik Charousset

Store max_throughput parameter in MM

parent e11edafb
...@@ -594,8 +594,6 @@ public: ...@@ -594,8 +594,6 @@ public:
behavior& fun, behavior& fun,
message_id awaited_response); message_id awaited_response);
//invoke_message_result invoke_message(mailbox_element_ptr& node);
using pending_response = std::pair<message_id, behavior>; using pending_response = std::pair<message_id, behavior>;
message_id new_request_id(message_priority mp); message_id new_request_id(message_priority mp);
......
...@@ -485,17 +485,6 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -485,17 +485,6 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
CAF_CRITICAL("invalid message type"); CAF_CRITICAL("invalid message type");
} }
/*
invoke_message_result local_actor::invoke_message(mailbox_element_ptr& node) {
if (! awaits_response() && bhvr_stack().empty())
return im_dropped;
auto& bhvr = awaits_response() ? awaited_response_handler()
: bhvr_stack().back();
auto mid = awaited_response_id();
return invoke_message(node, bhvr, mid);
}
*/
struct pending_response_predicate { struct pending_response_predicate {
public: public:
explicit pending_response_predicate(message_id mid) : mid_(mid) { explicit pending_response_predicate(message_id mid) : mid_(mid) {
......
...@@ -137,6 +137,10 @@ public: ...@@ -137,6 +137,10 @@ public:
middleman(const backend_factory&); middleman(const backend_factory&);
inline size_t max_throughput() const {
return max_throughput_;
}
/// @endcond /// @endcond
private: private:
...@@ -154,6 +158,8 @@ private: ...@@ -154,6 +158,8 @@ private:
hook_uptr hooks_; hook_uptr hooks_;
// actor offering asyncronous IO by managing this singleton instance // actor offering asyncronous IO by managing this singleton instance
middleman_actor manager_; middleman_actor manager_;
// stores the max_throughput parameter from the scheduler coordinator
size_t max_throughput_;
}; };
} // namespace io } // namespace io
......
...@@ -21,8 +21,6 @@ ...@@ -21,8 +21,6 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
...@@ -47,11 +45,10 @@ public: ...@@ -47,11 +45,10 @@ public:
inline void operator()() { inline void operator()() {
CAF_PUSH_AID(self_->id()); CAF_PUSH_AID(self_->id());
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto sc = detail::singletons::get_scheduling_coordinator(); auto mt = self_->parent().max_throughput();
if (self_->resume(nullptr, sc->max_throughput()) == resume_later) { // re-schedule broker if it reached its maximum message throughput
// re-schedule broker if (self_->resume(nullptr, mt) == resumable::resume_later)
self_->backend().post(std::move(*this)); self_->backend().post(std::move(*this));
}
} }
private: private:
......
...@@ -39,9 +39,10 @@ ...@@ -39,9 +39,10 @@
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp" #include "caf/io/basp_broker.hpp"
#include "caf/io/system_messages.hpp" #include "caf/io/system_messages.hpp"
#include "caf/io/network/interfaces.hpp" #include "caf/io/network/interfaces.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
#include "caf/detail/ripemd_160.hpp" #include "caf/detail/ripemd_160.hpp"
#include "caf/detail/safe_equal.hpp" #include "caf/detail/safe_equal.hpp"
...@@ -248,6 +249,8 @@ void middleman::add_broker(broker_ptr bptr) { ...@@ -248,6 +249,8 @@ void middleman::add_broker(broker_ptr bptr) {
void middleman::initialize() { void middleman::initialize() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto sc = detail::singletons::get_scheduling_coordinator();
max_throughput_ = sc->max_throughput();
backend_supervisor_ = backend_->make_supervisor(); backend_supervisor_ = backend_->make_supervisor();
if (backend_supervisor_ == nullptr) { if (backend_supervisor_ == nullptr) {
// the only backend that returns a `nullptr` is the `test_multiplexer` // the only backend that returns a `nullptr` is the `test_multiplexer`
......
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