Race condition v SafeStream?

IMHO problém je v té mezeře mezi flock(LOCK_UN) a rename(). V té době se k tomu starému nezamčenému souboru může dostat jiný proces, získat LOCK_EX a začít s ním pracovat. Následný rename() našeho procesu potom dle situace může proběhnout až poté, co ten jiný proces dokončil práci a provedl svůj rename(), čímž potenciálně starší verze souboru přepíše nověji vygenerovanou.

Řekl bych, že rename() proběhne hned, navíc na Windows selže.

Jasně, stát se to může, že závod vyhraje proces, který začal později.

Na Windows (resp. na NTFS) nejde mazat nebo přejmenovávat soubory, pokud jsou otevřené, na rozdíl od Linuxu, takže proto to selže.

No ale ten soubor se před tím uzavře, takže ani na windowsech by neměl selhat. Jako je možné, že na windowsech je celý ten postup korektní, to netuším. Hádám ale, že hodně lidí má produkční kód na linuxu a mít tam tiché a náhodně se projevující bugy asi není ideální…

To z toho ale pořád nedělá relevantní informaci.

Mně to nepřijde jako bug. Je to v kódu stream_close, takže při zavírání souboru. V tu chvíli už daný proces nemůže spoléhat na to, že je obsah konzistentní. To mohl, dokud byl soubor otevřený. Po zavření už je jedno, jaký proces vyhrál.

Použil jsem blbý slovo. Myslel jsem tím, že proces po zavření souboru nemůže spoléhat na obsah, který do něj sám zapsal.

Zkusím to jinak. @phpfm žádná race condition imho není, protože rename se spustí až po uzavření souboru a je jedno, jaký rename vyhraje.

OK, dáme si názornou ukázku. Pokud ani fopen("nette.safe://...", "a") ani následné zápisy nevyhodí chybu, měly by zapsaná data v souboru být. Následující kód spustí procesy, které současně do stejného souboru přidávají řádky.

	define('WORKERS', 2);
	define('MESSAGES', 10);
	define('FILE_PATH', 'test_SafeStream_' . rand() . '.txt');

	$children = [];
	for($i = 0; $i < WORKERS; ++$i) {
		$pid = pcntl_fork();
		if( $pid == -1 ) {
			echo "Fork failed, bailing out...\n";
			break;
		} elseif( $pid == 0 ) {
			// child

			// "seed" lcg_value() generator so SafeStream doesn't generate
			// the same .tmp filename (and then fails to open it)
			for($j = 0; $j < (WORKERS - $i) * MESSAGES; ++$j)
				lcg_value();

			for($j = 0; $j < MESSAGES; ++$j) {
				// open file for append, write message and close

				if( !$f = fopen('nette.safe://' . FILE_PATH, 'a') )
					die("Worker $i failed to open " . FILE_PATH . " for message $j\n");

				$msg =  "Worker $i, message $j\n";
				$r = fwrite($f, $msg);
				if( $r === false || $r != strlen($msg) )
					die("Worker $i failed to append message $j\n");

				if( !fclose($f) )
					die("Worker $i failed to close file properly\n");
			}

			// child proces ends here
			exit(0);
		} else {
			// main process

			$children[] = $pid;
		}
	}

	while( !empty($children) ) {
		$child_pid = pcntl_wait($status);

		$children = array_filter($children, fn($pid) => $pid != $child_pid);
	}

	echo "Expected " . (WORKERS * MESSAGES) . " messages, got " . substr_count(file_get_contents(FILE_PATH), "\n") . "\n";

	unlink(FILE_PATH);

Ten stream_close() asi nebyl nejlepší příklad, protože on je problém už při otvírání. Ve skutečnosti pokaždé, když na sebe narazí N procesů nad jednou cestou, tak z N verzí zapsaných souborů zůstane jen jedna náhodná, všechny ostatní se ztratí. Jde o to, že cesta je jen ukazatel na soubor s daty a fopen() vrátí ten ukazatel, který tam zrovna je. Když později rename() nebo unlink() + fopen(..., "c") změní, na jaká data cesta ukazuje, tak se o tom ti, kteří získali předchozí ukazatel, nedozví.

