[mqtt] Avoid parallel streams with common thread pool to avoid deadlocks #13621

ssalonen · 2022-10-30T11:25:56Z

To mitigate issue #13657 (common thread pool exhaustion when combining parallel streams with synchronization or locks)

All the parallel streams are anyways hitting resource contention / bottleneck since they all leads to MqttBrokerConnection subscribe or unsubscribe which is having synchronized block with subscribers of that broker connection. [*]

On the surface things looks like they are asynchronous (based on CompletableFuture an all) but only the lowest level operations seem to be asynchronous, and in mid-levels we have this synchronization and locking behavour currently.

I would argue typical users have only single MqttBrokerConnection, raising the question whether the parallel streams bring any true benefit really.

[*] It looks like we might be able to remove the whole synchronized block there, using ConcurrentHashMap.compute. But even then the synchronization would happen, just hidden from plain sight, inside java runtime. From official javadocs: Some attempted update operations on this map by other threads may be blocked while computation is in progress, so the computation should be short and simple, and must not attempt to update any other mappings of this Map.... Not worth the trouble, the code would be much more unreadable that way

Fixes #13657

Signed-off-by: Sami Salonen [email protected]

To mitigate issue https://github.com/openhab/openhab-core/issues/3125 (common thread pool exhaustion when combining parallel streams with synchronization or locks) Signed-off-by: Sami Salonen <[email protected]>

Signed-off-by: Sami Salonen <[email protected]>

ssalonen · 2022-10-30T11:32:43Z

...ng.mqtt.generic/src/main/java/org/openhab/binding/mqtt/generic/AbstractMQTTThingHandler.java

-        return availabilityStates.values().parallelStream().map(cChannel -> cChannel.start(connection, scheduler, 0))
-                .collect(FutureCollector.allOf());
+        @NonNull
+        Collector<@NonNull CompletableFuture<@Nullable Void>, @NonNull Set<@NonNull CompletableFuture<@Nullable Void>>, @NonNull CompletableFuture<@Nullable Void>> allOfCollector = FutureCollector
+                .allOf();
+        return availabilityStates.values().stream().map(cChannel -> {
+            final @NonNull CompletableFuture<@Nullable Void> fut;
+            fut = cChannel == null ? CompletableFuture.completedFuture(null) : cChannel.start(connection, scheduler, 0);
+            return fut;
+        }).collect(allOfCollector);


cChannel.start() i.e. ChannelState.start calls MqttBrokerConnection.subscribe which is internally synchronizing on all subscribers of the connection.

Making this non-parallel stream should reduce resource contention anyways.

Other changes are to remove null typing warnings

ssalonen · 2022-10-30T11:33:34Z

...istant/src/main/java/org/openhab/binding/mqtt/homeassistant/internal/DiscoverComponents.java

-        this.topics.parallelStream().map(t -> connection.subscribe(t, this)).collect(FutureCollector.allOf())
+        this.topics.stream().map(t -> connection.subscribe(t, this)).collect(allOfCollector)


MqttBrokerConnection.subscribe is internally synchronizing on all subscribers of the connection.

Making this non-parallel stream should reduce resource contention anyways.

ssalonen · 2022-10-30T11:33:48Z

...istant/src/main/java/org/openhab/binding/mqtt/homeassistant/internal/DiscoverComponents.java

-                this.topics.parallelStream().forEach(t -> connection.unsubscribe(t, this));
+                this.topics.stream().forEach(t -> connection.unsubscribe(t, this));


MqttBrokerConnection.unsubscribe is internally synchronizing on all subscribers of the connection.

Making this non-parallel stream should reduce resource contention anyways.

ssalonen · 2022-10-30T11:33:54Z

...istant/src/main/java/org/openhab/binding/mqtt/homeassistant/internal/DiscoverComponents.java

-            this.topics.parallelStream().forEach(t -> connection.unsubscribe(t, this));
+            this.topics.stream().forEach(t -> connection.unsubscribe(t, this));


MqttBrokerConnection.unsubscribe is internally synchronizing on all subscribers of the connection.

Making this non-parallel stream should reduce resource contention anyways.

