docs/pws2002/coro.sdf

!init OPT_STYLE="paper"

!define DOC_NAME        "Coroutinen und Continuations mit dem Coro-Modul"
!define DOC_AUTHOR      "Marc Lehmann <pcg@goof.com>"
!build_title

H1: Was sind Coroutinen und Continuations?

Prozedurale Programme laufen normalerweise sequenziell ab und springen
manchml in Unterprogramme/Funktionen. Es ist nicht möglich, aus einem
Unterprogramm heraus kurzzeitig in den Aufrufer zurückzuspringen und später
an derselben Stelle weiterzuarbeiten.

Dies liegt daran, dass Perl, wie viele andere Sprachen, Funktionsaufrufe
mit einem Stack implementieren, den man in der gleichen Reihenfolge
abbauen muss, wie man ihn aufbaut.

Continuations abstrahieren diesen Stack, d.h. es gibt nicht nur eine
Aufruffolge, sondern viele. Zwischen diesen kann an beliebigen Stellen
gewechselt werden. Exception-handling, nichtlokale Sprünge, Multitasking
und Threads können mit Continuations implementiert werden.

Coroutinen sind eine Anwendung von Continuations, bei denen meherere
unabhängige Ablaufinstanzen nacheinander kurzzeitig ausgeführt werden, so
dass der Eindruck gleichzeitigen Ablaufs entstehen kann (kooperatives
Multitasking). Der Unterschied zu Continuations ist im Wesentlichen der,
dass der Wechsel normalerweise nicht explizit erfolgt und keine Parameter
zwischen den Instanzen ausgetauscht werden, da die Reihenfolge nicht
festgelegt ist.

H1: Die Entstehung des Coro-Moduls

Continuations für Perl5 stehen seit ca. 1996 auf meinem "TODO"-Zettel,
aber da ich immer dachte, man könne das nicht schnell genug (Kopieren des
gesamten Stacks oder schlimmer?) implementieren bzw. die Implementation
wäre viel zu komplex um jemals stabil zu laufen.

H1: Das Problem

Um Coroutinen zu implementieren, reicht es, zwei Primitiven zu schreiben
(die Namen stammen aus Modula-2, die Perl-Syntax sieht etwas anders aus):

 NEWPROCESS(process,coderef)
 TRANSFER(oldprocess,newprocess)

C<NEWPROCESS> erzeugt einen neuen "Prozess", wobei ein "Prozess" im
Wesentlichen einen Aufrufstack (zum Schachteln von Funktionsaufrufen)
und einen Startpunkt (C<coderef>) besitzt. Diese Daten werden in einer
Struktur (C<process>) abgelegt.

Um einen solchen Prozess zu starten muss man die Kontrolle an ihn abgeben,
indem man C<TRANSFER> mit einer leeren C<process>-Struktur als erstes
Argument und einer gültigen (z.B. von C<NEWPROCESS>) C<process>-Struktur
als zweites Argument aufruft.

C<TRANSFER> sichert nun die Daten des {{aktuellen}} Prozesses in
C<oldprocess> und "springt" in den neuen.

Damit lassen sich Continuations und kooperatives und präemptives
Multitasking implementieren, wenn man es zu nutzen weiß.

H1: Teil 1: Implementation

H2: Erste Erfolge und erste Probleme

Im Juli 2001 war es dann so weit: In einem Anflug von Leichtsinn dachte
ich mir, "ein paar Pointer-Austauscher könnten doch zum Ziel führen".

(Es folgen einige haarsträubende Implementationsdetails, wer will, kann
die überspringen und direkt zum interessanten Teil übergehen ;)

Und tatsächlich, wenige Stunden (und Zeilen) später hatte ich ein Modul,
das die beiden obigen Primitiven implementiert. Es funktionierte sogar:

!block C
 save_state(pTHX_ Coro__State c, int flags)
 {
   c->curstackinfo = PL_curstackinfo;
   c->curstack = PL_curstack;
   c->mainstack = PL_mainstack;
   c->stack_sp = PL_stack_sp;
   c->op = PL_op;
   ...
   c->curcop = PL_curcop;
   c->top_env = PL_top_env;
 }

 load_state(pTHX_ Coro__State c)
 {
   PL_curstackinfo = c->curstackinfo;
   PL_curstack = c->curstack;
   PL_mainstack = c->mainstack;
   PL_stack_sp = c->stack_sp;
   PL_op = c->op;
   ...
   PL_curcop = c->curcop;
   PL_top_env = c->top_env;
 }
!endblock

Bis einmal zwei Coroutinen dieselbe Funktion aufriefen. Viele
Stunden später kam dann das "Aha! So macht Perl das also. Na so ein
Mist...". Leider reicht es nicht, einfach die vielen Stacks auszutauschen,
die Perl benutzt. Denn Perl modifiziert bei jedem Aufruf einer Funktion
diese (genauer: den CV) selbst:

