mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-15 08:18:42 +01:00
Fix/s3 abandoned chunk download (#12616)
* filer: cancel a chunk download that no reader waits for * filer: stop waiting for reader cache budget once cancelled * filer: keep a cancelled download's re-lookups off its context The refresh and retry lookups took the download's context. FilerClient coalesces concurrent lookups of a volume, so a cancelled download leading one failed the other readers waiting on it. Fetches still stop on cancel. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer: test that cancelling a download leaves its re-lookup running Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer: cancel unheld downloads when the reader cache is destroyed ReaderCache.destroy waited for every downloader, so closing a reader waited for a stalled prefetch nobody read. Downloads with no reader or stream on them are now cancelled first; ones still in use keep running. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer: test that destroying the reader cache cancels an unread prefetch Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer: cancel a destroyed download once its last holder leaves destroy skipped downloads a reader or stream still held, and once they were out of the map nothing cancelled them, so destroy waited for a fetch nobody used. removeConsumedLocked now cancels any consumable download, whether or not it is still in the map. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer: test a destroyed download cancelled after its last stream leaves Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer: release the stream only after destroy dropped the download Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer: test a cancelled stream leaving a waiting reader's download running Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
4 files changed
+515
-18
No files matched your search
+45
-17
@@ -49,6 +49,9 @@ type SingleChunkCacher struct {
|
||||
wg sync.WaitGroup
|
||||
cacheStartedCh chan struct{}
|
||||
done chan struct{} // signals when download is complete
|
||||
// ctx is cancelled once no reader can use the download, or when it ends
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
func NewReaderCache(limit int, chunkCache chunk_cache.ChunkCache, lookupFileIdFn wdclient.LookupFileIdFunctionType, cacheInvalidator CacheInvalidator, budgets ...*ReaderCacheBudget) *ReaderCache {
|
||||
@@ -392,12 +395,18 @@ func (rc *ReaderCache) removeConsumed(downloader *SingleChunkCacher) {
|
||||
}
|
||||
}
|
||||
|
||||
// removeConsumedLocked also cancels a running download, as no reader can attach to it any more,
|
||||
// including one destroy already dropped from the map.
|
||||
func (rc *ReaderCache) removeConsumedLocked(downloader *SingleChunkCacher) (removed bool) {
|
||||
if rc.downloaders[downloader.chunkFileId] == downloader && downloader.consumable() {
|
||||
delete(rc.downloaders, downloader.chunkFileId)
|
||||
return true
|
||||
if !downloader.consumable() {
|
||||
return false
|
||||
}
|
||||
return false
|
||||
downloader.cancel()
|
||||
if rc.downloaders[downloader.chunkFileId] != downloader {
|
||||
return false
|
||||
}
|
||||
delete(rc.downloaders, downloader.chunkFileId)
|
||||
return true
|
||||
}
|
||||
|
||||
func (rc *ReaderCache) destroy() {
|
||||
@@ -405,13 +414,24 @@ func (rc *ReaderCache) destroy() {
|
||||
downloaders := rc.downloaders
|
||||
rc.downloaders = make(map[string]*SingleChunkCacher)
|
||||
rc.Unlock()
|
||||
// out of the map no stream can come back to a download, so it is cancelled
|
||||
// now or once its last reader or stream leaves
|
||||
for _, downloader := range downloaders {
|
||||
atomic.StoreInt32(&downloader.left, 1)
|
||||
if downloader.consumable() {
|
||||
downloader.cancel()
|
||||
}
|
||||
}
|
||||
for _, downloader := range downloaders {
|
||||
downloader.destroy()
|
||||
}
|
||||
}
|
||||
|
||||
func newSingleChunkCacher(parent *ReaderCache, fileId string, cipherKey []byte, isGzipped bool, chunkSize int, shouldCache bool) *SingleChunkCacher {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &SingleChunkCacher{
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
parent: parent,
|
||||
chunkFileId: fileId,
|
||||
cipherKey: cipherKey,
|
||||
@@ -428,6 +448,7 @@ func newSingleChunkCacher(parent *ReaderCache, fileId string, cipherKey []byte,
|
||||
// downloaders map, so a removal can never observe the fetch as absent.
|
||||
func (s *SingleChunkCacher) startCaching() {
|
||||
defer func() {
|
||||
s.cancel()
|
||||
close(s.done)
|
||||
s.wg.Done()
|
||||
if s.hasCompletedError() {
|
||||
@@ -443,13 +464,7 @@ func (s *SingleChunkCacher) startCaching() {
|
||||
return
|
||||
}
|
||||
|
||||
// Intentionally use context.Background(), not a request-specific context.
|
||||
// The downloaded chunk is a shared resource: multiple concurrent readers may
|
||||
// wait on this same download via s.done. A request-scoped context that got
|
||||
// cancelled would abort the download and error every other waiting reader.
|
||||
// The download always runs to completion once started; readers that cancel
|
||||
// individually drop out via readChunkAt's select on ctx.Done().
|
||||
urlStrings, err := s.parent.lookupFileIdFn(context.Background(), s.chunkFileId)
|
||||
urlStrings, err := s.lookupFileId(s.ctx, s.chunkFileId)
|
||||
if err != nil {
|
||||
s.setError(fmt.Errorf("operation LookupFileId %s failed, err: %v", s.chunkFileId, err))
|
||||
return
|
||||
@@ -459,9 +474,16 @@ func (s *SingleChunkCacher) startCaching() {
|
||||
return
|
||||
}
|
||||
|
||||
data, fetchErr := s.fetchChunkData(context.Background(), urlStrings)
|
||||
// not a request's context: other readers may wait on this download via s.done
|
||||
// and it is cancelled only when removeConsumedLocked drops the cacher
|
||||
data, fetchErr := s.fetchChunkData(s.ctx, urlStrings)
|
||||
if fetchErr != nil {
|
||||
data, fetchErr = s.retryFetchAfterCacheInvalidation(context.Background(), urlStrings, fetchErr)
|
||||
data, fetchErr = s.retryFetchAfterCacheInvalidation(s.ctx, urlStrings, fetchErr)
|
||||
}
|
||||
// a cancelled fetch can return a partly filled buffer without an error
|
||||
if fetchErr == nil && s.ctx.Err() != nil {
|
||||
mem.Free(data)
|
||||
data, fetchErr = nil, s.ctx.Err()
|
||||
}
|
||||
|
||||
// Now acquire lock to update state
|
||||
@@ -498,7 +520,7 @@ func (s *SingleChunkCacher) fetchChunkData(ctx context.Context, urlStrings []str
|
||||
// Allocate buffer and download without holding the lock.
|
||||
// This allows multiple downloads to proceed in parallel.
|
||||
data := mem.Allocate(s.chunkSize)
|
||||
_, fetchErr := s.parent.fetchChunkDataFn(ctx, data, urlStrings, s.cipherKey, s.isGzipped, true, 0, s.chunkFileId, refreshUrls(ctx, s.parent.cacheInvalidator, s.parent.lookupFileIdFn, s.chunkFileId))
|
||||
_, fetchErr := s.parent.fetchChunkDataFn(ctx, data, urlStrings, s.cipherKey, s.isGzipped, true, 0, s.chunkFileId, refreshUrls(ctx, s.parent.cacheInvalidator, s.lookupFileId, s.chunkFileId))
|
||||
if fetchErr != nil {
|
||||
mem.Free(data)
|
||||
return nil, fetchErr
|
||||
@@ -508,7 +530,7 @@ func (s *SingleChunkCacher) fetchChunkData(ctx context.Context, urlStrings []str
|
||||
|
||||
func (s *SingleChunkCacher) retryFetchAfterCacheInvalidation(ctx context.Context, oldUrlStrings []string, originalErr error) ([]byte, error) {
|
||||
var data []byte
|
||||
err := retryFetchWithFreshLocations(ctx, s.parent.cacheInvalidator, s.parent.lookupFileIdFn, s.chunkFileId, oldUrlStrings, originalErr, func(newUrls []string) error {
|
||||
err := retryFetchWithFreshLocations(ctx, s.parent.cacheInvalidator, s.lookupFileId, s.chunkFileId, oldUrlStrings, originalErr, func(newUrls []string) error {
|
||||
var fetchErr error
|
||||
data, fetchErr = s.fetchChunkData(ctx, newUrls)
|
||||
return fetchErr
|
||||
@@ -519,6 +541,12 @@ func (s *SingleChunkCacher) retryFetchAfterCacheInvalidation(ctx context.Context
|
||||
return data, nil
|
||||
}
|
||||
|
||||
// lookupFileId ignores the download's context: FilerClient coalesces concurrent
|
||||
// lookups of a volume, so a cancelled download must not fail other readers' lookups.
|
||||
func (s *SingleChunkCacher) lookupFileId(_ context.Context, fileId string) ([]string, error) {
|
||||
return s.parent.lookupFileIdFn(context.Background(), fileId)
|
||||
}
|
||||
|
||||
func (s *SingleChunkCacher) destroy() {
|
||||
// wait for all reads to finish before destroying the data
|
||||
s.wg.Wait()
|
||||
@@ -533,8 +561,8 @@ func (s *SingleChunkCacher) destroy() {
|
||||
|
||||
// readChunkAt reads data from the cached chunk.
|
||||
// It waits for the download to complete if it's still in progress.
|
||||
// The ctx parameter allows the reader to cancel its wait (but the download continues
|
||||
// for other readers - see comment in startCaching about shared resource semantics).
|
||||
// The ctx parameter allows the reader to cancel its wait; the download continues
|
||||
// for the other readers and streams that still wait on it.
|
||||
// The caller must s.wg.Add(1) under the ReaderCache lock before calling; this only releases it.
|
||||
func (s *SingleChunkCacher) readChunkAt(ctx context.Context, buf []byte, offset int64) (n int, err error) {
|
||||
defer func() {
|
||||
|
||||
@@ -45,6 +45,11 @@ func (b *ReaderCacheBudget) reserve(s *SingleChunkCacher) error {
|
||||
return fmt.Errorf("chunk buffer needs %d bytes, exceeding reader cache budget %d; increase the readerCacheSizeMB budget", size, b.limit)
|
||||
}
|
||||
for {
|
||||
// a cancelled download must not evict a chunk for itself
|
||||
if err := s.ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
b.Lock()
|
||||
if size <= b.limit-b.used {
|
||||
b.used += size
|
||||
@@ -89,7 +94,11 @@ func (b *ReaderCacheBudget) reserve(s *SingleChunkCacher) error {
|
||||
}
|
||||
changed := b.changed
|
||||
b.Unlock()
|
||||
<-changed
|
||||
select {
|
||||
case <-changed:
|
||||
case <-s.ctx.Done():
|
||||
return s.ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -428,3 +428,60 @@ func TestReaderCacheReReadAfterEviction(t *testing.T) {
|
||||
t.Fatalf("fetchCount=%d, want 3 (a, b, a-re-read)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// A download still waiting for budget is cancelled when its last stream leaves.
|
||||
func TestReaderCacheBudgetWaitStopsWithItsLastStream(t *testing.T) {
|
||||
budget := NewReaderCacheBudget(4 << 10)
|
||||
var lookups sync.Map
|
||||
rc := NewReaderCache(256, newMockChunkCacheForReaderCache(), func(_ context.Context, fileId string) ([]string, error) {
|
||||
lookups.Store(fileId, true)
|
||||
return []string{"unused"}, nil
|
||||
}, nil, budget)
|
||||
defer rc.destroy()
|
||||
|
||||
holding := make(chan struct{})
|
||||
var holdingOnce sync.Once
|
||||
finish := make(chan struct{})
|
||||
rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) {
|
||||
holdingOnce.Do(func() { close(holding) })
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return 0, ctx.Err()
|
||||
case <-finish:
|
||||
return len(buffer), nil
|
||||
}
|
||||
}
|
||||
defer close(finish)
|
||||
|
||||
// "a" takes the whole budget and its download does not finish
|
||||
go rc.ReadChunkAt(context.Background(), make([]byte, 1), "a", nil, false, 0, 3<<10, false)
|
||||
<-holding
|
||||
|
||||
var stream chunkStream
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
readDone := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := rc.readChunkAt(ctx, &stream, make([]byte, 1), "b", nil, true, 0, 3<<10, false)
|
||||
readDone <- err
|
||||
}()
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
if _, looked := lookups.Load("b"); looked {
|
||||
t.Fatal(`"b" did not wait for budget`)
|
||||
}
|
||||
|
||||
cancel()
|
||||
if err := <-readDone; err != context.Canceled {
|
||||
t.Fatalf("cancelled read: got %v, want context.Canceled", err)
|
||||
}
|
||||
|
||||
released := make(chan struct{})
|
||||
go func() {
|
||||
rc.releaseStream(&stream)
|
||||
close(released)
|
||||
}()
|
||||
select {
|
||||
case <-released:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("releasing the stream waited for budget nobody needs")
|
||||
}
|
||||
}
|
||||
@@ -773,3 +773,406 @@ func TestReaderCacheBookkeepingOffCacheLock(t *testing.T) {
|
||||
}
|
||||
rc.Unlock()
|
||||
}
|
||||
|
||||
// Releasing the last stream of an unfinished download cancels it without waiting for it.
|
||||
func TestAbandonedChunkDownloadStopsWithItsLastStream(t *testing.T) {
|
||||
lookupFn := func(ctx context.Context, fileId string) ([]string, error) {
|
||||
return []string{"http://volume/" + fileId}, nil
|
||||
}
|
||||
rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, nil)
|
||||
defer rc.destroy()
|
||||
|
||||
fetchStarted := make(chan struct{})
|
||||
fetchEnded := make(chan error, 1)
|
||||
rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) {
|
||||
close(fetchStarted)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
fetchEnded <- ctx.Err()
|
||||
return 0, ctx.Err()
|
||||
case <-time.After(10 * time.Second):
|
||||
fetchEnded <- nil
|
||||
return 0, fmt.Errorf("volume server never answered")
|
||||
}
|
||||
}
|
||||
|
||||
// a gzipped chunk is read through the shared whole-chunk download
|
||||
var stream chunkStream
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
readDone := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := rc.readChunkAt(ctx, &stream, make([]byte, 100), "3077,02c3475d67aaf564", nil, true, 0, 4<<20, false)
|
||||
readDone <- err
|
||||
}()
|
||||
|
||||
<-fetchStarted
|
||||
cancel()
|
||||
if err := <-readDone; err != context.Canceled {
|
||||
t.Fatalf("cancelled read: got %v, want context.Canceled", err)
|
||||
}
|
||||
|
||||
released := make(chan struct{})
|
||||
go func() {
|
||||
rc.releaseStream(&stream)
|
||||
close(released)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-released:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("releasing the stream waited for a download nobody wants")
|
||||
}
|
||||
|
||||
select {
|
||||
case err := <-fetchEnded:
|
||||
if err != context.Canceled {
|
||||
t.Fatalf("abandoned download: got %v, want context.Canceled", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("the abandoned download kept running")
|
||||
}
|
||||
}
|
||||
|
||||
// One stream leaving a shared download must not cancel it for another stream.
|
||||
func TestAbandonedChunkDownloadKeepsRunningForOtherStream(t *testing.T) {
|
||||
content := []byte("gzipped chunk 01")
|
||||
fileId := "3077,02c3475d67aaf564"
|
||||
lookupFn := func(ctx context.Context, fileId string) ([]string, error) {
|
||||
return []string{"http://volume/" + fileId}, nil
|
||||
}
|
||||
rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, nil)
|
||||
defer rc.destroy()
|
||||
|
||||
fetchStarted := make(chan struct{})
|
||||
finishFetch := make(chan struct{})
|
||||
rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) {
|
||||
close(fetchStarted)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return 0, ctx.Err()
|
||||
case <-finishFetch:
|
||||
return copy(buffer, content), nil
|
||||
}
|
||||
}
|
||||
|
||||
read := func(ctx context.Context, stream *chunkStream, done chan<- error, buffer []byte) {
|
||||
_, err := rc.readChunkAt(ctx, stream, buffer, fileId, nil, true, 0, len(content), false)
|
||||
done <- err
|
||||
}
|
||||
|
||||
var leaving, staying chunkStream
|
||||
leavingCtx, cancel := context.WithCancel(context.Background())
|
||||
leavingDone, stayingDone := make(chan error, 1), make(chan error, 1)
|
||||
stayingBuffer := make([]byte, len(content))
|
||||
go read(leavingCtx, &leaving, leavingDone, make([]byte, len(content)))
|
||||
<-fetchStarted
|
||||
go read(context.Background(), &staying, stayingDone, stayingBuffer)
|
||||
|
||||
// wait until the second stream is attached to the same download
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for {
|
||||
rc.RLock()
|
||||
cacher := rc.downloaders[fileId]
|
||||
rc.RUnlock()
|
||||
if cacher != nil && atomic.LoadInt32(&cacher.pins) == 2 {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("second stream never attached to the download")
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
|
||||
cancel()
|
||||
if err := <-leavingDone; err != context.Canceled {
|
||||
t.Fatalf("cancelled read: got %v, want context.Canceled", err)
|
||||
}
|
||||
rc.releaseStream(&leaving)
|
||||
|
||||
close(finishFetch)
|
||||
select {
|
||||
case err := <-stayingDone:
|
||||
if err != nil {
|
||||
t.Fatalf("staying stream: got %v, want the chunk", err)
|
||||
}
|
||||
if string(stayingBuffer) != string(content) {
|
||||
t.Fatalf("staying stream read %q, want %q", stayingBuffer, content)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("staying stream never got its data")
|
||||
}
|
||||
rc.releaseStream(&staying)
|
||||
}
|
||||
|
||||
func TestAbandonedChunkDownloadKeepsRunningForWaitingReader(t *testing.T) {
|
||||
content := []byte("gzipped chunk 01")
|
||||
fileId := "3077,02c3475d67aaf564"
|
||||
lookupFn := func(ctx context.Context, fileId string) ([]string, error) {
|
||||
return []string{"http://volume/" + fileId}, nil
|
||||
}
|
||||
rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, nil)
|
||||
defer rc.destroy()
|
||||
|
||||
fetchStarted := make(chan struct{})
|
||||
finishFetch := make(chan struct{})
|
||||
rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) {
|
||||
close(fetchStarted)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return 0, ctx.Err()
|
||||
case <-finishFetch:
|
||||
return copy(buffer, content), nil
|
||||
}
|
||||
}
|
||||
|
||||
var leaving chunkStream
|
||||
leavingCtx, cancel := context.WithCancel(context.Background())
|
||||
leavingDone, waitingDone := make(chan error, 1), make(chan error, 1)
|
||||
go func() {
|
||||
_, err := rc.readChunkAt(leavingCtx, &leaving, make([]byte, len(content)), fileId, nil, true, 0, len(content), false)
|
||||
leavingDone <- err
|
||||
}()
|
||||
<-fetchStarted
|
||||
waitingBuffer := make([]byte, len(content))
|
||||
go func() {
|
||||
_, err := rc.ReadChunkAt(context.Background(), waitingBuffer, fileId, nil, true, 0, len(content), false)
|
||||
waitingDone <- err
|
||||
}()
|
||||
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for {
|
||||
rc.RLock()
|
||||
cacher := rc.downloaders[fileId]
|
||||
rc.RUnlock()
|
||||
if cacher != nil && atomic.LoadInt32(&cacher.readers) == 2 {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("waiting reader never attached to the download")
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
|
||||
cancel()
|
||||
if err := <-leavingDone; err != context.Canceled {
|
||||
t.Fatalf("cancelled read: got %v, want context.Canceled", err)
|
||||
}
|
||||
rc.releaseStream(&leaving)
|
||||
|
||||
close(finishFetch)
|
||||
select {
|
||||
case err := <-waitingDone:
|
||||
if err != nil {
|
||||
t.Fatalf("waiting reader: got %v, want the chunk", err)
|
||||
}
|
||||
if string(waitingBuffer) != string(content) {
|
||||
t.Fatalf("waiting reader read %q, want %q", waitingBuffer, content)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("waiting reader never got its data")
|
||||
}
|
||||
}
|
||||
|
||||
// A download cancelled mid-copy must not cache the partly filled buffer.
|
||||
func TestAbandonedChunkDownloadIsNotCached(t *testing.T) {
|
||||
fileId := "3077,02c3475d67aaf564"
|
||||
lookupFn := func(ctx context.Context, fileId string) ([]string, error) {
|
||||
return []string{"http://volume/" + fileId}, nil
|
||||
}
|
||||
cache := newMockChunkCacheForReaderCache()
|
||||
rc := NewReaderCache(10, cache, lookupFn, nil)
|
||||
defer rc.destroy()
|
||||
|
||||
fetchStarted := make(chan struct{})
|
||||
rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) {
|
||||
close(fetchStarted)
|
||||
<-ctx.Done()
|
||||
// like readEncryptedUrl: the body was read, the copy was skipped, no error
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
var stream chunkStream
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
readDone := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := rc.readChunkAt(ctx, &stream, make([]byte, 100), fileId, nil, true, 0, 100, true)
|
||||
readDone <- err
|
||||
}()
|
||||
|
||||
<-fetchStarted
|
||||
cancel()
|
||||
if err := <-readDone; err != context.Canceled {
|
||||
t.Fatalf("cancelled read: got %v, want context.Canceled", err)
|
||||
}
|
||||
rc.releaseStream(&stream)
|
||||
|
||||
cache.mu.Lock()
|
||||
defer cache.mu.Unlock()
|
||||
if _, cached := cache.data[fileId]; cached {
|
||||
t.Fatal("the partly filled buffer of a cancelled download was cached")
|
||||
}
|
||||
}
|
||||
|
||||
// Other readers can join a re-lookup through FilerClient, so cancelling the download must not cancel it.
|
||||
func TestAbandonedChunkDownloadKeepsItsRelookupAlive(t *testing.T) {
|
||||
fileId := "3077,02c3475d67aaf564"
|
||||
var lookups int32
|
||||
relookupStarted := make(chan struct{})
|
||||
downloadCancelled := make(chan struct{})
|
||||
relookupErr := make(chan error, 1)
|
||||
lookupFn := func(ctx context.Context, fileId string) ([]string, error) {
|
||||
if atomic.AddInt32(&lookups, 1) > 1 {
|
||||
close(relookupStarted)
|
||||
<-downloadCancelled
|
||||
relookupErr <- ctx.Err()
|
||||
}
|
||||
return []string{"http://volume/" + fileId}, nil
|
||||
}
|
||||
rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, &mockCacheInvalidatorForReaderCache{})
|
||||
defer rc.destroy()
|
||||
rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) {
|
||||
return 0, fmt.Errorf("connection reset by peer")
|
||||
}
|
||||
|
||||
var stream chunkStream
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
readDone := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := rc.readChunkAt(ctx, &stream, make([]byte, 100), fileId, nil, true, 0, 100, false)
|
||||
readDone <- err
|
||||
}()
|
||||
|
||||
<-relookupStarted
|
||||
cancel()
|
||||
if err := <-readDone; err != context.Canceled {
|
||||
t.Fatalf("cancelled read: got %v, want context.Canceled", err)
|
||||
}
|
||||
released := make(chan struct{})
|
||||
go func() {
|
||||
rc.releaseStream(&stream)
|
||||
close(released)
|
||||
}()
|
||||
// the cacher leaves the map under the same lock that cancels it
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for {
|
||||
rc.RLock()
|
||||
_, found := rc.downloaders[fileId]
|
||||
rc.RUnlock()
|
||||
if !found {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("releasing the last stream never dropped the download")
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
close(downloadCancelled)
|
||||
|
||||
if err := <-relookupErr; err != nil {
|
||||
t.Fatalf("re-lookup context: got %v, want it independent of the download", err)
|
||||
}
|
||||
select {
|
||||
case <-released:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("releasing the stream never returned")
|
||||
}
|
||||
}
|
||||
|
||||
// Closing the reader cache cancels a stalled prefetch nobody reads instead of waiting for it.
|
||||
func TestReaderCacheDestroyCancelsUnreadPrefetch(t *testing.T) {
|
||||
lookupFn := func(ctx context.Context, fileId string) ([]string, error) {
|
||||
return []string{"http://volume/" + fileId}, nil
|
||||
}
|
||||
rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, nil)
|
||||
fetchStarted := make(chan struct{})
|
||||
rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) {
|
||||
close(fetchStarted)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return 0, ctx.Err()
|
||||
case <-time.After(10 * time.Second):
|
||||
return 0, fmt.Errorf("volume server never answered")
|
||||
}
|
||||
}
|
||||
rc.MaybeCache(&Interval[*ChunkView]{Value: &ChunkView{FileId: "3077,02c3475d67aaf564", ViewSize: 1024, ChunkSize: 1024}}, 1)
|
||||
<-fetchStarted
|
||||
|
||||
destroyed := make(chan struct{})
|
||||
go func() {
|
||||
rc.destroy()
|
||||
close(destroyed)
|
||||
}()
|
||||
select {
|
||||
case <-destroyed:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("destroy waited for a prefetch nobody reads")
|
||||
}
|
||||
}
|
||||
|
||||
// A download destroy dropped while a stream still held it is cancelled once that stream leaves.
|
||||
func TestReaderCacheDestroyCancelsDownloadAfterItsLastStream(t *testing.T) {
|
||||
lookupFn := func(ctx context.Context, fileId string) ([]string, error) {
|
||||
return []string{"http://volume/" + fileId}, nil
|
||||
}
|
||||
rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, nil)
|
||||
fetchStarted := make(chan struct{})
|
||||
fetchEnded := make(chan error, 1)
|
||||
rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) {
|
||||
close(fetchStarted)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
fetchEnded <- ctx.Err()
|
||||
return 0, ctx.Err()
|
||||
case <-time.After(10 * time.Second):
|
||||
fetchEnded <- nil
|
||||
return 0, fmt.Errorf("volume server never answered")
|
||||
}
|
||||
}
|
||||
|
||||
var stream chunkStream
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
readDone := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := rc.readChunkAt(ctx, &stream, make([]byte, 100), "3077,02c3475d67aaf564", nil, true, 0, 4<<20, false)
|
||||
readDone <- err
|
||||
}()
|
||||
<-fetchStarted
|
||||
cancel()
|
||||
if err := <-readDone; err != context.Canceled {
|
||||
t.Fatalf("cancelled read: got %v, want context.Canceled", err)
|
||||
}
|
||||
|
||||
destroyed := make(chan struct{})
|
||||
go func() {
|
||||
rc.destroy()
|
||||
close(destroyed)
|
||||
}()
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for {
|
||||
rc.RLock()
|
||||
mapped := len(rc.downloaders)
|
||||
rc.RUnlock()
|
||||
if mapped == 0 {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("destroy never dropped the download from the map")
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
select {
|
||||
case err := <-fetchEnded:
|
||||
t.Fatalf("destroy cancelled a download a stream still holds: %v", err)
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
|
||||
rc.releaseStream(&stream)
|
||||
select {
|
||||
case <-destroyed:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("destroy kept waiting for a download its last stream left")
|
||||
}
|
||||
if err := <-fetchEnded; err != context.Canceled {
|
||||
t.Fatalf("orphaned download: got %v, want context.Canceled", err)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user