Welcome to aiomas’ documentation!¶

The agent layer¶

This section describes the agent layer and gives you enough information to implement your own distributed system without going too much into detail. For that, you should also read the section about the RPC layer.

Overview¶

You can think of agents as small, independent programs running in parallel. Each agent waits for input (e.g., incoming network messages), processes the input and creates, based on its internal state and the input, some output (like outgoing network messages).

You can also imagine them as being like normal objects that call other object’s methods. But instead of calling these methods directly, they do remote procedure calls (RPC) via a network connection.

In theory, that means that every agent has a little server with an event loop that waits for incoming messages and dispatches them to the corresponding method calls.

Using this model, you would quickly run out of resources with hundreds or thousands of interconnected agents. For this reason, agents are clustered in containers. A container provides the network server and event loop which all agents within the container share.

Agents are uniquely identified by the container’s address and an ID (which is unique within a container), for example: tcp://localhost:5555/42.

The following image illustrates this: If Agent C wants to send a message to Agent A, its container connects to A’s container. Agent C can now send a message to Agent A. If Agent C now wanted to send a message to Agent B, it would simply reuse the same connection:

As you can see in the figure above, containers also have a clock, but you can ignore that for the moment. We’ll come back to it later.

Components of a distributed system in aiomas¶

1. Agent: You implement your business logic in subclasses of aiomas.Agent. Agents can be reactive or proactive.

   Reactive agents only react to incoming messages. That means, they simply expose some methods that other agents can call.

   Proactive agents actively perform one ore more tasks, i.e., calling other agent’s methods.

   An agent can be both, proactive and reactive (that just means that your agent class exposes some methods and has one or more tasks running).

2. Container: All agents live in a container. The agent container implements everything networking related (e.g., a shared RPC server) so that the agent base class can be as light-weight as possible. It also defines the codec used for message (de)serialization and provides a clock for agents.

3. Codec: Codecs define how messages to other agents get serialized to byte strings that can be sent over the network. The base codecs can only serialize the most common object types (like numbers, strings, lists or dicts) but you can extend them with serializers for custom object types.

   The Codecs section explain all this in detail.

4. Clock: Every container provides a clock for agents. Clocks are important for operations with a timeout (like sleep()). The default clock is a real-time clock synchronized to your system’s time.

   However, if you want to integrate your MAS with a simulation, you may want to let the time pass faster then real-time (in order to decrease the duration of your simulation). For that use case, aiomas provides a clock that can be synchronized with external sources.

   All clocks provide functions to get the current time, sleep for some time or execute a task after a given timeout. If you use these function instead of the once asyncio provides, you can easily switch between different kinds of clocks. The Clocks section provides more details and examples.

Don’t worry if you feel a bit confused now. I’ll explore all of this with small, intuitive examples.

>>> import aiomas
>>>
>>> class HelloWorld(aiomas.Agent):
...     def __init__(self, container, name):
...         # We must pass a ref. to the container to "aiomas.Agent":
...         super().__init__(container)
...         self.name = name  # Our agent's name
...
...     async def run(self):
...         # This method defines the task that our agent will perform.
...         # It's usually called "run()" but you can name it as wou want.
...         print(self.name, 'says:')
...         clock = self.container.clock
...         for i in range(3):
...             await clock.sleep(0.1)
...             print('Hello, World!')

>>> # Containers need to be started via a factory function:
>>> container = aiomas.Container.create(('localhost', 5555))
>>>
>>> # Now we can instantiate an agent an start its task:
>>> agent = HelloWorld(container, 'Monty')
>>> aiomas.run(until=agent.run())
Monty says:
Hello, World!
Hello, World!
Hello, World!
>>> container.shutdown()  # Close all connections and shutdown the server

>>> import asyncio
>>> import aiomas
>>>
>>> class Callee(aiomas.Agent):
...     # This agent does not need to override "__init__()".
...
...     # "expose"d methods can be called by other agents:
...     @aiomas.expose
...     def spam(self, times):
...         """Return a lot of spam."""
...         return 'spam' * times
>>>
>>>
>>> class Caller(aiomas.Agent):
...
...     async def run(self, callee_addr):
...         print(self, 'connecting to', callee_addr)
...         # Ask the container to make a connection to the other agent:
...         callee = await self.container.connect(callee_addr)
...         print(self, 'connected to', callee)
...         # "callee" is a proxy to the other agent.  It allows us to call
...         # the exposed methods:
...         result = await callee.spam(3)
...         print(self, 'got', result)
>>>
>>>
>>> container = aiomas.Container.create(('localhost', 5555))
>>> callee = Callee(container)
>>> caller = Caller(container)
>>> aiomas.run(until=caller.run(callee.addr))
Caller('tcp://localhost:5555/1') connecting to tcp://localhost:5555/0
Caller('tcp://localhost:5555/1') connected to CalleeProxy('tcp://localhost:5555/0')
Caller('tcp://localhost:5555/1') got spamspamspam
>>> container.shutdown()

The RPC layer¶

Remote procedure calls let you, as the name suggest, call functions or methods of remote objects via a network connection (nearly) like you would call local functions. This often leads to more readable code compared to using the lower level channels.

