Commit a1a02b1c authored by Dominik Charousset's avatar Dominik Charousset

refactored network layer to use stream interfaces rather than sockets

parent e3ac52c8
......@@ -95,9 +95,10 @@ set(LIBCPPA_SRC
src/fiber.cpp
src/group.cpp
src/group_manager.cpp
src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp
src/local_actor.cpp
src/mailman.cpp
src/native_socket.cpp
src/network_manager.cpp
src/object.cpp
src/object_array.cpp
......
......@@ -59,7 +59,6 @@ cppa/detail/list_member.hpp
cppa/detail/mailman.hpp
cppa/detail/map_member.hpp
cppa/detail/matches.hpp
cppa/detail/native_socket.hpp
cppa/detail/network_manager.hpp
cppa/detail/object_array.hpp
cppa/detail/object_impl.hpp
......@@ -201,7 +200,6 @@ src/group.cpp
src/group_manager.cpp
src/local_actor.cpp
src/mailman.cpp
src/native_socket.cpp
src/network_manager.cpp
src/object.cpp
src/object_array.cpp
......@@ -257,3 +255,11 @@ cppa/message_id.hpp
cppa/cppa_fwd.hpp
cppa/timeout_definition.hpp
cppa/detail/behavior_impl.hpp
cppa/util/output_stream.hpp
cppa/util/input_stream.hpp
cppa/util/io_stream.hpp
cppa/util/acceptor.hpp
cppa/detail/ipv4_io_stream.hpp
src/ipv4_io_stream.cpp
cppa/detail/ipv4_acceptor.hpp
src/ipv4_acceptor.cpp
......@@ -81,4 +81,26 @@
#define CPPA_CRITICAL(error) CPPA_CRITICAL__(error, __FILE__, __LINE__)
#ifdef CPPA_WINDOWS
#else
# include <unistd.h>
#endif
namespace cppa {
#ifdef CPPA_WINDOWS
typedef SOCKET native_socket_type;
typedef const char* socket_send_ptr;
typedef char* socket_recv_ptr;
constexpr SOCKET invalid_socket = INVALID_SOCKET;
#else
typedef int native_socket_type;
typedef const void* socket_send_ptr;
typedef void* socket_recv_ptr;
constexpr int invalid_socket = -1;
inline void closesocket(native_socket_type fd) { close(fd); }
#endif
} // namespace cppa
#endif // CPPA_CONFIG_HPP
......@@ -31,12 +31,11 @@
#ifndef CPPA_BUFFER_HPP
#define CPPA_BUFFER_HPP
#include <ios> // std::ios_base::failure
#include <ios> // std::ios_base::failure
#include <cstring>
#include <iostream>
#include <string.h>
#include "cppa/detail/native_socket.hpp"
#include "cppa/util/input_stream.hpp"
namespace cppa { namespace detail {
......@@ -48,39 +47,6 @@ class buffer {
size_t m_allocated;
size_t m_final_size;
template<typename F>
bool append_impl(F&& fun, bool throw_on_error) {
auto recv_result = fun();
if (recv_result == 0) {
// connection closed
if (throw_on_error) {
throw std::ios_base::failure("cannot read from a closed pipe/socket");
}
return false;
}
else if (recv_result < 0) {
switch (errno) {
case EAGAIN: {
// rdflags or sfd is set to non-blocking,
// this is not treated as error
return true;
}
default: {
// a "real" error occured;
if (throw_on_error) {
char* cstr = strerror(errno);
std::string errmsg = cstr;
free(cstr);
throw std::ios_base::failure(std::move(errmsg));
}
return false;
}
}
}
inc_written(static_cast<size_t>(recv_result));
return true;
}
public:
buffer() : m_data(nullptr), m_written(0), m_allocated(0), m_final_size(0) {
......@@ -153,20 +119,13 @@ class buffer {
return remaining() == 0;
}
bool append_from_file_descriptor(int fd, bool throw_on_error = false) {
auto fun = [=]() -> int {
return ::read(fd, this->wr_ptr(), this->remaining());
};
return append_impl(fun, throw_on_error);
}
bool append_from(native_socket_type sfd,
int rdflags = 0,
bool throw_on_error = false) {
auto fun = [=]() -> int {
return ::recv(sfd, this->wr_ptr(), this->remaining(), rdflags);
};
return append_impl(fun, throw_on_error);
bool append_from(util::input_stream* istream) {
auto num_bytes = istream->read_some(wr_ptr(), remaining());
if (num_bytes > 0) {
inc_written(num_bytes);
return true;
}
return false;
}
};
......
......@@ -28,35 +28,39 @@
\******************************************************************************/
#ifndef CPPA_NATIVE_SOCKET_HPP
#define CPPA_NATIVE_SOCKET_HPP
#ifndef CPPA_IPV4_ACCEPTOR_HPP
#define CPPA_IPV4_ACCEPTOR_HPP
#include <memory>
#include <cstdint>
#include "cppa/config.hpp"
#ifdef CPPA_WINDOWS
#else
# include <netdb.h>
# include <unistd.h>
# include <sys/types.h>
# include <sys/socket.h>
# include <netinet/in.h>
#endif
#include "cppa/util/acceptor.hpp"
namespace cppa { namespace detail {
#ifdef CPPA_WINDOWS
typedef SOCKET native_socket_type;
typedef const char* socket_send_ptr;
typedef char* socket_recv_ptr;
constexpr SOCKET invalid_socket = INVALID_SOCKET;
#else
typedef int native_socket_type;
typedef const void* socket_send_ptr;
typedef void* socket_recv_ptr;
constexpr int invalid_socket = -1;
void closesocket(native_socket_type s);
#endif
class ipv4_acceptor : public util::acceptor {
public:
static std::unique_ptr<util::acceptor> create(std::uint16_t port);
~ipv4_acceptor();
native_socket_type acceptor_file_handle() const;
util::io_stream_ptr_pair accept_connection();
option<util::io_stream_ptr_pair> try_accept_connection();
private:
ipv4_acceptor(native_socket_type fd, bool nonblocking);
native_socket_type m_fd;
bool m_is_nonblocking;
};
} } // namespace cppa::detail
#endif // CPPA_NATIVE_SOCKET_HPP
#endif // CPPA_IPV4_ACCEPTOR_HPP
......@@ -28,37 +28,40 @@
\******************************************************************************/
#include "cppa/config.hpp"
#ifndef CPPA_IPV4_IO_STREAM_HPP
#define CPPA_IPV4_IO_STREAM_HPP
#include <ios> // ios_base::failure
#include <errno.h>
#include <sstream>
#include "cppa/detail/native_socket.hpp"
#include "cppa/config.hpp"
#include "cppa/util/io_stream.hpp"
namespace cppa { namespace detail {
#ifndef CPPA_WINDOWS
void closesocket(native_socket_type s) {
if(::close(s) != 0) {
switch(errno) {
case EBADF: {
throw std::ios_base::failure("EBADF: invalid socket");
}
case EINTR: {
throw std::ios_base::failure("EINTR: interrupted");
}
case EIO: {
throw std::ios_base::failure("EIO: an I/O error occured");
}
default: {
std::ostringstream oss;
throw std::ios_base::failure("");
}
}
}
}
#endif
class ipv4_io_stream : public util::io_stream {
public:
static util::io_stream_ptr connect_to(const char* host, std::uint16_t port);
ipv4_io_stream(native_socket_type fd);
native_socket_type read_file_handle() const;
native_socket_type write_file_handle() const;
void read(void* buf, size_t len);
size_t read_some(void* buf, size_t len);
void write(const void* buf, size_t len);
size_t write_some(const void* buf, size_t len);
private:
native_socket_type m_fd;
};
} } // namespace cppa::detail
#endif // CPPA_IPV4_IO_STREAM_HPP
......@@ -34,12 +34,35 @@
#include "cppa/any_tuple.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/process_information.hpp"
#include "cppa/detail/native_socket.hpp"
#include "cppa/detail/addressed_message.hpp"
#include "cppa/util/acceptor.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace detail {
enum class mm_message_type {
outgoing_message,
add_peer,
shutdown
};
struct mm_message {
mm_message* next;
mm_message_type type;
union {
std::pair<process_information_ptr, addressed_message> out_msg;
std::pair<util::io_stream_ptr_pair, process_information_ptr> peer;
};
mm_message();
mm_message(process_information_ptr, addressed_message);
mm_message(util::io_stream_ptr_pair, process_information_ptr);
~mm_message();
template<typename... Args>
static inline std::unique_ptr<mm_message> create(Args&&... args) {
return std::unique_ptr<mm_message>(new mm_message(std::forward<Args>(args)...));
}
};
void mailman_loop();
}} // namespace cppa::detail
......
......@@ -31,10 +31,15 @@
#ifndef CPPA_NETWORK_MANAGER_HPP
#define CPPA_NETWORK_MANAGER_HPP
#include "cppa/detail/post_office.hpp"
#include <memory>
//#include "cppa/detail/post_office.hpp"
namespace cppa { namespace detail {
struct po_message;
struct mm_message;
class network_manager {
public:
......@@ -45,11 +50,9 @@ class network_manager {
virtual void stop() = 0;
virtual void send_to_post_office(const po_message& msg) = 0;
virtual void send_to_post_office(any_tuple msg) = 0;
virtual void send_to_post_office(std::unique_ptr<po_message> msg) = 0;
virtual void send_to_mailman(any_tuple msg) = 0;
virtual void send_to_mailman(std::unique_ptr<mm_message> msg) = 0;
static network_manager* create_singleton();
......
......@@ -32,30 +32,75 @@
#define CPPA_POST_OFFICE_HPP
#include <memory>
#include <utility>
#include "cppa/atom.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/detail/native_socket.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
#include "cppa/util/acceptor.hpp"
#include "cppa/util/io_stream.hpp"
#include "cppa/detail/network_manager.hpp"
#include "cppa/detail/singleton_manager.hpp"
namespace cppa { namespace detail {
enum class po_message_type {
add_peer,
rm_peer,
publish,
unpublish,
shutdown
};
struct po_message {
atom_value flag;
native_socket_type fd;
actor_id aid;
po_message* next;
const po_message_type type;
union {
std::pair<util::io_stream_ptr_pair, process_information_ptr> new_peer;
util::io_stream_ptr_pair peer_streams;
std::pair<std::unique_ptr<util::acceptor>, actor_ptr> new_published_actor;
actor_ptr published_actor;
};
po_message();
po_message(util::io_stream_ptr_pair, process_information_ptr);
po_message(util::io_stream_ptr_pair);
po_message(std::unique_ptr<util::acceptor>, actor_ptr);
po_message(actor_ptr);
~po_message();
template<typename... Args>
static inline std::unique_ptr<po_message> create(Args&&... args) {
return std::unique_ptr<po_message>(new po_message(std::forward<Args>(args)...));
}
};
void post_office_loop(int input_fd);
void post_office_loop(int input_fd, intrusive::single_reader_queue<po_message>&);
template<typename... Args>
inline void send2po(Args&&... args) {
auto nm = singleton_manager::get_network_manager();
nm->send_to_post_office(std::unique_ptr<po_message>(new po_message(std::forward<Args>(args)...)));
}
void post_office_add_peer(native_socket_type peer_socket,
const process_information_ptr& peer_ptr);
inline void post_office_add_peer(util::io_stream_ptr_pair peer_streams,
process_information_ptr peer_ptr ) {
send2po(std::move(peer_streams), std::move(peer_ptr));
}
void post_office_publish(native_socket_type server_socket,
const actor_ptr& published_actor);
inline void post_office_close_peer_connection(util::io_stream_ptr_pair peer_streams) {
send2po(std::move(peer_streams));
}
void post_office_unpublish(actor_id whom);
inline void post_office_publish(std::unique_ptr<util::acceptor> server,
actor_ptr published_actor ) {
send2po(std::move(server), std::move(published_actor));
}
void post_office_close_socket(native_socket_type sfd);
inline void post_office_unpublish(actor_ptr whom) {
send2po(std::move(whom));
}
} } // namespace cppa::detail
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_ACCEPTOR_HPP
#define CPPA_ACCEPTOR_HPP
#include "cppa/config.hpp"
#include "cppa/option.hpp"
#include "cppa/util/input_stream.hpp"
#include "cppa/util/output_stream.hpp"
namespace cppa { namespace util {
typedef std::pair<input_stream_ptr, output_stream_ptr> io_stream_ptr_pair;
class acceptor {
public:
virtual ~acceptor() { }
virtual native_socket_type acceptor_file_handle() const = 0;
virtual io_stream_ptr_pair accept_connection() = 0;
virtual option<io_stream_ptr_pair> try_accept_connection() = 0;
};
} } // namespace cppa::util
#endif // CPPA_ACCEPTOR_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_INPUT_STREAM_HPP
#define CPPA_INPUT_STREAM_HPP
#include "cppa/config.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa { namespace util {
class input_stream : public virtual ref_counted {
public:
virtual native_socket_type read_file_handle() const = 0;
/**
* @throws std::ios_base::failure
*/
virtual void read(void* buf, size_t num_bytes) = 0;
virtual size_t read_some(void* buf, size_t num_bytes) = 0;
};
typedef intrusive_ptr<input_stream> input_stream_ptr;
} } // namespace cppa::util
#endif // CPPA_INPUT_STREAM_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_IO_STREAM_HPP
#define CPPA_IO_STREAM_HPP
#include "cppa/util/input_stream.hpp"
#include "cppa/util/output_stream.hpp"
namespace cppa { namespace util {
class io_stream : public input_stream, public output_stream { };
typedef intrusive_ptr<io_stream> io_stream_ptr;
} } // namespace cppa::util
#endif // CPPA_IO_STREAM_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_OUTPUT_STREAM_HPP
#define CPPA_OUTPUT_STREAM_HPP
#include "cppa/config.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa { namespace util {
class output_stream : public virtual ref_counted {
public:
virtual native_socket_type write_file_handle() const = 0;
virtual void write(const void* buf, size_t num_bytes) = 0;
virtual size_t write_some(const void* buf, size_t num_bytes) = 0;
};
typedef intrusive_ptr<output_stream> output_stream_ptr;
} } // namespace cppa::util
#endif // CPPA_OUTPUT_STREAM_HPP
......@@ -50,7 +50,7 @@ void actor_proxy::forward_message(const process_information_ptr& piptr,
message_id_t id ) {
detail::addressed_message amsg{sender, this, std::move(msg), id};
detail::singleton_manager::get_network_manager()
->send_to_mailman(make_any_tuple(piptr, std::move(amsg)));
->send_to_mailman(detail::mm_message::create(piptr, std::move(amsg)));
}
void actor_proxy::enqueue(actor* sender, any_tuple msg) {
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include <ios>
#include <cstring>
#include <errno.h>
#include <iostream>
#include "cppa/exception.hpp"
#include "cppa/util/io_stream.hpp"
#include "cppa/detail/ipv4_acceptor.hpp"
#include "cppa/detail/ipv4_io_stream.hpp"
#ifdef CPPA_WINDOWS
#else
# include <netdb.h>
# include <unistd.h>
# include <sys/types.h>
# include <sys/socket.h>
# include <netinet/in.h>
# include <netinet/tcp.h>
# include <fcntl.h>
#endif
namespace cppa { namespace detail {
namespace {
struct socket_guard {
bool m_released;
native_socket_type m_socket;
public:
socket_guard(native_socket_type sfd) : m_released(false), m_socket(sfd) { }
~socket_guard() {
if (!m_released) closesocket(m_socket);
}
void release() {
m_released = true;
}
};
int rd_flags(native_socket_type fd) {
auto flags = fcntl(fd, F_GETFL, 0);
if (flags == -1) {
throw network_error("unable to read socket flags");
}
return flags;
}
bool accept_impl(util::io_stream_ptr_pair& result, native_socket_type fd, bool nonblocking) {
sockaddr addr;
socklen_t addrlen;
memset(&addr, 0, sizeof(addr));
memset(&addrlen, 0, sizeof(addrlen));
auto sfd = ::accept(fd, &addr, &addrlen);
if (sfd < 0) {
if (nonblocking && (errno == EAGAIN || errno == EWOULDBLOCK)) {
// ok, try again
return false;
}
char* cstr = strerror(errno);
std::string errmsg = cstr;
free(cstr);
throw std::ios_base::failure(std::move(errmsg));
}
int flags = 1;
setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(int));
util::io_stream_ptr ptr(new ipv4_io_stream(sfd));
result.first = ptr;
result.second = ptr;
return true;
}
} // namespace <anonymous>
ipv4_acceptor::ipv4_acceptor(native_socket_type fd, bool nonblocking)
: m_fd(fd), m_is_nonblocking(nonblocking) { }
std::unique_ptr<util::acceptor> ipv4_acceptor::create(std::uint16_t port) {
native_socket_type sockfd;
struct sockaddr_in serv_addr;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == invalid_socket) {
throw network_error("could not create server socket");
}
// sguard closes the socket in case of an exception
socket_guard sguard(sockfd);
memset((char*) &serv_addr, 0, sizeof(serv_addr));
serv_addr.sin_family = AF_INET;
serv_addr.sin_addr.s_addr = INADDR_ANY;
serv_addr.sin_port = htons(port);
if (bind(sockfd, (struct sockaddr*) &serv_addr, sizeof(serv_addr)) < 0) {
throw bind_failure(errno);
}
if (listen(sockfd, 10) != 0) {
throw network_error("listen() failed");
}
int flags = fcntl(sockfd, F_GETFL, 0);
if (flags == -1) {
throw network_error("unable to get socket flags");
}
bool nonblocking = (flags & O_NONBLOCK) != 0;
// ok, no exceptions
sguard.release();
std::unique_ptr<ipv4_acceptor> result(new ipv4_acceptor(sockfd, nonblocking));
return std::move(result);
}
ipv4_acceptor::~ipv4_acceptor() {
closesocket(m_fd);
}
native_socket_type ipv4_acceptor::acceptor_file_handle() const {
return m_fd;
}
util::io_stream_ptr_pair ipv4_acceptor::accept_connection() {
if (m_is_nonblocking) {
auto flags = rd_flags(m_fd);
if (fcntl(m_fd, F_SETFL, flags & (~(O_NONBLOCK))) < 0) {
throw network_error("unable to set socket to nonblock");
}
m_is_nonblocking = false;
}
util::io_stream_ptr_pair result;
accept_impl(result, m_fd, m_is_nonblocking);
return result;
}
option<util::io_stream_ptr_pair> ipv4_acceptor::try_accept_connection() {
if (!m_is_nonblocking) {
auto flags = rd_flags(m_fd);
if (fcntl(m_fd, F_SETFL, flags | O_NONBLOCK) < 0) {
throw network_error("unable to set socket to nonblock");
}
m_is_nonblocking = true;
}
util::io_stream_ptr_pair result;
if (accept_impl(result, m_fd, m_is_nonblocking)) {
return result;
}
return {};
}
} } // namespace cppa::detail
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include <ios>
#include <cstring>
#include <errno.h>
#include "cppa/exception.hpp"
#include "cppa/detail/ipv4_io_stream.hpp"
#ifdef CPPA_WINDOWS
#else
# include <netdb.h>
# include <unistd.h>
# include <sys/types.h>
# include <sys/socket.h>
# include <netinet/in.h>
# include <netinet/tcp.h>
# include <fcntl.h>
#endif
namespace cppa { namespace detail {
namespace {
template<typename T>
void handle_syscall_result(T result, size_t num_bytes, bool nonblocking) {
if (result < 0) {
if (!nonblocking || errno != EAGAIN) {
char* cstr = strerror(errno);
std::string errmsg = cstr;
free(cstr);
throw std::ios_base::failure(std::move(errmsg));
}
}
else if (result == 0) {
throw std::ios_base::failure("cannot read/write from closed socket");
}
else if (!nonblocking && static_cast<size_t>(result) != num_bytes) {
throw std::ios_base::failure("IO error on IPv4 socket");
}
}
} // namespace <anonymous>
ipv4_io_stream::ipv4_io_stream(native_socket_type fd) : m_fd(fd) { }
native_socket_type ipv4_io_stream::read_file_handle() const {
return m_fd;
}
native_socket_type ipv4_io_stream::write_file_handle() const {
return m_fd;
}
void ipv4_io_stream::read(void* buf, size_t len) {
handle_syscall_result(::recv(m_fd, buf, len, 0), len, false);
}
size_t ipv4_io_stream::read_some(void* buf, size_t len) {
auto recv_result = ::recv(m_fd, buf, len, MSG_DONTWAIT);
handle_syscall_result(recv_result, len, true);
return static_cast<size_t>(recv_result);
}
void ipv4_io_stream::write(const void* buf, size_t len) {
handle_syscall_result(::send(m_fd, buf, len, 0), len, false);
}
size_t ipv4_io_stream::write_some(const void* buf, size_t len) {
auto send_result = ::send(m_fd, buf, len, MSG_DONTWAIT);
handle_syscall_result(send_result, len, true);
return static_cast<size_t>(send_result);
}
util::io_stream_ptr ipv4_io_stream::connect_to(const char* host,
std::uint16_t port) {
native_socket_type sockfd;
struct sockaddr_in serv_addr;
struct hostent* server;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == invalid_socket) {
throw network_error("socket creation failed");
}
server = gethostbyname(host);
if (!server) {
std::string errstr = "no such host: ";
errstr += host;
throw network_error(std::move(errstr));
}
memset(&serv_addr, 0, sizeof(serv_addr));
serv_addr.sin_family = AF_INET;
memmove(&serv_addr.sin_addr.s_addr, server->h_addr, server->h_length);
serv_addr.sin_port = htons(port);
if (connect(sockfd, (const sockaddr*) &serv_addr, sizeof(serv_addr)) != 0) {
throw network_error("could not connect to host");
}
int flags = 1;
setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(int));
return new ipv4_io_stream(sockfd);
}
} } // namespace cppa::detail
......@@ -31,6 +31,17 @@
#include <atomic>
#include <iostream>
#ifdef CPPA_WINDOWS
#else
# include <netdb.h>
# include <unistd.h>
# include <sys/types.h>
# include <sys/socket.h>
# include <netinet/in.h>
# include <netinet/tcp.h>
# include <fcntl.h>
#endif
#include "cppa/cppa.hpp"
#include "cppa/to_string.hpp"
#include "cppa/detail/mailman.hpp"
......@@ -52,6 +63,46 @@ using std::endl;
// implementation of mailman.hpp
namespace cppa { namespace detail {
namespace {
template<typename T, typename... Args>
void call_ctor(T& var, Args&&... args) {
new (&var) T (std::forward<Args>(args)...);
}
template<typename T>
void call_dtor(T& var) {
var.~T();
}
} // namespace <anonymous>
mm_message::mm_message() : next(0), type(mm_message_type::shutdown) { }
mm_message::mm_message(process_information_ptr a0, addressed_message a1)
: next(0), type(mm_message_type::outgoing_message) {
call_ctor(out_msg, std::move(a0), std::move(a1));
}
mm_message::mm_message(util::io_stream_ptr_pair a0, process_information_ptr a1)
: next(0), type(mm_message_type::add_peer) {
call_ctor(peer, std::move(a0), std::move(a1));
}
mm_message::~mm_message() {
switch (type) {
case mm_message_type::outgoing_message: {
call_dtor(out_msg);
break;
}
case mm_message_type::add_peer: {
call_dtor(peer);
break;
}
default: break;
}
}
void mailman_loop() {
bool done = false;
// serializes outgoing messages
......@@ -81,7 +132,7 @@ void mailman_loop() {
if (disconnect_peer) {
DEBUG("peer disconnected (error during send)");
//closesocket(peer);
post_office_close_socket(peer_fd);
//post_office_close_socket(peer_fd);
peers.erase(i);
}
bs.reset();
......
......@@ -42,6 +42,8 @@
#include "cppa/scheduler.hpp"
#include "cppa/thread_mapped_actor.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
#include "cppa/detail/mailman.hpp"
#include "cppa/detail/post_office.hpp"
#include "cppa/detail/network_manager.hpp"
......@@ -53,10 +55,10 @@ using namespace cppa::detail;
struct network_manager_impl : network_manager {
local_actor_ptr m_mailman;
intrusive::single_reader_queue<mm_message> m_mailman_queue;
std::thread m_mailman_thread;
local_actor_ptr m_post_office;
intrusive::single_reader_queue<po_message> m_post_office_queue;
std::thread m_post_office_thread;
int pipe_fd[2];
......@@ -66,46 +68,44 @@ struct network_manager_impl : network_manager {
CPPA_CRITICAL("cannot create pipe");
}
// create actors
m_post_office.reset(new thread_mapped_actor);
m_mailman.reset(new thread_mapped_actor);
//m_post_office.reset(new thread_mapped_actor);
//m_mailman.reset(new thread_mapped_actor);
// store some data in local variables for lambdas
int pipe_fd0 = pipe_fd[0];
auto po_ptr = m_post_office;
auto mm_ptr = m_mailman;
// start threads
m_post_office_thread = std::thread([pipe_fd0, po_ptr]() {
scoped_self_setter sss{po_ptr.get()};
post_office_loop(pipe_fd0);
m_post_office_thread = std::thread([this, pipe_fd0] {
//scoped_self_setter sss{po_ptr.get()};
post_office_loop(pipe_fd0, this->m_post_office_queue);
});
m_mailman_thread = std::thread([mm_ptr]() {
scoped_self_setter sss{mm_ptr.get()};
m_mailman_thread = std::thread([] {
//scoped_self_setter sss{mm_ptr.get()};
mailman_loop();
});
}
void stop() { // override
m_mailman->enqueue(nullptr, make_any_tuple(atom("DONE")));
//m_mailman->enqueue(nullptr, make_any_tuple(atom("DONE")));
m_mailman_thread.join();
// wait until mailman is done; post_office closes all sockets
std::atomic_thread_fence(std::memory_order_seq_cst);
send_to_post_office(po_message{atom("DONE"), -1, 0});
m_post_office_queue.push_back(new po_message);
//send_to_post_office(po_message{atom("DONE"), -1, 0});
m_post_office_thread.join();
close(pipe_fd[0]);
close(pipe_fd[0]);
close(pipe_fd[1]);
}
void send_to_post_office(const po_message& msg) {
if (write(pipe_fd[1], &msg, sizeof(po_message)) != sizeof(po_message)) {
void send_to_post_office(std::unique_ptr<po_message> msg) {
m_post_office_queue.push_back(msg.release());
std::uint32_t dummy = 0;
if (write(pipe_fd[1], &dummy, sizeof(dummy)) != sizeof(dummy)) {
CPPA_CRITICAL("cannot write to pipe");
}
}
void send_to_post_office(any_tuple msg) {
m_post_office->enqueue(nullptr, std::move(msg));
}
void send_to_mailman(any_tuple msg) {
m_mailman->enqueue(nullptr, std::move(msg));
void send_to_mailman(std::unique_ptr<mm_message> msg) {
m_mailman_queue.push_back(msg.release());
//m_mailman->enqueue(nullptr, std::move(msg));
}
};
......
......@@ -56,11 +56,15 @@
#include "cppa/deserializer.hpp"
#include "cppa/binary_deserializer.hpp"
#include "cppa/util/acceptor.hpp"
#include "cppa/util/io_stream.hpp"
#include "cppa/util/input_stream.hpp"
#include "cppa/util/output_stream.hpp"
#include "cppa/detail/buffer.hpp"
#include "cppa/detail/mailman.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/post_office.hpp"
#include "cppa/detail/native_socket.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/network_manager.hpp"
#include "cppa/detail/singleton_manager.hpp"
......@@ -79,9 +83,11 @@ using std::cout;
using std::cerr;
using std::endl;
namespace cppa { namespace detail {
namespace {
cppa::detail::types_array<cppa::atom_value, cppa::actor_ptr> t_atom_actor_ptr_types;
types_array<atom_value, actor_ptr> t_atom_actor_ptr_types;
// allocate in 1KB chunks (minimize reallocations)
constexpr size_t s_chunk_size = 1024;
......@@ -92,57 +98,91 @@ constexpr size_t s_max_buffer_size = (1024 * 1024);
static_assert((s_max_buffer_size % s_chunk_size) == 0,
"max_buffer_size is not a multiple of chunk_size");
static_assert(sizeof(cppa::detail::native_socket_type) == sizeof(std::uint32_t),
static_assert(sizeof(native_socket_type) == sizeof(std::uint32_t),
"sizeof(native_socket_t) != sizeof(std::uint32_t)");
template<typename... Args>
inline void send2po(Args&&... args) {
auto nm = singleton_manager::get_network_manager();
nm->send_to_post_office(std::unique_ptr<po_message>(new po_message(std::forward<Args>(args)...)));
}
template<typename T, typename... Args>
void call_ctor(T& var, Args&&... args) {
new (&var) T (std::forward<Args>(args)...);
}
template<typename T>
void call_dtor(T& var) {
var.~T();
}
} // namespace <anonmyous>
namespace cppa { namespace detail {
po_message::po_message() : next(0), type(po_message_type::shutdown) { }
inline void send2po_(network_manager*) { }
po_message::po_message(util::io_stream_ptr_pair a0, process_information_ptr a1)
: next(0), type(po_message_type::add_peer) {
call_ctor(new_peer, std::move(a0), std::move(a1));
}
template<typename Arg0, typename... Args>
inline void send2po_(network_manager* nm, Arg0&& arg0, Args&&... args) {
nm->send_to_post_office(make_any_tuple(std::forward<Arg0>(arg0),
std::forward<Args>(args)...));
po_message::po_message(util::io_stream_ptr_pair a0)
: next(0), type(po_message_type::rm_peer) {
call_ctor(peer_streams, std::move(a0));
}
po_message::po_message(std::unique_ptr<util::acceptor> a0, actor_ptr a1)
: next(0), type(po_message_type::publish) {
call_ctor(new_published_actor, std::move(a0), std::move(a1));
}
template<typename... Args>
inline void send2po(const po_message& msg, Args&&... args) {
auto nm = singleton_manager::get_network_manager();
nm->send_to_post_office(msg);
send2po_(nm, std::forward<Args>(args)...);
po_message::po_message(actor_ptr a0)
: next(0), type(po_message_type::unpublish) {
call_ctor(published_actor, std::move(a0));
}
class po_socket_handler {
po_message::~po_message() {
switch (type) {
case po_message_type::add_peer: {
call_dtor(new_peer);
break;
}
case po_message_type::rm_peer: {
call_dtor(peer_streams);
break;
}
case po_message_type::publish: {
call_dtor(new_published_actor);
break;
}
case po_message_type::unpublish: {
call_dtor(published_actor);
break;
}
default: break;
}
}
public:
class post_office_worker {
po_socket_handler(native_socket_type fd) : m_socket(fd) { }
public:
virtual ~po_socket_handler() { }
virtual ~post_office_worker() { }
// returns bool if either done or an error occured
// returns false if either done or an error occured
virtual bool read_and_continue() = 0;
native_socket_type get_socket() const {
return m_socket;
}
virtual native_socket_type get_socket() const = 0;
virtual bool is_doorman_of(actor_id) const { return false; }
protected:
native_socket_type m_socket;
};
typedef std::unique_ptr<po_socket_handler> po_socket_handler_ptr;
typedef std::vector<po_socket_handler_ptr> po_socket_handler_vector;
typedef std::unique_ptr<post_office_worker> po_worker_ptr;
typedef std::vector<po_worker_ptr> po_worker_vector;
// represents a TCP connection to another peer
class po_peer : public po_socket_handler {
class po_peer : public post_office_worker {
enum state {
// connection just established; waiting for process information
......@@ -153,6 +193,9 @@ class po_peer : public po_socket_handler {
read_message
};
util::input_stream_ptr m_istream;
util::output_stream_ptr m_ostream;
state m_state;
// caches process_information::get()
process_information_ptr m_pself;
......@@ -163,21 +206,32 @@ class po_peer : public po_socket_handler {
// manages socket input
buffer<512, (16 * 1024 * 1024)> m_buf;
public:
po_peer(native_socket_type fd, process_information_ptr peer = nullptr)
: po_socket_handler(fd)
, m_state((peer) ? wait_for_msg_size : wait_for_process_info)
, m_pself(process_information::get())
, m_peer(std::move(peer))
, m_meta_msg(uniform_typeid<addressed_message>()) {
void init() {
m_state = (m_peer) ? wait_for_msg_size : wait_for_process_info;
m_pself = process_information::get();
m_meta_msg = uniform_typeid<addressed_message>();
m_buf.reset(m_state == wait_for_process_info
? sizeof(std::uint32_t) + process_information::node_id_size
: sizeof(std::uint32_t));
}
public:
po_peer(util::io_stream_ptr ios, process_information_ptr peer = nullptr)
: m_istream(ios)
, m_ostream(ios)
, m_peer(std::move(peer)) {
init();
}
po_peer(util::io_stream_ptr_pair spair, process_information_ptr peer = nullptr)
: m_istream(std::move(spair.first))
, m_ostream(std::move(spair.second))
, m_peer(std::move(peer)) {
init();
}
~po_peer() {
closesocket(m_socket);
if (m_peer) {
// collect all children (proxies to actors of m_peer)
std::vector<actor_proxy_ptr> children;
......@@ -196,12 +250,14 @@ class po_peer : public po_socket_handler {
}
}
inline native_socket_type get_socket() const { return m_socket; }
native_socket_type get_socket() const {
return m_istream->read_file_handle();
}
// @returns false if an error occured; otherwise true
bool read_and_continue() {
for (;;) {
if (m_buf.append_from(m_socket) == false) {
if (m_buf.append_from(m_istream.get()) == false) {
DEBUG("cannot read from socket");
return false;
}
......@@ -216,9 +272,12 @@ class po_peer : public po_socket_handler {
memcpy(node_id.data(), m_buf.data() + sizeof(std::uint32_t),
process_information::node_id_size);
m_peer.reset(new process_information(process_id, node_id));
util::io_stream_ptr_pair iop(m_istream, m_ostream);
// inform mailman about new peer
singleton_manager::get_network_manager()
->send_to_mailman(make_any_tuple(m_socket, m_peer));
->send_to_mailman(mm_message::create(iop, m_peer));
// forget the output stream (initialization done)
m_ostream.reset();
m_state = wait_for_msg_size;
m_buf.reset(sizeof(std::uint32_t));
DEBUG("pinfo read: "
......@@ -259,7 +318,7 @@ class po_peer : public po_socket_handler {
receiver->attach_functor([=](std::uint32_t reason) {
addressed_message kmsg{receiver, receiver, make_any_tuple(atom("KILL_PROXY"), reason)};
singleton_manager::get_network_manager()
->send_to_mailman(make_any_tuple(m_peer, kmsg));
->send_to_mailman(mm_message::create(m_peer, kmsg));
});
}
else {
......@@ -323,17 +382,18 @@ class po_peer : public po_socket_handler {
};
// accepts new connections to a published actor
class po_doorman : public po_socket_handler {
class po_doorman : public post_office_worker {
std::unique_ptr<util::acceptor> m_acceptor;
actor_id m_actor_id;
// caches process_information::get()
process_information_ptr m_pself;
po_socket_handler_vector* new_handler;
po_worker_vector* new_handler;
public:
po_doorman(actor_id aid, native_socket_type fd, po_socket_handler_vector* v)
: po_socket_handler(fd)
po_doorman(actor_id aid, std::unique_ptr<util::acceptor>&& acceptor, po_worker_vector* v)
: m_acceptor(std::move(acceptor))
, m_actor_id(aid), m_pself(process_information::get())
, new_handler(v) {
}
......@@ -342,66 +402,129 @@ class po_doorman : public po_socket_handler {
return m_actor_id == aid;
}
~po_doorman() {
closesocket(m_socket);
native_socket_type get_socket() const {
return m_acceptor->acceptor_file_handle();
}
bool read_and_continue() {
for (;;) {
sockaddr addr;
socklen_t addrlen;
memset(&addr, 0, sizeof(addr));
memset(&addrlen, 0, sizeof(addrlen));
auto sfd = ::accept(m_socket, &addr, &addrlen);
if (sfd < 0) {
switch (errno) {
case EAGAIN:
# if EAGAIN != EWOULDBLOCK
case EWOULDBLOCK:
# endif
// just try again
return true;
default:
perror("accept()");
DEBUG("accept failed (actor unpublished?)");
return false;
auto opt = m_acceptor->try_accept_connection();
if (opt) {
auto& pair = *opt;
std::uint32_t process_id = m_pself->process_id();
pair.second->write(&m_actor_id, sizeof(actor_id));
pair.second->write(&process_id, sizeof(std::uint32_t));
pair.second->write(m_pself->node_id().data(), m_pself->node_id().size());
new_handler->emplace_back(new po_peer(pair));
DEBUG("socket accepted; published actor: " << id);
}
}
return true;
}
};
class po_overseer : public post_office_worker {
public:
po_overseer(bool& done,
int pipe_fd,
po_worker_vector& handler,
po_worker_vector& new_handler,
std::vector<native_socket_type>& closed_sockets,
intrusive::single_reader_queue<po_message>& q )
: m_done(done)
, m_pipe_fd(pipe_fd)
, m_handler(handler)
, m_new_handler(new_handler)
, m_closed_sockets(closed_sockets)
, m_queue(q) { }
native_socket_type get_socket() const {
return m_pipe_fd;
}
bool read_and_continue() {
std::uint32_t dummy;
if (::read(m_pipe_fd, &dummy, sizeof(dummy)) != sizeof(dummy)) {
CPPA_CRITICAL("cannot read from pipe");
}
std::atomic_thread_fence(std::memory_order_seq_cst);
std::unique_ptr<po_message> msg;
msg.reset(m_queue.pop());
switch (msg->type) {
case po_message_type::add_peer: {
DEBUG("post_office: add_peer");
auto& new_peer = msg->new_peer;
m_new_handler.emplace_back(new po_peer(new_peer.first, new_peer.second));
break;
}
case po_message_type::rm_peer: {
DEBUG("post_office: rm_peer");
auto istream = msg->peer_streams.first;
if (istream) {
m_closed_sockets.emplace_back(istream->read_file_handle());
}
break;
}
int flags = 1;
setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(int));
flags = fcntl(sfd, F_GETFL, 0);
if (flags == -1) {
throw network_error("unable to get socket flags");
case po_message_type::publish: {
DEBUG("post_office: publish");
auto& ptrs = msg->new_published_actor;
m_new_handler.emplace_back(new po_doorman(ptrs.second->id(),
std::move(ptrs.first),
&m_new_handler));
break;
}
if (fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
throw network_error("unable to set socket to nonblock");
case po_message_type::unpublish: {
DEBUG("post_office: unpublish");
if (msg->published_actor) {
auto aid = msg->published_actor->id();
auto i = std::find_if(m_handler.begin(), m_handler.end(),
[aid](const po_worker_ptr& hp) {
return hp->is_doorman_of(aid);
});
if (i != m_handler.end()) {
m_closed_sockets.emplace_back((*i)->get_socket());
}
}
break;
}
case po_message_type::shutdown: {
DEBUG("post_office: shutdown");
m_done = true;
}
auto id = m_actor_id;
std::uint32_t process_id = m_pself->process_id();
::send(sfd, &id, sizeof(std::uint32_t), 0);
::send(sfd, &process_id, sizeof(std::uint32_t), 0);
::send(sfd, m_pself->node_id().data(), m_pself->node_id().size(), 0);
new_handler->emplace_back(new po_peer(sfd));
DEBUG("socket accepted; published actor: " << id);
}
return true;
}
private:
bool& m_done;
int m_pipe_fd;
po_worker_vector& m_handler;
po_worker_vector& m_new_handler;
std::vector<native_socket_type>& m_closed_sockets;
intrusive::single_reader_queue<po_message>& m_queue;
};
inline constexpr std::uint64_t valof(atom_value val) {
return static_cast<std::uint64_t>(val);
}
void post_office_loop(int input_fd) {
void post_office_loop(int input_fd, intrusive::single_reader_queue<po_message>& q) {
int maxfd = 0;
fd_set readset;
bool done = false;
po_socket_handler_vector handler;
po_socket_handler_vector new_handler;
po_worker_vector handler;
po_worker_vector new_handler;
std::vector<native_socket_type> closed_sockets;
handler.emplace_back(new po_overseer(done, input_fd, handler,
new_handler, closed_sockets, q));
do {
FD_ZERO(&readset);
FD_SET(input_fd, &readset);
maxfd = input_fd;
maxfd = 0;
for (auto& hptr : handler) {
auto fd = hptr->get_socket();
maxfd = std::max(maxfd, fd);
......@@ -426,62 +549,17 @@ void post_office_loop(int input_fd) {
}
}
}
if (FD_ISSET(input_fd, &readset)) {
DEBUG("post_office: read from pipe");
po_message msg;
if (read(input_fd, &msg, sizeof(po_message)) != sizeof(po_message)) {
CPPA_CRITICAL("cannot read from pipe");
}
switch (valof(msg.flag)) {
case valof(atom("ADD_PEER")): {
receive (
on_arg_match >> [&](native_socket_type fd,
process_information_ptr piptr) {
DEBUG("post_office: add_peer");
handler.emplace_back(new po_peer(fd, piptr));
}
);
break;
}
case valof(atom("RM_PEER")): {
DEBUG("post_office: rm_peer");
auto i = std::find_if(handler.begin(), handler.end(),
[&](const po_socket_handler_ptr& hp) {
return hp->get_socket() == msg.fd;
});
if (i != handler.end()) handler.erase(i);
break;
}
case valof(atom("PUBLISH")): {
receive (
on_arg_match >> [&](native_socket_type sockfd,
actor_ptr whom) {
DEBUG("post_office: publish_actor");
CPPA_REQUIRE(sockfd > 0);
CPPA_REQUIRE(whom.get() != nullptr);
handler.emplace_back(new po_doorman(whom->id(), sockfd, &new_handler));
}
);
break;
}
case valof(atom("UNPUBLISH")): {
DEBUG("post_office: unpublish_actor");
auto i = std::find_if(handler.begin(), handler.end(),
[&](const po_socket_handler_ptr& hp) {
return hp->is_doorman_of(msg.aid);
});
if (i != handler.end()) handler.erase(i);
break;
}
case valof(atom("DONE")): {
done = true;
break;
}
default: {
CPPA_CRITICAL("illegal pipe message");
}
// erase all handlers with closed sockets
for (auto fd : closed_sockets) {
auto i = std::find_if(handler.begin(), handler.end(),
[fd](const po_worker_ptr& wptr) {
return wptr->get_socket() == fd;
});
if (i != handler.end()) {
handler.erase(i);
}
}
// insert new handlers
if (new_handler.empty() == false) {
std::move(new_handler.begin(), new_handler.end(),
std::back_inserter(handler));
......@@ -491,31 +569,4 @@ void post_office_loop(int input_fd) {
while (done == false);
}
/******************************************************************************
* remaining implementation of post_office.hpp *
* (forward each function call to our queue) *
******************************************************************************/
void post_office_add_peer(native_socket_type a0,
const process_information_ptr& a1) {
po_message msg{atom("ADD_PEER"), -1, 0};
send2po(msg, a0, a1);
}
void post_office_publish(native_socket_type server_socket,
const actor_ptr& published_actor) {
po_message msg{atom("PUBLISH"), -1, 0};
send2po(msg, server_socket, published_actor);
}
void post_office_unpublish(actor_id whom) {
po_message msg{atom("UNPUBLISH"), -1, whom};
send2po(msg);
}
void post_office_close_socket(native_socket_type sfd) {
po_message msg{atom("RM_PEER"), sfd, 0};
send2po(msg);
}
} } // namespace cppa::detail
......@@ -52,20 +52,23 @@
#include "cppa/detail/mailman.hpp"
#include "cppa/detail/post_office.hpp"
#include "cppa/detail/native_socket.hpp"
#include "cppa/detail/ipv4_acceptor.hpp"
#include "cppa/detail/ipv4_io_stream.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/network_manager.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
#include "cppa/detail/singleton_manager.hpp"
using std::cout;
using std::endl;
namespace cppa {
/*
namespace {
void read_from_socket(detail::native_socket_type sfd, void* buf, size_t buf_size) {
void read_from_socket(native_socket_type sfd, void* buf, size_t buf_size) {
char* cbuf = reinterpret_cast<char*>(buf);
size_t read_bytes = 0;
size_t left = buf_size;
......@@ -88,11 +91,11 @@ void read_from_socket(detail::native_socket_type sfd, void* buf, size_t buf_size
struct socket_guard {
bool m_released;
detail::native_socket_type m_socket;
native_socket_type m_socket;
public:
socket_guard(detail::native_socket_type sfd) : m_released(false), m_socket(sfd) {
socket_guard(native_socket_type sfd) : m_released(false), m_socket(sfd) {
}
~socket_guard() {
......@@ -104,11 +107,16 @@ struct socket_guard {
}
};
*/
void publish(actor_ptr whom, std::uint16_t port) {
if (!whom) return;
// throws on error
auto ptr = detail::ipv4_acceptor::create(port);
detail::singleton_manager::get_actor_registry()->put(whom->id(), whom);
detail::native_socket_type sockfd;
detail::post_office_publish(std::move(ptr), whom);
/*
native_socket_type sockfd;
struct sockaddr_in serv_addr;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == detail::invalid_socket) {
......@@ -138,10 +146,12 @@ void publish(actor_ptr whom, std::uint16_t port) {
// ok, no exceptions
sguard.release();
detail::post_office_publish(sockfd, whom);
*/
}
actor_ptr remote_actor(const char* host, std::uint16_t port) {
detail::native_socket_type sockfd;
/*
native_socket_type sockfd;
struct sockaddr_in serv_addr;
struct hostent* server;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
......@@ -161,33 +171,32 @@ actor_ptr remote_actor(const char* host, std::uint16_t port) {
if (connect(sockfd, (const sockaddr*) &serv_addr, sizeof(serv_addr)) != 0) {
throw network_error("could not connect to host");
}
*/
auto pinf = process_information::get();
std::uint32_t process_id = pinf->process_id();
int flags = 1;
/*int flags = 1;
setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(int));
::send(sockfd, &process_id, sizeof(std::uint32_t), 0);
::send(sockfd, pinf->node_id().data(), pinf->node_id().size(), 0);
*/
// throws on error
util::io_stream_ptr peer = detail::ipv4_io_stream::connect_to(host, port);
peer->write(&process_id, sizeof(std::uint32_t));
peer->write(pinf->node_id().data(), pinf->node_id().size());
std::uint32_t remote_actor_id;
std::uint32_t peer_pid;
process_information::node_id_type peer_node_id;
read_from_socket(sockfd, &remote_actor_id, sizeof(remote_actor_id));
read_from_socket(sockfd, &peer_pid, sizeof(std::uint32_t));
read_from_socket(sockfd, peer_node_id.data(), peer_node_id.size());
flags = fcntl(sockfd, F_GETFL, 0);
if (flags == -1) {
throw network_error("unable to get socket flags");
}
if (fcntl(sockfd, F_SETFL, flags | O_NONBLOCK) < 0) {
throw network_error("unable to set socket to nonblock");
}
peer->read(&remote_actor_id, sizeof(remote_actor_id));
peer->read(&peer_pid, sizeof(std::uint32_t));
peer->read(peer_node_id.data(), peer_node_id.size());
auto peer_pinf = new process_information(peer_pid, peer_node_id);
process_information_ptr pinfptr(peer_pinf);
//auto key = std::make_tuple(remote_actor_id, pinfptr->process_id(), pinfptr->node_id());
detail::singleton_manager::get_network_manager()
->send_to_mailman(make_any_tuple(sockfd, pinfptr));
detail::post_office_add_peer(sockfd, pinfptr);
util::io_stream_ptr_pair io_ptrs(peer, peer);
//detail::singleton_manager::get_network_manager()
//->send_to_mailman(make_any_tuple(util::io_stream_ptr_pair(peer, peer),
// pinfptr));
detail::post_office_add_peer(io_ptrs, pinfptr);
return detail::get_actor_proxy_cache().get(remote_actor_id,
pinfptr->process_id(),
pinfptr->node_id());
......
//#define CPPA_VERBOSE_CHECK
#define CPPA_VERBOSE_CHECK
#include <stack>
#include <chrono>
......@@ -207,7 +207,7 @@ void testee3(actor_ptr parent) {
void echo_actor() {
receive (
others() >> []() {
self->last_sender() << self->last_dequeued();
reply_tuple(self->last_dequeued());
}
);
}
......@@ -325,7 +325,12 @@ int main() {
CPPA_IF_VERBOSE(cout << "test echo actor ... " << std::flush);
auto mecho = spawn(echo_actor);
send(mecho, "hello echo");
receive(on("hello echo") >> []() { });
receive (
on("hello echo") >> []() { },
others() >> [] {
cout << "UNEXPECTED: " << to_string(self->last_dequeued()) << endl;
}
);
await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl);
......
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