Skip to content

Commit 23a5cca

Browse files
streamichstreamich
committed
feat: 🎸 start server and connection implementation
1 parent e20ff0a commit 23a5cca

File tree

9 files changed

+395
-187
lines changed

9 files changed

+395
-187
lines changed

‎src/nfs/v4/FullNfsv4Encoder.ts‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import {Writer} from '@jsonjoy.com/util/lib/buffers/Writer';
1+
import {Writer} from '@jsonjoy.com/buffers/lib/Writer';
22
import {Nfsv4Encoder} from './Nfsv4Encoder';
33
import {RpcMessageEncoder} from '../../rpc/RpcMessageEncoder';
44
import {RmRecordEncoder} from '../../rm/RmRecordEncoder';
@@ -9,12 +9,11 @@ import type * as msg from './messages';
99
import type {IWriter, IWriterGrowable} from '@jsonjoy.com/util/lib/buffers';
1010

1111
export class FullNfsv4Encoder<W extends IWriter & IWriterGrowable = IWriter & IWriterGrowable> {
12-
protected readonly nfsEncoder: Nfsv4Encoder<W>;
13-
protected readonly rpcEncoder: RpcMessageEncoder<W>;
14-
protected readonly rmEncoder: RmRecordEncoder<W>;
12+
public readonly nfsEncoder: Nfsv4Encoder<W>;
13+
public readonly rpcEncoder: RpcMessageEncoder<W>;
14+
public readonly rmEncoder: RmRecordEncoder<W>;
1515

1616
constructor(
17-
public program: number = Nfsv4Const.PROGRAM,
1817
public readonly writer: W = new Writer() as any,
1918
) {
2019
this.nfsEncoder = new Nfsv4Encoder(writer);

‎src/nfs/v4/Nfsv4Connection.ts‎

Lines changed: 175 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,175 @@
1+
import {Reader} from '@jsonjoy.com/buffers/lib/Reader';
2+
import {Nfsv4Decoder} from './Nfsv4Decoder';
3+
import {FullNfsv4Encoder} from './FullNfsv4Encoder';
4+
import {RmRecordDecoder, RmRecordEncoder} from '../../rm';
5+
import {RpcAcceptStat, RpcAuthFlavor, RpcCallMessage, RpcMessage, RpcMessageDecoder, RpcMessageEncoder, RpcOpaqueAuth} from '../../rpc';
6+
import {EMPTY_READER, Nfsv4Proc} from './constants';
7+
import {Nfsv4CompoundRequest} from './messages';
8+
import {getOpNameFromRequest} from './util';
9+
import type {Duplex} from 'node:stream';
10+
import type {IWriter, IWriterGrowable} from '@jsonjoy.com/buffers/lib/types';
11+
12+
export interface Nfsv4ConnectionOpts {
13+
/**
14+
* Normally this is a TCP socket, but any Duplex stream will do.
15+
*/
16+
duplex: Duplex;
17+
encoder?: FullNfsv4Encoder;
18+
decoder?: Nfsv4Decoder;
19+
debug?: boolean;
20+
logger?: Pick<typeof console, 'log' | 'error'>;
21+
}
22+
23+
export class Nfsv4Connection {
24+
public closed = false;
25+
public maxIncomingMessage: number = 2 * 1024 * 1024;
26+
public maxBackpressure: number = 2 * 1024 * 1024;
27+
28+
/** Last known RPC transaction ID. Used to emit fatal connection errors. */
29+
protected lastXid = 0;
30+
31+
public readonly duplex: Duplex;
32+
33+
protected readonly rmDecoder: RmRecordDecoder;
34+
protected readonly rpcDecoder: RpcMessageDecoder;
35+
protected readonly nfsDecoder: Nfsv4Decoder;
36+
protected readonly writer: IWriter & IWriterGrowable;
37+
protected readonly rmEncoder: RmRecordEncoder;
38+
protected readonly rpcEncoder: RpcMessageEncoder;
39+
protected readonly nfsEncoder: FullNfsv4Encoder;
40+
41+
public debug: boolean;
42+
public logger: Pick<typeof console, 'log' | 'error'>;
43+
44+
constructor(opts: Nfsv4ConnectionOpts) {
45+
this.debug = !!opts.debug;
46+
this.logger = opts.logger || console;
47+
const duplex = this.duplex = opts.duplex;
48+
this.rmDecoder = new RmRecordDecoder();
49+
this.rpcDecoder = new RpcMessageDecoder();
50+
this.nfsDecoder = new Nfsv4Decoder();
51+
const nfsEncoder = this.nfsEncoder = new FullNfsv4Encoder();
52+
this.writer = nfsEncoder.writer;
53+
this.rmEncoder = nfsEncoder.rmEncoder;
54+
this.rpcEncoder = nfsEncoder.rpcEncoder;
55+
duplex.on('data', this.onData.bind(this));
56+
duplex.on('timeout', () => this.close());
57+
duplex.on('close', (hadError: boolean): void => {
58+
this.close();
59+
});
60+
duplex.on('error', (err: Error) => {
61+
this.logger.error('SOCKET ERROR:', err);
62+
this.close();
63+
});
64+
}
65+
66+
protected onData(data: Uint8Array): void {
67+
const {rmDecoder, rpcDecoder} = this;
68+
rmDecoder.push(data);
69+
let record = rmDecoder.readRecord();
70+
while (record) {
71+
if (record.size()) {
72+
const rpcMessage = rpcDecoder.decodeMessage(record);
73+
if (rpcMessage) this.onRpcMessage(rpcMessage);
74+
else {
75+
this.close();
76+
return;
77+
}
78+
}
79+
record = rmDecoder.readRecord();
80+
}
81+
}
82+
83+
protected onRpcMessage(msg: RpcMessage): void {
84+
const debug = this.debug;
85+
if (msg instanceof RpcCallMessage) {
86+
const proc = msg.proc;
87+
switch (proc) {
88+
case Nfsv4Proc.NULL: {
89+
if (debug) this.logger.log('NULL procedure');
90+
const rmEncoder = this.rmEncoder;
91+
const state = rmEncoder.startRmRecord();
92+
this.rpcEncoder.writeAcceptedReply(
93+
msg.xid,
94+
new RpcOpaqueAuth(RpcAuthFlavor.AUTH_NONE, EMPTY_READER),
95+
RpcAcceptStat.SUCCESS,
96+
);
97+
rmEncoder.endRmRecord(state);
98+
this.write(this.writer.flush());
99+
return;
100+
}
101+
case Nfsv4Proc.COMPOUND: {
102+
if (!(msg.params instanceof Reader)) return;
103+
const compound = this.nfsDecoder.decodeCompoundRequest(msg.params);
104+
if (compound instanceof Nfsv4CompoundRequest) {
105+
console.log('\nNFS COMPOUND Request:');
106+
console.log(` Tag: "${compound.tag}"`);
107+
console.log(` Minor Version: ${compound.minorversion}`);
108+
console.log(` Operations (${compound.argarray.length}):`);
109+
compound.argarray.forEach((op: any, idx: number) => {
110+
console.log(` [${idx}] ${getOpNameFromRequest(op)}`);
111+
console.log(` ${JSON.stringify(op, null, 2).split('\n').slice(1).join('\n ')}`);
112+
});
113+
} else {
114+
console.log('Could not decode COMPOUND request');
115+
}
116+
return;
117+
}
118+
default: {
119+
console.log(`Unknown procedure: ${proc}`);
120+
}
121+
}
122+
}
123+
throw new Error('Not implemented non-RPCCallMessage');
124+
}
125+
126+
private closeWithError(error: RpcAcceptStat.PROG_UNAVAIL | RpcAcceptStat.PROC_UNAVAIL | RpcAcceptStat.GARBAGE_ARGS | RpcAcceptStat.SYSTEM_ERR): void {
127+
const xid = this.lastXid;
128+
if (xid) {
129+
const state = this.rmEncoder.startRmRecord();
130+
const verify = new RpcOpaqueAuth(RpcAuthFlavor.AUTH_NONE, EMPTY_READER);
131+
this.rpcEncoder.writeAcceptedReply(xid, verify, error);
132+
this.rmEncoder.endRmRecord(state);
133+
const bin = this.writer.flush();
134+
this.duplex.write(bin);
135+
}
136+
this.close();
137+
}
138+
139+
private close(): void {
140+
if (this.closed) return;
141+
this.closed = true;
142+
clearImmediate(this.__uncorkTimer);
143+
this.__uncorkTimer = null;
144+
const duplex = this.duplex;
145+
duplex.removeAllListeners();
146+
if (!duplex.destroyed) duplex.destroy();
147+
}
148+
149+
// ---------------------------------------------------------- Write to socket
150+
151+
private __uncorkTimer: any = null;
152+
153+
public write(buf: Uint8Array): void {
154+
if (this.closed) return;
155+
const duplex = this.duplex;
156+
if (duplex.writableLength > this.maxBackpressure) {
157+
this.closeWithError(RpcAcceptStat.SYSTEM_ERR);
158+
return;
159+
}
160+
const __uncorkTimer = this.__uncorkTimer;
161+
if (!__uncorkTimer) duplex.cork();
162+
duplex.write(buf);
163+
if (!__uncorkTimer) this.__uncorkTimer = setImmediate(() => {
164+
this.__uncorkTimer = null;
165+
duplex.uncork();
166+
});
167+
}
168+
169+
// ------------------------------------------------- Write WebSocket messages
170+
171+
// TODO: Execute NFS Callback...
172+
public send(): void {
173+
174+
}
175+
}

‎src/nfs/v4/Nfsv4Decoder.ts‎

Lines changed: 8 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -16,25 +16,16 @@ export class Nfsv4Decoder {
1616
reader: Reader,
1717
isRequest: boolean,
1818
): msg.Nfsv4CompoundRequest | msg.Nfsv4CompoundResponse | undefined {
19-
this.xdr.reader = reader;
20-
const startPos = reader.x;
21-
try {
22-
if (isRequest) {
23-
return this.decodeCompoundRequest();
24-
} else {
25-
return this.decodeCompoundResponse();
26-
}
27-
} catch (err) {
28-
if (err instanceof RangeError) {
29-
reader.x = startPos;
30-
return undefined;
31-
}
32-
throw err;
19+
if (isRequest) {
20+
return this.decodeCompoundRequest(reader);
21+
} else {
22+
return this.decodeCompoundResponse(reader);
3323
}
3424
}
3525

36-
private decodeCompoundRequest(): msg.Nfsv4CompoundRequest {
26+
public decodeCompoundRequest(reader: Reader): msg.Nfsv4CompoundRequest {
3727
const xdr = this.xdr;
28+
xdr.reader = reader;
3829
const tag = xdr.readString();
3930
const minorversion = xdr.readUnsignedInt();
4031
const argarray: msg.Nfsv4Request[] = [];
@@ -47,8 +38,9 @@ export class Nfsv4Decoder {
4738
return new msg.Nfsv4CompoundRequest(tag, minorversion, argarray);
4839
}
4940

50-
private decodeCompoundResponse(): msg.Nfsv4CompoundResponse {
41+
public decodeCompoundResponse(reader: Reader): msg.Nfsv4CompoundResponse {
5142
const xdr = this.xdr;
43+
xdr.reader = reader;
5244
const status = xdr.readUnsignedInt();
5345
const tag = xdr.readString();
5446
const resarray: msg.Nfsv4Response[] = [];

‎src/nfs/v4/Nfsv4TcpServer.ts‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
import * as net from 'net';
2+
import {Nfsv4Connection} from './Nfsv4Connection';
3+
4+
/* tslint:disable:no-console */
5+
6+
const PORT = Number(process.env.NFS_PORT) || Number(process.env.PORT) || 2049;
7+
const HOST = process.env.NFS_HOST
8+
? String(process.env.NFS_HOST)
9+
: process.env.HOST ? String(process.env.HOST) : '127.0.0.1';
10+
11+
export interface Nfsv4TcpServerOpts {
12+
port?: number;
13+
host?: string;
14+
debug?: boolean;
15+
logger?: Pick<typeof console, 'log' | 'error'>;
16+
onError?: (err: Error) => void;
17+
stopOnSigint?: boolean;
18+
}
19+
20+
export class Nfsv4TcpServer {
21+
public static start(opts: Nfsv4TcpServerOpts = {}): void {
22+
const server = new Nfsv4TcpServer(opts);
23+
server.start().catch(console.error);
24+
}
25+
26+
public readonly server: net.Server;
27+
public port: number = PORT;
28+
public host: string = HOST;
29+
public debug: boolean = false;
30+
public logger: Pick<typeof console, 'log' | 'error'> = console;
31+
private sigintHandler?: () => void;
32+
33+
constructor(opts: Nfsv4TcpServerOpts = {}) {
34+
this.port = opts.port ?? PORT;
35+
this.host = opts.host ?? HOST;
36+
this.debug = opts.debug ?? false;
37+
this.logger = opts.logger ?? console;
38+
const server = this.server = new net.Server();
39+
server.on('connection', (socket) => {
40+
if (this.debug) this.logger.log('New connection from', socket.remoteAddress, 'port', socket.remotePort);
41+
new Nfsv4Connection({
42+
duplex: socket,
43+
debug: this.debug,
44+
logger: this.logger,
45+
})
46+
});
47+
server.on('error', opts.onError ?? ((err) => {
48+
if (this.debug) this.logger.error('Server error:', err.message);
49+
process.exit(1);
50+
}));
51+
if (opts.stopOnSigint ?? true) {
52+
this.sigintHandler = () => {
53+
if (this.debug) this.logger.log('\nShutting down NFSv4 server...');
54+
this.cleanup();
55+
process.exit(0);
56+
};
57+
process.on('SIGINT', this.sigintHandler);
58+
}
59+
}
60+
61+
private cleanup(): void {
62+
if (this.sigintHandler) {
63+
process.off('SIGINT', this.sigintHandler);
64+
this.sigintHandler = undefined;
65+
}
66+
this.server.close((err) => {
67+
if (this.debug && err) this.logger.error('Error closing server:', err);
68+
});
69+
}
70+
71+
public stop(): Promise<void> {
72+
return new Promise((resolve) => {
73+
this.cleanup();
74+
this.server.close(() => {
75+
if (this.debug) this.logger.log('NFSv4 server closed');
76+
resolve();
77+
});
78+
});
79+
}
80+
81+
public start(port: number = this.port, host: string = this.host): Promise<void> {
82+
if (this.debug) this.logger.log(`Starting NFSv4 TCP server on ${host}:${port}...`);
83+
return new Promise((resolve, reject) => {
84+
const onError = (err: unknown) => reject(err);
85+
const server = this.server;
86+
server.on('error', onError);
87+
server.listen(port, host, () => {
88+
if (this.debug) this.logger.log(`NFSv4 TCP server listening on ${host}:${port}`);
89+
server.off('error', onError);
90+
resolve();
91+
});
92+
});
93+
}
94+
}

‎src/nfs/v4/__demos__/README.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,12 @@ Then mount an NFSv4 share from another terminal or machine:
4040
mount -t nfs -o vers=4,nfsvers=4,port=8777,mountport=8777,proto=tcp,sec=none 127.0.0.1:/export ~/mnt/test
4141
```
4242

43+
You might need to clean all hanging `mount_nfs` processes if previous mounts failed.
44+
45+
```bash
46+
sudo pkill -9 -f "ts-node.*tcp-server"; sudo pkill -9 mount_nfs
47+
```
48+
4349
## NFSv4 Protocol Structure
4450

4551
NFSv4 differs from NFSv3 in that it uses COMPOUND procedures to bundle multiple operations:

‎src/nfs/v4/__demos__/tcp-client.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,10 @@ import {Nfsv4Proc} from '../constants';
77

88
/* tslint:disable:no-console */
99

10-
const PORT = Number(process.env.PORT) || 2049;
11-
const HOST = '127.0.0.1';
10+
const PORT = Number(process.env.NFS_PORT) || Number(process.env.PORT) || 2049;
11+
const HOST = process.env.NFS_HOST
12+
? String(process.env.NFS_HOST)
13+
: process.env.HOST ? String(process.env.HOST) : '127.0.0.1';
1214

1315
const createTestCompoundRequest = (): Nfsv4CompoundRequest => {
1416
const fhData = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8]);

0 commit comments

Comments
 (0)