Fixed CommandReader hanging on shutdown, close #25 (#171)

Use stream_select to poll stdin status before reading
Add detection for FIFO pipes, rewrite half of the CommandReader (again)
Add timeout for CommandReader to prevent hang in Windows custom consoles (unknown reason)

src/pocketmine/PocketMine.php

@@ -2,11 +2,11 @@
 
 /*
  *
- *  ____            _        _   __  __ _                  __  __ ____  
- * |  _ \ ___   ___| | _____| |_|  \/  (_)_ __   ___      |  \/  |  _ \ 
+ *  ____            _        _   __  __ _                  __  __ ____
+ * |  _ \ ___   ___| | _____| |_|  \/  (_)_ __   ___      |  \/  |  _ \
  * | |_) / _ \ / __| |/ / _ \ __| |\/| | | '_ \ / _ \_____| |\/| | |_) |
- * |  __/ (_) | (__|   <  __/ |_| |  | | | | | |  __/_____| |  | |  __/ 
- * |_|   \___/ \___|_|\_\___|\__|_|  |_|_|_| |_|\___|     |_|  |_|_| 
+ * |  __/ (_) | (__|   <  __/ |_| |  | | | | | |  __/_____| |  | |  __/
+ * |_|   \___/ \___|_|\_\___|\__|_|  |_|_|_| |_|\___|     |_|  |_|_|
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Lesser General Public License as published by
@@ -15,7 +15,7 @@
  *
  * @author PocketMine Team
  * @link http://www.pocketmine.net/
- * 
+ *
  *
 */
 
@@ -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(){
-		global $stdin;
-		$stdin = fopen("php://stdin", "r");
-		$this->streamBlocking = (stream_set_blocking($stdin, 0) === false);
+	public function quit(){
+		$wait = microtime(true) + 0.5;
+		while(microtime(true) < $wait){
+			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(){
-		if(!$this->readline){
+	private function initStdin(){
+		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);
-			
-			if($line === false and $this->streamBlocking === true){ //windows sucks
-				$this->initStdin();
-				$line = fgets($stdin);
-			}
-			
-			return trim($line);
-		}else{
-			$line = trim(readline("> "));
-			if($line != ""){
-				readline_add_history($line);
-			}
+			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();
+					}
 
-			return $line;
+					if(($raw = fgets($stdin)) !== false){
+						$line = trim($raw);
+					}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;
+			}
 		}
+
+		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){
-			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);
-				});
-			}
+		while(!$this->shutdown and $this->readLine());
 
-			$lastLine = microtime(true);
-		}
-
-		if(!$this->readline){
+		if($this->type !== self::TYPE_READLINE){
 			global $stdin;
 			fclose($stdin);
 		}
+
 	}
 
 	public function getThreadName(){

src/spl

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