diff --git a/AUTHORS b/AUTHORS index 267d9bc57..8785e7f69 100644 --- a/AUTHORS +++ b/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) diff --git a/task/list.go b/task/list.go index 6a1a720d7..7afe9d3f4 100644 --- a/task/list.go +++ b/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 diff --git a/task/list_test.go b/task/list_test.go index eb6ebddd1..ecbc59d21 100644 --- a/task/list_test.go +++ b/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)) + } +} diff --git a/task/task.go b/task/task.go index 72f606997..bb8fd1514 100644 --- a/task/task.go +++ b/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