distributed-process-extras (empty) → 0.1.0

raw patch · 17 files changed

+2259/−0 lines, 17 filesdep +HUnitdep +QuickCheckdep +ansi-terminalsetup-changed

Dependencies added: HUnit, QuickCheck, ansi-terminal, base, binary, containers, deepseq, derive, distributed-process, distributed-process-extras, distributed-process-tests, fingertree, ghc-prim, hashable, mtl, network-transport, network-transport-tcp, rematch, stm, template-haskell, test-framework, test-framework-hunit, test-framework-quickcheck2, time, transformers, uniplate, unordered-containers

Files

• LICENCE +30/−0
• Setup.lhs +3/−0
• distributed-process-extras.cabal +112/−0
• src/Control/Concurrent/Utils.hs +65/−0
• src/Control/Distributed/Process/Extras.hs +124/−0
• src/Control/Distributed/Process/Extras/Call.hs +243/−0
• src/Control/Distributed/Process/Extras/Internal/Containers/MultiMap.hs +95/−0
• src/Control/Distributed/Process/Extras/Internal/Primitives.hs +290/−0
• src/Control/Distributed/Process/Extras/Internal/Queue/PriorityQ.hs +38/−0
• src/Control/Distributed/Process/Extras/Internal/Queue/SeqQ.hs +66/−0
• src/Control/Distributed/Process/Extras/Internal/Types.hs +277/−0
• src/Control/Distributed/Process/Extras/Internal/Unsafe.hs +128/−0
• src/Control/Distributed/Process/Extras/Time.hs +258/−0
• src/Control/Distributed/Process/Extras/Timer.hs +180/−0
• src/Control/Distributed/Process/Extras/UnsafePrimitives.hs +63/−0
• tests/TestQueues.hs +91/−0
• tests/TestTimer.hs +196/−0

@@ -0,0 +1,30 @@+Copyright Tim Watson, 2012-2013.++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 the author 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.

@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+> import Distribution.Simple+> main = defaultMain

@@ -0,0 +1,112 @@+name:           distributed-process-extras+version:        0.1.0+cabal-version:  >=1.8+build-type:     Simple+license:        BSD3+license-file:   LICENCE+stability:      experimental+Copyright:      Tim Watson 2012 - 2013+Author:         Tim Watson+Maintainer:     watson.timothy@gmail.com+Stability:      experimental+Homepage:       http://github.com/haskell-distributed/distributed-process-extras+Bug-Reports:    https://cloud-haskell.atlassian.net+synopsis:       Cloud Haskell Extras+description:    Supporting library, providing common types and utilities used by the+                various components that make up the distributed-process-platform package.+category:       Control+tested-with:    GHC == 7.4.2 GHC == 7.6.2+data-dir:       ""++source-repository head+  type:      git+  location:  https://github.com/haskell-distributed/distributed-process-extras++flag perf+  description: Build with profiling enabled+  default: False++library+  build-depends:+                   base >= 4.4 && < 5,+                   distributed-process >= 0.5.2 && < 0.6,+                   binary >= 0.6.3.0 && < 0.8,+                   deepseq >= 1.3.0.1 && < 1.4,+                   mtl,+                   containers >= 0.4 && < 0.6,+                   hashable >= 1.2.0.5 && < 1.3,+                   unordered-containers >= 0.2.3.0 && < 0.3,+                   fingertree < 0.2,+                   stm >= 2.4 && < 2.5,+                   time > 1.4 && < 1.5,+                   transformers+  if impl(ghc <= 7.5) +    Build-Depends:   template-haskell == 2.7.0.0,+                     derive == 2.5.5,+                     uniplate == 1.6.12,+                     ghc-prim+  extensions:      CPP+  hs-source-dirs:   src+  ghc-options:      -Wall+  exposed-modules:+                   Control.Distributed.Process.Extras,+                   Control.Distributed.Process.Extras.Call,+                   Control.Distributed.Process.Extras.Time,+                   Control.Distributed.Process.Extras.Timer,+                   Control.Distributed.Process.Extras.UnsafePrimitives,+                   Control.Concurrent.Utils,+                   Control.Distributed.Process.Extras.Internal.Containers.MultiMap,+                   Control.Distributed.Process.Extras.Internal.Primitives,+                   Control.Distributed.Process.Extras.Internal.Types,+                   Control.Distributed.Process.Extras.Internal.Queue.SeqQ,+                   Control.Distributed.Process.Extras.Internal.Queue.PriorityQ+                   Control.Distributed.Process.Extras.Internal.Unsafe++test-suite InternalQueueTests+  type:            exitcode-stdio-1.0+  x-uses-tf:       true+  build-depends:+                   base >= 4.4 && < 5,+                   ansi-terminal >= 0.5 && < 0.6,+                   distributed-process >= 0.5.2 && < 0.6,+                   distributed-process-extras,+                   distributed-process-tests >= 0.4.1 && < 0.5,+                   HUnit >= 1.2 && < 2,+                   test-framework >= 0.6 && < 0.9,+                   test-framework-hunit,+                   QuickCheck >= 2.4,+                   test-framework-quickcheck2,+                   rematch >= 0.2.0.0,+                   ghc-prim+  hs-source-dirs:+                   tests+  ghc-options:     -Wall -rtsopts+  extensions:      CPP+  main-is:         TestQueues.hs+  cpp-options:     -DTESTING++test-suite TimerTests+  type:            exitcode-stdio-1.0+  x-uses-tf:       true+  build-depends:+                   base >= 4.4 && < 5,+                   ansi-terminal >= 0.5 && < 0.6,+                   deepseq >= 1.3.0.1 && < 1.4,+                   distributed-process >= 0.5.2 && < 0.6,+                   distributed-process-extras,+                   distributed-process-tests >= 0.4.1 && < 0.5,+                   network-transport >= 0.4 && < 0.5,+                   network-transport-tcp >= 0.4 && < 0.5,+                   HUnit >= 1.2 && < 2,+                   test-framework >= 0.6 && < 0.9,+                   test-framework-hunit,+                   QuickCheck >= 2.4,+                   test-framework-quickcheck2,+                   rematch >= 0.2.0.0,+                   ghc-prim+  hs-source-dirs:+                   tests+  ghc-options:     -Wall -rtsopts+  extensions:      CPP+  main-is:         TestTimer.hs+  cpp-options:     -DTESTING

@@ -0,0 +1,65 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables   #-}++module Control.Concurrent.Utils+  ( Lock()+  , Exclusive(..)+  , Synchronised(..)+  , withLock+  ) where++import Control.Distributed.Process+  ( Process+  )+import qualified Control.Distributed.Process as Process (catch)+import Control.Exception (SomeException, throw)+import qualified Control.Exception as Exception (catch)+import Control.Concurrent.MVar+  ( MVar+  , tryPutMVar+  , newMVar+  , takeMVar+  )+import Control.Monad.IO.Class (MonadIO, liftIO)++newtype Lock = Lock { mvar :: MVar () }++class Exclusive a where+  new          :: IO a+  acquire      :: (MonadIO m) => a -> m ()+  release      :: (MonadIO m) => a -> m ()++instance Exclusive Lock where+  new       = return . Lock =<< newMVar ()+  acquire   = liftIO . takeMVar . mvar+  release l = liftIO (tryPutMVar (mvar l) ()) >> return ()++class Synchronised e m where+  synchronised :: (Exclusive e, Monad m) => e -> m b -> m b++  synchronized :: (Exclusive e, Monad m) => e -> m b -> m b+  synchronized = synchronised++instance Synchronised Lock IO where+  synchronised = withLock++instance Synchronised Lock Process where+  synchronised = withLockP++withLockP :: (Exclusive e) => e -> Process a -> Process a+withLockP excl act = do+  Process.catch (do { liftIO $ acquire excl+                    ; result <- act+                    ; liftIO $ release excl+                    ; return result+                    })+                (\(e :: SomeException) -> (liftIO $ release excl) >> throw e)++withLock :: (Exclusive e) => e -> IO a -> IO a+withLock excl act = do+  Exception.catch (do { acquire excl+                      ; result <- act+                      ; release excl+                      ; return result+                      })+                  (\(e :: SomeException) -> release excl >> throw e)

