diff --git a/internal/bft/viewchanger.go b/internal/bft/viewchanger.go index 58d9c9da..85d814a6 100644 --- a/internal/bft/viewchanger.go +++ b/internal/bft/viewchanger.go @@ -330,12 +330,25 @@ func (v *ViewChanger) processMsg(sender uint64, m *protos.Message) { } } -// InformNewView tells the view changer to advance to a new view number +// InformNewView tells the view changer to advance to a new view number. +// It is called from the controller's run loop after a sync, and so it must not wait for the view changer. +// If the view changer did not process a previous view yet, the pending view is replaced by the higher of the two, +// which loses nothing since informNewView ignores views lower than the current one. +// The pending view is held only in informChan, so concurrent calls are safe as well. func (v *ViewChanger) InformNewView(view uint64) { - select { - case v.informChan <- view: - case <-v.stopChan: - return + for { + select { + case v.informChan <- view: + return + case <-v.stopChan: + return + default: + } + select { + case pending := <-v.informChan: + view = max(view, pending) + default: + } } } diff --git a/internal/bft/viewchanger_test.go b/internal/bft/viewchanger_test.go index 0918d43f..f575a75b 100644 --- a/internal/bft/viewchanger_test.go +++ b/internal/bft/viewchanger_test.go @@ -1578,7 +1578,10 @@ func TestInformViewChanger(t *testing.T) { }) reqTimer := &mocks.RequestsTimer{} reqTimer.On("StopTimers").Once() - reqTimer.On("RestartTimers") + informed := make(chan struct{}, 1) + reqTimer.On("RestartTimers").Run(func(args mock.Arguments) { + informed <- struct{}{} // the view changer restarts the timers once it is informed of a new view + }) basicLog, err := zap.NewDevelopment() assert.NoError(t, err) log := basicLog.Sugar() @@ -1600,7 +1603,7 @@ func TestInformViewChanger(t *testing.T) { info := uint64(2) vc.InformNewView(info) // increase the view number - vc.InformNewView(info) // make sure that inform happened (channel size is 1) + <-informed // make sure that inform happened vc.StartViewChange(2, true) msg := <-msgChan @@ -1614,6 +1617,69 @@ func TestInformViewChanger(t *testing.T) { reqTimer.AssertCalled(t, "RestartTimers") } +// TestInformNewViewDoesNotBlock makes sure that informing the view changer of a new view never blocks, +// even when it is not processing the views it was already informed of, and that it ends up in the latest one. +func TestInformNewViewDoesNotBlock(t *testing.T) { + comm := &mocks.CommMock{} + msgChan := make(chan *protos.Message) + comm.On("BroadcastConsensus", mock.Anything).Run(func(args mock.Arguments) { + msgChan <- args.Get(0).(*protos.Message) + }) + reqTimer := &mocks.RequestsTimer{} + reqTimer.On("StopTimers") + informed := make(chan struct{}, 1) + reqTimer.On("RestartTimers").Run(func(args mock.Arguments) { + informed <- struct{}{} // the view changer restarts the timers once it is informed of a new view + }) + basicLog, err := zap.NewDevelopment() + assert.NoError(t, err) + controller := &mocks.ViewController{} + controller.On("AbortView", mock.Anything) + + vc := &bft.ViewChanger{ + N: 4, + NodesList: []uint64{0, 1, 2, 3}, + Comm: comm, + RequestsTimer: reqTimer, + Ticker: make(chan time.Time), + Logger: basicLog.Sugar(), + Controller: controller, + InMsqQSize: 100, + } + // the view changer does not run until the controller started, so it processes none of the views it is informed of + vc.ControllerStartedWG.Add(1) + vc.Start(0) + + returned := make(chan struct{}) + go func() { + defer close(returned) + for view := uint64(1); view <= 3; view++ { + vc.InformNewView(view) + } + }() + select { + case <-returned: + case <-time.After(3 * time.Second): + assert.Fail(t, "informing the view changer of a new view blocked") + } + + vc.ControllerStartedWG.Done() + select { + case <-informed: + case <-time.After(3 * time.Second): + assert.Fail(t, "the view changer did not process the views it was informed of") + } + vc.StartViewChange(3, true) + select { + case msg := <-msgChan: + assert.Equal(t, uint64(4), msg.GetViewChange().GetNextView(), "the view changer is not in the latest view it was informed of") + case <-time.After(3 * time.Second): + assert.Fail(t, "the view changer did not start a view change") + } + + vc.Stop() +} + func TestRestoreViewChange(t *testing.T) { comm := &mocks.CommMock{} broadcastChan := make(chan *protos.Message)