Commit 92de836b authored by Dominik Charousset's avatar Dominik Charousset

Implement scaffold for new multiplexer

parent 9791f0c9
......@@ -8,9 +8,12 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files
set(LIBCAF_NET_SRCS
src/host.cpp
src/multiplexer.cpp
src/network_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
src/socket.cpp
src/socket_manager.cpp
src/stream_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 <memory>
#include <mutex>
#include <thread>
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp"
extern "C" {
struct pollfd;
} // extern "C"
namespace caf {
namespace net {
/// Multiplexes any number of ::socket_manager objects with a ::socket.
class multiplexer : public std::enable_shared_from_this<multiplexer> {
public:
// -- member types -----------------------------------------------------------
using pollfd_list = std::vector<pollfd>;
using manager_list = std::vector<socket_manager_ptr>;
// -- constructors, destructors, and assignment operators --------------------
multiplexer();
~multiplexer();
error init();
// -- properties -------------------------------------------------------------
/// Returns the number of currently active socket managers.
size_t num_socket_managers() const noexcept;
/// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t index_of(const socket_manager_ptr& mgr);
// -- thread-safe signaling --------------------------------------------------
/// @thread-safe
void update(const socket_manager_ptr& mgr);
// -- control flow -----------------------------------------------------------
/// Polls I/O activity once and runs all socket event handlers that become
/// ready as a result.
bool poll_once(bool blocking);
/// Polls until no socket event handler remains.
void run();
void handle_updates();
void handle(const socket_manager_ptr& mgr, int mask);
protected:
// -- convenience functions --------------------------------------------------
void add(socket_manager_ptr mgr);
// -- member variables -------------------------------------------------------
/// Bookkeeping data for managed sockets.
std::vector<pollfd> pollset_;
/// Maps sockets to their owning managers by storing the managers in the same
/// order as their sockets appear in `pollset_`.
manager_list managers_;
/// Managers that updated their event mask and need updating.
manager_list dirty_managers_;
/// Stores the ID of the thread this multiplexer is running in. Must be set
/// by the subclass.
std::thread::id tid_;
///
pipe_socket write_handle_;
///
std::mutex write_lock_;
};
/// @relates multiplexer
using multiplexer_ptr = std::shared_ptr<multiplexer>;
/// @relates multiplexer
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
} // 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/pipe_socket.hpp"
#include "caf/net/socket_manager.hpp"
namespace caf {
namespace net {
class pollset_updater : public socket_manager {
public:
// -- member types -----------------------------------------------------------
using super = socket_manager;
// -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, multiplexer_ptr parent);
~pollset_updater() override;
// -- properties -------------------------------------------------------------
/// Returns the managed socket.
pipe_socket handle() const noexcept {
return socket_cast<pipe_socket>(handle_);
}
// -- interface functions ----------------------------------------------------
bool handle_read_event() override;
bool handle_write_event() override;
void handle_error(sec code) override;
};
} // 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 <atomic>
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp"
namespace caf {
namespace net {
/// Manages the lifetime of a single socket and handles any I/O events on it.
class socket_manager : public ref_counted {
public:
// -- constructors, destructors, and assignment operators --------------------
/// @pre `parent != nullptr`
/// @pre `handle != invalid_socket`
socket_manager(socket handle, const multiplexer_ptr& parent);
virtual ~socket_manager();
socket_manager(const socket_manager&) = delete;
socket_manager& operator=(const socket_manager&) = delete;
// -- properties -------------------------------------------------------------
/// Returns the managed socket.
socket handle() const noexcept {
return handle_;
}
/// Returns for which operations the socket is registered (read or write).
operation mask() const noexcept;
/// Adds given flag(s) to the event mask and updates the parent on success.
/// @returns `false` if `mask()` already contained `flag`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none`
bool mask_add(operation flag) noexcept;
/// Deletes given flag(s) from the event mask and updates the parent on
/// success.
/// @returns `false` if `mask()` already missed `flag`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none`
bool mask_del(operation flag) noexcept;
// -- pure virtual member functions ------------------------------------------
/// Called whenever the socket received new data.
virtual bool handle_read_event() = 0;
/// Called whenever the socket is allowed to send additional data.
virtual bool handle_write_event() = 0;
/// Called when the remote side becomes unreachable due to an error.
/// @param reason The error code as reported by the operating system.
virtual void handle_error(sec code) = 0;
protected:
// -- member variables -------------------------------------------------------
socket handle_;
std::atomic<operation> mask_;
weak_multiplexer_ptr parent_;
};
using socket_manager_ptr = intrusive_ptr<socket_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. *
******************************************************************************/
#include "caf/net/multiplexer.hpp"
#include "caf/config.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
#ifndef CAF_WINDOWS
# include <poll.h>
#else
# include "caf/detail/socket_sys_includes.hpp"
#endif // CAF_WINDOWS
namespace caf {
namespace net {
#ifndef POLLRDHUP
# define POLLRDHUP POLLHUP
#endif
#ifndef POLLPRI
# define POLLPRI POLLIN
#endif
namespace {
#ifdef CAF_WINDOWS
// From the MSDN: If the POLLPRI flag is set on a socket for the Microsoft
// Winsock provider, the WSAPoll function will fail.
const short input_mask = POLLIN;
#else
const short input_mask = POLLIN | POLLPRI;
#endif
const short error_mask = POLLRDHUP | POLLERR | POLLHUP | POLLNVAL;
const short output_mask = POLLOUT;
short to_bitmask(operation op) {
switch (op) {
case operation::read:
return input_mask;
case operation::write:
return output_mask;
case operation::read_write:
return input_mask | output_mask;
default:
return 0;
}
}
} // namespace
multiplexer::multiplexer() {
// nop
}
multiplexer::~multiplexer() {
// nop
}
error multiplexer::init() {
auto pipe_handles = make_pipe();
if (!pipe_handles)
return std::move(pipe_handles.error());
tid_ = std::this_thread::get_id();
add(make_counted<pollset_updater>(pipe_handles->first, shared_from_this()));
write_handle_ = pipe_handles->second;
return none;
}
size_t multiplexer::num_socket_managers() const noexcept {
return managers_.size();
}
ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) {
auto max_index = static_cast<ptrdiff_t>(managers_.size());
for (ptrdiff_t index = 0; index < max_index; ++index)
if (managers_[index] == mgr)
return index;
return -1;
}
void multiplexer::update(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) {
auto i = std::find(dirty_managers_.begin(), dirty_managers_.end(), mgr);
if (i == dirty_managers_.end())
dirty_managers_.emplace_back(mgr);
} else {
mgr->ref();
auto value = reinterpret_cast<intptr_t>(mgr.get());
variant<size_t, sec> res;
{ // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_};
res = write(write_handle_, &value, sizeof(intptr_t));
}
if (holds_alternative<sec>(res))
mgr->deref();
}
}
bool multiplexer::poll_once(bool blocking) {
// We'll call poll() until poll() succeeds or fails.
for (;;) {
int presult;
#ifdef CAF_WINDOWS
presult = ::WSAPoll(pollset_.data(), static_cast<ULONG>(pollset_.size()),
blocking ? -1 : 0);
#else
presult = ::poll(pollset_.data(), static_cast<nfds_t>(pollset_.size()),
blocking ? -1 : 0);
#endif
if (presult < 0) {
switch (last_socket_error()) {
case std::errc::interrupted: {
// A signal was caught. Simply try again.
CAF_LOG_DEBUG("received errc::interrupted, try again");
break;
}
case std::errc::not_enough_memory: {
CAF_LOG_ERROR("poll() failed due to insufficient memory");
// There's not much we can do other than try again in hope someone
// else releases memory.
break;
}
default: {
// Must not happen.
perror("poll() failed");
CAF_CRITICAL("poll() failed");
}
}
// Rinse and repeat.
continue;
}
CAF_LOG_DEBUG("poll() on" << pollset_.size() << "sockets reported"
<< presult << "event(s)");
// No activity.
if (presult == 0)
return false;
// Scan pollset for events.
CAF_LOG_DEBUG("scan pollset for socket events");
for (size_t i = 0; i < pollset_.size() && presult > 0; ++i) {
auto& x = pollset_[i];
if (x.revents != 0) {
handle(managers_[i], x.revents);
--presult;
}
}
handle_updates();
return true;
}
}
void multiplexer::handle_updates() {
for (auto mgr : dirty_managers_) {
auto index = index_of(mgr.get());
if (index == -1) {
add(std::move(mgr));
} else {
// Update or remove an existing manager in the pollset.
if (mgr->mask() == operation::none) {
pollset_.erase(pollset_.begin() + index);
managers_.erase(managers_.begin() + index);
} else {
pollset_[index].events = to_bitmask(mgr->mask());
}
}
}
dirty_managers_.clear();
}
void multiplexer::handle(const socket_manager_ptr& mgr, int mask) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()) << CAF_ARG(mask));
CAF_ASSERT(mgr != nullptr);
bool checkerror = true;
if ((mask & input_mask) != 0) {
checkerror = false;
if (!mgr->handle_read_event())
mgr->mask_del(operation::read);
}
if ((mask & output_mask) != 0) {
checkerror = false;
if (!mgr->handle_write_event())
mgr->mask_del(operation::write);
}
if (checkerror && ((mask & error_mask) != 0)) {
if (mask & POLLNVAL)
mgr->handle_error(sec::socket_invalid);
else if (mask & POLLHUP)
mgr->handle_error(sec::socket_disconnected);
else
mgr->handle_error(sec::socket_operation_failed);
mgr->mask_del(operation::read_write);
}
}
void multiplexer::add(socket_manager_ptr mgr) {
pollfd new_entry{socket_cast<socket_id>(mgr->handle()),
to_bitmask(mgr->mask()), 0};
pollset_.emplace_back(new_entry);
managers_.emplace_back(std::move(mgr));
}
} // 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/pollset_updater.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace net {
pollset_updater::pollset_updater(pipe_socket read_handle,
multiplexer_ptr parent)
: super(read_handle, std::move(parent)) {
mask_ = operation::read;
}
pollset_updater::~pollset_updater() {
// nop
}
bool pollset_updater::handle_read_event() {
for (;;) {
intptr_t value;
auto res = read(handle(), &value, sizeof(intptr_t));
if (auto num_bytes = get_if<size_t>(&res)) {
CAF_ASSERT(*num_bytes == sizeof(intptr_t));
socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false};
if (auto ptr = parent_.lock())
ptr->update(mgr);
}
return get<sec>(res) == sec::unavailable_or_would_block;
}
}
bool pollset_updater::handle_write_event() {
return false;
}
void pollset_updater::handle_error(sec) {
// nop
}
} // 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/socket_manager.hpp"
#include "caf/config.hpp"
#include "caf/net/multiplexer.hpp"
namespace caf {
namespace net {
socket_manager::socket_manager(socket handle, const multiplexer_ptr& parent)
: handle_(handle), mask_(operation::none), parent_(parent) {
CAF_ASSERT(parent != nullptr);
CAF_ASSERT(handle_ != invalid_socket);
}
socket_manager::~socket_manager() {
close(handle_);
}
operation socket_manager::mask() const noexcept {
return mask_.load();
}
bool socket_manager::mask_add(operation flag) noexcept {
CAF_ASSERT(flag != operation::none);
auto x = mask();
while ((x & flag) != flag)
if (mask_.compare_exchange_strong(x, x | flag)) {
if (auto ptr = parent_.lock())
ptr->update(this);
return true;
}
return false;
}
bool socket_manager::mask_del(operation flag) noexcept {
CAF_ASSERT(flag != operation::none);
auto x = mask();
while ((x & flag) != operation::none)
if (mask_.compare_exchange_strong(x, x & ~flag)) {
if (auto ptr = parent_.lock())
ptr->update(this);
return true;
}
return false;
}
} // 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 multiplexer
#include "caf/net/multiplexer.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <new>
#include <tuple>
#include <vector>
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp"
using namespace caf;
using namespace caf::net;
namespace {
class dummy_manager : public socket_manager {
public:
dummy_manager(size_t& manager_count, stream_socket handle,
multiplexer_ptr parent)
: socket_manager(handle, std::move(parent)),
count_(manager_count),
rd_buf_pos_(0) {
++count_;
rd_buf_.resize(1024);
}
~dummy_manager() {
--count_;
}
stream_socket handle() const noexcept {
return socket_cast<stream_socket>(socket_manager::handle());
}
bool handle_read_event() {
if (read_capacity() < 1024)
rd_buf_.resize(rd_buf_.size() + 2048);
auto res = read(handle(), read_position_begin(), read_capacity());
if (auto num_bytes = get_if<size_t>(&res)) {
CAF_ASSERT(*num_bytes > 0);
rd_buf_pos_ += *num_bytes;
return true;
}
return get<sec>(res) == sec::unavailable_or_would_block;
}
bool handle_write_event() {
if (wr_buf_.size() == 0)
return false;
auto res = write(handle(), wr_buf_.data(), wr_buf_.size());
if (auto num_bytes = get_if<size_t>(&res)) {
CAF_ASSERT(*num_bytes > 0);
wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + *num_bytes);
return wr_buf_.size() > 0;
}
return get<sec>(res) == sec::unavailable_or_would_block;
}
void handle_error(sec) {
}
void send(string_view x) {
wr_buf_.insert(wr_buf_.end(), x.begin(), x.end());
}
std::string receive() {
std::string result(rd_buf_.data(), read_position_begin());
rd_buf_pos_ = 0;
return result;
}
private:
char* read_position_begin() {
return rd_buf_.data() + rd_buf_pos_;
}
char* read_position_end() {
return rd_buf_.data() + rd_buf_.size();
}
size_t read_capacity() const {
return rd_buf_.size() - rd_buf_pos_;
}
size_t& count_;
size_t rd_buf_pos_;
std::vector<char> wr_buf_;
std::vector<char> rd_buf_;
};
using dummy_manager_ptr = intrusive_ptr<dummy_manager>;
struct fixture : host_fixture {
fixture() : manager_count(0) {
mpx = std::make_shared<multiplexer>();
CAF_REQUIRE_EQUAL(mpx->num_socket_managers(), 0u);
CAF_REQUIRE_EQUAL(mpx->init(), none);
CAF_REQUIRE_EQUAL(mpx->num_socket_managers(), 1u);
auto sockets = unbox(make_stream_socket_pair());
alice = make_counted<dummy_manager>(manager_count, sockets.first, mpx);
bob = make_counted<dummy_manager>(manager_count, sockets.second, mpx);
alice->mask_add(operation::read);
bob->mask_add(operation::read);
mpx->handle_updates();
CAF_REQUIRE_EQUAL(mpx->num_socket_managers(), 3u);
}
~fixture() {
alice.reset();
bob.reset();
mpx.reset();
CAF_REQUIRE_EQUAL(manager_count, 0u);
}
size_t manager_count;
multiplexer_ptr mpx;
dummy_manager_ptr alice;
dummy_manager_ptr bob;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(multiplexer_tests, fixture)
CAF_TEST(poll once) {
alice->send("hello bob");
alice->mask_add(operation::write);
mpx->handle_updates();
while (mpx->poll_once(false))
; // Repeat.
CAF_CHECK_EQUAL(bob->receive(), "hello bob");
}
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