Der Event in der Menge

Oft gibt es starke Unterschiede zwischen der abstrakten, ersten Sicht auf ein Problem und der endgültigen Implementierung. Gerade bei mehreren, weitgehend unabhängigen Akteuren ist das Beschreiben einer Aufgabenstellung durch Rollen oder Objekte sowie deren Kommunikation sinnvoll. Aktionen können dabei nebenläufig sein, und die Abhängigkeiten der Objekte beschränken sich auf die Kommunikation untereinander. Allerdings ist eine Implementierung solcher Modelle selten möglich oder erfordert einen relativ hohen Aufwand, etwa bei der Verwendung von Threads oder dem Verschicken von Nachrichten an beliebige Objekte, wenn die benutzte Sprache auf strenger Typprüfung besteht.

Das Perl Object Environment (POE) stellt für solche Aufgaben eine Umgebung bereit. Nachrichten beziehungsweise Events können an Objekte verschickt werden, in Ansätzen auch zwischen Prozessen oder Rechnern. POE betreibt kooperatives Multi-Tasking auf Ebene von Perl-Funktionen, überwacht IO-Ressourcen und wandelt deren Veränderungen sowie Timer und Signale in Nachrichten um. Zusätzlich gibt es eine Abstraktionsschicht für Ein- und Ausgaben, die zum Beispiel bei stark interaktiven Programmen hilft und die Verwendung von Komponenten einfacher gestaltet. Zum Ausprobieren muss nur POE installiert sein.

Wichtige Releaseschritte gibt es im CPAN, die aktuellen, kleineren auf den POE-Webseiten oder bei Sourceforge. Die Mailing-Liste ist eher ruhig, Auskunft und Hilfe sind dort aber immer zu bekommen.

Events bilden die Grundlage

Grundlage des Moduls ist die Event-Schicht. Sie sorgt unter anderem dafür, dass Events zum Aufruf der ihnen zugeordneten Routinen (Handler) führen. Zu deren Beschreibung dient lediglich ein Name, der bis auf wenige Einschränkungen frei wählbar ist (siehe Kasten ‘Vordefinierte Event-Namen’). Das Verschicken erfolgt entweder synchron (‘call’, mit Rückgabewerten, wie ein normaler Funktionsaufruf) oder asynchron (‘post’), und dies immer nur an eine bestimmte Session (dazu später mehr). Zwar bezeichnet die POE-Dokumentation Event-Handler als ‘States’, das ist aber in Bezug auf die eigentliche Bedeutung im Zusammenhang mit Automaten verwirrend. Beispiel 1 zeigt die wichtigsten Bestandteile der Event-Schicht. Es besteht aus zwei Sessions: Der Client schickt Strings an den Server, die dieser ausgibt.

#! /usr/bin/perl -w

use POE qw(Session);


#################
# Server Session
#

sub server_spawn {
    my $alias = shift;
    POE::Session->create
        ( 'inline_states' => 
       { '_start' => \&server_start,
         'print' => \&server_print,
         'shutdown' => \&server_shutdown,
       },
          'args' => [ $alias ],
        );
    undef;
}

# Handler für _start-Event
sub server_start {
    my ($kernel, $heap, $alias) = @_[KERNEL, HEAP, ARG0];

    print "Server start\n";
    $heap->{'alias'} = $alias;
    # als Argument nur der Name des Alias
    $kernel->alias_set($alias);
}

# Handler für print-Event
sub server_print {
    my ($string, $postback) = @_[ARG0, ARG1];

    print "Server print: $string\n";
    # ggf. Rückgabewert via postback, 
    # Aufruf der Code-Ref mit eigenen Argumenten
    &{$postback}("printed") if defined $postback;
    return "printed";
}

# Handler für shutdown-Event
sub server_shutdown {
    my ($kernel, $heap) = @_[KERNEL, HEAP];

    $kernel->alias_remove($heap->{alias});
    print "Server shutdown\n";
}
# client session
#

sub client_spawn {
    my $server = shift;

    POE::Session->create
        ( 'inline_states' => 
       { '_start' => \&client_start,
         'result' => \&client_result},
          'heap' => { 'server' => $server },
        );
    undef;
}

