CherryPy is a successful small web framework which over the years has built up its performances as well as its stability. To do so, Robert Brewer, the main CherryPy’s architect has introduced what is called the Web Site Process Bus (WSPBus). The idea is to manage a Python process by providing it with a bus to which one can publish or subscribe for events. CherryPy’s implementation of the bus comes with a set of pubsub handlers for very basic operations such as responding to system signals, handle thread creation and deletion, drop process privileges and handle PID files. The bus mechanism can help your handling of sub-processes so that they start, run and terminates gracefully. Let’s see how.
Create your bus
First, you need to create a bus instance. This could be as simple as this.
1
2
| from cherrypy.process import wspbus
bus = wspbus.Bus() |
If you want to log through the bus, you will need further work since the bus doesn’t create a logger by default. Let’s see an example.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
| import sys
import logging
from logging import handlers
from cherrypy.process import wspbus
class MyBus(wspbus.Bus):
def __init__(self, name=""):
wspbus.Bus.__init__(self)
self.open_logger(name)
self.subscribe("log", self._log)
def exit(self):
wspbus.Bus.exit(self)
self.close_logger()
def open_logger(self, name=""):
logger = logging.getLogger(name)
logger.setLevel(logging.INFO)
h = logging.StreamHandler(sys.stdout)
h.setLevel(logging.INFO)
h.setFormatter(logging.Formatter("[%(asctime)s] %(name)s - %(levelname)s - %(message)s"))
logger.addHandler(h)
self.logger = logger
def close_logger(self):
for handler in self.logger.handlers:
handler.flush()
handler.close()
def _log(self, msg="", level=logging.INFO):
self.logger.log(level, msg) |
Not much, just creating a logger and subscribing the bus log channel to an instance method.
Associate the bus with the main process
Before we move on to the management of sub-process, let’s see how we can manage the main Python process already with our bus above.
For this, let’s imagine a bank placing stock orders, those orders will be handled by a broker running in a sub-process.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| import random
import string
from multiprocessing import Process
class Bank(object):
def __init__(self, queue):
self.bus = MyBus(Bank.__name__)
self.queue = queue
self.bus.subscribe("main", self.randomly_place_order)
self.bus.subscribe("exit", self.terminate)
def randomly_place_order(self):
order = random.sample(['BUY', 'SELL'], 1)[0]
code = random.sample(string.ascii_uppercase, 4)
amount = random.randint(0, 100)
message = "%s %s %d" % (order, ''.join(code), amount)
self.bus.log("Placing order: %s" % message)
self.queue.put(message)
def run(self):
self.bus.start()
self.bus.block(interval=0.01)
def terminate(self):
self.bus.unsubscribe("main", self.randomly_place_order)
self.bus.unsubscribe("exit", self.terminate) |
As you can see, not much again here, we simply associate a bus with the bank object. We also register to the exit channel of the bus so that when we terminated, we can do some cleanup. It’s good use to unregister from the bus.
We don’t actually care where those orders come from so we randomly generate them. The orders are placed every time the bus iterates its loop. This is done by attaching to the main channel of the bus.
We use a process queue to communicate with the broker’s sub-process.
Associate the bus with a sub-process
Handling the sub-process is actually similar to handling the main process. Let’s see the broker implementation for example.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
| from Queue import Empty
class Broker(Process):
def __init__(self, queue):
Process.__init__(self)
self.queue = queue
self.bus = MyBus(Broker.__name__)
self.bus.subscribe("main", self.check)
def check(self):
try:
message = self.queue.get_nowait()
except Empty:
return
if message == "stop":
self.bus.unsubscribe("main", self.check)
self.bus.exit()
elif message.startswith("BUY"):
self.buy(*message.split(' ', 2)[1:])
elif message.startswith("SELL"):
self.sell(*message.split(' ', 2)[1:])
def run(self):
self.bus.start()
self.bus.block(interval=0.01)
def stop(self):
self.queue.put("stop")
def buy(self, code, amount):
self.bus.log("BUY order placed for %s %s" % (amount, code))
def sell(self, code, amount):
self.bus.log("SELL order placed for %s %s" % (amount, code)) |
Several things are to be noticed. First we register once again to the bus’ main channel a method that checks the shared queue for incoming data. Whenever the incoming message is “stop”, we exit the bus altogether, thus leaving the sub-process, since it was blocked on the bus loop.
Note that the stop method could be called by the parent process if you needed to programatically stop the sub-process.
Put it all together
Run the code above as follow:
1
2
3
4
5
6
7
8
9
| if __name__ == '__main__':
from multiprocessing import Queue
queue = Queue()
broker = Broker(queue)
broker.start()
bank = Bank(queue)
bank.run() |
This creates the shared queue, starts the sub-process that runs the broker and finally starts the bank within the main process.
You should see a bunch of messages in the console and if you hit Ctrl-C, this will stop both processes cleanly.
And here we are, we now manage processes and sub-processes with a clean solution. The CherryPy process bus is an elegant add-on to your toolbox that I can only highly advise to consider in the future. The WSPBus implementation is part of the main CherryPy package (CherryPy 3.x), so you’ll have to install it all, even if you don’t require the HTTP framework. But don’t let that hold you back since the HTTP framework isn’t required for the bus to be used.
Happy coding!
The code is here.
