Cancel a running scan

lib/scanner/blockqueue.go

@@ -19,13 +19,13 @@ import (
 // workers are used in parallel. The outbox will become closed when the inbox
 // is closed and all items handled.
 
-func newParallelHasher(dir string, blockSize, workers int, outbox, inbox chan protocol.FileInfo, counter *int64, done chan struct{}) {
+func newParallelHasher(dir string, blockSize, workers int, outbox, inbox chan protocol.FileInfo, counter *int64, done, cancel chan struct{}) {
 	wg := sync.NewWaitGroup()
 	wg.Add(workers)
 
 	for i := 0; i < workers; i++ {
 		go func() {
-			hashFiles(dir, blockSize, outbox, inbox, counter)
+			hashFiles(dir, blockSize, outbox, inbox, counter, cancel)
 			wg.Done()
 		}()
 	}
@@ -59,19 +59,33 @@ func HashFile(path string, blockSize int, sizeHint int64, counter *int64) ([]pro
 	return Blocks(fd, blockSize, sizeHint, counter)
 }
 
-func hashFiles(dir string, blockSize int, outbox, inbox chan protocol.FileInfo, counter *int64) {
-	for f := range inbox {
-		if f.IsDirectory() || f.IsDeleted() {
-			panic("Bug. Asked to hash a directory or a deleted file.")
-		}
+func hashFiles(dir string, blockSize int, outbox, inbox chan protocol.FileInfo, counter *int64, cancel chan struct{}) {
+	for {
+		select {
+		case f, ok := <-inbox:
+			if !ok {
+				return
+			}
 
-		blocks, err := HashFile(filepath.Join(dir, f.Name), blockSize, f.CachedSize, counter)
-		if err != nil {
-			l.Debugln("hash error:", f.Name, err)
-			continue
-		}
+			if f.IsDirectory() || f.IsDeleted() {
+				panic("Bug. Asked to hash a directory or a deleted file.")
+			}
 
-		f.Blocks = blocks
-		outbox <- f
+			blocks, err := HashFile(filepath.Join(dir, f.Name), blockSize, f.CachedSize, counter)
+			if err != nil {
+				l.Debugln("hash error:", f.Name, err)
+				continue
+			}
+
+			f.Blocks = blocks
+			select {
+			case outbox <- f:
+			case <-cancel:
+				return
+			}
+
+		case <-cancel:
+			return
+		}
 	}
 }

lib/scanner/walk.go

@@ -72,6 +72,8 @@ type Walker struct {
 	// Optional progress tick interval which defines how often FolderScanProgress
 	// events are emitted. Negative number means disabled.
 	ProgressTickIntervalS int
+	// Signals cancel from the outside - when closed, we should stop walking.
+	Cancel chan struct{}
 }
 
 type TempNamer interface {
@@ -121,7 +123,7 @@ func (w *Walker) Walk() (chan protocol.FileInfo, error) {
 	// We're not required to emit scan progress events, just kick off hashers,
 	// and feed inputs directly from the walker.
 	if w.ProgressTickIntervalS < 0 {
-		newParallelHasher(w.Dir, w.BlockSize, w.Hashers, finishedChan, toHashChan, nil, nil)
+		newParallelHasher(w.Dir, w.BlockSize, w.Hashers, finishedChan, toHashChan, nil, nil, w.Cancel)
 		return finishedChan, nil
 	}
 
@@ -149,7 +151,7 @@ func (w *Walker) Walk() (chan protocol.FileInfo, error) {
 
 		realToHashChan := make(chan protocol.FileInfo)
 		done := make(chan struct{})
-		newParallelHasher(w.Dir, w.BlockSize, w.Hashers, finishedChan, realToHashChan, &progress, done)
+		newParallelHasher(w.Dir, w.BlockSize, w.Hashers, finishedChan, realToHashChan, &progress, done, w.Cancel)
 
 		// A routine which actually emits the FolderScanProgress events
 		// every w.ProgressTicker ticks, until the hasher routines terminate.
@@ -168,13 +170,21 @@ func (w *Walker) Walk() (chan protocol.FileInfo, error) {
 						"current": current,
 						"total":   total,
 					})
+				case <-w.Cancel:
+					ticker.Stop()
+					return
 				}
 			}
 		}()
 
+	loop:
 		for _, file := range filesToHash {
 			l.Debugln("real to hash:", file.Name)
-			realToHashChan <- file
+			select {
+			case realToHashChan <- file:
+			case <-w.Cancel:
+				break loop
+			}
 		}
 		close(realToHashChan)
 	}()
@@ -329,7 +339,11 @@ func (w *Walker) walkAndHashFiles(fchan, dchan chan protocol.FileInfo) filepath.
 
 			l.Debugln("symlink changedb:", p, f)
 
-			dchan <- f
+			select {
+			case dchan <- f:
+			case <-w.Cancel:
+				return errors.New("cancelled")
+			}
 
 			return skip
 		}
@@ -363,7 +377,13 @@ func (w *Walker) walkAndHashFiles(fchan, dchan chan protocol.FileInfo) filepath.
 				Modified: mtime.Unix(),
 			}
 			l.Debugln("dir:", p, f)
-			dchan <- f
+
+			select {
+			case dchan <- f:
+			case <-w.Cancel:
+				return errors.New("cancelled")
+			}
+
 			return nil
 		}
 
@@ -406,7 +426,12 @@ func (w *Walker) walkAndHashFiles(fchan, dchan chan protocol.FileInfo) filepath.
 				CachedSize: info.Size(),
 			}
 			l.Debugln("to hash:", p, f)
-			fchan <- f
+
+			select {
+			case fchan <- f:
+			case <-w.Cancel:
+				return errors.New("cancelled")
+			}
 		}
 
 		return nil