<?php
/**
* SeekQuarry/Yioop --
* Open Source Pure PHP Search Engine, Crawler, and Indexer
*
* Copyright (C) 2009 - 2026 Chris Pollett chris@pollett.org
*
* LICENSE:
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*
* END LICENSE
*
* @author Chris Pollett chris@pollett.org
* @license https://www.gnu.org/licenses/ GPL3
* @link https://www.seekquarry.com/
* @copyright 2009 - 2026
* @filesource
*/
namespace seekquarry\yioop\library\data_structures;
use seekquarry\yioop\library as L;
/**
* UndoLog lets a step that changes several files of an index be undone
* when it did not finish, as when a crawl is stopped or a process is killed
* part way through. Before the step changes a file, it records how to put
* the file back: the length of a file it appends to, a copy of a file it
* rewrites, or that a file did not exist yet. The log is synced before the
* step goes on. A step that finishes appends a record saying so. undo,
* run when an index is opened, puts back every file of each step that did
* not finish, and returns the names of those steps.
*
* @author Chris Pollett
*/
class UndoLog
{
/**
* Name of the log file kept in the folder the log guards
*/
const LOG_NAME = "undo_log.txt";
/**
* Name of the folder, beside the log, that holds copies of files a
* step rewrites
*/
const COPIES_NAME = "undo_copies";
/**
* Name of the checkpoint file, beside the log, that holds for each file
* a finished step wrote the length known to be on the disk
*/
const CHECKPOINT_NAME = "undo_checkpoint.txt";
/**
* Folder whose files this log guards; the log is kept in it
* @var string
*/
public $folder;
/**
* What the last undo found on comparing files with the checkpoint, a
* line for each file it cut back or found short, for the caller to log
* @var array
*/
public $checkpoint_notes = [];
/**
* Makes an undo log for the files of one folder, such as an index
* bundle's.
*
* @param string $folder folder whose files the log guards
*/
public function __construct($folder)
{
$this->folder = $folder;
}
/**
* begin records, before a step changes anything, how to put back each
* file it will change, and syncs the record to disk. A file that does
* not exist yet is recorded as new, and is removed on undo. A file in
* $appended_files is recorded with its length; on undo it is cut back
* to it. A file in $rewritten_files is copied aside; on undo the copy
* is put back.
*
* @param string $step name of the step, used in the undo's log line
* @param array $appended_files paths of files the step only appends to
* @param array $rewritten_files paths of files, or folders, the step
* writes anew; a folder is recorded only if the step makes it
* @return string the id of the step, to pass to finish
*/
public function begin($step, $appended_files = [],
$rewritten_files = [])
{
$step_id = getmypid() . "-" . hrtime(true);
$lines = "BEGIN\t$step_id\t" . str_replace(["\t", "\n"], " ", $step) .
"\n";
foreach ($appended_files as $path) {
clearstatcache(true, $path);
$lines .= file_exists($path) ?
"LENGTH\t$step_id\t$path\t" . filesize($path) . "\n" :
"NEW\t$step_id\t$path\n";
}
$copies = $this->folder . "/" . self::COPIES_NAME;
foreach ($rewritten_files as $number => $path) {
clearstatcache(true, $path);
if (!file_exists($path)) {
$lines .= "NEW\t$step_id\t$path\n";
continue;
}
if (is_dir($path)) {
/* A folder that is already there is left as it is; only
one the step makes is removed on undo. */
continue;
}
if (!file_exists($copies)) {
@mkdir($copies);
}
$copy = "$copies/$step_id-$number";
copy($path, $copy);
self::syncPath($copy);
self::syncPath($copies);
$lines .= "COPY\t$step_id\t$path\t$copy\n";
}
$this->append($lines);
return $step_id;
}
/**
* commit marks the point past which a step is completed rather than
* undone: its new files are in place, and what remains is to remove
* old ones, listed in $remove_paths. If the step stops after this, the
* next undo removes those paths, finishing the step, instead of
* putting its files back. A tier merge commits once its merged tier is
* in place, before it removes the tiers it merged.
*
* @param string $step_id the id begin returned for the step
* @param array $remove_paths files or folders the step will remove
*/
public function commit($step_id, $remove_paths)
{
$lines = "COMMIT\t$step_id\n";
foreach ($remove_paths as $path) {
$lines .= "REMOVE\t$step_id\t$path\n";
}
$this->append($lines);
}
/**
* finish records that a step has written all its files, so undo leaves
* them as they are. When every step in the log has finished, the log
* and its copies are removed.
*
* @param string $step_id the id begin returned for the step
* @param array $written_paths files and folders the step wrote, which
* are synced to the disk and checkpointed before it is marked
* finished
* @param array $checkpoint_paths the files and folders among
* $written_paths whose lengths the checkpoint keeps, or null for
* all of them. A file written once and never grown, such as a
* dictionary tier's, has no use for a kept length.
*/
public function finish($step_id, $written_paths = [],
$checkpoint_paths = null)
{
/* Everything the step wrote reaches the disk, with the folders
that hold it, before the log says the step finished; then the
checkpoint records how long each file is. */
self::syncPaths($written_paths);
$lengths = [];
foreach ($checkpoint_paths ?? $written_paths as $path) {
foreach (self::filesUnder($path) as $file) {
clearstatcache(true, $file);
$lengths[$file] = filesize($file);
}
}
/* A step that checkpoints nothing leaves the checkpoint alone,
which saves reading every file it names. */
if ($checkpoint_paths !== []) {
$this->recordCheckpoint($lengths);
}
$this->append("FINISH\t$step_id\n");
$steps = $this->readSteps();
foreach ($steps as $record) {
if (!$record["FINISHED"]) {
return;
}
}
$this->clear();
}
/**
* undo puts back the files of every step in the log that did not
* finish and whose process has stopped: an appended file is cut back
* to its old length, a rewritten file is copied back, and a new file
* or folder is removed. A step that had committed is completed instead,
* by removing what it was to remove. A step whose process still runs
* is left in the log. An index
* calls it when it is opened, before it reads any file the log guards.
*
* @return array names of the steps that were undone
*/
public function undo()
{
$steps = $this->readSteps();
$running = false;
foreach ($steps as $step_id => $record) {
$holder = intval(explode("-", $step_id)[0]);
$running = $running || (!$record["FINISHED"] && $holder > 0 &&
($holder == getmypid() ||
L\CrawlDaemon::isProcessAlive($holder)));
}
/* A file may grow past its checkpoint only while a step that
writes it runs, so files are cut back only when none runs. */
$this->checkpoint_notes = $running ? [] : $this->cutToCheckpoint();
$undone = [];
$still_running = "";
foreach ($steps as $step_id => $record) {
if ($record["FINISHED"]) {
continue;
}
/* A step whose process still runs, this one included, is in
the middle of its work, not stopped part way, so its record
is kept. */
$holder = intval(explode("-", $step_id)[0]);
if ($holder > 0 && ($holder == getmypid() ||
L\CrawlDaemon::isProcessAlive($holder))) {
$still_running .= $record["LINES"];
continue;
}
if ($record["COMMITTED"]) {
foreach ($record["FILES"] as [$kind, $path]) {
if ($kind == "REMOVE") {
$this->removePath($path);
}
}
$undone[] = $record["STEP"] . " (completed)";
continue;
}
foreach (array_reverse($record["FILES"]) as $file) {
[$kind, $path, $detail] = $file;
if ($kind == "NEW") {
$this->removePath($path);
} else if ($kind == "LENGTH" && file_exists($path)) {
$handle = fopen($path, "r+");
ftruncate($handle, intval($detail));
fclose($handle);
} else if ($kind == "COPY" && file_exists($detail)) {
copy($detail, $path);
}
}
$undone[] = $record["STEP"];
}
$this->clear();
if ($still_running != "") {
$this->append($still_running);
}
return $undone;
}
/**
* Reads the log into its steps, each with its name, whether it
* finished, its files as [kind, path, length or copy], and its lines
* as written. A last line
* cut short by a crash is ignored; the step it belongs to did not get
* past writing its log, so it changed no file.
*
* @return array steps keyed by step id, in the order they began
*/
protected function readSteps()
{
$path = $this->folder . "/" . self::LOG_NAME;
$steps = [];
if (!file_exists($path)) {
return $steps;
}
$text = file_get_contents($path);
$end = strrpos($text, "\n");
$lines = ($end === false) ? [] :
explode("\n", substr($text, 0, $end));
foreach ($lines as $line) {
$fields = explode("\t", $line);
[$kind, $step_id] = [$fields[0], $fields[1] ?? ""];
if ($kind == "BEGIN") {
$steps[$step_id] = ["STEP" => $fields[2] ?? "",
"FINISHED" => false, "COMMITTED" => false,
"FILES" => [], "LINES" => ""];
} else if ($kind == "FINISH" && isset($steps[$step_id])) {
$steps[$step_id]["FINISHED"] = true;
} else if ($kind == "COMMIT" && isset($steps[$step_id])) {
$steps[$step_id]["COMMITTED"] = true;
} else if (isset($steps[$step_id])) {
$steps[$step_id]["FILES"][] = [$kind, $fields[2] ?? "",
$fields[3] ?? ""];
}
if (isset($steps[$step_id])) {
$steps[$step_id]["LINES"] .= "$line\n";
}
}
return $steps;
}
/**
* Appends lines to the log and syncs them to disk before returning, so
* a step never changes a file its log does not yet record.
*
* @param string $lines the records to add, each ending in a newline
*/
protected function append($lines)
{
$handle = fopen($this->folder . "/" . self::LOG_NAME, "a");
flock($handle, LOCK_EX);
fwrite($handle, $lines);
fsync($handle);
flock($handle, LOCK_UN);
fclose($handle);
}
/**
* syncPath asks the operating system to write a file, or a folder's
* list of names, to the disk before returning, with fsync, which on
* Windows is the C runtime's _commit. A folder is synced so a rename
* or a new file in it lasts through a power cut; PHP cannot open a
* folder on Windows, where NTFS keeps names in its own journal, so
* there a folder is left as it is.
*
* @param string $path the file or folder
*/
public static function syncPath($path)
{
$handle = @fopen($path, is_dir($path) ? "r" : "r+");
if ($handle === false) {
return;
}
@fsync($handle);
fclose($handle);
}
/**
* syncPaths makes every file under $paths, and the folders holding
* them, reach the disk before returning. UndoLog::finish and
* LSMTree::mergeTier call it for the files a step or a tier merge
* wrote. Each file is synced with syncPath, then each folder holding
* them once, using PHP's own fsync and starting no other program.
*
* @param array $paths files and folders whose files, and the folders
* holding them, are synced
*/
public static function syncPaths($paths)
{
if (empty($paths)) {
return;
}
$folders = [];
foreach ($paths as $path) {
foreach (self::filesUnder($path) as $file) {
self::syncPath($file);
}
$folders[is_dir($path) ? $path : dirname($path)] = true;
}
/* Each folder is synced once, however many of the files it
holds. */
foreach (array_keys($folders) as $folder) {
self::syncPath($folder);
}
}
/**
* filesUnder lists $path itself when it is a file, or every file in
* it, at any depth, when it is a folder.
*
* @param string $path a file or folder
* @return array the files
*/
public static function filesUnder($path)
{
if (!is_dir($path)) {
return file_exists($path) ? [$path] : [];
}
$files = [];
$entries = new \RecursiveIteratorIterator(
new \RecursiveDirectoryIterator($path,
\FilesystemIterator::SKIP_DOTS));
foreach ($entries as $entry) {
if ($entry->isFile()) {
$files[] = $entry->getPathname();
}
}
return $files;
}
/**
* recordCheckpoint adds the lengths of files a step has synced to the
* checkpoint file, keeping each file's latest length and dropping
* files no longer there. The file is written to a new name, synced and
* renamed into place, so the old checkpoint stands until the new one
* is whole on the disk. finish calls it. A process that has removed
* files the checkpoint names, as a tier merge in the dictionary
* removes the tiers it merged, calls it with no lengths, so an undo
* does not later report those files as lost.
*
* @param array $lengths file path => length known to be on the disk
*/
public function recordCheckpoint($lengths)
{
$checkpoint = $this->folder . "/" . self::CHECKPOINT_NAME;
$known = $this->readCheckpoint();
if (empty($known) && empty($lengths)) {
return;
}
foreach ($lengths as $file => $length) {
$known[$file] = $length;
}
$text = "";
$dropped = false;
foreach ($known as $file => $length) {
if (file_exists($file)) {
$text .= "$file\t$length\n";
} else {
$dropped = true;
}
}
if (empty($lengths) && !$dropped) {
return;
}
$writing = "$checkpoint.writing-" . getmypid();
file_put_contents($writing, $text);
self::syncPath($writing);
rename($writing, $checkpoint);
self::syncPath($this->folder);
}
/**
* Reads the checkpoint file into file path => length known to be on the
* disk. A last line cut short is ignored.
*
* @return array the lengths, or none when there is no checkpoint
*/
protected function readCheckpoint()
{
$checkpoint = $this->folder . "/" . self::CHECKPOINT_NAME;
$lengths = [];
if (!file_exists($checkpoint)) {
return $lengths;
}
$text = file_get_contents($checkpoint);
$end = strrpos($text, "\n");
foreach (($end === false) ? [] : explode("\n", substr($text, 0,
$end)) as $line) {
$fields = explode("\t", $line);
if (count($fields) == 2) {
$lengths[$fields[0]] = intval($fields[1]);
}
}
return $lengths;
}
/**
* cutToCheckpoint compares each checkpointed file with the length the
* checkpoint says reached the disk. A file longer than that holds bytes
* a stopped step wrote after it, and is cut back to it. A file shorter
* than that, or gone, lost bytes the disk was asked to keep, as a drive
* that holds writes in its own cache can on a power cut; it cannot be
* mended here, so it is named, for the caller to log. undo calls it
* first.
*
* @return array a line for each file cut back or found short
*/
protected function cutToCheckpoint()
{
$notes = [];
foreach ($this->readCheckpoint() as $file => $length) {
clearstatcache(true, $file);
$size = file_exists($file) ? filesize($file) : -1;
if ($size > $length) {
$handle = fopen($file, "r+");
ftruncate($handle, $length);
fsync($handle);
fclose($handle);
$notes[] = "$file cut back from $size to $length bytes";
} else if ($size < $length) {
$notes[] = "$file is " . (($size < 0) ? "gone" :
"$size bytes") . ", not the $length checkpointed";
}
}
return $notes;
}
/**
* Removes a file, or a folder with everything in it, as undo does for
* a file or folder a step made and for what a committed step was to
* remove.
*
* @param string $path the file or folder to remove
*/
protected function removePath($path)
{
if (!is_dir($path)) {
@unlink($path);
return;
}
$entries = new \RecursiveIteratorIterator(
new \RecursiveDirectoryIterator($path,
\FilesystemIterator::SKIP_DOTS),
\RecursiveIteratorIterator::CHILD_FIRST);
foreach ($entries as $entry) {
$entry->isDir() ? @rmdir($entry->getPathname()) :
@unlink($entry->getPathname());
}
@rmdir($path);
}
/**
* Removes the log and the copies of old files it kept.
*/
protected function clear()
{
@unlink($this->folder . "/" . self::LOG_NAME);
$copies = $this->folder . "/" . self::COPIES_NAME;
foreach (glob("$copies/*") ?: [] as $copy) {
@unlink($copy);
}
@rmdir($copies);
}
}
- Yioop Repo - Seekquarry