queueing: generalize DynamicClassifier for pull-based per-class structures

WHATSNEW

@@ -56,6 +56,22 @@ Notable backward compatible changes are the following:
    owns a MAC address, by the host's path relative to the network, and falls back
    to the MAC address string when no host owns the address.
 
+4. Dynamic packet classifier branches
+
+   DynamicClassifier creates the branch of each traffic class the first time a
+   packet of that class arrives. It creates it on the delivery path now, instead
+   of as a side effect of classifying a packet, which also happens whenever the
+   classifier is merely asked whether it can take one.
+
+   It also gained an aggregatorSubmoduleName parameter that names the submodule
+   its branches are wired into, which was previously a submodule named
+   "multiplexer"; the aggregator may now be a pull scheduler as well as a push
+   multiplexer. The inherited reverseOrder parameter is refused, because the
+   classifier no longer maps class indices through the output gate order.
+
+   Existing simulation models are unaffected: the defaults reproduce the previous
+   behavior.
+
 
 INET-4.7 (July 2026) — feature release
 --------------------------------------

src/inet/queueing/classifier/DynamicClassifier.cc

@@ -21,44 +21,134 @@ void DynamicClassifier::initialize(int stage)
     if (stage == INITSTAGE_LOCAL) {
         submoduleName = par("submoduleName");
         moduleType = cModuleType::get(par("moduleType"));
+        aggregatorSubmoduleName = par("aggregatorSubmoduleName");
         if (!getParentModule()->hasSubmoduleVector(submoduleName))
-             throw cRuntimeError("The submodule vector '%s' missing from %s", submoduleName, getParentModule()->getFullPath().c_str());
-     }
+            throw cRuntimeError("The submodule vector '%s' is missing from %s", submoduleName, getParentModule()->getFullPath().c_str());
+        if (getParentModule()->getSubmodule(aggregatorSubmoduleName) == nullptr)
+            throw cRuntimeError("The aggregator submodule '%s' is missing from %s", aggregatorSubmoduleName, getParentModule()->getFullPath().c_str());
+        if (reverseOrder)
+            throw cRuntimeError("The reverseOrder parameter is not supported: branches are created in the order the classes of the packets first appear");
+    }
+}
+
+int DynamicClassifier::getClassIndex(Packet *packet) const
+{
+    // The class of the packet, taken as the classifier function returns it, and not mapped
+    // through getOutputGateIndex(): that mapping depends on the number of output gates, which
+    // grows with each branch, so the same class would end up under a different key over time,
+    // and get a second branch.
+    return packetClassifierFunction->classifyPacket(packet);
 }
 
 int DynamicClassifier::classifyPacket(Packet *packet)
 {
-    int index = PacketClassifier::classifyPacket(packet);
-    auto it = classIndexToGateItMap.find(index);
-    if (it == classIndexToGateItMap.end()) {
-        auto parentModule = getParentModule();
-        int submoduleIndex = gateSize("out");
-        int origVectorSize = parentModule->getSubmoduleVectorSize(submoduleName);
-        parentModule->setSubmoduleVectorSize(submoduleName, std::max(origVectorSize, submoduleIndex + 1));
-        auto module = moduleType->create(submoduleName, parentModule, submoduleIndex);
-        auto moduleInputGate = module->gate("in");
-        auto moduleOutputGate = module->gate("out");
-        auto multiplexer = parentModule->getSubmodule("multiplexer");
-        multiplexer->setGateSize("in", multiplexer->gateSize("in") + 1);
-        auto multiplexerInputGate = multiplexer->gate("in", multiplexer->gateSize("in") - 1);
-        setGateSize("out", submoduleIndex + 1);
-        auto classifierOutputGate = gate("out", gateSize("out") - 1);
-        classifierOutputGate->connectTo(moduleInputGate);
-        outputGates.push_back(classifierOutputGate);
-        PassivePacketSinkRef consumer;
-        consumer.reference(classifierOutputGate, false);
-        consumers.push_back(consumer);
-        moduleOutputGate->connectTo(multiplexerInputGate);
-        module->finalizeParameters();
-        module->buildInside();
-        module->callInitialize();
-        classIndexToGateItMap[index] = submoduleIndex;
-        return submoduleIndex;
-    }
-    else
-        return it->second;
+    // a plain lookup, free of side effects: the branch of a class that is seen for the first
+    // time is created before the base class classifies, on the delivery path only
+    auto it = classIndexToBranchIndex.find(getClassIndex(packet));
+    return it != classIndexToBranchIndex.end() ? it->second : -1;
+}
+
+void DynamicClassifier::pushPacket(Packet *packet, const cGate *gate)
+{
+    Enter_Method("pushPacket");
+    // usually a no-op: a source that asked canPushPacket() first already had it created
+    createBranchIfAbsent(packet);
+    PacketClassifier::pushPacket(packet, gate);
+}
+
+void DynamicClassifier::startPacketStreaming(Packet *packet)
+{
+    // the one place all three streaming push operations classify through
+    createBranchIfAbsent(packet);
+    PacketClassifier::startPacketStreaming(packet);
+}
+
+bool DynamicClassifier::canPushPacket(Packet *packet, const cGate *gate) const
+{
+    // The answer has to hold for this very packet, because a source that asks may push exactly
+    // it next. Answering yes for a class that has no branch would promise on behalf of a branch
+    // that does not exist and may refuse its first packet, so the branch is created here and
+    // asked. This is the one query whose answer decides the fate of a packet, and it therefore
+    // has to build the thing that decides it; classification itself stays a pure lookup.
+    // KLUDGE the query is const, the model change it needs is not
+    const_cast<DynamicClassifier *>(this)->createBranchIfAbsent(packet);
+    return consumers[classIndexToBranchIndex.at(getClassIndex(packet))].canPushPacket(packet);
+}
+
+bool DynamicClassifier::canPushSomePacket(const cGate *gate) const
+{
+    // The inherited answer is "one of the existing branches can take a packet", which is false
+    // for a classifier that has not built any branch yet: an active source in front of such a
+    // classifier stops and waits for the notification that packets can be pushed again, and
+    // never gets it, because nothing else creates the first branch. Answer true instead, the
+    // branch of the class of the first packet being created for it.
+    //
+    // Once a branch exists the inherited answer is used, and deliberately so. An active source
+    // takes this as the licence to produce a packet and push it -- it cannot ask
+    // canPushPacket() about a packet it has not created yet -- so answering true while every
+    // branch is full pushes into a full queue, which a queue without a packet dropper refuses.
+    // The price is that a full branch also stops a source that would have opened a new class,
+    // but stopping is the safe error, and it is what a statically wired classifier does too.
+    return outputGates.empty() || PacketClassifierBase::canPushSomePacket(gate);
+}
+
+void DynamicClassifier::createBranchIfAbsent(Packet *packet)
+{
+    int classIndex = getClassIndex(packet);
+    if (classIndexToBranchIndex.find(classIndex) == classIndexToBranchIndex.end())
+        classIndexToBranchIndex[classIndex] = createBranch();
+}
+
+int DynamicClassifier::createBranch()
+{
+    cModule *parent = getParentModule();
+    int index = gateSize("out");
+    // grow this classifier's output gate vector
+    setGateSize("out", index + 1);
+    cGate *classifierOutputGate = gate("out", index);
+    // the branch module is built but not initialized yet: its initialization is deferred until
+    // the whole chain, including the aggregator connection, is wired
+    cModule *branch = createBranchModule(index, classifierOutputGate);
+    // Wire the branch output into the aggregator's next input gate. An aggregator that has
+    // to take notice of a runtime-added input (a pull scheduler, for example) learns about
+    // it from the model change notification of this very connection, so nothing here needs
+    // to know what kind of aggregator it is.
+    cModule *aggregator = parent->getSubmodule(aggregatorSubmoduleName);
+    aggregator->setGateSize("in", aggregator->gateSize("in") + 1);
+    cGate *aggregatorInputGate = aggregator->gate("in", aggregator->gateSize("in") - 1);
+    branch->gate("out")->connectTo(aggregatorInputGate);
+    // the sink references resolve the far end of the path eagerly, so they can only be taken
+    // now that the whole branch, up to and including the aggregator, is connected
+    outputGates.push_back(classifierOutputGate);
+    PassivePacketSinkRef consumer;
+    consumer.reference(classifierOutputGate, false);
+    consumers.push_back(consumer);
+    ActivePacketSinkRef collector;
+    collector.reference(classifierOutputGate, false);
+    collectors.push_back(collector);
+    branch->callInitialize();
+    // the branch is a complete path only now, so this is the earliest point where the packet
+    // operations of its modules can be checked, the way the base class checks the gates that
+    // are wired in NED
+    checkPacketOperationSupport(classifierOutputGate);
+    return index;
+}
+
+cModule *DynamicClassifier::createBranchModule(int index, cGate *classifierOutputGate)
+{
+    cModule *parent = getParentModule();
+    // the vector is only ever extended: it may have been declared larger in NED, and shrinking
+    // one that still holds submodules is an error
+    parent->setSubmoduleVectorSize(submoduleName, std::max(parent->getSubmoduleVectorSize(submoduleName), index + 1));
+    // the branch is created with its final name and index, so that its parameters (from the
+    // enclosing NED declaration and from the ini file), its display string and its result
+    // recording are all resolved for the module path it keeps
+    cModule *module = moduleType->create(submoduleName, parent, index);
+    classifierOutputGate->connectTo(module->gate("in"));
+    module->finalizeParameters();
+    module->buildInside();
+    return module;
 }
 
 } // namespace queueing
 } // namespace inet
