!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: 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.

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

!block "perl"
!endblock

!block "perl"
!endblock

!block "perl"
!endblock




A1: Module

- Coro (Coroutinen)
- Event (der Event-Standard, nur leider unmaintained)
- Linux::AIO (asynchrone Ein-/Ausgabe)
- Convert::Scalar (für C<weaken>)
- HTTP::Date (code-reuse made easy with CPAN, hat mir sicherlich 5 Minuten gespart)
- Date::Parse (siehe HTTP::Date, ich liebe CPAN ;)
- BerkeleyDB (Caching der Directory-Listings und WHOIS-Requests)