diff --git a/CMakeLists.txt b/CMakeLists.txt index 55469f8..67395e9 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -50,6 +50,7 @@ build_lib( utils/format-utils.cc utils/switch-api.cc utils/p4-queue.cc + utils/p4-traffic-manager.cc utils/fattree-topo-helper.cc model/switched-ethernet-channel.cc model/eth-net-device.cc @@ -64,6 +65,7 @@ build_lib( helper/build-flowtable-helper.cc HEADER_FILES # equivalent to headers.source utils/p4-queue.h + utils/p4-traffic-manager.h utils/format-utils.h utils/switch-api.h utils/register-access-v1model.h @@ -89,6 +91,7 @@ build_lib( ${third_party_libs} TEST_SOURCES # equivalent to module_test.source test/p4-switch-queue-item-test-suite.cc + test/p4-traffic-manager-test-suite.cc # test/p4-controller-test-suite.cc # test/p4sim-test-suite.cc # test/format-utils-test-suite.cc diff --git a/test/p4-traffic-manager-test-suite.cc b/test/p4-traffic-manager-test-suite.cc new file mode 100644 index 0000000..ba8e58d --- /dev/null +++ b/test/p4-traffic-manager-test-suite.cc @@ -0,0 +1,547 @@ +/* + * Copyright (c) 2025 TU Dresden + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 2 as + * published by the Free Software Foundation; + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * + * Authors: Vineet Goel + */ + +#include "ns3/boolean.h" +#include "ns3/data-rate.h" +#include "ns3/nstime.h" +#include "ns3/p4-traffic-manager.h" +#include "ns3/simulator.h" +#include "ns3/test.h" +#include "ns3/uinteger.h" + +#include +#include +#include + +using namespace ns3; + +namespace +{ + +/// Lightweight test payload so the TM can be exercised without a bm::Packet. +class DummyPayload : public TmPayload +{ + public: + explicit DummyPayload(uint32_t tag) + : m_tag(tag) + { + } + + uint32_t GetTag() const + { + return m_tag; + } + + private: + uint32_t m_tag; +}; + +/// Convenience: make a payload and enqueue it into the VOQ. +bool +Enq(Ptr tm, + uint32_t size, + uint32_t in, + uint32_t out, + uint8_t prio, + uint32_t tag = 0) +{ + return tm->EnqueueToVoq(std::make_unique(tag), size, in, out, prio); +} + +Ptr +MakeTm(uint32_t numPorts) +{ + Ptr tm = CreateObject(); + tm->SetNumPorts(numPorts); + return tm; +} + +} // namespace + +// --------------------------------------------------------------------------- +// 1. Basic enqueue / dequeue correctness + VOQ sizing +// --------------------------------------------------------------------------- +class TmBasicEnqueueDequeueTest : public TestCase +{ + public: + TmBasicEnqueueDequeueTest() + : TestCase("TM: basic enqueue/dequeue and accounting") + { + } + + void DoRun() override + { + Ptr tm = MakeTm(4); + + NS_TEST_ASSERT_MSG_EQ(tm->GetNumPorts(), 4, "NumPorts should be 4"); + NS_TEST_ASSERT_MSG_EQ(tm->GlobalBufferBytes(), 0, "global bytes start at 0"); + + NS_TEST_ASSERT_MSG_EQ(Enq(tm, 100, 0, 1, 3, 42), true, "enqueue should succeed"); + NS_TEST_ASSERT_MSG_EQ(tm->VoqLength(0, 1, 3), 1, "VOQ[0][1][3] len == 1"); + NS_TEST_ASSERT_MSG_EQ(tm->VoqBytes(0, 1, 3), 100, "VOQ bytes == 100"); + NS_TEST_ASSERT_MSG_EQ(tm->InputBufferBytes(0), 100, "input bytes == 100"); + NS_TEST_ASSERT_MSG_EQ(tm->GlobalBufferBytes(), 100, "global bytes == 100"); + + // Second packet, same VOQ. + NS_TEST_ASSERT_MSG_EQ(Enq(tm, 200, 0, 1, 3, 43), true, "enqueue 2 should succeed"); + NS_TEST_ASSERT_MSG_EQ(tm->VoqLength(0, 1, 3), 2, "VOQ len == 2"); + NS_TEST_ASSERT_MSG_EQ(tm->GlobalBufferBytes(), 300, "global bytes == 300"); + + // FIFO order within a VOQ: first out is tag 42. + TmItem item; + NS_TEST_ASSERT_MSG_EQ(tm->DequeueFromVoq(0, 1, 3, item), true, "dequeue should succeed"); + auto* p = dynamic_cast(item.payload.get()); + NS_TEST_ASSERT_MSG_NE(p, nullptr, "payload must survive as DummyPayload"); + NS_TEST_ASSERT_MSG_EQ(p->GetTag(), 42, "FIFO: first dequeued tag == 42"); + NS_TEST_ASSERT_MSG_EQ(tm->VoqBytes(0, 1, 3), 200, "VOQ bytes back to 200"); + NS_TEST_ASSERT_MSG_EQ(tm->GlobalBufferBytes(), 200, "global bytes == 200"); + + // Dequeue empty VOQ returns false. + NS_TEST_ASSERT_MSG_EQ(tm->DequeueFromVoq(2, 2, 0, item), false, "empty VOQ dequeue false"); + + const auto& s = tm->GetStats(); + NS_TEST_ASSERT_MSG_EQ(s.totalReceived, 2, "received == 2"); + NS_TEST_ASSERT_MSG_EQ(s.totalVoqEnqueued, 2, "enqueued == 2"); + NS_TEST_ASSERT_MSG_EQ(s.totalMovedToEgress, 1, "moved to egress == 1"); + NS_TEST_ASSERT_MSG_EQ(s.totalDropped, 0, "no drops"); + + Simulator::Destroy(); + } +}; + +// --------------------------------------------------------------------------- +// 2. Priority scheduling correctness (fabric picks highest priority first) +// --------------------------------------------------------------------------- +class TmPrioritySchedulingTest : public TestCase +{ + public: + TmPrioritySchedulingTest() + : TestCase("TM: fabric serves higher priority first") + { + } + + void DoRun() override + { + Ptr tm = MakeTm(2); + + // Same input->output, different priorities: 2 and 5. + Enq(tm, 100, 0, 1, 2, 1); + Enq(tm, 100, 0, 1, 5, 2); + + auto grants = tm->RunFabricScheduler(); + NS_TEST_ASSERT_MSG_EQ(grants.size(), 1, "one input -> one grant per round"); + NS_TEST_ASSERT_MSG_EQ(static_cast(grants[0].priority), 5, + "priority 5 must be served before 2"); + NS_TEST_ASSERT_MSG_EQ(grants[0].inPort, 0, "granted input 0"); + NS_TEST_ASSERT_MSG_EQ(grants[0].outPort, 1, "granted output 1"); + + Simulator::Destroy(); + } +}; + +// --------------------------------------------------------------------------- +// 3. VOQ correctness: two inputs target the same output +// --------------------------------------------------------------------------- +class TmVoqSameOutputTest : public TestCase +{ + public: + TmVoqSameOutputTest() + : TestCase("TM: two inputs to same output are isolated VOQs") + { + } + + void DoRun() override + { + Ptr tm = MakeTm(3); + + // input 0 and input 2 both target output 1, same priority. + Enq(tm, 100, 0, 1, 4, 10); + Enq(tm, 100, 2, 1, 4, 20); + + // Distinct VOQs. + NS_TEST_ASSERT_MSG_EQ(tm->VoqLength(0, 1, 4), 1, "VOQ[0][1][4] len 1"); + NS_TEST_ASSERT_MSG_EQ(tm->VoqLength(2, 1, 4), 1, "VOQ[2][1][4] len 1"); + NS_TEST_ASSERT_MSG_EQ(tm->InputBufferBytes(0), 100, "input 0 bytes"); + NS_TEST_ASSERT_MSG_EQ(tm->InputBufferBytes(2), 100, "input 2 bytes"); + + // Output contention: only ONE of them can be granted this round + // (one output receives at most one packet per round). + auto grants = tm->RunFabricScheduler(); + NS_TEST_ASSERT_MSG_EQ(grants.size(), 1, "output 1 contended -> single grant"); + NS_TEST_ASSERT_MSG_EQ(grants[0].outPort, 1, "grant is for output 1"); + + Simulator::Destroy(); + } +}; + +// --------------------------------------------------------------------------- +// 4. Fabric matching rules: +// - one input sends at most one packet per round +// - one output receives at most one packet per round +// - a full permutation yields a maximal matching +// --------------------------------------------------------------------------- +class TmFabricMatchingTest : public TestCase +{ + public: + TmFabricMatchingTest() + : TestCase("TM: fabric one-in/one-out matching constraints") + { + } + + void DoRun() override + { + Ptr tm = MakeTm(4); + + // A perfect permutation demand: in i -> out (i+1)%4, all same priority. + for (uint32_t i = 0; i < 4; ++i) + { + Enq(tm, 100, i, (i + 1) % 4, 3, i); + } + + auto grants = tm->RunFabricScheduler(); + NS_TEST_ASSERT_MSG_EQ(grants.size(), 4, "permutation -> 4 grants (maximal)"); + + std::vector inSeen(4, false), outSeen(4, false); + for (const auto& g : grants) + { + NS_TEST_ASSERT_MSG_EQ(inSeen[g.inPort], false, "each input granted at most once"); + NS_TEST_ASSERT_MSG_EQ(outSeen[g.outPort], false, "each output granted at most once"); + inSeen[g.inPort] = true; + outSeen[g.outPort] = true; + } + + // Now a contention case: inputs 0,1,2 all want output 0. + Ptr tm2 = MakeTm(4); + Enq(tm2, 100, 0, 0, 3, 0); + Enq(tm2, 100, 1, 0, 3, 1); + Enq(tm2, 100, 2, 0, 3, 2); + auto g2 = tm2->RunFabricScheduler(); + NS_TEST_ASSERT_MSG_EQ(g2.size(), 1, "3 inputs -> 1 output => single grant"); + NS_TEST_ASSERT_MSG_EQ(g2[0].outPort, 0, "grant for output 0"); + // Priority-first, then input order: input 0 wins the tie. + NS_TEST_ASSERT_MSG_EQ(g2[0].inPort, 0, "input 0 wins tie (lowest index first)"); + + Simulator::Destroy(); + } +}; + +// --------------------------------------------------------------------------- +// 5. Buffer overflow and drop-reason correctness +// --------------------------------------------------------------------------- +class TmBufferDropTest : public TestCase +{ + public: + TmBufferDropTest() + : TestCase("TM: finite buffers produce correct drop reasons") + { + } + + void DoRun() override + { + // ---- VOQ limit ---- + { + Ptr tm = MakeTm(2); + tm->SetAttribute("VoqLimit", UintegerValue(150)); + + uint8_t lastReason = 255; + tm->TraceConnectWithoutContext( + "Drop", + MakeCallback(&TmBufferDropTest::OnDrop, this)); + m_lastReason = &lastReason; + + NS_TEST_ASSERT_MSG_EQ(Enq(tm, 100, 0, 1, 0), true, "first fits under VoqLimit"); + NS_TEST_ASSERT_MSG_EQ(Enq(tm, 100, 0, 1, 0), false, "second exceeds VoqLimit -> drop"); + NS_TEST_ASSERT_MSG_EQ(lastReason, + static_cast(TmDropReason::VOQ_QUEUE_FULL), + "drop reason == VOQ_QUEUE_FULL"); + NS_TEST_ASSERT_MSG_EQ(tm->VoqLength(0, 1, 0), 1, "only one packet accepted"); + const auto& s = tm->GetStats(); + NS_TEST_ASSERT_MSG_EQ(s.totalDropped, 1, "one drop counted"); + NS_TEST_ASSERT_MSG_EQ( + s.dropsByReason[static_cast(TmDropReason::VOQ_QUEUE_FULL)], + 1, + "VOQ_QUEUE_FULL counter == 1"); + m_lastReason = nullptr; + } + + // ---- Input buffer limit (independent of a single VOQ) ---- + { + Ptr tm = MakeTm(3); + tm->SetAttribute("InputBufferLimit", UintegerValue(150)); + uint8_t lastReason = 255; + tm->TraceConnectWithoutContext( + "Drop", + MakeCallback(&TmBufferDropTest::OnDrop, this)); + m_lastReason = &lastReason; + + // Two different VOQs of the same input; second overflows input budget. + NS_TEST_ASSERT_MSG_EQ(Enq(tm, 100, 0, 1, 0), true, "input budget ok"); + NS_TEST_ASSERT_MSG_EQ(Enq(tm, 100, 0, 2, 5), false, "input budget exceeded -> drop"); + NS_TEST_ASSERT_MSG_EQ(lastReason, + static_cast(TmDropReason::VOQ_INPUT_BUFFER_FULL), + "drop reason == VOQ_INPUT_BUFFER_FULL"); + m_lastReason = nullptr; + } + + // ---- Global buffer limit ---- + { + Ptr tm = MakeTm(3); + tm->SetAttribute("GlobalBufferLimit", UintegerValue(150)); + uint8_t lastReason = 255; + tm->TraceConnectWithoutContext( + "Drop", + MakeCallback(&TmBufferDropTest::OnDrop, this)); + m_lastReason = &lastReason; + + // Different inputs, different VOQs; global budget is the binding limit. + NS_TEST_ASSERT_MSG_EQ(Enq(tm, 100, 0, 1, 0), true, "global budget ok"); + NS_TEST_ASSERT_MSG_EQ(Enq(tm, 100, 1, 2, 0), false, "global budget exceeded -> drop"); + NS_TEST_ASSERT_MSG_EQ(lastReason, + static_cast(TmDropReason::VOQ_GLOBAL_BUFFER_FULL), + "drop reason == VOQ_GLOBAL_BUFFER_FULL"); + m_lastReason = nullptr; + } + + Simulator::Destroy(); + } + + private: + void OnDrop(uint8_t reason, uint32_t, uint32_t, uint8_t, uint32_t) + { + if (m_lastReason) + { + *m_lastReason = reason; + } + } + + uint8_t* m_lastReason{nullptr}; +}; + +// --------------------------------------------------------------------------- +// 6. Delay measurement correctness (VOQ waiting delay via simulated time) +// --------------------------------------------------------------------------- +class TmDelayMeasurementTest : public TestCase +{ + public: + TmDelayMeasurementTest() + : TestCase("TM: VOQ waiting delay is measured in simulated time") + { + } + + void DoRun() override + { + Ptr tm = MakeTm(2); + tm->TraceConnectWithoutContext("VoqWaitingDelay", + MakeCallback(&TmDelayMeasurementTest::OnDelay, this)); + + // Enqueue at t=0, dequeue at t=500ns. + Simulator::Schedule(Time(0), [tm]() { Enq(tm, 100, 0, 1, 3, 7); }); + Simulator::Schedule(NanoSeconds(500), [tm]() { + TmItem item; + tm->DequeueFromVoq(0, 1, 3, item); + }); + + Simulator::Run(); + + NS_TEST_ASSERT_MSG_EQ(m_lastDelay, NanoSeconds(500), "VOQ waiting delay == 500ns"); + const auto& s = tm->GetStats(); + NS_TEST_ASSERT_MSG_EQ(s.AvgVoqDelay(), NanoSeconds(500), "avg VOQ delay == 500ns"); + NS_TEST_ASSERT_MSG_EQ(s.maxQueueingDelay, NanoSeconds(500), "max queueing delay == 500ns"); + + Simulator::Destroy(); + } + + private: + void OnDelay(Time d) + { + m_lastDelay = d; + } + + Time m_lastDelay; +}; + +// --------------------------------------------------------------------------- +// 7. Event-driven end-to-end drain (P3 fabric loop + P4 egress scheduler) +// --------------------------------------------------------------------------- +class TmEventDrivenDrainTest : public TestCase +{ + public: + TmEventDrivenDrainTest() + : TestCase("TM: event-driven fabric+egress drains all packets to the wire") + { + } + + void DoRun() override + { + Ptr tm = MakeTm(4); + tm->SetAttribute("EventDriven", BooleanValue(true)); + tm->SetAttribute("PortRate", DataRateValue(DataRate("1Gbps"))); + tm->SetAttribute("FabricRate", DataRateValue(DataRate("10Gbps"))); + + uint32_t delivered = 0; + tm->SetTransmitCallback( + [&delivered](uint32_t, uint8_t, std::unique_ptr) { delivered++; }); + + // A perfect permutation: in i -> out (i+1)%4, so no egress contention. + for (uint32_t i = 0; i < 4; ++i) + { + Enq(tm, 100, i, (i + 1) % 4, 3, i); + } + + Simulator::Run(); + + NS_TEST_ASSERT_MSG_EQ(delivered, 4, "all 4 packets delivered via TransmitCallback"); + NS_TEST_ASSERT_MSG_EQ(tm->GlobalBufferBytes(), 0, "buffers empty after drain"); + for (uint32_t out = 0; out < 4; ++out) + { + NS_TEST_ASSERT_MSG_EQ(tm->EgressPortBytes(out), 0, "egress port drained"); + } + + const auto& s = tm->GetStats(); + NS_TEST_ASSERT_MSG_EQ(s.totalVoqEnqueued, 4, "4 enqueued"); + NS_TEST_ASSERT_MSG_EQ(s.totalMovedToEgress, 4, "4 moved to egress"); + NS_TEST_ASSERT_MSG_EQ(s.totalEgressEnqueued, 4, "4 accepted at egress"); + NS_TEST_ASSERT_MSG_EQ(s.totalTransmitted, 4, "4 transmitted"); + NS_TEST_ASSERT_MSG_EQ(s.totalDropped, 0, "no drops"); + + Simulator::Destroy(); + } +}; + +// --------------------------------------------------------------------------- +// 8. Egress strict-priority: a packet already in service is not preempted, +// but among waiting packets the highest priority goes next. +// --------------------------------------------------------------------------- +class TmEgressStrictPriorityTest : public TestCase +{ + public: + TmEgressStrictPriorityTest() + : TestCase("TM: egress scheduler serves waiting packets in strict priority") + { + } + + void DoRun() override + { + Ptr tm = MakeTm(4); + tm->SetAttribute("EventDriven", BooleanValue(true)); + // Slow port so the first packet is still transmitting when the others + // pile up behind it in the egress queue of output 0. + tm->SetAttribute("PortRate", DataRateValue(DataRate("10Mbps"))); + tm->SetAttribute("FabricRate", DataRateValue(DataRate("100Gbps"))); + + std::vector order; + tm->SetTransmitCallback( + [&order](uint32_t, uint8_t prio, std::unique_ptr) { order.push_back(prio); }); + + // All target output 0 (distinct inputs so the fabric can move them). + // Low-priority packet enters service first; higher priorities queue. + Simulator::Schedule(Time(0), [tm]() { Enq(tm, 100, 0, 0, 2, 2); }); + Simulator::Schedule(MicroSeconds(10), [tm]() { Enq(tm, 100, 1, 0, 5, 5); }); + Simulator::Schedule(MicroSeconds(20), [tm]() { Enq(tm, 100, 2, 0, 7, 7); }); + + Simulator::Run(); + + NS_TEST_ASSERT_MSG_EQ(order.size(), 3, "all three packets transmitted"); + // First to arrive (prio 2) is already on the wire; then strict priority. + NS_TEST_ASSERT_MSG_EQ(static_cast(order[0]), 2, "prio 2 in service first"); + NS_TEST_ASSERT_MSG_EQ(static_cast(order[1]), 7, "prio 7 next (highest waiting)"); + NS_TEST_ASSERT_MSG_EQ(static_cast(order[2]), 5, "prio 5 last"); + + Simulator::Destroy(); + } +}; + +// --------------------------------------------------------------------------- +// 9. Egress buffer overflow produces EGRESS_QUEUE_FULL +// --------------------------------------------------------------------------- +class TmEgressDropTest : public TestCase +{ + public: + TmEgressDropTest() + : TestCase("TM: full egress queue drops with EGRESS_QUEUE_FULL") + { + } + + void DoRun() override + { + Ptr tm = MakeTm(4); + tm->SetAttribute("EventDriven", BooleanValue(true)); + // Very slow port so egress never drains during the test; fast fabric. + tm->SetAttribute("PortRate", DataRateValue(DataRate("1Kbps"))); + tm->SetAttribute("FabricRate", DataRateValue(DataRate("100Gbps"))); + // Room for one 100B packet waiting behind the one in service. + tm->SetAttribute("EgressQueueLimit", UintegerValue(150)); + + tm->TraceConnectWithoutContext("Drop", MakeCallback(&TmEgressDropTest::OnDrop, this)); + + // Three packets, distinct inputs, same output+priority. The fabric moves + // them one per round: #1 enters service, #2 waits (100B <= 150), + // #3 overflows the egress queue (200B > 150) -> EGRESS_QUEUE_FULL. + Enq(tm, 100, 0, 0, 3, 0); + Enq(tm, 100, 1, 0, 3, 1); + Enq(tm, 100, 2, 0, 3, 2); + + Simulator::Run(); + + NS_TEST_ASSERT_MSG_EQ(m_egressDrops, 1, "exactly one egress drop"); + const auto& s = tm->GetStats(); + NS_TEST_ASSERT_MSG_EQ( + s.dropsByReason[static_cast(TmDropReason::EGRESS_QUEUE_FULL)], + 1, + "EGRESS_QUEUE_FULL counter == 1"); + NS_TEST_ASSERT_MSG_EQ(s.totalMovedToEgress, 3, "all three granted out of the VOQ"); + NS_TEST_ASSERT_MSG_EQ(s.totalEgressEnqueued, 2, "only two accepted into egress"); + + Simulator::Destroy(); + } + + private: + void OnDrop(uint8_t reason, uint32_t, uint32_t, uint8_t, uint32_t) + { + if (reason == static_cast(TmDropReason::EGRESS_QUEUE_FULL)) + { + m_egressDrops++; + } + } + + uint32_t m_egressDrops{0}; +}; + +// --------------------------------------------------------------------------- +// Suite +// --------------------------------------------------------------------------- +class P4TrafficManagerTestSuite : public TestSuite +{ + public: + P4TrafficManagerTestSuite() + : TestSuite("p4-traffic-manager", Type::UNIT) + { + AddTestCase(new TmBasicEnqueueDequeueTest, TestCase::QUICK); + AddTestCase(new TmPrioritySchedulingTest, TestCase::QUICK); + AddTestCase(new TmVoqSameOutputTest, TestCase::QUICK); + AddTestCase(new TmFabricMatchingTest, TestCase::QUICK); + AddTestCase(new TmBufferDropTest, TestCase::QUICK); + AddTestCase(new TmDelayMeasurementTest, TestCase::QUICK); + AddTestCase(new TmEventDrivenDrainTest, TestCase::QUICK); + AddTestCase(new TmEgressStrictPriorityTest, TestCase::QUICK); + AddTestCase(new TmEgressDropTest, TestCase::QUICK); + } +}; + +static P4TrafficManagerTestSuite g_p4TrafficManagerTestSuite; diff --git a/utils/p4-traffic-manager.cc b/utils/p4-traffic-manager.cc new file mode 100644 index 0000000..392b760 --- /dev/null +++ b/utils/p4-traffic-manager.cc @@ -0,0 +1,744 @@ +/* + * Copyright (c) 2025 TU Dresden + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 2 as + * published by the Free Software Foundation; + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * + * Authors: Vineet Goel + */ + +#include "p4-traffic-manager.h" + +#include "ns3/abort.h" +#include "ns3/boolean.h" +#include "ns3/log.h" +#include "ns3/simulator.h" +#include "ns3/uinteger.h" + +#include + +namespace ns3 +{ + +NS_LOG_COMPONENT_DEFINE("P4TrafficManager"); + +NS_OBJECT_ENSURE_REGISTERED(P4TrafficManager); + +const char* +TmDropReasonToString(TmDropReason r) +{ + switch (r) + { + case TmDropReason::VOQ_GLOBAL_BUFFER_FULL: + return "VOQ_GLOBAL_BUFFER_FULL"; + case TmDropReason::VOQ_INPUT_BUFFER_FULL: + return "VOQ_INPUT_BUFFER_FULL"; + case TmDropReason::VOQ_QUEUE_FULL: + return "VOQ_QUEUE_FULL"; + case TmDropReason::EGRESS_PORT_BUFFER_FULL: + return "EGRESS_PORT_BUFFER_FULL"; + case TmDropReason::EGRESS_QUEUE_FULL: + return "EGRESS_QUEUE_FULL"; + } + return "UNKNOWN"; +} + +// --------------------------------------------------------------------------- +// TmStats helpers +// --------------------------------------------------------------------------- + +Time +P4TrafficManager::TmStats::AvgVoqDelay() const +{ + return (cntVoqDelay == 0) ? Time(0) : (sumVoqDelay / static_cast(cntVoqDelay)); +} + +Time +P4TrafficManager::TmStats::AvgEgressDelay() const +{ + return (cntEgressDelay == 0) ? Time(0) : (sumEgressDelay / static_cast(cntEgressDelay)); +} + +Time +P4TrafficManager::TmStats::AvgTotalDelay() const +{ + return (cntTotalDelay == 0) ? Time(0) : (sumTotalDelay / static_cast(cntTotalDelay)); +} + +// --------------------------------------------------------------------------- +// TypeId / construction +// --------------------------------------------------------------------------- + +TypeId +P4TrafficManager::GetTypeId() +{ + static TypeId tid = + TypeId("ns3::P4TrafficManager") + .SetParent() + .SetGroupName("P4sim") + .AddConstructor() + .AddAttribute("NumPorts", + "Number of switch ports N (allocates N*N*8 VOQs).", + UintegerValue(0), + MakeUintegerAccessor(&P4TrafficManager::SetNumPorts, + &P4TrafficManager::GetNumPorts), + MakeUintegerChecker()) + .AddAttribute("GlobalBufferLimit", + "Global buffer limit in bytes across VOQ and egress (0 = unlimited).", + UintegerValue(0), + MakeUintegerAccessor(&P4TrafficManager::m_globalBufferLimit), + MakeUintegerChecker()) + .AddAttribute("InputBufferLimit", + "Per-input-port buffer limit in bytes (0 = unlimited).", + UintegerValue(0), + MakeUintegerAccessor(&P4TrafficManager::m_inputBufferLimit), + MakeUintegerChecker()) + .AddAttribute("VoqLimit", + "Per-VOQ[in][out][prio] limit in bytes (0 = unlimited).", + UintegerValue(0), + MakeUintegerAccessor(&P4TrafficManager::m_voqLimit), + MakeUintegerChecker()) + .AddAttribute("EgressPortLimit", + "Per-output-port egress buffer limit in bytes (0 = unlimited).", + UintegerValue(0), + MakeUintegerAccessor(&P4TrafficManager::m_egressPortLimit), + MakeUintegerChecker()) + .AddAttribute("EgressQueueLimit", + "Per-egress-queue[out][prio] limit in bytes (0 = unlimited).", + UintegerValue(0), + MakeUintegerAccessor(&P4TrafficManager::m_egressQueueLimit), + MakeUintegerChecker()) + .AddAttribute("PortRate", + "Output-port serialization rate.", + DataRateValue(DataRate("1Gbps")), + MakeDataRateAccessor(&P4TrafficManager::m_portRate), + MakeDataRateChecker()) + .AddAttribute("FabricRate", + "Fabric transfer rate.", + DataRateValue(DataRate("10Gbps")), + MakeDataRateAccessor(&P4TrafficManager::m_fabricRate), + MakeDataRateChecker()) + .AddAttribute("IngressPipelineDelay", + "Fixed ingress pipeline processing delay.", + TimeValue(Time(0)), + MakeTimeAccessor(&P4TrafficManager::m_ingressPipelineDelay), + MakeTimeChecker()) + .AddAttribute("FabricArbitrationDelay", + "Fixed fabric arbitration delay per round.", + TimeValue(Time(0)), + MakeTimeAccessor(&P4TrafficManager::m_fabricArbitrationDelay), + MakeTimeChecker()) + .AddAttribute("EgressPipelineDelay", + "Fixed egress pipeline processing delay.", + TimeValue(Time(0)), + MakeTimeAccessor(&P4TrafficManager::m_egressPipelineDelay), + MakeTimeChecker()) + .AddAttribute("EventDriven", + "If true, EnqueueToVoq self-clocks the fabric + egress " + "scheduler via ns-3 events; if false, the fabric is driven " + "manually via RunFabricScheduler()/DequeueFromVoq().", + BooleanValue(false), + MakeBooleanAccessor(&P4TrafficManager::m_eventDriven), + MakeBooleanChecker()) + .AddTraceSource("VoqEnqueue", + "A packet was enqueued into a VOQ (in, out, prio, bytes).", + MakeTraceSourceAccessor(&P4TrafficManager::m_voqEnqueueTrace), + "ns3::P4TrafficManager::QueueOpCallback") + .AddTraceSource("VoqDequeue", + "A packet was dequeued from a VOQ (in, out, prio, bytes).", + MakeTraceSourceAccessor(&P4TrafficManager::m_voqDequeueTrace), + "ns3::P4TrafficManager::QueueOpCallback") + .AddTraceSource("EgressEnqueue", + "A packet was enqueued into an egress queue (in, out, prio, bytes).", + MakeTraceSourceAccessor(&P4TrafficManager::m_egressEnqueueTrace), + "ns3::P4TrafficManager::QueueOpCallback") + .AddTraceSource("EgressDequeue", + "A packet was dequeued from an egress queue (in, out, prio, bytes).", + MakeTraceSourceAccessor(&P4TrafficManager::m_egressDequeueTrace), + "ns3::P4TrafficManager::QueueOpCallback") + .AddTraceSource("Drop", + "A packet was dropped (reason, in, out, prio, bytes).", + MakeTraceSourceAccessor(&P4TrafficManager::m_dropTrace), + "ns3::P4TrafficManager::DropCallback") + .AddTraceSource("VoqWaitingDelay", + "VOQ waiting delay of a dequeued packet.", + MakeTraceSourceAccessor(&P4TrafficManager::m_voqDelayTrace), + "ns3::P4TrafficManager::DelayCallback") + .AddTraceSource("EgressWaitingDelay", + "Egress queue waiting delay of a dequeued packet.", + MakeTraceSourceAccessor(&P4TrafficManager::m_egressDelayTrace), + "ns3::P4TrafficManager::DelayCallback") + .AddTraceSource("TotalDelay", + "Total Traffic Manager delay of a packet.", + MakeTraceSourceAccessor(&P4TrafficManager::m_totalDelayTrace), + "ns3::P4TrafficManager::DelayCallback"); + return tid; +} + +P4TrafficManager::P4TrafficManager() +{ + NS_LOG_FUNCTION(this); +} + +P4TrafficManager::~P4TrafficManager() +{ + NS_LOG_FUNCTION(this); +} + +void +P4TrafficManager::DoDispose() +{ + NS_LOG_FUNCTION(this); + m_fabricEvent.Cancel(); + for (auto& ev : m_egressEvent) + { + ev.Cancel(); + } + m_transmitCallback = nullptr; + m_voq.clear(); + m_egress.clear(); + m_voqBytes.clear(); + m_inputBufferBytes.clear(); + m_egressPortBytes.clear(); + m_egressQueueBytes.clear(); + Object::DoDispose(); +} + +void +P4TrafficManager::SetTransmitCallback(TransmitCallback cb) +{ + m_transmitCallback = std::move(cb); +} + +// --------------------------------------------------------------------------- +// Setup +// --------------------------------------------------------------------------- + +void +P4TrafficManager::SetNumPorts(uint32_t numPorts) +{ + NS_LOG_FUNCTION(this << numPorts); + m_numPorts = numPorts; + AllocateStructures(); +} + +uint32_t +P4TrafficManager::GetNumPorts() const +{ + return m_numPorts; +} + +void +P4TrafficManager::AllocateStructures() +{ + NS_LOG_FUNCTION(this << m_numPorts); + + const uint32_t n = m_numPorts; + const size_t nn = static_cast(n) * n; + + // Reallocating cancels any in-flight scheduling from a previous sizing. + m_fabricEvent.Cancel(); + for (auto& ev : m_egressEvent) + { + ev.Cancel(); + } + m_fabricScheduled = false; + + // TmItem is move-only, so the VOQ element type cannot be relocated by + // vector::resize/assign (those instantiate a copy fallback). Direct sized + // construction value-initialises each element in place, which is fine. + m_voq = std::vector(nn); + m_egress = std::vector(n); + m_voqBytes = std::vector(nn, PriBytes{}); + m_inputBufferBytes.assign(n, 0); + m_egressPortBytes.assign(n, 0); + m_egressQueueBytes.assign(n, PriBytes{}); + m_egressBusy.assign(n, false); + m_egressEvent = std::vector(n); + + m_globalBufferBytes = 0; + m_stats.perPortTxBytes.assign(n, 0); + + NS_LOG_INFO("Allocated " << (static_cast(n) * n * P4_TM_NUM_PRIORITIES) + << " VOQs for " << n << " ports"); +} + +bool +P4TrafficManager::ValidPort(uint32_t p) const +{ + return p < m_numPorts; +} + +bool +P4TrafficManager::ValidPriority(uint8_t prio) const +{ + return prio < P4_TM_NUM_PRIORITIES; +} + +// --------------------------------------------------------------------------- +// Ingress boundary: EnqueueToVoq +// --------------------------------------------------------------------------- + +bool +P4TrafficManager::EnqueueToVoq(std::unique_ptr payload, + uint32_t sizeBytes, + uint32_t inPort, + uint32_t outPort, + uint8_t priority) +{ + NS_LOG_FUNCTION(this << sizeBytes << inPort << outPort << static_cast(priority)); + + NS_ABORT_MSG_IF(!ValidPort(inPort) || !ValidPort(outPort), + "EnqueueToVoq: port index out of range (numPorts=" << m_numPorts << ")"); + NS_ABORT_MSG_IF(!ValidPriority(priority), "EnqueueToVoq: priority must be 0..7"); + + m_stats.totalReceived++; + + // Admission control, in order: global -> input -> VOQ. + auto drop = [&](TmDropReason reason) { + m_stats.totalDropped++; + m_stats.dropsByReason[static_cast(reason)]++; + m_dropTrace(static_cast(reason), inPort, outPort, priority, sizeBytes); + NS_LOG_DEBUG("Dropped " << sizeBytes << "B pkt (" << TmDropReasonToString(reason) + << ") in=" << inPort << " out=" << outPort + << " prio=" << static_cast(priority)); + // payload released on return (unique_ptr goes out of scope) + return false; + }; + + if (m_globalBufferLimit != 0 && m_globalBufferBytes + sizeBytes > m_globalBufferLimit) + { + return drop(TmDropReason::VOQ_GLOBAL_BUFFER_FULL); + } + if (m_inputBufferLimit != 0 && m_inputBufferBytes[inPort] + sizeBytes > m_inputBufferLimit) + { + return drop(TmDropReason::VOQ_INPUT_BUFFER_FULL); + } + if (m_voqLimit != 0 && m_voqBytes[Idx(inPort, outPort)][priority] + sizeBytes > m_voqLimit) + { + return drop(TmDropReason::VOQ_QUEUE_FULL); + } + + // Accept. + TmItem item; + item.payload = std::move(payload); + item.sizeBytes = sizeBytes; + item.inPort = inPort; + item.outPort = outPort; + item.priority = priority; + item.voqEnqueueTime = Simulator::Now(); + item.uid = m_nextUid++; + + m_voq[Idx(inPort, outPort)][priority].push_back(std::move(item)); + m_voqBytes[Idx(inPort, outPort)][priority] += sizeBytes; + m_inputBufferBytes[inPort] += sizeBytes; + m_globalBufferBytes += sizeBytes; + + m_stats.totalVoqEnqueued++; + m_voqEnqueueTrace(inPort, outPort, priority, sizeBytes); + + NS_LOG_DEBUG("Enqueued " << sizeBytes << "B into VOQ[" << inPort << "][" << outPort << "][" + << static_cast(priority) << "]"); + + // Event-driven mode: wake the fabric so this packet gets scheduled. + ScheduleFabricRound(); + return true; +} + +// --------------------------------------------------------------------------- +// Fabric scheduler +// --------------------------------------------------------------------------- + +std::vector +P4TrafficManager::RunFabricScheduler() +{ + return DoRunFabricScheduler(); +} + +std::vector +P4TrafficManager::DoRunFabricScheduler() +{ + NS_LOG_FUNCTION(this); + + std::vector grants; + if (m_numPorts == 0) + { + return grants; + } + + std::vector inputUsed(m_numPorts, false); + std::vector outputUsed(m_numPorts, false); + + // Priority-first maximal matching: highest priority (7) first. + for (int p = P4_TM_NUM_PRIORITIES - 1; p >= 0; --p) + { + for (uint32_t in = 0; in < m_numPorts; ++in) + { + if (inputUsed[in]) + { + continue; + } + for (uint32_t out = 0; out < m_numPorts; ++out) + { + if (outputUsed[out]) + { + continue; + } + if (VoqNotEmpty(in, out, static_cast(p))) + { + grants.push_back({in, out, static_cast(p)}); + inputUsed[in] = true; + outputUsed[out] = true; + break; // this input is now used; move to next input + } + } + } + } + + NS_LOG_DEBUG("Fabric scheduler produced " << grants.size() << " grants"); + return grants; +} + +bool +P4TrafficManager::DequeueFromVoq(uint32_t inPort, + uint32_t outPort, + uint8_t priority, + TmItem& item) +{ + NS_LOG_FUNCTION(this << inPort << outPort << static_cast(priority)); + + NS_ABORT_MSG_IF(!ValidPort(inPort) || !ValidPort(outPort) || !ValidPriority(priority), + "DequeueFromVoq: index out of range"); + + auto& q = m_voq[Idx(inPort, outPort)][priority]; + if (q.empty()) + { + return false; + } + + item = std::move(q.front()); + q.pop_front(); + + const uint32_t sizeBytes = item.sizeBytes; + m_voqBytes[Idx(inPort, outPort)][priority] -= sizeBytes; + m_inputBufferBytes[inPort] -= sizeBytes; + m_globalBufferBytes -= sizeBytes; + + m_stats.totalMovedToEgress++; + + Time voqDelay = Simulator::Now() - item.voqEnqueueTime; + m_stats.sumVoqDelay += voqDelay; + m_stats.cntVoqDelay++; + if (voqDelay > m_stats.maxQueueingDelay) + { + m_stats.maxQueueingDelay = voqDelay; + } + + m_voqDequeueTrace(inPort, outPort, priority, sizeBytes); + m_voqDelayTrace(voqDelay); + + NS_LOG_DEBUG("Dequeued " << sizeBytes << "B from VOQ[" << inPort << "][" << outPort << "][" + << static_cast(priority) << "] voqDelay=" + << voqDelay.GetNanoSeconds() << "ns"); + return true; +} + +// --------------------------------------------------------------------------- +// P3: event-driven fabric loop +// --------------------------------------------------------------------------- + +void +P4TrafficManager::ScheduleFabricRound() +{ + if (!m_eventDriven || m_fabricScheduled) + { + return; + } + m_fabricScheduled = true; + m_fabricEvent = Simulator::Schedule(m_fabricArbitrationDelay, + &P4TrafficManager::RunFabricRoundEvent, + this); +} + +void +P4TrafficManager::RunFabricRoundEvent() +{ + NS_LOG_FUNCTION(this); + m_fabricScheduled = false; + + std::vector grants = RunFabricScheduler(); + + // Apply grants: move each granted head packet from its VOQ to egress. The + // fabric transfers all grants of a round in parallel, so its busy time is + // the transfer time of the largest granted packet. + uint32_t maxBytes = 0; + for (const Grant& g : grants) + { + TmItem item; + if (!DequeueFromVoq(g.inPort, g.outPort, g.priority, item)) + { + continue; // defensive: VOQ emptied out from under the grant + } + maxBytes = std::max(maxBytes, item.sizeBytes); + EnqueueToEgress(std::move(item)); + } + + // Re-arm the next round while there is still demand. The next arbitration + // happens after this round's fabric transfer plus the arbitration delay. + if (AnyVoqNonEmpty()) + { + Time transfer = (maxBytes > 0) ? m_fabricRate.CalculateBytesTxTime(maxBytes) : Time(0); + m_fabricScheduled = true; + m_fabricEvent = Simulator::Schedule(m_fabricArbitrationDelay + transfer, + &P4TrafficManager::RunFabricRoundEvent, + this); + } +} + +bool +P4TrafficManager::AnyVoqNonEmpty() const +{ + for (const PriQueues& pq : m_voq) + { + for (const std::deque& q : pq) + { + if (!q.empty()) + { + return true; + } + } + } + return false; +} + +// --------------------------------------------------------------------------- +// P4: egress queues + serialising egress scheduler +// --------------------------------------------------------------------------- + +bool +P4TrafficManager::EnqueueToEgress(TmItem&& item) +{ + const uint32_t in = item.inPort; + const uint32_t out = item.outPort; + const uint8_t prio = item.priority; + const uint32_t sizeBytes = item.sizeBytes; + + auto drop = [&](TmDropReason reason) { + m_stats.totalDropped++; + m_stats.dropsByReason[static_cast(reason)]++; + m_dropTrace(static_cast(reason), in, out, prio, sizeBytes); + NS_LOG_DEBUG("Egress drop " << sizeBytes << "B (" << TmDropReasonToString(reason) + << ") out=" << out << " prio=" + << static_cast(prio)); + // The packet has already been removed from the VOQ (global was + // decremented there); dropping it here simply releases the payload. + return false; + }; + + // Admission: per-output-port then per-egress-queue. Global is NOT + // re-checked: the VOQ->egress hand-off is byte-neutral for the global + // counter (DequeueFromVoq subtracted these bytes, we add them back below), + // so the packet was already counted globally throughout its residence. + if (m_egressPortLimit != 0 && m_egressPortBytes[out] + sizeBytes > m_egressPortLimit) + { + return drop(TmDropReason::EGRESS_PORT_BUFFER_FULL); + } + if (m_egressQueueLimit != 0 && m_egressQueueBytes[out][prio] + sizeBytes > m_egressQueueLimit) + { + return drop(TmDropReason::EGRESS_QUEUE_FULL); + } + + item.egressEnqueueTime = Simulator::Now(); + m_egress[out][prio].push_back(std::move(item)); + m_egressPortBytes[out] += sizeBytes; + m_egressQueueBytes[out][prio] += sizeBytes; + m_globalBufferBytes += sizeBytes; + + m_stats.totalEgressEnqueued++; + m_egressEnqueueTrace(in, out, prio, sizeBytes); + + ScheduleEgressService(out); + return true; +} + +int +P4TrafficManager::SelectEgressPriority(uint32_t outPort) const +{ + for (int p = P4_TM_NUM_PRIORITIES - 1; p >= 0; --p) + { + if (!m_egress[outPort][static_cast(p)].empty()) + { + return p; + } + } + return -1; +} + +void +P4TrafficManager::ScheduleEgressService(uint32_t outPort) +{ + if (!m_eventDriven || m_egressBusy[outPort]) + { + return; + } + m_egressBusy[outPort] = true; + m_egressEvent[outPort] = Simulator::Schedule(m_egressPipelineDelay, + &P4TrafficManager::EgressServiceEvent, + this, + outPort); +} + +void +P4TrafficManager::EgressServiceEvent(uint32_t outPort) +{ + NS_LOG_FUNCTION(this << outPort); + + const int prio = SelectEgressPriority(outPort); + if (prio < 0) + { + m_egressBusy[outPort] = false; // nothing to send: the port goes idle + return; + } + + std::deque& q = m_egress[outPort][static_cast(prio)]; + TmItem item = std::move(q.front()); + q.pop_front(); + + const uint32_t sizeBytes = item.sizeBytes; + m_egressPortBytes[outPort] -= sizeBytes; + m_egressQueueBytes[outPort][static_cast(prio)] -= sizeBytes; + m_globalBufferBytes -= sizeBytes; + + const Time now = Simulator::Now(); + const Time egressDelay = now - item.egressEnqueueTime; + const Time totalDelay = now - item.voqEnqueueTime; + + m_stats.totalTransmitted++; + m_stats.perPriorityTransmitted[static_cast(prio)]++; + m_stats.perPortTxBytes[outPort] += sizeBytes; + m_stats.sumEgressDelay += egressDelay; + m_stats.cntEgressDelay++; + m_stats.sumTotalDelay += totalDelay; + m_stats.cntTotalDelay++; + if (totalDelay > m_stats.maxQueueingDelay) + { + m_stats.maxQueueingDelay = totalDelay; + } + + m_egressDequeueTrace(item.inPort, outPort, static_cast(prio), sizeBytes); + m_egressDelayTrace(egressDelay); + m_totalDelayTrace(totalDelay); + + // Deliver the payload (integration wires this to the ns-3 NetDevice send). + if (m_transmitCallback) + { + m_transmitCallback(outPort, static_cast(prio), std::move(item.payload)); + } + + // Serialisation: the port is busy for this packet's transmission time; the + // next packet is serviced when transmission completes. The service event + // finding the port empty is what finally clears m_egressBusy. + const Time txTime = m_portRate.CalculateBytesTxTime(sizeBytes); + m_egressEvent[outPort] = Simulator::Schedule(txTime, + &P4TrafficManager::EgressServiceEvent, + this, + outPort); +} + +// --------------------------------------------------------------------------- +// Occupancy queries +// --------------------------------------------------------------------------- + +size_t +P4TrafficManager::VoqLength(uint32_t inPort, uint32_t outPort, uint8_t priority) const +{ + if (!ValidPort(inPort) || !ValidPort(outPort) || !ValidPriority(priority)) + { + return 0; + } + return m_voq[Idx(inPort, outPort)][priority].size(); +} + +bool +P4TrafficManager::VoqNotEmpty(uint32_t inPort, uint32_t outPort, uint8_t priority) const +{ + return VoqLength(inPort, outPort, priority) > 0; +} + +uint64_t +P4TrafficManager::GlobalBufferBytes() const +{ + return m_globalBufferBytes; +} + +uint64_t +P4TrafficManager::InputBufferBytes(uint32_t inPort) const +{ + return ValidPort(inPort) ? m_inputBufferBytes[inPort] : 0; +} + +uint64_t +P4TrafficManager::VoqBytes(uint32_t inPort, uint32_t outPort, uint8_t priority) const +{ + if (!ValidPort(inPort) || !ValidPort(outPort) || !ValidPriority(priority)) + { + return 0; + } + return m_voqBytes[Idx(inPort, outPort)][priority]; +} + +size_t +P4TrafficManager::EgressLength(uint32_t outPort, uint8_t priority) const +{ + if (!ValidPort(outPort) || !ValidPriority(priority)) + { + return 0; + } + return m_egress[outPort][priority].size(); +} + +uint64_t +P4TrafficManager::EgressPortBytes(uint32_t outPort) const +{ + return ValidPort(outPort) ? m_egressPortBytes[outPort] : 0; +} + +uint64_t +P4TrafficManager::EgressQueueBytes(uint32_t outPort, uint8_t priority) const +{ + if (!ValidPort(outPort) || !ValidPriority(priority)) + { + return 0; + } + return m_egressQueueBytes[outPort][priority]; +} + +// --------------------------------------------------------------------------- +// Statistics +// --------------------------------------------------------------------------- + +const P4TrafficManager::TmStats& +P4TrafficManager::GetStats() const +{ + return m_stats; +} + +void +P4TrafficManager::ResetStats() +{ + const size_t nPorts = m_stats.perPortTxBytes.size(); + m_stats = TmStats{}; + m_stats.perPortTxBytes.assign(nPorts, 0); +} + +} // namespace ns3 diff --git a/utils/p4-traffic-manager.h b/utils/p4-traffic-manager.h new file mode 100644 index 0000000..9274d69 --- /dev/null +++ b/utils/p4-traffic-manager.h @@ -0,0 +1,412 @@ +/* + * Copyright (c) 2025 TU Dresden + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 2 as + * published by the Free Software Foundation; + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * + * Authors: Vineet Goel + */ + +#ifndef P4_TRAFFIC_MANAGER_H +#define P4_TRAFFIC_MANAGER_H + +#include "ns3/data-rate.h" +#include "ns3/event-id.h" +#include "ns3/nstime.h" +#include "ns3/object.h" +#include "ns3/traced-callback.h" + +#include +#include +#include +#include +#include +#include + +namespace ns3 +{ + +/** + * \ingroup p4sim + * + * \brief High-fidelity switch Traffic Manager (VOQ + fabric arbitration). + * + * This is the first stage of a high-fidelity traffic-manager architecture that + * replaces the output-only priority scheduler (NSQueueingLogicPriRL). The + * intended packet flow is: + * + * ingress pipeline -> EnqueueToVoq(VOQ[in][out][prio]) + * -> fabric scheduler grants (in -> out) + * -> egress queue (per output port / priority) + * -> egress pipeline -> ns-3 NetDevice + * + * This file implements phases P1 (VOQ + finite-buffer accounting + drop + * reasons + stats/traces), P2 (priority-first maximal-matching fabric + * scheduler), P3 (ns-3 event-driven fabric-round timing) and P4 (per-output + * strict-priority egress queues + serialising egress scheduler). + * + * Two ways to drive the pipeline: + * - Manual (default): call RunFabricScheduler()/DequeueFromVoq() yourself. + * Nothing is scheduled on the ns-3 event queue. Used by the low-level + * unit tests. + * - Event-driven (attribute "EventDriven" = true): EnqueueToVoq() arms a + * self-clocking fabric loop (Simulator::Schedule, no threads) that moves + * packets VOQ -> egress -> wire, respecting fabric/port rates and the + * arbitration/pipeline delays. Transmitted packets are handed to the + * TransmitCallback (wired to an ns-3 NetDevice at integration time). + * + * Packet-format boundary: the Traffic Manager does NOT assume ns3::Packet. It + * carries an opaque, move-only ::ns3::TmPayload (see BmPacketPayload for the + * bm::Packet wrapper used by the real switch core). Scheduling and accounting + * are driven solely by the metadata in TmItem, never by the payload contents. + */ + +/// Number of priority levels. The priority field is 3 bits, so 8 levels. +/// Higher value = higher priority (priority 7 is the highest). +static constexpr uint8_t P4_TM_NUM_PRIORITIES = 8; + +/** + * \brief Reason a packet was dropped by the Traffic Manager. + */ +enum class TmDropReason : uint8_t +{ + VOQ_GLOBAL_BUFFER_FULL = 0, ///< global buffer would overflow + VOQ_INPUT_BUFFER_FULL, ///< per-input-port buffer would overflow + VOQ_QUEUE_FULL, ///< the target VOQ[in][out][prio] would overflow + EGRESS_PORT_BUFFER_FULL, ///< per-output-port egress buffer would overflow + EGRESS_QUEUE_FULL, ///< the target egress queue[out][prio] would overflow +}; + +/// Human-readable name for a drop reason (for logging/tracing). +const char* TmDropReasonToString(TmDropReason r); + +/** + * \brief Opaque, move-only payload carried by the Traffic Manager. + * + * The TM never inspects the payload; it only moves it from VOQ to egress. + * Concrete payloads (a bm::Packet wrapper, or a unit-test stub) derive from + * this base. This keeps the TM decoupled from any specific packet format. + */ +class TmPayload +{ + public: + virtual ~TmPayload() = default; +}; + +/** + * \brief A unit of work inside the Traffic Manager. + * + * Packet-level only: no cell segmentation. Move-only, because it owns the + * payload. + */ +struct TmItem +{ + std::unique_ptr payload; ///< opaque payload (bm::Packet, test stub, ...) + uint32_t sizeBytes{0}; ///< size in bytes, used for buffer accounting + uint32_t inPort{0}; ///< ingress port + uint32_t outPort{0}; ///< egress port chosen by ingress + uint8_t priority{0}; ///< 0..7, 7 = highest + Time voqEnqueueTime; ///< when the item entered the VOQ + Time egressEnqueueTime; ///< when the item entered the egress queue + uint64_t uid{0}; ///< monotonically increasing id (tracing / tie-break) + + TmItem() = default; + TmItem(TmItem&&) = default; + TmItem& operator=(TmItem&&) = default; + TmItem(const TmItem&) = delete; + TmItem& operator=(const TmItem&) = delete; +}; + +/** + * \brief A fabric matching decision produced by the scheduler. + * + * Instructs the fabric to move one packet from VOQ[inPort][outPort][priority] + * to the egress side. + */ +struct Grant +{ + uint32_t inPort; + uint32_t outPort; + uint8_t priority; +}; + +/** + * \ingroup p4sim + * \brief Traffic Manager: input-side VOQ + priority-first fabric scheduler. + */ +class P4TrafficManager : public Object +{ + public: + /// TracedCallback signature for VOQ/egress enqueue and dequeue events. + typedef void (*QueueOpCallback)(uint32_t inPort, + uint32_t outPort, + uint8_t priority, + uint32_t sizeBytes); + /// TracedCallback signature for drop events. + typedef void (*DropCallback)(uint8_t reason, + uint32_t inPort, + uint32_t outPort, + uint8_t priority, + uint32_t sizeBytes); + /// TracedCallback signature for waiting-delay measurements. + typedef void (*DelayCallback)(Time delay); + + /** + * \brief Delivery hook invoked when the egress scheduler serialises a + * packet onto the wire (event-driven mode only). + * + * The Traffic Manager transfers ownership of the payload to the callback. + * At integration time this is wired to the ns-3 NetDevice send path; the + * unit tests use it to observe transmit order. std::function (not + * ns3::Callback) because it carries a move-only unique_ptr argument. + */ + using TransmitCallback = + std::function payload)>; + + static TypeId GetTypeId(); + + P4TrafficManager(); + ~P4TrafficManager() override; + + // ---- Setup ---- + + /** + * \brief Set the number of switch ports and (re)allocate all structures. + * + * Allocates N*N*8 VOQs and the associated byte counters. Also settable + * via the "NumPorts" attribute. + * \param numPorts number of ports N. + */ + void SetNumPorts(uint32_t numPorts); + uint32_t GetNumPorts() const; + + /** + * \brief Set the delivery hook for transmitted packets (event-driven mode). + * \param cb callback receiving (outPort, priority, payload). + */ + void SetTransmitCallback(TransmitCallback cb); + + // ---- Ingress boundary ---- + + /** + * \brief Enqueue a packet into VOQ[inPort][outPort][priority]. + * + * Finite-buffer admission is checked in order: global, input, VOQ. If any + * limit would be exceeded the packet is dropped, m_dropTrace fires with the + * corresponding TmDropReason, and the function returns false (payload is + * released). + * + * \param payload the opaque payload (ownership taken on success). + * \param sizeBytes packet size in bytes (for accounting). + * \param inPort ingress port (< numPorts). + * \param outPort egress port (< numPorts). + * \param priority priority 0..7 (7 = highest). + * \return true if accepted into the VOQ, false if dropped. + */ + bool EnqueueToVoq(std::unique_ptr payload, + uint32_t sizeBytes, + uint32_t inPort, + uint32_t outPort, + uint8_t priority); + + // ---- Fabric ---- + + /** + * \brief Run one round of fabric matching. + * + * Delegates to DoRunFabricScheduler() (overridable to swap in iSLIP / + * round-robin). The default policy is priority-first maximal matching. + * \return the list of grants for this round. + */ + std::vector RunFabricScheduler(); + + /** + * \brief Remove the head item of VOQ[inPort][outPort][priority]. + * + * Used to apply a Grant. Updates all byte counters and fires the VOQ + * dequeue / VOQ-waiting-delay traces. + * \param inPort ingress port. + * \param outPort egress port. + * \param priority priority level. + * \param[out] item receives the dequeued item on success. + * \return true if an item was dequeued, false if the VOQ was empty. + */ + bool DequeueFromVoq(uint32_t inPort, uint32_t outPort, uint8_t priority, TmItem& item); + + // ---- Occupancy queries ---- + + /// Number of packets queued in VOQ[in][out][prio]. + size_t VoqLength(uint32_t inPort, uint32_t outPort, uint8_t priority) const; + /// True if VOQ[in][out][prio] holds at least one packet. + bool VoqNotEmpty(uint32_t inPort, uint32_t outPort, uint8_t priority) const; + /// Total bytes occupied across VOQ (and, later, egress). + uint64_t GlobalBufferBytes() const; + /// Bytes occupied by all VOQs of a given input port. + uint64_t InputBufferBytes(uint32_t inPort) const; + /// Bytes occupied by VOQ[in][out][prio]. + uint64_t VoqBytes(uint32_t inPort, uint32_t outPort, uint8_t priority) const; + + /// Number of packets queued in egress[out][prio]. + size_t EgressLength(uint32_t outPort, uint8_t priority) const; + /// Bytes occupied by all egress queues of a given output port. + uint64_t EgressPortBytes(uint32_t outPort) const; + /// Bytes occupied by egress[out][prio]. + uint64_t EgressQueueBytes(uint32_t outPort, uint8_t priority) const; + + // ---- Statistics ---- + + /** + * \brief Cumulative Traffic Manager statistics. + */ + struct TmStats + { + uint64_t totalReceived{0}; ///< packets offered to EnqueueToVoq + uint64_t totalVoqEnqueued{0}; ///< packets accepted into a VOQ + uint64_t totalMovedToEgress{0}; ///< packets dequeued from VOQ (granted) + uint64_t totalEgressEnqueued{0}; ///< packets accepted into an egress queue + uint64_t totalTransmitted{0}; ///< packets serialised onto the wire + uint64_t totalDropped{0}; ///< packets dropped for any reason + std::array dropsByReason{}; ///< indexed by TmDropReason + std::array perPriorityTransmitted{}; + + // delay accumulators (sums + counts -> averages via helpers) + Time sumVoqDelay; + uint64_t cntVoqDelay{0}; + Time sumEgressDelay; + uint64_t cntEgressDelay{0}; + Time sumTotalDelay; + uint64_t cntTotalDelay{0}; + Time maxQueueingDelay; + + std::vector perPortTxBytes; ///< bytes transmitted per output port + + Time AvgVoqDelay() const; + Time AvgEgressDelay() const; + Time AvgTotalDelay() const; + }; + + const TmStats& GetStats() const; + void ResetStats(); + + protected: + void DoDispose() override; + + /** + * \brief The fabric matching policy. + * + * Default: priority-first maximal matching. For priority 7 down to 0, for + * each unused input port, grant the first unused output port that has a + * non-empty VOQ at that priority. One input grants at most one packet and + * one output receives at most one packet per round. + * + * Override this method (or replace the whole object) to implement iSLIP, + * round-robin, etc. It only needs VoqNotEmpty() as its primitive. + * \return the grants for this round. + */ + virtual std::vector DoRunFabricScheduler(); + + private: + /// (Re)allocate VOQ and accounting structures for m_numPorts ports. + void AllocateStructures(); + /// True if idx < m_numPorts and priority < 8. + bool ValidPort(uint32_t p) const; + bool ValidPriority(uint8_t prio) const; + /// Flatten (inPort, outPort) into the [in*N + out] VOQ index. + size_t Idx(uint32_t inPort, uint32_t outPort) const + { + return static_cast(inPort) * m_numPorts + outPort; + } + + // ---- P3: event-driven fabric loop ---- + /// Arm a fabric round if one is not already pending (event-driven mode). + void ScheduleFabricRound(); + /// One fabric round: apply grants (VOQ -> egress), then re-arm if work remains. + void RunFabricRoundEvent(); + /// True if any VOQ holds at least one packet. + bool AnyVoqNonEmpty() const; + + // ---- P4: egress side ---- + /// Move a granted item into egress[out][prio]; false (and dropped) if full. + bool EnqueueToEgress(TmItem&& item); + /// Arm the serialising egress scheduler for one output port, if idle. + void ScheduleEgressService(uint32_t outPort); + /// Serialise the highest-priority egress packet of a port, then re-arm. + void EgressServiceEvent(uint32_t outPort); + /// Highest non-empty egress priority of a port, or -1 if all empty. + int SelectEgressPriority(uint32_t outPort) const; + + uint32_t m_numPorts{0}; + + /// Event-driven mode flag (attribute "EventDriven"); see class doc. + bool m_eventDriven{false}; + + // Finite-buffer limits (bytes). 0 means "no limit". + uint64_t m_globalBufferLimit{0}; + uint64_t m_inputBufferLimit{0}; + uint64_t m_voqLimit{0}; + uint64_t m_egressPortLimit{0}; + uint64_t m_egressQueueLimit{0}; + + // Timing configuration (used by later phases; part of the config surface). + DataRate m_portRate; + DataRate m_fabricRate; + Time m_ingressPipelineDelay; + Time m_fabricArbitrationDelay; + Time m_egressPipelineDelay; + + // ---- VOQ storage: logically queues[inPort][outPort][priority] ---- + // Stored flat (size N*N, indexed by Idx(in,out)) because the element type + // holds move-only TmItem and cannot be relocated by vector::resize/assign. + using PriQueues = std::array, P4_TM_NUM_PRIORITIES>; + std::vector m_voq; ///< [Idx(in,out)] -> 8 priority deques + + // ---- Egress storage: per output port, 8 strict-priority queues ---- + std::vector m_egress; ///< [out] -> 8 priority deques + + // ---- Byte accounting ---- + uint64_t m_globalBufferBytes{0}; + std::vector m_inputBufferBytes; ///< [in] + using PriBytes = std::array; + std::vector m_voqBytes; ///< [Idx(in,out)] -> 8 counters + std::vector m_egressPortBytes; ///< [out] + std::vector m_egressQueueBytes; ///< [out] -> 8 counters + + // ---- Event-driven scheduling state ---- + bool m_fabricScheduled{false}; ///< a fabric round is pending + EventId m_fabricEvent; ///< pending fabric-round event + std::vector m_egressBusy; ///< [out] port is serialising + std::vector m_egressEvent; ///< [out] pending egress-service event + TransmitCallback m_transmitCallback; ///< delivery hook (set by integration) + + uint64_t m_nextUid{0}; + TmStats m_stats; + + // ---- Trace sources ---- + TracedCallback m_voqEnqueueTrace; + TracedCallback m_voqDequeueTrace; + TracedCallback m_egressEnqueueTrace; + TracedCallback m_egressDequeueTrace; + TracedCallback m_dropTrace; + TracedCallback