Merge branch '1712-detect-dead-bluetooth-connections' into 'master'

Detect and close dead Bluetooth connections

See merge request briar/briar!1246

bramble-core/src/main/java/org/briarproject/bramble/BrambleCoreModule.java

@@ -9,6 +9,7 @@ import org.briarproject.bramble.db.DatabaseExecutorModule;
 import org.briarproject.bramble.db.DatabaseModule;
 import org.briarproject.bramble.event.EventModule;
 import org.briarproject.bramble.identity.IdentityModule;
+import org.briarproject.bramble.io.IoModule;
 import org.briarproject.bramble.keyagreement.KeyAgreementModule;
 import org.briarproject.bramble.lifecycle.LifecycleModule;
 import org.briarproject.bramble.plugin.PluginModule;
@@ -35,6 +36,7 @@ import dagger.Module;
 		DatabaseExecutorModule.class,
 		EventModule.class,
 		IdentityModule.class,
+		IoModule.class,
 		KeyAgreementModule.class,
 		LifecycleModule.class,
 		PluginModule.class,

bramble-core/src/main/java/org/briarproject/bramble/io/IoModule.java

@@ -0,0 +1,18 @@
+package org.briarproject.bramble.io;
+
+import org.briarproject.bramble.api.io.TimeoutMonitor;
+
+import javax.inject.Singleton;
+
+import dagger.Module;
+import dagger.Provides;
+
+@Module
+public class IoModule {
+
+	@Provides
+	@Singleton
+	TimeoutMonitor provideTimeoutMonitor(TimeoutMonitorImpl timeoutMonitor) {
+		return timeoutMonitor;
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/io/TimeoutInputStream.java

@@ -0,0 +1,104 @@
+package org.briarproject.bramble.io;
+
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.system.Clock;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import javax.annotation.concurrent.GuardedBy;
+
+@NotNullByDefault
+class TimeoutInputStream extends InputStream {
+
+	private final Clock clock;
+	private final InputStream in;
+	private final long timeoutMs;
+	private final CloseListener listener;
+	private final Object lock = new Object();
+	@GuardedBy("lock")
+	private long readStartedMs = -1;
+
+	TimeoutInputStream(Clock clock, InputStream in, long timeoutMs,
+			CloseListener listener) {
+		this.clock = clock;
+		this.in = in;
+		this.timeoutMs = timeoutMs;
+		this.listener = listener;
+	}
+
+	@Override
+	public int read() throws IOException {
+		synchronized (lock) {
+			readStartedMs = clock.currentTimeMillis();
+		}
+		int input = in.read();
+		synchronized (lock) {
+			readStartedMs = -1;
+		}
+		return input;
+	}
+
+	@Override
+	public int read(byte[] b) throws IOException {
+		return read(b, 0, b.length);
+	}
+
+	@Override
+	public int read(byte[] b, int off, int len) throws IOException {
+		synchronized (lock) {
+			readStartedMs = clock.currentTimeMillis();
+		}
+		int read = in.read(b, off, len);
+		synchronized (lock) {
+			readStartedMs = -1;
+		}
+		return read;
+	}
+
+	@Override
+	public void close() throws IOException {
+		try {
+			in.close();
+		} finally {
+			listener.onClose(this);
+		}
+	}
+
+	@Override
+	public int available() throws IOException {
+		return in.available();
+	}
+
+	@Override
+	public void mark(int readlimit) {
+		in.mark(readlimit);
+	}
+
+	@Override
+	public boolean markSupported() {
+		return in.markSupported();
+	}
+
+	@Override
+	public void reset() throws IOException {
+		in.reset();
+	}
+
+	@Override
+	public long skip(long n) throws IOException {
+		return in.skip(n);
+	}
+
+	boolean hasTimedOut() {
+		synchronized (lock) {
+			return readStartedMs != -1 &&
+					clock.currentTimeMillis() - readStartedMs > timeoutMs;
+		}
+	}
+
+	interface CloseListener {
+
+		void onClose(TimeoutInputStream closed);
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/io/TimeoutMonitorImpl.java

@@ -0,0 +1,96 @@
+package org.briarproject.bramble.io;
+
+import org.briarproject.bramble.api.io.TimeoutMonitor;
+import org.briarproject.bramble.api.lifecycle.IoExecutor;
+import org.briarproject.bramble.api.system.Clock;
+import org.briarproject.bramble.api.system.Scheduler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.logging.Logger;
+
+import javax.annotation.concurrent.GuardedBy;
+import javax.inject.Inject;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static java.util.logging.Level.INFO;
+import static java.util.logging.Logger.getLogger;
+import static org.briarproject.bramble.util.LogUtils.logException;
+
+class TimeoutMonitorImpl implements TimeoutMonitor {
+
+	private static final Logger LOG =
+			getLogger(TimeoutMonitorImpl.class.getName());
+
+	private static final long CHECK_INTERVAL_MS = SECONDS.toMillis(10);
+
+	private final ScheduledExecutorService scheduler;
+	private final Executor ioExecutor;
+	private final Clock clock;
+	private final Object lock = new Object();
+	@GuardedBy("lock")
+	private final List<TimeoutInputStream> streams = new ArrayList<>();
+
+	@GuardedBy("lock")
+	private Future<?> task = null;
+
+	@Inject
+	TimeoutMonitorImpl(@Scheduler ScheduledExecutorService scheduler,
+			@IoExecutor Executor ioExecutor, Clock clock) {
+		this.scheduler = scheduler;
+		this.ioExecutor = ioExecutor;
+		this.clock = clock;
+	}
+
+	@Override
+	public InputStream createTimeoutInputStream(InputStream in,
+			long timeoutMs) {
+		TimeoutInputStream stream = new TimeoutInputStream(clock, in,
+				timeoutMs, this::removeStream);
+		synchronized (lock) {
+			if (streams.isEmpty()) {
+				task = scheduler.scheduleWithFixedDelay(this::checkTimeouts,
+						CHECK_INTERVAL_MS, CHECK_INTERVAL_MS, MILLISECONDS);
+			}
+			streams.add(stream);
+		}
+		return stream;
+	}
+
+	private void removeStream(TimeoutInputStream stream) {
+		Future<?> toCancel = null;
+		synchronized (lock) {
+			if (streams.remove(stream) && streams.isEmpty()) {
+				toCancel = task;
+				task = null;
+			}
+		}
+		if (toCancel != null) toCancel.cancel(false);
+	}
+
+	@Scheduler
+	private void checkTimeouts() {
+		ioExecutor.execute(() -> {
+			List<TimeoutInputStream> snapshot;
+			synchronized (lock) {
+				snapshot = new ArrayList<>(streams);
+			}
+			for (TimeoutInputStream stream : snapshot) {
+				if (stream.hasTimedOut()) {
+					LOG.info("Input stream has timed out");
+					try {
+						stream.close();
+					} catch (IOException e) {
+						logException(LOG, INFO, e);
+					}
+				}
+			}
+		});
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/bluetooth/BluetoothPlugin.java

@@ -5,6 +5,7 @@ import org.briarproject.bramble.api.Pair;
 import org.briarproject.bramble.api.data.BdfList;
 import org.briarproject.bramble.api.event.Event;
 import org.briarproject.bramble.api.event.EventListener;
+import org.briarproject.bramble.api.io.TimeoutMonitor;
 import org.briarproject.bramble.api.keyagreement.KeyAgreementConnection;
 import org.briarproject.bramble.api.keyagreement.KeyAgreementListener;
 import org.briarproject.bramble.api.keyagreement.event.KeyAgreementListeningEvent;
@@ -60,12 +61,13 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
 			getLogger(BluetoothPlugin.class.getName());
 
 	final BluetoothConnectionLimiter connectionLimiter;
+	final TimeoutMonitor timeoutMonitor;
 
 	private final Executor ioExecutor;
 	private final SecureRandom secureRandom;
 	private final Backoff backoff;
 	private final PluginCallback callback;
-	private final int maxLatency;
+	private final int maxLatency, maxIdleTime;
 	private final AtomicBoolean used = new AtomicBoolean(false);
 
 	private volatile boolean running = false, contactConnections = false;
@@ -105,14 +107,17 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
 	abstract DuplexTransportConnection discoverAndConnect(String uuid);
 
 	BluetoothPlugin(BluetoothConnectionLimiter connectionLimiter,
-			Executor ioExecutor, SecureRandom secureRandom,
-			Backoff backoff, PluginCallback callback, int maxLatency) {
+			TimeoutMonitor timeoutMonitor, Executor ioExecutor,
+			SecureRandom secureRandom, Backoff backoff,
+			PluginCallback callback, int maxLatency, int maxIdleTime) {
 		this.connectionLimiter = connectionLimiter;
+		this.timeoutMonitor = timeoutMonitor;
 		this.ioExecutor = ioExecutor;
 		this.secureRandom = secureRandom;
 		this.backoff = backoff;
 		this.callback = callback;
 		this.maxLatency = maxLatency;
+		this.maxIdleTime = maxIdleTime;
 	}
 
 	void onAdapterEnabled() {
@@ -141,8 +146,7 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
 
 	@Override
 	public int getMaxIdleTime() {
-		// Bluetooth detects dead connections so we don't need keepalives
-		return Integer.MAX_VALUE;
+		return maxIdleTime;
 	}
 
 	@Override

bramble-core/src/test/java/org/briarproject/bramble/io/TimeoutInputStreamTest.java

@@ -0,0 +1,143 @@
+package org.briarproject.bramble.io;
+
+import org.briarproject.bramble.test.BrambleTestCase;
+import org.briarproject.bramble.test.SettableClock;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class TimeoutInputStreamTest extends BrambleTestCase {
+
+	private static final long TIMEOUT_MS = MINUTES.toMillis(1);
+
+	private final long now = System.currentTimeMillis();
+
+	private AtomicLong time;
+	private UnresponsiveInputStream in;
+	private AtomicBoolean listenerCalled;
+	private TimeoutInputStream stream;
+	private CountDownLatch readReturned;
+
+	@Before
+	public void setUp() {
+		time = new AtomicLong(now);
+		in = new UnresponsiveInputStream();
+		listenerCalled = new AtomicBoolean(false);
+		stream = new TimeoutInputStream(new SettableClock(time), in,
+				TIMEOUT_MS, stream -> listenerCalled.set(true));
+		readReturned = new CountDownLatch(1);
+	}
+
+	@Test
+	public void testTimeoutIsReportedIfReadDoesNotReturn() throws Exception {
+		startReading();
+		try {
+			// The stream should not report a timeout
+			assertFalse(stream.hasTimedOut());
+
+			// Time passes
+			time.set(now + TIMEOUT_MS);
+
+			// The stream still shouldn't report a timeout
+			assertFalse(stream.hasTimedOut());
+
+			// Time passes
+			time.set(now + TIMEOUT_MS + 1);
+
+			// The stream should report a timeout
+			assertTrue(stream.hasTimedOut());
+
+			// The listener should not have been called yet
+			assertFalse(listenerCalled.get());
+
+			// Close the stream
+			stream.close();
+
+			// The listener should have been called
+			assertTrue(listenerCalled.get());
+		} finally {
+			// Allow the read to return
+			in.readFinished.countDown();
+		}
+	}
+
+	@Test
+	public void testTimeoutIsNotReportedIfReadReturns() throws Exception {
+		startReading();
+		try {
+			// The stream should not report a timeout
+			assertFalse(stream.hasTimedOut());
+
+			// Time passes
+			time.set(now + TIMEOUT_MS);
+
+			// The stream still shouldn't report a timeout
+			assertFalse(stream.hasTimedOut());
+
+			// Allow the read to finish and wait for it to return
+			in.readFinished.countDown();
+			readReturned.await(10, SECONDS);
+
+			// Time passes
+			time.set(now + TIMEOUT_MS + 1);
+
+			// The stream should not report a timeout as the read has returned
+			assertFalse(stream.hasTimedOut());
+
+			// The listener should not have been called yet
+			assertFalse(listenerCalled.get());
+
+			// Close the stream
+			stream.close();
+
+			// The listener should have been called
+			assertTrue(listenerCalled.get());
+		} finally {
+			// Allow the read to return in case an assertion was thrown
+			in.readFinished.countDown();
+		}
+	}
+
+	private void startReading() throws Exception {
+		// Start a background thread to read from the unresponsive stream
+		new Thread(() -> {
+			try {
+				assertEquals(123, stream.read());
+				readReturned.countDown();
+			} catch (IOException e) {
+				fail();
+			}
+		}).start();
+		// Wait for the background thread to start reading
+		assertTrue(in.readStarted.await(10, SECONDS));
+	}
+
+	private class UnresponsiveInputStream extends InputStream {
+
+		private final CountDownLatch readStarted = new CountDownLatch(1);
+		private final CountDownLatch readFinished = new CountDownLatch(1);
+
+		@Override
+		public int read() throws IOException {
+			readStarted.countDown();
+			try {
+				readFinished.await();
+				return 123;
+			} catch (InterruptedException e) {
+				throw new IOException(e);
+			}
+		}
+	}
+}