reactor (empty) → 0.1.3
raw patch · 11 files changed
+761/−0 lines, 11 filesdep +arraydep +basedep +bits-atomicsetup-changed
Dependencies added: array, base, bits-atomic, comonad, contravariant, mtl, semigroupoids, transformers
Files
- LICENSE +30/−0
- Reactor/Atomic.hs +78/−0
- Reactor/Deque.hs +194/−0
- Reactor/Filtered.hs +14/−0
- Reactor/Moore.hs +34/−0
- Reactor/Observable.hs +216/−0
- Reactor/Observer.hs +38/−0
- Reactor/Subscription.hs +17/−0
- Reactor/Task.hs +96/−0
- Setup.lhs +7/−0
- reactor.cabal +37/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright 2011 Edward Kmett++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions+are met:++1. Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++2. 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.++3. Neither the name of the author nor the names of his contributors+ may be used to endorse or promote products derived from this software+ without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE AUTHORS ``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 AUTHORS 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.
+ Reactor/Atomic.hs view
@@ -0,0 +1,78 @@+{-# LANGUAGE DeriveDataTypeable #-}+module Reactor.Atomic where++import Control.Monad+import Control.Monad.IO.Class+import Data.Bits.Atomic+import Data.Data+import Foreign.Ptr+import Foreign.ForeignPtr+import Foreign.Storable+import System.IO.Unsafe++newtype Atomic a = Atomic (ForeignPtr a)+ deriving (Data, Typeable)++instance (Show a, Storable a) => Show (Atomic a) where+ showsPrec d (Atomic fp) = showsPrec d $ unsafePerformIO $ withForeignPtr fp peek++atomic :: (MonadIO m, AtomicBits a, Storable a) => a -> m (Atomic a)+atomic a = liftIO $ do+ fp <- mallocForeignPtr+ withForeignPtr fp $ \p -> poke p a+ return $ Atomic fp ++withAtomic :: MonadIO m => Atomic a -> (Ptr a -> IO b) -> m b+withAtomic (Atomic fp) = liftIO . withForeignPtr fp +{-# INLINE withAtomic #-}++atomicFetchAndAdd :: (MonadIO m, AtomicBits a) => Atomic a -> a -> m a+atomicFetchAndAdd fp a = withAtomic fp $ \p -> fetchAndAdd p a+{-# INLINE atomicFetchAndAdd #-}++atomicFetchAndAnd :: (MonadIO m, AtomicBits a) => Atomic a -> a -> m a+atomicFetchAndAnd fp a = withAtomic fp $ \p -> fetchAndAnd p a+{-# INLINE atomicFetchAndAnd #-}+ +atomicFetch :: (MonadIO m, AtomicBits a) => Atomic a -> m a +atomicFetch fp = atomicFetchAndAdd fp 0+{-# INLINE atomicFetch #-}++atomicFetchAndSub :: (MonadIO m, AtomicBits a) => Atomic a -> a -> m a+atomicFetchAndSub fp a = withAtomic fp $ \p -> fetchAndSub p a+{-# INLINE atomicFetchAndSub #-}++atomicSubAndFetch :: (MonadIO m, AtomicBits a) => Atomic a -> a -> m a+atomicSubAndFetch fp a = withAtomic fp $ \p -> subAndFetch p a+{-# INLINE atomicSubAndFetch #-}++atomicAddAndFetch :: (MonadIO m, AtomicBits a) => Atomic a -> a -> m a+atomicAddAndFetch fp a = withAtomic fp $ \p -> subAndFetch p a+{-# INLINE atomicAddAndFetch #-}++atomicCompareAndSwapBool :: (MonadIO m, AtomicBits a) => Atomic a -> a -> a -> m Bool+atomicCompareAndSwapBool fp old new = withAtomic fp $ \p -> compareAndSwapBool p old new+{-# INLINE atomicCompareAndSwapBool #-}++atomicCompareAndSwap :: (MonadIO m, AtomicBits a) => Atomic a -> a -> a -> m a+atomicCompareAndSwap fp old new = withAtomic fp $ \p -> compareAndSwap p old new+{-# INLINE atomicCompareAndSwap #-}++atomicLockTestAndSet :: (MonadIO m, AtomicBits a) => Atomic a -> m a+atomicLockTestAndSet fp = withAtomic fp lockTestAndSet+{-# INLINE atomicLockTestAndSet #-}++atomicLockRelease :: (MonadIO m, AtomicBits a) => Atomic a -> m ()+atomicLockRelease fp = withAtomic fp lockRelease+{-# INLINE atomicLockRelease #-}++given :: (MonadIO m, AtomicBits a) => Atomic a -> m () -> m ()+given flag task = do+ i <- atomicFetch flag+ when (i /= 0) task++clearing :: (MonadIO m, AtomicBits a) => Atomic a -> m () -> m ()+clearing flag task = do+ i <- atomicFetchAndAnd flag 0+ when (i /= 0) task+
+ Reactor/Deque.hs view
@@ -0,0 +1,194 @@+{-# LANGUAGE UndecidableInstances, FlexibleContexts, DeriveDataTypeable #-}++module Reactor.Deque (+ Deque ++ -- * Local stack operations+ , empty -- :: (MonadIO m, MArray a e IO) => IO (Deque a e)+ , push -- :: (MonadIO m, MArray a e IO) => e -> Deque a e -> IO ()+ , pop -- :: (MonadIO m, MArray a e IO) => Deque a e -> IO (Maybe e)++ -- * Performance tuning+ , withCapacity -- :: (MonadIO m, MArray a e IO) => Int -> IO (Deque a e)+ , minimumCapacity -- :: Int+ , defaultCapacity -- :: Int++ -- * Work stealing+ , steal -- :: (MonadIO m, MArray a e IO) => Deque a e -> IO (Stolen e)+ , Stolen(..) + ) where++-- | For an explanation of the implementation, see \"Dynamic Circular Work-Stealing Deque\" +-- by David Chase and Yossi Lev of Sun Microsystems.++import Prelude hiding (read)+import Control.Applicative hiding (empty)+import Data.Bits.Atomic+import Foreign.Ptr+import Foreign.ForeignPtr+import Foreign.Storable+import Data.IORef+import Data.Array.MArray+import Control.Monad+import Control.Monad.IO.Class+import Data.Data+import System.IO.Unsafe++data Buffer a e = Buffer {-# UNPACK #-} !Int !(a Int e) ++instance Typeable2 a => Typeable1 (Buffer a) where+ typeOf1 tae = mkTyConApp bufferTyCon [typeOf1 (aInte tae)]+ where aInte :: t a e -> a Int e+ aInte = undefined++bufferTyCon :: TyCon+bufferTyCon = mkTyCon "Reactor.Deque.Buffer"++size :: Buffer a e -> Int+size (Buffer i _) = i++data Deque a e = Deque + { _tb :: ForeignPtr Int + , _content :: IORef (Buffer a e)+ }++instance (MArray a e IO, Show e) => Show (Deque a e) where+ showsPrec d (Deque tb content) = unsafePerformIO $ do+ (t,b) <- withForeignPtr tb $ \p -> (,) <$> peekTop p <*> peekBottom p + buffer <- readIORef content+ contents <- forM [t..b-1] (read buffer)+ return $ showParen (d > 10) $ + showString "Deque (ptr " . showsPrec 11 t . showChar ' ' . showsPrec 11 b . showString ") (buffer " . showsPrec 11 contents . showChar ')'++instance Typeable2 a => Typeable1 (Deque a) where+ typeOf1 dae = mkTyConApp dequeTyCon [typeOf1 (aInte dae)]+ where aInte :: t a e -> a Int e+ aInte = undefined++dequeTyCon :: TyCon+dequeTyCon = mkTyCon "Reactor.Deque.Deque"++ptr :: Storable a => a -> a -> IO (ForeignPtr a)+ptr a b = do+ p <- mallocForeignPtrArray 2+ withForeignPtr p $ \q -> do + poke q a+ pokeElemOff q 1 b+ return p++minimumCapacity :: Int+minimumCapacity = 16++defaultCapacity :: Int+defaultCapacity = 32++bufferWithCapacity :: MArray a e IO => Int -> IO (Buffer a e)+bufferWithCapacity i = + Buffer i <$> newArray_ (0, (minimumCapacity `max` i) - 1)++withCapacity :: (MonadIO m, MArray a e IO) => Int -> m (Deque a e)+withCapacity i = liftIO (Deque <$> ptr 0 0 <*> (bufferWithCapacity i >>= newIORef))++empty :: (MonadIO m, MArray a e IO) => m (Deque a e)+empty = withCapacity defaultCapacity+{-# INLINE empty #-}+ +-- unsafeRead +read :: MArray a e IO => Buffer a e -> Int -> IO e+read (Buffer s c) i = do+ readArray c (i `mod` s)+{-# INLINE read #-}++-- unsafeWrite+write :: MArray a e IO => Buffer a e -> Int -> e -> IO ()+write (Buffer s c) i e = do+ writeArray c (i `mod` s) e+{-# INLINE write #-}++grow :: MArray a e IO => Buffer a e -> Int -> Int -> IO (Buffer a e) +grow c b t = do+ c' <- bufferWithCapacity (size c * 2)+ forM_ [t..b-1] $ \i -> read c i >>= write c' i + return c'+{-# INLINE grow #-}++peekBottom :: Ptr Int -> IO Int+peekBottom p = peekElemOff p 1++peekTop :: Ptr Int -> IO Int+peekTop p = peek p++pokeBottom :: Ptr Int -> Int -> IO ()+pokeBottom p = pokeElemOff p 1++push :: (MonadIO m, MArray a e IO) => e -> Deque a e -> m ()+push o (Deque tb content) = liftIO $ withForeignPtr tb $ \p -> do+ b <- peekBottom p+ t <- peekTop p+ a <- readIORef content+ let size' = b - t+ if size' >= size a+ then do + a' <- grow a b t + writeIORef content a' + go p a' b+ else go p a b+ where+ go p arr b = do+ write arr b o+ pokeBottom p (b + 1)++data Stolen e + = Empty + | Abort + | Stolen e+ deriving (Data,Typeable,Eq,Ord,Show,Read)++steal :: (MonadIO m, MArray a e IO) => Deque a e -> m (Stolen e)+steal (Deque tb content) = liftIO $ withForeignPtr tb $ \p -> do + t <- peekTop p+ b <- peekBottom p+ a <- readIORef content+ let size' = b - t+ if size' <= 0+ then return Empty+ else do+ o <- read a t+ result <- compareAndSwapBool p t (t + 1)+ return $ if result then Stolen o else Abort++{-+steal' :: MArray a e IO => Deque a e -> IO (Maybe e)+steal' deque = do+ s <- steal deque + case s of+ Stolen e -> return (Just e)+ Empty -> return Nothing+ Abort -> steal' deque+-}++pop :: (MonadIO m, MArray a e IO) => Deque a e -> m (Maybe e)+pop (Deque tb content) = liftIO $ withForeignPtr tb $ \p -> do+ b <- peekBottom p+ a <- readIORef content+ let b' = b - 1+ pokeBottom p b'+ t <- peekTop p+ let size' = b' - t+ if size' < 0 + then do+ pokeBottom p t+ return Nothing+ else do+ o <- read a b'+ if size' > 0 + then return (Just o)+ else do+ result <- compareAndSwapBool p t (t + 1)+ if result + then do+ pokeBottom p (t + 1)+ return (Just o)+ else do+ pokeBottom p (t + 1)+ return Nothing
+ Reactor/Filtered.hs view
@@ -0,0 +1,14 @@+module Reactor.Filtered where++import Prelude hiding (filter)+import qualified Prelude++class Filtered f where+ filter :: (a -> Bool) -> f a -> f a++instance Filtered [] where+ filter = Prelude.filter++instance Filtered Maybe where+ filter p m@(Just a) | p a = m+ filter _ _ = Nothing
+ Reactor/Moore.hs view
@@ -0,0 +1,34 @@+{-# LANGUAGE DeriveDataTypeable #-}+module Reactor.Moore+ ( Moore(..)+ ) where++import Control.Applicative+import Control.Comonad+import Data.Functor.Apply+import Data.Typeable++data Moore i o = Moore { step :: i -> Moore i o, current :: o }+ deriving Typeable++instance Functor (Moore i) where+ fmap g (Moore f o) = Moore (fmap g . f) (g o)+ b <$ _ = pure b++instance Extend (Moore i) where+ duplicate m = Moore (duplicate . step m) m+ extend g m = Moore (extend g . step m) (g m)++instance Comonad (Moore i) where+ extract (Moore _ o) = o+ +instance Apply (Moore i) where+ Moore ff f <.> Moore fa a = Moore (\i -> ff i <.> fa i) (f a)+ a <. _ = a+ _ .> b = b++instance Applicative (Moore i) where+ pure o = m where m = Moore (const m) o+ (<*>) = (<.>)+ (<* ) = (<. )+ ( *>) = ( .>)
+ Reactor/Observable.hs view
@@ -0,0 +1,216 @@+module Reactor.Observable+ ( Observable(..)+ , never+ , fby+ , take, drop+ , safe+ , filterMap, (!?)+ , observe+ , counted+ -- , andThen+ , (<||>)+ -- , feed+ -- , head+ -- , safeHead+ ) where++import Prelude hiding (filter, take, drop)+import Control.Applicative+import Control.Exception hiding (handle)+import Control.Monad+import Control.Monad.Error+import Data.Foldable+import qualified Data.Foldable as Foldable+import Data.Functor.Bind+import Data.Functor.Plus+import Data.Functor.Extend+import Data.Functor.Contravariant+import Data.Monoid+import Data.IORef+import Reactor.Atomic+import Reactor.Filtered+import Reactor.Observer+import Reactor.Task+import Reactor.Subscription++newtype Observable a = Observable { subscribe :: Observer a -> Task Subscription }++instance Functor Observable where+ fmap f s = Observable (subscribe s . contramap f)++instance Apply Observable where+ mf <.> ma = Observable $ \o -> do+ funs <- io $ newIORef []+ fflag <- atomic (1 :: Int)+ aflag <- atomic (1 :: Int)+ let discard pf sf = do+ i <- atomicFetchAndAnd pf 0+ when (i == 1) $ do+ j <- atomicFetch sf+ when (j == 0) $ complete o+ mappend + <$> subscribe_ mf+ (\f -> io $ atomicModifyIORef funs (\fs -> (f:fs, ())))+ (handle o)+ (discard fflag aflag)+ <*> subscribe_ ma+ (\a -> do+ fs <- io $ readIORef funs + spawn $ Foldable.mapM_ (\f -> o ! f a) fs)+ (handle o)+ (discard aflag fflag)++instance Filtered Observable where+ filter p s = Observable (subscribe s . filter p)++instance Applicative Observable where+ pure a = Observable $ \o -> (o ! a) *> complete o $> mempty + (<*>) = (<.>)++instance Bind Observable where+ mf >>- k = Observable $ \o -> do+ counter <- atomic (1 :: Int)+ topFlag <- atomic (1 :: Int)+ let detach flag = do+ clearing flag $ do+ i <- atomicSubAndFetch counter 1+ when (i == 0) $ complete o+ subscribe_ mf + (\f -> do+ flag <- atomic (1 :: Int)+ _ <- atomicAddAndFetch counter 1 + () <$ subscribe_ (k f)+ (o !)+ (handle o)+ (detach flag))+ (handle o)+ (detach topFlag)+ +instance Monad Observable where+ return = pure+ (>>=) = (>>-)++instance Alt Observable where+ a <!> b = Observable $ \o -> subscribe_ a+ (o !)+ (handle o)+ (subscribe b o $> ())++instance Plus Observable where+ zero = Observable (\o -> complete o $> mempty)+ +instance Alternative Observable where+ empty = zero+ (<|>) = (<!>)++instance MonadPlus Observable where+ mzero = zero+ mplus = (<!>)++instance Extend Observable where+ duplicate p = Observable $ \o -> subscribe_ p + (\a -> o ! fby a p)+ (handle o) + (complete o)+ extend f p = Observable $ \o -> subscribe_ p + (\a -> o ! f (fby a p))+ (handle o)+ (complete o)++safe :: Observable a -> Observable a +safe p = Observable $ \o -> do+ alive <- atomic (1 :: Int)+ subscribe_ p+ (\a -> given alive $ o ! a)+ (\e -> clearing alive $ handle o e)+ (clearing alive $ complete o)++subscribe_ :: Observable a -> (a -> Task ()) -> (SomeException -> Task ()) -> Task () -> Task Subscription+subscribe_ a f h c = subscribe a (Observer f h c)++filterMap :: (a -> Maybe b) -> Observable a -> Observable b+filterMap p s = Observable $ \o -> subscribe s $ o ?! p++(!?) :: Observable a -> (a -> Maybe b) -> Observable b+(!?) = flip filterMap ++never :: Observable a+never = Observable $ \_ -> return mempty++fby :: a -> Observable a -> Observable a +fby a as = Observable $ \o -> do+ o ! a+ subscribe as o++take :: Int -> Observable a -> Observable a +take n p = Observable $ \o -> do+ counter <- atomic n+ subscribe_ p + (\a -> do+ i <- io $ atomicSubAndFetch counter 1+ when (i >= 0) $ o ! a+ when (i == 0) $ complete o)+ (handle o)+ (do+ i <- io $ atomicFetchAndAnd counter 0+ when (i >= 0) $ complete o)+ +drop :: Int -> Observable a -> Observable a +drop n p = Observable $ \o -> do+ counter <- atomic n+ subscribe_ p+ (\a -> do+ i <- io $ atomicSubAndFetch counter 1+ when (i < 0) $ o ! a)+ (handle o)+ (complete o)++-- | Observe both at the same time.+(<||>) :: Observable a -> Observable a -> Observable a +p <||> q = Observable $ \o -> mappend <$> subscribe p o <*> subscribe q o++observe :: Foldable f => f a -> Observable a+observe t = Observable $ \o -> do+ spawn $ do+ Foldable.forM_ t (o !)+ complete o+ return mempty++counted :: Observable a -> Observable (Int, a)+counted p = Observable $ \o -> do+ counter <- atomic 0+ subscribe_ p + (\a -> do+ i <- atomicFetchAndAdd counter 1+ o ! (i, a))+ (handle o)+ (complete o)++-- do something when the observable completes++{-+andThen :: Observable a -> Task () -> Task ()+andThen p t = spawn $ do+ () <$ subscribe_ p (\_ -> return ()) (\_ -> t) t+-}++{-++feed :: Moore i o -> Observable i -> Task o+feed machine p = do+ m <- io $ newIORef machine+ callCC $ \resume -> + subscribe p + (\i -> io $ atomicModifyIORef m $ \m' -> (step m' i, ()))+ throwError + (do Moore _ o <- io $ readIORef m+ resume o)++safeHead :: Onservable a -> Task (Maybe a)+safeHead = feed (Moore (pure . Just) Nothing) . take 1++head :: Observable a -> Task a +head = feed (Moore pure (error "head: empty observable")) . take 1++-- uncons :: Observable a -> Task (Maybe (a, Observable a))+-}
+ Reactor/Observer.hs view
@@ -0,0 +1,38 @@+{-# LANGUAGE DeriveDataTypeable #-}+module Reactor.Observer+ ( Observer(..)+ , (?!)+ ) where++import Prelude hiding (filter)+import Control.Monad+import Control.Exception hiding (handle)+import Control.Monad.Error+import Data.Monoid+import Data.Functor.Contravariant+import Data.Data+import Reactor.Filtered+import Reactor.Task++data Observer a = Observer + { (!) :: a -> Task ()+ , handle :: SomeException -> Task ()+ , complete :: Task ()+ } deriving Typeable++instance Contravariant Observer where+ contramap g (Observer f h c) = Observer (f . g) h c++instance Filtered Observer where+ filter p (Observer f h c) = Observer (\a -> when (p a) (f a)) h c++instance Monoid (Observer a) where+ mempty = Observer (\_ -> return ()) throwError (return ())+ p `mappend` q = Observer+ (\a -> do p ! a; q ! a)+ (\e -> do handle p e; handle q e)+ (do complete p; complete q)++-- filter and map in one operation+(?!) :: Observer b -> (a -> Maybe b) -> Observer a+Observer f h c ?! p = Observer (maybe (return ()) f . p) h c
+ Reactor/Subscription.hs view
@@ -0,0 +1,17 @@+{-# LANGUAGE DeriveDataTypeable #-}+module Reactor.Subscription+ ( Subscription(..)+ ) where++import Control.Applicative+import Reactor.Task+import Data.Monoid+import Data.Typeable++-- Like in real life, cancelling a subscription may not stop it from sending you stuff immediately!+newtype Subscription = Subscription { cancel :: Task () }+ deriving Typeable++instance Monoid Subscription where+ mempty = Subscription (return ())+ Subscription a `mappend` Subscription b = Subscription (a *> b)
+ Reactor/Task.hs view
@@ -0,0 +1,96 @@+{-# LANGUAGE MultiParamTypeClasses, DeriveDataTypeable #-}+module Reactor.Task+ ( Task+ , run+ , spawn+ , io+ ) where++import Control.Applicative+import Control.Monad+import Control.Exception+import Control.Monad.Reader.Class+import Control.Monad.Error.Class+import Control.Monad.IO.Class+import Data.Array.IO+import Data.Functor.Bind+import Data.Functor.Plus+import Reactor.Deque (Deque)+import Data.Data+import qualified Reactor.Deque as Deque++newtype Env = Env { envDeque :: Deque IOArray (Task ()) }++mkEnv :: IO Env +mkEnv = Env <$> Deque.empty++newtype Task a = Task + { runTask :: (a -> IO ()) -> + (SomeException -> IO ()) -> + (Env -> IO ())+ } deriving Typeable++instance Functor Task where+ fmap f (Task m) = Task $ \ks -> m (ks . f)++instance Apply Task where+ Task mf <.> Task ma = Task $ \ks kf e -> mf (\f -> ma (ks . f) kf e) kf e++instance Applicative Task where + pure a = Task (\ks _kf _e -> ks a)+ (<*>) = (<.>) ++instance Bind Task where+ Task mf >>- k = Task (\ks kf e -> mf (\a -> runTask (k a) ks kf e) kf e)+ +instance Monad Task where+ return = pure+ (>>=) = (>>-)++instance MonadReader Env Task where+ ask = Task (\ks _kf e -> ks e)+ local f (Task ma) = Task (\ks kf e -> ma ks kf (f e))++instance MonadIO Task where+ liftIO = io++io :: IO a -> Task a +io act = Task (\ks _kf _e -> act >>= ks)++instance MonadError SomeException Task where+ throwError err = Task (\_ks kf _e -> kf err)+ catchError (Task m) h = Task (\ks kf e -> m ks (\err -> runTask (h err) ks kf e) e)++instance Alt Task where+ Task ma <!> Task mb = Task (\ks kf e -> ma ks (\_ -> mb ks kf e) e)++instance Plus Task where+ zero = Task (\_ks kf _e -> kf (toException (ErrorCall "empty")))++instance Alternative Task where+ (<|>) = (<!>)+ empty = zero++instance MonadPlus Task where+ mzero = zero+ mplus = (<!>)++spawn :: Task () -> Task ()+spawn task = Task (\_ks _kf e -> Deque.push task (envDeque e))++-- run a single threaded pump, all tasks are placed locally+run :: Task () -> IO ()+run task0 = do+ env <- mkEnv+ bracket_+ (register env)+ (go env task0)+ (unregister env)+ where+ go :: Env -> Task () -> IO ()+ go env (Task m) = m (success env) (failure env) env+ success env _ = Deque.pop (envDeque env) >>= maybe (return ()) (go env)+ failure _env = throw -- TODO: shut down workers?+ register _env = return () -- TODO: start up if necessary and tell worker threads about us+ unregister _env = return () -- TODO: shutdown if necessary and tell worker threads about us+
+ Setup.lhs view
@@ -0,0 +1,7 @@+#!/usr/bin/runhaskell+> module Main (main) where++> import Distribution.Simple++> main :: IO ()+> main = defaultMain
+ reactor.cabal view
@@ -0,0 +1,37 @@+name: reactor+category: Concurrency+version: 0.1.3+license: BSD3+cabal-version: >= 1.2+license-file: LICENSE+author: Edward A. Kmett+maintainer: Edward A. Kmett <ekmett@gmail.com>+stability: experimental+homepage: http://comonad.com/reader/+copyright: Copyright (C) 2011 Edward A. Kmett+synopsis: Reactor - task parallel reactive programming+description: Reactor - task parallel reactive programming+build-type: Simple++library+ build-depends: + base >= 4 && < 4.4,+ array >= 0.3.0 && < 0.4,+ semigroupoids >= 1.2.1 && < 1.3,+ bits-atomic >= 0.1.3 && < 0.2,+ comonad >= 1.1 && < 1.2,+ transformers >= 0.2.2 && < 0.3,+ mtl >= 2.0.1.0 && < 2.1,+ contravariant >= 0.1.0.1 && < 0.2++ exposed-modules:+ Reactor.Atomic+ Reactor.Deque+ Reactor.Filtered+ Reactor.Moore+ Reactor.Observer+ Reactor.Observable+ Reactor.Subscription+ Reactor.Task++ ghc-options: -Wall