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

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

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