nodejs / nodejs/node

quic: Bug in flow control for bidistreams and sessions?

オープン
#63,304 コメント 2 件 リアクション 0 件 担当者 0 名 GitHub で見る

まだ誰も着手していません。

never-stale quic
主要言語
JavaScript
スター
122k
フォーク
37.3k
平均マージ
4日 2時間
マージ済み PR(30日)
283

説明

I wrote a test in order to trigger stalls caused by session flow control. (So exhaustion of the flow control window of the session).
I am not sure if I achieved it, but it stalls. So it might be something else.
I built a node on Windows, not on the current main, but on https://github.com/nodejs/node/pull/63230, so that I am sure that it is another issue.

The test is based on portions from different other node.js tests (test-quic-stream-bidi-echo-concurrent.mjs):

// Flags: --experimental-quic --experimental-stream-iter --no-warnings

// Test: bidirectional stream echo with binary.
// The client sends a message, the server reads it and echoes it back.
// Both directions of the bidi stream carry data and are properly FIN'd.
// Verifies that both client and server can read and write on the same stream.

import { hasQuic, skip, mustCall } from '../common/index.mjs';
import assert from 'node:assert';

const { strictEqual } = assert;

if (!hasQuic) {
  skip('QUIC is not enabled');
}

const { listen, connect } = await import('../common/quic.mjs');
const { bytes, drainableProtocol: dp } = await import('stream/iter');

const chunkSizes = [60000, 12, 1000000, 50000, 1600, 20000, 1000000, 30000, 0, 100]
const numChunks = chunkSizes.length
const byteLength = chunkSizes.reduce((accumulator, currentValue) => accumulator + currentValue, 0);


// Build a deterministic payload so we can verify integrity.
function buildChunk(index) {
  const chunk = new Uint8Array(chunkSizes[index]);
  // Fill with a pattern derived from the chunk index.
  const val = index & 0xff;
  for (let i = 0; i < chunkSizes[index]; i++) {
    chunk[i] = (val + i) & 0xff;
  }
  return chunk;
}

function checksum(data) {
  let sum = 0;
  for (let i = 0; i < data.byteLength; i++) {
    sum = (sum + data[i]) | 0;
  }
  return sum;
}

// Compute expected checksum.
let expectedChecksum = 0;
for (let i = 0; i < numChunks; i++) {
  const chunk = buildChunk(i);
  expectedChecksum = (expectedChecksum + checksum(chunk)) | 0;
}

const maxStreams = 20;
let pendStreams = maxStreams;
const done = Promise.withResolvers();

const serverEndpoint = await listen(mustCall((serverSession) => {
  serverSession.onstream = mustCall(async (stream) => {
    const writer = stream.writer
    let written = 0;
    for await (const chunks of stream) {
      for (const chunk of chunks) {
        written += chunk.byteLength;

        while (!writer.writeSync(chunk)) {
          // Flow controlled — wait for drain before retrying.
          const drainable = writer[dp]();
          if (drainable) await drainable;
        }
      }
    }
    writer.end();

    await stream.closed;
    pendStreams--;
    console.log('end pendstreams', pendStreams)
    if (pendStreams === 0) {
      serverSession.close();
      done.resolve();
    }
  });
}));

const clientSession = await connect(serverEndpoint.address);
await clientSession.opened;





const readFromStream = async (stream) => {
  const readChunks = [];
  for await (const chunks of stream) {
    readChunks.push(...chunks);
  }
  const receivedBytes = readChunks.reduce((accu, curVal) => accu + curVal.byteLength, 0);

  strictEqual(receivedBytes, byteLength);

  let receivedChecksum = 0;
  for (const chunk of readChunks) {
    receivedChecksum = (receivedChecksum + checksum(chunk)) | 0;
  }

  strictEqual(receivedChecksum, expectedChecksum);
}


const writeToStream = async (stream) => {
  const w = stream.writer
  for (let i = 0; i < numChunks; i++) {
    const chunk = buildChunk(i);
    while (!w.writeSync(chunk)) {
      // Flow controlled — wait for drain before retrying.
      const drainable = w[dp]();
      if (drainable) await drainable;
    }
  }
  w.endSync();
}

const streamClosed = []

await Promise.all(Array.from({ length: 20 }, async () => {
  const stream = await clientSession.createBidirectionalStream()
  streamClosed.push(stream.closed);
  await Promise.all([readFromStream(stream), writeToStream(stream)])
}))


await Promise.all([...streamClosed, done.promise]);
await clientSession.close();
await serverEndpoint.close();

The results are flaky and stall after different number of streams. For example:

end pendstreams 19
end pendstreams 18
end pendstreams 17
end pendstreams 16
end pendstreams 15
end pendstreams 14
end pendstreams 13
end pendstreams 12
end pendstreams 11
end pendstreams 10
end pendstreams 9  <--- stall
node:internal/quic/quic:2122  <--- session time out I presume
      handle.endWrite();
             ^

Error: Stream is no longer writable
    at endSync (node:internal/quic/quic:2122:14)
    at Object.end (node:internal/quic/quic:2146:17)
    at QuicSession.<anonymous> (file:///C:/PrivateDaten/Projekte/failscomponents/failslib/thirdparty/nodepatch/node/test/parallel/test-quic-stream-bidi-echo-concurrent.mjs:70:12) {
  code: 'ERR_INVALID_STATE'
}

@jasnell That would be another issue.....

コントリビューションガイド

コントリビューションガイドを開く

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

調査の方向性

test/parallel/test-quic-stream-bidi-echo-concurrent.mjs から始め、issue に記載された並行双方向ストリームの再現を実行します。報告された停止とタイムアウトの周辺における session と stream のフロー制御動作を追跡します。テストが停止せず、無効な writable-state エラーも発生させずに、20 個すべてのストリームを確実に完了できれば完了です。

索引モデルが issue の本文から書いたものです。

評価

技術スタック
javascript, nodejs
領域
networking, testing-qa
issue の種類
バグ
難易度
4/5
見積もり時間
3〜5日
活発さ
活発
明瞭さ
おおむね明確
初心者へのやさしさ
45/100

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。