Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add ServerMessage.termination handler #635

Merged
merged 10 commits into from
Aug 23, 2024
Merged

Add ServerMessage.termination handler #635

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
999469b
Add ServerMessage.termination handler
nbsp Aug 16, 2024
3751710
types
nbsp Aug 16, 2024
d0a78af
Create metal-zebras-kick.md
nbsp Aug 16, 2024
47953d1
move job getter to proc_pool
nbsp Aug 16, 2024
6bf239b
Merge branch 'main' into nbsp/add/jobtermination
nbsp Aug 20, 2024
47c00d3
Update livekit-agents/livekit/agents/ipc/proc_pool.py
nbsp Aug 22, 2024
b4834e6
run job termination task asynchronously
nbsp Aug 22, 2024
e6c4000
get_by_job_id
nbsp Aug 22, 2024
2ab00e8
Update livekit-agents/livekit/agents/worker.py
nbsp Aug 23, 2024
6357572
ruff
nbsp Aug 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/metal-zebras-kick.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"livekit-agents": patch
---

Add ServerMessage.termination handler
10 changes: 10 additions & 0 deletions livekit-agents/livekit/agents/ipc/proc_pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,16 @@ def __init__(
def processes(self) -> list[SupervisedProc]:
return self._processes

def get(self, job_id: str) -> SupervisedProc | None:
nbsp marked this conversation as resolved.
Show resolved Hide resolved
return next(
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, by a map I mean to avoid iterating, and creating a dict instead

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i don't think it makes sense for us here, the proc is created and referred to before a job is assigned to it

(
x
for x in self._processes
if x.running_job and x.running_job.job.id == job_id
),
None,
)

def start(self) -> None:
if self._started:
return
Expand Down
9 changes: 9 additions & 0 deletions livekit-agents/livekit/agents/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -485,6 +485,8 @@ async def _recv_task():
self._handle_availability(msg.availability)
elif which == "assignment":
self._handle_assignment(msg.assignment)
elif which == "termination":
await self._handle_termination(msg.termination)
nbsp marked this conversation as resolved.
Show resolved Hide resolved

tasks = [
asyncio.create_task(_load_task()),
Expand Down Expand Up @@ -627,3 +629,10 @@ def _handle_assignment(self, assignment: agent.JobAssignment):
"received assignment for an unknown job",
extra={"job": assignment.job, "agent_name": self._opts.agent_name},
)

async def _handle_termination(self, msg: agent.JobTermination):
proc = self._proc_pool.get(msg.job_id)
if not proc:
# safe to ignore
return
await proc.aclose()
Loading