14:51 <jtfmumm> I recently uploaded a blog post about how we approached designing Wallaroo with performance in mind and some principles that could be useful when building your own performance-sensitive software systems. You can read it here: https://blog.wallaroolabs.com/2018/02/how-we-built-wallaroo-to-process-millions-of-messages/sec-with-microsecond-latencies/

21:24 <nemosupremo> I can't remember if I brought this up - but is there a way to provide your own kafka offsets?

21:25 <nemosupremo> If I make significant changes to my Python code (like even change class names), I notice I have to delete the entire state

21:25 <nemosupremo> and in that case, it seems I may lose data as Wallaroo will start again at the newest offset

22:04 <SeanTAllen> nemosupremo: that work is underway right now. and we are starting on work to allow you to migrate state with schema changes. because pickle includes the class name it does have issues if you change the class name. if you use a serialization format that isn't tried to the class name at all then that wouldnt be an issue (unless the state object schema changes).

22:25 <nemosupremo> I'm trying to understand the simple performance metrics I'm seeing. In my decode message from kafka, I decode I also include the time in the message. At the end, before encode, I print out the time taken from decode-encode

22:26 <nemosupremo> I see that it takes 2-3 seconds to go all the way through, but the numbers dont look right, the first message takes 4s, while the last takes 2s

22:27 <nemosupremo> Does wallaroo buffer messages in each stage?

22:44 <nemosupremo> nvm my computation just seems to be slow

23:11 <nemosupremo> wallaroo def seems to be buffering the messages or something, I'd just like an insight into how. Currently I see a stream of messages every 10 seconds

23:21 <slfritchie> What are you using for your TCP sink process, nemosupremo ?

23:21 <nemosupremo> giles, but I"m ignoring it

23:21 <nemosupremo> I basically print in the encoder function

23:22 <nemosupremo> and what I'm seeing is that the first print statement in a "stream" comes 5-6 seconds after it was decoded

23:22 <nemosupremo> (and the last comes like ~50ms, which is consistent with my compute)

23:23 <nemosupremo> I'm taking a dive into the source code to try and understand if theres any buffering going on after a message gets decoded and what the parameters are

23:24 <nemosupremo> more generally, I'm trying to understand how these buffers may affect our real time SLAs (as I'm trying to come up with a general number of when a message hits kafka how long will it take to get processed?)

23:24 <slfritchie> Do your print statements have a newline in them? Are you using stdout or stderr or another file descriptor?

23:25 <nemosupremo> yeah they have a new line, just print to stdout

23:25 <nemosupremo> def encoder(data): print time.time() - data[0]; return struct.pack(">If", 4, 0)

23:27 <slfritchie> There isn't any explicit buffering in wallaroo's internals that would cause that kind of delay.

23:28 <SeanTAllen> nemosupremo: its not completely linear. each underlying actor for something like decoding, encoding, state computation etc will process up to X number of messages in a go which improves performance.

23:28 <nemosupremo> SeanTAllen: is this a dynamic heuristic, or an internal one?

23:28 <SeanTAllen> it doesnt do 1 message all the way through. we'd read a chunk of data off of a source and decode those and they get routed on to the state computation or stateless computation that processes it.

23:28 <SeanTAllen> but that would not result in that kind of delay.

23:29 <nemosupremo> Why wouldn't that cause a delay? If decode holds my message until it sees X number of messages, isn't that a delay?

23:29 <SeanTAllen> it will process up to 100 messages at any computation at a time, if 100 were available. ditto for a sink.

23:30 <SeanTAllen> how many it reads off a source is different

23:30 <SeanTAllen> no no

23:30 <SeanTAllen> it doesnt wait for X number of messages

23:30 <SeanTAllen> it will process up to X number of messages

23:30 <SeanTAllen> so if there is only 1

23:30 <SeanTAllen> it is processed

23:30 <SeanTAllen> it doesnt wait for 100

23:31 <SeanTAllen> its not a batch, its a limit on the number before it has to give up the scheduler to other processing.

23:31 <SeanTAllen> so if you send 1 message, it will get processed by all steps immediately.

23:31 <SeanTAllen> it wouldnt be waiting for others to arrive.

23:32 <SeanTAllen> does that make sense?

23:32 <nemosupremo> how do you know there's another message and not just 1?

23:32 <nemosupremo> don't you have to poll a socket?

23:33 <SeanTAllen> no

23:33 <SeanTAllen> its all async io

23:33 <SeanTAllen> the tcp source is async io

23:33 <nemosupremo> I might have donkey brains so bear with me

23:33 <SeanTAllen> its alerted when there is data

23:33 <SeanTAllen> it read to...

23:34 <nemosupremo> what I'm currently seeing is every X seconds I get a stream of exactly 500 messages

