You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
At the moment, we have a less-than-ideal worker experience when using maximum_units_per_worker, as workers are able to see these tasks appear as available in their queues, but after they accept them they're unable to work on them. Instead, it would be great if we could push a qualification to these workers that prevents them from being eligible for the task altogether when they exceed this count.
Idea
One idea we can use is a local qualification to hold the number of tasks currently completed by an individual worker on a given task_name (task_id). We can construct this at the creation of any particular task_id (ie, completed_units_for_<task_name>). Then whenever a Unit is completed, we can increment this qualification value.
When a TaskRun is launched with maximum_units_per_worker set, we create a qualification for <task_name>_<task_run_id>_exceeded_maximum. We can then query all of the completed_units_for_<task_name> qualifications, and assign this disqualification to these workers.
If upon completion of a Unit, a worker now exceeds the maximum_units_per_worker value for a particular task, we can assign the qualification during the task as well.
Possible Implementation
For any new task_id, we should generate a completed_units_for_<task_name> qualification. We can do this by implementing a get_completed_units_qualification method for the Task class, relying on utils.qualifications.find_or_create_qualification.
The Operator class should call Task.get_completed_units_qualification() to ensure the qualification is initialized.
After creating the TaskRun, if maximum_units_per_worker is set, the Operator should call a new method utils.qualifications.create_exceed_max_for_task(TaskRun, max_units). It should create the <task_name>_<task_run_id>_exceeded_maximum qualification, then query for all of the workers with the completed_units_for_<task_name> greater than the provided value, and assign it to them.
To cover any Unit completion, the Agent class in handle_submit should call a WorkerPool method to increment the value for the completed_units_for_<task_name> qualification for its worker. NOTE this qualification increment should use bypass_crowd. If the new value exceeds the TaskRun's maximum_units_per_worker variable, it should assign the exceeded max qualification. The methodology for using such a method from the Agent class can be seen in the update_status method (if self.agent_in_active_run(): ... live_run.loop_wrap.execute_coro(...)
The text was updated successfully, but these errors were encountered:
Overview
At the moment, we have a less-than-ideal worker experience when using
maximum_units_per_worker
, as workers are able to see these tasks appear as available in their queues, but after they accept them they're unable to work on them. Instead, it would be great if we could push a qualification to these workers that prevents them from being eligible for the task altogether when they exceed this count.Idea
One idea we can use is a local qualification to hold the number of tasks currently completed by an individual worker on a given
task_name
(task_id
). We can construct this at the creation of any particulartask_id
(ie,completed_units_for_<task_name>
). Then whenever aUnit
is completed, we can increment this qualification value.When a
TaskRun
is launched withmaximum_units_per_worker
set, we create a qualification for<task_name>_<task_run_id>_exceeded_maximum
. We can then query all of thecompleted_units_for_<task_name>
qualifications, and assign this disqualification to these workers.If upon completion of a
Unit
, a worker now exceeds themaximum_units_per_worker
value for a particular task, we can assign the qualification during the task as well.Possible Implementation
task_id
, we should generate acompleted_units_for_<task_name>
qualification. We can do this by implementing aget_completed_units_qualification
method for theTask
class, relying onutils.qualifications.find_or_create_qualification
.Operator
class should callTask.get_completed_units_qualification()
to ensure the qualification is initialized.TaskRun
, ifmaximum_units_per_worker
is set, theOperator
should call a new methodutils.qualifications.create_exceed_max_for_task(TaskRun, max_units)
. It should create the<task_name>_<task_run_id>_exceeded_maximum
qualification, then query for all of the workers with thecompleted_units_for_<task_name>
greater than the provided value, and assign it to them.Unit
completion, theAgent
class inhandle_submit
should call aWorkerPool
method to increment the value for thecompleted_units_for_<task_name>
qualification for its worker. NOTE this qualification increment should usebypass_crowd
. If the new value exceeds theTaskRun
'smaximum_units_per_worker
variable, it should assign the exceeded max qualification. The methodology for using such a method from theAgent
class can be seen in theupdate_status
method (if self.agent_in_active_run(): ... live_run.loop_wrap.execute_coro(...
)The text was updated successfully, but these errors were encountered: