There is nothing new here. Why is the Disruptor pattern relevant?

Started out with my blog post about learnings from actor development Got a comment on DZone about the need for assigning threads to a core, laughed it off as a troll The commenter claimed that the JVM's life expectancy was limited if you couldn't assign a thread to a core I didn't get it - why would you want to take control of a core when an OS does that job for you so well? LMAX changed my thinking - I didn't understand high-throughput well enough LMAX is handling 6 MILLION transactions per second using a single-threaded computing model, and they're doing it with Java So the troll was half-right: there is a use-case for assigning a thread to a core, but you CAN do it in the JVM and get tremendous results

How Did They Arrive at the Disruptor

• Looked into J2EE, SEDA, Actors, etc - couldn't get the throughput they desired
• So named because it has elements for dealing with graphs of dependencies to the Java7 Phaser concurrency type, introduced in support of ForkJoin

ASK MARTIN THOMPSON: Why was this done on the JVM instead of C++ or another native implementation? Ola Bini talked to Martin Fowler, and they believe that the benefits of using the Java platform outweighed the potential for marginal throughput gains in C++

Mechanical Sympathy

• Martin Thompson, the lead architect at LMAX, believes in the concept of Mechanical Sympathy, a term coined by former F1 champion Jackie Stewart, who believed that all great drivers had to understand at some level how their machine worked to derive the fastest time driving it

Keys to the Disruptor's Performance

• Never release the core to the kernel (thus maintaining your L3 cache)
• Avoid lock arbitration
• Minimize usage of memory barriers
• Reuse pre-allocated memory to avoid GC and compaction

Thread Management

• In a low latency, high throughput system with balanced flow, you can assign a thread to a core from the JVM by never releasing control of the thread (wait, ForkJoin)

Avoiding Locks

• Locks, extremely non-performant due to context switching in the kernel which suspend threads waiting on a lock until it is released
• Note that during a kernel context switch, the OS may decide to perform other tasks not related to your process, thus losing valuable cycles
• CAS semantics are much better, since no kernel context switch is required for lock arbitration, but the processor must still lock its instruction pipeline to ensure atomicity and introduce a memory barrier to ensure that changes are made visible to all threads
• Memory barriers are used by processors to indicate sections of code where the ordering of memory updates matters - all memory changes appear in order at the point required (note: compilers can add software MBs in addtion to the processor's, which is how Java's volatile keyword works)

Minimal Use of Memory Barriers

• Processors only need to guarantee that the execution of instructions gives the same result, regardless of order, and thus perform instructions out of order frequently to enhance performance
• Memory barriers (volatile) specify where no optimizations can be performed to ensure that ordering is correct at runtime

What's Wrong With Queues

• Unbounded queues use linked lists, which we've already discussed with respect to memory contiguousness and strides
• Bounded queues use arrays, where the head (dequeuing) and tail (enqueuing) are the primary points of write contention, but also have a tendency to share a cache line
• In Java, queues are also a significant source of garbage - the memory for data must be allocated, and if unbounded, the linked list node must also be allocated; when dereferenced, all of these objects must be reclaimed

Memory Allocation

