<commit_before>/* Copyright (c) 2014-2016 Boundless and others.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Distribution License v1.0
 * which accompanies this distribution, and is available at
 * https://www.eclipse.org/org/documents/edl-v10.html
 *
 * Contributors:
 * <NAME> (Boundless) - initial implementation
 */
package org.locationtech.geogig.geotools.data.stresstest;

import static com.google.common.collect.ImmutableList.copyOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.geotools.data.DataUtilities;
import org.geotools.data.DefaultTransaction;
import org.geotools.data.Transaction;
import org.geotools.data.simple.SimpleFeatureCollection;
import org.geotools.data.simple.SimpleFeatureIterator;
import org.geotools.data.simple.SimpleFeatureSource;
import org.geotools.data.simple.SimpleFeatureStore;
import org.geotools.feature.SchemaException;
import org.geotools.feature.simple.SimpleFeatureBuilder;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.locationtech.geogig.geotools.data.GeoGigDataStore;
import org.locationtech.geogig.model.ObjectId;
import org.locationtech.geogig.model.RevCommit;
import org.locationtech.geogig.porcelain.ConfigOp;
import org.locationtech.geogig.porcelain.ConfigOp.ConfigAction;
import org.locationtech.geogig.porcelain.InitOp;
import org.locationtech.geogig.porcelain.LogOp;
import org.locationtech.geogig.repository.Context;
import org.locationtech.geogig.repository.Hints;
import org.locationtech.geogig.repository.impl.GeoGIG;
import org.locationtech.geogig.repository.impl.GlobalContextBuilder;
import org.locationtech.geogig.test.TestPlatform;
import org.locationtech.geogig.test.integration.TestContextBuilder;
import org.opengis.feature.simple.SimpleFeature;
import org.opengis.feature.simple.SimpleFeatureType;

import com.google.common.base.Optional;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.vividsolutions.jts.geom.Coordinate;
import com.vividsolutions.jts.geom.GeometryFactory;
import com.vividsolutions.jts.geom.Point;

public class DataStoreConcurrencyTest {

    private Context context;

    private GeoGigDataStore store;

    private static final SimpleFeatureType pointType;
    static {
        final String pointsTypeSpec = "sp:String,ip:Integer,pp:Point:srid=4326";
        try {
            pointType = DataUtilities.createType("point", pointsTypeSpec);
        } catch (SchemaException e) {
            throw Throwables.propagate(e);
        }
    }

    private ExecutorService editThreads;

    private ExecutorService readThreads;

    private final int writeThreadCount = 4, readThreadCount = 4;

    @Rule
    public TemporaryFolder tmp = new TemporaryFolder();

    private int initialCommitCount;

    @Before
    public void beforeTest() throws Exception {

        File workingDirectory = tmp.newFolder("repo");
        File userHomeDirectory = tmp.newFolder("home");
        TestPlatform platform = new TestPlatform(workingDirectory);
        platform.setUserHome(userHomeDirectory);

        GlobalContextBuilder.builder(new TestContextBuilder(platform));
        context = GlobalContextBuilder.builder().build(new Hints().platform(platform));

        GeoGIG repo = new GeoGIG(context);
        repo.command(InitOp.class).call();
        repo.command(ConfigOp.class).setAction(ConfigAction.CONFIG_SET).setName("user.name")
                .setValue("gabriel").call();
        repo.command(ConfigOp.class).setAction(ConfigAction.CONFIG_SET).setName("user.email")
                .setValue("<EMAIL>").call();

        store = new GeoGigDataStore(repo.getRepository());

        store.createSchema(pointType);

        editThreads = Executors.newFixedThreadPool(writeThreadCount,
                new ThreadFactoryBuilder().setNameFormat("edit-thread-%d").build());

        readThreads = Executors.newFixedThreadPool(readThreadCount,
                new ThreadFactoryBuilder().setNameFormat("read-thread-%d").build());
        initialCommitCount = copyOf(context.command(LogOp.class).call()).size();
    }

    @After
    public void afterTest() throws Exception {
        if (store != null) {
            store.dispose();
        }
        if (editThreads != null) {
            editThreads.shutdownNow();
        }
        if (readThreads != null) {
            readThreads.shutdownNow();
        }
    }

    @Test
    public void testConcurrentEdits() throws Exception {
        final int insertsPerTask = 20;
        List<Future<Integer>> insertResults = runInserts(writeThreadCount, insertsPerTask);
        for (Future<Integer> f : insertResults) {
            assertEquals(insertsPerTask, f.get().intValue());
        }

        List<RevCommit> commits = copyOf(context.command(LogOp.class).call());
        final int expectedCommitCount = initialCommitCount + insertsPerTask * writeThreadCount;
        assertEquals(expectedCommitCount, commits.size());
    }

    @Test
    public void testConcurrentReads() throws Exception {

        final int insertsPerTask = 20;
        assertEquals(insertsPerTask, runInserts(1, insertsPerTask).get(0).get().intValue());

        final int readsPerTask = 20;
        List<Future<Integer>> readResults = runReads(readThreadCount, readsPerTask);

        for (Future<Integer> f : readResults) {
            assertEquals(insertsPerTask * readsPerTask, f.get().intValue());
        }
    }

