Scheduling jobs with priority aging

Summary

In this post we describe how we used the technique of priority aging on a set of background jobs in order to avoid starvation. We provide some context about what background jobs are and how they are used. We then explore our specific use case, and assess different approaches against it. We are using Ruby on Rails, but the ideas in this post are stack-independent.

by Markos Fragkakis, Staff Engineer @ Workable

Background jobs and Rails

In this section we give some background on what background jobs are, and how they are supported in Ruby on Rails.

Background jobs

Background jobs are a queuing mechanism that is used to perform tasks asynchronously from the process that scheduled them. In web applications, background jobs are often scheduled in the controller. When responding to an HTTP request, we want to respond as fast as possible, so that the user feels that our application is responsive and has a smooth experience. There is, thus, the incentive to only perform in the controller the bare minimum, and defer anything that can be deferred for later. Any computationally expensive tasks that the user can live without for some time are good candidates to be executed in a background job. In practice, the background job will be processed some milliseconds to some minutes later, depending on the number of background jobs that have been enqueued and the number of workers that process them.

Background jobs in Rails

Like many mature web application frameworks, Rails has out of the box support for background jobs with ActiveJob. There are other background job runners, most of them compatible with ActiveJob, like sidekiq, que, and resque. Depending on the runner you use, you may be using a different store for the enqueued background jobs, as some of them use the database, others use redis. This article is directly applicable to database-backed runners, but you could apply the same ideas to redis-based runners too.

Order of execution

Background jobs have a priority. Jobs that are enqueued with the same priority are consumed with the order they were enqueued. However, jobs with higher priority are always consumed before jobs with lower priority, even if they were enqueued later.

Queues

For high traffic applications, it is a good idea to organize your jobs into groups, called queues (the terminology might be different in other frameworks). A different pool of workers processes the jobs in each queue. For example, you may have a queue for the jobs sending emails, another queue for the jobs performing indexing operations in Elasticsearch, another queue for sending webhooks to 3rd parties, another queue for sending messages to Kafka.

Grouping in queues has some key advantages:

• You can specify different hardware resources for workers of different queues, i.e. simpler jobs can be processed in cheaper workers
• You can define different criteria for scaling in/out for each queue
• You can momentarily turn off a single queue for an operation, without disrupting all the background jobs (i.e. to upgrade your Elasticsearch / Kafka cluster, if you expect downtime)

Who’s next?

With database-backed background job runners, where enqueued jobs are rows in a table, the query executed by a worker to get the next background job (of the queue corresponding to it’s worker pool) to process typically looks something like this:

SELECT *
FROM background_jobs
WHERE queue = 'kafka_queue'
ORDER BY priority, created_at
LIMIT 1

Regardless of the background job runner you use, the idea is the same. You enqueue background jobs, which are then consumed from a pool of workers.

Product backstory

Last year we reimplemented the core of our Applicant Tracking System, and part of this work were bulk actions. In the Workable ATS, our users process job applications, and depending on the customer, these applications may be thousands for a single job. In order to help our customers handle huge numbers of applications with ease, the Workable ATS supports bulk actions (i.e. send email, send assessment tests, move, disqualify and others).

The previous implementation of bulk actions was also the first one, since their introduction. We had already followed the paradigm where we did the bare minimum on the controller, and scheduled one or more background jobs for the deferrable tasks.

Although they had served us for years, our customers’ needs had outgrown them. Workable has more and more enterprise customers that receive huge volumes of applications. For them, bulk actions on 1000 candidates or more are commonplace.

For such cases, even the bare minimum of work became too much for the controller to handle in time. We started having cases of bulk action requests that took too long to respond, or even timed out. Furthermore, we had cases where the background job performing the asynchronous tasks timed out. Finally, there were cases of successful bulk actions that impacted the system in other ways (i.e. locked many rows in the DB for too long, causing timeouts in other requests and background jobs).

Apart from the bulk action-related issues, we had other scalability issues. So, we decided to rebuild the core of our ATS from scratch, having scalability in mind. Part of this reimplementation were bulk actions.

Implementing bulk actions

In terms of scalability, our target was to let our customers perform bulk actions on 1000s of candidates smoothly. In order to avoid controller timeouts altogether, we made our new background jobs completely asynchronous. The controller that serves the background job requests simply stores the parameters, and enqueues a preprocessor job, whose job is to break up the bulk action into smaller chunks, each of which is scheduled as a background job. We sized the chunks so that each takes approximately 5 seconds.

