diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,7 @@ +2017-06-13 Alexander Vershilov 0.3.5 +* Bump dependencies +* Cleanup code + 2017-02-05 Tim Watson 0.3.0 * re-implement whereIsRemote in terms of whereIsRemoteAsync diff --git a/distributed-process-extras.cabal b/distributed-process-extras.cabal --- a/distributed-process-extras.cabal +++ b/distributed-process-extras.cabal @@ -1,5 +1,5 @@ name: distributed-process-extras -version: 0.3.2 +version: 0.3.5 cabal-version: >=1.8 build-type: Simple license: BSD3 @@ -8,7 +8,6 @@ Copyright: Tim Watson 2012 - 2017 Author: Tim Watson Maintainer: Tim Watson -Stability: experimental Homepage: http://github.com/haskell-distributed/distributed-process-extras Bug-Reports: http://github.com/haskell-distributed/distributed-process-extras/issues synopsis: Cloud Haskell Extras @@ -30,8 +29,8 @@ default: False library - build-depends: base >= 4.6 && < 5, - distributed-process >= 0.6.6 && < 0.7, + build-depends: base >= 4.8 && < 5, + distributed-process >= 0.6.0 && < 0.8, binary >= 0.6.3.0 && < 0.9, deepseq >= 1.3.0.1 && < 1.6, mtl >= 2.0 && < 2.4, @@ -71,10 +70,10 @@ x-uses-tf: true build-depends: base >= 4.6 && < 5, - ansi-terminal >= 0.5 && < 0.7, - distributed-process >= 0.6.6 && < 0.7, + ansi-terminal >= 0.5 && < 0.9, + distributed-process >= 0.6.0 && < 0.8, distributed-process-extras, - distributed-process-systest >= 0.1.0 && < 0.2.0, + distributed-process-systest >= 0.1.0 && < 0.3.0, HUnit >= 1.2 && < 2, test-framework >= 0.6 && < 0.9, test-framework-hunit, @@ -94,11 +93,11 @@ x-uses-tf: true build-depends: base >= 4.6 && < 5, - ansi-terminal >= 0.5 && < 0.7, - distributed-process >= 0.6.6 && < 0.7, + ansi-terminal >= 0.5 && < 0.9, + distributed-process >= 0.6.0 && < 0.8, distributed-process-extras, - distributed-process-systest >= 0.1.0 && < 0.2.0, - network-transport >= 0.4 && < 0.5, + distributed-process-systest >= 0.1.0 && < 0.3.0, + network-transport >= 0.4 && < 0.6, mtl, containers >= 0.4 && < 0.6, network-transport-tcp >= 0.4 && < 0.6, @@ -122,12 +121,12 @@ x-uses-tf: true build-depends: base >= 4.6 && < 5, - ansi-terminal >= 0.5 && < 0.7, + ansi-terminal >= 0.5 && < 0.9, deepseq >= 1.3.0.1 && < 1.6, - distributed-process >= 0.6.6 && < 0.7, + distributed-process >= 0.6.0 && < 0.8, distributed-process-extras, - distributed-process-systest >= 0.1.0 && < 0.2.0, - network-transport >= 0.4 && < 0.5, + distributed-process-systest >= 0.1.0 && < 0.3.0, + network-transport >= 0.4 && < 0.6, network-transport-tcp >= 0.4 && < 0.6, HUnit >= 1.2 && < 2, test-framework >= 0.6 && < 0.9, @@ -148,18 +147,18 @@ -- x-uses-tf: true build-depends: base >= 4.6 && < 5, - ansi-terminal >= 0.5 && < 0.7, + ansi-terminal >= 0.5 && < 0.9, containers, hashable, unordered-containers >= 0.2.3.0 && < 0.3, - distributed-process >= 0.6.6 && < 0.7, + distributed-process >= 0.6.0 && < 0.8, distributed-process-extras, - distributed-process-systest >= 0.1.0 && < 0.2.0, + distributed-process-systest >= 0.1.0 && < 0.3.0, distributed-static, bytestring, data-accessor, fingertree < 0.2, - network-transport >= 0.4 && < 0.5, + network-transport >= 0.4 && < 0.6, deepseq >= 1.3.0.1 && < 1.6, mtl, network-transport-tcp >= 0.4 && < 0.6, diff --git a/src/Control/Distributed/Process/Extras.hs b/src/Control/Distributed/Process/Extras.hs --- a/src/Control/Distributed/Process/Extras.hs +++ b/src/Control/Distributed/Process/Extras.hs @@ -62,6 +62,7 @@ , Routable(..) , Linkable(..) , Killable(..) + , Monitored(..) , NFSerializable , Recipient(..) , Shutdown(..) diff --git a/src/Control/Distributed/Process/Extras/Internal/Containers/MultiMap.hs b/src/Control/Distributed/Process/Extras/Internal/Containers/MultiMap.hs --- a/src/Control/Distributed/Process/Extras/Internal/Containers/MultiMap.hs +++ b/src/Control/Distributed/Process/Extras/Internal/Containers/MultiMap.hs @@ -14,7 +14,9 @@ , delete , filter , filterWithKey + , foldrWithKey , toList + , size ) where import qualified Data.Foldable as Foldable @@ -46,6 +48,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{..} = @@ -71,7 +76,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) @@ -81,7 +87,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 diff --git a/src/Control/Distributed/Process/Extras/Internal/Primitives.hs b/src/Control/Distributed/Process/Extras/Internal/Primitives.hs --- a/src/Control/Distributed/Process/Extras/Internal/Primitives.hs +++ b/src/Control/Distributed/Process/Extras/Internal/Primitives.hs @@ -27,6 +27,7 @@ , Resolvable(..) , Linkable(..) , Killable(..) + , Monitored(..) -- * Spawning and Linking , spawnSignalled @@ -57,7 +58,7 @@ import Control.Concurrent (myThreadId, throwTo) import Control.Distributed.Process hiding (monitor, finally, catch) -import qualified Control.Distributed.Process as P (monitor) +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,52 +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 --- | deliver = flip sendTo +-- | Send message to @Addressable@ object. deliver :: (Addressable a, Serializable m) => m -> a -> Process () deliver = flip sendTo --- | True if getProcessInfo /= Nothing --- NB: only works for local processes. +-- | 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 @@ -201,6 +194,8 @@ , 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 @@ -272,9 +267,10 @@ 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') diff --git a/src/Control/Distributed/Process/Extras/Internal/Queue/SeqQ.hs b/src/Control/Distributed/Process/Extras/Internal/Queue/SeqQ.hs --- a/src/Control/Distributed/Process/Extras/Internal/Queue/SeqQ.hs +++ b/src/Control/Distributed/Process/Extras/Internal/Queue/SeqQ.hs @@ -8,16 +8,18 @@ , dequeue , peek , filter + , size ) where -- A simple FIFO queue implementation backed by @Data.Sequence@. -import Prelude hiding (filter) +import Prelude hiding (filter, length) import Data.Sequence ( Seq , ViewR(..) , (<|) , viewr + , length ) import qualified Data.Sequence as Seq (empty, singleton, null, filter) @@ -49,6 +51,10 @@ {-# INLINE peek #-} 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) diff --git a/src/Control/Distributed/Process/Extras/Internal/Types.hs b/src/Control/Distributed/Process/Extras/Internal/Types.hs --- a/src/Control/Distributed/Process/Extras/Internal/Types.hs +++ b/src/Control/Distributed/Process/Extras/Internal/Types.hs @@ -4,6 +4,8 @@ {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE UndecidableInstances #-} +{-# LANGUAGE AllowAmbiguousTypes #-} +{-# LANGUAGE FunctionalDependencies #-} -- | Types used throughout the Extras package -- @@ -18,6 +20,7 @@ , Killable(..) , Resolvable(..) , Routable(..) + , Monitored(..) , Addressable , Recipient(..) , RegisterSelf(..) @@ -49,6 +52,7 @@ import Control.Monad.Catch (catch) import Data.Binary import Data.Foldable (traverse_) +import Data.Maybe (fromJust) import Data.Typeable (Typeable) import GHC.Generics @@ -144,6 +148,18 @@ -- | Create a /link/ with the supplied object. linkTo :: (Resolvable a) => a -> Process () linkTo r = resolve r >>= traverse_ link + +class Monitored a r m | a r -> m where + mkMonitor :: a -> Process r + checkMonitor :: a -> r -> m -> Process Bool + +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 -- | Class of things that can be killed (or instructed to exit). class Killable p where