docs/event.sdf

!init OPT_STYLE="paper"

!define DOC_NAME           "Einführung in Event::"
!define DOC_TYPE           "[Vortrag]"
!define DOC_AUTHOR         "(c) 2000 Marc Lehmann <pcg@goof.com>"
!build_title

!block abstract

Wenn viele Jobs parallel ausgeführt werden sollen, eignet sich das
bekannte fork-Paradigma von Unix nicht mehr: Die Interprozeßkommunikation
und der Mehraufwand an Speicher und Ressourcen überwiegt dir Vorteile der
einfacheren Programmstruktur bei weitem. Diese kurze Einführung in die
Ereignis-gesteuerte Programmierung in Perl zeigt an einem konkreten Beispiel
(News-Scanner), wie einfach sich selbst komplexe Strukturen in Perl
realisieren lassen.

!endblock

H1: C<Event> in der Praxis --- oder wie man 500 Newsserver gleichzeitig scannt.

H2: Ereignis-gesteuerte Programmierung?

Zur Lösung paralleler ablaufender Prozesse sind heute drei Ansätze
gebräuchlich:

* Prozesse mit getrenntem Adressraum (z.B. mit C<fork>)
* eng gekoppelte Prozesse mit gemeinsamen Adreßraum (z.B. mit {{1:pthreads}})
* Ereignis-gesteuerte Prozesse

Jeder dieser Ansätze hat verschiedene Vor- und Nachteile: Das
C<fork>-Modell ist sehr einfach zu programmieren und eignet sich
besonders für einfache Probleme, die sozusagen in kleine "Stückzahlen"
anfallen. Durch die Abschottung der Prozesse wird eine einfach
Parallelisierung möglich, da die Prozesse (z.B.) auf unterschiedlichen
Rechnern arbeiten können. Größter Nachteil ist die relative aufwendige
Interprozeßkommunikation, die einen großen Overhead nach sich ziehen
kann.

{{1:Threads}} werden vielfach als das Mittel der Wahl angesehen. Der größte
Vorteil von Threads ist das Vorhandensein mehrere Ablauf-Instanzen,
die getrennt blockieren können. Leider werden Threads in den meisten
Fällen nur dazu mißbraucht, das Blockieren des gesamten Prozesses zu
verhindern (z.B. wenn Daten nicht sofort zur Verfügung stehen), werden
also effektiv nur als Krücke für asynchrone-EA verwendet. Diesen Vorteil
erkauft man sich durch eine zwar schnelle aber dafür extrem komplizierte
Synchronisation innerhalb der Threads. {{1:Threads sind in in den seltensten
Fällen die richtige Wahl für ein Problem.}}

Ereignis-gesteuerte Programmierung beruht auf dem
{{1:Callback}}-Prinzip: Eine zentrale Anlaufstelle innerhalb des Prozesses
wartet auf Ereignisse (engl. "Events", also z.B. "Daten angekommen",
"Zeit abgelaufen" etc...). Je nach Ereignis werden entsprechende
Callback-Funktionen aufgerufen. Der Vorteil dieses Ansatzes ist eine
übersichtliche Programmstruktur, eine extreme schnelle Kommunikation
(nur ein Prozeß) und ein ressourcenschonendes Endprodukt. Auch dieser
Ansatz hat seine Nachteile. Der größte ist, daß man bei vielen Problemen
"Umdenken" muß, da sich Callbacks eben keine lineare Programmstruktur
verwirklichen läßt ({{1:Closures}} können dabei jedoch helfen). Außerdem
muß man sich bewußt sein, das ein blockierender Funktionsaufruf
(z.B. C<read>) das gesamte Programm anhält.

H2: Das Problem...

{{...}}ist oberflächlich betrachtet, recht einfach: Eine (kleine) Menge
von Usenet-Servern soll nach Newsgruppen abgesucht werden. Das kann
auf faire Weise geschehen: man öffnet eine NNTP-Verbindung und schickt
Requests. Dies läßt sich durch Pipelining (senden mehrere Befehle
gleichzeitig) beschleunigen. Durch die Zeiten, die der News-Server
benötigt um Artikel zu suchen, wird die Datenrate in der Praxis allerdings
drastisch beschränkt.

Also die unfaire Weise: statt einer öffnet man 5, 10 oder gleich mehrere
hundert Verbindungen zu einem (oder mehreren) Servern und verteilt so die
Verbindungslatenz und die Antwortzeit.

