Worker.php 49 KB

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