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

=clu replace Set -- with diff and ++ with union

Browse source 
* better performance according to
  https://docs.google.com/presentation/d/1Qjryxoe-fYEM8ZPhM-98LKfbhnRcn5eAEMNlVVnixsA/pub
This commit is contained in:
Patrik Nordwall 2015-10-30 14:59:36 +01:00
parent 9380983d3c
commit c7c187f6b7
24 changed files with 46 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

2
akka-cluster-metrics/src/main/scala/akka/cluster/metrics/ClusterMetricsCollector.scala
View file

@ -211,7 +211,7 @@ private[metrics] class ClusterMetricsCollector extends Actor with ActorLogging {
* Updates the initial node ring for those nodes that are [[akka.cluster.MemberStatus]] `Up`. * Updates the initial node ring for those nodes that are [[akka.cluster.MemberStatus]] `Up`.
*/ */
def receiveState(state: CurrentClusterState): Unit = def receiveState(state: CurrentClusterState): Unit =
nodes = (state.members -- state.unreachable) collect { nodes = (state.members diff state.unreachable) collect {
case m if m.status == MemberStatus.Up || m.status == MemberStatus.WeaklyUp ⇒ m.address case m if m.status == MemberStatus.Up || m.status == MemberStatus.WeaklyUp ⇒ m.address
} }

4
akka-cluster-metrics/src/main/scala/akka/cluster/metrics/Metric.scala
View file

@ -282,7 +282,7 @@ final case class NodeMetrics(address: Address, timestamp: Long, metrics: Set[Met
if (timestamp >= that.timestamp) this // that is older if (timestamp >= that.timestamp) this // that is older
else { else {
// equality is based on the name of the Metric and Set doesn't replace existing element // equality is based on the name of the Metric and Set doesn't replace existing element
copy(metrics = that.metrics ++ metrics, timestamp = that.timestamp) copy(metrics = that.metrics union metrics, timestamp = that.timestamp)
} }
} }
@ -303,7 +303,7 @@ final case class NodeMetrics(address: Address, timestamp: Long, metrics: Set[Met
} }
// Append metrics missing from either latest or current. // Append metrics missing from either latest or current.
// Equality is based on the [[Metric.name]] and [[Set]] doesn't replace existing elements. // Equality is based on the [[Metric.name]] and [[Set]] doesn't replace existing elements.
val merged = updated ++ latestNode.metrics ++ currentNode.metrics val merged = updated union latestNode.metrics union currentNode.metrics
copy(metrics = merged, timestamp = latestNode.timestamp) copy(metrics = merged, timestamp = latestNode.timestamp)
} }

2
akka-cluster-metrics/src/main/scala/akka/cluster/metrics/MetricsCollector.scala
View file

