zfs: rewrite SendStream, fix bug in Close() on FreeBSD, add platformtests

This commit was motivated by https://github.com/zrepl/zrepl/issues/495
where, on FreeBSD with OpenZFS 2.0, a SendStream.Close() call might wait indefinitely for `zfs send` to exit.
The reason is that, due to the refactoring done for redacted send & recv
(https://github.com/openzfs/zfs/commit/30af21b02569ac192f52ce6e6511015f8a8d5729),
the `dump_bytes` function, which writes to the pipe, executes in a separate thread (synctask taskq) iff not `HAVE_LARGE_STACKS`.
The `zfs send` process/thread waits for that taskq thread using an uninterruptible primitive.
So when we SIGKILL `zfs send`, that signal doesn't reach the right thread to interrupt the pipe write.

Theoretically this affects both Linux and FreeBSD, but most Linux users `HAVE_LARGE_STACKS` and since https://github.com/penzfs/zfs/pull/12350/files OpenZFS on FreeBSD `HAVE_LARGE_STACKS` as well.
However, at least until FreeBSD 13.1, possibly for the entire 13 lifecycle, we're going to have to live with that oddity.

Measures taken in this commit:
- Report the behavior as an upstream bug https://github.com/openzfs/zfs/issues/12500
- Change SendStream code so that it closes zrepl's read-end of the pipe (see comment in code)
- Clean up and make explicit SendStream's state handling
- Write extensive platformtests for SendStream
    - They pass on my Linux install and on FreeBSD 12
    - FreeBSD 13 still needs testing.

fixes https://github.com/zrepl/zrepl/issues/495

client/status/viewmodel/bytesprogresshistory.go

@@ -11,6 +11,7 @@ type bytesProgressHistory struct {
 	last        *byteProgressMeasurement // pointer as poor man's optional
 	changeCount int
 	lastChange  time.Time
+	bpsAvg      float64
 }
 
 func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64, changeCount int) {
@@ -37,8 +38,11 @@ func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64
 	deltaT := time.Since(p.last.time)
 	rate := float64(deltaV) / deltaT.Seconds()
 
+	factor := 0.3
+	p.bpsAvg = (1-factor)*p.bpsAvg + factor*rate
+
 	p.last.time = time.Now()
 	p.last.val = currentVal
 
-	return int64(rate), p.changeCount
+	return int64(p.bpsAvg), p.changeCount
 }

config/config.go

@@ -13,7 +13,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/zrepl/yaml-config"
 
-	"github.com/zrepl/zrepl/util/datasizeunit"
 	zfsprop "github.com/zrepl/zrepl/zfs/property"
 )
 