!block C
  /* aus pp_entersub */
  CvDEPTH(cv)++;
  if (CvDEPTH(cv) < 2)
      (void)SvREFCNT_inc(cv);
  else {        /* save temporaries on recursion? */
      PERL_STACK_OVERFLOW_CHECK();
      if (CvDEPTH(cv) > AvFILLp(padlist)) {
         /* hier wird eine neue padlist erzeugt - 40 Zeilen Magie */
         ...
      }
  }
  ...
  PL_curpad = AvARRAY((AV*)svp[CvDEPTH(cv)]);
!endblock

Da wurde mir auch klar, weshalb Threads so langsam sind, denn die müssen
unglaubliche Anstrengungen unternehmen, damit sich einzelne Threads nicht
in die Quere kommen. C<pp_entersub> ist alles andere als trivial, ein
Wunder, dass Perl so schnell ist.

Im Wesentlichen geht es darum, für jede Rekursionsstufe eine C<padlist>
(in der werden lokale Variablen u.ä. gespeichert) zu erzeugen, die erste
schon beim Kompilieren.

Solange eine Coroutine nicht in eine Funktion springt, die schon rekursiv
aufgerufen wurde, ist das kein Problem, da C<CvDEPTH> dann 1 ist und die
erste padlist leer. Andernfalls wird die Rekursionsstufe erhöht, auch
kein Problem.  Wenn die erste Coroutine dann aber wieder aus der Funktion
zurückkehren will, gibt es ein Problem, denn das käme der Aufgabe
gleich, die n-te Rekursionsstufe zu beenden während die n+1-te noch immer
aktiv ist.

Die Lösung ist eklig, aber noch nicht unzumutbar langsam: Da ein
Perl-Modul nicht die Freiheit hat, C<pp_entersub> zu verändern (die
Perl-Thread-Implementation kann sich das leisten), muss man den Aufrufstack
hinaufgewandern und manuell eine Art "Rücksprung" organisieren.

!block C
 save_state(pTHX_ Coro__State c, int flags)
 ...
    /* this loop was inspired by pp_caller */
    for (;;)
      { 
        while (cxix >= 0)
          {
            PERL_CONTEXT *cx = &ccstk[cxix--];
            
            if (CxTYPE(cx) == CXt_SUB)
              {
                CV *cv = cx->blk_sub.cv;
                if (CvDEPTH(cv))
                  {
                    EXTEND (SP, CvDEPTH(cv)*2);

                    while (--CvDEPTH(cv))
                      {
                        /* this tells the restore code to increment CvDEPTH */
                        PUSHs (Nullsv);
                        PUSHs ((SV *)cv);
                      }

                    PUSHs ((SV *)CvPADLIST(cv));
                    PUSHs ((SV *)cv);

                    get_padlist (cv); /* this is a monster */
                  }
              }
        ...
        if (top_si->si_type == PERLSI_MAIN)
          break;

        top_si = top_si->si_prev;
        ccstk = top_si->si_cxstack;
        cxix = top_si->si_cxix;
      }
 ...

 load_state(pTHX_ Coro__State c)
 ...
    /* now do the ugly restore mess */
    while ((cv = (CV *)POPs))
      {
        AV *padlist = (AV *)POPs;
  
        if (padlist)
          {
            put_padlist (cv); /* mark this padlist as available */
            CvPADLIST(cv) = padlist;
          }
  
        ++CvDEPTH(cv);
      }
!endblock

Der lustigste Teil verbirgt sich hinter:

 get_padlist (cv); /* this is a monster */

Denn, während man die neu erzeugten padlists einfach sichern kann, muss
die erste neu erzeugt werden, ein höchst umständlicher Vorgang, der
deshalb auch (mit einem Hash) gecached wird.

Damit war bewiesen, dass Coroutinen tatsächlich so viele Probleme machen,
wie ich befürchtet hatte.

H2: Das vorläufige Ende

Aber es kam noch schlimmer. Beispiel: Perl ruft eine XS-Funktion auf, und
diese wieder einen Perl-Callback. Der führt einen Coroutinenwechsel
durch und die neue Coroutinen macht ein C<return> - wohin? In die
XS-Funktion, die natürlich Ergebnisse auf dem Stack erwartet, die ebenso
natürlich nicht da sind. dies ist auch der Grund dafür, dass der "fake
threads"-Ansatz von Perl (vorerst) zum Scheitern verurteilt ist.

An diesem Punkt habe ich erstmal aufgegeben. Die Aussicht, auch eine
portable Coroutinenimplementation für C zu finden, waren Null, sie werden
ja noch nicht einem in irgendwelchen Standards erwähnt (dachte ich).

Naja, selbst schreiben war angesagt - tatsächlich kann man, lediglich
mit UNIX95-Funktionen, portabel Coroutinen erzeugen. Leider ist mir
kein System bekannt, dass sich wirklich ganz an den Standard hielte
(von Grausamkeiten wie IRIX mal ganz abgesehen, siehe sigaltstack,
das vollkommen dokumentiert das vollkommen sinnlose macht), aber die
Unterschiede sind gering. Sogar unter der eingeschränkten Win32-API kann
man durch direktes herumpoken in C<jmp_buf> relativ schmerzlos eine
Implementation bekommen.

Im Coro-Modul gibt es im Verzeichnis C<Coro/libcoro> seitdem eine
relativ portable, sehr kleine (366 Zeilen inkl. Dokumentation ;)
Coroutinenimplementation für C.

Das Coro-Modul erzeugt entweder für jede Coroutine in Perl eine Coroutine
in C, oder optional nur "on demand", d.h. wenn sich der C-Stackpointer
verändert hat, eine gefährliche, aber in der Praxis gut funktionierende
Heuristik.

H1: Teil 2: Implementation von Coroutinen und Continuations in Perl

H2: Die Perl-API

C<NEWPROCESS> und C<TRANSFER> werden vom C<Coro::State>-Modul implementiert.
C<NEWPROCESS> sieht in Perl so aus:

!block perl
 # eine "leere" Coroutine, bereit, gefüllt zu werden
 $main = new Coro::State;

 # eine neue Coroutine, zu der man wechseln kann
 $new = new Coro::State sub {
    # ... coroutine
 }, @args;
!endblock

Und C<TRANSFER> sieht so aus:

!block perl
 # etwas asymetrisch:
 $main->transfer($new);
 # oder symmetrischer
 Coro::State::transfer($main,$new);
!endblock

H2: Continuations

Nicht, dass ich das Modul viel benutzen würde, aber ich wollte zeigen,
dass es geht: Das C<Coro::Cont>-Modul exportiert zwei Funktionen, C<csub>,
mit dem sich neue Continuations erzeugen lassen und C<yield>, das so
ähnlich wie C<return> wirkt. Außerdem werden Attribute unterstützt, so,
dass man Continuations so schreiben kann:

!block perl
 use Coro::Cont;

 sub generator : Cont {
    my $state = time;
    while() {
       $state = $state * 31231 % 65531;
       yield $state % $_[0];
    }
 }
!endblock

Der erste Aufruf initialisiert den Zustand (Zuweisung an C<$state>), bei
jedem weiteren wird eine (schlechte) Zufallszahl im Bereich C<0..$_[0]>
zurückgeliefert, wobei @_ sich jedesmal ändert. Das Attributsystem von
Perl5 ist ziemlich schwer zu benutzen (mehrere Module können es sich nicht
einfach teilen, Attribute sind automatisch global etc..), weshalb ich
darauf nicht näher eingehe ;)

