Scan refactor (#1816)

* Add file scanner
* Scan scene changes
* Split scan files
* Generalise scan
* Refactor ffprobe
* Refactor ffmpeg encoder
* Move scene scan code to scene package
* Move matchExtension to utils
* Refactor gallery scanning
* Refactor image scanning
* Prevent race conditions on identical hashes
* Refactor image thumbnail generation
* Perform count concurrently
* Allow progress increment before total set
* Make progress updates more frequent

pkg/utils/file.go

@@ -377,3 +377,16 @@ func FindInPaths(paths []string, baseName string) string {
 
 	return ""
 }
+
+// MatchExtension returns true if the extension of the provided path
+// matches any of the provided extensions.
+func MatchExtension(path string, extensions []string) bool {
+	ext := filepath.Ext(path)
+	for _, e := range extensions {
+		if strings.EqualFold(ext, "."+e) {
+			return true
+		}
+	}
+
+	return false
+}

pkg/utils/mutex.go

@@ -0,0 +1,64 @@
+package utils
+
+// MutexManager manages access to mutexes using a mutex type and key.
+type MutexManager struct {
+	mapChan chan map[string]<-chan struct{}
+}
+
+// NewMutexManager returns a new instance of MutexManager.
+func NewMutexManager() *MutexManager {
+	ret := &MutexManager{
+		mapChan: make(chan map[string]<-chan struct{}, 1),
+	}
+
+	initial := make(map[string]<-chan struct{})
+	ret.mapChan <- initial
+
+	return ret
+}
+
+// Claim blocks until the mutex for the mutexType and key pair is available.
+// The mutex is then claimed by the calling code until the provided done
+// channel is closed.
+func (csm *MutexManager) Claim(mutexType string, key string, done <-chan struct{}) {
+	mapKey := mutexType + "_" + key
+	success := false
+
+	var existing <-chan struct{}
+	for !success {
+		// grab the map
+		m := <-csm.mapChan
+
+		// get the entry for the given key
+		newEntry := m[mapKey]
+
+		// if its the existing entry or nil, then it's available, add our channel
+		if newEntry == nil || newEntry == existing {
+			m[mapKey] = done
+			success = true
+		}
+
+		// return the map
+		csm.mapChan <- m
+
+		// if there is an existing entry, now we can wait for it to
+		// finish, then repeat the process
+		if newEntry != nil {
+			existing = newEntry
+			<-newEntry
+		}
+	}
+
+	// add to goroutine to remove from the map only
+	go func() {
+		<-done
+
+		m := <-csm.mapChan
+
+		if m[mapKey] == done {
+			delete(m, mapKey)
+		}
+
+		csm.mapChan <- m
+	}()
+}

pkg/utils/mutex_test.go

@@ -0,0 +1,50 @@
+package utils
+
+import (
+	"sync"
+	"testing"
+)
+
+// should be run with -race
+func TestMutexManager(t *testing.T) {
+	m := NewMutexManager()
+
+	map1 := make(map[string]bool)
+	map2 := make(map[string]bool)
+	map3 := make(map[string]bool)
+	maps := []map[string]bool{
+		map1,
+		map2,
+		map3,
+	}
+
+	types := []string{
+		"foo",
+		"foo",
+		"bar",
+	}
+
+	const key = "baz"
+
+	const workers = 8
+	const loops = 300
+	var wg sync.WaitGroup
+	for k := 0; k < workers; k++ {
+		wg.Add(1)
+		go func(wk int) {
+			defer wg.Done()
+			for l := 0; l < loops; l++ {
+				func(l int) {
+					c := make(chan struct{})
+					defer close(c)
+
+					m.Claim(types[l%3], key, c)
+
+					maps[l%3][key] = true
+				}(l)
+			}
+		}(k)
+	}
+
+	wg.Wait()
+}

pkg/utils/oshash.go

@@ -4,6 +4,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"io"
 	"os"
 )
 
@@ -41,6 +42,40 @@ func oshash(size int64, head []byte, tail []byte) (string, error) {
 	return fmt.Sprintf("%016x", result), nil
 }
 
+func OSHashFromReader(src io.ReadSeeker, fileSize int64) (string, error) {
+	if fileSize == 0 {
+		return "", nil
+	}
+
+	fileChunkSize := chunkSize
+	if fileSize < fileChunkSize {
+		fileChunkSize = fileSize
+	}
+
+	head := make([]byte, fileChunkSize)
+	tail := make([]byte, fileChunkSize)
+
+	// read the head of the file into the start of the buffer
+	_, err := src.Read(head)
+	if err != nil {
+		return "", err
+	}
+
+	// seek to the end of the file - the chunk size
+	_, err = src.Seek(-fileChunkSize, 2)
+	if err != nil {
+		return "", err
+	}
+
+	// read the tail of the file
+	_, err = src.Read(tail)
+	if err != nil {
+		return "", err
+	}
+
+	return oshash(fileSize, head, tail)
+}
+
 // OSHashFromFilePath calculates the hash using the same algorithm that
 // OpenSubtitles.org uses.
 //
@@ -60,35 +95,5 @@ func OSHashFromFilePath(filePath string) (string, error) {
 
 	fileSize := fi.Size()
 
-	if fileSize == 0 {
-		return "", nil
-	}
-
-	fileChunkSize := chunkSize
-	if fileSize < fileChunkSize {
-		fileChunkSize = fileSize
-	}
-
-	head := make([]byte, fileChunkSize)
-	tail := make([]byte, fileChunkSize)
-
-	// read the head of the file into the start of the buffer
-	_, err = f.Read(head)
-	if err != nil {
-		return "", err
-	}
-
-	// seek to the end of the file - the chunk size
-	_, err = f.Seek(-fileChunkSize, 2)
-	if err != nil {
-		return "", err
-	}
-
-	// read the tail of the file
-	_, err = f.Read(tail)
-	if err != nil {
-		return "", err
-	}
-
-	return oshash(fileSize, head, tail)
+	return OSHashFromReader(f, fileSize)
 }