osBrain - 0.4.4¶

What is osBrain?¶

osBrain is a general-purpose multi-agent system module written in Python and developed by OpenSistemas. Agents run independently as system processes and communicate with each other using message passing.

osBrain uses ØMQ for efficient and flexible messsage passing between agents. It also uses Pyro4 to ease the configuration and deployment of complex systems.

Please read the Software License and Disclaimer.

Contents¶

About osBrain¶

Feature overview¶

osBrain is a general-purpose multi-agent system module written in Python.

Agents run independently as system processes and communicate with each other using message passing.
Message passing is implemented using ØMQ, and in particular, the PyZMQ Python bindings.
ØMQ allows for efficient, asynchronous communication using different commonly used communication patterns such as request-reply, push-pull and publish-subscribe.
osBrain integrates Pyro4 to ease the configuration and deployment of complex systems.
Thanks to Pyro4, remote agents can be treated as local objects and reconfigured even when they are running. Not just variables, but also new methods can be created in the remote agents.
osBrain provides the base for implementing robust, highly-available, flexible multi-agent systems.
Being implemented in Python, osBrain can take advantage of a huge set of packages for data analysis, statistics, numerical computing, etc. available in the Python ecosystem.

In order to fully understand osBrain capabilities, it is highly recommended to read the Pyro4 documentation and the ØMQ guide.

OsBrain’s history¶

osBrain was initially developed in OpenSistemas based on the need to create a real-time automated-trading platform. This platform needed to be able to process real-time market data updates fast and in a parallel way. Robustness was very important as well in order to prevent running trading strategies from being affected by a failure in another strategy.

Python was chosen for being a great language for fast prototyping and for having a huge data analysis ecosystem available. It was kept for its final performance and the beautiful source code created with it.

The appearance of osBrain was a consecuence of a series of steps that were taken during the development process:

Isolation of agents; creating separate system processes to avoid shared memory and any problems derived from multi-threading development.
Implementation of message passing; making use of the modern, efficient and flexible ØMQ library.
Ease of configuration/deployment; making use of the very convenient, well implemented and documented Pyro4 package.
Separation from the trading platform; what started as a basic architecture for implementing a real-time automated-trading platform, ended-up being a general-purpose multi-agent system architecture.

What can you use osBrain for?¶

osBrain has been successfully used to develop a real-time automated-trading platform in OpenSistemas, but being a general-purpose multi-agent system, it is not limited to this application. Other applications include:

Transportation.
Logistics.
Defense and military applications.
Networking.
Load balancing.
Self-healing networks.

In general, osBrain can be used whenever a multi-agent system architecture fits the application well:

Autonomy of the agents.
Local views.
Decentralization.

Performance¶

The performance of osBrain, just as the performance of any other system architecture, depends a lot on the actual application. The developer should always take this into account:

Pyro4 is used only for configuration, deployment, updating and debugging, which means that the actual performance of the system should not depend on this package.
ØMQ is used with the PyZMQ Python bindings, which means that the system performance depends on the PyZMQ performance.
osBrain uses pickle for serialization by default, which means that the system performance may as well depend on this package. Serialization is configurable, though.
osBrain default transport is IPC by default, but it can be changed globally or configured specifically for each bind. Note, however, that when using TCP, the network may have a great impact on performance.

Introduction and Example¶

Installation¶

This tutorial is a step-by-step introduction to osBrain with examples. In order to start playing with this module, you only need to install it.

osBrain requires Python 3. Most probably, Python 3 is already packaged for your favorite distribution (and maybe even installed by default in your system). If you do not have Python 3 available, consider using Conda to create a virtual environment with Python 3.

Installing osBrain is very simple with pip:

pip install osbrain

You should now be able to import osbrain from a python console:

>>> import osbrain

Hello world¶

The first example is, of course, a simple hello world! program. Three steps are taken here:

Run a name server.
Run an agent with an alias Example.
Log a Hello world message from the agent.

from osbrain import run_nameserver
from osbrain import run_agent


if __name__ == '__main__':

    # System deployment
    run_nameserver()
    agent = run_agent('Example')

    # Log a message
    agent.log_info('Hello world!')

Running this example from your terminal should simply show you a log message saying Hello world! but, what exactly is happening there?

Agents and proxies¶

An agent, in osBrain, is an entity that runs independly from other agents in the system. An agent, by default, will simply poll for incoming messages before executing the code defined by the developer. This means a single agent, as in the Hello world! example, makes little or no sense. Agents in a multi-agent system start to make sense when connected to each other.

The easiest way to run an agent in an osBrain architecture is by calling the function osbrain.run_agent:

>>> agent = run_agent(...)

This function will spawn a new agent and will return a osbrain.Proxy to it.

Proxies are simply local objects that allow us to easily have access to the remote agent. The fact that agents are run independently from each other justifies the need of a proxy.

A proxy allows us to call methods or access attributes of the remote agent in a very convenient way. See for example the previous call:

>>> agent.log_info('Hello world')

The method log_info() is implemented in osbrain.Agent so, when this method is called from the proxy, this call is actually being serialized to the remote running agent and gets executed there. The return value, if any, is then serialized back and returned by the proxy. So basically so get the impression of being working with a local object while your code is executed remotely.

The name server¶

A name server is just like any other agent, so it runs independently, but with a very specific role. Name servers are used as an address book. This means other agents can be run in the system and can be registered in the name server using a human-readable alias. Aliases help us accessing these agents easily even from remote locations.

Note that when calling the osbrain.run_agent() function, we are passing a string parameter. This parameter is the alias the agent will use to register itself in the name server.

When we run a name server calling the osbrain.run_nameserver(), we also get in return a proxy to this name server:

>>> ns = run_nameserver()

This proxy can be used to list the agents registered in the name server:

from osbrain import run_nameserver
from osbrain import run_agent


if __name__ == '__main__':

    # System deployment
    ns = run_nameserver()
    run_agent('Agent0')
    run_agent('Agent1')
    run_agent('Agent2')

    # Show agents registered in the name server
    for alias in ns.agents():
        print(alias)

The code above should simply print the aliases of all the agents registered in the name server.

A name server proxy can also be used to create proxies to registered agents. This is specially useful when accessing the multi-agent system from a different console or location, as it will reduce the number of addresses that we need to remember.

from osbrain import run_nameserver
from osbrain import run_agent


if __name__ == '__main__':

    # System deployment
    ns = run_nameserver()
    run_agent('Agent0')
    run_agent('Agent1')
    run_agent('Agent2')

    # Create a proxy to Agent1 and log a message
    agent = ns.proxy('Agent1')
    agent.log_info('Hello world!')

The code above creates (and registers) three different agents in a name server and then creates, through the name server proxy, a proxy to one of those agents simply using its alias. Then it uses the agent proxy to remotely call a method to log a Hello world! message.

Basic communication patterns¶

Push-Pull¶

Example¶

Now that we understand the basics of how proxies, agents and name servers work, let us jump into a more interesting example.

As mentioned before, a multi-agent system only makes sense if agents are connected with each other and share some information using message passing.

In this first example, we will create two agents: Alice and Bob, and we will make alice send messages to Bob using a simple push-pull communication pattern.

import time
from osbrain import run_agent
from osbrain import run_nameserver


def log_message(agent, message):
    agent.log_info('Received: %s' % message)


if __name__ == '__main__':

    # System deployment
    run_nameserver()
    alice = run_agent('Alice')
    bob = run_agent('Bob')

    # System configuration
    addr = alice.bind('PUSH', alias='main')
    bob.connect(addr, handler=log_message)

    # Send messages
    while True:
        time.sleep(1)
        alice.send('main', 'Hello, Bob!')

So, in this case, we are doing some more stuff. After we spawn Alice and Bob, we connect them.

First, we make Alice bind:

addr = alice.bind(‘PUSH’, alias=’main’)

There are three things to remark in that line:

The first parameter 'PUSH' represents the communication pattern we want to use. In this case we are using a simple push-pull (unidirectional) pattern to allow Alice to send messages to Bob.
The second parameter is, once again, an alias. We can use this alias to refer to this communication channel in an easier way.
The binding, as you already guessed, takes place in the remote agent, but it actually returns a value, which is the address the agent binded to. This address is serialized back to us so we can use it to connect other agents to it.

The next interesting line of code is the one in which Bob connects to Alice:

bob.connect(addr, handler=log_message)

There are two things to remark in here:

Calling connect() from an agent requires, first, an address. This address is, in this case, the one we got after binding Alice. This method will automatically select the appropriate communication pattern to connect to this pattern ('PULL' in this case).
Bob will be receiving messages from Alice, so we must set a handler function that will be executed when a message from Alice is received. This handler will be serialized and stored in the remote agent to be executed there when needed.

The handler function, in its most basic form, accepts two parameters:

def handler(agent, message):
    ...

The actual agent (can be named self as well, in an OOP way).
The message that is received.

In the example above, the handler simply logs the message received.

List of handlers¶

When using push-pull communication patterns we are allowed to set multiple handlers using a list. In example:

agent.connect('PULL', handler=[handler1, handler2, handler3])

Note that in this case all handlers will be executed in sequence.

Request-Reply¶

Example¶

Another common communication pattern is the request-reply, in which a requester sends a message to the replier and expects always a reply back. It is sometimes useful, specially when some kind of synchronization is required.

from osbrain import run_agent
from osbrain import run_nameserver


def reply(agent, message):
    return 'Received ' + str(message)


if __name__ == '__main__':

    run_nameserver()
    alice = run_agent('Alice')
    bob = run_agent('Bob')

    addr = alice.bind('REP', alias='main', handler=reply)
    bob.connect(addr, alias='main')

    for i in range(10):
        bob.send('main', i)
        reply = bob.recv('main')
        print(reply)

The main difference with respect to the push-pull pattern is that, in this case, Bob must run the recv method in order to get the reply back from Alice.

Note

Although the requester is not required to immediately await for the reply (i.e.: can do other stuff after sending the request and before receiving the response), it is required to receive a reply back before making another request through the same communication channel. Multiple requests can be made from the same agent as long as it uses different communication channels for each request.

Return versus yield¶

The easiest way to reply to a request is to return a value from the handler, as seen in Request-Reply:

def reply(agent, message):
    return 'Received ' + str(message)

However, using return the agent can only send a response after executing the handler. Instead, an agent can use yield to reply earlier if needed:

def reply(agent, message):
    yield 'Received' + str(message)  # Reply now
    agent.log_info('Already sent a reply back!')   # Do some stuff later

Publish-Subscribe¶

Example¶

One of the most useful communication patterns between agents is the publish and subscribe pattern. The publisher will send messages to all subscribed agents.

Here is an example in which Alice is the publisher and Bob and Eve subscribe to Alice. This way, when Alice sends a message, both Bob and Eve will receive it:

import time
from osbrain import run_agent
from osbrain import run_nameserver


def log_message(agent, message):
    agent.log_info('Received: %s' % message)


if __name__ == '__main__':

    # System deployment
    run_nameserver()
    alice = run_agent('Alice')
    bob = run_agent('Bob')
    eve = run_agent('Eve')

    # System configuration
    addr = alice.bind('PUB', alias='main')
    bob.connect(addr, handler=log_message)
    eve.connect(addr, handler=log_message)

    # Send messages
    while True:
        time.sleep(1)
        alice.send('main', 'Hello, all!')

Note the similarities between this example and the Sender-Receiver example. The only differences are that Alice is now binding using the 'PUB' pattern and that, instead of having just Bob connecting to Alice, we now have Eve as well connecting to Alice.

This communication pattern allows for easy filtering. Refer to the Filtering section in the tutorial for more details.

Filtering¶

The publish-subscribe pattern is very useful, but it is also very powerful when combined with filtering.

