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-registry.cabal b/distributed-process-registry.cabal new file mode 100644 --- /dev/null +++ b/distributed-process-registry.cabal @@ -0,0 +1,91 @@ +name: distributed-process-registry +version: 0.1.0 +cabal-version: >=1.8 +build-type: Simple +license: BSD3 +license-file: LICENCE +stability: experimental +Copyright: Tim Watson 2012 - 2015 +Author: Tim Watson +Maintainer: watson.timothy@gmail.com +Stability: experimental +Homepage: http://github.com/haskell-distributed/distributed-process-registry +Bug-Reports: http://github.com/haskell-distributed/distributed-process-registry/issues +synopsis: Cloud Haskell Extended Process Registry +description: This package provides an extended process registry. +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-registry + +flag perf + description: Build with profiling enabled + default: False + +library + build-depends: + base >= 4.4 && < 5, + data-accessor >= 0.2.2.3, + distributed-process >= 0.5.3 && < 0.6, + distributed-process-extras >= 0.2.0 && < 0.3, + distributed-process-supervisor >= 0.1.2 && < 0.2, + distributed-process-client-server >= 0.1.2 && < 0.2, + 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.Registry + +test-suite RegistryTests + 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, + distributed-process >= 0.5.3 && < 0.6, + distributed-process-extras >= 0.2.0 && < 0.3, + distributed-process-registry, + distributed-process-tests >= 0.4.3 && < 0.5, + distributed-static, + bytestring, + data-accessor, + fingertree < 0.2, + network-transport >= 0.4 && < 0.5, + deepseq >= 1.3.0.1 && < 1.4, + mtl, + network-transport-tcp >= 0.4 && < 0.5, + 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.5, + test-framework >= 0.6 && < 0.9, + test-framework-hunit, + transformers, + rematch >= 0.2.0.0 + if impl(ghc <= 7.5) + Build-Depends: ghc-prim + hs-source-dirs: + tests + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind -eventlog + extensions: CPP + main-is: TestRegistry.hs diff --git a/src/Control/Distributed/Process/Registry.hs b/src/Control/Distributed/Process/Registry.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Registry.hs @@ -0,0 +1,1076 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE PatternGuards #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE UndecidableInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.Registry +-- Copyright : (c) Tim Watson 2012 - 2013 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- The module provides an extended process registry, offering slightly altered +-- semantics to the built in @register@ and @unregister@ primitives and a richer +-- set of features: +-- +-- * Associate (unique) keys with a process /or/ (unique key per-process) values +-- * Use any 'Keyable' algebraic data type as keys +-- * Query for process with matching keys / values / properties +-- * Atomically /give away/ names +-- * Forceibly re-allocate names to/from a third party +-- +-- [Subscribing To Registry Events] +-- +-- It is possible to monitor a registry for changes and be informed whenever +-- changes take place. All subscriptions are /key based/, which means that +-- you can subscribe to name or property changes for any process, so that any +-- property changes matching the key you've subscribed to will trigger a +-- notification (i.e., regardless of the process to which the property belongs). +-- +-- The different types of event are defined by the 'KeyUpdateEvent' type. +-- +-- Processes subscribe to registry events using @monitorName@ or its counterpart +-- @monitorProperty@. If the operation succeeds, this will evaluate to an +-- opaque /reference/ that can be used when subsequently handling incoming +-- notifications, which will be delivered to the subscriber's mailbox as +-- @RegistryKeyMonitorNotification keyIdentity opaqueRef event@, where @event@ +-- has the type 'KeyUpdateEvent'. +-- +-- Subscribers can filter the types of event they receive by using the lower +-- level @monitor@ function (defined in /this/ module - not the one defined +-- in distributed-process' @Primitives@) and passing a list of filtering +-- 'KeyUpdateEventMask'. Without these filters in place, a monitor event will +-- be fired for /every/ pertinent change. +-- +----------------------------------------------------------------------------- +module Control.Distributed.Process.Registry + ( -- * Registry Keys + KeyType(..) + , Key(..) + , Keyable + -- * Defining / Starting A Registry + , Registry(..) + , start + , run + -- * Registration / Unregistration + , addName + , addProperty + , registerName + , registerValue + , giveAwayName + , RegisterKeyReply(..) + , unregisterName + , UnregisterKeyReply(..) + -- * Queries / Lookups + , lookupName + , lookupProperty + , registeredNames + , foldNames + , SearchHandle() + , member + , queryNames + , findByProperty + , findByPropertyValue + -- * Monitoring / Waiting + , monitor + , monitorName + , monitorProp + , unmonitor + , await + , awaitTimeout + , AwaitResult(..) + , KeyUpdateEventMask(..) + , KeyUpdateEvent(..) + , RegKeyMonitorRef + , RegistryKeyMonitorNotification(RegistryKeyMonitorNotification) + ) where + +{- DESIGN NOTES +This registry is a single process, parameterised by the types of key and +property value it can manage. It is, of course, possible to start multiple +registries and inter-connect them via registration, with one another. + +The /Service/ API is intended to be a declarative layer in which you define +the managed processes that make up your services, and each /Service Component/ +is registered and supervised appropriately for you, with the correct restart +strategies and start order calculated and so on. The registry is not only a +service locator, but provides the /wait for these dependencies to start first/ +bit of the puzzle. + +At some point, we'd like to offer a shared memory based registry, created on +behalf of a particular subsystem (i.e., some service or service group) and +passed implicitly using a reader monad or some such. This would allow multiple +processes to interact with the registry using STM (or perhaps a simple RWLock) +and could facilitate reduced contention. + +Even for the singleton-process based registry (i.e., this one) we /might/ also +be better off separating the monitoring (or at least the notifications) from +the registration/mapping parts into separate processes. +-} + +import Control.Distributed.Process hiding (call, monitor, unmonitor, mask) +import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe (send) +import qualified Control.Distributed.Process as P (monitor) +import Control.Distributed.Process.Serializable +import Control.Distributed.Process.Extras hiding (monitor, wrapMessage) +import qualified Control.Distributed.Process.Extras as PL + ( monitor + ) +import Control.Distributed.Process.ManagedProcess + ( call + , cast + , handleInfo + , reply + , continue + , input + , defaultProcess + , prioritised + , InitResult(..) + , ProcessAction + , ProcessReply + , ProcessDefinition(..) + , PrioritisedProcessDefinition(..) + , DispatchPriority + , CallRef + ) +import qualified Control.Distributed.Process.ManagedProcess as MP + ( pserve + ) +import Control.Distributed.Process.ManagedProcess.Server + ( handleCallIf + , handleCallFrom + , handleCallFromIf + , handleCast + ) +import Control.Distributed.Process.ManagedProcess.Server.Priority + ( prioritiseInfo_ + , setPriority + ) +import Control.Distributed.Process.ManagedProcess.Server.Restricted + ( RestrictedProcess + , Result + , getState + ) +import qualified Control.Distributed.Process.ManagedProcess.Server.Restricted as Restricted + ( handleCall + , reply + ) +import Control.Distributed.Process.Extras.Time +import Control.Monad (forM_, void) +import Data.Accessor + ( Accessor + , accessor + , (^:) + , (^=) + , (^.) + ) +import Data.Binary +import Data.Foldable (Foldable) +import qualified Data.Foldable as Foldable +import Data.Maybe (fromJust, isJust) +import Data.Hashable +import Data.HashMap.Strict (HashMap) +import qualified Data.HashMap.Strict as Map +import Control.Distributed.Process.Extras.Internal.Containers.MultiMap (MultiMap) +import qualified Control.Distributed.Process.Extras.Internal.Containers.MultiMap as MultiMap +import Data.HashSet (HashSet) +import qualified Data.HashSet as Set +import Data.Typeable (Typeable) + +import GHC.Generics + +-------------------------------------------------------------------------------- +-- Types -- +-------------------------------------------------------------------------------- + +-- | Describes how a key will be used - for storing names or properties. +data KeyType = + KeyTypeAlias -- ^ the key will refer to a name (i.e., named process) + | KeyTypeProperty -- ^ the key will refer to a (per-process) property + deriving (Typeable, Generic, Show, Eq) +instance Binary KeyType where +instance Hashable KeyType where + +-- | A registered key. Keys can be mapped to names or (process-local) properties +-- in the registry. The 'keyIdentity' holds the key's value (e.g., a string or +-- similar simple data type, which must provide a 'Keyable' instance), whilst +-- the 'keyType' and 'keyScope' describe the key's intended use and ownership. +data Key a = + Key + { keyIdentity :: !a + , keyType :: !KeyType + , keyScope :: !(Maybe ProcessId) + } + deriving (Typeable, Generic, Show, Eq) +instance (Serializable a) => Binary (Key a) where +instance (Hashable a) => Hashable (Key a) where + +-- | The 'Keyable' class describes types that can be used as registry keys. +-- The constraints ensure that the key can be stored and compared appropriately. +class (Show a, Eq a, Hashable a, Serializable a) => Keyable a +instance (Show a, Eq a, Hashable a, Serializable a) => Keyable a + +-- | A phantom type, used to parameterise registry startup +-- with the required key and value types. +data Registry k v = Registry { registryPid :: ProcessId } + deriving (Typeable, Generic, Show, Eq) +instance (Keyable k, Serializable v) => Binary (Registry k v) where + +instance Resolvable (Registry k v) where + resolve = return . Just . registryPid + +instance Linkable (Registry k v) where + linkTo = link . registryPid + +-- Internal/Private Request/Response Types + +data LookupKeyReq k = LookupKeyReq !(Key k) + deriving (Typeable, Generic) +instance (Serializable k) => Binary (LookupKeyReq k) where + +data LookupPropReq k = PropReq (Key k) + deriving (Typeable, Generic) +instance (Serializable k) => Binary (LookupPropReq k) where + +data LookupPropReply = + PropFound !Message + | PropNotFound + deriving (Typeable, Generic) +instance Binary LookupPropReply where + +data InvalidPropertyType = InvalidPropertyType + deriving (Typeable, Generic, Show, Eq) +instance Binary InvalidPropertyType where + +data RegNamesReq = RegNamesReq !ProcessId + deriving (Typeable, Generic) +instance Binary RegNamesReq where + +data UnregisterKeyReq k = UnregisterKeyReq !(Key k) + deriving (Typeable, Generic) +instance (Serializable k) => Binary (UnregisterKeyReq k) where + +-- | The result of an un-registration attempt. +data UnregisterKeyReply = + UnregisterOk -- ^ The given key was successfully unregistered + | UnregisterInvalidKey -- ^ The given key was invalid and could not be unregistered + | UnregisterKeyNotFound -- ^ The given key was not found (i.e., was not registered) + deriving (Typeable, Generic, Eq, Show) +instance Binary UnregisterKeyReply where + +-- Types used in (setting up and interacting with) key monitors + +-- | Used to describe a subset of monitoring events to listen for. +data KeyUpdateEventMask = + OnKeyRegistered -- ^ receive an event when a key is registered + | OnKeyUnregistered -- ^ receive an event when a key is unregistered + | OnKeyOwnershipChange -- ^ receive an event when a key's owner changes + | OnKeyLeaseExpiry -- ^ receive an event when a key's lease expires + deriving (Typeable, Generic, Eq, Show) +instance Binary KeyUpdateEventMask where +instance Hashable KeyUpdateEventMask where + +-- | An opaque reference used for matching monitoring events. See +-- 'RegistryKeyMonitorNotification' for more details. +newtype RegKeyMonitorRef = + RegKeyMonitorRef { unRef :: (ProcessId, Integer) } + deriving (Typeable, Generic, Eq, Show) +instance Binary RegKeyMonitorRef where +instance Hashable RegKeyMonitorRef where + +instance Resolvable RegKeyMonitorRef where + resolve = return . Just . fst . unRef + +-- | Provides information about a key monitoring event. +data KeyUpdateEvent = + KeyRegistered + { + owner :: !ProcessId + } + | KeyUnregistered + | KeyLeaseExpired + | KeyOwnerDied + { + diedReason :: !DiedReason + } + | KeyOwnerChanged + { + previousOwner :: !ProcessId + , newOwner :: !ProcessId + } + deriving (Typeable, Generic, Eq, Show) +instance Binary KeyUpdateEvent where + +-- | This message is delivered to processes which are monioring a +-- registry key. The opaque monitor reference will match (i.e., be equal +-- to) the reference returned from the @monitor@ function, which the +-- 'KeyUpdateEvent' describes the change that took place. +data RegistryKeyMonitorNotification k = + RegistryKeyMonitorNotification !k !RegKeyMonitorRef !KeyUpdateEvent !ProcessId + deriving (Typeable, Generic) +instance (Keyable k) => Binary (RegistryKeyMonitorNotification k) where +deriving instance (Keyable k) => Eq (RegistryKeyMonitorNotification k) +deriving instance (Keyable k) => Show (RegistryKeyMonitorNotification k) + +data RegisterKeyReq k = RegisterKeyReq !(Key k) + deriving (Typeable, Generic) +instance (Serializable k) => Binary (RegisterKeyReq k) where + +-- | The (return) value of an attempted registration. +data RegisterKeyReply = + RegisteredOk -- ^ The given key was registered successfully + | AlreadyRegistered -- ^ The key was already registered + deriving (Typeable, Generic, Eq, Show) +instance Binary RegisterKeyReply where + +-- | A cast message used to atomically give a name/key away to another process. +data GiveAwayName k = GiveAwayName !ProcessId !(Key k) + deriving (Typeable, Generic) +instance (Keyable k) => Binary (GiveAwayName k) where +deriving instance (Keyable k) => Eq (GiveAwayName k) +deriving instance (Keyable k) => Show (GiveAwayName k) + +data MonitorReq k = MonitorReq !(Key k) !(Maybe [KeyUpdateEventMask]) + deriving (Typeable, Generic) +instance (Keyable k) => Binary (MonitorReq k) where + +data UnmonitorReq = UnmonitorReq !RegKeyMonitorRef + deriving (Typeable, Generic) +instance Binary UnmonitorReq where + +-- | The result of an @await@ operation. +data AwaitResult k = + RegisteredName !ProcessId !k -- ^ The name was registered + | ServerUnreachable !DiedReason -- ^ The server was unreachable (or died) + | AwaitTimeout -- ^ The operation timed out + deriving (Typeable, Generic, Eq, Show) +instance (Keyable k) => Binary (AwaitResult k) where + +-- Server state + +-- On the server, a monitor reference consists of the actual +-- RegKeyMonitorRef which we can 'sendTo' /and/ the which +-- the client matches on, plus an optional list of event masks +data KMRef = KMRef { ref :: !RegKeyMonitorRef + , mask :: !(Maybe [KeyUpdateEventMask]) + -- use Nothing to monitor every event + } + deriving (Typeable, Generic, Show) +instance Hashable KMRef where +-- instance Binary KMRef where +instance Eq KMRef where + (KMRef a _) == (KMRef b _) = a == b + +data State k v = + State + { + _names :: !(HashMap k ProcessId) + , _properties :: !(HashMap (ProcessId, k) v) + , _monitors :: !(MultiMap k KMRef) + , _registeredPids :: !(HashSet ProcessId) + , _listeningPids :: !(HashSet ProcessId) + , _monitorIdCount :: !Integer + } + deriving (Typeable, Generic) + +-- Types used in \direct/ queries + +-- TODO: enforce QueryDirect's usage over only local channels + +data QueryDirect = QueryDirectNames | QueryDirectProperties | QueryDirectValues + deriving (Typeable, Generic) +instance Binary QueryDirect where + +-- NB: SHashMap is basically a shim, allowing us to copy a +-- pointer to our HashMap directly to the querying 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. +data SHashMap k v = SHashMap [(k, v)] (HashMap k v) + deriving (Typeable, Generic) + +instance (Keyable k, Serializable v) => + Binary (SHashMap k v) where + put = error "AttemptedToUseBinaryShim" + get = error "AttemptedToUseBinaryShim" +{- a real instance could look something like this: + + put (SHashMap _ hmap) = put (toList hmap) + get = do + hm <- get :: Get [(k, v)] + return $ SHashMap [] (fromList hm) +-} + +newtype SearchHandle k v = RS { getRS :: HashMap k v } + deriving (Typeable) + +instance (Keyable k) => Functor (SearchHandle k) where + fmap f (RS m) = RS $ Map.map f m +instance (Keyable k) => Foldable (SearchHandle k) where + foldr f acc = Foldable.foldr f acc . getRS +-- TODO: add Functor and Traversable instances + +-------------------------------------------------------------------------------- +-- Starting / Running A Registry -- +-------------------------------------------------------------------------------- + +start :: forall k v. (Keyable k, Serializable v) + => Process (Registry k v) +start = return . Registry =<< spawnLocal (run (undefined :: Registry k v)) + +run :: forall k v. (Keyable k, Serializable v) + => Registry k v + -> Process () +run _ = + MP.pserve () (const $ return $ InitOk initState Infinity) serverDefinition + where + initState = State { _names = Map.empty + , _properties = Map.empty + , _monitors = MultiMap.empty + , _registeredPids = Set.empty + , _listeningPids = Set.empty + , _monitorIdCount = (1 :: Integer) + } :: State k v + +-------------------------------------------------------------------------------- +-- Client Facing API -- +-------------------------------------------------------------------------------- + +-- -- | Sends a message to the process, or processes, corresponding to @key@. +-- -- If Key belongs to a unique object (name or aggregated counter), this +-- -- function will send a message to the corresponding process, or fail if there +-- -- is no such process. If Key is for a non-unique object type (counter or +-- -- property), Msg will be send to all processes that have such an object. +-- -- +-- dispatch svr ky msg = undefined +-- -- TODO: do a local-lookup and then sendTo the target + +-- | Associate the calling process with the given (unique) key. +addName :: forall k v. (Keyable k) + => Registry k v + -> k + -> Process RegisterKeyReply +addName s n = getSelfPid >>= registerName s n + +-- | Atomically transfer a (registered) name to another process. Has no effect +-- if the name does is not registered to the calling process! +-- +giveAwayName :: forall k v . (Keyable k) + => Registry k v + -> k + -> ProcessId + -> Process () +giveAwayName s n p = do + us <- getSelfPid + cast s $ GiveAwayName p $ Key n KeyTypeAlias (Just us) + +-- | Associate the given (non-unique) property with the current process. +-- If the property already exists, it will be overwritten with the new value. +addProperty :: (Keyable k, Serializable v) + => Registry k v -> k -> v -> Process RegisterKeyReply +addProperty s k v = do + call s $ (RegisterKeyReq (Key k KeyTypeProperty $ Nothing), v) + +-- | Register the item at the given address. +registerName :: forall k v . (Keyable k) + => Registry k v -> k -> ProcessId -> Process RegisterKeyReply +registerName s n p = do + call s $ RegisterKeyReq (Key n KeyTypeAlias $ Just p) + +-- | Register an item at the given address and associate it with a value. +-- If the property already exists, it will be overwritten with the new value. +registerValue :: (Resolvable b, Keyable k, Serializable v) + => Registry k v -> b -> k -> v -> Process RegisterKeyReply +registerValue s t n v = do + Just p <- resolve t + call s $ (RegisterKeyReq (Key n KeyTypeProperty $ Just p), v) + +-- | Un-register a (unique) name for the calling process. +unregisterName :: forall k v . (Keyable k) + => Registry k v + -> k + -> Process UnregisterKeyReply +unregisterName s n = do + self <- getSelfPid + call s $ UnregisterKeyReq (Key n KeyTypeAlias $ Just self) + +-- | Lookup the process identified by the supplied key. Evaluates to +-- @Nothing@ if the key is not registered. +lookupName :: forall k v . (Keyable k) + => Registry k v + -> k + -> Process (Maybe ProcessId) +lookupName s n = call s $ LookupKeyReq (Key n KeyTypeAlias Nothing) + +-- | Lookup the value of a named property for the calling process. Evaluates to +-- @Nothing@ if the property (key) is not registered. If the assignment to a +-- value of type @v@ does not correspond to the type of properties stored by +-- the registry, the calling process will exit with the reason set to +-- @InvalidPropertyType@. +lookupProperty :: (Keyable k, Serializable v) + => Registry k v + -> k + -> Process (Maybe v) +lookupProperty s n = do + us <- getSelfPid + res <- call s $ PropReq (Key n KeyTypeProperty (Just us)) + case res of + PropNotFound -> return Nothing + PropFound msg -> do + val <- unwrapMessage msg + if (isJust val) + then return val + else die InvalidPropertyType + +-- | Obtain a list of all registered keys. +registeredNames :: forall k v . (Keyable k) + => Registry k v + -> ProcessId + -> Process [k] +registeredNames s p = call s $ RegNamesReq p + +-- | Monitor changes to the supplied name. +-- +monitorName :: forall k v. (Keyable k) + => Registry k v -> k -> Process RegKeyMonitorRef +monitorName svr name = do + let key' = Key { keyIdentity = name + , keyScope = Nothing + , keyType = KeyTypeAlias + } + monitor svr key' Nothing + +-- | Monitor changes to the supplied (property) key. +-- +monitorProp :: forall k v. (Keyable k) + => Registry k v -> k -> ProcessId -> Process RegKeyMonitorRef +monitorProp svr key pid = do + let key' = Key { keyIdentity = key + , keyScope = Just pid + , keyType = KeyTypeProperty + } + monitor svr key' Nothing + +-- | Low level monitor operation. For the given key, set up a monitor +-- filtered by any 'KeyUpdateEventMask' entries that are supplied. +monitor :: forall k v. (Keyable k) + => Registry k v + -> Key k + -> Maybe [KeyUpdateEventMask] + -> Process RegKeyMonitorRef +monitor svr key' mask' = call svr $ MonitorReq key' mask' + +-- | Remove a previously set monitor. +-- +unmonitor :: forall k v. (Keyable k) + => Registry k v + -> RegKeyMonitorRef + -> Process () +unmonitor s = call s . UnmonitorReq + +-- | Await registration of a given key. This function will subsequently +-- block the evaluating process until the key is registered and a registration +-- event is dispatched to the caller's mailbox. +-- +await :: forall k v. (Keyable k) + => Registry k v + -> k + -> Process (AwaitResult k) +await a k = awaitTimeout a Infinity k + +-- | Await registration of a given key, but give up and return @AwaitTimeout@ +-- if registration does not take place within the specified time period (@delay@). +awaitTimeout :: forall k v. (Keyable k) + => Registry k v + -> Delay + -> k + -> Process (AwaitResult k) +awaitTimeout a d k = do + p <- forceResolve a + Just mRef <- PL.monitor p + kRef <- monitor a (Key k KeyTypeAlias Nothing) (Just [OnKeyRegistered]) + let matches' = matches mRef kRef k + let recv = case d of + Infinity -> receiveWait matches' >>= return . Just + Delay t -> receiveTimeout (asTimeout t) matches' + NoDelay -> receiveTimeout 0 matches' + recv >>= return . maybe AwaitTimeout id + where + forceResolve addr = do + mPid <- resolve addr + case mPid of + Nothing -> die "InvalidAddressable" + Just p -> return p + + matches mr kr k' = [ + matchIf (\(RegistryKeyMonitorNotification mk' kRef' ev' _) -> + (matchEv ev' && kRef' == kr && mk' == k')) + (\(RegistryKeyMonitorNotification _ _ (KeyRegistered pid) _) -> + return $ RegisteredName pid k') + , matchIf (\(ProcessMonitorNotification mRef' _ _) -> mRef' == mr) + (\(ProcessMonitorNotification _ _ dr) -> + return $ ServerUnreachable dr) + ] + + matchEv ev' = case ev' of + KeyRegistered _ -> True + _ -> False + +-- Local (non-serialised) shared data access. See note [sharing] below. + +findByProperty :: forall k v. (Keyable k) + => Registry k v + -> k + -> Process [ProcessId] +findByProperty r key = do + let pid = registryPid r + self <- getSelfPid + withMonitor pid $ do + cast r $ (self, QueryDirectProperties) + answer <- receiveWait [ + match (\(SHashMap _ m :: SHashMap ProcessId [k]) -> return $ Just m) + , matchIf (\(ProcessMonitorNotification _ p _) -> p == pid) + (\_ -> return Nothing) + , matchAny (\_ -> return Nothing) + ] + case answer of + Nothing -> die "DisconnectedFromServer" + Just m -> return $ Map.foldlWithKey' matchKey [] m + where + matchKey ps p ks + | key `elem` ks = p:ps + | otherwise = ps + +findByPropertyValue :: (Keyable k, Serializable v, Eq v) + => Registry k v + -> k + -> v + -> Process [ProcessId] +findByPropertyValue r key val = do + let pid = registryPid r + self <- getSelfPid + withMonitor pid $ do + cast r $ (self, QueryDirectValues) + answer <- receiveWait [ + match (\(SHashMap _ m :: SHashMap ProcessId [(k, v)]) -> return $ Just m) + , matchIf (\(ProcessMonitorNotification _ p _) -> p == pid) + (\_ -> return Nothing) + , matchAny (\_ -> return Nothing) -- TODO: logging? + ] + case answer of + Nothing -> die "DisconnectedFromServer" + Just m -> return $ Map.foldlWithKey' matchKey [] m + where + matchKey ps p ks + | (key, val) `elem` ks = p:ps + | otherwise = ps + +-- TODO: move to UnsafePrimitives over a passed {Send|Receive}Port here, and +-- avoid interfering with the caller's mailbox. + +-- | Monadic left fold over all registered names/keys. The fold takes place +-- in the evaluating process. +foldNames :: forall b k v . Keyable k + => Registry k v + -> b + -> (b -> (k, ProcessId) -> Process b) + -> Process b +foldNames pid acc fn = do + self <- getSelfPid + -- TODO: monitor @pid@ and die if necessary!!! + cast pid $ (self, QueryDirectNames) + -- Although we incur the cost of scanning our mailbox here (which we could + -- avoid by spawning an intermediary perhaps), the message is delivered to + -- us without any copying or serialisation overheads. + SHashMap _ m <- expect :: Process (SHashMap k ProcessId) + Foldable.foldlM fn acc (Map.toList m) + +-- | Evaluate a query on a 'SearchHandle', in the calling process. +queryNames :: forall b k v . Keyable k + => Registry k v + -> (SearchHandle k ProcessId -> Process b) + -> Process b +queryNames pid fn = do + self <- getSelfPid + cast pid $ (self, QueryDirectNames) + SHashMap _ m <- expect :: Process (SHashMap k ProcessId) + fn (RS m) + +-- | Tests whether or not the supplied key is registered, evaluated in the +-- calling process. +member :: (Keyable k, Serializable v) + => k + -> SearchHandle k v + -> Bool +member k = Map.member k . getRS + +-- note [sharing]: +-- We use the base library's UnsafePrimitives for these fold/query operations, +-- to pass a pointer to our internal HashMaps for read-only operations. There +-- is a potential cost to the caller, if their mailbox is full - we should move +-- to use unsafe channel's for this at some point. +-- + +-------------------------------------------------------------------------------- +-- Server Process -- +-------------------------------------------------------------------------------- + +serverDefinition :: forall k v. (Keyable k, Serializable v) + => PrioritisedProcessDefinition (State k v) +serverDefinition = prioritised processDefinition regPriorities + where + regPriorities :: [DispatchPriority (State k v)] + regPriorities = [ + prioritiseInfo_ (\(ProcessMonitorNotification _ _ _) -> setPriority 100) + ] + +processDefinition :: forall k v. (Keyable k, Serializable v) + => ProcessDefinition (State k v) +processDefinition = + defaultProcess + { + apiHandlers = + [ + handleCallIf + (input ((\(RegisterKeyReq (Key{..} :: Key k)) -> + keyType == KeyTypeAlias && (isJust keyScope)))) + handleRegisterName + , handleCallIf + (input ((\((RegisterKeyReq (Key{..} :: Key k)), _ :: v) -> + keyType == KeyTypeProperty && (isJust keyScope)))) + handleRegisterProperty + , handleCallFromIf + (input ((\((RegisterKeyReq (Key{..} :: Key k)), _ :: v) -> + keyType == KeyTypeProperty && (not $ isJust keyScope)))) + handleRegisterPropertyCR + , handleCast handleGiveAwayName + , handleCallIf + (input ((\(LookupKeyReq (Key{..} :: Key k)) -> + keyType == KeyTypeAlias))) + (\state (LookupKeyReq key') -> reply (findName key' state) state) + , handleCallIf + (input ((\(PropReq (Key{..} :: Key k)) -> + keyType == KeyTypeProperty && (isJust keyScope)))) + handleLookupProperty + , handleCallIf + (input ((\(UnregisterKeyReq (Key{..} :: Key k)) -> + keyType == KeyTypeAlias && (isJust keyScope)))) + handleUnregisterName + , handleCallFrom handleMonitorReq + , handleCallFrom handleUnmonitorReq + , Restricted.handleCall handleRegNamesLookup + , handleCast handleQuery + ] + , infoHandlers = [handleInfo handleMonitorSignal] + } :: ProcessDefinition (State k v) + +handleQuery :: forall k v. (Keyable k, Serializable v) + => State k v + -> (ProcessId, QueryDirect) + -> Process (ProcessAction (State k v)) +handleQuery st@State{..} (pid, qd) = do + case qd of + QueryDirectNames -> Unsafe.send pid shmNames + QueryDirectProperties -> Unsafe.send pid shmProps + QueryDirectValues -> Unsafe.send pid shmVals + continue st + where + shmNames = SHashMap [] $ st ^. names + shmProps = SHashMap [] xfmProps + shmVals = SHashMap [] xfmVals + + -- since we currently have to fold over our properties in order + -- answer remote queries, the sharing we do here seems a bit pointless, + -- however we'll be moving to a shared memory based registry soon + + xfmProps = Map.foldlWithKey' convProps Map.empty (st ^. properties) + xfmVals = Map.foldlWithKey' convVals Map.empty (st ^. properties) + + convProps m (p, k) _ = + case Map.lookup p m of + Nothing -> Map.insert p [k] m + Just ks -> Map.insert p (k:ks) m + + convVals m (p, k) v = + case Map.lookup p m of + Nothing -> Map.insert p [(k, v)] m + Just ks -> Map.insert p ((k, v):ks) m + +handleRegisterName :: forall k v. (Keyable k, Serializable v) + => State k v + -> RegisterKeyReq k + -> Process (ProcessReply RegisterKeyReply (State k v)) +handleRegisterName state (RegisterKeyReq Key{..}) = do + let found = Map.lookup keyIdentity (state ^. names) + case found of + Nothing -> do + let pid = fromJust keyScope + let refs = state ^. registeredPids + refs' <- ensureMonitored pid refs + notifySubscribers keyIdentity state (KeyRegistered pid) + reply RegisteredOk $ ( (names ^: Map.insert keyIdentity pid) + . (registeredPids ^= refs') + $ state) + Just pid -> + if (pid == (fromJust keyScope)) + then reply RegisteredOk state + else reply AlreadyRegistered state + +handleRegisterPropertyCR :: forall k v. (Keyable k, Serializable v) + => State k v + -> CallRef (RegisterKeyReply) + -> (RegisterKeyReq k, v) + -> Process (ProcessReply RegisterKeyReply (State k v)) +handleRegisterPropertyCR st cr req = do + pid <- resolve cr + doRegisterProperty (fromJust pid) st req + +handleRegisterProperty :: forall k v. (Keyable k, Serializable v) + => State k v + -> (RegisterKeyReq k, v) + -> Process (ProcessReply RegisterKeyReply (State k v)) +handleRegisterProperty state req@((RegisterKeyReq Key{..}), _) = do + doRegisterProperty (fromJust keyScope) state req + +doRegisterProperty :: forall k v. (Keyable k, Serializable v) + => ProcessId + -> State k v + -> (RegisterKeyReq k, v) + -> Process (ProcessReply RegisterKeyReply (State k v)) +doRegisterProperty scope state ((RegisterKeyReq Key{..}), v) = do + void $ P.monitor scope + notifySubscribers keyIdentity state (KeyRegistered scope) + reply RegisteredOk $ ( (properties ^: Map.insert (scope, keyIdentity) v) + $ state ) + +handleLookupProperty :: forall k v. (Keyable k, Serializable v) + => State k v + -> LookupPropReq k + -> Process (ProcessReply LookupPropReply (State k v)) +handleLookupProperty state (PropReq Key{..}) = do + let entry = Map.lookup (fromJust keyScope, keyIdentity) (state ^. properties) + case entry of + Nothing -> reply PropNotFound state + Just p -> reply (PropFound (wrapMessage p)) state + +handleUnregisterName :: forall k v. (Keyable k, Serializable v) + => State k v + -> UnregisterKeyReq k + -> Process (ProcessReply UnregisterKeyReply (State k v)) +handleUnregisterName state (UnregisterKeyReq Key{..}) = do + let entry = Map.lookup keyIdentity (state ^. names) + case entry of + Nothing -> reply UnregisterKeyNotFound state + Just pid -> + case (pid /= (fromJust keyScope)) of + True -> reply UnregisterInvalidKey state + False -> do + notifySubscribers keyIdentity state KeyUnregistered + let state' = ( (names ^: Map.delete keyIdentity) + . (monitors ^: MultiMap.filterWithKey (\k' _ -> k' /= keyIdentity)) + $ state) + reply UnregisterOk $ state' + +handleGiveAwayName :: forall k v. (Keyable k, Serializable v) + => State k v + -> GiveAwayName k + -> Process (ProcessAction (State k v)) +handleGiveAwayName state (GiveAwayName newPid Key{..}) = do + maybe (continue state) giveAway $ Map.lookup keyIdentity (state ^. names) + where + giveAway pid = do + let scope = fromJust keyScope + case (pid == scope) of + False -> continue state + True -> do + let state' = ((names ^: Map.insert keyIdentity newPid) $ state) + notifySubscribers keyIdentity state (KeyOwnerChanged pid newPid) + continue state' + +handleMonitorReq :: forall k v. (Keyable k, Serializable v) + => State k v + -> CallRef RegKeyMonitorRef + -> MonitorReq k + -> Process (ProcessReply RegKeyMonitorRef (State k v)) +handleMonitorReq state cRef (MonitorReq Key{..} mask') = do + let mRefId = (state ^. monitorIdCount) + 1 + Just caller <- resolve cRef + let mRef = RegKeyMonitorRef (caller, mRefId) + let kmRef = KMRef mRef mask' + let refs = state ^. listeningPids + refs' <- ensureMonitored caller refs + fireEventForPreRegisteredKey state keyIdentity keyScope kmRef + reply mRef $ ( (monitors ^: MultiMap.insert keyIdentity kmRef) + . (listeningPids ^= refs') + . (monitorIdCount ^= mRefId) + $ state + ) + where + fireEventForPreRegisteredKey st kId kScope KMRef{..} = do + let evMask = maybe [] id mask + case (keyType, elem OnKeyRegistered evMask) of + (KeyTypeAlias, True) -> do + let found = Map.lookup kId (st ^. names) + fireEvent found kId ref + (KeyTypeProperty, _) -> do + self <- getSelfPid + let scope = maybe self id kScope + let found = Map.lookup (scope, kId) (st ^. properties) + case found of + Nothing -> return () -- TODO: logging or some such!? + Just _ -> fireEvent (Just scope) kId ref + _ -> return () + + fireEvent fnd kId' ref' = do + case fnd of + Nothing -> return () + Just p -> do + us <- getSelfPid + sendTo ref' $ (RegistryKeyMonitorNotification kId' + ref' + (KeyRegistered p) + us) + +handleUnmonitorReq :: forall k v. (Keyable k, Serializable v) + => State k v + -> CallRef () + -> UnmonitorReq + -> Process (ProcessReply () (State k v)) +handleUnmonitorReq state _cRef (UnmonitorReq ref') = do + let pid = fst $ unRef ref' + reply () $ ( (monitors ^: MultiMap.filter ((/= ref') . ref)) + . (listeningPids ^: Set.delete pid) + $ state + ) + +handleRegNamesLookup :: forall k v. (Keyable k, Serializable v) + => RegNamesReq + -> RestrictedProcess (State k v) (Result [k]) +handleRegNamesLookup (RegNamesReq p) = do + state <- getState + Restricted.reply $ Map.foldlWithKey' (acc p) [] (state ^. names) + where + acc pid ns n pid' + | pid == pid' = (n:ns) + | otherwise = ns + +handleMonitorSignal :: forall k v. (Keyable k, Serializable v) + => State k v + -> ProcessMonitorNotification + -> Process (ProcessAction (State k v)) +handleMonitorSignal state@State{..} (ProcessMonitorNotification _ pid diedReason) = + do let state' = removeActiveSubscriptions pid state + (deadNames, deadProps) <- notifyListeners state' pid diedReason + continue $ ( (names ^= Map.difference _names deadNames) + . (properties ^= Map.difference _properties deadProps) + $ state) + where + removeActiveSubscriptions p s = + let subscriptions = (state ^. listeningPids) in + case (Set.member p subscriptions) of + False -> s + True -> ( (listeningPids ^: Set.delete p) + -- delete any monitors this (now dead) process held + . (monitors ^: MultiMap.filter ((/= p) . fst . unRef . ref)) + $ s) + + notifyListeners :: State k v + -> ProcessId + -> DiedReason + -> Process (HashMap k ProcessId, HashMap (ProcessId, k) v) + notifyListeners st pid' dr = do + let diedNames = Map.filter (== pid') (st ^. names) + let diedProps = Map.filterWithKey (\(p, _) _ -> p == pid') + (st ^. properties) + let nameSubs = MultiMap.filterWithKey (\k _ -> Map.member k diedNames) + (st ^. monitors) + let propSubs = MultiMap.filterWithKey (\k _ -> Map.member (pid', k) diedProps) + (st ^. monitors) + forM_ (MultiMap.toList nameSubs) $ \(kIdent, KMRef{..}) -> do + let kEvDied = KeyOwnerDied { diedReason = dr } + let mRef = RegistryKeyMonitorNotification kIdent ref + us <- getSelfPid + case mask of + Nothing -> sendTo ref (mRef kEvDied us) + Just mask' -> do + case (elem OnKeyOwnershipChange mask') of + True -> sendTo ref (mRef kEvDied us) + False -> do + if (elem OnKeyUnregistered mask') + then sendTo ref (mRef KeyUnregistered us) + else return () + forM_ (MultiMap.toList propSubs) (notifyPropSubscribers dr) + return (diedNames, diedProps) + + notifyPropSubscribers dr' (kIdent, KMRef{..}) = do + let died = maybe False (elem OnKeyOwnershipChange) mask + let event = case died of + True -> KeyOwnerDied { diedReason = dr' } + False -> KeyUnregistered + getSelfPid >>= sendTo ref . RegistryKeyMonitorNotification kIdent ref event + +ensureMonitored :: ProcessId -> HashSet ProcessId -> Process (HashSet ProcessId) +ensureMonitored pid refs = do + case (Set.member pid refs) of + True -> return refs + False -> P.monitor pid >> return (Set.insert pid refs) + +notifySubscribers :: forall k v. (Keyable k, Serializable v) + => k + -> State k v + -> KeyUpdateEvent + -> Process () +notifySubscribers k st ev = do + let subscribers = MultiMap.filterWithKey (\k' _ -> k' == k) (st ^. monitors) + forM_ (MultiMap.toList subscribers) $ \(_, KMRef{..}) -> do + if (maybe True (elem (maskFor ev)) mask) + then getSelfPid >>= sendTo ref . RegistryKeyMonitorNotification k ref ev + else {- (liftIO $ putStrLn "no mask") >> -} return () + +-------------------------------------------------------------------------------- +-- Utilities / Accessors -- +-------------------------------------------------------------------------------- + +maskFor :: KeyUpdateEvent -> KeyUpdateEventMask +maskFor (KeyRegistered _) = OnKeyRegistered +maskFor KeyUnregistered = OnKeyUnregistered +maskFor (KeyOwnerDied _) = OnKeyOwnershipChange +maskFor (KeyOwnerChanged _ _) = OnKeyOwnershipChange +maskFor KeyLeaseExpired = OnKeyLeaseExpiry + +findName :: forall k v. (Keyable k, Serializable v) + => Key k + -> State k v + -> Maybe ProcessId +findName Key{..} state = Map.lookup keyIdentity (state ^. names) + +names :: forall k v. Accessor (State k v) (HashMap k ProcessId) +names = accessor _names (\n' st -> st { _names = n' }) + +properties :: forall k v. Accessor (State k v) (HashMap (ProcessId, k) v) +properties = accessor _properties (\ps st -> st { _properties = ps }) + +monitors :: forall k v. Accessor (State k v) (MultiMap k KMRef) +monitors = accessor _monitors (\ms st -> st { _monitors = ms }) + +registeredPids :: forall k v. Accessor (State k v) (HashSet ProcessId) +registeredPids = accessor _registeredPids (\mp st -> st { _registeredPids = mp }) + +listeningPids :: forall k v. Accessor (State k v) (HashSet ProcessId) +listeningPids = accessor _listeningPids (\lp st -> st { _listeningPids = lp }) + +monitorIdCount :: forall k v. Accessor (State k v) Integer +monitorIdCount = accessor _monitorIdCount (\i st -> st { _monitorIdCount = i }) + diff --git a/tests/TestRegistry.hs b/tests/TestRegistry.hs new file mode 100644 --- /dev/null +++ b/tests/TestRegistry.hs @@ -0,0 +1,542 @@ +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE TemplateHaskell #-} + +module Main where + +import Control.Concurrent.MVar (newEmptyMVar, takeMVar) +import Control.Concurrent.Utils (Lock, Exclusive(..), Synchronised(..)) +import Control.Distributed.Process +import Control.Distributed.Process.Node +import Control.Distributed.Process.Extras + ( awaitExit + , spawnSignalled + , Killable(..) + ) +import Control.Distributed.Process.Registry + ( Registry(..) + , KeyUpdateEvent(..) + , RegistryKeyMonitorNotification(..) + , addName + , addProperty + , giveAwayName + , registerName + , registerValue + , unregisterName + , lookupName + , lookupProperty + , registeredNames + , foldNames + , queryNames + , findByProperty + , findByPropertyValue + , awaitTimeout + , await + , SearchHandle + , AwaitResult(..) + , RegisterKeyReply(..) + , UnregisterKeyReply(..) + ) +import qualified Control.Distributed.Process.Registry as Registry +import Control.Distributed.Process.Tests.Internal.Utils +import Control.Distributed.Process.Extras.Time +import Control.Distributed.Process.Extras.Timer (sleep) +import Control.Monad (void, forM_, forM) +import Control.Rematch + ( equalTo + ) + +import qualified Data.Foldable as Foldable +import qualified Data.List as List + +#if ! MIN_VERSION_base(4,6,0) +import Prelude hiding (catch) +#endif + +import Test.HUnit (Assertion, assertFailure) +import Test.Framework (Test, testGroup, defaultMain) +import Test.Framework.Providers.HUnit (testCase) + +import Network.Transport.TCP +import qualified Network.Transport as NT + +myRegistry :: Process (Registry String ()) +myRegistry = Registry.start + +counterReg :: Process (Registry String Int) +counterReg = Registry.start + +withRegistry :: LocalNode + -> (Registry String () -> Process ()) + -> Assertion +withRegistry node proc = do + runProcess node $ do + reg' <- myRegistry + (proc reg') `finally` (killProc reg' "goodbye") + +testAddLocalName :: TestResult RegisterKeyReply -> Process () +testAddLocalName result = do + reg <- myRegistry + stash result =<< addName reg "foobar" + +testAddLocalProperty :: TestResult (Maybe Int) -> Process () +testAddLocalProperty result = do + reg <- counterReg + addProperty reg "chickens" (42 :: Int) + stash result =<< lookupProperty reg "chickens" + +testAddRemoteProperty :: TestResult Int -> Process () +testAddRemoteProperty result = do + reg <- counterReg + p <- spawnLocal $ do + pid <- expect + Just i <- lookupProperty reg "ducks" :: Process (Maybe Int) + send pid i + RegisteredOk <- registerValue reg p "ducks" (39 :: Int) + getSelfPid >>= send p + expect >>= stash result + +testFindByPropertySet :: TestResult Bool -> Process () +testFindByPropertySet result = do + reg <- counterReg + p1 <- spawnLocal $ addProperty reg "animals" (1 :: Int) >> expect >>= return + p2 <- spawnLocal $ addProperty reg "animals" (1 :: Int) >> expect >>= return + sleep $ seconds 1 + found <- findByProperty reg "animals" + found `shouldContain` p1 + found `shouldContain` p2 + notFound <- findByProperty reg "foobar" + stash result $ length notFound == 0 + +testFindByPropertyValueSet :: TestResult Bool -> Process () +testFindByPropertyValueSet result = do + us <- getSelfPid + reg <- counterReg + p1 <- spawnLocal $ link us >> addProperty reg "animals" (1 :: Int) >> expect >>= return + _ <- spawnLocal $ link us >> addProperty reg "animals" (2 :: Int) >> expect >>= return + p3 <- spawnLocal $ link us >> addProperty reg "animals" (1 :: Int) >> expect >>= return + _ <- spawnLocal $ link us >> addProperty reg "ducks" (1 :: Int) >> expect >>= return + + sleep $ seconds 1 + found <- findByPropertyValue reg "animals" (1 :: Int) + found `shouldContain` p1 + found `shouldContain` p3 + stash result $ length found == 2 + +testCheckLocalName :: Registry String () -> Process () +testCheckLocalName reg = do + void $ addName reg "fwibble" + fwibble <- lookupName reg "fwibble" + selfPid <- getSelfPid + fwibble `shouldBe` equalTo (Just selfPid) + +testGiveAwayName :: Registry String () -> Process () +testGiveAwayName reg = do + testPid <- getSelfPid + void $ addName reg "cat" + pid <- spawnLocal $ link testPid >> (expect :: Process ()) + giveAwayName reg "cat" pid + cat <- lookupName reg "cat" + cat `shouldBe` equalTo (Just pid) + +testMultipleRegistrations :: Registry String () -> Process () +testMultipleRegistrations reg = do + self <- getSelfPid + forM_ names (addName reg) + forM_ names $ \name -> do + found <- lookupName reg name + found `shouldBe` equalTo (Just self) + where + names = ["foo", "bar", "baz"] + +testDuplicateRegistrations :: Registry String () -> Process () +testDuplicateRegistrations reg = do + void $ addName reg "foobar" + RegisteredOk <- addName reg "foobar" + pid <- spawnLocal $ (expect :: Process ()) >>= return + result <- registerName reg "foobar" pid + result `shouldBe` equalTo AlreadyRegistered + +testUnregisterName :: Registry String () -> Process () +testUnregisterName reg = do + self <- getSelfPid + void $ addName reg "fwibble" + void $ addName reg "fwobble" + Just self' <- lookupName reg "fwibble" + self' `shouldBe` equalTo self + unreg <- unregisterName reg "fwibble" + unregFwibble <- lookupName reg "fwibble" + unreg `shouldBe` equalTo UnregisterOk + -- fwibble is gone... + unregFwibble `shouldBe` equalTo Nothing + -- but fwobble is still registered + fwobble <- lookupName reg "fwobble" + fwobble `shouldBe` equalTo (Just self) + +testUnregisterUnknownName :: Registry String () -> Process () +testUnregisterUnknownName reg = do + result <- unregisterName reg "no.such.name" + result `shouldBe` equalTo UnregisterKeyNotFound + +testUnregisterAnothersName :: Registry String () -> Process () +testUnregisterAnothersName reg = do + (sp, rp) <- newChan + pid <- spawnLocal $ do + void $ addName reg "proc.name" + sendChan sp () + (expect :: Process ()) >>= return + () <- receiveChan rp + found <- lookupName reg "proc.name" + found `shouldBe` equalTo (Just pid) + unreg <- unregisterName reg "proc.name" + unreg `shouldBe` equalTo UnregisterInvalidKey + +testProcessDeathHandling :: Registry String () -> Process () +testProcessDeathHandling reg = do + (sp, rp) <- newChan + pid <- spawnLocal $ do + void $ addName reg "proc.name.1" + void $ addName reg "proc.name.2" + sendChan sp () + (expect :: Process ()) >>= return + () <- receiveChan rp + void $ monitor pid + regNames <- registeredNames reg pid + regNames `shouldContain` "proc.name.1" + regNames `shouldContain` "proc.name.2" + send pid () + receiveWait [ + match (\(ProcessMonitorNotification _ _ _) -> return ()) + ] + forM_ [1..2 :: Int] $ \n -> do + let name = "proc.name." ++ (show n) + found <- lookupName reg name + found `shouldBe` equalTo Nothing + regNames' <- registeredNames reg pid + regNames' `shouldBe` equalTo ([] :: [String]) + +testLocalRegNamesFold :: Registry String () -> Process () +testLocalRegNamesFold reg = do + parent <- getSelfPid + forM_ [1..1000] $ \(i :: Int) -> spawnLocal $ do + send parent i + addName reg (show i) >> expect :: Process () + waitForTestRegistrations + ns <- foldNames reg [] $ \acc (n, _) -> return ((read n :: Int):acc) + (List.sort ns) `shouldBe` equalTo [1..1000] + where + waitForTestRegistrations = do + void $ receiveWait [ matchIf (\(i :: Int) -> i == 1000) (\_ -> return ()) ] + sleep $ milliSeconds 150 + +testLocalQueryNamesFold :: Registry String () -> Process () +testLocalQueryNamesFold reg = do + pids <- forM [1..1000] $ \(i :: Int) -> spawnLocal $ do + addName reg (show i) >> expect :: Process () + waitRegs reg 1000 + ns <- queryNames reg $ \(sh :: SearchHandle String ProcessId) -> do + return $ Foldable.foldl (\acc pid -> (pid:acc)) [] sh + (List.sort ns) `shouldBe` equalTo pids + where + waitRegs :: Registry String () -> Int -> Process () + waitRegs _ 0 = return () + waitRegs reg' n = await reg' (show n) >> waitRegs reg' (n - 1) + +testMonitorName :: Registry String () -> Process () +testMonitorName reg = do + (sp, rp) <- newChan + pid <- spawnLocal $ do + void $ addName reg "proc.name.1" + void $ addName reg "proc.name.2" + sendChan sp () + (expect :: Process ()) >>= return + () <- receiveChan rp + mRef <- Registry.monitorName reg "proc.name.2" + send pid () + res <- receiveTimeout (after 2 Seconds) [ + matchIf (\(RegistryKeyMonitorNotification k ref _ _) -> + k == "proc.name.2" && ref == mRef) + (\(RegistryKeyMonitorNotification _ _ ev _) -> return ev) + ] + res `shouldBe` equalTo (Just (KeyOwnerDied DiedNormal)) + +testMonitorNameChange :: Registry String () -> Process () +testMonitorNameChange reg = do + let k = "proc.name.foo" + + lock <- liftIO $ new :: Process Lock + acquire lock + testPid <- getSelfPid + + pid <- spawnSignalled (addName reg k) $ const $ do + link testPid + synchronised lock $ giveAwayName reg k testPid + expect >>= return + + -- at this point, we know the child has grabbed the name k for itself + mRef <- Registry.monitorName reg k + release lock + + ev <- receiveWait [ matchIf (\(RegistryKeyMonitorNotification k' ref _ _) -> + k' == k && ref == mRef) + (\(RegistryKeyMonitorNotification _ _ ev' _) -> + return ev') ] + ev `shouldBe` equalTo (KeyOwnerChanged pid testPid) + +testUnmonitor :: TestResult Bool -> Process () +testUnmonitor result = do + let k = "chickens" + let name = "poultry" + + reg <- counterReg + (sp, rp) <- newChan + + pid <- spawnLocal $ do + void $ addProperty reg k (42 :: Int) + addName reg name + sendChan sp () + expect >>= return + + () <- receiveChan rp + mRef <- Registry.monitorProp reg k pid + + void $ receiveWait [ + matchIf (\(RegistryKeyMonitorNotification k' ref ev _) -> + k' == k && ref == mRef && ev == (KeyRegistered pid)) + return + ] + + Registry.unmonitor reg mRef + kill pid "goodbye!" + awaitExit pid + + Nothing <- lookupName reg name + t <- receiveTimeout (after 1 Seconds) [ + matchIf (\(RegistryKeyMonitorNotification k' ref ev _) -> + k' == k && ref == mRef && ev == (KeyRegistered pid)) + return + ] + t `shouldBe` (equalTo Nothing) + stash result True + +testMonitorPropertyChanged :: TestResult Bool -> Process () +testMonitorPropertyChanged result = do + let k = "chickens" + let name = "poultry" + + lock <- liftIO $ new :: Process Lock + reg <- counterReg + + -- yes, using the lock here without exception handling is risky... + acquire lock + + pid <- spawnSignalled (addProperty reg k (42 :: Int)) $ const $ do + addName reg name + synchronised lock $ addProperty reg k (45 :: Int) + expect >>= return + + -- at this point, the worker has already registered 42 (since we used + -- spawnSignalled to start it) and is waiting for us to unlock... + mRef <- Registry.monitorProp reg k pid + + -- we /must/ receive a monitor notification first, for the pre-existing + -- key and only then will we let the worker move on and update the value + void $ receiveWait [ + matchIf (\(RegistryKeyMonitorNotification k' ref ev _) -> + k' == k && ref == mRef && ev == (KeyRegistered pid)) + return + ] + + release lock + + kr <- receiveTimeout 1000000 [ + matchIf (\(RegistryKeyMonitorNotification k' ref ev _) -> + k' == k && ref == mRef && ev == (KeyRegistered pid)) + return + ] + + stash result (kr /= Nothing) + +testMonitorPropertyOwnerDied :: Registry String () -> Process () +testMonitorPropertyOwnerDied reg = do + let k = "foo.bar" + pid <- spawnSignalled (addProperty reg k ()) $ const $ do + expect >>= return + + mRef <- Registry.monitorProp reg k pid + kill pid "goodbye!" + + -- we /must/ receive a monitor notification first, for the pre-existing + -- key and only then will we let the worker move on and update the value + r <- receiveTimeout 1000000 [ + matchIf (\(RegistryKeyMonitorNotification k' ref ev _) -> + k' == k && ref == mRef && ev /= (KeyRegistered pid)) + (\(RegistryKeyMonitorNotification _ _ ev _) -> + return ev) + ] + case r of + Just (KeyOwnerDied (DiedException _)) -> return () + _ -> (liftIO $ putStrLn (show r)) >> return () + +-- testMonitorLeaseExpired :: ProcessId -> Process () + +-- testMonitorPropertyLeaseExpired :: ProcessId -> Process () + +testMonitorUnregistration :: Registry String () -> Process () +testMonitorUnregistration reg = do + (sp, rp) <- newChan + pid <- spawnLocal $ do + void $ addName reg "proc.1" + sendChan sp () + expect :: Process () + void $ unregisterName reg "proc.1" + sendChan sp () + (expect :: Process ()) >>= return + () <- receiveChan rp + mRef <- Registry.monitorName reg "proc.1" + send pid () + () <- receiveChan rp + res <- receiveTimeout (after 2 Seconds) [ + matchIf (\(RegistryKeyMonitorNotification k ref _ _) -> + k == "proc.1" && ref == mRef) + (\(RegistryKeyMonitorNotification _ _ ev _) -> return ev) + ] + res `shouldBe` equalTo (Just KeyUnregistered) + +testMonitorRegistration :: Registry String () -> Process () +testMonitorRegistration reg = do + kRef <- Registry.monitorName reg "my.proc" + pid <- spawnSignalled (addName reg "my.proc") $ const expect + res <- receiveTimeout (after 5 Seconds) [ + matchIf (\(RegistryKeyMonitorNotification k ref _ _) -> + k == "my.proc" && ref == kRef) + (\(RegistryKeyMonitorNotification _ _ ev _) -> return ev) + ] + res `shouldBe` equalTo (Just (KeyRegistered pid)) + +testAwaitRegistration :: Registry String () -> Process () +testAwaitRegistration reg = do + pid <- spawnLocal $ do + void $ addName reg "foo.bar" + expect :: Process () + res <- awaitTimeout reg (Delay $ within 5 Seconds) "foo.bar" + res `shouldBe` equalTo (RegisteredName pid "foo.bar") + +testAwaitRegistrationNoTimeout :: Registry String () -> Process () +testAwaitRegistrationNoTimeout reg = do + parent <- getSelfPid + barrier <- liftIO $ newEmptyMVar + void $ spawnLocal $ do + res <- await reg "baz.bog" + case res of + RegisteredName _ "baz.bog" -> stash barrier () + _ -> kill parent "BANG!" + void $ addName reg "baz.bog" + liftIO $ takeMVar barrier >>= return + +testAwaitServerDied :: Registry String () -> Process () +testAwaitServerDied reg = do + result <- liftIO $ newEmptyMVar + _ <- spawnLocal $ await reg "bobobob" >>= stash result + sleep $ milliSeconds 250 + killProc reg "bye!" + res <- liftIO $ takeMVar result + case res of + ServerUnreachable (DiedException _) -> return () + _ -> liftIO $ assertFailure (show res) + +tests :: NT.Transport -> IO [Test] +tests transport = do + localNode <- newLocalNode transport initRemoteTable + let testProc = withRegistry localNode + return [ + testGroup "Name Registration/Unregistration" + [ + testCase "Simple Registration" + (delayedAssertion + "expected the server to return the incremented state as 7" + localNode RegisteredOk testAddLocalName) + , testCase "Give Away Name" + (testProc testGiveAwayName) + , testCase "Verified Registration" + (testProc testCheckLocalName) + , testCase "Single Process, Multiple Registered Names" + (testProc testMultipleRegistrations) + , testCase "Duplicate Registration Fails" + (testProc testDuplicateRegistrations) + , testCase "Unregister Own Name" + (testProc testUnregisterName) + , testCase "Unregister Unknown Name" + (testProc testUnregisterUnknownName) + , testCase "Unregister Someone Else's Name" + (testProc testUnregisterAnothersName) + ] + , testGroup "Properties" + [ + testCase "Simple Property Registration" + (delayedAssertion + "expected the server to return the property value 42" + localNode (Just 42) testAddLocalProperty) + , testCase "Remote Property Registration" + (delayedAssertion + "expected the server to return the property value 39" + localNode 39 testAddRemoteProperty) + ] + , testGroup "Queries" + [ + testCase "Folding Over Registered Names (Locally)" + (testProc testLocalRegNamesFold) + , testCase "Querying Registered Names (Locally)" + (testProc testLocalQueryNamesFold) + , testCase "Querying Process Where Property Exists (Locally)" + (delayedAssertion + "expected the server to return only the relevant processes" + localNode True testFindByPropertySet) + , testCase "Querying Process Where Property Is Set To Specific Value (Locally)" + (delayedAssertion + "expected the server to return only the relevant processes" + localNode True testFindByPropertyValueSet) + ] + , testGroup "Named Process Monitoring/Tracking" + [ + testCase "Process Death Results In Unregistration" + (testProc testProcessDeathHandling) + , testCase "Monitoring Name Changes" + (testProc testMonitorName) + , testCase "Monitoring Name Changes (KeyOwnerChanged)" + (testProc testMonitorNameChange) + , testCase "Unmonitoring (Ignoring) Changes" + (delayedAssertion + "expected no further notifications after 'unmonitor' was called" + localNode True testUnmonitor) + , testCase "Monitoring Property Changes/Updates" + (delayedAssertion + "expected the server to send additional notifications for each change" + localNode True testMonitorPropertyChanged) + , testCase "Monitoring Property Owner Death" + (testProc testMonitorPropertyOwnerDied) + , testCase "Monitoring Registration" + (testProc testMonitorRegistration) + , testCase "Awaiting Registration" + (testProc testAwaitRegistration) + , testCase "Await without timeout" + (testProc testAwaitRegistrationNoTimeout) + , testCase "Server Died During Await" + (testProc testAwaitServerDied) + , testCase "Monitoring Unregistration" + (testProc testMonitorUnregistration) + ] + ] + +-- | 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 $ tests +