Make individual controller actions asynchronous.

pkg/controller/replication_controller.go

@@ -19,6 +19,7 @@ package controller
 import (
 	"encoding/json"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
@@ -182,15 +183,27 @@ func (rm *ReplicationManager) syncReplicationController(controllerSpec api.Repli
 	diff := len(filteredList) - controllerSpec.DesiredState.Replicas
 	if diff < 0 {
 		diff *= -1
+		wait := sync.WaitGroup{}
+		wait.Add(diff)
 		glog.Infof("Too few replicas, creating %d\n", diff)
 		for i := 0; i < diff; i++ {
+			go func() {
+				defer wait.Done()
 				rm.podControl.createReplica(controllerSpec)
+			}()
 		}
+		wait.Wait()
 	} else if diff > 0 {
 		glog.Infof("Too many replicas, deleting %d\n", diff)
+		wait := sync.WaitGroup{}
+		wait.Add(diff)
 		for i := 0; i < diff; i++ {
-			rm.podControl.deletePod(filteredList[i].ID)
+			go func(ix int) {
+				defer wait.Done()
+				rm.podControl.deletePod(filteredList[ix].ID)
+			}(i)
 		}
+		wait.Wait()
 	}
 	return nil
 }

pkg/controller/replication_controller_test.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"net/http/httptest"
 	"reflect"
+	"sync"
 	"testing"
 	"time"
 
@@ -41,13 +42,18 @@ func makeURL(suffix string) string {
 type FakePodControl struct {
 	controllerSpec []api.ReplicationController
 	deletePodID    []string
+	lock           sync.Mutex
 }
 
 func (f *FakePodControl) createReplica(spec api.ReplicationController) {
+	f.lock.Lock()
+	defer f.lock.Unlock()
 	f.controllerSpec = append(f.controllerSpec, spec)
 }
 
 func (f *FakePodControl) deletePod(podID string) error {
+	f.lock.Lock()
+	defer f.lock.Unlock()
 	f.deletePodID = append(f.deletePodID, podID)
 	return nil
 }