Kooperatives Multi-Threading

In meinem letzten Posting ging es um zwei Kontrolltechniken der Verzögerung: verzögerter Aufruf und verzögerte Ausführung. Hier möchte ich Ihnen zeigen, was Sie damit machen können. Wir werden kooperatives Mutil-Threading realisieren. Das Beispiel ist in Python programmiert.

Zur Erinnerung: Die zu einem zurückgestellten Aufruf benötigte Information speichern wir in einer Datenklasse "DelayedCall".

class DelayedCall(object):
    def __init__(self,func,*args,**kw):
        self.func, self.args, self.kw = func, args, kw

Bauen wir uns zunächst die Infrastruktur, die wir benötigen, um mehrere zurückgestellte Aufrufe zusammen mit ihren (optionalen) zurückgestellten Ausführungen zu speichern. Wir nennen einen solchen Speicher "Scheduler" und die Speicheroperation "schedule". Gespeicherte Aufruf/Ausführungs-Paare sind vom Scheduler via "next" abrufbar; mit dieser Operation wird das Paar außerdem aus dem Speicher gelöscht. Im Grunde handelt es sich um eine einfache Queue, die nach dem FIFO-Prinzip arbeitet (FIFO = First In First Out). Die Methoden "next" und "__iter__" erfüllen in Python das Iterator-Protokoll, so dass der Scheduler als Iterator z.B. in einem for-Statement eingesetzt werden kann. Wir werden das gleich beim Dispatcher im Einsatz sehen.

class Scheduler(object):

    def __init__(self):
        self.callList = []

    def schedule(self,call,callback=None):
        assert isinstance(call,DelayedCall) and \
               (callback == None or callable(callback))
        self.callList.append((call,callback))

    def next(self):
        if self.callList == []: raise StopIteration
        call, callback = self.callList.pop(0)
        return call, callback

    def __iter__(self):
        return self

Eine weitere Klasse namens "Dispatcher" soll verzögerte Aufrufe ausführen und das Ergebnis eines Aufrufs an die callback-Funktion übergeben, falls ein callback gegeben ist. Die callback-Funktion wird nicht direkt ausgeführt, sondern dem Scheduler übergeben. Der callback reiht sich ein in die Schlange der noch auszuführenden Aufrufe! Der Dispatcher läuft nach "run" so lange, bis ihm der Scheduler keine "Nahrung" mehr bietet.

class Dispatcher(object):
    
    def __init__(self,scheduler):
        self.scheduler = scheduler

    def run(self):
        for call, callback in self.scheduler:
            result = call.func(*call.args,**call.kw)
            if callback:
                self.scheduler.schedule(DelayedCall(callback,result))

Mit Scheduler und Dispatcher haben wir uns im Kern die Funktionalität aufgebaut, die z.B. in einem Betriebssystem steckt, um Multi-Tasking (threads) zu realisieren. Im Folgenden nutzen wir diese Infrastruktur, um uns ein eigenes "kooperatives Multi-Threading"-System zu bauen; "kooperativ" deshalb, weil der Aufgerufene freiwillig die Kontrolle wieder an den Dispatcher/Scheduler zurück gibt, denn zwingen kann man ihn nicht dazu. In Simulatoren und Echtzeit-Systemen werden Sie diese Techniken ebenfalls gebrauchen können.

Nun verzeihen Sie mir, dass mir kein besseres Beispiel eingefallen ist. Eine Player-Klasse implementiert ein mehr oder minder aggressives Wesen, das austeilen (fight) und auch einstecken (take_this) muss. Das Ergebnis eines angezettelten fights wird in "follow_up" via callback nachgehalten.

import random

class Player(object):
    def __init__(self,name,aggressiveness,scheduler):
        assert 0 <= aggressiveness <= 1
        self.name = name
        self.aggressiveness = aggressiveness
        self.scheduler = scheduler
        self.fear = 0
        self.action = ["hit","hurt","wounded","punched","scratched","annoyed"]
        self.comment = ["Hey, you seem to like it!",
                        "You Bastard",
                        "You getting serious",
                        "Ok, I got enough of that",
                        "I got the message!",
                        "Ok, I give up"]
        
    def fight(self,opponent):
        action = self.action[int(random.random()*len(self.action))]
        print '%s: "%s, you will get %s"' % (self.name, opponent.name, action)
        self.scheduler.schedule(\
            DelayedCall(opponent.take_this,self,action),\
            self.follow_up)
        
    def follow_up(self,result):
        self.fear += result
        try:
            comment = self.comment[int(self.fear)]
        except IndexError:
            comment = self.comment[-1]
        print '%s: "%s"' % (self.name,comment)

    def take_this(self,opponent,impact):
        self.fear += 1
        print '%s: "Ouch! %s, you %s me!"' % (self.name,opponent.name,impact)
        if random.random() <= self.aggressiveness:
            print '%s: "Hey %s, you will regret this"' % (self.name, opponent.name)
            self.scheduler.schedule(DelayedCall(self.fight,opponent))
            return self.aggressiveness * 2
        return 0

