a7d9712 · streamly · git.iko.soy

core/src/Streamly/Internal/Control/Concurrent.hs+1−7

compare with master

unifiedsplit

@@ -23,6 +23,7 @@ module Streamly.Internal.Control.Concurrent

23where

24

25import Control.Monad.Catch (MonadThrow)

26+import Streamly.Internal.Data.SVar.Type (RunInIO(..))

26

27

27#ifdef USE_UNLIFTIO

28#ifdef USE_UNLIFTIO

28import Control.Monad.IO.Unlift (MonadUnliftIO(..), UnliftIO(..), askUnliftIO)

29import Control.Monad.IO.Unlift (MonadUnliftIO(..), UnliftIO(..), askUnliftIO)

@@ -52,13 +53,6 @@ type MonadAsync m = (MonadUnliftIO m, MonadThrow m)

52type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

53type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

53#endif

54#endif

54

55

55−

56−#ifdef USE_UNLIFTIO

57−newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO b }

58−#else

59−newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }

60−#endif

61−

62-- | When we run computations concurrently, we completely isolate the state of

56-- | When we run computations concurrently, we completely isolate the state of

63-- the concurrent computations from the parent computation. The invariant is

57-- the concurrent computations from the parent computation. The invariant is

64-- that we should never be running two concurrent computations in the same

58-- that we should never be running two concurrent computations in the same

core/src/Streamly/Internal/Data/SVar/Type.hs+10−1

@@ -11,6 +11,7 @@ module Streamly.Internal.Data.SVar.Type

11 -- * Parent child communication

12 ThreadAbort (..)

13 , ChildEvent (..)

14+ , RunInIO(..)

14 , AheadHeapEntry (..)

15 , AheadHeapEntry (..)

15

16

16 -- * SVar

17 -- * SVar

@@ -55,6 +56,9 @@ where

55import Control.Concurrent (ThreadId)

56import Control.Concurrent (ThreadId)

56import Control.Concurrent.MVar (MVar)

57import Control.Concurrent.MVar (MVar)

57import Control.Exception (SomeException(..), Exception)

58import Control.Exception (SomeException(..), Exception)

59+#ifndef USE_UNLIFTIO

60+import Control.Monad.Trans.Control (MonadBaseControl(StM))

61+#endif

58import Data.Heap (Heap, Entry(..))

62import Data.Heap (Heap, Entry(..))

59import Data.Int (Int64)

63import Data.Int (Int64)

60import Data.IORef (IORef)

64import Data.IORef (IORef)

@@ -62,7 +66,6 @@ import Data.Kind (Type)

62import Data.Set (Set)

66import Data.Set (Set)

63

67

64import Streamly.Internal.Data.Time.Units (AbsTime, NanoSecond64(..))

68import Streamly.Internal.Data.Time.Units (AbsTime, NanoSecond64(..))

65−import Streamly.Internal.Control.Concurrent (RunInIO)

66

69

67newtype Count = Count Int64

70newtype Count = Count Int64

68 deriving ( Eq

71 deriving ( Eq

@@ -89,6 +92,12 @@ data ChildEvent a =

89 ChildYield a

92 ChildYield a

90 | ChildStop ThreadId (Maybe SomeException)

93 | ChildStop ThreadId (Maybe SomeException)

91

94

95+#ifdef USE_UNLIFTIO

96+newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO b }

97+#else

98+newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }

99+#endif

100+

92-- | Sorting out-of-turn outputs in a heap for Ahead style streams

101-- | Sorting out-of-turn outputs in a heap for Ahead style streams

93data AheadHeapEntry (t :: (Type -> Type) -> Type -> Type) m a =

102data AheadHeapEntry (t :: (Type -> Type) -> Type -> Type) m a =

94 AheadEntryNull

103 AheadEntryNull

src/Streamly/Internal/Data/SVar.hs+1−2

@@ -47,8 +47,7 @@ import Control.Monad.IO.Class (MonadIO(liftIO))

47import Data.Heap (Heap, Entry(..))

48import Data.IORef (newIORef, readIORef)

49import Data.IORef (IORef, atomicModifyIORef)

50−import Streamly.Internal.Control.Concurrent

50+import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)

51− (MonadAsync, askRunInIO, RunInIO)

52import Streamly.Internal.Data.Atomics

51import Streamly.Internal.Data.Atomics

53 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier)

52 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier)

54import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

53import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

src/Streamly/Internal/Data/Stream/Ahead.hs+1−1

@@ -47,7 +47,7 @@ import GHC.Exts (inline)

47import qualified Data.Heap as H

48

49import Streamly.Internal.Control.Concurrent

50− (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO, restoreM)

50+ (MonadRunInIO, MonadAsync, askRunInIO, restoreM)

51import Streamly.Internal.Data.Stream.StreamK.Type (Stream)

52

53import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

src/Streamly/Internal/Data/Stream/Async.hs+1−1

@@ -55,7 +55,7 @@ import Prelude hiding (map)

55import qualified Data.Set as S

56

57import Streamly.Internal.Control.Concurrent

58− (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO, restoreM)

58+ (MonadRunInIO, MonadAsync, askRunInIO, restoreM)

59import Streamly.Internal.Data.Atomics

60 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

61import Streamly.Internal.Data.Stream.StreamK.Type (Stream)