Disseminate: The Computer Science Research Podcast - Cuong Nguyen | Detock: High Performance Multi-region Transactions at Scale | #39

Episode Date: August 28, 2023

Summary: In this episode Cuong Nguyen tells us about Detock, a geographically replicated database system. Tune in to learn about its specialised concurrency control and deadlock resolution protocols that enable processing strictly-serializable multi-region transactions with near-zero performance degradation at extremely high conflict and improves latency by up to a factor of 5.Links: SIGMOD PaperDetock Github RepoCuong's Homepage Hosted on Acast. See acast.com/privacy for more information.

Hello and welcome to Disseminate the Computer Science Research Podcast. I'm your host, Jack Wardby. A reminder that if you enjoy the show, please do consider supporting us through buying me a coffee. It really helps us to keep making the show. Today, I'm joined by Kong Nguyen, who will be telling us everything we need to know about DTOC, high-performance multi-region transactions at scale. Kong is a PhD student at the University of Maryland. Kong, welcome to the show. Thank you, Jack. I'm very happy to be here. Great stuff. So let's jump straight in with some questions then. So tell us a little bit more about yourself and how you became interested in database research. Yeah, sure. So I am currently a PhD student from the University

of Maryland. And during my undergrad, I had the opportunity to work on a research project on time series data mining. But at the time, I wasn't sure whether I would pursue a PhD or not. So after my graduation, I started to work in the industry for a little over a year as a software engineer. And this is when I started to develop my interest in distributed systems and database systems. I want to learn more about them. And that was when I started to apply for a PhD and eventually ended up with my current advisor at Geophysics Maryland, Daniel Abadi. And we do research in transaction processing. So great. So let's talk. Let's have some background for the listeners.

So can you maybe tell them about transaction processing and geo-distributed databases which which are what detox is and so yeah why do we need them and what are the challenges that come with these types of systems yeah sure so transaction is a very important concept in database systems so it's the concept where you have either of the operations in the transaction to be executed successfully, or none of them would be successful. And geo-distributed database is when the database is spread across different geographical regions around the world. For example, you can have a database node running in a data center in the east coast of the US, another node running in the data center in Europe, and another one running in Asia. But eventually, you want to abstract away all this complexity and

present the database to the applications if the database is running on a single node. The reason why we need to geo-distribute the database is, in today's world, many consumer-facing applications have users located all around the world. And it has been common for these applications to require high availability and serve low latency requests to the users. These requirements are very hard to achieve if you only run your database in a single location. But additionally, this kind of setup also makes the database more resilient to regional failures because you have data in other regions for backup. The biggest challenge to this kind of setup is that you can have very high network latency on the order of a hundred milliseconds due to the fact that the database nodes could be very far away from each other.

And this network delay will not only impact the latency perceived by the end users, but it can also affect the throughput of the system. For example, if you use locking for concurrency control, the network delay may cause the transactions to hold the loss for a lot longer, preventing all the conflicting transactions to make progress. And therefore, it would reduce the number of transactions that you can process at a time. Awesome, cool.

So yeah, we want these for high availability and fault tolerance, but then obviously that comes with the trade-offs right there that we need. That comes with high network latency. And then like you said, it can degrade throughput if we don't get our scheme quite right. Cool. So with all these challenges in mind then, give us the elevator pitch for Detox. Yeah. So Detox is a geo-distributed database system that uses something called deterministic concurrency control to move this high network delay out of the contention windows of the transactions. And therefore, it can achieve high throughput for strictly serializable transactions. But additionally, it also partitioned the data across different regions. And it takes advantage of the localities of the workload

so that users can just access data that is closest to them. And we also use a board-free deadlock resolution protocol to reduce the latency of the transactions. Awesome. Cool. So you hit on a term there, deterministic databases. So what do you mean by this and how does that differ from like normal like normal non-deterministic databases i guess right so yeah okay sure um so they talk a part of a lot of research that involves this idea of using determinisms to speed up transaction processing and the high level idea is that um so you would try to perform the necessary coordination among the transactions as much as possible before executing the transactions. And this way, you don't have