# Handler für _start-Event
sub client_start {
    my ($kernel, $session, $heap) = @_[KERNEL, SESSION, HEAP];
    my $server = $heap->{'server'};

    # Parameter fuer post/call Aufrufe:
    #  Ziel-Session, Event-Name, dem Event zugeordnete Argumente
    $kernel->post($server, 'print',
                  'spaeter, da asynchron, ohne Rückgabewert');
    my $result = $kernel->call($server, 'print',
                               'zuerst, da synchron, mit Rückgabewert');
    print "client/call: $result\n";
    # Argumente fuer postback: Event-Namen den die (hier aktuelle) Session
    # erhalten soll, dem Event zugeordnete Argumente
    # zurueckgegeben wird eine Code-Referenz
    $kernel->post($server, 'print', 'zuletzt, mit späterem postback',
                 $session->postback('result', '2. post'));
    $kernel->post($server, 'shutdown');
}

# Handler für result
sub client_result {
    my ($eigene_args, $uebergebene_args) = @_[ARG0, ARG1];

    print "client/postback:\n eigene argumente: ".
      join(', ', @$eigene_args) . 
     "\n vom Aufrufenden uebergebene argumente: ".
       join(', ', @$uebergebene_args) . "\n";
}

#
# Sessions erstellen
server_spawn('server');
client_spawn('server');
#
# Start der POE Event-Loop der Kern-Instanz
$poe_kernel->run();

Pro laufendem POE-Programm gibt es eine Kerninstanz, die Events und Sessions verwaltet. Sie verfügt über eine Event-FIFO und überwacht IO-Ressourcen sowie Timer, für die sie gegebenenfalls Events generiert. Diese gehen durch die FIFO, ohne dass dabei eine Priorisierung einzelner Sessions stattfindet, und der Kernel stellt sie den zuständigen Handlern zu. Eine Ausnahme davon bilden IO-Events, die der Kern unter Umgehung der FIFO sofort abarbeitet. Statt der in POE eingebauten Event-Loop sind auch die von TK, GTK+ oder des Event-Moduls benutzbar. Nach dem Start des Kernels arbeitet dieser nur noch die FIFO ab, solange es noch Sessions gibt. Letztere bekommen die vollständige Kontrolle übergeben. Am Ende des Beispiels geschieht das durch den run()-Aufruf, direkt vorher erstellen Aufrufe von client_spawn() und server_spawn() die initialen Sessions.

Für einen Event sucht der Kern nur in der Session, für die er bestimmt ist, nach einem passenden Handler und ruft diesen auf. Diese Handler sind atomar, ihre Ausführung wird also nicht unterbrochen. Deshalb sollten sie abgeschlossene Aktionen darstellen, zum Beispiel die einzelnen Schritte eines Protokolls. Die Zuordnung zwischen Events und Handlern muss man beim Erstellen einer Session angeben, kann sie aber nachträglich verändern. client_spawn() und server_spawn() übernehmen diese Arbeit im Beispiel. Das Argument %inline_states enthält diese Zuordnungen, die die Schnittstelle der Session nach außen sind. So ordnet es beispielsweise beim Server dem Event ‘print’ die Funktion server_print() zu.

Sessions sind im Prinzip die POE-Objekte. Sie bestehen aus einem ihnen gehörenden Speicher und Event-Handlern. Eine Session ist jeweils Kind derjenigen, die sie erzeugte; es entsteht eine Baumstruktur wie bei Prozessen. Sessions kümmern sich hauptsächlich um ihre eigene Verwaltung, die Kommunikation mit dem Kern und den Aufruf der Event-Handler. Dabei bekommen Letztere den gesamten Laufzeitkontext in @_ übergeben: die Session selbst, ihren Speicher (Heap), den Event, den Kern und die Session, die den Event abschickte, samt deren Argumenten. Zumindestens einen Teil davon verwenden alle Handler im Beispiel. vom Absender des Events übergebene Argumente finden sich ab ARG0 in @_ wieder. Für alle anderen existieren ähnliche Konstanten mit sinnvollen Namen, etwa HEAP für den Heap der Session.

