Commit a3dac573 authored by Joseph Noir's avatar Joseph Noir

Merge branch 'unstable' of github.com:Neverlord/libcppa into unstable

parents ed5bfe58 30f0ba09
...@@ -84,7 +84,16 @@ if ("${CMAKE_BUILD_TYPE}" STREQUAL "") ...@@ -84,7 +84,16 @@ if ("${CMAKE_BUILD_TYPE}" STREQUAL "")
set(CMAKE_BUILD_TYPE RelWithDebInfo) set(CMAKE_BUILD_TYPE RelWithDebInfo)
endif ("${CMAKE_BUILD_TYPE}" STREQUAL "") endif ("${CMAKE_BUILD_TYPE}" STREQUAL "")
if (${CMAKE_SYSTEM_NAME} MATCHES "Linux")
set(LIBCPPA_PLATFORM_SRC src/middleman_event_handler_epoll.cpp)
elseif (${CMAKE_SYSTEM_NAME} MATCHES "Darwin")
set(LIBCPPA_PLATFORM_SRC src/middleman_event_handler_poll.cpp)
else ()
message(FATAL_ERROR "This platform is not supported by libcppa")
endif()
set(LIBCPPA_SRC set(LIBCPPA_SRC
${LIBCPPA_PLATFORM_SRC}
src/abstract_tuple.cpp src/abstract_tuple.cpp
src/actor.cpp src/actor.cpp
src/actor_addressing.cpp src/actor_addressing.cpp
...@@ -101,8 +110,7 @@ set(LIBCPPA_SRC ...@@ -101,8 +110,7 @@ set(LIBCPPA_SRC
src/buffer.cpp src/buffer.cpp
src/channel.cpp src/channel.cpp
src/context_switching_actor.cpp src/context_switching_actor.cpp
src/continuable_reader.cpp src/continuable_io.cpp
src/continuable_writer.cpp
src/decorated_tuple.cpp src/decorated_tuple.cpp
src/default_actor_addressing.cpp src/default_actor_addressing.cpp
src/default_actor_proxy.cpp src/default_actor_proxy.cpp
...@@ -133,6 +141,7 @@ set(LIBCPPA_SRC ...@@ -133,6 +141,7 @@ set(LIBCPPA_SRC
src/memory_managed.cpp src/memory_managed.cpp
src/message_header.cpp src/message_header.cpp
src/middleman.cpp src/middleman.cpp
src/middleman_event_handler.cpp
src/object.cpp src/object.cpp
src/object_array.cpp src/object_array.cpp
src/on.cpp src/on.cpp
......
...@@ -92,7 +92,6 @@ cppa/message_id.hpp ...@@ -92,7 +92,6 @@ cppa/message_id.hpp
cppa/message_priority.hpp cppa/message_priority.hpp
cppa/network/acceptor.hpp cppa/network/acceptor.hpp
cppa/network/continuable_io.hpp cppa/network/continuable_io.hpp
cppa/network/continuable_reader.hpp
cppa/network/default_actor_addressing.hpp cppa/network/default_actor_addressing.hpp
cppa/network/default_actor_proxy.hpp cppa/network/default_actor_proxy.hpp
cppa/network/default_message_queue.hpp cppa/network/default_message_queue.hpp
...@@ -104,7 +103,7 @@ cppa/network/io_stream.hpp ...@@ -104,7 +103,7 @@ cppa/network/io_stream.hpp
cppa/network/ipv4_acceptor.hpp cppa/network/ipv4_acceptor.hpp
cppa/network/ipv4_io_stream.hpp cppa/network/ipv4_io_stream.hpp
cppa/network/middleman.hpp cppa/network/middleman.hpp
cppa/network/middleman_event_handler_base.hpp cppa/network/middleman_event_handler.hpp
cppa/network/output_stream.hpp cppa/network/output_stream.hpp
cppa/network/protocol.hpp cppa/network/protocol.hpp
cppa/object.hpp cppa/object.hpp
...@@ -112,7 +111,9 @@ cppa/on.hpp ...@@ -112,7 +111,9 @@ cppa/on.hpp
cppa/opencl.hpp cppa/opencl.hpp
cppa/opencl/actor_facade.hpp cppa/opencl/actor_facade.hpp
cppa/opencl/command.hpp cppa/opencl/command.hpp
cppa/opencl/device_info.hpp
cppa/opencl/global.hpp cppa/opencl/global.hpp
cppa/opencl/opencl_metainfo.hpp
cppa/opencl/program.hpp cppa/opencl/program.hpp
cppa/opencl/smart_ptr.hpp cppa/opencl/smart_ptr.hpp
cppa/opt.hpp cppa/opt.hpp
...@@ -143,6 +144,7 @@ cppa/tpartial_function.hpp ...@@ -143,6 +144,7 @@ cppa/tpartial_function.hpp
cppa/tuple_cast.hpp cppa/tuple_cast.hpp
cppa/uniform_type_info.hpp cppa/uniform_type_info.hpp
cppa/util/abstract_uniform_type_info.hpp cppa/util/abstract_uniform_type_info.hpp
cppa/util/algorithm.hpp
cppa/util/arg_match_t.hpp cppa/util/arg_match_t.hpp
cppa/util/buffer.hpp cppa/util/buffer.hpp
cppa/util/call.hpp cppa/util/call.hpp
...@@ -151,6 +153,8 @@ cppa/util/compare_tuples.hpp ...@@ -151,6 +153,8 @@ cppa/util/compare_tuples.hpp
cppa/util/dptr.hpp cppa/util/dptr.hpp
cppa/util/duration.hpp cppa/util/duration.hpp
cppa/util/fiber.hpp cppa/util/fiber.hpp
cppa/util/get_mac_addresses.hpp
cppa/util/get_root_uuid.hpp
cppa/util/int_list.hpp cppa/util/int_list.hpp
cppa/util/left_or_right.hpp cppa/util/left_or_right.hpp
cppa/util/limited_vector.hpp cppa/util/limited_vector.hpp
...@@ -166,6 +170,7 @@ cppa/util/shared_spinlock.hpp ...@@ -166,6 +170,7 @@ cppa/util/shared_spinlock.hpp
cppa/util/tbind.hpp cppa/util/tbind.hpp
cppa/util/type_list.hpp cppa/util/type_list.hpp
cppa/util/type_pair.hpp cppa/util/type_pair.hpp
cppa/util/type_traits.hpp
cppa/util/upgrade_lock_guard.hpp cppa/util/upgrade_lock_guard.hpp
cppa/util/void_type.hpp cppa/util/void_type.hpp
cppa/util/wrapped.hpp cppa/util/wrapped.hpp
...@@ -185,6 +190,8 @@ examples/qtsupport/qt_group_chat.cpp ...@@ -185,6 +190,8 @@ examples/qtsupport/qt_group_chat.cpp
examples/remote_actors/distributed_calculator.cpp examples/remote_actors/distributed_calculator.cpp
examples/remote_actors/group_chat.cpp examples/remote_actors/group_chat.cpp
examples/remote_actors/group_server.cpp examples/remote_actors/group_server.cpp
examples/remote_actors/protobuf.cpp
examples/remote_actors/protobuf.scala
examples/type_system/announce_1.cpp examples/type_system/announce_1.cpp
examples/type_system/announce_2.cpp examples/type_system/announce_2.cpp
examples/type_system/announce_3.cpp examples/type_system/announce_3.cpp
...@@ -195,6 +202,7 @@ src/actor.cpp ...@@ -195,6 +202,7 @@ src/actor.cpp
src/actor_addressing.cpp src/actor_addressing.cpp
src/actor_proxy.cpp src/actor_proxy.cpp
src/actor_registry.cpp src/actor_registry.cpp
src/algorithm.cpp
src/any_tuple.cpp src/any_tuple.cpp
src/atom.cpp src/atom.cpp
src/attachable.cpp src/attachable.cpp
...@@ -205,8 +213,8 @@ src/binary_serializer.cpp ...@@ -205,8 +213,8 @@ src/binary_serializer.cpp
src/buffer.cpp src/buffer.cpp
src/channel.cpp src/channel.cpp
src/context_switching_actor.cpp src/context_switching_actor.cpp
src/continuable_reader.cpp src/continuable_io.cpp
src/continuable_writer.cpp src/decorated_tuple.cpp
src/default_actor_addressing.cpp src/default_actor_addressing.cpp
src/default_actor_proxy.cpp src/default_actor_proxy.cpp
src/default_peer.cpp src/default_peer.cpp
...@@ -222,6 +230,8 @@ src/exit_reason.cpp ...@@ -222,6 +230,8 @@ src/exit_reason.cpp
src/factory.cpp src/factory.cpp
src/fd_util.cpp src/fd_util.cpp
src/fiber.cpp src/fiber.cpp
src/get_mac_addresses.cpp
src/get_root_uuid.cpp
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
src/ipv4_acceptor.cpp src/ipv4_acceptor.cpp
...@@ -234,11 +244,15 @@ src/memory.cpp ...@@ -234,11 +244,15 @@ src/memory.cpp
src/memory_managed.cpp src/memory_managed.cpp
src/message_header.cpp src/message_header.cpp
src/middleman.cpp src/middleman.cpp
src/middleman_event_handler.cpp
src/middleman_event_handler_epoll.cpp
src/middleman_event_handler_poll.cpp
src/object.cpp src/object.cpp
src/object_array.cpp src/object_array.cpp
src/on.cpp src/on.cpp
src/opencl/actor_facade.cpp src/opencl/actor_facade.cpp
src/opencl/global.cpp src/opencl/global.cpp
src/opencl/opencl_metainfo.cpp
src/opencl/program.cpp src/opencl/program.cpp
src/opt.cpp src/opt.cpp
src/partial_function.cpp src/partial_function.cpp
...@@ -285,14 +299,3 @@ unit_testing/test_sync_send.cpp ...@@ -285,14 +299,3 @@ unit_testing/test_sync_send.cpp
unit_testing/test_tuple.cpp unit_testing/test_tuple.cpp
unit_testing/test_uniform_type.cpp unit_testing/test_uniform_type.cpp
unit_testing/test_yield_interface.cpp unit_testing/test_yield_interface.cpp
cppa/util/algorithm.hpp
src/algorithm.cpp
cppa/util/get_mac_addresses.hpp
src/get_mac_addresses.cpp
cppa/util/get_root_uuid.hpp
src/get_root_uuid.cpp
cppa/util/type_traits.hpp
src/decorated_tuple.cpp
cppa/opencl/opencl_metainfo.hpp
src/opencl/opencl_metainfo.cpp
cppa/opencl/device_info.hpp
...@@ -34,10 +34,19 @@ ...@@ -34,10 +34,19 @@
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include "cppa/ref_counted.hpp" #include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp" #include "cppa/intrusive_ptr.hpp"
#include "cppa/network/continuable_reader.hpp"
namespace cppa { namespace network { namespace cppa { namespace network {
/**
* @brief Denotes the return value of
* {@link continuable_io::continue_reading()}.
*/
enum continue_reading_result {
read_failure,
read_closed,
read_continue_later
};
/** /**
* @brief Denotes the return value of * @brief Denotes the return value of
* {@link continuable_io::continue_writing()}. * {@link continuable_io::continue_writing()}.
...@@ -52,32 +61,46 @@ enum continue_writing_result { ...@@ -52,32 +61,46 @@ enum continue_writing_result {
/** /**
* @brief An object performing asynchronous input and output. * @brief An object performing asynchronous input and output.
*/ */
class continuable_io : public continuable_reader { class continuable_io : public ref_counted {
typedef continuable_reader super;
public: public:
/**
* @brief Returns the file descriptor for incoming data.
*/
inline native_socket_type read_handle() const { return m_rd; }
/** /**
* @brief Returns the file descriptor for outgoing data. * @brief Returns the file descriptor for outgoing data.
*/ */
native_socket_type write_handle() const { inline native_socket_type write_handle() const {
return m_wr; return m_wr;
} }
continuable_io* as_io(); /**
* @brief Reads from {@link read_handle()} if valid.
*/
virtual continue_reading_result continue_reading();
/**
* @brief Writes to {@link write_handle()} if valid.
*/
virtual continue_writing_result continue_writing();
/** /**
* @brief Writes to {@link write_handle()}. * @brief Called from middleman before it removes this object
* due to an IO failure.
*/ */
virtual continue_writing_result continue_writing() = 0; virtual void io_failed() = 0;
protected: protected:
continuable_io(native_socket_type read_fd, native_socket_type write_fd); continuable_io(native_socket_type read_fd,
native_socket_type write_fd = invalid_socket);
private: private:
native_socket_type m_rd;
native_socket_type m_wr; native_socket_type m_wr;
}; };
......
...@@ -43,7 +43,6 @@ ...@@ -43,7 +43,6 @@
#include "cppa/network/input_stream.hpp" #include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp" #include "cppa/network/output_stream.hpp"
#include "cppa/network/continuable_reader.hpp"
#include "cppa/network/continuable_io.hpp" #include "cppa/network/continuable_io.hpp"
#include "cppa/network/default_message_queue.hpp" #include "cppa/network/default_message_queue.hpp"
...@@ -68,8 +67,6 @@ class default_peer : public continuable_io { ...@@ -68,8 +67,6 @@ class default_peer : public continuable_io {
continue_writing_result continue_writing() override; continue_writing_result continue_writing() override;
continuable_io* as_io() override;
void io_failed(); void io_failed();
void enqueue(const message_header& hdr, const any_tuple& msg); void enqueue(const message_header& hdr, const any_tuple& msg);
......
...@@ -34,15 +34,15 @@ ...@@ -34,15 +34,15 @@
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
#include "cppa/network/ipv4_acceptor.hpp" #include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/continuable_reader.hpp" #include "cppa/network/continuable_io.hpp"
namespace cppa { namespace network { namespace cppa { namespace network {
class default_protocol; class default_protocol;
class default_peer_acceptor : public continuable_reader { class default_peer_acceptor : public continuable_io {
typedef continuable_reader super; typedef continuable_io super;
public: public:
......
...@@ -36,7 +36,7 @@ ...@@ -36,7 +36,7 @@
#include <memory> #include <memory>
#include <functional> #include <functional>
#include "cppa/network/continuable_reader.hpp" #include "cppa/network/continuable_io.hpp"
namespace cppa { namespace detail { class singleton_manager; } } namespace cppa { namespace detail { class singleton_manager; } }
...@@ -92,11 +92,11 @@ class abstract_middleman : public middleman { ...@@ -92,11 +92,11 @@ class abstract_middleman : public middleman {
inline abstract_middleman() : m_done(false) { } inline abstract_middleman() : m_done(false) { }
void stop_writer(const continuable_reader_ptr& ptr); void stop_writer(const continuable_io_ptr& ptr);
void continue_writer(const continuable_reader_ptr& ptr); void continue_writer(const continuable_io_ptr& ptr);
void stop_reader(const continuable_reader_ptr& what); void stop_reader(const continuable_io_ptr& what);
void continue_reader(const continuable_reader_ptr& what); void continue_reader(const continuable_io_ptr& what);
protected: protected:
...@@ -104,7 +104,7 @@ class abstract_middleman : public middleman { ...@@ -104,7 +104,7 @@ class abstract_middleman : public middleman {
inline bool done() const { return m_done; } inline bool done() const { return m_done; }
bool m_done; bool m_done;
std::vector<continuable_reader_ptr> m_readers; std::vector<continuable_io_ptr> m_readers;
middleman_event_handler& handler(); middleman_event_handler& handler();
......
...@@ -28,71 +28,128 @@ ...@@ -28,71 +28,128 @@
\******************************************************************************/ \******************************************************************************/
#ifndef CONTINUABLE_READER_HPP #ifndef MIDDLEMAN_EVENT_HANDLER_HPP
#define CONTINUABLE_READER_HPP #define MIDDLEMAN_EVENT_HANDLER_HPP
#include <vector>
#include <utility>
#include "cppa/atom.hpp"
#include "cppa/actor.hpp"
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include "cppa/ref_counted.hpp" #include "cppa/logging.hpp"
#include "cppa/network/continuable_io.hpp"
namespace cppa { namespace network { namespace cppa { namespace network {
class middleman; typedef int event_bitmask;
namespace event { namespace {
constexpr event_bitmask none = 0x00;
constexpr event_bitmask read = 0x01;
constexpr event_bitmask write = 0x02;
constexpr event_bitmask both = 0x03;
constexpr event_bitmask error = 0x04;
} } // namespace event
/** /**
* @brief Denotes the return value of * @brief Converts an event bitmask to a human-readable string.
* {@link continuable_reader::continue_reading()}.
*/ */
enum continue_reading_result { inline const char* eb2str(event_bitmask e) {
read_failure, switch (e) {
read_closed, default: return "INVALID";
read_continue_later case event::none: return "event::none";
case event::read: return "event::read";
case event::write: return "event::write";
case event::both: return "event::both";
case event::error: return "event::error";
}
}
struct fd_meta_info {
native_socket_type fd;
continuable_io_ptr ptr;
event_bitmask mask;
fd_meta_info(native_socket_type a0,
const continuable_io_ptr& a1,
event_bitmask a2)
: fd(a0), ptr(a1), mask(a2) { }
};
struct fd_meta_info_less {
inline bool operator()(const fd_meta_info& lhs, native_socket_type rhs) const {
return lhs.fd < rhs;
}
}; };
class continuable_io; enum class fd_meta_event { add, erase, mod };
/** class middleman_event_handler {
* @brief An object performing asynchronous input.
*/
class continuable_reader : public ref_counted {
public: public:
/** typedef std::pair<event_bitmask, continuable_io*> io_event;
* @brief Returns the file descriptor for incoming data.
*/ virtual ~middleman_event_handler();
inline native_socket_type read_handle() const { return m_rd; }
void alteration(const continuable_io_ptr& ptr, event_bitmask e, fd_meta_event etype);
void add(const continuable_io_ptr& ptr, event_bitmask e);
/** void erase(const continuable_io_ptr& ptr, event_bitmask e);
* @brief Reads from {@link read_handle()}.
*/
virtual continue_reading_result continue_reading() = 0;
/** event_bitmask next_bitmask(event_bitmask old, event_bitmask arg, fd_meta_event op) const;
* @brief Casts @p this to a continuable_io or returns @p nullptr
* if cast fails.
*/
virtual continuable_io* as_io();
/** template<typename F>
* @brief Called from middleman before it removes this object void poll(const F& fun) {
* due to an IO failure. poll_impl();
*/ for (auto& p : m_events) fun(p.first, p.second);
virtual void io_failed() = 0; m_events.clear();
update();
}
// pure virtual member function
virtual void init() = 0;
virtual size_t num_sockets() const = 0;
// fills the event vector
virtual void poll_impl() = 0;
virtual void handle_event(fd_meta_event me,
native_socket_type fd,
event_bitmask old_bitmask,
event_bitmask new_bitmask,
continuable_io* ptr) = 0;
// implemented in platform-dependent .cpp file
static std::unique_ptr<middleman_event_handler> create();
void update();
protected: protected:
continuable_reader(native_socket_type read_fd); fd_meta_info_less m_less;
std::vector<fd_meta_info> m_meta; // this vector is *always* sorted
std::vector<std::pair<fd_meta_info, fd_meta_event> > m_alterations;
std::vector<io_event> m_events;
middleman_event_handler();
private: private:
native_socket_type m_rd; std::vector<fd_meta_info>::iterator find_meta(native_socket_type fd) {
auto last = end(m_meta);
auto iter = lower_bound(begin(m_meta), last, fd, m_less);
return (iter != last && iter->fd == fd) ? iter : last;
}
}; };
typedef intrusive_ptr<continuable_reader> continuable_reader_ptr; } } // namespace cppa::detail
} } // namespace cppa::network
#endif // CONTINUABLE_READER_HPP #endif // MIDDLEMAN_EVENT_HANDLER_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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 MIDDLEMAN_EVENT_HANDLER_BASE_HPP
#define MIDDLEMAN_EVENT_HANDLER_BASE_HPP
#include <vector>
#include <utility>
#include "cppa/config.hpp"
#include "cppa/logging.hpp"
#include "cppa/network/continuable_reader.hpp"
#include "cppa/network/continuable_io.hpp"
namespace cppa { namespace network {
typedef int event_bitmask;
namespace event { namespace {
constexpr event_bitmask none = 0x00;
constexpr event_bitmask read = 0x01;
constexpr event_bitmask write = 0x02;
constexpr event_bitmask both = 0x03;
constexpr event_bitmask error = 0x04;
} } // namespace event
/**
* @brief Converts an event bitmask to a human-readable string.
*/
inline const char* eb2str(event_bitmask e) {
switch (e) {
default: return "INVALID";
case event::none: return "event::none";
case event::read: return "event::read";
case event::write: return "event::write";
case event::both: return "event::both";
case event::error: return "event::error";
}
}
struct fd_meta_info {
native_socket_type fd;
continuable_reader_ptr ptr;
event_bitmask mask;
fd_meta_info(native_socket_type a0,
const continuable_reader_ptr& a1,
event_bitmask a2)
: fd(a0), ptr(a1), mask(a2) { }
};
struct fd_meta_info_less {
inline bool operator()(const fd_meta_info& lhs, native_socket_type rhs) const {
return lhs.fd < rhs;
}
};
enum class fd_meta_event { add, erase, mod };
template<typename Derived>
class middleman_event_handler_base {
public:
typedef std::vector<fd_meta_info> vector_type;
virtual ~middleman_event_handler_base() { }
void alteration(const continuable_reader_ptr& ptr,
event_bitmask e,
fd_meta_event etype) {
native_socket_type fd;
switch (e) {
case event::read:
fd = ptr->read_handle();
break;
case event::write: {
auto wptr = ptr->as_io();
if (wptr) fd = wptr->write_handle();
else {
CPPA_LOGMF(CPPA_ERROR, self, "ptr->downcast() returned nullptr");
return;
}
break;
}
case event::both: {
fd = ptr->read_handle();
auto wptr = ptr->as_io();
if (wptr) {
auto wrfd = wptr->write_handle();
if (fd != wrfd) {
CPPA_LOGMF(CPPA_DEBUG, self, "read_handle != write_handle, split "
"into two function calls");
// split into two function calls
e = event::read;
alteration(ptr, event::write, etype);
}
}
else {
CPPA_LOGMF(CPPA_ERROR, self, "ptr->downcast() returned nullptr");
return;
}
break;
}
default:
CPPA_LOGMF(CPPA_ERROR, self, "invalid bitmask");
return;
}
m_alterations.emplace_back(fd_meta_info(fd, ptr, e), etype);
}
void add(const continuable_reader_ptr& ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::add);
}
void erase(const continuable_reader_ptr& ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::erase);
}
inline event_bitmask next_bitmask(event_bitmask old, event_bitmask arg, fd_meta_event op) {
CPPA_REQUIRE(op == fd_meta_event::add || op == fd_meta_event::erase);
return (op == fd_meta_event::add) ? old | arg : old & ~arg;
}
void update() {
CPPA_LOG_TRACE("");
for (auto& elem_pair : m_alterations) {
auto& elem = elem_pair.first;
auto old = event::none;
auto last = end(m_meta);
auto iter = lower_bound(begin(m_meta), last, elem.fd, m_less);
if (iter != last) old = iter->mask;
auto mask = next_bitmask(old, elem.mask, elem_pair.second);
auto ptr = elem.ptr.get();
CPPA_LOGMF(CPPA_DEBUG, self, "new bitmask for "
<< elem.ptr.get() << ": " << eb2str(mask));
if (iter == last || iter->fd != elem.fd) {
CPPA_LOG_INFO_IF(mask == event::none,
"cannot erase " << ptr
<< " (not found in m_meta)");
if (mask != event::none) {
m_meta.insert(iter, elem);
d()->handle_event(fd_meta_event::add, elem.fd,
event::none, mask, ptr);
}
}
else if (iter->fd == elem.fd) {
CPPA_REQUIRE(iter->ptr == elem.ptr);
if (mask == event::none) {
m_meta.erase(iter);
d()->handle_event(fd_meta_event::erase, elem.fd, old, mask, ptr);
}
else {
iter->mask = mask;
d()->handle_event(fd_meta_event::mod, elem.fd, old, mask, ptr);
}
}
}
m_alterations.clear();
}
protected:
fd_meta_info_less m_less;
vector_type m_meta; // this vector is *always* sorted
std::vector<std::pair<fd_meta_info, fd_meta_event> > m_alterations;
private:
vector_type::iterator find_meta(native_socket_type fd) {
auto last = end(m_meta);
auto iter = lower_bound(begin(m_meta), last, fd, m_less);
return (iter != last && iter->fd == fd) ? iter : last;
}
Derived* d() { return static_cast<Derived*>(this); }
};
template<class BaseIter, class BasIterAccess>
class event_iterator_impl {
public:
event_iterator_impl(const BaseIter& iter) : m_i(iter) { }
inline event_iterator_impl& operator++() {
m_access.advance(m_i);
return *this;
}
inline event_iterator_impl* operator->() { return this; }
inline const event_iterator_impl* operator->() const { return this; }
inline event_bitmask type() const {
return m_access.type(m_i);
}
inline void io_failed() {
ptr()->io_failed();
}
inline continue_reading_result continue_reading() {
return ptr()->continue_reading();
}
inline continue_writing_result continue_writing() {
return ptr()->as_io()->continue_writing();
}
inline bool equal_to(const event_iterator_impl& other) const {
return m_access.equal(m_i, other.m_i);
}
inline void handled() { m_access.handled(m_i); }
inline continuable_reader* ptr() { return m_access.ptr(m_i); }
private:
BaseIter m_i;
BasIterAccess m_access;
};
template<class Iter, class Access>
inline bool operator==(const event_iterator_impl<Iter, Access>& lhs,
const event_iterator_impl<Iter, Access>& rhs) {
return lhs.equal_to(rhs);
}
template<class Iter, class Access>
inline bool operator!=(const event_iterator_impl<Iter, Access>& lhs,
const event_iterator_impl<Iter, Access>& rhs) {
return !lhs.equal_to(rhs);
}
} } // namespace cppa::detail
#endif // MIDDLEMAN_EVENT_HANDLER_BASE_HPP
...@@ -47,9 +47,8 @@ namespace cppa { class actor_addressing; } ...@@ -47,9 +47,8 @@ namespace cppa { class actor_addressing; }
namespace cppa { namespace network { namespace cppa { namespace network {
class abstract_middleman;
class continuable_reader;
class continuable_io; class continuable_io;
class abstract_middleman;
/** /**
* @brief Implements a communication protocol. * @brief Implements a communication protocol.
...@@ -86,16 +85,16 @@ class protocol : public ref_counted { ...@@ -86,16 +85,16 @@ class protocol : public ref_counted {
protected: protected:
// note: not thread-safe; call only in run_later functor! // note: not thread-safe; call only in run_later functor!
void continue_reader(continuable_reader* what); void continue_reader(continuable_io* what);
// note: not thread-safe; call only in run_later functor! // note: not thread-safe; call only in run_later functor!
void continue_writer(continuable_reader* what); void continue_writer(continuable_io* what);
// note: not thread-safe; call only in run_later functor! // note: not thread-safe; call only in run_later functor!
void stop_reader(continuable_reader* what); void stop_reader(continuable_io* what);
// note: not thread-safe; call only in run_later functor! // note: not thread-safe; call only in run_later functor!
void stop_writer(continuable_reader* what); void stop_writer(continuable_io* what);
inline abstract_middleman* parent() { return m_parent.get(); } inline abstract_middleman* parent() { return m_parent.get(); }
......
...@@ -33,8 +33,14 @@ ...@@ -33,8 +33,14 @@
namespace cppa { namespace network { namespace cppa { namespace network {
continuable_io::continuable_io(native_socket_type rd, native_socket_type wr) continuable_io::continuable_io(native_socket_type rd, native_socket_type wr)
: super(rd), m_wr(wr) { } : m_rd(rd), m_wr(wr) { }
continuable_io* continuable_io::as_io() { return this; } continue_reading_result continuable_io::continue_reading() {
return read_closed;
}
continue_writing_result continuable_io::continue_writing() {
return write_closed;
}
} } // namespace cppa::network } } // namespace cppa::network
...@@ -343,10 +343,6 @@ continue_writing_result default_peer::continue_writing() { ...@@ -343,10 +343,6 @@ continue_writing_result default_peer::continue_writing() {
return write_done; return write_done;
} }
continuable_io* default_peer::as_io() {
return this;
}
void default_peer::enqueue(const message_header& hdr, const any_tuple& msg) { void default_peer::enqueue(const message_header& hdr, const any_tuple& msg) {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
binary_serializer bs(&m_wr_buf, m_parent->addressing()); binary_serializer bs(&m_wr_buf, m_parent->addressing());
......
This diff is collapsed.
...@@ -28,14 +28,95 @@ ...@@ -28,14 +28,95 @@
\******************************************************************************/ \******************************************************************************/
#include "cppa/network/continuable_reader.hpp" #include "cppa/network/middleman_event_handler.hpp"
#include "cppa/network/continuable_io.hpp"
namespace cppa { namespace network { namespace cppa { namespace network {
continuable_reader::continuable_reader(native_socket_type rd) : m_rd(rd) { } middleman_event_handler::middleman_event_handler() { }
continuable_io* continuable_reader::as_io() { return nullptr; } middleman_event_handler::~middleman_event_handler() { }
void middleman_event_handler::alteration(const continuable_io_ptr& ptr,
event_bitmask e,
fd_meta_event etype) {
native_socket_type fd;
switch (e) {
case event::read:
fd = ptr->read_handle();
break;
case event::write: {
fd = ptr->write_handle();
break;
}
case event::both: {
fd = ptr->read_handle();
auto wrfd = ptr->write_handle();
if (fd != wrfd) {
CPPA_LOGMF(CPPA_DEBUG, self, "read_handle != write_handle, split "
"into two function calls");
// split into two function calls
e = event::read;
alteration(ptr, event::write, etype);
}
break;
}
default:
CPPA_LOGMF(CPPA_ERROR, self, "invalid bitmask");
return;
}
m_alterations.emplace_back(fd_meta_info(fd, ptr, e), etype);
}
void middleman_event_handler::add(const continuable_io_ptr& ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::add);
}
void middleman_event_handler::erase(const continuable_io_ptr& ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::erase);
}
event_bitmask middleman_event_handler::next_bitmask(event_bitmask old, event_bitmask arg, fd_meta_event op) const {
CPPA_REQUIRE(op == fd_meta_event::add || op == fd_meta_event::erase);
return (op == fd_meta_event::add) ? old | arg : old & ~arg;
}
void middleman_event_handler::update() {
CPPA_LOG_TRACE("");
for (auto& elem_pair : m_alterations) {
auto& elem = elem_pair.first;
auto old = event::none;
auto last = end(m_meta);
auto iter = lower_bound(begin(m_meta), last, elem.fd, m_less);
if (iter != last) old = iter->mask;
auto mask = next_bitmask(old, elem.mask, elem_pair.second);
auto ptr = elem.ptr.get();
CPPA_LOGMF(CPPA_DEBUG, self, "new bitmask for "
<< elem.ptr.get() << ": " << eb2str(mask));
if (iter == last || iter->fd != elem.fd) {
CPPA_LOG_INFO_IF(mask == event::none,
"cannot erase " << ptr
<< " (not found in m_meta)");
if (mask != event::none) {
m_meta.insert(iter, elem);
handle_event(fd_meta_event::add, elem.fd,
event::none, mask, ptr);
}
}
else if (iter->fd == elem.fd) {
CPPA_REQUIRE(iter->ptr == elem.ptr);
if (mask == event::none) {
m_meta.erase(iter);
handle_event(fd_meta_event::erase, elem.fd, old, mask, ptr);
}
else {
iter->mask = mask;
handle_event(fd_meta_event::mod, elem.fd, old, mask, ptr);
}
}
}
m_alterations.clear();
}
} } // namespace cppa::network } } // namespace cppa::network
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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 <string>
#include <vector>
#include <string.h>
#include <sys/epoll.h>
#include "cppa/network/middleman_event_handler.hpp"
namespace cppa { namespace network {
namespace {
class middleman_event_handler_impl : public middleman_event_handler {
public:
middleman_event_handler_impl() : m_epollfd(-1) { }
~middleman_event_handler_impl() { if (m_epollfd != -1) close(m_epollfd); }
void init() {
m_epollfd = epoll_create1(EPOLL_CLOEXEC);
if (m_epollfd == -1) {
throw std::ios_base::failure( std::string("epoll_create1: ")
+ strerror(errno));
}
// handle at most 64 events at a time
m_epoll_events.resize(64);
}
size_t num_sockets() const { return m_meta.size(); }
void poll_impl() {
CPPA_REQUIRE(m_meta.empty() == false);
int presult;
do {
CPPA_LOGMF(CPPA_DEBUG, self, "epoll_wait on " << num_sockets() << " sockets");
presult = epoll_wait(m_epollfd, m_epoll_events.data(), (int) m_epoll_events.size(), -1);
CPPA_LOGMF(CPPA_DEBUG, self, "epoll_wait returned " << presult);
if (presult < 0) {
// try again unless critical error occured
presult = 0;
switch (errno) {
// a signal was caught
case EINTR: {
// just try again
break;
}
default: {
perror("epoll() failed");
CPPA_CRITICAL("epoll() failed");
}
}
}
else {
for (size_t i = 0; i < static_cast<size_t>(presult); ++i) {
auto events = m_epoll_events[i].events;
auto mask = event::none;
if (events & (EPOLLRDHUP | EPOLLERR | EPOLLHUP)) mask = event::error;
else {
if (events & EPOLLIN) mask |= event::read;
if (events & EPOLLOUT) mask |= event::write;
}
CPPA_REQUIRE(mask != 0);
m_events.emplace_back(mask, reinterpret_cast<continuable_io*>(m_epoll_events[i].data.ptr));
}
}
} while (presult < 0);
}
void handle_event(fd_meta_event me,
native_socket_type fd,
event_bitmask old_bitmask,
event_bitmask new_bitmask,
continuable_io* ptr) {
static_cast<void>(old_bitmask); // no need for it
int operation;
epoll_event ee;
ee.data.ptr = ptr;
switch (new_bitmask) {
case event::none:
CPPA_REQUIRE(me == fd_meta_event::erase);
ee.events = 0;
break;
case event::read:
ee.events = EPOLLIN | EPOLLRDHUP;
break;
case event::write:
ee.events = EPOLLOUT;
case event::both:
ee.events = EPOLLIN | EPOLLRDHUP | EPOLLOUT;
break;
default: CPPA_CRITICAL("invalid event bitmask");
}
switch (me) {
case fd_meta_event::add:
operation = EPOLL_CTL_ADD;
break;
case fd_meta_event::erase:
operation = EPOLL_CTL_DEL;
break;
case fd_meta_event::mod:
operation = EPOLL_CTL_MOD;
break;
default: CPPA_CRITICAL("invalid fd_meta_event");
}
if (epoll_ctl(m_epollfd, operation, fd, &ee) < 0) {
switch (errno) {
// supplied file descriptor is already registered
case EEXIST: {
CPPA_LOGMF(CPPA_ERROR, self, "file descriptor registered twice");
break;
}
// op was EPOLL_CTL_MOD or EPOLL_CTL_DEL,
// and fd is not registered with this epoll instance.
case ENOENT: {
CPPA_LOGMF(CPPA_ERROR, self, "cannot delete file descriptor "
"because it isn't registered");
break;
}
default: {
CPPA_LOGMF(CPPA_ERROR, self, strerror(errno));
perror("epoll_ctl() failed");
CPPA_CRITICAL("epoll_ctl() failed");
}
}
}
}
private:
int m_epollfd;
std::vector<epoll_event> m_epoll_events;
};
} // namespace <anonymous>
std::unique_ptr<middleman_event_handler> middleman_event_handler::create() {
return std::unique_ptr<middleman_event_handler>{new middleman_event_handler_impl};
}
} } // namespace cppa::network
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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 <poll.h>
#include "cppa/network/middleman_event_handler.hpp"
#ifndef POLLRDHUP
#define POLLRDHUP POLLHUP
#endif
namespace cppa { namespace network {
namespace {
bool pollfd_less(const pollfd& lhs, native_socket_type rhs) {
return lhs.fd < rhs;
}
short to_poll_bitmask(event_bitmask mask) {
switch (mask) {
case event::read: return POLLIN;
case event::write: return POLLOUT;
case event::both: return (POLLIN|POLLOUT);
default: CPPA_CRITICAL("invalid event bitmask");
}
}
class middleman_event_handler_impl : public middleman_event_handler {
public:
void init() { }
size_t num_sockets() const { return m_pollset.size(); }
void poll_impl() {
CPPA_REQUIRE(m_pollset.empty() == false);
CPPA_REQUIRE(m_pollset.size() == m_meta.size());
int presult;
do {
presult = ::poll(m_pollset.data(), m_pollset.size(), -1);
CPPA_LOGMF(CPPA_DEBUG, self, "poll() on " << num_sockets()
<< " sockets returned " << presult);
if (presult < 0) {
switch (errno) {
// a signal was caught
case EINTR: {
// just try again
break;
}
case ENOMEM: {
CPPA_LOGMF(CPPA_ERROR, self, "poll() failed for reason ENOMEM");
// there's not much we can do other than try again
// in hope someone releases memory
//this_thread::yield();
break;
}
default: {
perror("poll() failed");
CPPA_CRITICAL("poll() failed");
}
}
}
else {
for (size_t i = 0; i < m_pollset.size(); ++i) {
event_bitmask eb = event::none;
auto& revents = m_pollset[i].revents;
if (revents & (POLLRDHUP | POLLERR | POLLHUP | POLLNVAL)) {
eb = event::error;
}
else {
if (revents & (POLLIN | POLLPRI)) eb |= event::read;
if (revents & POLLOUT) eb |= event::write;
}
revents = 0;
m_events.emplace_back(eb, m_meta[i].ptr.get());
}
}
} while (presult < 0);
}
void handle_event(fd_meta_event me,
native_socket_type fd,
event_bitmask old_bitmask,
event_bitmask new_bitmask,
continuable_io*) {
static_cast<void>(old_bitmask); // no need for it
switch (me) {
case fd_meta_event::add: {
pollfd tmp;
tmp.fd = fd;
tmp.events = to_poll_bitmask(new_bitmask);
tmp.revents = 0;
m_pollset.insert(lb(fd), tmp);
CPPA_LOGMF(CPPA_DEBUG, self, "inserted new element");
break;
}
case fd_meta_event::erase: {
auto last = end(m_pollset);
auto iter = lb(fd);
CPPA_LOG_ERROR_IF(iter == last || iter->fd != fd,
"m_meta and m_pollset out of sync; "
"no element found for fd (cannot erase)");
if (iter != last && iter->fd == fd) {
CPPA_LOGMF(CPPA_DEBUG, self, "erased element");
m_pollset.erase(iter);
}
break;
}
case fd_meta_event::mod: {
auto last = end(m_pollset);
auto iter = lb(fd);
CPPA_LOG_ERROR_IF(iter == last || iter->fd != fd,
"m_meta and m_pollset out of sync; "
"no element found for fd (cannot erase)");
if (iter != last && iter->fd == fd) {
CPPA_LOGMF(CPPA_DEBUG, self, "updated bitmask");
iter->events = to_poll_bitmask(new_bitmask);
}
break;
}
}
}
private:
std::vector<pollfd>::iterator lb(native_socket_type fd) {
return lower_bound(begin(m_pollset), end(m_pollset), fd, pollfd_less);
}
std::vector<pollfd> m_pollset; // always in sync with m_meta
};
} // namespace <anonymous>
std::unique_ptr<middleman_event_handler> middleman_event_handler::create() {
return std::unique_ptr<middleman_event_handler>{new middleman_event_handler_impl};
}
} } // namespace cppa::network
...@@ -43,23 +43,22 @@ void protocol::run_later(std::function<void()> fun) { ...@@ -43,23 +43,22 @@ void protocol::run_later(std::function<void()> fun) {
m_parent->run_later(std::move(fun)); m_parent->run_later(std::move(fun));
} }
void protocol::continue_reader(continuable_reader* ptr) { void protocol::continue_reader(continuable_io* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr)); CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_parent->continue_reader(ptr); m_parent->continue_reader(ptr);
} }
void protocol::continue_writer(continuable_reader* ptr) { void protocol::continue_writer(continuable_io* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr)); CPPA_LOG_TRACE(CPPA_ARG(ptr));
CPPA_REQUIRE(ptr->as_io() != nullptr);
m_parent->continue_writer(ptr); m_parent->continue_writer(ptr);
} }
void protocol::stop_reader(continuable_reader* ptr) { void protocol::stop_reader(continuable_io* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr)); CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_parent->stop_reader(ptr); m_parent->stop_reader(ptr);
} }
void protocol::stop_writer(continuable_reader* ptr) { void protocol::stop_writer(continuable_io* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr)); CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_parent->stop_writer(ptr); m_parent->stop_writer(ptr);
} }
......
...@@ -22,6 +22,86 @@ bool is_even(int i) { return i % 2 == 0; } ...@@ -22,6 +22,86 @@ bool is_even(int i) { return i % 2 == 0; }
* *
*/ */
struct dummy {
inline operator actor_ptr() const { return nullptr; }
inline actor_ptr as_facade() const { return nullptr; }
};
inline dummy operator|(const dummy&, const dummy&) { return {}; }
template<typename T>
dummy operator|(const dummy& lhs, const T&) {
return lhs;
}
template<typename T>
dummy operator|(const T&, const dummy& rhs) {
return rhs;
}
dummy slice(int,int) { return {}; }
dummy prepend(atom_value) { return {}; }
dummy consume(atom_value) { return {}; }
dummy split(int) { return {}; }
dummy join(int) { return {}; }
void example() {
// {}: tuple
// []: list
// 'a': atom 'a' == atom("a")
// x => y: replies to x with y
{
// for starters: get the internally used user-ID from a key-value store
// and get the user role, e.g., "admin", from the user database
// {'get', key} => {user_id, [tags]}
actor_ptr kvstore;
// {'info', user_id} => {name, address, phone, role}
actor_ptr userinfo;
// {username} => {role}
auto query = kvstore | slice(0,0) | prepend(atom("info")) | userinfo | slice(3,3);
// check whether Dirty Harry has admin access
sync_send(query, atom("get"), "Dirty Harry").then(
on("admin") >> [] {
// grant access
},
others() >> [] {
// deny access (well, no one denies Dirty Harry access anyways)
}
);
// creates a new actor providing the interface {username} => {role}
actor_ptr roles = query.as_facade();
}
{
// we have an actor representing a database for orders and we want
// to have the net price of all orders from Dirty Harry
// {'orders', user_id} => {[order_id]}
// {'price', order_id} => {net_price}
actor_ptr orders;
// {'prices', user_id} => {[net_price]}
auto query = consume(atom("prices"))
| prepend(atom("orders"))
| orders
| prepend(atom("price"))
| split(1)
| orders
| join(0);
sync_send(query, atom("prices"), "Dirty Harry").then(
[](const vector<float>& prices) {
cout << "Dirty Harry has ordered products "
<< "with a total net price of $"
<< accumulate(prices.begin(), prices.end(), 0)
<< endl;
}
);
}
}
int main() { int main() {
CPPA_TEST(test_match); CPPA_TEST(test_match);
......
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