Commit 61bcdaac authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #5

Implement endpoint managers
parents 0c22a39a d2af6398
......@@ -7,6 +7,7 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files
set(LIBCAF_NET_SRCS
src/endpoint_manager.cpp
src/host.cpp
src/multiplexer.cpp
src/network_socket.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 <cstddef>
#include <cstdint>
#include <memory>
#include "caf/actor.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace net {
/// Manages a communication endpoint.
class endpoint_manager : public socket_manager {
public:
// -- member types -----------------------------------------------------------
using super = socket_manager;
struct event : intrusive::singly_linked<event> {
struct resolve_request {
std::string path;
actor listener;
};
struct timeout {
atom_value type;
uint64_t id;
};
event(std::string path, actor listener);
event(atom_value type, uint64_t id);
/// Either contains a string for `resolve` requests or an `atom_value`
variant<resolve_request, timeout> value;
};
struct event_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = event;
using unique_pointer = std::unique_ptr<event>;
using queue_type = intrusive::drr_queue<event_policy>;
task_size_type task_size(const event&) const noexcept {
return 1;
}
};
using event_queue_type = intrusive::fifo_inbox<event_policy>;
struct message : intrusive::singly_linked<message> {
/// Original message to a remote actor.
mailbox_element_ptr msg;
/// Serialized representation of of `msg->content()`.
std::vector<char> payload;
};
struct message_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = message;
using unique_pointer = std::unique_ptr<message>;
using queue_type = intrusive::drr_queue<message_policy>;
task_size_type task_size(const message& x) const noexcept {
return x.payload.size();
}
};
using message_queue_type = intrusive::fifo_inbox<message_policy>;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager(socket handle, const multiplexer_ptr& parent,
actor_system& sys);
~endpoint_manager() override;
// -- properties -------------------------------------------------------------
event_queue_type& event_queue() {
return events_;
}
message_queue_type& message_queue() {
return messages_;
}
// -- event management -------------------------------------------------------
/// Resolves a path to a remote actor
void resolve(std::string path, actor listener);
// -- pure virtual member functions ------------------------------------------
/// Initializes the manager before adding it to the multiplexer's event loop.
virtual error init() = 0;
protected:
/// Points to the hosting actor system.
actor_system& sys_;
/// Stores control events.
event_queue_type events_;
/// Stores outbound messages.
message_queue_type messages_;
};
using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>;
} // namespace net
} // 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/net/endpoint_manager.hpp"
namespace caf {
namespace net {
template <class Transport, class Application>
class endpoint_manager_impl : public endpoint_manager {
public:
// -- member types -----------------------------------------------------------
using super = endpoint_manager;
using transport_type = Transport;
using application_type = Application;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys,
Transport trans, Application app)
: super(trans.handle(), parent, sys),
transport_(std::move(trans)),
application_(std::move(app)) {
// nop
}
~endpoint_manager_impl() override {
// nop
}
// -- properties -------------------------------------------------------------
transport_type& transport() {
return transport_;
}
application_type& application() {
return application_;
}
// -- interface functions ----------------------------------------------------
error init() override {
return transport_.init(*this);
}
bool handle_read_event() override {
return transport_.handle_read_event(*this);
}
bool handle_write_event() override {
if (!this->events_.empty()) {
this->events_.fetch_more();
auto& q = this->events_.queue();
q.inc_deficit(q.total_task_size());
for (auto ptr = q.next(); ptr != nullptr; ptr = q.next()) {
using timeout = endpoint_manager::event::timeout;
using resolve_request = endpoint_manager::event::resolve_request;
if (auto rr = get_if<resolve_request>(&ptr->value)) {
transport_.resolve(*this, std::move(rr->path),
std::move(rr->listener));
} else {
auto& t = get<timeout>(ptr->value);
transport_.timeout(*this, t.type, t.id);
}
}
}
return transport_.handle_write_event(*this);
}
void handle_error(sec code) override {
transport_.handle_error(application_, code);
}
private:
Transport transport_;
Application application_;
};
} // namespace net
} // 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/make_counted.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp"
namespace caf {
namespace net {
template <class Transport, class Application>
endpoint_manager_ptr make_endpoint_manager(const multiplexer_ptr& mpx,
actor_system& sys, Transport trans,
Application app) {
using impl = endpoint_manager_impl<Transport, Application>;
return make_counted<impl>(mpx, sys, std::move(trans), std::move(app));
}
} // namespace net
} // 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/net/endpoint_manager.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace caf {
namespace net {
endpoint_manager::event::event(std::string path, actor listener)
: value(resolve_request{std::move(path), std::move(listener)}) {
// nop
}
endpoint_manager::event::event(atom_value type, uint64_t id)
: value(timeout{type, id}) {
// nop
}
endpoint_manager::endpoint_manager(socket handle, const multiplexer_ptr& parent,
actor_system& sys)
: super(handle, parent),
sys_(sys),
events_(event_policy{}),
messages_(message_policy{}) {
// nop
}
endpoint_manager::~endpoint_manager() {
// nop
}
void endpoint_manager::resolve(std::string path, actor listener) {
using intrusive::inbox_result;
auto ptr = new event(std::move(path), std::move(listener));
switch (events_.push_back(ptr)) {
default:
break;
case inbox_result::unblocked_reader:
mask_add(operation::write);
break;
case inbox_result::queue_closed:
anon_send(listener, resolve_atom::value,
make_error(sec::request_receiver_down));
}
}
} // namespace net
} // 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 endpoint_manager
#include "caf/net/endpoint_manager.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
using namespace caf;
using namespace caf::net;
namespace {
string_view hello_manager{"hello manager!"};
string_view hello_test{"hello test!"};
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
}
multiplexer_ptr mpx;
};
class dummy_application {};
class dummy_transport {
public:
dummy_transport(stream_socket handle, std::shared_ptr<std::vector<char>> data)
: handle_(handle), data_(data), read_buf_(1024) {
// nop
}
stream_socket handle() {
return handle_;
}
template <class Manager>
error init(Manager& manager) {
write_buf_.insert(write_buf_.end(), hello_test.begin(), hello_test.end());
CAF_CHECK(manager.mask_add(operation::read_write));
return none;
}
template <class Manager>
bool handle_read_event(Manager&) {
auto res = read(handle_, read_buf_.data(), read_buf_.size());
if (auto num_bytes = get_if<size_t>(&res)) {
data_->insert(data_->end(), read_buf_.begin(),
read_buf_.begin() + *num_bytes);
return true;
}
return get<sec>(res) == sec::unavailable_or_would_block;
}
template <class Manager>
bool handle_write_event(Manager&) {
auto res = write(handle_, write_buf_.data(), write_buf_.size());
if (auto num_bytes = get_if<size_t>(&res)) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + *num_bytes);
return write_buf_.size() > 0;
}
return get<sec>(res) == sec::unavailable_or_would_block;
}
template <class Manager>
void handle_error(Manager&, sec) {
// nop
}
template <class Manager>
void resolve(Manager&, std::string path, actor listener) {
anon_send(listener, resolve_atom::value, std::move(path),
make_error(sec::feature_disabled));
}
template <class Manager>
void timeout(Manager&, atom_value, uint64_t) {
// nop
}
private:
stream_socket handle_;
std::shared_ptr<std::vector<char>> data_;
std::vector<char> read_buf_;
std::vector<char> write_buf_;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(send and receive) {
std::vector<char> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<char>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()),
sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf},
dummy_application{});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(sockets.second, hello_manager.data(),
hello_manager.size()),
hello_manager.size());
while (mpx->poll_once(false))
; // Repeat.
CAF_CHECK_EQUAL(string_view(buf->data(), buf->size()), hello_manager);
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()),
hello_test.size());
CAF_CHECK_EQUAL(string_view(read_buf.data(), hello_test.size()), hello_test);
}
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