<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.jackrabbit.oak.plugins.segment;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.collect.Iterables.get;
import static com.google.common.collect.Sets.newConcurrentHashSet;
import static com.google.common.util.concurrent.Futures.addCallback;
import static com.google.common.util.concurrent.Futures.immediateCancelledFuture;
import static com.google.common.util.concurrent.MoreExecutors.listeningDecorator;
import static java.io.File.createTempFile;
import static java.lang.String.valueOf;
import static java.lang.System.getProperty;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.commons.io.FileUtils.deleteDirectory;
import static org.apache.commons.lang.RandomStringUtils.randomAlphabetic;
import static org.apache.jackrabbit.oak.plugins.segment.CompactionMap.sum;
import static org.apache.jackrabbit.oak.plugins.segment.compaction.CompactionStrategy.CleanupType.CLEAN_OLD;
import static org.apache.jackrabbit.oak.plugins.segment.compaction.CompactionStrategy.MEMORY_THRESHOLD_DEFAULT;
import static org.apache.jackrabbit.oak.plugins.segment.file.FileStore.newFileStore;
import static org.junit.Assume.assumeTrue;
import static org.slf4j.helpers.MessageFormatter.arrayFormat;
import static org.slf4j.helpers.MessageFormatter.format;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.lang.management.ManagementFactory;
import java.util.Date;
import java.util.Iterator;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.annotation.Nonnull;
import javax.management.InstanceAlreadyExistsException;
import javax.management.MBeanRegistrationException;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.NotCompliantMBeanException;
import javax.management.ObjectName;

import com.google.common.base.Predicate;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListenableScheduledFuture;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import org.apache.jackrabbit.oak.api.Blob;
import org.apache.jackrabbit.oak.api.CommitFailedException;
import org.apache.jackrabbit.oak.api.PropertyState;
import org.apache.jackrabbit.oak.commons.jmx.AnnotatedStandardMBean;
import org.apache.jackrabbit.oak.plugins.segment.compaction.CompactionStrategy;
import org.apache.jackrabbit.oak.plugins.segment.compaction.DefaultCompactionStrategyMBean;
import org.apache.jackrabbit.oak.plugins.segment.file.FileStore;
import org.apache.jackrabbit.oak.plugins.segment.file.FileStoreGCMonitor;
import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.apache.jackrabbit.oak.spi.state.NodeStore;
import org.apache.jackrabbit.oak.spi.whiteboard.CompositeRegistration;
import org.apache.jackrabbit.oak.spi.whiteboard.Registration;
import org.apache.jackrabbit.oak.stats.Clock;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This is a longeivity test for SegmentMK compaction. The test schedules a number
 * of readers, writers, a compactor and holds some references for a certain time.
 * All of which can be interactively modified through the accompanying
 * {@link SegmentCompactionITMBean}, the
 * {@link org.apache.jackrabbit.oak.plugins.segment.compaction.CompactionStrategyMBean} and the
 * {@link org.apache.jackrabbit.oak.plugins.segment.file.GCMonitorMBean}.
 *
 * TODO Leverage longeivity test support from OAK-2771 once we have it.
 */
public class SegmentCompactionIT {
    private static final boolean PERSIST_COMPACTION_MAP = Boolean.getBoolean("persist-compaction-map");

    /** Only run if explicitly asked to via -Dtest=SegmentCompactionIT */
    private static final boolean ENABLED =
            SegmentCompactionIT.class.getSimpleName().equals(getProperty("test"));
    private static final Logger LOG = LoggerFactory.getLogger(SegmentCompactionIT.class);

    private final MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();

    private final Random rnd = new Random();
    private final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(50);
    private final ListeningScheduledExecutorService scheduler = listeningDecorator(executor);
    private final FileStoreGCMonitor fileStoreGCMonitor = new FileStoreGCMonitor(Clock.SIMPLE);
    private final TestGCMonitor gcMonitor = new TestGCMonitor(fileStoreGCMonitor);
    private final Set<ListenableScheduledFuture<?>> writers = newConcurrentHashSet();
    private final Set<ListenableScheduledFuture<?>> readers = newConcurrentHashSet();
    private final Set<Reference> references = newConcurrentHashSet();
    private final SegmentCompactionITMBean segmentCompactionMBean = new SegmentCompactionITMBean();
    private final CompactionStrategy compactionStrategy = new CompactionStrategy(
            false, false, CLEAN_OLD, 60000, MEMORY_THRESHOLD_DEFAULT) {
        @Override
        public boolean compacted(@Nonnull Callable<Boolean> setHead) throws Exception {
            return nodeStore.locked(setHead, lockWaitTime, SECONDS);
        }
    };

    private File directory;
    private FileStore fileStore;
    private SegmentNodeStore nodeStore;
    private Registration mBeanRegistration;