@@ -88,8 +87,6 @@ type SendOptions struct {
 	Compressed       bool `yaml:"compressed,optional,default=false"`
 	EmbeddedData     bool `yaml:"embbeded_data,optional,default=false"`
 	Saved            bool `yaml:"saved,optional,default=false"`
-
-	BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
 }
 
 type RecvOptions struct {
@@ -99,15 +96,6 @@ type RecvOptions struct {
 	// Reencrypt bool `yaml:"reencrypt"`
 
 	Properties *PropertyRecvOptions `yaml:"properties,fromdefaults"`
-
-	BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
-}
-
-var _ yaml.Unmarshaler = &datasizeunit.Bits{}
-
-type BandwidthLimit struct {
-	Max            datasizeunit.Bits `yaml:"max,default=-1 B"`
-	BucketCapacity datasizeunit.Bits `yaml:"bucket_capacity,default=128 KiB"`
 }
 
 type Replication struct {
@@ -125,6 +113,10 @@ type ReplicationOptionsConcurrency struct {
 	SizeEstimates int `yaml:"size_estimates,optional,default=4"`
 }
 
+func (l *RecvOptions) SetDefault() {
+	*l = RecvOptions{Properties: &PropertyRecvOptions{}}
+}
+
 type PropertyRecvOptions struct {
 	Inherit  []zfsprop.Property          `yaml:"inherit,optional"`
 	Override map[zfsprop.Property]string `yaml:"override,optional"`

config/samples/bandwidth_limit.yml

@@ -1,41 +0,0 @@
-
-jobs:
-  - type: sink
-    name: "limited_sink"
-    root_fs: "fs0"
-    recv:
-      bandwidth_limit:
-        max: 12345 B
-    serve:
-      type: local
-      listener_name: localsink
-
-  - type: push
-    name: "limited_push"
-    connect:
-      type: local
-      listener_name: localsink
-      client_identity: local_backup
-    filesystems: {
-      "root<": true,
-    }
-    send:
-      bandwidth_limit:
-        max: 54321 B
-        bucket_capacity: 1024 B
-    snapshotting:
-      type: manual
-    pruning:
-      keep_sender:
-        - type: last_n
-          count: 1
-      keep_receiver:
-        - type: last_n
-          count: 1
-
-  - type: sink
-    name: "nolimit_sink"
-    root_fs: "fs1"
-    serve:
-      type: local
-      listener_name: localsink

daemon/job/build_jobs.go

@@ -8,7 +8,6 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/util/bandwidthlimit"
 )
 
 func JobsFromConfig(c *config.Config) ([]Job, error) {
@@ -108,13 +107,3 @@ func validateReceivingSidesDoNotOverlap(receivingRootFSs []string) error {
 	}
 	return nil
 }
-
-func buildBandwidthLimitConfig(in *config.BandwidthLimit) (c bandwidthlimit.Config, _ error) {
-	if in.Max.ToBytes() > 0 && int64(in.Max.ToBytes()) == 0 {
-		return c, fmt.Errorf("bandwidth limit `max` is too small, must at least specify one byte")
-	}
-	return bandwidthlimit.Config{
-		Max:            int64(in.Max.ToBytes()),
-		BucketCapacity: int64(in.BucketCapacity.ToBytes()),
-	}, nil
-}

daemon/job/build_jobs_sendrecvoptions.go

@@ -22,12 +22,7 @@ func buildSenderConfig(in SendingJobConfig, jobID endpoint.JobID) (*endpoint.Sen
 		return nil, errors.Wrap(err, "cannot build filesystem filter")
 	}
 	sendOpts := in.GetSendOptions()
-	bwlim, err := buildBandwidthLimitConfig(sendOpts.BandwidthLimit)
-	if err != nil {
-		return nil, errors.Wrap(err, "cannot build bandwith limit config")
-	}
-
-	sc := &endpoint.SenderConfig{
+	return &endpoint.SenderConfig{
 		FSF:   fsf,
 		JobID: jobID,
 
@@ -39,15 +34,7 @@ func buildSenderConfig(in SendingJobConfig, jobID endpoint.JobID) (*endpoint.Sen
 		SendCompressed:       sendOpts.Compressed,
 		SendEmbeddedData:     sendOpts.EmbeddedData,
 		SendSaved:            sendOpts.Saved,
-
-		BandwidthLimit: bwlim,
-	}
-
-	if err := sc.Validate(); err != nil {
-		return nil, errors.Wrap(err, "cannot build sender config")
-	}
-
-	return sc, nil
+	}, nil
 }
 
 type ReceivingJobConfig interface {
@@ -66,12 +53,6 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
 	}
 
 	recvOpts := in.GetRecvOptions()
-
-	bwlim, err := buildBandwidthLimitConfig(recvOpts.BandwidthLimit)
-	if err != nil {
-		return rc, errors.Wrap(err, "cannot build bandwith limit config")
-	}
-
 	rc = endpoint.ReceiverConfig{
 		JobID:                      jobID,
 		RootWithoutClientComponent: rootFs,
@@ -79,8 +60,6 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
 
 		InheritProperties:  recvOpts.Properties.Inherit,
 		OverrideProperties: recvOpts.Properties.Override,
-
-		BandwidthLimit: bwlim,
 	}
 	if err := rc.Validate(); err != nil {
 		return rc, errors.Wrap(err, "cannot build receiver config")

daemon/job/build_jobs_test.go

@@ -119,14 +119,6 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 		t.Errorf("glob failed: %+v", err)
 	}
 
-	type additionalCheck struct {
-		state int
-		test  func(t *testing.T, jobs []Job)
-	}
-	additionalChecks := map[string]*additionalCheck{
-		"bandwidth_limit.yml": {test: testSampleConfig_BandwidthLimit},
-	}
-
 	for _, p := range paths {
 
 		if path.Ext(p) != ".yml" {
@@ -134,20 +126,10 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 			continue
 		}
 
-		filename := path.Base(p)
-		t.Logf("checking for presence additonal checks for file %q", filename)
-		additionalCheck := additionalChecks[filename]
-		if additionalCheck == nil {
-			t.Logf("no additional checks")
-		} else {
-			t.Logf("additional check present")
-			additionalCheck.state = 1
-		}
-
 		t.Run(p, func(t *testing.T) {
 			c, err := config.ParseConfig(p)
 			if err != nil {
-				t.Fatalf("error parsing %s:\n%+v", p, err)
+				t.Errorf("error parsing %s:\n%+v", p, err)
 			}
 
 			t.Logf("file: %s", p)
@@ -156,57 +138,11 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 			tls.FakeCertificateLoading(t)
 			jobs, err := JobsFromConfig(c)
 			t.Logf("jobs: %#v", jobs)
-			require.NoError(t, err)
-
-			if additionalCheck != nil {
-				additionalCheck.test(t, jobs)
-				additionalCheck.state = 2
-			}
+			assert.NoError(t, err)
 		})
 
 	}
 
-	for basename, c := range additionalChecks {
-		if c.state == 0 {
-			panic("univisited additional check " + basename)
-		}
-	}
-
-}
-
-func testSampleConfig_BandwidthLimit(t *testing.T, jobs []Job) {
-	require.Len(t, jobs, 3)
-
-	{
-		limitedSink, ok := jobs[0].(*PassiveSide)
-		require.True(t, ok, "%T", jobs[0])
-		limitedSinkMode, ok := limitedSink.mode.(*modeSink)
-		require.True(t, ok, "%T", limitedSink)
-
-		assert.Equal(t, int64(12345), limitedSinkMode.receiverConfig.BandwidthLimit.Max)
-		assert.Equal(t, int64(1<<17), limitedSinkMode.receiverConfig.BandwidthLimit.BucketCapacity)
-	}
-
-	{
-		limitedPush, ok := jobs[1].(*ActiveSide)
-		require.True(t, ok, "%T", jobs[1])
-		limitedPushMode, ok := limitedPush.mode.(*modePush)
-		require.True(t, ok, "%T", limitedPush)
-
-		assert.Equal(t, int64(54321), limitedPushMode.senderConfig.BandwidthLimit.Max)
-		assert.Equal(t, int64(1024), limitedPushMode.senderConfig.BandwidthLimit.BucketCapacity)
-	}
-
-	{
-		unlimitedSink, ok := jobs[2].(*PassiveSide)
-		require.True(t, ok, "%T", jobs[2])
-		unlimitedSinkMode, ok := unlimitedSink.mode.(*modeSink)
-		require.True(t, ok, "%T", unlimitedSink)
-
-		max := unlimitedSinkMode.receiverConfig.BandwidthLimit.Max
-		assert.Less(t, max, int64(0), max, "unlimited mode <=> negative value for .Max, see bandwidthlimit.Config")
-	}
-
 }
 
 func TestReplicationOptions(t *testing.T) {

docs/configuration/sendrecvoptions.rst

@@ -36,9 +36,6 @@ See the `upstream man page <https://openzfs.github.io/openzfs-docs/man/8/zfs-sen
     * - ``encrypted``
       -
       - Specific to zrepl, :ref:`see below <job-send-options-encrypted>`.
-    * - ``bandwidth_limit``
-      -
-      - Specific to zrepl, :ref:`see below <job-send-recv-options-bandwidth-limit>`.
     * - ``raw``
       - ``-w``
       - Use ``encrypted`` to only allow encrypted sends.
@@ -141,7 +138,6 @@ Recv Options
          override: {
            "org.openzfs.systemd:ignore": "on"
          }
-       bandwidth_limit: ... # see below
      ...
 
 .. _job-recv-options--inherit-and-override:
@@ -216,25 +212,3 @@ and property replication is enabled, the receiver must :ref:`inherit the followi
 * ``keylocation``
 * ``keyformat``
 * ``encryption``
-
-Common Options
-~~~~~~~~~~~~~~
-
-.. _job-send-recv-options-bandwidth-limit:
-
-Bandwidth Limit (send & recv)
------------------------------
-
-::
-
-   bandwidth_limit:
-     max: 23.5 MiB # -1 is the default and disabled rate limiting
-     bucket_capacity: # token bucket capacity in bytes; defaults to 128KiB
-
-Both ``send`` and ``recv`` can be limited to a maximum bandwidth through ``bandwidth_limit``.
-For most users, it should be sufficient to just set ``bandwidth_limit.max``.
-The ``bandwidth_limit.bucket_capacity`` refers to the `token bucket size <https://github.com/juju/ratelimit>`_.
-
-The bandwidth limit only applies to the payload data, i.e., the ZFS send stream.
-It does not account for transport protocol overheads.
-The scope is the job level, i.e., all :ref:`concurrent <replication-option-concurrency>` sends or incoming receives of a job share the bandwidth limit.

docs/supporters.rst

@@ -32,8 +32,6 @@ We would like to thank the following people and organizations for supporting zre
 
    <div class="fa fa-code" style="width: 1em;"></div>
 
-* |supporter-gold| Prominic.NET, Inc.
-* |supporter-std| Torsten Blum
 * |supporter-gold| Cyberiada GmbH
 * |supporter-std| `Gordon Schulz <https://github.com/azmodude>`_
 * |supporter-std| `@jwittlincohen <https://github.com/jwittlincohen>`_

endpoint/endpoint.go

@@ -14,7 +14,6 @@ import (
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/replication/logic/pdu"
-	"github.com/zrepl/zrepl/util/bandwidthlimit"
 	"github.com/zrepl/zrepl/util/chainedio"
 	"github.com/zrepl/zrepl/util/chainlock"
 	"github.com/zrepl/zrepl/util/envconst"
@@ -35,8 +34,6 @@ type SenderConfig struct {
 	SendCompressed       bool
 	SendEmbeddedData     bool
 	SendSaved            bool
-
-	BandwidthLimit bandwidthlimit.Config
 }
 
 func (c *SenderConfig) Validate() error {
@@ -47,9 +44,6 @@ func (c *SenderConfig) Validate() error {
 	if _, err := StepHoldTag(c.JobID); err != nil {
 		return fmt.Errorf("JobID cannot be used for hold tag: %s", err)
 	}
-	if err := bandwidthlimit.ValidateConfig(c.BandwidthLimit); err != nil {
-		return errors.Wrap(err, "`BandwidthLimit` field invalid")
-	}
 	return nil
 }
 
@@ -60,21 +54,16 @@ type Sender struct {
 	FSFilter zfs.DatasetFilter
 	jobId    JobID
 	config   SenderConfig
-	bwLimit  bandwidthlimit.Wrapper
 }
 
 func NewSender(conf SenderConfig) *Sender {
 	if err := conf.Validate(); err != nil {
 		panic("invalid config" + err.Error())
 	}
-
-	ratelimiter := bandwidthlimit.WrapperFromConfig(conf.BandwidthLimit)
-
 	return &Sender{
 		FSFilter: conf.FSF,
 		jobId:    conf.JobID,
 		config:   conf,
-		bwLimit:  ratelimiter,
 	}
 }
 
@@ -312,16 +301,12 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		abstractionsCacheSingleton.TryBatchDestroy(ctx, s.jobId, sendArgs.FS, destroyTypes, keep, check)
 	}()
 
-	var sendStream io.ReadCloser
-	sendStream, err = zfs.ZFSSend(ctx, sendArgs)
+	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
 	if err != nil {
 		// it's ok to not destroy the abstractions we just created here, a new send attempt will take care of it
 		return nil, nil, errors.Wrap(err, "zfs send failed")
 	}
 
-	// apply rate limit
-	sendStream = s.bwLimit.WrapReadCloser(sendStream)
-
 	return res, sendStream, nil
 }
 
@@ -454,8 +439,6 @@ type ReceiverConfig struct {
 
 	InheritProperties  []zfsprop.Property
 	OverrideProperties map[zfsprop.Property]string
-
-	BandwidthLimit bandwidthlimit.Config
 }
 
 func (c *ReceiverConfig) copyIn() {
@@ -492,11 +475,6 @@ func (c *ReceiverConfig) Validate() error {
 	if c.RootWithoutClientComponent.Length() <= 0 {
 		return errors.New("RootWithoutClientComponent must not be an empty dataset path")
 	}
-
-	if err := bandwidthlimit.ValidateConfig(c.BandwidthLimit); err != nil {
-		return errors.Wrap(err, "`BandwidthLimit` field invalid")
-	}
-
 	return nil
 }
 
@@ -506,8 +484,6 @@ type Receiver struct {
 
 	conf ReceiverConfig // validated
 
-	bwLimit bandwidthlimit.Wrapper
-
 	recvParentCreationMtx *chainlock.L
 }
 
@@ -519,7 +495,6 @@ func NewReceiver(config ReceiverConfig) *Receiver {
 	return &Receiver{
 		conf:                  config,
 		recvParentCreationMtx: chainlock.New(),
-		bwLimit:               bandwidthlimit.WrapperFromConfig(config.BandwidthLimit),
 	}
 }
 
@@ -812,9 +787,6 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 		return nil, errors.Wrap(err, "cannot determine whether we can use resumable send & recv")
 	}
 
-	// apply rate limit
-	receive = s.bwLimit.WrapReadCloser(receive)
-
 	var peek bytes.Buffer
 	var MaxPeek = envconst.Int64("ZREPL_ENDPOINT_RECV_PEEK_SIZE", 1<<20)
 	log.WithField("max_peek_bytes", MaxPeek).Info("peeking incoming stream")

go.mod

@@ -15,7 +15,6 @@ require (
 	github.com/golang/protobuf v1.4.3
 	github.com/google/uuid v1.1.2
 	github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8
-	github.com/juju/ratelimit v1.0.1
 	github.com/kisielk/gotool v1.0.0 // indirect
 	github.com/kr/pretty v0.1.0
 	github.com/leodido/go-urn v1.2.1 // indirect

go.sum

@@ -164,8 +164,6 @@ github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8 h1:+dKzeuiDYbD/Cfi/s
 github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8/go.mod h1:yL958EeXv8Ylng6IfnvG4oflryUi3vgA3xPs9hmII1s=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
-github.com/juju/ratelimit v1.0.1 h1:+7AIFJVQ0EQgq/K9+0Krm7m530Du7tIz0METWzN0RgY=
-github.com/juju/ratelimit v1.0.1/go.mod h1:qapgC/Gy+xNh9UxzV13HGGl/6UXNN+ct+vwSgWNm/qk=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 h1:uC1QfSlInpQF+M0ao65imhwqKnz3Q2z/d8PWZRMQvDM=

packaging/deb/Dockerfile

@@ -1,13 +1,9 @@
 FROM debian:latest
 
-# binutils are for cross-compilation to work in bullseye
 RUN apt-get update && apt-get install -y \
 	build-essential \
     devscripts \
-    dh-exec \
-	binutils-aarch64-linux-gnu \
-	binutils-arm-linux-gnueabihf  \
-	binutils-i686-linux-gnu
+    dh-exec
 
 RUN mkdir -p /build/src && chmod -R 0777 /build
 

platformtest/tests/generated_cases.go

@@ -34,5 +34,10 @@ var Cases = []Case{BatchDestroy,
 	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden__EncryptionSupported_true,
 	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
 	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
+	SendStreamCloseAfterBlockedOnPipeWrite,
+	SendStreamCloseAfterEOFRead,
+	SendStreamMultipleCloseAfterEOF,
+	SendStreamMultipleCloseBeforeEOF,
+	SendStreamNonEOFReadErrorHandling,
 	UndestroyableSnapshotParsing,
 }

platformtest/tests/replication.go

@@ -21,7 +21,6 @@ import (
 	"github.com/zrepl/zrepl/replication/logic"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/replication/report"
-	"github.com/zrepl/zrepl/util/bandwidthlimit"
 	"github.com/zrepl/zrepl/util/limitio"
 	"github.com/zrepl/zrepl/util/nodefault"
 	"github.com/zrepl/zrepl/zfs"
@@ -65,10 +64,9 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 	}
 
 	senderConfig := endpoint.SenderConfig{
-		FSF:            i.sfilter.AsFilter(),
-		Encrypt:        &nodefault.Bool{B: false},
-		JobID:          i.sjid,
-		BandwidthLimit: bandwidthlimit.NoLimitConfig(),
+		FSF:     i.sfilter.AsFilter(),
+		Encrypt: &nodefault.Bool{B: false},
+		JobID:   i.sjid,
 	}
 	if i.senderConfigHook != nil {
 		i.senderConfigHook(&senderConfig)
@@ -78,7 +76,6 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		JobID:                      i.rjid,
 		AppendClientIdentity:       false,
 		RootWithoutClientComponent: mustDatasetPath(i.rfsRoot),
-		BandwidthLimit:             bandwidthlimit.NoLimitConfig(),
 	}
 	if i.receiverConfigHook != nil {
 		i.receiverConfigHook(&receiverConfig)

platformtest/tests/sendStream.go

@@ -0,0 +1,186 @@
+package tests
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"os/exec"
+	"path"
+	"syscall"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/unix"
+
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/util/nodefault"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+func sendStreamTest(ctx *platformtest.Context) *zfs.SendStream {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		DESTROYROOT
+		CREATEROOT
+		+  "sender"
+	`)
+
+	fs := fmt.Sprintf("%s/sender", ctx.RootDataset)
+
+	fsmpo, err := zfs.ZFSGetMountpoint(ctx, fs)
+	require.NoError(ctx, err)
+
+	writeDummyData(path.Join(fsmpo.Mountpoint, "dummy.data"), 1<<26)
+	mustSnapshot(ctx, fs+"@1")
+	snap := fsversion(ctx, fs, "@1")
+	snapSendArg := snap.ToSendArgVersion()
+
+	sendArgs, err := zfs.ZFSSendArgsUnvalidated{
+		FS:   fs,
+		From: nil,
+		To:   &snapSendArg,
+		ZFSSendFlags: zfs.ZFSSendFlags{
+			Encrypted: &nodefault.Bool{B: false},
+		},
+	}.Validate(ctx)
+	require.NoError(ctx, err)
+
+	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
+	require.NoError(ctx, err)
+
+	return sendStream
+
+}
+
+func SendStreamCloseAfterBlockedOnPipeWrite(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	// let the pipe buffer fill and the zfs process block uninterruptibly
+
+	ctx.Logf("waiting for pipe write to block")
+	time.Sleep(5 * time.Second) // XXX need a platform-neutral way to detect that the pipe is full and the writer is blocked
+
+	ctx.Logf("closing send stream")
+	err := sendStream.Close() // this is what this test case is about
+	ctx.Logf("close error: %T %s", err, err)
+	require.NoError(ctx, err)
+
+	exitErrZfsError := sendStream.TestOnly_ExitErr()
+	require.Contains(ctx, exitErrZfsError.Error(), "signal")
+	require.Error(ctx, exitErrZfsError)
+	exitErr, ok := exitErrZfsError.WaitErr.(*exec.ExitError)
+	require.True(ctx, ok)
+	if exitErr.Exited() {
+		// some ZFS impls (FreeBSD 12) behaves that way
+		return
+	}
+
+	// ProcessState is only available after exit
+	// => use as proxy that the process was wait()ed upon and is gone
+	ctx.Logf("%#v", exitErr.ProcessState)
+	require.NotNil(ctx, exitErr.ProcessState)
+	// and let's verify that the process got killed, so that we know it was the call to .Close() above
+	waitStatus := exitErr.ProcessState.Sys().(syscall.WaitStatus)
+	ctx.Logf("wait status: %#v", waitStatus)
+	ctx.Logf("exit status: %v", waitStatus.ExitStatus())
+	require.True(ctx, waitStatus.Signaled())
+	switch waitStatus.Signal() {
+	case unix.SIGKILL:
+		fallthrough
+	case unix.SIGPIPE:
+		// ok
+
+	default:
+		ctx.Errorf("%T %s\n%v", waitStatus.Signal(), waitStatus.Signal(), waitStatus.Signal())
+		ctx.FailNow()
+	}
+}
+
+func SendStreamCloseAfterEOFRead(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	_, err := io.Copy(ioutil.Discard, sendStream)
+	require.NoError(ctx, err)
+
+	var buf [128]byte
+	n, err := sendStream.Read(buf[:])
+	require.Zero(ctx, n)
+	require.Equal(ctx, io.EOF, err)
+
+	err = sendStream.Close()
+	require.NoError(ctx, err)
+
+	n, err = sendStream.Read(buf[:])
+	require.Zero(ctx, n)
+	require.Equal(ctx, os.ErrClosed, err, "same read error should be returned")
+}
+
+func SendStreamMultipleCloseAfterEOF(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	_, err := io.Copy(ioutil.Discard, sendStream)
+	require.NoError(ctx, err)
+
+	var buf [128]byte
+	n, err := sendStream.Read(buf[:])
+	require.Zero(ctx, n)
+	require.Equal(ctx, io.EOF, err)
+
+	err = sendStream.Close()
+	require.NoError(ctx, err)
+
+	err = sendStream.Close()
+	require.Equal(ctx, os.ErrClosed, err)
+}
+
+func SendStreamMultipleCloseBeforeEOF(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	err := sendStream.Close()
+	require.NoError(ctx, err)
+
+	err = sendStream.Close()
+	require.Equal(ctx, os.ErrClosed, err)
+}
+
+type failingReadCloser struct {
+	err error
+}
+
+var _ io.ReadCloser = &failingReadCloser{}
+
+func (c *failingReadCloser) Read(p []byte) (int, error) { return 0, c.err }
+func (c *failingReadCloser) Close() error               { return c.err }
+
+func SendStreamNonEOFReadErrorHandling(ctx *platformtest.Context) {
+
+	sendStream := sendStreamTest(ctx)
+
+	var buf [128]byte
+	n, err := sendStream.Read(buf[:])
+	require.Equal(ctx, len(buf), n)
+	require.NoError(ctx, err)
+
+	var mockError = fmt.Errorf("taeghaefow4piesahwahjocu7ul5tiachaiLipheijae8ooZ8Pies8shohGee9feeTeirai5aiFeiyaecai4kiaLoh4azeih0tea")
+	mock := &failingReadCloser{err: mockError}
+	orig := sendStream.TestOnly_ReplaceStdoutReader(mock)
+
+	n, err = sendStream.Read(buf[:])
+	require.Equal(ctx, 0, n)
+	require.Equal(ctx, mockError, err)
+
+	if sendStream.TestOnly_ReplaceStdoutReader(orig) != mock {
+		panic("incorrect test impl")
+	}
+
+	err = sendStream.Close()
+	require.NoError(ctx, err) // if we can't kill the child then this will be a flaky test, but let's assume we can kill the child
+
+	err = sendStream.Close()
+	require.Equal(ctx, os.ErrClosed, err)
+}

rpc/dataconn/dataconn_server.go

@@ -241,7 +241,7 @@ func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *strea
 		err := c.SendStream(ctx, sendStream, ZFSStream)
 		closeErr := sendStream.Close()
 		if closeErr != nil {
-			s.log.WithError(closeErr).Error("cannot close send stream")
+			s.log.WithError(err).Error("cannot close send stream")
 		}
 		if err != nil {
 			s.log.WithError(err).Error("cannot write send stream")

rpc/transportmux/transportmux.go

@@ -7,7 +7,6 @@ package transportmux
 
 import (
 	"context"
-	"os"
 	"sync/atomic"
 	"syscall"
 
@@ -16,7 +15,6 @@ import (
 	"net"
 	"time"
 
-	"github.com/kr/pretty"
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/transport"
@@ -153,7 +151,6 @@ func Demux(ctx context.Context, rawListener transport.AuthenticatedListener, lab
 
 			rawConn, err := rawListener.Accept(ctx)
 			if err != nil {
-				fmt.Fprintf(os.Stderr, "transportmux.Demux: rawListener.Accept() returned error: %T %s\n%s\n", err, err, pretty.Sprint(err))
 				if ctx.Err() != nil {
 					return
 				}

rpc/versionhandshake/versionhandshake.go

@@ -10,12 +10,9 @@ import (
 	"fmt"
 	"io"
 	"net"
-	"os"
 	"strings"
 	"time"
 	"unicode/utf8"
-
-	"github.com/kr/pretty"
 )
 
 type HandshakeMessage struct {
@@ -96,7 +93,6 @@ func (m *HandshakeMessage) Encode() ([]byte, error) {
 func (m *HandshakeMessage) DecodeReader(r io.Reader, maxLen int) error {
 	var lenAndSpace [11]byte
 	if _, err := io.ReadFull(r, lenAndSpace[:]); err != nil {
-		fmt.Fprintf(os.Stderr, "HandshakeMessage.DecodeReader error: %T\n%s", err, pretty.Sprint(err))
 		return hsIOErr(err, "error reading protocol banner length: %s", err)
 	}
 	if !utf8.Valid(lenAndSpace[:]) {

rpc/versionhandshake/versionhandshake_transport_wrappers.go

@@ -2,9 +2,7 @@ package versionhandshake
 
 import (
 	"context"
-	"fmt"
 	"net"
-	"os"
 	"time"
 
 	"github.com/zrepl/zrepl/transport"
@@ -25,7 +23,6 @@ func (c HandshakeConnecter) Connect(ctx context.Context) (transport.Wire, error)
 		dl = time.Now().Add(c.timeout)
 	}
 	if err := DoHandshakeCurrentVersion(conn, dl); err != nil {
-		fmt.Fprintf(os.Stderr, "HandshakeConnecter error: %T\n\t%s\n\t%v\n\t%#v\n\n", err, err, err, err)
 		conn.Close()
 		return nil, err
 	}

transport/tcp/serve_tcp_ipmap_test.go

@@ -91,12 +91,15 @@ func TestIPMap(t *testing.T) {
 				"fde4:8dba:82e1::/64": "sub64-*",
 			},
 			expect: map[string]testCaseExpect{
-				"10.1.2.3":      {expectNoMapping: true},
-				"192.168.23.1":  {expectIdent: "db-192.168.23.1"},
-				"192.168.42.1":  {expectIdent: "web-192.168.42.1"},
-				"192.168.23.23": {expectIdent: "db-twentythree"},
-				"10.1.4.5":      {expectIdent: "my-10.1.4.5-server"},
+				"10.1.2.3":        {expectNoMapping: true},
+				"192.168.23.1":    {expectIdent: "db-192.168.23.1"},
+				"192.168.23.23":   {expectIdent: "db-twentythree"},
+				"192.168.023.001": {expectIdent: "db-192.168.23.1"},
+				"10.1.4.5":        {expectIdent: "my-10.1.4.5-server"},
 
+				// normalization
+				"192.168.42.1":    {expectIdent: "web-192.168.42.1"},
+				"192.168.042.001": {expectIdent: "web-192.168.42.1"},
 				// v6 matching
 				"fe80::23:42%eth1": {expectIdent: "san-fe80::23:42-eth1"},
 				"fe80::23:42%eth2": {expectNoMapping: true},
@@ -176,8 +179,7 @@ func TestIPMap(t *testing.T) {
 			for input, expect := range c.expect {
 				// reuse newIPMapEntry to parse test case input
 				// "test" is not used during testing but must not be empty.
-				ipMapEntry, err := newIPMapEntry(input, "test")
-				require.NoError(t, err)
+				ipMapEntry, _ := newIPMapEntry(input, "test")
 				ones, bits := ipMapEntry.subnet.Mask.Size()
 				require.Equal(t, bits, net.IPv6len*8, "and we know ipMapEntry always expands its IPs to 16bytes")
 				require.Equal(t, ones, net.IPv6len*8, "test case addresses must be fully specified")

transport/tls/connect_tls.go

@@ -3,9 +3,7 @@ package tls
 import (
 	"context"
 	"crypto/tls"
-	"fmt"
 	"net"
-	"os"
 
 	"github.com/pkg/errors"
 
@@ -50,7 +48,6 @@ func TLSConnecterFromConfig(in *config.TLSConnect) (*TLSConnecter, error) {
 func (c *TLSConnecter) Connect(dialCtx context.Context) (transport.Wire, error) {
 	conn, err := c.dialer.DialContext(dialCtx, "tcp", c.Address)
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "tls connecter error %T\n\t%s\n\t%v\n\t%#v\n\n", err, err, err, err)
 		return nil, err
 	}
 	tcpConn := conn.(*net.TCPConn)

transport/tls/issue-517/main.go

@@ -1,140 +0,0 @@
-package main
-
-import (
-	"context"
-	"flag"
-	"io"
-	"log"
-	"os"
-	"strings"
-	"time"
-
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/transport/tls"
-)
-
-var servConf = config.TLSServe{
-	ServeCommon: config.ServeCommon{
-		Type: "tls",
-	},
-	HandshakeTimeout: 10 * time.Second,
-}
-
-var clientConf = config.TLSConnect{
-	ConnectCommon: config.ConnectCommon{
-		Type: "",
-	},
-	Address:     "",
-	Ca:          "",
-	Cert:        "",
-	Key:         "",
-	ServerCN:    "",
-	DialTimeout: 10 * time.Second,
-}
-
-var ca string
-var mode string
-
-func main() {
-
-	flag.StringVar(&mode, "mode", "", "server|client")
-
-	flag.StringVar(&ca, "ca", "", "path")
-
-	flag.StringVar(&servConf.Listen, "serve.listen", "", "")
-	flag.StringVar(&servConf.Cert, "serve.cert", "", "path")
-	flag.StringVar(&servConf.Key, "serve.key", "", "path")
-	var clientCN string
-	flag.StringVar(&clientCN, "serve.client_cn", "", "")
-
-	flag.StringVar(&clientConf.Address, "client.address", "", "")
-	flag.StringVar(&clientConf.Cert, "client.cert", "", "path")
-	flag.StringVar(&clientConf.Key, "client.key", "", "path")
-	flag.StringVar(&clientConf.ServerCN, "client.server_cn", "", "")
-
-	flag.Parse()
-
-	servConf.ClientCNs = append(servConf.ClientCNs, clientCN)
-
-	servConf.Ca = ca
-	clientConf.Ca = ca
-
-	switch mode {
-	case "server":
-		server()
-	case "client":
-		client()
-	default:
-		panic(mode)
-	}
-
-}
-
-func server() {
-
-	servFactory, err := tls.TLSListenerFactoryFromConfig(nil, &servConf)
-	if err != nil {
-		panic(err)
-	}
-
-	listener, err := servFactory()
-	if err != nil {
-		panic(err)
-	}
-
-	ctx := context.Background()
-
-	for {
-		conn, err := listener.Accept(ctx)
-		if err != nil {
-			log.Printf("accept error: %s", err)
-			continue
-		}
-		go func() {
-			defer conn.Close()
-
-			log.Printf("handling connection %s", conn.RemoteAddr())
-			_, err = io.Copy(conn, strings.NewReader("here is the server\n"))
-			if err != nil {
-				log.Printf("%s: respond to client error: %s", conn.RemoteAddr(), err)
-				return
-			}
-
-			err = conn.CloseWrite()
-			if err != nil {
-				log.Printf("%s: failed to close write connection: err", conn.RemoteAddr(), err)
-			}
-
-			log.Printf("%s: waiting for client to close connection", conn.RemoteAddr())
-
-			_, err = io.Copy(io.Discard, conn)
-			if err != nil {
-				log.Printf("%s: error draining client connection: %s", conn.RemoteAddr(), err)
-				return
-			}
-
-			log.Printf("%s: done", conn.RemoteAddr())
-			return
-		}()
-	}
-
-}
-
-func client() {
-	connecter, err := tls.TLSConnecterFromConfig(&clientConf)
-	if err != nil {
-		panic(err)
-	}
-
-	ctx := context.Background()
-	conn, err := connecter.Connect(ctx)
-	if err != nil {
-		panic(err)
-	}
-	defer conn.Close()
-
-	_, err = io.Copy(os.Stdout, conn)
-	if err != nil {
-		panic(err)
-	}
-}

transport/tls/serve_tls.go

@@ -4,10 +4,8 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
-	"os"
 	"time"
 
-	"github.com/kr/pretty"
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/config"
@@ -70,7 +68,6 @@ type tlsAuthListener struct {
 func (l tlsAuthListener) Accept(ctx context.Context) (*transport.AuthConn, error) {
 	tcpConn, tlsConn, cn, err := l.ClientAuthListener.Accept()
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "tlsAuthListener.Accept: l.ClientAuthListener.Accept returned error %T %s\n%s\n", err, err, pretty.Sprint(err))
 		return nil, err
 	}
 	if _, ok := l.clientCNs[cn]; !ok {

util/bandwidthlimit/bandwidthlimit.go

@@ -1,79 +0,0 @@
-package bandwidthlimit
-
-import (
-	"errors"
-	"io"
-
-	"github.com/juju/ratelimit"
-)
-
-type Wrapper interface {
-	WrapReadCloser(io.ReadCloser) io.ReadCloser
-}
-
-type Config struct {
-	// Units in this struct are in _bytes_.
-
-	Max            int64 // < 0 means no limit, BucketCapacity is irrelevant then
-	BucketCapacity int64
-}
-
-func NoLimitConfig() Config {
-	return Config{
-		Max:            -1,
-		BucketCapacity: -1,
-	}
-}
-
-func ValidateConfig(conf Config) error {
-	if conf.BucketCapacity == 0 {
-		return errors.New("BucketCapacity must not be zero")
-	}
-	return nil
-}
-
-func WrapperFromConfig(conf Config) Wrapper {
-	if err := ValidateConfig(conf); err != nil {
-		panic(err)
-	}
-
-	if conf.Max < 0 {
-		return noLimit{}
-	}
-
-	return &withLimit{
-		bucket: ratelimit.NewBucketWithRate(float64(conf.Max), conf.BucketCapacity),
-	}
-}
-
-type noLimit struct{}
-
-func (_ noLimit) WrapReadCloser(rc io.ReadCloser) io.ReadCloser { return rc }
-
-type withLimit struct {
-	bucket *ratelimit.Bucket
-}
-
-func (l *withLimit) WrapReadCloser(rc io.ReadCloser) io.ReadCloser {
-	return WrapReadCloser(rc, l.bucket)
-}
-
-type withLimitReadCloser struct {
-	orig    io.Closer
-	limited io.Reader
-}
-
-func (r *withLimitReadCloser) Read(buf []byte) (int, error) {
-	return r.limited.Read(buf)
-}
-
-func (r *withLimitReadCloser) Close() error {
-	return r.orig.Close()
-}
-
-func WrapReadCloser(rc io.ReadCloser, bucket *ratelimit.Bucket) io.ReadCloser {
-	return &withLimitReadCloser{
-		limited: ratelimit.Reader(rc, bucket),
-		orig:    rc,
-	}
-}

util/bandwidthlimit/bandwidthlimit_test.go

@@ -1,19 +0,0 @@
-package bandwidthlimit
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/require"
-)
-
-func TestNoLimitConfig(t *testing.T) {
-
-	conf := NoLimitConfig()
-
-	err := ValidateConfig(conf)
-	require.NoError(t, err)
-
-	require.NotPanics(t, func() {
-		_ = WrapperFromConfig(conf)
-	})
-}

util/datasizeunit/datasizeunit.go

@@ -1,86 +0,0 @@
-package datasizeunit
-
-import (
-	"errors"
-	"fmt"
-	"math"
-	"regexp"
-	"strconv"
-	"strings"
-)
-
-type Bits struct {
-	bits float64
-}
-
-func (b Bits) ToBits() float64    { return b.bits }
-func (b Bits) ToBytes() float64   { return b.bits / 8 }
-func FromBytesInt64(i int64) Bits { return Bits{float64(i) * 8} }
-
-var datarateRegex = regexp.MustCompile(`^([-0-9\.]*)\s*(bit|(|K|Ki|M|Mi|G|Gi|T|Ti)([bB]))$`)
-
-func (r *Bits) UnmarshalYAML(u func(interface{}, bool) error) (_ error) {
-
-	var s string
-	err := u(&s, false)
-	if err != nil {
-		return err
-	}
-
-	genericErr := func(err error) error {
-		var buf strings.Builder
-		fmt.Fprintf(&buf, "cannot parse %q using regex %s", s, datarateRegex)
-		if err != nil {
-			fmt.Fprintf(&buf, ": %s", err)
-		}
-		return errors.New(buf.String())
-	}
-
-	match := datarateRegex.FindStringSubmatch(s)
-	if match == nil {
-		return genericErr(nil)
-	}
-
-	bps, err := strconv.ParseFloat(match[1], 64)
-	if err != nil {
-		return genericErr(err)
-	}
-
-	if match[2] == "bit" {
-		if math.Round(bps) != bps {
-			return genericErr(fmt.Errorf("unit bit must be an integer value"))
-		}
-		r.bits = bps
-		return nil
-	}
-
-	factorMap := map[string]uint64{
-		"": 1,
-
-		"K": 1e3,
-		"M": 1e6,
-		"G": 1e9,
-		"T": 1e12,
-
-		"Ki": 1 << 10,
-		"Mi": 1 << 20,
-		"Gi": 1 << 30,
-		"Ti": 1 << 40,
-	}
-	factor, ok := factorMap[match[3]]
-	if !ok {
-		panic(match)
-	}
-
-	baseUnitFactorMap := map[string]uint64{
-		"b": 1,
-		"B": 8,
-	}
-	baseUnitFactor, ok := baseUnitFactorMap[match[4]]
-	if !ok {
-		panic(match)
-	}
-
-	r.bits = bps * float64(factor) * float64(baseUnitFactor)
-	return nil
-}

util/datasizeunit/datasizeunit_test.go

@@ -1,57 +0,0 @@
-package datasizeunit
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/yaml-config"
-)
-
-func TestBits(t *testing.T) {
-
-	tcs := []struct {
-		input      string
-		expectRate float64
-		expectErr  string
-	}{
-		{`23 bit`, 23, ""}, // bit special case works
-		{`23bit`, 23, ""},  // also without space
-
-		{`10MiB`, 10 * (1 << 20) * 8, ""},  // integer unit without space
-		{`10 MiB`, 8 * 10 * (1 << 20), ""}, // integer unit with space
-
-		{`10.5 Kib`, 10.5 * (1 << 10), ""}, // floating point with bit unit works with space
-		{`10.5Kib`, 10.5 * (1 << 10), ""},  // floating point with bit unit works without space
-
-		// unit checks
-		{`1 bit`, 1, ""},
-		{`1 B`, 1 * 8, ""},
-		{`1 Kb`, 1e3, ""},
-		{`1 Kib`, 1 << 10, ""},
-		{`1 Mb`, 1e6, ""},
-		{`1 Mib`, 1 << 20, ""},
-		{`1 Gb`, 1e9, ""},
-		{`1 Gib`, 1 << 30, ""},
-		{`1 Tb`, 1e12, ""},
-		{`1 Tib`, 1 << 40, ""},
-	}
-
-	for _, tc := range tcs {
-		t.Run(tc.input, func(t *testing.T) {
-
-			var bits Bits
-			err := yaml.Unmarshal([]byte(tc.input), &bits)
-			if tc.expectErr != "" {
-				assert.Error(t, err)
-				assert.Regexp(t, tc.expectErr, err.Error())
-				assert.Zero(t, bits.bits)
-			} else {
-				require.NoError(t, err)
-				assert.Equal(t, tc.expectRate, bits.bits)
-			}
-		})
-
-	}
-
-}

zfs/zfs.go

@@ -14,7 +14,6 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"time"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -334,60 +333,122 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	return stdoutReader, stdoutWriter, nil
 }
 
-type SendStream struct {
-	cmd  *zfscmd.Cmd
-	kill context.CancelFunc
+type sendStreamState int
 
-	closeMtx     sync.Mutex
-	stdoutReader *os.File
+const (
+	sendStreamOpen sendStreamState = iota
+	sendStreamClosed
+)
+
+type SendStream struct {
+	cmd          *zfscmd.Cmd
+	kill         context.CancelFunc
+	stdoutReader io.ReadCloser // not *os.File for mocking during platformtest
 	stderrBuf    *circlog.CircularLog
-	opErr        error
+
+	mtx     sync.Mutex
+	state   sendStreamState
+	exitErr *ZFSError
 }
 
-func (s *SendStream) Read(p []byte) (n int, err error) {
-	s.closeMtx.Lock()
-	opErr := s.opErr
-	s.closeMtx.Unlock()
-	if opErr != nil {
-		return 0, opErr
-	}
+func (s *SendStream) Read(p []byte) (n int, _ error) {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
 
-	n, err = s.stdoutReader.Read(p)
-	if err != nil {
-		debug("sendStream: read err: %T %s", err, err)
-		// TODO we assume here that any read error is permanent
-		// which is most likely the case for a local zfs send
-		kwerr := s.killAndWait(err)
-		debug("sendStream: killAndWait n=%v err= %T %s", n, kwerr, kwerr)
-		// TODO we assume here that any read error is permanent
-		return n, kwerr
+	switch s.state {
+	case sendStreamClosed:
+		return 0, os.ErrClosed
+
+	case sendStreamOpen:
+		n, readErr := s.stdoutReader.Read(p)
+		if readErr != nil {
+			debug("sendStream: read: readErr=%T %s", readErr, readErr)
+			if readErr == io.EOF {
+				// io.EOF must be bubbled up as is so that consumers can handle it properly.
+				return n, readErr
+			}
+			// Assume that the error is not retryable.
+			// Try to kill now so that we can return a nice *ZFSError with captured stderr.
+			// If the kill doesn't work, it doesn't matter because the caller must by contract call Close() anyways.
+			killErr := s.killAndWait()
+			debug("sendStream: read: killErr=%T %s", killErr, killErr)
+			if killErr == nil {
+				s.state = sendStreamClosed
+				return n, s.exitErr // return the nice error
+			} else {
+				// we remain open so that we retry
+				return n, readErr // return the normal error
+			}
+		}
+		return n, readErr
+
+	default:
+		panic("unreachable")
 	}
-	return n, err
 }
 
 func (s *SendStream) Close() error {
 	debug("sendStream: close called")
-	return s.killAndWait(nil)
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	switch s.state {
+	case sendStreamOpen:
+		err := s.killAndWait()
+		if err != nil {
+			return err
+		} else {
+			s.state = sendStreamClosed
+			return nil
+		}
+	case sendStreamClosed:
+		return os.ErrClosed
+	default:
+		panic("unreachable")
+	}
 }
 
-func (s *SendStream) killAndWait(precedingReadErr error) error {
+// returns nil iff the child process is gone (has been successfully waited upon)
+// in that case, s.exitErr is set
+func (s *SendStream) killAndWait() error {
 
 	debug("sendStream: killAndWait enter")
 	defer debug("sendStream: killAndWait leave")
-	if precedingReadErr == io.EOF {
-		// give the zfs process a little bit of time to terminate itself
-		// if it holds this deadline, exitErr will be nil
-		time.AfterFunc(200*time.Millisecond, s.kill)
-	} else {
-		s.kill()
-	}
 
-	// allow async kills from Close(), that's why we only take the mutex here
-	s.closeMtx.Lock()
-	defer s.closeMtx.Unlock()
+	// send SIGKILL
+	s.kill()
 
-	if s.opErr != nil {
-		return s.opErr
+	// Close our read-end of the pipe.
+	//
+	// We must do this before .Wait() because in some (not all) versions/build configs of ZFS,
+	// `zfs send` uses a separate kernel thread (taskq) to write the send stream (function `dump_bytes`).
+	// The `zfs send` thread then waits uinterruptably for the taskq thread to finish the write.
+	// And signalling the `zfs send` thread doesn't propagate to the taskq thread.
+	// So we end up in a state where we .Wait() forever.
+	// (See https://github.com/openzfs/zfs/issues/12500 and
+	//  https://github.com/zrepl/zrepl/issues/495#issuecomment-902530043)
+	//
+	// By closing our read end of the pipe before .Wait(), we unblock the taskq thread if there is any.
+	// If there is no separate taskq thread, the SIGKILL to `zfs end` would suffice and be most precise,
+	// but due to the circumstances above, there is no other portable & robust way.
+	//
+	// However, the fallout from closing the pipe is that (in non-taskq builds) `zfs sends` will get a SIGPIPE.
+	// And on Linux, that SIGPIPE appears to win over the previously issued SIGKILL.
+	// And thus, on Linux, the `zfs send` will be killed by the default SIGPIPE handler.
+	// We can observe this in the WaitStatus below.
+	// This behavior is slightly annoying because the *exec.ExitError's message ("signal: broken pipe")
+	// isn't as clear as ("signal: killed").
+	// However, it seems like we just have to live with that. (covered by platformtest)
+	var closePipeErr error
+	if s.stdoutReader != nil {
+		closePipeErr = s.stdoutReader.Close()
+		if closePipeErr == nil {
+			// avoid double-closes in case waiting below doesn't work
+			// and someone attempts Close again
+			s.stdoutReader = nil
+		} else {
+			return closePipeErr
+		}
 	}
 
 	waitErr := s.cmd.Wait()
@@ -402,39 +463,30 @@ func (s *SendStream) killAndWait(precedingReadErr error) error {
 		}
 	}
 
-	// now, after we know the program exited do we close the pipe
-	var closePipeErr error
-	if s.stdoutReader != nil {
-		closePipeErr = s.stdoutReader.Close()
-		if closePipeErr == nil {
-			// avoid double-closes in case anything below doesn't work
-			// and someone calls Close again
-			s.stdoutReader = nil
-		} else {
-			return closePipeErr
-		}
-	}
+	// invariant: at this point, the child is gone and we cleaned up everything related to the SendStream
 
-	// we managed to tear things down, no let's give the user some pretty *ZFSError
 	if exitErr != nil {
-		s.opErr = &ZFSError{
+		// zfs send exited with an error or was killed by a signal.
+		s.exitErr = &ZFSError{
 			Stderr:  []byte(s.stderrBuf.String()),
 			WaitErr: exitErr,
 		}
 	} else {
-		s.opErr = precedingReadErr
+		// zfs send exited successfully (we know that since waitErr was either nil or wasn't an *exec.ExitError)
+		s.exitErr = nil
 	}
 
-	// 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
-	if exitErr == nil && precedingReadErr == io.EOF {
-		return precedingReadErr
-	}
-
-	return s.opErr
+	return nil
 }
 
+func (s *SendStream) TestOnly_ReplaceStdoutReader(f io.ReadCloser) (prev io.ReadCloser) {
+	prev = s.stdoutReader
+	s.stdoutReader = f
+	return prev
+}
+
+func (s *SendStream) TestOnly_ExitErr() *ZFSError { return s.exitErr }
+
 // NOTE: When updating this struct, make sure to update funcs Validate ValidateCorrespondsToResumeToken
 type ZFSSendArgVersion struct {
 	RelName string

zfs/zfscmd/zfscmd.go

@@ -209,3 +209,7 @@ func (c *Cmd) Runtime() time.Duration {
 	}
 	return c.waitReturnedAt.Sub(c.startedAt)
 }
+
+func (c *Cmd) TestOnly_ExecCmd() *exec.Cmd {
+	return c.cmd
+}