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 sequentiell 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 gleicher Stelle weiterzuarbeiten.

Dies liegt daran, daß 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
daß der Eindruck von gleichzeitigem Ablauf entstehen kann (kooperatives
Multitasking). Der Unterschied zu Continuations ist im wesentlichen der,
daß 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, es wäre viel zu langsam (Kopieren des gesamten
Stacks oder schlimmer?) und kompliziert, um jemals implementiert zu
werden.

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 muß 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 preemptives
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,
daß 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
die Funktion (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 Queue kommen. C<pp_entersub> ist alles andere als trivial, ein
Wunder, daß 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, wobei die
erste schon beim Kompilieren erzeugt wird.

Solange eine Coroutine nicht in eine Funktion springt, die schon rekursiv
aufgerufen wurde, ist das kein Problem, da 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 dem Problem
gleich, die n-te Rekursionsstufe zu beenden während die n+1-te nich 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 der Aufrufstack
hinaufgewandert werden und manuell eine Art "Rücksprung" organisiert werden:

!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 wegsichern kann,
muss die erste neu erzeugt werden, ein höchst umständlicher Vorgang, der
deshalb auch (mit einem Hash) gecached wird.

Damit war bewiesen, daß 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. Dieser 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
natürlich nicht da sind. dies ist auch der Grund dafür, daß 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, daß sich wirklich an den Standard hielte (von
Grausamkeiten wie IRIX mal ganz abgesehen, dessen Programmierer vom
Programmieren herzlich wenig Ahnung haben (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, daß ich das Modul viel benutzen würde, aber ich wollte zeigen,
daß 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. Ausserdem werden Attribute unterstützt, so,
daß 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

Beim ersten Aufruf wird der Zustand initialisiert, bei jedem weiteren
Aufruf 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 einigermassen 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 Kode-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 eine Referenz
auf ein Array, in dem die "Rücksprung"-Coroutinen gespeichetr 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 Kode-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, muß 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, istd as 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, daß 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 Prozeß mehr am laufen ist, haben wir einen
Deadlock, zumindest, solange man Signale außer acht läßt. 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 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.

Interessanter ist, daß C<async> den C<$manager> in Bereitschaft versetzt
und nicht nur den neuen Prozess. Der einzige Grund dafür ist, daß so die
meisten Prozesse mit einem definierten Stackpointer erzeugt werden, so
daß 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
in 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 automatishc 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 ein Prozess
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ändig 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 von wartenden
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, ansonsten reiht sich der aktuelle Prozess in die
Warteschlange und legt sich schlafen.

H1: Teil 3: Die Aussenwelt

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 maintainte und auf vielen Systemen
nicht mehr lauffähige) Event-Modul, das jedoch Callback erfordert.

Das C<Coro::Event>-Modul benutzt Event (oder ein anderes Modul, es gibt
nur keines ;), statt einem Callback 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 garnicht, da es die Event-Objekte intern
recycled 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, daß 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, daß 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 Eriegnis 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 sehr 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 viel zu viel 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 garnicht 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 (manche jedoch tun das), sie prüfen
lediglich auf einen gewünschten Zustand, ist dieser shcon erreicht, wird
ein "Ereignis" ausgelöst, auch, wenn eigentlich nichts passiert ist.

Und aus der Abteilung Kurioses: Wenn ein Filehandle in einen String
gewandelt wird, 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 auch 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 = new Listen-Socket;

 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
- das 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 Serevr 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, daß ich Perl-5.6-Features für XS
verwende. Ein Port auf ältere Perls sollte nicht schwer 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.2
Committed:	Wed Jan 16 03:23:37 2002 UTC (24 years, 9 months ago) by root
Branch:	MAIN
Changes since 1.1:	+523 -100 lines
Log Message:	* empty log message *