Commit 04de0609 authored by Joseph Noir's avatar Joseph Noir

Socket does not stop reading the same packet

parent 9b548fbb
...@@ -45,6 +45,8 @@ ...@@ -45,6 +45,8 @@
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include <iomanip> // TODO: Remove this
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
# ifndef WIN32_LEAN_AND_MEAN # ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN # define WIN32_LEAN_AND_MEAN
...@@ -178,6 +180,20 @@ bool write_some(size_t& result, native_socket fd, const void* buf, size_t len); ...@@ -178,6 +180,20 @@ bool write_some(size_t& result, native_socket fd, const void* buf, size_t len);
/// as long as /// as long as
bool try_accept(native_socket& result, native_socket fd); 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`.
bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
sockaddr_storage& sa, size_t sa_len);
/// 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 receive_datagram(size_t& result, native_socket fd, void* buf, size_t len,
sockaddr_storage& sender_addr, socklen_t& sender_len);
class default_multiplexer; class default_multiplexer;
/// A socket I/O event handler. /// A socket I/O event handler.
...@@ -361,16 +377,20 @@ private: ...@@ -361,16 +377,20 @@ private:
CAF_LOG_TRACE(CAF_ARG(op) << CAF_ARG(fd) << CAF_ARG(old_bf)); CAF_LOG_TRACE(CAF_ARG(op) << CAF_ARG(fd) << CAF_ARG(old_bf));
auto last = events_.end(); auto last = events_.end();
auto i = std::lower_bound(events_.begin(), last, fd, event_less{}); auto i = std::lower_bound(events_.begin(), last, fd, event_less{});
// std::cerr << "Events: " << std::distance(events_.begin(), i) << std::endl;
if (i != last && i->fd == fd) { if (i != last && i->fd == fd) {
CAF_ASSERT(ptr == i->ptr); CAF_ASSERT(ptr == i->ptr);
// squash events together // squash events together
CAF_LOG_DEBUG("squash events:" << CAF_ARG(i->mask) CAF_LOG_DEBUG("squash events:" << CAF_ARG(i->mask)
<< CAF_ARG(fun(op, i->mask))); << CAF_ARG(fun(op, i->mask)));
//std::cerr << "[NE] Squash events:" << std::ios::hex << i->mask
// << " --> " << std::ios::hex << fun(op, i->mask) << std::endl;
auto bf = i->mask; auto bf = i->mask;
i->mask = fun(op, bf); i->mask = fun(op, bf);
if (i->mask == bf) { if (i->mask == bf) {
// didn't do a thing // didn't do a thing
CAF_LOG_DEBUG("squashing did not change the event"); CAF_LOG_DEBUG("squashing did not change the event");
//std::cerr << "[NE] Squashing did not change the event" << std::endl;
} else if (i->mask == old_bf) { } else if (i->mask == old_bf) {
// just turned into a nop // just turned into a nop
CAF_LOG_DEBUG("squashing events resulted in a NOP"); CAF_LOG_DEBUG("squashing events resulted in a NOP");
...@@ -380,10 +400,16 @@ private: ...@@ -380,10 +400,16 @@ private:
// insert new element // insert new element
auto bf = fun(op, old_bf); auto bf = fun(op, old_bf);
if (bf == old_bf) { if (bf == old_bf) {
CAF_LOG_DEBUG("event has no effect (discarded): " CAF_LOG_DEBUG("[NE] Event has no effect (discarded): "
<< CAF_ARG(bf) << ", " << CAF_ARG(old_bf)); << CAF_ARG(bf) << ", " << CAF_ARG(old_bf));
//std::cerr << "[NE] Event has no effect (discarded): "
// << std::ios::hex << bf << ", "
// << std::ios::hex << old_bf << std::endl;
} else { } else {
CAF_LOG_DEBUG("added handler:" << CAF_ARG(fd) << CAF_ARG(op)); CAF_LOG_DEBUG("added handler:" << CAF_ARG(fd) << CAF_ARG(op));
//std::cerr << "[NE] Added handler: "
// << std::ios::hex << bf << ", "
// << std::ios::hex << to_string(op) << std::endl;
events_.insert(i, event{fd, bf, ptr}); events_.insert(i, event{fd, bf, ptr});
} }
} }
...@@ -619,8 +645,6 @@ public: ...@@ -619,8 +645,6 @@ public:
void handle_event(operation op) override; void handle_event(operation op) override;
void sender_from_sockaddr(const sockaddr_storage& sa, size_t len);
void set_endpoint_addr(const sockaddr_storage& sa, size_t len, void set_endpoint_addr(const sockaddr_storage& sa, size_t len,
bool is_tmp_endpoint); bool is_tmp_endpoint);
...@@ -703,8 +727,6 @@ public: ...@@ -703,8 +727,6 @@ public:
void handle_event(operation op) override; void handle_event(operation op) override;
void sender_from_sockaddr(const sockaddr_storage& sa, size_t len);
std::pair<const sockaddr_storage&,size_t> last_sender(); std::pair<const sockaddr_storage&,size_t> last_sender();
private: private:
...@@ -738,6 +760,8 @@ new_dgram_scribe_impl(const std::string& host, uint16_t port, ...@@ -738,6 +760,8 @@ new_dgram_scribe_impl(const std::string& host, uint16_t port,
expected<std::pair<native_socket, uint16_t>> expected<std::pair<native_socket, uint16_t>>
new_dgram_doorman_impl(uint16_t port, const char* addr, bool reuse_addr); new_dgram_doorman_impl(uint16_t port, const char* addr, bool reuse_addr);
std::tuple<std::string,uint16_t> sender_from_sockaddr(const sockaddr_storage& sa, size_t len);
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
...@@ -463,8 +463,11 @@ expected<uint16_t> remote_port_of_fd(native_socket fd); ...@@ -463,8 +463,11 @@ expected<uint16_t> remote_port_of_fd(native_socket fd);
presult = ::WSAPoll(pollset_.data(), presult = ::WSAPoll(pollset_.data(),
static_cast<ULONG>(pollset_.size()), -1); static_cast<ULONG>(pollset_.size()), -1);
# else # else
std::cerr << std::endl << "[R] Poll time!" << std::endl;
presult = ::poll(pollset_.data(), presult = ::poll(pollset_.data(),
static_cast<nfds_t>(pollset_.size()), -1); static_cast<nfds_t>(pollset_.size()), -1);
std::cerr << "[R] Structures with nonzero revents fields: "
<< presult << std::endl;
# endif # endif
if (presult < 0) { if (presult < 0) {
switch (last_socket_error()) { switch (last_socket_error()) {
...@@ -495,6 +498,9 @@ expected<uint16_t> remote_port_of_fd(native_socket fd); ...@@ -495,6 +498,9 @@ expected<uint16_t> remote_port_of_fd(native_socket fd);
if (pfd.revents != 0) { if (pfd.revents != 0) {
CAF_LOG_DEBUG("event on socket:" << CAF_ARG(pfd.fd) CAF_LOG_DEBUG("event on socket:" << CAF_ARG(pfd.fd)
<< CAF_ARG(pfd.revents)); << CAF_ARG(pfd.revents));
std::cerr << "[R] Event on socket " << pfd.fd
<< " --> " << std::ios::hex
<< pfd.revents << std::endl;
poll_res.push_back({pfd.fd, pfd.revents, shadow_[i]}); poll_res.push_back({pfd.fd, pfd.revents, shadow_[i]});
pfd.revents = 0; pfd.revents = 0;
--presult; // stop as early as possible --presult; // stop as early as possible
...@@ -508,6 +514,8 @@ expected<uint16_t> remote_port_of_fd(native_socket fd); ...@@ -508,6 +514,8 @@ expected<uint16_t> remote_port_of_fd(native_socket fd);
handle_socket_event(e.fd, e.mask, e.ptr); handle_socket_event(e.fd, e.mask, e.ptr);
} }
CAF_LOG_DEBUG(CAF_ARG(events_.size())); CAF_LOG_DEBUG(CAF_ARG(events_.size()));
std::cerr << "[R] After poll we have " << events_.size()
<< " to handle" << std::endl;
poll_res.clear(); poll_res.clear();
for (auto& me : events_) { for (auto& me : events_) {
handle(me); handle(me);
...@@ -666,17 +674,22 @@ void default_multiplexer::close_pipe() { ...@@ -666,17 +674,22 @@ void default_multiplexer::close_pipe() {
void default_multiplexer::handle_socket_event(native_socket fd, int mask, void default_multiplexer::handle_socket_event(native_socket fd, int mask,
event_handler* ptr) { event_handler* ptr) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(mask)); CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(mask));
std::cerr << "[HSE] For " << fd << " with mask " << std::ios::hex << mask
<< std::endl;
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
bool checkerror = true; bool checkerror = true;
if (mask & input_mask) { if (mask & input_mask) {
checkerror = false; checkerror = false;
// ignore read events if a previous event caused // ignore read events if a previous event caused
// this socket to be shut down for reading // this socket to be shut down for reading
if (!ptr->read_channel_closed()) if (!ptr->read_channel_closed()) {
std::cerr << "[HSE] Next up: read event" << std::endl;
ptr->handle_event(operation::read); ptr->handle_event(operation::read);
} }
}
if (mask & output_mask) { if (mask & output_mask) {
checkerror = false; checkerror = false;
std::cerr << "[HSE] Next up: write event" << std::endl;
ptr->handle_event(operation::write); ptr->handle_event(operation::write);
} }
if (checkerror && (mask & error_mask)) { if (checkerror && (mask & error_mask)) {
...@@ -1253,6 +1266,12 @@ bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len, ...@@ -1253,6 +1266,12 @@ bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
sockaddr_storage& sa, size_t sa_len) { sockaddr_storage& sa, size_t sa_len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(buf_len)); CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(buf_len));
std::cerr << "[SD] Sending datagram ... " << std::endl; std::cerr << "[SD] Sending datagram ... " << std::endl;
{
std::string host;
uint16_t port;
std::tie(host,port) = sender_from_sockaddr(sa, sa_len);
std::cerr << "[SD] Addressing: " << host << ":" << port << std::endl;
}
auto sres = ::sendto(fd, reinterpret_cast<socket_send_ptr>(buf), buf_len, auto sres = ::sendto(fd, reinterpret_cast<socket_send_ptr>(buf), buf_len,
no_sigpipe_flag, reinterpret_cast<sockaddr*>(&sa), no_sigpipe_flag, reinterpret_cast<sockaddr*>(&sa),
sa_len); sa_len);
...@@ -1277,11 +1296,22 @@ bool receive_datagram(size_t& result, native_socket fd, void* buf, size_t len, ...@@ -1277,11 +1296,22 @@ bool receive_datagram(size_t& result, native_socket fd, void* buf, size_t len,
auto sres = ::recvfrom(fd, buf, len, no_sigpipe_flag, auto sres = ::recvfrom(fd, buf, len, no_sigpipe_flag,
reinterpret_cast<struct sockaddr *>(&sender_addr), reinterpret_cast<struct sockaddr *>(&sender_addr),
&sender_len); &sender_len);
if (is_error(sres, true) || sres == 0) { {
std::cerr << "[RD] Nothing to receive (res: " << sres << ")" << std::endl; std::string host;
uint16_t port;
std::tie(host,port) = sender_from_sockaddr(sender_addr, sender_len);
std::cerr << "[RD] recvfrom returned '" << sres << "' with senderaddr '"
<< host << ":" << port << "'" << std::endl;
}
// TODO: Check if sres > len and do some error handling ...
if (is_error(sres, true)) {
std::cerr << "[RD] error: " << sres << std::endl;
// Nothing to receive // Nothing to receive
return false; return false;
} }
if (sres == 0) {
std::cerr << "[RD] 0 length datagram is permitted" << std::endl;
}
result = (sres > 0) ? static_cast<size_t>(sres) : 0; result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return true; return true;
} }
...@@ -1654,27 +1684,35 @@ void dgram_communicator::handle_event(operation op) { ...@@ -1654,27 +1684,35 @@ void dgram_communicator::handle_event(operation op) {
// incoming message should signify a new remote enpoint // incoming message should signify a new remote enpoint
// --> create a new dgram_scribe // --> create a new dgram_scribe
size_t rb; size_t rb;
//rd_buf_.clear();
//rd_buf_.resize(dgram_size_);
if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(), if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(),
sockaddr_, sockaddr_len_)) { sockaddr_, sockaddr_len_)) {
reader_->io_failure(&backend(), operation::read); reader_->io_failure(&backend(), operation::read);
passivate(); passivate();
return; return;
} }
// currently handles the change from acceptor
// to communicator, TODO: Find a better solution
// Either keep sending to the same endpoint,
// Which would require some logic to determin if
// the sender is a new endpoint ...
if (waiting_for_remote_endpoint) { if (waiting_for_remote_endpoint) {
sender_from_sockaddr(sockaddr_, sockaddr_len_); std::tie(host_, port_) = sender_from_sockaddr(sockaddr_, sockaddr_len_);
remote_endpoint_addr_ = std::move(sockaddr_); remote_endpoint_addr_ = std::move(sockaddr_);
remote_endpoint_addr_len_ = sockaddr_len_; remote_endpoint_addr_len_ = sockaddr_len_;
sockaddr_len_ = 0; sockaddr_len_ = 0;
std::cerr << "[DC] new endpoint: " << host_ << ":" << port_ << std::endl;
} }
if (rb == 0) if (rb > 0) {
return;
auto res = reader_->consume(&backend(), rd_buf_.data(), rb); auto res = reader_->consume(&backend(), rd_buf_.data(), rb);
bytes_read_ = rb; bytes_read_ = rb;
prepare_next_read();
if (!res) { if (!res) {
passivate(); passivate();
return; return;
} }
}
prepare_next_read();
break; break;
} }
case operation::write: { case operation::write: {
...@@ -1722,32 +1760,6 @@ void dgram_communicator::prepare_next_read() { ...@@ -1722,32 +1760,6 @@ void dgram_communicator::prepare_next_read() {
rd_buf_.resize(dgram_size_); rd_buf_.resize(dgram_size_);
} }
void dgram_communicator::sender_from_sockaddr(const sockaddr_storage& sa,
size_t) {
char addr[INET6_ADDRSTRLEN];
switch(sa.ss_family) {
case AF_INET:
port_ = ntohs(reinterpret_cast<const sockaddr_in*>(&sa)->sin_port);
inet_ntop(AF_INET,
&reinterpret_cast<const sockaddr_in*>(&sa)->sin_addr,
addr, INET_ADDRSTRLEN);
host_.insert(std::begin(host_), std::begin(addr),
std::begin(addr) + INET_ADDRSTRLEN);
break;
case AF_INET6:
port_ = ntohs(reinterpret_cast<const sockaddr_in6*>(&sa)->sin6_port);
inet_ntop(AF_INET6,
&reinterpret_cast<const sockaddr_in*>(&sa)->sin_addr,
addr, INET6_ADDRSTRLEN);
host_.insert(std::begin(host_), std::begin(addr),
std::begin(addr) + INET6_ADDRSTRLEN);
break;
default:
// nop
break;
}
}
void dgram_communicator::set_endpoint_addr(const sockaddr_storage& sa, void dgram_communicator::set_endpoint_addr(const sockaddr_storage& sa,
size_t len, size_t len,
bool is_tmp_endpoint) { bool is_tmp_endpoint) {
...@@ -1800,16 +1812,21 @@ void dgram_acceptor::handle_event(operation op) { ...@@ -1800,16 +1812,21 @@ void dgram_acceptor::handle_event(operation op) {
case operation::read: { case operation::read: {
// incoming message should signify a new remote enpoint // incoming message should signify a new remote enpoint
// --> create a new dgram_scribe // --> create a new dgram_scribe
// TODO: some idenfification to prevent arbitrary
// message from creating endpoints?
rd_buf_.clear();
rd_buf_.resize(dgram_size_);
size_t rb; size_t rb;
memset(&sockaddr_,0, sockaddr_len_);
if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(), if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(),
sockaddr_, sockaddr_len_)) { sockaddr_, sockaddr_len_)) {
mgr_->io_failure(&backend(), operation::read); mgr_->io_failure(&backend(), operation::read);
passivate(); passivate();
return; return;
} }
sender_from_sockaddr(sockaddr_, sockaddr_len_); bytes_read_ = rb;
if (rb == 0) if (rb > 0) {
return; std::tie(host_,port_) = sender_from_sockaddr(sockaddr_, sockaddr_len_);
std::cerr << "[DA] received " << rb << " bytes data from " std::cerr << "[DA] received " << rb << " bytes data from "
<< host_ << ":" << port_ << std::endl; << host_ << ":" << port_ << std::endl;
auto res = mgr_->new_endpoint(rd_buf_.data(), rb); auto res = mgr_->new_endpoint(rd_buf_.data(), rb);
...@@ -1819,7 +1836,8 @@ void dgram_acceptor::handle_event(operation op) { ...@@ -1819,7 +1836,8 @@ void dgram_acceptor::handle_event(operation op) {
std::cerr << "[DA] Failure during creation of new udp endpoint" std::cerr << "[DA] Failure during creation of new udp endpoint"
<< std::endl;; << std::endl;;
} }
bytes_read_ = rb; }
prepare_next_read();
break; break;
} }
case operation::write: { case operation::write: {
...@@ -1835,29 +1853,6 @@ void dgram_acceptor::handle_event(operation op) { ...@@ -1835,29 +1853,6 @@ void dgram_acceptor::handle_event(operation op) {
} }
} }
void dgram_acceptor::sender_from_sockaddr(const sockaddr_storage& sa, size_t) {
char addr[INET6_ADDRSTRLEN];
switch(sa.ss_family) {
case AF_INET:
port_ = ntohs(reinterpret_cast<const sockaddr_in*>(&sa)->sin_port);
inet_ntop(AF_INET,
&reinterpret_cast<const sockaddr_in*>(&sa)->sin_addr,
addr, INET_ADDRSTRLEN);
host_ = addr;
break;
case AF_INET6:
port_ = ntohs(reinterpret_cast<const sockaddr_in6*>(&sa)->sin6_port);
inet_ntop(AF_INET6,
&reinterpret_cast<const sockaddr_in*>(&sa)->sin_addr,
addr, INET6_ADDRSTRLEN);
host_ = addr;
break;
default:
// nop
break;
}
}
std::pair<const sockaddr_storage&,size_t> dgram_acceptor::last_sender() { std::pair<const sockaddr_storage&,size_t> dgram_acceptor::last_sender() {
return std::make_pair(std::ref(sockaddr_), sockaddr_len_); return std::make_pair(std::ref(sockaddr_), sockaddr_len_);
} }
...@@ -1893,6 +1888,30 @@ private: ...@@ -1893,6 +1888,30 @@ private:
native_socket fd_; native_socket fd_;
}; };
std::tuple<std::string,uint16_t>
sender_from_sockaddr(const sockaddr_storage& sa, size_t) {
uint16_t port = 0;
char addr[INET6_ADDRSTRLEN];
switch(sa.ss_family) {
case AF_INET:
port = ntohs(reinterpret_cast<const sockaddr_in*>(&sa)->sin_port);
inet_ntop(AF_INET,
&reinterpret_cast<const sockaddr_in*>(&sa)->sin_addr,
addr, INET_ADDRSTRLEN);
break;
case AF_INET6:
port = ntohs(reinterpret_cast<const sockaddr_in6*>(&sa)->sin6_port);
inet_ntop(AF_INET6,
&reinterpret_cast<const sockaddr_in*>(&sa)->sin_addr,
addr, INET6_ADDRSTRLEN);
break;
default:
addr[0] = '\0';
break;
}
return std::make_tuple(std::string(addr),port);
}
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& { auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr; return what.sin_addr;
} }
......
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