func doDestroyBatchedRec(ctx context.Context, fsbatch []*DestroySnapOp, d destroyer) {
	if len(fsbatch) <= 1 {
		doDestroySeq(ctx, fsbatch, d)
		return
	}
	err := tryBatch(ctx, fsbatch, d)
	if err == nil {
		setDestroySnapOpErr(fsbatch, nil)
		return
	}
	if pe, ok := err.(*os.PathError); ok && pe.Err == syscall.E2BIG {
		debug("batch destroy: E2BIG encountered: %s", err)
		doDestroyBatchedRec(ctx, fsbatch[0:len(fsbatch)/2], d)
		doDestroyBatchedRec(ctx, fsbatch[len(fsbatch)/2:], d)
		return
	}
	singleRun := fsbatch  
	if err, ok := err.(*DestroySnapshotsError); ok {
		strippedBatch, remaining := make([]*DestroySnapOp, 0, len(fsbatch)), make([]*DestroySnapOp, 0, len(fsbatch))
		for _, b := range fsbatch {
			isUndestroyable := false
			for _, undestroyable := range err.Undestroyable {
				if undestroyable == b.Name {
					isUndestroyable = true
					break
				}
			}
			if isUndestroyable {
				remaining = append(remaining, b)
			} else {
				strippedBatch = append(strippedBatch, b)
			}
		}
		err := tryBatch(ctx, strippedBatch, d)
		if err != nil {
			singleRun = fsbatch  
		} else {
			setDestroySnapOpErr(strippedBatch, nil)  
			singleRun = remaining                    
		}
	}
	doDestroySeq(ctx, singleRun, d)
}
fsbatch must be on same filesystem
package zfs
import (
	"context"
	"fmt"
	"os"
	"os/exec"
	"sort"
	"strings"
	"sync"
	"syscall"
	"github.com/zrepl/zrepl/util/envconst"
	"github.com/zrepl/zrepl/zfs/zfscmd"
)
func ZFSDestroyFilesystemVersion(ctx context.Context, filesystem *DatasetPath, version *FilesystemVersion) (err error) {
	datasetPath := version.ToAbsPath(filesystem)
	if !strings.ContainsAny(datasetPath, "@#") {
		return fmt.Errorf("sanity check failed: no @ or # character found in %q", datasetPath)
	}
	return ZFSDestroy(ctx, datasetPath)
}
var destroyerSingleton = destroyerImpl{}
type DestroySnapOp struct {
	Filesystem string
	Name       string
	ErrOut     *error
}
func (o *DestroySnapOp) String() string {
	return fmt.Sprintf("destroy operation %s@%s", o.Filesystem, o.Name)
}
func ZFSDestroyFilesystemVersions(ctx context.Context, reqs []*DestroySnapOp) {
	doDestroy(ctx, reqs, destroyerSingleton)
}
func setDestroySnapOpErr(b []*DestroySnapOp, err error) {
	for _, r := range b {
		*r.ErrOut = err
	}
}
type destroyer interface {
	Destroy(ctx context.Context, args []string) error
	DestroySnapshotsCommaSyntaxSupported(context.Context) (bool, error)
}
func doDestroy(ctx context.Context, reqs []*DestroySnapOp, e destroyer) {
	var validated []*DestroySnapOp
	for _, req := range reqs {
		if req.Filesystem == "" {
			*req.ErrOut = fmt.Errorf("Filesystem must not be an empty string")
		} else if req.Name == "" {
			*req.ErrOut = fmt.Errorf("Name must not be an empty string")
		} else {
			validated = append(validated, req)
		}
	}
	reqs = validated
	commaSupported, err := e.DestroySnapshotsCommaSyntaxSupported(ctx)
	if err != nil {
		debug("destroy: comma syntax support detection failed: %s", err)
		setDestroySnapOpErr(reqs, err)
		return
	}
	if !commaSupported {
		doDestroySeq(ctx, reqs, e)
	} else {
		doDestroyBatched(ctx, reqs, e)
	}
}
func doDestroySeq(ctx context.Context, reqs []*DestroySnapOp, e destroyer) {
	for _, r := range reqs {
		*r.ErrOut = e.Destroy(ctx, []string{fmt.Sprintf("%s@%s", r.Filesystem, r.Name)})
	}
}
func doDestroyBatched(ctx context.Context, reqs []*DestroySnapOp, d destroyer) {
	perFS := buildBatches(reqs)
	for _, fsbatch := range perFS {
		doDestroyBatchedRec(ctx, fsbatch, d)
	}
}
func buildBatches(reqs []*DestroySnapOp) [][]*DestroySnapOp {
	if len(reqs) == 0 {
		return nil
	}
	sorted := make([]*DestroySnapOp, len(reqs))
	copy(sorted, reqs)
	sort.SliceStable(sorted, func(i, j int) bool {
		fscmp := strings.Compare(sorted[i].Filesystem, sorted[j].Filesystem)
		if fscmp != 0 {
			return fscmp == -1
		}
		return strings.Compare(sorted[i].Name, sorted[j].Name) == -1
	})
	var perFS [][]*DestroySnapOp
	consumed := 0
	for consumed < len(sorted) {
		batchConsumedUntil := consumed
		for ; batchConsumedUntil < len(sorted) && sorted[batchConsumedUntil].Filesystem == sorted[consumed].Filesystem; batchConsumedUntil++ {
		}
		perFS = append(perFS, sorted[consumed:batchConsumedUntil])
		consumed = batchConsumedUntil
	}
	return perFS
}
func tryBatch(ctx context.Context, batch []*DestroySnapOp, d destroyer) error {
	if len(batch) == 0 {
		return nil
	}
	batchFS := batch[0].Filesystem
	batchNames := make([]string, len(batch))
	for i := range batchNames {
		batchNames[i] = batch[i].Name
		if batchFS != batch[i].Filesystem {
			panic("inconsistent batch")
		}
	}
	batchArg := fmt.Sprintf("%s@%s", batchFS, strings.Join(batchNames, ","))
	return d.Destroy(ctx, []string{batchArg})
}