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

src/inet/queueing/base/PacketClassifierBase.cc

@@ -60,11 +60,21 @@ int PacketClassifierBase::callClassifyPacket(Packet *packet) const
 {
     // KLUDGE
     int index = const_cast<PacketClassifierBase *>(this)->classifyPacket(packet);
-    if (index < 0 || static_cast<unsigned int>(index) >= outputGates.size())
+    if (index < -1 || index >= (int)outputGates.size())
         throw cRuntimeError("Packet is classified to invalid output gate: %d", index);
     return index;
 }
 
+int PacketClassifierBase::createGateForPacket(Packet *packet)
+{
+    throw cRuntimeError("Packet cannot be classified to any output gate");
+}
+
+bool PacketClassifierBase::canCreateGateForPacket(Packet *packet) const
+{
+    return false;
+}
+
 void PacketClassifierBase::checkPacketStreaming(Packet *packet)
 {
     if (inProgressStreamId != -1 && (packet == nullptr || packet->getTreeId() != inProgressStreamId))
@@ -76,6 +86,8 @@ void PacketClassifierBase::startPacketStreaming(Packet *packet)
     EV_INFO << "Classifying packet" << EV_FIELD(packet) << EV_ENDL;
     inProgressStreamId = packet->getTreeId();
     inProgressGateIndex = callClassifyPacket(packet);
+    if (inProgressGateIndex == -1)
+        inProgressGateIndex = createGateForPacket(packet);
 }
 
 void PacketClassifierBase::endPacketStreaming(Packet *packet)
@@ -97,6 +109,8 @@ bool PacketClassifierBase::canPushSomePacket(const cGate *gate) const
 bool PacketClassifierBase::canPushPacket(Packet *packet, const cGate *gate) const
 {
     int index = callClassifyPacket(packet);
+    if (index == -1)
+        return canCreateGateForPacket(packet);
     return consumers[index].canPushPacket(packet);
 }
 
@@ -107,6 +121,8 @@ void PacketClassifierBase::pushPacket(Packet *packet, const cGate *gate)
     checkPacketStreaming(nullptr);
     EV_INFO << "Classifying packet" << EV_FIELD(packet) << EV_ENDL;
     int index = callClassifyPacket(packet);
+    if (index == -1)
+        index = createGateForPacket(packet);
     handlePacketProcessed(packet);
     emit(packetPushedSignal, packet);
     pushOrSendPacket(packet, outputGates[index], consumers[index]);
@@ -219,6 +235,8 @@ void PacketClassifierBase::handleCanPullPacketChanged(const cGate *gate)
     auto packet = provider.canPullPacket();
     if (packet != nullptr) {
         int index = callClassifyPacket(packet);
+        if (index == -1)
+            return; // the packet routes to no existing gate, so there is no collector to notify
         auto collector = collectors[index];
         if (collector != nullptr)
             collector.handleCanPullPacketChanged();

src/inet/queueing/base/PacketClassifierBase.h

@@ -43,9 +43,33 @@ class INET_API PacketClassifierBase : public PacketProcessorBase, public Transpa
     virtual void mapRegistrationForwardingGates(cGate *gate, std::function<void(cGate *)> f) override;
 
     virtual size_t getOutputGateIndex(size_t i) const { return reverseOrder ? outputGates.size() - i - 1 : i; }
+
+    /**
+     * Returns the index of the output gate the packet is classified to, or -1
+     * if no existing output gate suits the packet. Classification is a query
+     * and must be free of side effects: the capacity checks (canPushPacket(),
+     * canPullPacket()) classify the same packet as its eventual delivery, and
+     * the pull path classifies it more than once.
+     */
     virtual int classifyPacket(Packet *packet) = 0;
     virtual int callClassifyPacket(Packet *packet) const;
 
+    /**
+     * Called when a packet being pushed is classified to no existing output
+     * gate. This is where side effects of taking such a packet belong: a
+     * classifier that extends itself on demand creates the new output gate
+     * here and returns its index. Called from packet delivery only, never
+     * from a query. The default refuses the packet with an error.
+     */
+    virtual int createGateForPacket(Packet *packet);
+
+    /**
+     * Returns true if createGateForPacket() would provide an output gate for
+     * the packet: the query pair of createGateForPacket(), consulted by
+     * canPushPacket() when classifyPacket() finds no gate.
+     */
+    virtual bool canCreateGateForPacket(Packet *packet) const;
+
     virtual bool isStreamingPacket() const { return inProgressStreamId != -1; }
     virtual void startPacketStreaming(Packet *packet);
     virtual void endPacketStreaming(Packet *packet);

src/inet/queueing/classifier/DynamicClassifier.cc

@@ -21,44 +21,105 @@ 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());
+    }
+}
+
+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();
+    // a class seen for the first time has no gate yet; its branch is created by
+    // createGateForPacket(), which the base class calls on packet delivery only
+    auto it = classIndexToGateItMap.find(getClassIndex(packet));
+    return it != classIndexToGateItMap.end() ? it->second : -1;
+}
+
+int DynamicClassifier::createGateForPacket(Packet *packet)
+{
+    int branchIndex = createBranch();
+    classIndexToGateItMap[getClassIndex(packet)] = branchIndex;
+    return branchIndex;
+}
+
+bool DynamicClassifier::canCreateGateForPacket(Packet *packet) const
+{
+    // a branch can be created for every class, so every packet gets an output gate
+    return true;
+}
+
+bool DynamicClassifier::canPushSomePacket(const cGate *gate) const
+{
+    // Not the inherited "one of the existing branches can take a packet": a packet of a class
+    // that has not been seen yet is taken by the branch created for it, and there may always be
+    // such a class, the range of the classifier function not being known here. Without this, a
+    // classifier that has no branch yet answers that it cannot accept anything, and an active
+    // source in front of it stops before the first branch is ever created. Whether a particular
+    // packet can be pushed is answered by the inherited canPushPacket(), through
+    // canCreateGateForPacket() above.
+    return true;
+}
+
+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);
+    // build the branch and collect the modules whose initialization is deferred until the
+    // whole chain (including the aggregator connection) is wired
+    std::vector<cModule *> modulesToInitialize;
+    cGate *branchOutputGate = createModuleBranch(index, classifierOutputGate, modulesToInitialize);
+    // 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);
+    branchOutputGate->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);
+    for (auto module : modulesToInitialize)
         module->callInitialize();
