ten thousand hours

I still remember vividly my first encounter with an SMP computer. This was around 1996 and  it was a dual socket Pentium Pro monster. It blew my mind. Growing up along with x86 evolution I was always curious about hardware operation. My first 8088 board was simple enough to understand each signal line in it. Now this monster machine had two processors, two sets of L1 and L2 caches and RAM shared between them. I was wondering how it was possible to keep caches and RAM in sync as well as how would someone write software for this incredible beast.

Fast-forward to 2012. Two things happened: Internet and power wall.

The Internet ignited renaissance of distributed computing. We often cite а scale of operation that became unprecedented in pre-Internet information systems. But this is true for just a handful of companies. What truly undid traditional information systems is how fast one's capacity requirements can grow when а product hits the hockey stick. This made old systems terribly uneconomical. The scale up of expensive proprietary system was soon replaced with scaling out of generic consumer-level kits.

In the first part of this post I described our thought process for evolving connect.me from a single database to a partitioned database with a fixed number of partitions. We decided to partition on the dimension of user profiles because we are expecting this dimension to exhibit the most growth. We call partition a cell and we use an external key provided during authentication to choose a cell where a profile is or will be located. Such a key is a combination of a provider name, such as twitter or facebook, and a unique identifier within a provider, which we call a proid. Internally we identify a profile by a 64-bit integer identifier called a puid. We use puids to express relationships between profiles, as in vouching. Each cell handles a dedicated range of puids from which it allocates new ones sequentially starting with some puid base. It is possible to quickly find a cell for a puid by checking puid ranges. Routing is the process of resolving a provider-proid pair or a puid to a corresponding cell.

At connect.me we started building our web application with one relational database implemented in Postgres. The idea was to quickly iterate on the feature set to solidify our data model. In the later stage we planned to work on a scale-out approach that would allow us to add server machines without major service disruption. The objective was to support exponential growth, which we could face with viral elements in our product. It turned out to be an interesting and challenging problem and I want to share our thought process and the resulting design in this blog post.

The essense of connect.me data model is to represent people profiles and vouches. A vouch is a directed edge in the reputation graph--there is one giving and one receiving profile. Also the vouch always has a specified tag, such as “guitarist” or “cyclist”. There are other aspects that make it more complicated, but for the purpose of this discussion this simplified model is sufficient. With a single relational database this model has a trivial implementation. The profiles table has a unique 64-bit integer id--we called it a puid--along with other profile attributes. A new puid is generated with a sequence starting at 1. The tags table also has a sequence generated 64-bit integer id along with a tag name. The vouches table has source puid, target puid and tag id.

When a database is limited to running on a single computer, only certain load can be served with acceptable latency. Adding hardware to this single computer will help only so much. Doubling or tripling the load may require significantly more than twice or thrice cost of hardware to scale up. Such approach is expensive to scale and it eventually hits its limit. Ideally, we would start with a single inexpensive computer and as load increases we would keep on adding same inexpensive computers resulting in a near-linear function between load and cost. Such horizontal scaling out is common place in today's web applications because it provides a more predictable cost model. 

Clusters of inexpensive commodity hardware lead to a disruption in the database ecosystem. What further amplifies this disruption in falling prices of RAM and solid state storage. A number NoSQL of databases are truly leveraging this disruption. At Meshin, one of our key requirements is the lightning-fast delivery of query results. Think of Google Instant search. Showing results with "as you type" latency enables a whole new class of use cases. This lead us to considering various in-memory database engines. Redis, with its simple and elegant data model and very transparent performance characteristics came out on top. For in-memory databases, single-threaded design becomes an obvious choice, essentially removing significant overhead common to traditional database architectures. Another assumption in Meshin design is that if the index is partially or completely lost it can be recovered by re-indexing original data. Sure, this introduces short-term inconsistencies, but on the other hand it allows to relax durability and further simplify the design. Such a trade-off is a perfect fit to the Redis persistence strategy.

Single-threaded design of Redis brings up another interesting issue. Scaling of a single instance of Redis is not limited just by the computer it runs on, which would amount to RAM size, but is further limited by the single core or hardware thread it will utilize. So if your load exceeds the single hardware thread or size of available RAM, you need a second instance of Redis. This requires some approach to clustering. Effective clustering is all about figuring out the right partitioning scheme. Ideally, you need to split load into identically-sized partitions. If load becomes biased towards one or more partitions you will have a bottleneck. Such a bottleneck will limit your system's scalability or in other words will make your load-to-cost function non-linear. It is important that partitions are as independent as possible. If an operation spans a group of partitions its scalability will be limited by the number of such groups in the cluster. In extreme cases of spanning all partitions in a cluster, the scalability will be as good as running on a single partition. 

