Event system, or how to decouple processors

[Lehnart]

When using esper, I like to have all my processors decoupled. I mean no processor needs to python import another_processor to make his job. To achieve this, I use an event system.

I have a python package containing Event classes that have no dependencies, not on Component neither on Processor. A processor can import any Event and can publish it thanks to a function of World. Here is a snippet of a processor that only publishes an event (Maybe I've should used message instead of event :D ):

from event.my_event import AnEvent
class MyProcessor(Processor):
    def __init__():
        super().__init__()

    def process():
         self.world.publish(AnEvent())

Another snippet where a processor receive the event :

from event.my_event import AnEvent
class MySecondProcessor(Processor):
    def __init__():
        super().__init__()

    def process():
         for an_event in self.world.receive(AnEvent):
             do_something()

It could be great to have this mechanism inside the ECS library :

• each processor should have a method to publish/receive events.
• It is the responsibility of the World to ensure the dispatch of events to the processors.

[benmoran56]

Thanks for opening the ticket. I wasn't planning on adding Event Dispatching to esper. Mostly because other libraries exist, and are easy enough to mix in. For example the pyglet EventDispatcher framework, which I use myself.

That said, this is not the first request for this that I've received. I'll have to think about this for a little bit.

[benmoran56]

It wouldn't be too hard to add a light-weight event dispatching system to esper. I'm not sure how far I would want to take this though. Consider the following simple implimentation:

Event handlers could be registered like this:

self.world.set_handler('on_pickup', self.my_method)
self.world.set_hanlder('on_pickup', my_func)

Events can be dispatched by name, and from anywhere that has access to the World:

self.world.dispatch_event('on_pickup', arg1, etc)

Events would be unregistered like this:

self.world.remove_hanlder('on_pickup', my_func)

[Lehnart]

Thanks for your answer. If I found time to do this, I would come back with a proposition on this.

[benmoran56]

I mocked up a quick prototype of this. The API is essentially the same as what I posted above. If you'd like to have a look, it's in the eventsystem branch:

https://github.com/benmoran56/esper/tree/eventsystem

[Lehnart]

Looks nice and simple.

don't you think the event processing should be inside the processor? It bugs me that data/components may be read and modified in an event processing rather than in a processor.

[benmoran56]

(Just a heads-up, but that code I posted earlier is probably non-functional. Maybe it would be better to do esper.set_handler(), etc ?)

What do you think that might look like? Could you share some pseudocode maybe?

EDIT: nevermind, you already did in your first post :)

If Events are iterated over, instead of being dispatched directly, then I think some kind of internal event caching would be necessary. For example:

ProcessorA
ProcessorB.receive('on_coin') --> None
ProcessorC
ProcessorD.dispatch('on_coin', 3)
ProcessorE
ProcessorF.receive('on_coin') --> 3

If ProcessorD dispatches an event, it would have to be stored internally so that ProcessorB has a chance to receive it. What's not clear to me is how long is this cached for? I haven't used any event systems that work this way myself.

[Lehnart]

Actually, I have a working prototype on my side. I will push it there and we can discuss it.

[Lehnart]

#Somewhere in esper.py

class Event:
    """ Not sure it is useful to have an Event class but let s go with it ... """
    def __init__(self):
        pass

    def key(self) -> Type:
        return self.__class__

class EventQueue:

    def __init__(self):
    """ Event queue is a dictionary. Keys are event class and values are list of event instances and their living duration as an integer """
        self._queue = {}

    def add(self, key: Type, event: object):
    """ Add a new event of a given type inside the queue """
        if key not in self._queue:
            self._queue[key] = []
        self._queue[key].append([event, 0])

    def tick(self):
    """ Increment lifetime of events """
        for key in self._queue.keys():
            for event in self._queue[key]:
                event[1] += 1

        for key in self._queue.keys():
            self._queue[key] = [msg for msg in self._queue[key] if msg[1] < 2]

    def get(self, key: Type) -> List:
    """ get all events of a given type if they have been their for exactly one loop of process """
        if key not in self._queue:
            return []
        return [event[0] for event in self._queue[key] if event[1] == 1]

EventQueue is the class responsible for the dispatch of the events. Event have a kind of life time :

• When they are created their lifetime is 0. They can't be read or processed by a processor.
• At the next esper.World process loop, their lifetime is 1 and they are ready to be received by the processors.
• Next esper.World process loop, their lifetime is 2 and they are destroyed.

This way, each event can be read only once by a processor (assuming that each processor run during a world process loop).

[Lehnart]

World Class :

class World:
    # ...
    def publish(self, event: Event):
        """ When someone want to publish a new event """
        self._message_queue.add(event.key(), event)

    def receive(self, event_class) -> List:
        """ When  a processor want to access given type of events """
        return self._message_queue.get(event_class)
    # ...

[Lehnart]

An example in a processor :

class MoveProcessor(Processor):

    def process(self, *args, **kwargs):
        move_events = self.world.receive(MoveEvent)
        for move_event in move_events:
            r = self.world.component_for_entity(move_event.ent, RectComponent)
            r.move(move_event.dx, move_event.dy)

[Lehnart]

Let me know what you think of this.

2 more comments on this:

• What matters for me is really to decouple processors. So I don't want my application to crash if a given event is not sent or if an event is sent but no one is receiving it. Just keeping the event for one process loop iteration helps for that.
• The caching can't be easier I guess. I assume we don't require a real-time event system. It should be ok to wait for a process loop iteration to receive an event. Maybe we can do something fancier with processors subscribing to the event queue but I didn't need it.

[benmoran56]

I just pushed out a new Esper release, with an event system. To keep it simple, it's direct dispatch. This is similar to event systems in libraries like pyglet.

For lazily handling events, you can easily cache them for later iteration:

class MoveProcessor(Processor):
    def __init__(self):
        self.event_queue = []
        esper.set_handler('on_move', self.move_handler)

    def move_handler(self, *args):
        self.event_queue.append(args)

    def process(self, *args, **kwargs):
        for args in self.event_queue:
            ...
            do_something_with_event()
            ...
        self.event_queue.clear()

Let me know if you have any questions about how to make it work.

[Lehnart]

Perfectly fine with this. Thanks a lot for your work!

[benmoran56]

No problem, and I hope it's useful. By the way, if you didn't see it, there is a short explanation at the bottom of the README.

[Lehnart]

Yep. I did read that and it was crystal clear. I ll test it soon in my own projects.