Master thesis merge

.gitignore

@@ -20,7 +20,7 @@ __debug_bin
 # binaries
 /hotstuff
 /plot
-
+**/__debug_bin*
 
 # other
 *.in
@@ -35,3 +35,5 @@ measurements.json
 *.pdf
 
 twins.json
+
+/debug_logs

.vscode/launch.json

@@ -12,7 +12,7 @@
 			"program": "${workspaceRoot}/cmd/hotstuff/main.go",
 			"cwd": "${workspaceRoot}",
 			"args": [
-				"run"
+				"run",
 			]
 		},
 		{
@@ -55,4 +55,4 @@
 			"host": "127.0.0.1"
 		}
 	]
-}
+}

README.md

@@ -1,3 +1,7 @@
+# NOTICE
+
+This implementation was done as part of Alan Rostem's master's thesis. 
+
 # hotstuff
 
 [![Go Reference](https://pkg.go.dev/badge/github.com/relab/consensus.svg)](https://pkg.go.dev/github.com/relab/hotstuff)

backend/backend_test.go

@@ -46,8 +46,33 @@ func TestConnect(t *testing.T) {
 	runBoth(t, run)
 }
 
+// Mainly test initialization of scoped modules and how they depend on each other.
+func TestConnectScoped(t *testing.T) {
+	run := func(t *testing.T, setup setupFunc) {
+		const n = 4
+		ctrl := gomock.NewController(t)
+		td := setup(t, ctrl, n)
+		builder := modules.NewBuilder(1, td.keys[0])
+		testutil.TestModulesScoped(t, ctrl, 1, td.keys[0], &builder, 4)
+		teardown := createServers(t, td, ctrl)
+		defer teardown()
+		td.builders.Build()
+
+		cfg := NewConfig(td.creds, gorums.WithDialTimeout(time.Second))
+
+		builder.Add(cfg)
+		builder.Build()
+
+		err := cfg.Connect(td.replicas)
+		if err != nil {
+			t.Error(err)
+		}
+	}
+	runBoth(t, run)
+}
+
 // testBase is a generic test for a unicast/multicast call
-func testBase(t *testing.T, typ any, send func(modules.Configuration), handle eventloop.EventHandler) {
+func testBase(t *testing.T, typ any, send func(modules.Configuration), handle eventloop.EventHandler, opts ...eventloop.HandlerOption) {
 	run := func(t *testing.T, setup setupFunc) {
 		const n = 4
 		ctrl := gomock.NewController(t)
@@ -69,11 +94,11 @@ func testBase(t *testing.T, typ any, send func(modules.Configuration), handle ev
 		ctx, cancel := context.WithCancel(context.Background())
 		for _, hs := range hl[1:] {
 			var (
-				eventLoop    *eventloop.EventLoop
+				eventLoop    *eventloop.ScopedEventLoop
 				synchronizer modules.Synchronizer
 			)
 			hs.Get(&eventLoop, &synchronizer)
-			eventLoop.RegisterHandler(typ, handle)
+			eventLoop.RegisterHandler(typ, handle, opts...)
 			synchronizer.Start(ctx)
 			go eventLoop.Run(ctx)
 		}
@@ -89,10 +114,16 @@ func TestPropose(t *testing.T) {
 		ID: 1,
 		Block: hotstuff.NewBlock(
 			hotstuff.GetGenesis().Hash(),
-			hotstuff.NewQuorumCert(nil, 0, hotstuff.GetGenesis().Hash()),
-			"foo", 1, 1,
+			hotstuff.NewQuorumCert(
+				nil,
+				0,
+				hotstuff.NullPipe, // TODO: Verify if this code conflicts with pipelining
+				hotstuff.GetGenesis().Hash()),
+			"foo", 1, 1, 0,
 		),
+		Pipe: 0,
 	}
+
 	testBase(t, want, func(cfg modules.Configuration) {
 		wg.Add(3)
 		cfg.Propose(want)
@@ -109,13 +140,47 @@ func TestPropose(t *testing.T) {
 	})
 }
 
+func TestProposeScoped(t *testing.T) {
+	var wg sync.WaitGroup
+	pipe := hotstuff.Pipe(123)
+	want := hotstuff.ProposeMsg{
+		ID: 1,
+		Block: hotstuff.NewBlock(
+			hotstuff.GetGenesis().Hash(),
+			hotstuff.NewQuorumCert(
+				nil,
+				0,
+				hotstuff.NullPipe, // TODO: Verify if this code conflicts with pipelining
+				hotstuff.GetGenesis().Hash()),
+			"foo", 1, 1, pipe,
+		),
+		Pipe: pipe,
+	}
+
+	testBase(t, want, func(cfg modules.Configuration) {
+		wg.Add(3)
+		cfg.Propose(want)
+		wg.Wait()
+	}, func(event any) {
+		got := event.(hotstuff.ProposeMsg)
+		if got.ID != want.ID {
+			t.Errorf("wrong id in proposal: got: %d, want: %d", got.ID, want.ID)
+		}
+		if got.Block.Hash() != want.Block.Hash() {
+
+			t.Errorf("block hashes do not match. want %d got %d", got.Block.Pipe(), want.Block.Pipe())
+		}
+		wg.Done()
+	}, eventloop.RespondToScope(pipe))
+}
+
 func TestTimeout(t *testing.T) {
 	var wg sync.WaitGroup
 	want := hotstuff.TimeoutMsg{
 		ID:            1,
 		View:          1,
 		ViewSignature: nil,
-		SyncInfo:      hotstuff.NewSyncInfo(),
+		SyncInfo:      hotstuff.NewSyncInfo(hotstuff.NullPipe),
 	}
 	testBase(t, want, func(cfg modules.Configuration) {
 		wg.Add(3)
@@ -133,6 +198,33 @@ func TestTimeout(t *testing.T) {
 	})
 }
 
+func TestTimeoutScoped(t *testing.T) {
+	var wg sync.WaitGroup
+
+	pipe := hotstuff.Pipe(1)
+	want := hotstuff.TimeoutMsg{
+		ID:            1,
+		View:          1,
+		ViewSignature: nil,
+		SyncInfo:      hotstuff.NewSyncInfo(pipe),
+		Pipe:          pipe,
+	}
+	testBase(t, want, func(cfg modules.Configuration) {
+		wg.Add(3)
+		cfg.Timeout(want)
+		wg.Wait()
+	}, func(event any) {
+		got := event.(hotstuff.TimeoutMsg)
+		if got.ID != want.ID {
+			t.Errorf("wrong id in proposal: got: %d, want: %d", got.ID, want.ID)
+		}
+		if got.View != want.View {
+			t.Errorf("wrong view in proposal: got: %d, want: %d", got.View, want.View)
+		}
+		wg.Done()
+	}, eventloop.RespondToScope(pipe))
+}
+
 type testData struct {
 	n         int
 	creds     credentials.TransportCredentials

backend/config.go

@@ -24,7 +24,7 @@ import (
 
 // Replica provides methods used by hotstuff to send messages to replicas.
 type Replica struct {
-	eventLoop *eventloop.EventLoop
+	eventLoop *eventloop.ScopedEventLoop
 	node      *hotstuffpb.Node
 	id        hotstuff.ID
 	pubKey    hotstuff.PublicKey
@@ -46,7 +46,7 @@ func (r *Replica) Vote(cert hotstuff.PartialCert) {
 	if r.node == nil {
 		return
 	}
-	ctx, cancel := synchronizer.TimeoutContext(r.eventLoop.Context(), r.eventLoop)
+	ctx, cancel := synchronizer.ScopedTimeoutContext(r.eventLoop.Context(), r.eventLoop, cert.Pipe())
 	defer cancel()
 	pCert := hotstuffpb.PartialCertToProto(cert)
 	r.node.Vote(ctx, pCert)
@@ -57,7 +57,7 @@ func (r *Replica) NewView(msg hotstuff.SyncInfo) {
 	if r.node == nil {
 		return
 	}
-	ctx, cancel := synchronizer.TimeoutContext(r.eventLoop.Context(), r.eventLoop)
+	ctx, cancel := synchronizer.ScopedTimeoutContext(r.eventLoop.Context(), r.eventLoop, msg.Pipe())
 	defer cancel()
 	r.node.NewView(ctx, hotstuffpb.SyncInfoToProto(msg))
 }
@@ -78,7 +78,7 @@ type Config struct {
 }
 
 type subConfig struct {
-	eventLoop *eventloop.EventLoop
+	eventLoop *eventloop.ScopedEventLoop
 	logger    logging.Logger
 	opts      *modules.Options
 
@@ -87,7 +87,7 @@ type subConfig struct {
 }
 
 // InitModule initializes the configuration.
-func (cfg *Config) InitModule(mods *modules.Core) {
+func (cfg *Config) InitModule(mods *modules.Core, _ modules.ScopeInfo) {
 	mods.Get(
 		&cfg.eventLoop,
 		&cfg.logger,
@@ -287,7 +287,7 @@ func (cfg *subConfig) Propose(proposal hotstuff.ProposeMsg) {
 	if cfg.cfg == nil {
 		return
 	}
-	ctx, cancel := synchronizer.TimeoutContext(cfg.eventLoop.Context(), cfg.eventLoop)
+	ctx, cancel := synchronizer.ScopedTimeoutContext(cfg.eventLoop.Context(), cfg.eventLoop, proposal.Pipe)
 	defer cancel()
 	cfg.cfg.Propose(
 		ctx,
@@ -302,7 +302,7 @@ func (cfg *subConfig) Timeout(msg hotstuff.TimeoutMsg) {
 	}
 
 	// will wait until the second timeout before canceling
-	ctx, cancel := synchronizer.TimeoutContext(cfg.eventLoop.Context(), cfg.eventLoop)
+	ctx, cancel := synchronizer.ScopedTimeoutContext(cfg.eventLoop.Context(), cfg.eventLoop, msg.Pipe)
 	defer cancel()
 
 	cfg.cfg.Timeout(

backend/server.go

@@ -22,20 +22,21 @@ import (
 )
 
 // Server is the Server-side of the gorums backend.
-// It is responsible for calling handler methods on the consensus instance.
+// It is responsible for calling handler methods on the pipe.
 type Server struct {
 	blockChain    modules.BlockChain
 	configuration modules.Configuration
-	eventLoop     *eventloop.EventLoop
+	eventLoop     *eventloop.ScopedEventLoop
 	logger        logging.Logger
 	location      string
 	locationInfo  map[hotstuff.ID]string
 	latencyMatrix map[string]time.Duration
+	hackyLatency  time.Duration
 	gorumsSrv     *gorums.Server
 }
 
 // InitModule initializes the Server.
-func (srv *Server) InitModule(mods *modules.Core) {
+func (srv *Server) InitModule(mods *modules.Core, _ modules.ScopeInfo) {
 	mods.Get(
 		&srv.eventLoop,
 		&srv.configuration,
@@ -74,6 +75,26 @@ func (srv *Server) induceLatency(sender hotstuff.ID) {
 	<-timer1.C
 }
 
+func (srv *Server) SetHackyLatency(amount time.Duration) {
+	srv.hackyLatency = amount
+}
+
+// Alan: Hacky version of induceLatency which just blocks execution for an arbitrary time.
+func (srv *Server) induceLatencyHacky(sender hotstuff.ID, callback func()) {
+	if srv.hackyLatency == 0 {
+		callback()
+		return
+	}
+	go func() {
+		senderLatency := srv.hackyLatency
+		senderLocation := srv.locationInfo[sender]
+		srv.logger.Debugf("latency from server %s to server %s is %s\n", srv.location, senderLocation, senderLatency)
+		timer1 := time.NewTimer(senderLatency)
+		<-timer1.C
+		callback()
+	}()
+}
+
 // GetGorumsServer returns the underlying gorums Server.
 func (srv *Server) GetGorumsServer() *gorums.Server {
 	return srv.gorumsSrv
@@ -162,8 +183,14 @@ func (impl *serviceImpl) Propose(ctx gorums.ServerCtx, proposal *hotstuffpb.Prop
 	proposal.Block.Proposer = uint32(id)
 	proposeMsg := hotstuffpb.ProposalFromProto(proposal)
 	proposeMsg.ID = id
-	impl.srv.induceLatency(id)
-	impl.srv.eventLoop.AddEvent(proposeMsg)
+	impl.srv.induceLatencyHacky(id, func() {
+		if proposeMsg.Pipe.IsNull() {
+			impl.srv.eventLoop.AddScopedEvent(proposeMsg.Pipe, proposeMsg)
+			return
+		}
+
+		impl.srv.eventLoop.AddEvent(proposeMsg)
+	})
 }
 
 // Vote handles an incoming vote message.
@@ -173,10 +200,20 @@ func (impl *serviceImpl) Vote(ctx gorums.ServerCtx, cert *hotstuffpb.PartialCert
 		impl.srv.logger.Infof("Failed to get client ID: %v", err)
 		return
 	}
-	impl.srv.induceLatency(id)
-	impl.srv.eventLoop.AddEvent(hotstuff.VoteMsg{
-		ID:          id,
-		PartialCert: hotstuffpb.PartialCertFromProto(cert),
+	impl.srv.induceLatencyHacky(id, func() {
+		pipe := hotstuff.Pipe(cert.Pipe)
+		if pipe.IsNull() {
+			impl.srv.eventLoop.AddScopedEvent(pipe, hotstuff.VoteMsg{
+				ID:          id,
+				PartialCert: hotstuffpb.PartialCertFromProto(cert),
+			})
+			return
+		}
+
+		impl.srv.eventLoop.AddEvent(hotstuff.VoteMsg{
+			ID:          id,
+			PartialCert: hotstuffpb.PartialCertFromProto(cert),
+		})
 	})
 }
 
@@ -187,10 +224,20 @@ func (impl *serviceImpl) NewView(ctx gorums.ServerCtx, msg *hotstuffpb.SyncInfo)
 		impl.srv.logger.Infof("Failed to get client ID: %v", err)
 		return
 	}
-	impl.srv.induceLatency(id)
-	impl.srv.eventLoop.AddEvent(hotstuff.NewViewMsg{
-		ID:       id,
-		SyncInfo: hotstuffpb.SyncInfoFromProto(msg),
+	impl.srv.induceLatencyHacky(id, func() {
+		pipe := hotstuff.Pipe(msg.Pipe)
+		if pipe.IsNull() {
+			impl.srv.eventLoop.AddScopedEvent(pipe, hotstuff.NewViewMsg{
+				ID:       id,
+				SyncInfo: hotstuffpb.SyncInfoFromProto(msg),
+			})
+			return
+		}
+
+		impl.srv.eventLoop.AddEvent(hotstuff.NewViewMsg{
+			ID:       id,
+			SyncInfo: hotstuffpb.SyncInfoFromProto(msg),
+		})
 	})
 }
 
@@ -217,8 +264,15 @@ func (impl *serviceImpl) Timeout(ctx gorums.ServerCtx, msg *hotstuffpb.TimeoutMs
 	if err != nil {
 		impl.srv.logger.Infof("Could not get ID of replica: %v", err)
 	}
-	impl.srv.induceLatency(timeoutMsg.ID)
-	impl.srv.eventLoop.AddEvent(timeoutMsg)
+
+	impl.srv.induceLatencyHacky(timeoutMsg.ID, func() {
+		if timeoutMsg.Pipe.IsNull() {
+			impl.srv.eventLoop.AddScopedEvent(timeoutMsg.Pipe, timeoutMsg)
+			return
+		}
+
+		impl.srv.eventLoop.AddEvent(timeoutMsg)
+	})
 }
 
 type replicaConnected struct {