DO NOT MERGE: timeoutconn: drop the only use of "unsafe" in the code base

Unless we're using the global source for math/rand, (*rand.Rand).Read
should not be called concurrently. We seed the rng in daemon.Run to
avoid ambiguity or hiding global side effects inside packages.

closes #414

.circleci/config.yml

@@ -204,7 +204,7 @@ workflows:
   periodic:
     triggers:
       - schedule:
-          cron: "45 15 * * *"
+          cron: "00 17 * * *"
           filters:
             branches:
               only:
@@ -212,7 +212,7 @@ workflows:
                 - stable
                 - problame/circleci-build
     jobs:
-      - periodic-trigger-pipeline
+      - periodic-full-pipeline-run
 
   zrepl.github.io:
     jobs:
@@ -342,14 +342,12 @@ jobs:
           description: CI-generated release artifacts
           minio-dst: ""
 
-  periodic-trigger-pipeline:
+  periodic-full-pipeline-run:
     docker:
       - image: cimg/base:2020.08
     steps:
       - trigger-pipeline:
-          body_no_shell_subst: '{"branch":"master", "parameters": { "do_ci": true, "do_release": true }}'
+          body_no_shell_subst: '{"branch":"<<pipeline.git.branch>>", "parameters": { "do_ci": true, "do_release": true }}'
-      - trigger-pipeline:
-          body_no_shell_subst: '{"branch":"stable", "parameters": { "do_ci": true, "do_release": true }}'
 
   publish-zrepl-github-io:
     docker:

README.md