to do these expensive coordinations during the execution of the transactions, in which a long delay might increase the contention window of the transaction. So for example, let's say in a non-deterministic database system, you might use two-phase locking for contingency control, and you also want to replicate your data, right? But during the commit of the transaction, you have to invoke the replication protocol to persist the commit record. But this replication process will increase the lock holding time of your transaction and thus prevent other conflicting transactions to progress. Now, on the other hand, in deterministic database systems like Calvin,

which was published in 2012, you have an ordering layer at the very top and every transaction will be ordered in this layer. And then every node in the system will then execute these transactions in a deterministic manner following this predetermined order. So by doing this, you have minimal to no further coordination during their executions. But note that to do this pre-coordination, we need to make an assumption that the transactions will be sent to the system in one shot. So unlike in conventional database system where you may make many back and forth trip to the base system, like you start the transaction with a begin statement and then send a couple

of SQL statements and then finally send a commit to finish your transaction. In a one shot transaction, you package all of the statement into a single request a single request and send this request to the system awesome yeah i've often wondered whilst i've um obviously been working this area how big of an assumption is that that transactions and are one shot and you don't have this interactive pattern of begin sending statements and then issue a commit or abort or whatever like like yeah i guess how big of an assumption is like in reality is does that reflect the way most applications are actually developed or not yeah uh so to to my knowledge uh first of all in uh academic the first i think one one of the first

uh papers that proposed this idea of using one shot for ol OLTP workload is the paper on HStore. And they argue that usually in OLTP workload, you will have short transactions and you usually know ahead of time what the code for the application would want to send to the database. So you can just package all this into a one-shot request. And in reality, I know of some systems

that use this one-shot models like TAO from Meta. It's one example for a system that use one-shot transaction. Nice, cool. So let's dig into the details of DTOC a little bit more then. So maybe let's kick things off with, give us a high-level overview of the system architecture, then we can go into a few of the important features of the system. Sure. So DTOC is deployed across multiple geographic

regions. And the way you can think of these regions is that the servers within one region are connected with fast network links, but the connection across two regions can be a lot slower. Now, for every data item in the database, we assign exactly one region, which we call the home regions of the data item. And by assigning the home regions for every data item, you effectively split the database into separate partitions across different regions. So there are two types of transactions in DTOC, single home transactions and multi-home transactions. A transaction is single-home if all of the data items that are referred to in the transactions have the same

home region. And a transaction is multi-home when its data items are from more than one home region. Within each server, we structure the DTOC process as a stack of modules. So first, we have a forwarder module which ensures forwarding the transactions among regions. In each region, we maintain a local input log for the transactions. So we have a sequencer module that uses a consensus protocol like multipaxels to append the transactions to this log. Next, we have the log manager module that manages the local log, as well as the logs that are asynchronously replicated from all other regions. And then we have a scheduler module, which will create an execution module for the transactions in the log.

And finally, the transaction will be disp closely with the idea of deterministic database system where you have an ordering layer on the top and then an execution layer after that. Awesome, cool. I just wanted to touch on the replication. So you've seen that data items have a home region, every data item has one distinct home region. Are data replicated across regions still? that you assign i don't know

like you see you've got three regions and some data right and x and it's home is it is is that the a region is it replicated across b and c as well or is it all within one region yeah so there are two types of replication happening here so first you have synchronous replication, which is preferably done within the same regions or within neighboring regions. And this replication is used for fault tolerance in the case if a node in a region failed or the whole region failed. We also have another replication mechanism that is done asynchronously. So each region will asynchronously replicate their log to other regions. However, this asynchronous replication mechanism is optional, but we highly encourage the usage of this mechanism because it will speed up the processing of multi-home transactions, and it will also provide the possibility of snapshot grid.

So yeah, so there are two types of publications in the system. Awesome, cool. Yeah, another thing that sort of jumped out to me when you were describing the architecture there, Kong, was how does a user decide where to put certain data? How do I know to make my data item foo in this region like oh is that totally down to the the user to express locality of where it wants that data item to be yeah so we don't focus a lot on like the uh the decision on where to look to

to place the data it's it's mostly up to the application to decide on that we focus more on the transaction processing aspect of the system cool yeah well that's a good segue into the the concurrency control protocol and this abort free deterministic deadlock resolution protocol you mentioned earlier on so can can you maybe illustrate how these two protocols work and maybe do this by, I don't know, like you walk through a lifecycle of a transaction over a single home or multi-home. The floor is yours, take it away.

