Compare commits

..

37 Commits

Author SHA1 Message Date
Zlatin Balevsky
c79e8712d0 correctly determine if uploader has requested piece 2019-06-21 20:36:33 +01:00
Zlatin Balevsky
ed12d78a48 clear pieces on cancel 2019-06-21 17:22:55 +01:00
Zlatin Balevsky
d27872cc8b investigate StringIndexOutOfBounds 2019-06-21 16:29:52 +01:00
Zlatin Balevsky
f794c39760 personas not destinations 2019-06-21 16:15:35 +01:00
Zlatin Balevsky
2be9c425f7 compute which pieces are requested 2019-06-21 16:09:57 +01:00
Zlatin Balevsky
ab5fea9216 416 if piece not downloaded 2019-06-21 16:03:20 +01:00
Zlatin Balevsky
d1c8328080 do not send alts if there aren't any 2019-06-21 15:39:00 +01:00
Zlatin Balevsky
89e761f53b write personas on the wire part1 2019-06-21 15:26:18 +01:00
Zlatin Balevsky
40410eba63 fix constructor 2019-06-21 14:57:53 +01:00
Zlatin Balevsky
85466a8e80 fix npe 2019-06-21 14:45:14 +01:00
Zlatin Balevsky
c210af7870 source partial uploads from incompletes file 2019-06-21 14:39:20 +01:00
Zlatin Balevsky
38ff49d28f downloaders get pieces from mesh manager 2019-06-21 14:17:10 +01:00
Zlatin Balevsky
710f9f52a8 send X-Have and X-Alts from uploader 2019-06-21 13:58:21 +01:00
Zlatin Balevsky
1b6eda5a40 skeleton of mesh manager 2019-06-21 13:34:00 +01:00
Zlatin Balevsky
1ee9ccf098 parse X-Have on uploader side 2019-06-21 12:55:25 +01:00
Zlatin Balevsky
0f07562de3 pass new sources to active downloaders 2019-06-21 12:39:16 +01:00
Zlatin Balevsky
6eb1aa07f5 key downloaders by infohash 2019-06-21 12:29:32 +01:00
Zlatin Balevsky
05b02834af parse X-Alt 2019-06-21 12:25:04 +01:00
Zlatin Balevsky
56125f6df8 refactor X-Have decoding logic 2019-06-21 09:32:10 +01:00
Zlatin Balevsky
8f9996848b send X-Have from downloader too 2019-06-21 09:25:28 +01:00
Zlatin Balevsky
dd655ed60f test for re-requesting available pieces 2019-06-21 09:12:42 +01:00
Zlatin Balevsky
807ab22f8e test parsing of X-Have 2019-06-21 06:43:48 +01:00
Zlatin Balevsky
a26ad229ee more tests 2019-06-21 05:56:42 +01:00
Zlatin Balevsky
5504dd2251 tighten conditions 2019-06-21 05:45:11 +01:00
Zlatin Balevsky
f9777d29f4 get existing tests to pass 2019-06-21 05:41:49 +01:00
Zlatin Balevsky
b23226e8c6 wip on parsing X-Have from uploader 2019-06-21 05:30:56 +01:00
Zlatin Balevsky
1249ad29e0 claim pieces from list of available pieces 2019-06-21 04:42:02 +01:00
Zlatin Balevsky
7bb5e5b632 Release 0.3.4 2019-06-20 21:07:50 +01:00
Zlatin Balevsky
b2e43f9765 update split pattern and add unit test 2019-06-20 21:06:39 +01:00
Zlatin Balevsky
2aa73c203a Release 0.3.3 2019-06-20 18:08:02 +01:00
Zlatin Balevsky
18d2b56563 fix indexing 2019-06-20 17:57:36 +01:00
Zlatin Balevsky
a455b4ad6e redirect exceptions in result sender to log 2019-06-20 17:22:59 +01:00
Zlatin Balevsky
761b683a81 Release 0.3.2 2019-06-20 16:04:46 +01:00
Zlatin Balevsky
1d41bcd825 prevent empty tokens in search index 2019-06-20 16:02:48 +01:00
Zlatin Balevsky
f1ac038b55 update split pattern 2019-06-20 15:47:00 +01:00
Zlatin Balevsky
396c636e42 prevent empty search terms 2019-06-20 15:29:27 +01:00
Zlatin Balevsky
e32c858e90 update README with quick FAQ 2019-06-20 14:18:37 +01:00
25 changed files with 681 additions and 115 deletions

View File

@@ -31,3 +31,19 @@ The first time you run MuWire it will ask you to select a nickname. This nickna
### Known bugs and limitations
* Many UI features you would expect are not there yet
### Quick FAQ
* why is MuWire slow ?
- too few sources you're downloading from
- you can increse the number of tunnels by using more tunnels via Options->I2P Inbound/Outbound Quantity
the default is 4 and you could raise it and even can go up as high as 16 ( Caution !!!!)
* my search is not returning (enough) results !
- search is keyword or hash based
- keywords and hash(es) are NOT regexed or wildcarded so they have to be complete
so searching for 'musi' will not return results with 'music' - you have to search for 'music'
- ALL keywords have to match
- only use <SPACE> for keyword separation

View File

