tasklist: start every unblocked task when a task finishes

AUTHORS

@@ -89,3 +89,4 @@ List of contributors, in chronological order:
 * James Munson (https://github.com/jmunson)
 * Luan Dang (https://github.com/luanmdang)
 * Martin Simon (https://github.com/barnumbirr)
+* Nathan Gray (https://github.com/NathanGray-ChurchillNavigation)

task/list.go

@@ -72,23 +72,23 @@ func (list *List) consumer() {
 					task.wgTask.Done()
 					list.wg.Done()
 
-					unlocked := false
+					// start every idle task whose resources are now free
+					var ready []*Task
 					for _, t := range list.tasks {
-						if t.State == IDLE {
+						if t.State == IDLE && !t.queued {
 							// check resources
 							blockingTasks := list.usedResources.UsedBy(t.Resources)
 							if len(blockingTasks) == 0 {
 								list.usedResources.MarkInUse(t.Resources, t)
-								// unlock list since queueing may block
-								list.Unlock()
-								unlocked = true
-								list.queue <- t
-								break
+								t.queued = true
+								ready = append(ready, t)
 							}
 						}
 					}
-					if !unlocked {
-						list.Unlock()
+					// unlock list since queueing may block
+					list.Unlock()
+					for _, t := range ready {
+						list.queue <- t
 					}
 				}
 			}()
@@ -220,6 +220,7 @@ func (list *List) RunTaskInBackground(name string, resources []string, process P
 	tasks := list.usedResources.UsedBy(resources)
 	if len(tasks) == 0 {
 		list.usedResources.MarkInUse(task.Resources, task)
+		task.queued = true
 		// queueing task might block if channel not ready, unlock list before queueing
 		list.Unlock()
 		list.queue <- task

task/list_test.go

@@ -2,6 +2,9 @@ package task
 
 import (
 	"errors"
+	"fmt"
+	"sync/atomic"
+	"time"
 
 	"github.com/aptly-dev/aptly/aptly"
 
@@ -52,3 +55,72 @@ func (s *ListSuite) TestList(c *check.C) {
 	c.Check(deleteErr, check.IsNil)
 	list.Stop()
 }
+
+// A holds r1 and r2; B (r1) and C (r2) queue behind it, D (r1) behind B.
+// When A finishes, C must not wait for the long-running B, while D still does.
+func (s *ListSuite) TestAllUnblockedTasksStart(c *check.C) {
+	list := NewList()
+	releaseA := make(chan struct{})
+	releaseB := make(chan struct{})
+	defer func() {
+		close(releaseB)
+		list.Wait()
+		list.Stop()
+	}()
+
+	a, _ := list.RunTaskInBackground("A", []string{"r1", "r2"}, func(aptly.Progress, *Detail) (*ProcessReturnValue, error) {
+		<-releaseA
+		return nil, nil
+	})
+	_, _ = list.RunTaskInBackground("B", []string{"r1"}, func(aptly.Progress, *Detail) (*ProcessReturnValue, error) {
+		<-releaseB
+		return nil, nil
+	})
+	taskC, _ := list.RunTaskInBackground("C", []string{"r2"}, func(aptly.Progress, *Detail) (*ProcessReturnValue, error) {
+		return nil, nil
+	})
+	taskD, _ := list.RunTaskInBackground("D", []string{"r1"}, func(aptly.Progress, *Detail) (*ProcessReturnValue, error) {
+		return nil, nil
+	})
+
+	close(releaseA)
+	_, _ = list.WaitForTaskByID(a.ID)
+
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if t, _ := list.GetTaskByID(taskC.ID); t.State == SUCCEEDED {
+			t, _ = list.GetTaskByID(taskD.ID)
+			c.Check(t.State, check.Equals, IDLE)
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t, _ := list.GetTaskByID(taskC.ID)
+	c.Fatalf("C still in state %d 2s after its resources were freed", t.State)
+}
+
+// Resource-free tasks waiting behind a global holder are released together;
+// each must run once even when earlier ones finish and rescan before the rest are received.
+func (s *ListSuite) TestReleasedTasksRunOnce(c *check.C) {
+	list := NewList()
+	defer list.Stop()
+	releaseA := make(chan struct{})
+
+	_, _ = list.RunTaskInBackground("A", []string{AllResourcesKey}, func(aptly.Progress, *Detail) (*ProcessReturnValue, error) {
+		<-releaseA
+		return nil, nil
+	})
+	var runs [50]int32
+	for i := range runs {
+		_, _ = list.RunTaskInBackground(fmt.Sprintf("C%d", i), nil, func(aptly.Progress, *Detail) (*ProcessReturnValue, error) {
+			atomic.AddInt32(&runs[i], 1)
+			return nil, nil
+		})
+	}
+
+	close(releaseA)
+	list.Wait()
+	for i := range runs {
+		c.Check(atomic.LoadInt32(&runs[i]), check.Equals, int32(1), check.Commentf("C%d", i))
+	}
+}

task/task.go

@@ -54,6 +54,7 @@ type Task struct {
 	State              State
 	Resources          []string
 	wgTask             *sync.WaitGroup
+	queued             bool // handed to the consumer, so it must not be queued again
 }
 
 // NewTask creates new task