6 Commits

Author SHA1 Message Date
c435e2f79e gradle wrapper should not require bash
Some checks failed
continuous-integration/drone/push Build is failing
2021-12-08 14:39:47 +01:00
d5d3640991 Add '.drone.yml'
Some checks failed
continuous-integration/drone/push Build is failing
2021-12-08 10:39:25 +01:00
64ee41aee8 Merge branch 'release/2.0.4' 2016-11-01 06:28:44 +01:00
d3205336ed Some tweaks to determine what objects should be requested from other nodes, so they may arrive as quickly as possible. 2016-10-31 06:21:36 +01:00
7b14081c63 Set 'send time' on outgoing messages 2016-10-28 07:52:05 +02:00
e1173d0619 Merge tag '2.0.3' into develop
Version 2.0.3
2016-10-24 08:13:18 +02:00
11 changed files with 151 additions and 100 deletions

9
.drone.yml Normal file
View File

@ -0,0 +1,9 @@
kind: pipeline
name: default
steps:
- name: test
image: adoptopenjdk/openjdk8:alpine
commands:
- ./gradlew assemble
- ./gradlew check

View File

@ -264,6 +264,9 @@ public class Plaintext implements Streamable {
} }
public void setStatus(Status status) { public void setStatus(Status status) {
if (status != Status.RECEIVED && sent == null && status != Status.DRAFT) {
sent = UnixTime.now();
}
this.status = status; this.status = status;
} }

139
gradlew vendored
View File