Jinými slovy proces P otveře cestu s daty S1, získá zámek a dá se do práce. Přijde dalších N procesů, otevřou cestu, získají ukazatel na ten samý S1 a zablokují se na zámku. Proces P dokončí práci, uvolní zámek nad S1 a provede rename(), takže nyní cesta ukazuje na soubor S2. Uvolnění zámku nad S1 probudilo jeden z čekajících procesů, který se dá do práce, ovšem nad původními daty S1. O tom, že cesta teď ukazuje na S2 od procesu P, neví. Pracující proces se dopracuje k výsledku S3, uvolní zámek nad S1 a udělá rename(), takže cesta ukazuje na jeho data S3 a S2 jsou fuč. Uvolnění zámku probudí další čekající proces, který ovšem opět drží ukazatel na S1. O S2 ani o S3 neví a dá se vesele do práce…

S tímhle názorná ukázka funguje o něco lépe.

class EvenSaferStream
{
	/** Name of stream protocol - nette.safe:// */
	public const PROTOCOL = 'nette.safe';

	private $path;

	private $work_path;
	private $work_fd;

	private $lock_path;
	private $lock_fd;

	private const OK = 0, Error = -1, Blocked = -2;


	public static function register(): bool
	{
		return stream_wrapper_register(self::PROTOCOL, self::class);
	}


	public function stream_open(string $path, string $mode, int $options): bool
	{
		$path = substr($path, strpos($path, ':') + 3);  // trim protocol nette.safe://

		if( $mode != 'a' ) // support only append
			return false;

		if( $this->lock($path) != self::OK )
			return false;

		$ok = false;
		try {
			$work_path = $path . '.tmp';

			if( file_exists($path) ) {
				if( !$fd = fopen($path, 'r+') )
					return false;

				$copy = stream_get_contents($fd);
				if( $copy === false )
					return false;

				if( file_put_contents($work_path, $copy) !== strlen($copy) )
					return false;

				fclose($fd);
			}

			if( !$work_fd = fopen($work_path, 'a') ) {
				@ unlink($work_path);
				return false;
			}

			$this->path = $path;
			$this->work_path = $work_path;
			$this->work_fd = $work_fd;

			$ok = true;

			return true;
		} finally {
			if( !$ok ) {
				$this->unlock();
			}
		}
	}

	private const Attempts = 10, TokenLength = 10;

	private function lock($path)
	{
		$lock_path = $path . '.lock';

		for($i=0; $i < self::Attempts; ++$i) {
			if( !$lock_fd = fopen($lock_path, 'c') ) {
				return self::Error;
			}

			if( !flock($lock_fd, LOCK_EX) ) {
				fclose($lock_fd);
				return self::Error;
			}

			if( !$stat = fstat($lock_fd) ) {
				fclose($lock_fd);
				return self::Error;
			}

			if( $stat['size'] > 0 ) {
				// op on this lock fd was successfully committed, but we need to
				// check if the lock file was also removed.
				if( file_exists($lock_path) ) {
					// Lock file is still there, so now we need to figure out if $lock_path points
					// to our $lock_fd or it's new (which means there is another process with lock() on this path).
					// We will do it by writing random challenge string into $lock_fd and see if
					// it appears in $lock_path file.
					$challenge = random_bytes(self::TokenLength);
					fwrite($lock_fd, $challenge);
					fflush($lock_fd);

					if( $challenge === @ file_get_contents($lock_path) )
						unlink($lock_path);
				}

				fclose($lock_fd);
				continue;
			}

			$this->lock_path = $lock_path;
			$this->lock_fd = $lock_fd;

			return self::OK;
		}

		return self::Blocked;
	}

	private function unlock()
	{
		fwrite($this->lock_fd, 'c'); // signal we released the file correctly
		fflush($this->lock_fd);
		unlink($this->lock_path);
		fclose($this->lock_fd); // release lock

		$this->lock_path = null;
		$this->lock_fd = null;
	}


	public function stream_close(): void
	{
		fclose($this->work_fd);
		rename($this->work_path, $this->path);

		$this->work_path = null;
		$this->work_fd = null;

		$this->unlock();
	}

	public function stream_read(int $length)
	{
		return fread($this->work_fd, $length);
	}

	public function stream_write(string $data)
	{
		return fwrite($this->work_fd, $data, strlen($data));
	}