@@ -1,7 +1,7 @@
 [![GitHub license](https://img.shields.io/github/license/zrepl/zrepl.svg)](https://github.com/zrepl/zrepl/blob/master/LICENSE)
 [![Language: Go](https://img.shields.io/badge/language-Go-6ad7e5.svg)](https://golang.org/)
 [![User Docs](https://img.shields.io/badge/docs-web-blue.svg)](https://zrepl.github.io)
-[![Donate via Patreon](https://img.shields.io/endpoint.svg?url=https%3A%2F%2Fshieldsio-patreon.vercel.app%2Fapi%3Fusername%3Dzrepl%26type%3Dpatrons&style=flat&color=yellow)](https://patreon.com/zrepl)
+[![Support me on Patreon](https://img.shields.io/badge/dynamic/json?color=yellow&label=Patreon&query=data.attributes.patron_count&suffix=%20patrons&url=https%3A%2F%2Fwww.patreon.com%2Fapi%2Fcampaigns%2F3095079)](https://patreon.com/zrepl)
 [![Donate via GitHub Sponsors](https://img.shields.io/static/v1?label=Sponsor&message=%E2%9D%A4&logo=GitHub&style=flat&color=yellow)](https://github.com/sponsors/problame)
 [![Donate via Liberapay](https://img.shields.io/liberapay/patrons/zrepl.svg?logo=liberapay)](https://liberapay.com/zrepl/donate)
 [![Donate via PayPal](https://img.shields.io/badge/donate-paypal-yellow.svg)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96)

daemon/daemon.go

@@ -3,6 +3,7 @@ package daemon
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"os"
 	"os/signal"
 	"strings"
@@ -38,6 +39,12 @@ func Run(ctx context.Context, conf *config.Config) error {
 		cancel()
 	}()
 
+	// The math/rand package is used presently for generating trace IDs, we
+	// seed it with the current time and pid so that the IDs are mostly
+	// unique.
+	rand.Seed(time.Now().UnixNano())
+	rand.Seed(int64(os.Getpid()))
+
 	outlets, err := logging.OutletsFromConfig(*conf.Global.Logging)
 	if err != nil {
 		return errors.Wrap(err, "cannot build logging from config")

daemon/job/snapjob.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"sync"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -29,7 +30,8 @@ type SnapJob struct {
 
 	promPruneSecs *prometheus.HistogramVec // labels: prune_side
 
-	pruner *pruner.Pruner
+	prunerMtx sync.Mutex
+	pruner    *pruner.Pruner
 }
 
 func (j *SnapJob) Name() string { return j.name.String() }
@@ -77,9 +79,11 @@ type SnapJobStatus struct {
 func (j *SnapJob) Status() *Status {
 	s := &SnapJobStatus{}
 	t := j.Type()
+	j.prunerMtx.Lock()
 	if j.pruner != nil {
 		s.Pruning = j.pruner.Report()
 	}
+	j.prunerMtx.Unlock()
 	s.Snapshotting = j.snapper.Report()
 	return &Status{Type: t, JobSpecific: s}
 }
@@ -177,7 +181,9 @@ func (j *SnapJob) doPrune(ctx context.Context) {
 		// FIXME encryption setting is irrelevant for SnapJob because the endpoint is only used as pruner.Target
 		Encrypt: &zfs.NilBool{B: true},
 	})
+	j.prunerMtx.Lock()
 	j.pruner = j.prunerFactory.BuildLocalPruner(ctx, sender, alwaysUpToDateReplicationCursorHistory{sender})
+	j.prunerMtx.Unlock()
 	log.Info("start pruning")
 	j.pruner.Prune()
 	log.Info("finished pruning")

daemon/logging/trace/trace_genID.go

@@ -3,20 +3,11 @@ package trace
 import (
 	"encoding/base64"
 	"math/rand"
-	"os"
 	"strings"
-	"time"
 
 	"github.com/zrepl/zrepl/util/envconst"
 )
 
-var genIdPRNG = rand.New(rand.NewSource(1))
-
-func init() {
-	genIdPRNG.Seed(time.Now().UnixNano())
-	genIdPRNG.Seed(int64(os.Getpid()))
-}
-
 var genIdNumBytes = envconst.Int("ZREPL_TRACE_ID_NUM_BYTES", 3)
 
 func init() {
@@ -30,7 +21,7 @@ func genID() string {
 	enc := base64.NewEncoder(base64.RawStdEncoding, &out)
 	buf := make([]byte, genIdNumBytes)
 	for i := 0; i < len(buf); {
-		n, err := genIdPRNG.Read(buf[i:])
+		n, err := rand.Read(buf[i:])
 		if err != nil {
 			panic(err)
 		}

docs/configuration/overview.rst

@@ -188,7 +188,7 @@ Encoding the job name in the names ensures that multiple sending jobs can replic
 
 .. _tentative-replication-cursor-bookmarks:
 
-**Tentative replication cursor bookmarks** are short-lived boomkarks that protect the atomic moving-forward of the replication cursor and last-received-hold (see :issue:`this issue <340>`).
+**Tentative replication cursor bookmarks** are short-lived bookmarks that protect the atomic moving-forward of the replication cursor and last-received-hold (see :issue:`this issue <340>`).
 They are only necessary if step holds are not used as per the :ref:`replication.protection <replication-option-protection>` setting.
 The tentative replication cursor has the format ``#zrepl_CUSORTENTATIVE_G_<GUID>_J_<JOBNAME>``.
 The ``zrepl zfs-abstraction list`` command provides a listing of all bookmarks and holds managed by zrepl.
@@ -255,7 +255,7 @@ On your setup, ensure that
 
 * all ``filesystems`` filter specifications are disjoint
 * no ``root_fs`` is a prefix or equal to another ``root_fs``
-* no ``filesystems`` filter maches any ``root_fs``
+* no ``filesystems`` filter matches any ``root_fs``
 
 **Exceptions to the rule**:
 

docs/global.rst.inc

@@ -13,7 +13,7 @@
    :target: https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96
 .. |Donate via Liberapay| image:: https://img.shields.io/liberapay/patrons/zrepl.svg?logo=liberapay
    :target: https://liberapay.com/zrepl/donate
-.. |Donate via Patreon| image:: https://img.shields.io/endpoint.svg?url=https%3A%2F%2Fshieldsio-patreon.vercel.app%2Fapi%3Fusername%3Dzrepl%26type%3Dpatrons&style=flat&color=yellow
+.. |Donate via Patreon| image:: https://img.shields.io/badge/dynamic/json?color=yellow&label=Patreon&query=data.attributes.patron_count&suffix=%20patrons&url=https%3A%2F%2Fwww.patreon.com%2Fapi%2Fcampaigns%2F3095079
    :target: https://www.patreon.com/zrepl
 .. |Donate via GitHub Sponsors| image:: https://img.shields.io/static/v1?label=Sponsor&message=%E2%9D%A4&logo=GitHub&style=flat&color=yellow
    :target: https://github.com/sponsors/problame

docs/supporters.rst

@@ -32,6 +32,7 @@ We would like to thank the following people and organizations for supporting zre
 
    <div class="fa fa-code" style="width: 1em;"></div>
 
+* |supporter-gold| Cyberiada GmbH
 * |supporter-std| `Gordon Schulz <https://github.com/azmodude>`_
 * |supporter-std| `@jwittlincohen <https://github.com/jwittlincohen>`_
 * |supporter-std| `Michael D. Schmitt <https://waterbendingscroll.dancingdragons.org>`_

lazy.sh

@@ -39,12 +39,7 @@ godep() {
     export GOOS="$GOHOSTOS"
     export GOARCH="$GOHOSTARCH"
     # TODO GOARM=$GOHOSTARM?
-    go build -v -mod=readonly -o "$GOPATH/bin/stringer"      golang.org/x/tools/cmd/stringer
+    cat tools.go | grep _ | awk -F'"' '{print $2}' | tee | xargs -tI '{}' go install '{}'
-    go build -v -mod=readonly -o "$GOPATH/bin/protoc-gen-go" github.com/golang/protobuf/protoc-gen-go
-    go build -v -mod=readonly -o "$GOPATH/bin/enumer"        github.com/alvaroloes/enumer
-    go build -v -mod=readonly -o "$GOPATH/bin/goimports"     golang.org/x/tools/cmd/goimports
-    go build -v -mod=readonly -o "$GOPATH/bin/golangci-lint" github.com/golangci/golangci-lint/cmd/golangci-lint
-    go build -v -mod=readonly -o "$GOPATH/bin/gocovmerge"    github.com/wadey/gocovmerge
     set +x
     popd
     if ! type stringer || ! type protoc-gen-go || ! type enumer || ! type goimports || ! type golangci-lint || ! type gocovmerge; then

replication/driver/replication_driver.go

@@ -492,57 +492,37 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 	// => don't set f.planning.done just yet
 	f.debug("initial len(fs.planned.steps) = %d", len(f.planned.steps))
 
-	// for not-first attempts, only allow fs.planned.steps
+	// for not-first attempts that succeeded in planning, only allow fs.planned.steps
-	// up to including the originally planned target snapshot
+	// up to and including the originally planned target snapshot
 	if prev != nil && prev.planning.done && prev.planning.err == nil {
+		f.debug("attempting to correlate plan with previous attempt to find out what is left to do")
+		// find the highest of the previously uncompleted steps for which we can also find a step
+		// in our current plan
 		prevUncompleted := prev.planned.steps[prev.planned.step:]
-		if len(prevUncompleted) == 0 {
+		var target struct{ prev, cur int }
-			f.debug("prevUncompleted is empty")
+		target.prev = -1
-			return
+		target.cur = -1
-		}
+	out:
-		if len(f.planned.steps) == 0 {
+		for p := len(prevUncompleted) - 1; p >= 0; p-- {
-			f.debug("fs.planned.steps is empty")
+			for q := len(f.planned.steps) - 1; q >= 0; q-- {
-			return
+				if prevUncompleted[p].step.TargetEquals(f.planned.steps[q].step) {
-		}
+					target.prev = p
-		prevFailed := prevUncompleted[0]
+					target.cur = q
-		curFirst := f.planned.steps[0]
+					break out
-		// we assume that PlanFS retries prevFailed (using curFirst)
-		if !prevFailed.step.TargetEquals(curFirst.step) {
-			f.debug("Targets don't match")
-			// Two options:
-			// A: planning algorithm is broken
-			// B: manual user intervention inbetween
-			// Neither way will we make progress, so let's error out
-			stepFmt := func(step *step) string {
-				r := step.report()
-				s := r.Info
-				if r.IsIncremental() {
-					return fmt.Sprintf("%s=>%s", s.From, s.To)
-				} else {
-					return fmt.Sprintf("full=>%s", s.To)
 				}
 			}
-			msg := fmt.Sprintf("last attempt's uncompleted step %s does not correspond to this attempt's first planned step %s",
+		}
-				stepFmt(prevFailed), stepFmt(curFirst))
+		if target.prev == -1 || target.cur == -1 {
-			f.planned.stepErr = newTimedError(errors.New(msg), time.Now())
+			f.debug("no correlation possible between previous attempt and this attempt's plan")
+			f.planning.err = newTimedError(fmt.Errorf("cannot correlate previously failed attempt to current plan"), time.Now())
 			return
 		}
-		// only allow until step targets diverge
+
-		min := len(prevUncompleted)
+		f.planned.steps = f.planned.steps[0:target.cur]
-		if min > len(f.planned.steps) {
+		f.debug("found correlation, new steps are len(fs.planned.steps) = %d", len(f.planned.steps))
-			min = len(f.planned.steps)
+	} else {
-		}
+		f.debug("previous attempt does not exist or did not finish planning, no correlation possible, taking this attempt's plan as is")
-		diverge := 0
-		for ; diverge < min; diverge++ {
-			f.debug("diverge compare iteration %d", diverge)
-			if !f.planned.steps[diverge].step.TargetEquals(prevUncompleted[diverge].step) {
-				break
-			}
-		}
-		f.debug("diverge is %d", diverge)
-		f.planned.steps = f.planned.steps[0:diverge]
 	}
-	f.debug("post-prev-merge len(fs.planned.steps) = %d", len(f.planned.steps))
 
 	// now we are done planning (f.planned.steps won't change from now on)
 	f.planning.done = true

rpc/dataconn/frameconn/frameconn.go

@@ -47,7 +47,7 @@ func (f *FrameHeader) Unmarshal(buf []byte) {
 
 type Conn struct {
 	readMtx, writeMtx sync.Mutex
-	nc                timeoutconn.Conn
+	nc                *timeoutconn.Conn
 	readNextValid     bool
 	readNext          FrameHeader
 	nextReadErr       error
@@ -55,7 +55,7 @@ type Conn struct {
 	shutdown          shutdownFSM
 }
 
-func Wrap(nc timeoutconn.Conn) *Conn {
+func Wrap(nc *timeoutconn.Conn) *Conn {
 	return &Conn{
 		nc: nc,
 		//		ncBuf: bufio.NewReadWriter(bufio.NewReaderSize(nc, 1<<23), bufio.NewWriterSize(nc, 1<<23)),

rpc/dataconn/stream/stream_conn.go

@@ -144,6 +144,7 @@ func (c *Conn) ReadStream(frameType uint32, closeConnOnClose bool) (_ *StreamRea
 
 	c.readMtx.Lock()
 	if !c.readClean {
+		c.readMtx.Unlock()
 		return nil, errWriteStreamToErrorUnknownState
 	}
 

rpc/dataconn/timeoutconn/timeoutconn.go

@@ -11,7 +11,7 @@ import (
 	"errors"
 	"io"
 	"net"
-	"sync/atomic"
+	"sync"
 	"syscall"
 	"time"
 )
@@ -54,39 +54,60 @@ type Wire interface {
 }
 
 type Conn struct {
+	// immutable state
+
 	Wire
-	renewDeadlinesDisabled int32
+	idleTimeout time.Duration
-	idleTimeout            time.Duration
+
+	// mutable state (protected by mtx)
+
+	mtx                    sync.RWMutex
+	renewDeadlinesDisabled bool
 }
 
-func Wrap(conn Wire, idleTimeout time.Duration) Conn {
+func Wrap(conn Wire, idleTimeout time.Duration) *Conn {
-	return Conn{Wire: conn, idleTimeout: idleTimeout}
+	return &Conn{
+		Wire:        conn,
+		idleTimeout: idleTimeout,
+	}
 }
 
 // DisableTimeouts disables the idle timeout behavior provided by this package.
-// Existing deadlines are cleared iff the call is the first call to this method.
+// Existing deadlines are cleared iff the call is the first call to this method
+// or if the previous call produced an error.
 func (c *Conn) DisableTimeouts() error {
-	if atomic.CompareAndSwapInt32(&c.renewDeadlinesDisabled, 0, 1) {
+	c.mtx.Lock()
-		return c.SetDeadline(time.Time{})
+	defer c.mtx.Unlock()
+	if c.renewDeadlinesDisabled {
+		return nil
 	}
+	err := c.SetDeadline(time.Time{})
+	if err != nil {
+		return err
+	}
+	c.renewDeadlinesDisabled = true
 	return nil
 }
 
 func (c *Conn) renewReadDeadline() error {
-	if atomic.LoadInt32(&c.renewDeadlinesDisabled) != 0 {
+	c.mtx.RLock()
+	defer c.mtx.RUnlock()
+	if c.renewDeadlinesDisabled {
 		return nil
 	}
 	return c.SetReadDeadline(time.Now().Add(c.idleTimeout))
 }
 
 func (c *Conn) RenewWriteDeadline() error {
-	if atomic.LoadInt32(&c.renewDeadlinesDisabled) != 0 {
+	c.mtx.RLock()
+	defer c.mtx.RUnlock()
+	if c.renewDeadlinesDisabled {
 		return nil
 	}
 	return c.SetWriteDeadline(time.Now().Add(c.idleTimeout))
 }
 
-func (c Conn) Read(p []byte) (n int, err error) {
+func (c *Conn) Read(p []byte) (n int, err error) {
 	n = 0
 	err = nil
 restart:
@@ -103,7 +124,7 @@ restart:
 	return n, err
 }
 
-func (c Conn) Write(p []byte) (n int, err error) {
+func (c *Conn) Write(p []byte) (n int, err error) {
 	n = 0
 restart:
 	if err := c.RenewWriteDeadline(); err != nil {
@@ -123,7 +144,7 @@ restart:
 // but is guaranteed to use the writev system call if the wrapped Wire
 // support it.
 // Note the Conn does not support writev through io.Copy(aConn, aNetBuffers).
-func (c Conn) WritevFull(bufs net.Buffers) (n int64, err error) {
+func (c *Conn) WritevFull(bufs net.Buffers) (n int64, err error) {
 	n = 0
 restart:
 	if err := c.RenewWriteDeadline(); err != nil {
@@ -163,12 +184,12 @@ var _ SyscallConner = (*net.TCPConn)(nil)
 // If the connection returned io.EOF, the number of bytes written until
 // then + io.EOF is returned. This behavior is different to io.ReadFull
 // which returns io.ErrUnexpectedEOF.
-func (c Conn) ReadvFull(buffers net.Buffers) (n int64, err error) {
+func (c *Conn) ReadvFull(buffers net.Buffers) (n int64, err error) {
 	return c.readv(buffers)
 }
 
 // invoked by c.readv if readv system call cannot be used
-func (c Conn) readvFallback(nbuffers net.Buffers) (n int64, err error) {
+func (c *Conn) readvFallback(nbuffers net.Buffers) (n int64, err error) {
 	buffers := [][]byte(nbuffers)
 	for i := range buffers {
 		curBuf := buffers[i]

rpc/dataconn/timeoutconn/timeoutconn_readv_unsupported.go

@@ -1,10 +1,8 @@
-// +build illumos solaris
-
 package timeoutconn
 
 import "net"
 
-func (c Conn) readv(buffers net.Buffers) (n int64, err error) {
+func (c *Conn) readv(buffers net.Buffers) (n int64, err error) {
 	// Go does not expose the SYS_READV symbol for Solaris / Illumos - do they have it?
 	// Anyhow, use the fallback
 	return c.readvFallback(buffers)

rpc/dataconn/timeoutconn/timoutconn_readv.go

@@ -1,133 +0,0 @@
-// +build !illumos
-// +build !solaris
-
-package timeoutconn
-
-import (
-	"io"
-	"net"
-	"syscall"
-	"unsafe"
-)
-
-func buildIovecs(buffers net.Buffers) (totalLen int64, vecs []syscall.Iovec) {
-	vecs = make([]syscall.Iovec, 0, len(buffers))
-	for i := range buffers {
-		totalLen += int64(len(buffers[i]))
-		if len(buffers[i]) == 0 {
-			continue
-		}
-
-		v := syscall.Iovec{
-			Base: &buffers[i][0],
-		}
-		// syscall.Iovec.Len has platform-dependent size, thus use SetLen
-		v.SetLen(len(buffers[i]))
-
-		vecs = append(vecs, v)
-	}
-	return totalLen, vecs
-}
-
-func (c Conn) readv(buffers net.Buffers) (n int64, err error) {
-
-	scc, ok := c.Wire.(SyscallConner)
-	if !ok {
-		return c.readvFallback(buffers)
-	}
-	rawConn, err := scc.SyscallConn()
-	if err == SyscallConnNotSupported {
-		return c.readvFallback(buffers)
-	}
-	if err != nil {
-		return 0, err
-	}
-
-	_, iovecs := buildIovecs(buffers)
-
-	for len(iovecs) > 0 {
-		if err := c.renewReadDeadline(); err != nil {
-			return n, err
-		}
-		oneN, oneErr := c.doOneReadv(rawConn, &iovecs)
-		n += oneN
-		if netErr, ok := oneErr.(net.Error); ok && netErr.Timeout() && oneN > 0 { // TODO likely not working
-			continue
-		} else if oneErr == nil && oneN > 0 {
-			continue
-		} else {
-			return n, oneErr
-		}
-	}
-	return n, nil
-}
-
-func (c Conn) doOneReadv(rawConn syscall.RawConn, iovecs *[]syscall.Iovec) (n int64, err error) {
-	rawReadErr := rawConn.Read(func(fd uintptr) (done bool) {
-		// iovecs, n and err must not be shadowed!
-
-		// NOTE: unsafe.Pointer safety rules
-		// 		https://tip.golang.org/pkg/unsafe/#Pointer
-		//
-		//		(4) Conversion of a Pointer to a uintptr when calling syscall.Syscall.
-		// 		...
-		//		uintptr() conversions must appear within the syscall.Syscall argument list.
-		//      (even though we are not the escape analysis Likely not )
-		thisReadN, _, errno := syscall.Syscall(
-			syscall.SYS_READV,
-			fd,
-			uintptr(unsafe.Pointer(&(*iovecs)[0])),
-			uintptr(len(*iovecs)),
-		)
-		if thisReadN == ^uintptr(0) {
-			if errno == syscall.EAGAIN {
-				return false
-			}
-			err = syscall.Errno(errno)
-			return true
-		}
-		if int(thisReadN) < 0 {
-			panic("unexpected return value")
-		}
-		n += int64(thisReadN) // TODO check overflow
-
-		// shift iovecs forward
-		for left := int(thisReadN); left > 0; {
-			// conversion to uint does not change value, see TestIovecLenFieldIsMachineUint, and left > 0
-			thisIovecConsumedCompletely := uint((*iovecs)[0].Len) <= uint(left)
-			if thisIovecConsumedCompletely {
-				// Update left, cannot go below 0 due to
-				// a) definition of thisIovecConsumedCompletely
-				// b) left > 0 due to loop invariant
-				// Converting .Len to int64 is thus also safe now, because it is < left < INT_MAX
-				left -= int((*iovecs)[0].Len)
-				*iovecs = (*iovecs)[1:]
-			} else {
-				// trim this iovec to remaining length
-
-				// NOTE: unsafe.Pointer safety rules
-				// 		https://tip.golang.org/pkg/unsafe/#Pointer
-				// 		(3) Conversion of a Pointer to a uintptr and back, with arithmetic.
-				// 		...
-				//		Note that both conversions must appear in the same expression,
-				//		with only the intervening arithmetic between them:
-				(*iovecs)[0].Base = (*byte)(unsafe.Pointer(uintptr(unsafe.Pointer((*iovecs)[0].Base)) + uintptr(left)))
-				curVecNewLength := uint((*iovecs)[0].Len) - uint(left) // casts to uint do not change value
-				(*iovecs)[0].SetLen(int(curVecNewLength))              // int and uint have the same size, no change of value
-
-				break // inner
-			}
-		}
-		if thisReadN == 0 {
-			err = io.EOF
-			return true
-		}
-		return true
-	})
-
-	if rawReadErr != nil {
-		err = rawReadErr
-	}
-
-	return n, err
-}