@@ -0,0 +1,124 @@+{- | [Cloud Haskell Extras]++[Evaluation Strategies and Support for NFData]++When sending messages to a local process (i.e., intra-node), the default+approach is to encode (i.e., serialise) the message /anyway/, just to+ensure that no unevaluated thunks are passed to the receiver.+In distributed-process, you must explicitly choose to use /unsafe/ primitives+that do nothing to ensure evaluation, since this might cause an error in the+receiver which would be difficult to debug. Using @NFData@, it is possible+to force evaluation, but there is no way to ensure that both the @NFData@+and @Binary@ instances do so in the same way (i.e., to the same depth, etc)+therefore automatic use of @NFData@ is not possible in distributed-process.++By contrast, distributed-process-platform makes extensive use of @NFData@+to force evaluation (and avoid serialisation overheads during intra-node+communication), via the @NFSerializable@ type class. This does nothing to+fix the potential disparity between @NFData@ and @Binary@ instances, so you+should verify that your data is being handled as expected (e.g., by sticking+to strict fields, or some such) and bear in mind that things could go wrong.++The @UnsafePrimitives@ module in /this/ library will force evaluation before+calling the @UnsafePrimitives@ in distributed-process, which - if you've+vetted everything correctly - should provide a bit more safety, whilst still+keeping performance at an acceptable level.++Users of the various service and utility models (such as @ManagedProcess@ and+the @Service@ and @Task@ APIs) should consult the sub-system specific+documentation for instructions on how to utilise these features.++IMPORTANT NOTICE: Despite the apparent safety of forcing evaluation before+sending, we /still/ cannot make any actual guarantees about the evaluation+semantics of these operations, and therefore the /unsafe/ moniker will remain+in place, in one form or another, for all functions and modules that use them.++[Error/Exception Handling]++It is /important/ not to be too general when catching exceptions in+cloud haskell application, because asynchonous exceptions provide cloud haskell+with its process termination mechanism. Two exception types in particular,+signal the instigator's intention to stop a process immediately, which are+raised (i.e., thrown) in response to the @kill@ and @exit@ primitives provided+by the base distributed-process package.++You should generally try to keep exception handling code to the lowest (i.e.,+most specific) scope possible. If you wish to trap @exit@ signals, use the+various flavours of @catchExit@ primitive from distributed-process.++-}+module Control.Distributed.Process.Extras+  (+    -- * Exported Types+    Addressable+  , sendToRecipient+  , Resolvable(..)+  , Routable(..)+  , Linkable(..)+  , Killable(..)+  , NFSerializable+  , Recipient(..)+  , Shutdown(..)+  , ExitReason(..)+  , CancelWait(..)+  , ServerDisconnected(..)+  , Channel+  , Tag+  , TagPool++    -- * Primitives overriding those in distributed-process+  , monitor+  , module Control.Distributed.Process.Extras.UnsafePrimitives++    -- * Utilities and Extended Primitives+  , spawnSignalled+  , spawnLinkLocal+  , spawnMonitorLocal+  , linkOnFailure+  , times+  , isProcessAlive+  , matchCond+  , deliver+  , awaitExit+  , awaitResponse++    -- * Call/Tagging support+  , newTagPool+  , getTag++    -- * Registration and Process Lookup+  , whereisOrStart+  , whereisOrStartRemote++    -- remote call table+  , __remoteTable+  ) where++import Control.Distributed.Process (RemoteTable)+import Control.Distributed.Process.Extras.Internal.Types+  ( NFSerializable+  , sendToRecipient+  , NFSerializable+  , Recipient(..)+  , Shutdown(..)+  , ExitReason(..)+  , CancelWait(..)+  , ServerDisconnected(..)+  , Channel+  , Tag+  , TagPool+  , newTagPool+  , getTag+  )+import Control.Distributed.Process.Extras.UnsafePrimitives+import Control.Distributed.Process.Extras.Internal.Primitives hiding (__remoteTable)+import qualified Control.Distributed.Process.Extras.Internal.Primitives (__remoteTable)+import qualified Control.Distributed.Process.Extras.Internal.Types      (__remoteTable)++-- remote table++__remoteTable :: RemoteTable -> RemoteTable+__remoteTable =+  Control.Distributed.Process.Extras.Internal.Primitives.__remoteTable .+  Control.Distributed.Process.Extras.Internal.Types.__remoteTable+