-

src/inet/queueing/classifier/DynamicClassifier.h

@@ -15,20 +15,36 @@ namespace queueing {
 
 using namespace inet::queueing;
 
+/**
+ * Creates the branch of each traffic class on demand, the first time a packet of
+ * that class is seen. See the NED file for what the branches are built from and
+ * how they are wired.
+ */
 class INET_API DynamicClassifier : public PacketClassifier
 {
   protected:
-    const char *submoduleName = nullptr;
-    cModuleType *moduleType = nullptr;
-    std::map<int, int> classIndexToGateItMap;
+    const char *submoduleName = nullptr; // submodule vector that holds the branches
+    cModuleType *moduleType = nullptr; // type of the per-class branch module (may be a compound)
+    const char *aggregatorSubmoduleName = nullptr; // downstream aggregator submodule (multiplexer or scheduler)
+    std::map<int, int> classIndexToBranchIndex; // the branch of a class, keyed by the index the classifier function returns
 
   protected:
     virtual void initialize(int stage) override;
+    virtual int getClassIndex(Packet *packet) const;
     virtual int classifyPacket(Packet *packet) override;
+    virtual void startPacketStreaming(Packet *packet) override;
+
+    virtual void createBranchIfAbsent(Packet *packet);
+    virtual int createBranch();
+    virtual cModule *createBranchModule(int index, cGate *classifierOutputGate);
+
+  public:
+    virtual bool canPushSomePacket(const cGate *gate) const override;
+    virtual bool canPushPacket(Packet *packet, const cGate *gate) const override;
+    virtual void pushPacket(Packet *packet, const cGate *gate) override;
 };
 
 } // namespace queueing
 } // namespace inet
 
 #endif