Any time we publish a message from an agent, a topic can be specified. If a topic is specified, then only the agents that are subscribed to that topic will receive the message. This filtering is done in the publisher side, meaning that the network does not suffer from excessive message passing.

In the following example we have Alice publishing messages using topic a or b at random. Then we have Bob subscribed to both topics, Eve subscribed to topic a only and Dave subscribed to topic b only.

import time
import random
from osbrain import run_agent
from osbrain import run_nameserver


def log_a(agent, message):
    agent.log_info('Log a: %s' % message)


def log_b(agent, message):
    agent.log_info('Log b: %s' % message)


if __name__ == '__main__':

    # System deployment
    run_nameserver()
    alice = run_agent('Alice')
    bob = run_agent('Bob')
    eve = run_agent('Eve')
    dave = run_agent('Dave')

    # System configuration
    addr = alice.bind('PUB', alias='main')
    bob.connect(addr, handler={'a': log_a, 'b': log_b})
    eve.connect(addr, handler={'a': log_a})
    dave.connect(addr, handler={'b': log_b})

    # Send messages
    while True:
        time.sleep(1)
        topic = random.choice(['a', 'b'])
        message = 'Hello, %s!' % topic
        alice.send('main', message, topic=topic)

Note how we can specify different handlers for different topics when subscribing agents.

Considerations¶

Clients versus servers¶

When using Basic communication patterns we have a lot of flexibility:

We are allowed to connect multiple clients to a server.
The server can play any role (i.e.: does not need to be always REP, but can be REQ as well). Servers are only defined by the action of binding, not by the role they play in the communication pattern.

For example, if we bind using PUSH and we connect multiple clients to this server, then messages pushed will be distributed among the clients in a Round-robin fashion, which means the first message will be received by the first client, the second message will be received by the second client, and so on.

If we bind using PULL and we connect multiple clients to this server, then messages pushed will all be received by the single server, as expected.

For more information simply refer to the ØMQ guide.

Adding new methods¶

Note that proxies can not only be used to execute methods remotely in the agent, but they can also be used to add new methods or change already existing methods in the remote agent.

In the following example you can see how we can create a couple of functions that are then added to the remote agent as new methods.

In order to add new methods (or change current methods) we only need to call set_method() from the proxy.

from osbrain import run_agent
from osbrain import run_nameserver


def set_x(self, value):
    self.x = value


def set_y(self, value):
    self.y = value


def add_xy(self):
    return self.x + self.y


if __name__ == '__main__':

    # System deployment
    run_nameserver()
    agent = run_agent('Example')

    # System configuration
    agent.set_method(set_x, set_y, add=add_xy)

    # Trying the new methods
    agent.set_x(1)
    agent.set_y(2)
    print(agent.add())

Note that set_method() accepts any number of parameters:

In case they are not named parameters, the function names will be used as the method names in the remote agent.
In case they are named parameters, then the method in the remote agent will be named after the parameter name.

Lambdas¶

osBrain uses dill for serialization when communicating with remote agents through a proxy. This means that almost anything can be serialized to an agent using a proxy.

In order to further simplify some tasks, lambda functions can be used to configure remote agents:

from osbrain import run_agent
from osbrain import run_nameserver


if __name__ == '__main__':

    run_nameserver()
    alice = run_agent('Alice')
    bob = run_agent('Bob')

    addr = alice.bind('REP', handler=lambda agent, msg: 'Received ' + str(msg))
    bob.connect(addr, alias='main')

    for i in range(10):
        bob.send('main', i)
        reply = bob.recv('main')
        print(reply)

See the similarities between this example and the one showed in Request-Reply. In fact, the only difference is the binding from Alice, in which we are using a lambda function for the handler.

Shutting down¶

Although not covered in the examples until now (because many times you just want the multi-agent system to run forever until, perhaps, an event occurs), it is possible to actively kill the system using proxies:

import time

from osbrain import run_agent
from osbrain import run_nameserver


def tick(agent):
    agent.log_info('tick')


if __name__ == '__main__':

    ns = run_nameserver()
    a0 = run_agent('Agent0')
    a1 = run_agent('Agent1')

    a0.each(1, tick)
    a1.each(1, tick)
    time.sleep(3)

    a0.shutdown()
    time.sleep(3)

    ns.shutdown()

Note

Shutting down the nameserver will result in all agents registered in the name server being shut down as well.

OOP¶

Although the approach of using proxies for the whole configuration process is valid, sometimes the developer may prefer to use OOP to define the behavior of an agent.

This, of course, can be done with osBrain:

import time
from osbrain import run_agent
from osbrain import run_nameserver
from osbrain import Agent


class Greeter(Agent):
    def on_init(self):
        self.bind('PUSH', alias='main')

    def hello(self, name):
        self.send('main', 'Hello, %s!' % name)


def log_message(agent, message):
    agent.log_info('Received: %s' % message)


if __name__ == '__main__':

    # System deployment
    run_nameserver()
    alice = run_agent('Alice', base=Greeter)
    bob = run_agent('Bob')

    # System configuration
    bob.connect(alice.addr('main'), handler=log_message)

    # Send messages
    while True:
        time.sleep(1)
        alice.hello('Bob')

Most of the code is similar to the one presented in the Push-Pull example, however you may notice some differences:

When runing Alice, a new parameter base is passed to the osbrain.run_agent() function. This means that, instead of running the default agent class, the user-defined agent class will be used instead. In this case, this class is named Greeter.
The Greeter class implements two methods:
1. on_init(): which is executed on initialization and will, in this case, simply bind a 'PUSH' communication channel.
2. hello(): which simply logs a Hello message when it is executed.
When connecting Bob to Alice, we need the address where Alice binded to. As the binding was executed on initialization, we need to use the addr() method, which will return the address associated to the alias passed as parameter (in the example above it is main).

Timers¶

Repeated actions¶