In Zusammenarbeit mit den Sessions führt der POE-Kern eine Garbage Collection durch. In deren Verlauf entfernt er eine Session automatisch, sobald sie weder etwas zu tun hat noch etwas am Leben erhält. Zu diesen lebensverlängernden Dingen gehören die für die Session bestimmten Events in der Kernel-FIFO, ihre Kinder, beobachtete Ressourcen und weitere Referenzen auf die Session, etwa durch Postbacks oder Aliases. In der Praxis ist das nützlich und harmoniert mit Perls eigener Speicherverwaltung. Erzwingt der Kern das Ende einer Session, entfernt er alle erwähnten Ressourcen, Kinder sterbender Sessions werden ihren Großeltern zugeordnet.

Postbacks und Aliases als Referenzen

Postbacks sind Referenzen auf Code, der einen Event samt Argumenten an die Session schickt. Sie erhöhen die Anzahl der Referenzen auf die Session und sorgen so dafür, dass sie aktiv bleibt. Ein Alias erhöht diese auch und ordnet der Session einen String zu, unter dem sie im Weiteren erreichbar ist.

Beides setzt das Beispiel zur Adressierung einer Session ein: Zur Ansprache des Servers benutzt der Client nur dessen Alias; umgekehrt verwendet der Server zur Kommunikation mit der Client-Session den ihm in client_start() übergebenen Postback. POE::Session->create liefert zwar eine Referenz auf die erstellte Session zurück. Ebenso wie alle anderen direkten Referenzen auf Sessions sollte man diese aber nur mit Vorsicht verwenden, da von ihnen die Garbage Collection bei Sessions abhängt. Weitere Möglichkeiten zur Adressierung von Sessions existieren bisher leider nicht.

client_start() zeigt die Unterschiede zwischen dem synchronen call() und asynchronem post(). Letzteres ruft die Routine direkt nach dem Start der Session auf (siehe Kasten ‘Vordefinierte Events’). Nach dem Behandeln des ‘result’-Events des Postback terminiert die Client-Session automatisch, da keine Events oder weiteren Referenzen mehr vorhanden sind.

Die Server-Session bricht schon nach dem Entfernen des Alias im Handler des ‘shutdown’-Events ab. Der Event wurde direkt nach den ‘print’-Events geschickt und steht somit im Event-FIFO vor dem Postback-Event (‘result’). Weitere Informationen, unter anderem über Parameter für POE::Session->create finden sich in der Dokumentation zu den POE-Modulen.

Dreigeteilte Ein- und Ausgabe

Zurzeit besteht die Ein-/Ausgabe-Schicht aus drei Teilen: Wheels, Filter und Driver. Wheels sind die Objekte, mit denen Entwickler direkt arbeiten, Driver sorgen für das systemnahe Schreiben/Lesen und Filter übersetzen Datenströme, meistens aus einem Anwendungsformat in einen Byte-Strom oder umgekehrt. Beispiel 2 zeigt einen Port-Forwarder. Die listen/accept-Session lauscht mit einem Wheel auf einem Port und erstellt für jede neue Verbindung eine weiterleitende Session, die die hereinkommmenden Daten auf einen anderen Port schickt und Daten von dort auch wieder zurück. Von der Kommandozeile aus gestartet, kann man beispielsweise durch telnet localhost 12345 Verbindung zum lokalen FTP-Server aufnehmen: Das Skript setzt 12345 auf dessen Port um.

#! /usr/bin/perl -w

use POE qw(Session Filter::Stream Wheel::ReadWrite Wheel::SocketFactory
           Driver::SysRW);

### host:port auf dem gehoert werden soll
my $bind_host = 'localhost';
my $bind_port = 12345;
### host:port auf den weitergeleitet werden soll
my $dest_host = 'localhost';
my $dest_port = 21;


##################
# listen/accept Session
#
sub la_spawn {
    POE::Session->create
        ( 'inline_states' => 
       { '_start'   => \&la_start,
         'connection' => \&la_connection,
         'failure'    => \&la_failure,
       },
        );
    undef;
}


sub la_start {
    my $heap = $_[HEAP];
    # neue Socket-Factory erstellen
    $heap->{'socketfactory'} = POE::Wheel::SocketFactory->new
      ( 'BindAddress'   => $bind_host,
     'BindPort'      => $bind_port,
     'SuccessState'  => 'connection',
     'FailureState'  => 'failure',
     Reuse => 'yes',
      );
    warn 'created '.$heap->{'socketfactory'}->ID();
}


