/**
  * Constructor
  *
  * @param   string command default NULL
  * @param   string[] arguments default []
  * @param   string cwd default NULL the working directory
  * @param   [:string] default NULL the environment
  * @throws  io.IOException in case the command could not be executed
  */
 public function __construct($command = NULL, $arguments = array(), $cwd = NULL, $env = NULL)
 {
     static $spec = array(0 => array('pipe', 'r'), 1 => array('pipe', 'w'), 2 => array('pipe', 'w'));
     // For `new self()` used in getProcessById()
     if (NULL === $command) {
         return;
     }
     // Check whether the given command is executable.
     $binary = self::resolve($command);
     if (!is_file($binary) || !is_executable($binary)) {
         throw new IOException('Command "' . $binary . '" is not an executable file');
     }
     // Open process
     $cmd = CommandLine::forName(PHP_OS)->compose($binary, $arguments);
     if (!is_resource($this->_proc = proc_open($cmd, $spec, $pipes, $cwd, $env, array('bypass_shell' => TRUE)))) {
         throw new IOException('Could not execute "' . $cmd . '"');
     }
     $this->status = proc_get_status($this->_proc);
     $this->status['exe'] = $binary;
     $this->status['arguments'] = $arguments;
     $this->status['owner'] = TRUE;
     // Assign in, out and err members
     $this->in = new File($pipes[0]);
     $this->out = new File($pipes[1]);
     $this->err = new File($pipes[2]);
 }

 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;
             }
         }
     }
 }

 private function compile_using_docker($code_text, &$return_code, &$msg)
 {
     if (!file_exists($this->_compile_workingdir)) {
         mkdir($this->_compile_workingdir, $recursive = true);
     }
     chdir($this->_compile_workingdir);
     file_put_contents($this->_compile_fn, $code_text);
     #var_dump('output compile file');
     $command = $this->_compile_docker_cmd;
     $descriptorspec = array(0 => array("pipe", "r"), 1 => array("pipe", "w"), 2 => array("pipe", "w"));
     $process = proc_open($command, $descriptorspec, $pipes, dirname(__FILE__), null);
     /*$r = exec($command, $output, $return_var);
       var_dump($r);
       var_dump($output);
       var_dump($return_var);*/
     $status = proc_get_status($process);
     while ($status["running"]) {
         sleep(1);
         $status = proc_get_status($process);
     }
     $return_code = $status['exitcode'];
     if ($return_code !== 0) {
         $msg = stream_get_contents($pipes[2]);
         fclose($pipes[2]);
     }
 }

 public function execute()
 {
     global $CFG;
     $connstr = '';
     switch ($CFG->dbtype) {
         case 'mysqli':
         case 'mariadb':
             $connstr = "mysql -h {$CFG->dbhost} -u {$CFG->dbuser} -p{$CFG->dbpass} {$CFG->dbname}";
             break;
         case 'pgsql':
             $portoption = '';
             if (!empty($CFG->dboptions['dbport'])) {
                 $portoption = '-p ' . $CFG->dboptions['dbport'];
             }
             putenv("PGPASSWORD={$CFG->dbpass}");
             $connstr = "psql -h {$CFG->dbhost} -U {$CFG->dbuser} {$portoption} {$CFG->dbname}";
             break;
         default:
             cli_error("Sorry, database type '{$CFG->dbtype}' is not supported yet.  Feel free to contribute!");
             break;
     }
     if ($this->verbose) {
         echo "Connecting to database using '{$connstr}'";
     }
     $process = proc_open($connstr, array(0 => STDIN, 1 => STDOUT, 2 => STDERR), $pipes);
     $proc_status = proc_get_status($process);
     $exit_code = proc_close($process);
     return $proc_status["running"] ? $exit_code : $proc_status["exitcode"];
 }

function waitToProcess($proc)
{
    $allowedFailures = 5;
    $r = proc_get_status($proc);
    for ($failures = 0; $failures < $allowedFailures; $failures++) {
        while ($r["running"]) {
            $r = proc_get_status($proc);
        }
        // Break if the process died normally.
        if ($r["termsig"] == 0) {
            break;
        }
        // If the process was killed. Fire it up again.
        disp("Processed Killed", 7);
        // Throw a high level warning.
        $descriptorspec = array(0 => array("file", "/dev/null", "r"), 1 => array("file", "/dev/null", "w"), 2 => array("file", "/dev/null", "a"));
        // Reopen the process
        $proc = proc_open($r["command"], $descriptorspec, $pipes);
        // Get the process status.
        $r = proc_get_status($proc);
    }
    // If we've failed the maximum number of times, give up.
    // There is probably something wrong with the image or it's too large to be converted and keeps getting killed.
    if ($failures == $allowedFailures) {
        disp("Process killed too many times. Try executing it via command line and see what the issue is.", 5);
        return 1;
    }
    return 0;
}

 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;
 }

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;
}

 public function status()
 {
     if (is_resource($this->_process)) {
         $this->_status = proc_get_status($this->_process);
     }
     return $this->_status;
 }

 /**
  * Runs a command and handles its output in real-time, line-by-line.
  * $callback is called for each line of output.
  * If $cwd is not provided, will use the working dir of the current PHP process.
  *
  * @param string   $cmd
  * @param callable $callback
  * @param string   $cwd
  *
  * @return array with two keys: exit_code and output.
  */
 public static function run($cmd, $callback = null, $cwd = null)
 {
     $pipes = [];
     $output = '';
     $descriptor_spec = [0 => ['pipe', 'r'], 1 => ['pipe', 'w']];
     $process = proc_open("{$cmd} 2>&1", $descriptor_spec, $pipes, $cwd);
     fclose($pipes[0]);
     stream_set_blocking($pipes[1], 0);
     $status = proc_get_status($process);
     while (true) {
         # Handle pipe output.
         $result = fgets($pipes[1]);
         $result = trim($result);
         while (!empty($result)) {
             $output .= $result;
             if ($callback) {
                 call_user_func($callback, trim($result));
             }
             $result = fgets($pipes[1]);
         }
         if (!$status['running']) {
             # Exit the loop.
             break;
         }
         $status = proc_get_status($process);
     }
     return ['exit_code' => $status['exitcode'], 'output' => $output];
 }

 public function poll()
 {
     for ($i = 0; $i < $this->n_workers; $i++) {
         // if a job is active in this slot then check if it is still running
         if ($this->pool[$i] !== FALSE) {
             $status = proc_get_status($this->pool[$i]['proc']);
             if ($status['running'] == FALSE) {
                 proc_close($this->pool[$i]['proc']);
                 call_user_func($this->job_cleanup, $this->pool[$i]['job']);
                 $this->pool[$i] = FALSE;
                 $this->n_running--;
             }
         }
         // start new workers when there are empty slots
         if ($this->pool[$i] === FALSE) {
             if (count($this->jobs) != 0) {
                 $job = array_shift($this->jobs);
                 $this->pool[$i]['job'] = $job;
                 $command = call_user_func($this->job_prepare, $job);
                 $this->pool[$i]['proc'] = proc_open($command, array(), $dummy);
                 $this->n_running++;
             }
         }
     }
     return $this->n_running;
 }

 /**
  * Run specific job script
  *
  * @param string
  */
 public function run($jobScript)
 {
     $this->jobStarted($jobScript);
     // -----
     $phpBin = 'php';
     $proc = proc_open($phpBin . ' ' . escapeshellarg($jobScript), array(array('pipe', 'r'), array('pipe', 'w'), array('pipe', 'w')), $pipes, dirname($jobScript), NULL, array('bypass_shell' => TRUE));
     list($stdin, $stdout, $stderr) = $pipes;
     fclose($stdin);
     do {
         $status = proc_get_status($proc);
     } while ($status['running']);
     // -----
     $result = $status['exitcode'] == 0;
     if ($result) {
         $this->jobFinished($jobScript, $stdout, $stderr);
     } else {
         $this->jobFailed($jobScript, $stdout, $stderr);
     }
     fclose($stdout);
     fclose($stderr);
     proc_close($proc);
     /// @todo error handling
     if ($result) {
         unlink($jobScript);
     } else {
         $dir = $this->storage->directory . '/failed';
         FileSystem::createDirIfNotExists($dir);
         rename($jobScript, $dir . '/' . basename($jobScript));
     }
     return $result;
 }