Another aspect is differentiating the load by reads and writes. If the Redis hardware thread capacity is exceeded by reads, it is easy to scale out by putting additional read-only replicas of the same data. It is much harder to scale out writes by replication where trading off consistency is often required. Redis again takes simple a approach with its master-slave replication. A writable master replica asynchronously updates one or more read-only slave replicas. Right after replying to a write operation the master notifies all replicas. No acknowledgment is required by the master. This means that there is short period of time when slave replicas may return old data. This provides with write-write consistency guarantees that are as strong as without replication. However, the write-read consistency guarantees are now less strong or "eventual". It is important to note that replication not only helps scalability of reads, but also improves reliability when replicas are placed on different machines. For the Meshin application eventual write-read consistency is acceptable tradeoff for higher reliability.

With the Meshin application, we have a large number of user indexes, each of roughly equal size. Meshin keeps a sliding window index of email, Twitter, Facebook etc. messages to maintain predictable maximum size of index. This provides a great opportunity for partitioning. If a partition handles roughly an equal number of users, we have balanced storage and load requirements and at the same time made the most frequent operations directed at only one partition. One approach is choosing between N partitions is hashing a unique user identifier, for example an email address, such that the hash space is in the [0..N) range. If the quality of the hash function is sufficient, we will get a well-balanced distribution.

Now we can place our Redis instances on M = N / K computer nodes in the cluster. K is number of Redis instances that we can comfortably run on single node. This number is limited by the number of cores or hardware threads. With a large number of hardware threads (8 and up) it makes sense to map each Redis instance to one thread. RAM is then equally divided between these instances. Now, knowing each user's maximum required RAM, we can calculate how many users our cluster can accommodate. One thing to keep in mind is that if you use Redis snapshotting for persistence -- which employs the OS copy-on-write mechanism -- it may require as much as twice amount the RAM dedicated to a partition to take a snapshot.

It is important to note that in the case of the Meshin application with regard to specific RAM and thread count per computer the load is "memory-bound". Meaning that a partition having a full memory pool sill won't saturate the hardware thread with Redis. It is important to note that for some applications the load may become "thread-bound". In this case the hardware thread is being saturated before the memory pool is maxed out. Interestingly, with the "memory-bound" case we have underutilized CPUs and with the "thread-bound" case we may have underutilized RAM.

Great so far, but what if we keep on adding users and they start exceeding the capacity of a partition? In the "memory-bound" case they exceed the memory pool allocated for a partition and in the "thread-bound" case they fully saturate the corresponding hardware thread. In our case, the number of partitions is fixed. Also, since partitions are equally populated with same size users they all about to overflow. As we add new nodes to the cluster, we may try to move some partitions from existing ones and rebalance such that our nodes have an equal number of partitions. There are two problems with this approach. In the "thread-bound" case this won't help at all -- we can't really utilize new cores, since partitions are single threaded! In "memory-bound" case the machines will have fewer number of Redis instances than number of hardware threads. This leaves those threads idle and CPUs even more underutilized! One way to fix this problem is to overprovision partitions initially, so that K is a multiple of the number of hardware threads. The multiplier is essentially the number of times we want to grow our cluster by. Other approach is to allow changing number of partitions.

At Meshin we took approach where we are able to add partitions over time. This is accomplished by changing the way initial partitioning is done. Instead of limiting the hash function space to (0..N] we allow it to generate 64-bit unsigned integers. These hashes are mapped to a ring and wrapped from 2^64-1 to 0. We compared several hash functions homogeneity performance for a large set of email addresses as unique user identifiers. The best homogeneity was achieved by the SHA1 hash reduced to 64 bits by XOR-ing its parts.

Each partition is placed on the ring with equal distances in between. When a user identifier is hashed we get its position on the ring and thus can find the closest partition where user's index resides by going clockwise.

When new partitions are added they are placed in between existing partitions to maintain equal distances. We also have the flexibility to add partitions in multiple points on the ring if partitions are of variable size to enable more users' indexes allocated in larger partitions. Now, to lookup an existing user's index you may hit a newly added partition and miss the old partition where the index was originally allocated. One way to address this problem is to move the corresponding user indexes to new partitions or rebalance the whole hash ring. We decided to employ simple routing instead. If a partition doesn't have the user index, it proceeds to the next partition in clockwise direction.

