<commit_before>package org.broadinstitute.hellbender.utils.nio;

import com.google.common.base.Stopwatch;
import org.broadinstitute.hellbender.exceptions.GATKException;

import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.NonWritableChannelException;
import java.nio.channels.SeekableByteChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.UnknownFormatConversionException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * SeekableByteChannelPrefetcher wraps an existing SeekableByteChannel to add prefetching.
 * The prefetching is done on a different thread, so you can write simple code that repeatedly
 * calls read() to get data, processes it, and then calls read again -- and yet this
 * simple code overlaps computation and communication for you.
 * (Of course this is only worthwhile if the underlying SeekableByteChannel doesn't already
 * implement prefetching).
 */
public final class SeekableByteChannelPrefetcher implements SeekableByteChannel {

    // Only one thread at a time should use chan.
    // To ensure this is the case, only the prefetching thread uses it.
    private final SeekableByteChannel chan;
    private final int bufSize;
    private final ExecutorService exec;
    private final long size;
    private List<WorkUnit> full = new ArrayList<>();
    private WorkUnit fetching = null;
    // total number of buffers
    private final static int BUF_COUNT = 2;
    // where we pretend to be, wrt returning bytes from read()
    private long position = 0;
    private boolean open;
    private Stopwatch betweenCallsToRead = Stopwatch.createUnstarted();

    // statistics, for profiling
    // time spent blocking the user because we're waiting on the network
    public long msWaitingForData = 0;
    // time spent blocking the user because we're copying bytes
    public long msCopyingData = 0;
    // total number of bytes returned by read (if the user asks for the same bytes multiple times, they count)
    public long bytesReturned = 0;
    // total number of bytes read over the network (whether returned to the user or not)
    public long bytesRead = 0;
    // time spend in between calls to Read, ie. presumably while the user is processing the data we returned.
    public long msBetweenCallsToRead = 0;
    // number of times we had the user's data already ready, didn't have to grab it from the net.
    public long nbHit = 0;
    // number of times we had already started to prefetch the user's data (but it hadn't arrived yet).
    public long nbNearHit = 0;
    // number of times we don't have what the user's asking for, we have to wait for a prefetch to finish,
    // and the prefetch didn't return what the user wanted (either they are going backward, or jumping forward)
    public long nbMiss = 0;
    // number of times the user asks for data with a lower index than what we already have
    // (so they're not following the expected pattern of increasing indexes)
    public long nbGoingBack = 0;
    // number of times the user asks for data past the end of the file
    public long nbReadsPastEnd = 0;
    // timing statistics have an overhead, so only turn them on when debugging performance
    // issues.
    private static final boolean trackTime = false;

    /**
     * Wraps the provided SeekableByteChannel within a SeekableByteChannelPrefetcher, using the provided buffer size
     *
     * @param bufferSizeMB buffer size in MB
     * @param channel channel to wrap in the prefetcher
     * @return wrapped channel
     */
    public static SeekableByteChannel addPrefetcher(int bufferSizeMB, SeekableByteChannel channel) {
        try {
            return new SeekableByteChannelPrefetcher(channel, bufferSizeMB * 1024 * 1024);
        } catch (IOException ex) {
            throw new GATKException("Unable to initialize the prefetcher: " + ex);
        }
    }

    /**
     * WorkUnit holds a buffer and the instructions for what to put in it.
     *
     * <p>Use it like this:
     * <ol>
     *   <li> call()
     *   <li> the data is now in buf, you can access it directly
     *   <li> if need more, call resetForIndex(...) and go back to the top.
     *   <li> else, call close()
     * </ol>
     */
    private static class WorkUnit implements Callable<ByteBuffer>, Closeable {
        public final ByteBuffer buf;
        public long blockIndex;
        private final SeekableByteChannel chan;
        private final int blockSize;
        private Future<ByteBuffer> futureBuf;

        public WorkUnit(SeekableByteChannel chan, int blockSize, long blockIndex) {
            this.chan = chan;
            this.buf = ByteBuffer.allocate(blockSize);
            this.futureBuf = null;
            this.blockSize = blockSize;
            this.blockIndex = blockIndex;
        }

        @Override
        public ByteBuffer call() throws IOException {
            long pos = ((long)blockSize) * blockIndex;
            if (pos > chan.size()) {
                return null;
            }
            if (pos < 0) {
                // This should never happen, if the code's correct.
                throw new IllegalArgumentException("blockIndex " + blockIndex + " has position " + pos + ": negative position is not valid.");
            }
            chan.position(pos);
            // read until buffer is full, or EOF
            while (chan.read(buf) >= 0 && buf.hasRemaining()) {}
            return buf;
        }

        public ByteBuffer getBuf() throws ExecutionException, InterruptedException {
            return futureBuf.get();
        }

