Part 1 - Building a Fast Event-Driven Simulator for Concurrent Systems of Hardware Processes Using Curio and Python 3.6

3.1   The Process class

First thing, let's start with a simple class to represent a Process (reminder that all code I'm writing probably requires Python 3.6)

class Process:

    next_id = 0

    def __init__(self, name):
        self.name = name
        self.id = Process.next_id
        Process.next_id += 1

    def __str__(self):
        return "{}.{}".format(self.name, self.id)

    def __repr__(self):
        return "{}('{}')".format(type(self).__name__, self.name)

    def message(self, m):
        print("{}: {}".format(str(self), m))

Nothing too crazy here; just a constructor that keeps track of a unique process id for every single Process that's created, plus some utility functions to create nice string representations and also a message method to hand output from the process in a standard way.

I want to segue here and introduce a new feature in 3.6 called fstrings, which are faster, cleaner to read, and just plain awesome once you start using them.

class Process:

    next_id = 0

    def __init__(self, name):
        self.name = name
        self.id = Process.next_id
        Process.next_id += 1

    def __str__(self):
        return f"{self.name}.{self.id}"

    def __repr__(self):
        return f"{type(self).__name__}('{self.name}')"

    def message(self, m):
        print(f"{self}: {m}")

Look at that, interpolation of code inside strings!

Alright, now let's create a Process subclass that actually does something useful: a Source process that sends a specified value on a channel a fixed number of times.

class Source(Process):

    def __init__(self, name, length, srcval):
        super().__init__(name)
        self.val = srcval
        self.length = length

    def connect(self, chan):
        self.out_chan = chan

    async def exec(self):
        for i in range(self.length):
            self.message(f"sending {self.val}")
            await self.out_chan.send(self.val)
            self.message(f"sent {self.val}")
        self.message("terminated")

The constructor lets us set the value and sequence length to output, and the connect rather naively just connects the passed in channel to an instance variable. We'll define channels in the next section, but for now, let's focus on the exec method which does the actual work. The async keyword marks this as an asynchronously executing coroutine (technically, it marks the method as an awaitable). This is the code that will get executed in Curio's event loop.

The code itself iterates the length of the sequence and calls a send method on the output channel. The method call is preceded by an await keyword, which is how you call async methods (the send method will be an async method). You can read up on the semantics of this (things like yield from), but if you just remember to use await before calling an async method as a rule, you'll be all set.

Now, let's further introduce a Sink process that 'eats' values (say from a source), and the Buffer process described in the previous section that just receives and passes on a value.

from curio import CancelledError
class Sink(Process):

    def __init__(self, name):
        super().__init__(name)

    def connect(self, chan):
        self.in_chan = chan

    async def exec(self):
        tok_count = 0
        try:
            while True:
                tok = await self.in_chan.recv()
                tok_count += 1
                self.message(f"received {tok}")
        except CancelledError:
            self.message(f"{tok_count} tokens received")

class Buffer(Process):

    def __init__(self, name):
        super().__init__(name)

    def connect(self, chan_l, chan_r):
        self.l_chan = chan_l
        self.r_chan = chan_r

    async def exec(self):
        while True:
            tok = await self.l_chan.recv()
            self.message(f"received {tok}")
            self.message(f"sending {tok}")
            await self.r_chan.send(tok)

One new thing here is how we receive data from an async method. The incoming data (or token) is assigned to the result of the await self.in_chan.recv() call. (Recall that the shortcut is, wherever we'd do a function call in synchronous code, we do an await on the function call instead in async code). The other point to note is that we can catch a CancelledError like in the Sink process to print something out if this process is terminated by the environment (which it will be, since otherwise it will just hang waiting for more input).

3.2   The Channel class

One more thing before we can make our first attempt at simulating a system of processes: We need a way to implement the communication channels. Luckily, Curio has a built-in Queue type that's explicitly designed for this kind of communication.

Here's our first stab at the Channel class, assuming we've done a from curio import Queue:

from curio import Queue
class Channel:

    def __init__(self, name):
        self.name = name
        self.q = Queue(maxsize=1)  # Max buffering of 1

    async def send(self, val):
        await self.q.put(val)

    async def recv(self):
        tok =  await self.q.get()
        await self.q.task_done()
        return tok

    async def close(self):
        await self.q.join()

Here, we've defined a Channel that has the actions send and recv (and a utility function close to terminate it). The send inserts a value into the internal Queue, and the recv will await on the value (and block if none is available). Once the recv is done, it's important to call the task_done method on the Queue to signal that the receive is complete.