Garbage-collect timer and fix timer data race

Signed-off-by: Matej Pavlovic <matopavlovic@gmail.com>

pkg/iss/iss.go

@@ -586,6 +586,9 @@ func (iss *ISS) applyStableCheckpoint(stableCheckpoint *isspb.StableCheckpoint)
 		// stable checkpoint is associated with.
 		eventsOut.PushBack(events.WALTruncate(walModuleName, t.WALRetIndex(stableCheckpoint.Epoch)))
 
+		// Prune old Timer tasks
+		eventsOut.PushBack(events.TimerGarbageCollect(timerModuleName, t.TimerRetIndex(stableCheckpoint.Epoch)))
+
 		// Clean up the global ISS state from all the epoch
 		// instances that are associated with epoch numbers
 		// less than the epoch number of the new stable

pkg/timer/timer.go

@@ -6,6 +6,7 @@ import (
 	"github.com/filecoin-project/mir/pkg/events"
 	"github.com/filecoin-project/mir/pkg/pb/eventpb"
 	t "github.com/filecoin-project/mir/pkg/types"
+	"sync"
 	"time"
 )
 
@@ -14,7 +15,8 @@ import (
 type Timer struct {
 	eventsOut chan *events.EventList
 
-	retIndex t.TimerRetIndex
+	retIndexMutex sync.RWMutex
+	retIndex      t.TimerRetIndex
 }
 
 func New() *Timer {
@@ -107,7 +109,7 @@ func (tm *Timer) Repeat(
 		defer ticker.Stop()
 
 		// Repeat as long as this repetition task has not been garbage-collected.
-		for retIndex >= tm.retIndex {
+		for retIndex >= tm.getRetIndex() {
 
 			// Try to write events to the output channel until the context is canceled.
 			select {
@@ -135,9 +137,18 @@ func (tm *Timer) Repeat(
 // smaller than retIndex.
 // If GarbageCollect already has been invoked with the same or higher retention index, the call has no effect.
 func (tm *Timer) GarbageCollect(retIndex t.TimerRetIndex) {
+	tm.retIndexMutex.Lock()
+	defer tm.retIndexMutex.Unlock()
 
 	// Retention index must be monotonically increasing over calls to GarbageCollect.
 	if retIndex > tm.retIndex {
 		tm.retIndex = retIndex
 	}
 }
+
+func (tm *Timer) getRetIndex() t.TimerRetIndex {
+	tm.retIndexMutex.RLock()
+	defer tm.retIndexMutex.RUnlock()
+
+	return tm.retIndex
+}