Co-op streams from scratch in 33 rows on Haskell

Haskell distinguishes itself from most functional languages ​​in that it has deep cultural roots from the field of mathematics and computer science, which give the deceptive impression that Haskell is not well suited for solving practical problems. However, the more you know Haskell, the more you appreciate that theory is often the most practical solution to many common programming problems. This article would like to emphasize this point of view by the fact that we mix the existing theoretical foundations and create a clean user flow system.

Type of

Haskell is a language where types are primary, so we start by choosing the appropriate type to represent streams. First of all, we have to indicate in simple language which flows we want to make:

• Threads must extend existing instruction sequences.
• Threads must support a set of operations: branching, transfer of control, and completion
• Threads must allow different types of schedulers.

Now we translate these concepts into Haskel:

• When you hear "several interpreters / planners / backends" you should think "free" (as in "free object")
• When you hear the “sequence of commands” you should think: “monads”.
• When you want to “expand” something you have to think: “transformers”.

Combine these words together and you will get the right mathematical solution: “free monad transformer”.

Syntax tree

“Free monad transformer” is a fancy name for a mathematical abstract syntax tree, where consistency plays an important role. We provide it with a set of instructions and it builds us a syntax tree of these instructions.

We said that we want our stream to either branch, or transfer control, or stop, so let's do a data type with forks, returns, and termination:

{-# LANGUAGE DeriveFunctor #-} data ThreadF next = Fork next next | Yield next | Done deriving (Functor) 

ThreadF presents our instruction set. We wanted to add three new instructions, so ThreadF has three constructors, one for each command: Fork , Yield , and Done .
')
Our ThreadF type represents a single node in the syntax tree. next fields from constructors represent where the children of the nodes should go. Fork creates two ways of performing, so he has two children. Done completes the current execution path, so he has no children. Yield neither branches nor stops, so he has one child. The deriving (Functor) part simply tells the free monad transformer that next field is where the children should go.

Now the free monad transformer FreeT can build the syntax tree of our commands. We will call this tree a thread:

 --  `free`  import Control.Monad.Trans.Free type Thread = FreeT ThreadF 

An experienced Haskel programmer will read this code, as if saying “ Thread is a syntax tree built from ThreadF instructions”.

Instructions

Now we need primitive instructions. free package provides a liftF operation that converts one command into a syntax tree one node deeper:

 yield :: (Monad m) => Thread m () yield = liftF (Yield ()) done :: (Monad m) => Thread mr done = liftF Done cFork :: (Monad m) => Thread m Bool cFork = liftF (Fork False True) 

You do not need to fully understand how this works, except to notice that the return value of each command corresponds to what we store in the child of the node field:

• yield command saves () as its child, so the return value of the function is ()
• done command has no children, so the compiler infers that it has a polymorphic return value (ie, r ), which means that it will never end.
• The cFork command stores logical values ​​as children, so it returns a Bool

cFork gets its name because it behaves like a fork function from C, which means the returned boolean value tells us which branch we are on after branching. If we get False , then we are on the left branch and if we get True , then we are on the right branch.

We can combine cFork and done anew by implementing fork in a more traditional Haskell style, using the convention that the left branch is the “parent” and the right branch is the “child”:

 import Control.Monad fork :: (Monad m) => Thread ma -> Thread m () fork thread = do child <- cFork when child $ do thread done 

The code above calls cFork , and then cFork says, "If I'm a child, start the split action, and then stop, otherwise just continue as usual."

Free monads

Notice how something unusual happened in the last code snippet. We assembled cFork and done functions from the primitive Thread thread instructions using the do notation, and we got the new Thread back. This is because Haskell allows us to use do notation of any type that implements the monad interface ( Monad ) and our free monad transformer automatically determines the necessary monad instance for Thread . Amazing!

In fact, our free monadny transformer is not at all super-smart. When we compile a free monad transformer using do notation, all that is done is to connect these primitive syntax trees into one node of depth (i.e., instructions) into a larger syntax tree. A sequence of two commands:

 do yield done 

... is discarded simply into storing the second command (ie, done ) as a child of the first command (ie, yield ).

Cyclic Thread Manager

Now we are going to write our own thread scheduler. This will be a naive circular planner:

 --   O(1)      import Data.Sequence roundRobin :: (Monad m) => Thread ma -> m () roundRobin t = go (singleton t) --     where go ts = case (viewl ts) of --   : ! EmptyL -> return () --   :      t :< ts' -> do x <- runFreeT t --     case x of --       Free (Fork t1 t2) -> go (t1 <| (ts' |> t2)) --       Free (Yield t') -> go (ts' |> t') --  :     Free Done -> go ts' Pure _ -> go ts' 

... and you're done! No, really, that's all! This is a complete streaming implementation.

Custom streams

Let's try our brave new streaming system. Let's start with something simple.

 mainThread :: Thread IO () mainThread = do lift $ putStrLn "Forking thread #1" fork thread1 lift $ putStrLn "Forking thread #1" fork thread2 thread1 :: Thread IO () thread1 = forM_ [1..10] $ \i -> do lift $ print i yield thread2 :: Thread IO () thread2 = replicateM_ 3 $ do lift $ putStrLn "Hello" yield 

Each of these threads is of type Thread IO () . Thread is a “monad transformer”, which means that it expands the existing monad with additional functionality. In our case, we extend the IO monad with user threads, and this, in turn, means that every time we need to invoke an IO action, we use lift to insert this action into the Thread .

When we call the roundRobin function, we pull out our Thread monad transformer, and our stream program collapses to a linear sequence of instructions in IO

 >>> roundRobin mainThread :: IO () Forking thread #1 Forking thread #1 1 Hello 2 Hello 3 Hello 4 5 6 7 8 9 10 

Moreover, our streaming system is clean! We can extend other monads, not just IO , and still get stream effects! For example, we can build stream Writer computations, where Writer is one of the many pure monads (for more information about it, see on Habré ):

 import Control.Monad.Trans.Writer logger :: Thread (Writer [String]) () logger = do fork helper lift $ tell ["Abort"] yield lift $ tell ["Fail"] helper :: Thread (Writer [String]) () helper = do lift $ tell ["Retry"] yield lift $ tell ["!"] 

This time, the roundRobin function roundRobin a clean Writer action when we start the logger :

 roundRobin logger :: Writer [String] () 

... and we can extract the results of the logging command as well:

 execWriter (roundRobin logger) :: [String] 

Notice how the type computes the net value, the list is String in our case. And we can still get real streams of logged values:

 >>> execWriter (roundRobin logger) ["Abort","Retry","Fail","!"] 

Conclusion

You may think that I am a cheater, that the main work went to the free library, but all the functionality that I used can fit in 12 lines of a very common code, suitable for secondary use.

 data FreeF fax = Pure a | Free (fx) newtype FreeT fma = FreeT { runFreeT :: m (FreeF fa (FreeT fma)) } instance (Functor f, Monad m) => Monad (FreeT fm) where return a = FreeT (return (Pure a)) FreeT m >>= f = FreeT $ m >>= \v -> case v of Pure a -> runFreeT (fa) Free w -> return (Free (fmap (>>= f) w)) instance MonadTrans (FreeT f) where lift = FreeT . liftM Pure liftF :: (Functor f, Monad m) => fr -> FreeT fmr liftF x = FreeT (return (Free (fmap return x))) 

This is a common trend in Haskell: when we use theory, we get a frequently used, elegant and powerful solution in a shocking little code.

The writing of the article was inspired by the article by Peng Lee and Steve Zhdantevich “Methods of language for combining flows and events”. The main difference is that the continuation methods were replaced by simpler methods of the free monad.