Forks and joins in Scala

A short while ago Alex Miller (aka puredanger) wrote up a blog entry detailing how to use Clojure to interact with the new fork/join concurrency framework to be included with Java7 (and already available from Doug Lea's site). The meat of Alex's solution is a Java wrapper for Clojure's IFn type which integrates nicely with the fork/join framework. At the time there was some discussion about whether a Scala implementation would require an equivalent amount of scaffolding. It turns out that supporting this functionality in Scala requires a bit more effort than one might expect, although the problems one faces are quite different than what Alex worked through in Clojure. We review the steps to a working Scala implementation and see what we can learn from them.

Before we move on, a few notes about fork/join itself. A full overview is outside the scope of this post, however details can be found in the jsr166y section of Doug Lea's site. A solid (although slightly out-of-date) review can be found in a developerWorks article by Brian Goetz.

We begin as simply as possible; a straight port of Alex's code (in this case an implementation of Fibonacci using fork/join) to Scala. We create a named function fib that will be recursively called by our RecursiveTasks. We also use an implicit conversion to map no-arg anonymous functions onto RecursiveTask instances using the technique for supporting SAM interfaces discussed in an earlier post. The resulting code looks pretty good but an attempt to build with sbt gives an unexpected error:

[info] == test-compile ==
[info]   Source analysis: 2 new/modified, 0 indirectly invalidated, 0 removed.
[info] Compiling test sources...
[error] .../src/test/scala/org/fencepost/forkjoin/ForkJoinTestFailOne.scala:23: method compute cannot be accessed in jsr166y.RecursiveTask[Int]
[error]       (f2 compute) + (f1 join)
[error]           ^
[error] one error found

A bit of investigation reveals the problem; RecursiveTask.compute() is a protected abstract method in the fork/join framework. The compute method in the class returned by our implicit conversion consists of nothing more than a call to fib.apply() but there's no way to know this when fib is defined. It follows that an attempt to access RecursiveTask.compute() from within the body of fib is (correctly) understood as an attempt to access a protected method.

How does Alex get around this problem in Clojure? He doesn't; the problem is actually resolved via the Java wrapper. Note that in his Java code compute() has been "elevated" to be a public method. He's thus able to call this method without issue from his "fib-task" function. Without this modification his code fails with similar errors [1].

Strangely, there's no obvious way to resolve this issue within Scala itself. The language provides protected and private keywords to restrict access to certain methods but there is no public keyword; methods are assumed to be public if not annotated otherwise. As a consequence there's no obvious way to "raise" the access level of a method in a subclass. We work around this constraint by way of an egregious hack; we define a new subclass of RecursiveTask containing a surrogate method that provides public access to compute. We then return this subclass from our implicit conversion.

Our new implementation now compiles, but when we try to actually execute the test with sbt we see a curious error; the test starts up and then simply hangs:

[info] == test-compile ==
[info] 
[info] == copy-test-resources ==
[info] == copy-test-resources ==
[info] 
[info] == test-start ==
[info] == test-start ==
[info] 
[info] == org.fencepost.forkjoin.ForkJoinTestFailTwo ==
[info] Test Starting: testFibonacci

Well, this isn't good. What's going on here?

The key point to understand is that the implicit conversion is called every time an existing value needs to be converted to a different type. As written the calls to f1.fork() and f1.join() both require conversion, and the implicit conversion function in play here will return a new instance on each invocation. This means that even though the input to the conversion function is identical in both cases (f1) the object that invokes fork() will be a different object than that which invokes join(). For fork/join tasks this matters; join() is called on an object that was never forked(). Voila, an unterminated wait.

We can resolve this issue one of two ways:

• Caching previous values returned by the implicit conversion function and re-using them when appropriate


• Explicitly specifying the type of f1 and f2 as our wrapper subclass. This has the effect of forcing implicit conversion at the time of assignment rather than when we call fork() and join()

I chose the first approach, mainly because it seemed a bit cooler. Oh, and it's a bit cleaner logically as well. The final result runs without issue.





[1] Using Alex's code (as published on his blog):

bartok ~/git/puredanger_forkjoin $ clojure process.clj 
(1 1 2 3 5 8 13 21 34 55 89 144 233 377 610 987 1597 2584 4181)

After changing IFnTask.compute() to protected (with no other changes):

bartok ~/git/puredanger_forkjoin $ clojure process.clj 
(Exception in thread "main" java.lang.reflect.InvocationTargetException
 at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
 at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
 at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
 at java.lang.reflect.Method.invoke(Method.java:616)
 at jline.ConsoleRunner.main(Unknown Source)
Caused by: java.lang.RuntimeException: java.lang.IllegalArgumentException: No matching field found: compute for class revelytix.federator.process.IFnTask (process.clj:0)
 at clojure.lang.Compiler.eval(Compiler.java:5440)
 at clojure.lang.Compiler.load(Compiler.java:5857)
 at clojure.lang.Compiler.loadFile(Compiler.java:5820)
 at clojure.main$load_script.invoke(main.clj:221)
 at clojure.main$script_opt.invoke(main.clj:273)

Beautiful Code 1: On Pointers in Go and Regular Expressions

Before we proceed permit me a word or two of explanation.

A few weeks ago I began working through Beautiful Code and almost immediately it became clear that this was going to be a problem. The first chapter documents an implementation of regular expressions in C, yet pondering the ideas contained within this code led to an digression on pointers and reference types in Go. That digression forms the meat of the post that follows. A quick review of the remaining chapters suggests that this probably isn't a one-time thing. This post likely represents the first of a series; thus the title. We'll see how that works out.

Thanks for listening. We now return you to your regularly scheduled blog post.

The first chapter of Oram & Wilson's work contains an analysis by Brian Kernighan (yes, that Brian Kernighan) of an elegant bit of C code. The code in question was written by Rob Pike a little over a decade ago as a simple yet useful example for The Practice of Programming. The code combines a number of C's features (strings as pointers to the first character in an array of characters, null-terminated strings, pointer arithmetic) with recursion to create a compact and elegant implementation of a simple regular expression syntax. Kernighan observes that:

"Rob's implementation itself is a superb example of beautiful code: compact, elegant, efficient and useful. It's one of the best examples of recursion that I have ever seen, and it shows the power of C pointers."

That's not hyperbole, folks.

The algorithm implemented by this code is subtle without being cryptic. In order to determine whether a given regex matches a string s with length n and consisting of characters s(0) through s(n - 1) we compare the regex to every substring of s that also ends in s(n - 1). C uses null-terminated character arrays to represent strings and a pointer to the first character of the array to refer to the string so we can enumerate all relevant substrings of s simply by incrementing the pointer to the first character via pointer arithmetic and recursively calling our comparison function with every substring generated by this process.

Go also supports pointers in order to provide call-by-reference semantics so in theory the same algorithm could be implemented in this language as well. Attempting to do so seemed like an interesting experiment, not least because Rob is somewhat involved with the Go language. Unfortunately according to the specification Go does not allow pointers to refer to individual characters in a string and it isn't clear whether pointer arithmetic is supported in any capacity [1]. All is not lost, however; Go does include a reference type for arrays and their sub-arrays in the form of slices. Using slices we can perform the processing described above in an (arguably) more explicit syntax. The results can be found on github.

A curious side effect: Python also supports slices on lists so in the end our implementation looks quite a bit like the Python implementation of this algorithm.

[1] If you don't support referencing individual array elements a lot of the arguments for pointer arithmetic don't make quite as much sense.

Transmutation of Scala closures into SAM instances

Okay, it's not actually alchemy. But it is pretty cool nonetheless.

A brief refresher on terminology: a single abstract method (or SAM) interface or abstract method contains exactly one method that concrete implementations must define. Java includes a number of interfaces that satisfy this property, including Runnable, Callable and Comparable.

The basic idea at play here is that a closure or anonymous function (either will do here) can be used in place of a SAM implementation if the parameters and return type of the closure match up to those of the method. That assumes your language cares about such things, of course; duck typing makes this less of an issue (as we'll see).

JRuby automatically performs this operation via closure conversion (scroll down to the bottom of the page). Stuart Sierra has recently published a macro for doing something similar in Clojure. Even Java is considering including this feature in an eventual closure implementation (see Brian Goetz's writeup for details). Why should Scala miss out on all the fun?

Let's take a look at some code to make this discussion more tangible. A simple example of closure conversion in JRuby can be found here. This is where duck typing helps us out; the only real requirement on our closure is that we actually return something (in order to clearly distinguish between Runnable and Callable). Our objective is to implement a Scala unit test that does something similar. Any such approach will be built around Scala's support for implicit conversion of types, but in this case a bit of care and feeding is required to line up the parameters and return types of the closure with that of the contents of the SAM interface. The basic approach works as follows:

1. The implicit conversion accepts a function with the same parameter list and return value as the lone method in the SAM interface


2. The conversion then returns a new concrete instance of the SAM interface. The implementation of the method doesn't need to be anything other than invoking apply() on the input function

The resulting ScalaTest class can be found here.

base64 in Clojure: Once more, with feeling

In a previous post we sketched out a simple base64 implementation in Clojure and measured it's performance against clojure-contrib and commons-codec. Our initial implementation used the partition function to split the input into blocks and then applied a sequence of interdependent functions to translate each block into base64-encoded data. This approach worked well enough but it's not terribly efficient; it traverses the input data once to do the partitioning and then that same data is traversed again (in block form) when the encoding function is applied. There's no reason to make multiple passes if we attack the problem recursively rather than iteratively. Time for some further investigation.

Clojure optimizes recursive calls when the loop/recur constructs are used so we deploy them here. In addition we make significant use of destructuring. Each input chunk of data is routed to the appropriate function via a sequence of cond expressions. These functions are implemented as let statements which break the chunk into bytes (via destructuring) and then create bindings for the base64 characters representing these bytes by applying the relevant bit manipulations. The body of the let doesn't need to do much beyond combining these character bindings into an appropriate return value.

This refactoring offers the following benefits:

• The code is a lot cleaner and much easier to read and understand


• Using the same techniques we should be able to easily implement a decoding function


• We should see some increase in performance

So, how'd we do? The following is fairly representative:

bartok ~/git/base64-clojure $ clojure fencepost/test/compare_base64.clj 
Encoding
Commons-codec
"Elapsed time: 85.219348 msecs"
Xm91U1RMbHRAK1ZCcWFqeEsmJEJrfkhIdSsxfStJdGBNc2txUTBkUyQrdCM7Wk9rInN2XFBtKGNTX002aw==
clojure-contrib
"Elapsed time: 896.538841 msecs"
Xm91U1RMbHRAK1ZCcWFqeEsmJEJrfkhIdSsxfStJdGBNc2txUTBkUyQrdCM7Wk9rInN2XFBtKGNTX002aw==
fencepost/base64_letfn
"Elapsed time: 315.582591 msecs"
Xm91U1RMbHRAK1ZCcWFqeEsmJEJrfkhIdSsxfStJdGBNc2txUTBkUyQrdCM7Wk9rInN2XFBtKGNTX002aw==
fencepost-recur
"Elapsed time: 215.516695 msecs"
Xm91U1RMbHRAK1ZCcWFqeEsmJEJrfkhIdSsxfStJdGBNc2txUTBkUyQrdCM7Wk9rInN2XFBtKGNTX002aw==

The recursive encoding operation is just north of a third faster. We still take about twice as long as commons-codec (an improvement on our previous performance) but we're now over four times faster than clojure-contrib.

As referenced above a decoding function was also completed using the same techniques described for the recursive encoding function. The performance increase for that operation was just as stark:

Decoding
Commons-codec
"Elapsed time: 33.148648 msecs"
^ouSTLlt@+VBqajxK&$Bk~HHu+1}+It`MskqQ0dS$+t#;ZOk"sv\Pm(cS_M6k
clojure-contrib
"Elapsed time: 1494.2707 msecs"
^ouSTLlt@+VBqajxK&$Bk~HHu+1}+It`MskqQ0dS$+t#;ZOk"sv\Pm(cS_M6k
fencepost-recur
"Elapsed time: 268.112751 msecs"
^ouSTLlt@+VBqajxK&$Bk~HHu+1}+It`MskqQ0dS$+t#;ZOk"sv\Pm(cS_M6k

Still nowhere near as fast as commons-codec but we're a bit better than five times faster than clojure-contrib.

As always code can be found on github.

A Funny Thing Happened While Writing Some Haskell....

Suppose that you've just finished Graham Hutton's solid introduction to Haskell and now want to take the language for a quick test drive. You decide to stick with something familiar; after noting that the Prelude doesn't have an equivalent to Python's range function you decide to whip one up yourself. Haskell's syntax is still a bit unfamiliar, however, so you decide to implement this function in a different "mostly functional" language that you have some experience with. You've been looking for a reason to dip your toes into Scheme again (and it seems to be a good fit here [1]) so you install Gambit Scheme and Chicken Scheme and dig in.

Perhaps you begin with a straightforward recursive definition in Scheme:

(define (range start stop step)
 (if (< start stop)
  (cons start (range (+ start step) stop step))
  '()))

This works well enough when start < stop but fails completely when start > stop and step is a negative value (a use case supported by Python's range()). To fix this problem we define two procedures via letrec and use whichever is appropriate based on the passed arguments:

(define (newrange start stop step)
  (letrec 
   ((upto (lambda (art op ep) 
     (if (< art op)
      (cons art (upto (+ art ep) op ep)) 
      '())))
    (downto (lambda (art op ep)
     (if (> art op)
      (cons art (downto (+ art ep) op ep)) 
      '()))))
 (if (< start stop)
  (upto start stop step)
  (downto start stop step))))

This function now works for both start < stop (with positive step) and start > stop (with negative step). There's still a problem, though; if we change the sign of step in either of these cases we find ourselves in an infinite loop. Python's range() returns an empty list in this case so we probably should too:

(define (newerrange start stop step)
 (letrec
  ((upto (lambda (art op ep)
    (if (< art op)
     (cons art (upto (+ art ep) op ep)) 
     '())))
   (downto (lambda (art op ep)
    (if (> art op)
     (cons art (downto (+ art ep) op ep)) 
     '()))))
    (cond ((and (< start stop) (< step 0)) '())
          ((and (> start stop) (> step 0)) '())
          ((< start stop) (upto start stop step))
          (else (downto start stop step)))))

While working on these implementations you discover that SRFI-1 includes a function iota which looks similar to Python's range(), although this function takes the number of elements in the return list rather than the start/stop/step values we're looking for. Still, both Chicken Scheme and Gambit Scheme support SFRI-1 as extensions or modules and we should be able to whip up a thin wrapper around iota to get what we need. Of course we first need to figure out how to load up these modules. And after some initial experimentation we see some curious behaviour in the iota implementation on both platforms...

But wait... weren't we supposed to be writing some Haskell code?

Oh, yeah.

Haskell's guarded equations allow for an easy expression of the same logic we were getting from cons in Scheme. We have no need to translate the recursive logic found in the Scheme version, however, since the list-related functions in the Prelude combined with a few sections gives us everything we need:

range art op ep | art < op && ep < 0 = []
                | art > op && ep > 0 = []
                | art < op = takeWhile (<op) (iterate (+ep) art)
                | art > op = takeWhile (>op) (iterate (+ep) art)

In all honesty this is a fairly elegant expression of the algorithm. And while there's clearly a lot to like in Haskell this exercise has rekindled a long-dormant romance with Scheme rather than encouraging exploration of something new.

[1] Yes, I realize Scheme isn't "purely functional". I'm not interested in taking a stand in this particular holy war, but there's little doubt that Scheme's bias against set! and it's kin justify the "mostly functional" qualifier.

Making Hadoop more "functional" with Scala

Introduction

After spending more and more time working with Scala I found myself attempting to apply the language to more and more domains. One candidate was a long-standing issue with the mechanism used by Hadoop to express map, reduce and/or combine functions. These functions are expressed within the framework via a pair of interfaces, Mapper and Reducer. Jobs then add class references to concrete implementations of these interfaces.

The problem

So what's the problem? These interfaces largely exist to define a single function to perform either a map or reduce operation. Both Mapper and Reducer include a setup() and cleanup() method (as well as a run() method to change runtime behaviour if needed) but the heavy lifting for both is performed within a single method (map() and reduce() respectively). Strip away the overhead and it's clear that in a fair number of cases we really only need an anonymous block of code to implement our map or reduce functions. At least some of the setup work could be removed if this anonymous block of code could somehow "contain" the environment that was present when the function was defined.

It should be obvious that what's being described here is a closure. Scala supports closures, but so does Groovy, JRuby and several other JVM languages. Why prefer Scala to these options? It's all a matter of typing; my belief was that the typing of closures in Scala would provide easier integration with Hadoop. This assumption may prove to be incorrect in the long run (Clojure's strong Java integration might make it a better fit for example) but it does provide a starting point.

Note that defining these functions inline would prevent the kind of re-use provided by defining this functionality in distinct classes. This is only a concern if the map and/or reduce functions are sufficiently complex; if they reduce down to something in the neighbourhood of a one-liner then re-use isn't a large concern.

Finally, it's worth noting that the problem here does not stem from Hadoop itself. Implementing this functionality in discrete classes is a reasonable approach when the language doesn't provide a clear mechanism for expressing anonymous functions.

A sample

To make this more concrete let's analyze the "grep" sample job referenced in the document describing a single-node Hadoop setup. The main work is performed by a single job using a regex-based map implementation and a reducer which does little more than sum up long values. The meat of the mapping class is as follows:

org.apache.hadoop.mapreduce.lib.map.RegexMapper

public class RegexMapper extends Mapper {

  public static String PATTERN = "mapreduce.mapper.regex";
  public static String GROUP = "mapreduce.mapper.regexmapper..group";
  private Pattern pattern;
  private int group;

  public void setup(Context context) {
    Configuration conf = context.getConfiguration();
    pattern = Pattern.compile(conf.get(PATTERN));
    group = conf.getInt(GROUP, 0);
  }

  public void map(K key, Text value,
                  Context context)
    throws IOException, InterruptedException {
    String text = value.toString();
    Matcher matcher = pattern.matcher(text);
    while (matcher.find()) {
      context.write(new Text(matcher.group(group)), new LongWritable(1));
    }
  }
}

The reducer implementation doesn't fare much better:


org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer

public class LongSumReducer extends Reducer                                                 KEY,LongWritable> {

  private LongWritable result = new LongWritable();

  public void reduce(KEY key, Iterable values,
                     Context context) throws IOException, InterruptedException {
    long sum = 0;
    for (LongWritable val : values) {
      sum += val.get();
    }
    result.set(sum);
    context.write(key, result);
  }

}

That's a lot of boilerplate code in order to implement a pair of very simple functions. Surely we can do better with a language that offers native support for many elements of functional programming.

An initial implementation

As a first step we convert the "grep" example into Scala and run it using Hadoop 0.21. The conversion itself was fairly straightforward and the results can be found on github. At present the port isn't much more than a translation of the existing Java code into more idiomatic Scala; the hope is that this code will provide a basis for future work. The code has been verified to run using the following steps:

1. Add the Scala runtime to the Hadoop classpath by editing hadoop-env.sh:
   export HADOOP_CLASSPATH=$HOME/local/scala/lib/scala-library.jar


2. Replace org.apache.hadoop.examples.Grep in mapred-examples with our Scala implementation:
   jar uvf hadoop-mapred-examples-0.21.0.jar org/apache/hadoop/examples/Grep*
   
3. bin/hdfs namenode -format


4. bin/start-dfs.sh


5. bin/start-mapred.sh


6. bin/hadoop fs -put conf input


7. bin/hadoop jar hadoop-mapred-examples-0.21.0.jar grep input output-scala 'dfs[a-z.]+'


8. bin/hadoop fs -cat output-scala/*

Future work

In order to get the most out of writing Hadoop jobs in Scala we need to be able to define these functions directly within the job class. This is complicated somewhat by the current Hadoop architecture; jobs are configured to reference mapper and/or reducer classes rather than instances of those classes. Something akin to Groovy's support for using a closure as an implementation of a class with a Single Abstract Method (SAM) might be useful here, but near as I can tell Scala does not offer such support.

base64 in Clojure: An initial attempt

After working through most of Stuart Halloway's excellent Programming Clojure I found myself in search of a meaty example to exercise my Clojure chops. Re-implementing the Project Euler examples in another language didn't seem terribly appealing. Maybe a base64 implementation; I hadn't implemented the encoding in a while and it is a reasonable exercise when learning a new language. So what kind of support does Clojure already have?

As expected there's no base64 support in the core. clojure-contrib does include some support, although the documentation doesn't inspire a great deal of confidence:

This is mainly here as an example.  It is much slower
 than the Apache Commons Codec implementation or 
sun.misc.BASE64Encoder.

Intriguing; looks like we have a winner!

The only complexity in a base64 implementation is handling the one and two byte cases. Any code must support returning the last two bits of the first byte (or the last four bits of the second byte) as distinct values in these cases as well as combining these bits with the leading bits of the second and third bytes (respectively) in other cases. My initial implementation was a fairly straightforward adaptation of this principle. The core consists of a letfn definition with functions for handling the one, two and three byte cases. Each function returns two values: a list of all 6-bit elements discovered so far as well a "remainder" value representing the trailing bits in the one and two byte cases. This structure leads to better function re-use; as an example the two-byte function can build on the one-byte function. A wrapper function calls the correct value for a given set of bytes. Now we need only partition our byte array into groups of three or less and apply the wrapper function to each group. A combination of the partition and map functions do exactly that.

So how'd we do? We implement a simple test program to generate a set of random data, map a base64 implementation on to each sample and then realize the lazy sequence by converting it into a vector. We time this conversion (for a common set of sample data) for our implementation, the clojure-contrib implementation and commons-codec. The following run is fairly representative:

bartok ~/git/base64-clojure $ source set_classpath.sh 
bartok ~/git/base64-clojure $ clojure fencepost/test/compare_base64.clj 
Commons-codec
"Elapsed time: 69.074926 msecs"
RFJlYWkqIF10RCc1PXlpO3FcIDY6dkJuQkhHNWxRJkwkOFdgciFGfHwhYz5EYG15PjxWdFJ9Ml83TGFoeltHSTs+ST9mdj0rfSZrcVNIKn5oKSdTI3U5a1FqIzBvIkRIV0BkK3xCZjtrSWYuTiM1XWB9UW14W2dVLFM=
clojure-contrib
"Elapsed time: 1221.124746 msecs"
RFJlYWkqIF10RCc1PXlpO3FcIDY6dkJuQkhHNWxRJkwkOFdgciFGfHwhYz5EYG15PjxWdFJ9Ml83TGFoeltHSTs+ST9mdj0rfSZrcVNIKn5oKSdTI3U5a1FqIzBvIkRIV0BkK3xCZjtrSWYuTiM1XWB9UW14W2dVLFM=
fencepost/base64_letfn
"Elapsed time: 375.829153 msecs"
RFJlYWkqIF10RCc1PXlpO3FcIDY6dkJuQkhHNWxRJkwkOFdgciFGfHwhYz5EYG15PjxWdFJ9Ml83TGFoeltHSTs+ST9mdj0rfSZrcVNIKn5oKSdTI3U5a1FqIzBvIkRIV0BkK3xCZjtrSWYuTiM1XWB9UW14W2dVLFM=

These results do confirm that that clojure-contrib implementation is very, very slow in relation to commons-codec. We've reduced that runtime by roughly a third but we're still five or six times slower than commons-codec. Surely we can do better; in the next installment we'll see how to do so.

As usual all code can be found on github.