>>> import aiomas
>>>
>>>
>>> class MathServer:
...     # The "Service" creates a router for each instance of "MathServer":
...     router = aiomas.rpc.Service()
...
...     # Exposed methods can be called by clients:
...     @aiomas.expose
...     def add(self, a, b):
...         return a + b
...
>>>
>>> async def client():
...     """Client coroutine: Call the server's "add()" method."""
...     # Connect to the RPC server and get an "RpcClient":
...     rpc_con = await aiomas.rpc.open_connection(('localhost', 5555))
...     # "remote" is a Proxy to the remote service.
...     # We cann call its "add()" method:
...     rep = await rpc_con.remote.add(3, 4)
...     print('What’s 3 + 4?', rep)
...     await rpc_con.close()
>>>
>>> # Start the RPC server with an instance of the "MathServer" service:
>>> server = aiomas.run(aiomas.rpc.start_server(('localhost', 5555), MathServer()))
>>>
>>> aiomas.run(client())
What’s 3 + 4? 7
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> @aiomas.expose
... def add(a, b):
...     return a + b
...
>>> math_service = aiomas.rpc.ServiceDict({
...     'add': add,
... })
>>>
>>> # Start the RPC server with the math service:
>>> server = aiomas.run(aiomas.rpc.start_server(('localhost', 5555), math_service))
>>>
>>> # The client stays the same as in our last example:
>>> aiomas.run(client())
What’s 3 + 4? 7
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> @aiomas.expose
... def add(a, b):
...     return a + b
...
>>> # A Sub-service for addition
>>> adding_service = aiomas.rpc.ServiceDict({
...     'add': add,
... })
>>>
>>> # A Sub-service for subtraction
>>> class SubService:
...     router = aiomas.rpc.Service()
...
...     @aiomas.expose
...     def sub(self, a, b):
...         return a - b
...
>>> # Service dict with two sub-services:
>>> root_service = aiomas.rpc.ServiceDict({
...     'addition': adding_service,   # Service dict
...     'subtraction': SubService(),  # Instance(!) of service class
... })
>>>
>>> async def client():
...     rpc_con = await aiomas.rpc.open_connection(('localhost', 5555))
...     # Call the addition service:
...     rep = await rpc_con.remote.addition.add(3, 4)
...     print('What’s 3 + 4?', rep)
...     # Call the subtraction service:
...     rep = await rpc_con.remote.subtraction.sub(4, 3)
...     print('What’s 4 - 3?', rep)
...     await rpc_con.close()
>>>
>>> server = aiomas.run(aiomas.rpc.start_server(('localhost', 5555), root_service))
>>>
>>> aiomas.run(client())
What’s 3 + 4? 7
What’s 4 - 3? 1
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> @aiomas.expose
... def add(a, b):
...     return a + b
...
>>> # A Sub-service for addition
>>> adding_service = aiomas.rpc.ServiceDict({
...     'add': add,
... })
>>>
>>> # A Sub-service for subtraction
>>> class SubService:
...     router = aiomas.rpc.Service()
...
...     @aiomas.expose
...     def sub(self, a, b):
...         return a - b
...
>>> class RootService:
...     # You first have to declare that instances of this class will have
...     # the following sub-services:
...     router = aiomas.rpc.Service(['addition', 'subtraction'])
...
...     def __init__(self):
...         # For each(!) instance, you have to add instances of the
...         # declared sub-services:
...         self.addition = adding_service
...         self.subtraction = SubService()
>>>
>>>
>>> server = aiomas.run(aiomas.rpc.start_server(('localhost', 5555), RootService()))
>>>
>>> # The client remains the same
>>> aiomas.run(client())
What’s 3 + 4? 7
What’s 4 - 3? 1
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> @aiomas.expose
... def add(a, b):
...     return a + b
...
>>> # A Sub-service for addition
>>> adding_service = aiomas.rpc.ServiceDict({
...     'add': add,
... })
>>>
>>> # A Sub-service for subtraction
>>> class SubService:
...     router = aiomas.rpc.Service()
...
...     @aiomas.expose
...     def sub(self, a, b):
...         return a - b
...
>>> class RootService:
...     # In contrast to the last example, we don't declare any sub-services:
...     router = aiomas.rpc.Service()
...
...     def __init__(self):
...         # Add a sub-services via the router's "add()" method:
...         self.addition = adding_service
...         self.router.add('addition')
...
...         # Add a sub-service via the router's "set_sub_router()" method:
...         self.subtraction = SubService()
...         self.router.set_sub_router(self.subtraction.router, 'subtraction')
>>>
>>>
>>> server = aiomas.run(aiomas.rpc.start_server(('localhost', 5555), RootService()))
>>>
>>> # The client remains the same
>>> aiomas.run(client())
What’s 3 + 4? 7
What’s 4 - 3? 1
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> import aiomas
>>>
>>>
>>> class Client:
...     # The client needs to be marked as RPC service:
...     router = aiomas.rpc.Service()
...
...     def __init__(self, name):
...         self.name = name
...
...     async def run(self):
...         # When we open a connection, we need to pass the service instance
...         # ("self" in this case) so that a background task for it can be
...         # started:
...         rpc_con = await aiomas.rpc.open_connection(('localhost', 5555),
...                                                    rpc_service=self)
...
...         # We can now pass the service to the server when we call one of its
...         # methods:
...         rep = await rpc_con.remote.server_method(self)
...         print('Server reply:', rep)
...
...         await rpc_con.close()
...
...     # This method is exposed to the server:
...     @aiomas.expose
...     def get_client_name(self):
...         return self.name
>>>
>>>
>>> class Server:
...     router = aiomas.rpc.Service()
...
...     @aiomas.expose
...     async def server_method(self, client_proxy):
...         # When a client passes a reference to its service, we'll receive it as
...         # a proxy object which we can use to call a client method:
...         client_name = await client_proxy.get_client_name()
...         return 'Client name is "%s"' % client_name
>>>
>>>
>>> server = aiomas.run(aiomas.rpc.start_server(('localhost', 5555), Server()))
>>>
>>> aiomas.run(Client('Monty').run())
Server reply: Client name is "Monty"
>>>
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> @aiomas.expose
... def fail_badly():
...     raise ValueError('"spam" is not a number')
>>>
>>> service = aiomas.rpc.ServiceDict({'fail_badly': fail_badly})
>>>
>>> async def client():
...     rpc_con = await aiomas.rpc.open_connection(('127.0.0.1', 5555))
...     try:
...         await rpc_con.remote.fail_badly()
...     except aiomas.RemoteException as exc:
...         print('Origin:', exc.origin)
...         print('Traceback:', exc.remote_traceback)
>>>
>>> server = aiomas.run(aiomas.rpc.start_server(('127.0.0.1', 5555), service))
>>>
>>> aiomas.run(client())
Origin: ('127.0.0.1', 5555)
Traceback: Traceback (most recent call last):
  ...
ValueError: "spam" is not a number

