<commit_before><?php

namespace Firehed\ProcessControl;

abstract class ProcessManager {

	use \Psr\Log\LoggerAwareTrait;

	private $managerPid;
	private $workerProcesses = [];
	private $shouldWork = true;
	private $workers = 1;


	protected $myPid;

	public function __construct(\Psr\Log\LoggerInterface $logger = null) {
		$this->managerPid = $this->myPid = getmypid();
		if ($logger) {
			$this->setLogger($logger);
		}
		else {
			$this->setLogger(new \Psr\Log\NullLogger);
		}
		$this->installSignals();
	}

	protected function getLogger() {
		return $this->logger;
	}

	public function setWorkerCount($count) {
		if (!is_int($count)) {
			throw new InvalidArgumentException("Integer argument is required");
		}
		if ($count < 1) {
			throw new InvalidArgumentException("Must have at least 1 worker");
		}
		$this->workers = $count;
	}

	final public function start() {
		$this->manageWorkers();
	}

	private function cleanChildren() {
		$status = null;
		if ($exited = pcntl_wait($status, WNOHANG)) {
			unset($this->workerProcesses[$exited]);
			$this->getLogger()->info("Worker $exited got WNOHANG during normal operation");
		}
	}

	abstract protected function doWork();

	private function installSignals() {
		$this->getLogger()->debug("$this->myPid SIGTERM handler installation");
		pcntl_signal(SIGTERM, [$this,'signal']);
		pcntl_signal(SIGINT,  [$this,'signal']);
	}

	private function isParent() {
		return $this->myPid == $this->managerPid;
	}

	private function manageWorkers() {
		while ($this->shouldWork) {
			// Do nothing other than wait for SIGTERM/SIGIN
			if (count($this->workerProcesses) < $this->workers) {
				$this->spawnWorker();
			}
			else {
				$this->cleanChildren();
				sleep(5);
			}
		}
	}

	public function signal($signo) {
		if ($this->isParent()) {
			$this->stopWorking();
			$this->getLogger()->info('Parent got sigterm');
			$this->getLogger()->debug("Children: " . print_r($this->workerProcesses, true));
			$this->stopChildren(SIGTERM);
			while ($this->workerProcesses) {
				$status = null;
				if ($exited = pcntl_wait($status, WNOHANG)) {
					unset($this->workerProcesses[$exited]);
					$this->getLogger()->debug("Worker $exited got WNOHANG");
				}
				else {
					sleep(1);
				}
			}
			$this->getLogger()->info("Parent shutting down");
			exit;
		}
		else {
			$this->getLogger()->info("Child $this->myPid received SIGTERM; stopping work");
			$this->stopWorking();
		}
	}

	private function spawnWorker() {
		$this->getLogger()->info("Creating a new worker");
		switch ($pid = pcntl_fork()) {
		case -1: // Failed
			$this->getLogger()->error("Spawning worker failed");
			exit(2);
		case 0:  // Child
			$this->myPid = getmypid();
			$this->getLogger()->info("$this->myPid created");
			$this->installSignals();
			$this->work();
			break;
		default: // Parent
			$this->getLogger()->debug("Parent created child with pid $pid");
			$this->workerProcesses[$pid] = $pid;
			break;
		}
	}

	private function stopChildren($sig = SIGTERM) {
		foreach ($this->workerProcesses as $pid) {
			$this->getLogger()->debug("Sending SIGTERM to $pid");
			posix_kill($pid, $sig);
			if (!posix_kill($pid, 0)) {
				$this->getLogger()->debug("$pid is dead already");
			}
		}
	}

	protected function stopWorking() {
		$this->shouldWork = false;
	}

