dvdrw/actors
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add global registry actor to system

Browse Source
This commit is contained in:
dvdrw
2026-06-19 17:02:05 +02:00
parent 25a27eda92
commit 9a52fc5089
8 changed files with 302 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
actors.cabal
+1
View File
@@ -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:
src/Control/Actor.hs
+35 -35
View File
@@ -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
src/Control/Actor/Core.hs
+53 -12
View File
@@ -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
@@ -96,23 +107,27 @@ 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 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
src/Control/Actor/Network.hs
+17 -34
View File
@@ -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
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,8 +170,8 @@ 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 ->
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
src/Control/Actor/Registry.hs
+131
View File
@@ -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
src/Control/Actor/Runtime.hs
+34 -2
View File
@@ -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
src/Control/Actor/Supervision.hs
+1 -1
View File
@@ -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)
src/Control/Actor/Types.hs
+27 -2
View File
@@ -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