Worker.php 49 KB

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