endpoint: concurrent queries

client/holds.go

@@ -33,6 +33,7 @@ type holdsFilterFlags struct {
 	Filesystems FilesystemsFilterFlag
 	Job         JobIDFlag
 	Types       AbstractionTypesFlag
+	Concurrency int64
 }
 
 // produce a query from the CLI flags
@@ -42,6 +43,7 @@ func (f holdsFilterFlags) Query() (endpoint.ListZFSHoldsAndBookmarksQuery, error
 		What:        f.Types.FlagValue(),
 		JobID:       f.Job.FlagValue(),
 		Until:       nil, // TODO support this as a flag
+		Concurrency: f.Concurrency,
 	}
 	return q, q.Validate()
 }
@@ -58,6 +60,8 @@ func (f *holdsFilterFlags) registerHoldsFilterFlags(s *pflag.FlagSet, verb strin
 	variants = sort.StringSlice(variants)
 	variantsJoined := strings.Join(variants, "|")
 	s.Var(&f.Types, "type", fmt.Sprintf("only %s holds of the specified type [default: all] [comma-separated list of %s]", verb, variantsJoined))
+
+	s.Int64VarP(&f.Concurrency, "concurrency", "p", 1, "number of concurrently queried filesystems")
 }
 
 type JobIDFlag struct{ J *endpoint.JobID }