@ -1,4 +1,20 @@
#!/usr/bin/env bash #!/usr/bin/env sh
#
# Copyright 2015 the original author or authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
############################################################################## ##############################################################################
## ##
@ -6,42 +22,6 @@
## ##
############################################################################## ##############################################################################
# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
DEFAULT_JVM_OPTS=""
APP_NAME="Gradle"
APP_BASE_NAME=`basename "$0"`
# Use the maximum available, or set MAX_FD != -1 to use that value.
MAX_FD="maximum"
warn ( ) {
echo "$*"
}
die ( ) {
echo
echo "$*"
echo
exit 1
}
# OS specific support (must be 'true' or 'false').
cygwin=false
msys=false
darwin=false
case "`uname`" in
CYGWIN* )
cygwin=true
;;
Darwin* )
darwin=true
;;
MINGW* )
msys=true
;;
esac
# Attempt to set APP_HOME # Attempt to set APP_HOME
# Resolve links: $0 may be a link # Resolve links: $0 may be a link
PRG="$0" PRG="$0"
@ -60,8 +40,49 @@ cd "`dirname \"$PRG\"`/" >/dev/null
APP_HOME="`pwd -P`" APP_HOME="`pwd -P`"
cd "$SAVED" >/dev/null cd "$SAVED" >/dev/null
APP_NAME="Gradle"
APP_BASE_NAME=`basename "$0"`
# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
# Use the maximum available, or set MAX_FD != -1 to use that value.
MAX_FD="maximum"
warn () {
echo "$*"
}
die () {
echo
echo "$*"
echo
exit 1
}
# OS specific support (must be 'true' or 'false').
cygwin=false
msys=false
darwin=false
nonstop=false
case "`uname`" in
CYGWIN* )
cygwin=true
;;
Darwin* )
darwin=true
;;
MSYS* | MINGW* )
msys=true
;;
NONSTOP* )
nonstop=true
;;
esac
CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
# Determine the Java command to use to start the JVM. # Determine the Java command to use to start the JVM.
if [ -n "$JAVA_HOME" ] ; then if [ -n "$JAVA_HOME" ] ; then
if [ -x "$JAVA_HOME/jre/sh/java" ] ; then if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
@ -85,7 +106,7 @@ location of your Java installation."
fi fi
# Increase the maximum file descriptors if we can. # Increase the maximum file descriptors if we can.
if [ "$cygwin" = "false" -a "$darwin" = "false" ] ; then if [ "$cygwin" = "false" -a "$darwin" = "false" -a "$nonstop" = "false" ] ; then
MAX_FD_LIMIT=`ulimit -H -n` MAX_FD_LIMIT=`ulimit -H -n`
if [ $? -eq 0 ] ; then if [ $? -eq 0 ] ; then
if [ "$MAX_FD" = "maximum" -o "$MAX_FD" = "max" ] ; then if [ "$MAX_FD" = "maximum" -o "$MAX_FD" = "max" ] ; then
@ -105,10 +126,11 @@ if $darwin; then
GRADLE_OPTS="$GRADLE_OPTS \"-Xdock:name=$APP_NAME\" \"-Xdock:icon=$APP_HOME/media/gradle.icns\"" GRADLE_OPTS="$GRADLE_OPTS \"-Xdock:name=$APP_NAME\" \"-Xdock:icon=$APP_HOME/media/gradle.icns\""
fi fi
# For Cygwin, switch paths to Windows format before running java # For Cygwin or MSYS, switch paths to Windows format before running java
if $cygwin ; then if [ "$cygwin" = "true" -o "$msys" = "true" ] ; then
APP_HOME=`cygpath --path --mixed "$APP_HOME"` APP_HOME=`cygpath --path --mixed "$APP_HOME"`
CLASSPATH=`cygpath --path --mixed "$CLASSPATH"` CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
JAVACMD=`cygpath --unix "$JAVACMD"` JAVACMD=`cygpath --unix "$JAVACMD"`
# We build the pattern for arguments to be converted via cygpath # We build the pattern for arguments to be converted via cygpath
@ -134,27 +156,30 @@ if $cygwin ; then
else else
eval `echo args$i`="\"$arg\"" eval `echo args$i`="\"$arg\""
fi fi
i=$((i+1)) i=`expr $i + 1`
done done
case $i in case $i in
(0) set -- ;; 0) set -- ;;
(1) set -- "$args0" ;; 1) set -- "$args0" ;;
(2) set -- "$args0" "$args1" ;; 2) set -- "$args0" "$args1" ;;
(3) set -- "$args0" "$args1" "$args2" ;; 3) set -- "$args0" "$args1" "$args2" ;;
(4) set -- "$args0" "$args1" "$args2" "$args3" ;; 4) set -- "$args0" "$args1" "$args2" "$args3" ;;
(5) set -- "$args0" "$args1" "$args2" "$args3" "$args4" ;; 5) set -- "$args0" "$args1" "$args2" "$args3" "$args4" ;;
(6) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" ;; 6) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" ;;
(7) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" ;; 7) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" ;;
(8) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" ;; 8) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" ;;
(9) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" "$args8" ;; 9) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" "$args8" ;;
esac esac
fi fi
# Split up the JVM_OPTS And GRADLE_OPTS values into an array, following the shell quoting and substitution rules # Escape application args
function splitJvmOpts() { save () {
JVM_OPTS=("$@") for i do printf %s\\n "$i" | sed "s/'/'\\\\''/g;1s/^/'/;\$s/\$/' \\\\/" ; done
echo " "
} }
eval splitJvmOpts $DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS APP_ARGS=`save "$@"`
JVM_OPTS[${#JVM_OPTS[*]}]="-Dorg.gradle.appname=$APP_BASE_NAME"
exec "$JAVACMD" "${JVM_OPTS[@]}" -classpath "$CLASSPATH" org.gradle.wrapper.GradleWrapperMain "$@" # Collect all arguments for the java command, following the shell quoting and substitution rules
eval set -- $DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS "\"-Dorg.gradle.appname=$APP_BASE_NAME\"" -classpath "\"$CLASSPATH\"" org.gradle.wrapper.GradleWrapperMain "$APP_ARGS"
exec "$JAVACMD" "$@"

View File

@ -54,7 +54,7 @@ public abstract class AbstractConnection {
protected final NetworkHandler.MessageListener listener; protected final NetworkHandler.MessageListener listener;
protected final Map<InventoryVector, Long> ivCache; protected final Map<InventoryVector, Long> ivCache;
protected final Deque<MessagePayload> sendingQueue; protected final Deque<MessagePayload> sendingQueue;
protected final Set<InventoryVector> commonRequestedObjects; protected final Map<InventoryVector, Long> commonRequestedObjects;
protected final Set<InventoryVector> requestedObjects; protected final Set<InventoryVector> requestedObjects;
protected volatile State state; protected volatile State state;
@ -71,7 +71,7 @@ public abstract class AbstractConnection {
public AbstractConnection(InternalContext context, Mode mode, public AbstractConnection(InternalContext context, Mode mode,
NetworkAddress node, NetworkAddress node,
Set<InventoryVector> commonRequestedObjects, Map<InventoryVector, Long> commonRequestedObjects,
long syncTimeout) { long syncTimeout) {
this.ctx = context; this.ctx = context;
this.mode = mode; this.mode = mode;
@ -143,7 +143,7 @@ public abstract class AbstractConnection {
int originalSize = inv.getInventory().size(); int originalSize = inv.getInventory().size();
updateIvCache(inv.getInventory()); updateIvCache(inv.getInventory());
List<InventoryVector> missing = ctx.getInventory().getMissing(inv.getInventory(), streams); List<InventoryVector> missing = ctx.getInventory().getMissing(inv.getInventory(), streams);
missing.removeAll(commonRequestedObjects); missing.removeAll(commonRequestedObjects.keySet());
LOG.trace("Received inventory with " + originalSize + " elements, of which are " LOG.trace("Received inventory with " + originalSize + " elements, of which are "
+ missing.size() + " missing."); + missing.size() + " missing.");
send(new GetData.Builder().inventory(missing).build()); send(new GetData.Builder().inventory(missing).build());
@ -175,7 +175,7 @@ public abstract class AbstractConnection {
} catch (IOException e) { } catch (IOException e) {
LOG.error("Stream " + objectMessage.getStream() + ", object type " + objectMessage.getType() + ": " + e.getMessage(), e); LOG.error("Stream " + objectMessage.getStream() + ", object type " + objectMessage.getType() + ": " + e.getMessage(), e);
} finally { } finally {
if (!commonRequestedObjects.remove(objectMessage.getInventoryVector())) { if (commonRequestedObjects.remove(objectMessage.getInventoryVector()) == null) {
LOG.debug("Received object that wasn't requested."); LOG.debug("Received object that wasn't requested.");
} }
} }
@ -205,6 +205,10 @@ public abstract class AbstractConnection {
return ivCache.containsKey(iv); return ivCache.containsKey(iv);
} }
public boolean requested(InventoryVector iv) {
return requestedObjects.contains(iv);
}
private void cleanupIvCache() { private void cleanupIvCache() {
Long fiveMinutesAgo = UnixTime.now(-5 * MINUTE); Long fiveMinutesAgo = UnixTime.now(-5 * MINUTE);
for (Map.Entry<InventoryVector, Long> entry : ivCache.entrySet()) { for (Map.Entry<InventoryVector, Long> entry : ivCache.entrySet()) {

View File

@ -36,9 +36,9 @@ import java.net.InetAddress;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.net.Socket; import java.net.Socket;
import java.net.SocketTimeoutException; import java.net.SocketTimeoutException;
import java.util.HashSet; import java.util.HashMap;
import java.util.Map;
import java.util.Objects; import java.util.Objects;
import java.util.Set;
import static ch.dissem.bitmessage.networking.AbstractConnection.Mode.CLIENT; import static ch.dissem.bitmessage.networking.AbstractConnection.Mode.CLIENT;
import static ch.dissem.bitmessage.networking.AbstractConnection.Mode.SYNC; import static ch.dissem.bitmessage.networking.AbstractConnection.Mode.SYNC;
@ -64,20 +64,20 @@ class Connection extends AbstractConnection {
private boolean socketInitialized; private boolean socketInitialized;
public Connection(InternalContext context, Mode mode, Socket socket, public Connection(InternalContext context, Mode mode, Socket socket,
Set<InventoryVector> requestedObjectsMap) throws IOException { Map<InventoryVector, Long> requestedObjectsMap) throws IOException {
this(context, mode, socket, requestedObjectsMap, this(context, mode, socket, requestedObjectsMap,
new NetworkAddress.Builder().ip(socket.getInetAddress()).port(socket.getPort()).stream(1).build(), new NetworkAddress.Builder().ip(socket.getInetAddress()).port(socket.getPort()).stream(1).build(),
0); 0);
} }
public Connection(InternalContext context, Mode mode, NetworkAddress node, public Connection(InternalContext context, Mode mode, NetworkAddress node,
Set<InventoryVector> requestedObjectsMap) { Map<InventoryVector, Long> requestedObjectsMap) {
this(context, mode, new Socket(), requestedObjectsMap, this(context, mode, new Socket(), requestedObjectsMap,
node, 0); node, 0);
} }
private Connection(InternalContext context, Mode mode, Socket socket, private Connection(InternalContext context, Mode mode, Socket socket,
Set<InventoryVector> commonRequestedObjects, NetworkAddress node, long syncTimeout) { Map<InventoryVector, Long> commonRequestedObjects, NetworkAddress node, long syncTimeout) {
super(context, mode, node, commonRequestedObjects, syncTimeout); super(context, mode, node, commonRequestedObjects, syncTimeout);
this.startTime = UnixTime.now(); this.startTime = UnixTime.now();
this.socket = socket; this.socket = socket;
@ -86,9 +86,9 @@ class Connection extends AbstractConnection {
public static Connection sync(InternalContext ctx, InetAddress address, int port, MessageListener listener, public static Connection sync(InternalContext ctx, InetAddress address, int port, MessageListener listener,
long timeoutInSeconds) throws IOException { long timeoutInSeconds) throws IOException {
return new Connection(ctx, SYNC, new Socket(address, port), return new Connection(ctx, SYNC, new Socket(address, port),
new HashSet<InventoryVector>(), new HashMap<InventoryVector, Long>(),
new NetworkAddress.Builder().ip(address).port(port).stream(1).build(), new NetworkAddress.Builder().ip(address).port(port).stream(1).build(),
timeoutInSeconds); timeoutInSeconds);
} }
public long getStartTime() { public long getStartTime() {

View File

@ -18,7 +18,6 @@ package ch.dissem.bitmessage.networking;
import ch.dissem.bitmessage.InternalContext; import ch.dissem.bitmessage.InternalContext;
import ch.dissem.bitmessage.entity.valueobject.NetworkAddress; import ch.dissem.bitmessage.entity.valueobject.NetworkAddress;
import ch.dissem.bitmessage.ports.NetworkHandler;
import ch.dissem.bitmessage.utils.UnixTime; import ch.dissem.bitmessage.utils.UnixTime;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -32,12 +31,13 @@ import static ch.dissem.bitmessage.networking.DefaultNetworkHandler.NETWORK_MAGI
/** /**
* @author Christian Basler * @author Christian Basler
*/ */
@Deprecated
@SuppressWarnings("deprecation")
public class ConnectionOrganizer implements Runnable { public class ConnectionOrganizer implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(ConnectionOrganizer.class); private static final Logger LOG = LoggerFactory.getLogger(ConnectionOrganizer.class);
private final InternalContext ctx; private final InternalContext ctx;
private final DefaultNetworkHandler networkHandler; private final DefaultNetworkHandler networkHandler;
private final NetworkHandler.MessageListener listener;
private Connection initialConnection; private Connection initialConnection;
@ -45,7 +45,6 @@ public class ConnectionOrganizer implements Runnable {
DefaultNetworkHandler networkHandler) { DefaultNetworkHandler networkHandler) {
this.ctx = ctx; this.ctx = ctx;
this.networkHandler = networkHandler; this.networkHandler = networkHandler;
this.listener = ctx.getNetworkListener();
} }
@Override @Override
@ -87,7 +86,7 @@ public class ConnectionOrganizer implements Runnable {
if (active < NETWORK_MAGIC_NUMBER) { if (active < NETWORK_MAGIC_NUMBER) {
List<NetworkAddress> addresses = ctx.getNodeRegistry().getKnownAddresses( List<NetworkAddress> addresses = ctx.getNodeRegistry().getKnownAddresses(
NETWORK_MAGIC_NUMBER - active, ctx.getStreams()); NETWORK_MAGIC_NUMBER - active, ctx.getStreams());
boolean first = active == 0 && initialConnection == null; boolean first = active == 0 && initialConnection == null;
for (NetworkAddress address : addresses) { for (NetworkAddress address : addresses) {
Connection c = new Connection(ctx, CLIENT, address, networkHandler.requestedObjects); Connection c = new Connection(ctx, CLIENT, address, networkHandler.requestedObjects);

View File

@ -39,7 +39,6 @@ import static ch.dissem.bitmessage.networking.AbstractConnection.Mode.SERVER;
import static ch.dissem.bitmessage.networking.AbstractConnection.State.ACTIVE; import static ch.dissem.bitmessage.networking.AbstractConnection.State.ACTIVE;
import static ch.dissem.bitmessage.utils.DebugUtils.inc; import static ch.dissem.bitmessage.utils.DebugUtils.inc;
import static ch.dissem.bitmessage.utils.ThreadFactoryBuilder.pool; import static ch.dissem.bitmessage.utils.ThreadFactoryBuilder.pool;
import static java.util.Collections.newSetFromMap;
/** /**
* Handles all the networky stuff. * Handles all the networky stuff.
@ -59,7 +58,7 @@ public class DefaultNetworkHandler implements NetworkHandler, ContextHolder {
private ServerRunnable server; private ServerRunnable server;
private volatile boolean running; private volatile boolean running;
final Set<InventoryVector> requestedObjects = newSetFromMap(new ConcurrentHashMap<InventoryVector, Boolean>(50_000)); final Map<InventoryVector, Long> requestedObjects = new ConcurrentHashMap<>(50_000);
@Override @Override
public void setContext(InternalContext context) { public void setContext(InternalContext context) {

View File

@ -31,6 +31,7 @@ import static ch.dissem.bitmessage.networking.AbstractConnection.Mode.SERVER;
/** /**
* @author Christian Basler * @author Christian Basler
*/ */
@Deprecated
public class ServerRunnable implements Runnable, Closeable { public class ServerRunnable implements Runnable, Closeable {
private static final Logger LOG = LoggerFactory.getLogger(ServerRunnable.class); private static final Logger LOG = LoggerFactory.getLogger(ServerRunnable.class);
private final InternalContext ctx; private final InternalContext ctx;

View File

@ -26,11 +26,10 @@ import ch.dissem.bitmessage.entity.valueobject.NetworkAddress;
import ch.dissem.bitmessage.exception.NodeException; import ch.dissem.bitmessage.exception.NodeException;
import ch.dissem.bitmessage.factory.V3MessageReader; import ch.dissem.bitmessage.factory.V3MessageReader;
import ch.dissem.bitmessage.networking.AbstractConnection; import ch.dissem.bitmessage.networking.AbstractConnection;
import ch.dissem.bitmessage.utils.UnixTime;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.util.Iterator; import java.util.*;
import java.util.Queue;
import java.util.Set;
import static ch.dissem.bitmessage.networking.AbstractConnection.Mode.CLIENT; import static ch.dissem.bitmessage.networking.AbstractConnection.Mode.CLIENT;
import static ch.dissem.bitmessage.networking.AbstractConnection.Mode.SYNC; import static ch.dissem.bitmessage.networking.AbstractConnection.Mode.SYNC;
@ -46,7 +45,7 @@ public class ConnectionInfo extends AbstractConnection {
private long lastUpdate = System.currentTimeMillis(); private long lastUpdate = System.currentTimeMillis();
public ConnectionInfo(InternalContext context, Mode mode, NetworkAddress node, public ConnectionInfo(InternalContext context, Mode mode, NetworkAddress node,
Set<InventoryVector> commonRequestedObjects, long syncTimeout) { Map<InventoryVector, Long> commonRequestedObjects, long syncTimeout) {
super(context, mode, node, commonRequestedObjects, syncTimeout); super(context, mode, node, commonRequestedObjects, syncTimeout);
headerOut.flip(); headerOut.flip();
if (mode == CLIENT || mode == SYNC) { if (mode == CLIENT || mode == SYNC) {
@ -147,8 +146,12 @@ public class ConnectionInfo extends AbstractConnection {
protected void send(MessagePayload payload) { protected void send(MessagePayload payload) {
sendingQueue.add(payload); sendingQueue.add(payload);
if (payload instanceof GetData) { if (payload instanceof GetData) {
requestedObjects.addAll(((GetData) payload).getInventory()); Long now = UnixTime.now();
commonRequestedObjects.addAll(((GetData) payload).getInventory()); List<InventoryVector> inventory = ((GetData) payload).getInventory();
requestedObjects.addAll(inventory);
for (InventoryVector iv : inventory) {
commonRequestedObjects.put(iv, now);
}
} }
} }

View File

@ -46,14 +46,14 @@ import static ch.dissem.bitmessage.utils.Collections.selectRandom;
import static ch.dissem.bitmessage.utils.DebugUtils.inc; import static ch.dissem.bitmessage.utils.DebugUtils.inc;
import static ch.dissem.bitmessage.utils.ThreadFactoryBuilder.pool; import static ch.dissem.bitmessage.utils.ThreadFactoryBuilder.pool;
import static java.nio.channels.SelectionKey.*; import static java.nio.channels.SelectionKey.*;
import static java.util.Collections.newSetFromMap;
/** /**
* Network handler using java.nio, resulting in less threads. * Network handler using java.nio, resulting in less threads.
*/ */
public class NioNetworkHandler implements NetworkHandler, InternalContext.ContextHolder { public class NioNetworkHandler implements NetworkHandler, InternalContext.ContextHolder {
private static final Logger LOG = LoggerFactory.getLogger(NioNetworkHandler.class); private static final Logger LOG = LoggerFactory.getLogger(NioNetworkHandler.class);
private static final long REQUESTED_OBJECTS_MAX_TIME = 30 * 60_000; // 30 minutes private static final long REQUESTED_OBJECTS_MAX_TIME = 2 * 60_000; // 2 minutes
private static final Long DELAYED = Long.MIN_VALUE;
private final ExecutorService threadPool = Executors.newCachedThreadPool( private final ExecutorService threadPool = Executors.newCachedThreadPool(
pool("network") pool("network")
@ -66,8 +66,7 @@ public class NioNetworkHandler implements NetworkHandler, InternalContext.Contex
private ServerSocketChannel serverChannel; private ServerSocketChannel serverChannel;
private Queue<NetworkAddress> connectionQueue = new ConcurrentLinkedQueue<>(); private Queue<NetworkAddress> connectionQueue = new ConcurrentLinkedQueue<>();
private Map<ConnectionInfo, SelectionKey> connections = new ConcurrentHashMap<>(); private Map<ConnectionInfo, SelectionKey> connections = new ConcurrentHashMap<>();
private final Set<InventoryVector> requestedObjects = newSetFromMap(new ConcurrentHashMap<InventoryVector, Boolean>(10_000)); private final Map<InventoryVector, Long> requestedObjects = new ConcurrentHashMap<>(10_000);
private long requestedObjectsTimeout = 0;
private Thread starter; private Thread starter;
@ -80,7 +79,7 @@ public class NioNetworkHandler implements NetworkHandler, InternalContext.Contex
channel.configureBlocking(false); channel.configureBlocking(false);
ConnectionInfo connection = new ConnectionInfo(ctx, SYNC, ConnectionInfo connection = new ConnectionInfo(ctx, SYNC,
new NetworkAddress.Builder().ip(server).port(port).stream(1).build(), new NetworkAddress.Builder().ip(server).port(port).stream(1).build(),
new HashSet<InventoryVector>(), timeoutInSeconds); new HashMap<InventoryVector, Long>(), timeoutInSeconds);
while (channel.isConnected() && !connection.isSyncFinished()) { while (channel.isConnected() && !connection.isSyncFinished()) {
write(channel, connection); write(channel, connection);
read(channel, connection); read(channel, connection);
@ -147,7 +146,6 @@ public class NioNetworkHandler implements NetworkHandler, InternalContext.Contex
} catch (IOException e) { } catch (IOException e) {
throw new ApplicationException(e); throw new ApplicationException(e);
} }
requestedObjectsTimeout = System.currentTimeMillis() + REQUESTED_OBJECTS_MAX_TIME;
requestedObjects.clear(); requestedObjects.clear();
starter = thread("connection manager", new Runnable() { starter = thread("connection manager", new Runnable() {
@ -189,15 +187,22 @@ public class NioNetworkHandler implements NetworkHandler, InternalContext.Contex
// The list 'requested objects' helps to prevent downloading an object // The list 'requested objects' helps to prevent downloading an object
// twice. From time to time there is an error though, and an object is // twice. From time to time there is an error though, and an object is
// never downloaded. To prevent a large list of failed objects and give // never downloaded. To prevent a large list of failed objects and give
// them a chance to get downloaded again, let's clear the list from time // them a chance to get downloaded again, we will attempt to download an
// to time. The timeout should be such that most of the initial object // object from another node after some time out.
// sync should be done by then, but small enough to prevent objects with long timedOut = System.currentTimeMillis() - REQUESTED_OBJECTS_MAX_TIME;
// a normal time out from not being downloaded at all. List<InventoryVector> delayed = new LinkedList<>();
long now = System.currentTimeMillis(); Iterator<Map.Entry<InventoryVector, Long>> iterator = requestedObjects.entrySet().iterator();
if (now > requestedObjectsTimeout) { while (iterator.hasNext()) {
requestedObjectsTimeout = now + REQUESTED_OBJECTS_MAX_TIME; Map.Entry<InventoryVector, Long> e = iterator.next();
requestedObjects.clear(); //noinspection NumberEquality
if (e.getValue() == DELAYED) {
iterator.remove();
} else if (e.getValue() < timedOut) {
delayed.add(e.getKey());
e.setValue(DELAYED);
}
} }
request(delayed);
try { try {
Thread.sleep(30_000); Thread.sleep(30_000);
@ -422,7 +427,7 @@ public class NioNetworkHandler implements NetworkHandler, InternalContext.Contex
break; break;
} }
} }
if (connection.knowsOf(next)) { if (connection.knowsOf(next) && !connection.requested(next)) {
List<InventoryVector> ivs = distribution.get(connection); List<InventoryVector> ivs = distribution.get(connection);
if (ivs.size() == GetData.MAX_INVENTORY_SIZE) { if (ivs.size() == GetData.MAX_INVENTORY_SIZE) {
connection.send(new GetData.Builder().inventory(ivs).build()); connection.send(new GetData.Builder().inventory(ivs).build());
@ -442,7 +447,9 @@ public class NioNetworkHandler implements NetworkHandler, InternalContext.Contex
} while (iterator.hasNext()); } while (iterator.hasNext());
// remove objects nobody knows of // remove objects nobody knows of
requestedObjects.removeAll(inventoryVectors); for (InventoryVector iv : inventoryVectors) {
requestedObjects.remove(iv);
}
for (ConnectionInfo connection : distribution.keySet()) { for (ConnectionInfo connection : distribution.keySet()) {
List<InventoryVector> ivs = distribution.get(connection); List<InventoryVector> ivs = distribution.get(connection);

View File

@ -76,6 +76,7 @@ public class NetworkHandlerTest {
} }
@Parameterized.Parameters @Parameterized.Parameters
@SuppressWarnings("deprecation")
public static List<Object[]> parameters() { public static List<Object[]> parameters() {
return Arrays.asList(new Object[][]{ return Arrays.asList(new Object[][]{
{new DefaultNetworkHandler(), new DefaultNetworkHandler()}, {new DefaultNetworkHandler(), new DefaultNetworkHandler()},