• What if you designed your system to utilize on-board caches for a core as effectively as possible?
• Allocate a large ring buffer of byte arrays on startup, and copy data into those arrays as received/handled
• Sequencing of data in main memory is also important - as the core understands your usage of data from memory, it can pre-load the cache with data it knows you need next. You need to understand how you are allocating data and what that means (see padding)
• The sequential nature of the ring buffer allows you to introduce dependencies between processes that need something to happen before they move on (how you compose Consumers by ConsumerBarrier)
• Maximizing the use of "cache lines" (64 bytes, 8 longs of 8 bytes each, avoiding misses and "false sharing" - which aids with...

Caching

• Processors are much faster than memory now, and to optimize their performance, they use varying levels of caches (registers, store buffers, L1, L2, L3 and main memory) to support the execution of instructions, and they are kept coherent via message passing protocols
• One of the most expensive operations for a process is a cache miss - when data is looked for in one of the caches and not found, so it must go to the next level
• Data is not moved in bytes or words, but in cache lines (32-256 bytes depending on the processor, usually 64)
• If two variables are on the same cache line and written to by different threads, they present the same problems of write contention as if they were one variable ("false sharing")
• For performance, you must ensure that independent but concurrently written data are on separate cache lines
• When data is accessed from main memory in a predictable fashion (such as walking the data in a predictable "stride"), the processor can optimize by pre-fetching data it expects will be needed shortly. This ring buffer has a predicatable pattern of access
• Note that data structures such as linked lists and trees tend to have nodes that are more widely distributed (non-contiguous) in memory and therefore no predictable strides for performance optimization, which forces the processor to perform main memory direct access more often at the time the data is needed at significant performance cost

Garbage Collection

• The reuse of array buffers guarantees no GC of mature objects, thus no global GC compaction to "stop the world" - limit GC to short- and long-lived objects, not those in the middle
• Latency occurs as those objects are copied between generations, which is minimized by reusing objects so that they only traverse the generations once *GC: Restart the machine every day to clear the heap, guarantee no compaction (can go 3-4 days without worrying about it, but this is a safety measure)
• How much time do we all waste trying to find the optimal GC strategy for a system?
• What if you were able to design your system so that you didn't have to think about it, because YOU CONTROL how much GC and compaction is taking place?

Sequential Data

• The "ring" characteristics of the array buffer promote resequencing - you never stop traversing the queue (sequence number increases, use MOD by ring size to get slot)

Use cases for Disruptor

• Note that the key is BALANCED FLOW - if your flow is unbalanced, you need to weigh the cost of losing local L3 cache with the reuse of cores

RING BUFFER The Ring Buffer is a bounded, pre-allocated data structure, and the allocated data elements will exist for the life of the Disruptor instance Per Daniel Spiewak, was considered for the core data structure in Clojure before the bit-mapped vector trie was selected by Rich Hickey On most processors, there is a high cost for a remainder calculation on a sequence number which determins the slot in the ring, but it can be greatly reduced by making the ring size a power of 2; use a bit mask of ring size minus one to perform the remainder operation efficiently The data elements are merely storage for the data to be handled, not the data itself Since the data is allocated all at once on startup, it is highly likely that the memory will be contiguous in main memory and will support effective striding for the caches When an entry in the ring buffer is claimed by a producer, it copies data into one of the pre-allocated elements Sequence number % number of slots = slot in use, no pointer to end, ok to wrap managed by producer/consumer instance In most Disruptor usages, there is only one producer (network IO, file system reads, etc), which means no contention on sequence entry/allocation; if more than one producer, they can race each other for slots and use CAS on the sequence number for next available slot to use Producers copy data into the claimed element and make it public to consumers by "committing" the sequence Consumers wait for a sequence to become available in the ring buffer before they read the entry using a WaitStrategy defined in the ConsumerBarrier; note that various strategies exist for waiting, and the choice depends on the priority of CPU resource versus latency and throughput If CPU resource is more important, the consumer can wait on a condition variable protected by a lock that is signalled by a producer, which as mentioned before comes with a contention performance penalty Consumers loop, checking the cursor representing the current available sequence in the ring buffer, which can be done with or without thread yield by trading CPU resource against latency - no lock or CAS to slow it down Consumers that represent the same dependency share a ConsumerBarrier instance, but only one consumer per CB can have write access to any field in the entry SEQUENCING Basic counter for single producer, atomic int/long for multiple producers (using CAS to protect the counter) When a producer finishes copying data to a ring buffer element, it "commits" the transaction by updating a separate counter used by consumers to find out the next available data to use Consumers merely provide a BatchHandler implementation that receives callbacks when data is available for consumption Consumers can be constructed into a graph of dependencies representing multiple stages in a processing pipeline Read/Writes are minimized due to the performance cost of the volatile memory barrier BATCHING EFFECT When a consumer falls behind due to latency, it has the ability to process all ring buffer elements up to the last committed by the producer, a capability not found in queues Lagging consumers can therefore "catch up", increasing throughput and reducing/smoothing latency; near constant time for latency regardless of load, until memory subsystem is saturated, at which point the profile is linear following Little's Law Producers also batch, and can write to the point in the ring buffer where the slowest consumer is currently working Producers also have to manage a wait strategy when there are multiples of them; no "commits" to the ring buffer occur until the current sequence number is the one before the claimed slot Compared to "J" curve effect on latency observed with queues as load increases DEPENDENCY GRAPHS With a graph like model of producers and consumers (such as actors), queues are required to manage interactions between each of the elements The single ring buffer replaces this in a single data structure for all of the elements, resulting in greatly reduced fixed costs of execution, increasing throughput and reducing latency Care must be taken to ensure that state written by independent consumers doesn't result in the false sharing of cache lines

TO EXECUTE PERFORMANCE TESTS, YOU NEED A MACHINE THAT CAN EXECUTE 4 THREADS SIMULTANEOUSLY (I need a box)

EVENT SOURCING Daily snapshot and restart to clear all memory Replay events from a snapshot to see what happened when something goes awry

Links: Blog: Processing 1M TPS with Axon Framework and the Disruptor: http://blog.jteam.nl/2011/07/20/processing-1m-tps-with-axon-framework-and-the-disruptor/ QCon presentation: http://www.infoq.com/presentations/LMAX Google Group: http://groups.google.com/group/lmax-disruptor Martin Fowler's Bliki post: http://martinfowler.com/articles/lmax.html Martin Thompson's Mechanical Sympathy blog: http://mechanical-sympathy.blogspot.com/ Trisha Gee's Mechanitis Blog: http://mechanitis.blogspot.com/ Disruptor Wizard (simplifying dependency wiring): http://github.com/ajsutton/disruptorWizard My Scala port: http://github.com/jamie-allen/sdisruptor Presenting at JavaOne 2011

Disclaimer

Just like Brian, I am not a Querulous expert, and have never used it in anger.

Querulous

• Written by Matt Freels, Nick Kallen, Robey Pointer, Utkarsh Srivastava and Ed Caesar at Twitter
• Exposes SQL directly. No DSL. Sits on top of JDBC. Can call stored procs.
• Designed primarily to meet the extreme demands of FlockDB (Twitter's distributed, fault-tolerant graph database), which demands very low-latency (sub-millisecond) response times for individual queries.
• A very minimalist database querying library. Any excessive indirection would be intolerable in this environment.
• Designed for querying databases at low-latency, massive scale and with easy operability.

Major Features

• Flexible timeouts
• Extensive logging
• Rich statistics
• Extremely modular, to support DIFFERENT timeout and health check strategies based on context
• As a result, it is extremely configurable

Nick's Design Patterns of Modularity

• Must have very few assumptions hard-coded
• No concrete types!

• Ruby's open classes, method aliasing and even metaclasses do not solve this problem

• Dependency Injection

• Factories

• Decorators

• Querulous achieves modularity by providing an "injection point" for the programmer to layer on custom functionality.

• "Why I Love Everything You Hate About Java"

Querulous is MySQL-Specific

• FlockDB uses MySQL as the storage engine

• There are several generic JDBC forks on GitHub, including Rose Toomey, forked from Brendan McAdams, forked from Rhys Keepence

• I'm using Rose's querulous-generic fork for the purposes of this talk, and SQLite as the database. Olle Kullberg has another fork as well.

• Sean Rhea has a project called scalaqlite if you want to talk directly to SQLite without Querulous.

Making a Connection and Performing Queries

• All queries go through a QueryEvaluator instance

  import com.twitter.querulous.evaluator.QueryEvaluator

  trait TestConnection { val queryEvaluator = QueryEvaluator("org.sqlite.JDBC", "jdbc:sqlite:phase", "", "") }

• Basic Query Semantics - YOU write the SQL

  case class Author(id: Int, firstName: String, lastName: String) object ShowAuthors extends TestConnection { def main(args: Array[String]) { getAuthors }

    def getAuthors() {
        val users = queryEvaluator.select(
            "SELECT * FROM author WHERE id IN (?) OR first_name = ?", List(1,2), "Bill") { 
                row =>
                    new Author(row.getInt("id"), 
                        row.getString("first_name"), 
                        row.getString("last_name"))
            }
  
            users.map(author => println(author))
    }
  

  }

Transaction Support

queryEvaluator.transaction { transaction =>
    transaction.select("SELECT ... FOR UPDATE", ...)
    transaction.execute("INSERT INTO author VALUES (?, ?, ?, ?, ?, ?)", 5, "Brendan",
        "McAdams", "Michelangelo", "US", "1980")
    transaction.execute("INSERT INTO author VALUES (?, ?, ?, ?, ?, ?)", 6, "Debasish",
        "Ghosh", null, "India", "1980")
}

• Transactions will be rolled back in the case of an exception

Composable Decorators

• Advanced features can be layered on using composing QueryFactory types

  val queryFactory = new RetryingQueryFactory( new TimingOutQueryFactory(new SqlQueryFactory, 3000.millis), 5)

  val queryFactory = new DebuggingQueryFactory( new RetryingQueryFactory( new TimingOutQueryFactory(new SqlQueryFactory, 3.seconds), 5), println)

Statistics

• Use factories to get database stats

  val stats = new StatsCollector { def incr(name: String, count: Int) = Stats.incr(name, count) def timeA(f: => A): A = Stats.time(name)(f) } val databaseFactory = new StatsCollectingDatabaseFactory( new ApachePoolingDatabaseFactory(...), stats)

• To memoize DB connections and maintain connection limits (if you are dynamically connecting to dozens of hosts)

  val databaseFactory = new MemoizingDatabaseFactory(new ApachePoolingDatabaseFactory(...))

Async API

• Based on Twitter's com.twitter.util.Future, a non-Actor re-implementation of Scala Futures
• Uses the AsyncQueryEvaluator rather than the standard QueryEvaluator
• Methods immediately return values wrapped in a Future
• Blocking JDBC calls are executed within a thread pool

Async API Example

// returns Future[Seq[User]]
val future = queryEvaluator.select(
                "SELECT * FROM users WHERE id IN (?) OR name = ?", List(1,2,3), "Jacques") 
                    { row => new User(row.getInt("id"), row.getString("name"))}

// Futures support a functional, monadic interface:
val tweetsFuture = future flatMap { users =>
                        queryEvaluator.select("SELECT * FROM tweets WHERE user_id IN (?)", 
                            users.map(_.id)) { 
                                row => new Tweet(row.getInt("id"), row.getString("text"))}
}

// futures only block when unwrapped.
val tweets = tweetsFuture.apply()

External Configuration

• AutoDisabling due to failure
• Database pooling
• Retries
• Timeouts

My Impressions

Pros

• Modularity is a plus
• Some very nice composable decorators for standard behavior (timeouts, retrying, debug, etc)
• Definitely has been tested under load
• Futures support looks very cool, async is hot

Cons

• Limited support for Maven
• No support from Twitter for any database other than MySQL
• If you can't write SQL, you better learn

Additional info

• This presentation and some sample code can be found at https://github.com/scala-phase/scala-orms/tree/master/Querulous
• Querulous's GitHub repo: https://github.com/twitter/querulous
• Rose's querulous-generic GitHub repo: https://github.com/novus/querulous-generic
• Sean Rhea's scalaqlite: https://github.com/srhea/scalaqlite
• Twitter Blog: Why I Love Everything You Hate About Java: http://magicscalingsprinkles.wordpress.com/2010/02/08/why-i-love-everything-you-hate-about-java/
• No Google Groups group
• No tag on StackOverflow

Like Brian, I created this presentation with Scott Chacon's ShowOff tool. See https://github.com/schacon/showoff.