Transcript

Yu: We'll be talking about leveraging determinism, or in more words, trusting deterministic execution to stabilize, scale, and simplify systems. My name is Frank. I'm on the engineering team that built and runs the Coinbase Derivatives Exchange. I've got two things I'd like you to come away with. First, I'll try to make the case that if you've got important logic, make it deterministic. You'll be happy you did. If you've already got deterministic logic, don't be afraid to rerun it anywhere in your stack, in production, even, for efficiency and profit. As a disclaimer, we'll largely be going over practices directly from our experience running financial exchanges. We don't rigorously prove any of the concepts we present here. Any references to the canon or any prior art we're bringing together is for inspiration, and as a case study for what we've done. Just remember, n equals 1 here, it's all anecdotal.

About Us and Our Problems

Let's talk us and our problems. We built and run what is now the Coinbase Derivatives Exchange, formerly a startup called FairX, which built a CFTC government regulated trading exchange for folks to trade financial contracts with each other. You can submit us an order to buy or sell something for some price, and we'll let you and the market know asynchronously when you submit that order, and when you trade, and what price you get. Think of exchanges as financial infrastructure. Everyone who participates in finance relies on us to provide the real prices of things traded on our platform, and relies on us to fairly assign trades between folks who want to enter into financial transactions. First thing to think about when running an exchange is mission critical code. A bug on our system could result in catastrophic problems for both ourselves and all of the clients in our platform, because we're dealing directly with up to millions of dollars of client funds. Potential losses we risk are multiple orders of magnitude more than any revenue we might make from a transaction. In a previous life running a currency exchange, our clients would trust us with millions of dollars for each request, and maybe we'd be happy to earn 0.00001%, give or take a zero on that transaction. Business logic has got to be rock solid. Exchanges absolutely have to be correct.

The second thing about running a financial exchange is that many of our partners expect us to operate as close to instantaneously as possible, and also to be as consistent as possible. If we're the ground that folks stand on for financial transactions, we've got to be predictable and fast. Specifically, here, we expect our 99th percentile response time latencies to stay comfortably under a millisecond. Finally, we got to keep records. We need to allow regulators, customers, debuggers like me to reconstruct the state of the market or the system exactly as it was at any given microsecond instant, for the last seven years. All of this adds up to a super high bar of quality on our core business logic. You might ask here, if risks are so high, correctness bugs are catastrophic, and performance bugs get instantly noticed, how do we actually add features to the system in any reasonable timeframes? We've got to make sure our core logic stays simple. In principle, we want to minimize the number of arrows and boxes in our architecture diagram. We don't want a lot of services that can fail independently in a transaction and result in a partially failed system. When we handle financial transactions, we handle them in an atomic and transactional manner. We also avoid concurrency at all costs in our logic. In the last 10 to 20 years, multiprocessor systems have offered programs an interesting deal with the devil. Run stuff in parallel and you get some more throughput and performance. You've got multiple CPUs, why not assign work to all of them at the same time? You just have to make sure you don't have any race conditions. Your logic is correct for every possible interleaving of every line of code in your concurrent system. Then, if you're trying to reproduce a bug by running code over again, you can't count on it to happen, because you've left things up to the OS scheduler. Here, I claim that simplicity through deterministic execution, it'll get you simplicity, but it can also get you more performance and throughput than concurrent logic.

Deterministic Execution

Let's describe deterministic execution a bit. You can largely consider a traditional program with a variable set to various values in the program counter set to some line of code as a state machine. The idea is this, inputs come into the system and contact with the state of your system and may change it. Inputs also come into your system and may result in some new data or output getting created. In practice, we'll talk about inputs as requests into the system, API calls, and the change state given a request as the new state of the system. For example, if Alice deposits $1,000 into her account, then the new state of the system would reflect an additional $1,000 in her balance. We also talk about the output of a system as events. This is when Alice submits an order, and we have to asynchronously send a push notification to the rest of the market, informing them of the new bid for a given thing at some given price. You'll hear me use inputs and requests, and then outputs and events interchangeably here. We're drawing out that equivalency. A system is deterministic now if given an exact set of inputs, in the same order, we get the exact same state and outputs. In practice, if you take all of your requests, and you sequence them all into a log, and then you apply determinism here, [inaudible 00:07:01], we'll effectively get replicated state and output events. Again, if we sequence Alice first depositing $1,000, then submitting an order to buy something using that $1,000, our deterministic system should never have Alice using money for an order before she deposited it, even if I rerun my service over again, 1000 times at any point.