H2: Die Planung

Die (für mich) naheligende Idee, dies mit mehreren Scanprozessen zu
implementieren, scheiterte an zwei Problemen:

* Die Scanprozesse müssen sich untereinander absprechen, um Duplikate zu
vermeiden. Dies ist zwangsläufig Interprozeßkommunikation (z.B. über eine
SQL-Datenbank), die sehr aufwendig zu implementieren ist. Hinzu kommt, das
einzelne Jobs zuerst markiert werden müssen ("in Arbeit"), damit sie nicht
von mehreren Prozessen gleichzeitig bearbeitet werden, was jedoch sehr
schwierig ist, wenn man Wert darauf legt, Prozesse beliebig abbrechen zu
können, ohne Artikel zu verlieren.

* Das Zielsystem, ein Pentium-166-System, hat weder unendlich Rechen-
noch Speicherressourcen.  Da Perl von beidem gerne viel nimmt, wäre die
Sättigung schon bei relativ wenigen Verbindungen erreicht. Stichwort Speicher:
jeder Prozeß benötigt einen Interpreter, eine Kopie der
libc-Variablen, eine eigene Kopie des Scanprogramms und seine eigene
SQL-Anbindung. Auch moderne Systeme mit einem effizienten C<fork()> leiden
darunter, da gerade Perl nicht zimperlich mit dem Speicher umgeht.

Die Lösung (klar!) lag im Event-Modul. Da alle Verbindungen von einem
Prozeß bearbeitet werden, gibt es keine Synchronisationsprobleme. Der
Overhead pro Verbindung beschränkt sich ebenfalls auf einen Hash, und das
umschalten von Prozessen entfällt ebenfalls (schneller).

H1: Die Implementation

Die folgenden Abschnitte stellen die wichtigsten "Knotenpunkte" des
Scanprogrammes vor. Jedesmal wird kurz das Problem erläutert und die
Lösung mit Hilfe des C<Event>-Moduls diskutiert.

H2: Der "Scheduler"

Der komplizierteste Teil des Programmes ist der Scheduler: Er verteilt
einzelne Jobs auf die Scanner, bzw. beendet das Programm, wenn alle Jobs
abgearbeitet wurden. Es gibt nur zwei Typen von "Jobs":

* 'S': {{1:S}}canne eine Gruppe. Der Scanner sucht eine bestimmte
Newsgruppe auf dem Server und stellt mit Hilfe einer SQL-Tabelle fest,
welche Artikel(-nummern) noch nicht gescannt wurden.

* 'A': {{1:A}}rtikel holen. Da Gruppen Tausende von Artikeln enthalten
können, wird nur ein "Job" pro Gruppe erzeugt. Ein Scanner sucht sich eine
Artikelnummer aus, bearbeitet sie und legt die restlichen wieder zurück in
die Warteschlange.

Ein "Scanner" ist dabei kein Prozeß, sondern nur eine Instanz der
C<Scanner>-Klasse, in der im wesentlichen der Zustand einer Verbindung
gespeichert wird (Server, Port, aktuelle Gruppe...). Für jede potentielle
Verbindung wird ein solches Objekt erzeugt. Für hundert Verbindungen sieht
das z.B. so aus:

>new Scanner for 1..100;

Die Objekte reihen sich automatisch in die C<idle>-Warteschlange ein.

Beim Programmstart werden alle Server- und Gruppen aus einer Datei gelesen und in die Job-Warteschlange eingefügt. Dann wird
in die Hauptschleife gesprungen:

>Scanner::loop();       # Hauptschleife

!block perl
sub loop {
   while (@queue || @idle < $scanners) {
      runq;
      Event::loop;
   }
}
!endblock

Dabei stehen die zu bearbeitenden Jobs in C<@queue> und die verfügbaren
Scanner-Objekte in C<@idle>. Solange noch Jobs vorhanden sind (C<@queue !=
0>) und nicht alle (C<$scanners>) Scanner idlen, wird C<runq> aufgerufen
und in die Hauptschleife von C<Event> gesprungen.

