Execute tasks with BEP Worker

The Worker scaling strategy is a mechanism for calculating an optimal Worker quantity to satisfy, to the maximum possible extent, the multiple task execution requirements on a cluster with fixed limited resources (non-elastic cluster). The strategy is supposed to adapt when cluster resources are added or removed.

Resource pool balancing

Single task type

If BEP executes only one task type, for example, only CT tasks, there is no contention for cluster resources as there is a single Worker pool. Therefore, the CT Worker pool is scaled up to consume the needed cluster resources. BEP reacts as fast as possible to any peak load, processing a new huge batch of tasks without a warm-up period.

Multiple task types

Submission of multiple task types to BEP, for example, CT and AutoML Execution tasks, can lead to resource contention. In this case, BEP tries to balance Worker pools so that estimated task completion times based on the average task processing speed are approximately the same.

WMS provides more resources to a slower pool, but the maximum number of Workers does not exceed the maximum number of tasks during the observed time. If the number of tasks decreases, the number of Workers decreases too.

In the simple case, when average task processing times for different types are roughly equal, the Worker pool size is proportionate to task counts:

BEP worker scaling strategy monitors tasks continuously and rebalances Worker pools based on several metrics: task count, task publishing speed, task consuming speed, and so on.

Task execution requirements

The following requirements apply to task execution on a cluster:

Requirement	Details
Ensure effective cluster resource utilization.	That means, provide maximum resource utilization for maximum total task processing throughput. For instance, you have ten nodes and 106 tasks for a Worker type. In this case, WMS needs to start as many Workers to process these tasks and as fast as possible. And these are not just one or two Workers. The fewer Workers are employed, the longer it takes to process all tasks, which deteriorates cluster utilization.
Guarantee fair-share and progression.	Each task type must get its amount of resources without having to wait until previously submitted tasks are completed to free up resources.
For instance, you have 106 tasks submitted for a Worker type. WMS already started a large pool of Workers for another task type, consuming all cluster resources. In this case, the system needs to re-balance Worker pools—that means, to reduce the Worker pool for the first task type and to provide processing resources for the second one.
Avoid frequent switching.	For instance, you have a cluster with limited resources, tasks for two Worker types, and roughly identical task streams. In this case, constant back-and-forth switching could look like this. First, you start Worker A with more tasks. Five seconds later, when Worker A processed enough tasks, and Worker B has much more tasks, you stop Worker A and start Worker B. Another five seconds later, you stop worker B and start worker A. The same cycle repeats over and over again. In this case, you are likely to spend more time on Worker starts or stops and less time on processing tasks.
The scaling mechanism reacts fast enough to task stream changes.	For example, as soon as tasks appear for a new Worker pool, the latter is to be started fast enough even if the operation requires stopping some existing Workers. The pool must not wait for the completion of any existing task streams.
Take into consideration fragmentation of free resources.	For instance, you need 16 GB RAM to start a new Worker, and it is possible to provide the amount of memory by stopping a particular Worker on a specific node. Assume, some node already has 12 GB free RAM, and the Worker process on it has 4 GB RAM allocated. Thus, stopping the single Worker is more preferable than stopping multiple Worker processes on random machines.
Build tasks as short as possible.	Currently, the task processing time is configured to 12 hours maximum. The restriction relates to the RabbitMQ server's delivery acknowledgment timeout. Though, the general recommendation for successful task execution is the shortest possible processing time. You can set up the execution time limit while designing a task.

note

By default, a graceful shutdown using a "death pill" is applied to kill a Worker. Though, there is a case when a "force kill" is used (equivalent to SIGKILL in a Unix-like OS), for example, if a pool has waiting tasks, none of its Workers is launched, and the same situation was in the previous iteration. For more details, see Manage BEP Worker execution | Graceful Worker shutdown.

Scaling flow

Below, you can see a sample scaling flow with three iterations.

Workers are scaled up or down as described below:

1. WMS triggers scaling iterations every five seconds, which you can adjust by changing the bep.wms.schedule.task-monitor-fixed-delay property. It makes requests to TDS, Marathon, and Mesos to collect the following data:

   • Queue statistics from TDS
   • Job info from Marathon
   • Cluster resources info from Mesos

   Those are three different requests, so the collected states can be inconsistent.

2. Some pools are excluded from the collected stats when they meet any of the following criteria:

   • A pool is in the error mode.
   • There are failures in Workers, in which case the pool is excluded, and the error mode is enabled.
   • A Worker jar is not available.
   • worker.yml is invalid.

3. An iteration proceeds only if there are pools without running or waiting Workers, or 60 seconds (bep.wms.scaling.scale-after) passed after a prior iteration. It is essential for fast reaction to new pools. In other cases, avoid frequent switching.

4. Orphan Workers are killed. Those are workers that were launched in Marathon but have no associated tasks or queues.

5. Waiting Workers are killed. Those are Workers that were launched in a prior scaling iteration but are still in the Waiting status. It's better to kill them to calculate new states from scratch.

6. The best state for the scaling strategy is deduced.

7. The best state is applied to the cluster:

   1. Calculations are made to find out how many Workers of each type are started or killed on each node.

   2. Requests are sent to Marathon to start or kill Workers. Killing is based on the bep.wms.scaling.supervisor setting:

      - If `bep.wms.scaling.supervisor=smart` (by default), Workers are killed with ["death pills"](/platform/docs/10.2.8/automate/workers/worker-execution#graceful-worker-shutdown). Workers are not killed instantly, so started Workers can be in the Waiting status until the next iteration is started.
      - If `bep.wms.scaling.supervisor=direct`, Workers are killed via Marathon.

   3. If the first Worker for a pool cannot be started for too long, the algorithm selects a few nodes and kills Workers on them without "death pills."