Worker.php 49 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540
  1. <?php
  2. /**
  3. * This file is part of workerman.
  4. *
  5. * Licensed under The MIT License
  6. * For full copyright and license information, please see the MIT-LICENSE.txt
  7. * Redistributions of files must retain the above copyright notice.
  8. *
  9. * @author walkor<walkor@workerman.net>
  10. * @copyright walkor<walkor@workerman.net>
  11. * @link http://www.workerman.net/
  12. * @license http://www.opensource.org/licenses/mit-license.php MIT License
  13. */
  14. namespace Workerman;
  15. require_once __DIR__.'/Lib/Constants.php';
  16. use \Workerman\Events\Libevent;
  17. use \Workerman\Events\Select;
  18. use \Workerman\Events\EventInterface;
  19. use \Workerman\Connection\ConnectionInterface;
  20. use \Workerman\Connection\TcpConnection;
  21. use \Workerman\Connection\UdpConnection;
  22. use \Workerman\Lib\Timer;
  23. use \Exception;
  24. /**
  25. * Worker 类
  26. * 是一个容器,用于监听端口,维持客户端连接
  27. */
  28. class Worker
  29. {
  30. /**
  31. * 版本号
  32. * @var string
  33. */
  34. const VERSION = '3.2.6';
  35. /**
  36. * 状态 启动中
  37. * @var int
  38. */
  39. const STATUS_STARTING = 1;
  40. /**
  41. * 状态 运行中
  42. * @var int
  43. */
  44. const STATUS_RUNNING = 2;
  45. /**
  46. * 状态 停止
  47. * @var int
  48. */
  49. const STATUS_SHUTDOWN = 4;
  50. /**
  51. * 状态 平滑重启中
  52. * @var int
  53. */
  54. const STATUS_RELOADING = 8;
  55. /**
  56. * 给子进程发送重启命令 KILL_WORKER_TIMER_TIME 秒后
  57. * 如果对应进程仍然未重启则强行杀死
  58. * @var int
  59. */
  60. const KILL_WORKER_TIMER_TIME = 1;
  61. /**
  62. * 默认的backlog,即内核中用于存放未被进程认领(accept)的连接队列长度
  63. * @var int
  64. */
  65. const DEFAUL_BACKLOG = 1024;
  66. /**
  67. * udp最大包长
  68. * @var int
  69. */
  70. const MAX_UDP_PACKEG_SIZE = 65535;
  71. /**
  72. * worker id
  73. * @var int
  74. */
  75. public $id = 0;
  76. /**
  77. * worker的名称,用于在运行status命令时标记进程
  78. * @var string
  79. */
  80. public $name = 'none';
  81. /**
  82. * 设置当前worker实例的进程数
  83. * @var int
  84. */
  85. public $count = 1;
  86. /**
  87. * 设置当前worker进程的运行用户,需要root超级权限
  88. * @var string
  89. */
  90. public $user = '';
  91. /**
  92. * 设置当前worker进程的运行用户组,需要root超级权限
  93. * @var string
  94. */
  95. public $group = '';
  96. /**
  97. * 设置当前worker进程的文件系统根目录,需要root超级权限
  98. * @var string
  99. */
  100. public $chroot = '';
  101. /**
  102. * 当前worker进程是否可以平滑重启
  103. * @var bool
  104. */
  105. public $reloadable = true;
  106. /**
  107. * reuse port
  108. * @var bool
  109. */
  110. public $reusePort = false;
  111. /**
  112. * 当worker进程启动时,如果设置了$onWorkerStart回调函数,则运行
  113. * 此钩子函数一般用于进程启动后初始化工作
  114. * @var callback
  115. */
  116. public $onWorkerStart = null;
  117. /**
  118. * 当有客户端连接时,如果设置了$onConnect回调函数,则运行
  119. * @var callback
  120. */
  121. public $onConnect = null;
  122. /**
  123. * 当客户端连接上发来数据时,如果设置了$onMessage回调,则运行
  124. * @var callback
  125. */
  126. public $onMessage = null;
  127. /**
  128. * 当客户端的连接关闭时,如果设置了$onClose回调,则运行
  129. * @var callback
  130. */
  131. public $onClose = null;
  132. /**
  133. * 当客户端的连接发生错误时,如果设置了$onError回调,则运行
  134. * 错误一般为客户端断开连接导致数据发送失败、服务端的发送缓冲区满导致发送失败等
  135. * 具体错误码及错误详情会以参数的形式传递给回调,参见手册
  136. * @var callback
  137. */
  138. public $onError = null;
  139. /**
  140. * 当连接的发送缓冲区满时,如果设置了$onBufferFull回调,则执行
  141. * @var callback
  142. */
  143. public $onBufferFull = null;
  144. /**
  145. * 当链接的发送缓冲区被清空时,如果设置了$onBufferDrain回调,则执行
  146. * @var callback
  147. */
  148. public $onBufferDrain = null;
  149. /**
  150. * 当前进程退出时(由于平滑重启或者服务停止导致),如果设置了此回调,则运行
  151. * @var callback
  152. */
  153. public $onWorkerStop = null;
  154. /**
  155. * 当收到reload命令时的回调函数
  156. * @var callback
  157. */
  158. public $onWorkerReload = null;
  159. /**
  160. * 传输层协议
  161. * @var string
  162. */
  163. public $transport = 'tcp';
  164. /**
  165. * 所有的客户端连接
  166. * @var array
  167. */
  168. public $connections = array();
  169. /**
  170. * 应用层协议,由初始化worker时指定
  171. * 例如 new worker('http://0.0.0.0:8080');指定使用http协议
  172. * @var string
  173. */
  174. protected $_protocol = '';
  175. /**
  176. * 当前worker实例初始化目录位置,用于设置应用自动加载的根目录
  177. * @var string
  178. */
  179. protected $_appInitPath = '';
  180. /**
  181. * 是否以守护进程的方式运行。运行start时加上-d参数会自动以守护进程方式运行
  182. * 例如 php start.php start -d
  183. * @var bool
  184. */
  185. public static $daemonize = false;
  186. /**
  187. * 重定向标准输出,即将所有echo、var_dump等终端输出写到对应文件中
  188. * 注意 此参数只有在以守护进程方式运行时有效
  189. * @var string
  190. */
  191. public static $stdoutFile = '/dev/null';
  192. /**
  193. * pid文件的路径及名称
  194. * 例如 Worker::$pidFile = '/tmp/workerman.pid';
  195. * 注意 此属性一般不必手动设置,默认会放到php临时目录中
  196. * @var string
  197. */
  198. public static $pidFile = '';
  199. /**
  200. * 日志目录,默认在workerman根目录下,与Applications同级
  201. * 可以手动设置
  202. * 例如 Worker::$logFile = '/tmp/workerman.log';
  203. * @var mixed
  204. */
  205. public static $logFile = '';
  206. /**
  207. * 全局事件轮询库,用于监听所有资源的可读可写事件
  208. * @var Select/Libevent
  209. */
  210. public static $globalEvent = null;
  211. /**
  212. * 主进程pid
  213. * @var int
  214. */
  215. protected static $_masterPid = 0;
  216. /**
  217. * 监听的socket
  218. * @var resource
  219. */
  220. protected $_mainSocket = null;
  221. /**
  222. * socket名称,包括应用层协议+ip+端口号,在初始化worker时设置
  223. * 值类似 http://0.0.0.0:80
  224. * @var string
  225. */
  226. protected $_socketName = '';
  227. /**
  228. * socket的上下文,具体选项设置可以在初始化worker时传递
  229. * @var array
  230. */
  231. protected $_context = null;
  232. /**
  233. * 所有的worker实例
  234. * @var array
  235. */
  236. protected static $_workers = array();
  237. /**
  238. * 所有worker进程的pid
  239. * 格式为 [worker_id=>[pid=>pid, pid=>pid, ..], ..]
  240. * @var array
  241. */
  242. protected static $_pidMap = array();
  243. /**
  244. * 所有需要重启的进程pid
  245. * 格式为 [pid=>pid, pid=>pid]
  246. * @var array
  247. */
  248. protected static $_pidsToRestart = array();
  249. /**
  250. * 所有进程pid到id的映射
  251. * 格式为[worker_id=>[0=>$pid, 1=>$pid, ..], ..]
  252. * @var array
  253. */
  254. protected static $_idMap = array();
  255. /**
  256. * 当前worker状态
  257. * @var int
  258. */
  259. protected static $_status = self::STATUS_STARTING;
  260. /**
  261. * 所有worke名称(name属性)中的最大长度,用于在运行 status 命令时格式化输出
  262. * @var int
  263. */
  264. protected static $_maxWorkerNameLength = 12;
  265. /**
  266. * 所有socket名称(_socketName属性)中的最大长度,用于在运行 status 命令时格式化输出
  267. * @var int
  268. */
  269. protected static $_maxSocketNameLength = 12;
  270. /**
  271. * 所有user名称(user属性)中的最大长度,用于在运行 status 命令时格式化输出
  272. * @var int
  273. */
  274. protected static $_maxUserNameLength = 12;
  275. /**
  276. * 运行 status 命令时用于保存结果的文件名
  277. * @var string
  278. */
  279. protected static $_statisticsFile = '';
  280. /**
  281. * 启动的全局入口文件
  282. * 例如 php start.php start ,则入口文件为start.php
  283. * @var string
  284. */
  285. protected static $_startFile = '';
  286. /**
  287. * 全局统计数据,用于在运行 status 命令时展示
  288. * 统计的内容包括 workerman启动的时间戳及每组worker进程的退出次数及退出状态码
  289. * @var array
  290. */
  291. protected static $_globalStatistics = array(
  292. 'start_timestamp' => 0,
  293. 'worker_exit_info' => array()
  294. );
  295. /**
  296. * 运行所有worker实例
  297. * @return void
  298. */
  299. public static function runAll()
  300. {
  301. // 初始化环境变量
  302. self::init();
  303. // 解析命令
  304. self::parseCommand();
  305. // 尝试以守护进程模式运行
  306. self::daemonize();
  307. // 初始化所有worker实例,主要是监听端口
  308. self::initWorkers();
  309. // 初始化所有信号处理函数
  310. self::installSignal();
  311. // 保存主进程pid
  312. self::saveMasterPid();
  313. // 创建子进程(worker进程)并运行
  314. self::forkWorkers();
  315. // 展示启动界面
  316. self::displayUI();
  317. // 尝试重定向标准输入输出
  318. self::resetStd();
  319. // 监控所有子进程(worker进程)
  320. self::monitorWorkers();
  321. }
  322. /**
  323. * 初始化一些环境变量
  324. * @return void
  325. */
  326. protected static function init()
  327. {
  328. // 如果没设置$pidFile,则生成默认值
  329. if(empty(self::$pidFile))
  330. {
  331. $backtrace = debug_backtrace();
  332. self::$_startFile = $backtrace[count($backtrace)-1]['file'];
  333. self::$pidFile = __DIR__ . "/../".str_replace('/', '_', self::$_startFile).".pid";
  334. }
  335. // 没有设置日志文件,则生成一个默认值
  336. if(empty(self::$logFile))
  337. {
  338. self::$logFile = __DIR__ . '/../workerman.log';
  339. }
  340. // 标记状态为启动中
  341. self::$_status = self::STATUS_STARTING;
  342. // 启动时间戳
  343. self::$_globalStatistics['start_timestamp'] = time();
  344. // 设置status文件位置
  345. self::$_statisticsFile = sys_get_temp_dir().'/workerman.status';
  346. // 尝试设置进程名称(需要php>=5.5或者安装了proctitle扩展)
  347. self::setProcessTitle('WorkerMan: master process start_file=' . self::$_startFile);
  348. // 初始化id
  349. self::initId();
  350. // 初始化定时器
  351. Timer::init();
  352. }
  353. /**
  354. * 初始化所有的worker实例,主要工作为获得格式化所需数据及监听端口
  355. * @return void
  356. */
  357. protected static function initWorkers()
  358. {
  359. /** @var static $worker */
  360. foreach(self::$_workers as $worker)
  361. {
  362. // 没有设置worker名称,则使用none代替
  363. if(empty($worker->name))
  364. {
  365. $worker->name = 'none';
  366. }
  367. // 获得所有worker名称中最大长度
  368. $worker_name_length = strlen($worker->name);
  369. if(self::$_maxWorkerNameLength < $worker_name_length)
  370. {
  371. self::$_maxWorkerNameLength = $worker_name_length;
  372. }
  373. // 获得所有_socketName中最大长度
  374. $socket_name_length = strlen($worker->getSocketName());
  375. if(self::$_maxSocketNameLength < $socket_name_length)
  376. {
  377. self::$_maxSocketNameLength = $socket_name_length;
  378. }
  379. // 获得运行用户名的最大长度
  380. if(empty($worker->user))
  381. {
  382. $worker->user = self::getCurrentUser();
  383. }
  384. $user_name_length = strlen($worker->user);
  385. if(self::$_maxUserNameLength < $user_name_length)
  386. {
  387. self::$_maxUserNameLength = $user_name_length;
  388. }
  389. // 如果端口不可复用,则直接在主进程就监听
  390. if(!$worker->reusePort)
  391. {
  392. // 监听端口
  393. $worker->listen();
  394. }
  395. }
  396. }
  397. /**
  398. * 初始化idMap
  399. * return void
  400. */
  401. protected static function initId()
  402. {
  403. foreach(self::$_workers as $worker_id=>$worker)
  404. {
  405. self::$_idMap[$worker_id] = array_fill(0, $worker->count, 0);
  406. }
  407. }
  408. /**
  409. * 获得运行当前进程的用户名
  410. * @return string
  411. */
  412. protected static function getCurrentUser()
  413. {
  414. $user_info = posix_getpwuid(posix_getuid());
  415. return $user_info['name'];
  416. }
  417. /**
  418. * 展示启动界面
  419. * @return void
  420. */
  421. protected static function displayUI()
  422. {
  423. echo "\033[1A\n\033[K-----------------------\033[47;30m WORKERMAN \033[0m-----------------------------\n\033[0m";
  424. echo 'Workerman version:' , Worker::VERSION , " PHP version:",PHP_VERSION,"\n";
  425. echo "------------------------\033[47;30m WORKERS \033[0m-------------------------------\n";
  426. echo "\033[47;30muser\033[0m",str_pad('', self::$_maxUserNameLength+2-strlen('user')), "\033[47;30mworker\033[0m",str_pad('', self::$_maxWorkerNameLength+2-strlen('worker')), "\033[47;30mlisten\033[0m",str_pad('', self::$_maxSocketNameLength+2-strlen('listen')), "\033[47;30mprocesses\033[0m \033[47;30m","status\033[0m\n";
  427. /** @var static $worker */
  428. foreach(self::$_workers as $worker)
  429. {
  430. echo str_pad($worker->user, self::$_maxUserNameLength+2),str_pad($worker->name, self::$_maxWorkerNameLength+2),str_pad($worker->getSocketName(), self::$_maxSocketNameLength+2), str_pad(' '.$worker->count, 9), " \033[32;40m [OK] \033[0m\n";;
  431. }
  432. echo "----------------------------------------------------------------\n";
  433. if(self::$daemonize)
  434. {
  435. global $argv;
  436. $start_file = $argv[0];
  437. echo "Input \"php $start_file stop\" to quit. Start success.\n";
  438. }
  439. else
  440. {
  441. echo "Press Ctrl-C to quit. Start success.\n";
  442. }
  443. }
  444. /**
  445. * 解析运行命令
  446. * php yourfile.php start | stop | restart | reload | status
  447. * @return void
  448. */
  449. protected static function parseCommand()
  450. {
  451. // 检查运行命令的参数
  452. global $argv;
  453. $start_file = $argv[0];
  454. if(!isset($argv[1]))
  455. {
  456. exit("Usage: php yourfile.php {start|stop|restart|reload|status|kill}\n");
  457. }
  458. // 命令
  459. $command = trim($argv[1]);
  460. // 子命令,目前只支持-d
  461. $command2 = isset($argv[2]) ? $argv[2] : '';
  462. // 记录日志
  463. $mode = '';
  464. if($command === 'start')
  465. {
  466. if($command2 === '-d')
  467. {
  468. $mode = 'in DAEMON mode';
  469. }
  470. else
  471. {
  472. $mode = 'in DEBUG mode';
  473. }
  474. }
  475. self::log("Workerman[$start_file] $command $mode");
  476. // 检查主进程是否在运行
  477. $master_pid = @file_get_contents(self::$pidFile);
  478. $master_is_alive = $master_pid && @posix_kill($master_pid, 0);
  479. if($master_is_alive)
  480. {
  481. if($command === 'start')
  482. {
  483. self::log("Workerman[$start_file] already running");
  484. exit;
  485. }
  486. }
  487. elseif($command !== 'start' && $command !== 'restart')
  488. {
  489. self::log("Workerman[$start_file] not run");
  490. }
  491. // 根据命令做相应处理
  492. switch($command)
  493. {
  494. case 'kill':
  495. exec("ps aux | grep $start_file | grep -v grep | awk '{print $2}' |xargs kill -SIGINT");
  496. exec("ps aux | grep $start_file | grep -v grep | awk '{print $2}' |xargs kill -SIGKILL");
  497. break;
  498. // 启动 workerman
  499. case 'start':
  500. if($command2 === '-d')
  501. {
  502. Worker::$daemonize = true;
  503. }
  504. break;
  505. // 显示 workerman 运行状态
  506. case 'status':
  507. // 尝试删除统计文件,避免脏数据
  508. if(is_file(self::$_statisticsFile))
  509. {
  510. @unlink(self::$_statisticsFile);
  511. }
  512. // 向主进程发送 SIGUSR2 信号 ,然后主进程会向所有子进程发送 SIGUSR2 信号
  513. // 所有进程收到 SIGUSR2 信号后会向 $_statisticsFile 写入自己的状态
  514. posix_kill($master_pid, SIGUSR2);
  515. // 睡眠100毫秒,等待子进程将自己的状态写入$_statisticsFile指定的文件
  516. usleep(100000);
  517. // 展示状态
  518. readfile(self::$_statisticsFile);
  519. exit(0);
  520. // 重启 workerman
  521. case 'restart':
  522. // 停止 workeran
  523. case 'stop':
  524. self::log("Workerman[$start_file] is stoping ...");
  525. // 想主进程发送SIGINT信号,主进程会向所有子进程发送SIGINT信号
  526. $master_pid && posix_kill($master_pid, SIGINT);
  527. // 如果 $timeout 秒后主进程没有退出则展示失败界面
  528. $timeout = 5;
  529. $start_time = time();
  530. while(1)
  531. {
  532. // 检查主进程是否存活
  533. $master_is_alive = $master_pid && posix_kill($master_pid, 0);
  534. if($master_is_alive)
  535. {
  536. // 检查是否超过$timeout时间
  537. if(time() - $start_time >= $timeout)
  538. {
  539. self::log("Workerman[$start_file] stop fail");
  540. exit;
  541. }
  542. usleep(10000);
  543. continue;
  544. }
  545. self::log("Workerman[$start_file] stop success");
  546. // 是restart命令
  547. if($command === 'stop')
  548. {
  549. exit(0);
  550. }
  551. // -d 说明是以守护进程的方式启动
  552. if($command2 === '-d')
  553. {
  554. Worker::$daemonize = true;
  555. }
  556. break;
  557. }
  558. break;
  559. // 平滑重启 workerman
  560. case 'reload':
  561. posix_kill($master_pid, SIGUSR1);
  562. self::log("Workerman[$start_file] reload");
  563. exit;
  564. // 未知命令
  565. default :
  566. exit("Usage: php yourfile.php {start|stop|restart|reload|status|kill}\n");
  567. }
  568. }
  569. /**
  570. * 安装信号处理函数
  571. * @return void
  572. */
  573. protected static function installSignal()
  574. {
  575. // stop
  576. pcntl_signal(SIGINT, array('\Workerman\Worker', 'signalHandler'), false);
  577. // reload
  578. pcntl_signal(SIGUSR1, array('\Workerman\Worker', 'signalHandler'), false);
  579. // status
  580. pcntl_signal(SIGUSR2, array('\Workerman\Worker', 'signalHandler'), false);
  581. // ignore
  582. pcntl_signal(SIGPIPE, SIG_IGN, false);
  583. }
  584. /**
  585. * 为子进程重新安装信号处理函数,使用全局事件轮询监听信号
  586. * @return void
  587. */
  588. protected static function reinstallSignal()
  589. {
  590. // uninstall stop signal handler
  591. pcntl_signal(SIGINT, SIG_IGN, false);
  592. // uninstall reload signal handler
  593. pcntl_signal(SIGUSR1, SIG_IGN, false);
  594. // uninstall status signal handler
  595. pcntl_signal(SIGUSR2, SIG_IGN, false);
  596. // reinstall stop signal handler
  597. self::$globalEvent->add(SIGINT, EventInterface::EV_SIGNAL, array('\Workerman\Worker', 'signalHandler'));
  598. // uninstall reload signal handler
  599. self::$globalEvent->add(SIGUSR1, EventInterface::EV_SIGNAL,array('\Workerman\Worker', 'signalHandler'));
  600. // uninstall status signal handler
  601. self::$globalEvent->add(SIGUSR2, EventInterface::EV_SIGNAL, array('\Workerman\Worker', 'signalHandler'));
  602. }
  603. /**
  604. * 信号处理函数
  605. * @param int $signal
  606. */
  607. public static function signalHandler($signal)
  608. {
  609. switch($signal)
  610. {
  611. // stop
  612. case SIGINT:
  613. self::stopAll();
  614. break;
  615. // reload
  616. case SIGUSR1:
  617. self::$_pidsToRestart = self::getAllWorkerPids();
  618. self::reload();
  619. break;
  620. // show status
  621. case SIGUSR2:
  622. self::writeStatisticsToStatusFile();
  623. break;
  624. }
  625. }
  626. /**
  627. * 尝试以守护进程的方式运行
  628. * @throws Exception
  629. */
  630. protected static function daemonize()
  631. {
  632. if(!self::$daemonize)
  633. {
  634. return;
  635. }
  636. umask(0);
  637. $pid = pcntl_fork();
  638. if(-1 === $pid)
  639. {
  640. throw new Exception('fork fail');
  641. }
  642. elseif($pid > 0)
  643. {
  644. exit(0);
  645. }
  646. if(-1 === posix_setsid())
  647. {
  648. throw new Exception("setsid fail");
  649. }
  650. // fork again avoid SVR4 system regain the control of terminal
  651. $pid = pcntl_fork();
  652. if(-1 === $pid)
  653. {
  654. throw new Exception("fork fail");
  655. }
  656. elseif(0 !== $pid)
  657. {
  658. exit(0);
  659. }
  660. }
  661. /**
  662. * 重定向标准输入输出
  663. * @throws Exception
  664. */
  665. protected static function resetStd()
  666. {
  667. if(!self::$daemonize)
  668. {
  669. return;
  670. }
  671. global $STDOUT, $STDERR;
  672. $handle = fopen(self::$stdoutFile,"a");
  673. if($handle)
  674. {
  675. unset($handle);
  676. @fclose(STDOUT);
  677. @fclose(STDERR);
  678. $STDOUT = fopen(self::$stdoutFile,"a");
  679. $STDERR = fopen(self::$stdoutFile,"a");
  680. }
  681. else
  682. {
  683. throw new Exception('can not open stdoutFile ' . self::$stdoutFile);
  684. }
  685. }
  686. /**
  687. * 保存pid到文件中,方便运行命令时查找主进程pid
  688. * @throws Exception
  689. */
  690. protected static function saveMasterPid()
  691. {
  692. self::$_masterPid = posix_getpid();
  693. if(false === @file_put_contents(self::$pidFile, self::$_masterPid))
  694. {
  695. throw new Exception('can not save pid to ' . self::$pidFile);
  696. }
  697. }
  698. /**
  699. * 获得所有子进程的pid
  700. * @return array
  701. */
  702. protected static function getAllWorkerPids()
  703. {
  704. $pid_array = array();
  705. foreach(self::$_pidMap as $worker_pid_array)
  706. {
  707. foreach($worker_pid_array as $worker_pid)
  708. {
  709. $pid_array[$worker_pid] = $worker_pid;
  710. }
  711. }
  712. return $pid_array;
  713. }
  714. /**
  715. * 创建子进程
  716. * @return void
  717. */
  718. protected static function forkWorkers()
  719. {
  720. /** @var static $worker */
  721. foreach(self::$_workers as $worker)
  722. {
  723. // 启动过程中需要得到运行用户名的最大长度,在status时格式化展示
  724. if(self::$_status === self::STATUS_STARTING)
  725. {
  726. if(empty($worker->name))
  727. {
  728. $worker->name = $worker->getSocketName();
  729. }
  730. $worker_name_length = strlen($worker->name);
  731. if(self::$_maxWorkerNameLength < $worker_name_length)
  732. {
  733. self::$_maxWorkerNameLength = $worker_name_length;
  734. }
  735. }
  736. // 创建子进程
  737. while(count(self::$_pidMap[$worker->workerId]) < $worker->count)
  738. {
  739. static::forkOneWorker($worker);
  740. }
  741. }
  742. }
  743. /**
  744. * 创建一个子进程
  745. * @param Worker $worker
  746. * @throws Exception
  747. */
  748. protected static function forkOneWorker($worker)
  749. {
  750. $pid = pcntl_fork();
  751. // 获得可用的id
  752. $id = self::getId($worker->workerId, 0);
  753. // 主进程记录子进程pid
  754. if($pid > 0)
  755. {
  756. self::$_pidMap[$worker->workerId][$pid] = $pid;
  757. self::$_idMap[$worker->workerId][$id] = $pid;
  758. }
  759. // 子进程运行
  760. elseif(0 === $pid)
  761. {
  762. // 如果设置了端口复用,则在子进程执行监听
  763. if($worker->reusePort)
  764. {
  765. $worker->listen();
  766. }
  767. // 启动过程中尝试重定向标准输出
  768. if(self::$_status === self::STATUS_STARTING)
  769. {
  770. self::resetStd();
  771. }
  772. self::$_pidMap = array();
  773. self::$_workers = array($worker->workerId => $worker);
  774. Timer::delAll();
  775. self::setProcessTitle('WorkerMan: worker process ' . $worker->name . ' ' . $worker->getSocketName());
  776. self::setProcessUserAndRoot();
  777. $worker->id = $id;
  778. $worker->run();
  779. exit(250);
  780. }
  781. else
  782. {
  783. throw new Exception("forkOneWorker fail");
  784. }
  785. }
  786. /**
  787. * 获得可用的worker->id,以便传递给子进程
  788. * @param int $worker_id
  789. * @param int $pid
  790. */
  791. protected static function getId($worker_id, $pid)
  792. {
  793. $id = array_search($pid, self::$_idMap[$worker_id]);
  794. if($id === false)
  795. {
  796. echo "getId fail\n";
  797. }
  798. return $id;
  799. }
  800. /**
  801. * 尝试设置运行当前进程的用户、用户组、文件系统根目录
  802. *
  803. * @param $user_name
  804. */
  805. protected static function setProcessUserAndRoot()
  806. {
  807. // set chroot
  808. if($this->chroot)
  809. {
  810. if (posix_getuid() != 0)
  811. {
  812. self::log('Waring: You must have the root privileges to change root.', true);
  813. }
  814. else if(!chroot($this->chroot))
  815. {
  816. return self::log( "Notice: chroot({$this->chroot}) fail.", true);
  817. }
  818. }
  819. // get uid
  820. $user_info = posix_getpwnam($this->user);
  821. if(!$user_info)
  822. {
  823. return self::log( "Waring: User {$this->user} not exsits", true);
  824. }
  825. $uid = $user_info['uid'];
  826. // get gid
  827. if($this->group)
  828. {
  829. $group_info = posix_getgrnam($this->group);
  830. if(!$group_info)
  831. {
  832. return self::log( "Waring: Group {$this->group} not exsits", true);
  833. }
  834. $gid = $group_info['gid'];
  835. }
  836. else
  837. {
  838. $gid = $user_info['gid'];
  839. }
  840. // set uid and gid
  841. if($uid != posix_getuid() || $gid != posix_getgid())
  842. {
  843. if (posix_getuid() != 0)
  844. {
  845. self::log('Waring: You must have the root privileges to change uid and gid.', true);
  846. }
  847. elseif(!posix_setgid($gid) || !posix_initgroups($user_info['name'], $gid) || !posix_setuid($uid))
  848. {
  849. self::log( "Waring: change gid or uid fail.", true);
  850. }
  851. }
  852. }
  853. /**
  854. * 设置当前进程的名称,在ps aux命令中有用
  855. * 注意 需要php>=5.5或者安装了protitle扩展
  856. * @param string $title
  857. * @return void
  858. */
  859. protected static function setProcessTitle($title)
  860. {
  861. // >=php 5.5
  862. if (function_exists('cli_set_process_title'))
  863. {
  864. @cli_set_process_title($title);
  865. }
  866. // 需要扩展
  867. elseif(extension_loaded('proctitle') && function_exists('setproctitle'))
  868. {
  869. @setproctitle($title);
  870. }
  871. }
  872. /**
  873. * 监控所有子进程的退出事件及退出码
  874. * @return void
  875. */
  876. protected static function monitorWorkers()
  877. {
  878. self::$_status = self::STATUS_RUNNING;
  879. while(1)
  880. {
  881. // 如果有信号到来,尝试触发信号处理函数
  882. pcntl_signal_dispatch();
  883. // 挂起进程,直到有子进程退出或者被信号打断
  884. $status = 0;
  885. $pid = pcntl_wait($status, WUNTRACED);
  886. // 如果有信号到来,尝试触发信号处理函数
  887. pcntl_signal_dispatch();
  888. // 有子进程退出
  889. if($pid > 0)
  890. {
  891. // 查找是哪个进程组的,然后再启动新的进程补上
  892. foreach(self::$_pidMap as $worker_id => $worker_pid_array)
  893. {
  894. if(isset($worker_pid_array[$pid]))
  895. {
  896. $worker = self::$_workers[$worker_id];
  897. // 检查退出状态
  898. if($status !== 0)
  899. {
  900. self::log("worker[".$worker->name.":$pid] exit with status $status");
  901. }
  902. // 统计,运行status命令时使用
  903. if(!isset(self::$_globalStatistics['worker_exit_info'][$worker_id][$status]))
  904. {
  905. self::$_globalStatistics['worker_exit_info'][$worker_id][$status] = 0;
  906. }
  907. self::$_globalStatistics['worker_exit_info'][$worker_id][$status]++;
  908. // 清除子进程信息
  909. unset(self::$_pidMap[$worker_id][$pid]);
  910. // 标记$id为可用id
  911. $id = self::getId($worker_id, $pid);
  912. self::$_idMap[$worker_id][$id] = 0;
  913. break;
  914. }
  915. }
  916. // 如果不是关闭状态,则补充新的进程
  917. if(self::$_status !== self::STATUS_SHUTDOWN)
  918. {
  919. self::forkWorkers();
  920. // 如果该进程是因为运行reload命令退出,则继续执行reload流程
  921. if(isset(self::$_pidsToRestart[$pid]))
  922. {
  923. unset(self::$_pidsToRestart[$pid]);
  924. self::reload();
  925. }
  926. }
  927. else
  928. {
  929. // 如果是关闭状态,并且所有进程退出完毕,则主进程退出
  930. if(!self::getAllWorkerPids())
  931. {
  932. self::exitAndClearAll();
  933. }
  934. }
  935. }
  936. else
  937. {
  938. // 如果是关闭状态,并且所有进程退出完毕,则主进程退出
  939. if(self::$_status === self::STATUS_SHUTDOWN && !self::getAllWorkerPids())
  940. {
  941. self::exitAndClearAll();
  942. }
  943. }
  944. }
  945. }
  946. /**
  947. * 退出当前进程
  948. * @return void
  949. */
  950. protected static function exitAndClearAll()
  951. {
  952. @unlink(self::$pidFile);
  953. self::log("Workerman[".basename(self::$_startFile)."] has been stopped");
  954. exit(0);
  955. }
  956. /**
  957. * 执行平滑重启流程
  958. * @return void
  959. */
  960. protected static function reload()
  961. {
  962. // 主进程部分
  963. if(self::$_masterPid === posix_getpid())
  964. {
  965. // 设置为平滑重启状态
  966. if(self::$_status !== self::STATUS_RELOADING && self::$_status !== self::STATUS_SHUTDOWN)
  967. {
  968. self::log("Workerman[".basename(self::$_startFile)."] reloading");
  969. self::$_status = self::STATUS_RELOADING;
  970. }
  971. // 如果有worker设置了reloadable=false,则过滤掉
  972. $reloadable_pid_array = array();
  973. foreach(self::$_pidMap as $worker_id =>$worker_pid_array)
  974. {
  975. $worker = self::$_workers[$worker_id];
  976. if($worker->reloadable)
  977. {
  978. foreach($worker_pid_array as $pid)
  979. {
  980. $reloadable_pid_array[$pid] = $pid;
  981. }
  982. }
  983. else
  984. {
  985. foreach($worker_pid_array as $pid)
  986. {
  987. // 给reloadable=false的进程也发送一个reload信号,触发onWorkerReload
  988. posix_kill($pid, SIGUSR1);
  989. }
  990. }
  991. }
  992. // 得到所有可以重启的进程
  993. self::$_pidsToRestart = array_intersect(self::$_pidsToRestart , $reloadable_pid_array);
  994. // 平滑重启完毕
  995. if(empty(self::$_pidsToRestart))
  996. {
  997. if(self::$_status !== self::STATUS_SHUTDOWN)
  998. {
  999. self::$_status = self::STATUS_RUNNING;
  1000. }
  1001. return;
  1002. }
  1003. // 继续执行平滑重启流程
  1004. $one_worker_pid = current(self::$_pidsToRestart );
  1005. // 给子进程发送平滑重启信号
  1006. posix_kill($one_worker_pid, SIGUSR1);
  1007. // 定时器,如果子进程在KILL_WORKER_TIMER_TIME秒后没有退出,则强行杀死
  1008. Timer::add(self::KILL_WORKER_TIMER_TIME, 'posix_kill', array($one_worker_pid, SIGKILL), false);
  1009. }
  1010. // 子进程部分
  1011. else
  1012. {
  1013. // 如果当前worker的reloadable属性为真,则执行退出
  1014. $worker = current(self::$_workers);
  1015. // 如果有设置Reload回调,则执行
  1016. if($worker->onWorkerReload)
  1017. {
  1018. call_user_func($worker->onWorkerReload, $worker);
  1019. }
  1020. if($worker->reloadable)
  1021. {
  1022. self::stopAll();
  1023. }
  1024. }
  1025. }
  1026. /**
  1027. * 执行关闭流程
  1028. * @return void
  1029. */
  1030. public static function stopAll()
  1031. {
  1032. self::$_status = self::STATUS_SHUTDOWN;
  1033. // 主进程部分
  1034. if(self::$_masterPid === posix_getpid())
  1035. {
  1036. self::log("Workerman[".basename(self::$_startFile)."] Stopping ...");
  1037. $worker_pid_array = self::getAllWorkerPids();
  1038. // 向所有子进程发送SIGINT信号,表明关闭服务
  1039. foreach($worker_pid_array as $worker_pid)
  1040. {
  1041. posix_kill($worker_pid, SIGINT);
  1042. Timer::add(self::KILL_WORKER_TIMER_TIME, 'posix_kill', array($worker_pid, SIGKILL),false);
  1043. }
  1044. }
  1045. // 子进程部分
  1046. else
  1047. {
  1048. // 执行stop逻辑
  1049. /** @var static $worker */
  1050. foreach(self::$_workers as $worker)
  1051. {
  1052. $worker->stop();
  1053. }
  1054. exit(0);
  1055. }
  1056. }
  1057. /**
  1058. * 将当前进程的统计信息写入到统计文件
  1059. * @return void
  1060. */
  1061. protected static function writeStatisticsToStatusFile()
  1062. {
  1063. // 主进程部分
  1064. if(self::$_masterPid === posix_getpid())
  1065. {
  1066. $loadavg = sys_getloadavg();
  1067. file_put_contents(self::$_statisticsFile, "---------------------------------------GLOBAL STATUS--------------------------------------------\n");
  1068. file_put_contents(self::$_statisticsFile, 'Workerman version:' . Worker::VERSION . " PHP version:".PHP_VERSION."\n", FILE_APPEND);
  1069. file_put_contents(self::$_statisticsFile, 'start time:'. date('Y-m-d H:i:s', self::$_globalStatistics['start_timestamp']).' run ' . floor((time()-self::$_globalStatistics['start_timestamp'])/(24*60*60)). ' days ' . floor(((time()-self::$_globalStatistics['start_timestamp'])%(24*60*60))/(60*60)) . " hours \n", FILE_APPEND);
  1070. file_put_contents(self::$_statisticsFile, 'load average: ' . implode(", ", $loadavg) . "\n", FILE_APPEND);
  1071. file_put_contents(self::$_statisticsFile, count(self::$_pidMap) . ' workers ' . count(self::getAllWorkerPids())." processes\n", FILE_APPEND);
  1072. file_put_contents(self::$_statisticsFile, str_pad('worker_name', self::$_maxWorkerNameLength) . " exit_status exit_count\n", FILE_APPEND);
  1073. foreach(self::$_pidMap as $worker_id =>$worker_pid_array)
  1074. {
  1075. $worker = self::$_workers[$worker_id];
  1076. if(isset(self::$_globalStatistics['worker_exit_info'][$worker_id]))
  1077. {
  1078. foreach(self::$_globalStatistics['worker_exit_info'][$worker_id] as $worker_exit_status=>$worker_exit_count)
  1079. {
  1080. file_put_contents(self::$_statisticsFile, str_pad($worker->name, self::$_maxWorkerNameLength) . " " . str_pad($worker_exit_status, 16). " $worker_exit_count\n", FILE_APPEND);
  1081. }
  1082. }
  1083. else
  1084. {
  1085. file_put_contents(self::$_statisticsFile, str_pad($worker->name, self::$_maxWorkerNameLength) . " " . str_pad(0, 16). " 0\n", FILE_APPEND);
  1086. }
  1087. }
  1088. file_put_contents(self::$_statisticsFile, "---------------------------------------PROCESS STATUS-------------------------------------------\n", FILE_APPEND);
  1089. file_put_contents(self::$_statisticsFile, "pid\tmemory ".str_pad('listening', self::$_maxSocketNameLength)." ".str_pad('worker_name', self::$_maxWorkerNameLength)." connections ".str_pad('total_request', 13)." ".str_pad('send_fail', 9)." ".str_pad('throw_exception', 15)."\n", FILE_APPEND);
  1090. chmod(self::$_statisticsFile, 0722);
  1091. foreach(self::getAllWorkerPids() as $worker_pid)
  1092. {
  1093. posix_kill($worker_pid, SIGUSR2);
  1094. }
  1095. return;
  1096. }
  1097. // 子进程部分
  1098. $worker = current(self::$_workers);
  1099. $wrker_status_str = posix_getpid()."\t".str_pad(round(memory_get_usage(true)/(1024*1024),2)."M", 7)." " .str_pad($worker->getSocketName(), self::$_maxSocketNameLength) ." ".str_pad(($worker->name === $worker->getSocketName() ? 'none' : $worker->name), self::$_maxWorkerNameLength)." ";
  1100. $wrker_status_str .= str_pad(ConnectionInterface::$statistics['connection_count'], 11)." ".str_pad(ConnectionInterface::$statistics['total_request'], 14)." ".str_pad(ConnectionInterface::$statistics['send_fail'],9)." ".str_pad(ConnectionInterface::$statistics['throw_exception'],15)."\n";
  1101. file_put_contents(self::$_statisticsFile, $wrker_status_str, FILE_APPEND);
  1102. }
  1103. /**
  1104. * 检查错误
  1105. * @return void
  1106. */
  1107. public static function checkErrors()
  1108. {
  1109. if(self::STATUS_SHUTDOWN != self::$_status)
  1110. {
  1111. $error_msg = "WORKER EXIT UNEXPECTED ";
  1112. $errors = error_get_last();
  1113. if($errors && ($errors['type'] === E_ERROR ||
  1114. $errors['type'] === E_PARSE ||
  1115. $errors['type'] === E_CORE_ERROR ||
  1116. $errors['type'] === E_COMPILE_ERROR ||
  1117. $errors['type'] === E_RECOVERABLE_ERROR ))
  1118. {
  1119. $error_msg .= self::getErrorType($errors['type']) . " {$errors['message']} in {$errors['file']} on line {$errors['line']}";
  1120. }
  1121. self::log($error_msg);
  1122. }
  1123. }
  1124. /**
  1125. * 获取错误类型对应的意义
  1126. * @param integer $type
  1127. * @return string
  1128. */
  1129. protected static function getErrorType($type)
  1130. {
  1131. switch($type)
  1132. {
  1133. case E_ERROR: // 1 //
  1134. return 'E_ERROR';
  1135. case E_WARNING: // 2 //
  1136. return 'E_WARNING';
  1137. case E_PARSE: // 4 //
  1138. return 'E_PARSE';
  1139. case E_NOTICE: // 8 //
  1140. return 'E_NOTICE';
  1141. case E_CORE_ERROR: // 16 //
  1142. return 'E_CORE_ERROR';
  1143. case E_CORE_WARNING: // 32 //
  1144. return 'E_CORE_WARNING';
  1145. case E_COMPILE_ERROR: // 64 //
  1146. return 'E_COMPILE_ERROR';
  1147. case E_COMPILE_WARNING: // 128 //
  1148. return 'E_COMPILE_WARNING';
  1149. case E_USER_ERROR: // 256 //
  1150. return 'E_USER_ERROR';
  1151. case E_USER_WARNING: // 512 //
  1152. return 'E_USER_WARNING';
  1153. case E_USER_NOTICE: // 1024 //
  1154. return 'E_USER_NOTICE';
  1155. case E_STRICT: // 2048 //
  1156. return 'E_STRICT';
  1157. case E_RECOVERABLE_ERROR: // 4096 //
  1158. return 'E_RECOVERABLE_ERROR';
  1159. case E_DEPRECATED: // 8192 //
  1160. return 'E_DEPRECATED';
  1161. case E_USER_DEPRECATED: // 16384 //
  1162. return 'E_USER_DEPRECATED';
  1163. }
  1164. return "";
  1165. }
  1166. /**
  1167. * 记录日志
  1168. * @param string $msg
  1169. * @return void
  1170. */
  1171. protected static function log($msg)
  1172. {
  1173. $msg = $msg."\n";
  1174. if(!self::$daemonize)
  1175. {
  1176. echo $msg;
  1177. }
  1178. file_put_contents(self::$logFile, date('Y-m-d H:i:s') . " " . $msg, FILE_APPEND | LOCK_EX);
  1179. }
  1180. /**
  1181. * worker构造函数
  1182. *
  1183. * @param string $socket_name
  1184. * @param array $context_option
  1185. */
  1186. public function __construct($socket_name = '', $context_option = array())
  1187. {
  1188. // 保存worker实例
  1189. $this->workerId = spl_object_hash($this);
  1190. self::$_workers[$this->workerId] = $this;
  1191. self::$_pidMap[$this->workerId] = array();
  1192. // 获得实例化文件路径,用于自动加载设置根目录
  1193. $backrace = debug_backtrace();
  1194. $this->_appInitPath = dirname($backrace[0]['file']);
  1195. // 设置socket上下文
  1196. if($socket_name)
  1197. {
  1198. $this->_socketName = $socket_name;
  1199. if(!isset($context_option['socket']['backlog']))
  1200. {
  1201. $context_option['socket']['backlog'] = self::DEFAUL_BACKLOG;
  1202. }
  1203. $this->_context = stream_context_create($context_option);
  1204. }
  1205. // 设置一个空的onMessage,当onMessage未设置时用来消费socket数据
  1206. $this->onMessage = function(){};
  1207. }
  1208. /**
  1209. * 监听端口
  1210. * @throws Exception
  1211. */
  1212. public function listen()
  1213. {
  1214. if(!$this->_socketName || $this->_mainSocket)
  1215. {
  1216. return;
  1217. }
  1218. // 设置自动加载根目录
  1219. Autoloader::setRootPath($this->_appInitPath);
  1220. // 获得应用层通讯协议以及监听的地址
  1221. list($scheme, $address) = explode(':', $this->_socketName, 2);
  1222. // 如果有指定应用层协议,则检查对应的协议类是否存在
  1223. if($scheme != 'tcp' && $scheme != 'udp')
  1224. {
  1225. $scheme = ucfirst($scheme);
  1226. $this->_protocol = '\\Protocols\\'.$scheme;
  1227. if(!class_exists($this->_protocol))
  1228. {
  1229. $this->_protocol = "\\Workerman\\Protocols\\$scheme";
  1230. if(!class_exists($this->_protocol))
  1231. {
  1232. throw new Exception("class \\Protocols\\$scheme not exist");
  1233. }
  1234. }
  1235. }
  1236. elseif($scheme === 'udp')
  1237. {
  1238. $this->transport = 'udp';
  1239. }
  1240. // flag
  1241. $flags = $this->transport === 'udp' ? STREAM_SERVER_BIND : STREAM_SERVER_BIND | STREAM_SERVER_LISTEN;
  1242. $errno = 0;
  1243. $errmsg = '';
  1244. // 如果设置了端口复用,则设置SO_REUSEPORT选项为1
  1245. if($this->reusePort)
  1246. {
  1247. stream_context_set_option($this->_context, 'socket', 'so_reuseport', 1);
  1248. }
  1249. $this->_mainSocket = stream_socket_server($this->transport.":".$address, $errno, $errmsg, $flags, $this->_context);
  1250. if(!$this->_mainSocket)
  1251. {
  1252. throw new Exception($errmsg);
  1253. }
  1254. // 尝试打开tcp的keepalive,关闭TCP Nagle算法
  1255. if(function_exists('socket_import_stream'))
  1256. {
  1257. $socket = socket_import_stream($this->_mainSocket );
  1258. @socket_set_option($socket, SOL_SOCKET, SO_KEEPALIVE, 1);
  1259. @socket_set_option($socket, SOL_SOCKET, TCP_NODELAY, 1);
  1260. }
  1261. // 设置非阻塞
  1262. stream_set_blocking($this->_mainSocket, 0);
  1263. // 放到全局事件轮询中监听_mainSocket可读事件(客户端连接事件)
  1264. if(self::$globalEvent)
  1265. {
  1266. if($this->transport !== 'udp')
  1267. {
  1268. self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ, array($this, 'acceptConnection'));
  1269. }
  1270. else
  1271. {
  1272. self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ, array($this, 'acceptUdpConnection'));
  1273. }
  1274. }
  1275. }
  1276. /**
  1277. * 获得 socket name
  1278. * @return string
  1279. */
  1280. public function getSocketName()
  1281. {
  1282. return $this->_socketName ? $this->_socketName : 'none';
  1283. }
  1284. /**
  1285. * 运行worker实例
  1286. */
  1287. public function run()
  1288. {
  1289. //更新 Worker 状态
  1290. self::$_status = self::STATUS_RUNNING;
  1291. // 注册进程退出回调,用来检查是否有错误
  1292. register_shutdown_function(array("\\Workerman\\Worker", 'checkErrors'));
  1293. // 设置自动加载根目录
  1294. Autoloader::setRootPath($this->_appInitPath);
  1295. // 如果没有全局事件轮询,则创建一个
  1296. if(!self::$globalEvent)
  1297. {
  1298. if(extension_loaded('libevent'))
  1299. {
  1300. self::$globalEvent = new Libevent();
  1301. }
  1302. else
  1303. {
  1304. self::$globalEvent = new Select();
  1305. }
  1306. // 监听_mainSocket上的可读事件(客户端连接事件)
  1307. if($this->_socketName)
  1308. {
  1309. if($this->transport !== 'udp')
  1310. {
  1311. self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ, array($this, 'acceptConnection'));
  1312. }
  1313. else
  1314. {
  1315. self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ, array($this, 'acceptUdpConnection'));
  1316. }
  1317. }
  1318. }
  1319. // 重新安装事件处理函数,使用全局事件轮询监听信号事件
  1320. self::reinstallSignal();
  1321. // 用全局事件轮询初始化定时器
  1322. Timer::init(self::$globalEvent);
  1323. // 如果有设置进程启动回调,则执行
  1324. if($this->onWorkerStart)
  1325. {
  1326. call_user_func($this->onWorkerStart, $this);
  1327. }
  1328. // 子进程主循环
  1329. self::$globalEvent->loop();
  1330. }
  1331. /**
  1332. * 停止当前worker实例
  1333. * @return void
  1334. */
  1335. public function stop()
  1336. {
  1337. // 如果有设置进程终止回调,则执行
  1338. if($this->onWorkerStop)
  1339. {
  1340. call_user_func($this->onWorkerStop, $this);
  1341. }
  1342. // 删除相关监听事件,关闭_mainSocket
  1343. self::$globalEvent->del($this->_mainSocket, EventInterface::EV_READ);
  1344. @fclose($this->_mainSocket);
  1345. }
  1346. /**
  1347. * 接收一个客户端连接
  1348. * @param resource $socket
  1349. * @return void
  1350. */
  1351. public function acceptConnection($socket)
  1352. {
  1353. // 获得客户端连接
  1354. $new_socket = @stream_socket_accept($socket, 0, $remote_address);
  1355. // 惊群现象,忽略
  1356. if(false === $new_socket)
  1357. {
  1358. return;
  1359. }
  1360. // 初始化连接对象
  1361. $connection = new TcpConnection($new_socket, $remote_address);
  1362. $this->connections[$connection->id] = $connection;
  1363. $connection->worker = $this;
  1364. $connection->protocol = $this->_protocol;
  1365. $connection->onMessage = $this->onMessage;
  1366. $connection->onClose = $this->onClose;
  1367. $connection->onError = $this->onError;
  1368. $connection->onBufferDrain = $this->onBufferDrain;
  1369. $connection->onBufferFull = $this->onBufferFull;
  1370. // 如果有设置连接回调,则执行
  1371. if($this->onConnect)
  1372. {
  1373. try
  1374. {
  1375. call_user_func($this->onConnect, $connection);
  1376. }
  1377. catch(Exception $e)
  1378. {
  1379. ConnectionInterface::$statistics['throw_exception']++;
  1380. self::log($e);
  1381. }
  1382. }
  1383. }
  1384. /**
  1385. * 处理udp连接(udp其实是无连接的,这里为保证和tcp连接接口一致)
  1386. *
  1387. * @param resource $socket
  1388. * @return bool
  1389. */
  1390. public function acceptUdpConnection($socket)
  1391. {
  1392. $recv_buffer = stream_socket_recvfrom($socket , self::MAX_UDP_PACKEG_SIZE, 0, $remote_address);
  1393. if(false === $recv_buffer || empty($remote_address))
  1394. {
  1395. return false;
  1396. }
  1397. // 模拟一个连接对象
  1398. $connection = new UdpConnection($socket, $remote_address);
  1399. if($this->onMessage)
  1400. {
  1401. if($this->_protocol)
  1402. {
  1403. /** @var \Workerman\Protocols\ProtocolInterface $parser */
  1404. $parser = $this->_protocol;
  1405. $recv_buffer = $parser::decode($recv_buffer, $connection);
  1406. }
  1407. ConnectionInterface::$statistics['total_request']++;
  1408. try
  1409. {
  1410. call_user_func($this->onMessage, $connection, $recv_buffer);
  1411. }
  1412. catch(Exception $e)
  1413. {
  1414. ConnectionInterface::$statistics['throw_exception']++;
  1415. }
  1416. }
  1417. }
  1418. }