Agent Timing

[ryanstwrt]

Hi, I'm using osbrain for creating a blackboard system with multiple agents, but I am running into some timing issues. I've created two proxy classes that are able to reproduce the error, they are posted below.

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

class Blackboard(Agent):

    def on_init(self):
        self.ka_to_execute = (None, 0)
        self.trigger_event_num = 0
        self.trigger_events = {}
        self.pub_trigger_alias = 'trigger'
        self.pub_trigger_addr = self.bind('PUB', alias=self.pub_trigger_alias)

    def connect_pub_trigger(self, message):
        agent_name, response_addr, response_alias = message
        self.connect(response_addr, alias=response_alias, handler=self.response_trigger_handler)
        return (self.pub_trigger_alias, self.pub_trigger_addr)

    def send_pub_trigger(self):
        self.send(self.pub_trigger_alias, 'sending trigger')

    def response_trigger_handler(self, message):
        self.log_info('Writing Trigger Response')
        agent, trig_val = message

    def run_problem(self):

        for i in range(2):
            #self.log('Starting Timer')
            self.send_pub_trigger()
            time.sleep(1)

class KaBase(Agent):

    def on_init(self):
        self.trigger_val = 0
        self.trigger_response_alias = None
        self.trigger_response_addr = None

    def add_blackboard(self, blackboard):
        self.bb = blackboard

    def connect_pub_trigger(self):
        if self.bb:
            self.trigger_response_alias = 'trigger_response_{}'.format(self.name)
            self.trigger_response_addr = self.bind('PUSH', alias=self.trigger_response_alias)
            self.pub_trigger_alias, self.pub_trigger_addr = self.bb.connect_pub_trigger((self.name, self.trigger_response_addr, self.trigger_response_alias))
            self.connect(self.pub_trigger_addr, alias=self.pub_trigger_alias, handler=self.pub_trigger_handler)

    def pub_trigger_handler(self, message):
        self.log_info('Sent trigger handler')
        self.send(self.trigger_response_alias, (self.name, self.trigger_val))

I am running into an issue with timing when I attempt to use the run_problem method in Blackboard. Below is my attempt to run the method, where the output is posted below

Input

ns = run_nameserver()
bb = run_agent(name='blackboard', base=Blackboard)
ka_rp = run_agent(name='ka_rp', base=KaBase)

ka_rp.add_blackboard(bb)
ka_rp.connect_pub_trigger()

bb.run_problem()

ns.shutdown()

Output

Broadcast server running on 0.0.0.0:9091
NS running on 127.0.0.1:18727 (127.0.0.1)
URI = PYRO:Pyro.NameServer@127.0.0.1:18727
INFO [2020-02-26 16:47:34.983973] (ka_rp): Sent trigger handler
INFO [2020-02-26 16:47:35.985655] (ka_rp): Sent trigger handler
INFO [2020-02-26 16:47:36.987399] (blackboard): Writing Trigger Response
INFO [2020-02-26 16:47:36.990119] (blackboard): Writing Trigger Response
NS shut down.

We notice that both the blackboard publishes the trigger to our agent, however we do not get the push response (i.e. Writing Trigger Response) from the agent until after both triggers are sent. The intended function should send the trigger via publish and receive the push response before the next trigger event is send out. However, I've noticed that if I perform this type of action outside of the Blackboard agent, I get the correct functionality. The loop below is essentially mimicking the run_problem method.

Input

ns = run_nameserver()
bb = run_agent(name='blackboard', base=Blackboard)
ka_rp = run_agent(name='ka_rp', base=KaBase)

ka_rp.add_blackboard(bb)
ka_rp.connect_pub_trigger()

for i in range(2):
    bb.send_pub_trigger()
    time.sleep(1)

ns.shutdown()

Output

Broadcast server running on 0.0.0.0:9091
NS running on 127.0.0.1:17733 (127.0.0.1)
URI = PYRO:Pyro.NameServer@127.0.0.1:17733
INFO [2020-02-26 16:47:29.409920] (ka_rp): Sent trigger handler
INFO [2020-02-26 16:47:29.411943] (blackboard): Writing Trigger Response
INFO [2020-02-26 16:47:30.413244] (ka_rp): Sent trigger handler
INFO [2020-02-26 16:47:30.415158] (blackboard): Writing Trigger Response
NS shut down.

We notice that the trigger is sent by the blackboard, and immediately afterwards the KaBase agent sends its push response. I imagine that I am missing something relating to the timing of sending/receiving messages. Any help would be appreciated. Thank you.

-Ryan

[Peque]

Yeah, your run_problem() method takes control of the main thread, so the agent does not process incoming messages until it finishes the run_problem() execution. That means, your Blackboard is receiving the messages on time, only it does not process them on time.

You want to do that asynchronously.

An alternative (not recommended) would be to call bb.unsafe.run_problem(). See the documentation on one-way, unsafe calls. This is not recommended since it is "unsafe", which means you will have two threads running concurrently. Python's GIL protects you, but better be sure!

So you may want to use timers instead:

# bb.run_problem()
bb.each(1, 'send_pub_trigger')

time.sleep(3)
ns.shutdown()

[Peque]

PS: if timers do not fit your purpose well, consider splitting your Blackboard agent in two: make one agent "active", which just sleeps and sends triggers. Then the other agent is completely reactive: it receives triggers and reacts to them by sending PUB messages. Once they are sent, its reaction is complete, so it goes back to listening to new incoming events. This way it will process the reply back from the agent on time.

Alternatively, you can safely execute tasks in parallel if you run bb.unsafe.run_task() or bb.after(0, 'task'), but in order to be safe you would need to communicate with the main thread using the loopback socket, or another inproc socket that you create. This is unfortunately not documented. If you want to contribute documentation on that, welcome! :blush:

[ryanstwrt]

Wonderful, that is definitely what I was missing! Thank you for the information. I think I might try this a couple of different ways to see which is most effective for my problem. If I end up using the last method you suggested, I'll be sure to write something up. Thanks again.

[Peque]

Closing this for now then. Feel free to comment/reopen if you need more help. :blush: