Merge branch 'master' into mcpe-1.0

src/pocketmine/PocketMine.php

@@ -496,16 +496,31 @@ namespace pocketmine {
 	$killer = new ServerKiller(8);
 	$killer->start();
 
+	$erroredThreads = 0;
+
 	foreach(ThreadManager::getInstance()->getAll() as $id => $thread){
 		$logger->debug("Stopping " . $thread->getThreadName() . " thread");
-		$thread->quit();
+		try{
+			$thread->quit();
+			$logger->debug($thread->getThreadName() . " thread stopped successfully.");
+		}catch(\ThreadException $e){
+			++$erroredThreads;
+			$logger->debug("Could not stop " . $thread->getThreadName() . " thread: " . $e->getMessage());
+		}
 	}
 
 	$logger->shutdown();
 	$logger->join();
 
-	echo Terminal::$FORMAT_RESET . "\n";
+	echo Terminal::$FORMAT_RESET . PHP_EOL;
 
-	exit(0);
+	if($erroredThreads > 0){
+		if(\pocketmine\DEBUG > 1){
+			echo "Some threads could not be stopped, performing a force-kill" . PHP_EOL . PHP_EOL;
+		}
+		kill(getmypid());
+	}else{
+		exit(0);
+	}
 
 }

src/pocketmine/command/CommandReader.php

@@ -19,21 +19,32 @@
  *
 */
 
+declare(strict_types = 1);
+
 namespace pocketmine\command;
 
 use pocketmine\Thread;
 
 class CommandReader extends Thread{
-	private $readline;
+
+	const TYPE_READLINE = 0;
+	const TYPE_STREAM = 1;
+	const TYPE_PIPED = 2;
+
 	/** @var \Threaded */
 	protected $buffer;
 	private $shutdown = false;
-	private $streamBlocking = false;
+	private $type = self::TYPE_STREAM;
+	private $lastLine = -1;
 
 	public function __construct(){
 		$this->buffer = new \Threaded;
 		$opts = getopt("", ["disable-readline"]);
-		$this->readline = (extension_loaded("readline") and !isset($opts["disable-readline"]));
+
+		if((extension_loaded("readline") and !isset($opts["disable-readline"]) and !$this->isPipe(STDIN))){
+			$this->type = self::TYPE_READLINE;
+		}
+
 		$this->start();
 	}
 
@@ -41,36 +52,105 @@ class CommandReader extends Thread{
 		$this->shutdown = true;
 	}
 
-	private function initStdin(){
+	public function quit(){
-		global $stdin;
+		$wait = microtime(true) + 0.5;
-		$stdin = fopen("php://stdin", "r");
+		while(microtime(true) < $wait){
-		$this->streamBlocking = (stream_set_blocking($stdin, 0) === false);
+			if($this->isRunning()){
+				usleep(100000);
+			}else{
+				parent::quit();
+				return;
+			}
+		}
+
+		$message = "Thread blocked for unknown reason";
+		if($this->type === self::TYPE_PIPED){
+			$message = "STDIN is being piped from another location and the pipe is blocked, cannot stop safely";
+		}
+
+		throw new \ThreadException($message);
 	}
 
-	private function readLine(){
+	private function initStdin(){
-		if(!$this->readline){
+		global $stdin;
+
+		if(is_resource($stdin)){
+			fclose($stdin);
+		}
+
+		$stdin = fopen("php://stdin", "r");
+		if($this->isPipe($stdin)){
+			$this->type = self::TYPE_PIPED;
+		}else{
+			$this->type = self::TYPE_STREAM;
+		}
+	}
+
+	/**
+	 * Checks if the specified stream is a FIFO pipe.
+	 *
+	 * @return bool
+	 */
+	private function isPipe($stream) : bool{
+		return is_resource($stream) and ((function_exists("posix_isatty") and !posix_isatty($stream)) or ((fstat($stream)["mode"] & 0170000) === 0010000));
+	}
+
+	/**
+	 * Reads a line from the console and adds it to the buffer. This method may block the thread.
+	 *
+	 * @return bool if the main execution should continue reading lines
+	 */
+	private function readLine() : bool{
+		$line = "";
+		if($this->type === self::TYPE_READLINE){
+			$line = trim(readline("> "));
+			if($line !== ""){
+				readline_add_history($line);
+			}else{
+				return true;
+			}
+		}else{
 			global $stdin;
 
 			if(!is_resource($stdin)){
 				$this->initStdin();
 			}
 
-			$line = fgets($stdin);
+			switch($this->type){
+				case self::TYPE_STREAM:
+					$r = [$stdin];
+					if(($count = stream_select($r, $w, $e, 0, 200000)) === 0){ //nothing changed in 200000 microseconds
+						return true;
+					}elseif($count === false){ //stream error
+						$this->initStdin();
+					}
 
-			if($line === false and $this->streamBlocking === true){ //windows sucks
+					if(($raw = fgets($stdin)) !== false){
-				$this->initStdin();
+						$line = trim($raw);
-				$line = fgets($stdin);
+					}else{
+						return false; //user pressed ctrl+c?
+					}
+
+					break;
+				case self::TYPE_PIPED:
+					if(($raw = fgets($stdin)) === false){ //broken pipe or EOF
+						$this->initStdin();
+						$this->synchronized(function(){
+							$this->wait(200000);
+						}); //prevent CPU waste if it's end of pipe
+						return true; //loop back round
+					}else{
+						$line = trim($raw);
+					}
+					break;
 			}
-			
-			return trim($line);
-		}else{
-			$line = trim(readline("> "));
-			if($line != ""){
-				readline_add_history($line);
-			}
-
-			return $line;
 		}
+
+		if($line !== ""){
+			$this->buffer[] = preg_replace("#\\x1b\\x5b([^\\x1b]*\\x7e|[\\x40-\\x50])#", "", $line);
+		}
+
+		return true;
 	}
 
 	/**
@@ -87,27 +167,17 @@ class CommandReader extends Thread{
 	}
 
 	public function run(){
-		if(!$this->readline){
+		if($this->type !== self::TYPE_READLINE){
 			$this->initStdin();
 		}
 
-		$lastLine = microtime(true);
+		while(!$this->shutdown and $this->readLine());
-		while(!$this->shutdown){
-			if(($line = $this->readLine()) !== ""){
-				$this->buffer[] = preg_replace("#\\x1b\\x5b([^\\x1b]*\\x7e|[\\x40-\\x50])#", "", $line);
-			}elseif(!$this->shutdown and (microtime(true) - $lastLine) <= 0.1){ //Non blocking! Sleep to save CPU
-				$this->synchronized(function(){
-					$this->wait(10000);
-				});
-			}
 
-			$lastLine = microtime(true);
+		if($this->type !== self::TYPE_READLINE){
-		}
-
-		if(!$this->readline){
 			global $stdin;
 			fclose($stdin);
 		}
+
 	}
 
 	public function getThreadName(){

src/spl

@@ -1 +1 @@
-Subproject commit 18e2f081d4803d014ab04f53d6a881791a24a384
+Subproject commit 23ac7b8b84a27232ac4af89927b63936d7c1928b