fix is_uplink_burst. implement continous fragmentation rebuilding for…

… the uplink.

include/l2/upper_mac.hpp

@@ -66,12 +66,14 @@ class UpperMac {
     std::unique_ptr<UpperMacMetrics> metrics_;
 
     /// The prometheus metrics for the fragmentation
-    std::shared_ptr<UpperMacFragmentsPrometheusCounters> fragmentation_metrics_continous_;
-    std::shared_ptr<UpperMacFragmentsPrometheusCounters> fragmentation_metrics_stealing_channel_;
+    std::shared_ptr<UpperMacFragmentsPrometheusCounters> fragmentation_metrics_downlink_continous_;
+    std::shared_ptr<UpperMacFragmentsPrometheusCounters> fragmentation_metrics_uplink_continous_;
+    std::shared_ptr<UpperMacFragmentsPrometheusCounters> fragmentation_metrics_downlink_stealing_channel_;
 
     LogicalLinkControlParser logical_link_control_;
 
-    std::unique_ptr<UpperMacFragmentation> fragmentation_;
+    std::unique_ptr<UpperMacDownlinkFragmentation> downlink_fragmentation_;
+    std::unique_ptr<UpperMacUplinkFragmentation> uplink_fragmentation_;
 
     /// The worker thread
     std::thread worker_thread_;

include/l2/upper_mac_fragments.hpp

@@ -10,6 +10,7 @@
 
 #include "l2/upper_mac_packet.hpp"
 #include "prometheus.h"
+#include "utils/address.hpp"
 #include <cassert>
 #include <memory>
 #include <optional>
@@ -49,10 +50,8 @@ class UpperMacFragmentsPrometheusCounters {
     auto increment_fragment_count() -> void { fragment_count_total_.Increment(); }
 };
 
-/// Class that provides the fragment reconstruction for uplink and downlink packets.
-/// TODO: Uplink fragmentation may include reserved slots and is therefore harder to reconstruct. This is not handled
-/// with this class.
-class UpperMacFragmentation {
+/// Class that provides the fragment reconstruction for downlink packets.
+class UpperMacDownlinkFragmentation {
   private:
     /// Holds the internal state of the fragment rebuilder
     enum class State {
@@ -76,8 +75,8 @@ class UpperMacFragmentation {
     /// \param new_state the new state into which the state machine would be transfered with this fragment
     /// \param fragment the control plane signalling packet that is fragmented
     /// \return an optional reconstructed control plane signalling packet when reconstuction was successful
-    auto change_state(State new_state, const UpperMacCPlaneSignallingPacket& fragment)
-        -> std::optional<UpperMacCPlaneSignallingPacket> {
+    auto change_state(State new_state,
+                      const UpperMacCPlaneSignallingPacket& fragment) -> std::optional<UpperMacCPlaneSignallingPacket> {
         const auto& valid_state_changes = allowed_state_changes_[state_];
 
         // increment the total fragment counters
@@ -125,12 +124,12 @@ class UpperMacFragmentation {
     };
 
   public:
-    UpperMacFragmentation() = delete;
+    UpperMacDownlinkFragmentation() = delete;
 
     /// Constructor for the fragmentations. Optionally specify if an arbitraty numner of continuation fragments are
     /// allowed
-    explicit UpperMacFragmentation(const std::shared_ptr<UpperMacFragmentsPrometheusCounters>& metrics,
-                                   bool continuation_fragments_allowed = true)
+    explicit UpperMacDownlinkFragmentation(const std::shared_ptr<UpperMacFragmentsPrometheusCounters>& metrics,
+                                           bool continuation_fragments_allowed = true)
         : state_(State::kStart)
         , metrics_(metrics) {
         if (continuation_fragments_allowed) {
@@ -153,8 +152,8 @@ class UpperMacFragmentation {
     /// Push a fragment for reconstruction.
     /// \param fragment the control plane signalling packet that is fragmented
     /// \return an optional reconstructed control plane signalling packet when reconstuction was successful
-    auto push_fragment(const UpperMacCPlaneSignallingPacket& fragment)
-        -> std::optional<UpperMacCPlaneSignallingPacket> {
+    auto
+    push_fragment(const UpperMacCPlaneSignallingPacket& fragment) -> std::optional<UpperMacCPlaneSignallingPacket> {
         switch (fragment.type_) {
         case MacPacketType::kMacResource:
             assert(fragment.fragmentation_);
@@ -167,13 +166,142 @@ class UpperMacFragmentation {
             throw std::runtime_error("No fragmentation in MacDBlck");
         case MacPacketType::kMacBroadcast:
             throw std::runtime_error("No fragmentation in MacBroadcast");
+        case MacPacketType::kMacAccess:
+            throw std::runtime_error("MacAccess is not handled by UpperMacDownlinkFragmentation");
+        case MacPacketType::kMacData:
+            throw std::runtime_error("MacData is not handled by UpperMacDownlinkFragmentation");
+        case MacPacketType::kMacFragmentUplink:
+            throw std::runtime_error("MacFragmentUplink is not handled by UpperMacDownlinkFragmentation");
+        case MacPacketType::kMacEndHu:
+            throw std::runtime_error("MacEndHu is not handled by UpperMacDownlinkFragmentation");
+        case MacPacketType::kMacEndUplink:
+            throw std::runtime_error("MacEndUplink is not handled by UpperMacDownlinkFragmentation");
+        case MacPacketType::kMacUBlck:
+            throw std::runtime_error("No fragmentation in MacUBlck");
+        case MacPacketType::kMacUSignal:
+            throw std::runtime_error("No fragmentation in MacUSignal");
+        }
+    };
+};
+
+/// Class that provides the fragment reconstruction for uplink packets.
+/// Uplink fragmentation may include reserved slots and is therefore harder to reconstruct. This is not handled
+/// with this class.
+class UpperMacUplinkFragmentation {
+  private:
+    /// Holds the internal state of the fragment rebuilder
+    enum class State {
+        kStart,
+        kStartFragmentReceived,
+        kContinuationFragmentReceived,
+        kEndFragmentReceived,
+    };
+    /// The vector that holds the accumulated fragments for each mobile station by its address
+    std::map<Address, std::vector<UpperMacCPlaneSignallingPacket>> fragments_per_address_;
+    /// Are continuation allowed in the state machine?
+    std::map<State, std::set<State>> allowed_state_changes_;
+    /// The current state of the fragment reassembler for each mobile station by its address
+    std::map<Address, State> state_per_address_;
+
+    /// the metrics for the fragmentation
+    std::shared_ptr<UpperMacFragmentsPrometheusCounters> metrics_;
+
+    /// Try the state transtition with a fragment. Increment the error metrics if there is an invalid state transition
+    /// attempted
+    /// \param new_state the new state into which the state machine would be transfered with this fragment
+    /// \param fragment the control plane signalling packet that is fragmented
+    /// \return an optional reconstructed control plane signalling packet when reconstuction was successful
+    auto change_state(State new_state,
+                      const UpperMacCPlaneSignallingPacket& fragment) -> std::optional<UpperMacCPlaneSignallingPacket> {
+        const auto& address = fragment.address_;
+        auto& state = state_per_address_[address];
+        auto& fragments = fragments_per_address_[address];
+
+        const auto& valid_state_changes = allowed_state_changes_[state];
+
+        // increment the total fragment counters
+        if (metrics_) {
+            metrics_->increment_fragment_count();
+        }
+
+        if (valid_state_changes.count(new_state)) {
+            // valid state change. perform and add fragment
+            fragments.emplace_back(fragment);
+            state = new_state;
+        } else {
+            // increment the invalid state metrics
+            if (metrics_) {
+                metrics_->increment_fragment_reconstruction_error();
+            }
+
+            // always save the start segment
+            if (new_state == State::kStartFragmentReceived) {
+                fragments = {fragment};
+                state = State::kStartFragmentReceived;
+            } else {
+                fragments.clear();
+                state = State::kStart;
+            }
+        }
+
+        // if we are in the end state reassmeble the packet.
+        if (state == State::kEndFragmentReceived) {
+            std::optional<UpperMacCPlaneSignallingPacket> packet;
+            for (const auto& fragment : fragments) {
+                if (packet) {
+                    packet->tm_sdu_->append(*fragment.tm_sdu_);
+                } else {
+                    packet = fragment;
+                }
+            }
+            fragments.clear();
+            state = State::kStart;
+
+            return packet;
+        }
+
+        return std::nullopt;
+    };
+
+  public:
+    UpperMacUplinkFragmentation() = delete;
+
+    /// Constructor for the fragmentations. Optionally specify if an arbitraty numner of continuation fragments are
+    /// allowed
+    explicit UpperMacUplinkFragmentation(const std::shared_ptr<UpperMacFragmentsPrometheusCounters>& metrics)
+        : metrics_(metrics) {
+        allowed_state_changes_ = {
+            {State::kStart, {State::kStartFragmentReceived}},
+            {State::kStartFragmentReceived, {State::kContinuationFragmentReceived, State::kEndFragmentReceived}},
+            {State::kContinuationFragmentReceived, {State::kContinuationFragmentReceived, State::kEndFragmentReceived}},
+            {State::kEndFragmentReceived, {State::kStart}}};
+    };
+
+    /// Push a fragment for reconstruction.
+    /// \param fragment the control plane signalling packet that is fragmented
+    /// \return an optional reconstructed control plane signalling packet when reconstuction was successful
+    auto
+    push_fragment(const UpperMacCPlaneSignallingPacket& fragment) -> std::optional<UpperMacCPlaneSignallingPacket> {
+        switch (fragment.type_) {
+        case MacPacketType::kMacResource:
+            throw std::runtime_error("MacResource is not handled by UpperMacUplinkFragmentation");
+        case MacPacketType::kMacFragmentDownlink:
+            throw std::runtime_error("MacFragmentDownlink is not handled by UpperMacUplinkFragmentation");
+        case MacPacketType::kMacEndDownlink:
+            throw std::runtime_error("MacEndDownlink is not handled by UpperMacUplinkFragmentation");
+        case MacPacketType::kMacDBlck:
+            throw std::runtime_error("No fragmentation in MacDBlck");
+        case MacPacketType::kMacBroadcast:
+            throw std::runtime_error("No fragmentation in MacBroadcast");
         case MacPacketType::kMacAccess:
         case MacPacketType::kMacData:
             assert(fragment.fragmentation_);
             return change_state(State::kStartFragmentReceived, fragment);
         case MacPacketType::kMacFragmentUplink:
             return change_state(State::kContinuationFragmentReceived, fragment);
         case MacPacketType::kMacEndHu:
+            throw std::runtime_error("MacEndHu is in a reserverd subslot and not handled since there is no "
+                                     "integration between the uplink and downlink processing");
         case MacPacketType::kMacEndUplink:
             return change_state(State::kEndFragmentReceived, fragment);
         case MacPacketType::kMacUBlck:

include/l2/upper_mac_packet.hpp

@@ -462,7 +462,8 @@ struct UpperMacCPlaneSignallingPacket {
     [[nodiscard]] auto is_uplink_fragment() const -> bool {
         return (type_ == MacPacketType::kMacData && fragmentation_) ||
                (type_ == MacPacketType::kMacData && fragmentation_on_stealling_channel_) ||
-               (type_ == MacPacketType::kMacFragmentUplink) || (type_ == MacPacketType::kMacEndUplink);
+               (type_ == MacPacketType::kMacFragmentUplink) || (type_ == MacPacketType::kMacEndUplink) ||
+               (type_ == MacPacketType::kMacEndHu);
     };
 
     /// check if this packet is sent on downlink

include/utils/address.hpp

@@ -85,6 +85,13 @@ class Address {
         }
     }
 
+    // Overload this operator for usage of the Address as a map key
+    auto operator<(const Address& other) const -> bool {
+        return std::tie(country_code_, network_code_, sna_, ssi_, event_label_, ussi_, smi_, usage_marker_) <
+               std::tie(other.country_code_, other.network_code_, other.sna_, other.ssi_, other.event_label_,
+                        other.ussi_, other.smi_, other.usage_marker_);
+    }
+
     friend auto operator<<(std::ostream& stream, const Address& address_type) -> std::ostream&;
 
     NLOHMANN_DEFINE_TYPE_INTRUSIVE(Address, country_code_, network_code_, sna_, ssi_, event_label_, ussi_, smi_,

src/l2/upper_mac.cpp

@@ -28,12 +28,17 @@ UpperMac::UpperMac(const std::shared_ptr<StreamingOrderedOutputThreadPoolExecuto
     , logical_link_control_(prometheus_exporter) {
     if (prometheus_exporter) {
         metrics_ = std::make_unique<UpperMacMetrics>(prometheus_exporter);
-        fragmentation_metrics_continous_ =
-            std::make_shared<UpperMacFragmentsPrometheusCounters>(prometheus_exporter, "Continous");
-        fragmentation_metrics_stealing_channel_ =
-            std::make_shared<UpperMacFragmentsPrometheusCounters>(prometheus_exporter, "Stealing Channel");
+        fragmentation_metrics_downlink_continous_ =
+            std::make_shared<UpperMacFragmentsPrometheusCounters>(prometheus_exporter, "Continous Downlink");
+        fragmentation_metrics_uplink_continous_ =
+            std::make_shared<UpperMacFragmentsPrometheusCounters>(prometheus_exporter, "Continous Uplink");
+        fragmentation_metrics_downlink_stealing_channel_ =
+            std::make_shared<UpperMacFragmentsPrometheusCounters>(prometheus_exporter, "Stealing Channel Downlink");
     }
-    fragmentation_ = std::make_unique<UpperMacFragmentation>(fragmentation_metrics_continous_);
+    downlink_fragmentation_ =
+        std::make_unique<UpperMacDownlinkFragmentation>(fragmentation_metrics_downlink_continous_);
+    uplink_fragmentation_ = std::make_unique<UpperMacUplinkFragmentation>(fragmentation_metrics_uplink_continous_);
+
     worker_thread_ = std::thread(&UpperMac::worker, this);
 
 #if defined(__linux__)
@@ -110,9 +115,10 @@ auto UpperMac::process(const Slots& slots) -> void {
 
 auto UpperMac::processPackets(UpperMacPackets&& packets) -> void {
     // the fragmentation reconstructor for over two stealing channel in the same burst
-    auto& fragmentation = *fragmentation_;
+    auto& downlink_fragmentation = *downlink_fragmentation_;
     auto stealling_channel_fragmentation =
-        UpperMacFragmentation(fragmentation_metrics_stealing_channel_, /*continuation_fragments_allowed=*/false);
+        UpperMacDownlinkFragmentation(fragmentation_metrics_downlink_stealing_channel_,
+                                      /*continuation_fragments_allowed=*/false);
 
     std::vector<UpperMacCPlaneSignallingPacket> c_plane_packets;
 
@@ -122,13 +128,18 @@ auto UpperMac::processPackets(UpperMacPackets&& packets) -> void {
             metrics_->increment_c_plane_packet_counters(packet);
         }
 
-        if (packet.is_downlink_fragment() || packet.is_uplink_fragment()) {
+        if (packet.is_downlink_fragment()) {
             /// populate the fragmenter for stealing channel
             if (packet.fragmentation_on_stealling_channel_) {
-                fragmentation = stealling_channel_fragmentation;
+                downlink_fragmentation = stealling_channel_fragmentation;
             }
 
-            auto reconstructed_fragment = fragmentation.push_fragment(packet);
+            auto reconstructed_fragment = downlink_fragmentation.push_fragment(packet);
+            if (reconstructed_fragment) {
+                c_plane_packets.emplace_back(std::move(*reconstructed_fragment));
+            }
+        } else if (packet.is_uplink_fragment()) {
+            auto reconstructed_fragment = uplink_fragmentation_->push_fragment(packet);
             if (reconstructed_fragment) {
                 c_plane_packets.emplace_back(std::move(*reconstructed_fragment));
             }
@@ -138,8 +149,8 @@ auto UpperMac::processPackets(UpperMacPackets&& packets) -> void {
     }
 
     /// increment the reconstruction error counter if we could not complete the fragmentation over stealing channel
-    if (!stealling_channel_fragmentation.is_in_start_state() && fragmentation_metrics_stealing_channel_) {
-        fragmentation_metrics_stealing_channel_->increment_fragment_reconstruction_error();
+    if (!stealling_channel_fragmentation.is_in_start_state() && fragmentation_metrics_downlink_stealing_channel_) {
+        fragmentation_metrics_downlink_stealing_channel_->increment_fragment_reconstruction_error();
     }
 
     /// increment the packet counter