Replace sorted_list/vector with callback_list doubly-linked list for timers and frame callbacks
This commit is contained in:
@@ -43,39 +43,25 @@ static void discover_peer(peer_callback on_found, fail_callback on_timeout) {
|
||||
ipv4::ip4_addr our_ip = ns.ip;
|
||||
|
||||
auto timer = std::make_shared<timer_handle>(nullptr);
|
||||
auto cb = std::make_shared<std::function<void(std::span<const uint8_t>)>>();
|
||||
*cb = [our_ip, timer, cb, on_found = std::move(on_found)](std::span<const uint8_t> frame) {
|
||||
auto cb_id = std::make_shared<frame_cb_handle>(nullptr);
|
||||
*cb_id = net_add_frame_callback([our_ip, timer, cb_id, on_found = std::move(on_found)](std::span<const uint8_t> frame) -> bool {
|
||||
parse_buffer pb(frame);
|
||||
auto* eth_hdr = pb.consume<eth::header>();
|
||||
if (!eth_hdr || eth_hdr->ethertype != eth::ETH_IPV4) {
|
||||
net_add_frame_callback(*cb);
|
||||
return;
|
||||
}
|
||||
if (!eth_hdr || eth_hdr->ethertype != eth::ETH_IPV4) return false;
|
||||
auto* ip = pb.consume<ipv4::header>();
|
||||
if (!ip || ip->protocol != 17) {
|
||||
net_add_frame_callback(*cb);
|
||||
return;
|
||||
}
|
||||
if (!ip || ip->protocol != 17) return false;
|
||||
size_t options_len = ip->header_len() - sizeof(ipv4::header);
|
||||
if (options_len > 0 && !pb.skip(options_len)) {
|
||||
net_add_frame_callback(*cb);
|
||||
return;
|
||||
}
|
||||
if (options_len > 0 && !pb.skip(options_len)) return false;
|
||||
auto* uhdr = pb.consume<udp::header>();
|
||||
if (!uhdr || uhdr->src_port != PICOMAP_PORT) {
|
||||
net_add_frame_callback(*cb);
|
||||
return;
|
||||
}
|
||||
if (ip->src == our_ip) {
|
||||
net_add_frame_callback(*cb);
|
||||
return;
|
||||
}
|
||||
if (!uhdr || uhdr->src_port != PICOMAP_PORT) return false;
|
||||
if (ip->src == our_ip) return false;
|
||||
dispatch_cancel_timer(*timer);
|
||||
on_found({eth_hdr->src, ip->src});
|
||||
};
|
||||
net_add_frame_callback(*cb);
|
||||
return true;
|
||||
});
|
||||
|
||||
*timer = dispatch_schedule_ms(5000, [on_timeout = std::move(on_timeout)]() {
|
||||
*timer = dispatch_schedule_ms(5000, [cb_id, on_timeout = std::move(on_timeout)]() {
|
||||
net_remove_frame_callback(*cb_id);
|
||||
on_timeout();
|
||||
});
|
||||
}
|
||||
@@ -88,23 +74,18 @@ static void test_discovery_igmp(const responder& resp) {
|
||||
net_send_raw(buf.span());
|
||||
|
||||
auto timer = std::make_shared<timer_handle>(nullptr);
|
||||
auto cb = std::make_shared<std::function<void(std::span<const uint8_t>)>>();
|
||||
*cb = [resp, timer, cb](std::span<const uint8_t> frame) {
|
||||
auto cb_id = std::make_shared<frame_cb_handle>(nullptr);
|
||||
*cb_id = net_add_frame_callback([resp, timer](std::span<const uint8_t> frame) -> bool {
|
||||
ipv4::ip4_addr group;
|
||||
if (!igmp::parse_report(frame, group)) {
|
||||
net_add_frame_callback(*cb);
|
||||
return;
|
||||
}
|
||||
if (group != igmp::PICOMAP_DISCOVERY_GROUP) {
|
||||
net_add_frame_callback(*cb);
|
||||
return;
|
||||
}
|
||||
if (!igmp::parse_report(frame, group)) return false;
|
||||
if (group != igmp::PICOMAP_DISCOVERY_GROUP) return false;
|
||||
dispatch_cancel_timer(*timer);
|
||||
resp.respond(ResponseTest{true, {"got IGMP report for " + ipv4::to_string(group)}});
|
||||
};
|
||||
net_add_frame_callback(*cb);
|
||||
return true;
|
||||
});
|
||||
|
||||
*timer = dispatch_schedule_ms(5000, [resp]() {
|
||||
*timer = dispatch_schedule_ms(5000, [cb_id, resp]() {
|
||||
net_remove_frame_callback(*cb_id);
|
||||
resp.respond(ResponseTest{false, {"no IGMP report within 5s"}});
|
||||
});
|
||||
}
|
||||
@@ -131,23 +112,21 @@ static void test_ping(const responder& resp, ipv4::ip4_addr dst_ip) {
|
||||
ipv4::ip4_addr our_ip = ns.ip;
|
||||
|
||||
auto timer = std::make_shared<timer_handle>(nullptr);
|
||||
auto cb = std::make_shared<std::function<void(std::span<const uint8_t>)>>();
|
||||
*cb = [resp, ping_id, our_ip, timer, cb](std::span<const uint8_t> frame) {
|
||||
auto cb_id = std::make_shared<frame_cb_handle>(nullptr);
|
||||
*cb_id = net_add_frame_callback([resp, ping_id, our_ip, timer](std::span<const uint8_t> frame) -> bool {
|
||||
ipv4::ip4_addr src_ip;
|
||||
if (!icmp::parse_echo_reply(frame, src_ip, ping_id)) {
|
||||
net_add_frame_callback(*cb);
|
||||
return;
|
||||
}
|
||||
if (!icmp::parse_echo_reply(frame, src_ip, ping_id)) return false;
|
||||
dispatch_cancel_timer(*timer);
|
||||
if (src_ip == our_ip) {
|
||||
resp.respond(ResponseTest{false, {"got reply from self: " + ipv4::to_string(src_ip)}});
|
||||
return;
|
||||
return true;
|
||||
}
|
||||
resp.respond(ResponseTest{true, {"reply from " + ipv4::to_string(src_ip)}});
|
||||
};
|
||||
net_add_frame_callback(*cb);
|
||||
return true;
|
||||
});
|
||||
|
||||
*timer = dispatch_schedule_ms(5000, [resp]() {
|
||||
*timer = dispatch_schedule_ms(5000, [cb_id, resp]() {
|
||||
net_remove_frame_callback(*cb_id);
|
||||
resp.respond(ResponseTest{false, {"no reply from non-self host within 5s"}});
|
||||
});
|
||||
}
|
||||
@@ -163,6 +142,7 @@ static void test_ping_global(const responder& resp) {
|
||||
struct ping_rate_state {
|
||||
responder resp;
|
||||
timer_handle timer = nullptr;
|
||||
frame_cb_handle cb_handle = nullptr;
|
||||
uint16_t ping_id;
|
||||
uint16_t sent = 0;
|
||||
uint16_t received = 0;
|
||||
@@ -190,44 +170,6 @@ static void ping_rate_send_one(std::shared_ptr<ping_rate_state> st) {
|
||||
st->sent++;
|
||||
}
|
||||
|
||||
static void ping_rate_recv(std::shared_ptr<ping_rate_state> st,
|
||||
std::shared_ptr<std::function<void(std::span<const uint8_t>)>> cb,
|
||||
std::span<const uint8_t> frame) {
|
||||
ipv4::ip4_addr src_ip;
|
||||
if (!icmp::parse_echo_reply(frame, src_ip, st->ping_id)) {
|
||||
net_add_frame_callback(*cb);
|
||||
return;
|
||||
}
|
||||
if (src_ip == st->our_ip) {
|
||||
net_add_frame_callback(*cb);
|
||||
return;
|
||||
}
|
||||
|
||||
st->received++;
|
||||
if (st->received >= st->target) {
|
||||
dispatch_cancel_timer(st->timer);
|
||||
uint32_t elapsed_us = time_us_32() - st->start_us;
|
||||
uint32_t elapsed_ms = elapsed_us / 1000;
|
||||
uint32_t pps = static_cast<uint32_t>(
|
||||
static_cast<uint64_t>(st->received) * 1000000 / elapsed_us);
|
||||
uint64_t total_bytes = static_cast<uint64_t>(st->received) * 2 * st->frame_size();
|
||||
uint32_t kbps = static_cast<uint32_t>(total_bytes * 1000 / elapsed_us);
|
||||
char msg[128];
|
||||
snprintf(msg, sizeof(msg),
|
||||
"%u rt in %lu ms, %lu pps, %lu bytes, %lu KB/s",
|
||||
st->received, static_cast<unsigned long>(elapsed_ms),
|
||||
static_cast<unsigned long>(pps),
|
||||
static_cast<unsigned long>(total_bytes),
|
||||
static_cast<unsigned long>(kbps));
|
||||
st->resp.respond(ResponseTest{true, {msg}});
|
||||
return;
|
||||
}
|
||||
|
||||
if (st->sent < st->target)
|
||||
ping_rate_send_one(st);
|
||||
net_add_frame_callback(*cb);
|
||||
}
|
||||
|
||||
static void start_ping_rate(const responder& resp, uint16_t target,
|
||||
uint16_t payload_len, uint16_t pipeline) {
|
||||
auto& ns = net_get_state();
|
||||
@@ -245,16 +187,41 @@ static void start_ping_rate(const responder& resp, uint16_t target,
|
||||
st->peer = peer;
|
||||
st->start_us = time_us_32();
|
||||
|
||||
auto cb = std::make_shared<std::function<void(std::span<const uint8_t>)>>();
|
||||
*cb = [st, cb](std::span<const uint8_t> frame) {
|
||||
ping_rate_recv(st, cb, frame);
|
||||
};
|
||||
|
||||
for (uint16_t i = 0; i < st->pipeline && st->sent < st->target; i++)
|
||||
ping_rate_send_one(st);
|
||||
net_add_frame_callback(*cb);
|
||||
|
||||
st->cb_handle = net_add_frame_callback([st](std::span<const uint8_t> frame) -> bool {
|
||||
ipv4::ip4_addr src_ip;
|
||||
if (!icmp::parse_echo_reply(frame, src_ip, st->ping_id)) return false;
|
||||
if (src_ip == st->our_ip) return false;
|
||||
|
||||
st->received++;
|
||||
if (st->received >= st->target) {
|
||||
dispatch_cancel_timer(st->timer);
|
||||
uint32_t elapsed_us = time_us_32() - st->start_us;
|
||||
uint32_t elapsed_ms = elapsed_us / 1000;
|
||||
uint32_t pps = static_cast<uint32_t>(
|
||||
static_cast<uint64_t>(st->received) * 1000000 / elapsed_us);
|
||||
uint64_t total_bytes = static_cast<uint64_t>(st->received) * 2 * st->frame_size();
|
||||
uint32_t kbps = static_cast<uint32_t>(total_bytes * 1000 / elapsed_us);
|
||||
char msg[128];
|
||||
snprintf(msg, sizeof(msg),
|
||||
"%u rt in %lu ms, %lu pps, %lu bytes, %lu KB/s",
|
||||
st->received, static_cast<unsigned long>(elapsed_ms),
|
||||
static_cast<unsigned long>(pps),
|
||||
static_cast<unsigned long>(total_bytes),
|
||||
static_cast<unsigned long>(kbps));
|
||||
st->resp.respond(ResponseTest{true, {msg}});
|
||||
return true;
|
||||
}
|
||||
|
||||
if (st->sent < st->target)
|
||||
ping_rate_send_one(st);
|
||||
return false;
|
||||
});
|
||||
|
||||
st->timer = dispatch_schedule_ms(10000, [st]() {
|
||||
net_remove_frame_callback(st->cb_handle);
|
||||
uint32_t elapsed_us = time_us_32() - st->start_us;
|
||||
char msg[64];
|
||||
snprintf(msg, sizeof(msg), "timeout after %u/%u rt in %lu ms",
|
||||
@@ -269,7 +236,7 @@ static void start_ping_rate(const responder& resp, uint16_t target,
|
||||
}
|
||||
|
||||
static void test_ping_rate(const responder& resp) {
|
||||
start_ping_rate(resp, 8192, 0, 1);
|
||||
start_ping_rate(resp, 8192, 0, 2);
|
||||
}
|
||||
|
||||
static void test_ping_rate_1k(const responder& resp) {
|
||||
|
||||
Reference in New Issue
Block a user