    private volatile ListenableFuture<?> compactor = immediateCancelledFuture();
    private volatile int lockWaitTime = 60;
    private volatile int maxReaders = 10;
    private volatile int maxWriters = 10;
    private volatile long maxStoreSize = 200000000000L;
    private volatile int maxBlobSize = 1000000;
    private volatile int maxStringSize = 10000;
    private volatile int maxReferences = 10;
    private volatile int maxWriteOps = 10000;
    private volatile int maxNodeCount = 1000;
    private volatile int maxPropertyCount = 1000;
    private volatile int nodeRemoveRatio = 10;
    private volatile int propertyRemoveRatio = 10;
    private volatile int nodeAddRatio = 40;
    private volatile int addStringRatio = 20;
    private volatile int addBinaryRatio = 20;
    private volatile int compactionInterval = 1;
    private volatile boolean stopping;
    private volatile Reference rootReference;
    private volatile long fileStoreSize;

    public synchronized void stop() {
        stopping = true;
        notifyAll();
    }

    public void addReaders(int count) {
        for (int c = 0; c < count; c++) {
            scheduleReader();
        }
    }

    public void removeReaders(int count) {
        remove(readers, count);
    }

    public void addWriters(int count) {
        for (int c = 0; c < count; c++) {
            scheduleWriter();
        }
    }

    public void removeWriters(int count) {
        remove(writers, count);
    }

    public void removeReferences(int count) {
        Iterator<Reference> it = references.iterator();
        while (it.hasNext() && count-- > 0) {
            it.next().run();
            it.remove();
        }
    }

    private static void remove(Set<ListenableScheduledFuture<?>> futures, int count) {
        Iterator<ListenableScheduledFuture<?>> it = futures.iterator();
        while (it.hasNext() && count-- > 0) {
            it.next().cancel(false);
        }
    }

    private Registration registerMBean(Object mBean, final ObjectName objectName)
            throws NotCompliantMBeanException, InstanceAlreadyExistsException,
            MBeanRegistrationException {
        mBeanServer.registerMBean(mBean, objectName);
        return new Registration(){
            @Override
            public void unregister() {
                try {
                    mBeanServer.unregisterMBean(objectName);
                } catch (Exception e) {
                    LOG.error("Error unregistering Segment Compaction MBean", e);
                }
            }
        };
    }

