pthreads properties

[veryhot]

Lädt man sich die binaries zu pthreads runter, werden auch Beispiele mitgeliefert. Darunter gibt es auch das folgende Beispiel:

<?php
/*
* Sharing symbols 101
* @NOTE Thread::fetch was never included in a release and was superceeded by object handlers
*     pthreads allows read access to thread data from any context
    pthreads allows write access to thread data from any context
    carry on reading ...
    work in progress ...
*/
class TestObject {
    public $val;
}

class Fetching extends Thread {
    public function run(){
        /*
        * of course ...
        */
        $this->sym = 10245;
        $this->arr = array(
            "1", "2", "3"
        );
        
        /*
        * objects do work, no preparation needed ...
        * read/write objects isn't finalized ..
        * so do the dance to make it work ...
        */
        $obj = new TestObject();
        $obj->val = "testval";
        $this->obj = $obj;
        
        /*
        * will always work
        */
        $this->objs = serialize($this->obj);
        
        /*
        * nooooooo
        */
        $this->res = fopen("php://stdout", "w");
        
        /*
        * tell the waiting process we have created symbols and fetch will succeed
        */
        $this->synchronized(function(){
            $this->notify();
        });
        
        /* wait for the process to be finished with the stream */
        $this->synchronized(function(){
            $this->wait();
        });
    }
}

$thread = new Fetching();

$thread->start();

$thread->synchronized(function($me){
    $me->wait();
}, $thread);

/*
* we just got notified that there are symbols waiting
*/
foreach(array("sym", "arr", "obj", "objs", "res") as $symbol){
    printf("\$thread->%s: ", $symbol);    
    $fetched = $thread->$symbol;
    if ($fetched) {
        switch($symbol){
            /*
            * manual unserialize
            */
            case "objs":
                var_dump(unserialize($fetched));
            break;
            
            default: var_dump($fetched);
        }
    }
    printf("\n");
}

/* notify the thread so it can destroy resource */
$thread->synchronized(function($me){
    $me->notify();
}, $thread);
?>

Warum benötigt man die ganzen synchronized-Anweisungen?

Ganz besonders irritiert mich, dass es in der run-Methode gleich hintereinander ein notify und wait gibt. Die Kommentare helfen mit leider auch nicht weiter.

[tkausl]

Das ganze geht tief ins Multithreading, hartes Thema...

Kurz gesagt, du kannst nur auf Objekte warten/notifyen auf die du ein Lock hast. Beim warten wird der Lock dann freigegeben und es wird gewartet. Beim Notify wird der wartende Thread (EIN wartender thread, sofern mehrere warten. notifyAll würde ALLE wartenden Threads aufwecken) dann wieder den Lock anfordern und weiterarbeiten.

(So ist es bei Java zumindest, in PHP hab ich noch keinerlei erfahrung mit Multithreading, denke aber dass es dort nicht anders ist).

[veryhot]

Das muss ich erst mal verdauen.

[Asterixus]

notifyAll gibt es in PHP anscheinend nicht. Deshalb rate ich dir auch an, nicht synchronized zu benutzen sondern die Mutex-Klasse. Macht im Grunde dasselbe, aber es ist interessanter, die Locks als die Threads überall rumzutragen.

Ich glaube aber, es wäre besser, wenn du dich mit Multithreading in Java auseinandersetzen würdest, denn in PHP sieht mir das Ganze etwas zu gebastelt aus.

(PHPs Mutex ist dasselbe wie ein ReentrantLock in Java.)

[Tiberius]

Der Unterschied vom php synchronized zum reinen Mutex lock/unlock ist, dass ein wait() in einem synchronized den Lock vom synchronized atomar freigibt

[Asterixus]

Keine Ahung, was du genau damit meinst, aber Mutex alle Aktionen in einem lock/unlock-Prozess sind atomar für diesen Lock, darum geht es ja. Statt wait/notify benutzt man dann Cond::wait/Cond::broadcast/Cond::signal. In keinen meiner neuen Java-Anwendungen kommt das Schlüsselwort synchronized mehr vor, ich benutze nur noch explizite Locks und wenn nicht zu umständlich Compare-and-Set Strukturen.

