09283c4 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Implement pull based concurrency rate control

09283c4Harendra Kumar9 years ago

src/Asyncly/AsyncT.hs+126−46

compare with master

unifiedsplit

@@ -25,9 +25,11 @@ module Asyncly.AsyncT

25where

27import Control.Applicative (Alternative (..))

28−import Control.Concurrent (ThreadId, forkIO, myThreadId)

28+import Control.Concurrent (ThreadId, forkIO,

29+ myThreadId, threadDelay)

29import Control.Concurrent.STM (TChan, atomically, newTChan,

30import Control.Concurrent.STM (TChan, atomically, newTChan,

31+ readTChan, tryReadTChan,

30− readTChan, writeTChan)

32+ writeTChan)

31import Control.Exception (fromException, try,

33import Control.Exception (fromException, try,

32 SomeException (..),

34 SomeException (..),

33 BlockedIndefinitelyOnSTM(..))

35 BlockedIndefinitelyOnSTM(..))

@@ -44,6 +46,8 @@ import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),

44 defaultLiftBaseWith,

46 defaultLiftBaseWith,

45 defaultRestoreM, liftBaseWith)

47 defaultRestoreM, liftBaseWith)

46 -}

48 -}

49+import Data.IORef (IORef, writeIORef, newIORef,

50+ readIORef)

47import Data.Maybe (maybe)

51import Data.Maybe (maybe)

48import Data.Monoid ((<>))

52import Data.Monoid ((<>))

@@ -78,7 +82,11 @@ data ChildEvent a =

78-- State threaded around the monad for thread management

82-- State threaded around the monad for thread management

79------------------------------------------------------------------------------

83------------------------------------------------------------------------------

85+data Context m a =

81−newtype Context a = Context { childChannel :: TChan (ChildEvent a) }

86+ Context { childChannel :: TChan (ChildEvent a)

87+ , pullSide :: Bool

88+ , pendingWork :: IORef (AsyncT m a)

89+ }

83-- The 'Maybe (AsyncT m a)' is redundant as we can use 'stop' value for the

91-- The 'Maybe (AsyncT m a)' is redundant as we can use 'stop' value for the

84-- Nothing case, but it makes the fold using '<|>' 25% faster. Need to try

92-- Nothing case, but it makes the fold using '<|>' 25% faster. Need to try

@@ -91,9 +99,9 @@ newtype Context a = Context { childChannel :: TChan (ChildEvent a) }

91newtype AsyncT m a =

99newtype AsyncT m a =

92 AsyncT {

100 AsyncT {

93 runAsyncT :: forall r.

101 runAsyncT :: forall r.

94− Maybe (Context a) -- state

102+ Maybe (Context m a) -- state

95 -> m r -- stop

103 -> m r -- stop

96− -> (a -> Maybe (Context a) -> Maybe (AsyncT m a) -> m r) -- yield

104+ -> (a -> Maybe (Context m a) -> Maybe (AsyncT m a) -> m r) -- yield

97 -> m r

105 -> m r

98 }

106 }

107

@@ -170,7 +178,7 @@ doFork action preExit =

170 return ()

178 return ()

171

179

