orc 1.1.2.0 → 1.2.1.1

raw patch · 4 files changed

+170/−26 lines, 4 filesdep ~basedep ~monadIOdep ~mtlPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base, monadIO, mtl, stm

API changes (from Hackage documentation)

Files

• orc.cabal +7/−7
• src/Control/Concurrent/Hierarchical.hs +37/−10
• src/Orc/Combinators.hs +64/−8
• src/Orc/Monad.hs +62/−1

@@ -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

@@ -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            

@@ -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 

@@ -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