-        classIndexToGateItMap[index] = submoduleIndex;
-        return submoduleIndex;
-    }
-    else
-        return it->second;
+    return index;
+}
+
+cGate *DynamicClassifier::createModuleBranch(int index, cGate *classifierOutputGate, std::vector<cModule *>& modulesToInitialize)
+{
+    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();
+    modulesToInitialize.push_back(module);
+    return module->gate("out");
 }
 
 } // namespace queueing
 } // namespace inet
-

src/inet/queueing/classifier/DynamicClassifier.h

@@ -8,27 +8,51 @@
 #ifndef __INET_DYNAMICCLASSIFIER_H
 #define __INET_DYNAMICCLASSIFIER_H
 
+#include <vector>
+
 #include "inet/queueing/classifier/PacketClassifier.h"
 
 namespace inet {
 namespace queueing {
 
 using namespace inet::queueing;
 
+/**
+ * A packet classifier that creates the branch for each traffic class on demand, the first
+ * time a packet of that class is seen. Each branch is one element of a submodule vector
+ * (submoduleName) of a configurable type (moduleType), wired between this classifier's output
+ * and a downstream aggregator submodule (aggregatorSubmoduleName).
+ *
+ * The aggregator may be either a push ~PacketMultiplexer (the traditional use) or a pull
+ * scheduler. An aggregator that needs to take notice of an input appearing at runtime picks
+ * it up from the POST_MODEL_CHANGE notification of the connection itself (see
+ * cPostPathCreateNotification), so no extra contract is needed between the two. This lets the
+ * same classifier build both push demux/remux chains and pull per-class queue/scheduler
+ * structures.
+ */
 class INET_API DynamicClassifier : public PacketClassifier
 {
   protected:
-    const char *submoduleName = nullptr;
-    cModuleType *moduleType = nullptr;
+    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> classIndexToGateItMap;
 
   protected:
     virtual void initialize(int stage) override;
+    virtual int getClassIndex(Packet *packet) const;
     virtual int classifyPacket(Packet *packet) override;
+    virtual int createGateForPacket(Packet *packet) override;
+    virtual bool canCreateGateForPacket(Packet *packet) const override;
+
+    virtual int createBranch();
+    virtual cGate *createModuleBranch(int index, cGate *classifierOutputGate, std::vector<cModule *>& modulesToInitialize);
+
+  public:
+    virtual bool canPushSomePacket(const cGate *gate) const override;
 };
 
 } // namespace queueing
 } // namespace inet
 
 #endif
-

src/inet/queueing/classifier/DynamicClassifier.ned

@@ -7,10 +7,26 @@
 
 package inet.queueing.classifier;
 
+//
+// A packet classifier that creates the branch for 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 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)
+        string submoduleName; // the submodule vector that holds the branches
+        string aggregatorSubmoduleName = default("multiplexer"); // downstream aggregator submodule to wire branches into
         @class(DynamicClassifier);
 }