Natürlich ist das Beispiel etwas gekünstelt, es soll einfach nur zeigen, wie die Ideen des letzten Posts zum Einsatz kommen. Bereiten wir für unsere Spieler noch die Bühne, sich zu produzieren:

class Game(object):
    def __init__(self):
        self.scheduler = Scheduler()
        self.dispatcher = Dispatcher(self.scheduler)
        self.player1 = Player("Hulk",0.7,self.scheduler)
        self.player2 = Player("Hogan",0.5,self.scheduler)

    def round(self,attacker,attacked):
        attacker.fear = 0
        attacked.fear = 0
        attacker.fight(attacked)
        self.dispatcher.run()     

Schauen wir uns einmal das Gehabe zwischen Hulk und Hogan ;-) an. Wir nutzen dazu die Textkonsole in Python:

>>> g.round(g.player1,g.player2)
Hulk: "Hogan, you will get annoyed"
Hogan: "Ouch! Hulk, you annoyed me!"
Hogan: "Hey Hulk, you will regret this"
Hogan: "Hulk, you will get hit"
Hulk: "You Bastard"
Hulk: "Ouch! Hogan, you hit me!"
Hulk: "Hey Hogan, you will regret this"
Hulk: "Hogan, you will get hit"
Hogan: "You getting serious"
Hogan: "Ouch! Hulk, you hit me!"
Hogan: "Hey Hulk, you will regret this"
Hogan: "Hulk, you will get annoyed"
Hulk: "Ok, I got enough of that"
Hulk: "Ouch! Hogan, you annoyed me!"
Hogan: "Ok, I got enough of that"
>>>

Nett, gell? Warum haben wir diesen ganzen Aufwand eigentlich betrieben? Verändern Sie einmal den Dispatcher geringfügig, indem der callback nicht gescheduled, sondern sofort zur Ausführung gebracht wird.

class Dispatcher(object):
    
    def __init__(self,scheduler):
        self.scheduler = scheduler

    def run(self):
        for call, callback in self.scheduler:
            result = call.func(*call.args,**call.kw)
            if callback:
                callback(result)

Plötzlich verhalten sich Hulk und Hogan anders (abgesehen von den Zufallszuteilungen)!

>>> g.round(g.player1,g.player2)
Hulk: "Hogan, you will get hurt"
Hogan: "Ouch! Hulk, you hurt me!"
Hogan: "Hey Hulk, you will regret this"
Hulk: "You Bastard"
Hogan: "Hulk, you will get punched"
Hulk: "Ouch! Hogan, you punched me!"
Hulk: "Hey Hogan, you will regret this"
Hogan: "You getting serious"
Hulk: "Hogan, you will get scratched"
Hogan: "Ouch! Hulk, you scratched me!"
Hogan: "Hey Hulk, you will regret this"
Hulk: "Ok, I got enough of that"
Hogan: "Hulk, you will get hit"
Hulk: "Ouch! Hogan, you hit me!"
Hogan: "Ok, I got enough of that"
>>>

Diese Änderung im Dispatcher zeigt eines ganz deutlich: Wir können die Strategie der Ausführung nun selbst und an zentraler Stelle verändern. Ebenso kann der Scheduler die Strategie der Einreihung und Auslieferung verändern. Z.B. gibt es in Echtzeit-Systemen Vorgänge, die mit höherer Priorität abgearbeitet werden müssen als "normale" Vorgänge. Ein Prioritäten-Scheduler kommt in einem solchen Fall zum Einsatz, der die Priorität eines verzögerten Aufrufs mit berücksichtigt.

Wie Sie sehen: Wir haben das "Gesetz des nächsten Befehls" im übertragenen Sinne "durchbrochen".