StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move fields to atomic (#573)

Browse Source
This commit is contained in:
KirillPamPam
2024-09-25 15:16:43 +04:00
committed by GitHub
parent 89a11b85ae
commit 2b141f7d08
2 changed files with 124 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHead.kt
View File

@@ -37,6 +37,7 @@ import reactor.core.publisher.Sinks
import reactor.core.scheduler.Scheduler import reactor.core.scheduler.Scheduler
import reactor.kotlin.core.publisher.switchIfEmpty import reactor.kotlin.core.publisher.switchIfEmpty
import java.time.Duration import java.time.Duration
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.atomic.AtomicReference
@@ -64,34 +65,34 @@ class GenericWsHead(
} }
private val chainIdValidator = chainSpecific.chainSettingsValidator(upstream.getChain(), upstream, jsonRpcWsClient) private val chainIdValidator = chainSpecific.chainSettingsValidator(upstream.getChain(), upstream, jsonRpcWsClient)
private var connectionId: String? = null private val connectionId = AtomicReference<String?>(null)
private var subscribed = false private val subscribed = AtomicBoolean(false)
private var connected = false private val connected = AtomicBoolean(false)
private var isSyncing = false private val isSyncing = AtomicBoolean(false)
private var subscription: Disposable? = null private val subscription = AtomicReference<Disposable?>()
private var headResubSubscription: Disposable? = null private val headResubSubscription = AtomicReference<Disposable?>()
private val noHeadUpdatesSink = Sinks.many().multicast().directBestEffort<Boolean>() private val noHeadUpdatesSink = Sinks.many().multicast().directBestEffort<Boolean>()
private var subscriptionId = AtomicReference("") private val subscriptionId = AtomicReference("")
override fun isRunning(): Boolean { override fun isRunning(): Boolean {
return subscription != null return subscription.get() != null
} }
override fun start() { override fun start() {
super.start() super.start()
this.subscription?.dispose() this.subscription.get()?.dispose()
this.subscribed = true this.subscribed.set(true)
val heads = Flux.merge( val heads = Flux.merge(
// get the current block, not just wait for the next update // get the current block, not just wait for the next update
getLatestBlock(api), getLatestBlock(api),
listenNewHeads(), listenNewHeads(),
) )
this.subscription = super.follow(heads) this.subscription.set(super.follow(heads))
if (headResubSubscription == null) { if (headResubSubscription.get() == null) {
headResubSubscription = registerHeadResubscribeFlux() headResubSubscription.set(registerHeadResubscribeFlux())
} }
} }
@@ -100,10 +101,10 @@ class GenericWsHead(
} }
override fun onSyncingNode(isSyncing: Boolean) { override fun onSyncingNode(isSyncing: Boolean) {
if (isSyncing && !this.isSyncing) { if (isSyncing && !this.isSyncing.get()) {
cancelSub() cancelSub()
} }
this.isSyncing = isSyncing this.isSyncing.set(isSyncing)
} }
private fun listenNewHeads(): Flux<BlockContainer> { private fun listenNewHeads(): Flux<BlockContainer> {
@@ -129,7 +130,7 @@ class GenericWsHead(
} }
UPSTREAM_SETTINGS_ERROR -> { UPSTREAM_SETTINGS_ERROR -> {
log.warn("Couldn't check chain settings via ws connection for {}, ws sub will be recreated", upstreamId) log.warn("Couldn't check chain settings via ws connection for {}, ws sub will be recreated", upstreamId)
subscribed = false subscribed.set(false)
Mono.empty() Mono.empty()
} }
UPSTREAM_FATAL_SETTINGS_ERROR -> { UPSTREAM_FATAL_SETTINGS_ERROR -> {
@@ -144,8 +145,7 @@ class GenericWsHead(
override fun stop() { override fun stop() {
super.stop() super.stop()
cancelSub() cancelSub()
headResubSubscription?.dispose() headResubSubscription.getAndSet(null)?.dispose()
headResubSubscription = null
} }
override fun chainIdValidator(): SingleValidator<ValidateUpstreamSettingsResult>? { override fun chainIdValidator(): SingleValidator<ValidateUpstreamSettingsResult>? {
@@ -153,7 +153,7 @@ class GenericWsHead(
} }
private fun unsubscribe(): Mono<BlockContainer> { private fun unsubscribe(): Mono<BlockContainer> {
subscribed = false subscribed.set(false)
return wsSubscriptions.unsubscribe(chainSpecific.unsubscribeNewHeadsRequest(subscriptionId.get()).copy(id = ids.getAndIncrement())) return wsSubscriptions.unsubscribe(chainSpecific.unsubscribeNewHeadsRequest(subscriptionId.get()).copy(id = ids.getAndIncrement()))
.flatMap { it.requireResult() } .flatMap { it.requireResult() }
.doOnNext { log.warn("{} has just unsubscribed from newHeads", upstreamId) } .doOnNext { log.warn("{} has just unsubscribed from newHeads", upstreamId) }
@@ -170,10 +170,10 @@ class GenericWsHead(
return try { return try {
wsSubscriptions.subscribe(chainSpecific.listenNewHeadsRequest().copy(id = ids.getAndIncrement())) wsSubscriptions.subscribe(chainSpecific.listenNewHeadsRequest().copy(id = ids.getAndIncrement()))
.also { .also {
connectionId = it.connectionId connectionId.set(it.connectionId)
subscriptionId = it.subId subscriptionId.set(it.subId.get())
if (!connected) { if (!connected.get()) {
connected = true connected.set(true)
} }
}.data }.data
} catch (e: Exception) { } catch (e: Exception) {
@@ -184,13 +184,13 @@ class GenericWsHead(
private fun registerHeadResubscribeFlux(): Disposable { private fun registerHeadResubscribeFlux(): Disposable {
val connectionStates = wsSubscriptions.connectionInfoFlux() val connectionStates = wsSubscriptions.connectionInfoFlux()
.map { .map {
if (it.connectionId == connectionId && it.connectionState == WsConnection.ConnectionState.DISCONNECTED) { if (it.connectionId == connectionId.get() && it.connectionState == WsConnection.ConnectionState.DISCONNECTED) {
headLivenessSink.emitNext(HeadLivenessState.DISCONNECTED) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } headLivenessSink.emitNext(HeadLivenessState.DISCONNECTED) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
subscribed = false subscribed.set(false)
connected = false connected.set(false)
connectionId = null connectionId.set(null)
} else if (it.connectionState == WsConnection.ConnectionState.CONNECTED) { } else if (it.connectionState == WsConnection.ConnectionState.CONNECTED) {
connected = true connected.set(true)
return@map true return@map true
} }
return@map false return@map false
@@ -200,7 +200,7 @@ class GenericWsHead(
noHeadUpdatesSink.asFlux(), noHeadUpdatesSink.asFlux(),
connectionStates, connectionStates,
).publishOn(wsConnectionResubscribeScheduler) ).publishOn(wsConnectionResubscribeScheduler)
.filter { it && !subscribed && connected && !isSyncing } .filter { it && !subscribed.get() && connected.get() && !isSyncing.get() }
.subscribe { .subscribe {
log.warn("Restart ws head, upstreamId: $upstreamId") log.warn("Restart ws head, upstreamId: $upstreamId")
start() start()
@@ -208,8 +208,7 @@ class GenericWsHead(
} }
private fun cancelSub() { private fun cancelSub() {
subscription?.dispose() subscription.getAndSet(null)?.dispose()
subscription = null subscribed.set(false)
subscribed = false
} }
} }

179
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt
View File

@@ -99,13 +99,14 @@ open class GenericUpstream(
} }
private val validator: UpstreamValidator? = validatorBuilder(chain, this, getOptions(), chainConfig, versionRules) private val validator: UpstreamValidator? = validatorBuilder(chain, this, getOptions(), chainConfig, versionRules)
private var validatorSubscription: Disposable? = null private val validatorSubscription = AtomicReference<Disposable?>()
private var validationSettingsSubscription: Disposable? = null private val validationSettingsSubscription = AtomicReference<Disposable?>()
private var lowerBlockDetectorSubscription: Disposable? = null private val lowerBlockDetectorSubscription = AtomicReference<Disposable?>()
private val settingsDetectorSubscription = AtomicReference<Disposable?>()
private val hasLiveSubscriptionHead: AtomicBoolean = AtomicBoolean(false) private val hasLiveSubscriptionHead: AtomicBoolean = AtomicBoolean(false)
protected val connector: GenericConnector = connectorFactory.create(this, chain) protected val connector: GenericConnector = connectorFactory.create(this, chain)
private var livenessSubscription: Disposable? = null private val livenessSubscription = AtomicReference<Disposable?>()
private val settingsDetector = upstreamSettingsDetectorBuilder(chain, this) private val settingsDetector = upstreamSettingsDetectorBuilder(chain, this)
private var rpcMethodsDetector: UpstreamRpcMethodsDetector? = null private var rpcMethodsDetector: UpstreamRpcMethodsDetector? = null
@@ -116,7 +117,7 @@ open class GenericUpstream(
private val clientVersion = AtomicReference(UNKNOWN_CLIENT_VERSION) private val clientVersion = AtomicReference(UNKNOWN_CLIENT_VERSION)
private val finalizationDetector = finalizationDetectorBuilder() private val finalizationDetector = finalizationDetectorBuilder()
private var finalizationDetectorSubscription: Disposable? = null private val finalizationDetectorSubscription = AtomicReference<Disposable?>()
private val headLivenessState = Sinks.many().multicast().directBestEffort<ValidateUpstreamSettingsResult>() private val headLivenessState = Sinks.many().multicast().directBestEffort<ValidateUpstreamSettingsResult>()
@@ -204,63 +205,67 @@ open class GenericUpstream(
private fun validateUpstreamSettings() { private fun validateUpstreamSettings() {
if (validator != null) { if (validator != null) {
validationSettingsSubscription = Flux.merge( validationSettingsSubscription.set(
Flux.interval( Flux.merge(
Duration.ofSeconds(20), Flux.interval(
).flatMap { Duration.ofSeconds(20),
validator.validateUpstreamSettings() ).flatMap {
}, validator.validateUpstreamSettings()
headLivenessState.asFlux(), },
headLivenessState.asFlux(),
)
.subscribeOn(upstreamSettingsScheduler)
.distinctUntilChanged()
.subscribe {
when (it) {
UPSTREAM_FATAL_SETTINGS_ERROR -> {
if (isUpstreamValid.get()) {
log.warn("There is a fatal error after upstream settings validation, removing ${getId()}...")
partialStop()
sendUpstreamStateEvent(UpstreamChangeEvent.ChangeType.FATAL_SETTINGS_ERROR_REMOVED)
}
isUpstreamValid.set(false)
}
UPSTREAM_VALID -> {
if (!isUpstreamValid.get()) {
log.warn("Upstream ${getId()} is now valid, adding to the multistream...")
upstreamStart()
sendUpstreamStateEvent(UpstreamChangeEvent.ChangeType.ADDED)
}
isUpstreamValid.set(true)
}
else -> {
log.warn("Continue validation of upstream ${getId()}")
}
}
},
) )
.subscribeOn(upstreamSettingsScheduler)
.distinctUntilChanged()
.subscribe {
when (it) {
UPSTREAM_FATAL_SETTINGS_ERROR -> {
if (isUpstreamValid.get()) {
log.warn("There is a fatal error after upstream settings validation, removing ${getId()}...")
partialStop()
sendUpstreamStateEvent(UpstreamChangeEvent.ChangeType.FATAL_SETTINGS_ERROR_REMOVED)
}
isUpstreamValid.set(false)
}
UPSTREAM_VALID -> {
if (!isUpstreamValid.get()) {
log.warn("Upstream ${getId()} is now valid, adding to the multistream...")
upstreamStart()
sendUpstreamStateEvent(UpstreamChangeEvent.ChangeType.ADDED)
}
isUpstreamValid.set(true)
}
else -> {
log.warn("Continue validation of upstream ${getId()}")
}
}
}
} }
} }
private fun detectSettings() { private fun detectSettings() {
Flux.interval( settingsDetectorSubscription.set(
Duration.ZERO, Flux.interval(
Duration.ofSeconds(getOptions().validationInterval.toLong() * 5), Duration.ZERO,
).flatMap { Duration.ofSeconds(getOptions().validationInterval.toLong() * 5),
Flux.merge( ).flatMap {
settingsDetector?.detectLabels() Flux.merge(
?.doOnNext { label -> settingsDetector?.detectLabels()
updateLabels(label) ?.doOnNext { label ->
sendUpstreamStateEvent(UPDATED) updateLabels(label)
}, sendUpstreamStateEvent(UPDATED)
settingsDetector?.detectClientVersion() },
?.doOnNext { settingsDetector?.detectClientVersion()
log.info("Detected node version $it for upstream ${getId()}") ?.doOnNext {
clientVersion.set(it) log.info("Detected node version $it for upstream ${getId()}")
}, clientVersion.set(it)
) },
.subscribeOn(settingsScheduler) )
}.subscribe() .subscribeOn(settingsScheduler)
}.subscribe(),
)
} }
private fun detectRpcMethods( private fun detectRpcMethods(
@@ -311,22 +316,26 @@ open class GenericUpstream(
this.setStatus(UpstreamAvailability.OK) this.setStatus(UpstreamAvailability.OK)
} else { } else {
log.debug("Start validation for upstream ${this.getId()}") log.debug("Start validation for upstream ${this.getId()}")
validatorSubscription = validator?.start() validatorSubscription.set(
?.subscribe(this::setStatus) validator?.start()
?.subscribe(this::setStatus),
)
} }
livenessSubscription = connector.headLivenessEvents().subscribe( livenessSubscription.set(
{ connector.headLivenessEvents().subscribe(
val hasSub = it == HeadLivenessState.OK {
hasLiveSubscriptionHead.set(hasSub) val hasSub = it == HeadLivenessState.OK
if (it == HeadLivenessState.FATAL_ERROR) { hasLiveSubscriptionHead.set(hasSub)
headLivenessState.emitNext(UPSTREAM_FATAL_SETTINGS_ERROR) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } if (it == HeadLivenessState.FATAL_ERROR) {
} else { headLivenessState.emitNext(UPSTREAM_FATAL_SETTINGS_ERROR) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
sendUpstreamStateEvent(UPDATED) } else {
} sendUpstreamStateEvent(UPDATED)
}, }
{ },
log.debug("Error while checking live subscription for ${getId()}", it) {
}, log.debug("Error while checking live subscription for ${getId()}", it)
},
),
) )
detectSettings() detectSettings()
@@ -337,21 +346,17 @@ open class GenericUpstream(
override fun stop() { override fun stop() {
partialStop() partialStop()
validationSettingsSubscription?.dispose() validationSettingsSubscription.getAndSet(null)?.dispose()
validationSettingsSubscription = null
connector.stop() connector.stop()
started.set(false) started.set(false)
} }
private fun partialStop() { private fun partialStop() {
validatorSubscription?.dispose() validatorSubscription.getAndSet(null)?.dispose()
validatorSubscription = null livenessSubscription.getAndSet(null)?.dispose()
livenessSubscription?.dispose() lowerBlockDetectorSubscription.getAndSet(null)?.dispose()
livenessSubscription = null finalizationDetectorSubscription.getAndSet(null)?.dispose()
lowerBlockDetectorSubscription?.dispose() settingsDetectorSubscription.getAndSet(null)?.dispose()
lowerBlockDetectorSubscription = null
finalizationDetectorSubscription?.dispose()
finalizationDetectorSubscription = null
connector.getHead().stop() connector.getHead().stop()
} }
@@ -373,21 +378,23 @@ open class GenericUpstream(
} }
private fun detectFinalization() { private fun detectFinalization() {
finalizationDetectorSubscription = finalizationDetectorSubscription.set(
finalizationDetector.detectFinalization(this, chainConfig.expectedBlockTime, getChain()) finalizationDetector.detectFinalization(this, chainConfig.expectedBlockTime, getChain())
.subscribeOn(finalizationScheduler) .subscribeOn(finalizationScheduler)
.subscribe { .subscribe {
sendUpstreamStateEvent(UPDATED) sendUpstreamStateEvent(UPDATED)
} },
)
} }
private fun detectLowerBlock() { private fun detectLowerBlock() {
lowerBlockDetectorSubscription = lowerBlockDetectorSubscription.set(
lowerBoundService.detectLowerBounds() lowerBoundService.detectLowerBounds()
.subscribeOn(lowerScheduler) .subscribeOn(lowerScheduler)
.subscribe { .subscribe {
sendUpstreamStateEvent(UPDATED) sendUpstreamStateEvent(UPDATED)
} },
)
} }
fun getIngressSubscription(): IngressSubscription { fun getIngressSubscription(): IngressSubscription {