Benefits of Determinism

For some of you, replicated state machines might ring a bell, and you might think about consensus. This is a segue to some of the benefits for having a replicated state machine through deterministic execution. The first one here is high availability. We all have high availability concerns these days. Let's say I've got my service running in production, and then some hardware failure crashes my service. Now I'm concerned. I've got an outage. I'm getting paged in the middle of the night. I've got to start stuff and recover lost data. It's no fun. It wasn't even my code that necessarily caused that hardware failure. If I have a deterministic system, I can actually replicate my log of incoming requests to other services running the same logic, and they'll have the exact same state. If invariably, my service goes down, through no fault of my code, I'm still fine. I can pick a new primary service. Good thing, we have redundancy through determinism. That said, replicating every request into our system is a ton of work. The Raft algorithm says, don't even start changing your system until you're sure that your state machine is replicated to most of your machines. If we're going to operate at microsecond timescales, we'll need an implementation of consensus that's really fast. Thanks to Todd Montgomery, and Martin Thompson of Real Logic, we've got a very fast open source Raft cluster that we can put our deterministic logic on and get all these nice high availability properties with negligible performance overhead. Hopefully, the civility benefits to determinism are clear.

Let's talk performance. Another thing that's nice when your logic is deterministic and serialized, is you can actually run everything in one thread and make use of the consistency and throughput you get from single threaded execution. One barrier to that consistent performance is actually giving up control to the operating system scheduler and hoping that you won't get context switch out when you start handling a client request. Again, in the vein of keeping things simple, let's tell that operating system to pin our business logic thread directly to the physical core or a hyper threaded core and not have our logic be preempted for anything else, and remove the complication of having to think about OS scheduling when developing for performance.

Changing tack, it should go without saying that a deterministic program should be easier to test than a non-deterministic program. Certainly, in cases of concurrency where the spaces of tests you need to write expands by n times m of the interleavings of the lines of code running in parallel. There's some other nice things you can do if you've got one deterministic monolith that handles every request as a transaction to test. Each of your user stories basically turns into a test case. You can make use of time travel to forward your system to a very specific date so you don't crash or have issues every leap year or something like that. Despite all that testing, when something still seems weird with the system, let's say I'm in prod again, or staging, or anywhere, and something weird starts happening on my service. If I'm in prod, I'm concerned. I don't know what's going on. I wish I added log lines for this case. I hope this doesn't impact too many people. If I've got deterministic logic now, when the weird behavior starts, I can go ahead and copy the request log, run it on my laptop. When the problem comes up, maybe I was running the system in a debugger. Now I know exactly why my program got into that weird state, because I'm inspecting all the states in memory and I'm figuring out the problem. I've still got some weird behavior in prod, but at least I maybe have a plan for how to deal with it. This idea of being able to take a running system in production, and rerun patch with log lines on your development environment, it's a superpower. It also can't be overstated how useful it is for maybe finding some pathological performance situations or tracing bugs. You rarely say, it works on my machine, again, for logic bugs, if your system is deterministic like this.

Business Logic

All this stuff. Great, sold. We've built this. What you're seeing on the screen here is our business logic running in Raft consensus that's in blue, surrounded by gateways or API endpoints, just like an HTTP JSON endpoint, or special message endpoints that you send finance messages to, those are all in green. Here's a simple message flow to illustrate our system. Clients are going to send us requests through our gateways, like Alice wants to buy 2 nano-Bitcoin contracts for $20,000. That request gets forwarded onto our business logic service, the Raft monolith, and we go ahead and output events via multicast here to clients that connect to our bare metal data center. This is where the latency sensitive nature of our system impacts our architecture. For the non-latency sensitive use case, or the high throughput use case, or the cold storage use cases, we're going to go ahead and send events over some transport to the cloud. Then forward them to regulators sitting behind the gateway or a database, so we can actually keep track of what happened long term. This big, red pipe, in other stacks, you might think of them like Kafka or a message queue. In general, you've got requests coming in to your business object in blue, and then events coming out of the system, and sometimes over a big transport to another region.

