<commit_before>/*
 * Copyright 2016 <EMAIL>
 *
 * Licensed 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 net.openhft.chronicle.queue.impl.single;

import net.openhft.chronicle.bytes.MappedFile;
import net.openhft.chronicle.core.Jvm;
import net.openhft.chronicle.core.OS;
import net.openhft.chronicle.core.io.IOTools;
import net.openhft.chronicle.core.onoes.ExceptionKey;
import net.openhft.chronicle.core.onoes.LogLevel;
import net.openhft.chronicle.core.threads.ThreadDump;
import net.openhft.chronicle.core.time.SetTimeProvider;
import net.openhft.chronicle.queue.*;
import net.openhft.chronicle.queue.impl.RollingChronicleQueue;
import net.openhft.chronicle.wire.DocumentContext;
import org.jetbrains.annotations.NotNull;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.lang.reflect.Field;
import java.nio.file.NoSuchFileException;
import java.util.*;

import static junit.framework.Assert.assertEquals;
import static org.junit.Assert.*;

public class ToEndTest {
    long lastCycle;
    private ThreadDump threadDump;
    private Map<ExceptionKey, Integer> exceptionKeyIntegerMap;
    private List<File> pathsToDelete = new LinkedList<>();

    @Before
    public void before() {
        threadDump = new ThreadDump();
        threadDump.ignore(StoreComponentReferenceHandler.THREAD_NAME);
        threadDump.ignore("main/" + SingleChronicleQueue.DISK_SPACE_CHECKER_NAME);
        exceptionKeyIntegerMap = Jvm.recordExceptions();
    }

    @After
    public void after() {
        threadDump.assertNoNewThreads();

        if (Jvm.hasException(exceptionKeyIntegerMap)) {
            Jvm.dumpException(exceptionKeyIntegerMap);
            fail();
        }
        Jvm.resetExceptionHandlers();

        for (File file : pathsToDelete) {
            IOTools.shallowDeleteDirWithFiles(file);
        }
    }

    @Test
    public void missingCyclesToEndTest() throws InterruptedException {
        String path = OS.TARGET + "/missingCyclesToEndTest-" + System.nanoTime();
        IOTools.shallowDeleteDirWithFiles(path);

        final SetTimeProvider timeProvider = new SetTimeProvider();
        long now = 1470757797000L;
        long timeIncMs = 1001;
        timeProvider.currentTimeMillis(now);

        final RollingChronicleQueue queue = SingleChronicleQueueBuilder.binary(path)
                .testBlockSize()
                .rollCycle(RollCycles.TEST_SECONDLY)
                .timeProvider(timeProvider)
                .build();

        final ExcerptAppender appender = queue.acquireAppender();

        appender.writeDocument(wire -> wire.write(() -> "msg").int32(1));

        // roll
        timeProvider.currentTimeMillis(now += timeIncMs);

        appender.writeDocument(wire -> wire.write(() -> "msg").int32(2));
        appender.writeDocument(wire -> wire.write(() -> "msg").int32(3));

        final ExcerptTailer tailer = queue.createTailer().toEnd();
        try (DocumentContext dc = tailer.readingDocument()) {
            if (dc.isPresent()) {
                fail("Should be at the end of the queue but dc.isPresent and we read: " + String.valueOf(dc.wire().read(() -> "msg").int32()));
            }
        }

        // append same cycle.
        appender.writeDocument(wire -> wire.write(() -> "msg").int32(4));

        try (DocumentContext dc = tailer.readingDocument()) {
            assertTrue("Should be able to read entry in this cycle. Got NoDocumentContext.", dc.isPresent());
            int i = dc.wire().read(() -> "msg").int32();
            assertEquals("Should've read 4, instead we read: " + i, 4, i);
        }

        // read from the beginning
        tailer.toStart();

        for (int j = 1; j <= 4; j++) {
            try (DocumentContext dc = tailer.readingDocument()) {
                assertTrue(dc.isPresent());
                int i = dc.wire().read(() -> "msg").int32();
                assertEquals(j, i);
            }
        }

        try (DocumentContext dc = tailer.readingDocument()) {
            if (dc.isPresent()) {
                fail("Should be at the end of the queue but dc.isPresent and we read: " + String.valueOf(dc.wire().read(() -> "msg").int32()));
            }
        }

        // write another
        appender.writeDocument(wire -> wire.write(() -> "msg").int32(5));

        // roll 5 cycles
        timeProvider.currentTimeMillis(now += timeIncMs * 5);

        try (DocumentContext dc = tailer.readingDocument()) {
            assertTrue(dc.isPresent());
            assertEquals(5, dc.wire().read(() -> "msg").int32());
        }
        try (DocumentContext dc = tailer.readingDocument()) {
            assertFalse(dc.isPresent());
        }
    }

    @Test
    public void tailerToEndIncreasesRefCount() throws Exception {
        String path = OS.TARGET + "/toEndIncRefCount-" + System.nanoTime();
        IOTools.shallowDeleteDirWithFiles(path);

        SetTimeProvider time = new SetTimeProvider();
        long now = System.currentTimeMillis();
        time.currentTimeMillis(now);

        RollingChronicleQueue queue = SingleChronicleQueueBuilder.binary(path)
                .testBlockSize()
                .rollCycle(RollCycles.TEST_SECONDLY)
                .timeProvider(time)
                .build();

        final SingleChronicleQueueExcerpts.StoreAppender appender = (SingleChronicleQueueExcerpts.StoreAppender) queue.acquireAppender();
        Field storeF1 = SingleChronicleQueueExcerpts.StoreAppender.class.getDeclaredField("store");
        storeF1.setAccessible(true);
        SingleChronicleQueueStore store1 = (SingleChronicleQueueStore) storeF1.get(appender);
        System.out.println(store1);

        appender.writeDocument(wire -> wire.write(() -> "msg").int32(1));

        final SingleChronicleQueueExcerpts.StoreTailer tailer = (SingleChronicleQueueExcerpts.StoreTailer) queue.createTailer();
        System.out.println(tailer);
        tailer.toEnd();
        System.out.println(tailer);

        Field storeF2 = SingleChronicleQueueExcerpts.StoreTailer.class.getDeclaredField("store");
        storeF2.setAccessible(true);
        SingleChronicleQueueStore store2 = (SingleChronicleQueueStore) storeF2.get(tailer);

        // the reference count here is 1, the queue itself
        assertEquals(1, store2.refCount());
    }

    @Test
    public void toEndTest() {
        File baseDir = DirectoryUtils.tempDir("toEndTest");

        List<Integer> results = new ArrayList<>();
        final SingleChronicleQueue singleChronicleQueue = null;
        try (RollingChronicleQueue queue = SingleChronicleQueueBuilder.binary(baseDir)
                .testBlockSize()
                .rollCycle(RollCycles.TEST_DAILY)
                .indexCount(8)
                .indexSpacing(1)
                .build()) {

            checkOneFile(baseDir);
            ExcerptAppender appender = queue.acquireAppender();
            checkOneFile(baseDir);

            for (int i = 0; i < 10; i++) {
                final int j = i;
                appender.writeDocument(wire -> wire.write(() -> "msg").int32(j));
            }

            checkOneFile(baseDir);

            ExcerptTailer tailer = queue.createTailer();
            checkOneFile(baseDir);

            ExcerptTailer atEnd = tailer.toEnd();
            assertEquals(10, queue.rollCycle().toSequenceNumber(atEnd.index()));
            checkOneFile(baseDir);
            fillResults(atEnd, results);
            checkOneFile(baseDir);
            assertEquals(0, results.size());

            tailer.toStart();
            checkOneFile(baseDir);
            fillResults(tailer, results);
            assertEquals(10, results.size());
            checkOneFile(baseDir);
        }
        System.gc();
        pathsToDelete.add(baseDir);
    }

    @Test
    public void toEndBeforeWriteTest() {
        File baseDir = DirectoryUtils.tempDir("toEndBeforeWriteTest");
        IOTools.shallowDeleteDirWithFiles(baseDir);

        try (ChronicleQueue queue = SingleChronicleQueueBuilder.binary(baseDir)
                .testBlockSize()
                .build()) {
            checkOneFile(baseDir);

            // if this appender isn't created, the tailer toEnd doesn't cause a roll.
            ExcerptAppender appender = queue.acquireAppender();
            checkOneFile(baseDir);

            ExcerptTailer tailer = queue.createTailer();
            checkOneFile(baseDir);

            ExcerptTailer tailer2 = queue.createTailer();
            checkOneFile(baseDir);

            tailer.toEnd();
            checkOneFile(baseDir);

            tailer2.toEnd();
            checkOneFile(baseDir);
        }
        System.gc();

        /*for (int i = 0; i < 10; i++) {
            final int j = i;
            appender.writeDocument(wire -> wire.write(() -> "msg").int32(j));
        }*/
        pathsToDelete.add(baseDir);
    }

    @Test
    public void toEndAfterWriteTest() {
        File file = DirectoryUtils.tempDir("toEndAfterWriteTest");
        IOTools.shallowDeleteDirWithFiles(file);

        final SetTimeProvider stp = new SetTimeProvider();
        stp.currentTimeMillis(1470757797000L);

        try (RollingChronicleQueue wqueue = SingleChronicleQueueBuilder
                .binary(file)
                .testBlockSize()
                .rollCycle(RollCycles.TEST_SECONDLY)
                .timeProvider(stp)
                .build()) {
            ExcerptAppender appender = wqueue.acquireAppender();

            for (int i = 0; i < 10; i++) {
                try (DocumentContext dc = appender.writingDocument()) {
                    dc.wire().write().text("hi-" + i);
                    lastCycle = wqueue.rollCycle().toCycle(dc.index());
                }

                stp.currentTimeMillis(stp.currentTimeMillis() + 1000);
            }
        }

        try (ChronicleQueue rqueue = SingleChronicleQueueBuilder
                .binary(file)
                .testBlockSize()
                .rollCycle(RollCycles.TEST_SECONDLY)
                .timeProvider(stp)
                .build()) {

            ExcerptTailer tailer = rqueue.createTailer();
            stp.currentTimeMillis(stp.currentTimeMillis() + 1000);

            //noinspection StatementWithEmptyBody
            while (tailer.readText() != null) ;

            assertNull(tailer.readText());
            stp.currentTimeMillis(stp.currentTimeMillis() + 1000);

            ExcerptTailer tailer1 = rqueue.createTailer();
            ExcerptTailer excerptTailer = tailer1.toEnd();
            assertNull(excerptTailer.readText());
        }
        System.gc();
        pathsToDelete.add(file);
    }

    private void checkOneFile(@NotNull File baseDir) {
        String[] files = baseDir.list((d, n) -> n.endsWith(SingleChronicleQueue.SUFFIX));

        if (files == null || files.length == 0)
            return;

        if (files.length == 1)
            assertTrue(files[0], files[0].startsWith("2"));
        else
            fail("Too many files " + Arrays.toString(files));
    }

    @NotNull
    private List<Integer> fillResults(@NotNull ExcerptTailer tailer, @NotNull List<Integer> results) {
        for (int i = 0; i < 10; i++) {

            try (DocumentContext documentContext = tailer.readingDocument()) {
                if (!documentContext.isPresent())
                    break;
                results.add(documentContext.wire().read(() -> "msg").int32());
            }

        }
        return results;
    }

    @After
    public void checkMappedFiles() {
        System.gc();
        Jvm.pause(50);

        MappedFile.checkMappedFiles();
    }
}
<commit_msg>Attempt to beat asynchronousness - 2<commit_after>/*
 * Copyright 2016 <EMAIL>
 *
 * Licensed 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 net.openhft.chronicle.queue.impl.single;

import net.openhft.chronicle.bytes.MappedFile;
import net.openhft.chronicle.core.Jvm;
import net.openhft.chronicle.core.OS;
import net.openhft.chronicle.core.io.IOTools;
import net.openhft.chronicle.core.onoes.ExceptionKey;
import net.openhft.chronicle.core.onoes.LogLevel;
import net.openhft.chronicle.core.threads.ThreadDump;
import net.openhft.chronicle.core.time.SetTimeProvider;
import net.openhft.chronicle.queue.*;
import net.openhft.chronicle.queue.impl.RollingChronicleQueue;
import net.openhft.chronicle.wire.DocumentContext;
import org.jetbrains.annotations.NotNull;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.lang.reflect.Field;
import java.nio.file.NoSuchFileException;
import java.util.*;

import static junit.framework.Assert.assertEquals;
import static org.junit.Assert.*;

public class ToEndTest {
    long lastCycle;
    private ThreadDump threadDump;
    private Map<ExceptionKey, Integer> exceptionKeyIntegerMap;
    private List<File> pathsToDelete = new LinkedList<>();

    @Before
    public void before() {
        threadDump = new ThreadDump();
        threadDump.ignore(StoreComponentReferenceHandler.THREAD_NAME);
        threadDump.ignore("main/" + SingleChronicleQueue.DISK_SPACE_CHECKER_NAME);
        exceptionKeyIntegerMap = Jvm.recordExceptions();
    }

    @After
    public void after() {
        threadDump.assertNoNewThreads();

        if (Jvm.hasException(exceptionKeyIntegerMap)) {
            Jvm.dumpException(exceptionKeyIntegerMap);
            fail();
        }
        Jvm.resetExceptionHandlers();
    }

    @AfterClass
    public void afterClass() {
        for (File file : pathsToDelete) {
            IOTools.shallowDeleteDirWithFiles(file);
        }
    }

    @Test
    public void missingCyclesToEndTest() throws InterruptedException {
        String path = OS.TARGET + "/missingCyclesToEndTest-" + System.nanoTime();
        IOTools.shallowDeleteDirWithFiles(path);

        final SetTimeProvider timeProvider = new SetTimeProvider();
        long now = 1470757797000L;
        long timeIncMs = 1001;
        timeProvider.currentTimeMillis(now);

        final RollingChronicleQueue queue = SingleChronicleQueueBuilder.binary(path)
                .testBlockSize()
                .rollCycle(RollCycles.TEST_SECONDLY)
                .timeProvider(timeProvider)
                .build();

        final ExcerptAppender appender = queue.acquireAppender();

        appender.writeDocument(wire -> wire.write(() -> "msg").int32(1));

        // roll
        timeProvider.currentTimeMillis(now += timeIncMs);

        appender.writeDocument(wire -> wire.write(() -> "msg").int32(2));
        appender.writeDocument(wire -> wire.write(() -> "msg").int32(3));

        final ExcerptTailer tailer = queue.createTailer().toEnd();
        try (DocumentContext dc = tailer.readingDocument()) {
            if (dc.isPresent()) {
                fail("Should be at the end of the queue but dc.isPresent and we read: " + String.valueOf(dc.wire().read(() -> "msg").int32()));
            }
        }

        // append same cycle.
        appender.writeDocument(wire -> wire.write(() -> "msg").int32(4));

        try (DocumentContext dc = tailer.readingDocument()) {
            assertTrue("Should be able to read entry in this cycle. Got NoDocumentContext.", dc.isPresent());
            int i = dc.wire().read(() -> "msg").int32();
            assertEquals("Should've read 4, instead we read: " + i, 4, i);
        }

        // read from the beginning
        tailer.toStart();

        for (int j = 1; j <= 4; j++) {
            try (DocumentContext dc = tailer.readingDocument()) {
                assertTrue(dc.isPresent());
                int i = dc.wire().read(() -> "msg").int32();
                assertEquals(j, i);
            }
        }

        try (DocumentContext dc = tailer.readingDocument()) {
            if (dc.isPresent()) {
                fail("Should be at the end of the queue but dc.isPresent and we read: " + String.valueOf(dc.wire().read(() -> "msg").int32()));
            }
        }

        // write another
        appender.writeDocument(wire -> wire.write(() -> "msg").int32(5));

        // roll 5 cycles
        timeProvider.currentTimeMillis(now += timeIncMs * 5);

        try (DocumentContext dc = tailer.readingDocument()) {
            assertTrue(dc.isPresent());
            assertEquals(5, dc.wire().read(() -> "msg").int32());
        }
        try (DocumentContext dc = tailer.readingDocument()) {
            assertFalse(dc.isPresent());
        }
    }

    @Test
    public void tailerToEndIncreasesRefCount() throws Exception {
        String path = OS.TARGET + "/toEndIncRefCount-" + System.nanoTime();
        IOTools.shallowDeleteDirWithFiles(path);

        SetTimeProvider time = new SetTimeProvider();
        long now = System.currentTimeMillis();
        time.currentTimeMillis(now);

        RollingChronicleQueue queue = SingleChronicleQueueBuilder.binary(path)
                .testBlockSize()
                .rollCycle(RollCycles.TEST_SECONDLY)
                .timeProvider(time)
                .build();

        final SingleChronicleQueueExcerpts.StoreAppender appender = (SingleChronicleQueueExcerpts.StoreAppender) queue.acquireAppender();
        Field storeF1 = SingleChronicleQueueExcerpts.StoreAppender.class.getDeclaredField("store");
        storeF1.setAccessible(true);
        SingleChronicleQueueStore store1 = (SingleChronicleQueueStore) storeF1.get(appender);
        System.out.println(store1);

        appender.writeDocument(wire -> wire.write(() -> "msg").int32(1));

        final SingleChronicleQueueExcerpts.StoreTailer tailer = (SingleChronicleQueueExcerpts.StoreTailer) queue.createTailer();
        System.out.println(tailer);
        tailer.toEnd();
        System.out.println(tailer);

        Field storeF2 = SingleChronicleQueueExcerpts.StoreTailer.class.getDeclaredField("store");
        storeF2.setAccessible(true);
        SingleChronicleQueueStore store2 = (SingleChronicleQueueStore) storeF2.get(tailer);

        // the reference count here is 1, the queue itself
        assertEquals(1, store2.refCount());
    }

    @Test
    public void toEndTest() {
        File baseDir = DirectoryUtils.tempDir("toEndTest");

        List<Integer> results = new ArrayList<>();
        final SingleChronicleQueue singleChronicleQueue = null;
        try (RollingChronicleQueue queue = SingleChronicleQueueBuilder.binary(baseDir)
                .testBlockSize()
                .rollCycle(RollCycles.TEST_DAILY)
                .indexCount(8)
                .indexSpacing(1)
                .build()) {

            checkOneFile(baseDir);
            ExcerptAppender appender = queue.acquireAppender();
            checkOneFile(baseDir);

            for (int i = 0; i < 10; i++) {
                final int j = i;
                appender.writeDocument(wire -> wire.write(() -> "msg").int32(j));
            }

            checkOneFile(baseDir);

            ExcerptTailer tailer = queue.createTailer();
            checkOneFile(baseDir);

            ExcerptTailer atEnd = tailer.toEnd();
            assertEquals(10, queue.rollCycle().toSequenceNumber(atEnd.index()));
            checkOneFile(baseDir);
            fillResults(atEnd, results);
            checkOneFile(baseDir);
            assertEquals(0, results.size());

            tailer.toStart();
            checkOneFile(baseDir);
            fillResults(tailer, results);
            assertEquals(10, results.size());
            checkOneFile(baseDir);
        }
        System.gc();
        pathsToDelete.add(baseDir);
    }

    @Test
    public void toEndBeforeWriteTest() {
        File baseDir = DirectoryUtils.tempDir("toEndBeforeWriteTest");
        IOTools.shallowDeleteDirWithFiles(baseDir);

        try (ChronicleQueue queue = SingleChronicleQueueBuilder.binary(baseDir)
                .testBlockSize()
                .build()) {
            checkOneFile(baseDir);

            // if this appender isn't created, the tailer toEnd doesn't cause a roll.
            ExcerptAppender appender = queue.acquireAppender();
            checkOneFile(baseDir);

            ExcerptTailer tailer = queue.createTailer();
            checkOneFile(baseDir);

            ExcerptTailer tailer2 = queue.createTailer();
            checkOneFile(baseDir);

            tailer.toEnd();
            checkOneFile(baseDir);

            tailer2.toEnd();
            checkOneFile(baseDir);
        }
        System.gc();

        /*for (int i = 0; i < 10; i++) {
            final int j = i;
            appender.writeDocument(wire -> wire.write(() -> "msg").int32(j));
        }*/
        pathsToDelete.add(baseDir);
    }

    @Test
    public void toEndAfterWriteTest() {
        File file = DirectoryUtils.tempDir("toEndAfterWriteTest");
        IOTools.shallowDeleteDirWithFiles(file);

        final SetTimeProvider stp = new SetTimeProvider();
        stp.currentTimeMillis(1470757797000L);

        try (RollingChronicleQueue wqueue = SingleChronicleQueueBuilder
                .binary(file)
                .testBlockSize()
                .rollCycle(RollCycles.TEST_SECONDLY)
                .timeProvider(stp)
                .build()) {
            ExcerptAppender appender = wqueue.acquireAppender();

            for (int i = 0; i < 10; i++) {
                try (DocumentContext dc = appender.writingDocument()) {
                    dc.wire().write().text("hi-" + i);
                    lastCycle = wqueue.rollCycle().toCycle(dc.index());
                }

                stp.currentTimeMillis(stp.currentTimeMillis() + 1000);
            }
        }

        try (ChronicleQueue rqueue = SingleChronicleQueueBuilder
                .binary(file)
                .testBlockSize()
                .rollCycle(RollCycles.TEST_SECONDLY)
                .timeProvider(stp)
                .build()) {

            ExcerptTailer tailer = rqueue.createTailer();
            stp.currentTimeMillis(stp.currentTimeMillis() + 1000);

            //noinspection StatementWithEmptyBody
            while (tailer.readText() != null) ;

            assertNull(tailer.readText());
            stp.currentTimeMillis(stp.currentTimeMillis() + 1000);

            ExcerptTailer tailer1 = rqueue.createTailer();
            ExcerptTailer excerptTailer = tailer1.toEnd();
            assertNull(excerptTailer.readText());
        }
        System.gc();
        pathsToDelete.add(file);
    }

    private void checkOneFile(@NotNull File baseDir) {
        String[] files = baseDir.list((d, n) -> n.endsWith(SingleChronicleQueue.SUFFIX));

        if (files == null || files.length == 0)
            return;

        if (files.length == 1)
            assertTrue(files[0], files[0].startsWith("2"));
        else
            fail("Too many files " + Arrays.toString(files));
    }

    @NotNull
    private List<Integer> fillResults(@NotNull ExcerptTailer tailer, @NotNull List<Integer> results) {
        for (int i = 0; i < 10; i++) {

            try (DocumentContext documentContext = tailer.readingDocument()) {
                if (!documentContext.isPresent())
                    break;
                results.add(documentContext.wire().read(() -> "msg").int32());
            }

        }
        return results;
    }

    @After
    public void checkMappedFiles() {
        System.gc();
        Jvm.pause(50);

        MappedFile.checkMappedFiles();
    }
}

{'fraction_non_alphanumeric': 0.11420955505433668, 'fraction_numerical': 0.010949354111133893, 'mean_word_length': 2.3065762711864406, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 65, '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': '5034719', 'n_tokens_mistral': 7206, 'n_tokens_neox': 6777, 'n_words': 1370}