improve schedulingQueue to avoid making zombie scheduler

Author: sp-yduck

cloud/scheduler/queue/queue.go

@@ -8,8 +8,9 @@ import (
 )
 
 type SchedulingQueue struct {
-	activeQ []*qemuSpec
-	lock    *sync.Cond
+	activeQ      []*qemuSpec
+	lock         *sync.Cond
+	shuttingDown bool
 }
 
 func New() *SchedulingQueue {
@@ -30,22 +31,47 @@ type qemuSpec struct {
 func (s *SchedulingQueue) Add(ctx context.Context, config *api.VirtualMachineCreateOptions) {
 	s.lock.L.Lock()
 	defer s.lock.L.Unlock()
+
+	if s.shuttingDown {
+		return
+	}
+
 	s.activeQ = append(s.activeQ, &qemuSpec{ctx: ctx, config: config})
 	s.lock.Signal()
 }
 
-// return next qemuSpec
-// to do : break if context is done
-func (s *SchedulingQueue) NextQEMU(ctx context.Context) *qemuSpec {
-	// wait
+// return length of active queue
+// func (s *SchedulingQueue) Len() int {
+// 	s.lock.L.Lock()
+// 	defer s.lock.L.Unlock()
+// 	return len(s.activeQ)
+// }
+
+// return nex qemuSpec
+func (s *SchedulingQueue) Get() (spec *qemuSpec, shutdown bool) {
 	s.lock.L.Lock()
 	defer s.lock.L.Unlock()
-	for len(s.activeQ) == 0 {
+	for len(s.activeQ) == 0 && !s.shuttingDown {
 		s.lock.Wait()
 	}
-	spec := s.activeQ[0]
+	if len(s.activeQ) == 0 {
+		return nil, true
+	}
+
+	spec = s.activeQ[0]
+	// The underlying array still exists and reference this object,
+	// so the object will not be garbage collected.
+	s.activeQ[0] = nil
 	s.activeQ = s.activeQ[1:]
-	return spec
+	return spec, false
+}
+
+// shut down the queue
+func (s *SchedulingQueue) ShutDown() {
+	s.lock.L.Lock()
+	defer s.lock.L.Unlock()
+	s.shuttingDown = true
+	s.lock.Broadcast()
 }
 
 func (s *qemuSpec) Config() *api.VirtualMachineCreateOptions {

cloud/scheduler/queue/queue_test.go

@@ -0,0 +1,62 @@
+package queue_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/sp-yduck/cluster-api-provider-proxmox/cloud/scheduler/queue"
+	"github.com/sp-yduck/proxmox-go/api"
+)
+
+func TestQueue(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Queue Suite")
+}
+
+var _ = Describe("New", Label("unit", "queue"), func() {
+	It("shoud not error", func() {
+		q := queue.New()
+		Expect(q).ToNot(BeNil())
+	})
+})
+
+var _ = Describe("Add", Label("unit", "queue"), func() {
+	q := queue.New()
+
+	It("should not error", func() {
+		q.Add(context.Background(), &api.VirtualMachineCreateOptions{Name: "foo"})
+	})
+})
+
+var _ = Describe("Get", Label("unit", "queue"), func() {
+	var q *queue.SchedulingQueue
+
+	BeforeEach(func() {
+		q = queue.New()
+	})
+
+	Context("normal", func() {
+		It("should run properly", func() {
+			c := &api.VirtualMachineCreateOptions{Name: "foo"}
+			q.Add(context.Background(), c)
+			qemu, shutdown := q.Get()
+			Expect(qemu.Config()).To(Equal(c))
+			Expect(shutdown).To(BeFalse())
+		})
+	})
+
+	Context("shutdown empty queue after 1 sec", func() {
+		It("should get nil", func() {
+			go func() {
+				time.Sleep(1 * time.Second)
+				q.ShutDown()
+			}()
+			qemu, shutdown := q.Get()
+			Expect(qemu).To(BeNil())
+			Expect(shutdown).To(BeTrue())
+		})
+	})
+})

cloud/scheduler/scheduler.go

@@ -47,7 +47,7 @@ func (m *Manager) GetOrCreateScheduler(client *proxmox.Service) *Scheduler {
 	if err != nil {
 		// create new scheduler without registering
 		// to not make it zombie scheduler set timeout to context
-		sched := m.NewScheduler(client, WithCancelContext(context.WithTimeout(m.ctx, 5*time.Minute)))
+		sched := m.NewScheduler(client, WithTimeout(1*time.Minute))
 		return sched
 	}
 	m.params.Logger = m.params.Logger.WithValues("scheduler ID", *schedID)
@@ -90,11 +90,17 @@ func (m *Manager) NewScheduler(client *proxmox.Service, opts ...SchedulerOption)
 }
 
 type SchedulerOption func(s *Scheduler)
+type CancelFunc func()
 
-func WithCancelContext(ctx context.Context, cancel context.CancelFunc) SchedulerOption {
+func (s *Scheduler) WithCancel() (*Scheduler, CancelFunc) {
+	return s, s.Stop
+}
+
+// set timeout to scheduler
+func WithTimeout(timeout time.Duration) SchedulerOption {
 	return func(s *Scheduler) {
-		s.ctx = ctx
-		s.cancel = cancel
+		s, cancel := s.WithCancel()
+		go time.AfterFunc(timeout, cancel)
 	}
 }
 
@@ -123,7 +129,7 @@ type Scheduler struct {
 
 	// to do : cache
 
-	// map[qemu name]*framework.CycleState
+	// map[qemu name]chan *framework.CycleState
 	resultMap map[string]chan *framework.CycleState
 	logger    logr.Logger
 
@@ -170,14 +176,18 @@ func (s *Scheduler) RunAsync() {
 }
 
 // stop scheduler
-// func (s *Scheduler) Stop() {
-// 	defer s.cancel()
-// }
+func (s *Scheduler) Stop() {
+	defer s.cancel()
+	s.schedulingQueue.ShutDown()
+}
 
 // retrieve one qemuSpec from queue and try to create
 // new qemu according to the qemuSpec
 func (s *Scheduler) ScheduleOne(ctx context.Context) {
-	qemu := s.schedulingQueue.NextQEMU(ctx)
+	qemu, shutdown := s.schedulingQueue.Get()
+	if shutdown {
+		return
+	}
 	config := qemu.Config()
 	qemuCtx := qemu.Context()
 	s.logger = s.logger.WithValues("qemu", config.Name)
@@ -230,7 +240,9 @@ func (s *Scheduler) WaitStatus(ctx context.Context, config *api.VirtualMachineCr
 		delete(s.resultMap, config.Name)
 		return *state, nil
 	case <-ctx.Done():
-		return framework.CycleState{}, fmt.Errorf("exceed timeout deadline")
+		err := fmt.Errorf("exceed timeout deadline. schedulingQueue might be shutdowned")
+		s.logger.Error(err, fmt.Sprintf("schedulingQueue: %v", *s.schedulingQueue))
+		return framework.CycleState{}, err
 	}
 }
 

cloud/scheduler/scheduler_test.go

@@ -6,12 +6,11 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"github.com/sp-yduck/proxmox-go/api"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
 	"github.com/sp-yduck/cluster-api-provider-proxmox/cloud/scheduler"
 	"github.com/sp-yduck/cluster-api-provider-proxmox/cloud/scheduler/framework"
-	"github.com/sp-yduck/proxmox-go/api"
-	// "github.com/sp-yduck/cluster-api-provider-proxmox/cloud/scheduler/framework"
 )
 
 var _ = Describe("NewManager", Label("unit", "scheduler"), func() {
@@ -40,19 +39,36 @@ var _ = Describe("GetOrCreateScheduler", Label("integration", "scheduler"), func
 	})
 })
 
-var _ = Describe("Run (RunAsync) / IsRunning", Label("unit", "scheduler"), func() {
+var _ = Describe("Run (RunAsync) / IsRunning / Stop", Label("unit", "scheduler"), func() {
 	manager := scheduler.NewManager(scheduler.SchedulerParams{zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true))})
 
 	Context("with minimal scheduler", func() {
 		It("should not error", func() {
 			sched := manager.NewScheduler(proxmoxSvc)
 			sched.RunAsync()
 			time.Sleep(1 * time.Second)
-			Expect(sched.IsRunning()).To(Equal(true))
+			Expect(sched.IsRunning()).To(BeTrue())
+			sched.Stop()
+			time.Sleep(1 * time.Second)
+			Expect(sched.IsRunning()).To(BeFalse())
 		})
 	})
 })
 
+var _ = Describe("WithTimeout", Label("integration", "scheduler"), func() {
+	manager := scheduler.NewManager(scheduler.SchedulerParams{})
+
+	It("should not error", func() {
+		sched := manager.NewScheduler(proxmoxSvc, scheduler.WithTimeout(2*time.Second))
+		Expect(sched).NotTo(BeNil())
+		sched.RunAsync()
+		time.Sleep(1 * time.Second)
+		Expect(sched.IsRunning()).To(BeTrue())
+		time.Sleep(2 * time.Second)
+		Expect(sched.IsRunning()).To(BeFalse())
+	})
+})
+
 var _ = Describe("CreateQEMU", Label("integration"), func() {
 	manager := scheduler.NewManager(scheduler.SchedulerParams{zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true))})
 	var result framework.SchedulerResult