Use iterators in place of QuorumSpec quorum functions.

async.go

@@ -1,31 +1,26 @@
 package gorums
 
-import (
-	"context"
-
-	"github.com/relab/gorums/ordering"
-	"google.golang.org/protobuf/reflect/protoreflect"
-)
+import "google.golang.org/protobuf/proto"
 
 // Async encapsulates the state of an asynchronous quorum call,
 // and has methods for checking the status of the call or waiting for it to complete.
 //
 // This struct should only be used by generated code.
-type Async struct {
-	reply protoreflect.ProtoMessage
+type Async[resultType any] struct {
+	reply resultType
 	err   error
 	c     chan struct{}
 }
 
 // Get returns the reply and any error associated with the called method.
 // The method blocks until a reply or error is available.
-func (f *Async) Get() (protoreflect.ProtoMessage, error) {
+func (f *Async[resultType]) Get() (resultType, error) {
 	<-f.c
 	return f.reply, f.err
 }
 
 // Done reports if a reply and/or error is available for the called method.
-func (f *Async) Done() bool {
+func (f *Async[resultType]) Done() bool {
 	select {
 	case <-f.c:
 		return true
@@ -34,74 +29,19 @@ func (f *Async) Done() bool {
 	}
 }
 
-type asyncCallState struct {
-	md              *ordering.Metadata
-	data            QuorumCallData
-	replyChan       <-chan response
-	expectedReplies int
-}
-
-// AsyncCall starts an asynchronous quorum call, returning an Async object that can be used to retrieve the results.
-//
-// This function should only be used by generated code.
-func (c RawConfiguration) AsyncCall(ctx context.Context, d QuorumCallData) *Async {
-	expectedReplies := len(c)
-	md := ordering.NewGorumsMetadata(ctx, c.getMsgID(), d.Method)
-	replyChan := make(chan response, expectedReplies)
-
-	for _, n := range c {
-		msg := d.Message
-		if d.PerNodeArgFn != nil {
-			msg = d.PerNodeArgFn(d.Message, n.id)
-			if !msg.ProtoReflect().IsValid() {
-				expectedReplies--
-				continue // don't send if no msg
-			}
-		}
-		n.channel.enqueue(request{ctx: ctx, msg: &Message{Metadata: md, Message: msg}}, replyChan, false)
+// NewAsync lets you use a quorum call as an async call.
+func NewAsync[responseType proto.Message, resultType any](
+	iter Responses[responseType],
+	asyncFunc func(Responses[responseType]) resultType,
+) Async[resultType] {
+	async := Async[resultType]{
+		c: make(chan struct{}),
 	}
 
-	fut := &Async{c: make(chan struct{}, 1)}
-
-	go c.handleAsyncCall(ctx, fut, asyncCallState{
-		md:              md,
-		data:            d,
-		replyChan:       replyChan,
-		expectedReplies: expectedReplies,
-	})
+	go func() {
+		async.reply = asyncFunc(iter)
+		close(async.c)
+	}()
 
-	return fut
-}
-
-func (c RawConfiguration) handleAsyncCall(ctx context.Context, fut *Async, state asyncCallState) {
-	defer close(fut.c)
-
-	var (
-		resp    protoreflect.ProtoMessage
-		errs    []nodeError
-		quorum  bool
-		replies = make(map[uint32]protoreflect.ProtoMessage)
-	)
-
-	for {
-		select {
-		case r := <-state.replyChan:
-			if r.err != nil {
-				errs = append(errs, nodeError{nodeID: r.nid, cause: r.err})
-				break
-			}
-			replies[r.nid] = r.msg
-			if resp, quorum = state.data.QuorumFunction(state.data.Message, replies); quorum {
-				fut.reply, fut.err = resp, nil
-				return
-			}
-		case <-ctx.Done():
-			fut.reply, fut.err = resp, QuorumCallError{cause: ctx.Err(), errors: errs, replies: len(replies)}
-			return
-		}
-		if len(errs)+len(replies) == state.expectedReplies {
-			fut.reply, fut.err = resp, QuorumCallError{cause: Incomplete, errors: errs, replies: len(replies)}
-			return
-		}
-	}
+	return async
 }

benchmark/benchmark.go

@@ -8,6 +8,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	gorums "github.com/relab/gorums"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -31,13 +32,12 @@ type Bench struct {
 }
 
 type (
-	benchFunc   func(Options) (*Result, error)
-	qcFunc      func(context.Context, *Echo) (*Echo, error)
-	asyncQCFunc func(context.Context, *Echo) *AsyncEcho
-	serverFunc  func(context.Context, *TimedMsg)
+	benchFunc  func(Options) (*Result, error)
+	qcFunc     func(context.Context, *Echo) gorums.Responses[*Echo]
+	serverFunc func(context.Context, *TimedMsg)
 )
 
-func runQCBenchmark(opts Options, cfg *Configuration, f qcFunc) (*Result, error) {
+func runQCBenchmark(opts Options, cfg *Configuration, quorum int, f qcFunc) (*Result, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	msg := Echo_builder{Payload: make([]byte, opts.Payload)}.Build()
@@ -48,7 +48,7 @@ func runQCBenchmark(opts Options, cfg *Configuration, f qcFunc) (*Result, error)
 		g.Go(func() error {
 			warmupEnd := time.Now().Add(opts.Warmup)
 			for !time.Now().After(warmupEnd) {
-				_, err := f(ctx, msg)
+				err := qf(f(ctx, msg), quorum)
 				if err != nil {
 					return err
 				}
@@ -63,7 +63,7 @@ func runQCBenchmark(opts Options, cfg *Configuration, f qcFunc) (*Result, error)
 	}
 
 	if opts.Remote {
-		_, err := cfg.StartBenchmark(ctx, &StartRequest{})
+		err := qf(cfg.StartBenchmark(ctx, &StartRequest{}), quorum)
 		if err != nil {
 			return nil, err
 		}
@@ -75,7 +75,7 @@ func runQCBenchmark(opts Options, cfg *Configuration, f qcFunc) (*Result, error)
 			endTime := time.Now().Add(opts.Duration)
 			for !time.Now().After(endTime) {
 				start := time.Now()
-				_, err := f(ctx, msg)
+				err := qf(f(ctx, msg), quorum)
 				if err != nil {
 					return err
 				}
@@ -93,17 +93,17 @@ func runQCBenchmark(opts Options, cfg *Configuration, f qcFunc) (*Result, error)
 
 	result := s.GetResult()
 	if opts.Remote {
-		memStats, err := cfg.StopBenchmark(ctx, &StopRequest{})
+		memStats, err := stopBenchmarkQF(cfg.StopBenchmark(ctx, &StopRequest{}), cfg.Size())
 		if err != nil {
 			return nil, err
 		}
-		result.SetServerStats(memStats.GetMemoryStats())
+		result.SetServerStats(memStats)
 	}
 
 	return result, nil
 }
 
-func runAsyncQCBenchmark(opts Options, cfg *Configuration, f asyncQCFunc) (*Result, error) {
+func runAsyncQCBenchmark(opts Options, cfg *Configuration, quorum int) (*Result, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	msg := Echo_builder{Payload: make([]byte, opts.Payload)}.Build()
@@ -116,9 +116,9 @@ func runAsyncQCBenchmark(opts Options, cfg *Configuration, f asyncQCFunc) (*Resu
 	var warmupFunc func() error
 	warmupFunc = func() error {
 		for ; !time.Now().After(warmupEnd) && atomic.LoadUint64(&async) < uint64(opts.MaxAsync); atomic.AddUint64(&async, 1) {
-			fut := f(ctx, msg)
+			responses := cfg.AsyncQuorumCall(ctx, msg)
 			g.Go(func() error {
-				_, err := fut.Get()
+				err := qf(responses, quorum)
 				if err != nil {
 					return err
 				}
@@ -139,7 +139,7 @@ func runAsyncQCBenchmark(opts Options, cfg *Configuration, f asyncQCFunc) (*Resu
 	}
 
 	if opts.Remote {
-		_, err := cfg.StartBenchmark(ctx, &StartRequest{})
+		err := qf(cfg.StartBenchmark(ctx, &StartRequest{}), cfg.Size())
 		if err != nil {
 			return nil, err
 		}
@@ -150,9 +150,9 @@ func runAsyncQCBenchmark(opts Options, cfg *Configuration, f asyncQCFunc) (*Resu
 	benchmarkFunc = func() error {
 		for ; !time.Now().After(endTime) && atomic.LoadUint64(&async) < uint64(opts.MaxAsync); atomic.AddUint64(&async, 1) {
 			start := time.Now()
-			fut := f(ctx, msg)
+			responses := cfg.AsyncQuorumCall(ctx, msg)
 			g.Go(func() error {
-				_, err := fut.Get()
+				err := qf(responses, quorum)
 				if err != nil {
 					return err
 				}
@@ -177,11 +177,11 @@ func runAsyncQCBenchmark(opts Options, cfg *Configuration, f asyncQCFunc) (*Resu
 
 	result := s.GetResult()
 	if opts.Remote {
-		memStats, err := cfg.StopBenchmark(ctx, &StopRequest{})
+		memStats, err := stopBenchmarkQF(cfg.StopBenchmark(ctx, &StopRequest{}), cfg.Size())
 		if err != nil {
 			return nil, err
 		}
-		result.SetServerStats(memStats.GetMemoryStats())
+		result.SetServerStats(memStats)
 	}
 
 	return result, nil
@@ -211,7 +211,7 @@ func runServerBenchmark(opts Options, cfg *Configuration, f serverFunc) (*Result
 		return nil, err
 	}
 
-	_, err = cfg.StartServerBenchmark(ctx, &StartRequest{})
+	err = qf(cfg.StartServerBenchmark(ctx, &StartRequest{}), cfg.Size())
 	if err != nil {
 		return nil, err
 	}
@@ -227,7 +227,7 @@ func runServerBenchmark(opts Options, cfg *Configuration, f serverFunc) (*Result
 	}
 	runtime.ReadMemStats(&end)
 
-	resp, err := cfg.StopServerBenchmark(ctx, &StopRequest{})
+	resp, err := stopServerBenchmarkQF(cfg.StopServerBenchmark(ctx, &StopRequest{}), cfg.Size())
 	if err != nil {
 		return nil, err
 	}
@@ -241,22 +241,22 @@ func runServerBenchmark(opts Options, cfg *Configuration, f serverFunc) (*Result
 }
 
 // GetBenchmarks returns a list of Benchmarks that can be performed on the configuration
-func GetBenchmarks(cfg *Configuration) []Bench {
+func GetBenchmarks(cfg *Configuration, quorum int) []Bench {
 	m := []Bench{
 		{
 			Name:        "QuorumCall",
 			Description: "NodeStream based quorum call implementation with FIFO ordering",
-			runBench:    func(opts Options) (*Result, error) { return runQCBenchmark(opts, cfg, cfg.QuorumCall) },
+			runBench:    func(opts Options) (*Result, error) { return runQCBenchmark(opts, cfg, quorum, cfg.QuorumCall) },
 		},
 		{
 			Name:        "AsyncQuorumCall",
 			Description: "NodeStream based async quorum call implementation with FIFO ordering",
-			runBench:    func(opts Options) (*Result, error) { return runAsyncQCBenchmark(opts, cfg, cfg.AsyncQuorumCall) },
+			runBench:    func(opts Options) (*Result, error) { return runAsyncQCBenchmark(opts, cfg, quorum) },
 		},
 		{
 			Name:        "SlowServer",
 			Description: "Quorum Call with a 10s processing time on the server",
-			runBench:    func(opts Options) (*Result, error) { return runQCBenchmark(opts, cfg, cfg.SlowServer) },
+			runBench:    func(opts Options) (*Result, error) { return runQCBenchmark(opts, cfg, quorum, cfg.SlowServer) },
 		},
 		{
 			Name:        "Multicast",
@@ -270,8 +270,8 @@ func GetBenchmarks(cfg *Configuration) []Bench {
 }
 
 // RunBenchmarks runs all the benchmarks that match the given regex with the given options
-func RunBenchmarks(benchRegex *regexp.Regexp, options Options, cfg *Configuration) ([]*Result, error) {
-	benchmarks := GetBenchmarks(cfg)
+func RunBenchmarks(benchRegex *regexp.Regexp, options Options, cfg *Configuration, quorum int) ([]*Result, error) {
+	benchmarks := GetBenchmarks(cfg, quorum)
 	var results []*Result
 	for _, b := range benchmarks {
 		if benchRegex.MatchString(b.Name) {