Such routing is fairly scalable. It does not have a single point of contention since each partition participates in routing. We also use a caching layer on the application server to remember partition addresses for recent user identities. As with lookups, it is also possible to route new user index allocations. When a partition is full it will simply forward to next partition clockwise until a partition available for allocation is found. If new partitions' points are added in between old ones this will keep routing paths short. 

To add reliability and further scale reads we replicate each partition to a different node in the cluster. Replicas are placed so that each node has an equal number of masters and slaves. All writes are directed at the master replica, while reads are balanced between master and slave replicas. Balancing uses stickiness by user identity to keep CPU caches warm. For example, replicating N partitions twice, K partitions per node, 2 * K replicas per node and M nodes in cluster.

Partitioned replicas are taking snapshots asynchronously. We place the snapshots on redundant networked storage. All snapshot traffic goes through a dedicated network. Taking snapshots requires additional memory to be allocated by Redis. The amount of this memory depends on write activity on a partition and I/O latency of snapshot operation. To minimize snapshotting overhead on a specific node in the cluster we run only one snapshotting operation at the time. A dedicated agent calls the SAVE command for each partition on a node in a round-robin fashion. The SAVE command initiates snapshotting and returns when it is completed. The period of full cycle of snapshotting defines the maximum time slice of changes that could be lost if cluster a node fails.

Cluster configuration, including partition IP addresses and ports, positions on hash ring and health check status are stored in dedicated Redis instances. We call this dedicated instance the "directory". The directory is replicated as well with the master doing the writes, which are quite rare. The directory reads are served by the master and one or more slave replicas in round-robin manner. When the application fails to access a certain partition replica, it marks it as offline in the directory. This means other instances will stop trying to access it since it's been removed from the list of available replicas for a partition. If a failed replica was in slave role the cluster will remain fully functional, just with less redundancy for failed partition. We call this partial outage. In turn, if the replica was in master role, the partition will switch to read-only outage mode. Read-only outage mode still allows the partition to handle read queries. If all replicas in a partition have failed cluster is in full outage. Full outage is a rare event since it requires two nodes to fail simultaneously. A special agent monitors partitions for outages by polling the directory. If there is master-less partition it will elect one of the slave replicas to become master, reconfigure the rest of the replicas and adjust the directory to clear the read-only outage. While we attempt to recover from read-only outages automatically, partial outages and full outages must be addressed manually. At some point in scaling this will not work and we realize the need for a fully-automated recovery system down the road.

Meshin application relies on back-end triple store for holding person semantic index and for processing front-end queries. This back-end store is built on top of open source in-memory key-value database Redis. Before getting into details of how triples are represented and queried I will briefly introduce essential Redis features. Fill free to skip following paragraph if you are familiar with Redis.

Redis is key-value store where keys are binary strings and values can be either simple binary strings or higher order data structures. These data structures include ordered lists, unordered unique sets, secondary level hash tables (hsets) and weight sorted sets (zsets). Redis exposes its functionality through simple text based protocol. Protocol defines number of commands and corresponding replies. Commands are either general for all kinds of key-values or specialized for value type. For example SET A X associates binary string value X with key A.

RPUSH A X appends value X to the end of list at key A.

SADD A X inserts value X into unordered set at key A.

HSET A P X inserts field P with value X into hset at key A.

There are several commands that allow to combine values in interesting ways. For sets it is possible to union, intersect or subtract sets and store the resulting set with new key. Corresponding commands are SUNIONSTORE, SINTERSTORE, SDIFFSTORE. Even more interesting is "for each" operation with patterns expanded with each element of input set or list. When expanded star (*) character is replaced with each value from input. Pattern can be just "key" for binary strings lookup or "key->field" for hsets field lookup. For example if pattern is A* and input set is A, B, C command will produce output from string values of AA, AB, AC keys. Now if pattern is *->P and input set it A, B, C command will produce output from values of A->P, B->P, C->P hset fields. In addition "for each" operation allows sorting and paging of the result by yet another specified pattern, therefore the name of command is SORT. SORT command also allows to store resulting list with new key. 

