From c005469c392552bfe24400623e4cda181d8c93ef Mon Sep 17 00:00:00 2001 From: "J. Ritchie Carroll" Date: Tue, 15 Sep 2026 14:34:40 -0700 Subject: [PATCH] Add opt-in STTP startup diagnostics and benchmarks --- docs/MetadataScaling.md | 56 +++ docs/StartupDiagnostics.md | 111 ++++++ src/lib/transport/DataSubscriber.cpp | 16 + src/lib/transport/StartupTrace.h | 85 +++++ src/lib/transport/SubscriberInstance.cpp | 49 +++ .../MetadataBenchmark/MetadataBenchmark.cpp | 340 ++++++++++++++++++ src/samples/MetadataBenchmark/README.md | 27 ++ src/samples/MetadataBenchmark/build.cmd | 20 ++ src/samples/MetadataBenchmark/run-suite.ps1 | 54 +++ src/samples/StartupTiming/README.md | 30 ++ src/samples/StartupTiming/StartupTiming.cpp | 126 +++++++ src/samples/StartupTiming/build.cmd | 19 + 12 files changed, 933 insertions(+) create mode 100644 docs/MetadataScaling.md create mode 100644 docs/StartupDiagnostics.md create mode 100644 src/lib/transport/StartupTrace.h create mode 100644 src/samples/MetadataBenchmark/MetadataBenchmark.cpp create mode 100644 src/samples/MetadataBenchmark/README.md create mode 100644 src/samples/MetadataBenchmark/build.cmd create mode 100644 src/samples/MetadataBenchmark/run-suite.ps1 create mode 100644 src/samples/StartupTiming/README.md create mode 100644 src/samples/StartupTiming/StartupTiming.cpp create mode 100644 src/samples/StartupTiming/build.cmd diff --git a/docs/MetadataScaling.md b/docs/MetadataScaling.md new file mode 100644 index 0000000..d8b96b3 --- /dev/null +++ b/docs/MetadataScaling.md @@ -0,0 +1,56 @@ +# Metadata scaling and 32-bit indexes + +## Local reproduction + +Windows x64 Release, MSVC v145, Boost 1.92. The benchmark generates deterministic metadata and passes it through the actual SubscriberInstance metadata parser and configuration builder. Its timer excludes XML generation, compression, network transfer, and post-parse validation. + +480,000 analog measurements concentrated on eight devices (60,000 each), 185,450,687 XML bytes: + +| Stage | Original implementation | Indexed frame construction | +|---|---:|---:| +| Complete metadata processing | 107.081 s | 10.947 s | +| Configuration frames | 92.934 s | 0.348 s | +| XML parsing | 0.504 s | 0.369 s | + +A final run after widening metadata indexes completed in **9.783 s**, with all output validation passing. These are individual runs and vary with machine load. Three alternating 50,000-point comparisons gave median totals of 5.479 s original and 1.211 s optimized. + +This reproduces a substantial client-side delay locally and identifies configuration construction as the dominant cost for this shape. It does not establish the cause of the client's reported 2.5-minute connection delay: their metadata distribution, server work, transfer, and hardware remain unmeasured. + +## Fix + +ConstructConfigurationFrames previously scanned a device's measurements for every analog/digital index and scanned its phasors for every phasor index. It now builds lookup vectors once per device. Ordering, missing-index placeholders, and first-match handling of duplicate indexes are preserved. + +SignalReference.Index, MeasurementMetadata.PhasorSourceIndex, and PhasorMetadata.SourceIndex now use signed int32_t. Parsing, helper signatures, publisher phasor maps, and metadata filtering preserve these values. Frame loops use int64_t counters so neither 65,535 nor INT32_MAX causes counter wraparound. The aggregate phasor count uses size_t. + +Runtime signal-cache and compact-measurement indexes were already int32_t; their wire layout is unchanged. Cache comments have been corrected accordingly. + +## Validation + +The native benchmark regression mode checks: + +- Dense and sparse analog, digital, and phasor metadata, including gzip input. +- Indexes 65,535, 65,536, and 100,000 for each kind; placeholder positions and signal identities. +- Duplicate indexes retain the original first-match behavior. +- A complete 70,000-measurement metadata set. +- A 70,001-entry wire cache and compact-measurement round trips, including INT32_MAX. +- Publisher metadata round trips for indexes 34,464 and 100,000, which collided when narrowed to 16 bits. +- TSSC compressed round trips through indexes 65,535, 65,536, and 100,000. + +Build and execution instructions: [MetadataBenchmark](../src/samples/MetadataBenchmark/README.md). + +Local evidence is under build/startup-diagnostics (ignored build artifacts): baseline-summary.txt, optimized-summary.txt, massive-baseline.out/.log, massive-optimized.out/.log, massive-int32.out/.log, repeat-baseline-*.out/.log, repeat-optimized-*.out/.log, and int32-regressions.out/.log. MetadataBenchmark-baseline.exe preserves the original frame algorithm for local comparisons. + +## Limits and integration + +- Configuration frames still allocate placeholders through the maximum index. A sparse enormous index can exhaust memory; signed 32-bit fields do not make billions of frame slots practical. INT32_MAX was tested in parsing, cache, and compact records, not by allocating that many configuration slots or TSSC states. +- Per-phasor angle/magnitude association still scans device measurements and can become costly on a device with many phasors. This optimization addresses configuration-frame construction. +- The separately observed duplicate connection initialization remains unchanged; see StartupDiagnostics.md. +- These changes alter the public native metadata ABI. Rebuild all native consumers. Before using this library with net-cppapi, update its three SWIG field declarations, regenerate bindings, and rebuild both managed and native wrapper components together. Existing .NET binaries have not been replaced. +- Boost compatibility uses explicit version checks; pre-1.66 branches were not compiled in this environment. + +## Final native live check + +After the performance and int32 changes, the rebuilt StartupTiming.exe connected to openHistorian at 127.0.0.1:7175 with metadata enabled. First measurement arrived 143.550 ms after connection; the five-second run received 19,860 measurements at approximately 4,100/sec and exited successfully. Evidence: build/startup-diagnostics/final-live.out and final-live.log. This checks compatibility with the local publisher, whose small metadata set does not exercise the large-index cases covered by the synthetic regressions. + +Production correctness tests are also available in [ConfigurationFramesTest](../src/samples/ConfigurationFramesTest/README.md), independently of this diagnostics branch. + diff --git a/docs/StartupDiagnostics.md b/docs/StartupDiagnostics.md new file mode 100644 index 0000000..b4b94b3 --- /dev/null +++ b/docs/StartupDiagnostics.md @@ -0,0 +1,111 @@ +# Diagnosing connection-to-first-data delays + +## Branch scope + +This diagnostics branch is based on production commit 5a5edac. Performance and int32 fixes, together with samples/ConfigurationFramesTest, are already in main. This branch adds opt-in tracing, StartupTiming, MetadataBenchmark, and investigation notes. Measurements below were collected during the original investigation unless explicitly marked as branch validation. + + +## Findings from source inspection + +The default SubscriberInstance path is sequential: + +connection ready -> request metadata -> receive entire response -> callback queue -> decompress -> parse XML -> construct device/measurement/phasor objects -> construct configuration frames -> install maps -> ParsedMetadata callback -> destroy temporary metadata structures -> send Subscribe -> publisher processes subscription -> receive/decode signal index cache -> receive/decode data -> measurement callback. + +`SubscriberInstance::HandleMetadata` waits for `ReceivedMetadata` to return before calling `Subscribe`. The subscription filter does not limit the metadata request: that uses the separate `MetadataFilters` setting. A filter excluding STAT therefore still downloads/processes the default metadata set. + +### Findings in the original implementation + +1. **Configuration-frame construction.** `ConstructConfigurationFrames` calls `TryFindMeasurement` (a linear scan of the device's measurement vector) for every analog/digital index. `GetSignalKindCount` actually returns the maximum signal-reference index, not the number of records. Cost grows roughly with `measurements-per-device * (maximum-analog-index + maximum-digital-index)`. Dense 50,000 analogs on one device can mean roughly 1.25 billion comparisons. Sparse high indexes also generate placeholder records. The phasor configuration loop repeatedly scans the phasor list. This is per-device scaling, not automatically quadratic in the total metadata count across small devices. +2. **Phasor association.** For each phasor, ReceivedMetadata scans the associated device's measurements to locate angle/magnitude. Approximately `sum(phasors-per-device * measurements-per-device)` in the worst case. +3. **Metadata materialization and memory pressure.** Every record creates objects/strings and parses UpdatedOn using string processing and locale-based streams. The metadata maps are copied into member maps; temporary maps, XML storage, and the expanded payload are then destroyed before Subscribe. Raw XML parsing is only one part of this work. +4. **Transfer/decompression and signal cache.** Metadata and cache decompression use CopyStream, which appends bytes one at a time to a vector. Cache Decode performs per-record allocations/hash insertions, with no pre-reservation. These deserve measurement, but do not show the repeated whole-device scans above. + +The client Subscribe method builds and sends a small connection string containing the filter; it does not expand the filter to every measurement locally. The publisher performs selection and subscription setup. A long gap after sending Subscribe and before receiving its response/cache points toward publisher work or transfer, rather than this client's filter evaluation. The C++ publisher code does not establish what openHistorian's publisher spends time doing. + +**Fixed:** configuration-frame construction now builds per-device index lookups instead of repeatedly scanning each device. Metadata indexes and publisher mappings use int32_t; frame loops use int64_t counters to avoid wraparound. See [MetadataScaling.md](MetadataScaling.md) for the local reproduction, measurements, and boundary tests. Phasor association remains a separate potential hotspot. + +## Opt-in instrumentation + +Rebuild the native C++ library/client with the modified sources, then launch from PowerShell: + +```powershell +$env:STTP_STARTUP_TRACE = '1' +# Run your rebuilt native client with its normal IP/port arguments, redirecting stderr: +# .\YourClient.exe 127.0.0.1 7175 2> startup-trace.log +``` + +The metadata field changes alter the native ABI. Update the three corresponding SWIG field declarations to int32_t and regenerate the bindings before rebuilding the wrapper. For the .NET wrapper, rebuild/relink `sttp.net.lib.dll` against these instrumented sources and redeploy it before testing ConnectionTest. Merely republishing the .NET executable does not incorporate these C++ changes. The old prebuilt native DLL will not emit these diagnostics. + +Unset STTP_STARTUP_TRACE or set it to 0 to disable tracing. The setting is read once per process. Events go directly to stderr and are flushed; they are independent of StatusMessage overrides and the callback queue. No measurement values or metadata content are logged. Before constructing configuration frames, a diagnostic-only scan reports maximum per-device measurement/phasor counts and maximum analog/digital/phasor reference indexes. Its duration is reported separately. + +Log format contains monotonic milliseconds since the first trace event, native DataSubscriber pointer (to separate connections), stage duration, and item/byte count where applicable. Scope labels appear on entry and exit; marks report elapsed time since the previous mark. Logs can interleave across threads, so compare timestamps and subscriber pointers. Diagnostic I/O adds some overhead; redirect to a local file. + +| Compare stages | What the interval covers | +|---|---| +| metadata request sending -> metadata response received | Request sending, publisher preparation, response transfer | +| metadata response received -> response queued -> metadata callback scope entry | Payload copying and callback scheduling; queue and callback may overlap | +| metadata decompress/copy complete | Gzip expansion or uncompressed payload copy | +| XML parse complete | pugixml load_buffer_inplace | +| device / measurement objects complete | Record conversion, timestamps, strings and map insertion | +| phasor objects and measurement matching complete | Phasor record conversion and association scans | +| configuration frames complete | Indexed lookups and placeholder construction | +| metadata maps installed | Configuration lock, map copying/replacement | +| ParsedMetadata callback complete -> metadata processing scope exit | Temporary metadata/XML/buffer destruction | +| subscribe command built -> subscribe send returned | Local command construction/send; not publisher completion | +| subscribe send -> acknowledged / signal cache scope entry | Publisher processing and transfer; response ordering may vary | +| cache decompress/copy -> cache decode complete | Cache expansion and decoding | +| first nonempty packet received -> first packet decoded | First packet's decode before measurement callback | + +The first-packet marker uses the existing per-subscription receive counter. It is a packet diagnostic, not a guarantee of successful measurements if the packet cannot be decoded or its cache is absent. + +## Quick comparison test + +Use the same endpoint/filter with one counting subscriber, once with default metadata parsing and once with: + +```cpp +subscriber.SetAutoParseMetadata(false); // before Connect/ConnectAsync +``` + +This skips both metadata retrieval and metadata construction and immediately subscribes with the configured filter. Signal-index-cache decoding is still required. It is suitable for the counting test; code that depends on parsed device/configuration metadata will lose that metadata. A large improvement isolates the metadata path, but does not distinguish server metadata preparation/transfer from client parsing; use the stage timings for that distinction. + +Then compare the same subscriber with a small filter versus the full filter while metadata is disabled to assess publisher subscription/cache scaling. Keep compression, endpoint and build configuration fixed. Use Release builds for representative performance and record metadata size, per-device point counts and maximum analog/digital/phasor indexes. + +## Native build and measured results + +The Windows x64 Release diagnostic client is now available at `build/startup-diagnostics/StartupTiming.exe`. Build and run instructions are in `src/samples/StartupTiming/README.md`. Its `--no-metadata` option runs the comparison above without changing the filter. The native library and client were successfully compiled using MSVC v145 and Boost 1.92. + +Required compatibility changes: use Boost's built-in UUID std::hash from 1.86 onward, use modern Asio resolver/address APIs with explicit Boost version checks, and explicitly include chrono in ANTLR's profiling source. The configuration-frame optimization is described in MetadataScaling.md. The pre-1.66 Boost branches have not been compiled in this environment. + +Initial five-second tests against openHistorian at 127.0.0.1:7175: + +| Mode | Connection to first measurement | Received | Exit code | +|---|---:|---:|---:| +| Metadata enabled | 241.080 ms | 19,584 | 0 | +| Metadata disabled | 35.175 ms | 20,402 | 0 | + +Both streamed approximately 4,100 measurements/sec. These are individual diagnostic runs, not benchmark averages. Raw traces and console output are in `build/startup-diagnostics/metadata.log`, `metadata-output.log`, `no-metadata.log`, and `no-metadata-output.log`. + +For the first metadata response (50,802 compressed bytes, 448,356 expanded bytes, 747 measurement records across 5 devices), the timings were approximately: + +- Metadata request to complete response: 116 ms. +- Decompression/copy: 14.2 ms. +- XML parsing: 0.8 ms. +- Measurement-object construction: 15.6 ms. +- Phasor conversion/matching: 3.6 ms. +- Configuration frames: 0.23 ms. +- Signal cache decompression/decoding: approximately 1 ms combined. + +This small metadata set does not reproduce the large-device configuration scaling hypothesis. Its maximum per-device measurement count was 169, with maximum analog/digital/phasor indexes of 14/2/59. A trace from the slow client is still needed to locate the 2.5-minute delay. + +### Additional observed defect: duplicate startup requests + +The trace shows two `connection ready` events, two metadata requests, and two metadata responses/subscriptions for one connection. With metadata disabled it shows two immediate Subscribe commands. Source inspection confirms that `SubscriberInstance::Connect()` calls `ConnectionEstablished()` and `HandleConnect()`, and the registered `HandleConnectionEstablished()` callback also calls them. This duplicates startup work and can trigger a resubscription while the first stream is starting. + +The diagnostic patch deliberately retains this behavior so it can be measured. This finding is specific to the current C++ checkout; it does not prove that the older prebuilt SWIG DLL has the same behavior. The diagnostic client retains the earliest connection callback timestamp so duplicate notifications do not reset its timer. + + + + +## Diagnostics branch validation + +Rebuilt on codex/startup-diagnostics, based on production commit 5a5edac. With tracing enabled, the local openHistorian check at 127.0.0.1:7175 received first data in 171.366 ms and 19,856 measurements over the five-second run (roughly 4,100/sec). Output and trace: build/review-stages/04-live.out and 04-live.log. Timings are individual runs, not benchmark averages. diff --git a/src/lib/transport/DataSubscriber.cpp b/src/lib/transport/DataSubscriber.cpp index 9821f71..1a74700 100644 --- a/src/lib/transport/DataSubscriber.cpp +++ b/src/lib/transport/DataSubscriber.cpp @@ -24,6 +24,7 @@ //****************************************************************************************************** #include "DataSubscriber.h" +#include "StartupTrace.h" #include "Constants.h" #include "CompactMeasurement.h" #include "../Convert.h" @@ -386,6 +387,8 @@ void DataSubscriber::HandleSucceeded(const uint8_t commandCode, uint8_t* data, c // Do not break on these messages because there is // still an associated message to be processed. m_subscribed = (commandCode == ServerCommand::Subscribe); + if (m_subscribed) + diagnostics::StartupEvent(this, "subscribe acknowledged"); [[fallthrough]]; case ServerCommand::UpdateProcessingInterval: case ServerCommand::RotateCipherKeys: @@ -435,7 +438,9 @@ void DataSubscriber::HandleFailed(const uint8_t commandCode, uint8_t* data, cons // Handles metadata refresh messages from the server. void DataSubscriber::HandleMetadataRefresh(const uint8_t* data, const uint32_t offset, const uint32_t length) { + diagnostics::StartupEvent(this, "metadata response received (bytes)", 0.0, length); Dispatch(&MetadataDispatcher, data, offset, length); + diagnostics::StartupEvent(this, "metadata response queued"); } // Handles data start time reported by the server at the beginning of a subscription. @@ -456,6 +461,7 @@ void DataSubscriber::HandleUpdateSignalIndexCache(const uint8_t* data, uint32_t if (data == nullptr) return; + diagnostics::StartupPhase startup(this, "signal cache processing scope"); vector uncompressedBuffer; int32_t cacheIndex = 0; @@ -486,8 +492,10 @@ void DataSubscriber::HandleUpdateSignalIndexCache(const uint8_t* data, uint32_t WriteBytes(uncompressedBuffer, data, offset, length); } + startup.Mark("signal cache decompress/copy complete (bytes)", uncompressedBuffer.size()); SignalIndexCachePtr signalIndexCache = NewSharedPtr(); signalIndexCache->Decode(uncompressedBuffer, m_subscriberID); + startup.Mark("signal cache decode complete"); m_signalIndexCacheMutex.lock(); m_signalIndexCache[cacheIndex].swap(signalIndexCache); @@ -548,6 +556,9 @@ void DataSubscriber::HandleDataPacket(uint8_t* data, uint32_t offset, const uint // Read measurement count and gather statistics const uint32_t count = EndianConverter::ToBigEndian(data, offset); + const bool firstPacket = m_totalMeasurementsReceived == 0 && count > 0; + if (firstPacket) + diagnostics::StartupEvent(this, "first nonempty data packet received", 0.0, count); m_totalMeasurementsReceived += count; offset += 4; //-V112 @@ -569,6 +580,8 @@ void DataSubscriber::HandleDataPacket(uint8_t* data, uint32_t offset, const uint else ParseCompactMeasurements(signalIndexCache, data, offset, length, includeTime, info.UseMillisecondResolution, frameLevelTimestamp, measurements); + if (firstPacket) + diagnostics::StartupEvent(this, "first packet decoded; invoking measurement callback", 0.0, measurements.size()); newMeasurementsCallback(this, measurements); } } @@ -1414,6 +1427,7 @@ void DataSubscriber::Subscribe(const SubscriptionInfo& info) // Subscribe to publisher in order to start receiving data. void DataSubscriber::Subscribe() { + diagnostics::StartupPhase startup(this, "subscribe send scope"); stringstream connectionStream; vector buffer; uint32_t bigEndianConnectionStringSize; @@ -1488,7 +1502,9 @@ void DataSubscriber::Subscribe() for (size_t i = 0; i < connectionStringSize; ++i) buffer[5 + i] = connectionStringPtr[i]; + startup.Mark("subscribe command built (bytes)", bufferSize); SendServerCommand(ServerCommand::Subscribe, buffer.data(), 0, bufferSize); + startup.Mark("subscribe send returned"); // Reset TSSC decompresser on successful (re)subscription m_tsscLastOOSReportMutex.lock(); diff --git a/src/lib/transport/StartupTrace.h b/src/lib/transport/StartupTrace.h new file mode 100644 index 0000000..01dc52c --- /dev/null +++ b/src/lib/transport/StartupTrace.h @@ -0,0 +1,85 @@ +#pragma once + +#include +#include +#include +#include + +namespace sttp { namespace diagnostics { + +// Opt-in, synchronous startup diagnostics. Bypasses the callback queue so a busy +// metadata callback cannot delay the timestamps. Set STTP_STARTUP_TRACE=1 before launch. +inline bool StartupTraceEnabled() +{ + static const bool enabled = [] { +#ifdef _MSC_VER + char* value = nullptr; + size_t length = 0; + const bool found = _dupenv_s(&value, &length, "STTP_STARTUP_TRACE") == 0 && + value != nullptr && value[0] == '1' && value[1] == '\0'; + std::free(value); + return found; +#else + const char* value = std::getenv("STTP_STARTUP_TRACE"); + return value != nullptr && value[0] == '1' && value[1] == '\0'; +#endif + }(); + return enabled; +} + +using StartupClock = std::chrono::steady_clock; + +inline void StartupEvent(const void* subscriber, const char* stage, double elapsedMs = 0.0, size_t items = 0) +{ + if (!StartupTraceEnabled()) + return; + + static const auto origin = StartupClock::now(); + static std::mutex outputMutex; + const auto now = StartupClock::now(); + const double sinceOrigin = std::chrono::duration(now - origin).count(); + const std::lock_guard lock(outputMutex); + std::fprintf(stderr, "[STTP startup +%.3f ms subscriber=%p] %s: %.3f ms; items=%zu\n", + sinceOrigin, const_cast(subscriber), stage, elapsedMs, items); + std::fflush(stderr); +} + +class StartupPhase +{ +public: + StartupPhase(const void* subscriber, const char* stage) : + m_subscriber(subscriber), m_stage(stage), m_enabled(StartupTraceEnabled()) + { + if (m_enabled) + { + m_start = m_previous = StartupClock::now(); + StartupEvent(m_subscriber, m_stage); + } + } + + // Each mark reports time since the previous mark; destruction reports total scope time. + void Mark(const char* stage, size_t items = 0) + { + if (!m_enabled) + return; + const auto now = StartupClock::now(); + StartupEvent(m_subscriber, stage, + std::chrono::duration(now - m_previous).count(), items); + m_previous = now; + } + + ~StartupPhase() + { + if (m_enabled) + StartupEvent(m_subscriber, m_stage, + std::chrono::duration(StartupClock::now() - m_start).count()); + } + +private: + const void* m_subscriber; + const char* m_stage; + bool m_enabled; + StartupClock::time_point m_start{}, m_previous{}; +}; + +} } diff --git a/src/lib/transport/SubscriberInstance.cpp b/src/lib/transport/SubscriberInstance.cpp index a831fca..92384db 100644 --- a/src/lib/transport/SubscriberInstance.cpp +++ b/src/lib/transport/SubscriberInstance.cpp @@ -22,6 +22,7 @@ //****************************************************************************************************** #include "SubscriberInstance.h" +#include "StartupTrace.h" #include "Constants.h" #include "../Convert.h" #include "../EndianConverter.h" @@ -189,6 +190,7 @@ void SubscriberInstance::SetMetadataFilters(const std::string& metadataFilters) void SubscriberInstance::HandleConnect() { + diagnostics::StartupEvent(m_subscriber.get(), "connection ready", 0.0, m_autoParseMetadata ? 1 : 0); // If automatically parsing metadata, request metadata upon successful connection, // after metadata is received the SubscriberInstance will then initiate subscribe; // otherwise, subscribe is initiated immediately (when auto subscribe requested) @@ -831,6 +833,7 @@ void SubscriberInstance::ReceivedMetadata(const vector& payload) return; } + diagnostics::StartupPhase startup(m_subscriber.get(), "metadata processing scope"); vector uncompressedBuffer; // Step 1: Decompress meta-data if needed @@ -852,6 +855,8 @@ void SubscriberInstance::ReceivedMetadata(const vector& payload) uncompressedBuffer.push_back(byte); } + startup.Mark("metadata decompress/copy complete (bytes)", uncompressedBuffer.size()); + // Step 2: Load string into an XML parser xml_document document; @@ -865,6 +870,8 @@ void SubscriberInstance::ReceivedMetadata(const vector& payload) return; } + startup.Mark("XML parse complete"); + // Find root node xml_node rootNode = document.document_element(); @@ -892,6 +899,8 @@ void SubscriberInstance::ReceivedMetadata(const vector& payload) devices.insert_or_assign(deviceMetadata->Acronym, deviceMetadata); } + startup.Mark("device objects complete", devices.size()); + // Query MeasurementDetail records from metadata unordered_map measurements; @@ -923,6 +932,8 @@ void SubscriberInstance::ReceivedMetadata(const vector& payload) } } + startup.Mark("measurement objects complete", measurements.size()); + // Query PhasorDetail records from metadata size_t phasorCount = 0; @@ -996,9 +1007,42 @@ void SubscriberInstance::ReceivedMetadata(const vector& payload) phasorCount++; } + startup.Mark("phasor objects and measurement matching complete"); + + // Diagnose per-device scaling and sparse indexes without recording metadata content. + if (diagnostics::StartupTraceEnabled()) + { + size_t maxDeviceMeasurements = 0; + int32_t maxAnalogIndex = 0, maxDigitalIndex = 0, maxPhasorIndex = 0; + size_t maxDevicePhasors = 0; + + for (const auto& entry : devices) + { + const auto& device = entry.second; + if (device->Measurements.size() > maxDeviceMeasurements) + maxDeviceMeasurements = device->Measurements.size(); + if (device->Phasors.size() > maxDevicePhasors) + maxDevicePhasors = device->Phasors.size(); + + const int32_t analogIndex = GetSignalKindCount(device->Measurements, SignalKind::Analog); + const int32_t digitalIndex = GetSignalKindCount(device->Measurements, SignalKind::Digital); + const int32_t phasorIndex = GetSignalKindCount(device->Measurements, SignalKind::Angle); + if (analogIndex > maxAnalogIndex) maxAnalogIndex = analogIndex; + if (digitalIndex > maxDigitalIndex) maxDigitalIndex = digitalIndex; + if (phasorIndex > maxPhasorIndex) maxPhasorIndex = phasorIndex; + } + + diagnostics::StartupEvent(m_subscriber.get(), "maximum measurements on one device", 0.0, maxDeviceMeasurements); + diagnostics::StartupEvent(m_subscriber.get(), "maximum phasors on one device", 0.0, maxDevicePhasors); + diagnostics::StartupEvent(m_subscriber.get(), "maximum analog reference index", 0.0, maxAnalogIndex); + diagnostics::StartupEvent(m_subscriber.get(), "maximum digital reference index", 0.0, maxDigitalIndex); + diagnostics::StartupEvent(m_subscriber.get(), "maximum phasor reference index", 0.0, maxPhasorIndex); + startup.Mark("diagnostic-only metadata shape scan complete"); + } // Construct a "configuration frame" for each of the devices StringMap configurationFrames; ConstructConfigurationFrames(devices, measurements, configurationFrames); + startup.Mark("configuration frames complete", configurationFrames.size()); m_configurationUpdateLock.lock(); @@ -1009,15 +1053,18 @@ void SubscriberInstance::ReceivedMetadata(const vector& payload) m_configurationUpdateLock.unlock(); stringstream message; + startup.Mark("metadata maps installed"); message << "Loaded " << devices.size() << " devices, " << measurements.size() << " measurements and " << phasorCount << " phasors from STTP meta data..."; StatusMessage(message.str()); // Notify derived class that meta-data has been parsed and is now available ParsedMetadata(); + startup.Mark("ParsedMetadata callback complete"); } void SubscriberInstance::SendMetadataRefreshCommand() { + diagnostics::StartupEvent(m_subscriber.get(), "metadata request sending"); if (m_metadataFilters.empty()) { m_subscriber->SendServerCommand(ServerCommand::MetadataRefresh); @@ -1384,8 +1431,10 @@ void SubscriberInstance::HandleMetadata(DataSubscriber* source, const vectorReceivedMetadata(payload); + startup.Mark("ReceivedMetadata returned; ready to subscribe", payload.size()); // When auto-parsing metadata, start subscription after successful user meta-data handling if (instance->m_autoParseMetadata) diff --git a/src/samples/MetadataBenchmark/MetadataBenchmark.cpp b/src/samples/MetadataBenchmark/MetadataBenchmark.cpp new file mode 100644 index 0000000..186d461 --- /dev/null +++ b/src/samples/MetadataBenchmark/MetadataBenchmark.cpp @@ -0,0 +1,340 @@ +#include "../../lib/transport/SubscriberInstance.h" +#include "../../lib/Convert.h" +#include "../../lib/EndianConverter.h" +#include "../../lib/transport/CompactMeasurement.h" +#include "../../lib/transport/DataPublisher.h" +#include "../../lib/transport/tssc/TSSCEncoder.h" +#include "../../lib/transport/tssc/TSSCDecoder.h" +#include +#include +#include +#include +#include +#include +#include + +using namespace sttp; +using namespace sttp::transport; +using BenchClock = std::chrono::steady_clock; + +static std::string GuidText(uint64_t id) +{ + char buffer[40]; + std::snprintf(buffer, sizeof(buffer), "00000000-0000-0000-0000-%012llx", static_cast(id)); + return buffer; +} + +static Guid GuidValue(uint64_t id) { return ParseGuid(GuidText(id).c_str()); } + +struct Shape +{ + std::string kind; + unsigned devices, points, stride; + bool gzip; +}; + +static std::vector Metadata(const Shape& shape, size_t& xmlBytes) +{ + std::ostringstream xml; + xml << ""; + for (unsigned d = 0; d < shape.devices; ++d) + xml << "D" << d << "Test device" + << GuidText(10000000 + d) << "30" + << "2026-09-15T12:34:56.789Z"; + uint64_t id = 0; + for (unsigned d = 0; d < shape.devices; ++d) + for (unsigned p = 1; p <= shape.points; ++p) + for (unsigned part = 0; part < (shape.kind == "phasor" ? 2U : 1U); ++part) + { + const char* suffix = shape.kind == "analog" ? "AV" : shape.kind == "digital" ? "DV" : part == 0 ? "PA" : "PM"; + ++id; + xml << "D" << d << "TEST:" << id + << "" << GuidText(id) << "TEST_" << id + << "D" << d << '-' << suffix << p * shape.stride + << "" << (shape.kind == "phasor" ? p * shape.stride : 0) + << "Synthetic measurement for metadata scaling test" + << "2026-09-15T12:34:56.789Z"; + } + if (shape.kind == "phasor") + for (unsigned d = 0; d < shape.devices; ++d) + for (unsigned p = 1; p <= shape.points; ++p) + xml << "D" << d << "" + << "V+" << p * shape.stride + << "2026-09-15T12:34:56.789Z"; + xml << ""; + const auto text = xml.str(); + xmlBytes = text.size(); + if (!shape.gzip) return {text.begin(), text.end()}; + std::vector compressed; + { + boost::iostreams::filtering_ostream output; + output.push(boost::iostreams::gzip_compressor()); + output.push(boost::iostreams::back_inserter(compressed)); + output.write(text.data(), static_cast(text.size())); + } + return {compressed.begin(), compressed.end()}; +} + +class BenchmarkSubscriber final : public SubscriberInstance +{ +public: + void Parse(const std::vector& payload) { ReceivedMetadata(payload); } + bool parsed = false; +protected: + void StatusMessage(const std::string&) override {} + void ErrorMessage(const std::string& message) override { throw std::runtime_error(message); } + void ParsedMetadata() override { parsed = true; } +}; + +static void Require(bool condition, const char* message) +{ + if (!condition) throw std::runtime_error(message); +} + +static void Validate(BenchmarkSubscriber& subscriber, const Shape& shape) +{ + Require(subscriber.parsed, "ParsedMetadata was not called"); + size_t measured = 0, frames = 0; + const unsigned parts = shape.kind == "phasor" ? 2U : 1U; + subscriber.IterateMeasurementMetadata([&](MeasurementMetadataPtr, void*) { ++measured; }, nullptr); + Require(measured == size_t(shape.devices) * shape.points * parts, "Measurement count mismatch"); + subscriber.IterateConfigurationFrames([&](ConfigurationFramePtr frame, void*) { + ++frames; + const auto d = std::stoul(frame->DeviceAcronym.substr(1)); + const unsigned slots = shape.points * shape.stride; + Require(frame->Measurements.size() == size_t(shape.points) * parts, "Frame signal count mismatch"); + if (shape.kind == "phasor") + { + Require(frame->Phasors.size() == slots, "Phasor slot count mismatch"); + for (unsigned i = 1; i <= slots; ++i) + { + const auto& phasor = frame->Phasors[i - 1]; + Require(phasor->Phasor->SourceIndex == i, "Phasor index mismatch"); + if (i % shape.stride == 0) + { + const uint64_t id = (uint64_t(d) * shape.points + i / shape.stride - 1) * 2 + 1; + Require(phasor->Angle && phasor->Magnitude, "Missing phasor component"); + Require(phasor->Angle->SignalID == GuidValue(id), "Wrong phasor angle"); + Require(phasor->Magnitude->SignalID == GuidValue(id + 1), "Wrong phasor magnitude"); + } + else Require(!phasor->Angle && !phasor->Magnitude && phasor->Phasor->Label == "UNDEFINED", "Wrong phasor placeholder"); + } + } + else + { + const auto& values = shape.kind == "analog" ? frame->Analogs : frame->Digitals; + Require(values.size() == slots, "Scalar slot count mismatch"); + for (unsigned i = 1; i <= slots; ++i) + { + Require(values[i - 1]->Reference.Index == i, "Scalar index mismatch"); + if (i % shape.stride == 0) + Require(values[i - 1]->SignalID == GuidValue(uint64_t(d) * shape.points + i / shape.stride), "Wrong scalar signal"); + else Require(values[i - 1]->SignalID == Empty::Guid && values[i - 1]->PointTag == "UNDEFINED", "Wrong scalar placeholder"); + } + } + }, nullptr); + Require(frames == shape.devices, "Device/frame count mismatch"); +} + +static void Run(const Shape& shape) +{ + size_t xmlBytes = 0; + const auto payload = Metadata(shape, xmlBytes); + BenchmarkSubscriber subscriber; + subscriber.SetMetadataCompressed(shape.gzip); + const auto start = BenchClock::now(); + subscriber.Parse(payload); + const double elapsed = std::chrono::duration(BenchClock::now() - start).count(); + Validate(subscriber, shape); + std::cout << "RESULT," << shape.kind << ',' << shape.devices << ',' << shape.points << ',' << shape.stride + << ',' << shape.gzip << ',' << xmlBytes << ',' << payload.size() << ',' << elapsed << ",PASS" << std::endl; +} + +static void DuplicateRegression(const std::string& kind) +{ + const Shape shape{kind, 1, 1, 1, false}; + size_t size; + const auto payload = Metadata(shape, size); + std::string xml(payload.begin(), payload.end()); + const std::string suffix = kind == "analog" ? "AV" : kind == "digital" ? "DV" : "PA"; + std::string extra = "D0TEST:999" + GuidText(999) + + "DUPLICATED0-" + suffix + + "112026-09-15T12:34:56.789Z"; + if (kind == "phasor") + extra += "D0V+12026-09-15T12:34:56.789Z"; + xml.insert(xml.rfind(""), extra); + BenchmarkSubscriber subscriber; + subscriber.SetMetadataCompressed(false); + subscriber.Parse({xml.begin(), xml.end()}); + ConfigurationFramePtr frame; + Require(subscriber.TryGetConfigurationFrame("D0", frame), "Missing duplicate regression frame"); + if (kind == "phasor") + Require(frame->Phasors[0]->Phasor->Label == "Test phasor" && frame->Phasors[0]->Angle->SignalID == GuidValue(1), "First phasor match changed"); + else + { + const auto& values = kind == "analog" ? frame->Analogs : frame->Digitals; + Require(values[0]->SignalID == GuidValue(1), "First scalar match changed"); + } + std::cout << "REGRESSION," << kind << ",duplicate,PASS\n"; +} +static void WideIndexRegression() +{ + for (const int32_t index : {65535, 65536, 100000, std::numeric_limits::max()}) + { + const std::string text = "D0-AV" + std::to_string(index); + const SignalReference reference(text); + Require(reference.Index == index, "32-bit signal-reference parsing failed"); + std::ostringstream formatted; + formatted << reference; + Require(formatted.str() == text, "32-bit signal-reference formatting failed"); + } + Require(SignalReference("D0-AV2147483648").Index == 0, "Overflowed reference did not retain default index"); + + // Build a genuine signal-index-cache wire payload containing over 65535 entries. + std::vector wire; + const Guid subscriberID = GuidValue(123456789); + WriteBytes(wire, uint32_t(0)); + WriteBytes(wire, subscriberID); + constexpr uint32_t entries = 70001; + EndianConverter::WriteBigEndianBytes(wire, entries); + for (uint32_t i = 0; i < entries; ++i) + { + const int32_t index = i == entries - 1 ? std::numeric_limits::max() : static_cast(65535 + i); + EndianConverter::WriteBigEndianBytes(wire, index); + WriteBytes(wire, GuidValue(uint64_t(index) + 1)); + EndianConverter::WriteBigEndianBytes(wire, uint32_t(4)); + for (const char c : std::string("TEST")) wire.push_back(static_cast(c)); + EndianConverter::WriteBigEndianBytes(wire, uint64_t(index)); + } + EndianConverter::WriteBigEndianBytes(wire, uint32_t(0)); + const uint32_t length = EndianConverter::Default.ConvertBigEndian(static_cast(wire.size())); + std::memcpy(wire.data(), &length, sizeof(length)); + const auto cache = NewSharedPtr(); + Guid decodedSubscriber; + cache->Decode(wire, decodedSubscriber); + Require(decodedSubscriber == subscriberID && cache->Count() == entries, "Large wire cache count/identity mismatch"); + CompactMeasurement codec(cache, nullptr, false); + for (const int32_t index : {65535, 65536, 100000, std::numeric_limits::max()}) + { + const auto id = GuidValue(uint64_t(index) + 1); + Require(cache->GetSignalID(index) == id && cache->GetSignalIndex(id) == index, "Wide cache lookup failed"); + Measurement input; + input.SignalID = id; + input.Value = 12.5; + input.Multiplier = 1.0; + std::vector bytes; + codec.SerializeMeasurement(input, bytes, index); + uint32_t offset = 0; + MeasurementPtr output; + Require(codec.TryParseMeasurement(bytes.data(), offset, static_cast(bytes.size()), output), "Wide compact measurement decode failed"); + Require(offset == bytes.size() && output->SignalID == id && output->Value == input.Value && output->ID == uint64_t(index), "Wide compact measurement mismatch"); + } + std::cout << "REGRESSION,int32,70001-entry-wire-cache-and-compact-roundtrip,PASS\n"; +} +static void PublisherIndexRegression() +{ + const auto device = NewSharedPtr(); + device->Acronym = "D0"; + device->Name = "Wide index regression"; + device->UniqueID = GuidValue(987654); + device->FramesPerSecond = 30; + device->UpdatedOn = UtcNow(); + std::vector measurements; + std::vector phasors; + for (const int32_t index : {34464, 100000}) + { + const auto phasor = NewSharedPtr(); + phasor->DeviceAcronym = "D0"; + phasor->Label = "Test"; + phasor->SourceIndex = index; + phasor->Type = index == 100000 ? "V" : "I"; + phasor->Phase = "+"; + phasor->UpdatedOn = UtcNow(); + phasors.push_back(phasor); + const auto measurement = NewSharedPtr(); + measurement->DeviceAcronym = "D0"; + measurement->ID = "TEST:" + std::to_string(index); + measurement->PointTag = "TEST_" + std::to_string(index); + measurement->SignalID = GuidValue(index); + measurement->Reference = SignalReference("D0-PA" + std::to_string(index)); + measurement->PhasorSourceIndex = index; + measurement->UpdatedOn = UtcNow(); + measurements.push_back(measurement); + } + DataPublisher publisher; + publisher.DefineMetadata({device}, measurements, phasors); + const auto voltage = publisher.FilterMetadata("FILTER MeasurementDetail WHERE SignalAcronym = 'VPHA'"); + const auto current = publisher.FilterMetadata("FILTER MeasurementDetail WHERE SignalAcronym = 'IPHA'"); + Require(voltage.size() == 1 && current.size() == 1, "Publisher phasor index collision"); + Require(voltage[0]->PhasorSourceIndex == 100000 && voltage[0]->Reference.Index == 100000, "Publisher truncated 32-bit metadata index"); + Require(current[0]->PhasorSourceIndex == 34464, "Publisher confused indexes differing by 65536"); + std::cout << "REGRESSION,int32,publisher-index-roundtrip-and-collision,PASS\n"; +} + +static void TSSCIndexRegression() +{ + using namespace sttp::transport::tssc; + std::vector bytes(4096); + TSSCEncoder encoder; + encoder.SetBuffer(bytes.data(), 0, static_cast(bytes.size())); + const std::vector indexes{65535, 65536, 100000, 0, 100000}; + for (const auto index : indexes) + Require(encoder.TryAddMeasurement(index, 123456789, 0, 12.5f), "TSSC encode failed"); + const auto length = encoder.FinishBlock(); + TSSCDecoder decoder; + decoder.SetBuffer(bytes.data(), 0, length); + for (const auto expected : indexes) + { + int32_t index; + int64_t timestamp; + uint32_t quality; + float32_t value; + Require(decoder.TryGetMeasurement(index, timestamp, quality, value), "TSSC decode failed"); + Require(index == expected && timestamp == 123456789 && quality == 0 && value == 12.5f, "TSSC wide index mismatch"); + } + std::cout << "REGRESSION,int32,TSSC-wide-index-roundtrip,PASS\n"; +} +int main(int argc, char** argv) +{ + try + { + if (argc == 2 && std::string(argv[1]) == "--regression") + { + for (const auto& kind : {"analog", "digital", "phasor"}) + { + Run({kind, 3, 10, 1, false}); + Run({kind, 2, 3, 7, true}); + Run({kind, 1, 1, 65535, false}); + Run({kind, 1, 1, 65536, false}); + Run({kind, 1, 1, 100000, true}); + DuplicateRegression(kind); + } + Run({"analog", 1, 70000, 1, true}); + WideIndexRegression(); + PublisherIndexRegression(); + TSSCIndexRegression(); + return 0; + } + if (argc < 4) + { + std::cout << "Usage: MetadataBenchmark analog|digital|phasor DEVICES POINTS_PER_DEVICE [STRIDE=1] [gzip]\n"; + return 1; + } + Shape shape{argv[1], static_cast(std::stoul(argv[2])), static_cast(std::stoul(argv[3])), + argc > 4 ? static_cast(std::stoul(argv[4])) : 1, argc > 5 && std::string(argv[5]) == "gzip"}; + Require(shape.kind == "analog" || shape.kind == "digital" || shape.kind == "phasor", "Invalid kind"); + Require(shape.devices > 0 && shape.points > 0 && shape.stride > 0, "Counts must be positive"); + Require(uint64_t(shape.points) * shape.stride <= std::numeric_limits::max(), "Index exceeds int32 range"); + Require(uint64_t(shape.devices) * shape.points * (shape.kind == "phasor" ? 2 : 1) <= 500000, "Limit this benchmark to 500,000 measurements"); + Require(uint64_t(shape.devices) * shape.points * shape.stride <= 1000000, "Limit total configuration slots to 1,000,000"); + Run(shape); + return 0; + } + catch (const std::exception& ex) { std::cerr << "FAIL: " << ex.what() << std::endl; return 1; } +} + + + + + + diff --git a/src/samples/MetadataBenchmark/README.md b/src/samples/MetadataBenchmark/README.md new file mode 100644 index 0000000..77c48e2 --- /dev/null +++ b/src/samples/MetadataBenchmark/README.md @@ -0,0 +1,27 @@ +# Native metadata benchmark + +Build from the C++ repository root: + +```powershell +.\src\samples\MetadataBenchmark\build.cmd +.\build\startup-diagnostics\MetadataBenchmark.exe --regression +``` + +Uses the same Visual Studio / Boost setup as ../StartupTiming/build.cmd. Output is build/startup-diagnostics/MetadataBenchmark.exe. Tests return nonzero on failure. + +Generate and validate metadata without a publisher: + +```powershell +$env:STTP_STARTUP_TRACE = '1' +.\build\startup-diagnostics\MetadataBenchmark.exe analog 8 60000 1 +.\build\startup-diagnostics\MetadataBenchmark.exe phasor 1 5000 1 gzip +.\src\samples\MetadataBenchmark\run-suite.ps1 -Label current +``` + +Arguments: analog|digital|phasor DEVICES POINTS_PER_DEVICE [STRIDE=1] [gzip]. A phasor point creates two measurement records. Stride produces sparse indexes and tests placeholders. The harness caps input at 500,000 measurements and 1,000,000 configuration slots to bound memory usage; these are benchmark safeguards, not protocol limits. + +RESULT rows contain kind, devices, points per device, stride, gzip flag, XML bytes, payload bytes, processing milliseconds, and PASS. Timing includes the real metadata-processing call and its local cleanup, but excludes generation, compression, validation, and network work. Large cases can use over 1 GB of memory. Run comparisons sequentially in Release builds on an otherwise quiet machine. + +The suite writes separate output/trace files per case under build/startup-diagnostics/LABEL, imposes a per-case timeout (default 180 seconds), and fails if any case fails or times out. Use -Executable to compare a separately preserved baseline binary. Do not run old binaries on index 65,535: the original frame loop can wrap indefinitely. + +Regression mode verifies dense/sparse/duplicate metadata, 65K boundaries, 70,000 measurements, publisher index collisions, signal-cache and compact wire decoding, and TSSC wide-index round trips. See ../../../docs/MetadataScaling.md for measured results and integration limits. diff --git a/src/samples/MetadataBenchmark/build.cmd b/src/samples/MetadataBenchmark/build.cmd new file mode 100644 index 0000000..0392903 --- /dev/null +++ b/src/samples/MetadataBenchmark/build.cmd @@ -0,0 +1,20 @@ +@echo off +setlocal +for %%I in ("%~dp0..\..\..") do set "REPO=%%~fI" +for /f "usebackq tokens=*" %%I in (`"%ProgramFiles(x86)%\Microsoft Visual Studio\Installer\vswhere.exe" -latest -products * -requires Microsoft.VisualStudio.Component.VC.Tools.x86.x64 -property installationPath`) do set "VS=%%I" +if not defined VS exit /b 1 +call "%VS%\VC\Auxiliary\Build\vcvars64.bat" +if errorlevel 1 exit /b 1 +if not defined STTP_BOOST_ROOT set "STTP_BOOST_ROOT=%REPO%\..\boost" +if not defined STTP_PLATFORM_TOOLSET set "STTP_PLATFORM_TOOLSET=v145" +"%VS%\MSBuild\Current\Bin\MSBuild.exe" "%REPO%\src\lib\sttp.cpp.vcxproj" /p:Configuration=Release /p:Platform=x64 /p:PlatformToolset=%STTP_PLATFORM_TOOLSET% /p:SolutionDir="%REPO%\src\\" /nologo /v:minimal +if errorlevel 1 exit /b 1 +if not exist "%REPO%\build\startup-diagnostics" mkdir "%REPO%\build\startup-diagnostics" +pushd "%REPO%\build\startup-diagnostics" +cl /nologo /std:c++latest /O2 /MD /EHsc /D_WIN32_WINNT=0x0601 /DBOOST_CONFIG_SUPPRESS_OUTDATED_MESSAGE /DUSE_UTF8_INSTEAD_OF_CODECVT /D_HAS_AUTO_PTR_ETC /DANTLR4CPP_STATIC /I"%STTP_BOOST_ROOT%" /Fe:MetadataBenchmark.exe /Fo:MetadataBenchmark.obj "%REPO%\src\samples\MetadataBenchmark\MetadataBenchmark.cpp" "%REPO%\build\output\x64\Release\lib\sttp.cpp.lib" /link /LTCG /LIBPATH:"%STTP_BOOST_ROOT%\stage\lib" ws2_32.lib +set "RESULT=%ERRORLEVEL%" +popd +exit /b %RESULT% + + + diff --git a/src/samples/MetadataBenchmark/run-suite.ps1 b/src/samples/MetadataBenchmark/run-suite.ps1 new file mode 100644 index 0000000..2f90c3d --- /dev/null +++ b/src/samples/MetadataBenchmark/run-suite.ps1 @@ -0,0 +1,54 @@ +param( + [string]$Executable = '', + [string]$Label = 'baseline', + [int]$TimeoutSeconds = 180 +) +$ErrorActionPreference = 'Stop' +$repo = [System.IO.Path]::GetFullPath((Join-Path $PSScriptRoot '../../..')) +if (!$Executable) { $Executable = Join-Path $repo 'build/startup-diagnostics/MetadataBenchmark.exe' } +$Executable = (Resolve-Path -LiteralPath $Executable).Path +if ($Label -notmatch '^[A-Za-z0-9_-]+$') { throw 'Label must contain only letters, numbers, underscore or hyphen' } +$output = Join-Path $repo "build/startup-diagnostics/$Label" +New-Item -ItemType Directory -Force -Path $output | Out-Null +$cases = @( + @('analog','1','5000','1'), + @('analog','1','10000','1'), + @('analog','1','20000','1'), + @('analog','1','50000','1'), + @('analog','500','100','1'), + @('digital','1','20000','1'), + @('phasor','1','5000','1'), + @('analog','1','10','6000'), + @('analog','500','100','1','gzip') +) +$failures = 0 +$oldTrace = $env:STTP_STARTUP_TRACE +$env:STTP_STARTUP_TRACE = '1' +try { + foreach ($arguments in $cases) { + $name = $arguments -join '-' + $stdout = Join-Path $output "$name.out" + $stderr = Join-Path $output "$name.log" + $process = Start-Process -FilePath $Executable -ArgumentList $arguments -WindowStyle Hidden -PassThru -RedirectStandardOutput $stdout -RedirectStandardError $stderr + $elapsed = [System.Diagnostics.Stopwatch]::StartNew() + while (!$process.WaitForExit(1000)) { + if ($elapsed.Elapsed.TotalSeconds -ge $TimeoutSeconds) { + $process.Kill() + $process.WaitForExit() + break + } + } + $process.Refresh() + if ($process.ExitCode -ne 0) { + $failures++ + Write-Output "$name FAILED or timed out after $($elapsed.Elapsed.TotalSeconds.ToString('F1')) seconds, exit $($process.ExitCode)" + Get-Content -LiteralPath $stderr -Tail 3 + } else { + Get-Content -LiteralPath $stdout | Where-Object { $_ -like 'RESULT,*' } + } + $process.Dispose() + } +} finally { $env:STTP_STARTUP_TRACE = $oldTrace } + +if ($failures -gt 0) { throw "$failures benchmark cases failed; see $output" } + diff --git a/src/samples/StartupTiming/README.md b/src/samples/StartupTiming/README.md new file mode 100644 index 0000000..e3c3392 --- /dev/null +++ b/src/samples/StartupTiming/README.md @@ -0,0 +1,30 @@ +# Native startup timing client (Windows x64) + +Uses one subscriber with `FILTER ActiveMeasurements WHERE SignalType <> 'STAT'`. +Reports first-data timing and measurements/sec; any key or Ctrl+C exits. `--seconds N` limits a diagnostic run; exit code 2 means no measurements arrived before exit, 1 means an argument/startup exception, and 0 means measurements were received. + +## Build + +From the repository root: + +```powershell +.\src\samples\StartupTiming\build.cmd +``` + +Builds the instrumented C++ library and the client in Release with the installed Visual Studio C++ tools. The default toolset is v145; set `STTP_PLATFORM_TOOLSET` to select another installed toolset. Boost headers and compiled libraries must be in the sibling `..\boost` folder as expected by the existing library project. The standalone client link step also accepts `STTP_BOOST_ROOT` to override its Boost lookup. + +Output: `build\startup-diagnostics\StartupTiming.exe`. + +## Compare startup with and without metadata + +```powershell +$env:STTP_STARTUP_TRACE = '1' +.\build\startup-diagnostics\StartupTiming.exe 127.0.0.1 7175 --seconds 10 2> build\startup-diagnostics\metadata.log +.\build\startup-diagnostics\StartupTiming.exe 127.0.0.1 7175 --no-metadata --seconds 10 2> build\startup-diagnostics\no-metadata.log +``` + +For a slow remote system, omit `--seconds` or use a duration long enough to include the reported 2.5-minute startup (for example 300). IP/hostname is required; port defaults to 7165. `--no-metadata` skips metadata retrieval and configuration construction; the subscription filter remains identical and runtime signal-cache decoding still occurs. + +Set `STTP_STARTUP_TRACE=0` or remove the environment variable to turn off native diagnostics. Trace output goes to stderr and rate output to stdout. Detailed stage interpretation is in `docs/StartupDiagnostics.md`. + +This is a native executable linked to the C++ static library. It does not update the .NET test executable or its SWIG DLL. The Visual C++ runtime must be available on the target machine. diff --git a/src/samples/StartupTiming/StartupTiming.cpp b/src/samples/StartupTiming/StartupTiming.cpp new file mode 100644 index 0000000..c3c7c2e --- /dev/null +++ b/src/samples/StartupTiming/StartupTiming.cpp @@ -0,0 +1,126 @@ +#include "../../lib/transport/SubscriberInstance.h" +#include +#include +#include +#include +#include +#include +#include + +using Clock = std::chrono::steady_clock; +using namespace sttp; +using namespace sttp::transport; + +static volatile std::sig_atomic_t stopped = 0; +static long long SteadyNanoseconds() { return std::chrono::duration_cast(Clock::now().time_since_epoch()).count(); } + +class TimingSubscriber final : public SubscriberInstance +{ +public: + std::atomic connected{0}, first{0}; + std::atomic count{0}; +protected: + void StatusMessage(const std::string&) override {} + void ErrorMessage(const std::string& message) override { std::cerr << "STTP: " << message << std::endl; } + void ConnectionEstablished() override + { + // Capture the earliest callback even if the library reports establishment twice. + long long expected = 0; + connected.compare_exchange_strong(expected, SteadyNanoseconds()); + } + void ConnectionTerminated() override { std::cerr << "Connection terminated." << std::endl; } + void ReceivedNewMeasurements(const std::vector& measurements) override + { + if (measurements.empty()) return; + const auto now = SteadyNanoseconds(); + count.fetch_add(measurements.size()); + long long expected = 0; + first.compare_exchange_strong(expected, now); + } +}; + +int main(int argc, char** argv) +{ + try + { + if (argc < 2 || std::string(argv[1]) == "--help") + { + std::cout << "Usage: StartupTiming IP [port=7165] [--no-metadata] [--seconds N]\n"; + return argc < 2 ? 1 : 0; + } + unsigned long port = 7165; + bool metadata = true; + int seconds = 0; + int index = 2; + if (index < argc && argv[index][0] != '-') + { + std::string value(argv[index++]); + size_t consumed = 0; + port = std::stoul(value, &consumed); + if (consumed != value.size() || port == 0 || port > 65535) throw std::runtime_error("Invalid port"); + } + while (index < argc) + { + const std::string option(argv[index++]); + if (option == "--no-metadata") metadata = false; + else if (option == "--seconds" && index < argc) + { + const std::string value(argv[index++]); + size_t consumed = 0; + seconds = std::stoi(value, &consumed); + if (consumed != value.size() || seconds <= 0) throw std::runtime_error("Invalid duration"); + } + else throw std::runtime_error("Unknown/incomplete option: " + option); + } + std::signal(SIGINT, [](int) { stopped = 1; }); + TimingSubscriber subscriber; + subscriber.Initialize(argv[1], static_cast(port)); + subscriber.SetFilterExpression("FILTER ActiveMeasurements WHERE SignalType <> 'STAT'"); + subscriber.SetAutoParseMetadata(metadata); + subscriber.SetAutoReconnect(false); + std::cout << "Endpoint: " << argv[1] << ':' << port << "; metadata: " << (metadata ? "enabled" : "disabled") << '\n' + << "Press any key to exit.\n"; + const auto start = SteadyNanoseconds(); + subscriber.ConnectAsync(); + bool reported = false; + unsigned long long previousCount = 0; + long long previousTime = start; + unsigned spinner = 0; + while (!stopped) + { + if (_kbhit()) { _getch(); break; } + const auto now = SteadyNanoseconds(); + if (seconds > 0 && (now - start) / 1e9 >= seconds) break; + const auto first = subscriber.first.load(); + if (!reported && first != 0) + { + std::cout << std::fixed << std::setprecision(3) + << "Connection to first measurement: " << (first - subscriber.connected.load()) / 1e6 << " ms\n" + << "Total from connection attempt: " << (first - start) / 1e6 << " ms\n\n"; + previousTime = first; + reported = true; + } + const double elapsed = (now - previousTime) / 1e9; + if (reported && elapsed >= 1.0) + { + const auto count = subscriber.count.load(); + if (count != previousCount) spinner = (spinner + 1) % 4; + std::cout << '\r' << "\\|/-"[spinner] << ' ' << std::setw(12) << std::setprecision(0) + << (count - previousCount) / elapsed << " measurements/sec" << std::flush; + previousCount = count; + previousTime = now; + } + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + } + subscriber.Disconnect(); + std::cout << "\nTotal measurements: " << subscriber.count.load() << std::endl; + return subscriber.count.load() == 0 ? 2 : 0; + } + catch (const std::exception& ex) + { + std::cerr << ex.what() << std::endl; + return 1; + } +} + + diff --git a/src/samples/StartupTiming/build.cmd b/src/samples/StartupTiming/build.cmd new file mode 100644 index 0000000..4afc728 --- /dev/null +++ b/src/samples/StartupTiming/build.cmd @@ -0,0 +1,19 @@ +@echo off +setlocal +for %%I in ("%~dp0..\..\..") do set "REPO=%%~fI" +for /f "usebackq tokens=*" %%I in (`"%ProgramFiles(x86)%\Microsoft Visual Studio\Installer\vswhere.exe" -latest -products * -requires Microsoft.VisualStudio.Component.VC.Tools.x86.x64 -property installationPath`) do set "VS=%%I" +if not defined VS exit /b 1 +call "%VS%\VC\Auxiliary\Build\vcvars64.bat" +if errorlevel 1 exit /b 1 +if not defined STTP_BOOST_ROOT set "STTP_BOOST_ROOT=%REPO%\..\boost" +if not defined STTP_PLATFORM_TOOLSET set "STTP_PLATFORM_TOOLSET=v145" +"%VS%\MSBuild\Current\Bin\MSBuild.exe" "%REPO%\src\lib\sttp.cpp.vcxproj" /p:Configuration=Release /p:Platform=x64 /p:PlatformToolset=%STTP_PLATFORM_TOOLSET% /p:SolutionDir="%REPO%\src\\" /nologo /v:minimal +if errorlevel 1 exit /b 1 +if not exist "%REPO%\build\startup-diagnostics" mkdir "%REPO%\build\startup-diagnostics" +pushd "%REPO%\build\startup-diagnostics" +cl /nologo /std:c++latest /O2 /MD /EHsc /D_WIN32_WINNT=0x0601 /DBOOST_CONFIG_SUPPRESS_OUTDATED_MESSAGE /DUSE_UTF8_INSTEAD_OF_CODECVT /D_HAS_AUTO_PTR_ETC /DANTLR4CPP_STATIC /I"%STTP_BOOST_ROOT%" /Fe:StartupTiming.exe /Fo:StartupTiming.obj "%REPO%\src\samples\StartupTiming\StartupTiming.cpp" "%REPO%\build\output\x64\Release\lib\sttp.cpp.lib" /link /LTCG /LIBPATH:"%STTP_BOOST_ROOT%\stage\lib" ws2_32.lib +set "RESULT=%ERRORLEVEL%" +popd +exit /b %RESULT% + +