Tim Bray started the WideFinder meme spreading, so I’m going to bite.
But I’m going to talk about something different. Not a language shoot-out or micro benchmark. I’ll let others talk about that. I’m going to use the WideFinder as an example to show you some principles of concurrent processes and how you can use them in your code. I’ll use Ruby here, link to a Java implementation at the end.
All the recent interest in coreness and RESTful services, it might be the right time to bring back a favorite subject of mine.
Process Calculus
This is something I got involved with around the turn of the century. I like how that sounds, turn of the century, all credible and aw inspiring, don’t you think? Sad part is, it took me a decade before I found out about pi-calculus, and only while trailing down the wrong path. I started with and recommend reading A Calculus of Mobile Processes (Milner, Parrow and Walker, 1989).
The essential features of process calculus, courtesy of the Wikipedia entry:
- Representing interactions between independent processes as communication (message-passing), rather than as the modification of shared variables
- Describing processes and systems using a small collection of primitives, and operators for combining those primitives
- Defining algebraic laws for the process operators, which allow process expressions to be manipulated using equational reasoning
Process calculus is to concurrent systems what relational algebra is to databases. And just for saying that, half of you moved on to the next post, and the rest, I can hear you snoring. But hang on, this is seriously cool stuff, especially if you like the science part of computers and like to learn new skills. I’ll do my part by showing more code than concepts.
In process calculus, we describe everything as concurrent processes interacting through message passing, with the ability to pass channels in messages. We can use that to describe protocols from TCP through HTTP all the way to a Web of services talking to each other. We can also use that to describe low-level constructs like threads, locks, semaphors and such. Also stuff that doesn’t happen in parallel.
These processes are not distributed services. They’re not operating system processes. They’re not threads. They’re a conceptual way we can describe any and all of these. So the processes I’m going to show you here deal with writing lock-free asynchronous code and letting go of shared state. It doesn’t prescribe a particular way for distributing the workload.
Why Should I Care?
I find it fascinating, but I guess you need something a bit more concrete.
When you’re building concurrent systems, you can work hard or work smart. Threads, locks and other synchronization constructs help you work hard. Smart is avoiding any shared state. Think about it as functional programming for concurrency. And it’s a great way to pull off asynchronous processing and anything you need to scale larger than a method call.
So that’s one reason.
It will also help you build distributed systems and deal with distributed fallacies. Those are all easier when you’re using messages to transfer state. Substitute process for resource and you get REST. And you know why that one is important. So there.
Pi-calculus Cliff Notes
Pi-calculus was my introduction into process calculus, so I’m going to show it briefly before moving on to code. I do recommend you spend some time reading about it in detail. My apologies to math geeks worldwide, though pi-calculus uses mathematical notation, I’ll approximate it using ASCII.
The smallest process you can have sends a message on a channel, receives a message from a channel, does something we don’t care to describe (call it, magic) or nothing at all:
P = x!y # Send value y on channel x Q = x?y # Receive on channel x, store in y 0 # Nothing
The simplest composition is just a sequence of processes, separated with dots:
P = x?y Q = y!z S = P.Q
Therefore:
S = x?y.y!z
Interesting? Not so much. So let’s add the parallel composition:
P = {x} Q | R
Q = x!y
R = x?y
The process P consists of two other processes and a new channel known to both (think of it as a scoped variable). Process Q sends a message on that channel and then reduces to nothing (0). Process R receives that message and reduces to nothing, at which point process P reduces to … I’ll let you guess. Think of reduction as single-stepping through your code.
We’re almost there, but the world is non-deterministic, so we also need to consider those cases:
P = x?y.Q | z?y.R
This just means process P either receives a message on channel x and reduces to Q, or receives a message on channel z and reduces to R. Not both. So now we have conditions.
(Wondering what happened to good old if? It will take a few minutes to think this through, Google may help, but there is a way to express if with these constructs alone)
But we still can’t reduce more than once, so let’s introduce the bang:
!P = !P | P
The bang just says that we can reduce the process any number of times, once for each message. And that we can use to implement a Web server, 8080?req, and to handle recursion (and therefore loops). Here’s an example for an infinite loop:
P = x?y.x!y
Q = {x} !P | x!0
So with inputs, outputs, conditions and recursion we can express all sort of functions (the lambda part) but also describe concurrency and distributed systems.
Code Time
So let’s see what this looks like in code. Obviously we’re working higher level than pi-calculus, we have things like variables, functions, loops, ifs, etc. We only use the process model for concurrent work.
We can start by writing a simple DSL. I did. One of my first experiences meta-programming in Ruby, ended up as a miserable failure and important lesson: abstractions are good when they help you express more, not when they limit what you can express. Here, an API will do just fine.
Sending a message to a process:
process.send *args
Receiving a message:
message = receive
This is always called inside the process.
We need conditional receives. In pi-calculus we keep the conceptual model simple, but end up with a boatload of channels. In the real world we don’t want to be in the business of tracking all these channels, so we’ll multiplex different messages on the same communication channel (one per process) and use pattern matching to tell them apart:
receive do |match|
match.when :foo do |args|
. . .
end
match.when :bar do |args|
. . .
end
end
Parallel work is easy, since fork is already in use, we’ll call it spawn instead:
foo = spawn { ... }
spawn { ..., foo, ... }
There’s no bang. We don’t need it when we can just spawn, loop and recurse. (Although, when I implented a bang method I ran into scoping issues when using it and just defaulted to spawn; perhaps there’s a better way to bang)
Speaking of recursion, have you heard of the stack limit? Let’s get around that with stack-less recursion:
tail { ... }
Now we’re ready, let’s use all of these to write a WideFinder.
WideFinder In Ruby Flavored With Pi-C
First thing I’m going to do is pull out counting and reporting into separate functions. I’m using the same code from Tim’s Ruby WideFinder, so we can get that out of the way and move to the more interesting parts:
# I map from lines to hash.
def count(lines)
lines.inject(Hash.new(0)) do |counts, line|
if line =~ %r{GET /ongoing/When/\d\d\dx/(\d\d\d\d/\d\d/\d\d/[^ .]+) }
counts[$1] += 1
end
counts
end
end
# I just output the top-ten results.
def report(counts)
puts "Results are in"
keys = counts.keys.sort { |a, b| counts[b] <=> counts[a] }
keys[0 .. 9].each do |key|
puts "#{counts[key]}: #{key}"
end
end
Next we’re going to write a process that reads the log line by line and splits it into chunks, then spawns a process to count lines in each chunk, so all the chunks counting happens in parallel.
There are so many ways to write that. I decided to pick one that has no shared state or synchronization blocks. But we do need to wait for all counters and collect the results, so we’ll add another process for dealing with that. One maps, the other reduces.
Here’s a spawn that starts a process (everything in the block) to count a chunk of lines and send the results to the collecting process:
spawn { collector.send :result, count(lines) }
I’ll do one better and insist on the process not holding any mutable state, only using messages to change from one state to the other. That’s how you build concurrent systems.
So no Object, global variables, and all local variables are immutable. There’s more overhead, but I’m not shooting for maximum performance, I’m trying to show you how to think asynchronously.
So without variables that can change state, we’ll use the oldest trick in the book and recurse. Since the Ruby stack can only take so much abuse, we’ll optimize tail recursion:
tail { split(collector, limit, source, lines + , sets) }
The process looks like this:
def split(collector, limit, source, lines = [], sets = 0)
if lines.size >= limit
# Over the limit, spawn a new process to count the lines and pass the
# results back to collector. The repeat for a new set of lines.
spawn { collector.send :result, count(lines) }
tail { split(collector, limit, source, [], sets + 1) }
elsif source.eof?
# Send whatever lines we collected so far to collector. Also tell collector
# how many sets of results we have.
spawn { collector.send :result, count(lines) }
collector.send :expecting, sets + 1
else
# Collect the new line, repeat.
tail { split(collector, limit, source, lines + , sets) }
end
end
The collector receives all the results, groups them together and prints out the report. It follows the same process-functional style.
Since messages may arrive in any order, we can’t tell the collection when we’re done. Instead, we tell it how many result sets to expect and let it figure things out. Here’s what it looks like:
def collect(counts = {}, sets = 0, expecting = nil)
if expecting == sets
# All sets are in, report and return.
report counts
else
receive do |match|
match.when :result do |_, result|
# Results from counter, combine with what we already have, and loop back.
counts = result.keys.inject(counts) { |h,k| h.merge!(k=>(h[k] || 0) + result[k]) }
tail { collect(counts, sets + 1, expecting) }
end
match.when :expecting do |_, expecting|
# If we know how many sets there are, we know when to end.
tail { collect(counts, sets, expecting) }
end
end
end
end
Now we need to tie it all together:
collector = spawn { collect }
spawn { split(collector, 10000, ARGF) }
And run:
ruby wf.rb o10k.ap => Results are in 42: 2006/09/29/Dynamic-IDE 8: 2006/07/28/Open-Data 3: 2003/07/25/NotGaming 3: 2004/04/27/RSSticker 2: 2003/09/18/NXML 2: 2004/10/01/AutumnLeaves 2: 2006/09/07/JRuby-guys 2: 2004/02/27/RSS-Unreal 2: 2003/04/10/Concorde 2: 2005/12/29/Selling-Art
Misc
The entire example is here, and the pic library here.
If Ruby is not your deal and you’d much rather use Java, have a look at Jacob. It’s a much larger framework, so a big harder to use, but as bonus it can persist state in the database and pull other cool tricks.
Labnotes » WideFinder, Ruby Pic, and Scaling Up, Out and Away