core.actor: Mailbox class (with status), send() function

scopes/core/actor.py

@@ -4,22 +4,37 @@ import asyncio
 from collections.abc import Awaitable, Callable
 from typing import Any, Optional
 
-from scopes.util import keywords as k
+from scopes.util import keywords as k, Symbol
+
+# keywords:
+k('active full ok quit retired suspended')
 
 type Behavior = Callable[[Any], Awaitable[Behavior | None]]
 
+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:
         msg = await q.get()
         if msg is k.quit: 
             break
         bhv = await bhv(msg) or bhv
+        q.task_done()
 
-def create(bhv: Behavior, group: Optional[asyncio.TaskGroup] = None) -> asyncio.Queue:
-    q: asyncio.Queue = asyncio.Queue()
-    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 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 mb.queue.put(msg)
+    return k.ok
 

scopes/core/core.py

@@ -14,18 +14,17 @@ from scopes import util
 from scopes.util import keywords as k, Symbol, Symbols
 
 # keywords (domains, actions, classes, stages, data keys):
-k('csys ctl env',
+k('csys env _meta',
   'connect create created data default query set value',
-  'c00 e00 s00 rcvr op',
+  '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]
 type ProgramSpec = dict[tuple[Symbol, Symbol], Proc]
-type Syn = Callable[[Message], Awaitable[None]]
+type Syn = Callable[[Message], Awaitable[Symbol | None]]
 type SynOp = Callable[[Message], Message | None]
 
 type Handler = Callable[[Message, Scope], 
@@ -33,27 +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:
-    async def syn(msg: Message) -> None:
-        # op = update_op(msg) (=> return?)
-        # rv = check_return(msg) => return rv
+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 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,8 +87,8 @@ class Scope:
     stage: Symbol = k.initial
     proc: Proc = field(init=False)
     program: Program = gen_program({})
-    syns: list[Syn] = field(default_factory=list)
-    environ: Mailbox = env_dummy
+    syns: list[Syn | None] = field(default_factory=list)
+    environ: actor.Mailbox = env_dummy
 
     def __post_init__(self):
         self.set_proc()
@@ -99,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:
@@ -109,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)
@@ -117,10 +115,22 @@ async def create_zero(env: Mailbox) -> Mailbox:
 
 # predefined action handlers and Action sets
 
-def forward() -> Handler:
+def noop(no_forward: bool = False) -> Handler:
+    async def handle(msg: Message, scope: Scope) -> Message | None:
+        return None if no_forward else msg
+    return handle
+
+def forward(no_syns_notify: bool = False) -> Handler:
+    do_notify = notify() if no_syns_notify else None
     async def handle(msg: Message, scope: Scope) -> None:
-        for syn in scope.syns:
-            await syn(msg)
+        count = 0
+        for idx, syn in enumerate(scope.syns):
+            res = await syn(msg) if syn is not None else None
+            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
 
 def value_add() -> Handler:
@@ -159,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 scope.environ.put(msg)
+    await notify()(msg, scope)
 

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)

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