@@ -0,0 +1,243 @@+{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Extras.Call+-- Copyright   :  (c) Parallel Scientific (Jeff Epstein) 2012+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainers :  Jeff Epstein, Tim Watson+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- This module provides a facility for Remote Procedure Call (rpc) style+-- interactions with Cloud Haskell processes.+--+-- Clients make synchronous calls to a running process (i.e., server) using the+-- 'callAt', 'callTimeout' and 'multicall' functions. Processes acting as the+-- server are constructed using Cloud Haskell's 'receive' family of primitives+-- and the 'callResponse' family of functions in this module.+-----------------------------------------------------------------------------++module Control.Distributed.Process.Extras.Call+  ( -- client API+    callAt+  , callTimeout+  , multicall+    -- server API+  , callResponse+  , callResponseIf+  , callResponseDefer+  , callResponseDeferIf+  , callForward+  , callResponseAsync+  ) where++import Control.Distributed.Process+import Control.Distributed.Process.Serializable (Serializable)+import Control.Monad (forM, forM_, join)+import Data.List (delete)+import qualified Data.Map as M+import Data.Maybe (listToMaybe)+import Data.Binary (Binary,get,put)+import Data.Typeable (Typeable)++import Control.Distributed.Process.Extras hiding (monitor, send)+import Control.Distributed.Process.Extras.Time++----------------------------------------------+-- * Multicall+----------------------------------------------++-- | Sends a message of type a to the given process, to be handled by a+-- corresponding callResponse... function, which will send back a message of+-- type b. The tag is per-process unique identifier of the transaction. If the+-- timeout expires or the target process dies, Nothing will be returned.+callTimeout :: (Serializable a, Serializable b)+            => ProcessId -> a -> Tag -> Timeout -> Process (Maybe b)+callTimeout pid msg tag time =+  do res <- multicall [pid] msg tag time+     return $ join (listToMaybe res)++-- | Like 'callTimeout', but with no timeout.+-- Returns Nothing if the target process dies.+callAt :: (Serializable a, Serializable b)+       => ProcessId -> a -> Tag -> Process (Maybe b)+callAt pid msg tag = callTimeout pid msg tag infiniteWait++-- | Like 'callTimeout', but sends the message to multiple+-- recipients and collects the results.+multicall :: forall a b.(Serializable a, Serializable b)+             => [ProcessId] -> a -> Tag -> Timeout -> Process [Maybe b]+multicall nodes msg tag time =+  do caller <- getSelfPid+     receiver <- spawnLocal $+         do receiver_pid <- getSelfPid+            mon_caller <- monitor caller+            () <- expect+            monitortags <- forM nodes monitor+            forM_ nodes $ \node -> send node (Multicall, node,+                                              receiver_pid, tag, msg)+            maybeTimeout time tag receiver_pid+            results <- recv nodes monitortags mon_caller+            send caller (MulticallResponse,tag,results)+     mon_receiver <- monitor receiver+     send receiver ()+     receiveWait [+         matchIf (\(MulticallResponse,mtag,_) -> mtag == tag)+                 (\(MulticallResponse,_,val) -> return val),+         matchIf (\(ProcessMonitorNotification ref _pid reason)+                  -> ref == mon_receiver && reason /= DiedNormal)+                 (\_ -> error "multicall: unexpected termination of worker")+       ]+  where+    recv nodes' monitortags mon_caller = do+      resultmap <- recv1 mon_caller+                         (nodes', monitortags, M.empty) :: Process (M.Map ProcessId b)+      return $ ordered nodes' resultmap++    ordered []     _ = []+    ordered (x:xs) m = M.lookup x m : ordered xs m++    recv1 _   ([],_,results) = return results+    recv1 _   (_,[],results) = return results+    recv1 ref (nodesleft,monitortagsleft,results) =+          receiveWait [+              matchIf (\(ProcessMonitorNotification ref' _ _)+                         -> ref' == ref)+                      (\_ -> return Nothing)+            , matchIf (\(ProcessMonitorNotification ref' pid reason) ->+                      ref' `elem` monitortagsleft &&+                      pid `elem` nodesleft+                      && reason /= DiedNormal)+                      (\(ProcessMonitorNotification ref' pid _reason) ->+                        return $ Just (delete pid nodesleft,+                                       delete ref' monitortagsleft, results))+            , matchIf (\(MulticallResponse, mtag, _, _) -> mtag == tag)+                      (\(MulticallResponse, _, responder, msgx) ->+                        return $ Just (delete responder nodesleft,+                                       monitortagsleft,+                                       M.insert responder (msgx :: b) results))+            , matchIf (\(TimeoutNotification mtag) -> mtag == tag )+                      (\_ -> return Nothing)+          ]+          >>= maybe (return results) (recv1 ref)++data MulticallResponseType a =+         MulticallAccept+       | MulticallForward ProcessId a+       | MulticallReject deriving Eq++callResponseImpl :: (Serializable a,Serializable b)+                 => (a -> MulticallResponseType c) ->+                    (a -> (b -> Process())-> Process c) -> Match c+callResponseImpl cond proc =+  matchIf (\(Multicall,_responder,_,_,msg) ->+            case cond msg of+              MulticallReject -> False+              _               -> True)+          (\wholemsg@(Multicall,responder,sender,tag,msg) ->+            case cond msg of+              -- TODO: sender should get a ProcessMonitorNotification if+              -- our target dies, or we should link to it (?)+              MulticallForward target ret -> send target wholemsg >> return ret+              -- TODO: use `die Reason` when issue #110 is resolved+              MulticallReject -> error "multicallResponseImpl: Indecisive condition"+              MulticallAccept ->+                let resultSender tosend =+                      send sender (MulticallResponse,+                                   tag::Tag,+                                   responder::ProcessId,+                                   tosend)+                in proc msg resultSender)++-- | Produces a Match that can be used with the 'receiveWait' family of+-- message-receiving functions. @callResponse@ will respond to a message of+-- type a sent by 'callTimeout', and will respond with a value of type b.+callResponse :: (Serializable a,Serializable b)+                => (a -> Process (b,c)) -> Match c+callResponse = callResponseIf (const True)++callResponseDeferIf  :: (Serializable a,Serializable b)+                     => (a -> Bool)+                     -> (a -> (b -> Process()) -> Process c)+                     -> Match c+callResponseDeferIf cond =+  callResponseImpl (\msg ->+                     if cond msg+                     then MulticallAccept+                     else MulticallReject)++callResponseDefer  :: (Serializable a,Serializable b)+                   => (a -> (b -> Process())-> Process c) -> Match c+callResponseDefer = callResponseDeferIf (const True)++-- | Produces a Match that can be used with the 'receiveWait' family of+-- message-receiving functions. When calllForward receives a message of type+-- from from 'callTimeout' (and similar), it will forward the message to another+-- process, who will be responsible for responding to it. It is the user's+-- responsibility to ensure that the forwarding process is linked to the+-- destination process, so that if it fails, the sender will be notified.+callForward :: Serializable a => (a -> (ProcessId, c)) -> Match c+callForward proc =+   callResponseImpl+     (\msg -> let (pid, ret) = proc msg+              in MulticallForward pid ret )+     (\_ sender ->+       (sender::(() -> Process ())) `mention`+          error "multicallForward: Indecisive condition")++-- | The message handling code is started in a separate thread. It's not+-- automatically linked to the calling thread, so if you want it to be+-- terminated when the message handling thread dies, you'll need to call+-- link yourself.+callResponseAsync :: (Serializable a,Serializable b)+                  => (a -> Maybe c) -> (a -> Process b) -> Match c+callResponseAsync cond proc =+   callResponseImpl+         (\msg ->+            case cond msg of+              Nothing -> MulticallReject+              Just _ -> MulticallAccept)+         (\msg sender ->+            do _ <- spawnLocal $ -- TODO linkOnFailure to spawned procss+                 do val <- proc msg+                    sender val+               case cond msg of+                 Nothing -> error "multicallResponseAsync: Indecisive condition"+                 Just ret -> return ret )++callResponseIf :: (Serializable a,Serializable b)+               => (a -> Bool) -> (a -> Process (b,c)) -> Match c+callResponseIf cond proc =+    callResponseImpl+             (\msg ->+                 case cond msg of+                   True -> MulticallAccept+                   False -> MulticallReject)+             (\msg sender ->+                 do (tosend,toreturn) <- proc msg+                    sender tosend+                    return toreturn)++maybeTimeout :: Timeout -> Tag -> ProcessId -> Process ()+maybeTimeout Nothing _ _ = return ()+maybeTimeout (Just time) tag p = timeout time tag p++----------------------------------------------+-- * Private types+----------------------------------------------++mention :: a -> b -> b+mention _a b = b++data Multicall = Multicall+       deriving (Typeable)+instance Binary Multicall where+       get = return Multicall+       put _ = return ()+data MulticallResponse = MulticallResponse+       deriving (Typeable)+instance Binary MulticallResponse where+       get = return MulticallResponse+       put _ = return ()

@@ -0,0 +1,95 @@+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE ExistentialQuantification  #-}+{-# LANGUAGE FlexibleInstances          #-}+{-# LANGUAGE UndecidableInstances       #-}++module Control.Distributed.Process.Extras.Internal.Containers.MultiMap+  ( MultiMap+  , Insertable+  , empty+  , insert+  , member+  , lookup+  , filter+  , filterWithKey+  , toList+  ) where++import qualified Data.Foldable as Foldable++import Data.Hashable+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as Map+import Data.HashSet (HashSet)+import qualified Data.HashSet as Set+import Data.Foldable (Foldable(..))+import Prelude hiding (lookup, filter, pred)++-- | Class of things that can be inserted in a map or+-- a set (of mapped values), for which instances of+-- @Eq@ and @Hashable@ must be present.+--+class (Eq a, Hashable a) => Insertable a+instance (Eq a, Hashable a) => Insertable a++-- | Opaque type of MultiMaps.+data MultiMap k v = M { hmap :: !(HashMap k (HashSet v)) }++-- instance Foldable++instance Foldable (MultiMap k) where+  foldr f = foldrWithKey (const f)++empty :: MultiMap k v+empty = M $ Map.empty++insert :: forall k v. (Insertable k, Insertable v)+       => k -> v -> MultiMap k v -> MultiMap k v+insert k' v' M{..} =+  case Map.lookup k' hmap of+    Nothing -> M $ Map.insert k' (Set.singleton v') hmap+    Just s  -> M $ Map.insert k' (Set.insert v' s) hmap+{-# INLINE insert #-}++member :: (Insertable k) => k -> MultiMap k a -> Bool+member k = Map.member k . hmap++lookup :: (Insertable k) => k -> MultiMap k v -> Maybe [v]+lookup k M{..} = maybe Nothing (Just . Foldable.toList) $ Map.lookup k hmap+{-# INLINE lookup #-}++filter :: forall k v. (Insertable k)+       => (v -> Bool)+       -> MultiMap k v+       -> MultiMap k v+filter p M{..} = M $ Map.foldlWithKey' (matchOn p) hmap hmap+  where+    matchOn pred acc key valueSet =+      Map.insert key (Set.filter pred valueSet) acc+{-# INLINE filter #-}++filterWithKey :: forall k v. (Insertable k)+              => (k -> v -> Bool)+              -> MultiMap k v+              -> MultiMap k v+filterWithKey p M{..} = M $ Map.foldlWithKey' (matchOn p) hmap hmap+  where+    matchOn pred acc key valueSet =+      Map.insert key (Set.filter (pred key) valueSet) acc+{-# INLINE filterWithKey #-}++-- | /O(n)/ Reduce this map by applying a binary operator to all+-- elements, using the given starting value (typically the+-- right-identity of the operator).+foldrWithKey :: (k -> v -> a -> a) -> a -> MultiMap k v -> a+foldrWithKey f a M{..} =+  let wrap = \k' v' acc' -> f k' v' acc'+  in Map.foldrWithKey (\k v acc -> Set.foldr (wrap k) acc v) a hmap+{-# INLINE foldrWithKey #-}++toList :: MultiMap k v -> [(k, v)]+toList M{..} = Map.foldlWithKey' explode [] hmap+  where+    explode xs k vs = Set.foldl' (\ys v -> ((k, v):ys)) xs vs+{-# INLINE toList #-}+

@@ -0,0 +1,290 @@+{-# LANGUAGE DeriveDataTypeable    #-}+{-# LANGUAGE DeriveGeneric         #-}+{-# LANGUAGE StandaloneDeriving    #-}+{-# LANGUAGE ScopedTypeVariables   #-}+{-# LANGUAGE TemplateHaskell       #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances     #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Extras.Internal.Primitives+-- Copyright   :  (c) Tim Watson 2013, Parallel Scientific (Jeff Epstein) 2012+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainers :  Jeff Epstein, Tim Watson+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- This module provides a set of additional primitives that add functionality+-- to the basic Cloud Haskell APIs.+-----------------------------------------------------------------------------++module Control.Distributed.Process.Extras.Internal.Primitives+  ( -- * General Purpose Process Addressing+    Addressable+  , Routable(..)+  , Resolvable(..)+  , Linkable(..)+  , Killable(..)++    -- * Spawning and Linking+  , spawnSignalled+  , spawnLinkLocal+  , spawnMonitorLocal+  , linkOnFailure++    -- * Registered Processes+  , whereisRemote+  , whereisOrStart+  , whereisOrStartRemote++    -- * Selective Receive/Matching+  , matchCond+  , awaitResponse++    -- * General Utilities+  , times+  , monitor+  , awaitExit+  , isProcessAlive+  , forever'+  , deliver++    -- * Remote Table+  , __remoteTable+  ) where++import Control.Concurrent (myThreadId, throwTo)+import Control.Distributed.Process hiding (monitor)+import qualified Control.Distributed.Process as P (monitor)+import Control.Distributed.Process.Closure (seqCP, remotable, mkClosure)+import Control.Distributed.Process.Serializable (Serializable)+import Control.Distributed.Process.Extras.Internal.Types+  ( Addressable+  , Linkable(..)+  , Killable(..)+  , Resolvable(..)+  , Routable(..)+  , RegisterSelf(..)+  , ExitReason(ExitOther)+  , whereisRemote+  )+import Control.Monad (void)+import Data.Maybe (isJust, fromJust)++-- utility++-- | Monitor any @Resolvable@ object.+--+monitor :: Resolvable a => a -> Process (Maybe MonitorRef)+monitor addr = do+  mPid <- resolve addr+  case mPid of+    Nothing -> return Nothing+    Just p  -> return . Just =<< P.monitor p++awaitExit :: Resolvable a => a -> Process ()+awaitExit addr = do+  mPid <- resolve addr+  case mPid of+    Nothing -> return ()+    Just p  -> do+      mRef <- P.monitor p+      receiveWait [+          matchIf (\(ProcessMonitorNotification r p' _) -> r == mRef && p == p')+                  (\_ -> return ())+        ]++deliver :: (Addressable a, Serializable m) => m -> a -> Process ()+deliver = flip sendTo++isProcessAlive :: ProcessId -> Process Bool+isProcessAlive pid = getProcessInfo pid >>= \info -> return $ info /= Nothing++-- | Apply the supplied expression /n/ times+times :: Int -> Process () -> Process ()+n `times` proc = runP proc n+  where runP :: Process () -> Int -> Process ()+        runP _ 0 = return ()+        runP p n' = p >> runP p (n' - 1)++-- | Like 'Control.Monad.forever' but sans space leak+forever' :: Monad m => m a -> m b+forever' a = let a' = a >> a' in a'+{-# INLINE forever' #-}++-- spawning, linking and generic server startup++-- | Spawn a new (local) process. This variant takes an initialisation+-- action and a secondary expression from the result of the initialisation+-- to @Process ()@. The spawn operation synchronises on the completion of the+-- @before@ action, such that the calling process is guaranteed to only see+-- the newly spawned @ProcessId@ once the initialisation has successfully+-- completed.+spawnSignalled :: Process a -> (a -> Process ()) -> Process ProcessId+spawnSignalled before after = do+  (sigStart, recvStart) <- newChan+  (pid, mRef) <- spawnMonitorLocal $ do+    initProc <- before+    sendChan sigStart ()+    after initProc+  receiveWait [+      matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)+              (\(ProcessMonitorNotification _ _ dr) -> die $ ExitOther (show dr))+    , matchChan recvStart (\() -> return pid)+    ]++-- | Node local version of 'Control.Distributed.Process.spawnLink'.+-- Note that this is just the sequential composition of 'spawn' and 'link'.+-- (The "Unified" semantics that underlies Cloud Haskell does not even support+-- a synchronous link operation)+spawnLinkLocal :: Process () -> Process ProcessId+spawnLinkLocal p = do+  pid <- spawnLocal p+  link pid+  return pid++-- | Like 'spawnLinkLocal', but monitors the spawned process.+--+spawnMonitorLocal :: Process () -> Process (ProcessId, MonitorRef)+spawnMonitorLocal p = do+  pid <- spawnLocal p+  ref <- P.monitor pid+  return (pid, ref)++-- | CH's 'link' primitive, unlike Erlang's, will trigger when the target+-- process dies for any reason. This function has semantics like Erlang's:+-- it will trigger 'ProcessLinkException' only when the target dies abnormally.+--+linkOnFailure :: ProcessId -> Process ()+linkOnFailure them = do+  us <- getSelfPid+  tid <- liftIO $ myThreadId+  void $ spawnLocal $ do+    callerRef <- P.monitor us+    calleeRef <- P.monitor them+    reason <- receiveWait [+             matchIf (\(ProcessMonitorNotification mRef _ _) ->+                       mRef == callerRef) -- nothing left to do+                     (\_ -> return DiedNormal)+           , matchIf (\(ProcessMonitorNotification mRef' _ _) ->+                       mRef' == calleeRef)+                     (\(ProcessMonitorNotification _ _ r') -> return r')+         ]+    case reason of+      DiedNormal -> return ()+      _ -> liftIO $ throwTo tid (ProcessLinkException us reason)++-- | Returns the pid of the process that has been registered+-- under the given name. This refers to a local, per-node registration,+-- not @global@ registration. If that name is unregistered, a process+-- is started. This is a handy way to start per-node named servers.+--+whereisOrStart :: String -> Process () -> Process ProcessId+whereisOrStart name proc =+  do mpid <- whereis name+     case mpid of+       Just pid -> return pid+       Nothing ->+         do caller <- getSelfPid+            pid <- spawnLocal $+                 do self <- getSelfPid+                    register name self+                    send caller (RegisterSelf,self)+                    () <- expect+                    proc+            ref <- P.monitor pid+            ret <- receiveWait+               [ matchIf (\(ProcessMonitorNotification aref _ _) -> ref == aref)+                         (\(ProcessMonitorNotification _ _ _) -> return Nothing),+                 matchIf (\(RegisterSelf,apid) -> apid == pid)+                         (\(RegisterSelf,_) -> return $ Just pid)+               ]+            case ret of+              Nothing -> whereisOrStart name proc+              Just somepid ->+                do unmonitor ref+                   send somepid ()+                   return somepid++registerSelf :: (String, ProcessId) -> Process ()+registerSelf (name,target) =+  do self <- getSelfPid+     register name self+     send target (RegisterSelf, self)+     () <- expect+     return ()++$(remotable ['registerSelf])++-- | A remote equivalent of 'whereisOrStart'. It deals with the+-- node registry on the given node, and the process, if it needs to be started,+-- will run on that node. If the node is inaccessible, Nothing will be returned.+--+whereisOrStartRemote :: NodeId -> String -> Closure (Process ()) -> Process (Maybe ProcessId)+whereisOrStartRemote nid name proc =+     do mRef <- monitorNode nid+        whereisRemoteAsync nid name+        res <- receiveWait+          [ matchIf (\(WhereIsReply label _) -> label == name)+                    (\(WhereIsReply _ mPid) -> return (Just mPid)),+            matchIf (\(NodeMonitorNotification aref _ _) -> aref == mRef)+                    (\(NodeMonitorNotification _ _ _) -> return Nothing)+          ]+        case res of+           Nothing -> return Nothing+           Just (Just pid) -> unmonitor mRef >> return (Just pid)+           Just Nothing ->+              do self <- getSelfPid+                 sRef <- spawnAsync nid ($(mkClosure 'registerSelf) (name,self) `seqCP` proc)+                 ret <- receiveWait [+                      matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)+                              (\(NodeMonitorNotification _ _ _) -> return Nothing),+                      matchIf (\(DidSpawn ref _) -> ref==sRef )+                              (\(DidSpawn _ pid) ->+                                  do pRef <- P.monitor pid+                                     receiveWait+                                       [ matchIf (\(RegisterSelf, apid) -> apid == pid)+                                                 (\(RegisterSelf, _) -> do unmonitor pRef+                                                                           send pid ()+                                                                           return $ Just pid),+                                         matchIf (\(NodeMonitorNotification aref _ _) -> aref == mRef)+                                                 (\(NodeMonitorNotification _aref _ _) -> return Nothing),+                                         matchIf (\(ProcessMonitorNotification ref _ _) -> ref==pRef)+                                                 (\(ProcessMonitorNotification _ _ _) -> return Nothing)+                                       ] )+                      ]+                 unmonitor mRef+                 case ret of+                   Nothing -> whereisOrStartRemote nid name proc+                   Just pid -> return $ Just pid++-- advanced messaging/matching++-- | An alternative to 'matchIf' that allows both predicate and action+-- to be expressed in one parameter.+matchCond :: (Serializable a) => (a -> Maybe (Process b)) -> Match b+matchCond cond =+   let v n = (isJust n, fromJust n)+       res = v . cond+    in matchIf (fst . res) (snd . res)++-- | Safe (i.e., monitored) waiting on an expected response/message.+awaitResponse :: Addressable a+              => a+              -> [Match (Either ExitReason b)]+              -> Process (Either ExitReason b)+awaitResponse addr matches = do+  mPid <- resolve addr+  case mPid of+    Nothing -> return $ Left $ ExitOther "UnresolvedAddress"+    Just p  -> do+      mRef <- P.monitor p+      receiveWait ((matchRef mRef):matches)+  where+    matchRef :: MonitorRef -> Match (Either ExitReason b)+    matchRef r = matchIf (\(ProcessMonitorNotification r' _ _) -> r == r')+                         (\(ProcessMonitorNotification _ _ d) -> do+                             return (Left (ExitOther (show d))))+

@@ -0,0 +1,38 @@+{-# LANGUAGE BangPatterns       #-}+{-# LANGUAGE StandaloneDeriving #-}+module Control.Distributed.Process.Extras.Internal.Queue.PriorityQ where++-- NB: we might try this with a skewed binomial heap at some point,+-- but for now, we'll use this module from the fingertree package+import qualified Data.PriorityQueue.FingerTree as PQ+import Data.PriorityQueue.FingerTree (PQueue)++newtype PriorityQ k a = PriorityQ { q :: PQueue k a }++{-# INLINE empty #-}+empty :: Ord k => PriorityQ k v+empty = PriorityQ $ PQ.empty++{-# INLINE isEmpty #-}+isEmpty :: Ord k => PriorityQ k v -> Bool+isEmpty = PQ.null . q++{-# INLINE singleton #-}+singleton :: Ord k => k -> a -> PriorityQ k a+singleton !k !v = PriorityQ $ PQ.singleton k v++{-# INLINE enqueue #-}+enqueue :: Ord k => k -> v -> PriorityQ k v -> PriorityQ k v+enqueue !k !v p = PriorityQ (PQ.add k v $ q p)++{-# INLINE dequeue #-}+dequeue :: Ord k => PriorityQ k v -> Maybe (v, PriorityQ k v)+dequeue p = maybe Nothing (\(v, pq') -> Just (v, pq')) $+              case (PQ.minView (q p)) of+                Nothing     -> Nothing+                Just (v, q') -> Just (v, PriorityQ $ q')++{-# INLINE peek #-}+peek :: Ord k => PriorityQ k v -> Maybe v+peek p = maybe Nothing (\(v, _) -> Just v) $ dequeue p+

@@ -0,0 +1,66 @@+-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Extras.Internal.Queue.SeqQ+-- Copyright   :  (c) Tim Watson 2012 - 2013+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+--+-- A simple FIFO queue implementation backed by @Data.Sequence@.+-----------------------------------------------------------------------------++module Control.Distributed.Process.Extras.Internal.Queue.SeqQ+  ( SeqQ+  , empty+  , isEmpty+  , singleton+  , enqueue+  , dequeue+  , peek+  )+  where++import Data.Sequence+  ( Seq+  , ViewR(..)+  , (<|)+  , viewr+  )+import qualified Data.Sequence as Seq (empty, singleton, null)++newtype SeqQ a = SeqQ { q :: Seq a }+  deriving (Show)++instance Eq a => Eq (SeqQ a) where+  a == b = (q a) == (q b)++{-# INLINE empty #-}+empty :: SeqQ a+empty = SeqQ Seq.empty++isEmpty :: SeqQ a -> Bool+isEmpty = Seq.null . q++{-# INLINE singleton #-}+singleton :: a -> SeqQ a+singleton = SeqQ . Seq.singleton++{-# INLINE enqueue #-}+enqueue :: SeqQ a -> a -> SeqQ a+enqueue s a = SeqQ $ a <| q s++{-# INLINE dequeue #-}+dequeue :: SeqQ a -> Maybe (a, SeqQ a)+dequeue s = maybe Nothing (\(s' :> a) -> Just (a, SeqQ s')) $ getR s++{-# INLINE peek #-}+peek :: SeqQ a -> Maybe a+peek s = maybe Nothing (\(_ :> a) -> Just a) $ getR s++getR :: SeqQ a -> Maybe (ViewR a)+getR s =+  case (viewr (q s)) of+    EmptyR -> Nothing+    a      -> Just a+

@@ -0,0 +1,277 @@+{-# LANGUAGE DeriveDataTypeable     #-}+{-# LANGUAGE DeriveGeneric          #-}+{-# LANGUAGE StandaloneDeriving     #-}+{-# LANGUAGE TemplateHaskell        #-}+{-# LANGUAGE FlexibleInstances      #-}+{-# LANGUAGE UndecidableInstances   #-}+{-# LANGUAGE OverlappingInstances   #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables   #-}++-- | Types used throughout the Extras package+--+module Control.Distributed.Process.Extras.Internal.Types+  ( -- * Tagging+    Tag+  , TagPool+  , newTagPool+  , getTag+    -- * Addressing+  , Linkable(..)+  , Killable(..)+  , Resolvable(..)+  , Routable(..)+  , Addressable+  , sendToRecipient+  , Recipient(..)+  , RegisterSelf(..)+    -- * Interactions+  , whereisRemote+  , resolveOrDie+  , CancelWait(..)+  , Channel+  , Shutdown(..)+  , ExitReason(..)+  , ServerDisconnected(..)+  , NFSerializable+    -- remote table+  , __remoteTable+  ) where++import Control.Concurrent.MVar+  ( MVar+  , newMVar+  , modifyMVar+  )+import Control.DeepSeq (NFData, ($!!))+import Control.Distributed.Process hiding (send)+import qualified Control.Distributed.Process as P+  ( send+  , unsafeSend+  , unsafeNSend+  )+import Control.Distributed.Process.Closure+  ( remotable+  , mkClosure+  , functionTDict+  )+import Control.Distributed.Process.Serializable++import Data.Binary+import Data.Typeable (Typeable)+import GHC.Generics++--------------------------------------------------------------------------------+-- API                                                                        --+--------------------------------------------------------------------------------++-- | Introduces a class that brings NFData into scope along with Serializable,+-- such that we can force evaluation. Intended for use with the UnsafePrimitives+-- module (which wraps "Control.Distributed.Process.UnsafePrimitives"), and+-- guarantees evaluatedness in terms of @NFData@. Please note that we /cannot/+-- guarantee that an @NFData@ instance will behave the same way as a @Binary@+-- one with regards evaluation, so it is still possible to introduce unexpected+-- behaviour by using /unsafe/ primitives in this way.+--+class (NFData a, Serializable a) => NFSerializable a+instance (NFData a, Serializable a) => NFSerializable a++-- | Tags provide uniqueness for messages, so that they can be+-- matched with their response.+type Tag = Int++-- | Generates unique 'Tag' for messages and response pairs.+-- Each process that depends, directly or indirectly, on+-- the call mechanisms in "Control.Distributed.Process.Global.Call"+-- should have at most one TagPool on which to draw unique message+-- tags.+type TagPool = MVar Tag++-- | Create a new per-process source of unique+-- message identifiers.+newTagPool :: Process TagPool+newTagPool = liftIO $ newMVar 0++-- | Extract a new identifier from a 'TagPool'.+getTag :: TagPool -> Process Tag+getTag tp = liftIO $ modifyMVar tp (\tag -> return (tag+1,tag))++-- | Wait cancellation message.+data CancelWait = CancelWait+    deriving (Eq, Show, Typeable, Generic)+instance Binary CancelWait where+instance NFData CancelWait where++-- | Simple representation of a channel.+type Channel a = (SendPort a, ReceivePort a)++-- | Used internally in whereisOrStart. Sent as (RegisterSelf,ProcessId).+data RegisterSelf = RegisterSelf+  deriving (Typeable, Generic)+instance Binary RegisterSelf where+instance NFData RegisterSelf where++-- | A ubiquitous /shutdown signal/ that can be used+-- to maintain a consistent shutdown/stop protocol for+-- any process that wishes to handle it.+data Shutdown = Shutdown+  deriving (Typeable, Generic, Show, Eq)+instance Binary Shutdown where+instance NFData Shutdown where++-- | Provides a /reason/ for process termination.+data ExitReason =+    ExitNormal        -- ^ indicates normal exit+  | ExitShutdown      -- ^ normal response to a 'Shutdown'+  | ExitOther !String -- ^ abnormal (error) shutdown+  deriving (Typeable, Generic, Eq, Show)+instance Binary ExitReason where+instance NFData ExitReason where++-- | A simple means of mapping to a receiver.+data Recipient =+    Pid !ProcessId+  | Registered !String+  | RemoteRegistered !String !NodeId+--  | ProcReg !ProcessId !String+--  | RemoteProcReg NodeId String+--  | GlobalReg String+  deriving (Typeable, Generic, Show, Eq)+instance Binary Recipient where++-- useful exit reasons++-- | Given when a server is unobtainable.+data ServerDisconnected = ServerDisconnected !DiedReason+  deriving (Typeable, Generic)+instance Binary ServerDisconnected where+instance NFData ServerDisconnected where++$(remotable ['whereis])++-- | A synchronous version of 'whereis', this relies on 'call'+-- to perform the relevant monitoring of the remote node.+whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)+whereisRemote node name =+  call $(functionTDict 'whereis) node ($(mkClosure 'whereis) name)++sendToRecipient :: (Serializable m) => Recipient -> m -> Process ()+sendToRecipient (Pid p) m                = P.send p m+sendToRecipient (Registered s) m         = nsend s m+sendToRecipient (RemoteRegistered s n) m = nsendRemote n s m++unsafeSendToRecipient :: (NFSerializable m) => Recipient -> m -> Process ()+unsafeSendToRecipient (Pid p) m                = P.unsafeSend p $!! m+unsafeSendToRecipient (Registered s) m         = P.unsafeNSend s $!! m+unsafeSendToRecipient (RemoteRegistered s n) m = nsendRemote n s m++baseAddressableErrorMessage :: (Routable a) => a -> String+baseAddressableErrorMessage _ = "CannotResolveAddressable"++-- | Class of things to which a @Process@ can /link/ itself.+class Linkable a where+  -- | Create a /link/ with the supplied object.+  linkTo :: a -> Process ()++-- | Class of things that can be resolved to a 'ProcessId'.+--+class Resolvable a where+  -- | Resolve the reference to a process id, or @Nothing@ if resolution fails+  resolve :: a -> Process (Maybe ProcessId)++-- | Class of things that can be killed (or instructed to exit).+class Killable a where+  killProc :: a -> String -> Process ()+  exitProc :: (Serializable m) => a -> m -> Process ()++instance Killable ProcessId where+  killProc = kill+  exitProc = exit++instance Resolvable r => Killable r where+  killProc r s = resolve r >>= maybe (return ()) (flip kill $ s)+  exitProc r m = resolve r >>= maybe (return ()) (flip exit $ m)++-- | Provides a unified API for addressing processes.+--+class Routable a where+  -- | Send a message to the target asynchronously+  sendTo  :: (Serializable m) => a -> m -> Process ()++  -- | Send some @NFData@ message to the target asynchronously,+  -- forcing evaluation (i.e., @deepseq@) beforehand.+  unsafeSendTo :: (NFSerializable m) => a -> m -> Process ()++  -- | Unresolvable @Addressable@ Message+  unresolvableMessage :: a -> String+  unresolvableMessage = baseAddressableErrorMessage++instance (Resolvable a) => Routable a where+  sendTo a m = do+    mPid <- resolve a+    maybe (die (unresolvableMessage a))+          (\p -> P.send p m)+          mPid++  unsafeSendTo a m = do+    mPid <- resolve a+    maybe (die (unresolvableMessage a))+          (\p -> P.unsafeSend p $!! m)+          mPid++  -- | Unresolvable Addressable Message+  unresolvableMessage = baseAddressableErrorMessage++instance Resolvable Recipient where+  resolve (Pid                p) = return (Just p)+  resolve (Registered         n) = whereis n+  resolve (RemoteRegistered s n) = whereisRemote n s++instance Routable Recipient where+  sendTo = sendToRecipient+  unsafeSendTo = unsafeSendToRecipient++  unresolvableMessage (Pid                p) = unresolvableMessage p+  unresolvableMessage (Registered         n) = unresolvableMessage n+  unresolvableMessage (RemoteRegistered s n) = unresolvableMessage (n, s)++instance Resolvable ProcessId where+  resolve p = return (Just p)++instance Routable ProcessId where+  sendTo                 = P.send+  unsafeSendTo pid msg   = P.unsafeSend pid $!! msg+  unresolvableMessage p  = "CannotResolvePid[" ++ (show p) ++ "]"++instance Resolvable String where+  resolve = whereis++instance Routable String where+  sendTo                = nsend+  unsafeSendTo name msg = P.unsafeNSend name $!! msg+  unresolvableMessage s = "CannotResolveRegisteredName[" ++ s ++ "]"++instance Resolvable (NodeId, String) where+  resolve (nid, pname) = whereisRemote nid pname++instance Routable (NodeId, String) where+  sendTo  (nid, pname) msg   = nsendRemote nid pname msg+  unsafeSendTo               = sendTo -- because serialisation *must* take place+  unresolvableMessage (n, s) =+    "CannotResolveRemoteRegisteredName[name: " ++ s ++ ", node: " ++ (show n) ++ "]"++instance Routable (Message -> Process ()) where+  sendTo f       = f . wrapMessage+  unsafeSendTo f = f . unsafeWrapMessage++class (Resolvable a, Routable a) => Addressable a+instance (Resolvable a, Routable a) => Addressable a++-- TODO: this probably belongs somewhere other than in ..Types.+-- | resolve the Resolvable or die with specified msg plus details of what didn't resolve+resolveOrDie  :: (Routable a, Resolvable a) => a -> String -> Process ProcessId+resolveOrDie resolvable failureMsg = do+  result <- resolve resolvable+  case result of+    Nothing -> die $ failureMsg ++ " " ++ unresolvableMessage resolvable+    Just pid -> return pid

@@ -0,0 +1,128 @@+{-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE DeriveGeneric       #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving  #-}++-- | If you don't know exactly what this module is for and precisely+-- how to use the types within, you should move on, quickly!+--+module Control.Distributed.Process.Extras.Internal.Unsafe+  ( -- * Copying non-serializable data+    PCopy()+  , pCopy+  , matchP+  , matchChanP+  , pUnwrap+    -- * Arbitrary (unmanaged) message streams+  , InputStream(Null)+  , newInputStream+  , matchInputStream+  , readInputStream+  , InvalidBinaryShim(..)+  ) where++import Control.Concurrent.STM (STM, atomically)+import Control.Distributed.Process+  ( matchAny+  , matchChan+  , matchSTM+  , match+  , handleMessage+  , receiveChan+  , liftIO+  , die+  , Match+  , ReceivePort+  , Message+  , Process+  )+import Control.Distributed.Process.Serializable (Serializable)+import Data.Binary+import Control.DeepSeq (NFData)+import Data.Typeable (Typeable)+import GHC.Generics++data InvalidBinaryShim = InvalidBinaryShim+  deriving (Typeable, Show, Eq)++-- NB: PCopy is a shim, allowing us to copy a pointer to otherwise+-- non-serializable data directly to another local process'+-- mailbox with no serialisation or even deepseq evaluation+-- required. We disallow remote queries (i.e., from other nodes)+-- and thus the Binary instance below is never used (though it's+-- required by the type system) and will in fact generate errors if+-- you attempt to use it at runtime. In other words, if you attempt+-- to make a @Message@ out of this, you'd better make sure you're+-- calling @unsafeCreateUnencodedMessage@, otherwise /BOOM/! You have+-- been warned.+--+data PCopy a = PCopy !a+  deriving (Typeable, Generic)+instance (NFData a) => NFData (PCopy a) where++instance (Typeable a) => Binary (PCopy a) where+  put _ = error "InvalidBinaryShim"+  get   = error "InvalidBinaryShim"++-- | Wrap any @Typeable@ datum in a @PCopy@. We hide the constructor to+-- discourage arbitrary uses of the type, since @PCopy@ is a specialised+-- and potentially dangerous construct.+pCopy :: (Typeable a) => a -> PCopy a+pCopy = PCopy++-- | Matches on @PCopy m@ and returns the /m/ within.+-- This potentially allows us to bypass serialization (and the type constraints+-- it enforces) for local message passing (i.e., with @UnencodedMessage@ data),+-- since PCopy is just a shim.+matchP :: (Typeable m) => Match (Maybe m)+matchP = matchAny pUnwrap++-- | Given a raw @Message@, attempt to unwrap a @Typeable@ datum from+-- an enclosing @PCopy@ wrapper.+pUnwrap :: (Typeable m) => Message -> Process (Maybe m)+pUnwrap m = handleMessage m (\(PCopy m' :: PCopy m) -> return m')++-- | Matches on a @TypedChannel (PCopy a)@.+matchChanP :: (Typeable m) => ReceivePort (PCopy m) -> Match m+matchChanP rp = matchChan rp (\(PCopy m' :: PCopy m) -> return m')++-- | A generic input channel that can be read from in the same fashion+-- as a typed channel (i.e., @ReceivePort@). To read from an input stream+-- in isolation, see 'readInputStream'. To compose an 'InputStream' with+-- reads on a process' mailbox (and/or typed channels), see 'matchInputStream'.+--+data InputStream a = ReadChan (ReceivePort a) | ReadSTM (STM a) | Null+  deriving (Typeable)++data NullInputStream = NullInputStream+  deriving (Typeable, Generic, Show, Eq)+instance Binary NullInputStream where+instance NFData NullInputStream where++-- [note: InputStream]+-- InputStream wraps either a ReceivePort or an arbitrary STM action. Used+-- internally when we want to allow internal clients to completely bypass+-- regular messaging primitives (which is rare but occaisionally useful),+-- the type (only, minus its constructors) is exposed to users of some+-- @Exchange@ APIs.++-- | Create a new 'InputStream'.+newInputStream :: forall a. (Typeable a)+               => Either (ReceivePort a) (STM a)+               -> InputStream a+newInputStream (Left rp)   = ReadChan rp+newInputStream (Right stm) = ReadSTM stm++-- | Read from an 'InputStream'. This is a blocking operation.+readInputStream :: (Serializable a) => InputStream a -> Process a+readInputStream (ReadChan rp) = receiveChan rp+readInputStream (ReadSTM stm) = liftIO $ atomically stm+readInputStream Null          = die $ NullInputStream++-- | Constructs a @Match@ for a given 'InputChannel'.+matchInputStream :: InputStream a -> Match a+matchInputStream (ReadChan rp) = matchChan rp return+matchInputStream (ReadSTM stm) = matchSTM stm return+matchInputStream Null          = match (\NullInputStream -> do+                                           error "NullInputStream")+

@@ -0,0 +1,258 @@+{-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE DeriveGeneric       #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Extras.Time+-- Copyright   :  (c) Tim Watson, Jeff Epstein, Alan Zimmerman+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- This module provides facilities for working with time delays and timeouts.+-- The type 'Timeout' and the 'timeout' family of functions provide mechanisms+-- for working with @threadDelay@-like behaviour that operates on microsecond+-- values.+--+-- The 'TimeInterval' and 'TimeUnit' related functions provide an abstraction+-- for working with various time intervals, whilst the 'Delay' type provides a+-- corrolary to 'timeout' that works with these.+-----------------------------------------------------------------------------++module Control.Distributed.Process.Extras.Time+  ( -- * Time interval handling+    microSeconds+  , milliSeconds+  , seconds+  , minutes+  , hours+  , asTimeout+  , after+  , within+  , timeToMicros+  , TimeInterval+  , TimeUnit(..)+  , Delay(..)++  -- * Conversion To/From NominalDiffTime+  , timeIntervalToDiffTime+  , diffTimeToTimeInterval+  , diffTimeToDelay+  , delayToDiffTime+  , microsecondsToNominalDiffTime++    -- * (Legacy) Timeout Handling+  , Timeout+  , TimeoutNotification(..)+  , timeout+  , infiniteWait+  , noWait+  ) where++import Control.Concurrent (threadDelay)+import Control.DeepSeq (NFData)+import Control.Distributed.Process+import Control.Distributed.Process.Extras.Internal.Types+import Control.Monad (void)+import Data.Binary+import Data.Ratio ((%))+import Data.Time.Clock+import Data.Typeable (Typeable)++import GHC.Generics++--------------------------------------------------------------------------------+-- API                                                                        --+--------------------------------------------------------------------------------++-- | Defines the time unit for a Timeout value+data TimeUnit = Days | Hours | Minutes | Seconds | Millis | Micros+    deriving (Typeable, Generic, Eq, Show)++instance Binary TimeUnit where+instance NFData TimeUnit where++-- | A time interval.+data TimeInterval = TimeInterval TimeUnit Int+    deriving (Typeable, Generic, Eq, Show)++instance Binary TimeInterval where+instance NFData TimeInterval where++-- | Represents either a delay of 'TimeInterval', an infinite wait or no delay+-- (i.e., non-blocking).+data Delay = Delay TimeInterval | Infinity | NoDelay+    deriving (Typeable, Generic, Eq, Show)++instance Binary Delay where+instance NFData Delay where++-- | Represents a /timeout/ in terms of microseconds, where 'Nothing' stands for+-- infinity and @Just 0@, no-delay.+type Timeout = Maybe Int++-- | Send to a process when a timeout expires.+data TimeoutNotification = TimeoutNotification Tag+       deriving (Typeable)++instance Binary TimeoutNotification where+  get = fmap TimeoutNotification $ get+  put (TimeoutNotification n) = put n++-- time interval/unit handling++-- | converts the supplied @TimeInterval@ to microseconds+asTimeout :: TimeInterval -> Int+asTimeout (TimeInterval u v) = timeToMicros u v++-- | Convenience for making timeouts; e.g.,+--+-- > receiveTimeout (after 3 Seconds) [ match (\"ok" -> return ()) ]+--+after :: Int -> TimeUnit -> Int+after n m = timeToMicros m n++-- | Convenience for making 'TimeInterval'; e.g.,+--+-- > let ti = within 5 Seconds in .....+--+within :: Int -> TimeUnit -> TimeInterval+within n m = TimeInterval m n++-- | given a number, produces a @TimeInterval@ of microseconds+microSeconds :: Int -> TimeInterval+microSeconds = TimeInterval Micros++-- | given a number, produces a @TimeInterval@ of milliseconds+milliSeconds :: Int -> TimeInterval+milliSeconds = TimeInterval Millis++-- | given a number, produces a @TimeInterval@ of seconds+seconds :: Int -> TimeInterval+seconds = TimeInterval Seconds++-- | given a number, produces a @TimeInterval@ of minutes+minutes :: Int -> TimeInterval+minutes = TimeInterval Minutes++-- | given a number, produces a @TimeInterval@ of hours+hours :: Int -> TimeInterval+hours = TimeInterval Hours++-- TODO: is timeToMicros efficient enough?++-- | converts the supplied @TimeUnit@ to microseconds+{-# INLINE timeToMicros #-}+timeToMicros :: TimeUnit -> Int -> Int+timeToMicros Micros  us   = us+timeToMicros Millis  ms   = ms  * (10 ^ (3 :: Int)) -- (1000µs == 1ms)+timeToMicros Seconds secs = timeToMicros Millis  (secs * milliSecondsPerSecond)+timeToMicros Minutes mins = timeToMicros Seconds (mins * secondsPerMinute)+timeToMicros Hours   hrs  = timeToMicros Minutes (hrs  * minutesPerHour)+timeToMicros Days    days = timeToMicros Hours   (days * hoursPerDay)++{-# INLINE hoursPerDay #-}+hoursPerDay :: Int+hoursPerDay = 60++{-# INLINE minutesPerHour #-}+minutesPerHour :: Int+minutesPerHour = 60++{-# INLINE secondsPerMinute #-}+secondsPerMinute :: Int+secondsPerMinute = 60++{-# INLINE milliSecondsPerSecond #-}+milliSecondsPerSecond :: Int+milliSecondsPerSecond = 1000++{-# INLINE microSecondsPerSecond #-}+microSecondsPerSecond :: Int+microSecondsPerSecond = 1000000++-- timeouts/delays (microseconds)++-- | Constructs an inifinite 'Timeout'.+infiniteWait :: Timeout+infiniteWait = Nothing++-- | Constructs a no-wait 'Timeout'+noWait :: Timeout+noWait = Just 0++-- | Sends the calling process @TimeoutNotification tag@ after @time@ microseconds+timeout :: Int -> Tag -> ProcessId -> Process ()+timeout time tag p =+  void $ spawnLocal $+               do liftIO $ threadDelay time+                  send p (TimeoutNotification tag)++-- Converting to/from Data.Time.Clock NominalDiffTime++-- | given a @TimeInterval@, provide an equivalent @NominalDiffTim@+timeIntervalToDiffTime :: TimeInterval -> NominalDiffTime+timeIntervalToDiffTime ti = microsecondsToNominalDiffTime (fromIntegral $ asTimeout ti)++-- | given a @NominalDiffTim@@, provide an equivalent @TimeInterval@+diffTimeToTimeInterval :: NominalDiffTime -> TimeInterval+diffTimeToTimeInterval dt = microSeconds $ (fromIntegral (round (dt * 1000000) :: Integer))++-- | given a @Delay@, provide an equivalent @NominalDiffTim@+delayToDiffTime :: Delay -> NominalDiffTime+delayToDiffTime (Delay ti) = timeIntervalToDiffTime ti+delayToDiffTime Infinity   = error "trying to convert Delay.Infinity to a NominalDiffTime"+delayToDiffTime (NoDelay)  = microsecondsToNominalDiffTime 0++-- | given a @NominalDiffTim@@, provide an equivalent @Delay@+diffTimeToDelay :: NominalDiffTime -> Delay+diffTimeToDelay dt = Delay $ diffTimeToTimeInterval dt++-- | Create a 'NominalDiffTime' from a number of microseconds.+microsecondsToNominalDiffTime :: Integer -> NominalDiffTime+microsecondsToNominalDiffTime x = fromRational (x % (fromIntegral microSecondsPerSecond))++-- tenYearsAsMicroSeconds :: Integer+-- tenYearsAsMicroSeconds = 10 * 365 * 24 * 60 * 60 * 1000000++-- | Allow @(+)@ and @(-)@ operations on @TimeInterval@s+instance Num TimeInterval where+  t1 + t2 = microSeconds $ asTimeout t1 + asTimeout t2+  t1 - t2 = microSeconds $ asTimeout t1 - asTimeout t2+  _ * _ = error "trying to multiply two TimeIntervals"+  abs t = microSeconds $ abs (asTimeout t)+  signum t = if (asTimeout t) == 0+              then 0+              else if (asTimeout t) < 0 then -1+                                        else 1+  fromInteger _ = error "trying to call fromInteger for a TimeInterval. Cannot guess units"++-- | Allow @(+)@ and @(-)@ operations on @Delay@s+instance Num Delay where+  NoDelay     + x          = x+  Infinity    + _          = Infinity+  x           + NoDelay    = x+  _           + Infinity   = Infinity+  (Delay t1 ) + (Delay t2) = Delay (t1 + t2)++  NoDelay     - x          = x+  Infinity    - _          = Infinity+  x           - NoDelay    = x+  _           - Infinity   = Infinity+  (Delay t1 ) - (Delay t2) = Delay (t1 - t2)++  _ * _ = error "trying to multiply two Delays"++  abs NoDelay   = NoDelay+  abs Infinity  = Infinity+  abs (Delay t) = Delay (abs t)++  signum (NoDelay) = 0+  signum Infinity  = 1+  signum (Delay t) = Delay (signum t)++  fromInteger 0 = NoDelay+  fromInteger _ = error "trying to call fromInteger for a Delay. Cannot guess units"+

@@ -0,0 +1,180 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric      #-}+{-# LANGUAGE PatternGuards      #-}+{-# LANGUAGE TemplateHaskell    #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Extras.Timer+-- Copyright   :  (c) Tim Watson 2012+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- Provides an API for running code or sending messages, either after some+-- initial delay or periodically, and for cancelling, re-setting and/or+-- flushing pending /timers/.+-----------------------------------------------------------------------------++module Control.Distributed.Process.Extras.Timer+  (+    TimerRef+  , Tick(Tick)+  , sleep+  , sleepFor+  , sendAfter+  , runAfter+  , exitAfter+  , killAfter+  , startTimer+  , ticker+  , periodically+  , resetTimer+  , cancelTimer+  , flushTimer+  ) where++import Control.DeepSeq (NFData)+import Control.Distributed.Process hiding (send)+import Control.Distributed.Process.Serializable+import Control.Distributed.Process.Extras.UnsafePrimitives (send)+import Control.Distributed.Process.Extras.Internal.Types (NFSerializable)+import Control.Distributed.Process.Extras.Time+import Data.Binary+import Data.Typeable (Typeable)+import Prelude hiding (init)++import GHC.Generics++-- | an opaque reference to a timer+type TimerRef = ProcessId++-- | cancellation message sent to timers+data TimerConfig = Reset | Cancel+    deriving (Typeable, Generic, Eq, Show)+instance Binary TimerConfig where+instance NFData TimerConfig where++-- | represents a 'tick' event that timers can generate+data Tick = Tick+    deriving (Typeable, Generic, Eq, Show)+instance Binary Tick where+instance NFData Tick where++data SleepingPill = SleepingPill+    deriving (Typeable, Generic, Eq, Show)+instance Binary SleepingPill where+instance NFData SleepingPill where++--------------------------------------------------------------------------------+-- API                                                                        --+--------------------------------------------------------------------------------++-- | blocks the calling Process for the specified TimeInterval. Note that this+-- function assumes that a blocking receive is the most efficient approach to+-- acheiving this, however the runtime semantics (particularly with regards+-- scheduling) should not differ from threadDelay in practise.+sleep :: TimeInterval -> Process ()+sleep t =+  let ms = asTimeout t in do+  _ <- receiveTimeout ms [matchIf (\SleepingPill -> True)+                                  (\_ -> return ())]+  return ()++-- | Literate way of saying @sleepFor 3 Seconds@.+sleepFor :: Int -> TimeUnit -> Process ()+sleepFor i u = sleep (within i u)++-- | starts a timer which sends the supplied message to the destination+-- process after the specified time interval.+sendAfter :: (NFSerializable a)+          => TimeInterval+          -> ProcessId+          -> a+          -> Process TimerRef+sendAfter t pid msg = runAfter t proc+  where proc = do { send pid msg }++-- | runs the supplied process action(s) after @t@ has elapsed+runAfter :: TimeInterval -> Process () -> Process TimerRef+runAfter t p = spawnLocal $ runTimer t p True++-- | calls @exit pid reason@ after @t@ has elapsed+exitAfter :: (Serializable a)+             => TimeInterval+             -> ProcessId+             -> a+             -> Process TimerRef+exitAfter delay pid reason = runAfter delay $ exit pid reason++-- | kills the specified process after @t@ has elapsed+killAfter :: TimeInterval -> ProcessId -> String -> Process TimerRef+killAfter delay pid why = runAfter delay $ kill pid why++-- | starts a timer that repeatedly sends the supplied message to the destination+-- process each time the specified time interval elapses. To stop messages from+-- being sent in future, 'cancelTimer' can be called.+startTimer :: (NFSerializable a)+           => TimeInterval+           -> ProcessId+           -> a+           -> Process TimerRef+startTimer t pid msg = periodically t (send pid msg)++-- | runs the supplied process action(s) repeatedly at intervals of @t@+periodically :: TimeInterval -> Process () -> Process TimerRef+periodically t p = spawnLocal $ runTimer t p False++-- | resets a running timer. Note: Cancelling a timer does not guarantee that+-- all its messages are prevented from being delivered to the target process.+-- Also note that resetting an ongoing timer (started using the 'startTimer' or+-- 'periodically' functions) will only cause the current elapsed period to time+-- out, after which the timer will continue running. To stop a long-running+-- timer permanently, you should use 'cancelTimer' instead.+resetTimer :: TimerRef -> Process ()+resetTimer = (flip send) Reset++-- | permanently cancels a timer+cancelTimer :: TimerRef -> Process ()+cancelTimer = (flip send) Cancel++-- | cancels a running timer and flushes any viable timer messages from the+-- process' message queue. This function should only be called by the process+-- expecting to receive the timer's messages!+flushTimer :: (Serializable a, Eq a) => TimerRef -> a -> Delay -> Process ()+flushTimer ref ignore t = do+    mRef <- monitor ref+    cancelTimer ref+    performFlush mRef t+    return ()+  where performFlush mRef Infinity  = receiveWait $ filters mRef+        performFlush mRef NoDelay   = performFlush mRef (Delay $ microSeconds 0)+        performFlush mRef (Delay i) = receiveTimeout (asTimeout i) (filters mRef) >> return ()+        filters mRef = [+                matchIf (\x -> x == ignore)+                        (\_ -> return ())+              , matchIf (\(ProcessMonitorNotification mRef' _ _) -> mRef == mRef')+                        (\_ -> return ()) ]++-- | sets up a timer that sends 'Tick' repeatedly at intervals of @t@+ticker :: TimeInterval -> ProcessId -> Process TimerRef+ticker t pid = startTimer t pid Tick++--------------------------------------------------------------------------------+-- Implementation                                                             --+--------------------------------------------------------------------------------++-- runs the timer process+runTimer :: TimeInterval -> Process () -> Bool -> Process ()+runTimer t proc cancelOnReset = do+    cancel <- expectTimeout (asTimeout t)+    -- say $ "cancel = " ++ (show cancel) ++ "\n"+    case cancel of+        Nothing     -> runProc cancelOnReset+        Just Cancel -> return ()+        Just Reset  -> if cancelOnReset then return ()+                                        else runTimer t proc cancelOnReset+  where runProc True  = proc+        runProc False = proc >> runTimer t proc cancelOnReset

@@ -0,0 +1,63 @@+-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Extras.UnsafePrimitives+-- Copyright   :  (c) Tim Watson 2013+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- [Unsafe Messaging Primitives Using NFData]+--+-- This module mirrors "Control.Distributed.Process.UnsafePrimitives", but+-- attempts to provide a bit more safety by forcing evaluation before sending.+-- This is handled using @NFData@, by means of the @NFSerializable@ type class.+--+-- Note that we /still/ cannot guarantee that both the @NFData@ and @Binary@+-- instances will evaluate your data the same way, therefore these primitives+-- still have certain risks and potential side effects. Use with caution.+--+-----------------------------------------------------------------------------+module Control.Distributed.Process.Extras.UnsafePrimitives+  ( send+  , nsend+  , sendToAddr+  , sendChan+  , wrapMessage+  ) where++import Control.DeepSeq (($!!))+import Control.Distributed.Process+  ( Process+  , ProcessId+  , SendPort+  , Message+  )+import Control.Distributed.Process.Extras.Internal.Types+  ( NFSerializable+  , Addressable+  , Resolvable(..)+  )+import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe++send :: NFSerializable m => ProcessId -> m -> Process ()+send pid msg = Unsafe.send pid $!! msg++nsend :: NFSerializable a => String -> a -> Process ()+nsend name msg = Unsafe.nsend name $!! msg++sendToAddr :: (Addressable a, NFSerializable m) => a -> m -> Process ()+sendToAddr addr msg = do+  mPid <- resolve addr+  case mPid of+    Nothing -> return ()+    Just p  -> send p msg++sendChan :: (NFSerializable m) => SendPort m -> m -> Process ()+sendChan port msg = Unsafe.sendChan port $!! msg++-- | Create an unencoded @Message@ for any @Serializable@ type.+wrapMessage :: NFSerializable a => a -> Message+wrapMessage msg = Unsafe.wrapMessage $!! msg+

@@ -0,0 +1,91 @@+{-# LANGUAGE PatternGuards  #-}+module Main where++import qualified Control.Distributed.Process.Extras.Internal.Queue.SeqQ as FIFO+import Control.Distributed.Process.Extras.Internal.Queue.SeqQ ( SeqQ )+import qualified Control.Distributed.Process.Extras.Internal.Queue.PriorityQ as PQ++import Control.Rematch hiding (on)+import Control.Rematch.Run+import Data.Function (on)+import Data.List+import Test.Framework as TF (defaultMain, testGroup, Test)+import Test.Framework.Providers.HUnit+import Test.Framework.Providers.QuickCheck2 (testProperty)+import Test.HUnit (Assertion, assertFailure)++import Prelude++expectThat :: a -> Matcher a -> Assertion+expectThat a matcher = case res of+  MatchSuccess -> return ()+  (MatchFailure msg) -> assertFailure msg+  where res = runMatch matcher a++-- NB: these tests/properties are not meant to be complete, but rather+-- they exercise the small number of behaviours that we actually use!++-- TODO: some laziness vs. strictness tests, with error/exception checking++prop_pq_ordering :: [Int] -> Bool+prop_pq_ordering xs =+    let xs' = map (\x -> (x, show x)) xs+        q   = foldl (\q' x -> PQ.enqueue (fst x) (snd x) q') PQ.empty xs'+        ys  = drain q []+        zs  = [snd x | x <- reverse $ sortBy (compare `on` fst) xs']+        -- the sorted list should match the stuff we drained back out+    in zs == ys+  where+    drain q xs'+      | True <- PQ.isEmpty q = xs'+      | otherwise =+          let Just (x, q') = PQ.dequeue q in drain q' (x:xs')++prop_fifo_enqueue :: Int -> Int -> Int -> Bool+prop_fifo_enqueue a b c =+  let q1           = foldl FIFO.enqueue FIFO.empty [a,b,c]+      Just (a', q2) = FIFO.dequeue q1+      Just (b', q3) = FIFO.dequeue q2+      Just (c', q4) = FIFO.dequeue q3+      Nothing       = FIFO.dequeue q4+  in q4 `seq` [a',b',c'] == [a,b,c]  -- why seq here? to shut the compiler up.++prop_enqueue_empty :: String -> Bool+prop_enqueue_empty s =+  let q            = FIFO.enqueue FIFO.empty s+      Just (_, q') = FIFO.dequeue q+  in (FIFO.isEmpty q') == ((FIFO.isEmpty q) == False)++tests :: [TF.Test]+tests = [+     testGroup "Priority Queue Tests" [+        -- testCase "New Queue Should Be Empty"+        --   (expect (PQ.isEmpty $ PQ.empty) $ equalTo True),+        -- testCase "Singleton Queue Should Contain One Element"+        --   (expect (PQ.dequeue $ (PQ.singleton 1 "hello") :: PriorityQ Int String) $+        --      equalTo $ (Just ("hello", PQ.empty)) :: Maybe (PriorityQ Int String)),+        -- testCase "Dequeue Empty Queue Should Be Nothing"+        --   (expect (Q.isEmpty $ PQ.dequeue $+        --              (PQ.empty :: PriorityQ Int ())) $ equalTo True),+        testProperty "Enqueue/Dequeue should respect Priority order"+            prop_pq_ordering+     ],+     testGroup "FIFO Queue Tests" [+        testCase "New Queue Should Be Empty"+          (expectThat (FIFO.isEmpty $ FIFO.empty) $ equalTo True),+        testCase "Singleton Queue Should Contain One Element"+          (expectThat (FIFO.dequeue $ FIFO.singleton "hello") $+             equalTo $ Just ("hello", FIFO.empty)),+        testCase "Dequeue Empty Queue Should Be Nothing"+          (expectThat (FIFO.dequeue $ (FIFO.empty :: SeqQ ())) $+            is (Nothing :: Maybe ((), SeqQ ()))),+        testProperty "Enqueue/Dequeue should respect FIFO order"+            prop_fifo_enqueue,+        testProperty "Enqueue/Dequeue should respect isEmpty"+            prop_enqueue_empty+     ]+   ]++main :: IO ()+main = defaultMain tests+

@@ -0,0 +1,196 @@+module Main where++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif+import Control.Monad (forever)+import Control.Concurrent.MVar+  ( newEmptyMVar+  , putMVar+  , takeMVar+  , withMVar+  )+import qualified Network.Transport as NT (Transport)+import Network.Transport.TCP()+import Control.DeepSeq (NFData)+import Control.Distributed.Process+import Control.Distributed.Process.Node+import Control.Distributed.Process.Serializable()+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.Extras.Timer+import Control.Distributed.Process.Tests.Internal.Utils++import Test.Framework (Test, testGroup, defaultMain)+import Test.Framework.Providers.HUnit (testCase)+import Network.Transport.TCP+import qualified Network.Transport as NT++import GHC.Generics++-- orphan instance+instance NFData Ping where++testSendAfter :: TestResult Bool -> Process ()+testSendAfter result =+  let delay = seconds 1 in do+  sleep $ seconds 10+  pid <- getSelfPid+  _ <- sendAfter delay pid Ping+  hdInbox <- receiveTimeout (asTimeout (seconds 2)) [+                      match (\m@(Ping) -> return m)+                    ]+  case hdInbox of+      Just Ping -> stash result True+      Nothing   -> stash result False++testRunAfter :: TestResult Bool -> Process ()+testRunAfter result =+  let delay = seconds 2 in do++  parentPid <- getSelfPid+  _ <- spawnLocal $ do+    _ <- runAfter delay $ send parentPid Ping+    return ()++  msg <- expectTimeout ((asTimeout delay) * 4)+  case msg of+      Just Ping -> stash result True+      Nothing   -> stash result False+  return ()++testCancelTimer :: TestResult Bool -> Process ()+testCancelTimer result = do+  let delay = milliSeconds 50+  pid <- periodically delay noop+  ref <- monitor pid++  sleep $ seconds 1+  cancelTimer pid++  _ <- receiveWait [+          match (\(ProcessMonitorNotification ref' pid' _) ->+                  stash result $ ref == ref' && pid == pid')+        ]++  return ()++testPeriodicSend :: TestResult Bool -> Process ()+testPeriodicSend result = do+  let delay = milliSeconds 100+  self <- getSelfPid+  ref <- ticker delay self+  listener 0 ref+  liftIO $ putMVar result True+  where listener :: Int -> TimerRef -> Process ()+        listener n tRef | n > 10    = cancelTimer tRef+                        | otherwise = waitOne >> listener (n + 1) tRef+        -- get a single tick, blocking indefinitely+        waitOne :: Process ()+        waitOne = do+            Tick <- expect+            return ()++testTimerReset :: TestResult Int -> Process ()+testTimerReset result = do+  let delay = seconds 10+  counter <- liftIO $ newEmptyMVar++  listenerPid <- spawnLocal $ do+      stash counter 0+      -- we continually listen for 'ticks' and increment counter for each+      forever $ do+        Tick <- expect+        liftIO $ withMVar counter (\n -> (return (n + 1)))++  -- this ticker will 'fire' every 10 seconds+  ref <- ticker delay listenerPid++  sleep $ seconds 2+  resetTimer ref++  -- at this point, the timer should be back to roughly a 5 second count down+  -- so our few remaining cycles no ticks ought to make it to the listener+  -- therefore we kill off the timer and the listener now and take the count+  cancelTimer ref+  kill listenerPid "stop!"++  -- how many 'ticks' did the listener observer? (hopefully none!)+  count <- liftIO $ takeMVar counter+  liftIO $ putMVar result count++testTimerFlush :: TestResult Bool -> Process ()+testTimerFlush result = do+  let delay = seconds 1+  self <- getSelfPid+  ref  <- ticker delay self++  -- sleep so we *should* have a message in our 'mailbox'+  sleep $ milliSeconds 2++  -- flush it out if it's there+  flushTimer ref Tick (Delay $ seconds 3)++  m <- expectTimeout 10+  case m of+      Nothing   -> stash result True+      Just Tick -> stash result False++testSleep :: TestResult Bool -> Process ()+testSleep r = do+  sleep $ seconds 20+  stash r True++--------------------------------------------------------------------------------+-- Utilities and Plumbing                                                     --+--------------------------------------------------------------------------------++tests :: LocalNode  -> [Test]+tests localNode = [+    testGroup "Timer Tests" [+        testCase "testSendAfter"+                 (delayedAssertion+                  "expected Ping within 1 second"+                  localNode True testSendAfter)+      , testCase "testRunAfter"+                 (delayedAssertion+                  "expecting run (which pings parent) within 2 seconds"+                  localNode True testRunAfter)+      , testCase "testCancelTimer"+                 (delayedAssertion+                  "expected cancelTimer to exit the timer process normally"+                  localNode True testCancelTimer)+      , testCase "testPeriodicSend"+                 (delayedAssertion+                  "expected ten Ticks to have been sent before exiting"+                  localNode True testPeriodicSend)+      , testCase "testTimerReset"+                 (delayedAssertion+                  "expected no Ticks to have been sent before resetting"+                  localNode 0 testTimerReset)+      , testCase "testTimerFlush"+                 (delayedAssertion+                  "expected all Ticks to have been flushed"+                  localNode True testTimerFlush)+      , testCase "testSleep"+                 (delayedAssertion+                  "why am I not seeing a delay!?"+                  localNode True testTimerFlush)+      ]+  ]++timerTests :: NT.Transport -> IO [Test]+timerTests transport = do+  localNode <- newLocalNode transport initRemoteTable+  let testData = tests localNode+  return testData++-- | Given a @builder@ function, make and run a test suite on a single transport+testMain :: (NT.Transport -> IO [Test]) -> IO ()+testMain builder = do+  Right (transport, _) <- createTransportExposeInternals+                                     "127.0.0.1" "10501" defaultTCPParameters+  testData <- builder transport+  defaultMain testData++main :: IO ()+main = testMain $ timerTests