raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Union and diff for sets are more performant

Browse source
This commit is contained in:
Johan Andrén 2020-06-11 15:52:46 +02:00
parent 9c7f16a4db
commit 33b34da36c
3 changed files with 4 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

2
akka-cluster-sharding/src/main/scala/akka/cluster/sharding/Shard.scala
View file

@ -198,7 +198,7 @@ private[akka] object Shard {
else newState else newState
} else { } else {
if (r.ackTo.isEmpty) this if (r.ackTo.isEmpty) this
else RememberingStart(ackTo ++ r.ackTo) else RememberingStart(ackTo.union(r.ackTo))
} }
case _ => throw new IllegalArgumentException(s"Transition from $this to $newState not allowed") case _ => throw new IllegalArgumentException(s"Transition from $this to $newState not allowed")
} }

2
akka-cluster-sharding/src/main/scala/akka/cluster/sharding/internal/DDataRememberEntitiesShardStore.scala
View file

@ -178,7 +178,7 @@ private[akka] final class DDataRememberEntitiesShardStore(
} }
private def onUpdate(update: RememberEntitiesShardStore.Update): Unit = { private def onUpdate(update: RememberEntitiesShardStore.Update): Unit = {
val allEvts: Set[Evt] = (update.started.map(Started) ++ update.stopped.map(Stopped)) val allEvts: Set[Evt] = (update.started.map(Started(_): Evt).union(update.stopped.map(Stopped)))
// map from set of evts (for same ddata key) to one update that applies each of them // map from set of evts (for same ddata key) to one update that applies each of them
val ddataUpdates: Map[Set[Evt], (Update[ORSet[EntityId]], Int)] = val ddataUpdates: Map[Set[Evt], (Update[ORSet[EntityId]], Int)] =
allEvts.groupBy(evt => key(evt.id)).map { allEvts.groupBy(evt => key(evt.id)).map {

4
akka-cluster-sharding/src/main/scala/akka/cluster/sharding/internal/EventSourcedRememberEntitiesShardStore.scala
View file

@ -80,7 +80,7 @@ private[akka] final class EventSourcedRememberEntitiesShardStore(
override def receiveRecover: Receive = { override def receiveRecover: Receive = {
case EntitiesStarted(ids) => state = state.copy(state.entities.union(ids)) case EntitiesStarted(ids) => state = state.copy(state.entities.union(ids))
case EntitiesStopped(ids) => state = state.copy(state.entities -- ids) case EntitiesStopped(ids) => state = state.copy(state.entities.diff(ids))
case SnapshotOffer(_, snapshot: State) => state = snapshot case SnapshotOffer(_, snapshot: State) => state = snapshot
case RecoveryCompleted => case RecoveryCompleted =>
log.debug("Recovery completed for shard [{}] with [{}] entities", shardId, state.entities.size) log.debug("Recovery completed for shard [{}] with [{}] entities", shardId, state.entities.size)
@ -97,7 +97,7 @@ private[akka] final class EventSourcedRememberEntitiesShardStore(
left -= 1 left -= 1
if (left == 0) { if (left == 0) {
sender() ! RememberEntitiesShardStore.UpdateDone(started, stopped) sender() ! RememberEntitiesShardStore.UpdateDone(started, stopped)
state.copy(state.entities.union(started) -- stopped) state.copy(state.entities.union(started).diff(stopped))
saveSnapshotWhenNeeded() saveSnapshotWhenNeeded()
} }
} }