ssalonen · 2022-10-30T11:34:15Z

...c/main/java/org/openhab/binding/mqtt/homeassistant/internal/component/AbstractComponent.java

-        return channels.values().parallelStream().map(cChannel -> cChannel.start(connection, scheduler, timeout))
-                .collect(FutureCollector.allOf());
+        @NonNull
+        Collector<@NonNull CompletableFuture<@Nullable Void>, @NonNull Set<@NonNull CompletableFuture<@Nullable Void>>, @NonNull CompletableFuture<@Nullable Void>> allOfCollector = FutureCollector
+                .allOf();
+        return channels.values().stream().map(cChannel -> cChannel.start(connection, scheduler, timeout))
+                .collect(allOfCollector);


cChannel.start() i.e. ComponentChannel.start calls ChannelState.start which calls MqttBrokerConnection.subscribe which is internally synchronizing on all subscribers of the connection.

Making this non-parallel stream should reduce resource contention anyways.

Other changes are to remove null typing warnings

ssalonen · 2022-10-30T11:35:31Z

...c/main/java/org/openhab/binding/mqtt/homeassistant/internal/component/AbstractComponent.java

-        return channels.values().parallelStream().map(ComponentChannel::stop).collect(FutureCollector.allOf());
+        @NonNull
+        Collector<@NonNull CompletableFuture<@Nullable Void>, @NonNull Set<@NonNull CompletableFuture<@Nullable Void>>, @NonNull CompletableFuture<@Nullable Void>> allOfCollector = FutureCollector
+                .allOf();
+        return channels.values().stream().map(ComponentChannel::stop).collect(allOfCollector);


ComponentChannel.stop i.e. ComponentChannel.start calls ChannelState.stop which calls MqttBrokerConnection.unsubscribe which is internally synchronizing on all subscribers of the connection.

Making this non-parallel stream should reduce resource contention anyways.

Other changes are to remove null typing warnings

ssalonen · 2022-10-30T11:36:39Z

.../java/org/openhab/binding/mqtt/homeassistant/internal/handler/HomeAssistantThingHandler.java

-                haComponents.values().parallelStream().map(e -> e.start(connection, scheduler, attributeReceiveTimeout))
+                haComponents.values().stream().map(e -> e.start(connection, scheduler, attributeReceiveTimeout))