Combination of SUNIONSTORE, SINTERSTORE, SDIFFSTORE, SORT and similar commands allows for interesting use case. Sufficiently complex query can be expressed as dataflow between inputs, intermediate results stored as temporary keys and output result returning optionally sorted and paged values. Ability to store intermediate result in temporary key is essential for performance -- it allows to avoid roundtripping of intermediate results between Redis database and application server. In addition Redis enables pipelined execution where multiple commands are sent without waiting for replies. This creates condition for "stored procedure"-like execution with single roundtrip to database. Experimentally Redis supports embedded Lua scripting where instead of pipelining multiple do-and-store commands it is possible to submit single EVAL command with equivalent Lua script.

Now let's illustrate such dataflow approach with specific query and triples between person, address, participant and message objects from previous post. Let's say we want to retrieve ids for twitter messages exchanged with specific person ids: P0, P1, P2. Messages need to be sorted by their timestamp in descending order and paged to top 20. Before jumping into query details we first need to understand how triples are represented in Redis. For example we have the following object -> predicate -> subject triples that assumes uniqueness of predicate per object.

A0 -> person-of-address -> P0 A1 -> person-of-address -> P0 A2 -> person-of-address -> P1

This means that person P0 has two addresses A0 and A1 and person P1 has one address A2. In Redis we place the following hset keys to represent object to subject reference.

And following set keys to represent reverse subject to object reference.

Now let's look at query dataflow. 

First we initiate temporary set T0 with P0, P1, P2 in it.

Next we need to get set of corresponding address ids. To do this for each set pattern "*:person-of-address" we union and store in T1. If person object P0 has two address objects with ids A0 and A1 pointing to it there will be reverse reference set P1:person-of-address containing A0 and A1. Now T1 contains union of corresponding address ids.

Next we need to filter and leave only address ids of "twitter-address" class. To do this we intersect "twitter-address:class" set with T1 and store in T2. "twitter-address:class" set is similar reverse reference set for all twitter class ids.

Next we need to get set of corresponding participant ids. Again to do this for each set pattern "*:address-of-participant" we union and store in T3. Keeping in mind that "A0:address-of-participant" would contain all participant ids pointing to address A0.

Next we need to get corresponding message ids. This is done for each hset pattern "*->message-of-participant" we lookup field value and store into T4.

Finally we sort T4 by hset pattern "*->date-time".

Returned result contains pairs of message id and subject.

Note that SFOREACHUNIONSTORE command does not exist in Redis. Originally you would have to resort to forking Redis and coding new command or do it in application server code, which would hurt performance. Using Lua scripting extension of Redis it is very easy to implement SFOREACHUNIONSTORE with EVAL command and following Lua script.

local p = string.find(ARGV[1], '*') local pre = string.sub(ARGV[1], 1, p - 1) local post = string.sub(ARGV[1], p + 1, -1) local x = redis.call('SMEMBERS', KEYS[2]) for i = 1, #x do   local y = redis.call('SMEMBERS', pre .. x[i] .. post)   for j = 1, #y do      redis.call('SADD', KEYS[1], y[j])   end end return redis.call('SCARD', KEYS[1])

Meshin application attempts to unify various personal communication streams and provide value on top of that. User registers streams from cloud-based carriers and installs Meshin app on their smartphone. Cloud-based carriers including Gmail, Facebook messages, Twitter DMs, LinkedIn messages and smartphone data sources like phone book, call history and SMS messages are continuously crawled and indexed by Meshin. Separate index in maintained for each user. Index includes aggregated contacts and each contact's corresponding message stream across all the carriers and smart phone data sources. Contacts represent both people and organizations. Person contact is derived from display name of email address or similar metadata in other carriers. Organization contact currently is derived from email address Internet domain name. Once initial indexing is performed Meshin smartphone app allows user to group and filter communications in various ways. We call it slice and dice streams. Simple ones include streams from specific contacts -- people and organizations. Also it is possible to define explicit groups of contacts. Each slice and dice stream of messages allows further filtering by time and keyword search. With every message Meshin associates read/unread status, so that user can see what's new in each stream. Currently Meshin read/unread status is maintained independently of carrier read/unread status. In addition to explicit user defined grouping of contacts Meshin creates implicit group of contacts with high importance. Importance metric is calculated by proprietary algorithm and is based on interaction patterns accross all carriers. Factors like frequency of communication, recency, directness and reply timing are used.

Meshin index is implemented as triple store based on key-value backend database engine. Separate crawler and indexer exists for each carrier to deal with specifics of carrier API and data model. Indexers convert carrier data model into Meshin triples. Example of such triple is connection between person object and address object. For example email indexer produces following triples.

