From 9a52fc50897dc4e1e048f9658b323220f5d1a927 Mon Sep 17 00:00:00 2001 From: dvdrw Date: Fri, 19 Jun 2026 17:02:05 +0200 Subject: [PATCH] feat: add global registry actor to system --- actors.cabal | 1 + src/Control/Actor.hs | 70 ++++++++--------- src/Control/Actor/Core.hs | 67 +++++++++++++--- src/Control/Actor/Network.hs | 55 +++++-------- src/Control/Actor/Registry.hs | 131 +++++++++++++++++++++++++++++++ src/Control/Actor/Runtime.hs | 36 ++++++++- src/Control/Actor/Supervision.hs | 2 +- src/Control/Actor/Types.hs | 29 ++++++- 8 files changed, 302 insertions(+), 89 deletions(-) create mode 100644 src/Control/Actor/Registry.hs diff --git a/actors.cabal b/actors.cabal index 2b6afa1..da6d888 100644 --- a/actors.cabal +++ b/actors.cabal @@ -70,6 +70,7 @@ library Control.Actor.Core Control.Actor.Supervision Control.Actor.Network + Control.Actor.Registry -- Modules included in this library but not exported. -- other-modules: diff --git a/src/Control/Actor.hs b/src/Control/Actor.hs index c3d42e7..8e8c268 100644 --- a/src/Control/Actor.hs +++ b/src/Control/Actor.hs @@ -5,10 +5,6 @@ module Control.Actor , module Control.Actor.Core , module Control.Actor.Supervision , module Control.Actor.Network - , pass - , continue - , maybeContinue - , become -- Demo , pingActor , forwardActorWithCell @@ -29,22 +25,26 @@ import Control.Concurrent.STM ( TMVar, TVar , atomically, newEmptyTMVarIO, newTVarIO, readTMVar, readTVarIO, writeTQueue, writeTVar ) -import Control.Monad ((<=<)) +import Control.Monad ((<=<), void) import Control.Monad.Reader (MonadIO (..)) import Unsafe.Coerce (unsafeCoerce) +import Control.Exception (SomeException, try) +import Control.Actor.Registry (createRegistry, registerActor, lookupRemoteActor) -continue :: r -> Actor msg u r -continue x = (\u -> ActorResult (Just x) u Nothing) <$> state +initRuntime :: NodeAddr -> (NodeAddr -> IO (Transport, NodeAddr)) -> IO Runtime +initRuntime myAddr createTransport = do + (transport, actualAddr) <- createTransport myAddr + rt0 <- newRuntime actualAddr transport + let rt = rt0 { rtSendRemote = \addr nm -> getOrCreateConn addr >>= cast' nm } + withRuntime rt $ void $ spawnActor sysHandlerFn stopOnDeath () + withRuntime rt $ void createRegistry + tListen transport $ \ch -> do + result <- try @SomeException $ withRuntime rt $ handleNewConn ch + case result of + Left _ -> chClose ch + Right () -> return () + return rt -maybeContinue :: Maybe r -> Actor msg u r -maybeContinue Nothing = pass -maybeContinue (Just r) = continue r - -pass :: Actor msg u r -pass = (\u -> ActorResult Nothing u Nothing) <$> state - -become :: Handler msg u r -> Actor msg u r -become fn = (\u -> ActorResult Nothing u (Just fn)) <$> state -- Demo @@ -70,7 +70,7 @@ repeatActor r cell () = do system :: IO () system = do - rt <- initRuntime (NodeAddr "localhost" 9000) + rt <- initRuntime (NodeAddr "localhost" 9000) createTCPTransport withRuntime rt $ do pingCell <- liftIO newEmptyTMVarIO forwardCell <- liftIO newEmptyTMVarIO @@ -123,8 +123,8 @@ networkDemo = do -- Port 0 lets the OS pick a free port each run, avoiding stale-socket -- conflicts when re-running in the same GHCi session. - rt1 <- initRuntime (NodeAddr "localhost" 0) - rt2 <- initRuntime (NodeAddr "localhost" 0) + rt1 <- initRuntime (NodeAddr "localhost" 0) createTCPTransport + rt2 <- initRuntime (NodeAddr "localhost" 0) createTCPTransport let addr1 = rtNodeId rt1 addr2 = rtNodeId rt2 @@ -133,27 +133,27 @@ networkDemo = do node2 <- withRuntime rt1 $ connect Nothing addr2 node1 <- withRuntime rt2 $ connect Nothing addr1 - -- Helper: build a RemoteRef that is visible from one node for an actor - -- that was spawned on a different node. The actor's own id uses NodeId 0 - -- (self), but from the caller's node it is addressed by the assigned id. - let toRemote nid ref = - let ActorId _ u = someActorId (SomeActorRef ref) - in RemoteRef (ActorId nid u) - - -- Spawn workers on node 2. + -- Spawn and register workers on node 2. echoRef <- withRuntime rt2 $ spawnActor echoActor stopOnDeath () printRef <- withRuntime rt2 $ spawnActor printerActor stopOnDeath () + withRuntime rt2 $ registerActor "echo" echoRef + withRuntime rt2 $ registerActor "printer" printRef - let remoteEcho = toRemote node2 echoRef :: ActorRef String String - remotePrint = toRemote node2 printRef :: ActorRef String () + -- Small delay to let NMRegistryUpdate propagate to node 1. + threadDelay 100_000 -- 1. Remote cast: fire-and-forget from node 1 to node 2. putStrLn "\n[1] remote cast node 1 -> node 2" + Just remotePrint <- withRuntime rt1 $ lookupRemoteActor "printer" withRuntime rt1 $ cast' "hello from node 1" remotePrint threadDelay 300_000 -- 2. Remote call: request/reply across nodes. putStrLn "\n[2] remote call node 1 <-> node 2" + Just remoteEcho' <- withRuntime rt1 $ lookupRemoteActor "echo" + let + remoteEcho :: ActorRef String String + remoteEcho = (unsafeCoerce remoteEcho') reply <- withRuntime rt1 $ call' "ping" remoteEcho putStrLn $ " reply: " <> show reply @@ -161,7 +161,9 @@ networkDemo = do putStrLn "\n[3] reverse cast node 2 -> node 1" recvVar <- newTVarIO (Nothing :: Maybe String) recvRef <- withRuntime rt1 $ spawnActor (recvActorFn recvVar) stopOnDeath () - let remoteRecv = toRemote node1 recvRef :: ActorRef String () + withRuntime rt1 $ registerActor "receiver" recvRef + threadDelay 100_000 + Just remoteRecv <- withRuntime rt2 $ lookupRemoteActor "receiver" withRuntime rt2 $ cast' "hello from node 2" remoteRecv threadDelay 300_000 readTVarIO recvVar >>= \v -> putStrLn $ " received: " <> show v @@ -184,13 +186,11 @@ networkDemo = do let mortalId = someActorId (SomeActorRef mortalRef) watchId = someActorId (SomeActorRef watchRef) - -- mortalRef's links: on death, send NMDeath to node 1's address. - withRuntime rt2 $ - linkActorTo (RemoteTarget watchId addr1) mortalRef + -- mortalRef's links: on death, send NMDeath to node 1 (by NodeId). + withRuntime rt2 $ linkActorTo (RemoteTarget watchId node1) mortalRef -- watchRef's links: declare interest in mortalId so routeRemoteDeath -- on node 1 can find this actor when NMDeath arrives. - withRuntime rt1 $ - linkActorTo (RemoteTarget mortalId addr2) watchRef + withRuntime rt1 $ linkActorTo (RemoteTarget mortalId node2) watchRef -- Kill the mortal actor by posting directly to its death queue. case mortalRef of diff --git a/src/Control/Actor/Core.hs b/src/Control/Actor/Core.hs index aed4deb..02b77f8 100644 --- a/src/Control/Actor/Core.hs +++ b/src/Control/Actor/Core.hs @@ -11,6 +11,7 @@ module Control.Actor.Core , liftRuntime , notifyOfDeath , spawnActor + , spawnActorAs , linkActorTo , linkTo , killActor @@ -20,6 +21,12 @@ module Control.Actor.Core , cast , call , castIn + , pass + , passWith + , continue + , maybeContinue + , become + , lastMessageFrom ) where import Control.Actor.Runtime @@ -56,6 +63,7 @@ import Control.Monad.Reader ) import Data.Binary (Binary, decode, encode) import Data.Map qualified as Map +import Data.UUID (UUID) import Data.UUID.V4 (nextRandom) newtype ActorM u r = ActorM @@ -71,6 +79,9 @@ runActorM m rt s = runReaderT (unActorM m) (s, rt) state :: ActorM u u state = asks (asEnv . fst) +lastMessageFrom :: ActorM u NodeId +lastMessageFrom = asks (asLatestFrom . fst) + getSelf :: ActorM u SomeActorRef getSelf = do (as, rt) <- ask @@ -95,24 +106,28 @@ liftRuntime :: RuntimeM a -> ActorM u a liftRuntime = ActorM . withReaderT snd . unRuntimeM notifyOfDeath :: Runtime -> DeathMessage -> DeathTarget -> IO () -notifyOfDeath _ dm (LocalTarget q) = atomically $ writeTQueue q dm -notifyOfDeath rt dm (RemoteTarget _ peerAddr) = - withRuntime rt $ rtSendRemote rt peerAddr (NMDeath (dmActorId dm) (toRemoteExitReason (dmReason dm))) +notifyOfDeath _ dm (LocalTarget q) = atomically $ writeTQueue q dm +notifyOfDeath rt dm (RemoteTarget _ nodeId) = + withRuntime rt $ do + maybeAddr <- lookupNode nodeId + case maybeAddr of + Nothing -> return () + Just addr -> rtSendRemote rt addr (NMDeath (dmActorId dm) (toRemoteExitReason (dmReason dm))) -spawnActor :: +spawnActorAs :: (Binary m, Binary r) => + UUID -> Handler m u r -> (DeathMessage -> ActorM u (SupervisorAction u)) -> u -> RuntimeM (ActorRef m r) -spawnActor actorFn deathFn initState = do +spawnActorAs uuid actorFn deathFn initState = do rt <- ask mailbox <- liftIO newTQueueIO deathQ <- liftIO newTQueueIO links <- liftIO $ newTVarIO [] - uuid <- liftIO nextRandom let actorId = ActorId thisNodeId uuid - actorState = ActorState actorId links initState + actorState = ActorState actorId links initState thisNodeId actorRef = LocalRef mailbox deathQ actorState let loop fn as = do @@ -123,11 +138,11 @@ spawnActor actorFn deathFn initState = do case event of Left envelope -> case envelope of - Cast msg -> do - ActorResult _ u' mFn <- runActorM (fn msg) rt as + (from, Cast msg) -> do + ActorResult _ u' mFn <- runActorM (fn msg) rt as {asLatestFrom = from} loop (fromMaybe fn mFn) as {asEnv = u'} - Call msg mv -> do - ActorResult reply u' mFn <- runActorM (fn msg) rt as + (from, Call msg mv) -> do + ActorResult reply u' mFn <- runActorM (fn msg) rt as {asLatestFrom = from} putMVar mv reply loop (fromMaybe fn mFn) as {asEnv = u'} Right dm -> do @@ -154,6 +169,16 @@ spawnActor actorFn deathFn initState = do modifyTVar (rtActors rt) (Map.insert actorId (tid, SomeActorRef actorRef)) return actorRef +spawnActor :: + (Binary m, Binary r) => + Handler m u r -> + (DeathMessage -> ActorM u (SupervisorAction u)) -> + u -> + RuntimeM (ActorRef m r) +spawnActor actorFn deathFn initState = do + uuid <- liftIO nextRandom + spawnActorAs uuid actorFn deathFn initState + linkActorTo :: DeathTarget -> ActorRef m r -> RuntimeM () linkActorTo target (LocalRef {arState}) = liftIO $ atomically $ modifyTVar (asLinks arState) (target :) @@ -175,7 +200,7 @@ stopOnDeath _ = return Stop cast' :: (Binary msg) => msg -> ActorRef msg reply -> RuntimeM () cast' msg (LocalRef {arMsgQ}) = - liftIO $ atomically $ writeTQueue arMsgQ (Cast msg) + liftIO $ atomically $ writeTQueue arMsgQ (thisNodeId, Cast msg) cast' msg (RemoteRef (ActorId nodeId uuid)) = do rt <- ask maybeAddr <- lookupNode nodeId @@ -186,7 +211,7 @@ cast' msg (RemoteRef (ActorId nodeId uuid)) = do call' :: (Binary msg, Binary reply) => msg -> ActorRef msg reply -> RuntimeM (Maybe reply) call' msg (LocalRef {arMsgQ}) = liftIO $ do mv <- newEmptyMVar - atomically $ writeTQueue arMsgQ (Call msg mv) + atomically $ writeTQueue arMsgQ (thisNodeId, Call msg mv) takeMVar mv call' msg (RemoteRef (ActorId nodeId uuid)) = do rt <- ask @@ -220,3 +245,19 @@ castIn ms msg ref = do call :: (Binary msg, Binary reply) => msg -> ActorRef msg reply -> ActorM u (Maybe reply) call = (liftRuntime .) . call' + +continue :: r -> Actor msg u r +continue x = (\u -> ActorResult (Just x) u Nothing) <$> state + +maybeContinue :: Maybe r -> Actor msg u r +maybeContinue Nothing = pass +maybeContinue (Just r) = continue r + +pass :: Actor msg u r +pass = (\u -> ActorResult Nothing u Nothing) <$> state + +passWith :: u -> Actor msg u r +passWith u = return (ActorResult Nothing u Nothing) + +become :: Handler msg u r -> Actor msg u r +become fn = (\u -> ActorResult Nothing u (Just fn)) <$> state diff --git a/src/Control/Actor/Network.hs b/src/Control/Actor/Network.hs index 65ab6ea..de3ee19 100644 --- a/src/Control/Actor/Network.hs +++ b/src/Control/Actor/Network.hs @@ -10,16 +10,17 @@ module Control.Actor.Network , routeRemoteDeath , validateAndDispatch , sysHandlerFn - , initRuntime + , createTCPTransport ) where import Control.Actor.Core - ( Actor, ActorM, ActorResult (..), Handler, cast', liftRuntime, linkActorTo, spawnActor, state, stopOnDeath ) -import Control.Actor.Runtime (Runtime (..), RuntimeM, newRuntime, withRuntime) + ( ActorM, ActorResult (..), Handler, cast', liftRuntime, linkActorTo + , spawnActor, state, stopOnDeath ) +import Control.Actor.Runtime (Runtime (..), RuntimeM, addrToNodeId, withRuntime) import Control.Actor.Supervision (ChildSpec (..), RestartStrategy (..), childWithRef, supervise') import Control.Actor.Transport (ConnHandle (..), Transport (..), createTCPTransport) import Control.Actor.Types -import Control.Concurrent (ThreadId, forkIO) +import Control.Concurrent (forkIO) import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) import Control.Concurrent.STM ( atomically @@ -41,7 +42,7 @@ import Control.Monad.Reader (MonadIO (..), MonadReader (..)) import Data.Binary (decode, decodeOrFail, encode) import Data.ByteString.Lazy (ByteString) import Data.Map qualified as Map -import Data.UUID (UUID) +import Data.Maybe (fromMaybe) -- Connection actors @@ -152,20 +153,12 @@ getOrCreateConn peer = do -- Message dispatch -findByUUID :: UUID -> Map.Map ActorId (ThreadId, SomeActorRef) -> Maybe SomeActorRef -findByUUID uuid actors = - case Map.toList (Map.filterWithKey (\(ActorId _ u) _ -> u == uuid) actors) of - [] -> Nothing - (_, (_, r)):_ -> Just r - routeRemoteDeath :: NodeAddr -> ActorId -> RemoteExitReason -> RuntimeM () routeRemoteDeath senderAddr (ActorId _ uuid) reason = do - rt <- ask + rt <- ask + mSenderNid <- addrToNodeId senderAddr liftIO $ do - localNodeId <- atomically $ do - table <- readTVar (rtNodeTable rt) - return $ Map.foldrWithKey (\k v acc -> if v == senderAddr then Just k else acc) Nothing table - let deadId = ActorId (case localNodeId of { Just n -> n; Nothing -> 0 }) uuid + let deadId = ActorId (fromMaybe 0 mSenderNid) uuid exitReason = case reason of RNormal -> Normal RKilled -> Killed @@ -177,9 +170,9 @@ routeRemoteDeath senderAddr (ActorId _ uuid) reason = do LocalRef {arDeathQ, arState} -> do links <- readTVarIO (asLinks arState) forM_ links $ \case - RemoteTarget (ActorId _ uid) peerAddr - | uid == uuid && peerAddr == senderAddr -> - atomically $ writeTQueue arDeathQ dm + RemoteTarget (ActorId _ uid) nodeId + | uid == uuid && Just nodeId == mSenderNid -> + atomically $ writeTQueue arDeathQ dm _else -> return () RemoteRef _ -> return () @@ -198,6 +191,7 @@ validateAndDispatch senderAddr nm = case nm of NMCast uuid payload -> do rt <- ask + rid <- remoteNodeId liftIO $ do actors <- readTVarIO (rtActors rt) case findByUUID uuid actors of @@ -207,13 +201,14 @@ validateAndDispatch senderAddr nm = case nm of case decodeOrFail payload of Left _ -> return False Right (_, _, msg) -> do - atomically $ writeTQueue arMsgQ (Cast msg) + atomically $ writeTQueue arMsgQ (rid, Cast msg) return True Just (SomeActorRef (RemoteRef _)) -> return False NMCall uuid corrId returnAddr payload -> do rt <- ask + rid <- remoteNodeId liftIO $ do actors <- readTVarIO (rtActors rt) case findByUUID uuid actors of @@ -224,7 +219,7 @@ validateAndDispatch senderAddr nm = case nm of Left _ -> return False Right (_, _, msg) -> do mv <- newEmptyMVar - atomically $ writeTQueue arMsgQ (Call msg mv) + atomically $ writeTQueue arMsgQ (rid, Call msg mv) void $ forkIO $ do reply <- takeMVar mv case reply of @@ -240,6 +235,9 @@ validateAndDispatch senderAddr nm = case nm of routeRemoteDeath senderAddr deadId reason return True + where + remoteNodeId = fromMaybe thisNodeId <$> addrToNodeId senderAddr + -- System event handler sysHandlerFn :: Handler NetworkMessage () () @@ -264,18 +262,3 @@ connect suggestedId peer = do return assigned void $ getOrCreateConn peer return nodeId - --- Runtime initialization - -initRuntime :: NodeAddr -> IO Runtime -initRuntime myAddr = do - (transport, actualAddr) <- createTCPTransport myAddr - rt0 <- newRuntime actualAddr transport - let rt = rt0 { rtSendRemote = \addr nm -> getOrCreateConn addr >>= cast' nm } - withRuntime rt $ void $ spawnActor sysHandlerFn stopOnDeath () - tListen transport $ \ch -> do - result <- try @SomeException $ withRuntime rt $ handleNewConn ch - case result of - Left _ -> chClose ch - Right () -> return () - return rt diff --git a/src/Control/Actor/Registry.hs b/src/Control/Actor/Registry.hs new file mode 100644 index 0000000..c1f0bdb --- /dev/null +++ b/src/Control/Actor/Registry.hs @@ -0,0 +1,131 @@ +module Control.Actor.Registry + ( registerActor, + lookupActor, + lookupRemoteActor, + registryUUID, + registry, + registerDeath, + createRegistry, + ) +where + +import Control.Actor.Core (ActorM, ActorResult (..), Handler, call, call', cast', liftRuntime, linkActorTo, pass, spawnActorAs, state, lastMessageFrom, cast, passWith) +import Control.Actor.Runtime (Runtime (..), RuntimeM, getActorByUUID, withRuntime) +import Control.Actor.Types (ActorId (..), ActorRef (..), DeathMessage (..), DeathTarget (LocalTarget), RegistryMsg (..), SomeActorRef (..), SupervisorAction (Continue), actorRefId, thisNodeId, findByUUID) +import Control.Concurrent.STM (readTVarIO) +import Control.Monad (forM_, when) +import Control.Monad.IO.Class (MonadIO (liftIO)) +import Control.Monad.Reader (ask) +import Data.Map qualified as Map +import Data.UUID (UUID, fromWords) +import Unsafe.Coerce (unsafeCoerce) + +registryUUID :: UUID +registryUUID = fromWords 0 0 0 1 + +type RegistryState = Map.Map String ActorId + +registryHandlerFn :: Handler RegistryMsg RegistryState (Maybe ActorId) +registryHandlerFn msg@(RMRegister name uuid) = do + u <- state + from <- lastMessageFrom + when (from == thisNodeId) $ do + rt <- liftRuntime ask + nodeTable <- liftIO $ readTVarIO (rtNodeTable rt) + forM_ (Map.keys nodeTable) $ \nodeId -> do + cast msg (RemoteRef (ActorId nodeId registryUUID)) + passWith (Map.insert name (ActorId from uuid) u) +registryHandlerFn (RMLookup name) = do + u <- state + return $ ActorResult (Just (Map.lookup name u)) u Nothing +registryHandlerFn (RMDeregister name) = do + u <- state + return $ ActorResult Nothing (Map.delete name u) Nothing +registryHandlerFn msg@(RMDeath deadUUID) = do + u <- state + from <- lastMessageFrom + + let (_, kept) = Map.partition (\(ActorId _ uuid) -> uuid == deadUUID) u + + when (from == thisNodeId) $ do + rt <- liftRuntime ask + nodeTable <- liftIO $ readTVarIO (rtNodeTable rt) + forM_ (Map.keys nodeTable) $ \nodeId -> do + cast msg (RemoteRef (ActorId nodeId registryUUID)) + + passWith kept + +registryDeathFn :: DeathMessage -> ActorM RegistryState (SupervisorAction RegistryState) +registryDeathFn (DeathMessage (ActorId _ deadUUID) _) = do + u <- state + let (_, kept) = Map.partition (\(ActorId _ uuid) -> uuid == deadUUID) u + rt <- liftRuntime ask + nodeTable <- liftIO $ readTVarIO (rtNodeTable rt) + forM_ (Map.keys nodeTable) $ \nodeId -> do + cast (RMDeath deadUUID) (RemoteRef (ActorId nodeId registryUUID)) + + return $ Continue kept + +registry :: RuntimeM (Maybe (ActorRef RegistryMsg (Maybe ActorId))) +registry = do + unsafeCoerce $ getActorByUUID registryUUID + +registerDeath :: ActorId -> RuntimeM () +registerDeath (ActorId _ uuid) = do + r <- registry + mapM_ (cast' (RMDeath uuid)) r + +registerActor :: String -> ActorRef msg r -> RuntimeM () +registerActor name ref = do + rt <- ask + let ActorId _ uuid = actorRefId ref + actors <- liftIO $ readTVarIO (rtActors rt) + case findByUUID registryUUID actors of + Just (SomeActorRef regRef) -> do + let reg = unsafeCoerce regRef :: ActorRef RegistryMsg (Maybe ActorId) + cast' (RMRegister name uuid) reg + case reg of + LocalRef {arDeathQ} -> linkActorTo (LocalTarget arDeathQ) ref + RemoteRef _ -> return () + Nothing -> liftIO $ putStrLn "registerActor: registry not found" + +lookupActor :: String -> RuntimeM (Maybe (ActorRef msg r)) +lookupActor name = do + rt <- ask + actors <- liftIO $ readTVarIO (rtActors rt) + case findByUUID registryUUID actors of + Nothing -> return Nothing + Just (SomeActorRef regRef) -> do + result <- + call' + (RMLookup name) + (unsafeCoerce regRef :: ActorRef RegistryMsg (Maybe ActorId)) + case result of + Just (Just (ActorId 0 uuid)) -> do + actors' <- liftIO $ readTVarIO (rtActors rt) + return $ fmap (\(SomeActorRef r) -> unsafeCoerce r) (findByUUID uuid actors') + Just (Just aid) -> return $ Just (RemoteRef aid) + _else -> return Nothing + +lookupRemoteActor :: String -> RuntimeM (Maybe (ActorRef msg r)) +lookupRemoteActor name = do + rt <- ask + table <- liftIO $ readTVarIO (rtNodeTable rt) + go (Map.keys table) + where + go [] = return Nothing + go (peerNodeId : rest) = do + let remoteReg = + RemoteRef (ActorId peerNodeId registryUUID) :: + ActorRef RegistryMsg (Maybe ActorId) + result <- call' (RMLookup name) remoteReg + case result of + Just (Just (ActorId 0 uuid)) -> + return $ Just (RemoteRef (ActorId peerNodeId uuid)) + Just (Just aid) -> + return $ Just (RemoteRef aid) + _else -> go rest + +createRegistry :: RuntimeM (ActorRef RegistryMsg (Maybe ActorId)) +createRegistry = + spawnActorAs registryUUID registryHandlerFn registryDeathFn Map.empty diff --git a/src/Control/Actor/Runtime.hs b/src/Control/Actor/Runtime.hs index b706495..cbc39e7 100644 --- a/src/Control/Actor/Runtime.hs +++ b/src/Control/Actor/Runtime.hs @@ -6,6 +6,9 @@ module Control.Actor.Runtime , newRuntime , withRuntime , lookupNode + , addrToNodeId + , getActorRef + , getActorByUUID ) where import Control.Actor.Transport (Transport) @@ -13,11 +16,13 @@ import Control.Actor.Types import Control.Concurrent (ThreadId) import Control.Concurrent.MVar (MVar) import Control.Concurrent.STM - ( TMVar, TVar, atomically, newTVarIO, readTVar ) + ( TMVar, TVar, atomically, newTVarIO, readTVar, readTVarIO ) import Control.Monad.Reader - ( MonadIO (..), MonadReader (..), ReaderT (..), runReaderT ) + ( MonadIO (..), MonadReader (..), ReaderT (..), runReaderT, asks ) import Data.ByteString.Lazy (ByteString) +import Data.Foldable (foldl') import Data.Map qualified as Map +import Data.UUID (UUID) data Runtime = Runtime { rtNodeId :: NodeAddr @@ -68,3 +73,30 @@ lookupNode nodeId = do liftIO $ atomically $ do table <- readTVar (rtNodeTable rt) return $ Map.lookup nodeId table + +addrToNodeId :: NodeAddr -> RuntimeM (Maybe NodeId) +addrToNodeId addr = do + rt <- ask + liftIO $ atomically $ do + table <- readTVar (rtNodeTable rt) + return $ Map.foldrWithKey (\k v acc -> if v == addr then Just k else acc) Nothing table + +getActorRef :: ActorId -> RuntimeM (Maybe SomeActorRef) +getActorRef aid = do + actVar <- asks rtActors + actors <- liftIO $ readTVarIO actVar + return $ snd <$> Map.lookup aid actors + +firstByKey' :: Ord k => (k -> Bool) -> Map.Map k v -> Maybe v +firstByKey' p = snd . foldl' step (False, Nothing) . Map.toAscList + where + step (True, mv) _ = (True, mv) + step (False, _) (k,v) + | p k = (True, Just v) + | otherwise = (False, Nothing) + +getActorByUUID :: UUID -> RuntimeM (Maybe SomeActorRef) +getActorByUUID uuid = do + actVar <- asks rtActors + actors <- liftIO $ readTVarIO actVar + return $ snd <$> firstByKey' (\(ActorId _ x) -> x == uuid) actors diff --git a/src/Control/Actor/Supervision.hs b/src/Control/Actor/Supervision.hs index 33167a9..e7301ef 100644 --- a/src/Control/Actor/Supervision.hs +++ b/src/Control/Actor/Supervision.hs @@ -16,7 +16,7 @@ module Control.Actor.Supervision , killSlot ) where -import Control.Actor.Core (Actor, ActorM, Handler, liftRuntime, linkActorTo, spawnActor) +import Control.Actor.Core (ActorM, Handler, liftRuntime, linkActorTo, spawnActor) import Control.Actor.Runtime (Runtime (..), RuntimeM, withRuntime) import Control.Actor.Types import Data.Binary (Binary) diff --git a/src/Control/Actor/Types.hs b/src/Control/Actor/Types.hs index 16927b1..cc043a1 100644 --- a/src/Control/Actor/Types.hs +++ b/src/Control/Actor/Types.hs @@ -16,9 +16,12 @@ module Control.Actor.Types , ActorRef (..) , SomeActorRef (..) , someActorId + , actorRefId , SupervisorAction (..) , CorrelationId , Envelope (..) + , RegistryMsg (..) + , findByUUID ) where import Control.Concurrent.MVar (MVar) @@ -28,6 +31,8 @@ import Data.Binary (Binary) import Data.ByteString.Lazy (ByteString) import Data.UUID (UUID) import GHC.Generics (Generic) +import Data.Map qualified as Map +import Control.Concurrent (ThreadId) data NodeAddr = NodeAddr { nodeHost :: String @@ -59,7 +64,7 @@ data DeathMessage = DeathMessage data DeathTarget = LocalTarget (TQueue DeathMessage) - | RemoteTarget ActorId NodeAddr + | RemoteTarget ActorId NodeId data RemoteExitReason = RNormal @@ -88,6 +93,7 @@ data ActorState u = ActorState { asId :: ActorId , asLinks :: TVar [DeathTarget] , asEnv :: u + , asLatestFrom :: NodeId } type CorrelationId = Integer @@ -98,7 +104,7 @@ data Envelope msg reply data ActorRef msg reply = forall u. LocalRef - { arMsgQ :: TQueue (Envelope msg reply) + { arMsgQ :: TQueue (NodeId, Envelope msg reply) , arDeathQ :: TQueue DeathMessage , arState :: ActorState u } @@ -110,6 +116,25 @@ someActorId :: SomeActorRef -> ActorId someActorId (SomeActorRef (LocalRef {arState})) = asId arState someActorId (SomeActorRef (RemoteRef aid)) = aid +actorRefId :: ActorRef msg r -> ActorId +actorRefId (LocalRef {arState}) = asId arState +actorRefId (RemoteRef aid) = aid + data SupervisorAction u = Stop | Continue u + +data RegistryMsg + = RMRegister String UUID + | RMLookup String + | RMDeregister String + | RMDeath UUID + deriving (Generic) + +instance Binary RegistryMsg + +findByUUID :: UUID -> Map.Map ActorId (ThreadId, SomeActorRef) -> Maybe SomeActorRef +findByUUID uuid actors = + case Map.toList (Map.filterWithKey (\(ActorId _ u) _ -> u == uuid) actors) of + [] -> Nothing + (_, (_, r)):_ -> Just r