Here's one thing we noticed. At this point in time, we're in production. We're a startup at this point, and we noticed something interesting about our requests. That request, the tuple of Alice buying 2 Bitcoin contracts, it's 5 pieces of data, really. It's about maybe 100 bytes coming in, but what comes out after we handle that API call might be huge. What if Alice at this point successfully buys 2 Bitcoin, and let's say she trades with 2 different people. What are we sending over that big, red pipe? First, we're sending the event that we successfully handled Alice's or his order. Then, let's say she trades 1 Bitcoin with Bob. We tell Alice that she bought a Bitcoin from Bob. We tell Bob that he sold a Bitcoin contract to Alice. If Alice also trades with Charlie in that request, we got to tell Alice that she traded with Charlie, and we got to tell Charlie that she sold to Alice. One request to buy 2 Bitcoin contracts suddenly becomes a mouthful of events once it goes through a logic. Here's the kicker. We're sending events from our bare metal data center over to our AWS data center. We're finding out that AWS's compute charges are fairly reasonable. Then, network ingress and network egress costs were nuts, we were being charged for our data bandwidth. At the time we were a starving startup, at least burn rate conscious, we weren't super happy about lighting money on fire sending all this data. Yes, big question here, can we optimize? What's our problem?

What we're saying here is that in our situation, and in many situations, in general, business logic is fast and cheap, but reading and writing the data is what's actually slow and expensive. How does the fact that our business logic is deterministic help here? Here's a proposal. How about if when handling requests, we still go ahead, and instead of sending a ton of events over this expensive network connection, and lighting money on fire, we replicate only that request over the network. Then we continue on, we only replicate that request to our downstream services. That doesn't really work, our downstream services want to see the impacts of that request, not the request itself. Let's just rerun our business logic on the same machines as our downstream services, and send our red big blasts of events over IPC, or interprocess communication. That effectively means that we've now lighted all the network costs of egress events: way less network usage, less burning money.

Let's catch our breath here. For the first time, we had to blink a few times. We're saying, we can go from sending fat events everywhere, from one little request message, and we can go to just sending the requests everywhere? How does this work? It works because everything is deterministic. Our monolith is just a function that turns requests into their output events. It does the same thing every time as long as the ordering of requests are the same. If we didn't have a multicast in the data center, if our data center on the left here was actually just in the cloud, we could apply that same trick. We could replicate requests to a monolith that is localized to our gateways, and send all that through IPC. What this has opened up is a whole axis of tradeoffs you can make, where you can now replicate compute instead of data and cases.

Replay Logic to Scale and Stabilize

Why does this work? Here's what we noticed about inputs and requests. They're often smaller than outputs. When we receive an order or an API call, maybe one of the first things we do about it when we handle it, is we actually stick an ID on it, and we want to identify the state of that request. We're sticking a bunch of state on it. We're contextualizing that request with our own services' business context. Here's another interesting thing we noticed. Inputs can probably be more consistent than our output events. You all know about 99 percentile latencies. How often do we think about our 99th percentile network load given some request? Here's an example. Let's say Alice sends us one request to sell 1000 Bitcoin contracts at just any price, because she just wants to cash out. Those 100 bytes coming to your system, they're going to cause possibly thousands of events to happen, because she's going to trade with the entire market. What you get here is a ton of events coming out of one request. If you've got an event driven system, you've got a positive feedback loop in here somewhere where you're creating more events from handling events. This is how you get what's called thundering herd. This is how massive autoscaling systems in data centers and in the cloud, can go down. Your network buffers completely fill up, because you're blasting tons of events everywhere. Your services time out, and we're not able to keep health checks up, or things like that.

Why are we saying here that these input requests can be more predictable, and outputs can't be? The size of an input, a sequence into your log and the rate of them, they can be validated and rejected. Alice sending one API request is actually not a lot of network usage. If a client had a bug and was just spamming us with tons of requests, we can actually validate their input rate with rate limits, throttles. Your API gateways, your load balancers actually should be doing it for you. When you talk about the size of inputs, if there's malformed data, you can validate that before it gets into your business logic. It's not quite so easy with outputs and events. An event, in some sense of fact, it's a change in your system because of some request. You don't reject an effect of a request. You can, but it's pretty difficult. Also, it's often unclear when a request might cause just a landslide of events to come out. Those are what you would basically call pathological cases. You've got to analyze them to be able to predict them and mitigate them. In this case, if your entire network utilization is just your input rate that is validated by all of your external endpoints, then you'll actually get some pretty consistent network utilization for all of your replication here. You can protect from thundering herd here by not sending blasts of events. If you're replaying your logs in production, that's how you'll stabilize your system. Then, because it's less network utilization, it'll also help you scale.

