Commit 48b6303a authored by Joseph Noir's avatar Joseph Noir

Add datagram sender & receiver to def. multiplexer

parent efb2a167
...@@ -198,6 +198,41 @@ public: ...@@ -198,6 +198,41 @@ public:
/// Creates and assigns a new `doorman` from given native socked `fd`. /// Creates and assigns a new `doorman` from given native socked `fd`.
expected<accept_handle> add_tcp_doorman(network::native_socket fd); expected<accept_handle> add_tcp_doorman(network::native_socket fd);
/// Adds a `datagram_sink` instance to this broker.
void add_datagram_sink(const intrusive_ptr<datagram_sink>& ptr);
/// Tries to create a datgram sink for `host` on given `port` and creates
/// a new datagram sink describing the endpoint afterwards.
/// @returns The handle of the new `datagram_sink` on success.
expected<datagram_sink_handle> add_datagram_sink(const std::string& host,
uint16_t port);
/// Assigns a detached `datagram_sink` instance identified by `hdl`
/// from the `multiplexer` to this broker.
expected<void> assign_datagram_sink(connection_handle hdl);
/// Creates and assigns a new `datagram_sink` from given native socked `fd`.
expected<datagram_sink_handle> add_datagram_sink(network::native_socket fd);
/// Adds a `datagram_source` instance to this broker.
void add_datagram_source(const intrusive_ptr<datagram_source>& ptr);
/// Tries to open a local port and creates a `datagram_source` managing
/// it on success. If `port == 0`, then the broker will ask
/// the operating system to pick a random port.
/// @returns The handle of the new `datagram_source` and the assigned port.
expected<std::pair<datagram_source_handle, uint16_t>>
add_datagram_source(uint16_t port = 0, const char* in = nullptr,
bool reuse_addr = false);
/// Assigns a detached `datagram_source` instance identified by `hdl`
/// from the `multiplexer` to this broker.
expected<void> assign_datagram_source(accept_handle hdl);
/// Creates and assigns a new `datagram_source` from given native socked `fd`.
expected<datagram_source_handle>
add_datagram_source(network::native_socket fd);
/// Returns the remote address associated to `hdl` /// Returns the remote address associated to `hdl`
/// or empty string if `hdl` is invalid. /// or empty string if `hdl` is invalid.
std::string remote_addr(connection_handle hdl); std::string remote_addr(connection_handle hdl);
......
...@@ -44,8 +44,10 @@ public: ...@@ -44,8 +44,10 @@ public:
~datagram_source(); ~datagram_source();
/*
/// Implicitly starts the read loop on first call. /// Implicitly starts the read loop on first call.
virtual void configure_read(receive_policy::config config) = 0; virtual void configure_read(receive_policy::config config) = 0;
*/
/// Returns the current input buffer. /// Returns the current input buffer.
virtual std::vector<char>& rd_buf() = 0; virtual std::vector<char>& rd_buf() = 0;
......
...@@ -38,6 +38,8 @@ ...@@ -38,6 +38,8 @@
#include "caf/io/network/multiplexer.hpp" #include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/stream_manager.hpp" #include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp" #include "caf/io/network/acceptor_manager.hpp"
#include "caf/io/network/datagram_sink_manager.hpp"
#include "caf/io/network/datagram_source_manager.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
...@@ -405,6 +407,14 @@ inline accept_handle accept_hdl_from_socket(native_socket fd) { ...@@ -405,6 +407,14 @@ inline accept_handle accept_hdl_from_socket(native_socket fd) {
return accept_handle::from_int(int64_from_native_socket(fd)); return accept_handle::from_int(int64_from_native_socket(fd));
} }
inline datagram_sink_handle dg_sink_hdl_from_socket(native_socket fd) {
return datagram_sink_handle::from_int(int64_from_native_socket(fd));
}
inline datagram_source_handle dg_source_hdl_from_socket(native_socket fd) {
return datagram_source_handle::from_int(int64_from_native_socket(fd));
}
/// A stream capable of both reading and writing. The stream's input /// A stream capable of both reading and writing. The stream's input
/// data is forwarded to its {@link stream_manager manager}. /// data is forwarded to its {@link stream_manager manager}.
class stream : public event_handler { class stream : public event_handler {
...@@ -529,6 +539,109 @@ expected<native_socket> new_tcp_connection(const std::string& host, ...@@ -529,6 +539,109 @@ expected<native_socket> new_tcp_connection(const std::string& host,
expected<std::pair<native_socket, uint16_t>> expected<std::pair<native_socket, uint16_t>>
new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr); new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr);
/// A datagram_sender is responsible for sending datagrams to an endpoint.
class datagram_sender: public event_handler {
public:
/// A manger providing the TODO
using manager_type = datagram_sink_manager;
/// A smart pointer to a datagram sink manger
using manager_ptr = intrusive_ptr<datagram_sink_manager>;
/// A buffer class providing a compatible
/// interface to `std::vector`.
using buffer_type = std::vector<char>;
datagram_sender(default_multiplexer& backend_ref, native_socket sockfd);
void ack_writes(bool x);
/// Copies data to the write buffer.
/// @warning Not thread safe.
void write(const void* buf, size_t num_bytes);
/// Returns the write buffer of this datagram sender.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline buffer_type& wr_buf() {
return wr_offline_buf_;
}
/// Sends the content of the write buffer, calling the `io_failure`
/// member function of `mgr` in case of an error.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void flush(const manager_ptr& mgr);
/// Starts forwarding incoming data to `mgr`.
void start(manager_type* mgr);
/// Activates the datagram_sender.
void activate(manager_type* mgr);
/// Closes the network connection and removes this handler from its parent.
void stop_reading();
void removed_from_loop(operation op) override;
void handle_event(operation op) override;
private:
void prepare_next_write();
manager_ptr writer_;
bool ack_writes_;
bool writing_;
size_t written_;
buffer_type wr_buf_;
buffer_type wr_offline_buf_;
};
/// A datagram_receiver is responsible for receiving datagrams on an endpoint.
class datagram_receiver: public event_handler {
public:
/// A manger providing the TODO
using manager_type = datagram_source_manager;
/// A smart pointer to a datagram sink manger
using manager_ptr = intrusive_ptr<datagram_source_manager>;
/// A buffer class providing a compatible
/// interface to `std::vector`.
using buffer_type = std::vector<char>;
datagram_receiver(default_multiplexer& backend_ref, native_socket sockfd);
/// Returns the read buffer of this datagram receiver.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline buffer_type& rd_buf() {
return rd_buf_;
}
/// Starts reading data from the socket.
void start(manager_type* mgr);
/// Activates the datagram_receiver.
void activate(manager_type* mgr);
/// Closes the read channel of the underlying socket and removes
/// this handler from its parent.
void stop_reading();
void removed_from_loop(operation op) override;
void handle_event(operation op) override;
private:
manager_ptr reader_;
size_t read_threshold_;
size_t collected_;
size_t max_;
buffer_type rd_buf_;
};
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
...@@ -158,7 +158,8 @@ expected<void> abstract_broker::assign_tcp_doorman(accept_handle hdl) { ...@@ -158,7 +158,8 @@ expected<void> abstract_broker::assign_tcp_doorman(accept_handle hdl) {
return backend().assign_tcp_doorman(this, hdl); return backend().assign_tcp_doorman(this, hdl);
} }
expected<accept_handle> abstract_broker::add_tcp_doorman(network::native_socket fd) { expected<accept_handle>
abstract_broker::add_tcp_doorman(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd)); CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_tcp_doorman(this, fd); return backend().add_tcp_doorman(this, fd);
} }
......
...@@ -813,8 +813,8 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -813,8 +813,8 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (detached()) if (detached())
// we are already disconnected from the broker while the multiplexer // we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event causes // did not yet remove the socket, this can happen if an I/O event
// the broker to call close_all() while the pollset contained // causes the broker to call close_all() while the pollset contained
// further activities for the broker // further activities for the broker
return false; return false;
auto& dm = acceptor_.backend(); auto& dm = acceptor_.backend();
...@@ -849,7 +849,7 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -849,7 +849,7 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
void remove_from_loop() override { void remove_from_loop() override {
acceptor_.passivate(); acceptor_.passivate();
} }
private: private:
network::acceptor acceptor_; network::acceptor acceptor_;
}; };
auto ptr = make_counted<impl>(self, *this, fd); auto ptr = make_counted<impl>(self, *this, fd);
...@@ -857,6 +857,113 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -857,6 +857,113 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
return ptr->hdl(); return ptr->hdl();
} }
datagram_sink_handle
default_multiplexer::add_datagram_sink(abstract_broker* self,
native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
CAF_ASSERT(fd != network::invalid_native_socket);
class impl : public datagram_sink {
public:
impl(abstract_broker* ptr, default_multiplexer& mx, native_socket sockfd)
: datagram_sink(ptr, network::dg_sink_hdl_from_socket(sockfd)),
sender_(mx, sockfd) {
// nop
}
void stop_reading() override {
CAF_LOG_TRACE("");
sender_.stop_reading();
detach(&sender_.backend(), false);
}
void ack_writes(bool enable) override {
sender_.ack_writes(enable);
}
std::vector<char>& wr_buf() override {
return sender_.wr_buf();
}
std::string addr() const override {
auto x = remote_addr_of_fd(sender_.fd());
if (!x)
return "";
return *x;
}
uint16_t port() const override {
auto x = remote_port_of_fd(sender_.fd());
if (!x)
return 0;
return *x;
}
void launch() {
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
launched_ = true;
sender_.start(this);
}
void add_to_loop() override {
sender_.activate(this);
}
void remove_from_loop() override {
sender_.passivate();
}
private:
bool launched_;
network::datagram_sender sender_;
};
auto ptr = make_counted<impl>(self, *this, fd);
self->add_datagram_sink(ptr);
return ptr->hdl();
}
datagram_source_handle
default_multiplexer::add_datagram_source(abstract_broker* self,
native_socket fd) {
class impl : public datagram_source {
public:
impl(abstract_broker* ptr, default_multiplexer& mx, native_socket sockfd)
: datagram_source(ptr, network::dg_source_hdl_from_socket(sockfd)),
receiver_(mx, sockfd) {
// nop
}
void stop_reading() override {
CAF_LOG_TRACE("");
receiver_.stop_reading();
detach(&receiver_.backend(), false);
}
std::vector<char>& rd_buf() override {
return receiver_.rd_buf();
}
std::string addr() const override {
auto x = remote_addr_of_fd(receiver_.fd());
if (!x)
return "";
return *x;
}
uint16_t port() const override {
auto x = remote_port_of_fd(receiver_.fd());
if (!x)
return 0;
return *x;
}
void launch() {
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
launched_ = true;
receiver_.start(this);
}
void add_to_loop() override {
receiver_.activate(this);
}
void remove_from_loop() override {
receiver_.passivate();
}
private:
bool launched_;
network::datagram_receiver receiver_;
};
auto ptr = make_counted<impl>(self, *this, fd);
self->add_datagram_source(ptr);
return ptr->hdl();
}
expected<connection_handle> expected<connection_handle>
default_multiplexer::new_tcp_scribe(const std::string& host, uint16_t port) { default_multiplexer::new_tcp_scribe(const std::string& host, uint16_t port) {
auto fd = new_tcp_connection(host, port); auto fd = new_tcp_connection(host, port);
...@@ -892,9 +999,9 @@ default_multiplexer::new_tcp_doorman(uint16_t port, const char* in, ...@@ -892,9 +999,9 @@ default_multiplexer::new_tcp_doorman(uint16_t port, const char* in,
res->second); res->second);
} }
expected<void> default_multiplexer::assign_tcp_doorman(abstract_broker* ptr, expected<void> default_multiplexer::assign_tcp_doorman(abstract_broker* self,
accept_handle hdl) { accept_handle hdl) {
add_tcp_doorman(ptr, static_cast<native_socket>(hdl.id())); add_tcp_doorman(self, static_cast<native_socket>(hdl.id()));
return unit; return unit;
} }
...@@ -908,6 +1015,21 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port, ...@@ -908,6 +1015,21 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port,
return std::make_pair(add_tcp_doorman(self, acceptor->first), bound_port); return std::make_pair(add_tcp_doorman(self, acceptor->first), bound_port);
} }
expected<datagram_sink_handle>
default_multiplexer::new_datagram_sink(const std::string& host, uint16_t port) {
auto fd = new_datagram_sink_impl(host, port);
if (!fd)
return std::move(fd.error());
return datagram_sink_handle::from_int(int64_from_native_socket(*fd));
}
expected<void> default_multiplexer::assign_datagram_sink(abstract_broker* self,
datagram_sink_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(self->id()) << CAF_ARG(hdl));
add_datagram_sink(self, static_cast<native_socket>(hdl.id()));
return unit;
}
/****************************************************************************** /******************************************************************************
* platform-independent implementations (finally) * * platform-independent implementations (finally) *
******************************************************************************/ ******************************************************************************/
...@@ -973,6 +1095,19 @@ bool try_accept(native_socket& result, native_socket fd) { ...@@ -973,6 +1095,19 @@ bool try_accept(native_socket& result, native_socket fd) {
return true; return true;
} }
bool send_datagram(size_t& result, native_socket , void* , size_t ) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
// TODO: Send datgrams! (requires acceess to sockaddr)
// ssize_t sendto(int sockfd, const void *buf, size_t len, int flags,
// const struct sockaddr *dest_addr, socklen_t addrlen);
auto sres = 0; //::sendto(fd, reinterpret_cast<socket_send_ptr>(buf),
// len, );
if (is_error(sres, true))
return false;
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return true;
}
event_handler::event_handler(default_multiplexer& dm, native_socket sockfd) event_handler::event_handler(default_multiplexer& dm, native_socket sockfd)
: eventbf_(0), : eventbf_(0),
fd_(sockfd), fd_(sockfd),
...@@ -1268,6 +1403,101 @@ void acceptor::removed_from_loop(operation op) { ...@@ -1268,6 +1403,101 @@ void acceptor::removed_from_loop(operation op) {
mgr_.reset(); mgr_.reset();
} }
datagram_sender::datagram_sender(default_multiplexer& backend_ref,
native_socket sockfd)
: event_handler(backend_ref, sockfd),
ack_writes_(false),
writing_(false),
written_(0) {
// nop
}
void datagram_sender::ack_writes(bool x) {
ack_writes_ = x;
}
void datagram_sender::write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes));
auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes;
wr_offline_buf_.insert(wr_offline_buf_.end(), first, last);
}
void datagram_sender::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (!wr_offline_buf_.empty() && !writing_) {
backend().add(operation::write, fd(), this);
writer_ = mgr;
writing_ = true;
prepare_next_write();
}
}
void datagram_sender::start(manager_type* mgr) {
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void datagram_sender::activate(manager_type* mgr) {
if (!writer_) {
writer_.reset(mgr);
event_handler::activate();
// prepare_next_read();
}
}
void datagram_sender::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void datagram_sender::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG(fd()) << CAF_ARG(op));
if (op == operation::read)
writer_.reset();
}
void datagram_sender::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case operation::read: {
// This should not happen ...
break;
}
case operation::write: {
size_t wb; // written bytes
if (!send_datagram(wb, fd(),
wr_buf_.data() + written_,
wr_buf_.size() - written_)) {
writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this);
} else if (wb > 0) {
written_ += wb;
CAF_ASSERT(written_ <= wr_buf_.size());
auto remaining = wr_buf_.size() - written_;
if (ack_writes_)
writer_->datagram_sent(&backend(), wb);
// prepare next send (or stop sending)
if (remaining == 0)
prepare_next_write();
}
break;
}
case operation::propagate_error:
if (writer_)
writer_->io_failure(&backend(), operation::write);
// backend will delete this handler anyway,
// no need to call backend().del() here
break;
}
}
void datagram_sender::prepare_next_write() {
}
class socket_guard { class socket_guard {
public: public:
explicit socket_guard(native_socket fd) : fd_(fd) { explicit socket_guard(native_socket fd) : fd_(fd) {
......
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