SBTB FinagleCon 2015: Steve Cosenza and Neuman Vong, Streaming HTTP with Finatra and AsyncStream
hi everyone my name is Steve Cosenza I'm not the on the furniture team at twitter and i'm here with newman who is on the CSL team and today we're going to be talking about streaming HTTP and how fenêtre has used a sink stream to do it quick little summary of what is fenêtre furniture is a framework for easily building ap is on top of Twitter Scala stack right now we support building HTTP services but we have some internal experimental support for thrift servers and we plan to be releasing that at some point some highlights fenêtre has widespread use as Twitter's production HTTP framework if you've used for natural 12 is about 50 times faster in many benchmarks we have powerful feature and integration testing support optional support for jsr 330 using google juice we've got improved integration with jackson over the normal jackson scala module so we do things such as supporting required fields default values and validations on fields you're also able to use that same kind of JSON support for declarative request parsing so even if it's not chase on it's a get request you still can do the type conversions and validations and we have some nice integration with lock back if you're using that as you're logging provider which lets you do contextual logging across futures okay so here's what a request would look like infinite aura to parse JSON you use a case class to mirror the JSON you could see some of these annotations provide validation hints such as a size we have a default value for sensitive finish or controllers you simply extend controller and then you have methods for all your HTTP GET post post elite etc in this case we have posting a tweet and getting a tweet and here you can see the batch versions of those endpoints on the top would let you post multiple tweets in one post and on the bottom it would let you do multiple lookups of tweets a single post and notice how the incoming argument is tweets which is a sequence opposed to tweet and on the bottom a sequence of tweet IDs server you add filters add controllers and the bottom you have a warm-up method so that you can get all of your clients warmed up get the server cheated before you start accepting and coming traffic and then here's an example of the feature testing support where we'll start up your server locally and an ephemeral port and make it easy to send in requests in the cert responses so I'm the sample that I showed you earlier this would be what would happen if you posted a JSON body of a tweet that wasn't valid and you could see based on the validation fields are based on the validation annotations you'd get the errors below so notice that it doesn't fail it the first one it kind of collects all the errors and then give them to you okay so what are the issues with the controller just showed you mainly that we use the sequence for our tweets and tweet IDs that's going to cause all of the data to be loaded into memory at once and we're going to be increasing the latency of our response because we can't return anything until we've read all the tweets saved all the tweets and then returned so streaming to the rescue streaming essentially in this case is going to be the incremental reading and processing of the request and likewise the incremental writing of their spots now traditionally if you're using blocking api's this is pretty easy you use an input stream you're in Java to read your request use an output stream to generate your response and then you probably have something like an iterator that you use throughout your program however in the non-blocking world we need new primitives to be able to do this type of string parsing and for that I will cool hi everyone so yeah as Steve are saying what's really important for servers I guess is to not run out of memory and one of the ways to do that is to sort of like bound your inputs and streaming lets you kind of do that by turning something that's really big into like small pieces so for that we have this primitive called acing stream and it lives in not infinite row or finagle it's in Twitter YouTube which is a wonderful reusable code from Twitter and yeah it has many fans and you know so an async stream is really it's practically a lazy list of futures it's slightly different and I'll tell you how in just a moment but let's just go over several important properties one is that it's immutable so it can be shared by multiple threads when the stream is modified it doesn't get updated it creates like a new version of itself which points to like the old version so there's always this previous version that sort of just stays around and gets collected or whatever so that's called persistence it's also lazy so detail of the stream if you don't if you don't need it then it never gets evaluated so that's really important when processing infinite input so like let's say like a stream of events or something so the basic shape I just took that from wikipedias linked list page but it's sort of applies here you have these cells that have two parts to them the first part is the value and the second part is a pointer to the rest of this stream the only difference between let's say like a normal x kala stream and an async stream is that the tail the pointer it required it's behind a future so it requires like a rendezvous between whoever is supplying the future cool so some basic instructions to introduce you to the async stream it's it should not look alien at all there's there's an empty stream which is given by just empty and you can make streams of strings streams of ants and you can convert other things into a stream like sequences and options and also futures so this is the special one so how do we do this how do we how do we take a future and make it into an async stream I'll get into that just in a moment so because of all these things you can use them in a for comprehension and it behaves just like a list like a list a flat map so just to go through this I we have we had one two three four and it's as if these were like seeks you can you will get like a four and then two and three and then one and if we sum all of them will get like you know this seven and eight if we convert it to a sequence I hope that's so follow ball anyway so let's have a look at how its implemented so as I said before it it's a chain of these cells where one part is the value and the other part is the rest of the stream behind a future so how do we do that well we take this thing called node so a type node that has two constructors one is the mt node which is like that last box there so that's what it represents and then a cons which is the value and then the rest of the string in a tail so you can see the type of tail is a sink stream and how we define a Singh stream is it's a rapper of the future of a node so the node type is actually hidden and to use async stream that's like the public interface you don't have to think about note at all but that's how its implemented on the that's like the internals of it so to sort of illustrate all of this I would ask you to think of a queue API so we have something like this in finagle called a sink you in fact and but you know this can work for any kind of cue that you have some some sort of receive or pole and it returns to your future of some thing so you can imagine that an async stream is just a bunch of these operations like sequenced so you can turn any sort of like you know reading from a socket or I don't know reading from a cue like this into a stream so like right at the bottom that that's how you convert it you take receive which gives you a future of string but from future gives you an async stream of string and then you concatenate it with you know bye bye Bri cursing unreceived but that will only get called if it's demanded and that's important so what yet is this repeated calls to receive and with that you can imagine creating like an echo server of sorts we're like the basic component of it is that you take this one stream of messages that you've received and then you just write them out again to some send API and you can do that just with it for each f which is a a for each that's specialized to a future operation so what happens here is when the when the pole the receive completes we send it into this we write it into the send operation and when that completes we call pole again that's the mechanism for this you can imagine that to transform the messages you'll just map it so here is that same echo server but with all the messages upper caste so it's I guess one of the benefits to this as well is just to be able to program to sort of like structure your program in these modules and then you can like map the pipeline through them ending in some effect which send in this case so i want to say lastly that a sink stream is pretty much just like a list kind of just a little bit of its api you have all of the stuff like for each take while drop while concatenation and you have all of the Monad interfaces like flat map and then you have the functor interface map filter so you can use this like in for comprehensions and there's like tons more stuff and I think I'm going to pass this back to Steve and he'll tell you how he'll connect how this is used infinite row alright thanks Newman so we started integrating a cig stream into fenêtre in early June and we released response support in the milestone to release so that allows you to return an async stream from a controller and the async stream will be automatically rendered either as bites if you specialize a way to convert into bytes or it'll be seamlessly converted into a JSON list and today we merged experimental support for parsing requests with a sink stream so the way that looks is instead of a sequence of tweets you'll now see there is an async stream of tweets notice that now that we have tweets we can for each or for comprehend over the tweets so now for each tweet we can call the database save method that safe method returns the future we convert that future into an async stream we then get a saved tweet and we're able to then yield the ID that the database generated for that tweet so this is now a fully streaming non-blocking API where you can post an infinite list of tweets in as each one is saved will stream out the ideas likewise in the bottom you could post a list of ideas will look up hid one at a time if the ID is found it'll be returned if not it just won't admit anything for that running and the way that happens is database get returns a future option and when you call to async stream on that if it's a future none it's just an empty async stream if it's a future some it's an async stream with one element it here's what the output looks like if we were to post a list of tweets to that end point you'll notice we return the transfer encoding chunked header and then as we're writing we immediately started receiving the beginning JSON list we wrote about 5 tweets and then we read the ID of the first tweet and then at that point the operations are interleaved so the way this works under the hood is we start out reading from request reader so that's an asynchronous way of reading bites from the finagle request we turn that into an async stream of buffs so buff is our encapsulation of bytes you can see that the way we receive those might not be on proper boundaries and most likely is not on proper JSON boundaries so we parse that looking for valid JSON objects in the list so the top three buffs will end up turning into two buffs delimited on the elements in the JSON array and then for each one of these guys will object map it to the case class so we're streaming each tweet but for each tweet we're okay loading the entire tweet into memory an object mapping that so it's kind of this best of both worlds if you had a stream parse every individual field of the tweet that would be pretty onerous but this kind of gets you a good middle ground and in terms of future work we're hoping to have a similar api for multi-part file uploads so that the controller can ask for an async stream of buffs you could then in this example maybe sanitize the bytes in a certain way before let's say you stream them to a file or stream them into your database and that is all any questions she's sitting there json request that has multiple JSON objects and you're trying to match that is the case of us are you are you driving the person based on the case class that you're parsing for or you just mash in any blind object that you get and failed to match that how does that work we first take each individual element out of the JSON array coming in that element is then compared against the case class so that could be an arbitrarily complex you know case glass with inner case classes but we're just kind of doing a trivial chase on list farce first and then doing the parse and one of the things we're planning on adding is an ability to ask for an async stream of tries let's say tweets so that if one of them fails the holding this zone you're parsing are you operating a buffer and so can you visit disability for example for each buff that's read off of the requests that essentially Nettie is giving to us we'll find however many full JSON objects we can so if in that buff there were five tweets all five of those tweets would kind of be processed at that time and that and that kind of iteration but if a full object isn't parsed and we have to wait until the next request to get a little bit more that's what saved and in-memory just so the next time it could kind of fill it in are you set the shocking libraries the chunking boundaries so I think what's probably happening was there a right example because you can't write to any extreme times I'm assuming that you just produce like one element each time and then whatever that gets person so I guess is some sort of buff and then the buff gets written to the writer which is kind of an abstraction of the queue that wraps this solve it like so it would just be that buff and then that would get written to Nettie as a chunk like an HTTP check and then yeah and then thanks