P0: opt-in negotiated MMS NameList wire pipeline with exact response routing

include/ariec61850/mms/association_runtime.hpp

@@ -154,6 +154,13 @@ struct MmsConfirmedExchangeResult final {
     std::vector<std::uint8_t> presentation_payload;
 };
 
+// Caller supplies complete encoded P-DATA. Invoke IDs are validated against
+// the encoded requests before any bytes are sent to the association.
+struct MmsConfirmedBatchRequest final {
+    std::vector<std::uint8_t> presentation_payload;
+    std::uint32_t invoke_id{};
+};
+
 class MmsAssociationRuntime final {
 public:
     explicit MmsAssociationRuntime(
@@ -182,6 +189,22 @@ class MmsAssociationRuntime final {
         // A response timeout after a successful send still leaves this true.
         bool* request_sent = nullptr);
 
+    // Association-owned, synchronous *wire pipeline*, not a thread pool.
+    // Sends complete requests serially, then receives and dispatches responses
+    // by exact invoke-ID. Must never overlap poll/receive/exchange calls.
+    // Only available after a strict, positive negotiated calling window.
+    // sent_count records completed transmissions even when a later send/recv
+    // fails. The association is faulted/closed on any in-flight failure.
+    [[nodiscard]] std::vector<MmsConfirmedExchangeResult> exchange_confirmed_batch(
+        std::span<const MmsConfirmedBatchRequest> requests,
+        std::stop_token stop_token = {},
+        std::size_t* sent_count = nullptr);
+
+    [[nodiscard]] std::size_t verified_outstanding_calling() const noexcept {
+        return negotiated_outstanding_verified_
+            ? negotiated_.maximum_outstanding_calling : 1U;
+    }
+
     [[nodiscard]] MmsPduEnvelope poll_once(
         std::stop_token stop_token = {});
 
@@ -256,6 +279,8 @@ class MmsAssociationRuntime final {
     MmsAssociationRuntimeState state_{MmsAssociationRuntimeState::disconnected};
     MmsEndpoint endpoint_;
     MmsNegotiatedAssociation negotiated_;
+    // Unknown or tolerant Initiate-Response is strictly serial.
+    bool negotiated_outstanding_verified_{};
     MmsInvokeRouter invoke_router_;
     osi::TpktStreamDecoder tpkt_decoder_;
     std::deque<std::vector<std::uint8_t>> information_reports_;

include/ariec61850/mms/live_discovery.hpp

@@ -25,6 +25,11 @@ struct MmsLiveDiscoveryOptions final {
     std::size_t maximum_domains{4'096U};
     std::size_t maximum_names_per_domain{65'536U};
 
+    // Opt-in association-owned GetNameList wire pipelining (not threads).
+    // Default 1 preserves legacy serial order and behavior. Effective window
+    // is min(requested, verified peer calling limit, hard cap of 8).
+    std::size_t max_concurrent_name_list_chains{1U};
+
     bool probe_variable_types{true};
     std::size_t maximum_variable_type_probes{4'096U};
 
@@ -77,6 +82,8 @@ struct MmsReportControlEvidence final {
 // count. Pure accounting: no extra MMS service or scheduling side effects.
 struct MmsLiveDiscoveryWireBudget final {
     std::size_t get_name_list{};
+    // Observed send window, not a requested/conjectured concurrency value.
+    std::size_t name_list_peak_outstanding{};
     std::size_t get_variable_access_attributes{};
     std::size_t get_named_variable_list_attributes{};
     std::size_t read_report_controls{};
@@ -135,6 +142,13 @@ class MmsLiveDiscoveryClient final {
         MmsLiveDiscoveryWireBudget& wire_budget,
         std::stop_token stop_token);
 
+    void get_name_lists_pipelined(
+        const std::vector<std::string>& domains,
+        const MmsLiveDiscoveryOptions& options,
+        MmsLiveDiscoveryResult& result,
+        std::stop_token stop_token,
+        std::size_t window);
+
     MmsAssociationRuntime& association_;
 };
 

src/mms/association_runtime.cpp

@@ -11,6 +11,7 @@
 #include <array>
 #include <exception>
 #include <limits>
+#include <unordered_map>
 #include <utility>
 
 namespace ar::iec61850::mms {
@@ -391,6 +392,7 @@ void MmsAssociationRuntime::connect(
     information_reports_.clear();
     tpkt_decoder_.reset();
     next_invoke_id_ = 0U;
+    negotiated_outstanding_verified_ = false;
     remote_cotp_reference_ = 0U;
     negotiated_ = {};
     negotiated_.tpdu_size_code = options_.tpdu_size_code;
@@ -413,6 +415,7 @@ void MmsAssociationRuntime::connect(
             information_reports_.clear();
             tpkt_decoder_.reset();
             remote_cotp_reference_ = 0U;
+            negotiated_outstanding_verified_ = false;
             negotiated_ = {};
             negotiated_.tpdu_size_code = options_.tpdu_size_code;
             negotiated_.presentation_context_id = options_.presentation_context_id;
@@ -531,6 +534,8 @@ void MmsAssociationRuntime::connect(
                             initiate.negotiated_maximum_outstanding_called;
                         negotiated_.data_structure_nesting_level =
                             initiate.negotiated_data_structure_nesting_level;
+                        negotiated_outstanding_verified_ =
+                            initiate.negotiated_maximum_outstanding_calling > 0U;
                     } catch (const std::exception& exception) {
                         strict_failure =
                             std::string{"Strict MMS Initiate-Response decode failed: "} +
@@ -555,6 +560,7 @@ void MmsAssociationRuntime::connect(
                     throw MmsAssociationRuntimeError(strict_failure);
                 }
                 used_csharp_compatibility = true;
+                negotiated_outstanding_verified_ = false;
                 acceptance_message =
                     profile.name + ": " + compatibility_message +
                     (strict_failure.empty()
@@ -764,6 +770,122 @@ MmsConfirmedExchangeResult MmsAssociationRuntime::exchange_confirmed(
     }
 }
 
+std::vector<MmsConfirmedExchangeResult>
+MmsAssociationRuntime::exchange_confirmed_batch(
+    const std::span<const MmsConfirmedBatchRequest> requests,
+    const std::stop_token stop_token,
+    std::size_t* const sent_count) {
+    if (sent_count) *sent_count = 0U;
+    require_associated();
+    // Bound the collection independently of server claims. Caller must opt in.
+    constexpr std::size_t hard_window = 8U;
+    const auto window = std::min<std::size_t>(
+        hard_window, verified_outstanding_calling());
+    if (requests.empty() || requests.size() > window) {
+        throw std::invalid_argument(
+            "Confirmed MMS batch exceeds verified negotiated calling window.");
+    }
+    std::unordered_map<std::uint32_t, std::size_t> slots;
+    slots.reserve(requests.size());
+    for (std::size_t i = 0U; i < requests.size(); ++i) {
+        const auto& req = requests[i];
+        if (req.invoke_id == 0U ||
+            req.invoke_id > MmsPduCodec::maximum_invoke_id ||
+            !slots.emplace(req.invoke_id, i).second) {
+            throw std::invalid_argument(
+                "Confirmed MMS batch contains invalid or duplicate invoke ID.");
+        }
+        const auto pdu = MmsPduCodec::decode_envelope(req.presentation_payload);
+        if (pdu.kind != MmsPduKind::confirmed_request ||
+            !pdu.matches_invoke(req.invoke_id)) {
+            throw std::invalid_argument(
+                "Confirmed MMS batch PDU does not match its invoke ID.");
+        }
+    }
+
+    const auto deadline = deadline_after(options_.request_timeout);
+    try {
+        if (invoke_router_.queued_confirmed_count() != 0U) {
+            throw MmsAssociationRuntimeError(
+                "Orphan confirmed response before MMS batch send.");
+        }
+        for (const auto& req : requests) {
+            require_not_cancelled(stop_token);
+            send_application_payload(req.presentation_payload, deadline, stop_token);
+            if (sent_count) ++*sent_count;
+            add_event(MmsAssociationEventKind::confirmed_request_sent,
+                      "Batched confirmed MMS request sent.", req.invoke_id);
+        }
+
+        std::vector<std::optional<MmsConfirmedExchangeResult>> completed(requests.size());
+        std::size_t remaining = requests.size();
+        for (std::size_t received_count = 0U;
+             received_count < options_.maximum_receive_chunks_per_operation;
+             ++received_count) {
+            auto bytes = receive_application_payload(deadline, stop_token);
+            auto envelope = route_received_payload(bytes);
+            if (!envelope.confirmed_result()) continue; // InformationReport
+            const auto found = envelope.invoke_id
+                ? slots.find(*envelope.invoke_id) : slots.end();
+            if (found == slots.end() || completed[found->second].has_value()) {
+                throw MmsAssociationRuntimeError(
+                    "Unregistered or duplicate confirmed MMS batch invoke ID.");
+            }
+            MmsPduEnvelope matched;
+            if (!invoke_router_.try_dequeue(*envelope.invoke_id, matched)) {
+                throw MmsAssociationRuntimeError(
+                    "Batch response absent from authoritative invoke router.");
+            }
+            completed[found->second] = MmsConfirmedExchangeResult{
+                std::move(matched), std::move(bytes)};
+            add_event(MmsAssociationEventKind::confirmed_result_received,
+                      "Batched confirmed MMS result received.", *envelope.invoke_id);
+            if (--remaining == 0U) {
+                std::vector<MmsConfirmedExchangeResult> results;
+                results.reserve(completed.size());
+                for (auto& item : completed) results.push_back(std::move(*item));
+                return results;
+            }
+        }
+        throw MmsAssociationRuntimeError(
+            "MMS batch exhausted bounded receive iteration budget.");
+    } catch (const MmsTransportCancelledError& error) {
+        state_ = MmsAssociationRuntimeState::faulted;
+        last_fault_ = error.what();
+        add_event(MmsAssociationEventKind::cancelled, error.what());
+        transport_.close();
+        invoke_router_.clear();
+        information_reports_.clear();
+        tpkt_decoder_.reset();
+        remote_cotp_reference_ = 0U;
+        throw;
+    } catch (const MmsTransportTimeoutError& error) {
+        state_ = MmsAssociationRuntimeState::faulted;
+        last_fault_ = error.what();
+        add_event(MmsAssociationEventKind::timed_out, error.what());
+        transport_.close();
+        invoke_router_.clear();
+        information_reports_.clear();
+        tpkt_decoder_.reset();
+        remote_cotp_reference_ = 0U;
+        throw;
+    } catch (const std::exception& error) {
+        if (stop_token.stop_requested()) {
+            state_ = MmsAssociationRuntimeState::faulted;
+            last_fault_ = error.what();
+            add_event(MmsAssociationEventKind::cancelled, error.what());
+        } else {
+            fail(error.what());
+        }
+        transport_.close();
+        invoke_router_.clear();
+        information_reports_.clear();
+        tpkt_decoder_.reset();
+        remote_cotp_reference_ = 0U;
+        throw;
+    }
+}
+
 MmsPduEnvelope MmsAssociationRuntime::poll_once(
     const std::stop_token stop_token) {
     require_associated();