172{-# NOINLINE push #-}

180{-# NOINLINE push #-}

173−push :: MonadIO m => Context a -> AsyncT m a -> m ()

181+push :: MonadIO m => Context m a -> AsyncT m a -> m ()

174push context action = run (Just context) action

182push context action = run (Just context) action

175

183

176 where

184 where

@@ -203,52 +211,107 @@ handlePushException pchan e = do

203

211

204-- | run m1 in a new thread, pushing its results to a pull channel and then run

212-- | run m1 in a new thread, pushing its results to a pull channel and then run

205-- m2 in the parent thread. Any exceptions are also pushed to the channel.

213-- m2 in the parent thread. Any exceptions are also pushed to the channel.

206−{-# INLINE pushFork #-}

214+{-# INLINE pullSideDispatch #-}

207−pushFork :: (MonadIO m, MonadBaseControl IO m)

215+pullSideDispatch :: MonadAsync m

216+ => Context m a -> AsyncT m a -> AsyncT m a -> AsyncT m a

217+pullSideDispatch ctx m1 m2 = AsyncT $ \_ stp yld -> do

218+ let chan = childChannel ctx

219+ _ <- doFork (push (ctx {pullSide = False}) m1) (handlePushException chan)

220+ liftIO $ threadDelay 0

221+ liftIO $ writeIORef (pendingWork ctx) m2

222+ (runAsyncT (pullDispatch ctx False)) Nothing stp yld

223+

224+{-# NOINLINE pushSideDispatch #-}

225+pushSideDispatch :: MonadAsync m

208− => Context a -> AsyncT m a -> AsyncT m a -> AsyncT m a

226+ => Context m a -> AsyncT m a -> AsyncT m a -> AsyncT m a

209−pushFork c m1 m2 = AsyncT $ \ctx stp yld -> do

227+pushSideDispatch ctx m1 m2 = AsyncT $ \_ stp yld -> do

228+ let chan = childChannel ctx

210− _ <- doFork (push c m1) (handlePushException (childChannel c))

229+ _ <- doFork (push ctx m1) (handlePushException chan)

211− -- XXX run m2 only when the channel becomes empty

230+ (runAsyncT m2) (Just ctx) stp yld

212− (runAsyncT m2) ctx stp yld

231+

213−

232+-- XXX the TBQueue size should be proportional to the pending threads.

214-- We re-raise any exceptions received from the child threads, that way

233-- We re-raise any exceptions received from the child threads, that way

215-- exceptions get propagated to the top level computation and can be handled

234-- exceptions get propagated to the top level computation and can be handled

216-- there.

235-- there.

217−{-# NOINLINE pull #-}

236+{-# NOINLINE pullDispatch #-}

218−pull :: (MonadIO m, MonadThrow m) => Context a -> AsyncT m a

237+pullDispatch :: (MonadIO m, MonadThrow m)

238+ => Context m a -> Bool -> AsyncT m a

219−pull ctx = AsyncT $ \_ stp yld -> do

239+pullDispatch ctx dispatch = AsyncT $ \_ stp yld -> do

240+ let chan = childChannel ctx

241+ eHandler e = handleException e stp

242+ evHandler ev = handleEvent ev stp yld

243+ maybeEvHandler ev = maybe (continue stp yld) evHandler ev

244+

245+ if dispatch then do

246+ m <- liftIO $ readIORef (pendingWork ctx)

247+ (runAsyncT m) (Just ctx) stp yld

248+ else do

249+ res <- liftIO $ try $ atomically $ tryReadTChan chan

250+ either eHandler maybeEvHandler res

251+

252+ where

253+

254+ handleException e stp =

255+ case fromException e of

256+ Just BlockedIndefinitelyOnSTM -> stp

257+ Nothing -> throwM e

258+

259+ {-# INLINE continue #-}

260+ continue stp yld = (runAsyncT (pullDispatch ctx True)) (Just ctx) stp yld

261+

262+ {-# INLINE handleEvent #-}

263+ handleEvent ev stp yld =

264+ case ev of

265+ ChildYield a -> yld a Nothing (Just (pullDispatch ctx False))

266+ ChildDone _ a -> yld a Nothing (Just (pullDispatch ctx True))

267+ -- XXX kill threads

268+ ChildStop _ e -> maybe (continue stp yld) throwM e

269+

270+{-# NOINLINE pullDrain #-}

271+pullDrain :: (MonadIO m, MonadThrow m) => TChan (ChildEvent a) -> AsyncT m a

272+pullDrain chan = AsyncT $ \_ stp yld -> do

273+ let eHandler e = handleException e stp

274+ evHandler ev = handleEvent ev stp yld

275+

220− res <- liftIO $ try (atomically $ readTChan (childChannel ctx))

276+ res <- liftIO $ try (atomically $ readTChan chan)

221− case res of

277+ either eHandler evHandler res

222− Left e ->

278+

279+ where

280+

281+ handleException e stp =

223− case fromException e of

282+ case fromException e of

224− Just BlockedIndefinitelyOnSTM -> stp

283+ Just BlockedIndefinitelyOnSTM -> stp

225− Nothing -> throwM e

284+ Nothing -> throwM e

226− Right ev ->

285+

286+ {-# INLINE continue #-}

287+ continue stp yld = (runAsyncT (pullDrain chan)) Nothing stp yld

288+

289+ {-# INLINE handleEvent #-}

290+ handleEvent ev stp yld =

227− let yielder a = yld a Nothing (Just (pull ctx))

291+ let yielder a = yld a Nothing (Just (pullDrain chan))

228− in case ev of

292+ in case ev of

229− ChildYield a -> yielder a

293+ ChildYield a -> yielder a

230− ChildDone _ a -> yielder a

294+ ChildDone _ a -> yielder a

295+ -- XXX kill threads

231− ChildStop _ e -> do

296+ ChildStop _ e -> maybe (continue stp yld) throwM e

232− let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld

233− in maybe continue throwM e

234

297

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

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

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

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

237{-# NOINLINE pullFork #-}

300{-# NOINLINE pullFork #-}

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

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

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

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

240− c <- liftIO $ newContext

303+ ctx <- liftIO $ newContext (m1 <|> m2)

241− _ <- doFork (push c (m1 <|> m2)) (handlePushException (childChannel c))

242− (runAsyncT (pull c)) (Just c) stp yld

304+ (runAsyncT (pullDispatch ctx True)) Nothing stp yld

243

305

244 where

306 where

245

307

246− newContext = do

308+ newContext m = do

247− childChan <- atomically newTChan

309+ channel <- atomically newTChan

310+ work <- liftIO $ newIORef (m <|> pullDrain channel)

248− return $ Context { childChannel = childChan }

311+ return $ Context { childChannel = channel

249−

312+ , pullSide = True

250−canFork :: IO Bool

313+ , pendingWork = work

251−canFork = return False

314+ }

252

315

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

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

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

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

@@ -260,25 +323,42 @@ canFork = return False

260-- problematic if the initial production latency is high, we may end up

323-- problematic if the initial production latency is high, we may end up

261-- creating too many threads. So we need some way to monitor and use the

324-- creating too many threads. So we need some way to monitor and use the

262-- latency as well.

325-- latency as well.

326+--

263-- TBD For quick response we may have to increase the rate in the middle of

327-- TBD For quick response we may have to increase the rate in the middle of

264-- a serially running computation. For that we can use a state flag to fork

328-- a serially running computation. For that we can use a state flag to fork

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

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

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

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

267--

331--

332+-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.

333+-- If we have to dispatch in arbitrary chunksizes we will need to compose the

334+-- parallel actions using a data constructor instead so that we can divide it

335+-- in chunks of arbitrary size before dispatch. When batching we can convert

336+-- the structure into Alternative batches of Monoid composition. That will also

337+-- allow us to dispatch more work to existing threads rather than creating new

338+-- threads always.

339+--

340+-- TBD for pure work (when we are not in the IO monad) we can divide it into

341+-- just the number of CPUs.

342+--

343+-- XXX to rate control left folded structrues we will have to return the

344+-- residual work back to the dispatcher. It will also consume a lot of

345+-- memory due to queueing of all the work before execution starts.

346+--

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

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

269 empty = mempty

348 empty = mempty

270

349

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

350+ -- Note: This is designed to scale for right associated compositions,

272− -- avoid this automatically if the task is small?

351+ -- therefore always use a right fold for folding bigger structures.

273 {-# INLINE (<|>) #-}

352 {-# INLINE (<|>) #-}

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

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

275 case ctx of

354 case ctx of

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

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

277 Just c -> do

356 Just c -> do

278− doAsync <- liftIO $ canFork

279− if doAsync then

357+ if pullSide c then

280− (runAsyncT (pushFork c m1 m2)) ctx stp yld

358+ (runAsyncT (pullSideDispatch c m1 m2)) ctx stp yld

359+ else

360+ -- for left associated compositions

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

361+ (runAsyncT (pushSideDispatch c m1 m2)) ctx stp yld

282

362

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

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

284 mzero = empty

364 mzero = empty

test/Main.hs+15−1

@@ -1,7 +1,7 @@

1module Main (main) where

3import Asyncly

4−import Control.Applicative ((<|>))

4+import Control.Applicative ((<|>), empty)

5import Control.Concurrent (myThreadId, threadDelay)

6import Data.Monoid ((<>))

7import Control.Monad.IO.Class (liftIO)

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

28 toList (return 1 >> return 2) `shouldReturn` ([2] :: [Int])

29 it "Bind and toList" $

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

31+ it "Alternative - empty" $

32+ (toList empty) `shouldReturn` ([] :: [Int])

33+ it "Alternative composition - empty <|> empty" $

34+ (toList (empty <|> empty)) `shouldReturn` ([] :: [Int])

35+ it "Alternative composition - empty at the beginning" $

36+ (toList $ (empty <|> return 1)) `shouldReturn` ([1] :: [Int])

37+ it "Alternative composition - empty at the end" $

38+ (toList $ (return 1 <|> empty)) `shouldReturn` ([1] :: [Int])

31 it "Alternative composition" $

39 it "Alternative composition" $

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

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

41+ it "Alternative composition - empty in the middle" $

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

43+ it "Alternative composition - left associated" $

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

45+ it "Alternative composition - right associated" $

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

33 it "Alternative composition with bind" $

47 it "Alternative composition with bind" $

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

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

35 {-

49 {-