Commit 792c3213 authored by Dominik Charousset's avatar Dominik Charousset

Simplify send/receive API

parent 6c45312f
...@@ -11,7 +11,6 @@ set(LIBCAF_NET_SRCS ...@@ -11,7 +11,6 @@ set(LIBCAF_NET_SRCS
src/network_socket.cpp src/network_socket.cpp
src/pipe_socket.cpp src/pipe_socket.cpp
src/socket.cpp src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp src/stream_socket.cpp
) )
......
...@@ -16,20 +16,28 @@ ...@@ -16,20 +16,28 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/net/socket_manager.hpp" #pragma once
#include "caf/config.hpp" #include <memory>
#include "caf/intrusive_ptr.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
socket_manager::socket_manager(socket handle) : handle_(handle) { class multiplexer;
CAF_ASSERT(handle != invalid_socket); class socket_manager;
}
struct network_socket;
struct pipe_socket;
struct socket;
struct stream_socket;
using socket_manager_ptr = intrusive_ptr<socket_manager>;
using multiplexer_ptr = std::shared_ptr<multiplexer>;
socket_manager::~socket_manager() { using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
close(handle_);
}
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -18,51 +18,32 @@ ...@@ -18,51 +18,32 @@
#pragma once #pragma once
#include "caf/error.hpp"
#include "caf/net/socket.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
/// Manages the lifetime of a single socket and handles any I/O events on it. /// Values for representing bitmask of I/O operations.
class socket_manager { enum class operation {
public: none = 0x00,
// -- constructors, destructors, and assignment operators -------------------- read = 0x01,
write = 0x02,
explicit socket_manager(socket handle); read_write = 0x03,
};
virtual ~socket_manager();
socket_manager(const socket_manager&) = delete;
socket_manager& operator=(const socket_manager&) = delete;
// -- properties -------------------------------------------------------------
socket handle() const noexcept {
return handle_;
}
// -- pure virtual member functions ------------------------------------------
/// Called whenever the socket received new data.
virtual error handle_read_event() = 0;
/// Called whenever the socket is allowed to send additional data. constexpr operation operator|(operation x, operation y) {
virtual error handle_write_event() = 0; return static_cast<operation>(static_cast<int>(x) | static_cast<int>(y));
}
/// Called if the remote side becomes unreachable. constexpr operation operator&(operation x, operation y) {
/// @param reason A default-constructed error object (no error) if the remote return static_cast<operation>(static_cast<int>(x) & static_cast<int>(y));
/// host performed an ordinary shutdown (only for connection }
/// oriented sockets), otherwise an error code as reported by
/// the operating system.
virtual void handle_shutdown_event(error reason) = 0;
private: constexpr operation operator^(operation x, operation y) {
// -- member variables ------------------------------------------------------- return static_cast<operation>(static_cast<int>(x) ^ static_cast<int>(y));
}
socket handle_; constexpr operation operator~(operation x) {
}; return static_cast<operation>(~static_cast<int>(x));
}
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -52,8 +52,7 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe(); ...@@ -52,8 +52,7 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe();
/// @param buf_size Specifies the size of the buffer in bytes. /// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code. /// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket /// @relates pipe_socket
variant<size_t, std::errc> write(pipe_socket x, const void* buf, variant<size_t, sec> write(pipe_socket x, const void* buf, size_t buf_size);
size_t buf_size);
/// Receives data from `x`. /// Receives data from `x`.
/// @param x Connected endpoint. /// @param x Connected endpoint.
...@@ -61,7 +60,7 @@ variant<size_t, std::errc> write(pipe_socket x, const void* buf, ...@@ -61,7 +60,7 @@ variant<size_t, std::errc> write(pipe_socket x, const void* buf,
/// @param buf_size Specifies the maximum size of the buffer in bytes. /// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, otherwise an error code. /// @returns The number of received bytes on success, otherwise an error code.
/// @relates pipe_socket /// @relates pipe_socket
variant<size_t, std::errc> read(pipe_socket x, void* buf, size_t buf_size); variant<size_t, sec> read(pipe_socket x, void* buf, size_t buf_size);
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -68,5 +68,10 @@ error child_process_inherit(socket x, bool new_value); ...@@ -68,5 +68,10 @@ error child_process_inherit(socket x, bool new_value);
/// @relates socket /// @relates socket
error nonblocking(socket x, bool new_value); error nonblocking(socket x, bool new_value);
/// Converts the result from `recv` or `send` to either an error code or a
/// non-zero positive integer.
/// @relates socket
variant<size_t, sec> check_socket_io_res(std::make_signed<size_t>::type res);
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -57,10 +57,10 @@ error nodelay(stream_socket x, bool new_value); ...@@ -57,10 +57,10 @@ error nodelay(stream_socket x, bool new_value);
/// @param x Connected endpoint. /// @param x Connected endpoint.
/// @param buf Points to destination buffer. /// @param buf Points to destination buffer.
/// @param buf_size Specifies the maximum size of the buffer in bytes. /// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, 0 if the connection was /// @returns The number of received bytes on success, an error code otherwise.
/// closed and an error code otherwise.
/// @relates pipe_socket /// @relates pipe_socket
variant<size_t, std::errc> read(stream_socket x, void* buf, size_t buf_size); /// @post either the result is a `sec` or a positive (non-zero) integer.
variant<size_t, sec> read(stream_socket x, void* buf, size_t buf_size);
/// Transmits data from `x` to its peer. /// Transmits data from `x` to its peer.
/// @param x Connected endpoint. /// @param x Connected endpoint.
...@@ -68,8 +68,8 @@ variant<size_t, std::errc> read(stream_socket x, void* buf, size_t buf_size); ...@@ -68,8 +68,8 @@ variant<size_t, std::errc> read(stream_socket x, void* buf, size_t buf_size);
/// @param buf_size Specifies the size of the buffer in bytes. /// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code. /// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket /// @relates pipe_socket
variant<size_t, std::errc> write(stream_socket x, const void* buf, /// @post either the result is a `sec` or a positive (non-zero) integer.
size_t buf_size); variant<size_t, sec> write(stream_socket x, const void* buf, size_t buf_size);
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -49,13 +49,12 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() { ...@@ -49,13 +49,12 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
} }
} }
variant<size_t, std::errc> write(pipe_socket x, const void* buf, variant<size_t, sec> write(pipe_socket x, const void* buf, size_t buf_size) {
size_t buf_size) {
// On Windows, a pipe consists of two stream sockets. // On Windows, a pipe consists of two stream sockets.
return write(socket_cast<stream_socket>(x), buf, buf_size); return write(socket_cast<stream_socket>(x), buf, buf_size);
} }
variant<size_t, std::errc> read(pipe_socket x, void* buf, size_t buf_size) { variant<size_t, sec> read(pipe_socket x, void* buf, size_t buf_size) {
// On Windows, a pipe consists of two stream sockets. // On Windows, a pipe consists of two stream sockets.
return read(socket_cast<stream_socket>(x), buf, buf_size); return read(socket_cast<stream_socket>(x), buf, buf_size);
} }
...@@ -81,19 +80,14 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() { ...@@ -81,19 +80,14 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
return std::make_pair(pipe_socket{pipefds[0]}, pipe_socket{pipefds[1]}); return std::make_pair(pipe_socket{pipefds[0]}, pipe_socket{pipefds[1]});
} }
variant<size_t, std::errc> write(pipe_socket x, const void* buf, variant<size_t, sec> write(pipe_socket x, const void* buf, size_t buf_size) {
size_t buf_size) {
auto res = ::write(x.id, buf, buf_size); auto res = ::write(x.id, buf, buf_size);
if (res < 0) return check_socket_io_res(res);
return last_socket_error();
return static_cast<size_t>(res);
} }
variant<size_t, std::errc> read(pipe_socket x, void* buf, size_t buf_size) { variant<size_t, sec> read(pipe_socket x, void* buf, size_t buf_size) {
auto res = ::read(x.id, buf, buf_size); auto res = ::read(x.id, buf, buf_size);
if (res < 0) return check_socket_io_res(res);
return last_socket_error();
return static_cast<size_t>(res);
} }
#endif // CAF_WINDOWS #endif // CAF_WINDOWS
......
...@@ -24,6 +24,8 @@ ...@@ -24,6 +24,8 @@
#include "caf/detail/net_syscall.hpp" #include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
...@@ -178,5 +180,18 @@ error nonblocking(socket x, bool new_value) { ...@@ -178,5 +180,18 @@ error nonblocking(socket x, bool new_value) {
#endif // CAF_WINDOWS #endif // CAF_WINDOWS
variant<size_t, sec> check_socket_io_res(std::make_signed<size_t>::type res) {
if (res == 0)
return sec::socket_disconnected;
if (res < 0) {
auto code = last_socket_error();
if (code == std::errc::operation_would_block
|| code == std::errc::resource_unavailable_try_again)
return sec::unavailable_or_would_block;
return sec::socket_operation_failed;
}
return static_cast<size_t>(res);
}
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -161,21 +161,16 @@ error nodelay(stream_socket x, bool new_value) { ...@@ -161,21 +161,16 @@ error nodelay(stream_socket x, bool new_value) {
return none; return none;
} }
variant<size_t, std::errc> read(stream_socket x, void* buf, size_t buf_size) { variant<size_t, sec> read(stream_socket x, void* buf, size_t buf_size) {
auto res = ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf), buf_size, auto res = ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf), buf_size,
no_sigpipe_io_flag); no_sigpipe_io_flag);
if (res < 0) return check_socket_io_res(res);
return last_socket_error();
return static_cast<size_t>(res);
} }
variant<size_t, std::errc> write(stream_socket x, const void* buf, variant<size_t, sec> write(stream_socket x, const void* buf, size_t buf_size) {
size_t buf_size) {
auto res = ::send(x.id, reinterpret_cast<socket_send_ptr>(buf), buf_size, auto res = ::send(x.id, reinterpret_cast<socket_send_ptr>(buf), buf_size,
no_sigpipe_io_flag); no_sigpipe_io_flag);
if (res < 0) return check_socket_io_res(res);
return last_socket_error();
return static_cast<size_t>(res);
} }
} // namespace net } // namespace net
......
...@@ -47,9 +47,9 @@ CAF_TEST(connected socket pair) { ...@@ -47,9 +47,9 @@ CAF_TEST(connected socket pair) {
CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.second)), 0u); CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.second)), 0u);
CAF_MESSAGE("verify nonblocking communication"); CAF_MESSAGE("verify nonblocking communication");
CAF_CHECK_EQUAL(read(x.first, rd_buf.data(), rd_buf.size()), CAF_CHECK_EQUAL(read(x.first, rd_buf.data(), rd_buf.size()),
std::errc::operation_would_block); sec::unavailable_or_would_block);
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()),
std::errc::operation_would_block); sec::unavailable_or_would_block);
CAF_MESSAGE("transfer data from first to second socket"); CAF_MESSAGE("transfer data from first to second socket");
CAF_CHECK_EQUAL(write(x.first, wr_buf.data(), wr_buf.size()), wr_buf.size()); CAF_CHECK_EQUAL(write(x.first, wr_buf.data(), wr_buf.size()), wr_buf.size());
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), wr_buf.size()); CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), wr_buf.size());
...@@ -62,7 +62,8 @@ CAF_TEST(connected socket pair) { ...@@ -62,7 +62,8 @@ CAF_TEST(connected socket pair) {
rd_buf.assign(rd_buf.size(), 0); rd_buf.assign(rd_buf.size(), 0);
CAF_MESSAGE("shut down first socket and observe shutdown on the second one"); CAF_MESSAGE("shut down first socket and observe shutdown on the second one");
close(x.first); close(x.first);
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), size_t{0}); CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()),
sec::socket_disconnected);
CAF_MESSAGE("done (cleanup)"); CAF_MESSAGE("done (cleanup)");
close(x.second); close(x.second);
} }
......
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