diff --git a/.changeset/schema-binary-codec.md b/.changeset/schema-binary-codec.md index 4b95439c3a6..6b6c89fd3bc 100644 --- a/.changeset/schema-binary-codec.md +++ b/.changeset/schema-binary-codec.md @@ -5,3 +5,5 @@ Add `SchemaBinary`, a compact binary codec derived from the Schema AST. `SchemaBinary.toCodec(schema)` compiles a wire layout from the encoded-side AST on each side, so field names never appear on the payload. `SchemaBinary.parser(schema)` reads concatenated frames from a stream, and `SchemaBinary.fieldId(n)` pins a field's wire id so it survives a rename. + +Encoded results are stable views into a shared bump-allocated arena. Their byte range is exact, but their backing buffer may be larger, have a non-zero offset, and contain other encoded results. Copy a result before transferring its buffer when independent ownership is required. diff --git a/packages/effect/benchmark/schema/SchemaBinary.md b/packages/effect/benchmark/schema/SchemaBinary.md index 0d2d112fd96..bb5ae463777 100644 --- a/packages/effect/benchmark/schema/SchemaBinary.md +++ b/packages/effect/benchmark/schema/SchemaBinary.md @@ -6,7 +6,7 @@ Run from the repository root: nix develop -c pnpm --dir packages/effect exec node benchmark/schema/SchemaBinary.ts ``` -The benchmark compares `SchemaBinary`, JSON, and Effect's Msgpack schema codec with the same Schema values. It covers a small scalar-heavy record, a nested order payload, arrays and records, and 200 repeated records where framing and field-name overhead become visible. +The benchmark compares arena-backed `SchemaBinary`, `SchemaBinary` followed by an ownership copy, JSON, and Effect's Msgpack schema codec with the same Schema values. The copy case calls `.slice()` on every arena result, reproducing the allocation that the arena removes. It covers a small scalar-heavy record, a nested order payload, arrays and records, and 200 repeated records where framing and field-name overhead become visible. Codec and schema construction happen before timing. Decode uses payloads encoded before timing. Each one-shot encode and decode task gets 100 warmup samples followed by 1,000 measured samples. The output includes UTF-8 payload sizes, average operations per second, median latency, relative margin of error, and sample count. @@ -21,27 +21,33 @@ Treat the measurements as a per-machine comparison within one run. Runtime versi One run on Node 26.7.0, Linux x64. One-shot throughput is the average of 1,000 measured samples per task; treat these as a within-run comparison, not a portable score. -| Case | Direction | SchemaBinary | Msgpack | SchemaBinary vs Msgpack | -| ---------------------- | --------- | -----------: | ------: | ----------------------: | -| small record | encode | 413,942 | 468,259 | 0.88x | -| small record | decode | 472,582 | 474,553 | 1.00x | -| nested payload | encode | 230,222 | 205,683 | 1.12x | -| nested payload | decode | 227,531 | 199,069 | 1.14x | -| collections | encode | 33,596 | 21,846 | 1.54x | -| collections | decode | 37,822 | 21,604 | 1.75x | -| large repeated records | encode | 6,366 | 3,543 | 1.80x | -| large repeated records | decode | 6,248 | 4,156 | 1.50x | +The encode rows exercise both ownership models on every case. Decode does not depend on output ownership, so the arena rows below are compared with Msgpack. + +| Case | SchemaBinary arena | SchemaBinary copy | Arena vs copy | Msgpack | +| ---------------------- | -----------------: | ----------------: | ------------: | ------: | +| small record | 462,238 | 444,428 | 1.04x | 470,631 | +| nested payload | 235,971 | 223,358 | 1.06x | 209,940 | +| collections | 34,632 | 34,224 | 1.01x | 22,340 | +| large repeated records | 6,196 | 6,094 | 1.02x | 3,514 | + +| Case | SchemaBinary decode | Msgpack decode | SchemaBinary vs Msgpack | +| ---------------------- | ------------------: | -------------: | ----------------------: | +| small record | 485,041 | 471,327 | 1.03x | +| nested payload | 223,300 | 203,587 | 1.10x | +| collections | 37,881 | 21,442 | 1.77x | +| large repeated records | 6,129 | 4,066 | 1.51x | Both codecs run the same Schema parser, which costs roughly 140 us per direction on the largest case and sets a floor neither format can go below. The numbers above therefore understate the difference between the two serialization layers. -The small-record row is the one case where msgpackr stays ahead. It returns a -view into a reused 8 KiB buffer rather than allocating per call, so it skips -the output copy that `SchemaBinary` pays on every encode. `SchemaBinary` -returns a freshly allocated `Uint8Array` that the caller owns outright, which -costs roughly 250 ns and dominates a 72-byte payload. +Removing the output copy matters most for the 72-byte small record, where the +arena was 4% faster than the ownership-copy control and nearly closed the gap +with msgpackr. The other three cases were 1-6% faster than the copy control; +none showed a material regression. Results returned by the arena keep an exact +byte range but can share a larger backing buffer, as documented by +`SchemaBinary.toCodec`. ## Streaming decode comparison diff --git a/packages/effect/benchmark/schema/SchemaBinary.ts b/packages/effect/benchmark/schema/SchemaBinary.ts index 9617d22aa06..eb5e4e32b21 100644 --- a/packages/effect/benchmark/schema/SchemaBinary.ts +++ b/packages/effect/benchmark/schema/SchemaBinary.ts @@ -182,6 +182,7 @@ const prepare = >( const msgpackDecode = Schema.decodeUnknownSync(msgpackCodec) const binary = binaryEncode(value) + const binaryCopy = binary.slice() const json = jsonEncode(value) const msgpack = msgpackEncode(value) @@ -210,11 +211,17 @@ const prepare = >( return { formats: [ { - name: "SchemaBinary", + name: "SchemaBinary arena", encodedSize: binary.length, encode: () => binaryEncode(value), decode: () => binaryDecode(binary) }, + { + name: "SchemaBinary copy", + encodedSize: binaryCopy.length, + encode: () => binaryEncode(value).slice(), + decode: () => binaryDecode(binaryCopy) + }, { name: "JSON", encodedSize: textEncoder.encode(json).length, diff --git a/packages/effect/src/unstable/encoding/SchemaBinary.ts b/packages/effect/src/unstable/encoding/SchemaBinary.ts index 160ad1ff3b2..1aeca45aabc 100644 --- a/packages/effect/src/unstable/encoding/SchemaBinary.ts +++ b/packages/effect/src/unstable/encoding/SchemaBinary.ts @@ -170,46 +170,85 @@ function decodeUtf8( } } +const OUTPUT_ARENA_SIZE = 8 * 1024 + +interface OutputArena { + readonly buf: Uint8Array + offset: number + writing: boolean +} + +function makeOutputArena(size: number): OutputArena { + return { buf: new Uint8Array(size), offset: 0, writing: false } +} + +let outputArena = makeOutputArena(OUTPUT_ARENA_SIZE) + class Writer { - buf = new Uint8Array(1024) + buf: Uint8Array = new Uint8Array(0) view = new DataView(this.buf.buffer) + arena: OutputArena | undefined + start = 0 len = 0 + reset() { + let arena = outputArena + // A nested SchemaBinary codec can start while the outer writer still owns + // the current arena tail. Move the nested writer to a fresh arena rather + // than reserving a guessed range that the outer writer could outgrow. + if (arena.writing || arena.offset >= arena.buf.length) { + arena = outputArena = makeOutputArena(OUTPUT_ARENA_SIZE) + } + arena.writing = true + this.arena = arena + this.buf = arena.buf + this.view = new DataView(arena.buf.buffer) + this.start = arena.offset + this.len = 0 + } ensure(n: number) { - if (this.len + n > this.buf.length) { - const next = new Uint8Array(Math.max(this.buf.length * 2, this.len + n)) - next.set(this.buf) - this.buf = next - this.view = new DataView(next.buffer) + if (this.start + this.len + n > this.buf.length) { + const previous = this.arena! + const required = this.len + n + let size = OUTPUT_ARENA_SIZE + while (size < required) size *= 2 + const next = makeOutputArena(size) + next.writing = true + next.buf.set(this.buf.subarray(this.start, this.start + this.len)) + previous.writing = false + this.arena = outputArena = next + this.buf = next.buf + this.view = new DataView(next.buf.buffer) + this.start = 0 } } byte(b: number) { this.ensure(1) - this.buf[this.len++] = b + this.buf[this.start + this.len++] = b } bytes(b: Uint8Array) { this.ensure(b.length) - this.buf.set(b, this.len) + this.buf.set(b, this.start + this.len) this.len += b.length } uvarint(n: number) { this.ensure(10) const buf = this.buf - let p = this.len + let p = this.start + this.len while (n > 0x7F) { buf[p++] = (n & 0x7F) | 0x80 n = n < 0x80000000 ? n >>> 7 : Math.floor(n / 128) } buf[p++] = n - this.len = p + this.len = p - this.start } // Field ids are fixed by the layout, so their varint bytes are encoded once // at compile time and blitted here. raw(bytes: Uint8Array) { this.ensure(bytes.length) const buf = this.buf - let p = this.len + let p = this.start + this.len for (let i = 0; i < bytes.length; i++) buf[p++] = bytes[i] - this.len = p + this.len = p - this.start } uvarintBig(n: bigint) { while (n > BIGINT_VARINT_MASK) { @@ -223,22 +262,22 @@ class Writer { } f64(n: number) { this.ensure(8) - this.view.setFloat64(this.len, n, true) + this.view.setFloat64(this.start + this.len, n, true) this.len += 8 } i64(n: bigint) { this.ensure(8) - this.view.setBigInt64(this.len, n, true) + this.view.setBigInt64(this.start + this.len, n, true) this.len += 8 } u32le(n: number) { this.ensure(4) - this.view.setUint32(this.len, n >>> 0, true) + this.view.setUint32(this.start + this.len, n >>> 0, true) this.len += 4 } i32le(n: number) { this.ensure(4) - this.view.setInt32(this.len, n | 0, true) + this.view.setInt32(this.start + this.len, n | 0, true) this.len += 4 } // Reserves one byte for a length prefix that is not known yet. The payload is @@ -251,16 +290,17 @@ class Writer { endSized(mark: number) { const payload = this.len - mark - 1 if (payload < 0x80) { - this.buf[mark] = payload + this.buf[this.start + mark] = payload return } const size = uvarintSize(payload) const extra = size - 1 this.ensure(extra) - this.buf.copyWithin(mark + size, mark + 1, this.len) + const absoluteMark = this.start + mark + this.buf.copyWithin(absoluteMark + size, absoluteMark + 1, this.start + this.len) this.len += extra let n = payload - let p = mark + let p = absoluteMark while (n > 0x7F) { this.buf[p++] = (n & 0x7F) | 0x80 n = Math.floor(n / 128) @@ -272,19 +312,31 @@ class Writer { if (n === 0) return this.ensure(n * 3) const buf = this.buf - let p = this.len + let p = this.start + this.len for (let i = 0; i < n; i++) { const c = s.charCodeAt(i) if (c > 0x7F) { - this.len += utf8Encode.encodeInto(s, buf.subarray(this.len)).written + this.len += utf8Encode.encodeInto(s, buf.subarray(this.start + this.len)).written return } buf[p++] = c } - this.len = p + this.len = p - this.start } out(): Uint8Array { - return this.buf.slice(0, this.len) as Uint8Array + const arena = this.arena! + const out = new Uint8Array(arena.buf.buffer, this.start, this.len) + if (arena === outputArena) arena.offset = this.start + this.len + arena.writing = false + this.arena = undefined + return out + } + abort() { + const arena = this.arena + if (arena === undefined) return + if (arena === outputArena) arena.offset = this.start + arena.writing = false + this.arena = undefined } } @@ -1621,7 +1673,7 @@ function encodeFrame(layout: Layout, value: unknown, options: SchemaAST.ParseOpt const w = pooledWriter ?? new Writer() const pooled = w === pooledWriter if (pooled) pooledWriter = undefined - w.len = 0 + w.reset() const savedPathLen = issuePathLen issuePathLen = 0 try { @@ -1631,6 +1683,7 @@ function encodeFrame(layout: Layout, value: unknown, options: SchemaAST.ParseOpt w.endSized(mark) return w.out() } finally { + w.abort() issuePathLen = savedPathLen if (pooled) pooledWriter = w } @@ -2115,6 +2168,12 @@ export interface toCodec extends * One-shot encode/decode reuse the existing runners: exactly one frame, * leftover bytes are malformed. Use {@link parser} for concatenated frames. * + * Encoded results are stable views into a bump-allocated arena. A result's + * `byteLength` covers exactly one frame, but its `.buffer` may be larger, + * `byteOffset` may be non-zero, and unrelated encoded results may share the + * same buffer. Transferring or detaching that buffer affects every view into + * it. Use `bytes.slice()` when an independently owned buffer is required. + * * **Example** * * ```ts diff --git a/packages/effect/test/unstable/encoding/SchemaBinary.test.ts b/packages/effect/test/unstable/encoding/SchemaBinary.test.ts index fd8f39ee372..50f74f8cdf8 100644 --- a/packages/effect/test/unstable/encoding/SchemaBinary.test.ts +++ b/packages/effect/test/unstable/encoding/SchemaBinary.test.ts @@ -120,6 +120,65 @@ describe("SchemaBinary", () => { }) }) + describe("output arena", () => { + it("keeps an earlier result stable through later encodes and arena rollover", () => { + const codec = SchemaBinary.toCodec(Schema.String) + const encode = Schema.encodeUnknownSync(codec) + const first = encode("first") + const expected = first.slice() + let rolledOver = false + + for (let i = 0; i < 500; i++) { + const later = encode(`later-${i}-${"x".repeat(32)}`) + if (later.buffer !== first.buffer) rolledOver = true + assert.deepStrictEqual(first, expected) + } + + assert.isTrue(rolledOver) + assert.strictEqual(Schema.decodeUnknownSync(codec)(first), "first") + }) + + it("keeps different results independent inside a shared arena", () => { + const codec = SchemaBinary.toCodec(Schema.String) + const encode = Schema.encodeUnknownSync(codec) + const decode = Schema.decodeUnknownSync(codec) + const results = [encode("alpha"), encode("bravo"), encode("charlie")] + + assert.deepStrictEqual(results.map((bytes) => decode(bytes)), ["alpha", "bravo", "charlie"]) + const shared = results.flatMap((left, index) => results.slice(index + 1).map((right) => [left, right] as const)) + .find(([left, right]) => left.buffer === right.buffer) + assert.isDefined(shared) + assert.notStrictEqual(shared![0].byteOffset, shared![1].byteOffset) + }) + + it("preserves nested two-phase codec composition", () => { + const Inner = Schema.Struct({ id: Schema.Number, label: Schema.String }) + const Outer = Schema.Struct({ id: Schema.String, inner: SchemaBinary.toCodec(Inner) }) + const codec = SchemaBinary.toCodec(Outer) + const value = { id: "outer", inner: { id: 1, label: "inner" } } + + assert.deepStrictEqual(Schema.decodeUnknownSync(codec)(Schema.encodeUnknownSync(codec)(value)), value) + }) + + it.effect("keeps concurrent fiber results readable after their producing turn", () => { + const Value = Schema.Struct({ id: Schema.Number, value: Schema.String }) + const codec = SchemaBinary.toCodec(Value) + const encode = Schema.encodeUnknownEffect(codec) + const decode = Schema.decodeUnknownSync(codec) + const values = Array.from({ length: 100 }, (_, id) => ({ id, value: `value-${id}` })) + + return Effect.gen(function*() { + const encoded = yield* Effect.forEach( + values, + (value) => encode(value).pipe(Effect.flatMap((bytes) => Effect.yieldNow.pipe(Effect.as(bytes)))), + { concurrency: "unbounded" } + ) + yield* Effect.yieldNow + assert.deepStrictEqual(encoded.map((bytes) => decode(bytes)), values) + }) + }) + }) + describe("schema evolution", () => { it("skips unknown fields, accepts field reorder, and leaves missing optionals absent", () => { const Writer = Schema.Struct({ a: Schema.Number, extra: Schema.String, b: Schema.String })