PHP pThreads - 如何执行垃圾收集？

Question

给定以下代码，您如何确保已完成的 MyWorker 对象被销毁/释放它们的内存？

由于我的脚本需要大约 50 个线程不断地从 cURL 获取数据并进行处理。

我已经尝试过让线程永远不会离开 run()，或者如示例代码所示，它们离开运行并让收集函数生成它们的新副本。

但不管我在一分钟左右后达到内存限制是什么。你能告诉我我做错了什么吗？

class MyWorker extends Threaded
{
    public $complete;
    public function __construct() {$this->complete = false;}
    public function run() {$this->complete = true;}
}

$pool = new Pool(50);
for($i=0; $i<50; $i++)
    $pool->submit(new MyWorker());
$pool->collect(function($worker)
{
    global $pool;
    if($worker->complete == true)
        $pool->submit(new MyWorker());
    return $worker->complete;
});
$pool->shutdown();

Answer 1

为什么

我为什么要收集？

pthreads 提供的Worker 线程要求程序员保留对正在执行的Threaded 对象的正确引用。这对于程序员来说很难在用户空间可靠地实现，因此 pthreads 提供了 Workers 的 Pool 抽象，它为您维护引用。

为了维护那些引用 pthreads 需要知道对象何时是垃圾，它为此提供了 Pool::collect 接口(interface)。 Pool::collect 采用一个闭包，该闭包应该接受一个 Threaded 对象并在传递的对象执行完毕时返回 bool 值 true。

如何

手头的任务......

为了继续提交任务执行而不耗尽资源，您必须创建一个已完成任务的队列以重新提交到Pool

下面的代码展示了一种明智的做法:

<?php

define("LOG", Mutex::create());
/* thread safe log to stdout */
function slog($message, $args = []) {
    $args = func_get_args();
    if (($message = array_shift($args))) {
        Mutex::lock(LOG);
        echo vsprintf(
            "{$message}\n", $args);
        Mutex::unlock(LOG);
    }
}

class Request extends Threaded {
    public function __construct($url) {
        $this->url = $url;
    }

    public function run() {
        $response = @file_get_contents($this->url);

        slog("%s returned %d bytes",
            $this->url, strlen($response));

        $this->reQueue();
    }

    public function getURL()        { return $this->url; }

    public function isQueued()      { return $this->queued; }
    public function reQueue()       { $this->queued = true; }

    protected $url;
    protected $queued = false;
}

/* create a pool of 50 threads */
$pool = new Pool(50);

/* submit 50 requests for execution */
while (@$i++<50) {
    $pool->submit(new Request(sprintf(
        "http://google.com/?q=%s", md5($i))));
}

do {
    $queue = array();

    $pool->collect(function($request) use ($pool, &$queue) {
        /* check for items to requeue */
        if ($request->isQueued()) {
            /* get the url for the request, insert into queue */
            $queue[] = 
                $request->getURL();
            /* allow this job to be collected */
            return true;
        }
    });

    /* resubmit completed tasks to pool */
    if (count($queue)) {
        foreach ($queue as $queued)
            $pool->submit(new Request($queued));
    }

    /* sleep for a couple of seconds here ... because, be nice ! */
    usleep(2.5 * 1000000);
} while (true);
?>