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
33 -- * Concurrent Stream Vars (SVars)
45 -- * Concurrent Streams
51import Control.Applicative (Alternative (..))
52import Control.Concurrent (ThreadId, forkIO,
53 myThreadId, threadDelay)
54import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
56import Control.Exception (SomeException (..))
57import qualified Control.Exception.Lifted as EL
58import Control.Monad (MonadPlus(..), mzero, when)
59import Control.Monad.Base (MonadBase (..), liftBaseDefault)
60import Control.Monad.Catch (MonadThrow, throwM)
61import Control.Monad.Error.Class (MonadError(..))
62import Control.Monad.IO.Class (MonadIO(..))
63import Control.Monad.Reader.Class (MonadReader(..))
64import Control.Monad.State.Class (MonadState(..))
65import Control.Monad.Trans.Class (MonadTrans (lift))
66import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
67import Data.Atomics (atomicModifyIORefCAS,
68 atomicModifyIORefCAS_)
69import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
71import Data.Functor (void)
72import Data.IORef (IORef, modifyIORef, newIORef,
74import Data.Maybe (isNothing)
75import Data.Semigroup (Semigroup(..))
77import qualified Data.Set as S
79------------------------------------------------------------------------------
80-- Parent child thread communication type
81------------------------------------------------------------------------------
83-- | Events that a child thread may send to a parent thread.
86 | ChildStop ThreadId (Maybe SomeException)
88------------------------------------------------------------------------------
89-- State threaded around the monad for thread management
90------------------------------------------------------------------------------
92-- | Conjunction is used for monadic/product style composition. Disjunction is
93-- used for fold/sum style composition. We need to distinguish the two types of
94-- SVars so that the scheduling of the two is independent.
95data SVarTag = Conjunction | Disjunction deriving Eq
97-- | For fairly interleaved parallel composition the sched policy is FIFO
98-- whereas for left biased parallel composition it is LIFO.
99data SVarSched = LIFO | FIFO deriving Eq
101-- | Identify the type of the SVar. Two computations using the same style can
102-- be scheduled on the same SVar.
103data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
105-- | An SVar or a Stream Var is a conduit to the output from multiple streams
106-- running concurrently and asynchronously. An SVar can be thought of as an
107-- asynchronous IO handle. We can write any number of streams to an SVar in a
108-- non-blocking manner and then read them back at any time at any pace. The
109-- SVar would run the streams asynchronously and accumulate results. An SVar
110-- may not really execute the stream completely and accumulate all the results.
111-- However, it ensures that the reader can read the results at whatever paces
112-- it wants to read. The SVar monitors and adapts to the consumer's pace.
114-- An SVar is a mini scheduler, it has an associated runqueue that holds the
115-- stream tasks to be picked and run by a pool of worker threads. It has an
116-- associated output queue where the output stream elements are placed by the
117-- worker threads. A doorBell is used by the worker threads to intimate the
118-- consumer thread about availability of new results in the output queue. More
119-- workers are added to the SVar by 'fromStreamVar' on demand if the output
120-- produced is not keeping pace with the consumer. On bounded SVars, workers
121-- block on the output queue to provide throttling of the producer when the
122-- consumer is not pulling fast enough. The number of workers may even get
123-- reduced depending on the consuming pace.
125-- New work is enqueued either at the time of creation of the SVar or as a
126-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
127-- already enqueued computations get evaluated. See 'joinStreamVar2'.
130 SVar { outputQueue :: IORef [ChildEvent a]
131 , doorBell :: MVar Bool -- wakeup mechanism for outQ
132 , enqueue :: Stream m a -> IO ()
134 , runningThreads :: IORef (Set ThreadId)
135 , queueEmpty :: m Bool
136 , svarStyle :: SVarStyle
139------------------------------------------------------------------------------
141------------------------------------------------------------------------------
143-- TBD use a functor instead of the bare type a?
145-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
146-- constructed using actions in monad 'm'. It uses a stop continuation and a
147-- yield continuation. You can consider it a rough equivalent of direct style
150-- data Stream m a = Stop | Yield a (Maybe (Stream m a))
152-- Our goal is to be able to represent finite as well infinite streams and
153-- being able to compose a large number of small streams efficiently. In
154-- addition we want to compose streams in parallel, to facilitate that we
155-- maintain a local state in an SVar that is shared across and is used for
156-- synchronization of the streams being composed.
158-- Using this type, there are two ways to indicate the end of a stream, one is
159-- by calling the stop continuation and the other one is by yielding the last
160-- value along with 'Nothing' as the rest of the stream.
162-- Why do we have this redundancy? Why can't we use (a -> Stream m a -> m r) as
163-- the type of the yield continuation and always use the stop continuation to
164-- indicate the end of the stream? The reason is that when we compose a large
165-- number of short or singleton streams then using the stop continuation
166-- becomes expensive, just to know that there is no next element we have to
167-- call the continuation, introducing an indirection, it seems when using CPS
168-- GHC is not able to optimize this out as efficiently as it can be in direct
169-- style because of the function call involved. In direct style it will just be
170-- a constructor check and a memory access instead of a function call. So we
173-- data Stream m a = Stop | Yield a (Stream m a)
175-- In CPS style, when we use the 'Maybe' argument of yield to indicate the end
176-- then just like direct style we can figure out that there is no next element
177-- without a function call.
179-- Then why not get rid of the stop continuation and use only yield to indicate
180-- the end of stream? The answer is, in that case to indicate the end of the
181-- stream we would have to yield at least one element so there is no way to
182-- represent an empty stream.
184-- Whenever we make a singleton stream or in general when we build a stream
185-- strictly i.e. when we know all the elements of the stream in advance we can
186-- use the last yield to indicate th end of the stream, because we know in
187-- advance at the time of the last yield that the stream is ending. We build
188-- singleton streams in the implementation of 'pure' for Applicative and Monad,
189-- and in 'lift' for MonadTrans, in these places we use yield with 'Nothing' to
190-- indicate the end of the stream. Note that, the only advantage of Maybe is
191-- when we have to build a large number of singleton or short streams. For
192-- larger streams anyway the overhead of a separate stop continuation is not
193-- significant. This could be significant when we breakdown a large stream into
194-- its elements, process them in some way and then recompose it from the
195-- pieces. Zipping streams is one such example. Zipping with streamly is the
196-- fastest among all streaming libraries.
198-- However in a lazy computation we cannot know in advance that the stream is
199-- ending therefore we cannot use 'Maybe', we use the stop continuation in that
200-- case. For example when building a stream from a lazy container using a right
205 runStream :: forall r.
206 Maybe (SVar m a) -- local state
208 -> (a -> Maybe (Stream m a) -> m r) -- yield
212-- | A monad that can perform asynchronous/concurrent IO operations. Streams
213-- that can be composed concurrently require the underlying monad to be
215type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
217scons :: a -> Maybe (Stream m a) -> Stream m a
218scons a r = Stream $ \_ _ yld -> yld a r
221snil = Stream $ \_ stp _ -> stp
223------------------------------------------------------------------------------
225------------------------------------------------------------------------------
227-- Streams can be composed sequentially or in parallel; in product style
228-- compositions (monadic bind multiplies streams in a ListT fashion) or in sum
229-- style compositions like 'Semigroup', 'Monoid', 'Alternative' or variants of
232------------------------------------------------------------------------------
234------------------------------------------------------------------------------
236-- | '<>' concatenates two streams sequentially i.e. the first stream is
237-- exhausted completely before yielding any element from the second stream.
238instance Semigroup (Stream m a) where
241 go (Stream m) = Stream $ \_ stp yld ->
242 let stop = (runStream m2) Nothing stp yld
243 yield a Nothing = yld a (Just m2)
244 yield a (Just r) = yld a (Just (go r))
245 in m Nothing stop yield
247------------------------------------------------------------------------------
249------------------------------------------------------------------------------
251instance Monoid (Stream m a) where
252 mempty = Stream $ \_ stp _ -> stp
255------------------------------------------------------------------------------
257------------------------------------------------------------------------------
260interleave :: Stream m a -> Stream m a -> Stream m a
261interleave m1 m2 = Stream $ \_ stp yld -> do
262 let stop = (runStream m2) Nothing stp yld
263 yield a Nothing = yld a (Just m2)
264 yield a (Just r) = yld a (Just (interleave m2 r))
265 (runStream m1) Nothing stop yield
267------------------------------------------------------------------------------
268-- Spawning threads and collecting result in streamed fashion
269------------------------------------------------------------------------------
272doFork :: MonadBaseControl IO m
274 -> (SomeException -> m ())
276doFork action exHandler =
277 EL.mask $ \restore ->
278 liftBaseWith $ \runInIO -> forkIO $ do
279 -- XXX test the exception handling
280 _ <- runInIO $ EL.catch (restore action) exHandler
281 -- XXX restore state here?
284-- XXX exception safety of all atomic/MVar operations
287send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
288send sv msg = liftIO $ do
289 atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
290 -- XXX need a memory barrier? The wake up must happen only after the
291 -- store has finished otherwise we can have lost wakeup problems.
292 void $ tryPutMVar (doorBell sv) True
294{-# INLINE sendStop #-}
295sendStop :: MonadIO m => SVar m a -> m ()
296sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
298-- Note: Left associated compositions can grow this queue to a large size
299{-# INLINE enqueueLIFO #-}
300enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
301enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
303runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
304runqueueLIFO sv q = run
311 Nothing -> sendStop sv
312 Just m -> (runStream m) (Just sv) run yield
314 sendit a = send sv (ChildYield a)
315 yield a Nothing = sendit a >> run
316 yield a (Just r) = sendit a >> (runStream r) (Just sv) run yield
318 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
320 x : xs -> (xs, Just x)
322{-# INLINE enqueueFIFO #-}
323enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
326runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
327runqueueFIFO sv q = run
334 Nothing -> sendStop sv
335 Just m -> (runStream m) (Just sv) run yield
337 dequeue = liftIO $ tryPopR q
338 sendit a = send sv (ChildYield a)
339 yield a Nothing = sendit a >> run
340 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
342-- Thread tracking is needed for two reasons:
344-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
345-- themselves because they may run for significant times before going away or
346-- worse they may be stuck in IO and never go away.
348-- 2) To know when all threads are done.
350{-# NOINLINE addThread #-}
351addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
353 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
355{-# INLINE delThread #-}
356delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
358 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
360{-# INLINE allThreadsDone #-}
361allThreadsDone :: MonadIO m => SVar m a -> m Bool
362allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
364{-# NOINLINE handleChildException #-}
365handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
366handleChildException sv e = do
367 tid <- liftIO myThreadId
368 send sv (ChildStop tid (Just e))
370{-# NOINLINE pushWorker #-}
371pushWorker :: MonadAsync m => SVar m a -> m ()
373 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
375-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
376-- Also, if a worker blocks on the output queue we can decide if we want to
377-- block or make it go away entirely, depending on the number of workers and
378-- the type of the queue.
379{-# INLINE sendWorkerWait #-}
380sendWorkerWait :: MonadAsync m => SVar m a -> m ()
381sendWorkerWait sv = do
383 SVarStyle _ LIFO -> liftIO $ threadDelay 200
384 SVarStyle _ FIFO -> liftIO $ threadDelay 0
386 output <- liftIO $ readIORef (outputQueue sv)
387 when (null output) $ do
388 done <- queueEmpty sv
390 then pushWorker sv >> sendWorkerWait sv
391 else void (liftIO $ takeMVar (doorBell sv))
393-- | Pull a stream from an SVar.
394{-# NOINLINE fromStreamVar #-}
395fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
396fromStreamVar sv = Stream $ \_ stp yld -> do
397 -- XXX if reading the IORef is costly we can use a flag in the SVar to
398 -- indicate we are done.
399 done <- allThreadsDone sv
403 res <- liftIO $ tryTakeMVar (doorBell sv)
404 when (isNothing res) $ sendWorkerWait sv
405 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
406 -- To avoid lock overhead we read all events at once instead of reading
407 -- one at a time. We just reverse the list to process the events in the
408 -- order they arrived. Maybe we can use a queue instead?
409 (runStream $ processEvents (reverse list)) Nothing stp yld
413 handleException e tid = do
415 -- XXX implement kill async exception handling
416 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
419 {-# INLINE processEvents #-}
420 processEvents [] = Stream $ \_ stp yld -> do
421 done <- allThreadsDone sv
423 then (runStream (fromStreamVar sv)) Nothing stp yld
426 processEvents (ev : es) = Stream $ \_ stp yld -> do
427 let continue = (runStream (processEvents es)) Nothing stp yld
428 yield a = yld a (Just (processEvents es))
431 ChildYield a -> yield a
434 Nothing -> delThread sv tid >> continue
435 Just ex -> handleException ex tid
437getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
438getFifoSVar ctype = do
440 outQMv <- newEmptyMVar
441 running <- newIORef S.empty
444 SVar { outputQueue = outQ
446 , runningThreads = running
447 , runqueue = runqueueFIFO sv q
449 , queueEmpty = liftIO $ nullQ q
454getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
455getLifoSVar ctype = do
457 outQMv <- newEmptyMVar
458 running <- newIORef S.empty
460 let checkEmpty = null <$> liftIO (readIORef q)
462 SVar { outputQueue = outQ
464 , runningThreads = running
465 , runqueue = runqueueLIFO sv q
466 , enqueue = enqueueLIFO q
467 , queueEmpty = checkEmpty
472-- | Create a new empty SVar.
473newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
474newEmptySVar style = do
477 SVarStyle _ FIFO -> getFifoSVar style
478 SVarStyle _ LIFO -> getLifoSVar style
480-- | Create a new SVar and enqueue one stream computation on it.
481newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
482newStreamVar1 style m = do
483 sv <- newEmptySVar style
484 -- Note: We must have all the work on the queue before sending the
485 -- pushworker, otherwise the pushworker may exit before we even get a
487 liftIO $ (enqueue sv) m
491-- | Create a new SVar and enqueue two stream computations on it.
492newStreamVar2 :: MonadAsync m
493 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
494newStreamVar2 style m1 m2 = do
495 -- Note: We must have all the work on the queue before sending the
496 -- pushworker, otherwise the pushworker may exit before we even get a
500 SVarStyle _ FIFO -> do
501 c <- getFifoSVar style
502 (enqueue c) m1 >> (enqueue c) m2
504 SVarStyle _ LIFO -> do
505 c <- getLifoSVar style
506 (enqueue c) m2 >> (enqueue c) m1
511-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
512-- be read back from the SVar using 'fromSVar'.
513toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
515 liftIO $ (enqueue sv) m
516 done <- allThreadsDone sv
517 -- XXX there may be a race here unless we are running in the consumer
518 -- thread. This is safe only when called from the consumer thread or when
519 -- no consumer is present.
520 when done $ pushWorker sv
522------------------------------------------------------------------------------
523-- Running streams concurrently
524------------------------------------------------------------------------------
526-- Concurrency rate control. Our objective is to create more threads on demand
527-- if the consumer is running faster than us. As soon as we encounter an
528-- Alternative composition we create a push pull pair of threads. We use a
529-- channel for communication between the consumer pulling from the channel and
530-- the producer who pushing to the channel. The producer creates more threads
531-- if no output is seen on the channel, that is the consumer is running faster.
532-- However this mechanism can be problematic if the initial production latency
533-- is high, we may end up creating too many threads. So we need some way to
534-- monitor and use the latency as well.
536-- TBD We may run computations at the lower level of the composition tree
537-- serially even if they are composed using a parallel combinator. We can use
538-- <> in place of <| and <=> in place of <|>. If we find that a parallel
539-- channel immediately above a computation becomes empty we can switch to
540-- parallelizing the computation. For that we can use a state flag to fork the
541-- rest of the computation at any point of time inside the Monad bind operation
542-- if the consumer is running at a faster speed.
544-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
545-- If we have to dispatch in arbitrary chunksizes we will need to compose the
546-- parallel actions using a data constructor (Free Alternative) instead so that
547-- we can divide it in chunks of arbitrary size before dispatch. If the stream
548-- is composed of hierarchically composed grains of different sizes then we can
549-- always switch to a desired granularity depending on the consumer speed.
551-- TBD for pure work (when we are not in the IO monad) we can divide it into
552-- just the number of CPUs.
554{-# NOINLINE withNewSVar2 #-}
555withNewSVar2 :: MonadAsync m
556 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
557withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
558 sv <- newStreamVar2 style m1 m2
559 (runStream (fromStreamVar sv)) Nothing stp yld
561-- | Join two computations on the currently running 'SVar' queue for concurrent
562-- execution. The 'SVarStyle' required by the current composition context is
563-- passed as one of the parameters. If the style does not match with the style
564-- of the current 'SVar' we create a new 'SVar' and schedule the computations
565-- on that. The newly created SVar joins as one of the computations on the
566-- current SVar queue.
568-- When we are using parallel composition, an SVar is passed around as a state
569-- variable. We try to schedule a new parallel computation on the SVar passed
570-- to us. The first time, when no SVar exists, a new SVar is created.
571-- Subsequently, 'joinStreamVar2' may get called when a computation already
572-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
573-- evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on the current scheduler
574-- queue. However, if the scheduling and composition style of the new
575-- computation being scheduled is different than the style of the current SVar,
576-- then we create a new SVar and schedule it on that.
580-- * (x \<|> y) \<|> (t \<|> u) -- all of them get scheduled on the same SVar
581-- * (x \<|> y) \<|> (t \<| u) -- @t@ and @u@ get scheduled on a new child SVar
582-- because of the scheduling policy change.
583-- * if we 'adapt' a stream of type 'AsyncT' to a stream of type
584-- 'ParallelT', we create a new SVar at the transitioning bind.
585-- * When the stream is switching from disjunctive composition to conjunctive
586-- composition and vice-versa we create a new SVar to isolate the scheduling
589{-# INLINE joinStreamVar2 #-}
590joinStreamVar2 :: MonadAsync m
591 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
592joinStreamVar2 style m1 m2 = Stream $ \st stp yld ->
594 Just sv | svarStyle sv == style ->
595 liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
596 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp yld
598------------------------------------------------------------------------------
599-- Semigroup and Monoid style compositions for parallel actions
600------------------------------------------------------------------------------
604parAhead :: Stream m a -> Stream m a -> Stream m a
607-- | Sequential composition similar to '<>' except that it can execute the
608-- action on the right in parallel ahead of time. Returns the results in
609-- sequential order like '<>' from left to right.
610(<>|) :: Stream m a -> Stream m a -> Stream m a
614-- | Same as '<|>'. Since this schedules all the composed streams fairly you
615-- cannot fold infinite number of streams using this operation.
617parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
618parAlt = joinStreamVar2 (SVarStyle Disjunction FIFO)
620-- | Same as '<|'. Since this schedules the left side computation first you can
621-- right fold an infinite container using this operator. However a left fold
622-- will not work well as it first unpeels the whole structure before scheduling
623-- a computation requiring an amount of memory proportional to the size of the
625{-# INLINE parLeft #-}
626parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
627parLeft = joinStreamVar2 (SVarStyle Disjunction LIFO)
629-------------------------------------------------------------------------------
630-- Instances (only used for deriving newtype instances)
631-------------------------------------------------------------------------------
633-- Stream type is not exposed, these instances are only for deriving instances
634-- for the newtype wrappers based on Stream.
636-- Dummy Instances, defined to enable the definition of other instances that
637-- require a Monad constraint. Must be defined by the newtypes.
639instance Monad m => Functor (Stream m) where
642instance Monad m => Applicative (Stream m) where
646instance Monad m => Monad (Stream m) where
650------------------------------------------------------------------------------
651-- Alternative & MonadPlus
652------------------------------------------------------------------------------
654-- | `empty` represents an action that takes non-zero time to complete. Since
655-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
656-- monoidal composition executing all actions in parallel, it is similar to
657-- '<>' except that it runs all the actions in parallel and interleaves their
659instance MonadAsync m => Alternative (Stream m) where
663instance MonadAsync m => MonadPlus (Stream m) where
667-------------------------------------------------------------------------------
669-------------------------------------------------------------------------------
671instance MonadTrans Stream where
672 lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))
674instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where
675 liftBase = liftBaseDefault
677------------------------------------------------------------------------------
678-- Standard transformer instances
679------------------------------------------------------------------------------
681instance MonadIO m => MonadIO (Stream m) where
682 liftIO = lift . liftIO
684instance MonadThrow m => MonadThrow (Stream m) where
685 throwM = lift . throwM
687-- XXX handle and test cross thread state transfer
688instance MonadError e m => MonadError e (Stream m) where
689 throwError = lift . throwError
690 catchError m h = Stream $ \st stp yld ->
691 let handle r = r `catchError` \e -> (runStream (h e)) st stp yld
692 yield a Nothing = yld a Nothing
693 yield a (Just r) = yld a (Just (catchError r h))
694 in handle $ (runStream m) st stp yield
696instance MonadReader r m => MonadReader r (Stream m) where
698 local f m = Stream $ \st stp yld ->
699 let yield a Nothing = local f $ yld a Nothing
700 yield a (Just r) = local f $ yld a (Just (local f r))
701 in (runStream m) st (local f stp) yield
703instance MonadState s m => MonadState s (Stream m) where
706 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