Update pod evaluator to use round-robin to delegate work to krm functions (kptdev#497)

* Update pod evaluator to use round-robin to delegate work to krm functions

* Reduce the replica count to 1 for function-runner

* Remove 5m context as api-server context has a 291s deadline

* Add documentation for fn-runner round-robin mechanism

Author: kushnaidu

deployments/porch/2-function-runner.yaml

@@ -24,7 +24,7 @@ metadata:
   name: function-runner
   namespace: porch-system
 spec:
-  replicas: 2
+  replicas: 1
   selector:
     matchLabels:
       app: function-runner

docs/content/en/docs/5_architecture_and_components/function-runner/design.md

@@ -244,6 +244,57 @@ For detailed explanations of how these differences affect operations, see the in
 - May recreate pods for functions with irregular usage patterns
 - Configurable TTL allows tuning for specific workloads
 
+### Pod Selection Strategy
+
+**Decision**: Use round-robin distribution among pods with equal minimum load.
+
+**Rationale:**
+- Ensures even distribution of requests across equally-loaded pods
+- Prevents all requests from going to the same pod when multiple pods have equal capacity
+- Simple and predictable load balancing behavior
+- Works well with parallel execution within pods
+
+**Implementation:**
+- Find the minimum waitlist length across all pods for a function
+- Use round-robin to select among pods that have the minimum waitlist
+- Maintains round-robin index per function to track position
+
+**Alternatives considered:**
+- **Pure least-loaded**: Can cause all requests to go to same pod when loads are equal
+- **Random selection**: Less predictable distribution patterns
+- **Weighted round-robin**: Unnecessary complexity for uniform pod capacities
+
+**Trade-offs:**
+- Slightly more complex than pure least-loaded selection
+- Requires maintaining round-robin index per function
+- Better load distribution justifies the minimal added complexity
+
+### Parallel Execution Model
+
+**Decision**: Allow parallel function evaluations within the same pod.
+
+**Rationale:**
+- Improves throughput by utilizing pod capacity fully
+- Reduces wait times when multiple requests target the same function
+- Simplifies code by removing serialization mutex
+- Works well with round-robin pod selection for even load distribution
+
+**Implementation:**
+- Removed per-pod mutex that was serializing executions
+- Concurrent executions tracked via atomic counter
+- gRPC connection shared across parallel calls to same pod
+- Context timeout prevents goroutine leaks
+
+**Alternatives considered:**
+- **Serial execution per pod**: Simpler but underutilizes pod capacity
+- **Connection pooling**: More complex, unnecessary with gRPC multiplexing
+- **Per-function concurrency limits**: Added complexity without clear benefit
+
+**Trade-offs:**
+- Functions must be safe for concurrent execution (standard KRM function requirement)
+- Multiple evaluations share pod resources (CPU, memory)
+- Better resource utilization justifies the concurrency model
+
 ### Service Mesh Compatibility
 
 **Decision**: Use ClusterIP services as frontends for function pods.

docs/content/en/docs/5_architecture_and_components/function-runner/functionality/function-evaluation.md

@@ -437,14 +437,20 @@ The evaluation system employs several performance strategies.
 - Multiple requests can execute concurrently
 - Each request gets own gRPC connection
 - Pod cache manager coordinates access
-- Waitlist prevents duplicate pod creation
+- Round-robin load balancing across equal-load pods
 
 **Concurrency characteristics:**
-- Same function, same pod: Sequential (one at a time)
+- Same function, same pod: Parallel execution supported
 - Same function, different pods: Concurrent
 - Different functions: Fully concurrent
 - No artificial concurrency limits
 
+**Load balancing:**
+- Requests distributed to least-loaded pods
+- Round-robin among pods with equal load
+- Ensures even work distribution
+- Prevents hotspotting on single pod
+
 ### Resource Limits
 
 **Resource considerations:**

docs/content/en/docs/5_architecture_and_components/function-runner/functionality/pod-lifecycle-management.md

@@ -34,7 +34,7 @@ The Pod Cache Manager runs as a single goroutine managing all pod caching operat
 
 The manager maintains two data structures:
 
-**Cache Map** - Maps function image names to pod information and gRPC client connections. Each entry contains the pod namespace/name and an active gRPC client connection to the pod's wrapper server.
+**Cache Map** - Maps function image names to pod information and gRPC client connections. Each entry contains the pod namespace/name and an active gRPC client connection to the pod's wrapper server. When multiple pods serve the same function, the cache uses a round-robin index to distribute requests evenly across pods with equal load.
 
 **Waitlist Map** - Maps function image names to lists of channels waiting for pod readiness. When multiple evaluation requests arrive for the same function before a pod is ready, they queue in the waitlist to prevent duplicate pod creation.
 
@@ -346,9 +346,14 @@ The system supports concurrent operations efficiently:
 - Each request gets own gRPC connection
 - Pod cache manager coordinates access via channels
 - Waitlist prevents duplicate pod creation
+- Function evaluations can execute in parallel within the same pod
+
+**Pod Selection Strategy:**
+
+When multiple pods serve the same function, the cache manager selects the pod with the least load (smallest waitlist length). When multiple pods have equal load, the system uses round-robin distribution to ensure even load balancing across pods rather than always selecting the first available pod.
 
 **Concurrency benefits:**
-- Same function, same pod: Sequential (one at a time)
+- Same function, same pod: Parallel (multiple evaluations execute concurrently)
 - Same function, different pods: Concurrent
 - Different functions: Fully concurrent
 - No artificial concurrency limits

docs/content/en/docs/5_architecture_and_components/function-runner/interactions.md

@@ -141,10 +141,12 @@ Pod      Pod       Pod
 
 **Execution pattern:**
 - Pod cache checked for existing pod (reuse if available)
+- Pod selection uses round-robin among pods with minimum waitlist length
 - Cache miss triggers pod creation with wrapper server
 - ClusterIP service provides stable DNS-based access
 - gRPC connection to wrapper server in pod
 - Wrapper server executes function binary and returns results
+- Multiple evaluations can execute in parallel on the same pod
 
 **For detailed pod lifecycle, see [Pod Lifecycle Management]({{% relref "/docs/5_architecture_and_components/function-runner/functionality/pod-lifecycle-management.md" %}}).**
 
@@ -430,7 +432,8 @@ The Function Runner handles concurrent operations safely:
 
 **Concurrency characteristics:**
 - Multiple function executions can run concurrently
-- Each request gets own gRPC connection
+- Multiple evaluations can run in parallel on the same pod
+- Each request gets own gRPC connection to pod
 - Pod cache manager coordinates access via channels
 - Waitlist prevents duplicate pod creation
 

docs/content/en/docs/6_configuration_and_deployments/configurations/components/function-runner-config/_index.md

@@ -149,7 +149,7 @@ metadata:
   name: function-runner
   namespace: porch-system
 spec:
-  replicas: 2
+  replicas: 1
   selector:
     matchLabels:
       app: function-runner

func/internal/podcachemanager.go

@@ -71,6 +71,8 @@ type podCacheConfigEntry struct {
 type functionInfo struct {
 	// status of all pods belonging to the same KRM function image
 	pods []functionPodInfo
+	// roundRobinIdx is used to distribute requests across pods when all have equal load
+	roundRobinIdx int
 }
 
 // functionPodInfo represents the state of a single pod instance.
@@ -82,8 +84,6 @@ type functionPodInfo struct {
 	waitlist []chan<- *connectionResponse
 	// time of last function evaluation, used by the garbage collector to identify idle pods
 	lastActivity time.Time
-	// mutex used to prevent concurrent fn evaluations in the same pod
-	fnEvaluationMutex *sync.Mutex
 	// the number of currently ongoing and waiting fn evaluations in the pod
 	concurrentEvaluations *atomic.Int32
 }
@@ -369,25 +369,39 @@ func forEachConcurrently(m []podCacheConfigEntry, fn func(k podCacheConfigEntry)
 	wg.Wait()
 }
 
-// findBestPod returns with the index of the least loaded healthy pod for the given function.
+// findBestPod returns with the index of the best pod for the given function.
+// It uses round-robin among pods with equal load to ensure even distribution.
 // If there are no suitable pods, it returns with -1.
 func (pcm *podCacheManager) findBestPod(fn *functionInfo) (int, int) {
 	if fn == nil {
 		return -1, 0
 	}
-	if len(fn.pods) == 0 {
+	n := len(fn.pods)
+	if n == 0 {
 		return -1, 0
 	}
-	bestPodIdx := 0
-	bestWaitlistLen := fn.pods[0].WaitlistLen()
-	for i := 1; i < len(fn.pods); i++ {
-		waitlistLen := fn.pods[i].WaitlistLen()
-		if waitlistLen < bestWaitlistLen {
-			bestPodIdx = i
-			bestWaitlistLen = waitlistLen
+
+	minWaitlist := 0
+	// Find the minimum waitlist length across all pods
+	minWaitlist = fn.pods[0].WaitlistLen()
+	for i := 1; i < n; i++ {
+		wl := fn.pods[i].WaitlistLen()
+		if wl < minWaitlist {
+			minWaitlist = wl
 		}
 	}
-	return bestPodIdx, bestWaitlistLen
+
+	// Round-robin among pods that have the minimum waitlist length
+	for i := 0; i < n; i++ {
+		idx := (fn.roundRobinIdx + i) % n
+		if fn.pods[idx].WaitlistLen() == minWaitlist {
+			fn.roundRobinIdx = (idx + 1) % n
+			return idx, minWaitlist
+		}
+	}
+
+	// This should never happen since minWaitlist was calculated from these same pods
+	return -1, 0
 }
 
 // removeUnhealthyPods removes unhealthy pods from the function's pod list.
@@ -539,7 +553,6 @@ func NewPodInfo(firstResponseCh chan<- *connectionResponse) functionPodInfo {
 		podData:               nil, // This will be filled in when the pod is ready.
 		lastActivity:          time.Now(),
 		concurrentEvaluations: &atomic.Int32{},
-		fnEvaluationMutex:     &sync.Mutex{},
 	}
 	if firstResponseCh != nil {
 		pod.waitlist = append(pod.waitlist, firstResponseCh)
@@ -564,7 +577,6 @@ func (pod *functionPodInfo) SendResponse(responseCh chan<- *connectionResponse,
 	default:
 		responseCh <- &connectionResponse{
 			podData:               *pod.podData,
-			fnEvaluationMutex:     pod.fnEvaluationMutex,
 			concurrentEvaluations: pod.concurrentEvaluations,
 			err:                   nil,
 		}

func/internal/podcachemanager_unit_test.go

@@ -39,7 +39,6 @@ func makePodInfoWithLoad(load int32) functionPodInfo {
 	counter.Store(load)
 	return functionPodInfo{
 		concurrentEvaluations: counter,
-		fnEvaluationMutex:     &sync.Mutex{},
 		lastActivity:          time.Now(),
 		waitlist:              []chan<- *connectionResponse{},
 	}
@@ -57,7 +56,6 @@ func makeReadyPodInfo(image string, podKey, serviceKey client.ObjectKey, grpcCon
 			grpcConnection: grpcConn,
 		},
 		concurrentEvaluations: counter,
-		fnEvaluationMutex:     &sync.Mutex{},
 		lastActivity:          time.Now(),
 		waitlist:              []chan<- *connectionResponse{},
 	}
@@ -301,7 +299,6 @@ func TestNewPodInfo(t *testing.T) {
 		assert.Nil(t, pod.podData)
 		assert.Empty(t, pod.waitlist)
 		assert.Equal(t, int32(0), pod.concurrentEvaluations.Load())
-		assert.NotNil(t, pod.fnEvaluationMutex)
 	})
 
 	t.Run("non-nil channel adds to waitlist and increments counter", func(t *testing.T) {
@@ -318,7 +315,6 @@ func TestSendResponse(t *testing.T) {
 		pod := &functionPodInfo{
 			podData:               &podData{image: "test"},
 			concurrentEvaluations: &atomic.Int32{},
-			fnEvaluationMutex:     &sync.Mutex{},
 		}
 		ch := make(chan *connectionResponse, 1)
 		testErr := fmt.Errorf("test error")
@@ -334,7 +330,6 @@ func TestSendResponse(t *testing.T) {
 		pod := &functionPodInfo{
 			podData:               nil,
 			concurrentEvaluations: &atomic.Int32{},
-			fnEvaluationMutex:     &sync.Mutex{},
 		}
 		ch := make(chan *connectionResponse, 1)
 
@@ -359,7 +354,6 @@ func TestSendResponse(t *testing.T) {
 				serviceKey:     &serviceKey,
 			},
 			concurrentEvaluations: &atomic.Int32{},
-			fnEvaluationMutex:     &sync.Mutex{},
 		}
 		ch := make(chan *connectionResponse, 1)
 
@@ -369,7 +363,6 @@ func TestSendResponse(t *testing.T) {
 		assert.NoError(t, resp.err)
 		assert.Equal(t, "test-image", resp.podData.image)
 		assert.NotNil(t, resp.grpcConnection)
-		assert.NotNil(t, resp.fnEvaluationMutex)
 		assert.NotNil(t, resp.concurrentEvaluations)
 	})
 }
@@ -461,7 +454,6 @@ func TestRemoveUnhealthyPods(t *testing.T) {
 				{
 					podData:               nil, // under creation
 					concurrentEvaluations: &atomic.Int32{},
-					fnEvaluationMutex:     &sync.Mutex{},
 				},
 			},
 		}

func/internal/podevaluator.go

@@ -17,7 +17,6 @@ package internal
 import (
 	"context"
 	"fmt"
-	"sync"
 	"sync/atomic"
 	"time"
 
@@ -93,8 +92,6 @@ type connectionRequest struct {
 
 type connectionResponse struct {
 	podData
-	// mutex used to prevent concurrent fn evaluations in the same pod
-	fnEvaluationMutex *sync.Mutex
 	// the number of currently ongoing and waiting fn evaluations in the pod
 	concurrentEvaluations *atomic.Int32
 	// err indicates the error that prevents us to allocate a pod for the fn evaluator
@@ -215,23 +212,25 @@ func (pe *podEvaluator) EvaluateFunction(ctx context.Context, req *evaluator.Eva
 	}
 
 	// Waiting for the client from the channel. This step is blocking.
-	pod := <-responseChannel
-	if pod == nil || pod.grpcConnection == nil || pod.err != nil {
-		return nil, fmt.Errorf("unable to get the grpc client to the pod for %v: %w", req.Image, pod.err)
-	}
+	select {
+	case pod := <-responseChannel:
+		if pod == nil || pod.grpcConnection == nil || pod.err != nil {
+			return nil, fmt.Errorf("unable to get the grpc client to the pod for %v: %w", req.Image, pod.err)
+		}
 
-	defer pod.concurrentEvaluations.Add(-1)
-	pod.fnEvaluationMutex.Lock()
-	defer pod.fnEvaluationMutex.Unlock()
+		defer pod.concurrentEvaluations.Add(-1)
 
-	resp, err := evaluator.NewFunctionEvaluatorClient(pod.grpcConnection).EvaluateFunction(ctx, req)
-	if err != nil {
-		klog.V(4).Infof("Resource List: %s", req.ResourceList)
-		return nil, fmt.Errorf("unable to evaluate %v with pod evaluator: %w", req.Image, err)
-	}
-	// Log stderr when the function succeeded. If the function fails, stderr will be surfaced to the users.
-	if len(resp.Log) > 0 {
-		klog.Warningf("evaluating %v succeeded, but stderr is: %v", req.Image, string(resp.Log))
+		resp, err := evaluator.NewFunctionEvaluatorClient(pod.grpcConnection).EvaluateFunction(ctx, req)
+		if err != nil {
+			klog.V(4).Infof("Resource List: %s", req.ResourceList)
+			return nil, fmt.Errorf("unable to evaluate %v with pod evaluator: %w", req.Image, err)
+		}
+		// Log stderr when the function succeeded. If the function fails, stderr will be surfaced to the users.
+		if len(resp.Log) > 0 {
+			klog.Warningf("evaluating %v succeeded, but stderr is: %v", req.Image, string(resp.Log))
+		}
+		return resp, nil
+	case <-ctx.Done():
+		return nil, fmt.Errorf("function evaluation timed out for %v: %w", req.Image, ctx.Err())
 	}
-	return resp, nil
 }

…evaluator_porch_serial_execution_test.go …aluator_porch_parallel_execution_test.gofunc/internal/podevaluator_porch_serial_execution_test.go renamed to func/internal/podevaluator_porch_parallel_execution_test.go func/internal/podevaluator_porch_serial_execution_test.go renamed to func/internal/podevaluator_porch_parallel_execution_test.go

@@ -53,7 +53,7 @@ func startFakeServer(ctx context.Context, t *testing.T, delay time.Duration) (st
 	return addr, nil
 }
 
-func TestPodEvaluatorExecutionSerial(t *testing.T) {
+func TestPodEvaluatorExecutionParallel(t *testing.T) {
 	const sleep = 2 * time.Second
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -75,16 +75,16 @@ func TestPodEvaluatorExecutionSerial(t *testing.T) {
 
 	reqCh := make(chan *connectionRequest, 2)
 	go func() {
-		lock := &sync.Mutex{}
 		counter := &atomic.Int32{}
 		for req := range reqCh {
+			// Increment counter to simulate single pod with limited concurrency
+			counter.Add(1)
 			req.responseCh <- &connectionResponse{
 				podData: podData{
 					image:          req.image,
 					grpcConnection: conn,
 					podKey:         ptr.To(client.ObjectKey{}),
 				},
-				fnEvaluationMutex:     lock,
 				concurrentEvaluations: counter,
 				err:                   nil,
 			}
@@ -116,7 +116,13 @@ func TestPodEvaluatorExecutionSerial(t *testing.T) {
 	t.Logf("durations: %v", durations)
 
 	slices.Sort(durations)
-	if durations[funcCallCount-1] < time.Duration(funcCallCount)*sleep {
-		t.Errorf("expected serial but slowest took only %v", durations[funcCallCount-1])
+	// Since the current implementation allows parallel execution on the same connection,
+	// we should expect parallel behavior, not serial. Update the test expectation.
+	if durations[funcCallCount-1] > time.Duration(funcCallCount)*sleep {
+		t.Errorf("expected parallel but execution took too long: %v", durations[funcCallCount-1])
+	}
+	// Verify that all calls completed in roughly the same time (parallel execution)
+	if durations[funcCallCount-1]-durations[0] > sleep/2 {
+		t.Errorf("expected parallel execution but calls had significant time difference: %v", durations[funcCallCount-1]-durations[0])
 	}
 }