WIP: draft snapmgmt

client/signal.go

@@ -9,7 +9,7 @@ import (
 )
 
 var SignalCmd = &cli.Subcommand{
-	Use:   "signal [wakeup|reset] JOB [DATA]",
+	Use:   "signal [wakeup|reset] JOB",
 	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,13 +17,8 @@ var SignalCmd = &cli.Subcommand{
 }
 
 func runSignalCmd(config *config.Config, args []string) error {
-	if len(args) < 2 || len(args) > 3 {
+	if len(args) != 2 {
-		return errors.Errorf("Expected 2 arguments: [wakeup|reset|set-concurrency] JOB [DATA]")
+		return errors.Errorf("Expected 2 arguments: [wakeup|reset] JOB")
-	}
-
-	var data string
-	if len(args) == 3 {
-		data = args[2]
 	}
 
 	httpc, err := controlHttpClient(config.Global.Control.SockPath)
@@ -35,11 +30,9 @@ func runSignalCmd(config *config.Config, args []string) error {
 		struct {
 			Name string
 			Op   string
-			Data string
 		}{
 			Name: args[1],
 			Op:   args[0],
-			Data: data,
 		},
 		struct{}{},
 	)

daemon/control.go

@@ -8,7 +8,6 @@ import (
 	"io"
 	"net"
 	"net/http"
-	"strconv"
 	"time"
 
 	"github.com/pkg/errors"
@@ -48,8 +47,6 @@ 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
@@ -129,7 +126,6 @@ func (j *controlJob) Run(ctx context.Context) {
 			type reqT struct {
 				Name string
 				Op   string
-				Data string
 			}
 			var req reqT
 			if decoder(&req) != nil {
@@ -142,14 +138,6 @@ 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)
 			}

daemon/daemon.go

@@ -176,18 +176,6 @@ 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"

daemon/job/active.go

@@ -55,12 +55,9 @@ const (
 type activeSideTasks struct {
 	state ActiveSideState
 
-	concurrency int
-
 	// valid for state ActiveSideReplicating, ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
 	replicationReport driver.ReportFunc
 	replicationCancel context.CancelFunc
-	replicationSetConcurrency driver.SetConcurrencyFunc
 
 	// valid for state ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
 	prunerSender, prunerReceiver *pruner.Pruner
@@ -281,8 +278,6 @@ 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
@@ -380,22 +375,6 @@ 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)
@@ -457,14 +436,11 @@ func (j *ActiveSide) do(ctx context.Context) {
 		ctx, repCancel := context.WithCancel(ctx)
 		var repWait driver.WaitFunc
 		j.updateTasks(func(tasks *activeSideTasks) {
-			// reset it (almost)
+			// reset it
-			old := *tasks
+			*tasks = activeSideTasks{}
-			*tasks = activeSideTasks{
-				concurrency: old.concurrency,
-			}
 			tasks.replicationCancel = repCancel
-			tasks.replicationReport, repWait, tasks.replicationSetConcurrency = replication.Do(
+			tasks.replicationReport, repWait = replication.Do(
-				ctx, tasks.concurrency, logic.NewPlanner(j.promRepStateSecs, j.promBytesReplicated, sender, receiver, j.mode.PlannerPolicy()),
+				ctx, logic.NewPlanner(j.promRepStateSecs, j.promBytesReplicated, sender, receiver, j.mode.PlannerPolicy()),
 			)
 			tasks.state = ActiveSideReplicating
 		})

daemon/job/job.go

@@ -40,7 +40,6 @@ 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

daemon/job/passive.go

@@ -163,8 +163,6 @@ 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)

daemon/job/snapjob.go

@@ -117,8 +117,6 @@ 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.

daemon/prometheus.go

@@ -5,7 +5,6 @@ import (
 	"net"
 	"net/http"
 
-	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 
@@ -52,8 +51,6 @@ 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) {

draft_snapmgmt.md

@@ -0,0 +1,129 @@
+# Tiered Snapshotting, Pruning & Replication
+
+## Use Case
+
+* Differently paced snapshotting & replication
+    * 20 x 10min-spaced snapshots for rollback in case of adminstrative mistakes.
+        * Should never be replicated
+    * Daily Snapshots for off-site replication
+        * at 3:00 AM to avoid impacting production workload
+
+* Sometimes, out-of-routine snapshot for off-site replication
+    * e.g. before system update
+
+
+## Config Draft
+
+```yaml
+- type: push
+  connect:
+    type: tcp
+    address: "backup-server.foo.bar:8888"
+  filesystems: {
+    "<": true,
+    "tmp": false
+  }
+  send:
+    snap_filters:
+    - type: snapmgmt # auto-add this snap filter if snapmgmt != {} 
+      # every snapmgmt method produces a list of snaps_filters which this snap_filter type represents
+    - type: regex
+      regex: "manual_.*" # admin may create snapshots named like `manual_pre_upgrade` and have those replicated as well
+      # TODO: how does a manual `zfs snapshot` trigger replication? `zrepl signal wakeup ...` good enough for now
+  snapmgmt:
+    type: tiered
+    prefix: zrepl_
+    tiers:
+    - name: local
+      cron: every 10 minutes
+      keep: { local: 10, remote: 0 }
+    - name: daily
+      cron: every day at 3 AM
+      keep: { local: 14, remote: 30 }
+    - name: weekly
+      cron: every sunday at 3 AM
+      keep: { local: 0, remote: 20 }
+    - name: monthly
+      cron: every last sunday of the month at 3 AM
+      keep: { local: 0, remote: 12 }
+```
+
+TODO pull - likely breaks existing split of responsibilities about pruning between `pull` and `source`
+
+## Implementation
+
+A `snapmgmt` type implements both snapshotting and *local* pruning in one service (`RunSnapshotterAndLocalPruner()`)
+The `snapmgmt` type `tiered` works as follows:
+
+* Snapshots are created per tier:
+  * snapshot name = `${PREFIX}_${JOBNAME}_TIERED_${thistier.name}_${NOW}`
+  * property `zrepl:JOBNAME:tier=${thistier.name}`
+  * property `zrepl:JOBNAME:destroy_at=${thistier.destroy_at(NOW)}`
+* Sender-side pruning by
+  * filtering all snaps by userprop `zrepl:JOBNAME:tier=${thistier.name}`
+  * destroying all snaps in the filtered list with `(zrepl:JOBNAME:destroy_at).Before(NOW)`
+  * sleeping until MIN of
+    * `MIN(filtered_list.zrepl:JOBNAME:destroy_at)`
+    * `(<-newSnapshots).DestroyAt)`
+
+The replication enigne remains orthogonal to `snapmgmt`, but the protocol allows a sender to present a filtered view of snapshots in the `ListFilesystemVersions` RPC response.
+The filters are specified in a list, and have OR semantics, i.e., if one filter matches, the snapshot is presented tothe replication engine.
+The filter type supporting the feature proposed in this issue is the `tiered` filter type:
+it includes snapshots that are
+* named by the snapshotter above: `${PREFIX}_${JOBNAME}_${snapmgmt.tiers|.name}_${NOW}`
+* and the respective tier has a non-zero `keep.remote`
+* TODO: deriving the tier from the snapshot name is unclean.
+  We could instead allow that a replication filter is not pure (i.e. may use zfs.ZFSGet).
+  Would work for both `push` and `source` jobs.
+
+Receiver-side pruning is still unsolved (TODO):
+
+* We could replicate the `destroy_at` property, and have independent pruning on the receive-side.
+    * While there are cases where this is desirable (use case "guarantee that replicated data is destroyed after X months")
+    * it might be undesirable in cases where the receive-side is a backup (which should not self-destroy)
+* We could require each `snapmgmt` method to have a `PruneReceiver(receiver)` method, and require that `receiver` provides an RPC endpoint to get the `destroy_at` infromation required to evaluate the prune policy (e.g. a flag in the request, like `GetDestroyAt=true`)
+
+## Other future `snapmgmt` methods
+
+```yaml
+snapmgmt:
+  # zrepl signal ondemand JOBNAME SNAPNAME EXPIRE_AT
+  # => creates snapshot @SNAPNAME for all filesystems matched by `$JOB.filesystems` with expiration date `EXPIRE_AT`
+  # => snapshot has user prop `zrepl:JOBNAME:ondemand=on`
+  # replication filter `snapmgmt` filters by `ondemand` property
+  # pruning filter uses expieration data encoded in property
+  type: ondemand
+
+snapmgmt:
+  # no snapshots, never-matching filter
+  type: manual
+```
+
+## Existing Use-Cases
+
+### Existing Configurations & Migration to New Config
+
+`snapmgmt` replaces the previous `snapshotting` and `pruning` fields.
+Proposed migration for existing configs:
+
+```yaml
+snapmgmt:
+  type: pre-snapmgmt
+  snapshotting:
+    type: periodic
+    interval: 10m
+    prefix: zrepl_
+  pruning:
+    keep_sender:
+      - type: not_replicated
+      - type: last_n
+        count: 10
+      - type: grid
+        grid: 1x1h(keep=all) | 24x1h | 14x1d
+        regex: "^zrepl_.*"
+    keep_receiver:
+      - type: grid
+        grid: 1x1h(keep=all) | 24x1h | 35x1d | 6x30d
+        regex: "^zrepl_.*"
+```
+

replication/driver/replication_driver.go

@@ -89,8 +89,6 @@ 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 {
@@ -172,12 +170,11 @@ 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, initialConcurrency int, planner Planner) (ReportFunc, WaitFunc, SetConcurrencyFunc) {
+func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
 	log := getLog(ctx)
 	l := chainlock.New()
 	run := &run{
@@ -185,8 +182,6 @@ func Do(ctx context.Context, initialConcurrency int, planner Planner) (ReportFun
 		startedAt: time.Now(),
 	}
 
-	concurrencyChanges := make(chan concurrencyChange)
-
 	done := make(chan struct{})
 	go func() {
 		defer close(done)
@@ -203,21 +198,15 @@ func Do(ctx context.Context, initialConcurrency int, planner Planner) (ReportFun
 			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,
-				concurrency: prevConcurrency,
 			}
 			run.attempts = append(run.attempts, cur)
 			run.l.DropWhile(func() {
-				cur.do(ctx, prev, concurrencyChanges)
+				cur.do(ctx, prev)
 			})
 			prev = cur
 			if ctx.Err() != nil {
@@ -288,25 +277,10 @@ func Do(ctx context.Context, initialConcurrency int, planner Planner) (ReportFun
 		defer run.l.Lock().Unlock()
 		return run.report()
 	}
-	setConcurrency := func(concurrency int) (reterr error) {
+	return report, wait
-		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
 }
 
-type concurrencyChange struct {
+func (a *attempt) do(ctx context.Context, prev *attempt) {
-	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()
@@ -380,8 +354,8 @@ func (a *attempt) do(ctx context.Context, prev *attempt, setConcurrency <-chan c
 	}
 	// invariant: prevs contains an entry for each unambigious correspondence
 
-	stepQueue := newStepQueue(a.concurrency)
+	stepQueue := newStepQueue()
-	defer stepQueue.Start()()
+	defer stepQueue.Start(1)() // TODO parallel replication
 	var fssesDone sync.WaitGroup
 	for _, f := range a.fss {
 		fssesDone.Add(1)
@@ -390,27 +364,8 @@ func (a *attempt) do(ctx context.Context, prev *attempt, setConcurrency <-chan c
 			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()
 }
@@ -506,8 +461,7 @@ 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()
-			ctx, done := pq.WaitReady(ctx, fs, targetDate)()
+			defer pq.WaitReady(fs, targetDate)()
-			defer done()
 			err = s.step.Step(ctx) // no shadow
 			errTime = time.Now()   // no shadow
 		})
@@ -528,7 +482,6 @@ 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()

replication/driver/replication_driver_test.go

@@ -151,7 +151,7 @@ func TestReplication(t *testing.T) {
 	ctx := context.Background()
 
 	mp := &mockPlanner{}
-	getReport, wait, _ := Do(ctx, 1, mp)
+	getReport, wait := Do(ctx, mp)
 	begin := time.Now()
 	fireAt := []time.Duration{
 		// the following values are relative to the start

replication/driver/replication_stepqueue.go

@@ -2,8 +2,6 @@ package driver
 
 import (
 	"container/heap"
-	"fmt"
-	"sync"
 	"time"
 
 	"github.com/zrepl/zrepl/util/chainlock"
@@ -13,88 +11,51 @@ 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 {
-	res := h.items[i].req.targetDate.Before(h.items[j].req.targetDate)
+	return h[i].req.targetDate.Before(h[j].req.targetDate)
-	if h.reverse {
-		return !res
-	}
-	return res
 }
 
 func (h stepQueueHeap) Swap(i, j int) {
-	h.items[i], h.items[j] = h.items[j], h.items[i]
+	h[i], h[j] = h[j], h[i]
-	h.items[i].idx = i
+	h[i].idx = i
-	h.items[j].idx = j
+	h[j].idx = j
 }
 
 func (h stepQueueHeap) Len() int {
-	return len(h.items)
+	return len(h)
 }
 
 func (h *stepQueueHeap) Push(elem interface{}) {
 	hitem := elem.(*stepQueueHeapItem)
 	hitem.idx = h.Len()
-	h.items = append(h.items, hitem)
+	*h = append(*h, hitem)
 }
 
 func (h *stepQueueHeap) Pop() interface{} {
-	elem := h.items[h.Len()-1]
+	elem := (*h)[h.Len()-1]
 	elem.idx = -1
-	h.items = h.items[:h.Len()-1]
+	*h = (*h)[:h.Len()-1]
 	return elem
 }
 
 // returned stepQueue must be closed with method Close
-func newStepQueue(concurrency int) *stepQueue {
+func newStepQueue() *stepQueue {
-	l := chainlock.New()
 	q := &stepQueue{
 		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
 }
@@ -104,12 +65,25 @@ func newStepQueue(concurrency int) *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() (done func()) {
+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
 	go func() { // "stopper" goroutine
 		<-q.stop
-		defer q.l.Lock().Unlock()
+		defer l.Lock().Unlock()
-		q.stopped = true
+		stopped = true
-		q.pendingCond.Broadcast()
+		pendingCond.Broadcast()
 	}()
 	go func() { // "reqs" goroutine
 		for {
@@ -123,52 +97,41 @@ func (q *stepQueue) Start() (done func()) {
 				}
 			case req := <-q.reqs:
 				func() {
-					defer q.l.Lock().Unlock()
+					defer l.Lock().Unlock()
-					if _, ok := q.queueItems[req.ident]; ok {
+					if _, ok := queueItems[req.ident]; ok {
 						panic("WaitReady must not be called twice for the same ident")
 					}
 					qitem := &stepQueueHeapItem{
 						req: req,
 					}
-					q.queueItems[req.ident] = qitem
+					queueItems[req.ident] = qitem
-					heap.Push(q.pending, qitem)
+					heap.Push(pending, qitem)
-					q.pendingCond.Broadcast()
+					pendingCond.Broadcast()
 				}()
 			}
 		}
 	}()
 	go func() { // "wake" goroutine
-		defer q.l.Lock().Unlock()
+		defer l.Lock().Unlock()
 		for {
 
-			for !q.stopped && (q.active.Len() >= q.concurrency || q.pending.Len() == 0) {
+			for !stopped && (active >= concurrency || pending.Len() == 0) {
-				q.pendingCond.Wait()
+				pendingCond.Wait()
 			}
-			if q.stopped {
+			if stopped {
 				return
 			}
-			if q.pending.Len() <= 0 {
+			if pending.Len() <= 0 {
 				return
 			}
+			active++
+			next := heap.Pop(pending).(*stepQueueHeapItem).req
+			delete(queueItems, next.ident)
 
-			// pop from tracked items
+			next.wakeup <- func() {
-			next := heap.Pop(q.pending).(*stepQueueHeapItem)
+				defer l.Lock().Unlock()
-
+				active--
-			next.req.cancelDueToConcurrencyDownsize = 
+				pendingCond.Broadcast()
-
-			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()
 			}
 		}
 	}()
@@ -198,24 +161,3 @@ 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)
-}

replication/driver/replication_stepqueue_test.go

@@ -17,7 +17,7 @@ import (
 // (relies on scheduler responsivity of < 500ms)
 func TestPqNotconcurrent(t *testing.T) {
 	var ctr uint32
-	q := newStepQueue(1)
+	q := newStepQueue()
 	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()()
+	defer q.Start(1)()
 	time.Sleep(500 * time.Millisecond)
 
 	// while "1" is still running, queue in "2", "3" and "4"
@@ -78,8 +78,7 @@ func (r record) String() string {
 // times for each step are close together.
 func TestPqConcurrent(t *testing.T) {
 
-	concurrency := 5
+	q := newStepQueue()
-	q := newStepQueue(concurrency)
 	var wg sync.WaitGroup
 	filesystems := 100
 	stepsPerFS := 20
@@ -105,7 +104,8 @@ func TestPqConcurrent(t *testing.T) {
 			records <- recs
 		}(fs)
 	}
-	defer q.Start()()
+	concurrency := 5
+	defer q.Start(concurrency)()
 	wg.Wait()
 	close(records)
 	t.Logf("loop done")

replication/replication.go

@@ -8,6 +8,6 @@ import (
 	"github.com/zrepl/zrepl/replication/driver"
 )
 
-func Do(ctx context.Context, initialConcurrency int, planner driver.Planner) (driver.ReportFunc, driver.WaitFunc, driver.SetConcurrencyFunc) {
+func Do(ctx context.Context, planner driver.Planner) (driver.ReportFunc, driver.WaitFunc) {
-	return driver.Do(ctx, initialConcurrency, planner)
+	return driver.Do(ctx, planner)
 }

replication/report/replication_report.go

@@ -10,7 +10,6 @@ type Report struct {
 	WaitReconnectSince, WaitReconnectUntil time.Time
 	WaitReconnectError                     *TimedError
 	Attempts                               []*AttemptReport
-	Concurrency                            int
 }
 
 var _, _ = json.Marshal(&Report{})