Kaladaran/briar
0
0
Fork 0
mirror of https://code.briarproject.org/briar/briar.git synced 2026-02-11 18:29:05 +01:00
Code Issues Projects Releases Wiki Activity

Added the ability to remove neighbours from the database (untested).

Browse Source
This commit is contained in:
akwizgran
2011-06-29 12:54:00 +01:00
parent 6960f64982
commit ed0174a91b
6 changed files with 689 additions and 425 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
api/net/sf/briar/api/db/DatabaseComponent.java
View File

@@ -50,6 +50,9 @@ public interface DatabaseComponent {
*/
void receiveBundle(NeighbourId n, Bundle b) throws DbException;
/** Removes a neighbour (and all associated state) from the database. */
void removeNeighbour(NeighbourId n) throws DbException;
/** Records the user's rating for the given author. */
void setRating(AuthorId a, Rating r) throws DbException;

90
components/net/sf/briar/db/Database.java
View File

@@ -19,6 +19,16 @@ import net.sf.briar.api.protocol.MessageId;
* obtained by calling startTransaction(). Every transaction must be
* terminated by calling either abortTransaction() or commitTransaction(),
* even if an exception is thrown.
*
* Locking is provided by the DatabaseComponent implementation. To prevent
* deadlock, locks must be acquired in the following order:
* <ul>
* <li> contacts
* <li> messages
* <li> messageStatuses
* <li> ratings
* <li> subscriptions
* </ul>
*/
interface Database<T> {
@@ -49,63 +59,80 @@ interface Database<T> {
/**
* Records a received batch as needing to be acknowledged.
* Locking: neighbours write.
* <p>
* Locking: contacts read, messageStatuses write.
*/
void addBatchToAck(T txn, NeighbourId n, BatchId b) throws DbException;
/**
* Returns false if the given message is already in the database. Otherwise
* stores the message and returns true.
* <p>
* Locking: messages write.
*/
boolean addMessage(T txn, Message m) throws DbException;
/**
* Adds a new neighbour to the database.
* Locking: neighbours write.
* <p>
* Locking: contacts write, messageStatuses write.
*/
void addNeighbour(T txn, NeighbourId n) throws DbException;
/**
* Records a sent batch as needing to be acknowledged.
* Locking: neighbours write, messages read.
* <p>
* Locking: contacts read, messages read, messageStatuses write.
*/
void addOutstandingBatch(T txn, NeighbourId n, BatchId b, Set<MessageId> sent) throws DbException;
/**
* Records a received bundle. This should be called after processing the
* bundle's contents, and may result in outstanding messages becoming
* eligible for retransmittion.
* Locking: neighbours write, messages read.
* eligible for retransmission.
* <p>
* Locking: contacts read, messages read, messageStatuses write.
*/
Set<BatchId> addReceivedBundle(T txn, NeighbourId n, BundleId b) throws DbException;
/**
* Subscribes to the given group.
* <p>
* Locking: subscriptions write.
*/
void addSubscription(T txn, GroupId g) throws DbException;
/**
* Records a neighbour's subscription to a group.
* Locking: neighbours write.
* <p>
* Locking: contacts read, messageStatuses write.
*/
void addSubscription(T txn, NeighbourId n, GroupId g) throws DbException;
/**
* Removes all recorded subscriptions for the given neighbour.
* Locking: neighbours write.
* <p>
* Locking: contacts read, messageStatuses write.
*/
void clearSubscriptions(T txn, NeighbourId n) throws DbException;
/**
* Returns true iff the database contains the given message.
* <p>
* Locking: messages read.
*/
boolean containsMessage(T txn, MessageId m) throws DbException;
/**
* Returns true iff the database contains the given neighbour.
* <p>
* Locking: contacts read.
*/
boolean containsNeighbour(T txn, NeighbourId n) throws DbException;
/**
* Returns true iff the user is subscribed to the given group.
* <p>
* Locking: subscriptions read.
*/
boolean containsSubscription(T txn, GroupId g) throws DbException;
@@ -114,18 +141,21 @@ interface Database<T> {
* Returns the amount of free storage space available to the database, in
* bytes. This is based on the minimum of the space available on the device
* where the database is stored and the database's configured size.
* <p>
* Locking: messages read.
*/
long getFreeSpace() throws DbException;
/**
* Returns the message identified by the given ID.
* <p>
* Locking: messages read.
*/
Message getMessage(T txn, MessageId m) throws DbException;
/**
* Returns the IDs of all messages signed by the given author.
* <p>
* Locking: messages read.
*/
Iterable<MessageId> getMessagesByAuthor(T txn, AuthorId a) throws DbException;
@@ -133,31 +163,36 @@ interface Database<T> {
/**
* Returns the IDs of all children of the message identified by the given
* ID that are present in the database.
* <p>
* Locking: messages read.
*/
Iterable<MessageId> getMessagesByParent(T txn, MessageId m) throws DbException;
/**
* Returns the IDs of all neighbours
* Locking: neighbours read.
* Returns the IDs of all neighbours.
* <p>
* Locking: contacts read, messageStatuses read.
*/
Set<NeighbourId> getNeighbours(T txn) throws DbException;
/**
* Returns the IDs of the oldest messages in the database, with a total
* size less than or equal to the given size.
* <p>
* Locking: messages read.
*/
Iterable<MessageId> getOldMessages(T txn, long size) throws DbException;
/**
* Returns the parent of the given message.
* <p>
* Locking: messages read.
*/
MessageId getParent(T txn, MessageId m) throws DbException;
/**
* Returns the user's rating for the given author.
* <p>
* Locking: ratings read.
*/
Rating getRating(T txn, AuthorId a) throws DbException;
@@ -166,6 +201,7 @@ interface Database<T> {
* Returns the sendability score of the given message. Messages with
* sendability scores greater than zero are eligible to be sent to
* neighbours.
* <p>
* Locking: messages read.
*/
int getSendability(T txn, MessageId m) throws DbException;
@@ -173,12 +209,14 @@ interface Database<T> {
/**
* Returns the IDs of some messages that are eligible to be sent to the
* given neighbour, with a total size less than or equal to the given size.
* Locking: neighbours read, messages read.
* <p>
* Locking: contacts read, messages read, messageStatuses read.
*/
Iterable<MessageId> getSendableMessages(T txn, NeighbourId n, long capacity) throws DbException;
/**
* Returns the groups to which the user subscribes.
* <p>
* Locking: subscriptions read.
*/
Set<GroupId> getSubscriptions(T txn) throws DbException;
@@ -187,14 +225,16 @@ interface Database<T> {
* Removes an outstanding batch that has been acknowledged. Any messages in
* the batch that are still considered outstanding (Status.SENT) with
* respect to the given neighbour are now considered seen (Status.SEEN).
* Locking: neighbours write, messages read.
* <p>
* Locking: contacts read, messages read, messageStatuses write.
*/
void removeAckedBatch(T txn, NeighbourId n, BatchId b) throws DbException;
/**
* Removes and returns the IDs of any batches received from the given
* neighbour that need to be acknowledged.
* Locking: neighbours write.
* <p>
* Locking: contacts read, messageStatuses write.
*/
Set<BatchId> removeBatchesToAck(T txn, NeighbourId n) throws DbException;
@@ -202,38 +242,52 @@ interface Database<T> {
* Removes an outstanding batch that has been lost. Any messages in the
* batch that are still considered outstanding (Status.SENT) with respect
* to the given neighbour are now considered unsent (Status.NEW).
* Locking: neighbours write, messages read.
* <p>
* Locking: contacts read, messages read, messageStatuses write.
*/
void removeLostBatch(T txn, NeighbourId n, BatchId b) throws DbException;
/**
* Removes a message from the database.
* Locking: neighbours write, messages write.
* Removes a message (and all associated state) from the database.
* <p>
* Locking: contacts read, messages write, messageStatuses write.
*/
void removeMessage(T txn, MessageId m) throws DbException;
/**
* Removes a neighbour (and all associated state) from the database.
* <p>
* Locking: contacts write, messageStatuses write.
*/
void removeNeighbour(T txn, NeighbourId n) throws DbException;
/**
* Unsubscribes from the given group. Any messages belonging to the group
* are deleted from the database.
* Locking: subscriptions write, neighbours write, messages write.
* <p>
* Locking: contacts read, subscriptions write, messages write,
* messageStatuses write.
*/
void removeSubscription(T txn, GroupId g) throws DbException;
/**
* Records the user's rating for the given author.
* <p>
* Locking: ratings write.
*/
Rating setRating(T txn, AuthorId a, Rating r) throws DbException;
/**
* Records the sendability score of the given message.
* <p>
* Locking: messages write.
*/
void setSendability(T txn, MessageId m, int sendability) throws DbException;
/**
* Sets the status of the given message with respect to the given neighbour.
* Locking: neighbours write, messages read
* Sets the status of the given message with respect to the given neighbour.
* <p>
* Locking: contacts read, messages read, messageStatuses write.
*/
void setStatus(T txn, NeighbourId n, MessageId m, Status s) throws DbException;
}

17
components/net/sf/briar/db/DatabaseComponentImpl.java
View File

@@ -74,7 +74,20 @@ abstract class DatabaseComponentImpl<Txn> implements DatabaseComponent {
}
}
// Locking: messages write, neighbours write
// Locking: contacts read
protected boolean containsNeighbour(NeighbourId n) throws DbException {
Txn txn = db.startTransaction();
try {
boolean contains = db.containsNeighbour(txn, n);
db.commitTransaction(txn);
return contains;
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
// Locking: contacts read, messages write, messageStatuses write
protected void removeMessage(Txn txn, MessageId id) throws DbException {
Integer sendability = db.getSendability(txn, id);
assert sendability != null;
@@ -127,7 +140,7 @@ abstract class DatabaseComponentImpl<Txn> implements DatabaseComponent {
new Thread(cleaner).start();
}
// Locking: messages write, neighbours write
// Locking: contacts read, messages write, messageStatuses write
protected boolean storeMessage(Txn txn, Message m, NeighbourId sender)
throws DbException {
boolean added = db.addMessage(txn, m);

44
components/net/sf/briar/db/JdbcDatabase.java
View File

@@ -650,6 +650,33 @@ abstract class JdbcDatabase implements Database<Connection> {
}
}
public boolean containsNeighbour(Connection txn, NeighbourId n)
throws DbException {
PreparedStatement ps = null;
ResultSet rs = null;
try {
String sql = "SELECT COUNT(neighbourId) FROM neighbours"
+ " WHERE neighbourId = ?";
ps = txn.prepareStatement(sql);
ps.setInt(1, n.getInt());
rs = ps.executeQuery();
boolean found = rs.next();
assert found;
int count = rs.getInt(1);
assert count <= 1;
boolean more = rs.next();
assert !more;
rs.close();
ps.close();
return count > 0;
} catch(SQLException e) {
tryToClose(rs);
tryToClose(ps);
tryToClose(txn);
throw new DbException(e);
}
}
public boolean containsSubscription(Connection txn, GroupId g)
throws DbException {
PreparedStatement ps = null;
@@ -1070,6 +1097,23 @@ abstract class JdbcDatabase implements Database<Connection> {
}
}
public void removeNeighbour(Connection txn, NeighbourId n)
throws DbException {
PreparedStatement ps = null;
try {
String sql = "DELETE FROM neighbours WHERE neighbourId = ?";
ps = txn.prepareStatement(sql);
ps.setInt(1, n.getInt());
int rowsAffected = ps.executeUpdate();
assert rowsAffected == 1;
ps.close();
} catch(SQLException e) {
tryToClose(ps);
tryToClose(txn);
throw new DbException(e);
}
}
public void removeSubscription(Connection txn, GroupId g)
throws DbException {
PreparedStatement ps = null;

572
components/net/sf/briar/db/ReadWriteLockDatabaseComponent.java
View File

@@ -32,9 +32,11 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
* implementation can allow writers to starve.
*/
private final ReentrantReadWriteLock contactLock =
new ReentrantReadWriteLock(true);
private final ReentrantReadWriteLock messageLock =
new ReentrantReadWriteLock(true);
private final ReentrantReadWriteLock neighbourLock =
private final ReentrantReadWriteLock messageStatusLock =
new ReentrantReadWriteLock(true);
private final ReentrantReadWriteLock ratingLock =
new ReentrantReadWriteLock(true);
@@ -48,85 +50,16 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
}
protected void expireMessages(long size) throws DbException {
messageLock.writeLock().lock();
contactLock.readLock().lock();
try {
neighbourLock.writeLock().lock();
messageLock.writeLock().lock();
try {
Txn txn = db.startTransaction();
try {
for(MessageId m : db.getOldMessages(txn, size)) {
removeMessage(txn, m);
}
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
neighbourLock.writeLock().unlock();
}
} finally {
messageLock.writeLock().unlock();
}
}
public void close() throws DbException {
messageLock.writeLock().lock();
try {
neighbourLock.writeLock().lock();
try {
ratingLock.writeLock().lock();
try {
subscriptionLock.writeLock().lock();
try {
db.close();
} finally {
subscriptionLock.writeLock().unlock();
}
} finally {
ratingLock.writeLock().unlock();
}
} finally {
neighbourLock.writeLock().unlock();
}
} finally {
messageLock.writeLock().unlock();
}
}
public void addNeighbour(NeighbourId n) throws DbException {
if(LOG.isLoggable(Level.FINE)) LOG.fine("Adding neighbour " + n);
neighbourLock.writeLock().lock();
try {
Txn txn = db.startTransaction();
try {
db.addNeighbour(txn, n);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
neighbourLock.writeLock().unlock();
}
}
public void addLocallyGeneratedMessage(Message m) throws DbException {
waitForPermissionToWrite();
messageLock.writeLock().lock();
try {
neighbourLock.writeLock().lock();
try {
subscriptionLock.readLock().lock();
messageStatusLock.writeLock().lock();
try {
Txn txn = db.startTransaction();
try {
if(db.containsSubscription(txn, m.getGroup())) {
boolean added = storeMessage(txn, m, null);
assert added;
} else {
if(LOG.isLoggable(Level.FINE))
LOG.fine("Not subscribed");
for(MessageId m : db.getOldMessages(txn, size)) {
removeMessage(txn, m);
}
db.commitTransaction(txn);
} catch(DbException e) {
@@ -134,13 +67,102 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
throw e;
}
} finally {
subscriptionLock.readLock().unlock();
messageStatusLock.writeLock().unlock();
}
} finally {
neighbourLock.writeLock().unlock();
messageLock.writeLock().unlock();
}
} finally {
messageLock.writeLock().unlock();
contactLock.readLock().unlock();
}
}
public void close() throws DbException {
contactLock.writeLock().lock();
try {
messageLock.writeLock().lock();
try {
messageStatusLock.writeLock().lock();
try {
ratingLock.writeLock().lock();
try {
subscriptionLock.writeLock().lock();
try {
db.close();
} finally {
subscriptionLock.writeLock().unlock();
}
} finally {
ratingLock.writeLock().unlock();
}
} finally {
messageStatusLock.writeLock().unlock();
}
} finally {
messageLock.writeLock().unlock();
}
} finally {
contactLock.writeLock().unlock();
}
}
public void addNeighbour(NeighbourId n) throws DbException {
if(LOG.isLoggable(Level.FINE)) LOG.fine("Adding neighbour " + n);
contactLock.writeLock().lock();
try {
messageStatusLock.writeLock().lock();
try {
Txn txn = db.startTransaction();
try {
db.addNeighbour(txn, n);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
messageStatusLock.writeLock().unlock();
}
} finally {
contactLock.writeLock().unlock();
}
}
public void addLocallyGeneratedMessage(Message m) throws DbException {
waitForPermissionToWrite();
contactLock.readLock().lock();
try {
messageLock.writeLock().lock();
try {
messageStatusLock.writeLock().lock();
try {
subscriptionLock.readLock().lock();
try {
Txn txn = db.startTransaction();
try {
if(db.containsSubscription(txn, m.getGroup())) {
boolean added = storeMessage(txn, m, null);
assert added;
} else {
if(LOG.isLoggable(Level.FINE))
LOG.fine("Not subscribed");
}
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
subscriptionLock.readLock().unlock();
}
} finally {
messageStatusLock.writeLock().unlock();
}
} finally {
messageLock.writeLock().unlock();
}
} finally {
contactLock.readLock().unlock();
}
}
@@ -161,6 +183,28 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
}
}
public void removeNeighbour(NeighbourId n) throws DbException {
if(LOG.isLoggable(Level.FINE)) LOG.fine("Removing neighbour " + n);
contactLock.writeLock().lock();
try {
messageStatusLock.writeLock().lock();
try {
Txn txn = db.startTransaction();
try {
db.removeNeighbour(txn, n);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
messageStatusLock.writeLock().unlock();
}
} finally {
contactLock.writeLock().unlock();
}
}
public void setRating(AuthorId a, Rating r) throws DbException {
messageLock.writeLock().lock();
try {
@@ -224,72 +268,89 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
public void unsubscribe(GroupId g) throws DbException {
if(LOG.isLoggable(Level.FINE)) LOG.fine("Unsubscribing from " + g);
messageLock.writeLock().lock();
contactLock.readLock().lock();
try {
neighbourLock.writeLock().lock();
messageLock.writeLock().lock();
try {
subscriptionLock.writeLock().lock();
messageStatusLock.writeLock().lock();
try {
Txn txn = db.startTransaction();
subscriptionLock.writeLock().lock();
try {
db.removeSubscription(txn, g);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
Txn txn = db.startTransaction();
try {
db.removeSubscription(txn, g);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
subscriptionLock.writeLock().unlock();
}
} finally {
subscriptionLock.writeLock().unlock();
messageStatusLock.writeLock().unlock();
}
} finally {
neighbourLock.writeLock().unlock();
messageLock.writeLock().unlock();
}
} finally {
messageLock.writeLock().unlock();
contactLock.readLock().unlock();
}
}
public void generateBundle(NeighbourId n, Bundle b) throws DbException {
if(LOG.isLoggable(Level.FINE)) LOG.fine("Generating bundle for " + n);
// Ack all batches received from the neighbour
neighbourLock.writeLock().lock();
contactLock.readLock().lock();
try {
Txn txn = db.startTransaction();
if(!containsNeighbour(n)) return;
messageStatusLock.writeLock().lock();
try {
int numAcks = 0;
for(BatchId ack : db.removeBatchesToAck(txn, n)) {
b.addAck(ack);
numAcks++;
Txn txn = db.startTransaction();
try {
int numAcks = 0;
for(BatchId ack : db.removeBatchesToAck(txn, n)) {
b.addAck(ack);
numAcks++;
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Added " + numAcks + " acks");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Added " + numAcks + " acks");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
} finally {
messageStatusLock.writeLock().unlock();
}
} finally {
neighbourLock.writeLock().unlock();
contactLock.readLock().unlock();
}
// Add a list of subscriptions
subscriptionLock.readLock().lock();
contactLock.readLock().lock();
try {
Txn txn = db.startTransaction();
if(!containsNeighbour(n)) return;
subscriptionLock.readLock().lock();
try {
int numSubs = 0;
for(GroupId g : db.getSubscriptions(txn)) {
b.addSubscription(g);
numSubs++;
Txn txn = db.startTransaction();
try {
int numSubs = 0;
for(GroupId g : db.getSubscriptions(txn)) {
b.addSubscription(g);
numSubs++;
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Added " + numSubs + " subscriptions");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Added " + numSubs + " subscriptions");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
} finally {
subscriptionLock.readLock().unlock();
}
} finally {
subscriptionLock.readLock().unlock();
contactLock.readLock().unlock();
}
// Add as many messages as possible to the bundle
long capacity = b.getCapacity();
@@ -309,55 +370,61 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
}
private Batch fillBatch(NeighbourId n, long capacity) throws DbException {
messageLock.readLock().lock();
contactLock.readLock().lock();
try {
Set<MessageId> sent;
Batch b;
neighbourLock.readLock().lock();
if(!containsNeighbour(n)) return null;
messageLock.readLock().lock();
try {
Txn txn = db.startTransaction();
Set<MessageId> sent;
Batch b;
messageStatusLock.readLock().lock();
try {
capacity = Math.min(capacity, Batch.CAPACITY);
Iterator<MessageId> it =
db.getSendableMessages(txn, n, capacity).iterator();
if(!it.hasNext()) {
Txn txn = db.startTransaction();
try {
capacity = Math.min(capacity, Batch.CAPACITY);
Iterator<MessageId> it =
db.getSendableMessages(txn, n, capacity).iterator();
if(!it.hasNext()) {
db.commitTransaction(txn);
return null; // No more messages to send
}
sent = new HashSet<MessageId>();
b = batchProvider.get();
while(it.hasNext()) {
MessageId m = it.next();
b.addMessage(db.getMessage(txn, m));
sent.add(m);
}
b.seal();
db.commitTransaction(txn);
return null; // No more messages to send
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
sent = new HashSet<MessageId>();
b = batchProvider.get();
while(it.hasNext()) {
MessageId m = it.next();
b.addMessage(db.getMessage(txn, m));
sent.add(m);
}
b.seal();
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
} finally {
messageStatusLock.readLock().unlock();
}
} finally {
neighbourLock.readLock().unlock();
}
// Record the contents of the batch
neighbourLock.writeLock().lock();
try {
Txn txn = db.startTransaction();
// Record the contents of the batch
messageStatusLock.writeLock().lock();
try {
assert !sent.isEmpty();
db.addOutstandingBatch(txn, n, b.getId(), sent);
db.commitTransaction(txn);
return b;
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
Txn txn = db.startTransaction();
try {
assert !sent.isEmpty();
db.addOutstandingBatch(txn, n, b.getId(), sent);
db.commitTransaction(txn);
return b;
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
messageStatusLock.writeLock().unlock();
}
} finally {
neighbourLock.writeLock().unlock();
messageLock.readLock().unlock();
}
} finally {
messageLock.readLock().unlock();
contactLock.readLock().unlock();
}
}
@@ -366,133 +433,164 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
LOG.fine("Received bundle from " + n + ", "
+ b.getSize() + " bytes");
// Mark all messages in acked batches as seen
messageLock.readLock().lock();
contactLock.readLock().lock();
try {
neighbourLock.writeLock().lock();
if(!containsNeighbour(n)) return;
messageLock.readLock().lock();
try {
int acks = 0;
for(BatchId ack : b.getAcks()) {
acks++;
Txn txn = db.startTransaction();
try {
db.removeAckedBatch(txn, n, ack);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + acks + " acks");
} finally {
neighbourLock.writeLock().unlock();
}
} finally {
messageLock.readLock().unlock();
}
// Update the neighbour's subscriptions
neighbourLock.writeLock().lock();
try {
Txn txn = db.startTransaction();
try {
db.clearSubscriptions(txn, n);
int subs = 0;
for(GroupId g : b.getSubscriptions()) {
subs++;
db.addSubscription(txn, n, g);
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + subs + " subscriptions");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
neighbourLock.writeLock().unlock();
}
// Store the messages
int batches = 0;
for(Batch batch : b.getBatches()) {
batches++;
waitForPermissionToWrite();
messageLock.writeLock().lock();
try {
neighbourLock.writeLock().lock();
messageStatusLock.writeLock().lock();
try {
subscriptionLock.readLock().lock();
try {
int acks = 0;
for(BatchId ack : b.getAcks()) {
acks++;
Txn txn = db.startTransaction();
try {
int received = 0, stored = 0;
for(Message m : batch.getMessages()) {
received++;
if(db.containsSubscription(txn, m.getGroup())) {
if(storeMessage(txn, m, n)) stored++;
}
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + received
+ " messages, stored " + stored);
db.addBatchToAck(txn, n, batch.getId());
db.removeAckedBatch(txn, n, ack);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
subscriptionLock.readLock().unlock();
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + acks + " acks");
} finally {
neighbourLock.writeLock().unlock();
messageStatusLock.writeLock().unlock();
}
} finally {
messageLock.writeLock().unlock();
messageLock.readLock().unlock();
}
} finally {
contactLock.readLock().unlock();
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + batches + " batches");
// Find any lost batches that need to be retransmitted
Set<BatchId> lost;
messageLock.readLock().lock();
// Update the neighbour's subscriptions
contactLock.readLock().lock();
try {
neighbourLock.writeLock().lock();
if(!containsNeighbour(n)) return;
messageStatusLock.writeLock().lock();
try {
Txn txn = db.startTransaction();
try {
lost = db.addReceivedBundle(txn, n, b.getId());
db.clearSubscriptions(txn, n);
int subs = 0;
for(GroupId g : b.getSubscriptions()) {
subs++;
db.addSubscription(txn, n, g);
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + subs + " subscriptions");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
neighbourLock.writeLock().unlock();
messageStatusLock.writeLock().unlock();
}
} finally {
messageLock.readLock().unlock();
contactLock.readLock().lock();
}
for(BatchId batch : lost) {
// Store the messages
int batches = 0;
for(Batch batch : b.getBatches()) {
batches++;
waitForPermissionToWrite();
contactLock.readLock().lock();
try {
if(!containsNeighbour(n)) return;
messageLock.writeLock().lock();
try {
messageStatusLock.writeLock().lock();
try {
subscriptionLock.readLock().lock();
try {
Txn txn = db.startTransaction();
try {
int received = 0, stored = 0;
for(Message m : batch.getMessages()) {
received++;
GroupId g = m.getGroup();
if(db.containsSubscription(txn, g)) {
if(storeMessage(txn, m, n)) stored++;
}
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + received
+ " messages, stored " + stored);
db.addBatchToAck(txn, n, batch.getId());
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
subscriptionLock.readLock().unlock();
}
} finally {
messageStatusLock.writeLock().unlock();
}
} finally {
messageLock.writeLock().unlock();
}
} finally {
contactLock.readLock().unlock();
}
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + batches + " batches");
// Find any lost batches that need to be retransmitted
Set<BatchId> lost;
contactLock.readLock().lock();
try {
if(!containsNeighbour(n)) return;
messageLock.readLock().lock();
try {
neighbourLock.writeLock().lock();
messageStatusLock.writeLock().lock();
try {
Txn txn = db.startTransaction();
try {
if(LOG.isLoggable(Level.FINE))
LOG.fine("Removing lost batch");
db.removeLostBatch(txn, n, batch);
lost = db.addReceivedBundle(txn, n, b.getId());
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
neighbourLock.writeLock().unlock();
messageStatusLock.writeLock().unlock();
}
} finally {
messageLock.readLock().unlock();
}
} finally {
contactLock.readLock().unlock();
}
for(BatchId batch : lost) {
contactLock.readLock().lock();
try {
if(!containsNeighbour(n)) return;
messageLock.readLock().lock();
try {
messageStatusLock.writeLock().lock();
try {
Txn txn = db.startTransaction();
try {
if(LOG.isLoggable(Level.FINE))
LOG.fine("Removing lost batch");
db.removeLostBatch(txn, n, batch);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
messageStatusLock.writeLock().unlock();
}
} finally {
messageLock.readLock().unlock();
}
} finally {
contactLock.readLock().unlock();
}
}
System.gc();
}

388
components/net/sf/briar/db/SynchronizedDatabaseComponent.java
View File

@@ -30,8 +30,9 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
* interface to find out which calls require which locks.
*/
private final Object contactLock = new Object();
private final Object messageLock = new Object();
private final Object neighbourLock = new Object();
private final Object messageStatusLock = new Object();
private final Object ratingLock = new Object();
private final Object subscriptionLock = new Object();
@@ -42,13 +43,45 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
}
protected void expireMessages(long size) throws DbException {
synchronized(messageLock) {
synchronized(neighbourLock) {
synchronized(contactLock) {
synchronized(messageLock) {
synchronized(messageStatusLock) {
Txn txn = db.startTransaction();
try {
for(MessageId m : db.getOldMessages(txn, size)) {
removeMessage(txn, m);
}
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
}
}
}
public void close() throws DbException {
synchronized(contactLock) {
synchronized(messageLock) {
synchronized(messageStatusLock) {
synchronized(ratingLock) {
synchronized(subscriptionLock) {
db.close();
}
}
}
}
}
}
public void addNeighbour(NeighbourId n) throws DbException {
if(LOG.isLoggable(Level.FINE)) LOG.fine("Adding neighbour " + n);
synchronized(contactLock) {
synchronized(messageStatusLock) {
Txn txn = db.startTransaction();
try {
for(MessageId m : db.getOldMessages(txn, size)) {
removeMessage(txn, m);
}
db.addNeighbour(txn, n);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
@@ -58,50 +91,26 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
}
}
public void close() throws DbException {
synchronized(messageLock) {
synchronized(neighbourLock) {
synchronized(ratingLock) {
synchronized(subscriptionLock) {
db.close();
}
}
}
}
}
public void addNeighbour(NeighbourId n) throws DbException {
if(LOG.isLoggable(Level.FINE)) LOG.fine("Adding neighbour " + n);
synchronized(neighbourLock) {
Txn txn = db.startTransaction();
try {
db.addNeighbour(txn, n);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
}
public void addLocallyGeneratedMessage(Message m) throws DbException {
waitForPermissionToWrite();
synchronized(messageLock) {
synchronized(neighbourLock) {
synchronized(subscriptionLock) {
Txn txn = db.startTransaction();
try {
if(db.containsSubscription(txn, m.getGroup())) {
boolean added = storeMessage(txn, m, null);
assert added;
} else {
if(LOG.isLoggable(Level.FINE))
LOG.fine("Not subscribed");
synchronized(contactLock) {
synchronized(messageLock) {
synchronized(messageStatusLock) {
synchronized(subscriptionLock) {
Txn txn = db.startTransaction();
try {
if(db.containsSubscription(txn, m.getGroup())) {
boolean added = storeMessage(txn, m, null);
assert added;
} else {
if(LOG.isLoggable(Level.FINE))
LOG.fine("Not subscribed");
}
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
}
@@ -173,16 +182,18 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
public void unsubscribe(GroupId g) throws DbException {
if(LOG.isLoggable(Level.FINE)) LOG.fine("Unsubscribing from " + g);
synchronized(messageLock) {
synchronized(neighbourLock) {
synchronized(subscriptionLock) {
Txn txn = db.startTransaction();
try {
db.removeSubscription(txn, g);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
synchronized(contactLock) {
synchronized(messageLock) {
synchronized(messageStatusLock) {
synchronized(subscriptionLock) {
Txn txn = db.startTransaction();
try {
db.removeSubscription(txn, g);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
}
}
@@ -192,37 +203,43 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
public void generateBundle(NeighbourId n, Bundle b) throws DbException {
if(LOG.isLoggable(Level.FINE)) LOG.fine("Generating bundle for " + n);
// Ack all batches received from the neighbour
synchronized(neighbourLock) {
Txn txn = db.startTransaction();
try {
int numAcks = 0;
for(BatchId ack : db.removeBatchesToAck(txn, n)) {
b.addAck(ack);
numAcks++;
synchronized(contactLock) {
if(!containsNeighbour(n)) return;
synchronized(messageStatusLock) {
Txn txn = db.startTransaction();
try {
int numAcks = 0;
for(BatchId ack : db.removeBatchesToAck(txn, n)) {
b.addAck(ack);
numAcks++;
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Added " + numAcks + " acks");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Added " + numAcks + " acks");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
// Add a list of subscriptions
synchronized(subscriptionLock) {
Txn txn = db.startTransaction();
try {
int numSubs = 0;
for(GroupId g : db.getSubscriptions(txn)) {
b.addSubscription(g);
numSubs++;
synchronized(contactLock) {
if(!containsNeighbour(n)) return;
synchronized(subscriptionLock) {
Txn txn = db.startTransaction();
try {
int numSubs = 0;
for(GroupId g : db.getSubscriptions(txn)) {
b.addSubscription(g);
numSubs++;
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Added " + numSubs + " subscriptions");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Added " + numSubs + " subscriptions");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
// Add as many messages as possible to the bundle
@@ -243,30 +260,49 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
}
private Batch fillBatch(NeighbourId n, long capacity) throws DbException {
synchronized(messageLock) {
synchronized(neighbourLock) {
synchronized(contactLock) {
if(!containsNeighbour(n)) return null;
synchronized(messageLock) {
synchronized(messageStatusLock) {
Txn txn = db.startTransaction();
try {
capacity = Math.min(capacity, Batch.CAPACITY);
Iterator<MessageId> it =
db.getSendableMessages(txn, n, capacity).iterator();
if(!it.hasNext()) {
db.commitTransaction(txn);
return null; // No more messages to send
}
Batch b = batchProvider.get();
Set<MessageId> sent = new HashSet<MessageId>();
while(it.hasNext()) {
MessageId m = it.next();
b.addMessage(db.getMessage(txn, m));
sent.add(m);
}
b.seal();
// Record the contents of the batch
assert !sent.isEmpty();
db.addOutstandingBatch(txn, n, b.getId(), sent);
db.commitTransaction(txn);
return b;
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
}
}
}
public void removeNeighbour(NeighbourId n) throws DbException {
if(LOG.isLoggable(Level.FINE)) LOG.fine("Removing neighbour " + n);
synchronized(contactLock) {
synchronized(messageStatusLock) {
Txn txn = db.startTransaction();
try {
capacity = Math.min(capacity, Batch.CAPACITY);
Iterator<MessageId> it =
db.getSendableMessages(txn, n, capacity).iterator();
if(!it.hasNext()) {
db.commitTransaction(txn);
return null; // No more messages to send
}
Batch b = batchProvider.get();
Set<MessageId> sent = new HashSet<MessageId>();
while(it.hasNext()) {
MessageId m = it.next();
b.addMessage(db.getMessage(txn, m));
sent.add(m);
}
b.seal();
// Record the contents of the batch
assert !sent.isEmpty();
db.addOutstandingBatch(txn, n, b.getId(), sent);
db.removeNeighbour(txn, n);
db.commitTransaction(txn);
return b;
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
@@ -278,42 +314,48 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
public void receiveBundle(NeighbourId n, Bundle b) throws DbException {
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received bundle from " + n + ", "
+ b.getSize() + " bytes");
+ b.getSize() + " bytes");
// Mark all messages in acked batches as seen
synchronized(messageLock) {
synchronized(neighbourLock) {
int acks = 0;
for(BatchId ack : b.getAcks()) {
acks++;
Txn txn = db.startTransaction();
try {
db.removeAckedBatch(txn, n, ack);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
synchronized(contactLock) {
if(!containsNeighbour(n)) return;
synchronized(messageLock) {
synchronized(messageStatusLock) {
int acks = 0;
for(BatchId ack : b.getAcks()) {
acks++;
Txn txn = db.startTransaction();
try {
db.removeAckedBatch(txn, n, ack);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + acks + " acks");
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + acks + " acks");
}
}
// Update the neighbour's subscriptions
synchronized(neighbourLock) {
Txn txn = db.startTransaction();
try {
db.clearSubscriptions(txn, n);
int subs = 0;
for(GroupId g : b.getSubscriptions()) {
subs++;
db.addSubscription(txn, n, g);
synchronized(contactLock) {
if(!containsNeighbour(n)) return;
synchronized(messageStatusLock) {
Txn txn = db.startTransaction();
try {
db.clearSubscriptions(txn, n);
int subs = 0;
for(GroupId g : b.getSubscriptions()) {
subs++;
db.addSubscription(txn, n, g);
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + subs + " subscriptions");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + subs + " subscriptions");
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
// Store the messages
@@ -321,26 +363,30 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
for(Batch batch : b.getBatches()) {
batches++;
waitForPermissionToWrite();
synchronized(messageLock) {
synchronized(neighbourLock) {
synchronized(subscriptionLock) {
Txn txn = db.startTransaction();
try {
int received = 0, stored = 0;
for(Message m : batch.getMessages()) {
received++;
if(db.containsSubscription(txn, m.getGroup())) {
if(storeMessage(txn, m, n)) stored++;
synchronized(contactLock) {
if(!containsNeighbour(n)) return;
synchronized(messageLock) {
synchronized(messageStatusLock) {
synchronized(subscriptionLock) {
Txn txn = db.startTransaction();
try {
int received = 0, stored = 0;
for(Message m : batch.getMessages()) {
received++;
GroupId g = m.getGroup();
if(db.containsSubscription(txn, g)) {
if(storeMessage(txn, m, n)) stored++;
}
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + received
+ " messages, stored " + stored);
db.addBatchToAck(txn, n, batch.getId());
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
if(LOG.isLoggable(Level.FINE))
LOG.fine("Received " + received
+ " messages, stored " + stored);
db.addBatchToAck(txn, n, batch.getId());
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
}
@@ -350,26 +396,13 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
LOG.fine("Received " + batches + " batches");
// Find any lost batches that need to be retransmitted
Set<BatchId> lost;
synchronized(messageLock) {
synchronized(neighbourLock) {
Txn txn = db.startTransaction();
try {
lost = db.addReceivedBundle(txn, n, b.getId());
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
}
for(BatchId batch : lost) {
synchronized(contactLock) {
if(!containsNeighbour(n)) return;
synchronized(messageLock) {
synchronized(neighbourLock) {
synchronized(messageStatusLock) {
Txn txn = db.startTransaction();
try {
if(LOG.isLoggable(Level.FINE))
LOG.fine("Removing lost batch");
db.removeLostBatch(txn, n, batch);
lost = db.addReceivedBundle(txn, n, b.getId());
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
@@ -378,6 +411,25 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
}
}
}
for(BatchId batch : lost) {
synchronized(contactLock) {
if(!containsNeighbour(n)) return;
synchronized(messageLock) {
synchronized(messageStatusLock) {
Txn txn = db.startTransaction();
try {
if(LOG.isLoggable(Level.FINE))
LOG.fine("Removing lost batch");
db.removeLostBatch(txn, n, batch);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
}
}
}
System.gc();
}
}