sub la_connection {
    my ($socket, $remote_addr, 
     $remote_port, $wheel_id) = @_[ARG0..ARG3];

    warn "l/a: neue verbindung $wheel_id mit $remote_addr:$remote_port\n";
    fw_spawn($socket);
}


sub la_failure {
    my ($heap, $operation, $errnum, 
     $errstr, $wheel_id) = @_[HEAP, ARG0..ARG3];
    warn "l/a: socketfactory $wheel_id, $operation: $errnum/$errstr\n";
    delete $heap->{'socketfactory'};
}


##################
# Weiterleitende/forwarding Session
#
sub fw_spawn {
    my $socket = shift;
    POE::Session->create
        ( 'inline_states' => 
       { '_start'     => \&fw_start,
         'connected'  => \&fw_connected,
         'input_dest' => \&fw_input_dest,
        ' input_src'  => \&fw_input_src,
         'failure'    => \&fw_failure,
       },
          'heap' => { 'accepted_socket' => $socket },
        );
    undef;
}


sub fw_start {
    my $heap = $_[HEAP];
    $heap->{'socketfactory'} = POE::Wheel::SocketFactory->new
      ( 'RemoteAddress' => $dest_host,
        'RemotePort'    => $dest_port,
        'SuccessState'  => 'connected',
        'FailureState'  => 'failure',
      );
    warn 'fw: neue socketfactory '.$heap->{'socketfactory'}->ID()."\n";
}


sub fw_connected {
    my ($heap, $socket) = @_[HEAP, ARG0];
    # Wheels fuer Ziel und Quelle
    $heap->{'wheel_dest'} = POE::Wheel::ReadWrite->new
      ( 'Handle'     => $socket,
        'Filter'     => POE::Filter::Stream->new(),
        'Driver'     => POE::Driver::SysRW->new(),
        'InputState' => 'input_dest',
        'ErrorState' => 'failure',
      );
    $heap->{'wheel_src'} = POE::Wheel::ReadWrite->new
      ( 'Handle'     => delete $heap->{accepted_socket},
        'Filter'     => POE::Filter::Stream->new(),
        'Driver'     => POE::Driver::SysRW->new(),
        'InputState' => 'input_src',
        'ErrorState' => 'failure',
      );
    warn 'fw: neue Wheels fŸr Ziel/Quelle: '.
      $heap->{'wheel_dest'}->ID().'/'.$heap->{'wheel_src'}->ID();
}


sub fw_input_dest {
    $_[HEAP]->{'wheel_src'}->put($_[ARG0]);
}


sub fw_input_src {
    $_[HEAP]->{'wheel_dest'}->put($_[ARG0]);
}


sub fw_failure {
    my ($heap, $operation, $errnum, 
     $errstr, $wheel_id) = @_[HEAP, ARG0..ARG3];
    warn "fw: Wheel $wheel_id, $operation: $errnum/$errstr\n";
    foreach (qw(socketfactory wheel_dest wheel_src accepted_socket)) {
        delete $heap->{$_};
    }
}


la_spawn();
$poe_kernel->run();

Wheels sind reine Perl-Objekte; sie empfangen daher Befehle über Methodenaufrufe. Events registrieren und verschicken sie nur im Namen der Session, von der sie erstellt wurden, und sind somit nicht auf andere Sessions übertragbar. Benachrichtigungen über eingetroffene Daten und Fehler schicken sie als Events an ihre Session. Beim Lesen oder Schreiben von Daten blockieren sie nicht.

Filter existieren bisher unter anderem für zeilen-, block- und streambasierte Kommunikation; mit dem Reference-Filter lassen sich serialisierte Perl-Datenstrukturen verschicken und lesen. Filter sind kombinierbar.

Ports per POE-Script weiterleiten

Im Beispiel werden das SocketFactory- und das ReadWrite-Wheel verwendet. Die listen/accept-Session erstellt eine SocketFactory bei ihrem Start. Diese öffnet den angegebenen Port; Event-Namen für neue Verbindungen oder Fehler setzen die Handler für SuccessState und FailureState. la_connection(), der Handler für neue Verbindungen, erstellt nur eine weiterleitende Session für den neuen Socket mit fw_spawn(). Die Fehlerbehandlung in la_failure() beschränkt sich auf Ausgeben einer Meldung und Entfernen des Wheel-Objektes. Da dieses dabei das Beobachten des Sockets aufgibt, hält nichts mehr die Session am Leben, so dass sie terminiert.

