Twisted Tutorial

Twisted -- The Tutorial

• Welcome
• Gimmick -- Charmed quotes

Twisted -- Networking For Python

• Handles the icky socket stuff
• Handles the icky select stuff
• No threads, no blocking

Finger

• Send username
• Get back some stuff about user
• Will only implement subset of protocol here

Finger - Protocol code

from twisted.protocols import basic

class FingerClient(basic.LineReceiver):

    # This will be called when the connection is made
    def connectionMade(self): self.sendLine(self.factory.user)

    # This will be called when the server sends us a line.
    # IMPORTANT: line *without "\n" at end.
    # Yes, this means empty line does not mean EOF
    def lineReceived(self, line): print line

    # This will be called when the connection is terminated
    def connectionLost(self, _): print "-"*40

Finger - client factory

• Keep configuration information
• In this case, just the username

from twisted.internet import protocol

class FingerFactory(protocol.ClientFactory):
    protocol = FingerProtocol

    def __init__(self, user): self.user = user

    def clientConnectionFailed(self, _, reason):
        print "error", reason.value

Finger - tying it all together

• Actually run above code
• Use reactors

from twisted.internet import reactor
import sys

user, host = sys.argv[1].split('@')
port = 79
reactor.connectTCP(host, port, FingerFactory(port))
reactor.run()

Finger - a bug

• Succeed or fail, program doesn't exit
• Reactor continues in a loop
• Takes almost no CPU time...
• ...but still wrong behaviour

Digression - Deferreds

• In order to be more flexible, we want callbacks
• Common callbacks are too weak
• We used 'deferreds' as an abstraction for callbacks

Finger - reimplementing correctly

from twisted.protocols import basic
from twisted.internet import protocol, defer
import sys

class FingerClient(basic.LineReceiver):

    def connectionMade(self):
        self.transport.write(self.factory.user+"\n")

    def lineReceived(self, line):
        self.factory.gotLine(line)

