Commit 8bb07092 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'develop'

parents 33424d0f 77bda8b5
Subproject commit e63eeb168550f87095218ab8232b712b322155cb
Subproject commit ed985aed8bbd0f3eb6c149beff01a5a32276433f
Subproject commit 248c9beed366e856f60a8167671816f3688ec4dc
Subproject commit 8d709e6da3796161680c726802ea6152e88eaf08
......@@ -39,6 +39,7 @@
#include "caf/abstract_channel.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/functor_attachable.hpp"
namespace caf {
......@@ -90,16 +91,7 @@ class abstract_actor : public abstract_channel {
*/
template <class F>
void attach_functor(F f) {
struct functor_attachable : attachable {
F m_functor;
functor_attachable(F arg) : m_functor(std::move(arg)) {
// nop
}
void actor_exited(abstract_actor*, uint32_t reason) {
m_functor(reason);
}
};
attach(attachable_ptr{new functor_attachable(std::move(f))});
attach(attachable_ptr{new detail::functor_attachable<F>(std::move(f))});
}
/**
......
......@@ -96,6 +96,8 @@
# if LINUX_VERSION_CODE <= KERNEL_VERSION(2,6,16)
# define CAF_POLL_IMPL
# endif
#elif defined(__FreeBSD__)
# define CAF_BSD
#elif defined(WIN32) || defined(_WIN32)
# define CAF_WINDOWS
#else
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 LICENCE_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_FUNCTOR_ATTACHABLE_HPP
#define CAF_FUNCTOR_ATTACHABLE_HPP
#include "caf/attachable.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
namespace detail {
template <class F,
int Args = tl_size<typename get_callable_trait<F>::arg_types>::value>
struct functor_attachable : attachable {
static_assert(Args == 2, "Only 1 and 2 arguments for F are supported");
F m_functor;
functor_attachable(F arg) : m_functor(std::move(arg)) {
// nop
}
void actor_exited(abstract_actor* self, uint32_t reason) override {
m_functor(self, reason);
}
};
template <class F>
struct functor_attachable<F, 1> : attachable {
F m_functor;
functor_attachable(F arg) : m_functor(std::move(arg)) {
// nop
}
void actor_exited(abstract_actor*, uint32_t reason) override {
m_functor(reason);
}
};
} // namespace detail
} // namespace caf
#endif // CAF_FUNCTOR_ATTACHABLE_HPP
......@@ -29,48 +29,49 @@ namespace caf {
/**
* SI time units to specify timeouts.
* @relates duration
*/
enum class time_unit : uint32_t {
invalid = 0,
seconds = 1,
milliseconds = 1000,
microseconds = 1000000
};
// minutes are implicitly converted to seconds
/**
* Converts the ratio Num/Denom to a `time_unit` if the ratio describes
* seconds, milliseconds, microseconds, or minutes. Minutes are mapped
* to `time_unit::seconds`, any unrecognized ratio to `time_unit::invalid`.
* @relates duration
*/
template <intmax_t Num, intmax_t Denom>
struct ratio_to_time_unit_helper {
static constexpr time_unit value = time_unit::invalid;
};
template <>
struct ratio_to_time_unit_helper<1, 1> {
static constexpr time_unit value = time_unit::seconds;
};
template <>
struct ratio_to_time_unit_helper<1, 1000> {
static constexpr time_unit value = time_unit::milliseconds;
};
template <>
struct ratio_to_time_unit_helper<1, 1000000> {
static constexpr time_unit value = time_unit::microseconds;
};
template <>
struct ratio_to_time_unit_helper<60, 1> {
static constexpr time_unit value = time_unit::seconds;
};
/**
* Converts an STL time period to a `time_unit`.
* @relates duration
*/
template <class Period>
constexpr time_unit get_time_unit_from_period() {
......@@ -81,12 +82,14 @@ constexpr time_unit get_time_unit_from_period() {
* Time duration consisting of a `time_unit` and a 64 bit unsigned integer.
*/
class duration {
public:
constexpr duration() : unit(time_unit::invalid), count(0) {
// nop
}
constexpr duration() : unit(time_unit::invalid), count(0) {}
constexpr duration(time_unit u, uint32_t v) : unit(u), count(v) {}
constexpr duration(time_unit u, uint32_t v) : unit(u), count(v) {
// nop
}
/**
* Creates a new instance from an STL duration.
......@@ -94,54 +97,43 @@ class duration {
*/
template <class Rep, class Period>
duration(std::chrono::duration<Rep, Period> d)
: unit(get_time_unit_from_period<Period>()), count(rd(d)) {
: unit(get_time_unit_from_period<Period>()),
count(rd(d)) {
static_assert(get_time_unit_from_period<Period>() != time_unit::invalid,
"caf::duration supports only minutes, seconds, "
"milliseconds or microseconds");
"only minutes, seconds, milliseconds and "
"microseconds are supported");
static_assert(std::is_integral<Rep>::value,
"only integral durations are supported");
}
/**
* Creates a new instance from an STL duration given in minutes.
* @throws std::invalid_argument Thrown if `d.count() is negative.
*/
template <class Rep>
duration(std::chrono::duration<Rep, std::ratio<60, 1>> d)
: unit(time_unit::seconds), count(rd(d) * 60) {}
/**
* Returns `unit != time_unit::invalid`.
*/
inline bool valid() const { return unit != time_unit::invalid; }
inline bool valid() const {
return unit != time_unit::invalid;
}
/**
* Returns `count == 0`.
*/
inline bool is_zero() const { return count == 0; }
std::string to_string() const;
inline bool is_zero() const {
return count == 0;
}
time_unit unit;
uint64_t count;
private:
// reads d.count and throws invalid_argument if d.count < 0
template <class Rep, class Period>
static inline uint64_t rd(const std::chrono::duration<Rep, Period>& d) {
template <class Rep, intmax_t Num, intmax_t D>
static uint64_t rd(const std::chrono::duration<Rep, std::ratio<Num, D>>& d) {
// assertion (via ctors): Num == 1 || (Num == 60 && D == 1)
if (d.count() < 0) {
throw std::invalid_argument("negative durations are not supported");
}
return static_cast<uint64_t>(d.count());
}
template <class Rep>
static inline uint64_t
rd(const std::chrono::duration<Rep, std::ratio<60, 1>>& d) {
// convert minutes to seconds on-the-fly
return rd(std::chrono::duration<Rep, std::ratio<1, 1>>(d.count()) * 60);
return static_cast<uint64_t>(d.count()) * static_cast<uint64_t>(Num);
}
};
/**
......@@ -156,32 +148,28 @@ inline bool operator!=(const duration& lhs, const duration& rhs) {
return !(lhs == rhs);
}
} // namespace caf
/**
* @relates caf::duration
* @relates duration
*/
template <class Clock, class Duration>
std::chrono::time_point<Clock, Duration>&
operator+=(std::chrono::time_point<Clock, Duration>& lhs,
const caf::duration& rhs) {
operator+=(std::chrono::time_point<Clock, Duration>& lhs, const duration& rhs) {
switch (rhs.unit) {
case caf::time_unit::seconds:
case time_unit::seconds:
lhs += std::chrono::seconds(rhs.count);
break;
case caf::time_unit::milliseconds:
case time_unit::milliseconds:
lhs += std::chrono::milliseconds(rhs.count);
break;
case caf::time_unit::microseconds:
case time_unit::microseconds:
lhs += std::chrono::microseconds(rhs.count);
break;
case caf::time_unit::invalid:
case time_unit::invalid:
break;
}
return lhs;
}
} // namespace caf
#endif // CAF_DURATION_HPP
......@@ -67,6 +67,10 @@ class single_timeout : public Base {
return waits_for_timeout(tid);
}
uint32_t active_timeout_id() const {
return m_timeout_id;
}
void reset_timeout() {
this->has_timeout(false);
}
......
......@@ -112,11 +112,20 @@ class event_based_resume {
return resume_result::done;
}
}
auto had_tout = d->has_timeout();
auto tout = d->active_timeout_id();
int handled_msgs = 0;
auto reset_timeout_if_needed = [&] {
if (had_tout && handled_msgs > 0 && tout == d->active_timeout_id()) {
d->request_timeout(d->get_behavior().timeout());
}
};
// max_throughput = 0 means infinite
for (size_t i = 0; i < max_throughput; ++i) {
auto ptr = d->next_message();
if (ptr) {
if (d->invoke_message(ptr)) {
++handled_msgs;
if (actor_done()) {
CAF_LOG_DEBUG("actor exited");
return resume_result::done;
......@@ -140,12 +149,14 @@ class event_based_resume {
} else {
CAF_LOG_DEBUG("no more element in mailbox; going to block");
if (d->mailbox().try_block()) {
reset_timeout_if_needed();
return resumable::awaiting_message;
}
CAF_LOG_DEBUG("try_block() interrupted by new message");
}
}
if (!d->has_next_message() && d->mailbox().try_block()) {
reset_timeout_if_needed();
return resumable::awaiting_message;
}
// time's up
......
......@@ -23,40 +23,8 @@
namespace caf {
namespace {
inline uint64_t ui64_val(const duration& d) {
return static_cast<uint64_t>(d.unit) * d.count;
}
} // namespace <anonmyous>
bool operator==(const duration& lhs, const duration& rhs) {
return (lhs.unit == rhs.unit ? lhs.count == rhs.count
: ui64_val(lhs) == ui64_val(rhs));
}
std::string duration::to_string() const {
if (unit == time_unit::invalid) {
return "-invalid-";
}
std::ostringstream oss;
oss << count;
switch (unit) {
case time_unit::invalid:
oss << "?";
break;
case time_unit::seconds:
oss << "s";
break;
case time_unit::milliseconds:
oss << "ms";
break;
case time_unit::microseconds:
oss << "us";
break;
}
return oss.str();
return lhs.unit == rhs.unit && lhs.count == rhs.count;
}
} // namespace caf
#include "caf/config.hpp"
#include "caf/detail/get_mac_addresses.hpp"
#ifdef CAF_MACOS
#if defined(CAF_MACOS) || defined(CAF_BSD)
#include <sys/types.h>
#include <sys/socket.h>
......
......@@ -26,7 +26,7 @@ constexpr char uuid_format[] = "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF";
} // namespace <anonmyous>
#endif // CAF_MACOS
#ifdef CAF_MACOS
#if defined(CAF_MACOS)
namespace {
......@@ -61,23 +61,15 @@ std::string get_root_uuid() {
} // namespace detail
} // namespace caf
#elif defined(CAF_LINUX)
#elif defined(CAF_LINUX) || defined(CAF_BSD)
#include <vector>
#include <string>
#include <cctype>
#include <fstream>
#include <sstream>
#include <iostream>
#include <iterator>
#include <algorithm>
#include <stdio.h>
#include <sys/ioctl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <net/if.h>
#include <cstring>
#include <unistd.h>
#include <iostream>
#include "caf/string_algorithms.hpp"
......@@ -123,49 +115,6 @@ bool operator!=(const columns_iterator& lhs, const columns_iterator& rhs) {
} // namespace <anonymous>
std::string get_root_uuid() {
int sck = socket(AF_INET, SOCK_DGRAM, 0);
if (sck < 0) {
perror("socket");
return "";
}
// query available interfaces
char buf[1024];
ifconf ifc;
ifc.ifc_len = sizeof(buf);
ifc.ifc_buf = buf;
if (ioctl(sck, SIOCGIFCONF, &ifc) < 0) {
perror("ioctl(SIOCGIFCONF)");
return "";
}
vector<string> hw_addresses;
auto ctoi = [](char c)->unsigned {
return static_cast<unsigned char>(c);
};
// iterate through interfaces.
auto ifr = ifc.ifc_req;
size_t num_ifs = ifc.ifc_len / sizeof(ifreq);
for (size_t i = 0; i < num_ifs; i++) {
auto& item = ifr[i];
// get MAC address
if (ioctl(sck, SIOCGIFHWADDR, &item) < 0) {
perror("ioctl(SIOCGIFHWADDR)");
return "";
}
// convert MAC address to standard string representation
std::ostringstream oss;
oss << std::hex;
oss.width(2);
oss << ctoi(item.ifr_hwaddr.sa_data[0]);
for (size_t i = 1; i < 6; ++i) {
oss << ":";
oss.width(2);
oss << ctoi(item.ifr_hwaddr.sa_data[i]);
}
auto addr = oss.str();
if (addr != "00:00:00:00:00:00") {
hw_addresses.push_back(std::move(addr));
}
}
string uuid;
ifstream fs;
fs.open("/etc/fstab", std::ios_base::in);
......
......@@ -13,11 +13,14 @@ 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.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)
......@@ -52,4 +55,4 @@ include_directories(. ${INCLUDE_DIRS})
if(NOT WIN32)
install(DIRECTORY caf/ DESTINATION include/caf FILES_MATCHING PATTERN "*.hpp")
install(DIRECTORY cppa/ DESTINATION include/cppa FILES_MATCHING PATTERN "*.hpp")
endif()
\ No newline at end of file
endif()
......@@ -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;
......@@ -72,7 +67,6 @@ class basp_broker : public broker, public actor_namespace::backend {
struct client_handshake_data {
id_type remote_id;
std::promise<abstract_actor_ptr>* result;
std::string* error_msg;
const std::set<std::string>* expected_ifs;
};
......@@ -203,8 +197,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
......
......@@ -28,12 +28,26 @@
#include "caf/io/middleman.hpp"
#include "caf/io/remote_actor_impl.hpp"
#include "caf/io/typed_remote_actor_helper.hpp"
namespace caf {
namespace io {
abstract_actor_ptr remote_actor_impl(const std::set<std::string>& ifs,
const std::string& host, uint16_t port);
template <class List>
struct typed_remote_actor_helper;
template <template <class...> class List, class... Ts>
struct typed_remote_actor_helper<List<Ts...>> {
using return_type = typed_actor<Ts...>;
template <class... Vs>
return_type operator()(Vs&&... vs) {
auto iface = return_type::message_types();
auto tmp = remote_actor_impl(std::move(iface), std::forward<Vs>(vs)...);
return actor_cast<return_type>(tmp);
}
};
/**
* Establish a new connection to the actor at `host` on given `port`.
* @param host Valid hostname or IP address.
......
......@@ -17,36 +17,44 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_TYPED_REMOTE_ACTOR_HELPER_HPP
#define CAF_IO_TYPED_REMOTE_ACTOR_HELPER_HPP
#ifndef CAF_IO_UNPUBLISH_HPP
#define CAF_IO_UNPUBLISH_HPP
#include "caf/actor_cast.hpp"
#include "caf/typed_actor.hpp"
#include <cstdint>
#include "caf/actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/typed_actor.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/io/remote_actor_impl.hpp"
namespace caf {
namespace io {
template <class List>
struct typed_remote_actor_helper;
template <template <class...> class List, class... Ts>
struct typed_remote_actor_helper<List<Ts...>> {
using return_type = typed_actor<Ts...>;
template <class... Vs>
return_type operator()(Vs&&... vs) {
auto iface = return_type::message_types();
auto tmp = remote_actor_impl(std::move(iface), std::forward<Vs>(vs)...);
return actor_cast<return_type>(tmp);
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_TYPED_REMOTE_ACTOR_HELPER_HPP
#endif // CAF_IO_UNPUBLISH_HPP
......@@ -17,6 +17,7 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/exception.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp"
......@@ -26,6 +27,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,13 +63,21 @@ 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) {
CAF_LOGM_TRACE("make_behavior$connection_closed_msg",
CAF_MARG(msg.handle, id));
CAF_REQUIRE(m_ctx.count(msg.handle) > 0);
auto j = m_ctx.find(msg.handle);
if (j != m_ctx.end()) {
auto hd = j->second.handshake_data;
if (hd) {
network_error err{"disconnect during handshake"};
hd->result->set_exception(std::make_exception_ptr(err));
}
m_ctx.erase(j);
}
// purge handle from all routes
std::vector<id_type> lost_connections;
for (auto& kvp : m_routes) {
......@@ -96,12 +106,19 @@ behavior basp_broker::make_behavior() {
p->kill_proxy(exit_reason::remote_link_unreachable);
}
}
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,
......@@ -430,6 +447,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
auto e = what.end();
tmp += *i++;
while (i != e) {
tmp += ",";
tmp += *i++;
}
tmp += ">";
......@@ -437,7 +455,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
};
auto iface_str = tostr(remote_ifs);
auto expected_str = tostr(ifs);
auto& error_msg = *(ctx.handshake_data->error_msg);
std::string error_msg;
if (ifs.empty()) {
error_msg = "expected remote actor to be a "
"dynamically typed actor but found "
......@@ -457,7 +475,8 @@ basp_broker::handle_basp_header(connection_context& ctx,
+ iface_str;
}
// abort with error
ctx.handshake_data->result->set_value(nullptr);
std::runtime_error err{error_msg};
ctx.handshake_data->result->set_exception(std::make_exception_ptr(err));
return close_connection;
}
auto nid = ctx.handshake_data->remote_id;
......@@ -646,17 +665,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
......@@ -17,8 +17,7 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_REMOTE_ACTOR_IMPL_HPP
#define CAF_IO_REMOTE_ACTOR_IMPL_HPP
#include "caf/io/remote_actor.hpp"
#include <set>
#include <ios>
......@@ -40,14 +39,11 @@
namespace caf {
namespace io {
template <class... Ts>
abstract_actor_ptr remote_actor_impl(const std::set<std::string>& ifs,
const std::string& host, uint16_t port) {
auto mm = middleman::instance();
std::string error_msg;
std::promise<abstract_actor_ptr> result_promise;
basp_broker::client_handshake_data hdata{invalid_node_id, &result_promise,
&error_msg, &ifs};
std::promise<abstract_actor_ptr> res;
basp_broker::client_handshake_data hdata{invalid_node_id, &res, &ifs};
mm->run_later([&] {
try {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
......@@ -55,17 +51,12 @@ abstract_actor_ptr remote_actor_impl(const std::set<std::string>& ifs,
bro->init_client(hdl, &hdata);
}
catch (...) {
result_promise.set_exception(std::current_exception());
res.set_exception(std::current_exception());
}
});
auto result = result_promise.get_future().get();
if (!result) {
throw std::runtime_error(error_msg);
}
return result;
return res.get_future().get();
}
} // namespace io
} // namespace caf
#endif // CAF_IO_REMOTE_ACTOR_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
Subproject commit 89533ab743a443f8a4addcba1880a00d30920240
Subproject commit 5a11746201ff5dca5306278f1b07384d0cb4345a
Subproject commit 3b46188499434d3c071b69036923e50f0023eb00
Subproject commit abcc82a94eb67ee453365dc3b158fc523dce9015
Subproject commit 29bb67062293cfe88b17212bdf37c5dab7729390
Subproject commit bd18ef316da3cb23a9a1eed3a127d093e510e6aa
......@@ -39,5 +39,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