<commit_before>/* This file is part of VoltDB.
 * Copyright (C) 2008-2015 VoltDB Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */
package org.voltdb.utils;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;

import org.voltcore.logging.VoltLogger;
import org.voltcore.utils.Bits;
import org.voltcore.utils.DBBPool;
import org.voltcore.utils.DBBPool.BBContainer;
import org.voltcore.utils.DBBPool.MBBContainer;
import org.voltdb.utils.BinaryDeque.OutputContainerFactory;
import org.xerial.snappy.Snappy;

/**
 * Objects placed in the queue are stored in file segments that are up to 64 megabytes.
 * Segments only support appending objects. A segment will throw an IOException if an attempt
 * to insert an object that exceeds the remaining space is made. A segment can be used
 * for reading and writing, but not both at the same time.
 *
 */
class PBDSegment {
    private static final VoltLogger LOG = new VoltLogger("HOST");

    public static final int FLAG_COMPRESSED = 1;

    //Avoid unecessary sync with this flag
    private boolean m_syncedSinceLastEdit = true;
    final File m_file;
    private RandomAccessFile m_ras;
    private FileChannel m_fc;
    private MBBContainer m_buf;
    private ByteBuffer m_readBuf;

    //If this is the first time polling a segment, madvise the entire thing
    //into memory
    private boolean m_haveMAdvised = false;

    //Index of the next object to read, not an offset into the file
    //The offset is maintained by the ByteBuffer. Used to determine if there is another object
    int m_objectReadIndex = 0;
    private int m_bytesRead = 0;

    //ID of this segment
    final Long m_index;
    static final int m_chunkSize = (1024 * 1024) * 64;
    static final int m_objectHeaderBytes = 8;
    static final int m_segmentHeaderBytes = 8;

    static final int COUNT_OFFSET = 0;
    static final int SIZE_OFFSET = 4;

    private boolean m_closed = false;

    //How many entries that have been polled have from this file have been discarded.
    //Convenient to let PBQ maintain the counter here
    int m_discardCount = 0;

    public PBDSegment(Long index, File file ) {
        m_index = index;
        m_file = file;
        if (LOG.isDebugEnabled()) {
            LOG.debug("Creating Segment: " + file.getName() + " At Index: " + m_index);
        }
    }

    int getNumEntries() throws IOException {
        if (m_fc == null) {
            open(false);
        }
        if (m_fc.size() > m_segmentHeaderBytes) {
            final int numEntries = m_buf.b().getInt(0);
            return numEntries;
        } else {
            return 0;
        }
    }

    private void initNumEntries() throws IOException {
        final ByteBuffer buf = m_buf.b();
        buf.putInt(0, 0);
        buf.putInt(4, 0);
        m_syncedSinceLastEdit = false;
    }

    private void incrementNumEntries(int size) throws IOException {
        final ByteBuffer buf = m_buf.b();
        //First read the existing amount
        buf.putInt(COUNT_OFFSET, buf.getInt(COUNT_OFFSET) + 1);
        buf.putInt(SIZE_OFFSET, buf.getInt(SIZE_OFFSET) + size);
        m_syncedSinceLastEdit = false;
    }

    void open(boolean forWrite) throws IOException {
        if (!m_file.exists()) {
            m_syncedSinceLastEdit = false;
        }
        if (m_ras != null) {
            throw new IOException(m_file + " was already opened");
        }
        m_ras = new RandomAccessFile( m_file, "rw");
        m_fc = m_ras.getChannel();

        if (forWrite) {
            //If this is for writing, map the chunk size RW and put the buf positions at the start
            m_buf = DBBPool.wrapMBB(m_fc.map(MapMode.READ_WRITE, 0, m_chunkSize));
            m_buf.b().position(SIZE_OFFSET + 4);
            m_readBuf = m_buf.b().duplicate();
            initNumEntries();
        } else {
            //If it isn't for write, map read only to the actual size and put the write buf position at the end
            //so size is reported correctly
            final long size = m_fc.size();
            m_buf = DBBPool.wrapMBB(m_fc.map(MapMode.READ_ONLY, 0, size));
            m_readBuf = m_buf.b().duplicate();
            m_buf.b().position((int) size);
            m_readBuf.position(SIZE_OFFSET + 4);
        }
    }

    public void closeAndDelete() throws IOException {
        close();
        if (LOG.isDebugEnabled()) {
            LOG.debug("Deleting segment at Index " + m_index + " File: " + m_file.getAbsolutePath());
        }
        m_file.delete();
    }

    public void close() throws IOException {
        try {
            if (m_fc != null) {
                m_fc.close();
                m_ras = null;
                m_fc = null;
                m_buf.discard();
                m_buf = null;
                m_readBuf = null;
            }
        } finally {
            m_closed = true;
        }
    }

    void sync() throws IOException {
        if (m_closed) throw new IOException("closed");
        if (!m_syncedSinceLastEdit) {
            m_buf.b().force();
        }
        m_syncedSinceLastEdit = true;
    }

    boolean hasMoreEntries() throws IOException {
        if (m_closed) throw new IOException("closed");
        return m_objectReadIndex < m_buf.b().getInt(COUNT_OFFSET);
    }
<commit_msg>ENG-8237: Support reopening a closed PBD segment.<commit_after>/* This file is part of VoltDB.
 * Copyright (C) 2008-2015 VoltDB Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */
package org.voltdb.utils;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;

import org.voltcore.logging.VoltLogger;
import org.voltcore.utils.Bits;
import org.voltcore.utils.DBBPool;
import org.voltcore.utils.DBBPool.BBContainer;
import org.voltcore.utils.DBBPool.MBBContainer;
import org.voltdb.utils.BinaryDeque.OutputContainerFactory;
import org.xerial.snappy.Snappy;

/**
 * Objects placed in the queue are stored in file segments that are up to 64 megabytes.
 * Segments only support appending objects. A segment will throw an IOException if an attempt
 * to insert an object that exceeds the remaining space is made. A segment can be used
 * for reading and writing, but not both at the same time.
 *
 */
class PBDSegment {
    private static final VoltLogger LOG = new VoltLogger("HOST");