	private function work() {
		$this->getLogger()->debug("Child $this->myPid about to start work");
		while ( $this->shouldWork ) {
			$_SERVER['REQUEST_TIME'] = time();
			$_SERVER['REQUEST_TIME_FLOAT'] = microtime(true);
			$this->doWork();
		}
		$this->getLogger()->info("Child $this->myPid exiting");<commit_msg>tweak signal handling for more flexibility, document<commit_after><?php

namespace Firehed\ProcessControl;

abstract class ProcessManager {

	use \Psr\Log\LoggerAwareTrait;

	private $managerPid;
	private $workerProcesses = []; // pid => pid
	private $shouldWork = true;
	private $workers = 1;


	protected $myPid;

	public function __construct(\Psr\Log\LoggerInterface $logger = null) {
		$this->managerPid = $this->myPid = getmypid();
		if ($logger) {
			$this->setLogger($logger);
		}
		else {
			$this->setLogger(new \Psr\Log\NullLogger);
		}
		$this->installSignals();
	}

	protected function getLogger() {
		return $this->logger;
	}

	public function setWorkerCount($count) {
		if (!is_int($count)) {
			throw new InvalidArgumentException("Integer argument is required");
		}
		if ($count < 1) {
			throw new InvalidArgumentException("Must have at least 1 worker");
		}
		$this->workers = $count;
	}

	final public function start() {
		$this->manageWorkers();
	}

	private function cleanChildren() {
		$status = null;
		if ($exited = pcntl_wait($status, WNOHANG)) {
			unset($this->workerProcesses[$exited]);
			$this->getLogger()->info("Worker $exited got WNOHANG during normal operation");
		}
	}

	abstract protected function doWork();

	private function installSignals() {
		$this->getLogger()->debug("$this->myPid SIGTERM handler installation");
		pcntl_signal(SIGTERM, [$this,'signal']);
		pcntl_signal(SIGINT,  [$this,'signal']);
	}

	private function isParent() {
		return $this->myPid == $this->managerPid;
	}

	private function manageWorkers() {
		while ($this->shouldWork) {
			// Do nothing other than wait for SIGTERM/SIGIN
			if (count($this->workerProcesses) < $this->workers) {
				$this->spawnWorker();
			}
			else {
				$this->cleanChildren();
				sleep(5);
			}
		}
	}

	public function signal($signo) {
		switch ($signo) {
		case SIGTERM:
		case SIGINT:
			$this->handleSigterm();
			break;
		}
	}

	private function handleSigterm() {
		if ($this->isParent()) {
			$this->stopWorking();
			$this->getLogger()->info('Parent got sigterm');
			$this->getLogger()->debug("Children: ".
				print_r(array_keys($this->workerProcesses), true));
			$this->stopChildren(SIGTERM);
			while ($this->workerProcesses) {
				$status = null;
				if ($exited = pcntl_wait($status, WNOHANG)) {
					unset($this->workerProcesses[$exited]);
					$this->getLogger()->debug("Worker $exited got WNOHANG");
				}
				else {
					sleep(1);
				}
			}
			$this->getLogger()->info("Parent shutting down");
			exit;
		}
		else {
			$this->getLogger()->info("Child $this->myPid received SIGTERM; stopping work");
			$this->stopWorking();
		}
	}

	private function spawnWorker() {
		$this->getLogger()->info("Creating a new worker");
		switch ($pid = pcntl_fork()) {
		case -1: // Failed
			$this->getLogger()->error("Spawning worker failed");
			exit(2);
		case 0:  // Child
			$this->myPid = getmypid();
			$this->getLogger()->info("$this->myPid created");
			$this->installSignals();
			$this->work();
			break;
		default: // Parent
			$this->getLogger()->debug("Parent created child with pid $pid");
			$this->workerProcesses[$pid] = $pid;
			break;
		}
	}

	private function stopChildren($sig = SIGTERM) {
		foreach ($this->workerProcesses as $pid => $pidCopy) {
			$this->getLogger()->debug("Sending SIGTERM to $pid");
			posix_kill($pid, $sig);
			if (!posix_kill($pid, 0)) {
				$this->getLogger()->debug("$pid is dead already");
			}
		}
	}

	protected function stopWorking() {
		$this->shouldWork = false;
	}

	private function work() {
		$this->getLogger()->debug("Child $this->myPid about to start work");
		while ( $this->shouldWork ) {
			$_SERVER['REQUEST_TIME'] = time();
			$_SERVER['REQUEST_TIME_FLOAT'] = microtime(true);
			$this->doWork();
		}
		$this->getLogger()->info("Child $this->myPid exiting");
{'fraction_non_alphanumeric': 0.17721518987341772, 'fraction_numerical': 0.0024239159709130084, 'mean_word_length': 3.9678929765886286, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 138, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1780454', 'n_tokens_mistral': 2863, 'n_tokens_neox': 2466, 'n_words': 517}