Fix reconfiguration of fast modbus events

debian/changelog

@@ -1,3 +1,9 @@
+wb-mqtt-serial (2.147.1) stable; urgency=medium
+
+  * Reconfigure fast modbus events after device restart
+
+ -- Petr Krasnoshchekov <[email protected]>  Tue, 12 Nov 2024 21:25:35 +0500
+
 wb-mqtt-serial (2.147.0) stable; urgency=medium
 
   * Store press_counter register values as soon as possible.

src/poll_plan.h

@@ -12,6 +12,15 @@ template<typename TItem> class TPriorityQueue: public std::priority_queue<TItem>
     {
         return std::find_if(this->c.cbegin(), this->c.cend(), pred) != this->c.cend();
     }
+
+    template<typename Pred> void Remove(Pred pred)
+    {
+        auto it = std::find_if(this->c.begin(), this->c.end(), pred);
+        if (it != this->c.end()) {
+            this->c.erase(it);
+            std::make_heap(this->c.begin(), this->c.end(), this->comp);
+        }
+    }
 };
 
 template<class TEntry, typename ComparePredicate> class TPriorityQueueSchedule
@@ -72,6 +81,11 @@ template<class TEntry, typename ComparePredicate> class TPriorityQueueSchedule
         return Entries.Contains([&](const TItem& item) { return item.Data == entry; });
     }
 
+    void Remove(TEntry entry)
+    {
+        Entries.Remove([&](const TItem& item) { return item.Data == entry; });
+    }
+
 private:
     TPriorityQueue<TItem> Entries;
 };
@@ -293,6 +307,12 @@ template<class TEntry, class TComparePredicate = std::less<TEntry>> class TSched
         return LowPriorityQueue.Contains(entry) || HighPriorityQueue.Contains(entry);
     }
 
+    void Remove(TEntry entry)
+    {
+        LowPriorityQueue.Remove(entry);
+        HighPriorityQueue.Remove(entry);
+    }
+
     bool IsEmpty() const
     {
         return LowPriorityQueue.IsEmpty() && HighPriorityQueue.IsEmpty();

src/serial_client_register_poller.cpp

@@ -137,6 +137,8 @@ void TSerialClientRegisterPoller::ClosedPortCycle(steady_clock::time_point curre
 {
     Scheduler.ResetLoadBalancing();
 
+    RescheduleDisconnectedDevices();
+
     TClosedPortDeviceReader reader(currentTime);
     do {
         reader.ClearRegisters();
@@ -180,6 +182,8 @@ TPollResult TSerialClientRegisterPoller::OpenPortCycle(TPort& port,
                                                        TSerialClientDeviceAccessHandler& lastAccessedDevice,
                                                        TRegisterCallback callback)
 {
+    RescheduleDisconnectedDevices();
+
     TPollResult res;
 
     TDeviceReader reader(spentTime, maxPollingTime, readAtLeastOneRegister, lastAccessedDevice);
@@ -237,10 +241,7 @@ TPollResult TSerialClientRegisterPoller::OpenPortCycle(TPort& port,
 void TSerialClientRegisterPoller::OnDeviceConnectionStateChanged(PSerialDevice device)
 {
     if (device->GetConnectionState() == TDeviceConnectionState::DISCONNECTED) {
-        auto range = Devices.equal_range(device);
-        for (auto it = range.first; it != range.second; ++it) {
-            it->second->RescheduleAllRegisters();
-        }
+        DisconnectedDevicesWaitingForReschedule.push_back(device);
     }
 }
 
@@ -263,3 +264,16 @@ std::string TThrottlingStateLogger::GetMessage()
     }
     return std::string();
 }
+
+void TSerialClientRegisterPoller::RescheduleDisconnectedDevices()
+{
+    for (auto& device: DisconnectedDevicesWaitingForReschedule) {
+        auto range = Devices.equal_range(device);
+        for (auto it = range.first; it != range.second; ++it) {
+            Scheduler.Remove(it->second);
+            it->second->RescheduleAllRegisters();
+            Scheduler.AddEntry(it->second, it->second->GetDeadline(), it->second->GetPriority());
+        }
+    }
+    DisconnectedDevicesWaitingForReschedule.clear();
+}

src/serial_client_register_poller.h

@@ -53,6 +53,7 @@ class TSerialClientRegisterPoller
     std::chrono::steady_clock::time_point GetDeadline(bool lowPriorityRateLimitIsExceeded,
                                                       const util::TSpentTimeMeter& spentTime) const;
     void OnDeviceConnectionStateChanged(PSerialDevice device);
+    void RescheduleDisconnectedDevices();
 
     std::multimap<PSerialDevice, PPollableDevice> Devices;
 
@@ -61,4 +62,6 @@ class TSerialClientRegisterPoller
     TThrottlingStateLogger ThrottlingStateLogger;
 
     TRateLimiter LowPriorityRateLimiter;
+
+    std::vector<PSerialDevice> DisconnectedDevicesWaitingForReschedule;
 };