Deforestation is usually defined as the elimination of an intermediate data structure between a single producer and a single consumer. Jorge Adriano showed an interesting example of one producer feeding two independent consumers. The two streams of data are mutually dependent. Furthermore, the rate of production is non-uniform: Generating an element in one stream may require generating trillion items in the other stream. If the first consumer is being evaluated, that consumer causes the evaluation of the producer until the latter yields an item. The trillion of items incidentally produced in the other stream have to be stored somewhere. In more OS-centric terms, the evaluator can't generally reduce two expressions in parallel. If it chooses one stream consumer, the other consumer is set aside. While the first stream consumer "waits" for a value, the second stream consumer is "blocked", and hence the corresponding stream must be buffered, sometimes at great expense. We will consider the problem of one producer and two consumers in the general setting. We derive a deforested version, which no longer needs to buffer any produced items. The retaining profile shows no space leaks. We also consider "parallel" writing of streams into two distinct files. Our solution is safe but the i/o is effectively interleaved. The deforested solution is indeed _derived_, via a sequence of equivalent transformations. In fact, the derived code worked on the first try. In the original example, the two consumers were writing received data into the their own files. Let us first generalize the problem. We first assume that the two mutually-recursive streams are specified by a transition function transition:: (a,b) -> (a,b) one the state of a _pre-stream_. The state of the latter is a pair of values. We obtain two streams in question by splitting the pre-stream into the fst and the snd components, and applying two different filters. The filtering part makes the rate of production non-uniform.
import IO
infixr 0 $^$ -- Can't do pattern-matching-deconstruction on state because -- pattern-matching is eager! (f,g) $^$ state = (f $ fst state, g $ snd state)
stream transition (p,q) state@(xstate,ystate) = ((filter p) . (xstate :), (filter q) . (ystate :)) $^$ (stream transition (p,q) $ transition state)
For illustration, we will try the following transition function and the filters:
trans1 (x,y) = (y+1,x) st_p x = x `mod` 1997 == 0 st_q = const True
The streams are: without filtering Main> take 9 $ fst $ stream trans1 (\x->True,\x->True) (0,0) [0,1,1,2,2,3,3,4,4] Main> take 9 $ snd $ stream trans1 (\x->True,\x->True) (0,0) [0,0,1,1,2,2,3,3,4] and with filtering: Main> take 9 $ snd $ stream trans1 (st_p,st_q) (0,0) [0,0,1,1,2,2,3,3,4] (443 reductions, 782 cells) Main> take 9 $ fst $ stream trans1 (st_p,st_q) (0,0) [0,1997,1997,3994,3994,5991,5991,7988,7988] (6262603 reductions, 9198209 cells, 42 garbage collections) Generating each item in the first stream indeed generates quite a few (thousand) items in the second stream. In our first example, the consumers will sum the corresponding streams:
consumers_sum stream n = (f,f) $^$ stream trans1 (st_p,st_q) (0,0) where f = sum . take n
More generally, we can write a consumer using a function sfoldl (stream fold). The function is similar to the regular foldl but can be applied to infinite streams (and so it needs a predicate to tell the termination).
sfoldl sf st z (x:_) | sf x z = z sfoldl sf st z (x:xs) = sfoldl sf st (st z x) xs
Our consumers_sum function can be written as
consumers_sum' stream n = (fst,fst) $^$ (f,f) $^$ stream trans1 (st_p,st_q) (0,0) where f = sfoldl sf st (0,0) sf x (_,count) = count >= n st (sum,count) val = (val+sum,count+1)
In general, the essence of producing and consuming two streams is captured by an expression (sfoldl sf1 st1 z1, sfoldl sf2 st2 z2) $^$ stream trans (p,q) (x,y) === { definition of stream } === (sfoldl sf1 st1 z1, sfoldl sf2 st2 z2) $^$ ( ((filter p) . (x :), (filter q) . (y :)) $^$ stream trans (p,q) $ trans (x,y) ) === { We know a$(b$c) === (a.b) $ c Similarly, a $^$ (b $^$ c) === ((((.),(.)) $^$ a) $^$ b) $^$ c or, in a less bizarre notation a $^$ (b $^$ c) === ((fst a . fst b),(snd a . snd b)) $^$ c } === ((sfoldl sf1 st1 z1) . (filter p) . (x :), (sfoldl sf2 st2 z2) . (filter q) . (y :)) $^$ stream trans (p,q) $ trans (x,y) where we also took advantage of the associativity of composition. We note that two folds, sfoldl and the fold implicit in filter, can be fused: sfoldlf sf st p z =def= (sfoldl sf st z) . (filter p) Then sfoldlf sf st p z (x:xs) === (sfoldl sf st z) $ (filter p (x:xs)) === { definition of filter, lifting the condition } if p x then (sfoldl sf st z (x:filter p xs)) else (sfoldl sf st z (filter p xs)) === { definition of sfoldlf } if p x then (sfoldl sf st z (x:filter p xs)) else sfoldlf sf st p z xs === { definition of sfoldl } if p x then if sf x z then z else sfoldl sf st (st z x) (filter p xs) else sfoldlf sf st p z xs === if p x then if sf x z then z else sfoldlf sf st p (st z x) xs else sfoldlf sf st p z xs We have just obtained a recursive equation for sfoldlf. Note that sfoldlf (just as sfoldl) are undefined for empty lists. We can now write our producer-consumers problem as (sfoldl sf1 st1 z1, sfoldl sf2 st2 z2) $^$ stream trans (p,q) (x,y) === ((sfoldlf sf1 st1 p z1) . (x :), (sfoldlf sf2 st2 q z2) . (y :)) $^$ stream trans (p,q) $ trans (x,y) We already see a notable simplification. The streams are now producing at the same rate! So we can replace a pair of two streams with one stream of pairs. But we can do better. Let's look at the equation for sfoldlf again sfoldlf sf st p z (x:xs) === if p x then if sf x z then z else sfoldlf sf st p (st z x) xs else sfoldlf sf st p z xs The equation suggests a useful generalization of sfoldl:
sfoldl' sf st z True _ = z sfoldl' sf st z done (x:xs) | sf x z = sfoldl' sf st z True xs sfoldl' sf st z done (x:xs) = sfoldl' sf st done (st z x) xs
We can now introduce sfoldlf' sf st p done z =def= (sfoldl' sf st done z) . (filter p) which obeys the equation sfoldlf' sf st p done z (x:xs) === if done then z else if p x then if sf x z then sfoldlf' sf st z True xs else sfoldlf' sf st p done (st z x) xs else sfoldlf' sf st p done z xs === { pushing conditions inside } sfoldlf' sf st p (done || (p x && sf x z)) (ss sf st p done z x) xs where ss sf st p done z x = if done then z else if p x then if sf x z then z else (st z x) else z Something remarkable has happened: we've got the left recursion! We should note that the introduction of a 'done' argument is not unlike a common trick in physics of converting a value into a functional with the help of the Dirac delta-function. We can now write sfoldlfs sf1 st1 done1 z1 sf2 st2 done2 z2 (p,q) trans state@(x,y) =def= (sfoldl' sf1 st1 done1 z1, sfoldl' sf2 st2 done2 z2) $^$ stream trans (p,q) (x,y) === ((sfoldl' sf1 st1 done1 z1) . (filter p) . (x :), (sfoldl' sf2 st2 done2 z2) . (filter q) . (y :)) $^$ stream trans (p,q) $ trans (x,y) === (\tail-> sfoldlf' sf1 st1 p done1 z1 (x:tail), \tail-> sfoldlf' sf2 st2 q done2 z2 (y:tail) ) $^$ stream trans (p,q) $ trans (x,y) === (\tail-> sfoldlf' sf1 st1 p (done1 || (p x && sf1 x z1)) (ss sf1 st1 p done1 z1 x) tail, \tail-> sfoldlf' sf2 st2 q (done2 || (q y && sf2 y z2)) (ss sf2 st2 q done2 z2 y) tail ) $^$ stream trans (p,q) $ trans (x,y) === { eta reduction } (sfoldlf' sf1 st1 p (done1 || (p x && sf1 x z1)) (ss sf1 st1 p done1 z1 x), sfoldlf' sf2 st2 q (done2 || (q y && sf2 y z2)) (ss sf2 st2 q done2 z2 y) ) $^$ stream trans (p,q) $ trans (x,y) === { definition of sfoldlfs } sfoldlfs sf1 st1 (done1 || (p x && sf1 x z1)) (ss sf1 st1 p done1 z1 x) sf2 st2 (done2 || (q y && sf2 y z2)) (ss sf2 st2 q done2 z2 y) (p,q) trans (trans state) And we have obtained the primitive recursive solution with no lists whatsoever. We have achieved the full deforestation! Now we just have to program it:
ss:: (x->z->Bool) -> (z->x->z) -> (x->Bool) -> Bool -> z -> x -> z ss sf st p done z x = if done then z else if p x then (if sf x z then z else (st z x)) else z
*> sfoldlfs sf1 st1 True z1 sf2 st2 True z2 (p,q) trans state = (z1,z2) *> sfoldlfs sf1 st1 done1 z1 sf2 st2 done2 z2 flt@(p,q) trans state@(x,y) = *> sfoldlfs sf1 st1 (done1 || (p x && sf1 x z1)) (ss sf1 st1 p done1 z1 x) *> sf2 st2 (done2 || (q y && sf2 y z2)) (ss sf2 st2 q done2 z2 y) *> flt trans (trans state) Or, factoring out common arguments
sfoldlfs sf1 st1 done1 z1 sf2 st2 done2 z2 flt@(p,q) trans state = aux done1 z1 done2 z2 state where aux True z1 True z2 _ = (z1,z2) aux done1 z1 done2 z2 state@(x,y) = aux (done1 || (p x && sf1 x z1)) (ss sf1 st1 p done1 z1 x) (done2 || (q y && sf2 y z2)) (ss sf2 st2 q done2 z2 y) (trans state)
consumers_sum'' trans n = (fst,fst) $^$ sfoldlfs sf st False (0,0) sf st False (0,0) (st_p,st_q) trans (0,0) where sf x (_,count) = count >= n st (sum,count) val = (val+sum,count+1)
Let us now compare the forested version consumers_sum' with the deforested version consumers_sum'' in performance, using GHCi: *Main> consumers_sum' stream 8 (31952,12) (0.26 secs, 11934436 bytes) *Main> consumers_sum'' trans1 8 (31952,12) (0.18 secs, 7695084 bytes) *Main> consumers_sum' stream 16 (127808,56) (0.86 secs, 32660328 bytes) *Main> consumers_sum'' trans1 16 (127808,56) (0.32 secs, 14354768 bytes) *Main> consumers_sum'' trans1 16 (127808,56) (0.32 secs, 14579336 bytes) *Main> consumers_sum' stream 32 (511232,240) (3.32 secs, 97786900 bytes) *Main> consumers_sum'' trans1 32 (511232,240) (0.69 secs, 28490268 bytes) It appears that the deforestation was worth it. The resulting code needs less memory and executes notably faster. In another test, we compile the code with optimization and profiling enabled:
main = do print $ consumers_sum' stream 32; print $ consumers_sum'' trans1 32
The result indicate that consumers_sum' is responsible for only 15.4% of the time and 13.7% of the allocations (most of which are due to the transition function). What's more exhilarating, the retainer profile of "print $ consumers_sum'' trans1 32" shows nothing. Nothing is retained. In contrast, the retainer profile of "print $ consumers_sum' stream 32;" shows large a large space leak of 12 MB of retained memory. But what about the original question by Jorge Adriano? Can we write two stream into two different files, and _safely_ interleave the i/o? Yes, we can. We need a version of sfoldlfs that shares the singe seed between two consumers:
sfoldlfs1 sf1 st1 done1 sf2 st2 done2 flt@(p,q) trans z state = aux done1 done2 z state where aux True True z _ = z aux done1 done2 z state@(x,y) = aux (done1 || (p x && sf1 x z)) (done2 || (q y && sf2 y z)) (ss sf2 st2 q done2 (ss sf1 st1 p done1 z x) y) (trans state)
write_in_two_files file1 file2 trans max = (\(h1,h2) -> hClose h1 >> hClose h2) =<< sfoldlfs1 sf (st fst) False sf (st snd) False (st_p,st_q) trans (do h1 <- openFile file1 WriteMode; h2 <- openFile file2 WriteMode; return (h1,h2)) (0,0) where sf x _ = x >= max st selector two_handles val = do handles <- two_handles hPrint (selector handles) val; return handles
*> main = write_in_two_files "/tmp/a1" "/tmp/a2" trans1 80000 The profile shows that the i/o is interleaved indeed.