    @Test
    public void testConcurrentReadsWithIndex() throws Exception {

        final int insertsPerTask = 20;
        List<Future<Integer>> runInserts = runInserts(1, insertsPerTask);
        assertEquals(1, runInserts.size());
        Integer inserts = runInserts.get(0).get();
        assertEquals(insertsPerTask, inserts.intValue());

        // create an index
        Optional<ObjectId> createOrUpdateIndex = store.createOrUpdateIndex(pointType.getTypeName());
        assertTrue(createOrUpdateIndex.isPresent());
        ObjectId indexId = createOrUpdateIndex.get();
        assertNotNull(indexId);

        final int readsPerTask = 20;
        List<Future<Integer>> readResults = runReads(1, readsPerTask);
        for (Future<Integer> f : readResults) {
            assertEquals(insertsPerTask * readsPerTask, f.get().intValue());
        }
    }

    @Test
    public void testConcurrentEditsAndReads() throws Exception {

        final int insertsPerTask = 20;
        final int readsPerTask = 40;

        // have something to read
        runInserts(1, insertsPerTask).get(0).get();

        List<Future<Integer>> insertResults = runInserts(writeThreadCount, insertsPerTask);
        Thread.sleep(3000);
        List<Future<Integer>> readResults = runReads(readThreadCount, readsPerTask);

        for (Future<Integer> f : insertResults) {
            assertEquals(insertsPerTask, f.get().intValue());
        }
        for (Future<Integer> f : readResults) {
            assertEquals((insertsPerTask + writeThreadCount * insertsPerTask) * readsPerTask,
                    f.get().intValue());
        }

        List<RevCommit> commits = copyOf(context.command(LogOp.class).call());
        final int expectedCommitCount = insertsPerTask + initialCommitCount
                + insertsPerTask * writeThreadCount;
        assertEquals(expectedCommitCount, commits.size());
    }

