Improved encapsulation of thread synchronisation as follows

- replaced use of Object instance mutex with a private final Lock object
- replaced Object signaling with specific condition signalling
This commit is contained in:
Abraham Kiggundu
2014-12-23 23:55:56 +03:00
parent 276dcb1038
commit b074978472
19 changed files with 1001 additions and 478 deletions

1
.gitignore vendored
View File

@@ -1,3 +1,4 @@
build
.gradle
.metadata
*.tmp

View File

@@ -11,6 +11,8 @@ import static java.util.logging.Level.WARNING;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import javax.inject.Inject;
@@ -64,6 +66,8 @@ Service, EventListener {
private int nextRequestId = 0; // Locking: this
private volatile Settings settings = new Settings();
private final Lock synchLock = new ReentrantLock();
@Inject
public AndroidNotificationManagerImpl(DatabaseComponent db,
@@ -103,19 +107,31 @@ Service, EventListener {
if(e instanceof SettingsUpdatedEvent) loadSettings();
}
public synchronized void showPrivateMessageNotification(ContactId c) {
Integer count = contactCounts.get(c);
if(count == null) contactCounts.put(c, 1);
else contactCounts.put(c, count + 1);
privateTotal++;
updatePrivateMessageNotification();
public void showPrivateMessageNotification(ContactId c) {
synchLock.lock();
try{
Integer count = contactCounts.get(c);
if(count == null) contactCounts.put(c, 1);
else contactCounts.put(c, count + 1);
privateTotal++;
updatePrivateMessageNotification();
}
finally{
synchLock.unlock();
}
}
public synchronized void clearPrivateMessageNotification(ContactId c) {
Integer count = contactCounts.remove(c);
if(count == null) return; // Already cleared
privateTotal -= count;
updatePrivateMessageNotification();
public void clearPrivateMessageNotification(ContactId c) {
synchLock.lock();
try{
Integer count = contactCounts.remove(c);
if(count == null) return; // Already cleared
privateTotal -= count;
updatePrivateMessageNotification();
}
finally{
synchLock.unlock();
}
}
// Locking: this
@@ -180,19 +196,31 @@ Service, EventListener {
return defaults;
}
public synchronized void showGroupPostNotification(GroupId g) {
Integer count = groupCounts.get(g);
if(count == null) groupCounts.put(g, 1);
else groupCounts.put(g, count + 1);
groupTotal++;
updateGroupPostNotification();
public void showGroupPostNotification(GroupId g) {
synchLock.lock();
try{
Integer count = groupCounts.get(g);
if(count == null) groupCounts.put(g, 1);
else groupCounts.put(g, count + 1);
groupTotal++;
updateGroupPostNotification();
}
finally{
synchLock.unlock();
}
}
public synchronized void clearGroupPostNotification(GroupId g) {
public void clearGroupPostNotification(GroupId g) {
synchLock.lock();
try{
Integer count = groupCounts.remove(g);
if(count == null) return; // Already cleared
groupTotal -= count;
updateGroupPostNotification();
}
finally{
synchLock.unlock();
}
}
// Locking: this
@@ -238,18 +266,23 @@ Service, EventListener {
}
}
// Locking: this
private void clearGroupPostNotification() {
Object o = appContext.getSystemService(NOTIFICATION_SERVICE);
NotificationManager nm = (NotificationManager) o;
nm.cancel(GROUP_POST_NOTIFICATION_ID);
}
public synchronized void clearNotifications() {
contactCounts.clear();
groupCounts.clear();
privateTotal = groupTotal = 0;
clearPrivateMessageNotification();
clearGroupPostNotification();
public void clearNotifications() {
synchLock.lock();
try{
contactCounts.clear();
groupCounts.clear();
privateTotal = groupTotal = 0;
clearPrivateMessageNotification();
clearGroupPostNotification();
}
finally{
synchLock.unlock();
}
}
}

View File

@@ -4,6 +4,8 @@ import static java.util.logging.Level.INFO;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import org.briarproject.api.android.ReferenceManager;
@@ -13,49 +15,70 @@ class ReferenceManagerImpl implements ReferenceManager {
private static final Logger LOG =
Logger.getLogger(ReferenceManagerImpl.class.getName());
// Locking: this
private final Map<Class<?>, Map<Long, Object>> outerMap =
new HashMap<Class<?>, Map<Long, Object>>();
private long nextHandle = 0; // Locking: this
public synchronized <T> T getReference(long handle, Class<T> c) {
Map<Long, Object> innerMap = outerMap.get(c);
if(innerMap == null) {
private final Lock synchLock = new ReentrantLock();
public <T> T getReference(long handle, Class<T> c) {
synchLock.lock();
try{
Map<Long, Object> innerMap = outerMap.get(c);
if(innerMap == null) {
if(LOG.isLoggable(INFO))
LOG.info("0 handles for " + c.getName());
return null;
}
if(LOG.isLoggable(INFO))
LOG.info("0 handles for " + c.getName());
return null;
LOG.info(innerMap.size() + " handles for " + c.getName());
Object o = innerMap.get(handle);
return c.cast(o);
}
if(LOG.isLoggable(INFO))
LOG.info(innerMap.size() + " handles for " + c.getName());
Object o = innerMap.get(handle);
return c.cast(o);
finally{
synchLock.unlock();
}
}
public synchronized <T> long putReference(T reference, Class<T> c) {
Map<Long, Object> innerMap = outerMap.get(c);
if(innerMap == null) {
innerMap = new HashMap<Long, Object>();
outerMap.put(c, innerMap);
public <T> long putReference(T reference, Class<T> c) {
synchLock.lock();
try{
Map<Long, Object> innerMap = outerMap.get(c);
if(innerMap == null) {
innerMap = new HashMap<Long, Object>();
outerMap.put(c, innerMap);
}
long handle = nextHandle++;
innerMap.put(handle, reference);
if(LOG.isLoggable(INFO)) {
LOG.info(innerMap.size() + " handles for " + c.getName() +
" after put");
}
return handle;
}
long handle = nextHandle++;
innerMap.put(handle, reference);
if(LOG.isLoggable(INFO)) {
LOG.info(innerMap.size() + " handles for " + c.getName() +
" after put");
finally{
synchLock.unlock();
}
return handle;
}
public synchronized <T> T removeReference(long handle, Class<T> c) {
Map<Long, Object> innerMap = outerMap.get(c);
if(innerMap == null) return null;
Object o = innerMap.remove(handle);
if(innerMap.isEmpty()) outerMap.remove(c);
if(LOG.isLoggable(INFO)) {
LOG.info(innerMap.size() + " handles for " + c.getName() +
" after remove");
public <T> T removeReference(long handle, Class<T> c) {
synchLock.lock();
try{
Map<Long, Object> innerMap = outerMap.get(c);
if(innerMap == null) return null;
Object o = innerMap.remove(handle);
if(innerMap.isEmpty()) outerMap.remove(c);
if(LOG.isLoggable(INFO)) {
LOG.info(innerMap.size() + " handles for " + c.getName() +
" after remove");
}
return c.cast(o);
}
return c.cast(o);
finally{
synchLock.unlock();
}
}
}

View File

@@ -1,5 +1,8 @@
package org.briarproject.crypto;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.briarproject.api.crypto.MessageDigest;
import org.spongycastle.crypto.BlockCipher;
import org.spongycastle.crypto.digests.SHA256Digest;
@@ -23,61 +26,89 @@ class FortunaGenerator {
private final byte[] counter = new byte[BLOCK_BYTES];
private final byte[] buffer = new byte[BLOCK_BYTES];
private final byte[] newKey = new byte[KEY_BYTES];
private final Lock synchLock = new ReentrantLock();
FortunaGenerator(byte[] seed) {
reseed(seed);
}
synchronized void reseed(byte[] seed) {
digest.update(key);
digest.update(seed);
digest.digest(key, 0, KEY_BYTES);
incrementCounter();
void reseed(byte[] seed) {
synchLock.lock();
try{
digest.update(key);
digest.update(seed);
digest.digest(key, 0, KEY_BYTES);
incrementCounter();
}
finally{
synchLock.unlock();
}
}
// Package access for testing
synchronized void incrementCounter() {
counter[0]++;
for(int i = 0; counter[i] == 0; i++) {
if(i + 1 == BLOCK_BYTES)
throw new RuntimeException("Counter exhausted");
counter[i + 1]++;
void incrementCounter() {
synchLock.lock();
try{
counter[0]++;
for(int i = 0; counter[i] == 0; i++) {
if(i + 1 == BLOCK_BYTES)
throw new RuntimeException("Counter exhausted");
counter[i + 1]++;
}
}
}
finally{
synchLock.unlock();
}
}
// Package access for testing
synchronized byte[] getCounter() {
return counter;
byte[] getCounter() {
synchLock.lock();
try{
return counter;
}
finally{
synchLock.unlock();
}
}
synchronized int nextBytes(byte[] dest, int off, int len) {
// Don't write more than the maximum number of bytes in one request
if(len > MAX_BYTES_PER_REQUEST) len = MAX_BYTES_PER_REQUEST;
cipher.init(true, new KeyParameter(key));
// Generate full blocks directly into the output buffer
int fullBlocks = len / BLOCK_BYTES;
for(int i = 0; i < fullBlocks; i++) {
cipher.processBlock(counter, 0, dest, off + i * BLOCK_BYTES);
incrementCounter();
int nextBytes(byte[] dest, int off, int len) {
synchLock.lock();
try{
// Don't write more than the maximum number of bytes in one request
if(len > MAX_BYTES_PER_REQUEST) len = MAX_BYTES_PER_REQUEST;
cipher.init(true, new KeyParameter(key));
// Generate full blocks directly into the output buffer
int fullBlocks = len / BLOCK_BYTES;
for(int i = 0; i < fullBlocks; i++) {
cipher.processBlock(counter, 0, dest, off + i * BLOCK_BYTES);
incrementCounter();
}
// Generate a partial block if needed
int done = fullBlocks * BLOCK_BYTES, remaining = len - done;
assert remaining < BLOCK_BYTES;
if(remaining > 0) {
cipher.processBlock(counter, 0, buffer, 0);
incrementCounter();
// Copy the partial block to the output buffer and erase our copy
System.arraycopy(buffer, 0, dest, off + done, remaining);
for(int i = 0; i < BLOCK_BYTES; i++) buffer[i] = 0;
}
// Generate a new key
for(int i = 0; i < KEY_BYTES / BLOCK_BYTES; i++) {
cipher.processBlock(counter, 0, newKey, i * BLOCK_BYTES);
incrementCounter();
}
System.arraycopy(newKey, 0, key, 0, KEY_BYTES);
for(int i = 0; i < KEY_BYTES; i++) newKey[i] = 0;
// Return the number of bytes written
return len;
}
// Generate a partial block if needed
int done = fullBlocks * BLOCK_BYTES, remaining = len - done;
assert remaining < BLOCK_BYTES;
if(remaining > 0) {
cipher.processBlock(counter, 0, buffer, 0);
incrementCounter();
// Copy the partial block to the output buffer and erase our copy
System.arraycopy(buffer, 0, dest, off + done, remaining);
for(int i = 0; i < BLOCK_BYTES; i++) buffer[i] = 0;
finally{
synchLock.unlock();
}
// Generate a new key
for(int i = 0; i < KEY_BYTES / BLOCK_BYTES; i++) {
cipher.processBlock(counter, 0, newKey, i * BLOCK_BYTES);
incrementCounter();
}
System.arraycopy(newKey, 0, key, 0, KEY_BYTES);
for(int i = 0; i < KEY_BYTES; i++) newKey[i] = 0;
// Return the number of bytes written
return len;
}
}

View File

@@ -1,5 +1,8 @@
package org.briarproject.crypto;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.briarproject.api.crypto.MessageDigest;
import org.briarproject.api.crypto.PseudoRandom;
import org.briarproject.util.ByteUtils;
@@ -10,6 +13,8 @@ class PseudoRandomImpl implements PseudoRandom {
private byte[] state;
private int offset;
private final Lock synchLock = new ReentrantLock();
PseudoRandomImpl(MessageDigest messageDigest, int seed1, int seed2) {
this.messageDigest = messageDigest;
@@ -21,21 +26,27 @@ class PseudoRandomImpl implements PseudoRandom {
offset = 0;
}
public synchronized byte[] nextBytes(int bytes) {
byte[] b = new byte[bytes];
int half = state.length / 2;
int off = 0, len = b.length, available = half - offset;
while(available < len) {
System.arraycopy(state, offset, b, off, available);
off += available;
len -= available;
messageDigest.update(state, half, half);
state = messageDigest.digest();
offset = 0;
available = half;
public byte[] nextBytes(int bytes) {
synchLock.lock();
try{
byte[] b = new byte[bytes];
int half = state.length / 2;
int off = 0, len = b.length, available = half - offset;
while(available < len) {
System.arraycopy(state, offset, b, off, available);
off += available;
len -= available;
messageDigest.update(state, half, half);
state = messageDigest.digest();
offset = 0;
available = half;
}
System.arraycopy(state, offset, b, off, len);
offset += len;
return b;
}
finally{
synchLock.unlock();
}
System.arraycopy(state, offset, b, off, len);
offset += len;
return b;
}
}

View File

@@ -1,5 +1,8 @@
package org.briarproject.crypto;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.briarproject.api.crypto.SecretKey;
import org.briarproject.util.ByteUtils;
@@ -8,23 +11,38 @@ class SecretKeyImpl implements SecretKey {
private final byte[] key;
private boolean erased = false; // Locking: this
private final Lock synchLock = new ReentrantLock();
SecretKeyImpl(byte[] key) {
this.key = key;
}
public synchronized byte[] getEncoded() {
if(erased) throw new IllegalStateException();
return key;
public byte[] getEncoded() {
synchLock.lock();
try{
if(erased) throw new IllegalStateException();
return key;
}
finally{
synchLock.unlock();
}
}
public SecretKey copy() {
return new SecretKeyImpl(key.clone());
}
public synchronized void erase() {
if(erased) throw new IllegalStateException();
ByteUtils.erase(key);
erased = true;
public void erase() {
synchLock.lock();
try{
if(erased) throw new IllegalStateException();
ByteUtils.erase(key);
erased = true;
}
finally{
synchLock.unlock();
}
}
}

View File

@@ -27,6 +27,9 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import org.briarproject.api.Author;
@@ -322,6 +325,9 @@ abstract class JdbcDatabase implements Database<Connection> {
protected abstract Connection createConnection() throws SQLException;
protected abstract void flushBuffersToDisk(Statement s) throws SQLException;
private final Lock connectionsLock = new ReentrantLock();
private final Condition connectionsChanged = connectionsLock.newCondition();
JdbcDatabase(String hashType, String binaryType, String counterType,
String secretType, Clock clock) {
@@ -431,19 +437,28 @@ abstract class JdbcDatabase implements Database<Connection> {
public Connection startTransaction() throws DbException {
Connection txn = null;
synchronized(connections) {
connectionsLock.lock();
try {
if(closed) throw new DbClosedException();
txn = connections.poll();
}
finally{
connectionsLock.unlock();
}
try {
if(txn == null) {
// Open a new connection
txn = createConnection();
if(txn == null) throw new DbException();
txn.setAutoCommit(false);
synchronized(connections) {
connectionsLock.lock();
try {
openConnections++;
}
finally{
connectionsLock.unlock();
}
}
} catch(SQLException e) {
throw new DbException(e);
@@ -455,9 +470,13 @@ abstract class JdbcDatabase implements Database<Connection> {
public void abortTransaction(Connection txn) {
try {
txn.rollback();
synchronized(connections) {
connectionsLock.lock();
try {
connections.add(txn);
connections.notifyAll();
connectionsChanged.signalAll();
}
finally{
connectionsLock.unlock();
}
} catch(SQLException e) {
// Try to close the connection
@@ -468,11 +487,14 @@ abstract class JdbcDatabase implements Database<Connection> {
if(LOG.isLoggable(WARNING)) LOG.log(WARNING, e1.toString(), e1);
}
// Whatever happens, allow the database to close
synchronized(connections) {
connectionsLock.lock();
try {
openConnections--;
connections.notifyAll();
connectionsChanged.signalAll();
}
}
finally{
connectionsLock.unlock();
} }
}
public void commitTransaction(Connection txn) throws DbException {
@@ -486,9 +508,13 @@ abstract class JdbcDatabase implements Database<Connection> {
tryToClose(s);
throw new DbException(e);
}
synchronized(connections) {
connectionsLock.lock();
try{
connections.add(txn);
connections.notifyAll();
connectionsChanged.signalAll();
}
finally{
connectionsLock.unlock();
}
}
@@ -502,14 +528,15 @@ abstract class JdbcDatabase implements Database<Connection> {
protected void closeAllConnections() throws SQLException {
boolean interrupted = false;
synchronized(connections) {
connectionsLock.lock();
try{
closed = true;
for(Connection c : connections) c.close();
openConnections -= connections.size();
connections.clear();
while(openConnections > 0) {
try {
connections.wait();
connectionsChanged.await();
} catch(InterruptedException e) {
LOG.warning("Interrupted while closing connections");
interrupted = true;
@@ -519,6 +546,10 @@ abstract class JdbcDatabase implements Database<Connection> {
connections.clear();
}
}
finally{
connectionsLock.unlock();
}
if(interrupted) Thread.currentThread().interrupt();
}

View File

@@ -10,6 +10,8 @@ import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import org.briarproject.api.Author;
@@ -60,6 +62,8 @@ class ConnectorGroup extends Thread implements InvitationTask {
private final Collection<InvitationListener> listeners;
private final AtomicBoolean connected;
private final CountDownLatch localConfirmationLatch;
private final Lock synchLock = new ReentrantLock();
/*
* All of the following require locking: this. We don't want to call the
@@ -104,12 +108,18 @@ class ConnectorGroup extends Thread implements InvitationTask {
localConfirmationLatch = new CountDownLatch(1);
}
public synchronized InvitationState addListener(InvitationListener l) {
listeners.add(l);
return new InvitationState(localInvitationCode, remoteInvitationCode,
localConfirmationCode, remoteConfirmationCode, connected.get(),
connectionFailed, localCompared, remoteCompared, localMatched,
remoteMatched, remoteName);
public InvitationState addListener(InvitationListener l) {
synchLock.lock();
try{
listeners.add(l);
return new InvitationState(localInvitationCode, remoteInvitationCode,
localConfirmationCode, remoteConfirmationCode, connected.get(),
connectionFailed, localCompared, remoteCompared, localMatched,
remoteMatched, remoteName);
}
finally{
synchLock.unlock();
}
}
public void removeListener(InvitationListener l) {
@@ -130,9 +140,13 @@ class ConnectorGroup extends Thread implements InvitationTask {
localProps = db.getLocalProperties();
} catch(DbException e) {
if(LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
synchronized(this) {
synchLock.lock();
try {
connectionFailed = true;
}
finally{
synchLock.unlock();
}
for(InvitationListener l : listeners) l.connectionFailed();
return;
}
@@ -163,9 +177,13 @@ class ConnectorGroup extends Thread implements InvitationTask {
}
// If none of the threads connected, inform the listeners
if(!connected.get()) {
synchronized(this) {
synchLock.lock();
try {
connectionFailed = true;
}
finally{
synchLock.unlock();
}
for(InvitationListener l : listeners) l.connectionFailed();
}
}
@@ -193,18 +211,26 @@ class ConnectorGroup extends Thread implements InvitationTask {
}
public void localConfirmationSucceeded() {
synchronized(this) {
synchLock.lock();
try {
localCompared = true;
localMatched = true;
}
finally{
synchLock.unlock();
}
localConfirmationLatch.countDown();
}
public void localConfirmationFailed() {
synchronized(this) {
synchLock.lock();
try {
localCompared = true;
localMatched = false;
}
finally{
synchLock.unlock();
}
localConfirmationLatch.countDown();
}
@@ -216,10 +242,14 @@ class ConnectorGroup extends Thread implements InvitationTask {
}
void keyAgreementSucceeded(int localCode, int remoteCode) {
synchronized(this) {
synchLock.lock();
try {
localConfirmationCode = localCode;
remoteConfirmationCode = remoteCode;
}
finally{
synchLock.unlock();
}
for(InvitationListener l : listeners)
l.keyAgreementSucceeded(localCode, remoteCode);
}
@@ -230,32 +260,48 @@ class ConnectorGroup extends Thread implements InvitationTask {
boolean waitForLocalConfirmationResult() throws InterruptedException {
localConfirmationLatch.await(CONFIRMATION_TIMEOUT, MILLISECONDS);
synchronized(this) {
synchLock.lock();
try {
return localMatched;
}
finally{
synchLock.unlock();
}
}
void remoteConfirmationSucceeded() {
synchronized(this) {
synchLock.lock();
try {
remoteCompared = true;
remoteMatched = true;
}
finally{
synchLock.unlock();
}
for(InvitationListener l : listeners) l.remoteConfirmationSucceeded();
}
void remoteConfirmationFailed() {
synchronized(this) {
synchLock.lock();
try {
remoteCompared = true;
remoteMatched = false;
}
finally{
synchLock.unlock();
}
for(InvitationListener l : listeners) l.remoteConfirmationFailed();
}
void pseudonymExchangeSucceeded(Author remoteAuthor) {
String name = remoteAuthor.getName();
synchronized(this) {
synchLock.lock();
try {
remoteName = name;
}
finally{
synchLock.unlock();
}
for(InvitationListener l : listeners)
l.pseudonymExchangeSucceeded(name);
}

View File

@@ -2,6 +2,8 @@ package org.briarproject.lifecycle;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.briarproject.api.lifecycle.ShutdownManager;
@@ -10,26 +12,42 @@ class ShutdownManagerImpl implements ShutdownManager {
protected final Map<Integer, Thread> hooks; // Locking: this
private int nextHandle = 0; // Locking: this
private final Lock synchLock = new ReentrantLock();
ShutdownManagerImpl() {
hooks = new HashMap<Integer, Thread>();
}
public synchronized int addShutdownHook(Runnable r) {
int handle = nextHandle++;
Thread hook = createThread(r);
hooks.put(handle, hook);
Runtime.getRuntime().addShutdownHook(hook);
return handle;
public int addShutdownHook(Runnable r) {
synchLock.lock();
try{
int handle = nextHandle++;
Thread hook = createThread(r);
hooks.put(handle, hook);
Runtime.getRuntime().addShutdownHook(hook);
return handle;
}
finally{
synchLock.unlock();
}
}
protected Thread createThread(Runnable r) {
return new Thread(r, "ShutdownManager");
}
public synchronized boolean removeShutdownHook(int handle) {
Thread hook = hooks.remove(handle);
if(hook == null) return false;
else return Runtime.getRuntime().removeShutdownHook(hook);
public boolean removeShutdownHook(int handle) {
synchLock.lock();
try{
Thread hook = hooks.remove(handle);
if(hook == null) return false;
else return Runtime.getRuntime().removeShutdownHook(hook);
}
finally{
synchLock.unlock();
}
}
}

View File

@@ -8,6 +8,8 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import org.briarproject.api.ContactId;
@@ -29,6 +31,8 @@ class ConnectionRegistryImpl implements ConnectionRegistry {
private final Map<TransportId, Map<ContactId, Integer>> connections;
// Locking: this
private final Map<ContactId, Integer> contactCounts;
private final Lock synchLock = new ReentrantLock();
@Inject
ConnectionRegistryImpl(EventBus eventBus) {
@@ -40,7 +44,8 @@ class ConnectionRegistryImpl implements ConnectionRegistry {
public void registerConnection(ContactId c, TransportId t) {
LOG.info("Connection registered");
boolean firstConnection = false;
synchronized(this) {
synchLock.lock();
try {
Map<ContactId, Integer> m = connections.get(t);
if(m == null) {
m = new HashMap<ContactId, Integer>();
@@ -57,6 +62,10 @@ class ConnectionRegistryImpl implements ConnectionRegistry {
contactCounts.put(c, count + 1);
}
}
finally{
synchLock.unlock();
}
if(firstConnection) {
LOG.info("Contact connected");
eventBus.broadcast(new ContactConnectedEvent(c));
@@ -66,7 +75,8 @@ class ConnectionRegistryImpl implements ConnectionRegistry {
public void unregisterConnection(ContactId c, TransportId t) {
LOG.info("Connection unregistered");
boolean lastConnection = false;
synchronized(this) {
synchLock.lock();
try {
Map<ContactId, Integer> m = connections.get(t);
if(m == null) throw new IllegalArgumentException();
Integer count = m.remove(c);
@@ -85,22 +95,40 @@ class ConnectionRegistryImpl implements ConnectionRegistry {
contactCounts.put(c, count - 1);
}
}
finally{
synchLock.unlock();
}
if(lastConnection) {
LOG.info("Contact disconnected");
eventBus.broadcast(new ContactDisconnectedEvent(c));
}
}
public synchronized Collection<ContactId> getConnectedContacts(
public Collection<ContactId> getConnectedContacts(
TransportId t) {
Map<ContactId, Integer> m = connections.get(t);
if(m == null) return Collections.emptyList();
List<ContactId> ids = new ArrayList<ContactId>(m.keySet());
if(LOG.isLoggable(INFO)) LOG.info(ids.size() + " contacts connected");
return Collections.unmodifiableList(ids);
synchLock.lock();
try{
Map<ContactId, Integer> m = connections.get(t);
if(m == null) return Collections.emptyList();
List<ContactId> ids = new ArrayList<ContactId>(m.keySet());
if(LOG.isLoggable(INFO)) LOG.info(ids.size() + " contacts connected");
return Collections.unmodifiableList(ids);
}
finally{
synchLock.unlock();
}
}
public synchronized boolean isConnected(ContactId c) {
return contactCounts.containsKey(c);
public boolean isConnected(ContactId c) {
synchLock.lock();
try{
return contactCounts.containsKey(c);
}
finally{
synchLock.unlock();
}
}
}

View File

@@ -5,6 +5,10 @@ import java.util.Comparator;
import java.util.Iterator;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.briarproject.api.reliability.ReadHandler;
import org.briarproject.api.system.Clock;
@@ -23,6 +27,8 @@ class Receiver implements ReadHandler {
private long nextSequenceNumber = 1;
private volatile boolean valid = true;
private Lock synchLock = new ReentrantLock();
private Condition dataFrameAvailable = synchLock.newCondition();
Receiver(Clock clock, Sender sender) {
this.sender = sender;
@@ -30,36 +36,46 @@ class Receiver implements ReadHandler {
dataFrames = new TreeSet<Data>(new SequenceNumberComparator());
}
synchronized Data read() throws IOException, InterruptedException {
long now = clock.currentTimeMillis(), end = now + READ_TIMEOUT;
while(now < end && valid) {
if(dataFrames.isEmpty()) {
// Wait for a data frame
wait(end - now);
} else {
Data d = dataFrames.first();
if(d.getSequenceNumber() == nextSequenceNumber) {
dataFrames.remove(d);
// Update the window
windowSize += d.getPayloadLength();
sender.sendAck(0, windowSize);
nextSequenceNumber++;
return d;
Data read() throws IOException, InterruptedException {
synchLock.lock();
try{
long now = clock.currentTimeMillis(), end = now + READ_TIMEOUT;
while(now < end && valid) {
if(dataFrames.isEmpty()) {
// Wait for a data frame
dataFrameAvailable.await(end - now, TimeUnit.MILLISECONDS);
} else {
// Wait for the next in-order data frame
wait(end - now);
Data d = dataFrames.first();
if(d.getSequenceNumber() == nextSequenceNumber) {
dataFrames.remove(d);
// Update the window
windowSize += d.getPayloadLength();
sender.sendAck(0, windowSize);
nextSequenceNumber++;
return d;
} else {
// Wait for the next in-order data frame
dataFrameAvailable.await(end - now, TimeUnit.MILLISECONDS);
}
}
now = clock.currentTimeMillis();
}
now = clock.currentTimeMillis();
if(valid) throw new IOException("Read timed out");
throw new IOException("Connection closed");
}
finally{
synchLock.unlock();
}
if(valid) throw new IOException("Read timed out");
throw new IOException("Connection closed");
}
void invalidate() {
valid = false;
synchronized(this) {
notifyAll();
synchLock.lock();
try {
dataFrameAvailable.signalAll();
}
finally{
synchLock.unlock();
}
}
@@ -79,43 +95,49 @@ class Receiver implements ReadHandler {
}
}
private synchronized void handleData(byte[] b) throws IOException {
if(b.length < Data.MIN_LENGTH || b.length > Data.MAX_LENGTH) {
// Ignore data frame with invalid length
return;
}
Data d = new Data(b);
int payloadLength = d.getPayloadLength();
if(payloadLength > windowSize) return; // No space in the window
if(d.getChecksum() != d.calculateChecksum()) {
// Ignore data frame with invalid checksum
return;
}
long sequenceNumber = d.getSequenceNumber();
if(sequenceNumber == 0) {
// Window probe
} else if(sequenceNumber < nextSequenceNumber) {
// Duplicate data frame
} else if(d.isLastFrame()) {
finalSequenceNumber = sequenceNumber;
// Remove any data frames with higher sequence numbers
Iterator<Data> it = dataFrames.iterator();
while(it.hasNext()) {
Data d1 = it.next();
if(d1.getSequenceNumber() >= finalSequenceNumber) it.remove();
private void handleData(byte[] b) throws IOException {
synchLock.lock();
try{
if(b.length < Data.MIN_LENGTH || b.length > Data.MAX_LENGTH) {
// Ignore data frame with invalid length
return;
}
if(dataFrames.add(d)) {
windowSize -= payloadLength;
notifyAll();
Data d = new Data(b);
int payloadLength = d.getPayloadLength();
if(payloadLength > windowSize) return; // No space in the window
if(d.getChecksum() != d.calculateChecksum()) {
// Ignore data frame with invalid checksum
return;
}
} else if(sequenceNumber < finalSequenceNumber) {
if(dataFrames.add(d)) {
windowSize -= payloadLength;
notifyAll();
long sequenceNumber = d.getSequenceNumber();
if(sequenceNumber == 0) {
// Window probe
} else if(sequenceNumber < nextSequenceNumber) {
// Duplicate data frame
} else if(d.isLastFrame()) {
finalSequenceNumber = sequenceNumber;
// Remove any data frames with higher sequence numbers
Iterator<Data> it = dataFrames.iterator();
while(it.hasNext()) {
Data d1 = it.next();
if(d1.getSequenceNumber() >= finalSequenceNumber) it.remove();
}
if(dataFrames.add(d)) {
windowSize -= payloadLength;
dataFrameAvailable.signalAll();
}
} else if(sequenceNumber < finalSequenceNumber) {
if(dataFrames.add(d)) {
windowSize -= payloadLength;
dataFrameAvailable.signalAll();
}
}
// Acknowledge the data frame even if it's a duplicate
sender.sendAck(sequenceNumber, windowSize);
}
finally{
synchLock.unlock();
}
// Acknowledge the data frame even if it's a duplicate
sender.sendAck(sequenceNumber, windowSize);
}
private static class SequenceNumberComparator implements Comparator<Data> {

View File

@@ -5,6 +5,10 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.briarproject.api.reliability.WriteHandler;
import org.briarproject.api.system.Clock;
@@ -31,6 +35,9 @@ class Sender {
private long lastWindowUpdateOrProbe = Long.MAX_VALUE;
private boolean dataWaiting = false;
private Lock synchLock = new ReentrantLock();
private Condition sendWindowAvailable = synchLock.newCondition();
Sender(Clock clock, WriteHandler writeHandler) {
this.clock = clock;
this.writeHandler = writeHandler;
@@ -58,7 +65,8 @@ class Sender {
long sequenceNumber = a.getSequenceNumber();
long now = clock.currentTimeMillis();
Outstanding fastRetransmit = null;
synchronized(this) {
synchLock.lock();
try {
// Remove the acked data frame if it's outstanding
int foundIndex = -1;
Iterator<Outstanding> it = outstanding.iterator();
@@ -96,6 +104,9 @@ class Sender {
// If space has become available, notify any waiting writers
if(windowSize > oldWindowSize || foundIndex != -1) notifyAll();
}
finally{
synchLock.unlock();
}
// Fast retransmission
if(fastRetransmit != null)
writeHandler.handleWrite(fastRetransmit.data.getBuffer());
@@ -105,7 +116,8 @@ class Sender {
long now = clock.currentTimeMillis();
List<Outstanding> retransmit = null;
boolean sendProbe = false;
synchronized(this) {
synchLock.lock();
try {
if(outstanding.isEmpty()) {
if(dataWaiting && now - lastWindowUpdateOrProbe > rto) {
sendProbe = true;
@@ -135,6 +147,9 @@ class Sender {
}
}
}
finally{
synchLock.unlock();
}
// Send a window probe if necessary
if(sendProbe) {
byte[] buf = new byte[Data.MIN_LENGTH];
@@ -151,12 +166,13 @@ class Sender {
void write(Data d) throws IOException, InterruptedException {
int payloadLength = d.getPayloadLength();
synchronized(this) {
synchLock.lock();
try {
// Wait for space in the window
long now = clock.currentTimeMillis(), end = now + WRITE_TIMEOUT;
while(now < end && outstandingBytes + payloadLength >= windowSize) {
dataWaiting = true;
wait(end - now);
sendWindowAvailable.await(end - now, TimeUnit.MILLISECONDS);
now = clock.currentTimeMillis();
}
if(outstandingBytes + payloadLength >= windowSize)
@@ -165,11 +181,20 @@ class Sender {
outstandingBytes += payloadLength;
dataWaiting = false;
}
finally{
synchLock.unlock();
}
writeHandler.handleWrite(d.getBuffer());
}
synchronized void flush() throws IOException, InterruptedException {
while(dataWaiting || !outstanding.isEmpty()) wait();
void flush() throws IOException, InterruptedException {
synchLock.lock();
try{
while(dataWaiting || !outstanding.isEmpty()) sendWindowAvailable.await();
}
finally{
synchLock.unlock();
}
}
private static class Outstanding {

View File

@@ -11,6 +11,8 @@ import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.TimerTask;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import javax.inject.Inject;
@@ -56,6 +58,8 @@ class KeyManagerImpl extends TimerTask implements KeyManager, EventListener {
private final Map<EndpointKey, TemporarySecret> currentSecrets;
private final Map<EndpointKey, TemporarySecret> newSecrets;
private final Lock synchLock = new ReentrantLock();
@Inject
KeyManagerImpl(CryptoComponent crypto, DatabaseComponent db,
EventBus eventBus, TagRecogniser tagRecogniser, Clock clock,
@@ -72,41 +76,47 @@ class KeyManagerImpl extends TimerTask implements KeyManager, EventListener {
newSecrets = new HashMap<EndpointKey, TemporarySecret>();
}
public synchronized boolean start() {
eventBus.addListener(this);
// Load the temporary secrets and transport latencies from the database
Collection<TemporarySecret> secrets;
public boolean start() {
synchLock.lock();
try {
secrets = db.getSecrets();
maxLatencies.putAll(db.getTransportLatencies());
} catch(DbException e) {
if(LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
return false;
}
// Work out what phase of its lifecycle each secret is in
long now = clock.currentTimeMillis();
Collection<TemporarySecret> dead = assignSecretsToMaps(now, secrets);
// Replace any dead secrets
Collection<TemporarySecret> created = replaceDeadSecrets(now, dead);
if(!created.isEmpty()) {
// Store any secrets that have been created, removing any dead ones
eventBus.addListener(this);
// Load the temporary secrets and transport latencies from the database
Collection<TemporarySecret> secrets;
try {
db.addSecrets(created);
secrets = db.getSecrets();
maxLatencies.putAll(db.getTransportLatencies());
} catch(DbException e) {
if(LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
return false;
}
// Work out what phase of its lifecycle each secret is in
long now = clock.currentTimeMillis();
Collection<TemporarySecret> dead = assignSecretsToMaps(now, secrets);
// Replace any dead secrets
Collection<TemporarySecret> created = replaceDeadSecrets(now, dead);
if(!created.isEmpty()) {
// Store any secrets that have been created, removing any dead ones
try {
db.addSecrets(created);
} catch(DbException e) {
if(LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
return false;
}
}
// Pass the old, current and new secrets to the recogniser
for(TemporarySecret s : oldSecrets.values())
tagRecogniser.addSecret(s);
for(TemporarySecret s : currentSecrets.values())
tagRecogniser.addSecret(s);
for(TemporarySecret s : newSecrets.values())
tagRecogniser.addSecret(s);
// Schedule periodic key rotation
timer.scheduleAtFixedRate(this, MS_BETWEEN_CHECKS, MS_BETWEEN_CHECKS);
return true;
}
finally{
synchLock.unlock();
}
// Pass the old, current and new secrets to the recogniser
for(TemporarySecret s : oldSecrets.values())
tagRecogniser.addSecret(s);
for(TemporarySecret s : currentSecrets.values())
tagRecogniser.addSecret(s);
for(TemporarySecret s : newSecrets.values())
tagRecogniser.addSecret(s);
// Schedule periodic key rotation
timer.scheduleAtFixedRate(this, MS_BETWEEN_CHECKS, MS_BETWEEN_CHECKS);
return true;
}
// Assigns secrets to the appropriate maps and returns any dead secrets
@@ -215,15 +225,21 @@ class KeyManagerImpl extends TimerTask implements KeyManager, EventListener {
return created;
}
public synchronized boolean stop() {
eventBus.removeListener(this);
timer.cancel();
tagRecogniser.removeSecrets();
maxLatencies.clear();
removeAndEraseSecrets(oldSecrets);
removeAndEraseSecrets(currentSecrets);
removeAndEraseSecrets(newSecrets);
return true;
public boolean stop() {
synchLock.lock();
try{
eventBus.removeListener(this);
timer.cancel();
tagRecogniser.removeSecrets();
maxLatencies.clear();
removeAndEraseSecrets(oldSecrets);
removeAndEraseSecrets(currentSecrets);
removeAndEraseSecrets(newSecrets);
return true;
}
finally{
synchLock.unlock();
}
}
// Locking: this
@@ -232,98 +248,116 @@ class KeyManagerImpl extends TimerTask implements KeyManager, EventListener {
m.clear();
}
public synchronized StreamContext getStreamContext(ContactId c,
public StreamContext getStreamContext(ContactId c,
TransportId t) {
TemporarySecret s = currentSecrets.get(new EndpointKey(c, t));
if(s == null) {
LOG.info("No secret for endpoint");
return null;
}
long streamNumber;
try {
streamNumber = db.incrementStreamCounter(c, t, s.getPeriod());
if(streamNumber == -1) {
LOG.info("No counter for period");
synchLock.lock();
try{
TemporarySecret s = currentSecrets.get(new EndpointKey(c, t));
if(s == null) {
LOG.info("No secret for endpoint");
return null;
}
} catch(DbException e) {
if(LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
return null;
long streamNumber;
try {
streamNumber = db.incrementStreamCounter(c, t, s.getPeriod());
if(streamNumber == -1) {
LOG.info("No counter for period");
return null;
}
} catch(DbException e) {
if(LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
return null;
}
// Clone the secret - the original will be erased
byte[] secret = s.getSecret().clone();
return new StreamContext(c, t, secret, streamNumber, s.getAlice());
}
finally{
synchLock.unlock();
}
// Clone the secret - the original will be erased
byte[] secret = s.getSecret().clone();
return new StreamContext(c, t, secret, streamNumber, s.getAlice());
}
public synchronized void endpointAdded(Endpoint ep, long maxLatency,
public void endpointAdded(Endpoint ep, long maxLatency,
byte[] initialSecret) {
maxLatencies.put(ep.getTransportId(), maxLatency);
// Work out which rotation period we're in
long elapsed = clock.currentTimeMillis() - ep.getEpoch();
long rotation = maxLatency + MAX_CLOCK_DIFFERENCE;
long period = (elapsed / rotation) + 1;
if(period < 1) throw new IllegalStateException();
// Derive the old, current and new secrets
byte[] b1 = initialSecret;
for(long p = 0; p < period; p++) {
byte[] temp = crypto.deriveNextSecret(b1, p);
ByteUtils.erase(b1);
b1 = temp;
synchLock.lock();
try{
maxLatencies.put(ep.getTransportId(), maxLatency);
// Work out which rotation period we're in
long elapsed = clock.currentTimeMillis() - ep.getEpoch();
long rotation = maxLatency + MAX_CLOCK_DIFFERENCE;
long period = (elapsed / rotation) + 1;
if(period < 1) throw new IllegalStateException();
// Derive the old, current and new secrets
byte[] b1 = initialSecret;
for(long p = 0; p < period; p++) {
byte[] temp = crypto.deriveNextSecret(b1, p);
ByteUtils.erase(b1);
b1 = temp;
}
byte[] b2 = crypto.deriveNextSecret(b1, period);
byte[] b3 = crypto.deriveNextSecret(b2, period + 1);
TemporarySecret s1 = new TemporarySecret(ep, period - 1, b1);
TemporarySecret s2 = new TemporarySecret(ep, period, b2);
TemporarySecret s3 = new TemporarySecret(ep, period + 1, b3);
// Add the incoming secrets to their respective maps
EndpointKey k = new EndpointKey(ep);
oldSecrets.put(k, s1);
currentSecrets.put(k, s2);
newSecrets.put(k, s3);
// Store the new secrets
try {
db.addSecrets(Arrays.asList(s1, s2, s3));
} catch(DbException e) {
if(LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
return;
}
// Pass the new secrets to the recogniser
tagRecogniser.addSecret(s1);
tagRecogniser.addSecret(s2);
tagRecogniser.addSecret(s3);
}
byte[] b2 = crypto.deriveNextSecret(b1, period);
byte[] b3 = crypto.deriveNextSecret(b2, period + 1);
TemporarySecret s1 = new TemporarySecret(ep, period - 1, b1);
TemporarySecret s2 = new TemporarySecret(ep, period, b2);
TemporarySecret s3 = new TemporarySecret(ep, period + 1, b3);
// Add the incoming secrets to their respective maps
EndpointKey k = new EndpointKey(ep);
oldSecrets.put(k, s1);
currentSecrets.put(k, s2);
newSecrets.put(k, s3);
// Store the new secrets
try {
db.addSecrets(Arrays.asList(s1, s2, s3));
} catch(DbException e) {
if(LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
return;
finally{
synchLock.unlock();
}
// Pass the new secrets to the recogniser
tagRecogniser.addSecret(s1);
tagRecogniser.addSecret(s2);
tagRecogniser.addSecret(s3);
}
@Override
public synchronized void run() {
public void run() {
synchLock.lock();
try{
// Rebuild the maps because we may be running a whole period late
Collection<TemporarySecret> secrets = new ArrayList<TemporarySecret>();
secrets.addAll(oldSecrets.values());
secrets.addAll(currentSecrets.values());
secrets.addAll(newSecrets.values());
oldSecrets.clear();
currentSecrets.clear();
newSecrets.clear();
// Work out what phase of its lifecycle each secret is in
long now = clock.currentTimeMillis();
Collection<TemporarySecret> dead = assignSecretsToMaps(now, secrets);
// Remove any dead secrets from the recogniser
for(TemporarySecret s : dead) {
ContactId c = s.getContactId();
TransportId t = s.getTransportId();
long period = s.getPeriod();
tagRecogniser.removeSecret(c, t, period);
}
// Replace any dead secrets
Collection<TemporarySecret> created = replaceDeadSecrets(now, dead);
if(!created.isEmpty()) {
// Store any secrets that have been created
try {
db.addSecrets(created);
} catch(DbException e) {
if(LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
Collection<TemporarySecret> secrets = new ArrayList<TemporarySecret>();
secrets.addAll(oldSecrets.values());
secrets.addAll(currentSecrets.values());
secrets.addAll(newSecrets.values());
oldSecrets.clear();
currentSecrets.clear();
newSecrets.clear();
// Work out what phase of its lifecycle each secret is in
long now = clock.currentTimeMillis();
Collection<TemporarySecret> dead = assignSecretsToMaps(now, secrets);
// Remove any dead secrets from the recogniser
for(TemporarySecret s : dead) {
ContactId c = s.getContactId();
TransportId t = s.getTransportId();
long period = s.getPeriod();
tagRecogniser.removeSecret(c, t, period);
}
// Pass any secrets that have been created to the recogniser
for(TemporarySecret s : created) tagRecogniser.addSecret(s);
// Replace any dead secrets
Collection<TemporarySecret> created = replaceDeadSecrets(now, dead);
if(!created.isEmpty()) {
// Store any secrets that have been created
try {
db.addSecrets(created);
} catch(DbException e) {
if(LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
}
// Pass any secrets that have been created to the recogniser
for(TemporarySecret s : created) tagRecogniser.addSecret(s);
}
}
finally{
synchLock.unlock();
}
}
@@ -407,11 +441,15 @@ class KeyManagerImpl extends TimerTask implements KeyManager, EventListener {
public void run() {
ContactId c = event.getContactId();
tagRecogniser.removeSecrets(c);
synchronized(KeyManagerImpl.this) {
synchLock.lock();
try {
removeAndEraseSecrets(c, oldSecrets);
removeAndEraseSecrets(c, currentSecrets);
removeAndEraseSecrets(c, newSecrets);
}
finally{
synchLock.unlock();
}
}
}
@@ -425,9 +463,13 @@ class KeyManagerImpl extends TimerTask implements KeyManager, EventListener {
@Override
public void run() {
synchronized(KeyManagerImpl.this) {
synchLock.lock();
try {
maxLatencies.put(event.getTransportId(), event.getMaxLatency());
}
finally{
synchLock.unlock();
}
}
}
@@ -443,12 +485,16 @@ class KeyManagerImpl extends TimerTask implements KeyManager, EventListener {
public void run() {
TransportId t = event.getTransportId();
tagRecogniser.removeSecrets(t);
synchronized(KeyManagerImpl.this) {
synchLock.lock();
try {
maxLatencies.remove(t);
removeAndEraseSecrets(t, oldSecrets);
removeAndEraseSecrets(t, currentSecrets);
removeAndEraseSecrets(t, newSecrets);
}
finally{
synchLock.unlock();
}
}
}
}

View File

@@ -2,6 +2,8 @@ package org.briarproject.transport;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.inject.Inject;
@@ -20,6 +22,9 @@ class TagRecogniserImpl implements TagRecogniser {
private final DatabaseComponent db;
// Locking: this
private final Map<TransportId, TransportTagRecogniser> recognisers;
private final Lock synchLock = new ReentrantLock();
@Inject
TagRecogniserImpl(CryptoComponent crypto, DatabaseComponent db) {
@@ -31,9 +36,13 @@ class TagRecogniserImpl implements TagRecogniser {
public StreamContext recogniseTag(TransportId t, byte[] tag)
throws DbException {
TransportTagRecogniser r;
synchronized(this) {
synchLock.lock();
try {
r = recognisers.get(t);
}
finally{
synchLock.unlock();
}
if(r == null) return null;
return r.recogniseTag(tag);
}
@@ -41,35 +50,63 @@ class TagRecogniserImpl implements TagRecogniser {
public void addSecret(TemporarySecret s) {
TransportId t = s.getTransportId();
TransportTagRecogniser r;
synchronized(this) {
synchLock.lock();
try {
r = recognisers.get(t);
if(r == null) {
r = new TransportTagRecogniser(crypto, db, t);
recognisers.put(t, r);
}
}
finally{
synchLock.unlock();
}
r.addSecret(s);
}
public void removeSecret(ContactId c, TransportId t, long period) {
TransportTagRecogniser r;
synchronized(this) {
synchLock.lock();
try {
r = recognisers.get(t);
}
finally{
synchLock.unlock();
}
if(r != null) r.removeSecret(c, period);
}
public synchronized void removeSecrets(ContactId c) {
for(TransportTagRecogniser r : recognisers.values())
r.removeSecrets(c);
public void removeSecrets(ContactId c) {
synchLock.lock();
try{
for(TransportTagRecogniser r : recognisers.values())
r.removeSecrets(c);
}
finally{
synchLock.unlock();
}
}
public synchronized void removeSecrets(TransportId t) {
recognisers.remove(t);
public void removeSecrets(TransportId t) {
synchLock.lock();
try{
recognisers.remove(t);
}
finally{
synchLock.unlock();
}
}
public synchronized void removeSecrets() {
for(TransportTagRecogniser r : recognisers.values())
r.removeSecrets();
public void removeSecrets() {
synchLock.lock();
try{
for(TransportTagRecogniser r : recognisers.values())
r.removeSecrets();
}
finally{
synchLock.unlock();
}
}
}

View File

@@ -6,6 +6,8 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.briarproject.api.Bytes;
import org.briarproject.api.ContactId;
@@ -30,6 +32,8 @@ class TransportTagRecogniser {
private final Map<Bytes, TagContext> tagMap; // Locking: this
private final Map<RemovalKey, RemovalContext> removalMap; // Locking: this
private final Lock synchLock = new ReentrantLock();
TransportTagRecogniser(CryptoComponent crypto, DatabaseComponent db,
TransportId transportId) {
this.crypto = crypto;
@@ -39,62 +43,80 @@ class TransportTagRecogniser {
removalMap = new HashMap<RemovalKey, RemovalContext>();
}
synchronized StreamContext recogniseTag(byte[] tag) throws DbException {
TagContext t = tagMap.remove(new Bytes(tag));
if(t == null) return null; // The tag was not expected
// Update the reordering window and the expected tags
SecretKey key = crypto.deriveTagKey(t.secret, !t.alice);
for(long streamNumber : t.window.setSeen(t.streamNumber)) {
byte[] tag1 = new byte[TAG_LENGTH];
crypto.encodeTag(tag1, key, streamNumber);
if(streamNumber < t.streamNumber) {
TagContext removed = tagMap.remove(new Bytes(tag1));
assert removed != null;
} else {
TagContext added = new TagContext(t, streamNumber);
TagContext duplicate = tagMap.put(new Bytes(tag1), added);
StreamContext recogniseTag(byte[] tag) throws DbException {
synchLock.lock();
try{
TagContext t = tagMap.remove(new Bytes(tag));
if(t == null) return null; // The tag was not expected
// Update the reordering window and the expected tags
SecretKey key = crypto.deriveTagKey(t.secret, !t.alice);
for(long streamNumber : t.window.setSeen(t.streamNumber)) {
byte[] tag1 = new byte[TAG_LENGTH];
crypto.encodeTag(tag1, key, streamNumber);
if(streamNumber < t.streamNumber) {
TagContext removed = tagMap.remove(new Bytes(tag1));
assert removed != null;
} else {
TagContext added = new TagContext(t, streamNumber);
TagContext duplicate = tagMap.put(new Bytes(tag1), added);
assert duplicate == null;
}
}
key.erase();
// Store the updated reordering window in the DB
db.setReorderingWindow(t.contactId, transportId, t.period,
t.window.getCentre(), t.window.getBitmap());
// Clone the secret - the key manager will erase the original
byte[] secret = t.secret.clone();
return new StreamContext(t.contactId, transportId, secret,
t.streamNumber, t.alice);
}
finally{
synchLock.unlock();
}
}
void addSecret(TemporarySecret s) {
synchLock.lock();
try{
ContactId contactId = s.getContactId();
boolean alice = s.getAlice();
long period = s.getPeriod();
byte[] secret = s.getSecret();
long centre = s.getWindowCentre();
byte[] bitmap = s.getWindowBitmap();
// Create the reordering window and the expected tags
SecretKey key = crypto.deriveTagKey(secret, !alice);
ReorderingWindow window = new ReorderingWindow(centre, bitmap);
for(long streamNumber : window.getUnseen()) {
byte[] tag = new byte[TAG_LENGTH];
crypto.encodeTag(tag, key, streamNumber);
TagContext added = new TagContext(contactId, alice, period,
secret, window, streamNumber);
TagContext duplicate = tagMap.put(new Bytes(tag), added);
assert duplicate == null;
}
key.erase();
// Create a removal context to remove the window and the tags later
RemovalContext r = new RemovalContext(window, secret, alice);
removalMap.put(new RemovalKey(contactId, period), r);
}
finally{
synchLock.unlock();
}
key.erase();
// Store the updated reordering window in the DB
db.setReorderingWindow(t.contactId, transportId, t.period,
t.window.getCentre(), t.window.getBitmap());
// Clone the secret - the key manager will erase the original
byte[] secret = t.secret.clone();
return new StreamContext(t.contactId, transportId, secret,
t.streamNumber, t.alice);
}
synchronized void addSecret(TemporarySecret s) {
ContactId contactId = s.getContactId();
boolean alice = s.getAlice();
long period = s.getPeriod();
byte[] secret = s.getSecret();
long centre = s.getWindowCentre();
byte[] bitmap = s.getWindowBitmap();
// Create the reordering window and the expected tags
SecretKey key = crypto.deriveTagKey(secret, !alice);
ReorderingWindow window = new ReorderingWindow(centre, bitmap);
for(long streamNumber : window.getUnseen()) {
byte[] tag = new byte[TAG_LENGTH];
crypto.encodeTag(tag, key, streamNumber);
TagContext added = new TagContext(contactId, alice, period,
secret, window, streamNumber);
TagContext duplicate = tagMap.put(new Bytes(tag), added);
assert duplicate == null;
void removeSecret(ContactId contactId, long period) {
synchLock.lock();
try{
RemovalKey k = new RemovalKey(contactId, period);
RemovalContext removed = removalMap.remove(k);
if(removed == null) throw new IllegalArgumentException();
removeSecret(removed);
}
finally{
synchLock.unlock();
}
key.erase();
// Create a removal context to remove the window and the tags later
RemovalContext r = new RemovalContext(window, secret, alice);
removalMap.put(new RemovalKey(contactId, period), r);
}
synchronized void removeSecret(ContactId contactId, long period) {
RemovalKey k = new RemovalKey(contactId, period);
RemovalContext removed = removalMap.remove(k);
if(removed == null) throw new IllegalArgumentException();
removeSecret(removed);
}
// Locking: this
@@ -110,17 +132,29 @@ class TransportTagRecogniser {
key.erase();
}
synchronized void removeSecrets(ContactId c) {
Collection<RemovalKey> keysToRemove = new ArrayList<RemovalKey>();
for(RemovalKey k : removalMap.keySet())
if(k.contactId.equals(c)) keysToRemove.add(k);
for(RemovalKey k : keysToRemove) removeSecret(k.contactId, k.period);
void removeSecrets(ContactId c) {
synchLock.lock();
try{
Collection<RemovalKey> keysToRemove = new ArrayList<RemovalKey>();
for(RemovalKey k : removalMap.keySet())
if(k.contactId.equals(c)) keysToRemove.add(k);
for(RemovalKey k : keysToRemove) removeSecret(k.contactId, k.period);
}
finally{
synchLock.unlock();
}
}
synchronized void removeSecrets() {
for(RemovalContext r : removalMap.values()) removeSecret(r);
assert tagMap.isEmpty();
removalMap.clear();
void removeSecrets() {
synchLock.lock();
try{
for(RemovalContext r : removalMap.values()) removeSecret(r);
assert tagMap.isEmpty();
removalMap.clear();
}
finally{
synchLock.unlock();
}
}
private static class TagContext {

View File

@@ -8,6 +8,8 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import org.briarproject.util.OsUtils;
@@ -38,6 +40,9 @@ class WindowsShutdownManagerImpl extends ShutdownManagerImpl {
private final Map<String, Object> options;
private boolean initialised = false; // Locking: this
private final Lock synchLock = new ReentrantLock();
WindowsShutdownManagerImpl() {
// Use the Unicode versions of Win32 API calls
@@ -48,9 +53,15 @@ class WindowsShutdownManagerImpl extends ShutdownManagerImpl {
}
@Override
public synchronized int addShutdownHook(Runnable r) {
if(!initialised) initialise();
return super.addShutdownHook(r);
public int addShutdownHook(Runnable r) {
synchLock.lock();
try {
if(!initialised) initialise();
return super.addShutdownHook(r);
}
finally{
synchLock.unlock();
}
}
@Override
@@ -69,20 +80,26 @@ class WindowsShutdownManagerImpl extends ShutdownManagerImpl {
}
// Package access for testing
synchronized void runShutdownHooks() {
boolean interrupted = false;
// Start each hook in its own thread
for(Thread hook : hooks.values()) hook.start();
// Wait for all the hooks to finish
for(Thread hook : hooks.values()) {
try {
hook.join();
} catch(InterruptedException e) {
LOG.warning("Interrupted while running shutdown hooks");
interrupted = true;
void runShutdownHooks() {
synchLock.lock();
try {
boolean interrupted = false;
// Start each hook in its own thread
for(Thread hook : hooks.values()) hook.start();
// Wait for all the hooks to finish
for(Thread hook : hooks.values()) {
try {
hook.join();
} catch(InterruptedException e) {
LOG.warning("Interrupted while running shutdown hooks");
interrupted = true;
}
}
if(interrupted) Thread.currentThread().interrupt();
}
finally{
synchLock.unlock();
}
if(interrupted) Thread.currentThread().interrupt();
}
private class EventLoop extends Thread {

View File

@@ -4,6 +4,10 @@ import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
class PollingRemovableDriveMonitor implements RemovableDriveMonitor, Runnable {
@@ -14,11 +18,14 @@ class PollingRemovableDriveMonitor implements RemovableDriveMonitor, Runnable {
private final Executor ioExecutor;
private final RemovableDriveFinder finder;
private final long pollingInterval;
private final Object pollingLock = new Object();
private volatile boolean running = false;
private volatile Callback callback = null;
private final Lock synchLock = new ReentrantLock();
private final Condition stopPolling = synchLock.newCondition();
public PollingRemovableDriveMonitor(Executor ioExecutor,
RemovableDriveFinder finder, long pollingInterval) {
this.ioExecutor = ioExecutor;
@@ -34,8 +41,12 @@ class PollingRemovableDriveMonitor implements RemovableDriveMonitor, Runnable {
public void stop() throws IOException {
running = false;
synchronized(pollingLock) {
pollingLock.notifyAll();
synchLock.lock();
try {
stopPolling.signalAll();
}
finally {
synchLock.unlock();
}
}
@@ -43,8 +54,12 @@ class PollingRemovableDriveMonitor implements RemovableDriveMonitor, Runnable {
try {
Collection<File> drives = finder.findRemovableDrives();
while(running) {
synchronized(pollingLock) {
pollingLock.wait(pollingInterval);
synchLock.lock();
try {
stopPolling.await(pollingInterval, TimeUnit.MILLISECONDS);
}
finally{
synchLock.unlock();
}
if(!running) return;
Collection<File> newDrives = finder.findRemovableDrives();

View File

@@ -4,6 +4,9 @@ import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import net.contentobjects.jnotify.JNotify;
import net.contentobjects.jnotify.JNotifyListener;
@@ -21,7 +24,11 @@ JNotifyListener {
private Callback callback = null; // Locking: this
protected abstract String[] getPathsToWatch();
//TODO: rationalise this in a further refactor
private final Lock synchLock = new ReentrantLock();
private static final Lock staticSynchLock = new ReentrantLock();
private static Throwable tryLoad() {
try {
Class.forName("net.contentobjects.jnotify.JNotify");
@@ -33,12 +40,18 @@ JNotifyListener {
}
}
public static synchronized void checkEnabled() throws IOException {
if(!triedLoad) {
loadError = tryLoad();
triedLoad = true;
public static void checkEnabled() throws IOException {
staticSynchLock.lock();
try {
if(!triedLoad) {
loadError = tryLoad();
triedLoad = true;
}
if(loadError != null) throw new IOException(loadError.toString());
}
finally{
staticSynchLock.unlock();
}
if(loadError != null) throw new IOException(loadError.toString());
}
public void start(Callback callback) throws IOException {
@@ -49,34 +62,46 @@ JNotifyListener {
if(new File(path).exists())
watches.add(JNotify.addWatch(path, mask, false, this));
}
synchronized(this) {
assert !started;
assert this.callback == null;
started = true;
this.callback = callback;
this.watches.addAll(watches);
}
synchLock.lock();
try {
assert !started;
assert this.callback == null;
started = true;
this.callback = callback;
this.watches.addAll(watches);
}
finally{
synchLock.unlock();
}
}
public void stop() throws IOException {
checkEnabled();
List<Integer> watches;
synchronized(this) {
assert started;
assert callback != null;
started = false;
callback = null;
watches = new ArrayList<Integer>(this.watches);
this.watches.clear();
}
synchLock.lock();
try {
assert started;
assert callback != null;
started = false;
callback = null;
watches = new ArrayList<Integer>(this.watches);
this.watches.clear();
}
finally{
synchLock.unlock();
}
for(Integer w : watches) JNotify.removeWatch(w);
}
public void fileCreated(int wd, String rootPath, String name) {
Callback callback;
synchronized(this) {
callback = this.callback;
}
synchLock.lock();
try {
callback = this.callback;
}
finally{
synchLock.unlock();
}
if(callback != null)
callback.driveInserted(new File(rootPath + "/" + name));
}

View File

@@ -10,6 +10,10 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.util.concurrent.Executor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import jssc.SerialPortEvent;
@@ -44,6 +48,11 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
private ReliabilityLayer reliability = null; // Locking: this
private boolean initialised = false, connected = false; // Locking: this
private final Lock synchLock = new ReentrantLock();
private final Condition connectedStateChanged = synchLock.newCondition();
private final Condition initialisedStateChanged = synchLock.newCondition();
ModemImpl(Executor executor, ReliabilityLayerFactory reliabilityFactory,
Clock clock, Callback callback, SerialPort port) {
@@ -91,14 +100,18 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
// Wait for the event thread to receive "OK"
boolean success = false;
try {
synchronized(this) {
synchLock.lock();
try {
long now = clock.currentTimeMillis();
long end = now + OK_TIMEOUT;
while(now < end && !initialised) {
wait(end - now);
initialisedStateChanged.await(end - now, TimeUnit.MILLISECONDS);
now = clock.currentTimeMillis();
}
success = initialised;
}
finally{
synchLock.unlock();
}
} catch(InterruptedException e) {
tryToClose(port);
@@ -123,11 +136,16 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
public void stop() throws IOException {
LOG.info("Stopping");
// Wake any threads that are waiting to connect
synchronized(this) {
synchLock.lock();
try {
// Wake any threads that are waiting to connect
initialised = false;
connected = false;
notifyAll();
initialisedStateChanged.signalAll();
connectedStateChanged.signalAll();
}
finally{
synchLock.unlock();
}
// Hang up if necessary and close the port
try {
@@ -148,7 +166,8 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
// Locking: stateChange
private void hangUpInner() throws IOException {
ReliabilityLayer reliability;
synchronized(this) {
synchLock.lock();
try {
if(this.reliability == null) {
LOG.info("Not hanging up - already on the hook");
return;
@@ -156,6 +175,9 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
reliability = this.reliability;
this.reliability = null;
connected = false;
}
finally{
synchLock.unlock();
}
reliability.stop();
LOG.info("Hanging up");
@@ -182,7 +204,8 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
try {
ReliabilityLayer reliability =
reliabilityFactory.createReliabilityLayer(this);
synchronized(this) {
synchLock.lock();
try {
if(!initialised) {
LOG.info("Not dialling - modem not initialised");
return false;
@@ -192,6 +215,9 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
return false;
}
this.reliability = reliability;
}
finally{
synchLock.unlock();
}
reliability.start();
LOG.info("Dialling");
@@ -204,14 +230,18 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
}
// Wait for the event thread to receive "CONNECT"
try {
synchronized(this) {
synchLock.lock();
try {
long now = clock.currentTimeMillis();
long end = now + CONNECT_TIMEOUT;
while(now < end && initialised && !connected) {
wait(end - now);
connectedStateChanged.await(end - now, TimeUnit.MILLISECONDS);
now = clock.currentTimeMillis();
}
if(connected) return true;
}
finally{
synchLock.unlock();
}
} catch(InterruptedException e) {
tryToClose(port);
@@ -227,8 +257,12 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
public InputStream getInputStream() throws IOException {
ReliabilityLayer reliability;
synchronized(this) {
synchLock.lock();
try {
reliability = this.reliability;
}
finally{
synchLock.unlock();
}
if(reliability == null) throw new IOException("Not connected");
return reliability.getInputStream();
@@ -236,8 +270,12 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
public OutputStream getOutputStream() throws IOException {
ReliabilityLayer reliability;
synchronized(this) {
synchLock.lock();
try {
reliability = this.reliability;
}
finally{
synchLock.unlock();
}
if(reliability == null) throw new IOException("Not connected");
return reliability.getOutputStream();
@@ -288,8 +326,12 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
private boolean handleData(byte[] b) throws IOException {
ReliabilityLayer reliability;
synchronized(this) {
synchLock.lock();
try {
reliability = this.reliability;
}
finally{
synchLock.unlock();
}
if(reliability == null) return false;
reliability.handleRead(b);
@@ -309,9 +351,13 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
lineLen = 0;
if(LOG.isLoggable(INFO)) LOG.info("Modem status: " + s);
if(s.startsWith("CONNECT")) {
synchronized(this) {
synchLock.lock();
try {
connected = true;
notifyAll();
connectedStateChanged.signalAll();
}
finally{
synchLock.unlock();
}
// There might be data in the buffer as well as text
int off = i + 1;
@@ -323,14 +369,22 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
return;
} else if(s.equals("BUSY") || s.equals("NO DIALTONE")
|| s.equals("NO CARRIER")) {
synchronized(this) {
synchLock.lock();
try {
connected = false;
notifyAll();
connectedStateChanged.signalAll();
}
finally{
synchLock.unlock();
}
} else if(s.equals("OK")) {
synchronized(this) {
synchLock.lock();
try {
initialised = true;
notifyAll();
initialisedStateChanged.signalAll();
}
finally{
synchLock.unlock();
}
} else if(s.equals("RING")) {
executor.execute(new Runnable() {
@@ -358,7 +412,8 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
try {
ReliabilityLayer reliability =
reliabilityFactory.createReliabilityLayer(this);
synchronized(this) {
synchLock.lock();
try {
if(!initialised) {
LOG.info("Not answering - modem not initialised");
return;
@@ -368,6 +423,9 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
return;
}
this.reliability = reliability;
}
finally{
synchLock.unlock();
}
reliability.start();
LOG.info("Answering");
@@ -380,14 +438,18 @@ class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
// Wait for the event thread to receive "CONNECT"
boolean success = false;
try {
synchronized(this) {
synchLock.lock();
try {
long now = clock.currentTimeMillis();
long end = now + CONNECT_TIMEOUT;
while(now < end && initialised && !connected) {
wait(end - now);
connectedStateChanged.await(end - now, TimeUnit.MILLISECONDS);
now = clock.currentTimeMillis();
}
success = connected;
}
finally{
synchLock.unlock();
}
} catch(InterruptedException e) {
tryToClose(port);