Rama on Clojure’s terms, and the magic of continuation-passing style

You can follow along with the code in this post by cloning rama-demo-gallery and opening a REPL with lein repl . Run the following to set up your REPL:

Basic example

Let’s start by defining the equivalent of Clojure’s identity function in Rama:

Here we define a “Rama operation” called identity-rama that accepts one argument named *v . Variables in Rama code are symbols beginning with * . A “Rama operation” can do everything a regular Clojure function can – conditionals, loops, define anonymous operations with lexical closures, declare locals, etc. – plus it can do much more.

In this case, the body of identity-rama “emits” the value of *v to its caller using :> . This is equivalent to the following Clojure code:

In Rama code, the continuation is implicit and is invoked by calling :> like a function. A Rama operation does not return a value to its caller. It emits values to its continuation. This is a critical distinction, as part of what makes Rama operations more general than functions is how they can emit multiple times, not emit at all, or emit asynchronously.

Note that Rama does not compile to Clojure code like this. It compiles straight to bytecode, which is necessary to achieve high performance.

Now suppose you want to call identity-rama with the value “Hello world!” and print the result. In Rama you would write this like so:

?<- is called the “execution operator” and just dynamically executes some Rama code. It’s not used in production and is just for playing at the REPL like this. Here the string “Hello world!” is passed as input to our identity-rama operation. The :> *str part binds the output of the operation to the variable *str . The :> keyword distinguishes the input from the output and is called the “default output stream” (you’ll see soon how you can have more than one output stream). The variable *str is then passed to println .

Here’s the equivalent Clojure code in CPS:

So far, you can see from this example how Rama makes things more concise by eliminating nested callback functions from the code. Here’s a slightly more complicated example to show how unreadable CPS gets when done manually:

In Clojure with CPS versions of + and * , this looks like:

In addition to how unreadable all the nesting makes this, it’s also extremely inefficient. Every single “emit” uses up another stack frame, so it seems entire programs compiled this way will quickly overflow the stack. If this is all Rama was doing, that would indeed be the case. You’ll see later some important optimizations Rama makes so that dataflow code is just as efficient as idiomatic code in any other language. The continuation being implicit in Rama rather than explicit like in the Clojure CPS examples gives Rama critical flexibility to make those optimizations.

Emitting zero or multiple times

As mentioned, you don’t have to call the continuation exactly one time. You can call it multiple times, or you can call it zero times. You can also call it asynchronously, on a different thread, or even on a different machine. This is where the expressive power of dataflow starts to show itself.

Here’s an example of a deframaop that emits multiple times along with some code that uses it:

This is equivalent to this CPS Clojure function:

Let’s now take a look at another deframaop that does filtering:

We’ll look more at conditionals later on in this post. my-filter> is the same as the built-in operation filter> and is equivalent to:

As you can see here, my-filter> emits zero values to its continuation. Emits can be done with any number of values, and in the next section you’ll see examples of emitting multiple values.

You could combine my-filter> with emit-many-times to write code like this:

You can nest expressions in Rama code just like you can in Clojure code, and the above is the same as writing:

This is equivalent to:

Running any of these prints:

This code is kind of like doing a filter on a sequence followed by a doseq , except no sequences are materialized. It also reads kind of like a WHERE clause in SQL, in that the filter is expressed solely on the value in question with an arbitrary predicate, and computation only continues with values that match.

Operations can also emit a dynamic number of times. For example, Rama has an operation in its standard library called explode that’s equivalent to this CPS Clojure function:

You could use explode to print every element of a sequence like this:

This is the same as:

Emitting multiple values in one emit

Besides being able to emit multiple times, Rama operations can emit multiple values per emit. Here’s an example:

This is equivalent to the Clojure CPS code (for brevity, without doing inc , dec , * , or / in CPS):

Something important here is the caller needs to know how many fields are expected to be given to the continuation. In Rama that’s specified by the number of variables bound to the :> output stream, and in the Clojure version that’s specified by the arity of the passed continuation function. In both cases, you’ll get a runtime error if you bind the incorrect number of continuation outputs. Whereas with a Clojure function you only have to know what arities are valid for inputs, with Rama operations you also must know the arity of the output.

Anonymous operations

Just like how you can declare anonymous functions in Clojure and pass them around as values, you can do the same in Rama with Rama operations. Like anonymous Clojure functions, anonymous Rama operations capture their lexical scope. Here’s a basic example of this:

