diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,13 @@ +2016-08-21 Facundo Domínguez 0.7.0 + +* Change type of message sent by `say` from a 3-tuple to a proper +type (`SayMessage`) with a proper `UTCTime`. (#291) +* Expose the MonitorRef in the withMonitor call. +* Have unmonitor remove the monitor message in the inbox. (#268) +* Remove Mx Data Tables. This API isn't used, is easy to replace with various +other packages. (#276) +* Add Ord instance to SpawnRef. + 2016-10-13 Facundo Domínguez 0.6.6 * Remove monitors from remote nodes when a process dies. (#295) diff --git a/distributed-process.cabal b/distributed-process.cabal --- a/distributed-process.cabal +++ b/distributed-process.cabal @@ -1,5 +1,5 @@ Name: distributed-process -Version: 0.6.6 +Version: 0.7.0 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 @@ -21,7 +21,7 @@ You will probably also want to install a Cloud Haskell backend such as distributed-process-simplelocalnet. -Tested-With: GHC==7.2.2 GHC==7.4.1 GHC==7.4.2 GHC==7.6.2 +Tested-With: GHC==7.6.3 GHC==7.8.4 GHC==7.10.3 GHC==8.0.1 Category: Control extra-source-files: ChangeLog @@ -41,10 +41,10 @@ default: False Library - Build-Depends: base >= 4.4 && < 5, + Build-Depends: base >= 4.6 && < 5, binary >= 0.6.3 && < 0.9, hashable >= 1.2.0.5 && < 1.3, - network-transport >= 0.4.1.0 && < 0.5, + network-transport >= 0.4.1.0 && < 0.6, stm >= 2.4 && < 2.5, transformers >= 0.2 && < 0.6, mtl >= 2.0 && < 2.4, @@ -53,8 +53,10 @@ random >= 1.0 && < 1.2, distributed-static >= 0.2 && < 0.4, rank1dynamic >= 0.1 && < 0.4, - syb >= 0.3 && < 0.7, - exceptions >= 0.5 + syb >= 0.3 && < 0.8, + exceptions >= 0.5, + containers >= 0.5 && < 0.6, + deepseq >= 1.3.0.1 && < 1.6 Exposed-modules: Control.Distributed.Process, Control.Distributed.Process.Closure, Control.Distributed.Process.Debug, @@ -76,7 +78,6 @@ Control.Distributed.Process.UnsafePrimitives Control.Distributed.Process.Management.Internal.Agent, Control.Distributed.Process.Management.Internal.Bus, - Control.Distributed.Process.Management.Internal.Table, Control.Distributed.Process.Management.Internal.Types, Control.Distributed.Process.Management.Internal.Trace.Primitives, Control.Distributed.Process.Management.Internal.Trace.Remote, @@ -84,21 +85,33 @@ Control.Distributed.Process.Management.Internal.Trace.Tracer ghc-options: -Wall HS-Source-Dirs: src - if impl(ghc <= 7.4.2) - Build-Depends: containers >= 0.4 && < 0.5, - deepseq == 1.3.0.0 - else - Build-Depends: containers >= 0.4 && < 0.6, - deepseq >= 1.3.0.1 && < 1.6 + other-extensions: BangPatterns + CPP + DeriveDataTypeable + DeriveFunctor + DeriveGeneric + ExistentialQuantification + FlexibleInstances + GADTs + GeneralizedNewtypeDeriving + KindSignatures + MagicHash + PatternGuards + RankNTypes + RecordWildCards + ScopedTypeVariables + StandaloneDeriving + TypeFamilies + TypeSynonymInstances + UnboxedTuples + UndecidableInstances if flag(old-locale) Build-Depends: time < 1.5, old-locale >= 1.0 && <1.1 else Build-Depends: time >= 1.5 if flag(th) - if impl(ghc <= 7.4.2) - Build-Depends: template-haskell >= 2.7 && < 2.8 - else - Build-Depends: template-haskell >= 2.6 && < 2.12 + other-extensions: TemplateHaskell + Build-Depends: template-haskell >= 2.6 && < 2.12 Exposed-modules: Control.Distributed.Process.Internal.Closure.TH CPP-Options: -DTemplateHaskellSupport @@ -106,7 +119,7 @@ benchmark distributed-process-throughput Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.4 && < 5, + Build-Depends: base >= 4.6 && < 5, distributed-process, network-transport-tcp >= 0.3 && < 0.6, bytestring >= 0.9 && < 0.11, @@ -116,7 +129,7 @@ benchmark distributed-process-latency Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.4 && < 5, + Build-Depends: base >= 4.6 && < 5, distributed-process, network-transport-tcp >= 0.3 && < 0.6, bytestring >= 0.9 && < 0.11, @@ -126,7 +139,7 @@ benchmark distributed-process-channels Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.4 && < 5, + Build-Depends: base >= 4.6 && < 5, distributed-process, network-transport-tcp >= 0.3 && < 0.6, bytestring >= 0.9 && < 0.11, @@ -136,7 +149,7 @@ benchmark distributed-process-spawns Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.4 && < 5, + Build-Depends: base >= 4.6 && < 5, distributed-process, network-transport-tcp >= 0.3 && < 0.6, bytestring >= 0.9 && < 0.11, @@ -146,7 +159,7 @@ benchmark distributed-process-ring Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.4 && < 5, + Build-Depends: base >= 4.6 && < 5, distributed-process, network-transport-tcp >= 0.3 && < 0.6, bytestring >= 0.9 && < 0.11, diff --git a/src/Control/Distributed/Process.hs b/src/Control/Distributed/Process.hs --- a/src/Control/Distributed/Process.hs +++ b/src/Control/Distributed/Process.hs @@ -102,6 +102,7 @@ , monitorPort , unmonitor , withMonitor + , withMonitor_ , MonitorRef -- opaque , ProcessLinkException(..) , NodeLinkException(..) @@ -269,6 +270,7 @@ , monitorPort , unmonitor , withMonitor + , withMonitor_ -- Logging , say -- Registry diff --git a/src/Control/Distributed/Process/Internal/Primitives.hs b/src/Control/Distributed/Process/Internal/Primitives.hs --- a/src/Control/Distributed/Process/Internal/Primitives.hs +++ b/src/Control/Distributed/Process/Internal/Primitives.hs @@ -77,7 +77,9 @@ , monitor , unmonitor , withMonitor + , withMonitor_ -- * Logging + , SayMessage(..) , say -- * Registry , register @@ -125,8 +127,9 @@ import Prelude hiding (catch) #endif -import Data.Binary (decode) -import Data.Time.Clock (getCurrentTime) +import Data.Binary (Binary(..), Put, Get, decode) +import Data.Time.Clock (getCurrentTime, UTCTime(..)) +import Data.Time.Calendar (Day(..)) import Data.Time.Format (formatTime) #if MIN_VERSION_time(1,5,0) import Data.Time.Format (defaultTimeLocale) @@ -134,7 +137,7 @@ import System.Locale (defaultTimeLocale) #endif import System.Timeout (timeout) -import Control.Monad (when) +import Control.Monad (when, void) import Control.Monad.Reader (ask) import Control.Monad.IO.Class (liftIO) import Control.Monad.Catch @@ -184,6 +187,7 @@ , SpawnRef(..) , ProcessSignal(..) , NodeMonitorNotification(..) + , ProcessMonitorNotification(..) , monitorCounter , spawnCounter , SendPort(..) @@ -885,13 +889,20 @@ -- @withMonitor@ returns, there might still be unreceived monitor -- messages in the queue. -- -withMonitor :: ProcessId -> Process a -> Process a -withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code) - -- unmonitor blocks waiting for the response, so there's a possibility - -- that an exception might interrupt withMonitor before the unmonitor - -- has completed. I think that's better than making the unmonitor - -- uninterruptible. +withMonitor :: ProcessId -> (MonitorRef -> Process a) -> Process a +withMonitor pid = bracket (monitor pid) unmonitor +-- | Establishes temporary monitoring of another process. +-- +-- @withMonitor_ pid code@ sets up monitoring of @pid@ for the duration +-- of @code@. Note: although monitoring is no longer active when +-- @withMonitor_@ returns, there might still be unreceived monitor +-- messages in the queue. +-- +-- Since 0.6.1 +withMonitor_ :: ProcessId -> Process a -> Process a +withMonitor_ p = withMonitor p . const + -- | Remove a link -- -- This is synchronous in the sense that once it returns you are guaranteed @@ -928,12 +939,21 @@ -- | Remove a monitor -- -- This has the same synchronous/asynchronous nature as 'unlink'. +-- +-- ProcessMonitorNotification messages for the given MonitorRef are removed from +-- the mailbox. unmonitor :: MonitorRef -> Process () unmonitor ref = do unmonitorAsync ref receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref) (\_ -> return ()) ] + -- Discard the notification if any. With the current NC implementation at most + -- one notification is in the mailbox for any given ref. + void $ receiveTimeout 0 + [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref' == ref) + (const $ return ()) + ] -------------------------------------------------------------------------------- -- Exception handling -- @@ -947,7 +967,7 @@ -- | Lift 'Control.Exception.try' try :: Exception e => Process a -> Process (Either e a) try = Catch.try -{-# DEPRECATED try "Use Control.Monad.Catch.mask_ instead" #-} +{-# DEPRECATED try "Use Control.Monad.Catch.try instead" #-} -- | Lift 'Control.Exception.mask' mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b @@ -1059,17 +1079,52 @@ -- Logging -- -------------------------------------------------------------------------------- +data SayMessage = SayMessage { sayTime :: UTCTime + , sayProcess :: ProcessId + , sayMessage :: String } + deriving (Typeable) + +-- There is sadly no Show UTCTime instance +instance Show SayMessage where + showsPrec p msg = + showParen (p >= 11) + $ showString "SayMessage " + . showString (formatTime defaultTimeLocale "%c" (sayTime msg)) + . showChar ' ' + . showsPrec 11 (sayProcess msg) . showChar ' ' + . showsPrec 11 (sayMessage msg) . showChar ' ' + +instance Binary SayMessage where + put s = do + putUTCTime (sayTime s) + put (sayProcess s) + put (sayMessage s) + get = SayMessage <$> getUTCTime <*> get <*> get + +-- Sadly there is no Binary UTCTime instance +putUTCTime :: UTCTime -> Put +putUTCTime (UTCTime (ModifiedJulianDay day) tod) = do + put day + put (toRational tod) + +getUTCTime :: Get UTCTime +getUTCTime = do + day <- get + tod <- get + return $! UTCTime (ModifiedJulianDay day) + (fromRational tod) + -- | Log a string -- --- @say message@ sends a message (time, pid of the current process, message) --- to the process registered as 'logger'. By default, this process simply --- sends the string to 'stderr'. Individual Cloud Haskell backends might --- replace this with a different logger process, however. +-- @say message@ sends a message of type 'SayMessage' with the current time and +-- 'ProcessId' of the current process to the process registered as @logger@. By +-- default, this process simply sends the string to @stderr@. Individual Cloud +-- Haskell backends might replace this with a different logger process, however. say :: String -> Process () say string = do now <- liftIO getCurrentTime us <- getSelfPid - nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string) + nsend "logger" (SayMessage now us string) -------------------------------------------------------------------------------- -- Registry -- diff --git a/src/Control/Distributed/Process/Internal/Types.hs b/src/Control/Distributed/Process/Internal/Types.hs --- a/src/Control/Distributed/Process/Internal/Types.hs +++ b/src/Control/Distributed/Process/Internal/Types.hs @@ -629,7 +629,7 @@ -- | 'SpawnRef' are used to return pids of spawned processes newtype SpawnRef = SpawnRef Int32 - deriving (Show, Binary, Typeable, Eq) + deriving (Show, Binary, Typeable, Eq, Ord) -- | (Asynchronius) reply from 'spawn' data DidSpawn = DidSpawn SpawnRef ProcessId diff --git a/src/Control/Distributed/Process/Management.hs b/src/Control/Distributed/Process/Management.hs --- a/src/Control/Distributed/Process/Management.hs +++ b/src/Control/Distributed/Process/Management.hs @@ -83,28 +83,8 @@ -- system information. Agents maintain their own internal state privately (via a -- state transformer - see 'mxGetLocal' et al), however it is possible for -- agents to share additional data with each other (and the outside world) --- using /data tables/. --- --- Each agent is assigned its own data table, which acts as a shared map, where --- the keys are @String@s and the values are @Serializable@ datum of whatever --- type the agent or its clients stores. --- --- Because an agent's /data table/ stores its values in raw 'Message' format, --- it works effectively as an /un-typed dictionary/, into which data of varying --- types can be fed and later retrieved. The upside of this is that different --- keys can be mapped to various types without any additional work on the part --- of the developer. The downside is that the code reading these values must --- know in advance what type(s) to expect, and the API provides no additional --- support for handling that. --- --- Publishing is accomplished using the 'mxPublish' and 'mxSet' APIs, whilst --- querying and deletion are handled by 'mxGet', 'mxClear', 'mxPurgeTable' and --- 'mxDropTable' respectively. --- --- When a management agent terminates, their tables are left in memory despite --- termination, such that an agent may resume its role (by restarting) or have --- its 'MxAgentId' taken over by another subsequent agent, leaving the data --- originally captured in place. +-- using whatever mechanism the user wishes, e.g., acidstate, or shared memory +-- primitives. -- -- [Defining Agents] -- @@ -194,10 +174,6 @@ -- the event bus /and/ their own mailboxes, plus searching through the set of -- event sinks (for each agent) to determine the right handler for the event. -- --- Each management agent requires not only its own @Process@ (in which the agent --- code is run), but also a peer process that provides its /data table/. These --- data tables also have to be coordinated and manaaged on each agent's behalf. --- -- [Architecture Overview] -- -- The architecture of the management event bus is internal and subject to @@ -264,13 +240,6 @@ , mxGetLocal , mxUpdateLocal , liftMX - -- * Mx Data API - , mxPublish - , mxSet - , mxGet - , mxClear - , mxPurgeTable - , mxDropTable ) where import Control.Applicative @@ -281,10 +250,7 @@ , TChan ) import Control.Distributed.Process.Internal.Primitives - ( newChan - , nsend - , receiveWait - , matchChan + ( receiveWait , matchAny , matchSTM , unwrapMessage @@ -302,14 +268,12 @@ , unsafeCreateUnencodedMessage ) import Control.Distributed.Process.Management.Internal.Bus (publishEvent) -import qualified Control.Distributed.Process.Management.Internal.Table as Table import Control.Distributed.Process.Management.Internal.Types ( MxAgentId(..) , MxAgent(..) , MxAction(..) , ChannelSelector(..) , MxAgentState(..) - , MxAgentStart(..) , MxSink , MxEvent(..) ) @@ -335,42 +299,6 @@ bus <- localEventBus . processNode <$> ask liftIO $ publishEvent bus $ unsafeCreateUnencodedMessage msg --- | Publish an arbitrary @Message@ as a property in the management database. --- --- For publishing @Serializable@ data, use 'mxSet' instead. --- -mxPublish :: MxAgentId -> String -> Message -> Process () -mxPublish a k v = Table.set k v (Table.MxForAgent a) - --- | Sets an arbitrary @Serializable@ datum against a key in the management --- database. Note that /no attempt is made to force the argument/, therefore --- it is very important that you do not pass unevaluated thunks that might --- crash some other, arbitrary process (or management agent!) that obtains --- and attempts to force the value later on. --- -mxSet :: Serializable a => MxAgentId -> String -> a -> Process () -mxSet mxId key msg = do - Table.set key (unsafeCreateUnencodedMessage msg) (Table.MxForAgent mxId) - --- | Fetches a property from the management database for the given key. --- If the property is not set, or does not match the expected type when --- typechecked (at runtime), returns @Nothing@. -mxGet :: Serializable a => MxAgentId -> String -> Process (Maybe a) -mxGet = Table.fetch . Table.MxForAgent - --- | Clears a property from the management database using the given key. --- If the key does not exist in the database, this is a noop. -mxClear :: MxAgentId -> String -> Process () -mxClear mxId key = Table.clear key (Table.MxForAgent mxId) - --- | Purges a table in the management database of all its stored properties. -mxPurgeTable :: MxAgentId -> Process () -mxPurgeTable = Table.purge . Table.MxForAgent - --- | Deletes a table from the management database. -mxDropTable :: MxAgentId -> Process () -mxDropTable = Table.delete . Table.MxForAgent - -------------------------------------------------------------------------------- -- API for writing user defined management extensions (i.e., agents) -- -------------------------------------------------------------------------------- @@ -490,10 +418,7 @@ return pid where start (sendTChan, recvTChan) = do - (sp, rp) <- newChan - nsend Table.mxTableCoordinator (MxAgentStart sp mxId) - tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ] - let nState = MxAgentState mxId sendTChan tablePid initState + let nState = MxAgentState mxId sendTChan initState runAgent dtor handlers InputChan recvTChan nState runAgent :: MxAgent s () diff --git a/src/Control/Distributed/Process/Management/Internal/Table.hs b/src/Control/Distributed/Process/Management/Internal/Table.hs deleted file mode 100644 --- a/src/Control/Distributed/Process/Management/Internal/Table.hs +++ /dev/null @@ -1,222 +0,0 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE DeriveGeneric #-} -{-# LANGUAGE RecordWildCards #-} - -module Control.Distributed.Process.Management.Internal.Table - ( MxTableRequest(..) - , MxTableId(..) - , mxTableCoordinator - , startTableCoordinator - , delete - , purge - , clear - , set - , get - , fetch - ) where - -import Control.Distributed.Process.Internal.Primitives - ( receiveWait - , receiveChan - , match - , matchAny - , matchIf - , matchChan - , send - , nsend - , sendChan - , getSelfPid - , link - , monitor - , unwrapMessage - , newChan - , withMonitor - ) -import Control.Distributed.Process.Internal.Types - ( Process - , ProcessId - , ProcessMonitorNotification(..) - , SendPort - , ReceivePort - , Message - , unsafeCreateUnencodedMessage - ) -import Control.Distributed.Process.Management.Internal.Types - ( MxTableId(..) - , MxAgentId(..) - , MxAgentStart(..) - , Fork) -import Control.Distributed.Process.Serializable (Serializable) -import Control.Monad.IO.Class (liftIO) -import Data.Accessor (Accessor, accessor, (^=), (^:)) -import Data.Binary (Binary) -import Data.Map (Map) -import qualified Data.Map as Map -import Data.Typeable (Typeable) - -import GHC.Generics - --- An extremely lightweight shared Map implementation, for use --- by /management agents/ and their cohorts. Each agent is assigned --- a table, into which any serializable @Message@ can be inserted. --- Data are inserted, removed and searched for via their key, which --- is a string. Tables can be purged, values can be set, fetched or --- cleared/removed. --- - -data MxTableRequest = - Delete - | Purge - | Clear !String - | Set !String !Message - | Get !String !(SendPort (Maybe Message)) -- see [note: un-typed send port] - deriving (Typeable, Generic) -instance Binary MxTableRequest where - -data MxTableState = MxTableState { _name :: !String - , _entries :: !(Map String Message) - } - -type MxTables = Map MxAgentId ProcessId - -mxTableCoordinator :: String -mxTableCoordinator = "mx.table.coordinator" - -delete :: MxTableId -> Process () -delete = sendReq Delete - -purge :: MxTableId -> Process () -purge = sendReq Purge - -clear :: String -> MxTableId -> Process () -clear k = sendReq (Clear k) - -set :: String -> Message -> MxTableId -> Process () -set k v = sendReq (Set k v) - -fetch :: forall a. (Serializable a) - => MxTableId - -> String - -> Process (Maybe a) -fetch (MxForPid pid) key = get pid key -fetch mxId@(MxForAgent _) key = do - (sp, rp) <- newChan :: Process (SendPort (Maybe Message), - ReceivePort (Maybe Message)) - sendReq (Get key sp) mxId - receiveChan rp >>= maybe (return Nothing) - (unwrapMessage :: Message -> Process (Maybe a)) - --- [note: un-typed send port] --- Here, fetch uses a typed channel over a raw Message to obtain --- its result, so type checking is deferred until receipt and will --- be handled in the caller's thread. This is necessary because --- the server portion of the code knows nothing about the types --- involved, nor should it, since these tables can be used to --- store arbitrary serializable data. - -get :: forall a. (Serializable a) - => ProcessId - -> String - -> Process (Maybe a) -get pid key = do - safeFetch pid key >>= maybe (return Nothing) - (unwrapMessage :: Message -> Process (Maybe a)) - -safeFetch :: ProcessId -> String -> Process (Maybe Message) -safeFetch pid key = do - (sp, rp) <- newChan - send pid $ Get key sp - withMonitor pid $ do - receiveWait [ - matchChan rp return - , matchIf (\(ProcessMonitorNotification _ pid' _) -> pid' == pid) - (\_ -> return $ Just (unsafeCreateUnencodedMessage ())) - ] - -sendReq :: MxTableRequest -> MxTableId -> Process () -sendReq req tid = (resolve tid) req - -resolve :: Serializable a => MxTableId -> (a -> Process ()) -resolve (MxForAgent agent) = \msg -> nsend mxTableCoordinator (agent, msg) -resolve (MxForPid pid) = \msg -> send pid msg - -startTableCoordinator :: Fork -> Process () -startTableCoordinator fork = run Map.empty - where - run :: MxTables -> Process () - run tables = - receiveWait [ - -- note that this state change can race with MxAgentStart requests - match (\(ProcessMonitorNotification _ pid _) -> do - return $ Map.filter (/= pid) tables) - , match (\(MxAgentStart ch agent) -> do - lookupAgent tables agent >>= \(p, t) -> do - sendChan ch p >> return t) - , match (\req@(agent, tReq :: MxTableRequest) -> do - case tReq of - Get k sp -> do - lookupAgent tables agent >>= \(p, t) -> do - safeFetch p k >>= sendChan sp >> return t - _ -> do - handleRequest tables req) - , matchAny (\_ -> return tables) -- unrecognised messages are dropped - ] >>= run - - handleRequest :: MxTables - -> (MxAgentId, MxTableRequest) - -> Process MxTables - handleRequest tables' (agent, req) = do - lookupAgent tables' agent >>= \(p, t) -> send p req >> return t - - lookupAgent :: MxTables -> MxAgentId -> Process (ProcessId, MxTables) - lookupAgent tables' agentId' = do - case Map.lookup agentId' tables' of - Nothing -> launchNew agentId' tables' - Just p -> return (p, tables') - - launchNew :: MxAgentId - -> MxTables - -> Process (ProcessId, MxTables) - launchNew mxId tblMap = do - let initState = MxTableState { _name = (agentId mxId) - , _entries = Map.empty - } - (pid, _) <- spawnSup $ tableHandler initState - return $ (pid, mxId `seq` pid `seq` Map.insert mxId pid tblMap) - - spawnSup proc = do - us <- getSelfPid - -- we need to use that passed in "fork", in order to - -- break an import cycle with Node.hs courtesy of the - -- management agent, API and tracing modules - them <- liftIO $ fork $ link us >> proc - ref <- monitor them - return (them, ref) - -tableHandler :: MxTableState -> Process () -tableHandler state = do - ns <- receiveWait [ - match (handleTableRequest state) - , matchAny (\_ -> return (Just state)) - ] - case ns of - Nothing -> return () - Just s' -> tableHandler s' - where - handleTableRequest _ Delete = return Nothing - handleTableRequest st Purge = return $ Just $ (entries ^= Map.empty) $ st - handleTableRequest st (Clear k) = return $ Just $ (entries ^: (k `seq` Map.delete k)) $ st - handleTableRequest st (Set k v) = return $ Just $ (entries ^: (k `seq` v `seq` Map.insert k v)) st - handleTableRequest st (Get k c) = getEntry k c st >> return (Just st) - -getEntry :: String - -> SendPort (Maybe Message) - -> MxTableState - -> Process () -getEntry k m MxTableState{..} = do - sendChan m =<< return (Map.lookup k _entries) - -entries :: Accessor MxTableState (Map String Message) -entries = accessor _entries (\ls st -> st { _entries = ls }) diff --git a/src/Control/Distributed/Process/Management/Internal/Types.hs b/src/Control/Distributed/Process/Management/Internal/Types.hs --- a/src/Control/Distributed/Process/Management/Internal/Types.hs +++ b/src/Control/Distributed/Process/Management/Internal/Types.hs @@ -4,12 +4,10 @@ {-# LANGUAGE DeriveGeneric #-} module Control.Distributed.Process.Management.Internal.Types ( MxAgentId(..) - , MxTableId(..) , MxAgentState(..) , MxAgent(..) , MxAction(..) , ChannelSelector(..) - , MxAgentStart(..) , Fork , MxSink , MxEvent(..) @@ -24,7 +22,6 @@ ( Process , ProcessId , Message - , SendPort , DiedReason , NodeId ) @@ -99,17 +96,10 @@ newtype MxAgentId = MxAgentId { agentId :: String } deriving (Typeable, Binary, Eq, Ord) -data MxTableId = - MxForAgent !MxAgentId - | MxForPid !ProcessId - deriving (Typeable, Generic) -instance Binary MxTableId where - data MxAgentState s = MxAgentState { mxAgentId :: !MxAgentId , mxBus :: !(TChan Message) - , mxSharedTable :: !ProcessId , mxLocalState :: !s } @@ -126,14 +116,6 @@ , Typeable , Applicative ) - -data MxAgentStart = MxAgentStart - { - mxAgentTableChan :: SendPort ProcessId - , mxAgentIdStart :: MxAgentId - } - deriving (Typeable, Generic) -instance Binary MxAgentStart where data ChannelSelector = InputChan | Mailbox diff --git a/src/Control/Distributed/Process/Node.hs b/src/Control/Distributed/Process/Node.hs --- a/src/Control/Distributed/Process/Node.hs +++ b/src/Control/Distributed/Process/Node.hs @@ -37,6 +37,13 @@ , filterWithKey , foldlWithKey ) +import qualified Data.Set as Set +import Data.Time.Format (formatTime) +#if MIN_VERSION_time(1,5,0) +import Data.Time.Format (defaultTimeLocale) +#else +import System.Locale (defaultTimeLocale) +#endif import Data.Set (Set) import qualified Data.Set as Set ( empty @@ -178,10 +185,6 @@ import Control.Distributed.Process.Management.Internal.Agent ( mxAgentController ) -import qualified Control.Distributed.Process.Management.Internal.Table as Table - ( mxTableCoordinator - , startTableCoordinator - ) import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace ( remoteTable ) @@ -209,6 +212,7 @@ , match , sendChan , unwrapMessage + , SayMessage(..) ) import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..)) import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable) @@ -316,8 +320,6 @@ -- before /that/ process has started - this is a totally harmless race -- however, so we deliberably ignore it startDefaultTracer node - tableCoordinatorPid <- fork $ Table.startTableCoordinator fork - runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid logger <- forkProcess node loop runProcess node $ do register "logger" logger @@ -326,12 +328,11 @@ -- process which uses 'send' or other primitives which are traced. register "trace.logger" logger where - fork = forkProcess node - loop = do receiveWait - [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do - liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string + [ match $ \(SayMessage time pid string) -> do + let time' = formatTime defaultTimeLocale "%c" time + liftIO . hPutStrLn stderr $ time' ++ " " ++ show pid ++ ": " ++ string loop , match $ \((time, string) :: (String, String)) -> do -- this is a 'trace' message from the local node tracer