diff --git a/orc.cabal b/orc.cabal --- a/orc.cabal +++ b/orc.cabal @@ -1,5 +1,5 @@ name: orc -version: 1.1.2.0 +version: 1.2.1.1 synopsis: Orchestration-style co-ordination EDSL description: Provides an EDSL with Orc primitives. category: Web @@ -14,12 +14,12 @@ Library Extensions: GeneralizedNewtypeDeriving - Build-Depends: base >= 4.2.0.0 && <= 4.3, - stm >= 2.1.0.0 && <= 2.2, - process >= 1.0.1.0 && <= 1.1, - mtl >= 1.1.0.0 && <= 1.2, - monadIO >= 0.9.1.0 && <= 0.10, - deepseq >= 1.1.0.0 && <= 1.2 + Build-Depends: base >= 4.2.0.0 && <= 5.0, + stm >= 2.2.0.0 && <= 2.3, + process >= 1.0.1.0 && <= 1.1, + mtl >= 2.0.1.0 && <= 2.1, + monadIO >= 0.10.1.1 && <= 0.11, + deepseq >= 1.1.0.0 && <= 1.2 Exposed-modules: Control.Concurrent.Hierarchical Orc.Monad Orc.Combinators diff --git a/src/Control/Concurrent/Hierarchical.hs b/src/Control/Concurrent/Hierarchical.hs --- a/src/Control/Concurrent/Hierarchical.hs +++ b/src/Control/Concurrent/Hierarchical.hs @@ -8,7 +8,8 @@ -- Stability : -- Portability : concurrency, unsafeIntereaveIO -- --- Hierarchical concurrent threads +-- Hierarchical concurrent threads, which kill all of their descendants +-- when they are killed. {-# OPTIONS_GHC -fno-warn-unused-do-bind #-} @@ -37,9 +38,14 @@ --------------------------------------------------------------------------- --- | - -newtype HIO a = HIO {inGroup :: Group -> IO a} +-- | The 'HIO' monad is simply the 'IO' monad augmented with an +-- environment that tracks the current thread 'Group'. This permits us +-- to keep track of forked threads and kill them en mass when an +-- ancestor is killed. The 'HIO' monad is an instance of 'MonadIO', so +-- arbitrary 'IO' actions may be embedded in it; however, the user is +-- advised that any action may be summarily killed, and thus it is of +-- extra importance that appropriate bracketing functions are used. +newtype HIO a = HIO {inGroup :: Group -> IO a} -- isomorphic to ReaderT Group IO a instance Functor HIO where fmap f (HIO hio) = HIO (fmap (fmap f) hio) @@ -60,10 +66,15 @@ --------------------------------------------------------------------------- --- | The thread-registry environment is a hierarchical structure of local +-- ^ The thread-registry environment is a hierarchical structure of local -- thread neighborhoods. +-- | A thread 'Group' keeps tracks of its inhabitants, which may be +-- threads or other 'Group's. type Group = (TVar Int, TVar Inhabitants) +-- | A group can be closed, in which case it is empty and cannot accept +-- new inhabitants, or open, in which case it contains any number of +-- threads and groups and may have additional threads/groups registered. data Inhabitants = Closed | Open [Entry] data Entry = Thread ThreadId | Group Group @@ -80,37 +91,52 @@ decrement w) +-- | Creates a new thread group and registers the current environment's +-- thread group in it. If the current group is closed, immediately +-- terminates execution of the current thread. newGroup :: HIO Group newGroup = HIO $ \w -> do w' <- newPrimGroup register (Group w') w return w' +-- | Explicitly sets the current 'Group' environment for a 'HIO' monad. local :: Group -> HIO a -> HIO a local w p = liftIO (p `inGroup` w) +-- | Kill all threads which are descendants of a 'Group' and closes the +-- group, disallowing new threads or groups to be added to the group. +-- Doesn't do anything if the group is already closed. close :: Group -> HIO () close (c,t) = liftIO $ fork (kill (Group (c,t)) >> writeTVar c 0) >> return () +-- | Blocks until the 'Group' @w@ is finished executing. finished :: Group -> HIO () finished w = liftIO $ isZero w - -runHIO :: HIO b -> IO b +-- | Runs a 'HIO' operation inside a new thread group that has no +-- parent, and blocks until all subthreads of the operation are done +-- executing. If @countingThreads@ is @True@, it then prints some +-- debugging information about the threads run (XXX: this seems +-- suboptimal.) +runHIO :: HIO b -> IO () runHIO hio = do w <- newPrimGroup r <- hio `inGroup` w isZero w when countingThreads printThreadReport - return r + return () +-- | Creates a new, empty thread group. newPrimGroup :: IO Group newPrimGroup = do count <- newTVar 0 threads <- newTVar (Open []) return (count,threads) +-- | Registers a thread/group entry @tid@ in a 'Group', terminating the +-- current thread (suicide) if the group is closed. register :: Entry -> Group -> IO () register tid (_,t) = join $ atomically $ do ts <- readTVarSTM t @@ -119,7 +145,8 @@ Open tids -> writeTVarSTM t (Open (tid:tids)) >> -- register return (return ()) - +-- | Recursively kills a thread/group entry. Does not do anything the +-- entry is a closed group. kill :: Entry -> IO () kill (Thread tid) = killThread tid kill (Group (_,t)) = do @@ -139,7 +166,7 @@ -- Profiling code: Records how many threads were created countingThreads :: Bool -countingThreads = True -- set to False to disable reporting +countingThreads = False -- set to enable reporting or not threadCount :: TVar Integer threadCount = unsafePerformIO $ newTVar 0 diff --git a/src/Orc/Combinators.hs b/src/Orc/Combinators.hs --- a/src/Orc/Combinators.hs +++ b/src/Orc/Combinators.hs @@ -21,85 +21,118 @@ ------------------ +-- | Alternate phrasing of @return ()@, which can be placed at the end +-- of an Orc computation to signal that it has no more values to +-- produce. signal :: Orc () signal = return () ------------------ --- | Cut executes an orc expression, waits for the first result, and then --- suppresses the rest, including killing any threads involved --- in computing the remainder. +-- | Cut executes an orc expression, waits for the first result, and then +-- suppresses the rest, including killing any threads involved +-- in computing the remainder. cut :: Orc a -> Orc a cut = join . eagerly +-- | Executes the computation @p@ and @done@. Once @done@ returns its +-- first result, kill both computations and returns that result. This +-- discards the results of @p@. onlyUntil :: Orc a -> Orc b -> Orc b p `onlyUntil` done = cut (silent p <|> done) +-- | Immediately executes the computation @p@, but if it hasn't returned +-- a result in @t@ seconds, execute the computation @q@ and return +-- whichever computations returns a result first (killing the other +-- thread). butAfter :: Orc a -> (Float, Orc a) -> Orc a p `butAfter` (t,def) = cut (p <|> (delay t >> def)) +-- | Executes a computation @p@, but if it hasn't returned a result in +-- @n@ seconds return @a@ instead (killing the @p@ computation). timeout :: Float -> a -> Orc a -> Orc a timeout n a p = cut (p <|> (delay n >> return a)) +-- | Executes the computation @p@ but suppresses its results. silent :: Orc a -> Orc b silent p = p >> stop +-- | Lifts a list into an Orc monad. liftList :: (MonadPlus list) => [a] -> list a liftList ps = foldr mplus mzero $ map return ps --- repeating works best when p is single-valued +-- | Repeatedly executes the computation @p@ and returns its +-- results. 'repeating' works best when @p@ is single-valued: +-- if @p@ is multivalued Orc will spawn a repeating thread for every +-- result returned, resulting in an exponential blow-up of +-- threads (XXX: I don't think this was actually intended.) repeating :: Orc a -> Orc a repeating p = do x <- p return x <|> repeating p +-- | Runs a computation @p@ and writes its results to the channel @ch@. runChan :: Chan a -> Orc a -> IO () runChan ch p = runOrc $ (p >>= writeChan ch) -------------------- +-- | Takes the first result of @p@, the first result of +-- @q@, and applies them to @f@. The computations for @p@ and @q@ are +-- run in parallel. sync :: (a->b->c) -> Orc a -> Orc b -> Orc c sync f p q = do po <- eagerly p qo <- eagerly q pure f <*> po <*> qo +-- | Runs the computation @p@ and returns its first result, but doesn't +-- return before @w@ seconds have elapsed. notBefore:: Orc a -> Float -> Orc a p `notBefore` w = sync const p (delay w) +-- | Runs a list of Orc computations @ps@ in parallel until they produce +-- their first result, and returns a list of all these results. syncList :: [Orc a] -> Orc [a] syncList ps = sequence (map eagerly ps) >>= sequence --------------------------------------------------------------------------- --- | Wait for a period of w seconds, then continue processing. +-- | Wait for a period of w seconds, then continue processing. delay :: (RealFrac a) => a -> Orc () delay w = (liftIO $ threadDelay (round (w * 1000000))) <|> (silent $ do guard (w>100) putStrLine ("Just checking you meant to wait " ++show w++" seconds")) - + --------------------------------------------------------------------------- --- | 'printOrc' and 'prompt' uses the 'Stdinout' library to provide +-- 'printOrc' and 'prompt' uses the 'Stdinout' library to provide -- basic console input/output in a concurrent setting. 'runOrc' executes -- an orc expression and prints out the answers eagerly per line. +-- | Runs an Orc computation, eagerly printing out the results of an Orc +-- computation line-by-line. printOrc :: Show a => Orc a -> IO () printOrc p = S.setupStdInOut $ runOrc $ do a <- p putStrLine ("Ans = " ++ show a) +-- | Prompts the user for a string. Concurrency-safe. prompt :: String -> Orc String prompt str = liftIO $ S.prompt str +-- | Writes a string and newline to standard output. Concurrency-safe. putStrLine :: String -> Orc () putStrLine str = liftIO $ S.putStrLine str --------------------------------------------------------------------------- +-- | Analogous to the list scan function, but the order in which +-- the combining function is applied to the results produced by +-- @p@ is nondeterministic. scan :: (a -> s -> s) -> s -> Orc a -> Orc s scan f s p = do accum <- newTVar s @@ -107,6 +140,9 @@ (w,w') <- modifyTVar accum (f x) return w' +-- | A variant of '<+>', pronounced or-else, which performs and returns +-- the results of @p@, and if @p@ produced no answers go on and performa +-- dn return the results of @q@. () :: Orc a -> Orc a -> Orc a p q = do tripwire <- newEmptyMVar @@ -119,6 +155,9 @@ Nothing -> q Just _ -> stop +-- | For each value produced by @p@, return a @Left a@. Once @p@ has +-- finished, return a @Right Int@ containing the number of results +-- produced. count :: Orc a -> Orc (Either a Int) count p = do accum <- newTVar 0 @@ -128,6 +167,8 @@ <+> fmap Right (readTVar accum) +-- | Collects all of the values of the computation @p@ and delivers them +-- as a list when @p@ is completed. collect :: Orc a -> Orc [a] collect p = do accum <- newTVar [] @@ -140,12 +181,15 @@ --------------------------------------------------------------------------- -- | List-like functions +-- | Runs the computation @p@ and returns the first @n@ results. takeOrc :: Int -> Orc a -> Orc a takeOrc n p = do vals <- newEmptyMVar end <- newEmptyMVar echo n vals end <|> silent (sandbox p vals end) +-- | Drops the first @n@ results of the computation @p@, and then +-- returns the rest of the results. dropOrc :: Int -> Orc a -> Orc a dropOrc n p = do countdown <- newTVar n @@ -157,6 +201,8 @@ writeTVarSTM countdown (w-1) return stop +-- | Zips the results of two computations @p@ and @q@. When one +-- computation finishes, kill the other. zipOrc :: Orc a -> Orc b -> Orc (a,b) zipOrc p q = do pvals <- newEmptyMVar @@ -169,11 +215,18 @@ -------------- -- Auxilliary definitions +-- | Runs the computation @p@, and repeatedly puts its results (tagged +-- with 'Just' into the @vals@ 'MVar'. Puts 'Nothing' if there are no +-- results left. Stops executing when the @end@ MVar is filled. sandbox :: Orc a -> MVar (Maybe a) -> MVar () -> Orc () sandbox p vals end = ((p >>= (putMVar vals . Just)) <+> putMVar vals Nothing) `onlyUntil` takeMVar end +-- | The rough inverse of 'sandbox', repeatedly reads values from the +-- @vals@ 'MVar' until @j@ values have been read or the @vals@ MVar is +-- exhausted (a 'Nothing' is passed). When there are no more values to +-- be returned, fills the @end@ MVar. echo :: Int -> MVar (Maybe a) -> MVar () -> Orc a echo 0 _ end = silent (putMVar end ()) echo j vals end = do @@ -182,6 +235,9 @@ Nothing -> silent (putMVar end ()) Just x -> return x <|> echo (j-1) vals end +-- | Like 'echo', repeatedly reads values from the @pvals@ and @qvals@ +-- 'MVar', returning tuples of the values until one 'MVar' is exhausted. +-- When there are no more values to be returned, fills the @end@ MVar. zipp :: MVar (Maybe a) -> MVar (Maybe b) -> MVar () -> Orc (a,b) zipp pvals qvals end = do mx <- takeMVar pvals @@ -195,10 +251,10 @@ --------------------------------------------------------------------------- + -- | Publish is a hyperstrict form of return. It is useful -- for combining results from multiple 'val' computations, providing -- a synchronization point. - publish :: NFData a => a -> Orc a publish x = deepseq x $ return x diff --git a/src/Orc/Monad.hs b/src/Orc/Monad.hs --- a/src/Orc/Monad.hs +++ b/src/Orc/Monad.hs @@ -9,7 +9,7 @@ -- Stability : -- Portability : concurrency -- --- The Orc EDSL in Haskell +-- Primitive combinators for the Orc EDSL in Haskell. {-# OPTIONS_GHC -fno-warn-unused-do-bind #-} @@ -40,6 +40,10 @@ --------------------------------------------------------------------------- +-- | A monad for many-valued concurrency, external actions and managed +-- resources. An expression of type @Orc a@ may perform many actions +-- and return many results of type @a@. The 'MonadPlus' instance does +-- not obey the Right-Zero law (@p >> stop /= stop@). newtype Orc a = Orc {(#) :: (a -> HIO ()) -> HIO ()} instance Functor Orc where @@ -50,6 +54,9 @@ p >>= h = Orc $ \k -> p # (\x -> h x # k) fail _ = stop +-- | Finish the local thread of operations, so that anything sequenced +-- afterwards is not executed. It satisfies the following law: +-- @stop >>= k == stop@ stop :: Orc a stop = Orc $ \_ -> return () @@ -57,6 +64,16 @@ empty = stop (<|>) = par +-- | Parallel choice operator that performs the actions of @p@ and @q@ +-- and returns their results as they become available. Also written +-- as @<|>@. There is no left-right bias: the ordering between @p@ and +-- @q@ is unspecified. 'par' satisfies the following laws (identity, +-- commutativity, associativity and left-distributivity across bind): +-- +-- > p <|> stop == p +-- > p <|> q == q <|> p +-- > p <|> (q <|> r) == (p <|> q) <|> r +-- > ((p <|> q) >>= k) == ((p >>= k) <|> (q >>= k)) par :: Orc a -> Orc a -> Orc a par p q = Orc $ \k -> do fork (p # k) @@ -70,6 +87,10 @@ instance MonadIO Orc where liftIO io = Orc (liftIO io >>=) +-- | Runs an Orc computation, discarding the (many) results of the +-- computation. See @collect@ on a mechanism for collecting the results +-- of a computation into a list, which may then be passed to another IO +-- thread. runOrc :: Orc a -> IO () runOrc p = runHIO (p # \_ -> return ()) @@ -84,6 +105,9 @@ --------------------------------------------------------------------------- +-- | Biased choice operator (pronounced and-then) that performs the +-- action (and returns all the results) of @p@ first, and then once done +-- performs the action of @q@. (<+>) :: Orc a -> Orc a -> Orc a p <+> q = Orc $ \k -> do w <- newGroup @@ -91,6 +115,29 @@ finished w q # k +-- | Immediately fires up a thread for @p@, and then returns a handle to +-- the first result of that thread which is also of type @Orc a@. An +-- invocation to 'eagerly' is non-blocking, while an invocation of the +-- resulting handle is blocking. 'eagerly' satisfies the following +-- laws: +-- +-- Par-eagerly: +-- +-- > eagerly p >>= (\x -> k x <|> h) +-- > == (eagerly p >>= k) <|> h +-- +-- Eagerly-swap: +-- +-- > do y <- eagerly p +-- > x <- eagerly q +-- > k x y +-- > == do x <- eagerly q +-- > y <- eagerly p +-- > k x y +-- +-- Eagerly-IO: +-- +-- > eagerly (liftIO m) >> p == (liftIO m >> stop) <|> p eagerly :: Orc a -> Orc (Orc a) eagerly p = Orc $ \k -> do res <- newEmptyMVar @@ -98,6 +145,20 @@ local w $ fork (p `saveOnce` (res,w)) k (liftIO $ readMVar res) +-- | An alternate mechanism for 'eagerly', it fires up a thread for @p@ +-- and returns a lazy thunk that contains the single (trimmed) result +-- of the computation. Be careful to use this function with 'public' +-- when these lazy values need to be fully evaluated before proceeding +-- further. For example, the following code succeeds immediately: +-- +-- > do x <- val p +-- > return x +-- +-- Whereas this code waits until @p@ has generated one result before +-- returning: +-- +-- > do x <- val p +-- > publish p val :: Orc a -> Orc a val p = Orc $ \k -> do res <- newEmptyMVar