Merge pull request #331 from hearchco/as/fix/algo

fix(search): forgot a case when no workers return for a origin

src/config/defaults_cat_general.go

@@ -43,6 +43,6 @@ func generalRanking() CategoryRanking {
 }
 
 var generalTimings = CategoryTimings{
-	PreferredTimeout: 700 * time.Millisecond,
-	HardTimeout:      3 * time.Second,
+	PreferredTimeout: 500 * time.Millisecond,
+	HardTimeout:      1500 * time.Millisecond,
 }

src/config/defaults_cat_images.go

@@ -27,6 +27,6 @@ func imagesRanking() CategoryRanking {
 }
 
 var imagesTimings = CategoryTimings{
-	PreferredTimeout: 700 * time.Millisecond,
-	HardTimeout:      3 * time.Second,
+	PreferredTimeout: 500 * time.Millisecond,
+	HardTimeout:      1500 * time.Millisecond,
 }

src/search/run_origins.go

@@ -0,0 +1,36 @@
+package search
+
+import (
+	"sync"
+)
+
+// Waits on either c.Wait() or wg.Wait() to do final.Done().
+func waitForSuccessOrFinish(c *sync.Cond, wg *sync.WaitGroup, final *sync.WaitGroup) {
+	defer final.Done()
+	d := sync.Cond{L: &sync.Mutex{}}
+
+	// Wait for signal from any successful worker.
+	go func() {
+		c.L.Lock()
+		c.Wait()
+		c.L.Unlock()
+
+		d.L.Lock()
+		d.Signal()
+		d.L.Unlock()
+	}()
+
+	// Wait for all workers to finish (even if it's unsuccessful).
+	go func() {
+		wg.Wait()
+
+		d.L.Lock()
+		d.Signal()
+		d.L.Unlock()
+	}()
+
+	// Whichever of the above finishes first, signal the final wait group.
+	d.L.Lock()
+	d.Wait()
+	d.L.Unlock()
+}

src/search/run_preferred_origins.go

@@ -37,18 +37,19 @@ func runPreferredByOriginEngines(enginers []scraper.Enginer, wgPreferredByOrigin
 			continue
 		}
 
-		c := sync.Cond{L: &sync.Mutex{}}
-		go func() {
-			c.L.Lock()
-			c.Wait()
-			c.L.Unlock()
-			wgPreferredByOriginEngines.Done()
-		}()
+		var wgWorkers sync.WaitGroup
+		wgWorkers.Add(len(workers))
+		successOrigin := sync.Cond{L: &sync.Mutex{}}
+		go waitForSuccessOrFinish(&successOrigin, &wgWorkers, wgPreferredByOriginEngines)
+
 		for _, engName := range workers {
 			enginer := enginers[engName]
 			resChan := make(chan result.ResultScraped, 100)
 			engChan <- resChan
 			go func() {
+				// Indicate that the worker is done, successful or not.
+				defer wgWorkers.Done()
+
 				searchOnce[engName].Do(func() {
 					log.Trace().
 						Str("engine", engName.String()).
@@ -80,10 +81,11 @@ func runPreferredByOriginEngines(enginers []scraper.Enginer, wgPreferredByOrigin
 					}
 				})
 
+				// Indicate that the worker was successful.
 				if searchOnce[engName].Success() {
-					c.L.Lock()
-					c.Signal()
-					c.L.Unlock()
+					successOrigin.L.Lock()
+					successOrigin.Signal()
+					successOrigin.L.Unlock()
 				}
 			}()
 		}

src/search/run_required_origins.go

@@ -37,18 +37,19 @@ func runRequiredByOriginEngines(enginers []scraper.Enginer, wgRequiredByOriginEn
 			continue
 		}
 
-		c := sync.Cond{L: &sync.Mutex{}}
-		go func() {
-			c.L.Lock()
-			c.Wait()
-			c.L.Unlock()
-			wgRequiredByOriginEngines.Done()
-		}()
+		var wgWorkers sync.WaitGroup
+		wgWorkers.Add(len(workers))
+		successOrigin := sync.Cond{L: &sync.Mutex{}}
+		go waitForSuccessOrFinish(&successOrigin, &wgWorkers, wgRequiredByOriginEngines)
+
 		for _, engName := range workers {
 			enginer := enginers[engName]
 			resChan := make(chan result.ResultScraped, 100)
 			engChan <- resChan
 			go func() {
+				// Indicate that the worker is done, successful or not.
+				defer wgWorkers.Done()
+
 				searchOnce[engName].Do(func() {
 					log.Trace().
 						Str("engine", engName.String()).
@@ -79,10 +80,12 @@ func runRequiredByOriginEngines(enginers []scraper.Enginer, wgRequiredByOriginEn
 						searchOnce[engName].Scraped()
 					}
 				})
+
+				// Indicate that the worker was successful.
 				if searchOnce[engName].Success() {
-					c.L.Lock()
-					c.Signal()
-					c.L.Unlock()
+					successOrigin.L.Lock()
+					successOrigin.Signal()
+					successOrigin.L.Unlock()
 				}
 			}()
 		}