Add cycle detecting locks, fix Orchid deadlock 2

subgraph/Orchid#10

orchid/pom.xml

@@ -114,6 +114,11 @@
             <version>4.11</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>16.0.1</version>
+        </dependency>
     </dependencies>
 
 </project>

orchid/src/com/subgraph/orchid/Threading.java

@@ -0,0 +1,47 @@
+package com.subgraph.orchid;
+
+import com.google.common.util.concurrent.CycleDetectingLockFactory;
+
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Created by android on 8/22/14.
+ */
+public class Threading {
+    static {
+        // Default policy goes here. If you want to change this, use one of the static methods before
+        // instantiating any orchid objects. The policy change will take effect only on new objects
+        // from that point onwards.
+        throwOnLockCycles();
+    }
+
+    private static CycleDetectingLockFactory.Policy policy;
+    public static CycleDetectingLockFactory factory;
+
+    public static ReentrantLock lock(String name) {
+        return factory.newReentrantLock(name);
+    }
+
+    public static void warnOnLockCycles() {
+        setPolicy(CycleDetectingLockFactory.Policies.WARN);
+    }
+
+    public static void throwOnLockCycles() {
+        setPolicy(CycleDetectingLockFactory.Policies.THROW);
+    }
+
+    public static void ignoreLockCycles() {
+        setPolicy(CycleDetectingLockFactory.Policies.DISABLED);
+    }
+
+    public static void setPolicy(CycleDetectingLockFactory.Policy policy) {
+        Threading.policy = policy;
+        factory = CycleDetectingLockFactory.newInstance(policy);
+    }
+
+    public static CycleDetectingLockFactory.Policy getPolicy() {
+        return policy;
+    }
+
+
+}

orchid/src/com/subgraph/orchid/circuits/CircuitIO.java

@@ -9,6 +9,7 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -18,6 +19,7 @@
 import com.subgraph.orchid.ConnectionIOException;
 import com.subgraph.orchid.RelayCell;
 import com.subgraph.orchid.Stream;
+import com.subgraph.orchid.Threading;
 import com.subgraph.orchid.TorException;
 import com.subgraph.orchid.circuits.cells.CellImpl;
 import com.subgraph.orchid.circuits.cells.RelayCellImpl;
