julian.kiedaisch/mdg-filius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(core): Gnutella search and download after Java, measured against Java

Browse Source 
GnutellaRun adds a line of four peers: a search goes on as a query, each
peer answers with a hit for each matching file and the hits come back the
way the query came; a hit's file is fetched by HTTP on port 6346. The
file type names of the model move from the project export and import into
the core, so that the download uses them too.

Departures from Java: a peer that sent hits keeps its connection, so that
the hits of farther peers come back too (Java closes it and loses them),
and sizes are in characters.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
julian.kiedaisch
2026-10-02 14:59:26 +00:00
co-authored by Claude Opus 5.5
parent b4a95e6cac
commit ce5b10c1f3
13 changed files with 39934 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/src/runtime/apps/gnutella.ts
+237 -20
View File
@@ -2,21 +2,32 @@
// with its PeerToPeerServer on port 6346 and PeerToPeerClient): a host joins
// the network by a ping to one peer, which answers with a pong, takes the
// host as a peer and sends the ping on to its own peers, whose pongs come
// back the same way; each knows at most its number of peers. The messages
// are Java's text packets (PeerToPeerPaket.toString), each on a connection
// of its own that stays open, as Java's do.
// back the same way; each knows at most its number of peers. A search goes
// to the peers as a query, each sends it on and answers with a hit for each
// of its files in /peer2peer whose name holds the words; hits come back the
// way the query came. A hit's file is fetched from its host by HTTP on the
// same port. The messages are Java's text packets (PeerToPeerPaket
// .toString), each on a connection of its own that stays open, as Java's do.
//
// Departures from Java (gnutellaJava.test.ts names each):
// - a ping that comes a second time gets no pong, and its connection stays
// (Java means so, but its worker throws on the missing pong and closes the
// connection).
// connection);
// - a peer that answered a query with hits keeps its connection, so that the
// hits of the peers it sent the query on to come back too (Java closes it
// after its own hits, and theirs are lost);
// - a hit names a text file's size in characters (Java decodes the text as
// Base64: "Hallo Welt" is 6 B).
import type { Id } from '../../ids.js';
import { fileContentOfJavaType, javaFileTypeName, type VirtualFileContent } from '../../model/virtualFileSystem.js';
import type { GnutellaConfiguration } from '../config.js';
import { formatHttpRequest, formatHttpResponse, HTTP_GET, httpPath, parseHttpRequest, parseHttpResponse } from '../http.js';
import type { NetworkRuntimeEngine } from '../engine.js';
import type { Steps } from '../threads.js';
import { ProgramLog } from './program.js';
import { TcpServerProgram } from './server.js';
import { javaBase64 } from '../../util/javaBase64.js';
/** PeerToPeerServer, PeerToPeerClientMitarbeiter: Gnutella's port. */
export const GNUTELLA_PORT = 6346;
@@ -47,9 +58,27 @@ export interface PongPacket extends GnutellaHead {
readonly bytes: number;
}
export type GnutellaPacket = PingPacket | PongPacket;
/** QueryPaket: the least speed asked for (Java's search asks "1") and the words searched for. */
export interface QueryPacket extends GnutellaHead {
readonly kind: 'query';
readonly minimumSpeed: string;
readonly criteria: string;
}
const PAYLOAD_KIND = { ping: '0x00', pong: '0x01' } as const;
/** QueryHitPaket: how many hits its host has, its port and address, its speed, the hit ("name: size B"), its servent id. */
export interface QueryHitPacket extends GnutellaHead {
readonly kind: 'queryHit';
readonly hits: number;
readonly port: number;
readonly ipAddress: string;
readonly speed: string;
readonly result: string;
readonly servent: string;
}
export type GnutellaPacket = PingPacket | PongPacket | QueryPacket | QueryHitPacket;
const PAYLOAD_KIND = { ping: '0x00', pong: '0x01', query: '0x80', queryHit: '0x81' } as const;
/** PeerToPeerPaket.anzahlBenoetigterBits: the least exponent whose power of two is at least the number. */
function bitsFor(value: number): number {
@@ -63,10 +92,35 @@ export function pongPayloadLength(ipAddress: string, port: number, files: number
return ipAddress.length * 8 + bitsFor(port) + bitsFor(files) + bitsFor(bytes);
}
/** PeerToPeerPaket.toString and its subclasses': the fields joined by "//". */
/** QueryPaket.payloadLengthBerechnen. */
export function queryPayloadLength(minimumSpeed: string, criteria: string): number {
return (minimumSpeed.length + criteria.length) * 8;
}
/** QueryHitPaket.payloadLengthBerechnen, as Java's worker leaves it: counted before the hit is put in (an empty one). */
export function queryHitPayloadLength(hits: number, port: number, ipAddress: string, speed: string, servent: string): number {
return bitsFor(hits) + bitsFor(port) + ipAddress.length * 8 + speed.length * 8 + servent.length * 8;
}
/** PeerToPeerPaket.toString and its subclasses': the fields joined by "//"; an empty text written as one blank. */
export function gnutellaText(packet: GnutellaPacket): string {
const head = [packet.guid, PAYLOAD_KIND[packet.kind], packet.hops, packet.ttl, packet.payloadLength];
const body = packet.kind === 'ping' ? [packet.ipAddress] : [packet.ipAddress, packet.port, packet.files, packet.bytes];
const blankForEmpty = (text: string) => (text === '' ? ' ' : text);
let body: (string | number)[];
switch (packet.kind) {
case 'ping':
body = [packet.ipAddress];
break;
case 'pong':
body = [packet.ipAddress, packet.port, packet.files, packet.bytes];
break;
case 'query':
body = [blankForEmpty(packet.minimumSpeed), blankForEmpty(packet.criteria)];
break;
case 'queryHit':
body = [packet.hits, packet.port, packet.ipAddress, packet.speed, blankForEmpty(packet.result), blankForEmpty(packet.servent)];
break;
}
return [...head, ...body].join('//');
}
@@ -93,11 +147,16 @@ export function gnutellaPacket(text: string): GnutellaPacket | undefined {
if (fields[1] === PAYLOAD_KIND.pong && fields[5] !== undefined && port !== undefined && files !== undefined && bytes !== undefined) {
return { kind: 'pong', ...head, ipAddress: fields[5], port, files, bytes };
}
if (fields[1] === PAYLOAD_KIND.query && fields[5] !== undefined && fields[6] !== undefined) return { kind: 'query', ...head, minimumSpeed: fields[5], criteria: fields[6] };
const [hits, hitPort] = [number(5), number(6)];
if (fields[1] === PAYLOAD_KIND.queryHit && hits !== undefined && hitPort !== undefined && fields.length >= 11) {
return { kind: 'queryHit', ...head, hits, port: hitPort, ipAddress: fields[7]!, speed: fields[8]!, result: fields[9]!, servent: fields[10]! };
}
return undefined;
}
/** What the program tells its window: its known peers, whenever they change. */
export type GnutellaLogEntry = { readonly kind: 'peers'; readonly peers: readonly string[] };
/** What the program tells its window: its known peers, or its search results ("address/name: size B"), whenever they change. */
export type GnutellaLogEntry = { readonly kind: 'peers'; readonly peers: readonly string[] } | { readonly kind: 'results'; readonly results: readonly string[] };
/** PeerToPeerServer: answers pings on 6346; a connection's worker remembers the id of its last packet. */
class GnutellaServer extends TcpServerProgram {
@@ -125,8 +184,13 @@ class GnutellaServer extends TcpServerProgram {
return;
}
const text = new TextDecoder().decode(received.payload);
// PeerToPeerServerMitarbeiter.verarbeiteNachricht: a text without a packet's head throws, and the worker closes the
// connection; a packet of another kind is left.
const send = (answer: string) => this.engine.sendTcpSteps(socketId, new TextEncoder().encode(answer));
// PeerToPeerServerMitarbeiter.verarbeiteNachricht: a request for a file; else a text without a packet's head
// throws, and the worker closes the connection; a packet of another kind is left.
if (text.startsWith(HTTP_GET)) {
yield* send(this.peer.fileResponse(text));
continue;
}
if (!hasGnutellaHead(text)) {
this.lastGuidBySocket.delete(socketId);
return;
@@ -134,7 +198,8 @@ class GnutellaServer extends TcpServerProgram {
const packet = gnutellaPacket(text);
if (packet === undefined) continue;
this.lastGuidBySocket.set(socketId, packet.guid);
if (packet.kind === 'ping') yield* this.peer.answerPing(packet, address, (text) => this.engine.sendTcpSteps(socketId, new TextEncoder().encode(text)));
if (packet.kind === 'ping') yield* this.peer.answerPing(packet, address, send);
else if (packet.kind === 'query') yield* this.peer.answerQuery(packet, address, send);
}
}
@@ -155,10 +220,15 @@ class GnutellaServer extends TcpServerProgram {
export class GnutellaPeer {
readonly log: ProgramLog<GnutellaLogEntry>;
private peersValue: string[] = [];
private resultsValue: string[] = [];
private readonly maximumKnownPeers: number;
/** The ids of the program's own pings (eigeneAnfragen), of others' (fremdeAnfragen), and of those it sent on (schonmalVerschicktListe). */
/**
* The ids of the program's own pings and queries (eigeneAnfragen), and of
* those it sent on (schonmalVerschicktListe). Java also keeps others'
* (fremdeAnfragen) and sends back only their hits; a hit goes back only on
* a connection whose last packet had its id, which comes to the same.
*/
private readonly ownRequests = new Set<number>();
private readonly foreignRequests = new Set<number>();
private readonly alreadySent = new Set<number>();
private readonly server: GnutellaServer;
@@ -177,6 +247,11 @@ export class GnutellaPeer {
return this.peersValue;
}
/** The search results (holeErgebnisse): each hit's host address and its hit, "address/name: size B". */
get results(): readonly string[] {
return this.resultsValue;
}
/** The node's first address (Betriebssystem.primaryIPAdresse). */
private get ownAddress(): string {
return this.engine.networkInterfaces(this.nodeId)[0]?.ipAddress ?? '';
@@ -196,6 +271,46 @@ export class GnutellaPeer {
this.log.write({ kind: 'peers', peers: [] });
}
/** sucheDatei: no results, a query for the words to every peer. */
search(criteria: string): void {
this.setResults([]);
const query: QueryPacket = {
kind: 'query',
guid: this.engine.drawGnutellaGuid(),
hops: 0,
ttl: 8,
payloadLength: queryPayloadLength('1', criteria),
minimumSpeed: '1',
criteria,
};
this.ownRequests.add(query.guid);
this.sendQueryOn(query, this.ownAddress);
}
/** loescheSuchergebnisse. */
clearResults(): void {
this.setResults([]);
}
/**
* herunterladenDatei: the result's file asked for by HTTP from its host
* (the result's address, and its name up to the ":"), and kept in
* /peer2peer when the host has it.
*/
download(index: number): void {
const result = this.resultsValue[index];
if (result === undefined) return;
const [owner, rest] = result.split('/').filter((part) => part !== '');
const name = rest?.split(':').find((part) => part !== '');
if (owner === undefined || name === undefined) return;
this.engine.startThread(this.fetchFile(owner, name));
}
private setResults(results: string[]): void {
this.resultsValue = results;
this.log.write({ kind: 'results', results });
}
/** hinzuTeilnehmer: a new peer while there is room. */
private addPeer(address: string): void {
if (this.peersValue.includes(address) || this.peersValue.length >= this.maximumKnownPeers) return;
@@ -213,7 +328,7 @@ export class GnutellaPeer {
// erstellePong: none for a ping sent on before.
if (!this.alreadySent.has(ping.guid)) {
const files = this.sharedFiles();
const bytes = files.reduce((sum, size) => sum + size, 0);
const bytes = files.reduce((sum, file) => sum + file.size, 0);
const address = this.ownAddress;
const pong: PongPacket = {
kind: 'pong',
@@ -235,28 +350,129 @@ export class GnutellaPeer {
private sendPingOn(ping: PingPacket, from: string): void {
if (ping.ttl <= 0 || this.alreadySent.has(ping.guid)) return;
this.alreadySent.add(ping.guid);
this.foreignRequests.add(ping.guid);
for (const peer of this.peersValue) {
if (peer !== from && peer !== ping.ipAddress) this.engine.startThread(this.sendTo(peer, ping));
}
}
/**
* PeerToPeerServerMitarbeiter.verarbeiteQuery, PeerToPeerAnwendung
* .verarbeiteAnfrage: a query not seen before sent on to the other peers
* while it lives, and answered with a hit for each file of /peer2peer
* whose name holds the words (in any case).
*/
*answerQuery(query: QueryPacket, from: string, send: (text: string) => Steps<unknown>): Steps {
if (this.alreadySent.has(query.guid)) return;
this.alreadySent.add(query.guid);
const wanted = query.criteria.toLowerCase();
const hits = this.sharedFiles().filter((file) => file.name.toLowerCase().includes(wanted));
if (query.ttl > 0) this.sendQueryOn({ ...query, ttl: query.ttl - 1, hops: query.hops + 1 }, from);
const address = this.ownAddress;
for (const hit of hits) {
const packet: QueryHitPacket = {
kind: 'queryHit',
guid: query.guid,
hops: 0,
ttl: 8,
payloadLength: queryHitPayloadLength(hits.length, GNUTELLA_PORT, address, '2', ' '),
hits: hits.length,
port: GNUTELLA_PORT,
ipAddress: address,
speed: '2',
result: `${hit.name}: ${hit.size} B`,
servent: ' ',
};
yield* send(gnutellaText(packet));
}
}
/** PeerToPeerClient.sendeAnfrage: to every peer but the sender. */
private sendQueryOn(query: QueryPacket, from: string): void {
for (const peer of this.peersValue) if (peer !== from) this.engine.startThread(this.sendTo(peer, query));
}
/** verarbeiteQueryHit: a hit for the program's own search (a result, once), else sent back the way its query came while it lives. */
private takeQueryHit(hit: QueryHitPacket): void {
if (this.ownRequests.has(hit.guid)) {
const result = `${hit.ipAddress}/${hit.result}`;
if (!this.resultsValue.includes(result)) this.setResults([...this.resultsValue, result]);
} else if (hit.ttl > 0) {
this.server.sendBack({ ...hit, ttl: hit.ttl - 1, hops: hit.hops + 1 });
}
}
/** httpAnfrageVerarbeiten: the asked file of /peer2peer with its type, or 404. */
fileResponse(text: string): string {
const request = parseHttpRequest(text);
const path = request.type === 'request' ? `${GNUTELLA_FOLDER}/${httpPath(request.request.path)}` : undefined;
const file = path === undefined ? undefined : this.sharedFile(path);
if (file === undefined) return formatHttpResponse({ protocolVersion: 'HTTP/1.1', statusCode: 404, contentType: undefined, body: undefined });
const content = file.content;
return formatHttpResponse({
protocolVersion: 'HTTP/1.1',
statusCode: 200,
contentType: javaFileTypeName(file.name, content),
body: content.kind === 'text' ? content.text : javaBase64(content.data),
});
}
/** PeerToPeerClientMitarbeiter.run for HTTP: the request, the answer, the file kept when it came. */
private *fetchFile(owner: string, name: string): Steps {
const engine = this.engine;
const socketId = engine.openTcpClientSocket(this.nodeId, owner, GNUTELLA_PORT);
if (socketId === undefined || (yield* engine.connectTcpSteps(socketId)) !== 'connected') {
if (socketId !== undefined) engine.requestTcpClose(socketId);
return;
}
yield* engine.sendTcpSteps(socketId, new TextEncoder().encode(formatHttpRequest({ method: HTTP_GET, path: name, host: owner })));
const received = yield* engine.receiveTcpSteps(socketId);
const response = received.type === 'message' ? parseHttpResponse(new TextDecoder().decode(received.payload)) : undefined;
if (response?.statusCode === 200) {
const path = `${GNUTELLA_FOLDER}/${name}`;
this.keepFile(path, fileContentOfJavaType(response.contentType ?? '', response.body ?? '', path));
}
engine.requestTcpClose(socketId);
}
/** speicherDatei: over a file of that name in any case, keeping its name (as the file system does). */
private keepFile(path: string, content: VirtualFileContent): void {
const files = this.engine.state.topologySnapshot.fileSystemsByNodeId.get(this.nodeId);
if (files === undefined || content.kind === 'directory') return;
if (content.kind === 'text') files.writeTextFile(path, content.text);
else if (content.kind === 'image') files.writeImageFile(path, content.data, content.mediaType);
else if (content.mediaType === undefined) files.writeBinaryFile(path, content.data);
else files.writeBinaryFile(path, content.data, content.mediaType);
}
/** verarbeitePong: the program's own ping answered (a new peer), else the pong sent back the way its ping came. */
private takePong(pong: PongPacket): void {
if (this.ownRequests.has(pong.guid)) this.addPeer(pong.ipAddress);
else this.server.sendBack({ ...pong, ttl: pong.ttl - 1, hops: pong.hops + 1 });
}
/** The sizes of the files in /peer2peer (Datei.holeGroesse, a text in characters). */
private sharedFiles(): number[] {
/** The files in /peer2peer (Dateisystem.holeDateien), their names and sizes (Datei.holeGroesse, a text in characters). */
private sharedFiles(): { readonly name: string; readonly size: number }[] {
const files = this.engine.state.topologySnapshot.fileSystemsByNodeId.get(this.nodeId);
if (files === undefined || !files.contains(GNUTELLA_FOLDER)) return [];
return files.entriesInCreationOrder(GNUTELLA_FOLDER).flatMap((entry) => {
const content = entry.content;
return content.kind === 'directory' ? [] : [content.kind === 'text' ? content.text.length : content.data.length];
const name = entry.path.slice(entry.path.lastIndexOf('/') + 1);
return content.kind === 'directory' ? [] : [{ name, size: content.kind === 'text' ? content.text.length : content.data.length }];
});
}
/** A file at the path, found in any case (Dateisystem.holeDatei); none for a folder or nothing. */
private sharedFile(path: string): { readonly path: string; readonly name: string; readonly content: Exclude<VirtualFileContent, { kind: 'directory' }> } | undefined {
const files = this.engine.state.topologySnapshot.fileSystemsByNodeId.get(this.nodeId);
try {
const entry = files?.entry(path);
if (entry === undefined || entry.content.kind === 'directory') return undefined;
return { path: entry.path, name: entry.path.slice(entry.path.lastIndexOf('/') + 1), content: entry.content };
} catch {
return undefined;
}
}
/**
* PeerToPeerClientMitarbeiter.run: a connection to the peer's port 6346, the
* packet sent, then each answer taken until the connection ends.
@@ -274,6 +490,7 @@ export class GnutellaPeer {
if (received.type !== 'message') break;
const answer = gnutellaPacket(new TextDecoder().decode(received.payload));
if (answer?.kind === 'pong') this.takePong(answer);
else if (answer?.kind === 'queryHit') this.takeQueryHit(answer);
}
engine.requestTcpClose(socketId);
}