[FAB-5313] Leader election yield if deliver unavailable

This commit makes the peer relinquish its leadership if it can't
connect to the ordering service.

Change-Id: I5fe679d5f23e539828fea4a9398c7dd4d9fd0f93
Signed-off-by: yacovm <yacovm@il.ibm.com>

core/deliverservice/deliveryclient.go

@@ -34,12 +34,20 @@ var (
  reConnectBackoffThreshold = float64(time.Hour)
 )
 
+// SetReconnectTotalTimeThreshold sets the total time the delivery service
+// may spend in reconnection attempts until its retry logic gives up
+// and returns an error
+func SetReconnectTotalTimeThreshold(duration time.Duration) {
+ reConnectTotalTimeThreshold = duration
+}
+
 // DeliverService used to communicate with orderers to obtain
 // new blocks and send them to the committer service
 type DeliverService interface {
  // StartDeliverForChannel dynamically starts delivery of new blocks from ordering service
  // to channel peers.
- StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo) error
+ // When the delivery finishes, the finalizer func is called
+ StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo, finalizer func()) error
 
  // StopDeliverForChannel dynamically stops delivery of new blocks from ordering service
  // to channel peers.
@@ -117,7 +125,7 @@ func (d *deliverServiceImpl) validateConfiguration() error {
 // initializes the grpc stream for given chainID, creates blocks provider instance
 // that spawns in go routine to read new blocks starting from the position provided by ledger
 // info instance.
-func (d *deliverServiceImpl) StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo) error {
+func (d *deliverServiceImpl) StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo, finalizer func()) error {
  d.lock.Lock()
  defer d.lock.Unlock()
  if d.stopping {
@@ -133,7 +141,10 @@ func (d *deliverServiceImpl) StartDeliverForChannel(chainID string, ledgerInfo b
  client := d.newClient(chainID, ledgerInfo)
  logger.Debug("This peer will pass blocks from orderer service to other peers for channel", chainID)
  d.blockProviders[chainID] = blocksprovider.NewBlocksProvider(chainID, client, d.conf.Gossip, d.conf.CryptoSvc)
- go d.blockProviders[chainID].DeliverBlocks()
+ go func() {
+ d.blockProviders[chainID].DeliverBlocks()
+ finalizer()
+ }()
  }
  return nil
 }

core/deliverservice/deliveryclient_test.go

@@ -113,10 +113,10 @@ func TestNewDeliverService(t *testing.T) {
  ConnFactory: connFactory,
  })
  assert.NoError(t, err)
- assert.NoError(t, service.StartDeliverForChannel("TEST_CHAINID", &mocks.MockLedgerInfo{0}))
+ assert.NoError(t, service.StartDeliverForChannel("TEST_CHAINID", &mocks.MockLedgerInfo{0}, func() {}))
 
  // Lets start deliver twice
- assert.Error(t, service.StartDeliverForChannel("TEST_CHAINID", &mocks.MockLedgerInfo{0}), "can't start delivery")
+ assert.Error(t, service.StartDeliverForChannel("TEST_CHAINID", &mocks.MockLedgerInfo{0}, func() {}), "can't start delivery")
  // Lets stop deliver that not started
  assert.Error(t, service.StopDeliverForChannel("TEST_CHAINID2"), "can't stop delivery")
 
@@ -130,7 +130,7 @@ func TestNewDeliverService(t *testing.T) {
  assert.Equal(t, 0, connNumber)
  assertBlockDissemination(0, gossipServiceAdapter.GossipBlockDisseminations, t)
  assert.Equal(t, atomic.LoadInt32(&blocksDeliverer.RecvCnt), atomic.LoadInt32(&gossipServiceAdapter.AddPayloadsCnt))
- assert.Error(t, service.StartDeliverForChannel("TEST_CHAINID", &mocks.MockLedgerInfo{0}), "Delivery service is stopping")
+ assert.Error(t, service.StartDeliverForChannel("TEST_CHAINID", &mocks.MockLedgerInfo{0}, func() {}), "Delivery service is stopping")
  assert.Error(t, service.StopDeliverForChannel("TEST_CHAINID"), "Delivery service is stopping")
 }
 
