Compare commits

...

6 Commits

Author SHA1 Message Date
Christian Schwarz bdd0c9a199 build: use go 1.19 for testing & release builds
New docker image since the old one was deprecated, according
to https://discuss.circleci.com/t/go-lang-docker-image-circleci-golang-1-19-is-missing/44961
2022-10-24 22:37:35 +02:00
Christian Schwarz 498036194d build: update golangci-lint
The previous commits were done in response to updating to
the version that we now pin in this commit.
We do the update after the fixes so that each commit builds.
2022-10-24 22:22:41 +02:00
Christian Schwarz a91fb873e4 fix incorrect use of sort.StringSlice
A newer version of staticheck found these:

> SA4029: sort.StringSlice is a type, not a function, and
> sort.StringSlice(variants) doesn't sort your values; consider using
> sort.Strings instead (staticcheck)
2022-10-24 22:22:41 +02:00
Christian Schwarz a6aa610165 run go1.19 gofmt and make adjustments as needed
(Go 1.19 expanded doc comment syntax)
2022-10-24 22:22:41 +02:00
Christian Schwarz 6c87bdb9fb go1.19: switch to new nolint directive that is compatible with Go 1.19 gofmt 2022-10-24 22:22:11 +02:00
Christian Schwarz b9250a41a2 go1.18: address net.Error.Temporary() deprecation
Go 1.18 deprecated net.Error.Temporary().
This commit cleans up places where we use it incorrectly.
Also, the rpc layer defines some errors that implement

  interface { Temporary() bool }

I added comments to all of the implementations to indicate
whether they will be required if net.Error.Temporary is ever
ever removed in the future.

For HandshakeError, the Temporary() return value is actually
important. I moved & rewrote a (previously misplaced) comment
there.

The ReadStreamError changes were
1. necessary to pacify newer staticcheck and
2. technically, an error can implement Temporary()
   without being net.Err. This applies to some syscall
   errors in the standard library.

Reading list for those interested:
- https://github.com/golang/go/issues/45729
- https://groups.google.com/g/golang-nuts/c/-JcZzOkyqYI
- https://man7.org/linux/man-pages/man2/accept.2.html

Note: This change was prompted by staticheck:

