Fix peer database lookups

The previously implemented support of compact tracker announce responses
introduced a bug in the way peers are stored in the BitTorrent client's
peer database: they were always created by host identifier, even when we
knew their peer ID. This made subsequent lookups by peer ID return null,
and triggered the creation of a new peer object for the same peer,
leading to peer exchange disconnection problems, and stale outstanding
requests to no-longer connected peers.

The peer exchange timeout mechanism was also adjusted to better handle
keep-alives and real timeouts that were not detected before.

Version was bumped to 1.0.3 for these fixes.

Signed-off-by: Maxime Petazzoni <[email protected]>

build.xml

@@ -22,7 +22,7 @@
 	<property name="src.dir" location="src" />
 
 	<!-- Release version. -->
-	<property name="project.version" value="1.0.2" />
+	<property name="project.version" value="1.0.3" />
 
 	<path id="project.classpath">
 		<pathelement location="${build.dir}" />
@@ -46,7 +46,7 @@
 
 	<target name="build" depends="init">
 		<javac destdir="${build.dir}" includeantruntime="false"
-			debug="off" sourcepath="" srcdir="${src.dir}">
+			debug="on" sourcepath="" srcdir="${src.dir}">
 			<include name="com/turn/ttorrent/**/*.java" />
 			<compilerarg value="-Xlint" />
 			<classpath refid="project.classpath" />

src/com/turn/ttorrent/client/Client.java

@@ -28,6 +28,7 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.net.SocketException;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedByInterruptException;
@@ -404,6 +405,7 @@ private SharingPeer getOrCreatePeer(byte[] peerId, String ip, int port) {
 			if (search.hasPeerId()) {
 				peer = this.peers.get(search.getHostIdentifier());
 				if (peer != null) {
+					// Set peer ID for perviously known peer.
 					peer.setPeerId(search.getPeerId());
 
 					this.peers.remove(peer.getHostIdentifier());
@@ -413,10 +415,13 @@ private SharingPeer getOrCreatePeer(byte[] peerId, String ip, int port) {
 			}
 
 			peer = new SharingPeer(ip, port, search.getPeerId(), this.torrent);
-			this.peers.putIfAbsent(peer.getHostIdentifier(), peer);
+			this.peers.putIfAbsent(peer.hasPeerId()
+					? peer.getHexPeerId()
+					: peer.getHostIdentifier(),
+				peer);
+			logger.trace("Created new peer " + peer + ".");
 		}
 
-		peer.register(this.torrent);
 		return peer;
 	}
 
@@ -644,16 +649,23 @@ private void processAnnouncedPeer(byte[] peerId, String ip, int port) {
 	public void handleNewPeerConnection(Socket s, byte[] peerId) {
 		SharingPeer peer = this.getOrCreatePeer(peerId,
 				s.getInetAddress().getHostAddress(), s.getPort());
-		this.connected.put(peer.getHexPeerId(), peer);
 
-		synchronized (peer) {
-			peer.register(this);
-			peer.bind(s);
-		}
+		try {
+			synchronized (peer) {
+				peer.register(this);
+				peer.bind(s);
+			}
 
-		logger.info("New peer connection with " + peer +
-				" [" + this.connected.size() + "/" +
-				this.peers.size() + "].");
+			this.connected.put(peer.getHexPeerId(), peer);
+			peer.register(this.torrent);
+			logger.info("New peer connection with " + peer +
+					" [" + this.connected.size() + "/" +
+					this.peers.size() + "].");
+		} catch (SocketException se) {
+			this.connected.remove(peer.getHexPeerId());
+			logger.warn("Could not handle new peer connection " +
+					"with " + peer + ": " + se.getMessage());
+		}
 	}
 
 

src/com/turn/ttorrent/client/peer/PeerExchange.java

@@ -23,6 +23,7 @@
 import java.io.OutputStream;
 import java.lang.InterruptedException;
 import java.net.Socket;
+import java.net.SocketException;
 import java.nio.ByteBuffer;
 import java.text.ParseException;
 import java.util.BitSet;
@@ -80,11 +81,14 @@ class PeerExchange {
 	 * @param socket The connected socket to the peer.
 	 */
 	public PeerExchange(SharingPeer peer, SharedTorrent torrent,
-			Socket socket) {
+			Socket socket) throws SocketException {
 		this.peer = peer;
 		this.torrent = torrent;
 		this.socket = socket;
 
+		// Set the socket read timeout.
+		this.socket.setSoTimeout(PeerExchange.KEEP_ALIVE_FOR_MINUTES*60*1000);
+
 		this.listeners = new HashSet<MessageListener>();
 		this.sendQueue = new LinkedBlockingQueue<Message>();
 
@@ -227,7 +231,7 @@ public void run() {
 				}
 			} catch (IOException ioe) {
 				logger.trace("Could not read message from " + peer +
-						": " + ioe.getMessage());
+						": " + ioe.getMessage(), ioe);
 				peer.unbind(true);
 			}
 		}
@@ -244,12 +248,8 @@ public void run() {
 	 */
 	private class OutgoingThread extends Thread {
 
-		private long lastMessageTime;
-
 		@Override
 		public void run() {
-			this.lastMessageTime = 0;
-
 			try {
 				OutputStream os = socket.getOutputStream();
 
@@ -263,20 +263,11 @@ public void run() {
 								TimeUnit.MINUTES);
 
 						if (message == null) {
-							// If we have to send a keep alive, check it hasn't
-							// been more than KEEP_ALIVE_FOR_MINUTES since the
-							// last message, otherwise we can consider this
-							// peer as dead.
-							if (System.currentTimeMillis() -
-									this.lastMessageTime >
-										PeerExchange.KEEP_ALIVE_FOR_MINUTES) {
-								peer.unbind(true);
+							if (stop) {
 								return;
-							} else {
-								message = Message.KeepAliveMessage.craft();
 							}
-						} else {
-							this.lastMessageTime = 0;
+
+							message = Message.KeepAliveMessage.craft();
 						}
 
 						logger.trace("Sending " + message + " to " + peer + ".");
@@ -287,7 +278,7 @@ public void run() {
 				}
 			} catch (IOException ioe) {
 				logger.trace("Could not send message to " + peer +
-						": " + ioe.getMessage());
+						": " + ioe.getMessage(), ioe);
 				peer.unbind(true);
 			}
 		}

src/com/turn/ttorrent/client/peer/SharingPeer.java

@@ -22,6 +22,7 @@
 
 import java.io.IOException;
 import java.net.Socket;
+import java.net.SocketException;
 import java.nio.ByteBuffer;
 import java.util.BitSet;
 import java.util.Comparator;
@@ -235,7 +236,9 @@ public synchronized boolean isSeed() {
 	 *
 	 * @param socket The connected socket for this peer.
 	 */
-	public synchronized void bind(Socket socket) {
+	public synchronized void bind(Socket socket) throws SocketException {
+		this.unbind(true);
+
 		this.exchange = new PeerExchange(this, this.torrent, socket);
 		this.exchange.register(this);
 
@@ -367,6 +370,10 @@ private synchronized void requestNextBlocks() {
 	 * @param message The PIECE message received.
 	 */
 	private synchronized void removeBlockRequest(Message.PieceMessage message) {
+		if (this.requests == null) {
+			return;
+		}
+
 		for (Message.RequestMessage request : this.requests) {
 			if (request.getPiece() == message.getPiece() &&
 					request.getOffset() == message.getOffset()) {