diff --git a/modules/opcua_generic_client_module/USAGE.md b/modules/opcua_generic_client_module/USAGE.md new file mode 100644 index 00000000..5b0b7d78 --- /dev/null +++ b/modules/opcua_generic_client_module/USAGE.md @@ -0,0 +1,291 @@ +# OPC UA Generic Client — properties and usage + +The module connects openDAQ to any OPC UA server. It does not build a device tree: you add one +`MonitoredItem` function block per OPC UA node you want to read, and each block publishes a value +signal and a domain (timestamp) signal. + +--- + +## 1. Connect + +```cpp +auto instance = daq::Instance(); +auto device = instance.addDevice("daq.opcua.generic://192.168.1.50:4840"); +``` + +Connection string: `daq.opcua.generic://[:][/]` — port defaults to `4840`, path to +empty. IPv6 hosts go in brackets: `daq.opcua.generic://[::1]:4840`. + +To connect with settings, take the default config of the device type and pass it along: + +```cpp +PropertyObjectPtr config = instance.createDefaultAddDeviceConfig(); +PropertyObjectPtr opcuaSetting = config.getPropertyValue("Device.OPCUAGeneric"); +opcuaSetting.setPropertyValue("Username", "operator"); +opcuaSetting.setPropertyValue("Password", "secret"); +opcuaSetting.setPropertyValue("TimestampMode", 3); // LocalSystemTimestamp +auto device = instance.addDevice("daq.opcua.generic://192.168.1.50:4840", config); +``` + +### Device properties + +| Property | Type | Default | Applied | +|---|---|---|---| +| `Username` | String | `""` | at connect | +| `Password` | String | `""` | at connect | +| `LocalId` | String | `""` | at connect | +| `TimestampMode` | Selection | `2` — `SourceTimestamp` | at connect **and** at runtime | +| `DeviceNodeIDType` | Selection | `1` — `String` | at connect | +| `DeviceNodeIDString` | String | `""` | at connect | +| `DeviceNodeIDNumeric` | Int | `0` | at connect | +| `DeviceNamespaceIndex` | Int | `0` | at connect | + +Everything except `TimestampMode` is read once while the device is being created; changing those +values afterwards has no effect — remove the device and add it again. `TimestampMode` remains a +property of the device object and can be written at any time. + +--- + +**`Username` / `Password`** — how the client authenticates when it opens the OPC UA session. + +An empty `Username` means an **anonymous** session, and `Password` is then ignored entirely — a +password on its own never reaches the server. A non-empty `Username` switches the endpoint to a +username/password identity token. + +--- + +**`LocalId`** — the device's local ID: the component identifier it gets inside the parent folder, so +it is what shows up in component paths and in `device.getLocalId()`. + +Set it when you want the same device to keep the same identifier across application runs — for +example when configuration is stored per component path. Leave it empty to let the module derive one. + +The value is used as-is unless it is empty or already taken by a sibling device; in those cases the +module falls back, in order, to `_` read from the device root node, then +the server's `ApplicationUri` (with `/` replaced by `-`), then a generated +`GenericOPCUAClientPseudoDevice`. + +Note this is not the device *name*: the name comes from the server's application description, and is +independent of `LocalId`. + +--- + +**`TimestampMode`** — which clock the domain (time) signal of every `MonitoredItem` of this device +carries. It is a device-wide setting; individual blocks cannot override it. + +| Value | Name | What the domain signal carries | When to use it | +|---|---|---|---| +| `0` | `None` | nothing — no domain signal is created at all | you only care about values, or the consumer supplies its own time axis | +| `1` | `ServerTimestamp` | the time the OPC UA server produced the response | the server's clock is the reference, the source timestamp is unreliable | +| `2` | `SourceTimestamp` | the time the value originated at its source (default) | closest to when the data was actually measured | +| `3` | `LocalSystemTimestamp` | the client's system clock at the moment of the read | the server sends no usable timestamps; includes network + polling delay | + +With `ServerTimestamp` or `SourceTimestamp`, a server that does not deliver that timestamp puts the +block into `Error` and it publishes nothing — that is a real, and common, failure mode. +`LocalSystemTimestamp` always works, at the cost of accuracy. `None` removes the domain signal, so +readers must not expect a time axis. + +Writing the property at runtime takes effect immediately on all existing blocks: domain signals are +created or removed as needed. + +```cpp +device.setPropertyValue("TimestampMode", 1); // ServerTimestamp +``` + +--- + +**`DeviceNodeIDType` / `DeviceNodeIDString` / `DeviceNodeIDNumeric` / `DeviceNamespaceIndex`** — the +address of one node on the server that describes the device itself, typically a `DeviceType` or +`ComponentType` object from the OPC UA DI companion specification. + +The four properties together form a single NodeID: `DeviceNodeIDType` selects which identifier is +used (`1` = String → `DeviceNodeIDString`, `0` = Numeric → `DeviceNodeIDNumeric`), and +`DeviceNamespaceIndex` is the namespace of that identifier. Only the matching identifier property is +visible in a UI; the other one is hidden. + +The whole group is **optional** and serves two purposes: + +* filling in `device.getInfo()` — the module browses the node's `HasProperty` children and takes + `SerialNumber`, `Manufacturer`, `Model`, `DeviceRevision`, `SoftwareRevision`, `HardwareRevision`, + `DeviceManual`, `DeviceClass`, `RevisionCounter`, `ManufacturerUri`, `ProductCode`, + `ProductInstanceUri`, `AssetId`, `ComponentName` from it; +* deriving a stable `LocalId` from `Manufacturer` + `SerialNumber` when `LocalId` is empty. + +Leaving it unset (String type with an empty string, or numeric `0` in namespace `0`) simply skips +that step and logs a warning. A node that does not exist, or properties you have no rights to read, +are skipped as well — they never make `addDevice` fail. The data is read once, at connect time; it is +not refreshed after a reconnect. + +```cpp +cfg.setPropertyValue("DeviceNodeIDType", 1); // String +cfg.setPropertyValue("DeviceNodeIDString", "PLC1"); +cfg.setPropertyValue("DeviceNamespaceIndex", 2); // ns=2;s=PLC1 +``` + +--- + +## 2. Add a monitored node + +```cpp +auto fbType = device.getAvailableFunctionBlockTypes().get("MonitoredItem"); + +auto cfg = fbType.createDefaultConfig(); +cfg.setPropertyValue("NodeIDType", 1); // String +cfg.setPropertyValue("NodeIDString", ".temperature"); +cfg.setPropertyValue("NamespaceIndex", 1); +cfg.setPropertyValue("SamplingInterval", 100); // ms + +auto fb = device.addFunctionBlock("MonitoredItem", cfg); +``` + +### MonitoredItem properties + +| Property | Type | Default | Applied | +|---|---|---|---| +| `LocalId` | String | `""` | at creation only | +| `NodeIDType` | Selection | `1` — `String` | at creation **and** at runtime | +| `NodeIDString` | String | `""` | at creation **and** at runtime | +| `NodeIDNumeric` | Int | `0` | at creation **and** at runtime | +| `NamespaceIndex` | Int | `0` | at creation **and** at runtime | +| `SamplingInterval` | Int | `100` | at creation **and** at runtime | + +`LocalId` is consumed while the block is being created and does not become a property of it. The +other five do, and each write re-reads the configuration, re-validates the node, refreshes the block +status and reconfigures the signals if the data type changed: + +```cpp +fb.setPropertyValue("SamplingInterval", 500); +fb.setPropertyValue("NodeIDString", ".otherNode"); // repoints the block at another node +``` + +--- + +**`LocalId`** — the block's local ID inside the device's function block folder. It also prefixes the +signal names: `ValueSignal` and `DomainSignal`. + +Give it a meaningful value (`"temperature"`, `"pressure"`) to get readable signal names and stable +component paths. If it is empty, or a block with that ID already exists, the module generates +`MonitoredItemFb` instead — silently, so a collision does not fail `addFunctionBlock`. + +--- + +**`NodeIDType` / `NodeIDString` / `NodeIDNumeric` / `NamespaceIndex`** — the address of the OPC UA +node this block reads. They map directly onto an OPC UA NodeID: + +| Config | Resulting NodeID | +|---|---| +| `NodeIDType = 1`, `NodeIDString = ".temperature"`, `NamespaceIndex = 1` | `ns=1;s=.temperature` | +| `NodeIDType = 0`, `NodeIDNumeric = 1234`, `NamespaceIndex = 2` | `ns=2;i=1234` | + +`NodeIDType` decides which of the two identifier properties is used; the unused one is hidden in a UI +and ignored. + +The node must exist, be a `Variable`, and be readable. If it is not, the block goes to `Error` with a +message saying which of the three failed, and it stops polling until the configuration is written +again or the connection is re-established. + +An empty `NodeIDString` while `NodeIDType` is `String` is a configuration error — the most common +reason for a freshly added block to sit in `Error` and never produce data. + +--- + +**`SamplingInterval`** — how often, in milliseconds, this block issues one OPC UA `Read` for its +node. `100` by default. + +This is client-side polling, not an OPC UA subscription: nothing is configured on the server, and the +server's own sampling and publishing settings do not apply. Every successful read publishes a sample, +even when the value has not changed — there is no deadband or change filter. + +The value must be greater than `0` and fit into 32 bits. Anything else (`0`, a negative number, a +huge number) is rejected: the block reports a `Config` error and keeps running at the 100 ms default. + +The interval is a target, not a guarantee. All blocks of one device are polled by a single thread, so +their reads are serialized: with many blocks, short intervals, or a slow server, the actual periods +stretch. Deadlines are moved forward when that happens, so a stall is never followed by a burst of +catch-up reads. As a rule of thumb, keep `N_blocks × read_round_trip` well below the shortest +interval you configure. + +Add as many blocks as you need; they all share that one thread per device. + +--- + +## 3. Read the values + +```cpp +auto valueSignal = fb.getSignals()[0]; // the domain signal is hidden + +daq::BaseObjectPtr value; +daq::BaseObjectPtr timestamp = valueSignal.getLastValueWithTimestamp(value); + +if (value.assigned()) + std::cout << timestamp << " -> " << value << std::endl; +``` + +`getLastValueWithTimestamp` hands back the value and the moment it carries in one call, already +resolved against the domain signal: the timestamp is an integer number of **microseconds since +1970-01-01**. Both come back unassigned when nothing has been read yet, and the timestamp alone stays +unassigned when `TimestampMode` is `None`, because then there is no domain signal to resolve it +against. + +Streaming works with the usual openDAQ readers: + +```cpp +auto reader = daq::StreamReaderBuilder() + .setSignal(fb.getSignals()[0]) + .setValueReadType(daq::SampleType::Float64) + .setDomainReadType(daq::SampleType::UInt64) + .setSkipEvents(true) + .build(); + +double values[64]; +uint64_t domain[64]; + +daq::SizeT count = std::size(values); +reader.readWithDomain(values, domain, &count, 1000); // waits up to 1000 ms + +for (daq::SizeT i = 0; i < count; ++i) + std::cout << domain[i] << " -> " << values[i] << std::endl; +``` + +The reader delivers samples as they are polled, so `count` comes back with however many arrived +within the timeout — with a 100 ms `SamplingInterval` that is about ten per second. + +The value signal's sample type follows the node: floats → `Float32`/`Float64`, integers → the +matching `Int*`/`UInt*`, `String`/`LocalizedText`/`QualifiedName` → `String`, `DateTime` → `Int64` +microseconds since `1970-01-01T00:00:00Z` (the signal carries the matching time descriptor — unit +`s`, tick resolution `1 / 1'000'000`, that origin). Arrays, structures and booleans are not +supported. + +The domain signal is always `UInt64` microseconds since `1970-01-01T00:00:00Z`. + +--- + +## 4. Check that it works + +```cpp +// per node +fb.getStatusContainer().getStatus("ComponentStatus"); // Ok / Error +fb.getStatusContainer().getStatusMessage("ComponentStatus"); // why it is not Ok + +// per connection +device.getStatusContainer().getStatus("ConnectionStatus"); // Connected / Reconnecting / Unrecoverable +``` + +A block reports `Error` when its configuration is incomplete (empty `NodeIDString`, bad +`SamplingInterval`), when the node cannot be used (missing, not a Variable, not readable), when the +server's answer is unusable (bad status, no value, missing timestamp for the selected +`TimestampMode`), or when the value type is not supported. The message says which. + +A lost connection is detected within 5 seconds: the device goes to `Reconnecting`, polling pauses, +and after a successful reconnect every block re-validates its node and resumes. + +--- + +## 5. Clean up + +```cpp +device.removeFunctionBlock(fb); +instance.removeDevice(device); +``` + +--- diff --git a/shared/libraries/opcua/opcuashared/include/opcuashared/opcuadatavalue.h b/shared/libraries/opcua/opcuashared/include/opcuashared/opcuadatavalue.h index 42e69679..a5663a36 100644 --- a/shared/libraries/opcua/opcuashared/include/opcuashared/opcuadatavalue.h +++ b/shared/libraries/opcua/opcuashared/include/opcuashared/opcuadatavalue.h @@ -43,6 +43,10 @@ class OpcUaDataValue : public OpcUaObject bool hasSourceTimestamp() const; UA_DateTime getSourceTimestampUnixEpoch() const; // us + // Valid only when isDateTime(): the DateTime scalar re-based from OPC UA's 100 ns ticks since + // 1601-01-01 to microseconds since the UNIX epoch. Signed, so pre-1970 dates stay negative. + int64_t getDateTimeValueUnixEpoch() const; // us + bool isStatusOK() const; bool isInteger() const; @@ -51,6 +55,7 @@ class OpcUaDataValue : public OpcUaObject bool isNull() const; bool isReal() const; bool isNumber() const; + bool isDateTime() const; std::string toString() const; int64_t toInteger() const; diff --git a/shared/libraries/opcua/opcuashared/src/opcuadatavalue.cpp b/shared/libraries/opcua/opcuashared/src/opcuadatavalue.cpp index 5d47e000..1fa4221b 100644 --- a/shared/libraries/opcua/opcuashared/src/opcuadatavalue.cpp +++ b/shared/libraries/opcua/opcuashared/src/opcuadatavalue.cpp @@ -36,6 +36,14 @@ UA_DateTime OpcUaDataValue::getSourceTimestampUnixEpoch() const return (hasSourceTimestamp()) ? toUnixTimeUs(getDataValue().sourceTimestamp) : 0; } +int64_t OpcUaDataValue::getDateTimeValueUnixEpoch() const +{ + if (!isDateTime()) + return 0; + const UA_DateTime date = *static_cast(value.value.data); + return (date - UA_DATETIME_UNIX_EPOCH) / UA_DATETIME_USEC; +} + const UA_DataValue& OpcUaDataValue::getDataValue() const { return OpcUaObject::getValue(); @@ -92,6 +100,13 @@ bool OpcUaDataValue::isNumber() const return isInteger() || isReal(); } +bool OpcUaDataValue::isDateTime() const +{ + const UA_Variant& variant = value.value; + return VariantUtils::IsScalar(variant) && variant.type != nullptr && + variant.type->typeKind == UA_DATATYPEKIND_DATETIME; +} + std::string OpcUaDataValue::toString() const { return VariantUtils::ToString(value.value); diff --git a/shared/libraries/opcua/opcuashared/tests/test_opcuadatavalue.cpp b/shared/libraries/opcua/opcuashared/tests/test_opcuadatavalue.cpp index 090a672a..4a5ae228 100644 --- a/shared/libraries/opcua/opcuashared/tests/test_opcuadatavalue.cpp +++ b/shared/libraries/opcua/opcuashared/tests/test_opcuadatavalue.cpp @@ -97,4 +97,67 @@ TEST_F(OpcUaDataValueTest, TestCopyBehaviour) ASSERT_EQ(value.toInteger(), 5); } +static bool isDateTimeOf(const void* val, const UA_DataType* type) +{ + UA_DataValue dataValue; + UA_DataValue_init(&dataValue); + + UA_Variant_setScalarCopy(&dataValue.value, val, type); + dataValue.hasValue = true; + + OpcUaDataValue value(dataValue, true); + const bool result = value.isDateTime(); + + UA_DataValue_clear(&dataValue); + return result; +} + +TEST_F(OpcUaDataValueTest, IsDateTime) +{ + const UA_DateTime dateTime = UA_DateTime_fromUnixTime(1700000000); + ASSERT_TRUE(isDateTimeOf(&dateTime, &UA_TYPES[UA_TYPES_DATETIME])); + + // UtcTime is a subtype of DateTime carrying its own UA_DataType entry + const UA_UtcTime utcTime = UA_DateTime_fromUnixTime(1700000001); + ASSERT_TRUE(isDateTimeOf(&utcTime, &UA_TYPES[UA_TYPES_UTCTIME])); + + const UA_Int64 int64Value = 1700000000; + ASSERT_FALSE(isDateTimeOf(&int64Value, &UA_TYPES[UA_TYPES_INT64])); + + const UA_Double doubleValue = 1.5; + ASSERT_FALSE(isDateTimeOf(&doubleValue, &UA_TYPES[UA_TYPES_DOUBLE])); + + ASSERT_FALSE(OpcUaDataValue().isDateTime()); +} + +TEST_F(OpcUaDataValueTest, DateTimeValueToUnixEpoch) +{ + const auto unixUsOf = [](UA_DateTime date, const UA_DataType* type) + { + UA_DataValue dataValue; + UA_DataValue_init(&dataValue); + + UA_Variant_setScalarCopy(&dataValue.value, &date, type); + dataValue.hasValue = true; + + OpcUaDataValue value(dataValue, true); + const int64_t result = value.getDateTimeValueUnixEpoch(); + + UA_DataValue_clear(&dataValue); + return result; + }; + + ASSERT_EQ(unixUsOf(UA_DateTime_fromUnixTime(1700000000), &UA_TYPES[UA_TYPES_DATETIME]), 1700000000LL * 1000000); + ASSERT_EQ(unixUsOf(UA_DateTime_fromUnixTime(1700000001), &UA_TYPES[UA_TYPES_UTCTIME]), 1700000001LL * 1000000); + + // the UNIX epoch itself, and a date before it, which must stay negative rather than wrap + ASSERT_EQ(unixUsOf(UA_DATETIME_UNIX_EPOCH, &UA_TYPES[UA_TYPES_DATETIME]), 0); + ASSERT_EQ(unixUsOf(UA_DateTime_fromUnixTime(-1), &UA_TYPES[UA_TYPES_DATETIME]), -1000000); + + // 0 ticks is 1601-01-01, not a null timestamp, when it comes from a node's value + ASSERT_EQ(unixUsOf(0, &UA_TYPES[UA_TYPES_DATETIME]), -11644473600LL * 1000000); + + ASSERT_EQ(OpcUaDataValue().getDateTimeValueUnixEpoch(), 0); +} + END_NAMESPACE_OPENDAQ_OPCUA diff --git a/shared/libraries/opcuageneric/opcuageneric_client/include/opcuageneric_client/generic_client_device_impl.h b/shared/libraries/opcuageneric/opcuageneric_client/include/opcuageneric_client/generic_client_device_impl.h index 6eeea5b1..0f85aa42 100644 --- a/shared/libraries/opcuageneric/opcuageneric_client/include/opcuageneric_client/generic_client_device_impl.h +++ b/shared/libraries/opcuageneric/opcuageneric_client/include/opcuageneric_client/generic_client_device_impl.h @@ -25,6 +25,7 @@ #include #include #include +#include BEGIN_NAMESPACE_OPENDAQ_OPCUA_GENERIC @@ -69,6 +70,9 @@ class OpcuaGenericClientDeviceImpl : public Device daq::opcua::OpcUaClientPtr client; DomainSource domainSource; + // Drives every monitored item of this device from a single thread. + SamplingScheduler sampler; + // Reconnect monitor const uint32_t reconnectIntervalMs; std::thread reconnectThread; diff --git a/shared/libraries/opcuageneric/opcuageneric_client/include/opcuageneric_client/opcua_monitored_item_fb_impl.h b/shared/libraries/opcuageneric/opcuageneric_client/include/opcuageneric_client/opcua_monitored_item_fb_impl.h index 9b4e1ec4..beb61d89 100644 --- a/shared/libraries/opcuageneric/opcuageneric_client/include/opcuageneric_client/opcua_monitored_item_fb_impl.h +++ b/shared/libraries/opcuageneric/opcuageneric_client/include/opcuageneric_client/opcua_monitored_item_fb_impl.h @@ -18,13 +18,15 @@ #include #include #include +#include +#include #include #include #include "opcuaclient/opcuaclient.h" BEGIN_NAMESPACE_OPENDAQ_OPCUA_GENERIC -class OpcUaMonitoredItemFbImpl final : public FunctionBlock +class OpcUaMonitoredItemFbImpl final : public FunctionBlock, public ISampledItem { friend class GenericOpcuaMonitoredItemTest; @@ -35,12 +37,18 @@ class OpcUaMonitoredItemFbImpl final : public FunctionBlock daq::opcua::OpcUaClientPtr client, const std::string& localId, DomainSource defaultDomainSource, + SamplingScheduler* scheduler = nullptr, const PropertyObjectPtr& config = nullptr); ~OpcUaMonitoredItemFbImpl(); DAQ_OPCUA_GENERIC_MODULE_API static FunctionBlockTypePtr CreateType(); void setDomainSource(DomainSource domainSource); + uint32_t getSamplingInterval() const override; + void processSample() override; + void onConnectionRestored() override; + void onSchedulerDestroyed() override; + protected: struct DataPackets { @@ -51,7 +59,6 @@ class OpcUaMonitoredItemFbImpl final : public FunctionBlock struct FbConfig { OpcUaNodeId nodeId; - uint32_t samplingInterval; DomainSource domainSource; }; @@ -68,8 +75,12 @@ class OpcUaMonitoredItemFbImpl final : public FunctionBlock daq::opcua::OpcUaClientPtr client; OpcUaNodeId nodeDataType; - std::thread readerThread; - std::atomic running; + std::atomic samplingIntervalMs{DEFAULT_OPCUA_MIFB_SAMPLING_INTERVAL}; + + // Not owned. The device owns the scheduler and destroys it before the component tree releases this + // block, so the scheduler clears this pointer from its destructor. Atomic because removed() and that + // teardown can reach it from different threads. + std::atomic scheduler; std::recursive_mutex processingMutex; std::shared_ptr statuses; @@ -83,6 +94,7 @@ class OpcUaMonitoredItemFbImpl final : public FunctionBlock static std::string generateLocalId(); void initStatusContainer(); + static DataDescriptorPtr buildTimeDescriptor(daq::SampleType sampleType); void adjustSignalDescriptor(); void createSignal(); void reconfigureSignal(const FbConfig& prevConfig); @@ -98,8 +110,7 @@ class OpcUaMonitoredItemFbImpl final : public FunctionBlock bool validateResponse(const OpcUaDataValue& value); bool validateValueDataType(const OpcUaDataValue& value); - void runReaderThread(); - void readerLoop(); + void detachFromScheduler(); DataPackets buildDataPacket(const OpcUaDataValue& value); daq::DataPacketPtr buildDomainDataPacket(const OpcUaDataValue& value); diff --git a/shared/libraries/opcuageneric/opcuageneric_client/include/opcuageneric_client/sampling_scheduler.h b/shared/libraries/opcuageneric/opcuageneric_client/include/opcuageneric_client/sampling_scheduler.h new file mode 100644 index 00000000..be70d9b3 --- /dev/null +++ b/shared/libraries/opcuageneric/opcuageneric_client/include/opcuageneric_client/sampling_scheduler.h @@ -0,0 +1,88 @@ +#pragma once +#include +#include +#include +#include +#include +#include +#include +#include + +BEGIN_NAMESPACE_OPENDAQ_OPCUA_GENERIC + +class ISampledItem +{ +public: + virtual ~ISampledItem() = default; + + // Sampling period in milliseconds. + // It must not take any lock that could be held while processSample() runs. + virtual uint32_t getSamplingInterval() const = 0; + + // Performs one read and publishes the result. + virtual void processSample() = 0; + + // Called once per item after the connection has been re-established. + virtual void onConnectionRestored() = 0; + + // Called once for every still-registered item while the scheduler is being destroyed. The item + // must drop its back-pointer here: the scheduler is gone by the time the item itself is torn down. + virtual void onSchedulerDestroyed() = 0; +}; + +// Drives all monitored items of a device from a single thread. Each item keeps its own deadline. +class SamplingScheduler +{ +public: + using Clock = std::chrono::steady_clock; + using TimePoint = Clock::time_point; + + explicit SamplingScheduler(std::function isConnected); + ~SamplingScheduler(); + + SamplingScheduler(const SamplingScheduler&) = delete; + SamplingScheduler& operator=(const SamplingScheduler&) = delete; + + void start(); + void stop(); + void registerItem(ISampledItem* item); + + // Removes the item and waits for an in-progress processSample() on it to return, so that the + // caller can destroy the item afterwards. + void unregisterItem(ISampledItem* item); + + // Requests onConnectionRestored() for every item. + void onReconnected(); + + static TimePoint advanceDeadline(TimePoint due, TimePoint now, std::chrono::milliseconds interval); + +private: + struct Entry + { + ISampledItem* item; + TimePoint nextDue; + }; + + static constexpr std::chrono::milliseconds DISCONNECTED_POLL_INTERVAL{1000}; + + void loop(); + void revalidateItems(); + + // Runs fn on the item outside of the mutex while keeping unregisterItem() correct. + void invokeUnlocked(std::unique_lock& lock, ISampledItem* item, const std::function& fn); + + std::function isConnected; + + std::thread thread; + std::atomic running{false}; + std::atomic revalidatePending{false}; + + std::mutex mutex; + std::condition_variable cv; + std::vector items; + + ISampledItem* inFlight{nullptr}; + std::condition_variable inFlightCv; +}; + +END_NAMESPACE_OPENDAQ_OPCUA_GENERIC diff --git a/shared/libraries/opcuageneric/opcuageneric_client/src/CMakeLists.txt b/shared/libraries/opcuageneric/opcuageneric_client/src/CMakeLists.txt index b8173e8b..2094d23c 100644 --- a/shared/libraries/opcuageneric/opcuageneric_client/src/CMakeLists.txt +++ b/shared/libraries/opcuageneric/opcuageneric_client/src/CMakeLists.txt @@ -7,12 +7,14 @@ set(SRC_PublicHeaders constants.h opcuageneric.h generic_client_device_impl.h opcua_monitored_item_fb_impl.h + sampling_scheduler.h status_container.h status_adaptor.h property_helper.h ) set(SRC_Cpp generic_client_device_impl.cpp opcua_monitored_item_fb_impl.cpp + sampling_scheduler.cpp ) source_group("common" FILES ${HEADERS_DIR}/constants.h @@ -27,6 +29,9 @@ source_group("device" FILES ${HEADERS_DIR}/generic_client_device_impl.h source_group("function_block" FILES ${HEADERS_DIR}/opcua_monitored_item_fb_impl.h opcua_monitored_item_fb_impl.cpp ) +source_group("scheduler" FILES ${HEADERS_DIR}/sampling_scheduler.h + sampling_scheduler.cpp +) opendaq_prepend_include(${LIB_NAME} SRC_PublicHeaders) diff --git a/shared/libraries/opcuageneric/opcuageneric_client/src/generic_client_device_impl.cpp b/shared/libraries/opcuageneric/opcuageneric_client/src/generic_client_device_impl.cpp index 051510f0..7290131d 100644 --- a/shared/libraries/opcuageneric/opcuageneric_client/src/generic_client_device_impl.cpp +++ b/shared/libraries/opcuageneric/opcuageneric_client/src/generic_client_device_impl.cpp @@ -20,6 +20,7 @@ OpcuaGenericClientDeviceImpl::OpcuaGenericClientDeviceImpl(const ContextPtr& ctx : Device(ctx, parent, localId.empty() ? generateLocalId() : localId) , connectionStatus("ConnectionStatusType", "ConnectionStatus", statusContainer, "Connected", context.getTypeManager()) , client(client) + , sampler([this] { return this->client->isConnected(); }) , reconnectIntervalMs(reconnectIntervalMs) { if (this->client == nullptr) @@ -35,12 +36,14 @@ OpcuaGenericClientDeviceImpl::OpcuaGenericClientDeviceImpl(const ContextPtr& ctx initComponentStatus(); initNestedFbTypes(); + sampler.start(); startReconnectMonitor(); } OpcuaGenericClientDeviceImpl::~OpcuaGenericClientDeviceImpl() { stopReconnectMonitor(); + sampler.stop(); } PropertyObjectPtr OpcuaGenericClientDeviceImpl::createDefaultConfig() @@ -118,6 +121,8 @@ std::string OpcuaGenericClientDeviceImpl::getConnectionString() const void OpcuaGenericClientDeviceImpl::removed() { stopReconnectMonitor(); + // Stopped before the function blocks are torn down, so that no tick can reach a dying item. + sampler.stop(); Device::removed(); client->disconnect(false); } @@ -158,6 +163,7 @@ void OpcuaGenericClientDeviceImpl::reconnectMonitorLoop() client->connect(); client->runIterate(); connectionStatus.setStatus("Connected"); + sampler.onReconnected(); } catch (const OpcUaException& e) { @@ -209,7 +215,7 @@ FunctionBlockPtr OpcuaGenericClientDeviceImpl::onAddFunctionBlock(const StringPt userSpecifiedLocalId = config.getPropertyValue(PROPERTY_NAME_OPCUA_MI_LOCAL_ID).asPtr().toStdString(); const auto localId = buildMILocalId(userSpecifiedLocalId); nestedFunctionBlock = createWithImplementation( - context, functionBlocks, fbTypePtr, client, localId, domainSource, config); + context, functionBlocks, fbTypePtr, client, localId, domainSource, &sampler, config); } else { @@ -223,6 +229,7 @@ FunctionBlockPtr OpcuaGenericClientDeviceImpl::onAddFunctionBlock(const StringPt auto lock = this->getRecursiveConfigLock2(); addNestedFunctionBlock(nestedFunctionBlock); } + sampler.registerItem(static_cast(*nestedFunctionBlock)); setComponentStatus(ComponentStatus::Ok); } else diff --git a/shared/libraries/opcuageneric/opcuageneric_client/src/opcua_monitored_item_fb_impl.cpp b/shared/libraries/opcuageneric/opcuageneric_client/src/opcua_monitored_item_fb_impl.cpp index 65acb5e8..d149e437 100644 --- a/shared/libraries/opcuageneric/opcuageneric_client/src/opcua_monitored_item_fb_impl.cpp +++ b/shared/libraries/opcuageneric/opcuageneric_client/src/opcua_monitored_item_fb_impl.cpp @@ -4,6 +4,8 @@ #include "opendaq/binary_data_packet_factory.h" #include "opendaq/packet_factory.h" #include +#include +#include #define DISABLE_NODE_DATATYPE_VALIDATION @@ -66,10 +68,11 @@ OpcUaMonitoredItemFbImpl::OpcUaMonitoredItemFbImpl(const ContextPtr& ctx, daq::opcua::OpcUaClientPtr client, const std::string& localId, DomainSource defaultDomainSource, + SamplingScheduler* scheduler, const PropertyObjectPtr& config) : FunctionBlock(type, ctx, parent, localId.empty() ? generateLocalId() : localId) , client(client) - , running(false) + , scheduler(scheduler) , statuses(std::make_shared()) { initComponentStatus(); @@ -85,28 +88,32 @@ OpcUaMonitoredItemFbImpl::OpcUaMonitoredItemFbImpl(const ContextPtr& ctx, adjustSignalDescriptor(); createSignal(); updateStatuses(); - runReaderThread(); } OpcUaMonitoredItemFbImpl::~OpcUaMonitoredItemFbImpl() { - if (readerThread.joinable()) - { - running = false; - readerThread.join(); - } + detachFromScheduler(); } void OpcUaMonitoredItemFbImpl::removed() { - if (readerThread.joinable()) - { - running = false; - readerThread.join(); - } + detachFromScheduler(); FunctionBlock::removed(); } +void OpcUaMonitoredItemFbImpl::detachFromScheduler() +{ + // Returns only once an in-progress processSample() on this item has finished, so the object can + // be torn down afterwards. + if (auto* sched = scheduler.exchange(nullptr); sched != nullptr) + sched->unregisterItem(this); +} + +void OpcUaMonitoredItemFbImpl::onSchedulerDestroyed() +{ + scheduler.store(nullptr); +} + void OpcUaMonitoredItemFbImpl::initStatusContainer() { configErr = statuses->addStatus("Config"); @@ -176,7 +183,7 @@ FunctionBlockTypePtr OpcUaMonitoredItemFbImpl::CreateType() void OpcUaMonitoredItemFbImpl::setDomainSource(DomainSource domainSource) { - auto lock = this->getRecursiveConfigLock(); + auto lock = this->getRecursiveConfigLock2(); auto lockProcessing = std::scoped_lock(processingMutex); if (config.domainSource != domainSource) { @@ -191,12 +198,27 @@ std::string OpcUaMonitoredItemFbImpl::generateLocalId() return std::string(OPCUA_LOCAL_MONITORED_ITEM_FB_ID_PREFIX + std::to_string(localIndex++)); } +DataDescriptorPtr OpcUaMonitoredItemFbImpl::buildTimeDescriptor(daq::SampleType sampleType) +{ + return DataDescriptorBuilder() + .setSampleType(sampleType) + .setRule(ExplicitDataRule()) + .setUnit(Unit("s", -1, "seconds", "time")) + .setTickResolution(Ratio(1, 1'000'000)) + .setOrigin("1970-01-01T00:00:00Z") + .setName("Time") + .build(); +} + void OpcUaMonitoredItemFbImpl::adjustSignalDescriptor() { auto lockProcessing = std::scoped_lock(processingMutex); if (nodeValidationErr.ok() && supportedDataTypeNodeIds.count(nodeDataType) != 0) { - outputSignalDescriptor = DataDescriptorBuilder().setSampleType(supportedDataTypeNodeIds[nodeDataType]).build(); + if (nodeDataType == OpcUaNodeId(0, UA_NS0ID_DATETIME)) + outputSignalDescriptor = buildTimeDescriptor(supportedDataTypeNodeIds[nodeDataType]); + else + outputSignalDescriptor = DataDescriptorBuilder().setSampleType(supportedDataTypeNodeIds[nodeDataType]).build(); } else { @@ -233,7 +255,7 @@ void OpcUaMonitoredItemFbImpl::readProperties() { using namespace property_helper; - auto lock = this->getRecursiveConfigLock(); + auto lock = this->getRecursiveConfigLock2(); auto lockProcessing = std::scoped_lock(processingMutex); configErr.reset(); @@ -256,13 +278,17 @@ void OpcUaMonitoredItemFbImpl::readProperties() config.nodeId = OpcUaNodeId{static_cast(namespaceIndex), nodeIdNumeric}; } - config.samplingInterval = - readProperty(objPtr, PROPERTY_NAME_OPCUA_SAMPLING_INTERVAL, DEFAULT_OPCUA_MIFB_SAMPLING_INTERVAL); - if (config.samplingInterval <= 0) + const auto samplingInterval = + readProperty(objPtr, PROPERTY_NAME_OPCUA_SAMPLING_INTERVAL, DEFAULT_OPCUA_MIFB_SAMPLING_INTERVAL); + if (samplingInterval <= 0 || samplingInterval > static_cast(std::numeric_limits::max())) { configErr.add(fmt::format("Invalid value for the \"{}\" property! Sampling interval must be a positive integer.", PROPERTY_NAME_OPCUA_SAMPLING_INTERVAL)); - config.samplingInterval = DEFAULT_OPCUA_MIFB_SAMPLING_INTERVAL; + samplingIntervalMs = DEFAULT_OPCUA_MIFB_SAMPLING_INTERVAL; + } + else + { + samplingIntervalMs = static_cast(samplingInterval); } updateStatuses(); @@ -270,7 +296,7 @@ void OpcUaMonitoredItemFbImpl::readProperties() void OpcUaMonitoredItemFbImpl::propertyChanged() { - auto lock = this->getRecursiveConfigLock(); + auto lock = this->getRecursiveConfigLock2(); auto lockProcessing = std::scoped_lock(processingMutex); statuses->resetAll(); @@ -366,6 +392,11 @@ bool OpcUaMonitoredItemFbImpl::validateResponse(const OpcUaDataValue& value) responseValidationErr.set(std::string("Reading value error: response without a value.")); return false; } + if (value.isNull()) + { + responseValidationErr.set(std::string("Reading value error: response with an empty value.")); + return false; + } if (config.domainSource == DomainSource::ServerTimestamp && (!value.getValue().hasServerTimestamp || value.getValue().serverTimestamp == 0)) { responseValidationErr.set(std::string("Reading value error: there is no required server timestamp")); @@ -407,7 +438,7 @@ bool OpcUaMonitoredItemFbImpl::validateValueDataType(const OpcUaDataValue& value void OpcUaMonitoredItemFbImpl::createSignal() { - auto lock = this->getRecursiveConfigLock(); + auto lock = this->getRecursiveConfigLock2(); LOG_I("Creating a signal..."); outputSignal = createAndAddSignal(OPCUA_VALUE_SIGNAL_LOCAL_ID, outputSignalDescriptor); @@ -418,7 +449,7 @@ void OpcUaMonitoredItemFbImpl::createSignal() void OpcUaMonitoredItemFbImpl::reconfigureSignal(const FbConfig& prevConfig) { - auto lock = this->getRecursiveConfigLock(); + auto lock = this->getRecursiveConfigLock2(); auto lockProcessing = std::scoped_lock(processingMutex); if (config.domainSource == DomainSource::None) @@ -442,66 +473,77 @@ void OpcUaMonitoredItemFbImpl::reconfigureSignal(const FbConfig& prevConfig) SignalConfigPtr OpcUaMonitoredItemFbImpl::createDomainSignal() { - auto lock = this->getRecursiveConfigLock(); - - const auto domainSignalDsc = DataDescriptorBuilder() - .setSampleType(SampleType::UInt64) - .setRule(ExplicitDataRule()) - .setUnit(Unit("s", -1, "seconds", "time")) - .setTickResolution(Ratio(1, 1'000'000)) - .setOrigin("1970-01-01T00:00:00Z") - .setName("Time") - .build(); + auto lock = this->getRecursiveConfigLock2(); + + const auto domainSignalDsc = buildTimeDescriptor(SampleType::UInt64); outputDomainSignal = createAndAddSignal(OPCUA_TS_SIGNAL_LOCAL_ID, domainSignalDsc, false); outputDomainSignal.setName(localId.toStdString() + "DomainSignal"); return outputDomainSignal; } -void OpcUaMonitoredItemFbImpl::runReaderThread() +uint32_t OpcUaMonitoredItemFbImpl::getSamplingInterval() const { - running = true; - readerThread = std::thread([this] { readerLoop(); }); + return samplingIntervalMs.load(); } -void OpcUaMonitoredItemFbImpl::readerLoop() +void OpcUaMonitoredItemFbImpl::processSample() { - auto start = std::chrono::high_resolution_clock::now(); - while (running) { - auto nextTP = start; + auto lockProcessing = std::scoped_lock(processingMutex); + if (configErr.ok() && nodeValidationErr.ok()) { - auto lockProcessing = std::scoped_lock(processingMutex); - nextTP += std::chrono::milliseconds(config.samplingInterval); - if (configErr.ok() && nodeValidationErr.ok()) + OpcUaDataValue dataValue; + try { - OpcUaDataValue dataValue; - try - { - dataValue = client->readDataValue(config.nodeId); + dataValue = client->readDataValue(config.nodeId); - exceptionErr.reset(); - if (validateResponse(dataValue) && validateValueDataType(dataValue)) + exceptionErr.reset(); + if (validateResponse(dataValue) && validateValueDataType(dataValue)) + { + const auto dps = buildDataPacket(dataValue); + if (dps.dataPacket.assigned()) { - const auto dps = buildDataPacket(dataValue); if (dps.domainDataPacket.assigned() && outputDomainSignal.assigned()) outputDomainSignal.sendPacket(dps.domainDataPacket); outputSignal.sendPacket(dps.dataPacket); } + else + { + valueValidationErr.set(fmt::format("Failed to build a packet for value type ({}).", + static_cast(dataValue.getValue().value.type->typeKind))); + } } - catch (OpcUaException&) - { - exceptionErr.set("Exception while reading."); - } + } + catch (const OpcUaException&) + { + exceptionErr.set("Exception while reading."); + } + catch (const std::exception& e) + { + exceptionErr.set(fmt::format("Exception while reading: {}", e.what())); + } + catch (...) + { + exceptionErr.set("Unknown exception while reading."); } } - updateStatuses(); - auto now = std::chrono::high_resolution_clock::now(); - std::chrono::microseconds sleepTime(0); - if (now < nextTP) - sleepTime = std::chrono::duration_cast(nextTP - now); - start = nextTP; - std::this_thread::sleep_for(sleepTime); } + updateStatuses(); +} + +void OpcUaMonitoredItemFbImpl::onConnectionRestored() +{ + auto lock = this->getRecursiveConfigLock2(); + auto lockProcessing = std::scoped_lock(processingMutex); + + // The node may have disappeared or changed its data type while the connection was down, so the + // validation done at construction time is redone against the reconnected server. + statuses->resetAll(); + + validateNode(); + adjustSignalDescriptor(); + reconfigureSignal(config); + updateStatuses(); } OpcUaMonitoredItemFbImpl::DataPackets OpcUaMonitoredItemFbImpl::buildDataPacket(const OpcUaDataValue& value) @@ -515,7 +557,7 @@ OpcUaMonitoredItemFbImpl::DataPackets OpcUaMonitoredItemFbImpl::buildDataPacket( dps.dataPacket = daq::BinaryDataPacket(dps.domainDataPacket, outputSignalDescriptor, convertedValue.size()); std::memcpy(dps.dataPacket.getRawData(), convertedValue.data(), convertedValue.size()); } - else if (value.isInteger() || value.isReal()) + else if (value.isInteger() || value.isReal() || value.isDateTime()) { if (dps.domainDataPacket.assigned()) dps.dataPacket = daq::DataPacketWithDomain(dps.domainDataPacket, outputSignalDescriptor, 1); @@ -549,7 +591,8 @@ OpcUaMonitoredItemFbImpl::DataPackets OpcUaMonitoredItemFbImpl::buildDataPacket( *(static_cast(dps.dataPacket.getRawData())) = value.readScalar(); break; case UA_DATATYPEKIND_DATETIME: - *(static_cast(dps.dataPacket.getRawData())) = value.readScalar(); + // OPC UA counts 100 ns ticks from 1601-01-01; the descriptor declares us from the UNIX epoch + *(static_cast(dps.dataPacket.getRawData())) = value.getDateTimeValueUnixEpoch(); break; case UA_DATATYPEKIND_FLOAT: *(static_cast(dps.dataPacket.getRawData())) = value.readScalar(); diff --git a/shared/libraries/opcuageneric/opcuageneric_client/src/sampling_scheduler.cpp b/shared/libraries/opcuageneric/opcuageneric_client/src/sampling_scheduler.cpp new file mode 100644 index 00000000..dff0122b --- /dev/null +++ b/shared/libraries/opcuageneric/opcuageneric_client/src/sampling_scheduler.cpp @@ -0,0 +1,170 @@ +#include +#include + +BEGIN_NAMESPACE_OPENDAQ_OPCUA_GENERIC + +SamplingScheduler::SamplingScheduler(std::function isConnected) + : isConnected(std::move(isConnected)) +{ +} + +SamplingScheduler::~SamplingScheduler() +{ + stop(); + + // stop() has joined the thread, so nothing else can touch `items` from here on. Anything still in + // the list belongs to a component that was never removed(), and it outlives this object - tell it + // to forget us before its own destructor tries to unregister from freed memory. + std::vector leftover; + { + std::scoped_lock lock(mutex); + leftover.swap(items); + } + + for (const auto& entry : leftover) + entry.item->onSchedulerDestroyed(); +} + +SamplingScheduler::TimePoint SamplingScheduler::advanceDeadline(TimePoint due, TimePoint now, std::chrono::milliseconds interval) +{ + if (interval < std::chrono::milliseconds(1)) + interval = std::chrono::milliseconds(1); + + due += interval; + return due > now ? due : now + interval; +} + +void SamplingScheduler::start() +{ + if (thread.joinable()) + return; + + running = true; + thread = std::thread([this] { loop(); }); +} + +void SamplingScheduler::stop() +{ + { + std::scoped_lock lock(mutex); + running = false; + } + cv.notify_all(); + if (thread.joinable()) + thread.join(); +} + +void SamplingScheduler::registerItem(ISampledItem* item) +{ + if (item == nullptr) + return; + + { + std::scoped_lock lock(mutex); + items.push_back({item, Clock::now()}); + } + cv.notify_all(); +} + +void SamplingScheduler::unregisterItem(ISampledItem* item) +{ + { + std::unique_lock lock(mutex); + items.erase(std::remove_if(items.begin(), items.end(), [item](const Entry& e) { return e.item == item; }), items.end()); + inFlightCv.wait(lock, [this, item] { return inFlight != item; }); + } + cv.notify_all(); +} + +void SamplingScheduler::onReconnected() +{ + revalidatePending = true; + cv.notify_all(); +} + +void SamplingScheduler::invokeUnlocked(std::unique_lock& lock, ISampledItem* item, const std::function& fn) +{ + inFlight = item; + lock.unlock(); + + try + { + fn(item); + } + catch (...) + { + // Items report their own errors through the component status; swallow whatever still escapes + // so that one misbehaving item cannot tear down sampling for the whole device. + } + + lock.lock(); + inFlight = nullptr; + inFlightCv.notify_all(); +} + +void SamplingScheduler::revalidateItems() +{ + std::unique_lock lock(mutex); + + // Snapshot: the list can change while an item is revalidated outside of the mutex. + std::vector pending; + pending.reserve(items.size()); + for (const auto& entry : items) + pending.push_back(entry.item); + + for (auto* item : pending) + { + if (!running) + return; + + const bool stillRegistered = std::any_of(items.begin(), items.end(), [item](const Entry& e) { return e.item == item; }); + if (!stillRegistered) + continue; + + invokeUnlocked(lock, item, [](ISampledItem* i) { i->onConnectionRestored(); }); + } +} + +void SamplingScheduler::loop() +{ + while (running) + { + // Checked without holding the mutex: it takes the client lock, which the reconnect thread can + // be holding for the duration of a connect(). + if (isConnected && !isConnected()) + { + std::unique_lock lock(mutex); + cv.wait_for(lock, DISCONNECTED_POLL_INTERVAL, [this] { return !running.load() || revalidatePending.load(); }); + continue; + } + + if (revalidatePending.exchange(false)) + revalidateItems(); + + std::unique_lock lock(mutex); + if (!running) + break; + + if (items.empty()) + { + cv.wait(lock, [this] { return !running.load() || !items.empty(); }); + continue; + } + + const auto next = + std::min_element(items.begin(), items.end(), [](const Entry& a, const Entry& b) { return a.nextDue < b.nextDue; }); + + const auto now = Clock::now(); + if (next->nextDue > now) + { + cv.wait_until(lock, next->nextDue); + continue; + } + + next->nextDue = advanceDeadline(next->nextDue, now, std::chrono::milliseconds(next->item->getSamplingInterval())); + + invokeUnlocked(lock, next->item, [](ISampledItem* i) { i->processSample(); }); + } +} + +END_NAMESPACE_OPENDAQ_OPCUA_GENERIC diff --git a/shared/libraries/opcuageneric/opcuageneric_client/tests/CMakeLists.txt b/shared/libraries/opcuageneric/opcuageneric_client/tests/CMakeLists.txt index e5df0b36..896b99a8 100644 --- a/shared/libraries/opcuageneric/opcuageneric_client/tests/CMakeLists.txt +++ b/shared/libraries/opcuageneric/opcuageneric_client/tests/CMakeLists.txt @@ -8,6 +8,7 @@ set(TEST_SOURCES test_app.cpp timer.h test_opcua_generic_client_device.cpp test_opcua_monitored_item_fb.cpp + test_sampling_scheduler.cpp ) set(SUPPORTS_ASAN 0) diff --git a/shared/libraries/opcuageneric/opcuageneric_client/tests/opcuaservertesthelper.cpp b/shared/libraries/opcuageneric/opcuageneric_client/tests/opcuaservertesthelper.cpp index a3ee1d97..fcd2c735 100644 --- a/shared/libraries/opcuageneric/opcuageneric_client/tests/opcuaservertesthelper.cpp +++ b/shared/libraries/opcuageneric/opcuageneric_client/tests/opcuaservertesthelper.cpp @@ -172,6 +172,13 @@ void OpcUaServerTestHelper::createModel() UA_StatusCode myStatus = UA_STATUSCODE_GOODSUBSCRIPTIONTRANSFERRED; publishVariable(".sc", &myStatus, &UA_TYPES[UA_TYPES_STATUSCODE], &uaObjectsFolder); + UA_DateTime myDateTime = UA_DateTime_fromUnixTime(1700000000); + publishVariable(".dt", &myDateTime, &UA_TYPES[UA_TYPES_DATETIME], &uaObjectsFolder); + + // UtcTime is a subtype of DateTime and carries its own UA_DataType entry + UA_UtcTime myUtcTime = UA_DateTime_fromUnixTime(1700000001); + publishVariable(".utc", &myUtcTime, &UA_TYPES[UA_TYPES_UTCTIME], &uaObjectsFolder); + // vectors UA_Int32 myVecInt32[] = {12, 13, 15, 18}; diff --git a/shared/libraries/opcuageneric/opcuageneric_client/tests/test_daq_test_helper.h b/shared/libraries/opcuageneric/opcuageneric_client/tests/test_daq_test_helper.h index 93fa88e3..b7cf2d35 100644 --- a/shared/libraries/opcuageneric/opcuageneric_client/tests/test_daq_test_helper.h +++ b/shared/libraries/opcuageneric/opcuageneric_client/tests/test_daq_test_helper.h @@ -5,6 +5,8 @@ #include "opcuageneric_client/common.h" #include "opcuageneric_client/constants.h" #include +#include +#include namespace daq::opcua::generic { @@ -42,6 +44,22 @@ class DaqTestHelper return device; } + // Waits until the reader holds at least `count` packets. Tests wait for the evidence instead of + // assuming a sampling rate: a slow runner takes longer to deliver the packets, it does not deliver + // fewer of them, so a generous timeout keeps the assertion meaningful everywhere. + template + static bool waitForPackets(const ReaderPtr& reader, daq::SizeT count, std::chrono::milliseconds timeout) + { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (reader.getAvailableCount() < count) + { + if (std::chrono::steady_clock::now() >= deadline) + return false; + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + return true; + } + static daq::ModulePtr CreateModule() { daq::ModulePtr module; diff --git a/shared/libraries/opcuageneric/opcuageneric_client/tests/test_opcua_generic_client_device.cpp b/shared/libraries/opcuageneric/opcuageneric_client/tests/test_opcua_generic_client_device.cpp index 17a235ce..e437383c 100644 --- a/shared/libraries/opcuageneric/opcuageneric_client/tests/test_opcua_generic_client_device.cpp +++ b/shared/libraries/opcuageneric/opcuageneric_client/tests/test_opcua_generic_client_device.cpp @@ -10,6 +10,7 @@ #include "opcuageneric_client/constants.h" #include "opcuageneric_client/generic_client_device_impl.h" #include "opcuaservertesthelper.h" +#include "opendaq/reader_factory.h" #include "test_daq_test_helper.h" #include #include @@ -318,6 +319,85 @@ TEST_F(GenericOpcuaClientDeviceTest, ReconnectMonitor_ReconnectsAfterServerResta ASSERT_TRUE(waitForConnectionStatus("Connected")); } +TEST_F(GenericOpcuaClientDeviceTest, SamplingPausesWhileServerIsDownAndResumesWithoutBurst) +{ + constexpr uint32_t interval = 20; + constexpr auto downtime = std::chrono::milliseconds(600); + constexpr auto burstWindow = std::chrono::milliseconds(100); // five intervals + constexpr auto patience = std::chrono::seconds(10); + + DaqInstanceInit(); + createDeviceWithShortInterval(testHelper.getServerUrl()); + ASSERT_TRUE(waitForConnectionStatus("Connected")); + + daq::FunctionBlockPtr fb; + ASSERT_NO_THROW(fb = addMonitoredItemFB(".i32", 1, interval)); + + auto reader = daq::StreamReaderBuilder() + .setSignal(fb.getSignals()[0]) + .setValueReadType(daq::SampleType::Int64) + .setDomainReadType(daq::SampleType::UInt64) + .setSkipEvents(true) + .build(); + + // Wait for the packets rather than for a deadline: a slow runner needs longer, but it still gets + // there, so the assertion keeps its meaning without assuming the 20 ms rate is achieved. + ASSERT_TRUE(waitForPackets(reader, 6u, patience)); + + testHelper.stop(); + ASSERT_TRUE(waitForConnectionStatus("Reconnecting")); + + const auto afterDisconnect = reader.getAvailableCount(); + std::this_thread::sleep_for(downtime); + // Nothing is sampled while the client is down, so no packets appear. + EXPECT_LE(reader.getAvailableCount(), afterDisconnect + 1u); + + const auto beforeRestart = reader.getAvailableCount(); + const auto restartedAt = std::chrono::steady_clock::now(); + testHelper.startServer(); + ASSERT_TRUE(waitForConnectionStatus("Connected")); + std::this_thread::sleep_for(burstWindow); + + // The ~30 reads missed during the downtime must not arrive at once. Sampling resumes as soon as + // the client is back, which is slightly before the status property flips, so the span is measured + // from the restart itself and compared with what the interval allows over it. Being slow only + // lowers the packet count, never the span, so this cannot fail spuriously. + const auto span = std::chrono::duration_cast(std::chrono::steady_clock::now() - restartedAt); + EXPECT_LE(reader.getAvailableCount() - beforeRestart, static_cast(span.count() / interval) + 5u); + + // Sampling did resume, at whatever pace the runner manages. + EXPECT_TRUE(waitForPackets(reader, beforeRestart + 6u, patience)); + + ASSERT_NO_THROW(device.removeFunctionBlock(fb)); +} + +TEST_F(GenericOpcuaClientDeviceTest, DroppingDeviceWithLiveMonitoredItemDoesNotUseDestroyedScheduler) +{ + // The device here is standalone, so removed() never runs and teardown goes through the destructors: + // the device destroys its SamplingScheduler member, and only afterwards does the base Device release + // the function block. The block must not try to unregister from that dead scheduler. + DaqInstanceInit(); + createDeviceWithShortInterval(testHelper.getServerUrl()); + ASSERT_TRUE(waitForConnectionStatus("Connected")); + + daq::FunctionBlockPtr fb; + ASSERT_NO_THROW(fb = addMonitoredItemFB(".i32", 1, 20)); + + auto reader = daq::StreamReaderBuilder() + .setSignal(fb.getSignals()[0]) + .setValueReadType(daq::SampleType::Int64) + .setDomainReadType(daq::SampleType::UInt64) + .setSkipEvents(true) + .build(); + + // Let the scheduler pick the item up, so it is still registered when the device goes away. + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + + reader.release(); + ASSERT_NO_THROW(fb.release()); + ASSERT_NO_THROW(device.release()); +} + TEST_F(GenericOpcuaClientDeviceTest, ReconnectMonitor_StopsCleanlyOnDeviceRemoval) { DaqInstanceInit(); @@ -386,6 +466,85 @@ TEST_F(GenericOpcuaClientDeviceTest, RemovedFBIsIgnoredOnSubsequentTimestampMode ASSERT_NO_THROW(device.setPropertyValue(PROPERTY_NAME_OPCUA_TS_MODE, static_cast(DomainSource::None))); } +TEST_F(GenericOpcuaClientDeviceTest, AddDeviceWithDefaultAddDeviceConfig) +{ + const auto instance = DaqInstanceInit(); + + PropertyObjectPtr config; + ASSERT_NO_THROW(config = instance.createDefaultAddDeviceConfig()); + ASSERT_TRUE(config.assigned()); + + PropertyObjectPtr deviceTypeConfigs = config.getPropertyValue("Device"); + ASSERT_TRUE(deviceTypeConfigs.hasProperty("OPCUAGeneric")); + + PropertyObjectPtr ourConfig = deviceTypeConfigs.getPropertyValue("OPCUAGeneric"); + ourConfig.setPropertyValue(PROPERTY_NAME_OPCUA_TS_MODE, static_cast(DomainSource::ServerTimestamp)); + + ASSERT_NO_THROW(device = instance.addDevice("daq.opcua.generic://127.0.0.1:4842", config)); + ASSERT_EQ(device.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", instance.getContext().getTypeManager())); + + // the value set in the nested section must reach the device + ASSERT_TRUE(device.hasProperty(PROPERTY_NAME_OPCUA_TS_MODE)); + EXPECT_EQ(device.getPropertyValue(PROPERTY_NAME_OPCUA_TS_MODE).asPtr(), + static_cast(DomainSource::ServerTimestamp)); +} + +// A config that is not derived from the device type and carries only a subset of the properties. +TEST_F(GenericOpcuaClientDeviceTest, AddDeviceWithPlainPartialConfig) +{ + const auto instance = DaqInstanceInit(); + + auto config = PropertyObject(); + config.addProperty(IntProperty(PROPERTY_NAME_OPCUA_TS_MODE, static_cast(DomainSource::LocalSystemTimestamp))); + + ASSERT_NO_THROW(device = instance.addDevice("daq.opcua.generic://127.0.0.1:4842", config)); + ASSERT_EQ(device.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", instance.getContext().getTypeManager())); + + ASSERT_TRUE(device.hasProperty(PROPERTY_NAME_OPCUA_TS_MODE)); + EXPECT_EQ(device.getPropertyValue(PROPERTY_NAME_OPCUA_TS_MODE).asPtr(), + static_cast(DomainSource::LocalSystemTimestamp)); +} + +// Properties the module knows nothing about must be ignored, not rejected. +TEST_F(GenericOpcuaClientDeviceTest, AddDeviceWithUnknownPropertiesInConfig) +{ + const auto module = CreateModule(); + const auto instance = DaqInstanceInit(); + + auto config = module.getAvailableDeviceTypes().get("OPCUAGeneric").createDefaultConfig(); + config.addProperty(StringProperty("SomeForeignProperty", "value")); + config.addProperty(IntProperty("AnotherForeignProperty", 42)); + + ASSERT_NO_THROW(device = instance.addDevice("daq.opcua.generic://127.0.0.1:4842", config)); + ASSERT_EQ(device.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", instance.getContext().getTypeManager())); + + EXPECT_FALSE(device.hasProperty("SomeForeignProperty")); + EXPECT_FALSE(device.hasProperty("AnotherForeignProperty")); +} + +// Adding with an explicit config must behave the same as adding with a null config. +TEST_F(GenericOpcuaClientDeviceTest, AddDeviceWithConfigMatchesNullConfig) +{ + const auto module = CreateModule(); + const auto instance = DaqInstanceInit(); + const auto okStatus = Enumeration("ComponentStatusType", "Ok", instance.getContext().getTypeManager()); + + daq::DevicePtr withNullConfig; + ASSERT_NO_THROW(withNullConfig = instance.addDevice("daq.opcua.generic://127.0.0.1:4842", nullptr)); + ASSERT_EQ(withNullConfig.getStatusContainer().getStatus("ComponentStatus"), okStatus); + + auto config = module.getAvailableDeviceTypes().get("OPCUAGeneric").createDefaultConfig(); + daq::DevicePtr withConfig; + ASSERT_NO_THROW(withConfig = instance.addDevice("daq.opcua.generic://127.0.0.1:4842", config)); + ASSERT_EQ(withConfig.getStatusContainer().getStatus("ComponentStatus"), okStatus); + + EXPECT_EQ(withConfig.getInfo().getName(), withNullConfig.getInfo().getName()); + EXPECT_EQ(withConfig.getAllProperties().getCount(), withNullConfig.getAllProperties().getCount()); +} + TEST_F(GenericOpcuaClientDeviceTest, DeviceInfoFilledFromDeviceTypeNode) { using NT = NodeIDType; diff --git a/shared/libraries/opcuageneric/opcuageneric_client/tests/test_opcua_monitored_item_fb.cpp b/shared/libraries/opcuageneric/opcuageneric_client/tests/test_opcua_monitored_item_fb.cpp index f336a2f7..253d3512 100644 --- a/shared/libraries/opcuageneric/opcuageneric_client/tests/test_opcua_monitored_item_fb.cpp +++ b/shared/libraries/opcuageneric/opcuageneric_client/tests/test_opcua_monitored_item_fb.cpp @@ -10,6 +10,9 @@ #include "opendaq/reader_factory.h" #include "test_daq_test_helper.h" #include "timer.h" +#include +#include +#include #define ASSERT_DOUBLE_NE(val1, val2) ASSERT_GT(std::abs((val1) - (val2)), 1e-9) @@ -321,6 +324,39 @@ TEST_F(GenericOpcuaMonitoredItemTest, CreationWithCustomConfig) ASSERT_EQ(fb.getStatusContainer().getStatus("ComponentStatus"), okStatus()); } +// The caller may hand the very same config object to several addFunctionBlock calls. +TEST_F(GenericOpcuaMonitoredItemTest, AddFbWithReusedConfigObject) +{ + StartUp(); + auto config = device.getAvailableFunctionBlockTypes().get(GENERIC_OPCUA_MONITORED_ITEM_FB_NAME).createDefaultConfig(); + config.setPropertyValue(PROPERTY_NAME_OPCUA_NODE_ID_STRING, ".i32"); + config.setPropertyValue(PROPERTY_NAME_OPCUA_NAMESPACE_INDEX, 1); + + daq::FunctionBlockPtr first; + ASSERT_NO_THROW(first = device.addFunctionBlock(GENERIC_OPCUA_MONITORED_ITEM_FB_NAME, config)); + ASSERT_EQ(first.getStatusContainer().getStatus("ComponentStatus"), okStatus()); + + ASSERT_NO_THROW(fb = device.addFunctionBlock(GENERIC_OPCUA_MONITORED_ITEM_FB_NAME, config)); + ASSERT_EQ(fb.getStatusContainer().getStatus("ComponentStatus"), okStatus()); + EXPECT_NE(fb.getLocalId(), first.getLocalId()); + + device.removeFunctionBlock(first); +} + +// Properties the function block knows nothing about must be ignored, not rejected. +TEST_F(GenericOpcuaMonitoredItemTest, AddFbWithUnknownPropertiesInConfig) +{ + StartUp(); + auto config = device.getAvailableFunctionBlockTypes().get(GENERIC_OPCUA_MONITORED_ITEM_FB_NAME).createDefaultConfig(); + config.setPropertyValue(PROPERTY_NAME_OPCUA_NODE_ID_STRING, ".i32"); + config.setPropertyValue(PROPERTY_NAME_OPCUA_NAMESPACE_INDEX, 1); + config.addProperty(StringProperty("SomeForeignProperty", "value")); + + ASSERT_NO_THROW(fb = device.addFunctionBlock(GENERIC_OPCUA_MONITORED_ITEM_FB_NAME, config)); + ASSERT_EQ(fb.getStatusContainer().getStatus("ComponentStatus"), okStatus()); + EXPECT_FALSE(fb.hasProperty("SomeForeignProperty")); +} + TEST_F(GenericOpcuaMonitoredItemTest, TwoFbCreation) { StartUp(); @@ -868,6 +904,8 @@ TEST_F(GenericOpcuaMonitoredItemTest, SignalDescriptorSampleTypeMatchesOpcUaData {OpcUaNodeId(1, ".i32"), SampleType::Int32}, {OpcUaNodeId(1, ".i64"), SampleType::Int64}, {OpcUaNodeId(1, ".s"), SampleType::String}, + {OpcUaNodeId(1, ".dt"), SampleType::Int64}, + {OpcUaNodeId(1, ".utc"), SampleType::Int64}, }; for (const auto& [nodeId, expectedType] : cases) @@ -881,6 +919,42 @@ TEST_F(GenericOpcuaMonitoredItemTest, SignalDescriptorSampleTypeMatchesOpcUaData } } +TEST_F(GenericOpcuaMonitoredItemTest, ReadDateTimeValue) +{ + StartUp(); + + const std::vector> cases = { + {OpcUaNodeId(1, ".dt"), UA_DateTime_fromUnixTime(1700000000)}, + {OpcUaNodeId(1, ".utc"), UA_DateTime_fromUnixTime(1700000001)}, + }; + + for (const auto& [nodeId, expected] : cases) + { + CreateMonitoredItemFB(nodeId.getIdentifier(), nodeId.getNamespaceIndex(), 50); + + EXPECT_EQ(fb.getStatusContainer().getStatus("ComponentStatus"), okStatus()); + + const daq::BaseObjectPtr val = readValueWithTout(fb.getSignals()[0], 300); + ASSERT_TRUE(val.assigned()); + + // the descriptor is adjusted only once the first value has been read + const auto descriptor = fb.getSignals()[0].getDescriptor(); + EXPECT_EQ(descriptor.getSampleType(), SampleType::Int64); + + // a DateTime node gets openDAQ's time descriptor + EXPECT_EQ(descriptor.getUnit().getSymbol(), "s"); + EXPECT_EQ(descriptor.getTickResolution(), Ratio(1, 1'000'000)); + EXPECT_EQ(descriptor.getOrigin(), "1970-01-01T00:00:00Z"); + + // OPC UA 100 ns ticks since 1601-01-01 are rebased to us since the UNIX epoch + const int64_t expectedUnixUs = (expected - UA_DATETIME_UNIX_EPOCH) / UA_DATETIME_USEC; + EXPECT_EQ(val.asPtr().getValue(int64_t(0)), expectedUnixUs); + + device.removeFunctionBlock(fb); + fb = nullptr; + } +} + TEST_F(GenericOpcuaMonitoredItemTest, UnsupportedDataTypeNode) { StartUp(); @@ -927,6 +1001,42 @@ TEST_F(GenericOpcuaMonitoredItemTest, ZeroSamplingInterval) ASSERT_EQ(fb.getStatusContainer().getStatus("ComponentStatus"), errStatus()); } +TEST_F(GenericOpcuaMonitoredItemTest, NegativeSamplingInterval) +{ + StartUp(); + + auto config = device.getAvailableFunctionBlockTypes().get(GENERIC_OPCUA_MONITORED_ITEM_FB_NAME).createDefaultConfig(); + config.setPropertyValue(PROPERTY_NAME_OPCUA_NODE_ID_STRING, std::string(".i32")); + config.setPropertyValue(PROPERTY_NAME_OPCUA_NAMESPACE_INDEX, 1); + config.setPropertyValue(PROPERTY_NAME_OPCUA_SAMPLING_INTERVAL, -5); + + CreateMonitoredItemFB(config); + + ASSERT_EQ(fb.getStatusContainer().getStatus("ComponentStatus"), errStatus()); + + // The negative value must not wrap into a huge unsigned interval: removing the FB has to + // detach it from the scheduler promptly instead of waiting out a weeks-long deadline. + const auto t0 = std::chrono::steady_clock::now(); + ASSERT_NO_THROW(device.removeFunctionBlock(fb)); + fb = nullptr; + const auto elapsedMs = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); + EXPECT_LT(elapsedMs, 2 * DEFAULT_OPCUA_MIFB_SAMPLING_INTERVAL); +} + +TEST_F(GenericOpcuaMonitoredItemTest, TooLargeSamplingInterval) +{ + StartUp(); + + auto config = device.getAvailableFunctionBlockTypes().get(GENERIC_OPCUA_MONITORED_ITEM_FB_NAME).createDefaultConfig(); + config.setPropertyValue(PROPERTY_NAME_OPCUA_NODE_ID_STRING, std::string(".i32")); + config.setPropertyValue(PROPERTY_NAME_OPCUA_NAMESPACE_INDEX, 1); + config.setPropertyValue(PROPERTY_NAME_OPCUA_SAMPLING_INTERVAL, static_cast(std::numeric_limits::max()) + 1); + + CreateMonitoredItemFB(config); + + ASSERT_EQ(fb.getStatusContainer().getStatus("ComponentStatus"), errStatus()); +} + TEST_F(GenericOpcuaMonitoredItemTest, PropertyVisibilityTogglesWithNodeIdType) { using NT = NodeIDType; @@ -1011,3 +1121,69 @@ TEST_F(GenericOpcuaMonitoredItemTest, ReconfigureNodeIdTypeNumericToString) ASSERT_EQ(fb.getStatusContainer().getStatus("ComponentStatus"), okStatus()); } + + +namespace +{ +// Number of samples delivered on a signal, counted without draining the reader. +daq::StreamReaderPtr makeCountingReader(const daq::SignalPtr& signal) +{ + return daq::StreamReaderBuilder() + .setSignal(signal) + .setValueReadType(daq::SampleType::Int64) + .setDomainReadType(daq::SampleType::UInt64) + .setSkipEvents(true) + .build(); +} +} + +TEST_F(GenericOpcuaMonitoredItemTest, RemoveFunctionBlockWhileSampling) +{ + StartUp(); + + // A short interval keeps the scheduler busy on this item, so removal is likely to land while a + // sample is in progress. + for (int i = 0; i < 10; ++i) + { + daq::FunctionBlockPtr localFb; + auto config = device.getAvailableFunctionBlockTypes().get(GENERIC_OPCUA_MONITORED_ITEM_FB_NAME).createDefaultConfig(); + config.setPropertyValue(PROPERTY_NAME_OPCUA_NODE_ID_STRING, std::string(".i32")); + config.setPropertyValue(PROPERTY_NAME_OPCUA_NAMESPACE_INDEX, 1); + config.setPropertyValue(PROPERTY_NAME_OPCUA_SAMPLING_INTERVAL, 1); + + ASSERT_NO_THROW(localFb = device.addFunctionBlock(GENERIC_OPCUA_MONITORED_ITEM_FB_NAME, config)); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + ASSERT_NO_THROW(device.removeFunctionBlock(localFb)); + } + + EXPECT_EQ(device.getFunctionBlocks().getCount(), 0u); +} + +TEST_F(GenericOpcuaMonitoredItemTest, ChangedSamplingIntervalTakesEffect) +{ + constexpr uint32_t slowInterval = 500; + constexpr uint32_t fastInterval = 20; + constexpr daq::SizeT packets = 6; + StartUp(); + + CreateMonitoredItemFB(std::string(".i32"), 1, slowInterval); + ASSERT_EQ(fb.getStatusContainer().getStatus("ComponentStatus"), okStatus()); + + auto reader = makeCountingReader(fb.getSignals()[0]); + + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + const auto slowCount = reader.getAvailableCount(); + EXPECT_LE(slowCount, 3u); + + fb.setPropertyValue(PROPERTY_NAME_OPCUA_SAMPLING_INTERVAL, fastInterval); + + // Time the same packets at the new interval. At the old one they would need six times 500 ms, and + // the first of them still waits out the pending old deadline. Comparing the measured time against + // the old interval pits the runner against itself, so a slow machine cannot fail this spuriously. + const auto start = std::chrono::steady_clock::now(); + const bool arrived = waitForPackets(reader, slowCount + packets, std::chrono::seconds(20)); + const auto elapsedMs = std::chrono::duration_cast(std::chrono::steady_clock::now() - start).count(); + + ASSERT_TRUE(arrived); + EXPECT_LT(elapsedMs, packets * slowInterval); +} diff --git a/shared/libraries/opcuageneric/opcuageneric_client/tests/test_sampling_scheduler.cpp b/shared/libraries/opcuageneric/opcuageneric_client/tests/test_sampling_scheduler.cpp new file mode 100644 index 00000000..f920f85f --- /dev/null +++ b/shared/libraries/opcuageneric/opcuageneric_client/tests/test_sampling_scheduler.cpp @@ -0,0 +1,403 @@ +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace daq::opcua::generic; +using namespace std::chrono_literals; + +namespace +{ + using Clock = SamplingScheduler::Clock; + + class FakeItem : public ISampledItem + { + public: + explicit FakeItem(uint32_t intervalMs) + : interval(intervalMs) + { + } + + uint32_t getSamplingInterval() const override + { + return interval.load(); + } + + void setSamplingInterval(uint32_t intervalMs) + { + interval = intervalMs; + } + + void processSample() override + { + const auto delay = sampleDelay.load(); + { + std::scoped_lock lock(mutex); + sampleTimes.push_back(Clock::now()); + } + cv.notify_all(); + + if (delay.count() > 0) + { + sampleDelay = 0ms; + std::this_thread::sleep_for(delay); + slowSampleReturnedAt = Clock::now(); + } + } + + void onConnectionRestored() override + { + revalidations++; + cv.notify_all(); + } + + void onSchedulerDestroyed() override + { + schedulerDestroyedCalls++; + cv.notify_all(); + } + + size_t sampleCount() const + { + std::scoped_lock lock(mutex); + return sampleTimes.size(); + } + + std::vector takeSampleTimes() const + { + std::scoped_lock lock(mutex); + return sampleTimes; + } + + // Blocks until at least count samples were taken, or the timeout elapses. + bool waitForSamples(size_t count, std::chrono::milliseconds timeout) + { + std::unique_lock lock(mutex); + return cv.wait_for(lock, timeout, [&] { return sampleTimes.size() >= count; }); + } + + std::atomic sampleDelay{0ms}; + // Set when a delayed sample returned: the catch-up window after the stall starts there. + std::atomic slowSampleReturnedAt{Clock::time_point{}}; + std::atomic revalidations{0}; + std::atomic schedulerDestroyedCalls{0}; + + private: + std::atomic interval; + mutable std::mutex mutex; + std::condition_variable cv; + std::vector sampleTimes; + }; + + // Median gap between consecutive samples, ignoring the first `from` samples. A slow runner can + // only stretch gaps, and the median absorbs the odd stalled wakeup, so bounding it from below + // holds however loaded the machine is. + int64_t medianGapMs(const std::vector& times, size_t from = 0) + { + std::vector gaps; + for (size_t i = std::max(from, 1); i < times.size(); ++i) + gaps.push_back(std::chrono::duration_cast(times[i] - times[i - 1]).count()); + + if (gaps.empty()) + return std::numeric_limits::max(); + + std::sort(gaps.begin(), gaps.end()); + return gaps[gaps.size() / 2]; + } + + // Samples taken in [from, from + window). Bounding this from above is safe on any runner: being + // slow can only move samples out of the window, never into it. + size_t countWithin(const std::vector& times, Clock::time_point from, std::chrono::milliseconds window) + { + return static_cast( + std::count_if(times.begin(), times.end(), [&](Clock::time_point t) { return t >= from && t < from + window; })); + } + +} + +TEST(SamplingSchedulerTest, AdvanceDeadlineNotLate) +{ + const auto now = Clock::now(); + const auto due = now - 10ms; // due right now, no backlog + + EXPECT_EQ(SamplingScheduler::advanceDeadline(due, now, 100ms), due + 100ms); +} + +TEST(SamplingSchedulerTest, AdvanceDeadlineLateByLessThanOneInterval) +{ + const auto now = Clock::now(); + const auto due = now - 60ms; // one interval of debt: the shifted deadline is still in the future + + EXPECT_EQ(SamplingScheduler::advanceDeadline(due, now, 100ms), due + 100ms); +} + +TEST(SamplingSchedulerTest, AdvanceDeadlineDropsBacklog) +{ + const auto now = Clock::now(); + const auto interval = 100ms; + + // Whatever the depth of the backlog, the deadline is re-anchored to exactly one interval ahead. + for (const int intervalsLate : {2, 3, 10, 3000}) + { + const auto due = now - interval * intervalsLate; + EXPECT_EQ(SamplingScheduler::advanceDeadline(due, now, interval), now + interval) << "late by " << intervalsLate << " intervals"; + } +} + +TEST(SamplingSchedulerTest, AdvanceDeadlinePostcondition) +{ + const auto now = Clock::now(); + const auto interval = 100ms; + + for (const int intervalsLate : {0, 1, 2, 10, 3000}) + { + const auto due = now - interval * intervalsLate; + const auto next = SamplingScheduler::advanceDeadline(due, now, interval); + + EXPECT_GT(next, now) << "late by " << intervalsLate << " intervals"; // there is always a wait + EXPECT_LE(next, now + interval) << "late by " << intervalsLate << " intervals"; // and it never exceeds one interval + } +} + +TEST(SamplingSchedulerTest, AdvanceDeadlineGuardsAgainstZeroInterval) +{ + const auto now = Clock::now(); + EXPECT_GT(SamplingScheduler::advanceDeadline(now, now, 0ms), now); +} + +TEST(SamplingSchedulerTest, SamplesItemUntilStopped) +{ + FakeItem item(20); + SamplingScheduler scheduler(nullptr); + scheduler.start(); + scheduler.registerItem(&item); + + ASSERT_TRUE(item.waitForSamples(3, 1s)); + + scheduler.stop(); + const auto afterStop = item.sampleCount(); + std::this_thread::sleep_for(100ms); + EXPECT_EQ(item.sampleCount(), afterStop); +} + +TEST(SamplingSchedulerTest, IndependentIntervalsAreHonoured) +{ + FakeItem fast(20); + FakeItem medium(50); + FakeItem slow(500); + + SamplingScheduler scheduler(nullptr); + scheduler.start(); + scheduler.registerItem(&fast); + scheduler.registerItem(&medium); + scheduler.registerItem(&slow); + + // The slow item sets how long the run has to be; waiting for it beats a fixed window, which would + // assume the runner keeps up with the fast one. + ASSERT_TRUE(slow.waitForSamples(4, 10s)); + scheduler.stop(); + + // Every item stays on its own deadline: none of them is sampled faster than its own interval. If + // the items shared one deadline, the slow ones would run at the fast item's rate. + EXPECT_GE(medianGapMs(fast.takeSampleTimes()), 15); + EXPECT_GE(medianGapMs(medium.takeSampleTimes()), 40); + EXPECT_GE(medianGapMs(slow.takeSampleTimes()), 400); + + // And the fast item really does get more turns, whatever pace the runner manages overall. + EXPECT_GT(fast.sampleCount(), slow.sampleCount() + 2); +} + +TEST(SamplingSchedulerTest, SingleSlowSampleCatchesUpByOneTickOnly) +{ + FakeItem item(50); + SamplingScheduler scheduler(nullptr); + scheduler.start(); + scheduler.registerItem(&item); + + ASSERT_TRUE(item.waitForSamples(1, 1s)); + item.sampleDelay = 200ms; // stalls for four intervals, consumed by the next sample + + ASSERT_TRUE(item.waitForSamples(4, 10s)); + scheduler.stop(); + + // The stall costs at most one catch-up read. Without the backlog drop the four missed reads would + // all fire the moment the slow sample returns, i.e. within microseconds of that catch-up read, so + // measuring the window from the read itself makes the bound independent of the runner's speed. + const auto times = item.takeSampleTimes(); + const auto caughtUp = + std::find_if(times.begin(), times.end(), [&](Clock::time_point t) { return t >= item.slowSampleReturnedAt.load(); }); + ASSERT_NE(caughtUp, times.end()); + EXPECT_EQ(countWithin(times, *caughtUp, 25ms), 1u); +} + +TEST(SamplingSchedulerTest, ChangedIntervalTakesEffect) +{ + constexpr auto slowInterval = 500ms; + + FakeItem item(static_cast(slowInterval.count())); + SamplingScheduler scheduler(nullptr); + scheduler.start(); + scheduler.registerItem(&item); + + ASSERT_TRUE(item.waitForSamples(1, 1s)); + const auto beforeChange = item.sampleCount(); + item.setSamplingInterval(20); + + // The timeout only has to be generous; it is the gaps that carry the assertion. + ASSERT_TRUE(item.waitForSamples(beforeChange + 6, 10s)); + scheduler.stop(); + + // The new interval applies from the next deadline shift, so the first gap still carries the + // pending old deadline and is skipped. From there the gaps must sit well below the old interval - + // a scheduler that ignored the change would still be a full 500 ms apart. This compares the runner + // with itself instead of assuming it achieves the requested 20 ms. + EXPECT_LT(medianGapMs(item.takeSampleTimes(), beforeChange + 1), slowInterval.count() / 2); +} + +TEST(SamplingSchedulerTest, UnregisterWaitsForSampleInProgress) +{ + FakeItem item(20); + SamplingScheduler scheduler(nullptr); + scheduler.start(); + scheduler.registerItem(&item); + + ASSERT_TRUE(item.waitForSamples(1, 1s)); + item.sampleDelay = 300ms; + ASSERT_TRUE(item.waitForSamples(2, 1s)); // the sample that sleeps has just started + + const auto start = Clock::now(); + scheduler.unregisterItem(&item); + const auto elapsed = Clock::now() - start; + + // unregisterItem must not return while the item is still being sampled. + EXPECT_GE(elapsed, 100ms); + + const auto afterUnregister = item.sampleCount(); + std::this_thread::sleep_for(100ms); + EXPECT_EQ(item.sampleCount(), afterUnregister); + + scheduler.stop(); +} + +TEST(SamplingSchedulerTest, DoesNotSampleWhileDisconnected) +{ + std::atomic connected{false}; + FakeItem item(20); + + SamplingScheduler scheduler([&] { return connected.load(); }); + scheduler.start(); + scheduler.registerItem(&item); + + std::this_thread::sleep_for(200ms); + EXPECT_EQ(item.sampleCount(), 0u); + + connected = true; + scheduler.onReconnected(); // also wakes the loop out of its disconnected wait + + // How long the three samples take is up to the runner; that they arrive at all is the assertion. + EXPECT_TRUE(item.waitForSamples(3, 5s)); + scheduler.stop(); +} + +TEST(SamplingSchedulerTest, ResumingAfterPauseDoesNotBurst) +{ + std::atomic connected{true}; + FakeItem item(20); + + SamplingScheduler scheduler([&] { return connected.load(); }); + scheduler.start(); + scheduler.registerItem(&item); + + ASSERT_TRUE(item.waitForSamples(2, 1s)); + connected = false; + std::this_thread::sleep_for(600ms); // ~30 missed ticks + + const auto beforeResume = item.sampleCount(); + const auto resumedAt = Clock::now(); + connected = true; + scheduler.onReconnected(); + + ASSERT_TRUE(item.waitForSamples(beforeResume + 3, 5s)); + scheduler.stop(); + + // The backlog is dropped, so the resumed item samples at its normal rate instead of firing the ~30 + // missed reads at once. Two intervals' worth of window holds three samples at the normal rate and + // all thirty of a burst; counting inside it cannot overcount on a slow runner. + const auto times = item.takeSampleTimes(); + EXPECT_LE(countWithin(times, resumedAt, 40ms), 3u); +} + +TEST(SamplingSchedulerTest, OnReconnectedRevalidatesEveryItem) +{ + std::atomic connected{true}; + FakeItem first(50); + FakeItem second(50); + + SamplingScheduler scheduler([&] { return connected.load(); }); + scheduler.start(); + scheduler.registerItem(&first); + scheduler.registerItem(&second); + + ASSERT_TRUE(first.waitForSamples(1, 1s)); + scheduler.onReconnected(); + + for (int i = 0; i < 200 && (first.revalidations == 0 || second.revalidations == 0); ++i) + std::this_thread::sleep_for(10ms); + + EXPECT_EQ(first.revalidations.load(), 1u); + EXPECT_EQ(second.revalidations.load(), 1u); + + scheduler.stop(); +} + +TEST(SamplingSchedulerTest, StopIsIdempotentAndSafeWithoutStart) +{ + SamplingScheduler scheduler(nullptr); + EXPECT_NO_THROW(scheduler.stop()); + + scheduler.start(); + EXPECT_NO_THROW(scheduler.stop()); + EXPECT_NO_THROW(scheduler.stop()); +} + +TEST(SamplingSchedulerTest, DestructorDetachesItemsThatAreStillRegistered) +{ + // Items outlive the scheduler whenever the owning device is destroyed without removed() running + // first: the device destroys its scheduler member, and only afterwards does the base class release + // the function blocks. Without this callback each of those blocks would unregister from a destroyed + // scheduler, which on macOS surfaces as "mutex lock failed: Invalid argument" and terminates. + FakeItem item(20); + + { + SamplingScheduler scheduler(nullptr); + scheduler.start(); + scheduler.registerItem(&item); + ASSERT_TRUE(item.waitForSamples(1, 1s)); + EXPECT_EQ(item.schedulerDestroyedCalls.load(), 0u); + } + + EXPECT_EQ(item.schedulerDestroyedCalls.load(), 1u); +} + +TEST(SamplingSchedulerTest, DestructorDoesNotDetachItemsThatUnregisteredThemselves) +{ + FakeItem item(20); + + { + SamplingScheduler scheduler(nullptr); + scheduler.start(); + scheduler.registerItem(&item); + ASSERT_TRUE(item.waitForSamples(1, 1s)); + scheduler.unregisterItem(&item); + } + + EXPECT_EQ(item.schedulerDestroyedCalls.load(), 0u); +}