This commit is contained in:
akwizgran
2017-11-16 10:29:18 +00:00
committed by Torsten Grote
parent 27328afe3c
commit 5fa6b0ca1c
109 changed files with 2087 additions and 3450 deletions

View File

@@ -50,18 +50,15 @@ public class PoliteExecutor implements Executor {
@Override
public void execute(final Runnable r) {
final long submitted = System.currentTimeMillis();
Runnable wrapped = new Runnable() {
@Override
public void run() {
if (log.isLoggable(LOG_LEVEL)) {
long queued = System.currentTimeMillis() - submitted;
log.log(LOG_LEVEL, "Queue time " + queued + " ms");
}
try {
r.run();
} finally {
scheduleNext();
}
Runnable wrapped = () -> {
if (log.isLoggable(LOG_LEVEL)) {
long queued = System.currentTimeMillis() - submitted;
log.log(LOG_LEVEL, "Queue time " + queued + " ms");
}
try {
r.run();
} finally {
scheduleNext();
}
};
synchronized (lock) {

View File

@@ -31,16 +31,13 @@ public class TimeLoggingExecutor extends ThreadPoolExecutor {
public void execute(final Runnable r) {
if (log.isLoggable(LOG_LEVEL)) {
final long submitted = System.currentTimeMillis();
super.execute(new Runnable() {
@Override
public void run() {
long started = System.currentTimeMillis();
long queued = started - submitted;
log.log(LOG_LEVEL, "Queue time " + queued + " ms");
r.run();
long executing = System.currentTimeMillis() - started;
log.log(LOG_LEVEL, "Execution time " + executing + " ms");
}
super.execute(() -> {
long started = System.currentTimeMillis();
long queued = started - submitted;
log.log(LOG_LEVEL, "Queue time " + queued + " ms");
r.run();
long executing = System.currentTimeMillis() - started;
log.log(LOG_LEVEL, "Execution time " + executing + " ms");
});
} else {
super.execute(r);

View File

@@ -103,15 +103,11 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
@Override
public boolean open() throws DbException {
Runnable shutdownHook = new Runnable() {
@Override
public void run() {
try {
close();
} catch (DbException e) {
if (LOG.isLoggable(WARNING))
LOG.log(WARNING, e.toString(), e);
}
Runnable shutdownHook = () -> {
try {
close();
} catch (DbException e) {
if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
}
};
boolean reopened = db.open();

View File

@@ -266,12 +266,7 @@ class ConnectionManagerImpl implements ConnectionManager {
contactId = ctx.getContactId();
connectionRegistry.registerConnection(contactId, transportId, true);
// Start the outgoing session on another thread
ioExecutor.execute(new Runnable() {
@Override
public void run() {
runOutgoingSession();
}
});
ioExecutor.execute(this::runOutgoingSession);
try {
// Create and run the incoming session
incomingSession = createIncomingSession(ctx, reader);
@@ -368,12 +363,7 @@ class ConnectionManagerImpl implements ConnectionManager {
return;
}
// Start the incoming session on another thread
ioExecutor.execute(new Runnable() {
@Override
public void run() {
runIncomingSession();
}
});
ioExecutor.execute(this::runIncomingSession);
try {
// Create and run the outgoing session
outgoingSession = createDuplexOutgoingSession(ctx, writer);

View File

@@ -112,29 +112,23 @@ class Poller implements EventListener {
}
private void connectToContact(final ContactId c, final SimplexPlugin p) {
ioExecutor.execute(new Runnable() {
@Override
public void run() {
TransportId t = p.getId();
if (!connectionRegistry.isConnected(c, t)) {
TransportConnectionWriter w = p.createWriter(c);
if (w != null)
connectionManager.manageOutgoingConnection(c, t, w);
}
ioExecutor.execute(() -> {
TransportId t = p.getId();
if (!connectionRegistry.isConnected(c, t)) {
TransportConnectionWriter w = p.createWriter(c);
if (w != null)
connectionManager.manageOutgoingConnection(c, t, w);
}
});
}
private void connectToContact(final ContactId c, final DuplexPlugin p) {
ioExecutor.execute(new Runnable() {
@Override
public void run() {
TransportId t = p.getId();
if (!connectionRegistry.isConnected(c, t)) {
DuplexTransportConnection d = p.createConnection(c);
if (d != null)
connectionManager.manageOutgoingConnection(c, t, d);
}
ioExecutor.execute(() -> {
TransportId t = p.getId();
if (!connectionRegistry.isConnected(c, t)) {
DuplexTransportConnection d = p.createConnection(c);
if (d != null)
connectionManager.manageOutgoingConnection(c, t, d);
}
});
}
@@ -161,12 +155,8 @@ class Poller implements EventListener {
if (scheduled == null || due < scheduled.due) {
final PollTask task = new PollTask(p, due, randomiseNext);
tasks.put(t, task);
scheduler.schedule(new Runnable() {
@Override
public void run() {
ioExecutor.execute(task);
}
}, delay, MILLISECONDS);
scheduler.schedule(
() -> ioExecutor.execute(task), delay, MILLISECONDS);
}
} finally {
lock.unlock();

View File

@@ -260,16 +260,12 @@ class LanTcpPlugin extends TcpPlugin {
@Override
public Callable<KeyAgreementConnection> listen() {
return new Callable<KeyAgreementConnection>() {
@Override
public KeyAgreementConnection call() throws IOException {
Socket s = ss.accept();
if (LOG.isLoggable(INFO))
LOG.info(ID.getString() + ": Incoming connection");
return new KeyAgreementConnection(
new TcpTransportConnection(LanTcpPlugin.this, s),
ID);
}
return () -> {
Socket s = ss.accept();
if (LOG.isLoggable(INFO))
LOG.info(ID.getString() + ": Incoming connection");
return new KeyAgreementConnection(
new TcpTransportConnection(LanTcpPlugin.this, s), ID);
};
}

View File

@@ -50,12 +50,7 @@ class PortMapperImpl implements PortMapper {
succeeded = gateway.addPortMapping(port, port,
getHostAddress(internal), "TCP", "TCP");
if (succeeded) {
shutdownManager.addShutdownHook(new Runnable() {
@Override
public void run() {
deleteMapping(port);
}
});
shutdownManager.addShutdownHook(() -> deleteMapping(port));
}
String externalString = gateway.getExternalIPAddress();
if (LOG.isLoggable(INFO))

View File

@@ -110,41 +110,37 @@ abstract class TcpPlugin implements DuplexPlugin {
}
protected void bind() {
ioExecutor.execute(new Runnable() {
@Override
public void run() {
if (!running) return;
ServerSocket ss = null;
for (InetSocketAddress addr : getLocalSocketAddresses()) {
try {
ss = new ServerSocket();
ss.bind(addr);
break;
} catch (IOException e) {
if (LOG.isLoggable(INFO))
LOG.info("Failed to bind " +
scrubSocketAddress(addr));
tryToClose(ss);
}
}
if (ss == null || !ss.isBound()) {
LOG.info("Could not bind server socket");
return;
}
if (!running) {
ioExecutor.execute(() -> {
if (!running) return;
ServerSocket ss = null;
for (InetSocketAddress addr : getLocalSocketAddresses()) {
try {
ss = new ServerSocket();
ss.bind(addr);
break;
} catch (IOException e) {
if (LOG.isLoggable(INFO))
LOG.info("Failed to bind " + scrubSocketAddress(addr));
tryToClose(ss);
return;
}
socket = ss;
backoff.reset();
InetSocketAddress local =
(InetSocketAddress) ss.getLocalSocketAddress();
setLocalSocketAddress(local);
if (LOG.isLoggable(INFO))
LOG.info("Listening on " + scrubSocketAddress(local));
callback.transportEnabled();
acceptContactConnections();
}
if (ss == null || !ss.isBound()) {
LOG.info("Could not bind server socket");
return;
}
if (!running) {
tryToClose(ss);
return;
}
socket = ss;
backoff.reset();
InetSocketAddress local =
(InetSocketAddress) ss.getLocalSocketAddress();
setLocalSocketAddress(local);
if (LOG.isLoggable(INFO))
LOG.info("Listening on " + scrubSocketAddress(local));
callback.transportEnabled();
acceptContactConnections();
});
}
@@ -220,15 +216,12 @@ abstract class TcpPlugin implements DuplexPlugin {
private void connectAndCallBack(final ContactId c,
final TransportProperties p) {
ioExecutor.execute(new Runnable() {
@Override
public void run() {
if (!isRunning()) return;
DuplexTransportConnection d = createConnection(p);
if (d != null) {
backoff.reset();
callback.outgoingConnectionCreated(c, d);
}
ioExecutor.execute(() -> {
if (!isRunning()) return;
DuplexTransportConnection d = createConnection(p);
if (d != null) {
backoff.reset();
callback.outgoingConnectionCreated(c, d);
}
});
}

View File

@@ -54,36 +54,32 @@ class ReliabilityLayerImpl implements ReliabilityLayer, WriteHandler {
inputStream = new ReceiverInputStream(receiver);
outputStream = new SenderOutputStream(sender);
running = true;
executor.execute(new Runnable() {
@Override
public void run() {
long now = clock.currentTimeMillis();
long next = now + TICK_INTERVAL;
try {
while (running) {
byte[] b = null;
while (now < next && b == null) {
b = writes.poll(next - now, MILLISECONDS);
if (!running) return;
now = clock.currentTimeMillis();
}
if (b == null) {
sender.tick();
while (next <= now) next += TICK_INTERVAL;
} else {
if (b.length == 0) return; // Poison pill
writeHandler.handleWrite(b);
}
executor.execute(() -> {
long now = clock.currentTimeMillis();
long next = now + TICK_INTERVAL;
try {
while (running) {
byte[] b = null;
while (now < next && b == null) {
b = writes.poll(next - now, MILLISECONDS);
if (!running) return;
now = clock.currentTimeMillis();
}
if (b == null) {
sender.tick();
while (next <= now) next += TICK_INTERVAL;
} else {
if (b.length == 0) return; // Poison pill
writeHandler.handleWrite(b);
}
} catch (InterruptedException e) {
LOG.warning("Interrupted while waiting to write");
Thread.currentThread().interrupt();
running = false;
} catch (IOException e) {
if (LOG.isLoggable(WARNING))
LOG.log(WARNING, e.toString(), e);
running = false;
}
} catch (InterruptedException e) {
LOG.warning("Interrupted while waiting to write");
Thread.currentThread().interrupt();
running = false;
} catch (IOException e) {
if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
running = false;
}
});
}

View File

@@ -54,12 +54,7 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
private static final Logger LOG =
Logger.getLogger(DuplexOutgoingSession.class.getName());
private static final ThrowingRunnable<IOException> CLOSE =
new ThrowingRunnable<IOException>() {
@Override
public void run() {
}
};
private static final ThrowingRunnable<IOException> CLOSE = () -> {};
private final DatabaseComponent db;
private final Executor dbExecutor;

View File

@@ -43,12 +43,7 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
private static final Logger LOG =
Logger.getLogger(SimplexOutgoingSession.class.getName());
private static final ThrowingRunnable<IOException> CLOSE =
new ThrowingRunnable<IOException>() {
@Override
public void run() {
}
};
private static final ThrowingRunnable<IOException> CLOSE = () -> {};
private final DatabaseComponent db;
private final Executor dbExecutor;

View File

@@ -94,12 +94,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
}
private void validateOutstandingMessagesAsync(final ClientId c) {
dbExecutor.execute(new Runnable() {
@Override
public void run() {
validateOutstandingMessages(c);
}
});
dbExecutor.execute(() -> validateOutstandingMessages(c));
}
@DatabaseExecutor
@@ -121,12 +116,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
private void validateNextMessageAsync(final Queue<MessageId> unvalidated) {
if (unvalidated.isEmpty()) return;
dbExecutor.execute(new Runnable() {
@Override
public void run() {
validateNextMessage(unvalidated);
}
});
dbExecutor.execute(() -> validateNextMessage(unvalidated));
}
@DatabaseExecutor
@@ -159,12 +149,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
}
private void deliverOutstandingMessagesAsync(final ClientId c) {
dbExecutor.execute(new Runnable() {
@Override
public void run() {
deliverOutstandingMessages(c);
}
});
dbExecutor.execute(() -> deliverOutstandingMessages(c));
}
@DatabaseExecutor
@@ -187,12 +172,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
private void deliverNextPendingMessageAsync(
final Queue<MessageId> pending) {
if (pending.isEmpty()) return;
dbExecutor.execute(new Runnable() {
@Override
public void run() {
deliverNextPendingMessage(pending);
}
});
dbExecutor.execute(() -> deliverNextPendingMessage(pending));
}
@DatabaseExecutor
@@ -255,12 +235,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
}
private void validateMessageAsync(final Message m, final Group g) {
validationExecutor.execute(new Runnable() {
@Override
public void run() {
validateMessage(m, g);
}
});
validationExecutor.execute(() -> validateMessage(m, g));
}
@ValidationExecutor
@@ -285,12 +260,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
private void storeMessageContextAsync(final Message m, final ClientId c,
final MessageContext result) {
dbExecutor.execute(new Runnable() {
@Override
public void run() {
storeMessageContext(m, c, result);
}
});
dbExecutor.execute(() -> storeMessageContext(m, c, result));
}
@DatabaseExecutor
@@ -385,12 +355,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
}
private void shareOutstandingMessagesAsync(final ClientId c) {
dbExecutor.execute(new Runnable() {
@Override
public void run() {
shareOutstandingMessages(c);
}
});
dbExecutor.execute(() -> shareOutstandingMessages(c));
}
@DatabaseExecutor
@@ -418,12 +383,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
*/
private void shareNextMessageAsync(final Queue<MessageId> toShare) {
if (toShare.isEmpty()) return;
dbExecutor.execute(new Runnable() {
@Override
public void run() {
shareNextMessage(toShare);
}
});
dbExecutor.execute(() -> shareNextMessage(toShare));
}
@DatabaseExecutor
@@ -452,12 +412,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
private void invalidateNextMessageAsync(final Queue<MessageId> invalidate) {
if (invalidate.isEmpty()) return;
dbExecutor.execute(new Runnable() {
@Override
public void run() {
invalidateNextMessage(invalidate);
}
});
dbExecutor.execute(() -> invalidateNextMessage(invalidate));
}
@DatabaseExecutor
@@ -513,12 +468,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
}
private void loadGroupAndValidateAsync(final Message m) {
dbExecutor.execute(new Runnable() {
@Override
public void run() {
loadGroupAndValidate(m);
}
});
dbExecutor.execute(() -> loadGroupAndValidate(m));
}
@DatabaseExecutor

View File

@@ -157,12 +157,8 @@ class KeyManagerImpl implements KeyManager, Service, EventListener {
private void removeContact(final ContactId c) {
activeContacts.remove(c);
dbExecutor.execute(new Runnable() {
@Override
public void run() {
for (TransportKeyManager m : managers.values())
m.removeContact(c);
}
dbExecutor.execute(() -> {
for (TransportKeyManager m : managers.values()) m.removeContact(c);
});
}
}

View File

@@ -134,32 +134,22 @@ class TransportKeyManagerImpl implements TransportKeyManager {
}
private void scheduleKeyRotation(long now) {
Runnable task = new Runnable() {
@Override
public void run() {
rotateKeys();
}
};
long delay = rotationPeriodLength - now % rotationPeriodLength;
scheduler.schedule(task, delay, MILLISECONDS);
scheduler.schedule((Runnable) this::rotateKeys, delay, MILLISECONDS);
}
private void rotateKeys() {
dbExecutor.execute(new Runnable() {
@Override
public void run() {
dbExecutor.execute(() -> {
try {
Transaction txn = db.startTransaction(false);
try {
Transaction txn = db.startTransaction(false);
try {
rotateKeys(txn);
db.commitTransaction(txn);
} finally {
db.endTransaction(txn);
}
} catch (DbException e) {
if (LOG.isLoggable(WARNING))
LOG.log(WARNING, e.toString(), e);
rotateKeys(txn);
db.commitTransaction(txn);
} finally {
db.endTransaction(txn);
}
} catch (DbException e) {
if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
}
});
}