How Uber Manages a Million Writes per Second
highscalability.com
highscalability.com
--
I like the part where 10% performance overhead is not a big deal while at the same time making a big deal about how mesos can save you 30% performance overhead.
And notice how at first they had a group of clusters for completely separate things like API, storage, etc, but now with mesos they program against an entire datacenter by using a different specific API for separate things like APIs, storage, etc. Much... better.
Also now with mesos they can run different kinds of services on the same machine. Kind of like containers. Kind of like separate applications on any kind of machine at all.
"If, for example, one services uses a lot of CPU it matches well with a service that uses a lot of storage or memory, then these two services can be efficiently run on the same server. Machine utilization goes up."
Because if I have a machine pegging 90% CPU, I totally want an i/o heavy job running on that machine, because interrupts are, like, imaginary, man. Yeah, i'm sure the machine's utilization goes up... right up to triple-digit load.
"Agility is more important than performance."
So the ability to automate rolling out a virtual machine and application in as abstracted a way as possible is more important than their 1-million-writes-per-second goal. I think I know why they don't find 10% loss to be a problem now... they can just keep spinning up machines until they reach their performance goal. 150 machines per DC to store ride-sharing data seems a bit much, but maybe there's a lot more going on behind the scenes than the few features on their app.
Honest question: can't you dedicate some cores to handling the interrupts, leaving the others for the other processes?
I've mostly seen it for load balancers. Pin the network card interrupt on one core and the load balancer on another.
In fact, modern kernels support this quite extensively and the RHEL7 documentation[1] has great tips on it for someone new to it.
You can tune an interface for max throughput via coalescing[2] on Linux via ethtool -C to change and ethtool -c to view current coalescing. Setting interrupt per packet helps with latency for certain workload types in addition to SO_BUSY_POLL[3] and the global or per-interface busy polling sysctl. However, interrupt per packet can trivially overwhelm CPUs if you don't isolate those cpus using things like isolcpus= on the linux grub commandline or cpu_exclusive=1 in a cpuset. RFS/RSS and NICs with multiple receive queues make this much easier to tune.
It is guaranteed to use more power, but you can do idle=poll[4] on the Linux kernel boot command line to stay in the highest C state and help with network latency.
You don't really always need an interrupt either as apps can DMA directly to and from the NIC <---> Application bypassing the CPU. How do you think RDMA works at a lower level? Finally, Linux "kernel bypass" networking aka 100% userspace tcp/ip stacks can be written to be truly interrupt-less take a look at Mellanox's VMA or Solar Flare's openonload. Heck, the default openonload config is interrupt-less. The more you know :)
[1] https://access.redhat.com/documentation/en-US/Red_Hat_Enterp...
[2] https://en.wikipedia.org/wiki/Interrupt_coalescing
[3] http://man7.org/linux/man-pages/man7/socket.7.html
[4] https://access.redhat.com/articles/65410
EDIT: This is a fantastic and simple overview with actual netperf results showing how this helps with latency: http://www.intel.com/content/dam/www/public/us/en/documents/...
It also doesn't matter because heavy i/o will take up both system and user cpu, and the 90/10+ split (is the app even multi core??) puts you into full utilization, which is fine for bulk jobs, terrible for high performance requests. Even a single machine at 100% can (in unfortunate circumstances) cause domino effects. Better to build in excess capacity as a buffer for unexpected spikes, which means managing your clusters to not stack jobs which could compete for resources, but also not stack jobs that could unintentionally starve other jobs - this requires intelligent load balancing that's application and job specific. Or a cluster dedicated to specific jobs (which they have, ironically)
e.g. One core per network card.
Linux has some really impressive knobs[2] for optimizing these sorts of weird workloads.
[1] https://lwn.net/Articles/382428/
[2] https://www.kernel.org/doc/Documentation/networking/scaling....
They're clearly tracking their own latency and reliability and comparing it to their availability goals - there's numbers right in the article, so I would think they would notice this and handle it somehow.
So I guess either they're lying in the presentation, or your sarcasm is targeted towards a problem you conjured up on you end for your personal satisfaction.
I'm curious how they can start up one new node in a cluster per minute, when last I saw Cassandra required 2 minutes after bringing a new node online for things to settle before starting a new one. Its a pain point for me running tests that need to startup and teardown Cassandra clusters, and I'd love to know when and how it can be avoided.
(Maybe I need to watch the actual talk)
Under 32GB, the JVM uses 32 bits pointers. [actually slightly under 32GB]
Over 32GB, the JVM uses 64 bits pointers.
Pointers consume a lot of space in a running program. Basically, a 30 GB heap (with 32bits pointers) is the same as a 40GB heap (with 64bits pointers).
That means, there is a chasm between 30-40GB heap where it makes no sense to run a server. The additional space is wasted by the bigger pointers.
Second, 30GB is a sweet stop. It's simple to justify and enough for most applications. (A bigger heap will give longer GC time and other side effects, most people don't need to get into that kind of optimizations).
Why is read 10x more expensive than writes?
I've seen some databases work in "append only" mode. They write new data to the end of the file. They never erase existing data.
It's generally a very efficient write patterns (even on good old spinning drives) and it allows to always write in batch.
On the opposite, read are expensive, they require to "find" stuff from various places and read it and verify it and [if configured] repeat on multi nodes to compare the values.
I think newer data and tombstones would stay for a while in sstables, not in the log.
To get 1M/s, we need 15M cars.
[1] says there are about 2M trips per day, so does this mean Uber records the location of all users every 30 seconds, rather than only when they're waiting for or in a car?
If they all have 16 cores 100GB RAM it's really unimpressive and we just found out why Uber is not profitable.
What's the reason behind this?
Is there something obvious that I don't see?
What if some rider starts in city X and ends in city Y (New York to Hoboken)? How do you elegantly handle that transaction? You would need consistent transactions across two databases. Usually its not possible, or at the least extremely inefficient, to implement transactions at the application layer - especially in a distributed system under heavy load.
Also you end up with a lot more operational responsibility. If (when) nodes fail, how many nodes failed? How many instances of Casandra were effected? If there were X node failures, was there data loss to any of the effected instances? If there was data loss, how can we recover from a snapshot and replay all recent data to that instance without replaying it to others?
Meanwhile, is the application layer really where you want to solve these scaling issues? What happens if a single city becomes too large to handle, do you partition again by neighborhood? If you're going to invest into anything, invest into improving the database. Maybe create a specialized database to handle your usecase more generally. Ideally, you want the application layer to only provide heuristics that the database can use to efficiently partition its data.
If I define that incomplete data equals to corrupt data, which I do, I am incapable of conceiving a scenario where corrupt data would be acceptable.
Why would one define it so? Because I'm convicted that computers should always deliver correct data, and in my conviction, data can never be correct unless it is also complete. Therefore, atomicity or bust; therefore, Cassandra = false, and in such a case, milion writes per second is of no consolation.
Therefore, I do not accept the premise that it is impossible to design systems which are both correct and fast on a large scale, and neither should anyone else have to accept such a thing. It's just not right.
CAP theorem says otherwise. I suggest you try implementing something as simple as distributed counters.
You DONT need atomicity all the time, sometimes you do (Financial Transactions), but a lot of the time you dont:
Facebook likes. Do they need to be atomic? Does it matter if a web request gets a stale "like count" and the number of likes for an item is 12,544 instead of the 12,545 on someones page view because the database isn't atomic? No, it doesn't matter.
Uber GPS data, Uber writes the GPS locations as your trip progresses - does it matter that the last 1 or 2 locations might not be visible to the app or external reports for a few seconds? No, it doesn't matter.
Ad-click data, Does it matter that the last 1 or 2 hundred impressions are not visible in a report who's absolute values are in the millions or 10s of millions? No, it doesn't matter - even though this can represent money, the value of a couple of hundred impressions missing on the report is chump change, not even worth thinking about.
With the exception of financial transactions, some safety critical machine data, and a few other cases, atomicity is not required, and that's where these distributed databases shine.
Bingo. No atomicity, bad money or no money. I'd have a problem. I also have this extreme dislike of the "garbage in, garbage out" principle, so either what I'm delivering is correct, or I should be fired, because using computers at that point is pointless. If we don't care about correctness or completeness at all, then we just discounted two of the three reasons why computers exist at all. That leaves us with speed, and at that point, what use is it?
It doesn't have to be expensive to be fast, reliable and correct.
In a commercial context, computers exist to serve human needs. If increasing correctness or completeness serves neither a user need nor a business need, then the work to do that is just developer self-indulgence.
And honestly, I don't have a problem with a little developer self-indulgence now and again. Sometimes I'll just clean up something because it's ugly and I don't like it. But it's important to recognize when we're doing that, because it's decidedly not what we're paid for.
It's a lot harder to go out and build something that actually delivers your ideals.
If you think a distributed database should provide full correct completeness from all nodes while hitting 1 million writes per second, go build one that does it. If you succeed, not only will you make yourself happy, you will make yourself a shitload of money.
I do that for a living. If I didn't, I would not have commented.
In fact, financial transactions rather famously do not implement strict consistency.
> instead of the 12,545 on someones page view because the database isn't atomic?
> No, it doesn't matter.
Actually, it does matter. Atomicity isn't the specific issue there, but rather it's read-after-write consistency guarantees that are important. That is, product-data writes need to be immediately consistent (synchronous) in the context of whichever data store is serving subsequent read requests for the same user/session.
Read-after-write consistency is essential for social networks. If you take an action and then click a new link -- or even just refresh the current page -- the effect of your action definitely needs to still be visible.
Facebook doesn't use Cassandra, fwiw.