<commit_before>/*
 * This file is part of Bisq.
 *
 * Bisq is free software: you can redistribute it and/or modify it
 * under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or (at
 * your option) any later version.
 *
 * Bisq 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 Affero General Public
 * License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with Bisq. If not, see <http://www.gnu.org/licenses/>.
 */

package bisq.common.storage;

import bisq.common.UserThread;
import bisq.common.proto.persistable.PersistableEnvelope;
import bisq.common.proto.persistable.PersistenceProtoResolver;
import bisq.common.util.Utilities;

import com.google.common.util.concurrent.CycleDetectingLockFactory;

import java.nio.file.Paths;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.PrintWriter;

import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class FileManager<T extends PersistableEnvelope> {
    private final File dir;
    private final File storageFile;
    private final ScheduledThreadPoolExecutor executor;
    private final AtomicBoolean savePending;
    private final long delay;
    private final Callable<Void> saveFileTask;
    private T persistable;
    private final PersistenceProtoResolver persistenceProtoResolver;
    private final ReentrantLock writeLock = CycleDetectingLockFactory.newInstance(CycleDetectingLockFactory.Policies.THROW).newReentrantLock("writeLock");

    ///////////////////////////////////////////////////////////////////////////////////////////
    // Constructor
    ///////////////////////////////////////////////////////////////////////////////////////////

    public FileManager(File dir, File storageFile, long delay, PersistenceProtoResolver persistenceProtoResolver) {
        this.dir = dir;
        this.storageFile = storageFile;
        this.persistenceProtoResolver = persistenceProtoResolver;

        executor = Utilities.getScheduledThreadPoolExecutor("FileManager", 1, 10, 5);

        // File must only be accessed from the auto-save executor from now on, to avoid simultaneous access.
        savePending = new AtomicBoolean();
        this.delay = delay;

        saveFileTask = () -> {
            try {
                Thread.currentThread().setName("Save-file-task-" + new Random().nextInt(10000));
                // Runs in an auto save thread.
                // TODO: this looks like it could cause corrupt data as the savePending is unset before the actual
                // save. By moving to after the save there might be some persist operations that are not performed
                // and data would be lost. Probably all persist operations should happen sequencially rather than
                // skip one when there is already one scheduled
                if (!savePending.getAndSet(false)) {
                    // Some other scheduled request already beat us to it.
                    return null;
                }
                saveNowInternal(persistable);
            } catch (Throwable e) {
                log.error("Error during saveFileTask", e);
            }
            return null;
        };
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            UserThread.execute(FileManager.this::shutDown);
        }, "FileManager.ShutDownHook"));
    }


    ///////////////////////////////////////////////////////////////////////////////////////////
    // API
    ///////////////////////////////////////////////////////////////////////////////////////////

    /**
     * Actually write the wallet file to disk, using an atomic rename when possible. Runs on the current thread.
     */
    public void saveNow(T persistable) {
        saveNowInternal(persistable);
    }

    /**
     * Queues up a save in the background. Useful for not very important wallet changes.
     */
    public void saveLater(T persistable) {
        saveLater(persistable, delay);
    }

    public void saveLater(T persistable, long delayInMilli) {
        this.persistable = persistable;

        if (savePending.getAndSet(true))
            return;   // Already pending.

        executor.schedule(saveFileTask, delayInMilli, TimeUnit.MILLISECONDS);
    }

    @SuppressWarnings("unchecked")
    public synchronized T read(File file) {
        log.debug("Read from disc: {}", file.getName());

        try (final FileInputStream fileInputStream = new FileInputStream(file)) {
            protobuf.PersistableEnvelope persistable = protobuf.PersistableEnvelope.parseDelimitedFrom(fileInputStream);
            return (T) persistenceProtoResolver.fromProto(persistable);
        } catch (Throwable t) {
            String errorMsg = "Exception at proto read: " + t.getMessage() + " file:" + file.getAbsolutePath();
            log.error(errorMsg, t);
            //if(DevEnv.DEV_MODE)
            throw new RuntimeException(errorMsg);
        }
    }

    public synchronized void removeFile(String fileName) {
        File file = new File(dir, fileName);
        boolean result = file.delete();
        if (!result)
            log.warn("Could not delete file: " + file.toString());

        File backupDir = new File(Paths.get(dir.getAbsolutePath(), "backup").toString());
        if (backupDir.exists()) {
            File backupFile = new File(Paths.get(dir.getAbsolutePath(), "backup", fileName).toString());
            if (backupFile.exists()) {
                result = backupFile.delete();
                if (!result)
                    log.warn("Could not delete backupFile: " + file.toString());
            }
        }
    }


    /**
     * Shut down auto-saving.
     */
    void shutDown() {
        executor.shutdown();
        try {
            executor.awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static void removeAndBackupFile(File dbDir, File storageFile, String fileName, String backupFolderName)
            throws IOException {
        File corruptedBackupDir = new File(Paths.get(dbDir.getAbsolutePath(), backupFolderName).toString());
        if (!corruptedBackupDir.exists())
            if (!corruptedBackupDir.mkdir())
                log.warn("make dir failed");

        File corruptedFile = new File(Paths.get(dbDir.getAbsolutePath(), backupFolderName, fileName).toString());
        FileUtil.renameFile(storageFile, corruptedFile);
    }

    public synchronized void removeAndBackupFile(String fileName) throws IOException {
        removeAndBackupFile(dir, storageFile, fileName, "backup_of_corrupted_data");
    }

    public synchronized void backupFile(String fileName, int numMaxBackupFiles) {
        FileUtil.rollingBackup(dir, fileName, numMaxBackupFiles);
    }

    ///////////////////////////////////////////////////////////////////////////////////////////
    // Private
    ///////////////////////////////////////////////////////////////////////////////////////////

    private void saveNowInternal(T persistable) {
        long now = System.currentTimeMillis();
        saveToFile(persistable, dir, storageFile);
        log.debug("Save {} completed in {} msec", storageFile, System.currentTimeMillis() - now);
    }

    private synchronized void saveToFile(T persistable, File dir, File storageFile) {
        File tempFile = null;
        FileOutputStream fileOutputStream = null;
        PrintWriter printWriter = null;

        try {
            log.debug("Write to disc: {}", storageFile.getName());
            protobuf.PersistableEnvelope protoPersistable;
            try {
                protoPersistable = (protobuf.PersistableEnvelope) persistable.toProtoMessage();
                if (protoPersistable.toByteArray().length == 0)
                    log.error("protoPersistable is empty. persistable=" + persistable.getClass().getSimpleName());
            } catch (Throwable e) {
                log.error("Error in saveToFile toProtoMessage: {}, {}", persistable.getClass().getSimpleName(), storageFile);
                e.printStackTrace();
                throw new RuntimeException(e);
            }

            if (!dir.exists() && !dir.mkdir())
                log.warn("make dir failed");

            tempFile = File.createTempFile("temp", null, dir);
            tempFile.deleteOnExit();
            fileOutputStream = new FileOutputStream(tempFile);

            log.debug("Writing protobuffer class:{} to file:{}", persistable.getClass(), storageFile.getName());
            writeLock.lock();
            protoPersistable.writeDelimitedTo(fileOutputStream);

            // Attempt to force the bits to hit the disk. In reality the OS or hard disk itself may still decide
            // to not write through to physical media for at least a few seconds, but this is the best we can do.
            fileOutputStream.flush();
            fileOutputStream.getFD().sync();
            writeLock.unlock();

            // Close resources before replacing file with temp file because otherwise it causes problems on windows
            // when rename temp file
            fileOutputStream.close();
            FileUtil.renameFile(tempFile, storageFile);
        } catch (Throwable t) {
            log.error("Error at saveToFile, storageFile=" + storageFile.toString(), t);
        } finally {
            if (writeLock.isLocked())
                writeLock.unlock();
            if (tempFile != null && tempFile.exists()) {
                log.warn("Temp file still exists after failed save. We will delete it now. storageFile=" + storageFile);
                if (!tempFile.delete())
                    log.error("Cannot delete temp file.");
            }

            try {
                if (fileOutputStream != null)
                    fileOutputStream.close();
                //noinspection ConstantConditions,ConstantConditions
                if (printWriter != null)
                    printWriter.close();
            } catch (IOException e) {
                // We swallow that
                e.printStackTrace();
                log.error("Cannot close resources." + e.getMessage());
            }
        }
    }
}
<commit_msg>Clean up AtomicBoolean usage in FileManager<commit_after>/*
 * This file is part of Bisq.
 *
 * Bisq is free software: you can redistribute it and/or modify it
 * under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or (at
 * your option) any later version.
 *
 * Bisq 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 Affero General Public
 * License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with Bisq. If not, see <http://www.gnu.org/licenses/>.
 */

package bisq.common.storage;

import bisq.common.UserThread;
import bisq.common.proto.persistable.PersistableEnvelope;
import bisq.common.proto.persistable.PersistenceProtoResolver;
import bisq.common.util.Utilities;

import com.google.common.util.concurrent.CycleDetectingLockFactory;

import java.nio.file.Paths;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.PrintWriter;

import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import lombok.extern.slf4j.Slf4j;

import static com.google.common.base.Preconditions.checkNotNull;

@Slf4j
public class FileManager<T extends PersistableEnvelope> {
    private final File dir;
    private final File storageFile;
    private final ScheduledThreadPoolExecutor executor;
    private final long delay;
    private final Callable<Void> saveFileTask;
    private final AtomicReference<T> nextWrite;
    private final PersistenceProtoResolver persistenceProtoResolver;
    private final ReentrantLock writeLock = CycleDetectingLockFactory.newInstance(CycleDetectingLockFactory.Policies.THROW).newReentrantLock("writeLock");

    ///////////////////////////////////////////////////////////////////////////////////////////
    // Constructor
    ///////////////////////////////////////////////////////////////////////////////////////////

    public FileManager(File dir, File storageFile, long delay, PersistenceProtoResolver persistenceProtoResolver) {
        this.dir = dir;
        this.storageFile = storageFile;
        this.persistenceProtoResolver = persistenceProtoResolver;
        this.nextWrite = new AtomicReference<>(null);

        executor = Utilities.getScheduledThreadPoolExecutor("FileManager", 1, 10, 5);

        // File must only be accessed from the auto-save executor from now on, to avoid simultaneous access.
        this.delay = delay;

        saveFileTask = () -> {
            try {
                Thread.currentThread().setName("Save-file-task-" + new Random().nextInt(10000));

                // Atomically take the next object to write and set the value to null so `saveLater` callers can
                // determine if there is a pending write.
                T persistable = this.nextWrite.getAndSet(null);
                checkNotNull(persistable);

                saveNowInternal(persistable);
            } catch (Throwable e) {
                log.error("Error during saveFileTask", e);
            }
            return null;
        };
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            UserThread.execute(FileManager.this::shutDown);
        }, "FileManager.ShutDownHook"));
    }


    ///////////////////////////////////////////////////////////////////////////////////////////
    // API
    ///////////////////////////////////////////////////////////////////////////////////////////

    /**
     * Actually write the wallet file to disk, using an atomic rename when possible. Runs on the current thread.
     */
    public void saveNow(T persistable) {
        saveNowInternal(persistable);
    }

    /**
     * Queues up a save in the background. Useful for not very important wallet changes.
     */
    public void saveLater(T persistable) {
        saveLater(persistable, delay);
    }

    public void saveLater(T persistable, long delayInMilli) {
        // Atomically set the value of the next write. This allows batching of multiple writes of the same data
        // structure if there are multiple calls to saveLater within a given `delayInMillis`.
        T pendingWrite = this.nextWrite.getAndSet(persistable);

        // If there isn't a pending write. Schedule one for `persistable`.
        if (pendingWrite == null)
            executor.schedule(saveFileTask, delayInMilli, TimeUnit.MILLISECONDS);
    }

    @SuppressWarnings("unchecked")
    public synchronized T read(File file) {
        log.debug("Read from disc: {}", file.getName());

        try (final FileInputStream fileInputStream = new FileInputStream(file)) {
            protobuf.PersistableEnvelope persistable = protobuf.PersistableEnvelope.parseDelimitedFrom(fileInputStream);
            return (T) persistenceProtoResolver.fromProto(persistable);
        } catch (Throwable t) {
            String errorMsg = "Exception at proto read: " + t.getMessage() + " file:" + file.getAbsolutePath();
            log.error(errorMsg, t);
            //if(DevEnv.DEV_MODE)
            throw new RuntimeException(errorMsg);
        }
    }

    public synchronized void removeFile(String fileName) {
        File file = new File(dir, fileName);
        boolean result = file.delete();
        if (!result)
            log.warn("Could not delete file: " + file.toString());

        File backupDir = new File(Paths.get(dir.getAbsolutePath(), "backup").toString());
        if (backupDir.exists()) {
            File backupFile = new File(Paths.get(dir.getAbsolutePath(), "backup", fileName).toString());
            if (backupFile.exists()) {
                result = backupFile.delete();
                if (!result)
                    log.warn("Could not delete backupFile: " + file.toString());
            }
        }
    }


    /**
     * Shut down auto-saving.
     */
    void shutDown() {
        executor.shutdown();
        try {
            executor.awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static void removeAndBackupFile(File dbDir, File storageFile, String fileName, String backupFolderName)
            throws IOException {
        File corruptedBackupDir = new File(Paths.get(dbDir.getAbsolutePath(), backupFolderName).toString());
        if (!corruptedBackupDir.exists())
            if (!corruptedBackupDir.mkdir())
                log.warn("make dir failed");

        File corruptedFile = new File(Paths.get(dbDir.getAbsolutePath(), backupFolderName, fileName).toString());
        FileUtil.renameFile(storageFile, corruptedFile);
    }

    public synchronized void removeAndBackupFile(String fileName) throws IOException {
        removeAndBackupFile(dir, storageFile, fileName, "backup_of_corrupted_data");
    }

    public synchronized void backupFile(String fileName, int numMaxBackupFiles) {
        FileUtil.rollingBackup(dir, fileName, numMaxBackupFiles);
    }

    ///////////////////////////////////////////////////////////////////////////////////////////
    // Private
    ///////////////////////////////////////////////////////////////////////////////////////////

    private void saveNowInternal(T persistable) {
        long now = System.currentTimeMillis();
        saveToFile(persistable, dir, storageFile);
        log.debug("Save {} completed in {} msec", storageFile, System.currentTimeMillis() - now);
    }

    private synchronized void saveToFile(T persistable, File dir, File storageFile) {
        File tempFile = null;
        FileOutputStream fileOutputStream = null;
        PrintWriter printWriter = null;

        try {
            log.debug("Write to disc: {}", storageFile.getName());
            protobuf.PersistableEnvelope protoPersistable;
            try {
                protoPersistable = (protobuf.PersistableEnvelope) persistable.toProtoMessage();
                if (protoPersistable.toByteArray().length == 0)
                    log.error("protoPersistable is empty. persistable=" + persistable.getClass().getSimpleName());
            } catch (Throwable e) {
                log.error("Error in saveToFile toProtoMessage: {}, {}", persistable.getClass().getSimpleName(), storageFile);
                e.printStackTrace();
                throw new RuntimeException(e);
            }

            if (!dir.exists() && !dir.mkdir())
                log.warn("make dir failed");

            tempFile = File.createTempFile("temp", null, dir);
            tempFile.deleteOnExit();
            fileOutputStream = new FileOutputStream(tempFile);

            log.debug("Writing protobuffer class:{} to file:{}", persistable.getClass(), storageFile.getName());
            writeLock.lock();
            protoPersistable.writeDelimitedTo(fileOutputStream);

            // Attempt to force the bits to hit the disk. In reality the OS or hard disk itself may still decide
            // to not write through to physical media for at least a few seconds, but this is the best we can do.
            fileOutputStream.flush();
            fileOutputStream.getFD().sync();
            writeLock.unlock();

            // Close resources before replacing file with temp file because otherwise it causes problems on windows
            // when rename temp file
            fileOutputStream.close();
            FileUtil.renameFile(tempFile, storageFile);
        } catch (Throwable t) {
            log.error("Error at saveToFile, storageFile=" + storageFile.toString(), t);
        } finally {
            if (writeLock.isLocked())
                writeLock.unlock();
            if (tempFile != null && tempFile.exists()) {
                log.warn("Temp file still exists after failed save. We will delete it now. storageFile=" + storageFile);
                if (!tempFile.delete())
                    log.error("Cannot delete temp file.");
            }

            try {
                if (fileOutputStream != null)
                    fileOutputStream.close();
                //noinspection ConstantConditions,ConstantConditions
                if (printWriter != null)
                    printWriter.close();
            } catch (IOException e) {
                // We swallow that
                e.printStackTrace();
                log.error("Cannot close resources." + e.getMessage());
            }
        }
    }
}

{'fraction_non_alphanumeric': 0.1402748660610296, 'fraction_numerical': 0.001397624039133473, 'mean_word_length': 2.090856731461483, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4902960', 'n_tokens_mistral': 5286, 'n_tokens_neox': 5131, 'n_words': 1495}