diff options
| -rw-r--r-- | weed/pb/filer_pb/filer_client_bfs.go | 40 | ||||
| -rw-r--r-- | weed/storage/needle_map_sorted_file.go | 3 |
2 files changed, 25 insertions, 18 deletions
diff --git a/weed/pb/filer_pb/filer_client_bfs.go b/weed/pb/filer_pb/filer_client_bfs.go index 6ae4a50cf..cb9367a7d 100644 --- a/weed/pb/filer_pb/filer_client_bfs.go +++ b/weed/pb/filer_pb/filer_client_bfs.go @@ -9,37 +9,41 @@ import ( ) func TraverseBfs(filerClient FilerClient, parentPath util.FullPath, fn func(parentPath util.FullPath, entry *Entry)) (err error) { - K := 5 var jobQueueWg sync.WaitGroup queue := util.NewQueue() jobQueueWg.Add(1) queue.Enqueue(parentPath) - var isTerminating bool + terminates := make([]chan bool, K) for i := 0; i < K; i++ { - go func() { + terminates[i] = make(chan bool) + go func(j int) { for { - if isTerminating { - break - } - t := queue.Dequeue() - if t == nil { - time.Sleep(329 * time.Millisecond) - continue - } - dir := t.(util.FullPath) - processErr := processOneDirectory(filerClient, dir, queue, &jobQueueWg, fn) - if processErr != nil { - err = processErr + select { + case <-terminates[j]: + return + default: + t := queue.Dequeue() + if t == nil { + time.Sleep(329 * time.Millisecond) + continue + } + dir := t.(util.FullPath) + processErr := processOneDirectory(filerClient, dir, queue, &jobQueueWg, fn) + if processErr != nil { + err = processErr + } + jobQueueWg.Done() } - jobQueueWg.Done() } - }() + }(i) } jobQueueWg.Wait() - isTerminating = true + for i := 0; i < K; i++ { + close(terminates[i]) + } return } diff --git a/weed/storage/needle_map_sorted_file.go b/weed/storage/needle_map_sorted_file.go index c610f0d20..977237f85 100644 --- a/weed/storage/needle_map_sorted_file.go +++ b/weed/storage/needle_map_sorted_file.go @@ -94,6 +94,9 @@ func (m *SortedFileNeedleMap) Delete(key NeedleId, offset Offset) error { } func (m *SortedFileNeedleMap) Close() { + if m == nil { + return + } if m.indexFile != nil { m.indexFile.Close() } |