Adventures in Widefinding: Performance

I've done some basic performance tests against:

1. Scala Actor-based parallel Widefinder
2. Scala serial Widefinder using a BufferedReader
3. Tim Bray's Ruby Widefinder

The test platform was my 2.2 GHz MacBook with 4GB of RAM using a 6 million line file.  The times were as follows:

Scala Parallel:

• real 0m14.588s
• user 0m24.541s
• sys 0m1.383s

Scala Serial:

• real 0m20.095s
• user 0m18.821s
• sys 0m1.441s

Ruby:

• real 0m14.301s
• user 0m12.485s
• sys 0m1.813s

The good news is that the parallel Scala version is noticeably faster than the serial version.  The bad news is that it is roughly the same speed as the Ruby version, and takes significantly more CPU.  The Scala versions are doing substantially more work, because they have to transcode the 8-bit ASCII contained in the input file into 16-bit Unicode strings.  This requires a full scan of the data.  I believe a fair amount of performance could be gained by combining the transcoding pass over the input with the line-splitting pass.

For those that are curious, the source code for the parallel widefinder is available here: the parallel IO code the actual widefinder code

Sphere: Related Content

Adventures in Widefinding: Complexity

For the moment I'm going to ignore the complexity of my widefinder "under the hood," and just focus on the complexity of it at the top level.   The following is Tim Bray's goal.  The red asterisk indicates that the loop should be processed in parallel instead of in serial.

counts = {}
counts.default = 0

ARGF.each_line* do |line|
if line =~ %r{GET /ongoing/When/\d\d\dx/(\d\d\d\d/\d\d/\d\d/[^ .]+) }
counts[$1] += 1
end
end

keys_by_count = counts.keys.sort { |a, b| counts[b] <=> counts[a] }
keys_by_count[0 .. 9].each do |key|
puts "#{counts[key]}: #{key}"
end

So here's the first part of my Scala solution:

   val pat = Pattern.compile("GET /ongoing/When/\\d\\d\\dx/(\\d\\d\\d\\d/\\d\\d/\\d\\d/[^ .]+ )")
   val map = new ConcurrentHashMap[String, AtomicInteger]()
   for (file <- args; line <- LineReader(file)) {
       val mat = pat.matcher(line)
       if (mat.find()) {
         val s = mat.group().substring(23)
         map.putIfAbsent(s, new AtomicInteger(0))
         map.get(s).incrementAndGet()
       }
     }
 

Notice that this is not a map-reduce based solution. This is because I tried to make it look as similar to the original Ruby as possible. There are some minor differences due to the fact that regular expressions are not special in Scala the way they are in Ruby. The big, and in my opinion most intrusive difference, is that a ConcurrentHashMap has to be used to store the results, AtomicIntegers must be used for the counts, and that weird "putIfAbsent" call needs to be used for pulling counts out of the map.

So while the concurrency itself is implicit, and the code is certainly succinct, the programmer still needs to be very much aware of the fact that the block inside the for loop will be executed in a concurrent fashion. This is certainly less elegant than a pure functional approach such as map-reduce, where no synchronization would be necessary, but I also think it is less intrusive to a programmer accustomed to imperative programming.

Here's an ugly part. This is largely due to an impedance mismatch between Java collections and Scala collections. It would look a lot better if Scala had an equivalent to ConcurrentHashMap in its library. There's also probably lots of other ways to make it cleaner. I tried several different approaches and always ran into some problem.

    val array = new Array[(String, Int)](map.size)
    val iter = map.entrySet.iterator
    var i = 0
    while (iter.hasNext) {
      val e = iter.next
      array(i) = (e.getKey, e.getValue.intValue)
      i += 1
    }

And here's the final part, once the ugliness is done:

    quickSort(array)((e) => OrderedTuple(e))
    val limit = if (array.length < 10) array.length - 1 else 9
    for(i <- 0 to limit) {
      print(array(i)._2)
      print(": ")
      println(array(i)._1)
    }

Notice having less that ten results is cleanly handled, unlike in TB's code (unless Ruby magically ignores invalid indices).

Overall I would say I came pretty close to TB's goal without without requiring a fundamental change in programming methodology. I didn't alleviate the need for the programmer to be aware of concurrency, but I don't think that's possible. Sure, people can learn to program in ways that are fundamentally parallelizable, and that probably is what universities should start teaching, but that's going to require a major shift on of part of millions of programmers, or for several generations to retire and be replaced by ones trained in functional programming. Lots of cores are here NOW, so it's not practical to wait until skills catch up. We need to work with what we have.

Sphere: Related Content

Adventures in Widefinding

