Compare commits

..

No commits in common. "87c2a68ef6eddb2c210ec7e5c3aef994a39bbee7" and "bc12966b703644c4f2fcb93848ca84553af9af07" have entirely different histories.

4 changed files with 42 additions and 67 deletions

View file

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

View file

@ -14,17 +14,18 @@ from scopes import util
from scopes.util import keywords as k, Symbol, Symbols from scopes.util import keywords as k, Symbol, Symbols
# keywords (domains, actions, classes, stages, data keys): # keywords (domains, actions, classes, stages, data keys):
k('csys env _meta', k('csys ctl env',
'connect create created data default query set value', 'connect create created data default query set value',
'c00 e00 s00 op rcvr', 'c00 e00 s00 rcvr op',
'initial active retired', 'initial active retired',
'address pred succ') 'address pred succ')
type Mailbox = asyncio.Queue
type Categ = tuple[Symbol, Symbol] type Categ = tuple[Symbol, Symbol]
type Proc = Callable[[Message, Scope], Awaitable[Scope | None]] type Proc = Callable[[Message, Scope], Awaitable[Scope | None]]
type Program = Callable[[Scope], Proc] type Program = Callable[[Scope], Proc]
type ProgramSpec = dict[tuple[Symbol, Symbol], Proc] type ProgramSpec = dict[tuple[Symbol, Symbol], Proc]
type Syn = Callable[[Message], Awaitable[Symbol | None]] type Syn = Callable[[Message], Awaitable[None]]
type SynOp = Callable[[Message], Message | None] type SynOp = Callable[[Message], Message | None]
type Handler = Callable[[Message, Scope], type Handler = Callable[[Message, Scope],
@ -32,26 +33,27 @@ type Handler = Callable[[Message, Scope],
type Handlers = Handler | None | list[Handlers] type Handlers = Handler | None | list[Handlers]
type Actions = dict[Symbol, Handlers] type Actions = dict[Symbol, Handlers]
def neuron(scope: Scope) -> actor.Mailbox: def neuron(scope: Scope) -> Mailbox:
async def process(msg: Message) -> actor.Behavior | None: async def process(msg: Message) -> actor.Behavior | None:
nonlocal scope nonlocal scope
scope = await scope.proc(msg, scope) or scope scope = await scope.proc(msg, scope) or scope
return process return process
return actor.create(process) return actor.create(process)
def synapse(rcvr: actor.Mailbox, op: SynOp = lambda x: x) -> Syn: def synapse(rcvr: Mailbox, op: SynOp = lambda x: x) -> Syn:
async def syn(msg: Message) -> Symbol | None: async def syn(msg: Message) -> None:
# op = update_op(msg) (=> return?)
# rv = check_return(msg) => return rv
msg_new = op(msg) msg_new = op(msg)
if msg_new is not None: if msg_new is not None:
return await actor.send(rcvr, msg_new) await rcvr.put(msg_new)
return None
return syn return syn
env_dummy: actor.Mailbox = actor.Mailbox() env_dummy: Mailbox = asyncio.Queue()
def notify() -> Handler: def notify() -> Handler:
async def handle(msg: Message, scope: Scope) -> None: async def handle(msg: Message, scope: Scope) -> None:
await actor.send(scope.environ, msg) await scope.environ.put(msg)
return handle return handle
def gen_proc(actions: Actions, default: Handlers = notify(), def gen_proc(actions: Actions, default: Handlers = notify(),
@ -87,8 +89,8 @@ class Scope:
stage: Symbol = k.initial stage: Symbol = k.initial
proc: Proc = field(init=False) proc: Proc = field(init=False)
program: Program = gen_program({}) program: Program = gen_program({})
syns: list[Syn | None] = field(default_factory=list) syns: list[Syn] = field(default_factory=list)
environ: actor.Mailbox = env_dummy environ: Mailbox = env_dummy
def __post_init__(self): def __post_init__(self):
self.set_proc() self.set_proc()
@ -97,7 +99,7 @@ class Scope:
self.proc = self.program(self) self.proc = self.program(self)
return self return self
def create_scope(env: actor.Mailbox, cat: Categ = (k.csys, k.c00), **kw) -> Scope: def create_scope(env: Mailbox, cat: Categ = (k.csys, k.c00), **kw) -> Scope:
return Scope(environ=env, categ=cat, **kw) return Scope(environ=env, categ=cat, **kw)
def clone_scope(scope: Scope, **kw) -> Scope: def clone_scope(scope: Scope, **kw) -> Scope:
@ -107,7 +109,7 @@ def clone_scope(scope: Scope, **kw) -> Scope:
return Scope(environ=scope.environ, value=value, categ=categ, stage=stage, return Scope(environ=scope.environ, value=value, categ=categ, stage=stage,
program=scope.program, syns=[]) program=scope.program, syns=[])
async def create_zero(env: actor.Mailbox) -> actor.Mailbox: async def create_zero(env: Mailbox) -> Mailbox:
scope = create_scope(env, program=gen_program({}, default=std_proc())) scope = create_scope(env, program=gen_program({}, default=std_proc()))
zero = neuron(scope) zero = neuron(scope)
await notify_created(Message(k.csys), zero, scope) await notify_created(Message(k.csys), zero, scope)
@ -115,22 +117,10 @@ async def create_zero(env: actor.Mailbox) -> actor.Mailbox:
# predefined action handlers and Action sets # predefined action handlers and Action sets
def noop(no_forward: bool = False) -> Handler: def forward() -> 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: async def handle(msg: Message, scope: Scope) -> None:
count = 0 for syn in scope.syns:
for idx, syn in enumerate(scope.syns): await syn(msg)
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 return handle
def value_add() -> Handler: def value_add() -> Handler:
@ -169,7 +159,7 @@ def std_proc(actions: Actions | None = None, finalize: Handler = forward(),
# helpers # helpers
async def notify_created(msg: Message, new: actor.Mailbox, scope: Scope) -> None: async def notify_created(msg: Message, new: Mailbox, scope: Scope) -> None:
msg = Message(msg.domain, k.created, data=msg.data | {k.new: new}) msg = Message(msg.domain, k.created, data=msg.data | {k.new: new})
await notify()(msg, scope) await scope.environ.put(msg)

View file

@ -7,7 +7,7 @@ from scopes.core import actor
from scopes.core.message import Message from scopes.core.message import Message
from scopes.util import keywords as k from scopes.util import keywords as k
def create_env(meta_env: actor.Mailbox) -> actor.Mailbox: def create_env(meta_env: core.Mailbox) -> core.Mailbox:
scope = core.create_scope(meta_env, (k.env, k.c00), scope = core.create_scope(meta_env, (k.env, k.c00),
program=core.gen_program({}, default=env_proc())) program=core.gen_program({}, default=env_proc()))
return core.neuron(scope) return core.neuron(scope)

View file

@ -37,15 +37,15 @@ def handle_alt(coll):
@pytest.mark.anyio @pytest.mark.anyio
async def test_actor(collector): async def test_actor(collector):
mb1 = actor.create(handle_basic(collector)) async with asyncio.TaskGroup() as group:
mb2 = actor.create(handle_basic(collector)) 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"] 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 asyncio.gather(*[q1.put(m) for m in msgs])
await actor.send(mb2, "actor 2") await q2.put("actor 2")
await actor.send(mb2, k.show) await q2.put(k.show)
await actor.send(mb2, k.quit) await q1.put(k.quit)
await actor.send(mb2, k.quit) await q2.put(k.quit)
await asyncio.sleep(0.01)
#print(collector) #print(collector)
assert len(collector) == 6 assert len(collector) == 6
assert collector[-1] == 'actor 1: Goodbye' assert collector[-1] == 'actor 1: Goodbye'
@ -69,9 +69,9 @@ async def meta_env(collector):
async def test_core(meta_env, collector): async def test_core(meta_env, collector):
env = environ.create_env(meta_env) env = environ.create_env(meta_env)
zero = await core.create_zero(env) zero = await core.create_zero(env)
await actor.send(env, Message(k.env, k.hello)) await env.put(Message(k.env, k.hello))
await actor.send(zero, Message(k.csys, k.hello)) await zero.put(Message(k.csys, k.hello))
await actor.send(zero, Message(k.csys, k.create, await zero.put(Message(k.csys, k.create,
data={k.connect: k.pred, k.address: (1, 1)})) data={k.connect: k.pred, k.address: (1, 1)}))
await asyncio.sleep(0.01) await asyncio.sleep(0.01)
assert len(collector) == 4 assert len(collector) == 4