feat(netstring): Chunked mode for writer
Allows no-copy writing of netstring
MODIFY
packages/daemon/src/connection.js
packages/daemon/src/connection.js
@@ -71,7 +71,7 @@ export const makeNodeNetstringCapTP = (
bootstrap,
) => {
const writer = mapWriter(
- makeNetstringWriter(makeNodeWriter(nodeWriter)),
+ makeNetstringWriter(makeNodeWriter(nodeWriter), { chunked: true }),
messageToBytes,
);
const reader = mapReader(

MODIFY
packages/netstring/test/test-netstring.js
packages/netstring/test/test-netstring.js
@@ -103,9 +103,10 @@ function delay(ms) {
});
}
-const makeArrayWriter = () => {
+const makeArrayWriter = opts => {
const array = [];
- const writer = makeNetstringWriter({
+ const writer = makeNetstringWriter(
+ {
async next(value) {
// Provide some back pressure to give the producer an opportunity to make
// the mistake of overwriting the given slice.
@@ -120,12 +121,14 @@ const makeArrayWriter = () => {
async throw() {
return { done: true };
},
- });
+ },
+ opts,
+ );
return { array, writer };
};
-test('round-trip short messages', async t => {
- const { array, writer } = makeArrayWriter();
+const shortMessages = async (t, opts) => {
+ const { array, writer } = makeArrayWriter(opts);
await writer.next(encoder.encode(''));
await writer.next(encoder.encode('A'));
await writer.next(encoder.encode('hello'));
@@ -135,10 +138,12 @@ test('round-trip short messages', async t => {
[encoder.encode(''), encoder.encode('A'), encoder.encode('hello')],
await read(makeNetstringReader(array)),
);
-});
+};
+test('round-trip short messages', shortMessages);
+test('round-trip short messages (chunked)', shortMessages, { chunked: true });
-test('concurrent writes', async t => {
- const { array, writer } = makeArrayWriter();
+const concurrentWrites = async (t, opts) => {
+ const { array, writer } = makeArrayWriter(opts);
await Promise.all([
writer.next(encoder.encode('')),
writer.next(encoder.encode('A')),
@@ -150,9 +155,11 @@ test('concurrent writes', async t => {
[encoder.encode(''), encoder.encode('A'), encoder.encode('hello')],
await read(makeNetstringReader(array)),
);
-});
+};
+test('concurrent writes', concurrentWrites);
+test('concurrent writes (chunked)', concurrentWrites, { chunked: true });
-test('round-trip varying messages', async t => {
+const varyingMessages = async (t, opts) => {
const array = ['', 'A', 'hello'];
for (let i = 1020; i < 1030; i += 1) {
@@ -168,7 +175,7 @@ test('round-trip varying messages', async t => {
const producer = (async () => {
/** @type {import('@endo/stream').Writer<Uint8Array, undefined>} */
- const w = makeNetstringWriter(output);
+ const w = makeNetstringWriter(output, opts);
for (let i = 0; i < array.length; i += 1) {
// eslint-disable-next-line no-await-in-loop
await w.next(encoder.encode(array[i]));
@@ -191,4 +198,8 @@ test('round-trip varying messages', async t => {
})();
await Promise.all([producer, consumer]);
+};
+test('round-trip varying messages', varyingMessages);
+test('round-trip varying messages (chunked)', varyingMessages, {
+ chunked: true,
});

MODIFY
packages/netstring/writer.js
packages/netstring/writer.js
// @ts-check
/// <reference types="ses"/>
-const COLON = ':'.charCodeAt(0);
-const COMMA = ','.charCodeAt(0);
+const COMMA_BUFFER = new Uint8Array([','.charCodeAt(0)]);
-// The initial buffer length should be small enough to not be an imposition in
-// the common case but large enough to avoid reallocation in the common case.
-// It also must be long enough to fit an ASCII decimal representation of the
-// length of the longest possible message.
-// This is a guess:
-const initialBufferLength = 128;
-
-const encoder = new TextEncoder();
+/** @param {number} length */
+const getLengthPrefixCharCodes = length =>
+ // eslint-disable-next-line no-bitwise
+ [...`${length | 0}:`].map(char => char.charCodeAt(0));
/**
* @param {import('@endo/stream').Writer<Uint8Array, undefined>} output
+ * @param {object} [opts]
+ * @param {boolean} [opts.chunked]
* @returns {import('@endo/stream').Writer<Uint8Array, undefined>}
*/
-export const makeNetstringWriter = output => {
+export const makeNetstringWriter = (output, { chunked = false } = {}) => {
return harden({
async next(message) {
- // Must allocate to support concurrent writes.
- let buffer = new Uint8Array(initialBufferLength);
- const { written: colonAt = 0 } = encoder.encodeInto(
- `${message.byteLength}`,
- buffer,
- );
- const messageAt = colonAt + 1;
- const commaAt = messageAt + message.byteLength;
- const messageLength = commaAt + 1;
-
- // Grow buffer if necessary.
- if (messageLength > buffer.byteLength) {
- let newCapacity = buffer.byteLength;
- while (newCapacity < messageLength) {
- newCapacity *= 2;
- }
- const newBuffer = new Uint8Array(newCapacity);
- newBuffer.set(buffer, 0);
- buffer = newBuffer;
+ const prefix = getLengthPrefixCharCodes(message.length);
+
+ if (chunked) {
+ return Promise.all([
+ output.next(new Uint8Array(prefix)),
+ output.next(message),
+ output.next(COMMA_BUFFER),
+ ]).then(([r1, r2, r3]) => ({
+ done: !!(r1.done || r2.done || r3.done),
+ value: undefined,
+ }));
+ } else {
+ const buffer = new Uint8Array(prefix.length + message.length + 1);
+ buffer.set(prefix, 0);
+ buffer.set(message, prefix.length);
+ buffer.set(COMMA_BUFFER, buffer.length - 1);
+
+ return output.next(buffer);
}
-
- buffer[colonAt] = COLON;
- buffer.set(message, messageAt);
- buffer[commaAt] = COMMA;
-
- return output.next(buffer.subarray(0, messageLength));
},
async return() {
return output.return(undefined);

feat(netstring): Chunked mode for writer
Allows no-copy writing of netstring