<commit_before>package net.serverpeon.twitcharchiver.downloader;

import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import com.google.common.collect.MapMaker;
import com.google.common.collect.Maps;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAccumulator;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantLock;

public class ProgressTracker {
    private final static Logger logger = LogManager.getLogger(ProgressTracker.class);
    private static final Gson GSON = new GsonBuilder().setPrettyPrinting().create();

    private final Map<String, Status> currentStatus = Maps.newHashMap();

    private final ReentrantLock writeLock = new ReentrantLock();

    private final File trackerFile;
    private final int numberOfParts;
    private ProgressUpdater updater = null;

    public ProgressTracker(File storageFolder, int numberOfParts) {
        this.numberOfParts = numberOfParts;
        this.trackerFile = new File(storageFolder, "status.json");

        if (trackerFile.exists()) {
            try (final FileReader reader = new FileReader(trackerFile)) {
                final Map<?, ?> map = GSON.fromJson(reader, Map.class);
                for (Map.Entry<?, ?> entry : map.entrySet()) {
                    this.currentStatus.put(
                            entry.getKey().toString(),
                            Status.valueOf(entry.getValue().toString())
                    );
                }
            } catch (IOException e) {
                logger.debug("Error reading!", e);
            }

            Iterables.removeIf(currentStatus.values(), new Predicate<Status>() {
                @Override
                public boolean apply(Status status) {
                    return status != Status.DOWNLOADED;
                }
            });
        }
    }

    public void reset(final ProgressUpdater updater) {
        this.updater = updater;

        //Clean up downloading state
    }

    public Status getStatus(final String ident) {
        final Status status = currentStatus.get(ident);
        return status != null ? status : Status.UNTRACKED;
    }

    public int getStatusCount(final Status status) {
        return Maps.filterValues(currentStatus, new Predicate<Status>() {
            @Override
            public boolean apply(Status s) {
                return status == s;
            }
        }).size();
    }

    public Partial track(final String ident) {
        return new Partial(ident, updater);
    }

    private void write() {
        try (final FileWriter writer = new FileWriter(trackerFile, false)) {
            GSON.toJson(this.currentStatus, writer);
        } catch (IOException e) {
            logger.warn("Error saving!", e);
        }
    }

    public enum Status {
        UNTRACKED,
        FAILED,
        DOWNLOADED
    }

    public interface ProgressUpdater {
        void updateProgress(int counter, int total);

        void updateCount(Status type);
    }

    public class Partial {
        private final String identifier;
        private final ProgressUpdater updater;

        Partial(final String identifier, ProgressUpdater updater) {
            this.identifier = identifier;
            this.updater = updater;
        }

        public void update(final long soFar, final long expectedTotal) {
            //updater.updateProgress(soFar, expectedTotal);
        }

        private int count() {
            return getStatusCount(Status.DOWNLOADED);
        }

        public void invalidate() {
            writeLock.lock();
            try {
                currentStatus.put(identifier, Status.FAILED);
                updater.updateCount(Status.FAILED);
                updater.updateProgress(count(), numberOfParts);
                write();
            } finally {
                writeLock.unlock();
            }
        }

        public void finish() {
            writeLock.lock();
            try {
                currentStatus.put(identifier, Status.DOWNLOADED);
                updater.updateCount(Status.DOWNLOADED);
                updater.updateProgress(count(), numberOfParts);
                write();
            } finally {
                writeLock.unlock();
            }
        }
    }
}<commit_msg>Properly lock access to the status map<commit_after>package net.serverpeon.twitcharchiver.downloader;

import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ProgressTracker {
    private final static Logger logger = LogManager.getLogger(ProgressTracker.class);
    private static final Gson GSON = new GsonBuilder().setPrettyPrinting().create();

    private final Map<String, Status> currentStatus = Maps.newHashMap();
    private final ReentrantReadWriteLock mapLock = new ReentrantReadWriteLock();

    private final File trackerFile;
    private final int numberOfParts;
    private ProgressUpdater updater = null;

    public ProgressTracker(File storageFolder, int numberOfParts) {
        this.numberOfParts = numberOfParts;
        this.trackerFile = new File(storageFolder, "status.json");

        if (trackerFile.exists()) {
            try (final FileReader reader = new FileReader(trackerFile)) {
                final Map<?, ?> map = GSON.fromJson(reader, Map.class);
                for (Map.Entry<?, ?> entry : map.entrySet()) {
                    this.currentStatus.put(
                            entry.getKey().toString(),
                            Status.valueOf(entry.getValue().toString())
                    );
                }
            } catch (IOException e) {
                logger.debug("Error reading!", e);
            }

            this.reset(null); //Reset 'failed' entries
        }
    }

    public void reset(final ProgressUpdater updater) {
        this.updater = updater;

        //Clean up downloading state
        mapLock.writeLock().lock();
        try {
            Iterables.removeIf(currentStatus.values(), new Predicate<Status>() {
                @Override
                public boolean apply(Status status) {
                    return status != Status.DOWNLOADED;
                }
            });
        } finally {
            mapLock.writeLock().unlock();
        }
    }

    public Status getStatus(final String ident) {
        final Status status = currentStatus.get(ident);
        return status != null ? status : Status.UNTRACKED;
    }

    public int getStatusCount(final Status status) {
        mapLock.readLock().lock();
        try {
            return Maps.filterValues(currentStatus, new Predicate<Status>() {
                @Override
                public boolean apply(Status s) {
                    return status == s;
                }
            }).size();
        } finally {
            mapLock.readLock().unlock();
        }
    }

    public Partial track(final String ident) {
        return new Partial(ident, updater);
    }

    private void write() {
        mapLock.readLock().lock();
        try {
            try (final FileWriter writer = new FileWriter(trackerFile, false)) {
                GSON.toJson(this.currentStatus, writer);
            } catch (IOException e) {
                logger.warn("Error saving!", e);
            }
        } finally {
            mapLock.readLock().unlock();
        }
    }

    public enum Status {
        UNTRACKED,
        FAILED,
        DOWNLOADED
    }

    public interface ProgressUpdater {
        void updateProgress(int counter, int total);

        void updateCount(Status type);
    }

    public class Partial {
        private final String identifier;
        private final ProgressUpdater updater;

        Partial(final String identifier, ProgressUpdater updater) {
            this.identifier = identifier;
            this.updater = updater;
        }

        public void update(final long soFar, final long expectedTotal) {
            //updater.updateProgress(soFar, expectedTotal);
        }

        private int count() {
            return getStatusCount(Status.DOWNLOADED);
        }

        public void invalidate() {
            mapLock.writeLock().lock();
            try {
                currentStatus.put(identifier, Status.FAILED);
                updater.updateCount(Status.FAILED);
                updater.updateProgress(count(), numberOfParts);
                write();
            } finally {
                mapLock.writeLock().unlock();
            }
        }

        public void finish() {
            mapLock.writeLock().lock();
            try {
                currentStatus.put(identifier, Status.DOWNLOADED);
                updater.updateCount(Status.DOWNLOADED);
                updater.updateProgress(count(), numberOfParts);
                write();
            } finally {
                mapLock.writeLock().unlock();
            }
        }
    }
}
{'fraction_non_alphanumeric': 0.09707112970711297, 'fraction_numerical': 0.00041841004184100416, 'mean_word_length': 1.9646511627906977, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1221648', 'n_tokens_mistral': 2502, 'n_tokens_neox': 2418, 'n_words': 558}