From b0b3819a54aaa35b0af63e7c42803daff198f8cf Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Tue, 13 Dec 2022 18:43:04 +0300 Subject: [PATCH 1/2] fix concurrent modification --- .../dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index f85d561b..e7506326 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory import org.springframework.util.ConcurrentReferenceHashMap import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import java.util.Collections @Suppress("UNCHECKED_CAST") open class EthereumPosMultiStream( @@ -126,7 +127,7 @@ open class EthereumPosMultiStream( val newHead = MergedHead(heads, PriorityForkChoice(), "ETH Pos Multistream").apply { this.start() } - val lagObserver = EthereumPosHeadLagObserver(newHead, upstreams as Collection) + val lagObserver = EthereumPosHeadLagObserver(newHead, Collections.unmodifiableCollection(upstreams)) this.lagObserver = lagObserver lagObserver.start() newHead From fe987db1c9e837d0d2d35b64cede8b291bcbc4a4 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Tue, 13 Dec 2022 19:04:44 +0300 Subject: [PATCH 2/2] fix concurrent modification 2 --- .../upstream/ethereum_pos/EthereumPosMultiStream.kt | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index e7506326..abf12c63 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -21,8 +21,14 @@ import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.* +import io.emeraldpay.dshackle.upstream.ChainFees +import io.emeraldpay.dshackle.upstream.EmptyHead +import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Lifecycle +import io.emeraldpay.dshackle.upstream.MergedHead +import io.emeraldpay.dshackle.upstream.Multistream +import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -31,7 +37,6 @@ import org.slf4j.LoggerFactory import org.springframework.util.ConcurrentReferenceHashMap import reactor.core.publisher.Flux import reactor.core.publisher.Mono -import java.util.Collections @Suppress("UNCHECKED_CAST") open class EthereumPosMultiStream( @@ -127,7 +132,7 @@ open class EthereumPosMultiStream( val newHead = MergedHead(heads, PriorityForkChoice(), "ETH Pos Multistream").apply { this.start() } - val lagObserver = EthereumPosHeadLagObserver(newHead, Collections.unmodifiableCollection(upstreams)) + val lagObserver = EthereumPosHeadLagObserver(newHead, ArrayList(upstreams)) this.lagObserver = lagObserver lagObserver.start() newHead