/**
  *
  * Exec the command and return code
  *
  * @param string $cmd
  * @param string $stdout
  * @param string $stderr
  * @param int    $timeout
  * @return int|null
  */
 public static function exec($cmd, &$stdout, &$stderr, $timeout = 3600)
 {
     if ($timeout <= 0) {
         $timeout = 3600;
     }
     $descriptors = array(1 => array("pipe", "w"), 2 => array("pipe", "w"));
     $stdout = $stderr = $status = null;
     $process = proc_open($cmd, $descriptors, $pipes);
     $time_end = time() + $timeout;
     if (is_resource($process)) {
         do {
             $time_left = $time_end - time();
             $read = array($pipes[1]);
             stream_select($read, $null, $null, $time_left, NULL);
             $stdout .= fread($pipes[1], 2048);
         } while (!feof($pipes[1]) && $time_left > 0);
         fclose($pipes[1]);
         if ($time_left <= 0) {
             proc_terminate($process);
             $stderr = 'process terminated for timeout.';
             return -1;
         }
         while (!feof($pipes[2])) {
             $stderr .= fread($pipes[2], 2048);
         }
         fclose($pipes[2]);
         $status = proc_close($process);
     }
     return $status;
 }

function execute($cmd)
{
    $descriptors = array(0 => array('pipe', 'r'), 1 => array('pipe', 'w'), 2 => array('pipe', 'w'));
    $stdout = '';
    $proc = proc_open(escapeshellcmd($cmd), $descriptors, $pipes);
    if (!is_resource($proc)) {
        throw new \Exception('Could not execute process');
    }
    // Set the stdout stream to none-blocking.
    stream_set_blocking($pipes[1], 0);
    $timeout = 3000;
    // miliseconds.
    $forceKill = true;
    while ($timeout > 0) {
        $start = round(microtime(true) * 1000);
        // Wait until we have output or the timer expired.
        $status = proc_get_status($proc);
        $read = array($pipes[1]);
        stream_select($read, $other, $other, 0, $timeout);
        $stdout .= stream_get_contents($pipes[1]);
        if (!$status['running']) {
            // Break from this loop if the process exited before the timeout.
            $forceKill = false;
            break;
        }
        // Subtract the number of microseconds that we waited.
        $timeout -= round(microtime(true) * 1000) - $start;
    }
    if ($forceKill == true) {
        proc_terminate($proc, 9);
    }
    return $stdout;
}

/**
 * Execute a command and kill it if the timeout limit fired to prevent long php execution
 * 
 * @see http://stackoverflow.com/questions/2603912/php-set-timeout-for-script-with-system-call-set-time-limit-not-working
 * 
 * @param string $cmd Command to exec (you should use 2>&1 at the end to pipe all output)
 * @param integer $timeout
 * @return string Returns command output 
 */
function ExecWaitTimeout($cmd, $timeout = 5)
{
    echo $cmd . "\n";
    $descriptorspec = array(0 => array("pipe", "r"), 1 => array("pipe", "w"), 2 => array("pipe", "w"));
    $pipes = array();
    $timeout += time();
    $process = proc_open($cmd, $descriptorspec, $pipes);
    if (!is_resource($process)) {
        throw new Exception("proc_open failed on: " . $cmd);
    }
    $output = '';
    do {
        $timeleft = $timeout - time();
        $read = array($pipes[1]);
        //     if($timeleft > 0)
        stream_select($read, $write = NULL, $exeptions = NULL, $timeleft, NULL);
        if (!empty($read)) {
            $output .= fread($pipes[1], 8192);
        }
    } while (!feof($pipes[1]) && $timeleft > 0);
    if ($timeleft <= 0) {
        proc_terminate($process);
        throw new Exception("command timeout on: " . $cmd);
    } else {
        return $output;
    }
}

 protected function disposeWorker() : \Generator
 {
     try {
         $this->executor->cancel(new PoolShutdownException('Pool shut down'));
         yield from $this->transmitter->send('', SocketTransmitter::TYPE_EXIT);
         list($type) = (yield from $this->transmitter->receive());
     } catch (\Throwable $e) {
         // Cannot do anything about this...
     }
     $this->transmitter = null;
     try {
         $this->socket->close();
     } finally {
         $this->socket = null;
         if ($this->process !== null) {
             try {
                 if (empty($type) || $type !== SocketTransmitter::TYPE_EXIT) {
                     $status = @\proc_get_status($this->process);
                     if (!empty($status['running']) && empty($status['stopped'])) {
                         @\proc_terminate($this->process);
                     }
                 }
             } finally {
                 @\proc_close($this->process);
                 $this->process = null;
             }
         }
     }
 }

function proc_exec($cmd)
{
    $descriptorspec = array(0 => array("pipe", "r"), 1 => array("pipe", "w"), 2 => array("pipe", "w"));
    $ptr = proc_open($cmd, $descriptorspec, $pipes, NULL, $_ENV);
    if (!is_resource($ptr)) {
        return false;
    }
    while (!feof($ptr)) {
        $buffer = fgets($ptr);
        $buffer = trim(htmlspecialchars($buffer));
        echo "data: " . $buffer . "<br />";
        echo "data: " . str_pad('', 4096);
        ob_flush();
        flush();
    }
    while (($buffer = fgets($pipes[FD_READ], BUF_SIZ)) != NULL || ($errbuf = fgets($pipes[FD_ERR], BUF_SIZ)) != NULL) {
        if (!isset($flag)) {
            $pstatus = proc_get_status($ptr);
            $first_exitcode = $pstatus["exitcode"];
            $flag = true;
        }
        if (strlen($buffer)) {
            echo "data: INFO: have";
        }
        //. $buffer;
        ob_flush();
        flush();
        if (strlen($errbuf)) {
            ob_flush();
        }
        flush();
        echo "data: ERR: err ";
        // . $errbuf;
    }
    foreach ($pipes as $pipe) {
        fclose($pipe);
    }
    /* Get the expected *exit* code to return the value */
    $pstatus = proc_get_status($ptr);
    if (!strlen($pstatus["exitcode"]) || $pstatus["running"]) {
        /* we can trust the retval of proc_close() */
        if ($pstatus["running"]) {
            proc_terminate($ptr);
        }
        $ret = proc_close($ptr);
    } else {
        if (($first_exitcode + 256) % 256 == 255 && ($pstatus["exitcode"] + 256) % 256 != 255) {
            $ret = $pstatus["exitcode"];
        } elseif (!strlen($first_exitcode)) {
            $ret = $pstatus["exitcode"];
        } elseif (($first_exitcode + 256) % 256 != 255) {
            $ret = $first_exitcode;
        } else {
            $ret = 0;
        }
        /* we "deduce" an EXIT_SUCCESS ;) */
        proc_close($ptr);
    }
    return ($ret + 256) % 256;
}

 public function terminate($signal = 15)
 {
     $ret = proc_terminate($this->_process, $signal);
     if (!$ret) {
         throw new Kwf_Exception("terminate failed");
     }
 }

 public function stop()
 {
     if (!$this->process) {
         return;
     }
     $status = proc_get_status($this->process);
     if ($status['running']) {
         fclose($this->pipes[1]);
         //stdout
         fclose($this->pipes[2]);
         //stderr
         //get the parent pid of the process we want to kill
         $pPid = $status['pid'];
         //use ps to get all the children of this process, and kill them
         foreach (array_filter(preg_split('/\\s+/', `ps -o pid --no-heading --ppid {$pPid}`)) as $pid) {
             if (is_numeric($pid)) {
                 posix_kill($pid, 9);
                 // SIGKILL signal
             }
         }
     }
     fclose($this->pipes[0]);
     proc_terminate($this->process);
     $this->process = NULL;
 }

 public function close()
 {
     @\fclose($this->stdin);
     @\fclose($this->stdout);
     @\proc_terminate($this->process, 15);
     @\proc_close($this->process);
 }

