%!TEX root = ../Main.tex

\clearpage{}
\section{Sequential Data Flow Fusion}
\label{s:SeqFlowDef}
The main definitions for sequential data flow fusion are given in Figure \ref{f:SeqFlowDef}. 

Unlike in the stream fusion approach, the per-element functions @get1@ and @get8@ are in the @IO@ monad instead of being @ST@ style state threads. Hence, the effects of a function that evaluates a flow cannot be safely masked. We will need this when we move to the parallel case in Section \ref{s:ParFlowDef} because we will reuse this definitions and have the @get1@ and @get8@ functions perform intra-thread communication.


\begin{figure}
\begin{small}
\begin{code}
data FD -- Type tag for delayed flow.
data FS -- Type tag for stateful flow.

data Flow r a
 = forall state. Flow
 { flowStart  :: IO state
 , flowSize   :: state -> IO Size
 , flowReport :: state -> IO Report
 , flowGet1   :: state -> (Step1 a -> IO ()) -> IO ()
 , flowGet8   :: state -> (Step8 a -> IO ()) -> IO () }

data Step1 a  = Yield1 a Bool           | Done
data Step8 a  = Yield8 a a a a a a a a  | Pull1
data Size     = Exact Int               | Max   Int
date Report   = ...

flow   :: (Int -> a) -> Int -> Flow mode a
unflow :: Flow FD a  -> Vector a

drain  :: (Int   -> IO (vec a))        -- allocate vector
       -> (vec a -> Int -> a -> IO ()) -- write into vector
       -> Flow mode a                  -- flow to evaluate
       -> IO (vec a, Int)  -- ^ result vector, and num elts
\end{code}
\end{small}

\caption{Sequential Flows}
\label{f:SeqFlowDef}
\end{figure}


% -----------------------------------------------------------------------------
\subsection{Mapping Flows}
\label{s:SeqFlowMap}
The definition of @map@ for sequential flows is given in Figure \ref{f:SeqFlowMap}

The @map@ function is a pure flow transformer, whose application does not force any elements of the flow to be evaluated.

\TODO{explain how @map@ works}

\begin{figure}
\begin{small}
\begin{code}
map :: (a -> b) -> Flow mode a -> Flow mode b
map f (Flow start size report get1 get8)
 = Flow start size report' get1' get8'
 where 
  report' state = ...
  get1' state push1
   =  get1 state \$ \r 
   -> case r of
       Done            -> push1 \$ Done
       Yield1 x hint   -> push1 \$ Yield1 (f x) hint

  get8' state push8
   =  get8 state \$ \r
   -> case r of
       Pull1          -> push8 \$ Pull1
       Yield8 x0 x1 x2 x3 x4 x5 x6 x7
        -> push8 \$ Yield8 (f x0) (f x1) (f x2) (f x3)
                           (f x4) (f x5) (f x6) (f x7)
\end{code}
\end{small}
\caption{Map for sequential flows}
\label{f:SeqFlowMap}
\end{figure}


% -----------------------------------------------------------------------------
\subsection{Folding Flows}
\label{s:SeqFlowFold}
The definition of @foldl@ for sequential flows is given in Figure \ref{f:SeqFlowFoldl}. Unlike with @map@ from \S\ref{s:SeqFlowMap}, the fold function is an @IO@ action instead of being a pure flow transformer. This is because evaluating @foldlIO@ forces the elements of the flow to be demanded, and as we mentioned in \S\ref{s:SeqFlowDef} this may involve intra-thread communcation.

\TODO{explain how @foldlIO@ works}

The definition of @foldlIO@ in Figure \ref{f:SeqFlowFoldl} is not incremental, and therefore cannot be used to solve the space leak problem with multiple consumers outlined in \S\ref{s:IntroMultipleConsumers}. As mentioned there, to consume an unzipped vector in constant space we must be able to co-routine between the two consumers, but @foldlIO@ as presented demands the entire flow at once. We deal with this in the next section.


\begin{figure}
\begin{small}
\begin{code}
foldlIO :: (a -> b -> a) -> a -> Flow mode b -> IO a
foldlIO f z (Flow start _ _ get1 get8)
 = do outRef <- unew 1
      write outRef 0 z
      state  <- start
      let
       eat1 !acc
        = get1 state \$ \r 
        -> case r of
            Done  -> write outRef 0 acc
            Yield1 x1 hint
             -> let !acc' = (acc `f` x1)
                in  if hint then eat8 acc'
                            else eat1 acc'
       eat8 !acc 
        =  get8 state \$ \r 
        -> case r of
            Pull1 -> eat1 acc
            Yield8 x0 x1 x2 x3 x4 x5 x6 x7
              -> eat8 ( acc `f` x0 `f` x1 `f` x2 `f` x3
                            `f` x4 `f` x5 `f` x6 `f` x7) 
       eat8 z
       read outRef 0
\end{code}
\end{small}
\caption{Non-incremental Left fold for sequential flows}
\label{f:SeqFlowFoldl}
\end{figure}


\clearpage{}
% -----------------------------------------------------------------------------
\subsection{Switched Flow Fold}
\TODO{This is just a sketch. None of the stuff in this section is implemented.}

\begin{small}
\begin{code}
 foldlIO :: (a -> b -> a) -> a 
         -> Flow mode b -> IO (Switched a)
\end{code}
\end{small}

\begin{small}
\begin{code}
 data Step1 a    = Yield1 a Bool | Done  | Stall
 data Switched a = Give a        | Switch
\end{code}
\end{small}

The producer sends a @Stall@ constructor to indicate that no new elements are currently available (without causing a space leak). All flow functions like @map@ and @foldlIO@ must be modified so that on receiving a @Stall@ from the in-flow they pass this onto the out-flow.

We then use the following @unswitch@ function to alternate between two switched consumers.
\begin{small}
\begin{code}
 unswitch :: IO (Switched a) -> IO (Switched b) 
          -> (a -> b -> c) -> IO c
\end{code}
\end{small}

\begin{small}
\begin{code}
 unzipW
  :: Flow mode (a, b) 
  -> IO (Flow mode (Switched a), Flow mode (Switched b))

 do (ys, zs) <- unzipSwitched zs
    unswitch (sumW ys) (prodW zs) (,)
\end{code}
\end{small}

\TODO{Can we do the @unswitch@ part in a nice monad?. Something like:}

\begin{small}
\begin{code}
do (ys, zs) <- unzipSwitched zs
    do y <- sumW ys
       z <- prodW zs
       return (y, z)
\end{code}
\end{small}

\begin{itemize}

\item   Pros: external API of non-reduction flow transform functions does not need to change, because we just need to add the Stall constructor to the @Step1@ type.

\item   Cons: Requires programmer to manually manage the join point.

\item   We need to change drain to be switchable. All functions that force evaluation of flows must be switchable. 

\item   I don't know how we're going to introduce the explicit join-points into a vectorised program.

\item   Can we put functions that are locked to the same producer in a single ST-like monad to prevent space leaks?
\end{itemize}

