2-- Module : Streamly.Internal.Data.Stream.Concurrent
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- Non-parallelizable stream combinators like unfoldrM, iterateM etc. can be
10-- evaluated concurrently with the stream consumer by using `eval`.
11-- Parallelizable combinators like repeatM, replicateM can generate the stream
12-- concurrently using 'concatMap'.
14-- Single effects related functionality can be moved to
15-- Data.Async/Control.Async.
16-- Common Channel functionality to Data.Channel.
17-- Stream channel to Data.Stream.Channel.
19module Streamly.Internal.Data.Stream.Concurrent
46 -- | Stream combinators using a concurrent channel
49 -- | Evaluates a stream concurrently using a channel.
51 -- Add unfoldrM/iterateM?
54 -- | Uses a single channel to evaluate all actions.
59 -- | Uses a single channel to evaluate all actions.
64 -- | Use a channel for each pair.
72 -- | Shares a single channel across many streams.
81 -- ** Stream of streams
86 -- | Shares a single channel across many streams.
102import Control.Concurrent (myThreadId, killThread)
103import Control.Monad (void, when)
104import Control.Monad.IO.Class (MonadIO(liftIO))
105import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
106import Streamly.Internal.Control.ForkLifted (forkManaged)
107import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)
108import Streamly.Internal.Data.Stream.Channel.Types
112import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)
113import Streamly.Internal.Data.Stream.StreamD.Type (Stream)
114import Streamly.Internal.Data.Stream.StreamD (Step(..))
116import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
117import qualified Streamly.Internal.Data.Stream.StreamD as Stream
118import qualified Streamly.Internal.Data.Stream.StreamD as D
119import qualified Streamly.Internal.Data.Stream.StreamK as K
120import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
122import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)
123import Streamly.Internal.Data.Stream.Concurrent.Channel
127-- Imports for example snippets in this module.
130-- >>> {-# LANGUAGE FlexibleContexts #-}
131-- >>> import Control.Concurrent (threadDelay)
132-- >>> import qualified Streamly.Data.Array as Array
133-- >>> import qualified Streamly.Data.Fold as Fold
134-- >>> import qualified Streamly.Data.Parser as Parser
135-- >>> import qualified Streamly.Data.StreamK as StreamK
136-- >>> import qualified Streamly.Internal.Data.Stream as Stream hiding (append2)
137-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream
138-- >>> import Prelude hiding (concatMap, concat, zipWith)
141-- threadDelay (n * 1000000) -- sleep for n seconds
142-- putStrLn (show n ++ " sec") -- print "n sec"
146-------------------------------------------------------------------------------
147-- Evaluating a stream
148-------------------------------------------------------------------------------
151{-# INLINE_NORMAL parEvalD #-}
152parEvalD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
153parEvalD modifier m = D.Stream step Nothing
157 chan <- newChannel modifier
158 sendFirstWorker chan (D.toStreamK m)
159 -- XXX should use an unfold to make this efficient
160 return $ D.Skip $ Just $ fromChannelD chan
162 step gst (Just (D.UnStream step1 st)) = do
165 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
166 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
170-- | Evaluate a stream asynchronously. In a serial stream, each element of the
171-- stream is generated as it is demanded by the consumer. `parEval` evaluates
172-- multiple elements of the stream ahead of time and serves the results from a
175-- Note that the evaluation requires only one thread as only one stream needs
176-- to be evaluated. Therefore, the concurrency options that are relevant to
177-- multiple streams won't apply here e.g. maxThreads, eager, interleaved,
178-- ordered, stopWhen options won't have any effect.
182-- >>> parUnfoldrM step = Stream.parEval id . Stream.unfoldrM step
183-- >>> parIterateM step = Stream.parEval id . Stream.iterateM step
184{-# INLINE parEval #-}
185parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
186parEval modifier input = withChannel modifier input (const id)
187 -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream
189-------------------------------------------------------------------------------
190-- combining two streams
191-------------------------------------------------------------------------------
193{-# INLINE _appendGeneric #-}
194_appendGeneric :: MonadAsync m =>
195 ((Config -> Config) -> m (Channel m a))
196 -> (Config -> Config)
200_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
205 chan <- newChan modifier
206 let cfg = modifier defaultConfig
207 done = K.nilM (stopChannel chan)
208 case getStopWhen cfg of
210 toChannelK chan stream2
211 toChannelK chan stream1
213 toChannelK chan stream2
214 toChannelK chan (K.append stream1 done)
216 toChannelK chan (K.append stream2 done)
217 toChannelK chan (K.append stream1 done)
218 return $ Stream.toStreamK $ fromChannel chan
220-- | Create a new channel and add both the streams to it for async evaluation.
221-- The output stream is the result of the evaluation.
222{-# INLINE appendWithK #-}
223appendWithK :: MonadAsync m =>
224 (Config -> Config) -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
225appendWithK modifier stream1 stream2 =
227 if getOrdered (modifier defaultConfig)
228 then parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
229 else _appendGeneric Append.newChannel modifier stream1 stream2
231 parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
233-- | Evaluate the first stream in the current thread and add the second stream
234-- to the supplied channel. This is to be used by a worker thread.
236-- This can be used with parConcatMap:
239-- concatMap = K.parConcatMap (_appendWithChanK chan) f stream
242{-# INLINE _appendWithChanK #-}
243_appendWithChanK :: MonadAsync m =>
244 Channel m a -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
245_appendWithChanK chan stream1 stream2 =
246 K.before (toChannelK chan stream2) stream1
248-- | Binary operation to evaluate two streams concurrently using a channel.
250-- If you want to combine more than two streams you almost always want the
251-- 'parList' or `parConcat` operation instead. The performance of this
252-- operation degrades rapidly when more streams are combined as each operation
253-- adds one more concurrent channel. On the other hand, 'parConcat' uses a
254-- single channel for all streams. However, with this operation you can
255-- precisely control the scheduling by creating arbitrary shape expression
260-- >>> parTwo cfg x y = Stream.parList cfg [x, y]
262-- Example, the following code finishes in 4 seconds:
264-- >>> async = Stream.parTwo id
265-- >>> stream1 = Stream.fromEffect (delay 4)
266-- >>> stream2 = Stream.fromEffect (delay 2)
267-- >>> Stream.fold Fold.toList $ stream1 `async` stream2
273parTwo :: MonadAsync m =>
274 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
275parTwo modifier stream1 stream2 =
278 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
280-------------------------------------------------------------------------------
282-------------------------------------------------------------------------------
284-- | A runner function takes a queuing function @q@ and a stream, it splits the
285-- input stream, queuing the tail and using the head to generate a stream.
286-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
287-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
289{-# INLINE mkEnqueue #-}
290mkEnqueue :: MonadAsync m =>
292 -> ((K.StreamK m a -> m ()) -> K.StreamK m a -> K.StreamK m b)
293 -> m (K.StreamK m a -> m ())
294mkEnqueue chan runner = do
295 runInIO <- askRunInIO
298 -- Enqueue the outer loop
299 liftIO $ enqueue chan False (runInIO, runner q stream)
300 -- XXX In case of eager dispatch we can just directly dispatch
301 -- a worker with the tail stream here rather than first queuing
302 -- and then dispatching a worker which dequeues the work. The
303 -- older implementation did a direct dispatch here and its perf
304 -- characterstics looked much better.
308-- | Takes the head element of the input stream and queues the tail of the
309-- stream to the channel, then maps the supplied function on the head and
310-- evaluates the resulting stream.
312-- This function is designed to be used by worker threads on a channel to
313-- concurrently map and evaluate a stream.
314{-# INLINE parConcatMapChanK #-}
315parConcatMapChanK :: MonadAsync m =>
316 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
317parConcatMapChanK chan f stream =
318 let run q = concatMapDivK q f
319 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
320 -- K.parConcatMap (_appendWithChanK chan) f stream
322{-# INLINE parConcatMapChanKAny #-}
323parConcatMapChanKAny :: MonadAsync m =>
324 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
325parConcatMapChanKAny chan f stream =
326 let done = K.nilM (stopChannel chan)
327 run q = concatMapDivK q (\x -> K.append (f x) done)
328 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
330{-# INLINE parConcatMapChanKFirst #-}
331parConcatMapChanKFirst :: MonadAsync m =>
332 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
333parConcatMapChanKFirst chan f stream =
334 let done = K.nilM (stopChannel chan)
335 run q = concatMapDivK q f
336 in K.concatEffect $ do
337 res <- K.uncons stream
339 Nothing -> return K.nil
341 q <- mkEnqueue chan run
343 return $ K.append (f h) done
345{-# INLINE parConcatMapChanKGeneric #-}
346parConcatMapChanKGeneric :: MonadAsync m =>
349 -> (a -> K.StreamK m b)
352parConcatMapChanKGeneric modifier chan f stream = do
353 let cfg = modifier defaultConfig
354 case getStopWhen cfg of
355 AllStop -> parConcatMapChanK chan f stream
356 FirstStops -> parConcatMapChanKFirst chan f stream
357 AnyStops -> parConcatMapChanKAny chan f stream
359-- XXX Add a deep evaluation variant that evaluates individual elements in the
360-- generated streams in parallel.
362-- | Allocate a channel and use it to concurrently evaluate the streams
363-- generated by the mapped function.
365{-# INLINE parConcatMapK #-}
366parConcatMapK :: MonadAsync m =>
367 (Config -> Config) -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
368parConcatMapK modifier f input =
369 let g = parConcatMapChanKGeneric modifier
370 in withChannelK modifier input (`g` f)
372-- | Map each element of the input to a stream and then concurrently evaluate
373-- and concatenate the resulting streams. Multiple streams may be evaluated
374-- concurrently but earlier streams are perferred. Output from the streams are
375-- used as they arrive.
379-- >>> parConcatMap modifier f stream = Stream.parConcat modifier $ fmap f stream
383-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap cfg id $ Stream.fromList xs
385-- The following streams finish in 4 seconds:
387-- >>> stream1 = Stream.fromEffect (delay 4)
388-- >>> stream2 = Stream.fromEffect (delay 2)
389-- >>> stream3 = Stream.fromEffect (delay 1)
390-- >>> f id [stream1, stream2, stream3]
396-- Limiting threads to 2 schedules the third stream only after one of the first
397-- two has finished, releasing a thread:
399-- >>> f (Stream.maxThreads 2) [stream1, stream2, stream3]
403-- When used with a Single thread it behaves like serial concatMap:
405-- >>> f (Stream.maxThreads 1) [stream1, stream2, stream3]
409-- >>> stream1 = Stream.fromList [1,2,3]
410-- >>> stream2 = Stream.fromList [4,5,6]
411-- >>> f (Stream.maxThreads 1) [stream1, stream2]
414-- Schedule all streams in a round robin fashion over the available threads:
416-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs
418-- >>> stream1 = Stream.fromList [1,2,3]
419-- >>> stream2 = Stream.fromList [4,5,6]
420-- >>> f (Stream.maxThreads 1) [stream1, stream2]
423{-# INLINE parConcatMap #-}
424parConcatMap :: MonadAsync m =>
425 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
426parConcatMap modifier f stream =
429 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
431-- | Evaluate the streams in the input stream concurrently and combine them.
433-- >>> parConcat modifier = Stream.parConcatMap modifier id
435{-# INLINE parConcat #-}
436parConcat :: MonadAsync m =>
437 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
438parConcat modifier = parConcatMap modifier id
440-------------------------------------------------------------------------------
442-------------------------------------------------------------------------------
444-- | Like 'parConcat' but works on a list of streams.
446-- >>> parList modifier = Stream.parConcat modifier . Stream.fromList
448{-# INLINE parList #-}
449parList :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
450parList modifier = parConcat modifier . Stream.fromList
452-- | Like 'concat' but works on a list of streams.
454-- >>> parListLazy = Stream.parList id
456{-# INLINE parListLazy #-}
457parListLazy :: MonadAsync m => [Stream m a] -> Stream m a
458parListLazy = parList id
460-- | Like 'parListLazy' but interleaves the streams fairly instead of prioritizing
461-- the left stream. This schedules all streams in a round robin fashion over
462-- limited number of threads.
464-- >>> parListInterleaved = Stream.parList (Stream.interleaved True)
466{-# INLINE parListInterleaved #-}
467parListInterleaved :: MonadAsync m => [Stream m a] -> Stream m a
468parListInterleaved = parList (interleaved True)
470-- | Like 'parListLazy' but with 'ordered' on.
472-- >>> parListOrdered = Stream.parList (Stream.ordered True)
474{-# INLINE parListOrdered #-}
475parListOrdered :: MonadAsync m => [Stream m a] -> Stream m a
476parListOrdered = parList (ordered True)
478-- | Like 'parListLazy' but with 'eager' on.
480-- >>> parListEager = Stream.parList (Stream.eager True)
482{-# INLINE parListEager #-}
483parListEager :: MonadAsync m => [Stream m a] -> Stream m a
484parListEager = parList (eager True)
486-- | Like 'parListEager' but stops the output as soon as the first stream stops.
488-- >>> parListEagerFst = Stream.parList (Stream.eager True . Stream.stopWhen Stream.FirstStops)
490{-# INLINE parListEagerFst #-}
491parListEagerFst :: MonadAsync m => [Stream m a] -> Stream m a
492parListEagerFst = parList (eager True . stopWhen FirstStops)
494-- | Like 'parListEager' but stops the output as soon as any of the two streams
499-- >>> parListEagerMin = Stream.parList (Stream.eager True . Stream.stopWhen Stream.AnyStops)
501{-# INLINE parListEagerMin #-}
502parListEagerMin :: MonadAsync m => [Stream m a] -> Stream m a
503parListEagerMin = parList (eager True . stopWhen AnyStops)
505-------------------------------------------------------------------------------
507-------------------------------------------------------------------------------
509-- | Apply an argument stream to a function stream concurrently. Uses a
510-- shared channel for all individual applications within a stream application.
511{-# INLINE parApply #-}
512{-# SPECIALIZE parApply ::
513 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
514parApply :: MonadAsync m =>
515 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b
516parApply modifier stream1 stream2 =
519 (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)
522-------------------------------------------------------------------------------
524-------------------------------------------------------------------------------
529-- >>> parMapM modifier f = Stream.parConcatMap modifier (Stream.fromEffect . f)
531-- Example, the following example finishes in 1 second as all actions run in
532-- parallel. Even though results are available out of order they are ordered
533-- due to the config option::
535-- >>> f x = delay x >> return x
536-- >>> Stream.fold Fold.toList $ Stream.parMapM (Stream.ordered True) f $ Stream.fromList [3,2,1]
542{-# INLINE parMapM #-}
543parMapM :: MonadAsync m =>
544 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
545parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)
549-- >>> parSequence modifier = Stream.parMapM modifier id
553-- >>> parFromListM = Stream.parSequence id . Stream.fromList
554-- >>> parFromFoldableM = Stream.parSequence id . StreamK.toStream . StreamK.fromFoldable
556{-# INLINE parSequence #-}
557parSequence :: MonadAsync m =>
558 (Config -> Config) -> Stream m (m a) -> Stream m a
559parSequence modifier = parMapM modifier id
561-- | Evaluates the streams being zipped in separate threads than the consumer.
562-- The zip function is evaluated in the consumer thread.
564-- >>> parZipWithM cfg f m1 m2 = Stream.zipWithM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
566-- Multi-stream concurrency options won't apply here, see the notes in
569-- If you want to evaluate the zip function as well in a separate thread, you
570-- can use a 'parEval' on 'parZipWithM'.
572{-# INLINE parZipWithM #-}
573parZipWithM :: MonadAsync m
574 => (Config -> Config)
579parZipWithM cfg f m1 m2 = Stream.zipWithM f (parEval cfg m1) (parEval cfg m2)
582-- >>> parZipWith cfg f = Stream.parZipWithM cfg (\a b -> return $ f a b)
584-- >>> m1 = Stream.fromList [1,2,3]
585-- >>> m2 = Stream.fromList [4,5,6]
586-- >>> Stream.fold Fold.toList $ Stream.parZipWith id (,) m1 m2
587-- [(1,4),(2,5),(3,6)]
589{-# INLINE parZipWith #-}
590parZipWith :: MonadAsync m
591 => (Config -> Config)
596parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
598-- | Like 'mergeByM' but evaluates both the streams concurrently.
602-- >>> parMergeByM cfg f m1 m2 = Stream.mergeByM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
604{-# INLINE parMergeByM #-}
605parMergeByM :: MonadAsync m
606 => (Config -> Config)
607 -> (a -> a -> m Ordering)
611parMergeByM cfg f m1 m2 = Stream.mergeByM f (parEval cfg m1) (parEval cfg m2)
613-- | Like 'mergeBy' but evaluates both the streams concurrently.
617-- >>> parMergeBy cfg f = Stream.parMergeByM cfg (\a b -> return $ f a b)
619{-# INLINE parMergeBy #-}
620parMergeBy :: MonadAsync m
621 => (Config -> Config)
622 -> (a -> a -> Ordering)
626parMergeBy cfg f = parMergeByM cfg (\a b -> return $ f a b)
628-------------------------------------------------------------------------------
630-------------------------------------------------------------------------------
632-- | Same as 'concatIterate' but concurrent.
635{-# INLINE parConcatIterate #-}
636parConcatIterate :: MonadAsync m =>
641parConcatIterate modifier f input =
643 $ withChannelK modifier (Stream.toStreamK input) iterateStream
647 iterateStream channel stream =
648 parConcatMapChanKGeneric modifier channel (generate channel) stream
651 -- XXX The channel q should be FIFO for DFS, otherwise it is BFS
652 x `K.cons` iterateStream channel (Stream.toStreamK $ f x)
654-------------------------------------------------------------------------------
656-------------------------------------------------------------------------------
661-- >>> parRepeatM cfg = Stream.parSequence cfg . Stream.repeat
663-- Generate a stream by repeatedly executing a monadic action forever.
664{-# INLINE parRepeatM #-}
665parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a
666parRepeatM cfg = parSequence cfg . Stream.repeat
668-- | Generate a stream by concurrently performing a monadic action @n@ times.
672-- >>> parReplicateM cfg n = Stream.parSequence cfg . Stream.replicate n
674-- Example, 'parReplicateM' in the following example executes all the
675-- replicated actions concurrently, thus taking only 1 second:
677-- >>> Stream.fold Fold.drain $ Stream.parReplicateM id 10 $ delay 1
680{-# INLINE parReplicateM #-}
681parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a
682parReplicateM cfg n = parSequence cfg . Stream.replicate n
684-------------------------------------------------------------------------------
686-------------------------------------------------------------------------------
688-- Note: we can use another API with two callbacks stop and yield if we want
689-- the callback to be able to indicate end of stream.
691-- | Generates a callback and a stream pair. The callback returned is used to
692-- queue values to the stream. The stream is infinite, there is no way for the
693-- callback to indicate that it is done now.
697{-# INLINE_NORMAL newCallbackStream #-}
698newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
699newCallbackStream = do
700 chan <- newChannel (eager True)
702 -- XXX Add our own thread-id to the SVar as we can not know the callback's
703 -- thread-id and the callback is not run in a managed worker. We need to
704 -- handle this better.
706 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
712 (outputQueue chan) (outputDoorBell chan) (ChildYield a)
713 -- XXX Use fromChannelD?
714 return (callback, fromChannel chan)
716-- | Supplies a stream generating callback to a callback setter function. Each
717-- invocation of the callback results in a value being generated in the
722{-# INLINE fromCallback #-}
723fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> Stream m a
724fromCallback setCallback = Stream.concatEffect $ do
725 (callback, stream) <- newCallbackStream
729{-# INLINE_NORMAL tapCountD #-}
733 -> (D.Stream m Int -> m b)
736tapCountD predicate fld (D.Stream step state) = D.Stream step' Nothing
739 {-# INLINE_LATE step' #-}
741 -- As long as we are using an "Int" for counts lockfree reads from
742 -- Var should work correctly on both 32-bit and 64-bit machines.
743 -- However, an Int on a 32-bit machine may overflow quickly.
744 countVar <- liftIO $ Unboxed.newIORef (0 :: Int)
747 $ Unboxed.toStreamD countVar
748 return $ Skip (Just (countVar, tid, state))
750 step' gst (Just (countVar, tid, st)) = do
755 $ liftIO $ Unboxed.modifyIORef' countVar (+ 1)
756 return $ Yield x (Just (countVar, tid, s))
757 Skip s -> return $ Skip (Just (countVar, tid, s))
759 liftIO $ killThread tid
762-- | @tapCount predicate fold stream@ taps the count of those elements in the
763-- stream that pass the @predicate@. The resulting count stream is sent to
764-- another thread which folds it using @fold@.
766-- For example, to print the count of elements processed every second:
768-- >>> rate = Stream.rollingMap2 (flip (-)) . Stream.delayPost 1
769-- >>> report = Stream.fold (Fold.drainMapM print) . rate
770-- >>> tap = Stream.tapCount (const True) report
771-- >>> go = Stream.fold Fold.drain $ tap $ Stream.enumerateFrom 0
773-- Note: This may not work correctly on 32-bit machines because of Int
778{-# INLINE tapCount #-}
782 -> (Stream m Int -> m b)
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