-

src/inet/queueing/classifier/DynamicClassifier.ned

@@ -7,10 +7,29 @@
 
 package inet.queueing.classifier;
 
+//
+// Creates the branch of each traffic class on demand. Each branch is one element of
+// the `submoduleName` submodule vector, of type `moduleType` (which may be a
+// compound module), wired between this classifier's output and a downstream aggregator
+// submodule (`aggregatorSubmoduleName`, a push multiplexer by default). The aggregator may also
+// be a pull scheduler; one that has to take notice of an input appearing at runtime learns
+// about it from the model change notification of the connection being made, so no extra
+// contract is needed between the two.
+//
+// The inherited `reverseOrder` parameter is not supported and is refused, because the order of
+// the output gates is the order in which the classes of the packets first appear.
+//
+// The submodule vector must be declared in the enclosing compound module, where it may be
+// empty; the classifier extends it as branches are created. A branch is created with its final
+// name and index, so parameter assignments (both from the enclosing NED declaration and from
+// the ini file), display string configuration and result recording all address it as
+// `<vector name>[k]`.
+//
 simple DynamicClassifier extends PacketClassifier
 {
     parameters:
-        string submoduleName;
-        string moduleType;
+        string moduleType; // NED type of the per-class branch module, may be a compound module
+        string submoduleName; // Name of the submodule vector that holds the branches
+        string aggregatorSubmoduleName = default("multiplexer"); // Name of the downstream aggregator submodule the branches are wired into
         @class(DynamicClassifier);
 }

tests/queueing/DynamicClassifier_1.test

