Syndicated Actors for Conversational Concurrency

Syndicated Actors for Python 3

2 September 2021
Tony Garnock-Jones
#syndicate #implementation #system-layer
permalink

Previously, the mini-syndicate package for Python 3 implemented an older version of the Syndicate network protocol.

A couple of weeks ago, I dusted it off, updated it to the new capability-oriented Syndicate protocol, and fleshed out its nascent Syndicated Actor Model code to be a full implementation of the model, including capabilities, object references, actors, facets, assertions and so on.

The new implementation makes heavy use of Python decorators to work around Python’s limited lambda forms and its poor support for syntactic extensibility. The result is surprisingly not terrible!

The revised codebase is different enough to the previous one that it deserves its own new git repository:

git clone https://git.syndicate-lang.org/syndicate-lang/syndicate-py

It’s also available on pypi.org, as package syndicate-py.

Updated Preserves for Python

As part of the work, I updated the Python Preserves implementation (for both python 2 and python 3) to include the text-based Preserves syntax as well as the binary syntax, plus implementations of Preserves Schema and Preserves Path. Version 0.7.0 or newer of the preserves package on pypi.org has the new features.

A little “chat” demo

(Based on chat.py in the repository.)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

import sys
import asyncio
import random
import syndicate
from syndicate import patterns as P, actor, dataspace
from syndicate.schema import simpleChatProtocol, sturdy

Present = simpleChatProtocol.Present
Says = simpleChatProtocol.Says

ds_capability = syndicate.parse('<ref "syndicate" [] #[acowDB2/oI+6aSEC3YIxGg==]>')

@actor.run_system()
def main(turn):
    root_facet = turn._facet

    @syndicate.relay.connect(turn, '<tcp "localhost" 8001>', sturdy.SturdyRef.decode(ds_capability))
    def on_connected(turn, ds):
        me = 'user_' + str(random.randint(10, 1000))

        turn.publish(ds, Present(me))

        @dataspace.during(turn, ds, P.rec('Present', P.CAPTURE), inert_ok=True)
        def on_presence(turn, who):
            print('%s joined' % (who,))
            turn.on_stop(lambda turn: print('%s left' % (who,)))

        @dataspace.on_message(turn, ds, P.rec('Says', P.CAPTURE, P.CAPTURE))
        def on_says(turn, who, what):
            print('%s says %r' % (who, what))

        @turn.linked_task()
        async def accept_input(f):
            reader = asyncio.StreamReader()
            await actor.find_loop().connect_read_pipe(lambda: asyncio.StreamReaderProtocol(reader), sys.stdin)
            while line := (await reader.readline()).decode('utf-8'):
                actor.Turn.external(f, lambda turn: turn.send(ds, Says(me, line.strip())))
            actor.Turn.external(f, lambda turn: turn.stop(root_facet))