I am getting sick of talk about twitter and it’s scalability problems and also frankly unqualified people slagging the service for it’s unreliability and also coming up with stupid ignorant answers to how it should be fixed.

As part of our future we do see a many-to-many problem arising so I put some thought to how we would tackle it (like any of our other problems) by doing some ‘planning’.

In a recent post Robert Scoble tries to explaining how twitter works by saying that twitter is using some form of ‘pivot table‘ – (my terminology for what he explains) and says that a model that others have put forward (i.e. a de-normalized system of inserting messages into everyones queues) was akin to microsoft exchange, now these two examples are so horribly not connected – and I won’t rant about how BAD exchange is efficiency wise, but please Robert do not get into any technical arguments please.

How it should be done

So what did I come up with for solving the many-to-many problem?

Firstly let me explain my terminology, I use the term ’shard’ because a shard could be a physical server or a daemon running alongside other daemon’s running on a server or even a grid platform.

The basic idea is to build a system in which no one particular element is unscalable. So the design takes into account that however big things get just larger numbers of shards would be required.

First a diagram (which I hope makes some sense of the complexity described below)

Queue Shard

The queue shard stores the message queue for a set number of users the shard would be found by a hashing mechanism that would map the user id to the particular user. The queue system would use a purely in-memory storage that would allow the for the really fast insertion speeds that would required.

The other really important point to make is that this server is only dealing with dates and the id of the tweet that it references (tweet id’s covered below)

Friend Shard

The friend shard stores the friends list for a number of users, this is likely to be a larger number of users per shard than the queue shard purely because its job is a lot easier. So lets say each shard can work with 100,000 users. When this shard gets told about a new message for a user it looks up its list of friends.

Then using the hashing algorithm that maps each user id to a specific queue shard it builds up a list of friends that live on each of those queue shards.

So in the extreme example of Robert Scoble of 25k friends we hope to split up the task so that no queue shard would have to deal with more than 1000 inserts.

Joiner Shard

Because we are only storing the id’s of each tweet in the queue when a query comes in for a stream we fetch the list of the 20 (or however many) tweets from the queue shard for that particular user then we need to go get each actual text tweet from a tweet shard. The joiner shard would very likely use a memcache (across a lot of servers) so that the tweet storage would get hit as little as possible.

The second thing to note is that as described the queue shards are in-memory only – they are storing data very inefficiently (e.g. totally de-normalized) but it would be a serious overhead for them to store to disk. The joiner shard would need to be able to cope with only getting partial results back from the queue shard – and then have to look it up from the ‘normalized’ version of the database (described below)

Tweet Shards

When new tweets come in they are stored in a tweet shard, we use a hash algorithm to spread them across a set number of servers. The algorithm should take into account that over time tweets get moved onto archive servers which are designed more for mass storage rather than quick retrieval.

Normalized Queue Storage

Because the in-memory queue shards are very fast but very inefficient it would be not sensible to store more than a weeks worth of data in them (and maybe even less) – as the many-to-friendlist-to-queue system even when stored just using id’s is still a large storage overhead compared with storing the data in a standard pivot/normalized style. So for every request that comes in it is also passed into this shard to be archived and also for when you do get the rare request for older data.

Web Shard

This is the easy bit, the other shards described so far have all done the complex database work, the web shard is responsible for dealing with the actual requests be they API / HTTP / whatever, and then pass on the request into the relevant other shards.

I will describe below the order in which the shards are called for particular actions (the two that are most common)

Sending a Tweet
The following is a rough step list of how a new tweet gets insert into the database.

1. Web Shard – Receives the request – First it sends a request off to the Tweet shard to store the message itself which then returns an ID that will then be used for all the other requests. Then it looks up which friend shard stores the particular user that the request is coming from.
   1. Tweet Shard – Receives message to store the tweet and returns the new ID
2. Friend Shard – Receives the message – then sends out signals to all the relevant queue shards for all the friends of that user.
3. Queue Shard – Gets a list of friends (that it deals with) and inserts the id of the message into each of their queues.
4. Normalized Queue Shard – We also store the data in this archive database for long term storage (and slow retrieval)

Building the Stream
This is a few short steps of how we build up a stream for a particular user.

1. Web Shard – Receives the request – sends a request to a joiner shard asking for the stream for that user
2. Joiner Shard – Looks up which queue shard has the data it is looking for – gets the list of id’s – then looks up those messages within the tweet shards – it then puts them together and returns them back to the web shard
3. Tweet Shard – Returns the actual text tweets

The above could be more complex in circumstances when the data is not available within the queue shards – or within the tweet shards – at which point they both seamlessly go look up the data in the archive instead.

Conclusion

This looks complex – (well it is..) but in general terms of complexity of building large scale web applications this is pretty simple stuff (fav.or.it has far more complex problems!). I understand that the twitter people now have the extremely hard task of slowly improving each module rather than a complete redesign. Whatever the future I wish them well. btw – if you want you can follow me (and fav.or.it’s exploits) via my twitter feed

(just as an aside – the fav.or.it blog will be going through a major overhaul soon and more of our wonderful employees we start posting (they already have their own individual persona blogs) so we have had to cater for this in our new design, but fear not we will also be splitting up the feed so if you only want product updates you can still have that, if you want development/tips/programmer/design stuff you will also have the option to subscribe to an overall feed or to individual topics.)