	public function stream_truncate(int $size): bool
	{
		return ftruncate($this->work_fd, $size);
	}

	public function stream_tell(): int
	{
		return ftell($this->work_fd);
	}

	public function stream_eof(): bool
	{
		return feof($this->work_fd);
	}

	public function stream_seek(int $offset, int $whence = SEEK_SET): bool
	{
		return fseek($this->work_fd, $offset, $whence) === 0;
	}

	public function stream_stat()
	{
		return fstat($this->work_fd);
	}
}

(Všechny ukázky byly testovány jen a pouze pod linuxem.)

Editoval phpfm (2. 1. 2022 16:51)

@DavidGrudl Pokud by se pro append nepoužíval dočasný soubor, už by neplatilo „Atomicity: The file is written either as a whole or not written at all.“ z dokumentace.

To jsem myslel, že řeší stream_write(), kde se kontroluje délka zápisu.

https://github.com/…feStream.php#L199

Když ale dojde k chybě při zápisu, tak stream_close nikdy dočasný soubor nepřejmenuje na ostrý. S původním se tedy nic nestane.

To používání dočasných souborů jsem tam dával před 11 lety a nevím už přesně proč (tehdy bylo PHP 5.2, časté fatal errory, a fopen neuměl nebo neměl zdokumentovaný režim c, tedy vytvořit bez smazání). Zkusil jsem to revertnout https://github.com/…feStream.php

Ve stream_truncate() zbyl tempHandle, ale to jen pro info.

Aha. Já myslel, že ta vychytávka, je právě ten temp file. Že když se kdykoliv mezi otevřením a zavřením souboru něco podělá, tak původní soubor zůstane nedotčen, a že právě rename() vyvolaný při close zajistí celkovou konzistenci souboru.

Na atomicitu rezignuji, protože mi dochází, že ji nelze rozlišit na low-level úrovni. Ověřím úspěšnost fwrite(), ale jen aplikace ví, kolik těch fwrite() mělo proběhnout.

Tahle třída vznikla primárně kvůli izolaci, aby nezapisovalo více vláken zároveň nebo někdo nečetl nedopsaná data. Dočasné soubory řeší situaci, kdy někdo získá zámek ke čtení dřív než pro zápis a přečte prázdný soubor, ale přidávají mnohem víc komplikací, takže bych je asi opustil.

Tak pro úplnost verze nette.safe:// s atomicitou a serializací. Možno užívat, jak je libo. (Testováno na linuxu.)

<?php

class EvenSaferStream
{
	/** Name of stream protocol - nette.safe:// */
	public const PROTOCOL = 'nette.safe';

	// True if opened in r mode.
	private bool $readOnly;

	// Path to the file.
	private ?string $path;

	// Path & handle to working file that will be moved to $path at the end,
	// except in case of read only mode, where $workHandle points to opened
	// $path and $workPath is null.
	private ?string $workPath;
	private $workHandle;

	// Path & handle to lock, if opened in any other mode than read only.
	private ?string $lockPath;
	private $lockHandle;


	private static function getTemporaryPath($path)
	{
		$dir = dirname($path);
		return ($dir !== '' ? $dir . '/' : '') . '~~' . basename($path);
	}

	// Locking is done on separate $path.lock file, which preferably should be removed after
	// work is over. But it also should be there as long as possible to not disturb the FIFO
	// front of processes waiting on it for their turn. For that, before attempt to acquire
	// lock is made, one byte is appended to the opened lock file. This is done without
	// synchronization and some data may be lost, so after acquiring lock the process needs
	// to check $lock_path still points to the opened lock handle and it wasn't unlinked by
	// previous lock holder who thought there is no one else waiting.

	private const OK = 0, Error = -1, Blocked = -2;
	private const Attempts = 3;

