Reactor Pattern
Um was gehts
Das Reaktor Pattern erlaubt es, einer ereignis-getriebenen Anwendung eine oder mehrere Client Anfragen gleichzeitig anzunehmen und auf verschiedene Serviceanbieter zu verteilen ( demultiplex and dispatch ) .
Auch bekannt unter
Beispiel
- Ein Logging Server erhält mehrere Anfragen gleichzeitig und muß sie auf die verschieden Ausgabegeräte verteilen:
- Eventhandling ins GUIs
Anforderungen
Ein Server soll mehrere Clientanfragen gleichzeitig beantworten können. Jede Clientanfrage besitzt eine eindeutige Indikation, die sie einem Serviceprovider zuordnen läßt. Folgende Bedingungen müssen für die Applikation gewährleistet sein:
- sie soll nicht blockieren
- sie soll auf maximalen Durchsatz ausgelegt sein und somit unnötiges Kontext wechseln, Daten synchronisieren oder Daten kopieren vermeiden
- sie soll einfach um neue und verbesserte Service erweitert werden können
- sie soll ohne komplexe multithreading und synchronisations Mechanismen auskommen
Lösung
Führe für jeden Servicetyp, den die Applikation anbietet, einen Handler ein. Dieser Eventhandler verarbeitet die spezifische Clientanfragen. Registriere den Handler beim Reaktor, der einem synchronen Event Verteiler (
event demultiplexer) unterhält um auf eingehende Events zu reagieren. Wenn ein Event im synchronen
event demultiplexerauftritt, benachrichtigt dieser den Reaktor, der das Event auf den angefragen Service verteilt.
Struktur
Handles
- identifizieren Eventquellen wie Sockets, Filehandles oder Timersignale des OS-Systems
- die Eventquellen erzeugen Events wie connect, read, time-out an, die auf den assoziierten Handle geschoben werden
- der Handle kann nur die entsprechende Operation vollziehen
synchrone event demultiplex
- der Verteiler (demultiplexer) wartet auf Indikatoren (indication events), die auf einer Menge von Handles auftreten
- bis die indication events abgeholt werden, blockiert der event _demultiplexer
- auf dem assozierten Handle kann nun das eintreffende Ereignis aufgerufen
Event Handler
- definiert das Interface um indication events zu prozessieren
- deklarieren die Services der Applikation
Konkrete Event Handler
- verarbeiten indication events in einer applikationsspezifischen Art
- definieren die Services der Applikation
Reaktor
- stellt ein Interface zu Verfügung, damit die Event Handler inklusiver ihrer assozierten Handles registrieren und entfernen kann
- der Reaktor benützt den synchronen Verteiler (event demultiplexer) um auf die Indikatoren (indicaton events) der Handles zu warten
- beim Auftreten eines Indikators (indication events) ordnet der Reaktor dies Ereugnis dem entsprechenden Ereignis Handler zu
- nach der Zuordnung des Ereignis an den entsprechenden Ereignis Handler, ruft ( dispatch ) der Reaktor die assozierte Methode auf dem Event Handler auf
- der Reaktor startet und unterhält die event loop der Applikation
Nicht die Applikation, sonder der Reaktor wartet auf
indication events, die er auf die entsprechenden konkreten Event Handler verteilt (
demultiplex ) und dann deren assozierte Hook Methode aufruft (
dispatch ). Als Applikationsentwickler gilt es die spezifischen Event Handler zu implementieren und sie beim Reaktor zu registrieren.
Der Reaktor als Framework stellt eine Ablaufumgebung für die Eventverarbeitung bereit. Diese
inversion of control - die Applikation wird durch den Reaktor gesteurt - wird als
Hollywood Prinzip bezeichnet.
Don't call me, we call you.
- Reaktor Klassendiagramm:
Umsetzung
Der Reaktor reagiert auf Zeittakte. Diese Ereignisse werden auf die registrierten Handler abgebildet. Sobald die Eventloop des Reaktors mittels reactor.run() gestartet wird, können die Ereignisse verarbeitet werden.
import time
from twisted.internet import task
# http://twistedmatrix.com/trac/browser/trunk/twisted/internet/task.py
from twisted.internet import reactor
# http://twistedmatrix.com/trac/browser/trunk/twisted/internet/reactor.py
# define handler as object
class Handler():
def __init__(self, Id ):
self.__id= Id
def __call__(self):
print "Handler with id %s" % self.__id
print "at %d:%d:%d%s\n" %(time.localtime()[3:6] + ( str(time.time() % 1)[1:] ,))
# register handler as callable object
l1 = task.LoopingCall(Handler(1))
# start the task
# start calls implicit reactor.callLater(... ) to reschedule the task ( fire the time event )
l1.start(0.3) # call every 0.3 seconds
l2 = task.LoopingCall(Handler(2))
l2.start(1) # call every second
# running the event loop
reactor.run()
Timer mit ACE
- starte zwei Timer cb1 und cb2, die durch Signale SIGTSTP und SIGINT um den Faktor 10 abgebremst werden können
Die ACE Variante ist deutlicher verboser, da hier auf Time- und Signalevents mit den entsprechenden CB und Signalhandler reagiert wird. Insbsondere sorgt der TimerDispatcher für das explizite Feueren der Timeevents.
Durch die schedule Methode des Timers bzw. die register_handler des Reactors werden die Handler registriert. Während der CB Handler auf Timeevents mit handle_timeout reagiert, reagiert der Signalhandler auf Signalevents mit handle_signal. Beide Eventhandler werden durch die gleichen Verteiler bedient ( select ). Die ACE_Timer_Queue bzw. der konkrete Implementierung ACE_Timer_Heap merkt sich die die zukünftigen Zeitpunkte, zu denen sie expire an den Verteiler schickt.
Die Methode wait_for_event startet die Eventloop, die dann auf die Timer- und Signalevents reagiert.
#include "ace/Timer_Queue.h"
#include "ace/Timer_Heap.h"
#include "ace/Reactor.h"
#include "CB.h"
#include "SignalHandler.h"
#include "TimerDispatcher.h"
int main()
{
CB cb1, cb2;
cb1.setID(1);
cb2.setID(2);
int arg1 = 1, arg2 = 2;
ACE_Timer_Queue *timer_queue;
ACE_NEW_RETURN(timer_queue, ACE_Timer_Heap, -1);
// setup the timer queue
Timer::instance()->set(timer_queue);
ACE_Time_Value curr_time = ACE_OS::gettimeofday();
ACE_Time_Value threeSeconds = curr_time + ACE_Time_Value(3L);
ACE_Time_Value fourSeconds = curr_time + ACE_Time_Value(4L);
// start in 3 seconds, each second
long timerId1= Timer::instance()->schedule(&cb1, &arg1, threeSeconds, ACE_Time_Value(1));
// start in 4 seconds; each 0.3 secondcs
long timerId2=Timer::instance()->schedule(&cb2, &arg2, fourSeconds, ACE_Time_Value(0,300000));
// Strg c
SignalHandler *mutateTimer1= new SignalHandler( timerId1 );
// Strg z
SignalHandler *mutateTimer2= new SignalHandler( timerId2 );
ACE_Reactor::instance()->register_handler( SIGINT, mutateTimer1);
ACE_Reactor::instance()->register_handler( SIGTSTP, mutateTimer2);
// "run" the timer.
Timer::instance()->wait_for_event();
return 0;
}
Dynamische Aspekte
- Die Applikation registriert einen konkreten Eventhander beim Reaktor. Der Eventhandler drückt durch sein Implementierung aus, auf welche Art von Events er reagieren will. Typischerweise heißen die hook-Methoden handle_*, wie handle_input, handle_timeout, handle_put,... .
- Durch eine get_handle des konkreten Eventhandlers Methode holt sich der Reaktor den spezifischen Handler.
- Wenn alle Handles registriert sind, startet die Applikaton die Eventloop des Reaktors. Der Reaktor überwacht nun die Menge aller registrierten Handler auf das Eintreffen von indication Events .
- Sobals ein Event auftritt, übergibt der synchrone event demultiplexer die Kontrolle an den Reaktor.
- Der Reaktor benützt die Handles als Schlüssel, um die entsprechenden Eventhandler aufzurufen (demultiplex) und auf die hook Methode auszurufen (dispatch).
- Die spezifische Methode des Eventhandler bearbeitet die Anfrage direkt auf dem Handle.
Sichten des Programmierers
- Reactor Pattern:
Anwendungsentwickler
Ich will netzwerktransparent wissen, wie lange die Design Pattern Runde noch dauert? Oder ein bißchen formaler:
Implementiere einen Server, der auf eine Browser Anfrage ( HTTP-GET ) ein html Seite schickt, die die verbleibende Zeit bis zum Ende der Design Pattern Runde darstellt. Noch formaler
Client macht eine HTTP-GET Request
Server nimmt Request an und dispatcht sie auf den Event Handler
Event Handler schickt die Antwort zum Client Dazu müssen drei Schritte als Anwendungsentwickler und Nutzer der Reaktor Struktur implementiert werden.
Ich verwende den
BaseHTTPServervon Python.
Request Handler implementieren
class RequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
import datetime
actTime= datetime.datetime.now()
endTime= datetime.datetime( 2007,5,22,9)
diffSeconds= (endTime-actTime).seconds
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write("""<?xml version="1.0" ?>
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN"
"http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<head>
<title>Wie lange dauerts noch?</title>
<script src="https://scwiki.science-computing.de/twiki/pub/TWiki/ChecklistPlugin/itemstatechange.js"
language="javascript" type="text/javascript"></script></head>
<body>
<h1 align="center"><font size="10" color="#FF0000">Count down</font></h1>
<p align="center"> <font size="6"> %s Sekunden noch bis zum Ende der Design Pattern Runde.</font> </p>
</body>
</html>""" %(str(diffSeconds)))
Request Handler registrieren
srvr = HTTPServer(("",4711),RequestHandler)
Event loop laufen starten
srvr.serve_forever()
Frameworkentwickler
Ich will ein Reaktor Framework in Python entwickeln, das auf Input Events reagiert. Exemplarisch werden vier verschiedene Typen von Input Events gleichzeitig prozessiert und die entsprechenden Nachrichten in ähnlichnamige Dateien ins home geschrieben.
- stdin
- jede stdin Eingabe erzeugt ein Event, das zur Prozessierung des Events führt
- lesen einer Datei
- nach dem Einlesen der Datei wird der Handler wieder entfernt
- lesen einer url
- nach dem Einlesen der Internetressource wird der Handler wieder entfernt
- HTTP - GET Anfrage
- bei jedem stdin Event will ich wissen, wie lange die Design Pattern Runde noch dauert; dazu müssen periodisch folgende Schritte vollzogen werden
- registriere den Event Handler, um den ReactorPattern Server zu fragen
- darstellen des Ergebnisses auf stderr
- deregistrieren des Event Handlers
##################
#Event Handlers
##################
import os
import socket
import sys
import time
class EventHandler:
def handle_input(self,device):
raise NotImplementedError
def handle_output(self,device):
raise NotImplementedError
def handle_exception( self,device):
raise NotImplmentedError
def getHandle( self ):
return NotImplementedError
class InputEventHandler( EventHandler ):
def __init__(self,device,dest):
self.handleDevice= device
self.outFile=open(dest,"a")
firstLine= "Read from %s with handle %s \n" % ( device.name , device.fileno() )
self.outFile.writelines([firstLine,"\n"])
def handle_input(self,device):
inp= device.readline().strip()
self.outFile.write( inp + "\n" )
self.outFile.flush()
myReactor.registerHandler( InputToStderrEventHandler( urllib.urlopen("http://ackbar:4711"),
os.getenv("HOME")+"/test.WieLangNoch" ) , "r")
def getHandle( self ): return self.handleDevice
class FileInputEventHandler( InputEventHandler ):
def __init__(self,device,dest):
self.handleDevice= device
self.outFile=open(dest,"a")
name=""
try:
name= device.name
except:
name= device.geturl()
firstLine= "Read from %s with handle %s \n" % ( name , device.fileno() )
self.outFile.writelines([firstLine,"\n"])
def handle_input( self, device ):
for line in device.readlines():
self.outFile.write( line.strip() + "\n" )
self.outFile.flush()
Reactor().removeHandler( self ,"r" )
class InputToStderrEventHandler( FileInputEventHandler ):
def handle_input( self, device ):
for line in device.readlines():
self.outFile.write( line.strip() + "\n" )
if line.startswith("<p align"):
message= line.split(">")[2].split("<")[0]
sys.stderr.write( message )
self.outFile.flush()
Reactor().removeHandler( self,"r")
#############
# Reactor
#############
class Singleton(object):
def __new__(cls, *args, **kwds):
it = cls.__dict__.get("__it__")
if it is not None:
return it
cls.__it__ = it = object.__new__(cls)
it.init(*args, **kwds)
return it
def init(self, *args, **kwds):
pass
import select
class Reactor( Singleton ):
readHandles={}
writeHandles={}
exceptHandles={}
def registerHandler( self, eventHandler,eventTypes):
handle= eventHandler.getHandle()
handleId= handle.fileno()
if "r" in eventTypes: Reactor.readHandles[handleId]= (handle,eventHandler)
if "w" in eventTypes: Reactor.Reactor.writeHandles[handleId]= (handle,eventHandler)
if "e" in eventTypes: Reactor.Reactor.exceptHandles[handleId]= (handle,eventHandler)
def removeHandler( self, eventHandler ,eventTypes ):
handle= eventHandler.getHandle()
handleId= handle.fileno()
if "r" in eventTypes: del Reactor.readHandles[handleId]
if "w" in eventTypes: del Reactor.writeHandles[handleId]
if "e" in eventTypes: del Reactor.exceptHandles[handleId]
def handleEvents( self):
while ( 1 ):
rHandle, wHandle,eHandle= select.select( Reactor.readHandles.keys(), Reactor.writeHandles.keys(),
Reactor.exceptHandles.keys() )
print "all ready handle: Reactor.readHandles: %s Reactor.writeHandles: %s Reactor.exceptHandles: %s "
%(rHandle,wHandle,eHandle)
for han in rHandle:
handleDevice= Reactor.readHandles[han][0]
eventHandler= Reactor.readHandles[han][1]
eventHandler.handle_input( handleDevice )
for han in wHandle:
handleDevice= Reactor.writeHandles[han][0]
eventHandler= Reactor.writeHandles[han][1]
eventHandler.handle_output( handleDevice )
for han in eHandle:
handleDevice= Reactor.exceptHandles[han][0]
eventHandler= Reactor.exceptHandles[han][1]
eventHandler.handle_exception( handleDevice )
print "Reactor:handleEvents: waiting for input "
import urllib
myReactor=Reactor()
myReactor.registerHandler( InputEventHandler( sys.stdin ,
os.getenv("HOME")+"/test.stdin" ) ,"r")
myReactor.registerHandler( FileInputEventHandler( open("/etc/services"),
os.getenv("HOME")+"/test.services" ) ,"r")
myReactor.registerHandler( FileInputEventHandler( urllib.urlopen("http://www.heise.de"),
os.getenv("HOME")+"/test.heise" ) , "r")
myReactor.handleEvents()
Aspekte des Reaktor Frameworks
Event Handler Interface festlegen
class EventHandler:
def handle_input(self,device):
raise NotImplementedError
def handle_output(self,device):
raise NotImplementedError
def handle_exception( self,device):
raise NotImplmentedError
def getHandle( self ):
return NotImplementedError
Reaktor implementieren
import select
class Reactor( Singleton ):
readHandles={}
writeHandles={}
exceptHandles={}
def registerHandler( self, eventHandler,eventTypes):
handle= eventHandler.getHandle()
handleId= handle.fileno()
if "r" in eventTypes: Reactor.readHandles[handleId]= (handle,eventHandler)
if "w" in eventTypes: Reactor.Reactor.writeHandles[handleId]= (handle,eventHandler)
if "e" in eventTypes: Reactor.Reactor.exceptHandles[handleId]= (handle,eventHandler)
def removeHandler( self, eventHandler ,eventTypes ):
handle= eventHandler.getHandle()
handleId= handle.fileno()
if "r" in eventTypes: del Reactor.readHandles[handleId]
if "w" in eventTypes: del Reactor.writeHandles[handleId]
if "e" in eventTypes: del Reactor.exceptHandles[handleId]
def handleEvents( self):
while ( 1 ):
rHandle, wHandle,eHandle= select.select( Reactor.readHandles.keys(), Reactor.writeHandles.keys(),
Reactor.exceptHandles.keys() )
print "all ready handle: Reactor.readHandles: %s Reactor.writeHandles: %s Reactor.exceptHandles: %s "
%(rHandle,wHandle,eHandle)
for han in rHandle:
handleDevice= Reactor.readHandles[han][0]
eventHandler= Reactor.readHandles[han][1]
eventHandler.handle_input( handleDevice )
for han in wHandle:
handleDevice= Reactor.writeHandles[han][0]
eventHandler= Reactor.writeHandles[han][1]
eventHandler.handle_output( handleDevice )
for han in eHandle:
handleDevice= Reactor.exceptHandles[han][0]
eventHandler= Reactor.exceptHandles[han][1]
eventHandler.handle_exception( handleDevice )
print "Reactor:handleEvents: waiting for input "
Um ein Eventhandler zu registrieren wird zusätzlich der Event Typ benötigt, für den sich der Event Handler interessiert.
Durch die Registratur des Eventhandler ist es möglich, über den Filehandle ( z.B.: stdin = 0 ) sowohl das Fileobject, die Eventsource und den Eventhandler, die Anwendungslogik zu erhalten.
In handle Events bedient sich der Reaktor dem nativen select Befehl um auf relevante Events registrieren zu können. handleEvents stellt die Eventloop des Reaktor dar, die, einmal gestartet, immer auf eingehende Events lauscht.
Ausgabe, abhängig von den registrierten Event Handles
Erst durch die Eingabe der Zahl 4 wird die Eventloop prozessiert. Ein Eingabe Event
Reactor.readHandles: [0]liegt nun vor.
Die Ressourcen file und url sind beim Starten der Eventloop registriert, daher werden sie sofort prozessiert
all ready handle: Reactor.readHandles: [4, 6] Reactor.writeHandles: [] Reactor.exceptHandles: [] . Da ich sie explizit deregistriere sind sie bei in dem Eintreten eines stdin-Events nicht mehr vorhanden
all ready handle: Reactor.readHandles: [0] Reactor.writeHandles: [] Reactor.exceptHandles: []. Der Reaktor prozessiert nun nur noch den Filedescriptor 0, also stdin.
- stdin, Datei , Url und HTTP-GET Request werden registriert
all ready handle: Reactor.readHandles: [4, 6] Reactor.writeHandles: [] Reactor.exceptHandles: []
Reactor:handleEvents: waiting for input
4
all ready handle: Reactor.readHandles: [0] Reactor.writeHandles: [] Reactor.exceptHandles: []
Reactor:handleEvents: waiting for input
all ready handle: Reactor.readHandles: [4] Reactor.writeHandles: [] Reactor.exceptHandles: []
49661 Sekunden noch bis zum Ende der Design Pattern Runde.Reactor:handleEvents: waiting for input
Die stdin Abfrage registriet nun einen neuen Eventhandler, der die mir die Frage beantwortert: Wie lange dauert noch die Design Pattern Runde? Dieser Request erhält wieder den Filedescriptor 4.
all ready handle: Reactor.readHandles: [4] Reactor.writeHandles: [] Reactor.exceptHandles: []
Implementierung
Die Implementierung des Reaktor Patterns lässt sich in zwei Schichten unterteilen. Die Frameworkschicht, die die applikationsunabhängige demultiplex/dispatch Infrastruktur zur Verfügung stellt und die Applikationschicht, die die konkreten Eventhandler liefert. In der klassischen, einfachsten Form, geschieht das ganz Eventhandling in einem Prozeß.
Definiere das Event Handler Interfaces
Die Methoden des Event Handlers legen das Servcie-Interface des Reaktor Frameworks fest.
- bestimme den Typ des dispatchingZiels
- verwende eine Event Handler Objekt oder eine Event Handle Funktion
- bestimme die Event Handling dispatching Strategie
- dispatch auf eine einzelne Methode
...
virtual void handlle_event( Handle handle, Event_Type et)= 0;
...
- dispatch auf mehrere Methoden
...
virtual void handle_input( Handle handle )=0;
virtual void handle_output( Handle handle )=0;
virtual void handle_timeout( Handle handle )=0;
...
- das eine Methode Interface erlaubt es einfach, das Framework um neue Eventtypen zu erweitern
- während bei handle_event die ganze Verteilungsstrategie mittels Bedingungen auf Applikationsebene definiert werden muß, geschieht der dispatch auf dem reichhaltigeren Interface automatisch auf Frameworkebene
- insbesondere ist bei feingranularen Dispatch möglich, spezielle hook Methoden in konkreten Event Handlern zu überschreiben
Definiere das Reaktor Interface
Die Applikation nutzt einerseits das Reaktor Interface um die spezifischen Event Handler zu de/registrieren und andererseits die Event Loop zu starten. Gerne wird das Reaktor Interface als Singleton implementiert, das die Anfragen an die Reaktor Implementierung delegiert. Neben dem Event Handler erhält erhält die register_handler als zweites Argument den Event Type als Argument, für den sie sich interessiert.
void Select_Reactor_Implementation::register_handler( EventHandler* eventHandler, Event_Type event_type )
Implementiere das Reaktor Interface
- entkopple das Reaktor Interface von seiner Implementierung durch eine Brücke mehrere verschiedene Implementierung können unterstützt werden ( select, poll, WaitFormMultipleObjects , GuiEventLoops ,... )
- wähle einen synchronen event demutliplexer aus
int select( u_int max_handle_plus_1 ,
fd_set *read_fds, fd_set * write_fds, fd_set *except_fds,
timeval *timeout);
- implementiere ein demultiplexing table
- ein Eintrag soll von der Form < handle, event_handle, indication event type > sein, wobei handle als Schlüssel für den Event Handler bei einem indication event ( connect, expire, read,... ) verwendet wird
- definiere die Reaktor Implementierung
# select Server( only demultiplexing )
def get_request(self):
while self._running:
log.info('select on listen socket')
# demultiplex
rs, ws, es = select.select([self.socket], [], [], 10)
if rs:
log.info('accepting new connection')
# socketobject and address
return self.socket.accept()
log.info('ending request loop')
return (None, None)
Bestimme die Anzahl der Reaktoren, die man benötigt
- in der Regel sollte der Reaktor ein Singleton sein, jedoch erlaubt win32 nur 64 Handles pro Reaktor
- aus Echtzeitforderungen kann es nötig sein mehrere Reaktoren gleichzeitig laufen zu lassen; Trennung der Reaktoren nach Eventtypen
Implementiere die konkreten Eventhandler
- sie stellen die Anwendungslogik dar, im Gegensatz zu dem bisher vorgestellten Reaktor-Framework
- statte die Eventhandler gegebenfalls mit einem Zustand aus; vgl. Beispiel Timer mit ACE
- implementiere die Eventhandler Funktionalität
Kritik
- Vorteile
- klare Trennung von Framework- und Applikationslogik
- Modularität von eventgetriebenen Anwendungen durch verschieden Eventhandler
- Portabilität durch Trennung von Interface und Implementierung des Reaktors
- einfache Parallelität durch den synchronen event demutliplexers
- Nachteile
- setzt einen event demultiplexer voraus
- Durchsatzprobleme bei lang laufenden Event Handler in single Threaded Applikation, den der Event Handler blockiert den Reaktor
- schwierig zu debuggen und zu testen durch die inversion of control
Verwendete Patterns - Techniken
- ObserverPattern
- der Event Handler wird informiert, sobald ein für ihn spezifisches Event auftritt
- BridgePattern
- der Reaktor hält sich eine spezifische Reaktor Implementierung, an die er die Aufrufe delegiert
- TemplateMethodePattern
- die handle_* Methoden als hook Methoden werden wohl in statisch typisierten Programmiersprachen in einer definierten Reihenfolge prozessiert
- double Dispatch: registerHandler getHandle
Weiterlesen...