function execute($cmd, $stdin = null, &$stdout, &$stderr, $timeout = false)
{
    $pipes = array();
    $process = proc_open($cmd, array(array('pipe', 'r'), array('pipe', 'w'), array('pipe', 'w')), $pipes);
    $start = time();
    $stdout = '';
    $stderr = '';
    if (is_resource($process)) {
        stream_set_blocking($pipes[0], 0);
        stream_set_blocking($pipes[1], 0);
        stream_set_blocking($pipes[2], 0);
        fwrite($pipes[0], $stdin);
        fclose($pipes[0]);
    }
    while (is_resource($process)) {
        $stdout .= stream_get_contents($pipes[1]);
        $stderr .= stream_get_contents($pipes[2]);
        if ($timeout !== false && time() - $start > $timeout) {
            proc_terminate($process, 9);
            return 1;
        }
        $status = proc_get_status($process);
        if (!$status['running']) {
            fclose($pipes[1]);
            fclose($pipes[2]);
            proc_close($process);
            return $status['exitcode'];
        }
        usleep(100000);
    }
    return 1;
}

 /**
  * @param WorkerBootstrapProfile $bootstrapProfile
  * @param string                 $implementationExpression
  */
 protected function __construct(WorkerBootstrapProfile $bootstrapProfile, $implementationExpression)
 {
     $bootstrapProfile->getOrFindPhpExecutablePathAndArguments($php, $phpArgs);
     $bootstrapProfile->compileScriptWithExpression($implementationExpression, null, $scriptPath, $deleteScript);
     try {
         $line = array_merge([$php], $phpArgs, [$scriptPath]);
         $outPath = $bootstrapProfile->getOutputPath();
         $this->process = proc_open(implode(' ', array_map('escapeshellarg', $line)), [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => $outPath !== null ? ['file', $outPath, 'a'] : STDERR], $pipes);
         $inputSink = Sink::fromStream($pipes[0], true);
         $outputSource = Source::fromStream($pipes[1], true);
         $this->channel = $bootstrapProfile->getChannelFactory()->createChannel($outputSource, $inputSink);
     } catch (\Exception $e) {
         if (isset($pipes[1])) {
             fclose($pipes[1]);
         }
         if (isset($pipes[0])) {
             fclose($pipes[0]);
         }
         if (isset($this->process)) {
             proc_terminate($this->process);
             proc_close($this->process);
         }
         if ($deleteScript) {
             unlink($scriptPath);
         }
         throw $e;
     }
 }

 public function run($argv)
 {
     // fetch task info
     $taskId = $argv[0];
     $task = $this->taskDao->find($taskId);
     if ($task === false) {
         $this->logger->error('Task not found.');
         return;
     }
     // TODO update pid
     $jobId = $task['job_id'];
     $developTaskId = $this->jobDao->getDevelopTaskId($task['job_id']);
     // open process
     $taskLogsDir = Config::get('taskLogsDir');
     $desc = [1 => ['file', $taskLogsDir . '/scheduler_task_' . $taskId . '.out', 'w'], 2 => ['file', $taskLogsDir . '/scheduler_task_' . $taskId . '.err', 'w']];
     $process = proc_open('php ' . __DIR__ . '/launcher.php develop_run.php ' . $developTaskId, $desc, $pipes);
     $this->logger->info("Run [task_id: {$taskId}, job_id: {$jobId}, develop_task_id: {$developTaskId}]");
     // wait
     while (!$this->taskDao->isInterrupted($taskId)) {
         $processStatus = proc_get_status($process);
         if (!$processStatus['running']) {
             if ($processStatus['exitcode'] == 0) {
                 $taskStatus = SchedulerTaskDao::STATUS_SUCCESS;
             } else {
                 $taskStatus = SchedulerTaskDao::STATUS_FAILURE;
             }
             $this->taskDao->updateStatus($taskId, $taskStatus);
             $this->jobDao->updateTaskStatus($jobId, $taskStatus);
             if ($taskStatus == SchedulerTaskDao::STATUS_SUCCESS) {
                 $this->jobDao->insertSignal($jobId, $this->now->format('Y-m-d'));
             }
             $this->logger->info('Task finished, exitcode ' . $processStatus['exitcode']);
             proc_close($process);
             return;
         }
         sleep(1);
     }
     $this->logger->warn('Task is interrupted, send SIGTERM.');
     proc_terminate($process, SIGTERM);
     // stop gracefully
     $countDown = 5;
     while ($countDown > 0) {
         $processStatus = proc_get_status($process);
         if (!$processStatus['running']) {
             break;
         }
         sleep(1);
         --$countDown;
     }
     if ($countDown == 0) {
         $this->logger->warn('Send SIGKILL.');
         proc_terminate($process, SIGKILL);
     }
     $this->taskDao->updateStatus($taskId, SchedulerTaskDao::STATUS_FAILURE);
     $this->jobDao->updateTaskStatus($jobId, SchedulerTaskDao::STATUS_FAILURE);
     $this->logger->info('Task killed.');
     proc_close($process);
 }

 public function __destruct()
 {
     $status = proc_terminate($this->_server);
     if ($status) {
         echo "local server stopped\n";
     } else {
         echo "stop local server failed";
     }
 }

