-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Create MappedAllocationStrategy, improve switchover support in alloca…
…tion strategies, add support for ShardRegion.StartEntity
- Loading branch information
Showing
8 changed files
with
292 additions
and
38 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
5 changes: 5 additions & 0 deletions
5
src/main/scala/com/evolutiongaming/cluster/ActorRefExtension.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,5 @@ | ||
| package com.evolutiongaming.cluster | ||
|
|
||
| import akka.actor.{ActorRef, Extension} | ||
|
|
||
| class ActorRefExtension(val ref: ActorRef) extends Extension |
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
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
98 changes: 98 additions & 0 deletions
98
src/main/scala/com/evolutiongaming/cluster/MappedAllocationStrategy.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,98 @@ | ||
| package com.evolutiongaming.cluster | ||
|
|
||
| import akka.actor.{ActorRef, ActorSystem} | ||
| import akka.cluster.sharding.ShardCoordinator.ShardAllocationStrategy | ||
| import akka.cluster.sharding.ShardRegion | ||
| import com.typesafe.scalalogging.LazyLogging | ||
|
|
||
| import scala.collection.immutable | ||
| import scala.concurrent.Future | ||
|
|
||
| class MappedAllocationStrategy( | ||
| typeName: String, | ||
| fallbackStrategy: ShardAllocationStrategy, | ||
| proxy: ActorRef, | ||
| val maxSimultaneousRebalance: Int) | ||
| extends ShardAllocationStrategy with LazyLogging { | ||
|
|
||
| import MappedAllocationStrategy._ | ||
|
|
||
| def mapShardToRegion(shardId: ShardRegion.ShardId, regionRef: ActorRef) = | ||
| proxy ! UpdateMapping(typeName, shardId, regionRef) | ||
|
|
||
| def allocateShard( | ||
| requester: ActorRef, | ||
| shardId: ShardRegion.ShardId, | ||
| currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardRegion.ShardId]]): Future[ActorRef] = { | ||
|
|
||
| shardToRegionMapping get EntityKey(typeName, shardId) match { | ||
| case Some(toNode) => | ||
| logger debug s"AllocateShard $typeName\n\t" + | ||
| s"shardId:\t$shardId\n\t" + | ||
| s"on node:\t$toNode\n\t" + | ||
| s"requester:\t$requester\n\t" | ||
| Future successful toNode | ||
| case None => | ||
| logger debug s"AllocateShard fallback $typeName, shardId:\t$shardId" | ||
| fallbackStrategy.allocateShard(requester, shardId, currentShardAllocations) | ||
| } | ||
| } | ||
|
|
||
|
|
||
| def rebalance( | ||
| currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardRegion.ShardId]], | ||
| rebalanceInProgress: Set[ShardRegion.ShardId]): Future[Set[ShardRegion.ShardId]] = { | ||
|
|
||
| logger debug | ||
| s"rebalance $typeName: currentShardAllocations = $currentShardAllocations, rebalanceInProgress = $rebalanceInProgress" | ||
|
|
||
| val result = (for { | ||
| (ref, shards) <- currentShardAllocations | ||
| shardId <- shards if !(shardToRegionMapping get EntityKey(typeName, shardId) contains ref) | ||
| } yield shardId).toSet | ||
|
|
||
| if (result.nonEmpty) logger info s"Rebalance $typeName\n\t" + | ||
| s"current:${ currentShardAllocations.mkString("\n\t\t", "\n\t\t", "") }\n\t" + | ||
| s"rebalanceInProgress:\t$rebalanceInProgress\n\t" + | ||
| s"result:\t$result" | ||
|
|
||
| Future successful result | ||
| } | ||
| } | ||
|
|
||
| object MappedAllocationStrategy { | ||
|
|
||
| def apply( | ||
| typeName: String, | ||
| fallbackStrategy: ShardAllocationStrategy, | ||
| maxSimultaneousRebalance: Int) | ||
| (implicit system: ActorSystem): MappedAllocationStrategy = { | ||
| // proxy doesn't depend on typeName, it should just start once | ||
| val proxy = MappedAllocationStrategyDistributedDataProxy(system).ref | ||
| new MappedAllocationStrategy( | ||
| typeName = typeName, | ||
| fallbackStrategy = fallbackStrategy, | ||
| proxy = proxy, | ||
| maxSimultaneousRebalance = maxSimultaneousRebalance) | ||
| } | ||
|
|
||
| case class EntityKey(typeName: String, id: ShardRegion.ShardId) { | ||
| override def toString: String = s"$typeName#$id" | ||
| } | ||
|
|
||
| object EntityKey { | ||
| def unapply(arg: List[String]): Option[EntityKey] = arg match { | ||
| case typeName :: id :: Nil => Some(EntityKey(typeName, id)) | ||
| case _ => None | ||
| } | ||
|
|
||
| def unapply(arg: String): Option[EntityKey] = unapply((arg split "#").toList) | ||
| } | ||
|
|
||
| case class UpdateMapping(typeName: String, id: ShardRegion.ShardId, regionRef: ActorRef) | ||
| case class Clear(typeName: String, id: ShardRegion.ShardId) | ||
|
|
||
| // TODO: check for thread-safety | ||
| @volatile | ||
| private[cluster] var shardToRegionMapping: Map[EntityKey, ActorRef] = Map.empty | ||
| } |
70 changes: 70 additions & 0 deletions
70
...main/scala/com/evolutiongaming/cluster/MappedAllocationStrategyDistributedDataProxy.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,70 @@ | ||
| /* | ||
| * Copyright 2016-2017 Evolution Gaming Limited | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package com.evolutiongaming.cluster | ||
|
|
||
| import akka.actor.{Actor, ActorLogging, ActorRef, ExtendedActorSystem, ExtensionId, Props} | ||
| import akka.cluster.Cluster | ||
| import akka.cluster.ddata.Replicator._ | ||
| import akka.cluster.ddata._ | ||
|
|
||
| class MappedAllocationStrategyDistributedDataProxy extends Actor with ActorLogging { | ||
|
|
||
| import MappedAllocationStrategy._ | ||
| import MappedAllocationStrategyDistributedDataProxy._ | ||
|
|
||
| implicit lazy val node = Cluster(context.system) | ||
| lazy val replicator: ActorRef = DistributedData(context.system).replicator | ||
| private val emptyMap = LWWMap.empty[String, ActorRef] | ||
|
|
||
| replicator ! Subscribe(MappingKey, self) | ||
|
|
||
|
|
||
| def receive: Receive = { | ||
| case UpdateMapping(typeName, id, regionRef) => | ||
| val entityKey = EntityKey(typeName, id) | ||
| shardToRegionMapping += entityKey -> regionRef | ||
| replicator ! Update(MappingKey, emptyMap, WriteLocal)(_ + (entityKey.toString -> regionRef)) | ||
|
|
||
| case Clear(typeName, id) => | ||
| val entityKey = EntityKey(typeName, id) | ||
| shardToRegionMapping -= entityKey | ||
| replicator ! Update(MappingKey, emptyMap, WriteLocal)(_ - entityKey.toString) | ||
|
|
||
| case UpdateSuccess(_, _) => | ||
|
|
||
| case UpdateTimeout(key, _) => | ||
| // probably should never happen for local updates | ||
| log warning s"Update timeout for key $key" | ||
|
|
||
| case c@Changed(MappingKey) => | ||
| val newData = (c get MappingKey).entries flatMap { | ||
| case (key, ref) => | ||
| EntityKey unapply key map { entityKey => | ||
| entityKey -> ref | ||
| } | ||
| } | ||
| shardToRegionMapping = newData | ||
| } | ||
| } | ||
|
|
||
| object MappedAllocationStrategyDistributedDataProxy extends ExtensionId[ActorRefExtension] { | ||
| override def createExtension(system: ExtendedActorSystem): ActorRefExtension = | ||
| new ActorRefExtension(system actorOf Props[MappedAllocationStrategyDistributedDataProxy]) | ||
|
|
||
| // DData key of ShardToRegionMapping map | ||
| private[cluster] val MappingKey: LWWMapKey[String, ActorRef] = | ||
| LWWMapKey[String, ActorRef]("ShardToRegionMapping") | ||
| } |
Oops, something went wrong.