diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,26 @@ +Copyright (c) 2008-2010, Galois, Inc. + +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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,5 @@ +module Main where + +import Distribution.Simple + +main = defaultMain diff --git a/orc.cabal b/orc.cabal new file mode 100644 --- /dev/null +++ b/orc.cabal @@ -0,0 +1,36 @@ +name: orc +version: 1.1 +synopsis: Orchestration-style co-ordination EDSL +description: Provides an EDSL with Orc primitives. +category: Web +license: BSD3 +license-file: LICENSE +author: John Launchbury, Trevor Elliott +maintainer: John Launchbury, Trevor Elliott +Copyright: (c) 2008-2010, Galois, Inc. +cabal-version: >= 1.2.0 +build-type: Simple + + +Library + Extensions: GeneralizedNewtypeDeriving + Build-Depends: base >= 4.2.0.0 && <= 4.3, + stm >= 2.1.2.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 + Exposed-modules: Control.Concurrent.Hierarchical + Orc.Monad + Orc.Combinators + Orc + Hs-Source-Dirs: src + Ghc-Options: -Wall + +Executable orc + Extensions: GeneralizedNewtypeDeriving + main-is: test.hs + Build-Depends: deepseq, + random + hs-source-dirs: src src/Examples + ghc-options: -threaded diff --git a/src/Control/Concurrent/Hierarchical.hs b/src/Control/Concurrent/Hierarchical.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Hierarchical.hs @@ -0,0 +1,155 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Hierarchical IO Threads +-- Copyright : (c) 2008-2010 Galois, Inc. +-- License : BSD3 +-- +-- Maintainer : John Launchbury +-- Stability : +-- Portability : concurrency, unsafeIntereaveIO +-- +-- Hierarchical concurrent threads + +{-# OPTIONS_GHC -fno-warn-unused-do-bind #-} + + +module Control.Concurrent.Hierarchical ( + + HIO -- :: * -> * + + , runHIO -- :: HIO b -> IO b + , newPrimGroup + , newGroup -- :: HIO Group + + , local + , close + , Group + , finished + + ) where + +import Control.Monad +import Control.Applicative +import Control.Exception +import Control.Concurrent.MonadIO +import Control.Concurrent.STM.MonadIO +import System.IO.Unsafe -- Global variable for profiling code + + +--------------------------------------------------------------------------- +-- | + +newtype HIO a = HIO {inGroup :: Group -> IO a} + +instance Functor HIO where + fmap f (HIO hio) = HIO (fmap (fmap f) hio) + +instance Monad HIO where + return x = HIO $ \_ -> return x + m >>= k = HIO $ \w -> do + x <- m `inGroup` w + k x `inGroup` w + + +instance Applicative HIO where + pure = return + f <*> x = ap f x + +instance MonadIO HIO where + liftIO io = HIO $ const io + + +--------------------------------------------------------------------------- +-- | The thread-registry environment is a hierarchical structure of local +-- thread neighborhoods. + +type Group = (TVar Int, TVar Inhabitants) +data Inhabitants = Closed | Open [Entry] +data Entry = Thread ThreadId + | Group Group + + +instance HasFork HIO where + fork hio = HIO $ \w -> block $ do + when countingThreads incrementThreadCount + increment w + fork (block (do tid <- myThreadId + register (Thread tid) w + unblock (hio `inGroup` w)) + `finally` + decrement w) + + +newGroup :: HIO Group +newGroup = HIO $ \w -> do + w' <- newPrimGroup + register (Group w') w + return w' + +local :: Group -> HIO a -> HIO a +local w p = liftIO (p `inGroup` w) + +close :: Group -> HIO () +close (c,t) = liftIO $ fork (kill (Group (c,t)) >> writeTVar c 0) + >> return () + +finished :: Group -> HIO () +finished w = liftIO $ isZero w + + +runHIO :: HIO b -> IO b +runHIO hio = do + w <- newPrimGroup + r <- hio `inGroup` w + isZero w + when countingThreads printThreadReport + return r + +newPrimGroup :: IO Group +newPrimGroup = do + count <- newTVar 0 + threads <- newTVar (Open []) + return (count,threads) + +register :: Entry -> Group -> IO () +register tid (_,t) = join $ atomically $ do + ts <- readTVarSTM t + case ts of + Closed -> return (myThreadId >>= killThread) -- suicide + Open tids -> writeTVarSTM t (Open (tid:tids)) >> -- register + return (return ()) + + +kill :: Entry -> IO () +kill (Thread tid) = killThread tid +kill (Group (_,t)) = do + (ts,_) <- modifyTVar t (const Closed) + case ts of + Closed -> return () + Open tids -> sequence_ (map kill tids) + +increment, decrement, isZero :: Group -> IO () +increment (c,_) = modifyTVar_ c (+1) +decrement (c,_) = modifyTVar_ c (\x->x-1) +isZero (c,_) = atomically $ (readTVarSTM c >>= (check . (==0))) + -- block until set (i.e. when locality is finished) + + +--------------------------------------------------------------------------- +-- Profiling code: Records how many threads were created + +countingThreads :: Bool +countingThreads = True -- set to False to disable reporting + +threadCount :: TVar Integer +threadCount = unsafePerformIO $ newTVar 0 + +incrementThreadCount :: IO () +incrementThreadCount = modifyTVar_ threadCount (+1) + +printThreadReport :: IO () +printThreadReport = do + n <- readTVar threadCount + putStrLn "----------------------------" + putStrLn (show n ++ " HIO threads were forked") + diff --git a/src/Examples/test.hs b/src/Examples/test.hs new file mode 100644 --- /dev/null +++ b/src/Examples/test.hs @@ -0,0 +1,253 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : test +-- Copyright : (c) 2008-2010 Galois, Inc. +-- License : BSD3 +-- +-- Maintainer : John Launchbury +-- Stability : +-- Portability : concurrency +-- +-- A test file for the Orc EDSL +-- + +{-# LANGUAGE GeneralizedNewtypeDeriving #-} + +module Main ( + + module Orc + , main + + ) where + +import Orc + +import Control.DeepSeq +import System.Random(randomIO) +import Data.List(sortBy) + +main = do + putStrLn "Enter a number 1-9 to run that example:" + ans <- getLine + putStrLn "----------------------------" + case ans of + "1" -> printOrc baseball + "2" -> printOrc (collect patterns) + "3" -> printOrc firstmetro + "4" -> printOrc parvalueVal + "5" -> printOrc $ count queens + "6" -> printOrc numbers + "7" -> printOrc (scan (flip (++)) [] (first 10 queens)) + "8" -> printOrc $ count (chans [1 .. 12]) + "9" -> printOrc $ first 100 $ zipper (recursive 1) (recursive 1000) + _ -> putStrLn "Unknown" + putStrLn "----------------------------" + + + +------------------------------------- +-- Shows a rich and nested set of Orc interactions + +baseball :: Orc (String,String) +baseball = do + + team <- prompt "Name a baseball team" + `notBefore` 10 + <|> prompt "Name another team" + `butAfter` (12, return "Yankees") + <|> (delay 8 >> return "Mariners") + + agree <- prompt ("Do you like "++team++"?") + `butAfter` (20, guard (team/="Mets") + >> return "maybe") + + return (team, agree) + + +----------------------------- +-- Pattern guards + +patterns :: Orc Int +patterns = do + x <- liftList [1..10] + y <- liftList [2,4..10] + (n,True) <- return (x, x `mod` y /= 0) + return n + +------------------ +-- Metronome example. Shows recursion. Also demonstrates the killing +-- of the metronome thread when the output is killed with ^C, or +-- after 3 prompts are fulfilled + +firstmetro :: Orc String +firstmetro = return "Respond to 2 prompt within 15 seconds, then check all prompts are dead" + <|> (delay 1 >> first 2 (metronome >> prompt "Hello")) + <|> (delay 15 >> return "15 seconds...") + +metronome = signal <|> (delay 2 >> metronome) + +------------------ +-- Demonstrates the concurency of 'eagerly' + +parvalue :: Orc Quote +parvalue = quotes + (Query "Quote A: Enter a number") + (Query "Quote B: Enter a number") + +newtype Query = Query {text::String} deriving (Eq,Show) +newtype Quote = Quote {price::Int} deriving (Eq,Show,Num,NFData) +noQuote = Quote 0 + +quotes :: Query -> Query -> Orc Quote +quotes srcA srcB = do + quoteA <- eagerly $ getQuote srcA + quoteB <- eagerly $ getQuote srcB + cut ( (pure least <*> quoteA <*> quoteB) + <|> (quoteA >>= threshold) + <|> (quoteB >>= threshold) + <|> (delay 25 >> quoteA <|> quoteB) + <|> (delay 30 >> return noQuote)) + +least qa qb = if price qa < price qb then qa else qb +threshold q = guard (price q < 300) >> return q + +getQuote q = prompt (text q) >>= (return . Quote . read) + +parallelOr orc1 orc2 = do + ox <- eagerly orc1 + oy <- eagerly orc2 + cut ( (ox >>= guard >> return True) + <|> (oy >>= guard >> return True) + <|> (pure (||) <*> ox <*> oy)) + + +parvalueVal :: Orc Quote +parvalueVal = quotesVal + (Query "Quote A: Enter a number") + (Query "Quote B: Enter a number") + +quotesVal :: Query -> Query -> Orc Quote +quotesVal srcA srcB = do + quoteA <- val $ getQuote srcA + quoteB <- val $ getQuote srcB + cut ( publish (least quoteA quoteB) + <|> (threshold quoteA) + <|> (threshold quoteB) + <|> (delay 25 >> publish quoteA <|> publish quoteB) + <|> (delay 30 >> return noQuote)) + +parallelOrVal orc1 orc2 = do + x <- val orc1 + y <- val orc2 + cut ( (guard x >> publish True) + <|> (guard y >> publish True) + <|> publish (x || y) ) + + + +------------------ +-- N queens + +queens = return ("Computing "++show size++"-queens...") + <+> fmap show (extend []) + + +size = 13 :: Int + +extend :: [Int] -> Orc [Int] +extend xs + | length xs == 3 = liftList $ extendL xs + | otherwise = do + j <- liftList [1 .. size] + guard $ not (conflict xs j) + extend (j:xs) + +conflict :: [Int] -> Int -> Bool +conflict rs n + = n `elem` rs -- column clash + || n `elem` zipWith (+) rs [1 .. size] -- diagonal clash + || n `elem` zipWith (-) rs [1 .. size] -- other diagonal + +extendL :: (MonadPlus liftList) => [Int] -> liftList [Int] +extendL xs + | length xs == size = return xs + | otherwise = do + j <- liftList [1 .. size] + guard $ not (conflict xs j) + extendL (j:xs) + +------------------------ +-- Shows numbers interpreted as Orc terms + +numbers :: Orc String +numbers = tally $ map prompt ["one","two","three","four"] + +tally ps = do n <- fmap sum (syncList (map success ps)) + guard (n/=2) + return $ show n + + return ("You responded to two, you lose... :-)") + +success :: Orc String -> Orc Int +success m = (m >> return 1) `butAfter` (10, return 0) + + +----------------------- +-- Demonstrates lifting functions to complex Orc expressions. + +pairs :: Orc (String, String) +pairs = sync (,) numbers queens + +----------------------- +-- Demonstrate that recursion needs to be guarded, otherwise runs away. + +recursive :: Int -> Orc Int +recursive n = do + putStrLine ("Recursion: "++show n) + liftList [1,-1] <|> (recursive (n+1) >>= \x->return (x+sign(x))) + +sign x = if x<0 then -1 else 1 + +powerset :: [a] -> Orc [a] +powerset xs = liftList $ filterM (const $ liftList [False, True]) xs + +threeset :: [a] -> Orc [a] +threeset xs = do + ys <- powerset xs + guard (length ys == 3) + return ys + +hassle = (metronome >> putStrLine("Simon")) + `onlyUntil` (delay 15 >> signal) + +chans :: [a] -> Orc [a] +chans xs = do + ch <- newChan + readChans 4000 ch <|> + silent (count (powerset xs) >>= writeChan ch) + +readChans 0 ch = do + ch' <- dupChan ch + readChans' ch <|> readChans' ch' +readChans n ch = do + mxs <- readChan ch + case mxs of + Left xs -> readChans (n-1) ch + Right n -> stop + +readChans' ch = do + mxs <- readChan ch + case mxs of + Left xs -> return xs <|> readChans' ch + Right n -> stop + +------------------------------ + +shuffle :: [a] -> IO [a] +shuffle xs = do + ks <- sequence $ take (length xs) $ repeat (randomIO :: IO Int) + return $ map snd $ sortBy key $ zip ks xs + where + key (a,x) (b,y) = if a +-- Stability : +-- Portability : concurrency +-- +module Orc ( + module Orc.Monad + , module Orc.Combinators + ) where + +import Orc.Monad +import Orc.Combinators diff --git a/src/Orc/Combinators.hs b/src/Orc/Combinators.hs new file mode 100644 --- /dev/null +++ b/src/Orc/Combinators.hs @@ -0,0 +1,199 @@ + +-------------------------------------------------------------------------------- +-- | +-- Module : Orc Combinators +-- Copyright : (c) 2008-2010 Galois, Inc. +-- License : BSD3 +-- +-- Maintainer : John Launchbury +-- Stability : +-- Portability : concurrency +-- + +{-# OPTIONS_GHC -fno-warn-unused-do-bind #-} + +module Orc.Combinators where + +import Orc.Monad +import qualified Control.Concurrent.StdInOut as S +import Control.DeepSeq + + +------------------ + +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 :: Orc a -> Orc a +cut = join . eagerly + +onlyUntil :: Orc a -> Orc b -> Orc b +p `onlyUntil` done = cut (silent p <|> done) + +butAfter :: Orc a -> (Float, Orc a) -> Orc a +p `butAfter` (t,def) = cut (p <|> (delay t >> def)) + +timeout :: Float -> a -> Orc a -> Orc a +timeout n a p = cut (p <|> (delay n >> return a)) + +silent :: Orc a -> Orc b +silent p = p >> stop + +liftList :: (MonadPlus list) => [a] -> list a +liftList ps = foldr mplus mzero $ map return ps + +-- repeating works best when p is single-valued +repeating :: Orc a -> Orc a +repeating p = do + x <- p + return x <|> repeating p + +runChan :: Chan a -> Orc a -> IO () +runChan ch p = runOrc $ (p >>= writeChan ch) + +-------------------- + +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 + +notBefore:: Orc a -> Float -> Orc a +p `notBefore` w = sync const p (delay w) + +syncList :: [Orc a] -> Orc [a] +syncList ps = sequence (map eagerly ps) >>= sequence + + +--------------------------------------------------------------------------- +-- | 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 +-- basic console input/output in a concurrent setting. 'runOrc' executes +-- an orc expression and prints out the answers eagerly per line. + +printOrc :: Show a => Orc a -> IO () +printOrc p = S.setupStdInOut $ runOrc $ do + a <- p + putStrLine ("Ans = " ++ show a) + +prompt :: String -> Orc String +prompt str = liftIO $ S.prompt str + +putStrLine :: String -> Orc () +putStrLine str = liftIO $ S.putStrLine str + + +--------------------------------------------------------------------------- + +scan :: (a -> s -> s) -> s -> Orc a -> Orc s +scan f s p = do + accum <- newTVar s + x <- p + v <- readTVar accum + let w = f x v + writeTVar accum w + return w + +() :: Orc a -> Orc a -> Orc a +p q = do + tripwire <- newEmptyMVar + do x <- p + tryPutMVar tripwire () + return x + <+> + do triggered <- tryTakeMVar tripwire + case triggered of + Nothing -> q + Just _ -> stop + +count :: Orc a -> Orc (Either a Int) +count p = do + accum <- newTVar 0 + do x <- p + modifyTVar accum (+1) + return $ Left x + <+> + fmap Right (readTVar accum) + +collect :: Orc a -> Orc [a] +collect p = do + accum <- newTVar [] + silent (do x <- p + modifyTVar accum (x:)) + <+> + readTVar accum + +ignore :: Int -> Orc a -> Orc a +ignore n p = do + countdown <- newTVar n + x <- p + join $ atomically $ do + w <- readTVarSTM countdown + if w==0 then return $ return x + else do + writeTVarSTM countdown (w-1) + return stop + +first :: Int -> Orc a -> Orc a +first n p = do + vals <- newEmptyMVar + end <- newEmptyMVar + echo n vals end <|> silent (sandbox p vals end) + where + echo 0 _ end = silent (putMVar end ()) + echo j vals end = do + mx <- takeMVar vals + case mx of + Nothing -> silent (putMVar end ()) + Just x -> return x <|> echo (j-1) vals end + +sandbox :: Orc a -> MVar (Maybe a) -> MVar () -> Orc () +sandbox p vals end + = ((p >>= (putMVar vals . Just)) <+> putMVar vals Nothing) + `onlyUntil` takeMVar end + + +zipper :: Orc a -> Orc b -> Orc (a,b) +zipper p q = do + pvals <- newEmptyMVar + qvals <- newEmptyMVar + end <- newEmptyMVar + zipp pvals qvals end + <|> silent (sandbox p pvals end) + <|> silent (sandbox q qvals end) + +zipp :: MVar (Maybe a) -> MVar (Maybe b) -> MVar () -> Orc (a,b) +zipp pvals qvals end = do + mx <- takeMVar pvals + my <- takeMVar qvals + case mx of + Nothing -> silent (putMVar end () >> putMVar end ()) + Just x -> case my of + Nothing -> silent (putMVar end () >> putMVar end ()) + Just y -> return (x,y) <|> zipp pvals qvals end + + + +--------------------------------------------------------------------------- +-- | 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 new file mode 100644 --- /dev/null +++ b/src/Orc/Monad.hs @@ -0,0 +1,114 @@ + +-------------------------------------------------------------------------------- +-- | +-- Module : Orc Monad +-- Copyright : (c) 2008-2010 Galois, Inc. +-- License : BSD3 +-- +-- Maintainer : John Launchbury +-- Stability : +-- Portability : concurrency +-- +-- The Orc EDSL in Haskell + +{-# OPTIONS_GHC -fno-warn-unused-do-bind #-} + + +module Orc.Monad ( + Orc -- :: * -> * + , module Control.Monad + , module Control.Applicative + , module Control.Concurrent.MonadIO + , module Control.Concurrent.STM.MonadIO + , stop -- :: Orc a + , eagerly -- :: Orc a -> Orc (Orc a) + , val -- :: Orc a -> Orc a + , (<+>) -- :: Orc a -> Orc a -> Orc a + , runOrc -- :: Orc a -> IO () + + ) where + + +import Control.Monad +import Control.Applicative +import Control.Concurrent.MonadIO +import Control.Concurrent.STM.MonadIO +import Control.Concurrent.Hierarchical + +import System.IO.Unsafe + + +--------------------------------------------------------------------------- + +newtype Orc a = Orc {(#) :: (a -> HIO ()) -> HIO ()} + +instance Functor Orc where + fmap f p = Orc $ \k -> p # (k . f) + +instance Monad Orc where + return x = Orc $ \k -> k x + p >>= h = Orc $ \k -> p # (\x -> h x # k) + fail _ = stop + +stop :: Orc a +stop = Orc $ \_ -> return () + +instance Alternative Orc where + empty = stop + (<|>) = par + +par :: Orc a -> Orc a -> Orc a +par p q = Orc $ \k -> do + fork (p # k) + q # k +{- Fully symmetric version: relevant if using async exceptions etc. + fork (q # k) + return () +-} + + +instance MonadIO Orc where + liftIO io = Orc (liftIO io >>=) + +runOrc :: Orc a -> IO () +runOrc p = runHIO (p # \_ -> return ()) + +instance Applicative Orc where + pure = return + f <*> x = ap f x + +instance MonadPlus Orc where + mzero = empty + mplus = (<|>) + + +--------------------------------------------------------------------------- + +(<+>) :: Orc a -> Orc a -> Orc a +p <+> q = Orc $ \k -> do + w <- newGroup + local w $ fork (p # k) + finished w + q # k + +eagerly :: Orc a -> Orc (Orc a) +eagerly p = Orc $ \k -> do + res <- newEmptyMVar + w <- newGroup + local w $ fork (p `saveOnce` (res,w)) + k (liftIO $ readMVar res) + +val :: Orc a -> Orc a +val p = Orc $ \k -> do + res <- newEmptyMVar + w <- newGroup + local w $ fork (p `saveOnce` (res,w)) + k (unsafePerformIO $ readMVar res) -- Like unsafeInterleaveIO + +saveOnce :: Orc a -> (MVar a,Group) -> HIO () +p `saveOnce` (r,w) = do + ticket <- newMVar () + p # \x -> (takeMVar ticket >> putMVar r x >> close w) + + +---------------------------------------------------------------------------