lvish (empty) → 1.0
raw patch · 33 files changed
+6284/−0 lines, 33 filesdep +HUnitdep +abstract-pardep +asyncsetup-changed
Dependencies added: HUnit, abstract-par, async, atomic-primops, base, bits-atomic, bytestring, bytestring-mmap, chaselev-deque, containers, deepseq, ghc-prim, hashable, lattices, missing-foreign, parallel, random, rdtsc, split, test-framework, test-framework-hunit, test-framework-th, time, transformers, vector
Files
- Control/LVish.hs +264/−0
- Control/LVish/DeepFrz.hs +164/−0
- Control/LVish/DeepFrz/Internal.hs +41/−0
- Control/LVish/Internal.hs +103/−0
- Control/LVish/MonadToss.hs +14/−0
- Control/LVish/SchedIdempotent.hs +701/−0
- Control/LVish/SchedIdempotentInternal.hs +187/−0
- Control/LVish/Types.hs +20/−0
- Control/Reagent.hs +42/−0
- Data/Concurrent/AlignedIORef.hs +24/−0
- Data/Concurrent/Bag.hs +51/−0
- Data/Concurrent/Counter.hs +22/−0
- Data/Concurrent/LinkedMap.hs +117/−0
- Data/Concurrent/SNZI.hs +109/−0
- Data/Concurrent/SkipListMap.hs +203/−0
- Data/LVar/Generic.hs +80/−0
- Data/LVar/Generic/Internal.hs +94/−0
- Data/LVar/IStructure.hs +201/−0
- Data/LVar/IVar.hs +200/−0
- Data/LVar/Internal/Pure.hs +126/−0
- Data/LVar/MaxCounter.hs +79/−0
- Data/LVar/NatArray.hs +284/−0
- Data/LVar/Pair.hs +56/−0
- Data/LVar/PureMap.hs +370/−0
- Data/LVar/PureSet.hs +397/−0
- Data/LVar/SLMap.hs +361/−0
- Data/LVar/SLSet.hs +397/−0
- Data/UtilInternal.hs +34/−0
- LICENSE +30/−0
- Setup.hs +2/−0
- TestHelpers.hs +186/−0
- lvish.cabal +145/−0
- unit-tests.hs +1180/−0
+ Control/LVish.hs view
@@ -0,0 +1,264 @@+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE Rank2Types #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE DataKinds #-} -- For 'Determinism'+-- {-# LANGUAGE ConstraintKinds, KindSignatures #-}+{-# LANGUAGE MagicHash #-}+{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}++{-|++ The @lvish@ package provides a parallel programming model based on monotonically+ growing data structures.++ This module provides the core scheduler and basic control flow operations. + But to do anything useful you will need to import one of the data structure modules+ (@Data.LVar.*@).++ Here is a self-contained example that writes the same value to @num@+ twice and deterministically prints @4@ instead of raising an error, as+ it would if @num@ were a traditional IVar rather than an LVar. (You+ will need to compile using the @-XDataKinds@ extension.)++> {-# LANGUAGE DataKinds #-}+> import Control.LVish -- Generic scheduler; works with any lattice.+> import Data.LVar.IVar -- The particular lattice in question.+> +> p :: Par Det s Int+> p = do+> num <- new+> fork $ put num 4+> fork $ put num 4+> get num+> +> main = do+> print $ runPar $ p++ -}++-- This module reexports the default LVish scheduler, adding some type-level+-- wrappers to ensure propert treatment of determinism.+module Control.LVish+ (+ -- * CRITICAL OBLIGATIONS for the user: valid @Eq@ and total @Ord@+ {-| + We would like to tell you that if you're programming with Safe Haskell (@-XSafe@),+ that this library provides a formal guarantee that anything executed with `runPar` is+ guaranteed-deterministic. Unfortunately, as of this release there is still one back-door+ that hasn't yet been closed.++ If an adverserial user defines invalid `Eq` instances (claiming objects are equal when they're+ not), or if they define a `compare` function that is not a /pure, total function/,+ and then they store those types within `LVar`s,+ then nondeterminism may leak out of a parallel `runPar` computation.++ In future releases, we will strive to require alternate, safe versions of `Eq` and+ `Ord` that are derived automatically by our library and by the GHC compiler.+ -}++ -- * Par computations and their parameters+ Par(), + Determinism(..), liftQD,+ LVishException(..),+ + -- * Basic control flow+ fork,+ yield, + runPar, runParIO,+-- runParIO_, runParLogged,+-- quiesceAll,+ + -- * Various loop constructs+ parForL, parForSimple, parForTree, parForTiled, for_,++ -- * Synchronizing with handler pools+ L.HandlerPool(), + newPool, + withNewPool, withNewPool_, + quiesce, + + forkHP,+ + -- * Debug facilities and internal bits+ logStrLn, runParLogged, + LVar()+ ) where++import qualified Data.Foldable as F+import Control.Exception (Exception)+import Control.LVish.Internal+import Control.LVish.DeepFrz.Internal (Frzn, Trvrsbl)+import qualified Control.LVish.SchedIdempotent as L+import Control.LVish.Types+import System.IO.Unsafe (unsafePerformIO, unsafeDupablePerformIO)++import Prelude hiding (rem)+-- import GHC.Exts (Constraint)++--------------------------------------------------------------------------------++--------------------------------------------------------------------------------+-- Inline *everything*, because these are just wrappers:+{-# INLINE liftQD #-}+{-# INLINE yield #-}+{-# INLINE newPool #-}+{-# INLINE runParIO #-}+{-# INLINE runPar #-}+--{-# INLINE runParThenFreeze #-}+{-# INLINE fork #-}+{-# INLINE quiesce #-}+--------------------------------------------------------------------------------++-- | It is always safe to lift a deterministic computation to a+-- quasi-deterministic one.+liftQD :: Par Det s a -> Par QuasiDet s a+liftQD (WrapPar p) = (WrapPar p)++-- | Cooperatively schedule other threads.+yield :: Par d s ()+yield = WrapPar L.yield++-- | Block until a handler pool is quiescent, i.e., until all+-- associated parallel computations have completed.+quiesce :: L.HandlerPool -> Par d s ()+quiesce = WrapPar . L.quiesce++-- | A global barrier. Wait for all unblocked, active threads of work in the system+-- to complete, and then proceed after that point.+quiesceAll :: Par d s ()+quiesceAll = WrapPar L.quiesceAll++-- | Execute a computation in parallel.+fork :: Par d s () -> Par d s ()+fork (WrapPar f) = WrapPar$ L.fork f++-- | A version of `fork` that also allows the forked computation to be tracked in a+-- `HandlerPool`, that enables the programmer to synchronize on the completion of the+-- child computation. But be careful; this does not automatically wait for+-- all downstream forked computations (transitively).+forkHP :: Maybe L.HandlerPool -> Par d s () -> Par d s ()+forkHP mh (WrapPar f) = WrapPar$ L.forkHP mh f++-- | Create a new pool that can be used to synchronize on the completion of all+-- parallel computations associated with the pool.+newPool :: Par d s L.HandlerPool+newPool = WrapPar L.newPool++-- | Execute a Par computation in the context of a fresh handler pool.+withNewPool :: (L.HandlerPool -> Par d s a) -> Par d s (a, L.HandlerPool)+withNewPool f = WrapPar $ L.withNewPool $ unWrapPar . f++-- | Execute a Par computation in the context of a fresh handler pool, while+-- ignoring the result of the computation.+withNewPool_ :: (L.HandlerPool -> Par d s ()) -> Par d s L.HandlerPool+withNewPool_ f = WrapPar $ L.withNewPool_ $ unWrapPar . f++-- | If the input computation is quasi-deterministic (`QuasiDet`), then this may+-- throw a `LVishException` non-deterministically on the thread that calls it, but if+-- it returns without exception then it always returns the same answer.+--+-- If the input computation is deterministic (`Det`), then @runParIO@ will return the+-- same result as `runPar`. However, `runParIO` is still possibly useful for+-- avoiding an extra `unsafePerformIO` required inside the implementation of+-- `runPar`.+-- +-- In the future, /full/ non-determinism may be allowed as a third setting beyond+-- `Det` and `QuasiDet`.+runParIO :: (forall s . Par d s a) -> IO a+runParIO (WrapPar p) = L.runParIO p ++-- | Useful ONLY for timing.+runParIO_ :: (Par d s a) -> IO ()+runParIO_ (WrapPar p) = L.runParIO p >> return ()++-- | Useful for debugging. Returns debugging logs, in realtime order, in addition to+-- the final result.+runParLogged :: (forall s . Par d s a) -> IO ([String],a)+runParLogged (WrapPar p) = L.runParLogged p ++-- | If a computation is guaranteed-deterministic, then `Par` becomes a dischargeable+-- effect. This function will create new worker threads and do the work in parallel,+-- returning the final result.+--+-- (For now there is no sharing of workers with repeated invocations; so+-- keep in mind that @runPar@ is an expensive operation. [2013.09.27])+runPar :: (forall s . Par Det s a) -> a+runPar (WrapPar p) = L.runPar p ++-- | This is only used when compiled in debugging mode. It atomically adds a string+-- onto an in-memory log.+logStrLn :: String -> Par d s ()+#ifdef DEBUG_LVAR+logStrLn = WrapPar . L.logStrLn+#else +logStrLn _ = return ()+{-# INLINE logStrLn #-}+#endif++++--------------------------------------------------------------------------------+-- Extras+--------------------------------------------------------------------------------++{-# INLINE parForL #-}+-- | Left-biased parallel for loop. As worker threads beyond the first are added,+-- this hews closer to the sequential iteration order than an unbiased parallel loop.+--+-- Takes a range as inclusive-start, exclusive-end.+parForL :: (Int,Int) -> (Int -> Par d s ()) -> Par d s ()+parForL (start,end) _ | start > end = error$"parForL: start is greater than end: "++show (start,end)+parForL (start,end) body = do+ -- logStrLn$ " initial iters: "++show (end-start)+ loop 0 (end - start) 1+ where+ loop offset remain chunk+ | remain <= 0 = return () + | remain <= chunk = parForSimple (offset, offset+remain) body+ | otherwise = do+ let nxtstrt = offset+chunk+ -- logStrLn$ "loop: .. "++show (offset, remain, chunk)+ fork$ parForSimple (offset, nxtstrt) body+ loop nxtstrt (remain-chunk) (2*chunk)++{-# INLINE parForSimple #-}+-- | The least-sophisticated form of parallel loop. Fork iterations one at a time.+parForSimple :: (Int,Int) -> (Int -> Par d s ()) -> Par d s ()+parForSimple range fn = do+ for_ range $ \i -> fork (fn i) ++-- | Divide the iteration space recursively, but ultimately run every iteration in+-- parallel. That is, the loop body is permitted to block on other iterations.+parForTree :: (Int,Int) -> (Int -> Par d s ()) -> Par d s ()+parForTree (start,end) _+ | start > end = error$"parForTree: start is greater than end: "++show (start,end)+parForTree (start,end) body = do+ loop 0 (end - start)+ where+ loop offset remain + | remain == 1 = body offset+ | otherwise = do+ let (half,rem) = remain `quotRem` 2+ fork$ loop offset half+ loop (offset+half) (half+rem)+++-- | Split the work into a number of tiles, and fork it in a tree topology.+parForTiled :: Int -> (Int,Int) -> (Int -> Par d s ()) -> Par d s ()+parForTiled otiles (start,end) body = do + loop 0 (end - start) otiles+ where+ loop offset remain tiles+ | remain == 1 = body offset+ | tiles == 1 = for_ (offset,offset+remain) body+ | otherwise = do+ let (half,rem) = remain `quotRem` 2+ (halfT,remT) = tiles `quotRem` 2+ fork$ loop offset half halfT+ loop (offset+half) (half+rem) (halfT+remT)
+ Control/LVish/DeepFrz.hs view
@@ -0,0 +1,164 @@+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE MagicHash #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE DataKinds #-}++{-|++Provides a way to return arbitrarily complex data-structures containing LVars+from `Par` computations.++The important thing to know is that to use `runParThenFreeze` you must make sure that+all types you return from the parallel computation have `DeepFrz` instances. This+means if you introduce custom (non-LVar) datatypes, you will need to include a bit of+boilerplate to give them `DeepFrz` instances. Here is a complete example:++> {-# LANGUAGE TypeFamilies #-}+> import Control.LVish.DeepFrz+> +> data MyData = MyData Int deriving Show+> +> instance DeepFrz MyData where+> type FrzType MyData = MyData+> +> main = print (runParThenFreeze (return (MyData 3)))++-}++-- LK: TODO: another example of a recursive FrzType would be nice.++module Control.LVish.DeepFrz+ (+ -- * The functions you'll want to use+ runParThenFreeze,+ runParThenFreezeIO,++ -- * Some supporting types+ DeepFrz(), FrzType,+ Frzn, Trvrsbl,+ + ) where++import Data.Int+import Data.Word+import GHC.Prim (unsafeCoerce#)++-- import Control.LVish (LVarData1(..))+import Control.LVish.DeepFrz.Internal (DeepFrz(..), Frzn, Trvrsbl)+import Control.LVish.Internal (Determinism(..), Par(WrapPar))+import Control.LVish.SchedIdempotent (runPar, runParIO)+--------------------------------------------------------------------------------++-- | Under normal conditions, calling a `freeze` operation makes a `Par` computation+-- quasi-deterministic. However, if we freeze once all LVar operations are completed+-- (after the implicit global barrier of `runPar`), then we've avoided all data+-- races, and freezing is therefore safe.+-- +-- For this to be possible, the type returned from the `Par` computation must be a+-- member of the `DeepFrz` class. All LVar libraries should provide this instance+-- already. Further, you can create additional instances for custom, pure datatypes.+-- The result of a `runParThenFreeze` depends on the type-level function `FrzType`,+-- whose only purpose is to toggle the `s` parameters of all IVars to the `Frzn`+-- state.+--+-- Significantly, the freeze at the end of `runParThenFreeze` has /no/ runtime cost, in+-- spite of the fact that it enables a /deep/ (recursive) freeze of the value returned+-- by the `Par` computation.+runParThenFreeze :: DeepFrz a => Par Det s a -> FrzType a+runParThenFreeze (WrapPar p) = frz $ runPar p++-- | This version works for non-deterministic computations as well.+-- +-- Such computations may also do freezes internally, but this function has an+-- advantage vs. doing your own freeze at the end of your computation. Namely, when+-- you use `runParThenFreezeIO`, there is an implicit barrier before the final+-- freeze. Further, `DeepFrz` has no runtime overhead, whereas regular freezing has a cost.+runParThenFreezeIO :: DeepFrz a => Par d s a -> IO (FrzType a)+runParThenFreezeIO (WrapPar p) = do+ x <- runParIO p+ return $ frz x++{-+-- This won't work because it conflicts with other instances such as "Either":+instance (LVarData1 f, DeepFrz a) => DeepFrz (f s a) where+ type FrzType (f s a) = f Frzn (FrzType a)+ frz = unsafeCoerce#+-}++#define MKFRZINST(T) instance DeepFrz T where type FrzType T = T++MKFRZINST(Int)+MKFRZINST(Int8)+MKFRZINST(Int16)+MKFRZINST(Int32)+MKFRZINST(Int64)+MKFRZINST(Word)+MKFRZINST(Word8)+MKFRZINST(Word16)+MKFRZINST(Word32)+MKFRZINST(Word64)+MKFRZINST(Bool)+MKFRZINST(Char)+MKFRZINST(Integer)+MKFRZINST(Float)+MKFRZINST(Double)++MKFRZINST(())+MKFRZINST(Ordering)++instance DeepFrz a => DeepFrz [a] where+ type FrzType [a] = [FrzType a]+ frz = unsafeCoerce#++instance DeepFrz a => DeepFrz (Maybe a) where+ type FrzType (Maybe a) = Maybe (FrzType a)+ frz = unsafeCoerce#++instance (DeepFrz a, DeepFrz b) => DeepFrz (Either a b) where+ type FrzType (Either a b) = Either (FrzType a) (FrzType b)+ frz = unsafeCoerce#++instance (DeepFrz a, DeepFrz b) => DeepFrz (a,b) where+ type FrzType (a,b) = (FrzType a,FrzType b)+ frz = unsafeCoerce#++instance (DeepFrz a, DeepFrz b, DeepFrz c) => DeepFrz (a,b,c) where+ type FrzType (a,b,c) = (FrzType a,FrzType b,FrzType c)+ frz = unsafeCoerce#++instance (DeepFrz a, DeepFrz b, DeepFrz c, DeepFrz d) => DeepFrz (a,b,c,d) where+ type FrzType (a,b,c,d) = (FrzType a, FrzType b, FrzType c, FrzType d)+ frz = unsafeCoerce#++instance (DeepFrz a, DeepFrz b, DeepFrz c, DeepFrz d, DeepFrz e) => DeepFrz (a,b,c,d,e) where+ type FrzType (a,b,c,d,e) = (FrzType a, FrzType b, FrzType c, FrzType d, FrzType e)+ frz = unsafeCoerce#++instance (DeepFrz a, DeepFrz b, DeepFrz c, DeepFrz d, DeepFrz e,+ DeepFrz f) => DeepFrz (a,b,c,d,e,f) where+ type FrzType (a,b,c,d,e,f) = (FrzType a, FrzType b, FrzType c, FrzType d, FrzType e,+ FrzType f)+ frz = unsafeCoerce#++instance (DeepFrz a, DeepFrz b, DeepFrz c, DeepFrz d, DeepFrz e,+ DeepFrz f, DeepFrz g) => DeepFrz (a,b,c,d,e,f,g) where+ type FrzType (a,b,c,d,e,f,g) = (FrzType a, FrzType b, FrzType c, FrzType d, FrzType e,+ FrzType f, FrzType g)+ frz = unsafeCoerce#+++instance (DeepFrz a, DeepFrz b, DeepFrz c, DeepFrz d, DeepFrz e,+ DeepFrz f, DeepFrz g, DeepFrz h) => DeepFrz (a,b,c,d,e,f,g,h) where+ type FrzType (a,b,c,d,e,f,g,h) = (FrzType a, FrzType b, FrzType c, FrzType d, FrzType e,+ FrzType f, FrzType g, FrzType h)+ frz = unsafeCoerce#+++instance (DeepFrz a, DeepFrz b, DeepFrz c, DeepFrz d, DeepFrz e,+ DeepFrz f, DeepFrz g, DeepFrz h, DeepFrz i) => DeepFrz (a,b,c,d,e,f,g,h,i) where+ type FrzType (a,b,c,d,e,f,g,h,i) = (FrzType a, FrzType b, FrzType c, FrzType d, FrzType e,+ FrzType f, FrzType g, FrzType h, FrzType i)+ frz = unsafeCoerce#
+ Control/LVish/DeepFrz/Internal.hs view
@@ -0,0 +1,41 @@+{-# LANGUAGE Unsafe #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE EmptyDataDecls #-}++-- | This module is NOT Safe-Haskell, but it must be used to create+-- new LVar types.+module Control.LVish.DeepFrz.Internal+ (+ DeepFrz(..), Frzn, Trvrsbl + )+ where++-- | DeepFreezing is type-level (guaranteed O(1) time complexity)+-- operation. It marks an LVar and its contents (recursively) as+-- frozen. DeepFreezing is not an action that can be taken directly+-- by the user, however. Rather it is an optional final-step in a+-- `runPar` invocation.+class DeepFrz a where+ -- | This type function is public. It maps pre-frozen types to+ -- frozen ones. It should be idempotent.+ type FrzType a :: *++ -- | Private: not exported to the end user.+ frz :: a -> FrzType a++ -- | While `frz` is not exported, users may opt-in to the `DeepFrz`+ -- class for their datatypes and take advantage of the default instance.+ -- Doing so REQUIRES that `type FrzType a = a`.+ default frz :: a -> a + frz a = a ++-- | An uninhabited type that signals an LVar has been frozen.+-- LVars should use this inplace of their `s` parameter.+data Frzn++-- | An uninhabited type that signals an LVar is not only frozen, but+-- it may be traversed in whatever order its internal representation+-- dictates.+data Trvrsbl +
+ Control/LVish/Internal.hs view
@@ -0,0 +1,103 @@+{-# LANGUAGE Unsafe #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DataKinds #-} -- For Determinism+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE KindSignatures #-}++{-|++This module is note @SafeHaskell@; as an end-user, you shouldn't ever need to import it.++It is exposed only because it is necessary for writing /new/ LVars that live in their+own, separate packages.++-}++module Control.LVish.Internal+ (+ -- * Type-safe wrappers around internal components+ Par(..), LVar(..),+ Determinism(..),+ + -- * Unsafe conversions and lifting+ unWrapPar, unsafeRunPar,+ unsafeConvert, state,+ liftIO,++ -- * General utilities+ for_+ )+ where++import Control.Monad.IO.Class+import Control.LVish.MonadToss+import Control.Applicative+import qualified Control.LVish.SchedIdempotent as L+import Control.LVish.DeepFrz.Internal (Frzn, Trvrsbl)+import qualified Data.Foldable as F+import Data.List (sort)++{-# INLINE state #-}+{-# INLINE unsafeConvert #-}+{-# INLINE unWrapPar #-}+--------------------------------------------------------------------------------++-- | This datatype is promoted to type-level (@DataKinds@ extension)+-- and used to indicate whether a `Par` computation is+-- guaranteed-deterministic, or only quasi-deterministic (i.e., might+-- throw `NonDeterminismExn`).+data Determinism = Det | QuasiDet+ deriving Show++-- | The type of parallel computations. A computation @Par d s a@ may or may not be+-- deterministic based on the setting of the `d` parameter (of kind `Determinism`).+-- The `s` parameter is for preventing the escape of @LVar@s from @Par@ computations+-- (just like the @ST@ monad). +-- +-- Implementation note: This is a wrapper around the internal `Par` type, only with more type parameters. +newtype Par :: Determinism -> * -> * -> * where+ WrapPar :: L.Par a -> Par d s a+ deriving (Monad, Functor, Applicative)++-- | The generic representation of LVars used by the scheduler. The+-- end-user can't actually do anything with these and should not try+-- to.+newtype LVar s all delt = WrapLVar { unWrapLVar :: L.LVar all delt }++-- | Unsafe: drops type information to go from the safe `Par` monad to+-- the internal, dangerous one.+unWrapPar :: Par d s a -> L.Par a+unWrapPar (WrapPar p) = p ++-- | This is cheating! It pays no attention to session sealing (@s@) or to the+-- determinism level (@d@).+unsafeRunPar :: Par d s a -> a+unsafeRunPar p = L.runPar (unWrapPar p)++-- | Extract the state of an LVar. This should only be used by implementations of+-- new LVar data structures.+state :: LVar s a d -> a+state = L.state . unWrapLVar++-- | Ignore the extra type annotations regarding both determinism and session-sealing.+unsafeConvert :: Par d1 s1 a -> Par d2 s2 a+unsafeConvert (WrapPar p) = (WrapPar p)++instance MonadIO (Par d s) where+ liftIO = WrapPar . L.liftIO ++instance MonadToss (Par d s) where+ toss = WrapPar L.toss++-- | A simple for loop for numeric ranges (not requiring deforestation+-- optimizations like `forM`). Inclusive start, exclusive end.+{-# INLINE for_ #-}+for_ :: Monad m => (Int, Int) -> (Int -> m ()) -> m ()+for_ (start, end) _fn | start > end = error "for_: start is greater than end"+for_ (start, end) fn = loop start+ where+ loop !i | i == end = return ()+ | otherwise = do fn i; loop (i+1)+
+ Control/LVish/MonadToss.hs view
@@ -0,0 +1,14 @@+module Control.LVish.MonadToss where++import Control.Monad+import System.Random (randomIO)++-- | A typeclass for monads supporting a coin toss operation. NB: the coin is+-- expected to be core-local, so that flipping by multiple threads does not+-- cause contention.+class Monad m => MonadToss m where+ toss :: m Bool+ +instance MonadToss IO where + toss = randomIO+ -- TODO: FIXME: probably use mwc-random here instead...
+ Control/LVish/SchedIdempotent.hs view
@@ -0,0 +1,701 @@+{-# LANGUAGE Unsafe #-}+{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE CPP #-}+{-# LANGUAGE NamedFieldPuns, BangPatterns #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE DoAndIfThenElse #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE FlexibleInstances #-} -- For DeepFreeze++{-# OPTIONS_GHC -Wall -fno-warn-name-shadowing -fno-warn-unused-do-bind #-}++-- | This is an internal module that provides the core parallel scheduler.+-- It is NOT for end-users.++module Control.LVish.SchedIdempotent+ (+ -- * Basic types and accessors:+ LVar(), state, HandlerPool(),+ Par(..), ClosedPar(..),+ + -- * Safe, deterministic operations:+ yield, newPool, fork, forkHP,+ runPar, runParIO, runParLogged,+ withNewPool, withNewPool_,+ forkWithExceptions,+ + -- * Quasi-deterministic operations:+ quiesce, quiesceAll,++ -- * Debug facilities+ logStrLn, dbgLvl,+ + -- * UNSAFE operations. Should be used only by experts to build new abstractions.+ newLV, getLV, putLV, putLV_, freezeLV, freezeLVAfter,+ addHandler, liftIO, toss+ ) where++import Control.Monad hiding (sequence, join)+import Control.Concurrent hiding (yield)+import qualified Control.Exception as E+import Control.DeepSeq+import Control.Applicative+import Control.LVish.MonadToss+import Data.IORef+import Data.Atomics+import Data.Typeable+import qualified Data.Concurrent.Counter as C+import qualified Data.Concurrent.Bag as B+import GHC.Conc hiding (yield)+import System.IO+import System.IO.Unsafe (unsafePerformIO)+import System.Environment(getEnvironment)+import System.Mem.StableName (makeStableName, hashStableName)+import Debug.Trace(trace)+import Prelude hiding (mapM, sequence, head, tail)+import System.Random (random)++-- import Control.Compose ((:.), unO)+import Data.Traversable ++import Control.LVish.Types+import qualified Control.LVish.SchedIdempotentInternal as Sched+++----------------------------------------------------------------------------------------------------+-- THREAD-SAFE LOGGING+----------------------------------------------------------------------------------------------------++-- This should probably be moved into its own module...++globalLog :: IORef [String]+globalLog = unsafePerformIO $ newIORef []++-- | Atomically add a line to the given log.+logStrLn :: String -> Par ()+logStrLn_ :: String -> IO ()+logLnAt_ :: Int -> String -> IO ()+#ifdef DEBUG_LVAR+#warning "Compiling in LVish DEBUG mode."+logStrLn = liftIO . logStrLn_+logStrLn_ s = logLnAt_ 1 s+logLnAt_ lvl s | dbgLvl >= 5 = putStrLn s+ | dbgLvl >= lvl = atomicModifyIORef globalLog $ \ss -> (s:ss, ())+ | otherwise = return ()+#else +logStrLn _ = return ()+logStrLn_ _ = return ()+logLnAt_ _ _ = return ()+{-# INLINE logStrLn #-}+{-# INLINE logStrLn_ #-}+#endif++-- | Print all accumulated log lines+printLog :: IO ()+printLog = do+ -- Clear the log when we read it:+ lines <- atomicModifyIORef globalLog $ \ss -> ([], ss)+ mapM_ putStrLn $ reverse lines + +printLogThread :: IO (IO ())+printLogThread = do+ tid <- forkIO $+ E.catch loop (\ (e :: E.AsyncException) -> do+ -- One last time on kill:+ printLog+ putStrLn " [dbg-log-printer] Shutting down."+ )+ return (do killThread tid+ let wait = do+ stat <- threadStatus tid+ case stat of+ ThreadRunning -> threadDelay 1000 >> wait+ _ -> return ()+ wait)+ where+ loop = do+ -- Flush the log at 5Hz:+ printLog+ threadDelay (200 * 1000)+ loop++{-# NOINLINE theEnv #-}+theEnv :: [(String, String)]+theEnv = unsafePerformIO getEnvironment++{-# NOINLINE dbgLvl #-}+-- | Debugging flag shared by several modules.+-- This is activated by setting the environment variable DEBUG=1..5+dbgLvl :: Int+dbgLvl = case lookup "DEBUG" theEnv of+ Nothing -> defaultDbg+ Just "" -> defaultDbg+ Just "0" -> defaultDbg+ Just s ->+ case reads s of+ ((n,_):_) -> trace (" [!] Responding to env Var: DEBUG="++show n) n+ [] -> error$"Attempt to parse DEBUG env var as Int failed: "++show s++defaultDbg :: Int+defaultDbg = 0++------------------------------------------------------------------------------+-- LVar and Par monad representation+------------------------------------------------------------------------------++-- | LVars are parameterized by two types:+-- +-- * The first, @a@, characterizes the "state" of the LVar (i.e. the lattice+-- value), and should be a concurrently mutable data type. That means, in+-- particular, that only a /transient snapshot/ of the lattice value can be+-- obtained in general. But the information in such a snapshot is always a+-- lower bound on the current value of the LVar.+--+-- * The second, @d@, characterizes the "delta" associated with a @putLV@+-- operation (i.e. the actual change, if any, to the LVar's lattice value).+-- In many cases such deltas allow far more efficient communication between+-- @putLV@s and blocked @getLV@s or handlers. It is crucial, however, that+-- the behavior of a @get@ or handler does not depend on the /particular/+-- choice of @putLV@ operations (and hence deltas) that moved the LVar over+-- the threshold. For simple data structures, the delta may just be the+-- entire LVar state, but for e.g. collection data structures, delta will+-- generally represent a single insertion.+data LVar a d = LVar {+ state :: a, -- the current, "global" state of the LVar+ status :: {-# UNPACK #-} !(IORef (Status d)), -- is the LVar active or frozen? + name :: {-# UNPACK #-} !LVarID -- a unique identifier for this LVar+}++type LVarID = IORef ()+newLVID = newIORef ()++-- | a global ID that is *not* the name of any LVar. Makes it possible to+-- represent Maybe (LVarID) with the type LVarID -- i.e., without any allocation.+noName :: LVarID+noName = unsafePerformIO $ newLVID++-- | The frozen bit of an LVar is tied together with the bag of waiting listeners,+-- which allows the entire bag to become garbage immediately after freezing.+-- (Note, however, that outstanding @put@s that occurred just before freezing+-- may still reference the bag, which is necessary to ensure that all listeners+-- are informed of the @put@ prior to freezing.)+data Status d + = Frozen -- further changes to the state are forbidden+ | Active (B.Bag (Listener d)) -- bag of blocked threshold reads and handlers++-- | A listener for an LVar is informed of each change to the LVar's lattice value+-- (represented as a delta) and the event of the LVar freezing. The listener is+-- given access to a bag token, allowing it to remove itself from the bag of+-- listeners, after unblocking a threshold read, for example. It is also given+-- access to the scheduler queue for the CPU that generated the event, which it+-- can use to add threads.+data Listener d = Listener {+ onUpdate :: d -> B.Token (Listener d) -> SchedState -> IO (),+ onFreeze :: B.Token (Listener d) -> SchedState -> IO ()+}++-- | A HandlerPool contains a way to count outstanding parallel computations that+-- are affiliated with the pool. It detects the condition where all such threads+-- have completeed.+data HandlerPool = HandlerPool {+ numHandlers :: C.Counter, -- How many handler callbacks are currently+ -- running?+ blockedOnQuiesce :: B.Bag ClosedPar+}++-- | A monadic type constructor for parallel computations producing an answer @a@.+-- This is the internal, unsafe type.+newtype Par a = Par {+ -- the computation is represented in CPS+ close :: (a -> ClosedPar) -> ClosedPar +}++-- A "closed" Par computation is one that has been plugged into a continuation.+-- It is represented in a "Church encoded" style, i.e., directly in terms of its+-- interpretation into the IO monad. Since the continuation has already been+-- plugged into the computation, there is no answer type here.+newtype ClosedPar = ClosedPar {+ exec :: SchedState -> IO ()+}++type SchedState = Sched.State ClosedPar LVarID++instance Functor Par where+ fmap f m = Par $ \k -> close m (k . f)++instance Monad Par where+ return a = Par $ \k -> k a+ m >>= c = Par $ \k -> close m $ \a -> close (c a) k++instance Applicative Par where+ (<*>) = ap+ pure = return+++------------------------------------------------------------------------------+-- A few auxiliary functions+------------------------------------------------------------------------------ ++mkPar :: ((a -> ClosedPar) -> SchedState -> IO ()) -> Par a+mkPar f = Par $ \k -> ClosedPar $ \q -> f k q++whenJust :: Maybe a -> (a -> IO ()) -> IO ()+whenJust Nothing _ = return ()+whenJust (Just a) f = f a++isFrozen :: LVar a d -> IO Bool+isFrozen (LVar {status}) = do+ curStatus <- readIORef status+ case curStatus of+ Active _ -> return False+ Frozen -> return True+ +------------------------------------------------------------------------------+-- LVar operations+------------------------------------------------------------------------------+ +-- | Create an LVar+newLV :: IO a -> Par (LVar a d)+newLV init = mkPar $ \k q -> do+ state <- init+ listeners <- B.new+ status <- newIORef $ Active listeners+ name <- newLVID+ exec (k $ LVar {state, status, name}) q++-- | Do a threshold read on an LVar+getLV :: (LVar a d) -- ^ the LVar + -> (a -> Bool -> IO (Maybe b)) -- ^ already past threshold?+ -> (d -> IO (Maybe b)) -- ^ does d pass the threshold?+ -> Par b+getLV lv@(LVar {state, status}) globalThresh deltaThresh = mkPar $ \k q -> do+ -- tradeoff: we fastpath the case where the LVar is already beyond the+ -- threshhold by polling *before* enrolling the callback. The price is+ -- that, if we are not currently above the threshhold, we will have to poll+ -- /again/ after enrolling the callback. This race may also result in the+ -- continuation being executed twice, which is permitted by idempotence.++ curStatus <- readIORef status+ case curStatus of+ Frozen -> do + tripped <- globalThresh state True+ case tripped of+ Just b -> exec (k b) q -- already past the threshold; invoke the+ -- continuation immediately + Nothing -> sched q + Active listeners -> do+ tripped <- globalThresh state False+ case tripped of+ Just b -> exec (k b) q -- already past the threshold; invoke the+ -- continuation immediately ++ Nothing -> do -- /transiently/ not past the threshhold; block + +#if GET_ONCE+ execFlag <- newIORef False+#endif+ + let onUpdate d = unblockWhen $ deltaThresh d+ onFreeze = unblockWhen $ globalThresh state True+ + unblockWhen thresh tok q = do+ tripped <- thresh+ whenJust tripped $ \b -> do + B.remove tok+#if GET_ONCE+ ticket <- readForCAS execFlag+ unless (peekTicket ticket) $ do+ (winner, _) <- casIORef execFlag ticket True+ when winner $ Sched.pushWork q (k b) +#else + Sched.pushWork q (k b) +#endif+ + -- add listener, i.e., move the continuation to the waiting bag+ tok <- B.put listeners $ Listener onUpdate onFreeze++ -- but there's a race: the threshold might be passed (or the LVar+ -- frozen) between our check and the enrollment as a listener, so we+ -- must poll again+ frozen <- isFrozen lv+ tripped' <- globalThresh state frozen+ case tripped' of+ Just b -> do+ B.remove tok -- remove the listener we just added, and+ exec (k b) q -- execute the continuation. this work might be+ -- redundant, but by idempotence that's OK+ Nothing -> sched q+++-- | Update an LVar+putLV_ :: LVar a d -- ^ the LVar+ -> (a -> Par (Maybe d, b)) -- ^ how to do the put and whether the LVar's+ -- value changed+ -> Par b+putLV_ LVar {state, status, name} doPut = mkPar $ \k q -> do + Sched.setStatus q name -- publish our intent to modify the LVar+ let cont (delta, ret) = ClosedPar $ \q -> do+ curStatus <- readIORef status -- read the frozen bit *while q's status is marked*+ Sched.setStatus q noName -- retract our modification intent+ whenJust delta $ \d -> do+ case curStatus of+ Frozen -> E.throw$ PutAfterFreezeExn "Attempt to change a frozen LVar"+ Active listeners -> + B.foreach listeners $ \(Listener onUpdate _) tok -> onUpdate d tok q+ exec (k ret) q + exec (close (doPut state) cont) q -- possibly modify the LVar + +-- | Update an LVar without generating a result. +putLV :: LVar a d -- ^ the LVar+ -> (a -> IO (Maybe d)) -- ^ how to do the put, and whether the LVar's+ -- value changed+ -> Par ()+putLV lv doPut = putLV_ lv doPut'+ where doPut' a = do r <- liftIO (doPut a); return (r, ())++-- | Freeze an LVar (limited nondeterminism)+-- It is the data-structure implementors responsibility to expose this as qasi-deterministc.+freezeLV :: LVar a d -> Par ()+freezeLV LVar {name, status} = mkPar $ \k q -> do+ oldStatus <- atomicModifyIORef status $ \s -> (Frozen, s) + case oldStatus of+ Frozen -> return ()+ Active listeners -> do+ Sched.await q (name /=) -- wait until all currently-running puts have+ -- snapshotted the active status+ B.foreach listeners $ \Listener {onFreeze} tok -> onFreeze tok q+ exec (k ()) q+ +------------------------------------------------------------------------------+-- Handler pool operations+------------------------------------------------------------------------------ ++-- | Create a handler pool+newPool :: Par HandlerPool+newPool = mkPar $ \k q -> do+ cnt <- C.new+ bag <- B.new+ let hp = HandlerPool cnt bag+ hpMsg " [dbg-lvish] Created new pool" hp+ exec (k hp) q+ +-- | Convenience function. Execute a Par computation in the context of a fresh handler pool+withNewPool :: (HandlerPool -> Par a) -> Par (a, HandlerPool)+withNewPool f = do+ hp <- newPool+ a <- f hp+ return (a, hp)+ +-- | Convenience function. Execute a Par computation in the context of a fresh+-- handler pool, while ignoring the result of the computation+withNewPool_ :: (HandlerPool -> Par ()) -> Par HandlerPool+withNewPool_ f = do+ hp <- newPool+ f hp+ return hp++data DecStatus = HasDec | HasNotDec++-- | Close a Par task so that it is properly registered with a handler pool+closeInPool :: Maybe HandlerPool -> Par () -> IO ClosedPar+closeInPool Nothing c = return $ close c $ const (ClosedPar sched)+closeInPool (Just hp) c = do+ decRef <- newIORef HasNotDec -- in case the thread is duplicated, ensure+ -- that the counter is decremented only once+ -- on termination+ let cnt = numHandlers hp+ + tryDecRef = do -- attempt to claim the role of decrementer+ ticket <- readForCAS decRef+ case peekTicket ticket of+ HasDec -> return False+ HasNotDec -> do+ (firstToDec, _) <- casIORef decRef ticket HasDec+ return firstToDec+ + onFinishHandler _ = ClosedPar $ \q -> do+ shouldDec <- tryDecRef -- are we the first copy of the thread to+ -- terminate?+ when shouldDec $ do+ C.dec cnt -- record handler completion in pool+ quiescent <- C.poll cnt -- check for (transient) quiescence+ when quiescent $ do -- wake any threads waiting on quiescence+ hpMsg " [dbg-lvish] -> Quiescent now.. waking conts" hp + let invoke t tok = do+ B.remove tok+ Sched.pushWork q t + B.foreach (blockedOnQuiesce hp) invoke+ sched q+ C.inc $ numHandlers hp -- record handler invocation in pool+ return $ close c onFinishHandler -- close the task with a special "done"+ -- continuation that clears it from the+ -- handler pool++-- | Add a handler to an existing pool+{-# INLINE addHandler #-}+addHandler :: Maybe HandlerPool -- ^ pool to enroll in, if any+ -> LVar a d -- ^ LVar to listen to+ -> (a -> IO (Maybe (Par ()))) -- ^ initial callback+ -> (d -> IO (Maybe (Par ()))) -- ^ subsequent callbacks: updates+ -> Par ()+addHandler hp LVar {state, status} globalThresh updateThresh = + let spawnWhen thresh q = do+ tripped <- thresh+ whenJust tripped $ \cb -> do+ closed <- closeInPool hp cb+ Sched.pushWork q closed + onUpdate d _ q = spawnWhen (updateThresh d) q+ onFreeze _ _ = return () + in mkPar $ \k q -> do+ curStatus <- readIORef status + case curStatus of+ Active listeners -> -- enroll the handler as a listener+ do B.put listeners $ Listener onUpdate onFreeze; return ()+ Frozen -> return () -- frozen, so no need to enroll + spawnWhen (globalThresh state) q -- poll globally to see whether we should+ -- launch any callbacks now+ exec (k ()) q ++-- | Block until a handler pool is quiescent +quiesce :: HandlerPool -> Par ()+quiesce hp@(HandlerPool cnt bag) = mkPar $ \k q -> do+ hpMsg " [dbg-lvish] Begin quiescing pool, identity= " hp+ -- tradeoff: we assume that the pool is not yet quiescent, and thus enroll as+ -- a blocked thread prior to checking for quiescence+ tok <- B.put bag (k ())+ quiescent <- C.poll cnt+ if quiescent then do+ B.remove tok+ hpMsg " [dbg-lvish] -> Quiescent already!" hp+ exec (k ()) q + else do + hpMsg " [dbg-lvish] -> Not quiescent yet, back to sched" hp+ sched q++-- | A global barrier.+quiesceAll :: Par ()+quiesceAll = mkPar $ \k q -> do+ sched q+ logStrLn_ " [dbg-lvish] Return from global barrier."+ exec (k ()) q++-- | Freeze an LVar after a given handler quiesces+-- This is quasideterministic, but it +freezeLVAfter :: LVar a d -- ^ the LVar of interest+ -> (a -> IO (Maybe (Par ()))) -- ^ initial callback+ -> (d -> IO (Maybe (Par ()))) -- ^ subsequent callbacks: updates+ -> Par ()+freezeLVAfter lv globalCB updateCB = do+ let globalCB' = globalCB+ updateCB' = updateCB+ hp <- newPool+ addHandler (Just hp) lv globalCB' updateCB'+ quiesce hp+ freezeLV lv+ + ++------------------------------------------------------------------------------+-- Par monad operations+------------------------------------------------------------------------------++-- | Fork a child thread, optionally in the context of a handler pool+forkHP :: Maybe HandlerPool -> Par () -> Par ()+forkHP mh child = mkPar $ \k q -> do+ closed <- closeInPool mh child+ Sched.pushWork q (k ()) -- "Work-first" policy.+-- hpMsg " [dbg-lvish] incremented and pushed work in forkInPool, now running cont" hp + exec closed q + +-- | Fork a child thread+fork :: Par () -> Par ()+fork f = forkHP Nothing f++-- | Perform an IO action+liftIO :: IO a -> Par a+liftIO io = mkPar $ \k q -> do+ r <- io+ exec (k r) q+ +-- | Generate a random boolean in a core-local way. Fully nondeterministic!+instance MonadToss Par where + toss = mkPar $ \k q -> do + g <- readIORef $ Sched.prng q+ let (b, g' ) = random g+ writeIORef (Sched.prng q) g'+ exec (k b) q++-- | Cooperatively schedule other threads+yield :: Par () +yield = mkPar $ \k q -> do+ Sched.yieldWork q (k ())+ sched q+ +{-# INLINE sched #-}+sched :: SchedState -> IO ()+sched q = do+ n <- Sched.next q+ case n of+ Just t -> exec t q+ Nothing -> return ()++-- Forcing evaluation of a LVar is fruitless.+instance NFData (LVar a d) where+ rnf _ = ()++runPar_internal :: Par a -> IO a+runPar_internal c = do+ closeLogger <- if dbgLvl >= 1+ then printLogThread+ else return (return ()) + res <- runPar_internal2 c+ -- printLog+ closeLogger+ hFlush stdout+ return res++runPar_internal2 :: Par a -> IO a+runPar_internal2 c = do+ queues <- Sched.new numCapabilities noName+ + -- We create a thread on each CPU with forkOn. The CPU on which+ -- the current thread is running will host the main thread; the+ -- other CPUs will host worker threads.+ main_cpu <- Sched.currentCPU+ answerMV <- newEmptyMVar++#if 1+ wrkrtids <- newIORef []+ let forkit = forM_ (zip [0..] queues) $ \(cpu, q) -> do + tid <- forkWithExceptions (forkOn cpu) "worker thread" $+ if cpu == main_cpu + then let k x = ClosedPar $ \q -> do + sched q -- ensure any remaining, enabled threads run to + putMVar answerMV x -- completion prior to returning the result+ -- [TODO: ^ perhaps better to use a binary notification tree to signal the workers to stop...]+ in exec (close c k) q+ -- Note: The above is important: it is sketchy to leave any workers running after+ -- the main thread exits. Subsequent exceptions on child threads, even if+ -- forwarded asynchronously, can arrive much later at the main thread+ -- (e.g. after it has exited, or set up a new handler, etc).+ else sched q+ atomicModifyIORef_ wrkrtids (tid:)+ logStrLn_ " [dbg-lvish] About to fork workers..." + ans <- E.catch (forkit >> takeMVar answerMV)+ (\ (e :: E.SomeException) -> do + tids <- readIORef wrkrtids+ logStrLn_$ " [dbg-lvish] Killing off workers due to exception: "++show tids+ mapM_ killThread tids+ -- if length tids < length queues then do -- TODO: we could try to chase these down in the idle list.+ mytid <- myThreadId+ when (dbgLvl >= 1) printLog -- Unfortunately this races with the log printing thread.+ E.throw$ LVarSpecificExn ("EXCEPTION in runPar("++show mytid++"): "++show e)+ )+ logStrLn_ " [dbg-lvish] parent thread escaped unscathed"+ return ans+#else+ let runWorker (cpu, q) = do + if (cpu /= main_cpu)+ then sched q+ else let k x = ClosedPar $ \q -> do + sched q -- ensure any remaining, enabled threads run to + putMVar answerMV x -- completion prior to returning the result+ in exec (close c k) q++ -- Here we want a traditional, fork-join parallel loop with proper exception handling:+ let loop [] asyncs = mapM_ wait asyncs+ loop ((cpu,q):tl) asyncs = +-- withAsync (runWorker state)+ withAsyncOn cpu (runWorker (cpu,q))+ (\a -> loop tl (a:asyncs))++----------------------------------------+-- (1) There is a BUG in 'loop' presently:+-- "thread blocked indefinitely in an STM transaction"+-- loop (zip [0..] queues) []+----------------------------------------+-- (2) This has the same problem as 'loop':+-- ls <- mapM (\ pr@(cpu,_) -> Async.asyncOn cpu (runWorker pr)) (zip [0..] queues)+-- mapM_ wait ls+----------------------------------------+-- (3) Using this FOR NOW, but it does NOT pin to the right processors:+ mapConcurrently runWorker (zip [0..] queues)+----------------------------------------+ -- Now that child threads are done, it's safe for the main thread+ -- to call it quits.+ takeMVar answerMV +#endif+++-- | Run a deterministic parallel computation as pure.+runPar :: Par a -> a+runPar = unsafePerformIO . runPar_internal++-- | A version that avoids an internal `unsafePerformIO` for calling+-- contexts that are already in the `IO` monad.+runParIO :: Par a -> IO a+runParIO = runPar_internal++-- | Debugging aide. Return debugging logs, in realtime order, in addition to the+-- final result.+runParLogged :: Par a -> IO ([String],a)+runParLogged c =+ do res <- runPar_internal2 c+ lines <- atomicModifyIORef globalLog $ \ss -> ([], ss)+ return (reverse lines, res)++{-# INLINE atomicModifyIORef_ #-}+atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()+atomicModifyIORef_ ref fn = atomicModifyIORef ref (\ x -> (fn x,()))++{-# NOINLINE unsafeName #-}+unsafeName :: a -> Int+unsafeName x = unsafePerformIO $ do + sn <- makeStableName x+ return (hashStableName sn)++{-# INLINE hpMsg #-}+hpMsg msg hp = + when (dbgLvl >= 3) $ do+ s <- hpId_ hp+ logLnAt_ 3 $ msg++", pool identity= " ++s++{-# NOINLINE hpId #-} +hpId hp = unsafePerformIO (hpId_ hp)++hpId_ (HandlerPool cnt bag) = do+ sn1 <- makeStableName cnt+ sn2 <- makeStableName bag+ c <- readIORef cnt+ return $ show (hashStableName sn1) ++"/"++ show (hashStableName sn2) +++ " transient cnt "++show c+++-- | Exceptions that walk up the fork tree of threads:+forkWithExceptions :: (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId+forkWithExceptions forkit descr action = do + parent <- myThreadId+ forkit $ do+ tid <- myThreadId+ E.catch action+ (\ e -> + case E.fromException e of + Just E.ThreadKilled -> do+-- Killing worker threads is normal now when exception handling, so this chatter is restricted to debug mode:+#ifdef DEBUG_LVAR + printf "\nThreadKilled exception inside child thread, %s (not propagating!): %s\n" (show tid) (show descr)+#endif+ return ()+ _ -> do+#ifdef DEBUG_LVAR + printf "\nException inside child thread %s, %s: %s\n" (show descr) (show tid) (show e)+#endif+ E.throwTo parent (e :: E.SomeException)+ )+
+ Control/LVish/SchedIdempotentInternal.hs view
@@ -0,0 +1,187 @@+{-# LANGUAGE Unsafe #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE NamedFieldPuns, BangPatterns #-}+{-# LANGUAGE RecursiveDo #-}++module Control.LVish.SchedIdempotentInternal (+ State(), new, number, next, pushWork, yieldWork, currentCPU, setStatus, await, prng+ ) where+++import Prelude+import Control.Monad+import Control.Concurrent+import Control.DeepSeq+import Control.Applicative+import Data.IORef +import GHC.Conc+import System.Random (StdGen, mkStdGen)++#ifdef CHASE_LEV+#warning "Compiling with Chase-Lev work-stealing deque"++import Data.Concurrent.Deque.ChaseLev as CL++type Deque a = CL.ChaseLevDeque a+newDeque = CL.newQ+pushMine = CL.pushL+popMine = CL.tryPopL+popOther = CL.tryPopR +pushYield = pushMine -- for now... ++#else++------------------------------------------------------------------------------+-- A nonscalable deque for work-stealing+------------------------------------------------------------------------------++type Deque a = IORef [a]++-- | Create a new local work deque+newDeque :: IO (Deque a)+newDeque = newIORef []++-- | Add work to a thread's own work deque+pushMine :: Deque a -> a -> IO ()+pushMine deque t = + atomicModifyIORef deque $ \ts -> (t:ts, ())+ +-- | Take work from a thread's own work deque+popMine :: Deque a -> IO (Maybe a)+popMine deque = do+ atomicModifyIORef deque $ \ts ->+ case ts of+ [] -> ([], Nothing)+ (t:ts') -> (ts', Just t)++-- | Add low-priority work to a thread's own work deque+pushYield :: Deque a -> a -> IO ()+pushYield deque t = + atomicModifyIORef deque $ \ts -> (ts++[t], ()) ++-- | Take work from a different thread's work deque+popOther :: Deque a -> IO (Maybe a)+popOther = popMine++#endif++------------------------------------------------------------------------------+-- A scheduling framework+------------------------------------------------------------------------------++-- All the state relevant to a single worker thread+data State a s = State+ { no :: {-# UNPACK #-} !Int,+ prng :: IORef StdGen, -- core-local random number generation+ status :: IORef s,+ workpool :: Deque a, + idle :: IORef [MVar Bool], -- global list of idle workers+ states :: [State a s] -- global list of all worker states.+ }+ +-- | Process the next item on the work queue or, failing that, go into+-- work-stealing mode.+{-# INLINE next #-}+next :: State a s -> IO (Maybe a)+next state@State{ workpool } = do+ e <- popMine workpool+ case e of+ Nothing -> steal state+ Just t -> return e++-- RRN: Note -- NOT doing random work stealing breaks the traditional+-- Cilk time/space bounds if one is running strictly nested (series+-- parallel) programs.++-- | Attempt to steal work or, failing that, give up and go idle.+steal :: State a s -> IO (Maybe a)+steal State{ idle, states, no=my_no } = do+ -- printf "cpu %d stealing\n" my_no+ go states+ where+ go [] = do m <- newEmptyMVar+ r <- atomicModifyIORef idle $ \is -> (m:is, is)+ if length r == numCapabilities - 1+ then do+ -- printf "cpu %d initiating shutdown\n" my_no+ mapM_ (\m -> putMVar m True) r+ return Nothing+ else do+ done <- takeMVar m+ if done+ then do+ -- printf "cpu %d shutting down\n" my_no+ return Nothing+ else do+ -- printf "cpu %d woken up\n" my_no+ go states+ go (x:xs)+ | no x == my_no = go xs+ | otherwise = do+ r <- popOther (workpool x)+ case r of+ Just t -> do+ -- printf "cpu %d got work from cpu %d\n" my_no (no x)+ return r+ Nothing -> go xs++-- | If any worker is idle, wake one up and give it work to do.+pushWork :: State a s -> a -> IO ()+pushWork State { workpool, idle } t = do+ pushMine workpool t+ idles <- readIORef idle+ when (not (null idles)) $ do+ r <- atomicModifyIORef idle (\is -> case is of+ [] -> ([], return ())+ (i:is) -> (is, putMVar i False))+ r -- wake one up+ +yieldWork :: State a s -> a -> IO ()+yieldWork State { workpool } t = + pushYield workpool t -- AJT: should this also wake an idle thread?++new :: Int -> s -> IO [State a s]+new n s = do+ idle <- newIORef []+ let mkState states i = do + workpool <- newDeque+ status <- newIORef s+ prng <- newIORef $ mkStdGen i+ return State { no = i, workpool, idle, status, states, prng }+ rec states <- forM [0..(n-1)] $ mkState states+ return states++number :: State a s -> Int+number State { no } = no++setStatus :: State a s -> s -> IO ()+setStatus State { status } s = writeIORef status s++await :: State a s -> (s -> Bool) -> IO ()+await State { states } p = + let awaitOne state@(State { status }) = do+ cur <- readIORef status+ unless (p cur) $ awaitOne state+ in mapM_ awaitOne states++-- | the CPU executing the current thread (0 if not supported)+currentCPU :: IO Int+currentCPU = +#if __GLASGOW_HASKELL__ >= 701 /* 20110301 */+ --+ -- Note: GHC 7.1.20110301 is required for this to work, because that+ -- is when threadCapability was added.+ --+ do + tid <- myThreadId+ (main_cpu, _) <- threadCapability tid+ return main_cpu+#else+ --+ -- Lacking threadCapability, we always pick CPU #0 to run the main+ -- thread. If the current thread is not running on CPU #0, this+ -- will require some data to be shipped over the memory bus, and+ -- hence will be slightly slower than the version above.+ --+ return 0+#endif
+ Control/LVish/Types.hs view
@@ -0,0 +1,20 @@+{-# LANGUAGE DeriveDataTypeable #-}++-- | A simple internal module to factor out types that are used in many places.+module Control.LVish.Types+ (LVishException(..))+ where++import Data.Typeable (Typeable)+import Control.Exception++-- | All @LVar@s share a common notion of exceptions.+-- The two common forms of exception currently are conflicting-put and put-after-freeze.+-- There are also errors that correspond to particular invariants for particular LVars.+data LVishException = ConflictingPutExn String+ | PutAfterFreezeExn String+ | LVarSpecificExn String+ deriving (Show, Read, Eq, Ord, Typeable)++instance Exception LVishException +
+ Control/Reagent.hs view
@@ -0,0 +1,42 @@+{-# LANGUAGE ExplicitForAll, Rank2Types #-} ++-- | An implementation of Reagents (http://www.mpi-sws.org/~turon/reagents.pdf)+-- NOTE: currently this is just a very tiny core of the Reagent design. Needs+-- lots of work.++module Control.Reagent+where+ +import Data.IORef +import Data.Atomics++type Reagent a = forall b. (a -> IO b) -> IO b -> IO b++-- | Execute a Reagent.+{-# INLINE react #-}+react :: Reagent a -> IO a+react r = try where + try = r finish try+ finish x = return x+ +-- | Like atomicModifyIORef, but uses CAS and permits the update action to force+-- a retry by returning Nothing + +{-# INLINE atomicUpdate #-}+atomicUpdate :: IORef a -> (a -> Maybe (a, b)) -> Reagent b +atomicUpdate r f succ fail = do+ curTicket <- readForCAS r+ let cur = peekTicket curTicket+ case f cur of+ Just (new, out) -> do+ (done, _) <- casIORef r curTicket new+ if done then succ out else fail+ Nothing -> fail+atomicUpdate_ :: IORef a -> (a -> a) -> Reagent ()+atomicUpdate_ r f = atomicUpdate r (\x -> Just (f x, ()))+ +postCommit :: Reagent a -> (a -> IO b) -> Reagent b+postCommit r f succ fail = r (\x -> f x >>= succ) fail++choice :: Reagent a -> Reagent a -> Reagent a+choice = error "TODO"
+ Data/Concurrent/AlignedIORef.hs view
@@ -0,0 +1,24 @@+{-# LANGUAGE NamedFieldPuns #-}++-- | Cacheline-aligned wrappers around IORefs. Currently doing nothing.++module Data.Concurrent.AlignedIORef (AlignedIORef(), newAlignedIORef, ref)+where+ +import Data.IORef +import Control.Monad++data AlignedIORef a = AlignedIORef {+ -- pad out to 64 bytes to avoid false sharing (assuming 4 byte words and 64+ -- byte cachelines)+ -- padding :: [IORef a], + ref :: {-# UNPACK #-} !(IORef a)+}++newAlignedIORef v = do+ ref <- newIORef v+-- padding <- replicateM 15 $ newIORef v+ return AlignedIORef {+-- padding,+ ref+ }
+ Data/Concurrent/Bag.hs view
@@ -0,0 +1,51 @@+module Data.Concurrent.Bag(Bag, Token, new, put, remove, foreach) where++import Control.Monad+import Control.Concurrent+import System.IO.Unsafe (unsafePerformIO)+import Data.IORef+import qualified Data.Map as M++------------------------------------------------------------------------------+-- A nonscalable implementation of a concurrent bag+------------------------------------------------------------------------------++type UID = Int+type Token a = (Bag a, UID)+type Bag a = IORef (M.Map UID a)++-- Return the old value. Could replace with a true atomic op.+atomicIncr :: IORef Int -> IO Int+atomicIncr cntr = atomicModifyIORef' cntr (\c -> (c+1,c))++uidCntr :: IORef UID+uidCntr = unsafePerformIO (newIORef 0)++getUID :: IO UID+getUID = atomicIncr uidCntr++-- | Create an empty bag+new :: IO (Bag a)+new = newIORef (M.empty)++-- | Add an element to a bag, returning a token that can later be used to remove+-- that element.+put :: Bag a -> a -> IO (Token a)+put b x = do+ uid <- getUID+ atomicModifyIORef' b $ \m -> (M.insert uid x m, ()) + return (b, uid)++-- | foreach b f will traverse b (concurrently with updates), applying f to each+-- encountered element, together with a token that can be used to remove the+-- element.+foreach :: Bag a -> (a -> Token a -> IO ()) -> IO ()+foreach b f = do+ m <- readIORef b+ let invoke (k, a) = f a (b, k)+ mapM_ invoke $ M.toList m++-- | Remove the element associated with a given token. Repeated removals are+-- permitted.+remove :: Token a -> IO ()+remove (b, uid) = atomicModifyIORef' b $ \m -> (M.delete uid m, ())
+ Data/Concurrent/Counter.hs view
@@ -0,0 +1,22 @@+module Data.Concurrent.Counter(Counter, new, inc, dec, poll) where++import Control.Monad+import Control.Concurrent+import Data.IORef++type Counter = IORef Int++new :: IO Counter+new = newIORef 0++inc :: Counter -> IO ()+inc c = atomicModifyIORef' c $ \n -> (n+1,())++dec :: Counter -> IO ()+dec c = atomicModifyIORef' c $ \n -> (n-1,())++-- | Is the counter (transiently) zero?+poll :: Counter -> IO Bool+poll c = do+ n <- readIORef c+ return (n == 0)
+ Data/Concurrent/LinkedMap.hs view
@@ -0,0 +1,117 @@+{-# LANGUAGE NamedFieldPuns, BangPatterns #-}++-- | A concurrent finite map represented as a single linked list. +--+-- In contrast to standard maps, this one only allows lookups and insertions,+-- not modifications or removals. While modifications would be fairly easy to+-- add, removals would significantly complicate the logic, and aren't needed for+-- the primary application -- LVars.+--+-- The interface is also somewhat low-level: rather than a standard insert+-- function, @tryInsert@ takes a "token" (i.e. a pointer into the linked list)+-- and attempts to insert at that location (but may fail). Tokens are acquired+-- through the @find@ function, which yields a token in the case that a key is+-- *not* found; the token represents the location in the list where the key+-- *should* go. This low-level interface is intended for use in higher-level+-- data structures, e.g. SkipListMap.++module Data.Concurrent.LinkedMap (+ LMap(), newLMap, Token(), value, find, FindResult(..), tryInsert,+ foldlWithKey, map, reverse)+where+ +import Data.IORef+import Data.Atomics +import Control.Reagent -- AT: not yet using this, but would be nice to refactor+ -- to use it.+import Control.Monad.IO.Class+import Prelude hiding (reverse, map)++-- | A concurrent finite map, represented as a linked list+data LMList k v = + Node k v {-# UNPACK #-} !(IORef (LMList k v))+ | Empty ++type LMap k v = IORef (LMList k v)++-- | Create a new concurrent map+newLMap :: IO (LMap k v)+newLMap = newIORef Empty+ +-- | A position in the map into which a key/value pair can be inserted +data Token k v = Token {+ keyToInsert :: k, -- ^ what key were we looking up?+ value :: Maybe v, -- ^ the value at this position in the map+ nextRef :: IORef (LMList k v), -- ^ the reference at which to insert+ nextTicket :: Ticket (LMList k v) -- ^ a ticket for the old value of nextRef+}++-- | Either the value associated with a key, or else a token at the position+-- where that key should go.+data FindResult k v =+ Found v+ | NotFound (Token k v)++-- | Attempt to locate a key in the map+{-# INLINE find #-}+find :: Ord k => LMap k v -> k -> IO (FindResult k v)+find m k = findInner m Nothing + where + findInner m v = do+ nextTicket <- readForCAS m+ let stopHere = NotFound $ Token {keyToInsert = k, value = v, nextRef = m, nextTicket}+ case peekTicket nextTicket of+ Empty -> return stopHere+ Node k' v' next -> + case compare k k' of+ LT -> return stopHere+ EQ -> return $ Found v'+ GT -> findInner next (Just v')+ +-- | Attempt to insert a key/value pair at the given location (where the key is+-- given by the token). NB: tryInsert will *always* fail after the first attempt.+-- If successful, returns a (mutable!) view of the map beginning at the given key. +{-# INLINE tryInsert #-} +tryInsert :: Token k v -> v -> IO (Maybe (LMap k v))+tryInsert Token { keyToInsert, nextRef, nextTicket } v = do+ newRef <- newIORef $ peekTicket nextTicket+ (success, _) <- casIORef nextRef nextTicket $ Node keyToInsert v newRef+ return $ if success then Just nextRef else Nothing++-- | Concurrently fold over all key/value pairs in the map within the given+-- monad, in increasing key order. Inserts that arrive concurrently may or may+-- not be included in the fold.+foldlWithKey :: MonadIO m => (a -> k -> v -> m a) -> a -> LMap k v -> m a+foldlWithKey f a m = do+ n <- liftIO $ readIORef m+ case n of+ Empty -> return a+ Node k v next -> do+ a' <- f a k v+ foldlWithKey f a' next+++-- | Map over a snapshot of the list. Inserts that arrive concurrently may or may+-- not be included. This does not affect keys, so the physical structure remains the+-- same.+map :: MonadIO m => (a -> b) -> LMap k a -> m (LMap k b)+map fn mp = do + tmp <- foldlWithKey (\ acc k v -> do+ r <- liftIO (newIORef acc)+ return$! Node k (fn v) r)+ Empty mp+ tmp' <- liftIO (newIORef tmp)+ -- Here we suffer a reverse to avoid blowing the stack. + reverse tmp'++-- | Create a new linked map that is the reverse order from the input.+reverse :: MonadIO m => LMap k v -> m (LMap k v)+reverse mp = liftIO . newIORef =<< loop Empty mp+ where+ loop !acc mp = do+ n <- liftIO$ readIORef mp+ case n of+ Empty -> return acc+ Node k v next -> do+ r <- liftIO (newIORef acc)+ loop (Node k v r) next
+ Data/Concurrent/SNZI.hs view
@@ -0,0 +1,109 @@+-- | A Scalable Non-Zero Indicator+--+-- A SNZI is a kind of concurrent counter which can be incremented, decremented,+-- and queried for equality with 0. The interface is a bit more complex,+-- though: it is exposed as N values (where N = the number of CPUs), each+-- providing an @arrive@ and @depart@ operation, together with a single polling+-- action querying the value of the counter. The client MUST NOT invoke+-- @depart@ more times than @arrive@ on any single value.+--+-- The implementation is based on http://dl.acm.org/citation.cfm?id=1281106, but+-- significantly simplified by allowing a call to @arrive@ to block indefinitely+-- until other such calls complete. (Thus the algorithm is no longer+-- non-blocking in theory; its liveness depends on assuming that the OS-level+-- thread scheduler is fair.)+--+-- The basic design is to have a *tree* of counters; each child node in the tree+-- is allowed to invoke @arrive@/@depart@ on its parents. There are two invariants:+-- +-- * The number of @depart@s (decrements) must never outnumber the @arrive@s+-- (increments) at any point in the tree. This invariant is partially+-- dependent on the client, which must ensure it for the exposed leaf+-- counters.+--+-- * The number of @arrive@s a child has invoked on a parent can outnumber the+-- @depart@s iff the total number of arrives at the child outnumbers the departs+-- at the child.+--+-- The idea is that child nodes act as "filters": they only need to invoke+-- @arrive@/@depart@ on their parents when their own value changes from 0 to 1 or 1+-- to 0 (i.e., when they change to/from having a surplus).+--+-- To maintain the above invariants, however, child nodes use a special+-- representation: if n >= 0, it represents the counter, but if n = -1 the child+-- is "locked". The locked value is needed to handle races between @arrive@s+-- when the node is currently at 0. The thread that wins the race will move the+-- counter from 0 to -1, thereby effectively "locking" it. Subsequently, it+-- will invoke @arrive@ on the parent, and then finally "unlock" the counter by+-- setting it to the value 1. See the paper for details on why a protocol like+-- this is needed (the paper uses a more complex, lock-free protocol). Such a+-- protocol is *not* needed for @depart@, however.++module Data.Concurrent.SNZI+where+ +import System.IO.Unsafe+import Control.Reagent +import Control.Monad+import GHC.Conc+import Data.IORef+import Data.Atomics+import Data.Concurrent.AlignedIORef+ +-- | An entry point for a shared SNZI value+data SNZI = + Child (AlignedIORef Int) SNZI+ | Root (AlignedIORef Int)++-- | Signal the presence of a thread at a SNZI+arrive :: SNZI -> IO () +arrive (Root cnt) = react $ atomicUpdate_ (ref cnt) (+1)+arrive (Child cnt parent) = + let upd 0 = Just (-1, True)+ upd (-1) = Nothing+ upd n = Just (n+1, False)+ in do+ tellParent <- react $ atomicUpdate (ref cnt) upd+ when tellParent $ do+ arrive parent+ writeBarrier+ writeIORef (ref cnt) 1+ +data TellParent = Yes | No | Err+ +-- | Signal the departure of a thread at a SNZI. IMPORTANT: depart MUST NOT be+-- called more times than arrive for a given SNZI value.+depart :: SNZI -> IO () +depart (Root cnt) = react $ atomicUpdate_ (ref cnt) (\x -> x-1)+depart (Child cnt parent) = + let upd 0 = Just (0, Err)+ upd (-1) = Nothing+ upd 1 = Just (0, Yes)+ upd n = Just (n-1, No)+ in do+ tellParent <- react $ atomicUpdate (ref cnt) upd+ case tellParent of+ No -> return ()+ Yes -> depart parent+ Err -> do putStrLn "SNZI BUG: departs outnumber arrives"+ error "SNZI BUG: departs outnumber arrives"+ +-- Helper function to generate a tree of SNZI values.+makeTree :: Int -> [SNZI] -> [SNZI] -> IO [SNZI]+makeTree n parents children = + if n >= numCapabilities then return children + else case parents of + [] -> makeTree 0 children []+ (parent:parents') -> do+ c1 <- newAlignedIORef 0+ c2 <- newAlignedIORef 0+ makeTree (n+2) parents' $ Child c1 parent : Child c2 parent : children+ +-- | Create a shared SNZI values with numCapabilities number of entry points,+-- together with a polling action that returns "true" when no threads are+-- present.+newSNZI :: IO ([SNZI], IO Bool)+newSNZI = do+ rootRef <- newAlignedIORef 0+ leaves <- makeTree 1 [] [Root rootRef]+ return (leaves, readIORef (ref rootRef) >>= return . (== 0))
+ Data/Concurrent/SkipListMap.hs view
@@ -0,0 +1,203 @@+{-# LANGUAGE ExistentialQuantification, GADTs #-}++-- | An implementation of concurrent finite maps based on skip lists. Only+-- supports lookup and insertions, not modifications or removals.+--+-- Skip lists are a probabilistic data structure that roughly approximate+-- balanced trees. At the bottom layer is a standard linked list representation+-- of a finite map. Above this is some number of "index" lists that provide+-- shortcuts to the layer below them. When a key/value pair is added, it is+-- always added to the bottom layer, and is added with exponentially decreasing+-- probability to each index layer above it.+--+-- Skip lists are a very good match for lock-free programming, since the+-- linearization point can be taken as insertion into the bottom list, and index+-- nodes can be added *afterward* in a best-effort style (i.e., if there is+-- contention to add them, we can simply walk away, with the effect that the+-- probability of appearing in an index is partly a function of contention.)+-- +-- To implement skip lists in Haskell, we use a GADT to represent the layers,+-- each of which has a different type (since it indexes the layer below it).++module Data.Concurrent.SkipListMap (+ SLMap(), newSLMap, find, PutResult(..), putIfAbsent, putIfAbsentToss, foldlWithKey, counts+ -- map: is not exposed, because it has that FINISHME for now... [2013.10.01]+ )+where+ +import System.Random ++import Control.Applicative ((<$>))+import Control.Monad +import Control.Monad.IO.Class+import Control.LVish.MonadToss+import Control.LVish (Par)+ +import Data.IORef+import Data.Atomics+import qualified Data.Concurrent.LinkedMap as LM+import Prelude hiding (map)+++-- | The GADT representation. The type @t@ gives the type of nodes at a given+-- level in the skip list.+data SLMap_ k v t where+ Bottom :: LM.LMap k v -> SLMap_ k v (LM.LMap k v)+ Index :: LM.LMap k (t, v) -> SLMap_ k v t -> SLMap_ k v (LM.LMap k (t, v))++-- The complete multi-level SLMap always keeps a pointer to the bottom level (the+-- second field).+data SLMap k v = forall t. SLMap (SLMap_ k v t) (LM.LMap k v)++-- | Physical identity+instance Eq (SLMap k v) where+ SLMap _ lm1 == SLMap _ lm2 = lm1 == lm2++-- | Create a new skip list with the given number of levels.+newSLMap :: Int -> IO (SLMap k v)+newSLMap 0 = do+ lm <- LM.newLMap+ return $ SLMap (Bottom lm) lm+newSLMap n = do + SLMap slm lmBottom <- newSLMap (n-1)+ lm <- LM.newLMap+ return $ SLMap (Index lm slm) lmBottom++-- | Attempt to locate a key in the map.+find :: Ord k => SLMap k v -> k -> IO (Maybe v) +find (SLMap slm _) k = find_ slm Nothing k++-- Helper for @find@.+find_ :: Ord k => SLMap_ k v t -> Maybe t -> k -> IO (Maybe v)++-- At the bottom level: just lift the find from LinkedMap+find_ (Bottom m) shortcut k = do+ searchResult <- LM.find (maybe m id shortcut) k+ case searchResult of+ LM.Found v -> return $ Just v+ LM.NotFound tok -> return Nothing+ +-- At an indexing level: attempt to use the index to shortcut into the level+-- below. +find_ (Index m slm) shortcut k = do + searchResult <- LM.find (maybe m id shortcut) k+ case searchResult of + LM.Found (_, v) -> + return $ Just v -- the key is in the index itself; we're outta here+ LM.NotFound tok -> case LM.value tok of+ Just (m', _) -> find_ slm (Just m') k -- there's an index node+ -- preceeding our key; use it to+ -- shortcut into the level below.+ + Nothing -> find_ slm Nothing k -- no smaller key in the index,+ -- so start at the beginning of+ -- the level below.+ +data PutResult v = Added v | Found v++{-# SPECIALIZE putIfAbsent :: (Ord k) => SLMap k v -> k -> Par d s v -> Par d s (PutResult v) #-}++-- | Adds a key/value pair if the key is not present, all within a given monad.+-- Returns the value now associated with the key in the map.+putIfAbsent :: (Ord k, MonadIO m, MonadToss m) => + SLMap k v -- ^ The map+ -> k -- ^ The key to lookup/insert+ -> m v -- ^ A computation of the value to insert+ -> m (PutResult v)+putIfAbsent (SLMap slm _) k vc = + putIfAbsent_ slm Nothing k vc toss $ \_ _ -> return ()++{-# SPECIALIZE putIfAbsentToss :: (Ord k) => + SLMap k v -> k -> Par d s v -> Par d s Bool -> Par d s (PutResult v) #-}++-- | Adds a key/value pair if the key is not present, all within a given monad.+-- Returns the value now associated with the key in the map.+putIfAbsentToss :: (Ord k, MonadIO m) => SLMap k v -- ^ The map+ -> k -- ^ The key to lookup/insert+ -> m v -- ^ A computation of the value to insert+ -> m Bool -- ^ An explicit, thread-local coin to toss+ -> m (PutResult v)+putIfAbsentToss (SLMap slm _) k vc coin = + putIfAbsent_ slm Nothing k vc coin $ \_ _ -> return () + +-- Helper for putIfAbsent+putIfAbsent_ :: (Ord k, MonadIO m) => + SLMap_ k v t -- ^ The map + -> Maybe t -- ^ A shortcut into this skiplist level+ -> k -- ^ The key to lookup/insert+ -> m v -- ^ A computation of the value to insert+ -> m Bool -- ^ A (thread-local) coin tosser+ -> (t -> v -> m ()) -- ^ A thunk for inserting into the higher+ -- levels of the skiplist+ -> m (PutResult v)+ +-- At the bottom level, we use a retry loop around the find/tryInsert functions+-- provided by LinkedMap+putIfAbsent_ (Bottom m) shortcut k vc coin install = retryLoop vc where + -- The retry loop; ensures that vc is only executed once+ retryLoop vc = do+ searchResult <- liftIO $ LM.find (maybe m id shortcut) k+ case searchResult of+ LM.Found v -> return $ Found v+ LM.NotFound tok -> do+ v <- vc+ maybeMap <- liftIO $ LM.tryInsert tok v+ case maybeMap of+ Just m' -> do+ install m' v -- all set on the bottom level, now try indices+ return $ Added v+ Nothing -> retryLoop $ return v -- next time around, remember the value to insert+ +-- At an index level; try to shortcut into the level below, while remembering+-- where we were so that we can insert index nodes later on+putIfAbsent_ (Index m slm) shortcut k vc coin install = do + searchResult <- liftIO $ LM.find (maybe m id shortcut) k+ case searchResult of + LM.Found (_, v) -> return $ Found v -- key is in the index; bail out+ LM.NotFound tok -> + let install' mBelow v = do -- to add an index node here,+ shouldAdd <- coin -- first, see if we (probabilistically) should+ when shouldAdd $ do + maybeHere <- liftIO $ LM.tryInsert tok (mBelow, v) -- then, try it!+ case maybeHere of+ Just mHere -> install mHere v -- if we succeed, keep inserting+ -- into the levels above us+ + Nothing -> return () -- otherwise, oh well; we tried.+ in case LM.value tok of+ Just (m', _) -> putIfAbsent_ slm (Just m') k vc coin install'+ Nothing -> putIfAbsent_ slm Nothing k vc coin install'++-- | Concurrently fold over all key/value pairs in the map within the given+-- monad, in increasing key order. Inserts that arrive concurrently may or may+-- not be included in the fold.+foldlWithKey :: MonadIO m => (a -> k -> v -> m a) -> a -> SLMap k v -> m a+foldlWithKey f a (SLMap _ lm) = LM.foldlWithKey f a lm++-- | Create an identical copy of an (unchanging) SLMap with the keys unchanged and+-- the values replaced by the result of applying the provided function.+-- map :: MonadIO m => (a -> b) -> SLMap k a -> m (SLMap k b)+map :: MonadIO m => (a -> a) -> SLMap k a -> m (SLMap k a)+map fn (SLMap (Bottom lm) lm2) = do+ lm' <- LM.map fn lm+ return$! SLMap (Bottom lm') lm'++map fn (SLMap (Index lm slm) lmbot) = do+ SLMap slm2 bot2 <- map fn (SLMap slm lmbot)+ lm2 <- LM.map (\(t,a) -> (t,fn a)) lm+ error "FINISHME -- SkipListMap.map"+-- return$! SLMap (Index lm2 slm2) bot2+++-- | Returns the sizes of the skiplist levels; for performance debugging.+counts :: SLMap k v -> IO [Int]+counts (SLMap slm _) = counts_ slm++counts_ :: SLMap_ k v t -> IO [Int]+counts_ (Bottom m) = do+ c <- LM.foldlWithKey (\n _ _ -> return (n+1)) 0 m+ return [c]+counts_ (Index m slm) = do+ c <- LM.foldlWithKey (\n _ _ -> return (n+1)) 0 m+ cs <- counts_ slm+ return $ c:cs
+ Data/LVar/Generic.hs view
@@ -0,0 +1,80 @@+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE MagicHash #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-} -- For Determinism++-- | A generic interface providing operations that work on ALL LVars.++module Data.LVar.Generic+ (+ -- * The classes containing the generic interfaces+ LVarData1(..), OrderedLVarData1(..),+ + -- * Supporting types and utilities+ AFoldable(..),+ castFrzn, forFrzn+ )+ where++import Control.LVish+import Control.LVish.DeepFrz.Internal (Frzn, Trvrsbl)+import qualified Data.Foldable as F+import Data.List (sort)+import GHC.Prim (unsafeCoerce#)+import System.IO.Unsafe (unsafeDupablePerformIO)+import Data.LVar.Generic.Internal++--------------------------------------------------------------------------------++-- |/Some LVar datatypes are stored in an /internally/ ordered way so+-- that it is then possible to take /O(1)/ frozen snapshots and consume them+-- inexpensively in a deterministic order.+--+-- LVars with this additional property provide this class as well as `LVarData1`.+class LVarData1 f => OrderedLVarData1 (f :: * -> * -> *) where+ -- | Don't just freeze the LVar, but make the full contents+ -- completely available and Foldable. Guaranteed /O(1)/.+ snapFreeze :: f s a -> Par QuasiDet s (f Trvrsbl a)++{- +-- | Just like LVarData1 but for type constructors of kind `*`.+class LVarData0 (t :: *) where+ -- | This associated type models a picture of the "complete" contents of the data:+ -- e.g. a whole set instead of one element, or the full/empty information for an+ -- IVar, instead of just the payload.+ type Snapshot0 t+ freeze0 :: t -> Par QuasiDet s (Snapshot0 t)+ newBottom0 :: Par d s t+-}+++------------------------------------------------------------------------------+-- Dealing with frozen LVars.+------------------------------------------------------------------------------++-- | `Trvrsbl` is a stronger property than `Frzn` so it is always ok to \"upcast\" to+-- the weaker version.+castFrzn :: LVarData1 f => f Trvrsbl a -> f Frzn a+castFrzn x = unsafeCoerceLVar x++-- | LVish Par actions must commute, therefore one safe way to consume a frozen (but+-- unordered) LVar, /even in another runPar session/, is to run a par computation for+-- each element.+forFrzn :: LVarData1 f => f Frzn a -> (a -> Par d s ()) -> Par d s ()+forFrzn fzn fn =+ F.foldrM (\ a () -> fn a) () $ + unsafeDupablePerformIO $ -- ASSUME idempotence.+ unsafeTraversable fzn+++-- | For any LVar, we have a generic way to freeze it in a `runParThenFreeze`.+-- instance (DeepFrz a, LVarData1 f) => DeepFrz (f s a) where+-- type FrzType (f s a) = f Frzn a +-- frz = unsafeCoerceLVar++-- ^^^++-- Note that this doesn't work because it CONFLICTS with the other DeepFrz instances.+-- There's no way that we can prove to GHC that pure data will NEVER be an instance+-- of LVarData1, and therefore will never actually cause a conflict with e above+-- instance.
+ Data/LVar/Generic/Internal.hs view
@@ -0,0 +1,94 @@+{-# LANGUAGE Unsafe #-}+{-# LANGUAGE MagicHash #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE DataKinds #-} -- For Determinism+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}++{-|++This module contains the unsafe bits that we cannot expose from + "Data.LVar.Generic".++-}++module Data.LVar.Generic.Internal+ (LVarData1(..), AFoldable(..),+ unsafeCoerceLVar, unsafeTraversable)+ where++import Control.LVish+import Control.LVish.DeepFrz.Internal (Frzn, Trvrsbl)+import qualified Data.Foldable as F+import Data.List (sort, intersperse)+import GHC.Prim (unsafeCoerce#)+import System.IO.Unsafe (unsafeDupablePerformIO)++------------------------------------------------------------------------------+-- Interface for generic LVar handling+------------------------------------------------------------------------------++-- | A class representing monotonic data types that take one type+-- parameter as well as an `s` parameter for session safety.+--+-- LVars that fall into this class are typically collection types.+class (F.Foldable (f Trvrsbl)) => LVarData1 (f :: * -> * -> *)+ -- TODO: if there is a Par class to generalize LVar Par monads, then+ -- it needs to be a superclass of this.+ where + -- type LVCtxt (f :: * -> * -> *) (s :: *) (a :: *) :: Constraint+ -- I was not able to get abstracting over the constraints to work.++ -- | Add a handler function which is called whenever an element is+ -- added to the LVar.+ addHandler :: Maybe HandlerPool -> f s elt -> (elt -> Par d s ()) -> Par d s ()++ -- | An /O(1)/ operation that atomically switches the LVar into a+ -- frozen state. Any threads waiting on the freeze are woken.+ --+ -- The frozen LVar provides a complete picture of the contents:+ -- e.g. a whole set instead of one element, or the full/empty+ -- information for an IVar, instead of just the payload.+ --+ -- However, note that `Frzn` LVars cannot be folded, because they may have+ -- nondeterministic ordering after being frozen. See `sortFreeze`.+ freeze :: -- LVCtxt f s a =>+ f s a -> Par QuasiDet s (f Frzn a)++ -- | Perform a freeze followed by a /sort/ operation which guarantees+ -- that the elements produced will be produced in a deterministic order.+ -- The result is fully accessible to the user (`Foldable`).+ sortFrzn :: Ord a => f Frzn a -> AFoldable a+ sortFrzn lv = + let lv3 :: f Trvrsbl a+ lv3 = unsafeCoerceLVar lv+ ls = F.foldr (:) [] lv3+ ls' = sort ls+ -- Without a traversible instance we cannot reconstruct an ordered+ -- version of the LVar contents with its original type:+ in AFoldable ls'++-- | Carries a Foldable type, but you don't get to know which one.+-- The purpose of this type is that `sortFreeze` should not have+-- to impose a particular memory representation.+data AFoldable a = forall f2 . F.Foldable f2 => AFoldable (f2 a)++instance Show a => Show (AFoldable a) where+ show (AFoldable col) =+ "AFoldable ["++ (concat$ intersperse ", " $ map show $ F.foldr (:) [] col)++"]"++--------------------------------------------------------------------------------++{-# INLINE unsafeCoerceLVar #-}+-- | A safer version of `unsafeCoerce#` for LVars only.+-- Note that it needs to change the contents type, because freezing is recursive.+unsafeCoerceLVar :: LVarData1 f => f s1 a -> f s2 b+unsafeCoerceLVar = unsafeCoerce#++-- | Here we gain permission to expose the non-deterministic internal structure of an+-- LVar: namely, the order in which elements occur. We pay the piper with an IO+-- action.+unsafeTraversable :: LVarData1 f => f Frzn a -> IO (f Trvrsbl a)+unsafeTraversable x = return (unsafeCoerceLVar x) +
+ Data/LVar/IStructure.hs view
@@ -0,0 +1,201 @@+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE InstanceSigs #-}++-- | An I-Structure, aka an Array of IVars.+-- This uses a boxed array.++module Data.LVar.IStructure+ (+ IStructure,+ + -- * Basic operations + newIStructure, newIStructureWithCallback,+ put, put_, get, getLength,++ -- * Iteration and callbacks+ forEachHP+ -- forEach, + ) where++import Data.Vector as V++import Control.DeepSeq (NFData)+import Control.Applicative+import Data.Maybe (fromJust, isJust)+import qualified Data.LVar.IVar as IV+import Data.LVar.IVar (IVar(IVar))+import qualified Data.Foldable as F+import Data.List (intersperse)+-- import qualified Data.Traversable as T++import Control.LVish as LV +import Control.LVish.DeepFrz.Internal+import Control.LVish.Internal as LI+import Control.LVish.SchedIdempotent (newLV, putLV, getLV, freezeLV,+ freezeLVAfter, liftIO)+import Data.LVar.Generic as G+import Data.LVar.Generic.Internal (unsafeCoerceLVar)++------------------------------------------------------------------------------++-- | An I-Structure, aka an Array of IVars.+-- For now this really is a simple vector of IVars.+newtype IStructure s a = IStructure (V.Vector (IV.IVar s a))++-- unIStructure (IStructure lv) = lv++instance Eq (IStructure s v) where+ IStructure vec1 == IStructure vec2 = vec1 == vec2++-- | An @IStructure@ can be treated as a generic container LVar. However, the+-- polymorphic operations are less useful than the monomorphic ones exposed by this+-- module (e.g. @forEachHP@ vs. @addHandler@).+instance LVarData1 IStructure where+ freeze orig@(IStructure vec) = WrapPar$ do+ -- No new alloc here, just time:+ V.forM_ vec $ \ (IVar (WrapLVar lv)) -> freezeLV lv + return (unsafeCoerceLVar orig)++ -- | We can do better than the default here; this is /O(1)/: + sortFrzn = AFoldable+ + -- Unlike the IStructure-specific forEach, this takes only values, not indices.+ addHandler mh is fn = forEachHP mh is (\ _k v -> fn v)++-- | The @IStructure@s in this module also have the special property that they+-- support a freeze operation which immediately yields a `Foldable` container+-- without any sorting (see `snapFreeze`).+instance OrderedLVarData1 IStructure where+ -- No extra work here... + snapFreeze is = unsafeCoerceLVar <$> G.freeze is++-- | As with all LVars, after freezing, map elements can be consumed. In the case of+-- this @IStructure@ implementation, it need only be `Frzn`, not `Trvrsbl`.+instance F.Foldable (IStructure Frzn) where+ foldr fn zer (IStructure vec) = + F.foldr (\ iv acc ->+ case IV.fromIVar iv of+ Nothing -> acc+ Just x -> fn x acc)+ zer vec++-- | Of course, the stronger `Trvrsbl` state is still fine for folding.+instance F.Foldable (IStructure Trvrsbl) where+ foldr fn zer mp = F.foldr fn zer (castFrzn mp)++-- | @IStructure@ values can be returned as the result of a `runParThenFreeze`.+-- Hence they need a `DeepFrz` instace.+-- @DeepFrz@ is just a type-coercion. No bits flipped at runtime.+instance DeepFrz a => DeepFrz (IStructure s a) where+ type FrzType (IStructure s a) = IStructure Frzn (FrzType a)+ frz = unsafeCoerceLVar++instance (Show a) => Show (IStructure Frzn a) where+ show (IStructure vec) =+ -- individual elements are showable, and show returns a string, so+ -- we want to concatenate those.+ "{IStructure: " Prelude.+++ (Prelude.concat $ intersperse ", " $ Prelude.map show $ V.toList vec) Prelude.+++ "}"++-- | For convenience only; the user could define this.+instance Show a => Show (IStructure Trvrsbl a) where+ show = show . castFrzn++------------------------------------------------------------------------------++-- | Retrieve the number of slots in the I-Structure.+getLength :: IStructure s a -> Par d s Int+getLength (IStructure vec) = return $! V.length vec++-- | Physical identity, just as with IORefs.+-- instance Eq (IStructure s v) where+-- IStructure lv1 == IStructure lv2 = state lv1 == state lv2 ++-- | Create a new, empty, monotonically growing 'IStructure' of a given size.+-- All entries start off as zero, which must be BOTTOM.+newIStructure :: Int -> Par d s (IStructure s elt)+newIStructure len = fmap IStructure $+ V.generateM len (\_ -> IV.new)++-- | This registers handlers on each internal IVar as it is created.+-- It should be more efficient than `newIStructure` followed by `forEachHP`+newIStructureWithCallback :: Int -> (Int -> elt -> Par d s ()) -> Par d s (IStructure s elt)+newIStructureWithCallback len fn =+ fmap IStructure $+ V.generateM len $ \ix -> do + iv <- IV.new+ IV.whenFull Nothing iv (fn ix)+ return iv++-- | /O(N)/ complexity, unfortunately. This implementation of I-Structures requires+-- freezing each of the individual IVars stored in the array.+-- +freezeIStructure :: IStructure s a -> LV.Par QuasiDet s (V.Vector (Maybe a))+freezeIStructure (IStructure vec) = do+ v <- V.mapM IV.freezeIVar vec+ return v++{-# INLINE forEachHP #-}+-- | Add an (asynchronous) callback that listens for all new elements added to+-- the IStructure, optionally enrolled in a handler pool+forEachHP :: -- (Eq a) =>+ Maybe HandlerPool -- ^ pool to enroll in, if any+ -> IStructure s a -- ^ IStructure to listen to+ -> (Int -> a -> Par d s ()) -- ^ callback+ -> Par d s ()+forEachHP hp (IStructure vec) callb =+ -- F.traverse_ (\iv -> IV.addHandler hp iv callb) vec+ for_ (0, V.length vec) $ \ ix ->+ IV.whenFull hp (V.unsafeIndex vec ix) (callb ix)++{-++{-# INLINE forVec #-}+-- | Simple for-each loops over vector elements.+forVec :: Storable a =>+ M.IOVector a -> (Int -> a -> Par d s ()) -> Par d s ()+forVec vec fn = loop 0 + where+ len = M.length vec+ loop i | i == len = return ()+ | otherwise = do elm <- LI.liftIO$ M.unsafeRead vec i+ fn i elm+ loop (i+1)++{-# INLINE forEach #-}+-- | Add an (asynchronous) callback that listens for all new elements added to+-- the set+forEach :: (Num a, Storable a, Eq a) =>+ NatArray s a -> (Int -> a -> Par d s ()) -> Par d s ()+forEach = forEachHP Nothing+-}+++++{-# INLINE put #-}++-- | Put a single element in the array. That slot must be previously empty. (WHNF)+-- Strict in the element being put in the set.+put_ :: Eq elt => IStructure s elt -> Int -> elt -> Par d s ()+put_ (IStructure vec) !ix !elm = IV.put_ (vec ! ix) elm++-- | Put a single element in the array. This variant is deeply strict (`NFData`).+put :: (NFData elt, Eq elt) => IStructure s elt -> Int -> elt -> Par d s ()+put (IStructure vec) !ix !elm = IV.put (vec ! ix) elm++{-# INLINE get #-}+-- | Wait for the indexed entry to contain a value and return that value.+get :: Eq elt => IStructure s elt -> Int -> Par d s elt+get (IStructure vec) !ix = IV.get (vec ! ix)
+ Data/LVar/IVar.hs view
@@ -0,0 +1,200 @@+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE BangPatterns, MultiParamTypeClasses, TypeFamilies, TypeOperators #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE MagicHash #-} ++{-|++ `IVar`s are the very simplest form of `LVar`s. They are either empty, or full, and+ contain only at most a single value.++ For more explanation of using IVars in Haskell, see the @monad-par@ and+ @meta-par@ packages and papers:++ * <http://hackage.haskell.org/package/monad-par>++ * <http://www.cs.indiana.edu/~rrnewton/papers/haskell2011_monad-par.pdf>++ * <http://www.cs.indiana.edu/~rrnewton/papers/2012-ICFP_meta-par.pdf>++ -}++module Data.LVar.IVar+ (+ IVar(..),+ -- * Basic IVar operations, same as in monad-par+ new, get, put, put_,+ + -- * Derived IVar operations, same as in monad-par+ spawn, spawn_, spawnP,++ -- * LVar style operations+ freezeIVar, fromIVar, whenFull)+ where++import Data.IORef+import Control.DeepSeq+import System.Mem.StableName (makeStableName, hashStableName)+import System.IO.Unsafe (unsafePerformIO, unsafeDupablePerformIO)+import qualified Data.Foldable as F+import Control.Exception (throw)+import Control.LVish as LV +import Control.LVish.DeepFrz.Internal+import Control.LVish.Internal as I+import Control.LVish.SchedIdempotent (newLV, putLV, getLV, freezeLV)+import qualified Control.LVish.SchedIdempotent as LI +import Data.LVar.Generic+import Data.LVar.Generic.Internal (unsafeCoerceLVar)+import GHC.Prim (unsafeCoerce#)++#ifdef USE_ABSTRACT_PAR+import qualified Control.Monad.Par.Class as PC+#endif++------------------------------------------------------------------------------+-- IVars implemented on top of (the idempotent implementation of) LVars+------------------------------------------------------------------------------+ +-- | An `IVar` is the simplest type of `LVar`.+newtype IVar s a = IVar (LVar s (IORef (Maybe a)) a)+-- the global data for an IVar a is a reference to Maybe a, while deltas are+-- simply values of type a (taking the IVar from Nothing to Just):++-- | Physical equality just as with IORefs.+instance Eq (IVar s a) where+ (==) (IVar lv1) (IVar lv2) = state lv1 == state lv2++-- | An @IVar@ can be treated as a generic container LVar which happens to+-- contain at most one value! Note, however, that the polymorphic operations are+-- less useful than the monomorphic ones exposed by this module.+instance LVarData1 IVar where + freeze :: IVar s a -> Par QuasiDet s (IVar Frzn a)+ freeze orig@(IVar (WrapLVar lv)) = WrapPar $ do+ freezeLV lv+ return (unsafeCoerceLVar orig)+ addHandler = whenFull++-- | DeepFrz is just a type-coercion. No bits flipped at runtime:+instance DeepFrz a => DeepFrz (IVar s a) where+ type FrzType (IVar s a) = IVar Frzn (FrzType a)+ frz = unsafeCoerceLVar++-- | As with all other `Trvrsbl` LVars, the elements are traversable in a fixed+-- order.+instance F.Foldable (IVar Trvrsbl) where+ foldr fn zer (IVar lv) =+ case unsafeDupablePerformIO$ readIORef (state lv) of+ Just x -> fn x zer+ Nothing -> zer++instance (Show a) => Show (IVar Frzn a) where+ show (IVar lv) =+ show $ unsafeDupablePerformIO $ readIORef (state lv)++-- | For convenience only; the user could define this.+instance Show a => Show (IVar Trvrsbl a) where+ show = show . castFrzn ++--------------------------------------++{-# INLINE new #-}+-- | A new IVar that starts out empty. +new :: Par d s (IVar s a)+new = WrapPar$ fmap (IVar . WrapLVar) $+ newLV $ newIORef Nothing++{-# INLINE get #-}+-- | read the value in a @IVar@. The 'get' can only return when the+-- value has been written by a prior or concurrent @put@ to the same+-- @IVar@.+get :: IVar s a -> Par d s a+get (IVar (WrapLVar iv)) = WrapPar$ getLV iv globalThresh deltaThresh+ where globalThresh ref _ = readIORef ref -- past threshold iff Jusbt _+ deltaThresh x = return $ Just x -- always past threshold++{-# INLINE put_ #-}+-- | put a value into a @IVar@. Multiple 'put's to the same @IVar@+-- are not allowed, and result in a runtime error. (Unless both values put happen to be @(==)@.)+-- +-- This function is always at least strict up to WHNF in the element put.+put_ :: Eq a => IVar s a -> a -> Par d s ()+put_ (IVar (WrapLVar iv)) !x = WrapPar $ putLV iv putter+ where putter ref = atomicModifyIORef ref update+ update (Just y) | x == y = (Just y, Nothing)+ | otherwise = unsafePerformIO $+ do n1 <- fmap hashStableName $ makeStableName x+ n2 <- fmap hashStableName $ makeStableName y+ throw (LV.ConflictingPutExn$ "Multiple puts to an IVar! (obj "++show n2++" was "++show n1++")")+ update Nothing = (Just x, Just x)++-- | The specialized freeze just for IVars. It leaves the result in a natural format (`Maybe`).+freezeIVar :: IVar s a -> LV.Par QuasiDet s (Maybe a)+freezeIVar (IVar (WrapLVar lv)) = WrapPar $ + do freezeLV lv+ getLV lv globalThresh deltaThresh+ where+ globalThresh _ False = return Nothing+ globalThresh ref True = fmap Just $ readIORef ref+ deltaThresh _ = return Nothing+ +-- | Unpack a frozen `IVar` (as produced by a generic `freeze` operation) as a more+-- palatable data structure.+fromIVar :: IVar Frzn a -> Maybe a+fromIVar (IVar lv) = unsafeDupablePerformIO $ readIORef (state lv)++{-# INLINE whenFull #-}+-- | Register a handler that fires when the `IVar` is filled, which, of course, only+-- happens once.+whenFull :: Maybe HandlerPool -> IVar s elt -> (elt -> Par d s ()) -> Par d s ()+whenFull mh (IVar (WrapLVar lv)) fn = + WrapPar (LI.addHandler mh lv globalCB fn')+ where+ fn' x = return (Just (unWrapPar (fn x)))+ globalCB ref = do+ mx <- readIORef ref -- Snapshot+ case mx of+ Nothing -> return Nothing+ Just v -> fn' v+ +--------------------------------------------------------------------------------++{-# INLINE spawn #-}+-- | A simple future represented as an IVar. The result is fully evaluated before+-- the child computation returns.+spawn :: (Eq a, NFData a) => Par d s a -> Par d s (IVar s a)+spawn p = do r <- new; fork (p >>= put r); return r++{-# INLINE spawn_ #-}+-- | A version of `spawn` that uses only weak-head-normal form rather than full `NFData`.+spawn_ :: Eq a => Par d s a -> Par d s (IVar s a)+spawn_ p = do r <- new; fork (p >>= put_ r); return r++{-# INLINE spawnP #-}+-- | A variant that +spawnP :: (Eq a, NFData a) => a -> Par d s (IVar s a)+spawnP a = spawn (return a)++{-# INLINE put #-}+-- | Fill an `IVar`.+put :: (Eq a, NFData a) => IVar s a -> a -> Par d s ()+put v a = deepseq a (put_ v a)+++#ifdef USE_ABSTRACT_PAR+ -- MIN_VERSION_abstract_par(0,4,0)+#warning "Using the latest version of abstract par to activate ParFuture/IVar instances."+instance PC.ParFuture (IVar s) (Par d s) where+ spawn_ = spawn_+ get = get++instance PC.ParIVar (IVar s) (Par d s) where+ fork = fork + put_ = put_+ new = new+#endif+
+ Data/LVar/Internal/Pure.hs view
@@ -0,0 +1,126 @@+{-# LANGUAGE Unsafe #-}++{-# LANGUAGE DataKinds, BangPatterns #-}+{-# LANGUAGE TypeSynonymInstances, FlexibleInstances, MultiParamTypeClasses, TypeFamilies #-}+{-# LANGUAGE InstanceSigs, MagicHash #-}++{-|++This is NOT a datatype for the end-user.++Rather, this module is for building /new/ LVar types in a comparatively easy way: by+putting a pure value in a mutable container, and defining a LUB operation as a pure+function.++The proof-obligation for the library-writer who uses this module is that they must+guarantee that their LUB is a /true least-upper-bound/, obeying the appropriate laws+for a join-semilattice:++ * <http://en.wikipedia.org/wiki/Semilattice>++-}++module Data.LVar.Internal.Pure+ ( PureLVar(..),+ newPureLVar, putPureLVar, waitPureLVar, freezePureLVar+ ) where++import Control.LVish+import Control.LVish.DeepFrz.Internal+import Control.LVish.Internal+import Data.IORef+import qualified Control.LVish.SchedIdempotent as LI +import Algebra.Lattice+import GHC.Prim (unsafeCoerce#)++--------------------------------------------------------------------------------++-- | An LVar which consists merely of an immutable, pure value inside a mutable box.+newtype PureLVar s t = PureLVar (LVar s (IORef t) t)++-- data PureLVar s t = BoundedJoinSemiLattice t => PureLVar (LVar s (IORef t) t)++{-# INLINE newPureLVar #-}+{-# INLINE putPureLVar #-}+{-# INLINE waitPureLVar #-}+{-# INLINE freezePureLVar #-}++-- | A new pure LVar populated with the provided initial state.+newPureLVar :: BoundedJoinSemiLattice t =>+ t -> Par d s (PureLVar s t)+newPureLVar st = WrapPar$ fmap (PureLVar . WrapLVar) $+ LI.newLV $ newIORef st++-- | Wait until the Pure LVar has crossed a threshold and then unblock. (In the+-- semantics, this is a singleton query set.)+waitPureLVar :: (JoinSemiLattice t, Eq t) =>+ PureLVar s t -> t -> Par d s ()+waitPureLVar (PureLVar (WrapLVar iv)) thrsh =+ WrapPar$ LI.getLV iv globalThresh deltaThresh+ where globalThresh ref _ = do x <- readIORef ref+ deltaThresh x+ deltaThresh x | thrsh `joinLeq` x = return $ Just ()+ | otherwise = return Nothing ++-- | Put a new value which will be joined with the old.+putPureLVar :: JoinSemiLattice t =>+ PureLVar s t -> t -> Par d s ()+putPureLVar (PureLVar (WrapLVar iv)) !new =+ WrapPar $ LI.putLV iv putter+ where+ -- Careful, this must be idempotent...+ putter _ = return (Just new)++-- | Freeze the pure LVar, returning its exact value.+-- Subsequent puts will cause an error.+freezePureLVar :: PureLVar s t -> Par QuasiDet s t+freezePureLVar (PureLVar (WrapLVar lv)) = WrapPar$ + do LI.freezeLV lv+ LI.getLV lv globalThresh deltaThresh+ where+ globalThresh ref True = fmap Just $ readIORef ref+ globalThresh _ False = return Nothing+ deltaThresh _ = return Nothing++------------------------------------------------------------+++-- | Physical identity, just as with IORefs.+instance Eq (PureLVar s v) where+ PureLVar lv1 == PureLVar lv2 = state lv1 == state lv2 ++-- | A `PureLVar` can be treated as a generic container LVar which happens to+-- contain exactly one value!+ +-- instance LVarData1 PureLVar where+-- freeze orig@(PureLVar (WrapLVar lv)) = WrapPar$ do freezeLV lv; return (unsafeCoerceLVar orig)+-- sortFreeze is = AFoldable <$> freezeSet is+-- addHandler = forEachHP++-- -- | The `PureLVar`s in this module also have the special property that they support an+-- -- `O(1)` freeze operation which immediately yields a `Foldable` container+-- -- (`snapFreeze`).+-- instance OrderedLVarData1 PureLVar where+-- snapFreeze is = unsafeCoerceLVar <$> freeze is++-- -- | As with all LVars, after freezing, map elements can be consumed. In the case of+-- -- this `PureLVar` implementation, it need only be `Frzn`, not `Trvrsbl`.+-- instance F.Foldable (PureLVar Frzn) where+-- foldr fn zer (PureLVar lv) =+-- -- It's not changing at this point, no problem if duped:+-- let set = unsafeDupablePerformIO (readIORef (state lv)) in+-- F.foldr fn zer set ++-- -- | Of course, the stronger `Trvrsbl` state is still fine for folding.+-- instance F.Foldable (PureLVar Trvrsbl) where+-- foldr fn zer mp = F.foldr fn zer (castFrzn mp)++-- | `PureLVar` values can be returned as the result of a `runParThenFreeze`.+-- Hence they need a `DeepFrz` instace.+-- @DeepFrz@ is just a type-coercion. No bits flipped at runtime.+instance DeepFrz a => DeepFrz (PureLVar s a) where+ -- We can't be sure that someone won't put an LVar value inside a+ -- PureLVar! Therefore we have to apply FrzType recursively.+ type FrzType (PureLVar s a) = PureLVar Frzn (FrzType a)+ frz = unsafeCoerce#+
+ Data/LVar/MaxCounter.hs view
@@ -0,0 +1,79 @@+{-# LANGUAGE Trustworthy #-}++{-# LANGUAGE DataKinds, BangPatterns, MagicHash #-}+{-# LANGUAGE TypeSynonymInstances, FlexibleInstances, MultiParamTypeClasses, TypeFamilies #-}++-- | A counter that contains the maximum value of all puts.++-- TODO: Add 'Min', 'Or', 'And' and other idempotent ops...++module Data.LVar.MaxCounter+ ( MaxCounter,+ newMaxCounter, put, waitThresh, freezeMaxCounter+ ) where++import Control.LVish hiding (freeze)+import Control.LVish.Internal (state)+import Control.LVish.DeepFrz.Internal+import Data.IORef+import Data.LVar.Generic+import Data.LVar.Internal.Pure as P+import Algebra.Lattice+import System.IO.Unsafe (unsafeDupablePerformIO)+import GHC.Prim (unsafeCoerce#)++--------------------------------------------------------------------------------++-- | A @MaxCounter@ is really a constant-space ongoing @fold max@ operation.+-- +-- A @MaxCounter@ is an example of a `PureLVar`. It is implemented simply as a+-- pure value in a mutable box.+type MaxCounter s = PureLVar s MC++newtype MC = MC Int+ deriving (Eq, Show, Ord, Read)++instance JoinSemiLattice MC where + join (MC a) (MC b) = MC (a `max` b)++instance BoundedJoinSemiLattice MC where+ bottom = MC minBound++-- | Create a new counter with the given initial value.+newMaxCounter :: Int -> Par d s (MaxCounter s)+newMaxCounter n = newPureLVar (MC n)++-- | Incorporate a new value in the max-fold. If the previous maximum is less than+-- the new value, increase it.+put :: MaxCounter s -> Int -> Par d s ()+put lv n = putPureLVar lv (MC n)++-- | Wait until the maximum observed value reaches some threshold, then return.+waitThresh :: MaxCounter s -> Int -> Par d s ()+waitThresh lv n = waitPureLVar lv (MC n)++-- | Observe what the final value of the counter was.+freezeMaxCounter :: MaxCounter s -> Par QuasiDet s Int+freezeMaxCounter lv = do+ MC n <- freezePureLVar lv+ return n++-- | Once frozen, for example by `runParThenFreeze`, a MaxCounter can be converted+-- directly into an Int.+fromMaxCounter :: MaxCounter Frzn -> Int+fromMaxCounter (PureLVar lv) =+ case unsafeDupablePerformIO (readIORef (state lv)) of+ MC n -> n++instance DeepFrz MC where+ type FrzType MC = MC++-- Don't need this because there is an instance for `PureLVar`:+{-+-- | @MaxCounter@ values can be returned in the results of a+-- `runParThenFreeze`. Hence they need a `DeepFrz` instance.+-- @DeepFrz@ is just a type-coercion. No bits flipped at runtime.+instance DeepFrz (MaxCounter s) where+ type FrzType (MaxCounter s) = (MaxCounter Frzn)+ frz = unsafeCoerce#+-}
+ Data/LVar/NatArray.hs view
@@ -0,0 +1,284 @@+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE GADTs #-}++{-|++An I-structure (array) of /positive/ numbers. A `NatArray` cannot store zeros.++This particular implementation makes a trade-off between expressiveness (monomorphic+in array contents) and efficiency. The efficiency gained of course is that the array+may be unboxed, and we don't need extra bits to store empty/full status.++/However/, relative to "Data.LVar.IStructure", there is a performance disadvantage as+well. As of [2013.09.28] and their initial release, `NatArray`s are implemented as a+/single/ `LVar`, which means they share a single wait-list of blocked computations.+If there are many computations blocking on different elements within a `NatArray`,+scalability will be much worse than with other `IStructure` implementations.++The holy grail is to get unboxed arrays and scalable blocking, but we don't have this+yet.++Finally, note that this data-structure has an EXPERIMENTAL status and may be removed+in future releases as we find better ways to support unboxed array structures with+per-element synchronization.++-}++module Data.LVar.NatArray+ (+ -- * Basic operations+ NatArray,+ newNatArray, put, get,++ -- * Iteration and callbacks+ forEach, forEachHP++ -- -- * Quasi-deterministic operations+ -- freezeSetAfter, withCallbacksThenFreeze, freezeSet,++ -- -- * Higher-level derived operations+ -- copy, traverseSet, traverseSet_, union, intersection,+ -- cartesianProd, cartesianProds, ++ -- -- * Alternate versions of derived ops that expose HandlerPools they create.+ -- forEachHP, traverseSetHP, traverseSetHP_,+ -- cartesianProdHP, cartesianProdsHP+ ) where++-- import qualified Data.Vector.Unboxed as U+-- import qualified Data.Vector.Unboxed.Mutable as M++import qualified Data.Vector.Storable as U+import qualified Data.Vector.Storable.Mutable as M+import Foreign.Marshal.MissingAlloc (callocBytes)+import Foreign.Marshal.Alloc (finalizerFree)+import Foreign.Storable (sizeOf, Storable)+import Foreign.ForeignPtr (newForeignPtr, withForeignPtr)+import qualified Foreign.Ptr as P+import qualified Data.Bits.Atomic as B+import Data.Bits ((.&.))++import Control.Monad (void)+import Control.Exception (throw)+import Data.IORef+import Data.Maybe (fromMaybe)+import qualified Data.Set as S+import qualified Data.LVar.IVar as IV+import qualified Data.Foldable as F+import qualified Data.Traversable as T+import Data.LVar.Generic++import Control.LVish as LV hiding (addHandler)+import Control.LVish.DeepFrz.Internal as DF+import Control.LVish.Internal as LI+import Control.LVish.SchedIdempotent (newLV, putLV, getLV, freezeLV,+ freezeLVAfter, liftIO)+import qualified Control.LVish.SchedIdempotent as L+import System.IO.Unsafe (unsafeDupablePerformIO)++------------------------------------------------------------------------------+-- Toggles++#define USE_CALLOC+-- A low-level optimization below.++------------------------------------------------------------------------------++-- | An array of bit-fields with a monotonic OR operation. This can be used to model+-- a set of Ints by setting the vector entries to zero or one, but it can also+-- model other finite lattices for each index.+-- newtype NatArray s a = NatArray (LVar s (M.IOVector a) (Int,a))+data NatArray s a = Storable a => NatArray !(LVar s (M.IOVector a) (Int,a))++unNatArray (NatArray lv) = lv++-- | Physical identity, just as with IORefs.+-- instance Eq (NatArray s v) where+-- NatArray lv1 == NatArray lv2 = state lv1 == state lv2 ++-- | Create a new, empty, monotonically growing 'NatArray' of a given size.+-- All entries start off as zero, which must be BOTTOM.+newNatArray :: forall elt d s . (Storable elt, Num elt) =>+ Int -> Par d s (NatArray s elt)+newNatArray len = WrapPar $ fmap (NatArray . WrapLVar) $ newLV $ do+#ifdef USE_CALLOC+ let bytes = sizeOf (undefined::elt) * len+ mem <- callocBytes bytes+ fp <- newForeignPtr finalizerFree mem+ return $! M.unsafeFromForeignPtr0 fp len+#else+ M.replicate len 0+#endif++-- | /O(1)/ Freeze operation that directly returns a nice, usable, representation of+-- the array data.+freezeNatArray :: Storable a => NatArray s a -> LV.Par QuasiDet s (U.Vector a)+freezeNatArray (NatArray lv) =+ error "FINISHME"+ -- LI.liftIO $ U.unsafeFreeze (LI.state lv)++--------------------------------------------------------------------------------+-- Instances:++-- FIXME: there is a tension here.. should NatArray really be a generic LVarData1 at all?+-- Can it really store anything in Storable!?!? Or do we need to fix it to numbers+-- to ensure the zero-trick makes sense?++{-++instance DeepFrz a => DeepFrz (NatArray s a) where+ type FrzType (NatArray s a) = NatArray Frzn (FrzType a)+ frz = unsafeCoerceLVar++-- | /O(1)/: Convert from a frozen `NatArray` to a plain vector.+-- This is only permitted when the `NatArray` has already been frozen.+-- This is useful for processing the result of `Control.LVish.DeepFrz.runParThenFreeze`.+fromNatArray :: NatArray Frzn a -> U.Vector a+fromNatArray (NatArray lv) = unsafeDupablePerformIO (readIORef (state lv))++-}++--------------------------------------------------------------------------------++{-# INLINE forEachHP #-}+-- | Add an (asynchronous) callback that listens for all new elements added to+-- the array, optionally enrolled in a handler pool.+forEachHP :: (Storable a, Eq a, Num a) =>+ Maybe HandlerPool -- ^ pool to enroll in, if any+ -> NatArray s a -- ^ array to listen to+ -> (Int -> a -> Par d s ()) -- ^ callback+ -> Par d s ()+forEachHP hp (NatArray (WrapLVar lv)) callb = WrapPar $ do+ L.addHandler hp lv globalCB deltaCB+ return ()+ where+ deltaCB (ix,x) = return$ Just$ unWrapPar$ callb ix x+ globalCB vec = return$ Just$ unWrapPar$+ -- FIXME / TODO: need a better (parallel) for loop:+ forVec vec $ \ ix elm ->+ -- FIXME: When it starts off, it is SPARSE... there must be a good way to+ -- avoid testing each position for zero.+ if elm == 0+ then return () + else forkHP hp $ callb ix elm++{-# INLINE forVec #-}+-- | Simple for-each loops over vector elements.+forVec :: Storable a =>+ M.IOVector a -> (Int -> a -> Par d s ()) -> Par d s ()+forVec vec fn = loop 0 + where+ len = M.length vec+ loop i | i == len = return ()+ | otherwise = do elm <- LI.liftIO$ M.unsafeRead vec i+ fn i elm+ loop (i+1)++{-# INLINE forEach #-}+-- | Add an (asynchronous) callback that listens for all new elements added to+-- the set+forEach :: (Num a, Storable a, Eq a) =>+ NatArray s a -> (Int -> a -> Par d s ()) -> Par d s ()+forEach = forEachHP Nothing+++{-# INLINE put #-}+-- | Put a single element in the array. That slot must be previously empty. (WHNF)+-- Strict in the element being put in the set.+put :: forall s d elt . (Storable elt, B.AtomicBits elt, Num elt, Show elt) =>+ NatArray s elt -> Int -> elt -> Par d s ()+put _ !ix 0 = throw (LVarSpecificExn$ "NatArray: violation! Attempt to put zero to index: "++show ix)+put (NatArray (WrapLVar lv)) !ix !elm = WrapPar$ putLV lv (putter ix)+ where putter ix vec@(M.MVector offset fptr) =+ withForeignPtr fptr $ \ ptr -> do + let offset = sizeOf (undefined::elt) * ix+ -- ARG, if it weren't for the idempotency requirement we could use fetchAndAdd here:+ -- orig <- B.fetchAndAdd (P.plusPtr ptr offset) elm + orig <- B.compareAndSwap (P.plusPtr ptr offset) 0 elm+ case orig of+ 0 -> return (Just (ix, elm))+ i | i == elm -> return Nothing -- Allow repeated, equal puts.+ | otherwise -> throw$ ConflictingPutExn$ "Multiple puts to index of a NatArray: "+++ show ix++" new/old : "++show elm++"/"++show orig++{-# INLINE get #-}+-- | Wait for an indexed entry to contain a non-zero value.+-- +-- Warning: this is inefficient if it needs to block, because the deltaThresh must+-- monitor EVERY new addition.+get :: forall s d elt . (Storable elt, B.AtomicBits elt, Num elt) =>+ NatArray s elt -> Int -> Par d s elt+get (NatArray (WrapLVar lv)) !ix = WrapPar $+ getLV lv globalThresh deltaThresh+ where+ globalThresh ref _frzn = do + elm <- M.read ref ix + if elm == 0+ then return Nothing+ else return (Just elm)+ -- FIXME: we don't actually want to call the deltaThresh on every element...+ -- We want more locality than that...+ deltaThresh (ix2,e2) | ix == ix2 = return$! Just e2+ | otherwise = return Nothing +++-- | A sequential for-loop with a catch. The body of the loop gets access to a+-- special get function. This getter will not block subsequent iterations of the+-- loop. Parallelism will be introduced minimally, only as neccessary to avoid+-- blocking.+seqLoopNonblocking :: Int -> Int ->+ ((NatArray s elt -> Int -> Par d s elt) -> Int -> Par d s ()) ->+ Par d s ()+seqLoopNonblocking start end fn = do+ error "TODO - FINISHME: seqLoopNonblocking optimization"+ where+ par =+ L.Par $ \k -> L.ClosedPar $ \q -> do+ -- tripped <- globalThresh state False+-- case tripped of+ -- Just b -> exec (k b) q -- already past the threshold; invoke the+-- forkHP mh child = mkPar $ \k q -> do+-- closed <- closeInPool mh child+-- Sched.pushWork q (k ()) -- "Work-first" policy.+-- -- hpMsg " [dbg-lvish] incremented and pushed work in forkInPool, now running cont" hp +-- exec closed q + undefined++{-+parFor :: (ParFuture iv p) => InclusiveRange -> (Int -> p ()) -> p ()+parFor (InclusiveRange start end) body =+ do+ let run (x,y) = for_ x (y+1) body+ range_segments = splitInclusiveRange (4*numCapabilities) (start,end)++ vars <- M.forM range_segments (\ pr -> spawn_ (run pr))+ M.mapM_ get vars+ return ()++splitInclusiveRange :: Int -> (Int, Int) -> [(Int, Int)]+splitInclusiveRange pieces (start,end) =+ map largepiece [0..remain-1] +++ map smallpiece [remain..pieces-1]+ where+ len = end - start + 1 -- inclusive [start,end]+ (portion, remain) = len `quotRem` pieces+ largepiece i =+ let offset = start + (i * (portion + 1))+ in (offset, offset + portion)+ smallpiece i =+ let offset = start + (i * portion) + remain+ in (offset, offset + portion - 1)++data InclusiveRange = InclusiveRange Int Int+-}
+ Data/LVar/Pair.hs view
@@ -0,0 +1,56 @@+{-# LANGUAGE BangPatterns #-}++-- | Just for demonstration purposes. It's probably simpler to use a pair of IVars.++module Data.LVar.Pair (+ IPair, newPair, putFst, putSnd, getFst, getSnd+ ) where++import Data.IORef+import Control.Exception (throw)+import Control.LVish+import Control.LVish.Internal+import Control.LVish.SchedIdempotent (newLV, putLV, getLV)+import qualified Control.LVish.SchedIdempotent as L+import Data.LVar.Generic++------------------------------------------------------------------------------+-- IPairs implemented on top of (the idempotent implementation of) LVars:+------------------------------------------------------------------------------+ +type IPair s a b = LVar s (IORef (Maybe a), IORef (Maybe b)) (Either a b)++-- This can't be an intstance of LVarData1... we need LVarData2.++newPair :: Par d s (IPair s a b)+newPair = WrapPar $ fmap WrapLVar $ newLV $ do+ r1 <- newIORef Nothing+ r2 <- newIORef Nothing+ return (r1, r2)+ +putFst :: IPair s a b -> a -> Par d s ()+putFst (WrapLVar lv) !elt = WrapPar $ putLV lv putter+ where putter (r1, _) = atomicModifyIORef r1 update+ update (Just _) = throw$ ConflictingPutExn$ "Multiple puts to first element of an IPair!"+ update Nothing = (Just elt, Just $ Left elt)+ +putSnd :: IPair s a b -> b -> Par d s ()+putSnd (WrapLVar lv) !elt = WrapPar $ putLV lv putter+ where putter (_, r2) = atomicModifyIORef r2 update+ update (Just _) = throw$ ConflictingPutExn$ "Multiple puts to second element of an IPair!"+ update Nothing = (Just elt, Just $ Right elt) + +getFst :: IPair s a b -> Par d s a +getFst (WrapLVar lv) = WrapPar $ getLV lv globalThresh deltaThresh+ where globalThresh (r1, _) _ = readIORef r1+ deltaThresh (Left x) = return $ Just x+ deltaThresh (Right _) = return Nothing+ +getSnd :: IPair s a b -> Par d s b +getSnd (WrapLVar lv) = WrapPar $ getLV lv globalThresh deltaThresh+ where globalThresh (_, r2) _ = readIORef r2+ deltaThresh (Left _) = return Nothing + deltaThresh (Right x) = return $ Just x++-- TODO: LVarData2 instance??+
+ Data/LVar/PureMap.hs view
@@ -0,0 +1,370 @@+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}++{-|++ This module provides finite maps that only grow. It is based on the popular `Data.Map`+ balanced-tree representation of maps. Thus scalability is /not/ good for this+ implementation. However, there are some interoperability benefits. For example,+ after running a parallel computation with a map result, this module can produce a+ `Data.Map` in /O(1)/ without copying, which may be useful downstream.++ -}++module Data.LVar.PureMap+ (+ -- * Basic operations+ IMap, + newEmptyMap, newMap, newFromList,+ insert, + getKey, waitValue, waitSize, modify, ++ -- * Freezing results (Quasi-determinism) + freezeMap, fromIMap,+ + -- * Iteration and callbacks+ forEach, forEachHP,+ withCallbacksThenFreeze,++ -- * Higher-level derived operations+ copy, traverseMap, traverseMap_, union,+ + -- * Alternate versions of derived ops that expose HandlerPools they create.+ traverseMapHP, traverseMapHP_, unionHP+ ) where++import Control.Monad (void)+import Control.Exception (throw)+import Control.Applicative (Applicative, (<$>),(*>), pure, getConst, Const(Const))+import Data.Monoid (Monoid(..))+import Data.IORef+import qualified Data.Map.Strict as M+import qualified Data.LVar.IVar as IV+import qualified Data.Foldable as F+import Data.LVar.Generic+import Data.LVar.Generic.Internal (unsafeCoerceLVar)+import Data.UtilInternal (traverseWithKey_)+import Data.List (intersperse)+import Control.LVish.DeepFrz.Internal+import Control.LVish+import Control.LVish.Internal as LI+import Control.LVish.SchedIdempotent (newLV, putLV, putLV_, getLV, freezeLV, freezeLVAfter)+import qualified Control.LVish.SchedIdempotent as L+import System.IO.Unsafe (unsafePerformIO, unsafeDupablePerformIO)+import System.Mem.StableName (makeStableName, hashStableName)++type QPar = Par QuasiDet -- Shorthand.++------------------------------------------------------------------------------+-- IMaps implemented on top of LVars:+------------------------------------------------------------------------------++-- | The map datatype itself. Like all other LVars, it has an @s@ parameter (think+-- `STRef`) in addition to the @a@ parameter that describes the type of elements+-- in the set.+-- +-- Performance note: There is only ONE mutable location in this implementation. Thus+-- it is not a scalable implementation.+newtype IMap k s v = IMap (LVar s (IORef (M.Map k v)) (k,v))++-- | Equality is physical equality, as with @IORef@s.+instance Eq (IMap k s v) where+ IMap lv1 == IMap lv2 = state lv1 == state lv2 ++-- | An `IMap` can be treated as a generic container LVar. However, the polymorphic+-- operations are less useful than the monomorphic ones exposed by this module.+instance LVarData1 (IMap k) where+ freeze orig@(IMap (WrapLVar lv)) = WrapPar$ do freezeLV lv; return (unsafeCoerceLVar orig)+ -- Unlike the Map-specific forEach variants, this takes only values, not keys.+ addHandler mh mp fn = forEachHP mh mp (\ _k v -> fn v)+ sortFrzn (IMap lv) = AFoldable$ unsafeDupablePerformIO (readIORef (state lv))++-- | The `IMap`s in this module also have the special property that they support an+-- `O(1)` freeze operation which immediately yields a `Foldable` container+-- (`snapFreeze`).+instance OrderedLVarData1 (IMap k) where+ snapFreeze is = unsafeCoerceLVar <$> freeze is++-- | As with all LVars, after freezing, map elements can be consumed. In the case of+-- this `IMap` implementation, it need only be `Frzn`, not `Trvrsbl`.+instance F.Foldable (IMap k Frzn) where+ foldr fn zer (IMap lv) =+ let set = unsafeDupablePerformIO (readIORef (state lv)) in+ F.foldr fn zer set ++-- | Of course, the stronger `Trvrsbl` state is still fine for folding.+instance F.Foldable (IMap k Trvrsbl) where+ foldr fn zer mp = F.foldr fn zer (castFrzn mp)++-- | `IMap` values can be returned as the result of a `runParThenFreeze`.+-- Hence they need a `DeepFrz` instace.+-- @DeepFrz@ is just a type-coercion. No bits flipped at runtime.+instance DeepFrz a => DeepFrz (IMap k s a) where+ type FrzType (IMap k s a) = IMap k Frzn (FrzType a)+ frz = unsafeCoerceLVar++instance (Show k, Show a) => Show (IMap k Frzn a) where+ show (IMap lv) =+ let mp' = unsafeDupablePerformIO (readIORef (state lv)) in+ "{IMap: " +++ (concat $ intersperse ", " $ map show $+ M.toList mp') ++ "}"++-- | For convenience only; the user could define this.+instance (Show k, Show a) => Show (IMap k Trvrsbl a) where+ show lv = show (castFrzn lv)++--------------------------------------------------------------------------------++-- | Create a fresh map with nothing in it.+newEmptyMap :: Par d s (IMap k s v)+newEmptyMap = WrapPar$ fmap (IMap . WrapLVar) $ newLV$ newIORef M.empty++-- | Create a new map populated with initial elements.+newMap :: M.Map k v -> Par d s (IMap k s v)+newMap m = WrapPar$ fmap (IMap . WrapLVar) $ newLV$ newIORef m++-- | A convenience function that is equivalent to calling `Data.Map.fromList`+-- followed by `newMap`.+newFromList :: (Ord k, Eq v) =>+ [(k,v)] -> Par d s (IMap k s v)+newFromList = newMap . M.fromList++-- | Register a per-element callback, then run an action in this context, and freeze+-- when all (recursive) invocations of the callback are complete. Returns the final+-- valueof the Map variable.+withCallbacksThenFreeze :: forall k v b s . Eq b =>+ IMap k s v -> (k -> v -> QPar s ()) -> QPar s b -> QPar s b+withCallbacksThenFreeze (IMap (WrapLVar lv)) callback action =+ do hp <- newPool + res <- IV.new + WrapPar$ freezeLVAfter lv (initCB hp res) deltaCB+ -- We additionally have to quiesce here because we fork the inital set of+ -- callbacks on their own threads:+ quiesce hp+ IV.get res+ where+ deltaCB (k,v) = return$ Just$ unWrapPar $ callback k v+ initCB :: HandlerPool -> IV.IVar s b -> (IORef (M.Map k v)) -> IO (Maybe (L.Par ()))+ initCB hp resIV ref = do+ -- The implementation guarantees that all elements will be caught either here,+ -- or by the delta-callback:+ mp <- readIORef ref -- Snapshot+ return $ Just $ unWrapPar $ do + traverseWithKey_ (\ k v -> forkHP (Just hp)$ callback k v) mp+ res <- action -- Any additional puts here trigger the callback.+ IV.put_ resIV res++-- | Add an (asynchronous) callback that listens for all new key/value pairs added to+-- the map, optionally enrolled in a handler pool+forEachHP :: Maybe HandlerPool -- ^ optional pool to enroll in + -> IMap k s v -- ^ Map to listen to+ -> (k -> v -> Par d s ()) -- ^ callback+ -> Par d s ()+forEachHP mh (IMap (WrapLVar lv)) callb = WrapPar $ do+ L.addHandler mh lv globalCB deltaCB+ return ()+ where+ deltaCB (k,v) = return$ Just$ unWrapPar $ callb k v+ globalCB ref = do+ mp <- readIORef ref -- Snapshot+ return $ Just $ unWrapPar $ + traverseWithKey_ (\ k v -> forkHP mh$ callb k v) mp+ +-- | Add an (asynchronous) callback that listens for all new new key/value pairs added to+-- the map+forEach :: IMap k s v -> (k -> v -> Par d s ()) -> Par d s ()+forEach = forEachHP Nothing ++-- | Put a single entry into the map. Strict (WHNF) in the key and value.+-- +-- As with other container LVars, if an key is put multiple times, the values had+-- better be equal @(==)@, or a multiple-put error is raised.+insert :: (Ord k, Eq v) =>+ k -> v -> IMap k s v -> Par d s () +insert !key !elm (IMap (WrapLVar lv)) = WrapPar$ putLV lv putter+ where putter ref = atomicModifyIORef' ref update+ update mp =+ let mp' = M.insertWith fn key elm mp+ fn v1 v2 | v1 == v2 = v1+ | otherwise = throw$ ConflictingPutExn$ "Multiple puts to one entry in an IMap!"+ in+ -- Here we do a constant time check to see if we actually changed anything:+ -- For idempotency it is important that we return Nothing if not.+ if M.size mp' > M.size mp+ then (mp',Just (key,elm))+ else (mp, Nothing)++-- | IMap's containing other LVars have some additional capabilities compared to+-- those containing regular Haskell data. In particular, it is possible to modify+-- existing entries (monotonically). Further, this `modify` function implicitly+-- inserts a "bottom" element if there is no existing entry for the key.+--+-- Unfortunately, that means that this takes another computation for creating new+-- "bottom" elements for the nested LVars stored inside the Map.+modify :: forall f a b d s key . (Ord key, LVarData1 f, Show key, Ord a) =>+ IMap key s (f s a)+ -> key -- ^ The key to lookup.+ -> (Par d s (f s a)) -- ^ Create a new "bottom" element whenever an entry is not present.+ -> (f s a -> Par d s b) -- ^ The computation to apply on the right-hand-side of the keyed entry.+ -> Par d s b+modify (IMap lv) key newBottom fn = WrapPar $ do + let ref = state lv + mp <- L.liftIO$ readIORef ref+ case M.lookup key mp of+ Just lv2 -> do L.logStrLn$ " [Map.modify] key already present: "++show key+++ " adding to inner "++show(unsafeName lv2)+ unWrapPar$ fn lv2+ Nothing -> do + bot <- unWrapPar newBottom :: L.Par (f s a)+ L.logStrLn$ " [Map.modify] allocated new inner "++show(unsafeName bot)+ let putter _ = L.liftIO$ atomicModifyIORef' ref $ \ mp2 ->+ case M.lookup key mp2 of+ Just lv2 -> (mp2, (Nothing, unWrapPar$ fn lv2))+ Nothing -> (M.insert key bot mp2,+ (Just (key, bot), + do L.logStrLn$ " [Map.modify] key absent, adding the new one."+ unWrapPar$ fn bot))+ + act <- putLV_ (unWrapLVar lv) putter+ act++-- | Wait for the map to contain a specified key, and return the associated value.+getKey :: Ord k => k -> IMap k s v -> Par d s v+getKey !key (IMap (WrapLVar lv)) = WrapPar$ getLV lv globalThresh deltaThresh+ where+ globalThresh ref _frzn = do+ mp <- readIORef ref+ return (M.lookup key mp)+ deltaThresh (k,v) | k == key = return$ Just v+ | otherwise = return Nothing ++-- | Wait until the map contains a certain value (on any key).+waitValue :: (Ord k, Eq v) => v -> IMap k s v -> Par d s ()+waitValue !val (IMap (WrapLVar lv)) = WrapPar$ getLV lv globalThresh deltaThresh+ where+ globalThresh ref _frzn = do+ mp <- readIORef ref+ -- This is very inefficient:+ let fn Nothing v | v == val = Just ()+ | otherwise = Nothing+ fn just _ = just+ -- FIXME: no short-circuit for this fold:+ return $! M.foldl fn Nothing mp+ deltaThresh (_,v) | v == val = return$ Just ()+ | otherwise = return Nothing +++-- | Wait on the SIZE of the map, not its contents.+waitSize :: Int -> IMap k s v -> Par d s ()+waitSize !sz (IMap (WrapLVar lv)) = WrapPar $+ getLV lv globalThresh deltaThresh+ where+ globalThresh ref _frzn = do+ mp <- readIORef ref+ case M.size mp >= sz of+ True -> return (Just ())+ False -> return (Nothing)+ -- Here's an example of a situation where we CANNOT TELL if a delta puts it over+ -- the threshold.a+ deltaThresh _ = globalThresh (L.state lv) False++-- | Get the exact contents of the map Using this may cause your+-- program to exhibit a limited form of nondeterminism: it will never+-- return the wrong answer, but it may include synchronization bugs+-- that can (nondeterministically) cause exceptions.+--+-- This Data.Map based LVar has the property that you can+-- retrieve the full set without any IO, and without nondeterminism+-- leaking. (This is because the internal order is fixed for the+-- tree-based Data.Set.) +freezeMap :: IMap k s v -> QPar s (M.Map k v)+freezeMap (IMap (WrapLVar lv)) = WrapPar $+ do freezeLV lv+ getLV lv globalThresh deltaThresh+ where+ globalThresh _ False = return Nothing+ globalThresh ref True = fmap Just $ readIORef ref+ deltaThresh _ = return Nothing++-- | /O(1)/: Convert from an `IMap` to a plain `Data.Map`.+-- This is only permitted when the `IMap` has already been frozen.+-- This is useful for processing the result of `Control.LVish.DeepFrz.runParThenFreeze`. +fromIMap :: IMap k Frzn a -> M.Map k a +fromIMap (IMap lv) = unsafeDupablePerformIO (readIORef (state lv))++--------------------------------------------------------------------------------+-- Higher level routines that could (mostly) be defined using the above interface.+--------------------------------------------------------------------------------++-- | Establish monotonic map between the input and output sets. Produce a new result+-- based on each element, while leaving the keys the same.+traverseMap :: (Ord k, Eq b) =>+ (k -> a -> Par d s b) -> IMap k s a -> Par d s (IMap k s b)+traverseMap f s = traverseMapHP Nothing f s++-- | An imperative-style, inplace version of 'traverseMap' that takes the output set+-- as an argument.+traverseMap_ :: (Ord k, Eq b) =>+ (k -> a -> Par d s b) -> IMap k s a -> IMap k s b -> Par d s ()+traverseMap_ f s o = traverseMapHP_ Nothing f s o++-- | Return a new map which will (ultimately) contain everything in either input+-- map. Conflicting entries will result in a multiple put exception.+-- Optionally ties the handlers to a pool.+union :: (Ord k, Eq a) => IMap k s a -> IMap k s a -> Par d s (IMap k s a)+union = unionHP Nothing++-- TODO: Intersection++--------------------------------------------------------------------------------+-- Alternate versions of functions that EXPOSE the HandlerPools+--------------------------------------------------------------------------------++-- | Return a fresh map which will contain strictly more elements than the input.+-- That is, things put in the former go in the latter, but not vice versa.+copy :: (Ord k, Eq v) => IMap k s v -> Par d s (IMap k s v)+copy = traverseMap (\ _ x -> return x)++-- | Variant that optionally ties the handlers to a pool.+traverseMapHP :: (Ord k, Eq b) =>+ Maybe HandlerPool -> (k -> a -> Par d s b) -> IMap k s a ->+ Par d s (IMap k s b)+traverseMapHP mh fn set = do+ os <- newEmptyMap+ traverseMapHP_ mh fn set os + return os++-- | Variant that optionally ties the handlers to a pool.+traverseMapHP_ :: (Ord k, Eq b) =>+ Maybe HandlerPool -> (k -> a -> Par d s b) -> IMap k s a -> IMap k s b ->+ Par d s ()+traverseMapHP_ mh fn set os = do+ forEachHP mh set $ \ k x -> do + x' <- fn k x+ insert k x' os++-- | Variant that optionally ties the handlers in the resulting set to the same+-- handler pool as those in the two input sets.+unionHP :: (Ord k, Eq a) => Maybe HandlerPool ->+ IMap k s a -> IMap k s a -> Par d s (IMap k s a)+unionHP mh m1 m2 = do+ os <- newEmptyMap+ forEachHP mh m1 (\ k v -> insert k v os)+ forEachHP mh m2 (\ k v -> insert k v os)+ return os++{-# NOINLINE unsafeName #-}+unsafeName :: a -> Int+unsafeName x = unsafePerformIO $ do + sn <- makeStableName x+ return (hashStableName sn)+
+ Data/LVar/PureSet.hs view
@@ -0,0 +1,397 @@+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE MagicHash #-}++{-|++ This module provides sets that only grow. It is based on the popular `Data.Set`+ balanced-tree representation of sets. Thus scalability is not good for this+ implementation. However, there are some interoperability benefits. For exmaple,+ after running a parallel computation with a set result, this module can produce a+ `Data.Set` in /O(1)/ without copying, which may be useful downstream.++ -}++module Data.LVar.PureSet+ (+ -- * Basic operations+ ISet, + newEmptySet, newSet, newFromList,+ insert, waitElem, waitSize, ++ -- * Iteration and callbacks+ forEach, forEachHP, ++ -- * Quasi-deterministic operations+ freezeSetAfter, withCallbacksThenFreeze, freezeSet,+ fromISet,+ + -- * Higher-level derived operations+ copy, traverseSet, traverseSet_, union, intersection,+ cartesianProd, cartesianProds, ++ -- * Alternate versions of derived ops that expose HandlerPools they create.+ traverseSetHP, traverseSetHP_, unionHP, intersectionHP,+ cartesianProdHP, cartesianProdsHP+ ) where++import Control.Monad (void)+import Control.Applicative ((<$>))+import Data.IORef+import Data.List (intersperse)+import qualified Data.Set as S+import qualified Data.LVar.IVar as IV+import qualified Data.Foldable as F+import Data.LVar.Generic+import Data.LVar.Generic.Internal (unsafeCoerceLVar)+import Control.LVish as LV+import Control.LVish.DeepFrz.Internal+import Control.LVish.Internal as LI+import Control.LVish.SchedIdempotent (newLV, putLV, getLV, freezeLV, freezeLVAfter)+import qualified Control.LVish.SchedIdempotent as L+import System.IO.Unsafe (unsafeDupablePerformIO)+import Prelude hiding (insert)++------------------------------------------------------------------------------+-- ISets and setmap implemented on top of LVars:+------------------------------------------------------------------------------++-- | The set datatype itself. Like all other LVars, it has an @s@ parameter (like+-- an `STRef`) in addition to the @a@ parameter that describes the type of elements+-- in the set.+--+-- Performance note: There is only ONE mutable location in this implementation. Thus+-- it is not a scalable implementation.+newtype ISet s a = ISet (LVar s (IORef (S.Set a)) a)++-- | Physical identity, just as with IORefs.+instance Eq (ISet s v) where+ ISet lv1 == ISet lv2 = state lv1 == state lv2 ++-- | An `ISet` can be treated as a generic container LVar. However, the polymorphic+-- operations are less useful than the monomorphic ones exposed by this module.+instance LVarData1 ISet where+ freeze orig@(ISet (WrapLVar lv)) = WrapPar$ do freezeLV lv; return (unsafeCoerceLVar orig)+ addHandler = forEachHP+ -- | We can do better than the default here; this is /O(1)/:+ sortFrzn (ISet lv) = AFoldable$ unsafeDupablePerformIO (readIORef (state lv))++-- | The `ISet`s in this module also have the special property that they support an+-- `O(1)` freeze operation which immediately yields a `Foldable` container+-- (`snapFreeze`).+instance OrderedLVarData1 ISet where+ snapFreeze is = unsafeCoerceLVar <$> freeze is++-- | As with all LVars, after freezing, map elements can be consumed. In the case of+-- this `ISet` implementation, it need only be `Frzn`, not `Trvrsbl`.+instance F.Foldable (ISet Frzn) where+ foldr fn zer (ISet lv) =+ -- It's not changing at this point, no problem if duped:+ let set = unsafeDupablePerformIO (readIORef (state lv)) in+ F.foldr fn zer set ++-- | Of course, the stronger `Trvrsbl` state is still fine for folding.+instance F.Foldable (ISet Trvrsbl) where+ foldr fn zer mp = F.foldr fn zer (castFrzn mp)+++-- | `ISet` values can be returned as the result of a `runParThenFreeze`.+-- Hence they need a `DeepFrz` instace.+-- @DeepFrz@ is just a type-coercion. No bits flipped at runtime.+instance DeepFrz a => DeepFrz (ISet s a) where+ type FrzType (ISet s a) = ISet Frzn (FrzType a)+ frz = unsafeCoerceLVar++instance (Show a) => Show (ISet Frzn a) where+ show (ISet lv) =+ let set = S.toList $ unsafeDupablePerformIO $ readIORef (state lv) in+ "{ISet: " +++ (concat $ intersperse ", " $ map show set) ++ "}"++-- | For convenience; the user could define this.+instance Show a => Show (ISet Trvrsbl a) where+ show = show . castFrzn++-- | Create a new, empty, monotonically growing 'ISet'.+newEmptySet :: Par d s (ISet s a)+newEmptySet = newSet S.empty++-- | Create a new set populated with initial elements.+newSet :: S.Set a -> Par d s (ISet s a)+newSet s = WrapPar$ fmap (ISet . WrapLVar) $ newLV$ newIORef s++-- | Create a new 'ISet' drawing initial elements from an existing list.+newFromList :: Ord a => [a] -> Par d s (ISet s a)+newFromList ls = newSet (S.fromList ls)++-- (Todo: in production you might want even more ... like going from a Vector)++--------------------------------------------------------------------------------+-- Quasi-deterministic ops:+--------------------------------------------------------------------------------++-- Just a shorthand for below:+type QPar = Par QuasiDet ++-- | Freeze an 'ISet' after a specified callback/handler is done running. This+-- differs from withCallbacksThenFreeze by not taking an additional action to run in+-- the context of the handlers.+--+-- (@'freezeSetAfter' 's' 'f' == 'withCallbacksThenFreeze' 's' 'f' 'return ()' @)+freezeSetAfter :: ISet s a -> (a -> QPar s ()) -> QPar s ()+freezeSetAfter s f = withCallbacksThenFreeze s f (return ())++-- | Register a per-element callback, then run an action in this context, and freeze+-- when all (recursive) invocations of the callback are complete. Returns the final+-- value of the provided action.+withCallbacksThenFreeze :: Eq b => ISet s a -> (a -> QPar s ()) -> QPar s b -> QPar s b+withCallbacksThenFreeze (ISet (WrapLVar lv)) callback action =+ do+ hp <- newPool + res <- IV.new -- TODO, specialize to skip this when the init action returns ()+ WrapPar$ + freezeLVAfter lv (initCB hp res) deltCB+ -- We additionally have to quiesce here because we fork the inital set of+ -- callbacks on their own threads:+ quiesce hp+ IV.get res+ where+ deltCB x = return$ Just$ unWrapPar$ callback x+ initCB hp resIV ref = do+ -- The implementation guarantees that all elements will be caught either here,+ -- or by the delta-callback:+ set <- readIORef ref -- Snapshot+ return $ Just $ unWrapPar $ do+ F.foldlM (\() v -> forkHP (Just hp)$ callback v) () set -- Non-allocating traversal.+ res <- action -- Any additional puts here trigger the callback.+ IV.put_ resIV res++-- | Get the exact contents of the set. Using this may cause your+-- program to exhibit a limited form of nondeterminism: it will never+-- return the wrong answer, but it may include synchronization bugs+-- that can (nondeterministically) cause exceptions.+--+-- This Data.Set based LVar has the special property that you can+-- retrieve the full set without any IO, and without nondeterminism+-- leaking. (This is because the internal order is fixed for the+-- tree-based Data.Set.)+freezeSet :: ISet s a -> QPar s (S.Set a)+freezeSet (ISet (WrapLVar lv)) = WrapPar $ + do freezeLV lv+ getLV lv globalThresh deltaThresh+ where+ globalThresh _ False = return Nothing+ globalThresh ref True = fmap Just $ readIORef ref+ deltaThresh _ = return Nothing++-- | /O(1)/: Convert from an `ISet` to a plain `Data.Set`.+-- This is only permitted when the `ISet` has already been frozen.+-- This is useful for processing the result of `Control.LVish.DeepFrz.runParThenFreeze`. +fromISet :: ISet Frzn a -> S.Set a +-- Alternate names? -- toPure? toSet? fromFrzn??+fromISet (ISet lv) = unsafeDupablePerformIO (readIORef (state lv))+++--------------------------------------------------------------------------------++-- | Add an (asynchronous) callback that listens for all new elements added to+-- the set, optionally enrolled in a handler pool+forEachHP :: Maybe HandlerPool -- ^ pool to enroll in, if any+ -> ISet s a -- ^ Set to listen to+ -> (a -> Par d s ()) -- ^ callback+ -> Par d s ()+forEachHP hp (ISet (WrapLVar lv)) callb = WrapPar $ do+ L.addHandler hp lv globalCB (\x -> return$ Just$ unWrapPar$ callb x)+ return ()+ where+ globalCB ref = do+ set <- readIORef ref -- Snapshot+ return $ Just $ unWrapPar $ + F.foldlM (\() v -> forkHP hp $ callb v) () set -- Non-allocating traversal.++-- | Add an (asynchronous) callback that listens for all new elements added to+-- the set+forEach :: ISet s a -> (a -> Par d s ()) -> Par d s ()+forEach = forEachHP Nothing++-- | Put a single element in the set. (WHNF) Strict in the element being put in the+-- set. +insert :: Ord a => a -> ISet s a -> Par d s ()+insert !elm (ISet (WrapLVar lv)) = WrapPar$ putLV lv putter+ where putter ref = atomicModifyIORef ref update+ update set =+ let set' = S.insert elm set in+ -- Here we do a constant time check to see if we actually changed anything:+ -- For idempotency it is important that we return Nothing if not.+ if S.size set' > S.size set+ then (set',Just elm)+ else (set, Nothing)+++-- | Wait for the set to contain a specified element.+waitElem :: Ord a => a -> ISet s a -> Par d s ()+waitElem !elm (ISet (WrapLVar lv)) = WrapPar $+ getLV lv globalThresh deltaThresh+ where+ globalThresh ref _frzn = do+ set <- readIORef ref+ case S.member elm set of+ True -> return (Just ())+ False -> return (Nothing)+ deltaThresh e2 | e2 == elm = return $ Just ()+ | otherwise = return Nothing +++-- | Wait on the SIZE of the set, not its contents.+waitSize :: Int -> ISet s a -> Par d s ()+waitSize !sz (ISet lv) = WrapPar$+ getLV (unWrapLVar lv) globalThresh deltaThresh+ where+ globalThresh ref _frzn = do+ set <- readIORef ref+ case S.size set >= sz of+ True -> return (Just ())+ False -> return (Nothing)+ -- Here's an example of a situation where we CANNOT TELL if a delta puts it over+ -- the threshold.a+ deltaThresh _ = globalThresh (state lv) False++--------------------------------------------------------------------------------+-- Higher level routines that could be defined using the above interface.+--------------------------------------------------------------------------------++-- | Return a fresh set which will contain strictly more elements than the input set.+-- That is, things put in the former go in the latter, but not vice versa.+copy :: Ord a => ISet s a -> Par d s (ISet s a)+copy = traverseSet return++-- | Establish monotonic map between the input and output sets.+traverseSet :: Ord b => (a -> Par d s b) -> ISet s a -> Par d s (ISet s b)+traverseSet f s = traverseSetHP Nothing f s++-- | An imperative-style, inplace version of 'traverseSet' that takes the output set+-- as an argument.+traverseSet_ :: Ord b => (a -> Par d s b) -> ISet s a -> ISet s b -> Par d s ()+traverseSet_ f s o = void $ traverseSetHP_ Nothing f s o++-- | Return a new set which will (ultimately) contain everything in either input set.+union :: Ord a => ISet s a -> ISet s a -> Par d s (ISet s a)+union = unionHP Nothing++-- | Build a new set which will contain the intersection of the two input sets.+intersection :: Ord a => ISet s a -> ISet s a -> Par d s (ISet s a)+intersection = intersectionHP Nothing++-- | Cartesian product of two sets.+cartesianProd :: (Ord a, Ord b) => ISet s a -> ISet s b -> Par d s (ISet s (a,b))+cartesianProd s1 s2 = cartesianProdHP Nothing s1 s2 + +-- | Takes the cartesian product of several sets.+cartesianProds :: Ord a => [ISet s a] -> Par d s (ISet s [a])+cartesianProds ls = cartesianProdsHP Nothing ls++--------------------------------------------------------------------------------+-- Alternate versions of functions that EXPOSE the HandlerPools+--------------------------------------------------------------------------------++-- | Variant that optionally ties the handlers to a pool.+traverseSetHP :: Ord b => Maybe HandlerPool -> (a -> Par d s b) -> ISet s a ->+ Par d s (ISet s b)+traverseSetHP mh fn set = do+ os <- newEmptySet+ traverseSetHP_ mh fn set os + return os++-- | Variant that optionally ties the handlers to a pool.+traverseSetHP_ :: Ord b => Maybe HandlerPool -> (a -> Par d s b) -> ISet s a -> ISet s b ->+ Par d s ()+traverseSetHP_ mh fn set os = do+ forEachHP mh set $ \ x -> do + x' <- fn x+ insert x' os++-- | Variant that optionally ties the handlers in the resulting set to the same+-- handler pool as those in the two input sets.+unionHP :: Ord a => Maybe HandlerPool -> ISet s a -> ISet s a -> Par d s (ISet s a)+unionHP mh s1 s2 = do+ os <- newEmptySet+ forEachHP mh s1 (`insert` os)+ forEachHP mh s2 (`insert` os)+ return os++-- | Variant that optionally ties the handlers in the resulting set to the same+-- handler pool as those in the two input sets.+intersectionHP :: Ord a => Maybe HandlerPool -> ISet s a -> ISet s a -> Par d s (ISet s a)+-- Can we do intersection with only the public interface? It should be monotonic.+-- Well, for now we cheat and use liftIO:+intersectionHP mh s1 s2 = do+ os <- newEmptySet+ forEachHP mh s1 (fn os s2)+ forEachHP mh s2 (fn os s1)+ return os+ where + fn outSet (ISet lv) elm = do+ -- At this point 'elm' has ALREADY been added to "us", we check "them": + peek <- LI.liftIO$ readIORef (state lv)+ if S.member elm peek + then insert elm outSet+ else return ()++-- | Variant of 'cartesianProd' that optionally ties the handlers to a pool.+cartesianProdHP :: (Ord a, Ord b) => Maybe HandlerPool -> ISet s a -> ISet s b ->+ Par d s (ISet s (a,b))+cartesianProdHP mh s1 s2 = do+ -- This is implemented much like intersection:+ os <- newEmptySet+ forEachHP mh s1 (fn os s2 (\ x y -> (x,y)))+ forEachHP mh s2 (fn os s1 (\ x y -> (y,x)))+ return os+ where+ -- This is expensive, but we've got to do it from both sides to counteract races:+ fn outSet (ISet lv) cmbn elm1 = do+ peek <- LI.liftIO$ readIORef (state lv)+ F.foldlM (\() elm2 -> insert (cmbn elm1 elm2) outSet) () peek+++-- | Variant of 'cartesianProds' that optionally ties the handlers to a pool.+cartesianProdsHP :: Ord a => Maybe HandlerPool -> [ISet s a] ->+ Par d s (ISet s [a])+cartesianProdsHP _ [] = newEmptySet+cartesianProdsHP mh ls = do+#if 1+ -- Case 1: recursive definition in terms of pairwise products:+ -- It would be best to create a balanced tree of these, I believe:+ let loop [lst] = traverseSetHP mh (\x -> return [x]) lst -- Inefficient!+ loop (nxt:rst) = do+ partial <- loop rst+ p1 <- cartesianProdHP mh nxt partial+ traverseSetHP mh (\ (x,tl) -> return (x:tl)) p1 -- Inefficient!!+ loop ls+#else+ os <- newEmptySet+ let loop done [] acc = acc+ loop done (nxt:rest) acc =+ addHandler hp nxt (fn os done rest)+ +-- forM_ ls $ \ inSet -> do +-- addHandler hp s1 (fn os s2 (\ x y -> (x,y)))++ return os+ where+ fn outSet left right newElm = do+ peeksL <- liftIO$ mapM (readIORef . state . unISet) left+ peeksR <- liftIO$ mapM (readIORef . state . unISet) right++-- F.foldlM (\() elm2 -> insert (cmbn elm1 elm2) outSet) () peek+ return undefined+#endif+
+ Data/LVar/SLMap.hs view
@@ -0,0 +1,361 @@+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE MagicHash #-}++{-|++ This module provides finite maps that only grow. It is based on a concurrent-skip-list+ implementation of maps.++ Note that this module provides almost the same interface as "Data.LVar.PureMap",+ but this module is usually more efficient. However, it's always good to test muliple+ data structures if you have a performance-critical use case.++ -}+++module Data.LVar.SLMap+ (+ -- * The type and its basic operations+ IMap,+ newEmptyMap, newMap, newFromList,+ insert, + getKey, waitSize, modify, freezeMap,+ -- waitValue, ++ -- * Iteration and callbacks+ forEach, forEachHP, + withCallbacksThenFreeze,++ -- * Higher-level derived operations+ copy, traverseMap, traverseMap_, + + -- * Alternate versions of derived ops that expose HandlerPools they create.+ traverseMapHP, traverseMapHP_, unionHP,++ ) where++import Control.Exception (throw)+import Control.Applicative+import Data.Concurrent.SkipListMap as SLM+import qualified Data.Map.Strict as M+import qualified Data.LVar.IVar as IV+import qualified Data.Foldable as F+import Data.IORef (readIORef)+import Data.UtilInternal (traverseWithKey_)+import Data.List (intersperse)+import Data.LVar.Generic+import Data.LVar.Generic.Internal (unsafeCoerceLVar)+import Control.Monad+import Control.Monad.IO.Class+import Control.LVish+import Control.LVish.DeepFrz.Internal+import Control.LVish.Internal as LI+import Control.LVish.SchedIdempotent (newLV, putLV, putLV_, getLV, freezeLV)+import qualified Control.LVish.SchedIdempotent as L+import System.Random (randomIO)+import System.IO.Unsafe (unsafeDupablePerformIO)+import GHC.Prim (unsafeCoerce#)+import Prelude++type QPar = Par QuasiDet -- Shorthand used below.++------------------------------------------------------------------------------+-- IMaps implemented vis SkipListMap+------------------------------------------------------------------------------++-- | The map datatype itself. Like all other LVars, it has an @s@ parameter (think+-- `STRef`) in addition to the @a@ parameter that describes the type of elements+-- in the set.+--+-- Performance note: this data structure reduces contention between parallel+-- computations inserting into the map, but all /blocking/ computations are not as+-- scalable. All continuations waiting for not-yet-present elements will currently+-- share a single queue [2013.09.26].+data IMap k s v = Ord k => IMap {-# UNPACK #-} !(LVar s (SLM.SLMap k v) (k,v))++-- | Equality is physical equality, as with @IORef@s.+instance Eq (IMap k s v) where+ IMap lv1 == IMap lv2 = state lv1 == state lv2 ++-- | An `IMap` can be treated as a generic container LVar. However, the polymorphic+-- operations are less useful than the monomorphic ones exposed by this module.+instance LVarData1 (IMap k) where+ -- | Get the exact contents of the map. Using this may cause your+ -- program to exhibit a limited form of nondeterminism: it will never+ -- return the wrong answer, but it may include synchronization bugs+ -- that can (nondeterministically) cause exceptions. + freeze orig@(IMap (WrapLVar lv)) =+ WrapPar$ do freezeLV lv; return (unsafeCoerceLVar orig)+ + -- | We can do better than the default here; this is /O(1)/: + sortFrzn = AFoldable ++ -- | This generic version has the disadvantage that it does not observe the KEY,+ -- only the value.+ addHandler mh (IMap (WrapLVar lv)) callb = WrapPar $ + L.addHandler mh lv globalCB (\(_k,v) -> return$ Just$ unWrapPar$ callb v)+ where+ globalCB slm = + return $ Just $ unWrapPar $+ SLM.foldlWithKey (\() _k v -> forkHP mh $ callb v) () slm++-- | The `IMap`s in this module also have the special property that they support an+-- `O(1)` freeze operation which immediately yields a `Foldable` container+-- (`snapFreeze`).+instance OrderedLVarData1 (IMap k) where+ snapFreeze is = unsafeCoerceLVar <$> freeze is++-- | `IMap` values can be returned as the result of a `runParThenFreeze`.+-- Hence they need a `DeepFrz` instace.+-- @DeepFrz@ is just a type-coercion. No bits flipped at runtime.+instance DeepFrz a => DeepFrz (IMap k s a) where+ type FrzType (IMap k s a) = IMap k Frzn (FrzType a)+ frz = unsafeCoerceLVar++--------------------------------------------------------------------------------++-- | The default number of skiplist levels+defaultLevels :: Int+defaultLevels = 8++-- | Create a fresh map with nothing in it.+newEmptyMap :: Ord k => Par d s (IMap k s v)+newEmptyMap = newEmptyMap_ defaultLevels++-- | Create a fresh map with nothing in it, with the given number of skiplist+-- levels.+newEmptyMap_ :: Ord k => Int -> Par d s (IMap k s v)+newEmptyMap_ n = fmap (IMap . WrapLVar) $ WrapPar $ newLV $ SLM.newSLMap n++-- | Create a new map populated with initial elements.+newMap :: Ord k => M.Map k v -> Par d s (IMap k s v)+newMap mp =+ fmap (IMap . WrapLVar) $ WrapPar $ newLV $ do+ slm <- SLM.newSLMap defaultLevels + traverseWithKey_ (\ k v -> do Added _ <- SLM.putIfAbsent slm k (return v)+ return ()+ ) mp+ return slm++-- | Create a new 'IMap' drawing initial elements from an existing list.+newFromList :: (Ord k, Eq v) =>+ [(k,v)] -> Par d s (IMap k s v)+newFromList ls = newFromList_ ls defaultLevels++-- | Create a new 'IMap' drawing initial elements from an existing list, with+-- the given number of skiplist levels.+newFromList_ :: Ord k => [(k,v)] -> Int -> Par d s (IMap k s v)+newFromList_ ls n = do + m@(IMap lv) <- newEmptyMap_ n+ forM_ ls $ \(k,v) -> LI.liftIO $ SLM.putIfAbsent (state lv) k $ return v+ return m++-- | Register a per-element callback, then run an action in this context, and freeze+-- when all (recursive) invocations of the callback are complete. Returns the final+-- value of the provided action.+withCallbacksThenFreeze :: forall k v b s . Eq b =>+ IMap k s v -> (k -> v -> QPar s ()) -> QPar s b -> QPar s b+withCallbacksThenFreeze (IMap lv) callback action = do+ hp <- newPool + res <- IV.new + let deltCB (k,v) = return$ Just$ unWrapPar$ callback k v+ initCB slm = do+ -- The implementation guarantees that all elements will be caught either here,+ -- or by the delta-callback:+ return $ Just $ unWrapPar $ do+ SLM.foldlWithKey (\() k v -> forkHP (Just hp) $ callback k v) () slm+ x <- action -- Any additional puts here trigger the callback.+ IV.put_ res x+ WrapPar $ L.addHandler (Just hp) (unWrapLVar lv) initCB deltCB+ + -- We additionally have to quiesce here because we fork the inital set of+ -- callbacks on their own threads:+ quiesce hp+ IV.get res++-- | Add an (asynchronous) callback that listens for all new key/value pairs+-- added to the map, optionally tied to a handler pool.+forEachHP :: Maybe HandlerPool -- ^ optional pool to enroll in + -> IMap k s v -- ^ Map to listen to+ -> (k -> v -> Par d s ()) -- ^ callback+ -> Par d s ()+forEachHP mh (IMap (WrapLVar lv)) callb = WrapPar $ + L.addHandler mh lv globalCB (\(k,v) -> return$ Just$ unWrapPar$ callb k v)+ where+ globalCB slm = + return $ Just $ unWrapPar $+ SLM.foldlWithKey (\() k v -> forkHP mh $ callb k v) () slm+ +-- | Add an (asynchronous) callback that listens for all new new key/value pairs added to+-- the map+forEach :: IMap k s v -> (k -> v -> Par d s ()) -> Par d s ()+forEach = forEachHP Nothing ++-- | Put a single entry into the map. (WHNF) Strict in the key and value.+insert :: (Ord k, Eq v) =>+ k -> v -> IMap k s v -> Par d s () +insert !key !elm (IMap (WrapLVar lv)) = WrapPar$ putLV lv putter+ where putter slm = do+ putRes <- SLM.putIfAbsent slm key $ return elm+ case putRes of+ Added _ -> return $ Just (key, elm)+ Found _ -> throw$ ConflictingPutExn$ "Multiple puts to one entry in an IMap!"+ +-- | IMap's containing other LVars have some additional capabilities compared to+-- those containing regular Haskell data. In particular, it is possible to modify+-- existing entries (monotonically). Further, this `modify` function implicitly+-- inserts a "bottom" element if there is no existing entry for the key.+--+modify :: forall f a b d s key . (Ord key, LVarData1 f, Show key, Ord a) =>+ IMap key s (f s a)+ -> key -- ^ The key to lookup.+ -> (Par d s (f s a)) -- ^ Create a new "bottom" element whenever an entry is not present.+ -> (f s a -> Par d s b) -- ^ The computation to apply on the right-hand-side of the keyed entry.+ -> Par d s b+modify (IMap (WrapLVar lv)) key newBottom fn = do+ act <- WrapPar $ putLV_ lv putter+ act+ where putter slm = do+ putRes <- unWrapPar $ SLM.putIfAbsent slm key newBottom+ case putRes of+ Added v -> return (Just (key,v), fn v)+ Found v -> return (Nothing, fn v) + +-- | Wait for the map to contain a specified key, and return the associated value.+getKey :: Ord k => k -> IMap k s v -> Par d s v+getKey !key (IMap (WrapLVar lv)) = WrapPar$ getLV lv globalThresh deltaThresh+ where+ globalThresh slm _frzn = SLM.find slm key+ deltaThresh (k,v) | k == key = return $ Just v+ | otherwise = return Nothing ++-- | Wait until the map contains a certain value (on any key).+waitValue :: (Ord k, Eq v) => v -> IMap k s v -> Par d s ()+waitValue !val (IMap (WrapLVar lv)) = error "TODO / FINISHME SLMap.waitValue"++-- | Wait on the SIZE of the map, not its contents.+waitSize :: Int -> IMap k s v -> Par d s ()+waitSize !sz (IMap (WrapLVar lv)) = WrapPar $+ getLV lv globalThresh deltaThresh+ where+ globalThresh slm _ = do+ snapSize <- SLM.foldlWithKey (\n _ _ -> return $ n+1) 0 slm+ case snapSize >= sz of+ True -> return (Just ())+ False -> return (Nothing)+ -- Here's an example of a situation where we CANNOT TELL if a delta puts it over+ -- the threshold.a+ deltaThresh _ = globalThresh (L.state lv) False++-- | Get the exact contents of the map Using this may cause your+-- program to exhibit a limited form of nondeterminism: it will never+-- return the wrong answer, but it may include synchronization bugs+-- that can (nondeterministically) cause exceptions.+--+-- This is an O(1) operation that doesn't copy the in-memory representation of the+-- IMap.+freezeMap :: Ord k => IMap k s v -> QPar s (IMap k Frzn v)+-- freezeMap (IMap (WrapLVar lv)) = return (IMap (WrapLVar lv))+-- OR we can just do this:++freezeMap x@(IMap (WrapLVar lv)) = WrapPar $ do+ freezeLV lv+ -- For the final deepFreeze at the end of a runpar we can actually skip+ -- the freezeLV part.... + return (unsafeCoerce# x)++--------------------------------------------------------------------------------+-- Higher level routines that could (mostly) be defined using the above interface.+--------------------------------------------------------------------------------++-- | Establish monotonic map between the input and output sets. Produce a new result+-- based on each element, while leaving the keys the same.+traverseMap :: (Ord k, Eq b) =>+ (k -> a -> Par d s b) -> IMap k s a -> Par d s (IMap k s b)+traverseMap f s = traverseMapHP Nothing f s++-- | An imperative-style, inplace version of 'traverseMap' that takes the output set+-- as an argument.+traverseMap_ :: (Ord k, Eq b) =>+ (k -> a -> Par d s b) -> IMap k s a -> IMap k s b -> Par d s ()+traverseMap_ f s o = traverseMapHP_ Nothing f s o++--------------------------------------------------------------------------------+-- Alternate versions of functions that EXPOSE the HandlerPools+--------------------------------------------------------------------------------++-- | Return a fresh map which will contain strictly more elements than the input.+-- That is, things put in the former go in the latter, but not vice versa.+copy :: (Ord k, Eq v) => IMap k s v -> Par d s (IMap k s v)+copy = traverseMap (\ _ x -> return x)++-- | Variant that optionally ties the handlers to a pool.+traverseMapHP :: (Ord k, Eq b) =>+ Maybe HandlerPool -> (k -> a -> Par d s b) -> IMap k s a ->+ Par d s (IMap k s b)+traverseMapHP mh fn set = do+ os <- newEmptyMap+ traverseMapHP_ mh fn set os + return os++-- | Variant that optionally ties the handlers to a pool.+traverseMapHP_ :: (Ord k, Eq b) =>+ Maybe HandlerPool -> (k -> a -> Par d s b) -> IMap k s a -> IMap k s b ->+ Par d s ()+traverseMapHP_ mh fn set os = do+ forEachHP mh set $ \ k x -> do + x' <- fn k x+ insert k x' os++-- | Return a new map which will (ultimately) contain everything in either input+-- map. Conflicting entries will result in a multiple put exception.+-- Optionally ties the handlers to a pool.+unionHP :: (Ord k, Eq a) => Maybe HandlerPool ->+ IMap k s a -> IMap k s a -> Par d s (IMap k s a)+unionHP mh m1 m2 = do+ os <- newEmptyMap+ forEachHP mh m1 (\ k v -> insert k v os)+ forEachHP mh m2 (\ k v -> insert k v os)+ return os++--------------------------------------------------------------------------------+-- Operations on frozen Maps+--------------------------------------------------------------------------------++-- | As with all LVars, after freezing, map elements can be consumed. In the case of+-- this `IMap` implementation, it need only be `Frzn`, not `Trvrsbl`.+instance F.Foldable (IMap k Frzn) where+ -- Note: making these strict for now: + foldr fn zer (IMap (WrapLVar lv)) =+ unsafeDupablePerformIO $+ SLM.foldlWithKey (\ a _k v -> return (fn v a))+ zer (L.state lv)++-- | Of course, the stronger `Trvrsbl` state is still fine for folding.+instance F.Foldable (IMap k Trvrsbl) where+ foldr fn zer mp = F.foldr fn zer (castFrzn mp)++instance (Show k, Show a) => Show (IMap k Frzn a) where+ show (IMap (WrapLVar lv)) =+ "{IMap: " +++ (concat $ intersperse ", " $ + unsafeDupablePerformIO $+ SLM.foldlWithKey (\ acc k v -> return$ show (k, v) : acc)+ [] (L.state lv)+ ) ++ "}"++-- | For convenience only; the user could define this.+instance (Show k, Show a) => Show (IMap k Trvrsbl a) where+ show lv = show (castFrzn lv)+
+ Data/LVar/SLSet.hs view
@@ -0,0 +1,397 @@+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE GADTs #-}++{-|++ This module provides sets that only grow. It is based on a concurrent-skip-list+ implementation of sets.++ Note that this module provides almost the same interface as "Data.LVar.PureSet",+ but this module is usually more efficient. However, it's always good to test muliple+ data structures if you have a performance-critical use case.++ -}++module Data.LVar.SLSet+ (+ -- * Basic operations+ ISet, + newEmptySet, newSet, newFromList,+ insert, waitElem, waitSize, + member,+ + -- * Iteration and callbacks+ forEach, forEachHP,++ -- * Quasi-deterministic operations+ freezeSetAfter, withCallbacksThenFreeze, ++ -- * Higher-level derived operations+ copy, traverseSet, traverseSet_, union, intersection,+ cartesianProd, cartesianProds, ++ -- * Alternate versions of derived ops that expose HandlerPools they create.+ traverseSetHP, traverseSetHP_,+ cartesianProdHP, cartesianProdsHP+ ) where ++import Control.Applicative+import qualified Data.Foldable as F+import Data.Concurrent.SkipListMap as SLM+import Data.List (intersperse)+import qualified Data.Set as S+import qualified Data.LVar.IVar as IV+import Data.LVar.Generic+import Data.LVar.Generic.Internal (unsafeCoerceLVar)+import Control.Monad+import Control.LVish as LV+import Control.LVish.DeepFrz.Internal+import Control.LVish.Internal as LI+import Control.LVish.SchedIdempotent (newLV, putLV, getLV, freezeLV)+import qualified Control.LVish.SchedIdempotent as L+import System.IO.Unsafe (unsafeDupablePerformIO)+import Prelude hiding (insert)++------------------------------------------------------------------------------+-- ISets implemented via SkipListMap+------------------------------------------------------------------------------++-- | The set datatype itself. Like all other LVars, it has an @s@ parameter (think+-- `STRef`) in addition to the @a@ parameter that describes the type of elements+-- in the set.+--+-- Performance note: this data structure reduces contention between parallel+-- computations inserting into the map, but all /blocking/ computations are not as+-- scalable. All continuations waiting for not-yet-present elements will currently+-- share a single queue [2013.09.26].+data ISet s a = Ord a => ISet {-# UNPACK #-}!(LVar s (SLM.SLMap a ()) a)+-- TODO: Address the possible inefficiency of carrying Ord dictionaries at runtime.++-- | Physical identity, just as with IORefs.+instance Eq (ISet s v) where+ ISet slm1 == ISet slm2 = state slm1 == state slm2+ +-- | An `ISet` can be treated as a generic container LVar.+instance LVarData1 ISet where+ -- In order to make freeze an O(1) operation, freeze is just a cast from the+ -- mutable to the immutable form of the data structure.+ freeze orig@(ISet (WrapLVar lv)) =+ WrapPar$ do freezeLV lv; return (unsafeCoerceLVar orig)+ addHandler = forEachHP + -- | We can do better than the default here; this is /O(1)/: + sortFrzn (is :: ISet Frzn a) = AFoldable is+++-- | The `ISet`s in this module also have the special property that they support an+-- `O(1)` freeze operation which immediately yields a `Foldable` container+-- (`snapFreeze`).+instance OrderedLVarData1 ISet where+ snapFreeze is = unsafeCoerceLVar <$> freeze is++-- | `ISet` values can be returned as the result of a `runParThenFreeze`.+-- Hence they need a `DeepFrz` instance.+-- @DeepFrz@ is just a type-coercion. No bits flipped at runtime.+instance DeepFrz a => DeepFrz (ISet s a) where+ type FrzType (ISet s a) = ISet Frzn (FrzType a)+ frz = unsafeCoerceLVar++instance Show a => Show (ISet Frzn a) where+ show lv = "{ISet: " +++ (concat $ intersperse ", " $ map show $ + F.foldr (\ elm ls -> elm : ls) []+ (unsafeCoerceLVar lv :: ISet Trvrsbl a)) ++ "}"++-- | For convenience only; the user could define this.+instance Show a => Show (ISet Trvrsbl a) where+ show lv = show (castFrzn lv)++--------------------------------------------------------------------------------++-- | Test whether an element is in a frozen image of a set.+member :: a -> ISet Frzn a -> Bool+member elm (ISet (WrapLVar lv)) =+ case unsafeDupablePerformIO (SLM.find (L.state lv) elm) of+ Just () -> True+ Nothing -> False++-- | As with all LVars, after freezing, map elements can be consumed. In the case of+-- this `ISet` implementation, it need only be `Frzn`, not `Trvrsbl`.+instance F.Foldable (ISet Frzn) where+ foldr fn zer (ISet (WrapLVar lv)) =+ unsafeDupablePerformIO $+ SLM.foldlWithKey (\ a k _v -> return (fn k a))+ zer (L.state lv)++-- | Of course, the stronger `Trvrsbl` state is still fine for folding.+instance F.Foldable (ISet Trvrsbl) where+ foldr fn zer mp = F.foldr fn zer (castFrzn mp)+++-- | The default number of skiplist levels+defaultLevels :: Int+defaultLevels = 8++-- | Create a new, empty, monotonically growing 'ISet'.+newEmptySet :: Ord a => Par d s (ISet s a)+newEmptySet = newEmptySet_ defaultLevels++-- | Tuning: Create a new, empty, monotonically growing 'ISet', with the given number+-- of skiplist levels.+newEmptySet_ :: Ord a => Int -> Par d s (ISet s a)+newEmptySet_ n = fmap (ISet . WrapLVar) $ WrapPar $ newLV $ SLM.newSLMap n++-- | Create a new set populated with initial elements.+newSet :: Ord a => S.Set a -> Par d s (ISet s a)+newSet set = + fmap (ISet . WrapLVar) $ WrapPar $ newLV $ do+ slm <- SLM.newSLMap defaultLevels+ F.foldlM (\ () elm -> do+ SLM.Added _ <- SLM.putIfAbsent slm elm (return ())+ return ()+ ) () set+ return slm++-- | A simple convenience function. Create a new 'ISet' drawing initial elements from an existing list.+newFromList :: Ord a => [a] -> Par d s (ISet s a)+newFromList ls = newFromList_ ls defaultLevels++-- | Create a new 'ISet' drawing initial elements from an existing list, with+-- the given number of skiplist levels.+newFromList_ :: Ord a => [a] -> Int -> Par d s (ISet s a)+newFromList_ ls n = do + s@(ISet lv) <- newEmptySet_ n+ LI.liftIO $ forM_ ls $ \x ->+ SLM.putIfAbsent (state lv) x $ return ()+ return s++-- (Todo: in production you might want even more ... like going from a Vector)++--------------------------------------------------------------------------------+-- Quasi-deterministic ops:+--------------------------------------------------------------------------------++type QPar = Par QuasiDet ++-- | Freeze an 'ISet' after a specified callback/handler is done running. This+-- differs from withCallbacksThenFreeze by not taking an additional action to run in+-- the context of the handlers.+--+-- (@'freezeSetAfter' 's' 'f' == 'withCallbacksThenFreeze' 's' 'f' 'return ()' @)+freezeSetAfter :: ISet s a -> (a -> QPar s ()) -> QPar s ()+freezeSetAfter s f = withCallbacksThenFreeze s f (return ())+ +-- | Register a per-element callback, then run an action in this context, and freeze+-- when all (recursive) invocations of the callback are complete. Returns the final+-- value of the provided action.+withCallbacksThenFreeze :: Eq b => ISet s a -> (a -> QPar s ()) -> QPar s b -> QPar s b+withCallbacksThenFreeze (ISet lv) callback action = do+ hp <- newPool + res <- IV.new -- TODO, specialize to skip this when the init action returns ()+ let deltCB x = return$ Just$ unWrapPar$ callback x+ initCB slm = do+ -- The implementation guarantees that all elements will be caught either here,+ -- or by the delta-callback:+ return $ Just $ unWrapPar $ do+ SLM.foldlWithKey (\() v () -> forkHP (Just hp) $ callback v) () slm+ x <- action -- Any additional puts here trigger the callback.+ IV.put_ res x+ WrapPar $ L.addHandler (Just hp) (unWrapLVar lv) initCB deltCB+ + -- We additionally have to quiesce here because we fork the inital set of+ -- callbacks on their own threads:+ quiesce hp+ IV.get res+++--------------------------------------------------------------------------------++-- | Add an (asynchronous) callback that listens for all new elements added to+-- the set, optionally enrolled in a handler pool.+forEachHP :: Maybe HandlerPool -- ^ optional pool to enroll in + -> ISet s a -- ^ Set to listen to+ -> (a -> Par d s ()) -- ^ callback+ -> Par d s ()+forEachHP hp (ISet (WrapLVar lv)) callb = WrapPar $ + L.addHandler hp lv globalCB (\x -> return$ Just$ unWrapPar$ callb x)+ where+ globalCB slm = + return $ Just $ unWrapPar $+ SLM.foldlWithKey (\() v () -> forkHP hp $ callb v) () slm++-- | Add an (asynchronous) callback that listens for all new elements added to+-- the set+forEach :: ISet s a -> (a -> Par d s ()) -> Par d s ()+forEach = forEachHP Nothing++-- | Put a single element in the set. (WHNF) Strict in the element being put in the+-- set. +insert :: Ord a => a -> ISet s a -> Par d s ()+insert !elm (ISet lv) = WrapPar$ putLV (unWrapLVar lv) putter+ where putter slm = do+ putRes <- SLM.putIfAbsent slm elm $ return ()+ case putRes of+ Added _ -> return $ Just elm+ Found _ -> return Nothing ++-- | Wait for the set to contain a specified element.+waitElem :: Ord a => a -> ISet s a -> Par d s ()+waitElem !elm (ISet (WrapLVar lv)) = WrapPar $+ getLV lv globalThresh deltaThresh+ where+ globalThresh slm _frzn = SLM.find slm elm+ deltaThresh e2 | e2 == elm = return $ Just ()+ | otherwise = return Nothing++-- | Wait on the SIZE of the set, not its contents.+waitSize :: Int -> ISet s a -> Par d s ()+waitSize !sz (ISet (WrapLVar lv)) = WrapPar$+ getLV lv globalThresh deltaThresh+ where+ globalThresh slm _ = do+ snapSize <- SLM.foldlWithKey (\n _ _ -> return $ n+1) 0 slm+ case snapSize >= sz of+ True -> return (Just ())+ False -> return (Nothing)+ -- Here's an example of a situation where we CANNOT TELL if a delta puts it over+ -- the threshold.a+ deltaThresh _ = globalThresh (L.state lv) False++--------------------------------------------------------------------------------+-- Higher level routines that could be defined using the above interface.+--------------------------------------------------------------------------------++-- | Return a fresh set which will contain strictly more elements than the input set.+-- That is, things put in the former go in the latter, but not vice versa.+copy :: Ord a => ISet s a -> Par d s (ISet s a)+copy = traverseSet return ++-- | Establish monotonic map between the input and output sets.+traverseSet :: Ord b => (a -> Par d s b) -> ISet s a -> Par d s (ISet s b)+traverseSet f s = traverseSetHP Nothing f s++-- | An imperative-style, inplace version of 'traverseSet' that takes the output set+-- as an argument.+traverseSet_ :: Ord b => (a -> Par d s b) -> ISet s a -> ISet s b -> Par d s ()+traverseSet_ f s o = traverseSetHP_ Nothing f s o++-- | Return a new set which will (ultimately) contain everything in either input set.+union :: Ord a => ISet s a -> ISet s a -> Par d s (ISet s a)+union = unionHP Nothing++-- | Build a new set which will contain the intersection of the two input sets.+intersection :: Ord a => ISet s a -> ISet s a -> Par d s (ISet s a)+intersection = intersectionHP Nothing++-- | Cartesian product of two sets.+cartesianProd :: (Ord a, Ord b) => ISet s a -> ISet s b -> Par d s (ISet s (a,b))+cartesianProd s1 s2 = cartesianProdHP Nothing s1 s2 + +-- | Takes the cartesian product of several sets.+cartesianProds :: Ord a => [ISet s a] -> Par d s (ISet s [a])+cartesianProds ls = cartesianProdsHP Nothing ls++--------------------------------------------------------------------------------+-- Alternate versions of functions that EXPOSE the HandlerPools+--------------------------------------------------------------------------------++-- | Variant that optionally ties the handlers to a pool.+traverseSetHP :: Ord b => Maybe HandlerPool -> (a -> Par d s b) -> ISet s a ->+ Par d s (ISet s b)+traverseSetHP mh fn set = do+ os <- newEmptySet+ traverseSetHP_ mh fn set os + return os++-- | Variant that optionally ties the handlers to a pool.+traverseSetHP_ :: Ord b => Maybe HandlerPool -> (a -> Par d s b) -> ISet s a -> ISet s b ->+ Par d s ()+traverseSetHP_ mh fn set os = do+ forEachHP mh set $ \ x -> do + x' <- fn x+ insert x' os++-- | Variant that optionally ties the handlers in the resulting set to the same+-- handler pool as those in the two input sets.+unionHP :: Ord a => Maybe HandlerPool -> ISet s a -> ISet s a -> Par d s (ISet s a)+unionHP mh s1 s2 = do+ os <- newEmptySet+ forEachHP mh s1 (`insert` os)+ forEachHP mh s2 (`insert` os)+ return os++-- | Variant that optionally ties the handlers in the resulting set to the same+-- handler pool as those in the two input sets.+intersectionHP :: Ord a => Maybe HandlerPool -> ISet s a -> ISet s a -> Par d s (ISet s a)+-- Can we do intersection with only the public interface? It should be monotonic.+-- AJT: You could do it using cartesian product...+-- Well, for now we cheat and use liftIO:+intersectionHP mh s1 s2 = do+ os <- newEmptySet+ forEachHP mh s1 (fn os s2)+ forEachHP mh s2 (fn os s1)+ return os+ where + fn outSet other@(ISet lv) elm = do+ -- At this point 'elm' has ALREADY been added to "us", we check "them": + peek <- LI.liftIO $ SLM.find (state lv) elm+ case peek of+ Just _ -> insert elm outSet+ Nothing -> return ()++-- | Variant of 'cartesianProd' that optionally ties the handlers to a pool.+cartesianProdHP :: (Ord a, Ord b) => Maybe HandlerPool -> ISet s a -> ISet s b ->+ Par d s (ISet s (a,b))+cartesianProdHP mh s1 s2 = do+ -- This is implemented much like intersection:+ os <- newEmptySet+ forEachHP mh s1 (fn os s2 (\ x y -> (x,y)))+ forEachHP mh s2 (fn os s1 (\ x y -> (y,x)))+ return os+ where+ -- This is expensive, but we've got to do it from both sides to counteract races:+ fn outSet other@(ISet lv) cmbn elm1 = + SLM.foldlWithKey (\() elm2 () -> insert (cmbn elm1 elm2) outSet) () (state lv)++-- | Variant of 'cartesianProds' that optionally ties the handlers to a pool.+cartesianProdsHP :: Ord a => Maybe HandlerPool -> [ISet s a] ->+ Par d s (ISet s [a])+cartesianProdsHP mh [] = newEmptySet+cartesianProdsHP mh ls = do+#if 1+ -- Case 1: recursive definition in terms of pairwise products:+ -- It would be best to create a balanced tree of these, I believe:+ let loop [lst] = traverseSetHP mh (\x -> return [x]) lst -- Inefficient!+ loop (nxt:rst) = do+ partial <- loop rst+ p1 <- cartesianProdHP mh nxt partial+ traverseSetHP mh (\ (x,tl) -> return (x:tl)) p1 -- Inefficient!!+ loop ls+#else+ os <- newEmptySet+ let loop done [] acc = acc+ loop done (nxt:rest) acc =+ addHandler hp nxt (fn os done rest)+ +-- forM_ ls $ \ inSet -> do +-- addHandler hp s1 (fn os s2 (\ x y -> (x,y)))++ return os+ where+ fn outSet left right newElm = do+ peeksL <- liftIO$ mapM (readIORef . state . unISet) left+ peeksR <- liftIO$ mapM (readIORef . state . unISet) right++-- F.foldlM (\() elm2 -> insert (cmbn elm1 elm2) outSet) () peek+ return undefined+#endif+
+ Data/UtilInternal.hs view
@@ -0,0 +1,34 @@++-- | A module with helper functions that are used elsewhere in the LVish repository.++module Data.UtilInternal+ (+ traverseWithKey_+ )+ where++import Control.Applicative+import Control.Monad (void)+import Data.Monoid (Monoid(..))+import Data.Map as M++--------------------------------------------------------------------------------+-- Helper code.+--------------------------------------------------------------------------------+ +-- Version of traverseWithKey_ from Shachaf Ben-Kiki+-- (See thread on Haskell-cafe.)+-- Avoids O(N) allocation when traversing for side-effect.++newtype Traverse_ f = Traverse_ { runTraverse_ :: f () }+instance Applicative f => Monoid (Traverse_ f) where+ mempty = Traverse_ (pure ())+ Traverse_ a `mappend` Traverse_ b = Traverse_ (a *> b)+-- Since the Applicative used is Const (newtype Const m a = Const m), the+-- structure is never built up.+--(b) You can derive traverseWithKey_ from foldMapWithKey, e.g. as follows:+traverseWithKey_ :: Applicative f => (k -> a -> f ()) -> M.Map k a -> f ()+traverseWithKey_ f = runTraverse_ .+ foldMapWithKey (\k x -> Traverse_ (void (f k x)))+foldMapWithKey :: Monoid r => (k -> a -> r) -> M.Map k a -> r+foldMapWithKey f = getConst . M.traverseWithKey (\k x -> Const (f k x))
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Lindsey Kuper, Ryan Newton, Aaron Turon 2012++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Simon Marlow nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ TestHelpers.hs view
@@ -0,0 +1,186 @@+{-# LANGUAGE BangPatterns, CPP #-}++-- | To make it easier to build (multithreaded) tests++module TestHelpers+ ( + -- * Testing parameters+ numElems, getNumAgents, producerRatio,++ -- * Utility for controlling the number of threads used by generated tests.+ setTestThreads,++ -- * Test initialization, reading common configs+ stdTestHarness+ )+ where ++import Data.IORef+import Control.Monad+import qualified Data.Set as S+import Text.Printf+import Control.Concurrent (forkOS, forkIO, ThreadId)+-- import Control.Exception (catch, SomeException, fromException, bracket, AsyncException(ThreadKilled))+import Control.Exception (bracket)+import System.Environment (withArgs, getArgs, getEnvironment)+import System.IO (hFlush, stdout)+import System.IO.Unsafe (unsafePerformIO)+import qualified Test.Framework as TF+import Test.Framework.Providers.HUnit (hUnitTestToTests)+import Test.HUnit as HU++import Debug.Trace (trace)++--------------------------------------------------------------------------------+++#if __GLASGOW_HASKELL__ >= 704+import GHC.Conc (getNumCapabilities, setNumCapabilities, getNumProcessors)+#else+import GHC.Conc (numCapabilities)+getNumCapabilities :: IO Int+getNumCapabilities = return numCapabilities++setNumCapabilities :: Int -> IO ()+setNumCapabilities = error "setNumCapabilities not supported in this older GHC! Set NUMTHREADS and +RTS -N to match."++getNumProcessors :: IO Int+getNumProcessors = return 1 +#endif ++theEnv :: [(String, String)]+theEnv = unsafePerformIO getEnvironment++----------------------------------------------------------------------------------------------------+-- TODO: In addition to setting these parameters from environment+-- variables, it would be nice to route all of this through a+-- configuration record, so that it can be changed programmatically.++-- How many elements should each of the tests pump through the queue(s)?+numElems :: Int+numElems = case lookup "NUMELEMS" theEnv of + Nothing -> 100 * 1000 -- 500000+ Just str -> warnUsing ("NUMELEMS = "++str) $ + read str++forkThread :: IO () -> IO ThreadId+forkThread = case lookup "OSTHREADS" theEnv of + Nothing -> forkIO+ Just x -> warnUsing ("OSTHREADS = "++x) $ + case x of + "0" -> forkIO+ "False" -> forkIO+ "1" -> forkOS+ "True" -> forkOS+ oth -> error$"OSTHREAD environment variable set to unrecognized option: "++oth++-- | How many communicating agents are there? By default one per+-- thread used by the RTS.+getNumAgents :: IO Int+getNumAgents = case lookup "NUMAGENTS" theEnv of + Nothing -> getNumCapabilities+ Just str -> warnUsing ("NUMAGENTS = "++str) $ + return (read str)++-- | It is possible to have imbalanced concurrency where there is more+-- contention on the producing or consuming side (which corresponds to+-- settings of this parameter less than or greater than 1).+producerRatio :: Double+producerRatio = case lookup "PRODUCERRATIO" theEnv of + Nothing -> 1.0+ Just str -> warnUsing ("PRODUCERRATIO = "++str) $ + read str++warnUsing :: String -> a -> a+warnUsing str a = trace (" [Warning]: Using environment variable "++str) a+++-- | Dig through the test constructors to find the leaf IO actions and bracket them+-- with a thread-setting action.+setTestThreads :: Int -> HU.Test -> HU.Test+setTestThreads nm tst = loop False tst+ where+ loop flg x = + case x of+ TestLabel lb t2 -> TestLabel (decor flg lb) (loop True t2)+ TestList ls -> TestList (map (loop flg) ls)+ TestCase io -> TestCase (bracketThreads nm io)++ -- We only need to insert the numcapabilities in the description string ONCE:+ decor False lb = "N"++show nm++"_"++ lb+ decor True lb = lb++ bracketThreads :: Int -> IO a -> IO a+ bracketThreads n act =+ bracket (getNumCapabilities)+ setNumCapabilities+ (\_ -> do dbgPrint 1 ("\n [Setting # capabilities to "++show n++" before test] \n")+ setNumCapabilities n+ act)++-- | Repeat a group of tests while varying the number of OS threads used. Also,+-- read configuration info.+--+-- WARNING: uses setNumCapabilities.+stdTestHarness :: (IO Test) -> IO ()+stdTestHarness genTests = do + numAgents <- getNumAgents + putStrLn$ "Running with numElems "++show numElems++" and numAgents "++ show numAgents+ putStrLn "Use NUMELEMS, NUMAGENTS, NUMTHREADS to control the size of this benchmark."+ args <- getArgs++ np <- getNumProcessors+ putStrLn $"Running on a machine with "++show np++" hardware threads."++ -- We allow the user to set this directly, because the "-t" based regexp selection+ -- of benchmarks is quite limited.+ let all_threads = case lookup "NUMTHREADS" theEnv of+ Just str -> [read str]+ Nothing -> S.toList$ S.fromList$+ [1, 2, np `quot` 2, np, 2*np ]+ putStrLn $"Running tests for these thread settings: " ++show all_threads+ all_tests <- genTests ++ -- Don't allow concurent tests (the tests are concurrent!):+ withArgs (args ++ ["-j1","--jxml=test-results.xml"]) $ do ++ -- Hack, this shouldn't be necessary, but I'm having problems with -t:+ tests <- case all_threads of+ [one] -> do cap <- getNumCapabilities+ unless (cap == one) $ setNumCapabilities one+ return all_tests+ _ -> return$ TestList [ setTestThreads n all_tests | n <- all_threads ]+ TF.defaultMain$ hUnitTestToTests tests++----------------------------------------------------------------------------------------------------+-- DEBUGGING+----------------------------------------------------------------------------------------------------++-- | Debugging flag shared by all accelerate-backend-kit modules.+-- This is activated by setting the environment variable DEBUG=1..5+dbg :: Int+dbg = case lookup "DEBUG" theEnv of+ Nothing -> defaultDbg+ Just "" -> defaultDbg+ Just "0" -> defaultDbg+ Just s ->+ trace (" ! Responding to env Var: DEBUG="++s)$+ case reads s of+ ((n,_):_) -> n+ [] -> error$"Attempt to parse DEBUG env var as Int failed: "++show s++defaultDbg :: Int+defaultDbg = 0++-- | Print if the debug level is at or above a threshold.+dbgPrint :: Int -> String -> IO ()+dbgPrint lvl str = if dbg < lvl then return () else do+-- hPutStrLn stderr str+ -- hPrintf stderr str + -- hFlush stderr+ printf str+ hFlush stdout++dbgPrintLn :: Int -> String -> IO ()+dbgPrintLn lvl str = dbgPrint lvl (str++"\n")+
+ lvish.cabal view
@@ -0,0 +1,145 @@+-- Initial lvish.cabal generated by cabal init. For further +-- documentation, see http://haskell.org/cabal/users-guide/++-- The name of the package.+name: lvish++-- The package version. See the Haskell package versioning policy (PVP) +-- for standards guiding when and how versions should be incremented.+-- http://www.haskell.org/haskellwiki/Package_versioning_policy+-- PVP summary: +-+------- breaking API changes+-- | | +----- non-breaking API additions+-- | | | +--- code changes with no API change+version: 1.0++-- Changelog:+-- 0.2 -- switch SLMap over to O(1) freeze+++synopsis: Parallel scheduler, LVar data structures, and infrastructure to build more.++description: + .+ A programming model based on monotonically-growing concurrent data structures.+ .+ As a starting point, look at "Control.LVish", as well as one of these papers:+ .+ * /LVars: lattice-based data structures for deterministic parallelism/ (<http://dl.acm.org/citation.cfm?id=2502326>).+ .+ * /Freeze after handling: quasi-deterministic programming with LVars/ (<http://www.cs.indiana.edu/~lkuper/papers/2013-lvish-draft.pdf>).++license: BSD3+license-file: LICENSE+author: Aaron Turon, Lindsey Kuper, Ryan Newton+maintainer: lindsey@composition.al+category: Concurrency+build-type: Simple++-- Constraint on the version of Cabal needed to build this package.+cabal-version: >=1.8++flag debug+ description: Activate additional debug assertions, and printed output + if DEBUGLVL env var is set to 1 or higher.+ default: False++flag chaselev+ description: Use the Chase-Lev work-stealing deque+ default: True++flag quick+ description: Build some targets but not others. Omit apps and tests.+ default: False++flag abstract-par+ description: If enabled, provide instances for generic par operations using the establish type classes.+ default: False++flag getonce+ description: Ensure that continuations of get run at most once + (by using extra synchronization)+ default: False++--------------------------------------------------------------------------------+library+ -- Modules exported by the library.+ exposed-modules:+ ------------- End user modules ------------+ Control.LVish+ Control.LVish.DeepFrz+ Data.LVar.Generic+ Data.LVar.IVar+ Data.LVar.IStructure + Data.LVar.PureSet+ Data.LVar.PureMap+ Data.LVar.SLSet+ Data.LVar.SLMap+ -------------------------------------------+ -- End users should NOT USE THESE.+ -- These are only for developing new LVars:+ Data.LVar.Internal.Pure+ Data.LVar.Generic.Internal+ Control.LVish.SchedIdempotent+ Control.LVish.Internal+ Control.LVish.DeepFrz.Internal + + -- Modules included in this library but not exported.+ other-modules:+ Data.UtilInternal+ Data.LVar.Pair + Data.Concurrent.Bag+ Data.Concurrent.Counter+ Data.Concurrent.SNZI+ Data.Concurrent.LinkedMap+ Data.Concurrent.SkipListMap+ Data.Concurrent.AlignedIORef+ Control.Reagent+ Control.LVish.SchedIdempotentInternal + Control.LVish.MonadToss+ Control.LVish.Types+ -- Not ready for prime-time yet:+ Data.LVar.NatArray+ Data.LVar.MaxCounter++ -- Other library packages from which modules are imported.+ build-depends: base ==4.6.*, deepseq ==1.3.*, containers ==0.5.*, lattices ==1.2.*, + split ==0.2.*, bytestring ==0.10.*, time ==1.4.*, rdtsc ==1.3.*, vector ==0.10.*, + parallel ==3.2.*, async ==2.0.*,+ atomic-primops, hashable, transformers, random, chaselev-deque, bits-atomic, missing-foreign,+ ghc-prim+ -- TEMP:+ build-depends: HUnit, test-framework, test-framework-hunit, test-framework-th,+ bytestring-mmap+ -- Actually -threaded won't do anything for a library, this is just a reminder:+ ghc-options: -O2 -threaded -rtsopts+ if flag(abstract-par) + cpp-options: -DUSE_ABSTRACT_PAR+ build-depends: abstract-par >=0.4+ -- monad-par-extras >=0.4+ if flag(debug)+ cpp-options: -DDEBUG_LVAR+ if flag(chaselev)+ cpp-options: -DCHASE_LEV+ if flag(getonce)+ cpp-options: -DGET_ONCE+++--------------------------------------------------------------------------------+-- TODO: New tests here:+test-suite test-lvish+ type: exitcode-stdio-1.0+ main-is: unit-tests.hs+ other-modules: TestHelpers + + ghc-options: -O2 -threaded -rtsopts -with-rtsopts=-N4+ build-depends: base ==4.6.*, containers ==0.5.*, transformers, atomic-primops, chaselev-deque, + random, deepseq, vector, bits-atomic, missing-foreign, time, ghc-prim, HUnit, test-framework, + test-framework-hunit, test-framework-th+ if flag(debug)+ cpp-options: -DDEBUG_LVAR+ if flag(chaselev)+ cpp-options: -DCHASE_LEV+ if flag(getonce)+ cpp-options: -DGET_ONCE++
+ unit-tests.hs view
@@ -0,0 +1,1180 @@+{-# LANGUAGE TemplateHaskell, CPP, ScopedTypeVariables #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE NamedFieldPuns #-}++module Main where++import Test.Framework.Providers.HUnit +import Test.Framework (Test, defaultMain, testGroup)+-- [2013.09.26] Temporarily disabling template haskell due to GHC bug discussed here:+-- https://github.com/rrnewton/haskell-lockfree/issues/10+-- import Test.Framework.TH (testGroupGenerator, defaultMainGenerator)++import Test.HUnit (Assertion, assertEqual, assertBool, Counts(..))+import qualified Test.HUnit as HU+import Control.Applicative+import Control.Monad+import Control.Concurrent+import Control.Concurrent.MVar+import GHC.Conc+import Data.List (isInfixOf, intersperse)+import qualified Data.Vector as V+import qualified Data.Set as S+import Data.IORef+import Data.Time.Clock+import System.Environment (getArgs)+import System.IO+import System.Exit+import System.Random++import Control.Exception (catch, evaluate, SomeException)++import Data.Traversable (traverse)+import qualified Data.Set as S+import qualified Data.Map as M+import Data.Word++import qualified Data.LVar.Generic as G+import qualified Data.LVar.NatArray as NA+import Data.LVar.PureSet as IS+import Data.LVar.PureMap as IM++import qualified Data.LVar.SLMap as SM+import qualified Data.LVar.SLSet as SS++import qualified Data.LVar.IVar as IV+import qualified Data.LVar.IStructure as ISt+import qualified Data.LVar.Pair as IP++import Control.LVish+import Control.LVish.DeepFrz (DeepFrz(..), Frzn, Trvrsbl, runParThenFreeze, runParThenFreezeIO)+import qualified Control.LVish.Internal as I+import Control.LVish.SchedIdempotent (liftIO, dbgLvl, forkWithExceptions)+import qualified Control.LVish.SchedIdempotent as L++import qualified Data.Concurrent.SNZI as SNZI+import qualified Data.Concurrent.LinkedMap as LM+import qualified Data.Concurrent.SkipListMap as SLM++import TestHelpers as T++--------------------------------------------------------------------------------++-- Disabling thread-variation due to below bug:+#if 1+-- EEK! Just got this [2013.06.27]:+-- +-- unit-tests.exe: internal error: wakeup_gc_threads+-- (GHC version 7.6.3 for x86_64_unknown_linux)+-- Please report this as a GHC bug: http://www.haskell.org/ghc/reportabug+-- Aborted (core dumped)++main :: IO ()+main = do+ -- T.stdTestHarness $ return all_tests -- Version that varies threads.+ if True then -- Use test-framework:+ defaultMain $ hUnitTestToTests all_tests+ else do + -- Counts{errors,failures} <- HU.runTestTT all_tests+ (Counts{errors,failures},_) <- HU.runTestText (HU.putTextToHandle stdout False) all_tests + if errors+failures == 0 then exitSuccess else exitFailure++ where + all_tests :: HU.Test+ all_tests =+ HU.TestList+ [ HU.TestLabel "case_v0" $ HU.TestCase case_v0+ , HU.TestLabel "case_v1a" $ HU.TestCase case_v1a+ , HU.TestLabel "case_v1b" $ HU.TestCase case_v1b+ , HU.TestLabel "case_v2a" $ HU.TestCase case_v2a+ , HU.TestLabel "case_v2b" $ HU.TestCase case_v2b -- livelock? [2013.09.26]+-- , HU.TestLabel "case_v3a" $ HU.TestCase case_v3a+ , HU.TestLabel "case_v3b" $ HU.TestCase case_v3b+ , HU.TestLabel "case_i3c" $ HU.TestCase case_i3c+ , HU.TestLabel "case_v3d" $ HU.TestCase case_v3d+ , HU.TestLabel "case_v3e" $ HU.TestCase case_v3e+ , HU.TestLabel "case_i3f" $ HU.TestCase case_i3f+ , HU.TestLabel "case_i3g" $ HU.TestCase case_i3g+ , HU.TestLabel "case_v7a" $ HU.TestCase case_v7a+ , HU.TestLabel "case_i7b" $ HU.TestCase case_i7b+ , HU.TestLabel "case_v7c" $ HU.TestCase case_v7c+ , HU.TestLabel "case_v8a" $ HU.TestCase case_v8a+ , HU.TestLabel "case_v8b" $ HU.TestCase case_v8b+ , HU.TestLabel "case_v8c" $ HU.TestCase case_v8c+ , HU.TestLabel "case_v8d" $ HU.TestCase case_v8d+ , HU.TestLabel "case_v9a" $ HU.TestCase case_v9a+ , HU.TestLabel "case_i9c" $ HU.TestCase case_i9c+ , HU.TestLabel "case_v9d" $ HU.TestCase case_v9d+ , HU.TestLabel "case_v9e" $ HU.TestCase case_v9e+-- , HU.TestLabel "case_v9f" $ HU.TestCase case_v9f -- [2013.09.26] RRN: problems..+-- , HU.TestLabel "case_v9g" $ HU.TestCase case_v9g -- [2013.09.26] Blocked indefinitely+ , HU.TestLabel "case_i9h" $ HU.TestCase case_i9h+ , HU.TestLabel "case_lp01" $ HU.TestCase case_lp01+ , HU.TestLabel "case_lp02" $ HU.TestCase case_lp02+ , HU.TestLabel "case_lp03" $ HU.TestCase case_lp03+ , HU.TestLabel "case_lp04" $ HU.TestCase case_lp04++ -- [2013.09.26] RRN: Disabling for now. We don't depend on them yet and they are+ -- exhibiting bugs: + -- , HU.TestLabel "case_snzi1" $ HU.TestCase case_snzi1+ -- , HU.TestLabel "case_snzi2" $ HU.TestCase case_snzi2+ -- , HU.TestLabel "case_snzi3" $ HU.TestCase case_snzi3+ -- , HU.TestLabel "case_snzi4 " $ HU.TestCase case_snzi4 + , HU.TestLabel "case_lm1" $ HU.TestCase case_lm1+ , HU.TestLabel "case_slm1" $ HU.TestCase case_slm1+ , HU.TestLabel "case_slm2" $ HU.TestCase case_slm2+ , HU.TestLabel "case_dftest0" $ HU.TestCase case_dftest0+ , HU.TestLabel "case_dftest1" $ HU.TestCase case_dftest1+ , HU.TestLabel "case_dftest3" $ HU.TestCase case_dftest3++ , HU.TestLabel "case_show01" $ HU.TestCase case_show01+ , HU.TestLabel "case_show02" $ HU.TestCase case_show02+ , HU.TestLabel "case_show03" $ HU.TestCase case_show03+ , HU.TestLabel "case_show04" $ HU.TestCase case_show04+ , HU.TestLabel "case_show05" $ HU.TestCase case_show05+ , HU.TestLabel "case_show06" $ HU.TestCase case_show06++ , HU.TestLabel "case_show05B" $ HU.TestCase case_show05B+ , HU.TestLabel "case_show06B" $ HU.TestCase case_show06B+ ]+ -- Ugh, busted test bracketing in test-framework... thus no good way to do+ -- thread-parameterization and no good way to take advantage of test-framework-th: + -- $(testGroupGenerator)+#else+-- This is what we would do if not for the atomic-primops triggered GHC linking bug:+main :: IO ()+main = $(defaultMainGenerator)+#endif++case_v0 :: HU.Assertion+case_v0 = do res <- v0+ HU.assertEqual "useless fork" (4::Int) res+v0 = runParIO $ do i <- IV.new; fork (return ()); IV.put i 4; IV.get i+++case_v1a :: Assertion+case_v1a = assertEqual "fork put" (4::Int) =<< v1a+v1a :: IO Int+v1a = runParIO $ do i<-IV.new; fork (IV.put i 4); IV.get i++case_v1b :: Assertion+case_v1b = do ls <- v1b+ case length ls of+ 0 -> return () -- Ok, i guess debugging is off.+ 1 -> return () + _ -> error $ "Wrong number of log messages: \n" ++ concat (intersperse "\n" ls)++-- | In this sequential case there should be no data-race, and thus no duplication of the callback.+v1b :: IO [String]+v1b = do let tag = "callback on ivar "+ (logs,_) <- runParLogged $ do+ i <- IV.new+ IV.put i (3::Int) + IV.whenFull Nothing i (\x -> logStrLn$ tag++show x)+ IV.put i 3+ IV.put i 3+ return ()+ mapM_ putStrLn logs+ return (filter (isInfixOf tag) logs)++-- v1c++case_v2a :: Assertion+case_v2a = v2a >>= assertEqual "put 10 in & wait"+ (S.fromList [1..10] :: S.Set Int)++-- [2013.06.27] getting thread-blocked-indefinitely errors:+v2a :: IO (S.Set Int)+v2a = runParIO $+ do s <- IS.newEmptySet+ mapM_ (\n -> fork $ IS.insert n s) [1..10]+ IS.waitSize 10 s + IS.freezeSet s++-- | This version uses a fork-join so it doesn't need the waitSize:+case_v2b :: Assertion+case_v2b = v2b >>= assertEqual "t2 with spawn instead of fork"+ (S.fromList [1..10] :: S.Set Int)+ +v2b :: IO (S.Set Int)+v2b = runParIO $+ do s <- IS.newEmptySet+ ivs <- mapM (\n -> IV.spawn_ $ IS.insert n s) [1..10]+ mapM_ IV.get ivs -- Join point.+ IS.freezeSet s++-- FIMXE:++-- | This version uses deep freeze. +case_v2c :: Assertion+case_v2c = assertEqual "t2 with spawn instead of fork"+ (S.fromList [1..10] :: S.Set Int)+ (IS.fromISet v2c)+ +-- v2c :: S.Set Int+v2c :: IS.ISet Frzn Int+v2c = -- IS.fromISet $+ runParThenFreeze par+ where+ par :: Par Det s (IS.ISet s Int)+ par = + do s <- IS.newEmptySet + ivs <- mapM (\n -> IV.spawn_ $ IS.insert n s) [1..10::Int]+ mapM_ IV.get ivs -- Join point.+ return s++-- | Simple callback test.+-- case_v3a :: Assertion+-- case_v3a = v3a >>= assertEqual "simple callback test"+-- (S.fromList [10,20,30,40,50,60,70,80,90,100] :: S.Set Int)++-- [2013.06.27] This is failing just occasionally with a multiple-put:+v3a :: IO (S.Set Int) +v3a = runParIO $+ do s1 <- IS.newEmptySet+ s2 <- IS.newEmptySet+ let fn e = IS.insert (e*10) s2+ IS.withCallbacksThenFreeze s1 fn $ do+ -- Populate the first set:+ mapM_ (\n -> fork $ IS.insert n s1) [1..10] + -- We never read out of s1 directly. Instead, writes to s1 trigger the+ -- callback 'fn' to run, with the element written to s2. So eventually,+ -- ten elements are written to s2.+ IS.waitSize 10 s2+ IS.freezeSet s2++case_v3b :: Assertion+case_v3b = v3b >>= assertEqual "simple callback test"+ (S.fromList [10,20,30,40,50,60,70,80,90,100] :: S.Set Int)+ +v3b :: IO (S.Set Int) +v3b = runParIO $+ do s1 <- IS.newEmptySet+ s2 <- IS.newEmptySet+ let fn e = IS.insert (e*10) s2+ IS.withCallbacksThenFreeze s1 fn $ do+ -- Populate the first set:+ mapM_ (\n -> IS.insert n s1) [1..10]+ -- Because we filled s1 sequentially, we know it is full at this point.+ -- (If the above were forked we would need a finish/asnyc style construct)+ + -- After all of s1's callbacks are finished executing, s2 is full:+ IS.freezeSet s2+++-- | An under-synchronized test. This should always return the same+-- result OR throw an exception. In this case it should always return+-- a list of 10 elements, or throw an exception.+case_i3c :: Assertion+case_i3c = do + allowSomeExceptions ["Attempt to change a frozen LVar"] $ + do x <- i3c+ assertEqual "under-synchronized passed through"+ (S.fromList [10,20..100] :: S.Set Int) x+ return ()+ +i3c :: IO (S.Set Int)+i3c = runParIO $+ do s1 <- IS.newEmptySet+ s2 <- IS.newEmptySet+ let fn e = IS.insert (e*10) s2+ IS.withCallbacksThenFreeze s1 fn $ do+ mapM_ (\n -> fork $ IS.insert n s1) [1..10] + IS.waitSize 1 s2 -- Not ENOUGH synchronization!+ IS.freezeSet s2+ -- If this ^ freeze occurs *before* all the puts have happened,+ -- the a put happening after it will throw an exception. If,+ -- on the other hand, it occurs after they've all happened,+ -- then we won't notice that anything is wrong and we'll get+ -- the same result we would have in case_v3.++-- FIXME: currently if run enough times, i3c can get the following failure:+-- I think we need to use full Async's so the cancellation goes both ways:++ -- Main:+ -- Exception inside child thread "worker thread", ThreadId 12: Attempt to change a frozen LVar+ -- Exception inside child thread "worker thread", ThreadId 9: Attempt to change a frozen LVar+ -- Exception inside child thread "worker thread", ThreadId 11: Attempt to change a frozen LVar+ -- test-lvish: Attempt to change a frozen LVar+ -- Exception inside child thread "worker thread", ThreadId 10: thread blocked indefinitely in an MVar operation+++case_v3d :: Assertion+case_v3d = assertEqual "test of parallelism in freezeSetAfter"+ (S.fromList [1..5]) =<< v3d++-- | This test has interdependencies between callbacks (that are launched on+-- already-present data), which forces these to be handled in parallel.+v3d :: IO (S.Set Int)+v3d = runParIO $ + do s1 <- IS.newFromList [1..5]+ s2 <- IS.newEmptySet+ IS.freezeSetAfter s1 $ \ elm -> do+ let dep = case elm of+ 1 -> Just 2+ 2 -> Just 3+ 3 -> Nothing -- Foil either left-to-right or right-to-left+ 4 -> Just 3+ 5 -> Just 4+ case dep of+ Nothing -> logStrLn $ " [Invocation "++show elm++"] has no dependencies, running... "+ Just d -> do logStrLn $ " [Invocation "++show elm++"] waiting on "++show dep+ IS.waitElem d s2+ logStrLn $ " [Invocation "++show elm++"] dependency satisfied! "+ IS.insert elm s2 + logStrLn " [freezeSetAfter completed] "+ freezeSet s2++case_v3e :: Assertion+case_v3e = assertEqual "test of parallelism in forEachHP"+ (S.fromList [1..5]) =<< v3e++-- | Same as v3d but for forEachHP+v3e :: IO (S.Set Int)+v3e = runParIO $ IS.freezeSet =<<+ do s1 <- IS.newFromList [1..5]+ s2 <- IS.newEmptySet+ hp <- newPool+ IS.forEachHP (Just hp) s1 $ \ elm -> do+ let dep = case elm of+ 1 -> Just 2+ 2 -> Just 3+ 3 -> Nothing -- Foil either left-to-right or right-to-left+ 4 -> Just 3+ 5 -> Just 4+ case dep of+ Nothing -> logStrLn $ " [Invocation "++show elm++"] has no dependencies, running... "+ Just d -> do logStrLn $ " [Invocation "++show elm++"] waiting on "++show dep+ IS.waitElem d s2+ logStrLn $ " [Invocation "++show elm++"] dependency satisfied! "+ IS.insert elm s2+ quiesce hp+ logStrLn " [quiesce completed] "+ return s2++-- RRN: Currently we have a policy where leaving the seen with running threads is+-- disallowed, but blocked ones are tolerated.+case_i3f :: Assertion+case_i3f = exceptionOrTimeOut 0.3 ["test switched off"] i3f+#ifdef NO_DANGLING_THREADS+-- | A test to make sure that we get an error when we block on an unavailable ivar.+i3f :: IO ()+i3f = runParIO$ do+ iv <- IV.new+ fork $ do IV.get iv+ logStrLn "Unblocked! Shouldn't see this."+ return ()+ return ()+#else +i3f = error "test switched off"+#endif++case_i3g :: Assertion+case_i3g = exceptionOrTimeOut 0.3 [] i3g+-- | A still-running worker thread should NOT be allowed, because it may do a put that causes an exception.+i3g :: IO Word8+i3g = runParIO$ do+ iv <- IV.new+ fork $ do let loop !ls = loop [1 .. length ls]+ loop [1..10]+ return 9+++case_v7a :: Assertion+case_v7a = assertEqual "basic imap test"+ (M.fromList [(1,1.0),(2,2.0),(3,3.0),(100,100.1),(200,201.1)]) =<<+ v7a++v7a :: IO (M.Map Int Float)+v7a = runParIO $ IM.freezeMap =<<+ do mp <- IM.newEmptyMap+ fork $ do IM.waitSize 3 mp+ IM.insert 100 100.1 mp+ fork $ do IM.waitValue 100.1 mp+ v <- IM.getKey 1 mp+ IM.insert 200 (200.1 + v) mp+ IM.insert 1 1 mp+ IM.insert 2 2 mp+ logStrLn "[v7a] Did the first two puts.."+ I.liftIO$ threadDelay 1000+ IM.insert 3 3 mp+ logStrLn "[v7a] Did the first third put."+ IM.waitSize 5 mp+ return mp++-- [2013.08.05] RRN: Observing nondeterministic blocked-indefinitely+-- exception here.+case_i7b :: Assertion+case_i7b = do + allowSomeExceptions ["Multiple puts"] $ + assertEqual "racing insert and modify"+ (M.fromList [(1,S.fromList [3.33]),+ (2,S.fromList [0.11,4.44])]) =<<+ i7b+ return ()++-- | A quasi-deterministic example.+i7b :: IO (M.Map Int (S.Set Float))+-- Do we need a "deep freeze" that freezes nested structures?+i7b = runParIO $ do+ mp <- IM.newEmptyMap+ s1 <- IS.newEmptySet+ s2 <- IS.newEmptySet+ IS.insert 0.11 s2+ f1 <- IV.spawn_ $ do IM.insert 1 s1 mp + IM.insert 2 s2 mp+ f2 <- IV.spawn_ $ do s <- IM.getKey 1 mp+ IS.insert 3.33 s+ -- RACE: this modify is racing with the insert of s2:+ IM.modify mp 2 IS.newEmptySet (IS.insert 4.44) ++ IV.get f1; IV.get f2+ mp2 <- IM.freezeMap mp+ traverse IS.freezeSet mp2++case_v7c :: Assertion+case_v7c = assertEqual "imap test - racing modifies"+ (M.fromList [(1,S.fromList [3.33]),+ (2,S.fromList [4.44]),+ (3,S.fromList [5.55,6.6])]) =<<+ v7c++-- | This example is valid because two modifies may race.+v7c :: IO (M.Map Int (S.Set Float))+-- Do we need a "deep freeze" that freezes nested structures?+v7c = runParIO $ do+ mp <- IM.newEmptyMap+ s1 <- IS.newEmptySet+ f1 <- IV.spawn_ $ IM.insert 1 s1 mp + f2 <- IV.spawn_ $ do s <- IM.getKey 1 mp+ IS.insert 3.33 s+ IM.modify mp 2 IS.newEmptySet (IS.insert 4.44)+ f3 <- IV.spawn_ $ IM.modify mp 3 IS.newEmptySet (IS.insert 5.55)+ f4 <- IV.spawn_ $ IM.modify mp 3 IS.newEmptySet (IS.insert 6.6)+ -- No easy way to wait on the total size of all contained sets...+ -- + -- Need a barrier here.. should have a monad-transformer that provides cilk "sync"+ -- Global quiesce is convenient too..+ IV.get f1; IV.get f2; IV.get f3; IV.get f4+ mp2 <- IM.freezeMap mp+ traverse IS.freezeSet mp2++--------------------------------------------------------------------------------+-- Higher level derived ops+-------------------------------------------------------------------------------- ++case_v8a :: Assertion+case_v8a = assertEqual "simple cartesian product test"+ (S.fromList+ [(1,'a'),(1,'b'),(1,'c'),+ (2,'a'),(2,'b'),(2,'c'),+ (3,'a'),(3,'b'),(3,'c')])+ =<< v8a++-- v8a :: IO (S.Set (Integer, Char))+v8a :: IO (S.Set (Integer, Char))+v8a = runParIO $ do+ s1 <- IS.newFromList [1,2,3]+ s2 <- IS.newFromList ['a','b']+ logStrLn " [v8a] now to construct cartesian product..."+ h <- newPool+ s3 <- IS.cartesianProdHP (Just h) s1 s2+ logStrLn " [v8a] cartesianProd call finished... next quiesce"+ IS.forEach s3 $ \ elm ->+ logStrLn$ " [v8a] Got element: "++show elm+ IS.insert 'c' s2+ quiesce h+ logStrLn " [v8a] quiesce finished, next freeze::"+ freezeSet s3++case_v8b :: Assertion+case_v8b = assertEqual "3-way cartesian product"+ (S.fromList+ [[1,40,101],[1,40,102], [1,50,101],[1,50,102],+ [2,40,101],[2,40,102], [2,50,101],[2,50,102]]+ )+ =<< v8b++v8b :: IO (S.Set [Int])+v8b = runParIO $ do+ hp <- newPool+ s1 <- IS.newFromList [1,2]+ s2 <- IS.newFromList [40,50]+ -- (hp,s3) <- IS.traverseSetHP Nothing (return . (+100)) s1+ s3 <- IS.traverseSetHP (Just hp) (return . (+100)) s1+ s4 <- IS.cartesianProdsHP (Just hp) [s1,s2,s3]+ IS.forEachHP (Just hp) s4 $ \ elm ->+ logStrLn $ " [v8b] Got element: "++show elm+ -- [2013.07.03] Confirmed: this makes the bug(s) go away: + -- liftIO$ threadDelay$ 100*1000+ quiesce hp+ logStrLn " [v8b] quiesce finished, next freeze::"+ freezeSet s4++case_v8c :: Assertion+case_v8c = assertEqual "forEachHP on maps"+ (M.fromList [(1,101),(2,102)] ) =<< v8c++-- | Similar test with Maps instead of Sets.+v8c :: IO (M.Map Int Int)+v8c = runParIO $ do+ hp <- newPool+ m1 <- IM.newFromList [(1,1),(2,2)]+ m2 <- newEmptyMap+ let cb k v = do logStrLn$" [v8c] Inside callback for Map.. key="++show k+ IM.insert k (v+100) m2+ IM.forEachHP (Just hp) m1 cb + logStrLn " [v8c] Everything set up; about to quiesce..."+ quiesce hp+ logStrLn " [v8c] quiesce finished, next freeze:"+ freezeMap m2+++case_v8d :: Assertion+case_v8d = assertEqual "union on maps"+ (M.fromList [(1,101),(2,102),(40,40),(50,50)] )+ =<< v8d+v8d :: IO (M.Map Int Int)+v8d = runParIO $ do+ hp <- newPool+ logStrLn " [v8d] Got a new pool..." + m1 <- IM.newFromList [(1,1),(2,2)]+ m2 <- IM.newFromList [(40,40),(50,50)]+ logStrLn " [v8d] Got two fresh maps..."+ let cb k v = do logStrLn$" [v8d] Inside callback for traverse.. key="++show k+ return (v+100)+ m3 <- IM.traverseMapHP (Just hp) cb m1+ m4 <- IM.unionHP (Just hp) m2 m3+ IM.forEachHP (Just hp) m4 $ \ k elm ->+ logStrLn $ " [v8d] Got element: "++show (k,elm)+ logStrLn " [v8d] Everything set up; about to quiesce..."+ quiesce hp+-- quiesceAll + logStrLn " [v8d] quiesce finished, next freeze::"+ freezeMap m4++--------------------------------------------------------------------------------+-- NatArrays+--------------------------------------------------------------------------------++case_v9a :: Assertion+case_v9a = assertEqual "basic NatArray" 4 =<< v9a+v9a :: IO Word8+v9a = runParIO$ do+ arr <- NA.newNatArray 10+ NA.put arr 5 (4::Word8)+ NA.get arr 5+++-- #ifdef NO_DANGLING_THREADS+-- case_i9b :: Assertion+-- case_i9b = exceptionOrTimeOut 0.3 [] i9b+-- -- | A test to make sure that we get an error when we should.+-- i9b :: IO Word8+-- i9b = runParIO$ do+-- arr:: NA.NatArray s Word8 <- NA.newNatArray 10 +-- fork $ do NA.get arr 5+-- logStrLn "Unblocked! Shouldn't see this."+-- return ()+-- return 9+-- #endif++case_i9c :: Assertion+case_i9c = exceptionOrTimeOut 0.3 ["thread blocked indefinitely"] i9c+i9c :: IO Word8+i9c = runParIO$ do+ arr:: NA.NatArray s Word8 <- NA.newNatArray 10 + fork $ do NA.get arr 5+ logStrLn "Unblocked! Shouldn't see this."+ NA.put arr 6 99+ NA.get arr 6 ++case_v9d :: Assertion+case_v9d = assertEqual "NatArray blocking/unblocking" 99 =<< v9d+v9d :: IO Word8+v9d = runParIO$ do+ arr:: NA.NatArray s Word8 <- NA.newNatArray 10 + fork $ do NA.get arr 5+ logStrLn "Unblocked! Good."+ NA.put arr 6 99+ logStrLn "After fork."+ NA.put arr 5 5+ NA.get arr 6 ++-- WARNING: I'm seeing some livelocks here that depend on the number of threads+-- (e.g. at -N4 but not -N2). When deadlocked on -N4 it burns 250% cpu.+-- +-- [2013.08.05] Update... it can pass 100 iterations at -N4 BY ITSELF,+-- but fails much more rapidly when run together with other 'v9'+-- tests.+case_v9e :: Assertion++case_v9e = assertEqual "Scale up a bit" 5000050000 =<< v9e+v9e :: IO Word64+v9e = runParIO$ do+ let size = 100000+ arr <- NA.newNatArray size+ fork $+ forM_ [0..size-1] $ \ix ->+ NA.put arr ix (fromIntegral ix + 1) -- Can't put 0+ logStrLn "After fork."+ let loop !acc ix | ix == size = return acc+ | otherwise = do v <- NA.get arr ix+ loop (acc+v) (ix+1)+ loop 0 0+-- NOTE: this test takes about 0.03 seconds.+-- It is not faster with two threads, alas... but it is higher variance!++-- | Here's the same test with an actual array of IVars.+-- This one is reliable, but takes about 0.20-0.30 seconds.+case_v9f :: Assertion+-- [2013.08.05] RRN: Actually I'm seeing the same non-deterministic+-- thread-blocked-indefinitely problem here.+case_v9f = assertEqual "Array of ivars, compare effficiency:" 5000050000 =<< v9f+v9f :: IO Word64+v9f = runParIO$ do+ let size = 100000+ news = V.replicate size IV.new+ arr <- V.sequence news+ fork $+ forM_ [0..size-1] $ \ix ->+ IV.put_ (arr V.! ix) (fromIntegral ix + 1)+ logStrLn "After fork."+ let loop !acc ix | ix == size = return acc+ | otherwise = do v <- IV.get (arr V.! ix)+ loop (acc+v) (ix+1)+ loop 0 0++-- | One more time with a full IStructure.+case_v9g :: Assertion+case_v9g = assertEqual "IStructure, compare effficiency:" 5000050000 =<< v9g+v9g :: IO Word64+v9g = runParIO$ do+ let size = 100000+ arr <- ISt.newIStructure size + fork $+ forM_ [0..size-1] $ \ix ->+ ISt.put_ arr ix (fromIntegral ix + 1)+ logStrLn "After fork."+ let loop !acc ix | ix == size = return acc+ | otherwise = do v <- ISt.get arr ix+ loop (acc+v) (ix+1)+ loop 0 0+++-- Uh oh, this is blocking indefinitely sometimes...+-- BUT, only when I run the whole test suite.. via cabal install --enable-tests+case_i9h :: Assertion+case_i9h = exceptionOrTimeOut 0.3 ["Attempt to put zero"] i9i+i9i :: IO Word+i9i = runParIO$ do+ arr <- NA.newNatArray 1+ NA.put arr 0 0+ NA.get arr 0++--------------------------------------------------------------------------------+-- Looping constructs+--------------------------------------------------------------------------------++case_lp01 :: Assertion+case_lp01 = assertEqual "parForSimple test" "done" =<< lp01+lp01 = runParIO$ do+ logStrLn " [lp01] Starting parForSimple loop..."+ x <- IV.new + parForSimple (0,10) $ \ ix -> do+ logStrLn$ " [lp01] iter "++show ix+ when (ix == 9)$ IV.put x "done"+ IV.get x++case_lp02 :: Assertion+case_lp02 = assertEqual "parForL test" "done" =<< lp02+lp02 = runParIO$ do+ logStrLn " [lp02] Starting parForL loop..."+ x <- IV.new + parForL (0,10) $ \ ix -> do+ logStrLn$ " [lp02] iter "++show ix+ when (ix == 9)$ IV.put x "done"+ logStrLn$ " [lp02] after loop..."+ IV.get x++-- [2013.08.05] RRN: I'm seeing this hang sometimes. It live-locks+-- burning CPU. (But only 170% CPU with -N4.) Hmm, I can't get it to+-- freeze running BY ITSELF, however. In fact I can't get the problem+-- while running just the "lp" tests. I can get the problem running+-- just 'v' tests and even just 'v9' tests.+case_lp03 :: Assertion+case_lp03 = assertEqual "parForTree test" "done" =<< lp03+lp03 = runParIO$ do+ logStrLn " [lp03] Starting parForTree loop..."+ x <- IV.new + parForTree (0,10) $ \ ix -> do+ logStrLn$ " [lp03] iter "++show ix+ when (ix == 9)$ IV.put x "done"+ logStrLn$ " [lp03] after loop..."+ IV.get x++case_lp04 :: Assertion+case_lp04 = assertEqual "parForTree test" "done" =<< lp04+lp04 = runParIO$ do+ logStrLn " [lp04] Starting parForTiled loop..."+ x <- IV.new + parForTiled 16 (0,10) $ \ ix -> do+ logStrLn$ " [lp04] iter "++show ix+ when (ix == 9)$ IV.put x "done"+ logStrLn$ " [lp04] after loop..."+ IV.get x+++--------------------------------------------------------------------------------+-- TESTS FOR SNZI +--------------------------------------------------------------------------------+ +-- | Test snzi in a sequential setting+snzi1 :: IO (Bool)+snzi1 = do+ (cs, poll) <- SNZI.newSNZI+ forM_ cs SNZI.arrive + forM_ cs SNZI.arrive+ forM_ cs SNZI.depart + forM_ cs SNZI.depart+ poll+ +case_snzi1 :: Assertion +case_snzi1 = snzi1 >>= assertEqual "sequential use of SNZI" True++-- | Very simple sequential snzi test+snzi2a :: IO (Bool)+snzi2a = do+ (cs, poll) <- SNZI.newSNZI+ forM_ cs SNZI.arrive + poll+ +case_snzi2a :: Assertion +case_snzi2a = snzi2a >>= assertEqual "sequential use of SNZI" False++-- | Test snzi in a sequential setting+snzi2 :: IO (Bool)+snzi2 = do+ (cs, poll) <- SNZI.newSNZI+ forM_ cs SNZI.arrive + forM_ cs SNZI.arrive+ forM_ cs SNZI.depart + forM_ cs SNZI.depart+ forM_ cs SNZI.arrive+ poll+ +case_snzi2 :: Assertion +case_snzi2 = snzi2 >>= assertEqual "sequential use of SNZI" False++-- | Test snzi in a concurrent setting+snzi3 :: IO (Bool)+snzi3 = do+ (cs, poll) <- SNZI.newSNZI+ mvars <- forM cs $ \c -> do+ mv <- newEmptyMVar+ forkWithExceptions forkIO "snzi3 test thread" $ do + nTimes 1000000 $ \_ -> do+ SNZI.arrive c+ SNZI.depart c+ SNZI.arrive c+ SNZI.arrive c+ SNZI.depart c+ SNZI.depart c+ putMVar mv ()+ return mv+ forM_ mvars takeMVar+ poll+ +case_snzi3 :: Assertion +case_snzi3 = snzi3 >>= assertEqual "concurrent use of SNZI" True++-- | Test snzi in a concurrent setting+snzi4 :: IO (Bool)+snzi4 = do+ (cs, poll) <- SNZI.newSNZI+ mvars <- forM cs $ \c -> do+ mv <- newEmptyMVar+ internalMV <- newEmptyMVar+ forkWithExceptions forkIO "snzi4 test thread type A" $ do + SNZI.arrive c+ putMVar internalMV ()+ forkWithExceptions forkIO "snzi4 test thread type B" $ do + nTimes 1000000 $ \_ -> do+ SNZI.arrive c+ SNZI.depart c+ SNZI.arrive c+ SNZI.arrive c+ SNZI.depart c+ SNZI.depart c+ takeMVar internalMV+ putMVar mv ()+ return mv+ forM_ mvars takeMVar+ poll+ +case_snzi4 :: Assertion +case_snzi4 = snzi4 >>= assertEqual "concurrent use of SNZI" False++--------------------------------------------------------------------------------+-- TESTS FOR SKIPLIST+--------------------------------------------------------------------------------++lm1 :: IO (String)+lm1 = do+ lm <- LM.newLMap+ LM.NotFound tok <- LM.find lm 1+ LM.tryInsert tok "Hello"+ LM.NotFound tok <- LM.find lm 0+ LM.tryInsert tok " World"+ LM.Found s1 <- LM.find lm 1+ LM.Found s0 <- LM.find lm 0+ return $ s1 ++ s0+ +case_lm1 :: Assertion +case_lm1 = lm1 >>= assertEqual "test sequential insertion for LinkedMap" "Hello World"++slm1 :: IO (String)+slm1 = do+ slm <- SLM.newSLMap 5+ SLM.putIfAbsent slm 0 $ return "Hello "+ SLM.putIfAbsent slm 1 $ return "World"+ Just s0 <- SLM.find slm 0+ Just s1 <- SLM.find slm 1+ return $ s0 ++ s1+ +case_slm1 :: Assertion +case_slm1 = slm1 >>= assertEqual "test sequential insertion for SkipListMap" "Hello World" ++slm2 :: IO Bool+slm2 = do+ slm <- SLM.newSLMap 10+ mvars <- replicateM numCapabilities $ do+ mv <- newEmptyMVar+ forkWithExceptions forkIO "slm2 test thread" $ do+ rgen <- newIORef $ mkStdGen 0+ let flip = do+ g <- readIORef rgen+ let (b, g') = random g+ writeIORef rgen $! g'+ return b+ nTimes 10000 $ \n -> SLM.putIfAbsentToss slm n (return n) flip+ putMVar mv ()+ return mv + forM_ mvars takeMVar+ -- cs <- SLM.counts slm+ -- putStrLn $ show cs+ SLM.foldlWithKey (\b k v -> if k == v then return b else return False) True slm+-- Just n <- SLM.find slm (slm2Count/2) -- test find function+-- return n+ +case_slm2 :: Assertion +case_slm2 = slm2 >>= assertEqual "test concurrent insertion for SkipListMap" True++--------------------------------------------------------------------------------+-- TEMPLATE HASKELL BUG? -- if we have *block* commented case_foo decls, it detects+-- those when it shouldn't:+--------------------------------------------------------------------------------++-- -- | Simple test of pairs.+-- case_v4 :: Assertion+-- case_v4 = v4 >>= assertEqual "simple-pair" (3, "hi") ++-- v4 :: IO (Int,String)+-- v4 = runParIO $+-- do p <- newPair+-- putFst p 3+-- putSnd p "hi" +-- x <- getFst p+-- y <- getSnd p+-- return (x,y)++-- -- | This program should throw an exception due to multiple puts.+-- case_i5a :: Assertion+-- case_i5a = assertException ["Multiple puts to an IVar!"] i5a++-- i5a :: IO Int+-- i5a = runParIO (+-- do p <- newPair+-- putFst p 3+-- putSnd p "hi"+-- putSnd p "there" +-- getFst p)++-- -- | Another exception due to multiple puts. This tests whether the scheduler waits+-- -- around for a trailing (errorful) computation that is not on the main thread.+-- case_i5b :: Assertion+-- case_i5b = assertException ["Multiple puts to an IVar!"] i5b++-- i5b = +-- runParIO $+-- do p <- newPair+-- putFst p 3+-- putSnd p "hi"+-- fork $ do waste_time+-- putSnd p "there"+-- -- There's no 'consume' here; so we should really just get a+-- -- "Multiple puts to an IVar!" exception.+-- getSnd p++-- -- | Similar to 5b but with the branches flipped.+-- case_i5c :: Assertion+-- case_i5c = assertException ["Multiple puts to an IVar!"] i5c++-- i5c = runParIO $+-- do p <- newPair+-- putSnd p "hi"++-- -- The forked thread's value is not returned, so we go to a little extra work+-- -- here to bounce the value through the First of the pair.+-- fork $ putFst p =<< getSnd p+-- waste_time+ +-- putSnd p "there"+-- getFst p++-- -- | Another multiple put error. This one makes sure that ANY tops get thrown as+-- -- exceptions, or we have full nondeterminism (not even limited guarantees), the+-- -- program would return "a" or "b".+-- case_i6a :: Assertion+-- case_i6a = assertException ["Multiple puts to an IVar!"] i6a+-- i6a = runParIO (+-- do p <- newPair+-- putFst p 3++-- -- TODO: Randomize these amounts of time:+-- fork $ do waste_time+-- putSnd p "a"+-- fork $ do waste_time+-- putSnd p "b"+-- -- There's no 'consume' here; so we should really just get a+-- -- "Multiple puts to an IVar!" exception.+-- getSnd p)+++-- -- TODO:+-- --------------------------------+-- -- | This test, semantically, has two possible outcomes. It can return "hi" or an+-- -- error. That's quasi-determinism. In practice, we force it to have one outcome by+-- -- wasting a significant amount of time in one branch.+-- --------------------------------+++-- waste_time = loop 1000 3.3+-- where+-- loop 0 acc = if acc < 10 then return acc else return 0+-- loop i !acc = loop (i - 1) (sin acc + 1.0)++-- -- More pairs+-- case_v6 :: Assertion+-- case_v6 = assertEqual "fancy pairs"+-- 33 =<< runParIO (+-- do p1 <- newPair+-- p2 <- newPair+-- fork $ do x <- getFst p1+-- putSnd p2 x +-- fork $ do x <- getSnd p2+-- putSnd p1 x+-- putFst p1 33+-- getSnd p1)+++--------------------------------------------------------------------------------+-- Freeze-related tests:+--------------------------------------------------------------------------------++case_dftest0 = assertEqual "manual freeze, outer layer" "hello" =<< dftest0++dftest0 :: IO String+dftest0 = runParIO $ do+ iv1 <- IV.new+ iv2 <- IV.new+ IV.put_ iv1 iv2+ IV.put_ iv2 "hello"+ m <- IV.freezeIVar iv1+ case m of+ Just i -> IV.get i++case_dftest1 = assertEqual "deefreeze double ivar" (Just "hello") =<< dftest1++-- | Should return (Just (Just "hello"))+dftest1 :: IO (Maybe String)+dftest1 = runParIO $ do+ iv1 <- IV.new+ iv2 <- IV.new+ IV.put_ iv1 iv2+ IV.put_ iv2 "hello"+ Just x <- IV.freezeIVar iv1+ IV.freezeIVar x++case_dftest3 = assertEqual "freeze simple ivar" (Just 3) =<< dftest3+dftest3 :: IO (Maybe Int)+dftest3 = runParIO $ do+ iv1 <- IV.new+ IV.put_ iv1 (3::Int)+ IV.freezeIVar iv1 +++--FIXME:++-- -- | Polymorphic version of previous. DeepFrz is more flexible than regular+-- -- freeze, because we can pick multiple return types for the same code. But we must+-- -- be very careful with this kind of thing due to the 's' type variables.+-- dftest4_ :: DeepFrz (IV.IVar s1 Int) =>+-- Par QuasiDet s1 b+-- dftest4_ = do+-- iv1 <- newBottom +-- IV.put_ iv1 (3::Int)+-- res <- IV.freezeIVar iv1 +-- return res++-- case_dftest4a = assertEqual "freeze polymorphic 1" (Just 3) =<< dftest4a+-- dftest4a :: IO (Maybe Int)+-- dftest4a = runParIO dftest4_++------------------------------------------------------------------------------------------+-- Show instances+------------------------------------------------------------------------------------------++case_show01 :: Assertion+case_show01 = assertEqual "show for IVar" "Just 3" show01+show01 :: String+show01 = show$ runParThenFreeze $ do v <- IV.new; IV.put v (3::Int); return v++-- | It happens that these come out in the opposite order from the Pure one:+case_show02 :: Assertion+case_show02 = assertEqual "show for SLMap" "{IMap: (\"key2\",44), (\"key1\",33)}" show02+show02 :: String+show02 = show$ runParThenFreeze $ do+ mp <- SM.newEmptyMap+ SM.insert "key1" (33::Int) mp+ SM.insert "key2" (44::Int) mp + return mp++case_show03 :: Assertion+case_show03 = assertEqual "show for PureMap" "{IMap: (\"key1\",33), (\"key2\",44)}" show03+show03 :: String+show03 = show$ runParThenFreeze $ do+ mp <- IM.newEmptyMap+ IM.insert "key1" (33::Int) mp+ IM.insert "key2" (44::Int) mp + return mp++case_show04 :: Assertion+case_show04 = assertEqual "show for IStructure" "{IStructure: Just 33, Just 44}" show04+show04 :: String+show04 = show$ runParThenFreeze $ do+ ist <- ISt.newIStructure 2+ ISt.put ist 0 (33::Int)+ ISt.put ist 1 (44::Int)+ return ist++case_show05 :: Assertion+case_show05 = assertEqual "show for PureSet" "{ISet: 33, 44}" (show show05)+show05 :: ISet Frzn Int+show05 = runParThenFreeze $ do+ is <- IS.newEmptySet+ IS.insert (33::Int) is+ IS.insert (44::Int) is+ return is++-- | It happens that these come out in the opposite order from the Pure one:+case_show06 :: Assertion+case_show06 = assertEqual "show for SLSet" "{ISet: 44, 33}" (show show06)+show06 :: SS.ISet Frzn Int+show06 = runParThenFreeze $ do+ is <- SS.newEmptySet+ SS.insert (33::Int) is+ SS.insert (44::Int) is+ return is++----------------------------------------+-- Test sortFrzn instances:++case_show05B :: Assertion+case_show05B = assertEqual "show for PureSet/Trvrsbl" "AFoldable [33, 44]" (show show05B)+show05B :: G.AFoldable Int+show05B = G.sortFrzn show05++case_show06B :: Assertion+case_show06B = assertEqual "show for SLSet/Trvrsbl" "AFoldable [44, 33]" (show show06B)+show06B :: G.AFoldable Int+show06B = G.sortFrzn show06++------------------------------------------------------------------------------------------+-- Misc Helpers+------------------------------------------------------------------------------------------++-- | Ensure that executing an action returns an exception+-- containing one of the expected messages.+assertException :: [String] -> IO a -> IO ()+assertException msgs action = do+ x <- catch (do action; return Nothing) + (\e -> do putStrLn $ "Good. Caught exception: " ++ show (e :: SomeException)+ return (Just $ show e))+ case x of + Nothing -> error "Failed to get an exception!"+ Just s -> + if any (`isInfixOf` s) msgs+ then return () + else error $ "Got the wrong exception, expected one of the strings: "++ show msgs+ ++ "\nInstead got this exception:\n " ++ show s++-- | For testing quasi-deterministic programs: programs that always+-- either raise a particular exception or produce a particular answer.+allowSomeExceptions :: [String] -> IO a -> IO (Either SomeException a)+allowSomeExceptions msgs action = do+ catch (do a <- action; evaluate a; return (Right a))+ (\e ->+ let estr = show e in+ if any (`isInfixOf` estr) msgs+ then do when (dbgLvl>=1) $+ putStrLn $ "Caught allowed exception: " ++ show (e :: SomeException)+ return (Left e)+ else error $ "Got the wrong exception, expected one of the strings: "++ show msgs+ ++ "\nInstead got this exception:\n " ++ show estr)++exceptionOrTimeOut :: Double -> [String] -> IO a -> IO ()+exceptionOrTimeOut time msgs action = do+ x <- timeOut time $+ allowSomeExceptions msgs action+ case x of+ Just (Right _val) -> error "exceptionOrTimeOut: action returned successfully!" + Just (Left _exn) -> return () -- Error, yay!+ Nothing -> return () -- Timeout.++-- | Time-out an IO action by running it on a separate thread, which is killed when+-- the timer expires. This requires that the action do allocation, otherwise it will+-- be non-preemptable.+timeOut :: Double -> IO a -> IO (Maybe a)+timeOut interval act = do+ result <- newIORef Nothing+ tid <- forkIO (act >>= writeIORef result . Just)+ t0 <- getCurrentTime+ let loop = do+ stat <- threadStatus tid+ case stat of+ ThreadFinished -> readIORef result+ ThreadBlocked _ -> return Nothing+ ThreadDied -> return Nothing+ ThreadRunning -> do + now <- getCurrentTime+ let delt :: Double+ delt = fromRational$ toRational$ diffUTCTime now t0+ if delt >= interval+ then do killThread tid -- TODO: should probably wait for it to show up as dead.+ return Nothing+ else do threadDelay (10 * 1000)+ loop + loop+ +assertOr :: Assertion -> Assertion -> Assertion+assertOr act1 act2 = + catch act1 + (\(e::SomeException) -> act2)++nTimes :: Int -> (Int -> IO a) -> IO ()+nTimes 0 _ = return ()+nTimes n c = c n >> nTimes (n-1) c