Timers can be used to repeat an action after a period of time. To illustrate this, let us modify the Push-Pull example a bit and make use of the .each() method:

from osbrain import run_agent
from osbrain import run_nameserver


def log_message(agent, message):
    agent.log_info('Received: %s' % message)


def annoy(agent, say, more=None):
    message = say if not more else say + ' ' + more + '!'
    agent.send('annoy', message)


if __name__ == '__main__':

    run_nameserver()
    orange = run_agent('Orange')
    apple = run_agent('Apple')
    addr = orange.bind('PUSH', alias='annoy')
    apple.connect(addr, handler=log_message)

    # Multiple timers with parameters
    orange.each(1., annoy, 'Hey')
    orange.each(1.4142, annoy, 'Apple')
    orange.each(3.1415, annoy, 'Hey', more='Apple')

Note that if an action takes longer to run than the time available before the next execution, the timer will simply fall behind.

Delayed actions¶

Timers can be used to execute an action after a defined time delay using the .after() method:

from osbrain import run_agent
from osbrain import run_nameserver


def delayed(agent):
    agent.log_info('Logged later')


if __name__ == '__main__':

    run_nameserver()
    agent = run_agent('a0')

    agent.after(1, delayed)
    agent.log_info('Logged now')

Note that if an action takes longer to run than the time available before the next execution, the timer will simply fall behind.

Stopping timers¶

When executing the .each() or .after() methods, a timer is created and an identifier is returned that can be used later to refer to that timer (i.e.: for stopping it).

In the following example we can see how the returned identifier and an alias parameter can be used to identify the timer and stop it calling the .stop_timer() method:

from osbrain import run_agent
from osbrain import run_nameserver


def delayed(agent, message):
    agent.log_info(message)


if __name__ == '__main__':

    run_nameserver()
    agent = run_agent('a0')

    agent.after(1, delayed, 'Hello!')
    # Timer ID returned
    timer0 = agent.after(1, delayed, 'Never logged')
    # Timer alias set
    agent.after(1, delayed, 'Never logged either', alias='timer_alias')

    # Stop timers by ID and alias
    agent.stop_timer(timer0)
    agent.stop_timer('timer_alias')

You can call the .stop_all_timers() method if you would rather simply stop all started timers.

If you want to list the timers that are currently running, you can call the .list_timers() method.

Transport protocol¶

Available transports¶

Althought the default transport protocol is IPC, there are other transport protocols that can be used in osBrain:

tcp: common TCP. Can always be used, and must be used to communicate agents running in different machines.
ipc: common IPC. It is the default and the best suited for communication between agents that run on the same machine
inproc: for in-process communication (between threads). The fastest protocol, although it is limited to communication between threads that share the same Agent.context

The transport protocol can be changed on a per-socket basis, per-agent basis and also globally.

Changing the transport¶

It is possible to change the default global transport protocol by setting the osbrain.config['TRANSPORT'] configuration variable. So, for example, to set TCP as the default transport, we could set:

osbrain.config['TRANSPORT'] = 'tcp'

We can also set the default transport that a particular agent should use by default by passing the transport parameter to run_agent:

agent = run_agent('a0', transport='tcp')

If we do not want to change the global default nor any agent’s default, then we can still change the transport protocol when binding, passing the transport parameter again:

agent = run_agent('a0')
agent.bind('PULL', transport='inproc')

Note

It is also possible to change the global default transport protocol setting the OSBRAIN_DEFAULT_TRANSPORT environment variable.

Serialization¶

Introduction¶

osBrain uses pickle module for serialization when passing messages between agents internally and can use pickle, json and raw serialization (in which raw bytes are sent, hence the name) for serialization when configuring and deploying the multi-agent architectures.

It is well known that using pickle or json for this purpose is a security risk. The main problem is that allowing a program to unpickle or unjson arbitrary data can cause arbitrary code execution and this may wreck or compromise your system.

NOTE: Be aware that different serializers might have different limitations. For example, the json method does not support serializing an object of bytes type, while pickle does support it.

Defining the serializer¶

Specifying the serializer only makes sense in server sockets, since clients will automatically detect and set the type they need in order to communicate accordingly with the server.

There are three ways in which the serializer can be specified:

Global configuration.
Specifying it at per agent configuration.
Specifying it at per socket configuration.

Global configuration¶

By setting the osbrain.config['SERIALIZER'] configuration variable, we can change the default serializer. For example:

osbrain.config['SERIALIZER'] = 'json'

Note

It is also possible to change the global default serializer by setting the OSBRAIN_DEFAULT_SERIALIZER environment variable.

Per agent configuration¶

Specifying the serializer at per agent level will override the global configuration. This can be done as follows:

a1 = run_agent('a1', serializer='json')

Per socket configuration¶

Finally, we can specify the serializer at per socket level. This will override any other configuration (global/per agent). For example:

a1 = run_agent('a1', serializer='json')
# Raw serialization will override json for this socket
addr1 = a1.bind('PUB', 'alias1', serializer='raw')

PUBSUB messaging pattern¶

For the PUBSUB pattern, there is a special character (b’x80’ as of now, even though it could change at any time) that we use so as to let the agents know what is the topic and what is the message itself. Note that the special separator character is only required if there is a topic and the serialization option is NOT set to raw (read below for more information).

Considerations when using raw serialization and PUBSUB pattern¶

Special care must be taken when working with raw serialization and the PUBSUB messaging pattern. Under those conditions, we decided to replicate the raw ZeroMQ PUBSUB communication, in which the topic is sent along with the message and is the handler the one that must take care of separating the topic from the message it self.

Note that if we are using other type of serialization, it is safe to assume that what we are receiving only the original message, without any traces of the topic.

Advanced proxy handling¶

Note

Before reading this section make sure you have already understood Agents and proxies.

Understanding proxies better¶

When an agent is run, there is one thread executing the method Agent.run, which simply listens forever for incoming messages to process them. At the same time, the Agent object is served so it can be accessed through proxies.

In order to prevent concurrent access to the agent, proxy calls are by default serialized and sent to the main thread to be executed there. This way we avoid any concurrent access to memory (only the main thread will make changes to the agent).

However, the user has control and can change that default behavior.

Warning

The user can decide to change the default behavior and make unsafe calls. However, they should be aware of the possible risks they may bring. If not carefully selected, unsafe calls can lead to undefined behavior and hard-to-debug problems.

Executing unsafe calls¶

You can change the default behavior and make unsafe calls, which means the method call will be executed on another thread separate from the main thread. In order to do so, the Proxy provides us with access to the safe and unsafe attributes, which allows us to force safe or unsafe calls for single remote method calls:

agent = run_agent('test')

agent.unsafe.method_call()  # This forces an unsafe method_call()
agent.safe.method_call()  # This forces a safe method_call()

agent.method_call()  # This is a default (safe) method_call() again

The Proxy behavior can also be changed on a per-proxy basis, either when executing the run_agent function:

agent = run_agent('test', safe=False)

agent.method_call()  # This is now by default an unsafe method_call()

Or when creating a new Proxy to the agent:

run_agent('a0')
agent = Proxy('a0', safe=False)

agent.method_call()  # This is now by default an unsafe method_call()

Note

It is also possible, although totally unadvisable, to change the default proxy behavior globally. Setting the OSBRAIN_DEFAULT_SAFE environment variable to false would result in all proxies making unsafe calls by default. Another option is to change the osbrain.config['SAFE'] configuration variable.

Executing one-way, unsafe calls¶

In some situations, you might want to execute unsafe calls in parallel without waiting for any return. In that case, you can make use of the oneway proxy attribute:

agent_proxy.oneway.method_call()

This method call will return immediately, and the code will be executed in the remote agent concurrently with the main thread and any other unsafe or one-way calls.

Warning

Do note that oneway calls are actually executed in a separate thread, which means they behave like unsafe calls. If not used with care, this concurrency may result in unexpected hard-to-debug behavior.

Security¶

Warning

osBrain should be considered unsafe when used with remote machines. This package has some security risks. Understanding the risks is very important to avoid creating systems that are very easy to compromise by malicious entities.

Serialization in osBrain¶

osBrain uses pickle module for serialization when passing messages between agents and can use pickle and dill for serialization when configuring and deploying the multi-agent architectures. It is well known that using pickle or dill for this purpose is a security risk. The main problem is that allowing a program to unpickle or undill arbitrary data can cause arbitrary code execution and this may wreck or compromise your system.

Network interface binding¶

By default osBrain binds every server on localhost, to avoid exposing things on a public network or over the internet by mistake. If you want to expose your osBrain agents to anything other than localhost, you have to explicitly tell osBrain the network interface address it should use. This means it is a conscious effort to expose agents to remote machines.

Protocol encryption¶

osBrain doesn’t encrypt the data it sends over the network. This means you must not transfer sensitive data on untrusted networks (especially user data, passwords, and such) because it is possible to eavesdrop. Either encrypt the data yourself before passing, or run osBrain over a secure network (VPN or SSH tunnel).

Developers¶

Workflow¶

We are happy you like osBrain and we would love to receive contributions from you! Note that contributions do not necessarily need to include code: you can help telling us what problems you are having with osBrain or suggesting improvements to the documentation.

If you would like to help us with code, proceed to fork the project, make the changes you want and then submit a pull request to start a discussion. Take into account that we follow some rules for development:

We like to follow style standards (i.e.: PEP8). But do not worry, our test suite will help you with that and tell you if you wrote something wrong.
We like tests, so any new functionality should also include the corresponding tests.
We like documentation, so changes in the code should also include changes in the code docstrings and changes in the user documentation if they affect how the user may use osBrain.

Installing dependencies¶

To install the required dependencies for developing osBrain, you can make use of the provided requirements.txt file:

pip install -r requirements.txt

Running the tests¶

Running the tests locally is very simple, using Tox from the top level path of the project:

tox

That single command will run all the tests for all the supported Python versions available in your system or environment.

For faster results you may want to run all the tests just against a single Python version. This command will run all tests against Python 3.5 only:

tox -e py35

Note that those tests include style and static analysis checks. If you just want to run all the behavior tests (not recommended):

pytest -n 8

If you just want to run a handful of behavior tests (common when developing new functionality), just run:

pytest -k keyword

Note

Before submitting your changes for review, make sure all tests pass with tox, as the continuous integration system will run all those checks as well.

Generating documentation¶

Documentation is generated with Sphinx. In order to generate the documentation locally you need to run make from the docs directory:

make html

osBrain library API¶

This chapter describes osBrain’s library API. All osBrain classes and functions are defined in sub packages such as osbrain.agent, but for ease of use, the most important ones are also placed in the osbrain package scope.

`osbrain` — Main API package¶

osbrain is the main package of osBrain. It imports most of the other packages that it needs and provides shortcuts to the most frequently used objects and functions from those packages. This means you can mostly just import osbrain in your code to start using osBrain.

The classes and functions provided are:

symbol in `osbrain`	referenced location
class `osbrain.Agent`¶	`osbrain.agent.Agent`
`osbrain.run_agent`()¶	`osbrain.agent.run_agent()`
`osbrain.run_nameserver`()¶	`osbrain.nameserver.run_nameserver()`
class `osbrain.Proxy`¶	`osbrain.proxy.Proxy`
class `osbrain.NSProxy`¶	`osbrain.proxy.NSProxy`
class `osbrain.Logger`¶	`osbrain.logging.Logger`
`osbrain.run_logger`()¶	`osbrain.logging.run_logger()`
class `osbrain.SocketAddress`¶	`osbrain.address.SocketAddress`
class `osbrain.AgentAddress`¶	`osbrain.address.AgentAddress`

`osbrain.agent` — osBrain agent logic¶

Core agent classes.

class osbrain.agent.Agent(name=None, host=None, serializer=None, transport=None)¶