The sequence diagram above shows a simplified timeline of a bulk email action, but the same takes place for every bulk action.

For reasons mentioned earlier, we decided to introduce a dedicated queue for background jobs belonging to bulk actions.

We now had to decide how we could prioritize the jobs. We considered our default scenario, with big and small ongoing background jobs, and then assessed several approaches against it.

The scenario is the following:

1. Bulk action A is scheduled with 100 chunks
2. Bulk action B is scheduled with 10 chunks
3. Bulk action C is scheduled with 5 chunks

Approach 1: Same priority for all jobs (or, how to starve small bulk actions)

The first, naive approach is to schedule all background jobs with the same priority. Jobs will be processed by the workers in the order they are scheduled. On the surface, this may sound fair, as it satisfies the first-come-first-serve approach, which is commonly desired in queuing systems, but it can have negative consequences.

This is what would happen in our scenario. The chunks of bulk action A would be processed before the chunks of smaller bulk actions, starving them out. The graph below shows how A starves out B and C.

Even with multiple workers, it could take a considerable amount of time (minutes) before B and C start being processed, leaving a baffled user wondering why a small task takes so much time.

Verdict: ❌

Approach 2: High priority to small bulk action jobs (or, how to starve big bulk actions)

Another naive approach would be to utilize the priority mechanism, and promote the chunks of smaller bulk actions. We could do this by setting as priority the total number of chunks of the background job (the smaller the number, the higher the priority). This would ensure that even if there are thousands of scheduled chunks by big bulk actions, the chunks of smaller actions would take priority. However, this too has significant dangers. Let’s see how our scenario would play out.

After bulk action A is scheduled the picture would be the following:

Chunks A1 to A10 are consumed, and then bulk action B is scheduled.

Chunks B1 to B5 are consumed, and then bulk action C scheduled.

You can see that the remaining chunks from bulk action A keep getting pushed back.

The graph above shows that the processing of the chunks of B and C can halt altogether the processing of the chunks of A. In high-traffic applications, new high-priority chunks will keep getting scheduled, starving out the chunks of big bulk actions, who are waiting to get a chance to be processed. This will again result in baffled users, who may be OK if their large bulk action takes a lot of time to finish, but they do expect it to finish at some point.

Verdict: ❌

Approach 3: Incremental priority with aging

In order to prevent a bulk action from overwhelming the worker pool, we can give incremental priorities to its chunks. In our example, when bulk action A is scheduled, the priorities would be as follows:

Back to our scenario, if all bulk actions were issued at the same time, the priorities would be the following:

You can see that the next jobs in line for execution are A1, B1, and C1, belonging to different bulk actions. After they are successfully processed, the picture will be the following:

Regardless of how many background chunks are enqueued, and even if there is a single worker in the pool processing the bulk action chunks, the high priority chunks of the bulk actions will get a fair share of processing. Is that enough? Apparently, it isn’t.

The problem now is that as long as there are low priority chunks in a bulk action, they are in danger of being starved out because of new, small bulk actions with high priority chunks.

This can be solved with Aging, a technique used to avoid starvation of scheduled tasks. With aging, as time passes, the older the scheduled task, the higher priority it gets. In this manner, all the enqueued tasks will eventually get a high priority to be processed.

To implement aging in our context, we could increase (numerically decrease) the priority of pending chunks each time a chunk is processed. This could be done with the following query:

UPDATE background_jobs
SET priority = priority - 1
WHERE arguments->>'bulk_action_id' = '1234'

Aging could also be implemented with a time-based trigger instead. For example, we could increase (numerically decrease) the priority of all pending jobs every X seconds.

In the context of bulk action A, this is what the pending chunks would look like after A1 is successfully processed (notice that the remaining chunks have changed priority):

In the complete scenario, the following would be the state after A1, B1 and C1 were processed:

The graph below shows that the chunks of all bulk actions, regardless their size, keep getting processed, without one starving out the others. Now all the bulk actions are guaranteed to finish, at some point.

Our approach doesn’t come without disadvantages. That the update query temporarily locks the rows it updates in the background jobs table. This might not fit well in high-traffic applications with millions of scheduled jobs per minute, as it could block other updates to the table rows (the updates your runner will do, i.e. the updates marking the rows as being processed, or deleting the processed rows).

Verdict: ✅

Conclusion

We described how we used the aging mechanism to implement bulk actions in our system. We hope this post can serve as food for thought for your own queuing. Feel free to share, or drop a comment if you feel like it.