重要的部分在于loop函数：

loop [] var = put var Nil
  loop (x:xs) var = do
    tail <- new
    put var (Cons x tail)
    loop xs tail

我们需要添加分叉距离f和区块大小c作为参数：

loop _ _ [] var = put var Nil
  loop 0 c (x:xs) var = -- see below
  loop f c (x:xs) var = do
    tail <- new
    put var (Cons x tail)
    loop (f-1) c xs tail

分叉距离在每次迭代中减小。当分叉距离为零时，我们需要做什么？我们提供了一个Fork op t，其中op继续生成列表：

loop 0 c (x:xs) var = do
    tail <- new
    let op = loop c xs tail
    put var (Fork op (Cons x tail))

请注意，如果列表为空，则不使用Fork。这是可能的，但有点傻，毕竟，没有什么可生产的了。现在更改streamFromList很简单：

streamFromList :: NFData a => Int -> Int -> [a] -> Par (Stream a)
streamFromList f c xs = do
  var <- new                            
  fork $ loop f c xs var                 
  return var

现在，为了使用它，我们需要在streamFold中更改case：

streamFold :: (a -> b -> a) -> a -> Stream b -> Par a
streamFold fn acc instrm = acc `seq` do
  ilst <- get instrm
  case ilst of
    Cons h t          -> streamFold fn (fn acc h) t
    Fork p (Cons h t) -> -- see below
    _                 -> return acc

请记住，我们不允许streamFromList中的Fork中有空列表，但以防万一我们通过Nil匹配它（和Nil）。
如果我们遇到一个Fork数据，我们需要做什么？首先，我们需要使用fork来运行Par ()操作，以便传播t，然后我们可以开始使用它。我们最后一个案子是

Fork p (Cons h t) -> fork p >> streamFold fn (fn acc h) t

streamMap类似。只有在这种情况下，你才能像在streamFromList中一样在循环中使用额外的参数。

我认为以下是一个有效的实现。

