refactor(broadcast): removed async option

the async option is redundant in broadcast because running synchronously
can easily cause a deadlock

Author: aleksander-vedvik

broadcast.go

@@ -24,7 +24,6 @@ type broadcastServer struct {
 	timeout         time.Duration
 	clientReqs      *RequestMap
 	stopChan        chan struct{}
-	async           bool
 	logger          *slog.Logger
 }
 
@@ -43,6 +42,8 @@ func newBroadcastServer(logger *slog.Logger) *broadcastServer {
 	}
 }
 
+func (srv *broadcastServer) stop() {}
+
 func (srv *broadcastServer) alreadyBroadcasted(broadcastID string, method string) bool {
 	_, ok := srv.broadcastedMsgs[broadcastID]
 	if !ok {
@@ -57,8 +58,9 @@ func (srv *broadcastServer) alreadyBroadcasted(broadcastID string, method string
 
 func (srv *broadcastServer) run() {
 	for msg := range srv.broadcastChan {
-		if handler, ok := srv.handlers[msg.method]; ok {
-			handler(msg.ctx, msg.request, msg.metadata, msg.srvAddrs)
+		if broadcastCall, ok := srv.handlers[msg.method]; ok {
+			// it runs a interceptor prior to broadcastCall, hence a different signature.
+			broadcastCall(msg.ctx, msg.request, msg.metadata, msg.srvAddrs)
 		}
 		msg.setFinished()
 	}
@@ -106,11 +108,7 @@ func (srv *broadcastServer) handle(response *responseMsg) {
 	}
 }
 
-func (srv *broadcastServer) clientReturn(resp ResponseTypes, err error, metadata BroadcastMetadata) {
-	srv.returnToClient(metadata.BroadcastID, resp, err)
-}
-
-func (srv *broadcastServer) returnToClient(broadcastID string, resp ResponseTypes, err error) {
+func (srv *broadcastServer) sendToClient(broadcastID string, resp ResponseTypes, err error) {
 	if !srv.alreadyReturnedToClient(broadcastID) {
 		srv.responseChan <- newResponseMessage(resp, err, broadcastID)
 	}
@@ -121,12 +119,11 @@ func (srv *broadcastServer) alreadyReturnedToClient(broadcastID string) bool {
 }
 
 func (srv *broadcastServer) addClientRequest(metadata *ordering.Metadata, ctx ServerCtx, finished chan<- *Message) {
-	done := make(chan struct{})
 	srv.clientReqs.Add(metadata.BroadcastMsg.GetBroadcastID(), clientRequest{
 		id:       uuid.New().String(),
 		ctx:      ctx,
 		finished: finished,
 		metadata: metadata,
-		doneChan: done,
+		doneChan: make(chan struct{}),
 	})
 }

broadcastTypes.go

@@ -32,7 +32,7 @@ type ResponseTypes interface {
 }
 
 type BroadcastHandlerFunc func(method string, req RequestTypes, metadata BroadcastMetadata, data ...BroadcastOptions)
-type BroadcastReturnToClientHandlerFunc func(resp ResponseTypes, err error, metadata BroadcastMetadata)
+type BroadcastSendToClientHandlerFunc func(broadcastID string, resp ResponseTypes, err error)
 
 type defaultImplementationFunc[T RequestTypes, V ResponseTypes] func(ServerCtx, T) (V, error)
 
@@ -84,13 +84,23 @@ type clientRequest struct {
 	doneChan  chan struct{}
 }
 
-type SpBroadcast struct {
-	BroadcastHandler      BroadcastHandlerFunc
-	ReturnToClientHandler BroadcastReturnToClientHandlerFunc
+// The BroadcastOrchestrator is used as a container for all
+// broadcast handlers. The BroadcastHandler takes in a method
+// and schedules it for broadcasting. SendToClientHandler works
+// similarly but it sends the message to the calling client.
+//
+// It is necessary to use an orchestrator to hide certain
+// implementation details, such as internal methods on the
+// broadcaster struct. The BroadcastOrchestrator will thus
+// be an unimported field in the broadcaster struct in the
+// generated code.
+type BroadcastOrchestrator struct {
+	BroadcastHandler    BroadcastHandlerFunc
+	SendToClientHandler BroadcastSendToClientHandlerFunc
 }
 
-func NewSpBroadcastStruct() *SpBroadcast {
-	return &SpBroadcast{}
+func NewBroadcastOrchestrator() *BroadcastOrchestrator {
+	return &BroadcastOrchestrator{}
 }
 
 type BroadcastOption func(*BroadcastOptions)
@@ -132,12 +142,6 @@ func WithoutUniquenessChecks() BroadcastOption {
 	}
 }
 
-// not sure if this is necessary because the implementer
-// can decide to run the broadcast in a go routine.
-func WithoutWaiting() BroadcastOption {
-	return func(b *BroadcastOptions) {}
-}
-
 // returns a listener for the given address.
 // panics upon errors.
 func WithListener(listenAddr string) net.Listener {
@@ -162,22 +166,29 @@ func NewBroadcastOptions() BroadcastOptions {
 }
 
 type broadcaster interface {
-	setMetadataHandler(func(metadata BroadcastMetadata))
+	setMetadataHandler(func(metadata BroadcastMetadata), func())
 	setMetadata(metadata BroadcastMetadata)
+	resetMetadata()
 }
 
 type Broadcaster struct {
-	metadataHandler func(metadata BroadcastMetadata)
+	metadataHandler      func(metadata BroadcastMetadata)
+	resetMetadataHandler func()
 }
 
-func (b *Broadcaster) setMetadataHandler(handler func(metadata BroadcastMetadata)) {
+func (b *Broadcaster) setMetadataHandler(handler func(metadata BroadcastMetadata), resetHandler func()) {
 	b.metadataHandler = handler
+	b.resetMetadataHandler = resetHandler
 }
 
 func (b *Broadcaster) setMetadata(metadata BroadcastMetadata) {
 	b.metadataHandler(metadata)
 }
 
+func (b *Broadcaster) resetMetadata() {
+	b.resetMetadataHandler()
+}
+
 func NewBroadcaster() *Broadcaster {
 	return &Broadcaster{}
 }

cmd/protoc-gen-gorums/dev/mgr.go

@@ -37,7 +37,7 @@ func NewManager(opts ...gorums.ManagerOption) (mgr *Manager) {
 // A new configuration can also be created from an existing configuration,
 // using the And, WithNewNodes, Except, and WithoutNodes methods.
 func (m *Manager) NewConfiguration(opts ...gorums.ConfigOption) (c *Configuration, err error) {
-	if len(opts) < 1 || len(opts) > 2 {
+	if len(opts) < 1 || len(opts) > 3 {
 		return nil, fmt.Errorf("wrong number of options: %d", len(opts))
 	}
 	c = &Configuration{}
@@ -62,10 +62,10 @@ func (m *Manager) NewConfiguration(opts ...gorums.ConfigOption) (c *Configuratio
 		}
 	}
 	// return an error if the QuorumSpec interface is not empty and no implementation was provided.
-	var test interface{} = struct{}{}
-	if _, empty := test.(QuorumSpec); !empty && c.qspec == nil {
-		return nil, fmt.Errorf("missing required QuorumSpec")
-	}
+	//var test interface{} = struct{}{}
+	//if _, empty := test.(QuorumSpec); !empty && c.qspec == nil {
+	//	return nil, fmt.Errorf("missing required QuorumSpec")
+	//}
 	return c, nil
 }
 

cmd/protoc-gen-gorums/dev/server.go

@@ -5,22 +5,27 @@ import (
 
 	"github.com/relab/gorums"
 	grpc "google.golang.org/grpc"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
 )
 
 type Server struct {
 	*gorums.Server
-	View *Configuration
+	broadcast *Broadcast
+	View      *Configuration
 }
 
 func NewServer() *Server {
 	srv := &Server{
 		Server: gorums.NewServer(),
 	}
 	b := &Broadcast{
-		Broadcaster: gorums.NewBroadcaster(),
-		sp:          gorums.NewSpBroadcastStruct(),
+		Broadcaster:  gorums.NewBroadcaster(),
+		orchestrator: gorums.NewBroadcastOrchestrator(),
+		metadata:     gorums.BroadcastMetadata{},
 	}
-	srv.RegisterBroadcastStruct(b, configureHandlers(b), configureMetadata(b))
+	srv.broadcast = b
+	set, reset := configureMetadata(b)
+	srv.RegisterBroadcaster(b, configureHandlers(b), set, reset)
 	return srv
 }
 
@@ -32,21 +37,23 @@ func (srv *Server) SetView(config *Configuration) {
 
 type Broadcast struct {
 	*gorums.Broadcaster
-	sp       *gorums.SpBroadcast
-	metadata gorums.BroadcastMetadata
+	orchestrator *gorums.BroadcastOrchestrator
+	metadata     gorums.BroadcastMetadata
 }
 
-func configureHandlers(b *Broadcast) func(bh gorums.BroadcastHandlerFunc, ch gorums.BroadcastReturnToClientHandlerFunc) {
-	return func(bh gorums.BroadcastHandlerFunc, ch gorums.BroadcastReturnToClientHandlerFunc) {
-		b.sp.BroadcastHandler = bh
-		b.sp.ReturnToClientHandler = ch
+func configureHandlers(b *Broadcast) func(bh gorums.BroadcastHandlerFunc, ch gorums.BroadcastSendToClientHandlerFunc) {
+	return func(bh gorums.BroadcastHandlerFunc, ch gorums.BroadcastSendToClientHandlerFunc) {
+		b.orchestrator.BroadcastHandler = bh
+		b.orchestrator.SendToClientHandler = ch
 	}
 }
 
-func configureMetadata(b *Broadcast) func(metadata gorums.BroadcastMetadata) {
+func configureMetadata(b *Broadcast) (func(metadata gorums.BroadcastMetadata), func()) {
 	return func(metadata gorums.BroadcastMetadata) {
-		b.metadata = metadata
-	}
+			b.metadata = metadata
+		}, func() {
+			b.metadata = gorums.BroadcastMetadata{}
+		}
 }
 
 // Returns a readonly struct of the metadata used in the broadcast.
@@ -76,3 +83,11 @@ func (c *Configuration) RegisterClientServer(lis net.Listener, opts ...grpc.Serv
 	c.srv = srvImpl
 	return nil
 }
+
+func (b *Broadcast) SendToClient(resp protoreflect.ProtoMessage, err error) {
+	b.orchestrator.SendToClientHandler(b.metadata.BroadcastID, resp, err)
+}
+
+func (srv *Server) SendToClient(resp protoreflect.ProtoMessage, err error, broadcastID string) {
+	srv.RetToClient(resp, err, broadcastID)
+}