From ee2336a24b4049f48296aad1156f2e1d514bd915 Mon Sep 17 00:00:00 2001
From: Lukas Schauer <lukas@schauer.dev>
Date: Sun, 21 Mar 2021 21:46:29 +0100
Subject: [PATCH] zfs: pipe size: default to value of /proc/sys/fs/pipe-max-siz

Addition by @problame: move getPipeCapacityHint() into platform-specific
code. This has the added benefit of not recognizing the envvar as an
envconst on platform that do not support resizing pipes.  => won't show
up in (zrepl status --raw).Global.Envconst

fixes #424
cloes #449
---
 zfs/zfs.go            | 13 ++++---------
 zfs/zfs_pipe.go       |  6 +++++-
 zfs/zfs_pipe_linux.go | 22 ++++++++++++++++++++++
 3 files changed, 31 insertions(+), 10 deletions(-)

diff --git a/zfs/zfs.go b/zfs/zfs.go
index 574f67a..fca80b4 100644
--- a/zfs/zfs.go
+++ b/zfs/zfs.go
@@ -26,11 +26,6 @@ import (
 	"github.com/zrepl/zrepl/zfs/zfscmd"
 )
 
-var (
-	ZFSSendPipeCapacityHint = int(envconst.Int64("ZFS_SEND_PIPE_CAPACITY_HINT", 1<<25))
-	ZFSRecvPipeCapacityHint = int(envconst.Int64("ZFS_RECV_PIPE_CAPACITY_HINT", 1<<25))
-)
-
 type DatasetPath struct {
 	comps []string
 }
@@ -328,8 +323,8 @@ func absVersion(fs string, v *ZFSSendArgVersion) (full string, err error) {
 }
 
 func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
-	if capacity <= 0 {
-		panic(fmt.Sprintf("capacity must be positive %v", capacity))
+	if capacity < 0 {
+		panic(fmt.Sprintf("capacity must be non-negative, got %v", capacity))
 	}
 	stdoutReader, stdoutWriter, err := os.Pipe()
 	if err != nil {
@@ -858,7 +853,7 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*SendStream, e
 	ctx, cancel := context.WithCancel(ctx)
 
 	// setup stdout with an os.Pipe to control pipe buffer size
-	stdoutReader, stdoutWriter, err := pipeWithCapacityHint(ZFSSendPipeCapacityHint)
+	stdoutReader, stdoutWriter, err := pipeWithCapacityHint(getPipeCapacityHint("ZFS_SEND_PIPE_CAPACITY_HINT"))
 	if err != nil {
 		cancel()
 		return nil, err
@@ -1160,7 +1155,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 
 	stderr := bytes.NewBuffer(make([]byte, 0, RecvStderrBufSiz))
 
-	stdin, stdinWriter, err := pipeWithCapacityHint(ZFSRecvPipeCapacityHint)
+	stdin, stdinWriter, err := pipeWithCapacityHint(getPipeCapacityHint("ZFS_RECV_PIPE_CAPACITY_HINT"))
 	if err != nil {
 		return err
 	}
diff --git a/zfs/zfs_pipe.go b/zfs/zfs_pipe.go
index 4816441..5828d32 100644
--- a/zfs/zfs_pipe.go
+++ b/zfs/zfs_pipe.go
@@ -7,10 +7,14 @@ import (
 	"sync"
 )
 
+func getPipeCapacityHint(envvar string) int {
+	return 0 // not supported
+}
+
 var zfsPipeCapacityNotSupported sync.Once
 
 func trySetPipeCapacity(p *os.File, capacity int) {
-	if debugEnabled {
+	if debugEnabled && capacity != 0 {
 		zfsPipeCapacityNotSupported.Do(func() {
 			debug("trySetPipeCapacity error: OS does not support setting pipe capacity")
 		})
diff --git a/zfs/zfs_pipe_linux.go b/zfs/zfs_pipe_linux.go
index ab50020..104997c 100644
--- a/zfs/zfs_pipe_linux.go
+++ b/zfs/zfs_pipe_linux.go
@@ -3,11 +3,33 @@ package zfs
 import (
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"os"
+	"strconv"
+	"strings"
 
 	"golang.org/x/sys/unix"
+
+	"github.com/zrepl/zrepl/util/envconst"
 )
 
+func getPipeCapacityHint(envvar string) int {
+	var capacity int64 = 1 << 25
+
+	// Work around a race condition in Linux >= 5.8 related to pipe resizing.
+	// https://github.com/zrepl/zrepl/issues/424#issuecomment-800370928
+	// https://bugzilla.kernel.org/show_bug.cgi?id=212295
+	if _, err := os.Stat("/proc/sys/fs/pipe-max-size"); err == nil {
+		if dat, err := ioutil.ReadFile("/proc/sys/fs/pipe-max-size"); err == nil {
+			if capacity, err = strconv.ParseInt(strings.TrimSpace(string(dat)), 10, 64); err != nil {
+				capacity = 1 << 25
+			}
+		}
+	}
+
+	return int(envconst.Int64(envvar, capacity))
+}
+
 func trySetPipeCapacity(p *os.File, capacity int) {
 	res, err := unix.FcntlInt(p.Fd(), unix.F_SETPIPE_SZ, capacity)
 	if err != nil {