2b0548b · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Remove EndOfStream to make fromSVar behave like fromHandle

2b0548bHarendra Kumar9 years ago

src/Asyncly.hs−1

compare with master

unifiedsplit

@@ -82,7 +82,6 @@ module Asyncly

82 , runZipAsync

84 -- * Transformation

85− , EndOfStream

86 , async

85 , async

88 -- * Sum Style Composition

87 -- * Sum Style Composition

src/Asyncly/Core.hs+11−15

@@ -29,7 +29,6 @@ module Asyncly.Core

29 , SVarSched (..)

30 , SVarTag (..)

31 , SVarStyle (..)

32− , EndOfStream (..)

33 , newEmptySVar

32 , newEmptySVar

34 , newStreamVar1

33 , newStreamVar1

35 , newStreamVar2

34 , newStreamVar2

@@ -48,7 +47,7 @@ import Control.Concurrent (ThreadId, forkIO,

48 myThreadId, threadDelay)

47 myThreadId, threadDelay)

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

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

50 tryPutMVar, takeMVar)

49 tryPutMVar, takeMVar)

51−import Control.Exception (SomeException (..), Exception)

50+import Control.Exception (SomeException (..))

52import qualified Control.Exception.Lifted as EL

51import qualified Control.Exception.Lifted as EL

53import Control.Monad (MonadPlus(..), mzero, when)

52import Control.Monad (MonadPlus(..), mzero, when)

54import Control.Monad.Base (MonadBase (..), liftBaseDefault)

53import Control.Monad.Base (MonadBase (..), liftBaseDefault)

@@ -321,10 +320,6 @@ sendWorkerWait sv = do

321 then (pushWorker sv) >> sendWorkerWait sv

320 then (pushWorker sv) >> sendWorkerWait sv

322 else void (liftIO $ takeMVar (doorBell sv))

321 else void (liftIO $ takeMVar (doorBell sv))

323

322

324−-- | A stream being pulled from 'SVar' has ended.

325−data EndOfStream = EndOfStream deriving Show

326−instance Exception EndOfStream

327−

328-- | Pull a stream from an SVar.

323-- | Pull a stream from an SVar.

329{-# NOINLINE fromStreamVar #-}

324{-# NOINLINE fromStreamVar #-}

330fromStreamVar :: MonadAsync m => SVar m a -> Stream m a

325fromStreamVar :: MonadAsync m => SVar m a -> Stream m a

@@ -332,15 +327,16 @@ fromStreamVar sv = Stream $ \_ stp yld -> do

332 -- XXX if reading the IORef is costly we can use a flag in the SVar to

327 -- XXX if reading the IORef is costly we can use a flag in the SVar to

333 -- indicate we are done.

328 -- indicate we are done.

334 done <- allThreadsDone sv

329 done <- allThreadsDone sv

335− when done $ throwM EndOfStream

330+ if done

336−

331+ then stp

332+ else do

337− res <- liftIO $ tryTakeMVar (doorBell sv)

333+ res <- liftIO $ tryTakeMVar (doorBell sv)

338− when (isNothing res) $ sendWorkerWait sv

334+ when (isNothing res) $ sendWorkerWait sv

339− list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)

335+ list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)

340− -- To avoid lock overhead we read all events at once instead of reading one

336+ -- To avoid lock overhead we read all events at once instead of reading

341− -- at a time. We just reverse the list to process the events in the order

337+ -- one at a time. We just reverse the list to process the events in the

342− -- they arrived. Maybe we can use a queue instead?

338+ -- order they arrived. Maybe we can use a queue instead?

343− (runStream $ processEvents (reverse list)) Nothing stp yld

339+ (runStream $ processEvents (reverse list)) Nothing stp yld

344

340

345 where

341 where

346

342

src/Asyncly/Streams.hs+4−6

@@ -24,7 +24,6 @@ module Asyncly.Streams

24 , MonadAsync

26 -- * SVars

27− , EndOfStream (..)

28 , SVarSched (..)

27 , SVarSched (..)

29 , SVarTag (..)

28 , SVarTag (..)

30 , SVarStyle (..)

29 , SVarStyle (..)

@@ -761,8 +760,7 @@ streamBuild k = fromStream $ Stream $ \sv stp yld ->

761fromCallback :: (Streaming t) => (forall r. (a -> m r) -> m r) -> t m a

760fromCallback :: (Streaming t) => (forall r. (a -> m r) -> m r) -> t m a

762fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)

761fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)

763

762

764−-- | Convert an SVar to a stream. Throws 'EndOfStream' if the SVar is accessed

763+-- | Read an SVar to get a stream.

765−-- even after it has been fully drained.

766fromSVar :: (MonadAsync m, Streaming t) => SVar m a -> t m a

764fromSVar :: (MonadAsync m, Streaming t) => SVar m a -> t m a

767fromSVar sv = fromStream $ fromStreamVar sv

765fromSVar sv = fromStream $ fromStreamVar sv

768

766

@@ -830,9 +828,9 @@ runZipAsync = runStreaming

830-- XXX Get rid of this?

828-- XXX Get rid of this?

831-- | Make a stream asynchronous, triggers the computation and returns a stream

829-- | Make a stream asynchronous, triggers the computation and returns a stream

832-- in the underlying monad representing the output generated by the original

830-- in the underlying monad representing the output generated by the original

833−-- computation. The returned action must be drained once and only once. If not

831+-- computation. The returned action is exhaustible and must be drained once. If

834−-- drained fully we may have a thread blocked forever and once drained it will

832+-- not drained fully we may have a thread blocked forever and once exhausted it

835−-- throw an 'EndOfStream' exception if we try to run it again

833+-- will always return 'empty'.

836

834

837async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)

835async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)

838async m = do

836async m = do