<commit_before>package higgla.server;

import juglr.*;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.Fieldable;
import org.apache.lucene.document.NumericField;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.Term;
import org.apache.lucene.index.TermDocs;
import org.apache.lucene.store.FSDirectory;
import org.apache.lucene.util.Version;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Responsible for handling {@link Transaction}s for a particular base. There
 * will be one {@code BaseActor} for each base the Higgla server has written to.
 *
 * @author <NAME> <mailto:<EMAIL>>
 * @since Feb 3, 2010
 */
public class BaseActor extends Actor {

    /** The higgla.meta file must always be exactly 2048 bytes in order
     * to guard against disk-full scenarios.
     * We also reserve a byte array of that size in memory to provide
     * durability against OOM. */
    private static final int META_SIZE = 2048;
    private static final int HIGGLA_META_VERSION = 1;
    private ByteBuffer metaBuffer;
    private File metaFile;

    private Queue<Transaction> todo;
    private int actualTransactionLatch;
    private Transaction actualTransaction;
    private List<Box> actualTransactionErrors;
    private IndexWriter indexWriter;
    private Address writer;
    private Address baseAddress;
    private String baseName;
    private boolean started;
    private AtomicLong revisionCounter;

    public BaseActor(String baseName) {
        this.baseName = baseName;
        todo = new PriorityQueue<Transaction>();
        actualTransactionErrors = new LinkedList<Box>();

        // Pre-allocate resources for the higgla.meta file
        metaBuffer = ByteBuffer.allocate(META_SIZE);
        metaFile = new File(baseName, "higgla.meta");
    }

    @Override
    public void start() {
        try {
            baseAddress = getBus().allocateNamedAddress(this,
                                                        "__base__" + baseName);
        } catch (AddressAlreadyOwnedException e) {
            // There was a race creating this BaseActor and another actor
            // is already responsible for this base so we silently retract
            // this actor from the bus
            getBus().freeAddress(getAddress());
        }

        // We can now assume that we are the unique owner
        // of the name "__base__${baseName}". Thus it should be safe to create
        // an IndexWriter for the base

        try {
            revisionCounter = new AtomicLong(readLastRevision());
        } catch (IOException e) {
            e.printStackTrace();
            System.err.println("I/O Error detecting last revision number");
            shutdown();
        }

        renewWriter();  // requires revisionCounter to be set
    }

    /**
     * Commits a {@link Transaction} to the Lucene index
     * @param message
     */
    @Override
    public void react(Message message) {
        if (message instanceof Transaction) {
            handleTransaction((Transaction)message);
        } else if (message instanceof Check) {
            handleCheck((Check)message);
        } else {
            throw new MessageFormatException(
                    "Expected Storage.Transaction. Got "
                    + message.getClass().getName());
        }
    }

    private void handleCheck(Check check) {
        assert actualTransaction != null;
        assert check.transactionId == actualTransaction.getId();
        actualTransactionLatch--;

        if (check.error != null) {
            actualTransactionErrors.add(check.error);
        }

        if (actualTransactionLatch == 0) {
            if (actualTransactionErrors.size() != 0) {
                try {
                    indexWriter.rollback();
                    renewWriter();
                } catch (IOException e) {
                    e.printStackTrace();
                    System.err.println(String.format(
                         "I/O Error while rolling back transaction '%s': %s",
                         actualTransaction.getId(), e.getMessage()));
                    return;
                } finally {
                    Box reply = formatMsg(
                            Long.toString(actualTransaction.getId()), "error");
                    reply.put("transaction", actualTransaction.getId());
                    reply.put("error", actualTransactionErrors);
                    send(reply, actualTransaction.getReplyTo());

                    // Reset and prepare for next transaction. Note that we can
                    // not run actualTransactionErrors.clear() since this list
                    // is now owned by the reply message. Instead we create a
                    // new list
                    actualTransaction = null;
                    actualTransactionErrors = new LinkedList<Box>();
                    renewWriter();
                    return;
                }
            }

            try {
                indexWriter.commit();
                commitMeta();
                Box reply = formatMsg(
                                Long.toString(actualTransaction.getId()), "ok");
                reply.put("transaction", actualTransaction.getId());
                send(reply, actualTransaction.getReplyTo());
                actualTransaction = null;
                scheduleNextTransaction();
            } catch (IOException e) {
                try {
                    Box reply = formatMsg(
                            Long.toString(actualTransaction.getId()),
                            "Failed to roll back transaction after failed commit: %s",
                            e.getMessage());
                    reply.put("transaction", actualTransaction.getId());
                    send(reply, actualTransaction.getReplyTo());
                    indexWriter.rollback();
                    actualTransaction = null;
                    scheduleNextTransaction();
                } catch (IOException e1) {
                    e.printStackTrace();
                    System.err.println(
                         "Failed to roll back transaction after failed commit");
                    shutdown();
                }
            }
        }
    }

