17
$\begingroup$

Many modern languages, such as Haskell, feature lazy lists, which can model infinite sequences of values, albeit synchronously. Lately, the reactive programming movement has shown that asynchronous streams have the same structure as lists and streams, suggesting that the usual, functional operators like Map, Take, and, most importantly, Fold (see this paper by Graham Hutton about why Fold is extra important), can be written for streams, synchronous and asynchronous, in Mathematica. How should one go about doing that?

$\endgroup$
  • 3
    $\begingroup$ Rumor has it that we might see official support for lazy lists soon. Also related: (838). $\endgroup$ – WReach Apr 13 '15 at 15:00
  • $\begingroup$ I got a 404 on the Hutton paper link. $\endgroup$ – alancalvitti Sep 18 '15 at 13:28
  • $\begingroup$ @alancalvitti updated the link give it another try, please, or just Google for "graham hutton fold tutorial" the paper is easy to find $\endgroup$ – Reb.Cabin Sep 19 '15 at 14:51
16
$\begingroup$

Download a version of this post as a Mathematica notebook by evaluating the following expression in a fresh notebook:

Import["http://goo.gl/NaH6rM"]["http://i.stack.imgur.com/Yfan3.png"]

Folds over Lists, Lazy Lists (Iteractive) and Tasks (Reactive)

We'd like to have a symmetric set of operations over Lists, lazy lists (streams, iterables, or iteractive [sic] collections) and over asynchronous tasks (observables or reactive collections). Lists already have a rich set of operations, like Map, Scan, Take, Fold, FoldList, Drop, Append, and so on. For the sake of brevity, we'll work up examples just with Take and FoldList, but it's possible to build the others, too. Both Take and FoldList convert Lists---and some other arguments---into Lists, so they have the nice property of being composable within a "domain" of Lists.

We want similar operators for streams and observables. The following figure illustrates. Consider three domains: Observables, Lists, and Streams. We want a "take" and a "fold" for each domain. We also want a "way out" of each domain: an operation that yields individual values. For simplicity, we'll consider "ways-out" that work by side effect. The pertinent operator for Lists is Scan. We posit subscribe for observables and forEach for streams. These should have the same signatures as Scan, taking a collection and a unary function that operates on each element of the corresponding collections. We also would like ways to and from Lists for streams and observables. We posit reify and disperse for streams and harvest and dispense for observables.

Mathematica graphics

Lists

In general, a fold takes three arguments: a binary function, a starting value, and a collection of other values. There are versions that produce final values; there are versions that produce new collections. As said, we want the latter because they compose easily (and composition is good). For Mathematica Lists, this version is FoldList. Using Haskell-style functional type notation, the type of FoldList is as follows:

FoldList :: binaryFunction --> startingValue --> collection --> collection

Example: Fibonacci

This version of Fibonacci doesn't actually use the values in the third argument, the collection---it just needs them to be there.

FoldList[
 (* binary function *)
 {acc, v} |-> {acc[[2]], acc[[1]] + acc[[2]]},
 (* starting value *)
 {0, 1},
 (* collection of other values *)
 Range[0, 9]]
{{0, 1}, {1, 1}, {1, 2}, {2, 3}, {3, 5}, {5, 8}, {8, 13}, {13, 21}, {21, 34}, {34, 55}, {55, 89}}

This produces a new collection from the given collection. The new collection has one extra element, because the fold produces the starting value if the input collection is empty.

FoldList cannot produce an empty list from an empty list, no matter what the function produces.

FoldList[{a, b} |-> Null, 42, {}]
{42}

Streams

Let's make folds work on potentially infinite, lazy, iteractive collections (streams). Represent a stream as a pair of a value and a thunk; the thunk must produce another stream. We'll be a little sloppy with pin-perfect data abstraction---we should have properties for the parts rather than just lay out the implementation explicitly---but we're in a hurry.

Here's an example that produces the entire infinite set of integers:

integersFrom[n_Integer] := {n, integersFrom[n + 1] &}

extract :: stream --> value

Extract is a (tail-recursive) way to get the 1-indexed, n-th value from any stream (series in mathematics are usually 1-indexed, so we mimic). This is a temporary convenience method; the better ones are forEach and reify, and we get to them below.

By convention, a finite stream has a Null thunk at the end. Thus, the empty stream, obtained by invoking such a thunk, is Null[]. Extracting anything from such a stream produces nothing, that is, Null, which does not print in the notebook and looks like nothing.

