actor (empty) → 0.1
raw patch · 14 files changed
+1777/−0 lines, 14 filesdep +basedep +haskell98dep +stmsetup-changed
Dependencies added: base, haskell98, stm, time
Files
- Actor/ActorBase.hs +150/−0
- Actor/ActorCompiler.hs +352/−0
- Actor/ActorLinearSearch.hs +266/−0
- Actor/ActorSyntax.hs +103/−0
- Actor/QList.hs +126/−0
- Actor/SList.hs +142/−0
- Actor/Timeout.hs +34/−0
- Chain.hs +85/−0
- LICENSE +34/−0
- MarketPlace.hs +334/−0
- PingPong.hs +96/−0
- README +15/−0
- Setup.hs +4/−0
- actor.cabal +36/−0
+ Actor/ActorBase.hs view
@@ -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) }+++++
+ Actor/ActorCompiler.hs view
@@ -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 !!!!!+ }+ }+
+ Actor/ActorLinearSearch.hs view
@@ -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 ()++
+ Actor/ActorSyntax.hs view
@@ -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]+
+ Actor/QList.hs view
@@ -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+++
+ Actor/SList.hs view
@@ -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+++
+ Actor/Timeout.hs view
@@ -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 ()+
+ Chain.hs view
@@ -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+ }+++
+ LICENSE view
@@ -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.++-}
+ MarketPlace.hs view
@@ -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+ }
+ PingPong.hs view
@@ -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)+ }
+ README view
@@ -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.
+ Setup.hs view
@@ -0,0 +1,4 @@++import Distribution.Simple++main = defaultMain
+ actor.cabal view
@@ -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