function proc_terminate($signal)
{
    global $mockProcTerminateFail;
    if ($mockProcTerminateFail) {
        return false;
    } else {
        return \proc_terminate($signal);
    }
}

 /**
  * @return DriverService
  */
 public function stop()
 {
     if ($this->process === null) {
         return $this;
     }
     proc_terminate($this->process);
     $this->process = null;
     $checker = new URLChecker();
     $checker->waitUntilUnAvailable(3 * 1000, $this->url . '/shutdown');
     return $this;
 }

 public function shutdown()
 {
     pclose($this->config['proc-server']);
     pclose($this->config['proc-docs-server']);
     pclose($this->config['proc-watcher']);
     pclose($this->config['proc-logs']);
     proc_terminate($this->config['proc-server']);
     proc_terminate($this->config['proc-docs-server']);
     proc_terminate($this->config['proc-watcher']);
     proc_terminate($this->config['proc-logs']);
     exit;
 }

function handleShutdown()
{
    global $webSocketProcess;
    if (PHP_OS == "Linux") {
        proc_terminate($webSocketProcess, 9);
    }
    $error = error_get_last();
    if (!empty($error)) {
        $info = "[SHUTDOWN] date: " . date("d.m.y H:m", time()) . " file: " . $error['file'] . " | ln: " . $error['line'] . " | msg: " . $error['message'] . PHP_EOL;
        file_put_contents(APP_ROOT . 'logs' . DIRECTORY_SEPARATOR . 'error.log', $info, FILE_APPEND);
    }
}

 protected function stopProcessWithSignal($process, $signal, $callback)
 {
     list($process, $pipes) = $process;
     proc_terminate($process, $signal);
     usleep(100000);
     // Wait until the signal will be dispatched by the supervisor process
     $stdout = stream_get_contents($pipes[1]);
     $stderr = stream_get_contents($pipes[2]);
     fclose($pipes[1]);
     fclose($pipes[2]);
     proc_close($process);
     $callback($stdout, $stderr);
 }

 protected function call_maxima($command)
 {
     $ret = false;
     $err = '';
     $cwd = null;
     $env = array('why' => 'itworks');
     $descriptors = array(0 => array('pipe', 'r'), 1 => array('pipe', 'w'), 2 => array('pipe', 'w'));
     $casprocess = proc_open($this->command, $descriptors, $pipes, $cwd, $env);
     if (!is_resource($casprocess)) {
         throw new stack_exception('stack_cas_connection: could not open a CAS process');
     }
     if (!fwrite($pipes[0], $this->initcommand)) {
         throw new stack_exception('stack_cas_connection: could not write to the CAS process.');
     }
     fwrite($pipes[0], $command);
     fwrite($pipes[0], 'quit();' . "\n\n");
     $ret = '';
     // Read output from stdout.
     $starttime = microtime(true);
     $continue = true;
     if (!stream_set_blocking($pipes[1], false)) {
         $this->debug->log('', 'Warning: could not stream_set_blocking to be FALSE on the CAS process.');
     }
     while ($continue and !feof($pipes[1])) {
         $now = microtime(true);
         if ($now - $starttime > $this->timeout) {
             $procarray = proc_get_status($casprocess);
             if ($procarray['running']) {
                 proc_terminate($casprocess);
             }
             $continue = false;
         } else {
             $out = fread($pipes[1], 1024);
             if ('' == $out) {
                 // Pause.
                 usleep(1000);
             }
             $ret .= $out;
         }
     }
     if ($continue) {
         fclose($pipes[0]);
         fclose($pipes[1]);
         $this->debug->log('Timings', "Start: {$starttime}, End: {$now}, Taken = " . ($now - $starttime));
     } else {
         // Add sufficient closing ]'s to allow something to be un-parsed from the CAS.
         // WARNING: the string 'The CAS timed out' is used by the cache to search for a timeout occurrence.
         $ret .= ' The CAS timed out. ] ] ] ]';
     }
     return $ret;
 }

 public function exec_timeout($cmd, $timeout, &$output = '')
 {
     $fdSpec = [0 => ['file', '/dev/null', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w', 'a']];
     $pipes = [];
     $proc = proc_open($cmd, $fdSpec, $pipes);
     //        $status = proc_get_status($proc);
     //        var_dump($status);
     stream_set_blocking($pipes[1], false);
     $stop = time() + $timeout;
     while (true === true) {
         $in = [$pipes[1], $pipes[2]];
         $out = [];
         $err = [];
         //stream_select($in, $out, $err, min(1, $stop - time()));
         stream_select($in, $out, $err, 0, 200);
         if (count($in) !== 0) {
             foreach ($in as $socketToRead) {
                 while (feof($socketToRead) !== false) {
                     $output .= stream_get_contents($socketToRead);
                     continue;
                 }
             }
         }
         if ($stop <= time()) {
             break;
         } else {
             if ($this->isLockFileStillValid() === false) {
                 break;
             }
         }
     }
     fclose($pipes[1]);
     //close process's stdout, since we're done with it
     fclose($pipes[2]);
     //close process's stderr, since we're done with it
     //var_dump($output);
     $status = proc_get_status($proc);
     if (intval($status['running']) !== 0) {
         proc_terminate($proc);
         //terminate, since close will block until the process exits itself
         //This is the child process - so just exit
         exit(0);
         return -1;
     } else {
         proc_close($proc);
         //This is the child process - so just exit
         exit(0);
         return $status['exitcode'];
     }
 }

 /**
  * Executes the given test file, this time no other checking/setup is
  * involved, just run it and get the result.
  * 
  * This is the main bread and butter of the whole test framework. If theres
  * anything that needs to be polished or improved should be this method. 
  *
  * @param  string  $commands to execute the test file
  * @param  mixed   environment variable needed by the test file before firing
  *                 the $commands
  * @param  string  standard input
  * @return mixed   Returns data output after the $commands is invoked
  * 
  * @access public
  */
 public function console($cmd, $env = null, $stdin = null)
 {
     if (!empty($cmd)) {
         $data = "";
         $env = $env == null ? $_ENV : $env;
         $cwd = ($c = getcwd()) != false ? $c : null;
         $pipes = array();
         $descriptor = array(0 => array('pipe', 'r'), 1 => array('pipe', 'w'), 2 => array('pipe', 'w'));
         $proc = proc_open($cmd, $descriptor, $pipes, $cwd, $env, array("suppress_errors" => true));
         if ($proc && is_resource($proc)) {
             if (is_string($stdin)) {
                 fwrite($pipes[0], $stdin);
             }
             fclose($pipes[0]);
             while (true) {
                 /* hide errors from interrupted syscalls */
                 $r = $pipes;
                 $w = null;
                 $e = null;
                 $n = @stream_select($r, $w, $e, 300);
                 if ($n === 0) {
                     /* timed out */
                     $data .= "\n ** ERROR: process timed out **\n";
                     proc_terminate($proc);
                     return $data;
                 } else {
                     if ($n > 0) {
                         $line = fread($pipes[1], 8192);
                         if (strlen($line) == 0) {
                             /* EOF */
                             break;
                         }
                         $data .= $line;
                     }
                 }
             }
             # TODO: add more handler for $stat return value
             $stat = proc_get_status($proc);
             if ($stat['signaled']) {
                 $data .= "\nTermsig=" . $stat['stopsig'];
             }
             # TODO: implement (proc_close($proc) >> 8) & 0xff;
             $code = proc_close($proc);
             return $data;
         }
         /* close pipe */
         fclose($pipes[0]);
     }
     return false;
 }