@@ -157,7 +157,7 @@ func TestDeliverServiceRestart(t *testing.T) {
  li := &mocks.MockLedgerInfo{Height: uint64(100)}
  os.SetNextExpectedSeek(uint64(100))
 
- err = service.StartDeliverForChannel("TEST_CHAINID", li)
+ err = service.StartDeliverForChannel("TEST_CHAINID", li, func() {})
  assert.NoError(t, err, "can't start delivery")
  // Check that delivery client requests blocks in order
  go os.SendBlock(uint64(100))
@@ -203,7 +203,7 @@ func TestDeliverServiceFailover(t *testing.T) {
  os1.SetNextExpectedSeek(uint64(100))
  os2.SetNextExpectedSeek(uint64(100))
 
- err = service.StartDeliverForChannel("TEST_CHAINID", li)
+ err = service.StartDeliverForChannel("TEST_CHAINID", li, func() {})
  assert.NoError(t, err, "can't start delivery")
  // We need to discover to which instance the client connected to
  go os1.SendBlock(uint64(100))
@@ -278,7 +278,7 @@ func TestDeliverServiceServiceUnavailable(t *testing.T) {
  os1.SetNextExpectedSeek(li.Height)
  os2.SetNextExpectedSeek(li.Height)
 
- err = service.StartDeliverForChannel("TEST_CHAINID", li)
+ err = service.StartDeliverForChannel("TEST_CHAINID", li, func() {})
  assert.NoError(t, err, "can't start delivery")
 
  waitForConnectionToSomeOSN := func() (*mocks.Orderer, *mocks.Orderer) {
@@ -367,7 +367,7 @@ func TestDeliverServiceShutdown(t *testing.T) {
 
  li := &mocks.MockLedgerInfo{Height: uint64(100)}
  os.SetNextExpectedSeek(uint64(100))
- err = service.StartDeliverForChannel("TEST_CHAINID", li)
+ err = service.StartDeliverForChannel("TEST_CHAINID", li, func() {})
  assert.NoError(t, err, "can't start delivery")
 
  // Check that delivery service requests blocks in order

core/peer/peer_test.go

@@ -48,7 +48,7 @@ type mockDeliveryClient struct {
 
 // StartDeliverForChannel dynamically starts delivery of new blocks from ordering service
 // to channel peers.
-func (ds *mockDeliveryClient) StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo) error {
+func (ds *mockDeliveryClient) StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo, f func()) error {
  return nil
 }
 

core/scc/cscc/configure_test.go

@@ -55,7 +55,7 @@ type mockDeliveryClient struct {
 
 // StartDeliverForChannel dynamically starts delivery of new blocks from ordering service
 // to channel peers.
-func (ds *mockDeliveryClient) StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo) error {
+func (ds *mockDeliveryClient) StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo, f func()) error {
  return nil
 }
 

gossip/election/election.go

@@ -93,6 +93,10 @@ type LeaderElectionService interface {
 
  // Stop stops the LeaderElectionService
  Stop()
+
+ // Yield relinquishes the leadership until a new leader is elected,
+ // or a timeout expires
+ Yield()
 }
 
 type peerID []byte
@@ -150,10 +154,12 @@ type leaderElectionSvcImpl struct {
  isLeader int32
  toDie int32
  leaderExists int32
+ yield int32
  sleeping bool
  adapter LeaderElectionAdapter
  logger *logging.Logger
  callback leadershipCallback
+ yieldTimer *time.Timer
 }
 
 func (le *leaderElectionSvcImpl) start() {
@@ -239,6 +245,11 @@ func (le *leaderElectionSvcImpl) run() {
  if !le.isLeaderExists() {
  le.leaderElection()
  }
+ // If we are yielding and some leader has been elected,
+ // stop yielding
+ if le.isLeaderExists() && le.isYielding() {
+ le.stopYielding()
+ }
  if le.shouldStop() {
  return
  }
@@ -253,14 +264,26 @@ func (le *leaderElectionSvcImpl) run() {
 func (le *leaderElectionSvcImpl) leaderElection() {
  le.logger.Debug(le.id, ": Entering")
  defer le.logger.Debug(le.id, ": Exiting")
+ // If we're yielding to other peers, do not participate
+ // in leader election
+ if le.isYielding() {
+ return
+ }
+ // Propose ourselves as a leader
  le.propose()
+ // Collect other proposals
  le.waitForInterrupt(getLeaderElectionDuration())
  // If someone declared itself as a leader, give up
  // on trying to become a leader too
  if le.isLeaderExists() {
  le.logger.Debug(le.id, ": Some peer is already a leader")
  return
  }
+
+ if le.isYielding() {
+ le.logger.Debug(le.id, ": Aborting leader election because yielding")
+ return
+ }
  // Leader doesn't exist, let's see if there is a better candidate than us
  // for being a leader
  for _, o := range le.proposals.ToArray() {
@@ -364,6 +387,38 @@ func (le *leaderElectionSvcImpl) shouldStop() bool {
  return atomic.LoadInt32(&le.toDie) == int32(1)
 }
 
+func (le *leaderElectionSvcImpl) isYielding() bool {
+ return atomic.LoadInt32(&le.yield) == int32(1)
+}
+
+func (le *leaderElectionSvcImpl) stopYielding() {
+ le.logger.Debug("Stopped yielding")
+ le.Lock()
+ defer le.Unlock()
+ atomic.StoreInt32(&le.yield, int32(0))
+ le.yieldTimer.Stop()
+}
+
+// Yield relinquishes the leadership until a new leader is elected,
+// or a timeout expires
+func (le *leaderElectionSvcImpl) Yield() {
+ le.Lock()
+ defer le.Unlock()
+ if !le.IsLeader() || le.isYielding() {
+ return
+ }
+ // Turn on the yield flag
+ atomic.StoreInt32(&le.yield, int32(1))
+ // Stop being a leader
+ le.stopBeingLeader()
+ // Clear the leader exists flag since it could be that we are the leader
+ atomic.StoreInt32(&le.leaderExists, int32(0))
+ // Clear the yield flag in any case afterwards
+ le.yieldTimer = time.AfterFunc(getLeaderAliveThreshold()*6, func() {
+ atomic.StoreInt32(&le.yield, int32(0))
+ })
+}
+
 // Stop stops the LeaderElectionService
 func (le *leaderElectionSvcImpl) Stop() {
  le.logger.Debug(le.id, ": Entering")

gossip/election/election_test.go

@@ -308,6 +308,70 @@ func TestLeadershipTakeover(t *testing.T) {
  assert.Equal(t, "p2", leaders[0])
 }
 
+func TestYield(t *testing.T) {
+ t.Parallel()
+ // Scenario: Peers spawn and a leader is elected.
+ // After a while, the leader yields.
+ // (Call yield twice to ensure only one callback is called)
+ // Expected outcome:
+ // (1) A new leader is elected
+ // (2) The old leader doesn't take back its leadership
+ peers := createPeers(0, 0, 1, 2, 3, 4, 5)
+ leaders := waitForLeaderElection(t, peers)
+ assert.Len(t, leaders, 1, "Only 1 leader should have been elected")
+ assert.Equal(t, "p0", leaders[0])
+ peers[0].Yield()
+ // Ensure the callback was called with 'false'
+ assert.True(t, peers[0].isCallbackInvoked())
+ assert.False(t, peers[0].isLeaderFromCallback())
+ // Clear the callback invoked flag
+ peers[0].lock.Lock()
+ peers[0].callbackInvoked = false
+ peers[0].lock.Unlock()
+ // Yield again and ensure it isn't called again
+ peers[0].Yield()
+ assert.False(t, peers[0].isCallbackInvoked())
+
+ ensureP0isNotAleader := func() bool {
+ leaders := waitForLeaderElection(t, peers)
+ return len(leaders) == 1 && leaders[0] != "p0"
+ }
+ // A new leader is elected, and it is not p0
+ waitForBoolFunc(t, ensureP0isNotAleader, true)
+ time.Sleep(getLeaderAliveThreshold() * 2)
+ // After a while, p0 doesn't restore its leadership status
+ waitForBoolFunc(t, ensureP0isNotAleader, true)
+}
+
+func TestYieldSinglePeer(t *testing.T) {
+ t.Parallel()
+ // Scenario: spawn a single peer and have it yield.
+ // Ensure it recovers its leadership after a while.
+ peers := createPeers(0, 0)
+ waitForLeaderElection(t, peers)
+ peers[0].Yield()
+ assert.False(t, peers[0].IsLeader())
+ waitForLeaderElection(t, peers)
+}
+
+func TestYieldAllPeers(t *testing.T) {
+ t.Parallel()
+ // Scenario: spawn 2 peers and have them all yield after regaining leadership.
+ // Ensure the first peer is the leader in the end after both peers yield
+ peers := createPeers(0, 0, 1)
+ leaders := waitForLeaderElection(t, peers)
+ assert.Len(t, leaders, 1, "Only 1 leader should have been elected")
+ assert.Equal(t, "p0", leaders[0])
+ peers[0].Yield()
+ leaders = waitForLeaderElection(t, peers)
+ assert.Len(t, leaders, 1, "Only 1 leader should have been elected")
+ assert.Equal(t, "p1", leaders[0])
+ peers[1].Yield()
+ leaders = waitForLeaderElection(t, peers)
+ assert.Len(t, leaders, 1, "Only 1 leader should have been elected")
+ assert.Equal(t, "p0", leaders[0])
+}
+
 func TestPartition(t *testing.T) {
  t.Parallel()
  // Scenario: peers spawn together, and then after a while a network partition occurs

gossip/service/gossip_service.go

@@ -196,7 +196,7 @@ func (g *gossipServiceImpl) InitializeChannel(chainID string, committer committe
  g.leaderElection[chainID] = g.newLeaderElectionComponent(chainID, g.onStatusChangeFactory(chainID, committer))
  } else if isStaticOrgLeader {
  logger.Debug("This peer is configured to connect to ordering service for blocks delivery, channel", chainID)
- g.deliveryService.StartDeliverForChannel(chainID, committer)
+ g.deliveryService.StartDeliverForChannel(chainID, committer, func() {})
  } else {
  logger.Debug("This peer is not configured to connect to ordering service for blocks delivery, channel", chainID)
  }
@@ -282,8 +282,12 @@ func (g *gossipServiceImpl) amIinChannel(myOrg string, config Config) bool {
 func (g *gossipServiceImpl) onStatusChangeFactory(chainID string, committer blocksprovider.LedgerInfo) func(bool) {
  return func(isLeader bool) {
  if isLeader {
+ yield := func() {
+ le := g.leaderElection[chainID]
+ le.Yield()
+ }
  logger.Info("Elected as a leader, starting delivery service for channel", chainID)
- if err := g.deliveryService.StartDeliverForChannel(chainID, committer); err != nil {
+ if err := g.deliveryService.StartDeliverForChannel(chainID, committer, yield); err != nil {
  logger.Error("Delivery service is not able to start blocks delivery for chain, due to", err)
  }
  } else {

gossip/service/gossip_service_test.go

@@ -276,7 +276,7 @@ type mockDeliverService struct {
  running map[string]bool
 }
 
-func (ds *mockDeliverService) StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo) error {
+func (ds *mockDeliverService) StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo, finalizer func()) error {
  ds.running[chainID] = true
  return nil
 }