Ich habe zwei Anwendungen komplett umschrieben und explizite Locks zu benutzen, war wirklich effizienter (jeder Thread hat etwa 5% mehr arbeiten können) und ich finde das persönlich vom Stil her besser. Inwiefern das allerdings in PHP stimmt, kann ich nicht sagen. In C und C++ gibt es synchronized in der Standard-Bibliothek übrigens nicht.

[Tiberius]

Wenn man einen Bereich oder ein Objekt reservieren möchte, dafür Mutex::lock() nutzt und in diesem Bereich wait() aufruft, kann kein anderer Thread per lock() den Mutex akquirieren, da dieser vom wait() nicht freigegeben wurde - Deadlock. In dem Fall muss man vor dem wait() explizit Mutex::unlock() aufrufen. Das kann problematisch werden, da der Übergang vom unlock() zum wait() nicht atomar abläuft und ein anderer Thread in der Zeit durchrutschen kann. In einem synchronized Closure hingegen ist der wait() Aufruf und die damit verbundene Freigabe vom Mutex atomar, somit kann z.B. ein notify() von einem anderen Thread nicht versehentlich vor dem wait() im Closure ausgeführt werden.

[Asterixus]

Was du da sagst, klingt ehrlich gesagt, danach, dass du was falsch verstehst.

wait() in Cond verlangt nämlich erstens die Cond-Referenz und zweitens die Mutexlock-Referenz, um gerade diesen Lock freizugeben.


Ich habe übrigens mal pthreads installiert und kann nur anraten, davon die Finger zu lassen. Für einfache Aufgaben könnte das noch reichen, aber sobald es etwas komplizierter wird, kann man das vergessen.

<?php
error_reporting(-1);
define("HELPERS", 8);
class App {
    protected $helperLock = null;
    protected $helperCond = null;
    protected $appLock = null;
    protected $appCond = null;
    protected $lock = null;
    protected $closed = false;
    protected $sum;
    protected $computing = false;
    protected $count = 0;
    protected $helperStates;
    protected $threads;
    public function getHelperLock() {
        return $this->helperLock;
    }
    public function getHelperCond() {
        return $this->helperCond;
    }
    public function __construct() {
        $this->helperLock = Mutex::create();
        $this->helperCond = Cond::create();
        
        $this->appLock = Mutex::create();
        $this->appCond = Cond::create();
        
        $this->lock = Mutex::create();
    }
    public function launch() {
        $this->threads = array();
        $this->helperStates = array();
        
        for ($i = 0; $i < HELPERS; ++$i) {
            $this->helperStates[$i] = false;
            $thread = new Helper($this, $i);
            $thread->start();
            $this->threads[$i] = $thread;
        }
    }
    public function helperShouldBeWaiting($helperId) {
        if ($this->closed) {
            return false;
        }
        $computing = &$this->computing;
        return ($this->helperStates[$helperId] && $this->computing) || !$this->computing;
    }
    public function helperDone($helperId, $sum) {
        Mutex::lock($this->lock);
        ++$this->count;
        $this->helperStates[$helperId] = true;
        if ($this->count === HELPERS) {
            Mutex::lock($this->appLock);
            Cond::broadcast($this->appCond);
            Mutex::unlock($this->appLock);
        }
        Mutex::unlock($this->lock);
    }
    public function computeStuff() {
        $this->computing = true;
        $this->sum = 0;
        
        for ($i = 0; $i < HELPERS; ++$i) {
            $this->helperStates[$i] = false;
        }
        
        Mutex::lock($this->helperLock);
        Cond::broadcast($this->helperCond);
        Mutex::unlock($this->helperLock); 
        
        Mutex::lock($this->appLock);
        
        while ($this->count !== HELPERS) {
            Cond::wait($this->appCond, $this->appLock);
        }
        
        Mutex::unlock($this->appLock);
        $this->computing = false;
        return $this->sum;
    }
    public function close() {
        $this->closed = true;
        Mutex::lock($this->helperLock);
        Cond::broadcast($this->helperCond);
        Mutex::unlock($this->helperLock);
        /* Da lass ich mal den GC arbeiten, habe keinen Bock, funktioniert ja aber nicht.
        Mutex::destroy($this->helperLock);
        Mutex::destroy($this->appLock);
        Mutex::destroy($this->lock);*/
    }
    public function isClosed() {
        return $this->closed;
    }
}

