Merge branch 'concurrent-tools' into 'main'

Add concurrent processing to I/O-intensive tools

See merge request !22

cmd/archive-files/main.go

@@ -14,6 +14,7 @@ import (
 var (
 	target      = flag.String("to", "", "target repository name")
 	move        = flag.Bool("move", false, "delete original files")
+	concurrency = flag.Int("concurrency", 10, "concurrent workers")
 
 	docstring = `
 
@@ -137,7 +138,7 @@ func main() {
 	}
 
 	// Move files into the repository (that aren't already there).
-	err = coll.Each(func(obj model.Object) error {
+	err = collection.ConcurrentEach(coll, *concurrency, func(obj model.Object) error {
 		return archive(coll, fs, repo, obj)
 	})
 

cmd/generate-thumbnails/main.go

@@ -18,6 +18,7 @@ var (
 	target      = flag.String("to", "", "target repository name")
 	width       = flag.Int("width", 100, "width")
 	height      = flag.Int("height", 100, "height")
+	concurrency = flag.Int("concurrency", 10, "concurrent workers")
 
 	docstring = `
 
@@ -103,7 +104,7 @@ func main() {
 	}
 
 	// Move files into the repository (that aren't already there).
-	err = coll.Each(func(obj model.Object) error {
+	err = collection.ConcurrentEach(coll, *concurrency, func(obj model.Object) error {
 		return mkThumbnail(coll, fs, repo, obj)
 	})
 

cmd/refine-metadata/main.go

@@ -21,6 +21,7 @@ import (
 
 var (
 	force        = flag.Bool("force", false, "always request remote metadata")
+	concurrency  = flag.Int("concurrency", 10, "concurrent workers")
 	enableOL     = flag.Bool("enable-openlibrary", true, "enable the OpenLibrary metadata source")
 	enableGB     = flag.Bool("enable-googlebooks", true, "enable the Google Books metadata source")
 	enableSqlite = flag.String("enable-sqlite", "", "enable the generic SQLite source (argument is config file)")
@@ -128,7 +129,7 @@ func main() {
 	}
 
 	// Refine all Book objects in the collection.
-	err = coll.Each(func(obj model.Object) error {
+	err = collection.ConcurrentEach(coll, *concurrency, func(obj model.Object) error {
 		if obj.GetID().Class != model.ClassBook {
 			return nil
 		}

collection/parallel.go

+package collection
+
+import (
+	"io"
+	"sync"
+
+	"git.autistici.org/ale/ebooks/meta/model"
+)
+
+func ConcurrentEach(coll Collection, concurrency int, f func(model.Object) error) error {
+	ch := make(chan model.Object, concurrency)
+	errCh := make(chan error, 1)
+	defer close(errCh)
+
+	var wg sync.WaitGroup
+	for i := 0; i < concurrency; i++ {
+		wg.Add(1)
+		go func() {
+			for obj := range ch {
+				if err := f(obj); err != nil {
+					select {
+					case errCh <- err:
+					default:
+					}
+				}
+			}
+			wg.Done()
+		}()
+	}
+
+	stop := make(chan bool)
+	doneCh := make(chan error)
+	defer close(doneCh)
+	go func() {
+		err := coll.Each(func(obj model.Object) error {
+			select {
+			case <-stop:
+				return io.EOF
+			default:
+			}
+			ch <- obj
+			return nil
+		})
+		doneCh <- err
+		close(ch)
+	}()
+
+	var err error
+	select {
+	case err = <-doneCh:
+	case err = <-errCh:
+	}
+
+	close(stop)
+	wg.Wait()
+
+	return err
+}