	private function lock($path)
	{
		$lock_path = self::getTemporaryPath($path) . '.lock';

		for($i=0; $i < self::Attempts; ++$i) {
			if( !$lock_handle = fopen($lock_path, 'a') )
				return self::Error;

			// Append hint to lock file signaling there is someone waiting on it.
			fwrite($lock_handle, 'o');
			fflush($lock_handle);

			if( !flock($lock_handle, LOCK_EX) ) {
				fclose($lock_handle);
				return self::Error;
			}

			$lock_stat = fstat($lock_handle);

			// Decrement string of waiting hints.
			ftruncate($lock_handle, $lock_stat['size'] > 0 ? $lock_stat['size'] - 1 : 0);

			// Now check the lock we acquired is the same as under $lock_path. If inodes
			// are different (or $lock_path doesn't exists), it means someone else deleted
			// the lock and we have to throw it away and start again from current $lock_path.

			clearstatcache(true, $lock_path);

			$lock_path_ino = @ fileinode($lock_path);
			if( $lock_path_ino === false || $lock_path_ino != $lock_stat['ino'] ) {
				fclose($lock_handle);
				continue;
			}

			$this->lockPath = $lock_path;
			$this->lockHandle = $lock_handle;

			return self::OK;
		}

		return self::Blocked;
	}

	private function unlock()
	{
		$others_wait_on_lock = fstat($this->lockHandle)['size'] > 0;

		// This is rough approximation, it's possible there are others waiting on
		// the lock even if this says there are none. However when they wake up
		// they should detect that in lock() and run another attempt.
		if( !$others_wait_on_lock )
			unlink($this->lockPath);

		fclose($this->lockHandle); // release the lock

		$this->lockPath = null;
		$this->lockHandle = null;
	}


	public static function register(): bool
	{
		return stream_wrapper_register(self::PROTOCOL, self::class);
	}

	public function stream_open(string $path, string $mode, int $options): bool
	{
		$path = substr($path, strlen(self::PROTOCOL) + 3);  // trim protocol nette.safe://

		$tb_flag = ltrim($mode, 'acrxw+');
		$mode = rtrim($mode, 'tb');
		$use_include_path = (bool) (STREAM_USE_PATH & $options);

		if( $mode == 'r' ) {
			// for read only mode no need to lock anything, just use what is there
			// under $path
			if( !$fd = fopen($path, 'r' . $tb_flag, $use_include_path) )
				return false;

			$this->readOnly = true;
			$this->workHandle = $fd;

			return true;
		}

		$this->readOnly = false;

		if( $this->lock($path) != self::OK )
			return false;

		$ok = false;
		try {
			$work_path = self::getTemporaryPath($path) . '.tmp';

			if( $mode[0] != 'w' && $mode[0] != 'x' ) {
				// w & x mode truncate file, others don't => need to copy content
				// to working file
				clearstatcache(true, $path);

				// FIXME: this doesn't use $use_include_path
				if( file_exists($path) ) {
					if( !copy($path, $work_path) )
						return false;
				} elseif( $mode[0] == 'r' ) {
					return false;
				}
			} elseif( $mode[0] == 'x' ) {
				// make sure we are first to create $path
				if( !fopen($path, $mode . $tb_flag, $use_include_path) )
					return false;
			}

			if( !$work_handle = fopen($work_path, $mode . $tb_flag, $use_include_path) ) {
				@ unlink($work_path);
				return false;
			}

			$this->path = $path;
			$this->workPath = $work_path;
			$this->workHandle = $work_handle;

			$ok = true;

			return true;
		} finally {
			if( !$ok ) {
				$this->unlock();
			}
		}
	}

	public function stream_close(): void
	{
		fclose($this->workHandle);

		if( $this->readOnly )
			return;

		rename($this->workPath, $this->path);

		$this->workPath = null;
		$this->workHandle = null;

		$this->unlock();
	}

	public function stream_read(int $length)
	{
		return fread($this->workHandle, $length);
	}

	public function stream_write(string $data)
	{
		return fwrite($this->workHandle, $data, strlen($data));
	}

	public function stream_truncate(int $size): bool
	{
		return ftruncate($this->workHandle, $size);
	}

	public function stream_tell(): int
	{
		return ftell($this->workHandle);
	}

	public function stream_eof(): bool
	{
		return feof($this->workHandle);
	}

	public function stream_seek(int $offset, int $whence = SEEK_SET): bool
	{
		return fseek($this->workHandle, $offset, $whence) === 0;
	}

	public function stream_stat()
	{
		return fstat($this->workHandle);
	}
}

?>

Editoval phpfm (5. 1. 2022 23:31)