Bases: object

A base agent class which is to be served by an AgentProcess.

An AgentProcess runs a Pyro multiplexed server and serves one Agent object.

Parameters:

name : str, default is None: Name of the Agent.
host : str, default is None: Host address where the agent will bind to. When not set, ‘127.0.0.1’ (localhost) is used.
transport : str, AgentAddressTransport, default is None: Transport protocol.

Attributes:

name : str: Name of the agent.
host : str: Host address where the agent is binding to.
socket : dict: A dictionary in which the key is the address or the alias and the value is the actual socket.
adddress : dict: A dictionary in which the key is the address or the alias and the value is the actual address.
handler : dict: A dictionary in which the key is the socket and the values are the handlers for each socket.
poll_timeout : int: Polling timeout, in milliseconds. After this timeout, if no message is received, the agent executes de iddle() method before going back to polling.
keep_alive : bool: When set to True, the agent will continue executing the main loop.
running : bool: Set to True if the agent is running (executing the main loop).

Methods

addr(alias)

Parameters:

after(delay, method, *args[, alias]) Execute an action after a delay.

bind(kind[, alias, handler, addr, …]) Bind to an agent address.

close_sockets() Close all non-internal zmq sockets.

connect(server[, alias, handler]) Connect to a server agent address.

each(period, method, *args[, alias]) Execute a repeated action with a defined period.

execute_function(function, *args, **kwargs) Execute a function passed as parameter.

get_unique_external_zmq_sockets() Return an iterable containing all the zmq.Socket objects from self.socket which are not internal, without repetition.

iddle() This function is to be executed when the agent is iddle.

iterate() Agent’s main iteration.

list_timers()

Returns:

log_debug(message[, logger]) Log a debug message.

log_error(message[, logger]) Log an error message.

log_info(message[, logger]) Log an info message.

log_warning(message[, logger]) Log a warning message.

loop() Agent’s main loop.

on_init() This user-defined method is to be executed after initialization.

ping() A test method to check the readiness of the agent.

raise_exception() Raise an exception (for testing purposes).

recv(address) Receive a message from the specified address.

run() Run the agent.

safe_call(method, *args, **kwargs) A safe call to a method.

send(address, message[, topic, handler, …]) Send a message through the specified address.

send_recv(address, message) This method is only used in REQREP communication patterns.

set_attr(**kwargs) Set object attributes.

set_logger(logger[, alias]) Connect the agent to a logger and start logging messages to it.

set_method(*args, **kwargs) Set object methods.

stop() Stop the agent.

stop_all_timers() Stop all currently running timers.

stop_timer(alias) Stop a currently running timer.

subscribe(alias, handlers, str], Any]) Subscribe the agent to another agent.

get_attr
kill
register
registered
shutdown

addr(alias)¶

Parameters:	alias : str Alias of the socket whose address is to be retreived.
Returns:	AgentAddress Address of the agent socket associated with the alias.

after(delay, method, *args, alias=None, **kwargs)¶

Execute an action after a delay.

Parameters:	delay : float Execute the action after delay seconds. method Method (action) to be executed by the agent. alias : str, default is None An alias for the generated timer. args : tuple Parameters to pass for the method execution. *kwargs : dict Named parameters to pass for the method execution.
Returns:	str The timer alias or identifier.

bind(kind, alias=None, handler=None, addr=None, transport=None, serializer=None)¶

Bind to an agent address.

Parameters:	kind : str, AgentAddressKind The agent address kind: PUB, REQ… alias : str, default is None Optional alias for the socket. handler, default is None If the socket receives input messages, the handler/s is/are to be set with this parameter. addr : str, default is None The address to bind to. transport : str, AgentAddressTransport, default is None Transport protocol.
Returns:	AgentAddress The address where the agent binded to.

close_sockets()¶: Close all non-internal zmq sockets.

connect(server, alias=None, handler=None)¶

Connect to a server agent address.

Parameters:	server : AgentAddress Agent address to connect to. alias : str, default is None Optional alias for the new address. handler, default is None If the new socket receives input messages, the handler/s is/are to be set with this parameter.

each(period, method, *args, alias=None, **kwargs)¶

Execute a repeated action with a defined period.

Parameters:	period : float Repeat the action execution with a delay of period seconds between executions. method Method (action) to be executed by the agent. alias : str, default is None An alias for the generated timer. args : tuple Parameters to pass for the method execution. *kwargs : dict Named parameters to pass for the method execution.
Returns:	str The timer alias or identifier.

execute_function(function, *args, **kwargs)¶: Execute a function passed as parameter.

get_attr(name)¶

get_unique_external_zmq_sockets()¶

Return an iterable containing all the zmq.Socket objects from self.socket which are not internal, without repetition.

Originally, a socket was internal if its alias was one of the following:

loopback

_loopback_safe

inproc://loopback

inproc://_loopback_safe

However, since we are storing more than one entry in the self.socket dictionary per zmq.socket (by storing its AgentAddress, for example), we need a way to simply get all non-internal zmq.socket objects, and this is precisely what this function does.

iddle()¶

This function is to be executed when the agent is iddle.

After a timeout occurs when the agent’s poller receives no data in any of its sockets, the agent may execute this function.

iterate()¶

Agent’s main iteration.

This iteration is normally executed inside the main loop.

The agent is polling all its sockets for input data. It will wait for poll_timeout; after this period, the method iddle will be executed before polling again.

Returns:	int 1 if an error occurred during the iteration (we would expect this to happen if an interruption occurs during polling). 0 otherwise.

kill()¶

list_timers()¶

Returns:	list (str) A list with all the timer aliases currently running.

log_debug(message, logger='_logger')¶

Log a debug message.

Parameters:	message : str Message to log. logger : str Alias of the logger.

log_error(message, logger='_logger')¶

Log an error message.

Parameters:	message : str Message to log. logger : str Alias of the logger.

log_info(message, logger='_logger')¶