        public WorkUnit resetForIndex(long blockIndex) {
            this.blockIndex = blockIndex;
            buf.clear();
            futureBuf = null;
            return this;
        }


        @Override
        public void close() throws IOException {
            chan.close();
        }
    }

    public SeekableByteChannelPrefetcher(SeekableByteChannel chan, int bufSize) throws IOException {
        if (chan instanceof SeekableByteChannelPrefetcher) {
            throw new IllegalArgumentException("Cannot put two prefetchers on the same channel.");
        }
        if (!chan.isOpen()) {
            throw new IllegalArgumentException("channel must be open");
        }
        this.chan = chan;
        if (bufSize <= 0) {
            throw new IllegalArgumentException("bufSize must be positive");
        }
        this.size = chan.size();
        if (bufSize > this.size) {
            this.bufSize = (int)this.size;
        } else {
            this.bufSize = bufSize;
        }
        this.open = true;
        // We use a single thread to ensure no parallel accesses to the channel.
        exec = Executors.newFixedThreadPool(1);
    }

    public String getStatistics() {
        try {
            double returnedPct = (bytesRead > 0 ? (100.0 * bytesReturned / bytesRead) : 100.0);
            return String
                .format("Bytes read: %12d\n  returned: %12d ( %3.2f %% )", bytesRead, bytesReturned,
                    returnedPct)
                + String.format("\nReads past the end: %3d", nbReadsPastEnd)
                + String.format("\nReads forcing re-fetching of an earlier block: %3d", nbGoingBack)
                // A near-hit is when we're already fetching the data the user is asking for,
                // but we're not done loading it in.
                + String
                .format("\nCache\n hits:      %12d\n near-hits: %12d\n misses:    %12d", nbHit,
                    nbNearHit, nbMiss);
        } catch (UnknownFormatConversionException x) {
            // let's not crash the whole program, instead just return no info
            return "(error while formatting statistics)";
        }
    }

    // if we don't already have that block and the fetching thread is idle,<commit_msg>Name and daemonize prefetching thread (#2654)<commit_after>package org.broadinstitute.hellbender.utils.nio;

import com.google.common.base.Stopwatch;
import java.util.concurrent.ThreadFactory;
import org.broadinstitute.hellbender.exceptions.GATKException;

import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.NonWritableChannelException;
import java.nio.channels.SeekableByteChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.UnknownFormatConversionException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import shaded.cloud_nio.com.google.common.util.concurrent.ThreadFactoryBuilder;

/**
 * SeekableByteChannelPrefetcher wraps an existing SeekableByteChannel to add prefetching.
 * The prefetching is done on a different thread, so you can write simple code that repeatedly
 * calls read() to get data, processes it, and then calls read again -- and yet this
 * simple code overlaps computation and communication for you.
 * (Of course this is only worthwhile if the underlying SeekableByteChannel doesn't already
 * implement prefetching).
 */
public final class SeekableByteChannelPrefetcher implements SeekableByteChannel {

    // Only one thread at a time should use chan.
    // To ensure this is the case, only the prefetching thread uses it.
    private final SeekableByteChannel chan;
    private final int bufSize;
    private final ExecutorService exec;
    private final long size;
    private List<WorkUnit> full = new ArrayList<>();
    private WorkUnit fetching = null;
    // total number of buffers
    private final static int BUF_COUNT = 2;
    // where we pretend to be, wrt returning bytes from read()
    private long position = 0;
    private boolean open;
    private Stopwatch betweenCallsToRead = Stopwatch.createUnstarted();
    private static int prefetcherCount = 0;
    private final int prefetcherIndex;

    // statistics, for profiling
    // time spent blocking the user because we're waiting on the network
    public long msWaitingForData = 0;
    // time spent blocking the user because we're copying bytes
    public long msCopyingData = 0;
    // total number of bytes returned by read (if the user asks for the same bytes multiple times, they count)
    public long bytesReturned = 0;
    // total number of bytes read over the network (whether returned to the user or not)
    public long bytesRead = 0;
    // time spend in between calls to Read, ie. presumably while the user is processing the data we returned.
    public long msBetweenCallsToRead = 0;
    // number of times we had the user's data already ready, didn't have to grab it from the net.
    public long nbHit = 0;
    // number of times we had already started to prefetch the user's data (but it hadn't arrived yet).
    public long nbNearHit = 0;
    // number of times we don't have what the user's asking for, we have to wait for a prefetch to finish,
    // and the prefetch didn't return what the user wanted (either they are going backward, or jumping forward)
    public long nbMiss = 0;
    // number of times the user asks for data with a lower index than what we already have
    // (so they're not following the expected pattern of increasing indexes)
    public long nbGoingBack = 0;
    // number of times the user asks for data past the end of the file
    public long nbReadsPastEnd = 0;
    // timing statistics have an overhead, so only turn them on when debugging performance
    // issues.
    private static final boolean trackTime = false;