    private void renewWriter() {
        assert revisionCounter != null;

        if (writer != null) {
            send(WriterActor.SHUTDOWN, writer);
        }
        if (indexWriter != null) {
            try {
                indexWriter.close();
            } catch (IOException e) {
                e.printStackTrace();
                System.err.println("I/O Error renewing index writer");
            }
        }

        try {
            indexWriter = new IndexWriter(FSDirectory.open(new File(baseName)),
                                          new StandardAnalyzer(
                                                  Version.LUCENE_CURRENT,<commit_msg>Keep a ref to the underlying FSDirectory around so we don't waste file handles<commit_after>package higgla.server;

import juglr.*;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.Fieldable;
import org.apache.lucene.document.NumericField;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.Term;
import org.apache.lucene.index.TermDocs;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;
import org.apache.lucene.util.Version;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Responsible for handling {@link Transaction}s for a particular base. There
 * will be one {@code BaseActor} for each base the Higgla server has written to.
 *
 * @author <NAME> <mailto:<EMAIL>>
 * @since Feb 3, 2010
 */
public class BaseActor extends Actor {

    /** The higgla.meta file must always be exactly 2048 bytes in order
     * to guard against disk-full scenarios.
     * We also reserve a byte array of that size in memory to provide
     * durability against OOM. */
    private static final int META_SIZE = 2048;
    private static final int HIGGLA_META_VERSION = 1;
    private ByteBuffer metaBuffer;
    private File metaFile;

    private Queue<Transaction> todo;
    private int actualTransactionLatch;
    private Transaction actualTransaction;
    private List<Box> actualTransactionErrors;
    private IndexWriter indexWriter;
    private Address writer;
    private Address baseAddress;
    private String baseName;
    private Directory baseDir;
    private boolean started;
    private AtomicLong revisionCounter;

    public BaseActor(String baseName) {        
        this.baseName = baseName;
        todo = new PriorityQueue<Transaction>();
        actualTransactionErrors = new LinkedList<Box>();

        // Pre-allocate resources for the higgla.meta file
        metaBuffer = ByteBuffer.allocate(META_SIZE);
        metaFile = new File(baseName, "higgla.meta");
    }

    @Override
    public void start() {
        try {
            baseAddress = getBus().allocateNamedAddress(this,
                                                        "__base__" + baseName);
        } catch (AddressAlreadyOwnedException e) {
            // There was a race creating this BaseActor and another actor
            // is already responsible for this base so we silently retract
            // this actor from the bus
            getBus().freeAddress(getAddress());
            return;
        }
        try {
            baseDir = FSDirectory.open(new File(baseName));
        } catch (IOException e) {
            e.printStackTrace();
            System.err.println("I/O Error opening base directory");
            shutdown();
            return;
        }

        // We can now assume that we are the unique owner
        // of the name "__base__${baseName}". Thus it should be safe to create
        // an IndexWriter for the base

        try {
            revisionCounter = new AtomicLong(readLastRevision());
        } catch (IOException e) {
            e.printStackTrace();
            System.err.println("I/O Error detecting last revision number");
            shutdown();
            return;
        }

        renewWriter();  // requires revisionCounter to be set
    }

    /**
     * Commits a {@link Transaction} to the Lucene index
     * @param message
     */
    @Override
    public void react(Message message) {
        if (message instanceof Transaction) {
            handleTransaction((Transaction)message);
        } else if (message instanceof Check) {
            handleCheck((Check)message);
        } else {
            throw new MessageFormatException(
                    "Expected Storage.Transaction. Got "
                    + message.getClass().getName());
        }
    }

    private void handleCheck(Check check) {
        assert actualTransaction != null;
        assert check.transactionId == actualTransaction.getId();
        actualTransactionLatch--;

        if (check.error != null) {
            actualTransactionErrors.add(check.error);
        }

        if (actualTransactionLatch == 0) {
            if (actualTransactionErrors.size() != 0) {
                try {
                    indexWriter.rollback();
                    renewWriter();
                } catch (IOException e) {
                    e.printStackTrace();
                    System.err.println(String.format(
                         "I/O Error while rolling back transaction '%s': %s",
                         actualTransaction.getId(), e.getMessage()));
                    return;
                } finally {
                    Box reply = formatMsg(
                            Long.toString(actualTransaction.getId()), "error");
                    reply.put("transaction", actualTransaction.getId());
                    reply.put("error", actualTransactionErrors);
                    send(reply, actualTransaction.getReplyTo());

                    // Reset and prepare for next transaction. Note that we can
                    // not run actualTransactionErrors.clear() since this list
                    // is now owned by the reply message. Instead we create a
                    // new list
                    actualTransaction = null;
                    actualTransactionErrors = new LinkedList<Box>();
                    renewWriter();
                    return;
                }
            }

            try {
                indexWriter.commit();
                commitMeta();
                Box reply = formatMsg(
                                Long.toString(actualTransaction.getId()), "ok");
                reply.put("transaction", actualTransaction.getId());
                send(reply, actualTransaction.getReplyTo());
                actualTransaction = null;
                scheduleNextTransaction();
            } catch (IOException e) {
                try {
                    Box reply = formatMsg(
                            Long.toString(actualTransaction.getId()),
                            "Failed to roll back transaction after failed commit: %s",
                            e.getMessage());
                    reply.put("transaction", actualTransaction.getId());
                    send(reply, actualTransaction.getReplyTo());
                    indexWriter.rollback();
                    actualTransaction = null;
                    scheduleNextTransaction();
                } catch (IOException e1) {
                    e.printStackTrace();
                    System.err.println(
                         "Failed to roll back transaction after failed commit");
                    shutdown();
                }
            }
        }
    }

    private void renewWriter() {
        assert revisionCounter != null;

        if (writer != null) {
            send(WriterActor.SHUTDOWN, writer);
        }
        if (indexWriter != null) {
            try {
                indexWriter.close();
            } catch (IOException e) {
                e.printStackTrace();
                System.err.println("I/O Error renewing index writer");
            }
        }

        try {
            indexWriter = new IndexWriter(baseDir,
                                          new StandardAnalyzer(
                                                  Version.LUCENE_CURRENT,
{'fraction_non_alphanumeric': 0.0875985567285848, 'fraction_numerical': 0.002271816116530803, 'mean_word_length': 1.7322015334063527, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4807796', 'n_tokens_mistral': 3575, 'n_tokens_neox': 3384, 'n_words': 985}