engine: make the container progress concurrently

Previously the waitForContainerTransition will wait for all the
transitions to be done before the container can move to next state. This
PR changes the waitForContainerTransition to only wait for no more than
one transition to be done so that the first container that done the
transition doesn't need to wait for other containers.

Author: Peng Yin

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## 1.17.3-dev
 * Bug - Fixed a bug where a stale websocket connection could linger [#1310](https://github.com/aws/amazon-ecs-agent/pull/1310)
+* Enhancement - Parallize the container transition in the same task [#1305](https://github.com/aws/amazon-ecs-agent/pull/1306)
 
 ## 1.17.2
 * Enhancement - Update the `amazon-ecs-cni-plugins` to `2018.02.0` [#1272](https://github.com/aws/amazon-ecs-agent/pull/1272)

agent/api/container.go

@@ -153,7 +153,9 @@ type Container struct {
 
 	// AppliedStatus is the status that has been "applied" (e.g., we've called Pull,
 	// Create, Start, or Stop) but we don't yet know that the application was successful.
-	AppliedStatus ContainerStatus
+	// No need to save it in the state file, as agent will synchronize the container status
+	// on restart and for some operation eg: pull, it has to be recalled again.
+	AppliedStatus ContainerStatus `json:"-"`
 	// ApplyingError is an error that occurred trying to transition the container
 	// to its desired state. It is propagated to the backend in the form
 	// 'Name: ErrorString' as the 'reason' field.
@@ -241,12 +243,14 @@ func (c *Container) GetKnownStatus() ContainerStatus {
 	return c.KnownStatusUnsafe
 }
 
-// SetKnownStatus sets the known status of the container
+// SetKnownStatus sets the known status of the container and update the container
+// applied status
 func (c *Container) SetKnownStatus(status ContainerStatus) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
 	c.KnownStatusUnsafe = status
+	c.updateAppliedStatusUnsafe(status)
 }
 
 // GetDesiredStatus gets the desired status of the container
@@ -548,3 +552,38 @@ func (c *Container) GetHealthStatus() HealthStatus {
 
 	return copyHealth
 }
+
+// updateAppliedStatusUnsafe updates the container transitioning status
+func (c *Container) updateAppliedStatusUnsafe(knownStatus ContainerStatus) {
+	if c.AppliedStatus == ContainerStatusNone {
+		return
+	}
+
+	// Check if the container transition has already finished
+	if c.AppliedStatus <= knownStatus {
+		c.AppliedStatus = ContainerStatusNone
+	}
+}
+
+// SetAppliedStatus sets the applied status of container and returns whether
+// the container is already in a transition
+func (c *Container) SetAppliedStatus(status ContainerStatus) bool {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if c.AppliedStatus != ContainerStatusNone {
+		// return false to indicate the set operation failed
+		return false
+	}
+
+	c.AppliedStatus = status
+	return true
+}
+
+// GetAppliedStatus returns the transitioning status of container
+func (c *Container) GetAppliedStatus() ContainerStatus {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
+	return c.AppliedStatus
+}

agent/engine/engine_unix_integ_test.go

@@ -603,17 +603,25 @@ func TestVolumesFromRO(t *testing.T) {
 	}
 	testTask.Containers[1].VolumesFrom = []api.VolumeFrom{{SourceContainer: testTask.Containers[0].Name, ReadOnly: true}}
 	testTask.Containers[1].Command = []string{"touch /data/readonly-fs || exit 42"}
+	// make all the three containers non-essential to make sure all of the
+	// container can be transitioned to running even one of them finished first
+	testTask.Containers[1].Essential = false
 	testTask.Containers[2].VolumesFrom = []api.VolumeFrom{{SourceContainer: testTask.Containers[0].Name}}
 	testTask.Containers[2].Command = []string{"touch /data/notreadonly-fs-1 || exit 42"}
+	testTask.Containers[2].Essential = false
 	testTask.Containers[3].VolumesFrom = []api.VolumeFrom{{SourceContainer: testTask.Containers[0].Name, ReadOnly: false}}
 	testTask.Containers[3].Command = []string{"touch /data/notreadonly-fs-2 || exit 42"}
+	testTask.Containers[3].Essential = false
 
 	go taskEngine.AddTask(testTask)
 
+	verifyTaskIsRunning(stateChangeEvents, testTask)
+	taskEngine.(*DockerTaskEngine).stopContainer(testTask, testTask.Containers[0])
+
 	verifyTaskIsStopped(stateChangeEvents, testTask)
 
 	if testTask.Containers[1].GetKnownExitCode() == nil || *testTask.Containers[1].GetKnownExitCode() != 42 {
-		t.Error("Didn't exit due to failure to touch ro fs as expected: ", *testTask.Containers[1].GetKnownExitCode())
+		t.Error("Didn't exit due to failure to touch ro fs as expected: ", testTask.Containers[1].GetKnownExitCode())
 	}
 	if testTask.Containers[2].GetKnownExitCode() == nil || *testTask.Containers[2].GetKnownExitCode() != 0 {
 		t.Error("Couldn't touch with default of rw")

agent/engine/task_manager.go

@@ -618,8 +618,8 @@ func (mtask *managedTask) progressContainers() {
 	// We've kicked off one or more transitions, wait for them to
 	// complete, but keep reading events as we do. in fact, we have to for
 	// transitions to complete
-	mtask.waitForContainerTransitions(transitions, transitionChange, transitionChangeContainer)
-	seelog.Debugf("Managed task [%s]: done transitioning all containers", mtask.Arn)
+	mtask.waitForContainerTransition(transitions, transitionChange, transitionChangeContainer)
+	seelog.Debugf("Managed task [%s]: wait for container transition done", mtask.Arn)
 
 	// update the task status
 	changed := mtask.UpdateStatus()
@@ -663,16 +663,30 @@ func (mtask *managedTask) startContainerTransitions(transitionFunc containerTran
 			reasons = append(reasons, transition.reason)
 			continue
 		}
+
+		// If the container is already in a transition, skip
+		if transition.actionRequired && !cont.SetAppliedStatus(transition.nextState) {
+			// At least one container is able to be moved forwards, so we're not deadlocked
+			anyCanTransition = true
+			continue
+		}
+
 		// At least one container is able to be moved forwards, so we're not deadlocked
 		anyCanTransition = true
 
 		if !transition.actionRequired {
-			mtask.handleContainerChange(dockerContainerChange{
-				container: cont,
-				event: DockerContainerChangeEvent{
-					Status: transition.nextState,
-				},
-			})
+			// Updating the container status without calling any docker API, send in
+			// a goroutine so that it won't block here before the waitForContainerTransition
+			// was called after this function. And all the events sent to mtask.dockerMessages
+			// will be handled by handleContainerChange.
+			go func(cont *api.Container, status api.ContainerStatus) {
+				mtask.dockerMessages <- dockerContainerChange{
+					container: cont,
+					event: DockerContainerChangeEvent{
+						Status: status,
+					},
+				}
+			}(cont, transition.nextState)
 			continue
 		}
 		transitions[cont.Name] = transition.nextState
@@ -763,39 +777,23 @@ func (mtask *managedTask) onContainersUnableToTransitionState() {
 		mtask.emitTaskEvent(mtask.Task, taskUnableToTransitionToStoppedReason)
 		// TODO we should probably panic here
 	} else {
-		seelog.Criticalf("Managed task [%s]: voving task to stopped due to bad state", mtask.Arn)
+		seelog.Criticalf("Managed task [%s]: moving task to stopped due to bad state", mtask.Arn)
 		mtask.handleDesiredStatusChange(api.TaskStopped, 0)
 	}
 }
 
-func (mtask *managedTask) waitForContainerTransitions(transitions map[string]api.ContainerStatus,
+func (mtask *managedTask) waitForContainerTransition(transitions map[string]api.ContainerStatus,
 	transitionChange <-chan struct{},
 	transitionChangeContainer <-chan string) {
-
-	for len(transitions) > 0 {
-		if mtask.waitEvent(transitionChange) {
-			changedContainer := <-transitionChangeContainer
-			seelog.Debugf("Managed task [%s]: transition for container[%s] finished",
-				mtask.Arn, changedContainer)
-			delete(transitions, changedContainer)
-			seelog.Debugf("Managed task [%s]: still waiting for: %v", mtask.Arn, transitions)
-		}
-		if mtask.GetDesiredStatus().Terminal() || mtask.GetKnownStatus().Terminal() {
-			allWaitingOnPulled := true
-			for _, desired := range transitions {
-				if desired != api.ContainerPulled {
-					allWaitingOnPulled = false
-				}
-			}
-			if allWaitingOnPulled {
-				// We don't actually care to wait for 'pull' transitions to finish if
-				// we're just heading to stopped since those resources aren't
-				// inherently linked to this task anyways for e.g. gc and so on.
-				seelog.Debugf("Managed task [%s]: all containers are waiting for pulled transition; exiting early: %v",
-					mtask.Arn, transitions)
-				break
-			}
-		}
+	// There could be multiple transitions, but we just need to wait for one of them
+	// to ensure that there is at least one container can be processed in the next
+	// progressContainers. This is done by waiting for one transition/acs/docker message.
+	if mtask.waitEvent(transitionChange) {
+		changedContainer := <-transitionChangeContainer
+		seelog.Debugf("Managed task [%s]: transition for container[%s] finished",
+			mtask.Arn, changedContainer)
+		delete(transitions, changedContainer)
+		seelog.Debugf("Managed task [%s]: still waiting for: %v", mtask.Arn, transitions)
 	}
 }
 

agent/engine/task_manager_test.go

@@ -693,6 +693,7 @@ func TestStartContainerTransitionsInvokesHandleContainerChange(t *testing.T) {
 		},
 		stateChangeEvents:          stateChangeEvents,
 		containerChangeEventStream: containerChangeEventStream,
+		dockerMessages:             make(chan dockerContainerChange),
 	}
 
 	eventsGenerated := sync.WaitGroup{}
@@ -716,6 +717,7 @@ func TestStartContainerTransitionsInvokesHandleContainerChange(t *testing.T) {
 		eventsGenerated.Done()
 	}()
 
+	go task.waitEvent(nil)
 	canTransition, transitions, _ := task.startContainerTransitions(
 		func(cont *api.Container, nextStatus api.ContainerStatus) {
 			t.Error("Invalid code path. The transition function should not be invoked when transitioning container from CREATED -> STOPPED")
@@ -747,7 +749,7 @@ func TestWaitForContainerTransitionsForNonTerminalTask(t *testing.T) {
 
 	// populate the transitions map with transitions for two
 	// containers. We expect two sets of events to be consumed
-	// by `waitForContainerTransitions`
+	// by `waitForContainerTransition`
 	transitions := make(map[string]api.ContainerStatus)
 	transitions[firstContainerName] = api.ContainerRunning
 	transitions[secondContainerName] = api.ContainerRunning
@@ -762,13 +764,13 @@ func TestWaitForContainerTransitionsForNonTerminalTask(t *testing.T) {
 		transitionChangeContainer <- firstContainerName
 	}()
 
-	// waitForContainerTransitions will block until it receives events
+	// waitForContainerTransition will block until it receives events
 	// sent by the go routine defined above
-	task.waitForContainerTransitions(transitions, transitionChange, transitionChangeContainer)
+	task.waitForContainerTransition(transitions, transitionChange, transitionChangeContainer)
 }
 
 // TestWaitForContainerTransitionsForTerminalTask verifies that the
-// `waitForContainerTransitions` method doesn't wait for any container
+// `waitForContainerTransition` method doesn't wait for any container
 // transitions when the task's desired status is STOPPED and if all
 // containers in the task are in PULLED state
 func TestWaitForContainerTransitionsForTerminalTask(t *testing.T) {
@@ -796,13 +798,13 @@ func TestWaitForContainerTransitionsForTerminalTask(t *testing.T) {
 	transitions[secondContainerName] = api.ContainerPulled
 
 	// Event though there are two keys in the transitions map, send
-	// only one event. This tests that `waitForContainerTransitions` doesn't
+	// only one event. This tests that `waitForContainerTransition` doesn't
 	// block to receive two events and will still progress
 	go func() {
 		transitionChange <- struct{}{}
 		transitionChangeContainer <- secondContainerName
 	}()
-	task.waitForContainerTransitions(transitions, transitionChange, transitionChangeContainer)
+	task.waitForContainerTransition(transitions, transitionChange, transitionChangeContainer)
 }
 
 func TestOnContainersUnableToTransitionStateForDesiredStoppedTask(t *testing.T) {

agent/statemanager/state_manager.go

@@ -59,7 +59,9 @@ const (
 	//   d) Added task cgroup related fields ('CPU', 'Memory', 'MemoryCPULimitsEnabled') to 'api.Task'
 	// 9) Add 'ipToTask' map to state file
 	// 10) Add 'healthCheckType' field in 'api.Container'
-	// 11) Add 'PrivateDNSName' field to 'api.ENI'
+	// 11)
+	//  a) Add 'PrivateDNSName' field to 'api.ENI'
+	//  b)Remove `AppliedStatus` field form 'api.Container'
 	ECSDataVersion = 11
 
 	// ecsDataFile specifies the filename in the ECS_DATADIR