@ -218,7 +218,7 @@ class SigarMetricsCollector(address: Address, decayFactor: Double, sigar: SigarP
override def metrics(): Set[Metric] = { override def metrics(): Set[Metric] = {
// Must obtain cpuPerc in one shot. See https://github.com/akka/akka/issues/16121 // Must obtain cpuPerc in one shot. See https://github.com/akka/akka/issues/16121
val cpuPerc = sigar.getCpuPerc val cpuPerc = sigar.getCpuPerc
super.metrics ++ Set(cpuCombined(cpuPerc), cpuStolen(cpuPerc)).flatten super.metrics union Set(cpuCombined(cpuPerc), cpuStolen(cpuPerc)).flatten
} }
/** /**

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

@ -661,7 +661,7 @@ abstract class ShardCoordinator(typeName: String, settings: ClusterShardingSetti
rebalanceInProgress += shard rebalanceInProgress += shard
log.debug("Rebalance shard [{}] from [{}]", shard, rebalanceFromRegion) log.debug("Rebalance shard [{}] from [{}]", shard, rebalanceFromRegion)
context.actorOf(rebalanceWorkerProps(shard, rebalanceFromRegion, handOffTimeout, context.actorOf(rebalanceWorkerProps(shard, rebalanceFromRegion, handOffTimeout,
state.regions.keySet ++ state.regionProxies) state.regions.keySet union state.regionProxies)
.withDispatcher(context.props.dispatcher)) .withDispatcher(context.props.dispatcher))
case None ⇒ case None ⇒
log.debug("Rebalance of non-existing shard [{}] is ignored", shard) log.debug("Rebalance of non-existing shard [{}] is ignored", shard)

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

@ -318,7 +318,7 @@ class ShardRegion(
} }
def receiveClusterState(state: CurrentClusterState): Unit = { def receiveClusterState(state: CurrentClusterState): Unit = {
changeMembers(immutable.SortedSet.empty(ageOrdering) ++ state.members.filter(m ⇒ changeMembers(immutable.SortedSet.empty(ageOrdering) union state.members.filter(m ⇒
m.status == MemberStatus.Up && matchingRole(m))) m.status == MemberStatus.Up && matchingRole(m)))
} }

6
akka-cluster-tools/src/main/scala/akka/cluster/client/ClusterClient.scala
View file

@ -310,7 +310,7 @@ final class ClusterClient(settings: ClusterClientSettings) extends Actor with Ac
def sendGetContacts(): Unit = { def sendGetContacts(): Unit = {
val sendTo = val sendTo =
if (contacts.isEmpty) initialContactsSel if (contacts.isEmpty) initialContactsSel
else if (contacts.size == 1) (initialContactsSel ++ contacts) else if (contacts.size == 1) (initialContactsSel union contacts)
else contacts else contacts
if (log.isDebugEnabled) if (log.isDebugEnabled)
log.debug(s"""Sending GetContacts to [${sendTo.mkString(",")}]""") log.debug(s"""Sending GetContacts to [${sendTo.mkString(",")}]""")
@ -639,7 +639,7 @@ final class ClusterReceptionist(pubSubMediator: ActorRef, settings: ClusterRecep
val slice = { val slice = {
val first = nodes.from(a).tail.take(numberOfContacts) val first = nodes.from(a).tail.take(numberOfContacts)
if (first.size == numberOfContacts) first if (first.size == numberOfContacts) first
else first ++ nodes.take(numberOfContacts - first.size) else first union nodes.take(numberOfContacts - first.size)
} }
val contacts = Contacts(slice.map(a ⇒ self.path.toStringWithAddress(a))(collection.breakOut)) val contacts = Contacts(slice.map(a ⇒ self.path.toStringWithAddress(a))(collection.breakOut))
if (log.isDebugEnabled) if (log.isDebugEnabled)
@ -648,7 +648,7 @@ final class ClusterReceptionist(pubSubMediator: ActorRef, settings: ClusterRecep
} }
case state: CurrentClusterState ⇒ case state: CurrentClusterState ⇒
nodes = nodes.empty ++ state.members.collect { case m if m.status != MemberStatus.Joining && matchingRole(m) ⇒ m.address } nodes = nodes.empty union state.members.collect { case m if m.status != MemberStatus.Joining && matchingRole(m) ⇒ m.address }
consistentHash = ConsistentHash(nodes, virtualNodesFactor) consistentHash = ConsistentHash(nodes, virtualNodesFactor)
case MemberUp(m) ⇒ case MemberUp(m) ⇒

2
akka-cluster-tools/src/main/scala/akka/cluster/singleton/ClusterSingletonManager.scala
View file

@ -251,7 +251,7 @@ object ClusterSingletonManager {
} }
def handleInitial(state: CurrentClusterState): Unit = { def handleInitial(state: CurrentClusterState): Unit = {
membersByAge = immutable.SortedSet.empty(ageOrdering) ++ state.members.filter(m ⇒ membersByAge = immutable.SortedSet.empty(ageOrdering) union state.members.filter(m ⇒
(m.status == MemberStatus.Up || m.status == MemberStatus.Leaving) && matchingRole(m)) (m.status == MemberStatus.Up || m.status == MemberStatus.Leaving) && matchingRole(m))
val safeToBeOldest = !state.members.exists { m ⇒ (m.status == MemberStatus.Down || m.status == MemberStatus.Exiting) } val safeToBeOldest = !state.members.exists { m ⇒ (m.status == MemberStatus.Down || m.status == MemberStatus.Exiting) }
val initial = InitialOldestState(membersByAge.headOption.map(_.address), safeToBeOldest) val initial = InitialOldestState(membersByAge.headOption.map(_.address), safeToBeOldest)

2
akka-cluster-tools/src/main/scala/akka/cluster/singleton/ClusterSingletonProxy.scala
View file

@ -168,7 +168,7 @@ final class ClusterSingletonProxy(singletonManagerPath: String, settings: Cluste
def handleInitial(state: CurrentClusterState): Unit = { def handleInitial(state: CurrentClusterState): Unit = {
trackChange { trackChange {
() ⇒ () ⇒
membersByAge = immutable.SortedSet.empty(ageOrdering) ++ state.members.collect { membersByAge = immutable.SortedSet.empty(ageOrdering) union state.members.collect {
case m if m.status == MemberStatus.Up && matchingRole(m) ⇒ m case m if m.status == MemberStatus.Up && matchingRole(m) ⇒ m
} }
} }

6
akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala
View file

@ -926,11 +926,11 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
// handle changes // handle changes
// replace changed members // replace changed members
val newMembers = changedMembers ++ localMembers -- removedUnreachable val newMembers = changedMembers union localMembers diff removedUnreachable
// removing REMOVED nodes from the `seen` table // removing REMOVED nodes from the `seen` table
val removed = removedUnreachable.map(_.uniqueAddress) val removed = removedUnreachable.map(_.uniqueAddress)
val newSeen = localSeen -- removed val newSeen = localSeen diff removed
// removing REMOVED nodes from the `reachability` table // removing REMOVED nodes from the `reachability` table
val newReachability = localOverview.reachability.remove(removed) val newReachability = localOverview.reachability.remove(removed)
val newOverview = localOverview copy (seen = newSeen, reachability = newReachability) val newOverview = localOverview copy (seen = newSeen, reachability = newReachability)
@ -988,7 +988,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
if (changedMembers.nonEmpty) { if (changedMembers.nonEmpty) {
// replace changed members // replace changed members
val newMembers = changedMembers ++ localMembers val newMembers = changedMembers union localMembers
val newGossip = localGossip.copy(members = newMembers) val newGossip = localGossip.copy(members = newMembers)
updateLatestGossip(newGossip) updateLatestGossip(newGossip)

6
akka-cluster/src/main/scala/akka/cluster/ClusterEvent.scala
View file

@ -272,7 +272,7 @@ object ClusterEvent {
private[cluster] def diffMemberEvents(oldGossip: Gossip, newGossip: Gossip): immutable.Seq[MemberEvent] = private[cluster] def diffMemberEvents(oldGossip: Gossip, newGossip: Gossip): immutable.Seq[MemberEvent] =
if (newGossip eq oldGossip) Nil if (newGossip eq oldGossip) Nil
else { else {
val newMembers = newGossip.members -- oldGossip.members val newMembers = newGossip.members diff oldGossip.members
val membersGroupedByAddress = List(newGossip.members, oldGossip.members).flatten.groupBy(_.uniqueAddress) val membersGroupedByAddress = List(newGossip.members, oldGossip.members).flatten.groupBy(_.uniqueAddress)
val changedMembers = membersGroupedByAddress collect { val changedMembers = membersGroupedByAddress collect {
case (_, newMember :: oldMember :: Nil) if newMember.status != oldMember.status || newMember.upNumber != oldMember.upNumber ⇒ case (_, newMember :: oldMember :: Nil) if newMember.status != oldMember.status || newMember.upNumber != oldMember.upNumber ⇒
@ -285,7 +285,7 @@ object ClusterEvent {
// no events for other transitions // no events for other transitions
} }
val removedMembers = oldGossip.members -- newGossip.members val removedMembers = oldGossip.members diff newGossip.members
val removedEvents = removedMembers.map(m ⇒ MemberRemoved(m.copy(status = Removed), m.status)) val removedEvents = removedMembers.map(m ⇒ MemberRemoved(m.copy(status = Removed), m.status))
(new VectorBuilder[MemberEvent]() ++= memberEvents ++= removedEvents).result() (new VectorBuilder[MemberEvent]() ++= memberEvents ++= removedEvents).result()
@ -305,7 +305,7 @@ object ClusterEvent {
*/ */
private[cluster] def diffRolesLeader(oldGossip: Gossip, newGossip: Gossip, selfUniqueAddress: UniqueAddress): Set[RoleLeaderChanged] = { private[cluster] def diffRolesLeader(oldGossip: Gossip, newGossip: Gossip, selfUniqueAddress: UniqueAddress): Set[RoleLeaderChanged] = {
for { for {
role ← (oldGossip.allRoles ++ newGossip.allRoles) role ← (oldGossip.allRoles union newGossip.allRoles)
newLeader = newGossip.roleLeader(role, selfUniqueAddress) newLeader = newGossip.roleLeader(role, selfUniqueAddress)
if newLeader != oldGossip.roleLeader(role, selfUniqueAddress) if newLeader != oldGossip.roleLeader(role, selfUniqueAddress)
} yield RoleLeaderChanged(role, newLeader.map(_.address)) } yield RoleLeaderChanged(role, newLeader.map(_.address))

6
akka-cluster/src/main/scala/akka/cluster/ClusterHeartbeat.scala
View file

@ -184,7 +184,7 @@ private[cluster] final case class ClusterHeartbeatSenderState(
oldReceiversNowUnreachable: Set[UniqueAddress], oldReceiversNowUnreachable: Set[UniqueAddress],
failureDetector: FailureDetectorRegistry[Address]) { failureDetector: FailureDetectorRegistry[Address]) {
val activeReceivers: Set[UniqueAddress] = ring.myReceivers ++ oldReceiversNowUnreachable val activeReceivers: Set[UniqueAddress] = ring.myReceivers union oldReceiversNowUnreachable
def selfAddress = ring.selfAddress def selfAddress = ring.selfAddress
@ -212,7 +212,7 @@ private[cluster] final case class ClusterHeartbeatSenderState(
private def membershipChange(newRing: HeartbeatNodeRing): ClusterHeartbeatSenderState = { private def membershipChange(newRing: HeartbeatNodeRing): ClusterHeartbeatSenderState = {
val oldReceivers = ring.myReceivers val oldReceivers = ring.myReceivers
val removedReceivers = oldReceivers -- newRing.myReceivers val removedReceivers = oldReceivers diff newRing.myReceivers
var adjustedOldReceiversNowUnreachable = oldReceiversNowUnreachable var adjustedOldReceiversNowUnreachable = oldReceiversNowUnreachable
removedReceivers foreach { a ⇒ removedReceivers foreach { a ⇒
if (failureDetector.isAvailable(a.address)) if (failureDetector.isAvailable(a.address))
@ -260,7 +260,7 @@ private[cluster] final case class HeartbeatNodeRing(
ha < hb || (ha == hb && Member.addressOrdering.compare(a.address, b.address) < 0) ha < hb || (ha == hb && Member.addressOrdering.compare(a.address, b.address) < 0)
} }
immutable.SortedSet() ++ nodes immutable.SortedSet() union nodes
} }
/** /**

4
akka-cluster/src/main/scala/akka/cluster/ClusterMetricsCollector.scala
View file

@ -404,7 +404,7 @@ final case class NodeMetrics(address: Address, timestamp: Long, metrics: Set[Met
if (timestamp >= that.timestamp) this // that is older if (timestamp >= that.timestamp) this // that is older
else { else {
// equality is based on the name of the Metric and Set doesn't replace existing element // equality is based on the name of the Metric and Set doesn't replace existing element
copy(metrics = that.metrics ++ metrics, timestamp = that.timestamp) copy(metrics = that.metrics union metrics, timestamp = that.timestamp)
} }
} }
@ -742,7 +742,7 @@ class SigarMetricsCollector(address: Address, decayFactor: Double, sigar: AnyRef
} }
override def metrics: Set[Metric] = { override def metrics: Set[Metric] = {
super.metrics.filterNot(_.name == SystemLoadAverage) ++ Set(systemLoadAverage, cpuCombined).flatten super.metrics.filterNot(_.name == SystemLoadAverage) union Set(systemLoadAverage, cpuCombined).flatten
} }
/** /**

2
akka-cluster/src/main/scala/akka/cluster/ClusterRemoteWatcher.scala
View file

@ -72,7 +72,7 @@ private[cluster] class ClusterRemoteWatcher(
case state: CurrentClusterState ⇒ case state: CurrentClusterState ⇒
clusterNodes = state.members.collect { case m if m.address != selfAddress ⇒ m.address } clusterNodes = state.members.collect { case m if m.address != selfAddress ⇒ m.address }
clusterNodes foreach takeOverResponsibility clusterNodes foreach takeOverResponsibility
unreachable --= clusterNodes unreachable = unreachable diff clusterNodes
case MemberUp(m) ⇒ memberUp(m) case MemberUp(m) ⇒ memberUp(m)
case MemberWeaklyUp(m) ⇒ memberUp(m) case MemberWeaklyUp(m) ⇒ memberUp(m)
case MemberRemoved(m, previousStatus) ⇒ memberRemoved(m, previousStatus) case MemberRemoved(m, previousStatus) ⇒ memberRemoved(m, previousStatus)

8
akka-cluster/src/main/scala/akka/cluster/Gossip.scala
View file

@ -73,12 +73,12 @@ private[cluster] final case class Gossip(
throw new IllegalArgumentException(s"Live members must have status [${Removed}], " + throw new IllegalArgumentException(s"Live members must have status [${Removed}], " +
s"got [${members.filter(_.status == Removed)}]") s"got [${members.filter(_.status == Removed)}]")
val inReachabilityButNotMember = overview.reachability.allObservers -- members.map(_.uniqueAddress) val inReachabilityButNotMember = overview.reachability.allObservers diff members.map(_.uniqueAddress)
if (inReachabilityButNotMember.nonEmpty) if (inReachabilityButNotMember.nonEmpty)
throw new IllegalArgumentException("Nodes not part of cluster in reachability table, got [%s]" throw new IllegalArgumentException("Nodes not part of cluster in reachability table, got [%s]"
format inReachabilityButNotMember.mkString(", ")) format inReachabilityButNotMember.mkString(", "))
val seenButNotMember = overview.seen -- members.map(_.uniqueAddress) val seenButNotMember = overview.seen diff members.map(_.uniqueAddress)
if (seenButNotMember.nonEmpty) if (seenButNotMember.nonEmpty)
throw new IllegalArgumentException("Nodes not part of cluster have marked the Gossip as seen, got [%s]" throw new IllegalArgumentException("Nodes not part of cluster have marked the Gossip as seen, got [%s]"
format seenButNotMember.mkString(", ")) format seenButNotMember.mkString(", "))
@ -129,7 +129,7 @@ private[cluster] final case class Gossip(
* Merges the seen table of two Gossip instances. * Merges the seen table of two Gossip instances.
*/ */
def mergeSeen(that: Gossip): Gossip = def mergeSeen(that: Gossip): Gossip =
this copy (overview = overview copy (seen = overview.seen ++ that.overview.seen)) this copy (overview = overview copy (seen = overview.seen union that.overview.seen))
/** /**
* Merges two Gossip instances including membership tables, and the VectorClock histories. * Merges two Gossip instances including membership tables, and the VectorClock histories.
@ -141,7 +141,7 @@ private[cluster] final case class Gossip(
val mergedVClock = this.version merge that.version val mergedVClock = this.version merge that.version
// 2. merge members by selecting the single Member with highest MemberStatus out of the Member groups // 2. merge members by selecting the single Member with highest MemberStatus out of the Member groups
val mergedMembers = Gossip.emptyMembers ++ Member.pickHighestPriority(this.members, that.members) val mergedMembers = Gossip.emptyMembers union Member.pickHighestPriority(this.members, that.members)
// 3. merge reachability table by picking records with highest version // 3. merge reachability table by picking records with highest version
val mergedReachability = this.overview.reachability.merge(mergedMembers.map(_.uniqueAddress), val mergedReachability = this.overview.reachability.merge(mergedMembers.map(_.uniqueAddress),

4
akka-cluster/src/main/scala/akka/cluster/Reachability.scala
View file

@ -80,7 +80,7 @@ private[cluster] class Reachability private (
val observerRowsMap: Map[UniqueAddress, Map[UniqueAddress, Reachability.Record]] = mapBuilder.toMap val observerRowsMap: Map[UniqueAddress, Map[UniqueAddress, Reachability.Record]] = mapBuilder.toMap
val allTerminated: Set[UniqueAddress] = terminatedBuilder.result() val allTerminated: Set[UniqueAddress] = terminatedBuilder.result()
val allUnreachable: Set[UniqueAddress] = unreachableBuilder.result() -- allTerminated val allUnreachable: Set[UniqueAddress] = unreachableBuilder.result() diff allTerminated
(observerRowsMap, allUnreachable, allTerminated) (observerRowsMap, allUnreachable, allTerminated)
} }
@ -88,7 +88,7 @@ private[cluster] class Reachability private (
val allUnreachableOrTerminated: Set[UniqueAddress] = val allUnreachableOrTerminated: Set[UniqueAddress] =
if (allTerminated.isEmpty) allUnreachable if (allTerminated.isEmpty) allUnreachable
else allUnreachable ++ allTerminated else allUnreachable union allTerminated
} }

2
akka-cluster/src/main/scala/akka/cluster/protobuf/ClusterMessageSerializer.scala
View file

@ -195,7 +195,7 @@ class ClusterMessageSerializer(val system: ExtendedActorSystem) extends BaseSeri
val allMembers = gossip.members.toVector val allMembers = gossip.members.toVector
val allAddresses: Vector[UniqueAddress] = allMembers.map(_.uniqueAddress) val allAddresses: Vector[UniqueAddress] = allMembers.map(_.uniqueAddress)
val addressMapping = allAddresses.zipWithIndex.toMap val addressMapping = allAddresses.zipWithIndex.toMap
val allRoles = allMembers.foldLeft(Set.empty[String])((acc, m) ⇒ acc ++ m.roles).to[Vector] val allRoles = allMembers.foldLeft(Set.empty[String])((acc, m) ⇒ acc union m.roles).to[Vector]
val roleMapping = allRoles.zipWithIndex.toMap val roleMapping = allRoles.zipWithIndex.toMap
val allHashes = gossip.version.versions.keys.to[Vector] val allHashes = gossip.version.versions.keys.to[Vector]
val hashMapping = allHashes.zipWithIndex.toMap val hashMapping = allHashes.zipWithIndex.toMap

2
akka-cluster/src/multi-jvm/scala/akka/cluster/MultiNodeClusterSpec.scala
View file

@ -303,7 +303,7 @@ trait MultiNodeClusterSpec extends Suite with STMultiNodeSpec with WatchedByCoro
* Wait until the specified nodes have seen the same gossip overview. * Wait until the specified nodes have seen the same gossip overview.
*/ */
def awaitSeenSameState(addresses: Address*): Unit = def awaitSeenSameState(addresses: Address*): Unit =
awaitAssert((addresses.toSet -- clusterView.seenBy) should ===(Set.empty)) awaitAssert((addresses.toSet diff clusterView.seenBy) should ===(Set.empty))
/** /**
* Leader according to the address ordering of the roles. * Leader according to the address ordering of the roles.

6
akka-cluster/src/multi-jvm/scala/akka/cluster/StressSpec.scala
View file

@ -3,8 +3,8 @@
*/ */
package akka.cluster package akka.cluster
// TODO remove metrics // TODO remove metrics
// FIXME this test is not migrated to metrics extension // FIXME this test is not migrated to metrics extension
import language.postfixOps import language.postfixOps
import scala.annotation.tailrec import scala.annotation.tailrec
@ -981,7 +981,7 @@ abstract class StressSpec
timeout = remainingOrDefault) timeout = remainingOrDefault)
awaitAllReachable() awaitAllReachable()
} }
val nextAddresses = clusterView.members.map(_.address) -- usedAddresses val nextAddresses = clusterView.members.map(_.address) diff usedAddresses
runOn(usedRoles: _*) { runOn(usedRoles: _*) {
nextAddresses.size should ===(numberOfNodesJoinRemove) nextAddresses.size should ===(numberOfNodesJoinRemove)
} }

6
akka-cluster/src/multi-jvm/scala/akka/cluster/TransitionSpec.scala
View file

@ -46,7 +46,7 @@ abstract class TransitionSpec
def nonLeader(roles: RoleName*) = roles.toSeq.sorted.tail def nonLeader(roles: RoleName*) = roles.toSeq.sorted.tail
def memberStatus(address: Address): MemberStatus = { def memberStatus(address: Address): MemberStatus = {
val statusOption = (clusterView.members ++ clusterView.unreachableMembers).collectFirst { val statusOption = (clusterView.members union clusterView.unreachableMembers).collectFirst {
case m if m.address == address ⇒ m.status case m if m.address == address ⇒ m.status
} }
statusOption.getOrElse(Removed) statusOption.getOrElse(Removed)
@ -91,7 +91,7 @@ abstract class TransitionSpec
clusterView.latestStats.gossipStats.receivedGossipCount != oldCount // received gossip clusterView.latestStats.gossipStats.receivedGossipCount != oldCount // received gossip
} }
// gossip chat will synchronize the views // gossip chat will synchronize the views
awaitCond((Set(fromRole, toRole) -- seenLatestGossip).isEmpty) awaitCond((Set(fromRole, toRole) diff seenLatestGossip).isEmpty)
enterBarrier("after-gossip-" + gossipBarrierCounter) enterBarrier("after-gossip-" + gossipBarrierCounter)
} }
runOn(fromRole) { runOn(fromRole) {
@ -99,7 +99,7 @@ abstract class TransitionSpec
// send gossip // send gossip
cluster.clusterCore ! InternalClusterAction.SendGossipTo(toRole) cluster.clusterCore ! InternalClusterAction.SendGossipTo(toRole)
// gossip chat will synchronize the views // gossip chat will synchronize the views
awaitCond((Set(fromRole, toRole) -- seenLatestGossip).isEmpty) awaitCond((Set(fromRole, toRole) diff seenLatestGossip).isEmpty)
enterBarrier("after-gossip-" + gossipBarrierCounter) enterBarrier("after-gossip-" + gossipBarrierCounter)
} }
runOn(roles.filterNot(r ⇒ r == fromRole || r == toRole): _*) { runOn(roles.filterNot(r ⇒ r == fromRole || r == toRole): _*) {

2
akka-cluster/src/multi-jvm/scala/akka/cluster/routing/ClusterRoundRobinSpec.scala
View file

@ -342,7 +342,7 @@ abstract class ClusterRoundRobinSpec extends MultiNodeSpec(ClusterRoundRobinMult
def routeeAddresses = (routees map { case ActorRefRoutee(ref) ⇒ fullAddress(ref) }).toSet def routeeAddresses = (routees map { case ActorRefRoutee(ref) ⇒ fullAddress(ref) }).toSet
routees foreach { case ActorRefRoutee(ref) ⇒ watch(ref) } routees foreach { case ActorRefRoutee(ref) ⇒ watch(ref) }
val notUsedAddress = ((roles map address).toSet -- routeeAddresses).head val notUsedAddress = ((roles map address).toSet diff routeeAddresses).head
val downAddress = routeeAddresses.find(_ != address(first)).get val downAddress = routeeAddresses.find(_ != address(first)).get
val downRouteeRef = routees.collectFirst { val downRouteeRef = routees.collectFirst {
case ActorRefRoutee(ref) if ref.path.address == downAddress ⇒ ref case ActorRefRoutee(ref) if ref.path.address == downAddress ⇒ ref

6
akka-cluster/src/test/scala/akka/cluster/ClusterHeartbeatSenderStateSpec.scala
View file

@ -163,7 +163,7 @@ class ClusterHeartbeatSenderStateSpec extends WordSpec with Matchers {
val oldUnreachable = state.oldReceiversNowUnreachable val oldUnreachable = state.oldReceiversNowUnreachable
state = state.addMember(node) state = state.addMember(node)
// keep unreachable // keep unreachable
(oldUnreachable -- state.activeReceivers) should ===(Set.empty) (oldUnreachable diff state.activeReceivers) should ===(Set.empty)
state.failureDetector.isMonitoring(node.address) should ===(false) state.failureDetector.isMonitoring(node.address) should ===(false)
state.failureDetector.isAvailable(node.address) should ===(true) state.failureDetector.isAvailable(node.address) should ===(true)
} }
@ -174,9 +174,9 @@ class ClusterHeartbeatSenderStateSpec extends WordSpec with Matchers {
state = state.removeMember(node) state = state.removeMember(node)
// keep unreachable, unless it was the removed // keep unreachable, unless it was the removed
if (oldUnreachable(node)) if (oldUnreachable(node))
(oldUnreachable -- state.activeReceivers) should ===(Set(node)) (oldUnreachable diff state.activeReceivers) should ===(Set(node))
else else
(oldUnreachable -- state.activeReceivers) should ===(Set.empty) (oldUnreachable diff state.activeReceivers) should ===(Set.empty)
state.failureDetector.isMonitoring(node.address) should ===(false) state.failureDetector.isMonitoring(node.address) should ===(false)
state.failureDetector.isAvailable(node.address) should ===(true) state.failureDetector.isAvailable(node.address) should ===(true)

2
akka-persistence/src/main/scala/akka/persistence/journal/leveldb/LeveldbStore.scala
View file

@ -63,7 +63,7 @@ private[persistence] trait LeveldbStore extends Actor with WriteJournalBase with
case _ ⇒ (p, Set.empty[String]) case _ ⇒ (p, Set.empty[String])
} }
if (tags.nonEmpty && hasTagSubscribers) if (tags.nonEmpty && hasTagSubscribers)
allTags ++= tags allTags = allTags union tags
require(!p2.persistenceId.startsWith(tagPersistenceIdPrefix), require(!p2.persistenceId.startsWith(tagPersistenceIdPrefix),
s"persistenceId [${p.persistenceId}] must not start with $tagPersistenceIdPrefix") s"persistenceId [${p.persistenceId}] must not start with $tagPersistenceIdPrefix")

4
akka-remote/src/test/scala/akka/remote/RemoteInitErrorSpec.scala
View file

@ -41,7 +41,7 @@ class RemoteInitErrorSpec extends FlatSpec with Matchers {
} }
""").resolve() """).resolve()
def currentThreadIds() = { def currentThreadIds(): Set[Long] = {
val threads = Thread.getAllStackTraces().keySet() val threads = Thread.getAllStackTraces().keySet()
threads.collect({ case t: Thread if (!t.isDaemon()) ⇒ t.getId() }) threads.collect({ case t: Thread if (!t.isDaemon()) ⇒ t.getId() })
} }
@ -56,7 +56,7 @@ class RemoteInitErrorSpec extends FlatSpec with Matchers {
eventually(timeout(30 seconds), interval(800 milliseconds)) { eventually(timeout(30 seconds), interval(800 milliseconds)) {
val current = currentThreadIds() val current = currentThreadIds()
// no new threads should remain compared to the start state // no new threads should remain compared to the start state
(current -- start) should be(empty) (current diff start) should be(empty)
} }
} }
} }

