test ci

Author: lichuqiang

pkg/controller/source/source.go

@@ -115,10 +115,23 @@ func (cs *ChannelSource) Start(
 				select {
 				case <-cs.stop:
 					// Close destination channels
-					cs.doStop()
+					cs.destLock.Lock()
+					for _, dst := range cs.dest {
+						close(dst)
+					}
+					cs.destLock.Unlock()
 					return
 				case evt := <-cs.Source:
-					cs.distribute(evt)
+					cs.destLock.Lock()
+					for _, dst := range cs.dest {
+						// We cannot make it under goroutine here, or we'll meet the
+						// race condition of writing message to closed channels.
+						// To avoid blocking, the dest channels are expected to be of
+						// proper buffer size. If we still see it blocked, then
+						// the controller is thought to be in an abnormal state.
+						dst <- evt
+					}
+					cs.destLock.Unlock()
 				}
 			}
 		}()
@@ -149,6 +162,7 @@ func (cs *ChannelSource) Start(
 	return nil
 }
 
+/*
 func (cs *ChannelSource) doStop() {
 	cs.destLock.Lock()
 	defer cs.destLock.Unlock()
@@ -157,7 +171,9 @@ func (cs *ChannelSource) doStop() {
 		close(dst)
 	}
 }
+*/
 
+/*
 func (cs *ChannelSource) distribute(evt event.GenericEvent) {
 	cs.destLock.Lock()
 	defer cs.destLock.Unlock()
@@ -171,6 +187,7 @@ func (cs *ChannelSource) distribute(evt event.GenericEvent) {
 		dst <- evt
 	}
 }
+*/
 
 // KindSource is used to provide a source of events originating inside the cluster from Watches (eh.g. Pod Create)
 type KindSource struct {