Commit 6f1fb436 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Add BASP workers for parallel deserialization

parent 9ecfd120
......@@ -47,6 +47,8 @@ set(LIBCAF_IO_SRCS
src/tcp.cpp
src/test_multiplexer.cpp
src/udp.cpp
src/worker.cpp
src/worker_hub.cpp
)
add_custom_target(libcaf_io)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
namespace caf {
namespace io {
namespace basp {
struct header;
class worker;
class worker_hub;
class message_queue;
class instance;
class routing_table;
} // namespace basp
} // namespace io
} // 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 <atomic>
#include <cstdint>
#include <vector>
#include "caf/config.hpp"
#include "caf/fwd.hpp"
#include "caf/io/basp/fwd.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/io/basp/remote_message_handler.hpp"
#include "caf/node_id.hpp"
#include "caf/resumable.hpp"
namespace caf {
namespace io {
namespace basp {
/// Deserializes payloads for BASP messages asynchronously.
class worker : public resumable, public remote_message_handler<worker> {
public:
// -- friends ----------------------------------------------------------------
friend worker_hub;
friend remote_message_handler<worker>;
// -- member types -----------------------------------------------------------
using atomic_pointer = std::atomic<worker*>;
using scheduler_type = scheduler::abstract_coordinator;
using buffer_type = std::vector<char>;
// -- constructors, destructors, and assignment operators --------------------
~worker() override;
// -- management -------------------------------------------------------------
void launch(const node_id& last_hop, const basp::header& hdr,
const buffer_type& payload);
// -- 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*)
+ sizeof(message_queue*)
+ sizeof(proxy_registry*)
+ sizeof(actor_system*);
static_assert(CAF_CACHE_LINE_SIZE > pointer_members_size,
"invalid cache line size");
// -- member variables -------------------------------------------------------
/// Points to the next worker in the hub.
atomic_pointer next_;
/// Points to our home hub.
worker_hub* hub_;
/// Points to the queue for establishing strict ordering.
message_queue* queue_;
/// Points to our proxy registry / factory.
proxy_registry* proxies_;
/// Points to the parent system.
actor_system* system_;
/// Prevents false sharing when writing to `next`.
char pad_[CAF_CACHE_LINE_SIZE - pointer_members_size];
/// ID for local ordering.
uint64_t msg_id_;
/// Identifies the node that sent us `hdr_` and `payload_`.
node_id last_hop_;
/// The header for the next message. Either a direct_message or a
/// routed_message.
header hdr_;
/// Contains whatever this worker deserializes next.
buffer_type payload_;
};
} // namespace basp
} // namespace io
} // 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 <atomic>
#include "caf/fwd.hpp"
#include "caf/io/basp/fwd.hpp"
namespace caf {
namespace io {
namespace basp {
/// A central place where BASP 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 {
public:
// -- member types -----------------------------------------------------------
using pointer = worker*;
// -- constructors, destructors, and assignment operators --------------------
worker_hub();
~worker_hub();
// -- properties -------------------------------------------------------------
/// Creates a new worker and adds it to the hub.
void push_new_worker(message_queue&, proxy_registry&);
/// Add a worker to the hub.
void push(pointer ptr);
/// Get a worker from the hub.
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// hub is currently empty.
pointer pop();
/// Check which worker would `pop` currently return.
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// hub is currently empty.
pointer peek();
private:
// -- member variables -------------------------------------------------------
std::atomic<pointer> head_;
};
} // namespace basp
} // namespace io
} // 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. *
******************************************************************************/
#include "caf/io/basp/worker.hpp"
#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"
namespace caf {
namespace io {
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()) {
CAF_IGNORE_UNUSED(pad_);
}
worker::~worker() {
// nop
}
// -- management ---------------------------------------------------------------
void worker::launch(const node_id& last_hop, const basp::header& hdr,
const buffer_type& payload) {
CAF_ASSERT(hdr.dest_actor != 0);
CAF_ASSERT(hdr.operation == basp::message_type::direct_message
|| hdr.operation == basp::message_type::routed_message);
msg_id_ = queue_->new_id();
last_hop_ = last_hop;
memcpy(&hdr_, &hdr, sizeof(basp::header));
payload_.assign(payload.begin(), payload.end());
system_->scheduler().enqueue(this);
}
// -- 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);
hub_->push(this);
return resumable::awaiting_message;
}
void worker::intrusive_ptr_add_ref_impl() {
// The basp::instance owns the hub (which owns this object) and must make
// sure to wait for pending workers at exit.
}
void worker::intrusive_ptr_release_impl() {
// nop
}
} // namespace basp
} // namespace io
} // 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. *
******************************************************************************/
#include "caf/io/basp/worker_hub.hpp"
#include "caf/io/basp/worker.hpp"
namespace caf {
namespace io {
namespace basp {
// -- constructors, destructors, and assignment operators ----------------------
worker_hub::worker_hub() : head_(nullptr) {
// nop
}
worker_hub::~worker_hub() {
auto head = head_.load();
while (head != nullptr) {
auto next = head->next_.load();
delete head;
head = next;
}
}
// -- properties ---------------------------------------------------------------
void worker_hub::push_new_worker(message_queue& queue,
proxy_registry& proxies) {
push(new worker(*this, queue, proxies));
}
void worker_hub::push(pointer ptr) {
auto next = head_.load();
for (;;) {
ptr->next_ = next;
if (head_.compare_exchange_strong(next, ptr))
return;
}
}
worker_hub::pointer worker_hub::pop() {
auto result = head_.load();
if (result == nullptr)
return nullptr;
for (;;) {
auto next = result->next_.load();
if (head_.compare_exchange_strong(result, next))
return result;
}
}
worker_hub::pointer worker_hub::peek() {
return head_.load();
}
} // namespace basp
} // namespace io
} // 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. *
******************************************************************************/
#define CAF_SUITE io_basp_worker
#include "caf/io/basp/worker.hpp"
#include "caf/test/dsl.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_control_block.hpp"
#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"
using namespace caf;
namespace {
behavior testee_impl() {
return {
[](ok_atom) {
// nop
}
};
}
class mock_actor_proxy : public actor_proxy {
public:
explicit mock_actor_proxy(actor_config& cfg) : actor_proxy(cfg) {
// nop
}
void enqueue(mailbox_element_ptr, execution_unit*) override {
CAF_FAIL("mock_actor_proxy::enqueue called");
}
void kill_proxy(execution_unit*, error) override {
// nop
}
};
class mock_proxy_registry_backend : public proxy_registry::backend {
public:
mock_proxy_registry_backend(actor_system& sys) : sys_(sys) {
//nop
}
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override {
actor_config cfg;
return make_actor<mock_actor_proxy, strong_actor_ptr>(aid, nid, &sys_, cfg);
}
void set_last_hop(node_id*) override {
// nop
}
private:
actor_system& sys_;
};
struct fixture : test_coordinator_fixture<> {
io::basp::worker_hub hub;
io::basp::message_queue queue;
mock_proxy_registry_backend proxies_backend;
proxy_registry proxies;
node_id last_hop;
actor testee;
fixture()
: proxies_backend(sys),
proxies(sys, proxies_backend),
last_hop(123, "0011223344556677889900112233445566778899") {
testee = sys.spawn<lazy_init>(testee_impl);
sys.registry().put(testee.id(), testee);
}
~fixture() {
sys.registry().erase(testee.id());
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(worker_tests, fixture)
CAF_TEST(deliver serialized message) {
CAF_MESSAGE("create the BASP worker");
CAF_REQUIRE_EQUAL(hub.peek(), nullptr);
hub.push_new_worker(queue, proxies);
CAF_REQUIRE_NOT_EQUAL(hub.peek(), nullptr);
auto w = hub.pop();
CAF_MESSAGE("create a fake message + BASP header");
std::vector<char> payload;
std::vector<strong_actor_ptr> stages;
binary_serializer sink{sys, payload};
if (auto err = sink(stages, make_message(ok_atom::value)))
CAF_FAIL("unable to serialize message: " << sys.render(err));
io::basp::header hdr{io::basp::message_type::direct_message,
0,
static_cast<uint32_t>(payload.size()),
make_message_id().integer_value(),
42,
testee.id()};
CAF_MESSAGE("launch worker");
w->launch(last_hop, hdr, payload);
sched.run_once();
expect((ok_atom), from(_).to(testee));
}
CAF_TEST_FIXTURE_SCOPE_END()
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