<<ramaop defines an anonymous Rama operation with the given name, arguments, and body. Vars for anonymous operations are prefixed with % . There’s no difference in functionality between an anonymous Rama op and a top-level one. The above code is equivalent to this Clojure CPS code:

As alluded to before, Rama operations can be passed around as values. Here’s an example of passing around top-level Rama operations, anonymous Rama operations, and regular Clojure functions as values in Rama:

This prints “Result: 46”.

Emitting asynchronously

Rama operations being able to emit asynchronously is what makes Rama so good for writing parallel and asynchronous code. To demonstrate this, I’ll briefly introduce you to Rama’s cluster programming environment which implements the underlying infrastructure powering the parallel programming primitives you’re about to see. A “Rama module” is what you deploy to a Rama cluster, and it uses dataflow to define all the data ingestion, processing, and indexing for a backend. A module is launched with a configurable number of partitions called “tasks”, and these tasks run across the cluster in processes launched for the module. Dataflow code runs across all tasks in parallel and defines how to react to incoming data.

For a basic example of distributed programming with dataflow, here’s code doing a bank transfer from *from-user-id to *to-user-id in the amount of *amt dollars. This code is a stripped down version of our open-source atomic bank transfer example, and the code for reading/writing that information to durable storage is mocked out to focus on the distributed programming aspects. $$funds here refers to a durable index, similar to a database.

|hash is called a “partitioner”, and it relocates computation to a different thread/node. The only difference with the other Rama operations you’ve seen is it emits to its continuation asynchronously and potentially on a different thread/node. |hash computes the target task by modding the hash of its argument by the total number of tasks in the module. Hashing ensures the same argument always goes to the same task, while different arguments get evenly distributed across all tasks.

Computation and storage are colocated in Rama. By using partitioners to control where code is executing, you’re able to control to which partitions of durable storage you read and write. This lets you control in a fine-grained way how data is partitioned across durable storage.

What makes partitioners powerful is they’re just like any other Rama operation, and that uniformity enables composition. You can use partitioners just like any other code, such as within conditionals, loops, or helper operations. Code is read linearly without any callback functions even though you’re jumping around the cluster with impunity.

Rama’s implementation of partitioners is similar to this Clojure CPS version:

Internally, every task of a module has a queue that runs events in the order in which they arrive. The event sent in this case is the continuation, which when called continues computation where it left off. This is no different than if the emit was done synchronously.

The Rama operation definition is similar, though since we haven’t exposed manipulating continuations in Rama’s public API, the following code is only representative of what the definition looks like internally:

<<continuation defines an anonymous Rama operation just like any other, with the difference being that it emits to the caller of its parent rather than its own caller. This is just like the Clojure CPS version: when cont is eventually invoked on the other thread/node, it invokes the code following the call of |hash-clj .

Rama takes care of efficiently serializing the continuation, including any information in its closure. The Rama compiler analyzes what vars are used after every invoke of an operation, and it uses that information to only include in the closure vars that are referenced in downstream code. This minimizes the amount of information sent across the wire. This compiler analysis isn’t specific to partitioners, as its used for closure construction for all anonymous operations.

Partitioners don’t have to emit just one time at one location. Sometimes, for example, you want to run code like this:

Code like this is typical for queries that fetch and aggregate information stored across all partitions. |all partitions to all tasks in parallel, and |global always goes to the same task. |all is defined approximately like this:

And |global is defined approximately like this:

CPS and the ability to emit asynchronously unifies general purpose programming with distributed programming, by enabling parallel code to be expressed no differently than any other logic. Partitioners enable Rama code to precisely control not just what is executing, but where.

Emitting to multiple output streams

Emitting zero or multiple times, emitting multiple values, and emitting asynchronously are three ways Rama operations are more general than functions. Another is that Rama operations can emit to output streams besides :> .

Here’s an example Rama operation doing this:

This emits to three streams: :> , :a> , and :b> . Let’s take a look at an equivalent Clojure function in CPS. Instead of passing in one continuation function, we’ll now pass in a map from output stream to continuation function:

This isn’t totally accurate, as Rama does not require a caller to provide a continuation for each output stream. If there’s no continuation, then emitting to that output stream is a no-op. So the Clojure code that matches what Rama does is this:

Invoking a Rama operation that emits multiple output streams is a little bit different, as each output stream is a different code path. Here’s an example:

Rama code produces an “abstract syntax graph” (ASG), whereas Clojure (and most other languages) produce an “abstract syntax tree” (AST). <emitted-a> and <b> are called “anchors” and label part of the ASG. Those anchors are used by <<branch to specify where that code should attach. You can visualize this code like so:

