d765444 · streamly · git.iko.soy

git.iko.soy/streamly

Remove lifted-base dependency

Also got nice perf bump for parallel composition due to simpler fork

commitd76544497a24d1e884cd47601c44f0a7c5fe3e2f authorHarendra Kumar <harendra.kumar@gmail.com> dateSun 27 May 2018, 08:55 (8 years ago) parente603579

3 files changed+35−40

compare with master

unifiedsplit

src/Streamly/Core.hs+10−36 src/Streamly/Streams.hs+22−2 streamly.cabal+3−2

src/Streamly/Core.hs+10−36

@@ -56,11 +56,6 @@ module Streamly.Core

5656 , zipWith

5757 , zipAsyncWith

5858

59− -- XXX move to Streams.hs

60− -- * Transformers

61− , withLocal

62− , withCatchError

63−

6459 -- * Concurrent Stream Vars (SVars)

6560 , SVar

6661 , SVarStyle (..)

@@ -73,15 +68,12 @@ where

7368import Control.Concurrent (ThreadId, myThreadId, threadDelay)

7469import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,

7570 tryPutMVar, takeMVar, putMVar)

76−import Control.Exception (SomeException (..))

77−import qualified Control.Exception.Lifted as EL

71+import Control.Exception (SomeException (..), catch, mask)

7872import Control.Monad (when)

7973import Control.Monad.Catch (MonadThrow, throwM)

80−import Control.Monad.Error.Class (MonadError(..))

8174import Control.Monad.IO.Class (MonadIO(..))

82−import Control.Monad.Reader.Class (MonadReader(..))

8375import Control.Monad.Trans.Class (MonadTrans (lift))

84−import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)

76+import Control.Monad.Trans.Control (MonadBaseControl, control)

8577import Data.Atomics (casIORef, readForCAS, peekTicket

8678 ,atomicModifyIORefCAS_)

8779import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,

@@ -395,14 +387,14 @@ rawForkIO action = IO $ \ s ->

395387{-# INLINE doFork #-}

396388doFork :: MonadBaseControl IO m

397389 => m ()

398− -> (SomeException -> m ())

390+ -> (SomeException -> IO ())

399391 -> m ThreadId

400392doFork action exHandler =

401− EL.mask $ \restore ->

402− liftBaseWith $ \runInIO -> rawForkIO $ do

403− _ <- runInIO $ EL.catch (restore action) exHandler

404− -- XXX restore state here?

405− return ()

393+ control $ \runInIO ->

394+ mask $ \restore -> do

395+ tid <- rawForkIO $ catch (restore $ void $ runInIO action)

396+ exHandler

397+ runInIO (return tid)

406398

407399-- XXX exception safety of all atomic/MVar operations

408400

@@ -808,8 +800,8 @@ allThreadsDone :: MonadIO m => SVar m a -> m Bool

808800allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)

809801

810802{-# NOINLINE handleChildException #-}

811−handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()

812−handleChildException sv e = liftIO $ do

803+handleChildException :: SVar m a -> SomeException -> IO ()

804+handleChildException sv e = do

813805 tid <- myThreadId

814806 send sv (ChildStop tid (Just e))

815807

@@ -1335,21 +1327,3 @@ zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do

13351327

13361328instance MonadTrans Stream where

13371329 lift = once

1338−

1339−withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a

1340−withLocal f m =

1341− Stream $ \svr stp sng yld ->

1342− let single = local f . sng

1343− yield a r = local f $ yld a (withLocal f r)

1344− in (runStream m) svr (local f stp) single yield

1345−

1346−-- XXX handle and test cross thread state transfer

1347−withCatchError

1348− :: MonadError e m

1349− => Stream m a -> (e -> Stream m a) -> Stream m a

1350−withCatchError m h =

1351− Stream $ \svr stp sng yld ->

1352− let run x = runStream x svr stp sng yield

1353− handle r = r `catchError` \e -> run $ h e

1354− yield a r = yld a (withCatchError r h)

1355− in handle $ run m

src/Streamly/Streams.hs+22−2

@@ -404,6 +404,26 @@ x |> f = f <| x

404404-- XXX use template haskell instead and include Monoid and IsStream instances

405405-- as well.

406406

407+withLocal :: MonadReader r m => (r -> r) -> S.Stream m a -> S.Stream m a

408+withLocal f m =

409+ S.Stream $ \svr stp sng yld ->

410+ let single = local f . sng

411+ yield a r = local f $ yld a (withLocal f r)

412+ in (S.runStream m) svr (local f stp) single yield

413+

414+{-

415+-- XXX handle and test cross thread state transfer

416+withCatchError

417+ :: MonadError e m

418+ => S.Stream m a -> (e -> S.Stream m a) -> S.Stream m a

419+withCatchError m h =

420+ S.Stream $ \svr stp sng yld ->

421+ let run x = S.runStream x svr stp sng yield

422+ handle r = r `catchError` \e -> run $ h e

423+ yield a r = yld a (withCatchError r h)

424+ in handle $ run m

425+-}

426+

407427#define MONADPARALLEL , MonadAsync m

408428

409429#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \

@@ -425,12 +445,12 @@ instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \

425445instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \

426446 throwError = lift . throwError; \

427447 catchError m h = \

428− fromStream $ S.withCatchError (toStream m) (\e -> toStream $ h e) }; \

448+ fromStream $ withCatchError (toStream m) (\e -> toStream $ h e) }; \

429449-} \

430450 \

431451instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \

432452 ask = lift ask; \

433− local f m = fromStream $ S.withLocal f (toStream m) }; \

453+ local f m = fromStream $ withLocal f (toStream m) }; \

434454 \

435455instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \

436456 get = lift get; \

streamly.cabal+3−2

@@ -118,14 +118,15 @@ library

118118 -Wnoncanonical-monadfail-instances

119119

120120 build-depends: base >= 4.8 && < 5

121− , atomic-primops >= 0.8 && < 0.9

122121 , containers >= 0.5 && < 0.6

123122 , heaps >= 0.3 && < 0.4

123+

124+ -- concurrency

125+ , atomic-primops >= 0.8 && < 0.9

124126 , lockfree-queue >= 0.2.3 && < 0.3

125127

126128 -- transfomers

127129 , exceptions >= 0.8 && < 0.11

128− , lifted-base >= 0.2 && < 0.3

129130 , monad-control >= 1.0 && < 2

130131 , mtl >= 2.2 && < 3

131132 , transformers >= 0.4 && < 0.6