Log an info message.

Parameters:	message : str Message to log. logger : str Alias of the logger.

log_warning(message, logger='_logger')¶

Log a warning message.

Parameters:	message : str Message to log. logger : str Alias of the logger.

loop()¶

Agent’s main loop.

This loop is executed until the keep_alive attribute is False or until an error occurs.

on_init()¶: This user-defined method is to be executed after initialization.

ping()¶: A test method to check the readiness of the agent. Used for testing purposes, where timing is very important. Do not remove.

raise_exception()¶: Raise an exception (for testing purposes).

recv(address)¶

Receive a message from the specified address.

This method is only used in REQREP communication patterns.

Parameters:	address :
Returns:	anything The content received in the address.

register(socket, address, alias=None, handler=None)¶

registered(address)¶

run()¶: Run the agent.

safe_call(method, *args, **kwargs)¶

A safe call to a method.

A safe call is simply sent to be executed by the main thread.

Parameters:	method : str Method name to be executed by the main thread. args : arguments Method arguments. kwargs : keyword arguments Method keyword arguments.

send(address, message, topic=None, handler=None, wait=None, on_error=None)¶

Send a message through the specified address.

Note that replies in a REQREP pattern do not use this function in order to be sent.

Parameters:

address : AgentAddress or AgentChannel: The address to send the message through.
message: The message to be sent.
topic : str: The topic, in case it is relevant (i.e.: for PUB sockets).
handler : function, method or string: Code that will be executed on input messages if relevant (i.e.: for PULL sockets).
wait : float: For channel requests, wait at most this number of seconds for a response from the server.
on_error : function, method or string: Code to be executed if wait is passed and the response is not received.

send_recv(address, message)¶: This method is only used in REQREP communication patterns.

set_attr(**kwargs)¶

Set object attributes.

Parameters:	kwargs : [name, value] Keyword arguments will be used to set the object attributes.

set_logger(logger, alias='_logger')¶: Connect the agent to a logger and start logging messages to it.

set_method(*args, **kwargs)¶

Set object methods.

Parameters:	args : [function] New methods will be created for each function, taking the same name as the original function. kwargs : [name, function] New methods will be created for each function, taking the name specified by the parameter.
Returns:	str Name of the registered method in the agent.

shutdown()¶

stop()¶: Stop the agent. Agent will stop running.

stop_all_timers()¶: Stop all currently running timers.

stop_timer(alias)¶

Stop a currently running timer.

Parameters:	alias : str The alias or identifier of the timer.

subscribe(alias: str, handlers: Dict[Union[bytes, str], Any])¶

Subscribe the agent to another agent.

Parameters:	alias Alias of the new subscriber socket. handlers A dictionary in which the keys represent the different topics and the values the actual handlers. If ,instead of a dictionary, a single handler is given, it will be used to subscribe the agent to any topic.

class osbrain.agent.AgentProcess(name, nsaddr=None, addr=None, serializer=None, transport=None, base=<class 'osbrain.agent.Agent'>)¶

Bases: multiprocessing.context.Process

Agent class. Instances of an Agent are system processes which can be run independently.

Attributes:	authkey `daemon` Return whether process is a daemon `exitcode` Return exit code of process or None if it has yet to stop `ident` Return identifier (PID) of process or None if it has yet to start name `pid` Return identifier (PID) of process or None if it has yet to start `sentinel` Return a file descriptor (Unix) or handle (Windows) suitable for waiting for process termination.

Methods

`is_alive`()	Return whether process is alive
`join`([timeout])	Wait until child process terminates
`run`()	Method to be run in sub-process; can be overridden in sub-class
`sigint_handler`(signal, frame)	Handle interruption signals.
`start`()	Start child process
`terminate`()	Terminate process; sends SIGTERM signal or uses TerminateProcess()

kill

kill()¶

run()¶: Method to be run in sub-process; can be overridden in sub-class

sigint_handler(signal, frame)¶: Handle interruption signals.

start()¶: Start child process

osbrain.agent.compose_message(message: bytes, topic: bytes, serializer: osbrain.address.AgentAddressSerializer) → bytes¶

Compose a message and leave it ready to be sent through a socket.

This is used in PUB-SUB patterns to combine the topic and the message in a single bytes buffer.

Parameters:	message Message to be composed. topic Topic to combine the message with. serializer Serialization for the message part.
Returns:	The bytes representation of the final message to be sent.

osbrain.agent.deserialize_message(message, serializer)¶

Check if a message needs to be deserialized and do it if that is the case.

Parameters:	message : bytes, memoryview The serialized message. serializer : AgentAddressSerializer The type of (de)serializer that should be used.
Returns:	anything The deserialized message, or the same message in case no deserialization is needed.

osbrain.agent.execute_code_after_yield(generator)¶

Some responses are dispatched with yield (generator handler). In those cases we still want to execute the remaining code in the generator, and also make sure it does not yield any more.

Raises:	ValueError If the generator yielded once more, which is unexpected.

osbrain.agent.run_agent(name, nsaddr=None, addr=None, base=<class 'osbrain.agent.Agent'>, serializer=None, transport=None, safe=None)¶

Ease the agent creation process.

This function will create a new agent, start the process and then run its main loop through a proxy.

Parameters:	name : str Agent name or alias. nsaddr : SocketAddress, default is None Name server address. addr : SocketAddress, default is None New agent address, if it is to be fixed. transport : str, AgentAddressTransport, default is None Transport protocol. safe : bool, default is None Use safe calls by default from the Proxy.
Returns:	proxy A proxy to the new agent.

osbrain.agent.serialize_message(message, serializer)¶

Check if a message needs to be serialized and do it if that is the case.

Parameters:	message : anything The message to serialize. serializer : AgentAddressSerializer The type of serializer that should be used.
Returns:	bytes The serialized message, or the same message in case no serialization is needed.

`osbrain.address` — osBrain address logic¶

Implementation of address-related features.