Yeah, definitely. Okay, let me walk you through a single home, how a single home transaction is processed. Let's assume that we have two regions A and B. We start from a client sending the transaction, let's call it T, to a server in region A. The first thing that this server does is that it will extract the read and write set from T. And this can be done by static analysis of the transaction code. Or if this is not possible, then we can send a read-only reconnaissance query that will perform all of the next sets of read-reads to discover the transaction read and write set.

In this case, the read and write set would be speculative and T might be aborted later if the read-write set changed between the time that we do the reconnaissance read and when we start executing T. And with this read-write set, the servers can now determine whether T is a single home transaction or a multi-home transaction by looking up an index. And let's say in this example, we assume that T is a single home. So every key in T belongs to one region, let's say region B, right? Now T will be forwarded to its home region B. And then here it will be appended to the local log of that region. And at this point, the scheduler at region B will pick up T from the log and add T to the dependency graph. Let's talk a bit about this dependency graph. Remember that we have the read and write set of every transaction at this point. So we can construct a dependency graph where each vertex corresponds to a transaction, and an edge is drawn between two vertices.

If the two transactions corresponding to these two vertices both access the same key, and one of them right to the key. If we assume that every transaction in our system is single-home, it's easy to see that this dependency graph is a directed acyclic graph or a DAG. And you can find a topological order on this graph and execute the transaction following this order. Okay, so next I will talk about multi-home transaction. In our example, region A will asynchronously replicate its local log to B, and B will also asynchronously replicate

its local log to A, as I mentioned previously. And so eventually, every region will have the full set of logs from all other regions. And let's go back to our example. Let's say T is a multi-home transaction. And once this is determined that it is a multi-home transaction. And once this is determined that it is a multi-home transaction, it will be broken down into separate single-home transactions. And these separate transactions will be sent to their respective regions.

So in our case, we only have two regions, A and B. So T will be broken down into two pieces, and these two pieces will be sent to region A and region B. And these two single home pieces will be again added to the local log at these two regions. These two regions then exchange the local log asynchronously to each other. So each region will have the full copy of the local logs and it will have all the information that it needs for T. But now when you build the dependency graph out of this set of blocks, you might see cycles. Because you can have, let's say, chocolate T1 and T2. T1 might be ordered before T2 in A, but T2 might be ordered before T1 in B.

So when you combine these two logs together, you have a cycle in the graph between T1 and T2. And this is where we need our deadlock resolution protocol. The way that this protocol works is that it will determine a stable subgraph in this dependency graph. And this subgraph has the properties that it will never change as new transactions are sent to the system, because we don't want a raise condition where region A see a graph that is different from region B, and then the two regions will resolve that log in different way and they will arrive at different state of system, and we don't want that to happen. So we have an algorithm to determine a subgraph that would be stable, and it will be consistent

between the two regions so once we know that we determine this subgraph we will find all of the strongly connected components in this graph so a strongly connected component is a subgraph where every vertex in this subgraph will have a path to all other vertex in the same subgraph. And this strongly connected component is like a generalization of the cycle to determine deadlock in this graph. So for each of this component,

we will rearrange the edges such that there will be no cycle within this component. And the simplest way to do this is just order the transactions by their IDs, and then you can connect the edge between them in a linear fashion so that they no longer have any cycle. This has to be done deterministically so that two regions agree on the order of the transaction within a component. And once

you found all of the strongly connected components in the graph, you can find something called a condensation of the graph, which is you can think of each strongly connected components as a super vertex. And you construct them as

two vertex, And you construct them as to a vertex, but you retain the edges between them, the original edge between them. And what you end up with is another directed acyclic graph. And now on this graph, you will be able to find the topological order. So now you have two order. There are this topological order and the order within each

component. And by combining these two orders, you can find the total ordering between the transactions. And this order is deterministic across regions, and you can execute the transaction following this order to achieve determinism. Awesome. So I'm just going to repeat this back to you to make sure I've kind of grasped what's going on. So we kind of gather all of the logs from everybody else, basically, and we build up this sort of this graph. But the problem is the graph might have cycles in it because different people may have different transactions ordered in different ways. So the first thing we do is we'd find a stable subgraph across this,

