WIP runtime-controllable concurrency for replication

Changes done so far: - signal to route concurrency request up to the stepQueue - pretty hacky, no status reporting yet - stepQueue upsizing (confirmed that it works, no intermediary commit) Stuck at: stepQueue downsizing - idea was to have the stepQueue signal to the activated step that it should suspend - ideally, we'd just kill everything associated with the step and track it as restartable - the driver model doesn't allow for that though within an attempt - would need to start a separate run for the step - less perfect: tell the downsized steps to stop copying, but leave all zfs sends + rpc conns open - - doesn't give back the resoures that a step aquired before being selected as downsize victim (open zfs processe + TCP conn, some memory in the pipe) - - looks weird to user if the ps aux - + re-waking a step is easy: just tell it to proceed with copying - (the impl would likely pass a check function down into Step.do and have it check that functino periodically. the suspend should be acknowledged, and stepQueue should only remove the step from the active queue _after_ that step has acknowledged that is suspended)
2020-02-17 17:22:52 +01:00
96 changed files with 612 additions and 1427 deletions
@@ -6,7 +6,6 @@ workflows:
      - build-1.11
      - build-1.12
      - build-1.13
-      - build-1.14
      - build-latest
      - test-build-in-docker
 jobs:
@@ -112,11 +111,6 @@ jobs:
    docker:
    - image: circleci/golang:1.13

-  build-1.14:
-    <<: *build-latest
-    docker:
-    - image: circleci/golang:1.14
-
  # this job tries to mimic the build-in-docker instructions
  # given in docs/installation.rst
  #