@@ -35,7 +35,7 @@ class Cli {
Core core
try {
core = new Core(props, home, "0.3.1")
core = new Core(props, home, "0.3.4")
} catch (Exception bad) {
bad.printStackTrace(System.out)
println "Failed to initialize core, exiting"

View File

@@ -53,7 +53,7 @@ class CliDownloader {
Core core
try {
core = new Core(props, home, "0.3.1")
core = new Core(props, home, "0.3.4")
} catch (Exception bad) {
bad.printStackTrace(System.out)
println "Failed to initialize core, exiting"

View File

@@ -11,5 +11,5 @@ class Constants {
public static final float DOWNLOAD_SEQUENTIAL_RATIO = 0.8f
public static final String SPLIT_PATTERN = "[\\.,_-]"
public static final String SPLIT_PATTERN = "[\\+\\-,\\.:;\\(\\)=_/\\\\\\!\\\"\\\'\\\$%\\|]"
}

View File

@@ -13,6 +13,7 @@ import com.muwire.core.connection.I2PConnector
import com.muwire.core.connection.LeafConnectionManager
import com.muwire.core.connection.UltrapeerConnectionManager
import com.muwire.core.download.DownloadManager
import com.muwire.core.download.SourceDiscoveredEvent
import com.muwire.core.download.UIDownloadCancelledEvent
import com.muwire.core.download.UIDownloadEvent
import com.muwire.core.files.FileDownloadedEvent
@@ -28,6 +29,7 @@ import com.muwire.core.files.DirectoryWatcher
import com.muwire.core.hostcache.CacheClient
import com.muwire.core.hostcache.HostCache
import com.muwire.core.hostcache.HostDiscoveredEvent
import com.muwire.core.mesh.MeshManager
import com.muwire.core.search.QueryEvent
import com.muwire.core.search.ResultsEvent
import com.muwire.core.search.ResultsSender
@@ -165,6 +167,10 @@ public class Core {
eventBus.register(FileDownloadedEvent.class, fileManager)
eventBus.register(FileUnsharedEvent.class, fileManager)
eventBus.register(SearchEvent.class, fileManager)
log.info("initializing mesh manager")
MeshManager meshManager = new MeshManager(fileManager)
eventBus.register(SourceDiscoveredEvent.class, meshManager)
log.info "initializing persistence service"
persisterService = new PersisterService(new File(home, "files.json"), eventBus, 15000, fileManager)
@@ -203,14 +209,15 @@ public class Core {
eventBus.register(ResultsEvent.class, searchManager)
log.info("initializing download manager")
downloadManager = new DownloadManager(eventBus, i2pConnector, home, me)
downloadManager = new DownloadManager(eventBus, trustService, meshManager, props, i2pConnector, home, me)
eventBus.register(UIDownloadEvent.class, downloadManager)
eventBus.register(UILoadedEvent.class, downloadManager)
eventBus.register(FileDownloadedEvent.class, downloadManager)
eventBus.register(UIDownloadCancelledEvent.class, downloadManager)
eventBus.register(SourceDiscoveredEvent.class, downloadManager)
log.info("initializing upload manager")
UploadManager uploadManager = new UploadManager(eventBus, fileManager)
UploadManager uploadManager = new UploadManager(eventBus, fileManager, meshManager, downloadManager)
log.info("initializing connection establisher")
connectionEstablisher = new ConnectionEstablisher(eventBus, i2pConnector, props, connectionManager, hostCache)
@@ -284,7 +291,7 @@ public class Core {
}
}
Core core = new Core(props, home, "0.3.1")
Core core = new Core(props, home, "0.3.4")
core.startServices()
// ... at the end, sleep or execute script

View File

@@ -3,6 +3,10 @@ package com.muwire.core.download
import com.muwire.core.connection.I2PConnector
import com.muwire.core.files.FileDownloadedEvent
import com.muwire.core.files.FileHasher
import com.muwire.core.mesh.Mesh
import com.muwire.core.mesh.MeshManager
import com.muwire.core.trust.TrustLevel
import com.muwire.core.trust.TrustService
import com.muwire.core.util.DataUtil
import groovy.json.JsonBuilder
@@ -14,24 +18,33 @@ import net.i2p.util.ConcurrentHashSet
import com.muwire.core.EventBus
import com.muwire.core.InfoHash
import com.muwire.core.MuWireSettings
import com.muwire.core.Persona
import com.muwire.core.UILoadedEvent
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.Executor
import java.util.concurrent.Executors
public class DownloadManager {
private final EventBus eventBus
private final TrustService trustService
private final MeshManager meshManager
private final MuWireSettings muSettings
private final I2PConnector connector
private final Executor executor
private final File incompletes, home
private final Persona me
private final Set<Downloader> downloaders = new ConcurrentHashSet<>()
private final Map<InfoHash, Downloader> downloaders = new ConcurrentHashMap<>()
public DownloadManager(EventBus eventBus, I2PConnector connector, File home, Persona me) {
public DownloadManager(EventBus eventBus, TrustService trustService, MeshManager meshManager, MuWireSettings muSettings,
I2PConnector connector, File home, Persona me) {
this.eventBus = eventBus
this.trustService = trustService
this.meshManager = meshManager
this.muSettings = muSettings
this.connector = connector
this.incompletes = new File(home,"incompletes")
this.home = home
@@ -61,17 +74,19 @@ public class DownloadManager {
destinations.addAll(e.sources)
destinations.remove(me.destination)
Pieces pieces = getPieces(infohash, size, pieceSize)
def downloader = new Downloader(eventBus, this, me, e.target, size,
infohash, pieceSize, connector, destinations,
incompletes)
downloaders.add(downloader)
incompletes, pieces)
downloaders.put(infohash, downloader)
persistDownloaders()
executor.execute({downloader.download()} as Runnable)
eventBus.publish(new DownloadStartedEvent(downloader : downloader))
}
public void onUIDownloadCancelledEvent(UIDownloadCancelledEvent e) {
downloaders.remove(e.downloader)
downloaders.remove(e.downloader.infoHash)
persistDownloaders()
}
@@ -99,23 +114,50 @@ public class DownloadManager {
byte [] root = Base64.decode(json.hashRoot)
infoHash = new InfoHash(root)
}
Pieces pieces = getPieces(infoHash, (long)json.length, json.pieceSizePow2)
def downloader = new Downloader(eventBus, this, me, file, (long)json.length,
infoHash, json.pieceSizePow2, connector, destinations, incompletes)
downloaders.add(downloader)
infoHash, json.pieceSizePow2, connector, destinations, incompletes, pieces)
downloaders.put(infoHash, downloader)
downloader.download()
eventBus.publish(new DownloadStartedEvent(downloader : downloader))
}
}
private Pieces getPieces(InfoHash infoHash, long length, int pieceSizePow2) {
int pieceSize = 0x1 << pieceSizePow2
int nPieces = (int)(length / pieceSize)
if (length % pieceSize != 0)
nPieces++
Mesh mesh = meshManager.getOrCreate(infoHash, nPieces)
mesh.pieces
}
void onSourceDiscoveredEvent(SourceDiscoveredEvent e) {
Downloader downloader = downloaders.get(e.infoHash)
if (downloader == null)
return
boolean ok = false
switch(trustService.getLevel(e.source.destination)) {
case TrustLevel.TRUSTED: ok = true; break
case TrustLevel.NEUTRAL: ok = muSettings.allowUntrusted; break
case TrustLevel.DISTRUSTED: ok = false; break
}
if (ok)
downloader.addSource(e.source.destination)
}
void onFileDownloadedEvent(FileDownloadedEvent e) {
downloaders.remove(e.downloader)
downloaders.remove(e.downloader.infoHash)
persistDownloaders()
}
private void persistDownloaders() {
File downloadsFile = new File(home,"downloads.json")
downloadsFile.withPrintWriter { writer ->
downloaders.each { downloader ->
downloaders.values().each { downloader ->
if (!downloader.cancelled) {
def json = [:]
json.file = Base64.encode(DataUtil.encodei18nString(downloader.file.getAbsolutePath()))
@@ -139,7 +181,7 @@ public class DownloadManager {
}
public void shutdown() {
downloaders.each { it.stop() }
downloaders.values().each { it.stop() }
Downloader.executorService.shutdownNow()
}
}

View File

@@ -3,8 +3,12 @@ package com.muwire.core.download;
import net.i2p.data.Base64
import com.muwire.core.Constants
import com.muwire.core.EventBus
import com.muwire.core.InfoHash
import com.muwire.core.Persona
import com.muwire.core.connection.Endpoint
import com.muwire.core.util.DataUtil
import static com.muwire.core.util.DataUtil.readTillRN
import groovy.util.logging.Log
@@ -23,6 +27,7 @@ class DownloadSession {
private static int SAMPLES = 10
private final EventBus eventBus
private final String meB64
private final Pieces pieces
private final InfoHash infoHash
@@ -30,6 +35,7 @@ class DownloadSession {
private final File file
private final int pieceSize
private final long fileLength
private final Set<Integer> available
private final MessageDigest digest
private final LinkedList<Long> timestamps = new LinkedList<>()
@@ -37,8 +43,9 @@ class DownloadSession {
private ByteBuffer mapped
DownloadSession(String meB64, Pieces pieces, InfoHash infoHash, Endpoint endpoint, File file,
int pieceSize, long fileLength) {
DownloadSession(EventBus eventBus, String meB64, Pieces pieces, InfoHash infoHash, Endpoint endpoint, File file,
int pieceSize, long fileLength, Set<Integer> available) {
this.eventBus = eventBus
this.meB64 = meB64
this.pieces = pieces
this.endpoint = endpoint
@@ -46,6 +53,7 @@ class DownloadSession {
this.file = file
this.pieceSize = pieceSize
this.fileLength = fileLength
this.available = available
try {
digest = MessageDigest.getInstance("SHA-256")
} catch (NoSuchAlgorithmException impossible) {
@@ -63,7 +71,11 @@ class DownloadSession {
OutputStream os = endpoint.getOutputStream()
InputStream is = endpoint.getInputStream()
int piece = pieces.claim()
int piece
if (available.isEmpty())
piece = pieces.claim()
else
piece = pieces.claim(available)
if (piece == -1)
return false
boolean unclaim = true
@@ -79,45 +91,79 @@ class DownloadSession {
try {
os.write("GET $root\r\n".getBytes(StandardCharsets.US_ASCII))
os.write("Range: $start-$end\r\n".getBytes(StandardCharsets.US_ASCII))
os.write("X-Persona: $meB64\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
os.write("X-Persona: $meB64\r\n".getBytes(StandardCharsets.US_ASCII))
String xHave = DataUtil.encodeXHave(pieces.getDownloaded(), pieces.nPieces)
os.write("X-Have: $xHave\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
os.flush()
String code = readTillRN(is)
if (code.startsWith("404 ")) {
String codeString = readTillRN(is)
int space = codeString.indexOf(' ')
if (space > 0)
codeString = codeString.substring(0, space)
int code = Integer.parseInt(codeString.trim())
if (code == 404) {
log.warning("file not found")
endpoint.close()
return false
}
if (code.startsWith("416 ")) {
log.warning("range $start-$end cannot be satisfied")
return // leave endpoint open
}
if (!code.startsWith("200 ")) {
if (!(code == 200 || code == 416)) {
log.warning("unknown code $code")
endpoint.close()
return false
}
// parse all headers
Set<String> headers = new HashSet<>()
Map<String,String> headers = new HashMap<>()
String header
while((header = readTillRN(is)) != "" && headers.size() < Constants.MAX_HEADERS)
headers.add(header)
long receivedStart = -1
long receivedEnd = -1
for (String receivedHeader : headers) {
def group = (receivedHeader =~ /^Content-Range: (\d+)-(\d+)$/)
if (group.size() != 1) {
log.info("ignoring header $receivedHeader")
continue
}
receivedStart = Long.parseLong(group[0][1])
receivedEnd = Long.parseLong(group[0][2])
while((header = readTillRN(is)) != "" && headers.size() < Constants.MAX_HEADERS) {
int colon = header.indexOf(':')
if (colon == -1 || colon == header.length() - 1)
throw new IOException("invalid header $header")
String key = header.substring(0, colon)
String value = header.substring(colon + 1)
headers[key] = value.trim()
}
// prase X-Alt if present
if (headers.containsKey("X-Alt")) {
headers["X-Alt"].split(",").each {
if (it.length() > 0) {
byte [] raw = Base64.decode(it)
Persona source = new Persona(new ByteArrayInputStream(raw))
eventBus.publish(new SourceDiscoveredEvent(infoHash : infoHash, source : source))
}
}
}
// parse X-Have if present
if (headers.containsKey("X-Have")) {
DataUtil.decodeXHave(headers["X-Have"]).each {
available.add(it)
}
if (!available.contains(piece))
return true // try again next time
} else {
if (code != 200)
throw new IOException("Code $code but no X-Have")
available.clear()
}
if (code != 200)
return true
String range = headers["Content-Range"]
if (range == null)
throw new IOException("Code 200 but no Content-Range")
def group = (range =~ /^(\d+)-(\d+)$/)
if (group.size() != 1)
throw new IOException("invalid Content-Range header $range")
long receivedStart = Long.parseLong(group[0][1])
long receivedEnd = Long.parseLong(group[0][2])
if (receivedStart != start || receivedEnd != end) {
log.warning("We don't support mismatching ranges yet")
endpoint.close()

View File

@@ -60,7 +60,7 @@ public class Downloader {
public Downloader(EventBus eventBus, DownloadManager downloadManager,
Persona me, File file, long length, InfoHash infoHash,
int pieceSizePow2, I2PConnector connector, Set<Destination> destinations,
File incompletes) {
File incompletes, Pieces pieces) {
this.eventBus = eventBus
this.me = me
this.downloadManager = downloadManager
@@ -73,15 +73,8 @@ public class Downloader {
this.incompleteFile = new File(incompletes, file.getName()+".part")
this.pieceSizePow2 = pieceSizePow2
this.pieceSize = 1 << pieceSizePow2
int nPieces
if (length % pieceSize == 0)
nPieces = length / pieceSize
else
nPieces = length / pieceSize + 1
this.nPieces = nPieces
pieces = new Pieces(nPieces, Constants.DOWNLOAD_SEQUENTIAL_RATIO)
this.pieces = pieces
this.nPieces = pieces.nPieces
}
public synchronized InfoHash getInfoHash() {
@@ -187,6 +180,7 @@ public class Downloader {
piecesFile.delete()
}
incompleteFile.delete()
pieces.clearAll()
}
void stop() {
@@ -221,12 +215,21 @@ public class Downloader {
}
}
void addSource(Destination d) {
if (activeWorkers.containsKey(d))
return
DownloadWorker newWorker = new DownloadWorker(d)
activeWorkers.put(d, newWorker)
executorService.submit(newWorker)
}
class DownloadWorker implements Runnable {
private final Destination destination
private volatile WorkerState currentState
private volatile Thread downloadThread
private Endpoint endpoint
private volatile DownloadSession currentSession
private final Set<Integer> available = new HashSet<>()
DownloadWorker(Destination destination) {
this.destination = destination
@@ -247,7 +250,8 @@ public class Downloader {
currentState = WorkerState.DOWNLOADING
boolean requestPerformed
while(!pieces.isComplete()) {
currentSession = new DownloadSession(me.toBase64(), pieces, getInfoHash(), endpoint, incompleteFile, pieceSize, length)
currentSession = new DownloadSession(eventBus, me.toBase64(), pieces, getInfoHash(),
endpoint, incompleteFile, pieceSize, length, available)
requestPerformed = currentSession.request()
if (!requestPerformed)
break

View File

@@ -38,6 +38,18 @@ class Pieces {
}
}
synchronized int claim(Set<Integer> available) {
for (int i = claimed.nextSetBit(0); i >= 0; i = claimed.nextSetBit(i+1))
available.remove(i)
if (available.isEmpty())
return -1
List<Integer> toList = available.toList()
Collections.shuffle(toList)
int rv = toList[0]
claimed.set(rv)
rv
}
synchronized def getDownloaded() {
def rv = []
for (int i = done.nextSetBit(0); i >= 0; i = done.nextSetBit(i+1)) {
@@ -62,4 +74,13 @@ class Pieces {
synchronized int donePieces() {
done.cardinality()
}
synchronized boolean isDownloaded(int piece) {
done.get(piece)
}
synchronized void clearAll() {
done.clear()
claimed.clear()
}
}

View File

@@ -0,0 +1,10 @@
package com.muwire.core.download
import com.muwire.core.Event
import com.muwire.core.InfoHash
import com.muwire.core.Persona
class SourceDiscoveredEvent extends Event {
InfoHash infoHash
Persona source
}

View File

@@ -0,0 +1,28 @@
package com.muwire.core.mesh
import com.muwire.core.InfoHash
import com.muwire.core.Persona
import com.muwire.core.download.Pieces
import net.i2p.data.Destination
import net.i2p.util.ConcurrentHashSet
class Mesh {
private final InfoHash infoHash
private final Set<Persona> sources = new ConcurrentHashSet<>()
private final Pieces pieces
Mesh(InfoHash infoHash, Pieces pieces) {
this.infoHash = infoHash
this.pieces = pieces
}
Set<Persona> getRandom(int n, Persona exclude) {
List<Persona> tmp = new ArrayList<>(sources)
tmp.remove(exclude)
Collections.shuffle(tmp)
if (tmp.size() < n)
return tmp
tmp[0..n-1]
}
}

View File

@@ -0,0 +1,43 @@
package com.muwire.core.mesh
import com.muwire.core.Constants
import com.muwire.core.InfoHash
import com.muwire.core.download.Pieces
import com.muwire.core.download.SourceDiscoveredEvent
import com.muwire.core.files.FileManager
class MeshManager {
private final Map<InfoHash, Mesh> meshes = Collections.synchronizedMap(new HashMap<>())
private final FileManager fileManager
MeshManager(FileManager fileManager) {
this.fileManager = fileManager
}
Mesh get(InfoHash infoHash) {
meshes.get(infoHash)
}
Mesh getOrCreate(InfoHash infoHash, int nPieces) {
synchronized(meshes) {
if (meshes.containsKey(infoHash))
return meshes.get(infoHash)
Pieces pieces = new Pieces(nPieces, Constants.DOWNLOAD_SEQUENTIAL_RATIO)
if (fileManager.rootToFiles.containsKey(infoHash)) {
for (int i = 0; i < nPieces; i++)
pieces.markDownloaded(i)
}
Mesh rv = new Mesh(infoHash, pieces)
meshes.put(infoHash, rv)
return rv
}
}
void onSourceDiscoveredEvent(SourceDiscoveredEvent e) {
Mesh mesh = meshes.get(e.infoHash)
if (mesh == null)
return
mesh.sources.add(e.source)
}
}

View File

@@ -11,6 +11,7 @@ import java.util.concurrent.Executor
import java.util.concurrent.Executors
import java.util.concurrent.ThreadFactory
import java.util.concurrent.atomic.AtomicInteger
import java.util.logging.Level
import java.util.stream.Collectors
import com.muwire.core.DownloadedFile
@@ -83,50 +84,54 @@ class ResultsSender {
@Override
public void run() {
byte [] tmp = new byte[InfoHash.SIZE]
JsonOutput jsonOutput = new JsonOutput()
Endpoint endpoint = null;
try {
endpoint = connector.connect(target)
DataOutputStream os = new DataOutputStream(endpoint.getOutputStream())
os.write("POST $uuid\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
me.write(os)
os.writeShort((short)results.length)
results.each {
byte [] name = it.getFile().getName().getBytes(StandardCharsets.UTF_8)
def baos = new ByteArrayOutputStream()
def daos = new DataOutputStream(baos)
daos.writeShort((short) name.length)
daos.write(name)
daos.flush()
String encodedName = Base64.encode(baos.toByteArray())
def obj = [:]
obj.type = "Result"
obj.version = oobInfohash ? 2 : 1
obj.name = encodedName
obj.infohash = Base64.encode(it.getInfoHash().getRoot())
obj.size = it.getFile().length()
obj.pieceSize = it.getPieceSize()
if (!oobInfohash) {
byte [] hashList = it.getInfoHash().getHashList()
def hashListB64 = []
for (int i = 0; i < hashList.length / InfoHash.SIZE; i++) {
System.arraycopy(hashList, InfoHash.SIZE * i, tmp, 0, InfoHash.SIZE)
hashListB64 << Base64.encode(tmp)
byte [] tmp = new byte[InfoHash.SIZE]
JsonOutput jsonOutput = new JsonOutput()
Endpoint endpoint = null;
try {
endpoint = connector.connect(target)
DataOutputStream os = new DataOutputStream(endpoint.getOutputStream())
os.write("POST $uuid\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
me.write(os)
os.writeShort((short)results.length)
results.each {
byte [] name = it.getFile().getName().getBytes(StandardCharsets.UTF_8)
def baos = new ByteArrayOutputStream()
def daos = new DataOutputStream(baos)
daos.writeShort((short) name.length)
daos.write(name)
daos.flush()
String encodedName = Base64.encode(baos.toByteArray())
def obj = [:]
obj.type = "Result"
obj.version = oobInfohash ? 2 : 1
obj.name = encodedName
obj.infohash = Base64.encode(it.getInfoHash().getRoot())
obj.size = it.getFile().length()
obj.pieceSize = it.getPieceSize()
if (!oobInfohash) {
byte [] hashList = it.getInfoHash().getHashList()
def hashListB64 = []
for (int i = 0; i < hashList.length / InfoHash.SIZE; i++) {
System.arraycopy(hashList, InfoHash.SIZE * i, tmp, 0, InfoHash.SIZE)
hashListB64 << Base64.encode(tmp)
}
obj.hashList = hashListB64
}
obj.hashList = hashListB64
if (it instanceof DownloadedFile)
obj.sources = it.sources.stream().map({dest -> dest.toBase64()}).collect(Collectors.toSet())
def json = jsonOutput.toJson(obj)
os.writeShort((short)json.length())
os.write(json.getBytes(StandardCharsets.US_ASCII))
}
if (it instanceof DownloadedFile)
obj.sources = it.sources.stream().map({dest -> dest.toBase64()}).collect(Collectors.toSet())
def json = jsonOutput.toJson(obj)
os.writeShort((short)json.length())
os.write(json.getBytes(StandardCharsets.US_ASCII))
os.flush()
} finally {
endpoint?.close()
}
os.flush()
} finally {
endpoint?.close()
} catch (Exception e) {
log.log(Level.WARNING, "problem sending results",e)
}
}
}

View File

@@ -33,7 +33,10 @@ class SearchIndex {
private static String[] split(String source) {
source = source.replaceAll(Constants.SPLIT_PATTERN, " ").toLowerCase()
source.split(" ")
String [] split = source.split(" ")
def rv = []
split.each { if (it.length() > 0) rv << it }
rv.toArray(new String[0])
}
String[] search(List<String> terms) {

View File

@@ -2,4 +2,5 @@ package com.muwire.core.upload
class ContentRequest extends Request {
Range range
boolean have
}

View File

@@ -5,33 +5,57 @@ import java.nio.channels.FileChannel
import java.nio.charset.StandardCharsets
import java.nio.file.Files
import java.nio.file.StandardOpenOption
import java.util.stream.Collectors
import com.muwire.core.Persona
import com.muwire.core.connection.Endpoint
import com.muwire.core.mesh.Mesh
import com.muwire.core.util.DataUtil
import net.i2p.data.Destination
class ContentUploader extends Uploader {
private final File file
private final ContentRequest request
private final Mesh mesh
private final int pieceSize
ContentUploader(File file, ContentRequest request, Endpoint endpoint) {
ContentUploader(File file, ContentRequest request, Endpoint endpoint, Mesh mesh, int pieceSize) {
super(endpoint)
this.file = file
this.request = request
this.mesh = mesh
this.pieceSize = pieceSize
}
@Override
void respond() {
OutputStream os = endpoint.getOutputStream()
Range range = request.getRange()
if (range.start >= file.length() || range.end >= file.length()) {
os.write("416 Range Not Satisfiable\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
boolean satisfiable = true
final long length = file.length()
if (range.start >= length || range.end >= length)
satisfiable = false
if (satisfiable) {
int startPiece = range.start / (0x1 << pieceSize)
int endPiece = range.end / (0x1 << pieceSize)
for (int i = startPiece; i <= endPiece; i++)
satisfiable &= mesh.pieces.isDownloaded(i)
}
if (!satisfiable) {
os.write("416 Range Not Satisfiable\r\n".getBytes(StandardCharsets.US_ASCII))
writeMesh(request.downloader)
os.write("\r\n".getBytes(StandardCharsets.US_ASCII))
os.flush()
return
}
os.write("200 OK\r\n".getBytes(StandardCharsets.US_ASCII))
os.write("Content-Range: $range.start-$range.end\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
os.write("Content-Range: $range.start-$range.end\r\n".getBytes(StandardCharsets.US_ASCII))
writeMesh(request.downloader)
os.write("\r\n".getBytes(StandardCharsets.US_ASCII))
FileChannel channel
try {
channel = Files.newByteChannel(file.toPath(), EnumSet.of(StandardOpenOption.READ))
@@ -50,6 +74,17 @@ class ContentUploader extends Uploader {
endpoint.getOutputStream().flush()
}
}
private void writeMesh(Persona toExclude) {
String xHave = DataUtil.encodeXHave(mesh.pieces.getDownloaded(), mesh.pieces.nPieces)
endpoint.getOutputStream().write("X-Have: $xHave\r\n".getBytes(StandardCharsets.US_ASCII))
Set<Persona> sources = mesh.getRandom(3, toExclude)
if (!sources.isEmpty()) {
String xAlts = sources.stream().map({ it.toBase64() }).collect(Collectors.joining(","))
endpoint.getOutputStream().write("X-Alt: $xAlts\r\n".getBytes(StandardCharsets.US_ASCII))
}
}
@Override
public String getName() {

View File

@@ -5,6 +5,7 @@ import java.nio.charset.StandardCharsets
import com.muwire.core.Constants
import com.muwire.core.InfoHash
import com.muwire.core.Persona
import com.muwire.core.util.DataUtil
import groovy.util.logging.Log
import net.i2p.data.Base64
@@ -48,8 +49,14 @@ class Request {
def decoded = Base64.decode(encoded)
downloader = new Persona(new ByteArrayInputStream(decoded))
}
boolean have = false
if (headers.containsKey("X-Have")) {
def encoded = headers["X-Have"].trim()
have = DataUtil.decodeXHave(encoded).size() > 0
}
new ContentRequest( infoHash : infoHash, range : new Range(start, end),
headers : headers, downloader : downloader)
headers : headers, downloader : downloader, have : have)
}
static Request parseHashListRequest(InfoHash infoHash, InputStream is) throws IOException {

View File

@@ -6,7 +6,12 @@ import com.muwire.core.EventBus
import com.muwire.core.InfoHash
import com.muwire.core.SharedFile
import com.muwire.core.connection.Endpoint
import com.muwire.core.download.DownloadManager
import com.muwire.core.download.Downloader
import com.muwire.core.download.SourceDiscoveredEvent
import com.muwire.core.files.FileManager
import com.muwire.core.mesh.Mesh
import com.muwire.core.mesh.MeshManager
import groovy.util.logging.Log
import net.i2p.data.Base64
@@ -15,12 +20,17 @@ import net.i2p.data.Base64
public class UploadManager {
private final EventBus eventBus
private final FileManager fileManager
private final MeshManager meshManager
private final DownloadManager downloadManager
public UploadManager() {}
public UploadManager(EventBus eventBus, FileManager fileManager) {
public UploadManager(EventBus eventBus, FileManager fileManager,
MeshManager meshManager, DownloadManager downloadManager) {
this.eventBus = eventBus
this.fileManager = fileManager
this.meshManager = meshManager
this.downloadManager = downloadManager
}
public void processGET(Endpoint e) throws IOException {
@@ -44,8 +54,10 @@ public class UploadManager {
log.info("Responding to upload request for root $infoHashString")
byte [] infoHashRoot = Base64.decode(infoHashString)
InfoHash infoHash = new InfoHash(infoHashRoot)
Set<SharedFile> sharedFiles = fileManager.getSharedFiles(infoHashRoot)
if (sharedFiles == null || sharedFiles.isEmpty()) {
Downloader downloader = downloadManager.downloaders.get(infoHash)
if (downloader == null && (sharedFiles == null || sharedFiles.isEmpty())) {
log.info "file not found"
e.getOutputStream().write("404 File Not Found\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
e.getOutputStream().flush()
@@ -61,13 +73,31 @@ public class UploadManager {
return
}
Request request = Request.parseContentRequest(new InfoHash(infoHashRoot), e.getInputStream())
ContentRequest request = Request.parseContentRequest(infoHash, e.getInputStream())
if (request.downloader != null && request.downloader.destination != e.destination) {
log.info("Downloader persona doesn't match their destination")
e.close()
return
}
Uploader uploader = new ContentUploader(sharedFiles.iterator().next().file, request, e)
if (request.have)
eventBus.publish(new SourceDiscoveredEvent(infoHash : request.infoHash, source : request.downloader))
Mesh mesh
File file
int pieceSize
if (downloader != null) {
mesh = meshManager.get(infoHash)
file = downloader.incompleteFile
pieceSize = downloader.pieceSizePow2
} else {
SharedFile sharedFile = sharedFiles.iterator().next();
mesh = meshManager.getOrCreate(request.infoHash, sharedFile.NPieces)
file = sharedFile.file
pieceSize = sharedFile.pieceSize
}
Uploader uploader = new ContentUploader(file, request, e, mesh, pieceSize)
eventBus.publish(new UploadEvent(uploader : uploader))
try {
uploader.respond()
@@ -85,8 +115,10 @@ public class UploadManager {
log.info("Responding to hashlist request for root $infoHashString")
byte [] infoHashRoot = Base64.decode(infoHashString)
InfoHash infoHash = new InfoHash(infoHashRoot)
Downloader downloader = downloadManager.downloaders.get(infoHash)
Set<SharedFile> sharedFiles = fileManager.getSharedFiles(infoHashRoot)
if (sharedFiles == null || sharedFiles.isEmpty()) {
if (downloader == null && (sharedFiles == null || sharedFiles.isEmpty())) {
log.info "file not found"
e.getOutputStream().write("404 File Not Found\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
e.getOutputStream().flush()
@@ -102,13 +134,30 @@ public class UploadManager {
return
}
Request request = Request.parseHashListRequest(new InfoHash(infoHashRoot), e.getInputStream())
Request request = Request.parseHashListRequest(infoHash, e.getInputStream())
if (request.downloader != null && request.downloader.destination != e.destination) {
log.info("Downloader persona doesn't match their destination")
e.close()
return
}
Uploader uploader = new HashListUploader(e, sharedFiles.iterator().next().infoHash, request)
InfoHash fullInfoHash
if (downloader == null) {
fullInfoHash = sharedFiles.iterator().next().infoHash
} else {
byte [] hashList = downloader.getInfoHash().getHashList()
if (hashList != null && hashList.length > 0)
fullInfoHash = downloader.getInfoHash()
else {
log.info("infohash not found in downloader")
e.getOutputStream().write("404 File Not Found\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
e.getOutputStream().flush()
e.close()
return
}
}
Uploader uploader = new HashListUploader(e, fullInfoHash, request)
eventBus.publish(new UploadEvent(uploader : uploader))
try {
uploader.respond()
@@ -130,8 +179,10 @@ public class UploadManager {
log.info("Responding to upload request for root $infoHashString")
infoHashRoot = Base64.decode(infoHashString)
infoHash = new InfoHash(infoHashRoot)
sharedFiles = fileManager.getSharedFiles(infoHashRoot)
if (sharedFiles == null || sharedFiles.isEmpty()) {
downloader = downloadManager.downloaders.get(infoHash)
if (downloader == null && (sharedFiles == null || sharedFiles.isEmpty())) {
log.info "file not found"
e.getOutputStream().write("404 File Not Found\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
e.getOutputStream().flush()
@@ -153,7 +204,25 @@ public class UploadManager {
e.close()
return
}
uploader = new ContentUploader(sharedFiles.iterator().next().file, request, e)
if (request.have)
eventBus.publish(new SourceDiscoveredEvent(infoHash : request.infoHash, source : request.downloader))
Mesh mesh
File file
int pieceSize
if (downloader != null) {
mesh = meshManager.get(infoHash)
file = downloader.incompleteFile
pieceSize = downloader.pieceSizePow2
} else {
SharedFile sharedFile = sharedFiles.iterator().next();
mesh = meshManager.getOrCreate(request.infoHash, sharedFile.NPieces)
file = sharedFile.file
pieceSize = sharedFile.pieceSize
}
uploader = new ContentUploader(file, request, e, mesh, pieceSize)
eventBus.publish(new UploadEvent(uploader : uploader))
try {
uploader.respond()

View File

@@ -4,6 +4,8 @@ import java.nio.charset.StandardCharsets
import com.muwire.core.Constants
import net.i2p.data.Base64
class DataUtil {
private final static int MAX_SHORT = (0x1 << 16) - 1
@@ -79,4 +81,32 @@ class DataUtil {
}
new String(baos.toByteArray(), StandardCharsets.US_ASCII)
}
public static String encodeXHave(List<Integer> pieces, int totalPieces) {
int bytes = totalPieces / 8
if (totalPieces % 8 != 0)
bytes++
byte[] raw = new byte[bytes]
pieces.each {
int byteIdx = it / 8
int offset = it % 8
int mask = 0x80 >>> offset
raw[byteIdx] |= mask
}
Base64.encode(raw)
}
public static List<Integer> decodeXHave(String xHave) {
byte [] availablePieces = Base64.decode(xHave)
List<Integer> available = new ArrayList<>()
availablePieces.eachWithIndex {b, i ->
for (int j = 0; j < 8 ; j++) {
byte mask = 0x80 >>> j
if ((b & mask) == mask) {
available.add(i * 8 + j)
}
}
}
available
}
}

View File

@@ -26,6 +26,15 @@ public class SharedFile {
return pieceSize;
}
public int getNPieces() {
long length = file.length();
int rawPieceSize = 0x1 << pieceSize;
int rv = (int) (length / rawPieceSize);
if (length % pieceSize != 0)
rv++;
return rv;
}
@Override
public int hashCode() {
return file.hashCode() ^ infoHash.hashCode();

View File

@@ -1,17 +1,26 @@
package com.muwire.core.download
import static org.junit.Assert.fail
import org.junit.After
import org.junit.Before
import org.junit.Test
import com.muwire.core.EventBus
import com.muwire.core.InfoHash
import com.muwire.core.Persona
import com.muwire.core.Personas
import com.muwire.core.connection.Endpoint
import com.muwire.core.files.FileHasher
import static com.muwire.core.util.DataUtil.readTillRN
import static com.muwire.core.util.DataUtil.encodeXHave
import net.i2p.data.Base64
import net.i2p.util.ConcurrentHashSet
class DownloadSessionTest {
private EventBus eventBus
private File source, target
private InfoHash infoHash
private Endpoint endpoint
@@ -24,6 +33,16 @@ class DownloadSessionTest {
private InputStream fromDownloader, fromUploader
private OutputStream toDownloader, toUploader
private volatile boolean performed
private Set<Integer> available = new ConcurrentHashSet<>()
private volatile IOException thrown
@Before
public void setUp() {
eventBus = new EventBus()
}
private void initSession(int size, def claimedPieces = []) {
Random r = new Random()
byte [] content = new byte[size]
@@ -56,12 +75,20 @@ class DownloadSessionTest {
toUploader = new PipedOutputStream(fromDownloader)
endpoint = new Endpoint(null, fromUploader, toUploader, null)
session = new DownloadSession("",pieces, infoHash, endpoint, target, pieceSize, size)
downloadThread = new Thread( { session.request() } as Runnable)
session = new DownloadSession(eventBus, "",pieces, infoHash, endpoint, target, pieceSize, size, available)
downloadThread = new Thread( { perform() } as Runnable)
downloadThread.setDaemon(true)
downloadThread.start()
}
private void perform() {
try {
performed = session.request()
} catch (IOException e) {
thrown = e
}
}
@After
public void teardown() {
source?.delete()
@@ -76,6 +103,7 @@ class DownloadSessionTest {
assert "GET $rootBase64" == readTillRN(fromDownloader)
assert "Range: 0-19" == readTillRN(fromDownloader)
readTillRN(fromDownloader)
readTillRN(fromDownloader)
assert "" == readTillRN(fromDownloader)
toDownloader.write("200 OK\r\n".bytes)
@@ -87,6 +115,9 @@ class DownloadSessionTest {
assert pieces.isComplete()
assert target.bytes == source.bytes
assert performed
assert available.isEmpty()
assert thrown == null
}
@Test
@@ -98,6 +129,7 @@ class DownloadSessionTest {
assert "GET $rootBase64" == readTillRN(fromDownloader)
readTillRN(fromDownloader)
readTillRN(fromDownloader)
readTillRN(fromDownloader)
assert "" == readTillRN(fromDownloader)
toDownloader.write("200 OK\r\n".bytes)
@@ -108,6 +140,9 @@ class DownloadSessionTest {
Thread.sleep(150)
assert pieces.isComplete()
assert target.bytes == source.bytes
assert performed
assert available.isEmpty()
assert thrown == null
}
@Test
@@ -125,6 +160,7 @@ class DownloadSessionTest {
assert (start == 0 && end == ((1 << pieceSize) - 1)) ||
(start == (1 << pieceSize) && end == (1 << pieceSize))
readTillRN(fromDownloader)
readTillRN(fromDownloader)
assert "" == readTillRN(fromDownloader)
@@ -138,6 +174,9 @@ class DownloadSessionTest {
Thread.sleep(150)
assert !pieces.isComplete()
assert 1 == pieces.donePieces()
assert performed
assert available.isEmpty()
assert thrown == null
}
@Test
@@ -145,7 +184,10 @@ class DownloadSessionTest {
initSession(20, [0])
long now = System.currentTimeMillis()
downloadThread.join(100)
assert 100 > (System.currentTimeMillis() - now)
assert 100 >= (System.currentTimeMillis() - now)
assert !performed
assert available.isEmpty()
assert thrown == null
}
@Test
@@ -164,4 +206,128 @@ class DownloadSessionTest {
assert pieces.claimed.get(0)
assert start == 0 && end == (1 << pieceSize) - 1
}
@Test
public void test416NoHave() {
initSession(20)
readAllHeaders(fromDownloader)
toDownloader.write("416 don't have it\r\n\r\n".bytes)
toDownloader.flush()
Thread.sleep(150)
assert !performed
assert available.isEmpty()
assert thrown != null
}
@Test
public void test416Have() {
initSession(20)
readAllHeaders(fromDownloader)
toDownloader.write("416 don't have it\r\n".bytes)
toDownloader.write("X-Have: ${encodeXHave([0], 1)}\r\n\r\n".bytes)
toDownloader.flush()
Thread.sleep(150)
assert performed
assert available.contains(0)
assert thrown == null
}
@Test
public void test416Have2Pieces() {
int pieceSize = FileHasher.getPieceSize(1)
int size = (1 << pieceSize) + 1
initSession(size)
readAllHeaders(fromDownloader)
toDownloader.write("416 don't have it\r\n".bytes)
toDownloader.write("X-Have: ${encodeXHave([1], 2)}\r\n\r\n".bytes)
toDownloader.flush()
Thread.sleep(150)
assert performed
assert available.contains(1)
assert thrown == null
}
@Test
public void test200TwoPieces1Available() {
int pieceSize = FileHasher.getPieceSize(1)
int size = (1 << pieceSize) * 9 + 1
initSession(size)
Set<String> headers = readAllHeaders(fromDownloader)
def matcher = null
headers.each {
if (it.startsWith("Range"))
matcher = (it =~ /^Range: (\d+)-(\d+)$/)
}
assert matcher.groupCount() > 0
int start = Integer.parseInt(matcher[0][1])
int end = Integer.parseInt(matcher[0][2])
if (start == 0)
fail("inconlcusive")
toDownloader.write("416 don't have it \r\n".bytes)
toDownloader.write("X-Have: ${encodeXHave([0],2)}\r\n\r\n".bytes)
toDownloader.flush()
downloadThread.join()
assert performed
performed = false
assert available.contains(0)
assert thrown == null
// request same session
downloadThread = new Thread( { perform() } as Runnable)
downloadThread.setDaemon(true)
downloadThread.start()
Thread.sleep(150)
headers = readAllHeaders(fromDownloader)
matcher = null
headers.each {
if (it.startsWith("Range"))
matcher = (it =~ /^Range: (\d+)-(\d+)$/)
}
assert matcher.groupCount() > 0
start = Integer.parseInt(matcher[0][1])
end = Integer.parseInt(matcher[0][2])
assert start == 0
}
@Test
public void testXAlt() throws Exception {
Personas personas = new Personas()
def sources = []
def listener = new Object() {
public void onSourceDiscoveredEvent(SourceDiscoveredEvent e) {
sources << e.source
}
}
eventBus.register(SourceDiscoveredEvent.class, listener)
initSession(20)
readAllHeaders(fromDownloader)
toDownloader.write("416 don't have it\r\n".bytes)
toDownloader.write("X-Alt: ${personas.persona1.toBase64()},${personas.persona2.toBase64()}\r\n\r\n".bytes)
toDownloader.flush()
Thread.sleep(150)
assert sources.contains(personas.persona1)
assert sources.contains(personas.persona2)
assert 2 == sources.size()
}
private static Set<String> readAllHeaders(InputStream is) {
Set<String> rv = new HashSet<>()
String header
while((header = readTillRN(is)) != "")
rv.add(header)
rv
}
}

View File

@@ -34,4 +34,19 @@ class PiecesTest {
pieces.markDownloaded(piece2)
assert pieces.isComplete()
}
@Test
public void testClaimAvailable() {
pieces = new Pieces(2)
int claimed = pieces.claim([0].toSet())
assert claimed == 0
assert -1 == pieces.claim([0].toSet())
}
@Test
public void testClaimNoneAvailable() {
pieces = new Pieces(20)
int claimed = pieces.claim()
assert -1 == pieces.claim([claimed].toSet())
}
}

View File

@@ -83,4 +83,11 @@ class SearchIndexTest {
assert found.size() == 1
assert found.contains("b c.d")
}
@Test
void testDuplicateTerm() {
initIndex(["MuWire-0.3.3.jar"])
def found = index.search(["muwire", "0", "3", "jar"])
assert found.size() == 1
}
}

View File

@@ -1,5 +1,5 @@
group = com.muwire
version = 0.3.1
version = 0.3.4
groovyVersion = 2.4.15
slf4jVersion = 1.7.25
spockVersion = 1.1-groovy-2.4

View File

@@ -67,7 +67,9 @@ class MainFrameController {
// this can be improved a lot
def replaced = search.toLowerCase().trim().replaceAll(Constants.SPLIT_PATTERN, " ")
def terms = replaced.split(" ")
searchEvent = new SearchEvent(searchTerms : terms, uuid : uuid, oobInfohash: true)
def nonEmpty = []
terms.each { if (it.length() > 0) nonEmpty << it }
searchEvent = new SearchEvent(searchTerms : nonEmpty, uuid : uuid, oobInfohash: true)
}
core.eventBus.publish(new QueryEvent(searchEvent : searchEvent, firstHop : true,
replyTo: core.me.destination, receivedOn: core.me.destination,