class Helper extends Thread {
    /**
     * @var App
     */
    protected $app;
    protected $helperId;
    protected $sum;
    public function __construct(App $app, $helperId) {
        $this->app = $app;
        $this->helperId = $helperId;
    }
    public function run() {
        $app = $this->app;
        $lock = $app->getHelperLock();
        $cond = $app->getHelperCond();
        while (true) {
            Mutex::lock($lock);
            while ($app->helperShouldBeWaiting($this->helperId)) {
                Cond::wait($cond, $lock);
            }
            
            Mutex::unlock($lock);
            
            if ($app->isClosed()) {
                return;
            }
            
            $sum = 0;
            for ($i = 0; $i < 1000000; ++$i) {
                $sum+= mt_rand(0, 2);
            }
            $app->helperDone($this->helperId, $sum);
        }
    }
} 

$app = new App();
$app->launch();
echo $app->computeStuff();
//$app->close(); 

Ich habe hier ein eigentlich triviales sinnloses Programm geschrieben, um das zu testen. In diesem Programm erstelle ich 8 Threads, die warten, bis der Main-Thread eine Methode (computeStuff) aufruft, denn dann wird in jedem Thread 1 Millionen Mal iteriert. Wenn diese 8 Threads fertig sind, schlafen sie wieder ein, bis diese Methode erneut aufgerufen wird.
Was ich feststellen musste, ist dass man Threads keine Objekte übergeben kann. Diese werden nämlich irgendwie nur kopiert und sind keine eigentlichen Objektreferenzen mehr.
Denn hat man ein Objekt (hier $app, Instanz von App) und übergibt dieses Objekt an einen Thread, kann Thread A etwas in diesem Objekt ändern, aber Thread B wird dies nicht sehen können und wird nur den Zustand dieses Objekts sehen, wie es ihm zugeschickt wurde.
Ich präzisiere, dass dieses Programm 1:1 übersetzt wurde von einer Java-Anwendung, die ich benutze, um Leuten die Funktionalität expliziter Locks zu erklären. Ich habe das auch schon in C übersetzt, wo ich so ein Problem nie hatte.

Führe ich dies aus, wird $app->computing auf TRUE im Main-Thread gesetzt. Wenn ich allerdings helperShouldBeWaiting() aufrufe und dort ein var_dump($this->computing) setze, sagt es mir FALSE. Das wait/broadcast-Schema funktioniert, man kann nur nicht richtig arbeiten, weil die Threads kein gemeinsames Objekt teilen können. Da bringt Multithreading überhaupt nichts.

[rkr]

Beim Thema Nebenläufigkeit (Threading) kann man viel falsch machen. Ich habe leider den Link nicht mehr, aber ich hatte mal in einer Onlinepräsentation gelesen, dass sich in 80% aller Fälle in denen man threading benötigt, man mit Futures (bzw Promises) ebenfalls zur gleichen Lösung käme. Für PHP habe ich nach kurzer Recherche das hier gefunden:
https://github.com/reactphp/promise
Ich habe jetzt nicht nachgeschaut, ob es dort auch ein gelöstes "ruf mich an, wenn all diese Prozesse beendet wurden" Szenario gibt.

[Asterixus]

Das ist klar, in Java benutze ich dafür immer ThreadPools mit http://docs.oracle.com/javase/6/docs...orService.html

Hast du denn schonmal in PHP mit Pools gearbeitet? Ich nehme jedenfalls stark an, dass ich damit dasselbe Problem hätte. pthreads hat da einen enormen Designfehler.

[Tiberius]

@Asterixus, ich schrieb von Thread(ed)::wait() wie im Beispiel von veryhot zu sehen, ich ging jetzt nicht von Cond::wait() aus.

Man darf nicht vom JAVA Threading auf das Verhalten von pthreads schließen, da liegen Welten zwischen. Trotz pthreads bleibt das PHP Model "Share nothing" erhalten. Alle Objekte, die an einen Thread übergeben werden, müssen von Thread, Threaded, Worker oder Stackable erben, sonst wird das Objekt von einem Threadcontext in einen anderen Threadcontext dupliziert. Lese- und Schreibzugriffe auf Properties werden infolge soweit wie möglich serialisiert, kopiert(marshalling) und sind somit auch dupliziert.

