Commit f4c5842b authored by Dominik Charousset's avatar Dominik Charousset

Surrender to clang-format

parent f9451ffc
......@@ -29,20 +29,23 @@ namespace decorator {
sequencer::sequencer(strong_actor_ptr f, strong_actor_ptr g,
message_types_set msg_types)
: monitorable_actor(actor_config{}.add_flag(is_actor_dot_decorator_flag)),
f_(std::move(f)),
g_(std::move(g)),
msg_types_(std::move(msg_types)) {
: monitorable_actor(actor_config{}.add_flag(is_actor_dot_decorator_flag)),
f_(std::move(f)),
g_(std::move(g)),
msg_types_(std::move(msg_types)) {
CAF_ASSERT(f_);
CAF_ASSERT(g_);
// composed actor has dependency on constituent actors by default;
// if either constituent actor is already dead upon establishing
// the dependency, the actor is spawned dead
f_->get()->attach(default_attachable::make_monitor(actor_cast<actor_addr>(f_),
address()));
if (g_ != f_)
g_->get()->attach(default_attachable::make_monitor(actor_cast<actor_addr>(g_),
address()));
auto monitor1 = default_attachable::make_monitor(actor_cast<actor_addr>(f_),
address());
f_->get()->attach(std::move(monitor1));
if (g_ != f_) {
auto monitor2 = default_attachable::make_monitor(actor_cast<actor_addr>(g_),
address());
g_->get()->attach(std::move(monitor2));
}
}
void sequencer::enqueue(mailbox_element_ptr what, execution_unit* context) {
......
......@@ -19,10 +19,10 @@
#include "caf/decorator/splitter.hpp"
#include "caf/actor_system.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/response_promise.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/default_attachable.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/response_promise.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
......@@ -47,21 +47,21 @@ behavior fan_out_fan_in(stateful_actor<splitter_state>* self,
// request().await() has LIFO ordering
for (auto i = workers.rbegin(); i != workers.rend(); ++i)
// TODO: maybe infer some useful timeout or use config parameter?
self->request(actor_cast<actor>(*i), infinite, msg).await(
[=]() {
// nop
},
[=](error& err) mutable {
if (err == sec::unexpected_response) {
self->state.result += std::move(err.context());
if (--self->state.pending == 0)
self->state.rp.deliver(std::move(self->state.result));
} else {
self->state.rp.deliver(err);
self->quit();
}
}
);
self->request(actor_cast<actor>(*i), infinite, msg)
.await(
[=]() {
// nop
},
[=](error& err) mutable {
if (err == sec::unexpected_response) {
self->state.result += std::move(err.context());
if (--self->state.pending == 0)
self->state.rp.deliver(std::move(self->state.result));
} else {
self->state.rp.deliver(err);
self->quit();
}
});
return delegated<message>{};
};
self->set_default_handler(f);
......@@ -74,10 +74,10 @@ behavior fan_out_fan_in(stateful_actor<splitter_state>* self,
splitter::splitter(std::vector<strong_actor_ptr> workers,
message_types_set msg_types)
: monitorable_actor(actor_config{}.add_flag(is_actor_dot_decorator_flag)),
num_workers(workers.size()),
workers_(std::move(workers)),
msg_types_(std::move(msg_types)) {
: monitorable_actor(actor_config{}.add_flag(is_actor_dot_decorator_flag)),
num_workers(workers.size()),
workers_(std::move(workers)),
msg_types_(std::move(msg_types)) {
// composed actor has dependency on constituent actors by default;
// if either constituent actor is already dead upon establishing
// the dependency, the actor is spawned dead
......
......@@ -20,8 +20,8 @@
#include "caf/detail/behavior_impl.hpp"
#include "caf/message_handler.hpp"
#include "caf/make_type_erased_tuple_view.hpp"
#include "caf/message_handler.hpp"
namespace caf {
namespace detail {
......@@ -43,9 +43,7 @@ public:
}
combinator(pointer p0, const pointer& p1)
: behavior_impl(p1->timeout()),
first(std::move(p0)),
second(p1) {
: behavior_impl(p1->timeout()), first(std::move(p0)), second(p1) {
// nop
}
......@@ -82,9 +80,7 @@ behavior_impl::~behavior_impl() {
}
behavior_impl::behavior_impl(duration tout)
: timeout_(tout),
begin_(nullptr),
end_(nullptr) {
: timeout_(tout), begin_(nullptr), end_(nullptr) {
// nop
}
......
......@@ -18,9 +18,9 @@
#include <iterator>
#include "caf/none.hpp"
#include "caf/local_actor.hpp"
#include "caf/detail/behavior_stack.hpp"
#include "caf/local_actor.hpp"
#include "caf/none.hpp"
namespace caf {
namespace detail {
......
......@@ -24,12 +24,10 @@ namespace caf {
namespace detail {
decorated_tuple::decorated_tuple(cow_ptr&& d, vector_type&& v)
: decorated_(std::move(d)),
mapping_(std::move(v)),
type_token_(0xFFFFFFFF) {
: decorated_(std::move(d)), mapping_(std::move(v)), type_token_(0xFFFFFFFF) {
CAF_ASSERT(mapping_.empty()
|| *(std::max_element(mapping_.begin(), mapping_.end()))
< static_cast<const cow_ptr&>(decorated_)->size());
|| *(std::max_element(mapping_.begin(), mapping_.end()))
< static_cast<const cow_ptr&>(decorated_)->size());
// calculate type token
for (unsigned long i : mapping_) {
type_token_ <<= 6;
......
......@@ -30,15 +30,13 @@ dynamic_message_data::dynamic_message_data() : type_token_(0xFFFFFFFF) {
}
dynamic_message_data::dynamic_message_data(elements&& data)
: elements_(std::move(data)),
type_token_(0xFFFFFFFF) {
: elements_(std::move(data)), type_token_(0xFFFFFFFF) {
for (auto& e : elements_)
add_to_type_token(e->type().first);
}
dynamic_message_data::dynamic_message_data(const dynamic_message_data& other)
: detail::message_data(other),
type_token_(0xFFFFFFFF) {
: detail::message_data(other), type_token_(0xFFFFFFFF) {
for (auto& e : other.elements_) {
add_to_type_token(e->type().first);
elements_.emplace_back(e->copy());
......
......@@ -23,21 +23,21 @@
#if defined(CAF_MACOS) || defined(CAF_BSD) || defined(CAF_IOS)
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <sys/sysctl.h>
#include <net/if.h>
#include <net/if_dl.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <cerrno>
#include <cstdio>
#include <cstdlib>
#include <memory>
#include <sstream>
# include <arpa/inet.h>
# include <cerrno>
# include <cstdio>
# include <cstdlib>
# include <memory>
# include <net/if.h>
# include <net/if_dl.h>
# include <netinet/in.h>
# include <sstream>
# include <sys/ioctl.h>
# include <sys/socket.h>
# include <sys/sysctl.h>
# include <sys/types.h>
#include <iostream>
# include <iostream>
namespace caf {
namespace detail {
......@@ -99,22 +99,21 @@ std::vector<iface_info> get_mac_addresses() {
#elif defined(CAF_LINUX) || defined(CAF_ANDROID) || defined(CAF_CYGWIN)
#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 <algorithm>
# include <cctype>
# include <cstring>
# include <fstream>
# include <iostream>
# include <iterator>
# include <net/if.h>
# include <sstream>
# include <stdio.h>
# include <string>
# include <sys/ioctl.h>
# include <sys/socket.h>
# include <sys/types.h>
# include <unistd.h>
# include <vector>
namespace caf {
namespace detail {
......@@ -122,17 +121,15 @@ namespace detail {
std::vector<iface_info> get_mac_addresses() {
// get a socket handle
int socktype = SOCK_DGRAM;
#ifdef SOCK_CLOEXEC
# ifdef SOCK_CLOEXEC
socktype |= SOCK_CLOEXEC;
#endif
# endif
int sck = socket(AF_INET, socktype, 0);
if (sck < 0) {
perror("socket");
return {};
}
auto g = make_scope_guard([&] {
close(sck);
});
auto g = make_scope_guard([&] { close(sck); });
// query available interfaces
char buf[1024] = {0};
ifconf ifc;
......@@ -143,9 +140,7 @@ std::vector<iface_info> get_mac_addresses() {
return {};
}
std::vector<iface_info> result;
auto ctoi = [](char c) -> unsigned {
return static_cast<unsigned char>(c);
};
auto ctoi = [](char c) -> unsigned { return static_cast<unsigned char>(c); };
// iterate through interfaces
auto ifr = ifc.ifc_req;
auto num_ifaces = static_cast<size_t>(ifc.ifc_len) / sizeof(ifreq);
......@@ -180,22 +175,22 @@ std::vector<iface_info> get_mac_addresses() {
// windows
#include <ws2tcpip.h>
#include <winsock2.h>
#include <iphlpapi.h>
# include <iphlpapi.h>
# include <winsock2.h>
# include <ws2tcpip.h>
#include <memory>
#include <vector>
#include <string>
#include <cctype>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <fstream>
#include <sstream>
#include <iostream>
#include <iterator>
#include <algorithm>
# include <algorithm>
# include <cctype>
# include <cstdio>
# include <cstdlib>
# include <cstring>
# include <fstream>
# include <iostream>
# include <iterator>
# include <memory>
# include <sstream>
# include <string>
# include <vector>
namespace {
......@@ -230,7 +225,7 @@ std::vector<iface_info> get_mac_addresses() {
// break condition
size_t iterations = 0;
do {
addresses.reset((IP_ADAPTER_ADDRESSES*)malloc(addresses_len));
addresses.reset((IP_ADAPTER_ADDRESSES*) malloc(addresses_len));
if (!addresses) {
perror("Memory allocation failed for IP_ADAPTER_ADDRESSES struct");
exit(1);
......
......@@ -20,7 +20,7 @@
#include "caf/config.hpp"
#ifdef CAF_WINDOWS
#ifdef CAF_WINDOWS
# include <windows.h>
#else
# include <sys/types.h>
......@@ -31,11 +31,11 @@ namespace caf {
namespace detail {
unsigned get_process_id() {
# ifdef CAF_WINDOWS
#ifdef CAF_WINDOWS
return GetCurrentProcessId();
# else
#else
return static_cast<unsigned>(getpid());
# endif
#endif
}
} // namespace detail
......
......@@ -16,13 +16,13 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#include "caf/detail/get_root_uuid.hpp"
#include "caf/config.hpp"
#ifndef CAF_MACOS // not needed on Mac OS X
namespace {
constexpr char uuid_format[] = "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF";
} // namespace <anonmyous>
} // namespace
#endif // CAF_MACOS
#if defined(CAF_MACOS)
......@@ -35,9 +35,8 @@ inline void erase_trailing_newline(std::string& str) {
}
}
constexpr const char* s_get_uuid =
"/usr/sbin/diskutil info / | "
"/usr/bin/awk '$0 ~ /UUID/ { print $3 }'";
constexpr const char* s_get_uuid = "/usr/sbin/diskutil info / | "
"/usr/bin/awk '$0 ~ /UUID/ { print $3 }'";
} // namespace
......@@ -62,28 +61,27 @@ std::string get_root_uuid() {
#elif defined(CAF_LINUX) || defined(CAF_BSD) || defined(CAF_CYGWIN)
#include <vector>
#include <string>
#include <fstream>
#include <sstream>
#include <iostream>
#include <iterator>
#include <algorithm>
#include <iostream>
# include <algorithm>
# include <fstream>
# include <iostream>
# include <iterator>
# include <sstream>
# include <string>
# include <vector>
#include "caf/string_algorithms.hpp"
# include "caf/string_algorithms.hpp"
using std::vector;
using std::string;
using std::ifstream;
using std::string;
using std::vector;
namespace caf {
namespace detail {
namespace {
struct columns_iterator : std::iterator<std::forward_iterator_tag,
vector<string>> {
struct columns_iterator
: std::iterator<std::forward_iterator_tag, vector<string>> {
columns_iterator(ifstream* s = nullptr) : fs(s) {
// nop
}
......@@ -145,12 +143,12 @@ std::string get_root_uuid() {
#elif defined(CAF_WINDOWS)
#include <string>
#include <iostream>
#include <algorithm>
# include <algorithm>
# include <iostream>
# include <string>
#include <windows.h>
#include <tchar.h>
# include <tchar.h>
# include <windows.h>
namespace caf {
namespace detail {
......@@ -167,8 +165,8 @@ void mv(std::string& lhs, std::string&& rhs) {
// if TCHAR is defined as WCHAR, we have to do unicode conversion
void mv(std::string& lhs, const std::basic_string<WCHAR>& rhs) {
auto size_needed = WideCharToMultiByte(CP_UTF8, 0, rhs.c_str(),
static_cast<int>(rhs.size()),
nullptr, 0, nullptr, nullptr);
static_cast<int>(rhs.size()), nullptr,
0, nullptr, nullptr);
lhs.resize(size_needed);
WideCharToMultiByte(CP_UTF8, 0, rhs.c_str(), static_cast<int>(rhs.size()),
&lhs[0], size_needed, nullptr, nullptr);
......@@ -210,12 +208,11 @@ std::string get_root_uuid() {
} // namespace detail
} // namespace caf
#elif defined(CAF_IOS) || defined(CAF_ANDROID)
// return a randomly-generated UUID on mobile devices
#include <random>
# include <random>
namespace caf {
namespace detail {
......
......@@ -26,7 +26,7 @@ namespace detail {
// -- abstract_ini_consumer ----------------------------------------------------
abstract_ini_consumer::abstract_ini_consumer(abstract_ini_consumer* parent)
: parent_(parent) {
: parent_(parent) {
// nop
}
......@@ -45,13 +45,12 @@ ini_list_consumer abstract_ini_consumer::begin_list() {
// -- map_consumer -------------------------------------------------------------
ini_map_consumer::ini_map_consumer(abstract_ini_consumer* parent)
: super(parent),
i_(xs_.end()) {
: super(parent), i_(xs_.end()) {
CAF_ASSERT(parent != nullptr);
}
ini_map_consumer::ini_map_consumer(ini_map_consumer&& other)
: super(other.parent()) {
: super(other.parent()) {
}
ini_map_consumer::~ini_map_consumer() {
......@@ -74,13 +73,12 @@ void ini_map_consumer::value_impl(config_value&& x) {
// -- ini_list_consumer --------------------------------------------------------
ini_list_consumer::ini_list_consumer(abstract_ini_consumer* parent)
: super(parent) {
: super(parent) {
CAF_ASSERT(parent != nullptr);
}
ini_list_consumer::ini_list_consumer(ini_list_consumer&& other)
: super(other.parent()),
xs_(std::move(other.xs_)) {
: super(other.parent()), xs_(std::move(other.xs_)) {
// nop
}
......@@ -95,7 +93,7 @@ void ini_list_consumer::value_impl(config_value&& x) {
// -- ini_value_consumer -------------------------------------------------------
ini_value_consumer::ini_value_consumer(abstract_ini_consumer* parent)
: super(parent) {
: super(parent) {
// nop
}
......@@ -107,16 +105,15 @@ void ini_value_consumer::value_impl(config_value&& x) {
ini_category_consumer::ini_category_consumer(ini_consumer* parent,
std::string category)
: super(parent),
category_(std::move(category)) {
: super(parent), category_(std::move(category)) {
CAF_ASSERT(parent_ != nullptr);
}
ini_category_consumer::ini_category_consumer(ini_category_consumer&& other)
: super(other.parent()),
category_(std::move(other.category_)),
xs_(std::move(other.xs_)),
current_key(std::move(other.current_key)) {
: super(other.parent()),
category_(std::move(other.category_)),
xs_(std::move(other.xs_)),
current_key(std::move(other.current_key)) {
CAF_ASSERT(parent_ != nullptr);
}
......@@ -169,13 +166,14 @@ void ini_consumer::value_impl(config_value&& x) {
using dict_type = config_value::dictionary;
auto dict = get_if<dict_type>(&x);
if (dict == nullptr) {
warnings_.emplace_back(make_error(pec::type_mismatch,
"expected a dictionary at top level"));
warnings_.emplace_back(
make_error(pec::type_mismatch, "expected a dictionary at top level"));
return;
}
if (current_key_ != "global") {
auto& dst = cfg_.emplace(current_key_, dict_type{}).first->second;
if (dict != nullptr && !dict->empty() && holds_alternative<dict_type>(dst)) {
if (dict != nullptr && !dict->empty()
&& holds_alternative<dict_type>(dst)) {
auto& dst_dict = get<dict_type>(dst);
// We need to "merge" values into the destination, because it can already
// contain any number of unrelated entries.
......
......@@ -36,14 +36,12 @@ merged_tuple::cow_ptr merged_tuple::make(message x, message y) {
else
mapping.emplace_back(0, i);
}
return cow_ptr{make_counted<merged_tuple>(std::move(data),
std::move(mapping))};
return cow_ptr{
make_counted<merged_tuple>(std::move(data), std::move(mapping))};
}
merged_tuple::merged_tuple(data_type xs, mapping_type ys)
: data_(std::move(xs)),
type_token_(0xFFFFFFFF),
mapping_(std::move(ys)) {
: data_(std::move(xs)), type_token_(0xFFFFFFFF), mapping_(std::move(ys)) {
CAF_ASSERT(!data_.empty());
CAF_ASSERT(!mapping_.empty());
// calculate type token
......
......@@ -38,4 +38,3 @@ const char octal_chars[9] = "01234567";
} // namespace parser
} // namespace detail
} // namespace caf
......@@ -21,9 +21,9 @@
#include "caf/config.hpp"
#if defined(CAF_LINUX) || defined(CAF_MACOS)
#include <unistd.h>
#include <cxxabi.h>
#include <sys/types.h>
# include <cxxabi.h>
# include <sys/types.h>
# include <unistd.h>
#endif
#include "caf/string_algorithms.hpp"
......@@ -32,7 +32,7 @@ namespace caf {
namespace detail {
void prettify_type_name(std::string& class_name) {
//replace_all(class_name, " ", "");
// replace_all(class_name, " ", "");
replace_all(class_name, "::", ".");
replace_all(class_name, "(anonymous namespace)", "ANON");
replace_all(class_name, ".__1.", "."); // gets rid of weird Clang-lib names
......@@ -58,15 +58,15 @@ void prettify_type_name(std::string& class_name) {
}
void prettify_type_name(std::string& class_name, const char* c_class_name) {
# if defined(CAF_LINUX) || defined(CAF_MACOS)
#if defined(CAF_LINUX) || defined(CAF_MACOS)
int stat = 0;
std::unique_ptr<char, decltype(free)*> real_class_name{nullptr, free};
auto tmp = abi::__cxa_demangle(c_class_name, nullptr, nullptr, &stat);
real_class_name.reset(tmp);
class_name = stat == 0 ? real_class_name.get() : c_class_name;
# else
#else
class_name = c_class_name;
# endif
#endif
prettify_type_name(class_name);
}
......
......@@ -27,10 +27,10 @@ namespace caf {
namespace detail {
private_thread::private_thread(scheduled_actor* self)
: self_destroyed_(false),
self_(self),
state_(active),
system_(self->system()) {
: self_destroyed_(false),
self_(self),
state_(active),
system_(self->system()) {
intrusive_ptr_add_ref(self->ctrl());
system_.inc_detached_threads();
}
......
......@@ -54,8 +54,8 @@
* *
\******************************************************************************/
#include <cstring>
#include "caf/detail/ripemd_160.hpp"
#include <cstring>
namespace {
......@@ -378,7 +378,7 @@ void MDfinish(dword* MDbuf, const byte* strptr, dword lswlen, dword mswlen) {
compress(MDbuf, X);
}
} // namespace <anonmyous>
} // namespace
namespace caf {
namespace detail {
......
......@@ -21,13 +21,13 @@
#include "caf/config.hpp"
#ifndef CAF_WINDOWS
#include <pthread.h>
# include <pthread.h>
#endif // CAF_WINDOWS
#if defined(CAF_LINUX)
#include <sys/prctl.h>
# include <sys/prctl.h>
#elif defined(CAF_BSD)
#include <pthread_np.h>
# include <pthread_np.h>
#endif // defined(...)
#include <thread>
......@@ -43,14 +43,14 @@ void set_thread_name(const char* name) {
#else // CAF_WINDOWS
static_assert(std::is_same<std::thread::native_handle_type, pthread_t>::value,
"std::thread not based on pthread_t");
#if defined(CAF_MACOS)
# if defined(CAF_MACOS)
pthread_setname_np(name);
#elif defined(CAF_LINUX)
# elif defined(CAF_LINUX)
prctl(PR_SET_NAME, name, 0, 0, 0);
#elif defined(CAF_BSD)
# elif defined(CAF_BSD)
pthread_set_name_np(pthread_self(), name);
#endif // defined(...)
#endif // CAF_WINDOWS
# endif // defined(...)
#endif // CAF_WINDOWS
}
} // namespace detail
......
......@@ -177,9 +177,8 @@ size_t simple_actor_clock::trigger_expired_timeouts() {
return result;
}
void
simple_actor_clock::add_schedule_entry(time_point t,
std::unique_ptr<ordinary_timeout> x) {
void simple_actor_clock::add_schedule_entry(
time_point t, std::unique_ptr<ordinary_timeout> x) {
auto aid = x->self->id();
auto type = x->type;
auto pred = [&](const actor_lookup_map::value_type& kvp) {
......
......@@ -16,14 +16,14 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/sec.hpp"
#include "caf/atom.hpp"
#include "caf/actor.hpp"
#include "caf/config.hpp"
#include "caf/actor_cast.hpp"
#include "caf/message_id.hpp"
#include "caf/atom.hpp"
#include "caf/config.hpp"
#include "caf/exit_reason.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/message_id.hpp"
#include "caf/sec.hpp"
#include "caf/system_messages.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
......
......@@ -68,8 +68,9 @@ void thread_safe_actor_clock::schedule_message(time_point t,
void thread_safe_actor_clock::schedule_message(time_point t, group target,
strong_actor_ptr sender,
message content) {
push(new group_msg(t, std::move(target), std::move(sender),
std::move(content)));
auto ptr = new group_msg(t, std::move(target), std::move(sender),
std::move(content));
push(ptr);
}
void thread_safe_actor_clock::cancel_all() {
......
......@@ -25,9 +25,7 @@ namespace caf {
namespace detail {
tick_emitter::tick_emitter()
: start_(duration_type{0}),
interval_(0),
last_tick_id_(0) {
: start_(duration_type{0}), interval_(0), last_tick_id_(0) {
// nop
}
......@@ -56,8 +54,8 @@ void tick_emitter::interval(duration_type x) {
size_t tick_emitter::timeouts(time_point now,
std::initializer_list<size_t> periods) {
CAF_LOG_TRACE(CAF_ARG(now) << CAF_ARG(periods)
<< CAF_ARG(interval_) << CAF_ARG(start_));
CAF_LOG_TRACE(CAF_ARG(now)
<< CAF_ARG(periods) << CAF_ARG(interval_) << CAF_ARG(start_));
CAF_ASSERT(now >= start_);
size_t result = 0;
auto f = [&](size_t tick) {
......
......@@ -40,8 +40,7 @@ bool match_atom_constant(const meta_element& me, const type_erased_tuple& xs,
return me.v == *reinterpret_cast<const atom_value*>(ptr);
}
bool try_match(const type_erased_tuple& xs,
pattern_iterator iter, size_t ps) {
bool try_match(const type_erased_tuple& xs, pattern_iterator iter, size_t ps) {
if (xs.size() != ps)
return false;
for (size_t i = 0; i < ps; ++i, ++iter)
......
......@@ -57,8 +57,9 @@ auto downstream_messages::id_of(mailbox_element& x) noexcept -> key_type {
bool downstream_messages::enabled(const nested_queue_type& q) noexcept {
auto congested = q.policy().handler->mgr->congested();
CAF_LOG_DEBUG_IF(congested, "path is congested:"
<< CAF_ARG2( "slot", q.policy().handler->slots.receiver));
CAF_LOG_DEBUG_IF(congested,
"path is congested:"
<< CAF_ARG2("slot", q.policy().handler->slots.receiver));
return !congested;
}
......@@ -70,4 +71,3 @@ auto downstream_messages::quantum(const nested_queue_type& q,
} // namespace policy
} // namespace caf
......@@ -35,26 +35,26 @@ work_stealing::~work_stealing() {
}
work_stealing::worker_data::worker_data(scheduler::abstract_coordinator* p)
: rengine(std::random_device{}()),
// no need to worry about wrap-around; if `p->num_workers() < 2`,
// `uniform` will not be used anyway
uniform(0, p->num_workers() - 2),
strategies{{
{CONFIG("aggressive-poll-attempts", aggressive_poll_attempts), 1,
CONFIG("aggressive-steal-interval", aggressive_steal_interval),
timespan{0}},
{CONFIG("moderate-poll-attempts", moderate_poll_attempts), 1,
CONFIG("moderate-steal-interval", moderate_steal_interval),
CONFIG("moderate-sleep-duration", moderate_sleep_duration)},
{1, 0, CONFIG("relaxed-steal-interval", relaxed_steal_interval),
CONFIG("relaxed-sleep-duration", relaxed_sleep_duration)}}} {
: rengine(std::random_device{}()),
// no need to worry about wrap-around; if `p->num_workers() < 2`,
// `uniform` will not be used anyway
uniform(0, p->num_workers() - 2),
strategies{
{{CONFIG("aggressive-poll-attempts", aggressive_poll_attempts), 1,
CONFIG("aggressive-steal-interval", aggressive_steal_interval),
timespan{0}},
{CONFIG("moderate-poll-attempts", moderate_poll_attempts), 1,
CONFIG("moderate-steal-interval", moderate_steal_interval),
CONFIG("moderate-sleep-duration", moderate_sleep_duration)},
{1, 0, CONFIG("relaxed-steal-interval", relaxed_steal_interval),
CONFIG("relaxed-sleep-duration", relaxed_sleep_duration)}}} {
// nop
}
work_stealing::worker_data::worker_data(const worker_data& other)
: rengine(std::random_device{}()),
uniform(other.uniform),
strategies(other.strategies) {
: rengine(std::random_device{}()),
uniform(other.uniform),
strategies(other.strategies) {
// nop
}
......
......@@ -18,14 +18,14 @@
#include "caf/scheduler/abstract_coordinator.hpp"
#include <ios>
#include <thread>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <fstream>
#include <ios>
#include <iostream>
#include <thread>
#include <unordered_map>
#include <condition_variable>
#include "caf/actor_ostream.hpp"
#include "caf/actor_system.hpp"
......@@ -51,7 +51,7 @@ namespace scheduler {
namespace {
using string_sink = std::function<void (std::string&&)>;
using string_sink = std::function<void(std::string&&)>;
// the first value is the use count, the last ostream_handle that
// decrements it to 0 removes the ostream pointer from the map
......@@ -127,7 +127,7 @@ string_sink make_sink(actor_system& sys, const std::string& fn, int flags) {
: std::ios_base::out);
if (fs->is_open())
return [fs](std::string&& out) { *fs << out; };
std::cerr << "cannot open file: " << fn << std::endl;
std::cerr << "cannot open file: " << fn << std::endl;
return nullptr;
}
......@@ -197,9 +197,7 @@ public:
flush(d, false);
}
},
[&](flush_atom, actor_id aid) {
flush(get_data(aid, false), true);
},
[&](flush_atom, actor_id aid) { flush(get_data(aid, false), true); },
[&](delete_atom, actor_id aid) {
auto data_ptr = get_data(aid, false);
if (data_ptr != nullptr) {
......@@ -218,8 +216,8 @@ public:
[&](exit_msg& em) {
fail_state(std::move(em.reason));
done = true;
}
).until([&] { return done; });
})
.until([&] { return done; });
}
const char* name() const override {
......@@ -271,10 +269,7 @@ void abstract_coordinator::stop_actors() {
}
abstract_coordinator::abstract_coordinator(actor_system& sys)
: next_worker_(0),
max_throughput_(0),
num_workers_(0),
system_(sys) {
: next_worker_(0), max_throughput_(0), num_workers_(0), system_(sys) {
// nop
}
......
......@@ -33,8 +33,7 @@ namespace {
class dummy_worker : public execution_unit {
public:
dummy_worker(test_coordinator* parent)
: execution_unit(&parent->system()),
parent_(parent) {
: execution_unit(&parent->system()), parent_(parent) {
// nop
}
......@@ -50,10 +49,7 @@ class dummy_printer : public monitorable_actor {
public:
dummy_printer(actor_config& cfg) : monitorable_actor(cfg) {
mh_.assign(
[&](add_atom, actor_id, const std::string& str) {
std::cout << str;
}
);
[&](add_atom, actor_id, const std::string& str) { std::cout << str; });
}
void enqueue(mailbox_element_ptr what, execution_unit*) override {
......@@ -82,8 +78,9 @@ void test_coordinator::start() {
dummy_worker worker{this};
actor_config cfg{&worker};
auto& sys = system();
utility_actors_[printer_id] = make_actor<dummy_printer, actor>(
sys.next_actor_id(), sys.node(), &sys, cfg);
utility_actors_[printer_id] = make_actor<dummy_printer,
actor>(sys.next_actor_id(),
sys.node(), &sys, cfg);
}
void test_coordinator::stop() {
......@@ -161,6 +158,5 @@ void test_coordinator::inline_all_enqueues_helper() {
after_next_enqueue([=] { inline_all_enqueues_helper(); });
}
} // namespace caf
} // namespace scheduler
} // namespace caf
......@@ -24,8 +24,7 @@
#include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
#define ERROR_HANDLER [&](error& err) { CAF_FAIL(system.render(err)); }
using namespace caf;
......@@ -33,12 +32,8 @@ namespace {
behavior testee(event_based_actor* self) {
return {
[](int v) {
return 2 * v;
},
[=] {
self->quit();
}
[](int v) { return 2 * v; },
[=] { self->quit(); },
};
}
......@@ -46,14 +41,14 @@ using first_stage = typed_actor<replies_to<int>::with<double, double>>;
using second_stage = typed_actor<replies_to<double, double>::with<double>>;
first_stage::behavior_type typed_first_stage() {
return [](int i) {
return std::make_tuple(i * 2.0, i * 4.0);
return {
[](int i) { return std::make_tuple(i * 2.0, i * 4.0); },
};
}
second_stage::behavior_type typed_second_stage() {
return [](double x, double y) {
return x * y;
return {
[](double x, double y) { return x * y; },
};
}
......@@ -140,14 +135,12 @@ CAF_TEST(request_response_promise) {
auto h = f * g;
anon_send_exit(h, exit_reason::kill);
CAF_CHECK(exited(h));
self->request(h, infinite, 1).receive(
[](int) {
CAF_CHECK(false);
},
[](error err) {
CAF_CHECK_EQUAL(err.code(), static_cast<uint8_t>(sec::request_receiver_down));
}
);
self->request(h, infinite, 1)
.receive([](int) { CAF_CHECK(false); },
[](error err) {
CAF_CHECK_EQUAL(err.code(), static_cast<uint8_t>(
sec::request_receiver_down));
});
}
// single composition of distinct actors
......@@ -155,25 +148,17 @@ CAF_TEST(dot_composition_1) {
auto first = system.spawn(typed_first_stage);
auto second = system.spawn(typed_second_stage);
auto first_then_second = second * first;
self->request(first_then_second, infinite, 42).receive(
[](double res) {
CAF_CHECK_EQUAL(res, (42 * 2.0) * (42 * 4.0));
},
ERROR_HANDLER
);
self->request(first_then_second, infinite, 42)
.receive([](double res) { CAF_CHECK_EQUAL(res, (42 * 2.0) * (42 * 4.0)); },
ERROR_HANDLER);
}
// multiple self composition
CAF_TEST(dot_composition_2) {
auto dbl_actor = system.spawn(testee);
auto dbl_x4_actor = dbl_actor * dbl_actor
* dbl_actor * dbl_actor;
self->request(dbl_x4_actor, infinite, 1).receive(
[](int v) {
CAF_CHECK_EQUAL(v, 16);
},
ERROR_HANDLER
);
auto dbl_x4_actor = dbl_actor * dbl_actor * dbl_actor * dbl_actor;
self->request(dbl_x4_actor, infinite, 1)
.receive([](int v) { CAF_CHECK_EQUAL(v, 16); }, ERROR_HANDLER);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -24,8 +24,7 @@
#include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
#define ERROR_HANDLER [&](error& err) { CAF_FAIL(system.render(err)); }
using namespace caf;
......@@ -36,19 +35,15 @@ using second_stage = typed_actor<replies_to<double, double>::with<double>,
replies_to<double>::with<double>>;
first_stage::behavior_type typed_first_stage() {
return [](double x) {
return std::make_tuple(x * 2.0, x * 4.0);
return {
[](double x) { return std::make_tuple(x * 2.0, x * 4.0); },
};
}
second_stage::behavior_type typed_second_stage() {
return {
[](double x, double y) {
return x * y;
},
[](double x) {
return 23.0 * x;
}
[](double x, double y) { return x * y; },
[](double x) { return 23.0 * x; },
};
}
......@@ -105,14 +100,14 @@ CAF_TEST(kill_second) {
CAF_TEST(untyped_splicing) {
init_untyped();
self->request(first_and_second, infinite, 42.0).receive(
[](double x, double y, double z) {
CAF_CHECK_EQUAL(x, (42.0 * 2.0));
CAF_CHECK_EQUAL(y, (42.0 * 4.0));
CAF_CHECK_EQUAL(z, (23.0 * 42.0));
},
ERROR_HANDLER
);
self->request(first_and_second, infinite, 42.0)
.receive(
[](double x, double y, double z) {
CAF_CHECK_EQUAL(x, (42.0 * 2.0));
CAF_CHECK_EQUAL(y, (42.0 * 4.0));
CAF_CHECK_EQUAL(z, (23.0 * 42.0));
},
ERROR_HANDLER);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -28,13 +28,14 @@ using caf::detail::limited_vector;
CAF_TEST(basics) {
using std::equal;
int arr1[] {1, 2, 3, 4};
limited_vector<int, 4> vec1 {1, 2, 3, 4};
limited_vector<int, 5> vec2 {4, 3, 2, 1};
int arr1[]{1, 2, 3, 4};
limited_vector<int, 4> vec1{1, 2, 3, 4};
limited_vector<int, 5> vec2{4, 3, 2, 1};
limited_vector<int, 4> vec3;
for (int i = 1; i <= 4; ++i) vec3.push_back(i);
limited_vector<int, 4> vec4 {1, 2};
limited_vector<int, 2> vec5 {3, 4};
for (int i = 1; i <= 4; ++i)
vec3.push_back(i);
limited_vector<int, 4> vec4{1, 2};
limited_vector<int, 2> vec5{3, 4};
vec4.insert(vec4.end(), vec5.begin(), vec5.end());
auto vec6 = vec4;
CAF_CHECK_EQUAL(vec1.size(), 4u);
......@@ -54,22 +55,22 @@ CAF_TEST(basics) {
CAF_CHECK(std::equal(vec4.begin(), vec4.end(), arr1));
CAF_CHECK(std::equal(vec6.begin(), vec6.end(), arr1));
CAF_CHECK(std::equal(vec6.begin(), vec6.end(), vec2.rbegin()));
limited_vector<int, 10> vec7 {5, 9};
limited_vector<int, 10> vec8 {1, 2, 3, 4};
limited_vector<int, 10> vec9 {6, 7, 8};
limited_vector<int, 10> vec7{5, 9};
limited_vector<int, 10> vec8{1, 2, 3, 4};
limited_vector<int, 10> vec9{6, 7, 8};
vec7.insert(vec7.begin() + 1, vec9.begin(), vec9.end());
vec7.insert(vec7.begin(), vec8.begin(), vec8.end());
CAF_CHECK_EQUAL(vec7.full(), false);
limited_vector<int, 1> vec10 {10};
limited_vector<int, 1> vec10{10};
vec7.insert(vec7.end(), vec10.begin(), vec10.end());
CAF_CHECK_EQUAL(vec7.full(), true);
CAF_CHECK((std::is_sorted(vec7.begin(), vec7.end())));
int arr2[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
int arr2[]{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
CAF_CHECK((std::equal(vec7.begin(), vec7.end(), std::begin(arr2))));
vec7.assign(std::begin(arr2), std::end(arr2));
CAF_CHECK((std::equal(vec7.begin(), vec7.end(), std::begin(arr2))));
vec7.assign(5, 0);
CAF_CHECK_EQUAL(vec7.size(), 5u);
CAF_CHECK((std::all_of(vec7.begin(), vec7.end(),
[](int i) { return i == 0; })));
CAF_CHECK(
(std::all_of(vec7.begin(), vec7.end(), [](int i) { return i == 0; })));
}
......@@ -32,7 +32,7 @@ namespace {
struct atom_parser_consumer {
atom_value x;
inline void value(atom_value y) {
void value(atom_value y) {
x = y;
}
};
......@@ -70,7 +70,7 @@ CAF_TEST(empty atom) {
CAF_CHECK_EQUAL(p("\t '' \t\t\t "), atom(""));
}
CAF_TEST(non-empty atom) {
CAF_TEST(valid atom) {
CAF_CHECK_EQUAL(p("'abc'"), atom("abc"));
CAF_CHECK_EQUAL(p("'a b c'"), atom("a b c"));
CAF_CHECK_EQUAL(p(" 'abcdef' "), atom("abcdef"));
......
......@@ -261,7 +261,7 @@ CAF_TEST(empty inis) {
CAF_CHECK_EQUAL(parse(";hello\n;world"), make_log());
}
CAF_TEST(section with valid key-value pairs) {
CAF_TEST(section with valid key - value pairs) {
CAF_CHECK_EQUAL(parse("[foo]"), make_log("key: foo", "{", "}"));
CAF_CHECK_EQUAL(parse(" [foo]"), make_log("key: foo", "{", "}"));
CAF_CHECK_EQUAL(parse(" [ foo] "), make_log("key: foo", "{", "}"));
......
......@@ -132,4 +132,3 @@ CAF_TEST(invalid timespans) {
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -57,7 +57,7 @@ struct fixture {
};
// TODO: remove and use "..."s from the STL when switching to C++14
std::string operator "" _s(const char* str, size_t str_len) {
std::string operator"" _s(const char* str, size_t str_len) {
std::string result;
result.assign(str, str_len);
return result;
......@@ -77,7 +77,7 @@ CAF_TEST(empty string) {
CAF_CHECK_EQUAL(p("\t \"\" \t\t\t "), ""_s);
}
CAF_TEST(non-empty quoted string) {
CAF_TEST(non - empty quoted string) {
CAF_CHECK_EQUAL(p(R"("abc")"), "abc"_s);
CAF_CHECK_EQUAL(p(R"("a b c")"), "a b c"_s);
CAF_CHECK_EQUAL(p(R"( "abcdefABCDEF" )"), "abcdefABCDEF"_s);
......
......@@ -72,8 +72,8 @@ CAF_TEST(ticks) {
auto cycle = detail::gcd(credit_interval.count(),
force_batch_interval.count());
CAF_CHECK_EQUAL(cycle, 50);
auto force_batch_frequency =
static_cast<size_t>(force_batch_interval.count() / cycle);
auto force_batch_frequency = static_cast<size_t>(force_batch_interval.count()
/ cycle);
auto credit_frequency = static_cast<size_t>(credit_interval.count() / cycle);
detail::tick_emitter tctrl{time_point{timespan{100}}};
tctrl.interval(timespan{cycle});
......@@ -149,4 +149,3 @@ CAF_TEST(next_timeout) {
next = tctrl.next_timeout(now, {5, 7});
CAF_CHECK_EQUAL(next, start + timespan((2 * 7) * interval));
}
......@@ -30,7 +30,7 @@ int fourty_two() {
return 42;
}
class instance_counting_wrapper final: public int_fun::wrapper {
class instance_counting_wrapper final : public int_fun::wrapper {
public:
instance_counting_wrapper(size_t* instance_counter)
: instance_counter_(instance_counter) {
......
......@@ -117,14 +117,14 @@ CAF_TEST(default_constructed) {
}
CAF_TEST(initializer_list_constructed) {
unordered_flat_map<int, int> zs{{1, 10}, {2, 20}, {3, 30}, {4, 40}};
unordered_flat_map<int, int> zs{{1, 10}, {2, 20}, {3, 30}, {4, 40}};
CAF_CHECK_EQUAL(zs.size(), 4u);
CAF_CHECK_EQUAL(zs, ivec({{1, 10}, {2, 20}, {3, 30}, {4, 40}}));
}
CAF_TEST(range_constructed) {
kvp_vec<int> tmp{{1, 10}, {2, 20}, {3, 30}, {4, 40}};
unordered_flat_map<int, int> zs(tmp.begin(), tmp.end());
unordered_flat_map<int, int> zs(tmp.begin(), tmp.end());
CAF_CHECK_EQUAL(zs.size(), 4u);
CAF_CHECK_EQUAL(zs, tmp);
}
......@@ -171,11 +171,9 @@ CAF_TEST(exceptions) {
try {
auto x = xs.at(10);
CAF_FAIL("got an unexpected value: " << x);
}
catch (std::out_of_range&) {
} catch (std::out_of_range&) {
CAF_MESSAGE("got expected out_of_range exception");
}
catch (...) {
} catch (...) {
CAF_FAIL("got an expected exception");
}
}
......
......@@ -169,7 +169,7 @@ CAF_TEST(take_front) {
}
CAF_TEST(alternating_consumer) {
using fun_type = std::function<task_result (inode&)>;
using fun_type = std::function<task_result(inode&)>;
fun_type f;
fun_type g;
fun_type* selected = &f;
......@@ -191,9 +191,7 @@ CAF_TEST(alternating_consumer) {
return task_result::resume;
};
/// Define a function object that alternates between f and g.
auto h = [&](inode& x) {
return (*selected)(x);
};
auto h = [&](inode& x) { return (*selected)(x); };
// Fill and consume queue, h leaves 9 in the cache since it reads (odd, even)
// sequences and no odd value to read after 7 is available.
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
......@@ -236,5 +234,4 @@ CAF_TEST(to_string) {
CAF_CHECK_EQUAL(deep_to_string(queue), "[1, 2, 3, 4]");
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -132,9 +132,7 @@ CAF_TEST(unblock) {
CAF_TEST(await) {
std::mutex mx;
std::condition_variable cv;
std::thread t{[&] {
inbox.synchronized_emplace_back(mx, cv, 1);
}};
std::thread t{[&] { inbox.synchronized_emplace_back(mx, cv, 1); }};
inbox.synchronized_await(mx, cv);
CAF_REQUIRE_EQUAL(close_and_fetch(), "1");
t.join();
......@@ -152,9 +150,7 @@ CAF_TEST(timed_await) {
CAF_REQUIRE_EQUAL(res, true);
CAF_CHECK_EQUAL(fetch(), "1");
tout += std::chrono::hours(1000);
std::thread t{[&] {
inbox.synchronized_emplace_back(mx, cv, 2);
}};
std::thread t{[&] { inbox.synchronized_emplace_back(mx, cv, 2); }};
res = inbox.synchronized_await(mx, cv, tout);
CAF_REQUIRE_EQUAL(res, true);
CAF_REQUIRE_EQUAL(close_and_fetch(), "2");
......
......@@ -125,9 +125,7 @@ CAF_TEST(unblock) {
CAF_TEST(await) {
std::mutex mx;
std::condition_variable cv;
std::thread t{[&] {
inbox.synchronized_emplace_front(mx, cv, 1);
}};
std::thread t{[&] { inbox.synchronized_emplace_front(mx, cv, 1); }};
inbox.synchronized_await(mx, cv);
CAF_REQUIRE_EQUAL(close_and_fetch(), "1");
t.join();
......@@ -145,9 +143,7 @@ CAF_TEST(timed_await) {
CAF_REQUIRE_EQUAL(res, true);
CAF_CHECK_EQUAL(fetch(), "1");
tout += std::chrono::hours(1000);
std::thread t{[&] {
inbox.synchronized_emplace_front(mx, cv, 2);
}};
std::thread t{[&] { inbox.synchronized_emplace_front(mx, cv, 2); }};
res = inbox.synchronized_await(mx, cv, tout);
CAF_REQUIRE_EQUAL(res, true);
CAF_REQUIRE_EQUAL(close_and_fetch(), "2");
......
......@@ -35,7 +35,7 @@ namespace {
struct inode : singly_linked<inode> {
int value;
inode(int x = 0) : value(x) {
inode(int x = 0) noexcept : value(x) {
// nop
}
};
......@@ -55,7 +55,7 @@ struct nested_inode_policy {
using unique_pointer = std::unique_ptr<mapped_type, deleter_type>;
static inline task_size_type task_size(const mapped_type&) {
static task_size_type task_size(const mapped_type&) noexcept {
return 1;
}
......@@ -65,8 +65,8 @@ struct nested_inode_policy {
// nop
}
nested_inode_policy(nested_inode_policy&&) = default;
nested_inode_policy& operator=(nested_inode_policy&&) = default;
nested_inode_policy(nested_inode_policy&&) noexcept = default;
nested_inode_policy& operator=(nested_inode_policy&&) noexcept = default;
};
struct inode_policy {
......@@ -86,15 +86,15 @@ struct inode_policy {
using queue_map_type = std::map<key_type, queue_type>;
static inline key_type id_of(const inode& x) {
static key_type id_of(const inode& x) noexcept {
return x.value % 3;
}
static inline bool enabled(const queue_type&) {
static bool enabled(const queue_type&) noexcept {
return true;
}
deficit_type quantum(const queue_type& q, deficit_type x) {
deficit_type quantum(const queue_type& q, deficit_type x) noexcept {
return enable_priorities && *q.policy().queue_id == 0 ? 2 * x : x;
}
......@@ -210,5 +210,4 @@ CAF_TEST(to_string) {
CAF_CHECK_EQUAL(deep_to_string(queue), "[3, 1, 4, 2]");
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -86,10 +86,8 @@ public:
using nested_queue_type = drr_queue<inode_policy>;
using queue_type = wdrr_fixed_multiplexed_queue<inode_policy,
high_prio_queue,
nested_queue_type,
nested_queue_type>;
using queue_type = wdrr_fixed_multiplexed_queue<
inode_policy, high_prio_queue, nested_queue_type, nested_queue_type>;
struct fetch_helper {
std::string result;
......
......@@ -59,7 +59,7 @@ struct mailbox_policy {
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
struct fixture{};
struct fixture {};
struct consumer {
std::vector<int> ints;
......@@ -75,7 +75,7 @@ struct consumer {
template <class Key, class Queue, class... Ts>
intrusive::task_result operator()(const Key&, const Queue&, const Ts&...) {
CAF_FAIL("unexepected message type");// << typeid(Ts).name());
CAF_FAIL("unexepected message type"); // << typeid(Ts).name());
return intrusive::task_result::resume;
}
};
......
......@@ -19,9 +19,9 @@
#include "caf/detail/socket_guard.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
# include <winsock2.h>
#else
# include <unistd.h>
# include <unistd.h>
#endif
#include "caf/logger.hpp"
......@@ -52,4 +52,4 @@ void socket_guard::close() {
}
} // namespace detail
} // namespace detail
} // namespace caf
......@@ -16,11 +16,11 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/none.hpp"
#include "caf/actor_system.hpp"
#include "caf/config.hpp"
#include "caf/logger.hpp"
#include "caf/actor_system.hpp"
#include "caf/make_counted.hpp"
#include "caf/none.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/io/broker.hpp"
......@@ -34,8 +34,8 @@
namespace caf {
namespace io {
void abstract_broker::enqueue(strong_actor_ptr src, message_id mid,
message msg, execution_unit*) {
void abstract_broker::enqueue(strong_actor_ptr src, message_id mid, message msg,
execution_unit*) {
enqueue(make_mailbox_element(std::move(src), mid, {}, std::move(msg)),
&backend());
}
......@@ -137,8 +137,7 @@ void abstract_broker::enqueue_datagram(datagram_handle hdl,
x->enqueue_datagram(hdl, std::move(buf));
}
void abstract_broker::write(datagram_handle hdl, size_t bs,
const void* buf) {
void abstract_broker::write(datagram_handle hdl, size_t bs, const void* buf) {
auto& out = wr_buf(hdl);
auto first = reinterpret_cast<const char*>(buf);
auto last = first + bs;
......@@ -225,7 +224,8 @@ void abstract_broker::add_hdl_for_datagram_servant(datagram_servant_ptr ptr,
get_map(hdl).emplace(hdl, std::move(ptr));
}
datagram_handle abstract_broker::add_datagram_servant(network::native_socket fd) {
datagram_handle
abstract_broker::add_datagram_servant(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
auto ptr = backend().new_datagram_servant(fd);
auto hdl = ptr->hdl();
......@@ -233,9 +233,8 @@ datagram_handle abstract_broker::add_datagram_servant(network::native_socket fd)
return hdl;
}
datagram_handle
abstract_broker::add_datagram_servant_for_endpoint(network::native_socket fd,
const network::ip_endpoint& ep) {
datagram_handle abstract_broker::add_datagram_servant_for_endpoint(
network::native_socket fd, const network::ip_endpoint& ep) {
CAF_LOG_TRACE(CAF_ARG(fd));
auto ptr = backend().new_datagram_servant_for_endpoint(fd, ep);
auto hdl = ptr->hdl();
......@@ -356,8 +355,8 @@ resumable::subtype_t abstract_broker::subtype() const {
return io_actor;
}
resumable::resume_result
abstract_broker::resume(execution_unit* ctx, size_t mt) {
resumable::resume_result abstract_broker::resume(execution_unit* ctx,
size_t mt) {
CAF_ASSERT(ctx != nullptr);
CAF_ASSERT(ctx == &backend());
return scheduled_actor::resume(ctx, mt);
......
......@@ -33,26 +33,20 @@ std::string to_bin(uint8_t x) {
return res;
}
std::string to_string(const header &hdr) {
std::string to_string(const header& hdr) {
std::ostringstream oss;
oss << "{"
<< to_string(hdr.operation) << ", "
<< to_bin(hdr.flags) << ", "
<< hdr.payload_len << ", "
<< hdr.operation_data << ", "
<< hdr.source_actor << ", "
<< hdr.dest_actor
<< "}";
oss << "{" << to_string(hdr.operation) << ", " << to_bin(hdr.flags) << ", "
<< hdr.payload_len << ", " << hdr.operation_data << ", "
<< hdr.source_actor << ", " << hdr.dest_actor << "}";
return oss.str();
}
bool operator==(const header& lhs, const header& rhs) {
return lhs.operation == rhs.operation
&& lhs.flags == rhs.flags
&& lhs.payload_len == rhs.payload_len
&& lhs.operation_data == rhs.operation_data
&& lhs.source_actor == rhs.source_actor
&& lhs.dest_actor == rhs.dest_actor;
return lhs.operation == rhs.operation && lhs.flags == rhs.flags
&& lhs.payload_len == rhs.payload_len
&& lhs.operation_data == rhs.operation_data
&& lhs.source_actor == rhs.source_actor
&& lhs.dest_actor == rhs.dest_actor;
}
namespace {
......
......@@ -33,7 +33,7 @@ namespace io {
namespace basp {
instance::callee::callee(actor_system& sys, proxy_registry::backend& backend)
: namespace_(sys, backend) {
: namespace_(sys, backend) {
// nop
}
......@@ -42,9 +42,7 @@ instance::callee::~callee() {
}
instance::instance(abstract_broker* parent, callee& lstnr)
: tbl_(parent),
this_node_(parent->system().node()),
callee_(lstnr) {
: tbl_(parent), this_node_(parent->system().node()), callee_(lstnr) {
CAF_ASSERT(this_node_ != none);
auto workers = get_or(config(), "middleman.workers",
defaults::middleman::workers);
......@@ -52,9 +50,8 @@ instance::instance(abstract_broker* parent, callee& lstnr)
hub_.add_new_worker(queue_, proxies());
}
connection_state instance::handle(execution_unit* ctx,
new_data_msg& dm, header& hdr,
bool is_payload) {
connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
header& hdr, bool is_payload) {
CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload));
// function object providing cleanup code on errors
auto err = [&]() -> connection_state {
......@@ -90,7 +87,7 @@ connection_state instance::handle(execution_unit* ctx,
void instance::handle_heartbeat(execution_unit* ctx) {
CAF_LOG_TRACE("");
for (auto& kvp: tbl_.direct_by_hdl_) {
for (auto& kvp : tbl_.direct_by_hdl_) {
CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second));
write_heartbeat(ctx, callee_.get_buffer(kvp.first));
callee_.flush(kvp.first);
......@@ -116,8 +113,8 @@ void instance::write(execution_unit* ctx, const routing_table::route& r,
void instance::add_published_actor(uint16_t port,
strong_actor_ptr published_actor,
std::set<std::string> published_interface) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(published_actor)
<< CAF_ARG(published_interface));
CAF_LOG_TRACE(CAF_ARG(port)
<< CAF_ARG(published_actor) << CAF_ARG(published_interface));
using std::swap;
auto& entry = published_actors_[port];
swap(entry.first, published_actor);
......@@ -136,8 +133,7 @@ size_t instance::remove_published_actor(uint16_t port,
return 1;
}
size_t instance::remove_published_actor(const actor_addr& whom,
uint16_t port,
size_t instance::remove_published_actor(const actor_addr& whom, uint16_t port,
removed_published_actor* cb) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port));
size_t result = 0;
......@@ -169,23 +165,30 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
const std::vector<strong_actor_ptr>& forwarding_stack,
const node_id& dest_node, uint64_t dest_actor,
uint8_t flags, message_id mid, const message& msg) {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(dest_node) << CAF_ARG(mid)
<< CAF_ARG(msg));
CAF_LOG_TRACE(CAF_ARG(sender)
<< CAF_ARG(dest_node) << CAF_ARG(mid) << CAF_ARG(msg));
CAF_ASSERT(dest_node && this_node_ != dest_node);
auto path = lookup(dest_node);
if (!path)
return false;
auto& source_node = sender ? sender->node() : this_node_;
if (dest_node == path->next_hop && source_node == this_node_) {
header hdr{message_type::direct_message, flags, 0, mid.integer_value(),
sender ? sender->id() : invalid_actor_id, dest_actor};
auto writer = make_callback([&](serializer& sink) -> error {
return sink(forwarding_stack, msg);
});
header hdr{message_type::direct_message,
flags,
0,
mid.integer_value(),
sender ? sender->id() : invalid_actor_id,
dest_actor};
auto writer = make_callback(
[&](serializer& sink) -> error { return sink(forwarding_stack, msg); });
write(ctx, callee_.get_buffer(path->hdl), hdr, &writer);
} else {
header hdr{message_type::routed_message, flags, 0, mid.integer_value(),
sender ? sender->id() : invalid_actor_id, dest_actor};
header hdr{message_type::routed_message,
flags,
0,
mid.integer_value(),
sender ? sender->id() : invalid_actor_id,
dest_actor};
auto writer = make_callback([&](serializer& sink) -> error {
return sink(source_node, dest_node, forwarding_stack, msg);
});
......@@ -195,8 +198,8 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
return true;
}
void instance::write(execution_unit* ctx, buffer_type& buf,
header& hdr, payload_writer* pw) {
void instance::write(execution_unit* ctx, buffer_type& buf, header& hdr,
payload_writer* pw) {
CAF_LOG_TRACE(CAF_ARG(hdr));
binary_serializer sink{ctx, buf};
if (pw != nullptr) {
......@@ -235,26 +238,32 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& out_buf,
}
return sink(this_node_, app_ids, aid, iface);
});
header hdr{message_type::server_handshake, 0, 0, version,
invalid_actor_id, invalid_actor_id};
header hdr{message_type::server_handshake,
0,
0,
version,
invalid_actor_id,
invalid_actor_id};
write(ctx, out_buf, hdr, &writer);
}
void instance::write_client_handshake(execution_unit* ctx, buffer_type& buf) {
auto writer = make_callback([&](serializer& sink) -> error {
return sink(this_node_);
});
header hdr{message_type::client_handshake, 0, 0, 0,
invalid_actor_id, invalid_actor_id};
auto writer = make_callback(
[&](serializer& sink) -> error { return sink(this_node_); });
header hdr{message_type::client_handshake,
0,
0,
0,
invalid_actor_id,
invalid_actor_id};
write(ctx, buf, hdr, &writer);
}
void instance::write_monitor_message(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid));
auto writer = make_callback([&](serializer& sink) -> error {
return sink(this_node_, dest_node);
});
auto writer = make_callback(
[&](serializer& sink) -> error { return sink(this_node_, dest_node); });
header hdr{message_type::monitor_message, 0, 0, 0, invalid_actor_id, aid};
write(ctx, buf, hdr, &writer);
}
......@@ -322,8 +331,8 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
}
// Close this connection if we already have a direct connection.
if (tbl_.lookup_direct(source_node)) {
CAF_LOG_DEBUG("close redundant direct connection:"
<< CAF_ARG(source_node));
CAF_LOG_DEBUG(
"close redundant direct connection:" << CAF_ARG(source_node));
callee_.finalize_handshake(source_node, aid, sigs);
return false;
}
......@@ -354,8 +363,8 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
}
// Drop repeated handshakes.
if (tbl_.lookup_direct(source_node)) {
CAF_LOG_DEBUG("received repeated client handshake:"
<< CAF_ARG(source_node));
CAF_LOG_DEBUG(
"received repeated client handshake:" << CAF_ARG(source_node));
break;
}
// Add direct route to this node and remove any indirect entry.
......
......@@ -27,13 +27,8 @@ namespace basp {
namespace {
const char* message_type_strings[] = {
"server_handshake",
"client_handshake",
"direct_message",
"routed_message",
"proxy_creation",
"proxy_destruction",
"heartbeat"
"server_handshake", "client_handshake", "direct_message", "routed_message",
"proxy_creation", "proxy_destruction", "heartbeat",
};
} // namespace
......
......@@ -24,8 +24,7 @@ namespace caf {
namespace io {
namespace basp {
routing_table::routing_table(abstract_broker* parent)
: parent_(parent) {
routing_table::routing_table(abstract_broker* parent) : parent_(parent) {
// nop
}
......
......@@ -16,12 +16,12 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/none.hpp"
#include "caf/config.hpp"
#include "caf/make_counted.hpp"
#include "caf/none.hpp"
#include "caf/logger.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/logger.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
......@@ -37,7 +37,7 @@ void broker::initialize() {
init_broker();
auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:"
<< CAF_ARG2("alive", alive()));
<< CAF_ARG2("alive", alive()));
if (bhvr) {
// make_behavior() did return a behavior instead of using become()
CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
......@@ -49,7 +49,6 @@ broker::broker(actor_config& cfg) : super(cfg) {
// nop
}
behavior broker::make_behavior() {
behavior res;
if (initial_behavior_fac_) {
......
......@@ -49,39 +49,37 @@ behavior connection_helper(stateful_actor<connection_helper_state>* self,
CAF_LOG_DEBUG("received requested config:" << CAF_ARG(msg));
// whatever happens, we are done afterwards
self->quit();
msg.apply({
[&](uint16_t port, network::address_listing& addresses) {
if (item == "basp.default-connectivity-tcp") {
auto& mx = self->system().middleman().backend();
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto hdl = mx.new_tcp_scribe(addr, port);
if (hdl) {
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(b, connect_atom::value, *hdl, port);
return;
}
msg.apply({[&](uint16_t port, network::address_listing& addresses) {
if (item == "basp.default-connectivity-tcp") {
auto& mx = self->system().middleman().backend();
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto hdl = mx.new_tcp_scribe(addr, port);
if (hdl) {
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(b, connect_atom::value, *hdl, port);
return;
}
}
CAF_LOG_INFO("could not connect to node directly");
} else {
CAF_LOG_INFO("aborted direct connection attempt, unknown item: "
<< CAF_ARG(item));
}
CAF_LOG_INFO("could not connect to node directly");
} else {
CAF_LOG_INFO("aborted direct connection attempt, unknown item: "
<< CAF_ARG(item));
}
});
}});
},
after(autoconnect_timeout) >> [=] {
CAF_LOG_TRACE(CAF_ARG(""));
// nothing heard in about 10 minutes... just a call it a day, then
CAF_LOG_INFO("aborted direct connection attempt after 10min");
self->quit(exit_reason::user_shutdown);
}
after(autoconnect_timeout) >>
[=] {
CAF_LOG_TRACE(CAF_ARG(""));
// nothing heard in about 10 minutes... just a call it a day, then
CAF_LOG_INFO("aborted direct connection attempt after 10min");
self->quit(exit_reason::user_shutdown);
},
};
}
} // namespace io
} // namespace caf
......@@ -75,4 +75,3 @@ void datagram_servant::datagram_sent(execution_unit* ctx, datagram_handle hdl,
} // namespace io
} // namespace caf
......@@ -18,49 +18,49 @@
#include "caf/io/middleman.hpp"
#include <tuple>
#include <cerrno>
#include <memory>
#include <cstring>
#include <memory>
#include <sstream>
#include <stdexcept>
#include <tuple>
#include "caf/sec.hpp"
#include "caf/send.hpp"
#include "caf/actor.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/actor_registry.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/after.hpp"
#include "caf/config.hpp"
#include "caf/logger.hpp"
#include "caf/node_id.hpp"
#include "caf/defaults.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/make_counted.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/function_view.hpp"
#include "caf/actor_registry.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/function_view.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/node_id.hpp"
#include "caf/raw_event_based_actor.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/test_multiplexer.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/get_mac_addresses.hpp"
#include "caf/detail/get_root_uuid.hpp"
#include "caf/detail/ripemd_160.hpp"
#include "caf/detail/safe_equal.hpp"
#include "caf/detail/get_root_uuid.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/detail/get_mac_addresses.hpp"
#ifdef CAF_WINDOWS
#include <io.h>
#include <fcntl.h>
# include <fcntl.h>
# include <io.h>
#endif // CAF_WINDOWS
namespace caf {
......@@ -106,8 +106,8 @@ expected<strong_actor_ptr> middleman::remote_spawn_impl(const node_id& nid,
std::set<std::string> s,
duration timeout) {
auto f = make_function_view(actor_handle(), timeout);
return f(spawn_atom::value, nid, std::move(name),
std::move(args), std::move(s));
return f(spawn_atom::value, nid, std::move(name), std::move(args),
std::move(s));
}
expected<uint16_t> middleman::open(uint16_t port, const char* in, bool reuse) {
......@@ -151,7 +151,7 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port,
return {
[self](get_atom, const std::string& name) {
return self->system().groups().get_local(name);
}
},
};
};
auto gn = system().spawn<hidden + lazy_init>(group_nameserver);
......@@ -217,34 +217,25 @@ expected<group> middleman::remote_group(const std::string& group_identifier,
/// terminate the actor after both requests finish.
self->unbecome();
auto rp = self->make_response_promise();
self->request(mm, infinite, connect_atom::value, host, port).then(
[=](const node_id&, strong_actor_ptr& ptr,
const std::set<std::string>&) mutable {
self->request(mm, infinite, connect_atom::value, host, port)
.then([=](const node_id&, strong_actor_ptr& ptr,
const std::set<std::string>&) mutable {
auto hdl = actor_cast<actor>(ptr);
self->request(hdl, infinite, get_atom::value, group_identifier)
.then(
[=](group& result) mutable {
rp.deliver(std::move(result));
}
);
}
);
}
.then(
[=](group& result) mutable { rp.deliver(std::move(result)); });
});
},
};
};
// Spawn the helper actor and wait for the result.
expected<group> result{sec::cannot_connect_to_node};
scoped_actor self{system(), true};
self->request(self->spawn<lazy_init>(two_step_lookup, actor_handle()),
infinite, get_atom::value)
.receive(
[&](group& grp) {
result = std::move(grp);
},
[&](error& err) {
result = std::move(err);
}
);
self
->request(self->spawn<lazy_init>(two_step_lookup, actor_handle()), infinite,
get_atom::value)
.receive([&](group& grp) { result = std::move(grp); },
[&](error& err) { result = std::move(err); });
return result;
}
......@@ -257,14 +248,11 @@ strong_actor_ptr middleman::remote_lookup(atom_value name, const node_id& nid) {
scoped_actor self{system(), true};
self->send(basp, forward_atom::value, nid, atom("ConfigServ"),
make_message(get_atom::value, name));
self->receive(
[&](strong_actor_ptr& addr) {
result = std::move(addr);
},
after(std::chrono::minutes(5)) >> [] {
// nop
}
);
self->receive([&](strong_actor_ptr& addr) { result = std::move(addr); },
after(std::chrono::minutes(5)) >>
[] {
// nop
});
return result;
}
......@@ -281,7 +269,7 @@ void middleman::start() {
std::atomic<bool> init_done{false};
std::mutex mtx;
std::condition_variable cv;
thread_ = std::thread{[&,this] {
thread_ = std::thread{[&, this] {
CAF_SET_LOGGER_SYS(&system());
detail::set_thread_name("caf.multiplexer");
system().thread_started();
......@@ -342,14 +330,13 @@ void middleman::init(actor_system_config& cfg) {
defaults::middleman::network_backend);
if (network_backend == atom("testing")) {
cfg.set("middleman.attach-utility-actors", true)
.set("middleman.manual-multiplexing", true);
.set("middleman.manual-multiplexing", true);
}
// add remote group module to config
struct remote_groups : group_module {
public:
remote_groups(middleman& parent)
: group_module(parent.system(), "remote"),
parent_(parent) {
: group_module(parent.system(), "remote"), parent_(parent) {
// nop
}
......@@ -374,16 +361,16 @@ void middleman::init(actor_system_config& cfg) {
// logging not available at this stage
// add I/O-related types to config
cfg.add_message_type<network::protocol>("@protocol")
.add_message_type<network::address_listing>("@address_listing")
.add_message_type<network::receive_buffer>("@receive_buffer")
.add_message_type<new_data_msg>("@new_data_msg")
.add_message_type<new_connection_msg>("@new_connection_msg")
.add_message_type<acceptor_closed_msg>("@acceptor_closed_msg")
.add_message_type<connection_closed_msg>("@connection_closed_msg")
.add_message_type<accept_handle>("@accept_handle")
.add_message_type<connection_handle>("@connection_handle")
.add_message_type<connection_passivated_msg>("@connection_passivated_msg")
.add_message_type<acceptor_passivated_msg>("@acceptor_passivated_msg");
.add_message_type<network::address_listing>("@address_listing")
.add_message_type<network::receive_buffer>("@receive_buffer")
.add_message_type<new_data_msg>("@new_data_msg")
.add_message_type<new_connection_msg>("@new_connection_msg")
.add_message_type<acceptor_closed_msg>("@acceptor_closed_msg")
.add_message_type<connection_closed_msg>("@connection_closed_msg")
.add_message_type<accept_handle>("@accept_handle")
.add_message_type<connection_handle>("@connection_handle")
.add_message_type<connection_passivated_msg>("@connection_passivated_msg")
.add_message_type<acceptor_passivated_msg>("@acceptor_passivated_msg");
// Compute and set ID for this network node.
auto this_node = node_id::default_data::local(cfg);
system().node_.swap(this_node);
......
......@@ -18,13 +18,13 @@
#include "caf/io/middleman_actor.hpp"
#include <tuple>
#include <stdexcept>
#include <tuple>
#include <utility>
#include "caf/actor_system.hpp"
#include "caf/spawn_options.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/spawn_options.hpp"
#include "caf/io/middleman_actor_impl.hpp"
......@@ -33,8 +33,8 @@ namespace io {
middleman_actor make_middleman_actor(actor_system& sys, actor db) {
return get_or(sys.config(), "middleman.attach-utility-actors", false)
? sys.spawn<middleman_actor_impl, hidden>(std::move(db))
: sys.spawn<middleman_actor_impl, detached + hidden>(std::move(db));
? sys.spawn<middleman_actor_impl, hidden>(std::move(db))
: sys.spawn<middleman_actor_impl, detached + hidden>(std::move(db));
}
} // namespace io
......
......@@ -18,32 +18,31 @@
#include "caf/io/middleman_actor_impl.hpp"
#include <tuple>
#include <stdexcept>
#include <tuple>
#include <utility>
#include "caf/sec.hpp"
#include "caf/send.hpp"
#include "caf/actor.hpp"
#include "caf/logger.hpp"
#include "caf/node_id.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/logger.hpp"
#include "caf/node_id.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/interfaces.hpp"
namespace caf {
namespace io {
middleman_actor_impl::middleman_actor_impl(actor_config& cfg,
actor default_broker)
: middleman_actor::base(cfg),
broker_(std::move(default_broker)) {
: middleman_actor::base(cfg), broker_(std::move(default_broker)) {
set_down_handler([=](down_msg& dm) {
auto i = cached_tcp_.begin();
auto e = cached_tcp_.end();
......@@ -154,9 +153,9 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
[=](spawn_atom atm, node_id& nid, std::string& str, message& msg,
std::set<std::string>& ifs) -> delegated<strong_actor_ptr> {
CAF_LOG_TRACE("");
delegate(
broker_, forward_atom::value, nid, atom("SpawnServ"),
make_message(atm, std::move(str), std::move(msg), std::move(ifs)));
delegate(broker_, forward_atom::value, nid, atom("SpawnServ"),
make_message(atm, std::move(str), std::move(msg),
std::move(ifs)));
return {};
},
[=](get_atom atm,
......@@ -164,15 +163,15 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
CAF_LOG_TRACE("");
delegate(broker_, atm, std::move(nid));
return {};
}
},
};
}
middleman_actor_impl::put_res
middleman_actor_impl::put(uint16_t port, strong_actor_ptr& whom, mpi_set& sigs,
const char* in, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs)
<< CAF_ARG(in) << CAF_ARG(reuse_addr));
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs) << CAF_ARG(in)
<< CAF_ARG(reuse_addr));
uint16_t actual_port;
// treat empty strings like nullptr
if (in != nullptr && in[0] == '\0')
......@@ -190,8 +189,8 @@ middleman_actor_impl::put(uint16_t port, strong_actor_ptr& whom, mpi_set& sigs,
middleman_actor_impl::put_res
middleman_actor_impl::put_udp(uint16_t port, strong_actor_ptr& whom,
mpi_set& sigs, const char* in, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs)
<< CAF_ARG(in) << CAF_ARG(reuse_addr));
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs) << CAF_ARG(in)
<< CAF_ARG(reuse_addr));
uint16_t actual_port;
// treat empty strings like nullptr
if (in != nullptr && in[0] == '\0')
......@@ -240,8 +239,8 @@ middleman_actor_impl::contact(const std::string& host, uint16_t port) {
return system().middleman().backend().new_remote_udp_endpoint(host, port);
}
expected<doorman_ptr>
middleman_actor_impl::open(uint16_t port, const char* addr, bool reuse) {
expected<doorman_ptr> middleman_actor_impl::open(uint16_t port,
const char* addr, bool reuse) {
return system().middleman().backend().new_tcp_doorman(port, addr, reuse);
}
......
......@@ -25,8 +25,7 @@ namespace io {
namespace network {
acceptor::acceptor(default_multiplexer& backend_ref, native_socket sockfd)
: event_handler(backend_ref, sockfd),
sock_(invalid_native_socket) {
: event_handler(backend_ref, sockfd), sock_(invalid_native_socket) {
// nop
}
......
......@@ -39,13 +39,13 @@ namespace network {
datagram_handler::datagram_handler(default_multiplexer& backend_ref,
native_socket sockfd)
: event_handler(backend_ref, sockfd),
max_consecutive_reads_(
get_or(backend().system().config(), "middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads)),
max_datagram_size_(receive_buffer_size),
rd_buf_(receive_buffer_size),
send_buffer_size_(0) {
: event_handler(backend_ref, sockfd),
max_consecutive_reads_(get_or(backend().system().config(),
"middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads)),
max_datagram_size_(receive_buffer_size),
rd_buf_(receive_buffer_size),
send_buffer_size_(0) {
allow_udp_connreset(sockfd, false);
auto es = send_buffer_size(sockfd);
if (!es)
......@@ -74,7 +74,8 @@ void datagram_handler::write(datagram_handle hdl, const void* buf,
wr_offline_buf_.back().first = hdl;
auto cbuf = reinterpret_cast<const char*>(buf);
wr_offline_buf_.back().second.assign(cbuf,
cbuf + static_cast<ptrdiff_t>(num_bytes));
cbuf
+ static_cast<ptrdiff_t>(num_bytes));
}
void datagram_handler::flush(const manager_ptr& mgr) {
......@@ -88,7 +89,8 @@ void datagram_handler::flush(const manager_ptr& mgr) {
}
}
std::unordered_map<datagram_handle, ip_endpoint>& datagram_handler::endpoints() {
std::unordered_map<datagram_handle, ip_endpoint>&
datagram_handler::endpoints() {
return ep_by_hdl_;
}
......@@ -124,9 +126,13 @@ void datagram_handler::remove_endpoint(datagram_handle hdl) {
void datagram_handler::removed_from_loop(operation op) {
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: ; // nop
case operation::read:
reader_.reset();
break;
case operation::write:
writer_.reset();
break;
case operation::propagate_error:; // nop
};
}
......@@ -184,7 +190,8 @@ bool datagram_handler::handle_read_result(bool read_result) {
return true;
}
void datagram_handler::handle_write_result(bool write_result, datagram_handle id,
void datagram_handler::handle_write_result(bool write_result,
datagram_handle id,
std::vector<char>& buf, size_t wb) {
if (!write_result) {
writer_->io_failure(&backend(), operation::write);
......
......@@ -29,4 +29,3 @@ datagram_manager::~datagram_manager() {
} // namespace network
} // namespace io
} // namespace caf
......@@ -30,20 +30,20 @@ namespace network {
datagram_servant_impl::datagram_servant_impl(default_multiplexer& mx,
native_socket sockfd, int64_t id)
: datagram_servant(datagram_handle::from_int(id)),
launched_(false),
handler_(mx, sockfd) {
: datagram_servant(datagram_handle::from_int(id)),
launched_(false),
handler_(mx, sockfd) {
// nop
}
bool datagram_servant_impl::new_endpoint(network::receive_buffer& buf) {
CAF_LOG_TRACE("");
if (detached())
// We are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event
// causes the broker to call close_all() while the pollset contained
// further activities for the broker.
return false;
// We are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event
// causes the broker to call close_all() while the pollset contained
// further activities for the broker.
return false;
// A datagram that has a source port of zero is valid and never requires a
// reply. In the case of CAF we can simply drop it as nothing but the
// handshake could be communicated which we could not reply to.
......@@ -141,7 +141,6 @@ void datagram_servant_impl::remove_from_loop() {
handler_.passivate();
}
void datagram_servant_impl::detach_handles() {
for (auto& p : handler_.endpoints()) {
if (p.first != hdl())
......
......@@ -29,8 +29,7 @@ namespace io {
namespace network {
doorman_impl::doorman_impl(default_multiplexer& mx, native_socket sockfd)
: doorman(network::accept_hdl_from_socket(sockfd)),
acceptor_(mx, sockfd) {
: doorman(network::accept_hdl_from_socket(sockfd)), acceptor_(mx, sockfd) {
// nop
}
......
......@@ -23,9 +23,9 @@
#include "caf/io/network/default_multiplexer.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
# include <winsock2.h>
#else
# include <sys/socket.h>
# include <sys/socket.h>
#endif
namespace caf {
......
......@@ -20,28 +20,28 @@
#include "caf/config.hpp"
#include <algorithm>
#include <cerrno>
#include <cstdlib>
#include <cstring>
#include <algorithm>
#ifdef CAF_WINDOWS
# ifndef _WIN32_WINNT
# define _WIN32_WINNT 0x0600
# endif
# include <iostream>
# include <winsock2.h>
# include <ws2tcpip.h>
# include <iphlpapi.h>
# ifndef _WIN32_WINNT
# define _WIN32_WINNT 0x0600
# endif
# include <iostream>
# include <iphlpapi.h>
# include <winsock2.h>
# include <ws2tcpip.h>
#else
# include <sys/socket.h>
# include <netinet/in.h>
# include <net/if.h>
# include <unistd.h>
# include <netdb.h>
# include <ifaddrs.h>
# include <sys/ioctl.h>
# include <arpa/inet.h>
# include <arpa/inet.h>
# include <ifaddrs.h>
# include <net/if.h>
# include <netdb.h>
# include <netinet/in.h>
# include <sys/ioctl.h>
# include <sys/socket.h>
# include <unistd.h>
#endif
#include <memory>
......@@ -56,9 +56,8 @@ namespace io {
namespace network {
// {interface_name => {protocol => address}}
using interfaces_map = std::map<std::string,
std::map<protocol::network,
std::vector<std::string>>>;
using interfaces_map = std::map<
std::string, std::map<protocol::network, std::vector<std::string>>>;
template <class T>
void* vptr(T* ptr) {
......@@ -71,17 +70,16 @@ void* fetch_in_addr(int family, sockaddr* addr) {
return vptr(&reinterpret_cast<sockaddr_in6*>(addr)->sin6_addr);
}
int fetch_addr_str(bool get_ipv4, bool get_ipv6,
char (&buf)[INET6_ADDRSTRLEN],
int fetch_addr_str(bool get_ipv4, bool get_ipv6, char (&buf)[INET6_ADDRSTRLEN],
sockaddr* addr) {
if (addr == nullptr)
return AF_UNSPEC;
auto family = addr->sa_family;
auto in_addr = fetch_in_addr(family, addr);
return ((family == AF_INET && get_ipv4) || (family == AF_INET6 && get_ipv6))
&& inet_ntop(family, in_addr, buf, INET6_ADDRSTRLEN) == buf
? family
: AF_UNSPEC;
&& inet_ntop(family, in_addr, buf, INET6_ADDRSTRLEN) == buf
? family
: AF_UNSPEC;
}
#ifdef CAF_WINDOWS
......@@ -100,21 +98,21 @@ void for_each_address(bool get_ipv4, bool get_ipv6, F fun) {
tmp = reinterpret_cast<IP_ADAPTER_ADDRESSES*>(malloc(tmp_size));
if (!tmp)
CAF_RAISE_ERROR("malloc() failed");
retval = GetAdaptersAddresses(AF_UNSPEC, GAA_FLAG_INCLUDE_PREFIX,
nullptr, tmp, &tmp_size);
retval = GetAdaptersAddresses(AF_UNSPEC, GAA_FLAG_INCLUDE_PREFIX, nullptr,
tmp, &tmp_size);
} while (retval == ERROR_BUFFER_OVERFLOW && ++try_nr < max_tries);
std::unique_ptr<IP_ADAPTER_ADDRESSES, decltype(free)*> ifs{tmp, free};
if (retval != NO_ERROR) {
std::cerr << "Call to GetAdaptersAddresses failed with error: "
<< retval << std::endl;
std::cerr << "Call to GetAdaptersAddresses failed with error: " << retval
<< std::endl;
if (retval == ERROR_NO_DATA) {
std::cerr << "No addresses were found for the requested parameters"
<< std::endl;
} else {
void* msgbuf = nullptr;
if (FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER
| FORMAT_MESSAGE_FROM_SYSTEM
| FORMAT_MESSAGE_IGNORE_INSERTS,
| FORMAT_MESSAGE_FROM_SYSTEM
| FORMAT_MESSAGE_IGNORE_INSERTS,
nullptr, retval,
MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
(LPTSTR) &msgbuf, 0, nullptr)) {
......@@ -152,8 +150,7 @@ void for_each_address(bool get_ipv4, bool get_ipv6, F fun) {
auto family = fetch_addr_str(get_ipv4, get_ipv6, buffer, i->ifa_addr);
if (family != AF_UNSPEC)
fun(i->ifa_name, family == AF_INET ? protocol::ipv4 : protocol::ipv6,
(i->ifa_flags & IFF_LOOPBACK) != 0,
buffer);
(i->ifa_flags & IFF_LOOPBACK) != 0, buffer);
}
}
......@@ -181,24 +178,24 @@ void interfaces::traverse(consumer f) {
interfaces_map interfaces::list_all(bool include_localhost) {
interfaces_map result;
traverse_impl(
{protocol::ipv4, protocol::ipv6},
[&](const char* name, protocol::network p, bool lo, const char* addr) {
if (include_localhost || !lo)
result[name][p].emplace_back(addr);
});
traverse_impl({protocol::ipv4, protocol::ipv6},
[&](const char* name, protocol::network p, bool lo,
const char* addr) {
if (include_localhost || !lo)
result[name][p].emplace_back(addr);
});
return result;
}
std::map<protocol::network, std::vector<std::string>>
interfaces::list_addresses(bool include_localhost) {
std::map<protocol::network, std::vector<std::string>> result;
traverse_impl(
{protocol::ipv4, protocol::ipv6},
[&](const char*, protocol::network p, bool lo, const char* addr) {
if (include_localhost || !lo)
result[p].emplace_back(addr);
});
traverse_impl({protocol::ipv4, protocol::ipv6},
[&](const char*, protocol::network p, bool lo,
const char* addr) {
if (include_localhost || !lo)
result[p].emplace_back(addr);
});
return result;
}
......@@ -266,8 +263,7 @@ interfaces::server_address(uint16_t port, const char* host,
auto family = fetch_addr_str(true, true, buffer, i->ai_addr);
if (family != AF_UNSPEC) {
results.emplace_back(std::string{buffer},
family == AF_INET ? protocol::ipv4
: protocol::ipv6);
family == AF_INET ? protocol::ipv4 : protocol::ipv6);
}
}
std::stable_sort(std::begin(results), std::end(results),
......
......@@ -24,17 +24,17 @@
#include "caf/sec.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
# include <windows.h>
# include <ws2tcpip.h>
# include <ws2ipdef.h>
# include <windows.h>
# include <winsock2.h>
# include <ws2ipdef.h>
# include <ws2tcpip.h>
#else
# include <unistd.h>
# include <cerrno>
# include <arpa/inet.h>
# include <sys/socket.h>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <arpa/inet.h>
# include <cerrno>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <sys/socket.h>
# include <unistd.h>
#endif
#ifdef CAF_WINDOWS
......@@ -91,7 +91,7 @@ void ip_endpoint::clear() {
ptr_->len = 0;
}
void ip_endpoint::impl_deleter::operator()(ip_endpoint::impl *ptr) const {
void ip_endpoint::impl_deleter::operator()(ip_endpoint::impl* ptr) const {
delete ptr;
}
......@@ -126,8 +126,8 @@ size_t ep_hash::hash(const sockaddr_in6* sa) const noexcept {
bool operator==(const ip_endpoint& lhs, const ip_endpoint& rhs) {
auto same = false;
if (*lhs.clength() == *rhs.clength() &&
lhs.caddress()->sa_family == rhs.caddress()->sa_family) {
if (*lhs.clength() == *rhs.clength()
&& lhs.caddress()->sa_family == rhs.caddress()->sa_family) {
switch (lhs.caddress()->sa_family) {
case AF_INET: {
auto* la = reinterpret_cast<const sockaddr_in*>(lhs.caddress());
......@@ -158,15 +158,19 @@ std::string host(const ip_endpoint& ep) {
char addr[INET6_ADDRSTRLEN];
if (*ep.clength() == 0)
return "";
switch(ep.caddress()->sa_family) {
switch (ep.caddress()->sa_family) {
case AF_INET:
inet_ntop(AF_INET,
&const_cast<sockaddr_in*>(reinterpret_cast<const sockaddr_in*>(ep.caddress()))->sin_addr,
&const_cast<sockaddr_in*>(
reinterpret_cast<const sockaddr_in*>(ep.caddress()))
->sin_addr,
addr, static_cast<socket_size_type>(*ep.clength()));
break;
case AF_INET6:
inet_ntop(AF_INET6,
&const_cast<sockaddr_in6*>(reinterpret_cast<const sockaddr_in6*>(ep.caddress()))->sin6_addr,
&const_cast<sockaddr_in6*>(
reinterpret_cast<const sockaddr_in6*>(ep.caddress()))
->sin6_addr,
addr, static_cast<socket_size_type>(*ep.clength()));
break;
default:
......@@ -180,12 +184,14 @@ uint16_t port(const ip_endpoint& ep) {
uint16_t port = 0;
if (*ep.clength() == 0)
return 0;
switch(ep.caddress()->sa_family) {
switch (ep.caddress()->sa_family) {
case AF_INET:
port = ntohs(reinterpret_cast<const sockaddr_in*>(ep.caddress())->sin_port);
port = ntohs(
reinterpret_cast<const sockaddr_in*>(ep.caddress())->sin_port);
break;
case AF_INET6:
port = ntohs(reinterpret_cast<const sockaddr_in6*>(ep.caddress())->sin6_port);
port = ntohs(
reinterpret_cast<const sockaddr_in6*>(ep.caddress())->sin6_port);
break;
default:
// nop
......@@ -200,12 +206,12 @@ uint32_t family(const ip_endpoint& ep) {
return ep.caddress()->sa_family;
}
error load_endpoint(ip_endpoint& ep, uint32_t& f, std::string& h,
uint16_t& p, size_t& l) {
error load_endpoint(ip_endpoint& ep, uint32_t& f, std::string& h, uint16_t& p,
size_t& l) {
ep.clear();
if (l > 0) {
*ep.length() = l;
switch(f) {
switch (f) {
case AF_INET: {
auto* addr = reinterpret_cast<sockaddr_in*>(ep.address());
inet_pton(AF_INET, h.c_str(), &addr->sin_addr);
......@@ -227,8 +233,8 @@ error load_endpoint(ip_endpoint& ep, uint32_t& f, std::string& h,
return none;
}
error save_endpoint(ip_endpoint& ep, uint32_t& f, std::string& h,
uint16_t& p, size_t& l) {
error save_endpoint(ip_endpoint& ep, uint32_t& f, std::string& h, uint16_t& p,
size_t& l) {
if (*ep.length() > 0) {
f = family(ep);
h = host(ep);
......
......@@ -57,8 +57,8 @@ void manager::detach(execution_unit*, bool invoke_disconnect_message) {
ptr.swap(parent_);
detach_from(raw_ptr);
if (invoke_disconnect_message) {
auto mptr = make_mailbox_element(nullptr, make_message_id(),
{}, detach_message());
auto mptr = make_mailbox_element(nullptr, make_message_id(), {},
detach_message());
switch (raw_ptr->consume(*mptr)) {
case im_success:
raw_ptr->finalize();
......
......@@ -24,8 +24,7 @@ namespace io {
namespace network {
multiplexer::multiplexer(actor_system* sys)
: execution_unit(sys),
tid_(std::this_thread::get_id()) {
: execution_unit(sys), tid_(std::this_thread::get_id()) {
// nop
}
......
This diff is collapsed.
......@@ -20,14 +20,14 @@
#include <cstdint>
#include "caf/io/network/pipe_reader.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/pipe_reader.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
# include <winsock2.h>
#else
# include <unistd.h>
# include <sys/socket.h>
# include <sys/socket.h>
# include <unistd.h>
#endif
namespace caf {
......@@ -35,7 +35,7 @@ namespace io {
namespace network {
pipe_reader::pipe_reader(default_multiplexer& dm)
: event_handler(dm, invalid_native_socket) {
: event_handler(dm, invalid_native_socket) {
// nop
}
......@@ -50,12 +50,12 @@ void pipe_reader::graceful_shutdown() {
resumable* pipe_reader::try_read_next() {
std::intptr_t ptrval;
// on windows, we actually have sockets, otherwise we have file handles
# ifdef CAF_WINDOWS
auto res = recv(fd(), reinterpret_cast<socket_recv_ptr>(&ptrval),
sizeof(ptrval), 0);
# else
auto res = read(fd(), &ptrval, sizeof(ptrval));
# endif
#ifdef CAF_WINDOWS
auto res = recv(fd(), reinterpret_cast<socket_recv_ptr>(&ptrval),
sizeof(ptrval), 0);
#else
auto res = read(fd(), &ptrval, sizeof(ptrval));
#endif
if (res != sizeof(ptrval))
return nullptr;
return reinterpret_cast<resumable*>(ptrval);
......
......@@ -33,4 +33,3 @@ std::string to_string(const protocol& x) {
} // namespace network
} // namespace io
} // namespace caf
......@@ -33,9 +33,7 @@ namespace io {
namespace network {
receive_buffer::receive_buffer() noexcept
: buffer_(nullptr),
capacity_(0),
size_(0) {
: buffer_(nullptr), capacity_(0), size_(0) {
// nop
}
......@@ -44,12 +42,12 @@ receive_buffer::receive_buffer(size_type count) : receive_buffer() {
}
receive_buffer::receive_buffer(receive_buffer&& other) noexcept
: receive_buffer() {
: receive_buffer() {
swap(other);
}
receive_buffer::receive_buffer(const receive_buffer& other)
: receive_buffer(other.size()) {
: receive_buffer(other.size()) {
std::copy(other.cbegin(), other.cend(), buffer_.get());
}
......@@ -72,7 +70,7 @@ void receive_buffer::resize(size_type new_size) {
void receive_buffer::reserve(size_type new_size) {
if (new_size > capacity_)
increase_by(new_size - capacity_);
increase_by(new_size - capacity_);
}
void receive_buffer::shrink_to_fit() {
......@@ -126,7 +124,8 @@ void receive_buffer::shrink_by(size_t bytes) {
capacity_ = new_size;
}
receive_buffer::iterator receive_buffer::insert(iterator pos, value_type value) {
receive_buffer::iterator receive_buffer::insert(iterator pos,
value_type value) {
if (size_ == capacity_) {
auto dist = (pos == nullptr) ? 0 : std::distance(begin(), pos);
increase_by(std::max(capacity_, min_size));
......@@ -138,6 +137,6 @@ receive_buffer::iterator receive_buffer::insert(iterator pos, value_type value)
return pos;
}
} // namepsace network
} // namespace network
} // namespace io
} // namespace caf
......@@ -29,9 +29,9 @@ namespace io {
namespace network {
scribe_impl::scribe_impl(default_multiplexer& mx, native_socket sockfd)
: scribe(network::conn_hdl_from_socket(sockfd)),
launched_(false),
stream_(mx, sockfd) {
: scribe(network::conn_hdl_from_socket(sockfd)),
launched_(false),
stream_(mx, sockfd) {
// nop
}
......
......@@ -31,13 +31,13 @@ namespace io {
namespace network {
stream::stream(default_multiplexer& backend_ref, native_socket sockfd)
: event_handler(backend_ref, sockfd),
max_consecutive_reads_(
get_or(backend().system().config(), "middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads)),
read_threshold_(1),
collected_(0),
written_(0) {
: event_handler(backend_ref, sockfd),
max_consecutive_reads_(get_or(backend().system().config(),
"middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads)),
read_threshold_(1),
collected_(0),
written_(0) {
configure_read(receive_policy::at_most(1024));
}
......@@ -62,7 +62,7 @@ void stream::configure_read(receive_policy::config config) {
void stream::write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes));
auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes;
auto last = first + num_bytes;
wr_offline_buf_.insert(wr_offline_buf_.end(), first, last);
}
......@@ -80,9 +80,13 @@ void stream::flush(const manager_ptr& mgr) {
void stream::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG2("fd", fd_) << CAF_ARG(op));
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: ; // nop
case operation::read:
reader_.reset();
break;
case operation::write:
writer_.reset();
break;
case operation::propagate_error:; // nop
}
}
......@@ -159,8 +163,7 @@ bool stream::handle_read_result(rw_state read_result, size_t rb) {
return false;
collected_ += rb;
if (collected_ >= read_threshold_) {
auto res = reader_->consume(&backend(), rd_buf_.data(),
collected_);
auto res = reader_->consume(&backend(), rd_buf_.data(), collected_);
prepare_next_read();
if (!res) {
passivate();
......
......@@ -36,44 +36,39 @@ constexpr size_t receive_buffer_size = std::numeric_limits<uint16_t>::max();
test_multiplexer::scribe_data::scribe_data(shared_buffer_type input,
shared_buffer_type output)
: vn_buf_ptr(std::move(input)),
wr_buf_ptr(std::move(output)),
vn_buf(*vn_buf_ptr),
wr_buf(*wr_buf_ptr),
stopped_reading(false),
passive_mode(false),
ack_writes(false) {
: vn_buf_ptr(std::move(input)),
wr_buf_ptr(std::move(output)),
vn_buf(*vn_buf_ptr),
wr_buf(*wr_buf_ptr),
stopped_reading(false),
passive_mode(false),
ack_writes(false) {
// nop
}
test_multiplexer::doorman_data::doorman_data()
: port(0),
stopped_reading(false),
passive_mode(false) {
: port(0), stopped_reading(false), passive_mode(false) {
// nop
}
test_multiplexer::datagram_data::
datagram_data(shared_job_queue_type input,
shared_job_queue_type output)
: vn_buf_ptr(std::move(input)),
wr_buf_ptr(std::move(output)),
vn_buf(*vn_buf_ptr),
wr_buf(*wr_buf_ptr),
rd_buf(datagram_handle::from_int(0), receive_buffer_size),
stopped_reading(false),
passive_mode(false),
ack_writes(false),
port(0),
local_port(0),
datagram_size(receive_buffer_size) {
test_multiplexer::datagram_data::datagram_data(shared_job_queue_type input,
shared_job_queue_type output)
: vn_buf_ptr(std::move(input)),
wr_buf_ptr(std::move(output)),
vn_buf(*vn_buf_ptr),
wr_buf(*wr_buf_ptr),
rd_buf(datagram_handle::from_int(0), receive_buffer_size),
stopped_reading(false),
passive_mode(false),
ack_writes(false),
port(0),
local_port(0),
datagram_size(receive_buffer_size) {
// nop
}
test_multiplexer::test_multiplexer(actor_system* sys)
: multiplexer(sys),
inline_runnables_(0),
servant_ids_(0) {
: multiplexer(sys), inline_runnables_(0), servant_ids_(0) {
CAF_ASSERT(sys != nullptr);
}
......@@ -126,6 +121,7 @@ scribe_ptr test_multiplexer::new_scribe(connection_handle hdl) {
void remove_from_loop() override {
mpx_->passive_mode(hdl()) = true;
}
private:
test_multiplexer* mpx_;
};
......@@ -203,6 +199,7 @@ doorman_ptr test_multiplexer::new_doorman(accept_handle hdl, uint16_t port) {
void remove_from_loop() override {
mpx_->passive_mode(hdl()) = true;
}
private:
test_multiplexer* mpx_;
};
......@@ -251,15 +248,17 @@ expected<doorman_ptr> test_multiplexer::new_tcp_doorman(uint16_t desired_port,
datagram_servant_ptr test_multiplexer::new_datagram_servant(native_socket) {
CAF_ASSERT(std::this_thread::get_id() == tid_);
CAF_CRITICAL("test_multiplexer::new_datagram_servant called with native socket");
CAF_CRITICAL(
"test_multiplexer::new_datagram_servant called with native socket");
}
datagram_servant_ptr
test_multiplexer::new_datagram_servant_for_endpoint(native_socket,
const ip_endpoint&) {
CAF_ASSERT(std::this_thread::get_id() == tid_);
CAF_CRITICAL("test_multiplexer::new_datagram_servant_for_endpoint called with "
"native socket");
CAF_CRITICAL(
"test_multiplexer::new_datagram_servant_for_endpoint called with "
"native socket");
}
expected<datagram_servant_ptr>
......@@ -289,8 +288,8 @@ test_multiplexer::new_remote_udp_endpoint(const std::string& host,
}
expected<datagram_servant_ptr>
test_multiplexer::new_local_udp_endpoint(uint16_t desired_port,
const char*, bool) {
test_multiplexer::new_local_udp_endpoint(uint16_t desired_port, const char*,
bool) {
CAF_LOG_TRACE(CAF_ARG(desired_port));
datagram_handle hdl;
uint16_t port = 0;
......@@ -327,7 +326,7 @@ datagram_servant_ptr test_multiplexer::new_datagram_servant(datagram_handle hdl,
class impl : public datagram_servant {
public:
impl(datagram_handle dh, test_multiplexer* mpx)
: datagram_servant(dh), mpx_(mpx) {
: datagram_servant(dh), mpx_(mpx) {
// nop
}
bool new_endpoint(network::receive_buffer& buf) override {
......@@ -355,8 +354,7 @@ datagram_servant_ptr test_multiplexer::new_datagram_servant(datagram_handle hdl,
buf.first = dh;
return buf.second;
}
void enqueue_datagram(datagram_handle dh,
std::vector<char> buf) override {
void enqueue_datagram(datagram_handle dh, std::vector<char> buf) override {
auto& q = mpx_->output_queue(dh);
q.emplace_back(dh, std::move(buf));
}
......@@ -398,7 +396,8 @@ datagram_servant_ptr test_multiplexer::new_datagram_servant(datagram_handle hdl,
mpx_->passive_mode(hdl()) = true;
}
void add_endpoint(const ip_endpoint&, datagram_handle) override {
CAF_CRITICAL("datagram_servant impl::add_endpoint called with ip_endpoint");
CAF_CRITICAL(
"datagram_servant impl::add_endpoint called with ip_endpoint");
}
void remove_endpoint(datagram_handle dh) override {
auto data = mpx_->data_for_hdl(hdl());
......@@ -417,6 +416,7 @@ datagram_servant_ptr test_multiplexer::new_datagram_servant(datagram_handle hdl,
data->servants.clear();
data->servants.emplace(hdl());
}
private:
test_multiplexer* mpx_;
};
......@@ -442,7 +442,6 @@ int64_t test_multiplexer::next_endpoint_id() {
return servant_ids_++;
}
bool test_multiplexer::is_known_port(uint16_t x) const {
auto pred1 = [&](const doorman_data_map::value_type& y) {
return x == y.second.port;
......@@ -471,8 +470,9 @@ bool test_multiplexer::is_known_handle(datagram_handle x) const {
return x == y.second;
};
return datagram_data_.count(x) > 0
|| std::any_of(local_endpoints_.begin(), local_endpoints_.end(), pred1)
|| std::any_of(remote_endpoints_.begin(), remote_endpoints_.end(), pred2);
|| std::any_of(local_endpoints_.begin(), local_endpoints_.end(), pred1)
|| std::any_of(remote_endpoints_.begin(), remote_endpoints_.end(),
pred2);
}
auto test_multiplexer::make_supervisor() -> supervisor_ptr {
......@@ -669,7 +669,7 @@ void test_multiplexer::prepare_connection(accept_handle src,
CAF_ASSERT(std::this_thread::get_id() == tid_);
CAF_ASSERT(this != &peer);
CAF_LOG_TRACE(CAF_ARG(src) << CAF_ARG(hdl) << CAF_ARG(host) << CAF_ARG(port)
<< CAF_ARG(peer_hdl));
<< CAF_ARG(peer_hdl));
auto input = std::make_shared<buffer_type>();
auto output = std::make_shared<buffer_type>();
CAF_LOG_DEBUG("insert scribe data for" << CAF_ARG(hdl));
......@@ -681,8 +681,9 @@ void test_multiplexer::prepare_connection(accept_handle src,
if (!res2.second)
CAF_RAISE_ERROR("prepare_connection: peer handle already in use");
CAF_LOG_INFO("acceptor" << src << "has connection" << hdl
<< "ready for incoming connect from" << host << ":"
<< port << "from peer with connection handle" << peer_hdl);
<< "ready for incoming connect from" << host << ":"
<< port << "from peer with connection handle"
<< peer_hdl);
if (doormen_.count(port) == 0)
provide_acceptor(port, src);
add_pending_connect(src, hdl);
......@@ -711,8 +712,7 @@ bool test_multiplexer::has_pending_scribe(std::string x, uint16_t y) {
return scribes_.count(std::make_pair(std::move(x), y)) > 0;
}
bool test_multiplexer::has_pending_remote_endpoint(std::string x,
uint16_t y) {
bool test_multiplexer::has_pending_remote_endpoint(std::string x, uint16_t y) {
CAF_ASSERT(std::this_thread::get_id() == tid_);
guard_type guard{mx_};
return remote_endpoints_.count(std::make_pair(std::move(x), y)) > 0;
......@@ -894,7 +894,8 @@ bool test_multiplexer::read_data(datagram_handle hdl) {
return false;
auto ditr = datagram_data_.find(hdl);
if (ditr == datagram_data_.end() || ditr->second->ptr->parent() == nullptr
|| !ditr->second->ptr->parent()->getf(abstract_actor::is_initialized_flag))
|| !ditr->second->ptr->parent()->getf(
abstract_actor::is_initialized_flag))
return false;
auto& data = ditr->second;
if (data->vn_buf.back().second.empty())
......@@ -1032,8 +1033,7 @@ void test_multiplexer::exec(resumable_ptr& ptr) {
case resumable::awaiting_message:
intrusive_ptr_release(ptr.get());
break;
default:
; // ignored
default:; // ignored
}
}
......
......@@ -71,10 +71,10 @@ void scribe::data_transferred(execution_unit* ctx, size_t written,
mailbox_element::forwarding_stack{},
transferred_t{hdl(), written, remaining}};
invoke_mailbox_element_impl(ctx, tmp);
//data_transferred_msg tmp{hdl(), written, remaining};
//auto ptr = make_mailbox_element(nullptr, invalid_message_id, {}, tmp);
//parent()->context(ctx);
//parent()->consume(std::move(ptr));
// data_transferred_msg tmp{hdl(), written, remaining};
// auto ptr = make_mailbox_element(nullptr, invalid_message_id, {}, tmp);
// parent()->context(ctx);
// parent()->consume(std::move(ptr));
}
} // namespace io
......
......@@ -24,19 +24,19 @@
#include "caf/logger.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
# include <winsock2.h>
#else
# include <sys/types.h>
# include <sys/socket.h>
# include <sys/socket.h>
# include <sys/types.h>
#endif
using caf::io::network::is_error;
using caf::io::network::rw_state;
using caf::io::network::native_socket;
using caf::io::network::socket_size_type;
using caf::io::network::last_socket_error;
using caf::io::network::native_socket;
using caf::io::network::no_sigpipe_io_flag;
using caf::io::network::rw_state;
using caf::io::network::socket_error_as_string;
using caf::io::network::socket_size_type;
namespace caf {
namespace policy {
......
......@@ -37,7 +37,7 @@ namespace {
using caf::make_message_id;
struct anything { };
struct anything {};
anything any_vals;
......@@ -92,7 +92,9 @@ struct node {
std::string name;
node_id id;
connection_handle connection;
union { scoped_actor dummy_actor; };
union {
scoped_actor dummy_actor;
};
node() {
// nop
......@@ -147,7 +149,7 @@ public:
CAF_MESSAGE("Jupiter: " << to_string(jupiter().id));
CAF_MESSAGE("Mars: " << to_string(mars().id));
CAF_REQUIRE_NOT_EQUAL(this_node_, jupiter().id);
CAF_REQUIRE_NOT_EQUAL(jupiter().id, mars().id);
CAF_REQUIRE_NOT_EQUAL(jupiter().id, mars().id);
}
~fixture() {
......@@ -257,13 +259,13 @@ public:
return {hdr, std::move(payload)};
}
void connect_node(node& n,
optional<accept_handle> ax = none,
actor_id published_actor_id = invalid_actor_id,
const set<string>& published_actor_ifs = std::set<std::string>{}) {
void connect_node(
node& n, optional<accept_handle> ax = none,
actor_id published_actor_id = invalid_actor_id,
const set<string>& published_actor_ifs = std::set<std::string>{}) {
auto src = ax ? *ax : ahdl_;
CAF_MESSAGE("connect remote node " << n.name
<< ", connection ID = " << n.connection.id()
CAF_MESSAGE("connect remote node "
<< n.name << ", connection ID = " << n.connection.id()
<< ", acceptor ID = " << src.id());
auto hdl = n.connection;
mpx_->add_pending_connect(src, hdl);
......@@ -339,13 +341,10 @@ public:
}
template <class... Ts>
mock_t& receive(connection_handle hdl,
maybe<basp::message_type> operation,
maybe<uint8_t> flags,
maybe<uint32_t> payload_len,
mock_t& receive(connection_handle hdl, maybe<basp::message_type> operation,
maybe<uint8_t> flags, maybe<uint32_t> payload_len,
maybe<uint64_t> operation_data,
maybe<actor_id> source_actor,
maybe<actor_id> dest_actor,
maybe<actor_id> source_actor, maybe<actor_id> dest_actor,
const Ts&... xs) {
CAF_MESSAGE("expect #" << num);
buffer buf;
......@@ -368,7 +367,7 @@ public:
auto end = first + hdr.payload_len;
payload.assign(first, end);
CAF_MESSAGE("erase " << std::distance(ob.begin(), end)
<< " bytes from output buffer");
<< " bytes from output buffer");
ob.erase(ob.begin(), end);
} else {
ob.erase(ob.begin(), ob.begin() + basp::header_size);
......@@ -394,9 +393,9 @@ public:
mock_t mock(connection_handle hdl, basp::header hdr, const Ts&... xs) {
buffer buf;
to_buf(buf, hdr, nullptr, xs...);
CAF_MESSAGE("virtually send " << to_string(hdr.operation)
<< " with " << (buf.size() - basp::header_size)
<< " bytes payload");
CAF_MESSAGE("virtually send " << to_string(hdr.operation) << " with "
<< (buf.size() - basp::header_size)
<< " bytes payload");
mpx()->virtual_send(hdl, buf);
return {this};
}
......@@ -430,7 +429,6 @@ public:
scheduler_type& sched;
middleman_actor mma;
autoconn_enabled_fixture()
: fixture(true),
sched(dynamic_cast<scheduler_type&>(sys.scheduler())),
......@@ -462,9 +460,12 @@ CAF_TEST(empty_server_handshake) {
basp::header hdr;
buffer payload;
std::tie(hdr, payload) = from_buf(buf);
basp::header expected{basp::message_type::server_handshake, 0,
basp::header expected{basp::message_type::server_handshake,
0,
static_cast<uint32_t>(payload.size()),
basp::version, invalid_actor_id, invalid_actor_id};
basp::version,
invalid_actor_id,
invalid_actor_id};
CAF_CHECK(basp::valid(hdr));
CAF_CHECK(basp::is_handshake(hdr));
CAF_CHECK_EQUAL(to_string(hdr), to_string(expected));
......@@ -480,9 +481,12 @@ CAF_TEST(non_empty_server_handshake) {
basp::header hdr;
buffer payload;
std::tie(hdr, payload) = from_buf(buf);
basp::header expected{basp::message_type::server_handshake, 0,
basp::header expected{basp::message_type::server_handshake,
0,
static_cast<uint32_t>(payload.size()),
basp::version, invalid_actor_id, invalid_actor_id};
basp::version,
invalid_actor_id,
invalid_actor_id};
CAF_CHECK(basp::valid(hdr));
CAF_CHECK(basp::is_handshake(hdr));
CAF_CHECK_EQUAL(to_string(hdr), to_string(expected));
......@@ -509,8 +513,7 @@ CAF_TEST(remote_address_and_port) {
// all test nodes have address "test" and connection handle ID as port
CAF_CHECK_EQUAL(addr, "test");
CAF_CHECK_EQUAL(port, mars().connection.id());
}
);
});
}
CAF_TEST(client_handshake_and_dispatch) {
......@@ -528,23 +531,17 @@ CAF_TEST(client_handshake_and_dispatch) {
CAF_REQUIRE(proxies().count_proxies(jupiter().id) == 1);
// must've send remote node a message that this proxy is monitored now
// receive the message
self()->receive(
[](int a, int b, int c) {
CAF_CHECK_EQUAL(a, 1);
CAF_CHECK_EQUAL(b, 2);
CAF_CHECK_EQUAL(c, 3);
return a + b + c;
}
);
self()->receive([](int a, int b, int c) {
CAF_CHECK_EQUAL(a, 1);
CAF_CHECK_EQUAL(b, 2);
CAF_CHECK_EQUAL(c, 3);
return a + b + c;
});
CAF_MESSAGE("exec message of forwarding proxy");
mpx()->exec_runnable();
// deserialize and send message from out buf
dispatch_out_buf(jupiter().connection);
jupiter().dummy_actor->receive(
[](int i) {
CAF_CHECK_EQUAL(i, 6);
}
);
jupiter().dummy_actor->receive([](int i) { CAF_CHECK_EQUAL(i, 6); });
}
CAF_TEST(message_forwarding) {
......@@ -579,8 +576,8 @@ CAF_TEST(remote_actor_and_send) {
CAF_REQUIRE(mpx()->has_pending_scribe(lo, 4242));
auto mm1 = sys.middleman().actor_handle();
actor result;
auto f = self()->request(mm1, infinite,
connect_atom::value, lo, uint16_t{4242});
auto f = self()->request(mm1, infinite, connect_atom::value, lo,
uint16_t{4242});
// wait until BASP broker has received and processed the connect message
while (!aut()->valid(jupiter().connection))
mpx()->exec_runnable();
......@@ -621,14 +618,11 @@ CAF_TEST(remote_actor_and_send) {
CAF_REQUIRE(proxy == res);
result = actor_cast<actor>(res);
},
[&](error& err) {
CAF_FAIL("error: " << sys.render(err));
}
);
[&](error& err) { CAF_FAIL("error: " << sys.render(err)); });
CAF_MESSAGE("send message to proxy");
anon_send(actor_cast<actor>(result), 42);
mpx()->flush_runnables();
// mpx()->exec_runnable(); // process forwarded message in basp_broker
// mpx()->exec_runnable(); // process forwarded message in basp_broker
mock().receive(jupiter().connection, basp::message_type::direct_message,
no_flags, any_vals, default_operation_data, invalid_actor_id,
jupiter().dummy_actor->id(), std::vector<strong_actor_ptr>{},
......@@ -639,23 +633,19 @@ CAF_TEST(remote_actor_and_send) {
{basp::message_type::direct_message, 0, 0, 0,
jupiter().dummy_actor->id(), self()->id()},
std::vector<strong_actor_ptr>{}, make_message("hi there!"));
self()->receive(
[&](const string& str) {
CAF_CHECK_EQUAL(to_string(self()->current_sender()), to_string(result));
CAF_CHECK_EQUAL(self()->current_sender(), result.address());
CAF_CHECK_EQUAL(str, "hi there!");
}
);
self()->receive([&](const string& str) {
CAF_CHECK_EQUAL(to_string(self()->current_sender()), to_string(result));
CAF_CHECK_EQUAL(self()->current_sender(), result.address());
CAF_CHECK_EQUAL(str, "hi there!");
});
}
CAF_TEST(actor_serialize_and_deserialize) {
auto testee_impl = [](event_based_actor* testee_self) -> behavior {
testee_self->set_default_handler(reflect_and_quit);
return {
[] {
// nop
}
};
return {[] {
// nop
}};
};
connect_node(jupiter());
auto prx = proxies().get_or_put(jupiter().id, jupiter().dummy_actor->id());
......@@ -711,12 +701,10 @@ CAF_TEST(indirect_connections) {
any_vals, no_operation_data, invalid_actor_id,
jupiter().dummy_actor->id(), this_node(), jupiter().id);
CAF_MESSAGE("receive message from jupiter");
self()->receive(
[](const std::string& str) -> std::string {
CAF_CHECK_EQUAL(str, "hello from jupiter!");
return "hello from earth!";
}
);
self()->receive([](const std::string& str) -> std::string {
CAF_CHECK_EQUAL(str, "hello from jupiter!");
return "hello from earth!";
});
mpx()->exec_runnable(); // process forwarded message in basp_broker
mock().receive(mars().connection, basp::message_type::routed_message,
no_flags, any_vals, default_operation_data, self()->id(),
......@@ -807,12 +795,10 @@ CAF_TEST(automatic_connection) {
check_node_in_tbl(jupiter());
check_node_in_tbl(mars());
CAF_MESSAGE("receive message from jupiter");
self()->receive(
[](const std::string& str) -> std::string {
CAF_CHECK_EQUAL(str, "hello from jupiter!");
return "hello from earth!";
}
);
self()->receive([](const std::string& str) -> std::string {
CAF_CHECK_EQUAL(str, "hello from jupiter!");
return "hello from earth!";
});
mpx()->exec_runnable(); // process forwarded message in basp_broker
CAF_MESSAGE("response message must take direct route now");
mock().receive(jupiter().connection, basp::message_type::direct_message,
......
......@@ -22,8 +22,8 @@
#include "caf/test/dsl.hpp"
#include <memory>
#include <iostream>
#include <memory>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
......@@ -39,58 +39,53 @@ using pong_atom = caf::atom_constant<caf::atom("pong")>;
using publish_atom = caf::atom_constant<caf::atom("publish")>;
using kickoff_atom = caf::atom_constant<caf::atom("kickoff")>;
void ping(event_based_actor* self, size_t num_pings) {
behavior ping(event_based_actor* self, size_t num_pings) {
CAF_MESSAGE("num_pings: " << num_pings);
auto count = std::make_shared<size_t>(0);
self->become(
return {
[=](kickoff_atom, const actor& pong) {
CAF_MESSAGE("received `kickoff_atom`");
self->send(pong, ping_atom::value, 1);
self->become(
[=](pong_atom, int value)->std::tuple<atom_value, int> {
if (++*count >= num_pings) {
CAF_MESSAGE("received " << num_pings
<< " pings, call self->quit");
self->quit();
}
return std::make_tuple(ping_atom::value, value + 1);
self->become([=](pong_atom, int value) -> std::tuple<atom_value, int> {
if (++*count >= num_pings) {
CAF_MESSAGE("received " << num_pings << " pings, call self->quit");
self->quit();
}
);
}
);
return std::make_tuple(ping_atom::value, value + 1);
});
},
};
}
void pong(event_based_actor* self) {
behavior pong(event_based_actor* self) {
CAF_MESSAGE("pong actor started");
self->set_down_handler([=](down_msg& dm) {
CAF_MESSAGE("received down_msg{" << to_string(dm.reason) << "}");
self->quit(dm.reason);
});
self->become(
return {
[=](ping_atom, int value) -> std::tuple<atom_value, int> {
CAF_MESSAGE("received `ping_atom`");
self->monitor(self->current_sender());
// set next behavior
self->become(
[](ping_atom, int val) {
return std::make_tuple(pong_atom::value, val);
}
);
self->become([](ping_atom, int val) {
return std::make_tuple(pong_atom::value, val);
});
// reply to 'ping'
return std::make_tuple(pong_atom::value, value);
}
);
},
};
}
void peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
behavior peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
CAF_MESSAGE("peer_fun called");
CAF_REQUIRE(self->subtype() == resumable::io_actor);
CAF_CHECK(self != nullptr);
self->monitor(buddy);
// assume exactly one connection
CAF_REQUIRE(self->connections().size() == 1);
self->configure_read(
hdl, receive_policy::exactly(sizeof(atom_value) + sizeof(int)));
self->configure_read(hdl, receive_policy::exactly(sizeof(atom_value)
+ sizeof(int)));
auto write = [=](atom_value type, int value) {
auto& buf = self->wr_buf(hdl);
auto first = reinterpret_cast<char*>(&type);
......@@ -104,7 +99,7 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
if (dm.source == buddy)
self->quit(dm.reason);
});
self->become(
return {
[=](const connection_closed_msg&) {
CAF_MESSAGE("received connection_closed_msg");
self->quit();
......@@ -124,8 +119,8 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
[=](pong_atom, int value) {
CAF_MESSAGE("received: pong " << value);
write(pong_atom::value, value);
}
);
},
};
}
behavior peer_acceptor_fun(broker* self, const actor& buddy) {
......@@ -141,7 +136,7 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) {
if (!res)
return std::move(res.error());
return res->second;
}
},
};
}
......@@ -153,8 +148,8 @@ void run_client(int argc, char** argv, uint16_t port) {
actor_system system{cfg};
auto p = system.spawn(ping, size_t{10});
CAF_MESSAGE("spawn_client...");
auto cl = unbox(system.middleman().spawn_client(peer_fun,
"127.0.0.1", port, p));
auto cl = unbox(
system.middleman().spawn_client(peer_fun, "127.0.0.1", port, p));
CAF_MESSAGE("spawn_client finished");
anon_send(p, kickoff_atom::value, cl);
CAF_MESSAGE("`kickoff_atom` has been send");
......@@ -171,17 +166,15 @@ void run_server(int argc, char** argv) {
auto serv = system.middleman().spawn_broker(peer_acceptor_fun,
system.spawn(pong));
std::thread child;
self->request(serv, infinite, publish_atom::value).receive(
[&](uint16_t port) {
CAF_MESSAGE("server is running on port " << port);
child = std::thread([=] {
run_client(argc, argv, port);
self->request(serv, infinite, publish_atom::value)
.receive(
[&](uint16_t port) {
CAF_MESSAGE("server is running on port " << port);
child = std::thread([=] { run_client(argc, argv, port); });
},
[&](const error& err) {
CAF_ERROR("Error: " << self->system().render(err));
});
},
[&](const error& err) {
CAF_ERROR("Error: " << self->system().render(err));
}
);
self->await_all_other_actors_done();
child.join();
}
......
......@@ -21,14 +21,14 @@
#define CAF_SUITE io_http_broker
#include "caf/test/unit_test.hpp"
#include <cassert>
#include <algorithm>
#include <cassert>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using std::cout;
using std::cerr;
using std::cout;
using std::endl;
using namespace caf;
......@@ -131,17 +131,19 @@ behavior http_worker(http_broker* self, connection_handle hdl) {
} else if (msg.buf.back() == '\n') {
self->state.ps = receive_new_line; // we've got a clean cut
} else {
self->state.ps = receive_continued_line; // interrupted in the middle
self->state.ps = receive_continued_line; // interrupted in the
// middle
}
// we don't need to check for completion in any intermediate state
if (self->state.ps != receive_new_line)
return;
// we have received the HTTP header if we have an empty line at the end
// we have received the HTTP header if we have an empty line at the
// end
if (lines.size() > 1 && lines.back().empty()) {
auto& out = self->wr_buf(hdl);
// we only look at the first line in our example and reply with our
// OK message if we receive exactly "GET / HTTP/1.1", otherwise
// we send a 404 HTTP response
// we only look at the first line in our example and reply with
// our OK message if we receive exactly "GET / HTTP/1.1",
// otherwise we send a 404 HTTP response
if (lines.front() == http_valid_get)
out.insert(out.end(), std::begin(http_ok), std::end(http_ok));
else
......@@ -151,9 +153,7 @@ behavior http_worker(http_broker* self, connection_handle hdl) {
self->quit();
}
},
[=](const connection_closed_msg&) {
self->quit();
}
[=](const connection_closed_msg&) { self->quit(); },
};
}
......@@ -163,15 +163,16 @@ behavior server(broker* self) {
[=](const new_connection_msg& ncm) {
CAF_MESSAGE("fork on new connection");
self->fork(http_worker, ncm.handle);
}
},
};
}
class fixture {
public:
using multiplexer_type = network::test_multiplexer;
fixture() : system(cfg.load<io::middleman, network::test_multiplexer>()) {
mpx_ = dynamic_cast<network::test_multiplexer*>(&system.middleman().backend());
mpx_ = dynamic_cast<multiplexer_type*>(&system.middleman().backend());
CAF_REQUIRE(mpx_ != nullptr);
// spawn the actor-under-test
aut_ = system.middleman().spawn_broker(server);
......@@ -202,10 +203,10 @@ public:
mock_t& expect(const std::string& x) {
auto& buf = this_->mpx_->output_buffer(this_->connection_);
CAF_REQUIRE((buf.size() >= x.size()));
CAF_REQUIRE((std::equal(buf.begin(),
buf.begin() + static_cast<ptrdiff_t>(x.size()),
x.begin())));
CAF_REQUIRE(buf.size() >= x.size());
CAF_REQUIRE(std::equal(buf.begin(),
buf.begin() + static_cast<ptrdiff_t>(x.size()),
x.begin()));
buf.erase(buf.begin(), buf.begin() + static_cast<ptrdiff_t>(x.size()));
return *this;
}
......
......@@ -45,29 +45,26 @@ public:
};
struct fixture {
template <class T, class... Ts>
std::vector<char> serialize(T& x, Ts&... xs) {
std::vector<char> buf;
binary_serializer bs{&context, buf};
bs(x, xs...);
return buf;
}
template <class T, class... Ts>
std::vector<char> serialize(T& x, Ts&... xs) {
std::vector<char> buf;
binary_serializer bs{&context, buf};
bs(x, xs...);
return buf;
}
template <class T, class... Ts>
void deserialize(const std::vector<char>& buf, T& x, Ts&... xs) {
binary_deserializer bd{&context, buf};
bd(x, xs...);
}
fixture() : cfg(), system(cfg), context(&system) {
}
template <class T, class... Ts>
void deserialize(const std::vector<char>& buf, T& x, Ts&... xs) {
binary_deserializer bd{&context, buf};
bd(x, xs...);
}
config cfg;
actor_system system;
scoped_execution_unit context;
fixture() : cfg(), system(cfg), context(&system) {
}
config cfg;
actor_system system;
scoped_execution_unit context;
};
} // namespace
......
......@@ -35,7 +35,7 @@ struct fixture {
receive_buffer b;
std::vector<char> vec;
fixture() : b(1024ul), vec{'h', 'a', 'l', 'l', 'o'} {
fixture() : b(1024ul), vec{'h', 'a', 'l', 'l', 'o'} {
// nop
}
......@@ -101,27 +101,24 @@ CAF_TEST(resize) {
a.resize(1024);
std::fill(a.begin(), a.end(), 'a');
auto cnt = 0;
CAF_CHECK(std::all_of(a.begin(), a.end(),
[&](char c) {
++cnt;
return c == 'a';
}));
CAF_CHECK(std::all_of(a.begin(), a.end(), [&](char c) {
++cnt;
return c == 'a';
}));
CAF_CHECK_EQUAL(cnt, 1024);
a.resize(10);
cnt = 0;
CAF_CHECK(std::all_of(a.begin(), a.end(),
[&](char c) {
++cnt;
return c == 'a';
}));
CAF_CHECK(std::all_of(a.begin(), a.end(), [&](char c) {
++cnt;
return c == 'a';
}));
CAF_CHECK_EQUAL(cnt, 10);
a.resize(1024);
cnt = 0;
CAF_CHECK(std::all_of(a.begin(), a.end(),
[&](char c) {
++cnt;
return c == 'a';
}));
CAF_CHECK(std::all_of(a.begin(), a.end(), [&](char c) {
++cnt;
return c == 'a';
}));
CAF_CHECK_EQUAL(cnt, 1024);
}
......@@ -136,7 +133,7 @@ CAF_TEST(push_back) {
}
CAF_TEST(insert) {
for (auto c: vec)
for (auto c : vec)
a.insert(a.end(), c);
CAF_CHECK_EQUAL(as_string(a), "hallo");
a.insert(a.begin(), '!');
......
This diff is collapsed.
......@@ -21,8 +21,8 @@
#define CAF_SUITE io_dynamic_remote_group
#include "caf/test/io_dsl.hpp"
#include <vector>
#include <algorithm>
#include <vector>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
......@@ -49,11 +49,7 @@ size_t received_messages = 0u;
behavior group_receiver(event_based_actor* self) {
self->set_default_handler(reflect_and_quit);
return {
[](ok_atom) {
++received_messages;
}
};
return {[](ok_atom) { ++received_messages; }};
}
// Our server is `mars` and our client is `earth`.
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
......@@ -18,10 +18,10 @@
#include "caf/openssl/remote_actor.hpp"
#include "caf/sec.hpp"
#include "caf/atom.hpp"
#include "caf/expected.hpp"
#include "caf/function_view.hpp"
#include "caf/sec.hpp"
#include "caf/openssl/manager.hpp"
......@@ -54,4 +54,3 @@ expected<strong_actor_ptr> remote_actor(actor_system& sys,
} // namespace openssl
} // namespace caf
This diff is collapsed.
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