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

=clu #3660 Simple speedup of gossip in early phase

Browse source
This commit is contained in:
Patrik Nordwall 2013-10-14 14:50:57 +02:00 • committed by Björn Antonsson
parent b4f3c01206
commit ce329e48c1
1 changed files with 17 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -111,6 +111,8 @@ private[cluster] object InternalClusterAction {
case object GossipTick extends Tick
case object GossipSpeedupTick extends Tick
case object HeartbeatTick extends Tick
case object ReapUnreachableTick extends Tick
@ -334,7 +336,8 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
def initialized: Actor.Receive = {
case msg: GossipEnvelope ⇒ receiveGossip(msg)
case msg: GossipStatus ⇒ receiveGossipStatus(msg)
case GossipTick ⇒ gossip()
case GossipTick ⇒ gossipTick()
case GossipSpeedupTick ⇒ gossipSpeedupTick()
case ReapUnreachableTick ⇒ reapUnreachableMembers()
case LeaderActionsTick ⇒ leaderActions()
case PublishStatsTick ⇒ publishInternalStats()
@ -680,7 +683,19 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
}
}
def mergeRate(count: Long): Double = (count * 1000.0) / GossipInterval.toMillis
def gossipTick(): Unit = {
gossip()
if (isGossipSpeedupNeeded) {
scheduler.scheduleOnce(GossipInterval / 3, self, GossipSpeedupTick)
scheduler.scheduleOnce(GossipInterval * 2 / 3, self, GossipSpeedupTick)
}
}
def gossipSpeedupTick(): Unit =
if (isGossipSpeedupNeeded) gossip()
def isGossipSpeedupNeeded: Boolean =
(latestGossip.overview.seen.size < latestGossip.members.size / 2)
/**
* Initiates a new round of gossip.