diff --git a/fs/accounting/accounting.go b/fs/accounting/accounting.go index 0c6c0a50c..244df0784 100644 --- a/fs/accounting/accounting.go +++ b/fs/accounting/accounting.go @@ -128,6 +128,13 @@ func (acc *Account) GetAsyncReader() *asyncreader.AsyncReader { // StopBuffering stops the async buffer doing any more buffering func (acc *Account) StopBuffering() { + if asyncIn, ok := acc.in.(*asyncreader.AsyncReader); ok { + asyncIn.StopBuffering() + } +} + +// Abandon stops the async buffer doing any more buffering +func (acc *Account) Abandon() { if asyncIn, ok := acc.in.(*asyncreader.AsyncReader); ok { asyncIn.Abandon() } @@ -139,7 +146,7 @@ func (acc *Account) UpdateReader(in io.ReadCloser) { acc.mu.Lock() withBuf := acc.withBuf if withBuf { - acc.StopBuffering() + acc.Abandon() acc.withBuf = false } acc.in = in diff --git a/fs/asyncreader/asyncreader.go b/fs/asyncreader/asyncreader.go index 1df1603f9..6b97c8f6c 100644 --- a/fs/asyncreader/asyncreader.go +++ b/fs/asyncreader/asyncreader.go @@ -268,9 +268,15 @@ func (a *AsyncReader) SkipBytes(skip int) (ok bool) { } } -// Abandon will ensure that the underlying async reader is shut down. -// It will NOT close the input supplied on New. -func (a *AsyncReader) Abandon() { +// StopBuffering will ensure that the underlying async reader is shut +// down so no more is read from the input. +// +// This does not free the memory so Abandon() or Close() need to be +// called on the input. +// +// This does not wait for Read/WriteTo to complete so can be called +// concurrently to those. +func (a *AsyncReader) StopBuffering() { select { case <-a.exit: // Do nothing if reader routine already exited @@ -280,6 +286,14 @@ func (a *AsyncReader) Abandon() { // Close and wait for go routine close(a.exit) <-a.exited +} + +// Abandon will ensure that the underlying async reader is shut down +// and memory is returned. It does everything but close the input. +// +// It will NOT close the input supplied on New. +func (a *AsyncReader) Abandon() { + a.StopBuffering() // take the lock to wait for Read/WriteTo to complete a.mu.Lock() defer a.mu.Unlock()