<commit_before>package com.yahoo.omid.tso;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ExecutionException;

import com.google.common.base.Optional;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

import org.jboss.netty.channel.Channel;

import com.yahoo.omid.committable.CommitTable;
import com.lmax.disruptor.*;
import com.codahale.metrics.MetricRegistry;

import static com.codahale.metrics.MetricRegistry.name;

import com.codahale.metrics.Timer;
import com.codahale.metrics.Histogram;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

class PersistenceProcessorImpl
    implements EventHandler<PersistenceProcessorImpl.PersistEvent>,
               PersistenceProcessor, TimeoutHandler {
    private static final Logger LOG = LoggerFactory.getLogger(PersistenceProcessor.class);

    final ReplyProcessor reply;
    final CommitTable.Client commitTableClient;
    final CommitTable.Writer writer;
    final RingBuffer<PersistEvent> persistRing;

    final int maxBatchSize;

    final Batch batch;

    final Timer flushTimer;
    final Histogram batchSizeHistogram;
    long lastFlush = System.nanoTime();
    final static int BATCH_TIMEOUT_MS = 5;

    PersistenceProcessorImpl(MetricRegistry metrics,
                             CommitTable commitTable, ReplyProcessor reply, int maxBatchSize)
                                     throws InterruptedException, ExecutionException {

        this.commitTableClient = commitTable.getClient().get();
        this.writer = commitTable.getWriter().get();
        this.reply = reply;
        this.maxBatchSize = maxBatchSize;
        
        batch = new Batch(maxBatchSize);

        flushTimer = metrics.timer(name("tso", "persist", "flush"));
        batchSizeHistogram = metrics.histogram(name("tso", "persist", "batchsize"));

        // FIXME consider putting something more like a phased strategy here to avoid
        // all the syscalls
        final TimeoutBlockingWaitStrategy timeoutStrategy
            = new TimeoutBlockingWaitStrategy(BATCH_TIMEOUT_MS, TimeUnit.MILLISECONDS);

        persistRing = RingBuffer.<PersistEvent>createSingleProducer(
                PersistEvent.EVENT_FACTORY, 1<<12, timeoutStrategy);
        SequenceBarrier persistSequenceBarrier = persistRing.newBarrier();
        BatchEventProcessor<PersistEvent> persistProcessor = new BatchEventProcessor<PersistEvent>(
                persistRing,
                persistSequenceBarrier,
                this);
        persistRing.addGatingSequences(persistProcessor.getSequence());


        ExecutorService persistExec = Executors.newSingleThreadExecutor(
                new ThreadFactoryBuilder().setNameFormat("persist-%d").build());
        persistExec.submit(persistProcessor);
    }

    @Override
    public void onEvent(final PersistEvent event, final long sequence, final boolean endOfBatch)
        throws Exception {
        
        switch (event.getType()) {
        case COMMIT:
            // TODO: What happens when the IOException is thrown?
            writer.addCommittedTransaction(event.getStartTimestamp(), event.getCommitTimestamp());
            batch.addCommit(event.getStartTimestamp(), event.getCommitTimestamp(), event.getChannel());
            break;
        case ABORT:
            handleAbort(event.getStartTimestamp(), event.isRetry(), event.getChannel());
            break;
        case TIMESTAMP:
            batch.addTimestamp(event.getStartTimestamp(), event.getChannel());
            break;
        }

        if (batch.isFull() || endOfBatch) {
            maybeFlushBatch();
        }
    }

    private void handleAbort(long startTimestamp, boolean isRetry, Channel channel) {

        if(!isRetry) {
            reply.abortResponse(startTimestamp, channel);
            return;
        }

        // 1) Force to flush so the whole batch hits disk
        flush();
        // 2) Check the commit table
        final Optional<Long> commitTimestamp;
        try {
            commitTimestamp = commitTableClient.getCommitTimestamp(startTimestamp).get();
            if(!commitTimestamp.isPresent()) {
                reply.abortResponse(startTimestamp, channel);
            } else {
                reply.commitResponse(startTimestamp, commitTimestamp.get(), channel);
            }
        } catch (InterruptedException e) {
            LOG.error("Interrupted reading from commit table");
            Thread.currentThread().interrupt();
        } catch (ExecutionException e) {
            LOG.error("Error reading from commit table", e);
        }
    }

    // no event has been received in the timeout period
    @Override
    public void onTimeout(final long sequence) {
        maybeFlushBatch();
    }

    /**
     * Flush the current batch if it is larger than the max batch size,
     * or BATCH_TIMEOUT_MS milliseconds have elapsed since the last flush.
     */
    private void maybeFlushBatch() {
        if (batch.isFull() || (System.nanoTime() - lastFlush) > TimeUnit.MILLISECONDS.toNanos(BATCH_TIMEOUT_MS)) {
            flush();
        }<commit_msg>Increased persistence timeout and added metric<commit_after>package com.yahoo.omid.tso;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ExecutionException;

import com.google.common.base.Optional;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

import org.jboss.netty.channel.Channel;

import com.yahoo.omid.committable.CommitTable;
import com.lmax.disruptor.*;
import com.codahale.metrics.MetricRegistry;

import static com.codahale.metrics.MetricRegistry.name;

import com.codahale.metrics.Timer;
import com.codahale.metrics.Meter;
import com.codahale.metrics.Histogram;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

class PersistenceProcessorImpl
    implements EventHandler<PersistenceProcessorImpl.PersistEvent>,
               PersistenceProcessor, TimeoutHandler {
    private static final Logger LOG = LoggerFactory.getLogger(PersistenceProcessor.class);

    final ReplyProcessor reply;
    final CommitTable.Client commitTableClient;
    final CommitTable.Writer writer;
    final RingBuffer<PersistEvent> persistRing;

    final int maxBatchSize;

    final Batch batch;

    final Timer flushTimer;
    final Histogram batchSizeHistogram;
    final Meter timeoutMeter;
    long lastFlush = System.nanoTime();
    final static int BATCH_TIMEOUT_MS = 100;

    PersistenceProcessorImpl(MetricRegistry metrics,
                             CommitTable commitTable, ReplyProcessor reply, int maxBatchSize)
                                     throws InterruptedException, ExecutionException {

        this.commitTableClient = commitTable.getClient().get();
        this.writer = commitTable.getWriter().get();
        this.reply = reply;
        this.maxBatchSize = maxBatchSize;
        
        batch = new Batch(maxBatchSize);

        flushTimer = metrics.timer(name("tso", "persist", "flush"));
        batchSizeHistogram = metrics.histogram(name("tso", "persist", "batchsize"));
        timeoutMeter = metrics.meter(name("tso", "persist", "timeout"));

        // FIXME consider putting something more like a phased strategy here to avoid
        // all the syscalls
        final TimeoutBlockingWaitStrategy timeoutStrategy
            = new TimeoutBlockingWaitStrategy(BATCH_TIMEOUT_MS, TimeUnit.MILLISECONDS);

        persistRing = RingBuffer.<PersistEvent>createSingleProducer(
                PersistEvent.EVENT_FACTORY, 1<<20, timeoutStrategy); // 2^20 entries in ringbuffer
        SequenceBarrier persistSequenceBarrier = persistRing.newBarrier();
        BatchEventProcessor<PersistEvent> persistProcessor = new BatchEventProcessor<PersistEvent>(
                persistRing,
                persistSequenceBarrier,
                this);
        persistRing.addGatingSequences(persistProcessor.getSequence());


        ExecutorService persistExec = Executors.newSingleThreadExecutor(
                new ThreadFactoryBuilder().setNameFormat("persist-%d").build());
        persistExec.submit(persistProcessor);
    }

    @Override
    public void onEvent(final PersistEvent event, final long sequence, final boolean endOfBatch)
        throws Exception {
        
        switch (event.getType()) {
        case COMMIT:
            // TODO: What happens when the IOException is thrown?
            writer.addCommittedTransaction(event.getStartTimestamp(), event.getCommitTimestamp());
            batch.addCommit(event.getStartTimestamp(), event.getCommitTimestamp(), event.getChannel());
            break;
        case ABORT:
            handleAbort(event.getStartTimestamp(), event.isRetry(), event.getChannel());
            break;
        case TIMESTAMP:
            batch.addTimestamp(event.getStartTimestamp(), event.getChannel());
            break;
        }

        if (batch.isFull() || endOfBatch) {
            maybeFlushBatch();
        }
    }

    private void handleAbort(long startTimestamp, boolean isRetry, Channel channel) {

        if(!isRetry) {
            reply.abortResponse(startTimestamp, channel);
            return;
        }

        // 1) Force to flush so the whole batch hits disk
        flush();
        // 2) Check the commit table
        final Optional<Long> commitTimestamp;
        try {
            commitTimestamp = commitTableClient.getCommitTimestamp(startTimestamp).get();
            if(!commitTimestamp.isPresent()) {
                reply.abortResponse(startTimestamp, channel);
            } else {
                reply.commitResponse(startTimestamp, commitTimestamp.get(), channel);
            }
        } catch (InterruptedException e) {
            LOG.error("Interrupted reading from commit table");
            Thread.currentThread().interrupt();
        } catch (ExecutionException e) {
            LOG.error("Error reading from commit table", e);
        }
    }

    // no event has been received in the timeout period
    @Override
    public void onTimeout(final long sequence) {
        timeoutMeter.mark();
        maybeFlushBatch();
    }

    /**
     * Flush the current batch if it is larger than the max batch size,
     * or BATCH_TIMEOUT_MS milliseconds have elapsed since the last flush.
     */
    private void maybeFlushBatch() {
        if (batch.isFull() || (System.nanoTime() - lastFlush) > TimeUnit.MILLISECONDS.toNanos(BATCH_TIMEOUT_MS)) {
            flush();
        }
{'fraction_non_alphanumeric': 0.08882386256371531, 'fraction_numerical': 0.0019822541060977913, 'mean_word_length': 2.7906976744186047, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6126240', 'n_tokens_mistral': 2785, 'n_tokens_neox': 2711, 'n_words': 625}