diff --git a/Actor/ActorBase.hs b/Actor/ActorBase.hs new file mode 100644 --- /dev/null +++ b/Actor/ActorBase.hs @@ -0,0 +1,150 @@ + +{-# LANGUAGE UndecidableInstances, FlexibleInstances #-} + + +{- +-------------------------------------------------------------------------------- +-- +-- Copyright (C) 2008 Martin Sulzmann, Edmund Lam. 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 Isaac Jones 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. + +-} + +module Actor.ActorBase where + +{- +Provides basic functionality such as + + - fresh variable, tag creation, + we use tags to check if a variable has been bound to a value + + - generic pattern matcher, needs to be extended for each specific message type + + - internal matcher uses tags to keep track of already matched messages + (we assume multi-set matching!) + + +-} + +import IO +import Monad +import Data.IORef +import Control.Concurrent +import Control.Concurrent.STM + + +-- tags, a unique tag is attached to each variable and (internal) message + +type Tag = IORef () + +newTag :: IO (IORef ()) +newTag = newIORef () + +lookupTag :: [Tag] -> Tag -> Bool +lookupTag tags tag = or (map (\t -> t == tag) tags) + +-- pattern variables + +data VAR a = VAR { variable :: (IORef a), var_tag :: Tag } + | DontCare deriving Eq + +data L a = Val a + | Var (VAR a) deriving Eq + + +-- dont care vars are convenient and avoid to generate "too many" annonymous variables +dontCareVar :: IO (VAR a) +dontCareVar = return DontCare + +newVar :: IO (VAR a) +newVar = do { var <- newIORef undefined + ; tag <- newIORef () + ; return (VAR { variable = var, var_tag = tag}) + } +readVar :: VAR a -> IO a +readVar (VAR {variable = var}) = readIORef var + +writeVar :: VAR a -> a -> IO () +writeVar (VAR {variable = var}) x = writeIORef var x + +-- matching of ground messages against patterns +-- we thread through the set of already bound variables, +-- identified via their tags, and already matched messages + +-- external matcher, untagged message against pattern messages +class EMatch a where + match :: [Tag] -> a -> a -> IO (Bool,[Tag]) + +-- generic instance, remaining instances depend on (external) message +instance (Eq a, EMatch a,Show a) => EMatch (L a) where + match tags (Val x) (Val y) = match tags x y + match tags (Val x) (Var DontCare) = return (True, tags) + match tags (Val x) (Var (y@(VAR {var_tag = tag}))) = + if lookupTag tags tag + then -- y already bound + do { v <- readVar y + -- ; putStr "Debug: match \n" + -- ; putStr ("x= " ++ (show x) ++ " \n") + -- ; putStr ("v= " ++ (show v) ++ " \n") + ; return (x==v,tags) } + else -- y not bound yet + do { writeVar y x + -- ; putStr "Debug2: match \n" + -- ; putStr ("x= " ++ (show x) ++ " \n") + ; return (True,tag:tags) } + match tags (Var _) _ = error "the impossible has happened" + + +-- internal matcher, tagged messages against pattern message +class Match a where + internal_match :: [Tag] -> (InternalMsg a) -> a -> IO (Bool,[Tag]) + +-- each message carries a tag, so we can identify already matched messages +-- (we assume multi-set matching, ie a message is a resource and can only be matched once) + +data InternalMsg a = InternalMsg { message :: a, msg_tag :: Tag } deriving Eq + +instance Show a => Show (InternalMsg a) where + show (InternalMsg {message = msg}) = show msg + +-- the only generic instance +instance EMatch a => Match a where + internal_match tags (InternalMsg {message = msg1, msg_tag = msgtag}) msg2 = + if lookupTag tags msgtag + then -- msg already matched, multi-set matching + return (False,tags) + else do { (b,tags2) <- match tags msg1 msg2 + ; if b + then return (b,msgtag:tags2) -- we remember the tag of the match + else return (False,tags) } + + + + + diff --git a/Actor/ActorCompiler.hs b/Actor/ActorCompiler.hs new file mode 100644 --- /dev/null +++ b/Actor/ActorCompiler.hs @@ -0,0 +1,352 @@ + +{-# LANGUAGE MultiParamTypeClasses, FunctionalDependencies #-} + +{- +-------------------------------------------------------------------------------- +-- +-- Copyright (C) 2008 Martin Sulzmann, Edmund Lam. 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 Isaac Jones 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. + +-} + + +module Actor.ActorCompiler where + +{- version using memoization, the programmer must explicitly provide the "line number" -} + +import Actor.ActorBase +import Actor.ActorSyntax + + +-- the bare actor class, minimal functionality required to +-- implement the multi-set message matching algorithm + + +-- TODO: we need rm -> a (which seems too strong), otherwise some constraints cant be deduced + +class EMatch m => Actor a m rm idx st | a -> m rm idx st, rm -> a where -- we could use type functions here + -- a is the actor type + -- m is the (plain) message + -- rm is the (rich) message with more (internal) information, eg the message's location + -- idx is the search index where to start + -- st is the current state of our search + + getMessage :: a -> Maybe Int -> IO (Maybe rm) + deleteMsg :: a -> rm -> IO () + getIndex :: a -> m -> IO idx -- m is a pattern message + initSearch :: a -> idx -> IO st + nextMsg :: a -> st -> IO (Maybe rm) + resetMB :: a -> IO () + + extractMsg :: rm -> IO (InternalMsg m) + + -- memoization + + codeLookup :: a -> Int -> IO (Maybe [CompClause a rm ()]) + memoCode :: a -> (Int,[CompClause a rm ()]) -> IO () +{- +getMessage gets the most recent active message which is then put +into the store (pool of inactive messages) +a time-out (microseconds, action) is an additional parameter +if getMessage returns Nothing the time for waiting for an +incoming message has exceeded the time-out limit, +if Just x then x is the next incoming message + +deleteMsg deletes message in the store + +getIndex computes the index for each message + +initSearch sets up the search based on a given index idx + +nextMsg gets the next message matching a given index idx + +resetMB restores the mailbox by putting store contraints back into the queue/mailbox +-} + + +-- behavior of receive clauses can be specified via the following data type + +data MemoFlag = NoMemo | Memo Int + +data ReceiveParameters = + RecParm { memo :: MemoFlag + -- we'll use the provided Int to store the 'compiled' clauses + , resetAction :: IO () + -- the action to be executed before executing the body (rhs) + -- typically, resetMB + , timeOut :: Maybe (Int, IO ()) + -- waits number of microseconds for incoming message, then executes action + -- and 'aborts' receive clause + } + + + +-- (compiled) receive clauses representation +----------------------------------------------- + +type Code_RHS a = IO a + +-- external receive clauses are pairs of ([MatchTask msg], Code_RHS a) +-- we compile them to a function + +type CompClause act rmsg code = rmsg -> act -> IO (Maybe (Code_RHS code)) + -- rmsg is the currently active message (in rich format) + -- search for matching lhs performed in IO + -- if match found commit and return code for rhs + -- otherwise return Nothing + + +-- receive clauses execution scheme +------------------------------------------ + + + +-- NOTE: We can only support memo for receive bodies returning () "unit", +-- For any other return type, some types would be too polymorphic. + + +-- standard reveive, no memoization, always reset MB, no timeout +receive :: (Actor act msg rmsg idx st, Show msg) => act -> [([MatchTask msg], Code_RHS a)] -> IO a +receive act prog = receiveInternal act (build prog) + where + build [] = [] + build ((tasks,body):rest) = (compile (do {resetMB act; body}) tasks) ++ (build rest) + -- maintain order of clauses! + +-- generalized, parameterized receive +receiveParm :: (Actor act msg rmsg idx st, Show msg) => + act -> ReceiveParameters -> [([MatchTask msg], Code_RHS ())] -> IO () +receiveParm act parm prog = + let -- 'compilation' of match clauses + build [] = [] + build ((tasks,body):rest) = (compile (do {resetAction parm; body}) tasks) ++ (build rest) + -- we plug in the resetAction before we execute the body + -- maintain order of clauses (important in case of sequential execution) + + in case (memo parm) of + NoMemo -> receiveInternal2 act parm (build prog) -- no memoization + Memo idx -> do res <- codeLookup act idx -- memoization + case res of -- check if already stored + Just code -> receiveInternal2 act parm code + Nothing -> do let code = build prog + memoCode act (idx,code) + receiveInternal2 act parm code + +receiveInternal :: (Actor a m rm idx st, Show m) => a -> [CompClause a rm c] -> IO c +receiveInternal act comp = + do { Just active_msg <- getMessage act Nothing -- no timeout + --; putStr "** StartSearch with getMessage **\n" + ; res <- select active_msg act comp + ; case res of + Just action -> action + Nothing -> receiveInternal act comp + } +-- get active message (which we then put into the store, pool of inactive messages) +-- check if the active messages fires any of the receive clauses (tried from top to bottom) +-- if yes, simply execute rhs/body, otherwise repeat, get a next/new active message ... + + +receiveInternal2 :: (Actor a m rm idx st, Show m) => a -> ReceiveParameters -> [CompClause a rm ()] -> IO () +receiveInternal2 act parm compiled_clauses = + let setupTimeOut = case (timeOut parm) of + Nothing -> Nothing + Just (t,_) -> Just t + timeOutAction = let Just (_,action) = timeOut parm + in action + choose Nothing a1 a2 = a1 + choose (Just x) a1 a2 = a2 x + seqMsgSeqCls = + -- sequential processing of incoming messages, and + -- sequential, top to bottom, application of clauses + do active_msg_res <- getMessage act setupTimeOut + case active_msg_res of + Just active_msg -> + do --putStrLn $ "Msg arrived" + res <- select active_msg act compiled_clauses + case res of + Just action -> do --putStrLn "fire" + action + Nothing -> do --putStrLn "failed, try again" + seqMsgSeqCls + Nothing -> timeOutAction + in seqMsgSeqCls + + + +select :: Actor a m rm idx st => rm -> a -> [CompClause a rm c] -> IO (Maybe (Code_RHS c)) +select _ _ [] = return Nothing +select msg act (comp:comps) = + do { res <- comp msg act + ; case res of + Just action -> return (Just action) + Nothing -> select msg act comps + } +-- based on the given message, check if any of the (compiled) clauses can be executed + + + +-- compilation scheme for receive clauses +------------------------------------------------------ + +-- compilation of a single receive clause +-- yields several compiled clauses, any of the messages +-- in the pattern could be matched first + +compile :: (Actor act msg rmsg idx st, Show msg) => + Code_RHS a -> [MatchTask msg] -> [CompClause act rmsg a] +compile body tasks = + map compileClause (optimize (generateTasks tasks)) + where + +-- compileClause tries to match the first message pattern +-- the search for the remaining message patterns are done by compileSingle + + --compileClause :: [MatchTask msg] -> rmsg -> act -> IO (Maybe (IO a)) + compileClause tasks msg act = -- msg is a message in rich format + -- \msg -> \act -> + case (head tasks) of -- get the first task + Simp active_msg -> + do { -- putStr "Matchsearch start:\n" + -- ; putStr ("active_msg = " ++ show active_msg ++ "\n") + -- ; putStr ("msg = " ++ show msg ++ "\n") + plain_msg <- extractMsg msg + ;(b,var_env) <- internal_match [] plain_msg active_msg + -- multi-set matching, see definition of internal_match + -- where we use tags to remember already matched messages + ; if b + then compileSingle act + (do {deleteMsg act msg; body}) var_env (tail tasks) + -- compileSingle accumulates all deletes by putting them + -- in front of body + else return Nothing } + Prop active_msg -> + do { plain_msg <- extractMsg msg + ; (b,var_env) <- internal_match [] plain_msg active_msg + -- multi-set matching, see definition of internal_match + ; if b + then compileSingle act (do {body}) var_env (tail tasks) + -- no delete for props + else return Nothing } + Guard _ -> error "A guard can't be the first match task" + +-- nothing at the moment +-- early guard scheduling, alternative semantics (best match), ... +optimize :: [[MatchTask msg]] -> [[MatchTask msg]] +optimize = id + +-- we only need to guarantee that each element appears once in front +generateTasks [] = error "simps/props can't be empty" +generateTasks xs = + let go 1 xs = [xs] + go n xs = [xs] ++ go (n-1) (shuffle xs) + in go (length xs) xs + +shuffle [] = [] +shuffle (x:xs) = xs ++ [x] + +-- we permute tasks, +-- the first tasks must be either Simp or Prop +-- all guards are at the tail to ensure that there're no unbound variables +-- NOTE: But we don't catch cases like [ x > y ] ! +permuteTasks :: [MatchTask msg] -> [[MatchTask msg]] +permuteTasks tasks = [ ys ++ guards | ys <- permute simps_props ] + where + test (Prop _) = True + test (Simp _) = True + test (Guard _) = False + simps_props = filter (\t -> test t) tasks + guards = filter (\t -> not (test t)) tasks + +-- brute-force permutation, unnecessary +permute [] = [] +permute [x] = [[x]] +permute (x:xs) = + [ zs | ys <- permute xs, zs <- patch x ys ] + +patch x zs = [x:zs] ++ [(take n zs) ++ [x] ++ (drop n zs) | n <- [1..length zs]] + + +-- compilation for a fixed sequence of match tasks +-- we thread through a list of tags, var_env, to check for already bound pattern variables +-- and already matched messages +compileSingle :: (Actor act msg rmsg idx st, Show msg) => + act -> Code_RHS a -> [Tag] -> [MatchTask msg] -> IO (Maybe (Code_RHS a)) +compileSingle act body _ [] = return (Just body) +compileSingle act body var_env ((Guard guard):tasks) = + do { b <- guard + ; if b + then compileSingle act body var_env tasks + else return Nothing + } +compileSingle act body var_env (task:tasks) = + let getMsg (Simp x) = x + getMsg (Prop x) = x + getMsg _ = error "the impossible has happened, always check for guards first" + in case getMsg task of + active_msg -> + +-- we perform a linear search of the store to find a match for active_msg +-- st is a store pointer, points to the current "inactive" message in the store +-- nextMsg will update st to the next "inactive" message + + do { idx <- getIndex act active_msg + ; st <- initSearch act idx + ; search st } + where + search st = + do { result <- nextMsg act st + ; case result of + Nothing -> return Nothing -- back-track + Just msg -> + do { -- putStr "Matchsearch2 \n" + -- ; putStr ("active_msg = " ++ show active_msg ++ "\n") + -- ; putStr ("msg = " ++ show msg ++ "\n") + plain_msg <- extractMsg msg + ; (b,var_env2) <- internal_match var_env plain_msg active_msg + -- side-effect of binding pattern variables + -- multi-set matching, see definition of internal_match + ; if b then + do { rest <- compileSingle act body var_env2 tasks + -- choice point + ; case rest of + Nothing -> search st + Just code_rest -> + case task of + Simp _ -> + return (Just (do {deleteMsg act msg; code_rest})) + -- we must delete msg, not active_msg + -- we use syntactic equality testing for delete + Prop _ -> return (Just code_rest) + -- nothing to delete + } + else search st --return Nothing is wrong + -- we must continue the search !!!!! + } + } + diff --git a/Actor/ActorLinearSearch.hs b/Actor/ActorLinearSearch.hs new file mode 100644 --- /dev/null +++ b/Actor/ActorLinearSearch.hs @@ -0,0 +1,266 @@ + +{-# LANGUAGE MultiParamTypeClasses, TypeSynonymInstances, FlexibleInstances #-} + + +{- +-------------------------------------------------------------------------------- +-- +-- Copyright (C) 2008 Martin Sulzmann, Edmund Lam. 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 Isaac Jones 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. + +-} + + +module Actor.ActorLinearSearch where + + +{- + +An implementation of actors using channels for the (external) mailbox +and a linear search algorithm for implementing multi-set message matching. + +NOTE: We can only support memo for receive bodies returning () "unit", + For any other return type, some types would be too polymorphic. + +-} + +import IO +import Monad +import Data.IORef +import Control.Concurrent +import Control.Concurrent.Chan + +import Actor.Timeout +import Actor.ActorBase +import Actor.ActorCompiler +import Actor.SList +import Actor.QList + + +---------------------------------------- +-- primitives for hashing messages + +type HashIdx = Int + +data HashOp msg = + HashOp { numberOfTables :: HashIdx + , hashMsg :: msg -> HashIdx } + +-- we use QLists to represent each hash table + +----------------------------------- +-- a concrete actor instance + + +-- linear store representation +data Act msg = Act { external_mailbox :: Chan msg, + internal_mailbox :: SList (InternalMsg msg), + internal_mailbox_barrier :: Iterator (InternalMsg msg), + memoTable :: IORef [(Int,[CompClause (Act msg) (Location msg) ()])], + hashTable :: [QList (InternalMsg msg)], + hashOp :: HashOp msg, + thread_id :: MVar ThreadId + } + + +{- +The mailbox consists of two parts: + + The internal mailbox is represented as a singly-linked list. + The barrier (a iterator) separates active from inactive messages. + A message is inactive if it couldn't contribute in firing a receive clause. + We'll maintain a store of inactive messages because in combination with the + latest active message we may actually fire a receive clause. + + getMessage retrieves the next active message, either by pushing up the barrier. + If the list is empty, we read from the external mailbox. + + nextMsg scans the store (ie inactive messages) + + deleteMsg logically deletes a message. That is, we turn on a flag saying that the + message has been logically deleted. + + The actual physical delete happens during getMessage. + + resetMB re-actives all store messages, we assign the barrier to a new iterator + +-} + +-- abbreviations + +type PID msg = (Chan msg,MVar ThreadId) + +type Location msg = IORef (Item (InternalMsg msg)) + +-- actor interface + +actorToPID :: Act msg -> PID msg +actorToPID act = (external_mailbox act, thread_id act) + +kill :: PID msg -> IO () +kill (_,tid) = do { id <- readMVar tid + ; killThread id } + + +send :: PID msg -> msg -> IO () +send (emb,_) msg = writeChan emb msg + + +createActor :: HashOp msg -> IO (Act msg) +createActor hop = + do { emb <- newChan -- create empty external mailbox + ; imb <- newSList -- create empty internal store + ; barrier <- newIterator imb -- initial barrier + ; memTble <- newIORef [] + ; let n = numberOfTables hop + ; ht <- mapM (\ _ -> newQList) [1..n] + ; tid <- newEmptyMVar + -- only after runActor we can initialize this value + -- we use MVars to avoid access to the uninitialized value + ; let actor = Act {external_mailbox = emb, + internal_mailbox = imb, + internal_mailbox_barrier = barrier, + hashTable = ht, + hashOp = hop, + memoTable = memTble, + thread_id = tid} + ; return actor + } + +runActor :: Act msg -> (Act msg -> IO ()) -> IO () +runActor (actor@(Act {thread_id = tid})) compiled_actor = + do { forkIO ( do { actual_tid <- myThreadId + ; putMVar tid actual_tid + ; compiled_actor actor } ) + ; return () + } + +{- +newActor :: (Act msg -> IO ()) -> IO (Act msg) +we don't support this functionality anymore +must use createActor and runActor instead +-} + + + +instance (EMatch msg, Eq msg, Show msg) => Actor (Act msg) + msg (Location msg) + HashIdx (QIterator (InternalMsg msg)) where +-- get new active message: +-- (1) message may be available in the internal store, check barrier +-- Check if barrier points to active node +---(2) Otherwise, read from external mailbox (blocking) +--getMessage :: Act msg -> IO (Location msg) + getMessage (act@(Act {external_mailbox = emb, + internal_mailbox = imb, + internal_mailbox_barrier = barrier })) time = + do curNode <- iterateSList barrier + case curNode of + Just curPtr -> return (Just curPtr) + Nothing -> -- we must have hit the tail + let -- read from external mailbox, possibly apply time-out + checkTimeout Nothing = do + r <- readChan emb + + return (Just r) + checkTimeout (Just t) = timeout t (readChan emb) + in do check <- checkTimeout time + case check of + Nothing -> return Nothing -- time-out applied + + Just untagged_new_msg -> + do + -- add newly received message to store + new_tag <- newTag + let new_msg = InternalMsg {message = untagged_new_msg, msg_tag = new_tag} + curPtr <- addToTail imb new_msg + iterateSList barrier -- bump up the barrier + + -- add to appropriate hash table + let hashIdx = (hashMsg (hashOp act)) untagged_new_msg + let qlist = (hashTable act) !! (hashIdx -1) + enQueue qlist curPtr + + return (Just curPtr) + + +--getIndex :: Act msg -> msg -> IO HashIdx + getIndex act m = return ((hashMsg (hashOp act)) m) + + +-- we perform a systematic search of the hash table (a queue), +-- we use an iterator, a pointer to a pointer, via which we scan +-- the queue from 'oldest' (head) to 'newest' (tail) +--initSearch :: Act msg -> HashIdx -> IO (QIterator (InternalMsg msg)) + initSearch act idx = + do let qlist = (hashTable act) !! (idx -1) + it <- newQIterator qlist + return it + + +-- we use the iterator to scan through the queue +--nextMsg :: Act msg -> (QIterator (InternalMsg msg)) -> IO (Maybe (Location msg)) + nextMsg act curIterator = + do res <- iterateQList curIterator + return res + + +-- we only logically delete a message, the actual physical delete (in the store and hash table) +-- happens when applying getMessage (iterateSList) and newMsg (iterateQList) +--deleteMsg :: Act mst -> (Location msg) -> IO () + deleteMsg _ ptr = + do node <- readIORef ptr + case node of + Null -> error "deleteMsg: something went wrong" + Node {val = x, next = n} -> writeIORef ptr (Node {val = x, deleted = True, next = n}) + +-- we reset the barrier to the head/beginning of the store +-- NOTE: no need to "reset" hashtables, already stored messages remain +-- where they are, same applies to hashed messages +--resetMB :: Act msg -> IO () + resetMB (Act {internal_mailbox = imb, + internal_mailbox_barrier = barrier}) = + do restartIt <- newIterator imb + assignIterator barrier restartIt + +--extractMsg :: Location msg -> IO (InternalMsg msg) + extractMsg ptr = + do node <- readIORef ptr + return (val node) + + codeLookup (Act {memoTable = memTble}) line_no = + do tble <- readIORef memTble + return (lookup line_no tble) + + memoCode (Act {memoTable = memTble}) (x@(line_no,code)) = + do tble <- readIORef memTble + writeIORef memTble (x:tble) + return () + + diff --git a/Actor/ActorSyntax.hs b/Actor/ActorSyntax.hs new file mode 100644 --- /dev/null +++ b/Actor/ActorSyntax.hs @@ -0,0 +1,103 @@ + +{-# LANGUAGE MultiParamTypeClasses, FunctionalDependencies, FlexibleInstances, UndecidableInstances #-} + +{- +-------------------------------------------------------------------------------- +-- +-- Copyright (C) 2008 Martin Sulzmann, Edmund Lam. 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 Isaac Jones 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. + +-} + +module Actor.ActorSyntax where + +{- + +Describes external syntax of receive classes and translation to internal form + +-} + + + +-------------------------------------------------------------------- +-- Representation of receive clauses + + +-- pattern clauses are of the form +-- pat_clause := pat* (.\. pat*)? (when guard)? | +-- pat := message +-- guard := IO Bool + +-- receive clauses are of the form +-- rec_clause := pat_clause ".->." body + + +-------------------------------------------------------------------- +-- Translation to internal form + + +-- internal form of match tasks + +data MatchTask a = Prop a + | Simp a + | Guard (IO Bool) + + + +-- we immediately turn each pattern clause into a list of match tasks + +(.->.) x y =(convertHead x, y) + +(.\.) x y = (x,y) + +data WITHGUARD a = WITHGUARD a (IO Bool) + +when x y = WITHGUARD x y + +infixr 4 .->. + +infixr 9 .\. + + +class ConvertHead a b | a -> b where + convertHead :: a -> [MatchTask b] + + +-- Simp/Prop case +instance ConvertHead ([a],[a]) a where + convertHead (xs,ys) = (map Prop xs) ++ (map Simp ys) + +-- only Simp case +instance ConvertHead [a] a where + convertHead xs = map Simp xs + +-- additional guard +instance ConvertHead b a => ConvertHead (WITHGUARD b) a where + convertHead (WITHGUARD x y) = (convertHead x) ++ [Guard y] + diff --git a/Actor/QList.hs b/Actor/QList.hs new file mode 100644 --- /dev/null +++ b/Actor/QList.hs @@ -0,0 +1,126 @@ + + + +{- +-------------------------------------------------------------------------------- +-- +-- Copyright (C) 2008 Martin Sulzmann, Edmund Lam. 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 Isaac Jones 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. + +-} + +module Actor.QList where + +import IO +import Monad +import Data.IORef + +import Actor.SList + +-- a QList is a queue (represented as a list) of elements pointing to SList elements + +data QItem a = QNode { valQ :: IORef (Item a) + , nextQ :: IORef (QItem a) } + | QNull + | QHead { nextQ :: IORef (QItem a) } + +data QList a = QList { headQ :: IORef (IORef (QItem a)), + tailQ :: IORef (IORef (QItem a)) } + +-- we assume a static head pointer, pointing to the first node which must be Head +-- tail points to the last node which must be Null + +type QIterator a = IORef (IORef (QItem a)) + + +-- basic api + +-- we create a new Qlist +newQList :: IO (QList a) +newQList = + do null <- newIORef QNull + hd <- newIORef (QHead null) + hdPtr <- newIORef hd + tailPtr <- newIORef null + return (QList {headQ = hdPtr, tailQ = tailPtr}) + + +-- we add a new node, by overwriting the null tail node +-- we only need to adjust tail but not head because +-- of the static Head +-- we return the location of the newly added node +enQueue :: QList a -> IORef (Item a) -> IO () +enQueue (QList {tailQ = tailPtrPtr}) x = + do null <- newIORef QNull + tailPtr <- readIORef tailPtrPtr + writeIORef tailPtr (QNode {valQ = x, nextQ = null}) + writeIORef tailPtrPtr null + + +-- the iterator always points to the PREVIOUS node, +-- recall that there's a static dummy new Head +newQIterator :: QList a -> IO (QIterator a) +newQIterator (QList {headQ = hd}) = + do hdPtr <- readIORef hd + it <- newIORef hdPtr + return it + + + +-- we iterate through the list and return the first "not deleted" node +-- we delink deleted nodes +-- there's no need to adjust head, tail +-- cause head has a static Head and +-- tail points to Null +iterateQList :: QIterator a -> IO (Maybe (IORef (Item a))) +iterateQList itPtrPtr = + let go prevPtr = + do prevNode <- readIORef prevPtr + let curPtr = nextQ prevNode -- the prev node has always a next field + -- recall the static dummy head node + curNode <- readIORef curPtr + case curNode of + QNull -> return Nothing -- reached end of the list + QNode { valQ = itemPtr, nextQ = nextNode } -> + do Node { deleted = del } <- readIORef itemPtr -- check element in store + if not del -- if deleted + then -- node available + do writeIORef itPtrPtr curPtr -- adjust iterator + return (Just itemPtr) + else -- delete curNode by setting the next of prevNode to next of curNode + do case prevNode of + QHead {} -> writeIORef prevPtr (QHead {nextQ = nextNode}) + QNode {} -> writeIORef prevPtr + (QNode {valQ = valQ prevNode, nextQ = nextNode}) + go prevPtr + in do startPtr <- readIORef itPtrPtr + go startPtr + + + diff --git a/Actor/SList.hs b/Actor/SList.hs new file mode 100644 --- /dev/null +++ b/Actor/SList.hs @@ -0,0 +1,142 @@ + +{- +-------------------------------------------------------------------------------- +-- +-- Copyright (C) 2008 Martin Sulzmann, Edmund Lam. 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 Isaac Jones 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. + +-} + + +module Actor.SList where + +import IO +import Monad +import Data.IORef + +--------------------------------------- +-- API for a singly-linked list +-- We use this structure for representing the internal mailbox + + +data Item a = Node { val :: a + , deleted :: Bool + , next :: IORef (Item a) } + | Null + | Head { next :: IORef (Item a) } + +data SList a = SList { headList :: IORef (IORef (Item a)), + tailList :: IORef (IORef (Item a)) } + +-- we assume a static head pointer, pointing to the first node which must be Head +-- tail points to the last node which must be Null + +type Iterator a = IORef (IORef (Item a)) + + +-- basic api + +-- we create a new list +newSList :: IO (SList a) +newSList = + do null <- newIORef Null + hd <- newIORef (Head null) + hdPtr <- newIORef hd + tailPtr <- newIORef null + return (SList {headList = hdPtr, tailList = tailPtr}) + + +-- we add a new node, by overwriting the null tail node +-- we only need to adjust tailList but not headList because +-- of the static Head +-- we return the location of the newly added node +addToTail :: SList a -> a -> IO (IORef (Item a)) +addToTail (SList {tailList = tailPtrPtr}) x = + do null <- newIORef Null + tailPtr <- readIORef tailPtrPtr + writeIORef tailPtr (Node {val = x, deleted = False, next = null}) + writeIORef tailPtrPtr null + return tailPtr + + +-- the iterator always points to the PREVIOUS node, +-- recall that there's a static dummy new Head +newIterator :: SList a -> IO (Iterator a) +newIterator (SList {headList = hd}) = + do hdPtr <- readIORef hd + it <- newIORef hdPtr + return it + + +-- test if iterators point to same location +testEqIterator :: Iterator a -> Iterator a -> IO Bool +testEqIterator it1 it2 = + do it1Ptr <- readIORef it1 + it2Ptr <- readIORef it2 + return (it1Ptr == it2Ptr) + +-- assign the rhs iterator's current pointer to the lhs iterator +assignIterator :: Iterator a -> Iterator a -> IO () +assignIterator lhs rhs = + do rhsVal <- readIORef rhs + writeIORef lhs rhsVal + + +-- we iterate through the list and return the first "not deleted" node +-- we delink deleted nodes +-- there's no need to adjust headList, tailList +-- cause headList has a static Head and +-- tailList points to Null +iterateSList :: Iterator a -> IO (Maybe (IORef (Item a))) +iterateSList itPtrPtr = + let go prevPtr = + do prevNode <- readIORef prevPtr + let curPtr = next prevNode -- the prev node has always a next field + -- recall the static dummy head node + curNode <- readIORef curPtr + case curNode of + Null -> return Nothing -- reached end of the list + Node { deleted = del, next = nextNode } -> + if not del + then -- node available + do writeIORef itPtrPtr curPtr -- adjust iterator + return (Just curPtr) + else -- delete curNode by setting the next of prevNode to next of curNode + do case prevNode of + Head {} -> writeIORef prevPtr (Head {next = nextNode}) + Node {} -> writeIORef prevPtr + (Node {val = val prevNode, + deleted = deleted prevNode, + next = nextNode}) + go prevPtr + in do startPtr <- readIORef itPtrPtr + go startPtr + + + diff --git a/Actor/Timeout.hs b/Actor/Timeout.hs new file mode 100644 --- /dev/null +++ b/Actor/Timeout.hs @@ -0,0 +1,34 @@ + + +-- found somewhere on the web + +module Actor.Timeout where + +import IO +import Control.Concurrent +import Control.Exception + +parIO :: IO a -> IO a -> IO a +parIO a1 a2 = do { m <- newEmptyMVar ; + c1 <- forkIO (child m a1) ; + c2 <- forkIO (child m a2) ; + r <- takeMVar m ; + throwTo c1 (AsyncException ThreadKilled) ; + throwTo c2 (AsyncException ThreadKilled) ; + return r } + where + child m a = do { r <- a ; putMVar m r } + + +timeout :: Int -> IO a -> IO (Maybe a) +timeout n a = parIO ( do { r <- a ; return (Just r) } ) + ( do { threadDelay n ; return Nothing } ) + + +main = do m <- newEmptyMVar + res <- timeout 100 (readMVar m) + case res of + Nothing -> putStrLn "timeout" + Just _ -> putStrLn "result" + return () + diff --git a/Chain.hs b/Chain.hs new file mode 100644 --- /dev/null +++ b/Chain.hs @@ -0,0 +1,85 @@ + +{-# LANGUAGE MultiParamTypeClasses, PatternSignatures, TypeSynonymInstances, FlexibleInstances #-} + +-- creating some n number of actors which pass around some message + + +module Main where + +import IO +import Monad +import Data.IORef +import Control.Concurrent +import System.Environment +import Data.Time + +import Actor.ActorBase +import Actor.ActorSyntax +import Actor.ActorCompiler +import Actor.ActorLinearSearch + +-- boilerplate + +data Msg = Blah deriving (Eq,Show) + +valhashOp = HashOp { numberOfTables = 1, hashMsg = \ _ -> 1 } + +instance EMatch Msg where + match tags m1 m2 = return (m1 == m2, tags) + +-- auxilliary + +loop c = do { c; loop c} + +ifonly b c = if b then c + else return () + +-- actual actor code + +toPID = actorToPID + +chainedActor idx max exit neighbor self = + do { loop + ( do { --putStrLn ("Actor " ++ show idx ++ ": WAITING"); + receive self + [ [Blah] .->. ( do { --putStrLn ("Actor " ++ show idx ++ ": RECEIVED"); + if idx == max + then putMVar exit () -- EXIT + else send (toPID neighbor) Blah + } + ) + ] + } + ) + } + + + + +createActors :: Int -> Int -> MVar () -> Act Msg -> IO () +createActors idx max exit curr + | idx == max = runActor curr (chainedActor idx max exit curr) + -- last actor won't sent, only performs exit + | idx < max = do { (next :: Act Msg) <- createActor valhashOp + ; runActor curr (chainedActor idx max exit next) + ; createActors (idx+1) max exit next + } + + +main :: IO () +main = do { let n = 50000 + ; exit <- newEmptyMVar + ; start <- getCurrentTime + ; (initialActor :: Act Msg) <- createActor valhashOp + ; createActors 1 n exit initialActor + ; end <- getCurrentTime + ; putStrLn $ "creation time: " ++ show (diffUTCTime end start) + ; send (toPID initialActor) Blah + ; readMVar exit -- blocks till done + ; fin <- getCurrentTime + ; putStrLn $ "message time: " ++ show (diffUTCTime fin end) + ; putStrLn $ "Threads: " ++ show n + } + + + diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,34 @@ + +{- +-------------------------------------------------------------------------------- +-- +-- Copyright (C) 2008 Martin Sulzmann, Edmund Lam. 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 Isaac Jones 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. + +-} diff --git a/MarketPlace.hs b/MarketPlace.hs new file mode 100644 --- /dev/null +++ b/MarketPlace.hs @@ -0,0 +1,334 @@ + + +{-# LANGUAGE MultiParamTypeClasses, PatternSignatures, TypeSynonymInstances, FlexibleInstances #-} + + +-- a market-place using server and client actors + +import Network.Socket hiding (send) +import System.IO +import Control.Exception +import Control.Concurrent +import Control.Concurrent.Chan +import Control.Monad +import Control.Monad.Fix (fix) + + +import Actor.ActorBase +import Actor.ActorSyntax +import Actor.ActorCompiler +import Actor.ActorLinearSearch + + + +-- types of messages + +data MsgClient = Arrived (L String) + -- notifies about newly arrived clients + | DisConnected (L String) + -- disconnected clients + | Sold (L String) + | Bought (L String) + -- notifies about sold/bought items + +valHash_MsgClient = HashOp { numberOfTables = 4, + hashMsg = \ msg -> case msg of + Arrived _ -> 1 + DisConnected _ -> 2 + Sold _ -> 3 + Bought _ -> 4 } + +data MsgServer = + New (L (PID MsgClient)) (L String) + -- informs server about new clients + | Sell (L (PID MsgClient)) (L String) + -- sell request + | Buy (L (PID MsgClient)) (L String) + -- buy request + | Done (L (PID MsgClient)) (L String) + -- tell server I'm done + +valHash_MsgServer = HashOp { numberOfTables = 4, + hashMsg = \ msg -> case msg of + New _ _ -> 1 + Sell _ _ -> 2 + Buy _ _ -> 3 + Done _ _ -> 4 } + +instance Eq MsgClient where + (==) (Arrived x) (Arrived y) = x == y + (==) (DisConnected x) (DisConnected y) = x == y + (==) (Sold x) (Sold y) = x == y + (==) (Bought x) (Bought y) = x == y + (==) _ _ = False + +instance Eq MsgServer where + (==) (New _ x) (New _ y) = x == y + (==) (Sell _ x) (Sell _ y) = x == y + (==) (Buy _ x) (Buy _ y) = x == y + (==) (Done _ x) (Done _ y) = x == y + (==) _ _ = False + + +-- boiler plate + +instance Show MsgClient where + show (Arrived (Val x)) = "Arrived " ++ show x + show (DisConnected (Val x)) = "Disconnected " ++ show x + show (Sold (Val x)) = "Sold " ++ show x + show (Bought (Val x)) = "Bought " ++ show x + +instance Show MsgServer where + show (New _ (Val x)) = "New " ++ show x + show (Sell _ (Val x)) = "Sell " ++ show x + show (Buy _ (Val x)) = "Buy " ++ show x + show (Done _ (Val x)) = "Done " ++ show x + show _ = "vars are not showable" + + +instance EMatch MsgClient where + match tags (Arrived x) (Arrived y) = match tags x y + match tags (DisConnected x) (DisConnected y) = match tags x y + match tags (Sold x) (Sold y) = match tags x y + match tags (Bought x) (Bought y) = match tags x y + match tags _ _ = return (False, tags) + +instance EMatch MsgServer where + match tags (New x1 x2) (New y1 y2) = + do { (b1,t1) <- match tags x1 y1 + ; (b2,t2) <- match t1 x2 y2 + ; return (b1 && b2, t2) } + match tags (Sell x1 x2) (Sell y1 y2) = + do { (b1,t1) <- match tags x1 y1 + ; (b2,t2) <- match t1 x2 y2 + ; return (b1 && b2, t2) } + match tags (Buy x1 x2) (Buy y1 y2) = + do { (b1,t1) <- match tags x1 y1 + ; (b2,t2) <- match t1 x2 y2 + ; return (b1 && b2, t2) } + match tags (Done x1 x2) (Done y1 y2) = + do { (b1,t1) <- match tags x1 y1 + ; (b2,t2) <- match t1 x2 y2 + ; return (b1 && b2, t2) } + match tags _ _ = return (False, tags) + +-- we're only allowed to match a pid +-- but we can't compare two pids +instance EMatch (PID msg) where + match tags _ _ = return (False, tags) + +instance Eq (Chan msg) where + (==) = error "impossible: == among channels" +instance Show (Chan msg) where + show = error "impossible: show for channels" +instance Show (MVar ThreadId) where + show = error "impossible: show for thread id" + + +-- EMatch depends on Eq and Show +instance Eq (Act msg) where + (==) _ _ = False + +instance Show (Act msg) where + show _ = "Actor" + +instance EMatch String where + match tags x y = return (x==y,tags) + +-- client interface +class ArrivedMsg a where + arrivedMsg :: a -> MsgClient + +instance ArrivedMsg (VAR String) where + arrivedMsg x = Arrived (Var x) +instance ArrivedMsg String where + arrivedMsg s = Arrived (Val s) + +class DisConnectedMsg a where + disconnectedMsg :: a -> MsgClient + +instance DisConnectedMsg (VAR String) where + disconnectedMsg (x@(VAR{})) = DisConnected (Var x) +instance DisConnectedMsg String where + disconnectedMsg s = DisConnected (Val s) + +class SoldMsg a where + soldMsg :: a -> MsgClient + +instance SoldMsg (VAR String) where + soldMsg x = Sold (Var x) +instance SoldMsg String where + soldMsg x = Sold (Val x) + +class BoughtMsg a where + boughtMsg :: a -> MsgClient + +instance BoughtMsg (VAR String) where + boughtMsg x = Bought (Var x) +instance BoughtMsg String where + boughtMsg x = Bought (Val x) + +-- server interface + +class NewMsg a b where + newMsg :: a -> b -> MsgServer + +instance NewMsg (VAR (PID MsgClient)) (VAR String) where + newMsg x y = New (Var x) (Var y) +instance NewMsg (VAR (PID MsgClient)) String where + newMsg x y = New (Var x) (Val y) +instance NewMsg (PID MsgClient) (VAR String) where + newMsg x y = New (Val x) (Var y) +instance NewMsg (PID MsgClient) String where + newMsg x y = New (Val x) (Val y) + +class SellMsg a b where + sellMsg :: a -> b -> MsgServer + +instance SellMsg (VAR (PID MsgClient)) (VAR String) where + sellMsg x y = Sell (Var x) (Var y) +instance SellMsg (VAR (PID MsgClient)) String where + sellMsg x y = Sell (Var x) (Val y) +instance SellMsg (PID MsgClient) (VAR String) where + sellMsg x y = Sell (Val x) (Var y) +instance SellMsg (PID MsgClient) String where + sellMsg x y = Sell (Val x) (Val y) + +class BuyMsg a b where + buyMsg :: a -> b -> MsgServer + +instance BuyMsg (VAR (PID MsgClient)) (VAR String) where + buyMsg x y = Buy (Var x) (Var y) +instance BuyMsg (VAR (PID MsgClient)) String where + buyMsg x y = Buy (Var x) (Val y) +instance BuyMsg (PID MsgClient) (VAR String) where + buyMsg x y = Buy (Val x) (Var y) +instance BuyMsg (PID MsgClient) String where + buyMsg x y = Buy (Val x) (Val y) + + +class DoneMsg a b where + doneMsg :: a -> b -> MsgServer + +instance DoneMsg (VAR (PID MsgClient)) (VAR String) where + doneMsg x y = Done (Var x) (Var y) +instance DoneMsg (VAR (PID MsgClient)) String where + doneMsg x y = Done (Var x) (Val y) +instance DoneMsg (PID MsgClient) (VAR String) where + doneMsg x y = Done (Val x) (Var y) +instance DoneMsg (PID MsgClient) String where + doneMsg x y = Done (Val x) (Val y) + + + +-- auxilliary + +loop c = do { c; loop c} + +toPID = actorToPID + +-- each new connection creates a new client + + +eventLoop :: Socket -> Act MsgServer -> IO () +eventLoop sock serverActor = do + conn <- accept sock + forkIO (spawnClient conn serverActor) + eventLoop sock serverActor + +-- client asks for name, registers with server and +-- waits for messages from server + +spawnClient (sock,_) serverActor = + do { (client :: Act MsgClient) <- createActor valHash_MsgClient + ; hdl <- socketToHandle sock ReadWriteMode + ; hSetBuffering hdl NoBuffering + ; hPutStrLn hdl "Hi, what's your name?" + ; name <- liftM init (hGetLine hdl) + ; send (toPID serverActor) (newMsg (toPID client) name) + ; hPutStrLn hdl "Anything you'd like to sell/buy ([Sitem]/[Bitem])?" + ; cmd <- liftM init (hGetLine hdl) + ; case cmd of + ('S':item) -> send (toPID serverActor) (sellMsg (toPID client) item) + ('B':item) -> send (toPID serverActor) (buyMsg (toPID client) item) + _ -> hPutStrLn hdl "Invalid input" + ; x <- newVar :: IO (VAR String) + ; y <- newVar :: IO (VAR String) + ; let goodbye = do { hPutStrLn hdl ("\n Bye!") + ; send (toPID serverActor) (doneMsg (toPID client) name) + ; hClose hdl + } + ; let go = + receive client + [ [arrivedMsg x] .->. + do { v1 <- readVar x + ; hPutStrLn hdl ("\n Arrived: " ++ show v1) + ; go + } + , [disconnectedMsg x] .->. + do { v1 <- readVar x + ; hPutStrLn hdl ("\n Disconnected: " ++ show v1) + ; go + } + , [soldMsg y] .->. + do { v2 <- readVar y + ; hPutStrLn hdl ("\n Sold: " ++ show v2) + ; goodbye + } + , [boughtMsg y] .->. + do { v2 <- readVar y + ; hPutStrLn hdl ("\n Bought: " ++ show v2) + ; goodbye + } + ] + ; go + } + + +-- server, stores new client and informs others + +server self = + do { s <- newVar :: IO (VAR String) + ; c <- newVar :: IO (VAR (PID MsgClient)) + ; c2 <- newVar :: IO (VAR (PID MsgClient)) + ; let go clients = + receive self + [ [newMsg c s] .->. + do { v1 <- readVar s + ; v2 <- readVar c + ; putStrLn ("New user " ++ show v1 ++ " has arrived, will inform others"); + ; mapM (\ other -> send other (arrivedMsg v1)) clients + ; go (v2:clients) + } + , [sellMsg c s, buyMsg c2 s] .->. + do { v1 <- readVar s + ; putStrLn ("Item " ++ show v1 ++ "sold/bought") + ; cv <- readVar c + ; cv2 <- readVar c2 + ; send cv (soldMsg v1) + ; send cv2 (boughtMsg v1) + ; go clients + } + , [doneMsg c s] .->. + do { v1 <- readVar s + ; v2 <- readVar c + ; putStrLn ("User " ++ show v1 ++ "is done, will inform others"); + ; mapM (\ other -> send other (disconnectedMsg v1)) clients + ; go clients + -- we should actually delete c from the list of clients + -- there's no easy way at the moment + } + ] + ; go [] -- initially there are no clients + } + +main :: IO () +main = + do { (srv :: Act MsgServer) <- createActor valHash_MsgServer + ; runActor srv server + ; sock <- socket AF_INET Stream 0 + ; setSocketOption sock ReuseAddr 1 + ; bindSocket sock (SockAddrInet 4242 iNADDR_ANY) + ; listen sock 2 + ; eventLoop sock srv + } diff --git a/PingPong.hs b/PingPong.hs new file mode 100644 --- /dev/null +++ b/PingPong.hs @@ -0,0 +1,96 @@ + +{-# LANGUAGE MultiParamTypeClasses, PatternSignatures, TypeSynonymInstances, FlexibleInstances #-} + +module PingPong where + +-- standard ping-pong actor example + +import IO +import Monad +import Data.IORef +import Control.Concurrent + + +import Actor.ActorBase +import Actor.ActorSyntax +import Actor.ActorCompiler +import Actor.ActorLinearSearch + + + + +-- boilerplate + +data Msg + = Ping | Pong | Stop deriving (Eq,Show) + + +valHashOp_Msg = HashOp {numberOfTables = 3, + hashMsg = \ msg -> case msg of + Ping -> 1 + Pong -> 2 + Stop -> 3 } + + +instance EMatch Msg where + match tags m1 m2 = return (m1 == m2, tags) + + +-- auxilliary + +loop c = do { c; loop c} + +ifonly b c = if b then c + else return () + + +--wait = 2 +wait = 10000 + +pingAct count pong self = + do { pingsLeft <- newIORef (count -1) + ; let pong_pid = actorToPID pong + ; send pong_pid Ping + ; loop + ( receive self + [ [Pong] .->. ( do { v <- readIORef pingsLeft + ; ifonly (v `mod` wait == 0) (putStrLn "ping:Pong") + ; if v > 0 + then do {send pong_pid Ping; writeIORef pingsLeft (v-1) } + else do {putStrLn "ping:Stop" ; send pong_pid Stop } + } + ) + ] + ) + } + +pongAct pong exit self = + do { pongCount <- newIORef 0 + ; let pong_pid = actorToPID pong + ; loop + ( receive self + [ [Ping] .->. ( do { v <- readIORef pongCount + ; ifonly (v `mod` wait == 0) (putStrLn ("pong:Ping " ++ (show v))) + ; send pong_pid Pong + ; writeIORef pongCount (v+1) + } + ) + , + [Stop] .->. ( do { putStrLn "pong: Stop" + ; putMVar exit () -- EXIT + } + ) + ] + ) + } + + +main :: IO () +main = do { (ping :: Act Msg) <- createActor valHashOp_Msg + ; (pong :: Act Msg) <- createActor valHashOp_Msg + ; exit <- newEmptyMVar + ; runActor ping (pingAct 1000000 pong) + ; runActor pong (pongAct ping exit) + ; readMVar exit -- blocks till we hit exit + ; kill (actorToPID ping) ; kill (actorToPID pong) + } diff --git a/README b/README new file mode 100644 --- /dev/null +++ b/README @@ -0,0 +1,15 @@ + +The Haskell implementation of + +Actors with Multi-Headed Message Receive Patterns +Martin Sulzmann (martin.sulzmann@gmail.com) and Edmund Lam and Peter van Weert +In Coordination'08 + + +Installation via cabal: +runhaskell Setup.hs configure --ghc --user --prefix=$HOME +runhaskell Setup.hs build +runhaskell Setup.hs install + +Check out the examples (extra-source-files) for how to use the library extension. +I'll add more information later. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,4 @@ + +import Distribution.Simple + +main = defaultMain diff --git a/actor.cabal b/actor.cabal new file mode 100644 --- /dev/null +++ b/actor.cabal @@ -0,0 +1,36 @@ + +Name: actor +Synopsis: Actors with multi-headed receive clauses +Category: Concurrency +Version: 0.1 +Stability: experimental +Cabal-Version: >= 1.2 +License: BSD3 +License-File: LICENSE +Author: Martin Sulzmann +Maintainer: martin.sulzmann@gmail.com +Description: A Haskell library implementing + "Actors with Multi-Headed Message Receive Patterns" + COORDINATION'08 +homepage: http://sulzmann.blogspot.com/2008/10/actors-with-multi-headed-receive.html + +data-files: README +extra-source-files: PingPong.hs, Chain.hs, MarketPlace.hs + +Build-Type: Simple +Build-Depends: base, haskell98, stm, time +Exposed-modules: + Actor.ActorBase, + Actor.ActorSyntax, + Actor.ActorCompiler, + Actor.ActorLinearSearch +Other-modules: Actor.SList, + Actor.QList, + Actor.Timeout + +Extensions: UndecidableInstances, FlexibleInstances, TypeSynonymInstances + MultiParamTypeClasses, FunctionalDependencies + +GHC-Options: -threaded + +tested-with: GHC == 6.8.2