Eventually the streams synchronizes in MqttBrokerConnection.subscribe (see above comment with AbstractComponent leading to resource contention.

Making this non-parallel should reduce resource contention.

ssalonen · 2022-10-30T11:37:37Z

.../java/org/openhab/binding/mqtt/homeassistant/internal/handler/HomeAssistantThingHandler.java

-            haComponents.values().parallelStream().map(AbstractComponent::stop) //
+            haComponents.values().stream().map(AbstractComponent::stop) //


Eventually the streams synchronizes in MqttBrokerConnection.unsubscribe (see above comment with AbstractComponent leading to resource contention.

Making this non-parallel should reduce resource contention.

.../java/org/openhab/binding/mqtt/homeassistant/internal/handler/HomeAssistantThingHandler.java

...c/main/java/org/openhab/binding/mqtt/homeassistant/internal/component/AbstractComponent.java

lolodomo · 2022-11-21T18:15:37Z

This looks like an important fix.
@wborn @cweitkamp @kaikreuzer @fwolter @jlaur : maybe one of you is more confortable with these changes than me ?

ssalonen · 2022-12-03T06:40:46Z

Any chance getting this to official release?

I have been running this version (as prebuilt here) for several weeks without adverse effects

antroids · 2022-12-05T16:43:52Z

I agree that calling and getting results from asynchronous methods in parallel streams is not a best idea and could be replaced by sequential version.

Signed-off-by: Sami Salonen <[email protected]>

wborn

Thanks! Let's deal with the annotations another time. 🙂

…cks (openhab#13621) To mitigate issue https://github.com/openhab/openhab-core/issues/3125 (common thread pool exhaustion when combining parallel streams with synchronization or locks) Signed-off-by: Sami Salonen <[email protected]> Signed-off-by: Ben Rosenblum <[email protected]>

…cks (openhab#13621) To mitigate issue https://github.com/openhab/openhab-core/issues/3125 (common thread pool exhaustion when combining parallel streams with synchronization or locks) Signed-off-by: Sami Salonen <[email protected]> Signed-off-by: Andras Uhrin <[email protected]>

…cks (openhab#13621) To mitigate issue https://github.com/openhab/openhab-core/issues/3125 (common thread pool exhaustion when combining parallel streams with synchronization or locks) Signed-off-by: Sami Salonen <[email protected]>

…cks (openhab#13621) To mitigate issue https://github.com/openhab/openhab-core/issues/3125 (common thread pool exhaustion when combining parallel streams with synchronization or locks) Signed-off-by: Sami Salonen <[email protected]> Signed-off-by: Andras Uhrin <[email protected]>

ssalonen requested review from davidgraeff and antroids as code owners October 30, 2022 11:25

[mqtt] avoid null annotation warnings

62236d9

Signed-off-by: Sami Salonen <[email protected]>

ssalonen commented Oct 30, 2022

View reviewed changes

ssalonen mentioned this pull request Oct 30, 2022

[mqtt] Frequent deadlock, likely caused by parallel streams #13657

Closed

ssalonen requested a review from J-N-K November 1, 2022 19:25

wborn added the bug An unexpected problem or unintended behavior of an add-on label Nov 5, 2022

[mqtt] Opt for cleaner code and disregard warning from Eclipse IDE

85375d1

Signed-off-by: Sami Salonen <[email protected]>

wborn approved these changes Dec 6, 2022

View reviewed changes

wborn merged commit 923c0f1 into openhab:main Dec 6, 2022

wborn added this to the 3.4 milestone Dec 6, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[mqtt] Avoid parallel streams with common thread pool to avoid deadlocks #13621

[mqtt] Avoid parallel streams with common thread pool to avoid deadlocks #13621

ssalonen commented Oct 30, 2022 •

edited by jlaur

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

lolodomo commented Nov 21, 2022

ssalonen commented Dec 3, 2022

antroids commented Dec 5, 2022

wborn left a comment

		this.topics.parallelStream().map(t -> connection.subscribe(t, this)).collect(FutureCollector.allOf())
		this.topics.stream().map(t -> connection.subscribe(t, this)).collect(allOfCollector)

		this.topics.parallelStream().forEach(t -> connection.unsubscribe(t, this));
		this.topics.stream().forEach(t -> connection.unsubscribe(t, this));

		haComponents.values().parallelStream().map(e -> e.start(connection, scheduler, attributeReceiveTimeout))
		haComponents.values().stream().map(e -> e.start(connection, scheduler, attributeReceiveTimeout))

		haComponents.values().parallelStream().map(AbstractComponent::stop) //
		haComponents.values().stream().map(AbstractComponent::stop) //

[mqtt] Avoid parallel streams with common thread pool to avoid deadlocks #13621

[mqtt] Avoid parallel streams with common thread pool to avoid deadlocks #13621

Conversation

ssalonen commented Oct 30, 2022 • edited by jlaur Loading

ssalonen Oct 30, 2022 • edited Loading

Choose a reason for hiding this comment

ssalonen Oct 30, 2022 • edited Loading

Choose a reason for hiding this comment

ssalonen Oct 30, 2022 • edited Loading

Choose a reason for hiding this comment

ssalonen Oct 30, 2022 • edited Loading

Choose a reason for hiding this comment

ssalonen Oct 30, 2022 • edited Loading

Choose a reason for hiding this comment

ssalonen Oct 30, 2022 • edited Loading

Choose a reason for hiding this comment

ssalonen Oct 30, 2022 • edited Loading

Choose a reason for hiding this comment

ssalonen Oct 30, 2022 • edited Loading

Choose a reason for hiding this comment

lolodomo commented Nov 21, 2022

ssalonen commented Dec 3, 2022

antroids commented Dec 5, 2022

wborn left a comment

Choose a reason for hiding this comment

ssalonen commented Oct 30, 2022 •

edited by jlaur

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading

ssalonen Oct 30, 2022 •

edited

Loading