    @Before
    public void setUp() throws IOException, MalformedObjectNameException, NotCompliantMBeanException,
            InstanceAlreadyExistsException, MBeanRegistrationException {
        assumeTrue(ENABLED);

        mBeanRegistration = new CompositeRegistration(
            registerMBean(segmentCompactionMBean, new ObjectName("IT:TYPE=Segment Compaction")),
            registerMBean(new DefaultCompactionStrategyMBean(compactionStrategy),
                    new ObjectName("IT:TYPE=Compaction Strategy")),
            registerMBean(fileStoreGCMonitor, new ObjectName("IT:TYPE=GC Monitor")));

        scheduler.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                fileStoreGCMonitor.run();
            }
        }, 1, 1, SECONDS);

        directory = createTempFile(getClass().getSimpleName(), "dir", new File("target"));
        directory.delete();
        directory.mkdir();

        fileStore = newFileStore(directory).withGCMonitor(gcMonitor).create();
        nodeStore = new SegmentNodeStore(fileStore);
        compactionStrategy.setPersistCompactionMap(PERSIST_COMPACTION_MAP);
        fileStore.setCompactionStrategy(compactionStrategy);
    }

    @After
    public void tearDown() {
        try {
            if (mBeanRegistration != null) {
                mBeanRegistration.unregister();
            }
            scheduler.shutdown();
            if (fileStore != null) {
                fileStore.close();
            }
            if (directory != null) {
                deleteDirectory(directory);
            }
        } catch (IOException e) {
            LOG.error("Error cleaning directory", e);
        }
    }

    @Test
    public void run() throws InterruptedException {
        scheduleSizeMonitor();
        scheduleCompactor();
        addReaders(maxReaders);
        addWriters(maxWriters);

        synchronized (this) {
            while (!stopping) {
                wait();
            }
        }
    }

    private void scheduleSizeMonitor() {
        scheduler.schedule(new Runnable() {
            @Override
            public void run() {
                fileStoreSize = fileStore.size();
            }
        }, 1, TimeUnit.MINUTES);
    }

    private synchronized void scheduleCompactor() {
        LOG.info("Scheduling compaction after {} minutes", compactionInterval);<commit_msg>OAK-2849: Improve revision gc on SegmentMK SegmentCompactionIT improvement: configurable core pool size and expose via MBean<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.jackrabbit.oak.plugins.segment;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.collect.Iterables.get;
import static com.google.common.collect.Sets.newConcurrentHashSet;
import static com.google.common.util.concurrent.Futures.addCallback;
import static com.google.common.util.concurrent.Futures.immediateCancelledFuture;
import static com.google.common.util.concurrent.MoreExecutors.listeningDecorator;
import static java.io.File.createTempFile;
import static java.lang.String.valueOf;
import static java.lang.System.getProperty;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.commons.io.FileUtils.deleteDirectory;
import static org.apache.commons.lang.RandomStringUtils.randomAlphabetic;
import static org.apache.jackrabbit.oak.plugins.segment.CompactionMap.sum;
import static org.apache.jackrabbit.oak.plugins.segment.compaction.CompactionStrategy.CleanupType.CLEAN_OLD;
import static org.apache.jackrabbit.oak.plugins.segment.compaction.CompactionStrategy.MEMORY_THRESHOLD_DEFAULT;
import static org.apache.jackrabbit.oak.plugins.segment.file.FileStore.newFileStore;
import static org.junit.Assume.assumeTrue;
import static org.slf4j.helpers.MessageFormatter.arrayFormat;
import static org.slf4j.helpers.MessageFormatter.format;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.lang.management.ManagementFactory;
import java.util.Date;
import java.util.Iterator;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ScheduledThreadPoolExecutor;

import javax.annotation.Nonnull;
import javax.management.InstanceAlreadyExistsException;
import javax.management.MBeanRegistrationException;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.NotCompliantMBeanException;
import javax.management.ObjectName;

import com.google.common.base.Predicate;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListenableScheduledFuture;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import org.apache.jackrabbit.oak.api.Blob;
import org.apache.jackrabbit.oak.api.CommitFailedException;
import org.apache.jackrabbit.oak.api.PropertyState;
import org.apache.jackrabbit.oak.commons.jmx.AnnotatedStandardMBean;
import org.apache.jackrabbit.oak.plugins.segment.compaction.CompactionStrategy;
import org.apache.jackrabbit.oak.plugins.segment.compaction.DefaultCompactionStrategyMBean;
import org.apache.jackrabbit.oak.plugins.segment.file.FileStore;
import org.apache.jackrabbit.oak.plugins.segment.file.FileStoreGCMonitor;
import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.apache.jackrabbit.oak.spi.state.NodeStore;
import org.apache.jackrabbit.oak.spi.whiteboard.CompositeRegistration;
import org.apache.jackrabbit.oak.spi.whiteboard.Registration;
import org.apache.jackrabbit.oak.stats.Clock;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This is a longeivity test for SegmentMK compaction. The test schedules a number
 * of readers, writers, a compactor and holds some references for a certain time.
 * All of which can be interactively modified through the accompanying
 * {@link SegmentCompactionITMBean}, the
 * {@link org.apache.jackrabbit.oak.plugins.segment.compaction.CompactionStrategyMBean} and the
 * {@link org.apache.jackrabbit.oak.plugins.segment.file.GCMonitorMBean}.
 *
 * TODO Leverage longeivity test support from OAK-2771 once we have it.
 */
public class SegmentCompactionIT {
    private static final boolean PERSIST_COMPACTION_MAP = Boolean.getBoolean("persist-compaction-map");

    /** Only run if explicitly asked to via -Dtest=SegmentCompactionIT */
    private static final boolean ENABLED =
            SegmentCompactionIT.class.getSimpleName().equals(getProperty("test"));
    private static final Logger LOG = LoggerFactory.getLogger(SegmentCompactionIT.class);

    private final MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();

    private final Random rnd = new Random();
    private final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(50);
    private final ListeningScheduledExecutorService scheduler = listeningDecorator(executor);
    private final FileStoreGCMonitor fileStoreGCMonitor = new FileStoreGCMonitor(Clock.SIMPLE);
    private final TestGCMonitor gcMonitor = new TestGCMonitor(fileStoreGCMonitor);
    private final Set<ListenableScheduledFuture<?>> writers = newConcurrentHashSet();
    private final Set<ListenableScheduledFuture<?>> readers = newConcurrentHashSet();
    private final Set<Reference> references = newConcurrentHashSet();
    private final SegmentCompactionITMBean segmentCompactionMBean = new SegmentCompactionITMBean();
    private final CompactionStrategy compactionStrategy = new CompactionStrategy(
            false, false, CLEAN_OLD, 60000, MEMORY_THRESHOLD_DEFAULT) {
        @Override
        public boolean compacted(@Nonnull Callable<Boolean> setHead) throws Exception {
            return nodeStore.locked(setHead, lockWaitTime, SECONDS);
        }
    };

    private File directory;
    private FileStore fileStore;
    private SegmentNodeStore nodeStore;
    private Registration mBeanRegistration;

    private volatile ListenableFuture<?> compactor = immediateCancelledFuture();
    private volatile int lockWaitTime = 60;
    private volatile int maxReaders = 10;
    private volatile int maxWriters = 10;
    private volatile long maxStoreSize = 200000000000L;
    private volatile int maxBlobSize = 1000000;
    private volatile int maxStringSize = 10000;
    private volatile int maxReferences = 10;
    private volatile int maxWriteOps = 10000;
    private volatile int maxNodeCount = 1000;
    private volatile int maxPropertyCount = 1000;
    private volatile int nodeRemoveRatio = 10;
    private volatile int propertyRemoveRatio = 10;
    private volatile int nodeAddRatio = 40;
    private volatile int addStringRatio = 20;
    private volatile int addBinaryRatio = 20;
    private volatile int compactionInterval = 1;
    private volatile boolean stopping;
    private volatile Reference rootReference;
    private volatile long fileStoreSize;

    public synchronized void stop() {
        stopping = true;
        notifyAll();
    }

    public void addReaders(int count) {
        for (int c = 0; c < count; c++) {
            scheduleReader();
        }
    }

    public void removeReaders(int count) {
        remove(readers, count);
    }

    public void addWriters(int count) {
        for (int c = 0; c < count; c++) {
            scheduleWriter();
        }
    }

    public void removeWriters(int count) {
        remove(writers, count);
    }

    public void removeReferences(int count) {
        Iterator<Reference> it = references.iterator();
        while (it.hasNext() && count-- > 0) {
            it.next().run();
            it.remove();
        }
    }

    private static void remove(Set<ListenableScheduledFuture<?>> futures, int count) {
        Iterator<ListenableScheduledFuture<?>> it = futures.iterator();
        while (it.hasNext() && count-- > 0) {
            it.next().cancel(false);
        }
    }

    private Registration registerMBean(Object mBean, final ObjectName objectName)
            throws NotCompliantMBeanException, InstanceAlreadyExistsException,
            MBeanRegistrationException {
        mBeanServer.registerMBean(mBean, objectName);
        return new Registration(){
            @Override
            public void unregister() {
                try {
                    mBeanServer.unregisterMBean(objectName);
                } catch (Exception e) {
                    LOG.error("Error unregistering Segment Compaction MBean", e);
                }
            }
        };
    }

    @Before
    public void setUp() throws IOException, MalformedObjectNameException, NotCompliantMBeanException,
            InstanceAlreadyExistsException, MBeanRegistrationException {
        assumeTrue(ENABLED);

        mBeanRegistration = new CompositeRegistration(
            registerMBean(segmentCompactionMBean, new ObjectName("IT:TYPE=Segment Compaction")),
            registerMBean(new DefaultCompactionStrategyMBean(compactionStrategy),
                    new ObjectName("IT:TYPE=Compaction Strategy")),
            registerMBean(fileStoreGCMonitor, new ObjectName("IT:TYPE=GC Monitor")));

        scheduler.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                fileStoreGCMonitor.run();
            }
        }, 1, 1, SECONDS);

        directory = createTempFile(getClass().getSimpleName(), "dir", new File("target"));
        directory.delete();
        directory.mkdir();

        fileStore = newFileStore(directory).withGCMonitor(gcMonitor).create();
        nodeStore = new SegmentNodeStore(fileStore);
        compactionStrategy.setPersistCompactionMap(PERSIST_COMPACTION_MAP);
        fileStore.setCompactionStrategy(compactionStrategy);
    }

    @After
    public void tearDown() {
        try {
            if (mBeanRegistration != null) {
                mBeanRegistration.unregister();
            }
            scheduler.shutdown();
            if (fileStore != null) {
                fileStore.close();
            }
            if (directory != null) {
                deleteDirectory(directory);
            }
        } catch (IOException e) {
            LOG.error("Error cleaning directory", e);
        }
    }

    @Test
    public void run() throws InterruptedException {
        scheduleSizeMonitor();
        scheduleCompactor();
        addReaders(maxReaders);
        addWriters(maxWriters);

        synchronized (this) {
            while (!stopping) {
                wait();
            }
        }
    }

    private void scheduleSizeMonitor() {
        scheduler.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                fileStoreSize = fileStore.size();
            }
        }, 1, 1, MINUTES);
    }

    private synchronized void scheduleCompactor() {
        LOG.info("Scheduling compaction after {} minutes", compactionInterval);
{'fraction_non_alphanumeric': 0.08643707482993197, 'fraction_numerical': 0.007185374149659864, 'mean_word_length': 3.590359094457455, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, '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': '2897600', 'n_tokens_mistral': 6445, 'n_tokens_neox': 6215, 'n_words': 1529}