Here's a third benefit. When you replicate compute, you can also simplify your downstream code. All those blue boxes you saw on that diagram, monolith, monolith, monolith, it's all the same code. It's deduplicated. What that means is, you don't have a lot of endpoints where you might be reimplementing your business logic just so you can materialize it, or have a better view of it. That said, deduplication is optimization here, it's not architecture. We're not offering up our firstborn at the altar of DRY, but it's still pretty cool. Let's take a look at it in action. Again, here's my system, with monoliths everywhere. A common use case for any given system is you might want to accept user queries. If Alice wants to get her active orders, she'll send the request through one of our endpoints. Traditionally, you would forward that query to your datastore. In our case, that's the Raft monolith. The datastore responds with the data that you're asking for, so blue coming in, red coming out, and that's why these arrows are purple.

These requests might come in from all over the world. You might have clients in different regions. That might be another gateway talking to your datastore to go fetch data. We try to solve some of the scaling problems with this by maybe introducing caches in the middle. What that means is you still have to keep that cache up to date. What are you doing? Are you sending data to that cache? Are you doing some a lazy LRU type thing? It's all just a lot of complication here. There's a fairly obvious optimization that can be made, given that we've got monoliths everywhere. Instead of going to our main piece of business logic, our main datastore, we can just make an IPC call to the local monolith running alongside your endpoints and get the data back super quickly, and get fantastic query performance in this case, because you're totally localized. We can remove a box from our architecture diagram.

What if we have no choice? We've built this great system, we got monoliths everywhere, and then a lot of new features keep getting added: new query opportunities, new gateways, new endpoints, new functionality. Alice wants to go get her balances, and a bunch of other things. You've got all these requests coming in from the outside. To scale, we might need a cache again. A really great cache, in this case, is actually our monolith. What we can do is, instead of replicating events, or updating maps, key-value stores, we can actually just replicate that same sequence log of requests. Now that monolith can actually answer queries over the network to all of these gateways who don't necessarily need to be running the monolith in the same machine. Here's an example of maybe going a little more in the middle of the spectrum between replicating all the compute and replicating none of the compute.

Challenges and Considerations

What are some things we got to think about when we've built a system like this? Here's a listicle. One, we've got to make sure to replicate well tested code, or else your bugs will replicate too. This architecture assumes that you've got some battle tested, hardened logic that's really important to your business, and that you'll actually test and make sure that it is stable. Because if you don't, you'll just have bugs everywhere. Here's the other thing. When you've got a deterministic system, you got to make sure that all of your replaying services are actually deterministic, they cannot drift. A common source of drift is when you change your business logic. One point you've got to keep track of is you've got to respect old behavior when replaying inputs. What that means is, let's say on Thursday, you had some undesirable behavior on your system, and you handled some requests, and then you fixed it on Friday, and you deployed it. If your downstream monoliths are replaying logs from Thursday, then they need to make sure to do that incorrect, undesirable behavior before Friday comes and we actually enable that new behavior. What this means is deploying code should not change the behavior of your system, you've got to decouple that and enable behavior changes with an API request to your business logic.

Four, we often need randomness in systems. It gives us uniform distributions, and fairness, load balancing, maybe some security benefits. As it turns out, you can do randomness deterministically. There's tons of deterministic pseudo algorithms out there for you to pass the seed in. Then, all of your systems that are replicating the input log are going to get the same nicely distributed, random looking output, but it'll all be exactly the same. Five, this one's fairly subtle. If you've got a single threaded deterministic system, then any long task is going to block every other task in your input log. This is called the convoy effect. If you've got requests, especially that maybe will explode into more work and then do a bunch more work in a transaction, consider breaking up that computation into multiple stages and allow more request messages to come in, while you're tracking the state of that computation. This is a tradeoff, of course, between better consistent performance dealing with the convoy effect, and the complexity of having to manage additional state because now you've just split up your computation into multiple stages.

