diff --git a/docs/changelog.rst b/docs/changelog.rst index 88fb820..f08b951 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -19,6 +19,8 @@ Developers should consult the git commit log or GitHub issue tracker. * Make sure to understand the meaning bookmarks have for :ref:`maximum replication downtime `. * Example: :sampleconf:`pullbackup/productionhost.yml` +* |bugfix| :issue:`61`: fix excessive memory usage + 0.0.2 ----- diff --git a/zfs/mapping.go b/zfs/mapping.go index ca80318..4ccfc70 100644 --- a/zfs/mapping.go +++ b/zfs/mapping.go @@ -1,6 +1,9 @@ package zfs -import "fmt" +import ( + "context" + "fmt" +) type DatasetFilter interface { Filter(p *DatasetPath) (pass bool, err error) @@ -12,15 +15,21 @@ func ZFSListMapping(filter DatasetFilter) (datasets []*DatasetPath, err error) { panic("filter must not be nil") } - var lines [][]string - lines, err = ZFSList([]string{"name"}, "-r", "-t", "filesystem,volume") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + rchan := make(chan ZFSListResult) + go ZFSListChan(ctx, rchan, []string{"name"}, "-r", "-t", "filesystem,volume") - datasets = make([]*DatasetPath, 0, len(lines)) + datasets = make([]*DatasetPath, 0) + for r := range rchan { - for _, line := range lines { + if r.err != nil { + err = r.err + return + } var path *DatasetPath - if path, err = NewDatasetPath(line[0]); err != nil { + if path, err = NewDatasetPath(r.fields[0]); err != nil { return } diff --git a/zfs/versions.go b/zfs/versions.go index fb3efa7..643c74a 100644 --- a/zfs/versions.go +++ b/zfs/versions.go @@ -2,6 +2,7 @@ package zfs import ( "bytes" + "context" "errors" "fmt" "strconv" @@ -61,17 +62,23 @@ type FilesystemVersionFilter interface { } func ZFSListFilesystemVersions(fs *DatasetPath, filter FilesystemVersionFilter) (res []FilesystemVersion, err error) { - var fieldLines [][]string - fieldLines, err = ZFSList( + listResults := make(chan ZFSListResult) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go ZFSListChan(ctx, listResults, []string{"name", "guid", "createtxg", "creation"}, "-r", "-d", "1", "-t", "bookmark,snapshot", "-s", "createtxg", fs.ToString()) - if err != nil { - return - } - res = make([]FilesystemVersion, 0, len(fieldLines)) - for _, line := range fieldLines { + + res = make([]FilesystemVersion, 0) + for listResult := range listResults { + if listResult.err != nil { + return nil, listResult.err + } + + line := listResult.fields if len(line[0]) < 3 { err = errors.New(fmt.Sprintf("snapshot or bookmark name implausibly short: %s", line[0])) diff --git a/zfs/zfs.go b/zfs/zfs.go index 91a082f..f6a3b2b 100644 --- a/zfs/zfs.go +++ b/zfs/zfs.go @@ -10,6 +10,8 @@ import ( "os/exec" "strings" + "context" + "github.com/problame/go-rwccmd" "github.com/zrepl/zrepl/util" ) @@ -185,6 +187,68 @@ func ZFSList(properties []string, zfsArgs ...string) (res [][]string, err error) return } +type ZFSListResult struct { + fields []string + err error +} + +// ZFSListChan executes `zfs list` and sends the results to the `out` channel. +// The `out` channel is always closed by ZFSListChan: +// If an error occurs, it is closed after sending a result with the err field set. +// If no error occurs, it is just closed. +// If the operation is cancelled via context, the channel is just closed. +// +// However, if callers do not drain `out` or cancel via `ctx`, the process will leak either running because +// IO is pending or as a zombie. +func ZFSListChan(ctx context.Context, out chan ZFSListResult, properties []string, zfsArgs ...string) { + defer close(out) + + args := make([]string, 0, 4+len(zfsArgs)) + args = append(args, + "list", "-H", "-p", + "-o", strings.Join(properties, ",")) + args = append(args, zfsArgs...) + + sendResult := func(fields []string, err error) (done bool) { + select { + case <-ctx.Done(): + return true + case out <- ZFSListResult{fields, err}: + return false + } + } + + cmd, err := rwccmd.CommandContext(ctx, ZFS_BINARY, args, []string{}) + if err != nil { + sendResult(nil, err) + return + } + if err = cmd.Start(); err != nil { + sendResult(nil, err) + return + } + defer cmd.Close() + + s := bufio.NewScanner(cmd) + buf := make([]byte, 1024) // max line length + s.Buffer(buf, 0) + + for s.Scan() { + fields := strings.SplitN(s.Text(), "\t", len(properties)) + if len(fields) != len(properties) { + sendResult(nil, errors.New("unexpected output")) + return + } + if sendResult(fields, nil) { + return + } + } + if s.Err() != nil { + sendResult(nil, s.Err()) + } + return +} + func ZFSSend(fs *DatasetPath, from, to *FilesystemVersion) (stream io.Reader, err error) { args := make([]string, 0)