The Fair Queue - Part 2: A Deep Dive Into the System Design

I have introduced the background of the project. In this blog, I’d like to talk about how I designed the fair queue.

The Aim of the Fair Queue

Actually, this system was not only a fair queue, but also a way to control the number of parallel instances of each workflow. So, the system had the following goals:

1. 01Keep traffic isolated: A long-running workflow task should wait in its own queue without affecting other workflow tasks.
2. 02Control the number of parallel instances: Set a default limit on the number of running instances for each workflow and work with the workflow system to keep the two systems consistent.
3. 03Make the limit configurable: Support different limits for each workflow or tenant.
4. 04Adjust the limit automatically: Get feedback on system load from the workflow system and adjust the limit automatically. (Nice to have)

So, we needed more than a traditional fair queue. We also needed other components to support all these features.

System Design

The system overview is shown in the picture below. The Fair Queue Platform was responsible for traffic isolation, distribution management, priority management, and parallelism management. It also needed a configuration feature so we could set different limits for each workflow or tenant.

Distribution Management: We couldn’t use a single server to manage tens of millions of queues. Instead, each server maintained a portion of the queues, which meant the servers were stateful. I therefore needed to handle rebalancing, failover, concurrency, hot-queue scheduling, and so on.

Priority Management: Workflow tasks came in several types. For example, real-time tasks had a higher priority than scheduled tasks.

Parallelism Limitation: This was one of the most important features of the system. Because the execution time of each workflow was uncertain, we needed to communicate with the workflow system to know the current running state and keep the number of running instances of each queue below the limit.

When a workflow instance finished, it would send a callback, and I would immediately start the next task. However, if the callback failed due to network jitter or another issue, we still needed a way to keep the two systems consistent. So, I periodically checked the state of every running task. I polled all running states every 3 minutes, which did not put a heavy load on the system.

Configuration: We could set a limit for each workflow or tenant. The limit could also vary depending on the subscription plan. In the future, this feature could be exposed to tenants so they could configure the limits for their own workflows.

Dig into the Fair Queue Platform

Distributed Queue Cluster

At the core of the system, we needed to maintain tens of millions of queues. At that time, we didn’t have AWS SQS, and we didn’t want to build another Kafka-style system that stored queues on the file system. We also couldn’t use Kafka because its performance dropped significantly once the number of topics exceeded 200,000.

I talked to our infrastructure team to see whether they could modify Kafka to support our requirements. They said it would be difficult to modify Kafka to support this use case. So, we decided to build our own fair queue system.

I designed the system around in-memory queues backed by an RDB for persistence. When a task arrived, the system first saved it to MySQL and then put it into an in-memory queue.

Because each server maintained a different set of queues, the servers were stateful. When a task entered the cluster, it needed to be routed to the server responsible for that workflow. So, we needed a proxy layer.

The proxy layer could eventually be responsible for several things. It would route tasks to the right server, maintain cluster metadata, track which queues were assigned to which servers and how many tasks were in each queue, and handle rebalancing. In the future, it could also receive information from the workflow system and dynamically adjust the parallelism limit.

However, we didn’t need all of these capabilities in the first version. So, the Proxy Layer was very lightweight. I put it on the same servers as the Memory Queue Layer and used only consistent hashing to route tasks.

For more sophisticated management, the Proxy Layer could later become an independent cluster. In fact, I designed that version after finishing the first release, with a manager elected using Raft. Perhaps I’ll dive into the details of the Proxy Layer in another blog.

Redis vs. Implementing an In-Memory Queue

For the Memory Queue Layer, I initially chose Redis because it separated storage from compute. However, Redis could potentially lose data, so I still needed to maintain consistency, just as I would if I implemented the queues in memory myself.

One advantage of Redis was that I didn’t need to handle rebalancing when a server restarted or lost its connection to the cluster. I’ll come back to rebalancing later.

However, during the design review, my manager didn’t recommend using Redis. So, I listed the pros and cons of both approaches.

Redis

Pros:

1. 01Separated storage from compute.
2. 02I didn’t need to implement rebalancing myself.
3. 03I didn’t need to deal with the problem of duplicated queues when a server lost its connection to the cluster but continued running.

Cons:

1. 01Added another piece of infrastructure.
2. 02With only consistent hashing, it was difficult to handle a hot queue.
3. 03We needed to rebuild all queues when scaling out.

