dc2e1d8 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Start the CH layer with the new transport

dc2e1d8Edsko de Vries14 years ago

distributed-process/distributed-process.cabal+5−4

compare with master

unifiedsplit

@@ -1,9 +1,9 @@

1Name: distributed-process

2−Version: 0.0.1

2+Version: 0.1.0

3Description: Distributed processes for message based concurrency.

4License: BSD3

5License-file: LICENSE

6−Author: Duncan Coutts, Nicolas Wu

6+Author: Duncan Coutts, Nicolas Wu, Edsko de Vries

7Maintainer: duncan@well-typed.com

8Homepage: http://github.com/haskell-distributed/

9Build-Type: Simple

@@ -14,8 +14,9 @@ Library

14 bytestring >= 0.9,

15 containers >= 0.4,

16 mtl >= 2.0,

17+ binary,

17− network-transport >= 0.0

18+ network-transport >= 0.1

18 Exposed-modules: Control.Distributed.Process

19 Exposed-modules: Control.Distributed.Process

19− Extensions: ExplicitForAll ScopedTypeVariables

20+ Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances

20 ghc-options: -Wall

21 ghc-options: -Wall

21 HS-Source-Dirs: src

22 HS-Source-Dirs: src

distributed-process/src/Control/Distributed/Process.hs+35−29

@@ -1,3 +1,37 @@

1+module Control.Distributed.Process ( expect ) where

3+import Data.ByteString.Lazy (ByteString)

4+import Data.Typeable (Typeable(typeOf))

5+import Data.Binary (decode)

6+import Control.Monad.Reader (MonadReader(..), ReaderT, liftIO)

7+import Control.Applicative ((<$>))

8+import Control.Distributed.Process.Internal.CQueue ( CQueue

9+ , dequeueMatching

10+ )

11+import Control.Distributed.Process.Serializable ( Serializable

12+ , Fingerprint

13+ , encodeFingerprint

14+ , fingerprint

15+ )

16+

17+data Message = Message { messageFingerprint :: Fingerprint

18+ , messageEncoding :: ByteString

19+ }

20+

21+data ProcessState = ProcessState { processQueue :: CQueue Message }

22+

23+newtype Process a = Process (ReaderT ProcessState IO a)

24+ deriving (Monad)

25+

26+expect :: forall a. Serializable a => Process a

27+expect = Process $ do

28+ queue <- processQueue <$> ask

29+ let fp = fingerprint (undefined :: a)

30+ msg <- liftIO $ dequeueMatching queue ((== fp) . messageFingerprint)

31+ return (decode . messageEncoding $ msg)

32+

33+

34+{-

1{-# LANGUAGE CPP, ExplicitForAll, ScopedTypeVariables #-}

35{-# LANGUAGE CPP, ExplicitForAll, ScopedTypeVariables #-}

3module Control.Distributed.Process (

37module Control.Distributed.Process (

@@ -198,33 +232,5 @@ monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()

198

232

199-}

233-}

200

234

201−-- Concurrent queue for single reader, single writer

202--

235--

203−data CQueue a = CQueue (MVar [a]) -- arrived

236+-}

204− (Chan a) -- incomming

205−

206−newCQueue :: IO (CQueue a)

207−newCQueue = do

208− arrived <- newMVar []

209− incomming <- newChan

210− return (CQueue arrived incomming)

211−

212−enqueue :: CQueue a -> a -> IO ()

213−enqueue (CQueue _arrived incomming) a = writeChan incomming a

214−

215−dequeueMatching :: forall a. CQueue a -> (a -> Bool) -> IO a

216−dequeueMatching (CQueue arrived incomming) matches = do

217− modifyMVar arrived (checkArrived [])

218− where

219− checkArrived :: [a] -> [a] -> IO ([a], a)

220− checkArrived xs' [] = checkIncomming xs'

221− checkArrived xs' (x:xs)

222− | matches x = return (reverse xs' ++ xs, x)

223− | otherwise = checkArrived (x:xs') xs

224−

225− checkIncomming :: [a] -> IO ([a], a)

226− checkIncomming xs' = do

227− x <- readChan incomming

228− if matches x

229− then return (reverse xs', x)

230− else checkIncomming (x:xs')

distributed-process/src/Control/Distributed/Process/Internal/CQueue.hsnew file+39

@@ -0,0 +1,39 @@

1+-- | Concurrent queue for single reader, single writer

2+module Control.Distributed.Process.Internal.CQueue

3+ ( CQueue

4+ , newCQueue

5+ , enqueue

6+ , dequeueMatching

7+ ) where

9+import Control.Concurrent.MVar (MVar, newMVar, modifyMVar)

10+import Control.Concurrent.Chan (Chan, newChan, writeChan, readChan)

11+

12+data CQueue a = CQueue (MVar [a]) -- arrived

13+ (Chan a) -- incoming

14+

15+newCQueue :: IO (CQueue a)

16+newCQueue = do

17+ arrived <- newMVar []

18+ incoming <- newChan

19+ return (CQueue arrived incoming)

20+

21+enqueue :: CQueue a -> a -> IO ()

22+enqueue (CQueue _arrived incoming) = writeChan incoming

23+

24+dequeueMatching :: forall a. CQueue a -> (a -> Bool) -> IO a

25+dequeueMatching (CQueue arrived incoming) matches =

26+ modifyMVar arrived (checkArrived [])

27+ where

28+ checkArrived :: [a] -> [a] -> IO ([a], a)

29+ checkArrived xs' [] = checkIncomming xs'

30+ checkArrived xs' (x:xs)

31+ | matches x = return (reverse xs' ++ xs, x)

32+ | otherwise = checkArrived (x:xs') xs

33+

34+ checkIncomming :: [a] -> IO ([a], a)

35+ checkIncomming xs' = do

36+ x <- readChan incoming

37+ if matches x

38+ then return (reverse xs', x)

39+ else checkIncomming (x:xs')

distributed-process/src/Control/Distributed/Process/Serializable.hsnew file+29

@@ -0,0 +1,29 @@

1+module Control.Distributed.Process.Serializable

2+ ( Serializable

3+ , encodeFingerprint

4+ , fingerprint

5+ , Fingerprint

6+ ) where

8+import Data.Binary (Binary)

9+import Data.Typeable (Typeable(..))

10+import Data.Typeable.Internal (TypeRep(TypeRep))

11+import GHC.Fingerprint.Type (Fingerprint(Fingerprint))

12+import Data.ByteString (ByteString)

13+import qualified Data.ByteString.Internal as BSI (unsafeCreate)

14+import Foreign.Storable (pokeByteOff)

15+

16+class (Binary a, Typeable a) => Serializable a

17+instance (Binary a, Typeable a) => Serializable a

18+

19+-- | Encode type representation as a bytestring

20+encodeFingerprint :: Fingerprint -> ByteString

21+encodeFingerprint (Fingerprint hi lo) =

22+ -- Since all CH nodes will run precisely the same binary, we don't have to

23+ -- worry about cross-arch issues here (like endianness)

24+ BSI.unsafeCreate 16 $ \p -> do

25+ pokeByteOff p 0 hi

26+ pokeByteOff p 8 lo

27+

28+fingerprint :: Typeable a => a -> Fingerprint

29+fingerprint a = let TypeRep fp _ _ = typeOf a in fp