    public static final int FLAG_COMPRESSED = 1;

    //Avoid unecessary sync with this flag
    private boolean m_syncedSinceLastEdit;
    final File m_file;
    private RandomAccessFile m_ras;
    private FileChannel m_fc;
    private MBBContainer m_buf;
    private ByteBuffer m_readBuf;

    //If this is the first time polling a segment, madvise the entire thing
    //into memory
    private boolean m_haveMAdvised;

    //Index of the next object to read, not an offset into the file
    //The offset is maintained by the ByteBuffer. Used to determine if there is another object
    int m_objectReadIndex;
    private int m_bytesRead;

    //ID of this segment
    final Long m_index;
    static final int m_chunkSize = (1024 * 1024) * 64;
    static final int m_objectHeaderBytes = 8;
    static final int m_segmentHeaderBytes = 8;

    static final int COUNT_OFFSET = 0;
    static final int SIZE_OFFSET = 4;

    private boolean m_closed = true;

    //How many entries that have been polled have from this file have been discarded.
    //Convenient to let PBQ maintain the counter here
    int m_discardCount;

    public PBDSegment(Long index, File file ) {
        m_index = index;
        m_file = file;
        reset();
        if (LOG.isDebugEnabled()) {
            LOG.debug("Creating Segment: " + file.getName() + " At Index: " + m_index);
        }
    }

    private void reset() {
        m_syncedSinceLastEdit = true;
        m_haveMAdvised = false;
        m_objectReadIndex = 0;
        m_bytesRead = 0;
        m_discardCount = 0;
    }

    int getNumEntries() throws IOException {
        if (m_closed) {
            open(false);
        }
        if (m_fc.size() > m_segmentHeaderBytes) {
            final int numEntries = m_buf.b().getInt(0);
            return numEntries;
        } else {
            return 0;
        }
    }

    public boolean isBeingPolled() {
        return m_objectReadIndex != 0;
    }

    private void initNumEntries() throws IOException {
        final ByteBuffer buf = m_buf.b();
        buf.putInt(0, 0);
        buf.putInt(4, 0);
        m_syncedSinceLastEdit = false;
    }

    private void incrementNumEntries(int size) throws IOException {
        final ByteBuffer buf = m_buf.b();
        //First read the existing amount
        buf.putInt(COUNT_OFFSET, buf.getInt(COUNT_OFFSET) + 1);
        buf.putInt(SIZE_OFFSET, buf.getInt(SIZE_OFFSET) + size);
        m_syncedSinceLastEdit = false;
    }

    void open(boolean forWrite) throws IOException {
        if (!m_closed) {
            throw new IOException("Segment is already opened");
        }

        try {
            if (!m_file.exists()) {
                m_syncedSinceLastEdit = false;
            }
            if (m_ras != null) {
                throw new IOException(m_file + " was already opened");
            }
            m_ras = new RandomAccessFile(m_file, "rw");
            m_fc = m_ras.getChannel();

            if (forWrite) {
                //If this is for writing, map the chunk size RW and put the buf positions at the start
                m_buf = DBBPool.wrapMBB(m_fc.map(MapMode.READ_WRITE, 0, m_chunkSize));
                m_buf.b().position(SIZE_OFFSET + 4);
                m_readBuf = m_buf.b().duplicate();
                initNumEntries();
            } else {
                //If it isn't for write, map read only to the actual size and put the write buf position at the end
                //so size is reported correctly
                final long size = m_fc.size();
                m_buf = DBBPool.wrapMBB(m_fc.map(MapMode.READ_ONLY, 0, size));
                m_readBuf = m_buf.b().duplicate();
                m_buf.b().position((int) size);
                m_readBuf.position(SIZE_OFFSET + 4);
            }
        } finally {
            m_closed = false;
        }
    }

    public void closeAndDelete() throws IOException {
        close();
        if (LOG.isDebugEnabled()) {
            LOG.debug("Deleting segment at Index " + m_index + " File: " + m_file.getAbsolutePath());
        }
        m_file.delete();
    }

    public boolean isClosed() {
        return m_closed;
    }

    public void close() throws IOException {
        try {
            if (m_fc != null) {
                m_fc.close();
                m_ras = null;
                m_fc = null;
                m_buf.discard();
                m_buf = null;
                m_readBuf = null;
            }
        } finally {
            m_closed = true;
            reset();
        }
    }

    void sync() throws IOException {
        if (m_closed) throw new IOException("closed");
        if (!m_syncedSinceLastEdit) {
            m_buf.b().force();
        }
        m_syncedSinceLastEdit = true;
    }

    boolean hasMoreEntries() throws IOException {
        if (m_closed) throw new IOException("closed");
        return m_objectReadIndex < m_buf.b().getInt(COUNT_OFFSET);
    }

{'fraction_non_alphanumeric': 0.08640226628895184, 'fraction_numerical': 0.006688700031476235, 'mean_word_length': 2.288227684346701, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '925224', 'n_tokens_mistral': 3764, 'n_tokens_neox': 3599, 'n_words': 1215}