4
akka-remote/src/test/scala/akka/remote/Ticket1978CommunicationSpec.scala
View file

@ -74,8 +74,8 @@ object Configuration {
} }
val engine = NettySSLSupport.initializeClientSSL(settings, NoLogging).getEngine val engine = NettySSLSupport.initializeClientSSL(settings, NoLogging).getEngine
val gotAllSupported = enabled.toSet -- engine.getSupportedCipherSuites.toSet val gotAllSupported = enabled.toSet diff engine.getSupportedCipherSuites.toSet
val gotAllEnabled = enabled.toSet -- engine.getEnabledCipherSuites.toSet val gotAllEnabled = enabled.toSet diff engine.getEnabledCipherSuites.toSet
gotAllSupported.isEmpty || (throw new IllegalArgumentException("Cipher Suite not supported: " + gotAllSupported)) gotAllSupported.isEmpty || (throw new IllegalArgumentException("Cipher Suite not supported: " + gotAllSupported))
gotAllEnabled.isEmpty || (throw new IllegalArgumentException("Cipher Suite not enabled: " + gotAllEnabled)) gotAllEnabled.isEmpty || (throw new IllegalArgumentException("Cipher Suite not enabled: " + gotAllEnabled))
engine.getSupportedProtocols.contains(settings.SSLProtocol.get) || engine.getSupportedProtocols.contains(settings.SSLProtocol.get) ||