git.iko.soy/streamly
on master help sign in
Add bound config param & expose for concurrent folds
commit6804880b99e1828b94ef3e6f687525b0b35f24fa authorShlok Datye <shlok@quant.is> dateMon 1 May 2023, 19:09 (3 years ago) parentf1b5a21
5 files changed+35−7
compare with master
unifiedsplit
src/Streamly/Internal/Control/ForkLifted.hs+17−4 src/Streamly/Internal/Data/Fold/Concurrent.hs+1 src/Streamly/Internal/Data/Fold/Concurrent/Channel.hs+1 src/Streamly/Internal/Data/Fold/Concurrent/Channel/Type.hs+4−3 src/Streamly/Internal/Data/Stream/Channel/Types.hs+12
src/Streamly/Internal/Control/ForkLifted.hs+17−4
@@ -9,12 +9,13 @@
99module Streamly.Internal.Control.ForkLifted
1010 (
1111 doFork
12+ , doFork'
1213 , fork
1314 , forkManaged
1415 )
1516where
1617
17−import Control.Concurrent (ThreadId, forkIO)
18+import Control.Concurrent (ThreadId, forkIO, forkOS)
1819import Control.Exception (SomeException(..), catch, mask)
1920import Data.Functor (void)
2021import Streamly.Internal.Control.Concurrent (MonadRunInIO, RunInIO(..), withRunInIO, withRunInIONoRestore)
@@ -32,11 +33,23 @@ doFork :: MonadRunInIO m
3233 -> RunInIO m
3334 -> (SomeException -> IO ())
3435 -> m ThreadId
35−doFork action (RunInIO mrun) exHandler =
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
47+doFork' bound action (RunInIO mrun) exHandler =
3648 withRunInIO $ \run ->
3749 mask $ \restore -> do
38− tid <- rawForkIO $ catch (restore $ void $ mrun action)
39− exHandler
50+ tid <- (if bound then forkOS else rawForkIO) $
51+ catch (restore $ void $ mrun action)
52+ exHandler
4053 run (return tid)
4154
4255-- | '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
2424 -- * Configuration
2525 Config
2626 , maxBuffer
27+ , bound
2728 , inspect
2829
2930 -- * Combinators
src/Streamly/Internal/Data/Fold/Concurrent/Channel.hs+1
@@ -14,6 +14,7 @@ module Streamly.Internal.Data.Fold.Concurrent.Channel
1414 -- * Configuration
1515 , Config
1616 , maxBuffer
17+ , bound
1718 , inspect
1819
1920 -- * Fold operations
src/Streamly/Internal/Data/Fold/Concurrent/Channel/Type.hs+4−3
@@ -28,7 +28,7 @@ import Data.IORef (IORef, newIORef, readIORef)
2828import Data.List (intersperse)
2929import Streamly.Internal.Control.Concurrent
3030 (MonadAsync, MonadRunInIO, askRunInIO)
31−import Streamly.Internal.Control.ForkLifted (doFork)
31+import Streamly.Internal.Control.ForkLifted (doFork')
3232import Streamly.Internal.Data.Fold (Fold(..))
3333import Streamly.Internal.Data.Stream.Channel.Dispatcher (dumpSVarStats)
3434import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)
@@ -236,9 +236,10 @@ mkNewChannel cfg = do
236236newChannel :: (MonadRunInIO m) =>
237237 (Config -> Config) -> Fold m a b -> m (Channel m a b)
238238newChannel modifier f = do
239− sv <- liftIO $ mkNewChannel (modifier defaultConfig)
239+ let config = modifier defaultConfig
240+ sv <- liftIO $ mkNewChannel config
240241 mrun <- askRunInIO
241− void $ doFork (work sv) mrun (sendExceptionToDriver sv)
242+ void $ doFork' (getBound config) (work sv) mrun (sendExceptionToDriver sv)
242243 return sv
243244
244245 where
src/Streamly/Internal/Data/Stream/Channel/Types.hs+12
@@ -76,6 +76,7 @@ module Streamly.Internal.Data.Stream.Channel.Types
7676 , stopWhen
7777 , ordered
7878 , interleaved
79+ , bound
7980
8081 , rate
8182 , avgRate
@@ -95,6 +96,7 @@ module Streamly.Internal.Data.Stream.Channel.Types
9596 , getStopWhen
9697 , getOrdered
9798 , getInterleaved
99+ , getBound
98100
99101 -- * Cleanup
100102 , cleanupSVar
@@ -340,6 +342,7 @@ data Config = Config
340342 , _stopWhen :: StopWhen
341343 , _ordered :: Bool
342344 , _interleaved :: Bool
345+ , _bound :: Bool
343346 }
344347
345348-------------------------------------------------------------------------------
@@ -374,6 +377,7 @@ defaultConfig = Config
374377 , _stopWhen = AllStop
375378 , _ordered = False
376379 , _interleaved = False
380+ , _bound = False
377381 }
378382
379383-------------------------------------------------------------------------------
@@ -542,6 +546,14 @@ interleaved flag st = st { _interleaved = flag }
542546getInterleaved :: Config -> Bool
543547getInterleaved = _interleaved
544548
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+
545557-------------------------------------------------------------------------------
546558-- Initialization
547559-------------------------------------------------------------------------------