C<runq> (das steht für "run queue") nimmt Jobs aus der Warteschlange und
teilt sie verfügbaren Scannern zu. Der Algorithmus ist sehr primitiv
(FCFS) und könnte wesentlich verbessert werden. Wichtig ist, daß die
Lastverteilung in diesen wenigen Zeilen stattfindet und sehr gut
lokalisiert und damit sehr einfach änderbar ist.

!block perl
sub runq {
   while (@queue && @idle) {
      my $c = pop @queue;
      my $s = pop @idle;
      $s->run(@$c);
   }
   Event::unloop_all unless @queue || @idle < $scanners;
}
!endblock

Der Aufruf von C<unloop_all> beendet alle Event-Schleifen, wenn alle Jobs abgearbeitet wurden.

H2: Job Management & Rescheduling

Um neue Jobs in das System einzufügen, gibt die Funktion C<add_job>:

!block perl
sub add_job {
   push @queue, [@_];
   $reschedule->start if @idle;
}
!endblock

Die wichtigste Teil ist der Aufruf von {{C:$reschedule->start}}: Wenn
ein Scanner verfügbar ist (C<@idle> nicht leer ist), muß der Scheduler
aufgerufen werden. Da der Aufruf von C<add_job> sehr häufig ist, und
der Scheduler (C<loop>) eine Rekursion bedeutet, wird er nicht direkt
aufgerufen, sondern nur, wenn sonst keine Ereignisse anliegen. Dies wird
mit einem C<idle>-Event-Handler erreicht, der in der globalen Variable
C<$reschedule> steht:

!block perl
my $reschedule = Event->idle(
   desc => "reschedule hook",
   max => 5,
   cb => sub {
      $_[0]->w->stop;
      Event::unloop;
   }
);
$reschedule->stop;
!endblock

{{C:Event->idle}} ist der {{Konstruktor}}, der einen Ereignis-Handler
vom Typ "idle" erzeugt. Die einzelnen Attribute bedeuten:

!block table
Attribut   Beschreibung
desc       Eine Beschreibung, z.B. für das C<NetServer::ProcessTop>-Modul.
max        Zeit (in Sekunden) nach dem der Callback {{auf jeden Fall}} ausgeführt wird.
cb         Die Callback-Funktion, die aufgerufen wird.
!endblock

Übertragen auf den C<$rescheduler> bedeutet dies, daß aus der
Event-Schleife gesprungen wird, wenn gerade kein Datentransfer oder
sonstige Aufgaben anliegen, {{oder nach fünf Sekunden}}, je nachdem,
was früher eintrit. Diese Einschränkung verhindert, das ein schnell
eintreffender Artikel den gesamten Prozeß "am Laufen hält" und damit
verhindert, das freie (idle) Scanner nicht mit neuen Jobs versorgt werden.

Wenn der Callback angesprungen wird, bekommt er ein {{Ereignis-Objekt}}
übergeben (unter X entspricht dies einem C<XEvent>, bei Gtk ist es ein
C<Gdk::Event>). Als erstes sucht er über dieses Ereignis-Objekt (in
C<$_[0]>) den ursprünglichen {{Watcher}} ({{C:$_[0]->w}}, "w" steht für
"watcher") und ruft die C<stop>-Methode auf.  Damit wird erreicht, daß
der Callback nicht mehr aufgerufen wird, bis er das nächste mal gestartet
wird (z.B. in C<add_job>). {{C:$_[0]->w->stop}} ist übrigens das gleiche
wie {{C:$rescheduler->stop}}, die Variable C<$rescheduler> ist wegen C<my>
jedoch erst {{nach}} dem Aufruf des Konstruktors sichtbar.

Das zweite (und wichtigste) was der Callback unternimmt, ist, den
eigentlichen Scheduler wieder anzuspringen C<loop>.  In C<loop> wurde die
Hauptschleife des Event-Moduls aufgerufen (C<Event::loop>): C<unloop> ist
das Gegenstück dazu und springt aus dieser Schleife heraus, so daß der
Scheduler neue Jobs verteilen kann.

H3: Beendigung eines Jobs

Wenn ein Scanner-Objekt einen Job verarbeitet hat, muß es sich wieder in
die C<@idle>-Queue eintragen:

!block perl
sub idle {
   my $self = shift;
   push @idle, $self;
   $reschedule->start;
}
!endblock

der Aufbau gleicht C<add_job>.

H2: Die Jobschleife

