程序设计 – PHPor 的Blog

PHP redis session 阻塞问题

13,998

8月 152018

现象

php 进程： THREAD_PHP_RD_\Service\Process\Deploy 不干活了，进程堆栈如下：

# pstack 157380
#0 0x00007f4396a440d8 in poll () from /lib64/libc.so.6
#1 0x0000000000745639 in php_sockop_read ()
#2 0x00007f438c90c601 in php_openssl_sockop_io () from /data1/php/lib/php/extensions/no-debug-non-zts-20121212/openssl.so
#3 0x000000000073aed4 in _php_stream_fill_read_buffer ()
#4 0x000000000073b073 in _php_stream_get_line ()
#5 0x00007f438c4d938f in redis_sock_read () from /data1/php/lib/php/extensions/no-debug-non-zts-20121212/redis.so
#6 0x00007f438c4db9dc in ps_write_redis () from /data1/php/lib/php/extensions/no-debug-non-zts-20121212/redis.so
#7 0x0000000000613090 in php_session_save_current_state ()
#8 0x00000000006143d6 in zm_deactivate_session ()
#9 0x0000000000787914 in zend_deactivate_modules ()
#10 0x0000000000722f45 in php_request_shutdown ()
#11 0x000000000083200f in do_cli ()
#12 0x0000000000833478 in main ()

# pstack 157380

#0 0x00007f4396a440d8 in poll () from /lib64/libc.so.6

#1 0x0000000000745639 in php_sockop_read ()

#2 0x00007f438c90c601 in php_openssl_sockop_io () from /data1/php/lib/php/extensions/no-debug-non-zts-20121212/openssl.so

#3 0x000000000073aed4 in _php_stream_fill_read_buffer ()

#4 0x000000000073b073 in _php_stream_get_line ()

#5 0x00007f438c4d938f in redis_sock_read () from /data1/php/lib/php/extensions/no-debug-non-zts-20121212/redis.so

#6 0x00007f438c4db9dc in ps_write_redis () from /data1/php/lib/php/extensions/no-debug-non-zts-20121212/redis.so

#7 0x0000000000613090 in php_session_save_current_state ()

#8 0x00000000006143d6 in zm_deactivate_session ()

#9 0x0000000000787914 in zend_deactivate_modules ()

#10 0x0000000000722f45 in php_request_shutdown ()

#11 0x000000000083200f in do_cli ()

#12 0x0000000000833478 in main ()

4个cli进程都阻塞在了这里；

大概意思为，PHP要回写session，先要一些其它交互，期望能读一行，但是，怎么也读不够一行，于是就死等

问题1：

如果没有什么意外的话，不应该让死等

问题2：

debug发现，这个stream的数据结构中，设置的超时时间为86400秒:

难怪一直阻塞住不干活。

解决办法：

添加超时时间的配置：

session.save_path = "tcp://127.0.0.1:6379?weight=1&timeout=3

1	session.save_path = "tcp://127.0.0.1:6379?weight=1&timeout=3

问题：

你问我gdb的时候是如何知道stream中如何获取超时时间的？如何知道stream->abstract 的真实数据类型的？

翻代码看到的：

注：还以为此事为偶发，进程刚刚工作几个小时就全部进入到了这个场景了,还好我解决了；其实，这并没有找到问题的根源，不找到原因的话，后续会有更加莫名其妙的的事情发生的

Redis 版本号

（然而现在该扩展已经开发到了4.x.y 了

原文链接：https://phpor.net/blog/post/9484

PHP 并发执行命令之multi_cmd

PHP No Responses »

3,445

5月 152018

代码功能：

同时执行多条命令，花费时间接近于最长那条命令的时间，而不是每条命令执行的时间和。

<?php