which is like everyone's going to agree that this thing is stable, basically. This is fine. But there's still going to be some other bits that maybe aren't. So they'll be deterministically ordered, like, essentially by, like, their IDs. So you kind of have, like, everyone agrees. And you end up with, like, these connected components, basically. And then you can order the connected components.

So you end up with this sort of overall total order, essentially, of transactions. Is that kind of roughly how it works? Yes, yes. You get it right yeah okay cool awesome nice so so once we kind of we have we have this sort of this this nice bot free um protocol and everything there's this other sort of concept you have in in the system where you call them home movement transactions so kind of what are these and how do these differ

from like your single home and multi-home? What's the difference between these two? Yeah, so remember that we partition the data across different regions across the world. So the reason why we do this is to take advantage of the localities in access patterns. So in the real world, data access pattern can change. For example, in a social network application, you can have a user moving from one continent to the other. But you want to retain the benefit of data locality. So you want a mechanism to move the data from one region to another.

And these whole movement transactions is one mechanism to do that without having any downtime. And the idea is pretty simple. So you can construct transactions that have a single operation that will change the metadata of the data item from, let's say, region A to region B. However, the lock managers have to treat these transactions a bit differently from a normal transaction.

Because you want to prevent the race condition where some transactions might see data that belongs to the old regions, but some others see the data as belonging to the new regions. And this could happen during that data movement. And to do this, when we lock a data item, we do not only specify the ID of the data item, but we also specify the regions of the data item. And depending on where a transaction is located in the locks in relation to this home movement transaction, we can determine whether or not to abort this transaction for assessing the wrong region. Awesome. I see it kind of, it sounds like as the user moves, the ownership of the data can move around the system as well. That's really nice. Really nice feature. Cool.

So let's talk about some numbers then. So how fast is Detox? So tell us about your experiments and the setup, what you compared it against, and yeah, what you found. Yeah, so we ran the YCSB workload in our experiment that is modified for transactions. And we varied the amount of multi-home transactions and the concession level. We compared Detox with two other deterministic database systems called SLOC and Calvin. And in terms of throughput, Detox performs either similar or slightly higher than these systems.

But the metrics that we care more about is the latency. And we saw that most of the multi-home transactions in Detox is up to five times faster than these other systems. And it is easy to see why this is the case. So both Zlok and Calvin need to specify one region as the default region, and they have to route some or all of the transactions to this default region. And you can imagine that clients that are situated far away from these default regions have to make this trip to this default region for global ordering before it can actually access the data from the regions that it wants to access. And in DTOC, thanks to this deterministic deadlock resolution protocol,

we can completely eliminate this global ordering process. So the transactions can be sent directly to the regions that they need to access data from. Awesome. Yeah. Did you measure the impacts of home movement transactions?

And was there any sort of experiments on those as well? Like how, if you have a workload running whilst you're at, whilst the class, did you, I'm going to ask you, did you kind of simulate or kind of model people moving around the globe and seeing how the impact on performance of those?

Yeah, that would be a very interesting result. But unfortunately, we didn't do that experiment. We only described what the whole boom and transactions are in the paper, but we didn't do any experiment on that. But that would be a nice experiment to run in the future. Yeah, it'd be cool. It'd be interesting to see if that had any impact at all. I guess it'd be nice.

Cool, yeah. So let's talk about some limitations of Detox then because it sounds amazing, but these things always come with trade-offs and caveats, right? So maybe you could tell us a little bit more about the scenarios maybe in which that DTOC performance is suboptimal. And yeah, just the general limitations of this approach. Yeah. So as we touched on previously, so one limitation of DTOC is that it does not support interactive transactions. And because

we need one source transaction for deterministic concurrency control to work. This could be quite inconvenient because you lose the ability to interleave database code with the application code. But then again, we think that this is still a reasonable assumption to make because you can see all the systems like VogueDB or TAO that make the same assumption. Another limitation of DTOC is that it's based on Calvin. And in Calvin, it uses a single thread to schedule the transactions. And this thread could be a bottleneck and make it harder to utilize the cores in a multi-core system. So it would be interesting to think about how to apply this same deadlock resolution protocol,

