src/test/java/com/google/devtools/build/lib/graph/DigraphConcurrentTest.java - bazel - Git at Google

 // Copyright 2015 The Bazel Authors. All rights reserved.
 //
 // 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 com.google.devtools.build.lib.graph;

 import static com.google.common.truth.Truth.assertThat;
 import static java.util.Objects.requireNonNull;
 import static org.junit.Assert.fail;

 import com.google.common.base.Preconditions;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Stream;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;

 /**
  * Tests that DiGraph after concurrent access is in consistent state.
  *
  * <p>Inconsistent state means that any of edges is half added.
  */
 @RunWith(JUnit4.class)
 public class DigraphConcurrentTest {

   private static final Random RANDOM = new Random();

   // need to have contention.
   private static final int THREAD_COUNT = Runtime.getRuntime().availableProcessors() * 3;

   /** Created 100_000 nodes randomly. Adds 10 outgoing and 10 incoming edges for every node. */
   @Test
   public void testValidStateOfGraphAfterConcurrentAccess() throws InterruptedException {
     Digraph<Integer> digraph = new Digraph<>();
     final int numberOfNodes = 100_000;

     Runnable workerBoth10 = new CreateNodeWorker(digraph, 10, 10, new AtomicInteger());

     runWorkers(numberOfNodes, workerBoth10);
     assertThat(digraph.getNodeCount()).isEqualTo(numberOfNodes);

     assertGraphIsInConsistentState(digraph);
   }

   /**
    * Created 10_000 nodes randomly. Adds different number outgoing and incoming edges for every
    * node.
    */
   @Test
   public void testValidStateOfGraphAfterConcurrentAccessWithVariertyOfNodes()
       throws InterruptedException {

     Digraph<Integer> digraph = new Digraph<>();
     final int numberOfNodes = 50_000;
     final AtomicInteger idGenerator = new AtomicInteger();

     Runnable workerBoth3 = new CreateNodeWorker(digraph, 3, 3, idGenerator);
     Runnable workerBoth10 = new CreateNodeWorker(digraph, 10, 10, idGenerator);
     Runnable workerBoth100 = new CreateNodeWorker(digraph, 100, 100, idGenerator);
     Runnable workerOutgoing20 = new CreateNodeWorker(digraph, 20, 0, idGenerator);
     Runnable workerIncoming20 = new CreateNodeWorker(digraph, 0, 20, idGenerator);

     runWorkers(
         numberOfNodes / 5, // have already 5 workers
         workerBoth3,
         workerBoth10,
         workerBoth100,
         workerIncoming20,
         workerOutgoing20);

     assertThat(digraph.getNodeCount()).isEqualTo(numberOfNodes);
     assertGraphIsInConsistentState(digraph);
   }

   /**
    * Creates 20_000 nodes. Then iterate over node and for every node does: Creates 100 tasks: 50
    * tasks for adding 10 outgoing edges and 50 tasks for 10 incoming edges per each tasks. Then
    * executes all tasks in parallel in high contention. Only after all tasks for this node have
    * finished, switch to next node. this behaviour introduces very high contention around one single
    * node.
    */
   @Test
   public void testAddEdgeWithHighContention()
       throws InterruptedException, ExecutionException, TimeoutException {

     final int numberOfNodes = 20_000;
     final int edgePerNode = 1000;

     Digraph<Integer> digraph = new Digraph<>();
     CreateNodeWorker workerZeroEdges = new CreateNodeWorker(digraph, 0, 0, new AtomicInteger());
     runWorkers(numberOfNodes, workerZeroEdges);
     assertThat(digraph.getNodeCount()).isEqualTo(numberOfNodes);

     ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);

     for (Node<Integer> node : digraph.getNodes()) {

       // TODO(dbabkin): think about moving this interrupt callback logic to common library or make
       // research, may be one exists already in any open source project.
       AtomicBoolean isTestInterrupted = new AtomicBoolean(false);
       Runnable interruptedCallBack =
           () -> {
             isTestInterrupted.set(true);
             executorService.shutdownNow();
           };

       CountDownLatch countDownLatch = new CountDownLatch(1);
       List<Future<?>> futures = new ArrayList<>();
       // fork 100 tasks executed in parallel in high contention for one node:
       // 100 =  50 tasks for adding 10 outgoing edges and 50 tasks for adding 10 incoming edges
       for (int i = 0; i < edgePerNode / 20; i++) {
         // add 10 outgoing edges
         Runnable add10Outgoing =
             new CreateEdgeNightContention(
                 digraph, countDownLatch, node.getLabel(), 10, true, interruptedCallBack);
         futures.add(executorService.submit(add10Outgoing));

         // add 10 incoming edges
         Runnable add10Incoming =
             new CreateEdgeNightContention(
                 digraph, countDownLatch, node.getLabel(), 10, false, interruptedCallBack);
         futures.add(executorService.submit(add10Incoming));
       }

       // red lights go out, race has begun. http://www.formula1-dictionary.net/start_sequence.html
       countDownLatch.countDown();

       // wait for every tasks completed before switch to the next node.
       for (Future<?> future : futures) {
         if (isTestInterrupted.get()) {
           break;
         }
         future.get(1, TimeUnit.MINUTES);
       }
       if (isTestInterrupted.get()) {
         fail("Test had been interrupted.");
       }
     }

     assertGraphIsInConsistentState(digraph);
   }

   private void runWorkers(int count, Runnable... workers) throws InterruptedException {

     ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);

     for (int i = 0; i < count; i++) {
       Stream.of(workers).forEach(executorService::execute);
     }

     executorService.shutdown();
     while (!executorService.isTerminated()) {
       if (!executorService.awaitTermination(1, TimeUnit.HOURS)) {
         throw new IllegalStateException("executor service termination wait time out");
       }
     }
   }

   /** Asserts that there are no half added edge in the graph. */
   private void assertGraphIsInConsistentState(Digraph<Integer> digraph) {
     for (Node<Integer> node : digraph.getNodes()) {
       assertNodeIsInConsistentState(node);
     }
   }

   private void assertNodeIsInConsistentState(Node<Integer> node) {
     for (Node<Integer> succ : node.getSuccessors()) {
       assertThat(succ.getPredecessors()).contains(node);
     }

     for (Node<Integer> pred : node.getPredecessors()) {
       assertThat(pred.getSuccessors()).contains(node);
     }
   }

   private static final class CreateNodeWorker implements Runnable {

     private final Digraph<Integer> digraph;
     private final int outgoingEdgesPerNode;
     private final int incomingEdgesPerNode;
     private final AtomicInteger idGenerator;

     private CreateNodeWorker(
         Digraph<Integer> digraph,
         int outgoingEdgesPerNode,
         int incomingEdgesPerNode,
         AtomicInteger idGenerator) {
       this.digraph = requireNonNull(digraph);
       Preconditions.checkArgument(outgoingEdgesPerNode >= 0);
       this.outgoingEdgesPerNode = outgoingEdgesPerNode;
       Preconditions.checkArgument(incomingEdgesPerNode >= 0);
       this.incomingEdgesPerNode = incomingEdgesPerNode;
       this.idGenerator = idGenerator;
     }

     @Override
     public void run() {
       Integer newNodeId = idGenerator.getAndIncrement();
       digraph.createNode(newNodeId);
       addEdgesFromNew(newNodeId);
       addEdgesToNew(newNodeId);
     }

     private void addEdgesFromNew(int newNodeId) {
       int count = Math.min(newNodeId, outgoingEdgesPerNode);
       addEdges(digraph, newNodeId, count, /*outgoing*/ true);
     }

     private void addEdgesToNew(int newNodeId) {
       int count = Math.min(newNodeId, incomingEdgesPerNode);
       addEdges(digraph, newNodeId, count, /*outgoing*/ false);
     }
   }

   private static final class CreateEdgeNightContention implements Runnable {

     private final Digraph<Integer> digraph;
     private final CountDownLatch countDownLatch;
     private final int nodeId;
     private final int numberEdgeToAdd;
     private final boolean outgoing;
     private final Runnable inerruptCallBack;

     private CreateEdgeNightContention(
         Digraph<Integer> digraph,
         CountDownLatch countDownLatch,
         int nodeId,
         int numberEdgeToAdd,
         boolean outgoing,
         Runnable inerruptCallBack) {
       this.digraph = requireNonNull(digraph);
       this.countDownLatch = requireNonNull(countDownLatch);
       Preconditions.checkArgument(nodeId >= 0);
       this.nodeId = nodeId;
       Preconditions.checkArgument(numberEdgeToAdd >= 0);
       this.numberEdgeToAdd = numberEdgeToAdd;
       this.outgoing = outgoing;
       this.inerruptCallBack = requireNonNull(inerruptCallBack);
     }

     @Override
     public void run() {
       try {
         countDownLatch.await();
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
         inerruptCallBack.run();
         // hardly ever will see that in the log. Who cares?
         fail(e.getMessage());
       }
       addEdges(digraph, nodeId, numberEdgeToAdd, outgoing);
     }
   }

   private static void addEdges(Digraph<Integer> digraph, int nodeId, int count, boolean outgoing) {

     for (int i = 0; i < count; i++) {
       int id = RANDOM.nextInt(digraph.getNodeCount());
       if (outgoing) {
         digraph.addEdge(nodeId, id);
       } else {
         digraph.addEdge(id, nodeId);
       }
     }
   }
 }
	// Copyright 2015 The Bazel Authors. All rights reserved.
	//
	// 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 com.google.devtools.build.lib.graph;

	import static com.google.common.truth.Truth.assertThat;
	import static java.util.Objects.requireNonNull;
	import static org.junit.Assert.fail;

	import com.google.common.base.Preconditions;
	import java.util.ArrayList;
	import java.util.List;
	import java.util.Random;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.ExecutionException;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.Executors;
	import java.util.concurrent.Future;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.TimeoutException;
	import java.util.concurrent.atomic.AtomicBoolean;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.stream.Stream;
	import org.junit.Test;
	import org.junit.runner.RunWith;
	import org.junit.runners.JUnit4;

	/**
	* Tests that DiGraph after concurrent access is in consistent state.
	*
	* <p>Inconsistent state means that any of edges is half added.
	*/
	@RunWith(JUnit4.class)
	public class DigraphConcurrentTest {

	private static final Random RANDOM = new Random();

	// need to have contention.
	private static final int THREAD_COUNT = Runtime.getRuntime().availableProcessors() * 3;

	/** Created 100_000 nodes randomly. Adds 10 outgoing and 10 incoming edges for every node. */
	@Test
	public void testValidStateOfGraphAfterConcurrentAccess() throws InterruptedException {
	Digraph<Integer> digraph = new Digraph<>();
	final int numberOfNodes = 100_000;

	Runnable workerBoth10 = new CreateNodeWorker(digraph, 10, 10, new AtomicInteger());

	runWorkers(numberOfNodes, workerBoth10);
	assertThat(digraph.getNodeCount()).isEqualTo(numberOfNodes);

	assertGraphIsInConsistentState(digraph);
	}

	/**
	* Created 10_000 nodes randomly. Adds different number outgoing and incoming edges for every
	* node.
	*/
	@Test
	public void testValidStateOfGraphAfterConcurrentAccessWithVariertyOfNodes()
	throws InterruptedException {

	Digraph<Integer> digraph = new Digraph<>();
	final int numberOfNodes = 50_000;
	final AtomicInteger idGenerator = new AtomicInteger();

	Runnable workerBoth3 = new CreateNodeWorker(digraph, 3, 3, idGenerator);
	Runnable workerBoth10 = new CreateNodeWorker(digraph, 10, 10, idGenerator);
	Runnable workerBoth100 = new CreateNodeWorker(digraph, 100, 100, idGenerator);
	Runnable workerOutgoing20 = new CreateNodeWorker(digraph, 20, 0, idGenerator);
	Runnable workerIncoming20 = new CreateNodeWorker(digraph, 0, 20, idGenerator);

	runWorkers(
	numberOfNodes / 5, // have already 5 workers
	workerBoth3,
	workerBoth10,
	workerBoth100,
	workerIncoming20,
	workerOutgoing20);

	assertThat(digraph.getNodeCount()).isEqualTo(numberOfNodes);
	assertGraphIsInConsistentState(digraph);
	}

	/**
	* Creates 20_000 nodes. Then iterate over node and for every node does: Creates 100 tasks: 50
	* tasks for adding 10 outgoing edges and 50 tasks for 10 incoming edges per each tasks. Then
	* executes all tasks in parallel in high contention. Only after all tasks for this node have
	* finished, switch to next node. this behaviour introduces very high contention around one single
	* node.
	*/
	@Test
	public void testAddEdgeWithHighContention()
	throws InterruptedException, ExecutionException, TimeoutException {

	final int numberOfNodes = 20_000;
	final int edgePerNode = 1000;

	Digraph<Integer> digraph = new Digraph<>();
	CreateNodeWorker workerZeroEdges = new CreateNodeWorker(digraph, 0, 0, new AtomicInteger());
	runWorkers(numberOfNodes, workerZeroEdges);
	assertThat(digraph.getNodeCount()).isEqualTo(numberOfNodes);

	ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);

	for (Node<Integer> node : digraph.getNodes()) {

	// TODO(dbabkin): think about moving this interrupt callback logic to common library or make
	// research, may be one exists already in any open source project.
	AtomicBoolean isTestInterrupted = new AtomicBoolean(false);
	Runnable interruptedCallBack =
	() -> {
	isTestInterrupted.set(true);
	executorService.shutdownNow();
	};

	CountDownLatch countDownLatch = new CountDownLatch(1);
	List<Future<?>> futures = new ArrayList<>();
	// fork 100 tasks executed in parallel in high contention for one node:
	// 100 = 50 tasks for adding 10 outgoing edges and 50 tasks for adding 10 incoming edges
	for (int i = 0; i < edgePerNode / 20; i++) {
	// add 10 outgoing edges
	Runnable add10Outgoing =
	new CreateEdgeNightContention(
	digraph, countDownLatch, node.getLabel(), 10, true, interruptedCallBack);
	futures.add(executorService.submit(add10Outgoing));

	// add 10 incoming edges
	Runnable add10Incoming =
	new CreateEdgeNightContention(
	digraph, countDownLatch, node.getLabel(), 10, false, interruptedCallBack);
	futures.add(executorService.submit(add10Incoming));
	}

	// red lights go out, race has begun. http://www.formula1-dictionary.net/start_sequence.html
	countDownLatch.countDown();

	// wait for every tasks completed before switch to the next node.
	for (Future<?> future : futures) {
	if (isTestInterrupted.get()) {
	break;
	}
	future.get(1, TimeUnit.MINUTES);
	}
	if (isTestInterrupted.get()) {
	fail("Test had been interrupted.");
	}
	}

	assertGraphIsInConsistentState(digraph);
	}

	private void runWorkers(int count, Runnable... workers) throws InterruptedException {

	ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);

	for (int i = 0; i < count; i++) {
	Stream.of(workers).forEach(executorService::execute);
	}

	executorService.shutdown();
	while (!executorService.isTerminated()) {
	if (!executorService.awaitTermination(1, TimeUnit.HOURS)) {
	throw new IllegalStateException("executor service termination wait time out");
	}
	}
	}

	/** Asserts that there are no half added edge in the graph. */
	private void assertGraphIsInConsistentState(Digraph<Integer> digraph) {
	for (Node<Integer> node : digraph.getNodes()) {
	assertNodeIsInConsistentState(node);
	}
	}

	private void assertNodeIsInConsistentState(Node<Integer> node) {
	for (Node<Integer> succ : node.getSuccessors()) {
	assertThat(succ.getPredecessors()).contains(node);
	}

	for (Node<Integer> pred : node.getPredecessors()) {
	assertThat(pred.getSuccessors()).contains(node);
	}
	}

	private static final class CreateNodeWorker implements Runnable {

	private final Digraph<Integer> digraph;
	private final int outgoingEdgesPerNode;
	private final int incomingEdgesPerNode;
	private final AtomicInteger idGenerator;

	private CreateNodeWorker(
	Digraph<Integer> digraph,
	int outgoingEdgesPerNode,
	int incomingEdgesPerNode,
	AtomicInteger idGenerator) {
	this.digraph = requireNonNull(digraph);
	Preconditions.checkArgument(outgoingEdgesPerNode >= 0);
	this.outgoingEdgesPerNode = outgoingEdgesPerNode;
	Preconditions.checkArgument(incomingEdgesPerNode >= 0);
	this.incomingEdgesPerNode = incomingEdgesPerNode;
	this.idGenerator = idGenerator;
	}

	@Override
	public void run() {
	Integer newNodeId = idGenerator.getAndIncrement();
	digraph.createNode(newNodeId);
	addEdgesFromNew(newNodeId);
	addEdgesToNew(newNodeId);
	}

	private void addEdgesFromNew(int newNodeId) {
	int count = Math.min(newNodeId, outgoingEdgesPerNode);
	addEdges(digraph, newNodeId, count, /outgoing/ true);
	}

	private void addEdgesToNew(int newNodeId) {
	int count = Math.min(newNodeId, incomingEdgesPerNode);
	addEdges(digraph, newNodeId, count, /outgoing/ false);
	}
	}

	private static final class CreateEdgeNightContention implements Runnable {

	private final Digraph<Integer> digraph;
	private final CountDownLatch countDownLatch;
	private final int nodeId;
	private final int numberEdgeToAdd;
	private final boolean outgoing;
	private final Runnable inerruptCallBack;

	private CreateEdgeNightContention(
	Digraph<Integer> digraph,
	CountDownLatch countDownLatch,
	int nodeId,
	int numberEdgeToAdd,
	boolean outgoing,
	Runnable inerruptCallBack) {
	this.digraph = requireNonNull(digraph);
	this.countDownLatch = requireNonNull(countDownLatch);
	Preconditions.checkArgument(nodeId >= 0);
	this.nodeId = nodeId;
	Preconditions.checkArgument(numberEdgeToAdd >= 0);
	this.numberEdgeToAdd = numberEdgeToAdd;
	this.outgoing = outgoing;
	this.inerruptCallBack = requireNonNull(inerruptCallBack);
	}

	@Override
	public void run() {
	try {
	countDownLatch.await();
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	inerruptCallBack.run();
	// hardly ever will see that in the log. Who cares?
	fail(e.getMessage());
	}
	addEdges(digraph, nodeId, numberEdgeToAdd, outgoing);
	}
	}

	private static void addEdges(Digraph<Integer> digraph, int nodeId, int count, boolean outgoing) {

	for (int i = 0; i < count; i++) {
	int id = RANDOM.nextInt(digraph.getNodeCount());
	if (outgoing) {
	digraph.addEdge(nodeId, id);
	} else {
	digraph.addEdge(id, nodeId);
	}
	}
	}
	}