fix race

kubernetes · Apr 10, 2015 · 28c6769 · 28c6769
1 parent 6835318
commit 28c6769
Show file tree

Hide file tree

Showing 3 changed files with 50 additions and 26 deletions.
diff --git a/pkg/controller/framework/controller_test.go b/pkg/controller/framework/controller_test.go
@@ -218,16 +218,17 @@ func TestHammerController(t *testing.T) {
 	go controller.Run(stop)
 
 	wg := sync.WaitGroup{}
-	for i := 0; i < 10; i++ {
-		wg.Add(1)
+	const threads = 3
+	wg.Add(threads)
+	for i := 0; i < threads; i++ {
 		go func() {
 			defer wg.Done()
 			// Let's add a few objects to the source.
 			currentNames := util.StringSet{}
 			rs := rand.NewSource(rand.Int63())
 			f := fuzz.New().NilChance(.5).NumElements(0, 2).RandSource(rs)
 			r := rand.New(rs) // Mustn't use r and f concurrently!
-			for i := 0; i < 750; i++ {
+			for i := 0; i < 100; i++ {
 				var name string
 				var isNew bool
 				if currentNames.Len() == 0 || r.Intn(3) == 1 {
@@ -335,48 +336,45 @@ func TestUpdate(t *testing.T) {
 		}
 	}
 
-	var wg sync.WaitGroup
 	tests := []func(string){
 		func(name string) {
-			defer wg.Done()
-			testDoneWG.Add(1)
 			name = "a-" + name
 			source.Add(pod(name, FROM))
 			source.Modify(pod(name, TO))
 		},
 		func(name string) {
-			defer wg.Done()
-			testDoneWG.Add(1)
 			name = "b-" + name
 			source.Add(pod(name, FROM))
 			source.ModifyDropWatch(pod(name, TO))
 		},
 		func(name string) {
-			defer wg.Done()
-			testDoneWG.Add(1)
 			name = "c-" + name
 			source.AddDropWatch(pod(name, FROM))
 			source.Modify(pod(name, ADD_MISSED))
 			source.Modify(pod(name, TO))
 		},
 		func(name string) {
-			defer wg.Done()
-			testDoneWG.Add(1)
 			name = "d-" + name
 			source.Add(pod(name, FROM))
 		},
 	}
 
 	// run every test a few times, in parallel
 	fuzzer := fuzz.New()
-	for i := 0; i < 20; i++ {
+	const threads = 3
+	var wg sync.WaitGroup
+	wg.Add(threads * len(tests))
+	testDoneWG.Add(threads * len(tests))
+	for i := 0; i < threads; i++ {
 		for _, f := range tests {
-			wg.Add(1)
 			var name string
 			for len(name) < 10 {
 				fuzzer.Fuzz(&name)
 			}
-			go f(name)
+			go func() {
+				defer wg.Done()
+				f(name)
+			}()
 		}
 	}
 	wg.Wait()

diff --git a/pkg/controller/framework/fake_controller_source.go b/pkg/controller/framework/fake_controller_source.go
@@ -23,6 +23,7 @@ import (
 	"sync"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/conversion"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
@@ -125,8 +126,15 @@ func (f *FakeControllerSource) List() (runtime.Object, error) {
 	defer f.lock.RUnlock()
 	list := make([]runtime.Object, 0, len(f.items))
 	for _, obj := range f.items {
-		// TODO: should copy obj first
-		list = append(list, obj)
+		// Must make a copy to allow clients to modify the object.
+		// Otherwise, if they make a change and write it back, they
+		// will inadvertantly change the our canonical copy (in
+		// addition to racing with other clients).
+		objCopy, err := conversion.DeepCopy(obj)
+		if err != nil {
+			return nil, err
+		}
+		list = append(list, objCopy.(runtime.Object))
 	}
 	listObj := &api.List{}
 	if err := runtime.SetList(listObj, list); err != nil {
@@ -151,7 +159,20 @@ func (f *FakeControllerSource) Watch(resourceVersion string) (watch.Interface, e
 		return nil, err
 	}
 	if rc < len(f.changes) {
-		return f.broadcaster.WatchWithPrefix(f.changes[rc:]), nil
+		changes := []watch.Event{}
+		for _, c := range f.changes[rc:] {
+			// Must make a copy to allow clients to modify the
+			// object.  Otherwise, if they make a change and write
+			// it back, they will inadvertantly change the our
+			// canonical copy (in addition to racing with other
+			// clients).
+			objCopy, err := conversion.DeepCopy(c.Object)
+			if err != nil {
+				return nil, err
+			}
+			changes = append(changes, watch.Event{c.Type, objCopy.(runtime.Object)})
+		}
+		return f.broadcaster.WatchWithPrefix(changes), nil
 	} else if rc > len(f.changes) {
 		return nil, errors.New("resource version in the future not supported by this fake")
 	}

diff --git a/pkg/conversion/deep_copy.go b/pkg/conversion/deep_copy.go
@@ -17,22 +17,27 @@ limitations under the License.
 package conversion
 
 import (
+	"bytes"
+	"encoding/gob"
 	"reflect"
 )
 
-var deepCopier = NewConverter()
-
 // DeepCopy makes a deep copy of source. Won't work for any private fields!
 // For nil slices, will return 0-length slices. These are equivilent in
 // basically every way except for the way that reflect.DeepEqual checks.
 func DeepCopy(source interface{}) (interface{}, error) {
-	src := reflect.ValueOf(source)
-	v := reflect.New(src.Type()).Elem()
-	s := &scope{
-		converter: deepCopier,
+	v := reflect.New(reflect.TypeOf(source))
+
+	buff := &bytes.Buffer{}
+	enc := gob.NewEncoder(buff)
+	dec := gob.NewDecoder(buff)
+	err := enc.Encode(source)
+	if err != nil {
+		return nil, err
 	}
-	if err := deepCopier.convert(src, v, s); err != nil {
+	err = dec.Decode(v.Interface())
+	if err != nil {
 		return nil, err
 	}
-	return v.Interface(), nil
+	return v.Elem().Interface(), nil
 }