6804880 · streamly · git.iko.soy

src/Streamly/Internal/Control/ForkLifted.hs+17−4

compare with master

unifiedsplit

@@ -9,12 +9,13 @@

9module Streamly.Internal.Control.ForkLifted

10 (

11 doFork

12+ , doFork'

12 , fork

13 , fork

13 , forkManaged

14 , forkManaged

14 )

15 )

15where

16where

16

17

17−import Control.Concurrent (ThreadId, forkIO)

18+import Control.Concurrent (ThreadId, forkIO, forkOS)

18import Control.Exception (SomeException(..), catch, mask)

19import Control.Exception (SomeException(..), catch, mask)

19import Data.Functor (void)

20import Data.Functor (void)

20import Streamly.Internal.Control.Concurrent (MonadRunInIO, RunInIO(..), withRunInIO, withRunInIONoRestore)

21import Streamly.Internal.Control.Concurrent (MonadRunInIO, RunInIO(..), withRunInIO, withRunInIONoRestore)

@@ -32,11 +33,23 @@ doFork :: MonadRunInIO m

32 -> RunInIO m

33 -> RunInIO m

33 -> (SomeException -> IO ())

34 -> (SomeException -> IO ())

34 -> m ThreadId

35 -> m ThreadId

36+doFork = doFork' False

37+

38+-- | Similar to 'doFork', but has a "bound" boolean parameter for specifying

39+-- whether 'forkOS' should be used instead of 'rawForkIO'.

40+{-# INLINE doFork' #-}

41+doFork' :: MonadRunInIO m

42+ => Bool

43+ -> m ()

44+ -> RunInIO m

45+ -> (SomeException -> IO ())

46+ -> m ThreadId

35−doFork action (RunInIO mrun) exHandler =

47+doFork' bound action (RunInIO mrun) exHandler =

36 withRunInIO $ \run ->

48 withRunInIO $ \run ->

37 mask $ \restore -> do

49 mask $ \restore -> do

50+ tid <- (if bound then forkOS else rawForkIO) $

38− tid <- rawForkIO $ catch (restore $ void $ mrun action)

51+ catch (restore $ void $ mrun action)

39− exHandler

52+ exHandler

40 run (return tid)

53 run (return tid)

41

54

42-- | 'fork' lifted to any monad with 'MonadBaseControl IO m' capability.

55-- | 'fork' lifted to any monad with 'MonadBaseControl IO m' capability.

src/Streamly/Internal/Data/Fold/Concurrent.hs+1

@@ -24,6 +24,7 @@ module Streamly.Internal.Data.Fold.Concurrent

24 -- * Configuration

25 Config

26 , maxBuffer

27+ , bound

27 , inspect

28 , inspect

28

29

29 -- * Combinators

30 -- * Combinators

src/Streamly/Internal/Data/Fold/Concurrent/Channel.hs+1

@@ -14,6 +14,7 @@ module Streamly.Internal.Data.Fold.Concurrent.Channel

14 -- * Configuration

15 , Config

16 , maxBuffer

17+ , bound

17 , inspect

18 , inspect

18

19

19 -- * Fold operations

20 -- * Fold operations

src/Streamly/Internal/Data/Fold/Concurrent/Channel/Type.hs+4−3

@@ -28,7 +28,7 @@ import Data.IORef (IORef, newIORef, readIORef)

28import Data.List (intersperse)

29import Streamly.Internal.Control.Concurrent

30 (MonadAsync, MonadRunInIO, askRunInIO)

31−import Streamly.Internal.Control.ForkLifted (doFork)

31+import Streamly.Internal.Control.ForkLifted (doFork')

32import Streamly.Internal.Data.Fold (Fold(..))

33import Streamly.Internal.Data.Stream.Channel.Dispatcher (dumpSVarStats)

34import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)

@@ -236,9 +236,10 @@ mkNewChannel cfg = do

236newChannel :: (MonadRunInIO m) =>

237 (Config -> Config) -> Fold m a b -> m (Channel m a b)

238newChannel modifier f = do

239+ let config = modifier defaultConfig

239− sv <- liftIO $ mkNewChannel (modifier defaultConfig)

240+ sv <- liftIO $ mkNewChannel config

240 mrun <- askRunInIO

241 mrun <- askRunInIO

241− void $ doFork (work sv) mrun (sendExceptionToDriver sv)

242+ void $ doFork' (getBound config) (work sv) mrun (sendExceptionToDriver sv)

242 return sv

243 return sv

243

244

244 where

245 where

src/Streamly/Internal/Data/Stream/Channel/Types.hs+12

@@ -76,6 +76,7 @@ module Streamly.Internal.Data.Stream.Channel.Types

76 , stopWhen

77 , ordered

78 , interleaved

79+ , bound

79

80

80 , rate

81 , rate

81 , avgRate

82 , avgRate

@@ -95,6 +96,7 @@ module Streamly.Internal.Data.Stream.Channel.Types

95 , getStopWhen

96 , getStopWhen

96 , getOrdered

97 , getOrdered

97 , getInterleaved

98 , getInterleaved

99+ , getBound

98

100

99 -- * Cleanup

101 -- * Cleanup

100 , cleanupSVar

102 , cleanupSVar

@@ -340,6 +342,7 @@ data Config = Config

340 , _stopWhen :: StopWhen

342 , _stopWhen :: StopWhen

341 , _ordered :: Bool

343 , _ordered :: Bool

342 , _interleaved :: Bool

344 , _interleaved :: Bool

345+ , _bound :: Bool

343 }

346 }

344

347

345-------------------------------------------------------------------------------

348-------------------------------------------------------------------------------

@@ -374,6 +377,7 @@ defaultConfig = Config

374 , _stopWhen = AllStop

377 , _stopWhen = AllStop

375 , _ordered = False

378 , _ordered = False

376 , _interleaved = False

379 , _interleaved = False

380+ , _bound = False

377 }

381 }

378

382

379-------------------------------------------------------------------------------

383-------------------------------------------------------------------------------

@@ -542,6 +546,14 @@ interleaved flag st = st { _interleaved = flag }

542getInterleaved :: Config -> Bool

546getInterleaved :: Config -> Bool

543getInterleaved = _interleaved

547getInterleaved = _interleaved

544

548

549+-- | Spawn bound threads (i.e., spawn threads using 'forkOS' instead of

550+-- 'forkIO'). The default value is 'False'.

551+bound :: Bool -> Config -> Config

552+bound flag st = st { _bound = flag }

553+

554+getBound :: Config -> Bool

555+getBound = _bound

556+

545-------------------------------------------------------------------------------

557-------------------------------------------------------------------------------

546-- Initialization

558-- Initialization

547-------------------------------------------------------------------------------

559-------------------------------------------------------------------------------