function multi_cmd($arr) {
	$arrResult = array();
	$arrFp = array();
	foreach($arr as  $k => $cmd) {
		$arrFp[$k] = popen($cmd, "r");
		if(!$arrFp[$k]) {
			$arrResult[$k] = array("stdout" => null, "meta"=>array("elapse"=>0));
			unset($arrFp[$k]);
			continue;
		}
		stream_set_blocking($arrFp[$k], false);
	}
	$start = microtime(1);
	$write = null;
	$expect = null;
	while(count($arrFp) > 0) {
		$arrRead = array_values($arrFp);
		$ret = stream_select($arrRead, $write, $expect, 0, 200000);
		if($ret === false) break;
		if($ret === 0) continue;
		foreach($arrFp as $k=>$fp) {
			if (!in_array($fp, $arrRead)) continue;
			while(!feof($fp)) {
				$r = fread($fp, 1024);
				if ($r !== "" && $r !== false) $arrResult[$k]["stdout"] .= $r;
				if (feof($fp) || $r === false) {
					unset($arrFp[$k]);
					$arrResult[$k]["meta"]["elapse"] = microtime(1) - $start;
					break;
				}
				if ($r === "") break;
			}
		}
	}
	return $arrResult;
}
$arr = array(
	"cmd1" => "sleep 1; eacho 1",
	"cmd2" => "sleep 2; echo 2",
	"cmd3" => "sleep 3; echo 3",
);
$arrResult = multi_cmd($arr);
print_r($arrResult);

<?php

