replication/driver: simplify second-attempt step correlation code & fix statekeeping

Before this change, the step correlation code returned early in several cases:
- did not set f.planning.done in the cases where it was a no-op
- did not set f.planning.err in the cases where correlation did not
  succeed

Reported-by: InsanePrawn <insane.prawny@gmail.com>

daemon/daemon.go

@@ -3,7 +3,6 @@ package daemon
 import (
 	"context"
 	"fmt"
-	"math/rand"
 	"os"
 	"os/signal"
 	"strings"
@@ -39,12 +38,6 @@ 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,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"sort"
-	"sync"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -30,8 +29,7 @@ type SnapJob struct {
 
 	promPruneSecs *prometheus.HistogramVec // labels: prune_side
 
-	prunerMtx sync.Mutex
+	pruner *pruner.Pruner
-	pruner    *pruner.Pruner
 }
 
 func (j *SnapJob) Name() string { return j.name.String() }
@@ -79,11 +77,9 @@ 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}
 }
@@ -181,9 +177,7 @@ 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,11 +3,20 @@ 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() {
@@ -21,7 +30,7 @@ func genID() string {
 	enc := base64.NewEncoder(base64.RawStdEncoding, &out)
 	buf := make([]byte, genIdNumBytes)
 	for i := 0; i < len(buf); {
-		n, err := rand.Read(buf[i:])
+		n, err := genIdPRNG.Read(buf[i:])
 		if err != nil {
 			panic(err)
 		}

lazy.sh

@@ -39,7 +39,12 @@ godep() {
     export GOOS="$GOHOSTOS"
     export GOARCH="$GOHOSTARCH"
     # TODO GOARM=$GOHOSTARM?
-    cat tools.go | grep _ | awk -F'"' '{print $2}' | tee | xargs -tI '{}' go install '{}'
+    go build -v -mod=readonly -o "$GOPATH/bin/stringer"      golang.org/x/tools/cmd/stringer
+    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

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,7 +144,6 @@ 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"
+	"sync/atomic"
 	"syscall"
 	"time"
 )
@@ -54,60 +54,39 @@ type Wire interface {
 }
 
 type Conn struct {
-	// immutable state
-
 	Wire
-	idleTimeout time.Duration
+	renewDeadlinesDisabled int32
-
+	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{
+	return Conn{Wire: conn, idleTimeout: idleTimeout}
-		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 {
-	c.mtx.Lock()
+	if atomic.CompareAndSwapInt32(&c.renewDeadlinesDisabled, 0, 1) {
-	defer c.mtx.Unlock()
+		return c.SetDeadline(time.Time{})
-	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 {
-	c.mtx.RLock()
+	if atomic.LoadInt32(&c.renewDeadlinesDisabled) != 0 {
-	defer c.mtx.RUnlock()
-	if c.renewDeadlinesDisabled {
 		return nil
 	}
 	return c.SetReadDeadline(time.Now().Add(c.idleTimeout))
 }
 
 func (c *Conn) RenewWriteDeadline() error {
-	c.mtx.RLock()
+	if atomic.LoadInt32(&c.renewDeadlinesDisabled) != 0 {
-	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:
@@ -124,7 +103,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 {
@@ -144,7 +123,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 {
@@ -184,12 +163,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,8 +1,10 @@
+// +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

@@ -0,0 +1,133 @@
+// +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
+}