Also, if you want a fast Raft service, everything that you're persisting between requests, you must fit in memory. That said, there's a lot of commodity RAM out there now, you'd be surprised. You can fit quite a lot of customer data into hundreds of gigabytes of RAM, especially if you're being maybe judicious about how you're storing that. If you're storing data in fixed length fields, and you're avoiding arbitrarily nested and variable length data, you can easily store millions of accounts worth of information in a megabyte times the number of 64-bit integers you're using to actually persist that info. Also, you'd be very surprised what fast single threaded code running on one core can get done. If you're not concerned about writing to disk with each message and you're not IO blocked, you can easily keep your processing times of your API calls maybe even under a microsecond. That gets you a throughput of a million messages per second all on one box. Really think about that if you come to the decision where you might want to shard, because sharding introduces tons of complexity into your system.

Nine, keep your 99s and 99.9s down. You've got that single threaded service, your response rate for everything is impacted by your slowest requests. When running in this manner, and if you want consistent performance, you've really got to apply some craft to make sure you avoid pathological performance cases, and make sure you're not slowing down all of the other requests by iterating through too many things or something like that. Last, our inputs are more consistent if we validate them to a band of acceptable size. You've got to make sure that your log, your monolith is protected from any client that maybe suddenly goes nuts, because you can now easily denial of service your entire system if a client goes haywire, and you're not applying rate throttles.

Simplicity

All this determinism is motivated by a call to simplicity. In general, we say that simplicity is going to get you stability. It's going to get you performance and development speed. I'm sure you've all seen plenty of talks and had engineering mentors talk about the virtues of keep it simple. Hopefully, I don't need to sell simplicity to you all. What I'm hoping I got across is that deterministic execution help keeps things simple. Again, you've got some important logic, make it deterministic. If you've got deterministic logic, trust it, replay it, run it anywhere in production, on your laptop, all over the place.

Questions and Answers

Montgomery: In order for this to work, you need the same state everywhere, that is, no external state. How do you do that? I suspect having all the market data at hand with prices changing very frequently is too large to hold in your local data storage.

Yu: If you're going to run everything basically in a Raft monolith, you are relying on the fact that your performance and everything is basically all in memory, and you rely on a persisted append only log of the requests to give you all that state. What that means is any external state, you want to be persistent, and you want to have access to in between your messages. They've got to fit in memory. I think we addressed this one in the listicle. Basically, if you avoid variable length data, so avoiding strings, using lots of fixed length fields, keep on to registers and primitives, you can store quite a lot of data in memory on one process. The question is, when do you shard? The complexity with sharding, and setting up basically some broker to demultiplex, and all that, is a pretty huge decision. In general, cloud systems and all of that tend to go to that shard mechanic looking for shard axes maybe a little too quickly, because having fewer boxes really does help a lot. Yes, this does not work if all of your market data at hand does not fit in memory, but your market data should ideally be maybe three, four numbers, so that's 32 bytes, 64 bytes. You can fit quite a lot of market data in your systems.

Montgomery: I think also, too, there's one difference that I know, it's very common between a lot of the financial systems that operate in this space and a lot of more consumer facing systems. That is things like you're not storing names, you're not storing a whole bunch of other static data in those systems. Because you're just not actually operating on them. How often does someone change their address? Does your business logic at that point need to know address, for example? There is a lot more that you can get in terms of useful information into a system into memory, when you take a look at it and go, "I don't care about some of this external stuff, because it's never going to actually change my business logic," for example. There is an assumption that some of that information that's normal for other systems to just have available is usually not available, because it's pared down.

How do you keep determinism when executing the same input run at different times when replicating? On a stateful system, the market changes over time.

Yu: This piggybacks off an important piece of architecture, called the sequencer architecture, where you basically rely on one leader, the leader of your Raft monolith to timestamp and provide the order of the requests that come in. The problem is that having a gigantic consensus pool, where you're expecting the shard outward to get you easy reads can quickly backpressure your leader. This idea of basically rerunning monoliths outside of Raft consensus, basically, as an optimization, it's basically separating the writes, which is the writes to the sequencer, and the reads, which are queries of your state machine. The timestamps stay the same. What happens is Alice's request to come in is timestamped at Monday, 11:59 p.m. That request that is replicated over to a monolith, maybe it's replaying some historical stuff, like maybe it's replaying it days later on Wednesday, but that message still has that timestamp of Monday, 11:59.

