Skip to content

Commit fd956d6

Browse files
nwangfwnwangfw
committed
remove-async-from-planner-name
Signed-off-by: Ning Wang <n.wang.chn@hotmail.com>
1 parent f0c6140 commit fd956d6

3 files changed

Lines changed: 50 additions & 49 deletions

File tree

apps/console/api/planner/impl/async_planner.go renamed to apps/console/api/planner/impl/planner.go

Lines changed: 22 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -33,10 +33,11 @@ import (
3333
rmtypes "github.com/vllm-project/aibrix/apps/console/api/resource_manager/types"
3434
)
3535

36-
// AsyncPlanner is an asynchronous Planner. Enqueue records the job in memory,
37-
// returns a placeholder batch in "pending" status, and lets workers run
38-
// Provision, wait for the resource to reach Running, then CreateBatch.
39-
type AsyncPlanner struct {
36+
// Planner is an asynchronous implementation of plannerapi.Planner.
37+
// Enqueue records the job in memory, returns a placeholder batch in
38+
// "pending" status, and lets workers run Provision, wait for the
39+
// resource to reach Running, then CreateBatch.
40+
type Planner struct {
4041
bc plannerclient.BatchClient
4142
prov provisioner.Provisioner
4243

@@ -116,14 +117,14 @@ const defaultProvPollInterval = 5 * time.Second
116117
// the resource is released.
117118
const provReadyTimeout = 2 * time.Minute
118119

119-
// NewAsyncPlanner constructs an asynchronous AsyncPlanner Planner and starts
120-
// workerCount background workers. workerCount < 1 is floored to 1.
121-
func NewAsyncPlanner(bc plannerclient.BatchClient, prov provisioner.Provisioner, workerCount int) *AsyncPlanner {
120+
// NewPlanner constructs a Planner and starts workerCount background
121+
// workers. workerCount < 1 is floored to 1.
122+
func NewPlanner(bc plannerclient.BatchClient, prov provisioner.Provisioner, workerCount int) *Planner {
122123
if workerCount < 1 {
123124
workerCount = 1
124125
}
125126
ctx, cancel := context.WithCancel(context.Background())
126-
q := &AsyncPlanner{
127+
q := &Planner{
127128
bc: bc,
128129
prov: prov,
129130
submit: make(chan string, queueCapacity),
@@ -141,16 +142,16 @@ func NewAsyncPlanner(bc plannerclient.BatchClient, prov provisioner.Provisioner,
141142
return q
142143
}
143144

144-
var _ plannerapi.Planner = (*AsyncPlanner)(nil)
145+
var _ plannerapi.Planner = (*Planner)(nil)
145146

146147
// Close cancels in-flight work and waits for workers to exit.
147-
func (q *AsyncPlanner) Close() error {
148+
func (q *Planner) Close() error {
148149
q.baseCancel()
149150
q.wg.Wait()
150151
return nil
151152
}
152153

153-
func (q *AsyncPlanner) run() {
154+
func (q *Planner) run() {
154155
defer q.wg.Done()
155156
for {
156157
select {
@@ -162,7 +163,7 @@ func (q *AsyncPlanner) run() {
162163
}
163164
}
164165

165-
func (q *AsyncPlanner) process(jobID string) {
166+
func (q *Planner) process(jobID string) {
166167
// Atomic check-and-flip Pending → Provisioning.
167168
q.mu.Lock()
168169
job, ok := q.jobs[jobID]
@@ -253,7 +254,7 @@ func (q *AsyncPlanner) process(jobID string) {
253254
// or Failed, the timeout elapses, or the scheduler is shutting down.
254255
// Provisioner.Provision returns when the request is accepted, not when the
255256
// resource is ready; Planner must wait for Running before invoking CreateBatch.
256-
func (q *AsyncPlanner) waitForProvisionReady(provisionID string) error {
257+
func (q *Planner) waitForProvisionReady(provisionID string) error {
257258
filter := &rmtypes.ListOptions{ProvisionIDs: &[]string{provisionID}}
258259
deadline := time.Now().Add(provReadyTimeout)
259260
for {
@@ -285,14 +286,14 @@ func (q *AsyncPlanner) waitForProvisionReady(provisionID string) error {
285286
// releaseAfter performs a best-effort RM release and logs failures. The
286287
// reason string ("wait failure", "CreateBatch failure", "cancel-race")
287288
// appears in the log line so each call site is self-identifying.
288-
func (q *AsyncPlanner) releaseAfter(jobID, provisionID, reason string) {
289+
func (q *Planner) releaseAfter(jobID, provisionID, reason string) {
289290
if err := q.prov.Release(q.baseCtx, provisionID); err != nil {
290291
klog.Warningf("[planner] release after %s job_id=%q provision_id=%q: %v",
291292
reason, jobID, provisionID, err)
292293
}
293294
}
294295

295-
func (q *AsyncPlanner) markFailed(jobID string, err error) {
296+
func (q *Planner) markFailed(jobID string, err error) {
296297
q.mu.Lock()
297298
if job, ok := q.jobs[jobID]; ok {
298299
job.state = jobStateFailed
@@ -305,7 +306,7 @@ func (q *AsyncPlanner) markFailed(jobID string, err error) {
305306

306307
// Enqueue records the job, pushes it onto the worker channel, and returns
307308
// a placeholder batch in "pending" status.
308-
func (q *AsyncPlanner) Enqueue(ctx context.Context, req *plannerapi.EnqueueRequest) (*plannerapi.Job, error) {
309+
func (q *Planner) Enqueue(ctx context.Context, req *plannerapi.EnqueueRequest) (*plannerapi.Job, error) {
309310
if req == nil {
310311
return nil, fmt.Errorf("%w: nil request", plannerapi.ErrInvalidJob)
311312
}
@@ -357,7 +358,7 @@ func (q *AsyncPlanner) Enqueue(ctx context.Context, req *plannerapi.EnqueueReque
357358

358359
// GetJob resolves the JobID. Submitted jobs forward to MDS; others return
359360
// a placeholder batch with status derived from jobState.
360-
func (q *AsyncPlanner) GetJob(ctx context.Context, jobID string) (*plannerapi.Job, error) {
361+
func (q *Planner) GetJob(ctx context.Context, jobID string) (*plannerapi.Job, error) {
361362
if jobID == "" {
362363
return nil, fmt.Errorf("%w: empty job_id", plannerapi.ErrInvalidJob)
363364
}
@@ -392,7 +393,7 @@ func (q *AsyncPlanner) GetJob(ctx context.Context, jobID string) (*plannerapi.Jo
392393
// MDS for a submitted job. A cancel that lands mid-Provision or
393394
// mid-CreateBatch is honored at the worker's post-CreateBatch checkpoint
394395
// (which forwards CancelBatch and releases the resource).
395-
func (q *AsyncPlanner) Cancel(ctx context.Context, jobID string) (*plannerapi.Job, error) {
396+
func (q *Planner) Cancel(ctx context.Context, jobID string) (*plannerapi.Job, error) {
396397
if jobID == "" {
397398
return nil, fmt.Errorf("%w: empty job_id", plannerapi.ErrInvalidJob)
398399
}
@@ -435,7 +436,7 @@ func (q *AsyncPlanner) Cancel(ctx context.Context, jobID string) (*plannerapi.Jo
435436

436437
// ListJobs merges MDS batches with local not-yet-submitted jobs. Local jobs
437438
// are shown only on the first page so the MDS cursor remains valid.
438-
func (q *AsyncPlanner) ListJobs(ctx context.Context, req *plannerapi.ListJobsRequest) (*plannerapi.ListJobsResponse, error) {
439+
func (q *Planner) ListJobs(ctx context.Context, req *plannerapi.ListJobsRequest) (*plannerapi.ListJobsResponse, error) {
439440
listReq := &plannerclient.ListBatchesRequest{}
440441
if req != nil {
441442
listReq.Limit = req.Limit
@@ -461,7 +462,7 @@ func (q *AsyncPlanner) ListJobs(ctx context.Context, req *plannerapi.ListJobsReq
461462
// unsubmittedJobs returns the non-submitted planner-tracked jobs, newest
462463
// first. Mutable fields are snapshotted under the lock so the rendering
463464
// loop doesn't race against concurrent state transitions.
464-
func (q *AsyncPlanner) unsubmittedJobs() []*plannerapi.Job {
465+
func (q *Planner) unsubmittedJobs() []*plannerapi.Job {
465466
type snap struct {
466467
req *plannerapi.EnqueueRequest
467468
state jobState
@@ -497,7 +498,7 @@ func (q *AsyncPlanner) unsubmittedJobs() []*plannerapi.Job {
497498
// rollbackEnqueue undoes the q.jobs insert from Enqueue when Enqueue fails.
498499
// Not called on processing failures — markFailed keeps those entries in
499500
// q.jobs so callers can observe them.
500-
func (q *AsyncPlanner) rollbackEnqueue(jobID string) {
501+
func (q *Planner) rollbackEnqueue(jobID string) {
501502
q.mu.Lock()
502503
delete(q.jobs, jobID)
503504
q.mu.Unlock()

apps/console/api/planner/impl/async_planner_test.go renamed to apps/console/api/planner/impl/planner_test.go

Lines changed: 27 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -180,12 +180,12 @@ func (b *fakeBatchClient) snapshot() (creates, cancels []string) {
180180
// Helpers
181181
// =============================================================================
182182

183-
// newTestAsyncPlanner builds a AsyncPlanner with the given fakes and worker count
183+
// newTestPlanner builds a Planner with the given fakes and worker count
184184
// and registers a cleanup that calls Close so leaked workers can't bleed
185185
// across tests.
186-
func newTestAsyncPlanner(t *testing.T, bc plannerclient.BatchClient, prov *fakeProvisioner, workers int) *AsyncPlanner {
186+
func newTestPlanner(t *testing.T, bc plannerclient.BatchClient, prov *fakeProvisioner, workers int) *Planner {
187187
t.Helper()
188-
q := NewAsyncPlanner(bc, prov, workers)
188+
q := NewPlanner(bc, prov, workers)
189189
t.Cleanup(func() {
190190
_ = q.Close()
191191
})
@@ -227,7 +227,7 @@ func waitFor(t *testing.T, timeout time.Duration, cond func() bool, msg string)
227227
func TestEnqueueValidation(t *testing.T) {
228228
prov := &fakeProvisioner{}
229229
bc := &fakeBatchClient{}
230-
q := newTestAsyncPlanner(t, bc, prov, 1)
230+
q := newTestPlanner(t, bc, prov, 1)
231231

232232
cases := []struct {
233233
name string
@@ -263,7 +263,7 @@ func TestEnqueueValidation(t *testing.T) {
263263
}
264264

265265
func TestEnqueueWithNilProvisioner(t *testing.T) {
266-
q := NewAsyncPlanner(&fakeBatchClient{}, nil, 1)
266+
q := NewPlanner(&fakeBatchClient{}, nil, 1)
267267
t.Cleanup(func() { _ = q.Close() })
268268

269269
_, err := q.Enqueue(context.Background(), validReq("j1"))
@@ -284,7 +284,7 @@ func TestDuplicateJobIDRejected(t *testing.T) {
284284
return &rmtypes.ProvisionResult{ProvisionID: "p1"}, nil
285285
},
286286
}
287-
q := newTestAsyncPlanner(t, &fakeBatchClient{}, prov, 1)
287+
q := newTestPlanner(t, &fakeBatchClient{}, prov, 1)
288288

289289
if _, err := q.Enqueue(context.Background(), validReq("j-dup")); err != nil {
290290
t.Fatalf("first Enqueue: %v", err)
@@ -310,7 +310,7 @@ func TestEnqueueReturnsPendingPlaceholder(t *testing.T) {
310310
return nil, errors.New("provision aborted by test")
311311
},
312312
}
313-
q := newTestAsyncPlanner(t, &fakeBatchClient{}, prov, 1)
313+
q := newTestPlanner(t, &fakeBatchClient{}, prov, 1)
314314

315315
job, err := q.Enqueue(context.Background(), validReq("j1"))
316316
if err != nil {
@@ -328,7 +328,7 @@ func TestEnqueueReturnsPendingPlaceholder(t *testing.T) {
328328
func TestHappyPathReachesSubmitted(t *testing.T) {
329329
prov := &fakeProvisioner{} // default success
330330
bc := &fakeBatchClient{} // default success, batch.ID = "batch-<JobID>"
331-
q := newTestAsyncPlanner(t, bc, prov, 1)
331+
q := newTestPlanner(t, bc, prov, 1)
332332

333333
if _, err := q.Enqueue(context.Background(), validReq("j1")); err != nil {
334334
t.Fatalf("Enqueue: %v", err)
@@ -372,7 +372,7 @@ func TestSlowProvisionDoesNotBlockEnqueue(t *testing.T) {
372372
return &rmtypes.ProvisionResult{ProvisionID: "p-" + req.IdempotencyKey}, nil
373373
},
374374
}
375-
q := newTestAsyncPlanner(t, &fakeBatchClient{}, prov, 1)
375+
q := newTestPlanner(t, &fakeBatchClient{}, prov, 1)
376376

377377
start := time.Now()
378378
_, err := q.Enqueue(context.Background(), validReq("j-slow"))
@@ -412,7 +412,7 @@ func TestWorkerPoolReachesConcurrency(t *testing.T) {
412412
return &rmtypes.ProvisionResult{ProvisionID: "p-" + req.IdempotencyKey}, nil
413413
},
414414
}
415-
q := newTestAsyncPlanner(t, &fakeBatchClient{}, prov, workers)
415+
q := newTestPlanner(t, &fakeBatchClient{}, prov, workers)
416416

417417
for i := 0; i < submitted; i++ {
418418
if _, err := q.Enqueue(context.Background(), validReq(fmt.Sprintf("j%d", i))); err != nil {
@@ -462,7 +462,7 @@ func TestProvisionFailureMarksFailed(t *testing.T) {
462462
},
463463
}
464464
bc := &fakeBatchClient{}
465-
q := newTestAsyncPlanner(t, bc, prov, 1)
465+
q := newTestPlanner(t, bc, prov, 1)
466466

467467
if _, err := q.Enqueue(context.Background(), validReq("j-bad")); err != nil {
468468
t.Fatalf("Enqueue j-bad: %v", err)
@@ -521,7 +521,7 @@ func TestWaitsForProvisionRunningBeforeCreateBatch(t *testing.T) {
521521
},
522522
}
523523
bc := &fakeBatchClient{}
524-
q := newTestAsyncPlanner(t, bc, prov, 1)
524+
q := newTestPlanner(t, bc, prov, 1)
525525
q.provPollInterval = 10 * time.Millisecond // fast polling for the test
526526

527527
if _, err := q.Enqueue(context.Background(), validReq("j-wait")); err != nil {
@@ -558,7 +558,7 @@ func TestProvisionFailedDuringPollingMarksFailed(t *testing.T) {
558558
},
559559
}
560560
bc := &fakeBatchClient{}
561-
q := newTestAsyncPlanner(t, bc, prov, 1)
561+
q := newTestPlanner(t, bc, prov, 1)
562562
q.provPollInterval = 10 * time.Millisecond
563563

564564
if _, err := q.Enqueue(context.Background(), validReq("j-prov-fail")); err != nil {
@@ -596,7 +596,7 @@ func TestCreateBatchFailureReleasesResource(t *testing.T) {
596596
return nil, errors.New("mds 503")
597597
},
598598
}
599-
q := newTestAsyncPlanner(t, bc, prov, 1)
599+
q := newTestPlanner(t, bc, prov, 1)
600600

601601
if _, err := q.Enqueue(context.Background(), validReq("j-fail")); err != nil {
602602
t.Fatalf("Enqueue: %v", err)
@@ -631,7 +631,7 @@ func TestCreateBatchFailureReleaseErrorIsLoggedNotSurfaced(t *testing.T) {
631631
return nil, errors.New("mds 503")
632632
},
633633
}
634-
q := newTestAsyncPlanner(t, bc, prov, 1)
634+
q := newTestPlanner(t, bc, prov, 1)
635635

636636
if _, err := q.Enqueue(context.Background(), validReq("j-rfail")); err != nil {
637637
t.Fatalf("Enqueue: %v", err)
@@ -664,7 +664,7 @@ func TestCancelQueuedJobBeforeWorkerPicksUp(t *testing.T) {
664664
},
665665
}
666666
bc := &fakeBatchClient{}
667-
q := newTestAsyncPlanner(t, bc, prov, 1)
667+
q := newTestPlanner(t, bc, prov, 1)
668668

669669
// Job A occupies the single worker; job B sits in the channel.
670670
if _, err := q.Enqueue(context.Background(), validReq("j-A")); err != nil {
@@ -714,7 +714,7 @@ func TestCancelQueuedJobBeforeWorkerPicksUp(t *testing.T) {
714714
func TestCancelSubmittedJobForwardsToMDS(t *testing.T) {
715715
prov := &fakeProvisioner{}
716716
bc := &fakeBatchClient{}
717-
q := newTestAsyncPlanner(t, bc, prov, 1)
717+
q := newTestPlanner(t, bc, prov, 1)
718718

719719
if _, err := q.Enqueue(context.Background(), validReq("j-sub")); err != nil {
720720
t.Fatalf("Enqueue: %v", err)
@@ -740,7 +740,7 @@ func TestCancelSubmittedJobForwardsToMDS(t *testing.T) {
740740
}
741741

742742
func TestCancelUnknownJobReturnsNotFound(t *testing.T) {
743-
q := newTestAsyncPlanner(t, &fakeBatchClient{}, &fakeProvisioner{}, 1)
743+
q := newTestPlanner(t, &fakeBatchClient{}, &fakeProvisioner{}, 1)
744744
_, err := q.Cancel(context.Background(), "j-ghost")
745745
if !errors.Is(err, plannerapi.ErrJobNotFound) {
746746
t.Errorf("want ErrJobNotFound; got %v", err)
@@ -762,7 +762,7 @@ func TestCancelDuringProvisioningHonoredAfterCreateBatch(t *testing.T) {
762762
},
763763
}
764764
bc := &fakeBatchClient{}
765-
q := newTestAsyncPlanner(t, bc, prov, 1)
765+
q := newTestPlanner(t, bc, prov, 1)
766766

767767
if _, err := q.Enqueue(context.Background(), validReq("j-mid-prov")); err != nil {
768768
t.Fatalf("Enqueue: %v", err)
@@ -804,7 +804,7 @@ func TestCancelDuringCreateBatchHonored(t *testing.T) {
804804
},
805805
}
806806
prov := &fakeProvisioner{} // default returns ProvisionID="prov-<JobID>"
807-
q := newTestAsyncPlanner(t, bc, prov, 1)
807+
q := newTestPlanner(t, bc, prov, 1)
808808

809809
if _, err := q.Enqueue(context.Background(), validReq("j-mid-create")); err != nil {
810810
t.Fatalf("Enqueue: %v", err)
@@ -859,7 +859,7 @@ func TestCloseCancelsInflightProvision(t *testing.T) {
859859
return nil, ctx.Err()
860860
},
861861
}
862-
q := NewAsyncPlanner(&fakeBatchClient{}, prov, workers)
862+
q := NewPlanner(&fakeBatchClient{}, prov, workers)
863863

864864
for i := 0; i < workers; i++ {
865865
if _, err := q.Enqueue(context.Background(), validReq(fmt.Sprintf("j%d", i))); err != nil {
@@ -892,7 +892,7 @@ func TestCloseCancelsInflightProvision(t *testing.T) {
892892
}
893893

894894
func TestCloseIsIdempotent(t *testing.T) {
895-
q := NewAsyncPlanner(&fakeBatchClient{}, &fakeProvisioner{}, 2)
895+
q := NewPlanner(&fakeBatchClient{}, &fakeProvisioner{}, 2)
896896
if err := q.Close(); err != nil {
897897
t.Fatalf("first Close: %v", err)
898898
}
@@ -916,7 +916,7 @@ func TestCloseIsIdempotent(t *testing.T) {
916916
func TestWorkerCountFloor(t *testing.T) {
917917
prov := &fakeProvisioner{}
918918
bc := &fakeBatchClient{}
919-
q := newTestAsyncPlanner(t, bc, prov, 0) // explicitly degenerate
919+
q := newTestPlanner(t, bc, prov, 0) // explicitly degenerate
920920

921921
if _, err := q.Enqueue(context.Background(), validReq("j1")); err != nil {
922922
t.Fatalf("Enqueue: %v", err)
@@ -934,7 +934,7 @@ func TestWorkerCountFloor(t *testing.T) {
934934
// =============================================================================
935935

936936
func TestGetJobUnknownReturnsNotFound(t *testing.T) {
937-
q := newTestAsyncPlanner(t, &fakeBatchClient{}, &fakeProvisioner{}, 1)
937+
q := newTestPlanner(t, &fakeBatchClient{}, &fakeProvisioner{}, 1)
938938
_, err := q.GetJob(context.Background(), "j-ghost")
939939
if !errors.Is(err, plannerapi.ErrJobNotFound) {
940940
t.Errorf("want ErrJobNotFound; got %v", err)
@@ -949,7 +949,7 @@ func TestGetJobUnknownReturnsNotFound(t *testing.T) {
949949
// Provision and is blocked on the fake), so its status is "provisioning".
950950
// The pending window — between Enqueue and the worker picking the JobID
951951
// off the submit channel — is too narrow to assert on deterministically;
952-
// it's exercised by other tests that use a 0-worker AsyncPlanner.
952+
// it's exercised by other tests that use a 0-worker Planner.
953953
func TestListJobsMergesProvisioningAndMDS(t *testing.T) {
954954
// Block Provision so the local job stays unsubmitted while ListJobs
955955
// runs — that's how unsubmittedJobs picks it up.
@@ -974,7 +974,7 @@ func TestListJobsMergesProvisioningAndMDS(t *testing.T) {
974974
}, nil
975975
},
976976
}
977-
q := newTestAsyncPlanner(t, bc, prov, 1)
977+
q := newTestPlanner(t, bc, prov, 1)
978978

979979
if _, err := q.Enqueue(context.Background(), validReq("j-provisioning")); err != nil {
980980
t.Fatalf("Enqueue: %v", err)
@@ -1024,7 +1024,7 @@ func TestListJobsMergesProvisioningAndMDS(t *testing.T) {
10241024
func TestConcurrentEnqueuesNoRace(t *testing.T) {
10251025
prov := &fakeProvisioner{}
10261026
bc := &fakeBatchClient{}
1027-
q := newTestAsyncPlanner(t, bc, prov, 4)
1027+
q := newTestPlanner(t, bc, prov, 4)
10281028

10291029
const N = 50
10301030
var wg sync.WaitGroup

‎apps/console/api/server/server.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -118,7 +118,7 @@ func (s *Server) StartGRPC(addr string) error {
118118
if err != nil {
119119
return fmt.Errorf("resource manager init: %w", err)
120120
}
121-
s.planner = plannerimpl.NewAsyncPlanner(batchClient, rm.Provisioner, plannerimpl.DefaultWorkerCount)
121+
s.planner = plannerimpl.NewPlanner(batchClient, rm.Provisioner, plannerimpl.DefaultWorkerCount)
122122

123123
// Register all service handlers
124124
pb.RegisterDeploymentServiceServer(s.grpcServer, handler.NewDeploymentHandler(s.store))

0 commit comments

Comments
 (0)