@@ -1,9 +1,8 @@
 [![GitHub license](https://img.shields.io/github/license/zrepl/zrepl.svg)](https://github.com/zrepl/zrepl/blob/master/LICENSE)
 [![Language: Go](https://img.shields.io/badge/language-Go-6ad7e5.svg)](https://golang.org/)
 [![User Docs](https://img.shields.io/badge/docs-web-blue.svg)](https://zrepl.github.io)
-[![Donate via Patreon](https://img.shields.io/endpoint.svg?url=https%3A%2F%2Fshieldsio-patreon.herokuapp.com%2Fzrepl%2Fpledges&style=flat&color=yellow)](https://www.patreon.com/zrepl)
-[![Donate via Liberapay](https://img.shields.io/liberapay/receives/zrepl.svg?logo=liberapay)](https://liberapay.com/zrepl/donate)
 [![Donate via PayPal](https://img.shields.io/badge/donate-paypal-yellow.svg)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96)
+[![Donate via Liberapay](https://img.shields.io/liberapay/receives/zrepl.svg?logo=liberapay)](https://liberapay.com/zrepl/donate)
 [![Twitter](https://img.shields.io/twitter/url/https/github.com/zrepl/zrepl.svg?style=social)](https://twitter.com/intent/tweet?text=Wow:&url=https%3A%2F%2Fgithub.com%2Fzrepl%2Fzrepl)

 # zrepl
@@ -24,7 +23,6 @@ zrepl is a one-stop ZFS backup & replication solution.
   If so, think of an expressive configuration example.
 2. Think of at least one use case that generalizes from your concrete application.
 3. Open an issue on GitHub with example conf & use case attached.
-4. **Optional**: [Post a bounty](https://www.bountysource.com/teams/zrepl) on the issue, or [contact Christian Schwarz](https://cschwarz.com) for contract work.

 The above does not apply if you already implemented everything.
 Check out the *Coding Workflow* section below for details.
@@ -48,8 +46,11 @@ zrepl is written in [Go](https://golang.org) and uses [Go modules](https://githu
 The documentation is written in [ReStructured Text](http://docutils.sourceforge.net/rst.html) using the [Sphinx](https://www.sphinx-doc.org) framework.

 To get started, run `./lazy.sh devsetup` to easily install build dependencies and read `docs/installation.rst -> Compiling from Source`.
-`lazy.sh` uses `python3-pip` to fetch the build dependencies for the docs - you might want to use a [venv](https://docs.python.org/3/library/venv.html).
-If you just want to install the Go dependencies, run `./lazy.sh godep`.
+
+### Overall Architecture
+
+The application architecture is documented as part of the user docs in the *Implementation* section (`docs/content/impl`).
+Make sure to develop an understanding how zrepl is typically used by studying the user docs first.

 ### Project Structure

@@ -61,15 +62,14 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   └── samples
 ├── daemon                  # the implementation of `zrepl daemon` subcommand
 │   ├── filters
-│   ├── hooks               # snapshot hooks
 │   ├── job                 # job implementations
 │   ├── logging             # logging outlets + formatters
 │   ├── nethelpers
 │   ├── prometheus
 │   ├── pruner              # pruner implementation
 │   ├── snapper             # snapshotter implementation
-├── dist                    # supplemental material for users & package maintainers
 ├── docs                    # sphinx-based documentation
+├── dist                    # supplemental material for users & package maintainers
 │   ├── **/*.rst            # documentation in reStructuredText
 │   ├── sphinxconf
 │   │   └── conf.py         # sphinx config (see commit 445a280 why its not in docs/)
@@ -78,7 +78,6 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   └── public_git          # checkout of zrepl.github.io managed by above shell script
 ├── endpoint                # implementation of replication endpoints (=> package replication)
 ├── logger                  # our own logger package
-├── platformtest            # test suite for our zfs abstractions (error classification, etc)
 ├── pruning                 # pruning rules (the logic, not the actual execution)
 │   └── retentiongrid
 ├── replication
@@ -93,13 +92,14 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   ├── transportmux        # TCP connecter and listener used to split control & data traffic
 │   └── versionhandshake    # replication protocol version handshake perfomed on newly established connections
 ├── tlsconf                 # abstraction for Go TLS server + client config
-├── transport               # transport implementations
+├── transport               # transports implementation
 │   ├── fromconfig
 │   ├── local
 │   ├── ssh
 │   ├── tcp
 │   └── tls
 ├── util
+├── vendor                  # managed by dep
 ├── version                 # abstraction for versions (filled during build by Makefile)
 └── zfs                     # zfs(8) wrappers
 ```
@@ -134,15 +134,3 @@ There will not be a big refactoring (an attempt was made, but it's destroying to

 However, new contributions & patches should fix naming without further notice in the commit message.

-### RPC debugging
-
-Optionally, there are various RPC-related environment variables, that if set to something != `""` will produce additional debug output on stderr:
-
-https://github.com/zrepl/zrepl/blob/master/rpc/rpc_debug.go#L11
-
-https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/dataconn_debug.go#L11
-
-https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/stream/stream_debug.go#L11
-
-https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/heartbeatconn/heartbeatconn_debug.go#L11
-
@@ -99,7 +99,7 @@ func doMigratePlaceholder0_1(sc *cli.Subcommand, args []string) error {
 		}
 		for _, fs := range wi.fss {
 			fmt.Printf("\t%q ... ", fs.ToString())
-			r, err := zfs.ZFSMigrateHashBasedPlaceholderToCurrent(ctx, fs, migratePlaceholder0_1Args.dryRun)
+			r, err := zfs.ZFSMigrateHashBasedPlaceholderToCurrent(fs, migratePlaceholder0_1Args.dryRun)
 			if err != nil {
 				fmt.Printf("error: %s\n", err)
 			} else if !r.NeedsModification {
@@ -165,7 +165,7 @@ func doMigrateReplicationCursor(sc *cli.Subcommand, args []string) error {
 	var hadError bool
 	for _, fs := range fss {

-		bold.Printf("INSPECT FILESYSTEM %q\n", fs.ToString())
+		bold.Printf("INSPECT FILESYTEM %q\n", fs.ToString())

 		err := doMigrateReplicationCursorFS(ctx, v1cursorJobs, fs)
 		if err == migrateReplicationCursorSkipSentinel {
@@ -264,7 +264,7 @@ func doMigrateReplicationCursorFS(ctx context.Context, v1CursorJobs []job.Job, f
 	if migrateReplicationCursorArgs.dryRun {
 		succ.Printf("DRY RUN\n")
 	} else {
-		if err := zfs.ZFSDestroyFilesystemVersion(ctx, fs, oldCursor); err != nil {
+		if err := zfs.ZFSDestroyFilesystemVersion(fs, oldCursor); err != nil {
 			return err
 		}
 	}
@@ -9,7 +9,7 @@ import (
 )

 var SignalCmd = &cli.Subcommand{
-	Use:   "signal [wakeup|reset] JOB",
+	Use:   "signal [wakeup|reset] JOB [DATA]",
 	Short: "wake up a job from wait state or abort its current invocation",
 	Run: func(subcommand *cli.Subcommand, args []string) error {
 		return runSignalCmd(subcommand.Config(), args)
@@ -17,8 +17,13 @@ var SignalCmd = &cli.Subcommand{
 }

 func runSignalCmd(config *config.Config, args []string) error {
-	if len(args) != 2 {
-		return errors.Errorf("Expected 2 arguments: [wakeup|reset] JOB")
+	if len(args) < 2 || len(args) > 3 {
+		return errors.Errorf("Expected 2 arguments: [wakeup|reset|set-concurrency] JOB [DATA]")
+	}
+
+	var data string
+	if len(args) == 3 {
+		data = args[2]
 	}

 	httpc, err := controlHttpClient(config.Global.Control.SockPath)
@@ -30,9 +35,11 @@ func runSignalCmd(config *config.Config, args []string) error {
 		struct {
 			Name string
 			Op   string
+			Data string
 		}{
 			Name: args[1],
 			Op:   args[0],
+			Data: data,
 		},
 		struct{}{},
 	)
@@ -11,7 +11,7 @@ import (
 	"sync"
 	"time"

-	// tcell is the termbox-compatible library for abstracting away escape sequences, etc.
+	// tcell is the termbox-compatbile library for abstracting away escape sequences, etc.
 	// as of tcell#252, the number of default distributed terminals is relatively limited
 	// additional terminal definitions can be included via side-effect import
 	// See https://github.com/gdamore/tcell/blob/master/terminfo/base/base.go
@@ -81,7 +81,7 @@ type tui struct {
 	indent int

 	lock   sync.Mutex //For report and error
-	report map[string]*job.Status
+	report map[string]job.Status
 	err    error

 	jobFilter string
@@ -219,7 +219,7 @@ func runStatus(s *cli.Subcommand, args []string) error {
 	defer termbox.Close()

 	update := func() {
-		var m daemon.Status
+		m := make(map[string]job.Status)

 		err2 := jsonRequestResponse(httpc, daemon.ControlJobEndpointStatus,
 			struct{}{},
@@ -228,7 +228,7 @@ func runStatus(s *cli.Subcommand, args []string) error {

 		t.lock.Lock()
 		t.err = err2
-		t.report = m.Jobs
+		t.report = m
 		t.lock.Unlock()
 		t.draw()
 	}
@@ -264,7 +264,7 @@ loop:

 }

-func (t *tui) getReplicationProgressHistory(jobName string) *bytesProgressHistory {
+func (t *tui) getReplicationProgresHistory(jobName string) *bytesProgressHistory {
 	p, ok := t.replicationProgress[jobName]
 	if !ok {
 		p = &bytesProgressHistory{}
@@ -329,7 +329,7 @@ func (t *tui) draw() {
 				t.printf("Replication:")
 				t.newline()
 				t.addIndent(1)
-				t.renderReplicationReport(activeStatus.Replication, t.getReplicationProgressHistory(k))
+				t.renderReplicationReport(activeStatus.Replication, t.getReplicationProgresHistory(k))
 				t.addIndent(-1)

 				t.printf("Pruning Sender:")
@@ -697,7 +697,7 @@ func rightPad(str string, length int, pad string) string {

 var arrowPositions = `>\|/`

-// changeCount = 0 indicates stall / no progress
+// changeCount = 0 indicates stall / no progresss
 func (t *tui) drawBar(length int, bytes, totalBytes int64, changeCount int) {
 	var completedLength int
 	if totalBytes > 0 {
@@ -49,7 +49,6 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 	}

 	conf := subcommand.Config()
-	ctx := context.Background()

 	var confFilter config.FilesystemsFilter
 	job, err := conf.Job(testFilterArgs.job)
@@ -61,8 +60,6 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 		confFilter = j.Filesystems
 	case *config.PushJob:
 		confFilter = j.Filesystems
-	case *config.SnapJob:
-		confFilter = j.Filesystems
 	default:
 		return fmt.Errorf("job type %T does not have filesystems filter", j)
 	}
@@ -76,7 +73,7 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 	if testFilterArgs.input != "" {
 		fsnames = []string{testFilterArgs.input}
 	} else {
-		out, err := zfs.ZFSList(ctx, []string{"name"})
+		out, err := zfs.ZFSList([]string{"name"})
 		if err != nil {
 			return fmt.Errorf("could not list ZFS filesystems: %s", err)
 		}
@@ -140,11 +137,10 @@ var testPlaceholder = &cli.Subcommand{
 func runTestPlaceholder(subcommand *cli.Subcommand, args []string) error {

 	var checkDPs []*zfs.DatasetPath
-	ctx := context.Background()

 	// all actions first
 	if testPlaceholderArgs.all {
-		out, err := zfs.ZFSList(ctx, []string{"name"})
+		out, err := zfs.ZFSList([]string{"name"})
 		if err != nil {
 			return errors.Wrap(err, "could not list ZFS filesystems")
 		}
@@ -168,7 +164,7 @@ func runTestPlaceholder(subcommand *cli.Subcommand, args []string) error {

 	fmt.Printf("IS_PLACEHOLDER\tDATASET\tzrepl:placeholder\n")
 	for _, dp := range checkDPs {
-		ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, dp)
+		ph, err := zfs.ZFSGetFilesystemPlaceholderState(dp)
 		if err != nil {
 			return errors.Wrap(err, "cannot get placeholder state")
 		}
@@ -620,7 +620,7 @@ func ParseConfigBytes(bytes []byte) (*Config, error) {

 var durationStringRegex *regexp.Regexp = regexp.MustCompile(`^\s*(\d+)\s*(s|m|h|d|w)\s*$`)

-func parsePositiveDuration(e string) (d time.Duration, err error) {
+func parsePostitiveDuration(e string) (d time.Duration, err error) {
 	comps := durationStringRegex.FindStringSubmatch(e)
 	if len(comps) != 3 {
 		err = fmt.Errorf("does not match regex: %s %#v", e, comps)
@@ -21,7 +21,7 @@ jobs:
    clients: {
      "10.0.0.1":"foo"
    }
-  root_fs: zroot/foo
+  root_fs: zoot/foo
 `
 	_, err := ParseConfigBytes([]byte(jobdef))
 	require.NoError(t, err)
@@ -42,7 +42,7 @@ func TestSampleConfigsAreParsedWithoutErrors(t *testing.T) {

 }

-// template must be a template/text template with a single '{{ . }}' as placeholder for val
+// template must be a template/text template with a single '{{ . }}' as placehodler for val
 //nolint[:deadcode,unused]
 func testValidConfigTemplate(t *testing.T, tmpl string, val string) *Config {
 	tmp, err := template.New("master").Parse(tmpl)
@@ -64,7 +64,7 @@ func parseRetentionGridIntervalString(e string) (intervals []RetentionInterval,
 		return nil, fmt.Errorf("contains factor <= 0")
 	}

-	duration, err := parsePositiveDuration(comps[2])
+	duration, err := parsePostitiveDuration(comps[2])
 	if err != nil {
 		return nil, err
 	}
@@ -8,6 +8,7 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"strconv"
 	"time"

 	"github.com/pkg/errors"
@@ -20,7 +21,6 @@ import (
 	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/version"
 	"github.com/zrepl/zrepl/zfs"
-	"github.com/zrepl/zrepl/zfs/zfscmd"
 )

 type controlJob struct {
@@ -48,6 +48,8 @@ func (j *controlJob) OwnedDatasetSubtreeRoot() (p *zfs.DatasetPath, ok bool) { r

 func (j *controlJob) SenderConfig() *endpoint.SenderConfig { return nil }

+func (j *controlJob) SetConcurrency(concurrency int) error { return errors.Errorf("not supported") }
+
 var promControl struct {
 	requestBegin    *prometheus.CounterVec
 	requestFinished *prometheus.HistogramVec
@@ -65,7 +67,7 @@ func (j *controlJob) RegisterMetrics(registerer prometheus.Registerer) {
 		Namespace: "zrepl",
 		Subsystem: "control",
 		Name:      "request_finished",
-		Help:      "time it took a request to finish",
+		Help:      "time it took a request to finih",
 		Buckets:   []float64{1e-6, 10e-6, 100e-6, 500e-6, 1e-3, 10e-3, 100e-3, 200e-3, 400e-3, 800e-3, 1, 10, 20},
 	}, []string{"endpoint"})
 	registerer.MustRegister(promControl.requestBegin)
@@ -118,9 +120,7 @@ func (j *controlJob) Run(ctx context.Context) {
 	mux.Handle(ControlJobEndpointStatus,
 		// don't log requests to status endpoint, too spammy
 		jsonResponder{log, func() (interface{}, error) {
-			jobs := j.jobs.status()
-			globalZFS := zfscmd.GetReport()
-			s := Status{Jobs: jobs, Global: GlobalStatus{ZFSCmds: globalZFS}}
+			s := j.jobs.status()
 			return s, nil
 		}})

@@ -129,6 +129,7 @@ func (j *controlJob) Run(ctx context.Context) {
 			type reqT struct {
 				Name string
 				Op   string
+				Data string
 			}
 			var req reqT
 			if decoder(&req) != nil {
@@ -141,6 +142,14 @@ func (j *controlJob) Run(ctx context.Context) {
 				err = j.jobs.wakeup(req.Name)
 			case "reset":
 				err = j.jobs.reset(req.Name)
+			case "set-concurrency":
+				var concurrency int
+				concurrency, err = strconv.Atoi(req.Data) // shadow
+				if err != nil {
+					// fallthrough outer
+				} else {
+					err = j.jobs.setConcurrency(req.Name, concurrency)
+				}
 			default:
 				err = fmt.Errorf("operation %q is invalid", req.Op)
 			}
@@ -253,7 +262,7 @@ func (j jsonRequestResponder) ServeHTTP(w http.ResponseWriter, r *http.Request)
 	var buf bytes.Buffer
 	encodeErr := json.NewEncoder(&buf).Encode(res)
 	if encodeErr != nil {
-		j.log.WithError(producerErr).Error("control handler json marshal error")
+		j.log.WithError(producerErr).Error("control handler json marhsal error")
 		w.WriteHeader(http.StatusInternalServerError)
 		_, err := io.WriteString(w, encodeErr.Error())
 		logIoErr(err)
@@ -20,7 +20,6 @@ import (
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/version"
-	"github.com/zrepl/zrepl/zfs/zfscmd"
 )

 func Run(conf *config.Config) error {
@@ -82,9 +81,6 @@ func Run(conf *config.Config) error {
 		jobs.start(ctx, job, true)
 	}

-	// register global (=non job-local) metrics
-	zfscmd.RegisterMetrics(prometheus.DefaultRegisterer)
-
 	log.Info("starting daemon")

 	// start regular jobs
@@ -132,15 +128,6 @@ func (s *jobs) wait() <-chan struct{} {
 	return ch
 }

-type Status struct {
-	Jobs   map[string]*job.Status
-	Global GlobalStatus
-}
-
-type GlobalStatus struct {
-	ZFSCmds *zfscmd.Report
-}
-
 func (s *jobs) status() map[string]*job.Status {
 	s.m.RLock()
 	defer s.m.RUnlock()
@@ -189,6 +176,18 @@ func (s *jobs) reset(job string) error {
 	return wu()
 }

+func (s *jobs) setConcurrency(jobName string, concurrency int) error {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	job, ok := s.jobs[jobName]
+	if !ok {
+		return errors.Errorf("Job %q does not exist", job)
+	}
+
+	return job.SetConcurrency(concurrency)
+}
+
 const (
 	jobNamePrometheus = "_prometheus"
 	jobNameControl    = "_control"
@@ -220,7 +219,6 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) {

 	s.jobs[jobName] = j
 	ctx = job.WithLogger(ctx, jobLog)
-	ctx = zfscmd.WithJobID(ctx, j.Name())
 	ctx, wakeup := wakeup.Context(ctx)
 	ctx, resetFunc := reset.Context(ctx)
 	s.wakeups[jobName] = wakeup
@@ -161,7 +161,7 @@ func (m DatasetMapFilter) Filter(p *zfs.DatasetPath) (pass bool, err error) {
 }

 // Construct a new filter-only DatasetMapFilter from a mapping
-// The new filter allows exactly those paths that were not forbidden by the mapping.
+// The new filter allows excactly those paths that were not forbidden by the mapping.
 func (m DatasetMapFilter) InvertedFilter() (inv *DatasetMapFilter, err error) {

 	if m.filterMode {
@@ -55,9 +55,12 @@ const (
 type activeSideTasks struct {
 	state ActiveSideState

+	concurrency int
+
 	// valid for state ActiveSideReplicating, ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
-	replicationReport driver.ReportFunc
-	replicationCancel context.CancelFunc
+	replicationReport         driver.ReportFunc
+	replicationCancel         context.CancelFunc
+	replicationSetConcurrency driver.SetConcurrencyFunc

 	// valid for state ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
 	prunerSender, prunerReceiver *pruner.Pruner
@@ -147,7 +150,7 @@ func modePushFromConfig(g *config.Global, in *config.PushJob, jobID endpoint.Job

 	fsf, err := filters.DatasetMapFilterFromConfig(in.Filesystems)
 	if err != nil {
-		return nil, errors.Wrap(err, "cannot build filesystem filter")
+		return nil, errors.Wrap(err, "cannnot build filesystem filter")
 	}

 	m.senderConfig = &endpoint.SenderConfig{
@@ -278,6 +281,8 @@ func activeSide(g *config.Global, in *config.ActiveJob, configJob interface{}) (
 		return nil, errors.Wrap(err, "invalid job name")
 	}

+	j.tasks.concurrency = 1 // FIXME
+
 	switch v := configJob.(type) {
 	case *config.PushJob:
 		j.mode, err = modePushFromConfig(g, v, j.name) // shadow
@@ -375,6 +380,22 @@ func (j *ActiveSide) SenderConfig() *endpoint.SenderConfig {
 	return push.senderConfig
 }

+func (j *ActiveSide) SetConcurrency(concurrency int) (err error) {
+	j.updateTasks(func(tasks *activeSideTasks) {
+		if tasks.replicationSetConcurrency != nil {
+			err = tasks.replicationSetConcurrency(concurrency) // no shadow
+			if err == nil {
+				tasks.concurrency = concurrency
+			}
+		} else {
+			// FIXME this is not great, should always be able to set it
+			err = errors.Errorf("cannot set while not replicating")
+		}
+
+	})
+	return err
+}
+
 func (j *ActiveSide) Run(ctx context.Context) {
 	log := GetLogger(ctx)
 	ctx = logging.WithSubsystemLoggers(ctx, log)
@@ -436,11 +457,14 @@ func (j *ActiveSide) do(ctx context.Context) {
 		ctx, repCancel := context.WithCancel(ctx)
 		var repWait driver.WaitFunc
 		j.updateTasks(func(tasks *activeSideTasks) {
-			// reset it
-			*tasks = activeSideTasks{}
+			// reset it (almost)
+			old := *tasks
+			*tasks = activeSideTasks{
+				concurrency: old.concurrency,
+			}
 			tasks.replicationCancel = repCancel
-			tasks.replicationReport, repWait = replication.Do(
-				ctx, logic.NewPlanner(j.promRepStateSecs, j.promBytesReplicated, sender, receiver, j.mode.PlannerPolicy()),
+			tasks.replicationReport, repWait, tasks.replicationSetConcurrency = replication.Do(
+				ctx, tasks.concurrency, logic.NewPlanner(j.promRepStateSecs, j.promBytesReplicated, sender, receiver, j.mode.PlannerPolicy()),
 			)
 			tasks.state = ActiveSideReplicating
 		})
@@ -40,6 +40,7 @@ type Job interface {
 	// must return the root of that subtree as rfs and ok = true
 	OwnedDatasetSubtreeRoot() (rfs *zfs.DatasetPath, ok bool)
 	SenderConfig() *endpoint.SenderConfig
+	SetConcurrency(concurrency int) error
 }

 type Type string
@@ -75,7 +75,7 @@ func modeSourceFromConfig(g *config.Global, in *config.SourceJob, jobID endpoint
 	m = &modeSource{}
 	fsf, err := filters.DatasetMapFilterFromConfig(in.Filesystems)
 	if err != nil {
-		return nil, errors.Wrap(err, "cannot build filesystem filter")
+		return nil, errors.Wrap(err, "cannnot build filesystem filter")
 	}
 	m.senderConfig = &endpoint.SenderConfig{
 		FSF:     fsf,
@@ -163,6 +163,8 @@ func (j *PassiveSide) SenderConfig() *endpoint.SenderConfig {

 func (*PassiveSide) RegisterMetrics(registerer prometheus.Registerer) {}

+func (*PassiveSide) SetConcurrency(concurrency int) error { return errors.Errorf("not supported") }
+
 func (j *PassiveSide) Run(ctx context.Context) {

 	log := GetLogger(ctx)
@@ -117,6 +117,8 @@ outer:
 	}
 }

+func (*SnapJob) SetConcurrency(concurrency int) error { return errors.Errorf("not supported") }
+
 // Adaptor that implements pruner.History around a pruner.Target.
 // The ReplicationCursor method is Get-op only and always returns
 // the filesystem's most recent version's GUID.
@@ -22,7 +22,6 @@ import (
 	"github.com/zrepl/zrepl/rpc/transportmux"
 	"github.com/zrepl/zrepl/tlsconf"
 	"github.com/zrepl/zrepl/transport"
-	"github.com/zrepl/zrepl/zfs/zfscmd"
 )

 func OutletsFromConfig(in config.LoggingOutletEnumList) (*logger.Outlets, error) {
@@ -71,7 +70,7 @@ type Subsystem string

 const (
 	SubsysReplication  Subsystem = "repl"
-	SubsysEndpoint     Subsystem = "endpoint"
+	SubsyEndpoint      Subsystem = "endpoint"
 	SubsysPruning      Subsystem = "pruning"
 	SubsysSnapshot     Subsystem = "snapshot"
 	SubsysHooks        Subsystem = "hook"
@@ -80,19 +79,17 @@ const (
 	SubsysRPC          Subsystem = "rpc"
 	SubsysRPCControl   Subsystem = "rpc.ctrl"
 	SubsysRPCData      Subsystem = "rpc.data"
-	SubsysZFSCmd       Subsystem = "zfs.cmd"
 )

 func WithSubsystemLoggers(ctx context.Context, log logger.Logger) context.Context {
 	ctx = logic.WithLogger(ctx, log.WithField(SubsysField, SubsysReplication))
 	ctx = driver.WithLogger(ctx, log.WithField(SubsysField, SubsysReplication))
-	ctx = endpoint.WithLogger(ctx, log.WithField(SubsysField, SubsysEndpoint))
+	ctx = endpoint.WithLogger(ctx, log.WithField(SubsysField, SubsyEndpoint))
 	ctx = pruner.WithLogger(ctx, log.WithField(SubsysField, SubsysPruning))
 	ctx = snapper.WithLogger(ctx, log.WithField(SubsysField, SubsysSnapshot))
 	ctx = hooks.WithLogger(ctx, log.WithField(SubsysField, SubsysHooks))
 	ctx = transport.WithLogger(ctx, log.WithField(SubsysField, SubsysTransport))
 	ctx = transportmux.WithLogger(ctx, log.WithField(SubsysField, SubsysTransportMux))
-	ctx = zfscmd.WithLogger(ctx, log.WithField(SubsysField, SubsysZFSCmd))
 	ctx = rpc.WithLoggers(ctx,
 		rpc.Loggers{
 			General: log.WithField(SubsysField, SubsysRPC),
@@ -211,7 +211,7 @@ func logfmtTryEncodeKeyval(enc *logfmt.Encoder, field, value interface{}) error
 	case logfmt.ErrUnsupportedValueType:
 		err := enc.EncodeKeyval(field, fmt.Sprintf("<%T>", value))
 		if err != nil {
-			return errors.Wrap(err, "cannot encode unsupported value type Go type")
+			return errors.Wrap(err, "cannot encode unsuuported value type Go type")
 		}
 		return nil
 	}
@@ -63,7 +63,7 @@ func NewTCPOutlet(formatter EntryFormatter, network, address string, tlsConfig *
 		return
 	}

-	entryChan := make(chan *bytes.Buffer, 1) // allow one message in flight while previous is in io.Copy()
+	entryChan := make(chan *bytes.Buffer, 1) // allow one message in flight while previos is in io.Copy()

 	o := &TCPOutlet{
 		formatter: formatter,
@@ -5,6 +5,7 @@ import (
 	"net"
 	"net/http"

+	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"

@@ -51,6 +52,8 @@ func (j *prometheusJob) OwnedDatasetSubtreeRoot() (p *zfs.DatasetPath, ok bool)

 func (j *prometheusJob) SenderConfig() *endpoint.SenderConfig { return nil }

+func (j *prometheusJob) SetConcurrency(concurrency int) error { return errors.Errorf("not supported") }
+
 func (j *prometheusJob) RegisterMetrics(registerer prometheus.Registerer) {}

 func (j *prometheusJob) Run(ctx context.Context) {
@@ -18,13 +18,13 @@ import (
 	"github.com/zrepl/zrepl/util/envconst"
 )

-// Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
+// Try to keep it compatible with gitub.com/zrepl/zrepl/endpoint.Endpoint
 type History interface {
 	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
 	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
 }

-// Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
+// Try to keep it compatible with gitub.com/zrepl/zrepl/endpoint.Endpoint
 type Target interface {
 	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
 	ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error)
@@ -277,7 +277,7 @@ func snapshot(a args, u updater) state {

 		jobCallback := hooks.NewCallbackHookForFilesystem("snapshot", fs, func(_ context.Context) (err error) {
 			l.Debug("create snapshot")
-			err = zfs.ZFSSnapshot(a.ctx, fs, snapname, false) // TODO propagate context to ZFSSnapshot
+			err = zfs.ZFSSnapshot(fs, snapname, false) // TODO propagagte context to ZFSSnapshot
 			if err != nil {
 				l.WithError(err).Error("cannot create snapshot")
 			}
@@ -105,7 +105,7 @@ Actual changelog:
   |  You can support maintenance and feature development through one of the following services:
   |  |Donate via Patreon| |Donate via Liberapay| |Donate via PayPal|
   |  Note that PayPal processing fees are relatively high for small donations.
-   |  For SEPA wire transfer and **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.
+   |  For SEPA wire transfer and **commerical support**, please `contact Christian directly <https://cschwarz.com>`_.


 0.1.1
@@ -202,7 +202,7 @@ The latter is particularly useful in combination with log aggregation services.
 .. WARNING::

    zrepl drops log messages to the TCP outlet if the underlying connection is not fast enough.
-    Note that TCP buffering in the kernel must first run full before messages are dropped.
+    Note that TCP buffering in the kernel must first run full becfore messages are dropped.

    Make sure to always configure a ``stdout`` outlet as the special error outlet to be informed about problems
    with the TCP outlet (see :ref:`above <logging-error-outlet>` ).
@@ -15,7 +15,7 @@ Prometheus & Grafana
 zrepl can expose `Prometheus metrics <https://prometheus.io/docs/instrumenting/exposition_formats/>`_ via HTTP.
 The ``listen`` attribute is a `net.Listen <https://golang.org/pkg/net/#Listen>`_  string for tcp, e.g. ``:9091`` or ``127.0.0.1:9091``.
 The ``listen_freebind`` attribute is :ref:`explained here <listen-freebind-explanation>`.
-The Prometheus monitoring job appears in the ``zrepl control`` job list and may be specified **at most once**.
+The Prometheues monitoring job appears in the ``zrepl control`` job list and may be specified **at most once**.

 zrepl also ships with an importable `Grafana <https://grafana.com>`_ dashboard that consumes the Prometheus metrics:
 see :repomasterlink:`dist/grafana`.
@@ -26,9 +26,9 @@ Config File Structure
     type: push
   - ...

-zrepl is configured using a single YAML configuration file with two main sections: ``global`` and ``jobs``.
+zrepl is confgured using a single YAML configuration file with two main sections: ``global`` and ``jobs``.
 The ``global`` section is filled with sensible defaults and is covered later in this chapter.
-The ``jobs`` section is a list of jobs which we are going to explain now.
+The ``jobs`` section is a list of jobs which we are goind to explain now.

 .. _job-overview:

@@ -41,7 +41,7 @@ Jobs are identified by their ``name``, both in log files and the ``zrepl status`

 Replication always happens between a pair of jobs: one is the **active side**, and one the **passive side**.
 The active side connects to the passive side using a :ref:`transport <transport>` and starts executing the replication logic.
-The passive side responds to requests from the active side after checking its permissions.
+The passive side responds to requests from the active side after checking its persmissions.

 The following table shows how different job types can be combined to achieve **both push and pull mode setups**.
 Note that snapshot-creation denoted by "(snap)" is orthogonal to whether a job is active or passive.
@@ -120,7 +120,7 @@ The following steps take place during replication and can be monitored using the
  * Perform replication steps in the following order:
    Among all filesystems with pending replication steps, pick the filesystem whose next replication step's snapshot is the oldest.
  * Create placeholder filesystems on the receiving side to mirror the dataset paths on the sender to ``root_fs/${client_identity}``.
-  * Acquire send-side step-holds on the step's `from` and `to` snapshots.
+  * Aquire send-side step-holds on the step's `from` and `to` snapshots.
  * Perform the replication step.
  * Move the **replication cursor** bookmark on the sending side (see below).
  * Move the **last-received-hold** on the receiving side (see below).
@@ -141,7 +141,7 @@ The ``zrepl holds list`` provides a listing of all bookmarks and holds managed b
 .. _replication-placeholder-property:

 **Placeholder filesystems** on the receiving side are regular ZFS filesystems with the placeholder property ``zrepl:placeholder=on``.
-Placeholders allow the receiving side to mirror the sender's ZFS dataset hierarchy without replicating every filesystem at every intermediary dataset path component.
+Placeholders allow the receiving side to mirror the sender's ZFS dataset hierachy without replicating every filesystem at every intermediary dataset path component.
 Consider the following example: ``S/H/J`` shall be replicated to ``R/sink/job/S/H/J``, but neither ``S/H`` nor ``S`` shall be replicated.
 ZFS requires the existence of ``R/sink/job/S`` and ``R/sink/job/S/H`` in order to receive into ``R/sink/job/S/H/J``.
 Thus, zrepl creates the parent filesystems as placeholders on the receiving side.
@@ -181,7 +181,7 @@ No Overlapping

 Jobs run independently of each other.
 If two jobs match the same filesystem with their ``filesystems`` filter, they will operate on that filesystem independently and potentially in parallel.
-For example, if job A prunes snapshots that job B is planning to replicate, the replication will fail because B assumed the snapshot to still be present.
+For example, if job A prunes snapshots that job B is planning to replicate, the replication will fail because B asssumed the snapshot to still be present.
 However, the next replication attempt will re-examine the situation from scratch and should work.

 N push jobs to 1 sink
@@ -198,5 +198,5 @@ Multiple pull jobs pulling from the same source have potential for race conditio
 each pull job prunes the source side independently, causing replication-prune and prune-prune races.

 There is currently no way for a pull job to filter which snapshots it should attempt to replicate.
-Thus, it is not possible to just manually assert that the prune rules of all pull jobs are disjoint to avoid replication-prune and prune-prune races.
+Thus, it is not possibe to just manually assert that the prune rules of all pull jobs are disjoint to avoid replication-prune and prune-prune races.

@@ -154,7 +154,7 @@ Policy ``regex``
           negate: true
           regex: "^zrepl_.*"

-``regex`` keeps all snapshots whose names are matched by the regular expression in ``regex``.
+``regex`` keeps all snapshots whose names are matched by the regular expressionin ``regex``.
 Like all other regular expression fields in prune policies, zrepl uses Go's `regexp.Regexp <https://golang.org/pkg/regexp/#Compile>`_ Perl-compatible regular expressions (`Syntax <https://golang.org/pkg/regexp/syntax>`_).
 The optional `negate` boolean field inverts the semantics: Use it if you want to keep all snapshots that *do not* match the given regex.

@@ -24,7 +24,7 @@ Send Options
 ---------------------

 The ``encryption`` variable controls whether the matched filesystems are sent as `OpenZFS native encryption <http://open-zfs.org/wiki/ZFS-Native_Encryption>`_ raw sends.
-More specifically, if ``encryption=true``, zrepl
+More specificially, if ``encryption=true``, zrepl

 * checks for any of the filesystems matched by ``filesystems`` whether the ZFS ``encryption`` property indicates that the filesystem is actually encrypted with ZFS native encryption and
 * invokes the ``zfs send`` subcommand with the ``-w`` option (raw sends) and
@@ -101,9 +101,9 @@ and serves as a reference for build dependencies and procedure:

 ::

-    git clone https://github.com/zrepl/zrepl.git && \
-    cd zrepl && \
-    sudo docker build -t zrepl_build -f build.Dockerfile . && \
+    git clone https://github.com/zrepl/zrepl.git
+    cd zrepl
+    sudo docker build -t zrepl_build -f build.Dockerfile .
    sudo docker run -it --rm \
        -v "${PWD}:/src" \
        --user "$(id -u):$(id -g)" \
@@ -127,7 +127,7 @@ Either way, all build results are located in the ``artifacts/`` directory.

 .. NOTE::

-    It is your job to install the appropriate binary in the zrepl users's ``$PATH``, e.g. ``/usr/local/bin/zrepl``.
+    It is your job to install the apropriate binary in the zrepl users's ``$PATH``, e.g. ``/usr/local/bin/zrepl``.
    Otherwise, the examples in the :ref:`tutorial` may need to be adjusted.

 What next?
@@ -6,7 +6,7 @@

 zrepl is a spare-time project primarily developed by `Christian Schwarz <https://cschwarz.com>`_.
 You can support maintenance and feature development through one of the services listed above.
-For SEPA wire transfer and **commercial support**, please `contact Christian directly <https://cschwarz.com>`_.
+For SEPA wire transfer and **commerical support**, please `contact Christian directly <https://cschwarz.com>`_.

 **Thanks for your support!**

@@ -48,7 +48,7 @@ zrepl daemon
 All actual work zrepl does is performed by a daemon process.
 The daemon supports structured :ref:`logging <logging>` and provides :ref:`monitoring endpoints <monitoring>`.

-When installing from a package, the package maintainer should have provided an init script / systemd.service file.
+When installating from a package, the package maintainer should have provided an init script / systemd.service file.
 You should thus be able to start zrepl daemon using your init system.

 Alternatively, or for running zrepl in the foreground, simply execute ``zrepl daemon``.
@@ -73,6 +73,6 @@ The daemon exits as soon as all jobs have reported shut down.
 Systemd Unit File
 ~~~~~~~~~~~~~~~~~

-A systemd service definition template is available in :repomasterlink:`dist/systemd`.
+A systemd service defintion template is available in :repomasterlink:`dist/systemd`.
 Note that some of the options only work on recent versions of systemd.
 Any help & improvements are very welcome, see :issue:`145`.
@@ -170,7 +170,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 		// ok, fallthrough outer
 	case pdu.Tri_False:
 		if s.encrypt.B {
-			return nil, nil, errors.New("only encrypted sends allowed (send -w + encryption!= off), but unencrypted send requested")
+			return nil, nil, errors.New("only encrytped sends allowed (send -w + encryption!= off), but unencrytped send requested")
 		}
 		// fallthrough outer
 	case pdu.Tri_True:
@@ -206,7 +206,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 	}

 	// From now on, assume that sendArgs has been validated by ZFSSendDry
-	// (because validation involves shelling out, it's actually a little expensive)
+	// (because validation invovles shelling out, it's actually a little expensive)

 	var expSize int64 = 0      // protocol says 0 means no estimate
 	if si.SizeEstimate != -1 { // but si returns -1 for no size estimate
@@ -511,7 +511,7 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR
 	fss := make([]*pdu.Filesystem, 0, len(filtered))
 	for _, a := range filtered {
 		l := getLogger(ctx).WithField("fs", a)
-		ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, a)
+		ph, err := zfs.ZFSGetFilesystemPlaceholderState(a)
 		if err != nil {
 			l.WithError(err).Error("error getting placeholder state")
 			return nil, errors.Wrapf(err, "cannot get placeholder state for fs %q", a)
@@ -625,9 +625,9 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	//       ZFS dataset hierarchy subtrees.
 	var visitErr error
 	func() {
-		getLogger(ctx).Debug("begin acquire recvParentCreationMtx")
+		getLogger(ctx).Debug("begin aquire recvParentCreationMtx")
 		defer s.recvParentCreationMtx.Lock().Unlock()
-		getLogger(ctx).Debug("end acquire recvParentCreationMtx")
+		getLogger(ctx).Debug("end aquire recvParentCreationMtx")
 		defer getLogger(ctx).Debug("release recvParentCreationMtx")

 		f := zfs.NewDatasetPathForest()
@@ -637,7 +637,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 			if v.Path.Equal(lp) {
 				return false
 			}
-			ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, v.Path)
+			ph, err := zfs.ZFSGetFilesystemPlaceholderState(v.Path)
 			getLogger(ctx).
 				WithField("fs", v.Path.ToString()).
 				WithField("placeholder_state", fmt.Sprintf("%#v", ph)).
@@ -661,7 +661,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 				}
 				l := getLogger(ctx).WithField("placeholder_fs", v.Path)
 				l.Debug("create placeholder filesystem")
-				err := zfs.ZFSCreatePlaceholderFilesystem(ctx, v.Path)
+				err := zfs.ZFSCreatePlaceholderFilesystem(v.Path)
 				if err != nil {
 					l.WithError(err).Error("cannot create placeholder filesystem")
 					visitErr = err
@@ -681,19 +681,19 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	// determine whether we need to rollback the filesystem / change its placeholder state
 	var clearPlaceholderProperty bool
 	var recvOpts zfs.RecvOptions
-	ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, lp)
+	ph, err := zfs.ZFSGetFilesystemPlaceholderState(lp)
 	if err == nil && ph.FSExists && ph.IsPlaceholder {
 		recvOpts.RollbackAndForceRecv = true
 		clearPlaceholderProperty = true
 	}
 	if clearPlaceholderProperty {
-		if err := zfs.ZFSSetPlaceholder(ctx, lp, false); err != nil {
+		if err := zfs.ZFSSetPlaceholder(lp, false); err != nil {
 			return nil, fmt.Errorf("cannot clear placeholder property for forced receive: %s", err)
 		}
 	}

 	if req.ClearResumeToken && ph.FSExists {
-		if err := zfs.ZFSRecvClearResumeToken(ctx, lp.ToString()); err != nil {
+		if err := zfs.ZFSRecvClearResumeToken(lp.ToString()); err != nil {
 			return nil, errors.Wrap(err, "cannot clear resume token")
 		}
 	}
@@ -170,7 +170,7 @@ func MoveReplicationCursor(ctx context.Context, fs string, target *zfs.ZFSSendAr
 	// idempotently create bookmark (guid is encoded in it, hence we'll most likely add a new one
 	// cleanup the old one afterwards

-	err = zfs.ZFSBookmark(ctx, fs, *target, bookmarkname)
+	err = zfs.ZFSBookmark(fs, *target, bookmarkname)
 	if err != nil {
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			return nil, err // TODO go1.13 use wrapping
@@ -221,13 +221,13 @@ func HoldStep(ctx context.Context, fs string, v *zfs.ZFSSendArgVersion, jobID Jo
 		return errors.Wrap(err, "create step bookmark: determine bookmark name")
 	}
 	// idempotently create bookmark
-	err = zfs.ZFSBookmark(ctx, fs, *v, bmname)
+	err = zfs.ZFSBookmark(fs, *v, bmname)
 	if err != nil {
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			// TODO we could actually try to find a local snapshot that has the requested GUID
 			// 		however, the replication algorithm prefers snapshots anyways, so this quest
 			// 		is most likely not going to be successful. Also, there's the possibility that
-			//      the caller might want to filter what snapshots are eligible, and this would
+			//      the caller might want to filter what snapshots are eligibile, and this would
 			//      complicate things even further.
 			return err // TODO go1.13 use wrapping
 		}
@@ -269,7 +269,7 @@ func ReleaseStep(ctx context.Context, fs string, v *zfs.ZFSSendArgVersion, jobID
 	}
 	// idempotently destroy bookmark

-	if err := zfs.ZFSDestroyIdempotent(ctx, bmname); err != nil {
+	if err := zfs.ZFSDestroyIdempotent(bmname); err != nil {
 		return errors.Wrap(err, "step release: bookmark destroy: zfs")
 	}

@@ -393,7 +393,7 @@ type ListHoldsAndBookmarksOutputHold struct {
 // List all holds and bookmarks managed by endpoint
 func ListZFSHoldsAndBookmarks(ctx context.Context, fsfilter zfs.DatasetFilter) (*ListHoldsAndBookmarksOutput, error) {

-	// initialize all fields so that JSON serialization of output looks pretty (see client/holds.go)
+	// initialize all fields so that JSON serializion of output looks pretty (see client/holds.go)
 	// however, listZFSHoldsAndBookmarksImplFS shouldn't rely on it
 	out := &ListHoldsAndBookmarksOutput{
 		StepBookmarks:              make([]*ListHoldsAndBookmarksOutputBookmark, 0),
@@ -100,7 +100,7 @@ func destroyBookmarksOlderThan(ctx context.Context, fs string, mostRecent *zfs.Z

 	// FIXME use batch destroy, must adopt code to handle bookmarks
 	for _, v := range destroy {
-		if err := zfs.ZFSDestroyIdempotent(ctx, v.ToAbsPath(fsp)); err != nil {
+		if err := zfs.ZFSDestroyIdempotent(v.ToAbsPath(fsp)); err != nil {
 			return nil, errors.Wrap(err, "destroy bookmark")
 		}
 	}
@@ -9,8 +9,8 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )

-// JobID instances returned by MakeJobID() guarantee their JobID.String()
-// can be used in ZFS dataset names and hold tags.
+// An instance of this type returned by MakeJobID guarantees
+// that that instance's JobID.String() can be used in a ZFS dataset name and hold tag.
 type JobID struct {
 	jid string
 }
@@ -21,7 +21,7 @@ func MakeJobID(s string) (JobID, error) {
 	}

 	if err := zfs.ComponentNamecheck(s); err != nil {
-		return JobID{}, errors.Wrap(err, "must be usable as a dataset path component")
+		return JobID{}, errors.Wrap(err, "muse be usable as a dataset path component")
 	}

 	if _, err := stepBookmarkNameImpl("pool/ds", 0xface601d, s); err != nil {
@@ -34,7 +34,7 @@ func MakeJobID(s string) (JobID, error) {
 	}

 	if _, err := lastReceivedHoldImpl(s); err != nil {
-		return JobID{}, errors.Wrap(err, "must be usable as a last-received-hold tag")
+		return JobID{}, errors.Wrap(err, "must be usabel as a last-recieved-hold tag")
 	}

 	// FIXME replication cursor bookmark name
@@ -57,7 +57,7 @@ func MustMakeJobID(s string) JobID {

 func (j JobID) expectInitialized() {
 	if j.jid == "" {
-		panic("use of uninitialized JobID")
+		panic("use of unitialized JobID")
 	}
 }

@@ -172,21 +172,21 @@ type testCaseResult struct {
 func runTestCase(ctx *platformtest.Context, ex platformtest.Execer, c tests.Case) *testCaseResult {

 	// run case
-	var panicked = false
+	var paniced = false
 	var panicValue interface{} = nil
 	var panicStack error
 	func() {
 		defer func() {
 			if item := recover(); item != nil {
 				panicValue = item
-				panicked = true
+				paniced = true
 				panicStack = errors.Errorf("panic while running test: %v", panicValue)
 			}
 		}()
 		c(ctx)
 	}()

-	if panicked {
+	if paniced {
 		switch panicValue {
 		case platformtest.SkipNowSentinel:
 			return &testCaseResult{skipped: true}
@@ -180,8 +180,8 @@ func splitQuotedWords(data []byte, atEOF bool) (advance int, token []byte, err e
 				// unescaped quote, end of this string
 				// remove backslash-escapes
 				withBackslash := data[begin+1 : end]
-				withoutBackslash := bytes.Replace(withBackslash, []byte("\\\""), []byte("\""), -1)
-				return end + 1, withoutBackslash, nil
+				withoutBaskslash := bytes.Replace(withBackslash, []byte("\\\""), []byte("\""), -1)
+				return end + 1, withoutBaskslash, nil
 			} else {
 				// continue to next quote
 				end += 1
@@ -34,7 +34,7 @@ func (a ZpoolCreateArgs) Validate() error {
 		return errors.Errorf("Mountpoint must be an absolute path to a directory")
 	}
 	if a.PoolName == "" {
-		return errors.Errorf("PoolName must not be empty")
+		return errors.Errorf("PoolName must not be emtpy")
 	}
 	return nil
 }
@@ -45,7 +45,7 @@ func CreateOrReplaceZpool(ctx context.Context, e Execer, args ZpoolCreateArgs) (
 	}

 	// export pool if it already exists (idempotence)
-	if _, err := zfs.ZFSGetRawAnySource(ctx, args.PoolName, []string{"name"}); err != nil {
+	if _, err := zfs.ZFSGetRawAnySource(args.PoolName, []string{"name"}); err != nil {
 		if _, ok := err.(*zfs.DatasetDoesNotExist); ok {
 			// we'll create it shortly
 		} else {
@@ -54,7 +54,7 @@ func rollupReleaseTest(ctx *platformtest.Context, cb func(fs string) []rollupRel

 func RollupReleaseIncluding(ctx *platformtest.Context) {
 	rollupReleaseTest(ctx, func(fs string) []rollupReleaseExpectTags {
-		guid5, err := zfs.ZFSGetGUID(ctx, fs, "@5")
+		guid5, err := zfs.ZFSGetGUID(fs, "@5")
 		require.NoError(ctx, err)

 		err = zfs.ZFSReleaseAllOlderAndIncludingGUID(ctx, fs, guid5, "zrepl_platformtest")
@@ -73,7 +73,7 @@ func RollupReleaseIncluding(ctx *platformtest.Context) {

 func RollupReleaseExcluding(ctx *platformtest.Context) {
 	rollupReleaseTest(ctx, func(fs string) []rollupReleaseExpectTags {
-		guid5, err := zfs.ZFSGetGUID(ctx, fs, "@5")
+		guid5, err := zfs.ZFSGetGUID(fs, "@5")
 		require.NoError(ctx, err)

 		err = zfs.ZFSReleaseAllOlderThanGUID(ctx, fs, guid5, "zrepl_platformtest")
@@ -92,13 +92,13 @@ func RollupReleaseExcluding(ctx *platformtest.Context) {

 func RollupReleaseMostRecentIsBookmarkWithoutSnapshot(ctx *platformtest.Context) {
 	rollupReleaseTest(ctx, func(fs string) []rollupReleaseExpectTags {
-		guid5, err := zfs.ZFSGetGUID(ctx, fs, "#5")
+		guid5, err := zfs.ZFSGetGUID(fs, "#5")
 		require.NoError(ctx, err)

 		err = zfs.ZFSRelease(ctx, "zrepl_platformtest", fs+"@5")
 		require.NoError(ctx, err)

-		err = zfs.ZFSDestroy(ctx, fs+"@5")
+		err = zfs.ZFSDestroy(fs + "@5")
 		require.NoError(ctx, err)

 		err = zfs.ZFSReleaseAllOlderAndIncludingGUID(ctx, fs, guid5, "zrepl_platformtest")
@@ -117,7 +117,7 @@ func RollupReleaseMostRecentIsBookmarkWithoutSnapshot(ctx *platformtest.Context)

 func RollupReleaseMostRecentIsBookmarkAndSnapshotStillExists(ctx *platformtest.Context) {
 	rollupReleaseTest(ctx, func(fs string) []rollupReleaseExpectTags {
-		guid5, err := zfs.ZFSGetGUID(ctx, fs, "#5")
+		guid5, err := zfs.ZFSGetGUID(fs, "#5")
 		require.NoError(ctx, err)

 		err = zfs.ZFSReleaseAllOlderAndIncludingGUID(ctx, fs, guid5, "zrepl_platformtest")
@@ -17,14 +17,14 @@ func GetNonexistent(ctx *platformtest.Context) {
 	`)

 	// test raw
-	_, err := zfs.ZFSGetRawAnySource(ctx, fmt.Sprintf("%s/foo bar", ctx.RootDataset), []string{"name"})
+	_, err := zfs.ZFSGetRawAnySource(fmt.Sprintf("%s/foo bar", ctx.RootDataset), []string{"name"})
 	if err != nil {
 		panic(err)
 	}

 	// test nonexistent filesystem
 	nonexistent := fmt.Sprintf("%s/nonexistent filesystem", ctx.RootDataset)
-	props, err := zfs.ZFSGetRawAnySource(ctx, nonexistent, []string{"name"})
+	props, err := zfs.ZFSGetRawAnySource(nonexistent, []string{"name"})
 	if err == nil {
 		panic(props)
 	}
@@ -37,7 +37,7 @@ func GetNonexistent(ctx *platformtest.Context) {

 	// test nonexistent snapshot
 	nonexistent = fmt.Sprintf("%s/foo bar@non existent", ctx.RootDataset)
-	props, err = zfs.ZFSGetRawAnySource(ctx, nonexistent, []string{"name"})
+	props, err = zfs.ZFSGetRawAnySource(nonexistent, []string{"name"})
 	if err == nil {
 		panic(props)
 	}
@@ -50,7 +50,7 @@ func GetNonexistent(ctx *platformtest.Context) {

 	// test nonexistent bookmark
 	nonexistent = fmt.Sprintf("%s/foo bar#non existent", ctx.RootDataset)
-	props, err = zfs.ZFSGetRawAnySource(ctx, nonexistent, []string{"name"})
+	props, err = zfs.ZFSGetRawAnySource(nonexistent, []string{"name"})
 	if err == nil {
 		panic(props)
 	}
@@ -14,8 +14,8 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )

-func sendArgVersion(ctx *platformtest.Context, fs, relName string) zfs.ZFSSendArgVersion {
-	guid, err := zfs.ZFSGetGUID(ctx, fs, relName)
+func sendArgVersion(fs, relName string) zfs.ZFSSendArgVersion {
+	guid, err := zfs.ZFSGetGUID(fs, relName)
 	if err != nil {
 		panic(err)
 	}
@@ -33,7 +33,7 @@ func mustDatasetPath(fs string) *zfs.DatasetPath {
 	return p
 }

-func mustSnapshot(ctx *platformtest.Context, snap string) {
+func mustSnapshot(snap string) {
 	if err := zfs.EntityNamecheck(snap, zfs.EntityTypeSnapshot); err != nil {
 		panic(err)
 	}
@@ -41,14 +41,14 @@ func mustSnapshot(ctx *platformtest.Context, snap string) {
 	if len(comps) != 2 {
 		panic(comps)
 	}
-	err := zfs.ZFSSnapshot(ctx, mustDatasetPath(comps[0]), comps[1], false)
+	err := zfs.ZFSSnapshot(mustDatasetPath(comps[0]), comps[1], false)
 	if err != nil {
 		panic(err)
 	}
 }

-func mustGetProps(ctx *platformtest.Context, entity string) zfs.ZFSPropCreateTxgAndGuidProps {
-	props, err := zfs.ZFSGetCreateTXGAndGuid(ctx, entity)
+func mustGetProps(entity string) zfs.ZFSPropCreateTxgAndGuidProps {
+	props, err := zfs.ZFSGetCreateTXGAndGuid(entity)
 	check(err)
 	return props
 }
@@ -87,7 +87,7 @@ type resumeSituation struct {
 func makeDummyDataSnapshots(ctx *platformtest.Context, sendFS string) (situation dummySnapshotSituation) {

 	situation.sendFS = sendFS
-	sendFSMount, err := zfs.ZFSGetMountpoint(ctx, sendFS)
+	sendFSMount, err := zfs.ZFSGetMountpoint(sendFS)
 	require.NoError(ctx, err)
 	require.True(ctx, sendFSMount.Mounted)

@@ -95,13 +95,13 @@ func makeDummyDataSnapshots(ctx *platformtest.Context, sendFS string) (situation
 	situation.dummyDataLen = dummyLen

 	writeDummyData(path.Join(sendFSMount.Mountpoint, "dummy_data"), dummyLen)
-	mustSnapshot(ctx, sendFS+"@a snapshot")
-	snapA := sendArgVersion(ctx, sendFS, "@a snapshot")
+	mustSnapshot(sendFS + "@a snapshot")
+	snapA := sendArgVersion(sendFS, "@a snapshot")
 	situation.snapA = &snapA

 	writeDummyData(path.Join(sendFSMount.Mountpoint, "dummy_data"), dummyLen)
-	mustSnapshot(ctx, sendFS+"@b snapshot")
-	snapB := sendArgVersion(ctx, sendFS, "@b snapshot")
+	mustSnapshot(sendFS + "@b snapshot")
+	snapB := sendArgVersion(sendFS, "@b snapshot")
 	situation.snapB = &snapB

 	return situation
@@ -113,7 +113,7 @@ func makeResumeSituation(ctx *platformtest.Context, src dummySnapshotSituation,

 	situation.sendArgs = sendArgs
 	situation.recvOpts = recvOptions
-	require.True(ctx, recvOptions.SavePartialRecvState, "this method would be pointless otherwise")
+	require.True(ctx, recvOptions.SavePartialRecvState, "this method would be pointeless otherwise")
 	require.Equal(ctx, sendArgs.FS, src.sendFS)

 	copier, err := zfs.ZFSSend(ctx, sendArgs)
@@ -19,22 +19,22 @@ func IdempotentBookmark(ctx *platformtest.Context) {

 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)

-	asnap := sendArgVersion(ctx, fs, "@a snap")
-	anotherSnap := sendArgVersion(ctx, fs, "@another snap")
+	asnap := sendArgVersion(fs, "@a snap")
+	anotherSnap := sendArgVersion(fs, "@another snap")

-	err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	err := zfs.ZFSBookmark(fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}

 	// do it again, should be idempotent
-	err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	err = zfs.ZFSBookmark(fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}

 	// should fail for another snapshot
-	err = zfs.ZFSBookmark(ctx, fs, anotherSnap, "a bookmark")
+	err = zfs.ZFSBookmark(fs, anotherSnap, "a bookmark")
 	if err == nil {
 		panic(err)
 	}
@@ -43,12 +43,12 @@ func IdempotentBookmark(ctx *platformtest.Context) {
 	}

 	// destroy the snapshot
-	if err := zfs.ZFSDestroy(ctx, fmt.Sprintf("%s@a snap", fs)); err != nil {
+	if err := zfs.ZFSDestroy(fmt.Sprintf("%s@a snap", fs)); err != nil {
 		panic(err)
 	}

 	// do it again, should fail with special error type
-	err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	err = zfs.ZFSBookmark(fs, asnap, "a bookmark")
 	if err == nil {
 		panic(err)
 	}
@@ -18,8 +18,8 @@ func IdempotentDestroy(ctx *platformtest.Context) {
 	`)

 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
-	asnap := sendArgVersion(ctx, fs, "@a snap")
-	err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	asnap := sendArgVersion(fs, "@a snap")
+	err := zfs.ZFSBookmark(fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}
@@ -41,17 +41,17 @@ func IdempotentDestroy(ctx *platformtest.Context) {
 			log.Printf("SUBBEGIN testing idempotent destroy %q for path %q", c.description, c.path)

 			log.Println("destroy existing")
-			err = zfs.ZFSDestroy(ctx, c.path)
+			err = zfs.ZFSDestroy(c.path)
 			if err != nil {
 				panic(err)
 			}
 			log.Println("destroy again, non-idempotently, must error")
-			err = zfs.ZFSDestroy(ctx, c.path)
+			err = zfs.ZFSDestroy(c.path)
 			if _, ok := err.(*zfs.DatasetDoesNotExist); !ok {
 				panic(fmt.Sprintf("%T: %s", err, err))
 			}
 			log.Println("destroy again, idempotently, must not error")
-			err = zfs.ZFSDestroyIdempotent(ctx, c.path)
+			err = zfs.ZFSDestroyIdempotent(c.path)
 			if err != nil {
 				panic(err)
 			}
@@ -62,12 +62,12 @@ func IdempotentDestroy(ctx *platformtest.Context) {
 	}

 	// also test idempotent destroy for cases where the parent dataset does not exist
-	err = zfs.ZFSDestroyIdempotent(ctx, fmt.Sprintf("%s/not foo bar@nonexistent snapshot", ctx.RootDataset))
+	err = zfs.ZFSDestroyIdempotent(fmt.Sprintf("%s/not foo bar@nonexistent snapshot", ctx.RootDataset))
 	if err != nil {
 		panic(err)
 	}

-	err = zfs.ZFSDestroyIdempotent(ctx, fmt.Sprintf("%s/not foo bar#nonexistent bookmark", ctx.RootDataset))
+	err = zfs.ZFSDestroyIdempotent(fmt.Sprintf("%s/not foo bar#nonexistent bookmark", ctx.RootDataset))
 	if err != nil {
 		panic(err)
 	}
@@ -22,7 +22,7 @@ func IdempotentHold(ctx *platformtest.Context) {
 	`)

 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
-	v1 := sendArgVersion(ctx, fs, "@1")
+	v1 := sendArgVersion(fs, "@1")

 	tag := "zrepl_platformtest"
 	err := zfs.ZFSHold(ctx, fs, v1, tag)
@@ -5,7 +5,6 @@ import (

 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/platformtest"
 	"github.com/zrepl/zrepl/zfs"
@@ -32,7 +31,7 @@ func ReplicationCursor(ctx *platformtest.Context) {
 	}

 	fs := ds.ToString()
-	snap := sendArgVersion(ctx, fs, "@1 with space")
+	snap := sendArgVersion(fs, "@1 with space")

 	destroyed, err := endpoint.MoveReplicationCursor(ctx, fs, &snap, jobid)
 	if err != nil {
@@ -40,7 +39,7 @@ func ReplicationCursor(ctx *platformtest.Context) {
 	}
 	assert.Empty(ctx, destroyed)

-	snapProps, err := zfs.ZFSGetCreateTXGAndGuid(ctx, snap.FullPath(fs))
+	snapProps, err := zfs.ZFSGetCreateTXGAndGuid(snap.FullPath(fs))
 	if err != nil {
 		panic(err)
 	}
@@ -60,7 +59,7 @@ func ReplicationCursor(ctx *platformtest.Context) {
 	cursor1BookmarkName, err := endpoint.ReplicationCursorBookmarkName(fs, snap.GUID, jobid)
 	require.NoError(ctx, err)

-	snap2 := sendArgVersion(ctx, fs, "@2 with space")
+	snap2 := sendArgVersion(fs, "@2 with space")
 	destroyed, err = endpoint.MoveReplicationCursor(ctx, fs, &snap2, jobid)
 	require.NoError(ctx, err)
 	require.Equal(ctx, 1, len(destroyed))
@@ -40,7 +40,7 @@ func ResumableRecvAndTokenHandling(ctx *platformtest.Context) {
 		require.True(ctx, ok)

 		// we know that support on sendFS implies support on recvFS
-		// => assert that if we don't support resumed recv, the method returns ""
+		// => asser that if we don't support resumed recv, the method returns ""
 		tok, err := zfs.ZFSGetReceiveResumeTokenOrEmptyStringIfNotSupported(ctx, mustDatasetPath(recvFS))
 		check(err)
 		require.Equal(ctx, "", tok)
@@ -16,7 +16,7 @@ type resumeTokenTest struct {

 func (rtt *resumeTokenTest) Test(t *platformtest.Context) {

-	resumeSendSupported, err := zfs.ResumeSendSupported(t)
+	resumeSendSupported, err := zfs.ResumeSendSupported()
 	if err != nil {
 		t.Errorf("cannot determine whether resume supported: %T %s", err, err)
 		t.FailNow()
@@ -23,7 +23,7 @@ func SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden(ctx *platformt
 	`)

 	fs := fmt.Sprintf("%s/send er", ctx.RootDataset)
-	props := mustGetProps(ctx, fs+"@a snap")
+	props := mustGetProps(fs + "@a snap")

 	sendArgs := zfs.ZFSSendArgs{
 		FS: fs,
@@ -58,7 +58,7 @@ func SendArgsValidationResumeTokenEncryptionMismatchForbidden(ctx *platformtest.
 	if !supported {
 		ctx.SkipNow()
 	}
-	supported, err = zfs.ResumeSendSupported(ctx)
+	supported, err = zfs.ResumeSendSupported()
 	check(err)
 	if !supported {
 		ctx.SkipNow()
@@ -117,7 +117,7 @@ func SendArgsValidationResumeTokenEncryptionMismatchForbidden(ctx *platformtest.
 		require.Equal(ctx, mismatchError.What, zfs.ZFSSendArgsResumeTokenMismatchEncryptionNotSet)
 	}

-	// threat model: use of a crafted resume token that requests an encrypted send
+	// threat model: use of a crafted resume token that requests an encryped send
 	//               but send args require unencrypted send
 	{
 		var maliciousSend zfs.ZFSSendArgs = unencS.sendArgs
@@ -149,7 +149,7 @@ func SendArgsValidationResumeTokenDifferentFilesystemForbidden(ctx *platformtest
 	if !supported {
 		ctx.SkipNow()
 	}
-	supported, err = zfs.ResumeSendSupported(ctx)
+	supported, err = zfs.ResumeSendSupported()
 	check(err)
 	if !supported {
 		ctx.SkipNow()
@@ -20,7 +20,7 @@ func UndestroyableSnapshotParsing(t *platformtest.Context) {
 		R  zfs hold zrepl_platformtest "${ROOTDS}/foo bar@4 5 6"
 	`)

-	err := zfs.ZFSDestroy(t, fmt.Sprintf("%s/foo bar@1 2 3,4 5 6,7 8 9", t.RootDataset))
+	err := zfs.ZFSDestroy(fmt.Sprintf("%s/foo bar@1 2 3,4 5 6,7 8 9", t.RootDataset))
 	if err == nil {
 		panic("expecting destroy error due to hold")
 	}
@@ -35,7 +35,7 @@ The algorithm **ensures resumability** of the replication step in presence of
 * network failures at any time
 * other instances of this algorithm executing the same step in parallel (e.g. concurrent replication to different destinations)

-To accomplish this goal, the algorithm **assumes ownership of parts of the ZFS hold tag namespace and the bookmark namespace**:
+To accomplish this goal, the algorithm **assumes ownersip of parts of the ZFS hold tag namespace and the bookmark namespace**:
 * holds with prefix `zrepl_STEP` on any snapshot are reserved for zrepl
 * bookmarks with prefix `zrepl_STEP` are reserved for zrepl

@@ -55,7 +55,7 @@ The replication step (full `to` send or `from => to` send) is *complete* iff the
 Specifically, the algorithm may be invoked with the same `from` and `to` arguments, and potentially a `resume_token`, after a temporary (like network-related) failure:
 **Unless permanent errors occur, repeated invocations of the algorithm with updated resume token will converge monotonically (but not strictly monotonically) toward completion.**

-Note that the mere existence of `to` on the receiving side does not constitute completion, since there may still be post-recv actions to be performed on sender and receiver.
+Note that the mere existence of `to` on the receiving side does not constitue completion, since there may still be post-recv actions to be performed on sender and receiver.

 #### Job and Job ID
 This algorithm supports that *multiple* instance of it run in parallel on the *same* step (full `to` / `from => to` pair).
@@ -117,7 +117,7 @@ Recv-side: no-op
    # => doesn't work, because zfs recv is implemented as a `clone` internally, that's exactly what we want
    ```

- if recv-side `to` exists, goto cleanup-phase (no replication to do)
+- if recv-side `to` exists, goto cleaup-phase (no replication to do)
  
 - `to` cannot be destroyed while being received, because it isn't visible as a snapshot yet (it isn't yet one after all)

@@ -142,7 +142,7 @@ Network failures during replication can be recovered from using resumable send &
 - Network failure during the replication
  - send-side `from` and `to` are still present due to zfs holds
  - recv-side `from` is still present because the partial receive state prevents its destruction (see prepare-phase)
-  - if recv-side has a resume token, the resume token will continue to work on the sender because `from`s and `to` are still present
+  - if recv-side hasa resume token, the resume token will continue to work on the sender because `from`s and `to` are still present
 - Network failure at the end of the replication step stream transmission
  - Variant A: failure from the sender's perspective, success from the receiver's perspective
    - receive-side `to` doesn't have a hold and could be destroyed anytime
@@ -221,7 +221,7 @@ It builds a diff between the sender and receiver filesystem bookmarks+snapshots
 In case of conflict, the algorithm errors out with a conflict description that can be used to manually or automatically resolve the conflict.
 Otherwise, the algorithm builds a list of replication steps that are then worked on sequentially by the "Algorithm for a Single Replication Step".

-The algorithm ensures that a plan can be executed exactly as planned by acquiring appropriate zfs holds.
+The algorithm ensures that a plan can be executed exactly as planned by aquiring appropriate zfs holds.
 The algorithm can be configured to retry a plan when encountering non-permanent errors (e.g. network errors).
 However, permanent errors result in the plan being cancelled.

@@ -262,22 +262,22 @@ If fast-forward is not possible, produce a conflict description and ERROR OUT.<b
 TODOs:
 - make it configurable what snapshots are included in the list (i.e. every one we see, only most recent, at least one every X hours, ...)

-**Ensure that we will be able to carry out all steps** by acquiring holds or fsstep bookmarks on the sending side
+**Ensure that we will be able to carry out all steps** by aquiring holds or fsstep bookmarks on the sending side
 - `idempotent_hold([s.to for s in STEPS], zrepl_FS_J_${jobid})`
 - `if STEPS[0].from != nil: idempotent_FSSTEP_bookmark(STEPS[0].from, zrepl_FSSTEP_bm_G_${STEPS[0].from.guid}_J_${jobid})` 
  
 **Determine which steps have not been completed (`uncompleted_steps`)** (we might be in an second invocation of this algorithm after a network failure and some steps might already be done):
 - `res_tok := receiver.ResumeToken(fs)`
 - `rmrfsv := receiver.MostRecentFilesystemVersion(fs)`
- if `res_tok !=  nil`: ensure that `res_tok` has a corresponding step in `STEPS`, otherwise ERROR OUT
- if `rmrfsv != nil`: ensure that `res_tok` has a corresponding step in `STEPS`, otherwise ERROR OUT
+- if `res_tok !=  nil`: ensure that `res_tok` has a correspondinng step in `STEPS`, otherwise ERROR OUT
+- if `rmrfsv != nil`: ensure that `res_tok` has a correspondinng step in `STEPS`, otherwise ERROR OUT
 - if `(res_token != nil && rmrfsv != nil)`: ensure that `res_tok` is the subsequent step to the one we found for `rmrfsv`
 - if both are nil, we are at the beginning, `uncompleted_steps = STEPS` and goto next block
 - `rstep := if res_tok != nil { res_tok } else { rmrfsv }`
 - `uncompleted_steps := STEPS[find_step_idx(STEPS, rstep).expect("must exist, checked above"):]`
  - Note that we do not explicitly check for the completion of prior replication steps.
    All we care about is what needs to be done from `rstep`.
-  - This is intentional and necessary because we cumulatively release all holds and step bookmarks made for steps that precede a just-completed step (see next paragraph)
+  - This is intentional and necessary because we cummutatively release all holds and step bookmarks made for steps that preceed a just-completed step (see next paragraph)

 **Execute uncompleted steps**<br/>
 Invoke the "Algorithm for a Single Replication Step" for each step in `uncompleted_steps`.
@@ -89,6 +89,8 @@ type attempt struct {
 	// if both are nil, it must be assumed that Planner.Plan is active
 	planErr *timedError
 	fss     []*fs
+
+	concurrency int
 }

 type timedError struct {
@@ -170,11 +172,12 @@ type step struct {

 type ReportFunc func() *report.Report
 type WaitFunc func(block bool) (done bool)
+type SetConcurrencyFunc func(concurrency int) error

 var maxAttempts = envconst.Int64("ZREPL_REPLICATION_MAX_ATTEMPTS", 3)
 var reconnectHardFailTimeout = envconst.Duration("ZREPL_REPLICATION_RECONNECT_HARD_FAIL_TIMEOUT", 10*time.Minute)

-func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
+func Do(ctx context.Context, initialConcurrency int, planner Planner) (ReportFunc, WaitFunc, SetConcurrencyFunc) {
 	log := getLog(ctx)
 	l := chainlock.New()
 	run := &run{
@@ -182,6 +185,8 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
 		startedAt: time.Now(),
 	}

+	concurrencyChanges := make(chan concurrencyChange)
+
 	done := make(chan struct{})
 	go func() {
 		defer close(done)
@@ -198,15 +203,21 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
 			run.waitReconnect.SetZero()
 			run.waitReconnectError = nil

+			prevConcurrency := initialConcurrency // FIXME default concurrency
+			if prev != nil {
+				prevConcurrency = prev.concurrency
+			}
+
 			// do current attempt
 			cur := &attempt{
-				l:         l,
-				startedAt: time.Now(),
-				planner:   planner,
+				l:           l,
+				startedAt:   time.Now(),
+				planner:     planner,
+				concurrency: prevConcurrency,
 			}
 			run.attempts = append(run.attempts, cur)
 			run.l.DropWhile(func() {
-				cur.do(ctx, prev)
+				cur.do(ctx, prev, concurrencyChanges)
 			})
 			prev = cur
 			if ctx.Err() != nil {
@@ -277,10 +288,25 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
 		defer run.l.Lock().Unlock()
 		return run.report()
 	}
-	return report, wait
+	setConcurrency := func(concurrency int) (reterr error) {
+		var wg sync.WaitGroup
+		wg.Add(1)
+		concurrencyChanges <- concurrencyChange{ concurrency, func(err error) {
+			defer wg.Done()
+			reterr = err // shadow
+		}}
+		wg.Wait()
+		return reterr
+	}
+	return report, wait, setConcurrency
 }

-func (a *attempt) do(ctx context.Context, prev *attempt) {
+type concurrencyChange struct {
+	value          int
+	resultCallback func(error)
+}
+
+func (a *attempt) do(ctx context.Context, prev *attempt, setConcurrency <-chan concurrencyChange) {
 	pfss, err := a.planner.Plan(ctx)
 	errTime := time.Now()
 	defer a.l.Lock().Unlock()
@@ -352,10 +378,10 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 			}
 		}
 	}
-	// invariant: prevs contains an entry for each unambiguous correspondence
+	// invariant: prevs contains an entry for each unambigious correspondence

-	stepQueue := newStepQueue()
-	defer stepQueue.Start(1)() // TODO parallel replication
+	stepQueue := newStepQueue(a.concurrency)
+	defer stepQueue.Start()()
 	var fssesDone sync.WaitGroup
 	for _, f := range a.fss {
 		fssesDone.Add(1)
@@ -364,8 +390,27 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 			f.do(ctx, stepQueue, prevs[f])
 		}(f)
 	}
+	changeConcurrencyDone := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case change := <-setConcurrency:
+				err := stepQueue.SetConcurrency(change.value)
+				go change.resultCallback(err)
+				if err == nil {
+					a.l.Lock()
+					a.concurrency = change.value
+					a.l.Unlock()
+				}
+			case <-changeConcurrencyDone:
+				return
+				// not waiting for ctx.Done here, the main job are the fsses
+			}
+		}
+	}()
 	a.l.DropWhile(func() {
 		fssesDone.Wait()
+		close(changeConcurrencyDone)
 	})
 	a.finishedAt = time.Now()
 }
@@ -399,7 +444,7 @@ func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 		}
 		fs.planned.steps = append(fs.planned.steps, step)
 	}
-	debug("initial len(fs.planned.steps) = %d", len(fs.planned.steps))
+	debug("iniital len(fs.planned.steps) = %d", len(fs.planned.steps))

 	// for not-first attempts, only allow fs.planned.steps
 	// up to including the originally planned target snapshot
@@ -461,7 +506,8 @@ func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 		// lock must not be held while executing step in order for reporting to work
 		fs.l.DropWhile(func() {
 			targetDate := s.step.TargetDate()
-			defer pq.WaitReady(fs, targetDate)()
+			ctx, done := pq.WaitReady(ctx, fs, targetDate)()
+			defer done()
 			err = s.step.Step(ctx) // no shadow
 			errTime = time.Now()   // no shadow
 		})
@@ -482,6 +528,7 @@ func (r *run) report() *report.Report {
 		WaitReconnectSince: r.waitReconnect.begin,
 		WaitReconnectUntil: r.waitReconnect.end,
 		WaitReconnectError: r.waitReconnectError.IntoReportError(),
+		// Concurrency:        r.concurrency,
 	}
 	for i := range report.Attempts {
 		report.Attempts[i] = r.attempts[i].report()
@@ -8,12 +8,12 @@ import (

 type Logger = logger.Logger

-type contextKey int
+type contexKey int

-const contextKeyLogger contextKey = iota + 1
+const contexKeyLogger contexKey = iota + 1

 func getLog(ctx context.Context) Logger {
-	l, ok := ctx.Value(contextKeyLogger).(Logger)
+	l, ok := ctx.Value(contexKeyLogger).(Logger)
 	if !ok {
 		l = logger.NewNullLogger()
 	}
@@ -21,5 +21,5 @@ func getLog(ctx context.Context) Logger {
 }

 func WithLogger(ctx context.Context, log Logger) context.Context {
-	return context.WithValue(ctx, contextKeyLogger, log)
+	return context.WithValue(ctx, contexKeyLogger, log)
 }
@@ -151,7 +151,7 @@ func TestReplication(t *testing.T) {
 	ctx := context.Background()

 	mp := &mockPlanner{}
-	getReport, wait := Do(ctx, mp)
+	getReport, wait, _ := Do(ctx, 1, mp)
 	begin := time.Now()
 	fireAt := []time.Duration{
 		// the following values are relative to the start
@@ -174,7 +174,7 @@ func TestReplication(t *testing.T) {
 	waitBegin := time.Now()
 	wait(true)
 	waitDuration := time.Since(waitBegin)
-	assert.True(t, waitDuration < 10*time.Millisecond, "%v", waitDuration) // and that's gracious
+	assert.True(t, waitDuration < 10*time.Millisecond, "%v", waitDuration) // and that's gratious

 	prev, err := json.Marshal(reports[0])
 	require.NoError(t, err)
@@ -2,6 +2,8 @@ package driver

 import (
 	"container/heap"
+	"fmt"
+	"sync"
 	"time"

 	"github.com/zrepl/zrepl/util/chainlock"
@@ -11,51 +13,88 @@ type stepQueueRec struct {
 	ident      interface{}
 	targetDate time.Time
 	wakeup     chan StepCompletedFunc
+	cancelDueToConcurrencyDownsize interace{}
 }

 type stepQueue struct {
 	stop chan struct{}
 	reqs chan stepQueueRec
+
+	// l protects all members except the channels above
+
+	l           *chainlock.L
+	pendingCond *sync.Cond
+
+	// ident => queueItem
+	pending    *stepQueueHeap
+	active     *stepQueueHeap
+	queueItems map[interface{}]*stepQueueHeapItem // for tracking used idents in both pending and active
+
+	// stopped is used for cancellation of "wake" goroutine
+	stopped bool
+
+	concurrency int
 }

 type stepQueueHeapItem struct {
 	idx int
-	req stepQueueRec
+	req *stepQueueRec
+}
+type stepQueueHeap struct {
+	items   []*stepQueueHeapItem
+	reverse bool // never change after pushing first element
 }
-type stepQueueHeap []*stepQueueHeapItem

 func (h stepQueueHeap) Less(i, j int) bool {
-	return h[i].req.targetDate.Before(h[j].req.targetDate)
+	res := h.items[i].req.targetDate.Before(h.items[j].req.targetDate)
+	if h.reverse {
+		return !res
+	}
+	return res
 }

 func (h stepQueueHeap) Swap(i, j int) {
-	h[i], h[j] = h[j], h[i]
-	h[i].idx = i
-	h[j].idx = j
+	h.items[i], h.items[j] = h.items[j], h.items[i]
+	h.items[i].idx = i
+	h.items[j].idx = j
 }

 func (h stepQueueHeap) Len() int {
-	return len(h)
+	return len(h.items)
 }

 func (h *stepQueueHeap) Push(elem interface{}) {
 	hitem := elem.(*stepQueueHeapItem)
 	hitem.idx = h.Len()
-	*h = append(*h, hitem)
+	h.items = append(h.items, hitem)
 }

 func (h *stepQueueHeap) Pop() interface{} {
-	elem := (*h)[h.Len()-1]
+	elem := h.items[h.Len()-1]
 	elem.idx = -1
-	*h = (*h)[:h.Len()-1]
+	h.items = h.items[:h.Len()-1]
 	return elem
 }

 // returned stepQueue must be closed with method Close
-func newStepQueue() *stepQueue {
+func newStepQueue(concurrency int) *stepQueue {
+	l := chainlock.New()
 	q := &stepQueue{
-		stop: make(chan struct{}),
-		reqs: make(chan stepQueueRec),
+		stop:        make(chan struct{}),
+		reqs:        make(chan stepQueueRec),
+		l:           l,
+		pendingCond: l.NewCond(),
+		// priority queue
+		pending: &stepQueueHeap{reverse: false},
+		active:  &stepQueueHeap{reverse: true},
+		// ident => queueItem
+		queueItems: make(map[interface{}]*stepQueueHeapItem),
+		// stopped is used for cancellation of "wake" goroutine
+		stopped: false,
+	}
+	err := q.setConcurrencyLocked(concurrency)
+	if err != nil {
+		panic(err)
 	}
 	return q
 }
@@ -65,25 +104,12 @@ func newStepQueue() *stepQueue {
 //
 // No WaitReady calls must be active at the time done is called
 // The behavior of calling WaitReady after done was called is undefined
-func (q *stepQueue) Start(concurrency int) (done func()) {
-	if concurrency < 1 {
-		panic("concurrency must be >= 1")
-	}
-	// l protects pending and queueItems
-	l := chainlock.New()
-	pendingCond := l.NewCond()
-	// priority queue
-	pending := &stepQueueHeap{}
-	// ident => queueItem
-	queueItems := make(map[interface{}]*stepQueueHeapItem)
-	// stopped is used for cancellation of "wake" goroutine
-	stopped := false
-	active := 0
+func (q *stepQueue) Start() (done func()) {
 	go func() { // "stopper" goroutine
 		<-q.stop
-		defer l.Lock().Unlock()
-		stopped = true
-		pendingCond.Broadcast()
+		defer q.l.Lock().Unlock()
+		q.stopped = true
+		q.pendingCond.Broadcast()
 	}()
 	go func() { // "reqs" goroutine
 		for {
@@ -97,41 +123,52 @@ func (q *stepQueue) Start(concurrency int) (done func()) {
 				}
 			case req := <-q.reqs:
 				func() {
-					defer l.Lock().Unlock()
-					if _, ok := queueItems[req.ident]; ok {
+					defer q.l.Lock().Unlock()
+					if _, ok := q.queueItems[req.ident]; ok {
 						panic("WaitReady must not be called twice for the same ident")
 					}
 					qitem := &stepQueueHeapItem{
 						req: req,
 					}
-					queueItems[req.ident] = qitem
-					heap.Push(pending, qitem)
-					pendingCond.Broadcast()
+					q.queueItems[req.ident] = qitem
+					heap.Push(q.pending, qitem)
+					q.pendingCond.Broadcast()
 				}()
 			}
 		}
 	}()
 	go func() { // "wake" goroutine
-		defer l.Lock().Unlock()
+		defer q.l.Lock().Unlock()
 		for {

-			for !stopped && (active >= concurrency || pending.Len() == 0) {
-				pendingCond.Wait()
+			for !q.stopped && (q.active.Len() >= q.concurrency || q.pending.Len() == 0) {
+				q.pendingCond.Wait()
 			}
-			if stopped {
+			if q.stopped {
 				return
 			}
-			if pending.Len() <= 0 {
+			if q.pending.Len() <= 0 {
 				return
 			}
-			active++
-			next := heap.Pop(pending).(*stepQueueHeapItem).req
-			delete(queueItems, next.ident)

-			next.wakeup <- func() {
-				defer l.Lock().Unlock()
-				active--
-				pendingCond.Broadcast()
+			// pop from tracked items
+			next := heap.Pop(q.pending).(*stepQueueHeapItem)
+
+			next.req.cancelDueToConcurrencyDownsize = 
+
+			heap.Push(q.active, next)
+
+			next.req.wakeup <- func() {
+				defer q.l.Lock().Unlock()
+
+				//
+				qitem := &stepQueueHeapItem{
+					req: req,
+				}
+
+				// delete(q.queueItems, next.req.ident) // def
+
+				q.pendingCond.Broadcast()
 			}
 		}
 	}()
@@ -161,3 +198,24 @@ func (q *stepQueue) WaitReady(ident interface{}, targetDate time.Time) StepCompl
 	}
 	return q.sendAndWaitForWakeup(ident, targetDate)
 }
+
+// caller must hold lock
+func (q *stepQueue) setConcurrencyLocked(newConcurrency int) error {
+	if !(newConcurrency >= 1) {
+		return fmt.Errorf("concurrency must be >= 1 but requested %v", newConcurrency)
+	}
+	q.concurrency = newConcurrency
+	q.pendingCond.Broadcast() // wake up waiters who could make progress
+
+	for q.active.Len() > q.concurrency {
+		item := heap.Pop(q.active).(*stepQueueHeapItem)
+		item.req.cancelDueToConcurrencyDownsize()
+		heap.Push(q.pending, item)
+	}
+	return nil
+}
+
+func (q *stepQueue) SetConcurrency(new int) error {
+	defer q.l.Lock().Unlock()
+	return q.setConcurrencyLocked(new)
+}
@@ -14,10 +14,10 @@ import (
 )

 // FIXME: this test relies on timing and is thus rather flaky
-// (relies on scheduler responsiveness of < 500ms)
+// (relies on scheduler responsivity of < 500ms)
 func TestPqNotconcurrent(t *testing.T) {
 	var ctr uint32
-	q := newStepQueue()
+	q := newStepQueue(1)
 	var wg sync.WaitGroup
 	wg.Add(4)
 	go func() {
@@ -29,7 +29,7 @@ func TestPqNotconcurrent(t *testing.T) {
 	}()

 	// give goroutine "1" 500ms to enter queue, get the active slot and enter time.Sleep
-	defer q.Start(1)()
+	defer q.Start()()
 	time.Sleep(500 * time.Millisecond)

 	// while "1" is still running, queue in "2", "3" and "4"
@@ -77,8 +77,9 @@ func (r record) String() string {
 // Hence, perform some statistics on the wakeup times and assert that the mean wakeup
 // times for each step are close together.
 func TestPqConcurrent(t *testing.T) {
-
-	q := newStepQueue()
+	
+	concurrency := 5
+	q := newStepQueue(concurrency)
 	var wg sync.WaitGroup
 	filesystems := 100
 	stepsPerFS := 20
@@ -104,8 +105,7 @@ func TestPqConcurrent(t *testing.T) {
 			records <- recs
 		}(fs)
 	}
-	concurrency := 5
-	defer q.Start(concurrency)()
+	defer q.Start()()
 	wg.Wait()
 	close(records)
 	t.Logf("loop done")
@@ -17,7 +17,7 @@ func fsvlist(fsv ...string) (r []*FilesystemVersion) {
 	r = make([]*FilesystemVersion, len(fsv))
 	for i, f := range fsv {

-		// parse the id from fsvlist. it is used to derive Guid,CreateTXG and Creation attrs
+		// parse the id from fsvlist. it is used to derivce Guid,CreateTXG and Creation attrs
 		split := strings.Split(f, ",")
 		if len(split) != 2 {
 			panic("invalid fsv spec")
@@ -114,7 +114,7 @@ func TestIncrementalPath_BookmarkSupport(t *testing.T) {
 		assert.Equal(t, l("#a,1", "@b,2"), path)
 	})

-	// bookmarks are stripped from IncrementalPath (cannot send incrementally)
+	// boomarks are stripped from IncrementalPath (cannot send incrementally)
 	doTest(l("@a,1"), l("#a,1", "#b,2", "@c,3"), func(path []*FilesystemVersion, conflict error) {
 		assert.Equal(t, l("#a,1", "@c,3"), path)
 	})
@@ -91,7 +91,7 @@ message ReceiveReq {
  string Filesystem = 1;
  FilesystemVersion To = 2;

-  // If true, the receiver should clear the resume token before performing the
+  // If true, the receiver should clear the resume token before perfoming the
  // zfs recv of the stream in the request
  bool ClearResumeToken = 3;
 }
@@ -262,6 +262,7 @@ func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
 		return nil, err
 	}
 	sfss := slfssres.GetFilesystems()
+	// no progress here since we could run in a live-lock on connectivity issues

 	rlfssres, err := p.receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
 	if err != nil {
@@ -352,7 +353,7 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 	}

 	// give both sides a hint about how far the replication got
-	// This serves as a cumulative variant of SendCompleted and can be useful
+	// This serves as a cummulative variant of SendCompleted and can be useful
 	// for example to release stale holds from an earlier (interrupted) replication.
 	// TODO FIXME: enqueue this as a replication step instead of doing it here during planning
 	//             then again, the step should run regardless of planning success
@@ -8,6 +8,6 @@ import (
 	"github.com/zrepl/zrepl/replication/driver"
 )

-func Do(ctx context.Context, planner driver.Planner) (driver.ReportFunc, driver.WaitFunc) {
-	return driver.Do(ctx, planner)
+func Do(ctx context.Context, initialConcurrency int, planner driver.Planner) (driver.ReportFunc, driver.WaitFunc, driver.SetConcurrencyFunc) {
+	return driver.Do(ctx, initialConcurrency, planner)
 }
@@ -10,6 +10,7 @@ type Report struct {
 	WaitReconnectSince, WaitReconnectUntil time.Time
 	WaitReconnectError                     *TimedError
 	Attempts                               []*AttemptReport
+	Concurrency                            int
 }

 var _, _ = json.Marshal(&Report{})
@@ -52,7 +52,7 @@ func (s *streamCopier) WriteStreamTo(w io.Writer) zfs.StreamCopierError {
 	s.mtx.Lock()
 	defer s.mtx.Unlock()
 	if s.used {
-		panic("streamCopier used multiple times")
+		panic("streamCopier used mulitple times")
 	}
 	s.used = true
 	return s.streamConn.ReadStreamInto(w, ZFSStream)
@@ -100,7 +100,7 @@ func (c *Conn) ReadFrame() (Frame, error) {
 		return Frame{}, ErrShutdown
 	}

-	// only acquire readMtx now to prioritize the draining in Shutdown()
+	// only aquire readMtx now to prioritize the draining in Shutdown()
 	// over external callers (= drain public callers)

 	c.readMtx.Lock()
@@ -148,7 +148,7 @@ func (c *Conn) readFrame() (Frame, error) {
 		//                | |         |           |
 		//                | |         |           F3
 		//                | |         |
-		//                | F2        |significant time between frames because
+		//                | F2        |signficant time between frames because
 		//                F1           the peer has nothing to say to us
 		//
 		// Assume we're at the point were F2's header is in c.readNext.
@@ -246,7 +246,7 @@ func (c *Conn) Shutdown(deadline time.Time) error {
 	//
 	// 1. Naive Option: We just call Close() right after CloseWrite.
 	// This yields the same race condition as explained above (DIF, first
-	// paragraph): The situation just became a little more unlikely because
+	// paragraph): The situation just becomae a little more unlikely because
 	// our rstFrameType + CloseWrite dance gave the client a full RTT worth of
 	// time to read the data from its TCP recv buffer.
 	//
@@ -295,7 +295,7 @@ func (c *Conn) Shutdown(deadline time.Time) error {
 	defer prometheus.NewTimer(prom.ShutdownSeconds).ObserveDuration()

 	closeWire := func(step string) error {
-		// TODO SetLinger(0) or similar (we want RST frames here, not FINS)
+		// TODO SetLinger(0) or similiar (we want RST frames here, not FINS)
 		closeErr := c.nc.Close()
 		if closeErr == nil {
 			return nil
@@ -321,10 +321,10 @@ func (c *Conn) Shutdown(deadline time.Time) error {

 	c.shutdown.Begin()
 	// new calls to c.ReadFrame and c.WriteFrame will now return ErrShutdown
-	// Acquiring writeMtx and readMtx afterwards ensures that already-running calls exit successfully
+	// Aquiring writeMtx and readMtx afterwards ensures that already-running calls exit successfully

 	// disable renewing timeouts now, enforce the requested deadline instead
-	// we need to do this before acquiring locks to enforce the timeout on slow
+	// we need to do this before aquiring locks to enforce the timeout on slow
 	// clients / if something hangs (DoS mitigation)
 	if err := c.nc.DisableTimeouts(); err != nil {
 		return hardclose(err, "disable_timeouts")
@@ -1,4 +1,4 @@
-// microbenchmark to manually test rpc/dataconn performance
+// microbenchmark to manually test rpc/dataconn perforamnce
 //
 // With stdin / stdout on client and server, simulating zfs send|recv piping
 //
@@ -6,7 +6,7 @@
 //   ./microbenchmark -appmode client -direction recv < /dev/zero
 //
 //
-// Without the overhead of pipes (just protocol performance, mostly useful with perf bc no bw measurement)
+// Without the overhead of pipes (just protocol perforamnce, mostly useful with perf bc no bw measurement)
 //
 //   ./microbenchmark -appmode client -direction recv -devnoopWriter -devnoopReader
 //   ./microbenchmark -appmode server -devnoopReader -devnoopWriter
@@ -49,7 +49,7 @@ type Wire interface {
 	// No data that could otherwise be Read is lost as a consequence of this call.
 	// The use case for this API is abortive connection shutdown.
 	// To provide any value over draining Wire using io.Read, an implementation
-	// will likely use out-of-band messaging mechanisms.
+	// will likely use out-of-bounds messaging mechanisms.
 	// TODO WaitForPeerClose() (supported bool, err error)
 }

@@ -119,7 +119,7 @@ restart:
 	return n, err
 }

-// Writes the given buffers to Conn, following the semantics of io.Copy,
+// Writes the given buffers to Conn, following the sematincs of io.Copy,
 // but is guaranteed to use the writev system call if the wrapped Wire
 // support it.
 // Note the Conn does not support writev through io.Copy(aConn, aNetBuffers).
@@ -158,9 +158,9 @@ var _ SyscallConner = (*net.TCPConn)(nil)
 // Think of io.ReadvFull, but for net.Buffers + using the readv syscall.
 //
 // If the underlying Wire is not a SyscallConner, a fallback
-// implementation based on repeated Conn.Read invocations is used.
+// ipmlementation based on repeated Conn.Read invocations is used.
 //
-// If the connection returned io.EOF, the number of bytes written until
+// If the connection returned io.EOF, the number of bytes up ritten until
 // then + io.EOF is returned. This behavior is different to io.ReadFull
 // which returns io.ErrUnexpectedEOF.
 func (c Conn) ReadvFull(buffers net.Buffers) (n int64, err error) {
@@ -99,7 +99,7 @@ func (c Conn) doOneReadv(rawConn syscall.RawConn, iovecs *[]syscall.Iovec) (n in
 				// Update left, cannot go below 0 due to
 				// a) definition of thisIovecConsumedCompletely
 				// b) left > 0 due to loop invariant
-				// Converting .Len to int64 is thus also safe now, because it is < left < INT_MAX
+				// Convertion .Len to int64 is thus also safe now, because it is < left < INT_MAX
 				left -= int((*iovecs)[0].Len)
 				*iovecs = (*iovecs)[1:]
 			} else {
@@ -56,7 +56,7 @@ func NewClientAuthListener(
 }

 // Accept() accepts a connection from the *net.TCPListener passed to the constructor
-// and sets up the TLS connection, including handshake and peer CommonName validation
+// and sets up the TLS connection, including handshake and peer CommmonName validation
 // within the specified handshakeTimeout.
 //
 // It returns both the raw TCP connection (tcpConn) and the TLS connection (tlsConn) on top of it.
@@ -121,6 +121,7 @@ func ClientAuthClient(serverName string, rootCA *x509.CertPool, clientCert tls.C
 		ServerName:   serverName,
 		KeyLogWriter: keylogFromEnv(),
 	}
+	tlsConfig.BuildNameToCertificate()
 	return tlsConfig, nil
 }

@@ -100,7 +100,7 @@ func (l *LocalListener) Accept(ctx context.Context) (*transport.AuthConn, error)
 			WithField("res.conn", res.conn).WithField("res.err", res.err).
 			Debug("responding to client request")

-		// contract between Connect and Accept is that Connect sends a req.callback
+		// contract bewteen Connect and Accept is that Connect sends a req.callback
 		// into which we can send one result non-blockingly.
 		// We want to panic if that contract is violated (impl error)
 		//
@@ -112,7 +112,7 @@ func (l *LocalListener) Accept(ctx context.Context) (*transport.AuthConn, error)
 		defer func() {
 			errv := recover()
 			if errv == clientCallbackBlocked {
-				// this would be a violation of contract between Connect and Accept, see above
+				// this would be a violation of contract betwee Connect and Accept, see above
 				panic(clientCallbackBlocked)
 			} else {
 				transport.GetLogger(ctx).WithField("recover_err", errv).
@@ -116,7 +116,7 @@ func (m *MultiStdinserverListener) Close() error {
 	return oneErr
 }

-// a single stdinserverListener (part of multiStdinserverListener)
+// a single stdinserverListener (part of multiStinserverListener)
 type stdinserverListener struct {
 	l              *netssh.Listener
 	clientIdentity string
@@ -32,7 +32,7 @@ func TLSListenerFactoryFromConfig(c *config.Global, in *config.TLSServe) (transp

 	serverCert, err := tls.LoadX509KeyPair(in.Cert, in.Key)
 	if err != nil {
-		return nil, errors.Wrap(err, "cannot parse cert/key pair")
+		return nil, errors.Wrap(err, "cannot parse cer/key pair")
 	}

 	clientCNs := make(map[string]struct{}, len(in.ClientCNs))
@@ -61,7 +61,7 @@ func ValidateClientIdentity(in string) (err error) {
 		return err
 	}
 	if path.Length() != 1 {
-		return errors.New("client identity must be a single path component (not empty, no '/')")
+		return errors.New("client identity must be a single path comonent (not empty, no '/')")
 	}
 	return nil
 }
@@ -7,7 +7,7 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )

-// StreamCopier wraps a zfs.StreamCopier, reimplementing
+// StreamCopier wraps a zfs.StreamCopier, reimplemening
 // its interface and counting the bytes written to during copying.
 type StreamCopier interface {
 	zfs.StreamCopier
@@ -20,7 +20,7 @@ func TestSemaphore(t *testing.T) {

 	sem := New(concurrentSemaphore)

-	var acquisitions struct {
+	var aquisitions struct {
 		beforeT, afterT uint32
 	}

@@ -33,9 +33,9 @@ func TestSemaphore(t *testing.T) {
 			require.NoError(t, err)
 			defer res.Release()
 			if time.Since(begin) > sleepTime {
-				atomic.AddUint32(&acquisitions.beforeT, 1)
+				atomic.AddUint32(&aquisitions.beforeT, 1)
 			} else {
-				atomic.AddUint32(&acquisitions.afterT, 1)
+				atomic.AddUint32(&aquisitions.afterT, 1)
 			}
 			time.Sleep(sleepTime)
 		}()
@@ -43,7 +43,7 @@ func TestSemaphore(t *testing.T) {

 	wg.Wait()

-	assert.True(t, acquisitions.beforeT == concurrentSemaphore)
-	assert.True(t, acquisitions.afterT == numGoroutines-concurrentSemaphore)
+	assert.True(t, aquisitions.beforeT == concurrentSemaphore)
+	assert.True(t, aquisitions.afterT == numGoroutines-concurrentSemaphore)

 }
@@ -39,7 +39,7 @@ type DatasetPathsVisitor func(v DatasetPathVisit) (visitChildTree bool)

 // Traverse a list of DatasetPaths top down, i.e. given a set of datasets with same
 // path prefix, those with shorter prefix are traversed first.
-// If there are gaps, i.e. the intermediary component a/b between a and a/b/c,
+// If there are gaps, i.e. the intermediary component a/b bewtween a and a/b/c,
 // those gaps are still visited but the FilledIn property of the visit is set to true.
 func (f *DatasetPathForest) WalkTopDown(visitor DatasetPathsVisitor) {

@@ -56,7 +56,7 @@ func TestDatasetPathForestWalkTopDown(t *testing.T) {

 	buildForest(paths).WalkTopDown(v)

-	expectedVisits := []DatasetPathVisit{
+	expectedVisists := []DatasetPathVisit{
 		{toDatasetPath("pool1"), false},
 		{toDatasetPath("pool1/foo"), true},
 		{toDatasetPath("pool1/foo/bar"), false},
@@ -65,7 +65,7 @@ func TestDatasetPathForestWalkTopDown(t *testing.T) {
 		{toDatasetPath("pool2/test"), true},
 		{toDatasetPath("pool2/test/bar"), false},
 	}
-	assert.Equal(t, expectedVisits, rec.visits)
+	assert.Equal(t, expectedVisists, rec.visits)

 }

@@ -82,7 +82,7 @@ func TestDatasetPathWalkTopDownWorksUnordered(t *testing.T) {

 	buildForest(paths).WalkTopDown(v)

-	expectedVisits := []DatasetPathVisit{
+	expectedVisists := []DatasetPathVisit{
 		{toDatasetPath("pool1"), false},
 		{toDatasetPath("pool1/foo"), true},
 		{toDatasetPath("pool1/foo/bar"), false},
@@ -91,6 +91,6 @@ func TestDatasetPathWalkTopDownWorksUnordered(t *testing.T) {
 		{toDatasetPath("pool1/bang/baz"), false},
 	}

-	assert.Equal(t, expectedVisits, rec.visits)
+	assert.Equal(t, expectedVisists, rec.visits)

 }
@@ -10,7 +10,6 @@ import (
 	"github.com/pkg/errors"

 	"github.com/zrepl/zrepl/util/envconst"
-	"github.com/zrepl/zrepl/zfs/zfscmd"
 )

 var encryptionCLISupport struct {
@@ -22,7 +21,7 @@ var encryptionCLISupport struct {
 func EncryptionCLISupported(ctx context.Context) (bool, error) {
 	encryptionCLISupport.once.Do(func() {
 		// "feature discovery"
-		cmd := zfscmd.CommandContext(ctx, "zfs", "load-key")
+		cmd := exec.Command("zfs", "load-key")
 		output, err := cmd.CombinedOutput()
 		if ee, ok := err.(*exec.ExitError); !ok || ok && !ee.Exited() {
 			encryptionCLISupport.err = errors.Wrap(err, "native encryption cli support feature check failed")
@@ -51,7 +50,7 @@ func ZFSGetEncryptionEnabled(ctx context.Context, fs string) (enabled bool, err
 		return false, err
 	}

-	props, err := zfsGet(ctx, fs, []string{"encryption"}, sourceAny)
+	props, err := zfsGet(fs, []string{"encryption"}, sourceAny)
 	if err != nil {
 		return false, errors.Wrap(err, "cannot get `encryption` property")
 	}
@@ -6,6 +6,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"os/exec"
 	"sort"
 	"strconv"
 	"strings"
@@ -14,7 +15,6 @@ import (
 	"github.com/pkg/errors"

 	"github.com/zrepl/zrepl/util/envconst"
-	"github.com/zrepl/zrepl/zfs/zfscmd"
 )

 // no need for feature tests, holds have been around forever
@@ -48,7 +48,7 @@ func ZFSHold(ctx context.Context, fs string, v ZFSSendArgVersion, tag string) er
 		return err
 	}
 	fullPath := v.FullPath(fs)
-	output, err := zfscmd.CommandContext(ctx, "zfs", "hold", tag, fullPath).CombinedOutput()
+	output, err := exec.CommandContext(ctx, "zfs", "hold", tag, fullPath).CombinedOutput()
 	if err != nil {
 		if bytes.Contains(output, []byte("tag already exists on this dataset")) {
 			goto success
@@ -67,7 +67,7 @@ func ZFSHolds(ctx context.Context, fs, snap string) ([]string, error) {
 		return nil, fmt.Errorf("`snap` must not be empty")
 	}
 	dp := fmt.Sprintf("%s@%s", fs, snap)
-	output, err := zfscmd.CommandContext(ctx, "zfs", "holds", "-H", dp).CombinedOutput()
+	output, err := exec.CommandContext(ctx, "zfs", "holds", "-H", dp).CombinedOutput()
 	if err != nil {
 		return nil, &ZFSError{output, errors.Wrap(err, "zfs holds failed")}
 	}
@@ -104,13 +104,11 @@ func ZFSRelease(ctx context.Context, tag string, snaps ...string) error {
 		}
 		args := []string{"release", tag}
 		args = append(args, snaps[i:j]...)
-		output, err := zfscmd.CommandContext(ctx, "zfs", args...).CombinedOutput()
+		output, err := exec.CommandContext(ctx, "zfs", args...).CombinedOutput()
 		if pe, ok := err.(*os.PathError); err != nil && ok && pe.Err == syscall.E2BIG {
 			maxInvocationLen = maxInvocationLen / 2
 			continue
 		}
-		// further error handling part of error scraper below
-
 		maxInvocationLen = maxInvocationLen + os.Getpagesize()
 		i = j

@@ -168,7 +166,7 @@ func doZFSReleaseAllOlderAndIncOrExcludingGUID(ctx context.Context, fs string, s
 		return fmt.Errorf("`tag` must not be empty`")
 	}

-	output, err := zfscmd.CommandContext(ctx,
+	output, err := exec.CommandContext(ctx,
 		"zfs", "list", "-o", "type,name,createtxg,guid,userrefs",
 		"-H", "-t", "snapshot,bookmark", "-r", "-d", "1", fs).CombinedOutput()
 	if err != nil {
@@ -268,7 +266,7 @@ func doZFSReleaseAllOlderAndIncOrExcludingGUIDFindSnapshots(snapOrBookmarkGuid u
 				case EntityTypeBookmark:
 					return 1
 				default:
-					panic("unexpected entity type " + t.String())
+					panic("unepxected entity type " + t.String())
 				}
 			}
 			return iET(lines[i].entityType) < iET(lines[j].entityType)
@@ -292,7 +290,7 @@ func doZFSReleaseAllOlderAndIncOrExcludingGUIDFindSnapshots(snapOrBookmarkGuid u
 		}
 		if foundGuid {
 			// The secondary key in sorting (snap < bookmark) guarantees that we
-			//   A) either found the snapshot with snapOrBookmarkGuid
+			//   A) either found the snapshot with snapOrBoomkarkGuid
 			//   B) or no snapshot with snapGuid exists, but one or more bookmarks of it exists
 			// In the case of A, we already added the snapshot to releaseSnaps if includeGuid requests it,
 			// and can ignore possible subsequent bookmarks of the snapshot.
@@ -13,7 +13,7 @@ func TestDoZFSReleaseAllOlderAndIncOrExcludingGUIDFindSnapshots(t *testing.T) {
 	// what we test here: sort bookmark #3 before @3
 	// => assert that the function doesn't stop at the first guid match
 	//    (which might be a bookmark, depending on zfs list ordering)
-	//    but instead considers the entire stride of bookmarks and snapshots with that guid
+	//    but instead considers the entire stride of boomarks and snapshots with that guid
 	//
 	// also, throw in unordered createtxg for good measure
 	list, err := doZFSReleaseAllOlderAndIncOrExcludingGUIDParseListOutput(
@@ -1,12 +1,11 @@
 package zfs

 import (
-	"context"
+	"bytes"
 	"crypto/sha512"
 	"encoding/hex"
 	"fmt"
-
-	"github.com/zrepl/zrepl/zfs/zfscmd"
+	"os/exec"
 )

 const (
@@ -62,10 +61,10 @@ type FilesystemPlaceholderState struct {
 // is a placeholder. Note that the property source must be `local` for the returned value to be valid.
 //
 // For nonexistent FS, err == nil and state.FSExists == false
-func ZFSGetFilesystemPlaceholderState(ctx context.Context, p *DatasetPath) (state *FilesystemPlaceholderState, err error) {
+func ZFSGetFilesystemPlaceholderState(p *DatasetPath) (state *FilesystemPlaceholderState, err error) {
 	state = &FilesystemPlaceholderState{FS: p.ToString()}
 	state.FS = p.ToString()
-	props, err := zfsGet(ctx, p.ToString(), []string{PlaceholderPropertyName}, sourceLocal)
+	props, err := zfsGet(p.ToString(), []string{PlaceholderPropertyName}, sourceLocal)
 	var _ error = (*DatasetDoesNotExist)(nil) // weak assertion on zfsGet's interface
 	if _, ok := err.(*DatasetDoesNotExist); ok {
 		return state, nil
@@ -78,19 +77,25 @@ func ZFSGetFilesystemPlaceholderState(ctx context.Context, p *DatasetPath) (stat
 	return state, nil
 }

-func ZFSCreatePlaceholderFilesystem(ctx context.Context, p *DatasetPath) (err error) {
+func ZFSCreatePlaceholderFilesystem(p *DatasetPath) (err error) {
 	if p.Length() == 1 {
 		return fmt.Errorf("cannot create %q: pools cannot be created with zfs create", p.ToString())
 	}
-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "create",
+	cmd := exec.Command(ZFS_BINARY, "create",
 		"-o", fmt.Sprintf("%s=%s", PlaceholderPropertyName, placeholderPropertyOn),
 		"-o", "mountpoint=none",
 		p.ToString())

-	stdio, err := cmd.CombinedOutput()
-	if err != nil {
+	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
+	cmd.Stderr = stderr
+
+	if err = cmd.Start(); err != nil {
+		return err
+	}
+
+	if err = cmd.Wait(); err != nil {
 		err = &ZFSError{
-			Stderr:  stdio,
+			Stderr:  stderr.Bytes(),
 			WaitErr: err,
 		}
 	}
@@ -98,14 +103,14 @@ func ZFSCreatePlaceholderFilesystem(ctx context.Context, p *DatasetPath) (err er
 	return
 }

-func ZFSSetPlaceholder(ctx context.Context, p *DatasetPath, isPlaceholder bool) error {
+func ZFSSetPlaceholder(p *DatasetPath, isPlaceholder bool) error {
 	props := NewZFSProperties()
 	prop := placeholderPropertyOff
 	if isPlaceholder {
 		prop = placeholderPropertyOn
 	}
 	props.Set(PlaceholderPropertyName, prop)
-	return zfsSet(ctx, p.ToString(), props)
+	return zfsSet(p.ToString(), props)
 }

 type MigrateHashBasedPlaceholderReport struct {
@@ -114,8 +119,8 @@ type MigrateHashBasedPlaceholderReport struct {
 }

 // fs must exist, will panic otherwise
-func ZFSMigrateHashBasedPlaceholderToCurrent(ctx context.Context, fs *DatasetPath, dryRun bool) (*MigrateHashBasedPlaceholderReport, error) {
-	st, err := ZFSGetFilesystemPlaceholderState(ctx, fs)
+func ZFSMigrateHashBasedPlaceholderToCurrent(fs *DatasetPath, dryRun bool) (*MigrateHashBasedPlaceholderReport, error) {
+	st, err := ZFSGetFilesystemPlaceholderState(fs)
 	if err != nil {
 		return nil, fmt.Errorf("error getting placeholder state: %s", err)
 	}
@@ -132,7 +137,7 @@ func ZFSMigrateHashBasedPlaceholderToCurrent(ctx context.Context, fs *DatasetPat
 		return &report, nil
 	}

-	err = ZFSSetPlaceholder(ctx, fs, st.IsPlaceholder)
+	err = ZFSSetPlaceholder(fs, st.IsPlaceholder)
 	if err != nil {
 		return nil, fmt.Errorf("error re-writing placeholder property: %s", err)
 	}
@@ -13,10 +13,9 @@ import (
 	"github.com/pkg/errors"

 	"github.com/zrepl/zrepl/util/envconst"
-	"github.com/zrepl/zrepl/zfs/zfscmd"
 )

-// NOTE: Update ZFSSendARgs.Validate when changing fields (potentially SECURITY SENSITIVE)
+// NOTE: Update ZFSSendARgs.Validate when changning fields (potentially SECURITY SENSITIVE)
 type ResumeToken struct {
 	HasFromGUID, HasToGUID    bool
 	FromGUID, ToGUID          uint64
@@ -26,7 +25,6 @@ type ResumeToken struct {
 }

 var resumeTokenNVListRE = regexp.MustCompile(`\t(\S+) = (.*)`)
-
 var resumeTokenContentsRE = regexp.MustCompile(`resume token contents:\nnvlist version: 0`)
 var resumeTokenIsCorruptRE = regexp.MustCompile(`resume token is corrupt`)

@@ -40,10 +38,10 @@ var resumeSendSupportedCheck struct {
 	err       error
 }

-func ResumeSendSupported(ctx context.Context) (bool, error) {
+func ResumeSendSupported() (bool, error) {
 	resumeSendSupportedCheck.once.Do(func() {
 		// "feature discovery"
-		cmd := zfscmd.CommandContext(ctx, "zfs", "send")
+		cmd := exec.Command("zfs", "send")
 		output, err := cmd.CombinedOutput()
 		if ee, ok := err.(*exec.ExitError); !ok || ok && !ee.Exited() {
 			resumeSendSupportedCheck.err = errors.Wrap(err, "resumable send cli support feature check failed")
@@ -87,7 +85,7 @@ func ResumeRecvSupported(ctx context.Context, fs *DatasetPath) (bool, error) {
 	}

 	if !sup.flagSupport.checked {
-		output, err := zfscmd.CommandContext(ctx, ZFS_BINARY, "receive").CombinedOutput()
+		output, err := exec.CommandContext(ctx, "zfs", "receive").CombinedOutput()
 		upgradeWhile(func() {
 			sup.flagSupport.checked = true
 			if ee, ok := err.(*exec.ExitError); err != nil && (!ok || ok && !ee.Exited()) {
@@ -125,7 +123,7 @@ func ResumeRecvSupported(ctx context.Context, fs *DatasetPath) (bool, error) {
 	if poolSup, ok = sup.poolSupported[pool]; !ok || // shadow
 		(!poolSup.supported && time.Since(poolSup.lastCheck) > resumeRecvPoolSupportRecheckTimeout) {

-		output, err := zfscmd.CommandContext(ctx, "zpool", "get", "-H", "-p", "-o", "value", "feature@extensible_dataset", pool).CombinedOutput()
+		output, err := exec.CommandContext(ctx, "zpool", "get", "-H", "-p", "-o", "value", "feature@extensible_dataset", pool).CombinedOutput()
 		if err != nil {
 			debug("resume recv pool support check result: %#v", sup.flagSupport)
 			poolSup.supported = false
@@ -156,7 +154,7 @@ func ResumeRecvSupported(ctx context.Context, fs *DatasetPath) (bool, error) {
 // FIXME: implement nvlist unpacking in Go and read through libzfs_sendrecv.c
 func ParseResumeToken(ctx context.Context, token string) (*ResumeToken, error) {

-	if supported, err := ResumeSendSupported(ctx); err != nil {
+	if supported, err := ResumeSendSupported(); err != nil {
 		return nil, err
 	} else if !supported {
 		return nil, ResumeTokenDecodingNotSupported
@@ -182,7 +180,7 @@ func ParseResumeToken(ctx context.Context, token string) (*ResumeToken, error) {
 	//	toname = pool1/test@b
 	//cannot resume send: 'pool1/test@b' used in the initial send no longer exists

-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "send", "-nvt", string(token))
+	cmd := exec.CommandContext(ctx, ZFS_BINARY, "send", "-nvt", string(token))
 	output, err := cmd.CombinedOutput()
 	if err != nil {
 		if exitErr, ok := err.(*exec.ExitError); ok {
@@ -259,7 +257,7 @@ func ZFSGetReceiveResumeTokenOrEmptyStringIfNotSupported(ctx context.Context, fs
 		return "", nil
 	}
 	const prop_receive_resume_token = "receive_resume_token"
-	props, err := ZFSGet(ctx, fs, []string{prop_receive_resume_token})
+	props, err := ZFSGet(fs, []string{prop_receive_resume_token})
 	if err != nil {
 		return "", err
 	}
@@ -11,10 +11,9 @@ import (
 	"syscall"

 	"github.com/zrepl/zrepl/util/envconst"
-	"github.com/zrepl/zrepl/zfs/zfscmd"
 )

-func ZFSDestroyFilesystemVersion(ctx context.Context, filesystem *DatasetPath, version *FilesystemVersion) (err error) {
+func ZFSDestroyFilesystemVersion(filesystem *DatasetPath, version *FilesystemVersion) (err error) {

 	datasetPath := version.ToAbsPath(filesystem)

@@ -23,7 +22,7 @@ func ZFSDestroyFilesystemVersion(ctx context.Context, filesystem *DatasetPath, v
 		return fmt.Errorf("sanity check failed: no @ or # character found in %q", datasetPath)
 	}

-	return ZFSDestroy(ctx, datasetPath)
+	return ZFSDestroy(datasetPath)
 }

 var destroyerSingleton = destroyerImpl{}
@@ -49,8 +48,8 @@ func setDestroySnapOpErr(b []*DestroySnapOp, err error) {
 }

 type destroyer interface {
-	Destroy(ctx context.Context, args []string) error
-	DestroySnapshotsCommaSyntaxSupported(context.Context) (bool, error)
+	Destroy(args []string) error
+	DestroySnapshotsCommaSyntaxSupported() (bool, error)
 }

 func doDestroy(ctx context.Context, reqs []*DestroySnapOp, e destroyer) {
@@ -70,7 +69,7 @@ func doDestroy(ctx context.Context, reqs []*DestroySnapOp, e destroyer) {
 	}
 	reqs = validated

-	commaSupported, err := e.DestroySnapshotsCommaSyntaxSupported(ctx)
+	commaSupported, err := e.DestroySnapshotsCommaSyntaxSupported()
 	if err != nil {
 		debug("destroy: comma syntax support detection failed: %s", err)
 		setDestroySnapOpErr(reqs, err)
@@ -86,7 +85,7 @@ func doDestroy(ctx context.Context, reqs []*DestroySnapOp, e destroyer) {

 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)})
+		*r.ErrOut = e.Destroy([]string{fmt.Sprintf("%s@%s", r.Filesystem, r.Name)})
 	}
 }

@@ -140,7 +139,7 @@ func tryBatch(ctx context.Context, batch []*DestroySnapOp, d destroyer) error {
 		}
 	}
 	batchArg := fmt.Sprintf("%s@%s", batchFS, strings.Join(batchNames, ","))
-	return d.Destroy(ctx, []string{batchArg})
+	return d.Destroy([]string{batchArg})
 }

 // fsbatch must be on same filesystem
@@ -189,7 +188,7 @@ func doDestroyBatchedRec(ctx context.Context, fsbatch []*DestroySnapOp, d destro
 		err := tryBatch(ctx, strippedBatch, d)
 		if err != nil {
 			// run entire batch sequentially if the stripped one fails
-			// (it shouldn't because we stripped erroneous datasets)
+			// (it shouldn't because we stripped erronous datasets)
 			singleRun = fsbatch // shadow
 		} else {
 			setDestroySnapOpErr(strippedBatch, nil) // these ones worked
@@ -204,7 +203,7 @@ func doDestroyBatchedRec(ctx context.Context, fsbatch []*DestroySnapOp, d destro

 type destroyerImpl struct{}

-func (d destroyerImpl) Destroy(ctx context.Context, args []string) error {
+func (d destroyerImpl) Destroy(args []string) error {
 	if len(args) != 1 {
 		// we have no use case for this at the moment, so let's crash (safer than destroying something unexpectedly)
 		panic(fmt.Sprintf("unexpected number of arguments: %v", args))
@@ -213,7 +212,7 @@ func (d destroyerImpl) Destroy(ctx context.Context, args []string) error {
 	if !strings.ContainsAny(args[0], "@") {
 		panic(fmt.Sprintf("sanity check: expecting '@' in call to Destroy, got %q", args[0]))
 	}
-	return ZFSDestroy(ctx, args[0])
+	return ZFSDestroy(args[0])
 }

 var batchDestroyFeatureCheck struct {
@@ -222,10 +221,10 @@ var batchDestroyFeatureCheck struct {
 	err    error
 }

-func (d destroyerImpl) DestroySnapshotsCommaSyntaxSupported(ctx context.Context) (bool, error) {
+func (d destroyerImpl) DestroySnapshotsCommaSyntaxSupported() (bool, error) {
 	batchDestroyFeatureCheck.once.Do(func() {
 		// "feature discovery"
-		cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "destroy")
+		cmd := exec.Command(ZFS_BINARY, "destroy")
 		output, err := cmd.CombinedOutput()
 		if _, ok := err.(*exec.ExitError); !ok {
 			debug("destroy feature check failed: %T %s", err, err)
@@ -25,11 +25,11 @@ type mockBatchDestroy struct {
 	e2biglen         int
 }

-func (m *mockBatchDestroy) DestroySnapshotsCommaSyntaxSupported(_ context.Context) (bool, error) {
+func (m *mockBatchDestroy) DestroySnapshotsCommaSyntaxSupported() (bool, error) {
 	return !m.commaUnsupported, nil
 }

-func (m *mockBatchDestroy) Destroy(ctx context.Context, args []string) error {
+func (m *mockBatchDestroy) Destroy(args []string) error {
 	defer m.mtx.Lock().Unlock()
 	if len(args) != 1 {
 		panic("unexpected use of Destroy")
@@ -21,7 +21,6 @@ import (

 	"github.com/zrepl/zrepl/util/circlog"
 	"github.com/zrepl/zrepl/util/envconst"
-	"github.com/zrepl/zrepl/zfs/zfscmd"
 )

 var (
@@ -128,7 +127,7 @@ func NewDatasetPath(s string) (p *DatasetPath, err error) {
 		return p, nil // the empty dataset path
 	}
 	const FORBIDDEN = "@#|\t<>*"
-	/* Documentation of allowed characters in zfs names:
+	/* Documenation of allowed characters in zfs names:
 	https://docs.oracle.com/cd/E19253-01/819-5461/gbcpt/index.html
 	Space is missing in the oracle list, but according to
 	https://github.com/zfsonlinux/zfs/issues/439
@@ -165,7 +164,7 @@ func (e *ZFSError) Error() string {

 var ZFS_BINARY string = "zfs"

-func ZFSList(ctx context.Context, properties []string, zfsArgs ...string) (res [][]string, err error) {
+func ZFSList(properties []string, zfsArgs ...string) (res [][]string, err error) {

 	args := make([]string, 0, 4+len(zfsArgs))
 	args = append(args,
@@ -173,9 +172,13 @@ func ZFSList(ctx context.Context, properties []string, zfsArgs ...string) (res [
 		"-o", strings.Join(properties, ","))
 	args = append(args, zfsArgs...)

-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
-	stdout, stderrBuf, err := cmd.StdoutPipeWithErrorBuf()
-	if err != nil {
+	cmd := exec.Command(ZFS_BINARY, args...)
+
+	var stdout io.Reader
+	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
+	cmd.Stderr = stderr
+
+	if stdout, err = cmd.StdoutPipe(); err != nil {
 		return
 	}

@@ -202,7 +205,7 @@ func ZFSList(ctx context.Context, properties []string, zfsArgs ...string) (res [

 	if waitErr := cmd.Wait(); waitErr != nil {
 		err := &ZFSError{
-			Stderr:  stderrBuf.Bytes(),
+			Stderr:  stderr.Bytes(),
 			WaitErr: waitErr,
 		}
 		return nil, err
@@ -241,12 +244,15 @@ func ZFSListChan(ctx context.Context, out chan ZFSListResult, properties []strin
 		}
 	}

-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
-	stdout, stderrBuf, err := cmd.StdoutPipeWithErrorBuf()
+	cmd := exec.CommandContext(ctx, ZFS_BINARY, args...)
+	stdout, err := cmd.StdoutPipe()
 	if err != nil {
 		sendResult(nil, err)
 		return
 	}
+	// TODO bounded buffer
+	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
+	cmd.Stderr = stderr
 	if err = cmd.Start(); err != nil {
 		sendResult(nil, err)
 		return
@@ -274,7 +280,7 @@ func ZFSListChan(ctx context.Context, out chan ZFSListResult, properties []strin
 	if err := cmd.Wait(); err != nil {
 		if err, ok := err.(*exec.ExitError); ok {
 			sendResult(nil, &ZFSError{
-				Stderr:  stderrBuf.Bytes(),
+				Stderr:  stderr.Bytes(),
 				WaitErr: err,
 			})
 		} else {
@@ -413,7 +419,7 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 }

 type sendStream struct {
-	cmd  *zfscmd.Cmd
+	cmd  *exec.Cmd
 	kill context.CancelFunc

 	closeMtx     sync.Mutex
@@ -505,7 +511,7 @@ func (s *sendStream) killAndWait(precedingReadErr error) error {

 	// detect the edge where we're called from s.Read
 	// after the pipe EOFed and zfs send exited without errors
-	// this is actually the "hot" / nice path
+	// this is actullay the "hot" / nice path
 	if exitErr == nil && precedingReadErr == io.EOF {
 		return precedingReadErr
 	}
@@ -558,7 +564,7 @@ func (a ZFSSendArgVersion) ValidateExistsAndGetCheckedProps(ctx context.Context,
 		return ZFSPropCreateTxgAndGuidProps{}, nil
 	}

-	realProps, err := ZFSGetCreateTXGAndGuid(ctx, a.FullPath(fs))
+	realProps, err := ZFSGetCreateTXGAndGuid(a.FullPath(fs))
 	if err != nil {
 		return ZFSPropCreateTxgAndGuidProps{}, err
 	}
@@ -612,13 +618,13 @@ func (n *NilBool) String() string {
 	return fmt.Sprintf("%v", n.B)
 }

-// When updating this struct, check Validate and ValidateCorrespondsToResumeToken (POTENTIALLY SECURITY SENSITIVE)
+// When updating this struct, check Validate and ValidateCorrespondsToResumeToken (Potentiall SECURITY SENSITIVE)
 type ZFSSendArgs struct {
 	FS        string
 	From, To  *ZFSSendArgVersion // From may be nil
 	Encrypted *NilBool

-	// Preferred if not empty
+	// Prefereed if not empty
 	ResumeToken string // if not nil, must match what is specified in From, To (covered by ValidateCorrespondsToResumeToken)
 }

@@ -654,7 +660,7 @@ func (e ZFSSendArgsValidationError) Error() string {
 }

 // - Recursively call Validate on each field.
-// - Make sure that if ResumeToken != "", it reflects the same operation as the other parameters would.
+// - Make sure that if ResumeToken != "", it reflects the same operation as the other paramters would.
 //
 // This function is not pure because GUIDs are checked against the local host's datasets.
 func (a ZFSSendArgs) Validate(ctx context.Context) error {
@@ -673,7 +679,7 @@ func (a ZFSSendArgs) Validate(ctx context.Context) error {
 		if err := a.From.ValidateExists(ctx, a.FS); err != nil {
 			return newGenericValidationError(a, errors.Wrap(err, "`From` invalid"))
 		}
-		// fallthrough
+		// falthrough
 	}

 	if err := a.Encrypted.Validate(); err != nil {
@@ -737,7 +743,7 @@ func (a ZFSSendArgs) validateCorrespondsToResumeToken(ctx context.Context, valCt

 	debug("decoding resume token %q", a.ResumeToken)
 	t, err := ParseResumeToken(ctx, a.ResumeToken)
-	debug("decode resume token result: %#v %T %v", t, err, err)
+	debug("decode resumee token result: %#v %T %v", t, err, err)
 	if err != nil {
 		return err
 	}
@@ -830,6 +836,7 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgs) (*ReadCloserCopier, erro
 	args = append(args, sargs...)

 	ctx, cancel := context.WithCancel(ctx)
+	cmd := exec.CommandContext(ctx, ZFS_BINARY, args...)

 	// setup stdout with an os.Pipe to control pipe buffer size
 	stdoutReader, stdoutWriter, err := pipeWithCapacityHint(ZFSSendPipeCapacityHint)
@@ -837,14 +844,11 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgs) (*ReadCloserCopier, erro
 		cancel()
 		return nil, err
 	}
-	stderrBuf := circlog.MustNewCircularLog(zfsSendStderrCaptureMaxSize)

-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
-	cmd.SetStdio(zfscmd.Stdio{
-		Stdin:  nil,
-		Stdout: stdoutWriter,
-		Stderr: stderrBuf,
-	})
+	cmd.Stdout = stdoutWriter
+
+	stderrBuf := circlog.MustNewCircularLog(zfsSendStderrCaptureMaxSize)
+	cmd.Stderr = stderrBuf

 	if err := cmd.Start(); err != nil {
 		cancel()
@@ -852,7 +856,6 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgs) (*ReadCloserCopier, erro
 		stdoutReader.Close()
 		return nil, errors.Wrap(err, "cannot start zfs send command")
 	}
-	// close our writing-end of the pipe so that we don't wait for ourselves when reading from the reading  end
 	stdoutWriter.Close()

 	stream := &sendStream{
@@ -891,7 +894,7 @@ type DrySendInfo struct {
 }

 var (
-	// keep same number of capture groups for unmarshalInfoLine homogeneity
+	// keep same number of capture groups for unmarshalInfoLine homogenity

 	sendDryRunInfoLineRegexFull = regexp.MustCompile(`^(full)\t()([^\t]+@[^\t]+)\t([0-9]+)$`)
 	// cannot enforce '[#@]' in incremental source, see test cases
@@ -993,7 +996,7 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgs) (_ *DrySendInfo, err
 	}
 	args = append(args, sargs...)

-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
+	cmd := exec.Command(ZFS_BINARY, args...)
 	output, err := cmd.CombinedOutput()
 	if err != nil {
 		return nil, &ZFSError{output, err}
@@ -1087,11 +1090,11 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier
 			rollbackTarget := snaps[0]
 			rollbackTargetAbs := rollbackTarget.ToAbsPath(fsdp)
 			debug("recv: rollback to %q", rollbackTargetAbs)
-			if err := ZFSRollback(ctx, fsdp, rollbackTarget, "-r"); err != nil {
+			if err := ZFSRollback(fsdp, rollbackTarget, "-r"); err != nil {
 				return fmt.Errorf("cannot rollback %s to %s for forced receive: %s", fsdp.ToString(), rollbackTarget, err)
 			}
 			debug("recv: destroy %q", rollbackTargetAbs)
-			if err := ZFSDestroy(ctx, rollbackTargetAbs); err != nil {
+			if err := ZFSDestroy(rollbackTargetAbs); err != nil {
 				return fmt.Errorf("cannot destroy %s for forced receive: %s", rollbackTargetAbs, err)
 			}
 		}
@@ -1112,26 +1115,24 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier

 	ctx, cancelCmd := context.WithCancel(ctx)
 	defer cancelCmd()
-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
+	cmd := exec.CommandContext(ctx, ZFS_BINARY, args...)
+
+	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
+	cmd.Stderr = stderr

 	// TODO report bug upstream
 	// Setup an unused stdout buffer.
 	// Otherwise, ZoL v0.6.5.9-1 3.16.0-4-amd64 writes the following error to stderr and exits with code 1
 	//   cannot receive new filesystem stream: invalid backup stream
 	stdout := bytes.NewBuffer(make([]byte, 0, 1024))
-
-	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
+	cmd.Stdout = stdout

 	stdin, stdinWriter, err := pipeWithCapacityHint(ZFSRecvPipeCapacityHint)
 	if err != nil {
 		return err
 	}

-	cmd.SetStdio(zfscmd.Stdio{
-		Stdin:  stdin,
-		Stdout: stdout,
-		Stderr: stderr,
-	})
+	cmd.Stdin = stdin

 	if err = cmd.Start(); err != nil {
 		stdinWriter.Close()
@@ -1141,7 +1142,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier
 	stdin.Close()
 	defer stdinWriter.Close()

-	pid := cmd.Process().Pid
+	pid := cmd.Process.Pid
 	debug := func(format string, args ...interface{}) {
 		debug("recv: pid=%v: %s", pid, fmt.Sprintf(format, args...))
 	}
@@ -1224,12 +1225,12 @@ func (e ClearResumeTokenError) Error() string {
 }

 // always returns *ClearResumeTokenError
-func ZFSRecvClearResumeToken(ctx context.Context, fs string) (err error) {
+func ZFSRecvClearResumeToken(fs string) (err error) {
 	if err := validateZFSFilesystem(fs); err != nil {
 		return err
 	}

-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "recv", "-A", fs)
+	cmd := exec.Command(ZFS_BINARY, "recv", "-A", fs)
 	o, err := cmd.CombinedOutput()
 	if err != nil {
 		if bytes.Contains(o, []byte("does not have any resumable receive state to abort")) {
@@ -1266,11 +1267,11 @@ func (p *ZFSProperties) appendArgs(args *[]string) (err error) {
 	return nil
 }

-func ZFSSet(ctx context.Context, fs *DatasetPath, props *ZFSProperties) (err error) {
-	return zfsSet(ctx, fs.ToString(), props)
+func ZFSSet(fs *DatasetPath, props *ZFSProperties) (err error) {
+	return zfsSet(fs.ToString(), props)
 }

-func zfsSet(ctx context.Context, path string, props *ZFSProperties) (err error) {
+func zfsSet(path string, props *ZFSProperties) (err error) {
 	args := make([]string, 0)
 	args = append(args, "set")
 	err = props.appendArgs(&args)
@@ -1279,11 +1280,18 @@ func zfsSet(ctx context.Context, path string, props *ZFSProperties) (err error)
 	}
 	args = append(args, path)

-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
-	stdio, err := cmd.CombinedOutput()
-	if err != nil {
+	cmd := exec.Command(ZFS_BINARY, args...)
+
+	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
+	cmd.Stderr = stderr
+
+	if err = cmd.Start(); err != nil {
+		return err
+	}
+
+	if err = cmd.Wait(); err != nil {
 		err = &ZFSError{
-			Stderr:  stdio,
+			Stderr:  stderr.Bytes(),
 			WaitErr: err,
 		}
 	}
@@ -1291,12 +1299,12 @@ func zfsSet(ctx context.Context, path string, props *ZFSProperties) (err error)
 	return
 }

-func ZFSGet(ctx context.Context, fs *DatasetPath, props []string) (*ZFSProperties, error) {
-	return zfsGet(ctx, fs.ToString(), props, sourceAny)
+func ZFSGet(fs *DatasetPath, props []string) (*ZFSProperties, error) {
+	return zfsGet(fs.ToString(), props, sourceAny)
 }

 // The returned error includes requested filesystem and version as quoted strings in its error message
-func ZFSGetGUID(ctx context.Context, fs string, version string) (g uint64, err error) {
+func ZFSGetGUID(fs string, version string) (g uint64, err error) {
 	defer func(e *error) {
 		if *e != nil {
 			*e = fmt.Errorf("zfs get guid fs=%q version=%q: %s", fs, version, *e)
@@ -1312,7 +1320,7 @@ func ZFSGetGUID(ctx context.Context, fs string, version string) (g uint64, err e
 		return 0, errors.New("version does not start with @ or #")
 	}
 	path := fmt.Sprintf("%s%s", fs, version)
-	props, err := zfsGet(ctx, path, []string{"guid"}, sourceAny) // always local
+	props, err := zfsGet(path, []string{"guid"}, sourceAny) // always local
 	if err != nil {
 		return 0, err
 	}
@@ -1324,11 +1332,11 @@ type GetMountpointOutput struct {
 	Mountpoint string
 }

-func ZFSGetMountpoint(ctx context.Context, fs string) (*GetMountpointOutput, error) {
+func ZFSGetMountpoint(fs string) (*GetMountpointOutput, error) {
 	if err := EntityNamecheck(fs, EntityTypeFilesystem); err != nil {
 		return nil, err
 	}
-	props, err := zfsGet(ctx, fs, []string{"mountpoint", "mounted"}, sourceAny)
+	props, err := zfsGet(fs, []string{"mountpoint", "mounted"}, sourceAny)
 	if err != nil {
 		return nil, err
 	}
@@ -1344,8 +1352,8 @@ func ZFSGetMountpoint(ctx context.Context, fs string) (*GetMountpointOutput, err
 	return o, nil
 }

-func ZFSGetRawAnySource(ctx context.Context, path string, props []string) (*ZFSProperties, error) {
-	return zfsGet(ctx, path, props, sourceAny)
+func ZFSGetRawAnySource(path string, props []string) (*ZFSProperties, error) {
+	return zfsGet(path, props, sourceAny)
 }

 var zfsGetDatasetDoesNotExistRegexp = regexp.MustCompile(`^cannot open '([^)]+)': (dataset does not exist|no such pool or dataset)`) // verified in platformtest
@@ -1404,9 +1412,9 @@ func (s zfsPropertySource) zfsGetSourceFieldPrefixes() []string {
 	return prefixes
 }

-func zfsGet(ctx context.Context, path string, props []string, allowedSources zfsPropertySource) (*ZFSProperties, error) {
+func zfsGet(path string, props []string, allowedSources zfsPropertySource) (*ZFSProperties, error) {
 	args := []string{"get", "-Hp", "-o", "property,value,source", strings.Join(props, ","), path}
-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
+	cmd := exec.Command(ZFS_BINARY, args...)
 	stdout, err := cmd.Output()
 	if err != nil {
 		if exitErr, ok := err.(*exec.ExitError); ok {
@@ -1454,8 +1462,8 @@ type ZFSPropCreateTxgAndGuidProps struct {
 	CreateTXG, Guid uint64
 }

-func ZFSGetCreateTXGAndGuid(ctx context.Context, ds string) (ZFSPropCreateTxgAndGuidProps, error) {
-	props, err := zfsGetNumberProps(ctx, ds, []string{"createtxg", "guid"}, sourceAny)
+func ZFSGetCreateTXGAndGuid(ds string) (ZFSPropCreateTxgAndGuidProps, error) {
+	props, err := zfsGetNumberProps(ds, []string{"createtxg", "guid"}, sourceAny)
 	if err != nil {
 		return ZFSPropCreateTxgAndGuidProps{}, err
 	}
@@ -1466,8 +1474,8 @@ func ZFSGetCreateTXGAndGuid(ctx context.Context, ds string) (ZFSPropCreateTxgAnd
 }

 // returns *DatasetDoesNotExist if the dataset does not exist
-func zfsGetNumberProps(ctx context.Context, ds string, props []string, src zfsPropertySource) (map[string]uint64, error) {
-	sps, err := zfsGet(ctx, ds, props, sourceAny)
+func zfsGetNumberProps(ds string, props []string, src zfsPropertySource) (map[string]uint64, error) {
+	sps, err := zfsGet(ds, props, sourceAny)
 	if err != nil {
 		if _, ok := err.(*DatasetDoesNotExist); ok {
 			return nil, err // pass through as is
@@ -1549,7 +1557,7 @@ func tryParseDestroySnapshotsError(arg string, stderr []byte) *DestroySnapshotsE
 	}
 }

-func ZFSDestroy(ctx context.Context, arg string) (err error) {
+func ZFSDestroy(arg string) (err error) {

 	var dstype, filesystem string
 	idx := strings.IndexAny(arg, "@#")
@@ -1568,21 +1576,28 @@ func ZFSDestroy(ctx context.Context, arg string) (err error) {

 	defer prometheus.NewTimer(prom.ZFSDestroyDuration.WithLabelValues(dstype, filesystem))

-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "destroy", arg)
-	stdio, err := cmd.CombinedOutput()
-	if err != nil {
+	cmd := exec.Command(ZFS_BINARY, "destroy", arg)
+
+	var stderr bytes.Buffer
+	cmd.Stderr = &stderr
+
+	if err = cmd.Start(); err != nil {
+		return err
+	}
+
+	if err = cmd.Wait(); err != nil {
 		err = &ZFSError{
-			Stderr:  stdio,
+			Stderr:  stderr.Bytes(),
 			WaitErr: err,
 		}

-		if destroyOneOrMoreSnapshotsNoneExistedErrorRegexp.Match(stdio) {
+		if destroyOneOrMoreSnapshotsNoneExistedErrorRegexp.Match(stderr.Bytes()) {
 			err = &DatasetDoesNotExist{arg}
-		} else if match := destroyBookmarkDoesNotExist.FindStringSubmatch(string(stdio)); match != nil && match[1] == arg {
+		} else if match := destroyBookmarkDoesNotExist.FindStringSubmatch(stderr.String()); match != nil && match[1] == arg {
 			err = &DatasetDoesNotExist{arg}
-		} else if dsNotExistErr := tryDatasetDoesNotExist(filesystem, stdio); dsNotExistErr != nil {
+		} else if dsNotExistErr := tryDatasetDoesNotExist(filesystem, stderr.Bytes()); dsNotExistErr != nil {
 			err = dsNotExistErr
-		} else if dserr := tryParseDestroySnapshotsError(arg, stdio); dserr != nil {
+		} else if dserr := tryParseDestroySnapshotsError(arg, stderr.Bytes()); dserr != nil {
 			err = dserr
 		}

@@ -1592,15 +1607,15 @@ func ZFSDestroy(ctx context.Context, arg string) (err error) {

 }

-func ZFSDestroyIdempotent(ctx context.Context, path string) error {
-	err := ZFSDestroy(ctx, path)
+func ZFSDestroyIdempotent(path string) error {
+	err := ZFSDestroy(path)
 	if _, ok := err.(*DatasetDoesNotExist); ok {
 		return nil
 	}
 	return err
 }

-func ZFSSnapshot(ctx context.Context, fs *DatasetPath, name string, recursive bool) (err error) {
+func ZFSSnapshot(fs *DatasetPath, name string, recursive bool) (err error) {

 	promTimer := prometheus.NewTimer(prom.ZFSSnapshotDuration.WithLabelValues(fs.ToString()))
 	defer promTimer.ObserveDuration()
@@ -1609,12 +1624,18 @@ func ZFSSnapshot(ctx context.Context, fs *DatasetPath, name string, recursive bo
 	if err := EntityNamecheck(snapname, EntityTypeSnapshot); err != nil {
 		return errors.Wrap(err, "zfs snapshot")
 	}
+	cmd := exec.Command(ZFS_BINARY, "snapshot", snapname)

-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "snapshot", snapname)
-	stdio, err := cmd.CombinedOutput()
-	if err != nil {
+	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
+	cmd.Stderr = stderr
+
+	if err = cmd.Start(); err != nil {
+		return err
+	}
+
+	if err = cmd.Wait(); err != nil {
 		err = &ZFSError{
-			Stderr:  stdio,
+			Stderr:  stderr.Bytes(),
 			WaitErr: err,
 		}
 	}
@@ -1646,7 +1667,7 @@ var ErrBookmarkCloningNotSupported = fmt.Errorf("bookmark cloning feature is not
 //
 // does not destroy an existing bookmark, returns
 //
-func ZFSBookmark(ctx context.Context, fs string, v ZFSSendArgVersion, bookmark string) (err error) {
+func ZFSBookmark(fs string, v ZFSSendArgVersion, bookmark string) (err error) {

 	promTimer := prometheus.NewTimer(prom.ZFSBookmarkDuration.WithLabelValues(fs))
 	defer promTimer.ObserveDuration()
@@ -1666,15 +1687,23 @@ func ZFSBookmark(ctx context.Context, fs string, v ZFSSendArgVersion, bookmark s

 	debug("bookmark: %q %q", snapname, bookmarkname)

-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, "bookmark", snapname, bookmarkname)
-	stdio, err := cmd.CombinedOutput()
-	if err != nil {
-		if ddne := tryDatasetDoesNotExist(snapname, stdio); err != nil {
+	cmd := exec.Command(ZFS_BINARY, "bookmark", snapname, bookmarkname)
+
+	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
+	cmd.Stderr = stderr
+
+	if err = cmd.Start(); err != nil {
+		return err
+	}
+
+	if err = cmd.Wait(); err != nil {
+
+		if ddne := tryDatasetDoesNotExist(snapname, stderr.Bytes()); err != nil {
 			return ddne
-		} else if zfsBookmarkExistsRegex.Match(stdio) {
+		} else if zfsBookmarkExistsRegex.Match(stderr.Bytes()) {

 			// check if this was idempotent
-			bookGuid, err := ZFSGetGUID(ctx, fs, "#"+bookmark)
+			bookGuid, err := ZFSGetGUID(fs, "#"+bookmark)
 			if err != nil {
 				return errors.Wrap(err, "bookmark idempotency check") // guid error expressive enough
 			}
@@ -1685,13 +1714,13 @@ func ZFSBookmark(ctx context.Context, fs string, v ZFSSendArgVersion, bookmark s
 			}
 			return &BookmarkExists{
 				fs: fs, bookmarkOrigin: v, bookmark: bookmark,
-				zfsMsg:   string(stdio),
+				zfsMsg:   stderr.String(),
 				bookGuid: bookGuid,
 			}

 		} else {
 			return &ZFSError{
-				Stderr:  stdio,
+				Stderr:  stderr.Bytes(),
 				WaitErr: err,
 			}
 		}
@@ -1702,7 +1731,7 @@ func ZFSBookmark(ctx context.Context, fs string, v ZFSSendArgVersion, bookmark s

 }

-func ZFSRollback(ctx context.Context, fs *DatasetPath, snapshot FilesystemVersion, rollbackArgs ...string) (err error) {
+func ZFSRollback(fs *DatasetPath, snapshot FilesystemVersion, rollbackArgs ...string) (err error) {

 	snapabs := snapshot.ToAbsPath(fs)
 	if snapshot.Type != Snapshot {
@@ -1713,11 +1742,18 @@ func ZFSRollback(ctx context.Context, fs *DatasetPath, snapshot FilesystemVersio
 	args = append(args, rollbackArgs...)
 	args = append(args, snapabs)

-	cmd := zfscmd.CommandContext(ctx, ZFS_BINARY, args...)
-	stdio, err := cmd.CombinedOutput()
-	if err != nil {
+	cmd := exec.Command(ZFS_BINARY, args...)
+
+	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
+	cmd.Stderr = stderr
+
+	if err = cmd.Start(); err != nil {
+		return err
+	}
+
+	if err = cmd.Wait(); err != nil {
 		err = &ZFSError{
-			Stderr:  stdio,
+			Stderr:  stderr.Bytes(),
 			WaitErr: err,
 		}
 	}
@@ -1,7 +1,6 @@
 package zfs

 import (
-	"context"
 	"testing"

 	"github.com/stretchr/testify/assert"
@@ -12,11 +11,9 @@ func TestZFSListHandlesProducesZFSErrorOnNonZeroExit(t *testing.T) {

 	var err error

-	ctx := context.Background()
-
 	ZFS_BINARY = "./test_helpers/zfs_failer.sh"

-	_, err = ZFSList(ctx, []string{"fictionalprop"}, "nonexistent/dataset")
+	_, err = ZFSList([]string{"fictionalprop"}, "nonexistent/dataset")

 	assert.Error(t, err)
 	zfsError, ok := err.(*ZFSError)
@@ -107,7 +104,7 @@ nvlist version: 0
 incremental	zroot/test/a@1	zroot/test/a@2	5383936
 `

-	// # incremental send with token + bookmark
+	// # incremental send with token + bookmarmk
 	// $ zfs send -nvP -t 1-ef01e717e-e0-789c636064000310a501c49c50360710a715e5e7a69766a63040c1d904b9e342877e062900d9ec48eaf293b252934b181898a0ea30e4d3d28a534b40323e70793624f9a4ca92d46220fdc1ce0fabfe927c882bc46c8a0a9f71ad3baf8124cf0996cf4bcc4d6560a82acacf2fd1079a55a29fe86004710b00d8ae1f93
 	incSendBookmark := `
 resume token contents:
@@ -1,11 +0,0 @@
-The tool in this package (`go run . -h`) scrapes log lines produces by the `github.com/zrepl/zrepl/zfs/zfscmd` package
-into a stream of JSON objects.
-
-The `analysis.ipynb` then runs some basic analysis on the collected log output.
-
-## Deps for the `scrape_graylog_csv.bash` script
-
-```
-pip install --upgrade git+https://github.com/lk-jeffpeck/csvfilter.git@ec433f14330fbbf5d41f56febfeedac22868a949
-```
-
@@ -1,260 +0,0 @@
-{
- "cells": [
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "import pandas as pd\n",
-    "import seaborn as sns\n",
-    "import matplotlib.pyplot as plt\n",
-    "import re\n"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "# ./parsed.json is the stdout of the scraper tool in this directory\n",
-    "df = pd.read_json(\"./parsed.json\", lines=True)\n",
-    "df"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "def parse_ds(entity):\n",
-    "    m = re.search(r\"(?P<dataset>[^@#]*)([@#].+)?\", entity)\n",
-    "    return m.group(\"dataset\")\n",
-    "    \n",
-    "def parse_cmd(row):\n",
-    "    cmd  = row.Cmd\n",
-    "    binary, verb, *tail = re.split(r\"\\s+\", cmd) # NOTE whitespace in dataset names => don't use comp\n",
-    "    \n",
-    "    dataset = None\n",
-    "    if binary == \"zfs\":\n",
-    "        if verb == \"send\":      \n",
-    "            if len(tail) == 0:\n",
-    "                verb = \"send-feature-test\"\n",
-    "            else:\n",
-    "                dataset = parse_ds(tail[-1])\n",
-    "                if \"-n\" in tail:\n",
-    "                    verb = \"send-dry\"\n",
-    "        elif verb == \"recv\" or verb == \"receive\":\n",
-    "            verb = \"receive\"\n",
-    "            if len(tail) > 0:\n",
-    "                dataset = parse_ds(tail[-1])\n",
-    "            else:\n",
-    "                verb = \"receive-CLI-test\"\n",
-    "        elif verb == \"get\":\n",
-    "            dataset = parse_ds(tail[-1])\n",
-    "        elif verb == \"list\":\n",
-    "            if \"-r\" in tail and \"-d\" in tail and \"1\" in tail:\n",
-    "                dataset = parse_ds(tail[-1])\n",
-    "                verb = \"list-single-dataset\"\n",
-    "            else:\n",
-    "                dataset = \"!ALL_POOLS!\"\n",
-    "                verb = \"list-all-filesystems\"\n",
-    "        elif verb == \"bookmark\":\n",
-    "            dataset = parse_ds(tail[-2])\n",
-    "        elif verb == \"hold\":\n",
-    "            dataset = parse_ds(tail[-1])\n",
-    "        elif verb == \"snapshot\":\n",
-    "            dataset = parse_ds(tail[-1])\n",
-    "        elif verb == \"release\":\n",
-    "            dss = tail[-1].split(\",\")\n",
-    "            if len(dss) > 1:\n",
-    "                raise Exception(\"cannot handle batch-release\")\n",
-    "            dataset = parse_ds(dss[0])\n",
-    "        elif verb == \"holds\" and \"-H\" in tail:\n",
-    "            dss = tail[-1].split(\",\")\n",
-    "            if len(dss) > 1:\n",
-    "                raise Exception(\"cannot handle batch-holds\")\n",
-    "            dataset = parse_ds(dss[0])\n",
-    "        elif verb == \"destroy\":\n",
-    "            dss = tail[-1].split(\",\")\n",
-    "            if len(dss) > 1:\n",
-    "                raise Exception(\"cannot handle batch-holds\")\n",
-    "            dataset = parse_ds(dss[0])\n",
-    "    \n",
-    "    return {'action':binary + \"-\" + verb, 'dataset': dataset }\n",
-    "    \n",
-    "    \n",
-    "res = df.apply(parse_cmd, axis='columns', result_type='expand')\n",
-    "res = pd.concat([df, res], axis='columns')\n",
-    "for cat in [\"action\", \"dataset\"]:\n",
-    "    res[cat] = res[cat].astype('category')\n",
-    "\n",
-    "res[\"LogTimeUnix\"] = pd.to_datetime(res.LogTime)"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "res[\"OtherTime\"] = res.TotalTime - res.Usertime - res.Systime\n",
-    "x = res.melt(id_vars=[\"action\", \"dataset\"], value_vars=[\"TotalTime\", \"OtherTime\", \"Usertime\", \"Systime\"])\n",
-    "x"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "print(\"commands with NaN values\")\n",
-    "set(x[x.isna().any(axis=1)].action.values)"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "# (~x.action.astype('str').isin([\"zfs-send\", \"zfs-recv\"]))\n",
-    "totaltimes = x[(x.variable == \"TotalTime\")].groupby([\"action\", \"dataset\"]).sum().reset_index()\n",
-    "display(totaltimes)"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "totaltimes_by_action = totaltimes.groupby(\"action\").sum().sort_values(by=\"value\")\n",
-    "totaltimes_by_action.plot.barh()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "totaltimes.groupby(\"dataset\").sum().plot.barh(fontsize=5)"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "most_expensive_action = totaltimes_by_action.idxmax().value\n",
-    "display(most_expensive_action)\n",
-    "most_expensive_action_by_dataset = totaltimes[totaltimes.action == most_expensive_action].groupby(\"dataset\").sum().sort_values(by=\"value\")\n",
-    "most_expensive_action_by_dataset.plot.barh(rot=50, fontsize=5, figsize=(10, 20))\n",
-    "plt.savefig('most-expensive-command.pdf')"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "res"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "# %matplotlib notebook \n",
-    "\n",
-    "# res.index = res.LogTimeUnix\n",
-    "\n",
-    "# resampled = res.pivot(columns='action', values='TotalTime').resample(\"1s\").sum()\n",
-    "# resampled.cumsum().plot()\n",
-    "# res[\"BeginTime\"] = res.LogTimeUnix.dt.total_seconds()\n",
-    "# holds = res[res.action == \"zfs-holds\"]\n",
-    "# sns.stripplot(x=\"LogTimeUnix\", y=\"action\", data=res)\n",
-    "# res[\"LogTimeUnix\"].resample(\"20min\").sum()\n",
-    "# res[res.action == \"zfs-holds\"].plot.scatter(x=\"LogTimeUnix\", y=\"TotalTime\")\n",
-    "\n",
-    "#res[res.action == \"zfs-holds\"].pivot(columns='action', values=['TotalTime', 'Systime', \"Usertime\"]).resample(\"1s\").sum().cumsum().plot()\n",
-    "pivoted = res.reset_index(drop=True).pivot_table(values=['TotalTime', 'Systime', \"Usertime\"], index=\"LogTimeUnix\", columns=\"action\")\n",
-    "pivoted"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "pivoted.cumsum()[[(\"TotalTime\", \"zfs-holds\"),(\"Systime\", \"zfs-holds\"),(\"Usertime\", \"zfs-holds\")]].plot()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "pivoted = res.reset_index(drop=True).pivot_table(values=['TotalTime'], index=\"LogTimeUnix\", columns=\"action\")\n",
-    "cum_invocation_counts_per_action = pivoted.isna().astype(int).cumsum()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "cum_invocation_counts_per_action"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "# zfs-get as reference value\n",
-    "cum_invocation_counts_per_action[[(\"TotalTime\",\"zfs-holds\"),(\"TotalTime\",\"zfs-get\")]].plot()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": []
-  }
- ],
- "metadata": {
-  "kernelspec": {
-   "display_name": "Python 3",
-   "language": "python",
-   "name": "python3"
-  },
-  "language_info": {
-   "codemirror_mode": {
-    "name": "ipython",
-    "version": 3
-   },
-   "file_extension": ".py",
-   "mimetype": "text/x-python",
-   "name": "python",
-   "nbconvert_exporter": "python",
-   "pygments_lexer": "ipython3",
-   "version": "3.7.6"
-  }
- },
- "nbformat": 4,
- "nbformat_minor": 2
-}
@@ -1,16 +0,0 @@
-#!/usr/bin/env bash
-
-# This script converts output that was produced by zrepl and captured by Graylog
-# back to something that the scraper in this package's main can understand
-# Intended for human syslog
-# logging:
-# - type: syslog
-#   level: debug
-#   format: human
-
-
-csvfilter --skip 1 -f 0,2 -q '"' --out-quotechar=' ' /dev/stdin | sed -E 's/^\s*([^,]*), /\1 [LEVEL]/' | \
-    go run . -v \
-    --dateRE '^([^\[]+) (\[.*)' \
-    --dateFormat '2006-01-02T15:04:05.999999999Z'
-
@@ -1,129 +0,0 @@
-package main
-
-import (
-	"bufio"
-	"encoding/json"
-	"fmt"
-	"os"
-	"regexp"
-	"strings"
-	"time"
-
-	"github.com/go-logfmt/logfmt"
-	"github.com/pkg/errors"
-	"github.com/spf13/pflag"
-
-	"github.com/zrepl/zrepl/daemon/logging"
-)
-
-type RuntimeLine struct {
-	LogTime                      time.Time
-	Cmd                          string
-	TotalTime, Usertime, Systime time.Duration
-	Error                        string
-}
-
-var humanFormatterLineRE = regexp.MustCompile(`^(\[[^\]]+\]){2}\[zfs.cmd\]:\s+command\s+exited\s+(with|without)\s+error\s+(.+)`)
-
-func parseSecs(s string) (time.Duration, error) {
-	d, err := time.ParseDuration(s + "s")
-	if err != nil {
-		return 0, errors.Wrapf(err, "parse duration %q", s)
-	}
-	return d, nil
-}
-
-func parseHumanFormatterNodate(line string) (l RuntimeLine, err error) {
-	m := humanFormatterLineRE.FindStringSubmatch(line)
-	if m == nil {
-		return l, errors.New("human formatter regex does not match")
-	}
-
-	d := logfmt.NewDecoder(strings.NewReader(m[3]))
-	for d.ScanRecord() {
-		for d.ScanKeyval() {
-			k := string(d.Key())
-			v := string(d.Value())
-			switch k {
-			case "cmd":
-				l.Cmd = v
-			case "total_time_s":
-				l.TotalTime, err = parseSecs(v)
-			case "usertime_s":
-				l.Usertime, err = parseSecs(v)
-			case "systemtime_s":
-				l.Systime, err = parseSecs(v)
-			case "err":
-				l.Error = v
-			case "invocation":
-				continue // pass
-			default:
-				return l, errors.Errorf("unknown key %q", k)
-			}
-			if err != nil {
-				return l, err
-			}
-		}
-	}
-	if d.Err() != nil {
-		return l, errors.Wrap(d.Err(), "decode key value pairs")
-	}
-	return l, nil
-}
-
-func parseLogLine(line string) (l RuntimeLine, err error) {
-	m := dateRegex.FindStringSubmatch(line)
-	if len(m) != 3 {
-		return l, errors.Errorf("invalid date regex match %v", m)
-	}
-	dateTrimmed := strings.TrimSpace(m[1])
-	date, err := time.Parse(dateFormat, dateTrimmed)
-	if err != nil {
-		panic(fmt.Sprintf("cannot parse date %q: %s", dateTrimmed, err))
-	}
-	logLine := m[2]
-
-	l, err = parseHumanFormatterNodate(strings.TrimSpace(logLine))
-	l.LogTime = date
-	return l, err
-}
-
-var verbose bool
-var dateRegexArg string
-var dateRegex = regexp.MustCompile(`^([^\[]+)(.*)`)
-var dateFormat = logging.HumanFormatterDateFormat
-
-func main() {
-
-	pflag.StringVarP(&dateRegexArg, "dateRE", "d", "", "date regex")
-	pflag.StringVar(&dateFormat, "dateFormat", logging.HumanFormatterDateFormat, "go date format")
-	pflag.BoolVarP(&verbose, "verbose", "v", false, "verbose")
-	pflag.Parse()
-
-	if dateRegexArg != "" {
-		dateRegex = regexp.MustCompile(dateRegexArg)
-	}
-
-	input := bufio.NewScanner(os.Stdin)
-	input.Split(bufio.ScanLines)
-
-	enc := json.NewEncoder(os.Stdout)
-	for input.Scan() {
-
-		l, err := parseLogLine(input.Text())
-		if err != nil && verbose {
-			fmt.Fprintf(os.Stderr, "ignoring line after error %v\n", err)
-			fmt.Fprintf(os.Stderr, "offending line was: %s\n", input.Text())
-		}
-		if err == nil {
-			if err := enc.Encode(l); err != nil {
-				panic(err)
-			}
-		}
-	}
-
-	if input.Err() != nil {
-		panic(input.Err())
-	}
-
-}
@@ -1,86 +0,0 @@
-package main
-
-import (
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/zrepl/zrepl/daemon/logging"
-)
-
-func TestParseHumanFormatter(t *testing.T) {
-
-	type testCase struct {
-		Name      string
-		Input     string
-		Expect    *RuntimeLine
-		ExpectErr string
-	}
-
-	secs := func(s string) time.Duration {
-		d, err := parseSecs(s)
-		require.NoError(t, err)
-		return d
-	}
-
-	logTime, err := time.Parse(logging.HumanFormatterDateFormat, "2020-04-04T00:00:05+02:00")
-	require.NoError(t, err)
-
-	tcs := []testCase{
-		{
-			Name:  "human-formatter-noerror",
-			Input: `2020-04-04T00:00:05+02:00 [DEBG][jobname][zfs.cmd]: command exited without error usertime_s="0.008445" cmd="zfs list -H -p -o name -r -t filesystem,volume" systemtime_s="0.033783" invocation="84" total_time_s="0.037828619"`,
-			Expect: &RuntimeLine{
-				Cmd:       "zfs list -H -p -o name -r -t filesystem,volume",
-				TotalTime: secs("0.037828619"),
-				Usertime:  secs("0.008445"),
-				Systime:   secs("0.033783"),
-				Error:     "",
-				LogTime:   logTime,
-			},
-		},
-		{
-			Name:  "human-formatter-witherror",
-			Input: `2020-04-04T00:00:05+02:00 [DEBG][jobname][zfs.cmd]: command exited with error usertime_s="0.008445" cmd="zfs list -H -p -o name -r -t filesystem,volume" systemtime_s="0.033783" invocation="84" total_time_s="0.037828619" err="some error"`,
-			Expect: &RuntimeLine{
-				Cmd:       "zfs list -H -p -o name -r -t filesystem,volume",
-				TotalTime: secs("0.037828619"),
-				Usertime:  secs("0.008445"),
-				Systime:   secs("0.033783"),
-				Error:     "some error",
-				LogTime:   logTime,
-			},
-		},
-		{
-			Name:  "from graylog",
-			Input: `2020-04-04T00:00:05+02:00 [DEBG][csnas][zfs.cmd]:  command  exited  without  error  usertime_s="0"  cmd="zfs  send  -i  zroot/ezjail/synapse-12@zrepl_20200329_095518_000  zroot/ezjail/synapse-12@zrepl_20200329_102454_000"  total_time_s="0.101598591"  invocation="85"  systemtime_s="0.041581"`,
-			Expect: &RuntimeLine{
-				Cmd:       "zfs  send  -i  zroot/ezjail/synapse-12@zrepl_20200329_095518_000  zroot/ezjail/synapse-12@zrepl_20200329_102454_000",
-				TotalTime: secs("0.101598591"),
-				Systime:   secs("0.041581"),
-				Usertime:  secs("0"),
-				Error:     "",
-				LogTime:   logTime,
-			},
-		},
-	}
-
-	for _, c := range tcs {
-		t.Run(c.Name, func(t *testing.T) {
-			l, err := parseLogLine(c.Input)
-			t.Logf("l=%v", l)
-			t.Logf("err=%T %v", err, err)
-			if (c.Expect != nil && c.ExpectErr != "") || (c.Expect == nil && c.ExpectErr == "") {
-				t.Fatal("bad test case", c)
-			}
-			if c.Expect != nil {
-				require.Equal(t, *c.Expect, l)
-			}
-			if c.ExpectErr != "" {
-				require.EqualError(t, err, c.ExpectErr)
-			}
-		})
-	}
-
-}
@@ -1,143 +0,0 @@
-// Package zfscmd provides a wrapper around packate os/exec.
-// Functionality provided by the wrapper:
-// - logging start and end of command execution
-// - status report of active commands
-// - prometheus metrics of runtimes
-package zfscmd
-
-import (
-	"context"
-	"io"
-	"os"
-	"os/exec"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/zrepl/zrepl/util/circlog"
-)
-
-type Cmd struct {
-	cmd                       *exec.Cmd
-	ctx                       context.Context
-	mtx                       sync.RWMutex
-	startedAt, waitReturnedAt time.Time
-}
-
-func CommandContext(ctx context.Context, name string, arg ...string) *Cmd {
-	cmd := exec.CommandContext(ctx, name, arg...)
-	return &Cmd{cmd: cmd, ctx: ctx}
-}
-
-// err.(*exec.ExitError).Stderr will NOT be set
-func (c *Cmd) CombinedOutput() (o []byte, err error) {
-	c.startPre()
-	c.startPost(nil)
-	c.waitPre()
-	o, err = c.cmd.CombinedOutput()
-	c.waitPost(err)
-	return
-}
-
-// err.(*exec.ExitError).Stderr will be set
-func (c *Cmd) Output() (o []byte, err error) {
-	c.startPre()
-	c.startPost(nil)
-	c.waitPre()
-	o, err = c.cmd.Output()
-	c.waitPost(err)
-	return
-}
-
-// Careful: err.(*exec.ExitError).Stderr will not be set, even if you don't open an StderrPipe
-func (c *Cmd) StdoutPipeWithErrorBuf() (p io.ReadCloser, errBuf *circlog.CircularLog, err error) {
-	p, err = c.cmd.StdoutPipe()
-	errBuf = circlog.MustNewCircularLog(1 << 15)
-	c.cmd.Stderr = errBuf
-	return p, errBuf, err
-}
-
-type Stdio struct {
-	Stdin  io.ReadCloser
-	Stdout io.Writer
-	Stderr io.Writer
-}
-
-func (c *Cmd) SetStdio(stdio Stdio) {
-	c.cmd.Stdin = stdio.Stdin
-	c.cmd.Stderr = stdio.Stderr
-	c.cmd.Stdout = stdio.Stdout
-}
-
-func (c *Cmd) String() string {
-	return strings.Join(c.cmd.Args, " ") // includes argv[0] if initialized with CommandContext, that's the only way we o it
-}
-
-func (c *Cmd) log() Logger {
-	return getLogger(c.ctx).WithField("cmd", c.String())
-}
-
-func (c *Cmd) Start() (err error) {
-	c.startPre()
-	err = c.cmd.Start()
-	c.startPost(err)
-	return err
-}
-
-// only call this after a successful call to .Start()
-func (c *Cmd) Process() *os.Process {
-	if c.startedAt.IsZero() {
-		panic("calling Process() only allowed after successful call to Start()")
-	}
-	return c.cmd.Process
-}
-
-func (c *Cmd) Wait() (err error) {
-	c.waitPre()
-	err = c.cmd.Wait()
-	if !c.waitReturnedAt.IsZero() {
-		// ignore duplicate waits
-		return err
-	}
-	c.waitPost(err)
-	return err
-}
-
-func (c *Cmd) startPre() {
-	startPreLogging(c, time.Now())
-}
-
-func (c *Cmd) startPost(err error) {
-	now := time.Now()
-
-	c.mtx.Lock()
-	c.startedAt = now
-	c.mtx.Unlock()
-
-	startPostReport(c, err, now)
-	startPostLogging(c, err, now)
-}
-
-func (c *Cmd) waitPre() {
-	waitPreLogging(c, time.Now())
-}
-
-func (c *Cmd) waitPost(err error) {
-	now := time.Now()
-
-	c.mtx.Lock()
-	c.waitReturnedAt = now
-	c.mtx.Unlock()
-
-	waitPostReport(c, now)
-	waitPostLogging(c, err, now)
-	waitPostPrometheus(c, err, now)
-}
-
-// returns 0 if the command did not yet finish
-func (c *Cmd) Runtime() time.Duration {
-	if c.waitReturnedAt.IsZero() {
-		return 0
-	}
-	return c.waitReturnedAt.Sub(c.startedAt)
-}
@@ -1,39 +0,0 @@
-package zfscmd
-
-import (
-	"context"
-
-	"github.com/zrepl/zrepl/logger"
-)
-
-type contextKey int
-
-const (
-	contextKeyLogger contextKey = iota
-	contextKeyJobID
-)
-
-type Logger = logger.Logger
-
-func WithJobID(ctx context.Context, jobID string) context.Context {
-	return context.WithValue(ctx, contextKeyJobID, jobID)
-}
-
-func getJobIDOrDefault(ctx context.Context, def string) string {
-	ret, ok := ctx.Value(contextKeyJobID).(string)
-	if !ok {
-		return def
-	}
-	return ret
-}
-
-func WithLogger(ctx context.Context, log Logger) context.Context {
-	return context.WithValue(ctx, contextKeyLogger, log)
-}
-
-func getLogger(ctx context.Context) Logger {
-	if l, ok := ctx.Value(contextKeyLogger).(Logger); ok {
-		return l
-	}
-	return logger.NewNullLogger()
-}
@@ -1,54 +0,0 @@
-package zfscmd
-
-import (
-	"os/exec"
-	"time"
-)
-
-// Implementation Note:
-//
-// Pre-events logged with debug
-// Post-event without error logged with info
-// Post-events with error _also_ logged with info
-// (Not all errors we observe at this layer) are actual errors in higher-level layers)
-
-func startPreLogging(c *Cmd, now time.Time) {
-	c.log().Debug("starting command")
-}
-
-func startPostLogging(c *Cmd, err error, now time.Time) {
-	if err == nil {
-		c.log().Info("started command")
-	} else {
-		c.log().WithError(err).Error("cannot start command")
-	}
-}
-
-func waitPreLogging(c *Cmd, now time.Time) {
-	c.log().Debug("start waiting")
-}
-
-func waitPostLogging(c *Cmd, err error, now time.Time) {
-
-	var total, system, user float64
-
-	total = c.Runtime().Seconds()
-	if ee, ok := err.(*exec.ExitError); ok {
-		system = ee.ProcessState.SystemTime().Seconds()
-		user = ee.ProcessState.UserTime().Seconds()
-	} else {
-		system = -1
-		user = -1
-	}
-
-	log := c.log().
-		WithField("total_time_s", total).
-		WithField("systemtime_s", system).
-		WithField("usertime_s", user)
-
-	if err == nil {
-		log.Info("command exited without error")
-	} else {
-		log.WithError(err).Info("command exited with error")
-	}
-}
@@ -1,7 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-echo "to stderr" 1>&2
-echo "to stdout"
-
-exit "$1"
@@ -1,89 +0,0 @@
-package zfscmd
-
-import (
-	"bufio"
-	"bytes"
-	"context"
-	"io"
-	"os/exec"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-const testBin = "./zfscmd_platform_test.bash"
-
-func TestCmdStderrBehaviorOutput(t *testing.T) {
-
-	stdout, err := exec.Command(testBin, "0").Output()
-	require.NoError(t, err)
-	require.Equal(t, []byte("to stdout\n"), stdout)
-
-	stdout, err = exec.Command(testBin, "1").Output()
-	assert.Equal(t, []byte("to stdout\n"), stdout)
-	require.Error(t, err)
-	ee, ok := err.(*exec.ExitError)
-	require.True(t, ok)
-	require.Equal(t, ee.Stderr, []byte("to stderr\n"))
-}
-
-func TestCmdStderrBehaviorCombinedOutput(t *testing.T) {
-
-	stdio, err := exec.Command(testBin, "0").CombinedOutput()
-	require.NoError(t, err)
-	require.Equal(t, "to stderr\nto stdout\n", string(stdio))
-
-	stdio, err = exec.Command(testBin, "1").CombinedOutput()
-	require.Equal(t, "to stderr\nto stdout\n", string(stdio))
-	require.Error(t, err)
-	ee, ok := err.(*exec.ExitError)
-	require.True(t, ok)
-	require.Empty(t, ee.Stderr) // !!!! maybe not what one would expect
-}
-
-func TestCmdStderrBehaviorStdoutPipe(t *testing.T) {
-	cmd := exec.Command(testBin, "1")
-	stdoutPipe, err := cmd.StdoutPipe()
-	require.NoError(t, err)
-	err = cmd.Start()
-	require.NoError(t, err)
-	defer cmd.Wait()
-	var stdout bytes.Buffer
-	_, err = io.Copy(&stdout, stdoutPipe)
-	require.NoError(t, err)
-	require.Equal(t, "to stdout\n", stdout.String())
-
-	err = cmd.Wait()
-	require.Error(t, err)
-	ee, ok := err.(*exec.ExitError)
-	require.True(t, ok)
-	require.Empty(t, ee.Stderr) // !!!!! probably not what one would expect if we only redirect stdout
-}
-
-func TestCmdProcessState(t *testing.T) {
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	cmd := exec.CommandContext(ctx, "bash", "-c", "echo running; sleep 3600")
-	stdout, err := cmd.StdoutPipe()
-	require.NoError(t, err)
-	err = cmd.Start()
-	require.NoError(t, err)
-
-	r := bufio.NewReader(stdout)
-	line, err := r.ReadString('\n')
-	require.NoError(t, err)
-	require.Equal(t, "running\n", line)
-
-	// we know it's running and sleeping
-	cancel()
-	err = cmd.Wait()
-	t.Logf("wait err %T\n%s", err, err)
-	require.Error(t, err)
-	ee, ok := err.(*exec.ExitError)
-	require.True(t, ok)
-	require.NotNil(t, ee.ProcessState)
-	require.Contains(t, ee.Error(), "killed")
-
-}
@@ -1,73 +0,0 @@
-package zfscmd
-
-import (
-	"time"
-
-	"github.com/prometheus/client_golang/prometheus"
-)
-
-var metrics struct {
-	totaltime  *prometheus.HistogramVec
-	systemtime *prometheus.HistogramVec
-	usertime   *prometheus.HistogramVec
-}
-
-var timeLabels = []string{"jobid", "zfsbinary", "zfsverb"}
-var timeBuckets = []float64{0.01, 0.1, 0.2, 0.5, 0.75, 1, 2, 5, 10, 60}
-
-func init() {
-	metrics.totaltime = prometheus.NewHistogramVec(prometheus.HistogramOpts{
-		Namespace: "zrepl",
-		Subsystem: "zfscmd",
-		Name:      "runtime",
-		Help:      "number of seconds that the command took from start until wait returned",
-		Buckets:   timeBuckets,
-	}, timeLabels)
-	metrics.systemtime = prometheus.NewHistogramVec(prometheus.HistogramOpts{
-		Namespace: "zrepl",
-		Subsystem: "zfscmd",
-		Name:      "systemtime",
-		Help:      "https://golang.org/pkg/os/#ProcessState.SystemTime",
-		Buckets:   timeBuckets,
-	}, timeLabels)
-	metrics.usertime = prometheus.NewHistogramVec(prometheus.HistogramOpts{
-		Namespace: "zrepl",
-		Subsystem: "zfscmd",
-		Name:      "usertime",
-		Help:      "https://golang.org/pkg/os/#ProcessState.UserTime",
-		Buckets:   timeBuckets,
-	}, timeLabels)
-
-}
-
-func RegisterMetrics(r prometheus.Registerer) {
-	r.MustRegister(metrics.totaltime)
-	r.MustRegister(metrics.systemtime)
-	r.MustRegister(metrics.usertime)
-}
-
-func waitPostPrometheus(c *Cmd, err error, now time.Time) {
-
-	if len(c.cmd.Args) < 2 {
-		getLogger(c.ctx).WithField("args", c.cmd.Args).
-			Warn("prometheus: cannot turn zfs command into metric")
-		return
-	}
-
-	// Note: do not start parsing other aspects
-	// of the ZFS command line. This is not the suitable layer
-	// for such a task.
-
-	jobid := getJobIDOrDefault(c.ctx, "_nojobid")
-
-	labelValues := []string{jobid, c.cmd.Args[0], c.cmd.Args[1]}
-
-	metrics.totaltime.
-		WithLabelValues(labelValues...).
-		Observe(c.Runtime().Seconds())
-	metrics.systemtime.WithLabelValues(labelValues...).
-		Observe(c.cmd.ProcessState.SystemTime().Seconds())
-	metrics.usertime.WithLabelValues(labelValues...).
-		Observe(c.cmd.ProcessState.UserTime().Seconds())
-
-}
@@ -1,68 +0,0 @@
-package zfscmd
-
-import (
-	"fmt"
-	"sync"
-	"time"
-)
-
-type Report struct {
-	Active []ActiveCommand
-}
-
-type ActiveCommand struct {
-	Path      string
-	Args      []string
-	StartedAt time.Time
-}
-
-func GetReport() *Report {
-	active.mtx.RLock()
-	defer active.mtx.RUnlock()
-	var activeCommands []ActiveCommand
-	for c := range active.cmds {
-		c.mtx.RLock()
-		activeCommands = append(activeCommands, ActiveCommand{
-			Path:      c.cmd.Path,
-			Args:      c.cmd.Args,
-			StartedAt: c.startedAt,
-		})
-		c.mtx.RUnlock()
-	}
-	return &Report{
-		Active: activeCommands,
-	}
-}
-
-var active struct {
-	mtx  sync.RWMutex
-	cmds map[*Cmd]bool
-}
-
-func init() {
-	active.cmds = make(map[*Cmd]bool)
-}
-
-func startPostReport(c *Cmd, err error, now time.Time) {
-	if err != nil {
-		return
-	}
-
-	active.mtx.Lock()
-	prev := active.cmds[c]
-	if prev {
-		panic("impl error: duplicate active command")
-	}
-	active.cmds[c] = true
-	active.mtx.Unlock()
-}
-
-func waitPostReport(c *Cmd, now time.Time) {
-	active.mtx.Lock()
-	defer active.mtx.Unlock()
-	prev := active.cmds[c]
-	if !prev {
-		panic(fmt.Sprintf("impl error: onWaitDone must only be called on an active command: %s", c))
-	}
-	delete(active.cmds, c)
-}