Für die Abarbeitung der Jobs ist die Methode C<run> zuständig. Sie hat
mindestens drei Parameter: C<self> (das Scanner-Objekt), C<host> (der
NNTP-Server, inkl. Port) und C<cmd> (der Jobtyp).

Da das NNTP-Protokoll "stateful" ist, muß der aktuelle NNTP-Server und
die aktuelle Gruppe gespeichert werden. Gilt der neue Job für denselben
Rechner und dieselbe Gruppe (der Normalfall) passiert nichts, ansonsten
wird die Verbindung zum NNTP-Server neu aufgebaut, bzw. die Gruppe
gewechselt.

Das Aufbauen der NNTP-Verbindung ist ein Problem für den Event-Ansatz: ein
C<connect>-Aufruf {{blockiert}} den Prozeß, bis entweder die Verbindung
steht oder ein Fehler passiert. Da ein solcher C<connect> einige Sekunden
benötigen kann (bei Netzwerkproblemen auch wesentlich länger), müssen sog.
"non-blocking-calls" verwendet werden.

Das ist auch der Grund, weshalb das Programm auf Standardmodule wie
C<IO::Socket> oder C<Net::NNTP> verzichten muß: Unterstützung für
nicht-blockierende Aufrufe ist kaum oder überhaupt nicht vorhanden. Das
C<Net::NNTP>-Modul ist in dieser Hinsicht besonders schlecht, dnen
man kann die entsprechende Methoden nicht einfach in einer Subklasse
überschreiben.

Der schwierigste Teil war der Aufruf von C<connect>, der ebenfalls nicht
blockieren sollte:

!block perl
if (socket $fd, PF_INET, SOCK_STREAM, getprotobyname 'tcp') {
   sub TCP_NODELAY(){1} sub SOL_TCP(){6}; # linux-2.2
   setsockopt $fd, SOL_TCP, TCP_NODELAY, 1;
   fcntl $fd, F_SETFL, O_NONBLOCK;
   connect $fd, sockaddr_in $port, inet_aton($ip);
   fcntl $fd, F_SETFL, 0;
} else {
   undef $fd;
}
!endblock

Einige Konstanten (z.B. C<SOL_TCP>) sind in Perl nicht einfach zu
bekommen. Da das Script mehr ein Hack als eine professionelle Anwendung
ist, wurden sie einfach hardcodiert.

Wenn der Server gewechselt wird, wechselt auch der Filehandle, so daß eine neuer
Event-Watcher erzeugt werden muß:

!block perl
($self->{w} = Event->io(fd => fileno $fd, poll => 'r'))->stop;
!endblock

H2: NNTP-Befehle

Das NNTP-Protokoll ist sehr einfach: Kommandos bestehen aus einer
einzelnen Textzeile, Antworten aus einem Zifferncode und einer
beschreibenden Textzeile. Artikel werden als Textblock übertragen, wobei
die letzte Zeile einen einzelnen Punkt als Endekennung enthält.

Das Absetzen eines Befehls geschieht über die Methode C<rcb>. Ihr werden
zwei Argumente übergeben, das Kommando (ohne Zeilenende) und eine
{{1:Callback}}-Funktion. Das Kommando wird an den NNTP-Server geschickt,
die Callback-Funktion wird aufgerufen, wenn die erste Zeile der Antwort
angekommen ist (mit dem Statuscode).

Dies wird erreicht, indem der Event-Watcher für die NNTP-Verbindung
(C<$self->{w}>) gefüttert und gestartet wird:

!block perl
sub rcb {
   my $self = shift;
   my $cmd = shift;
   my $cb = shift;
   if ($cmd) {
      $self->command($cmd);
   } else {
      $cmd = "<anonymous command>";
   }

   $self->{w}->desc($cmd);
   $self->{w}->cb(sub {
      $self->{w}->stop;
      $cb->($self);
   });
   $self->{w}->start;
}
!endblock

Falls eine Befehl (C<$cmd>) übergeben wurde, wird dieser über die Leitung
gepustet (C<$self->command>) und als beschreibender Text verwendet.  Mit
C<desc> wird diese Beschreibung gesetzt (hilfreich zum Debuggen oder
Tollfühlen, wenn es hinterher funktioniert).

Dann wird der Callback (C<cb>) gesetzt, der lediglich den Watcher
stoppt (Befehle sind einmalige Angelegenheiten) und die {{eigentliche}}
Callback-Funktion aufruft, und schließlich wird der Watcher gestartet.

