我正在寻找一种安全快捷的方式来使用共享对象。

我已经在这里问了这个问题:https://github.com/krakjoe/pthreads/issues/470
但显然这不是正确的地方。

试图与许多其他上下文(线程)共享一个对象(线程)。
所有线程都在更新这个分片对象——它们可以设置自己的请求，也必须响应其他人的请求。

现在 krakjoe 回应说在 7 中无法使用锁定/解锁，我遇到了问题。

我知道 :.synchronized 但不知道如何使用它来满足我的需求。

我如何使用::synchronized 来编写类似的方法

<?php
/*
TEST:
    create n threads
    each will
        - Shared::set() its own ref
        - check if Shared::exists() its own ref
        - Shared::get() its ref back
        - call method ::isRunning() at returned val to easily check if is ref or got overwritten by another context

TODO:
    using ::synchronized to handle multi-context-access

NOTES:
    every method as public to prevent pthreads v2 "Method Modifiers - Special Behaviour"
        see: "Method Modifiers - Special Behaviour"
            at http://blog.krakjoe.ninja/2015/08/a-letter-from-future.html
*/
class Shared extends Threaded
{
    public $data;
    public function exists($ident)
    {
        return isset($this->data[$ident]);
    }
    public function set($ident, $ref)
    {
        $return = false;
        if(!isset($this->data[$ident])){
            $data = $this->data;
            $data[$ident] = $ref;
            $this->data = $data;
            $return = $this->data[$ident];
        }
        #echo __METHOD__ . '(' . $ident . ') => ' . gettype($return) . PHP_EOL;
        return $return;
    }
    public function get($ident)
    {
        $return = false;
        if($this->exists($ident) === true){
            $data = $this->data;
            $return = $data[$ident];
            unset($data[$ident]);
            $this->data = $data;
        }
        #echo __METHOD__ . '(' . $ident . ') => ' . gettype($return) . PHP_EOL;
        return $return;
    }
}

class T extends Thread
{
    public $count;
    public function __construct(Shared $Shared, $ident)
    {
        $this->Shared = $Shared;
        $this->ident = $ident;
    }
    public function run()
    {
        $slowdown = true;
        $this->count = 0;
        while(true){
            if($slowdown){
                // "don't allow usleep or sleep" : https://github.com/krakjoe/pthreads/commit/a157b34057b0f584b4db326f30961b5c760dead8
                //  loop a bit to simulate work:
                $start = microtime(true);
                $until = rand(1, 100000)/1000000;
                while(microtime(true)-$start < $until){
                    // ...
                }
            }

            if($this->Shared->exists($this->ident) === true){
                $ref = $this->Shared->get($this->ident);
            }
            else{
                $ref = $this->Shared->set($this->ident, $this);
            }
            // calling a method on $ref -- if not a ref we crash
            $ref->isRunning();
            unset($ref);
            $this->count++;
        }
    }
}


echo 'start ...' . PHP_EOL;

$n = 8;
$Shared = new Shared();
for($i = 0, $refs = array(); $i < $n; $i++){
    $refs[$i] = new T($Shared, $i);
    $refs[$i]->start();
}

while(!empty($refs)){
    // print status:
    if(!isset($t)or microtime(true)-$t > 1){
        $t = microtime(true);
        echo 'status: ' . count($refs) . ' running atm ...' . PHP_EOL;
    }

    // join crashed threads:
    foreach($refs as $i => $thread){
        if($thread->isRunning() === false){
            echo 'T-' . $i . ' stopped after ' . $thread->count . PHP_EOL;
            if($thread->isJoined() === false){
                $thread->join();
            }
            unset($refs[$i]);
        }
    }
}

echo 'no thread running anymore.' . PHP_EOL;

/* output
start ...
status: 8 running atm ...

Notice: Undefined offset: 6 in ...\shared_test.php on line 33

Fatal error: Call to a member function isRunning() on null in ...\shared_test.php on line 82
T-6 stopped after 10
status: 7 running atm ...

Notice: Undefined offset: 4 in ...\shared_test.php on line 33

Fatal error: Call to a member function isRunning() on null in ...\shared_test.php on line 82
T-4 stopped after 35
status: 6 running atm ...

Notice: Undefined offset: 7 in ...\shared_test.php on line 33

Fatal error: Call to a member function isRunning() on null in ...\shared_test.php on line 82
T-7 stopped after 43
status: 5 running atm ...
status: 5 running atm ...
status: 5 running atm ...

[...]
*/
?>