ClearAll[extract];
(* defaulting n: *) 
extract[{v_, thunk_}] := extract[{v, thunk}, 1];
(* base case: *) 
extract[Null[], _] := Null; (* somebody invoked my null thunk! *)
(* do this if you just want the \
first value; it's the default *)
extract[{v_, thunk_}, 1] := v;
(* tail recursion: *)
extract[{v_, thunk_}, n_Integer /; n > 0] := extract[thunk[], n - 1];
(* catch-all: *)extract[___] := Null;

Now we can get the 630000-th integer efficiently (if not terribly quickly):

Block[{$IterationLimit = Infinity},
  extract[integersFrom[1], 630000]] // AbsoluteTiming
{1.66595, 630000}

Extracting 0 values produces Null:

extract[integersFrom[0], 0]

take :: stream --> index --> stream

Here's the first of the results of this notebook: an operation that takes a finite number of elements from a stream and produces another stream. We might call it a first-class stream operator.

ClearAll[take];
take[stream_] := take[stream, 1];
take[_, 0] := Null[];
take[{v_, thunk_}, 1] := {v, Null};
take[{v_, thunk_}, n_Integer /; n > 0] := {v, take[thunk[], n - 1] &};

Here it is at work producing a finite stream of three integers, from which we extract the first value:

extract[take[integersFrom[1], 3], 1]
1

and the last value:

extract[take[integersFrom[1], 3], 3]
3

If we extract too far into a finite stream, we get Null, which doesn't print to the notebook:

extract[take[integersFrom[1], 3], 400]

Here is the first one of our ways out:

reify :: stream --> list

Reify a stream to a list, rattling out all thunks (this is not tail-recursive; we might have to do better if we overflow the stack):

ClearAll[reify];
reify[Null[]] := {};
reify[{v_, Null}] := {v};
reify[{v_, thunk_}] := Join[{v}, reify[thunk[]]];

reify[take[integersFrom[1], 3]]
{1, 2, 3}

disperse :: list --> stream

We'll need to go the other way

ClearAll[disperse];
disperse[{}] := Null[];
disperse[{x_}] := {x, Null};
disperse[{v_, xs__}] := {v, disperse[{xs}] &};

reify[disperse[{}]]
{}
reify[disperse[{1, 2, 3}]]
{1, 2, 3}

Fibonaccis as a Stream

Here's a direct, non-tail recursive, fibonacci stream (with memoization, i.e., dynamic programming)

ClearAll[fibonaccis];
fibonaccis[0] = {0, fibonaccis[2] &};
fibonaccis[1] = {1, fibonaccis[3] &};
fibonaccis[n_] :=
 (* Here's the dynamic programming: create a static value for a particular argument. *)
 fibonaccis[n] =
  {(* value *)
   extract[fibonaccis[n - 1]] + extract[fibonaccis[n - 2]],
   (* thunk *)
   fibonaccis[n + 1] &}

This efficient, modulo stack overflow: it can find large Fibonacci numbers quickly:

extract[fibonaccis[0], 200] // AbsoluteTiming
{0.001897, 280571172992510140037611932413038677189525}
extract[fibonaccis[200]] // AbsoluteTiming
{0.00001, 280571172992510140037611932413038677189525}

Here is a fold for streams with the same signature and semantics as FoldList for Lists:

foldrStream :: binaryFunction --> initialValue --> stream --> stream

ClearAll[foldrStream];
(* Produce a singleton from an empty stream *)
foldrStream[bf_, z_, Null[]] := {z, Null};
foldrStream[bf_, z_, {v_, thunk_}] :=
  {z, foldrStream[bf, bf[z, v], thunk[]] &};

This produces a stream from a given stream. It can't produce an empty stream, just as FoldList can't produce an empty list. It's first-class, as composable as take---it produces a stream from a stream.

This next is like our first FoldList example, producing a a stream of fibionaccis, only it's infinite. It produces all the fibonaccis, plus the starting value at the beginning. That's one more than an infinite number, which is exactly the same infinity:

allFibs =
  foldrStream[
   {acc, v} |-> {acc[[2]], acc[[1]] + acc[[2]]},
   {0, 1},
   integersFrom[0]];

Its first value is the starting value:

reify[take[allFibs, 1]]
{{0, 1}}

To reproduce the original example, we must take one more than in our FoldList example:

reify[take[allFibs, 11]]
{{0, 1}, {1, 1}, {1, 2}, {2, 3}, {3, 5}, {5, 8}, {8, 13}, {13, 21}, {21, 34}, {34, 55}, {55, 89}}

foldrStream also works on finite streams:

integersTo[n_] := take[integersFrom[1], n];

reify[integersTo[3]]
{1, 2, 3}
integersTo[0]
Null[]

No way to produce an empty result, even from an empty input:

reify[integersTo[0]]
{}
reify[foldrStream[{acc, v} |-> v, 42, integersTo[0]]]
{42}
reify[foldrStream[{acc, v} |-> v, 42, integersTo[3]]]
{42, 1, 2, 3}

Iterable

forEach :: stream --> unaryFunction --> Null

An iterable is an interface with one method: forEach, which takes an iterable and a unary function (a lambda) and invokes the lambda on every member of the iterable for side effect. A list is an iterable, a stream is an iterable. Iterate over any iterable with forEach. It's the legitimate way to pull values out of an iterable.

It's bad to invoke this on infinite streams, just as with reify.

We won't model interfaces explicitly; they're just functions that take objects as first arguments, by convention, just as you'd emulate OOP in C. We could do so very nicely with Associations, but we're in a hurry.

Here is the implementation of forEach for streams:

ClearAll[forEach];
forEach[Null[], lambda_] := Null;
forEach[{v_, thunk_}, lambda_] :=
  (lambda[v](* discard lambda's result *); 
   forEach[thunk[], lambda];);

and here it is at work, side-effecting the notebook via Print:

forEach[take[allFibs, 11], Print]
{0,1}
{1,1}
{1,2}
{2,3}
{3,5}
{5,8}
{8,13}
{13,21}
{21,34}
{34,55}
{55,89}

Observable

An observable is an interface with one method: subscribe, which has the same signature and semantics as forEach. The difference is that an observable produces its results asynchronously in real time. The big lesson is that the unary lambda function can't tell whether it's being invoked on values in a List, values from a stream, or values in an observable. Put the algorithm in the function, and you've separated the concerns of synchrony from the concerns of algorithms.

Model the asynchronous behavior of observables with tasks---we'll model the observables themselves differently, see below. Here's an infinite one (you MUST interact with this to appreciate it. That goes for most of the observables in this notebook. To follow along, you may download a version of this post as a Mathematica notebook by evaluating the following expression in a fresh notebook:

Import["http://goo.gl/NaH6rM"]["http://i.stack.imgur.com/Yfan3.png"]

Here is our first asynchronous task:

r$ = 0; Dynamic[r$]
RunScheduledTask[r$++, 1];
DynamicBox[...]

When you're tired watching it, clean up all tasks:

ClearAll[cleanup];
cleanup[] := (ClearAll[r$]; RemoveScheduledTask[ScheduledTasks[]];);
cleanup[]

Here is a finite one:

r$ = 42; Dynamic[r$]
RunScheduledTask[r$++;, {1, 3}];
DynamicBox[...]

Still must clean it up:

cleanup[]

Create a task without running it:

r$ = 42;
myTask$ = CreateScheduledTask[r$++;, {1, 3}]
ScheduledTaskObject[...]

Now run it and clean it up:

Dynamic[r$]
StartScheduledTask[myTask$];
DynamicBox[...]
cleanup[]

Observables must produce values every tick so we can subscribe lambdas to them. Model observables as stateful thunks that produce new values every time they're invoked, then invoke the thunks inside asynchronous tasks that start when you subscribe.

Such tasks implement cold observables that don't produce values until subscribed; there are also hot observables that produce values whether subscribed or not, and you can subscribe and unsubscribe on-the-fly. We'll finesse those now because the purpose here is to show that subscribe is almost the same as forEach. Our purpose here is illustrative, not to produce a complete observables library (see this book by Lee Campbell for much more).

Fibonacci as an Observable

Here's a factory for fibonacci observables:

ClearAll[fibonacciObservable];
fibonacciObservable[] :=
 Module[{a = 0, b = 1, oblFibs},
  oblFibs[] :=
   With[{a$ = a, b$ = b, result = {a, b}},
    a = b$;
    b = a$ + b$;
    result];
  oblFibs]

Here we show multiple calls resulting in statefully updated values:

oblFibs = fibonacciObservable[];
oblFibs[] & /@ Range[11]
{{0, 1}, {1, 1}, {1, 2}, {2, 3}, {3, 5}, {5, 8}, {8, 13}, {13, 21}, {21, 34}, {34, 55}, {55, 89}}

Of course, we've left some garbage in the environment (the local varirables created by the Module). But we'll finesse that now, too (we're still in a hurry).

Watch this until it stops changing, then clean it up.

Dynamic[r$]
oblFibs = fibonacciObservable[];
RunScheduledTask[r$ = oblFibs[], {1, 11}];
DynamicBox[...]
cleanup[]

Just as we forEach a unary lambda function over a synchronous collection (finite or infinite, eager or lazy), we subscribe a lambda to an observable. Such lambdas are also called observers, but they're just ordinary lambdas.

subscribe :: observable --> unaryFunction --> Null

We emphasize: the syntax and semantics of forEach and subscribe are virtually identical:

ClearAll[subscribe];
subscribe[observable_, observer_] :=
  (RunScheduledTask[observer[observable[]]];);

Watch this until you get tired (it's infinite), then clean it up:

Dynamic[r$]
subscribe[
 fibonacciObservable[],
 v |-> (r$ = v)]
DynamicBox[...]
cleanup[]

Here's an infinite observable of all the integers, just like our iterable integersFrom:

ClearAll[integersFromObservable];
integersFromObservable[from_] :=
  Module[{state = from, integersObl},
   integersObl[] :=
    With[{result = state},
     state = state + 1;
     result];
   integersObl];

Watch it for a while, then clean it up:

Dynamic[r$]
subscribe[
  integersFromObservable[42],
  v |-> (r$ = v)];
DynamicBox[...]
cleanup[]

Here are our ways to get back and forth between lists and observables (doing this right requires another aspect of observers, namely OnCompleted, not covered here (see the book by Campbell for more). We'll joke it with Null for now).

dispense :: list --> observable

ClearAll[dispense];
dispense[{}] := Null &;
dispense[l_List] :=
 Module[{state = l},
  If[{} === state,
    Null,
    With[{v = First[state]},
     state = Rest[state];
     v]] &]

Here's a finite instance. Clean it up once it starts to produce Null.

Dynamic[r$]
subscribe[dispense[{1, 2, 3}], v |-> (r$ = v)];
DynamicBox[...]
cleanup[]

harvest :: observable --> list

Mathematica doesn't have general reference semantics, so we'll just update a global variable now, for illustrative purposes (by now you know we're still in a hurry, but see this StackExchange question for work-in-progress to employ Reap and Sow). Also not that this uses AppendTo, not the most efficient way (see this blog entry for more), but we digress.

ClearAll[harvest];
harvest[obl_] :=
  (r$ = {}; 
   subscribe[obl, v |-> If[v =!= Null, AppendTo[r$, v]]]);

Dynamic[r$]
harvest[dispense[{1, 2, 3}]];
DynamicBox[...]

We can capture the output list by evaluating r$ non-dynamically before doing the cleanup:

r$
{1, 2, 3}
cleanup[]

takeObservable :: observable --> integer --> observable

Here's our take version for observables:

ClearAll[takeObservable];
takeObservable[obl_, n_Integer /; n >= 0] :=
  Module[{newObl, state = Null, count = n},
   newObl[] :=
    If[count === 0, Null,
     count--;
     obl[]];
   newObl];

and the usual test:

Dynamic[r$]
subscribe[
  takeObservable[integersFromObservable[42], 3],
  v |-> (r$ = v)];
DynamicBox[...]
cleanup[]

All that's left is to produce a new observable by folding a binary function over another observable.

foldObservable :: binaryFunction --> initialValue --> observable --> observable

ClearAll[foldObservable];
foldObservable[bf_, z_, obl_] :=
 Module[{newObl, z$ = z},
  newObl[] := With[
    {result = z$},
    z$ = bf[z$, obl[]];
    result];
  newObl]

Dynamic[r$]
subscribe[
  foldObservable[
   {acc, v} |-> {acc[[2]], acc[[1]] + acc[[2]]},
   {0, 1},
   integersFromObservable[0]],
  v |-> (r$ = v)];

cleanup[]

There are lots of better ways to do this, but we've captured the gist.

Conclusion

We've established that folding over lists produces lists, folding over iterables produces iterables, and that folding over observables produces observables. All our algorithms, expressed as mere binary functions, can run over finite, infinite, synchronous, and asynchronous collections.

Again, you may download a version of this post as a Mathematica notebook by evaluating the following expression in a fresh notebook:

Import["http://goo.gl/NaH6rM"]["http://i.stack.imgur.com/Yfan3.png"]
$\endgroup$
  • $\begingroup$ Is "Iteractive" a typo or jargon I am unfamiliar with? (Iterative? Interactive? Interactively-iterative?) $\endgroup$ – Mr.Wizard Apr 13 '15 at 1:52
  • $\begingroup$ It's a real (made-up) word. Some of the reactive crowd goes about calling synchronous things "interactive," but I've pointed out that they are simply misspelling "iteractive" [sic], which is much more precise :) $\endgroup$ – Reb.Cabin Apr 13 '15 at 3:21

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.