Repository: rclone/rclone
Pull request #3413 (2019-08-06 10:37:40)

backend/local: Avoid polluting page cache when uploading local files to remote backends
#### What is the purpose of this change?

This patch makes rclone keep linux page cache usage under control when
uploading local files to remote backends. When opening a file it issues
FADV_SEQUENTIAL to configure read ahead strategy. While reading
the file it issues FADV_DONTNEED every 128kB to free page cache from
already consumed pages.

The read pattern is the following, the 32768 buffer size comes from io.Copy function.

```
fadvise64(5, 0, 0, POSIX_FADV_SEQUENTIAL) = 0
read(5, "\324\375\251\376\213\361\240\224>\t5E\301\331X\274^\203oA\353\303.2'\206z\177N\27fB"..., 32768) = 32768
read(5, "\361\311\vW!\354_\317hf\276t\307\30L\351\272T\342C\243\370\240\213\355\210\v\221\201\177[\333"..., 32768) = 32768
read(5, ":\371\337Gn\355C\322\334 \253f\373\277\301;\215\n\240\347\305\6N\257\313\4\365\276ANq!"..., 32768) = 32768
read(5, "\312\243\360P\263\242\267H\304\240Y\310\367sT\321\256\6[b\310\224\361\344$Ms\234\5\314\306i"..., 32768) = 32768
fadvise64(5, 0, 131072, POSIX_FADV_DONTNEED) = 0
read(5, "m\251\7a\306\226\366-\v~\"\216\353\342~0\fht\315DK0\236.\\\201!A#\177\320"..., 32768) = 32768
read(5, "\7\324\207,\205\360\376\307\276\254\250\232\21G\323n\255\354\234\257P\322y\3502\37\246\21\334^42"..., 32768) = 32768
read(5, "e{*\225\223R\320\212EG:^\302\377\242\337\10\222J\16A\305\0\353\354\326P\336\357A|-"..., 32768) = 32768
read(5, "n\23XA4*R\352\234\257\364\355Y\204t9T\363\33\357\333\3674\246\221T\360\226\326G\354\374"..., 32768) = 32768
fadvise64(5, 131072, 131072, POSIX_FADV_DONTNEED) = 0
read(5, "SX\331\251}\24\353\37\310#\307|h%\372\34\310\3070YX\250s\2269\242\236\371\302z\357_"..., 32768) = 32768
read(5, "\177\3500\236Y\245\376NIY\177\360p!\337L]\2726\206@\240\246pG\213\254N\274\226\303\357"..., 32768) = 32768
read(5, "\242$*\364\217U\264]\221Y\245\342r\t\253\25Hr\363\263\364\336\322\t\325\325\f\37z\324\201\351"..., 32768) = 32768
read(5, "\2305\242\366\370\203tM\226<\230\25\316(9\25x\2\376\212\346Q\223 \353\225\323\264jf|\216"..., 32768) = 32768
fadvise64(5, 262144, 131072, POSIX_FADV_DONTNEED) = 0
```

