Commit b721ecfb authored by Dominik Charousset's avatar Dominik Charousset

Add async delete/unpublish operation to MM actor

parent 3abbb933
......@@ -57,7 +57,7 @@ struct is_convertible_to_actor {
/**
* Identifies an untyped actor. Can be used with derived types
* of `event_based_actor`, `blocking_actor`, `actor_proxy`.
* of `event_based_actor`, `blocking_actor`, and `actor_proxy`.
*/
class actor : detail::comparable<actor>,
detail::comparable<actor, actor_addr>,
......
......@@ -79,6 +79,11 @@ using get_atom = atom_constant<atom("GET")>;
*/
using put_atom = atom_constant<atom("PUT")>;
/**
* Generic 'DELETE' atom for request operations.
*/
using delete_atom = atom_constant<atom("DELETE")>;
/**
* Generic 'OK' atom for response messages.
*/
......
......@@ -62,6 +62,7 @@ enum class atom_value : uint64_t;
using actor_id = uint32_t;
// intrusive pointer types
using abstract_actor_ptr = intrusive_ptr<abstract_actor>;
using abstract_group_ptr = intrusive_ptr<abstract_group>;
using actor_proxy_ptr = intrusive_ptr<actor_proxy>;
......
......@@ -51,7 +51,9 @@ class basp_broker : public broker, public actor_namespace::backend {
void add_published_actor(accept_handle hdl, const abstract_actor_ptr& whom,
uint16_t port);
void remove_published_actor(const abstract_actor_ptr& whom, uint16_t port);
bool remove_published_actor(const abstract_actor_ptr& whom);
bool remove_published_actor(const abstract_actor_ptr& whom, uint16_t port);
actor_proxy_ptr make_proxy(const node_id&, actor_id) override;
......
......@@ -32,40 +32,12 @@
#include "caf/io/hook.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman_actor.hpp"
#include "caf/io/network/multiplexer.hpp"
namespace caf {
namespace io {
/**
* API for asynchronous networking operations.
*/
using middleman_actor =
typed_actor<
replies_to<put_atom, actor_addr, uint16_t, std::string, bool>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<put_atom, actor_addr, uint16_t, std::string>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<put_atom, actor_addr, uint16_t, bool>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<put_atom, actor_addr, uint16_t>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<get_atom, std::string, uint16_t, std::set<std::string>>
::with_either<ok_atom, actor_addr>
::or_else<error_atom, std::string>,
replies_to<get_atom, std::string, uint16_t>
::with_either<ok_atom, actor_addr>
::or_else<error_atom, std::string>>;
/**
* Returns a handle for asynchronous networking operations.
*/
middleman_actor get_middleman_actor();
/**
* Manages brokers and network backends.
*/
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_MIDDLEMAN_ACTOR_HPP
#define CAF_IO_MIDDLEMAN_ACTOR_HPP
#include "caf/fwd.hpp"
#include "caf/typed_actor.hpp"
namespace caf {
namespace io {
/**
* A message passing interface for asynchronous networking operations.
*
* The interface implements the following pseudo code.
* ~~~
* interface middleman_actor {
* using put_result = either (ok_atom, uint16_t port)
* or (error_atom, std::string error_string);
*
* put_result <- (put_atom, actor_addr whom, uint16_t port,
* std::string addr, bool reuse_addr);
*
* put_result <- (put_atom, actor_addr whom,
* uint16_t port, std::string addr);
*
* put_result <- (put_atom, actor_addr whom, uint16_t port, bool reuse_addr);
*
* put_result <- (put_atom, actor_addr whom, uint16_t port);
*
* using get_result = either (ok_atom, actor_addr remote_address)
* or (error_atom, std::string error_string);
*
* get_result <- (get_atom, std::string hostname, uint16_t port);
*
* get_result <- (get_atom, std::string hostname, uint16_t port,
* std::set<std::string> expected_ifs);
*
* using delete_result = either (ok_atom)
* or (error_atom, std::string error_string);
*
* delete_result <- (delete_atom, actor_addr whom);
*
* delete_result <- (delete_atom, actor_addr whom, uint16_t port);
* }
* ~~~
*
* The `middleman_actor` actor offers the following operations:
* * `PUT` establishes a new `port <-> actor` mapping and returns the actual
* port in use on success. When passing 0 as `port` parameter,
* this is the only way to learn which port was used.
* Parameter | Description
* --------------|------------------------------------------------------------
* whom | Actor that should be published at port.
* port | Unused TCP port or 0 for any.
* addr | Optional; the IP address to listen to or INADDR_ANY
* reuse_addr | Optional; enable SO_REUSEPORT option (default: false)
* * `GET` queries remote node and returns an `actor_addr` to the remote
* actor on success. This handle must be cast to either `actor` or
* `typed_actor` using `actor_cast`.
* Parameter | Description
* --------------|------------------------------------------------------------
* hostname | Valid hostname or IP address.
* port | TCP port.
* expected_ifs | Optional; Interface of typed remote actor.
* * `DELETE` removes either all `port <-> actor` mappings for an actor or
* only a single one if the optional `port` parameter is set.
* Parameter | Description
* --------------|------------------------------------------------------------
* whom | Published actor.
* port | Optional; remove only a single mapping.
*/
using middleman_actor =
typed_actor<
replies_to<put_atom, actor_addr, uint16_t, std::string, bool>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<put_atom, actor_addr, uint16_t, std::string>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<put_atom, actor_addr, uint16_t, bool>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<put_atom, actor_addr, uint16_t>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<get_atom, std::string, uint16_t>
::with_either<ok_atom, actor_addr>
::or_else<error_atom, std::string>,
replies_to<get_atom, std::string, uint16_t, std::set<std::string>>
::with_either<ok_atom, actor_addr>
::or_else<error_atom, std::string>,
replies_to<delete_atom, actor_addr>
::with_either<ok_atom>
::or_else<error_atom, std::string>,
replies_to<delete_atom, actor_addr, uint16_t>
::with_either<ok_atom>
::or_else<error_atom, std::string>>;
/**
* Returns a handle for asynchronous networking operations.
*/
middleman_actor get_middleman_actor(); // implemented in middleman.cpp
} // namespace io
} // namespace caf
#endif // CAF_IO_MIDDLEMAN_ACTOR_HPP
......@@ -24,9 +24,9 @@
#include <string>
#include <cstdint>
#include "caf/fwd.hpp"
#include "caf/actor_cast.hpp"
#include "caf/io/middleman.hpp"
#include "caf/typed_actor.hpp"
namespace caf {
namespace io {
......
......@@ -29,29 +29,19 @@
namespace caf {
namespace io {
void unpublish_impl(abstract_actor_ptr whom, uint16_t port, bool block_caller);
void unpublish_impl(const actor_addr& 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) {
template <class Handle>
void unpublish(const Handle& 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);
unpublish_impl(whom.address(), port, true);
}
} // namespace io
......
......@@ -189,6 +189,23 @@ behavior basp_broker::make_behavior() {
ctx.handshake_data->expected_ifs.swap(expected_ifs);
init_handshake_as_client(ctx);
},
[=](delete_atom, int64_t request_id, const actor_addr& whom, uint16_t port)
-> message {
if (whom == invalid_actor_addr) {
return make_message(error_atom::value, "whom == invalid_actor_addr");
}
auto ptr = actor_cast<abstract_actor_ptr>(whom);
if (port == 0) {
if (!remove_published_actor(ptr)) {
return make_message(error_atom::value, "no mapping found");
}
} else {
if (!remove_published_actor(ptr, port)) {
return make_message(error_atom::value, "port not bound to actor");
}
}
return make_message(ok_atom::value, request_id);
},
// catch-all error handler
others() >> [=] {
CAF_LOG_ERROR("received unexpected message: "
......@@ -729,34 +746,52 @@ void basp_broker::add_published_actor(accept_handle hdl,
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);
unpublish_impl(self->address(), 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,
bool basp_broker::remove_published_actor(const abstract_actor_ptr& whom) {
size_t erased_elements = 0;
auto last = m_acceptors.end();
auto i = m_acceptors.begin();
while (i != last) {
if (i->second.first == whom) {
close(i->first);
m_open_ports.erase(i->second.second);
i = m_acceptors.erase(i);
}
else {
++i;
}
}
return erased_elements > 0;
}
bool 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;
return false;
}
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;
return false;
}
if (j->second.first != whom) {
CAF_LOG_INFO("port has been bound to a different actor already");
return;
return false;
}
close(j->first);
m_open_ports.erase(i);
m_acceptors.erase(j);
return true;
}
} // namespace io
......
......@@ -176,6 +176,7 @@ void do_announce(const char* tname) {
using detail::make_counted;
using middleman_actor_base = middleman_actor::extend<
reacts_to<ok_atom, int64_t>,
reacts_to<ok_atom, int64_t, actor_addr>,
reacts_to<error_atom, int64_t, std::string>
>::type;
......@@ -184,8 +185,8 @@ class middleman_actor_impl : public middleman_actor_base::base {
public:
middleman_actor_impl(middleman& mref, actor default_broker)
: m_broker(default_broker),
m_next_request_id(0),
m_parent(mref) {
m_parent(mref),
m_next_request_id(0) {
// nop
}
......@@ -196,6 +197,10 @@ class middleman_actor_impl : public middleman_actor_base::base {
using get_op_promise = typed_response_promise<get_op_result>;
using del_op_result = either<ok_atom>::or_else<error_atom, std::string>;
using del_op_promise = typed_response_promise<del_op_result>;
middleman_actor_base::behavior_type make_behavior() {
return {
[=](put_atom, const actor_addr& whom, uint16_t port,
......@@ -219,13 +224,28 @@ class middleman_actor_impl : public middleman_actor_base::base {
[=](get_atom, const std::string& hostname, uint16_t port) {
return get(hostname, port, std::set<std::string>());
},
[=](delete_atom, const actor_addr& whom) {
return del(whom);
},
[=](delete_atom, const actor_addr& whom, uint16_t port) {
return del(whom, port);
},
[=](ok_atom ok, int64_t request_id) {
auto i = m_pending_requests.find(request_id);
if (i == m_pending_requests.end()) {
CAF_LOG_ERROR("invalid request id: " << request_id);
return;
}
i->second.deliver(del_op_result{ok}.value);
m_pending_requests.erase(i);
},
[=](ok_atom ok, int64_t request_id, actor_addr result) {
auto i = m_pending_requests.find(request_id);
if (i == m_pending_requests.end()) {
CAF_LOG_ERROR("invalid request id: " << request_id);
return;
}
i->second.deliver(get_op_result{ok, result});
i->second.deliver(get_op_result{ok, result}.value);
m_pending_requests.erase(i);
},
[=](error_atom error, int64_t request_id, std::string& reason) {
......@@ -234,7 +254,7 @@ class middleman_actor_impl : public middleman_actor_base::base {
CAF_LOG_ERROR("invalid request id: " << request_id);
return;
}
i->second.deliver(get_op_result{error, std::move(reason)});
i->second.deliver(get_op_result{error, std::move(reason)}.value);
m_pending_requests.erase(i);
}
};
......@@ -267,7 +287,7 @@ class middleman_actor_impl : public middleman_actor_base::base {
get_op_promise get(const std::string& hostname, uint16_t port,
std::set<std::string> expected_ifs) {
get_op_promise result = make_response_promise();
auto result = make_response_promise();
try {
auto hdl = m_parent.backend().new_tcp_scribe(hostname, port);
auto req_id = m_next_request_id++;
......@@ -279,15 +299,23 @@ class middleman_actor_impl : public middleman_actor_base::base {
// fullfil promise immediately
std::string msg = "network_error: ";
msg += err.what();
result.deliver(get_op_result{error_atom{}, std::move(msg)});
result.deliver(get_op_result{error_atom{}, std::move(msg)}.value);
}
return result;
}
del_op_promise del(const actor_addr& whom, uint16_t port = 0) {
auto result = make_response_promise();
auto req_id = m_next_request_id++;
send(m_broker, delete_atom::value, req_id, whom, port);
m_pending_requests.insert(std::make_pair(req_id, result));
return result;
}
actor m_broker;
int64_t m_next_request_id;
middleman& m_parent;
std::map<int64_t, get_op_promise> m_pending_requests;
int64_t m_next_request_id;
std::map<int64_t, response_promise> m_pending_requests;
};
middleman_actor_impl::~middleman_actor_impl() {
......
......@@ -27,8 +27,8 @@
#include "caf/detail/singletons.hpp"
#include "caf/detail/actor_registry.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/middleman_actor.hpp"
namespace caf {
namespace io {
......
......@@ -32,8 +32,8 @@
#include "caf/abstract_actor.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/middleman_actor.hpp"
#include "caf/io/network/multiplexer.hpp"
#include "caf/detail/logging.hpp"
......
......@@ -22,32 +22,27 @@
#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"
#include "caf/io/middleman_actor.hpp"
namespace caf {
namespace io {
void unpublish_impl(abstract_actor_ptr whom, uint16_t port, bool block_caller) {
auto mm = middleman::instance();
void unpublish_impl(const actor_addr& whom, uint16_t port, bool block_caller) {
auto mm = get_middleman_actor();
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")) >> [] {
self->sync_send(mm, delete_atom::value, whom, port).await(
[](ok_atom) {
// ok, basp_broker is done
},
[](error_atom, const std::string&) {
// 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);
});
anon_send(mm, delete_atom::value, whom, port);
}
}
......
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