Re-send messages in critical path to other nodes

Author: ranchalp

pkg/accountability/simpleacc/accountability.go

@@ -125,6 +125,7 @@ func NewReconfigurableModule(mc ModuleConfig, paramsTemplate ModuleParams, logge
 				// Fill in instance-specific parameters.
 				moduleParams := paramsTemplate
 				moduleParams.Membership = accParams.Membership
+				moduleParams.RetentionIndex = accParams.RetentionIndex
 
 				// Create a new instance of the multisig collector.
 				accountabilityModule, err := NewModule(

pkg/accountability/simpleacc/common/common.go

@@ -6,6 +6,8 @@ import (
 	"github.com/filecoin-project/mir/pkg/logging"
 	accpbtypes "github.com/filecoin-project/mir/pkg/pb/accountabilitypb/types"
 	trantorpbtypes "github.com/filecoin-project/mir/pkg/pb/trantorpb/types"
+	timertypes "github.com/filecoin-project/mir/pkg/timer/types"
+	tt "github.com/filecoin-project/mir/pkg/trantor/types"
 	t "github.com/filecoin-project/mir/pkg/types"
 )
 
@@ -17,15 +19,18 @@ type ModuleConfig struct {
 	Ordering t.ModuleID // provides Predecisions
 	App      t.ModuleID // receives Decisions and/or PoMs
 	Crypto   t.ModuleID // provides cryptographic primitives
+	Timer    t.ModuleID // provides Timing primitives
 	Net      t.ModuleID // provides network primitives
 }
 
 // ModuleParams sets the values for the parameters of an instance of the protocol.
 // All replicas are expected to use identical module parameters.
 type ModuleParams struct {
-	Membership        *trantorpbtypes.Membership // the list of participating nodes
+	Membership        *trantorpbtypes.Membership // The list of participating nodes.
 	LightCertificates bool
-	PomsHandler       func(m dsl.Module, // function to be called when PoMs detected
+	ResendFrequency   timertypes.Duration // Frequency with which messages in the critical path are re-sent
+	RetentionIndex    tt.RetentionIndex
+	PoMsHandler       func(m dsl.Module, // Function to be called when PoMs detected.
 		mc *ModuleConfig,
 		params *ModuleParams,
 		state *incommon.State,

pkg/accountability/simpleacc/internal/poms/poms.go

@@ -93,7 +93,7 @@ func HandlePoMs(
 	logger.Log(logging.LevelWarn, "Found valid PoMs! sending...")
 
 	// Handle PoMs according to the application's logic defined when creating the accountability factory
-	params.PomsHandler(m, mc, params, state, state.UnhandledPoMs, logger)
+	params.PoMsHandler(m, mc, params, state, state.UnhandledPoMs, logger)
 
 	for _, pom := range state.UnhandledPoMs {
 		state.HandledPoMs[pom.NodeId] = pom

pkg/accountability/simpleacc/internal/predecisions/predecisions.go

@@ -3,6 +3,9 @@ package predecisions
 import (
 	"reflect"
 
+	eventpbdsl "github.com/filecoin-project/mir/pkg/pb/eventpb/dsl"
+	eventpbtypes "github.com/filecoin-project/mir/pkg/pb/eventpb/types"
+
 	"github.com/filecoin-project/mir/pkg/accountability/simpleacc/internal/certificates/lightcertificates"
 	isspbdsl "github.com/filecoin-project/mir/pkg/pb/isspb/dsl"
 	isspbtypes "github.com/filecoin-project/mir/pkg/pb/isspb/types"
@@ -19,6 +22,7 @@ import (
 	cryptopbdsl "github.com/filecoin-project/mir/pkg/pb/cryptopb/dsl"
 	cryptopbtypes "github.com/filecoin-project/mir/pkg/pb/cryptopb/types"
 	transportpbdsl "github.com/filecoin-project/mir/pkg/pb/transportpb/dsl"
+	transportpbevents "github.com/filecoin-project/mir/pkg/pb/transportpb/events"
 	t "github.com/filecoin-project/mir/pkg/types"
 	"github.com/filecoin-project/mir/pkg/util/maputil"
 	"github.com/filecoin-project/mir/pkg/util/membutil"
@@ -79,7 +83,13 @@ func IncludePredecisions(
 		state.LocalPredecision.SignedPredecision.Signature = signature
 
 		// Broadcast signed predecision to all participants (including oneself).
-		transportpbdsl.SendMessage(m, mc.Net, accpbmsgs.SignedPredecision(mc.Self, sr.data, signature), maputil.GetKeys(params.Membership.Nodes))
+		eventpbdsl.TimerRepeat(m,
+			mc.Timer,
+			[]*eventpbtypes.Event{transportpbevents.SendMessage(mc.Net, accpbmsgs.SignedPredecision(mc.Self, sr.data, signature), maputil.GetKeys(params.Membership.Nodes))},
+			params.ResendFrequency,
+			params.RetentionIndex,
+		)
+
 		return nil
 	})
 
@@ -230,13 +240,16 @@ func decide(m dsl.Module, mc *common.ModuleConfig, params *common.ModuleParams,
 	}
 
 	// Find the actual predecision from other nodes
-	transportpbdsl.SendMessage(
-		m,
-		mc.Net,
-		accpbmsgs.RequestSBMessage(mc.Self,
-			predecision),
-		state.PredecisionNodeIDs[string(predecision)])
-
+	eventpbdsl.TimerRepeat(m,
+		mc.Timer,
+		[]*eventpbtypes.Event{transportpbevents.SendMessage(
+			mc.Net,
+			accpbmsgs.RequestSBMessage(mc.Self,
+				predecision),
+			state.PredecisionNodeIDs[string(predecision)])},
+		params.ResendFrequency,
+		params.RetentionIndex,
+	)
 }
 
 func finishWithDecision(

pkg/pb/accountabilitypb/accountabilitypb.pb.go

@@ -63,7 +63,8 @@ message FullCertificate {
 message InstanceParams {
   option (mir.struct) = true;
 
-  trantorpb.Membership membership = 3;
+  trantorpb.Membership membership = 1;
+  uint64      retention_index     = 2 [(mir.type) = "github.com/filecoin-project/mir/pkg/trantor/types.RetentionIndex"];
 }