From 87c2a68ef6eddb2c210ec7e5c3aef994a39bbee7 Mon Sep 17 00:00:00 2001 From: Helmut Merz Date: Tue, 27 Oct 2026 09:40:05 +0100 Subject: [PATCH] core.actor: Mailbox class (with status), send() function --- scopes/core/actor.py | 33 +++++++++++++++++++-------------- scopes/core/core.py | 28 +++++++++++++--------------- scopes/core/environ.py | 2 +- tests/test_core.py | 24 ++++++++++++------------ 4 files changed, 45 insertions(+), 42 deletions(-) diff --git a/scopes/core/actor.py b/scopes/core/actor.py index b2787f6..32c1aef 100644 --- a/scopes/core/actor.py +++ b/scopes/core/actor.py @@ -5,10 +5,17 @@ from collections.abc import Awaitable, Callable from typing import Any, Optional from scopes.util import keywords as k, Symbol -k('full quit') + +# keywords: +k('active full ok quit retired suspended') type Behavior = Callable[[Any], Awaitable[Behavior | None]] -type Mailbox = asyncio.Queue + +class Mailbox: + + def __init__(self, maxsize:int = 0, status: Symbol = k.active): + self.queue: asyncio.Queue = asyncio.Queue(maxsize) + self.status: Symbol = status async def loop(q: asyncio.Queue, bhv: Behavior) -> None: while True: @@ -18,18 +25,16 @@ async def loop(q: asyncio.Queue, bhv: Behavior) -> None: bhv = await bhv(msg) or bhv q.task_done() -def create(bhv: Behavior, maxsize: int = 0, - group: Optional[asyncio.TaskGroup] = None) -> asyncio.Queue: - q: asyncio.Queue = asyncio.Queue(maxsize) - if group is None: - asyncio.create_task(loop(q, bhv)) - else: - group.create_task(loop(q, bhv)) - return q +def create(bhv: Behavior, maxsize: int = 0) -> Mailbox: + mb = Mailbox(maxsize) + asyncio.create_task(loop(mb.queue, bhv)) + return mb -async def try_send(rcvr: Mailbox, msg: Any) -> Symbol | None: - if rcvr.full(): +async def send(mb: Mailbox, msg: Any, no_wait: bool = False) -> Symbol: + if mb.status is k.retired: + return k.retired + if no_wait and mb.queue.full(): return k.full - await rcvr.put(msg) - return None + await mb.queue.put(msg) + return k.ok diff --git a/scopes/core/core.py b/scopes/core/core.py index e5532ba..bb53ba1 100644 --- a/scopes/core/core.py +++ b/scopes/core/core.py @@ -15,12 +15,11 @@ from scopes.util import keywords as k, Symbol, Symbols # keywords (domains, actions, classes, stages, data keys): k('csys env _meta', - 'connect create created data default query remove set value', + 'connect create created data default query set value', 'c00 e00 s00 op rcvr', 'initial active retired', 'address pred succ') -type Mailbox = asyncio.Queue type Categ = tuple[Symbol, Symbol] type Proc = Callable[[Message, Scope], Awaitable[Scope | None]] type Program = Callable[[Scope], Proc] @@ -33,26 +32,26 @@ type Handler = Callable[[Message, Scope], type Handlers = Handler | None | list[Handlers] type Actions = dict[Symbol, Handlers] -def neuron(scope: Scope) -> Mailbox: +def neuron(scope: Scope) -> actor.Mailbox: async def process(msg: Message) -> actor.Behavior | None: nonlocal scope scope = await scope.proc(msg, scope) or scope return process return actor.create(process) -def synapse(rcvr: Mailbox, op: SynOp = lambda x: x) -> Syn: +def synapse(rcvr: actor.Mailbox, op: SynOp = lambda x: x) -> Syn: async def syn(msg: Message) -> Symbol | None: msg_new = op(msg) if msg_new is not None: - await rcvr.put(msg_new) # return actor.send(rcvr, msg_new) + return await actor.send(rcvr, msg_new) return None return syn -env_dummy: Mailbox = asyncio.Queue() +env_dummy: actor.Mailbox = actor.Mailbox() def notify() -> Handler: async def handle(msg: Message, scope: Scope) -> None: - await scope.environ.put(msg) + await actor.send(scope.environ, msg) return handle def gen_proc(actions: Actions, default: Handlers = notify(), @@ -89,7 +88,7 @@ class Scope: proc: Proc = field(init=False) program: Program = gen_program({}) syns: list[Syn | None] = field(default_factory=list) - environ: Mailbox = env_dummy + environ: actor.Mailbox = env_dummy def __post_init__(self): self.set_proc() @@ -98,7 +97,7 @@ class Scope: self.proc = self.program(self) return self -def create_scope(env: Mailbox, cat: Categ = (k.csys, k.c00), **kw) -> Scope: +def create_scope(env: actor.Mailbox, cat: Categ = (k.csys, k.c00), **kw) -> Scope: return Scope(environ=env, categ=cat, **kw) def clone_scope(scope: Scope, **kw) -> Scope: @@ -108,7 +107,7 @@ def clone_scope(scope: Scope, **kw) -> Scope: return Scope(environ=scope.environ, value=value, categ=categ, stage=stage, program=scope.program, syns=[]) -async def create_zero(env: Mailbox) -> Mailbox: +async def create_zero(env: actor.Mailbox) -> actor.Mailbox: scope = create_scope(env, program=gen_program({}, default=std_proc())) zero = neuron(scope) await notify_created(Message(k.csys), zero, scope) @@ -127,10 +126,9 @@ def forward(no_syns_notify: bool = False) -> Handler: count = 0 for idx, syn in enumerate(scope.syns): res = await syn(msg) if syn is not None else None - if res == k.remove: - scope.syns[idx] = None - else: - count += 1 + match res: + case k.retired: scope.syns[idx] = None + case k.ok: count += 1 if count == 0 and do_notify: do_notify(msg, scope) return handle @@ -171,7 +169,7 @@ def std_proc(actions: Actions | None = None, finalize: Handler = forward(), # helpers -async def notify_created(msg: Message, new: Mailbox, scope: Scope) -> None: +async def notify_created(msg: Message, new: actor.Mailbox, scope: Scope) -> None: msg = Message(msg.domain, k.created, data=msg.data | {k.new: new}) await notify()(msg, scope) diff --git a/scopes/core/environ.py b/scopes/core/environ.py index 437b49b..ef0ea7a 100644 --- a/scopes/core/environ.py +++ b/scopes/core/environ.py @@ -7,7 +7,7 @@ from scopes.core import actor from scopes.core.message import Message from scopes.util import keywords as k -def create_env(meta_env: core.Mailbox) -> core.Mailbox: +def create_env(meta_env: actor.Mailbox) -> actor.Mailbox: scope = core.create_scope(meta_env, (k.env, k.c00), program=core.gen_program({}, default=env_proc())) return core.neuron(scope) diff --git a/tests/test_core.py b/tests/test_core.py index 291eb0e..697cd53 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -37,15 +37,15 @@ def handle_alt(coll): @pytest.mark.anyio async def test_actor(collector): - async with asyncio.TaskGroup() as group: - q1 = actor.create(handle_basic(collector), group) - q2 = actor.create(handle_basic(collector), group) - msgs = [Message(k.test, k.hello), k.show, k.switch, k.show, "actor 1: Goodbye"] - await asyncio.gather(*[q1.put(m) for m in msgs]) - await q2.put("actor 2") - await q2.put(k.show) - await q1.put(k.quit) - await q2.put(k.quit) + mb1 = actor.create(handle_basic(collector)) + mb2 = actor.create(handle_basic(collector)) + msgs = [Message(k.test, k.hello), k.show, k.switch, k.show, "actor 1: Goodbye"] + await asyncio.gather(*[actor.send(mb1, m) for m in msgs]) + await actor.send(mb2, "actor 2") + await actor.send(mb2, k.show) + await actor.send(mb2, k.quit) + await actor.send(mb2, k.quit) + await asyncio.sleep(0.01) #print(collector) assert len(collector) == 6 assert collector[-1] == 'actor 1: Goodbye' @@ -69,9 +69,9 @@ async def meta_env(collector): async def test_core(meta_env, collector): env = environ.create_env(meta_env) zero = await core.create_zero(env) - await env.put(Message(k.env, k.hello)) - await zero.put(Message(k.csys, k.hello)) - await zero.put(Message(k.csys, k.create, + await actor.send(env, Message(k.env, k.hello)) + await actor.send(zero, Message(k.csys, k.hello)) + await actor.send(zero, Message(k.csys, k.create, data={k.connect: k.pred, k.address: (1, 1)})) await asyncio.sleep(0.01) assert len(collector) == 4