Auf github findet man über pthreads vieles zum Verhalten und dessen Funktionsweise. Ist leider etwas verstreut. In der Beta war es auch noch so, dass die Sichtbarkeit public und protected geändert wurde. Ob sich das zwischenzeitlich in der finalen Version geändert hat, kann ich derzeit nicht sagen.

[Asterixus]

Es geht hier nicht nur um einen PHP/Java-Vergleich. Ich rede hier von allen Programmiersprachen in denen ich mit Multithreading schon zu tun hatte. Ich hatte übrigens bereits die App als Threaded extended. Geändert hat es was, aber nur, dass die Threads gar nicht mehr auf broadcast() reagierten.

pthreads (allein schon der Name ist die erste Dummheit, denn ich habe erst geglaubt, es ginge hier um Posix Threads) sieht für mich viel zu sehr nach zusammengebasteltem Kram aus, denn "Share Nothing" macht zwischen verschiedenen Scripts Sinn, wenn Threads aber nicht miteinander kommunizieren dürfen, ergibt das einfach keinen Sinn. Ich habe mich eben nochmals ein bisschen eingelesen und die Tatsache, dass Objekte serialisiert werden, wenn sie einem Thread übergeben werden, spricht doch schon sehr für meine These, dass pthreads schlecht zusammengebastelt wurde. Ich bin der Meinung, wenn man sowas macht: Entweder richtig oder gar nicht.

Im Grunde interesse ich mich für pthreads und allgmein PHP immer weniger. Wollte das hier nur mal testen und erklären.

[tkausl]

denn "Share Nothing" macht zwischen verschiedenen Scripts Sinn, wenn Threads aber nicht miteinander kommunizieren dürfen, ergibt das einfach keinen Sinn.

Es ergibt alleine deswegen schon kein Sinn, da Threads Shared-Memory besitzen (sollten), im gegensatz zu Prozessen

[Asterixus]

Ja, und gerade das ist nicht der Fall.

[veryhot]

Könnte mir mal bitte wer ein ganz einfaches Beispiel nur mit synchronized zeigen? Damit man sieht, wie was blockiert wird.

Versucht habe ich es mit:

<?php
   error_reporting(~0);
   class T extends Thread 
   {
      private $id;
      private $t;
      function __construct($id, $t)
      {
         $this->id = $id;
         $this->t = $t;
      }
      public function run() 
      {
         $oDT = new DateTime();
         var_dump('run '.$this->id.' '.$oDT->format('h i s u'));
         $this->synchronized
         (
            function($thread)
            {
               $oDT = new DateTime();
               var_dump('synchronized0 '.$thread->id.' '.$oDT->format('h i s u'));
               sleep($thread->t);
               $oDT = new DateTime();
               var_dump('synchronized1 '.$thread->id.' '.$oDT->format('h i s u'));
            }, 
            $this
         );
      }
   }
   
   $oT0 = new T(0, 6);
   $oT1 = new T(1, 4);
   $oDT = new DateTime();
   var_dump('main1 '.$oDT->format('h i s u'));
   
   $oT0->start();
   $oDT = new DateTime();
   var_dump('main2 '.$oDT->format('h i s u'));
   
   sleep(1);
   
   $oDT = new DateTime();
   var_dump('main3 '.$oDT->format('h i s u'));
   
   $oT1->start();
   
   $oT0->join();
   $oT1->join();
?>

Leider schaut das Ergebnis nicht so aus, wie ich gerne hätte. Es wird nämlich nichts blockiert. Ergebnis:

string(21) "run 1 02 27 35 000000"
string(31) "synchronized0 1 02 27 35 000000"
string(31) "synchronized1 1 02 27 39 000000"
string(21) "run 0 02 27 34 000000"
string(31) "synchronized0 0 02 27 34 000000"
string(31) "synchronized1 0 02 27 40 000000"
string(21) "main1 02 27 34 000000"
string(21) "main2 02 27 34 000000"
string(21) "main3 02 27 35 000000"