@@ -0,0 +1,130 @@
+%description:
+
+In this test, packets are produced periodically by an active packet source (ActivePacketSource)
+and are classified into two classes by a dynamic classifier (DynamicClassifier). The classifier
+creates the branch of a class when the first packet of that class arrives, as one element of the
+branch submodule vector, and wires it into the packet multiplexer that aggregates the branches.
+
+The producer is connected to the classifier directly, so the test also covers that a classifier
+which has no branch yet accepts a packet, rather than stopping the producer before the first
+branch is created.
+
+The branch is created with its final name and index, so the test checks that an ini file
+assignment addressing a submodule of a branch (the delay of the packet delayer in it) takes
+effect, and that the statistics of the branch submodules are recorded under the branch path.
+Empty output vectors are turned off, so a vector appears in the result file only if data was
+recorded into it.
+
+%file: test.ned
+
+import inet.queueing.classifier.DynamicClassifier;
+import inet.queueing.common.BackPressureBarrier;
+import inet.queueing.common.PacketDelayer;
+import inet.queueing.common.PacketMultiplexer;
+import inet.queueing.sink.PassivePacketSink;
+import inet.queueing.source.ActivePacketSource;
+
+module TestBranch
+{
+    gates:
+        input in;
+        output out;
+    submodules:
+        first: BackPressureBarrier {
+            @display("p=100,100");
+        }
+        second: PacketDelayer {
+            delay = default(0s);
+            @display("p=200,100");
+        }
+    connections:
+        in --> first.in;
+        first.out --> second.in;
+        second.out --> out;
+}
+
+module TestDemultiplexer
+{
+    gates:
+        input in;
+        output out;
+    submodules:
+        classifier: DynamicClassifier {
+            moduleType = "TestBranch";
+            submoduleName = "branch";
+            @display("p=100,100");
+        }
+        branch[0]: TestBranch { // grown on demand, one branch per class
+            @display("p=250,100,column,80");
+        }
+        multiplexer: PacketMultiplexer {
+            @display("p=400,100");
+        }
+    connections allowunconnected:
+        in --> classifier.in;
+        multiplexer.out --> out;
+}
+
+network TestDynamicClassifier
+{
+    submodules:
+        producer: ActivePacketSource {
+            @display("p=100,100");
+        }
+        demultiplexer: TestDemultiplexer {
+            @display("p=200,100");
+        }
+        consumer: PassivePacketSink {
+            @display("p=300,100");
+        }
+    connections:
+        producer.out --> demultiplexer.in;
+        demultiplexer.out --> consumer.in;
+}
+
+%file: Test.cc
+#include "inet/queueing/function/PacketClassifierFunction.h"
+#include "inet/common/packet/Packet.h"
+
+using namespace inet;
+
+static int testClassify(Packet *packet)
+{
+    return packet->getId() % 2;
+}
+
+Register_Packet_Classifier_Function(TestClassifier, testClassify);
+
+%inifile: omnetpp.ini
+
+[General]
+network = TestDynamicClassifier
+sim-time-limit = 10s
+cmdenv-event-banners = false
+cmdenv-log-prefix = "At %ts %N: "
+**.vector-record-empty = false
+*.producer.packetLength = 1B
+*.producer.productionInterval = 1s
+*.demultiplexer.classifier.classifierClass = "TestClassifier"
+*.demultiplexer.branch[*].second.delay = 2s
+
+%# remove formatting
+%subst: /\x1B\[[0-9;]*m//
+%# remove method call lines added in OMNeT++ 6.4
+%subst: /^At \S+ \S+: Method call [^\n]*\n//m
+%#--------------------------------------------------------------------------------------------------------------
+%# the delay assigned to the branch submodule from the ini file must be applied
+%contains-regex: stdout
+At 0s producer: Producing packet, .*?producer-0.*?
+At 1s producer: Producing packet, .*?producer-1.*?
+At 2s consumer: Consuming packet, .*?producer-0.*?
+At 3s consumer: Consuming packet, .*?producer-1.*?
+%#--------------------------------------------------------------------------------------------------------------
+%# the modules that have data in an output vector, and the branch submodules among them
+%postrun-command: grep "^vector " results/*.vec | cut -d ' ' -f 3 | sort -u > modules.out
+%postrun-command: grep -E "\.branch\[" modules.out > branchmodules.out || true
+%#--------------------------------------------------------------------------------------------------------------
+%contains: branchmodules.out
+TestDynamicClassifier.demultiplexer.branch[0].first
+TestDynamicClassifier.demultiplexer.branch[1].first
+%#--------------------------------------------------------------------------------------------------------------