{-# LANGUAGE BangPatterns #-}

import Control.Monad.Par (IVar, Par, fork, get, new, put, put_, runPar)
import Control.DeepSeq   (NFData, rnf)

data IList a
  = Nil
  | Cons a (IVar (IList a))
  | Fork (Par ()) (IVar (IList a))

instance NFData a => NFData (IList a) where
  rnf Nil = ()
  rnf (Cons a b) = rnf a `seq` rnf b
  rnf (Fork a b) = rnf (runPar a) `seq` rnf b

type Stream a = IVar (IList a)

main :: IO ()
main = print $ sum (pipeline [1 .. 10000])

pipeline :: [Int] -> [Int]
pipeline list = runPar $ do
  strm <- streamFromList list 100 200
  xs   <- streamFold (\x y -> (y : x)) [] strm
  return (reverse xs)

streamFromList :: NFData a => [a] -> Int -> Int -> Par (Stream a)
streamFromList xs k n = do
    var <- new
    fork $ loop xs var k
    return var
  where
    loop [] var _ = put var Nil
    loop xs var 0 = do
      var' <- new
      put_ var (Fork (loop xs var' n) var')
    loop (x:xs) var i = do
      tail <- new
      put var (Cons x tail)
      loop xs tail (i - 1)

streamFold :: (a -> b -> a) -> a -> Stream b -> Par a
streamFold fn !acc strm = do
  ilst <- get strm
  case ilst of
    Nil      -> return acc
    Cons h t -> streamFold fn (fn acc h) t
    Fork p s -> fork p >> streamFold fn acc s

在这里，streamFromList（生产者）将值分配给流，而streamFold并行使用它们。在第一个k值之后，streamFromList将Fork放入流中。这个Fork包含产生下一个n值的计算，以及可以从中消费这些值的流。
在这一点上，消费者有机会赶上，如果它落后于生产者。在到达Fork时，它是fork所包含的生产者。同样，生产者和消费者可以并行进行，直到生产者在另一个n值之后将另一个Fork添加到流中，并且循环重复。

在这个实现中，fork被放置在生成列表的中间。

import Control.DeepSeq
import Control.Monad.Par

data IList a
    = Nil -- need to be NFData
    | Cons a (IVar (IList a))
    | Fork (Par ()) (IList a)

instance NFData a => NFData (IList a) where
    rnf Nil = ()
    rnf (Cons x xs) = rnf x `seq` rnf xs
    rnf (Fork c l) = rnf l

type Stream a = IVar (IList a)

-- >>> runPar $ (streamFromList 3 [1 .. 10]) >>= (streamFold (+) 0)
-- 55

streamFromList :: NFData a => Int -> [a] -> Par (Stream a)
streamFromList chunkSize xs = do
    dt <- new
    dl <- new
    put dl xs
    fork $ next chunkSize dt dl
    return dt
  where
    next :: NFData a => Int -> Stream a -> IVar [a] -> Par ()
    next 1 dt dl = do 
        ilist <- get dl 
        case ilist of 
            [] -> put dt Nil 
            (x:xs) -> do 
                delaytail <- new 
                delaylist <- new 
                put delaylist xs
                put dt (Fork (next 1 delaytail delaylist) (Cons x delaytail))
    next chunkSize dt dl = do
        ilist <- get dl
        case ilist of
            [] -> put dt Nil
            (x : xs) -> do
                delaytail <- new
                delaylist <- new
                tail <- new
                put
                    dt
                    ( Fork
                        (next chunkSize delaytail delaylist)
                        (Cons x tail)
                    )
                loop xs tail delaytail delaylist (chunkSize - 2)
    loop :: NFData a => [a] -> Stream a -> Stream a -> IVar [a] -> Int -> Par ()
    loop [] var _ dl _ = do
        put var Nil
        put dl []
    loop (x : xs) var dt dl count =
        if count /= 0
            then do
                tail <- new
                put var (Cons x tail)
                loop xs tail dt dl (count - 1)
            else do
                put var (Cons x dt)
                put dl xs

streamFold :: (a -> b -> a) -> a -> Stream b -> Par a
streamFold fn acc instrm = do
    ilist <- get instrm
    case ilist of
        Nil -> return acc
        Cons h t -> streamFold fn (fn acc h) t
        Fork p Nil -> return acc
        Fork p (Cons h t) -> do
            fork p
            streamFold fn (fn acc h) t

-- >>> runPar $ (streamFromList 3 [1 .. 10]) >>= (streamMap (*2)) >>= (streamFold (+) 0)
-- 110

streamMap :: (NFData a, NFData b) => (a -> b) -> Stream a -> Par (Stream b)
streamMap fn instrm = do
    outstrm <- new
    fork $ init fn instrm outstrm
    return outstrm
  where
    init :: (NFData a, NFData b) => (a -> b) -> Stream a -> Stream b -> Par ()
    init fn instrm outstrm = do
        ilst <- get instrm
        case ilst of
            Nil -> put outstrm Nil
            Cons h t -> do
                newtl <- new
                put outstrm (Cons (fn h) newtl)
                init fn t newtl
            Fork p Nil -> put outstrm Nil
            Fork p (Cons h t) -> do
                fork p
                slist <- get t
                case slist of
                    Nil -> do
                        newtl <- new
                        put newtl Nil
                        put outstrm (Cons (fn h) newtl)
                    Cons h1 t1 -> do
                        newtl <- new
                        delaytail <- new
                        delaystrm <- new
                        put outstrm (Fork (init fn delaystrm delaytail) (Cons (fn h) newtl))
                        loopCons fn h1 t1 newtl delaytail delaystrm
                    Fork p1 Nil -> do
                        delaytail <- new
                        put outstrm (Fork (put delaytail Nil) (Cons (fn h) delaytail))
                    Fork p1 (Cons h1 t1) -> do
                        delaytail <- new
                        delaystrm <- new
                        put outstrm (Fork (init fn delaystrm delaytail) (Cons (fn h) delaytail))
    loopCons :: (NFData a, NFData b) => (a -> b) -> a -> Stream a -> Stream b -> Stream b -> Stream a -> Par ()
    loopCons fn h t var dl ds = do
        tlist <- get t
        case tlist of
            Nil -> do
                newtl <- new
                put newtl Nil
                put var (Cons (fn h) newtl)
                put ds Nil
            Cons h1 t1 -> do
                newtl <- new
                put var (Cons (fn h) newtl)
                loopCons fn h1 t1 newtl dl ds
            Fork p Nil -> do
                newtl <- new
                put newtl Nil
                put var (Cons (fn h) newtl)
            Fork p (Cons h1 t1) -> do
                put ds tlist
                put var (Cons (fn h) dl)