Implementing In-Memory Queues

Pros:

1. 01Easier to extend with more sophisticated management and handle hot-queue problems.
2. 02Easier to scale because queues could be rebalanced automatically.

Cons:

1. 01The overall complexity increased significantly. Handling consistency and rebalancing at the same time was particularly complicated.
2. 02I had to handle rebalancing during restarts and deployments.
3. 03I still had to deal with duplicated queues when a server lost its connection to the cluster but continued running.

I preferred Redis because rebalancing was quite complicated. With an in-memory implementation, I would need to rebuild queues and update metadata repeatedly during deployments. Keeping the queues in Redis would simplify this part of the system.

However, one important issue changed my decision: the Redis service on our company’s cloud platform was not stable at that time. Its SLA was only 99.9%, and it had frequent incidents. Our system needed to support a 99.99% SLA, so I decided to implement the queues in memory instead.

Interestingly, about a year after I left the company, another team needed the same fair queue system. Our team didn’t have enough resources to support their requirements, so they built their own version. They used Redis as the Memory Queue Layer, and it worked well.

In my opinion, there was a better option: use AWS MemoryDB, or a similar managed service from a cloud provider, and use RDB persistence as a disaster-recovery mechanism.

Our company did not allow us to use another cloud provider, and we didn’t have a similar service to AWS MemoryDB internally. So, we had no other choice at the time.

Maintaining Consistency

Because Redis could potentially lose data, I needed to maintain consistency regardless of which approach I used.

I started 10 background threads to scan all the queues and check their consistency on each server. The background threads checked the records in batches, using IN and GROUP BYand made sure the queries could use an index. Additionally, to reduce memory usage, each queue loaded a maximum of 1,000 records into memory, and those background threads are also responsible for maintaining this queue size.

Rebalancing

Rebalancing was a real challenge.

Imagine that one server goes down. All the queues on that server need to be moved to other servers, and their in-memory queues need to be rebuilt.

To do that, I had to maintain metadata for all queues so that I knew which queues needed to be rebuilt. I also needed to update that metadata during rebalancing.

This meant we needed a master server to coordinate the queues, manage the metadata, and dispatch commands to rebuild them.

There was another problem: when servers restarted or were deployed, rebalancing could be triggered many times within a few minutes. That made it clear that the proxy layer should eventually own this logic and be separated from the Memory Queue Layer.

I didn’t implement this design in the first version.

In the first version, I only used consistent hashing and stored the metadata in Redis. When a server restarted, it did not trigger a full rebalancing process. The metadata simply recorded which queues belonged to each server and told the server which queues it needed to rebuild after restarting.

Race Condition

Imagine that a server lost its connection to the cluster but continued running. This could result in the same workflow being owned by two servers, which meant two queues could exist for the same workflow. Both queues could then trigger tasks at the same time.

We could not allow the same task to be executed twice, so I needed to handle this race condition.

When triggering a task, I first updated its state to "processing" and used optimistic locking to prevent two servers from processing the same task at the same time.

Additionally, I also asked the workflow system to maintain idempotency with the trigger ID.

Dead Letter

There was another problem. After I triggered a task, the workflow system might not actually start executing it, perhaps because of network jitter or another issue.

I would retry it three times, and if it still failed, I would put the queue ID into a dedicated table.

I deliberately did not wrap these operations in a transaction. One reason was performance. More importantly, a transaction could not fully solve the problem where a queue had no task running.

For example, the server could crash during the transaction. In that case, the transaction might be rolled back, but the task would still not have been triggered. There would be no one left to trigger it again.

So, remember that I had a background thread maintaining consistency every 3 minutes. It would also find queues that had no task running, then try to trigger them again or put them into the dead-letter table.

For queues in the dead-letter table, I retried them 10 more times at longer intervals. I didn’t implement exponential backoff, either. If a task still failed after 10 attempts, I sent an alert.

Cleaning Up Old Data

As there would be billions of tasks in the tables, performance would decrease as the number of tasks grew. So, once a task had been completed for seven days, I transferred it to HBase and deleted it from MySQL.

Well, that’s the design of the Fair Queue System I implemented for our project. Some other features, such as configuration and automatic limit adjustment based on system load, would be handled by the independent proxy cluster. Perhaps I will summarize that part in the future.

Next, I’ll move on to the financial system I worked on at Alibaba. Its business logic was significantly more complex.