class osbrain.address.AgentAddress(transport, address, kind, role, serializer)¶

Bases: object

Agent address information consisting on the transport protocol, address, kind and role.

Parameters:	transport : str, AgentAddressTransport Agent transport protocol. address : str Agent address. kind : str, AgentAddressKind Agent kind. role : str, AgentAddressRole Agent role. serializer : str Agent serializer type.
Attributes:	transport : str, AgentAddressTransport Agent transport protocol. address : str, SocketAddress Agent address. kind : AgentAddressKind Agent kind. role : AgentAddressRole Agent role. serializer : AgentAddressSerializer Agent serializer.

Methods

twin()

Returns:

twin()¶

Returns:	AgentAddress The twin address of the current one; while the host and port are kept for the twin, the kind and role change to their corresponding twins, according to the rules defined in the respective classes.

class osbrain.address.AgentAddressKind¶

Bases: str

Agent’s address kind class.

This kind represents the communication pattern being used by the agent address: REP, PULL, PUB…

Methods

capitalize() Return a capitalized version of S, i.e.

casefold() Return a version of S suitable for caseless comparisons.

center(width[, fillchar]) Return S centered in a string of length width.

count(sub[, start[, end]]) Return the number of non-overlapping occurrences of substring sub in string S[start:end].

encode([encoding, errors]) Encode S using the codec registered for encoding.

endswith(suffix[, start[, end]]) Return True if S ends with the specified suffix, False otherwise.

expandtabs([tabsize]) Return a copy of S where all tab characters are expanded using spaces.

find(sub[, start[, end]]) Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end].

format(*args, **kwargs) Return a formatted version of S, using substitutions from args and kwargs.

format_map(mapping) Return a formatted version of S, using substitutions from mapping.

index(sub[, start[, end]]) Like S.find() but raise ValueError when the substring is not found.

isalnum() Return True if all characters in S are alphanumeric and there is at least one character in S, False otherwise.

isalpha() Return True if all characters in S are alphabetic and there is at least one character in S, False otherwise.

isdecimal() Return True if there are only decimal characters in S, False otherwise.

isdigit() Return True if all characters in S are digits and there is at least one character in S, False otherwise.

isidentifier() Return True if S is a valid identifier according to the language definition.

islower() Return True if all cased characters in S are lowercase and there is at least one cased character in S, False otherwise.

isnumeric() Return True if there are only numeric characters in S, False otherwise.

isprintable() Return True if all characters in S are considered printable in repr() or S is empty, False otherwise.

isspace() Return True if all characters in S are whitespace and there is at least one character in S, False otherwise.

istitle() Return True if S is a titlecased string and there is at least one character in S, i.e.

isupper() Return True if all cased characters in S are uppercase and there is at least one cased character in S, False otherwise.

join(iterable) Return a string which is the concatenation of the strings in the iterable.

ljust(width[, fillchar]) Return S left-justified in a Unicode string of length width.

lower() Return a copy of the string S converted to lowercase.

lstrip([chars]) Return a copy of the string S with leading whitespace removed.

maketrans(x[, y, z]) Return a translation table usable for str.translate().

partition(sep) Search for the separator sep in S, and return the part before it, the separator itself, and the part after it.

replace(old, new[, count]) Return a copy of S with all occurrences of substring old replaced by new.

requires_handler()

Returns:

rfind(sub[, start[, end]]) Return the highest index in S where substring sub is found, such that sub is contained within S[start:end].

rindex(sub[, start[, end]]) Like S.rfind() but raise ValueError when the substring is not found.

rjust(width[, fillchar]) Return S right-justified in a string of length width.

rpartition(sep) Search for the separator sep in S, starting at the end of S, and return the part before it, the separator itself, and the part after it.

rsplit([sep, maxsplit]) Return a list of the words in S, using sep as the delimiter string, starting at the end of the string and working to the front.

rstrip([chars]) Return a copy of the string S with trailing whitespace removed.

split([sep, maxsplit]) Return a list of the words in S, using sep as the delimiter string.

splitlines([keepends]) Return a list of the lines in S, breaking at line boundaries.

startswith(prefix[, start[, end]]) Return True if S starts with the specified prefix, False otherwise.

strip([chars]) Return a copy of the string S with leading and trailing whitespace removed.

swapcase() Return a copy of S with uppercase characters converted to lowercase and vice versa.

title() Return a titlecased version of S, i.e.

translate(table) Return a copy of the string S in which each character has been mapped through the given translation table.

twin()

Returns:

upper() Return a copy of S converted to uppercase.

zfill(width) Pad a numeric string S with zeros on the left, to fill a field of the specified width.

zmq()

Returns:

REQUIRE_HANDLER = ('REP', 'PULL', 'SUB', 'PULL_SYNC_PUB')¶

TWIN = {'PUB': 'SUB', 'PULL': 'PUSH', 'PULL_SYNC_PUB': 'PUSH_SYNC_SUB', 'PUSH': 'PULL', 'PUSH_SYNC_SUB': 'PULL_SYNC_PUB', 'REP': 'REQ', 'REQ': 'REP', 'SUB': 'PUB'}¶

ZMQ_KIND_CONVERSION = {'PUB': 1, 'PULL': 7, 'PULL_SYNC_PUB': 7, 'PUSH': 8, 'PUSH_SYNC_SUB': 8, 'REP': 4, 'REQ': 3, 'SUB': 2}¶

requires_handler()¶

Returns:	bool Whether the Agent’s address kind requires a handler or not. A socket which processes incoming messages would require a handler (i.e. ‘REP’, ‘PULL’, ‘SUB’…).

twin()¶

Returns:	AgentAddressKind The twin kind of the current one; REQ would be the twin of REP and viceversa, PUB would be the twin of SUB and viceversa, etc.

zmq()¶

Returns:	int The equivalent ZeroMQ socket kind.

class osbrain.address.AgentAddressRole¶

Bases: str

Agent’s address role class. It can either be ‘server’ or ‘client’.