<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.flume.channel.file;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.LongBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel.MapMode;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

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

import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;

/**
 * Queue of events in the channel. This queue stores only
 * {@link FlumeEventPointer} objects which are represented
 * as 8 byte longs internally. Additionally the queue itself
 * of longs is stored as a memory mapped file with a fixed
 * header and circular queue semantics. The header of the queue
 * contains the timestamp of last sync, the queue size and
 * the head position.
 */
class FlumeEventQueue {
  private static final Logger LOG = LoggerFactory
  .getLogger(FlumeEventQueue.class);
  private static final int VERSION = 2;
  private static final int EMPTY = 0;
  private static final int INDEX_VERSION = 0;
  private static final int INDEX_TIMESTAMP = 1;
  private static final int INDEX_SIZE = 2;
  private static final int INDEX_HEAD = 3;
  private static final int INDEX_ACTIVE_LOG = 4;
  private static final int MAX_ACTIVE_LOGS = 1024;
  private static final int HEADER_SIZE = 1028;
  private final Map<Integer, AtomicInteger> fileIDCounts = Maps.newHashMap();
  private final MappedByteBuffer mappedBuffer;
  private final LongBuffer elementsBuffer;
  private LongBufferWrapper elements;
  private final RandomAccessFile checkpointFile;
  private final java.nio.channels.FileChannel checkpointFileHandle;
  private final int queueCapacity;

  private int queueSize;
  private int queueHead;
  private long timestamp;

  /**
   * @param capacity max event capacity of queue
   * @throws IOException
   */
  FlumeEventQueue(int capacity, File file) throws IOException {
    Preconditions.checkArgument(capacity > 0,
        "Capacity must be greater than zero");
    this.queueCapacity = capacity;

    if (!file.exists()) {
      Preconditions.checkState(file.createNewFile(), "Unable to create file: "
          + file);
    }

    boolean freshlyAllocated = false;
    checkpointFile = new RandomAccessFile(file, "rw");
    if (checkpointFile.length() == 0) {
      // Allocate
      LOG.info("Event queue has zero allocation. Initializing to capacity. "
          + "Please wait...");
      checkpointFile.writeLong(VERSION);
      int absoluteCapacity = capacity + HEADER_SIZE;
      for (int i = 1; i < absoluteCapacity; i++) {
        checkpointFile.writeLong(EMPTY);
      }
      LOG.info("Event queue allocation complete");
      freshlyAllocated = true;
    } else {
      int fileCapacity = (int) checkpointFile.length() / 8;
      int expectedCapacity = capacity + HEADER_SIZE;

      Preconditions.checkState(fileCapacity == expectedCapacity,
          "Capacity cannot be reduced once the channel is initialized");
    }

    checkpointFileHandle = checkpointFile.getChannel();

    mappedBuffer = checkpointFileHandle.map(MapMode.READ_WRITE, 0,
        file.length());
<commit_msg>FLUME-1304. Allow for faster allocation of checkpoint file.<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.flume.channel.file;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.LongBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel.MapMode;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

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

import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;

/**
 * Queue of events in the channel. This queue stores only
 * {@link FlumeEventPointer} objects which are represented
 * as 8 byte longs internally. Additionally the queue itself
 * of longs is stored as a memory mapped file with a fixed
 * header and circular queue semantics. The header of the queue
 * contains the timestamp of last sync, the queue size and
 * the head position.
 */
class FlumeEventQueue {
  private static final Logger LOG = LoggerFactory
  .getLogger(FlumeEventQueue.class);
  private static final long VERSION = 2;
  private static final int EMPTY = 0;
  private static final int INDEX_VERSION = 0;
  private static final int INDEX_TIMESTAMP = 1;
  private static final int INDEX_SIZE = 2;
  private static final int INDEX_HEAD = 3;
  private static final int INDEX_ACTIVE_LOG = 4;
  private static final int MAX_ACTIVE_LOGS = 1024;
  private static final int HEADER_SIZE = 1028;
  private static final int MAX_ALLOC_BUFFER_SIZE = 2*1024*1024; // 2MB
  private final Map<Integer, AtomicInteger> fileIDCounts = Maps.newHashMap();
  private final MappedByteBuffer mappedBuffer;
  private final LongBuffer elementsBuffer;
  private LongBufferWrapper elements;
  private final RandomAccessFile checkpointFile;
  private final java.nio.channels.FileChannel checkpointFileHandle;
  private final int queueCapacity;

  private int queueSize;
  private int queueHead;
  private long timestamp;

  /**
   * @param capacity max event capacity of queue
   * @throws IOException
   */
  FlumeEventQueue(int capacity, File file) throws IOException {
    Preconditions.checkArgument(capacity > 0,
        "Capacity must be greater than zero");
    this.queueCapacity = capacity;

    if (!file.exists()) {
      Preconditions.checkState(file.createNewFile(), "Unable to create file: "
          + file);
    }

    boolean freshlyAllocated = false;
    checkpointFile = new RandomAccessFile(file, "rw");
    if (checkpointFile.length() == 0) {
      // Allocate
      LOG.info("Event queue has zero allocation. Initializing to capacity. "
          + "Please wait...");
      int totalBytes = (capacity + HEADER_SIZE)*8;
      if (totalBytes <= MAX_ALLOC_BUFFER_SIZE) {
        checkpointFile.write(new byte[totalBytes]);
      } else {
        byte[] initBuffer = new byte[MAX_ALLOC_BUFFER_SIZE];
        int remainingBytes = totalBytes;
        while (remainingBytes >= MAX_ALLOC_BUFFER_SIZE) {
          checkpointFile.write(initBuffer);
          remainingBytes -= MAX_ALLOC_BUFFER_SIZE;
        }
        if (remainingBytes > 0) {
          checkpointFile.write(initBuffer, 0, remainingBytes);
        }
      }

      LOG.info("Event queue allocation complete");
      freshlyAllocated = true;
    } else {
      int fileCapacity = (int) checkpointFile.length() / 8;
      int expectedCapacity = capacity + HEADER_SIZE;

      Preconditions.checkState(fileCapacity == expectedCapacity,
          "Capacity cannot be reduced once the channel is initialized");
    }

    checkpointFileHandle = checkpointFile.getChannel();

    mappedBuffer = checkpointFileHandle.map(MapMode.READ_WRITE, 0,
        file.length());

{'fraction_non_alphanumeric': 0.07956104252400549, 'fraction_numerical': 0.00800182898948331, 'mean_word_length': 3.7678474114441416, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, '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': '1924623', 'n_tokens_mistral': 2414, 'n_tokens_neox': 2336, 'n_words': 856}