H2: Lesen der Antwort

Der schwierigste Teil des Skriptes ist das zeilenweise Lesen, das vom
NNTP-Protokoll vorausgesetzt wird.  Da Perl von sich aus (noch) keinerlei
Support dafür anbietet ({{C:<>}} blockiert den Prozeß oder liefert
keine ganzen Zeilen zurück), mußte das Zusammensetzen der Zeilen selbst
implementiert werden.

Grundlage dafür ist die Methode C<refill>, die alle Zeichen liest, die
angekommen sind (ohne zu blockieren) und sie in einem Puffer ablegt:

!block perl
sub refill {
   my $self = shift;
   my $wait = shift;
   my $fd = $self->{fd};
   fcntl $fd, F_SETFL, O_NONBLOCK;
   for(;;) {
      my $r = sysread $fd, $self->{buff}, 32768, length $self->{buff};
      if ($r>0) {
         last;
      } elsif (!defined $r && $! == EAGAIN) {
         last unless $wait;
         $self->{w}->cb(sub { $self->{w}->stop; Event::unloop });
         $self->{w}->start;
         Event::loop();
      } else {
         $self->{buff} = "500 I/O error: $!\015\012.\015\012";
         delete $self->{host};
         last;
      }
   }
   fcntl $fd, F_SETFL, 0;
}
!endblock

Das Argument C<$wait> bestimmt, ob auf jeden Fall gewartet werden soll,
oder ob C<refill> zurückkehren soll, auch wenn keine neuen Daten verfügbar
sind. Letzteres ist außerst selten der Fall und wurde entsprechend
ineffizient implementiert, indem ein "leerer" Watcher gestartet wird und
dann auf dessen Unloop gewartet wird.

Als nächstes in der Hierarchy steht C<getline>, das einfach die nächste
Zeile liefert, notfalls durch Warten:

!block perl
sub getline {
   my $self = shift;
   $self->refill(1) while $self->{buff} !~ s/^([^\015\012]*)\015\012//o;
   $1;
}
!endblock

Sie ist sehr einfach: gibt es schon eine ganze Zeile im Puffer, dann
schneide sie heraus und gib sie zurück. Nicht sehr effizient, aber einfach
zu benutzen.

Sie wird benutzt von C<response>, wo die Zeile in ihre beiden Kompomenten
(Statuscode, Meldung) zerlegt wird, und die erste Ziffer des Statuscodes
zurückgegeben wird (der für das weitere Vorgehen am entscheidensten ist).

!block perl
sub response {
   my $self = shift;
   @{$self}{'code','message'} = split m/ /, $self->getline, 2;
   substr $self->{code}, 0, 1;
}
!endblock

H2: Scannen einer Gruppe

Um herauszufinden, welche Artikel seit dem letzten Mal neu hinzugekommen sind,
wird die Statusmeldung ausgewertet, die der Server beim Wechsel in eine Gruppe liefert:

!block example
{{BEFEHL }}  GROUP comp.lang.perl.moderated
{{ANTWORT}}  211 125 4886 5010 comp.lang.perl.moderated group selected
!endblock

C<211> ist der Statuscode für "O.K.", C<125> ist die Zahl der Artikel,
C<4886> ist die erste und C<5010> die letzte Artikelnummer.

Dies ist eine ideale Anwendung für C<rcb>:

!block perl
$self->rcb("GROUP $group", sub {
   if ($self->response == 2 && $self->{message} =~ /(\d+)\s+(\d+)\s+(\d+)/) {
      my($count, $first, $last, $name) = ($1, $2, $3, $3);
      if ($count) {
         $self->slog("selected group $group");
         $self->{group} = $group;
         $self->{first} = $first;
         $self->{last} = $last;
         $cb->($self);
         return;
      } else {
         $self->slog("SKIPPED empty group $group: ", substr($self->{message},0,-1));
      }
   } else {
      $self->slog("SKIPPED bogus group $group on ".$self->{host}[0].": ", substr($self->{message},0,-1));
   }
   $self->idle;
});
!endblock

C<rcb> bekommt zwei Argumente übergeben: C<"GROUP $group"> ist das
NNTP-Kommando zum Wechseln der (News-) Gruppe, das zweite Argument ist die
Callback-Funktion, die die NNTP-Anwtort als Argument bekommt.

