diff --git a/LICENCE b/LICENCE new file mode 100644 --- /dev/null +++ b/LICENCE @@ -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. diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,3 @@ +#!/usr/bin/env runhaskell +> import Distribution.Simple +> main = defaultMain diff --git a/distributed-process-extras.cabal b/distributed-process-extras.cabal new file mode 100644 --- /dev/null +++ b/distributed-process-extras.cabal @@ -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 diff --git a/src/Control/Concurrent/Utils.hs b/src/Control/Concurrent/Utils.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Utils.hs @@ -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) diff --git a/src/Control/Distributed/Process/Extras.hs b/src/Control/Distributed/Process/Extras.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras.hs @@ -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 + diff --git a/src/Control/Distributed/Process/Extras/Call.hs b/src/Control/Distributed/Process/Extras/Call.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras/Call.hs @@ -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 () diff --git a/src/Control/Distributed/Process/Extras/Internal/Containers/MultiMap.hs b/src/Control/Distributed/Process/Extras/Internal/Containers/MultiMap.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras/Internal/Containers/MultiMap.hs @@ -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 #-} + diff --git a/src/Control/Distributed/Process/Extras/Internal/Primitives.hs b/src/Control/Distributed/Process/Extras/Internal/Primitives.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras/Internal/Primitives.hs @@ -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)))) + diff --git a/src/Control/Distributed/Process/Extras/Internal/Queue/PriorityQ.hs b/src/Control/Distributed/Process/Extras/Internal/Queue/PriorityQ.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras/Internal/Queue/PriorityQ.hs @@ -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 + diff --git a/src/Control/Distributed/Process/Extras/Internal/Queue/SeqQ.hs b/src/Control/Distributed/Process/Extras/Internal/Queue/SeqQ.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras/Internal/Queue/SeqQ.hs @@ -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 +-- 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 + diff --git a/src/Control/Distributed/Process/Extras/Internal/Types.hs b/src/Control/Distributed/Process/Extras/Internal/Types.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras/Internal/Types.hs @@ -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 diff --git a/src/Control/Distributed/Process/Extras/Internal/Unsafe.hs b/src/Control/Distributed/Process/Extras/Internal/Unsafe.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras/Internal/Unsafe.hs @@ -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") + diff --git a/src/Control/Distributed/Process/Extras/Time.hs b/src/Control/Distributed/Process/Extras/Time.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras/Time.hs @@ -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" + diff --git a/src/Control/Distributed/Process/Extras/Timer.hs b/src/Control/Distributed/Process/Extras/Timer.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras/Timer.hs @@ -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 +-- 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 diff --git a/src/Control/Distributed/Process/Extras/UnsafePrimitives.hs b/src/Control/Distributed/Process/Extras/UnsafePrimitives.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras/UnsafePrimitives.hs @@ -0,0 +1,63 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.Extras.UnsafePrimitives +-- Copyright : (c) Tim Watson 2013 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- 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 + diff --git a/tests/TestQueues.hs b/tests/TestQueues.hs new file mode 100644 --- /dev/null +++ b/tests/TestQueues.hs @@ -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 + diff --git a/tests/TestTimer.hs b/tests/TestTimer.hs new file mode 100644 --- /dev/null +++ b/tests/TestTimer.hs @@ -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