From 9ec3cf37feeaa7a3daa8e21df2c7580597aacf0b Mon Sep 17 00:00:00 2001 From: =?utf8?q?David=20=E2=80=98Bombe=E2=80=99=20Roden?= Date: Tue, 13 Jan 2015 20:38:46 +0100 Subject: [PATCH] Add backoff to reconnects. --- .../pterodactylus/xdcc/core/ConnectionBackoff.java | 105 +++++++++++++++++++++ .../java/net/pterodactylus/xdcc/core/Core.java | 83 +++++++++++----- .../xdcc/core/ConnectionBackoffTest.java | 61 ++++++++++++ 3 files changed, 228 insertions(+), 21 deletions(-) create mode 100644 src/main/java/net/pterodactylus/xdcc/core/ConnectionBackoff.java create mode 100644 src/test/java/net/pterodactylus/xdcc/core/ConnectionBackoffTest.java diff --git a/src/main/java/net/pterodactylus/xdcc/core/ConnectionBackoff.java b/src/main/java/net/pterodactylus/xdcc/core/ConnectionBackoff.java new file mode 100644 index 0000000..5ba44a6 --- /dev/null +++ b/src/main/java/net/pterodactylus/xdcc/core/ConnectionBackoff.java @@ -0,0 +1,105 @@ +package net.pterodactylus.xdcc.core; + +import java.time.Clock; +import java.time.Instant; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import net.pterodactylus.xdcc.data.Network; + +/** + * Manages backoff times for connections. + * + * @author David ‘Bombe’ Roden + */ +public class ConnectionBackoff { + + private static final long MILLIS_PER_SECOND = TimeUnit.MINUTES.toMillis(1); + private static final long MILLIS_PER_HOUR = TimeUnit.HOURS.toMillis(1); + + private final Clock clock; + private final Map connectionFailures = new HashMap<>(); + + public ConnectionBackoff() { + this(Clock.systemDefaultZone()); + } + + public ConnectionBackoff(Clock clock) { + this.clock = clock; + } + + public void connectionFailed(Network network) { + ConnectionFailureCounter connectionFailureCounter = getConnectionFailureCounter(network); + connectionFailureCounter.countFailure(); + } + + public void connectionSuccessful(Network network) { + ConnectionFailureCounter connectionFailureCounter = getConnectionFailureCounter(network); + connectionFailureCounter.reset(); + } + + public long getBackoff(Network network) { + ConnectionFailureCounter connectionFailureCounter = getConnectionFailureCounter(network); + if (!connectionFailureCounter.hasFailure()) { + return 0; + } + long delay = (long) (MILLIS_PER_SECOND * + Math.pow(1.2, connectionFailureCounter.getFailureCount() - 1)); + return Math.min(MILLIS_PER_HOUR, Math.max(0, + (connectionFailureCounter.getLastFailureTime() + delay) - Instant.now(clock) + .toEpochMilli())); + } + + private ConnectionFailureCounter getConnectionFailureCounter(Network network) { + ConnectionFailureCounter connectionFailureCounter; + synchronized (connectionFailures) { + if (!connectionFailures.containsKey(network)) { + connectionFailures.put(network, new ConnectionFailureCounter()); + } + connectionFailureCounter = connectionFailures.get(network); + } + return connectionFailureCounter; + } + + private class ConnectionFailureCounter { + + private final Object lock = new Object(); + private int count; + private long lastFailureTime; + + public void reset() { + synchronized (lock) { + count = 0; + lastFailureTime = 0; + } + } + + public void countFailure() { + synchronized (lock) { + count++; + lastFailureTime = Instant.now(clock).toEpochMilli(); + } + } + + public boolean hasFailure() { + synchronized (lock) { + return (count > 0); + } + } + + public int getFailureCount() { + synchronized (lock) { + return count; + } + } + + public long getLastFailureTime() { + synchronized (lock) { + return lastFailureTime; + } + } + + } + +} diff --git a/src/main/java/net/pterodactylus/xdcc/core/Core.java b/src/main/java/net/pterodactylus/xdcc/core/Core.java index 6da5dbd..ca91805 100644 --- a/src/main/java/net/pterodactylus/xdcc/core/Core.java +++ b/src/main/java/net/pterodactylus/xdcc/core/Core.java @@ -31,11 +31,15 @@ import java.io.IOException; import java.io.OutputStream; import java.util.Collection; import java.util.Collections; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Set; +import java.util.TreeMap; import java.util.concurrent.TimeUnit; +import java.util.function.Function; import java.util.stream.Collectors; import net.pterodactylus.irc.Connection; @@ -111,6 +115,7 @@ public class Core extends AbstractExecutionThreadService { private final ConnectionFactory connectionFactory; private final ChannelBanManager channelBanManager = new ChannelBanManager(); + private final ConnectionBackoff connectionBackoff = new ConnectionBackoff(); /** The temporary directory to download files to. */ private final String temporaryDirectory; @@ -123,6 +128,7 @@ public class Core extends AbstractExecutionThreadService { /** The channels that are currentlymonitored. */ private final Collection joinedChannels = Sets.newHashSet(); + private final Set channelsBeingJoined = new HashSet<>(); /** The channels that are joined but not configured. */ private final Collection extraChannels = Sets.newHashSet(); @@ -377,39 +383,68 @@ public class Core extends AbstractExecutionThreadService { @Override protected void run() throws Exception { while (isRunning()) { - synchronized (syncObject) { - try { - syncObject.wait(TimeUnit.MINUTES.toMillis(1)); - } catch (InterruptedException ie1) { - /* ignore. */ - } - } - if (!isRunning()) { - break; - } - /* find channels that should be monitored but are not. */ + Set missingChannels = new HashSet<>(); for (Channel channel : channels) { - if (joinedChannels.contains(channel)) { + if (joinedChannels.contains(channel) || channelsBeingJoined.contains(channel)) { continue; } - - /* are we banned from this channel? */ if (channelBanManager.isBanned(channel)) { continue; } - - connectNetwork(channel.network()); - Connection connection = networkConnections.get(channel.network()); - if (connection.established()) { - eventBus.post(new GenericMessage(String.format("Trying to join %s on %s.", channel.name(), channel.network().name()))); + if (networkConnections.containsKey(channel.network())) { + if (networkConnections.get(channel.network()).established()) { + missingChannels.add(channel); + } + } + } + Set missingNetworks = missingChannels.stream() + .map(Channel::network) + .distinct() + .filter((network) -> !networkConnections.containsKey(network)) + .collect(Collectors.toSet()); + + if (!missingChannels.isEmpty()) { + for (Channel missingChannel : missingChannels) { + Network network = missingChannel.network(); + eventBus.post(new GenericMessage(String.format("Trying to join %s on %s...", missingChannel.name(), network))); try { - connection.joinChannel(channel.name()); + channelsBeingJoined.add(missingChannel); + networkConnections.get(network).joinChannel(missingChannel.name()); } catch (IOException ioe1) { - eventBus.post(new GenericMessage(String.format("Could not join %s on %s.", channel.name(), channel.network().name()))); + logger.warn(String.format("Could not join %s on %s!", missingChannel.name(), network.name()), ioe1); } } + } else if (missingNetworks.isEmpty()) { + synchronized (syncObject) { + try { + syncObject.wait(TimeUnit.MINUTES.toMillis(1)); + } catch (InterruptedException ie1) { + /* ignore. */ + } + } + continue; + } + + Map timesForNextConnects = new TreeMap<>(missingNetworks.stream() + .collect(Collectors.toMap(connectionBackoff::getBackoff, Function.identity(), (network, ignore) -> network))); + + Entry firstNetwork = timesForNextConnects.entrySet().stream().findFirst().get(); + if (firstNetwork.getKey() > 0) { + eventBus.post(new GenericMessage(String.format("Waiting %d seconds to connect to %s...", TimeUnit.MILLISECONDS.toMinutes(firstNetwork.getKey()), firstNetwork.getValue().name()))); + synchronized (syncObject) { + try { + syncObject.wait(firstNetwork.getKey()); + } catch (InterruptedException ie1) { + /* ignore. */ + } + } + if (!isRunning()) { + break; + } } + + connectNetwork(firstNetwork.getValue()); } } @@ -510,6 +545,7 @@ public class Core extends AbstractExecutionThreadService { return; } + connectionBackoff.connectionSuccessful(network.get()); eventBus.post(new GenericMessage(String.format("Connected to network %s.", network.get().name()))); /* join all channels on this network. */ @@ -534,6 +570,7 @@ public class Core extends AbstractExecutionThreadService { @Subscribe public void connectionClosed(ConnectionClosed connectionClosed) { removeConnection(connectionClosed.connection()); + connectionBackoff.connectionFailed(getNetwork(connectionClosed.connection()).get()); eventBus.post(new GenericMessage(String.format("Connection closed by %s.", connectionClosed.connection().hostname()))); } @@ -546,6 +583,7 @@ public class Core extends AbstractExecutionThreadService { @Subscribe public void connectionFailed(ConnectionFailed connectionFailed) { removeConnection(connectionFailed.connection()); + connectionBackoff.connectionFailed(getNetwork(connectionFailed.connection()).get()); eventBus.post(new GenericMessage(String.format("Could not connect to %s: %s.", connectionFailed.connection().hostname(), connectionFailed.cause()))); } @@ -573,6 +611,7 @@ public class Core extends AbstractExecutionThreadService { channelBanManager.unban(channel.get()); joinedChannels.add(channel.get()); + channelsBeingJoined.remove(channel.get()); logger.info(String.format("Joined Channel %s on %s.", channelJoined.channel(), network.get().name())); } } @@ -590,6 +629,8 @@ public class Core extends AbstractExecutionThreadService { return; } + channelsBeingJoined.remove(channel.get()); + /* remove all bots for this channel, we might have been kicked. */ Collection botsToRemove = networkBots.row(network.get()) .values().stream() diff --git a/src/test/java/net/pterodactylus/xdcc/core/ConnectionBackoffTest.java b/src/test/java/net/pterodactylus/xdcc/core/ConnectionBackoffTest.java new file mode 100644 index 0000000..adaae8b --- /dev/null +++ b/src/test/java/net/pterodactylus/xdcc/core/ConnectionBackoffTest.java @@ -0,0 +1,61 @@ +package net.pterodactylus.xdcc.core; + +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneId; + +import net.pterodactylus.xdcc.data.Network; + +import org.hamcrest.MatcherAssert; +import org.hamcrest.Matchers; +import org.junit.Test; +import org.mockito.Mockito; + +/** + * Unit test for {@link ConnectionBackoff}. + */ +public class ConnectionBackoffTest { + + private final Network network = Mockito.mock(Network.class); + private final Clock fixedClock = Clock.fixed(Instant.now(), ZoneId.systemDefault()); + private final ConnectionBackoff connectionBackoff = new ConnectionBackoff(fixedClock); + + @Test + public void defaultBackoffCanBeCreated() { + new ConnectionBackoff(); + } + + @Test + public void firstConnectionIsImmediatelyPossible() { + MatcherAssert.assertThat(connectionBackoff.getBackoff(network), Matchers.is(0L)); + } + + @Test + public void afterTheFirstFailedConnectionBackoffIsOneMinute() { + connectionBackoff.connectionFailed(network); + MatcherAssert.assertThat(connectionBackoff.getBackoff(network), Matchers.is(60000L)); + } + + @Test + public void secondFailureIncreasesBackoffTo72Seconds() { + connectionBackoff.connectionFailed(network); + connectionBackoff.connectionFailed(network); + MatcherAssert.assertThat(connectionBackoff.getBackoff(network), Matchers.is(72000L)); + } + + @Test + public void successfulConnectionResetsTheTimer() { + connectionBackoff.connectionFailed(network); + connectionBackoff.connectionSuccessful(network); + MatcherAssert.assertThat(connectionBackoff.getBackoff(network), Matchers.is(0L)); + } + + @Test + public void backoffIsCappedToOneHour() { + for (int i = 0; i < 45; i++) { + connectionBackoff.connectionFailed(network); + } + MatcherAssert.assertThat(connectionBackoff.getBackoff(network), Matchers.is(3600000L)); + } + +} -- 2.7.4