Skip to content

control packet support in test::Proxy #802

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
GabrielPerezCSDev wants to merge 4 commits into
base: develop
Choose a base branch
from
Open

control packet support in test::Proxy #802

Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f2afaeb
Added optional control packet support to test::Proxy
GabrielPerez0522 Feb 12, 2025
32dae4c
Updated implementation for bi-directional control packets
GabrielPerez0522 Mar 6, 2025
afeeaf5
Updated tests for bi-directional control packets
GabrielPerez0522 Mar 6, 2025
01f2bb7
Finalized formatting for bi-directional control packet handling
GabrielPerez0522 Mar 6, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
144 changes: 141 additions & 3 deletions src/tests/public_api/test_helpers/proxy.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,8 +43,10 @@ class Proxy : public core::Thread, private packet::IWriter {
, net_loop_(packet_pool_, buffer_pool_, arena_)
, n_source_packets_(n_source_packets)
, n_repair_packets_(n_repair_packets)
, n_control_packets_(0)
, flags_(flags)
, pos_(0) {
input_control_endp_ = NULL;
LONGS_EQUAL(status::StatusOK, net_loop_.init_status());

roc_protocol source_proto;
Expand Down Expand Up @@ -119,9 +121,129 @@ class Proxy : public core::Thread, private packet::IWriter {
== 0);
}

Proxy(const roc_endpoint* receiver_source_endp,
const roc_endpoint* receiver_repair_endp,
const roc_endpoint* receiver_control_endp,
size_t n_source_packets,
size_t n_repair_packets,
size_t n_control_packets,
Comment thread
gavv marked this conversation as resolved.
Outdated
unsigned flags)
: packet_pool_("proxy_packet_pool", arena_)
, buffer_pool_("proxy_buffer_pool", arena_, 2000)
, queue_(packet::ConcurrentQueue::Blocking)
, net_loop_(packet_pool_, buffer_pool_, arena_)
Comment thread
gavv marked this conversation as resolved.
Outdated
, n_source_packets_(n_source_packets)
, n_repair_packets_(n_repair_packets)
, n_control_packets_(n_control_packets)
, flags_(flags)
, pos_(0) {
LONGS_EQUAL(status::StatusOK, net_loop_.init_status());

roc_protocol source_proto;
CHECK(roc_endpoint_get_protocol(receiver_source_endp, &source_proto) == 0);

roc_protocol repair_proto;
CHECK(roc_endpoint_get_protocol(receiver_repair_endp, &repair_proto) == 0);

roc_protocol control_proto;
CHECK(roc_endpoint_get_protocol(receiver_control_endp, &control_proto) == 0);

int source_port = 0;
CHECK(roc_endpoint_get_port(receiver_source_endp, &source_port) == 0);

int repair_port = 0;
CHECK(roc_endpoint_get_port(receiver_repair_endp, &repair_port) == 0);

int control_port = 0;
CHECK(roc_endpoint_get_port(receiver_control_endp, &control_port) == 0);

CHECK(receiver_source_endp_.set_host_port(address::Family_IPv4, "127.0.0.1",
source_port));
CHECK(receiver_repair_endp_.set_host_port(address::Family_IPv4, "127.0.0.1",
repair_port));

CHECK(receiver_control_endp_.set_host_port(address::Family_IPv4, "127.0.0.1",
control_port));

CHECK(send_config_.bind_address.set_host_port(address::Family_IPv4, "127.0.0.1",
0));

CHECK(recv_source_config_.bind_address.set_host_port(address::Family_IPv4,
"127.0.0.1", 0));
CHECK(recv_repair_config_.bind_address.set_host_port(address::Family_IPv4,
"127.0.0.1", 0));
CHECK(recv_control_config_.bind_address.set_host_port(address::Family_IPv4,
"127.0.0.1", 0));

netio::NetworkLoop::PortHandle send_port = NULL;

{
netio::NetworkLoop::Tasks::AddUdpPort add_task(send_config_);
CHECK(net_loop_.schedule_and_wait(add_task));

send_port = add_task.get_handle();
CHECK(send_port);

netio::NetworkLoop::Tasks::StartUdpSend send_task(add_task.get_handle());
CHECK(net_loop_.schedule_and_wait(send_task));
writer_ = &send_task.get_outbound_writer();
}

{
netio::NetworkLoop::Tasks::AddUdpPort add_task(recv_source_config_);
CHECK(net_loop_.schedule_and_wait(add_task));

netio::NetworkLoop::Tasks::StartUdpRecv recv_task(add_task.get_handle(),
*this);
CHECK(net_loop_.schedule_and_wait(recv_task));
}

{
netio::NetworkLoop::Tasks::AddUdpPort add_task(recv_repair_config_);
CHECK(net_loop_.schedule_and_wait(add_task));

netio::NetworkLoop::Tasks::StartUdpRecv recv_task(add_task.get_handle(),
*this);
CHECK(net_loop_.schedule_and_wait(recv_task));
}

{
netio::NetworkLoop::Tasks::AddUdpPort add_task(recv_control_config_);
CHECK(net_loop_.schedule_and_wait(add_task));

netio::NetworkLoop::Tasks::StartUdpRecv recv_task(add_task.get_handle(),
*this);
CHECK(net_loop_.schedule_and_wait(recv_task));
}

CHECK(roc_endpoint_allocate(&input_source_endp_) == 0);
CHECK(roc_endpoint_set_protocol(input_source_endp_, source_proto) == 0);
CHECK(roc_endpoint_set_host(input_source_endp_, "127.0.0.1") == 0);
CHECK(roc_endpoint_set_port(input_source_endp_,
recv_source_config_.bind_address.port())
== 0);

CHECK(roc_endpoint_allocate(&input_repair_endp_) == 0);
CHECK(roc_endpoint_set_protocol(input_repair_endp_, repair_proto) == 0);
CHECK(roc_endpoint_set_host(input_repair_endp_, "127.0.0.1") == 0);
CHECK(roc_endpoint_set_port(input_repair_endp_,
recv_repair_config_.bind_address.port())
== 0);

CHECK(roc_endpoint_allocate(&input_control_endp_) == 0);
CHECK(roc_endpoint_set_protocol(input_control_endp_, control_proto) == 0);
CHECK(roc_endpoint_set_host(input_control_endp_, "127.0.0.1") == 0);
CHECK(roc_endpoint_set_port(input_control_endp_,
recv_control_config_.bind_address.port())
== 0);
}

~Proxy() {
CHECK(roc_endpoint_deallocate(input_source_endp_) == 0);
CHECK(roc_endpoint_deallocate(input_repair_endp_) == 0);
if (input_control_endp_ != NULL) {
CHECK(roc_endpoint_deallocate(input_control_endp_) == 0);
}
}

const roc_endpoint* source_endpoint() const {
Expand All @@ -132,6 +254,10 @@ class Proxy : public core::Thread, private packet::IWriter {
return input_repair_endp_;
}

const roc_endpoint* control_endpoint() const {
return input_control_endp_;
}
Comment thread
gavv marked this conversation as resolved.

size_t n_dropped_packets() const {
return n_dropped_packets_;
}
Expand Down Expand Up @@ -180,26 +306,33 @@ class Proxy : public core::Thread, private packet::IWriter {
if (pp->udp()->dst_addr == recv_source_config_.bind_address) {
pp->udp()->dst_addr = receiver_source_endp_;
LONGS_EQUAL(status::StatusOK, source_queue_.write(pp));
} else if (pp->udp()->dst_addr == recv_control_config_.bind_address) {
pp->udp()->dst_addr = receiver_control_endp_;
LONGS_EQUAL(status::StatusOK, control_queue_.write(pp));
} else {
pp->udp()->dst_addr = receiver_repair_endp_;
LONGS_EQUAL(status::StatusOK, repair_queue_.write(pp));
}
Comment thread
gavv marked this conversation as resolved.
Comment on lines +228 to 236

@gavv gavv Jun 8, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm, have you tested this, could you describe your testing approach? Or if you haven't, please do.

So we need to proxy control packets in two directions: receiver[control_endpoint] -> proxy -> sender[control_endpoint] and sender[control_endpoint] -> proxy -> receiver[control_endpoint].

In both cases dst_addr is the same - it's proxy.

In the second direction, proxy receives control (feedback) packets from sender on the same address from which proxy sends control packets to sender. But in constructor, I don't see that we create any UDP port that is both sending and receiving?

I guess the easiest way is to make the port that is currently sending (the very first AddUdpPort task) to be also receiving, and match packets by source address. You can find example of setting up bi-directional port in node::Sender::configure (look for StartUdpSend and StartUdpRecv).


for (;;) {
const size_t block_pos = pos_ % (n_source_packets_ + n_repair_packets_);
const size_t block_pos =
pos_ % (n_source_packets_ + n_repair_packets_ + n_control_packets_);
Comment thread
gavv marked this conversation as resolved.
Outdated

if (block_pos < n_source_packets_) {
const bool drop_packet = (flags_ & FlagLoseSomePkts) && (block_pos == 1);

if (!send_packet_(source_queue_, drop_packet)) {
break;
}
} else {
} else if (block_pos < (n_source_packets_ + n_repair_packets_)) {
const bool drop_packet = (flags_ & FlagLoseAllRepairPkts);

if (!send_packet_(repair_queue_, drop_packet)) {
break;
}
} else {
if (!send_packet_(control_queue_, false)) {
break;
}
}
}

Expand Down Expand Up @@ -244,15 +377,19 @@ class Proxy : public core::Thread, private packet::IWriter {
netio::UdpConfig send_config_;
netio::UdpConfig recv_source_config_;
netio::UdpConfig recv_repair_config_;
netio::UdpConfig recv_control_config_;

roc_endpoint* input_source_endp_;
roc_endpoint* input_repair_endp_;
roc_endpoint* input_control_endp_;

address::SocketAddr receiver_source_endp_;
address::SocketAddr receiver_repair_endp_;
address::SocketAddr receiver_control_endp_;

packet::FifoQueue source_queue_;
packet::FifoQueue repair_queue_;
packet::FifoQueue control_queue_;

packet::ConcurrentQueue queue_;
packet::IWriter* writer_;
Expand All @@ -261,6 +398,7 @@ class Proxy : public core::Thread, private packet::IWriter {

const size_t n_source_packets_;
const size_t n_repair_packets_;
const size_t n_control_packets_;
core::Atomic<size_t> n_dropped_packets_;

const unsigned flags_;
Expand Down