Commit 776d9956 authored by Dominik Charousset's avatar Dominik Charousset

Add missing functions, docs, and tests

parent 92de836b
...@@ -64,9 +64,15 @@ public: ...@@ -64,9 +64,15 @@ public:
// -- thread-safe signaling -------------------------------------------------- // -- thread-safe signaling --------------------------------------------------
/// Causes the multiplexer to update its event bitmask for `mgr`.
/// @thread-safe /// @thread-safe
void update(const socket_manager_ptr& mgr); void update(const socket_manager_ptr& mgr);
/// Closes the pipe for signaling updates to the multiplexer. After closing
/// the pipe, calls to `update` no longer have any effect.
/// @thread-safe
void close_pipe();
// -- control flow ----------------------------------------------------------- // -- control flow -----------------------------------------------------------
/// Polls I/O activity once and runs all socket event handlers that become /// Polls I/O activity once and runs all socket event handlers that become
...@@ -76,19 +82,22 @@ public: ...@@ -76,19 +82,22 @@ public:
/// Polls until no socket event handler remains. /// Polls until no socket event handler remains.
void run(); void run();
/// Processes all updates on the socket managers.
void handle_updates(); void handle_updates();
void handle(const socket_manager_ptr& mgr, int mask);
protected: protected:
// -- convenience functions -------------------------------------------------- // -- utility functions ------------------------------------------------------
/// Handles an I/O event on given manager.
void handle(const socket_manager_ptr& mgr, int mask);
/// Adds a new socket manager to the pollset.
void add(socket_manager_ptr mgr); void add(socket_manager_ptr mgr);
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Bookkeeping data for managed sockets. /// Bookkeeping data for managed sockets.
std::vector<pollfd> pollset_; pollfd_list pollset_;
/// Maps sockets to their owning managers by storing the managers in the same /// Maps sockets to their owning managers by storing the managers in the same
/// order as their sockets appear in `pollset_`. /// order as their sockets appear in `pollset_`.
...@@ -97,14 +106,14 @@ protected: ...@@ -97,14 +106,14 @@ protected:
/// Managers that updated their event mask and need updating. /// Managers that updated their event mask and need updating.
manager_list dirty_managers_; manager_list dirty_managers_;
/// Stores the ID of the thread this multiplexer is running in. Must be set /// Stores the ID of the thread this multiplexer is running in. Set when
/// by the subclass. /// calling `init()`.
std::thread::id tid_; std::thread::id tid_;
/// /// Used for pushing updates to the multiplexer's thread.
pipe_socket write_handle_; pipe_socket write_handle_;
/// /// Guards `write_handle_`.
std::mutex write_lock_; std::mutex write_lock_;
}; };
......
...@@ -115,13 +115,24 @@ void multiplexer::update(const socket_manager_ptr& mgr) { ...@@ -115,13 +115,24 @@ void multiplexer::update(const socket_manager_ptr& mgr) {
variant<size_t, sec> res; variant<size_t, sec> res;
{ // Lifetime scope of guard. { // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_}; std::lock_guard<std::mutex> guard{write_lock_};
res = write(write_handle_, &value, sizeof(intptr_t)); if (write_handle_ != invalid_socket)
res = write(write_handle_, &value, sizeof(intptr_t));
else
res = sec::socket_invalid;
} }
if (holds_alternative<sec>(res)) if (holds_alternative<sec>(res))
mgr->deref(); mgr->deref();
} }
} }
void multiplexer::close_pipe() {
std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket) {
close(write_handle_);
write_handle_ = pipe_socket{};
}
}
bool multiplexer::poll_once(bool blocking) { bool multiplexer::poll_once(bool blocking) {
// We'll call poll() until poll() succeeds or fails. // We'll call poll() until poll() succeeds or fails.
for (;;) { for (;;) {
...@@ -174,6 +185,11 @@ bool multiplexer::poll_once(bool blocking) { ...@@ -174,6 +185,11 @@ bool multiplexer::poll_once(bool blocking) {
} }
} }
void multiplexer::run() {
while (!pollset_.empty())
poll_once(true);
}
void multiplexer::handle_updates() { void multiplexer::handle_updates() {
for (auto mgr : dirty_managers_) { for (auto mgr : dirty_managers_) {
auto index = index_of(mgr.get()); auto index = index_of(mgr.get());
......
...@@ -52,10 +52,10 @@ public: ...@@ -52,10 +52,10 @@ public:
} }
stream_socket handle() const noexcept { stream_socket handle() const noexcept {
return socket_cast<stream_socket>(socket_manager::handle()); return socket_cast<stream_socket>(handle_);
} }
bool handle_read_event() { bool handle_read_event() override {
if (read_capacity() < 1024) if (read_capacity() < 1024)
rd_buf_.resize(rd_buf_.size() + 2048); rd_buf_.resize(rd_buf_.size() + 2048);
auto res = read(handle(), read_position_begin(), read_capacity()); auto res = read(handle(), read_position_begin(), read_capacity());
...@@ -67,7 +67,7 @@ public: ...@@ -67,7 +67,7 @@ public:
return get<sec>(res) == sec::unavailable_or_would_block; return get<sec>(res) == sec::unavailable_or_would_block;
} }
bool handle_write_event() { bool handle_write_event() override {
if (wr_buf_.size() == 0) if (wr_buf_.size() == 0)
return false; return false;
auto res = write(handle(), wr_buf_.data(), wr_buf_.size()); auto res = write(handle(), wr_buf_.data(), wr_buf_.size());
...@@ -79,7 +79,8 @@ public: ...@@ -79,7 +79,8 @@ public:
return get<sec>(res) == sec::unavailable_or_would_block; return get<sec>(res) == sec::unavailable_or_would_block;
} }
void handle_error(sec) { void handle_error(sec code) override {
CAF_FAIL("handle_error called with code " << code);
} }
void send(string_view x) { void send(string_view x) {
...@@ -117,46 +118,57 @@ private: ...@@ -117,46 +118,57 @@ private:
using dummy_manager_ptr = intrusive_ptr<dummy_manager>; using dummy_manager_ptr = intrusive_ptr<dummy_manager>;
struct fixture : host_fixture { struct fixture : host_fixture {
fixture() : manager_count(0) { fixture() : manager_count(0), mpx(std::make_shared<multiplexer>()) {
mpx = std::make_shared<multiplexer>(); // nop
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() { ~fixture() {
alice.reset();
bob.reset();
mpx.reset(); mpx.reset();
CAF_REQUIRE_EQUAL(manager_count, 0u); CAF_REQUIRE_EQUAL(manager_count, 0u);
} }
void exhaust() {
while (mpx->poll_once(false))
; // Repeat.
}
size_t manager_count; size_t manager_count;
multiplexer_ptr mpx; multiplexer_ptr mpx;
dummy_manager_ptr alice;
dummy_manager_ptr bob;
}; };
} // namespace } // namespace
CAF_TEST_FIXTURE_SCOPE(multiplexer_tests, fixture) CAF_TEST_FIXTURE_SCOPE(multiplexer_tests, fixture)
CAF_TEST(poll once) { CAF_TEST(default construction) {
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 0u);
}
CAF_TEST(init) {
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 0u);
CAF_REQUIRE_EQUAL(mpx->init(), none);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
mpx->close_pipe();
exhaust();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 0u);
// Calling run must have no effect now.
mpx->run();
}
CAF_TEST(send and receive) {
CAF_REQUIRE_EQUAL(mpx->init(), none);
auto sockets = unbox(make_stream_socket_pair());
auto alice = make_counted<dummy_manager>(manager_count, sockets.first, mpx);
auto bob = make_counted<dummy_manager>(manager_count, sockets.second, mpx);
alice->mask_add(operation::read);
bob->mask_add(operation::read);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
alice->send("hello bob"); alice->send("hello bob");
alice->mask_add(operation::write); alice->mask_add(operation::write);
mpx->handle_updates(); mpx->handle_updates();
while (mpx->poll_once(false)) exhaust();
; // Repeat.
CAF_CHECK_EQUAL(bob->receive(), "hello bob"); CAF_CHECK_EQUAL(bob->receive(), "hello bob");
} }
......
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