distributed-process 0.6.6 → 0.7.0

raw patch · 9 files changed

+133/−367 lines, 9 filesdep ~basedep ~containersdep ~deepseqPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base, containers, deepseq, network-transport, syb, template-haskell

API changes (from Hackage documentation)

Files

• ChangeLog +10/−0
• distributed-process.cabal +35/−22
• src/Control/Distributed/Process.hs +2/−0
• src/Control/Distributed/Process/Internal/Primitives.hs +70/−15
• src/Control/Distributed/Process/Internal/Types.hs +1/−1
• src/Control/Distributed/Process/Management.hs +4/−79
• src/Control/Distributed/Process/Management/Internal/Table.hs +0/−222
• src/Control/Distributed/Process/Management/Internal/Types.hs +0/−18
• src/Control/Distributed/Process/Node.hs +11/−10

@@ -1,3 +1,13 @@+2016-08-21 Facundo Domínguez <facundo.dominguez@tweag.io> 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 <facundo.dominguez@tweag.io> 0.6.6  * Remove monitors from remote nodes when a process dies. (#295)

@@ -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,

@@ -102,6 +102,7 @@   , monitorPort   , unmonitor   , withMonitor+  , withMonitor_   , MonitorRef -- opaque   , ProcessLinkException(..)   , NodeLinkException(..)@@ -269,6 +270,7 @@   , monitorPort   , unmonitor   , withMonitor+  , withMonitor_     -- Logging   , say     -- Registry

@@ -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                                                                   --

@@ -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

@@ -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 ()

@@ -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 })

@@ -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 

@@ -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