Page cache consumption per file can be checked with tools like [pcstat](https://github.com/tobert/pcstat).

#### Was the change discussed in an issue or in the forum before?

No.

#### Checklist

- [x] I have read the [contribution guidelines](https://github.com/rclone/rclone/blob/master/CONTRIBUTING.md#submitting-a-pull-request).
- [x] I have added tests for all changes in this PR if appropriate.
- [x] I have added documentation for the changes if appropriate.
- [x] All commit messages are in [house style](https://github.com/rclone/rclone/blob/master/CONTRIBUTING.md#commit-messages).
- [x] I'm done, this Pull Request is ready for review :-)


mmatczuk: @ncw thanks for review.

- added comments in the code
- added test results in the commit message

With the patch the copy is actually faster, however I think I found a bug and opened issue https://github.com/rclone/rclone/issues/3419. 

Patch:
diff --git a/backend/local/aaaa b/backend/local/aaaa
new file mode 100644
index 000000000..e69de29bb
diff --git a/backend/local/fadvise_other.go b/backend/local/fadvise_other.go
new file mode 100644
index 000000000..18ac0d750
--- /dev/null
+++ b/backend/local/fadvise_other.go
@@ -0,0 +1,12 @@
+//+build !linux
+
+package local
+
+import (
+	"io"
+	"os"
+)
+
+func newFadviseReadCloser(o *Object, f *os.File, offset, limit int64) io.ReadCloser {
+	return f
+}
diff --git a/backend/local/fadvise_unix.go b/backend/local/fadvise_unix.go
new file mode 100644
index 000000000..7a6f4bad5
--- /dev/null
+++ b/backend/local/fadvise_unix.go
@@ -0,0 +1,129 @@
+//+build linux
+
+package local
+
+import (
+	"io"
+	"os"
+
+	"github.com/rclone/rclone/fs"
+	"golang.org/x/sys/unix"
+)
+
+// fadvise provides means to automate freeing pages in kernel page cache for
+// a given file descriptor as the file is sequentially processed (read or
+// written).
+//
+// When copying a file to a remote backend all the file content is read by
+// kernel and put to page cache to make future reads faster.
+// This causes memory pressure visible in both memory usage and CPU consumption
+// and can even cause OOM errors in applications consuming large amounts memory.
+//
+// In case of an upload to a remote backend, there is no benefits from caching.
+//
+// fadvise would orchestrate calling POSIX_FADV_DONTNEED
+//
+// POSIX_FADV_DONTNEED attempts to free cached pages associated
+// with the specified region.  This is useful, for example, while
+// streaming large files.  A program may periodically request the
+// kernel to free cached data that has already been used, so that
+// more useful cached pages are not discarded instead.
+//
+// Requests to discard partial pages are ignored.  It is
+// preferable to preserve needed data than discard unneeded data.
+// If the application requires that data be considered for
+// discarding, then offset and len must be page-aligned.
+//
+// The implementation may attempt to write back dirty pages in
+// the specified region, but this is not guaranteed.  Any
+// unwritten dirty pages will not be freed.  If the application
+// wishes to ensure that dirty pages will be released, it should
+// call fsync(2) or fdatasync(2) first.
+type fadvise struct {
+	o          *Object
+	fd         int
+	lastPos    int64
+	curPos     int64
+	windowSize int64
+}
+
+// sequential configures readahead strategy in Linux kernel.
+//
+// Under Linux, POSIX_FADV_NORMAL sets the readahead window to the
+// default size for the backing device; POSIX_FADV_SEQUENTIAL doubles
+// this size, and POSIX_FADV_RANDOM disables file readahead entirely.
+func (f *fadvise) sequential(limit int64) bool {
+	l := int64(0)
+	if limit > 0 {
+		l = limit
+	}
+	if err := unix.Fadvise(f.fd, f.curPos, l, unix.FADV_SEQUENTIAL); err != nil {
+		fs.Debugf(f.o, "fadvise sequential failed on file descriptor %d: %s", f.fd, err)
+		return false
+	}
+
+	return true
+}
+
+func (f *fadvise) next(n int) {
+	f.curPos += int64(n)
+	f.freePagesIfNeeded()
+}
+
+func (f *fadvise) freePagesIfNeeded() {
+	if f.curPos >= f.lastPos+f.windowSize {
+		f.freePages()
+	}
+}
+
+func (f *fadvise) freePages() {
+	if err := unix.Fadvise(f.fd, f.lastPos, f.curPos-f.lastPos, unix.FADV_DONTNEED); err != nil {
+		fs.Debugf(f.o, "fadvise dontneed failed on file descriptor %d: %s", f.fd, err)
+	}
+	f.lastPos = f.curPos
+}
+
+type fadviseReadCloser struct {
+	*fadvise
+	inner io.ReadCloser
+}
+
+var defaultWindowSize = int64(32 * os.Getpagesize())
+
+// newFadviseReadCloser wraps os.File so that reading from that file would
+// remove already consumed pages from kernel page cache.
+// In addition to that it instructs kernel to double the readahead window to
+// make sequential reads faster.
+// See also fadvise.
+func newFadviseReadCloser(o *Object, f *os.File, offset, limit int64) io.ReadCloser {
+	r := fadviseReadCloser{
+		fadvise: &fadvise{
+			o:          o,
+			fd:         int(f.Fd()),
+			lastPos:    offset,
+			curPos:     offset,
+			windowSize: defaultWindowSize,
+		},
+		inner: f,
+	}
+
+	// If syscall failed it's likely that the subsequent syscalls to that
+	// file descriptor would also fail. In that case return the provided os.File
+	// pointer.
+	if !r.sequential(limit) {
+		return f
+	}
+
+	return r
+}
+
+func (f fadviseReadCloser) Read(p []byte) (n int, err error) {
+	n, err = f.inner.Read(p)
+	f.next(n)
+	return
+}
+
+func (f fadviseReadCloser) Close() error {
+	f.freePages()
+	return f.inner.Close()
+}
diff --git a/backend/local/local.go b/backend/local/local.go
index 12c76985f..f9e861977 100644
--- a/backend/local/local.go
+++ b/backend/local/local.go
@@ -777,7 +777,11 @@ func (o *Object) Hash(ctx context.Context, r hash.Type) (string, error) {
 		var in io.ReadCloser
 
 		if !o.translatedLink {
-			in, err = file.Open(o.path)
+			var fd *os.File
+			fd, err = file.Open(o.path)
+			if fd != nil {
+				in = newFadviseReadCloser(o, fd, 0, 0)
+			}
 		} else {
 			in, err = o.openTranslatedLink(0, -1)
 		}
@@ -938,7 +942,7 @@ func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (in io.Read
 	if err != nil {
 		return
 	}
-	wrappedFd := readers.NewLimitedReadCloser(fd, limit)
+	wrappedFd := readers.NewLimitedReadCloser(newFadviseReadCloser(o, fd, offset, limit), limit)
 	if offset != 0 {
 		// seek the object
 		_, err = fd.Seek(offset, io.SeekStart)
diff --git a/backend/swift/swift.go b/backend/swift/swift.go
index b0431aeba..7c1cfbe64 100644
--- a/backend/swift/swift.go
+++ b/backend/swift/swift.go
@@ -25,6 +25,7 @@ import (
 	"github.com/rclone/rclone/fs/operations"
 	"github.com/rclone/rclone/fs/walk"
 	"github.com/rclone/rclone/lib/pacer"
+	"github.com/rclone/rclone/lib/readers"
 )
 
 // Constants
@@ -1224,8 +1225,13 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 		}
 		o.headers = nil // wipe old metadata
 	} else {
+		var inCount *readers.CountingReader
 		if size >= 0 {
 			headers["Content-Length"] = strconv.FormatInt(size, 10) // set Content-Length if we know it
+		} else {
+			// otherwise count the size for later
+			inCount = readers.NewCountingReader(in)
+			in = inCount
 		}
 		var rxHeaders swift.Headers
 		err = o.fs.pacer.CallNoRetry(func() (bool, error) {
@@ -1242,6 +1248,10 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 		o.md5 = rxHeaders["ETag"]
 		o.contentType = contentType
 		o.headers = headers
+		if inCount != nil {
+			// update the size if streaming from the reader
+			o.size = int64(inCount.BytesRead())
+		}
 	}
 
 	// If file was a dynamic large object then remove old/all segments
diff --git a/cmd/serve/sftp/connection.go b/cmd/serve/sftp/connection.go
index 51408a8b5..5d50963ae 100644
--- a/cmd/serve/sftp/connection.go
+++ b/cmd/serve/sftp/connection.go
@@ -167,7 +167,7 @@ func (c *conn) handleChannel(newChannel ssh.NewChannel) {
 	}
 	defer func() {
 		err := channel.Close()
-		if err != nil {
+		if err != nil && err != io.EOF {
 			fs.Debugf(c.what, "Failed to close channel: %v", err)
 		}
 	}()
@@ -218,7 +218,7 @@ func (c *conn) handleChannel(newChannel ssh.NewChannel) {
 		server := sftp.NewRequestServer(channel, c.handlers)
 		defer func() {
 			err := server.Close()
-			if err != nil {
+			if err != nil && err != io.EOF {
 				fs.Debugf(c.what, "Failed to close server: %v", err)
 			}
 		}()
diff --git a/cmd/serve/sftp/server.go b/cmd/serve/sftp/server.go
index 6bf48d42b..ca507882e 100644
--- a/cmd/serve/sftp/server.go
+++ b/cmd/serve/sftp/server.go
@@ -130,8 +130,8 @@ func (s *server) serve() (err error) {
 		fs.Logf(nil, "Loaded %d authorized keys from %q", len(authorizedKeysMap), authKeysFile)
 	}
 
-	if !s.opt.NoAuth && len(authorizedKeysMap) == 0 && s.opt.User == "" && s.opt.Pass == "" {
-		return errors.New("no authorization found, use --user/--pass or --authorized-keys or --no-auth")
+	if !s.opt.NoAuth && len(authorizedKeysMap) == 0 && s.opt.User == "" && s.opt.Pass == "" && s.proxy == nil {
+		return errors.New("no authorization found, use --user/--pass or --authorized-keys or --no-auth or --auth-proxy")
 	}
 
 	// An SSH server is represented by a ServerConfig, which holds
diff --git a/fs/accounting/transfer.go b/fs/accounting/transfer.go
index 4741e170e..e6b6d8d6a 100644
--- a/fs/accounting/transfer.go
+++ b/fs/accounting/transfer.go
@@ -40,16 +40,17 @@ func (as TransferSnapshot) MarshalJSON() ([]byte, error) {
 // accounting functions.
 // Transfer needs to be closed on completion.
 type Transfer struct {
-	stats    *StatsInfo
-	remote   string
-	size     int64
-	checking bool
-
-	// Protects all bellow
-	mu          sync.Mutex
+	// these are initialised at creation and may be accessed without locking
+	stats     *StatsInfo
+	remote    string
+	size      int64
+	startedAt time.Time
+	checking  bool
+
+	// Protects all below
+	mu          sync.RWMutex
 	acc         *Account
 	err         error
-	startedAt   time.Time
 	completedAt time.Time
 }
 
@@ -79,7 +80,6 @@ func newTransferRemoteSize(stats *StatsInfo, remote string, size int64, checking
 // Must be called after transfer is finished to run proper cleanups.
 func (tr *Transfer) Done(err error) {
 	tr.mu.Lock()
-	defer tr.mu.Unlock()
 
 	if err != nil {
 		tr.stats.Error(err)
@@ -90,46 +90,47 @@ func (tr *Transfer) Done(err error) {
 			fs.LogLevelPrintf(fs.Config.StatsLogLevel, nil, "can't close account: %+v\n", err)
 		}
 	}
+
+	tr.completedAt = time.Now()
+
+	tr.mu.Unlock()
+
 	if tr.checking {
 		tr.stats.DoneChecking(tr.remote)
 	} else {
 		tr.stats.DoneTransferring(tr.remote, err == nil)
 	}
-
-	tr.completedAt = time.Now()
 }
 
 // Account returns reader that knows how to keep track of transfer progress.
 func (tr *Transfer) Account(in io.ReadCloser) *Account {
 	tr.mu.Lock()
-	defer tr.mu.Unlock()
-
 	if tr.acc == nil {
 		tr.acc = newAccountSizeName(tr.stats, in, tr.size, tr.remote)
-
 	}
+	tr.mu.Unlock()
 	return tr.acc
 }
 
 // TimeRange returns the time transfer started and ended at. If not completed
 // it will return zero time for end time.
 func (tr *Transfer) TimeRange() (time.Time, time.Time) {
-	tr.mu.Lock()
-	defer tr.mu.Unlock()
+	tr.mu.RLock()
+	defer tr.mu.RUnlock()
 	return tr.startedAt, tr.completedAt
 }
 
 // IsDone returns true if transfer is completed.
 func (tr *Transfer) IsDone() bool {
-	tr.mu.Lock()
-	defer tr.mu.Unlock()
+	tr.mu.RLock()
+	defer tr.mu.RUnlock()
 	return !tr.completedAt.IsZero()
 }
 
 // Snapshot produces stats for this account at point in time.
 func (tr *Transfer) Snapshot() TransferSnapshot {
-	tr.mu.Lock()
-	defer tr.mu.Unlock()
+	tr.mu.RLock()
+	defer tr.mu.RUnlock()
 
 	var s, b int64 = tr.size, 0
 	if tr.acc != nil {
diff --git a/fs/filter/filter.go b/fs/filter/filter.go
index aa9d45ce1..b693c2129 100644
--- a/fs/filter/filter.go
+++ b/fs/filter/filter.go
@@ -22,9 +22,8 @@ var Active = mustNewFilter(nil)
 
 // rule is one filter rule
 type rule struct {
-	Include          bool
-	Regexp           *regexp.Regexp
-	boundedRecursion bool
+	Include bool
+	Regexp  *regexp.Regexp
 }
 
 // Match returns true if rule matches path
@@ -48,14 +47,13 @@ type rules struct {
 }
 
 // add adds a rule if it doesn't exist already
-func (rs *rules) add(Include bool, re *regexp.Regexp, boundedRecursion bool) {
+func (rs *rules) add(Include bool, re *regexp.Regexp) {
 	if rs.existing == nil {
 		rs.existing = make(map[string]struct{})
 	}
 	newRule := rule{
-		Include:          Include,
-		Regexp:           re,
-		boundedRecursion: boundedRecursion,
+		Include: Include,
+		Regexp:  re,
 	}
 	newRuleString := newRule.String()
 	if _, ok := rs.existing[newRuleString]; ok {
@@ -76,23 +74,6 @@ func (rs *rules) len() int {
 	return len(rs.rules)
 }
 
-// boundedRecursion returns true if the set of filters would only
-// need bounded recursion to evaluate
-func (rs *rules) boundedRecursion() bool {
-	var (
-		excludeAll       = false
-		boundedRecursion = true
-	)
-	for _, rule := range rs.rules {
-		if rule.Include {
-			boundedRecursion = boundedRecursion && rule.boundedRecursion
-		} else if rule.Regexp.String() == `^.*$` {
-			excludeAll = true
-		}
-	}
-	return excludeAll && boundedRecursion
-}
-
 // FilesMap describes the map of files to transfer
 type FilesMap map[string]struct{}
 
@@ -252,8 +233,7 @@ func (f *Filter) addDirGlobs(Include bool, glob string) error {
 		if err != nil {
 			return err
 		}
-		boundedRecursion := globBoundedRecursion(dirGlob)
-		f.dirRules.add(Include, dirRe, boundedRecursion)
+		f.dirRules.add(Include, dirRe)
 	}
 	return nil
 }
@@ -269,9 +249,8 @@ func (f *Filter) Add(Include bool, glob string) error {
 	if err != nil {
 		return err
 	}
-	boundedRecursion := globBoundedRecursion(glob)
 	if isFileRule {
-		f.fileRules.add(Include, re, boundedRecursion)
+		f.fileRules.add(Include, re)
 		// If include rule work out what directories are needed to scan
 		// if exclude rule, we can't rule anything out
 		// Unless it is `*` which matches everything
@@ -284,7 +263,7 @@ func (f *Filter) Add(Include bool, glob string) error {
 		}
 	}
 	if isDirRule {
-		f.dirRules.add(Include, re, boundedRecursion)
+		f.dirRules.add(Include, re)
 	}
 	return nil
 }
@@ -365,12 +344,6 @@ func (f *Filter) InActive() bool {
 		len(f.Opt.ExcludeFile) == 0)
 }
 
-// BoundedRecursion returns true if the filter can be evaluated with
-// bounded recursion only.
-func (f *Filter) BoundedRecursion() bool {
-	return f.fileRules.boundedRecursion()
-}
-
 // includeRemote returns whether this remote passes the filter rules.
 func (f *Filter) includeRemote(remote string) bool {
 	for _, rule := range f.fileRules.rules {
@@ -570,3 +543,19 @@ func (f *Filter) MakeListR(ctx context.Context, NewObject func(ctx context.Conte
 		return g.Wait()
 	}
 }
+
+// UsesDirectoryFilters returns true if the filter uses directory
+// filters and false if it doesn't.
+//
+// This is used in deciding whether to walk directories or use ListR
+func (f *Filter) UsesDirectoryFilters() bool {
+	if len(f.dirRules.rules) == 0 {
+		return false
+	}
+	rule := f.dirRules.rules[0]
+	re := rule.Regexp.String()
+	if rule.Include == true && re == "^.*$" {
+		return false
+	}
+	return true
+}
diff --git a/fs/filter/glob.go b/fs/filter/glob.go
index 9fff1ca73..96a48d3d4 100644
--- a/fs/filter/glob.go
+++ b/fs/filter/glob.go
@@ -167,15 +167,3 @@ func globToDirGlobs(glob string) (out []string) {
 
 	return out
 }
-
-// globBoundedRecursion returns true if the glob only needs bounded
-// recursion in the file tree to evaluate.
-func globBoundedRecursion(glob string) bool {
-	if strings.Contains(glob, "**") {
-		return false
-	}
-	if strings.HasPrefix(glob, "/") {
-		return true
-	}
-	return false
-}
diff --git a/fs/operations/operations.go b/fs/operations/operations.go
index ad5d20b0a..81c723174 100644
--- a/fs/operations/operations.go
+++ b/fs/operations/operations.go
@@ -1222,8 +1222,9 @@ func Rcat(ctx context.Context, fdst fs.Fs, dstFileName string, in io.ReadCloser,
 	}()
 	in = tr.Account(in).WithBuffer()
 
-	hashOption := &fs.HashesOption{Hashes: fdst.Hashes()}
-	hash, err := hash.NewMultiHasherTypes(fdst.Hashes())
+	hashes := hash.NewHashSet(fdst.Hashes().GetOne()) // just pick one hash
+	hashOption := &fs.HashesOption{Hashes: hashes}
+	hash, err := hash.NewMultiHasherTypes(hashes)
 	if err != nil {
 		return nil, err
 	}
diff --git a/fs/walk/walk.go b/fs/walk/walk.go
index bf61a819a..a424e3a33 100644
--- a/fs/walk/walk.go
+++ b/fs/walk/walk.go
@@ -51,7 +51,7 @@ type Func func(path string, entries fs.DirEntries, err error) error
 //
 // Parent directories are always listed before their children
 //
-// This is implemented by WalkR if Config.UseUseListR is true
+// This is implemented by WalkR if Config.UseListR is true
 // and f supports it and level > 1, or WalkN otherwise.
 //
 // If --files-from and --no-traverse is set then a DirTree will be
@@ -146,7 +146,7 @@ func ListR(ctx context.Context, f fs.Fs, path string, includeAll bool, maxLevel
 		filter.Active.HaveFilesFrom() || // ...using --files-from
 		maxLevel >= 0 || // ...using bounded recursion
 		len(filter.Active.Opt.ExcludeFile) > 0 || // ...using --exclude-file
-		filter.Active.BoundedRecursion() { // ...filters imply bounded recursion
+		filter.Active.UsesDirectoryFilters() { // ...using any directory filters
 		return listRwalk(ctx, f, path, includeAll, maxLevel, listType, fn)
 	}
 	return listR(ctx, f, path, includeAll, listType, fn, doListR, listType.Dirs() && f.Features().BucketBased)
diff --git a/vfs/write.go b/vfs/write.go
index 68ad2b7b9..7939a4823 100644
--- a/vfs/write.go
+++ b/vfs/write.go
@@ -5,6 +5,7 @@ import (
 	"io"
 	"os"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/rclone/rclone/fs"
@@ -15,7 +16,8 @@ import (
 type WriteFileHandle struct {
 	baseHandle
 	mu          sync.Mutex
-	closed      bool // set if handle has been closed
+	cond        *sync.Cond // cond lock for out of sequence writes
+	closed      bool       // set if handle has been closed
 	remote      string
 	pipeWriter  *io.PipeWriter
 	o           fs.Object
@@ -42,6 +44,7 @@ func newWriteFileHandle(d *Dir, f *File, remote string, flags int) (*WriteFileHa
 		result: make(chan error, 1),
 		file:   f,
 	}
+	fh.cond = sync.NewCond(&fh.mu)
 	fh.file.addWriter(fh)
 	return fh, nil
 }
@@ -127,14 +130,28 @@ func (fh *WriteFileHandle) writeAt(p []byte, off int64) (n int, err error) {
 		fs.Errorf(fh.remote, "WriteFileHandle.Write: error: %v", EBADF)
 		return 0, ECLOSED
 	}
-	// Wait a short time for sequential writes to appear
-	const maxTries = 1000
-	const sleepTime = 1 * time.Millisecond
-	for try := 1; fh.offset != off && try <= maxTries; try++ {
-		//fs.Debugf(fh.remote, "waiting for in sequence write %d/%d", try, maxTries)
-		fh.mu.Unlock()
-		time.Sleep(sleepTime)
-		fh.mu.Lock()
+	if fh.offset != off {
+		// Set a background timer so we don't wait forever
+		timeout := time.NewTimer(10 * time.Second)
+		done := make(chan struct{})
+		abort := int32(0)
+		go func() {
+			select {
+			case <-timeout.C:
+				// set abort flag an give all the waiting goroutines a kick on timeout
+				atomic.StoreInt32(&abort, 1)
+				fh.cond.Broadcast()
+			case <-done:
+			}
+		}()
+		// Wait for an in-sequence write or abort
+		for fh.offset != off && atomic.LoadInt32(&abort) == 0 {
+			// fs.Debugf(fh.remote, "waiting for in-sequence write to %d", off)
+			fh.cond.Wait()
+		}
+		// tidy up end timer
+		close(done)
+		timeout.Stop()
 	}
 	if fh.offset != off {
 		fs.Errorf(fh.remote, "WriteFileHandle.Write: can't seek in file without --vfs-cache-mode >= writes")
@@ -152,6 +169,7 @@ func (fh *WriteFileHandle) writeAt(p []byte, off int64) (n int, err error) {
 		return 0, err
 	}
 	// fs.Debugf(fh.remote, "WriteFileHandle.Write OK (%d bytes written)", n)
+	fh.cond.Broadcast() // wake everyone up waiting for an in-sequence read
 	return n, nil
 }
 


Test patch:
diff --git a/backend/swift/swift_test.go b/backend/swift/swift_test.go
index 7b182230a..f7f4e0ecb 100644
--- a/backend/swift/swift_test.go
+++ b/backend/swift/swift_test.go
@@ -2,10 +2,19 @@
 package swift
 
 import (
+	"bytes"
+	"context"
+	"io"
 	"testing"
 
 	"github.com/rclone/rclone/fs"
+	"github.com/rclone/rclone/fs/hash"
+	"github.com/rclone/rclone/fs/object"
+	"github.com/rclone/rclone/fstest"
 	"github.com/rclone/rclone/fstest/fstests"
+	"github.com/rclone/rclone/lib/random"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 // TestIntegration runs integration tests against the remote
@@ -21,3 +30,50 @@ func (f *Fs) SetUploadChunkSize(cs fs.SizeSuffix) (fs.SizeSuffix, error) {
 }
 
 var _ fstests.SetUploadChunkSizer = (*Fs)(nil)
+
+// Check that PutStream works with NoChunk as it is the major code
+// deviation
+func (f *Fs) testNoChunk(t *testing.T) {
+	ctx := context.Background()
+	f.opt.NoChunk = true
+	defer func() {
+		f.opt.NoChunk = false
+	}()
+
+	file := fstest.Item{
+		ModTime: fstest.Time("2001-02-03T04:05:06.499999999Z"),
+		Path:    "piped data no chunk.txt",
+		Size:    -1, // use unknown size during upload
+	}
+
+	const contentSize = 100
+
+	contents := random.String(contentSize)
+	buf := bytes.NewBufferString(contents)
+	uploadHash := hash.NewMultiHasher()
+	in := io.TeeReader(buf, uploadHash)
+
+	file.Size = -1
+	obji := object.NewStaticObjectInfo(file.Path, file.ModTime, file.Size, true, nil, nil)
+	obj, err := f.Features().PutStream(ctx, in, obji)
+	require.NoError(t, err)
+
+	file.Hashes = uploadHash.Sums()
+	file.Size = int64(contentSize) // use correct size when checking
+	file.Check(t, obj, f.Precision())
+
+	// Re-read the object and check again
+	obj, err = f.NewObject(ctx, file.Path)
+	require.NoError(t, err)
+	file.Check(t, obj, f.Precision())
+
+	// Delete the object
+	assert.NoError(t, obj.Remove(ctx))
+}
+
+// Additional tests that aren't in the framework
+func (f *Fs) InternalTest(t *testing.T) {
+	t.Run("NoChunk", f.testNoChunk)
+}
+
+var _ fstests.InternalTester = (*Fs)(nil)
diff --git a/fs/filter/filter_test.go b/fs/filter/filter_test.go
index e5c360ab2..36c425cba 100644
--- a/fs/filter/filter_test.go
+++ b/fs/filter/filter_test.go
@@ -26,7 +26,6 @@ func TestNewFilterDefault(t *testing.T) {
 	assert.Len(t, f.dirRules.rules, 0)
 	assert.Nil(t, f.files)
 	assert.True(t, f.InActive())
-	assert.False(t, f.BoundedRecursion())
 }
 
 // testFile creates a temp file with the contents
@@ -105,38 +104,6 @@ func TestNewFilterFull(t *testing.T) {
 		}
 	}
 	assert.False(t, f.InActive())
-	assert.False(t, f.BoundedRecursion())
-}
-
-func TestFilterBoundedRecursion(t *testing.T) {
-	for _, test := range []struct {
-		in   string
-		want bool
-	}{
-		{"", false},
-		{"- /**", true},
-		{"+ *.jpg", false},
-		{"+ *.jpg\n- /**", false},
-		{"+ /*.jpg\n- /**", true},
-		{"+ *.png\n+ /*.jpg\n- /**", false},
-		{"+ /*.png\n+ /*.jpg\n- /**", true},
-		{"- *.jpg\n- /**", true},
-		{"+ /*.jpg\n- /**", true},
-		{"+ /*dir/\n- /**", true},
-		{"+ /*dir/\n", false},
-		{"+ /*dir/**\n- /**", false},
-		{"+ **/pics*/*.jpg\n- /**", false},
-	} {
-		f, err := NewFilter(nil)
-		require.NoError(t, err)
-		for _, rule := range strings.Split(test.in, "\n") {
-			if rule != "" {
-				require.NoError(t, f.AddRule(rule))
-			}
-		}
-		got := f.BoundedRecursion()
-		assert.Equal(t, test.want, got, test.in)
-	}
 }
 
 type includeTest struct {
@@ -185,7 +152,6 @@ func TestNewFilterIncludeFiles(t *testing.T) {
 		{"file3.jpg", 3, 0, false},
 	})
 	assert.False(t, f.InActive())
-	assert.False(t, f.BoundedRecursion())
 }
 
 func TestNewFilterIncludeFilesDirs(t *testing.T) {
@@ -313,7 +279,6 @@ func TestNewFilterMinSize(t *testing.T) {
 		{"potato/file2.jpg", 99, 0, false},
 	})
 	assert.False(t, f.InActive())
-	assert.False(t, f.BoundedRecursion())
 }
 
 func TestNewFilterMaxSize(t *testing.T) {
@@ -589,3 +554,76 @@ func TestFilterMatchesFromDocs(t *testing.T) {
 		}
 	}
 }
+
+func TestNewFilterUsesDirectoryFilters(t *testing.T) {
+	for i, test := range []struct {
+		rules []string
+		want  bool
+	}{
+		{
+			rules: []string{},
+			want:  false,
+		},
+		{
+			rules: []string{
+				"+ *",
+			},
+			want: false,
+		},
+		{
+			rules: []string{
+				"+ *.jpg",
+				"- *",
+			},
+			want: false,
+		},
+		{
+			rules: []string{
+				"- *.jpg",
+			},
+			want: false,
+		},
+		{
+			rules: []string{
+				"- *.jpg",
+				"+ *",
+			},
+			want: false,
+		},
+		{
+			rules: []string{
+				"+ dir/*.jpg",
+				"- *",
+			},
+			want: true,
+		},
+		{
+			rules: []string{
+				"+ dir/**",
+			},
+			want: true,
+		},
+		{
+			rules: []string{
+				"- dir/**",
+			},
+			want: true,
+		},
+		{
+			rules: []string{
+				"- /dir/**",
+			},
+			want: true,
+		},
+	} {
+		what := fmt.Sprintf("#%d", i)
+		f, err := NewFilter(nil)
+		require.NoError(t, err)
+		for _, rule := range test.rules {
+			err := f.AddRule(rule)
+			require.NoError(t, err, what)
+		}
+		got := f.UsesDirectoryFilters()
+		assert.Equal(t, test.want, got, fmt.Sprintf("%s: %s", what, f.DumpFilters()))
+	}
+}
diff --git a/fs/filter/glob_test.go b/fs/filter/glob_test.go
index 372fa63e3..008d4bfd3 100644
--- a/fs/filter/glob_test.go
+++ b/fs/filter/glob_test.go
@@ -108,45 +108,3 @@ func TestGlobToDirGlobs(t *testing.T) {
 		assert.Equal(t, test.want, got, test.in)
 	}
 }
-
-func TestGlobBoundedRecursion(t *testing.T) {
-	for _, test := range []struct {
-		in   string
-		want bool
-	}{
-		{`*`, false},
-		{`/*`, true},
-		{`/**`, false},
-		{`*.jpg`, false},
-		{`/*.jpg`, true},
-		{`/a/*.jpg`, true},
-		{`/a/b/*.jpg`, true},
-		{`*/*/*.jpg`, false},
-		{`a/b/`, false},
-		{`a/b`, false},
-		{`a/b/*.{png,gif}`, false},
-		{`/a/{jpg,png,gif}/*.{jpg,true,gif}`, true},
-		{`a/{a,a*b,a**c}/d/`, false},
-		{`/a/{a,a*b,a/c,d}/d/`, true},
-		{`**`, false},
-		{`a**`, false},
-		{`a**b`, false},
-		{`a**b**c**d`, false},
-		{`a**b/c**d`, false},
-		{`/A/a**b/B/c**d/C/`, false},
-		{`/var/spool/**/ncw`, false},
-		{`var/spool/**/ncw/`, false},
-		{"/file1.jpg", true},
-		{"/file2.png", true},
-		{"/*.jpg", true},
-		{"/*.png", true},
-		{"/potato", true},
-		{"/sausage1", true},
-		{"/sausage2*", true},
-		{"/sausage3**", false},
-		{"/a/*.jpg", true},
-	} {
-		got := globBoundedRecursion(test.in)
-		assert.Equal(t, test.want, got, test.in)
-	}
-}