Commit e9ce980c authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

WIP

parent f5fdb730
...@@ -20,20 +20,59 @@ ...@@ -20,20 +20,59 @@
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/policy/categorized.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
#include "caf/resumable.hpp"
namespace caf { namespace caf {
/// Implements a simple proxy forwarding all operations to a manager. /// Implements a simple proxy forwarding all operations to a manager.
class forwarding_actor_proxy : public actor_proxy { class forwarding_actor_proxy : public actor_proxy, public resumable {
public: public:
using forwarding_stack = std::vector<strong_actor_ptr>; // -- member types -----------------------------------------------------------
/// Stores asynchronous messages with default priority.
using normal_queue = intrusive::drr_queue<policy::normal_messages>;
/// Stores asynchronous messages with hifh priority.
using urgent_queue = intrusive::drr_queue<policy::urgent_messages>;
/// Configures the FIFO inbox with two nested queues:
///
/// 1. Default asynchronous messages
/// 2. High-priority asynchronous messages
struct mailbox_policy {
using deficit_type = size_t;
using mapped_type = mailbox_element;
using unique_pointer = mailbox_element_ptr;
using queue_type = intrusive::wdrr_fixed_multiplexed_queue<
policy::categorized, normal_queue, urgent_queue>;
static constexpr size_t normal_queue_index = 0;
static constexpr size_t urgent_queue_index = 1;
};
/// A queue optimized for single-reader-many-writers.
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
// -- constructors and destructors -------------------------------------------
forwarding_actor_proxy(actor_config& cfg, actor dest); forwarding_actor_proxy(actor_config& cfg, actor dest);
~forwarding_actor_proxy() override; ~forwarding_actor_proxy() override;
// -- overridden member functions --------------------------------------------
void enqueue(mailbox_element_ptr what, execution_unit* context) override; void enqueue(mailbox_element_ptr what, execution_unit* context) override;
bool add_backlink(abstract_actor* x) override; bool add_backlink(abstract_actor* x) override;
...@@ -42,9 +81,17 @@ public: ...@@ -42,9 +81,17 @@ public:
void kill_proxy(execution_unit* ctx, error rsn) override; void kill_proxy(execution_unit* ctx, error rsn) override;
resume_result resume(execution_unit*, size_t max_throughput) override;
void intrusive_ptr_add_ref_impl() override;
void intrusive_ptr_release_impl() override;
private: private:
void forward_msg(strong_actor_ptr sender, message_id mid, message msg, // -- member variables -------------------------------------------------------
const forwarding_stack* fwd = nullptr);
// used by both event-based and blocking actors
mailbox_type mailbox_;
mutable detail::shared_spinlock broker_mtx_; mutable detail::shared_spinlock broker_mtx_;
actor broker_; actor broker_;
......
...@@ -80,7 +80,6 @@ bool forwarding_actor_proxy::remove_backlink(abstract_actor* x) { ...@@ -80,7 +80,6 @@ bool forwarding_actor_proxy::remove_backlink(abstract_actor* x) {
} }
return false; return false;
} }
void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, error rsn) { void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, error rsn) {
CAF_ASSERT(ctx != nullptr); CAF_ASSERT(ctx != nullptr);
actor tmp; actor tmp;
...@@ -91,4 +90,34 @@ void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, error rsn) { ...@@ -91,4 +90,34 @@ void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, error rsn) {
cleanup(std::move(rsn), ctx); cleanup(std::move(rsn), ctx);
} }
resume_result forwarding_actor_proxy::resume(execution_unit* ctx,
size_t max_throughput) {
size_t handled_msgs = 0;
auto f = [](mailbox_element& x) {
std::vector<char> buf;
binary_serializer sink{ctx, buf};
if (auto err = sink(x.stages, x.content())) {
CAF_LOG_ERROR("unable to serialize message:" << CAF_ARG(x));
} else {
send(broker_, std::move(x.sender), x.mid, std::move(buf));
}
++handled_msgs;
};
while (handled_msgs < max_throughput) {
if (!mailbox_.new_round(3, f).consumed_items && mailbox().try_block())
return resumable::awaiting_message;
}
if (mailbox().try_block())
return resumable::awaiting_message;
return resumable::resume_later;
}
void forwarding_actor_proxy::intrusive_ptr_add_ref_impl() {
ref();
}
void forwarding_actor_proxy::intrusive_ptr_release_impl() {
deref();
}
} // 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