Weiterleitende Sessions sehen ähnlich aus. fw_start() benutzt wieder eine Socket-Factory, diesmal aber zum Aufbauen einer Verbindung mit dem Zielrechner und -port. Steht die Verbindung, erstellt fw_connected() jeweils ein ReadWrite-Wheel für die Verbindung zum Ziel und zum Nutzer des Port-Forwarder ($heap->{wheel_dest} beziehungsweise $heap->{wheel_src}). Die hierfür verwendeten Sockets kommen von der Socket-Factory und vom Heap, den fw_spawn() initialisiert hat. Der Stream-Filter gibt Daten unbearbeitet weiter. Bisher ist der hier benutzte SysRW-Driver der einzig existierende. Einen Treiber muss man nur verwenden, wenn das Wheel Daten liest oder schreibt - in la_start() weiter oben erstellte es lediglich einen Socket, so dass der Treiber verzichtbar war.

Zur Verarbeitung ihrer Eingabe-Events benutzen beide Wheels die Handler fw_input_dest() beziehungsweise fw_input_src(). Sie geben die empfangenen Daten, hier die vom Stream-Filter nicht veränderten Stücke, direkt wieder auf das jeweils andere Wheel aus.

Damit leitet die Session alle Daten zwischen Quelle und Ziel weiter. Allerdings unterstützt POE Flusskontrolle beim ReadWrite-Wheel bisher nur bei der Ausgabe. Bei jedem Durchlauf der Event-Loop liest der Kern so viel wie möglich von den Sockets und erzeugt synchrone Eingabe-Events. Das kann leicht zum Aushungern von Events im Event-FIFO führen oder durch nötiges Puffern viel Speicher belegen. Wenn man also von unsicheren Quellen oder zum Beispiel aus Dateien lesen will, sollte man sich die Funktionen select_pause_read() und select_resume_read() des POE-Kernels ansehen und selbst eine Flusskontrolle implementieren. Eine derartige Erweiterung von Wheel::ReadWrite wird es aber hoffentlich bald geben.

Die Fehlerbehandlung ähnelt der der anderen Session; sie beendet die Verbindung und damit die Session. Am Ende des Beispiels findet sich wieder das Erstellen einer initialen Session und das Starten der Event-Loop.

Komponenten erleichtern die Arbeit

Fertige Komponenten gibt es zum Beispiel auf Clientseite für HTTP, IRC, DNS und ICMP-Ping oder auf Serverseite mit dem HTTP-Server aus den POE-Beispielen. Eine weitere interessante Komponente ist IKC (Inter-Kernel Communication). Events können damit über Prozess- und Rechnergrenzen hinweg verschickt werden, Sessions sind per IKC bisher entweder direkt oder über generierte Proxy-Sessions ansprechbar.

Solche auf POE basierenden Komponenten haben einige Vorzüge:

• durch das Verschicken asynchroner Nachrichten lassen sich Aufgaben zum Teil einfacher und übersichtlicher strukturieren;
• gerade bei Ein- und Ausgabe helfen Nebenläufigkeit und eine einheitliche IO-Schicht, sonst braucht man oft entweder umfangreichere Schnittstellen oder mehr Aufwand zum Aufbrechen des linearen Ablaufs;
• es gibt weniger Synchronisationsbedarf als bei der Verwendung von Threads, da einzelne Handler im günstigsten Fall abgeschlossene Aktionen sind;
• der Aufwand für den Einsatz von POE ist gering, sodass es sich auch für Skripts eignet.

POEs Objektschicht, ursprünglich ein Hauptteil, will der Entwickler in Zukunft ausgliedern und separat weiter entwickeln. Obwohl sich viele Aspekte von POE noch verändern und es gelegentlich kleinere Unstimmigkeiten gibt, bemüht sich der Autor, das Modul stabil zu halten. In Zukunft wird es hoffentlich an Popularität gewinnen und sich das Angebot an Komponenten vergrößern. Auf jeden Fall kann man mit POE jetzt schon viele Aufgaben schnell und übersichtlich lösen.

Torvald Riegel
studiert Informatik an der TU Dresden.

(ck)