最佳答案

Threaded对象已经是线程安全的，也就是说，任何时候你读、写、检查是否存在或删除(取消设置)一个成员，操作都是原子的——没有其他上下文可以执行任何上述操作，而第一个操作发生。对于用户不知道的引擎处理程序也是如此，低至最低级别的所有内容都是隐式安全的。

然而，相当令人放心......当逻辑变得更复杂时，这有明显的限制，例如在设置之前检查成员的存在或对其进行其他操作，就像您正在做的那样:虽然对对象的操作是原子的，没有什么可以阻止另一个上下文 unset调用 isset 之间的成员以及读取属性/维度的调用。

这适用于 PHP7 (pthreads v3+)

安全和诚信在这里是两件不同的事情。当完整性很重要时，您可以使用 Threaded::synchronized在 PHP7 中正确保存它。在 PHP5 中，您也可以保留它，但代码会更复杂，解释也会更复杂。

如果我理解它的逻辑，你的第二个例子应该无限期地运行。所以我使用这个假设来构建正确的代码，我将进一步假设您可能想要在这个无限循环中做什么，并在似乎需要的地方提供一些见解。

<?php
class Referee extends Threaded {

    public function find(string $ident, Threaded $reference) {
        return $this->synchronized(function () use($ident, $reference) {
            if (isset($this[$ident])) {
                return $this[$ident];
            } else return ($this[$ident] = $reference);
        });
    }

    public function foreach(Closure $closure) {
        $this->synchronized(function() use($closure) {
            foreach ($this as $ident => $reference) {
                $closure($ident, $reference);
            }
        });
    }
}

class Test extends Thread {

    public function __construct(Referee $referee, string $ident, bool $delay) {
        $this->referee = $referee;
        $this->ident   = $ident;
        $this->delay   = $delay;
    }

    public function run() {
        while (1) {
            if ($this->delay) {
                $this->synchronized(function(){
                    $this->wait(1000000);
                });
            }

            $reference = 
                $this->referee->find($this->ident, $this);

            /* do something with reference here, I guess */         

            /* do something with all references here */
            $this->referee->foreach(function($ident, $reference){
                var_dump(Thread::getCurrentThreadId(),
                        $reference->getIdent(), 
                        $reference->isRunning());
            });
        }
    }

    public function getIdent() {
        return $this->ident;
    }

    private $referee;
    private $ident;
    private $delay;
}

$referee = new Referee();
$threads = [];
$thread = 0;
$idents = [
    "smelly",
    "dopey",
    "bashful",
    "grumpy",
    "sneezy",
    "sleepy",
    "happy",
    "naughty"
];

while ($thread < 8) {
    $threads[$thread] = new Test($referee, $idents[$thread], rand(0, 1));
    $threads[$thread]->start();
    $thread++;
}

foreach ($threads as $thread)
    $thread->join();
?>

if ($this->delay) {
    $this->synchronized(function(){
        $this->wait(1000000);
    });
}

if ($this->delay) {
    $this->synchronized(function(){
        while ($this->condition) {
            $this->wait(1000000);
        }
    });
}

$reference = 
    $this->referee->find($this->ident, $this);

public function find(string $ident, Threaded $reference) {
    return $this->synchronized(function () use($ident, $reference) {
        if (isset($this[$ident])) {
            return $this[$ident];
        } else return ($this[$ident] = $reference);
    });
}

$this->referee->foreach(function($ident, $reference){
    var_dump(Thread::getCurrentThreadId(),
            $reference->getIdent(), 
            $reference->isRunning());
});

public function foreach(Closure $closure) {
    $this->synchronized(function() use($closure) {
        foreach ($this as $ident => $reference) {
            $closure($ident, $reference);
        }
    });
}

关于PHP pthreads - 共享对象，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/32341744/