>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> service = aiomas.rpc.ServiceDict({})
>>>
>>> async def client():
...     rpc_con = await aiomas.rpc.open_connection(('127.0.0.1', 5555))
...     await rpc_con.close()
>>>
>>> def client_connected_cb(rpc_client):
...     print('Client connected:', rpc_client)
>>>
>>> server = aiomas.run(aiomas.rpc.start_server(('127.0.0.1', 5555), service,
...                                             client_connected_cb))
>>>
>>> aiomas.run(client())
Client connected: <aiomas.rpc.RpcClient object at 0x...>
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> import aiomas
>>>
>>>
>>> async def client():
...     rpc_con = await aiomas.rpc.open_connection(('localhost', 5555))
...     # The server will close the connection when we make the following call:
...     try:
...         await rpc_con.remote.close_connection()
...     except ConnectionResetError:
...         print('Connection lost :(')
>>>
>>>
>>> class Server:
...     router = aiomas.rpc.Service()
...
...     def __init__(self):
...         self.clients = []
...
...     def client_connected(self, client):
...         """*Client connected cb.* that adds new clients to ``self.clients``"""
...         self.clients.append(client)
...
...     @aiomas.expose
...     async def close_connection(self):
...         """Close all open connections and remove them from ``self.clients``."""
...         while self.clients:
...             client = self.clients.pop()
...             await client.close()
>>>
>>> server_service = Server()
>>> server = aiomas.run(aiomas.rpc.start_server(('localhost', 5555),
...                                             server_service,
...                                             server_service.client_connected))
>>>
>>> aiomas.run(client())
Connection lost :(
>>>
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> import aiomas
>>>
>>>
>>> async def client():
...     rpc_con = await aiomas.rpc.open_connection(('localhost', 5555))
...     await rpc_con.close()
>>>
>>>
>>> class Server:
...     router = aiomas.rpc.Service()
...
...     def __init__(self):
...         self.clients = []
...
...     def client_connected(self, client):
...         # Register a callback that removes the client from our list
...         # when it disconnects:
...         def remove_client(exc):
...             print('Client disconnected :(')
...             self.clients.remove(client)
...
...         client.on_connection_reset(remove_client)
...         print('Client connected :)')
...         self.clients.append(client)
>>>
>>> server_service = Server()
>>> server = aiomas.run(aiomas.rpc.start_server(('localhost', 5555),
...                                             server_service,
...                                             server_service.client_connected))
>>>
>>> aiomas.run(client())
Client connected :)
Client disconnected :(
>>>
>>> server.close()
>>> aiomas.run(server.wait_closed())

The channel layer¶

The channel layer is aiomas’ lowest layer of abstraction. It lets you send and receive complete messages. In contrast to asyncio’s built-in stream protocol which just sends byte strings, messages are JSON-encoded [*] data (which is a lot more convenient).

Here is a minimal example that shows how the Channel can be used:

>>> import aiomas
>>>
>>>
>>> async def client():
...     """Client coroutine: Send a greeting to the server and wait for a
...     reply."""
...     channel = await aiomas.channel.open_connection(('localhost', 5555))
...     rep = await channel.send('ohai')
...     print(rep)
...     await channel.close()
>>>
>>>
>>> async def handle_client(channel):
...     """Handle a client connection."""
...     req = await channel.recv()
...     print(req.content)
...     await req.reply('cya')
...     await channel.close()
>>>
>>>
>>> server = aiomas.run(aiomas.channel.start_server(('localhost', 5555), handle_client))
>>> aiomas.run(client())
ohai
cya
>>> server.close()
>>> aiomas.run(server.wait_closed())

A communication channel has two sides: The client side is created and returned by open_connection(). For each client connection, the server creates a Channel instance and starts a new background task of the client connected callback (client_connected_cb) to which it passes that channel instance.

Both, the client and server side, can send and receive messages. In the example above, the client starts to send a request and the server side waits for incoming requests. A request has a content attribute which holds the actual message. To send a reply, you can either use Request.reply() or Request.fail(). Channel.send() and Request.reply() take any data that the channel’s codec can serialize (e.g., strings, numbers, lists, dicts, ...). Request.fail() takes an exception instance which is raised at the requesting side as RemoteException, as the following example demonstrates:

>>> import aiomas
>>>
>>>
>>> async def client():
...     """Client coroutine: Send a greeting to the server and wait for a
...     reply."""
...     channel = await aiomas.channel.open_connection(('localhost', 5555))
...     try:
...         rep = await channel.send('ohai')
...         print(rep)
...     except aiomas.RemoteException as e:
...         print('Got an error:', str(e))
...     finally:
...         await channel.close()
>>>
>>>
>>> async def handle_client(channel):
...     """Handle a client connection."""
...     req = await channel.recv()
...     print(req.content)
...     await req.fail(ValueError(42))
...     await channel.close()
>>>
>>>
>>> server = aiomas.run(aiomas.channel.start_server(('127.0.0.1', 5555), handle_client))
>>> aiomas.run(client())
ohai
Got an error: Origin: ('127.0.0.1', 5555)
ValueError: 42

>>> server.close()
>>> aiomas.run(server.wait_closed())

These are the basics of the channel layer. The following sections answer some detail questions.

>>> import aiomas
>>>
>>> CODEC = aiomas.codecs.MsgPack
>>>
>>> async def client():
...     """Client coroutine: Send a greeting to the server and wait for a
...     reply."""
...     channel = await aiomas.channel.open_connection(('localhost', 5555),
...                                                    codec=CODEC)
...     rep = await channel.send('ohai')
...     print(rep)
...     await channel.close()
>>>
>>>
>>> async def handle_client(channel):
...     """Handle a client connection."""
...     req = await channel.recv()
...     print(req.content)
...     await req.reply('cya')
...     await channel.close()
>>>
>>>
>>> server = aiomas.run(aiomas.channel.start_server(('localhost', 5555), handle_client,
...                                                 codec=CODEC))
>>> aiomas.run(client())
ohai
cya
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> import aiomas
>>>
>>>
>>> class MyType:
...     """Our serializable type."""
...     def __init__(self, value):
...         self.value = value
...
...     def __repr__(self):
...         return '%s(%r)' % (self.__class__.__name__, self.value)
>>>
>>>
>>> def serialize_mytype(obj):
...     """Return a JSON serializable version "MyType" instances."""
...     return obj.value
>>>
>>>
>>> def deserialize_mytype(value):
...     """Make a "MyType" instance from *value*."""
...     return MyType(value)
>>>
>>>
>>> def mytype_serializer():
...     return (MyType, serialize_mytype, deserialize_mytype)
>>>
>>>
>>> EXTRA_SERIALIZERS = [mytype_serializer]
>>>
>>>
>>> async def client():
...     """Client coroutine: Send a greeting to the server and wait for a
...     reply."""
...     channel = await aiomas.channel.open_connection(
...         ('localhost', 5555), extra_serializers=EXTRA_SERIALIZERS)
...     rep = await channel.send(['ohai', MyType(42)])
...     print(rep)
...     await channel.close()
>>>
>>>
>>> async def handle_client(channel):
...     """Handle a client connection."""
...     req = await channel.recv()
...     print(req.content)
...     await req.reply(MyType('cya'))
...     await channel.close()
>>>
>>>
>>> server = aiomas.run(aiomas.channel.start_server(('localhost', 5555), handle_client,
...                                                 extra_serializers=EXTRA_SERIALIZERS))
>>> aiomas.run(client())
['ohai', MyType(42)]
MyType('cya')
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> import aiomas
>>>
>>>
>>> @aiomas.codecs.serializable
... class MyType:
...     """Our serializable type."""
...     def __init__(self, value):
...         self.value = value
>>>
>>>
>>> EXTRA_SERIALIZERS = [MyType.__serializer__]
>>>
>>>
>>> async def client():
...     """Client coroutine: Send a greeting to the server and wait for a
...     reply."""
...     channel = await aiomas.channel.open_connection(
...         ('localhost', 5555), extra_serializers=EXTRA_SERIALIZERS)
...     rep = await channel.send(['ohai', MyType(42)])
...     print(rep)
...     await channel.close()
>>>
>>>
>>> async def handle_client(channel):
...     """Handle a client connection."""
...     req = await channel.recv()
...     print(req.content)
...     await req.reply(MyType('cya'))
...     await channel.close()
>>>
>>>
>>> server = aiomas.run(aiomas.channel.start_server(('localhost', 5555), handle_client,
...                                                 extra_serializers=EXTRA_SERIALIZERS))
>>> aiomas.run(client())
['ohai', MyType(value=42)]
MyType(value='cya')
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> import errno
>>> import random
>>>
>>> max_tries = 100
>>> port_range = (49152, 65536)
>>>
>>> async def random_server(host, port_range, max_tries):
...     for i in range(max_tries):
...         try:
...             port = random.randrange(*port_range)
...             server = await aiomas.channel.start_server(
...                (host, port), handle_client)
...         except OSError as oe:
...             if oe.errno != errno.EADDRINUSE:
...                 # Re-raise if not errno 48 ("address already in use")
...                 raise
...         else:
...             return server, port
...     raise RuntimeError('Could not bind server to a random port.')
>>>
>>> server, port = aiomas.run(random_server('localhost', port_range, max_tries))
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> import asyncio
>>> import aiomas
>>>
>>>
>>> async def client():
...     """Client coroutine: Send a greeting to the server and wait for a
...     reply."""
...     # Try to connect for 1s:
...     channel = await aiomas.channel.open_connection(('localhost', 5555),
...                                                    timeout=1)
...     rep = await channel.send('ohai')
...     print(rep)
...     await channel.close()
>>>
>>>
>>> async def handle_client(channel):
...     """Handle a client connection."""
...     req = await channel.recv()
...     print(req.content)
...     await req.reply('cya')
...     await channel.close()
>>>
>>>
>>> # Start the client in background, ...
>>> t_client = asyncio.async(client())
>>> # wait 0.5 seconds, ...
>>> aiomas.run(asyncio.sleep(0.5))
>>> # and finally start the server:
>>> server = aiomas.run(aiomas.channel.start_server(('localhost', 5555), handle_client))
>>> aiomas.run(t_client)
ohai
cya
>>> server.close()
>>> aiomas.run(server.wait_closed())

How exactly do messages look like?¶

This section explains how aiomas messages look and how they are constructed. You can easily implement this protocol in other languages, too, and write programs that can communicate with aiomas.

Network messages consists of a four bytes long header and a payload of arbitrary length. The header is an unsigned integer (uint32) in network byte order (big-endian) and stores the number of bytes in the payload. The payload itself is an encoded [†] list containing the message type, a message ID and the actual content:

[†]	Depending on the codec you use, the payload may be a UTF-8 encoded JSON string (json.dumps().encode('utf-8')) (this is the default), a MsgPack list (msgpack.packb()), or whatever else the codec produces.

Messages send between two peers must follow the request-reply pattern. That means, every request that one peer makes must be responded to by the other peer. Request use the message type 0, replies use 1 for success or 2 to indicate a failure. The message ID is an integer that is unique for every request that a network socket makes. Replies (no matter if successful or failed) need to use the message ID of the corresponding request.

On the channel layer, the content of a request can be anything. On the RPC level, it a three-tuple (function_path, args, kwargs), e.g.:

[function, [arg0, arg1, ...], {kwarg0: val0, kwarg1: val1}]

Thereby, function is always a string containing the name of an exposed functions; if you use nested services, sub-services and the function names are separated by slashes (/) as in URLs. The type of the arguments and keyword arguments may vary depending on the function.

The content types of replies are the same for both, the channel layer and the RPC layer. Normal (successful) replies can be anything. The content of failure replies are strings with the error message and/or a stack trace.

Note

If the JSON codec is used, aiomas messages are compatible with simpy.io (and therewith with the co-simulation framework mosaik, too).

Codecs for message serialization¶

Codecs are used to convert the objects that you are going to send over the network to bytes and the bytes that you received back to the original objects. This is called serialization and deserialization.

A codec specifies, how the text representation of a certain object looks like. It can also recreate the object based on its text representation.

For example, the JSON encoded representation of the list ['spam', 3.14] would be b'["spam",3.14]'.

Many different codecs exists. Some of the most widely used ones are JSON, XML or MsgPack. They mainly differ in their:

• verbosity or compactness: How many bytes are needed to encode an object?
• performance: How fast can they encode and decode objects?
• readability: Can the result easily be read by humans?
• availability on different platforms: For which programming languages do libraries or bindings exist?
• security: Is it possible to decode bytes to arbitrary objects?

Which codec is the best very much depends on your specific requirements. An evaluation of different codecs and serialization formats is beyond the scope of this document, though.

$ pip install -U aiomas[mp]  # Install aiomas with MsgPack
$ # or
$ pip install -U aiomas msgpack-python

$ pip install -U aiomas[mpb]  # Install aiomas with MsgPack-Blosc
$ # or
$ pip install -U aiomas msgpack-python blosc

import aiomas
import numpy as np

def get_np_serializer():
   """Return a tuple *(type, serialize(), deserialize())* for NumPy arrays
   for usage with an :class:`aiomas.codecs.MsgPack` codec.

   """
   return np.ndarray, _serialize_ndarray, _deserialize_ndarray


def _serialize_ndarray(obj):
   return {
      'type': obj.dtype.str,
      'shape': obj.shape,
      'data': obj.tostring(),
   }


def _deserialize_ndarray(obj):
   array = np.fromstring(obj['data'], dtype=np.dtype(obj['type']))
   return array.reshape(obj['shape'])


# Usage:
c = aiomas.Container(('localhost', 5555), codec=aiomas.MsgPack,
                     extra_serializers=[get_np_serializer])

Container clocks¶

Clocks and time are a very important instrument and required, if your agents want to delay the execution of a task for some time, schedule a task at a certain time or just need to define a timeout.

Usually, the real (or wall-clock) time is used for this. In some contexts, however, you need a different notion of time – for example if you want to couple a multi-agent system with external simulators that usually run faster than real-time.

For this reason, every agent container provides a clock via its clock attribute. The default clock is the real-time clock that asyncio uses (AsyncioClock).

An alternative clock is the ExternalClock. The time of this clock can be set by external processes so that the time within your agent system passes as fast (or slow) as in that external process.

The benefit of using aiomas’ clocks compared to just using what asyncio offers is, that you can easily switch clocks (e.g., from the AsyncioClock to the ExternalClock) without touching the agents:

>>> import aiomas
>>>
>>>
>>> CLOCK = aiomas.AsyncioClock()
>>> # CLOCK = aiomas.ExternalClock('2016-01-01T00:00:00')
>>>
>>> class Sleeper(aiomas.Agent):
...     async def run(self):
...          # await asyncio.sleep(.5)  # <-- Don't use this!
...          # Depending on the clock used, this sleeps for a "real" half
...          # second or whatever the ExternalClock tells you:
...          await self.container.clock.sleep(.5)
>>>
>>> container = aiomas.Container.create(('127.0.0.1', 5555), clock=CLOCK)
>>> agent = Sleeper(container)
>>> aiomas.run(agent.run())
>>> container.shutdown()

(If you uncomment the ExternalClock in the example above, your program won’t terminate because there’s no process that sets its time.)

>>> import asyncio
>>> import time
>>>
>>> import aiomas
>>>
>>>
>>> CLOCK = aiomas.ExternalClock('2016-01-01T00:00:00')
>>>
>>> class Sleeper(aiomas.Agent):
...     async def run(self):
...          print('Gonna sleep for 1s ...')
...          await self.container.clock.sleep(1)
>>>
>>>
>>> async def clock_setter(factor=0.5):
...     """Let the time pass *factor* as fast as real-time."""
...     while True:
...         await asyncio.sleep(factor)
...         CLOCK.set_time(CLOCK.time() + 1)
>>>
>>> container = aiomas.Container.create(('127.0.0.1', 5555), clock=CLOCK)
>>>
>>> # Start the process that sets the clock:
>>> t_clock_setter = asyncio.async(clock_setter())
>>>
>>> # Start the agent an measure how long he runs in real-time:
>>> agent = Sleeper(container)
>>> start = time.monotonic()
>>> aiomas.run(agent.run())
Gonna sleep for 1s ...
>>> print('Agent process finished after %.1fs' % (time.monotonic() - start))
Agent process finished after 0.5s
>>>
>>> _ = t_clock_setter.cancel()
>>> container.shutdown()

Testing and debugging¶

Here are some general rules and ideas for developing and debugging distributed systems with aiomas:

• Distributed systems are complex. Always start as simple as possible. Examine and understand the behavior of that system. Start adding a bit more complexity. Repeat.
• I find using a debugger does not work very well with async., distributed systems, so I tend to add a lot of logging and or print()s to my code for debugging purposes.
• Read Develop with asyncio.
• If you enable asyncio’s debug mode, aiomas also falls into debug mode. It gives you better / more detailed exceptions in some cases. This impacts performance, so it isn’t activated always.
• Write unit and integration tests and run them as often as possible. Also check that your tests will fail if they should.

import pytest
import aiomas

#
# Production code (exampleagent.py)
#


class ExampleAgent(aiomas.Agent):
    async def run(self, target_addr, num):
        remote_agent = await self.container.connect(target_addr)
        return (await remote_agent.service(num))

    @aiomas.expose
    async def service(self, val):
        await self.container.clock.sleep(0.001)
        return val


#
# Testing code (test_exampleagent.py)
#


@pytest.yield_fixture
def container(event_loop, unused_tcp_port):
    """This fixture creates a new Container instance for every test and binds
    it to a random port.

    It requires the *event_loop" fixture, so every test will also have a fresh
    event loop.

    """
    # Create container and bind its server socket to a random port:
    c = aiomas.Container.create(('127.0.0.1', unused_tcp_port))

    # Yield the container to the test case:
    yield c

    # Clean-up that is run after the test finished:
    c.shutdown()


# The "@pytest.mark.asyncio" decorator allows you do use "await"/"yield from"
# directly within your test case.
#
# The "container" argument tells pytest to pass the return/yield value of the
# corresponding fixture to your test.
@pytest.mark.asyncio
async def test_example_agent(container):
    num = 42
    # Start two agents:
    agents = [ExampleAgent(container) for _ in range(2)]
    # Run the 1st one and let it connect to the 2nd one.  Check the return
    # value of the 1st one's run() task:
    res = await agents[0].run(agents[1].addr, num)
    assert res == num

Enabling transport security (TLS)¶

This guide explains how you can encrypt all messages sent with aiomas. Transport layer security (TLS, formerly known as SSL) can be applied in a similar fashion to all three layers (channel, RPC, agent) of aiomas and the following sections will show you how.

$ openssl genrsa -aes256 -out ca.key 4096

$ openssl req -new -x509 -nodes -key ca.key -out ca.pem -days 1000

$ openssl genrsa -out device.key 4096
$ openssl req -new -key device.key -out device.csr

$ openssl x509 -CA ca.pem -CAkey ca.key -CAcreateserial -req -in device.csr -out device.pem -days 365

>>> import asyncio
>>> import ssl
>>>
>>> import aiomas
>>>
>>>
>>> async def client(addr, ssl):
...     """Connect to *addr* and use the *ssl* context to enable TLS.
...     Send "ohai" to the server, print its reply and terminate."""
...     channel = await aiomas.channel.open_connection(addr, ssl=ssl)
...     reply = await channel.send('ohai')
...     print(reply)
...     await channel.close()
>>>
>>>
>>> async def handle_client(channel):
...     """Handle client requests by printing them.  Send a reply and
...     terminate."""
...     request = await channel.recv()
...     print(request.content)
...     await request.reply('cya')
...     await channel.close()
>>>
>>>
>>> addr = ('127.0.0.1', 5555)
>>>
>>> # Create an SSLContext for the server supporting (only) TLS 1.2 with
>>> # Eliptic Curve Diffie-Hellman and AES in Galois/Counter Mode
>>> server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
>>> server_ctx.set_ciphers('ECDH+AESGCM')
>>> # Load the cert and key for authentication against clients
>>> server_ctx.load_cert_chain(certfile='device.pem', keyfile='device.key')
>>> # The client also needs to authenticate itself with a cert signed by ca.pem
>>> server_ctx.verify_mode = ssl.CERT_REQUIRED
>>> server_ctx.load_verify_locations(cafile='ca.pem')
>>> # Only use ECDH keys once per SSL session
>>> server_ctx.options |= ssl.OP_SINGLE_ECDH_USE
>>> # Disable TLS compression
>>> server_ctx.options |= ssl.OP_NO_COMPRESSION
>>>
>>> # Start the server.
>>> # It will use "server_ctx" to enable TLS for each connection.
>>> server = aiomas.run(aiomas.channel.start_server(addr, handle_client,
...                                                 ssl=server_ctx))
>>>
>>> # Create an SSLContext for the client supporting (only) TLS 1.2 with
>>> # Eliptic Curve Diffie-Hellman and AES in Galois/Counter Mode
>>> client_ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
>>> client_ctx.set_ciphers('ECDH+AESGCM')
>>> # The server needs to authenticate itself with a cert signed by ca.pem.
>>> # And we also want ot verify its hostname.
>>> client_ctx.verify_mode = ssl.CERT_REQUIRED
>>> client_ctx.load_verify_locations(cafile='ca.pem')
>>> client_ctx.check_hostname = True
>>> # Load the cert and key for authentication against the server
>>> client_ctx.load_cert_chain(certfile='device.pem', keyfile='device.key')
>>>
>>> # Run the client.  It will use "client_ctx" to enable TLS.
>>> aiomas.run(client(addr, client_ctx))
ohai
cya
>>>
>>> # Shutdown the server
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> server_ctx = aiomas.make_ssl_server_context('ca.pem', 'device.pem', 'device.key')
>>> server = aiomas.run(aiomas.channel.start_server(
...     addr, handle_client, ssl=server_ctx))
>>>
>>> client_ctx = aiomas.make_ssl_client_context('ca.pem', 'device.pem', 'device.key')
>>> aiomas.run(client(addr, client_ctx))
ohai
cya
>>> server.close()
>>> aiomas.run(server.wait_closed())

>>> import aiomas
>>>
>>> sslcerts = aiomas.SSLCerts('ca.pem', 'device.pem', 'device.key')
>>> c = aiomas.Container.create(('127.0.0.1', 5555), ssl=sslcerts)
>>>
>>> # Start agents and run your system
>>> # ...
>>>
>>> c.shutdown()

>>> import aiomas
>>>
>>> server_ctx = aiomas.make_ssl_server_context('ca.pem', 'device.pem', 'device.key')
>>> client_ctx = aiomas.make_ssl_client_context('ca.pem', 'device.pem', 'device.key')
>>> c = aiomas.Container.create(('127.0.0.1', 5555), ssl=(server_ctx, client_ctx))
>>>
>>> # Start agents and run your system
>>> # ...
>>>
>>> c.shutdown()

aiomas¶

This module provides easier access to the most used components of aiomas. This purely for your convenience and you can, of cource, also import everything from its actual submodule.

aiomas.agent¶

This module implements the base class for agents (Agent) and containers for agents (Container).

Every agent must live in a container. A container can contain one ore more agents. Containers are responsible for making connections to other containers and agents. They also provide a factory function for spawning new agent instances and registering them with the container.

Thus, the Agent base class is very light-weight. It only has a name, a reference to its container and an RPC router (see aiomas.rpc).

class aiomas.agent.SSLCerts(cafile, certfile, keyfile)¶

namedtuple() storing the names of a CA file, a certificate file and the associated private key file.

See also aiomas.util.make_ssl_server_context() and aiomas.util.make_ssl_client_context().

cafile¶

Alias for field number 0

certfile¶

Alias for field number 1

keyfile¶

Alias for field number 2

class aiomas.agent.Container(base_url, clock, connect_kwargs)[source]¶

Container for agents.

You should not instantiate containers directly but use the create() method/coroutine instead. This makes sure that the container’s server socket is fully operational when it is created.

The container allows its agents to create connections to other agents (via connect()).

In order to destroy a container and close all of its sockets, call shutdown().

classmethod create(addr, *, clock=None, codec=None, extra_serializers=None, ssl=None, as_coro=False)[source]¶

Instantiate a container and create a server socket for it.

This function is a classmethod and coroutine.

Parameters:

addr – is the address that the server socket is bound to. It may be a (host, port) tuple for a TCP socket, a path for a Unix domain socket, or a LocalQueue instance as returned by the aiomas.local_queue.get_queue() function. TCP sockets If host is '0.0.0.0' or '::', the server is bound to all available IPv4 or IPv6 interfaces respectively. If host is None or '', the server is bound to all available IPv4 and IPv6 interfaces. In these cases, the machine’s FQDN (see socket.getfqdn()) should be resolvable and point to that machine as it will be used for the agent’s addresses. If host is a simple (IPv4 or IPv6) IP address, it will be used for the agent’s addresses as is. LocalQueue In contrast to TCP, multiple LocalQueue connections between containers (within the same thread and OS process) send and receive message in a deterministic order, which is useful for testing and debugging. LocalQueue instances should be retrieved via the aiomas.local_queue.get_queue() function (which also available as aiomas.get_queue()). This function always returns the same instance for a given queue ID. clock – can be an instance of BaseClock. It allows you to decouple the container’s (and thus, its agent’s) time from the system clock. This makes it easier to integrate your system with other simulators that may provide a clock for you or to let your MAS run as fast as possible. By default, the real-time AsyncioClock will be used. codec – can be a Codec subclass (not an instance!). JSON is used by default. extra_serializers – is an optional list of extra serializers for the codec. The list entries need to be callables that return a tuple with the arguments for add_serializer(). ssl – allows you to enable TLS for all incoming and outgoing TCP connections. It may either be an SSLCerts instance or a tuple containing two SSLContext instances, where the first one will be used for the server socket, the second one for client sockets. as_coro – must be set to True if the event loop is already running when you call this method. This function then returns a coroutine that you need to await in order to get the container. By default it will block until the server has been started and return the container.

Returns:

a fully initialized Container instance if async is False or else a coroutine returning the instance when it is done.

Invocation examples:

# Synchronous:
container = Container.create(...)

# Asynchronous:
container = await Container.create(..., as_coro=True)

clock¶

The clock of the container. Instance of aiomas.clocks.BaseClock.

connect(url, timeout=0)[source]¶

Connect to the argent available at url and return a proxy to it.

url is a string <protocol>://<addr>//<agent-id> (e.g., 'tcp://localhost:5555/0').

With a timeout of 0 (the default), there will only be one connection attempt before an error is raised (ConnectionRefusedError for TCP sockets and LocalQueue, FileNotFoundError for Unix domain sockets). If you set timeout to a number > 0 or None, this function will try to connect repeatedly for at most that many seconds (or indefinitely) before an error is raised. Use this if the remote agent’s container may not yet exist.

This function is a coroutine.

shutdown(as_coro=False)[source]¶

Close the container’s server socket and the RPC services for all outgoing TCP connections.

If async is left to False, this method calls asyncio.BaseEventLoop.run_until_complete() in order to wait until all sockets are closed.

Set async to True if the event loop is already running (e.g., because you are in a coroutine). The return value then is a coroutine that you need to await in order to actually shut the container down:

await container.shutdown(as_coro=True)

validate_aid(aid)[source]¶

Return the class name for the agent represented by aid if it exists or None.

class aiomas.agent.Agent(container)[source]¶

Base class for all agents.

router¶

Descriptor that creates an RPC Router for every agent instance.

You can override this in a sub-class if you need to. (Usually, you don’t.)

container¶

The Container that the agent lives in.

addr¶

The agent’s address.

aiomas.channel¶

This module implements and asyncio asyncio.Protocol protocol for a request-reply Channel.

aiomas.channel.DEFAULT_CODEC¶

Default codec: JSON

aiomas.channel.open_connection(addr, *, loop=None, codec=None, extra_serializers=(), timeout=0, **kwds)[source]¶

Return a Channel connected to addr.

This is a convenience wrapper for asyncio.BaseEventLoop.create_connection(), asyncio.BaseEventLoop.create_unix_connection(), and aiomas.local_queue.create_connection().

If addr is a tuple (host, port), a TCP connection will be created. If addr is a string, it should be a path name pointing to the unix domain socket to connect to. If addr is a aiomas.local_queue instance, a LocalQueue connection will be created.

You can optionally provide the event loop to use.

By default, the JSON codec is used. You can override this by passing any subclass of aiomas.codecs.Codec as codec.

You can also pass a list of extra_serializers for the codec. The list entires need to be callables that return a tuple with the arguments for add_serializer().

With a timeout of 0 (the default), there will only be one connection attempt before an error is raised (ConnectionRefusedError for TCP sockets and LocalQueue, FileNotFoundError for Unix domain sockets). If you set timeout to a number > 0 or None, this function will try to connect repeatedly for at most that many seconds (or indefinitely) before an error is raised. Use this if you need to start the client before the server.

The remaining keyword argumens kwds are forwarded to asyncio.BaseEventLoop.create_connection() and asyncio.BaseEventLoop.create_unix_connection() respectively.

This function is a coroutine.

aiomas.channel.start_server(addr, client_connected_cb, *, loop=None, codec=None, extra_serializers=(), **kwds)[source]¶

Start a server listening on addr and call client_connected_cb for every client connecting to it.

This function is a convenience wrapper for asyncio.BaseEventLoop.create_server(), asyncio.BaseEventLoop.create_unix_server(), and aiomas.local_queue.create_server().

If addr is a tuple (host, port), a TCP socket will be created. If addr is a string, a unix domain socket at this path will be created. If addr is a aiomas.local_queue instance, a LocalQueue server will be created.

The single argument of the callable client_connected_cb is a new instance of Channel.

You can optionally provide the event loop to use.

By default, the JSON codec is used. You can override this by passing any subclass of aiomas.codecs.Codec as codec.

You can also pass a list of extra_serializers for the codec. The list entires need to be callables that return a tuple with the arguments for add_serializer().

The remaining keyword argumens kwds are forwarded to asyncio.BaseEventLoop.create_server() and asyncio.BaseEventLoop.create_unix_server() respectively.

This function is a coroutine.

class aiomas.channel.ChannelProtocol(codec, client_connected_cb=None, *, loop)[source]¶

Asyncio asyncio.Protocol which connects the low level transport with the high level Channel API.

The codec is used to (de)serialize messages. It should be a sub-class of aiomas.codecs.Codec.

Optionally you can also pass a function/coroutine client_connected_cb that will be executed when a new connection is made (see start_server()).

connection_made(transport)[source]¶

Create a new Channel instance for a new connection.

Also call the client_connected_cb if one was passed to this class.

connection_lost(exc)[source]¶

Set a ConnectionError to the Channel to indicate that the connection is closed.

data_received(data)[source]¶

Buffer incomming data until we have a complete message and then pass it to Channel.

Messages are fixed length. The first four bytes (in network byte order) encode the length of the following payload. The payload is a triple (msg_type, msg_id, content) encoded with the specified codec.

eof_received()[source]¶

Set a ConnectionResetError to the Channel.

write(len_bytes, content)[source]¶

Serialize content and write the result to the transport.

This method is a coroutine.

pause_writing()[source]¶

Set the paused flag to True.

Can only be called if we are not already paused.

resume_writing()[source]¶

Set the paused flat to False and trigger the waiter future.

Can only be called if we are paused.

class aiomas.channel.Request(content, message_id, protocol)[source]¶

Represents a request returned by Channel.recv(). You shoudn’t instantiate it yourself.

content contains the incoming message.

msg_id is the ID for that message. It is unique within a channel.

protocol is the channel’s ChannelProtocol instance that is used for writing back the reply.

To reply to that request you can yield from Request.reply() or Request.fail().

content¶

The content of the incoming message.

reply(result)[source]¶

Reply to the request with the provided result.

This method is a coroutine.

fail(exception)[source]¶

Indicate a failure described by the exception instance.

This will raise a RemoteException on the other side of the channel.

This method is a coroutine.

class aiomas.channel.Channel(protocol, codec, transport, loop)[source]¶

A Channel represents a request-reply channel between two endpoints. An instance of it is returned by open_connection() or is passed to the callback of start_server().

protocol is an instance of ChannelProtocol.

transport is an asyncio.BaseTransport.

loop is an instance of an asyncio.BaseEventLoop.

codec¶

The codec used to de-/encode messages send via the channel.

transport¶

The transport of this channel (see the Python documentation for details).

send(content)[source]¶

Send a request content to the other end and return a future which is triggered when a reply arrives.

One of the following exceptions may be raised:

• ValueError if the message is too long (the length of the encoded message does not fit into a long, which is ~ 4 GiB).
• RemoteException: The remote site raised an exception during the computation of the result.
• ConnectionError (or its subclass ConnectionResetError): The connection was closed during the request.
• RuntimeError:
  • If an invalid message type was received.
  • If the future returned by this method was already triggered or canceled by a third party when an answer to the request arrives (e.g., if a task containing the future is cancelled). You get more detailed exception messages if you enable asyncio’s debug mode

try:
    result = yield from channel.request('ohai')
except RemoteException as exc:
    print(exc)

recv()[source]¶

Wait for an incoming Request and return it.

May raise one of the following exceptions:

• ConnectionError (or its subclass ConnectionResetError): The connection was closed during the request.
• RuntimeError: If two processes try to read from the same channel or if an invalid message type was received.

This method is a coroutine.

close()[source]¶

Coroutine that closes the channel and waits for all sub tasks to finish.

get_extra_info(name, default=None)[source]¶

Wrapper for asyncio.BaseTransport.get_extra_info().

aiomas.clocks¶

Clocks to be used with aiomas.agent.Container.

All clocks should subclass BaseClock. Currently available clock types are:

• AsyncioClock: a real-time clock synchronized with the asyncio event loop.
• ExternalClock: a clock that can be set by external tasks / processes in order to synchronize it with external systems or simulators.

class aiomas.clocks.BaseClock[source]¶

Interface for clocks.

Clocks must at least implement time() and utcnow().

time()[source]¶

Return the value (in seconds) of a monotonic clock.

The return value of consecutive calls is guaranteed to be greater or equal then the results of previous calls.

The initial value may not be defined. Don’t depend on it.

utcnow()[source]¶

Return an arrow.arrow.Arrow date with the current time in UTC.

sleep(dt, result=None)[source]¶

Sleep for a period dt in seconds. Return an asyncio.Future.

If result is provided, it will be passed back to the caller when the coroutine has finished.

sleep_until(t, result=None)[source]¶

Sleep until the time t. Return an asyncio.Future.

t may either be a number in seconds or an arrow.arrow.Arrow date.

If result is provided, it will be passed back to the caller when the coroutine has finished.

call_in(dt, func, *args)[source]¶

Schedule the execution of func(*args) in dt seconds and return immediately.

Return an opaque handle which lets you cancel the scheduled call via its cancel() method.

call_at(t, func, *args)[source]¶

Schedule the execution of func(*args) at t and return immediately.

t may either be a number in seconds or an arrow.arrow.Arrow date.

Return an opaque handle which lets you cancel the scheduled call via its cancel() method.

class aiomas.clocks.AsyncioClock[source]¶

asyncio based real-time clock.

class aiomas.clocks.ExternalClock(utc_start, init_time=0)[source]¶

A clock that can be set by external process in order to synchronize it with other systems.

The initial UTC date utc_start may either be an arrow.arrow.Arrow instance or something that arrow.factory.ArrowFactory.get() can parse.

class aiomas.clocks.TimerHandle(future, callback)[source]¶

This class lets you cancel calls scheduled by ExternalClock.

cancel()[source]¶

Cancel the scheduled call represented by this handle.

aiomas.codecs¶

This package imports the codecs that can be used for de- and encoding incoming and outgoing messages:

• JSON uses JSON
• MsgPack uses msgpack
• MsgPackBlosc uses msgpack and Blosc

All codecs should implement the base class Codec.

aiomas.codecs.serializable(repr=True)[source]¶

Class decorator that makes the decorated class serializable by aiomas.codecs.

The decorator tries to extract all arguments to the class’ __init__(). That means, the arguments must be available as attributes with the same name.

The decorator adds the following methods to the decorated class:

• __asdict__(): Returns a dict with all __init__ parameters
• __fromdict__(dict): Creates a new class instance from dict
• __serializer__(): Returns a tuple with args for Codec.add_serializer()
• __repr__(): Returns a generic instance representation. Adding this method can be deactivated by passing repr=False to the decorator.

Example:

>>> import aiomas.codecs
>>>
>>> @aiomas.codecs.serializable
... class A:
...     def __init__(self, x, y):
...         self.x = x
...         self._y = y
...
...     @property
...     def y(self):
...         return self._y
>>>
>>> codec = aiomas.codecs.JSON()
>>> codec.add_serializer(*A.__serializer__())
>>> a = codec.decode(codec.encode(A(1, 2)))
>>> a
A(x=1, y=2)

class aiomas.codecs.Codec[source]¶

Base class for all Codecs.

Subclasses must implement encode() and decode().

encode(data)[source]¶

Encode the given data and return a bytes object.

decode(data)[source]¶

Decode data from bytes to the original data structure.

add_serializer(type, serialize, deserialize)[source]¶

Add methods to serialize and deserialize objects typed type.

This can be used to de-/encode objects that the codec otherwise couldn’t encode.

serialize will receive the unencoded object and needs to return an encodable serialization of it.

deserialize will receive an objects representation and should return an instance of the original object.

serialize_obj(obj)[source]¶

Serialize obj to something that the codec can encode.

deserialize_obj(obj_repr)[source]¶

Deserialize the original object from obj_repr.

class aiomas.codecs.JSON[source]¶

A Codec that uses JSON to encode and decode messages.

class aiomas.codecs.MsgPack[source]¶

A Codec that uses msgpack to encode and decode messages.

class aiomas.codecs.MsgPackBlosc[source]¶

A Codec that uses msgpack to encode and decode messages and blosc to compress them.

aiomas.exceptions¶

Exception types used by aiomas.

exception aiomas.exceptions.AiomasError[source]¶

Base class for all exceptions defined by aiomas.

exception aiomas.exceptions.RemoteException(origin, remote_traceback)[source]¶

Wraps a traceback of an exception on the other side of a channel.

origin is the remote peername.

remote_traceback is the remote exception’s traceback.

origin = None¶

Peername (producer of the exception)

remote_traceback = None¶

Original traceback

exception aiomas.exceptions.SerializationError[source]¶

Raised when an object cannot be serialized.

aiomas.local_queue¶

The local queue transport roughly mimics a normal TCP transport, but it sends and receives messages via two asyncio.Queue instances.

Its purpose is to aid the development and debugging of complex networking algorithms and distributed or multi-agent systems. In contrast to normal network transports, messages send via the LocalQueueTransport will always arrive in a deterministic order [1].

This transport does not work across multiple processes and is not thread safe, so it should only be used within a single thread and process.

The easiest way to use it is to create a LocalQueue instance via the get_queue() function and pass it to aiomas.channel.start_server()/aiomas.channel.open_connection() or aiomas.agent.Container.create() as addr argument.

aiomas.local_queue.get_queue(queue_id)[source]¶

Return a LocalQueue instance for the given queue_id.

If no instance is cached yet, create a new one.

Queue IDs must be strings and must not contain the / character. Raise a ValueError if these rules are violated.

aiomas.local_queue.clear_queue_cache()[source]¶

Clear the global queue cache.

aiomas.local_queue.create_connection(protocol_factory, lq, *, loop=None, **kwds)[source]¶

Connect to a LocalQueue lq.

The protocol_factory must be a callable returning a protocol instance.

Before a connection to lq can be made, a server must be started for this instance (see create_server()).

This function is a coroutine which will try to establish the connection in the background. When successful, the coroutine returns a (transport, protocol) pair.

aiomas.local_queue.create_server(protocol_factory, lq, **kwds)[source]¶

Create a LocalQueue server bound to lq.

The protocol_factory must be a callable returning a protocol instance.

Return a LocalQueueServer instance. That instance is also set as server for lq.

This function is a coroutine.

class aiomas.local_queue.LocalQueue(queue_id)[source]¶

An instance of this class serves as transport description when creating a server or connection.

The functions create_server() and create_connection() both require an instance of this class. Alternatively, instances of this class can be passed as addr argument to aiomas.channel.start_server() and aiomas.channel.open_connection()

A server needs to be started before any connections can be made.

queue_id¶

The queue’s ID.

server¶

The LocalQueueServer instance that was bound to this instance or None if no server has yet been started.

set_server(server)[source]¶

Set a LocalQueueServer instance.

Raise a RuntimeError if a server has already been bound to this instance.

This method is called by create_server().

unset_server()[source]¶

Unset the server from this instance.

This method is called when the server is closed (see LocalQueueServer.close()).

new_connection(sendq, recvq)[source]¶

Create a connection endpoint on the server side.

This method is called by create_connection().

sendq and recvq are the queues used for sending and receiving messages to and from the client.

class aiomas.local_queue.LocalQueueServer(protocol_factory, lq)[source]¶

Implements asyncio.events.AbstractServer. An instance of this class is returned by create_server().

lq is the LocalQueue instance that this server was bound to.

protocol_factory is a callable that is called for each new client connection in order to create a new protocol instance.

lq¶

The LocalQueue the server is bound to.

new_connection(sendq, recvq)[source]¶

Create a new protocol and transport instance.

Call the protocol factory, create a new LocalQueueTransport with sendq and recvq and wire them together.

Called by create_connection() via LocalQueue.new_connection().

close()[source]¶

Close the server and unset this instance from the associated LocalQueue instance.

wait_closed()[source]¶

Immediately return (there’s nothing to wait for).

This method is a coroutine.

class aiomas.local_queue.LocalQueueTransport(lq, sendq, recvq, protocol)[source]¶

Implements asyncio.transports.Transport.

A LocalQueueTransport has to asynchronous queues (instances of asyncio.Queue) – one for sending messages to the other side and one for receiving messages from it.

close()[source]¶

Close the transport.

Buffered data will be flushed asynchronously. No more data will be received. After all buffered data is flushed, the protocol’s connection_lost() method will (eventually) be called with None as its argument.

write(data)[source]¶

Write some data bytes to the transport.

This does not block; it buffers the data and arranges for it to be sent out asynchronously.

can_write_eof()[source]¶

Return False. This transport does not support write_eof().

abort()[source]¶

Close the transport immediately.

Buffered data will be lost. No more data will be received. The protocol’s connection_lost() method will (eventually) be called with None as its argument.

aiomas.rpc¶

This module implements remote procedure calls (RPC) on top of request-reply channels (see aiomas.channel).

RPC connections are represented by instances of RpcClient (one for each side of a aiomas.channel.Channel). They provide access to the functions served by the peer via Proxy instances. Optionally, they can provide their own RPC service so that the peer can make calls as well.

An RPC service is an object with a router attribute which is an instance of Router. A router resolves paths requested by the peer. It can also handle sub-routers (which allows you to build hierarchies for nested calls) and is able to perform a reverse-lookup of a router (mapping a fuction to its path).

Routers an be attached to both, classes and dictionaries with functions. Dictionaires need to be wrapped with a ServiceDict. Classes need to have a Service class attribute named router. Service is a descriptor which creates a Router for every instance of that class.

Functions that should be callable from the remote side must be decorated with expose(); Router.expose() and Service.expose() are aliases for it.

aiomas.rpc.open_connection(addr, *, rpc_service=None, **kwds)[source]¶

Return an RpcClient connected to addr.

This is a convenience wrapper for aiomas.channel.open_connection(). All keyword arguments (kwds) are forwared to it.

You can optionally pass a rpc_service to allow the peer to call back to us.

This function is a coroutine.

aiomas.rpc.start_server(addr, rpc_service, client_connected_cb=None, **kwds)[source]¶

Start a server socket on host:port and create an RPC service with the provided handler for each new client.

This is a convenience wrapper for aiomas.channel.start_server(). All keyword arguments (kwds) are forwared to it.

rpc_service must be an RPC service (an object with a router attribute that is an instance of Router).

client_connected_cb is an optional callback that will be called with with the RpcClient instance for each new connection.

Raise a ValueError if handler is not decorated properly.

This function is a coroutine.

aiomas.rpc.rpc_service_process(rpc_client, router, channel)[source]¶

RPC service process for a connection rpc_lient.

Serves the functions provided by the Router router via the Channel channel.

Forward errors raised by the handler to the caller.

Stop running when the connection closes.

This function is a coroutine.