Then Twitter indexer produces.

Now if both indexer agree on the way of person object identity generation we achieve disambiguation where now both address objects point to the same person object.

This can be done by applying hash function to person name. But what if Twitter metadata included variation of the name that is not exactly the same with one in email metadata. Variation can be simple capitalization alternative (john smith), inclusion of middle names and prefixes and suffixes (Mr John K Smith III PhD) or simply variations of first name (Johnny Smith). Meshin employs several normalization steps in order to get unambiguous identity. But often this approach still does not work and we end up with several person objects representing same person. In this case Meshin allows user to manually unite several person objects into one. Another problem may arise when false disambiguation happened. In previous example this may be when John Smith via email and John Smith on Twitter are two different people. In this case, again we allow user to split person object into several where each has dedicated address association. As I mentioned above organization object is recognized based on Internet domain name of email address using correspondence dictionary. For example email indexer produces following triples.

Meshin organization dictionary is mainly derived from Dbpedia dataset. Each organization can have multiple Internet domain names associated with it. Organization node identity is generated as hash of Dbpedia identity. Additional attributes such as Wikipedia page link, home page link and short abstract can be included.

Messages are represented by corresponding objects with identity derived from of unique reference from corresponding carrier or, if such does not exist, hash of combination of key attributes. Message object has attributes such as time stamp, subject, body preview and also references participating addresses through intermediate object. For typical email messages indexer produces following triples.

In addition message objects are referenced by inverted keyword index on all text attributes such as subject and body. Keyword index is stored within the same key-value back-end database engine. The combination of "person-of-address", "organization-of-address", "address-of-participant" and "message-of-participant" predicates, ordered date-time predicate and inverted keyword index allow Meshin to express variety of slice and dice queries mentioned above. Such queries are expressed as dataflows specific to Meshin back-end database engine.

Indexer submits triples to Meshin store in batches. Sizes of such batches differ depending on mode that indexer operates in. When indexing new user and carrier combination batches can be large, while if indexer "catches up" with newly arrived messages they are small. After each indexer submission store checks for affected predicates and may trigger additional analysis within user's shard. Such "trigger" for example is used to calculate person prominence (or importance in previous terms). We decided to perform this calculation in the store instead of indexer because we wanted computation to be closer to the data and to enable ability to compute with data across carriers.

Semantic web is beginning to mature with adoption in many mainstream applications. I would like to propose vision for personal semantic web or network. It takes mechanisms of former and applies them in more person centric and real-time way. One key driving force for personal semantic network is proliferation of connected sensors that are carried with us at all time. But often sensors are not necessary anymore because activities are getting fully virtualized as with digital communication over email, phones and social networks. Next is continuously falling cost of storing and analyzing large amount of data in real-time. Last but most important is mass adoption of mobile apps that can truly leverage advances in previous two. It is important to note that the nature of data collected in such personal semantic network can subject to serious privacy concerns.

Connected sensors and real-time crawling of digital communications enable to collect and continuously push to centralized store various person centric signals. These include interaction with individuals and organizations. Such interaction may include messaging via communication carriers like email, phone and SMS, social networking. They also include explicit meetings recorded in personal calendar as well as implicit collocation through geolocation checkins. Signals could also include current geographic location and movements through continuous geolocation. Variety of signals can appear from person generated content such as written and voice notes, photographs.

Such signals collected in real time represent fire hose with little value by itself. Being aggregated and analyzed over time it may deliver important deductions about person context. Just several examples. Real-time social importance for already connected individuals and organization as well as ability to predict importance of new ones. Another is predictive traffic alerts based on over time moving behaviour. Yet another is ability to use person generated content such as photographs or notes as automatic tags for individuals, organizations and places.

Personal semantic network is a vision for system that can deliver likes of mentioned deductions to layer of specific applications. Often application usefulness is a function of explicit input that it requires. Fine balance is formed between amounts of input needed to produce useful result. Good example is search.  With keyword input it takes us this far. But once you are adding more facets like time, location or richer tagging it may produce better or more relevant results. Although this gets offset by requirement of more explicit input and end-user value deteriorates. Key is to hide these additional inputs from the end-user providing them implicitly derived from user context. Systems like personal semantic network can help in establishing such context, which may dramatically shift the balance for end-user to less inputs with more relevant results.