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

ParlAI Assignment Fixes pt 2 #878

Merged
merged 3 commits into from
Aug 2, 2022
Merged

ParlAI Assignment Fixes pt 2 #878

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
32e9320
bump hydra
JackUrb Aug 1, 2022
1a293af
Unused print
JackUrb Aug 1, 2022
de3fd7a
different concurrency approach for worker pool assignments
JackUrb Aug 1, 2022
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
1 change: 0 additions & 1 deletion mephisto/abstractions/blueprints/parlai_chat/parlai_chat_agent_state.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,6 @@ def _load_data(self) -> None:

def get_data(self) -> Dict[str, Any]:
"""Return dict with the messages of this agent"""
print(self.metadata)
return {
"outputs": {
"messages": self.messages,
Expand Down
20 changes: 12 additions & 8 deletions mephisto/operations/worker_pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -265,26 +265,30 @@ async def _assign_unit_to_agent(
agent,
)
else:
# See if the concurrent unit is ready to launch
assignment = await loop.run_in_executor(None, unit.get_assignment)

# Set status to waiting
agent.update_status(AgentState.STATUS_WAITING)

# See if the concurrent assignment is ready to launch
logger.debug(f"Attempting to launch {assignment}.")
agents = await loop.run_in_executor(None, assignment.get_agents)
if None in agents:
agent.update_status(AgentState.STATUS_WAITING)
return # need to wait for all agents to be here to launch

for queried_agent in agents:
if queried_agent.get_status() != AgentState.STATUS_WAITING:
Copy link
Contributor

Choose a reason for hiding this comment

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

Under what condition would this not be true? When things haven't synced to the worker threads?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

When two workers connect at the same time, and both have had an _assign_unit_to_agent thread called, and have thus both had Agents created, but only one of those threads has made it to the update_status call.

logger.debug(f"Delaying launch of {assignment}, should retry.")
return # Need to wait for all agents to be waiting to launch

# Mypy not-null cast
non_null_agents = [a for a in agents if a is not None]
# Launch the backend for this assignment
registered_agents = [
self.agents.get(a.get_agent_id())
self.agents[a.get_agent_id()]
for a in non_null_agents
if a is not None
]
if None in registered_agents:
# an Agent has been matched to the Assignment, but isn't yet watched
# by the pool. Wait for that thread to catch up
logger.debug(f"Delaying launch of {assignment}, should retry.")
return

live_run.task_runner.execute_assignment(assignment, registered_agents)

Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ exclude = [
python = "^3.7"
requests = "^2.22"
sh = "^1.12"
hydra-core = "^1.1.0"
hydra-core = "^1.2.0"
tqdm = "^4.50.2"
websockets = "^10.1"
pyyaml = "^5.4.0"
Expand Down