but with other deterministic concurrency control that uses LILI, like MVCC or OCC, so that we can get rid of this limitation. Nice. Yeah, kind of on that then. So how does Det compare to like other sort of state-of-the-art geodistribute databases be that in either academic the academic world or in the commercial world and just i mean when we've obviously been talking about this calvin and slog and then there's and there's other sort of systems i can't remember there's one recently out of um

oh what was the thing called uh it was like a h star c uh sequel i can't remember what the pit what the system's called um uh oh what's it called i'll try and find it while you're answering the question but yes i know there's other systems out there like aria as well which is the deterministic database which i know addresses some of the limitations and this when you were talking about about the data moving around the system, this kind of ownership thing, it reminded me of a protocol called Zeus that I read a while back. So yeah, how does DTOC compare to these other approaches?

Where do you position it in this sort of space? Yeah, so we focus on comparing with other systems that support true repetition database. So we didn't really look into comparing with ARIA because it's more focused on single data center deployment. But we did compare with CockroachDB, which is not a deterministic database system, but it supports geo-partitioning.

And we found that when we increased the contention level of the workload, CodeCrushDB would have an increased grace of transaction abort due to deadlock. And this is because CodeCrushDB uses a two-phase commit and some sort of locking mechanism for the write operation in the transactions. And it has to detect deadlocks and abort the transactions to break this deadlock. And since Detox does not abort any transaction due to deadlock, it can sustain a much higher level of throughput. Additionally, Detox supports strict serializable transactions,

but the highest level, isolation level of CodeCrushDB is serializable. But again, one caveat is that CodeCrushDB supports interactive transactions. It is still an amazing database system. But we believe that our assumption would not be too restrictive interactive transactions, it is still an amazing database system. But we believe that our assumption would not be too restrictive for OLTP application. And if the application had the characteristics that can be supported by Detox, then I think it's still a good choice for a database system. Cool, yeah.

I mean, the ability to to especially in a geo distributed deployment right of having no bots is like a massive win because the wasted work that goes for having a bot so they're like yeah it yes and then i can see why it has it has a massive win there the system i was thinking of that was recent it was called lotus i don't know if you've seen that paper it was a vldb oh yeah it's just a very recent paper yeah yeah i have that paper but uh i think lotus again uh only uh consider the case of a single data center single region deployment right okay yeah yeah so they haven't gone they haven't gone geo distributed cool yeah so awesome so what's next on your research agenda for detox then so what's the

next steps right now i'm working on a different project unrelated to Detox. So we don't have a specific plan for extending the research on Detox. But some of the ideas that I found interesting is, as I said before, to see if we can combine different deterministic concurrency control protocol

with this deadlock resolution protocol so that we can avoid any current shortcomings of the talk. Nice yeah because I guess the University of Maryland is the home of deterministic databases right like it was your advice was original idea so I guess there'll be there'll be more research to keep coming out of it on this sort of topic so now I look forward to reading that great stuff. So yeah, so kind of, again, more generally sort of speaking here, as a software developer, data engineer, DBA, how do you think I can leverage the findings of your research and

kind of what impact do you think your work with DTOC can have? I guess one way to leverage this research is if you want to build a new database system and if your requirements match with the transaction model that Detox can work on, then certainly you have this new option for your implementation. Or even if you have specific problems and it happens to match with the problem that Detox is trying to solve, that you can use the techniques presented in the paper to solve your problem. And we also open-source our code. So you are welcome to reuse the code in your work. Awesome. Nice. We'll drop a link in the show notes to that

so the interested listeners can go and find that for sure. Cool. So, yeah, I mean, obviously, whilst you've been working on Detox, has there been anything that's jumped out to you and been like, oh, this is the most interesting thing I've learned while working on this project? I think, yeah, most of the things that I learned

was related to the software engineering aspect of when building a system. And I think I got two big lessons that I learned from this project. And one is, you should instrument your code with observability tools early on. It helps a lot in finding performance bugs and just generally understand your system better. I didn't do this in the early stage of my project. So I have to spend countless time trying to figure out why something was too slow or why something is not behaving the way that we want it to be.

