Commit 58ca2f3b authored by Joseph Noir's avatar Joseph Noir

Add doorman policy

This is a transport policy for endpoint managers to accept TCP
connections. A new scribe will be created to handle subsequent
communication via the socket returned by accept.
parent 7f2798df
...@@ -11,11 +11,13 @@ set(LIBCAF_NET_SRCS ...@@ -11,11 +11,13 @@ set(LIBCAF_NET_SRCS
src/datagram_socket.cpp src/datagram_socket.cpp
src/endpoint_manager.cpp src/endpoint_manager.cpp
src/host.cpp src/host.cpp
src/interfaces.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/network_socket.cpp src/network_socket.cpp
src/pipe_socket.cpp src/pipe_socket.cpp
src/pollset_updater.cpp src/pollset_updater.cpp
src/socket.cpp src/socket.cpp
src/socket_guard.cpp
src/socket_manager.cpp src/socket_manager.cpp
src/stream_socket.cpp src/stream_socket.cpp
src/scribe.cpp src/scribe.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <vector>
#include "caf/net/ip.hpp"
#include "caf/optional.hpp"
namespace caf {
namespace net {
/// Utility class bundling access to network interface names and addresses.
class interfaces {
public:
/// Returns the host and protocol available for a local server socket
static std::vector<std::pair<std::string, ip>>
server_address(uint16_t port, const char* host,
optional<ip> preferred = none);
};
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <cstddef>
#include <string>
namespace caf {
namespace net {
// IP version tag.
enum class ip {
v4,
v6
};
/// @relates ip
inline std::string to_string(ip x) {
return x == ip::v4 ? "IPv4" : "IPv6";
}
} // namespace net
} // namespace caf
...@@ -55,6 +55,10 @@ void close(socket x); ...@@ -55,6 +55,10 @@ void close(socket x);
/// @relates socket /// @relates socket
std::errc last_socket_error(); std::errc last_socket_error();
/// Returns the string representation of a given socket error code.
/// @relates socket
std::string socket_error_as_string(std::errc err);
/// Returns the last socket error as human-readable string. /// Returns the last socket error as human-readable string.
/// @relates socket /// @relates socket
std::string last_socket_error_as_string(); std::string last_socket_error_as_string();
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/net/socket.hpp"
namespace caf {
namespace net {
/// Closes the guarded socket when destroyed.
class socket_guard {
public:
explicit socket_guard(net::socket fd);
~socket_guard();
net::socket release();
void close();
private:
net::socket fd_;
};
} // namespace net
} // namespace caf
...@@ -52,6 +52,10 @@ public: ...@@ -52,6 +52,10 @@ public:
return handle_; return handle_;
} }
multiplexer_ptr multiplexer() const {
return parent_.lock();
}
/// Returns registered operations (read, write, or both). /// Returns registered operations (read, write, or both).
operation mask() const noexcept; operation mask() const noexcept;
......
...@@ -18,8 +18,13 @@ ...@@ -18,8 +18,13 @@
#pragma once #pragma once
#include <cstdint>
#include "caf/expected.hpp"
#include "caf/variant.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/network_socket.hpp" #include "caf/net/network_socket.hpp"
#include "caf/sec.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
...@@ -77,5 +82,13 @@ variant<size_t, sec> write(stream_socket x, span<const byte> buf); ...@@ -77,5 +82,13 @@ variant<size_t, sec> write(stream_socket x, span<const byte> buf);
variant<size_t, sec> variant<size_t, sec>
check_stream_socket_io_res(std::make_signed<size_t>::type res); check_stream_socket_io_res(std::make_signed<size_t>::type res);
/// Creates a new TCP socket to accept connections on a given `port`
/// optionally limiting connections initiated from `addr`. Setting
/// `reuse_addr` to true allows rebinding to a port already in use.
/// @relates stream_socket
expected<stream_socket> make_accept_socket(uint16_t port,
const char* addr = nullptr,
bool reuse_addr = false);
} // namespace net } // namespace net
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <vector>
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/logger.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/policy/scribe.hpp"
#include "caf/send.hpp"
namespace caf {
namespace policy {
/// A doorman accepts TCP connections and creates scribes to handle them.
class doorman {
public:
doorman(net::stream_socket acceptor) : acceptor_(acceptor) {
// nop
}
net::socket acceptor_;
net::socket handle() {
return acceptor_;
}
template <class Parent>
error init(Parent&) {
return none;
}
template <class Parent>
bool handle_read_event(Parent& parent) {
auto sck = accept(acceptor_.id, nullptr, nullptr);
if (sck == net::invalid_socket_id) {
auto err = net::last_socket_error();
if (err != std::errc::operation_would_block
&& err != std::errc::resource_unavailable_try_again) {
CAF_LOG_ERROR("accept failed:" << net::socket_error_as_string(err));
return false;
}
return false;
}
auto mpx = parent.multiplexer();
if (!mpx) {
CAF_LOG_DEBUG("could not acquire multiplexer to create a new endpoint manager");
return false;
}
auto child = make_endpoint_manager(std::move(mpx), parent.system(),
scribe{sck},
parent.application().make());
if (auto err = child->init()) {
return false;
}
return true;
}
template <class Parent>
bool handle_write_event(Parent&) {
CAF_LOG_ERROR("doorman received write event");
return false;
}
template <class Parent>
void resolve(Parent&, const std::string& path, actor listener) {
CAF_LOG_ERROR("doorman called to resolve" << CAF_ARG(path));
anon_send(listener, resolve_atom::value, "doorman cannot resolve paths");
}
template <class Parent>
void timeout(Parent&, atom_value x, uint64_t id) {
CAF_LOG_ERROR("doorman received timeout" << CAF_ARG(x) << CAF_ARG(id));
CAF_IGNORE_UNUSED(x);
CAF_IGNORE_UNUSED(id);
}
template <class Application>
void handle_error(Application&, sec) {
close(acceptor_);
}
};
} // namespace policy
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/interfaces.hpp"
#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>
#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>
#endif
#include <map>
#include <string>
#include <memory>
#include <utility>
#include "caf/detail/socket_sys_includes.hpp"
namespace caf {
namespace net {
namespace {
// {interface_name => {protocol => address}}
using interfaces_map = std::map<std::string,
std::map<ip, std::vector<std::string>>>;
template <class T>
void* vptr(T* ptr) {
return static_cast<void*>(ptr);
}
void* fetch_in_addr(int family, sockaddr* addr) {
if (family == AF_INET)
return vptr(&reinterpret_cast<sockaddr_in*>(addr)->sin_addr);
return vptr(&reinterpret_cast<sockaddr_in6*>(addr)->sin6_addr);
}
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;
}
} // namespace
std::vector<std::pair<std::string, ip>>
interfaces::server_address(uint16_t port, const char* host,
optional<ip> preferred) {
using addr_pair = std::pair<std::string, ip>;
addrinfo hint;
memset(&hint, 0, sizeof(hint));
hint.ai_socktype = SOCK_STREAM;
if (preferred)
hint.ai_family = *preferred == ip::v4 ? AF_INET : AF_INET6;
else
hint.ai_family = AF_UNSPEC;
if (host == nullptr)
hint.ai_flags = AI_PASSIVE;
auto port_str = std::to_string(port);
addrinfo* tmp = nullptr;
if (getaddrinfo(host, port_str.c_str(), &hint, &tmp) != 0)
return {};
std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo};
char buffer[INET6_ADDRSTRLEN];
// Take the first ipv6 address or the first available address otherwise
std::vector<addr_pair> results;
for (auto i = addrs.get(); i != nullptr; i = i->ai_next) {
auto family = fetch_addr_str(true, true, buffer, i->ai_addr);
if (family != AF_UNSPEC) {
results.emplace_back(std::string{buffer},
family == AF_INET ? ip::v4 : ip::v6);
}
}
std::stable_sort(std::begin(results), std::end(results),
[](const addr_pair& lhs, const addr_pair& rhs) {
return lhs.second > rhs.second;
});
return results;
}
} // namespace net
} // namespace caf
...@@ -150,6 +150,10 @@ std::errc last_socket_error() { ...@@ -150,6 +150,10 @@ std::errc last_socket_error() {
return static_cast<std::errc>(errno); return static_cast<std::errc>(errno);
} }
std::string socket_error_as_string(std::errc err) {
return strerror(static_cast<int>(err));
}
std::string last_socket_error_as_string() { std::string last_socket_error_as_string() {
return strerror(errno); return strerror(errno);
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/socket_guard.hpp"
#include "caf/logger.hpp"
namespace caf {
namespace net {
socket_guard::socket_guard(net::socket fd) : fd_(fd) {
// nop
}
socket_guard::~socket_guard() {
close();
}
net::socket socket_guard::release() {
auto fd = fd_;
fd_ = net::invalid_socket;
return fd;
}
void socket_guard::close() {
if (fd_ != net::invalid_socket) {
CAF_LOG_DEBUG("close socket" << CAF_ARG(fd_));
net::close(fd_);
fd_ = net::invalid_socket;
}
}
} // namespace net
} // namespace caf
...@@ -22,14 +22,104 @@ ...@@ -22,14 +22,104 @@
#include "caf/detail/net_syscall.hpp" #include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp" #include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/interfaces.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
namespace {
// Save ourselves some typing.
constexpr auto ipv4 = caf::net::ip::v4;
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
expected<void> set_inaddr_any(socket, sockaddr_in& sa) {
sa.sin_addr.s_addr = INADDR_ANY;
return unit;
}
expected<void> set_inaddr_any(socket x, sockaddr_in6& sa) {
sa.sin6_addr = in6addr_any;
// also accept ipv4 requests on this socket
int off = 0;
CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<setsockopt_ptr>(&off),
static_cast<socket_size_type>(sizeof(off))));
return unit;
}
template <int Family, int SockType = SOCK_STREAM>
caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr, bool any) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
int socktype = SockType;
#ifdef SOCK_CLOEXEC
socktype |= SOCK_CLOEXEC;
#endif
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(Family, socktype, 0));
child_process_inherit(fd, false);
// sguard closes the socket in case of exception
socket_guard sguard{fd};
if (reuse_addr) {
int on = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<setsockopt_ptr>(&on),
static_cast<socket_size_type>(sizeof(on))));
}
using sockaddr_type =
typename std::conditional<
Family == AF_INET,
sockaddr_in,
sockaddr_in6
>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
family_of(sa) = Family;
if (any)
set_inaddr_any(fd, sa);
CAF_NET_SYSCALL("inet_pton", tmp, !=, 1,
inet_pton(Family, addr, &addr_of(sa)));
port_of(sa) = htons(port);
CAF_NET_SYSCALL("bind", res, !=, 0,
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<socket_size_type>(sizeof(sa))));
return sguard.release();
}
} // namespace
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
constexpr int no_sigpipe_io_flag = 0; constexpr int no_sigpipe_io_flag = 0;
...@@ -189,5 +279,40 @@ check_stream_socket_io_res(std::make_signed<size_t>::type res) { ...@@ -189,5 +279,40 @@ check_stream_socket_io_res(std::make_signed<size_t>::type res) {
return static_cast<size_t>(res); return static_cast<size_t>(res);
} }
expected<stream_socket> make_accept_socket(uint16_t port, const char* addr,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
auto addrs = interfaces::server_address(port, addr);
auto addr_str = std::string{addr == nullptr ? "" : addr};
if (addrs.empty())
return make_error(sec::cannot_open_port, "No local interface available",
addr_str);
bool any = addr_str.empty() || addr_str == "::" || addr_str == "0.0.0.0";
socket fd = invalid_socket;
for (auto& elem : addrs) {
auto hostname = elem.first.c_str();
auto p = elem.second == ipv4
? new_ip_acceptor_impl<AF_INET>(port, hostname, reuse_addr, any)
: new_ip_acceptor_impl<AF_INET6>(port, hostname, reuse_addr, any);
if (!p) {
CAF_LOG_DEBUG(p.error());
continue;
}
fd = *p;
break;
}
if (fd == invalid_socket_id) {
CAF_LOG_WARNING("could not open tcp socket on:" << CAF_ARG(port)
<< CAF_ARG(addr_str));
return make_error(sec::cannot_open_port, "tcp socket creation failed",
port, addr_str);
}
socket_guard sguard{fd.id};
CAF_NET_SYSCALL("listen", tmp2, !=, 0, listen(fd.id, SOMAXCONN));
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd.id));
return socket_cast<stream_socket>(sguard.release());
}
} // namespace net } // namespace net
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE doorman
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/policy/doorman.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
using namespace caf;
using namespace caf::net;
namespace {
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
multiplexer_ptr mpx;
};
class dummy_application {
public:
static expected<std::vector<char>> serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<char> result;
binary_serializer sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
}
template <class Transport>
error init(Transport&) {
return none;
}
template <class Transport>
bool handle_read_event(Transport&) {
return false;
}
template <class Transport>
bool handle_write_event(Transport&) {
return false;
}
template <class Transport>
void resolve(Transport&, std::string path, actor listener) {
anon_send(listener, resolve_atom::value, "the resolved path is still " + path);
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
// nop
}
void handle_error(sec) {
// nop
}
};
class dummy_application_factory {
public:
static expected<std::vector<char>> serialize(actor_system& sys,
const type_erased_tuple& x) {
return dummy_application::serialize(sys, x);
}
dummy_application make() const {
return dummy_application{};
}
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(doorman_tests, fixture)
CAF_TEST(doorman creation) {
auto acceptor = unbox(make_accept_socket(0, nullptr, false));
auto mgr = make_endpoint_manager(mpx, sys,
policy::doorman{acceptor},
dummy_application_factory{});
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -71,4 +71,9 @@ CAF_TEST(connected socket pair) { ...@@ -71,4 +71,9 @@ CAF_TEST(connected socket pair) {
close(x.second); close(x.second);
} }
CAF_TEST(create accept socket) {
auto acceptor = unbox(make_accept_socket(0, nullptr, false));
close(acceptor);
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
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