Commit 3a760f89 authored by Dominik Charousset's avatar Dominik Charousset

Add make_network_socket_pair function

parent 0d121f35
......@@ -62,11 +62,11 @@ error allow_udp_connreset(network_socket x, bool new_value);
/// Get the socket buffer size for `x`.
/// @pre `x != invalid_network_socket`
/// @relates network_socket
expected<int> send_buffer_size(network_socket x);
expected<size_t> send_buffer_size(network_socket x);
/// Set the socket buffer size for `x`.
/// @relates network_socket
error send_buffer_size(network_socket x, int new_value);
error send_buffer_size(network_socket x, size_t capacity);
/// Returns the locally assigned port of `x`.
/// @relates network_socket
......@@ -109,9 +109,15 @@ variant<size_t, std::errc> write(network_socket x, const void* buf,
/// @param x Connected endpoint.
/// @param buf Points to destination buffer.
/// @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, 0 if the connection was
/// closed and an error code otherwise.
/// @relates pipe_socket
variant<size_t, std::errc> read(network_socket x, void* buf, size_t buf_size);
/// Creates two connected sockets to mimic network communication (usually for
/// testing purposes).
/// @relates network_socket
expected<std::pair<network_socket, network_socket>> make_network_socket_pair();
} // namespace net
} // namespace caf
......@@ -40,6 +40,12 @@ struct socket : abstract_socket<socket> {
/// Denotes the invalid socket.
constexpr auto invalid_socket = socket{invalid_socket_id};
/// Converts between different socket types.
template <class To, class From>
To socket_cast(From x) {
return To{x.id};
}
/// Close socket `x`.
/// @relates socket
void close(socket x);
......
......@@ -140,17 +140,18 @@ error allow_udp_connreset(network_socket x, bool) {
#endif // CAF_WINDOWS
expected<int> send_buffer_size(network_socket x) {
expected<size_t> send_buffer_size(network_socket x) {
int size = 0;
socket_size_type ret_size = sizeof(size);
CAF_NET_SYSCALL("getsockopt", res, !=, 0,
getsockopt(x.id, SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<getsockopt_ptr>(&size),
&ret_size));
return size;
return static_cast<size_t>(size);
}
error send_buffer_size(network_socket x, int new_value) {
error send_buffer_size(network_socket x, size_t capacity) {
auto new_value = static_cast<int>(capacity);
CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<setsockopt_ptr>(&new_value),
......@@ -253,5 +254,102 @@ variant<size_t, std::errc> read(network_socket x, void* buf, size_t buf_size) {
return static_cast<size_t>(res);
}
#ifdef CAF_WINDOWS
/**************************************************************************\
* Based on work of others; *
* original header: *
* *
* Copyright 2007, 2010 by Nathan C. Myers <ncm@cantrip.org> *
* Redistribution and use in source and binary forms, with or without *
* modification, are permitted provided that the following conditions *
* are met: *
* *
* Redistributions of source code must retain the above copyright notice, *
* this list of conditions and the following disclaimer. *
* *
* Redistributions in binary form must reproduce the above copyright *
* notice, this list of conditions and the following disclaimer in the *
* documentation and/or other materials provided with the distribution. *
* *
* The name of the author must not be used to endorse or promote products *
* derived from this software without specific prior written permission. *
* *
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS *
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT *
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR *
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT *
* HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, *
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT *
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, *
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY *
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT *
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE *
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
\**************************************************************************/
expected<std::pair<network_socket, network_socket>> make_network_socket_pair() {
auto addrlen = static_cast<int>(sizeof(sockaddr_in));
socket_id socks[2] = {invalid_socket_id, invalid_socket_id};
CAF_NET_SYSCALL("socket", listener, ==, invalid_socket_id,
::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
union {
sockaddr_in inaddr;
sockaddr addr;
} a;
memset(&a, 0, sizeof(a));
a.inaddr.sin_family = AF_INET;
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_port = 0;
// makes sure all sockets are closed in case of an error
auto guard = detail::make_scope_guard([&] {
auto e = WSAGetLastError();
close(socket{listener});
close(socket{socks[0]});
close(socket{socks[1]});
WSASetLastError(e);
});
// bind listener to a local port
int reuse = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<char*>(&reuse),
static_cast<int>(sizeof(reuse))));
CAF_NET_SYSCALL("bind", tmp2, !=, 0,
bind(listener, &a.addr, static_cast<int>(sizeof(a.inaddr))));
// Read the port in use: win32 getsockname may only set the port number
// (http://msdn.microsoft.com/library/ms738543.aspx).
memset(&a, 0, sizeof(a));
CAF_NET_SYSCALL("getsockname", tmp3, !=, 0,
getsockname(listener, &a.addr, &addrlen));
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_family = AF_INET;
// set listener to listen mode
CAF_NET_SYSCALL("listen", tmp5, !=, 0, listen(listener, 1));
// create read-only end of the pipe
DWORD flags = 0;
CAF_NET_SYSCALL("WSASocketW", read_fd, ==, invalid_socket_id,
WSASocketW(AF_INET, SOCK_STREAM, 0, nullptr, 0, flags));
CAF_NET_SYSCALL("connect", tmp6, !=, 0,
connect(read_fd, &a.addr,
static_cast<int>(sizeof(a.inaddr))));
// get write-only end of the pipe
CAF_NET_SYSCALL("accept", write_fd, ==, invalid_socket_id,
accept(listener, nullptr, nullptr));
close(socket{listener});
guard.disable();
return std::make_pair(read_fd, write_fd);
}
#else // CAF_WINDOWS
expected<std::pair<network_socket, network_socket>> make_network_socket_pair() {
int sockets[2];
CAF_NET_SYSCALL("socketpair", spair_res, !=, 0,
socketpair(AF_UNIX, SOCK_STREAM, 0, sockets));
return std::make_pair(network_socket{sockets[0]}, network_socket{sockets[1]});
}
#endif // CAF_WINDOWS
} // namespace net
} // namespace caf
......@@ -20,6 +20,7 @@
#include <cstdio>
#include <cstdlib>
#include <utility>
#include "caf/config.hpp"
#include "caf/detail/net_syscall.hpp"
......@@ -35,88 +36,17 @@ namespace net {
#ifdef CAF_WINDOWS
/**************************************************************************\
* Based on work of others; *
* original header: *
* *
* Copyright 2007, 2010 by Nathan C. Myers <ncm@cantrip.org> *
* Redistribution and use in source and binary forms, with or without *
* modification, are permitted provided that the following conditions *
* are met: *
* *
* Redistributions of source code must retain the above copyright notice, *
* this list of conditions and the following disclaimer. *
* *
* Redistributions in binary form must reproduce the above copyright *
* notice, this list of conditions and the following disclaimer in the *
* documentation and/or other materials provided with the distribution. *
* *
* The name of the author must not be used to endorse or promote products *
* derived from this software without specific prior written permission. *
* *
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS *
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT *
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR *
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT *
* HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, *
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT *
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, *
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY *
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT *
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE *
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
\**************************************************************************/
expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
auto addrlen = static_cast<int>(sizeof(sockaddr_in));
socket_id socks[2] = {invalid_socket_id, invalid_socket_id};
CAF_NET_SYSCALL("socket", listener, ==, invalid_socket_id,
::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
union {
sockaddr_in inaddr;
sockaddr addr;
} a;
memset(&a, 0, sizeof(a));
a.inaddr.sin_family = AF_INET;
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_port = 0;
// makes sure all sockets are closed in case of an error
auto guard = detail::make_scope_guard([&] {
auto e = WSAGetLastError();
close(socket{listener});
close(socket{socks[0]});
close(socket{socks[1]});
WSASetLastError(e);
});
// bind listener to a local port
int reuse = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<char*>(&reuse),
static_cast<int>(sizeof(reuse))));
CAF_NET_SYSCALL("bind", tmp2, !=, 0,
bind(listener, &a.addr, static_cast<int>(sizeof(a.inaddr))));
// Read the port in use: win32 getsockname may only set the port number
// (http://msdn.microsoft.com/library/ms738543.aspx).
memset(&a, 0, sizeof(a));
CAF_NET_SYSCALL("getsockname", tmp3, !=, 0,
getsockname(listener, &a.addr, &addrlen));
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_family = AF_INET;
// set listener to listen mode
CAF_NET_SYSCALL("listen", tmp5, !=, 0, listen(listener, 1));
// create read-only end of the pipe
DWORD flags = 0;
CAF_NET_SYSCALL("WSASocketW", read_fd, ==, invalid_socket_id,
WSASocketW(AF_INET, SOCK_STREAM, 0, nullptr, 0, flags));
CAF_NET_SYSCALL("connect", tmp6, !=, 0,
connect(read_fd, &a.addr,
static_cast<int>(sizeof(a.inaddr))));
// get write-only end of the pipe
CAF_NET_SYSCALL("accept", write_fd, ==, invalid_socket_id,
accept(listener, nullptr, nullptr));
close(socket{listener});
guard.disable();
return std::make_pair(read_fd, write_fd);
// Windows has no support for unidirectional pipes. Emulate pipes by using a
// pair of regular TCP sockets with read/write channels closed appropriately.
if (auto result = make_network_socket_pair()) {
shutdown_write(result->first);
shutdown_read(result->second);
return std::make_pair(socket_cast<pipe_socket>(result->first),
socket_cast<pipe_socket>(result->second));
} else {
return std::move(result.error());
}
}
variant<size_t, std::errc> write(pipe_socket x, const void* buf,
......
......@@ -42,4 +42,35 @@ CAF_TEST(invalid socket) {
CAF_CHECK_EQUAL(remote_addr(x), sec::network_syscall_failed);
}
CAF_TEST(connected socket pair) {
std::vector<char> wr_buf{1, 2, 4, 8, 16, 32, 64};
std::vector<char> rd_buf(124);
CAF_MESSAGE("create sockets and configure nonblocking I/O");
auto x = unbox(make_network_socket_pair());
CAF_CHECK_EQUAL(nonblocking(x.first, true), caf::none);
CAF_CHECK_EQUAL(nonblocking(x.second, true), caf::none);
CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.first)), 0u);
CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.second)), 0u);
CAF_MESSAGE("verify nonblocking communication");
CAF_CHECK_EQUAL(read(x.first, rd_buf.data(), rd_buf.size()),
std::errc::operation_would_block);
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()),
std::errc::operation_would_block);
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(read(x.second, rd_buf.data(), rd_buf.size()), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), 0);
CAF_MESSAGE("transfer data from second to first socket");
CAF_CHECK_EQUAL(write(x.second, wr_buf.data(), wr_buf.size()), wr_buf.size());
CAF_CHECK_EQUAL(read(x.first, rd_buf.data(), rd_buf.size()), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), 0);
CAF_MESSAGE("shut down first socket and observe shutdown on the second one");
close(x.first);
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), size_t{0});
CAF_MESSAGE("done (cleanup)");
close(x.second);
}
CAF_TEST_FIXTURE_SCOPE_END()
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