Commit 1aeb0dd6 authored by Jakob Otto's avatar Jakob Otto

Implement write(span<vector>) on Windows

parent 9a364460
...@@ -52,5 +52,6 @@ ...@@ -52,5 +52,6 @@
# include <netinet/tcp.h> # include <netinet/tcp.h>
# include <sys/socket.h> # include <sys/socket.h>
# include <unistd.h> # include <unistd.h>
# include <sys/uio.h>
#endif #endif
// clang-format on // clang-format on
...@@ -316,7 +316,7 @@ private: ...@@ -316,7 +316,7 @@ private:
size_t max_consecutive_reads_; size_t max_consecutive_reads_;
size_t read_threshold_; size_t read_threshold_;
// size_t collected_;
size_t max_; size_t max_;
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
......
...@@ -17,7 +17,6 @@ ...@@ -17,7 +17,6 @@
******************************************************************************/ ******************************************************************************/
#include "caf/net/udp_datagram_socket.hpp" #include "caf/net/udp_datagram_socket.hpp"
#include <sys/uio.h>
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/convert_ip_endpoint.hpp" #include "caf/detail/convert_ip_endpoint.hpp"
...@@ -111,11 +110,57 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, ...@@ -111,11 +110,57 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
} }
} }
variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep) {
sockaddr_storage addr = {};
detail::convert(ep, addr);
auto len = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6);
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len);
auto ret = check_udp_datagram_socket_io_res(res);
if (auto num_bytes = get_if<size_t>(&ret))
return *num_bytes;
else
return get<sec>(ret);
}
#ifdef CAF_WINDOWS
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
ip_endpoint ep) {
CAF_ASSERT(bufs.size() < 10);
WSABUF buf_array[10];
auto convert = [](std::vector<byte>* buf) {
return WSABUF{static_cast<ULONG>(buf->size()),
reinterpret_cast<CHAR*>(buf->data())};
};
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
sockaddr_storage addr = {};
detail::convert(ep, addr);
auto len = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6);
DWORD bytes_sent = 0;
auto res = WSASendTo(x.id, buf_array, static_cast<DWORD>(bufs.size()),
&bytes_sent, 0, reinterpret_cast<sockaddr*>(&addr), len,
nullptr, nullptr);
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>(bytes_sent);
}
#else // CAF_WINDOWS
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs, variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
ip_endpoint ep) { ip_endpoint ep) {
CAF_ASSERT(bufs.size() < 10); CAF_ASSERT(bufs.size() < 10);
auto convert = [](std::vector<byte>* buf) { auto convert = [](std::vector<byte>* buf) {
return iovec{const_cast<byte*>(buf->data()), buf->size()}; return iovec{buf->data(), buf->size()};
}; };
sockaddr_storage addr = {}; sockaddr_storage addr = {};
detail::convert(ep, addr); detail::convert(ep, addr);
...@@ -132,20 +177,7 @@ variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs, ...@@ -132,20 +177,7 @@ variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
return check_udp_datagram_socket_io_res(res); return check_udp_datagram_socket_io_res(res);
} }
variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf, #endif // CAF_WINDOWS
ip_endpoint ep) {
sockaddr_storage addr = {};
detail::convert(ep, addr);
auto len = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6);
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len);
auto ret = check_udp_datagram_socket_io_res(res);
if (auto num_bytes = get_if<size_t>(&ret))
return *num_bytes;
else
return get<sec>(ret);
}
variant<size_t, sec> variant<size_t, sec>
check_udp_datagram_socket_io_res(std::make_signed<size_t>::type res) { check_udp_datagram_socket_io_res(std::make_signed<size_t>::type res) {
......
...@@ -72,6 +72,23 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -72,6 +72,23 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
return mpx->poll_once(false); return mpx->poll_once(false);
} }
error read_from_socket(udp_datagram_socket sock, std::vector<byte>& buf) {
uint8_t receive_attempts = 0;
variant<std::pair<size_t, ip_endpoint>, sec> read_ret;
do {
read_ret = read(sock, make_span(buf));
if (auto read_res = get_if<std::pair<size_t, ip_endpoint>>(&read_ret)) {
buf.resize(read_res->first);
} else if (get<sec>(read_ret) != sec::unavailable_or_would_block) {
return make_error(get<sec>(read_ret), "read failed");
}
if (++receive_attempts > 100)
return make_error(sec::runtime_error,
"too many unavailable_or_would_blocks");
} while (read_ret.index() != 0);
return none;
}
multiplexer_ptr mpx; multiplexer_ptr mpx;
std::shared_ptr<std::vector<byte>> recv_buf; std::shared_ptr<std::vector<byte>> recv_buf;
ip_endpoint ep; ip_endpoint ep;
...@@ -214,12 +231,7 @@ CAF_TEST(resolve and proxy communication) { ...@@ -214,12 +231,7 @@ CAF_TEST(resolve and proxy communication) {
after(std::chrono::seconds(0)) >> after(std::chrono::seconds(0)) >>
[&] { CAF_FAIL("manager did not respond with a proxy."); }); [&] { CAF_FAIL("manager did not respond with a proxy."); });
run(); run();
auto read_res = read(recv_socket, make_span(*recv_buf)); CAF_CHECK_EQUAL(read_from_socket(recv_socket, *recv_buf), none);
if (!holds_alternative<std::pair<size_t, ip_endpoint>>(read_res))
CAF_FAIL("read() returned an error: " << sys.render(get<sec>(read_res)));
recv_buf->resize(get<std::pair<size_t, ip_endpoint>>(read_res).first);
CAF_MESSAGE("received message from " << to_string(
get<std::pair<size_t, ip_endpoint>>(read_res).second));
CAF_MESSAGE("receive buffer contains " << recv_buf->size() << " bytes"); CAF_MESSAGE("receive buffer contains " << recv_buf->size() << " bytes");
message msg; message msg;
binary_deserializer source{sys, *recv_buf}; binary_deserializer source{sys, *recv_buf};
......
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