forked from I2P_Developers/i2p.i2p
* i2psnark:
- Defer piece loading until required - Stub out Extension message support
This commit is contained in:
14
apps/i2psnark/java/src/org/klomp/snark/DataLoader.java
Normal file
14
apps/i2psnark/java/src/org/klomp/snark/DataLoader.java
Normal file
@ -0,0 +1,14 @@
|
|||||||
|
package org.klomp.snark;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Callback used to fetch data
|
||||||
|
* @since 0.8.2
|
||||||
|
*/
|
||||||
|
interface DataLoader
|
||||||
|
{
|
||||||
|
/**
|
||||||
|
* This is the callback that PeerConnectionOut calls to get the data from disk
|
||||||
|
* @return bytes or null for errors
|
||||||
|
*/
|
||||||
|
public byte[] loadData(int piece, int begin, int length);
|
||||||
|
}
|
@ -39,23 +39,28 @@ class Message
|
|||||||
final static byte REQUEST = 6;
|
final static byte REQUEST = 6;
|
||||||
final static byte PIECE = 7;
|
final static byte PIECE = 7;
|
||||||
final static byte CANCEL = 8;
|
final static byte CANCEL = 8;
|
||||||
|
final static byte EXTENSION = 20;
|
||||||
|
|
||||||
// Not all fields are used for every message.
|
// Not all fields are used for every message.
|
||||||
// KEEP_ALIVE doesn't have a real wire representation
|
// KEEP_ALIVE doesn't have a real wire representation
|
||||||
byte type;
|
byte type;
|
||||||
|
|
||||||
// Used for HAVE, REQUEST, PIECE and CANCEL messages.
|
// Used for HAVE, REQUEST, PIECE and CANCEL messages.
|
||||||
|
// low byte used for EXTENSION message
|
||||||
int piece;
|
int piece;
|
||||||
|
|
||||||
// Used for REQUEST, PIECE and CANCEL messages.
|
// Used for REQUEST, PIECE and CANCEL messages.
|
||||||
int begin;
|
int begin;
|
||||||
int length;
|
int length;
|
||||||
|
|
||||||
// Used for PIECE and BITFIELD messages
|
// Used for PIECE and BITFIELD and EXTENSION messages
|
||||||
byte[] data;
|
byte[] data;
|
||||||
int off;
|
int off;
|
||||||
int len;
|
int len;
|
||||||
|
|
||||||
|
// Used to do deferred fetch of data
|
||||||
|
DataLoader dataLoader;
|
||||||
|
|
||||||
SimpleTimer.TimedEvent expireEvent;
|
SimpleTimer.TimedEvent expireEvent;
|
||||||
|
|
||||||
/** Utility method for sending a message through a DataStream. */
|
/** Utility method for sending a message through a DataStream. */
|
||||||
@ -68,6 +73,13 @@ class Message
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Get deferred data
|
||||||
|
if (data == null && dataLoader != null) {
|
||||||
|
data = dataLoader.loadData(piece, begin, length);
|
||||||
|
if (data == null)
|
||||||
|
return; // hmm will get retried, but shouldn't happen
|
||||||
|
}
|
||||||
|
|
||||||
// Calculate the total length in bytes
|
// Calculate the total length in bytes
|
||||||
|
|
||||||
// Type is one byte.
|
// Type is one byte.
|
||||||
@ -85,8 +97,12 @@ class Message
|
|||||||
if (type == REQUEST || type == CANCEL)
|
if (type == REQUEST || type == CANCEL)
|
||||||
datalen += 4;
|
datalen += 4;
|
||||||
|
|
||||||
|
// length is 1 byte
|
||||||
|
if (type == EXTENSION)
|
||||||
|
datalen += 1;
|
||||||
|
|
||||||
// add length of data for piece or bitfield array.
|
// add length of data for piece or bitfield array.
|
||||||
if (type == BITFIELD || type == PIECE)
|
if (type == BITFIELD || type == PIECE || type == EXTENSION)
|
||||||
datalen += len;
|
datalen += len;
|
||||||
|
|
||||||
// Send length
|
// Send length
|
||||||
@ -105,8 +121,11 @@ class Message
|
|||||||
if (type == REQUEST || type == CANCEL)
|
if (type == REQUEST || type == CANCEL)
|
||||||
dos.writeInt(length);
|
dos.writeInt(length);
|
||||||
|
|
||||||
|
if (type == EXTENSION)
|
||||||
|
dos.writeByte((byte) piece & 0xff);
|
||||||
|
|
||||||
// Send actual data
|
// Send actual data
|
||||||
if (type == BITFIELD || type == PIECE)
|
if (type == BITFIELD || type == PIECE || type == EXTENSION)
|
||||||
dos.write(data, off, len);
|
dos.write(data, off, len);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -135,6 +154,8 @@ class Message
|
|||||||
return "PIECE(" + piece + "," + begin + "," + length + ")";
|
return "PIECE(" + piece + "," + begin + "," + length + ")";
|
||||||
case CANCEL:
|
case CANCEL:
|
||||||
return "CANCEL(" + piece + "," + begin + "," + length + ")";
|
return "CANCEL(" + piece + "," + begin + "," + length + ")";
|
||||||
|
case EXTENSION:
|
||||||
|
return "EXTENSION(" + piece + ',' + data.length + ')';
|
||||||
default:
|
default:
|
||||||
return "<UNKNOWN>";
|
return "<UNKNOWN>";
|
||||||
}
|
}
|
||||||
|
@ -59,6 +59,11 @@ public class Peer implements Comparable
|
|||||||
private long uploaded_old[] = {-1,-1,-1};
|
private long uploaded_old[] = {-1,-1,-1};
|
||||||
private long downloaded_old[] = {-1,-1,-1};
|
private long downloaded_old[] = {-1,-1,-1};
|
||||||
|
|
||||||
|
// bytes per bt spec: 0011223344556677
|
||||||
|
static final long OPTION_EXTENSION = 0x0000000000100000l;
|
||||||
|
static final long OPTION_FAST = 0x0000000000000004l;
|
||||||
|
private long options;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Creates a disconnected peer given a PeerID, your own id and the
|
* Creates a disconnected peer given a PeerID, your own id and the
|
||||||
* relevant MetaInfo.
|
* relevant MetaInfo.
|
||||||
@ -285,9 +290,8 @@ public class Peer implements Comparable
|
|||||||
// Handshake write - header
|
// Handshake write - header
|
||||||
dout.write(19);
|
dout.write(19);
|
||||||
dout.write("BitTorrent protocol".getBytes("UTF-8"));
|
dout.write("BitTorrent protocol".getBytes("UTF-8"));
|
||||||
// Handshake write - zeros
|
// Handshake write - options
|
||||||
byte[] zeros = new byte[8];
|
dout.writeLong(OPTION_EXTENSION);
|
||||||
dout.write(zeros);
|
|
||||||
// Handshake write - metainfo hash
|
// Handshake write - metainfo hash
|
||||||
byte[] shared_hash = metainfo.getInfoHash();
|
byte[] shared_hash = metainfo.getInfoHash();
|
||||||
dout.write(shared_hash);
|
dout.write(shared_hash);
|
||||||
@ -312,8 +316,8 @@ public class Peer implements Comparable
|
|||||||
+ "'Bittorrent protocol', got '"
|
+ "'Bittorrent protocol', got '"
|
||||||
+ bittorrentProtocol + "'");
|
+ bittorrentProtocol + "'");
|
||||||
|
|
||||||
// Handshake read - zeros
|
// Handshake read - options
|
||||||
din.readFully(zeros);
|
options = din.readLong();
|
||||||
|
|
||||||
// Handshake read - metainfo hash
|
// Handshake read - metainfo hash
|
||||||
bs = new byte[20];
|
bs = new byte[20];
|
||||||
@ -325,6 +329,15 @@ public class Peer implements Comparable
|
|||||||
din.readFully(bs);
|
din.readFully(bs);
|
||||||
if (_log.shouldLog(Log.DEBUG))
|
if (_log.shouldLog(Log.DEBUG))
|
||||||
_log.debug("Read the remote side's hash and peerID fully from " + toString());
|
_log.debug("Read the remote side's hash and peerID fully from " + toString());
|
||||||
|
|
||||||
|
// if ((options & OPTION_EXTENSION) != 0) {
|
||||||
|
if (options != 0) {
|
||||||
|
// send them something
|
||||||
|
if (_log.shouldLog(Log.DEBUG))
|
||||||
|
//_log.debug("Peer supports extension message, what should we say? " + toString());
|
||||||
|
_log.debug("Peer supports options 0x" + Long.toString(options, 16) + ", what should we say? " + toString());
|
||||||
|
}
|
||||||
|
|
||||||
return bs;
|
return bs;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -171,6 +171,13 @@ class PeerConnectionIn implements Runnable
|
|||||||
if (_log.shouldLog(Log.DEBUG))
|
if (_log.shouldLog(Log.DEBUG))
|
||||||
_log.debug("Received cancel(" + piece + "," + begin + ") from " + peer + " on " + peer.metainfo.getName());
|
_log.debug("Received cancel(" + piece + "," + begin + ") from " + peer + " on " + peer.metainfo.getName());
|
||||||
break;
|
break;
|
||||||
|
case 20: // Extension message
|
||||||
|
int id = din.readUnsignedByte();
|
||||||
|
byte[] payload = new byte[i-2];
|
||||||
|
din.readFully(payload);
|
||||||
|
ps.extensionMessage(id, payload);
|
||||||
|
if (_log.shouldLog(Log.DEBUG))
|
||||||
|
_log.debug("Received extension message from " + peer + " on " + peer.metainfo.getName());
|
||||||
default:
|
default:
|
||||||
byte[] bs = new byte[i-1];
|
byte[] bs = new byte[i-1];
|
||||||
din.readFully(bs);
|
din.readFully(bs);
|
||||||
|
@ -430,6 +430,33 @@ class PeerConnectionOut implements Runnable
|
|||||||
return total;
|
return total;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** @since 0.8.2 */
|
||||||
|
void sendPiece(int piece, int begin, int length, DataLoader loader)
|
||||||
|
{
|
||||||
|
boolean sendNow = false;
|
||||||
|
// are there any cases where we should?
|
||||||
|
|
||||||
|
if (sendNow) {
|
||||||
|
// queue the real thing
|
||||||
|
byte[] bytes = loader.loadData(piece, begin, length);
|
||||||
|
if (bytes != null)
|
||||||
|
sendPiece(piece, begin, length, bytes);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
// queue a fake message... set everything up,
|
||||||
|
// except save the PeerState instead of the bytes.
|
||||||
|
Message m = new Message();
|
||||||
|
m.type = Message.PIECE;
|
||||||
|
m.piece = piece;
|
||||||
|
m.begin = begin;
|
||||||
|
m.length = length;
|
||||||
|
m.dataLoader = loader;
|
||||||
|
m.off = 0;
|
||||||
|
m.len = length;
|
||||||
|
addMessage(m);
|
||||||
|
}
|
||||||
|
|
||||||
void sendPiece(int piece, int begin, int length, byte[] bytes)
|
void sendPiece(int piece, int begin, int length, byte[] bytes)
|
||||||
{
|
{
|
||||||
Message m = new Message();
|
Message m = new Message();
|
||||||
@ -488,4 +515,16 @@ class PeerConnectionOut implements Runnable
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** @since 0.8.2 */
|
||||||
|
void sendExtension(int id, byte[] bytes) {
|
||||||
|
Message m = new Message();
|
||||||
|
m.type = Message.EXTENSION;
|
||||||
|
m.piece = id;
|
||||||
|
m.data = bytes;
|
||||||
|
m.begin = 0;
|
||||||
|
m.length = bytes.length;
|
||||||
|
addMessage(m);
|
||||||
|
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
@ -20,14 +20,20 @@
|
|||||||
|
|
||||||
package org.klomp.snark;
|
package org.klomp.snark;
|
||||||
|
|
||||||
|
import java.io.ByteArrayInputStream;
|
||||||
|
import java.io.InputStream;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.Iterator;
|
import java.util.Iterator;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
|
|
||||||
import net.i2p.I2PAppContext;
|
import net.i2p.I2PAppContext;
|
||||||
import net.i2p.util.Log;
|
import net.i2p.util.Log;
|
||||||
|
|
||||||
class PeerState
|
import org.klomp.snark.bencode.BDecoder;
|
||||||
|
import org.klomp.snark.bencode.BEValue;
|
||||||
|
|
||||||
|
class PeerState implements DataLoader
|
||||||
{
|
{
|
||||||
private final Log _log = I2PAppContext.getGlobalContext().logManager().getLog(PeerState.class);
|
private final Log _log = I2PAppContext.getGlobalContext().logManager().getLog(PeerState.class);
|
||||||
final Peer peer;
|
final Peer peer;
|
||||||
@ -201,13 +207,28 @@ class PeerState
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (_log.shouldLog(Log.INFO))
|
||||||
|
_log.info("Queueing (" + piece + ", " + begin + ", "
|
||||||
|
+ length + ")" + " to " + peer);
|
||||||
|
|
||||||
|
// don't load the data into mem now, let PeerConnectionOut do it
|
||||||
|
out.sendPiece(piece, begin, length, this);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* This is the callback that PeerConnectionOut calls
|
||||||
|
*
|
||||||
|
* @return bytes or null for errors
|
||||||
|
* @since 0.8.2
|
||||||
|
*/
|
||||||
|
public byte[] loadData(int piece, int begin, int length) {
|
||||||
byte[] pieceBytes = listener.gotRequest(peer, piece, begin, length);
|
byte[] pieceBytes = listener.gotRequest(peer, piece, begin, length);
|
||||||
if (pieceBytes == null)
|
if (pieceBytes == null)
|
||||||
{
|
{
|
||||||
// XXX - Protocol error-> diconnect?
|
// XXX - Protocol error-> diconnect?
|
||||||
if (_log.shouldLog(Log.WARN))
|
if (_log.shouldLog(Log.WARN))
|
||||||
_log.warn("Got request for unknown piece: " + piece);
|
_log.warn("Got request for unknown piece: " + piece);
|
||||||
return;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
// More sanity checks
|
// More sanity checks
|
||||||
@ -219,13 +240,13 @@ class PeerState
|
|||||||
+ ", " + begin
|
+ ", " + begin
|
||||||
+ ", " + length
|
+ ", " + length
|
||||||
+ "' message from " + peer);
|
+ "' message from " + peer);
|
||||||
return;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (_log.shouldLog(Log.INFO))
|
if (_log.shouldLog(Log.INFO))
|
||||||
_log.info("Sending (" + piece + ", " + begin + ", "
|
_log.info("Sending (" + piece + ", " + begin + ", "
|
||||||
+ length + ")" + " to " + peer);
|
+ length + ")" + " to " + peer);
|
||||||
out.sendPiece(piece, begin, length, pieceBytes);
|
return pieceBytes;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -413,6 +434,24 @@ class PeerState
|
|||||||
out.cancelRequest(piece, begin, length);
|
out.cancelRequest(piece, begin, length);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** @since 0.8.2 */
|
||||||
|
void extensionMessage(int id, byte[] bs)
|
||||||
|
{
|
||||||
|
if (id == 0) {
|
||||||
|
InputStream is = new ByteArrayInputStream(bs);
|
||||||
|
try {
|
||||||
|
BDecoder dec = new BDecoder(is);
|
||||||
|
BEValue bev = dec.bdecodeMap();
|
||||||
|
Map map = bev.getMap();
|
||||||
|
if (_log.shouldLog(Log.DEBUG))
|
||||||
|
_log.debug("Got extension handshake message " + bev.toString());
|
||||||
|
} catch (Exception e) {}
|
||||||
|
} else {
|
||||||
|
if (_log.shouldLog(Log.DEBUG))
|
||||||
|
_log.debug("Got extended message type: " + id + " length: " + bs.length);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
void unknownMessage(int type, byte[] bs)
|
void unknownMessage(int type, byte[] bs)
|
||||||
{
|
{
|
||||||
if (_log.shouldLog(Log.WARN))
|
if (_log.shouldLog(Log.WARN))
|
||||||
|
Reference in New Issue
Block a user