23:34 <SeanTAllen> ok, so, does that async io for the source make sense or should i go into that?

23:34 <SeanTAllen> are you using the tcp source or the kafka source?

23:34 <nemosupremo> I understand 100% what you are saying theoretically, I'm trying to match that with the output I'm seeing from my application

23:34 <nemosupremo> Kafka Source

23:35 <nemosupremo> In my stream the first message is

23:35 <nemosupremo> 6.519, ('DEVICEID=248170905114908013378', 1519772495.34687)

23:35 <nemosupremo> and the last massage is 0.052, ('DEVICEID=177170626113313718856', 1519772501.857092)

23:35 <nemosupremo> where 6.519 and 0.052 is the number of seconds between that message was decoded

23:36 <nemosupremo> To *me* that looks like a message could be decoded, hang around for 5 seconds, before its ultimately computed and/or encoded

23:37 <nemosupremo> and given that a stream has exactly 500 messages, I'm thinking that wallaroo is waiting for 500 messages before it shuffles data across through the stream

23:37 <SeanTAllen> thats not how it works so that is odd

23:37 <SeanTAllen> ill need to dig into the kafka source and consumer though and see if there is anything with that.

23:37 <SeanTAllen> and talk to Dipin who has done most of that work.

23:38 <SeanTAllen> im going to refer him to his conversation and what you are reporting and get his thoughts nemosupremo

23:39 <nemosupremo> ok, FWIW, I don't care that there is a delay, I'm simply trying to understand how the underlying system works

23:39 <nemosupremo> I don't even think I should be calling it a delay

23:39 <SeanTAllen> yeah that isnt how the underlying system works

23:39 <SeanTAllen> if it happening, its not intended and could be a bug.

23:39 <SeanTAllen> well

23:39 <SeanTAllen> i know it is happening, just not sure why

23:39 <SeanTAllen> i trust that you are seeing what you say you are seeing

23:39 <SeanTAllen> i dont have an immediate explanation

23:40 <SeanTAllen> do you computations take a long time to run? or your messages a long time to decode?

23:40 <SeanTAllen> for some definition of "long time"

23:41 <nemosupremo> I have a to(compute) step that takes like ~50ms, which is all CPU (which is why I consider it long)

23:41 <slfritchie> I wonder if there's an interaction with computation -> actor eliding + Machida's use of `--ponythreads=1` could be the "batching" factor. Multiple computation steps could be computed by a singleactor without the message passing & yielding that an unelided pipeline with multiple actors would have.

23:42 <SeanTAllen> i agree that could happen slfritchie but 500 is weird, given the internal max for messages to process at a step is 100.

23:43 <SeanTAllen> its a single `to` right nemosupremo?

23:43 <nemosupremo> thats what I thought as well, given that even though its async, it still has to "defer" to something else to get new messages

23:43 <nemosupremo> right

23:43 <nemosupremo> source -> to -> sink

23:43 <SeanTAllen> o yeah...

23:43 <SeanTAllen> hmmm

23:43 <slfritchie> 100 input messages x 0.050 seconds total processing time is 5 seconds, which is pretty near 6 seconds. {shrug}

23:44 <SeanTAllen> so something it could be, i need to check with jtfmumm when he starts (hes in europe), is that the `to` is being coalesced into the source. but 500 still seems weird.

23:44 <SeanTAllen> sorry, to coalesced is a rather technical internal term nemosupremo

23:44 <SeanTAllen> it basically means because there is no state that the `to` operates on, we turn into a single step with the proceeding step.

23:45 <nemosupremo> What slfritchie said sounds about right

23:45 <SeanTAllen> that might result in what you are seeing. although batch of 500, still seems odd.

23:45 <slfritchie> (sorry I used "elided" instead of "coalesced")

23:45 <SeanTAllen> if it was 100 at a time that could make sense

23:45 <SeanTAllen> @jtfmumm ^ does that seem correct to you?

23:46 <SeanTAllen> nemosupremo: i suspect that is what is happening.

23:46 <SeanTAllen> high computation times could end up looking like batching.

23:47 <SeanTAllen> we should probably open up a way to change that, either allowing a user to set or based on computation time.

23:47 <SeanTAllen> given that right now the python uses a single thread, it would be much more noticeable there.

23:47 <nemosupremo> If I make my compute function a no-op I see those numbers become .021 and .106 respectively, so I think he's right

23:47 <SeanTAllen> ya

23:48 <SeanTAllen> so that is the "up to 100"

23:48 <SeanTAllen> so theres 100 available so it processes them

23:48 <SeanTAllen> which is generally good for cache coherence

23:48 <SeanTAllen> particularly with state computations

23:50 <SeanTAllen> im off to eat dinner now. slfritchie i leave nemosupremo in your hands.