Saltycrane logo

SaltyCrane Blog

Notes on Python, Django, and web development on Ubuntu Linux


Notes on parallel processing with Python and Twisted

Twisted is a networking engine written in Python, that among many other things, can be used to do parallel processing. It is very big, though, so I had a hard time finding what I needed. I browsed through the Twisted Documentation and the Twisted O'Reilly book. There is also a Recipe in the Python Cookbook. However, I found Bruce Eckel's article, Concurrency with Python, Twisted, and Flex to be the most helpful. (See also Bruce Eckel's initial article on Twisted: Grokking Twisted)

Here are my notes on running Bruce Eckel's example. I removed the Flex part because I didn't need or know anything about it. This example runs a Controller which starts a number of separate parallel processes running Solvers (a.ka. workers). It also allows for communication between the Controller and Solvers. Though this example only runs on one machine, the article said extending this to multiple machines is not difficult. For a good explanation of how this works, please see the original article.

Here is which is copied from the original article. The actual "work" is done in the step method. I only added some debugging print statements for myself.

Original version by Bruce Eckel
Solves one portion of a problem, in a separate process on a separate CPU
import sys, random, math
from twisted.spread import pb
from twisted.internet import reactor

class Solver(pb.Root):

    def __init__(self, id):
        print " %s: solver init" % id = id

    def __str__(self): # String representation
        return "Solver %s" %

    def remote_initialize(self, initArg):
        return "%s initialized" % self

    def step(self, arg):
        print " %s: solver step" %
        "Simulate work and return result"
        result = 0
        for i in range(random.randint(1000000, 3000000)):
            angle = math.radians(random.randint(0, 45))
            result += math.tanh(angle)/math.cosh(angle)
        return "%s, %s, result: %.2f" % (self, str(arg), result)

    # Alias methods, for demonstration version:
    remote_step1 = step
    remote_step2 = step
    remote_step3 = step

    def remote_status(self):
        print " %s: remote_status" %
        return "%s operational" % self

    def remote_terminate(self):
        print " %s: remote_terminate" %
        reactor.callLater(0.5, reactor.stop)
        return "%s terminating..." % self

if __name__ == "__main__":
    port = int(sys.argv[1])
    reactor.listenTCP(port, pb.PBServerFactory(Solver(sys.argv[1])))

Here is This is also copied from the original article but I removed the Flex interface and created calls to start and terminate in the Controller class. I'm not sure if this makes sense, but at least this allowed me to run the example. I also moved the terminate method from the FlexInterface to the Controller.

Original version by Bruce Eckel
Starts and manages solvers in separate processes for parallel processing.
import sys
from subprocess import Popen
from twisted.spread import pb
from twisted.internet import reactor, defer


class Controller(object):

    def broadcastCommand(self, remoteMethodName, arguments, nextStep, failureMessage):
        print " broadcasting..."
        deferreds = [solver.callRemote(remoteMethodName, arguments) 
                     for solver in self.solvers.values()]
        print " broadcasted"
        reactor.callLater(3, self.checkStatus)

        defer.DeferredList(deferreds, consumeErrors=True).addCallbacks(
            nextStep, self.failed, errbackArgs=(failureMessage))
    def checkStatus(self):
        print " checkStatus"
        for solver in self.solvers.values():
                lambda r: sys.stdout.write(r + "\n"), self.failed, 
                errbackArgs=("Status Check Failed"))
    def failed(self, results, failureMessage="Call Failed"):
        print " failed"
        for (success, returnValue), (address, port) in zip(results, self.solvers):
            if not success:
                raise Exception("address: %s port: %d %s" % (address, port, failureMessage))

    def __init__(self):
        print " init"
        self.solvers = dict.fromkeys(
            [("localhost", i) for i in range(START_PORT, START_PORT+MAX_PROCESSES)])
        self.pids = [Popen(["python", "", str(port)]).pid
                     for ip, port in self.solvers]
        print "PIDS: ", self.pids
        self.connected = False
        reactor.callLater(1, self.connect)

    def connect(self):
        print " connect"
        connections = []
        for address, port in self.solvers:
            factory = pb.PBClientFactory()
            reactor.connectTCP(address, port, factory)
        defer.DeferredList(connections, consumeErrors=True).addCallbacks(
            self.storeConnections, self.failed, errbackArgs=("Failed to Connect"))

        print " starting parallel jobs"

    def storeConnections(self, results):
        print " storeconnections"
        for (success, solver), (address, port) in zip(results, self.solvers):
            self.solvers[address, port] = solver
        print " Connected; self.solvers:", self.solvers
        self.connected = True

    def start(self):
        " Begin the solving process"
        if not self.connected:
            return reactor.callLater(0.5, self.start)
        self.broadcastCommand("step1", ("step 1"), self.step2, "Failed Step 1")

    def step2(self, results):
        print " step 1 results:", results
        self.broadcastCommand("step2", ("step 2"), self.step3, "Failed Step 2")

    def step3(self, results):
        print " step 2 results:", results
        self.broadcastCommand("step3", ("step 3"), self.collectResults, "Failed Step 3")

    def collectResults(self, results):
        print " step 3 results:", results
    def terminate(self):
        print " terminate"
        for solver in self.solvers.values():
            solver.callRemote("terminate").addErrback(self.failed, "Termination Failed")
        reactor.callLater(1, reactor.stop)
        return "Terminating remote solvers"

if __name__ == "__main__":
    controller = Controller()

To run it, put the two files in the same directory and run python You should see 2 CPUs (if you have 2) go up to 100% usage. And here is the screen output: init
PIDS:  [12173, 12174] 5567: solver init 5566: solver init connect starting parallel jobs storeconnections Connected; self.solvers: {('localhost', 5567): , ('localhost', 5566): } broadcasting... broadcasted 5566: solver step 5567: solver step checkStatus 5566: remote_status
Solver 5566 operational 5567: remote_status step 1 results: [(True, 'Solver 5567, step 1, result: 683825.75'), (True, 'Solver 5566, step 1, result: 543177.17')] broadcasting... broadcasted
Solver 5567 operational 5566: solver step 5567: solver step checkStatus 5566: remote_status
Solver 5566 operational 5567: remote_status step 2 results: [(True, 'Solver 5567, step 2, result: 636793.90'), (True, 'Solver 5566, step 2, result: 335358.16')] broadcasting... broadcasted
Solver 5567 operational 5566: solver step 5567: solver step checkStatus 5566: remote_status
Solver 5566 operational 5567: remote_status step 3 results: [(True, 'Solver 5567, step 3, result: 847386.43'), (True, 'Solver 5566, step 3, result: 512120.15')] terminate
Solver 5567 operational 5566: remote_terminate 5567: remote_terminate

Post a comment

Required, but not displayed

Format using Markdown. (No HTML.)
  • Code blocks: prefix each line by at least 4 spaces or 1 tab (and a blank line before and after)
  • Code span: surround with backticks
  • Blockquotes: prefix lines to be quoted with >
  • Links: <URL>
  • Links w/ description: [description](URL)
Created with Django | Hosted by Linode