Finger - reimplementing correctly (cont'd)

class FingerFactory(protocol.ClientFactory):
    protocol = FingerProtocol

    def __init__(self, user):
        self.user, self.d = user, defer.Deferred()

    def gotLine(self, line): print line

    def clientConnectionLost(self, _, why): self.d.callback(None)

    def clientConnectionFailed(self, _, why): self.d.errback(why)

Finger - reimplementing correctly (cont'd 2)

if __name__ == '__main__':
    from twisted.internet import reactor
    from twisted.python import util
    user, host = sys.argv[1].split('@')
    f = FingerFactory(user)
    port = 79
    reactor.connectTCP(host, port, FingerFactory(port))
    f.d.addCallback(lambda _: reactor.stop())
    f.d.addErrback(lambda _: (util.println("could not connect"),
                              reactor.stop()))
    reactor.run()

Servers

• Servers are actually easier
• Servers meant to wait for events
• Most of concepts similar to clients

Finger - protocol

class FingerServer(basic.LineReceiver):

    def lineReceived(self, line):
        self.transport.write(self.factory.getUser(line))
        self.transport.loseConnection()

Finger - factory

class FingerServerFactory(protocol.Factory):

    protocol = FingerServer

    def __init__(self):
        self.users = {}
        self.message = "No such user\n"

    def getUser(self, name):
        return self.users.get(name, self.message)

    def setUser(self, user, status):
        self.users[user] = status

Finger - glue

factory = FingerServerFactory()
factory.setUser("moshez", "Online - Sitting at computer\n")
factory.setUser("spiv", "Offline - Surfing the waves\n")

reactor.listenTCP(79, factory)

Finger Server - problem

• What if server has to actually work to find user's status?
• For example, read status from a website
• API forces us to block -- not good

Finger server -- new protocol

class FingerServer(basic.LineReceiver):

    def lineReceived(self, line):
        d = self.factory.getUser(line)
        d.addCallback(self.writeResponse)
        d.addErrback(self.writeError)

    def writeResponse(self, response):
        self.transport.write(response)
        self.transport.loseConnection()

    def writeError(self, error):
        self.transport.write("Server error -- try later\n")
        self.transport.loseConnection()

Finger - factory

class FingerServerFactory(protocol.Factory):

    protocol = FingerServer

    def __init__(self):
        self.users = {}
        self.message = "No such user\n"

    def getUser(self, name):
        return defer.succeed(self.users.get(name, self.message))

    def setUser(self, user, status):
        self.users[user] = status

Finger - web factory

from twisted.web import client

class FingerWebFactory(protocol.Factory):
    protocol = FingerServer

    def getUser(self, name):
        url = "http://example.com/~%s/online" % name
        d = client.getPage(url)
        d.addErrback(lambda _: "No such user\n")
        return d

Application

• The Twisted way of configuration files
• Decouple configuration from running

Application (Example)

# File: finger.tpy
from twisted.internet import app
import fingerserver

factory = fingerserver.FingerServerFactory()
factory.setUser("moshez", "Online - Sitting at computer\n")
factory.setUser("spiv", "Offline - Surfing the waves\n")
application = app.Application("finger")
application.listenTCP(79, factory)

twistd

• TWISTed Daemonizer
• Daemonizes Twisted servers
• Takes care of log files, PID files, etc.
• twistd -y finger.tpy

twistd examples

• twistd -y finger.tpy -l /var/finger/log
• twistd -y finger.tpy --pidfile /var/run/finger.pid
• twistd -y finger.tpy --chroot /var/run

Writing Plugins

• Automatically create application configurations
• Accessible via commandline or GUI

Writing Plugins (Example)

# File finger/tap.py
from twisted.python import usage

class Options(usage.Options):
    synopsis = "Usage: mktap finger [options]"
    optParameters = [["port", "p", 6666,"Set the port number."]]
    longdesc = 'Finger Server'
    users = ()

    def opt_user(self, user):
        if not '=' in user: status = "Online"
        else: user, status = user.split('=', 1)
        self.users += ((user, status+"\n"),)

Writing Plugins (Example cont'd)

def updateApplication(app, config):
    f = FingerFactory()
    for (user, status) in config.users:
        f.setUser(user, status)
    app.listenTCP(int(config.opts['port']), s)

Writing Plugins (Example cont'd 2)

# File finger/plugins.tml
register("Finger",
         "finger.tap",
         description="Finger Server",
         type='tap',
         tapname="finger")

Using mktap

• mktap finger --user moshez --user spiv=Offline
• twistd -f finger.tap

Delayed execution

• Basic interface: reactor.callLater(<time>, <function>, [<arg>, [<arg> ...]])
• reactor.callLater(10, reactor.stop)
• reactor.callLater(5, util.println, 'hello', 'world')

callLater(0,) -- An idiom

• Use to set up a call in next iteration of loop
• Can be used in algorithm-heavy code to let other code run

def calculateFact(cur, acc=1, d=None):
    d = d or defer.Deferred()
    if cur<=1: d.callback(acc)
    else: reactor.callLater(0, calculateFact, acc*cur, cur-1, d)

calculateFact(10
).addCallback(lambda n: (util.println(n), reactor.stop()))
reactor.run()

UNIX Domain Sockets

• Servers
  • reactor.listenUNIX('/var/run/finger.sock', FingerWebFactory())
• Clients
  • reactor.connectUNIX('/var/run/finger.sock', FingerFactory())

SSL Servers

from OpenSSL import SSL

class ServerContextFactory:

    def getContext(self):
        ctx = SSL.Context(SSL.SSLv23_METHOD)
        ctx.use_certificate_file('server.pem')
        ctx.use_privatekey_file('server.pem')
        return ctx

reactor.listenSSL(111, FingerWebFactory(), ServerContextFactory())

SSL Clients

• from twisted.internet import ssl
• reactor.connectSSL(111, 'localhost', FingerFactory(), ssl.ClientContextFactory())

Running Processes

• A process has two outputs: stdout and stderr
• Protocol to interface with it is different

class Advertizer(protocol.ProcessProtocol):
    def outReceived(self, data): print "out", `data`

    def errReceived(self, data): print "error", `data`

    def processEnded(self, reason): print "ended", reason

reactor.spawnProcess(Advertizer(),
                     "echo", ["echo", "hello", "world"])

Further Reading

• Twisted Docs

Questions?

Bonus Slides

Perspective Broker

• Meant to be worked async
• Can transfer references or copies
• Secure (no pickles or other remote execution mechanisms)
• Lightweight (bandwidth and CPU)
• Translucent

PB Remote Control Finger (Server)

from twisted.spread import pb

class FingerSetter(pb.Root):

    def __init__(self, ff): self.ff = ff

    def remote_setUser(self, name, status):
        self.ff.setUser(name, status+"\n")

ff = FingerServerFactory()
setter = FingerSetter(ff)
reactor.listenUNIX("/var/run/finger.control",
                  pb.BrokerFactory(setter))

PB Remote Control Finger (Client)

from twisted.spread import pb
from twisted.internet import reactor
import sys

def failed(reason):
    print "failed:", reason.value;reactor.stop()

pb.getObjectAt("unix", "/var/run/finger.control", 30
).addCallback(lambda o: o.callRemote("setUser", *sys.argv[1:3],
).addCallbacks(lambda _: reactor.stop(), failed)

reactor.run()

Perspective Broker (Trick)

• Add to the application something which will call reactor.stop()
• Portable (works on Windows)
• Gets around OS security limitations
• Need to add application-level security
• The docs have the answers (see 'cred')

Perspective Broker (Authentication)

• pb.cred
• Perspectives
• Can get remote user with every call
  • Inherit from pb.Perpsective
  • Call methods perspective_<name>(self, remoteUser, ...)

Perspective Broker - About Large Data Streams

• Sending large (>640kb) strings is impossible -- feature, not bug.
• It stops DoSes
• Nobody would ever need...
  • JokeTooOldError
• Use twisted.spread.utils.Pager -- sends the data in managable chunks.

Producers and Consumers

• Use for things like sending a big file
• A good alternative to manually reactor.callLater(0,)-ing
• See twisted.internet.interfaces.{IProducer,IConsumer}

Threads (callInThread)

• Use for long running calculations
• Use for blocking calls you can't do without
• deferred = reactor.callInThread(function, arg, arg)

Threads (callFromThread)

• Use from a function running in a different thread
• Always thread safe
• Interface to non-thread-safe APIs
• reactor.callFromThread(protocol.transport.write, s)

Using ApplicationService

• Keep useful data...
• ...or useful volatile objects
• Support start/stop notification
• Example: process monitor

Playing With Persistence

• Shutdown taps are useful
• Even if you use twistd -y
• So remember
  • Classes belong in modules
  • Functions belong in modules
  • Modifying class attributes should be avoided