500d889 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Simplify push-pull split, fix monoid instance

500d889Harendra Kumar9 years ago

src/Asyncly/AsyncT.hs+22−26

compare with master

unifiedsplit

@@ -106,8 +106,8 @@ type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

106instance Monad m => Monoid (AsyncT m a) where

107 mempty = AsyncT $ \_ stp _ -> stp

108 mappend (AsyncT m1) m2 = AsyncT $ \ctx stp yld ->

109− m1 ctx ((runAsyncT m2) ctx stp yld) $ \a c r ->

109+ m1 ctx ((runAsyncT m2) ctx stp yld) $ \a _ r ->

110− let yield x = yld a c (Just x)

110+ let yield x = yld a ctx (Just x)

111 in maybe (yield m2) (\rx -> yield $ mappend rx m2) r

112

113-- We do not use bind for parallelism. That is, we do not start each iteration

@@ -152,7 +152,7 @@ instance Monad m => Applicative (AsyncT m) where

152-- Alternative

153------------------------------------------------------------------------------

154

155−{-# NOINLINE doFork #-}

155+{-# INLINE doFork #-}

156doFork :: (MonadIO m, MonadBaseControl IO m)

157 => m ()

158 -> (SomeException -> IO ())

@@ -160,6 +160,7 @@ doFork :: (MonadIO m, MonadBaseControl IO m)

160doFork action preExit =

161 EL.mask $ \restore ->

162 liftBaseWith $ \runInIO -> forkIO $ do

163+ -- XXX test the exception handling

163 _ <- runInIO $ EL.catch (restore action) (liftIO . preExit)

164 _ <- runInIO $ EL.catch (restore action) (liftIO . preExit)

164 -- XXX restore state here?

165 -- XXX restore state here?

165 return ()

166 return ()

@@ -219,20 +220,21 @@ pull ctx = AsyncT $ \_ stp yld -> do

219 Just BlockedIndefinitelyOnSTM -> stp

220 Just BlockedIndefinitelyOnSTM -> stp

220 Nothing -> throwM e

221 Nothing -> throwM e

221 Right ev ->

222 Right ev ->

223+ let yield a = yld a Nothing (Just (pull ctx))

222− case ev of

224+ in case ev of

223− ChildYield a -> yld a Nothing (Just (pull ctx))

225+ ChildYield a -> yield a

224− ChildDone _ a -> yld a Nothing (Just (pull ctx))

226+ ChildDone _ a -> yield a

225 ChildStop _ e -> do

227 ChildStop _ e -> do

226 let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld

228 let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld

227− maybe continue throwM e

229+ in maybe continue throwM e

228

230

229-- | Split the original computation in a pull-push pair. The original

231-- | Split the original computation in a pull-push pair. The original

230-- computation pulls from a Channel while m1 and m2 push to the channel.

232-- computation pulls from a Channel while m1 and m2 push to the channel.

231−{-# INLINE pullFork #-}

233+{-# NOINLINE pullFork #-}

232pullFork :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a

234pullFork :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a

233pullFork m1 m2 = AsyncT $ \_ stp yld -> do

235pullFork m1 m2 = AsyncT $ \_ stp yld -> do

234 c <- liftIO $ newContext

236 c <- liftIO $ newContext

235− _ <- doFork (pushBoth c m1 m2) (handlePushException (childChannel c))

237+ _ <- doFork (push c (m1 <|> m2)) (handlePushException (childChannel c))

236 (runAsyncT (pull c)) (Just c) stp yld

238 (runAsyncT (pull c)) (Just c) stp yld

237

239

238 where

240 where

@@ -241,13 +243,8 @@ pullFork m1 m2 = AsyncT $ \_ stp yld -> do

241 childChan <- atomically newTChan

243 childChan <- atomically newTChan

242 return $ Context { childChannel = childChan }

244 return $ Context { childChannel = childChan }

243

245

244− pushBoth c ma mb = do

245− _ <- doFork (push c ma) (handlePushException (childChannel c))

246− -- XXX push from m2 only when the channel becomes empty

247− push c mb

248−

249canFork :: IO Bool

246canFork :: IO Bool

250−canFork = return True

247+canFork = return False

251

248

252-- Concurrency rate control. Our objective is to create more threads on

249-- Concurrency rate control. Our objective is to create more threads on

253-- demand if the consumer is running faster than us. As soon as we

250-- demand if the consumer is running faster than us. As soon as we

@@ -264,21 +261,20 @@ canFork = return True

264-- the rest of the computation at any point of time inside the Monad bind

261-- the rest of the computation at any point of time inside the Monad bind

265-- operation if the consumer is running at a faster speed.

262-- operation if the consumer is running at a faster speed.

266--

263--

267−{-# NOINLINE fork #-}

268−fork :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a

269−fork m1 m2 = AsyncT $ \ctx stp yld -> do

270− case ctx of

271− Nothing -> (runAsyncT (pullFork m1 m2)) ctx stp yld

272− Just c -> (runAsyncT (pushFork c m1 m2)) ctx stp yld

273−

274instance MonadAsync m => Alternative (AsyncT m) where

264instance MonadAsync m => Alternative (AsyncT m) where

275 empty = mempty

265 empty = mempty

276

266

267+ -- XXX pullFork can have more than 2x penalty for smallish tasks. Can we

268+ -- avoid this automatically if the task is small?

269+ {-# INLINE (<|>) #-}

277 m1 <|> m2 = AsyncT $ \ctx stp yld -> do

270 m1 <|> m2 = AsyncT $ \ctx stp yld -> do

271+ case ctx of

272+ Nothing -> (runAsyncT (pullFork m1 m2)) ctx stp yld

273+ Just c -> do

278− doAsync <- liftIO $ canFork

274+ doAsync <- liftIO $ canFork

279− if doAsync then

275+ if doAsync then

280− (runAsyncT $ fork m1 m2) ctx stp yld

276+ (runAsyncT (pushFork c m1 m2)) ctx stp yld

281− else (runAsyncT (m1 <> m2)) ctx stp yld

277+ else (runAsyncT (m1 <> m2)) ctx stp yld

282

278

283instance MonadAsync m => MonadPlus (AsyncT m) where

279instance MonadAsync m => MonadPlus (AsyncT m) where

284 mzero = empty

280 mzero = empty

stack.yaml+1−1

@@ -1,4 +1,4 @@

1−resolver: lts-7.7

1+resolver: lts-8.23

2packages:

3- '.'

4- location:

test/Main.hs+4−2

@@ -20,6 +20,8 @@ main = hspec $ do

20 toList (return 0) `shouldReturn` ([0] :: [Int])

21 it "Monoid composition" $

22 toList (return 0 <> return 1) `shouldReturn` ([0, 1] :: [Int])

23+ it "Monoid composition - each" $

24+ ((toList $ each [1..100]) >>= return . sort) `shouldReturn` ([1..100] :: [Int])

23 it "simple runAsyncly and 'then' with IO" $

25 it "simple runAsyncly and 'then' with IO" $

24 runAsyncly (liftIO (putStrLn "hello") >> liftIO (putStrLn "world")) `shouldReturn` ()

26 runAsyncly (liftIO (putStrLn "hello") >> liftIO (putStrLn "world")) `shouldReturn` ()

25 it "Then and toList" $

27 it "Then and toList" $

@@ -28,8 +30,8 @@ main = hspec $ do

28 toList (do x <- return 1; y <- return 2; return (x + y)) `shouldReturn` ([3] :: [Int])

30 toList (do x <- return 1; y <- return 2; return (x + y)) `shouldReturn` ([3] :: [Int])

29 it "Alternative composition" $

31 it "Alternative composition" $

30 ((toList $ (return 0 <|> return 1)) >>= return . sort) `shouldReturn` ([0, 1] :: [Int])

32 ((toList $ (return 0 <|> return 1)) >>= return . sort) `shouldReturn` ([0, 1] :: [Int])

31− it "Alternative composition - each" $

33+ it "Alternative composition with bind" $

32− ((toList $ each [1..100]) >>= return . sort) `shouldReturn` ([1..100] :: [Int])

34+ (toList (for [1..10 :: Int] $ \x -> return x >>= return . id) >>= return . sort) `shouldReturn` ([1..10] :: [Int])

33 {-

35 {-

34 it "Alternative composition of async and sync tasks" $

36 it "Alternative composition of async and sync tasks" $

35 ((wait (threads 0 ((async (return 0) <|> return 1)))) >>= return . sort)

37 ((wait (threads 0 ((async (return 0) <|> return 1)))) >>= return . sort)