diff options
author | GitLab Bot <gitlab-bot@gitlab.com> | 2023-03-06 21:08:12 +0300 |
---|---|---|
committer | GitLab Bot <gitlab-bot@gitlab.com> | 2023-03-06 21:08:12 +0300 |
commit | e22c3819ad2321a0cf825877fe3b60e41268c5b3 (patch) | |
tree | fcd143b30bdd7b42d439cd0b2fc5c6c4268d8d97 /app/assets/javascripts/streaming | |
parent | 49b16b71778148e9f9c579bf7bf69853c780c827 (diff) |
Add latest changes from gitlab-org/gitlab@master
Diffstat (limited to 'app/assets/javascripts/streaming')
8 files changed, 380 insertions, 0 deletions
diff --git a/app/assets/javascripts/streaming/chunk_writer.js b/app/assets/javascripts/streaming/chunk_writer.js new file mode 100644 index 00000000000..4bbd0a5f843 --- /dev/null +++ b/app/assets/javascripts/streaming/chunk_writer.js @@ -0,0 +1,144 @@ +import { throttle } from 'lodash'; +import { RenderBalancer } from '~/streaming/render_balancer'; +import { + BALANCE_RATE, + HIGH_FRAME_TIME, + LOW_FRAME_TIME, + MAX_CHUNK_SIZE, + MIN_CHUNK_SIZE, + TIMEOUT, +} from '~/streaming/constants'; + +const defaultConfig = { + balanceRate: BALANCE_RATE, + minChunkSize: MIN_CHUNK_SIZE, + maxChunkSize: MAX_CHUNK_SIZE, + lowFrameTime: LOW_FRAME_TIME, + highFrameTime: HIGH_FRAME_TIME, + timeout: TIMEOUT, +}; + +function concatUint8Arrays(a, b) { + const array = new Uint8Array(a.length + b.length); + array.set(a, 0); + array.set(b, a.length); + return array; +} + +// This class is used to write chunks with a balanced size +// to avoid blocking main thread for too long. +// +// A chunk can be: +// 1. Too small +// 2. Too large +// 3. Delayed in time +// +// This class resolves all these problems by +// 1. Splitting or concatenating chunks to met the size criteria +// 2. Rendering current chunk buffer immediately if enough time has passed +// +// The size of the chunk is determined by RenderBalancer, +// It measures execution time for each chunk write and adjusts next chunk size. +export class ChunkWriter { + buffer = null; + decoder = new TextDecoder('utf-8'); + timeout = null; + + constructor(htmlStream, config) { + this.htmlStream = htmlStream; + + const { balanceRate, minChunkSize, maxChunkSize, lowFrameTime, highFrameTime, timeout } = { + ...defaultConfig, + ...config, + }; + + // ensure we still render chunks over time if the size criteria is not met + this.scheduleAccumulatorFlush = throttle(this.flushAccumulator.bind(this), timeout); + + const averageSize = Math.round((maxChunkSize + minChunkSize) / 2); + this.size = Math.max(averageSize, minChunkSize); + + this.balancer = new RenderBalancer({ + lowFrameTime, + highFrameTime, + decrease: () => { + this.size = Math.round(Math.max(this.size / balanceRate, minChunkSize)); + }, + increase: () => { + this.size = Math.round(Math.min(this.size * balanceRate, maxChunkSize)); + }, + }); + } + + write(chunk) { + this.scheduleAccumulatorFlush.cancel(); + + if (this.buffer) { + this.buffer = concatUint8Arrays(this.buffer, chunk); + } else { + this.buffer = chunk; + } + + // accumulate chunks until the size is fulfilled + if (this.size > this.buffer.length) { + this.scheduleAccumulatorFlush(); + return Promise.resolve(); + } + + return this.balancedWrite(); + } + + balancedWrite() { + let cursor = 0; + + return this.balancer.render(() => { + const chunkPart = this.buffer.subarray(cursor, cursor + this.size); + // accumulate chunks until the size is fulfilled + // this is a hot path for the last chunkPart of the chunk + if (chunkPart.length < this.size) { + this.buffer = chunkPart; + this.scheduleAccumulatorFlush(); + return false; + } + + this.writeToDom(chunkPart); + + cursor += this.size; + if (cursor >= this.buffer.length) { + this.buffer = null; + return false; + } + // continue render + return true; + }); + } + + writeToDom(chunk, stream = true) { + // stream: true allows us to split chunks with multi-part words + const decoded = this.decoder.decode(chunk, { stream }); + this.htmlStream.write(decoded); + } + + flushAccumulator() { + if (this.buffer) { + this.writeToDom(this.buffer); + this.buffer = null; + } + } + + close() { + this.scheduleAccumulatorFlush.cancel(); + if (this.buffer) { + // last chunk should have stream: false to indicate the end of the stream + this.writeToDom(this.buffer, false); + this.buffer = null; + } + this.htmlStream.close(); + } + + abort() { + this.scheduleAccumulatorFlush.cancel(); + this.buffer = null; + this.htmlStream.abort(); + } +} diff --git a/app/assets/javascripts/streaming/constants.js b/app/assets/javascripts/streaming/constants.js new file mode 100644 index 00000000000..224d93a7ac1 --- /dev/null +++ b/app/assets/javascripts/streaming/constants.js @@ -0,0 +1,9 @@ +// Lower min chunk numbers can make the page loading take incredibly long +export const MIN_CHUNK_SIZE = 128 * 1024; +export const MAX_CHUNK_SIZE = 2048 * 1024; +export const LOW_FRAME_TIME = 32; +// Tasks that take more than 50ms are considered Long +// https://web.dev/optimize-long-tasks/ +export const HIGH_FRAME_TIME = 64; +export const BALANCE_RATE = 1.2; +export const TIMEOUT = 500; diff --git a/app/assets/javascripts/streaming/handle_streamed_anchor_link.js b/app/assets/javascripts/streaming/handle_streamed_anchor_link.js new file mode 100644 index 00000000000..315dc9bb0a0 --- /dev/null +++ b/app/assets/javascripts/streaming/handle_streamed_anchor_link.js @@ -0,0 +1,26 @@ +import { throttle } from 'lodash'; +import { scrollToElement } from '~/lib/utils/common_utils'; +import LineHighlighter from '~/blob/line_highlighter'; + +const noop = () => {}; + +export function handleStreamedAnchorLink(rootElement) { + // "#L100-200" → ['L100', 'L200'] + const [anchorStart, end] = window.location.hash.substring(1).split('-'); + const anchorEnd = end ? `L${end}` : anchorStart; + if (!anchorStart || document.getElementById(anchorEnd)) return noop; + + const handler = throttle((mutationList, instance) => { + if (!document.getElementById(anchorEnd)) return; + scrollToElement(document.getElementById(anchorStart)); + // eslint-disable-next-line no-new + new LineHighlighter(); + instance.disconnect(); + }, 300); + + const observer = new MutationObserver(handler); + + observer.observe(rootElement, { childList: true, subtree: true }); + + return () => observer.disconnect(); +} diff --git a/app/assets/javascripts/streaming/html_stream.js b/app/assets/javascripts/streaming/html_stream.js new file mode 100644 index 00000000000..8182f69a607 --- /dev/null +++ b/app/assets/javascripts/streaming/html_stream.js @@ -0,0 +1,33 @@ +import { ChunkWriter } from '~/streaming/chunk_writer'; + +export class HtmlStream { + constructor(element) { + const streamDocument = document.implementation.createHTMLDocument('stream'); + + streamDocument.open(); + streamDocument.write('<streaming-element>'); + + const virtualStreamingElement = streamDocument.querySelector('streaming-element'); + element.appendChild(document.adoptNode(virtualStreamingElement)); + + this.streamDocument = streamDocument; + } + + withChunkWriter(config) { + return new ChunkWriter(this, config); + } + + write(chunk) { + // eslint-disable-next-line no-unsanitized/method + this.streamDocument.write(chunk); + } + + close() { + this.streamDocument.write('</streaming-element>'); + this.streamDocument.close(); + } + + abort() { + this.streamDocument.close(); + } +} diff --git a/app/assets/javascripts/streaming/polyfills.js b/app/assets/javascripts/streaming/polyfills.js new file mode 100644 index 00000000000..a9a044a3e99 --- /dev/null +++ b/app/assets/javascripts/streaming/polyfills.js @@ -0,0 +1,5 @@ +import { createReadableStreamWrapper } from '@mattiasbuelens/web-streams-adapter'; +import { ReadableStream as PolyfillReadableStream } from 'web-streams-polyfill'; + +// TODO: remove this when our WebStreams API reaches 100% support +export const toPolyfillReadable = createReadableStreamWrapper(PolyfillReadableStream); diff --git a/app/assets/javascripts/streaming/rate_limit_stream_requests.js b/app/assets/javascripts/streaming/rate_limit_stream_requests.js new file mode 100644 index 00000000000..04a592baa16 --- /dev/null +++ b/app/assets/javascripts/streaming/rate_limit_stream_requests.js @@ -0,0 +1,87 @@ +const consumeReadableStream = (stream) => { + return new Promise((resolve, reject) => { + stream.pipeTo( + new WritableStream({ + close: resolve, + abort: reject, + }), + ); + }); +}; + +const wait = (timeout) => + new Promise((resolve) => { + setTimeout(resolve, timeout); + }); + +// this rate-limiting approach is specific to Web Streams +// because streams only resolve when they're fully consumed +// so we need to split each stream into two pieces: +// one for the rate-limiter (wait for all the bytes to be sent) +// another for the original consumer +export const rateLimitStreamRequests = ({ + factory, + total, + maxConcurrentRequests, + immediateCount = maxConcurrentRequests, + timeout = 0, +}) => { + if (total === 0) return []; + + const unsettled = []; + + const pushUnsettled = (promise) => { + let res; + let rej; + const consume = new Promise((resolve, reject) => { + res = resolve; + rej = reject; + }); + unsettled.push(consume); + return promise.then((stream) => { + const [first, second] = stream.tee(); + // eslint-disable-next-line promise/no-nesting + consumeReadableStream(first) + .then(() => { + unsettled.splice(unsettled.indexOf(consume), 1); + res(); + }) + .catch(rej); + return second; + }, rej); + }; + + const immediate = Array.from({ length: Math.min(immediateCount, total) }, (_, i) => + pushUnsettled(factory(i)), + ); + + const queue = []; + const flushQueue = () => { + const promises = + unsettled.length > maxConcurrentRequests ? unsettled : [...unsettled, wait(timeout)]; + // errors are handled by the caller + // eslint-disable-next-line promise/catch-or-return + Promise.race(promises).then(() => { + const cb = queue.shift(); + cb?.(); + if (queue.length !== 0) { + // wait for stream consumer promise to be removed from unsettled + queueMicrotask(flushQueue); + } + }); + }; + + const throttled = Array.from({ length: total - immediateCount }, (_, i) => { + return new Promise((resolve, reject) => { + queue.push(() => { + pushUnsettled(factory(i + immediateCount)) + .then(resolve) + .catch(reject); + }); + }); + }); + + flushQueue(); + + return [...immediate, ...throttled]; +}; diff --git a/app/assets/javascripts/streaming/render_balancer.js b/app/assets/javascripts/streaming/render_balancer.js new file mode 100644 index 00000000000..66929ff3a54 --- /dev/null +++ b/app/assets/javascripts/streaming/render_balancer.js @@ -0,0 +1,36 @@ +export class RenderBalancer { + previousTimestamp = undefined; + + constructor({ increase, decrease, highFrameTime, lowFrameTime }) { + this.increase = increase; + this.decrease = decrease; + this.highFrameTime = highFrameTime; + this.lowFrameTime = lowFrameTime; + } + + render(fn) { + return new Promise((resolve) => { + const callback = (timestamp) => { + this.throttle(timestamp); + if (fn()) requestAnimationFrame(callback); + else resolve(); + }; + requestAnimationFrame(callback); + }); + } + + throttle(timestamp) { + const { previousTimestamp } = this; + this.previousTimestamp = timestamp; + if (previousTimestamp === undefined) return; + + const duration = Math.round(timestamp - previousTimestamp); + if (!duration) return; + + if (duration >= this.highFrameTime) { + this.decrease(); + } else if (duration < this.lowFrameTime) { + this.increase(); + } + } +} diff --git a/app/assets/javascripts/streaming/render_html_streams.js b/app/assets/javascripts/streaming/render_html_streams.js new file mode 100644 index 00000000000..7201e541777 --- /dev/null +++ b/app/assets/javascripts/streaming/render_html_streams.js @@ -0,0 +1,40 @@ +import { HtmlStream } from '~/streaming/html_stream'; + +async function pipeStreams(domWriter, streamPromises) { + try { + for await (const stream of streamPromises.slice(0, -1)) { + await stream.pipeTo(domWriter, { preventClose: true }); + } + const stream = await streamPromises[streamPromises.length - 1]; + await stream.pipeTo(domWriter); + } catch (error) { + domWriter.abort(error); + } +} + +// this function (and the rest of the pipeline) expects polyfilled streams +// do not pass native streams here unless our browser support allows for it +// TODO: remove this notice when our WebStreams API support reaches 100% +export function renderHtmlStreams(streamPromises, element, config) { + if (streamPromises.length === 0) return Promise.resolve(); + + const chunkedHtmlStream = new HtmlStream(element).withChunkWriter(config); + + return new Promise((resolve, reject) => { + const domWriter = new WritableStream({ + write(chunk) { + return chunkedHtmlStream.write(chunk); + }, + close() { + chunkedHtmlStream.close(); + resolve(); + }, + abort(error) { + chunkedHtmlStream.abort(); + reject(error); + }, + }); + + pipeStreams(domWriter, streamPromises); + }); +} |