diff --git a/api/graphql/resolvers/scanner.go b/api/graphql/resolvers/scanner.go index f14c48dd..788994a1 100644 --- a/api/graphql/resolvers/scanner.go +++ b/api/graphql/resolvers/scanner.go @@ -3,7 +3,9 @@ package resolvers import ( "context" + "github.com/pkg/errors" "github.com/viktorstrate/photoview/api/graphql/models" + "github.com/viktorstrate/photoview/api/scanner" ) func (r *mutationResolver) ScanAll(ctx context.Context) (*models.ScannerResult, error) { @@ -36,12 +38,18 @@ func (r *mutationResolver) ScanUser(ctx context.Context, userID int) (*models.Sc // }, nil // } - // startMessage := "Scanner started" + row := r.Database.QueryRow("SELECT * FROM user WHERE user_id = ?", userID) + user, err := models.NewUserFromRow(row) + if err != nil { + return nil, errors.Wrap(err, "get user from database") + } - // return &models.ScannerResult{ - // Finished: false, - // Success: true, - // Message: &startMessage, - // }, nil - panic("not implemented") + scanner.AddUserToQueue(user) + + startMessage := "Scanner started" + return &models.ScannerResult{ + Finished: false, + Success: true, + Message: &startMessage, + }, nil } diff --git a/api/scanner/cache.go b/api/scanner/cache.go index ce2be92d..47a1da82 100644 --- a/api/scanner/cache.go +++ b/api/scanner/cache.go @@ -1,10 +1,14 @@ package scanner -import "path" +import ( + "path" + "sync" +) type AlbumScannerCache struct { path_contains_photos map[string]bool photo_types map[string]ImageType + mutex sync.Mutex } func MakeAlbumCache() *AlbumScannerCache { @@ -16,6 +20,8 @@ func MakeAlbumCache() *AlbumScannerCache { // Insert single album directory in cache func (c *AlbumScannerCache) InsertAlbumPath(path string, contains_photo bool) { + c.mutex.Lock() + defer c.mutex.Unlock() c.path_contains_photos[path] = contains_photo } @@ -24,6 +30,9 @@ func (c *AlbumScannerCache) InsertAlbumPaths(end_path string, root string, conta curr_path := path.Clean(end_path) root_path := path.Clean(root) + c.mutex.Lock() + defer c.mutex.Unlock() + for curr_path != root_path || curr_path == "." { c.InsertAlbumPath(curr_path, contains_photo) @@ -33,6 +42,9 @@ func (c *AlbumScannerCache) InsertAlbumPaths(end_path string, root string, conta } func (c *AlbumScannerCache) AlbumContainsPhotos(path string) *bool { + c.mutex.Lock() + defer c.mutex.Unlock() + contains_photo, found := c.path_contains_photos[path] if found { // log.Printf("Album cache hit: %s\n", path) @@ -43,10 +55,16 @@ func (c *AlbumScannerCache) AlbumContainsPhotos(path string) *bool { } func (c *AlbumScannerCache) InsertPhotoType(path string, content_type ImageType) { + c.mutex.Lock() + defer c.mutex.Unlock() + (c.photo_types)[path] = content_type } func (c *AlbumScannerCache) GetPhotoType(path string) *ImageType { + c.mutex.Lock() + defer c.mutex.Unlock() + result, found := c.photo_types[path] if found { // log.Printf("Image cache hit: %s\n", path) diff --git a/api/scanner/queue.go b/api/scanner/queue.go index f73d1a60..1ca5cce2 100644 --- a/api/scanner/queue.go +++ b/api/scanner/queue.go @@ -46,21 +46,25 @@ func InitializeScannerQueue(db *sql.DB) { func (queue *ScannerQueue) startBackgroundWorker() { for { + log.Println("Queue waiting") <-queue.idle_chan + log.Println("Queue waiting for lock") queue.mutex.Lock() - defer queue.mutex.Unlock() + log.Println("Queue running") for len(queue.in_progress) < queue.settings.max_concurrent_tasks && len(queue.up_next) > 0 { + log.Println("Queue starting job") nextJob := queue.up_next[0] queue.up_next = queue.up_next[1:] queue.in_progress = append(queue.in_progress, nextJob) go func() { + log.Println("Starting job") nextJob.Run(queue.db) - queue.mutex.Lock() - defer queue.mutex.Unlock() + log.Println("Job finished") // Delete finished job from queue + queue.mutex.Lock() for i, x := range queue.in_progress { if x == nextJob { queue.in_progress[i] = queue.in_progress[len(queue.in_progress)-1] @@ -68,15 +72,20 @@ func (queue *ScannerQueue) startBackgroundWorker() { break } } + queue.mutex.Unlock() - queue.Notify() + queue.notify() }() } + + log.Printf("Waiting jobs: %d\n", len(queue.up_next)) + + queue.mutex.Unlock() } } // Notifies the queue that the jobs has changed -func (queue *ScannerQueue) Notify() bool { +func (queue *ScannerQueue) notify() bool { select { case queue.idle_chan <- true: return true @@ -85,21 +94,21 @@ func (queue *ScannerQueue) Notify() bool { } } -func (queue *ScannerQueue) ScanUser(user *models.User) { +func AddUserToQueue(user *models.User) { album_cache := MakeAlbumCache() - albums, album_errors := findAlbumsForUser(queue.db, user, album_cache) + albums, album_errors := findAlbumsForUser(global_scanner_queue.db, user, album_cache) for _, err := range album_errors { log.Printf("User scanner error: %s", err) } - queue.mutex.Lock() + global_scanner_queue.mutex.Lock() for _, album := range albums { - queue.addJob(&ScannerJob{ + global_scanner_queue.addJob(&ScannerJob{ album: album, cache: album_cache, }) } - queue.mutex.Unlock() + global_scanner_queue.mutex.Unlock() } // Queue should be locked prior to calling this function @@ -108,7 +117,7 @@ func (queue *ScannerQueue) addJob(job *ScannerJob) error { return err } queue.up_next = append(queue.up_next, *job) - queue.Notify() + queue.notify() return nil } diff --git a/api/scanner/scanner_album.go b/api/scanner/scanner_album.go index 7084c2bb..07f4af63 100644 --- a/api/scanner/scanner_album.go +++ b/api/scanner/scanner_album.go @@ -24,17 +24,23 @@ func scanAlbum(album *models.Album, cache *AlbumScannerCache, db *sql.DB) { ScannerError("Failed to find photos for album (%s): %s", album.Path, err) } - tx, err := db.Begin() - if err != nil { - ScannerError("Failed to begin database transaction: %s", err) - } - for _, photo := range albumPhotos { + tx, err := db.Begin() + if err != nil { + ScannerError("Failed to begin database transaction: %s", err) + } + err = ProcessPhoto(tx, photo) if err != nil { + tx.Rollback() ScannerError("Failed to process photo (%s): %s", photo.Path, err) } + err = tx.Commit() + if err != nil { + ScannerError("Failed to commit database transaction: %s", err) + } + // TODO: Broadcast progress } } @@ -60,7 +66,7 @@ func findPhotosForAlbum(album *models.Album, cache *AlbumScannerCache, db *sql.D photo, isNewPhoto, err := ScanPhoto(tx, photoPath, album.AlbumID) if err != nil { - ScannerError("Scanning image %s: %s", photoPath, err) + ScannerError("Scanning image (%s): %s", photoPath, err) tx.Rollback() continue } diff --git a/api/scanner/scanner_user.go b/api/scanner/scanner_user.go index 36817ad5..9b42d0b6 100644 --- a/api/scanner/scanner_user.go +++ b/api/scanner/scanner_user.go @@ -164,7 +164,7 @@ func deleteOldUserAlbums(db *sql.DB, scannedAlbums []*models.Album, user *models albumPaths := make([]interface{}, len(scannedAlbums)) for i, album := range scannedAlbums { - albumPaths[i] = album.AlbumID + albumPaths[i] = album.Path } // Delete old albums diff --git a/api/server.go b/api/server.go index f47dc871..7d4dc83c 100644 --- a/api/server.go +++ b/api/server.go @@ -13,6 +13,7 @@ import ( "github.com/viktorstrate/photoview/api/database" "github.com/viktorstrate/photoview/api/graphql/auth" "github.com/viktorstrate/photoview/api/routes" + "github.com/viktorstrate/photoview/api/scanner" "github.com/viktorstrate/photoview/api/server" "github.com/viktorstrate/photoview/api/utils" @@ -40,6 +41,8 @@ func main() { log.Panicf("Could not migrate database: %s\n", err) } + scanner.InitializeScannerQueue(db) + rootRouter := mux.NewRouter() rootRouter.Use(auth.Middleware(db))