-
Notifications
You must be signed in to change notification settings - Fork 75
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Change BlockFetcher to typed actor Split fetcher message handling among child actors Abstract the fetch trait Scalafmt Refactor blockFetcherState
- Loading branch information
1 parent
c38c2c2
commit 9876ed2
Showing
12 changed files
with
590 additions
and
748 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
568 changes: 214 additions & 354 deletions
568
src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcher.scala
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
72 changes: 72 additions & 0 deletions
72
src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BodiesFetcher.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| package io.iohk.ethereum.blockchain.sync.regular | ||
|
|
||
| import akka.actor.typed.{ActorRef, Behavior} | ||
| import akka.actor.typed.scaladsl.{AbstractBehavior, ActorContext, Behaviors} | ||
| import akka.actor.{ActorRef => ClassicActorRef} | ||
| import akka.util.ByteString | ||
| import io.iohk.ethereum.blockchain.sync.PeersClient.{BestPeer, Request} | ||
| import io.iohk.ethereum.blockchain.sync.regular.BlockFetcher.FetchCommand | ||
| import io.iohk.ethereum.blockchain.sync.regular.BodiesFetcher.BodiesFetcherCommand | ||
| import io.iohk.ethereum.network.Peer | ||
| import io.iohk.ethereum.network.p2p.Message | ||
| import io.iohk.ethereum.network.p2p.messages.PV62.{BlockBodies, GetBlockBodies} | ||
| import io.iohk.ethereum.utils.Config.SyncConfig | ||
| import monix.execution.Scheduler | ||
|
|
||
| import scala.util.{Failure, Success} | ||
|
|
||
| class BodiesFetcher( | ||
| val peersClient: ClassicActorRef, | ||
| val syncConfig: SyncConfig, | ||
| val supervisor: ActorRef[FetchCommand], | ||
| context: ActorContext[BodiesFetcher.BodiesFetcherCommand] | ||
| ) extends AbstractBehavior[BodiesFetcher.BodiesFetcherCommand](context) | ||
| with FetchRequest[BodiesFetcherCommand] { | ||
|
|
||
| val log = context.log | ||
| implicit val ec: Scheduler = Scheduler(context.executionContext) | ||
|
|
||
| import BodiesFetcher._ | ||
|
|
||
| override def makeAdaptedMessage[T <: Message](peer: Peer, msg: T): BodiesFetcherCommand = AdaptedMessage(peer, msg) | ||
|
|
||
| override def onMessage(message: BodiesFetcherCommand): Behavior[BodiesFetcherCommand] = { | ||
| message match { | ||
| case FetchBodies(hashes) => | ||
| log.debug("Start fetching bodies") | ||
| requestBodies(hashes) | ||
| Behaviors.same | ||
| case AdaptedMessage(peer, BlockBodies(bodies)) => | ||
| log.debug(s"Received ${bodies.size} block bodies") | ||
| supervisor ! BlockFetcher.ReceivedBodies(peer, bodies) | ||
| Behaviors.same | ||
| case BodiesFetcher.RetryBodiesRequest => | ||
| supervisor ! BlockFetcher.RetryBodiesRequest | ||
| Behaviors.same | ||
| case _ => Behaviors.unhandled | ||
| } | ||
| } | ||
|
|
||
| private def requestBodies(hashes: Seq[ByteString]): Unit = { | ||
| val resp = makeRequest(Request.create(GetBlockBodies(hashes), BestPeer), BodiesFetcher.RetryBodiesRequest) | ||
| context.pipeToSelf(resp.runToFuture) { | ||
| case Success(res) => res | ||
| case Failure(_) => BodiesFetcher.RetryBodiesRequest | ||
| } | ||
| } | ||
| } | ||
|
|
||
| object BodiesFetcher { | ||
|
|
||
| def apply( | ||
| peersClient: ClassicActorRef, | ||
| syncConfig: SyncConfig, | ||
| supervisor: ActorRef[FetchCommand] | ||
| ): Behavior[BodiesFetcherCommand] = | ||
| Behaviors.setup(context => new BodiesFetcher(peersClient, syncConfig, supervisor, context)) | ||
|
|
||
| sealed trait BodiesFetcherCommand | ||
| final case class FetchBodies(hashes: Seq[ByteString]) extends BodiesFetcherCommand | ||
| final case object RetryBodiesRequest extends BodiesFetcherCommand | ||
| private final case class AdaptedMessage[T <: Message](peer: Peer, msg: T) extends BodiesFetcherCommand | ||
| } |
56 changes: 56 additions & 0 deletions
56
src/main/scala/io/iohk/ethereum/blockchain/sync/regular/FetchRequest.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| package io.iohk.ethereum.blockchain.sync.regular | ||
|
|
||
| import akka.actor.ActorRef | ||
| import io.iohk.ethereum.blockchain.sync.PeersClient | ||
| import io.iohk.ethereum.blockchain.sync.PeersClient.{BlacklistPeer, NoSuitablePeer, Request, RequestFailed} | ||
| import io.iohk.ethereum.network.Peer | ||
| import io.iohk.ethereum.network.p2p.Message | ||
| import io.iohk.ethereum.utils.Config.SyncConfig | ||
| import monix.eval.Task | ||
| import org.slf4j.Logger | ||
| import akka.pattern.ask | ||
| import akka.util.Timeout | ||
| import io.iohk.ethereum.utils.FunctorOps._ | ||
|
|
||
| import scala.concurrent.duration._ | ||
| import scala.util.Failure | ||
|
|
||
| trait FetchRequest[A] { | ||
| val peersClient: ActorRef | ||
| val syncConfig: SyncConfig | ||
| val log: Logger | ||
|
|
||
| def makeAdaptedMessage[T <: Message](peer: Peer, msg: T): A | ||
|
|
||
| implicit val timeout: Timeout = syncConfig.peerResponseTimeout + 2.second // some margin for actor communication | ||
|
|
||
| def makeRequest(request: Request[_], responseFallback: A): Task[A] = | ||
| Task | ||
| .deferFuture(peersClient ? request) | ||
| .tap(blacklistPeerOnFailedRequest) | ||
| .flatMap(handleRequestResult(responseFallback)) | ||
| .onErrorHandle { error => | ||
| log.error("Unexpected error while doing a request", error) | ||
| responseFallback | ||
| } | ||
|
|
||
| def blacklistPeerOnFailedRequest(msg: Any): Unit = msg match { | ||
| case RequestFailed(peer, reason) => peersClient ! BlacklistPeer(peer.id, reason) | ||
| case _ => () | ||
| } | ||
|
|
||
| def handleRequestResult(fallback: A)(msg: Any): Task[A] = { | ||
| msg match { | ||
| case failed: RequestFailed => | ||
| log.debug("Request failed due to {}", failed) | ||
| Task.now(fallback) | ||
| case NoSuitablePeer => | ||
| Task.now(fallback).delayExecution(syncConfig.syncRetryInterval) | ||
| case Failure(cause) => | ||
| log.error("Unexpected error on the request result", cause) | ||
| Task.now(fallback) | ||
| case PeersClient.Response(peer, msg) => | ||
| Task.now(makeAdaptedMessage(peer, msg)) | ||
| } | ||
| } | ||
| } |
81 changes: 81 additions & 0 deletions
81
src/main/scala/io/iohk/ethereum/blockchain/sync/regular/HeadersFetcher.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| package io.iohk.ethereum.blockchain.sync.regular | ||
| import akka.actor.typed.{ActorRef, Behavior} | ||
| import akka.actor.typed.scaladsl.{AbstractBehavior, ActorContext, Behaviors} | ||
| import akka.actor.{ActorRef => ClassicActorRef} | ||
| import io.iohk.ethereum.blockchain.sync.PeersClient.{BestPeer, Request} | ||
| import io.iohk.ethereum.blockchain.sync.regular.BlockFetcher.FetchCommand | ||
| import io.iohk.ethereum.blockchain.sync.regular.HeadersFetcher.HeadersFetcherCommand | ||
| import io.iohk.ethereum.network.Peer | ||
| import io.iohk.ethereum.network.p2p.Message | ||
| import io.iohk.ethereum.network.p2p.messages.PV62.{BlockHeaders, GetBlockHeaders} | ||
| import io.iohk.ethereum.utils.Config.SyncConfig | ||
| import monix.eval.Task | ||
| import monix.execution.Scheduler | ||
| import org.slf4j.Logger | ||
|
|
||
| import scala.util.{Failure, Success} | ||
|
|
||
| class HeadersFetcher( | ||
| val peersClient: ClassicActorRef, | ||
| val syncConfig: SyncConfig, | ||
| val supervisor: ActorRef[FetchCommand], | ||
| context: ActorContext[HeadersFetcher.HeadersFetcherCommand] | ||
| ) extends AbstractBehavior[HeadersFetcher.HeadersFetcherCommand](context) | ||
| with FetchRequest[HeadersFetcherCommand] { | ||
|
|
||
| override val log: Logger = context.log | ||
| implicit val ec: Scheduler = Scheduler(context.executionContext) | ||
|
|
||
| import HeadersFetcher._ | ||
|
|
||
| override def makeAdaptedMessage[T <: Message](peer: Peer, msg: T): HeadersFetcherCommand = AdaptedMessage(peer, msg) | ||
|
|
||
| override def onMessage(message: HeadersFetcherCommand): Behavior[HeadersFetcherCommand] = | ||
| message match { | ||
| case FetchHeaders(blockNumber: BigInt, amount: BigInt) => | ||
| log.debug("Start fetching headers from block {}", blockNumber) | ||
| requestHeaders(blockNumber, amount) | ||
| Behaviors.same | ||
| case AdaptedMessage(_, BlockHeaders(headers)) => | ||
| log.debug("Fetched {} headers starting from block {}", headers.size, headers.headOption.map(_.number)) | ||
| supervisor ! BlockFetcher.ReceivedHeaders(headers) | ||
| Behaviors.same | ||
| case HeadersFetcher.RetryHeadersRequest => | ||
| supervisor ! BlockFetcher.RetryHeadersRequest | ||
| Behaviors.same | ||
| case _ => Behaviors.unhandled | ||
| } | ||
|
|
||
| private def requestHeaders(blockNr: BigInt, amount: BigInt): Unit = { | ||
| log.debug("Fetching headers from block {}", blockNr) | ||
| val msg = GetBlockHeaders(Left(blockNr), amount, skip = 0, reverse = false) | ||
|
|
||
| val resp = makeRequest(Request.create(msg, BestPeer), HeadersFetcher.RetryHeadersRequest) | ||
| .flatMap { | ||
| case AdaptedMessage(_, BlockHeaders(headers)) if headers.isEmpty => | ||
| log.debug("Empty BlockHeaders response. Retry in {}", syncConfig.syncRetryInterval) | ||
| Task.now(HeadersFetcher.RetryHeadersRequest).delayResult(syncConfig.syncRetryInterval) | ||
| case res => Task.now(res) | ||
| } | ||
|
|
||
| context.pipeToSelf(resp.runToFuture) { | ||
| case Success(res) => res | ||
| case Failure(_) => HeadersFetcher.RetryHeadersRequest | ||
| } | ||
| } | ||
| } | ||
|
|
||
| object HeadersFetcher { | ||
|
|
||
| def apply( | ||
| peersClient: ClassicActorRef, | ||
| syncConfig: SyncConfig, | ||
| supervisor: ActorRef[FetchCommand] | ||
| ): Behavior[HeadersFetcherCommand] = | ||
| Behaviors.setup(context => new HeadersFetcher(peersClient, syncConfig, supervisor, context)) | ||
|
|
||
| sealed trait HeadersFetcherCommand | ||
| final case class FetchHeaders(blockNumber: BigInt, amount: BigInt) extends HeadersFetcherCommand | ||
| final case object RetryHeadersRequest extends HeadersFetcherCommand | ||
| private final case class AdaptedMessage[T <: Message](peer: Peer, msg: T) extends HeadersFetcherCommand | ||
| } |
Oops, something went wrong.