/**
 *等待wait_close_count个进程执行结束
 *
 */
function process_monitor(&$running_cmds, &$running_process, &$running_pipes, $wait_close_count, $logfile, $logfunc)
{
    $unfinished = count($running_process);
    if ($unfinished < $wait_close_count) {
        $wait_close_count = $unfinished;
    }
    $failed = 0;
    while ($wait_close_count > 0) {
        sleep(1);
        foreach ($running_cmds as $key => $each_cmd) {
            $status = proc_get_status($running_process[$key]);
            if (!$status['running']) {
                $remained = count($running_cmds);
                $errcode = $status['exitcode'];
                $isfinish = exec("grep \"{$key}\" {$logfile} | tail -n 1 | grep Finish");
                if ($errcode == 0 || $isfinish != "") {
                    $wait_close_count--;
                    $logfunc("{$key} finished, remain {$wait_close_count}\n");
                    unset($running_cmds[$key]);
                    break;
                } else {
                    $wait_close_count--;
                    $failed++;
                    $logfunc("{$key} error with unknow reason({$errcode}), remain {$wait_close_count}\n");
                    unset($running_cmds[$key]);
                    return false;
                    //break;
                }
            }
        }
    }
    return true;
}

 public function start()
 {
     if (GlobalState::$TYPE == 'LOCAL') {
         $this->register();
         $this->doInstructions($instruction = null);
         $this->tryStopDebugIdUpdater();
         $cmd = 'php -d xdebug.remote_autostart=0 ..' . DS . 'core' . DS . 'DebugIdUpdater.php ' . Config::$DEBUG_ID;
         // start background script for updating in redis expire of debugId
         if (Config::$CORE['os_type'] != "WIN") {
             Config::$DEBUG_PID = exec($cmd . ' > /dev/null 2>&1 & echo $!');
         } else {
             $descriptorspec = [0 => ["pipe", "r"], 1 => ["pipe", "w"]];
             $pipes = '';
             $proc = proc_open("start /B " . $cmd, $descriptorspec, $pipes);
             $info = proc_get_status($proc);
             Config::$DEBUG_PID = $info['pid'];
             //proc_close( $proc );
         }
         // put pid into file for try kill DebugIdUpdater.php it before next run CodeRunner
         file_put_contents(".run", Config::$DEBUG_PID);
         for (;;) {
             $this->message_processor->run();
             $this->responder_processor->localRun();
             stream_set_blocking(STDIN, false);
             $command = trim(fgets(STDIN));
             stream_set_blocking(STDIN, true);
             if ($command == 'terminate') {
                 $this->terminateRunner($signal = 'SIGINT');
             }
         }
     } else {
         $this->message_processor->run();
         $this->responder_processor->cloudRun();
     }
 }

 /**
  * close stream
  * @return mixed
  */
 public function close()
 {
     if (!is_resource($this->process)) {
         return;
     }
     $status = proc_get_status($this->process);
     if ($status['running'] == true) {
         //process ran too long, kill it
         //close all pipes that are still open
         @fclose($this->pipes[0]);
         //stdin
         @fclose($this->pipes[1]);
         //stdout
         @fclose($this->pipes[2]);
         //stderr
         //get the parent pid of the process we want to kill
         $parent_pid = $status['pid'];
         //use ps to get all the children of this process, and kill them
         $pids = preg_split('/\\s+/', `ps -o pid --no-heading --ppid {$parent_pid}`);
         foreach ($pids as $pid) {
             if (is_numeric($pid)) {
                 posix_kill($pid, 9);
                 //9 is the SIGKILL signal
             }
         }
         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;
}

 public static function tearDownAfterClass()
 {
     $s = proc_get_status(self::$phpserver);
     posix_kill($s['pid'], SIGKILL);
     proc_close(self::$phpserver);
     self::$phpserver = null;
 }

 function handle($msg, $params)
 {
     $args = trim($msg->args);
     if (empty($args)) {
         return new BotMsg('Funkcja <b>ort</b> wymaga argumentu.<br />' . '<br />' . "\n" . '<u>Przykłady:</u><br />' . "\n" . 'ort grzegżółka<br />' . "\n" . 'ort warsawa');
     }
     $args = strtr($args, array("\r\n" => ' ', "\r" => ' ', "\n" => ' '));
     $proc = proc_open('aspell --lang=pl --encoding=utf-8 --ignore-case=true pipe', array(array('pipe', 'r'), array('pipe', 'w'), array('file', '/dev/null', 'w')), $pipe);
     fwrite($pipe[0], $args . "\n");
     fclose($pipe[0]);
     do {
         usleep(1);
         $status = proc_get_status($proc);
     } while ($status['running']);
     fgets($pipe[1], 1024);
     $spell = fgets($pipe[1], 4096);
     fclose($pipe[1]);
     proc_close($proc);
     if (empty($spell)) {
         return new BotMsg('Błąd podczas sprawdzania słowa w słowniku. Przepraszamy.');
     } elseif (substr($spell, 0, 1) == '*') {
         return new BotMsg('<span style="color:#060;">Pisownia poprawna.</span>');
     } elseif (substr($spell, 0, 1) == '#') {
         return new BotMsg('Brak propozycji poprawnej pisowni.');
     } else {
         $spell = explode(': ', $spell, 2);
         $spell = explode(',', $spell[1]);
         $txt = '<p>Prawdopobnie chodziło ci o:</p>' . "\n" . '<ul>' . "\n";
         foreach ($spell as $val) {
             $txt .= '<li>' . htmlspecialchars(trim($val)) . '</li>' . "\n";
         }
         $txt .= '</ul>';
         return new BotMsg($txt);
     }
 }

function test_me($desc)
{
    $pipes = null;
    $process = proc_open(__DIR__ . "/test_proc_open.sh", $desc, $pipes);
    $status = proc_get_status($process);
    pcntl_waitpid($status["pid"], $child_status);
}

 /**
  * @param InputInterface $input
  * @param OutputInterface $output
  * @return int|null|void
  */
 protected function execute(InputInterface $input, OutputInterface $output)
 {
     parent::execute($input, $output);
     $output->writeln('Starting Chamilo SQL cli');
     $_configuration = $this->getConfigurationArray();
     $cmd = 'mysql -h ' . $_configuration['db_host'] . ' -u ' . $_configuration['db_user'] . ' -p' . $_configuration['db_password'] . ' ' . $_configuration['main_database'];
     $process = proc_open($cmd, array(0 => STDIN, 1 => STDOUT, 2 => STDERR), $pipes);
     $proc_status = proc_get_status($process);
     $exit_code = proc_close($process);
     return $proc_status["running"] ? $exit_code : $proc_status["exitcode"];
     /*$output->writeln('<comment>Starting Chamilo process</comment>');
       $output->writeln('<info>Chamilo process ended succesfully</info>');
       */
     /*
             $progress = $this->getHelperSet()->get('progress');
     
             $progress->start($output, 50);
             $i = 0;
             while ($i++ < 50) {
                 // ... do some work
     
                 // advance the progress bar 1 unit
                 $progress->advance();
             }
             $progress->finish();*/
     // Inside execute function
     //$output->getFormatter()->setStyle('fcbarcelona', new OutputFormatterStyle('red', 'blue', array('blink', 'bold', 'underscore')));
     //$output->writeln('<fcbarcelona>Messi for the win</fcbarcelona>');
 }

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;
}