    private List<Future<Integer>> runInserts(final int writeThreadCount, final int insertsPerTask) {
        List<Future<Integer>> insertResults = Lists.newArrayList();
        for (int i = 0; i < writeThreadCount; i++) {
            insertResults.add(editThreads.submit(new InsertTask(store, insertsPerTask)));<commit_msg>Ignore flawed test until it's rewritten to test what it's supposed to<commit_after>/* Copyright (c) 2014-2016 Boundless and others.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Distribution License v1.0
 * which accompanies this distribution, and is available at
 * https://www.eclipse.org/org/documents/edl-v10.html
 *
 * Contributors:
 * <NAME> (Boundless) - initial implementation
 */
package org.locationtech.geogig.geotools.data.stresstest;

import static com.google.common.collect.ImmutableList.copyOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.geotools.data.DataUtilities;
import org.geotools.data.DefaultTransaction;
import org.geotools.data.Transaction;
import org.geotools.data.simple.SimpleFeatureCollection;
import org.geotools.data.simple.SimpleFeatureIterator;
import org.geotools.data.simple.SimpleFeatureSource;
import org.geotools.data.simple.SimpleFeatureStore;
import org.geotools.feature.SchemaException;
import org.geotools.feature.simple.SimpleFeatureBuilder;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.locationtech.geogig.geotools.data.GeoGigDataStore;
import org.locationtech.geogig.model.ObjectId;
import org.locationtech.geogig.model.RevCommit;
import org.locationtech.geogig.porcelain.ConfigOp;
import org.locationtech.geogig.porcelain.ConfigOp.ConfigAction;
import org.locationtech.geogig.porcelain.InitOp;
import org.locationtech.geogig.porcelain.LogOp;
import org.locationtech.geogig.repository.Context;
import org.locationtech.geogig.repository.Hints;
import org.locationtech.geogig.repository.impl.GeoGIG;
import org.locationtech.geogig.repository.impl.GlobalContextBuilder;
import org.locationtech.geogig.test.TestPlatform;
import org.locationtech.geogig.test.integration.TestContextBuilder;
import org.opengis.feature.simple.SimpleFeature;
import org.opengis.feature.simple.SimpleFeatureType;

import com.google.common.base.Optional;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.vividsolutions.jts.geom.Coordinate;
import com.vividsolutions.jts.geom.GeometryFactory;
import com.vividsolutions.jts.geom.Point;

public class DataStoreConcurrencyTest {

    private Context context;

    private GeoGigDataStore store;

    private static final SimpleFeatureType pointType;
    static {
        final String pointsTypeSpec = "sp:String,ip:Integer,pp:Point:srid=4326";
        try {
            pointType = DataUtilities.createType("point", pointsTypeSpec);
        } catch (SchemaException e) {
            throw Throwables.propagate(e);
        }
    }

    private ExecutorService editThreads;

    private ExecutorService readThreads;

    private final int writeThreadCount = 4, readThreadCount = 4;

    @Rule
    public TemporaryFolder tmp = new TemporaryFolder();

    private int initialCommitCount;

    @Before
    public void beforeTest() throws Exception {

        File workingDirectory = tmp.newFolder("repo");
        File userHomeDirectory = tmp.newFolder("home");
        TestPlatform platform = new TestPlatform(workingDirectory);
        platform.setUserHome(userHomeDirectory);

        GlobalContextBuilder.builder(new TestContextBuilder(platform));
        context = GlobalContextBuilder.builder().build(new Hints().platform(platform));

        GeoGIG repo = new GeoGIG(context);
        repo.command(InitOp.class).call();
        repo.command(ConfigOp.class).setAction(ConfigAction.CONFIG_SET).setName("user.name")
                .setValue("gabriel").call();
        repo.command(ConfigOp.class).setAction(ConfigAction.CONFIG_SET).setName("user.email")
                .setValue("<EMAIL>").call();

        store = new GeoGigDataStore(repo.getRepository());

        store.createSchema(pointType);

        editThreads = Executors.newFixedThreadPool(writeThreadCount,
                new ThreadFactoryBuilder().setNameFormat("edit-thread-%d").build());

        readThreads = Executors.newFixedThreadPool(readThreadCount,
                new ThreadFactoryBuilder().setNameFormat("read-thread-%d").build());
        initialCommitCount = copyOf(context.command(LogOp.class).call()).size();
    }

    @After
    public void afterTest() throws Exception {
        if (store != null) {
            store.dispose();
        }
        if (editThreads != null) {
            editThreads.shutdownNow();
        }
        if (readThreads != null) {
            readThreads.shutdownNow();
        }
    }

    @Test
    public void testConcurrentEdits() throws Exception {
        final int insertsPerTask = 20;
        List<Future<Integer>> insertResults = runInserts(writeThreadCount, insertsPerTask);
        for (Future<Integer> f : insertResults) {
            assertEquals(insertsPerTask, f.get().intValue());
        }

        List<RevCommit> commits = copyOf(context.command(LogOp.class).call());
        final int expectedCommitCount = initialCommitCount + insertsPerTask * writeThreadCount;
        assertEquals(expectedCommitCount, commits.size());
    }

    @Test
    public void testConcurrentReads() throws Exception {

        final int insertsPerTask = 20;
        assertEquals(insertsPerTask, runInserts(1, insertsPerTask).get(0).get().intValue());

        final int readsPerTask = 20;
        List<Future<Integer>> readResults = runReads(readThreadCount, readsPerTask);

        for (Future<Integer> f : readResults) {
            assertEquals(insertsPerTask * readsPerTask, f.get().intValue());
        }
    }

    @Test
    public void testConcurrentReadsWithIndex() throws Exception {

        final int insertsPerTask = 20;
        List<Future<Integer>> runInserts = runInserts(1, insertsPerTask);
        assertEquals(1, runInserts.size());
        Integer inserts = runInserts.get(0).get();
        assertEquals(insertsPerTask, inserts.intValue());

        // create an index
        Optional<ObjectId> createOrUpdateIndex = store.createOrUpdateIndex(pointType.getTypeName());
        assertTrue(createOrUpdateIndex.isPresent());
        ObjectId indexId = createOrUpdateIndex.get();
        assertNotNull(indexId);

        final int readsPerTask = 20;
        List<Future<Integer>> readResults = runReads(1, readsPerTask);
        for (Future<Integer> f : readResults) {
            assertEquals(insertsPerTask * readsPerTask, f.get().intValue());
        }
    }

    @Ignore //this test is flawed (the test itself), ignoring until it's fixed
    @Test
    public void testConcurrentEditsAndReads() throws Exception {

        final int insertsPerTask = 20;
        final int readsPerTask = 40;

        // have something to read
        runInserts(1, insertsPerTask).get(0).get();

        List<Future<Integer>> insertResults = runInserts(writeThreadCount, insertsPerTask);
        Thread.sleep(3000);
        List<Future<Integer>> readResults = runReads(readThreadCount, readsPerTask);

        for (Future<Integer> f : insertResults) {
            assertEquals(insertsPerTask, f.get().intValue());
        }
        for (Future<Integer> f : readResults) {
            assertEquals((insertsPerTask + writeThreadCount * insertsPerTask) * readsPerTask,
                    f.get().intValue());
        }

        List<RevCommit> commits = copyOf(context.command(LogOp.class).call());
        final int expectedCommitCount = insertsPerTask + initialCommitCount
                + insertsPerTask * writeThreadCount;
        assertEquals(expectedCommitCount, commits.size());
    }

    private List<Future<Integer>> runInserts(final int writeThreadCount, final int insertsPerTask) {
        List<Future<Integer>> insertResults = Lists.newArrayList();
        for (int i = 0; i < writeThreadCount; i++) {
            insertResults.add(editThreads.submit(new InsertTask(store, insertsPerTask)));
{'fraction_non_alphanumeric': 0.1045625075638388, 'fraction_numerical': 0.005445963935616604, 'mean_word_length': 3.6424157303370785, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 55, 'https://': 2, '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': '3756943', 'n_tokens_mistral': 4729, 'n_tokens_neox': 4562, 'n_words': 904}