From a5b034a992141c049c41a77d3fe65615a8655fbd Mon Sep 17 00:00:00 2001 From: Nick Craig-Wood Date: Sat, 18 Nov 2017 11:57:40 +0000 Subject: [PATCH] vfs: add WaitForWriters to wait until all writers have finished --- vfs/file.go | 7 +++++++ vfs/vfs.go | 42 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/vfs/file.go b/vfs/file.go index d403d4b4c..f2bce87e0 100644 --- a/vfs/file.go +++ b/vfs/file.go @@ -115,6 +115,13 @@ func (f *File) delWriter(h Handle) { f.mu.Unlock() } +// activeWriters returns the number of writers on the file +func (f *File) activeWriters() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.writers) +} + // ModTime returns the modified time of the file // // if NoModTime is set then it returns the mod time of the directory diff --git a/vfs/vfs.go b/vfs/vfs.go index 74c478514..858ea6ae6 100644 --- a/vfs/vfs.go +++ b/vfs/vfs.go @@ -248,6 +248,48 @@ func (vfs *VFS) CleanUp() error { return vfs.cache.cleanUp() } + +// WaitForWriters sleeps until all writers have finished or +// time.Duration has elapsed +func (vfs *VFS) WaitForWriters(timeout time.Duration) { + defer fs.Trace(nil, "timeout=%v", timeout)("") + const tickTime = 1 * time.Second + deadline := time.NewTimer(timeout) + defer deadline.Stop() + tick := time.NewTimer(tickTime) + defer tick.Stop() + tick.Stop() + for { + writers := 0 + vfs.root.walk("", func(d *Dir) { + fs.Debugf(d.path, "Looking for writers") + // NB d.mu is held by walk() here + for leaf, item := range d.items { + fs.Debugf(leaf, "reading active writers") + if file, ok := item.(*File); ok { + n := file.activeWriters() + if n != 0 { + fs.Debugf(file, "active writers %d", n) + } + writers += n + } + } + }) + if writers == 0 { + return + } + fs.Debugf(nil, "Still %d writers active, waiting %v", writers, tickTime) + tick.Reset(tickTime) + select { + case <-tick.C: + break + case <-deadline.C: + fs.Errorf(nil, "Exiting even though %d writers are active after %v", writers, timeout) + return + } + } +} + // Root returns the root node func (vfs *VFS) Root() (*Dir, error) { // fs.Debugf(vfs.f, "Root()")