Die Verwendung einer Closure erlaubt es, Befehl (C<rcb>) und die Reaktion
(das C<sub {}>) direkt hintereinander zu schreiben, so, als wäre C<rcb>
ein "normaler", blockierender Aufruf zum Lesen einer Zeile, mit dem einzigen Unterschied, daß
die Auswertung des Ergebnisses im einem eingerückten Block stattfindent. Anders gesagt, aus:

!block perl
$response = $self->rcb("GROUP $group");

if ($response....) {
}
!endblock

wird:

!block perl
$self->rcb(GROUP $group", sub {

   if ($response....) {
   }
});
!endblock

C<rcb> kehrt jedoch sofort zurück (ein C<sub>, daß C<rcb> verwendet, kann
deshalb nicht sofort ein Resultat an den Aufrufer zurückliefern.

Die Information über die Gruppe (C<first> und C<last>, wird aus der
NNTP-Antwort genommen) wird später mit den Daten aus der SQL-Datenbank
verglichen (das hat allerdings nichts mit C<Event> zu tun):

!block perl
sub group_scan {
   my $self = shift;
   my $group = $self->{group};
   my $todo = new Set::IntSpan $self->{first}."-".$self->{last};
   $todo = $todo->intersect($self->gs_done->complement);
   if ($todo->empty) {
      $self->slog("[no new articles in $group]");
   } else {
      $self->slog("scanning group $group: ", $todo->run_list);
      add_job($self->{host},'A',$group,$todo);
   }
   $self->idle;
}
!endblock

Das C<Set::IntSpan>-Modul wird dazu benutzt, um aus der Menge der
vorhandenen Artikel die bereits gescannten (die von C<gs_done>
zurückgegeben werden) zu entfernen. Ist die resultierende Menge nicht
leer, wird ein neuer Job ("hole alle diese Artikel") erzeugt.

H2: Holen eines Artikels

Das Holen geschieht in zwei Stufen. Zuerst wird die {{Message-Id}} mit
einem C<STAT>-Befehl ausgewertet. Damit wird außerdem festgestellt, ob ein
bestimmter Artikel überhaupt existiert.

!block perl
$self->rcb("STAT ".$self->{num}, \&got_stat);
!endblock

Ein Protokollbeispiel:

!block example
{{BEFEHL }} STAT 5010
{{ANTWORT}} 223 5010 <85j7jc$68n@junior.apk.net> article retrieved - request text separately
{{BEFEHL }} STAT 4977
{{ANTWORT}} 430 No such article: 4977
!endblock

Der Callback C<got_stat> wertet diese Information aus:

!block perl
sub got_stat {
   my $self = shift;
   my $r = $self->response;
   $self->mark_article_done;

   ($self->{mid}) = $self->{message} =~ /<([^>]+)>/g;

   if ($r == 2) {
      my $aid = sql_fetch("select count(*) from art where mid=? limit 1", "".$self->{mid});
      $self->mark_article_present;
      if ($aid) {
         sql_exec("replace into lnk values (?,?)", $self->gid, $aid);
         $self->idle;
      } else {
         $busy{$self->{mid}}++;
         $stat_article++;
         $self->rcb_dot("ARTICLE ".$self->{num}, \&got_article);
      }
   } else {
      $self->idle;
   }
}
!endblock

Existiert der Artikel nicht, ist der Job beendet und es wird in den
idle-Modus gegangen. Wurde er schon einmal geholt (z.B. in einer anderen Gruppe)
wird er nicht noch einmal geholt, sondern lediglich in die Gruppe "gelinkt" (Artikel können sehr groß werden).

Ansonsten wird ein C<ARTICLE>-Befehl abgesetzt, mit dem der gesamte Artikel geholt wird.

!block example
{{BEFEHL }} ARTICLE 5010
{{ANTWORT}} 220 5010 <85j7jc$68n@junior.apk.net> article retrieved - text follows
{{ANTWORT}} From: allbery@apk.net (Brandon S. Allbery KF8NH)
{{ANTWORT}} Newsgroups: comp.lang.perl.moderated
{{ANTWORT}} Subject: Re: Usefulness of Pseudo Hashes
{{ANTWORT}} Message-ID: <85j7jc$68n@junior.apk.net>
{{ANTWORT}}
{{ANTWORT}} Also sprach Alex Rhomberg <rhomberg@ife.ee.ethz.ch> (<384E39B8.D8635949@ife.ee.ethz.ch>):
{{ANTWORT}} +-----
{{ANTWORT}} | I wonder why pseudo hashes were invented
{{ANTWORT}} +--->8
{{ANTWORT}}
{{ANTWORT}} Sometimes you need an ordered list (so you can't use hashes) with keyed access
{{ANTWORT}} to the list (so lists/arrays are slow and a pain in the butt to use).  Pseudo
{{ANTWORT}} hashes are a better solution than the usual hack of maintaining duplicate
{{ANTWORT}} information in a hash and an array/list.
{{ANTWORT}}
{{ANTWORT}} --
{{ANTWORT}} brandon s. allbery      [os/2][linux][solaris][japh]     allbery@kf8nh.apk.net
{{ANTWORT}} system administrator         [WAY too many hats]           allbery@ece.cmu.edu
{{ANTWORT}} carnegie mellon / electrical and computer engineering                    KF8NH
{{ANTWORT}}                           Kiss my bits, Billy-boy.
{{ANTWORT}} .
!endblock

Hierbei tritt das Problem auf, daß nach der Statuszeile ein Artikel
folgt. Deshalb wird statt C<rcb> die Methode C<rcb_dot> benutzt (das steht
für "read callback + data read until dot"):

!block perl
sub rcb_dot {
   my $self = shift;
   my $cmd = shift;
   $self->{rcb_cb} = shift;
   delete $self->{body};
   $self->rcb($cmd, sub {
      if ($self->response == 2) {
         $self->{w}->cb([$self, 'rcb_cb']);
         $self->{w}->start;
         $self->rcb_cb;
      } else {
         $self->{rcb_cb}->($self);
      }
   });
}

sub rcb_cb {
   my $self = shift;
   $self->refill(0);
   if ($self->{buff} =~ s/^\.\015\012|^(.*?)\015\012\.\015\012//s) {
      $self->{body} .= $1;
      $self->{w}->stop;
      $self->{body} =~ s/\015\012/\n/g;
      $self->{rcb_cb}->($self, delete $self->{body});
   } elsif ($self->{buff} =~ s/^(.*\015\012)//s) {
      $self->{body} .= $1;
   }
}
!endblock

Der komplizierteste Teil ist C<rcb_cb>, in der die Artikeldaten
akkumuliert werden, wozu furchtbare regexes benutzt wurden. Im Gegensatz
zu vielen anderen Stellen wurden die Callbacks nicht durch Closures
implementiert, da Event+Closures im allgemeinen ein großes Memory-Leak
ist (soll ab Event-0.59 besser sein, aber man kann sichs nicht immer
ausssuchen).

H2: Updaten von SQL-Tabellen

Die Aufrufe C<mark_article_done> und C<mark_article_present> markieren
einen Artikel in der Datenbank als bearbeitet bzw. vorhanden. Sie setzen
einfach ein Element in der entsprechenden C<Set::IntSpan>-Menge.

Diese Mengen werden in einer SQL-Tabelle gespeichert. Da sie relativ groß
sind (einige Kilobytes), serh häufig geändert werden (bis zu 100 mal pro
Sekunde) und der Zielrechner sehr langsam ist, sollten die Tabellen nicht
bei jeder Änderung gespeichert werden. Dies wird mit einem C<idle>-Watcher
erreicht, der jedesmal gestartet wird, wenn sich die Daten ändern:

!block perl
my $save_gs = Event->idle(
                     desc => "groupstatus saver",
                     max => 60,
                     cb => sub {
                        $_[0]->w->stop;
                        # zurückschreiben der Tabellen
                     }
                 );
$save_gs->stop;

sub mark_article_done {
   my $self = shift;
   $gs{$self->hid,$self->gid}[0]->insert($self->{num});
   $save_gs->start;
}
!endblock

Sollte der Draht so richtig dampfen, sorgt der Timeout von 60 Sekunden
dafür, daß bei einem Absturz maximal die letzte Minute fehlt. In der
Praxis wird er viel häufiger aufgerufen, nämlich dann, wenn alle
einkommenden Verbindungen einmal bedient wurden und noch keine weiteren
Daten angekommen sind.

H2: Künstliche "Lastsimulation"

Da der Test-Server auf der lokalen Maschine lief, mußte künstlich Last
erzeugt werden, um einigermaßen wirklichkeitsnahe Ergebnisse zu erhalten.
Die größten Zeitfaktoren bei NNTP sind die Latenz zum Server (abhängig von
der Entfernung) und die Bandbreite.

Um eine künstliche Latenz einzuführen, wird die C<command>-Funktion leicht abgeändert:

!block perl
sub command {
   my ($self, $cmd) = @_;
   Event->timer(after => rand, cb => sub {
      $_[0]->w->cancel;
      syswrite $self->{fd}, "$cmd\015\012";
   });
}
!endblock

Statt das Kommando sofort zu verschicken, wird ein kurzer Timer
gestartet. Die Verzögerung liegt zwischen 0 und 1 Sekunde (C<rand>) und
sorgt für eine Streuung. Ohne diese zufällige Verzögerung würde ein
unerwünschtes Bearbietungsmuster entstehen, bei dem effektiv nur ein
Scan-Vorgang gleichzeitig stattfindet.

Die obige Version von C<command> schneidet in ihrer Kürze recht gut gegen
die "normale" Version ab:

!block perl
sub command {
   my ($self, $cmd) = @_;
   syswrite $self->{fd}, "$cmd\015\012";
}
!endblock

H2: C<NetServer::ProcessTop>

Ein recht interessantes Modul ist C<NetServer::ProcessTop>. Wird es
benutzt, bindet es sich auf einen TCP-Port, den man per C<telnet>
ansprechen kann, um ein C<top>-artiges Listing der Event-Watcher zu
bekommen, Außerdem kann man die Watcher edieren.

Die Benutzung ist denkbar einfach:

!block perl
eval {
   require NetServer::ProcessTop;
   NetServer::ProcessTop->new(7000);
};
!endblock

Ein C<telnet localhost 7000> erzeugt dann dieses Bild:

!block verbatim
get PID=3407 @ cerebro                                       | 14:26:46 [  60s]
10 events; load averages: 0.75, 0.73, 0.00; lag  0%

  EID PRI STATE   RAN  TIME   CPU TYPE DESCRIPTION                           P1
    0   7         912  0:00 26.6%  sys idle
    3   4 zomb    227  0:00 16.9%   io ARTICLE 273573
    6   4 zomb    236  0:00 16.6%   io ARTICLE 273572
    4   4 sleep   232  0:00 16.4%   io ARTICLE 273575
    5   4 sleep   221  0:00 16.0%   io ARTICLE 273574
    9   4 wait    117  0:00  7.3% idle groupstatus saver
   10   4 wait    180  0:00  0.3% idle reschedule hook
    2   3 sleep     1  0:00  0.0% time Event::Stats
    1   3 cpu       0  0:00  0.0%   io NetServer::ProcessTop::Client localhost
    7   3 sleep     0  0:00  0.0%   io NetServer::ProcessTop
    8   4 sleep     0  0:00  0.0%   io user input
    0  -1           0  0:00  0.0%  sys other processes

%
!endblock

Weil das Modul aber ein potentielles Sicherheitsproblem sein kann, sollte
es nur zum Debuggen/Erfreuen verwendet werden.

A1: Der Quellcode

!include "get"; perl

A2: Mehr!

Die folgenden Module/Programme/RFCs wurden für das Projekt verwendet.

* C<Event> - Event loop processing. {{URL:http://www.cpan.org/}}
* C<Set::IntSpan> - Manages sets of integers. {{URL:http://www.cpan.org/}}
* C<RFC-977> Network News Transfer Protocol. {{URL:ftp://ftp.isi.edu/in-notes/rfc977.txt}}
* C<NetServer::ProcessTop> - Make event loop statistics easily available. {{URL:http://www.cpan.org/}}
* C<Time::HiRes> - High resolution ualarm, usleep, and gettimeofday. {{URL:http://www.cpan.org/}}
* C<Socket> - load the C socket.h defines and structure manipulators. (Teil der Perl-Distribution).
* C<DBI> - Database independent interface for Perl
* C<MySQL> SQL-Datebank. {{URL:http://www.mysql.com}}.


Revision:	1.1
Committed:	Mon Mar 19 11:08:50 2001 UTC (23 years, 2 months ago) by root
Branch:	MAIN
CVS Tags:	HEAD
Log Message:	* empty log message *