@@ -106,7 +110,7 @@ type FilesystemsFilterFlag struct {
 
 func (flag *FilesystemsFilterFlag) Set(s string) error {
 	mappings := strings.Split(s, ",")
-	if len(mappings) == 1 {
+	if len(mappings) == 1 && !strings.Contains(mappings[0], ":") {
 		flag.F = endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter{
 			FS: &mappings[0],
 		}

client/holds_list.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"sync"
 
 	"github.com/fatih/color"
 	"github.com/pkg/errors"
@@ -12,6 +13,7 @@ import (
 
 	"github.com/zrepl/zrepl/cli"
 	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/util/chainlock"
 )
 
 var holdsListFlags struct {
@@ -43,30 +45,56 @@ func doHoldsList(sc *cli.Subcommand, args []string) error {
 		return errors.Wrap(err, "invalid filter specification on command line")
 	}
 
-	abstractions, errors, err := endpoint.ListAbstractions(ctx, q)
+	abstractions, errors, err := endpoint.ListAbstractionsStreamed(ctx, q)
 	if err != nil {
 		return err // context clear by invocation of command
 	}
 
-	// always print what we got
+	var line chainlock.L
-	if holdsListFlags.Json {
+	var wg sync.WaitGroup
+	defer wg.Wait()
+	wg.Add(1)
+
+	// print results
+	go func() {
+		defer wg.Done()
 		enc := json.NewEncoder(os.Stdout)
+		for a := range abstractions {
+			func() {
+				defer line.Lock().Unlock()
+				if holdsListFlags.Json {
 					enc.SetIndent("", "  ")
 					if err := enc.Encode(abstractions); err != nil {
 						panic(err)
 					}
 					fmt.Println()
 				} else {
-		for _, a := range abstractions {
 					fmt.Println(a)
 				}
+			}()
 		}
+	}()
 
-	// then potential errors, so that users always see them
+	// print errors to stderr
-	if len(errors) > 0 {
+	errorColor := color.New(color.FgRed)
-		color.New(color.FgRed).Fprintf(os.Stderr, "there were errors in listing the abstractions:\n%s\n", errors)
+	var errorsSlice []endpoint.ListAbstractionsError
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for err := range errors {
+			func() {
+				defer line.Lock().Unlock()
+				errorsSlice = append(errorsSlice, err)
+				errorColor.Fprintf(os.Stderr, "%s\n", err)
+			}()
+		}
+	}()
+	wg.Wait()
+	if len(errorsSlice) > 0 {
+		errorColor.Add(color.Bold).Fprintf(os.Stderr, "there were errors in listing the abstractions")
 		return fmt.Errorf("")
 	} else {
 		return nil
 	}
+
 }

endpoint/endpoint_zfs_abstraction.go

@@ -6,10 +6,12 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"sync"
 
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/util/envconst"
+	"github.com/zrepl/zrepl/util/semaphore"
 	"github.com/zrepl/zrepl/zfs"
 )
 
@@ -207,8 +209,8 @@ type ListZFSHoldsAndBookmarksQuery struct {
 	// else: CreateTXG of the hold or bookmark can be any value
 	Until *InclusiveExclusiveCreateTXG
 
-	// TODO
+	// Number of concurrently queried filesystems. Must be >= 1
-	// Concurrent: uint > 0
+	Concurrency int64
 }
 
 type InclusiveExclusiveCreateTXG struct {
@@ -237,6 +239,9 @@ func (q *ListZFSHoldsAndBookmarksQuery) Validate() error {
 	if err := q.What.Validate(); err != nil {
 		return err
 	}
+	if q.Concurrency < 1 {
+		return errors.New("Concurrency must be >= 1")
+	}
 	return nil
 }
 
@@ -330,6 +335,33 @@ func (e ListAbstractionsErrors) Error() string {
 }
 
 func ListAbstractions(ctx context.Context, query ListZFSHoldsAndBookmarksQuery) (out []Abstraction, outErrs []ListAbstractionsError, err error) {
+	outChan, outErrsChan, err := ListAbstractionsStreamed(ctx, query)
+	if err != nil {
+		return nil, nil, err
+	}
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for a := range outChan {
+			out = append(out, a)
+		}
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for err := range outErrsChan {
+			outErrs = append(outErrs, err)
+		}
+	}()
+	wg.Wait()
+	return out, outErrs, nil
+}
+
+// if err != nil, the returned channels are both nil
+// if err == nil, both channels must be fully drained by the caller to avoid leaking goroutines
+func ListAbstractionsStreamed(ctx context.Context, query ListZFSHoldsAndBookmarksQuery) (<-chan Abstraction, <-chan ListAbstractionsError, error) {
+
 	// impl note: structure the query processing in such a way that
 	// a minimum amount of zfs shell-outs needs to be done
 
@@ -342,8 +374,11 @@ func ListAbstractions(ctx context.Context, query ListZFSHoldsAndBookmarksQuery)
 		return nil, nil, errors.Wrap(err, "list filesystems")
 	}
 
+	outErrs := make(chan ListAbstractionsError)
+	out := make(chan Abstraction)
+
 	errCb := func(err error, fs string, what string) {
-		outErrs = append(outErrs, ListAbstractionsError{Err: err, FS: fs, What: what})
+		outErrs <- ListAbstractionsError{Err: err, FS: fs, What: what}
 	}
 	emitAbstraction := func(a Abstraction) {
 		jobIdMatches := query.JobID == nil || a.GetJobID() == nil || *a.GetJobID() == *query.JobID
@@ -358,15 +393,34 @@ func ListAbstractions(ctx context.Context, query ListZFSHoldsAndBookmarksQuery)
 		}
 
 		if jobIdMatches && untilMatches {
-			out = append(out, a)
+			out <- a
 		}
 	}
-	for _, fs := range fss {
+
-		listAbstractionsImplFS(ctx, fs, &query, emitAbstraction, errCb)
+	sem := semaphore.New(int64(query.Concurrency))
+	go func() {
+		defer close(out)
+		defer close(outErrs)
+		var wg sync.WaitGroup
+		defer wg.Wait()
+		for i := range fss {
+			wg.Add(1)
+			go func(i int) {
+				defer wg.Done()
+				g, err := sem.Acquire(ctx)
+				if err != nil {
+					errCb(err, fss[i], err.Error())
+					return
 				}
+				func() {
+					defer g.Release()
+					listAbstractionsImplFS(ctx, fss[i], &query, emitAbstraction, errCb)
+				}()
+			}(i)
+		}
+	}()
 
 	return out, outErrs, nil
-
 }
 
 func listAbstractionsImplFS(ctx context.Context, fs string, query *ListZFSHoldsAndBookmarksQuery, emitCandidate putListAbstraction, errCb putListAbstractionErr) {

endpoint/endpoint_zfs_abstraction_cursor_and_last_received_hold.go

@@ -84,6 +84,7 @@ func GetReplicationCursors(ctx context.Context, dp *zfs.DatasetPath, jobID JobID
 		},
 		JobID:       &jobID,
 		Until:       nil,
+		Concurrency: 1,
 	}
 	abs, absErr, err := ListAbstractions(ctx, q)
 	if err != nil {
@@ -176,6 +177,7 @@ func DestroyObsoleteReplicationCursors(ctx context.Context, fs string, current R
 			CreateTXG: current.GetCreateTXG(),
 			Inclusive: &zfs.NilBool{B: false},
 		},
+		Concurrency: 1,
 	}
 	abs, absErr, err := ListAbstractions(ctx, q)
 	if err != nil {
@@ -264,6 +266,7 @@ func MoveLastReceivedHold(ctx context.Context, fs string, to zfs.FilesystemVersi
 			CreateTXG: to.GetCreateTXG(),
 			Inclusive: &zfs.NilBool{B: false},
 		},
+		Concurrency: 1,
 	}
 	abs, absErrs, err := ListAbstractions(ctx, q)
 	if err != nil {

endpoint/endpoint_zfs_abstraction_step.go

@@ -154,6 +154,7 @@ func ReleaseStepCummulativeInclusive(ctx context.Context, fs string, mostRecent
 			CreateTXG: mostRecent.CreateTXG,
 			Inclusive: &zfs.NilBool{B: true},
 		},
+		Concurrency: 1,
 	}
 	abs, absErrs, err := ListAbstractions(ctx, q)
 	if err != nil {
@@ -195,6 +196,7 @@ func TryReleaseStepStaleFS(ctx context.Context, fs string, jobID JobID) {
 			AbstractionStepHold:     true,
 			AbstractionStepBookmark: true,
 		},
+		Concurrency: 1,
 	}
 	staleness, err := ListStale(ctx, q)
 	if err != nil {