2b0548b · streamly · git.iko.soy

git.iko.soy/streamly

Remove EndOfStream to make fromSVar behave like fromHandle

commit2b0548bb34c6bd01de43d5fb5865665aeed69c5a authorHarendra Kumar <harendra.kumar@gmail.com> dateSat 28 Oct 2017, 19:16 (9 years ago) parentd845877

3 files changed+15−22

compare with master

unifiedsplit

src/Asyncly.hs−1 src/Asyncly/Core.hs+11−15 src/Asyncly/Streams.hs+4−6

src/Asyncly.hs−1

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

8282 , runZipAsync

8383

8484 -- * Transformation

85− , EndOfStream

8685 , async

8786

8887 -- * Sum Style Composition

src/Asyncly/Core.hs+11−15

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

2929 , SVarSched (..)

3030 , SVarTag (..)

3131 , SVarStyle (..)

32− , EndOfStream (..)

3332 , newEmptySVar

3433 , newStreamVar1

3534 , newStreamVar2

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

4847 myThreadId, threadDelay)

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

5049 tryPutMVar, takeMVar)

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

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

5251import qualified Control.Exception.Lifted as EL

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

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

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

321320 then (pushWorker sv) >> sendWorkerWait sv

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

323322

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

325−data EndOfStream = EndOfStream deriving Show

326−instance Exception EndOfStream

327−

328323-- | Pull a stream from an SVar.

329324{-# NOINLINE fromStreamVar #-}

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

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

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

333328 -- indicate we are done.

334329 done <- allThreadsDone sv

335− when done $ throwM EndOfStream

336−

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

338− when (isNothing res) $ sendWorkerWait sv

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

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

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

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

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

330+ if done

331+ then stp

332+ else do

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

334+ when (isNothing res) $ sendWorkerWait sv

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

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

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

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

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

344340

345341 where

346342

src/Asyncly/Streams.hs+4−6

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

2424 , MonadAsync

2525

2626 -- * SVars

27− , EndOfStream (..)

2827 , SVarSched (..)

2928 , SVarTag (..)

3029 , SVarStyle (..)

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

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

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

763762

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

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

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

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

767765fromSVar sv = fromStream $ fromStreamVar sv

768766

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

830828-- XXX Get rid of this?

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

832830-- 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

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

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

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

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

833+-- will always return 'empty'.

836834

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

838836async m = do