1
0
mirror of https://github.com/redis/node-redis.git synced 2025-08-16 08:41:57 +03:00

use socket.setNoDelay and queueMicrotask to improve latency

This commit is contained in:
leibale
2021-07-19 17:38:48 -04:00
parent 81456fc1fc
commit a84d9eeefb
3 changed files with 22 additions and 3 deletions

View File

@@ -444,20 +444,29 @@ export default class RedisClient<M extends RedisModules = RedisModules, S extend
return this.#socket.disconnect(); return this.#socket.disconnect();
} }
#isTickQueued = false;
#tick(): void { #tick(): void {
const {chunkRecommendedSize} = this.#socket; const {chunkRecommendedSize} = this.#socket;
if (!chunkRecommendedSize) { if (!chunkRecommendedSize) {
return; return;
} }
// TODO: batch using process.nextTick? maybe socket.setNoDelay(false)? if (!this.#isTickQueued && this.#queue.waitingToBeSentCommandsLength < chunkRecommendedSize) {
queueMicrotask(() => this.#tick());
this.#isTickQueued = true;
return;
}
const isBuffering = this.#queue.executeChunk(chunkRecommendedSize); const isBuffering = this.#queue.executeChunk(chunkRecommendedSize);
if (isBuffering === true) { if (isBuffering === true) {
this.#socket.once('drain', () => this.#tick()); this.#socket.once('drain', () => this.#tick());
} else if (isBuffering === false) { } else if (isBuffering === false) {
this.#tick(); this.#tick();
return;
} }
this.#isTickQueued = false;
} }
} }

View File

@@ -74,6 +74,12 @@ export default class RedisCommandsQueue {
readonly #waitingToBeSent = new LinkedList<CommandWaitingToBeSent>(); readonly #waitingToBeSent = new LinkedList<CommandWaitingToBeSent>();
#waitingToBeSentCommandsLength = 0;
get waitingToBeSentCommandsLength() {
return this.#waitingToBeSentCommandsLength;
}
readonly #waitingForReply = new LinkedList<CommandWaitingForReply>(); readonly #waitingForReply = new LinkedList<CommandWaitingForReply>();
readonly #pubSubState = { readonly #pubSubState = {
@@ -97,7 +103,7 @@ export default class RedisCommandsQueue {
reply[2], reply[2],
reply[1] reply[1]
); );
case 'pmessage': case 'pmessage':
return RedisCommandsQueue.#emitPubSubMessage( return RedisCommandsQueue.#emitPubSubMessage(
this.#pubSubListeners.patterns.get(reply[1])!, this.#pubSubListeners.patterns.get(reply[1])!,
@@ -108,7 +114,7 @@ export default class RedisCommandsQueue {
case 'subscribe': case 'subscribe':
case 'psubscribe': case 'psubscribe':
if (--this.#waitingForReply.head!.value.channelsCounter! === 0) { if (--this.#waitingForReply.head!.value.channelsCounter! === 0) {
this.#shiftWaitingForReply().resolve(); this.#shiftWaitingForReply().resolve();
} }
return; return;
} }
@@ -185,6 +191,8 @@ export default class RedisCommandsQueue {
} else { } else {
this.#waitingToBeSent.pushNode(node); this.#waitingToBeSent.pushNode(node);
} }
this.#waitingToBeSentCommandsLength += encodedCommand.length;
}); });
} }
@@ -325,6 +333,7 @@ export default class RedisCommandsQueue {
} }
this.#chainInExecution = lastCommandChainId; this.#chainInExecution = lastCommandChainId;
this.#waitingToBeSentCommandsLength -= size;
} }
parseResponse(data: Buffer): void { parseResponse(data: Buffer): void {

View File

@@ -159,6 +159,7 @@ export default class RedisSocket extends EventEmitter {
this.#createNetSocket(); this.#createNetSocket();
socket socket
.setNoDelay()
.once('error', (err) => reject(err)) .once('error', (err) => reject(err))
.once(connectEvent, () => { .once(connectEvent, () => {
socket socket