Die Implementation ist einigermaßen trickreich, da die Standardimplementation
von C<transfer> das Array C<@_> mitspeichert, es also nicht so einfach übergeben
werden kann. Deshalb kann man C<transfer> ein drittes Argument, eine Bitmaske
mit Flags, übergeben, die angibt, ob C<@_> oder C<$_> u.ä. Teil des
"Prozesszustandes" werden oder nicht.

C<csub> erzeugt zwei C<Coro::State>-Objekte, eines für den Aufrufer und eines
für die Continuation:

!block perl
 sub csub(&) {
    my $code = $_[0];
    my $prev = new Coro::State;

    my $coro = new Coro::State sub {
       # we do this superfluous switch just to
       # avoid the parameter passing problem
       # on the first call
       &yield;
       &$code while 1;
    };
!endblock

D.h. die Code-Referenz wird in einer Schleife aufgerufen. Da die Parameter
beim ersten Aufruf der Coroutine etwas anders übergeben werden als
später bei C<yield>, wird als erstes ein C<yield> ausgeführt, damit alle
Aufrufe identisch sind.

Als nächstes wird die neu erzeugte Coroutine {{einmal}} aufgerufen (damit
sie "yield"-en kann):

!block perl
   push @$$return, [$coro, $prev];
   &Coro::State::transfer($prev, $coro, 0);
!endblock

In der gobalen Variablen C<$return> befindet sich eine Referenz
auf ein Array, in dem die "Rücksprung"-Coroutinen gespeichert sind,
d.h. C<Coro::Cont> macht fast den gleichen "Fehler" wie C<pp_entersub>,
aber C<$$return> ist nicht wirklich global (wie ich gleich zeige ;)

Als letztes wird eine Code-Referenz zurückgegeben, die das Aufrufen der
Coroutine für uns erledigt:

!block perl
   return sub {
      push @$$return, [$coro, $prev];
      &Coro::State::transfer($prev, $coro, 0);
      wantarray ? @_ : $_[0];
   };
!endblock

Auch hier wird die "Rücksprungaddresse" für C<yield> in C<@$$return>
gespeichert und dann die Coroutine aufgerufen, die beim nächsten C<yield>
zurückkehrt. Die Argumente, die die Coroutine an C<yield> übergeben hat,
stehen immer noch in C<@_>.

Die Funktion C<yield> ist in XS implementiert, die
Prototyp-Implementation in Perl sah jedoch so aus:

!block perl
 sub yield(@) {
    &Coro::State::transfer(@{pop @$$return}, 0);
    wantarray ? @_ : $_[0];
 }
!endblock

Auch nichts Überraschendes, es wird zum Aufrufer zurückgegeben. Beim
nächsten Aufruf wird das neue C<@_> zurückgegeben, d.h. der Aufruf
müsste lauten: C<@_ = yield ...>, und das ist der Grund, weshalb es
in XS implementiert wurde, denn in XS kann man das C<@_> des Aufrufers
verändern. Wäre nicht dieser Umstand. so würde auch C<yield> in reinem
Perl implementiert sein.

Das verbleibende Problem ist die globale Variable C<$return>. Tatsächlich
ist sie ein C<Coro::Specific>-Objekt und ist "Coroutinen"spezifisch, d.h.
kann in jeder Coroutine (hier meine ich ein C<Coro>-Objekt und nicht das
C<Coro::State>-Objekt) einen anderen Wert besitzen.

H2: C<Coro::Specific>

Mit C<Coro::State> kann man eigene Prozess-Abstraktionen basteln. Damit
C<Coro::Specific> funktioniert, muss man lediglich einen
coroutinen-/prozessspezifischen Hash in C<$Coro::current> speichern, und
genau das macht das C<Coro>-Modul.

C<Coro::Specific->new> gibt eine Referenz auf einen ge'tie'ten Scalar zurück, der
wiederum bei C<FETCH> und C<STORE> einen Skalar im Array
C<@{$Coro::current->{specific}}> verändert. Solange das C<Coro>-Modul nicht
geladen ist, ist das immer derselbe, danach ist er coroutinenspezifisch.

!block perl
 my $idx;

 sub new {
    my $var;
    tie $var, Coro::Specific::;
    \$var;
 }

 sub TIESCALAR {
    my $idx = $idx++;
    bless \$idx, $_[0];
 }

 sub FETCH {
    $Coro::current->{specific}[${$_[0]}];
 }

 sub STORE {
    $Coro::current->{specific}[${$_[0]}] = $_[1];
 }
!endblock

Auf diese Weise kann man in anderen Modulen ohne schlechtes Gewissen
C<Coro::Specific> verwenden, da weder C<Coro::State> noch C<Coro> geladen
sein müssen.

H2: Prozesse selbst gemacht

Der nächste Schritt, nach Continuations, Variablen und (primitiven)
Coroutinen sind richtige Prozesse. Diese werden im C<Coro>-Modul
implementiert, das Prozesserzeugung, Kommunikation, Prioritäten etc.
anbietet.

Das sieht dann so aus:

!block perl
 use Coro:

 async {
    # ein neuer "Prozess"
 };

 my $coro = async {
    terminate;
 };

 $coro->prio(PRIO_MAX);
 $coro->join;

 cede; # gib die CPU frei für andere Prozesse

 # usw.. hat ja "jeder" schon mal gesehen ;-]
!endblock

Die Implementation ist in einigen Teilen in XS, der Geschwindigkeit
wegen. Wenn das Modul geladen wird, erzeugt es zuerst einen
"Hauptprozess", der für das Hauptprogramm steht (das Hauptprogramm
korrekt auf eine Coroutine abzubilden, ist sowieso ein sehr schwieriges
Unterfangen - welche Coroutine DESTROY't die anderen Coroutinen am
Programmende, wenn das Hauptprogramm nicht mehr existiert?) und sorgt für
die Übernahme des bisherigen C<@{$Coro::current->{specific}}>.

!block perl
 our $main = new Coro;

 # maybe some other module used Coro::Specific before...
 if ($current) {
     $main->{specific} = $current->{specific};
 }  

 our $current = $main;
!endblock

Dann folgt die Erzeugung eines Idle-Prozesses, der Rechenzeit verbrät,
wenn kein Prozess mehr läuft:

!block perl
 our $idle = new Coro sub {
     print STDERR "FATAL: deadlock detected\n";
     exit(51);
 };
!endblock

Jaja, reingefallen. Wenn kein Prozess mehr läuft, haben wir einen
Deadlock, zumindest, solange man Signale außer acht lässt. Das ändert
sich, wenn man Events benutzt (z.B. mit C<Coro::Event>), weshalb der
Idle-Prozess auswechselbar ist.

Neue Prozesse werden mit C<async> oder C<Coro->new> erzeugt:

!block perl
 sub _newcoro {
    terminate &{+shift};
 }

 sub new {
    my $class = shift;
    bless {
       _coro_state => (new Coro::State $_[0] && \&_newcoro, @_),
    }, $class;
 }  

 sub async(&@) {
    my $pid = new Coro @_;
    $manager->ready; # this ensures that the stack is cloned from the manager
    $pid->ready;
    $pid;
 }
!endblock

Ein Prozess ist also wenig mehr als ein nackter Hash, in dessen
C<_coro_state>-Slot ein C<Coro::State>-Objekt gespeichert ist. Ähnlich
wie PDL sieht C<transfer> in diesem Slot nach einem C<Coro::State>-Objekt.

Dass C<async> den C<$manager> in Bereitschaft versetzt und nicht nur den
neuen Prozess, ist nicht auf Anhieb verständlich: Der Grund dafür ist,
dass so die meisten Prozesse mit einem definierten Stackpointer erzeugt
werden, so dass derselbe C-Stack für alle Prozesse verwendet werden kann,
zumindest anfänglich.

Am Ende eines Prozesslebens steht ein C<terminate>, oder ein C<cancel>:

!block perl
 sub cancel {
    push @destroy, $_[0];
    $manager->ready;
    &schedule if $current == $_[0];
 }
!endblock

C<schedule> springt direkt in den Scheduler, der den nächsten
lauffähigen Prozess aus der Warteschlange nimmt und zum aktuellen
macht. Er ist (jaja, "the need for speed") in XS implementiert, sieht aber
etwa (keine Prioritäten) so aus:

!block perl
 sub schedule {
    my $previous = $current;
    $current = shift @runqueue;
    Coro::State::transfer($current, $previous);
 }
!endblock

Der aktuelle Prozess wird nicht automatisch wieder aufgerufen, dazu muss
man ihn wieder mit der C<ready>-Methode in die Warteschlange setzen. Um
nur kurz andere Prozesse laufen zu lassen, konnte man früher C<yield>
benutzen, aber das ging schon für Continuations drauf, und Damian Conway
fand das wichtiger. Also musste ich einen neuen Namen finden:

!block perl
 sub cede {
    $current->ready;
    schedule;
 }
!endblock

Einen Prozess direkt zu zerstören, ist keine gute Idee. Zum einen könnte
er mehrfach in der C<ready>-Queue stehen, zum anderen sollte er sich
selbst C<DESTROY>en können, aber da er sich dabei selbst den C-Stack
wegzieht, muss dies von einem anderen Prozess aus stattfinden. Der
Manager-Prozess stellt genau dies sicher (und verständigt auch gleich
eventuell wartende Prozesse):

!block perl
 my @destroy;
 my $manager;
 $manager = new Coro sub {
    while() {
       # by overwriting the state object with the manager we destroy it
       # while still being able to schedule this coroutine (in case it has
       # been readied multiple times. this is harmless since the manager
       # can be called as many times as neccessary and will always
       # remove itself from the runqueue
       while (@destroy) {
          my $coro = pop @destroy;
          $coro->{status} ||= [];
          $_->ready for @{delete $coro->{join} || []};
          $coro->{_coro_state} = $manager->{_coro_state};
       }
       &schedule;
    }
 };
!endblock

H2: Signale, Semaphoren, Channels und Locks

Hat man Prozesse, die man erzeugen, anhalten, schedulen und wieder beenden
kann, so muss man sie koordinieren. Zum Beispiel mit C<Coro::Signal>'s:

!block perl
 sub send {
    if (@{$_[0][1]}) {
       (shift @{$_[0][1]})->ready;
    } else {
       $_[0][0] = 1;
    }
 }

 sub wait {
    if ($_[0][0]) {
       $_[0][0] = 0;
    } else {
       push @{$_[0][1]}, $Coro::current;
       Coro::schedule;
    }
 }
!endblock

Getreu dem Motto "Pseudo-Hashes müssen sterben!", etwas ungewöhnlich
ausgelegt, verwendet C<Coro::Signal> ein Array und keinen Hash. Das erste
Element ist nur ein Flag, das angibt, ob das Signal gesendet wurde (falls
niemand darauf wartet). Das zweite Element ist ein Array wartender
Prozessen.

C<send> weckt den ersten wartenden Prozess auf oder - falls es keinen gibt
- setzt das Flag. Wenn man auf ein Signal wartet, wird zuerst geprüft, ob
es schon aufgetreten ist, sonst reiht sich der aktuelle Prozess in die
Warteschlange und legt sich schlafen.

H1: Teil 3: Die Außenwelt

H2: C<Event> für Coro: C<Coro::Event>

Multitasking innerhalb von Perl ist ja ganz nett, aber das erste

!block perl
   my $line = <STDIN>;
!endblock

bringt das ganze System zum Erliegen. Wenn man viele asynchrone
Ein-/Ausgabekanäle hat, die in unterschiedlicher Reihenfolge bedient
werden sollen, braucht man eine Ereignissteuerung. Das Standardmodul für
diesen Fall ist das (leider nicht mehr gewartete und auf vielen Systemen
nicht mehr lauffähige) Event-Modul, das jedoch für jedes Ereigenis einen
Callback erfordert.

Das C<Coro::Event>-Modul benutzt Event (oder ein anderes Modul, es gibt
nur keines ;), statt eines Callbacks wird der Prozess solange blockiert,
bis das Ereignis eintritt:

!block perl
 my $event = Coro::Event->timer(after => 1, interval => 1, hard => 1);

 while() {
    $event->next;
    print "one event every second\n";
 }
!endblock

C<next> gibt dabei ein Event-Objekt zurück, ähnlich dem Event-Objekt,
das Callbacks des Event-Moduls erhalten. Eine frühe Implementation von
C<next> gab tatsächlich genau dieses Objekt zurück:

!block perl
 # dies ist der Callback, der jedem Event-Watcher zugewiesen wird:
 sub std_cb {
    # das private-Feld enthält ein Array,
    # das im ersten Element den wartenden Prozess
    # und im zweiten das Event-Objekt speichert
    my $w = $_[0]->w;
    my $q = $w->private;
    $q->[1] = $_[0];
    if ($q->[0]) { # somebody waiting?
       $q->[0]->ready;
       &Coro::schedule;
    } else {
       $w->stop;
    }
 }

 sub next {
    my $w = $_[0];
    my $q = $w->private;
    if ($q->[1]) { # event waiting?
       $w->again unless $w->is_cancelled;
    } elsif ($q->[0]) {
       croak "only one coroutine can wait for an event";
    } else {
       local $q->[0] = $Coro::current;
       &Coro::schedule;
    }
    pop @$q;
 }
!endblock

Dies war nicht nur relativ ineffizient (zwei Prozesswechsel pro Ereignis),
das Event-Modul mochte das auch gar nicht, da es die Event-Objekte intern
wiederverwendet und Zugriffe darauf nach Rückkehr aus dem Callback recht
zufällig bearbeitete.

Die Lösung für beide Probleme ist wieder einmal XS. Zum Glück
exportiert C<Event> eine fast dokumentierte Schnittstelle dafür (wie auch
Coro eine XS-Schnittstelle bereitstellt). C<Coro::Event> speichert die
Event-Daten in einem Array, wobei nur die Werte für C<hits>, C<prio> und
C<got> gesichert werden, da kein Event mehr als diese benötigt.

Durch Events kann es passieren, dass alle Prozesse auf ein Ereignis warten, also muss
der Idle-Prozess überschrieben werden:

!block perl
 $Coro::idle = new Coro sub {
    while () {
       Event::one_event; # inefficient
       Coro::schedule;
    }
 };
!endblock

C<one_event> ist nicht sehr effizient, weshalb man das Hauptprogramm am
besten durch einen Aufruf von C<Event::loop> zur Hauptschleife degradiert
und der Idle-Prozess nie laufen muss.

Um nicht bei jedem Ereignis zwei Prozesswechsel durchführen zu müssen,
legt der Callback die wartenden Prozesse nur in die Ready-Queue. Dabei
ergibt sich das Problem, dass Event keine dokumentierte Möglichkeit
bietet, {{vor}} dem Aufruf von C<poll> oder C<select> (oder was auch
immer intern verwendet wird) einen Callback auszuführen, der die
bereitstehenden Prozesse ausführt.

Zwar kann man mit:

!block perl
 Event->add_hooks(prepare => ..);
!endblock

Einen Callback für diesen Fall einbauen, nur wird er zu spät aufgerufen,
es gehen also Ereignisse verloren. Ausserdem quittiert Event einige
Aufrufe innerhalb dieses Callbacks mit Segmentation Faults.

Die Lösung liegt in einem speziellen Idle-Watcher mit der Priorität
C<PRIO_NORMAL>. Idle-Watcher mit dieser Priorität werden kurz vor dem
Aufruf von C<poll> aufgerufen, andere Prioritäten führen zu zu häufigen
oder gar keinen Aufrufen.

Wenn ein Ereignis auftritt, wird mit C<$watcher->now> dieser Watcher
"gestartet", der dann solange Prozesse scheduled bis keine mehr laufen:

!block perl
 #define NEED_SCHEDULE if (!do_schedule)                                 \
                         {                                               \
                           do_schedule = 1;                              \
                           GEventAPI->now ((pe_watcher *)scheduler);     \
                         }

 scheduler_cb(pe_event *pe)
 {
   while (CORO_NREADY)
     CORO_CEDE;
     
   do_schedule = 0;
 } 

 # overwrite the ready function
 void 
 ready(self)
         SV *    self
         PROTOTYPE: $
         CODE:
         NEED_SCHEDULE;
         CORO_READY (self);
!endblock

Wenn das Hauptprogramm artig C<Event::loop> benutzt, läuft dieser
Callback immer im Kontext des Hauptprogramms, der Idle-Prozess wird nicht
benutzt.

H2: Non-Blocking but Blocking Filehandles

Damit man das Ganze sinnvoll (und natürlich) nutzen kann, braucht man
non-blocking-IO. Das Ziel ist es, Programme so schreiben zu können:

!block perl
 # Ein finger-Prozess, der andere Prozesse nicht blockiert
 sub finger {
    my $user = shift;
    my $host = shift;

    my $fh = new Coro::Socket PeerHost => $host, PeerPort => "finger"
       or die;

    print $fh "$user\n";

    print "$user\@$host: $_" while <$fh>;
    print "$user\@$host: done\n";
 }
!endblock

Dazu muss man Filehandles tie'en, ein komplizierter Prozess, da
Filehandles in Perl sehr merkwürdige Objekte, und vor allem keine Objekte
sind. Dies erledigt das C<Coro::Handle>-Modul, das im Wesentlichen eine
Funktion C<unblock> exportiert, die aus einem normalen Filehandle einen
nichtblockierenden macht:

!block perl
 my $stdin = unblock \*STDIN;

 while(<$stdin>) {
    ...
 }
!endblock

Anders als andere Module (C<Data::Location>) werden dabei keine
echten Filehandles erzeugt, da dies zu viele Probleme
mit sich bringt. Stattdessen wird ein anonymer GLOB an das
C<Coro::Handle::FH>-Package getied, um C<syswrite>, C<readline> etc.
abzufangen, zusätzlich wird der GLOB in das C<Coro::Handle>-Package
ge'bless'ed, um auch Methoden abzufangen:

!block perl
  my $fh = shift or return;
  my $self = do { local *Coro::Handle };

  my ($package, $filename, $line) = caller;
  $filename =~ s/^.*[\/\\]//;

  tie $self, Coro::Handle::FH, fh => $fh, desc => "$filename:$line", @_;

  my $_fh = select bless \$self, $class; $| = 1; select $_fh;
!endblock

In C<Coro::Handle> stehen dann "glue"-Methoden:

!block perl
 sub readline    { tied(${+shift})->READLINE(@_) }
 sub read        { Coro::Handle::FH::READ  (tied ${$_[0]}, $_[1], $_[2], $_[3]) }
 sub sysread     { Coro::Handle::FH::READ  (tied ${$_[0]}, $_[1], $_[2], $_[3]) }
 sub syswrite    { Coro::Handle::FH::WRITE (tied ${$_[0]}, $_[1], $_[2], $_[3]) }
 sub print       { Coro::Handle::FH::WRITE (tied ${+shift}, join "", @_) }
 sub printf      { Coro::Handle::FH::PRINTF(tied ${+shift}, @_) }
 sub fileno      { Coro::Handle::FH::FILENO(tied ${$_[0]}) }
 sub close       { Coro::Handle::FH::CLOSE (tied ${$_[0]}) }
!endblock

Und einige weitere, die nur mit C<Coro::Handle>s Sinn machen:

!block perl
 sub readable    { Coro::Handle::FH::readable(tied ${$_[0]}) }
 sub writable    { Coro::Handle::FH::writable(tied ${$_[0]}) }
!endblock

Die eigentliche Implementation ist dann in C<Coro::Handle::FH> zu finden,
z.B. für C<WRITE>:

!block perl
 sub writable {
    ($_[0][6] ||= Coro::Event->io(
       fd      => $_[0][0],
       desc    => "$_[0][1] W",
       timeout => $_[0][2],
       poll    => W+E,
    ))->next->{Coro::Event}[5] & W;
 }

 sub WRITE {
    my $len = defined $_[2] ? $_[2] : length $_[1];
    my $ofs = $_[3];
    my $res = 0;

    while() {
       my $r = syswrite $_[0][0], $_[1], $len, $ofs;
       if (defined $r) {
          $len -= $r;
          $ofs += $r;
          $res += $r;
          last unless $len;
       } elsif ($! != Errno::EAGAIN) {
          last;
       }
       last unless &writable;
    }

    return $res;
 }
!endblock

Ja, die Hässlichkeit (C<READLINE> habe ich noch gar nicht gezeigt ;) ist
ein Grund mehr, das Ganze zu abstrahieren. Die eigentliche Arbeit wird von
der C<writable>-Methode erledigt, die beim ersten Aufruf einen Watcher
erzeugt und dann mit C<next> auf das nächste Ereignis wartet.

Am Rande muss ich noch bemerken, dass "Ereignis" nicht ganz korrekt
ist. Weder C<select> noch C<poll> warten auf "Ereignisse", noch tun
das die meisten anderen Event-Modelle, sie prüfen lediglich auf einen
gewünschten Zustand, ist dieser schon erreicht, wird ein "Ereignis"
ausgelöst, auch, wenn eigentlich nichts passiert ist.

Und aus der Abteilung Kurioses: Beim Wandeln eines Filehandle in einen
String wird offenbar die C<FETCH>-Methode aufgerufen.

H2: Von Handles zu Sockets

Da Dateien niemals "blockieren" und Perl kein AIO unterstützt, ist
C<Coro::Handle> eigentlich nur für Pipes und Sockets interessant. Damit
man auch ohne blockieren C<connect>en kann, gibt es mit C<Coro::Socket>
ein Modul, das sich so nah wie möglich an die C<IO::Socket::INET>-API
hält.

Der Konstruktor benutzt die folgende Methode für ein
non-blocking-connect:

!block perl
 sub _prepare_socket {
    my ($class, $arg) = @_;
 
    socket my $fh, PF_INET, $arg->{Type}, _proto($arg->{Proto})
       or return;
    
    $fh = bless Coro::Handle->new_from_fh($fh, timeout => $arg{Timeout}), $class
       or return;

    ...
 }

 sub new {
    ...
       for (@sa) {
          $fh = $class->_prepare_socket(\%arg)
             or return;

          $! = 0;

          if ($fh->connect($_)) {
             next unless writable $fh;
             $! = unpack "i", $fh->getsockopt(SOL_SOCKET, SO_ERROR);
          }

          $! or last;

          $!{ECONNREFUSED} or $!{ENETUNREACH} or $!{ETIMEDOUT} or $!{EHOSTUNREACH}
             or return;
       }
    ...
 }

 sub connect     { connect tied(${$_[0]})->[0], $_[1] or $! == Errno::EINPROGRESS }
!endblock

H1: Beispiele

H2: Ein alter Traum: LWP non-blocking

Wenn man Coro für sonst nichts braucht, auch in "normalen"
Programmen kann man Coro dazu benutzen, LWP (blockiert bei Anfragen) und
Ereignisgesteuerte Programme (z.B. Tk) zu kombinieren. Dazu muss man
"lediglich" den C<IO::Socket::INET>-Konstruktor überschreiben:

!block perl
 use Coro::Socket;
 use IO::Socket::INET;
 sub IO::Socket::INET::new {
    shift; new Coro::Socket @_;
 };
!endblock

Danach kann man loslegen und gleichzeitig mehrere LWP-Requests starten:

!block perl
 use Coro;
 use Coro::Event;
 use LWP::Simple;

 $SIG{PIPE} = 'IGNORE';

 async {
    get "http://.../";

 };

 async {
    get "http://.../";
 };

 loop;
!endblock

H2: Ein Prototyp für einen Server

Ein "herkömmlicher" Server sieht im Pseudocode so aus:

!block perl
 my $socket = <neue listen-Socket erzeugen>

 while () {
    my $fh = $socket->accpt;
    if (fork == 0) {
       serve_request;
       exit;
    }
 }
!endblock

Mit Coroutinen sieht das genauso aus, nur verwendet man statt C<fork> ein
C<async>:

!block perl
 sub handle_connection {
    my $fh = $_[0];
    while (<$fh>) {
       ...
    }
 }

 my $port = new Coro::Socket
         LocalPort => "imap2",
         ReuseAddr => 1,
         Listen => 1,
    or die;

 async {
    slog 1, "accepting connections";
    while () {
       my $fh = $port->accept;
       slog 3, "accepted $fh on $port";
       async \&handle_connection, $fh;
       undef $fh;
    }
 };

 loop;
!endblock

Die Vorteile sind - neben einem geringeren Verbrauch an Systemressourcen
- dass die einzelnen für die Clients zuständigen Prozesse untereinander
kommunizieren können. Mein Webserver z.B. kann Informationen über Clients
austauschen, mein IMAP-Server kann problemlos die asynchronen "neue Mail
empfangen"-Nachrichten an angeschlossene Clients weitergeben. Man kann
solche Server auch problemlos zusammen in denselben Prozess laden (dazu
muss man nur eines der beiden C<loop>-Statements entfernen).

A1: Verfügbarkeit

Das Coro-Modul gibts auf CPAN - getestet wurde es bisher nur unter
Linux/x86, IRIX und Cygwin, wobei es unter Cygwin wohl noch einige
Probleme gibt, die ich aber nicht debuggen kann. Perl-5.6 ist
Minimalversion, was daran liegt, dass ich Perl-5.6-Features für XS
verwende. Ein Port auf ältere Perls sollte nicht schwierig sein.

A2: Interessante URLs

Etwas Off-Topic, aber durchaus lesbar:

* Etwas Geschichte (Achtung, lustig!): http://brics.dk/~cw97/ProceedingS/01.ps.gz
* Subcontinuations: http://indiana.edu/~dyb/papers/subK.ps


Revision:	1.5
Committed:	Sun Feb 10 16:39:35 2002 UTC (24 years, 7 months ago) by root
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.4:	+1 -1 lines
Log Message:	* empty log message *