This is an automated email from the ASF dual-hosted git repository. wilfred-s pushed a commit to branch branch-1.9 in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
commit 9f190370a461f6a4c98551fe900305cc2e7d9bab Author: Aditya Maheshwari <[email protected]> AuthorDate: Mon Jun 22 11:32:00 2026 +1000 [YUNIKORN-3239] handling infra events and allocation events sepately (#1095) Adding or updating a node when in the same queue could get blocked by the scheduling cycle behind application and allocation updates. Changes to handle infra events, node events and allocation events separately. This will prevent node or infra events from getting blocked during a spike in allocation events or the scheduling cycle.. Closes: #1095 Signed-off-by: Wilfred Spiegelenburg <[email protected]> (cherry picked from commit ea164ef95d8afda657401e6d0827a2b20cfa841d) --- pkg/scheduler/scheduler.go | 84 +++++++++++++++++----- pkg/scheduler/scheduler_test.go | 149 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 216 insertions(+), 17 deletions(-) diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index ece83dac..430346d4 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -32,32 +32,43 @@ import ( "github.com/apache/yunikorn-scheduler-interface/lib/go/si" ) -// Main Scheduler service that starts the needed sub services +// Scheduler service that starts the needed sub services type Scheduler struct { - clusterContext *ClusterContext // main context - pendingEvents chan interface{} // queue for events - activityPending chan bool // activity pending channel - stop chan struct{} // channel to signal stop request - healthChecker *HealthChecker - nodesMonitor *nodesResourceUsageMonitor + clusterContext *ClusterContext // main context + pendingAllocEvents chan interface{} // queue for allocation and application events + pendingInfraEvents chan interface{} // queue for config and registration events + pendingNodeEvents chan interface{} // queue for node events + activityPending chan bool // activity pending channel + stop chan struct{} // channel to signal stop request + healthChecker *HealthChecker + nodesMonitor *nodesResourceUsageMonitor } func NewScheduler() *Scheduler { m := &Scheduler{} m.clusterContext = newClusterContext() - m.pendingEvents = make(chan interface{}, 1024*1024) + // Creating 3 channels for different types of events, the buffer size is set based on the expected event volume and processing speed. + // This can help to smooth out the event processing and avoid blocking the RM proxy when there is a sudden burst of events. + // A cluster can have hundreds of thousands of allocations, and each allocation can trigger multiple events, so the buffer size for pendingAllocEvents is set to 1 million. + m.pendingAllocEvents = make(chan interface{}, 1024*1024) + // There can be thousands of infra events in a cluster, so the buffer size for pendingInfraEvents is set to 1000. + m.pendingInfraEvents = make(chan interface{}, 1000) + // There can be tens of thousands of nodes in a cluster, so the buffer size for pendingNodeEvents is set to 100 thousand. + m.pendingNodeEvents = make(chan interface{}, 100*1000) m.activityPending = make(chan bool, 1) m.stop = make(chan struct{}) return m } -// Start service +// StartService starts the scheduler service, it will start the event handlers and the main scheduling loop. func (s *Scheduler) StartService(handlers handler.EventHandlers, manualSchedule bool) { // set the proxy handler in the context s.clusterContext.setEventHandler(handlers.RMProxyEventHandler) // Start event handlers - go s.handleRMEvent() + go s.handleAllocEvent() + go s.handleInfraEvent() + go s.handleNodeEvent() // Start resource monitor if necessary (majorly for testing) s.nodesMonitor = newNodesResourceUsageMonitor(s.clusterContext) @@ -118,9 +129,16 @@ func (s *Scheduler) internalQuotaPreemption() { } } -// Implement methods for Scheduler events +// HandleEvent is the main entry for handling events from RM proxy, it will dispatch events to different queues based on event type. func (s *Scheduler) HandleEvent(ev interface{}) { - enqueueAndCheckFull(s.pendingEvents, ev) + switch ev.(type) { + case *rmevent.RMUpdateAllocationEvent, *rmevent.RMUpdateApplicationEvent: + enqueueAndCheckFull(s.pendingAllocEvents, ev) + case *rmevent.RMUpdateNodeEvent: + enqueueAndCheckFull(s.pendingNodeEvents, ev) + default: + enqueueAndCheckFull(s.pendingInfraEvents, ev) + } } func enqueueAndCheckFull(queue chan interface{}, ev interface{}) { @@ -136,17 +154,31 @@ func enqueueAndCheckFull(queue chan interface{}, ev interface{}) { } } -func (s *Scheduler) handleRMEvent() { +func (s *Scheduler) handleAllocEvent() { for { select { - case ev := <-s.pendingEvents: + case ev := <-s.pendingAllocEvents: switch v := ev.(type) { case *rmevent.RMUpdateAllocationEvent: s.clusterContext.handleRMUpdateAllocationEvent(v) case *rmevent.RMUpdateApplicationEvent: s.clusterContext.handleRMUpdateApplicationEvent(v) - case *rmevent.RMUpdateNodeEvent: - s.clusterContext.handleRMUpdateNodeEvent(v) + default: + log.Log(log.Scheduler).Error("Received type is not an acceptable type for allocation event.", + zap.Stringer("received type", reflect.TypeOf(v))) + } + s.registerActivity() + case <-s.stop: + return + } + } +} + +func (s *Scheduler) handleInfraEvent() { + for { + select { + case ev := <-s.pendingInfraEvents: + switch v := ev.(type) { case *rmevent.RMPartitionsRemoveEvent: s.clusterContext.removePartitionsByRMID(v) case *rmevent.RMRegistrationEvent: @@ -154,7 +186,25 @@ func (s *Scheduler) handleRMEvent() { case *rmevent.RMConfigUpdateEvent: s.clusterContext.processRMConfigUpdateEvent(v) default: - log.Log(log.Scheduler).Error("Received type is not an acceptable type for RM event.", + log.Log(log.Scheduler).Error("Received type is not an acceptable type for infrastructure event.", + zap.Stringer("received type", reflect.TypeOf(v))) + } + s.registerActivity() + case <-s.stop: + return + } + } +} + +func (s *Scheduler) handleNodeEvent() { + for { + select { + case ev := <-s.pendingNodeEvents: + switch v := ev.(type) { + case *rmevent.RMUpdateNodeEvent: + s.clusterContext.handleRMUpdateNodeEvent(v) + default: + log.Log(log.Scheduler).Error("Received type is not an acceptable type for node event.", zap.Stringer("received type", reflect.TypeOf(v))) } s.registerActivity() diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go index 19eb7ecf..46a17825 100644 --- a/pkg/scheduler/scheduler_test.go +++ b/pkg/scheduler/scheduler_test.go @@ -167,3 +167,152 @@ func TestTriggerQuotaPreemption(t *testing.T) { }) } } + +// TestHandleEventRouting verifies that HandleEvent routes allocation and application +// events to pendingAllocEvents, and all other events to pendingInfraEvents. +func TestHandleEventRouting(t *testing.T) { + scheduler := NewScheduler() + + allocCases := []interface{}{ + &rmevent.RMUpdateAllocationEvent{Request: &si.AllocationRequest{}}, + &rmevent.RMUpdateApplicationEvent{Request: &si.ApplicationRequest{}}, + } + for _, ev := range allocCases { + scheduler.HandleEvent(ev) + } + assert.Equal(t, len(scheduler.pendingAllocEvents), 2, "expected 2 events on pendingAllocEvents") + assert.Equal(t, len(scheduler.pendingInfraEvents), 0, "expected 0 events on pendingInfraEvents") + assert.Equal(t, len(scheduler.pendingNodeEvents), 0, "expected 0 events on pendingNodeEvents") + + infraCases := []interface{}{ + &rmevent.RMRegistrationEvent{Channel: make(chan *rmevent.Result, 1)}, + &rmevent.RMConfigUpdateEvent{Channel: make(chan *rmevent.Result, 1)}, + &rmevent.RMPartitionsRemoveEvent{Channel: make(chan *rmevent.Result, 1)}, + } + for _, ev := range infraCases { + scheduler.HandleEvent(ev) + } + assert.Equal(t, len(scheduler.pendingAllocEvents), 2, "alloc channel should remain unchanged") + assert.Equal(t, len(scheduler.pendingInfraEvents), 3, "expected 3 events on pendingInfraEvents") + + nodeCase := &rmevent.RMUpdateNodeEvent{Request: &si.NodeRequest{}} + scheduler.HandleEvent(nodeCase) + assert.Equal(t, len(scheduler.pendingAllocEvents), 2, "alloc channel should remain unchanged") + assert.Equal(t, len(scheduler.pendingInfraEvents), 3, "infra channel should remain unchanged") + assert.Equal(t, len(scheduler.pendingNodeEvents), 1, "expected 1 event on pendingNodeEvents") +} + +// TestHandleAllocEventGoroutine verifies that the handleAllocEvent goroutine drains +// pendingAllocEvents and calls registerActivity and stops when signaled. +func TestHandleAllocEventGoroutine(t *testing.T) { + scheduler := NewScheduler() + + done := make(chan struct{}) + go func() { + defer close(done) + scheduler.handleAllocEvent() + }() + + // send an allocation update event; an empty request processes cleanly + scheduler.pendingAllocEvents <- &rmevent.RMUpdateAllocationEvent{ + Request: &si.AllocationRequest{}, + } + + // wait for activity signal which proves the event was dequeued and processed + select { + case <-scheduler.activityPending: + // success + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for activity from handleAllocEvent") + } + + close(scheduler.stop) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("handleAllocEvent goroutine did not stop") + } +} + +// TestHandleInfraEventGoroutine verifies that the handleInfraEvent goroutine drains +// pendingInfraEvents and calls registerActivity and stops when signaled. +func TestHandleInfraEventGoroutine(t *testing.T) { + scheduler := NewScheduler() + + resultCh := make(chan *rmevent.Result, 1) + scheduler.pendingInfraEvents <- &rmevent.RMRegistrationEvent{ + Registration: &si.RegisterResourceManagerRequest{ + RmID: rmID, + PolicyGroup: "default-policy-group", + Version: "0.0.2", + }, + Channel: resultCh, + } + + done := make(chan struct{}) + go func() { + defer close(done) + scheduler.handleInfraEvent() + }() + + // wait for activity signal which proves the event was processed + select { + case <-scheduler.activityPending: + // success + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for activity from handleInfraEvent") + } + + close(scheduler.stop) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("handleInfraEvent goroutine did not stop") + } +} + +// TestHandleNodeEventGoroutine verifies that the handleNodeEvent goroutine drains +// pendingNodeEvents and calls registerActivity. +func TestHandleNodeEventGoroutine(t *testing.T) { + scheduler := NewScheduler() + + scheduler.pendingNodeEvents <- &rmevent.RMUpdateNodeEvent{ + Request: &si.NodeRequest{}, + } + + done := make(chan struct{}) + go func() { + defer close(done) + scheduler.handleNodeEvent() + }() + + // wait for activity signal which proves the event was processed + select { + case <-scheduler.activityPending: + // success + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for activity from handleNodeEvent") + } + + close(scheduler.stop) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("handleNodeEvent goroutine did not stop") + } +} + +// TestNodeEventsNotBlockedByAllocEvents verifies that a full pendingAllocEvents channel +// does not prevent node events from being enqueued or processed. +func TestNodeEventsNotBlockedByAllocEvents(t *testing.T) { + scheduler := NewScheduler() + // fill the alloc channel to capacity so it cannot accept more events + for i := 0; i < cap(scheduler.pendingAllocEvents); i++ { + scheduler.pendingAllocEvents <- &rmevent.RMUpdateAllocationEvent{Request: &si.AllocationRequest{}} + } + + // a node event must still be enqueued without blocking + nodeEv := &rmevent.RMUpdateNodeEvent{Request: &si.NodeRequest{}} + scheduler.HandleEvent(nodeEv) + assert.Equal(t, len(scheduler.pendingNodeEvents), 1, "node event should be queued even when alloc channel is full") +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