:>> is called an inline hook and automatically handles setting up anchors and branching.

The above can be written with Clojure CPS like so:

All of these print:

As mentioned, you can also call emit-multiple-streams without providing continuations for every output stream. For example:

This is equivalent to:

Both of these print:

Each continuation and the code attached to it is fully executed before the subsequent line of emit-multiple-streams is run. The behavior of Rama is exactly the same as the Clojure CPS version in this respect.

Rama provides an operation called if> which is the basic primitive for specifying conditional behavior. if> takes in a value and emits to :then> or :else> depending on the truthiness of that value. The operation <<if mentioned before is a Rama macro (called “segmacro”) implemented using if> . Here’s an example of usage of if> :

This prints “False”. Unlike if in Clojure (as well as the equivalent in pretty much every other programming language), if> is not a special form in Rama. It doesn’t have to be, since it’s no different than any other Rama operation. So it can be passed around just like any Rama operation, like so:

This prints “Result: True branch”.

So far, I’ve never found a reason to pass if> around dynamically like this. What this demonstrates is how Rama’s richer language primitives provide greater uniformity and less special cases.

Also important to note is that if> produces exactly the same bytecode as Clojure’s if when invoked directly (not as an anonymous operation). Rama accomplishes this with an “intrinsic” implementation for if> in its compiler. This doesn’t change anything about semantics and is purely an optimization.

Unification

You can visualize this code like so:

unify> is a compile-time directive on how to construct the abstract syntax graph. When either :a> or :b> emit, the code continues on the shared code after the unify> . unify> can merge any number of branches, not just two.

Since the code after a unify> is shared among all its parent branches, there are rules regarding what vars are in scope after a unify> . Only vars defined in all parent branches are in scope. So if you try to reference a var after a unify> that doesn’t exist in all of the parent branches, you’ll get a compile-time error.

The above code can be written in Clojure CPS like so:

As you can see, the only way to share code across the two branches is to factor out a helper function defined before the code that executes first. The dataflow version reads much nicer since the code is ordered the same way it executes.

Loops

Like a Clojure loop, a dataflow loop has bindings along with initial values. Here the variable *a is initialized to 10. The bindings vector also binds emits from this loop to output variables which will be in scope after the loop –  *v in this case. A loop is recurred with continue> , and emits are done with :> just like emitting from a Rama operation.

This is equivalent to the following Clojure code:

Loops compose with everything else in dataflow, including partitioners. This makes it trivial to do distributed loops that hop around the cluster. Such loops are common with graph algorithms where you may be traversing from node to node fetching connections from each partition. For example:

Like before, the details of storage are mocked out to focus on the computation aspects. Here $$parents represents a datastore mapping the parent for every node. This code fetches all ancestors for a node ID by traversing the graph across the cluster in a loop.

Optimizations in dataflow compiler

When a Rama operation emits exactly one time, synchronously, and as the last thing it does, then it’s like a function. Rama provides two ways of defining operations: deframaop , as already shown, and deframafn . deframafn is just like deframaop except its implementation must synchronously emit exactly one time to the :> stream as the last thing it does.

For every invoke, Rama determines if it’s executing a deframaop or a deframafn . If it’s a deframafn , then it invokes it just like how functions are invoked in Clojure by unrolling the stack frame with the return value. For example, consider this Rama code:

The Rama execution will be very similar to this Clojure CPS:

The calls to double and println do not pass a continuation and instead do optimized invokes as functions.

It’s also worth noting that since a deframafn works just like a Clojure function, it can be invoked directly from Clojure like any Clojure function.

deframafn only has restrictions on emits to the :> output stream – it can emit to all other streams any number of times and/or asynchronously. Internally, we refer to operations like that as “semi-functions”. Rama determines whether a deframafn is a semi-function or regular function by statically analyzing whether emits are done to any other output stream beside :> . If so, invokes of that operation will unroll the stack and pass a continuation for the other output streams.

Finally, the ramafn> annotation can be used to tell Rama an anonymous operation is a ramafn .

The ramafn optimization is critical because the majority of code is still best written with function semantics. So most of a codebase will compile to stack-efficient invokes. Emitting multiples times, zero times, or asynchronously is powerful but less common.

The general term we use to refer to an object which is either a ramafn or ramaop is “fragment”. A ramafn is a fragment that has restrictions on the :> stream, while a ramaop has no restrictions.

Conclusion