mirror of
https://github.com/vercel-labs/scriptc.git
synced 2026-11-08 08:44:37 +08:00
Answer net.Socket write() with Node's backpressure boolean, and fire 'drain' (#2198)
This commit is contained in:
Chris Tate
@@ -72,7 +72,7 @@ describe.runIf(supported)("macOS arm64 native link info", () => {
|
||||
minimum_os: "14.0",
|
||||
},
|
||||
program: { object, entry_symbol: "main" },
|
||||
runtime_abi: { version: 44, marker: "scr_runtime_abi_v44" },
|
||||
runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" },
|
||||
runtime_pack: {
|
||||
kind: "precompiled",
|
||||
package: "@scriptc/runtime-darwin-arm64",
|
||||
@@ -165,6 +165,6 @@ describe.runIf(supported)("macOS arm64 native link info", () => {
|
||||
() => null,
|
||||
(failure: { stderr?: string }) => failure,
|
||||
);
|
||||
expect(error?.stderr).toContain("scr_runtime_abi_v44");
|
||||
expect(error?.stderr).toContain("scr_runtime_abi_v45");
|
||||
});
|
||||
});
|
||||
|
||||
+14
-5
@@ -5345,10 +5345,10 @@ declare module "node:zlib" {
|
||||
* with the real port discoverable as address().port (the composed read
|
||||
* is the lowered form — bare address() has no lowering); close([cb])
|
||||
* firing once connections drain; socket write/end/destroy/pipe and the
|
||||
* data/end/close/error/connect events (on and once — listeners take at
|
||||
* most the declared parameter). write/end return void here (Node returns
|
||||
* the backpressure boolean / the socket); on/once return void (chaining
|
||||
* is fenced). connect's host is a numeric IP or "localhost" (pinned to
|
||||
* data/end/close/error/connect/drain events (on and once — listeners
|
||||
* take at most the declared parameter). write returns Node's
|
||||
* backpressure boolean; end returns void here (Node returns the socket);
|
||||
* on/once return void (chaining is fenced). connect's host is a numeric IP or "localhost" (pinned to
|
||||
* 127.0.0.1 — SEMANTICS.md); every failure is the async 'error' event,
|
||||
* and an 'error' with no listener exits 1 like an unhandled EventEmitter
|
||||
* 'error'. */
|
||||
@@ -5380,7 +5380,13 @@ declare module "net" {
|
||||
readonly bytesWritten: number;
|
||||
/* write's callback runs once the data is flushed, with null; end's
|
||||
* once the socket finishes. */
|
||||
write(data: string | Uint8Array, callback?: (err?: Error | null) => void): void;
|
||||
/* false when the bytes this write left unsent reach the 65536-byte
|
||||
* high-water mark, as in Node; 'drain' fires when they are all sent. */
|
||||
write(data: string | Uint8Array, callback?: (err?: Error | null) => void): boolean;
|
||||
/* the bytes written and not yet sent */
|
||||
readonly writableLength: number;
|
||||
/* a write answered false and 'drain' has not fired yet */
|
||||
readonly writableNeedDrain: boolean;
|
||||
end(callback?: () => void): void;
|
||||
end(data: string | Uint8Array, callback?: () => void): void;
|
||||
destroy(): void;
|
||||
@@ -5428,6 +5434,7 @@ declare module "net" {
|
||||
| "connect"
|
||||
| "ready"
|
||||
| "timeout"
|
||||
| "drain"
|
||||
| "readable"
|
||||
| "finish"
|
||||
| "secureConnect"
|
||||
@@ -5452,6 +5459,7 @@ declare module "net" {
|
||||
| "connect"
|
||||
| "ready"
|
||||
| "timeout"
|
||||
| "drain"
|
||||
| "readable"
|
||||
| "finish"
|
||||
| "secureConnect"
|
||||
@@ -5475,6 +5483,7 @@ declare module "net" {
|
||||
| "connect"
|
||||
| "ready"
|
||||
| "timeout"
|
||||
| "drain"
|
||||
| "readable"
|
||||
| "finish"
|
||||
| "secureConnect"
|
||||
|
||||
@@ -309,6 +309,7 @@ export function emitNetworkHttpLibCall(host: LlvmEmitterContext, e: LibCallExpr)
|
||||
e.fn === "net.sockOnClose" ||
|
||||
e.fn === "net.sockOnConnect" ||
|
||||
e.fn === "net.sockOnTimeout" ||
|
||||
e.fn === "net.sockOnDrain" ||
|
||||
e.fn === "net.sockOnReadable" ||
|
||||
e.fn === "net.sockOnReady" ||
|
||||
e.fn === "http.reqOnEnd" ||
|
||||
@@ -329,6 +330,7 @@ export function emitNetworkHttpLibCall(host: LlvmEmitterContext, e: LibCallExpr)
|
||||
"net.sockOnClose": "scr_net_sock_on_close",
|
||||
"net.sockOnConnect": "scr_net_sock_on_connect",
|
||||
"net.sockOnTimeout": "scr_net_sock_on_timeout",
|
||||
"net.sockOnDrain": "scr_net_sock_on_drain",
|
||||
"net.sockOnReadable": "scr_net_sock_on_readable",
|
||||
"net.sockOnReady": "scr_net_sock_on_ready",
|
||||
"http.reqOnEnd": "scr_http_req_on_end",
|
||||
|
||||
@@ -1337,6 +1337,8 @@ export const LIB_FN_SYMS: Record<string, string> = {
|
||||
"net.sockResetAndDestroy": "scr_net_sock_reset_and_destroy",
|
||||
"net.sockAddress": "scr_net_sock_address",
|
||||
"net.sockBytesWritten": "scr_net_sock_bytes_written",
|
||||
"net.sockWritableLength": "scr_net_sock_writable_length",
|
||||
"net.sockWritableNeedDrain": "scr_net_sock_writable_need_drain",
|
||||
"net.sockAddrField": "scr_net_sock_addr_field",
|
||||
"net.sockBytesRead": "scr_net_sock_bytes_read",
|
||||
"net.sockConnecting": "scr_net_sock_connecting",
|
||||
|
||||
@@ -2,8 +2,8 @@
|
||||
* of the link contract: a program object keeps this symbol undefined and the
|
||||
* matching runtime defines it, so incompatible manual links fail before the
|
||||
* program can start. */
|
||||
export const RUNTIME_ABI_VERSION = 44 as const;
|
||||
export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v44" as const;
|
||||
export const RUNTIME_ABI_VERSION = 45 as const;
|
||||
export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v45" as const;
|
||||
|
||||
/** Object consumption is useful today, but the complete scr_* surface may
|
||||
* still change before 1.0. The versioned marker prevents accidental mixing;
|
||||
|
||||
@@ -52,7 +52,7 @@ async function fixture(target: NativeTargetSpec, workers = false) {
|
||||
object_format: target.objectFormat,
|
||||
minimum_os: target.minimumOs,
|
||||
},
|
||||
runtime_abi: { version: 44, marker: "scr_runtime_abi_v44" },
|
||||
runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" },
|
||||
compiler: { command: "fixture", identity: "fixture", target: target.llvmTriple },
|
||||
macros: { executable: [], excluded: ["SCR_LIB"], sanitizer: "external-toolchain-required" },
|
||||
flavors: {
|
||||
|
||||
@@ -129,7 +129,7 @@ async function fixture(target: NativeTargetSpec = MACOS_ARM64_TARGET) {
|
||||
object_format: target.objectFormat,
|
||||
minimum_os: target.minimumOs,
|
||||
},
|
||||
runtime_abi: { version: 44, marker: "scr_runtime_abi_v44" },
|
||||
runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" },
|
||||
compiler: {
|
||||
command: "clang",
|
||||
identity: "fixture clang",
|
||||
|
||||
@@ -2425,7 +2425,9 @@ function lowerNetSocketMethodCall(
|
||||
};
|
||||
}
|
||||
if (name === "write" || name === "end") {
|
||||
requireStatementPosition(lowerer, call, `socket.${name}(...)`);
|
||||
// write() answers Node's backpressure boolean; end() stays a statement
|
||||
if (name === "end") requireStatementPosition(lowerer, call, `socket.${name}(...)`);
|
||||
const resultT = name === "write" ? BOOL : VOID;
|
||||
// write(chunk, encoding) — the two-argument encoding form: 'buffer'
|
||||
// beside a string chunk is Writable's encoding check (a synchronous
|
||||
// ERR_UNKNOWN_ENCODING TypeError, "Unknown encoding: buffer" — the
|
||||
@@ -2451,7 +2453,7 @@ function lowerNetSocketMethodCall(
|
||||
const socket = coerceToHandle(lowerer, access.expression, NETSOCKET_T);
|
||||
const chunk = lowerer.lowerExpr(args[0]!);
|
||||
const fn: IrLibFn = chunk.type.kind === "string" ? "net.sockWrite" : "net.sockWriteBytes";
|
||||
return { kind: "libCall", fn, args: [socket, chunk], type: VOID, loc };
|
||||
return { kind: "libCall", fn, args: [socket, chunk], type: BOOL, loc };
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2510,7 +2512,7 @@ function lowerNetSocketMethodCall(
|
||||
NARROW_DATA_HINT,
|
||||
);
|
||||
}
|
||||
main = { kind: "libCall", fn, args: [receiver(), data], type: VOID, loc };
|
||||
main = { kind: "libCall", fn, args: [receiver(), data], type: resultT, loc };
|
||||
}
|
||||
const cb = lowerer.lowerExpr(last);
|
||||
let register: IrExpr;
|
||||
@@ -2531,8 +2533,24 @@ function lowerNetSocketMethodCall(
|
||||
loc,
|
||||
};
|
||||
}
|
||||
stmts.push({ kind: "exprStmt", expr: main, loc });
|
||||
return { kind: "seqExpr", stmts, result: register, type: VOID, loc };
|
||||
if (resultT.kind === "void") {
|
||||
stmts.push({ kind: "exprStmt", expr: main, loc });
|
||||
return { kind: "seqExpr", stmts, result: register, type: VOID, loc };
|
||||
}
|
||||
// write(data, cb): the data is written, the callback registered, and
|
||||
// the write's answer is the value
|
||||
const okLocal = lowerer.declareHiddenLocal("%sockWriteOk", BOOL);
|
||||
stmts.push(
|
||||
{ kind: "varDecl", localId: okLocal.id, init: main, loc },
|
||||
{ kind: "exprStmt", expr: register, loc },
|
||||
);
|
||||
return {
|
||||
kind: "seqExpr",
|
||||
stmts,
|
||||
result: { kind: "varRef", localId: okLocal.id, type: BOOL, loc },
|
||||
type: BOOL,
|
||||
loc,
|
||||
};
|
||||
}
|
||||
const maxArgs = name === "write" ? 1 : 1;
|
||||
const minArgs = name === "write" ? 1 : 0;
|
||||
@@ -2552,17 +2570,17 @@ function lowerNetSocketMethodCall(
|
||||
const data = lowerer.lowerExpr(args[0]!);
|
||||
if (data.type.kind === "string") {
|
||||
const fn: IrLibFn = name === "write" ? "net.sockWrite" : "net.sockEndStr";
|
||||
return { kind: "libCall", fn, args: [receiver, data], type: VOID, loc };
|
||||
return { kind: "libCall", fn, args: [receiver, data], type: resultT, loc };
|
||||
}
|
||||
if (data.type.kind === "bytes" && data.type.elem === "u8") {
|
||||
const fn: IrLibFn = name === "write" ? "net.sockWriteBytes" : "net.sockEndBytes";
|
||||
return { kind: "libCall", fn, args: [receiver, data], type: VOID, loc };
|
||||
return { kind: "libCall", fn, args: [receiver, data], type: resultT, loc };
|
||||
}
|
||||
if (data.type.kind === "dyn") {
|
||||
// An untyped JS payload (a helper's parameter): the runtime
|
||||
// dispatches STR/BYTES and throws Node's chunk TypeError otherwise.
|
||||
const fn: IrLibFn = name === "write" ? "net.sockWriteDyn" : "net.sockEndDyn";
|
||||
return { kind: "libCall", fn, args: [receiver, data], type: VOID, loc };
|
||||
return { kind: "libCall", fn, args: [receiver, data], type: resultT, loc };
|
||||
}
|
||||
lowerer.noLowering(
|
||||
`${name} of '${lowerer.fmt(data.type)}' data`,
|
||||
@@ -2944,6 +2962,7 @@ function lowerNetSocketMethodCall(
|
||||
event === "connect" ||
|
||||
event === "ready" ||
|
||||
event === "timeout" ||
|
||||
event === "drain" ||
|
||||
event === "readable"
|
||||
) {
|
||||
const { cb } = lowerCallbackArg(
|
||||
@@ -2962,11 +2981,13 @@ function lowerNetSocketMethodCall(
|
||||
? "net.sockOnClose"
|
||||
: event === "timeout"
|
||||
? "net.sockOnTimeout"
|
||||
: event === "readable"
|
||||
? "net.sockOnReadable"
|
||||
: event === "ready"
|
||||
? "net.sockOnReady"
|
||||
: "net.sockOnConnect";
|
||||
: event === "drain"
|
||||
? "net.sockOnDrain"
|
||||
: event === "readable"
|
||||
? "net.sockOnReadable"
|
||||
: event === "ready"
|
||||
? "net.sockOnReady"
|
||||
: "net.sockOnConnect";
|
||||
return { kind: "libCall", fn, args: [receiver, cb, once], type: VOID, loc };
|
||||
}
|
||||
if (event === "lookup") {
|
||||
@@ -3957,6 +3978,18 @@ export function lowerServerProperty(
|
||||
const receiver = coerceToHandle(lowerer, expr.expression, NETSOCKET_T);
|
||||
return { kind: "libCall", fn: "net.sockBytesWritten", args: [receiver], type: F64, loc };
|
||||
}
|
||||
if (
|
||||
recvKind === "netSocket" &&
|
||||
lowerer.isStdlibMember(expr) &&
|
||||
(expr.name.text === "writableLength" || expr.name.text === "writableNeedDrain")
|
||||
) {
|
||||
// The write buffer's unsent bytes, and whether a write answered false
|
||||
// and 'drain' has not fired yet.
|
||||
const receiver = coerceToHandle(lowerer, expr.expression, NETSOCKET_T);
|
||||
return expr.name.text === "writableLength"
|
||||
? { kind: "libCall", fn: "net.sockWritableLength", args: [receiver], type: F64, loc }
|
||||
: { kind: "libCall", fn: "net.sockWritableNeedDrain", args: [receiver], type: BOOL, loc };
|
||||
}
|
||||
if (recvKind === "netSocket" && lowerer.isStdlibMember(expr) && expr.name.text === "readable") {
|
||||
// true until the read half is done (peer FIN / destroy).
|
||||
const receiver = coerceToHandle(lowerer, expr.expression, NETSOCKET_T);
|
||||
|
||||
@@ -1547,15 +1547,18 @@ function buildLibFnSignatures() {
|
||||
* true } and a callback taking an address string or list, then dials
|
||||
* the answers in order. MAY THROW (host validation, a lookup throw). */
|
||||
"net.connectLookupDyn": { argTypes: [F64, STRING, DYN], result: NETSOCKET_T },
|
||||
"net.sockWrite": { argTypes: [NETSOCKET_T, STRING], result: VOID },
|
||||
"net.sockWriteBytes": { argTypes: [NETSOCKET_T, BYTES_U8], result: VOID },
|
||||
/** socket.write: Node's Writable answer — false when the writes not yet
|
||||
* completely sent, each counted in full, reach the 65536-byte
|
||||
* high-water mark ('drain' follows), and on an ended or closed socket. */
|
||||
"net.sockWrite": { argTypes: [NETSOCKET_T, STRING], result: BOOL },
|
||||
"net.sockWriteBytes": { argTypes: [NETSOCKET_T, BYTES_U8], result: BOOL },
|
||||
"net.sockEnd": { argTypes: [NETSOCKET_T], result: VOID },
|
||||
"net.sockEndStr": { argTypes: [NETSOCKET_T, STRING], result: VOID },
|
||||
"net.sockEndBytes": { argTypes: [NETSOCKET_T, BYTES_U8], result: VOID },
|
||||
/** write/end with a CHECKED-DYNAMIC chunk (an untyped JS payload into a
|
||||
* typed socket): the runtime dispatches STR/BYTES and throws Node's
|
||||
* ERR_INVALID_ARG_TYPE chunk TypeError on any other kind. */
|
||||
"net.sockWriteDyn": { argTypes: [NETSOCKET_T, DYN], result: VOID },
|
||||
"net.sockWriteDyn": { argTypes: [NETSOCKET_T, DYN], result: BOOL },
|
||||
"net.sockEndDyn": { argTypes: [NETSOCKET_T, DYN], result: VOID },
|
||||
"net.sockDestroy": { argTypes: [NETSOCKET_T], result: VOID },
|
||||
"net.sockPipe": { argTypes: [NETSOCKET_T, NETSOCKET_T], result: VOID },
|
||||
@@ -2013,6 +2016,12 @@ function buildLibFnSignatures() {
|
||||
argTypes: [NETSOCKET_T, { kind: "func", params: [], ret: VOID }, BOOL],
|
||||
result: VOID,
|
||||
},
|
||||
/** on('drain') / once('drain'): fires after a write answered false, once
|
||||
* the write buffer empties. */
|
||||
"net.sockOnDrain": {
|
||||
argTypes: [NETSOCKET_T, { kind: "func", params: [], ret: VOID }, BOOL],
|
||||
result: VOID,
|
||||
},
|
||||
"net.sockRemoteAddress": { argTypes: [NETSOCKET_T], result: VOID },
|
||||
/** socket.encrypted — `boolean | undefined`: the true arm iff the
|
||||
* socket carries a TLS transport (Node types `encrypted: true` on
|
||||
@@ -2139,6 +2148,12 @@ function buildLibFnSignatures() {
|
||||
/** socket.address(): { address, family, port } or {} as a dyn. */
|
||||
"net.sockAddress": { argTypes: [NETSOCKET_T], result: DYN },
|
||||
"net.sockBytesWritten": { argTypes: [NETSOCKET_T], result: F64 },
|
||||
/** socket.writableLength: Node's state.length, the length of every
|
||||
* write not yet completely sent. */
|
||||
"net.sockWritableLength": { argTypes: [NETSOCKET_T], result: F64 },
|
||||
/** socket.writableNeedDrain: a write answered false and 'drain' has not
|
||||
* fired, with the write half open. */
|
||||
"net.sockWritableNeedDrain": { argTypes: [NETSOCKET_T], result: BOOL },
|
||||
/** remotePort (0), remoteFamily (1), localAddress (2), localPort (3),
|
||||
* localFamily (4): dyn values, undefined when unknown. */
|
||||
"net.sockAddrField": { argTypes: [NETSOCKET_T, F64], result: DYN },
|
||||
|
||||
@@ -378,7 +378,7 @@ done:
|
||||
expect(await symbols(helperObject, ["-u"])).toEqual(await symbols(clangObject, ["-u"]));
|
||||
expect(await symbols(helperObject, ["-gU"])).toEqual(await symbols(clangObject, ["-gU"]));
|
||||
expect(await symbols(helperObject, ["-gU"])).toEqual(["0000000000000000 T _main"]);
|
||||
expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v44");
|
||||
expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v45");
|
||||
|
||||
const clangExe = join(dir, "clang-program");
|
||||
const linkDriver = join(dir, "clang-driver.c");
|
||||
@@ -712,7 +712,7 @@ done:
|
||||
.split("\n");
|
||||
expect(undefinedSymbols).toContain("_sf_scale");
|
||||
expect(undefinedSymbols).toContain("_sf_callback_mix");
|
||||
expect(undefinedSymbols).toContain("_scr_runtime_abi_v44");
|
||||
expect(undefinedSymbols).toContain("_scr_runtime_abi_v45");
|
||||
});
|
||||
|
||||
test("helper and clang object paths retain the same outbound FFI C ABI", async () => {
|
||||
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/7515-net-socket-backpressure.ts"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/7519-net-socket-drain-listener-close.ts"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
@@ -366,7 +366,7 @@ async function build() {
|
||||
package: packageManifest.name,
|
||||
version: packageManifest.version,
|
||||
target: RUNTIME_PACK_MATRIX.target,
|
||||
runtime_abi: { version: 44, marker: "scr_runtime_abi_v44" },
|
||||
runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" },
|
||||
compiler: {
|
||||
command: compiler,
|
||||
identity: compilerVersion,
|
||||
|
||||
@@ -469,7 +469,7 @@ async function build() {
|
||||
package: packageManifest.name,
|
||||
version: packageManifest.version,
|
||||
target: matrix.target,
|
||||
runtime_abi: { version: 44, marker: "scr_runtime_abi_v44" },
|
||||
runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" },
|
||||
compiler: {
|
||||
command: compiler,
|
||||
identity: compilerVersion,
|
||||
|
||||
@@ -148,7 +148,7 @@ static void scr_enable_utf8_console_output(void) {
|
||||
}
|
||||
#endif
|
||||
|
||||
void scr_runtime_abi_v44(void) {}
|
||||
void scr_runtime_abi_v45(void) {}
|
||||
|
||||
void scr_init(void) {
|
||||
#ifdef SCR_WORKERS
|
||||
|
||||
@@ -81,7 +81,7 @@ void scr_init(void);
|
||||
/* Program objects emitted by the bundled LLVM helper reference this symbol.
|
||||
* Its versioned spelling makes a mismatched manual runtime link fail before
|
||||
* the program can start. */
|
||||
void scr_runtime_abi_v44(void);
|
||||
void scr_runtime_abi_v45(void);
|
||||
void scr_runtime_workers_v8(void);
|
||||
|
||||
/* ── the trap funnel (scr_console.c; scr_library.c under -DSCR_LIB) ──────
|
||||
|
||||
+153
-24
@@ -516,6 +516,16 @@ struct ScrNetSocket {
|
||||
/* write buffer: [whead, wlen) of wbuf is unsent */
|
||||
char *wbuf;
|
||||
size_t whead, wlen, wcap;
|
||||
/* Node's writable state.length: each write with bytes in the buffer
|
||||
* counts in full until its last byte is sent (a partly sent write
|
||||
* included). wq[wq_head, wq_n) holds those writes in order, each with
|
||||
* the offset where its bytes end, counted from wbase, the offset of
|
||||
* wbuf[0] in everything buffered since the buffer last emptied. */
|
||||
struct ScrNetPendingWrite {
|
||||
size_t end, len;
|
||||
} *wq;
|
||||
size_t wq_head, wq_n, wq_cap;
|
||||
size_t wbase, wpending;
|
||||
/* receive buffer: [rhead, rlen) of rbuf is arrived-but-unconsumed —
|
||||
* fed by paused-mode reads ('readable' consumers) and unshift; drained
|
||||
* FIRST by every consumer (data listeners, the parser attach, the TLS
|
||||
@@ -565,6 +575,11 @@ struct ScrNetSocket {
|
||||
ScrNetLs lookup_ls;
|
||||
struct ScrNetLookupEv *lookup_evs;
|
||||
ScrNetLs data_ls, end_ls, close_ls, err_ls, conn_ls, timeout_ls, readable_ls;
|
||||
/* Writable backpressure: a user write answered false (the bytes it left
|
||||
* unsent reached SCR_NET_WRITE_HWM), so 'drain' fires once the write
|
||||
* buffer empties */
|
||||
ScrNetLs drain_ls;
|
||||
bool need_drain;
|
||||
/* Flow control (pause()/resume()): user_paused holds reads OFF even
|
||||
* with consumers (kernel/TCP backpressure is the buffer); flowing
|
||||
* counts as a consumer even with no 'data' listener
|
||||
@@ -747,6 +762,7 @@ void scr_net_sock_release(ScrNetSocket *s) {
|
||||
scr_net_ls_drop(&s->conn_ls);
|
||||
scr_net_ls_drop(&s->timeout_ls);
|
||||
scr_net_ls_drop(&s->readable_ls);
|
||||
scr_net_ls_drop(&s->drain_ls);
|
||||
scr_net_ls_drop(&s->finish_ls);
|
||||
scr_net_ls_drop(&s->wcb_ls);
|
||||
scr_net_ls_drop(&s->ready_ls);
|
||||
@@ -764,6 +780,7 @@ void scr_net_sock_release(ScrNetSocket *s) {
|
||||
if (s->lookup_evs != NULL) scr_net_lookup_ops->free(s);
|
||||
scr_net_tunnel_free(s->tunnel);
|
||||
free(s->wbuf);
|
||||
free(s->wq);
|
||||
free(s->rbuf);
|
||||
if (s->pipe_dst) scr_net_sock_release(s->pipe_dst);
|
||||
if (s->conn_pending) scr_net_server_release(s->conn_pending);
|
||||
@@ -1082,13 +1099,34 @@ static void scr_net_sock_maybe_finish_write(ScrNetSocket *s) {
|
||||
else scr_net_sock_update_read(s); /* the drain-to-EOF mode arms here */
|
||||
}
|
||||
|
||||
/* The write buffer emptied (or its bytes moved elsewhere): every pending
|
||||
* write is complete. */
|
||||
static void scr_net_sock_wq_clear(ScrNetSocket *s) {
|
||||
s->whead = s->wlen = 0;
|
||||
s->wq_head = s->wq_n = 0;
|
||||
s->wbase = 0;
|
||||
s->wpending = 0;
|
||||
}
|
||||
|
||||
/* Completes the pending writes whose last byte the flush has sent. */
|
||||
static void scr_net_sock_wq_settle(ScrNetSocket *s) {
|
||||
size_t sent_to = s->wbase + s->whead;
|
||||
while (s->wq_head < s->wq_n && s->wq[s->wq_head].end <= sent_to) {
|
||||
s->wpending -= s->wq[s->wq_head].len;
|
||||
s->wq_head++;
|
||||
}
|
||||
if (s->wq_head == s->wq_n) s->wq_head = s->wq_n = 0;
|
||||
}
|
||||
|
||||
/* Append bytes to the write buffer (post-immediate-write remainder, or
|
||||
* everything while connecting/buffered). */
|
||||
static void scr_net_sock_buffer(ScrNetSocket *s, const char *data, size_t len) {
|
||||
* everything while connecting/buffered). `counted` is the length of the
|
||||
* write they belong to, which counts in full until they are sent. */
|
||||
static void scr_net_sock_buffer(ScrNetSocket *s, const char *data, size_t len, size_t counted) {
|
||||
if (len == 0) return;
|
||||
/* compact the consumed head first */
|
||||
if (s->whead > 0) {
|
||||
memmove(s->wbuf, s->wbuf + s->whead, s->wlen - s->whead);
|
||||
s->wbase += s->whead;
|
||||
s->wlen -= s->whead;
|
||||
s->whead = 0;
|
||||
}
|
||||
@@ -1101,6 +1139,63 @@ static void scr_net_sock_buffer(ScrNetSocket *s, const char *data, size_t len) {
|
||||
}
|
||||
memcpy(s->wbuf + s->wlen, data, len);
|
||||
s->wlen += len;
|
||||
if (s->wq_n == s->wq_cap && s->wq_head > 0) {
|
||||
/* reclaim the completed writes before growing: a buffer that never
|
||||
* empties completes old writes while new ones arrive */
|
||||
memmove(s->wq, s->wq + s->wq_head, (s->wq_n - s->wq_head) * sizeof *s->wq);
|
||||
s->wq_n -= s->wq_head;
|
||||
s->wq_head = 0;
|
||||
}
|
||||
if (s->wq_n == s->wq_cap) {
|
||||
s->wq_cap = s->wq_cap ? s->wq_cap * 2 : 8;
|
||||
s->wq = realloc(s->wq, s->wq_cap * sizeof *s->wq);
|
||||
if (!s->wq) scr_net_oom();
|
||||
}
|
||||
s->wq[s->wq_n].end = s->wbase + s->wlen;
|
||||
s->wq[s->wq_n].len = counted;
|
||||
s->wq_n++;
|
||||
s->wpending += counted;
|
||||
}
|
||||
|
||||
/* Node's net.Socket writableHighWaterMark (the Writable default,
|
||||
* getDefaultHighWaterMark()). */
|
||||
#define SCR_NET_WRITE_HWM 65536
|
||||
|
||||
/* 'drain' once a false write's bytes are all out: Node's afterWrite emits
|
||||
* it when needDrain is set and nothing is pending, unless the stream is
|
||||
* ending or destroyed. */
|
||||
static void scr_net_sock_maybe_drain(ScrNetSocket *s) {
|
||||
if (!s->need_drain || s->wlen > s->whead) return;
|
||||
if (s->wr_ending || s->close_emitted || s->fd < 0) return;
|
||||
s->need_drain = false;
|
||||
scr_net_fire0_this(&s->drain_ls, s, SCR_DYNH_NET_SOCKET);
|
||||
}
|
||||
|
||||
static void scr_net_sock_write_raw(ScrNetSocket *s, const char *data, size_t len);
|
||||
|
||||
/* The write half takes writes: not ending or closed, and connected,
|
||||
* connecting, or waiting to dial (the agent's deferred dial, the caller
|
||||
* lookup), which buffers them until the connection is established. */
|
||||
static bool scr_net_sock_write_open(ScrNetSocket *w) {
|
||||
if (w->wr_ending || w->close_emitted) return false;
|
||||
return w->fd >= 0 || w->dial_deferred || w->lookup_wait;
|
||||
}
|
||||
|
||||
/* A user write: Writable.write's answer, as Node 24's writeOrBuffer
|
||||
* computes it after the socket writes. Each write with bytes still
|
||||
* unsent counts in full (wpending, Node's state.length), so the answer is
|
||||
* false when the pending writes reach the high-water mark, and needDrain
|
||||
* is set. A write the kernel takes at once answers true whatever its
|
||||
* size. A write to an ended or closed socket answers false, and so does
|
||||
* one whose send fails and closes the socket. */
|
||||
static bool scr_net_sock_user_write(ScrNetSocket *s, const char *data, size_t len) {
|
||||
ScrNetSocket *w = s->adopted_by != NULL ? s->adopted_by : s;
|
||||
bool open = scr_net_sock_write_open(w);
|
||||
scr_net_sock_write_raw(s, data, len);
|
||||
if (!open || !scr_net_sock_write_open(w)) return false;
|
||||
bool ok = w->wpending < SCR_NET_WRITE_HWM;
|
||||
if (!ok) w->need_drain = true;
|
||||
return ok;
|
||||
}
|
||||
|
||||
/* The write path shared by write()/end(data)/pipe delivery. Ignored on a
|
||||
@@ -1136,7 +1231,7 @@ static void scr_net_sock_write_raw(ScrNetSocket *s, const char *data, size_t len
|
||||
(s->tops && !s->t_est)) {
|
||||
/* mid-connect and mid-handshake bytes buffer; the flush after
|
||||
* 'connect'/establishment sends them */
|
||||
scr_net_sock_buffer(s, data, len);
|
||||
scr_net_sock_buffer(s, data, len, len);
|
||||
scr_net_sock_update_write(s);
|
||||
return;
|
||||
}
|
||||
@@ -1155,7 +1250,7 @@ static void scr_net_sock_write_raw(ScrNetSocket *s, const char *data, size_t len
|
||||
}
|
||||
}
|
||||
if ((size_t)sent < len) {
|
||||
scr_net_sock_buffer(s, data + sent, len - (size_t)sent);
|
||||
scr_net_sock_buffer(s, data + sent, len - (size_t)sent, len);
|
||||
scr_net_sock_update_write(s);
|
||||
}
|
||||
}
|
||||
@@ -1220,7 +1315,7 @@ void scr_net_sock_writev_native(ScrNetSocket *s, const ScrNetSlice *slices, size
|
||||
skip -= len;
|
||||
continue;
|
||||
}
|
||||
scr_net_sock_buffer(s, (const char *)iov[i].iov_base + skip, len - skip);
|
||||
scr_net_sock_buffer(s, (const char *)iov[i].iov_base + skip, len - skip, len);
|
||||
skip = 0;
|
||||
}
|
||||
scr_net_sock_update_write(s);
|
||||
@@ -1247,11 +1342,15 @@ static void scr_net_sock_flush(ScrNetSocket *s) {
|
||||
}
|
||||
if (sent > 0) scr_net_sock_touch(s);
|
||||
s->whead += (size_t)sent;
|
||||
scr_net_sock_wq_settle(s);
|
||||
}
|
||||
if (s->wlen == s->whead) {
|
||||
s->whead = s->wlen = 0;
|
||||
scr_net_sock_wq_clear(s);
|
||||
scr_net_sock_update_write(s);
|
||||
scr_net_sock_maybe_finish_write(s);
|
||||
/* Node's afterWrite for a write the socket completed later: 'drain'
|
||||
* now, unless the write half is ending or the socket is gone */
|
||||
scr_net_sock_maybe_drain(s);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2735,7 +2834,7 @@ ScrNetSocket *scr_net_connect_deferred(double port, ScrStr *host /*borrowed, nul
|
||||
* for a slot takes a pooled socket instead of dialing. */
|
||||
void scr_net_sock_move_writes(ScrNetSocket *from, ScrNetSocket *to) {
|
||||
if (from->wlen > from->whead) scr_net_sock_write_native(to, from->wbuf + from->whead, from->wlen - from->whead);
|
||||
from->whead = from->wlen = 0;
|
||||
scr_net_sock_wq_clear(from);
|
||||
}
|
||||
|
||||
void scr_net_sock_dial_start(ScrNetSocket *s) {
|
||||
@@ -3150,16 +3249,42 @@ void scr_net_lookup_answer(ScrClosure *self, bool has_err, ScrStr *msg /*borrowe
|
||||
|
||||
/* ── the socket surface ──────────────────────────────────────────────── */
|
||||
|
||||
void scr_net_sock_write_str(ScrNetSocket *s, ScrStr *data /*borrowed*/) {
|
||||
bool scr_net_sock_write_str(ScrNetSocket *s, ScrStr *data /*borrowed*/) {
|
||||
char *owned;
|
||||
const char *bytes = scr_str_utf8(data, &owned);
|
||||
scr_net_sock_write_raw(s, bytes, data->len);
|
||||
bool ok = scr_net_sock_user_write(s, bytes, data->len);
|
||||
scr_str_utf8_end(owned);
|
||||
return ok;
|
||||
}
|
||||
|
||||
void scr_net_sock_write_bytes(ScrNetSocket *s, ScrBytes *data /*borrowed*/) {
|
||||
bool scr_net_sock_write_bytes(ScrNetSocket *s, ScrBytes *data /*borrowed*/) {
|
||||
SCR_BYTES_SNAPSHOT(data);
|
||||
scr_net_sock_write_raw(s, (const char *)data->data, (data->len * scr_bytes_elem_size(data->elem)));
|
||||
return scr_net_sock_user_write(s, (const char *)data->data,
|
||||
(data->len * scr_bytes_elem_size(data->elem)));
|
||||
}
|
||||
|
||||
/* socket.writableLength: Node's state.length, the length of every write
|
||||
* not yet completely sent. */
|
||||
double scr_net_sock_writable_length(ScrNetSocket *s) {
|
||||
ScrNetSocket *w = s->adopted_by != NULL ? s->adopted_by : s;
|
||||
return (double)w->wpending;
|
||||
}
|
||||
|
||||
/* socket.writableNeedDrain: a write answered false and 'drain' has not
|
||||
* fired, on a socket whose write half is open (including one still
|
||||
* waiting to dial). */
|
||||
bool scr_net_sock_writable_need_drain(ScrNetSocket *s) {
|
||||
ScrNetSocket *w = s->adopted_by != NULL ? s->adopted_by : s;
|
||||
return w->need_drain && scr_net_sock_write_open(w);
|
||||
}
|
||||
|
||||
/* on('drain') / once('drain'). */
|
||||
void scr_net_sock_on_drain(ScrNetSocket *s, ScrClosure *cb /*moves*/, bool once) {
|
||||
if (s->close_emitted) {
|
||||
scr_closure_release(cb);
|
||||
return;
|
||||
}
|
||||
scr_net_ls_add(&s->drain_ls, cb, NULL, once);
|
||||
}
|
||||
|
||||
void scr_net_sock_end(ScrNetSocket *s) {
|
||||
@@ -5264,7 +5389,7 @@ ScrNetSocket *scr_net_sock_adopt(ScrNetSocket *raw) {
|
||||
tun->send_only = true;
|
||||
t->tunnel = tun;
|
||||
}
|
||||
raw->whead = raw->wlen = 0;
|
||||
scr_net_sock_wq_clear(raw);
|
||||
raw->adopted_by = scr_net_sock_retain(t);
|
||||
t->adopted_from = scr_net_sock_retain(raw);
|
||||
scr_net_sock_register(t);
|
||||
@@ -5523,6 +5648,7 @@ static bool scr_net_sock_settle(ScrNetSocket *sock) {
|
||||
scr_net_ls_drop(&sock->conn_ls);
|
||||
scr_net_ls_drop(&sock->timeout_ls);
|
||||
scr_net_ls_drop(&sock->readable_ls);
|
||||
scr_net_ls_drop(&sock->drain_ls);
|
||||
scr_net_ls_drop(&sock->finish_ls);
|
||||
scr_net_ls_drop(&sock->wcb_ls);
|
||||
scr_net_ls_drop(&sock->ready_ls);
|
||||
@@ -5860,6 +5986,7 @@ static void scr_net_cleanup_atexit(void) {
|
||||
scr_net_ls_drop(&s->conn_ls);
|
||||
scr_net_ls_drop(&s->timeout_ls);
|
||||
scr_net_ls_drop(&s->readable_ls);
|
||||
scr_net_ls_drop(&s->drain_ls);
|
||||
scr_net_ls_drop(&s->lookup_ls);
|
||||
/* Tear down the protocol layer's parser NOW: an in-flight http
|
||||
* exchange (a handler that threw before res.end()) holds req/res,
|
||||
@@ -5984,8 +6111,7 @@ static ScrDyn *scr_net_dynh_sock_invoke(void *h, ScrDyn *self, const char *metho
|
||||
/* fires once when the FIN goes out — once either way */
|
||||
scr_net_sock_on_finish(s, scr_dyn_listener_closure0(cb));
|
||||
} else if (scr_net_dynh_name_is(name, "drain")) {
|
||||
/* This surface never backpressures (write answers true) — an
|
||||
* accepted, never-fired registration is the consistent answer. */
|
||||
scr_net_sock_on_drain(s, scr_dyn_listener_closure0(cb), once);
|
||||
} else if (scr_net_dynh_tls != NULL && name->kind == SCR_DYN_STR &&
|
||||
scr_net_dynh_tls->on(s, name->v.str->data, cb, once)) {
|
||||
/* a TLS event name ('secureConnect'), handled by the TLS unit */
|
||||
@@ -6008,13 +6134,13 @@ static ScrDyn *scr_net_dynh_sock_invoke(void *h, ScrDyn *self, const char *metho
|
||||
scr_dyn_arg_type_fail("chunk", "of type string or an instance of Buffer or Uint8Array", chunk);
|
||||
return NULL;
|
||||
}
|
||||
if (chunk->kind == SCR_DYN_STR) scr_net_sock_write_str(s, chunk->v.str);
|
||||
else scr_net_sock_write_bytes(s, chunk->v.bytes);
|
||||
bool ok = chunk->kind == SCR_DYN_STR ? scr_net_sock_write_str(s, chunk->v.str)
|
||||
: scr_net_sock_write_bytes(s, chunk->v.bytes);
|
||||
if (argc > 1 && args[1]->kind == SCR_DYN_FUNC) {
|
||||
/* write(chunk, cb): fires when the buffer drains (sweep-deferred) */
|
||||
scr_net_sock_on_write_flush(s, scr_dyn_listener_closure0(args[1]));
|
||||
}
|
||||
return scr_dyn_new_bool(true); /* backpressure is not modeled (SEMANTICS.md) */
|
||||
return scr_dyn_new_bool(ok);
|
||||
}
|
||||
if (strcmp(method, "end") == 0) {
|
||||
const ScrDyn *chunk = argc > 0 ? args[0] : scr_dyn_undefined();
|
||||
@@ -6193,10 +6319,12 @@ static ScrDyn *scr_net_dynh_sock_get(void *h, const char *key, size_t key_len) {
|
||||
}
|
||||
}
|
||||
if (strcmp(key, "readable") == 0) return scr_dyn_new_bool(scr_net_sock_readable(s));
|
||||
if (strcmp(key, "writableLength") == 0) return scr_dyn_new_num(scr_net_sock_writable_length(s));
|
||||
if (strcmp(key, "writableNeedDrain") == 0)
|
||||
return scr_dyn_new_bool(scr_net_sock_writable_need_drain(s));
|
||||
if (strcmp(key, "writableHighWaterMark") == 0 || strcmp(key, "readableHighWaterMark") == 0) {
|
||||
/* Node's default stream highWaterMark — a constant here (backpressure
|
||||
* is not modeled; SEMANTICS.md) */
|
||||
return scr_dyn_new_num(16384);
|
||||
/* Node's default stream highWaterMark (getDefaultHighWaterMark()) */
|
||||
return scr_dyn_new_num(SCR_NET_WRITE_HWM);
|
||||
}
|
||||
if (scr_net_dynh_tls != NULL) {
|
||||
ScrDyn *out = NULL;
|
||||
@@ -6500,10 +6628,11 @@ void scr_net_dyn_install(void) {
|
||||
* STR/BYTES dispatch onto the typed entries; undefined/null end()s plain
|
||||
* (Node accepts them); anything else throws Node's ERR_INVALID_ARG_TYPE
|
||||
* chunk TypeError. Both borrowed. */
|
||||
void scr_net_sock_write_dynv(ScrNetSocket *s, const ScrDyn *d) {
|
||||
if (d->kind == SCR_DYN_STR) scr_net_sock_write_str(s, d->v.str);
|
||||
else if (d->kind == SCR_DYN_BYTES) scr_net_sock_write_bytes(s, d->v.bytes);
|
||||
else scr_dyn_arg_type_fail("chunk", "of type string or an instance of Buffer or Uint8Array", d);
|
||||
bool scr_net_sock_write_dynv(ScrNetSocket *s, const ScrDyn *d) {
|
||||
if (d->kind == SCR_DYN_STR) return scr_net_sock_write_str(s, d->v.str);
|
||||
if (d->kind == SCR_DYN_BYTES) return scr_net_sock_write_bytes(s, d->v.bytes);
|
||||
scr_dyn_arg_type_fail("chunk", "of type string or an instance of Buffer or Uint8Array", d);
|
||||
return false;
|
||||
}
|
||||
|
||||
void scr_net_sock_end_dynv(ScrNetSocket *s, const ScrDyn *d) {
|
||||
|
||||
@@ -227,15 +227,21 @@ ScrNetSocket *scr_net_connect_lookup(double port, ScrStr *host /*borrowed*/,
|
||||
ScrClosure *lookup /*moves*/, void *answer_fn); /* +1 */
|
||||
void scr_net_lookup_answer(ScrClosure *self, bool has_err, ScrStr *msg /*borrowed, nullable*/,
|
||||
ScrArr *ips /*borrowed, nullable*/);
|
||||
void scr_net_sock_write_str(ScrNetSocket *s, ScrStr *data /*borrowed*/);
|
||||
void scr_net_sock_write_bytes(ScrNetSocket *s, ScrBytes *data /*borrowed*/);
|
||||
/* write(): Node's Writable answer — false when the bytes the write left
|
||||
* unsent reach the 65536-byte high-water mark, after which 'drain' fires
|
||||
* when the write buffer empties; false on an ended or closed socket. */
|
||||
bool scr_net_sock_write_str(ScrNetSocket *s, ScrStr *data /*borrowed*/);
|
||||
bool scr_net_sock_write_bytes(ScrNetSocket *s, ScrBytes *data /*borrowed*/);
|
||||
double scr_net_sock_writable_length(ScrNetSocket *s);
|
||||
bool scr_net_sock_writable_need_drain(ScrNetSocket *s);
|
||||
void scr_net_sock_on_drain(ScrNetSocket *s, ScrClosure *cb /*moves*/, bool once);
|
||||
void scr_net_sock_end(ScrNetSocket *s);
|
||||
void scr_net_sock_end_str(ScrNetSocket *s, ScrStr *data /*borrowed*/);
|
||||
void scr_net_sock_end_bytes(ScrNetSocket *s, ScrBytes *data /*borrowed*/);
|
||||
/* Checked-dynamic chunk forms (an untyped JS payload into a typed
|
||||
* socket): STR/BYTES dispatch; end() takes undefined/null as no chunk;
|
||||
* anything else throws Node's ERR_INVALID_ARG_TYPE chunk TypeError. */
|
||||
void scr_net_sock_write_dynv(ScrNetSocket *s, const ScrDyn *d /*borrowed*/);
|
||||
bool scr_net_sock_write_dynv(ScrNetSocket *s, const ScrDyn *d /*borrowed*/);
|
||||
void scr_net_sock_end_dynv(ScrNetSocket *s, const ScrDyn *d /*borrowed*/);
|
||||
void scr_net_sock_destroy(ScrNetSocket *s);
|
||||
/* Flow control (pause/resume — see the struct's flag comments), the
|
||||
|
||||
@@ -0,0 +1,136 @@
|
||||
// net.Socket write() backpressure, as in Node 24: write answers false when
|
||||
// the writes not yet completely sent reach the 65536-byte high-water mark,
|
||||
// each counting in full, which a write made before the connection does;
|
||||
// writableNeedDrain stays true until 'drain', and 'drain' fires once the
|
||||
// buffered bytes are sent. The same holds while a custom lookup is pending,
|
||||
// writableLength counts a large write the kernel took in part in full,
|
||||
// untyped code reads the same properties, and a write whose send fails
|
||||
// after the peer reset the connection answers false. An echo client that
|
||||
// waits for 'drain' when write answers false moves every byte.
|
||||
import { connect, createServer, Socket } from "node:net";
|
||||
import type { LookupAddress, LookupAllOptions, LookupOneOptions } from "node:dns";
|
||||
|
||||
const CHUNK = 65536;
|
||||
const TOTAL = 64 * CHUNK;
|
||||
const chunk = Buffer.alloc(CHUNK, 120);
|
||||
|
||||
function untyped(socket: any): string {
|
||||
return `${socket.writableLength} ${socket.writableNeedDrain} ${socket.writableHighWaterMark}`;
|
||||
}
|
||||
|
||||
function echo(port: number): Promise<void> {
|
||||
return new Promise<void>((resolve) => {
|
||||
const client = connect(port, "127.0.0.1");
|
||||
console.log("small write before connect", client.write("hi"), client.writableLength);
|
||||
console.log("large write before connect", client.write(Buffer.alloc(70000, 121)), client.writableLength);
|
||||
console.log("need drain", client.writableNeedDrain, "untyped", untyped(client));
|
||||
let sent = 70002;
|
||||
let received = 0;
|
||||
const pump = (): void => {
|
||||
while (sent < TOTAL) {
|
||||
sent += CHUNK;
|
||||
if (!client.write(chunk)) {
|
||||
client.once("drain", pump);
|
||||
return;
|
||||
}
|
||||
}
|
||||
client.end();
|
||||
};
|
||||
client.once("drain", () => {
|
||||
console.log("drain fired, need drain", client.writableNeedDrain, "pending", client.writableLength);
|
||||
pump();
|
||||
});
|
||||
client.on("data", (data: Buffer) => {
|
||||
received += data.length;
|
||||
});
|
||||
client.on("end", () => {
|
||||
console.log("received", received, "of", sent);
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function lookupPending(port: number): Promise<void> {
|
||||
return new Promise<void>((resolve) => {
|
||||
const client = connect({
|
||||
port,
|
||||
host: "localhost",
|
||||
lookup: (
|
||||
_host: string,
|
||||
_options: LookupOneOptions | LookupAllOptions | number | undefined,
|
||||
callback: (err: NodeJS.ErrnoException | null, address: string | LookupAddress[], family?: number) => void,
|
||||
) => {
|
||||
const answer: LookupAddress[] = [{ address: "127.0.0.1", family: 4 }];
|
||||
setTimeout(() => callback(null, answer), 20);
|
||||
},
|
||||
});
|
||||
console.log("lookup write", client.write(Buffer.alloc(70000, 97)));
|
||||
console.log("lookup state", client.writableLength, client.writableNeedDrain, "untyped", untyped(client));
|
||||
client.once("drain", () => {
|
||||
console.log("lookup drain", client.writableLength, client.writableNeedDrain);
|
||||
client.destroy();
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function largeWrite(port: number): Promise<void> {
|
||||
return new Promise<void>((resolve) => {
|
||||
const client = connect(port, "127.0.0.1", () => {
|
||||
const size = 16 * 1024 * 1024;
|
||||
console.log("large write", client.write(Buffer.alloc(size, 98)));
|
||||
console.log("large state", client.writableLength === size, client.writableNeedDrain);
|
||||
client.once("drain", () => {
|
||||
console.log("large drain", client.writableLength, client.writableNeedDrain);
|
||||
client.destroy();
|
||||
resolve();
|
||||
});
|
||||
setTimeout(() => {
|
||||
if (paused !== null) paused.resume();
|
||||
}, 30);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function afterReset(port: number): Promise<void> {
|
||||
return new Promise<void>((resolve) => {
|
||||
const client = connect(port, "127.0.0.1", () => {
|
||||
client.pause();
|
||||
setTimeout(() => {
|
||||
console.log("first write after the peer closed", client.write("a"));
|
||||
setTimeout(() => {
|
||||
console.log("write after reset", client.write("b"));
|
||||
}, 50);
|
||||
}, 50);
|
||||
});
|
||||
client.on("error", () => {});
|
||||
client.on("close", () => {
|
||||
console.log("closed, destroyed", client.destroyed);
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
let paused: Socket | null = null;
|
||||
let mode = "echo";
|
||||
const server = createServer((socket: Socket) => {
|
||||
if (mode === "echo") socket.pipe(socket);
|
||||
else if (mode === "drain") socket.resume();
|
||||
else if (mode === "pause") {
|
||||
socket.pause();
|
||||
paused = socket;
|
||||
} else socket.destroy();
|
||||
});
|
||||
|
||||
server.listen(0, "127.0.0.1", async () => {
|
||||
const addr = server.address();
|
||||
const port = typeof addr === "object" && addr !== null ? addr.port : 0;
|
||||
await echo(port);
|
||||
mode = "drain";
|
||||
await lookupPending(port);
|
||||
mode = "pause";
|
||||
await largeWrite(port);
|
||||
mode = "end";
|
||||
await afterReset(port);
|
||||
server.close();
|
||||
});
|
||||
@@ -0,0 +1,62 @@
|
||||
// 'drain' listeners that capture their socket do not keep it after it
|
||||
// closes: a once('drain') listener still pending when the peer disconnects
|
||||
// before the buffer drains, and an on('drain') listener on a socket that
|
||||
// drained and then closed. The sanitized lane's exit audit checks that
|
||||
// both sockets are freed.
|
||||
import { connect, createServer, Socket } from "node:net";
|
||||
|
||||
function disconnectBeforeDrain(): Promise<void> {
|
||||
return new Promise<void>((resolve) => {
|
||||
const server = createServer((socket: Socket) => {
|
||||
socket.on("error", () => {});
|
||||
socket.once("drain", () => console.log("unexpected drain", socket.writableLength));
|
||||
console.log("server write", socket.write(Buffer.alloc(16 * 1024 * 1024, 97)));
|
||||
socket.on("close", () => {
|
||||
console.log("server socket closed, need drain", socket.writableNeedDrain);
|
||||
server.close(() => resolve());
|
||||
});
|
||||
});
|
||||
server.listen(0, "127.0.0.1", () => {
|
||||
const addr = server.address();
|
||||
const port = typeof addr === "object" && addr !== null ? addr.port : 0;
|
||||
const client = connect(port, "127.0.0.1", () => {
|
||||
client.pause();
|
||||
setTimeout(() => client.destroy(), 30);
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function drainThenClose(): Promise<void> {
|
||||
return new Promise<void>((resolve) => {
|
||||
const server = createServer((socket: Socket) => {
|
||||
socket.pause();
|
||||
setTimeout(() => socket.resume(), 30);
|
||||
});
|
||||
server.listen(0, "127.0.0.1", () => {
|
||||
const addr = server.address();
|
||||
const port = typeof addr === "object" && addr !== null ? addr.port : 0;
|
||||
const client = connect(port, "127.0.0.1", () => {
|
||||
let drains = 0;
|
||||
client.on("drain", () => {
|
||||
drains++;
|
||||
console.log("drain", drains, client.writableLength);
|
||||
client.end();
|
||||
});
|
||||
console.log("client write", client.write(Buffer.alloc(16 * 1024 * 1024, 98)));
|
||||
});
|
||||
client.on("close", () => {
|
||||
console.log("client closed");
|
||||
server.close(() => resolve());
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
await disconnectBeforeDrain();
|
||||
await drainThenClose();
|
||||
console.log("done");
|
||||
}
|
||||
|
||||
main();
|
||||
@@ -0,0 +1 @@
|
||||
{ "plain": 2800, "san": 4200 }
|
||||
@@ -0,0 +1 @@
|
||||
{ "plain": 2300, "san": 3100 }
|
||||
Reference in New Issue
Block a user