Generalize ErrorChannel to other underlying types

pkg/scheduler/framework/parallelize/error_channel.go

@@ -1,59 +0,0 @@
-/*
-Copyright 2019 The Kubernetes Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package parallelize
-
-import "context"
-
-// ErrorChannel supports non-blocking send and receive operation to capture error.
-// A maximum of one error is kept in the channel and the rest of the errors sent
-// are ignored, unless the existing error is received and the channel becomes empty
-// again.
-type ErrorChannel struct {
-	errCh chan error
-}
-
-// SendError sends an error without blocking the sender.
-func (e *ErrorChannel) SendError(err error) {
-	select {
-	case e.errCh <- err:
-	default:
-	}
-}
-
-// SendErrorWithCancel sends an error without blocking the sender and calls
-// cancel function.
-func (e *ErrorChannel) SendErrorWithCancel(err error, cancel context.CancelFunc) {
-	e.SendError(err)
-	cancel()
-}
-
-// ReceiveError receives an error from channel without blocking on the receiver.
-func (e *ErrorChannel) ReceiveError() error {
-	select {
-	case err := <-e.errCh:
-		return err
-	default:
-		return nil
-	}
-}
-
-// NewErrorChannel returns a new ErrorChannel.
-func NewErrorChannel() *ErrorChannel {
-	return &ErrorChannel{
-		errCh: make(chan error, 1),
-	}
-}

pkg/scheduler/framework/parallelize/result_channel.go

@@ -0,0 +1,60 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package parallelize
+
+import "context"
+
+// ResultChannel supports non-blocking send and receive operation to capture result.
+// A maximum of one result is kept in the channel and the rest of the results sent
+// are ignored, unless the existing result is received and the channel becomes empty
+// again.
+type ResultChannel[T comparable] struct {
+	ch chan T
+}
+
+// Send sends a result without blocking the sender.
+func (e *ResultChannel[T]) Send(result T) {
+	select {
+	case e.ch <- result:
+	default:
+	}
+}
+
+// SendWithCancel sends a result without blocking the sender and calls
+// cancel function.
+func (e *ResultChannel[T]) SendWithCancel(result T, cancel context.CancelFunc) {
+	e.Send(result)
+	cancel()
+}
+
+// Receive receives a result from channel without blocking on the receiver.
+func (e *ResultChannel[T]) Receive() T {
+	select {
+	case result := <-e.ch:
+		return result
+	default:
+		var zeroValue T
+		return zeroValue
+	}
+}
+
+// NewResultChannel returns a new ResultChannel.
+func NewResultChannel[T comparable]() *ResultChannel[T] {
+	return &ResultChannel[T]{
+		ch: make(chan T, 1),
+	}
+}

pkg/scheduler/framework/parallelize/result_channel_test.go

@@ -25,22 +25,22 @@ import (
 )
 
 func TestErrorChannel(t *testing.T) {
-	errCh := NewErrorChannel()
+	errCh := NewResultChannel[error]()
 
-	if actualErr := errCh.ReceiveError(); actualErr != nil {
+	if actualErr := errCh.Receive(); actualErr != nil {
 		t.Errorf("expect nil from err channel, but got %v", actualErr)
 	}
 
 	err := errors.New("unknown error")
-	errCh.SendError(err)
-	if actualErr := errCh.ReceiveError(); actualErr != err {
+	errCh.Send(err)
+	if actualErr := errCh.Receive(); !errors.Is(actualErr, err) {
 		t.Errorf("expect %v from err channel, but got %v", err, actualErr)
 	}
 
 	_, ctx := ktesting.NewTestContext(t)
 	ctx, cancel := context.WithCancel(ctx)
-	errCh.SendErrorWithCancel(err, cancel)
-	if actualErr := errCh.ReceiveError(); actualErr != err {
+	errCh.SendWithCancel(err, cancel)
+	if actualErr := errCh.Receive(); !errors.Is(actualErr, err) {
 		t.Errorf("expect %v from err channel, but got %v", err, actualErr)
 	}
 

pkg/scheduler/framework/preemption/preemption.go

@@ -470,7 +470,7 @@ func (ev *Evaluator) prepareCandidate(ctx context.Context, c Candidate, pod *v1.
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	logger := klog.FromContext(ctx)
-	errCh := parallelize.NewErrorChannel()
+	errCh := parallelize.NewResultChannel[error]()
 	fh.Parallelizer().Until(ctx, len(c.Victims().Pods), func(index int) {
 		victim := c.Victims().Pods[index]
 		if victim.DeletionTimestamp != nil {
@@ -479,10 +479,10 @@ func (ev *Evaluator) prepareCandidate(ctx context.Context, c Candidate, pod *v1.
 			return
 		}
 		if err := ev.PreemptPod(ctx, c, pod, victim, pluginName); err != nil {
-			errCh.SendErrorWithCancel(err, cancel)
+			errCh.SendWithCancel(err, cancel)
 		}
 	}, ev.PluginName)
-	if err := errCh.ReceiveError(); err != nil {
+	if err := errCh.Receive(); err != nil {
 		return fwk.AsStatus(err)
 	}
 
@@ -553,11 +553,11 @@ func (ev *Evaluator) prepareCandidateAsync(c Candidate, pod *v1.Pod, pluginName
 
 	metrics.PreemptionVictims.Observe(float64(len(c.Victims().Pods)))
 
-	errCh := parallelize.NewErrorChannel()
+	errCh := parallelize.NewResultChannel[error]()
 	preemptPod := func(index int) {
 		victim := victimPods[index]
 		if err := ev.PreemptPod(ctx, c, pod, victim, pluginName); err != nil {
-			errCh.SendErrorWithCancel(err, cancel)
+			errCh.SendWithCancel(err, cancel)
 		}
 	}
 
@@ -605,7 +605,7 @@ func (ev *Evaluator) prepareCandidateAsync(c Candidate, pod *v1.Pod, pluginName
 			// the preemptor completes the deletion API calls and is removed from the `preempting` map - that way
 			// the preemptor could end up stuck in the unschedulable pool, with all pod removal events being ignored.
 			ev.Handler.Parallelizer().Until(ctx, len(victimPods)-1, preemptPod, ev.PluginName)
-			if err := errCh.ReceiveError(); err != nil {
+			if err := errCh.Receive(); err != nil {
 				utilruntime.HandleErrorWithContext(ctx, err, "Error occurred during async preemption")
 				result = metrics.GoroutineResultError
 				preemptLastVictim = false

pkg/scheduler/framework/runtime/framework.go

@@ -1301,7 +1301,7 @@ func (f *frameworkImpl) RunScorePlugins(ctx context.Context, state fwk.CycleStat
 	}
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	errCh := parallelize.NewErrorChannel()
+	errCh := parallelize.NewResultChannel[error]()
 
 	if len(plugins) > 0 {
 		logger := klog.FromContext(ctx)
@@ -1326,7 +1326,7 @@ func (f *frameworkImpl) RunScorePlugins(ctx context.Context, state fwk.CycleStat
 				s, status := f.runScorePlugin(ctx, pl, state, pod, nodeInfo)
 				if !status.IsSuccess() {
 					err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
-					errCh.SendErrorWithCancel(err, cancel)
+					errCh.SendWithCancel(err, cancel)
 					return
 				}
 				pluginToNodeScores[pl.Name()][index] = fwk.NodeScore{
@@ -1335,7 +1335,7 @@ func (f *frameworkImpl) RunScorePlugins(ctx context.Context, state fwk.CycleStat
 				}
 			}
 		}, metrics.Score)
-		if err := errCh.ReceiveError(); err != nil {
+		if err := errCh.Receive(); err != nil {
 			return nil, fwk.AsStatus(fmt.Errorf("running Score plugins: %w", err))
 		}
 	}
@@ -1350,11 +1350,11 @@ func (f *frameworkImpl) RunScorePlugins(ctx context.Context, state fwk.CycleStat
 		status := f.runScoreExtension(ctx, pl, state, pod, nodeScoreList)
 		if !status.IsSuccess() {
 			err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
-			errCh.SendErrorWithCancel(err, cancel)
+			errCh.SendWithCancel(err, cancel)
 			return
 		}
 	}, metrics.Score)
-	if err := errCh.ReceiveError(); err != nil {
+	if err := errCh.Receive(); err != nil {
 		return nil, fwk.AsStatus(fmt.Errorf("running Normalize on Score plugins: %w", err))
 	}
 
@@ -1373,7 +1373,7 @@ func (f *frameworkImpl) RunScorePlugins(ctx context.Context, state fwk.CycleStat
 
 			if score > fwk.MaxNodeScore || score < fwk.MinNodeScore {
 				err := fmt.Errorf("plugin %q returns an invalid score %v, it should in the range of [%v, %v] after normalizing", pl.Name(), score, fwk.MinNodeScore, fwk.MaxNodeScore)
-				errCh.SendErrorWithCancel(err, cancel)
+				errCh.SendWithCancel(err, cancel)
 				return
 			}
 			weightedScore := score * int64(weight)
@@ -1385,7 +1385,7 @@ func (f *frameworkImpl) RunScorePlugins(ctx context.Context, state fwk.CycleStat
 		}
 		allNodePluginScores[index] = nodePluginScores
 	}, metrics.Score)
-	if err := errCh.ReceiveError(); err != nil {
+	if err := errCh.Receive(); err != nil {
 		return nil, fwk.AsStatus(fmt.Errorf("applying score defaultWeights on Score plugins: %w", err))
 	}
 

pkg/scheduler/schedule_one.go

@@ -651,7 +651,7 @@ func (sched *Scheduler) findNodesThatPassFilters(
 		return feasibleNodes, nil
 	}
 
-	errCh := parallelize.NewErrorChannel()
+	errCh := parallelize.NewResultChannel[error]()
 	var feasibleNodesLen int32
 	ctx, cancel := context.WithCancelCause(ctx)
 	defer cancel(errors.New("findNodesThatPassFilters has completed"))
@@ -667,7 +667,7 @@ func (sched *Scheduler) findNodesThatPassFilters(
 		nodeInfo := nodes[(sched.nextStartNodeIndex+i)%numAllNodes]
 		status := schedFramework.RunFilterPluginsWithNominatedPods(ctx, state, pod, nodeInfo)
 		if status.Code() == fwk.Error {
-			errCh.SendErrorWithCancel(status.AsError(), func() {
+			errCh.SendWithCancel(status.AsError(), func() {
 				cancel(errors.New("some other Filter operation failed"))
 			})
 			return
@@ -705,7 +705,7 @@ func (sched *Scheduler) findNodesThatPassFilters(
 		diagnosis.NodeToStatus.Set(item.node, item.status)
 		diagnosis.AddPluginStatus(item.status)
 	}
-	if err := errCh.ReceiveError(); err != nil {
+	if err := errCh.Receive(); err != nil {
 		statusCode = fwk.Error
 		return feasibleNodes, err
 	}