ce6719a · streamly · git.iko.soy

git.iko.soy/streamly

More SPECIALIZE and CAS for atomicmodify

commitce6719a6b8a7d7eed370bec3889b1a459ef3b8ea authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 19 Jul 2017, 11:17 (9 years ago) parentb7d651f

3 files changed+15−9

compare with master

unifiedsplit

asyncly.cabal+1 src/Asyncly/AsyncT.hs+7−6 src/Asyncly/Threads.hs+7−3

asyncly.cabal+1

@@ -38,6 +38,7 @@ library

3838 , ghc-prim

3939

4040 -- libraries not bundled with GHC

41+ , atomic-primops

4142 , exceptions

4243 , lifted-base

4344 , monad-control

src/Asyncly/AsyncT.hs+7−6

@@ -70,9 +70,9 @@ import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),

7070 MonadTransControl (..),

7171 defaultLiftBaseWith,

7272 defaultRestoreM, liftBaseWith)

73−import Data.IORef (IORef, atomicModifyIORef,

74− modifyIORef, newIORef, readIORef,

75− writeIORef)

73+import Data.Atomics (atomicModifyIORefCAS)

74+import Data.IORef (IORef, modifyIORef, newIORef,

75+ readIORef, writeIORef)

7676import Data.List (delete)

7777import Data.Maybe (isJust)

7878import GHC.Prim (Any)

@@ -338,13 +338,13 @@ waitForOneEvent ctx = do

338338-- instead it returns whether the value is zero or non-zero.

339339--

340340waitQSemB :: IORef Int -> IO Bool

341−waitQSemB sem = atomicModifyIORef sem $ \n ->

341+waitQSemB sem = atomicModifyIORefCAS sem $ \n ->

342342 if n > 0

343343 then (n - 1, True)

344344 else (n, False)

345345

346346signalQSemB :: IORef Int -> IO ()

347−signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ())

347+signalQSemB sem = atomicModifyIORefCAS sem $ \n -> (n + 1, ())

348348

349349-- Allocation of threads

350350--

@@ -361,6 +361,7 @@ signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ())

361361-- high.

362362--

363363

364+{-# INLINE handleResult #-}

364365handleResult :: MonadIO m

365366 => Context -> Either SomeException a -> m (Maybe SomeException)

366367handleResult ctx res =

@@ -376,7 +377,7 @@ handleResult ctx res =

376377 where

377378

378379 handleRecording recording ref = do

379− liftIO $ atomicModifyIORef ref $ \logs -> (recording : logs, ())

380+ liftIO $ atomicModifyIORefCAS ref $ \logs -> (recording : logs, ())

380381 waitForChildren ctx

381382

382383 passException e = do

src/Asyncly/Threads.hs+7−3

@@ -26,8 +26,8 @@ import Control.Monad.Catch (MonadCatch, throwM, try)

2626import Control.Monad.IO.Class (MonadIO (..))

2727import Control.Monad.State (mzero, runStateT)

2828import Control.Monad.Trans.Class (MonadTrans (lift))

29−import Data.IORef (IORef, atomicModifyIORef,

30− newIORef, readIORef)

29+import Data.Atomics (atomicModifyIORefCAS)

30+import Data.IORef (IORef, newIORef, readIORef)

3131

3232import Control.Monad.Trans.Recorder (MonadRecorder(..), RecorderT,

3333 Recording, blank, runRecorderT)

@@ -39,6 +39,8 @@ import Asyncly.AsyncT

3939

4040-- | Run an 'AsyncT m' computation and collect the results generated by each

4141-- thread of the computation in a list.

42+{-# SPECIALIZE waitAsync :: (a -> AsyncT IO a) -> Maybe (IORef [Recording])

43+ -> AsyncT IO a -> IO () #-}

4244waitAsync :: (MonadAsync m, MonadCatch m)

4345 => (a -> AsyncT m a) -> Maybe (IORef [Recording]) -> AsyncT m a -> m ()

4446waitAsync finalizer lref m = do

@@ -56,9 +58,10 @@ waitAsync finalizer lref m = do

5658

5759-- | Invoked to store the result of the computation in the context and finish

5860-- the computation when the computation is done

61+{-# SPECIALIZE gatherResult :: IORef [a] -> a -> AsyncT IO a #-}

5962gatherResult :: MonadAsync m => IORef [a] -> a -> AsyncT m a

6063gatherResult ref r = do

61− liftIO $ atomicModifyIORef ref $ \rs -> (r : rs, ())

64+ liftIO $ atomicModifyIORefCAS ref $ \rs -> (r : rs, ())

6265 mzero

6366

6467gather :: (MonadAsync m, MonadCatch m) => AsyncT m a -> AsyncT m [a]

@@ -70,6 +73,7 @@ gather m = AsyncT $ do

7073

7174-- | Run an 'AsyncT m' computation and collect the results generated by each

7275-- thread of the computation in a list.

76+{-# SPECIALIZE wait :: AsyncT IO a -> IO [a] #-}

7377wait :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m [a]

7478wait m = do

7579 resultsRef <- liftIO $ newIORef []