function multi_cmd($arr) {

$arrResult = array();

$arrFp = array();

foreach($arr as $k => $cmd) {

$arrFp[$k] = popen($cmd, "r");

if(!$arrFp[$k]) {

$arrResult[$k] = array("stdout" => null, "meta"=>array("elapse"=>0));

unset($arrFp[$k]);

continue;

}

stream_set_blocking($arrFp[$k], false);

}

$start = microtime(1);

$write = null;

$expect = null;

while(count($arrFp) > 0) {

$arrRead = array_values($arrFp);

$ret = stream_select($arrRead, $write, $expect, 0, 200000);

if($ret === false) break;

if($ret === 0) continue;

foreach($arrFp as $k=>$fp) {

if (!in_array($fp, $arrRead)) continue;

while(!feof($fp)) {

$r = fread($fp, 1024);

if ($r !== "" && $r !== false) $arrResult[$k]["stdout"] .= $r;

if (feof($fp) || $r === false) {

unset($arrFp[$k]);

$arrResult[$k]["meta"]["elapse"] = microtime(1) - $start;

break;

}

if ($r === "") break;

}

return $arrResult;

}

$arr = array(

"cmd1" => "sleep 1; eacho 1",

"cmd2" => "sleep 2; echo 2",

"cmd3" => "sleep 3; echo 3",

);

$arrResult = multi_cmd($arr);

print_r($arrResult);

原文链接：https://phpor.net/blog/post/8328

gops 工作原理

go No Responses »

8,142

11月 152017

gops用来查看系统中存在的go进程。

注意事项：

gops有可能没发现你的go进程：

一方面可能进程是藏在容器中的，在宿主机上gops就发现不了

另一种情况，就是gops参考的是进程文件的特征，该特征在不同go版本可能不一样，比如，我的gops能发现go 1.11的go进程，就发现不了go 1.14的go进程

如下：

gops能list出来系统中的go进程，如果内置了agent的话，还能添加个星号，带星号的进程可以进行深度的分析（因为有agent呀）

那么，gops是如何找到所有go进程，又是如何识别出来agent的呢？

关键代码是这么写的：

// isGo looks up the runtime.buildVersion symbol
// in the process' binary and determines if the process
// if a Go process or not. If the process is a Go process,
// it reports PID, binary name and full path of the binary.
func isGo(pr ps.Process) (path, version string, agent, ok bool, err error) {
	if pr.Pid() == 0 {
		// ignore system process
		return
	}
	path, err = pr.Path()
	if err != nil {
		return
	}
	var versionInfo goversion.Version
	versionInfo, err = goversion.ReadExe(path)
	if err != nil {
		return
	}
	ok = true
	version = versionInfo.Release
	pidfile, err := internal.PIDFile(pr.Pid())
	if err == nil {
		_, err := os.Stat(pidfile)
		agent = err == nil
	}
	return path, version, agent, ok, nil
}

// isGo looks up the runtime.buildVersion symbol

// in the process' binary and determines if the process

// if a Go process or not. If the process is a Go process,

// it reports PID, binary name and full path of the binary.

func isGo(pr ps.Process) (path, version string, agent, ok bool, err error) {

if pr.Pid() == 0 {

// ignore system process

return

}

path, err = pr.Path()

if err != nil {

return

}

var versionInfo goversion.Version

versionInfo, err = goversion.ReadExe(path)

if err != nil {

return

}

ok = true

version = versionInfo.Release

pidfile, err := internal.PIDFile(pr.Pid())

if err == nil {

_, err := os.Stat(pidfile)

agent = err == nil

}

return path, version, agent, ok, nil

}

拿到所有的系统进程
分析进程文件可以检测是否go进程
在进程的用户目录下查找 .config/gops/{$pid} 文件，如果存在，就认为是有内置了agent了，该文件中存放的是agentlisten的端口号

如下：

原文链接：https://phpor.net/blog/post/6760

bash 之携程（coproc)

Linux & Unix, shell No Responses »

5,945

8月 292017

参考：

https://unix.stackexchange.com/questions/86270/how-do-you-use-the-command-coproc-in-bash

http://www.gnu.org/software/bash/manual/bashref.html#Coprocesses

原文链接：https://phpor.net/blog/post/5755

python web开发框架

Django, python No Responses »

5,229

8月 242017

Python web开发框架：

Django 、Flask、Tornado、Bottle、Twisted

Tornado: http://www.tornadoweb.org/en/stable/

Tornado is a Python web framework and asynchronous networking library, originally developed at FriendFeed.

Twisted: https://pypi.python.org/pypi/Twisted/ http://twistedmatrix.com/trac/

An asynchronous networking framework written in Python

twisted/tornado 更是一个server，框架部分只是方便使用该server而附带的，不是重点；你可以选择使用tornado server + Django框架，而对于Bottle框架而言，是允许指定使用哪种server的，支持的server有：

server_names = {
    'cgi': CGIServer,
    'flup': FlupFCGIServer,
    'wsgiref': WSGIRefServer,
    'waitress': WaitressServer,
    'cherrypy': CherryPyServer,
    'paste': PasteServer,
    'fapws3': FapwsServer,
    'tornado': TornadoServer,
    'gae': AppEngineServer,
    'twisted': TwistedServer,
    'diesel': DieselServer,
    'meinheld': MeinheldServer,
    'gunicorn': GunicornServer,
    'eventlet': EventletServer,
    'gevent': GeventServer,
    'geventSocketIO':GeventSocketIOServer,
    'rocket': RocketServer,
    'bjoern' : BjoernServer,
    'auto': AutoServer,
}

server_names = {

'cgi': CGIServer,

'flup': FlupFCGIServer,

'wsgiref': WSGIRefServer,

'waitress': WaitressServer,

'cherrypy': CherryPyServer,

'paste': PasteServer,

'fapws3': FapwsServer,

'tornado': TornadoServer,

'gae': AppEngineServer,

'twisted': TwistedServer,

'diesel': DieselServer,

'meinheld': MeinheldServer,

'gunicorn': GunicornServer,

'eventlet': EventletServer,

'gevent': GeventServer,

'geventSocketIO':GeventSocketIOServer,

'rocket': RocketServer,

'bjoern' : BjoernServer,

'auto': AutoServer,

}

Bottle默认使用的server为：如下：

def run(app=None, server='wsgiref', host='127.0.0.1', port=8080,
        interval=1, reloader=False, quiet=False, plugins=None,
        debug=None, **kargs):

def run(app=None, server='wsgiref', host='127.0.0.1', port=8080,

interval=1, reloader=False, quiet=False, plugins=None,

debug=None, **kargs):

所以，有些技术不是同一个层面的，不能相提并论，非此即彼；

Django or Flask：

https://www.zhihu.com/question/33538127

http://bottlepy.org/docs/dev/index.html

http://www.cnblogs.com/suzhigang/p/6208244.html

原文链接：https://phpor.net/blog/post/5718

logstash-output-syslog 之 facility定义问题

Logstash, ruby No Responses »

7,928

8月 222017

问题：

当我们在logstash-output-syslog中配置facility为local6的时候，实际上走的是local4；而定义facility为local7的时候，实际上走的是local5，但是，定义为daemon的时候，确实是daemon

原因：

参考源码： https://github.com/phpor/logstash-output-syslog/blob/master/lib/logstash/outputs/syslog.rb 关于facility的定义有一个数组：

  FACILITY_LABELS = [
    "kernel",
    "user-level",
    "mail",
    "daemon",
    "security/authorization",
    "syslogd",
    "line printer",
    "network news",
    "uucp",
    "clock",
    "ftp",
    "ntp",
    "log audit",
    "log alert",
    "local0",
    "local1",
    "local2",
    "local3",
    "local4",
    "local5",
    "local6",
    "local7",
  ]

FACILITY_LABELS = [

"kernel",

"user-level",

"mail",

"daemon",

"security/authorization",

"syslogd",

"line printer",

"network news",

"uucp",

"clock",

"ftp",

"ntp",

"log audit",

"log alert",

"local0",

"local1",

"local2",

"local3",

"local4",

"local5",

"local6",

"local7",

]

而rfc中的定义如下： https://tools.ietf.org/html/rfc3164#section-4.1.1

       Numerical             Facility
          Code

           0             kernel messages
           1             user-level messages
           2             mail system
           3             system daemons
          <strong> 4             security/authorization messages (note 1)</strong>



<span class="grey">Lonvick                      Informational                      [Page 8]</span>

Numerical Facility

Code

0 kernel messages

1 user-level messages

2 mail system

3 system daemons

4 security/authorization messages (note 1)

Lonvick Informational [Page 8]

<a id="page-9" class="invisible" href="https://tools.ietf.org/html/rfc3164#page-9" name="page-9"> </a>
<span class="grey"><a href="https://tools.ietf.org/html/rfc3164">RFC 3164</a>                The BSD syslog Protocol              August 2001</span>


           5             messages generated internally by syslogd
           6             line printer subsystem
           7             network news subsystem
           8             UUCP subsystem
         <strong>  9             clock daemon (note 2)</strong>
         <strong> 10         </strong>    <strong>security/authorization messages (note 1)</strong>
          11             FTP daemon
          12             NTP subsystem
          13             log audit (note 1)
          14             log alert (note 1)
         <strong> 15             clock daemon (note 2)</strong>
          16             local use 0  (local0)
          17             local use 1  (local1)
          18             local use 2  (local2)
          19             local use 3  (local3)
          20             local use 4  (local4)
          21             local use 5  (local5)
          22             local use 6  (local6)
          23             local use 7  (local7)

<a href="https://tools.ietf.org/html/rfc3164">RFC 3164</a> The BSD syslog Protocol August 2001

5 messages generated internally by syslogd

6 line printer subsystem

7 network news subsystem

8 UUCP subsystem

9 clock daemon (note 2)

10 security/authorization messages (note 1)

11 FTP daemon

12 NTP subsystem

13 log audit (note 1)

14 log alert (note 1)

15 clock daemon (note 2)

16 local use 0 (local0)

17 local use 1 (local1)

18 local use 2 (local2)

19 local use 3 (local3)

20 local use 4 (local4)

21 local use 5 (local5)

22 local use 6 (local6)

23 local use 7 (local7)

我们发现，4和10的定义是重复的，9和15的定义是重复的；而插件源码中把这两个重复的条目就简单去掉了，于是就导致facility-code就对不上了

解决办法：把重复的两项补回去就可以了

原文链接：https://phpor.net/blog/post/5699

ngx.resp.get_headers()

lua, Nginx No Responses »

5,689

8月 022017

syntax: headers = ngx.resp.get_headers(max_headers?, raw?)

context: set_by_lua*, rewrite_by_lua*, access_by_lua*, content_by_lua*, header_filter_by_lua*, body_filter_by_lua*, log_by_lua*, balancer_by_lua*

Returns a Lua table holding all the current response headers for the current request.

 <span class="pl-k">local</span> h <span class="pl-k">=</span> ngx.<span class="pl-smi">resp</span>.<span class="pl-c1">get_headers</span>()
 <span class="pl-k">for</span> k, v <span class="pl-k">in</span> <span class="pl-c1">pairs</span>(h) <span class="pl-k">do</span>
     <span class="pl-c1">...</span>
 <span class="pl-k">end</span>

				
					
				1
2
3
4

						 <span class="pl-k">local</span> h <span class="pl-k">=</span> ngx.<span class="pl-smi">resp</span>.<span class="pl-c1">get_headers</span>()
 <span class="pl-k">for</span> k, v <span class="pl-k">in</span> <span class="pl-c1">pairs</span>(h) <span class="pl-k">do</span>
     <span class="pl-c1">...</span>
 <span class="pl-k">end</span>

					

			

This function has the same signature as ngx.req.get_headers except getting response headers instead of request headers.

参考: https://github.com/openresty/lua-nginx-module#ngxrespget_headers

注意：字段名都会转换成小写

expires:Thu, 19 Nov 1981 08:52:00 GMT
content-type:text/html; charset=utf-8
connection:keep-alive
cache-control:no-store, no-cache, must-revalidate, post-check=0, pre-check=0
pragma:no-cache
content-encoding:gzip

expires:Thu, 19 Nov 1981 08:52:00 GMT

content-type:text/html; charset=utf-8

connection:keep-alive

cache-control:no-store, no-cache, must-revalidate, post-check=0, pre-check=0

pragma:no-cache

content-encoding:gzip

原文链接：https://phpor.net/blog/post/5602

golang 中的只读变量

go No Responses »

4,736

7月 182017

示例：

// Copyright 2011 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package proxy

import (
	"net"
)

type direct struct{}

// Direct is a direct proxy: one that makes network connections directly.
var Direct = direct{}

func (direct) Dial(network, addr string) (net.Conn, error) {
	return net.Dial(network, addr)
}

// Use of this source code is governed by a BSD-style

// license that can be found in the LICENSE file.

package proxy

import (

"net"

)

type direct struct{}

// Direct is a direct proxy: one that makes network connections directly.

var Direct = direct{}

func (direct) Dial(network, addr string) (net.Conn, error) {

return net.Dial(network, addr)

}

上面是 golang.org/x/net/proxy/direct.go 里面的代码

我们显然是可以在proxy包外访问到Direct变量的
我们显然是无法在proxy包外new 一个 direct 结构变量的
由于2，而且我们无法改变Direct的变量类型，所以，我们做不到给Direct进行重新赋值

原文链接：https://phpor.net/blog/post/5522

php 获取数组中前N个值

PHP No Responses »

4,122

6月 072017

如下，获取数组中的前2个元素：

bash-4.1# php -r 'print_r(array_chunk(array('a'=>1,'b'=>2,'c'=>3),2,true)[0]);'
Array
(
    [a] => 1
    [b] => 2
)

bash-4.1# php -r 'print_r(array_chunk(array('a'=>1,'b'=>2,'c'=>3),2,true)[0]);'

Array

(

[a] => 1

[b] => 2

)

难道还有更方便的方法？

原文链接：https://phpor.net/blog/post/5307

PHP 协程示例

PHP No Responses »

3,128

4月 102017

在PHP没有协程的时候，我们也玩过并发多线程，但是，对于结果的实时处理就没那么方便了（尽管也可以实现），有了协程之后，代码看起来就会舒服许多，参考下面的multi_cmd.php

multi_cmd.php :

<?php
$arr_task = array(
		array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 1 && echo -n succ 1"'),
		array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 2 && echo -n succ 2"'),
		array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 3 && echo -n succ 3"'),
		);
$start = microtime(1);
$arr_result = multi_cmd($arr_task);
foreach($arr_result as $k=>$v) {
	echo "$k: {$v['cmd']}\n";
	echo "\tstdout: ".$v['output']['stdout']."\n";
	echo "\tstderr: ".$v['output']['stderr']."\n";
}
$end = microtime(1);
echo "time_used:". ($end - $start) ."\n";

function multi_cmd($arr_task) {
	$arr_result = array();
	foreach ($arr_task as $k => $arr) {
		$cmd = $arr['cmd'];
		$arr_result[$k] = array("cmd"=>$cmd, "result"=>0, "output"=>array("stdout"=>null, "stderr"=>null), "pipes"=>array());
		$descriptorspec[$k] = array(
				0 => array("pipe", "r"),  // stdin is a pipe that the child will read from
				1 => array("pipe", "w"),  // stdout is a pipe that the child will write to
				2 => array("pipe", "w") // stderr is a file to write to
				);


		$process[$k] = proc_open($cmd, $descriptorspec[$k], $arr_result[$k]["pipes"]);
		if (!is_resource($process[$k])) {
			$arr_result[$k] = 3;
			continue;
		}
		stream_set_blocking($arr_result[$k]["pipes"][1], 0);
		stream_set_blocking($arr_result[$k]["pipes"][2], 0);

	}
	while(!empty($arr_result)) {
		foreach($arr_result as $k=>&$v) {
			if ($v['result'] == 3) {
				yield $k=>$v;
				unset($arr_result[$k]);
				continue;
			}
			$stdout = $arr_result[$k]["pipes"][1];
			$stderr = $arr_result[$k]["pipes"][2];
			if ($v['result'] != 1) {
				$str = @fread($stdout, 1024);
				$arr_result[$k]["output"]["stdout"] .= $str;
				if (@feof($stdout)) {
					$v['result'] += 1;
					fclose($stdout);
				}
			}
			if ($v['result'] != 2) {
				$str = @fread($stderr, 1024);
				$arr_result[$k]["output"]["stderr"] .= $str;
				if (@feof($stderr)) {
					$v['result'] += 2;
					fclose($stderr);
				}
			}

		}
	}
}

<?php

$arr_task = array(

array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 1 && echo -n succ 1"'),

array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 2 && echo -n succ 2"'),

array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 3 && echo -n succ 3"'),

);

$start = microtime(1);

$arr_result = multi_cmd($arr_task);

foreach($arr_result as $k=>$v) {

echo "$k: {$v['cmd']}\n";

echo "\tstdout: ".$v['output']['stdout']."\n";

echo "\tstderr: ".$v['output']['stderr']."\n";

}

$end = microtime(1);

echo "time_used:". ($end - $start) ."\n";

function multi_cmd($arr_task) {

$arr_result = array();

foreach ($arr_task as $k => $arr) {

$cmd = $arr['cmd'];

$arr_result[$k] = array("cmd"=>$cmd, "result"=>0, "output"=>array("stdout"=>null, "stderr"=>null), "pipes"=>array());

$descriptorspec[$k] = array(

0 => array("pipe", "r"), // stdin is a pipe that the child will read from

1 => array("pipe", "w"), // stdout is a pipe that the child will write to

2 => array("pipe", "w") // stderr is a file to write to

);

$process[$k] = proc_open($cmd, $descriptorspec[$k], $arr_result[$k]["pipes"]);

if (!is_resource($process[$k])) {

$arr_result[$k] = 3;

continue;

}

stream_set_blocking($arr_result[$k]["pipes"][1], 0);

stream_set_blocking($arr_result[$k]["pipes"][2], 0);

}

while(!empty($arr_result)) {

foreach($arr_result as $k=>&$v) {

if ($v['result'] == 3) {

yield $k=>$v;

unset($arr_result[$k]);

continue;

}

$stdout = $arr_result[$k]["pipes"][1];

$stderr = $arr_result[$k]["pipes"][2];

if ($v['result'] != 1) {

$str = @fread($stdout, 1024);

$arr_result[$k]["output"]["stdout"] .= $str;

if (@feof($stdout)) {

$v['result'] += 1;

fclose($stdout);

}

if ($v['result'] != 2) {

$str = @fread($stderr, 1024);

$arr_result[$k]["output"]["stderr"] .= $str;

if (@feof($stderr)) {

$v['result'] += 2;

fclose($stderr);

}

我们见到的更多的可能是并发执行多个任务，每个任务都完成后（如： curl异步多请求）在一并处理结果，如果处理结果本身也是需要时间的话，就会比较浪费时间，如果能完成一个处理一个的结果，效果会好一些

上面脚本存在瑕疵：

1、为了避免某个流阻塞整个进程，上面使用了非阻塞；但是，后面的死循环却导致了大量cpu的占用，所以，考虑使用stream_select会更好一些

2、为了能控制并发数就更好了，并发太大也可能不是好事

改进的脚本如下：

<?php
$arr_task = array(
		array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 2 && echo -n succ 1"'),
		array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 2 && echo -n succ 2"'),
		array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 3 && echo -n succ 3"'),
		);
$start = microtime(1);
$arr_result = multi_cmd($arr_task, 2);
foreach($arr_result as $k=>$v) {
	echo "$k: {$v['cmd']}\n";
	echo "\tretcode: ".$v['retcode']."\n";
	echo "\tstdout: ".$v['output']['stdout']."\n";
	echo "\tstderr: ".$v['output']['stderr']."\n";
}
$end = microtime(1);
echo "time_used:". ($end - $start) ."\n";

function add_task(&$arr_task, &$arr_result, $concurent) {
	if (empty($arr_task)) return;
	foreach ($arr_task as $k => $arr) {
		if ($concurent != -1 && count($arr_result) >= $concurent ) return;
		echo "add task $k\n";
		$cmd = $arr['cmd'];
		unset($arr_task[$k]);
		$arr_result[$k] = array("cmd"=>$cmd, "result"=>0, "output"=>array("stdout"=>null, "stderr"=>null), "pipes"=>array());
		$arr_result[$k]['filedesc'] = array(
				0 => array("pipe", "r"),  // stdin is a pipe that the child will read from
				1 => array("pipe", "w"),  // stdout is a pipe that the child will write to
				2 => array("pipe", "w") // stderr is a file to write to
				);


		$arr_result[$k]['handle'] = proc_open($cmd, $arr_result[$k]['filedesc'], $arr_result[$k]["pipes"]);
		if (!is_resource($arr_result[$k]['handle'])) {
			$arr_result[$k] = 3;
			continue;
		}
		//stream_set_blocking($arr_result[$k]["pipes"][1], 0);
		//stream_set_blocking($arr_result[$k]["pipes"][2], 0);
	}
}
function multi_cmd($arr_task, $concurent = -1) {
	$arr_result = array();
	add_task($arr_task, $arr_result, $concurent);
	while(!empty($arr_result)) {
		$arr_read_fds = array();
		foreach($arr_result as $k=>&$v) {
			if ($v['result'] == 3) {
				unset($arr_result[$k]);
				$v['retcode'] = @proc_close($v['handle']);
				add_task($arr_task, $arr_result, $concurent);
				yield $k=>$v;
				continue;
			}
			if (!($v['result'] & 1)) {
				$arr_read_fds[] = $v["pipes"][1];
			}
			if (!($v['result'] & 2)) {
				$arr_read_fds[] = $v["pipes"][2];
			}
		}
		if (empty($arr_read_fds)) break;
		$arr_expect = array();
		$arr_write_fds = $arr_expect_fds = array();
		$arr_fds = $arr_read_fds;
		while( 0 === stream_select($arr_fds, $arr_write_fds, $arr_expect_fds, 0, 500000)) {
			$arr_fds = $arr_read_fds;
		}
		foreach($arr_result as $k=>&$v) {
			$stdout = $arr_result[$k]["pipes"][1];
			$stderr = $arr_result[$k]["pipes"][2];
			if (!($v['result'] & 1) && in_array($stdout, $arr_fds)) {
				$str = @fread($stdout, 1024);
				$arr_result[$k]["output"]["stdout"] .= $str;
				if (@feof($stdout)) {
					$v['result'] |= 1;
					@fclose($stdout);
				}
			}
			if (!($v['result'] & 2) && in_array($stderr, $arr_fds)) {
				$str = @fread($stderr, 1024);
				$arr_result[$k]["output"]["stderr"] .= $str;
				if (@feof($stderr)) {
					$v['result'] |= 2;
					@fclose($stderr);
				}
			}

		}
	}
}

<?php

$arr_task = array(

array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 2 && echo -n succ 1"'),

array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 2 && echo -n succ 2"'),

array('cmd'=>'ssh -i /home/phpor/.ssh/id_rsa phpor@localhost "sleep 3 && echo -n succ 3"'),

);

$start = microtime(1);

$arr_result = multi_cmd($arr_task, 2);

foreach($arr_result as $k=>$v) {

echo "$k: {$v['cmd']}\n";

echo "\tretcode: ".$v['retcode']."\n";

echo "\tstdout: ".$v['output']['stdout']."\n";

echo "\tstderr: ".$v['output']['stderr']."\n";

}

$end = microtime(1);

echo "time_used:". ($end - $start) ."\n";

function add_task(&$arr_task, &$arr_result, $concurent) {

if (empty($arr_task)) return;

foreach ($arr_task as $k => $arr) {

if ($concurent != -1 && count($arr_result) >= $concurent ) return;

echo "add task $k\n";

$cmd = $arr['cmd'];

unset($arr_task[$k]);

$arr_result[$k] = array("cmd"=>$cmd, "result"=>0, "output"=>array("stdout"=>null, "stderr"=>null), "pipes"=>array());

$arr_result[$k]['filedesc'] = array(

0 => array("pipe", "r"), // stdin is a pipe that the child will read from

1 => array("pipe", "w"), // stdout is a pipe that the child will write to

2 => array("pipe", "w") // stderr is a file to write to

);

$arr_result[$k]['handle'] = proc_open($cmd, $arr_result[$k]['filedesc'], $arr_result[$k]["pipes"]);

if (!is_resource($arr_result[$k]['handle'])) {

$arr_result[$k] = 3;

continue;

}

//stream_set_blocking($arr_result[$k]["pipes"][1], 0);

//stream_set_blocking($arr_result[$k]["pipes"][2], 0);

}

function multi_cmd($arr_task, $concurent = -1) {

$arr_result = array();

add_task($arr_task, $arr_result, $concurent);

while(!empty($arr_result)) {

$arr_read_fds = array();

foreach($arr_result as $k=>&$v) {

if ($v['result'] == 3) {

unset($arr_result[$k]);

$v['retcode'] = @proc_close($v['handle']);

add_task($arr_task, $arr_result, $concurent);

yield $k=>$v;

continue;

}

if (!($v['result'] & 1)) {

$arr_read_fds[] = $v["pipes"][1];

}

if (!($v['result'] & 2)) {

$arr_read_fds[] = $v["pipes"][2];

}

if (empty($arr_read_fds)) break;

$arr_expect = array();

$arr_write_fds = $arr_expect_fds = array();

$arr_fds = $arr_read_fds;

while( 0 === stream_select($arr_fds, $arr_write_fds, $arr_expect_fds, 0, 500000)) {

$arr_fds = $arr_read_fds;

}

foreach($arr_result as $k=>&$v) {

$stdout = $arr_result[$k]["pipes"][1];

$stderr = $arr_result[$k]["pipes"][2];

if (!($v['result'] & 1) && in_array($stdout, $arr_fds)) {

$str = @fread($stdout, 1024);

$arr_result[$k]["output"]["stdout"] .= $str;

if (@feof($stdout)) {

$v['result'] |= 1;

@fclose($stdout);

}

if (!($v['result'] & 2) && in_array($stderr, $arr_fds)) {

$str = @fread($stderr, 1024);

$arr_result[$k]["output"]["stderr"] .= $str;

if (@feof($stderr)) {

$v['result'] |= 2;

@fclose($stderr);

}

注意：

在使用stream_select 的时候，是否阻塞也就不重要了? 也不完全，加入其中一个比较靠前的任务执行时间很长，就算第一批的大部分任务执行时间都很短，也会因为fread而阻塞在执行时间长的任务身上而无法快速完成其它任务，进而加入更多的任务；所以，这里可能的办法有：

办法一：给stderr和stdout添加read timeout限制，但是测试发现，stream_set_timeout 给这两个流添加超时时间是失败的，stream_set_timeout 返回false
办法二：依然使用非阻塞模式, 依然不行，尽管stream_set_blocking(stdout, false) 返回true，也是无效的
注意：

也就是说仅有的两个函数都是不能用的，也就是说，不是函数不行，而是，这事儿就不可行

原文链接：https://phpor.net/blog/post/4972

Older Entries