Unverified Commit 6620e854 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #860

Move worker_hub to the core library
parents d2952af5 c81ccf0a
......@@ -11,6 +11,8 @@ set(LIBCAF_CORE_SRCS
src/abstract_composable_behavior.cpp
src/abstract_coordinator.cpp
src/abstract_group.cpp
src/abstract_worker.cpp
src/abstract_worker_hub.cpp
src/actor.cpp
src/actor_addr.cpp
src/actor_clock.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/fwd.hpp"
#include "caf/ref_counted.hpp"
#include "caf/resumable.hpp"
namespace caf {
namespace detail {
class abstract_worker : public ref_counted, public resumable {
public:
// -- friends ----------------------------------------------------------------
friend abstract_worker_hub;
// -- constructors, destructors, and assignment operators --------------------
abstract_worker();
~abstract_worker() override;
// -- implementation of resumable --------------------------------------------
subtype_t subtype() const override;
void intrusive_ptr_add_ref_impl() override;
void intrusive_ptr_release_impl() override;
private:
// -- member variables -------------------------------------------------------
/// Points to the next worker in the hub.
std::atomic<abstract_worker*> next_;
};
} // namespace detail
} // namespace caf
......@@ -23,54 +23,50 @@
#include <mutex>
#include "caf/fwd.hpp"
#include "caf/io/basp/fwd.hpp"
namespace caf {
namespace io {
namespace basp {
namespace detail {
/// A central place where BASP workers return to after finishing a task. A hub
/// A central place where workers return to after finishing a task. A hub
/// supports any number of workers that call `push`, but only a single master
/// that calls `pop`. The hub takes ownership of all workers. Workers register
/// at the hub during construction and get destroyed when the hub gets
/// destroyed.
class worker_hub {
class abstract_worker_hub {
public:
// -- member types -----------------------------------------------------------
// -- constructors, destructors, and assignment operators --------------------
using pointer = worker*;
abstract_worker_hub();
// -- constructors, destructors, and assignment operators --------------------
virtual ~abstract_worker_hub();
worker_hub();
// -- synchronization --------------------------------------------------------
~worker_hub();
/// Waits until all workers are back at the hub.
void await_workers();
// -- properties -------------------------------------------------------------
protected:
// -- worker management ------------------------------------------------------
/// Creates a new worker and adds it to the hub.
void add_new_worker(message_queue&, proxy_registry&);
/// Adds a new worker to the hub.
void push_new(abstract_worker* ptr);
/// Add a worker to the hub.
void push(pointer ptr);
/// Returns a worker to the hub.
void push_returning(abstract_worker* ptr);
/// Get a worker from the hub.
/// Tries to retrieve a worker from the hub.
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// hub is currently empty.
pointer pop();
abstract_worker* pop_impl();
/// Check which worker would `pop` currently return.
/// Checks which worker would `pop` currently return.
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// hub is currently empty.
pointer peek();
/// Waits until all workers are back at the hub.
void await_workers();
abstract_worker* peek_impl();
private:
// -- member variables -------------------------------------------------------
std::atomic<pointer> head_;
std::atomic<abstract_worker*> head_;
std::atomic<size_t> running_;
......@@ -79,6 +75,5 @@ private:
std::condition_variable cv_;
};
} // namespace basp
} // namespace io
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/detail/abstract_worker_hub.hpp"
namespace caf {
namespace detail {
template <class Worker>
class worker_hub : public abstract_worker_hub {
public:
// -- member types -----------------------------------------------------------
using super = abstract_worker_hub;
using worker_type = Worker;
// -- worker management ------------------------------------------------------
/// Creates a new worker and adds it to the hub.
template <class... Ts>
void add_new_worker(Ts&&... xs) {
super::push_new(new worker_type(*this, std::forward<Ts>(xs)...));
}
/// Returns a worker to the hub.
void push(worker_type* ptr) {
super::push_returning(ptr);
}
/// Gets a worker from the hub.
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// hub is currently empty.
worker_type* pop() {
return static_cast<worker_type*>(super::pop_impl());
}
/// Checks which worker would `pop` currently return.
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// hub is currently empty.
worker_type* peek() {
return static_cast<worker_type*>(super::peek_impl());
}
};
} // namespace detail
} // namespace caf
......@@ -250,6 +250,8 @@ namespace detail {
template <class> class type_erased_value_impl;
template <class> class stream_distribution_tree;
class abstract_worker;
class abstract_worker_hub;
class disposer;
class dynamic_message_data;
class group_manager;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/abstract_worker.hpp"
namespace caf {
namespace detail {
// -- constructors, destructors, and assignment operators ----------------------
abstract_worker::abstract_worker() : next_(nullptr) {
// nop
}
abstract_worker::~abstract_worker() {
// nop
}
// -- implementation of resumable ----------------------------------------------
resumable::subtype_t abstract_worker::subtype() const {
return resumable::function_object;
}
void abstract_worker::intrusive_ptr_add_ref_impl() {
ref();
}
void abstract_worker::intrusive_ptr_release_impl() {
deref();
}
} // namespace detail
} // namespace caf
......@@ -16,21 +16,20 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/basp/worker_hub.hpp"
#include "caf/detail/abstract_worker_hub.hpp"
#include "caf/io/basp/worker.hpp"
#include "caf/detail/abstract_worker.hpp"
namespace caf {
namespace io {
namespace basp {
namespace detail {
// -- constructors, destructors, and assignment operators ----------------------
worker_hub::worker_hub() : head_(nullptr), running_(0) {
abstract_worker_hub::abstract_worker_hub() : head_(nullptr), running_(0) {
// nop
}
worker_hub::~worker_hub() {
abstract_worker_hub::~abstract_worker_hub() {
await_workers();
auto head = head_.load();
while (head != nullptr) {
......@@ -40,10 +39,17 @@ worker_hub::~worker_hub() {
}
}
// -- properties ---------------------------------------------------------------
// -- synchronization ----------------------------------------------------------
void worker_hub::add_new_worker(message_queue& queue, proxy_registry& proxies) {
auto ptr = new worker(*this, queue, proxies);
void abstract_worker_hub::await_workers() {
std::unique_lock<std::mutex> guard{mtx_};
while (running_ != 0)
cv_.wait(guard);
}
// -- worker management --------------------------------------------------------
void abstract_worker_hub::push_new(abstract_worker* ptr) {
auto next = head_.load();
for (;;) {
ptr->next_ = next;
......@@ -52,7 +58,7 @@ void worker_hub::add_new_worker(message_queue& queue, proxy_registry& proxies) {
}
}
void worker_hub::push(pointer ptr) {
void abstract_worker_hub::push_returning(abstract_worker* ptr) {
auto next = head_.load();
for (;;) {
ptr->next_ = next;
......@@ -66,7 +72,7 @@ void worker_hub::push(pointer ptr) {
}
}
worker_hub::pointer worker_hub::pop() {
abstract_worker* abstract_worker_hub::pop_impl() {
auto result = head_.load();
if (result == nullptr)
return nullptr;
......@@ -80,16 +86,9 @@ worker_hub::pointer worker_hub::pop() {
}
}
worker_hub::pointer worker_hub::peek() {
abstract_worker* abstract_worker_hub::peek_impl() {
return head_.load();
}
void worker_hub::await_workers() {
std::unique_lock<std::mutex> guard{mtx_};
while (running_ != 0)
cv_.wait(guard);
}
} // namespace basp
} // namespace io
} // namespace detail
} // namespace caf
......@@ -47,7 +47,6 @@ set(LIBCAF_IO_SRCS
src/test_multiplexer.cpp
src/udp.cpp
src/worker.cpp
src/worker_hub.cpp
)
add_custom_target(libcaf_io)
......
......@@ -23,6 +23,7 @@
#include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/callback.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/error.hpp"
#include "caf/io/basp/buffer_type.hpp"
#include "caf/io/basp/connection_state.hpp"
......@@ -30,7 +31,7 @@
#include "caf/io/basp/message_queue.hpp"
#include "caf/io/basp/message_type.hpp"
#include "caf/io/basp/routing_table.hpp"
#include "caf/io/basp/worker_hub.hpp"
#include "caf/io/basp/worker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/variant.hpp"
......@@ -208,7 +209,7 @@ public:
return this_node_;
}
worker_hub& hub() {
detail::worker_hub<worker>& hub() {
return hub_;
}
......@@ -236,7 +237,7 @@ private:
node_id this_node_;
callee& callee_;
message_queue queue_;
worker_hub hub_;
detail::worker_hub<worker> hub_;
};
/// @}
......
......@@ -23,6 +23,8 @@
#include <vector>
#include "caf/config.hpp"
#include "caf/detail/abstract_worker.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/fwd.hpp"
#include "caf/io/basp/fwd.hpp"
#include "caf/io/basp/header.hpp"
......@@ -35,26 +37,28 @@ namespace io {
namespace basp {
/// Deserializes payloads for BASP messages asynchronously.
class worker : public resumable,
public remote_message_handler<worker>,
public ref_counted {
class worker : public detail::abstract_worker,
public remote_message_handler<worker> {
public:
// -- friends ----------------------------------------------------------------
friend worker_hub;
friend remote_message_handler<worker>;
// -- member types -----------------------------------------------------------
using atomic_pointer = std::atomic<worker*>;
using super = detail::abstract_worker;
using scheduler_type = scheduler::abstract_coordinator;
using buffer_type = std::vector<char>;
using hub_type = detail::worker_hub<worker>;
// -- constructors, destructors, and assignment operators --------------------
/// Only the ::worker_hub has access to the construtor.
worker(hub_type& hub, message_queue& queue, proxy_registry& proxies);
~worker() override;
// -- management -------------------------------------------------------------
......@@ -64,25 +68,13 @@ public:
// -- implementation of resumable --------------------------------------------
subtype_t subtype() const override;
resume_result resume(execution_unit* ctx, size_t) override;
void intrusive_ptr_add_ref_impl() override;
void intrusive_ptr_release_impl() override;
private:
// -- constructors, destructors, and assignment operators --------------------
/// Only the ::worker_hub has access to the construtor.
worker(worker_hub& hub, message_queue& queue, proxy_registry& proxies);
// -- constants and assertions -----------------------------------------------
/// Stores how many bytes the "first half" of this object requires.
static constexpr size_t pointer_members_size = sizeof(atomic_pointer)
+ sizeof(worker_hub*)
static constexpr size_t pointer_members_size = sizeof(hub_type*)
+ sizeof(message_queue*)
+ sizeof(proxy_registry*)
+ sizeof(actor_system*);
......@@ -92,11 +84,8 @@ private:
// -- member variables -------------------------------------------------------
/// Points to the next worker in the hub.
atomic_pointer next_;
/// Points to our home hub.
worker_hub* hub_;
hub_type* hub_;
/// Points to the queue for establishing strict ordering.
message_queue* queue_;
......
......@@ -20,7 +20,6 @@
#include "caf/actor_system.hpp"
#include "caf/io/basp/message_queue.hpp"
#include "caf/io/basp/worker_hub.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
......@@ -30,12 +29,8 @@ namespace basp {
// -- constructors, destructors, and assignment operators ----------------------
worker::worker(worker_hub& hub, message_queue& queue, proxy_registry& proxies)
: next_(nullptr),
hub_(&hub),
queue_(&queue),
proxies_(&proxies),
system_(&proxies.system()) {
worker::worker(hub_type& hub, message_queue& queue, proxy_registry& proxies)
: hub_(&hub), queue_(&queue), proxies_(&proxies), system_(&proxies.system()) {
CAF_IGNORE_UNUSED(pad_);
}
......@@ -60,10 +55,6 @@ void worker::launch(const node_id& last_hop, const basp::header& hdr,
// -- implementation of resumable ----------------------------------------------
resumable::subtype_t worker::subtype() const {
return resumable::function_object;
}
resumable::resume_result worker::resume(execution_unit* ctx, size_t) {
ctx->proxy_registry_ptr(proxies_);
handle_remote_message(ctx);
......@@ -71,14 +62,6 @@ resumable::resume_result worker::resume(execution_unit* ctx, size_t) {
return resumable::awaiting_message;
}
void worker::intrusive_ptr_add_ref_impl() {
ref();
}
void worker::intrusive_ptr_release_impl() {
deref();
}
} // namespace basp
} // namespace io
} // namespace caf
......@@ -27,7 +27,6 @@
#include "caf/actor_system.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/io/basp/message_queue.hpp"
#include "caf/io/basp/worker_hub.hpp"
#include "caf/make_actor.hpp"
#include "caf/proxy_registry.hpp"
......@@ -76,7 +75,7 @@ private:
};
struct fixture : test_coordinator_fixture<> {
io::basp::worker_hub hub;
detail::worker_hub<io::basp::worker> hub;
io::basp::message_queue queue;
mock_proxy_registry_backend proxies_backend;
proxy_registry proxies;
......
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