Montgomery: In other words, you can say that time is actually injected into the log, you're not grabbing time some other way. It's just a function of the log itself.

Yu: Agree. If you are comfortable relying on that one clock, you also avoid things like a clock skew. Fundamentally, if you have clocks running on multiple machines, that's another distributed consensus problem. Ideally, you try to rely on your solution to the consensus problem to give you the things you need to keep things simple.

Montgomery: This can only work if there is no external state, that is, services are purely compute reliant and nothing else. That's really how services exist, services rely on external state. Is there a version of this that works with services with external state?

Yu: Just refining what we mean by external state, I think of that as maybe a source of truth, or something that is basically feeding in information into your state machine. What this means is, let's say, we've got some clients who have programs that are interacting with our system, strictly the state of their programs is external to our system. Then, how does that state that they've got in their system come to us? They come to us in the form of requests. What that means is, if you want any state to end up in your state machine, you make that an explicit part of the state machine's API. Therefore, the state comes in. When you need to reconcile that application-level protocol, it's not data reconciliation protocol. It's fundamentally business logic that needs to reconcile, you end up actually opening APIs to actually support, say, I need to maybe pass in a snapshot of the old state, or maybe please replay me the stuff that came before. You've already got that, because we're replaying. The world is actually this way, and that state comes into a state machine through the external API of the state machine. The state machine itself is a source of truth for all things that it is keeping track of, and taking snapshots of. Obviously, systems don't exist in a vacuum, they only have any meaning with regards to their interaction with external systems. If you've got some external state that is updating, you will go ahead and pass that into your state machine via API calls. You got to expect that your state machine is fast enough to basically handle all of those changes in external state.

Montgomery: To go back to the idea of an account address, maybe your service does accept changing an address. Could it in a way, execute that by sending that off to something else, and then it comes back in through the log of events that says, this has been updated? In other words, it doesn't really do it itself, it passes it off to something else, and then it comes back in through the input sequence. Would that work?

Yu: Yes, this works for everything that isn't secret. I think static data can pass through a system. The throughput of data coming in that's not really being read that just pass back through is some function of your network card and your disk storage or whatnot. Secrets, I think, don't really fit. We try not to put secrets into our Raft monolith. What that means is, ok, when a secret changes, the business domain impact of that change should be expressed in some feature. Like if a password changed and has been invalidated, if we've invalidated some API key, what has been passed in to the Raft monolith is not the new password or anything, but rather, this thing is now invalidated. If your business logic says validate for a valid API key or something, it should then reject requests that are keyed off that API key. I think this model computation isn't well suited for being the source of truth on just any secrets or things that you don't want to basically be spreading everywhere.

Montgomery: Is it safe to say then that those things are really our external services, and whether they influence the logic or not, they influence it in such a way that you can send something into the service to change their behavior, like you were saying, invalidating a key. You're just invalidating that to give its current impact into the state logic. It means what you need to do then is to make sure that you don't have a lot of leakage of those kinds of things.

Yu: Absolutely.

Montgomery: Is there other places that you have thought about, since you made the presentation that you could possibly be pushing your logic that is not obvious, since you found this pushing this logic into other places, are you looking at certain problems a little differently now?

Yu: I think in general you want to separate out functional logic that is mission critical. In general, I think the reason we've spent a lot of time on trying to make a big portion of our logic mission critical is that running exchanges, almost all of it is. This is quite a big hammer, and you don't want everything to look like a nail here. When going back to any system, like say, if you're running something that keeps track of customer funds, or is keeping track of an important case where you have highly concurrent access, that determinism is incredibly useful, like scheduling, anything like that. We try to say we should localize this particular high throughput, low latency approach to only things that you intend to have very good test coverage for, and would really benefit from an explicit ordering of all events that come in. When you have massively parallel systems that are highly independent from each other, you don't really get as much benefit from having one sequence deterministic stream of requests.

See more presentations with transcripts