distributed-process-extras 0.2.1.2 → 0.3.9
raw patch · 20 files changed
Files
- ChangeLog +38/−9
- Setup.lhs +0/−3
- distributed-process-extras.cabal +92/−129
- src/Control/Concurrent/Utils.hs +27/−50
- src/Control/Distributed/Process/Extras.hs +9/−7
- src/Control/Distributed/Process/Extras/Internal/Containers/MultiMap.hs +14/−4
- src/Control/Distributed/Process/Extras/Internal/Primitives.hs +43/−55
- src/Control/Distributed/Process/Extras/Internal/Queue/PriorityQ.hs +0/−1
- src/Control/Distributed/Process/Extras/Internal/Queue/SeqQ.hs +14/−14
- src/Control/Distributed/Process/Extras/Internal/Types.hs +128/−118
- src/Control/Distributed/Process/Extras/Internal/Unsafe.hs +10/−2
- src/Control/Distributed/Process/Extras/Monitoring.hs +1/−2
- src/Control/Distributed/Process/Extras/SystemLog.hs +47/−65
- src/Control/Distributed/Process/Extras/Time.hs +1/−1
- src/Control/Distributed/Process/Extras/Timer.hs +11/−11
- src/Control/Distributed/Process/Extras/UnsafePrimitives.hs +1/−2
- tests/TestLog.hs +11/−17
- tests/TestPrimitives.hs +20/−24
- tests/TestQueues.hs +12/−21
- tests/TestTimer.hs +11/−17
ChangeLog view
@@ -1,3 +1,41 @@+2025-02-04 Laurent P. René de Cotret <laurent.decotret@outlook.com> 0.3.9++* Ported test suite to use `tasty` rather than `test-framework`.+* Addressed some compilation warnings (#467).++2024-10-30 David Simmons-Duffin <dsd@caltech.edu> 0.3.8++* Bump dependency bound for ansi-terminal++2024-10-30 Laurent P. René de Cotret <laurent.decotret@outlook.com> 0.3.7++* Removed dependency on `rematch` (#459)++2024-09-03 Laurent P. René de Cotret <laurent.decotret@outlook.com> 0.3.6++* Bumped dependency bounds to support GHC 8.10.7 - GHC 9.10.1+* Updated links to point to Distributed Haskell monorepo++2017-06-13 Alexander Vershilov <alexander.vershilov@gmail.com> 0.3.5+* Bump dependencies+* Cleanup code++2017-02-05 Tim Watson <watson.timothy @ google mail service dot com> 0.3.0++* re-implement whereIsRemote in terms of whereIsRemoteAsync+* re-implement whereisOrStart to avoid leaking zombie processes+* implement general NFSerializable instances+* make Resolvable instance of (NodeId, String) exception safe+* remove dependency on data-accessor+* Relax upper bound on time for testing+* documentation fixes and improvements++2017-02-05 Tim Watson <watson.timothy @ google mail service dot com> 0.3.0++* Update dependency bounds - drop support for distributed-process < 0.6.6+* Drop support for GHC < 7.10+* Fixes for testing with latest stack and CI changes+ 2016-02-16 Facundo Domínguez <facundo.dominguez@tweag.io> 0.2.1.2 * Update dependency bounds.@@ -8,12 +46,3 @@ * Add compatibility with ghc-7.10. * Fix dependency bounds. * Add missing NFData instances.--# HEAD--* Added initial GenServer module-* Added Timer Module-* Moved time functions into Time.hs-* Added Async API-* Added GenProcess API (subsumes lower level GenServer API)-
− Setup.lhs
@@ -1,3 +0,0 @@-#!/usr/bin/env runhaskell-> import Distribution.Simple-> main = defaultMain
distributed-process-extras.cabal view
@@ -1,176 +1,139 @@+cabal-version: 3.0 name: distributed-process-extras-version: 0.2.1.2-cabal-version: >=1.8+version: 0.3.9 build-type: Simple-license: BSD3+license: BSD-3-Clause license-file: LICENCE stability: experimental-Copyright: Tim Watson 2012 - 2013+Copyright: Tim Watson 2012 - 2017 Author: Tim Watson-Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>-Stability: experimental-Homepage: http://github.com/haskell-distributed/distributed-process-extras-Bug-Reports: https://cloud-haskell.atlassian.net+maintainer: The Distributed Haskell team+Homepage: http://github.com/haskell-distributed/distributed-process+Bug-Reports: http://github.com/haskell-distributed/distributed-process/issues 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.+ various libraries built on top of distributed-process category: Control-tested-with: GHC == 7.4.2 GHC == 7.6.2-data-dir: ""-extra-source-files: ChangeLog+tested-with: GHC==8.10.7 GHC==9.0.2 GHC==9.2.8 GHC==9.4.5 GHC==9.6.4 GHC==9.8.2 GHC==9.10.1 GHC==9.12.1+extra-doc-files: ChangeLog source-repository head- type: git- location: https://github.com/haskell-distributed/distributed-process-extras+ Type: git+ Location: https://github.com/haskell-distributed/distributed-process+ SubDir: packages/distributed-process-extras +common warnings+ ghc-options: -Wall+ -Wcompat+ -Widentities+ -Wincomplete-uni-patterns+ -Wincomplete-record-updates+ -Wredundant-constraints+ -fhide-source-paths+ -Wpartial-fields+ -Wunused-packages+ library- build-depends:- base >= 4.4 && < 5,- data-accessor >= 0.2.2.3,- distributed-process >= 0.5.3 && < 0.7,- binary >= 0.6.3.0 && < 0.8,- deepseq >= 1.3.0.1 && < 1.6,- mtl,- containers >= 0.4 && < 0.6,- hashable >= 1.2.0.5 && < 1.3,+ import: warnings+ build-depends: base >= 4.14 && < 5,+ distributed-process >= 0.6.0 && < 0.8,+ binary >= 0.8 && < 0.9,+ deepseq >= 1.4 && < 1.7,+ containers >= 0.6 && < 0.8,+ exceptions >= 0.10,+ hashable >= 1.2.0.5 && < 1.6, unordered-containers >= 0.2.3.0 && < 0.3, fingertree < 0.2,- stm >= 2.4 && < 2.5,- time > 1.4 && < 1.6,- 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+ stm >= 2.4 && < 2.6,+ time >= 1.5+ other-extensions: ExistentialQuantification+ HS-Source-Dirs: src exposed-modules:- Control.Distributed.Process.Extras,- Control.Distributed.Process.Extras.Call,- Control.Distributed.Process.Extras.Monitoring,- Control.Distributed.Process.Extras.SystemLog,- 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+ Control.Distributed.Process.Extras.Call+ Control.Distributed.Process.Extras.Monitoring+ Control.Distributed.Process.Extras.SystemLog+ 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+ default-language: Haskell2010 test-suite InternalQueueTests+ import: warnings type: exitcode-stdio-1.0 x-uses-tf: true build-depends:- base >= 4.4 && < 5,- ansi-terminal >= 0.5 && < 0.7,- distributed-process >= 0.5.3 && < 0.7,+ base >= 4.14 && < 5, distributed-process-extras,- distributed-process-tests >= 0.4.2 && < 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+ tasty >= 1.5 && <1.6,+ tasty-hunit >=0.10 && <0.11,+ tasty-quickcheck >=0.11 && <0.12+ hs-source-dirs: tests+ ghc-options: -rtsopts main-is: TestQueues.hs cpp-options: -DTESTING+ default-language: Haskell2010 test-suite PrimitivesTests+ import: warnings type: exitcode-stdio-1.0 x-uses-tf: true build-depends:- base >= 4.4 && < 5,- ansi-terminal >= 0.5 && < 0.7,- distributed-process >= 0.5.3 && < 0.7,+ base >= 4.14 && < 5,+ distributed-process >= 0.6.0 && < 0.8, distributed-process-extras,- distributed-process-tests >= 0.4.2 && < 0.5,- network-transport >= 0.4 && < 0.5,- mtl,- containers >= 0.4 && < 0.6,- network-transport-tcp >= 0.4 && < 0.6,- binary >= 0.6.3.0 && < 0.8,- deepseq >= 1.3.0.1 && < 1.6,- network >= 2.3 && < 2.7,- HUnit >= 1.2 && < 2,- stm >= 2.3 && < 2.5,- test-framework >= 0.6 && < 0.9,- test-framework-hunit,- rematch >= 0.2.0.0,- transformers- hs-source-dirs:- tests- ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind- extensions: CPP+ distributed-process-systest ^>= 0.4,+ network-transport >= 0.4 && < 0.6,+ network-transport-tcp >= 0.4 && < 0.9,+ tasty >= 1.5 && <1.6,+ tasty-hunit >=0.10 && <0.11,+ hs-source-dirs: tests+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind main-is: TestPrimitives.hs+ default-language: Haskell2010 test-suite TimerTests+ import: warnings type: exitcode-stdio-1.0 x-uses-tf: true build-depends:- base >= 4.4 && < 5,- ansi-terminal >= 0.5 && < 0.7,- deepseq >= 1.3.0.1 && < 1.6,- distributed-process >= 0.5.3 && < 0.7,+ base >= 4.14 && < 5,+ deepseq,+ distributed-process >= 0.6.0 && < 0.8, distributed-process-extras,- distributed-process-tests >= 0.4.2 && < 0.5,- network-transport >= 0.4 && < 0.5,- network-transport-tcp >= 0.4 && < 0.6,- 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+ distributed-process-systest ^>= 0.4,+ network-transport >= 0.4 && < 0.6,+ network-transport-tcp >= 0.4 && < 0.9,+ tasty >= 1.5 && <1.6,+ tasty-hunit >=0.10 && <0.11,+ hs-source-dirs: tests+ ghc-options: -rtsopts main-is: TestTimer.hs cpp-options: -DTESTING+ default-language: Haskell2010 test-suite LoggerTests+ import: warnings type: exitcode-stdio-1.0 -- x-uses-tf: true build-depends:- base >= 4.4 && < 5,- ansi-terminal >= 0.5 && < 0.7,- containers,- hashable,- unordered-containers >= 0.2.3.0 && < 0.3,- distributed-process >= 0.5.3 && < 0.7,+ base >= 4.14 && < 5,+ distributed-process >= 0.6.0 && < 0.8, distributed-process-extras,- distributed-process-tests >= 0.4.2 && < 0.5,- distributed-static,- bytestring,- data-accessor,- fingertree < 0.2,- network-transport >= 0.4 && < 0.5,- deepseq >= 1.3.0.1 && < 1.6,- mtl,- network-transport-tcp >= 0.4 && < 0.6,- binary >= 0.6.3.0 && < 0.8,- network >= 2.3 && < 2.7,- HUnit >= 1.2 && < 2,- stm >= 2.3 && < 2.5,- time > 1.4 && < 1.6,- test-framework >= 0.6 && < 0.9,- test-framework-hunit,- transformers,- rematch >= 0.2.0.0,- ghc-prim- hs-source-dirs:- tests- ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind -eventlog- extensions: CPP+ distributed-process-systest ^>= 0.4,+ network-transport >= 0.4 && < 0.6,+ network-transport-tcp >= 0.4 && < 0.9,+ stm,+ tasty >= 1.5 && <1.6,+ tasty-hunit >=0.10 && <0.11,+ hs-source-dirs: tests+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind main-is: TestLog.hs+ default-language: Haskell2010
src/Control/Concurrent/Utils.hs view
@@ -1,65 +1,42 @@-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE ScopedTypeVariables #-}-+{-# LANGUAGE ExistentialQuantification #-} module Control.Concurrent.Utils ( Lock()- , Exclusive(..)- , Synchronised(..)+ , mkExclusiveLock+ , mkQLock , 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.Monad.Catch (MonadMask)+import qualified Control.Monad.Catch as Catch import Control.Concurrent.MVar- ( MVar- , tryPutMVar- , newMVar+ ( newMVar , takeMVar+ , putMVar )+import Control.Concurrent.QSem 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+-- | Opaque lock.+data Lock = forall l . Lock l (l -> IO ()) (l -> IO ()) - synchronized :: (Exclusive e, Monad m) => e -> m b -> m b- synchronized = synchronised+-- | Take a lock.+acquire :: MonadIO m => Lock -> m ()+acquire (Lock l acq _) = liftIO $ acq l -instance Synchronised Lock IO where- synchronised = withLock+-- | Release lock.+release :: MonadIO m => Lock -> m ()+release (Lock l _ rel) = liftIO $ rel l -instance Synchronised Lock Process where- synchronised = withLockP+-- | Create exclusive lock. Only one process could take such lock.+mkExclusiveLock :: IO Lock+mkExclusiveLock = Lock <$> newMVar () <*> pure takeMVar <*> pure (flip putMVar ()) -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)+-- | Create quantity lock. A fixed number of processes can take this lock simultaniously.+mkQLock :: Int -> IO Lock+mkQLock n = Lock <$> newQSem n <*> pure waitQSem <*> pure signalQSem -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)+-- | Run action under a held lock.+withLock :: (MonadMask m, MonadIO m) => Lock -> m a -> m a+withLock excl =+ Catch.bracket_ (acquire excl)+ (release excl)
src/Control/Distributed/Process/Extras.hs view
@@ -33,6 +33,13 @@ 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. +[Addressing/Interaction Tools]++The various type classes exposed here, along with some common data types (such+as @Shutdown@, @ServerDisconnected@, etc.) are intended to simplify your CH+programs, and facilitate easily plugging code into higher level libraries such+as distributed-process-client-server and distributed-process-supervisor.+ [Error/Exception Handling] It is /important/ not to be too general when catching exceptions in@@ -51,11 +58,11 @@ ( -- * Exported Types Addressable- , sendToRecipient , Resolvable(..) , Routable(..) , Linkable(..) , Killable(..)+ , Monitored(..) , NFSerializable , Recipient(..) , Shutdown(..)@@ -97,8 +104,6 @@ import Control.Distributed.Process (RemoteTable) import Control.Distributed.Process.Extras.Internal.Types ( NFSerializable- , sendToRecipient- , NFSerializable , Recipient(..) , Shutdown(..) , ExitReason(..)@@ -113,12 +118,9 @@ 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-+ Control.Distributed.Process.Extras.Internal.Primitives.__remoteTable
src/Control/Distributed/Process/Extras/Internal/Containers/MultiMap.hs view
@@ -2,6 +2,7 @@ {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE TupleSections #-} module Control.Distributed.Process.Extras.Internal.Containers.MultiMap ( MultiMap@@ -10,20 +11,21 @@ , insert , member , lookup+ , delete , filter , filterWithKey+ , foldrWithKey , toList+ , size ) where import qualified Data.Foldable as Foldable import Data.Foldable (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(foldr)) import Prelude hiding (lookup, filter, pred) -- | Class of things that can be inserted in a map or@@ -44,6 +46,9 @@ empty :: MultiMap k v empty = M $ Map.empty +size :: MultiMap k v -> Int+size = Map.size . hmap+ insert :: forall k v. (Insertable k, Insertable v) => k -> v -> MultiMap k v -> MultiMap k v insert k' v' M{..} =@@ -59,6 +64,9 @@ lookup k M{..} = maybe Nothing (Just . Foldable.toList) $ Map.lookup k hmap {-# INLINE lookup #-} +delete :: (Insertable k) => k -> MultiMap k v -> Maybe ([v], MultiMap k v)+delete k m@M{..} = maybe Nothing (Just . (, M $ Map.delete k hmap)) $ lookup k m+ filter :: forall k v. (Insertable k) => (v -> Bool) -> MultiMap k v@@ -66,7 +74,8 @@ filter p M{..} = M $ Map.foldlWithKey' (matchOn p) hmap hmap where matchOn pred acc key valueSet =- Map.insert key (Set.filter pred valueSet) acc+ let vs = Set.filter pred valueSet in+ if Set.null vs then acc else Map.insert key vs acc {-# INLINE filter #-} filterWithKey :: forall k v. (Insertable k)@@ -76,7 +85,8 @@ 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+ let vs = Set.filter (pred key) valueSet in+ if Set.null vs then acc else Map.insert key vs acc {-# INLINE filterWithKey #-} -- | /O(n)/ Reduce this map by applying a binary operator to all
src/Control/Distributed/Process/Extras/Internal/Primitives.hs view
@@ -9,7 +9,7 @@ ----------------------------------------------------------------------------- -- | -- Module : Control.Distributed.Process.Extras.Internal.Primitives--- Copyright : (c) Tim Watson 2013, Parallel Scientific (Jeff Epstein) 2012+-- Copyright : (c) Tim Watson 2013 - 2017, Parallel Scientific (Jeff Epstein) 2012 -- License : BSD3 (see the file LICENSE) -- -- Maintainers : Jeff Epstein, Tim Watson@@ -27,6 +27,7 @@ , Resolvable(..) , Linkable(..) , Killable(..)+ , Monitored(..) -- * Spawning and Linking , spawnSignalled@@ -56,8 +57,8 @@ ) where import Control.Concurrent (myThreadId, throwTo)-import Control.Distributed.Process hiding (monitor)-import qualified Control.Distributed.Process as P (monitor)+import Control.Distributed.Process hiding (monitor, finally, catch)+import qualified Control.Distributed.Process as P (monitor, unmonitor) import Control.Distributed.Process.Closure (seqCP, remotable, mkClosure) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Extras.Internal.Types@@ -66,48 +67,44 @@ , Killable(..) , Resolvable(..) , Routable(..)+ , Monitored(..) , RegisterSelf(..) , ExitReason(ExitOther) , whereisRemote )-import Control.Monad (void)+import Control.Monad (void, (>=>), replicateM_)+import Control.Monad.Catch (finally, catchIf) import Data.Maybe (isJust, fromJust)+import Data.Foldable (traverse_) -- 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+monitor = resolve >=> traverse P.monitor +-- | Wait until @Resolvable@ object will exit. Return immediately+-- if object can't be resolved. 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 ())- ]+awaitExit = resolve >=> traverse_ await where+ await pid = withMonitorRef pid $ \ref -> receiveWait+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == ref)+ (\_ -> return ())+ ]+ withMonitorRef pid = bracket (P.monitor pid) P.unmonitor +-- | Send message to @Addressable@ object. deliver :: (Addressable a, Serializable m) => m -> a -> Process () deliver = flip sendTo +-- | Check if specified process is alive. Information may be outdated. isProcessAlive :: ProcessId -> Process Bool-isProcessAlive pid = getProcessInfo pid >>= \info -> return $ info /= Nothing+isProcessAlive pid = isJust <$> getProcessInfo pid -- | 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)+times = replicateM_+{-# DEPRECATED times "use replicateM_ instead" #-} -- | Like 'Control.Monad.forever' but sans space leak forever' :: Monad m => m a -> m b@@ -133,7 +130,7 @@ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef) (\(ProcessMonitorNotification _ _ dr) -> die $ ExitOther (show dr)) , matchChan recvStart (\() -> return pid)- ]+ ] `finally` (unmonitor mRef) -- | Node local version of 'Control.Distributed.Process.spawnLink'. -- Note that this is just the sequential composition of 'spawn' and 'link'.@@ -182,32 +179,23 @@ -- 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+whereisOrStart name proc = do+ (sigStart, recvStart) <- newChan+ (_, mRef) <- spawnMonitorLocal $ do+ us <- getSelfPid+ catchIf (\(ProcessRegistrationException _ r) -> isJust r)+ (register name us >> sendChan sigStart us)+ (\(ProcessRegistrationException _ rPid) ->+ sendChan sigStart $ fromJust rPid)+ proc+ receiveWait [+ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)+ (\(ProcessMonitorNotification _ _ dr) -> die $ ExitOther (show dr))+ , matchChan recvStart return+ ] `finally` (unmonitor mRef) +-- | Helper function will register itself under a given name and send+-- result to given @Process@. registerSelf :: (String, ProcessId) -> Process () registerSelf (name,target) = do self <- getSelfPid@@ -279,12 +267,12 @@ mPid <- resolve addr case mPid of Nothing -> return $ Left $ ExitOther "UnresolvedAddress"- Just p -> do- mRef <- P.monitor p- receiveWait ((matchRef mRef):matches)+ Just p ->+ bracket (P.monitor p)+ P.unmonitor+ $ \mRef -> 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))))-
src/Control/Distributed/Process/Extras/Internal/Queue/PriorityQ.hs view
@@ -35,4 +35,3 @@ {-# INLINE peek #-} peek :: Ord k => PriorityQ k v -> Maybe v peek p = maybe Nothing (\(v, _) -> Just v) $ dequeue p-
src/Control/Distributed/Process/Extras/Internal/Queue/SeqQ.hs view
@@ -1,15 +1,4 @@--------------------------------------------------------------------------------- |--- 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@.-------------------------------------------------------------------------------+{-# LANGUAGE NoImplicitPrelude #-} module Control.Distributed.Process.Extras.Internal.Queue.SeqQ ( SeqQ , empty@@ -18,16 +7,21 @@ , enqueue , dequeue , peek+ , filter+ , size ) where +-- A simple FIFO queue implementation backed by @Data.Sequence@.+import Prelude hiding (filter, length) import Data.Sequence ( Seq , ViewR(..) , (<|) , viewr+ , length )-import qualified Data.Sequence as Seq (empty, singleton, null)+import qualified Data.Sequence as Seq (empty, singleton, null, filter) newtype SeqQ a = SeqQ { q :: Seq a } deriving (Show)@@ -58,9 +52,15 @@ peek :: SeqQ a -> Maybe a peek s = maybe Nothing (\(_ :> a) -> Just a) $ getR s +{-# INLINE size #-}+size :: SeqQ a -> Int+size = length . q++filter :: (a -> Bool) -> SeqQ a -> SeqQ a+filter c s = SeqQ $ Seq.filter c (q s)+ getR :: SeqQ a -> Maybe (ViewR a) getR s = case (viewr (q s)) of EmptyR -> Nothing a -> Just a-
src/Control/Distributed/Process/Extras/Internal/Types.hs view
@@ -1,12 +1,11 @@ {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE StandaloneDeriving #-}-{-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE UndecidableInstances #-}-{-# LANGUAGE OverlappingInstances #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE FunctionalDependencies #-} -- | Types used throughout the Extras package --@@ -21,8 +20,8 @@ , Killable(..) , Resolvable(..) , Routable(..)+ , Monitored(..) , Addressable- , sendToRecipient , Recipient(..) , RegisterSelf(..) -- * Interactions@@ -34,8 +33,6 @@ , ExitReason(..) , ServerDisconnected(..) , NFSerializable- -- remote table- , __remoteTable ) where import Control.Concurrent.MVar@@ -44,20 +41,18 @@ , modifyMVar ) import Control.DeepSeq (NFData(..), ($!!))-import Control.Distributed.Process hiding (send)+import Control.Distributed.Process hiding (send, catch) import qualified Control.Distributed.Process as P ( send , unsafeSend , unsafeNSend )-import Control.Distributed.Process.Closure- ( remotable- , mkClosure- , functionTDict- ) import Control.Distributed.Process.Serializable-+import Control.Exception (SomeException)+import Control.Monad.Catch (catch) import Data.Binary+import Data.Foldable (traverse_)+import Data.Maybe (fromJust) import Data.Typeable (Typeable) import GHC.Generics @@ -76,6 +71,8 @@ class (NFData a, Serializable a) => NFSerializable a instance (NFData a, Serializable a) => NFSerializable a +instance (NFSerializable a) => NFSerializable (SendPort a)+ -- | Tags provide uniqueness for messages, so that they can be -- matched with their response. type Tag = Int@@ -96,6 +93,21 @@ getTag :: TagPool -> Process Tag getTag tp = liftIO $ modifyMVar tp (\tag -> return (tag+1,tag)) +-- | A synchronous version of 'whereis', this monitors the remote node+-- and returns @Nothing@ if the node goes down (since a remote node failing+-- or being non-contactible has the same effect as a process not being+-- registered from the caller's point of view).+whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)+whereisRemote node name = do+ mRef <- monitorNode node+ whereisRemoteAsync node name+ receiveWait [ matchIf (\(NodeMonitorNotification ref nid _) -> ref == mRef &&+ nid == node)+ (\NodeMonitorNotification{} -> return Nothing)+ , matchIf (\(WhereIsReply n _) -> n == name)+ (\(WhereIsReply _ mPid) -> return mPid)+ ]+ -- | Wait cancellation message. data CancelWait = CancelWait deriving (Eq, Show, Typeable, Generic)@@ -128,54 +140,46 @@ 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-instance NFData Recipient where- rnf (Pid p) = rnf p `seq` ()- rnf (Registered s) = rnf s `seq` ()- rnf (RemoteRegistered s n) = rnf s `seq` rnf n `seq` ()---- useful exit reasons+baseAddressableErrorMessage :: (Resolvable a) => a -> String+baseAddressableErrorMessage _ = "CannotResolveAddressable" --- | Given when a server is unobtainable.-data ServerDisconnected = ServerDisconnected !DiedReason- deriving (Typeable, Generic)-instance Binary ServerDisconnected where-instance NFData ServerDisconnected where+-- | Class of things to which a @Process@ can /link/ itself.+class Linkable a where+ -- | Create a /link/ with the supplied object.+ linkTo :: (Resolvable a) => a -> Process ()+ linkTo r = resolve r >>= traverse_ link -$(remotable ['whereis])+class Monitored a r m | a r -> m where+ mkMonitor :: a -> Process r+ checkMonitor :: a -> r -> m -> Process Bool --- | 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)+instance (Resolvable a) => Monitored a MonitorRef ProcessMonitorNotification where+ mkMonitor a = monitor . fromJust =<< resolve a+ checkMonitor p r (ProcessMonitorNotification ref pid _) = do+ p' <- resolve p+ case p' of+ Nothing -> return False+ Just pr -> return $ ref == r && pid == pr -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+-- | Class of things that can be killed (or instructed to exit).+class Killable p where+ -- | Kill (instruct to exit) generic process, using 'kill' primitive.+ killProc :: Resolvable p => p -> String -> Process ()+ killProc r s = resolve r >>= traverse_ (flip kill $ s) -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+ -- | Kill (instruct to exit) generic process, using 'exit' primitive.+ exitProc :: (Resolvable p, Serializable m) => p -> m -> Process ()+ exitProc r m = resolve r >>= traverse_ (flip exit $ m) -baseAddressableErrorMessage :: (Routable a) => a -> String-baseAddressableErrorMessage _ = "CannotResolveAddressable"+instance Resolvable p => Killable p --- | Class of things to which a @Process@ can /link/ itself.-class Linkable a where- -- | Create a /link/ with the supplied object.- linkTo :: a -> Process ()+-- | resolve the Resolvable or die with specified msg plus details of what didn't resolve+resolveOrDie :: (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 -- | Class of things that can be resolved to a 'ProcessId'. --@@ -183,99 +187,105 @@ -- | 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 ()+ -- | Unresolvable @Addressable@ Message+ unresolvableMessage :: (Resolvable a) => a -> String+ unresolvableMessage = baseAddressableErrorMessage -instance Killable ProcessId where- killProc = kill- exitProc = exit+instance Resolvable ProcessId where+ resolve p = return (Just p)+ unresolvableMessage p = "CannotResolvePid[" ++ (show p) ++ "]" -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)+instance Resolvable String where+ resolve = whereis+ unresolvableMessage s = "CannotResolveRegisteredName[" ++ s ++ "]" --- | Provides a unified API for addressing processes.----class Routable a where- -- | Send a message to the target asynchronously- sendTo :: (Serializable m) => a -> m -> Process ()+instance Resolvable (NodeId, String) where+ resolve (nid, pname) =+ whereisRemote nid pname `catch` (\(_ :: SomeException) -> return Nothing)+ unresolvableMessage (n, s) =+ "CannotResolveRemoteRegisteredName[name: " ++ s ++ ", node: " ++ (show n) ++ "]" - -- | Send some @NFData@ message to the target asynchronously,- -- forcing evaluation (i.e., @deepseq@) beforehand.- unsafeSendTo :: (NFSerializable m) => a -> m -> Process ()+-- Provide a unified API for addressing processes. - -- | Unresolvable @Addressable@ Message- unresolvableMessage :: a -> String- unresolvableMessage = baseAddressableErrorMessage+-- | Class of things that you can route/send serializable message to+class Routable a where -instance (Resolvable a) => Routable a where+ -- | Send a message to the target asynchronously+ sendTo :: (Serializable m, Resolvable a) => a -> m -> Process () sendTo a m = do mPid <- resolve a maybe (die (unresolvableMessage a)) (\p -> P.send p m) mPid + -- | Send some @NFData@ message to the target asynchronously,+ -- forcing evaluation (i.e., @deepseq@) beforehand.+ unsafeSendTo :: (NFSerializable m, Resolvable a) => a -> m -> Process () 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) ++ "]"+ sendTo (nid, pname) = nsendRemote nid pname+ unsafeSendTo = sendTo -- because serialisation *must* take place 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+instance Addressable ProcessId --- 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+-- | 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+instance NFData Recipient where+ rnf (Pid p) = rnf p `seq` ()+ rnf (Registered s) = rnf s `seq` ()+ rnf (RemoteRegistered s n) = rnf s `seq` rnf n `seq` ()++instance Resolvable Recipient where+ resolve (Pid p) = return (Just p)+ resolve (Registered n) = whereis n+ resolve (RemoteRegistered s n) = whereisRemote n s++ unresolvableMessage (Pid p) = unresolvableMessage p+ unresolvableMessage (Registered n) = unresolvableMessage n+ unresolvableMessage (RemoteRegistered s n) = unresolvableMessage (n, s)++-- although we have an instance of Routable for Resolvable, it really+-- makes no sense to do remote lookups on a pid, only to then send to it!+instance Routable Recipient where++ sendTo (Pid p) m = P.send p m+ sendTo (Registered s) m = nsend s m+ sendTo (RemoteRegistered s n) m = nsendRemote n s m++ unsafeSendTo (Pid p) m = P.unsafeSend p $!! m+ unsafeSendTo (Registered s) m = P.unsafeNSend s $!! m+ unsafeSendTo (RemoteRegistered s n) m = nsendRemote n s m++-- useful exit reasons++-- | Given when a server is unobtainable.+newtype ServerDisconnected = ServerDisconnected DiedReason+ deriving (Typeable, Generic)+instance Binary ServerDisconnected where+instance NFData ServerDisconnected where
src/Control/Distributed/Process/Extras/Internal/Unsafe.hs view
@@ -1,11 +1,20 @@ {-# 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! --+-- [Implementation Notes]+-- This module provides facilities for forcibly sending non-serializable+-- data via cloud haskell's messaging primitives, such as @send@ et al.+-- Of course, if you attmept to do this when interacting with a remote process,+-- your application will break.+--+-- NB: this module will be deprecated in the next dot release, pending rewrite+-- of the libraries that currently rely on it, to use the new supporting APIs+-- for STM interactions in distributed-process-client-server.+-- module Control.Distributed.Process.Extras.Internal.Unsafe ( -- * Copying non-serializable data PCopy()@@ -125,4 +134,3 @@ matchInputStream (ReadSTM stm) = matchSTM stm return matchInputStream Null = match (\NullInputStream -> do error "NullInputStream")-
src/Control/Distributed/Process/Extras/Monitoring.hs view
@@ -4,7 +4,7 @@ ----------------------------------------------------------------------------- -- | -- Module : Control.Distributed.Process.Extras.Monitoring--- Copyright : (c) Tim Watson 2013 - 2014+-- Copyright : (c) Tim Watson 2013 - 2017 -- License : BSD3 (see the file LICENSE) -- -- Maintainer : Tim Watson <watson.timothy@gmail.com>@@ -136,4 +136,3 @@ nodeUp = NodeUp . NodeId nodeDown = NodeDown . NodeId-
src/Control/Distributed/Process/Extras/SystemLog.hs view
@@ -1,17 +1,17 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE PatternGuards #-}+{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE TypeSynonymInstances #-} {-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE OverlappingInstances #-}-{-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ConstrainedClassMethods #-} ----------------------------------------------------------------------------- -- | -- Module : Control.Distributed.Process.Extras.SystemLog--- Copyright : (c) Tim Watson 2013 - 2014+-- Copyright : (c) Tim Watson 2013 - 2017 -- License : BSD3 (see the file LICENSE) -- -- Maintainer : Tim Watson <watson.timothy@gmail.com>@@ -44,7 +44,7 @@ , LogFormat , LogClient , LogChan- , LogText(..)+ , LogText , ToLog(..) , Logger(..) -- * Mx Agent Configuration / Startup@@ -69,7 +69,7 @@ ) where import Control.DeepSeq (NFData(..))-import Control.Distributed.Process+import Control.Distributed.Process hiding (catch) import Control.Distributed.Process.Management ( MxEvent(MxConnected, MxDisconnected, MxLog, MxUser) , MxAgentId(..)@@ -86,25 +86,16 @@ import Control.Distributed.Process.Extras ( Resolvable(..) , Routable(..)+ , Addressable ) import Control.Distributed.Process.Serializable import Control.Exception (SomeException)-import Data.Accessor- ( Accessor- , accessor- , (^:)- , (^=)- , (^.)- )+import Control.Monad.Catch (catch) import Data.Binary import Data.Typeable (Typeable) import GHC.Generics -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch, error, Read)-#else import Prelude hiding (error, Read)-#endif import System.IO ( IOMode(AppendMode)@@ -140,11 +131,11 @@ instance Binary AddFormatter data LogState =- LogState { output :: !(String -> Process ())- , cleanup :: !(Process ())- , _level :: !LogLevel- , _format :: !(String -> Process String)- , _formatters :: ![Message -> Process (Maybe String)]+ LogState { output :: !(String -> Process ())+ , cleanup :: !(Process ())+ , level :: !LogLevel+ , format :: !(String -> Process String)+ , formatters :: ![Message -> Process (Maybe String)] } data LogMessage =@@ -156,25 +147,26 @@ type LogFormat = String -> Process String -type LogChan = ()+type LogChanT = ()++newtype LogChan = LogChan LogChanT instance Routable LogChan where sendTo _ = mxNotify unsafeSendTo _ = mxNotify -data LogText = LogText { txt :: !String }+type LogText = String newtype LogClient = LogClient { agent :: ProcessId } instance Resolvable LogClient where resolve = return . Just . agent+instance Routable LogClient class ToLog m where- toLog :: m -> Process (LogLevel -> LogMessage)+ toLog :: (Serializable m) => m -> Process (LogLevel -> LogMessage)+ toLog = return . LogData . unsafeWrapMessage instance ToLog LogText where- toLog = return . LogMessage . txt--instance (Serializable a) => ToLog a where- toLog = return . LogData . unsafeWrapMessage+ toLog = return . LogMessage instance ToLog Message where toLog = return . LogData@@ -195,46 +187,46 @@ mxLogId = MxAgentId logProcessName logChannel :: LogChan-logChannel = ()+logChannel = LogChan () report :: (Logger l) => (l -> LogText -> Process ()) -> l -> String -> Process ()-report f l = f l . LogText+report f l = f l client :: Process (Maybe LogClient) client = resolve logProcessName >>= return . maybe Nothing (Just . LogClient) -debug :: (Logger l, ToLog m) => l -> m -> Process ()+debug :: (Logger l, Serializable m, ToLog m) => l -> m -> Process () debug l m = sendLog l m Debug -info :: (Logger l, ToLog m) => l -> m -> Process ()+info :: (Logger l, Serializable m, ToLog m) => l -> m -> Process () info l m = sendLog l m Info -notice :: (Logger l, ToLog m) => l -> m -> Process ()+notice :: (Logger l, Serializable m, ToLog m) => l -> m -> Process () notice l m = sendLog l m Notice -warning :: (Logger l, ToLog m) => l -> m -> Process ()+warning :: (Logger l, Serializable m, ToLog m) => l -> m -> Process () warning l m = sendLog l m Warning -error :: (Logger l, ToLog m) => l -> m -> Process ()+error :: (Logger l, Serializable m, ToLog m) => l -> m -> Process () error l m = sendLog l m Error -critical :: (Logger l, ToLog m) => l -> m -> Process ()+critical :: (Logger l, Serializable m, ToLog m) => l -> m -> Process () critical l m = sendLog l m Critical -alert :: (Logger l, ToLog m) => l -> m -> Process ()+alert :: (Logger l, Serializable m, ToLog m) => l -> m -> Process () alert l m = sendLog l m Alert -emergency :: (Logger l, ToLog m) => l -> m -> Process ()+emergency :: (Logger l, Serializable m, ToLog m) => l -> m -> Process () emergency l m = sendLog l m Emergency -sendLog :: (Logger l, ToLog m) => l -> m -> LogLevel -> Process ()+sendLog :: (Logger l, Serializable m, ToLog m) => l -> m -> LogLevel -> Process () sendLog a m lv = toLog m >>= \m' -> logMessage a $ m' lv -addFormatter :: (Routable r)+addFormatter :: (Addressable r) => r -> Closure (Message -> Process (Maybe String)) -> Process ()@@ -265,17 +257,17 @@ systemLog o c l f = go $ LogState o c l f defaultFormatters where go :: LogState -> Process ProcessId- go st = do+ go st = mxAgentWithFinalize mxLogId st [ -- these are the messages we're /really/ interested in- (mxSink $ \(m :: LogMessage) -> do+ (mxSink $ \(m :: LogMessage) -> case m of- (LogMessage msg lvl) -> do+ (LogMessage msg lvl) -> mxGetLocal >>= outputMin lvl msg >> mxReceive (LogData dat lvl) -> handleRawMsg dat lvl) -- complex messages rely on properly registered formatters- , (mxSink $ \(ev :: MxEvent) -> do+ , (mxSink $ \(ev :: MxEvent) -> case ev of (MxUser msg) -> handleRawMsg msg Debug -- we treat trace/log events like regular log events at@@ -285,14 +277,14 @@ -- command message handling , (mxSink $ \(SetLevel lvl) ->- mxGetLocal >>= mxSetLocal . (level ^= lvl) >> mxReceive)+ mxGetLocal >>= \st' -> mxSetLocal st' { level = lvl } >> mxReceive) , (mxSink $ \(AddFormatter f') -> do fmt <- liftMX $ catch (unClosure f' >>= return . Just) (\(_ :: SomeException) -> return Nothing) case fmt of Nothing -> mxReady Just mf -> do- mxUpdateLocal (formatters ^: (mf:))+ mxUpdateLocal (\s -> s { formatters = mf:formatters s }) mxReceive) ] runCleanup @@ -305,15 +297,15 @@ Just str -> outputMin lvl' str st >> mxReceive Nothing -> mxReceive -- we cannot format a Message, so we ignore it - handleEvent (MxConnected _ ep) = do+ handleEvent (MxConnected _ ep) = mxGetLocal >>= outputMin Notice- ("Endpoint: " ++ (show ep) ++ " Disconnected")- handleEvent (MxDisconnected _ ep) = do+ ("Endpoint: " ++ show ep ++ " Disconnected")+ handleEvent (MxDisconnected _ ep) = mxGetLocal >>= outputMin Notice- ("Endpoint " ++ (show ep) ++ " Connected")+ ("Endpoint " ++ show ep ++ " Connected") handleEvent _ = return () - formatMsg m st = let fms = st ^. formatters in formatMsg' m fms+ formatMsg m LogState{..} = let fms = formatters in formatMsg' m fms formatMsg' _ [] = return Nothing formatMsg' m (f':fs) = do@@ -322,22 +314,12 @@ ok@(Just _) -> return ok Nothing -> formatMsg' m fs - outputMin minLvl msgData st =- case minLvl >= (st ^. level) of- True -> liftMX $ ((st ^. format) msgData >>= (output st))+ outputMin minLvl msgData LogState{..} =+ case minLvl >= level of+ True -> liftMX (format msgData >>= output) False -> return () defaultFormatters = [basicDataFormat] basicDataFormat :: Message -> Process (Maybe String) basicDataFormat = unwrapMessage--level :: Accessor LogState LogLevel-level = accessor _level (\l s -> s { _level = l })--format :: Accessor LogState LogFormat-format = accessor _format (\f s -> s { _format = f })--formatters :: Accessor LogState [Message -> Process (Maybe String)]-formatters = accessor _formatters (\n' st -> st { _formatters = n' })-
src/Control/Distributed/Process/Extras/Time.hs view
@@ -155,7 +155,7 @@ {-# INLINE hoursPerDay #-} hoursPerDay :: Int-hoursPerDay = 60+hoursPerDay = 24 {-# INLINE minutesPerHour #-} minutesPerHour :: Int
src/Control/Distributed/Process/Extras/Timer.hs view
@@ -1,12 +1,10 @@ {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE PatternGuards #-}-{-# LANGUAGE TemplateHaskell #-} ----------------------------------------------------------------------------- -- | -- Module : Control.Distributed.Process.Extras.Timer--- Copyright : (c) Tim Watson 2012+-- Copyright : (c) Tim Watson 2012 - 2017 -- License : BSD3 (see the file LICENSE) -- -- Maintainer : Tim Watson <watson.timothy@gmail.com>@@ -36,12 +34,13 @@ , flushTimer ) where -import Control.DeepSeq (NFData)+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 Control.Monad (unless, void) import Data.Binary import Data.Typeable (Typeable) import Prelude hiding (init)@@ -56,12 +55,14 @@ deriving (Typeable, Generic, Eq, Show) instance Binary TimerConfig where instance NFData TimerConfig where+ rnf tc = tc `seq` () -- | represents a 'tick' event that timers can generate data Tick = Tick deriving (Typeable, Generic, Eq, Show) instance Binary Tick where instance NFData Tick where+ rnf t = t `seq` () data SleepingPill = SleepingPill deriving (Typeable, Generic, Eq, Show)@@ -95,7 +96,7 @@ -> a -> Process TimerRef sendAfter t pid msg = runAfter t proc- where proc = do { send pid msg }+ where proc = send pid msg -- | runs the supplied process action(s) after @t@ has elapsed runAfter :: TimeInterval -> Process () -> Process TimerRef@@ -134,11 +135,11 @@ -- 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+resetTimer = flip send Reset -- | permanently cancels a timer cancelTimer :: TimerRef -> Process ()-cancelTimer = (flip send) Cancel+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@@ -151,9 +152,9 @@ 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 ()+ performFlush mRef (Delay i) = void (receiveTimeout (asTimeout i) (filters mRef)) filters mRef = [- matchIf (\x -> x == ignore)+ matchIf (== ignore) (\_ -> return ()) , matchIf (\(ProcessMonitorNotification mRef' _ _) -> mRef == mRef') (\_ -> return ()) ]@@ -174,7 +175,6 @@ case cancel of Nothing -> runProc cancelOnReset Just Cancel -> return ()- Just Reset -> if cancelOnReset then return ()- else runTimer t proc cancelOnReset+ Just Reset -> unless cancelOnReset $ runTimer t proc cancelOnReset where runProc True = proc runProc False = proc >> runTimer t proc cancelOnReset
src/Control/Distributed/Process/Extras/UnsafePrimitives.hs view
@@ -1,7 +1,7 @@ ----------------------------------------------------------------------------- -- | -- Module : Control.Distributed.Process.Extras.UnsafePrimitives--- Copyright : (c) Tim Watson 2013+-- Copyright : (c) Tim Watson 2013 - 2017 -- License : BSD3 (see the file LICENSE) -- -- Maintainer : Tim Watson <watson.timothy@gmail.com>@@ -60,4 +60,3 @@ -- | Create an unencoded @Message@ for any @Serializable@ type. wrapMessage :: NFSerializable a => a -> Message wrapMessage msg = Unsafe.wrapMessage $!! msg-
tests/TestLog.hs view
@@ -1,9 +1,9 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE ScopedTypeVariables #-} module Main where --- import Control.Exception (SomeException) import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar, newEmptyMVar) import Control.Concurrent.STM (atomically) import Control.Concurrent.STM.TChan@@ -11,22 +11,18 @@ import Control.Distributed.Process.Closure (remotable, mkStaticClosure) import Control.Distributed.Process.Node import Control.Distributed.Process.Extras hiding (__remoteTable)-import qualified Control.Distributed.Process.Extras.SystemLog as Log (Logger, error)+import qualified Control.Distributed.Process.Extras.SystemLog as Log (Logger) import Control.Distributed.Process.Extras.SystemLog hiding (Logger, error)-import Control.Distributed.Process.Tests.Internal.Utils+import Control.Distributed.Process.SysTest.Utils import Control.Distributed.Process.Extras.Time import Control.Distributed.Process.Extras.Timer import Control.Monad (void) import Data.List (delete) -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch, drop, Read)-#else import Prelude hiding (drop, read, Read)-#endif -import Test.Framework (Test, testGroup, defaultMain)-import Test.Framework.Providers.HUnit (testCase)+import Test.Tasty (TestTree, testGroup, defaultMain)+import Test.Tasty.HUnit (testCase) import Network.Transport.TCP import qualified Network.Transport as NT @@ -34,7 +30,6 @@ import Text.ParserCombinators.ReadP as P import Text.ParserCombinators.ReadPrec -import qualified Network.Transport as NT logLevelFormatter :: Message -> Process (Maybe String) logLevelFormatter m = handleMessage m showLevel@@ -59,7 +54,7 @@ where writeLog chan = liftIO . atomically . writeTChan chan -testLogLevels :: (Log.Logger logger, ToLog tL)+testLogLevels :: (Log.Logger logger, NFSerializable tL, ToLog tL) => MVar () -> TChan String -> logger@@ -103,7 +98,7 @@ lift P.skipSpaces return x -tests :: NT.Transport -> IO [Test]+tests :: NT.Transport -> IO TestTree tests transport = do let ch = logChannel localNode <- newLocalNode transport $ __remoteTable initRemoteTable@@ -112,7 +107,7 @@ void $ forkProcess localNode $ do (_, chan) <- testLoggingProcess liftIO $ putMVar ex chan chan <- takeMVar ex- return [+ return $ testGroup "TestLog" [ testGroup "Log Reports / LogText" (map (mkTestCase lock chan ch simpleShowToLog localNode) (enumFromTo Debug Emergency)) , testGroup "Logging Raw Messages"@@ -126,15 +121,14 @@ testCase l (delayedAssertion ("Expected up to " ++ l) ln True $ testLogLevels lck chan ch' Debug lvl rdr) - simpleShowToLog = (LogText . show)+ simpleShowToLog = show messageToLog = unsafeWrapMessage . show messageRaw = unsafeWrapMessage -- | Given a @builder@ function, make and run a test suite on a single transport-testMain :: (NT.Transport -> IO [Test]) -> IO ()+testMain :: (NT.Transport -> IO TestTree) -> IO () testMain builder = do- Right (transport, _) <- createTransportExposeInternals- "127.0.0.1" "0" defaultTCPParameters+ Right (transport, _) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters testData <- builder transport defaultMain testData
tests/TestPrimitives.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE TemplateHaskell #-} @@ -15,19 +16,13 @@ import Control.Distributed.Process.Extras.Monitoring import Control.Distributed.Process.Extras.Time import Control.Monad (void)-import Control.Rematch hiding (match)-import qualified Network.Transport as NT (Transport) import Network.Transport.TCP()-#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif -import Test.HUnit (Assertion)-import Test.Framework (Test, testGroup, defaultMain)-import Test.Framework.Providers.HUnit (testCase)+import Test.Tasty (TestTree, testGroup, defaultMain)+import Test.Tasty.HUnit (Assertion, assertEqual, assertBool, testCase) import Network.Transport.TCP import qualified Network.Transport as NT-import Control.Distributed.Process.Tests.Internal.Utils+import Control.Distributed.Process.SysTest.Utils testLinkingWithNormalExits :: TestResult DiedReason -> Process () testLinkingWithNormalExits result = do@@ -115,8 +110,9 @@ mn1 <- liftIO $ takeMVar nid2 mn2 <- liftIO $ takeMVar nid3 - [mn1, mn2] `shouldContain` n1- [mn1, mn2] `shouldContain` n2+ liftIO $ do+ assertBool mempty $ n1 `elem` [mn1, mn2]+ assertBool mempty $ n2 `elem` [mn1, mn2] nid4 <- liftIO $ newEmptyMVar node4 <- liftIO $ newLocalNode transport initRemoteTable@@ -127,7 +123,7 @@ mn3 <- liftIO $ takeMVar nid4 NodeUp n3 <- expect- mn3 `shouldBe` (equalTo n3)+ liftIO $ assertEqual mempty n3 mn3 liftIO $ closeLocalNode node4 stash result ()@@ -181,8 +177,8 @@ -- Utilities and Plumbing -- -------------------------------------------------------------------------------- -tests :: NT.Transport -> LocalNode -> [Test]-tests transport localNode = [+tests :: NT.Transport -> LocalNode -> TestTree+tests transport localNode = testGroup "TestPrimitives" [ testGroup "Linking Tests" [ testCase "testLinkingWithNormalExits" (delayedAssertion@@ -195,26 +191,26 @@ ], testGroup "Call/RPC" [ testCase "multicallTest" (multicallTest transport)- ],- testGroup "Node Monitoring" [- testCase "Death Notifications"- (delayedAssertion- "subscribers should both have received NodeDown twice"- localNode () (testMonitorNodeDeath transport)) ]+ -- TODO: the test below has been very flaky in CI+ -- testGroup "Node Monitoring" [+ -- testCase "Death Notifications"+ -- (delayedAssertion+ -- "subscribers should both have received NodeDown twice"+ -- localNode () (testMonitorNodeDeath transport))+ -- ] ] -primitivesTests :: NT.Transport -> IO [Test]+primitivesTests :: NT.Transport -> IO TestTree primitivesTests transport = do localNode <- newLocalNode transport initRemoteTable let testData = tests transport localNode return testData -- | Given a @builder@ function, make and run a test suite on a single transport-testMain :: (NT.Transport -> IO [Test]) -> IO ()+testMain :: (NT.Transport -> IO TestTree) -> IO () testMain builder = do- Right (transport, _) <- createTransportExposeInternals- "127.0.0.1" "0" defaultTCPParameters+ Right (transport, _) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters testData <- builder transport defaultMain testData
tests/TestQueues.hs view
@@ -1,26 +1,19 @@ {-# LANGUAGE PatternGuards #-}+{-# OPTIONS_GHC -Wno-incomplete-uni-patterns #-} 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 Data.List ( sortBy )+import Test.Tasty (TestTree, testGroup, defaultMain)+import Test.Tasty.HUnit (assertEqual, assertBool, testCase)+import Test.Tasty.QuickCheck (testProperty) 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!@@ -47,8 +40,8 @@ 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.+ d = FIFO.dequeue q4+ in (d == Nothing) && (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 =@@ -56,8 +49,8 @@ Just (_, q') = FIFO.dequeue q in (FIFO.isEmpty q') == ((FIFO.isEmpty q) == False) -tests :: [TF.Test]-tests = [+tests :: TestTree+tests = testGroup "TestQueues" [ testGroup "Priority Queue Tests" [ -- testCase "New Queue Should Be Empty" -- (expect (PQ.isEmpty $ PQ.empty) $ equalTo True),@@ -72,13 +65,11 @@ ], testGroup "FIFO Queue Tests" [ testCase "New Queue Should Be Empty"- (expectThat (FIFO.isEmpty $ FIFO.empty) $ equalTo True),+ (assertBool mempty (FIFO.isEmpty $ FIFO.empty)), testCase "Singleton Queue Should Contain One Element"- (expectThat (FIFO.dequeue $ FIFO.singleton "hello") $- equalTo $ Just ("hello", FIFO.empty)),+ (assertEqual mempty (FIFO.dequeue $ FIFO.singleton "hello") $ Just ("hello", FIFO.empty)), testCase "Dequeue Empty Queue Should Be Nothing"- (expectThat (FIFO.dequeue $ (FIFO.empty :: SeqQ ())) $- is (Nothing :: Maybe ((), SeqQ ()))),+ (assertEqual mempty (FIFO.dequeue $ (FIFO.empty :: SeqQ ())) $ (Nothing :: Maybe ((), SeqQ ()))), testProperty "Enqueue/Dequeue should respect FIFO order" prop_fifo_enqueue, testProperty "Enqueue/Dequeue should respect isEmpty"
tests/TestTimer.hs view
@@ -1,8 +1,7 @@+{-# OPTIONS_GHC -Wno-orphans #-}+{-# LANGUAGE CPP #-} module Main where -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif import Control.Monad (forever) import Control.Concurrent.MVar ( newEmptyMVar@@ -10,7 +9,6 @@ , takeMVar , withMVar )-import qualified Network.Transport as NT (Transport) import Network.Transport.TCP() import Control.DeepSeq (NFData) import Control.Distributed.Process@@ -18,16 +16,13 @@ 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 Control.Distributed.Process.SysTest.Utils -import Test.Framework (Test, testGroup, defaultMain)-import Test.Framework.Providers.HUnit (testCase)+import Test.Tasty (TestTree, testGroup, defaultMain)+import Test.Tasty.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 ()@@ -63,8 +58,8 @@ let delay = milliSeconds 50 pid <- periodically delay noop ref <- monitor pid+ _ <- getProcessInfo pid - sleep $ seconds 1 cancelTimer pid _ <- receiveWait [@@ -144,8 +139,8 @@ -- Utilities and Plumbing -- -------------------------------------------------------------------------------- -tests :: LocalNode -> [Test]-tests localNode = [+tests :: LocalNode -> TestTree+tests localNode = testGroup "TestTimer" [ testGroup "Timer Tests" [ testCase "testSendAfter" (delayedAssertion@@ -178,17 +173,16 @@ ] ] -timerTests :: NT.Transport -> IO [Test]+timerTests :: NT.Transport -> IO TestTree 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 :: (NT.Transport -> IO TestTree) -> IO () testMain builder = do- Right (transport, _) <- createTransportExposeInternals- "127.0.0.1" "0" defaultTCPParameters+ Right (transport, _) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters testData <- builder transport defaultMain testData