packages feed

consistent (empty) → 0.0.1

raw patch · 5 files changed

+279/−0 lines, 5 filesdep +basedep +consistentdep +lifted-asyncsetup-changed

Dependencies added: base, consistent, lifted-async, lifted-base, monad-control, stm, transformers, transformers-base, unordered-containers

Files

+ Control/Concurrent/Consistent.hs view
@@ -0,0 +1,167 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TypeFamilies #-}++{-# OPTIONS_GHC -fno-warn-unused-binds #-}++module Control.Concurrent.Consistent+    ( ConsistentT+    , CTMT+    , runConsistentT+    , consistently+    , CVar+    , newCVar+    , dupCVar+    , readCVar+    , writeCVar+    , swapCVar+    , modifyCVar+    ) where++import Control.Applicative+import Control.Concurrent+import Control.Concurrent.Async.Lifted+import Control.Concurrent.STM+import Control.Exception.Lifted (bracket_)+import Control.Monad+import Control.Monad.Base+import Control.Monad.IO.Class+import Control.Monad.Trans.Control+import Control.Monad.Trans.Reader+import Control.Monad.Trans.Writer+import Data.HashMap.Strict as Map+import Data.Maybe+import Prelude++data ConsistentState = ConsistentState+    { csActiveThreads :: TVar Int+    , csJournal       :: TQueue [STM (Maybe (STM ()))]+    }++newtype ConsistentT m a = ConsistentT+    { getConsistentT :: ReaderT ConsistentState m a }+    deriving (Functor, Applicative, Monad, MonadIO)++instance MonadBase IO m => MonadBase IO (ConsistentT m) where+    liftBase b = ConsistentT $ liftBase b++instance MonadBaseControl IO m => MonadBaseControl IO (ConsistentT m) where+    newtype StM (ConsistentT m) a =+        StMConsistentT (StM (ReaderT ConsistentState m) a)+    liftBaseWith f =+        ConsistentT $ liftBaseWith $ \runInBase -> f $ \k ->+            liftM StMConsistentT $ runInBase $ getConsistentT k+    restoreM (StMConsistentT m) = ConsistentT . restoreM $ m++newtype CTMT m a = CTMT (WriterT [STM (Maybe (STM ()))] (ConsistentT m) a)+    deriving (Functor, Applicative, Monad, MonadIO)++runConsistentT :: (MonadBaseControl IO m, MonadIO m) => ConsistentT m a -> m a+runConsistentT action = do+    cs <- liftIO $ ConsistentState <$> newTVarIO 0 <*> newTQueueIO+    flip runReaderT cs+        $ withAsync (liftIO (applyChanges cs))+        $ \worker -> do+            link worker+            getConsistentT action+  where+    applyChanges ConsistentState {..} = forever $ atomically $ do+        -- Process only if no threads are in a "consistently" block.+        active <- readTVar csActiveThreads+        check (active == 0)++        -- Read the next set of updates to apply.+        updates <- sequence =<< readTQueue csJournal++        -- If any component of the update fails the generational check (see+        -- 'postUpdate'), drop the update for consistency's sake.+        when (all isJust updates) $ sequence_ (catMaybes updates)++consistently :: (MonadBaseControl IO m, MonadIO m)+           => CTMT m a -> ConsistentT m a+consistently (CTMT f) = do+    ConsistentState {..} <- ConsistentT ask+    let active = liftIO . atomically . modifyTVar csActiveThreads+    bracket_ (active succ) (active pred) $ do+        (a, updates) <- runWriterT f+        liftIO $ atomically $ writeTQueue csJournal updates+        return a++data CVar a = CVar+    { cdVars    :: TVar (HashMap ThreadId (TVar a, TVar Int))+      -- ^ The @TVar Int@ in cdVars is cvCurrGen from each thread's CVar.+    , cvMyData  :: TVar a+    , cdBaseGen :: TVar Int+    , cdCurrGen :: TVar Int+    }++newCVar :: MonadIO m => a -> m (CVar a)+newCVar a = liftIO $ do+    me   <- newTVarIO a+    bgen <- newTVarIO 0+    cgen <- newTVarIO 0+    tid  <- myThreadId+    vars <- newTVarIO (Map.singleton tid (me, cgen))+    return $ CVar vars me bgen cgen++dupCVar :: MonadIO m => CVar a -> m (CVar a)+dupCVar (CVar vs v _ _) = liftIO $ do+    tid <- myThreadId+    atomically $ do+        me   <- newTVar =<< readTVar v+        bgen <- newTVar 0+        cgen <- newTVar 0+        modifyTVar' vs (Map.insert tid (me, cgen))+        return $ CVar vs me bgen cgen++readCVar :: MonadIO m => CVar a -> m a+readCVar = liftIO . readTVarIO . cvMyData++writeCVar :: MonadIO m => CVar a -> a -> CTMT m ()+writeCVar v a = do+    liftIO $ atomically $ writeTVar (cvMyData v) a+    CTMT $ tell [updateData v a]++updateData :: CVar a -> a -> STM (Maybe (STM ()))+updateData (CVar vs _ bgen cgen) a = do+    -- Get the base generation for this CVar, or what we last knew it to+    -- be, and the current generation, which can be updated by other+    -- threads if their updates succeed.+    base <- readTVar bgen+    curr <- readTVar cgen++    -- If base is not the same as curr, another thread has changed the CVar,+    -- and so any changes in the current consistent block are now invalid.+    -- Return Nothing so that 'applyChanges' throws them away.+    return $+        if base /= curr+        then Nothing+        else Just $ do+            -- Otherwise, move on to the next generation...+            let next = succ base+            writeTVar bgen next+            writeTVar cgen next++            -- Update every thread with the new value and generation.+            -- This causes pending updates in other threads involving the+            -- same variable to become invalid.+            vars <- Map.elems <$> readTVar vs+            forM_ vars $ \(o, ogen) -> do+                writeTVar o a+                writeTVar ogen next++            -- jww (2014-04-07): Performance issue: Multiple calls to+            -- writeCVar from the same block will cause this loop to+            -- execute that many times.++swapCVar :: MonadIO m => CVar a -> a -> CTMT m a+swapCVar v y = do+    x <- readCVar v+    writeCVar v y+    return x++modifyCVar :: MonadIO m => CVar a -> (a -> a) -> CTMT m ()+modifyCVar v f = writeCVar v . f =<< readCVar v
+ LICENSE view
@@ -0,0 +1,19 @@+opyright (c) 2014 John Wiegley++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in+all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN+THE SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ consistent.cabal view
@@ -0,0 +1,50 @@+Name:                consistent+Version:             0.0.1+Synopsis:            Eventually consistent STM transactions.+License-file:        LICENSE+License:             MIT+Author:              John Wiegley+Maintainer:          johnw@newartisans.com+Build-Type:          Simple+Cabal-Version:       >=1.10+Category:            System+Description:+  Eventually consistent STM transactions.+  \+  Consistent provides eventually consistent atomic transactions, by delaying+  updates until no threads is mutating a shared variable.+  \+  This comes at a cost of having a separate TVar for every thread, but has the+  advantage that no thread will ever lock or retry except for the manager actor+  responsible for performing the updates.++Source-repository head+  type: git+  location: git://github.com/jwiegley/consistent.git++Library+    default-language:   Haskell98+    ghc-options: -Wall+    build-depends:+        base                 >= 3 && < 5+      , monad-control        >= 0.3.2.3+      , transformers         >= 0.3.0.0+      , transformers-base    >= 0.4.1+      , lifted-base          >= 0.2.2.0+      , lifted-async         >= 0.1.1+      , unordered-containers >= 0.2.3.0+      , stm                  >= 2.4.2+    exposed-modules:+        Control.Concurrent.Consistent++test-suite test+    hs-source-dirs: test+    default-language: Haskell2010+    main-is: main.hs+    type: exitcode-stdio-1.0+    ghc-options: -Wall -threaded+    build-depends:+        base+      , consistent+      , transformers         >= 0.3.0.0+      , lifted-async         >= 0.1.1
+ test/main.hs view
@@ -0,0 +1,41 @@+module Main where++import Control.Concurrent+import Control.Concurrent.Async.Lifted+import Control.Concurrent.Consistent+-- import Control.Exception+import Control.Monad hiding (forM_, mapM_)+import Control.Monad.IO.Class+import Debug.Trace+import Prelude hiding (log)+import Prelude hiding (log, mapM_)+--import Test.Hspec++-- tryAny :: IO a -> IO (Either SomeException a)+-- tryAny = try++-- main :: IO ()+-- main = hspec $ do+--     describe "simple logging" $ do+--         it "logs output" $ True `shouldBe` True++main :: IO ()+main = do+    test <- async $ void $ runConsistentT $ do+        u <- newCVar 0+        v <- newCVar 0+        mapConcurrently worker $+            flip map [1..100 :: Int] $ \i -> (u, v, i)+    wait test+  where+    worker (pu, pv, i) = do+        u <- dupCVar pu+        v <- dupCVar pv+        replicateM_ 100 $ do+            consistently $ do+                x <- readCVar u+                writeCVar u i+                y <- readCVar v+                writeCVar v i+                trace (show (x, y)) $ return ()+            liftIO $ threadDelay (i * 1000)