@@ -36,11 +38,10 @@ public class CircuitIO implements DashboardRenderable {
 	private final BlockingQueue<RelayCell> relayCellResponseQueue;
 	private final BlockingQueue<Cell> controlCellResponseQueue;
 	private final Map<Integer, StreamImpl> streamMap;
-	private final Object relaySendLock = new Object();
+    private final ReentrantLock streamLock = Threading.lock("stream");
+    private final ReentrantLock relaySendLock = Threading.lock("relaySend");
 
-    /** LOCKING: streamMap */
 	private boolean isMarkedForClose;
-    /** LOCKING: streamMap */
 	private boolean isClosed;
 
 	CircuitIO(CircuitImpl circuit, Connection connection, int circuitId) {
@@ -171,23 +172,27 @@ private void processRelayDataCell(RelayCell cell) {
 			}
 		}
 
-		synchronized(streamMap) {
+		streamLock.lock();
+        try {
 			final StreamImpl stream = streamMap.get(cell.getStreamId());
 			// It's not unusual for the stream to not be found.  For example, if a RELAY_CONNECTED arrives after
 			// the client has stopped waiting for it, the stream will never be tracked and eventually the edge node
 			// will send a RELAY_END for this stream.
 			if(stream != null) {
 				stream.addInputCell(cell);
 			}
-		}
+		} finally {
+            streamLock.unlock();
+        }
 	}
 
 	RelayCell createRelayCell(int relayCommand, int streamId, CircuitNode targetNode) {
 		return new RelayCellImpl(targetNode, circuitId, streamId, relayCommand);
 	}
 
 	void sendRelayCellTo(RelayCell cell, CircuitNode targetNode) {
-		synchronized(relaySendLock) {
+		relaySendLock.lock();
+        try {
 			logRelayCell("Sending:     ", cell);
 			cell.setLength();
 			targetNode.updateForwardDigest(cell);
@@ -200,7 +205,9 @@ void sendRelayCellTo(RelayCell cell, CircuitNode targetNode) {
 				targetNode.waitForSendWindowAndDecrement();
 
 			sendCell(cell);
-		}
+		} finally {
+            relaySendLock.unlock();
+        }
 	}
 
 
@@ -236,20 +243,26 @@ void sendCell(Cell cell) {
 
 	void markForClose() {
         boolean shouldClose;
-		synchronized (streamMap) {
+        streamLock.lock();
+        try {
 			if(isMarkedForClose) {
 				return;
 			}
 			isMarkedForClose = true;
             shouldClose = streamMap.isEmpty();
-		}
+		} finally {
+            streamLock.unlock();
+        }
         if(shouldClose)
             closeCircuit();
 	}
 
 	boolean isMarkedForClose() {
-        synchronized (streamMap) {
+        streamLock.lock();
+        try {
             return isMarkedForClose;
+        } finally {
+            streamLock.unlock();
         }
 	}
 
@@ -276,7 +289,8 @@ private void processCircuitSendme(RelayCell cell) {
 	}
 
 	void destroyCircuit() {
-		synchronized(streamMap) {
+        streamLock.lock();
+        try {
 			if(isClosed) {
 				return;
 			}
@@ -287,32 +301,43 @@ void destroyCircuit() {
 				s.close();
 			}
 			isClosed = true;
-		}
+		} finally {
+            streamLock.unlock();
+        }
 	}
 
 	StreamImpl createNewStream(boolean autoclose) {
-		synchronized(streamMap) {
+        streamLock.lock();
+        try {
 			final int streamId = circuit.getStatus().nextStreamId();
 			final StreamImpl stream = new StreamImpl(circuit, circuit.getFinalCircuitNode(), streamId, autoclose);
 			streamMap.put(streamId, stream);
 			return stream;
-		}
+		} finally {
+            streamLock.unlock();
+        }
 	}
 
 	void removeStream(StreamImpl stream) {
         boolean shouldClose;
-		synchronized(streamMap) {
+        streamLock.lock();
+        try {
 			streamMap.remove(stream.getStreamId());
             shouldClose = streamMap.isEmpty() && isMarkedForClose;
-		}
+		} finally {
+            streamLock.unlock();
+        }
         if(shouldClose)
             closeCircuit();
 	}
 
 	List<Stream> getActiveStreams() {
-		synchronized (streamMap) {
+        streamLock.lock();
+        try {
 			return new ArrayList<Stream>(streamMap.values());
-		}
+		} finally {
+            streamLock.unlock();
+        }
 	}
 
 	public void dashboardRender(DashboardRenderer renderer, PrintWriter writer, int flags) throws IOException {

orchid/src/com/subgraph/orchid/circuits/CircuitManagerImpl.java

@@ -13,6 +13,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.locks.ReentrantLock;
 
 import com.subgraph.orchid.Circuit;
 import com.subgraph.orchid.CircuitBuildHandler;
@@ -29,6 +30,7 @@
 import com.subgraph.orchid.Router;
 import com.subgraph.orchid.Stream;
 import com.subgraph.orchid.StreamConnectFailedException;
+import com.subgraph.orchid.Threading;
 import com.subgraph.orchid.Tor;
 import com.subgraph.orchid.TorConfig;
 import com.subgraph.orchid.circuits.guards.EntryGuards;
@@ -62,6 +64,7 @@ interface CircuitFilter {
 	private final TorInitializationTracker initializationTracker;
 	private final CircuitPathChooser pathChooser;
 	private final HiddenServiceManager hiddenServiceManager;
+    private final ReentrantLock lock = Threading.lock("circuitManager");
 
 	public CircuitManagerImpl(TorConfig config, DirectoryDownloaderImpl directoryDownloader, Directory directory, ConnectionCache connectionCache, TorInitializationTracker initializationTracker) {
 		this.config = config;
@@ -87,14 +90,20 @@ public void startBuildingCircuits() {
 		scheduledExecutor.scheduleAtFixedRate(circuitCreationTask, 0, 1000, TimeUnit.MILLISECONDS);
 	}
 
-	public synchronized void stopBuildingCircuits(boolean killCircuits) {
-		scheduledExecutor.shutdownNow();
-		if(killCircuits) {
-			List<CircuitImpl> circuits = new ArrayList<CircuitImpl>(activeCircuits);
-			for(CircuitImpl c: circuits) {
-				c.destroyCircuit();
-			}
-		}
+	public void stopBuildingCircuits(boolean killCircuits) {
+        lock.lock();
+
+        try {
+            scheduledExecutor.shutdownNow();
+            if (killCircuits) {
+                List<CircuitImpl> circuits = new ArrayList<CircuitImpl>(activeCircuits);
+                for (CircuitImpl c : circuits) {
+                    c.destroyCircuit();
+                }
+            }
+        } finally {
+            lock.unlock();
+        }
 	}
 
 	public ExitCircuit createNewExitCircuit(Router exitRouter) {
@@ -114,8 +123,10 @@ void removeActiveCircuit(CircuitImpl circuit) {
 		}
 	}
 
-	synchronized int getActiveCircuitCount() {
-		return activeCircuits.size();
+	int getActiveCircuitCount() {
+        synchronized (activeCircuits) {
+            return activeCircuits.size();
+        }
 	}
 
 	Set<Circuit> getPendingCircuits() {
@@ -126,19 +137,31 @@ public boolean filter(Circuit circuit) {
 		});
 	}
 
-	synchronized int getPendingCircuitCount() {
-		return getPendingCircuits().size();
+	int getPendingCircuitCount() {
+        lock.lock();
+
+        try {
+            return getPendingCircuits().size();
+        } finally {
+            lock.unlock();
+        }
 	}
 
 	Set<Circuit> getCircuitsByFilter(CircuitFilter filter) {
 		final Set<Circuit> result = new HashSet<Circuit>();
+        final Set<CircuitImpl> circuits = new HashSet<CircuitImpl>();
+
 		synchronized (activeCircuits) {
-			for(CircuitImpl c: activeCircuits) {
-				if(filter == null || filter.filter(c)) {
-					result.add(c);
-				}
-			}
+            // the filter might lock additional objects, causing a deadlock, so don't
+            // call it inside the monitor
+            circuits.addAll(activeCircuits);
 		}
+
+        for(CircuitImpl c: circuits) {
+            if(filter == null || filter.filter(c)) {
+                result.add(c);
+            }
+        }
 		return result;
 	}