Worker.php 81 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623
  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. declare(strict_types=1);
  15. namespace Workerman;
  16. use AllowDynamicProperties;
  17. use Exception;
  18. use Revolt\EventLoop;
  19. use RuntimeException;
  20. use stdClass;
  21. use Throwable;
  22. use Workerman\Connection\ConnectionInterface;
  23. use Workerman\Connection\TcpConnection;
  24. use Workerman\Connection\UdpConnection;
  25. use Workerman\Events\Event;
  26. use Workerman\Events\EventInterface;
  27. use Workerman\Events\Revolt;
  28. use Workerman\Events\Select;
  29. use Workerman\Protocols\ProtocolInterface;
  30. use function method_exists;
  31. use function restore_error_handler;
  32. use function set_error_handler;
  33. use function stream_socket_accept;
  34. use function stream_socket_recvfrom;
  35. use function substr;
  36. /**
  37. * Worker class
  38. * A container for listening ports
  39. */
  40. #[AllowDynamicProperties]
  41. class Worker
  42. {
  43. /**
  44. * Version.
  45. *
  46. * @var string
  47. */
  48. public const VERSION = '5.0.0-beta.3';
  49. /**
  50. * Status starting.
  51. *
  52. * @var int
  53. */
  54. public const STATUS_STARTING = 1;
  55. /**
  56. * Status running.
  57. *
  58. * @var int
  59. */
  60. public const STATUS_RUNNING = 2;
  61. /**
  62. * Status shutdown.
  63. *
  64. * @var int
  65. */
  66. public const STATUS_SHUTDOWN = 4;
  67. /**
  68. * Status reloading.
  69. *
  70. * @var int
  71. */
  72. public const STATUS_RELOADING = 8;
  73. /**
  74. * Default backlog. Backlog is the maximum length of the queue of pending connections.
  75. *
  76. * @var int
  77. */
  78. public const DEFAULT_BACKLOG = 102400;
  79. /**
  80. * The safe distance for columns adjacent
  81. *
  82. * @var int
  83. */
  84. public const UI_SAFE_LENGTH = 4;
  85. /**
  86. * Worker id.
  87. *
  88. * @var int
  89. */
  90. public int $id = 0;
  91. /**
  92. * Name of the worker processes.
  93. *
  94. * @var string
  95. */
  96. public string $name = 'none';
  97. /**
  98. * Number of worker processes.
  99. *
  100. * @var int
  101. */
  102. public int $count = 1;
  103. /**
  104. * Unix user of processes, needs appropriate privileges (usually root).
  105. *
  106. * @var string
  107. */
  108. public string $user = '';
  109. /**
  110. * Unix group of processes, needs appropriate privileges (usually root).
  111. *
  112. * @var string
  113. */
  114. public string $group = '';
  115. /**
  116. * reloadable.
  117. *
  118. * @var bool
  119. */
  120. public bool $reloadable = true;
  121. /**
  122. * reuse port.
  123. *
  124. * @var bool
  125. */
  126. public bool $reusePort = false;
  127. /**
  128. * Emitted when worker processes is starting.
  129. *
  130. * @var ?callable
  131. */
  132. public $onWorkerStart = null;
  133. /**
  134. * Emitted when a socket connection is successfully established.
  135. *
  136. * @var ?callable
  137. */
  138. public $onConnect = null;
  139. /**
  140. * Emitted when websocket handshake did complete (Only called when protocol is ws).
  141. *
  142. * @var ?callable
  143. */
  144. public $onWebSocketConnect = null;
  145. /**
  146. * Emitted when data is received.
  147. *
  148. * @var callable
  149. */
  150. public $onMessage = null;
  151. /**
  152. * Emitted when the other end of the socket sends a FIN packet.
  153. *
  154. * @var ?callable
  155. */
  156. public $onClose = null;
  157. /**
  158. * Emitted when an error occurs with connection.
  159. *
  160. * @var ?callable
  161. */
  162. public $onError = null;
  163. /**
  164. * Emitted when the send buffer becomes full.
  165. *
  166. * @var ?callable
  167. */
  168. public $onBufferFull = null;
  169. /**
  170. * Emitted when the send buffer becomes empty.
  171. *
  172. * @var ?callable
  173. */
  174. public $onBufferDrain = null;
  175. /**
  176. * Emitted when worker processes has stopped.
  177. *
  178. * @var ?callable
  179. */
  180. public $onWorkerStop = null;
  181. /**
  182. * Emitted when worker processes receives reload signal.
  183. *
  184. * @var ?callable
  185. */
  186. public $onWorkerReload = null;
  187. /**
  188. * Transport layer protocol.
  189. *
  190. * @var string
  191. */
  192. public string $transport = 'tcp';
  193. /**
  194. * Store all connections of clients.
  195. *
  196. * @var TcpConnection[]
  197. */
  198. public array $connections = [];
  199. /**
  200. * Application layer protocol.
  201. *
  202. * @var ?string
  203. */
  204. public ?string $protocol = null;
  205. /**
  206. * Pause accept new connections or not.
  207. *
  208. * @var bool
  209. */
  210. protected bool $pauseAccept = true;
  211. /**
  212. * Is worker stopping ?
  213. * @var bool
  214. */
  215. public bool $stopping = false;
  216. /**
  217. * Daemonize.
  218. *
  219. * @var bool
  220. */
  221. public static bool $daemonize = false;
  222. /**
  223. * Stdout file.
  224. *
  225. * @var string
  226. */
  227. public static string $stdoutFile = '/dev/null';
  228. /**
  229. * The file to store master process PID.
  230. *
  231. * @var string
  232. */
  233. public static string $pidFile = '';
  234. /**
  235. * The file used to store the master process status file.
  236. *
  237. * @var string
  238. */
  239. public static string $statusFile = '';
  240. /**
  241. * Log file.
  242. *
  243. * @var mixed
  244. */
  245. public static mixed $logFile = '';
  246. /**
  247. * Global event loop.
  248. *
  249. * @var ?EventInterface
  250. */
  251. public static ?EventInterface $globalEvent = null;
  252. /**
  253. * Emitted when the master process get reload signal.
  254. *
  255. * @var ?callable
  256. */
  257. public static $onMasterReload = null;
  258. /**
  259. * Emitted when the master process terminated.
  260. *
  261. * @var ?callable
  262. */
  263. public static $onMasterStop = null;
  264. /**
  265. * Emitted when worker processes exited.
  266. *
  267. * @var ?callable
  268. */
  269. public static $onWorkerExit = null;
  270. /**
  271. * EventLoopClass
  272. *
  273. * @var class-string
  274. */
  275. public static string $eventLoopClass = '';
  276. /**
  277. * Process title
  278. *
  279. * @var string
  280. */
  281. public static string $processTitle = 'WorkerMan';
  282. /**
  283. * After sending the stop command to the child process stopTimeout seconds,
  284. * if the process is still living then forced to kill.
  285. *
  286. * @var int
  287. */
  288. public static int $stopTimeout = 2;
  289. /**
  290. * Command
  291. * @var string
  292. */
  293. public static string $command = '';
  294. /**
  295. * The PID of master process.
  296. *
  297. * @var int
  298. */
  299. protected static int $masterPid = 0;
  300. /**
  301. * Listening socket.
  302. *
  303. * @var resource
  304. */
  305. protected $mainSocket = null;
  306. /**
  307. * Socket name. The format is like this http://0.0.0.0:80 .
  308. *
  309. * @var string
  310. */
  311. protected string $socketName = '';
  312. /**
  313. * parse from socketName avoid parse again in master or worker
  314. * LocalSocket The format is like tcp://0.0.0.0:8080
  315. * @var ?string
  316. */
  317. protected ?string $localSocket = null;
  318. /**
  319. * Context of socket.
  320. *
  321. * @var resource
  322. */
  323. protected $socketContext = null;
  324. /**
  325. * @var stdClass
  326. */
  327. protected stdClass $context;
  328. /**
  329. * All worker instances.
  330. *
  331. * @var Worker[]
  332. */
  333. protected static array $workers = [];
  334. /**
  335. * All worker processes pid.
  336. * The format is like this [worker_id=>[pid=>pid, pid=>pid, ..], ..]
  337. *
  338. * @var array
  339. */
  340. protected static array $pidMap = [];
  341. /**
  342. * All worker processes waiting for restart.
  343. * The format is like this [pid=>pid, pid=>pid].
  344. *
  345. * @var array
  346. */
  347. protected static array $pidsToRestart = [];
  348. /**
  349. * Mapping from PID to worker process ID.
  350. * The format is like this [worker_id=>[0=>$pid, 1=>$pid, ..], ..].
  351. *
  352. * @var array
  353. */
  354. protected static array $idMap = [];
  355. /**
  356. * Current status.
  357. *
  358. * @var int
  359. */
  360. protected static int $status = self::STATUS_STARTING;
  361. /**
  362. * Maximum length of the worker names.
  363. *
  364. * @var int
  365. */
  366. protected static int $maxWorkerNameLength = 12;
  367. /**
  368. * Maximum length of the socket names.
  369. *
  370. * @var int
  371. */
  372. protected static int $maxSocketNameLength = 12;
  373. /**
  374. * Maximum length of the process usernames.
  375. *
  376. * @var int
  377. */
  378. protected static int $maxUserNameLength = 12;
  379. /**
  380. * Maximum length of the Proto names.
  381. *
  382. * @var int
  383. */
  384. protected static int $maxProtoNameLength = 4;
  385. /**
  386. * Maximum length of the Processes names.
  387. *
  388. * @var int
  389. */
  390. protected static int $maxProcessesNameLength = 9;
  391. /**
  392. * Maximum length of the state names.
  393. *
  394. * @var int
  395. */
  396. protected static int $maxStateNameLength = 1;
  397. /**
  398. * The file to store status info of current worker process.
  399. *
  400. * @var string
  401. */
  402. protected static string $statisticsFile = '';
  403. /**
  404. * Start file.
  405. *
  406. * @var string
  407. */
  408. protected static string $startFile = '';
  409. /**
  410. * Processes for windows.
  411. *
  412. * @var array
  413. */
  414. protected static array $processForWindows = [];
  415. /**
  416. * Status info of current worker process.
  417. *
  418. * @var array
  419. */
  420. protected static array $globalStatistics = [
  421. 'start_timestamp' => 0,
  422. 'worker_exit_info' => []
  423. ];
  424. /**
  425. * Available event loops.
  426. *
  427. * @var array<string, string>
  428. */
  429. protected static array $availableEventLoops = [
  430. 'event' => Event::class,
  431. ];
  432. /**
  433. * PHP built-in protocols.
  434. *
  435. * @var array<string,string>
  436. */
  437. public const BUILD_IN_TRANSPORTS = [
  438. 'tcp' => 'tcp',
  439. 'udp' => 'udp',
  440. 'unix' => 'unix',
  441. 'ssl' => 'tcp'
  442. ];
  443. /**
  444. * PHP built-in error types.
  445. *
  446. * @var array<int,string>
  447. */
  448. public const ERROR_TYPE = [
  449. E_ERROR => 'E_ERROR', // 1
  450. E_WARNING => 'E_WARNING', // 2
  451. E_PARSE => 'E_PARSE', // 4
  452. E_NOTICE => 'E_NOTICE', // 8
  453. E_CORE_ERROR => 'E_CORE_ERROR', // 16
  454. E_CORE_WARNING => 'E_CORE_WARNING', // 32
  455. E_COMPILE_ERROR => 'E_COMPILE_ERROR', // 64
  456. E_COMPILE_WARNING => 'E_COMPILE_WARNING', // 128
  457. E_USER_ERROR => 'E_USER_ERROR', // 256
  458. E_USER_WARNING => 'E_USER_WARNING', // 512
  459. E_USER_NOTICE => 'E_USER_NOTICE', // 1024
  460. E_STRICT => 'E_STRICT', // 2048
  461. E_RECOVERABLE_ERROR => 'E_RECOVERABLE_ERROR', // 4096
  462. E_DEPRECATED => 'E_DEPRECATED', // 8192
  463. E_USER_DEPRECATED => 'E_USER_DEPRECATED' // 16384
  464. ];
  465. /**
  466. * Graceful stop or not.
  467. *
  468. * @var bool
  469. */
  470. protected static bool $gracefulStop = false;
  471. /**
  472. * Standard output stream
  473. * @var resource
  474. */
  475. protected static $outputStream = null;
  476. /**
  477. * If $outputStream support decorated
  478. * @var bool
  479. */
  480. protected static bool $outputDecorated = false;
  481. /**
  482. * Worker object's hash id(unique identifier).
  483. *
  484. * @var ?string
  485. */
  486. protected ?string $workerId = null;
  487. /**
  488. * Run all worker instances.
  489. *
  490. * @return void
  491. * @throws Throwable
  492. */
  493. public static function runAll(): void
  494. {
  495. static::checkSapiEnv();
  496. static::init();
  497. static::parseCommand();
  498. static::lock();
  499. static::daemonize();
  500. static::initWorkers();
  501. static::installSignal();
  502. static::saveMasterPid();
  503. static::lock(LOCK_UN);
  504. static::displayUI();
  505. static::forkWorkers();
  506. static::resetStd();
  507. static::monitorWorkers();
  508. }
  509. /**
  510. * Check sapi.
  511. *
  512. * @return void
  513. */
  514. protected static function checkSapiEnv(): void
  515. {
  516. // Only for cli.
  517. if (PHP_SAPI !== 'cli') {
  518. exit("Only run in command line mode \n");
  519. }
  520. }
  521. /**
  522. * Init.
  523. *
  524. * @return void
  525. */
  526. protected static function init(): void
  527. {
  528. set_error_handler(function ($code, $msg, $file, $line) {
  529. static::safeEcho("$msg in file $file on line $line\n");
  530. });
  531. // Start file.
  532. $backtrace = debug_backtrace(DEBUG_BACKTRACE_IGNORE_ARGS);
  533. static::$startFile = end($backtrace)['file'];
  534. $uniquePrefix = str_replace('/', '_', static::$startFile);
  535. // Pid file.
  536. if (empty(static::$pidFile)) {
  537. static::$pidFile = __DIR__ . "/../$uniquePrefix.pid";
  538. }
  539. // Log file.
  540. if (empty(static::$logFile)) {
  541. static::$logFile = __DIR__ . '/../../workerman.log';
  542. }
  543. if (!is_file(static::$logFile)) {
  544. // if /runtime/logs default folder not exists
  545. if (!is_dir(dirname(static::$logFile))) {
  546. @mkdir(dirname(static::$logFile), 0777, true);
  547. }
  548. touch(static::$logFile);
  549. chmod(static::$logFile, 0622);
  550. }
  551. // State.
  552. static::$status = static::STATUS_STARTING;
  553. // For statistics.
  554. static::$globalStatistics['start_timestamp'] = time();
  555. // Process title.
  556. static::setProcessTitle(static::$processTitle . ': master process start_file=' . static::$startFile);
  557. // Init data for worker id.
  558. static::initId();
  559. // Timer init.
  560. Timer::init();
  561. }
  562. /**
  563. * Lock.
  564. *
  565. * @param int $flag
  566. * @return void
  567. */
  568. protected static function lock(int $flag = LOCK_EX): void
  569. {
  570. global $argv;
  571. static $fd;
  572. if (DIRECTORY_SEPARATOR !== '/' || empty($argv)) {
  573. return;
  574. }
  575. $lockFile = static::$pidFile . '.lock';
  576. $fd = $fd ?: fopen($lockFile, 'a+');
  577. if ($fd) {
  578. flock($fd, $flag);
  579. if ($flag === LOCK_UN) {
  580. fclose($fd);
  581. $fd = null;
  582. clearstatcache();
  583. if (is_file($lockFile)) {
  584. unlink($lockFile);
  585. }
  586. }
  587. }
  588. }
  589. /**
  590. * Init All worker instances.
  591. *
  592. * @return void
  593. * @throws Exception
  594. */
  595. protected static function initWorkers(): void
  596. {
  597. if (DIRECTORY_SEPARATOR !== '/') {
  598. return;
  599. }
  600. static::$statisticsFile = static::$statusFile ?: __DIR__ . '/../workerman-' . posix_getpid() . '.status';
  601. foreach (static::$workers as $worker) {
  602. // Worker name.
  603. if (empty($worker->name)) {
  604. $worker->name = 'none';
  605. }
  606. // Get unix user of the worker process.
  607. if (empty($worker->user)) {
  608. $worker->user = static::getCurrentUser();
  609. } else {
  610. if (posix_getuid() !== 0 && $worker->user !== static::getCurrentUser()) {
  611. static::log('Warning: You must have the root privileges to change uid and gid.');
  612. }
  613. }
  614. // Socket name.
  615. $worker->context->statusSocket = $worker->getSocketName();
  616. // Status name.
  617. $worker->context->statusState = '<g> [OK] </g>';
  618. // Get column mapping for UI
  619. foreach (static::getUiColumns() as $columnName => $prop) {
  620. !isset($worker->$prop) && !isset($worker->context->$prop) && $worker->context->$prop = 'NNNN';
  621. $propLength = strlen((string)($worker->$prop ?? $worker->context->$prop));
  622. $key = 'max' . ucfirst(strtolower($columnName)) . 'NameLength';
  623. static::$$key = max(static::$$key, $propLength);
  624. }
  625. // Listen.
  626. if (!$worker->reusePort) {
  627. $worker->listen();
  628. }
  629. }
  630. }
  631. /**
  632. * Get all worker instances.
  633. *
  634. * @return Worker[]
  635. */
  636. public static function getAllWorkers(): array
  637. {
  638. return static::$workers;
  639. }
  640. /**
  641. * Get global event-loop instance.
  642. *
  643. * @return EventInterface
  644. */
  645. public static function getEventLoop(): EventInterface
  646. {
  647. return static::$globalEvent;
  648. }
  649. /**
  650. * Get main socket resource
  651. * @return resource
  652. */
  653. public function getMainSocket()
  654. {
  655. return $this->mainSocket;
  656. }
  657. /**
  658. * Init idMap.
  659. *
  660. * @return void
  661. */
  662. protected static function initId(): void
  663. {
  664. foreach (static::$workers as $workerId => $worker) {
  665. $newIdMap = [];
  666. $worker->count = max($worker->count, 1);
  667. for ($key = 0; $key < $worker->count; $key++) {
  668. $newIdMap[$key] = static::$idMap[$workerId][$key] ?? 0;
  669. }
  670. static::$idMap[$workerId] = $newIdMap;
  671. }
  672. }
  673. /**
  674. * Get unix user of current porcess.
  675. *
  676. * @return string
  677. */
  678. protected static function getCurrentUser(): string
  679. {
  680. $userInfo = posix_getpwuid(posix_getuid());
  681. return $userInfo['name'] ?? 'unknown';
  682. }
  683. /**
  684. * Display staring UI.
  685. *
  686. * @return void
  687. */
  688. protected static function displayUI(): void
  689. {
  690. $tmpArgv = static::getArgv();
  691. if (in_array('-q', $tmpArgv)) {
  692. return;
  693. }
  694. if (DIRECTORY_SEPARATOR !== '/') {
  695. static::safeEcho("----------------------- WORKERMAN -----------------------------\r\n");
  696. static::safeEcho('Workerman version:' . static::VERSION . ' PHP version:' . PHP_VERSION . "\r\n");
  697. static::safeEcho("------------------------ WORKERS -------------------------------\r\n");
  698. static::safeEcho("worker listen processes status\r\n");
  699. return;
  700. }
  701. //show version
  702. $lineVersion = 'Workerman version:' . static::VERSION . str_pad('PHP version:', 16, ' ', STR_PAD_LEFT) . PHP_VERSION . str_pad('Event-loop:', 16, ' ', STR_PAD_LEFT) . static::getEventLoopName() . PHP_EOL;
  703. !defined('LINE_VERSIOIN_LENGTH') && define('LINE_VERSIOIN_LENGTH', strlen($lineVersion));
  704. $totalLength = static::getSingleLineTotalLength();
  705. $lineOne = '<n>' . str_pad('<w> WORKERMAN </w>', $totalLength + strlen('<w></w>'), '-', STR_PAD_BOTH) . '</n>' . PHP_EOL;
  706. $lineTwo = str_pad('<w> WORKERS </w>', $totalLength + strlen('<w></w>'), '-', STR_PAD_BOTH) . PHP_EOL;
  707. static::safeEcho($lineOne . $lineVersion . $lineTwo);
  708. //Show title
  709. $title = '';
  710. foreach (static::getUiColumns() as $columnName => $prop) {
  711. $key = 'max' . ucfirst(strtolower($columnName)) . 'NameLength';
  712. //just keep compatible with listen name
  713. $columnName === 'socket' && $columnName = 'listen';
  714. $title .= "<w>$columnName</w>" . str_pad('', static::$$key + static::UI_SAFE_LENGTH - strlen($columnName));
  715. }
  716. $title && static::safeEcho($title . PHP_EOL);
  717. //Show content
  718. foreach (static::$workers as $worker) {
  719. $content = '';
  720. foreach (static::getUiColumns() as $columnName => $prop) {
  721. $propValue = (string)($worker->$prop ?? $worker->context->$prop);
  722. $key = 'max' . ucfirst(strtolower($columnName)) . 'NameLength';
  723. preg_match_all("/(<n>|<\/n>|<w>|<\/w>|<g>|<\/g>)/i", $propValue, $matches);
  724. $placeHolderLength = !empty($matches) ? strlen(implode('', $matches[0])) : 0;
  725. $content .= str_pad($propValue, static::$$key + static::UI_SAFE_LENGTH + $placeHolderLength);
  726. }
  727. $content && static::safeEcho($content . PHP_EOL);
  728. }
  729. //Show last line
  730. $lineLast = str_pad('', static::getSingleLineTotalLength(), '-') . PHP_EOL;
  731. !empty($content) && static::safeEcho($lineLast);
  732. if (static::$daemonize) {
  733. global $argv;
  734. $startFile = $argv[0];
  735. static::safeEcho('Input "php ' . $startFile . ' stop" to stop. Start success.' . "\n\n");
  736. } else {
  737. static::safeEcho("Press Ctrl+C to stop. Start success.\n");
  738. }
  739. }
  740. /**
  741. * Get UI columns to be shown in terminal
  742. *
  743. * 1. $columnMap: ['ui_column_name' => 'clas_property_name']
  744. * 2. Consider move into configuration in future
  745. *
  746. * @return array
  747. */
  748. public static function getUiColumns(): array
  749. {
  750. return [
  751. 'proto' => 'transport',
  752. 'user' => 'user',
  753. 'worker' => 'name',
  754. 'socket' => 'statusSocket',
  755. 'processes' => 'count',
  756. 'state' => 'statusState',
  757. ];
  758. }
  759. /**
  760. * Get single line total length for ui
  761. *
  762. * @return int
  763. */
  764. public static function getSingleLineTotalLength(): int
  765. {
  766. $totalLength = 0;
  767. foreach (static::getUiColumns() as $columnName => $prop) {
  768. $key = 'max' . ucfirst(strtolower($columnName)) . 'NameLength';
  769. $totalLength += static::$$key + static::UI_SAFE_LENGTH;
  770. }
  771. //Keep beauty when show less columns
  772. !defined('LINE_VERSIOIN_LENGTH') && define('LINE_VERSIOIN_LENGTH', 0);
  773. $totalLength <= LINE_VERSIOIN_LENGTH && $totalLength = LINE_VERSIOIN_LENGTH;
  774. return $totalLength;
  775. }
  776. /**
  777. * Parse command.
  778. *
  779. * @return void
  780. */
  781. protected static function parseCommand(): void
  782. {
  783. global $argv;
  784. if (DIRECTORY_SEPARATOR !== '/' || empty($argv)) {
  785. return;
  786. }
  787. // Check argv;
  788. $startFile = $argv[0];
  789. $usage = "Usage: php yourfile <command> [mode]\nCommands: \nstart\t\tStart worker in DEBUG mode.\n\t\tUse mode -d to start in DAEMON mode.\nstop\t\tStop worker.\n\t\tUse mode -g to stop gracefully.\nrestart\t\tRestart workers.\n\t\tUse mode -d to start in DAEMON mode.\n\t\tUse mode -g to stop gracefully.\nreload\t\tReload codes.\n\t\tUse mode -g to reload gracefully.\nstatus\t\tGet worker status.\n\t\tUse mode -d to show live status.\nconnections\tGet worker connections.\n";
  790. $availableCommands = [
  791. 'start',
  792. 'stop',
  793. 'restart',
  794. 'reload',
  795. 'status',
  796. 'connections',
  797. ];
  798. $availableMode = [
  799. '-d',
  800. '-g'
  801. ];
  802. $command = $mode = '';
  803. foreach (static::getArgv() as $value) {
  804. if (in_array($value, $availableCommands)) {
  805. $command = $value;
  806. } elseif (in_array($value, $availableMode)) {
  807. $mode = $value;
  808. }
  809. }
  810. if (!$command) {
  811. exit($usage);
  812. }
  813. // Start command.
  814. $modeStr = '';
  815. if ($command === 'start') {
  816. if ($mode === '-d' || static::$daemonize) {
  817. $modeStr = 'in DAEMON mode';
  818. } else {
  819. $modeStr = 'in DEBUG mode';
  820. }
  821. }
  822. static::log("Workerman[$startFile] $command $modeStr");
  823. // Get master process PID.
  824. $masterPid = is_file(static::$pidFile) ? (int)file_get_contents(static::$pidFile) : 0;
  825. // Master is still alive?
  826. if (static::checkMasterIsAlive($masterPid)) {
  827. if ($command === 'start') {
  828. static::log("Workerman[$startFile] already running");
  829. exit;
  830. }
  831. } elseif ($command !== 'start' && $command !== 'restart') {
  832. static::log("Workerman[$startFile] not run");
  833. exit;
  834. }
  835. $statisticsFile = static::$statusFile ?: __DIR__ . "/../workerman-$masterPid.$command";
  836. // execute command.
  837. switch ($command) {
  838. case 'start':
  839. if ($mode === '-d') {
  840. static::$daemonize = true;
  841. }
  842. break;
  843. case 'status':
  844. while (1) {
  845. if (is_file($statisticsFile)) {
  846. @unlink($statisticsFile);
  847. }
  848. // Master process will send SIGIOT signal to all child processes.
  849. posix_kill($masterPid, SIGIOT);
  850. // Sleep 1 second.
  851. sleep(1);
  852. // Clear terminal.
  853. if ($mode === '-d') {
  854. static::safeEcho("\33[H\33[2J\33(B\33[m", true);
  855. }
  856. // Echo status data.
  857. static::safeEcho(static::formatStatusData($statisticsFile));
  858. if ($mode !== '-d') {
  859. @unlink($statisticsFile);
  860. exit(0);
  861. }
  862. static::safeEcho("\nPress Ctrl+C to quit.\n\n");
  863. }
  864. case 'connections':
  865. if (is_file($statisticsFile) && is_writable($statisticsFile)) {
  866. unlink($statisticsFile);
  867. }
  868. // Master process will send SIGIO signal to all child processes.
  869. posix_kill($masterPid, SIGIO);
  870. // Waiting amoment.
  871. usleep(500000);
  872. // Display statisitcs data from a disk file.
  873. if (is_readable($statisticsFile)) {
  874. readfile($statisticsFile);
  875. }
  876. exit(0);
  877. case 'restart':
  878. case 'stop':
  879. if ($mode === '-g') {
  880. static::$gracefulStop = true;
  881. $sig = SIGQUIT;
  882. static::log("Workerman[$startFile] is gracefully stopping ...");
  883. } else {
  884. static::$gracefulStop = false;
  885. $sig = SIGINT;
  886. static::log("Workerman[$startFile] is stopping ...");
  887. }
  888. // Send stop signal to master process.
  889. $masterPid && posix_kill($masterPid, $sig);
  890. // Timeout.
  891. $timeout = static::$stopTimeout + 3;
  892. $startTime = time();
  893. // Check master process is still alive?
  894. while (1) {
  895. $masterIsAlive = $masterPid && posix_kill($masterPid, 0);
  896. if ($masterIsAlive) {
  897. // Timeout?
  898. if (!static::$gracefulStop && time() - $startTime >= $timeout) {
  899. static::log("Workerman[$startFile] stop fail");
  900. exit;
  901. }
  902. // Waiting amoment.
  903. usleep(10000);
  904. continue;
  905. }
  906. // Stop success.
  907. static::log("Workerman[$startFile] stop success");
  908. if ($command === 'stop') {
  909. exit(0);
  910. }
  911. if ($mode === '-d') {
  912. static::$daemonize = true;
  913. }
  914. break;
  915. }
  916. break;
  917. case 'reload':
  918. if ($mode === '-g') {
  919. $sig = SIGUSR2;
  920. } else {
  921. $sig = SIGUSR1;
  922. }
  923. posix_kill($masterPid, $sig);
  924. exit;
  925. default :
  926. static::safeEcho('Unknown command: ' . $command . "\n");
  927. exit($usage);
  928. }
  929. }
  930. /**
  931. * Get argv.
  932. *
  933. * @return array
  934. */
  935. public static function getArgv(): array
  936. {
  937. global $argv;
  938. return isset($argv[1]) ? $argv : (static::$command ? explode(' ', static::$command) : $argv);
  939. }
  940. /**
  941. * Format status data.
  942. *
  943. * @param $statisticsFile
  944. * @return string
  945. */
  946. protected static function formatStatusData($statisticsFile): string
  947. {
  948. static $totalRequestCache = [];
  949. if (!is_readable($statisticsFile)) {
  950. return '';
  951. }
  952. $info = file($statisticsFile, FILE_IGNORE_NEW_LINES);
  953. if (!$info) {
  954. return '';
  955. }
  956. $statusStr = '';
  957. $currentTotalRequest = [];
  958. $workerInfo = unserialize($info[0]);
  959. ksort($workerInfo, SORT_NUMERIC);
  960. unset($info[0]);
  961. $dataWaitingSort = [];
  962. $readProcessStatus = false;
  963. $totalRequests = 0;
  964. $totalQps = 0;
  965. $totalConnections = 0;
  966. $totalFails = 0;
  967. $totalMemory = 0;
  968. $totalTimers = 0;
  969. $maxLen1 = static::$maxSocketNameLength;
  970. $maxLen2 = static::$maxWorkerNameLength;
  971. foreach ($info as $value) {
  972. if (!$readProcessStatus) {
  973. $statusStr .= $value . "\n";
  974. if (preg_match('/^pid.*?memory.*?listening/', $value)) {
  975. $readProcessStatus = true;
  976. }
  977. continue;
  978. }
  979. if (preg_match('/^[0-9]+/', $value, $pidMath)) {
  980. $pid = $pidMath[0];
  981. $dataWaitingSort[$pid] = $value;
  982. if (preg_match('/^\S+?\s+?(\S+?)\s+?(\S+?)\s+?(\S+?)\s+?(\S+?)\s+?(\S+?)\s+?(\S+?)\s+?(\S+?)\s+?/', $value, $match)) {
  983. $totalMemory += (int)str_ireplace('M', '', $match[1]);
  984. $maxLen1 = max($maxLen1, strlen($match[2]));
  985. $maxLen2 = max($maxLen2, strlen($match[3]));
  986. $totalConnections += (int)$match[4];
  987. $totalFails += (int)$match[5];
  988. $totalTimers += (int)$match[6];
  989. $currentTotalRequest[$pid] = $match[7];
  990. $totalRequests += (int)$match[7];
  991. }
  992. }
  993. }
  994. foreach ($workerInfo as $pid => $info) {
  995. if (!isset($dataWaitingSort[$pid])) {
  996. $statusStr .= "$pid\t" . str_pad('N/A', 7) . " "
  997. . str_pad($info['listen'], static::$maxSocketNameLength) . " "
  998. . str_pad((string)$info['name'], static::$maxWorkerNameLength) . " "
  999. . str_pad('N/A', 11) . " " . str_pad('N/A', 9) . " "
  1000. . str_pad('N/A', 7) . " " . str_pad('N/A', 13) . " N/A [busy] \n";
  1001. continue;
  1002. }
  1003. //$qps = isset($totalRequestCache[$pid]) ? $currentTotalRequest[$pid]
  1004. if (!isset($totalRequestCache[$pid]) || !isset($currentTotalRequest[$pid])) {
  1005. $qps = 0;
  1006. } else {
  1007. $qps = $currentTotalRequest[$pid] - $totalRequestCache[$pid];
  1008. $totalQps += $qps;
  1009. }
  1010. $statusStr .= $dataWaitingSort[$pid] . " " . str_pad((string)$qps, 6) . " [idle]\n";
  1011. }
  1012. $totalRequestCache = $currentTotalRequest;
  1013. $statusStr .= "----------------------------------------------PROCESS STATUS---------------------------------------------------\n";
  1014. $statusStr .= "Summary\t" . str_pad($totalMemory . 'M', 7) . " "
  1015. . str_pad('-', $maxLen1) . " "
  1016. . str_pad('-', $maxLen2) . " "
  1017. . str_pad((string)$totalConnections, 11) . " " . str_pad((string)$totalFails, 9) . " "
  1018. . str_pad((string)$totalTimers, 7) . " " . str_pad((string)$totalRequests, 13) . " "
  1019. . str_pad((string)$totalQps, 6) . " [Summary] \n";
  1020. return $statusStr;
  1021. }
  1022. /**
  1023. * Install signal handler.
  1024. *
  1025. * @return void
  1026. */
  1027. protected static function installSignal(): void
  1028. {
  1029. if (DIRECTORY_SEPARATOR !== '/') {
  1030. return;
  1031. }
  1032. $signals = [SIGINT, SIGTERM, SIGHUP, SIGTSTP, SIGQUIT, SIGUSR1, SIGUSR2, SIGIOT, SIGIO];
  1033. foreach ($signals as $signal) {
  1034. pcntl_signal($signal, [static::class, 'signalHandler'], false);
  1035. }
  1036. // ignore
  1037. pcntl_signal(SIGPIPE, SIG_IGN, false);
  1038. }
  1039. /**
  1040. * Reinstall signal handler.
  1041. *
  1042. * @return void
  1043. * @throws Throwable
  1044. */
  1045. protected static function reinstallSignal(): void
  1046. {
  1047. if (DIRECTORY_SEPARATOR !== '/') {
  1048. return;
  1049. }
  1050. $signals = [SIGINT, SIGTERM, SIGHUP, SIGTSTP, SIGQUIT, SIGUSR1, SIGUSR2, SIGIOT, SIGIO];
  1051. foreach ($signals as $signal) {
  1052. pcntl_signal($signal, SIG_IGN, false);
  1053. static::$globalEvent->onSignal($signal, [static::class, 'signalHandler']);
  1054. }
  1055. }
  1056. /**
  1057. * Signal handler.
  1058. *
  1059. * @param int $signal
  1060. * @throws Exception
  1061. */
  1062. public static function signalHandler(int $signal): void
  1063. {
  1064. switch ($signal) {
  1065. // Stop.
  1066. case SIGINT:
  1067. case SIGTERM:
  1068. case SIGHUP:
  1069. case SIGTSTP:
  1070. static::$gracefulStop = false;
  1071. static::stopAll();
  1072. break;
  1073. // Graceful stop.
  1074. case SIGQUIT:
  1075. static::$gracefulStop = true;
  1076. static::stopAll();
  1077. break;
  1078. // Reload.
  1079. case SIGUSR2:
  1080. case SIGUSR1:
  1081. if (static::$status === static::STATUS_RELOADING || static::$status === static::STATUS_SHUTDOWN) {
  1082. return;
  1083. }
  1084. static::$gracefulStop = $signal === SIGUSR2;
  1085. static::$pidsToRestart = static::getAllWorkerPids();
  1086. static::reload();
  1087. break;
  1088. // Show status.
  1089. case SIGIOT:
  1090. static::writeStatisticsToStatusFile();
  1091. break;
  1092. // Show connection status.
  1093. case SIGIO:
  1094. static::writeConnectionsStatisticsToStatusFile();
  1095. break;
  1096. }
  1097. }
  1098. /**
  1099. * Run as daemon mode.
  1100. *
  1101. * @throws Exception
  1102. */
  1103. protected static function daemonize(): void
  1104. {
  1105. if (!static::$daemonize || DIRECTORY_SEPARATOR !== '/') {
  1106. return;
  1107. }
  1108. umask(0);
  1109. $pid = pcntl_fork();
  1110. if (-1 === $pid) {
  1111. throw new RuntimeException('Fork fail');
  1112. } elseif ($pid > 0) {
  1113. exit(0);
  1114. }
  1115. if (-1 === posix_setsid()) {
  1116. throw new RuntimeException("Setsid fail");
  1117. }
  1118. // Fork again avoid SVR4 system regain the control of terminal.
  1119. $pid = pcntl_fork();
  1120. if (-1 === $pid) {
  1121. throw new RuntimeException("Fork fail");
  1122. } elseif (0 !== $pid) {
  1123. exit(0);
  1124. }
  1125. }
  1126. /**
  1127. * Redirect standard input and output.
  1128. *
  1129. * @param bool $throwException
  1130. * @return void
  1131. * @throws Exception
  1132. */
  1133. public static function resetStd(bool $throwException = true): void
  1134. {
  1135. if (!static::$daemonize || DIRECTORY_SEPARATOR !== '/') {
  1136. return;
  1137. }
  1138. global $STDOUT, $STDERR;
  1139. $handle = fopen(static::$stdoutFile, "a");
  1140. if ($handle) {
  1141. unset($handle);
  1142. set_error_handler(function () {
  1143. });
  1144. if ($STDOUT) {
  1145. fclose($STDOUT);
  1146. }
  1147. if ($STDERR) {
  1148. fclose($STDERR);
  1149. }
  1150. if (is_resource(STDOUT)) {
  1151. fclose(STDOUT);
  1152. }
  1153. if (is_resource(STDERR)) {
  1154. fclose(STDERR);
  1155. }
  1156. $STDOUT = fopen(static::$stdoutFile, "a");
  1157. $STDERR = fopen(static::$stdoutFile, "a");
  1158. // Fix standard output cannot redirect of PHP 8.1.8's bug
  1159. if (function_exists('posix_isatty') && posix_isatty(2)) {
  1160. ob_start(function ($string) {
  1161. file_put_contents(static::$stdoutFile, $string, FILE_APPEND);
  1162. }, 1);
  1163. }
  1164. // change output stream
  1165. static::$outputStream = null;
  1166. static::outputStream($STDOUT);
  1167. restore_error_handler();
  1168. return;
  1169. }
  1170. if ($throwException) {
  1171. throw new RuntimeException('Can not open stdoutFile ' . static::$stdoutFile);
  1172. }
  1173. }
  1174. /**
  1175. * Save pid.
  1176. *
  1177. * @throws Exception
  1178. */
  1179. protected static function saveMasterPid(): void
  1180. {
  1181. global $argv;
  1182. if (DIRECTORY_SEPARATOR !== '/' || empty($argv)) {
  1183. return;
  1184. }
  1185. static::$masterPid = posix_getpid();
  1186. if (false === file_put_contents(static::$pidFile, static::$masterPid)) {
  1187. throw new RuntimeException('can not save pid to ' . static::$pidFile);
  1188. }
  1189. }
  1190. /**
  1191. * Get event loop name.
  1192. *
  1193. * @return string
  1194. */
  1195. protected static function getEventLoopName(): string
  1196. {
  1197. if (static::$eventLoopClass) {
  1198. return static::$eventLoopClass;
  1199. }
  1200. if (class_exists(EventLoop::class)) {
  1201. static::$eventLoopClass = Revolt::class;
  1202. return static::$eventLoopClass;
  1203. }
  1204. $loopName = '';
  1205. foreach (static::$availableEventLoops as $name => $class) {
  1206. if (extension_loaded($name)) {
  1207. $loopName = $name;
  1208. break;
  1209. }
  1210. }
  1211. if ($loopName) {
  1212. static::$eventLoopClass = static::$availableEventLoops[$loopName];
  1213. } else {
  1214. static::$eventLoopClass = Select::class;
  1215. }
  1216. return static::$eventLoopClass;
  1217. }
  1218. /**
  1219. * Get all pids of worker processes.
  1220. *
  1221. * @return array
  1222. */
  1223. protected static function getAllWorkerPids(): array
  1224. {
  1225. $pidArray = [];
  1226. foreach (static::$pidMap as $workerPidArray) {
  1227. foreach ($workerPidArray as $workerPid) {
  1228. $pidArray[$workerPid] = $workerPid;
  1229. }
  1230. }
  1231. return $pidArray;
  1232. }
  1233. /**
  1234. * Fork some worker processes.
  1235. *
  1236. * @return void
  1237. * @throws Throwable
  1238. */
  1239. protected static function forkWorkers(): void
  1240. {
  1241. if (DIRECTORY_SEPARATOR === '/') {
  1242. static::forkWorkersForLinux();
  1243. } else {
  1244. static::forkWorkersForWindows();
  1245. }
  1246. }
  1247. /**
  1248. * Fork some worker processes.
  1249. *
  1250. * @return void
  1251. * @throws Throwable
  1252. */
  1253. protected static function forkWorkersForLinux(): void
  1254. {
  1255. foreach (static::$workers as $worker) {
  1256. if (static::$status === static::STATUS_STARTING) {
  1257. if (empty($worker->name)) {
  1258. $worker->name = $worker->getSocketName();
  1259. }
  1260. $workerNameLength = strlen($worker->name);
  1261. if (static::$maxWorkerNameLength < $workerNameLength) {
  1262. static::$maxWorkerNameLength = $workerNameLength;
  1263. }
  1264. }
  1265. while (count(static::$pidMap[$worker->workerId]) < $worker->count) {
  1266. static::forkOneWorkerForLinux($worker);
  1267. }
  1268. }
  1269. }
  1270. /**
  1271. * Fork some worker processes.
  1272. *
  1273. * @return void
  1274. * @throws Throwable
  1275. */
  1276. protected static function forkWorkersForWindows(): void
  1277. {
  1278. $files = static::getStartFilesForWindows();
  1279. if (in_array('-q', static::getArgv()) || count($files) === 1) {
  1280. if (count(static::$workers) > 1) {
  1281. static::safeEcho("@@@ Error: multi workers init in one php file are not support @@@\r\n");
  1282. static::safeEcho("@@@ See https://www.workerman.net/doc/workerman/faq/multi-woker-for-windows.html @@@\r\n");
  1283. } elseif (count(static::$workers) <= 0) {
  1284. exit("@@@no worker inited@@@\r\n\r\n");
  1285. }
  1286. reset(static::$workers);
  1287. /** @var Worker $worker */
  1288. $worker = current(static::$workers);
  1289. Timer::delAll();
  1290. //Update process state.
  1291. static::$status = static::STATUS_RUNNING;
  1292. // Register shutdown function for checking errors.
  1293. register_shutdown_function([__CLASS__, 'checkErrors']);
  1294. // Create a global event loop.
  1295. if (!static::$globalEvent) {
  1296. $eventLoopClass = static::getEventLoopName();
  1297. static::$globalEvent = new $eventLoopClass;
  1298. static::$globalEvent->setErrorHandler(function ($exception) {
  1299. static::stopAll(250, $exception);
  1300. });
  1301. }
  1302. // Reinstall signal.
  1303. static::reinstallSignal();
  1304. // Init Timer.
  1305. Timer::init(static::$globalEvent);
  1306. restore_error_handler();
  1307. // Display UI.
  1308. static::safeEcho(str_pad($worker->name, 21) . str_pad($worker->getSocketName(), 36) . str_pad((string)$worker->count, 10) . "[ok]\n");
  1309. $worker->listen();
  1310. $worker->run();
  1311. static::$globalEvent->run();
  1312. if (static::$status !== self::STATUS_SHUTDOWN) {
  1313. $err = new Exception('event-loop exited');
  1314. static::log($err);
  1315. exit(250);
  1316. }
  1317. exit(0);
  1318. } else {
  1319. static::$globalEvent = new Select();
  1320. static::$globalEvent->setErrorHandler(function ($exception) {
  1321. static::stopAll(250, $exception);
  1322. });
  1323. Timer::init(static::$globalEvent);
  1324. foreach ($files as $startFile) {
  1325. static::forkOneWorkerForWindows($startFile);
  1326. }
  1327. }
  1328. }
  1329. /**
  1330. * Get start files for windows.
  1331. *
  1332. * @return array
  1333. */
  1334. public static function getStartFilesForWindows(): array
  1335. {
  1336. $files = [];
  1337. foreach (static::getArgv() as $file) {
  1338. if (is_file($file)) {
  1339. $files[$file] = $file;
  1340. }
  1341. }
  1342. return $files;
  1343. }
  1344. /**
  1345. * Fork one worker process.
  1346. *
  1347. * @param string $startFile
  1348. */
  1349. public static function forkOneWorkerForWindows(string $startFile): void
  1350. {
  1351. $startFile = realpath($startFile);
  1352. $descriptor_spec = array(
  1353. STDIN, STDOUT, STDOUT
  1354. );
  1355. $pipes = array();
  1356. $process = proc_open('"' . PHP_BINARY . '" ' . " \"$startFile\" -q", $descriptor_spec, $pipes, null, null, ['bypass_shell' => true]);
  1357. if (empty(static::$globalEvent)) {
  1358. static::$globalEvent = new Select();
  1359. static::$globalEvent->setErrorHandler(function ($exception) {
  1360. static::stopAll(250, $exception);
  1361. });
  1362. Timer::init(static::$globalEvent);
  1363. }
  1364. // 保存子进程句柄
  1365. static::$processForWindows[$startFile] = array($process, $startFile);
  1366. }
  1367. /**
  1368. * check worker status for windows.
  1369. * @return void
  1370. */
  1371. public static function checkWorkerStatusForWindows(): void
  1372. {
  1373. foreach (static::$processForWindows as $processData) {
  1374. $process = $processData[0];
  1375. $startFile = $processData[1];
  1376. $status = proc_get_status($process);
  1377. if (isset($status['running'])) {
  1378. if (!$status['running']) {
  1379. static::safeEcho("process $startFile terminated and try to restart\n");
  1380. proc_close($process);
  1381. static::forkOneWorkerForWindows($startFile);
  1382. }
  1383. } else {
  1384. static::safeEcho("proc_get_status fail\n");
  1385. }
  1386. }
  1387. }
  1388. /**
  1389. * Fork one worker process.
  1390. *
  1391. * @param self $worker
  1392. * @throws Exception|Throwable
  1393. */
  1394. protected static function forkOneWorkerForLinux(self $worker): void
  1395. {
  1396. // Get available worker id.
  1397. $id = static::getId($worker->workerId, 0);
  1398. $pid = pcntl_fork();
  1399. // For master process.
  1400. if ($pid > 0) {
  1401. static::$pidMap[$worker->workerId][$pid] = $pid;
  1402. static::$idMap[$worker->workerId][$id] = $pid;
  1403. } // For child processes.
  1404. elseif (0 === $pid) {
  1405. srand();
  1406. mt_srand();
  1407. static::$gracefulStop = false;
  1408. if (static::$status === static::STATUS_STARTING) {
  1409. static::resetStd();
  1410. }
  1411. static::$pidsToRestart = static::$pidMap = [];
  1412. // Remove other listener.
  1413. foreach (static::$workers as $key => $oneWorker) {
  1414. if ($oneWorker->workerId !== $worker->workerId) {
  1415. $oneWorker->unlisten();
  1416. unset(static::$workers[$key]);
  1417. }
  1418. }
  1419. Timer::delAll();
  1420. //Update process state.
  1421. static::$status = static::STATUS_RUNNING;
  1422. // Register shutdown function for checking errors.
  1423. register_shutdown_function(["\\Workerman\\Worker", 'checkErrors']);
  1424. // Create a global event loop.
  1425. if (!static::$globalEvent) {
  1426. $eventLoopClass = static::getEventLoopName();
  1427. static::$globalEvent = new $eventLoopClass;
  1428. static::$globalEvent->setErrorHandler(function ($exception) {
  1429. static::stopAll(250, $exception);
  1430. });
  1431. }
  1432. // Reinstall signal.
  1433. static::reinstallSignal();
  1434. // Init Timer.
  1435. Timer::init(static::$globalEvent);
  1436. restore_error_handler();
  1437. static::setProcessTitle(self::$processTitle . ': worker process ' . $worker->name . ' ' . $worker->getSocketName());
  1438. $worker->setUserAndGroup();
  1439. $worker->id = $id;
  1440. $worker->run();
  1441. // Main loop.
  1442. static::$globalEvent->run();
  1443. if (static::$status !== self::STATUS_SHUTDOWN) {
  1444. $err = new Exception('event-loop exited');
  1445. static::log($err);
  1446. exit(250);
  1447. }
  1448. exit(0);
  1449. } else {
  1450. throw new RuntimeException("forkOneWorker fail");
  1451. }
  1452. }
  1453. /**
  1454. * Get worker id.
  1455. *
  1456. * @param string $workerId
  1457. * @param int $pid
  1458. *
  1459. * @return false|int|string
  1460. */
  1461. protected static function getId(string $workerId, int $pid): bool|int|string
  1462. {
  1463. return array_search($pid, static::$idMap[$workerId]);
  1464. }
  1465. /**
  1466. * Set unix user and group for current process.
  1467. *
  1468. * @return void
  1469. */
  1470. public function setUserAndGroup(): void
  1471. {
  1472. // Get uid.
  1473. $userInfo = posix_getpwnam($this->user);
  1474. if (!$userInfo) {
  1475. static::log("Warning: User $this->user not exists");
  1476. return;
  1477. }
  1478. $uid = $userInfo['uid'];
  1479. // Get gid.
  1480. if ($this->group) {
  1481. $groupInfo = posix_getgrnam($this->group);
  1482. if (!$groupInfo) {
  1483. static::log("Warning: Group $this->group not exists");
  1484. return;
  1485. }
  1486. $gid = $groupInfo['gid'];
  1487. } else {
  1488. $gid = $userInfo['gid'];
  1489. }
  1490. // Set uid and gid.
  1491. if ($uid !== posix_getuid() || $gid !== posix_getgid()) {
  1492. if (!posix_setgid($gid) || !posix_initgroups($userInfo['name'], $gid) || !posix_setuid($uid)) {
  1493. static::log("Warning: change gid or uid fail.");
  1494. }
  1495. }
  1496. }
  1497. /**
  1498. * Set process name.
  1499. *
  1500. * @param string $title
  1501. * @return void
  1502. */
  1503. protected static function setProcessTitle(string $title): void
  1504. {
  1505. set_error_handler(function () {
  1506. });
  1507. cli_set_process_title($title);
  1508. restore_error_handler();
  1509. }
  1510. /**
  1511. * Monitor all child processes.
  1512. *
  1513. * @return void
  1514. * @throws Throwable
  1515. */
  1516. protected static function monitorWorkers(): void
  1517. {
  1518. if (DIRECTORY_SEPARATOR === '/') {
  1519. static::monitorWorkersForLinux();
  1520. } else {
  1521. static::monitorWorkersForWindows();
  1522. }
  1523. }
  1524. /**
  1525. * Monitor all child processes.
  1526. *
  1527. * @return void
  1528. * @throws Throwable
  1529. */
  1530. protected static function monitorWorkersForLinux(): void
  1531. {
  1532. static::$status = static::STATUS_RUNNING;
  1533. while (1) {
  1534. // Calls signal handlers for pending signals.
  1535. pcntl_signal_dispatch();
  1536. // Suspends execution of the current process until a child has exited, or until a signal is delivered
  1537. $status = 0;
  1538. $pid = pcntl_wait($status, WUNTRACED);
  1539. // Calls signal handlers for pending signals again.
  1540. pcntl_signal_dispatch();
  1541. // If a child has already exited.
  1542. if ($pid > 0) {
  1543. // Find out which worker process exited.
  1544. foreach (static::$pidMap as $workerId => $workerPidArray) {
  1545. if (isset($workerPidArray[$pid])) {
  1546. $worker = static::$workers[$workerId];
  1547. // Fix exit with status 2 for php8.2
  1548. if ($status === SIGINT && static::$status === static::STATUS_SHUTDOWN) {
  1549. $status = 0;
  1550. }
  1551. // Exit status.
  1552. if ($status !== 0) {
  1553. static::log("worker[$worker->name:$pid] exit with status $status");
  1554. }
  1555. // onWorkerExit
  1556. if (static::$onWorkerExit) {
  1557. try {
  1558. (static::$onWorkerExit)($worker, $status, $pid);
  1559. } catch (Throwable $exception) {
  1560. static::log("worker[$worker->name] onWorkerExit $exception");
  1561. }
  1562. }
  1563. // For Statistics.
  1564. if (!isset(static::$globalStatistics['worker_exit_info'][$workerId][$status])) {
  1565. static::$globalStatistics['worker_exit_info'][$workerId][$status] = 0;
  1566. }
  1567. ++static::$globalStatistics['worker_exit_info'][$workerId][$status];
  1568. // Clear process data.
  1569. unset(static::$pidMap[$workerId][$pid]);
  1570. // Mark id is available.
  1571. $id = static::getId($workerId, $pid);
  1572. static::$idMap[$workerId][$id] = 0;
  1573. break;
  1574. }
  1575. }
  1576. // Is still running state then fork a new worker process.
  1577. if (static::$status !== static::STATUS_SHUTDOWN) {
  1578. static::forkWorkers();
  1579. // If reloading continue.
  1580. if (isset(static::$pidsToRestart[$pid])) {
  1581. unset(static::$pidsToRestart[$pid]);
  1582. static::reload();
  1583. }
  1584. }
  1585. }
  1586. // If shutdown state and all child processes exited then master process exit.
  1587. if (static::$status === static::STATUS_SHUTDOWN && !static::getAllWorkerPids()) {
  1588. static::exitAndClearAll();
  1589. }
  1590. }
  1591. }
  1592. /**
  1593. * Monitor all child processes.
  1594. *
  1595. * @return void
  1596. * @throws Throwable
  1597. */
  1598. protected static function monitorWorkersForWindows(): void
  1599. {
  1600. Timer::add(1, "\\Workerman\\Worker::checkWorkerStatusForWindows");
  1601. static::$globalEvent->run();
  1602. }
  1603. /**
  1604. * Exit current process.
  1605. */
  1606. protected static function exitAndClearAll(): void
  1607. {
  1608. foreach (static::$workers as $worker) {
  1609. $socketName = $worker->getSocketName();
  1610. if ($worker->transport === 'unix' && $socketName) {
  1611. list(, $address) = explode(':', $socketName, 2);
  1612. $address = substr($address, strpos($address, '/') + 2);
  1613. @unlink($address);
  1614. }
  1615. }
  1616. @unlink(static::$pidFile);
  1617. static::log("Workerman[" . basename(static::$startFile) . "] has been stopped");
  1618. if (static::$onMasterStop) {
  1619. call_user_func(static::$onMasterStop);
  1620. }
  1621. exit(0);
  1622. }
  1623. /**
  1624. * Execute reload.
  1625. *
  1626. * @return void
  1627. * @throws Exception
  1628. */
  1629. protected static function reload(): void
  1630. {
  1631. // For master process.
  1632. if (static::$masterPid === posix_getpid()) {
  1633. $sig = static::$gracefulStop ? SIGUSR2 : SIGUSR1;
  1634. // Set reloading state.
  1635. if (static::$status !== static::STATUS_RELOADING && static::$status !== static::STATUS_SHUTDOWN) {
  1636. static::log("Workerman[" . basename(static::$startFile) . "] reloading");
  1637. static::$status = static::STATUS_RELOADING;
  1638. static::resetStd(false);
  1639. // Try to emit onMasterReload callback.
  1640. if (static::$onMasterReload) {
  1641. try {
  1642. call_user_func(static::$onMasterReload);
  1643. } catch (Throwable $e) {
  1644. static::stopAll(250, $e);
  1645. }
  1646. static::initId();
  1647. }
  1648. // Send reload signal to all child processes.
  1649. $reloadablePidArray = [];
  1650. foreach (static::$pidMap as $workerId => $workerPidArray) {
  1651. $worker = static::$workers[$workerId];
  1652. if ($worker->reloadable) {
  1653. foreach ($workerPidArray as $pid) {
  1654. $reloadablePidArray[$pid] = $pid;
  1655. }
  1656. } else {
  1657. foreach ($workerPidArray as $pid) {
  1658. // Send reload signal to a worker process which reloadable is false.
  1659. posix_kill($pid, $sig);
  1660. }
  1661. }
  1662. }
  1663. // Get all pids that are waiting reload.
  1664. static::$pidsToRestart = array_intersect(static::$pidsToRestart, $reloadablePidArray);
  1665. }
  1666. // Reload complete.
  1667. if (empty(static::$pidsToRestart)) {
  1668. if (static::$status !== static::STATUS_SHUTDOWN) {
  1669. static::$status = static::STATUS_RUNNING;
  1670. }
  1671. return;
  1672. }
  1673. // Continue reload.
  1674. $oneWorkerPid = current(static::$pidsToRestart);
  1675. // Send reload signal to a worker process.
  1676. posix_kill($oneWorkerPid, $sig);
  1677. // If the process does not exit after stopTimeout seconds try to kill it.
  1678. if (!static::$gracefulStop) {
  1679. Timer::add(static::$stopTimeout, '\posix_kill', [$oneWorkerPid, SIGKILL], false);
  1680. }
  1681. } // For child processes.
  1682. else {
  1683. reset(static::$workers);
  1684. $worker = current(static::$workers);
  1685. // Try to emit onWorkerReload callback.
  1686. if ($worker->onWorkerReload) {
  1687. try {
  1688. call_user_func($worker->onWorkerReload, $worker);
  1689. } catch (Throwable $e) {
  1690. static::stopAll(250, $e);
  1691. }
  1692. }
  1693. if ($worker->reloadable) {
  1694. static::stopAll();
  1695. } else {
  1696. static::resetStd(false);
  1697. }
  1698. }
  1699. }
  1700. /**
  1701. * Stop all.
  1702. *
  1703. * @param int $code
  1704. * @param mixed $log
  1705. */
  1706. public static function stopAll(int $code = 0, mixed $log = ''): void
  1707. {
  1708. if ($log) {
  1709. static::log($log);
  1710. }
  1711. static::$status = static::STATUS_SHUTDOWN;
  1712. // For master process.
  1713. if (DIRECTORY_SEPARATOR === '/' && static::$masterPid === posix_getpid()) {
  1714. static::log("Workerman[" . basename(static::$startFile) . "] stopping ...");
  1715. $workerPidArray = static::getAllWorkerPids();
  1716. // Send stop signal to all child processes.
  1717. $sig = static::$gracefulStop ? SIGQUIT : SIGINT;
  1718. foreach ($workerPidArray as $workerPid) {
  1719. // Fix exit with status 2 for php8.2
  1720. if ($sig === SIGINT && !static::$daemonize) {
  1721. Timer::add(1, '\posix_kill', [$workerPid, SIGINT], false);
  1722. } else {
  1723. posix_kill($workerPid, $sig);
  1724. }
  1725. if (!static::$gracefulStop) {
  1726. Timer::add(ceil(static::$stopTimeout), '\posix_kill', [$workerPid, SIGKILL], false);
  1727. }
  1728. }
  1729. Timer::add(1, "\\Workerman\\Worker::checkIfChildRunning");
  1730. // Remove statistics file.
  1731. if (is_file(static::$statisticsFile)) {
  1732. @unlink(static::$statisticsFile);
  1733. }
  1734. } // For child processes.
  1735. else {
  1736. // Execute exit.
  1737. $workers = array_reverse(static::$workers);
  1738. foreach ($workers as $worker) {
  1739. if (!$worker->stopping) {
  1740. $worker->stop();
  1741. $worker->stopping = true;
  1742. }
  1743. }
  1744. if (!static::$gracefulStop || ConnectionInterface::$statistics['connection_count'] <= 0) {
  1745. static::$workers = [];
  1746. static::$globalEvent?->stop();
  1747. try {
  1748. exit($code);
  1749. } catch (Exception $e) {
  1750. }
  1751. }
  1752. }
  1753. }
  1754. /**
  1755. * check if child processes is really running
  1756. */
  1757. public static function checkIfChildRunning(): void
  1758. {
  1759. foreach (static::$pidMap as $workerId => $workerPidArray) {
  1760. foreach ($workerPidArray as $pid => $workerPid) {
  1761. if (!posix_kill($pid, 0)) {
  1762. unset(static::$pidMap[$workerId][$pid]);
  1763. }
  1764. }
  1765. }
  1766. }
  1767. /**
  1768. * Get process status.
  1769. *
  1770. * @return int
  1771. */
  1772. public static function getStatus(): int
  1773. {
  1774. return static::$status;
  1775. }
  1776. /**
  1777. * If stop gracefully.
  1778. *
  1779. * @return bool
  1780. */
  1781. public static function getGracefulStop(): bool
  1782. {
  1783. return static::$gracefulStop;
  1784. }
  1785. /**
  1786. * Write statistics data to disk.
  1787. *
  1788. * @return void
  1789. */
  1790. protected static function writeStatisticsToStatusFile(): void
  1791. {
  1792. // For master process.
  1793. if (static::$masterPid === posix_getpid()) {
  1794. $allWorkerInfo = [];
  1795. foreach (static::$pidMap as $workerId => $pidArray) {
  1796. /** @var /Workerman/Worker $worker */
  1797. $worker = static::$workers[$workerId];
  1798. foreach ($pidArray as $pid) {
  1799. $allWorkerInfo[$pid] = ['name' => $worker->name, 'listen' => $worker->getSocketName()];
  1800. }
  1801. }
  1802. file_put_contents(static::$statisticsFile, serialize($allWorkerInfo) . "\n", FILE_APPEND);
  1803. $loadavg = function_exists('sys_getloadavg') ? array_map('round', sys_getloadavg(), [2, 2, 2]) : ['-', '-', '-'];
  1804. file_put_contents(static::$statisticsFile,
  1805. "----------------------------------------------GLOBAL STATUS----------------------------------------------------\n", FILE_APPEND);
  1806. file_put_contents(static::$statisticsFile,
  1807. 'Workerman version:' . static::VERSION . " PHP version:" . PHP_VERSION . "\n", FILE_APPEND);
  1808. file_put_contents(static::$statisticsFile, 'start time:' . date('Y-m-d H:i:s',
  1809. static::$globalStatistics['start_timestamp']) . ' run ' . floor((time() - static::$globalStatistics['start_timestamp']) / (24 * 60 * 60)) . ' days ' . floor(((time() - static::$globalStatistics['start_timestamp']) % (24 * 60 * 60)) / (60 * 60)) . " hours \n",
  1810. FILE_APPEND);
  1811. $loadStr = 'load average: ' . implode(", ", $loadavg);
  1812. file_put_contents(static::$statisticsFile,
  1813. str_pad($loadStr, 33) . 'event-loop:' . static::getEventLoopName() . "\n", FILE_APPEND);
  1814. file_put_contents(static::$statisticsFile,
  1815. count(static::$pidMap) . ' workers ' . count(static::getAllWorkerPids()) . " processes\n",
  1816. FILE_APPEND);
  1817. file_put_contents(static::$statisticsFile,
  1818. str_pad('worker_name', static::$maxWorkerNameLength) . " exit_status exit_count\n", FILE_APPEND);
  1819. foreach (static::$pidMap as $workerId => $workerPidArray) {
  1820. $worker = static::$workers[$workerId];
  1821. if (isset(static::$globalStatistics['worker_exit_info'][$workerId])) {
  1822. foreach (static::$globalStatistics['worker_exit_info'][$workerId] as $workerExitStatus => $workerExitCount) {
  1823. file_put_contents(static::$statisticsFile,
  1824. str_pad($worker->name, static::$maxWorkerNameLength) . " " . str_pad((string)$workerExitStatus,
  1825. 16) . " $workerExitCount\n", FILE_APPEND);
  1826. }
  1827. } else {
  1828. file_put_contents(static::$statisticsFile,
  1829. str_pad($worker->name, static::$maxWorkerNameLength) . " " . str_pad('0', 16) . " 0\n",
  1830. FILE_APPEND);
  1831. }
  1832. }
  1833. file_put_contents(static::$statisticsFile,
  1834. "----------------------------------------------PROCESS STATUS---------------------------------------------------\n",
  1835. FILE_APPEND);
  1836. file_put_contents(static::$statisticsFile,
  1837. "pid\tmemory " . str_pad('listening', static::$maxSocketNameLength) . " " . str_pad('worker_name',
  1838. static::$maxWorkerNameLength) . " connections " . str_pad('send_fail', 9) . " "
  1839. . str_pad('timers', 8) . str_pad('total_request', 13) . " qps status\n", FILE_APPEND);
  1840. chmod(static::$statisticsFile, 0722);
  1841. foreach (static::getAllWorkerPids() as $workerPid) {
  1842. posix_kill($workerPid, SIGIOT);
  1843. }
  1844. return;
  1845. }
  1846. // For child processes.
  1847. gc_collect_cycles();
  1848. if (function_exists('gc_mem_caches')) {
  1849. gc_mem_caches();
  1850. }
  1851. reset(static::$workers);
  1852. /** @var static $worker */
  1853. $worker = current(static::$workers);
  1854. $workerStatusStr = posix_getpid() . "\t" . str_pad(round(memory_get_usage() / (1024 * 1024), 2) . "M", 7)
  1855. . " " . str_pad($worker->getSocketName(), static::$maxSocketNameLength) . " "
  1856. . str_pad(($worker->name === $worker->getSocketName() ? 'none' : $worker->name), static::$maxWorkerNameLength)
  1857. . " ";
  1858. $workerStatusStr .= str_pad((string)ConnectionInterface::$statistics['connection_count'], 11)
  1859. . " " . str_pad((string)ConnectionInterface::$statistics['send_fail'], 9)
  1860. . " " . str_pad((string)static::$globalEvent->getTimerCount(), 7)
  1861. . " " . str_pad((string)ConnectionInterface::$statistics['total_request'], 13) . "\n";
  1862. file_put_contents(static::$statisticsFile, $workerStatusStr, FILE_APPEND);
  1863. }
  1864. /**
  1865. * Write statistics data to disk.
  1866. *
  1867. * @return void
  1868. */
  1869. protected static function writeConnectionsStatisticsToStatusFile(): void
  1870. {
  1871. // For master process.
  1872. if (static::$masterPid === posix_getpid()) {
  1873. file_put_contents(static::$statisticsFile, "--------------------------------------------------------------------- WORKERMAN CONNECTION STATUS --------------------------------------------------------------------------------\n", FILE_APPEND);
  1874. file_put_contents(static::$statisticsFile, "PID Worker CID Trans Protocol ipv4 ipv6 Recv-Q Send-Q Bytes-R Bytes-W Status Local Address Foreign Address\n", FILE_APPEND);
  1875. chmod(static::$statisticsFile, 0722);
  1876. foreach (static::getAllWorkerPids() as $workerPid) {
  1877. posix_kill($workerPid, SIGIO);
  1878. }
  1879. return;
  1880. }
  1881. // For child processes.
  1882. $bytesFormat = function ($bytes) {
  1883. if ($bytes > 1024 * 1024 * 1024 * 1024) {
  1884. return round($bytes / (1024 * 1024 * 1024 * 1024), 1) . "TB";
  1885. }
  1886. if ($bytes > 1024 * 1024 * 1024) {
  1887. return round($bytes / (1024 * 1024 * 1024), 1) . "GB";
  1888. }
  1889. if ($bytes > 1024 * 1024) {
  1890. return round($bytes / (1024 * 1024), 1) . "MB";
  1891. }
  1892. if ($bytes > 1024) {
  1893. return round($bytes / (1024), 1) . "KB";
  1894. }
  1895. return $bytes . "B";
  1896. };
  1897. $pid = posix_getpid();
  1898. $str = '';
  1899. reset(static::$workers);
  1900. $currentWorker = current(static::$workers);
  1901. $defaultWorkerName = $currentWorker->name;
  1902. /** @var static $worker */
  1903. foreach (TcpConnection::$connections as $connection) {
  1904. /** @var TcpConnection $connection */
  1905. $transport = $connection->transport;
  1906. $ipv4 = $connection->isIpV4() ? ' 1' : ' 0';
  1907. $ipv6 = $connection->isIpV6() ? ' 1' : ' 0';
  1908. $recvQ = $bytesFormat($connection->getRecvBufferQueueSize());
  1909. $sendQ = $bytesFormat($connection->getSendBufferQueueSize());
  1910. $localAddress = trim($connection->getLocalAddress());
  1911. $remoteAddress = trim($connection->getRemoteAddress());
  1912. $state = $connection->getStatus(false);
  1913. $bytesRead = $bytesFormat($connection->bytesRead);
  1914. $bytesWritten = $bytesFormat($connection->bytesWritten);
  1915. $id = $connection->id;
  1916. $protocol = $connection->protocol ?: $connection->transport;
  1917. $pos = strrpos($protocol, '\\');
  1918. if ($pos) {
  1919. $protocol = substr($protocol, $pos + 1);
  1920. }
  1921. if (strlen($protocol) > 15) {
  1922. $protocol = substr($protocol, 0, 13) . '..';
  1923. }
  1924. $workerName = isset($connection->worker) ? $connection->worker->name : $defaultWorkerName;
  1925. if (strlen($workerName) > 14) {
  1926. $workerName = substr($workerName, 0, 12) . '..';
  1927. }
  1928. $str .= str_pad((string)$pid, 9) . str_pad($workerName, 16) . str_pad((string)$id, 10) . str_pad($transport, 8)
  1929. . str_pad($protocol, 16) . str_pad($ipv4, 7) . str_pad($ipv6, 7) . str_pad($recvQ, 13)
  1930. . str_pad($sendQ, 13) . str_pad($bytesRead, 13) . str_pad($bytesWritten, 13) . ' '
  1931. . str_pad($state, 14) . ' ' . str_pad($localAddress, 22) . ' ' . str_pad($remoteAddress, 22) . "\n";
  1932. }
  1933. if ($str) {
  1934. file_put_contents(static::$statisticsFile, $str, FILE_APPEND);
  1935. }
  1936. }
  1937. /**
  1938. * Check errors when current process exited.
  1939. *
  1940. * @return void
  1941. */
  1942. public static function checkErrors(): void
  1943. {
  1944. if (static::STATUS_SHUTDOWN !== static::$status) {
  1945. $errorMsg = DIRECTORY_SEPARATOR === '/' ? 'Worker[' . posix_getpid() . '] process terminated' : 'Worker process terminated';
  1946. $errors = error_get_last();
  1947. if ($errors && ($errors['type'] === E_ERROR ||
  1948. $errors['type'] === E_PARSE ||
  1949. $errors['type'] === E_CORE_ERROR ||
  1950. $errors['type'] === E_COMPILE_ERROR ||
  1951. $errors['type'] === E_RECOVERABLE_ERROR)
  1952. ) {
  1953. $errorMsg .= ' with ERROR: ' . static::getErrorType($errors['type']) . " \"{$errors['message']} in {$errors['file']} on line {$errors['line']}\"";
  1954. }
  1955. static::log($errorMsg);
  1956. }
  1957. }
  1958. /**
  1959. * Get error message by error code.
  1960. *
  1961. * @param int $type
  1962. * @return string
  1963. */
  1964. protected static function getErrorType(int $type): string
  1965. {
  1966. return self::ERROR_TYPE[$type] ?? '';
  1967. }
  1968. /**
  1969. * Log.
  1970. *
  1971. * @param mixed $msg
  1972. * @param bool $decorated
  1973. * @return void
  1974. */
  1975. public static function log(mixed $msg, bool $decorated = false): void
  1976. {
  1977. $msg = $msg . "\n";
  1978. if (!static::$daemonize) {
  1979. static::safeEcho($msg, $decorated);
  1980. }
  1981. file_put_contents(static::$logFile, date('Y-m-d H:i:s') . ' ' . 'pid:'
  1982. . (DIRECTORY_SEPARATOR === '/' ? posix_getpid() : 1) . ' ' . $msg, FILE_APPEND | LOCK_EX);
  1983. }
  1984. /**
  1985. * Safe Echo.
  1986. * @param string $msg
  1987. * @param bool $decorated
  1988. * @return bool
  1989. */
  1990. public static function safeEcho(string $msg, bool $decorated = false): bool
  1991. {
  1992. $stream = static::outputStream();
  1993. if (!$stream) {
  1994. return false;
  1995. }
  1996. if ($decorated) {
  1997. $line = $white = $green = $end = '';
  1998. if (static::$outputDecorated) {
  1999. $line = "\033[1A\n\033[K";
  2000. $white = "\033[47;30m";
  2001. $green = "\033[32;40m";
  2002. $end = "\033[0m";
  2003. }
  2004. $msg = str_replace(['<n>', '<w>', '<g>'], [$line, $white, $green], $msg);
  2005. $msg = str_replace(['</n>', '</w>', '</g>'], $end, $msg);
  2006. } elseif (!static::$outputDecorated) {
  2007. return false;
  2008. }
  2009. fwrite($stream, $msg);
  2010. fflush($stream);
  2011. return true;
  2012. }
  2013. /**
  2014. * set and get output stream.
  2015. *
  2016. * @param resource|null $stream
  2017. * @return false|resource
  2018. */
  2019. private static function outputStream($stream = null)
  2020. {
  2021. if (!$stream) {
  2022. $stream = static::$outputStream ?: STDOUT;
  2023. }
  2024. if (!$stream || !is_resource($stream) || 'stream' !== get_resource_type($stream)) {
  2025. return false;
  2026. }
  2027. $stat = fstat($stream);
  2028. if (!$stat) {
  2029. return false;
  2030. }
  2031. if (($stat['mode'] & 0170000) === 0100000) { // whether is regular file
  2032. static::$outputDecorated = false;
  2033. } else {
  2034. static::$outputDecorated =
  2035. DIRECTORY_SEPARATOR === '/' && // linux or unix
  2036. function_exists('posix_isatty') &&
  2037. posix_isatty($stream); // whether is interactive terminal
  2038. }
  2039. return static::$outputStream = $stream;
  2040. }
  2041. /**
  2042. * Constructor.
  2043. *
  2044. * @param string|null $socketName
  2045. * @param array $socketContext
  2046. */
  2047. public function __construct(string $socketName = null, array $socketContext = [])
  2048. {
  2049. // Save all worker instances.
  2050. $this->workerId = spl_object_hash($this);
  2051. $this->context = new stdClass();
  2052. static::$workers[$this->workerId] = $this;
  2053. static::$pidMap[$this->workerId] = [];
  2054. // Context for socket.
  2055. if ($socketName) {
  2056. $this->socketName = $socketName;
  2057. if (!isset($socketContext['socket']['backlog'])) {
  2058. $socketContext['socket']['backlog'] = static::DEFAULT_BACKLOG;
  2059. }
  2060. $this->socketContext = stream_context_create($socketContext);
  2061. }
  2062. // Try to turn reusePort on.
  2063. /*if (\DIRECTORY_SEPARATOR === '/' // if linux
  2064. && $socketName
  2065. && \version_compare(php_uname('r'), '3.9', 'ge') // if kernel >=3.9
  2066. && \strtolower(\php_uname('s')) !== 'darwin' // if not Mac OS
  2067. && strpos($socketName,'unix') !== 0 // if not unix socket
  2068. && strpos($socketName,'udp') !== 0) { // if not udp socket
  2069. $address = \parse_url($socketName);
  2070. if (isset($address['host']) && isset($address['port'])) {
  2071. try {
  2072. \set_error_handler(function(){});
  2073. // If address not in use, turn reusePort on automatically.
  2074. $server = stream_socket_server("tcp://{$address['host']}:{$address['port']}");
  2075. if ($server) {
  2076. $this->reusePort = true;
  2077. fclose($server);
  2078. }
  2079. \restore_error_handler();
  2080. } catch (\Throwable $e) {}
  2081. }
  2082. }*/
  2083. }
  2084. /**
  2085. * Listen.
  2086. *
  2087. * @throws Exception
  2088. */
  2089. public function listen(): void
  2090. {
  2091. if (!$this->socketName) {
  2092. return;
  2093. }
  2094. if (!$this->mainSocket) {
  2095. $localSocket = $this->parseSocketAddress();
  2096. // Flag.
  2097. $flags = $this->transport === 'udp' ? STREAM_SERVER_BIND : STREAM_SERVER_BIND | STREAM_SERVER_LISTEN;
  2098. $errno = 0;
  2099. $errmsg = '';
  2100. // SO_REUSEPORT.
  2101. if ($this->reusePort) {
  2102. stream_context_set_option($this->socketContext, 'socket', 'so_reuseport', 1);
  2103. }
  2104. // Create an Internet or Unix domain server socket.
  2105. $this->mainSocket = stream_socket_server($localSocket, $errno, $errmsg, $flags, $this->socketContext);
  2106. if (!$this->mainSocket) {
  2107. throw new Exception($errmsg);
  2108. }
  2109. if ($this->transport === 'ssl') {
  2110. stream_socket_enable_crypto($this->mainSocket, false);
  2111. } elseif ($this->transport === 'unix') {
  2112. $socketFile = substr($localSocket, 7);
  2113. if ($this->user) {
  2114. chown($socketFile, $this->user);
  2115. }
  2116. if ($this->group) {
  2117. chgrp($socketFile, $this->group);
  2118. }
  2119. }
  2120. // Try to open keepalive for tcp and disable Nagle algorithm.
  2121. if (function_exists('socket_import_stream') && self::BUILD_IN_TRANSPORTS[$this->transport] === 'tcp') {
  2122. set_error_handler(function () {
  2123. });
  2124. $socket = socket_import_stream($this->mainSocket);
  2125. socket_set_option($socket, SOL_SOCKET, SO_KEEPALIVE, 1);
  2126. socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1);
  2127. restore_error_handler();
  2128. }
  2129. // Non blocking.
  2130. stream_set_blocking($this->mainSocket, false);
  2131. }
  2132. $this->resumeAccept();
  2133. }
  2134. /**
  2135. * Unlisten.
  2136. *
  2137. * @return void
  2138. */
  2139. public function unlisten(): void
  2140. {
  2141. $this->pauseAccept();
  2142. if ($this->mainSocket) {
  2143. set_error_handler(function () {
  2144. });
  2145. fclose($this->mainSocket);
  2146. restore_error_handler();
  2147. $this->mainSocket = null;
  2148. }
  2149. }
  2150. /**
  2151. * Parse local socket address.
  2152. *
  2153. * @throws Exception
  2154. */
  2155. protected function parseSocketAddress(): ?string
  2156. {
  2157. if (!$this->socketName) {
  2158. return null;
  2159. }
  2160. // Get the application layer communication protocol and listening address.
  2161. list($scheme, $address) = explode(':', $this->socketName, 2);
  2162. // Check application layer protocol class.
  2163. if (!isset(self::BUILD_IN_TRANSPORTS[$scheme])) {
  2164. $scheme = ucfirst($scheme);
  2165. $this->protocol = $scheme[0] === '\\' ? $scheme : 'Protocols\\' . $scheme;
  2166. if (!class_exists($this->protocol)) {
  2167. $this->protocol = "Workerman\\Protocols\\$scheme";
  2168. if (!class_exists($this->protocol)) {
  2169. throw new RuntimeException("class \\Protocols\\$scheme not exist");
  2170. }
  2171. }
  2172. if (!isset(self::BUILD_IN_TRANSPORTS[$this->transport])) {
  2173. throw new RuntimeException('Bad worker->transport ' . var_export($this->transport, true));
  2174. }
  2175. } else if ($this->transport === 'tcp') {
  2176. $this->transport = $scheme;
  2177. }
  2178. //local socket
  2179. return self::BUILD_IN_TRANSPORTS[$this->transport] . ":" . $address;
  2180. }
  2181. /**
  2182. * Pause accept new connections.
  2183. *
  2184. * @return void
  2185. */
  2186. public function pauseAccept(): void
  2187. {
  2188. if (static::$globalEvent && false === $this->pauseAccept && $this->mainSocket) {
  2189. static::$globalEvent->offReadable($this->mainSocket);
  2190. $this->pauseAccept = true;
  2191. }
  2192. }
  2193. /**
  2194. * Resume accept new connections.
  2195. *
  2196. * @return void
  2197. */
  2198. public function resumeAccept(): void
  2199. {
  2200. // Register a listener to be notified when server socket is ready to read.
  2201. if (static::$globalEvent && true === $this->pauseAccept && $this->mainSocket) {
  2202. if ($this->transport !== 'udp') {
  2203. static::$globalEvent->onReadable($this->mainSocket, [$this, 'acceptTcpConnection']);
  2204. } else {
  2205. static::$globalEvent->onReadable($this->mainSocket, [$this, 'acceptUdpConnection']);
  2206. }
  2207. $this->pauseAccept = false;
  2208. }
  2209. }
  2210. /**
  2211. * Get socket name.
  2212. *
  2213. * @return string
  2214. */
  2215. public function getSocketName(): string
  2216. {
  2217. return $this->socketName ? lcfirst($this->socketName) : 'none';
  2218. }
  2219. /**
  2220. * Run worker instance.
  2221. *
  2222. * @return void
  2223. * @throws Throwable
  2224. */
  2225. public function run(): void
  2226. {
  2227. $this->listen();
  2228. // Try to emit onWorkerStart callback.
  2229. if ($this->onWorkerStart) {
  2230. try {
  2231. ($this->onWorkerStart)($this);
  2232. } catch (Throwable $e) {
  2233. // Avoid rapid infinite loop exit.
  2234. sleep(1);
  2235. static::stopAll(250, $e);
  2236. }
  2237. }
  2238. }
  2239. /**
  2240. * Stop current worker instance.
  2241. *
  2242. * @return void
  2243. */
  2244. public function stop(): void
  2245. {
  2246. // Try to emit onWorkerStop callback.
  2247. if ($this->onWorkerStop) {
  2248. try {
  2249. ($this->onWorkerStop)($this);
  2250. } catch (Throwable $e) {
  2251. static::log($e);
  2252. }
  2253. }
  2254. // Remove listener for server socket.
  2255. $this->unlisten();
  2256. // Close all connections for the worker.
  2257. if (!static::$gracefulStop) {
  2258. foreach ($this->connections as $connection) {
  2259. $connection->close();
  2260. }
  2261. }
  2262. // Remove worker.
  2263. foreach (static::$workers as $key => $one_worker) {
  2264. if ($one_worker->workerId === $this->workerId) {
  2265. unset(static::$workers[$key]);
  2266. }
  2267. }
  2268. // Clear callback.
  2269. $this->onMessage = $this->onClose = $this->onError = $this->onBufferDrain = $this->onBufferFull = null;
  2270. }
  2271. /**
  2272. * Accept a connection.
  2273. *
  2274. * @param resource $socket
  2275. * @return void
  2276. */
  2277. public function acceptTcpConnection($socket): void
  2278. {
  2279. // Accept a connection on server socket.
  2280. set_error_handler(function () {
  2281. });
  2282. $newSocket = stream_socket_accept($socket, 0, $remoteAddress);
  2283. restore_error_handler();
  2284. // Thundering herd.
  2285. if (!$newSocket) {
  2286. return;
  2287. }
  2288. // TcpConnection.
  2289. $connection = new TcpConnection(static::$globalEvent, $newSocket, $remoteAddress);
  2290. $this->connections[$connection->id] = $connection;
  2291. $connection->worker = $this;
  2292. $connection->protocol = $this->protocol;
  2293. $connection->transport = $this->transport;
  2294. $connection->onMessage = $this->onMessage;
  2295. $connection->onClose = $this->onClose;
  2296. $connection->onError = $this->onError;
  2297. $connection->onBufferDrain = $this->onBufferDrain;
  2298. $connection->onBufferFull = $this->onBufferFull;
  2299. // Try to emit onConnect callback.
  2300. if ($this->onConnect) {
  2301. try {
  2302. ($this->onConnect)($connection);
  2303. } catch (Throwable $e) {
  2304. static::stopAll(250, $e);
  2305. }
  2306. }
  2307. }
  2308. /**
  2309. * For udp package.
  2310. *
  2311. * @param resource $socket
  2312. * @return bool
  2313. */
  2314. public function acceptUdpConnection($socket): bool
  2315. {
  2316. set_error_handler(function () {
  2317. });
  2318. $recvBuffer = stream_socket_recvfrom($socket, UdpConnection::MAX_UDP_PACKAGE_SIZE, 0, $remoteAddress);
  2319. restore_error_handler();
  2320. if (false === $recvBuffer || empty($remoteAddress)) {
  2321. return false;
  2322. }
  2323. // UdpConnection.
  2324. $connection = new UdpConnection($socket, $remoteAddress);
  2325. $connection->protocol = $this->protocol;
  2326. $messageCallback = $this->onMessage;
  2327. if ($messageCallback) {
  2328. try {
  2329. if ($this->protocol !== null) {
  2330. /** @var ProtocolInterface $parser */
  2331. $parser = $this->protocol;
  2332. if ($parser && method_exists($parser, 'input')) {
  2333. while ($recvBuffer !== '') {
  2334. $len = $parser::input($recvBuffer, $connection);
  2335. if ($len === 0)
  2336. return true;
  2337. $package = substr($recvBuffer, 0, $len);
  2338. $recvBuffer = substr($recvBuffer, $len);
  2339. $data = $parser::decode($package, $connection);
  2340. if ($data === false) {
  2341. continue;
  2342. }
  2343. $messageCallback($connection, $data);
  2344. }
  2345. } else {
  2346. $data = $parser::decode($recvBuffer, $connection);
  2347. // Discard bad packets.
  2348. if ($data === false) {
  2349. return true;
  2350. }
  2351. $messageCallback($connection, $data);
  2352. }
  2353. } else {
  2354. $messageCallback($connection, $recvBuffer);
  2355. }
  2356. ++ConnectionInterface::$statistics['total_request'];
  2357. } catch (Throwable $e) {
  2358. static::stopAll(250, $e);
  2359. }
  2360. }
  2361. return true;
  2362. }
  2363. /**
  2364. * Check master process is alive
  2365. *
  2366. * @param int $masterPid
  2367. * @return bool
  2368. */
  2369. protected static function checkMasterIsAlive(int $masterPid): bool
  2370. {
  2371. if (empty($masterPid)) {
  2372. return false;
  2373. }
  2374. $masterIsAlive = posix_kill($masterPid, 0) && posix_getpid() !== $masterPid;
  2375. if (!$masterIsAlive) {
  2376. return false;
  2377. }
  2378. $cmdline = "/proc/$masterPid/cmdline";
  2379. if (!is_readable($cmdline) || empty(static::$processTitle)) {
  2380. return true;
  2381. }
  2382. $content = file_get_contents($cmdline);
  2383. if (empty($content)) {
  2384. return true;
  2385. }
  2386. return stripos($content, static::$processTitle) !== false || stripos($content, 'php') !== false;
  2387. }
  2388. }