    /**
     * Wraps the provided SeekableByteChannel within a SeekableByteChannelPrefetcher, using the provided buffer size
     *
     * @param bufferSizeMB buffer size in MB
     * @param channel channel to wrap in the prefetcher
     * @return wrapped channel
     */
    public static SeekableByteChannel addPrefetcher(int bufferSizeMB, SeekableByteChannel channel) {
        try {
            return new SeekableByteChannelPrefetcher(channel, bufferSizeMB * 1024 * 1024);
        } catch (IOException ex) {
            throw new GATKException("Unable to initialize the prefetcher: " + ex);
        }
    }

    /**
     * WorkUnit holds a buffer and the instructions for what to put in it.
     *
     * <p>Use it like this:
     * <ol>
     *   <li> call()
     *   <li> the data is now in buf, you can access it directly
     *   <li> if need more, call resetForIndex(...) and go back to the top.
     *   <li> else, call close()
     * </ol>
     */
    private static class WorkUnit implements Callable<ByteBuffer>, Closeable {
        public final ByteBuffer buf;
        public long blockIndex;
        private final SeekableByteChannel chan;
        private final int blockSize;
        private Future<ByteBuffer> futureBuf;

        public WorkUnit(SeekableByteChannel chan, int blockSize, long blockIndex) {
            this.chan = chan;
            this.buf = ByteBuffer.allocate(blockSize);
            this.futureBuf = null;
            this.blockSize = blockSize;
            this.blockIndex = blockIndex;
        }

        @Override
        public ByteBuffer call() throws IOException {
            long pos = ((long)blockSize) * blockIndex;
            if (pos > chan.size()) {
                return null;
            }
            if (pos < 0) {
                // This should never happen, if the code's correct.
                throw new IllegalArgumentException("blockIndex " + blockIndex + " has position " + pos + ": negative position is not valid.");
            }
            chan.position(pos);
            // read until buffer is full, or EOF
            while (chan.read(buf) >= 0 && buf.hasRemaining()) {}
            return buf;
        }

        public ByteBuffer getBuf() throws ExecutionException, InterruptedException {
            return futureBuf.get();
        }

        public WorkUnit resetForIndex(long blockIndex) {
            this.blockIndex = blockIndex;
            buf.clear();
            futureBuf = null;
            return this;
        }


        @Override
        public void close() throws IOException {
            chan.close();
        }
    }

    public SeekableByteChannelPrefetcher(SeekableByteChannel chan, int bufSize) throws IOException {
        if (chan instanceof SeekableByteChannelPrefetcher) {
            throw new IllegalArgumentException("Cannot put two prefetchers on the same channel.");
        }
        if (!chan.isOpen()) {
            throw new IllegalArgumentException("channel must be open");
        }
        this.chan = chan;
        if (bufSize <= 0) {
            throw new IllegalArgumentException("bufSize must be positive");
        }
        this.size = chan.size();
        if (bufSize > this.size) {
            this.bufSize = (int)this.size;
        } else {
            this.bufSize = bufSize;
        }
        this.open = true;
        this.prefetcherIndex = (prefetcherCount++);
        // Make sure the prefetching thread's name indicate what it is and
        // which prefetcher it belongs to (for debugging purposes only, naturally).
        String nameFormat = "nio-prefetcher-" + prefetcherIndex + "-thread-%d";
        ThreadFactory threadFactory = new ThreadFactoryBuilder()
            .setNameFormat(nameFormat)
            .setDaemon(true)
            .build();
        // Single thread to ensure no concurrent access to chan.
        exec = Executors.newFixedThreadPool(1, threadFactory);
    }

    public String getStatistics() {
        try {
            double returnedPct = (bytesRead > 0 ? (100.0 * bytesReturned / bytesRead) : 100.0);
            return String
                .format("Bytes read: %12d\n  returned: %12d ( %3.2f %% )", bytesRead, bytesReturned,
                    returnedPct)
                + String.format("\nReads past the end: %3d", nbReadsPastEnd)
                + String.format("\nReads forcing re-fetching of an earlier block: %3d", nbGoingBack)
                // A near-hit is when we're already fetching the data the user is asking for,
                // but we're not done loading it in.
                + String
                .format("\nCache\n hits:      %12d\n near-hits: %12d\n misses:    %12d", nbHit,
                    nbNearHit, nbMiss);
        } catch (UnknownFormatConversionException x) {
            // let's not crash the whole program, instead just return no info
            return "(error while formatting statistics)";
        }
    }

    // if we don't already have that block and the fetching thread is idle,
{'fraction_non_alphanumeric': 0.07700503107428233, 'fraction_numerical': 0.005859721811186742, 'mean_word_length': 2.4837113402061854, 'pattern_counts': {'":': 2, '<': 29, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '7664731', 'n_tokens_mistral': 4559, 'n_tokens_neox': 4329, 'n_words': 1689}