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

Move tcp/udp policy related code to their files

parent 48f7baff
...@@ -261,38 +261,6 @@ private: ...@@ -261,38 +261,6 @@ private:
size_t max_throughput_; size_t max_throughput_;
}; };
/// Reads up to `len` bytes from `fd,` writing the received data
/// to `buf`. Returns `true` as long as `fd` is readable and `false`
/// if the socket has been closed or an IO error occured. The number
/// of read bytes is stored in `result` (can be 0).
rw_state read_some(size_t& result, native_socket fd, void* buf, size_t len);
/// Writes up to `len` bytes from `buf` to `fd`.
/// Returns `true` as long as `fd` is readable and `false`
/// if the socket has been closed or an IO error occured. The number
/// of written bytes is stored in `result` (can be 0).
rw_state write_some(size_t& result, native_socket fd, const void* buf,
size_t len);
/// Tries to accept a new connection from `fd`. On success,
/// the new connection is stored in `result`. Returns true
/// as long as
bool try_accept(native_socket& result, native_socket fd);
/// Write a datagram containing `buf_len` bytes to `fd` addressed
/// at the endpoint in `sa` with size `sa_len`. Returns true as long
/// as no IO error occurs. The number of written bytes is stored in
/// `result` and the sender is stored in `ep`.
bool read_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
ip_endpoint& ep);
/// Reveice a datagram of up to `len` bytes. Larger datagrams are truncated.
/// Up to `sender_len` bytes of the receiver address is written into
/// `sender_addr`. Returns `true` if no IO error occurred. The number of
/// received bytes is stored in `result` (can be 0).
bool write_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
const ip_endpoint& ep);
inline connection_handle conn_hdl_from_socket(native_socket fd) { inline connection_handle conn_hdl_from_socket(native_socket fd) {
return connection_handle::from_int(int64_from_native_socket(fd)); return connection_handle::from_int(int64_from_native_socket(fd));
} }
......
...@@ -18,25 +18,35 @@ ...@@ -18,25 +18,35 @@
#pragma once #pragma once
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/rw_state.hpp"
#include "caf/io/network/native_socket.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
/// Function signature of `read_some`.
using read_some_fun = decltype(io::network::read_some)*;
/// Function signature of `wite_some`.
using write_some_fun = decltype(io::network::write_some)*;
/// Function signature of `try_accept`.
using try_accept_fun = decltype(io::network::try_accept)*;
/// Policy object for wrapping default TCP operations. /// Policy object for wrapping default TCP operations.
struct tcp { struct tcp {
static read_some_fun read_some; /// Reads up to `len` bytes from `fd,` writing the received data
static write_some_fun write_some; /// to `buf`. Returns `true` as long as `fd` is readable and `false`
static try_accept_fun try_accept; /// if the socket has been closed or an IO error occured. The number
/// of read bytes is stored in `result` (can be 0).
static io::network::rw_state read_some(size_t& result,
io::network::native_socket fd,
void* buf, size_t len);
/// Writes up to `len` bytes from `buf` to `fd`.
/// Returns `true` as long as `fd` is readable and `false`
/// if the socket has been closed or an IO error occured. The number
/// of written bytes is stored in `result` (can be 0).
static io::network::rw_state write_some(size_t& result,
io::network::native_socket fd,
const void* buf, size_t len);
/// Tries to accept a new connection from `fd`. On success,
/// the new connection is stored in `result`. Returns true
/// as long as
static bool try_accept(io::network::native_socket& result,
io::network::native_socket fd);
}; };
} // namespace policy } // namespace policy
......
...@@ -18,21 +18,29 @@ ...@@ -18,21 +18,29 @@
#pragma once #pragma once
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/ip_endpoint.hpp"
#include "caf/io/network/native_socket.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
/// Function signature of read_datagram
using read_datagram_fun = decltype(io::network::read_datagram)*;
/// Function signature of write_datagram
using write_datagram_fun = decltype(io::network::write_datagram)*;
/// Policy object for wrapping default UDP operations /// Policy object for wrapping default UDP operations
struct udp { struct udp {
static read_datagram_fun read_datagram; /// Write a datagram containing `buf_len` bytes to `fd` addressed
static write_datagram_fun write_datagram; /// at the endpoint in `sa` with size `sa_len`. Returns true as long
/// as no IO error occurs. The number of written bytes is stored in
/// `result` and the sender is stored in `ep`.
static bool read_datagram(size_t& result, io::network::native_socket fd,
void* buf, size_t buf_len,
io::network::ip_endpoint& ep);
/// Reveice a datagram of up to `len` bytes. Larger datagrams are truncated.
/// Up to `sender_len` bytes of the receiver address is written into
/// `sender_addr`. Returns `true` if no IO error occurred. The number of
/// received bytes is stored in `result` (can be 0).
static bool write_datagram(size_t& result, io::network::native_socket fd,
void* buf, size_t buf_len,
const io::network::ip_endpoint& ep);
}; };
} // namespace policy } // namespace policy
......
...@@ -419,84 +419,6 @@ namespace network { ...@@ -419,84 +419,6 @@ namespace network {
} }
#endif // CAF_EPOLL_MULTIPLEXER #endif // CAF_EPOLL_MULTIPLEXER
// -- Free functions for sending and receiving data
rw_state read_some(size_t& result, native_socket fd, void* buf, size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
auto sres = ::recv(fd, reinterpret_cast<socket_recv_ptr>(buf),
len, no_sigpipe_io_flag);
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres));
if (is_error(sres, true) || sres == 0) {
// recv returns 0 when the peer has performed an orderly shutdown
return rw_state::failure;
}
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return rw_state::success;
}
rw_state write_some(size_t& result, native_socket fd, const void* buf,
size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
auto sres = ::send(fd, reinterpret_cast<socket_send_ptr>(buf),
len, no_sigpipe_io_flag);
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres));
if (is_error(sres, true))
return rw_state::failure;
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return rw_state::success;
}
bool try_accept(native_socket& result, native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
sockaddr_storage addr;
memset(&addr, 0, sizeof(addr));
socklen_t addrlen = sizeof(addr);
result = ::accept(fd, reinterpret_cast<sockaddr*>(&addr), &addrlen);
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(result));
if (result == invalid_native_socket) {
auto err = last_socket_error();
if (!would_block_or_temporarily_unavailable(err)) {
return false;
}
}
return true;
}
bool read_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
ip_endpoint& ep) {
CAF_LOG_TRACE(CAF_ARG(fd));
memset(ep.address(), 0, sizeof(sockaddr_storage));
socklen_t len = sizeof(sockaddr_storage);
auto sres = ::recvfrom(fd, static_cast<socket_recv_ptr>(buf), buf_len, 0, ep.address(), &len);
if (is_error(sres, true)) {
CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres));
return false;
}
if (sres == 0)
CAF_LOG_INFO("Received empty datagram");
else if (sres > static_cast<ssize_t>(buf_len))
CAF_LOG_WARNING("recvfrom cut of message, only received " << CAF_ARG(buf_len)
<< " of " << CAF_ARG(sres) << " bytes");
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
*ep.length() = static_cast<size_t>(len);
return true;
}
bool write_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
const ip_endpoint& ep) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(buf_len));
socklen_t len = static_cast<socklen_t>(*ep.clength());
auto sres = ::sendto(fd, reinterpret_cast<socket_send_ptr>(buf), buf_len,
0, ep.caddress(),
len);
if (is_error(sres, true)) {
CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres));
return false;
}
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return true;
}
// -- Helper functions for defining bitmasks of event handlers ----------------- // -- Helper functions for defining bitmasks of event handlers -----------------
...@@ -527,7 +449,7 @@ int del_flag(operation op, int bf) { ...@@ -527,7 +449,7 @@ int del_flag(operation op, int bf) {
// weird stuff going on // weird stuff going on
return 0; return 0;
} }
// -- Platform-independent parts of the default_multiplexer -------------------- // -- Platform-independent parts of the default_multiplexer --------------------
bool default_multiplexer::try_run_once() { bool default_multiplexer::try_run_once() {
...@@ -776,6 +698,8 @@ int64_t default_multiplexer::next_endpoint_id() { ...@@ -776,6 +698,8 @@ int64_t default_multiplexer::next_endpoint_id() {
return servant_ids_++; return servant_ids_++;
} }
// -- Related helper functions -------------------------------------------------
class socket_guard { class socket_guard {
public: public:
explicit socket_guard(native_socket fd) : fd_(fd) { explicit socket_guard(native_socket fd) : fd_(fd) {
......
...@@ -18,14 +18,59 @@ ...@@ -18,14 +18,59 @@
#include "caf/policy/tcp.hpp" #include "caf/policy/tcp.hpp"
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
using caf::io::network::is_error;
using caf::io::network::rw_state;
using caf::io::network::native_socket;
using caf::io::network::no_sigpipe_io_flag;
namespace caf { namespace caf {
namespace policy { namespace policy {
read_some_fun tcp::read_some = io::network::read_some; rw_state tcp::read_some(size_t& result, native_socket fd, void* buf,
size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
auto sres = ::recv(fd, reinterpret_cast<io::network::socket_recv_ptr>(buf),
len, no_sigpipe_io_flag);
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres));
if (is_error(sres, true) || sres == 0) {
// recv returns 0 when the peer has performed an orderly shutdown
return rw_state::failure;
}
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return rw_state::success;
}
rw_state tcp::write_some(size_t& result, native_socket fd, const void* buf,
size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
auto sres = ::send(fd, reinterpret_cast<io::network::socket_send_ptr>(buf),
len, no_sigpipe_io_flag);
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres));
if (is_error(sres, true))
return rw_state::failure;
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return rw_state::success;
}
write_some_fun tcp::write_some = io::network::write_some; bool tcp::try_accept(native_socket& result, native_socket fd) {
using namespace io::network;
CAF_LOG_TRACE(CAF_ARG(fd));
sockaddr_storage addr;
memset(&addr, 0, sizeof(addr));
socklen_t addrlen = sizeof(addr);
result = ::accept(fd, reinterpret_cast<sockaddr*>(&addr), &addrlen);
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(result));
if (result == invalid_native_socket) {
auto err = last_socket_error();
if (!would_block_or_temporarily_unavailable(err)) {
return false;
}
}
return true;
}
try_accept_fun tcp::try_accept = io::network::try_accept; } // namespace policy
} // policy
} // namespace caf } // namespace caf
...@@ -18,12 +18,50 @@ ...@@ -18,12 +18,50 @@
#include "caf/policy/udp.hpp" #include "caf/policy/udp.hpp"
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
using caf::io::network::is_error;
using caf::io::network::ip_endpoint;
using caf::io::network::native_socket;
namespace caf { namespace caf {
namespace policy { namespace policy {
read_datagram_fun udp::read_datagram = io::network::read_datagram;
write_datagram_fun udp::write_datagram = io::network::write_datagram; bool udp::read_datagram(size_t& result, native_socket fd, void* buf,
size_t buf_len, ip_endpoint& ep) {
CAF_LOG_TRACE(CAF_ARG(fd));
memset(ep.address(), 0, sizeof(sockaddr_storage));
socklen_t len = sizeof(sockaddr_storage);
auto sres = ::recvfrom(fd, static_cast<io::network::socket_recv_ptr>(buf),
buf_len, 0, ep.address(), &len);
if (is_error(sres, true)) {
CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres));
return false;
}
if (sres == 0)
CAF_LOG_INFO("Received empty datagram");
else if (sres > static_cast<ssize_t>(buf_len))
CAF_LOG_WARNING("recvfrom cut of message, only received "
<< CAF_ARG(buf_len) << " of " << CAF_ARG(sres) << " bytes");
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
*ep.length() = static_cast<size_t>(len);
return true;
}
bool udp::write_datagram(size_t& result, native_socket fd, void* buf,
size_t buf_len, const ip_endpoint& ep) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(buf_len));
socklen_t len = static_cast<socklen_t>(*ep.clength());
auto sres = ::sendto(fd, reinterpret_cast<io::network::socket_send_ptr>(buf),
buf_len, 0, ep.caddress(), len);
if (is_error(sres, true)) {
CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres));
return false;
}
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return true;
}
} // policy } // namespace policy
} // namespace caf } // namespace caf
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