Job queues are deceptively tricky
typesanitizer.com
typesanitizer.com
A buffer to accumulate data that blocks when it’s full allows you to handle bursty loads. It solves the back pressure problem of readers and writers operating at different speeds. It doesn’t over consume resources.
It also solves the architecture problem of when to trigger work. Both consumer and producer act on the pipe imperatively, rather than one end being imperative and the other being a declarative graph of callbacks (all those “reactive” libraries).
Even using a term like “back pressure” is a tell to me that someone is confused snd doing something architecturally wrong.
Since the pipe is generally unidirectional, the sending process has no way of knowing whether the receiving process has received, or even more successfully processed, anything sent. For that, one needs to make a pipe in the opposite direction and that is not as easy anymore; it also requires building your own protocol to identify and acknowledge sent work items.
There are solutions for that, though:
* You can observe and/or persist intermediate data with `tee`
* You can use a named pipe, whose lifetime is bound to an inode instead of the processes
* You can dup the read end of the pipe and pass the extra file descriptor to a crash handler process
None of these are esoteric. You do need a basic understanding of Unix primitives, but any half-decent computer science course will cover them.
So are most message queues.
If the producer needs to know when the consumer has finished work, that's not a queue; that's an RPC.
Most message queues want to be able to distribute work to multiple parallel consumers, either for performance, redundancy, re-deployment of the consumers, and so on. Like, sure, you can do that with named pipes, but they're not durable; their state can be lost in the event of a crash or reboot.
Other than like ... in-process, thread-to-thread message queues, almost all message queues are used because they provide some form of durability (either by persisting messages to disk, or just by virtue of running on a separate server/process than much more crash-prone producers/consumers).
> imperatively, rather than one end being imperative and the other being a declarative graph of callbacks (all those “reactive” libraries).
MQ client libraries are heavily reactive because they need to work with MQs that are network services. If every consumer that ever wanted to fetch a message had to issue a network RPC poll/timeout cycle, and they were all doing that constantly, that'd be both a lot of traffic for the MQ to handle and a lot of network chatter. Some MQs do use (or support) the RPC poll model, but especially among the more performant ones, most of them are push-based so clients can just sit in an epoll/select statement waiting for the message queue to send them traffic.
That said, you're totally right that plenty of MQ client frameworks/libraries are way over complicated and want to own your whole program's entry point and design. Callbacks are probably a necessary outgrowth of the underlying network behavior, but tortured whole-program-event-loop-ownership architecture is not.
> Even using a term like “back pressure” is a tell to me that someone is confused snd doing something architecturally wrong.
I don't think your other points support this; you said yourself that pipes' bounded nature is good, so what's the problem here?
One of my takeaways is that if you do not handle on data transfer in the http request/response then you are choosing to exit the pipe model. Once you do that, yes all kinds of crazy async problems exist which require very complex tools to wrangle.
See the sibiling comment from Google engineer. It’s better to do work synchronously than delay it until later.
So now we get your list of requirements that MQ is for. Ok but how did we get here? What problem are we trying to solve? The answer is usually a performance problem for which this is a bandaid.
Another possible model for this problem is email. You can find some great threads here about large amounts of infrastructure being replaced with Unix email tools. Although I have yet to try it!
> If every consumer that ever wanted to fetch a message had to issue a network RPC poll/timeout cycle, and they were all doing that constantly, that'd be both a lot of traffic for the MQ to handle and a lot of network chatter
This makes no sense to me. A while loop blocked on a socket read with a producer with a write does not generate any extra network “chatter”.
Anytime you are using callbacks you are choosing not to have a thread with imperative logic.
> I don't think your other points support this
Correct that was not a conclusion.
What I’m trying to say is if you want to send data between two systems and you have producers and consumers at different rates, you don’t have to do anything special. The UNIX kernel is designed to solve this problem.
So when I hear that, my inclination is that we don’t understand the capabilities UNIX already offers and I’ve yet to be wrong.
I agree that if that's the problem you're trying to solve, MQs are a poor stop-gap and not a general-purpose solution.
But many (most?) MQs aren't deployed for that reason; rather, they're used to either a) defer work whose latency characteristics are incompatible with the producer's runtime (e.g. sending email which might take minutes/hours to be accepted upstream can't be done synchronously from an HTTP request handler), b) handle work which doesn't need to be done transactionally and which has a high probability of needing to connect to resources whose uptime you don't control (queues make retries easy), or c) work that needs to be batched or otherwise completed on a schedule or dimensionality that's not available in the producer.
> A while loop blocked on a socket read with a producer with a write does not generate any extra network “chatter”.
True, and some MQs support that pattern, but it's rare for the same reason that most network server applications' core loop doesn't wrap a blocking read. While-blocking-read plays poorly with timeouts, restarts, and multiplexing/wait-any. This isn't unique to MQs.
> Anytime you are using callbacks you are choosing not to have a thread with imperative logic.
I mean, it's pretty easy in most languages/runtimes to turn a callback into an imperative wait. When it comes to network servers specifically, I think exposing the lowest-runtime-overhead model as the primary API (selectors and callbacks) and letting users add control flow primitives on top of that is preferable, but reasonable minds can differ here. This also isn't unique to MQs.
> if you want to send data between two systems and you have producers and consumers at different rates, you don’t have to do anything special. The UNIX kernel is designed to solve this problem.
For ephemeral/can-be-lost local traffic, you're partly right (pipes/fifos are a simple and widely available tool that can help with this, though you have to layer some protocol-ish logic on top to make them work reliably between consumers that expect particular structures and don't speak ASCII/newlines alone). For durable local traffic, or remote traffic, UNIX doesn't have a prebuilt primitive. TCP tx/rx window sizes aren't surfaced such that buffers and backpressure can be reliably interpreted by userland to make delivery decisions.
> latency characteristics are incompatible with the producer's runtime
> handle work which doesn't need to be done transactionally and which has a high probability of needing to connect to resources whose uptime you don't control (queues make retries easy
I agree. I think you do need to break the request/response pipe for this use case.
I traditionally have treated this as a state in a database. I know there are limitations and I can see MQ being a solution. I want to try the email idea sometime.
I do think queues are a poor technology for dealing with performance problems in a web stack.
> This also isn't unique to MQs.
Agreed. My ranting about async and callbacks is an ongoing project.
I think pipes solve producer/consumer problems in a way most engineers don’t appreciate.
> While-blocking-read plays poorly with timeouts, restarts, and multiplexing/wait-any.
The primary problem solved by not using blocking primitives is to try to free up OS resources from threads (green threading). Why would an internal queue have that problem? It’s not accepting arbitrary connections.
Likewise; thanks for engaging constructively as well.
> I traditionally have treated this as a state in a database.
Which is a super appropriate tool many (most) times! I've been using "MQ" in this context refers to the conceptual capability to externalize/distribute/persist work with a push/pop API. That can be provided by a database, a UNIX pipe, or a more traditional message broker--the reasons-to-adopt and costs/benefits are largely the same.
> I want to try the email idea sometime.
Email (and the print spool) are some of the oldest message-queue primitives on UNIX systems, I think? I'm not sure if POSIX/shmem MQs predate them or not, but they're all quite venerable proofs that the pattern has its uses. I'm sure you can (maybe others already have?) use SMTP and the mailq to manage application-internal queue communication. I'm not sure how low-overhead or performant it'll be given that it's highly optimized for one/very few consumers and delivery attempts >1 being attempted on a pretty large time scale compared to application traffic, but it'd be an interesting experiment to be sure!
> If you’re anything like me, you would probably have said Parallel Spawn, Prefer New, and Wait are perhaps defensible, whereas Prefer Old feels weird/backward.
it was pretty obvious I was not anything like him. My intuitive answers are pretty different.
- Parallel Spawn is useful, but it's orthogonal to the rest. Even if you have 4 workers, you'll still have to worry about hitting concurrency limit once you have enough jobs. What is it even doing in this list?
- Wait is very useful for non-scheduled tasks: if user uploaded 100 files to process, you better process them all. Sometimes you need limit, sometimes you do not (let them queue for a while until devops notices and either allocate more workers or clear them and has some harsh words with consumer).
For scheduled tasks, "Wait" seems much less useful. I can come up with a reason but they are all somewhat convoluted - perhaps you are hitting 3rd-party service, and it has a quota, so you've decided to use scheduler to avoid hitting ratelimits?
- "Prefer Old" is normally the best way. You repack takes 3-8 hours, so you set your timer to "every 1 hours, skip if running already" and you can be sure that your job finishes and the next one will start.
- "Prefer New" seems almost useless. You've already spent all this effort doing the job, why are you cancelling it and throwing away the results? If you want to add a timeout, add a timeout, preferably to specific operation. For example, if there job starts by fetching data, and this fetch can be super slow, use 'Prefer Old' and put a timeout on the fetch part. This way your job won't be interrupted if the fetch just finally succeed minutes before next scheduled interval hit.
Oh, and re "If the Prefer Old semantics are not offered, you can’t really emulate them using the two primitives of regular scheduling and limiting concurrency." - you totally can. Set concurrency to 2, and add as a first thing: "fetch the list of jobs running; if there is anyone except me, exit right away".
Really, not that tricky at all.
One example was running certain ad auctions when rendering websites, or something like that. You don't want to delay serving the side, if the ads are delayed.
So you have a certain wall clock time budget until the rest of the page is assembled to be sent to the user, and if you can fit your ad-serving in there, that's good.
If you have more work than you can currently handle, then it makes sense to continue with the newest open request after you handled the previous request.
The actual system was a lot more complicated, and combined a short, bounded queue on the inside with a large stack on the outside or something like that.
Similar considerations can apply, when you are one of many competing market makers for some financial assets on an exchange.
Basically, if you have a situation where serving quick is a lot more important than the distinction between late and very late (or even dropping the request).
The obvious use-case is if the result depends on a state and the state has changed since the job has started. Examples would be updating the landing page when a new article has come in (you don't need the outdated landing page w/o the new article) or if you did a code change while compiling (assuming it's not a commit).
(because with scheduled events, if you missed deadline once, there is a good chance that restarting with exactly the same deadline will miss again, and again, and you will never finish. Better do "prefer old" and at least have _some_ finished data)
1.) What they do to your 95th percentile latency. Users are often very sensitive to tail latency: a service that responds in 150ms 19 times and then takes 2s on the 20th is still perceived as annoyingly slow. With job queues, the reason for that slowness could be as simple as "it was the 20th request to arrive during a period of high demand, and backends couldn't keep up". The whole point of queues is so you can gracefully handle this case without overprovisioning your backends by a factor of 20x, but if the user is still going to consider this a miss anyway, you have to overprovision the backends anyway. There isn't really another way to handle this other than having spare backend capacity. Also note that in many cases the user hitting "refresh" doesn't cancel the existing queued job, it just adds another one to the queue. Which brings us to...
2.) They can turn simple failures into cascading failures. There were several postmortems that went something like "Service X became overloaded because of an unexpected flood of requests, leading to several individual replicas shutting down. This led to more requests being routed to the remaining replicas, which overloaded them too and led to all of Service X going down. When SRE attempted restart Service X, requests queued in the job queue were all retried en masse, which led to an overload of the partially-restarted service and a subsequent failure. SRE had to limit requests upstream and manually drain all job queues and bring Service X back cluster by cluster to restore service health."
The root principle here is that any distributed system needs a concept of backpressure. When critical downstream dependencies are overloaded, they need to pass this information back up the stack to the entry point, which needs to start denying requests from the user or do a simpler fallback that doesn't put load on the overloaded service. Naive queueing does not work, because the requests are still sitting there in the queue waiting to overload the downstream service once it becomes available again.
You can bolt backpressure onto a job queue system (by eg. rate-limiting requests to a service that has just come back up, or rate-limiting based on response time, and/or falling back to simpler algorithms), but at that point, it's a backpressure system, not a job queue. The semantics are very different from a system that guarantees eventual delivery, just not sure when. You need to be able to handle partial failures and adapt with different algorithms at multiple points within the system.
I think one subtlety worth bearing in mind here is that Google, at least during my tenure (2006-2014) didn't actually have a proper message queueing system outside of Gmail, which was used only for email delivery. It offered engineers:
1. RPC with infinite backoff/retry (a fun default).
2. Batch jobs that processed files.
but there was no equivalent to a standard enterprise MQ product like ActiveMQ, Artemis, Oracle AQ and so on.
So when we talk about "job queues" it's worth being very precise about what is meant. A standard enterprise message queue broker doesn't have problems with overload because workers pop work off the queue at whatever speed they can operate. The problem of cascading failures and services coming back only to be immediately overloaded again was a problem caused by the design of Google's infrastructure, in which RPCs would back up in memory in the clients and be retried in a loop until the service came back. So of course this required a lot of custom work to create proper backpressuring, which wasn't normally done and especially not on interactive serving paths, leading to this kind of repetitive failure mode. It was all made harder by the practice of making everything fully async, so thread pool sizes also didn't exert backpressure.
Looking back on my time there, one of the pieces of "normal" enterprise infrastructure that I really think Google could have benefited from was a proper JMS compliant scalable message broker. You can buy these - the Oracle Database has one - but Google neither bought one nor developed its own.
Probably they have long since rectified this oversight.
It’s a hard/interesting problem, and harder still once you’re running across a lot of machines — but if they all get slower under the load it turns out that it’s easier to scale / work out a good balance of idle capacity to guarantee x time sla under x requests.
Fast ramp up for additional capacity is important too, but less so if you know to start the process once median execution time drops to some % of your worst target
Well, memory is one of these resources that you are often locked on.
If you have enough memory, running 20x the load just goes 20x slower. But if memory is congested, then this can go arbitrarily slower than running your jobs one after another. Eg when you are swapping to a spinning disk.
Shared memory can be a problem, sure, which is why I don’t generally use that either. Erlang, of course, generally does not use shared memory outside of some cases with ETS etc which must be used carefully but i’d rather solve those problems myself.
Concurrency systems in other languages i’ve written, for example Go, there are ways to architect to avoid it also. I’d rather go slightly slower and copy be value than have to solve mutex contention and trying to make everything atomic and so on. YMMV, I don’t work in HPC just large complex busy systems.
Multiple tasks can share the CPU and just get a bit slower. But if you are out of RAM, you are better off running one thing after another.
Whether you hit swap or OOM was a distraction.
Indeed you need backpressure but the traditional methods (CPU usage or similar metrics) are difficult because many consumers aren't high on those metrics --imagine a messaging plaform, pure IO. Also you'd have to tailor to the consumer itself and that's difficult, which is what you mention on the next-to-last paragraph.
In the end I helped solving it by scaling based on queue size and input/output rates, agnostic to the consumer itself, but with the hypothesis that you can scale consumers linearly (or at least monotonically, some sublinearity is allowed). The queue scaler watches for incoming and outgoing traffic on the queue, plus items on the queue itself, and it can scale from 0 to 11 in seconds for gusts, then shutting everything down.
It's a satisfying problem to work on, but its proper solution demanded quite the investigation. Now every queue we've got in the system is managed by this autoscaler -- except when we can't ensure linearity.
I’m skeptical. You can support a pretty massive messaging system with one box.
What did you try? What went wrong?
When you want to switch from no processing to say 2M messages than need to be sent within 5 minutes it's difficult to appropriately provision without fast scale out.
This reminds me of some research I read about in the 1980s or 1990s on perceptions of the speed of command line commands. If command time varied over a range from nearly instantaneous to say 100 ms fairly uniformly through that range, people would perceive the system as overall being faster when the researchers added a variable delay to all the commands that made them all take 100 ms.
Humans apparently really like consistency.
> When SRE attempted restart Service X, requests queued in the job queue were all retried en masse, which led to an overload of the partially-restarted service and a subsequent failure
...and this reminds me of something else, from around 1983. I was working at a small Unix workstation maker. The guy in the office across the hall found one morning that the battery for the clock on his workstation had died, and the system time had come up after boot as the Unix epoch.
He shut down, put in a new battery, booted, and then set the clock to the current time, 13 or 14 years after the epoch.
Almost immediately his hard disk light came steadily along, and he could hear the disk furiously seeking, and the system became completely unresponsive.
It turned out AT&T cron in the early '80s wasn't smart about time changes. It had tried to all at once every cron job that should have run in the 13 or 14 years that the time just jumped.
SchedMD the leading developer of Slurm has recently being acquired by Nvidia, while Slurm remaining free and open source, but somehow it's Wikipedia entry is not yet updated accordingly.
[1] Slurm Workload Manager:
https://en.wikipedia.org/wiki/Slurm_Workload_Manager
[2] Nvidia Acquires SchedMD (7 comments):
The single most important lesson from queuing theory for software systems is the non-linear relationship between utilisation and latency.
As system utilisation approaches 1.0 (100% capacity), the average waiting time does not scale linearly, it scales hyperbolically. A system running at 95% utilisation is vastly more fragile and slow than one running at 80%, even though the load difference is minor.
Think about a secretarial pool with five secretaries. The pool is at 95% utilization. This might mean that one of the secretaries is at 100% utilization and the other four are at 93.75%. It might mean that all of the secretaries are at 95%. It might mean that four of them are at 100% and one is at 75%. It might mean something else.
Going back to that third case, we can easily see that with the system at 95% utilization, at any given moment in time there is a 75% chance that the system is busy and unable to begin a new task, and a 25% chance that the system has capacity to begin a new task right away.
In the full-average case where all five secretaries are at 95% utilization, the odds that the system has spare capacity at any given moment are 22.62% (= 1 - 0.95 ^ 5), not too different from the case where four secretaries are kept busy 100% of the time.
Are you talking about trying to interpret malformed data?