pruning/keep_last_n: correctly handle the case where count > matching snaps

fixes #446

client/signal.go

@@ -11,8 +11,8 @@ import (
 )
 
 var SignalCmd = &cli.Subcommand{
-	Use:   "signal [replication|reset|snapshot] JOB",
+	Use:   "signal [wakeup|reset] JOB",
-	Short: "run a job replication, abort its current invocation, run a snapshot job",
+	Short: "wake up a job from wait state or abort its current invocation",
 	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
 		return runSignalCmd(subcommand.Config(), args)
 	},
@@ -20,7 +20,7 @@ var SignalCmd = &cli.Subcommand{
 
 func runSignalCmd(config *config.Config, args []string) error {
 	if len(args) != 2 {
-		return errors.Errorf("Expected 2 arguments: [replication|reset|snapshot] JOB")
+		return errors.Errorf("Expected 2 arguments: [wakeup|reset] JOB")
 	}
 
 	httpc, err := controlHttpClient(config.Global.Control.SockPath)

client/status/client/client.go

@@ -55,12 +55,8 @@ func (c *Client) signal(job, sig string) error {
 	)
 }
 
-func (c *Client) SignalReplication(job string) error {
+func (c *Client) SignalWakeup(job string) error {
-	return c.signal(job, "replication")
+	return c.signal(job, "wakeup")
-}
-
-func (c *Client) SignalSnapshot(job string) error {
-	return c.signal(job, "snapshot")
 }
 
 func (c *Client) SignalReset(job string) error {

client/status/status.go

@@ -19,8 +19,7 @@ import (
 type Client interface {
 	Status() (daemon.Status, error)
 	StatusRaw() ([]byte, error)
-	SignalReplication(job string) error
+	SignalWakeup(job string) error
-	SignalSnapshot(job string) error
 	SignalReset(job string) error
 }
 
@@ -71,12 +70,16 @@ func runStatusV2Command(ctx context.Context, config *config.Config, args []strin
 
 	mode := statusv2Flags.Mode.Value().(statusv2Mode)
 
-	if !isatty.IsTerminal(os.Stdout.Fd()) && mode != StatusV2ModeDump {
+	if !isatty.IsTerminal(os.Stdout.Fd()) && mode != StatusV2ModeDump && mode != StatusV2ModeRaw {
-		usemode, err := statusv2Flags.Mode.InputForChoice(StatusV2ModeDump)
+		dumpmode, err := statusv2Flags.Mode.InputForChoice(StatusV2ModeDump)
 		if err != nil {
 			panic(err)
 		}
-		return errors.Errorf("error: stdout is not a tty, please use --mode %s", usemode)
+		rawmode, err := statusv2Flags.Mode.InputForChoice(StatusV2ModeRaw)
+		if err != nil {
+			panic(err)
+		}
+		return errors.Errorf("error: stdout is not a tty, please use --mode %s or --mode %s", dumpmode, rawmode)
 	}
 
 	switch mode {

client/status/status_interactive.go

@@ -281,8 +281,8 @@ func interactive(c Client, flag statusFlags) error {
 			if !ok {
 				return nil
 			}
-			signals := []string{"replication", "snapshot", "reset"}
+			signals := []string{"wakeup", "reset"}
-			clientFuncs := []func(job string) error{c.SignalReplication, c.SignalSnapshot, c.SignalReset}
+			clientFuncs := []func(job string) error{c.SignalWakeup, c.SignalReset}
 			sigMod := tview.NewModal()
 			sigMod.SetBackgroundColor(tcell.ColorDefault)
 			sigMod.SetBorder(true)

config/samples/quickstart_backup_to_external_disk.yml

@@ -40,7 +40,7 @@ jobs:
 
 # This job pushes to the local sink defined in job `backuppool_sink`.
 # We trigger replication manually from the command line / udev rules using
-#  `zrepl signal replication push_to_drive`
+#  `zrepl signal wakeup push_to_drive`
 - type: push
   name: push_to_drive
   connect:

daemon/control.go

@@ -143,12 +143,10 @@ func (j *controlJob) Run(ctx context.Context) {
 
 			var err error
 			switch req.Op {
-			case "replication":
+			case "wakeup":
-				err = j.jobs.doreplication(req.Name)
+				err = j.jobs.wakeup(req.Name)
 			case "reset":
 				err = j.jobs.reset(req.Name)
-			case "snapshot":
-				err = j.jobs.dosnapshot(req.Name)
 			default:
 				err = fmt.Errorf("operation %q is invalid", req.Op)
 			}

daemon/daemon.go

@@ -20,9 +20,8 @@ import (
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
-	"github.com/zrepl/zrepl/daemon/job/doreplication"
-	"github.com/zrepl/zrepl/daemon/job/dosnapshot"
 	"github.com/zrepl/zrepl/daemon/job/reset"
+	"github.com/zrepl/zrepl/daemon/job/wakeup"
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/version"
@@ -132,19 +131,17 @@ type jobs struct {
 	wg sync.WaitGroup
 
 	// m protects all fields below it
-	m              sync.RWMutex
+	m       sync.RWMutex
-	doreplications map[string]doreplication.Func // by Job.Name
+	wakeups map[string]wakeup.Func // by Job.Name
-	resets         map[string]reset.Func         // by Job.Name
+	resets  map[string]reset.Func  // by Job.Name
-	dosnapshots    map[string]dosnapshot.Func    // by Job.Name
+	jobs    map[string]job.Job
-	jobs           map[string]job.Job
 }
 
 func newJobs() *jobs {
 	return &jobs{
-		doreplications: make(map[string]doreplication.Func),
+		wakeups: make(map[string]wakeup.Func),
-		resets:         make(map[string]reset.Func),
+		resets:  make(map[string]reset.Func),
-		dosnapshots:    make(map[string]dosnapshot.Func),
+		jobs:    make(map[string]job.Job),
-		jobs:           make(map[string]job.Job),
 	}
 }
 
@@ -193,11 +190,11 @@ func (s *jobs) status() map[string]*job.Status {
 	return ret
 }
 
-func (s *jobs) doreplication(job string) error {
+func (s *jobs) wakeup(job string) error {
 	s.m.RLock()
 	defer s.m.RUnlock()
 
-	wu, ok := s.doreplications[job]
+	wu, ok := s.wakeups[job]
 	if !ok {
 		return errors.Errorf("Job %s does not exist", job)
 	}
@@ -215,17 +212,6 @@ func (s *jobs) reset(job string) error {
 	return wu()
 }
 
-func (s *jobs) dosnapshot(job string) error {
-	s.m.RLock()
-	defer s.m.RUnlock()
-
-	wu, ok := s.dosnapshots[job]
-	if !ok {
-		return errors.Errorf("Job %s does not exist", job)
-	}
-	return wu()
-}
-
 const (
 	jobNamePrometheus = "_prometheus"
 	jobNameControl    = "_control"
@@ -256,12 +242,10 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) {
 
 	s.jobs[jobName] = j
 	ctx = zfscmd.WithJobID(ctx, j.Name())
-	ctx, doreplication := doreplication.Context(ctx)
+	ctx, wakeup := wakeup.Context(ctx)
 	ctx, resetFunc := reset.Context(ctx)
-	ctx, dosnapshotFunc := dosnapshot.Context(ctx)
+	s.wakeups[jobName] = wakeup
-	s.doreplications[jobName] = doreplication
 	s.resets[jobName] = resetFunc
-	s.dosnapshots[jobName] = dosnapshotFunc
 
 	s.wg.Add(1)
 	go func() {

daemon/job/active.go

@@ -14,8 +14,8 @@ import (
 	"github.com/zrepl/zrepl/util/envconst"
 
 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/job/doreplication"
 	"github.com/zrepl/zrepl/daemon/job/reset"
+	"github.com/zrepl/zrepl/daemon/job/wakeup"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
@@ -89,7 +89,7 @@ type activeMode interface {
 	SenderReceiver() (logic.Sender, logic.Receiver)
 	Type() Type
 	PlannerPolicy() logic.PlannerPolicy
-	RunPeriodic(ctx context.Context, replicationCommon chan<- struct{})
+	RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{})
 	SnapperReport() *snapper.Report
 	ResetConnectBackoff()
 }
@@ -131,8 +131,8 @@ func (m *modePush) Type() Type { return TypePush }
 
 func (m *modePush) PlannerPolicy() logic.PlannerPolicy { return *m.plannerPolicy }
 
-func (m *modePush) RunPeriodic(ctx context.Context, replicationCommon chan<- struct{}) {
+func (m *modePush) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) {
-	m.snapper.Run(ctx, replicationCommon)
+	m.snapper.Run(ctx, wakeUpCommon)
 }
 
 func (m *modePush) SnapperReport() *snapper.Report {
@@ -214,10 +214,10 @@ func (*modePull) Type() Type { return TypePull }
 
 func (m *modePull) PlannerPolicy() logic.PlannerPolicy { return *m.plannerPolicy }
 
-func (m *modePull) RunPeriodic(ctx context.Context, replicationCommon chan<- struct{}) {
+func (m *modePull) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) {
 	if m.interval.Manual {
 		GetLogger(ctx).Info("manual pull configured, periodic pull disabled")
-		// "waiting for wakeup replications" is printed in common ActiveSide.do
+		// "waiting for wakeups" is printed in common ActiveSide.do
 		return
 	}
 	t := time.NewTicker(m.interval.Interval)
@@ -226,12 +226,12 @@ func (m *modePull) RunPeriodic(ctx context.Context, replicationCommon chan<- str
 		select {
 		case <-t.C:
 			select {
-			case replicationCommon <- struct{}{}:
+			case wakeUpCommon <- struct{}{}:
 			default:
 				GetLogger(ctx).
 					WithField("pull_interval", m.interval).
 					Warn("pull job took longer than pull interval")
-				replicationCommon <- struct{}{} // block anyways, to queue up the wakeup replication
+				wakeUpCommon <- struct{}{} // block anyways, to queue up the wakeup
 			}
 		case <-ctx.Done():
 			return
@@ -435,13 +435,13 @@ func (j *ActiveSide) Run(ctx context.Context) {
 	invocationCount := 0
 outer:
 	for {
-		log.Info("wait for replications")
+		log.Info("wait for wakeups")
 		select {
 		case <-ctx.Done():
 			log.WithError(ctx.Err()).Info("context")
 			break outer
 
-		case <-doreplication.Wait(ctx):
+		case <-wakeup.Wait(ctx):
 			j.mode.ResetConnectBackoff()
 		case <-periodicDone:
 		}

daemon/job/doreplication/doreplication.go

@@ -1,35 +0,0 @@
-package doreplication
-
-import (
-	"context"
-	"errors"
-)
-
-type contextKey int
-
-const contextKeyReplication contextKey = iota
-
-func Wait(ctx context.Context) <-chan struct{} {
-	wc, ok := ctx.Value(contextKeyReplication).(chan struct{})
-	if !ok {
-		wc = make(chan struct{})
-	}
-	return wc
-}
-
-type Func func() error
-
-var AlreadyReplicating = errors.New("already replicating")
-
-func Context(ctx context.Context) (context.Context, Func) {
-	wc := make(chan struct{})
-	wuf := func() error {
-		select {
-		case wc <- struct{}{}:
-			return nil
-		default:
-			return AlreadyReplicating
-		}
-	}
-	return context.WithValue(ctx, contextKeyReplication, wc), wuf
-}

daemon/job/snapjob.go

@@ -9,12 +9,12 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 
-	"github.com/zrepl/zrepl/daemon/job/doreplication"
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"github.com/zrepl/zrepl/util/nodefault"
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/daemon/job/wakeup"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
@@ -112,13 +112,13 @@ func (j *SnapJob) Run(ctx context.Context) {
 	invocationCount := 0
 outer:
 	for {
-		log.Info("wait for replications")
+		log.Info("wait for wakeups")
 		select {
 		case <-ctx.Done():
 			log.WithError(ctx.Err()).Info("context")
 			break outer
 
-		case <-doreplication.Wait(ctx):
+		case <-wakeup.Wait(ctx):
 		case <-periodicDone:
 		}
 		invocationCount++

daemon/job/wakeup/wakeup.go

@@ -1,4 +1,4 @@
-package dosnapshot
+package wakeup
 
 import (
 	"context"
@@ -7,10 +7,10 @@ import (
 
 type contextKey int
 
-const contextKeyDosnapshot contextKey = iota
+const contextKeyWakeup contextKey = iota
 
 func Wait(ctx context.Context) <-chan struct{} {
-	wc, ok := ctx.Value(contextKeyDosnapshot).(chan struct{})
+	wc, ok := ctx.Value(contextKeyWakeup).(chan struct{})
 	if !ok {
 		wc = make(chan struct{})
 	}
@@ -19,7 +19,7 @@ func Wait(ctx context.Context) <-chan struct{} {
 
 type Func func() error
 
-var AlreadyDosnapshot = errors.New("already snapshotting")
+var AlreadyWokenUp = errors.New("already woken up")
 
 func Context(ctx context.Context) (context.Context, Func) {
 	wc := make(chan struct{})
@@ -28,8 +28,8 @@ func Context(ctx context.Context) (context.Context, Func) {
 		case wc <- struct{}{}:
 			return nil
 		default:
-			return AlreadyDosnapshot
+			return AlreadyWokenUp
 		}
 	}
-	return context.WithValue(ctx, contextKeyDosnapshot, wc), wuf
+	return context.WithValue(ctx, contextKeyWakeup, wc), wuf
 }

daemon/snapper/snapper.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/zrepl/zrepl/daemon/job/dosnapshot"
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/config"
@@ -211,10 +210,6 @@ func syncUp(a args, u updater) state {
 		return u(func(s *Snapper) {
 			s.state = Planning
 		}).sf()
-	case <-dosnapshot.Wait(a.ctx):
-		return u(func(s *Snapper) {
-			s.state = Planning
-		}).sf()
 	case <-a.ctx.Done():
 		return onMainCtxDone(a.ctx, u)
 	}
@@ -383,10 +378,6 @@ func wait(a args, u updater) state {
 		return u(func(snapper *Snapper) {
 			snapper.state = Planning
 		}).sf()
-	case <-dosnapshot.Wait(a.ctx):
-		return u(func(snapper *Snapper) {
-			snapper.state = Planning
-		}).sf()
 	case <-a.ctx.Done():
 		return onMainCtxDone(a.ctx, u)
 	}

daemon/snapper/snapper_all.go

@@ -18,9 +18,9 @@ type PeriodicOrManual struct {
 	s *Snapper
 }
 
-func (s *PeriodicOrManual) Run(ctx context.Context, replicationCommon chan<- struct{}) {
+func (s *PeriodicOrManual) Run(ctx context.Context, wakeUpCommon chan<- struct{}) {
 	if s.s != nil {
-		s.s.Run(ctx, replicationCommon)
+		s.s.Run(ctx, wakeUpCommon)
 	}
 }
 

docs/changelog.rst

@@ -32,7 +32,6 @@ We use the following annotations for classifying changes:
 0.4.0
 -----
 
-* |break| Change syntax to trigger a job replication, rename ``zrepl signal wakeup JOB`` to ``zrepl signal replication JOB``
 * |feature| support setting zfs send / recv flags in the config (send: ``-wLcepbS`` , recv: ``-ox`` ).
   Config docs :ref:`here <job-send-options>` and :ref:`here <job-recv-options>` .
 * |feature| parallel replication is now configurable (disabled by default, :ref:`config docs here <replication-option-concurrency>` ).

docs/configuration/jobs.rst

@@ -78,7 +78,7 @@ Job Type ``pull``
         ``$root_fs/$source_path``
     * - ``interval``
       - | Interval at which to pull from the source job (e.g. ``10m``).
-        | ``manual`` disables periodic pulling, replication then only happens on :ref:`replication <cli-signal-replication>`.
+        | ``manual`` disables periodic pulling, replication then only happens on :ref:`wakeup <cli-signal-wakeup>`.
     * - ``pruning``
       - |pruning-spec|
 

docs/configuration/snapshotting.rst

@@ -15,7 +15,7 @@ A filesystem that does not have snapshots by the snapshotter has lower priority
 
 For ``push`` jobs, replication is automatically triggered after all filesystems have been snapshotted.
 
-Note that the ``zrepl signal replication JOB`` subcommand does not trigger snapshotting.
+Note that the ``zrepl signal wakeup JOB`` subcommand does not trigger snapshotting.
 
 
 ::
@@ -38,7 +38,7 @@ There is also a ``manual`` snapshotting type, which covers the following use cas
 * Existing infrastructure for automatic snapshots: you only want to use this zrepl job for replication.
 * Handling snapshotting through a separate ``snap`` job.
 
-Note that you will have to trigger replication manually using the ``zrepl signal replication JOB`` subcommand in that case.
+Note that you will have to trigger replication manually using the ``zrepl signal wakeup JOB`` subcommand in that case.
 
 ::
 

docs/quickstart.rst

@@ -55,8 +55,8 @@ Watch it Work
 =============
 
 Run ``zrepl status`` on the active side of the replication setup to monitor snaphotting, replication and pruning activity.
-To re-trigger replication (snapshots are separate!), use ``zrepl signal replication JOBNAME``.
+To re-trigger replication (snapshots are separate!), use ``zrepl signal wakeup JOBNAME``.
-(refer to the example use case document if you are uncertain which job you want to start replication).
+(refer to the example use case document if you are uncertain which job you want to wake up).
 
 You can also use basic UNIX tools to inspect see what's going on.
 If you like tmux, here is a handy script that works on FreeBSD: ::

docs/usage.rst

@@ -13,7 +13,7 @@ CLI Overview
     The zrepl binary is self-documenting:
     run ``zrepl help`` for an overview of the available subcommands or ``zrepl SUBCOMMAND --help`` for information on available flags, etc.
 
-.. _cli-signal-replication:
+.. _cli-signal-wakeup:
 
 .. list-table::
     :widths: 30 70
@@ -29,7 +29,7 @@ CLI Overview
       - show job activity, or with ``--raw`` for JSON output
     * - ``zrepl stdinserver``
       - see :ref:`transport-ssh+stdinserver`
-    * - ``zrepl signal replication JOB``
+    * - ``zrepl signal wakeup JOB``
       - manually trigger replication + pruning of JOB
     * - ``zrepl signal reset JOB``
       - manually abort current replication + pruning of JOB

pruning/keep_last_n.go

@@ -57,6 +57,11 @@ func (k KeepLastN) KeepRule(snaps []Snapshot) (destroyList []Snapshot) {
 		// then lexicographically descending (e.g. b, a)
 		return strings.Compare(matching[i].Name(), matching[j].Name()) == 1
 	})
-	destroyList = append(destroyList, matching[k.n:]...)
+
+	n := k.n
+	if n > len(matching) {
+		n = len(matching)
+	}
+	destroyList = append(destroyList, matching[n:]...)
 	return destroyList
 }

pruning/keep_last_n_test.go

@@ -83,6 +83,19 @@ func TestKeepLastN(t *testing.T) {
 				"b1": true,
 			},
 		},
+		"keep_more_than_matching": {
+			inputs: []Snapshot{
+				stubSnap{"a1", false, o(10)},
+				stubSnap{"b1", false, o(11)},
+				stubSnap{"a2", false, o(12)},
+			},
+			rules: []KeepRule{
+				MustKeepLastN(3, "a"),
+			},
+			expDestroy: map[string]bool{
+				"b1": true,
+			},
+		},
 	}
 
 	testTable(tcs, t)

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 {
+	if capacity < 0 {
-		panic(fmt.Sprintf("capacity must be positive %v", capacity))
+		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
 	}

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")
 		})

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 {