Сложности проектирования систем очередей задач: важность учета пределов и моделей отказов

Job queues are deceptively tricky One of the fun things about being a programmer is that as I look more into systems that I didn’t know much about, what superficially appears to be a simple system actually reveals interesting facets of underlying complexity. In other words, reality has a surprising amount of detail. In this post, I want to talk about job queues, which I’ve been thinking about for the past few days (and while drafting this post in my head, I realized I’d thought about them for longer at a previous job, but not nearly with as much clarity.) What do I mean by “job queue”? I mean a system where there is some notion of submitting batch jobs, scheduling them, and running them. Generally, the system is expected to be FIFO or FIFO-like, but that’s not required. Usually, the queue bundles together a native way to schedule jobs periodically – this lets you specify submissions using configuration files (e.g. using JSON or YAML). Job queues arise naturally in all sorts of situations where throughput requirements are high, but latency requirements are not that high. Continuous integration is a common example. Another example is summaries for the purpose of data analysis. Over the past year or two, a handful of lenses that I’ve found useful when thinking about system design are: - Being wary of queues: Queues tend to typically be close to full or close to empty, requiring some thought around appropriate sizing. I’ve also learnt a bunch of counter-intuitive behavior of queues when it comes to latency from Marc Brooker’s blog.In blog posts I’ve read about queues, the focus is generally on latency, not on throughput, which is I think partly why I never really mentally linked “queues” and “job queues” earlier. - Limits: This is based on the Tiger Style guide written by the folks at TigerBeetle, which talks about explicit limits on various things. On generalization, this leads to the notion of tolerances and having budgets for modular reasoning.Funnily enough, near the start of my career at Apple, I remember being mildly bemused at certain discussions where teams would discuss how much memory budget they would have to bargain for, especially for low-level code. I have grown more respect for that approach over time. - Fault models: Roughly speaking, a fault model describes your assumptions related to errors and reliability in your dependencies. - I’ve learnt a bunch on this topic from the talks and tweets by the TigerBeetle CEO Joran Dirk Greef, as well as writing by Alex Miller. I will try to demonstrate how these lenses can apply for system design later in this post. The problem at hand At $WORK, we have a background job for packing “reference repos”. A reference repo is a git repo which has been aggressively repacked – if you’re unfamiliar with repacking, you can think of this as more “aggressively compressed”. These are stored in object storage, and a new repo can be set up on a machine by downloading the reference repo, and fetching a delta of changes for the tip of the default branch. When it comes to very large repos, for the downloader, this approach helps cut down latency compared to the more typical approach of doing a git clone operation. The actual details of how git does aggressive repacking is not super relevant for this post. What does matter is that there are, roughly speaking, two forms of repacking available: - Wholesale repacking: This will lead to git ignoring any baseline information about how packing should be done, and recompute everything from scratch. Say for the repo under test, this takes 7 hours.If you’re surprised at the thought of a git operation taking several hours, (1) you should be happy that you don’t have this problem and (2) some of the sub-operations here are seemingly single-threaded, regardless of repo size. - Incremental repacking: This will lead to git reusing the baseline information already stored in the repo about packing, so it will only repack things which have changed. So if you did a clone, then did afetch, then only the newer changes fromfetchwill be repacked, and only some cursory checks will be performed on the history received from thecloneoperation. Say this takes 2 hours. OK. So that’s the setup. We have the option of doing the more expensive thing, which takes 7 hours. This buys us a smaller reference repo, which means faster downloads, faster un-tarring and lower disk usage. To give a size ballpark, a wholesale repacked repo can be up around 50-60% smaller than an incrementally packed repo. We have the option of doing the cheaper thing, which takes 2 hours. This buys us more up-to-date reference repos, which means that if the downloader still cares about getting the latest changes, the delta it will fetch subsequently will be smaller, so it’ll be faster and put less load on the server. But also, if the extra disk usage is in the hundreds of megabytes or gigabytes, then that’s not great. One more thing to note is that since this is a code repository, downstream consumers are much more active on weekdays than on weekends. The best of both worlds? A natural next step upon seeing the above dichotomy is to propose the following: why not do the wholesale repacking on the weekends, and do the incremental repacking on the weekdays? That seems like it buys up-to-dateness on weekdays (where more consumers are active), and the wholesale repacking on the weekend would ensure that the reference repo size grows more slowly. For simplicity, let’s say we’re writing the reference repos with some key <myrepo>-<timestamp>.tar into some bucket. To keep the size of the incrementally packed repo smaller, one can either: - Write the wholesale repacked repos using the same key scheme, and have the incremental repacking jobs bootstrap from the last successful repack (wholesale or incremental), instead of from a cloneoperation. - Write the wholesale repacked repos using an altered key scheme, say <myrepo>-<timestamp>-packed.tar, and bootstrap the incremental repack from that key location. However, this potentially creates the need to update other consumers so that they don’t lag behind too much on weekends (assuming incremental repacking is not running on weekends). Let’s say we go with the first one because it seems simpler. So now the question is, how are we going to handle this scheduling? Typical job queues only expose limited control of scheduling to clients for good reason – providing a large number of knobs increases the risk of surprising scheduling decisions. When I wrote the “natural next step” above, I’m essentially thinking from the point of view of writing the control loop. But when I’m using a job queue, I do not (by definition) have access to the control loop – someone else has written that loop, and I need to see what configuration knobs are available to me to customize the behavior. For now, let’s say the queue exposes two configuration knobs: - Scheduling interval: The job is started on a timer configured with this interval. - Concurrency limit: The maximum number of running jobs for the particular configuration. Before we started this optimization journey, let’s say these configuration knobs were set as: - Scheduling interval: 9 hours, to leave some headroom from the 7 hour running time of the job. - Concurrency limit: 1, because there’s not much point in having concurrent jobs do the same thing here. Again, a natural next step here might be to think: “So the 9 hour interval was sufficient for a 7 hour running time job. Since the incremental repack job on weekdays will take 2 hours, let’s set the interval to be 3 hours. On the weekends, when the wholesale repacking job is running, it will not have finished in 3 hours, so even if the timer triggers again, the concurrency limit will prevent any other jobs from running, so things should be fine.” Unfortunately, dear reader, this simplistic reasoning does not work. The set of possible semantics For a moment, let’s stop thinking about trying to think about how we can use the job queue. Rather, let’s think about implementing a job queue. Let’s say there’s a job which is running J1 based on some configuration J. After the scheduling interval elapses, say we want to create a new job J2, and J1 hasn’t completed yet. What should we do with J2? Essentially, there are only four options: - Parallel Spawn: If the concurrency limit is higher than 2, then we can just start running J2. - If the concurrency limit is 1, then we have three choices: - Prefer New: Stop J1and startJ2. - Wait: Wait for J1to finish and then startJ2. - Prefer Old: Auto-cancel J2and letJ1continue running. - Prefer New: Stop Pause for a moment here, and ask yourself what’s your intuitive answer to the following question: which of these possible semantics are worth implementing or could be called reasonable? 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. For Prefer New vs Prefer Old in particular, you may have justified this to yourself with reasoning like: “if this situation happens, the job owner probably cares about more recent results than older results so it makes sense to support Prefer New but it seems strange to support Prefer Old, who would ever want that?” Recall the lenses from earlier: vigilance about queues, limits and fault models. So the first thing to note is that the Wait option requires a (logical) per-job-config queue. The queue should probably be bounded (per Limits). OK, so how should the limit be decided? What should happen when the limit is hit? At what utilization level should we start alerting (i.e. what’s a soft limit for the queue size)? Should the queue be strictly FIFO? Those are good questions to ask! There’s no one-size-fits-all answer, but an important thing to realize is that these questions should be asked, if the Wait semantics are supported. If we think about limits, the Prefer New semantics imply that the scheduling interval is also a hard limit. Conceptually, these are different things – you could imagine varying them independently if the latter was offered as a separate configuration knob. Lastly, what are the fault models under which these semantics make sense? - Prefer New: - J1’s running time exceeded the scheduling interval because of some non-deterministic reason (e.g. a rate limit being hit). We are optimistic, so it’s possible that- J2will not hit the same issue.- In other words, we expect the job to be successful again sooner rather than later. - Wait: Same as above. Additionally, there is a non-functional requirement (or assumption!) that dropping jobs is not preferable. - Prefer Old: Assume that - J1is not hung, so it will probably finish if given more time.You would still want to have a hard limit separate from the scheduling interval to prevent runaway resource usage. We are pessimistic, so we assume that- J2’s running time will also exceed the scheduling interval. So it’s wisest to cancel- J2, so that at least- J1has a chance to finish. I just wanted to do a simple thing To jog your memory on what we were trying to do: - We wanted to run the same job with a fast path on weekdays (taking 2 hours) and a slow path on weekends (taking 7 hours). - We had one configuration knob for the scheduling interval. We were thinking of setting it to 3 hours to get more up-to-date results on weekdays. - We had a concurrency limit of 1 to avoid running multiple jobs at the same time. Let’s try to see how such a configuration would fare on the weekend for the different semantics we discussed:For simplicity, assume that scheduling operations are instantaneous. - Prefer New: When the 7 hour job would hit the 3 hour mark, it would be canceled. A new 7 hour job would be triggered, and canceled after 3 hours again. Essentially, we’d waste 48 hours of compute across 16 jobs, because none of these would ever finish. - Wait: Over the 48 hours of the weekend, we’d trigger 16 jobs, with a total compute demand of 16×7 = 112 hours. If the queue limit was higher than (112-48)/7 = 64/7 = 9.15, we’d emerge out of the weekend with about 64 hours of pending work. However, the weekdays only offer (24×5×(3-2)/3) = 40 spare hours, so we wouldn’t finish draining this queue before the start of the subsequent weekend. - Prefer Old: In 48 hours, we could almost finish seven 7 hour jobs, with ~1 hour of pending work. We would’ve canceled 9/16 jobs. We’d get back on schedule quickly on Monday since we only have 1 hour of extra work left from the weekend. So it seems like Prefer Old is the best fit here. Yes, it might still seem counter-intuitive; we’ll get to that shortly. The other extra fun thing here is that if you were using the Wait strategy, you can easily end up masking the problem without actually helping. Maybe you got paged on the weekend for the queue being non-empty for an extended period of time. So you decide to increase the concurrency. You pick a good round number like 4. The new runners start to chew through the jobs, great! Now, you have these expensive jobs running for 7 hours, but most of their results are only useful for ~3 hours on the weekend, before they get superseded. 😬 Coming back to the appropriateness of Prefer Old, let’s look again at what we wrote for the fault models: - Prefer New: [..] we expect the job to be successful again sooner rather than later. - Wait: Same as above. Additionally, there is a non-functional requirement (or assumption!) that dropping jobs is not preferable. - Prefer Old: [..] we assume that J2itself will also exceed the scheduling interval. [..] For the schedule that we were trying to have on the weekend (7 hour jobs with a 3 hour interval), it closely matches the assumptions of Prefer Old very well, but it directly goes against the assumptions of Prefer New and Wait. Additionally, our weekend workload is totally fine with dropping jobs in case something is running. But the Wait semantics would make us pay dearly for something we did not care about, and would likely require more expensive solutions on top. If the Prefer Old semantics are not offered, you can’t really emulate them using the two primitives of regular scheduling and limiting concurrency. For the workload above, if you had a richer primitive such as cron-like scheduling, you could split the work into two separate jobs – one which only runs on weekdays, and one which only runs on weekends. Things could’ve been worse Earlier, we were thinking about fault models. That was cute, right? Imagine a separate system compared to the one we’ve been talking about. There’s a global concurrency limit to keep costs under control. You have a bunch of different jobs coming in, mostly kicked off in the background. Everything goes in a FIFO queue. Nobody put in a global size limit. Every few months, a customer comes in and tries the feature related to the job queue. They kick off some new jobs manually and see the queue position as 100K+ (or even, 1M+). They’re like “uhh, this feature doesn’t seem to work?” They escalate to a support engineer. It becomes a ticket on your kanban board. An engineer goes in and clears the queue by running a DELETE manually, with some displeasure. The new manager asks “what can we do to solve this problem in a systemic way.” You spend a few days analyzing the problem. You look at the running times. You look at the queue growth charts. Ah, it’s hard to predict how long these jobs take, or how many of them come along. Maybe we reduce the jobs created? You look at the database columns, and you realize you don’t actually have enough information to go on to properly check which jobs came from the same configuration. The flexibility of job creation makes it seem quite tricky to retrofit a notion of job de-duplication. The manager comes back “can we have an 80-20 solution?” You add a hack which tries to pick an already queued job which looks similar-ish, and overwrite it in-place when doing the enqueue operation. The queue is now mostly bounded in size, because newly queued jobs overwrite old ones if present, but there are still some configuration knobs which a customer could tune and break the system out of equilibrium. You declare it “user error” to tweak those configuration knobs, and move on to the next ticket. While writing this, I was reminded of apenwarr’s line from The log/event processing pipeline you can’t have: (I suddenly feel a lot of pity for myself after reading that paragraph. I think I am more scars than person at this point.) Just randomly, for no good reason whatsoever. ANYWAY Here’s some hope as life’d be too depressing otherwise It seems possible to design systems while thinking explicitly about queues, limits and fault models – and thinking about these seems to lend itself to designs which degrade more gracefully, and handle unexpected situations better, especially if the design is made under pessimistic assumptions. In particular, explicitly spelling these out is valuable as a system designer, because it lets potential consumers evaluate the system quickly for their own workloads even before they go through a detailed step-by-step simulation in their head about what would happen in different situations. Conversely, when you’re using a new system that you haven’t used before, it’s useful to find out this information if it’s documented (or you have access to the code). That way, you can check how well the designer’s assumptions apply to (or are violated by) your intended workload. In particular, when there are configuration files involved, especially those which tweak control flow, not just data that flows through, it’s valuable to pause for a bit to dig into the various error cases that are involved in the corresponding control loop, because configuration files tend to hide this information away. For systems which lend themselves to simple models, you might even be able to model the different cases using just pen-and-paper, like we’ve done above, without the need for simulations. Doing so can help illustrate gaps in our own thinking, where it’s easy to gloss over what happens when things don’t go quite as swimmingly as we expect.