Over the past few weeks in my copiuos free time I've been trying to come up with a fast Widefinder using Scala Actors where the parallelization magic is hidden behind what looks like a nice friendly API. Based on timings on my oh-so-fast Athlon XP 1900+ I seem to have failed on the first account (Ruby is faster) and the second part requires some awareness of parallel programming. But I'm hopeful that when Tim runs it on the T5120 it will be competive with the other solutions, although I doubt that it will be the fastest. For those of you who don't want to listen to me ramble, here's a jar containing all of the class files needed to run my solution along with the Scala source code. It's so big because I put the Scala libraries in there for Tim so he wouldn't have to worry about CLASSPATH error or installing Scala. Ok, so why isn't it the fastest? What's the point? Well, first let me define what I think a good widefinder solution should demonstrate.

1. The "application code" should be similar in complexity and readability to Tim Bray's example in Ruby. Anything else should be hidden behind a general purpose API, preferably an out-of-the-box one.
2. It should be environmentally friendly. This means that it shouldn't hog all the available RAM and make other processes start paging. It should also be well behaved on a machine under load. This is not relative to input file size. It should ALWAYS be environmentally friendly.
3. It should maintain the abstractions to which programmers are accustomed. For example, it shouldn't bypass regexes or use non-standard string types. Also, if a language normally deals well with unicode and the potential for multiple encodings, it should deal well with unicode and multiple encodings as well.
4. The person running it shouldn't have to tell it about details like how many worker processes to spawn. It should figure it out based on its environment. Ideally the programmer shouldn't have to worry about it either.
5. It should be sufficiently performant on a single processor box.
6. It should run faster as more processors are made available.

I think my solution is reasonable according to all of these measure. Hopefully we'll see about execution on lots of cores shortly. #1 is a little tricky, because the using a concurrent map was tricky and there is a definite impedance mismatch between Java collections and idiomatic Scala. Depending on your perspective a map-reduce based solution would have been more elegant, but unless map-reduce becomes required learning for all professional programmers then I'm not so sure. #2 is one area where, in my opinion, many of the current leading solutions fall down. A favorite way of achieving easy parallel IO (well, IO on most disk can't really be done in parallel) is to spawn a bunch of processes (or threads) and have them each mmap a large chunk of the input file and process it. The issue with this is that, as Fredrik Lundh noticed (BTW - he has an excellent discussion - go read it), this can cause serious paging on some operating systems. Other solutions allocate buffers measured in megabytes for each worker thread/process. I personally don't see this as very environmentally friendly when only kilobytes are really needed. My solution reads the file sequentially using one set of buffers per worker (about 128k total per worker). Each worker reads in 32k and then passes the FileChannel to the next worker. #3 is the real kicker, and what kicked down the performance of Scala (and, I believe, any other JVM based solution). Java strings are unicode and based on 16 bit characters. The input file is ASCII and based on 8 bit characters. Consequently, a String-oriented solution on the JVM has to transcode the characters. Futhermore, it should be able to transcode the characters from any common file format, not just ASCII or UTF8. This turns out to take a lot of cycles, around 10% depending on the charset and the cost of some other functions. The good news is that each buffer can be transcoded in parallel. #4 is more about maintaining abstractions. My solution does it automatically, the other solutions don't seem to...although adding it probably wouldn't be a big deal. #5 is probably true of most of the solutions...or true if you substitute dual-core for uniprocessor, because that's what most of the people used to develop their solutions. In the case of my solution I think there is about a 30% performance penalty on a single processor box (that and transcoding are enough to make it slower than Ruby). #6 is interesting because this is an I/O bound task. Assuming the data is actually being read from disk (I believe it is cached in all or most of his trials), then this task should only be able to keep a small handful of processors busy. But that's not to say it isn't important. The reason it's important is because from the normal programmer's point of view, file I/O isn't just shuttling information from the disk to a buffer, it's everything that happens between him making the request and a string representing a line being returned. There's a lot of work that goes into making that string. Here's a list (more or less) for Java and Scala.

1. Read the data into a direct ByteBuffer.
2. Copy the data into a heap ByteBuffer**
3. Transcode the ByteBuffer into a CharBuffer
4. Find the characters that represent a line
5. Make a String from those characters

Your operating system probably uses something called readahead to asynchronously fetch the next N bytes of a file before you actually request it, on the assumption that you will. The Java or Scala library could asynchrously fill a fixed-sized queue of lines, assuming concurrency is cheap enough. I learned a lot in this excersise, but my two big takeways are:

