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

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
 

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)
 

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