1{-# LANGUAGE ConstraintKinds #-}
2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE LambdaCase #-}
5{-# LANGUAGE MultiParamTypeClasses #-}
6{-# LANGUAGE RankNTypes #-}
7{-# LANGUAGE UndecidableInstances #-} -- XXX
10-- Module : Streamly.Core
11-- Copyright : (c) 2017 Harendra Kumar
14-- Maintainer : harendra.kumar@gmail.com
15-- Stability : experimental
34 -- * Concurrent Stream Vars (SVars)
46 -- * Concurrent Streams
52import Control.Applicative (Alternative (..))
53import Control.Concurrent (ThreadId, forkIO,
54 myThreadId, threadDelay)
55import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
57import Control.Exception (SomeException (..))
58import qualified Control.Exception.Lifted as EL
59import Control.Monad (MonadPlus(..), mzero, when)
60import Control.Monad.Base (MonadBase (..), liftBaseDefault)
61import Control.Monad.Catch (MonadThrow, throwM)
62import Control.Monad.Error.Class (MonadError(..))
63import Control.Monad.IO.Class (MonadIO(..))
64import Control.Monad.Reader.Class (MonadReader(..))
65import Control.Monad.State.Class (MonadState(..))
66import Control.Monad.Trans.Class (MonadTrans (lift))
67import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
68import Data.Atomics (atomicModifyIORefCAS,
69 atomicModifyIORefCAS_)
70import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
72import Data.Functor (void)
73import Data.IORef (IORef, modifyIORef, newIORef,
75import Data.Maybe (isNothing)
76import Data.Semigroup (Semigroup(..))
78import qualified Data.Set as S
80------------------------------------------------------------------------------
81-- Parent child thread communication type
82------------------------------------------------------------------------------
84-- | Events that a child thread may send to a parent thread.
87 | ChildStop ThreadId (Maybe SomeException)
89------------------------------------------------------------------------------
90-- State threaded around the monad for thread management
91------------------------------------------------------------------------------
93-- | Conjunction is used for monadic/product style composition. Disjunction is
94-- used for fold/sum style composition. We need to distinguish the two types of
95-- SVars so that the scheduling of the two is independent.
96data SVarTag = Conjunction | Disjunction deriving Eq
98-- | For fairly interleaved parallel composition the sched policy is FIFO
99-- whereas for left biased parallel composition it is LIFO.
100data SVarSched = LIFO | FIFO deriving Eq
102-- | Identify the type of the SVar. Two computations using the same style can
103-- be scheduled on the same SVar.
104data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
106-- | An SVar or a Stream Var is a conduit to the output from multiple streams
107-- running concurrently and asynchronously. An SVar can be thought of as an
108-- asynchronous IO handle. We can write any number of streams to an SVar in a
109-- non-blocking manner and then read them back at any time at any pace. The
110-- SVar would run the streams asynchronously and accumulate results. An SVar
111-- may not really execute the stream completely and accumulate all the results.
112-- However, it ensures that the reader can read the results at whatever paces
113-- it wants to read. The SVar monitors and adapts to the consumer's pace.
115-- An SVar is a mini scheduler, it has an associated runqueue that holds the
116-- stream tasks to be picked and run by a pool of worker threads. It has an
117-- associated output queue where the output stream elements are placed by the
118-- worker threads. A doorBell is used by the worker threads to intimate the
119-- consumer thread about availability of new results in the output queue. More
120-- workers are added to the SVar by 'fromStreamVar' on demand if the output
121-- produced is not keeping pace with the consumer. On bounded SVars, workers
122-- block on the output queue to provide throttling of the producer when the
123-- consumer is not pulling fast enough. The number of workers may even get
124-- reduced depending on the consuming pace.
126-- New work is enqueued either at the time of creation of the SVar or as a
127-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
128-- already enqueued computations get evaluated. See 'joinStreamVar2'.
131 SVar { outputQueue :: IORef [ChildEvent a]
132 , doorBell :: MVar Bool -- wakeup mechanism for outQ
133 , enqueue :: Stream m a -> IO ()
135 , runningThreads :: IORef (Set ThreadId)
136 , queueEmpty :: m Bool
137 , svarStyle :: SVarStyle
140------------------------------------------------------------------------------
142------------------------------------------------------------------------------
144-- TBD use a functor instead of the bare type a?
146-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
147-- constructed using actions in monad 'm'. It uses a stop continuation and a
148-- yield continuation. You can consider it a rough equivalent of direct style
151-- data Stream m a = Stop | Yield a (Maybe (Stream m a))
153-- Our goal is to be able to represent finite as well infinite streams and
154-- being able to compose a large number of small streams efficiently. In
155-- addition we want to compose streams in parallel, to facilitate that we
156-- maintain a local state in an SVar that is shared across and is used for
157-- synchronization of the streams being composed.
159-- Using this type, there are two ways to indicate the end of a stream, one is
160-- by calling the stop continuation and the other one is by yielding the last
161-- value along with 'Nothing' as the rest of the stream.
163-- Why do we have this redundancy? Why can't we use (a -> Stream m a -> m r) as
164-- the type of the yield continuation and always use the stop continuation to
165-- indicate the end of the stream? The reason is that when we compose a large
166-- number of short or singleton streams then using the stop continuation
167-- becomes expensive, just to know that there is no next element we have to
168-- call the continuation, introducing an indirection, it seems when using CPS
169-- GHC is not able to optimize this out as efficiently as it can be in direct
170-- style because of the function call involved. In direct style it will just be
171-- a constructor check and a memory access instead of a function call. So we
174-- data Stream m a = Stop | Yield a (Stream m a)
176-- In CPS style, when we use the 'Maybe' argument of yield to indicate the end
177-- then just like direct style we can figure out that there is no next element
178-- without a function call.
180-- Then why not get rid of the stop continuation and use only yield to indicate
181-- the end of stream? The answer is, in that case to indicate the end of the
182-- stream we would have to yield at least one element so there is no way to
183-- represent an empty stream.
185-- Whenever we make a singleton stream or in general when we build a stream
186-- strictly i.e. when we know all the elements of the stream in advance we can
187-- use the last yield to indicate th end of the stream, because we know in
188-- advance at the time of the last yield that the stream is ending. We build
189-- singleton streams in the implementation of 'pure' for Applicative and Monad,
190-- and in 'lift' for MonadTrans, in these places we use yield with 'Nothing' to
191-- indicate the end of the stream. Note that, the only advantage of Maybe is
192-- when we have to build a large number of singleton or short streams. For
193-- larger streams anyway the overhead of a separate stop continuation is not
194-- significant. This could be significant when we breakdown a large stream into
195-- its elements, process them in some way and then recompose it from the
196-- pieces. Zipping streams is one such example. Zipping with streamly is the
197-- fastest among all streaming libraries.
199-- However in a lazy computation we cannot know in advance that the stream is
200-- ending therefore we cannot use 'Maybe', we use the stop continuation in that
201-- case. For example when building a stream from a lazy container using a right
206 runStream :: forall r.
207 Maybe (SVar m a) -- local state
209 -> (a -> Maybe (Stream m a) -> m r) -- yield
213-- | A monad that can perform asynchronous/concurrent IO operations. Streams
214-- that can be composed concurrently require the underlying monad to be
216type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
218scons :: a -> Maybe (Stream m a) -> Stream m a
219scons a r = Stream $ \_ _ yld -> yld a r
221srepeat :: a -> Stream m a
222srepeat a = let x = scons a (Just x) in x
225snil = Stream $ \_ stp _ -> stp
227------------------------------------------------------------------------------
229------------------------------------------------------------------------------
231-- Streams can be composed sequentially or in parallel; in product style
232-- compositions (monadic bind multiplies streams in a ListT fashion) or in sum
233-- style compositions like 'Semigroup', 'Monoid', 'Alternative' or variants of
236------------------------------------------------------------------------------
238------------------------------------------------------------------------------
240-- | '<>' concatenates two streams sequentially i.e. the first stream is
241-- exhausted completely before yielding any element from the second stream.
242instance Semigroup (Stream m a) where
245 go (Stream m) = Stream $ \_ stp yld ->
246 let stop = (runStream m2) Nothing stp yld
247 yield a Nothing = yld a (Just m2)
248 yield a (Just r) = yld a (Just (go r))
249 in m Nothing stop yield
251------------------------------------------------------------------------------
253------------------------------------------------------------------------------
255instance Monoid (Stream m a) where
256 mempty = Stream $ \_ stp _ -> stp
259------------------------------------------------------------------------------
261------------------------------------------------------------------------------
264interleave :: Stream m a -> Stream m a -> Stream m a
265interleave m1 m2 = Stream $ \_ stp yld -> do
266 let stop = (runStream m2) Nothing stp yld
267 yield a Nothing = yld a (Just m2)
268 yield a (Just r) = yld a (Just (interleave m2 r))
269 (runStream m1) Nothing stop yield
271------------------------------------------------------------------------------
272-- Spawning threads and collecting result in streamed fashion
273------------------------------------------------------------------------------
276doFork :: MonadBaseControl IO m
278 -> (SomeException -> m ())
280doFork action exHandler =
281 EL.mask $ \restore ->
282 liftBaseWith $ \runInIO -> forkIO $ do
283 -- XXX test the exception handling
284 _ <- runInIO $ EL.catch (restore action) exHandler
285 -- XXX restore state here?
288-- XXX exception safety of all atomic/MVar operations
291send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
292send sv msg = liftIO $ do
293 atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
294 -- XXX need a memory barrier? The wake up must happen only after the
295 -- store has finished otherwise we can have lost wakeup problems.
296 void $ tryPutMVar (doorBell sv) True
298{-# INLINE sendStop #-}
299sendStop :: MonadIO m => SVar m a -> m ()
300sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
302-- Note: Left associated compositions can grow this queue to a large size
303{-# INLINE enqueueLIFO #-}
304enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
305enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
307runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
308runqueueLIFO sv q = run
315 Nothing -> sendStop sv
316 Just m -> (runStream m) (Just sv) run yield
318 sendit a = send sv (ChildYield a)
319 yield a Nothing = sendit a >> run
320 yield a (Just r) = sendit a >> (runStream r) (Just sv) run yield
322 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
324 x : xs -> (xs, Just x)
326{-# INLINE enqueueFIFO #-}
327enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
330runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
331runqueueFIFO sv q = run
338 Nothing -> sendStop sv
339 Just m -> (runStream m) (Just sv) run yield
341 dequeue = liftIO $ tryPopR q
342 sendit a = send sv (ChildYield a)
343 yield a Nothing = sendit a >> run
344 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
346-- Thread tracking is needed for two reasons:
348-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
349-- themselves because they may run for significant times before going away or
350-- worse they may be stuck in IO and never go away.
352-- 2) To know when all threads are done.
354{-# NOINLINE addThread #-}
355addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
357 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
359{-# INLINE delThread #-}
360delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
362 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
364{-# INLINE allThreadsDone #-}
365allThreadsDone :: MonadIO m => SVar m a -> m Bool
366allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
368{-# NOINLINE handleChildException #-}
369handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
370handleChildException sv e = do
371 tid <- liftIO myThreadId
372 send sv (ChildStop tid (Just e))
374{-# NOINLINE pushWorker #-}
375pushWorker :: MonadAsync m => SVar m a -> m ()
377 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
379-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
380-- Also, if a worker blocks on the output queue we can decide if we want to
381-- block or make it go away entirely, depending on the number of workers and
382-- the type of the queue.
383{-# INLINE sendWorkerWait #-}
384sendWorkerWait :: MonadAsync m => SVar m a -> m ()
385sendWorkerWait sv = do
387 SVarStyle _ LIFO -> liftIO $ threadDelay 200
388 SVarStyle _ FIFO -> liftIO $ threadDelay 0
390 output <- liftIO $ readIORef (outputQueue sv)
391 when (null output) $ do
392 done <- queueEmpty sv
394 then pushWorker sv >> sendWorkerWait sv
395 else void (liftIO $ takeMVar (doorBell sv))
397-- | Pull a stream from an SVar.
398{-# NOINLINE fromStreamVar #-}
399fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
400fromStreamVar sv = Stream $ \_ stp yld -> do
401 -- XXX if reading the IORef is costly we can use a flag in the SVar to
402 -- indicate we are done.
403 done <- allThreadsDone sv
407 res <- liftIO $ tryTakeMVar (doorBell sv)
408 when (isNothing res) $ sendWorkerWait sv
409 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
410 -- To avoid lock overhead we read all events at once instead of reading
411 -- one at a time. We just reverse the list to process the events in the
412 -- order they arrived. Maybe we can use a queue instead?
413 (runStream $ processEvents (reverse list)) Nothing stp yld
417 handleException e tid = do
419 -- XXX implement kill async exception handling
420 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
423 {-# INLINE processEvents #-}
424 processEvents [] = Stream $ \_ stp yld -> do
425 done <- allThreadsDone sv
427 then (runStream (fromStreamVar sv)) Nothing stp yld
430 processEvents (ev : es) = Stream $ \_ stp yld -> do
431 let continue = (runStream (processEvents es)) Nothing stp yld
432 yield a = yld a (Just (processEvents es))
435 ChildYield a -> yield a
438 Nothing -> delThread sv tid >> continue
439 Just ex -> handleException ex tid
441getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
442getFifoSVar ctype = do
444 outQMv <- newEmptyMVar
445 running <- newIORef S.empty
448 SVar { outputQueue = outQ
450 , runningThreads = running
451 , runqueue = runqueueFIFO sv q
453 , queueEmpty = liftIO $ nullQ q
458getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
459getLifoSVar ctype = do
461 outQMv <- newEmptyMVar
462 running <- newIORef S.empty
464 let checkEmpty = null <$> liftIO (readIORef q)
466 SVar { outputQueue = outQ
468 , runningThreads = running
469 , runqueue = runqueueLIFO sv q
470 , enqueue = enqueueLIFO q
471 , queueEmpty = checkEmpty
476-- | Create a new empty SVar.
477newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
478newEmptySVar style = do
481 SVarStyle _ FIFO -> getFifoSVar style
482 SVarStyle _ LIFO -> getLifoSVar style
484-- | Create a new SVar and enqueue one stream computation on it.
485newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
486newStreamVar1 style m = do
487 sv <- newEmptySVar style
488 -- Note: We must have all the work on the queue before sending the
489 -- pushworker, otherwise the pushworker may exit before we even get a
491 liftIO $ (enqueue sv) m
495-- | Create a new SVar and enqueue two stream computations on it.
496newStreamVar2 :: MonadAsync m
497 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
498newStreamVar2 style m1 m2 = do
499 -- Note: We must have all the work on the queue before sending the
500 -- pushworker, otherwise the pushworker may exit before we even get a
504 SVarStyle _ FIFO -> do
505 c <- getFifoSVar style
506 (enqueue c) m1 >> (enqueue c) m2
508 SVarStyle _ LIFO -> do
509 c <- getLifoSVar style
510 (enqueue c) m2 >> (enqueue c) m1
515-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
516-- be read back from the SVar using 'fromSVar'.
517toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
519 liftIO $ (enqueue sv) m
520 done <- allThreadsDone sv
521 -- XXX there may be a race here unless we are running in the consumer
522 -- thread. This is safe only when called from the consumer thread or when
523 -- no consumer is present.
524 when done $ pushWorker sv
526------------------------------------------------------------------------------
527-- Running streams concurrently
528------------------------------------------------------------------------------
530-- Concurrency rate control. Our objective is to create more threads on demand
531-- if the consumer is running faster than us. As soon as we encounter an
532-- Alternative composition we create a push pull pair of threads. We use a
533-- channel for communication between the consumer pulling from the channel and
534-- the producer who pushing to the channel. The producer creates more threads
535-- if no output is seen on the channel, that is the consumer is running faster.
536-- However this mechanism can be problematic if the initial production latency
537-- is high, we may end up creating too many threads. So we need some way to
538-- monitor and use the latency as well.
540-- TBD We may run computations at the lower level of the composition tree
541-- serially even if they are composed using a parallel combinator. We can use
542-- <> in place of <| and <=> in place of <|>. If we find that a parallel
543-- channel immediately above a computation becomes empty we can switch to
544-- parallelizing the computation. For that we can use a state flag to fork the
545-- rest of the computation at any point of time inside the Monad bind operation
546-- if the consumer is running at a faster speed.
548-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
549-- If we have to dispatch in arbitrary chunksizes we will need to compose the
550-- parallel actions using a data constructor (Free Alternative) instead so that
551-- we can divide it in chunks of arbitrary size before dispatch. If the stream
552-- is composed of hierarchically composed grains of different sizes then we can
553-- always switch to a desired granularity depending on the consumer speed.
555-- TBD for pure work (when we are not in the IO monad) we can divide it into
556-- just the number of CPUs.
558{-# NOINLINE withNewSVar2 #-}
559withNewSVar2 :: MonadAsync m
560 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
561withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
562 sv <- newStreamVar2 style m1 m2
563 (runStream (fromStreamVar sv)) Nothing stp yld
565-- | Join two computations on the currently running 'SVar' queue for concurrent
566-- execution. The 'SVarStyle' required by the current composition context is
567-- passed as one of the parameters. If the style does not match with the style
568-- of the current 'SVar' we create a new 'SVar' and schedule the computations
569-- on that. The newly created SVar joins as one of the computations on the
570-- current SVar queue.
572-- When we are using parallel composition, an SVar is passed around as a state
573-- variable. We try to schedule a new parallel computation on the SVar passed
574-- to us. The first time, when no SVar exists, a new SVar is created.
575-- Subsequently, 'joinStreamVar2' may get called when a computation already
576-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
577-- evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on the current scheduler
578-- queue. However, if the scheduling and composition style of the new
579-- computation being scheduled is different than the style of the current SVar,
580-- then we create a new SVar and schedule it on that.
584-- * (x \<|> y) \<|> (t \<|> u) -- all of them get scheduled on the same SVar
585-- * (x \<|> y) \<|> (t \<| u) -- @t@ and @u@ get scheduled on a new child SVar
586-- because of the scheduling policy change.
587-- * if we 'adapt' a stream of type 'AsyncT' to a stream of type
588-- 'ParallelT', we create a new SVar at the transitioning bind.
589-- * When the stream is switching from disjunctive composition to conjunctive
590-- composition and vice-versa we create a new SVar to isolate the scheduling
593{-# INLINE joinStreamVar2 #-}
594joinStreamVar2 :: MonadAsync m
595 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
596joinStreamVar2 style m1 m2 = Stream $ \st stp yld ->
598 Just sv | svarStyle sv == style ->
599 liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
600 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp yld
602------------------------------------------------------------------------------
603-- Semigroup and Monoid style compositions for parallel actions
604------------------------------------------------------------------------------
608parAhead :: Stream m a -> Stream m a -> Stream m a
611-- | Sequential composition similar to '<>' except that it can execute the
612-- action on the right in parallel ahead of time. Returns the results in
613-- sequential order like '<>' from left to right.
614(<>|) :: Stream m a -> Stream m a -> Stream m a
618-- | Same as '<|>'. Since this schedules all the composed streams fairly you
619-- cannot fold infinite number of streams using this operation.
621parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
622parAlt = joinStreamVar2 (SVarStyle Disjunction FIFO)
624-- | Same as '<|'. Since this schedules the left side computation first you can
625-- right fold an infinite container using this operator. However a left fold
626-- will not work well as it first unpeels the whole structure before scheduling
627-- a computation requiring an amount of memory proportional to the size of the
629{-# INLINE parLeft #-}
630parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
631parLeft = joinStreamVar2 (SVarStyle Disjunction LIFO)
633-------------------------------------------------------------------------------
634-- Instances (only used for deriving newtype instances)
635-------------------------------------------------------------------------------
637-- Stream type is not exposed, these instances are only for deriving instances
638-- for the newtype wrappers based on Stream.
640-- Dummy Instances, defined to enable the definition of other instances that
641-- require a Monad constraint. Must be defined by the newtypes.
643instance Monad m => Functor (Stream m) where
646instance Monad m => Applicative (Stream m) where
650instance Monad m => Monad (Stream m) where
654------------------------------------------------------------------------------
655-- Alternative & MonadPlus
656------------------------------------------------------------------------------
658-- | `empty` represents an action that takes non-zero time to complete. Since
659-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
660-- monoidal composition executing all actions in parallel, it is similar to
661-- '<>' except that it runs all the actions in parallel and interleaves their
663instance MonadAsync m => Alternative (Stream m) where
667instance MonadAsync m => MonadPlus (Stream m) where
671-------------------------------------------------------------------------------
673-------------------------------------------------------------------------------
675instance MonadTrans Stream where
676 lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))
678instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where
679 liftBase = liftBaseDefault
681------------------------------------------------------------------------------
682-- Standard transformer instances
683------------------------------------------------------------------------------
685instance MonadIO m => MonadIO (Stream m) where
686 liftIO = lift . liftIO
688instance MonadThrow m => MonadThrow (Stream m) where
689 throwM = lift . throwM
691-- XXX handle and test cross thread state transfer
692instance MonadError e m => MonadError e (Stream m) where
693 throwError = lift . throwError
694 catchError m h = Stream $ \st stp yld ->
695 let handle r = r `catchError` \e -> (runStream (h e)) st stp yld
696 yield a Nothing = yld a Nothing
697 yield a (Just r) = yld a (Just (catchError r h))
698 in handle $ (runStream m) st stp yield
700instance MonadReader r m => MonadReader r (Stream m) where
702 local f m = Stream $ \st stp yld ->
703 let yield a Nothing = local f $ yld a Nothing
704 yield a (Just r) = local f $ yld a (Just (local f r))
705 in (runStream m) st (local f stp) yield
707instance MonadState s m => MonadState s (Stream m) where
710 state k = lift (state k)
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list