mirror of
https://github.com/livekit/livekit.git
synced 2026-10-23 13:02:53 +00:00
agent: run the agent name lookup off the worker pool (#5043)
This commit is contained in:
1 parent
6342d8fd1b
commit
752bd339b7
2 files changed
+37
-2
No files matched your search
@@ -60,6 +60,41 @@ func TestAgent(t *testing.T) {
|
||||
require.Fail(t, "job assignment timeout")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("dispatch burst does not wait for the CheckEnabled timeout", func(t *testing.T) {
|
||||
bus := psrpc.NewLocalMessageBus()
|
||||
server := testutils.NewTestServer(bus)
|
||||
t.Cleanup(server.Close)
|
||||
|
||||
rpcClient := must.Get(rpc.NewAgentInternalClient(bus))
|
||||
registered := must.Get(rpcClient.SubscribeWorkerRegistered(context.Background(), agent.DefaultHandlerNamespace))
|
||||
defer registered.Close()
|
||||
|
||||
worker := server.SimulateAgentWorker(testutils.WithJobLoad(testutils.NewStableJobLoad(0)))
|
||||
worker.Register(testAgentName, livekit.JobType_JT_ROOM)
|
||||
select {
|
||||
case <-registered.Channel():
|
||||
case <-time.After(time.Second):
|
||||
require.Fail(t, "registration timeout")
|
||||
}
|
||||
|
||||
client := must.Get(agent.NewAgentClient(bus, agent.Config{}))
|
||||
t.Cleanup(func() { _ = client.Stop() })
|
||||
|
||||
const jobCount = 60 // more than the client's 50 pool workers
|
||||
start := time.Now()
|
||||
for range jobCount {
|
||||
go client.LaunchJob(context.Background(), &agent.JobRequest{
|
||||
JobType: livekit.JobType_JT_ROOM,
|
||||
Room: &livekit.Room{},
|
||||
AgentName: testAgentName,
|
||||
})
|
||||
}
|
||||
require.Eventually(t, func() bool {
|
||||
return len(worker.Jobs()) == jobCount
|
||||
}, 2*agent.CheckEnabledTimeout, 10*time.Millisecond)
|
||||
require.Less(t, time.Since(start), 2*time.Second)
|
||||
})
|
||||
}
|
||||
|
||||
func testBatchJobRequest(t require.TestingT, batchSize int, totalJobs int, client rpc.AgentInternalClient, workers []*testutils.AgentWorker) <-chan struct{} {
|
||||
|
||||
+2
-2
@@ -240,7 +240,7 @@ func (c *agentClient) getDispatcher(agName string, jobType livekit.JobType) *ser
|
||||
}
|
||||
|
||||
done := make(chan *serverutils.IncrementalDispatcher[string], 1)
|
||||
c.workers.Submit(func() {
|
||||
go func() {
|
||||
agentNames.ForEach(func(ag string) {
|
||||
if ag == agName {
|
||||
select {
|
||||
@@ -253,7 +253,7 @@ func (c *agentClient) getDispatcher(agName string, jobType livekit.JobType) *ser
|
||||
case done <- nil:
|
||||
default:
|
||||
}
|
||||
})
|
||||
}()
|
||||
|
||||
return <-done
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user