SBTB 2023: Shadaj Laddad, Hydro: A Stack for Faster Streaming Applications

SBTB 2023: Shadaj Laddad, Hydro: A Stack for Faster Streaming Applications

Recording: SBTB 2023: Shadaj Laddad, Hydro: A Stack for Faster Streaming Applications

[Music] um all right hi everyone uh I'm shaj I'm a thiry year PhD student at UC Berkeley where I work on programing languages for distributed systems and today I'm really excited to share with you some of the research we've been developing um on making it dramatically easier to build distributed streaming applications um so before I dive into this this is a research project um with lot of amazing people I'm going to be presenting some of the work of other folks in our group um including several of the PhD students postdocs engineers and faculty involved with this project um and so before we jump into what is hydro um let's start by discussing kind of what what's the problem we're trying to solve here today when you're building a cloud application you you really start by focusing on developer productivity right especially if you're a small startup you want to ship your features as fast as possible um and just get something something out there and then as you start to get more users you'll want to scale up your app so thankfully we have the cloud and it's pretty easy to to scale simple applications just by adding more machines but at some point you're going to run into issues you're going to discover bugs because you scaled out your application using replicas and your application didn't realize that there were going to be replicas involved and you might run into consistency issues um or bottlenecks um where now you have all of your distributed microservices that are locked on on one common service that they need to talk to before they make a decision and in all these situations you the really the only solution today is hire distributed systems PhD which is what all the large companies do um and this is great because now you can kind of nail down what are the protocols you need what are the consistency levels you need at different Services um and at the end of the day you do get a scalable app that efficiently uses Cloud resources so that's kind of how applications end up being developed today now although I would love to have a job as a distributed systems PhD this is not really a great way to develop apps um and it means that unless you have kind of the resources of a giganta company um to be able to make these protocol level optimizations you can't really get the peak uh performance that you should be able to get get out of your app and that's where Hydro comes in the goal of hydro is to automate this process um to have a computer make the decisions that traditionally distributed systems uh engineer would make um and automatically optimize your app to take advantage of the vast amount of cloud compute we have available the hydro stock contains several components um I won't be talking about all of them today um but a quick overview Hydro deoy is the lowest level of first stack it's a uh tool for provisioning cloud resources and deploying your app and orchestrating these Services one of the key focuses of hydro deoy as part of the Sky Computing Lab at Berkeley is you can deploy your apps to span across multiple clouds um and this is kind of one of the key goals of our research project hydroflow which is what I'm going to be talking about today is our Ultra fast rust kernel for data flow um it's a low-level IR for describing streaming data flow um with an optimizing compiler that can make it go really really fast and hydrologic which I'll T touch on later on in this talk um is a step above that because Hydro flow is meant to go fast but hydr logic is meant to make it easy to write your distributed systems applications um and the idea is with hydraflow you just write your application as if it's going to run on a single machine we make the decisions to automatically distribute it as I mentioned before Hydra is a research project but it's all open source um and you can actually play around with Hydra today um at Hydra run so in today's talk I'm going to be kind of uh bouncing back and forth between the research side and the production open- source side of what we've been building um and we'll kind of touch on three key topics first we'll take a look at stage programming um which is a new way to develop uh program generators and we're going to look at how we use that to enable efficient data flow and R second we'll look at this really exciting new optimization technique called EG graphs and how that makes it easy to write optimizers over streaming data flow and lastly we'll look at program synthesis today has been all about large language models and this talk um does not shy away from that we're going to be taking a look though at a different angle of how we can use these to generate provably correct uh crdts so start with the first so hydroflow as I mentioned is a low-level rust uh IR for writing streaming dataflow applications um it's meant to write the types of things you would write as normal streaming queries or or spark queries but it's all implemented in Rust now hydroflow takes a slightly different design than some of the high level apis like you might see in Spar or Flink actually each hydroflow program just runs on a single thread on a single machine but compositions of hydroflow programs form a distributed system this gives us a lot of flexibility because if you want to scale out on a single Machine by using multiple threads you just launch multiple hydroflow programs but if you want to create a distributed system you just run your hydroflow programs across multiple machines and the reason for this architectural decision um has been a lot of research within our group on shared nothing architectures so for example we have a previous paper in our group called Anna where we developed a key value store that scales dramatically better than any other key Value Store out there because none of the none of the threads handling requests actually have any shared memory or shared State everything is just gossip in the background um and the goal of hydro is to make designing applications like these super trivial so actually at the top you'll see the original C++ um in the original an paper and at the bottom is our reimplementation of this in Hydro where we had dramatically less code and it scills even better so when you're writing rust code traditionally your pipeline is relatively straightforward just like any compile language you write your rust code you run it through the compiler and it spits at optimized native code but this has one downside if I write my rust code in a specific structure even though the optimizer is going to make some decisions about how to allocate memory where to place data the optimize code is still generally going to follow the structure of the input code just because we want to preserve kind of the way the developer intended the application to be executed but in the case of streaming data flow we actually have a different programming model we're not running iterative for Loops we have a more flexible declarative language and so we would love to be able to do optimizations that rearrange your code in order to achieve uh higher throughput that's where stage programming comes in stage programming is a technique where you actually compile your code through multiple stages of compilation in hydraflow you start by writing this highle functional rust code that looks very similar to what you would write in something like spark but we don't actually run this code directly instead we compile it to this intermedi binary and that binary gets gets fed through our own optimizing hydroflow compiler which can make these decisions to rearrange your code as necessary and at the end of the day we feed that back out to the rust compiler to spit out optimized native code which means that you do get to preserve all of the me memory safety benefits of rust but we get to make more aggressive optimization decisions that gets you a lot better performance so to understand how hydrofill Works let's take a look at an example so um in this application we're going to write a simple chat app so let's start by looking at the entry point um when you write a hydraflow program you specify a couple of inputs and outputs um in this case we're just using standard standard Tokyo streams Tokyo is the the Asic networking uh Library uh for rust so in this case we're going to have an incoming stream of users who are Jo joining our chat Channel um this is just a bunch of user IDs um 32-bit ins and we're going to have messages that are being sent to this channel which are just strings and we're going to have an output Channel which receives topples of user IDs and strings and this corresponds to sending a notification to a user that a message has arrived in the channel so let's take a look at how we would imple and and one one thing is you'll notice that there's a lot of kind of wrapping around this I'm not just taking the user stream as directly a a Tokyo stream I have this runtime data parameter um and you'll notice there's this stage left uh annotation at the top and this is where the stage programming comes in so when you actually invoke this function for constructing this data flow it's not a regular rust function it instead it actually expands to a lot of optimized code so under the hood we're doing a of code Generation Um to emit this optimized Rewritten data flow implementation so with that let's take a look at the actual body the program once you get past kind of the the entry point and outputs of this program everything else just looks like standard rust code so we're going to initialize two asynchronous dreams um for the users and messages now what we want to do is every time a message comes in we want to send a notification to all the users who've joined the channel so we can do that by doing a cross product so for every pair we'll emit uh a toule of those and that's actually all we need to send out so in this case our our streaming data flow is very simple but we've used hydraflow to implement much more complex protocols like pus which I'll talk about later now there's one issue with this and the program we actually wrote will not do any of the behavior that we expected it to actually it'll probably not send any notifications to the user and to understand why we need to dive a little bit deeper into hydr flow's execution model which differs pretty dramatically from traditional streaming data flow now there are two major problems with writing distributive system reasoning about time and asynchrony and reasoning about space where your data lives and hydroflow is designed to make it very explicit what decisions you're making with regard to both of these aspects so questions about time will the network reorder my data am I sending over UDP or TCP is my data batched and do I need to take that into account how long do I keep around data can I flush it out is it persistent to disk decisions like these are super critical and often times they're made kind of automatically in the background in ways that the developer did not intend similarly for for space when do I move around data is it okay if my data is just sitting on one machine or do I need to shuffle it between other machines is it partitioned or replicated these dramatically affect how you actually write your application so hydroflow is designed to have a runtime model that makes decisions about these very clear hydroflow executes in a single event Loop where you run iterations that we call ticks on each tick you take in a batch of inputs from the network or any other IO sources process all of your computation until fixed point hyro does allow Cycles within your streaming data flow graphs and we emit a batch of outputs that's all now the thing that makes the distributed systems challenges much more clear here is hydroflow uh uh event Loops are stateless by default that means at the end of a tick any state that's been accumulated will be cleared out so now it might be might be a bit more clear why this application might not work so let's say we receive three events um on our incoming stream so we receive a user with ID zero who's joined and then on the second tick we receive a message hello and then on the third tick we a user with id1 joins now because hydraflow is stateless we're just going to process each of these ticks separately and so each of on each of these ticks when we do the cross product we're just going to get no values because we're missing data in one of the two halves and this actually exposes a implementation flaw we didn't specify which users we should be joining with which messages so one decision might be to say hey once a user a user joins the channel the user is going to be there uh for for the rest of this channel lifetime so we add a persist annotation so we're going to take in the Stream of users and persist that and all persist does from the the developer point of view is add a little bit of state right to there so now when on every tick the user stream will not only emit the new data that's come in from the network but also all the previous data received on previous tis and so now we get the behavior that when the Hello message is sent it's sent out to user zero and this is the semantics of a chat app like signal where if a message gets sent it's gets sent to all the current users um of that channel but if you want to implement the semantics of an application like slack this is not actually what we want right if a new user joins they should also be notified of the existing messages that happen now implementing this is also relatively simple we just persist both we persist the users and the messages but this actually also gives us some behavior that W quite not looking for right now that we've accumulated both what's going to happen is on every single tick we're going to compute the cross product of data that arrived in previous ticks and so we're going to get redundant notifications um in particular on the final tick although we're now sending out the notification to the new user we also send a duplicated notification to the previous user so we send the notification to user zero twice and that's not what we want so hydroflow has a third uh a second stateless uh State flow operator called Delta and Delta just says filter out all the messages that appeared on previous text and these are the only two stateful operators in all of hyrl everything else um just falls out of this tick model um and we're able to show through optimizations that although implementing things like persistent Delta naively would be super uh bad for performance we're able to optimize that cost away so as I mentioned if you have a a pipeline like taking a source aing source of messages persisting it transforming the data and then reducing it there are actually ways that we can optimize this away first of all if we're accumulating this data and then transforming it why not transform it first so we don't have to do that over and over again on every tick but at this point if we're accumulating some data and then performing a reduction over it why not just keep around the result of that reduce from the previous tick and start from there instead of replaying all the messages over and over again on every step and these types of optimizations are are what we perform within the hydroflow optimizer so that the developer gets a nice high level API but their code is executed with high performance so let's dive into how we write the optimizer for this part I won't be taking a look at this particular example but we'll we'll take a look at a at a similar example of optimizing joins um which show up a lot in databases and in hydroflow as well now what's really exciting about this is the technique we're using to implement this Optimizer if you've looked at a optimizing compiler before you've probably seen compiler passes like those that show up in llvm um and that's kind of a traditional greedy approach to optimization where you look at your code you analyze for some pattern if you detect that you know that there's a better way to run that code um but in the the case of applications like hydraflow there actually a wide space of possible optimizations and the optimal optimization actually depends on many factors beyond just the code itself it depends on what type of data you're receiving on your inputs how frequently you're receiving data on different inputs and so EG graphs have become a really popular technique recently in the past two or three years in the programming languages Community as a new architecture for writing optimizing compilers um and in fact the the uh person who popularized EGS with the rust Library egg is actually joining Brickley as a faculty uh starting in the spring so let's take a deep dive into how EGS work and how they allow it us to optimize programs so consider a program where we're doing a bunch of natural joints we're going to join the table a with the result of joining B with a now when you populate this into an eaph um you're going to kind of take the as of this um and and dump them as a bunch of nodes but you'll notice some weird things happening here first of all if you have the same expression twice we actually consolidate those into a single node so when you see these black uh boxes those are called e-classes and all the contents of an E-class represent different ways to write a program that return the same output this is what's going to allow us to optimize our program because we're going to get different results in the root node corresponding to different ways to to write our program um and so you'll see here that because we had a referred twice and we know that a is an expression that returns a single value we just have two pointers to that equivalence class um of of relations now the second step is we're going to write Define our optimization rules um these are really simple rewrite rules where we say okay if I have joins joints are associative so I know how to transform those joints are communative so I can just swap um any pair of of arguments to ad join and if I'm joining a a relation with itself I know that that's redundant so I can just output the relation itself and that's it we only need these three rules and the egph will take care of actually figuring out which which order to apply these rules to get an optimized result so let's start by applying uh these rules so let's take a look at associativity will notice that this rule matches our entire expression right um we have something joined with another join um and so we if we apply this re rule you'll expand the Egra um so now let's take a look at what what happened here so we added a new node to our eaph um in the root and that corresponds to another expression right so now we have two different Vari of her program which compute the same result now this graph is going to get very messy but if we take a look here we'll notice that now this new expression that we added says we could join a with b and then join the result of that with a right this is applying the associativity rule next let's apply the commutativity rule so I'll notice that I have two different e-classes and so far according to the eaph these compute potentially different result but one of them is joining B with a and the other one is joining a with b and I noticed that these are can be proven to be equivalent according to our commutativity role so I can just merge those two together um into a single e uh E-class now what's really cool here is when you have pointers to an E-class that contains several e noes you get this combinatorial explosion of possible Pro Rewritten versions of your program but in memory you have this compact representation because we're not actually expanding all of them greedily uh we're we're keeping this lazy information around to make an optimization decision later next next let's apply the associativity rule again so now I'll notice that I'm joining a with a join of A and B if I applied this rule then I should I can move the join around so now I get a join of a with a joined with B again applying the associativity rule now you'll probably guess what's the next rule I'm going to apply it's the Redundant joint rule because I'm joining a with a and I can I I just know that that's the same thing as doing a so I can merge those two e-classes together because they represent Expressions which compute this compute the same result now this is a really cool thing because now we have a cycle in our egph and those are allowed this is saying that I can keep on joining a with a as many times as I want and it'll always produce the same result so EG graphs can actually capture unbounded programs U through these rewrites and we actually take advantage of these in some of our recent research projects now we'll notice that we actually ended up with two other uh uh Expressions within our egph which again are the same thing so here we have one node which is joining a with b and another node joining a with B now this is an interesting case right I have two expressions within equivalence classes and they're the exact same thing and because I know that an equivalence class captures programs that compute the exact same output I actually know that all five of these expressions are Computing the exact same result so I can merge them all together into the top level um equivalence class so now I have a bunch of different ways to compute this join result and so I'll notice that one of the paths after I did all this transformation in my eaph actually just captures the join of a with b um and that's our optimized program so EGS are a really powerful tool I encourage you if you're curious about program optimization to try it out it's very easy to get started with it it's a rust library that you can play around with um but we're applying this uh pretty extensively uh throughout the hydr optimizer so I don't have too much time to get into the details of exactly how we apply these um and the different optimizations we Implement um but I'll point you to two papers that we have um the one on the left um shows how you can do incremental computation through optimizations um in this egap technique so things like streaming joins or incremental view maintenance we actually show can be derived from scratch by applying these simple rear rules and on the right is a really exciting paper it'll appear at Sigma 2024 it's not published yet but we actually show how you can take implementations of classic protocols like paxis two-phase commit things that are foundational to distributed systems and scale them to achieve things like 5x throughput in the case of paxis just by applying Transformations um of the style you saw here now one of the most interesting uh optimizations that we'd love to apply is around folds so folds which accumulate state by processing a stream of inputs um and applying some aggregation function are very uh fundamental to writing streaming data flow um but mutable state is really hard to optimize um even when it's wrapped in a nice functional wrapper this is still something that's changing and and optimizing where this data is placed or how it's accumulated um is really hard to Reas about in particular something we would love to do is scale this out so instead of accumulating some uh result on a single machine what if we could employ multiple machines to uh compute this aggregation together um this is really hard because if for example these operations I care about the order um then I might not be able to arbitrarily split this task up among multiple machines um but we actually have some exciting uh results in making these optimizations go through as well um so crdt which show up on Hacker News almost every week U on the front page um have been gting a lot of popularity for making it easy to write local first software so this means software where you can interact with things like text documents locally and they'll sync in the background with other machines we actually have a different take on crdts we think crdts are especially useful for backend classic distributed systems because they give you similar guarantees on being able to immediately perform actions on a single machine without having to coordinate with others so let's take a look a traditional application is the shopping cart um popularized by a bunch of classic Amazon papers and distributed systems um where you have a user that's adding and removing items and in a classic implementation where using a regular transactional database you might just have all the requests for a single user go to a single machine to process them in order something we would love to do is replicate this uh shopping cart so instead of having a single machine uh serving these requests you can have multiple machines this means not only serving read requests which is something a lot of applications do today with replicas but also serving right requests so that means I might have one request to add a Ferrari go to one shopping cart a request to remove it go to another replica of the same shopping cart and then a query go to a third and what's in so if you implement this in a traditional database like if you use something like cockroach DB which is a distributed database the way this is achieved is by using locking where if I have two different replicas that receive requests to to write to the same entity they'll make sure that there are no conflicts with other replicas but locking is slow and it eliminates the benefit of Distributing uh this replica because now one replica is processing data at a time and so we really want to avoid um any sort of locking and that's where crdts come in so crdts are Conflict Free replicated data types are a special type of data structure that allow you to process these mutations locally without having to coordinate with any other replicas but they store enough metadata so that eventually you can combine the results um of of right on multiple machines so in the case of our application if you have concurrent rights on different machines to insert your Ferrari and remove the Ferrari we actually store these in two separate sets within your data structure so that when you merge them together you know about the effects of both of these operations and can process those appropriately so crdts are have been used in a lot of applications so if you've ever used Apple notes you're using crdts under the hood um but they've also been used in a lot of high skill distributed applications for example League of Legends but crdts are really really hard to actually make work in practice um so all the crdts that are kind of commonly used in practice were originally developed at research Labs um in Portugal and France um and these are distributed systems phds and those are the type people we're trying to put out of jobs in the first place so that's not really a win um and even distributed systems experts like Martin kman make mistakes when writing crdts um so it's really hard to adopt crdts in your own application especially if you have some domain specific data structure that you need to be replicating and so to tackle that we developed uh a tool called Katara so qara is a tool for automatically synthesizing crdts rather than writing them from scratch qara leverages a line of research we've been developing at Brickley called verified lifting um and the idea is if you have some Legacy code written in C or C++ or Python and you want to translate in into a new domain so for example past research projects have included taking imperative Java code and automatically translating it into spark we're using this pipeline to automatically take tradition data structures that are implemented for single threaded usage and translate those into uh replicated data types so using Katara has three steps uh you import an existing sequential data type so this can be written in C or C++ so in this case we're just describing the semantics of acept um and one of the cool things is in our optimizing compiler we actually automatically uh grab these specs because we analyze your code for where you have a fold and the fold gives us all the semantics we need for your initial State and how we apply any streaming uh mutations the second step is you have to specify a conflict resolution policy so in the case of your uh classic data structure so for example if I have a set what happens if I receive an insertion and a deletion at the same time how should they be ordered um our solution is to have developer specify a very lightweight annotation which says if they arrive at the same time how should I order them um and again this can usually be derived just by looking at your existing data pipeline um because you might be ordering your events anyways um Upstream um so so in this case we're going to specify that if you receive an insertion and a removal concurrently you should process them as if the removal occurred later and based on just these two pieces of information you can run a python script um let your machine get really hot for 20 minutes and it'll spit out a fully verified crdt um and the cool thing about this is um we are using kind of techniques like llms under the hood to make this process go faster um but we guarantee that this crdt will behave identically to your original data structure with this conflict resolution policy um and we do that by using traditional formal method solvers under the hood to actually perform this verification so in our optimizing compiler this actually happens entirely automatically the developer doesn't even know that we applied this entire program synthesis technique because we can transparently apply it since uh since we guarantee correctness at the end of the day um so just some cool results we got out of this if you've seen crdts before read a couple blog posts about them you might have seen SE the classic Two Face set crdt so this is the one I showed earlier which says if you want to store a set of items and you want to support insertions and removals you store them in two separate sets one to track the insertions and one to track the removals um so that you you keep around the necessary metadata turns out if you actually throw this at our system and try to ask it to derive the crdt from scratch from first principles it doesn't actually generate this one it generates a more efficient one where we say actually you only need a single map from the the elements you have to booleans so instead of having kind of four different states for each element you only actually need three um to capture this this is a simple example um but it shows kind of how you're able to use uh these techniques to derive uh Nic rdts for domain specific applications from scratch now crdts are a really exciting line of research and something we've been digging to a lot more I'll point you to two more papers um the first is keep common crdt on um which is a paper we recently published at uh vldb crdts have gained a lot of popularity in the in spaces like collaborative editors where even if you have some latency or some Inc temporary inconsistency it's a human at the other other end and so they can just get kind of annoyed or talk to the the other person over Zoom being like why is our data jumbled together but if you're developing a distributed system and especially if you're developing services on top of these crdts those types of inconsistencies cannot be tolerated because your application might be sending money to someone or starting a shipment um and so in this paper we actually to take a look at what is necessary to kind of layer on top of crdts to give you the strong consistency guarantees that we traditionally receive from um transactional databases um and if you want kind of a broader overview of kind of where crdts fit into distributed systems um I'll point you to the new directions in Cloud programming uh paper which is this Vision paper that kind of lays out um the the goals of the hydro project overall and some of the key kind of technical uh pieces where we're building into that all right so that's that's a quick overview of the type of research we've been developing in hydraflow and hopefully that kind of inspired you to to take a look at at some of these lines poke around with eaps try out hydroflow on your own laptop um because you can install it today and follow our tutorials um but there's a lot of exciting work happening um and if you're in the space of distributed systems and looking to optimize uh your your services uh we'd love to chat because we're always looking to take our research and apply it and and tune it to use cases in industry um as I mentioned uh you can check out Hydro hydra.cc of playground where you can interact with our compiler which has been compiled to web assembly so it runs entirely in your browser um and check out also our our recent Publications but with that I'd love to answer any questions so that uh program synthesis and making the crdts from the uh the single use data structures it's really awesome are you planning on expanding that to any other languages in the near future with your research yeah so we're in particular looking at python um now one of the challenges is python involves lots of external libraries and stuff in classic code and so figuring out how to formally model those is quite a big challenge but for example we're looking at things like Ray actors so Ray actors if you've uh used the ray programming uh the ray library before those run on a single machine and they single threaded but what if you could automatically replicate those so you can have fa tolerance and and higher throughput so we're looking at using the same technique in those domains all right if there are no more questions uh I'll be at the table outside I don't know which table I'll be at but yeah table two um so happy to answer more questions there thank [Music] you