Commit 91e2761f authored by Dominik Charousset's avatar Dominik Charousset

Properly track published actors, add unpublish

Extend the BASP broker to monitor published actors and automatically unpublish
them using attachables. Also add missing io::unpublish function to allow users
to release ports manually. Fixes #189 and closes #197.
parent 3d596c39
......@@ -13,11 +13,13 @@ set (LIBCAF_IO_SRCS
src/middleman.cpp
src/hook.cpp
src/default_multiplexer.cpp
src/publish.cpp
src/publish_local_groups.cpp
src/remote_actor_proxy.cpp
src/remote_group.cpp
src/manager.cpp
src/stream_manager.cpp
src/unpublish.cpp
src/acceptor_manager.cpp
src/multiplexer.cpp)
......
......@@ -24,8 +24,8 @@
#include "caf/io/publish.hpp"
#include "caf/io/spawn_io.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/unpublish.hpp"
#include "caf/io/max_msg_size.hpp"
#include "caf/io/publish_impl.hpp"
#include "caf/io/remote_actor.hpp"
#include "caf/io/remote_group.hpp"
#include "caf/io/receive_policy.hpp"
......
......@@ -52,16 +52,11 @@ class basp_broker : public broker, public actor_namespace::backend {
behavior make_behavior() override;
/*
template <class SocketAcceptor>
void publish(abstract_actor_ptr whom, SocketAcceptor fd) {
auto hdl = add_acceptor(std::move(fd));
announce_published_actor(hdl, whom);
}
*/
void add_published_actor(accept_handle hdl,
const abstract_actor_ptr& whom,
uint16_t port);
void announce_published_actor(accept_handle hdl,
const abstract_actor_ptr& whom);
void remove_published_actor(const abstract_actor_ptr& whom, uint16_t port);
actor_proxy_ptr make_proxy(const id_type&, actor_id) override;
......@@ -203,8 +198,8 @@ class basp_broker : public broker, public actor_namespace::backend {
actor_namespace m_namespace; // manages proxies
std::map<connection_handle, connection_context> m_ctx;
std::map<accept_handle, abstract_actor_ptr>
m_published_actors;
std::map<accept_handle, std::pair<abstract_actor_ptr, uint16_t>> m_acceptors;
std::map<uint16_t, accept_handle> m_open_ports;
routing_table m_routes; // stores non-direct routes
std::set<blacklist_entry, blacklist_less> m_blacklist; // stores invalidated
// routes
......
......@@ -24,12 +24,13 @@
#include "caf/actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/io/publish_impl.hpp"
#include "caf/typed_actor.hpp"
namespace caf {
namespace io {
void publish_impl(abstract_actor_ptr whom, uint16_t port, const char* in);
/**
* Publishes `whom` at `port`. The connection is managed by the middleman.
* @param whom Actor that should be published at `port`.
......@@ -41,7 +42,7 @@ inline void publish(caf::actor whom, uint16_t port, const char* in = nullptr) {
if (!whom) {
return;
}
io::publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in);
publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in);
}
/**
......@@ -53,7 +54,7 @@ void typed_publish(typed_actor<Rs...> whom, uint16_t port,
if (!whom) {
return;
}
io::publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in);
publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in);
}
} // namespace io
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_IO_UNPUBLISH_HPP
#define CAF_IO_UNPUBLISH_HPP
#include <cstdint>
#include "caf/actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/typed_actor.hpp"
namespace caf {
namespace io {
void unpublish_impl(abstract_actor_ptr whom, uint16_t port, bool block_caller);
/**
* Unpublishes `whom` by closing `port`.
* @param whom Actor that should be unpublished at `port`.
* @param port TCP port.
*/
inline void unpublish(caf::actor whom, uint16_t port) {
if (!whom) {
return;
}
unpublish_impl(actor_cast<abstract_actor_ptr>(whom), port, true);
}
/**
* @copydoc unpublish(actor,uint16_t)
*/
template <class... Rs>
void typed_unpublish(typed_actor<Rs...> whom, uint16_t port) {
if (!whom) {
return;
}
unpublish_impl(actor_cast<abstract_actor_ptr>(whom), port, true);
}
} // namespace io
} // namespace caf
#endif // CAF_IO_UNPUBLISH_HPP
......@@ -26,6 +26,7 @@
#include "caf/io/basp.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/unpublish.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/remote_actor_proxy.hpp"
......@@ -61,7 +62,7 @@ behavior basp_broker::make_behavior() {
ctx.hdl = msg.handle;
ctx.handshake_data = nullptr;
ctx.state = await_client_handshake;
init_handshake_as_sever(ctx, m_published_actors[msg.source]->address());
init_handshake_as_sever(ctx, m_acceptors[msg.source].first->address());
},
// received from underlying broker implementation
[=](const connection_closed_msg& msg) {
......@@ -99,9 +100,17 @@ behavior basp_broker::make_behavior() {
m_ctx.erase(msg.handle);
},
// received from underlying broker implementation
[=](const acceptor_closed_msg&) {
[=](const acceptor_closed_msg& msg) {
CAF_LOGM_TRACE("make_behavior$acceptor_closed_msg", "");
// nop
auto i = m_acceptors.find(msg.handle);
if (i == m_acceptors.end()) {
CAF_LOG_INFO("accept handle no longer in use");
return;
}
if (m_open_ports.erase(i->second.second) == 0) {
CAF_LOG_INFO("accept handle was not bound to a port");
}
m_acceptors.erase(i);
},
// received from proxy instances
on(atom("_Dispatch"), arg_match) >> [=](const actor_addr& sender,
......@@ -646,17 +655,46 @@ void basp_broker::init_handshake_as_sever(connection_context& ctx,
configure_read(ctx.hdl, receive_policy::exactly(basp::header_size));
}
void basp_broker::announce_published_actor(accept_handle hdl,
const abstract_actor_ptr& ptr) {
void basp_broker::add_published_actor(accept_handle hdl,
const abstract_actor_ptr& ptr,
uint16_t port) {
CAF_LOG_TRACE("");
if (!ptr) {
return;
}
CAF_LOG_TRACE("");
m_published_actors.insert(std::make_pair(hdl, ptr));
m_acceptors.insert(std::make_pair(hdl, std::make_pair(ptr, port)));
m_open_ports.insert(std::make_pair(port, hdl));
ptr->attach_functor([port](abstract_actor* self, uint32_t) {
unpublish_impl(self, port, false);
});
if (ptr->node() == node()) {
singletons::get_actor_registry()->put(ptr->id(), ptr);
}
}
void basp_broker::remove_published_actor(const abstract_actor_ptr& whom,
uint16_t port) {
CAF_LOG_TRACE("");
auto i = m_open_ports.find(port);
if (i == m_open_ports.end()) {
return;
}
auto j = m_acceptors.find(i->second);
if (j == m_acceptors.end()) {
CAF_LOG_ERROR("accept handle for port " << port
<< " not found in m_published_actors");
m_open_ports.erase(i);
return;
}
if (j->second.first != whom) {
CAF_LOG_INFO("port has been bound to a different actor already");
return;
}
close(j->first);
m_open_ports.erase(i);
m_acceptors.erase(j);
}
} // namespace io
} // namespace caf
......@@ -17,9 +17,6 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_PUBLISH_IMPL_HPP
#define CAF_IO_PUBLISH_IMPL_HPP
#include <future>
#include "caf/actor_cast.hpp"
......@@ -28,13 +25,13 @@
#include "caf/detail/singletons.hpp"
#include "caf/detail/actor_registry.hpp"
#include "caf/io/publish.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp"
namespace caf {
namespace io {
template <class... Ts>
void publish_impl(abstract_actor_ptr whom, uint16_t port, const char* in) {
using namespace detail;
auto mm = middleman::instance();
......@@ -43,7 +40,7 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port, const char* in) {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
try {
auto hdl = mm->backend().add_tcp_doorman(bro.get(), port, in);
bro->announce_published_actor(hdl, whom);
bro->add_published_actor(hdl, whom, port);
mm->notify<hook::actor_published>(whom->address(), port);
res.set_value(true);
}
......@@ -57,5 +54,3 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port, const char* in) {
} // namespace io
} // namespace caf
#endif // CAF_IO_PUBLISH_IMPL_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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/unpublish.hpp"
#include "caf/send.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/unpublish.hpp"
#include "caf/io/basp_broker.hpp"
namespace caf {
namespace io {
void unpublish_impl(abstract_actor_ptr whom, uint16_t port, bool block_caller) {
auto mm = middleman::instance();
if (block_caller) {
scoped_actor self;
mm->run_later([&] {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
bro->remove_published_actor(whom, port);
anon_send(self, atom("done"));
});
self->receive(
on(atom("done")) >> [] {
// ok, basp_broker is done
}
);
} else {
mm->run_later([whom, port, mm] {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
bro->remove_published_actor(whom, port);
});
}
}
} // namespace io
} // namespace caf
......@@ -38,5 +38,6 @@ add_unit_test(sync_send)
add_unit_test(broker)
add_unit_test(remote_actor ping_pong.cpp)
add_unit_test(typed_remote_actor)
add_unit_test(unpublish)
add_unit_test(optional)
add_unit_test(fixed_stack_actor)
......@@ -19,6 +19,8 @@ using namespace caf;
namespace {
atomic<long> s_destructors_called;
using string_pair = std::pair<std::string, std::string>;
using actor_vector = vector<actor>;
......@@ -162,6 +164,10 @@ class client : public event_based_actor {
return spawn_ping();
}
~client() {
++s_destructors_called;
}
private:
behavior spawn_ping() {
......@@ -230,9 +236,17 @@ class server : public event_based_actor {
public:
behavior make_behavior() override { return await_spawn_ping(); }
behavior make_behavior() override {
return await_spawn_ping();
}
server(bool run_in_loop = false) : m_run_in_loop(run_in_loop) {}
server(bool run_in_loop = false) : m_run_in_loop(run_in_loop) {
// nop
}
~server() {
++s_destructors_called;
}
private:
......@@ -351,7 +365,6 @@ void test_remote_actor(std::string app_path, bool run_remote_actor) {
auto serv2 = io::remote_actor("127.0.0.1", port);
CAF_CHECK(serv2 != invalid_actor && !serv2->is_remote());
CAF_CHECK(serv == serv2);
CAF_TEST(test_remote_actor);
thread child;
ostringstream oss;
oss << app_path << " -c " << port << " " << port0 << " " << gport;
......@@ -387,6 +400,7 @@ void test_remote_actor(std::string app_path, bool run_remote_actor) {
} // namespace <anonymous>
int main(int argc, char** argv) {
CAF_TEST(test_remote_actor);
announce<actor_vector>();
cout << "this node is: " << to_string(caf::detail::singletons::get_node_id())
<< endl;
......@@ -426,5 +440,8 @@ int main(int argc, char** argv) {
});
await_all_actors_done();
shutdown();
// we either spawn a server or a client, in both cases
// there must have been exactly one dtor called
CAF_CHECK_EQUAL(s_destructors_called.load(), 1);
return CAF_TEST_RESULT();
}
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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 <thread>
#include <atomic>
#include "test.hpp"
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using namespace caf;
namespace {
std::atomic<long> s_dtor_called;
class dummy : public event_based_actor {
public:
~dummy() {
++s_dtor_called;
}
behavior make_behavior() override {
return {
others() >> CAF_UNEXPECTED_MSG_CB(this)
};
}
};
uint16_t publish_at_some_port(uint16_t first_port, actor whom) {
auto port = first_port;
for (;;) {
try {
io::publish(whom, port);
return port;
}
catch (bind_failure&) {
// try next port
++port;
}
}
}
} // namespace <anonymous>
int main() {
CAF_TEST(test_unpublish);
auto d = spawn<dummy>();
auto port = publish_at_some_port(4242, d);
std::this_thread::sleep_for(std::chrono::milliseconds(50));
io::unpublish(d, port);
// must fail now
try {
auto oops = io::remote_actor("127.0.0.1", port);
CAF_FAILURE("unexpected: remote actor succeeded!");
} catch (network_error&) {
CAF_CHECKPOINT();
}
anon_send_exit(d, exit_reason::user_shutdown);
d = invalid_actor;
await_all_actors_done();
shutdown();
CAF_CHECK_EQUAL(s_dtor_called.load(), 1);
return CAF_TEST_RESULT();
}
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