js-waku
js-waku copied to clipboard
chore: investigate if possible to send many messages in a short amount of time (throughput is very limited in js-waku)
Problem
When attempting to send many messages back to back we run into an error: ERR_TOO_MANY_OUTBOUND_PROTOCOL_STREAMS
Steps to reproduce
import {
createLightNode,
waitForRemotePeer,
createEncoder,
utf8ToBytes,
} from "@waku/sdk";
import { Protocols } from "@waku/interfaces";
const contentTopic = "/light-guide/1/message/proto";
const encoder = createEncoder({ contentTopic });
const createNode = async () => {
const node = await createLightNode({
defaultBootstrap: true,
});
await node.start();
await waitForRemotePeer(node, [Protocols.LightPush, Protocols.Filter]);
setInterval(() => {
node.lightPush.send(encoder, utf8ToBytes("ping"));
}, 300);
};
createNode();
Expected Results
Expected to be able to send many messages back to back with no error (I can do this with libp2p directly with both floodsub and gossipsub implementations
Actual results
file:///Users/user/src/waku/node_modules/libp2p/dist/src/upgrader.js:325
const err = errCode(new Error(`Too many outbound protocol streams for protocol "${protocol}" - limit ${outgoingLimit}`), codes.ERR_TOO_MANY_OUTBOUND_PROTOCOL_STREAMS);
^
Error: Too many outbound protocol streams for protocol "/vac/waku/lightpush/2.0.0-beta1" - limit 64
at ConnectionImpl.newStream [as _newStream] (file:///Users/user/src/waku/node_modules/libp2p/dist/src/upgrader.js:325:45)
at process.processTicksAndRejections (node:internal/process/task_queues:95:5)
at async ConnectionImpl.newStream (file:///Users/user/src/waku/node_modules/libp2p/dist/src/connection/index.js:55:24)
at async LightPush.send (file:///Users/user/src/waku/node_modules/@waku/core/dist/lib/light_push/index.js:29:24) {
code: 'ERR_TOO_MANY_OUTBOUND_PROTOCOL_STREAMS'
}
Node.js v20.1.0
Libp2p example with no issues
/* eslint-disable no-console */
import { noise } from "@chainsafe/libp2p-noise";
import { yamux } from "@chainsafe/libp2p-yamux";
import { gossipsub } from "@chainsafe/libp2p-gossipsub";
import { mplex } from "@libp2p/mplex";
import { tcp } from "@libp2p/tcp";
import { createLibp2p } from "libp2p";
import { identifyService } from "libp2p/identify";
import { fromString as uint8ArrayFromString } from "uint8arrays/from-string";
import { toString as uint8ArrayToString } from "uint8arrays/to-string";
const createNode = async () => {
const node = await createLibp2p({
addresses: {
listen: ["/ip4/0.0.0.0/tcp/0"],
},
transports: [tcp()],
streamMuxers: [yamux(), mplex()],
connectionEncryption: [noise()],
services: {
pubsub: gossipsub(),
identify: identifyService(),
},
});
return node;
};
(async () => {
const topic = "news";
const [node1, node2] = await Promise.all([createNode(), createNode()]);
await node1.peerStore.patch(node2.peerId, {
multiaddrs: node2.getMultiaddrs(),
});
await node1.dial(node2.peerId);
node1.services.pubsub.subscribe(topic);
node1.services.pubsub.addEventListener("message", (evt) => {
console.log(
`node1 received: ${uint8ArrayToString(evt.detail.data)} on topic ${
evt.detail.topic
}`
);
});
setInterval(() => {
node2.services.pubsub
.publish(topic, uint8ArrayFromString("Bird bird bird, bird is the word!"))
.catch((err) => {
console.error(err);
});
}, 10);
})();
Note
This is mentioned in a couple places: https://github.com/ChainSafe/js-libp2p-gossipsub/issues/306 https://github.com/ChainSafe/js-libp2p-gossipsub/pull/293
Suggested fix
Currently lightPush.send
makes a new stream for every method. In the comments in the two issues above it seems js-libp2p's expectation is that a protocol should only have one open stream and send all messages over that stream instead of opening a new stream per message.
https://github.com/waku-org/js-waku/blob/89392dbfdf5471e540ec563980f34ecbfcc9981f/packages/core/src/lib/light_push/index.ts#L49