And the second lesson is that cross-region requests in AWS are very expensive. So in my first few experiments, I was very generous with sending cross-region messages across regions. And I ended up with a quite hefty view and my advisor wasn't very happy with that uh so i have to be more thrifty with that and um that bring down the costs on the first point you made there about the uh the observability i mean i've i've fallen into that same trap before as well and it's kind of you

think oh what need it's fine just hacking away developing and then all of a sudden you're like oh shit i need to figure out what's going wrong here and then you're observing it and it's yeah yeah yeah it's cool so yeah i mean kind of maybe loosely related to that then um was there like a few things you tried along the way with the development of detalk or maybe the idea or the actual implementation and things that failed and yeah kind of maybe some war stories about this this process you went through with detalk yeah so i think that the only thing that we tried uh a couple of time is the implementation of the deadlock resolution protocol itself. In our very first revision, we used a single graph that is shared between different threads. And this is a very bad idea because every time you want to modify this graph,

because in our protocols, we need to rearrange the edges and all that. You have to acquire a lock on this graph and that's just bring down the performance of the whole system. And in the end, we end up with an implementation where each thread, actually there are only two threads here, that will build their own graph. And so they don't have any contention over the entire graph. And they will share the information about what to change on the graph

by using a message queue. Okay, nice call. Yeah, I see. So yeah, I guess you mentioned it a second ago about that you're no longer working on Detox. You're on a different project. Maybe you can tell the listener a second ago about that you're no longer working on DTOC. You're on a different project. Maybe you can tell a listener a little bit about that project and maybe some about your other research.

So as part of the research on deterministic database system, we are working on a project where we investigate the practicalities of these deterministic database system in real application. And we do this by looking at the code of a bunch of database systems in real applications. And we do this by looking at the code of a bunch of open-source projects on GitHub and just study their transaction code

and write down what matches with the assumption that we need to make in deterministic database systems and whatnot. The second project I'm working on is another geo-distributed database project, but it's not a database system.

But we look into supporting geo-distributed transactions on existing single-node database system. Ah, nice. So is that sort of, what are these systems you look at? Are they like cloud-based systems you're looking at?

And like, I don't know, taking something like Aurora or one of these systems? Yeah, yeah, yeah. That's also, it's actually Aurora. It's actually the Aurora-like architecture. Ah, okay, cool. Yeah, sorry, continue. Yeah, because Aurora is right now fully deployed

in a single region. So we want to take it a step further and see if we can support geo-distributed transaction for this kind of system. Nice, nice. And obviously Aurora as well is like single writer, right? You only have one node. It's sort of like, can you make this multi-writer?

There's a lot of... I read an interesting paper at CIDR actually. I actually had it to be on the show who talks about that stuff. So it's interesting to see how that project comes along. Nice, for sure. Cool. Yeah, great. Yeah, so I've got just two more questions for you now.

And the penultimate one is, tell me a little bit more about your creative process. How do you go about generating these ideas, these really cool projects you work on, and then actually filtering out the ones to pursue. So how do you go about doing that? Yeah, so as a student, I'm still figuring it out myself. And obviously, my advisor is a great source of ideas

and things that I might be able to work on because he has a broader view of what is going on in the industry or in this area. But one approach that I am experimenting with is to look into the gap between research and what's happening on the real world and ask what are the obstacles that prevent this research to be applied to the real world and about selecting what projects to work on i try to evaluate the implication of solving each of these problems and then choose the one that have that would make the most impact nice that sounds really nice process i can see how that how that works it's really cool um awesome cool yes it's time for the last question now, the last word.

So what's the one takeaway you want the listener to get from this podcast today? A key takeaway from this work is deterministic concurrency control provides a framework for minimizing the overhead of coordination within a transaction. And this work on DTOC once again highlights the viability of this framework in improving the performance of geo-distributed transactions. And additionally, I believe that there are still more opportunities to explore within this framework. And I hope to see future work that addresses any current limitations of the frameworks, as well as works

that would try to integrate the techniques from this framework in the real systems. Fantastic. Let's end it there then. Thanks so much, Kong, for coming on the show. I've really enjoyed this conversation. If the listeners are interested in knowing more about his work, we'll put links in the show notes. And again, a reminder, if you enjoyed the show, please consider supporting us through

Bamya Coffee. And we'll see you all next time for some more awesome computer science research