> SA1019: neterr.Temporary has been deprecated since Go 1.18 because it
> shouldn't be used: Temporary errors are not well-defined. Most
> "temporary" errors are timeouts, and the few exceptions are surprising.
> Do not use this method. (staticcheck)
2022-10-24 22:21:52 +02:00
35 changed files with 1373 additions and 237 deletions
+6 -6
View File
@@ -150,7 +150,7 @@ parameters:
release_docker_baseimage_tag: release_docker_baseimage_tag:
type: string type: string
default: "1.17" default: "1.19"
workflows: workflows:
version: 2 version: 2
@@ -160,15 +160,15 @@ workflows:
jobs: jobs:
- quickcheck-docs - quickcheck-docs
- quickcheck-go: &quickcheck-go-smoketest - quickcheck-go: &quickcheck-go-smoketest
name: quickcheck-go-amd64-linux-1.17 name: quickcheck-go-amd64-linux-1.19
goversion: &latest-go-release "1.17" goversion: &latest-go-release "1.19"
goos: linux goos: linux
goarch: amd64 goarch: amd64
- test-go-on-latest-go-release: - test-go-on-latest-go-release:
goversion: *latest-go-release goversion: *latest-go-release
- quickcheck-go: - quickcheck-go:
requires: requires:
- quickcheck-go-amd64-linux-1.17 #quickcheck-go-smoketest.name - quickcheck-go-amd64-linux-1.19 #quickcheck-go-smoketest.name
matrix: &quickcheck-go-matrix matrix: &quickcheck-go-matrix
alias: quickcheck-go-matrix alias: quickcheck-go-matrix
parameters: parameters:
@@ -249,7 +249,7 @@ jobs:
goarch: goarch:
type: string type: string
docker: docker:
- image: circleci/golang:<<parameters.goversion>> - image: cimg/go:<<parameters.goversion>>
environment: environment:
GOOS: <<parameters.goos>> GOOS: <<parameters.goos>>
GOARCH: <<parameters.goarch>> GOARCH: <<parameters.goarch>>
@@ -286,7 +286,7 @@ jobs:
goversion: goversion:
type: string type: string
docker: docker:
- image: circleci/golang:<<parameters.goversion>> - image: cimg/go:<<parameters.goversion>>
steps: steps:
- checkout - checkout
- restore-cache-gomod - restore-cache-gomod
+1 -1
View File
@@ -28,7 +28,7 @@ GO_BUILDFLAGS := $(GO_MOD_READONLY) $(GO_EXTRA_BUILDFLAGS)
GO_BUILD := $(GO_ENV_VARS) $(GO) build $(GO_BUILDFLAGS) -ldflags $(GO_LDFLAGS) GO_BUILD := $(GO_ENV_VARS) $(GO) build $(GO_BUILDFLAGS) -ldflags $(GO_LDFLAGS)
GOLANGCI_LINT := golangci-lint GOLANGCI_LINT := golangci-lint
GOCOVMERGE := gocovmerge GOCOVMERGE := gocovmerge
RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.17 RELEASE_DOCKER_BASEIMAGE_TAG ?= 1.19
RELEASE_DOCKER_BASEIMAGE ?= golang:$(RELEASE_DOCKER_BASEIMAGE_TAG) RELEASE_DOCKER_BASEIMAGE ?= golang:$(RELEASE_DOCKER_BASEIMAGE_TAG)
ifneq ($(GOARM),) ifneq ($(GOARM),)
+3 -6
View File
@@ -4,13 +4,10 @@ go 1.12
require ( require (
github.com/alvaroloes/enumer v1.1.1 github.com/alvaroloes/enumer v1.1.1
github.com/golangci/golangci-lint v1.35.2 github.com/golangci/golangci-lint v1.50.1
github.com/golangci/misspell v0.3.4 // indirect
github.com/golangci/revgrep v0.0.0-20180812185044-276a5c0a1039 // indirect
github.com/spf13/afero v1.2.2 // indirect
github.com/spf13/jwalterweatherman v1.1.0 // indirect github.com/spf13/jwalterweatherman v1.1.0 // indirect
github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad
golang.org/x/tools v0.0.0-20210105210202-9ed45478a130 golang.org/x/tools v0.2.0
google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0 // indirect google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0 // indirect
google.golang.org/protobuf v1.25.0 google.golang.org/protobuf v1.28.0
) )
+1128
View File
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -57,7 +57,7 @@ func (f *zabsFilterFlags) registerZabsFilterFlags(s *pflag.FlagSet, verb string)
for v := range endpoint.AbstractionTypesAll { for v := range endpoint.AbstractionTypesAll {
variants = append(variants, string(v)) variants = append(variants, string(v))
} }
variants = sort.StringSlice(variants) sort.Strings(variants)
variantsJoined := strings.Join(variants, "|") variantsJoined := strings.Join(variants, "|")
s.Var(&f.Types, "type", fmt.Sprintf("only %s holds of the specified type [default: all] [comma-separated list of %s]", verb, variantsJoined)) s.Var(&f.Types, "type", fmt.Sprintf("only %s holds of the specified type [default: all] [comma-separated list of %s]", verb, variantsJoined))
+2 -1
View File
@@ -44,7 +44,8 @@ func TestSampleConfigsAreParsedWithoutErrors(t *testing.T) {
} }
// template must be a template/text template with a single '{{ . }}' as placeholder for val // template must be a template/text template with a single '{{ . }}' as placeholder for val
//nolint[:deadcode,unused] //
//nolint:deadcode,unused
func testValidConfigTemplate(t *testing.T, tmpl string, val string) *Config { func testValidConfigTemplate(t *testing.T, tmpl string, val string) *Config {
tmp, err := template.New("master").Parse(tmpl) tmp, err := template.New("master").Parse(tmpl)
if err != nil { if err != nil {
+2 -3
View File
@@ -5,7 +5,7 @@
// //
// This package also provides all supported hook type implementations and abstractions around them. // This package also provides all supported hook type implementations and abstractions around them.
// //
// Use For Other Kinds Of ExpectStepReports // # Use For Other Kinds Of ExpectStepReports
// //
// This package REQUIRES REFACTORING before it can be used for other activities than snapshots, e.g. pre- and post-replication: // This package REQUIRES REFACTORING before it can be used for other activities than snapshots, e.g. pre- and post-replication:
// //
@@ -15,7 +15,7 @@
// The hook implementations should move out of this package. // The hook implementations should move out of this package.
// However, there is a lot of tight coupling which to untangle isn't worth it ATM. // However, there is a lot of tight coupling which to untangle isn't worth it ATM.
// //
// How This Package Is Used By Package Snapper // # How This Package Is Used By Package Snapper
// //
// Deserialize a config.List using ListFromConfig(). // Deserialize a config.List using ListFromConfig().
// Then it MUST filter the list to only contain hooks for a particular filesystem using // Then it MUST filter the list to only contain hooks for a particular filesystem using
@@ -30,5 +30,4 @@
// Command hooks make it available in the environment variable ZREPL_DRYRUN. // Command hooks make it available in the environment variable ZREPL_DRYRUN.
// //
// Plan.Report() can be called while Plan.Run() is executing to give an overview of plan execution progress (future use in "zrepl status"). // Plan.Report() can be called while Plan.Run() is executing to give an overview of plan execution progress (future use in "zrepl status").
//
package hooks package hooks
+11 -8
View File
@@ -17,19 +17,22 @@ import (
"github.com/zrepl/zrepl/zfs" "github.com/zrepl/zrepl/zfs"
) )
// Hook to implement the following recommmendation from MySQL docs
// https://dev.mysql.com/doc/mysql-backup-excerpt/5.7/en/backup-methods.html // https://dev.mysql.com/doc/mysql-backup-excerpt/5.7/en/backup-methods.html
// //
// Making Backups Using a File System Snapshot: // Making Backups Using a File System Snapshot:
// //
// If you are using a Veritas file system, you can make a backup like this: // If you are using a Veritas file system, you can make a backup like this:
// //
// From a client program, execute FLUSH TABLES WITH READ LOCK. // From a client program, execute FLUSH TABLES WITH READ LOCK.
// From another shell, execute mount vxfs snapshot. // From another shell, execute mount vxfs snapshot.
// From the first client, execute UNLOCK TABLES. // From the first client, execute UNLOCK TABLES.
// Copy files from the snapshot. // Copy files from the snapshot.
// Unmount the snapshot. // Unmount the snapshot.
// //
// Similar snapshot capabilities may be available in other file systems, such as LVM or ZFS. // Similar snapshot capabilities may be available in other file systems, such as LVM or ZFS.
//
type MySQLLockTables struct { type MySQLLockTables struct {
errIsFatal bool errIsFatal bool
connector sqldriver.Connector connector sqldriver.Connector
+1 -1
View File
@@ -138,7 +138,7 @@ outer:
// TODO: // TODO:
// This is a work-around for the current package daemon/pruner // This is a work-around for the current package daemon/pruner
// and package pruning.Snapshot limitation: they require the // and package pruning.Snapshot limitation: they require the
// `Replicated` getter method be present, but obviously, // `Replicated` getter method be present, but obviously,
// a local job like SnapJob can't deliver on that. // a local job like SnapJob can't deliver on that.
// But the pruner.Pruner gives up on an FS if no replication // But the pruner.Pruner gives up on an FS if no replication
// cursor is present, which is why this pruner returns the // cursor is present, which is why this pruner returns the
+38 -41
View File
@@ -1,6 +1,6 @@
// package trace provides activity tracing via ctx through Tasks and Spans // package trace provides activity tracing via ctx through Tasks and Spans
// //
// Basic Concepts // # Basic Concepts
// //
// Tracing can be used to identify where a piece of code spends its time. // Tracing can be used to identify where a piece of code spends its time.
// //
@@ -10,51 +10,50 @@
// to tech-savvy users (albeit not developers). // to tech-savvy users (albeit not developers).
// //
// This package provides the concept of Tasks and Spans to express what activity is happening within an application: // This package provides the concept of Tasks and Spans to express what activity is happening within an application:
// // - Neither task nor span is really tangible but instead contained within the context.Context tree
// - Neither task nor span is really tangible but instead contained within the context.Context tree // - Tasks represent concurrent activity (i.e. goroutines).
// - Tasks represent concurrent activity (i.e. goroutines). // - Spans represent a semantic stack trace within a task.
// - Spans represent a semantic stack trace within a task.
//
// As a consequence, whenever a context is propagated across goroutine boundary, you need to create a child task: // As a consequence, whenever a context is propagated across goroutine boundary, you need to create a child task:
// //
// go func(ctx context.Context) { // go func(ctx context.Context) {
// ctx, endTask = WithTask(ctx, "what-happens-inside-the-child-task") // ctx, endTask = WithTask(ctx, "what-happens-inside-the-child-task")
// defer endTask() // defer endTask()
// // ... // // ...
// }(ctx) // }(ctx)
// //
// Within the task, you can open up a hierarchy of spans. // Within the task, you can open up a hierarchy of spans.
// In contrast to tasks, which have can multiple concurrently running child tasks, // In contrast to tasks, which have can multiple concurrently running child tasks,
// spans must nest and not cross the goroutine boundary. // spans must nest and not cross the goroutine boundary.
// //
// ctx, endSpan = WithSpan(ctx, "copy-dir") // ctx, endSpan = WithSpan(ctx, "copy-dir")
// defer endSpan() // defer endSpan()
// for _, f := range dir.Files() { // for _, f := range dir.Files() {
// func() { // func() {
// ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f)) // ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f))
// defer endspan() // defer endspan()
// b, _ := ioutil.ReadFile(f) // b, _ := ioutil.ReadFile(f)
// _ = ioutil.WriteFile(f + ".copy", b, 0600) // _ = ioutil.WriteFile(f + ".copy", b, 0600)
// }() // }()
// } // }
// //
// In combination: // In combination:
// ctx, endTask = WithTask(ctx, "copy-dirs") //
// defer endTask() // ctx, endTask = WithTask(ctx, "copy-dirs")
// for i := range dirs { // defer endTask()
// go func(dir string) { // for i := range dirs {
// ctx, endTask := WithTask(ctx, "copy-dir") // go func(dir string) {
// defer endTask() // ctx, endTask := WithTask(ctx, "copy-dir")
// for _, f := range filesIn(dir) { // defer endTask()
// func() { // for _, f := range filesIn(dir) {
// ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f)) // func() {
// defer endspan() // ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f))
// b, _ := ioutil.ReadFile(f) // defer endspan()
// _ = ioutil.WriteFile(f + ".copy", b, 0600) // b, _ := ioutil.ReadFile(f)
// }() // _ = ioutil.WriteFile(f + ".copy", b, 0600)
// } // }()
// }() // }
// } // }()
// }
// //
// Note that a span ends at the time you call endSpan - not before and not after that. // Note that a span ends at the time you call endSpan - not before and not after that.
// If you violate the stack-like nesting of spans by forgetting an endSpan() invocation, // If you violate the stack-like nesting of spans by forgetting an endSpan() invocation,
@@ -65,8 +64,7 @@
// //
// Recovering from endSpan() or endTask() panics will corrupt the trace stack and lead to corrupt tracefile output. // Recovering from endSpan() or endTask() panics will corrupt the trace stack and lead to corrupt tracefile output.
// //
// // # Best Practices For Naming Tasks And Spans
// Best Practices For Naming Tasks And Spans
// //
// Tasks should always have string constants as names, and must not contain the `#` character. WHy? // Tasks should always have string constants as names, and must not contain the `#` character. WHy?
// First, the visualization by chrome://tracing draws a horizontal bar for each task in the trace. // First, the visualization by chrome://tracing draws a horizontal bar for each task in the trace.
@@ -74,8 +72,7 @@
// Note that the `#NUM` suffix will be reused if a task has ended, in order to avoid an // Note that the `#NUM` suffix will be reused if a task has ended, in order to avoid an
// infinite number of horizontal bars in the visualization. // infinite number of horizontal bars in the visualization.
// //
// // # Chrome-compatible Tracefile Support
// Chrome-compatible Tracefile Support
// //
// The activity trace generated by usage of WithTask and WithSpan can be rendered to a JSON output file // The activity trace generated by usage of WithTask and WithSpan can be rendered to a JSON output file
// that can be loaded into chrome://tracing . // that can be loaded into chrome://tracing .
@@ -11,8 +11,6 @@ import (
// use like this: // use like this:
// //
// defer WithSpanFromStackUpdateCtx(&existingCtx)() // defer WithSpanFromStackUpdateCtx(&existingCtx)()
//
//
func WithSpanFromStackUpdateCtx(ctx *context.Context) DoneFunc { func WithSpanFromStackUpdateCtx(ctx *context.Context) DoneFunc {
childSpanCtx, end := WithSpan(*ctx, getMyCallerOrPanic()) childSpanCtx, end := WithSpan(*ctx, getMyCallerOrPanic())
*ctx = childSpanCtx *ctx = childSpanCtx
+1 -1
View File
@@ -439,7 +439,7 @@ type FSMap interface { // FIXME unused
} }
// NOTE: when adding members to this struct, remember // NOTE: when adding members to this struct, remember
// to add them to `ReceiverConfig.copyIn()` // to add them to `ReceiverConfig.copyIn()`
type ReceiverConfig struct { type ReceiverConfig struct {
JobID JobID JobID JobID
+1 -1
View File
@@ -168,7 +168,7 @@ func (s AbstractionTypeSet) String() string {
for i := range s { for i := range s {
sts = append(sts, string(i)) sts = append(sts, string(i))
} }
sts = sort.StringSlice(sts) sort.Strings(sts)
return strings.Join(sts, ",") return strings.Join(sts, ",")
} }
@@ -23,7 +23,7 @@ func replicationCursorBookmarkNameImpl(fs string, guid uint64, jobid string) (st
var ErrV1ReplicationCursor = fmt.Errorf("bookmark name is a v1-replication cursor") var ErrV1ReplicationCursor = fmt.Errorf("bookmark name is a v1-replication cursor")
//err != nil always means that the bookmark is not a valid replication bookmark // err != nil always means that the bookmark is not a valid replication bookmark
// //
// Returns ErrV1ReplicationCursor as error if the bookmark is a v1 replication cursor // Returns ErrV1ReplicationCursor as error if the bookmark is a v1 replication cursor
func ParseReplicationCursorBookmarkName(fullname string) (uint64, JobID, error) { func ParseReplicationCursorBookmarkName(fullname string) (uint64, JobID, error) {
+1 -1
View File
@@ -797,7 +797,7 @@ func (a *attempt) errorReport() *errorReport {
r.byClass[class] = errs r.byClass[class] = errs
} }
for _, err := range r.flattened { for _, err := range r.flattened {
if neterr, ok := err.Err.(net.Error); ok && neterr.Temporary() { if neterr, ok := err.Err.(net.Error); ok && neterr.Timeout() {
putClass(err, errorClassTemporaryConnectivityRelated) putClass(err, errorClassTemporaryConnectivityRelated)
continue continue
} }
@@ -13,7 +13,7 @@ func init() {
} }
} }
//nolint[:deadcode,unused] //nolint:deadcode,unused
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "repl: driver: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "repl: driver: %s\n", fmt.Sprintf(format, args...))
@@ -22,7 +22,7 @@ func debug(format string, args ...interface{}) {
type debugFunc func(format string, args ...interface{}) type debugFunc func(format string, args ...interface{})
//nolint[:deadcode,unused] //nolint:deadcode,unused
func debugPrefix(prefixFormat string, prefixFormatArgs ...interface{}) debugFunc { func debugPrefix(prefixFormat string, prefixFormatArgs ...interface{}) debugFunc {
prefix := fmt.Sprintf(prefixFormat, prefixFormatArgs...) prefix := fmt.Sprintf(prefixFormat, prefixFormatArgs...)
return func(format string, args ...interface{}) { return func(format string, args ...interface{}) {
+1 -1
View File
@@ -14,7 +14,7 @@ func init() {
} }
} }
//nolint[:deadcode,unused] //nolint:deadcode,unused
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "rpc/dataconn: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "rpc/dataconn: %s\n", fmt.Sprintf(format, args...))
@@ -24,6 +24,9 @@ func (e HeartbeatTimeout) Error() string {
return "heartbeat timeout" return "heartbeat timeout"
} }
// This function is deprecated in net.Error and since this
// function is not involved in .Accept() code path, nothing
// really needs this method to be here.
func (e HeartbeatTimeout) Temporary() bool { return true } func (e HeartbeatTimeout) Temporary() bool { return true }
func (e HeartbeatTimeout) Timeout() bool { return true } func (e HeartbeatTimeout) Timeout() bool { return true }
@@ -13,7 +13,7 @@ func init() {
} }
} }
//nolint[:deadcode,unused] //nolint:deadcode,unused
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "rpc/dataconn/heartbeatconn: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "rpc/dataconn/heartbeatconn: %s\n", fmt.Sprintf(format, args...))
@@ -6,12 +6,12 @@
// In commit 082335df5d85e1b0b9faa35ff182c71886142d3e and earlier, heartbeatconn would fail // In commit 082335df5d85e1b0b9faa35ff182c71886142d3e and earlier, heartbeatconn would fail
// this benchmark with a writev I/O timeout (here the ss(8) output at the time of failure) // this benchmark with a writev I/O timeout (here the ss(8) output at the time of failure)
// //
// ESTAB 33369 0 127.0.0.1:12345 127.0.0.1:57282 users:(("heartbeatconn_i",pid=25953,fd=5)) // ESTAB 33369 0 127.0.0.1:12345 127.0.0.1:57282 users:(("heartbeatconn_i",pid=25953,fd=5))
// cubic wscale:7,7 rto:203 rtt:2.992/5.849 ato:162 mss:32768 pmtu:65535 rcvmss:32741 advmss:65483 cwnd:10 bytes_sent:48 bytes_acked:48 bytes_received:195401 segs_out:44 segs_in:57 data_segs_out:6 data_segs_in:34 send 876.1Mbps lastsnd:125 lastrcv:9390 lastack:125 pacing_rate 1752.0Mbps delivery_rate 6393.8Mbps delivered:7 app_limited busy:42ms rcv_rtt:1 rcv_space:65483 rcv_ssthresh:65483 minrtt:0.029 // cubic wscale:7,7 rto:203 rtt:2.992/5.849 ato:162 mss:32768 pmtu:65535 rcvmss:32741 advmss:65483 cwnd:10 bytes_sent:48 bytes_acked:48 bytes_received:195401 segs_out:44 segs_in:57 data_segs_out:6 data_segs_in:34 send 876.1Mbps lastsnd:125 lastrcv:9390 lastack:125 pacing_rate 1752.0Mbps delivery_rate 6393.8Mbps delivered:7 app_limited busy:42ms rcv_rtt:1 rcv_space:65483 rcv_ssthresh:65483 minrtt:0.029
// -- // --
// ESTAB 0 3956805 127.0.0.1:57282 127.0.0.1:12345 users:(("heartbeatconn_i",pid=26100,fd=3)) // ESTAB 0 3956805 127.0.0.1:57282 127.0.0.1:12345 users:(("heartbeatconn_i",pid=26100,fd=3))
// cubic wscale:7,7 rto:211 backoff:5 rtt:10.38/16.937 ato:40 mss:32768 pmtu:65535 rcvmss:536 advmss:65483 cwnd:10 bytes_sent:195401 bytes_acked:195402 bytes_received:48 segs_out:57 segs_in:45 data_segs_out:34 data_segs_in:6 send 252.5Mbps lastsnd:9390 lastrcv:125 lastack:125 pacing_rate 505.1Mbps delivery_rate 1971.0Mbps delivered:35 busy:30127ms rwnd_limited:30086ms(99.9%) rcv_space:65495 rcv_ssthresh:65495 notsent:3956805 minrtt:0.007 // cubic wscale:7,7 rto:211 backoff:5 rtt:10.38/16.937 ato:40 mss:32768 pmtu:65535 rcvmss:536 advmss:65483 cwnd:10 bytes_sent:195401 bytes_acked:195402 bytes_received:48 segs_out:57 segs_in:45 data_segs_out:34 data_segs_in:6 send 252.5Mbps lastsnd:9390 lastrcv:125 lastack:125 pacing_rate 505.1Mbps delivery_rate 1971.0Mbps delivered:35 busy:30127ms rwnd_limited:30086ms(99.9%) rcv_space:65495 rcv_ssthresh:65495 notsent:3956805 minrtt:0.007
// panic: writev tcp 127.0.0.1:57282->127.0.0.1:12345: i/o timeout // panic: writev tcp 127.0.0.1:57282->127.0.0.1:12345: i/o timeout
// //
// The assumed reason for those writev timeouts is the following: // The assumed reason for those writev timeouts is the following:
// - Sporadic server stalls (sever data handling, usually I/O) cause TCP exponential backoff on the client for client->server // - Sporadic server stalls (sever data handling, usually I/O) cause TCP exponential backoff on the client for client->server
@@ -22,27 +22,23 @@
// The fix contained in the commit this message was committed with resets the deadline whenever // The fix contained in the commit this message was committed with resets the deadline whenever
// a heartbeat is received from the server. // a heartbeat is received from the server.
// //
//
// How to run this integration test: // How to run this integration test:
// //
// Terminal 1:
// $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go -mode server -addr 127.0.0.1:12345
// rpc/dataconn/heartbeatconn: send heartbeat
// rpc/dataconn/heartbeatconn: send heartbeat
// ...
// //
// Terminal 1: // Terminal 2:
// $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go -mode server -addr 127.0.0.1:12345 // $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go -mode client -addr 127.0.0.1:12345
// rpc/dataconn/heartbeatconn: send heartbeat // rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
// rpc/dataconn/heartbeatconn: send heartbeat // rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
// ... // rpc/dataconn/heartbeatconn: send heartbeat
// // rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
// Terminal 2: // rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
// $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go -mode client -addr 127.0.0.1:12345 // rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
// rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout // ...
// rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
// rpc/dataconn/heartbeatconn: send heartbeat
// rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
// rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
// rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
// ...
//
// You should observe
package main package main
import ( import (
@@ -2,15 +2,13 @@
// //
// With stdin / stdout on client and server, simulating zfs send|recv piping // With stdin / stdout on client and server, simulating zfs send|recv piping
// //
// ./microbenchmark -appmode server | pv -r > /dev/null // ./microbenchmark -appmode server | pv -r > /dev/null
// ./microbenchmark -appmode client -direction recv < /dev/zero // ./microbenchmark -appmode client -direction recv < /dev/zero
//
// //
// Without the overhead of pipes (just protocol performance, mostly useful with perf bc no bw measurement) // Without the overhead of pipes (just protocol performance, mostly useful with perf bc no bw measurement)
// //
// ./microbenchmark -appmode client -direction recv -devnoopWriter -devnoopReader // ./microbenchmark -appmode client -direction recv -devnoopWriter -devnoopReader
// ./microbenchmark -appmode server -devnoopReader -devnoopWriter // ./microbenchmark -appmode server -devnoopReader -devnoopWriter
//
package main package main
import ( import (
+7 -11
View File
@@ -29,7 +29,7 @@ func WithLogger(ctx context.Context, log Logger) context.Context {
return context.WithValue(ctx, contextKeyLogger, log) return context.WithValue(ctx, contextKeyLogger, log)
} }
//nolint[:deadcode,unused] //nolint:deadcode,unused
func getLog(ctx context.Context) Logger { func getLog(ctx context.Context) Logger {
log, ok := ctx.Value(contextKeyLogger).(Logger) log, ok := ctx.Value(contextKeyLogger).(Logger)
if !ok { if !ok {
@@ -181,23 +181,19 @@ func (e *ReadStreamError) Error() string {
var _ net.Error = &ReadStreamError{} var _ net.Error = &ReadStreamError{}
func (e ReadStreamError) netErr() net.Error {
if netErr, ok := e.Err.(net.Error); ok {
return netErr
}
return nil
}
func (e ReadStreamError) Timeout() bool { func (e ReadStreamError) Timeout() bool {
if netErr := e.netErr(); netErr != nil { if netErr, ok := e.Err.(net.Error); ok {
return netErr.Timeout() return netErr.Timeout()
} }
return false return false
} }
// This function is deprecated in net.Error and since this
// function is not involved in .Accept() code path, nothing
// really needs this method to be here.
func (e ReadStreamError) Temporary() bool { func (e ReadStreamError) Temporary() bool {
if netErr := e.netErr(); netErr != nil { if te, ok := e.Err.(interface{ Temporary() bool }); ok {
return netErr.Temporary() return te.Temporary()
} }
return false return false
} }
+6 -1
View File
@@ -232,7 +232,12 @@ var _ net.Error = (*closeStateErrConnectionClosed)(nil)
func (e *closeStateErrConnectionClosed) Error() string { func (e *closeStateErrConnectionClosed) Error() string {
return "connection closed" return "connection closed"
} }
func (e *closeStateErrConnectionClosed) Timeout() bool { return false }
func (e *closeStateErrConnectionClosed) Timeout() bool { return false }
// This function is deprecated in net.Error and since this
// function is not involved in .Accept() code path, nothing
// really needs this method to be here.
func (e *closeStateErrConnectionClosed) Temporary() bool { return false } func (e *closeStateErrConnectionClosed) Temporary() bool { return false }
func (s *closeState) CloseEntry() error { func (s *closeState) CloseEntry() error {
+1 -1
View File
@@ -13,7 +13,7 @@ func init() {
} }
} }
//nolint[:deadcode,unused] //nolint:deadcode,unused
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "rpc/dataconn/stream: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "rpc/dataconn/stream: %s\n", fmt.Sprintf(format, args...))
@@ -1,25 +1,5 @@
// Package netadaptor implements an adaptor from // Package netadaptor implements an adaptor from
// transport.AuthenticatedListener to net.Listener. // transport.AuthenticatedListener to net.Listener.
//
// In contrast to transport.AuthenticatedListener,
// net.Listener is commonly expected (e.g. by net/http.Server.Serve),
// to return errors that fulfill the Temporary interface:
// interface Temporary { Temporary() bool }
// Common behavior of packages consuming net.Listener is to return
// from the serve-loop if an error returned by Accept is not Temporary,
// i.e., does not implement the interface or is !net.Error.Temporary().
//
// The zrepl transport infrastructure was written with the
// idea that Accept() may return any kind of error, and the consumer
// would just log the error and continue calling Accept().
// We have to adapt these listeners' behavior to the expectations
// of net/http.Server.
//
// Hence, Listener does not return an error at all but blocks the
// caller of Accept() until we get a (successfully authenticated)
// connection without errors from the transport.
// Accept errors returned from the transport are logged as errors
// to the logger passed on initialization.
package netadaptor package netadaptor
import ( import (
+2 -2
View File
@@ -188,8 +188,8 @@ func (c *Client) WaitForConnectivity(ctx context.Context) error {
data, dataErr := c.dataClient.ReqPing(ctx, &req) data, dataErr := c.dataClient.ReqPing(ctx, &req)
// dataClient uses transport.Connecter, which doesn't expose WaitForReady(true) // dataClient uses transport.Connecter, which doesn't expose WaitForReady(true)
// => we need to mask dial timeouts // => we need to mask dial timeouts
if err, ok := dataErr.(interface{ Temporary() bool }); ok && err.Temporary() { if err, ok := dataErr.(interface{ Timeout() bool }); ok && err.Timeout() {
// Rate-limit pings here in case Temporary() is a mis-classification // Rate-limit pings here in case Timeout() == true is a mis-classification
// or returns immediately (this is a tight loop in that case) // or returns immediately (this is a tight loop in that case)
// TODO keep this in lockstep with controlClient // TODO keep this in lockstep with controlClient
// => don't use FailFast for control, but check that both control and data worked // => don't use FailFast for control, but check that both control and data worked
+1 -1
View File
@@ -14,7 +14,7 @@ func init() {
} }
} }
//nolint[:deadcode,unused] //nolint:deadcode,unused
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "rpc: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "rpc: %s\n", fmt.Sprintf(format, args...))
+62 -64
View File
@@ -3,7 +3,7 @@
// The zrepl documentation refers to the client as the // The zrepl documentation refers to the client as the
// `active side` and to the server as the `passive side`. // `active side` and to the server as the `passive side`.
// //
// Design Considerations // # Design Considerations
// //
// zrepl has non-standard requirements to remote procedure calls (RPC): // zrepl has non-standard requirements to remote procedure calls (RPC):
// whereas the coordination of replication (the planning phase) mostly // whereas the coordination of replication (the planning phase) mostly
@@ -35,7 +35,7 @@
// //
// Hence, this package attempts to combine the best of both worlds: // Hence, this package attempts to combine the best of both worlds:
// //
// GRPC for Coordination and Dataconn for Bulk Data Transfer // # GRPC for Coordination and Dataconn for Bulk Data Transfer
// //
// This package's Client uses its transport.Connecter to maintain // This package's Client uses its transport.Connecter to maintain
// separate control and data connections to the Server. // separate control and data connections to the Server.
@@ -47,68 +47,66 @@
// The following ASCII diagram gives an overview of how the individual // The following ASCII diagram gives an overview of how the individual
// building blocks are glued together: // building blocks are glued together:
// //
// +------------+ // +------------+
// | rpc.Client | // | rpc.Client |
// +------------+ // +------------+
// | | // | |
// +--------+ +------------+ // +--------+ +------------+
// | | // | |
// +---------v-----------+ +--------v------+ // +---------v-----------+ +--------v------+
// |pdu.ReplicationClient| |dataconn.Client| // |pdu.ReplicationClient| |dataconn.Client|
// +---------------------+ +--------v------+ // +---------------------+ +--------v------+
// | label: label: | // | label: label: |
// | zrepl_control zrepl_data | // | zrepl_control zrepl_data |
// +--------+ +------------+ // +--------+ +------------+
// | | // | |
// +--v---------v---+ // +--v---------v---+
// | transportmux | // | transportmux |
// +-------+--------+ // +-------+--------+
// | uses // | uses
// +-------v--------+ // +-------v--------+
// |versionhandshake| // |versionhandshake|
// +-------+--------+ // +-------+--------+
// | uses // | uses
// +------v------+ // +------v------+
// | transport | // | transport |
// +------+------+ // +------+------+
// | // |
// NETWORK // NETWORK
// | // |
// +------+------+ // +------+------+
// | transport | // | transport |
// +------^------+ // +------^------+
// | uses // | uses
// +-------+--------+ // +-------+--------+
// |versionhandshake| // |versionhandshake|
// +-------^--------+ // +-------^--------+
// | uses // | uses
// +-------+--------+ // +-------+--------+
// | transportmux | // | transportmux |
// +--^--------^----+ // +--^--------^----+
// | | // | |
// +--------+ --------------+ --- // +--------+ --------------+ ---
// | | | // | | |
// | label: label: | | // | label: label: | |
// | zrepl_control zrepl_data | | // | zrepl_control zrepl_data | |
// +-----+----+ +-----------+---+ | // +-----+----+ +-----------+---+ |
// |netadaptor| |dataconn.Server| | rpc.Server // |netadaptor| |dataconn.Server| | rpc.Server
// | + | +------+--------+ | // | + | +------+--------+ |
// |grpcclient| | | // |grpcclient| | |
// |identity | | | // |identity | | |
// +-----+----+ | | // +-----+----+ | |
// | | | // | | |
// +---------v-----------+ | | // +---------v-----------+ | |
// |pdu.ReplicationServer| | | // |pdu.ReplicationServer| | |
// +---------+-----------+ | | // +---------+-----------+ | |
// | | --- // | | ---
// +----------+ +------------+ // +----------+ +------------+
// | | // | |
// +---v--v-----+ // +---v--v-----+
// | Handler | // | Handler |
// +------------+ // +------------+
// (usually endpoint.{Sender,Receiver}) // (usually endpoint.{Sender,Receiver})
//
//
package rpc package rpc
// edit trick for the ASCII art above: // edit trick for the ASCII art above:
+1 -1
View File
@@ -9,7 +9,7 @@ import (
type Logger = logger.Logger type Logger = logger.Logger
/// All fields must be non-nil // All fields must be non-nil
type Loggers struct { type Loggers struct {
General Logger General Logger
Control Logger Control Logger
+41 -2
View File
@@ -33,8 +33,47 @@ var _ net.Error = &HandshakeError{}
func (e HandshakeError) Error() string { return e.msg } func (e HandshakeError) Error() string { return e.msg }
// Like with net.OpErr (Go issue 6163), a client failing to handshake // When a net.Listener.Accept() returns an error, the server must
// should be a temporary Accept error toward the Listener . // decide whether to retry calling Accept() or not.
// On some platforms (e.g., Linux), Accept() can return errors
// related to the specific protocol connection that was supposed
// to be returned as asocket FD. Obviously, we want to ignore,
// maybe log, those errors and retry Accept() immediately to
// serve other connections.
// But there are also conditions where we get Accept() errors because
// the process has run out of file descriptors. In that case, retrying
// won't help. We need to close some file descriptor to make progress.
// Note that there could be lots of open file descriptors because we
// have accepted, and not yet closed, lots of connections in the past.
// And then, of course there can be errors where we just want
// to return, e.g., if there's a programming error and we're getting
// an EBADFD or whatever.
//
// So, the serve loops in net/http.Server.Serve() or gRPC's server.Serve()
// must inspect the error and decide what to do.
// The vehicle for this is the
//
// interface { Temporary() bool }
//
// Behavior in both of the aforementioned Serve() loops:
//
// - if the error doesn't implement the interface, stop serving and return
// - `Temporary() == true`: retry with back-off
// - `Temporary() == false`: stop serving and return
//
// So, to make this package's HandshakeListener work with these
// Serve() loops, we return Temporary() == true if the handshake fails.
// In the aforementioned categories, that's the case of a per-connection
// protocol error.
//
// Note: the net.Error interface has deprecated the Temporary() method
// in go.dev/issue/45729, but there is no replacement for users of .Accept().
// Existing users of .Accept() continue to check for the interface.
// So, we need to continue supporting Temporary() until there's a different
// mechanism for serve loops to decide whether to retry or not.
// The following mailing list post proposes to eliminate the retries
// completely, but it seems like the effort has stalled.
// https://groups.google.com/g/golang-nuts/c/-JcZzOkyqYI/m/xwaZzjCgAwAJ
func (e HandshakeError) Temporary() bool { func (e HandshakeError) Temporary() bool {
if e.isAcceptError { if e.isAcceptError {
return true return true
+6 -7
View File
@@ -3,13 +3,12 @@
// //
// Intended Usage // Intended Usage
// //
// defer s.lock().unlock() // defer s.lock().unlock()
// // drop lock while waiting for wait group // // drop lock while waiting for wait group
// func() { // func() {
// defer a.l.Unlock().Lock() // defer a.l.Unlock().Lock()
// fssesDone.Wait() // fssesDone.Wait()
// }() // }()
//
package chainlock package chainlock
import "sync" import "sync"
+11 -12
View File
@@ -7,26 +7,25 @@
// //
// Example: // Example:
// //
// type Config struct { // type Config struct {
// // This field must be set to a non-nil value, // // This field must be set to a non-nil value,
// // forcing the caller to make their mind up // // forcing the caller to make their mind up
// // about this field. // // about this field.
// CriticalSetting *nodefault.Bool // CriticalSetting *nodefault.Bool
// } // }
// //
// An function that takes such a Config should _not_ check for nil-ness: // An function that takes such a Config should _not_ check for nil-ness:
// and instead unconditionally dereference: // and instead unconditionally dereference:
// //
// func f(c Config) { // func f(c Config) {
// if (c.CriticalSetting) { } // if (c.CriticalSetting) { }
// } // }
// //
// If the caller of f forgot to specify the .CriticalSetting // If the caller of f forgot to specify the .CriticalSetting
// field, the Go runtime will issue a nil-pointer deref panic // field, the Go runtime will issue a nil-pointer deref panic
// and it'll be clear that the caller did not read the docs of Config. // and it'll be clear that the caller did not read the docs of Config.
// //
// f(Config{}) // crashes // f(Config{}) // crashes
//
// f Config{ CriticalSetting: &nodefault.Bool{B: false}} // doesn't crash
// //
// f Config{ CriticalSetting: &nodefault.Bool{B: false}} // doesn't crash
package nodefault package nodefault
+2 -3
View File
@@ -53,7 +53,6 @@ var componentValidChar = regexp.MustCompile(`^[0-9a-zA-Z-_\.: ]+$`)
// characters: // characters:
// //
// [-_.: ] // [-_.: ]
//
func ComponentNamecheck(datasetPathComponent string) error { func ComponentNamecheck(datasetPathComponent string) error {
if len(datasetPathComponent) == 0 { if len(datasetPathComponent) == 0 {
return fmt.Errorf("path component must not be empty") return fmt.Errorf("path component must not be empty")
@@ -91,8 +90,8 @@ func (e *PathValidationError) Error() string {
// combines // combines
// //
// lib/libzfs/libzfs_dataset.c: zfs_validate_name // lib/libzfs/libzfs_dataset.c: zfs_validate_name
// module/zcommon/zfs_namecheck.c: entity_namecheck // module/zcommon/zfs_namecheck.c: entity_namecheck
// //
// The '%' character is not allowed because it's reserved for zfs-internal use // The '%' character is not allowed because it's reserved for zfs-internal use
func EntityNamecheck(path string, t EntityType) (err *PathValidationError) { func EntityNamecheck(path string, t EntityType) (err *PathValidationError) {
+4 -3
View File
@@ -1022,8 +1022,10 @@ func (s *DrySendInfo) unmarshalZFSOutput(output []byte) (err error) {
} }
// unmarshal info line, looks like this: // unmarshal info line, looks like this:
// full zroot/test/a@1 5389768 //
// incremental zroot/test/a@1 zroot/test/a@2 5383936 // full zroot/test/a@1 5389768
// incremental zroot/test/a@1 zroot/test/a@2 5383936
//
// => see test cases // => see test cases
func (s *DrySendInfo) unmarshalInfoLine(l string) (regexMatched bool, err error) { func (s *DrySendInfo) unmarshalInfoLine(l string) (regexMatched bool, err error) {
@@ -1855,7 +1857,6 @@ var ErrBookmarkCloningNotSupported = fmt.Errorf("bookmark cloning feature is not
// unless a bookmark with the name `bookmark` exists and has the same idenitty (zfs.FilesystemVersionEqualIdentity) // unless a bookmark with the name `bookmark` exists and has the same idenitty (zfs.FilesystemVersionEqualIdentity)
// //
// v must be validated by the caller // v must be validated by the caller
//
func ZFSBookmark(ctx context.Context, fs string, v FilesystemVersion, bookmark string) (bm FilesystemVersion, err error) { func ZFSBookmark(ctx context.Context, fs string, v FilesystemVersion, bookmark string) (bm FilesystemVersion, err error) {
bm = FilesystemVersion{ bm = FilesystemVersion{
+1 -1
View File
@@ -13,7 +13,7 @@ func init() {
} }
} }
//nolint[:deadcode,unused] //nolint:deadcode,unused
func debug(format string, args ...interface{}) { func debug(format string, args ...interface{}) {
if debugEnabled { if debugEnabled {
fmt.Fprintf(os.Stderr, "zfs: %s\n", fmt.Sprintf(format, args...)) fmt.Fprintf(os.Stderr, "zfs: %s\n", fmt.Sprintf(format, args...))