1. Abstraction has a real cost in terms of CPU cycles, and existing abstractions in mainstream languages (ok, Scala isn't mainstream yet, but Java is) are not optimized for multi-core environments. This will need to change.
2. Abstraction has a big cost in terms of programmer cycles if the programmer wants something different. Without the source available from OpenJDK I would have been lost when I was trying to figure out why approaches that seem like they require less CPU work actually require more. The Client and Server VMs make a huge difference in performance that varies widely depending on file size and program structure. With the libraries, JVM, and OS all trying different things to make your code go faster it is extremely difficult to figure out exactly what is happening and why.

**I looked at the OpenJDK source and if you use NIO to read into a non-direct buffer, what actually happens is the Java library reads into an available direct buffer and then copies it into a direct buffer. This is essentially what happens with a regular FileInputStream as well. At first I thought I could speed things up by skipping the copy from the direct buffer to the non-direct buffer, but it doesn't. You see, decoders are optimized to use the array backing the buffer if it is available because direct array accesses are much faster than function calls. Array copies are done using low-level native code. Consequently, copying the direct buffer and the transcoding the in-heap copy is much faster than directly transcoding the direct buffer.

Sphere: Related Content

Why test parallelism on a simple function?

On my last blog anonymous asked:

would a more expensive line-match-function make it more obvious if you are working in parallel?

I would say that one should be able to demonstrate that transparently supporting the potential for parallelism should be near free. If you can use a parallel algorithm to solve a problem that doesn't benefit much from parallelism with roughly the same or better performance characteristics as the serial code then it should be a lot better when you actually give it a more complex problem. Basically, parallelism should be free. A lot of people have commented on Tim Bray's blogs that his test is unsuited for demonstrating the benefits of parallelism because it is IO bound. Tim claims this isn't true, and I suspect there's some truth to that if you have really optimized IO, but I do think the benefits of parallelization for him problem are very limited. That being said, one thing that it he has clearly demonstrated is that parallelism isn't free. His "obvious" newbie solution in Erlang performed horribly and was considerably longer than the Ruby solution. Others have greatly improved the performance with extremely long, complicated chunks of code, but have yet to match Ruby. I find that really sad. So I would like to prove that parallelism can be almost free, meaning:

1. Leveraging it does not impose a significant additional cognitive load on the programmer.
2. Problems that are not effectively parallelizable should execute "about as fast" when parallel functionality is used as with serial code.

From an interface perspective I think I have it with the monadic interface to the file. I just need to work out some bugs or change the interface to make them go away. I'll write more on this when I've worked out some of the wrinkles. So that leaves the performance problem. One of the big challenges with parallelization is that spawning new threads or processes is very expensive, and synchronization is somewhat expensive, so it's very easy for the cost of parallelization to overwhelm the cost of the actual solution. The most straight forward way to address this problem is to not parallelize when the function is not complex enough or the input data set isn't large enough to justify it, but that is back to imposing a cognitive load on the programmer because he has to figure that out. Either that or always "start serial" and use runtime profiling tricks to detect if the problem is worth parallelizing, which sounds expensive put probably has merit. Another challenge is knowing how to divide up the problem to avoid excessive synchronization and/or messaging. When processing a file line-by-line, one could send each line out to be processed independently, but that requires a lot of messaging and synchronization if you don't have lock-free messaging. So really you want to break the problem into properly sized chunks and send each chunk as a message rather than simply use the most natural division. Figuring out how big a chunk should be (or how many chunks you should have) is a challenge because it is problem and runtime dependent. Again, this creates the potential to burden the programmer, use complex and potentially expensive runtime profiling, or somehow come up with a magic cheap hueristic. So you can either solve the problems above, or you can have sufficiently cheap parallelism that you don't need good solutions. Right now I'm going after the sufficiently cheap approach. What I have so far is a mapreduce-style function using Scala Actors that breaks a file into chunks of lines and sends them off to be processed by an Actors. I plan on adding a parallel foreach function that could be used for problems like Widefinder using a parallel hash map. Performance wise it's looking promising. Here's some numbers (using my 5+ year old machine): Serial:

Count: 185300 Serial: 11592 real 0m12.107s user 0m11.254s sys 0m0.784s

Parallel:

Count: 185300 Serial: 11722 real 0m12.225s user 0m11.441s sys 0m0.723s

As you can see the parallel code is slightly slower than the serial code. Across runs their times actually overlap a bit, but serial generally times to be a tad faster. One thing I've noticed is that the deltas between the serial and parallel implementations don't really grow - and to some extent shrink - with increasing input sizes. I believe this is because there is a fixed penalty for setting up the thread pool for the actors. This only has to be done once per process invokation, and appears to cost about 200ms on my machine. In other words, parallelization for file processing can be almost free. I actually think it could be better-than-free, even on a single processor box, if IO was more efficient. My current solution is using a BufferedReader to read in the file one line at a time. This means the IO is probably being done in a less-than-optimal way, and that a lot of work is being done in serial for each line (converting from 8bit ASCII to 16-bit Unicode strings, splitting it into lines). I'd like to use nio to read the file in a block at a time, and then let all this work be done in separate threads. I think then there would be a performance increase because one thread would be doing nothing but reading in buffers as fast as the OS and JVM and provide them, and others would be doing all the computation while the IO thread is blocking. But before that I'm going to get the interface cleaned up and solve the memory problem on large files.

Sphere: Related Content