Skip to content

R
railgun-routing-client
Commit 7d5ae443 authored by 神楽坂玲奈's avatar 神楽坂玲奈
Browse files
Options

test

parent 688ab046
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 331 additions and 231 deletions
.gitignore
View file @ 7d5ae443
/.idea/ /.idea/
\ No newline at end of file /node_modules/
config.json → config/config.json
View file @ 7d5ae443
{ {
"server_address": "127.0.0.1", "server_address": "127.0.0.1",
"server_port": 9999, "server_port": 500,
"port": 9998, "port": 495,
"timeout": 10, "timeout": 10,
"interval": 1000 "interval": 1000,
"table": 1,
"proto": 250
} }
config/routers.json 0 → 100644
View file @ 7d5ae443
[
{
"id": 1,
"address": "10.198.0.1",
"interface": "mc-yangtze",
"subnets": [
"192.168.1.0/24",
"10.198.1.0/24",
"10.198.2.0/24"
]
},
{
"id": 3,
"address": "10.198.0.3",
"interface": "mc-south",
"subnets": [
"10.198.3.0/24"
]
},
{
"id": 7,
"address": "10.198.0.7",
"interface": "mc-fractal",
"subnets": [
"10.198.7.0/24"
]
},
{
"id": 9,
"address": "10.198.0.9",
"interface": "mc-marcia",
"subnets": [
"10.198.9.0/24"
]
}
]
package-lock.json
View file @ 7d5ae443
This diff is collapsed.
package.json
View file @ 7d5ae443
{ {
"name": "railgun-network-client", "name": "railgun-routing-client",
"version": "1.0.0", "version": "0.0.1",
"description": "", "author": "zh99998 <zh99998@gmail.com>",
"scripts": { "scripts": {
"start": "ts-node src/main.ts" "start": "ts-node src/main.ts"
}, },
"main": "index.js", "main": "index.js",
"devDependencies": { "devDependencies": {
"@types/node": "^14.14.14", "@types/node": "^16.6.1",
"ts-node": "^9.1.1" "ts-node": "^10.2.0",
"typescript": "^4.3.5"
} }
} }
peers.json deleted 100644 → 0
View file @ 688ab046
{
"id": 3,
"peers": [
{
"id": 23,
"address": "10.200.23.3",
"subnets": ["10.198.0.25"],
"mark": 18003
}
]
}
protocol.ts 0 → 100644
View file @ 7d5ae443
// 这个文件两端共用
export interface PeerQuality {
delay: number;
jitter: number;
reliability: number;
}
// 路由器向中心服务器发送的消息
export interface UploadMessage {
id: number; // router id
ack: number;
peers?: Record<number, PeerQuality>;
}
// 中心服务器向路由器发送的消息
export interface DownloadMessage {
seq: number,
to: number,
nextHop: number;
}
// 路由器向路由器发送的消息
export interface PeerMessage {
id: number;
seq: number;
time: number;
}
src/Peer.ts
View file @ 7d5ae443
import config from '../config.json'; import config from '../config.json';
import { Socket } from 'dgram'; import { PeerMessage, PeerQuality } from '../protocol';
import { PeerMeasure } from './protocol';
export interface PeerHello { export interface RouterConfig {
id: number;
seq: number;
time: number;
}
export interface PeerConfig {
id: number; id: number;
address: string; address: string;
subnets: string[]; subnets: string[];
} }
export class Peer implements PeerHello, PeerMeasure, PeerConfig { export class Peer implements PeerMessage, PeerQuality, RouterConfig {
id: number; id: number;
address: string; address: string;
subnets: string[]; subnets: string[];
delay: number = 0; delay: number = 0;
jitter: number = 0;
reliability: number = 0; reliability: number = 0;
seq: number = 0; seq: number = 0;
time: number = 0; time: number = 0;
constructor(config: PeerConfig) { constructor(config: RouterConfig) {
Object.assign(this, config); Object.assign(this, config);
} }
...@@ -37,7 +29,7 @@ export class Peer implements PeerHello, PeerMeasure, PeerConfig { ...@@ -37,7 +29,7 @@ export class Peer implements PeerHello, PeerMeasure, PeerConfig {
this.time = 0; this.time = 0;
} }
onMessage(data: PeerHello) { onMessage(data: PeerMessage) {
if (data.seq == 0 || data.seq < this.seq - config.timeout || data.seq > this.seq + config.timeout) { if (data.seq == 0 || data.seq < this.seq - config.timeout || data.seq > this.seq + config.timeout) {
// 收到 seq = 0 或 seq 与之前差距较大，就 reset // 收到 seq = 0 或 seq 与之前差距较大，就 reset
this.reset(); this.reset();
...@@ -57,23 +49,14 @@ export class Peer implements PeerHello, PeerMeasure, PeerConfig { ...@@ -57,23 +49,14 @@ export class Peer implements PeerHello, PeerMeasure, PeerConfig {
this.time = data.time; this.time = data.time;
} }
update(socket: Socket, self: PeerHello) { update(time: number) {
if (this.reliability == 0) { if (this.reliability == 0) {
return; return;
} }
// 有几个包没到 // 有几个包没到
const step = Math.floor((self.time - this.time + this.delay - config.interval) / config.interval); const step = Math.floor((time - this.time + this.delay - config.interval) / config.interval);
if (step > config.timeout) { if (step > config.timeout) {
this.reset(); this.reset();
return;
} }
// if(self.time - this.time )
socket.send(JSON.stringify(self), config.port, this.address, (error, bytes) => error && console.warn(error));
} }
// pickServer(): PeerMeasure {
// const { id, delay, reliability } = this;
// return { id, delay, reliability };
// }
} }
src/RouteWriter.ts
View file @ 7d5ae443
import { Peer } from './Peer'; import config from '../config.json';
import routers from '../routers.json';
const tableOffset = 10000;
const groupTableOffset = 11000;
export class RouteWriter { export class RouteWriter {
routes: Route[];
constructor() {}
reset() {
// TODO: 所有 neighbor 重设为直连，不是 neighbor 的设为 unreachable
// 只是记下来，commit的时候再写入
}
add(to: Peer, via: Peer) { static reset() {
for (const subnet of to.subnets) { console.log(`route flush table ${config.table} proto ${config.proto}`);
this.routes.push({ to: subnet, via: via.address, table: 'main' }); for (const peer of routers.filter(r => r.interface)) {
this.set(peer.id, peer.id);
} }
this.routes.push({ to: 'default', via: via.address, table: tableOffset + to.id });
} }
// groupAdd(to: Peer, via: Peer) { static set(toId: number, viaId: number) {
// this.routes.push({ to: 'default', via: via.address, table: groupTableOffset + to.id }); const to = routers.find(r => r.id == toId);
// } const via = routers.find(r => r.id == viaId);
for (const address of [to.address, ...to.subnets]) {
commit() { console.log(`route replace ${address} dev ${via.interface}`);
// TODO: 写进系统，用 ip -batch }
} }
} }
export interface Route {
to: string;
via: string;
table: number | string;
}
src/Server.ts
View file @ 7d5ae443
import { RouteWriter } from './RouteWriter';
import { Peer, PeerHello, PeerMeasure } from './Peer';
import { Socket } from 'dgram'; import { Socket } from 'dgram';
import config from '../config.json'; import config from '../config.json';
import { RouteWriter } from './RouteWriter';
import { Peer } from './Peer';
import { DownloadMessage, PeerMessage, PeerQuality, UploadMessage } from '../protocol';
export class Server { export class Server {
ack = 0; ack = 0;
onMessage(data: ServerMessage, socket: Socket, self: PeerHello, peers: Map<number, Peer>) { onMessage(socket: Socket, message: DownloadMessage, self: PeerMessage) {
if (data.seq && this.ack != data.seq) { if (message.seq && this.ack != message.seq) {
return; return;
} }
const routeWriter = new RouteWriter(); if (message.seq === 0) {
RouteWriter.reset();
if (data.seq === 0) {
routeWriter.reset();
} }
for (const [to, via] of Object.entries(data.routes).map(([to, via]) => [peers.get(parseInt(to)), peers.get(via)])) {
routeWriter.add(to, via);
}
// for (const [to, via] of Object.entries(data.groups).map(([to, via]) => [peers.get(parseInt(to)), peers.get(via)])) {
// routeWriter.groupAdd(to, via);
// }
this.ack = data.seq + 1; RouteWriter.set(message.to, message.nextHop);
this.ack = message.seq + 1;
const response: UploadMessage = {
id: self.id,
ack: this.ack
};
socket.send( socket.send(JSON.stringify(response), config.server_port, config.server_address);
JSON.stringify({
id: self.id,
ack: this.ack,
}),
config.server_port,
config.server_address
);
} }
update(socket: Socket, self: PeerHello, peers: Map<number, Peer>) { update(socket: Socket, self: PeerMessage, peers: Peer[]) {
const p: PeerMeasure[] = []; const p: Record<number, PeerQuality> = {};
for (const peer of peers.values()) { for (const peer of peers) {
if (peer.reliability === 0) { if (peer.reliability === 0) {
continue; continue;
} }
...@@ -52,15 +46,12 @@ export class Server { ...@@ -52,15 +46,12 @@ export class Server {
const reliability = (peer.reliability * (config.timeout - step)) / config.timeout; const reliability = (peer.reliability * (config.timeout - step)) / config.timeout;
const { id, delay } = peer; const { id, delay } = peer;
p.push({ id, delay, reliability });
// jitter 还没算
p[id] = { delay, jitter: 0, reliability };
} }
socket.send(JSON.stringify({ id: self.id, ack: this.ack, peers: p }), config.server_port, config.server_address); const message: UploadMessage = { id: self.id, ack: this.ack, peers: p };
socket.send(JSON.stringify(message), config.server_port, config.server_address);
} }
} }
export interface ServerMessage {
seq: number;
routes: { [to: number]: number };
groups: { [to: number]: number };
}
src/main.ts
View file @ 7d5ae443
import dgram from 'dgram'; import dgram from 'dgram';
import assert from 'assert'; import assert from 'assert';
import { Peer, PeerHello } from './Peer';
import config from '../config.json'; import config from '../config.json';
import { Server, ServerMessage } from './Server'; import routers from '../routers.json';
import { id, peers as peersConfig } from '../peers.json'; import { Server } from './Server';
import { Peer } from './Peer';
import { DownloadMessage, PeerMessage } from '../protocol';
const self: PeerHello = { id, seq: 0, time: 0 }; const self: PeerMessage = { id: parseInt(process.env.RAILGUN_ID), seq: 0, time: 0 };
const server = new Server(); const server = new Server();
const peers = new Map<number, Peer>(); const peers = routers.filter(r => r.id !== self.id && r.interface).map(r => new Peer(r));
for (const peer of peersConfig) {
peers.set(peer.id, new Peer(peer));
}
const socket = dgram.createSocket('udp4'); const socket = dgram
socket.on('message', (msg, rinfo) => { .createSocket('udp4')
try { .on('listening', () => {
if (rinfo.address == config.server_address && rinfo.port == config.server_port) { const address = socket.address();
// from server console.log(`listening ${address.address}:${address.port}`);
const hello: ServerMessage = JSON.parse(msg.toString()); })
server.onMessage(hello, socket, self, peers); .on('message', (msg, rinfo) => {
} else { try {
// from client if (rinfo.address == config.server_address && rinfo.port == config.server_port) {
const hello: PeerHello = JSON.parse(msg.toString()); // from server
assert.ok(hello.id); const message: DownloadMessage = JSON.parse(msg.toString());
const peer = peers.get(hello.id); server.onMessage(socket, message, self);
assert.ok(peer && rinfo.address == peer.address && rinfo.port == config.port); } else {
peer.onMessage(hello); // from client
const message: PeerMessage = JSON.parse(msg.toString());
assert(message.id);
const peer = peers.find(p => p.id === message.id);
assert(peer && rinfo.address == peer.address && rinfo.port == config.port);
peer.onMessage(message);
}
} catch (e) {
console.warn(e);
} }
} catch (e) { });
console.warn(e);
}
});
socket.on('listening', () => { socket.bind(config.port, routers.find(r => r.id == self.id).address);
const address = socket.address();
console.log(`listening ${address.address}:${address.port}`);
});
socket.bind(config.port);
setInterval(() => { setInterval(() => {
self.time = Date.now(); self.time = Date.now();
for (const peer of peers.values()) { const message = JSON.stringify(self);
peer.update(socket, self); for (const peer of peers) {
peer.update(self.time);
socket.send(message, config.port, peer.address);
} }
server.update(socket, self, peers); server.update(socket, self, peers);
self.seq++; self.seq++;
......
src/protocol.ts deleted 100644 → 0
View file @ 688ab046
export interface PeerMeasure {
id: number;
delay: number;
reliability: number;
}
export interface ServerMessage {
id: number;
ack: number;
peers: PeerMeasure[];
}
tsconfig.json
View file @ 7d5ae443
{ {
"compilerOptions": { "compilerOptions": {
"module": "commonjs", "module": "commonjs",
"target": "es2020", "target": "esnext",
"sourceMap": true, "sourceMap": true,
"esModuleInterop": true, "esModuleInterop": true,
"resolveJsonModule": true, "resolveJsonModule": true,
"lib": [] "lib": []
}, }
"exclude": [
"node_modules"
]
} }
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment