third_party/perfetto/ui/src/common/recordingV2/traced_tracing_session.ts - cobalt - Git at Google

 // Copyright (C) 2022 The Android Open Source Project
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //      http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.

 import protobuf from 'protobufjs/minimal';

 import {defer, Deferred} from '../../base/deferred';
 import {assertExists, assertFalse, assertTrue} from '../../base/logging';
 import {
   DisableTracingRequest,
   DisableTracingResponse,
   EnableTracingRequest,
   EnableTracingResponse,
   FreeBuffersRequest,
   FreeBuffersResponse,
   GetTraceStatsRequest,
   GetTraceStatsResponse,
   IBufferStats,
   IMethodInfo,
   IPCFrame,
   ISlice,
   QueryServiceStateRequest,
   QueryServiceStateResponse,
   ReadBuffersRequest,
   ReadBuffersResponse,
   TraceConfig,
 } from '../protos';

 import {RecordingError} from './recording_error_handling';
 import {
   ByteStream,
   DataSource,
   TracingSession,
   TracingSessionListener,
 } from './recording_interfaces_v2';
 import {
   BUFFER_USAGE_INCORRECT_FORMAT,
   BUFFER_USAGE_NOT_ACCESSIBLE,
   PARSING_UNABLE_TO_DECODE_METHOD,
   PARSING_UNKNWON_REQUEST_ID,
   PARSING_UNRECOGNIZED_MESSAGE,
   PARSING_UNRECOGNIZED_PORT,
   RECORDING_IN_PROGRESS,
 } from './recording_utils';

 // See wire_protocol.proto for more details.
 const WIRE_PROTOCOL_HEADER_SIZE = 4;
 // See basic_types.h (kIPCBufferSize) for more details.
 const MAX_IPC_BUFFER_SIZE = 128 * 1024;

 const PROTO_LEN_DELIMITED_WIRE_TYPE = 2;
 const TRACE_PACKET_PROTO_ID = 1;
 const TRACE_PACKET_PROTO_TAG =
     (TRACE_PACKET_PROTO_ID << 3) | PROTO_LEN_DELIMITED_WIRE_TYPE;

 function parseMessageSize(buffer: Uint8Array) {
   const dv = new DataView(buffer.buffer, buffer.byteOffset, buffer.length);
   return dv.getUint32(0, true);
 }

 // This class implements the protocol described in
 // https://perfetto.dev/docs/design-docs/api-and-abi#tracing-protocol-abi
 export class TracedTracingSession implements TracingSession {
   // Buffers received wire protocol data.
   private incomingBuffer = new Uint8Array(MAX_IPC_BUFFER_SIZE);
   private bufferedPartLength = 0;
   private currentFrameLength?: number;

   private availableMethods: IMethodInfo[] = [];
   private serviceId = -1;

   private resolveBindingPromise!: Deferred<void>;
   private requestMethods = new Map<number, string>();

   // Needed for ReadBufferResponse: all the trace packets are split into
   // several slices. |partialPacket| is the buffer for them. Once we receive a
   // slice with the flag |lastSliceForPacket|, a new packet is created.
   private partialPacket: ISlice[] = [];
   // Accumulates trace packets into a proto trace file..
   private traceProtoWriter = protobuf.Writer.create();

   // Accumulates DataSource objects from QueryServiceStateResponse,
   // which can have >1 replies for each query
   // go/codesearch/android/external/perfetto/protos/
   // perfetto/ipc/consumer_port.proto;l=243-246
   private pendingDataSources: DataSource[] = [];

   // For concurrent calls to 'QueryServiceState', we return the same value.
   private pendingQssMessage?: Deferred<DataSource[]>;

   // Wire protocol request ID. After each request it is increased. It is needed
   // to keep track of the type of request, and parse the response correctly.
   private requestId = 1;

   private pendingStatsMessages = new Array<Deferred<IBufferStats[]>>();

   // The bytestream is obtained when creating a connection with a target.
   // For instance, the AdbStream is obtained from a connection with an Adb
   // device.
   constructor(
       private byteStream: ByteStream,
       private tracingSessionListener: TracingSessionListener) {
     this.byteStream.addOnStreamDataCallback(
         (data) => this.handleReceivedData(data));
     this.byteStream.addOnStreamCloseCallback(() => this.clearState());
   }

   queryServiceState(): Promise<DataSource[]> {
     if (this.pendingQssMessage) {
       return this.pendingQssMessage;
     }

     const requestProto =
         QueryServiceStateRequest.encode(new QueryServiceStateRequest())
             .finish();
     this.rpcInvoke('QueryServiceState', requestProto);

     return this.pendingQssMessage = defer<DataSource[]>();
   }

   start(config: TraceConfig): void {
     const duration = config.durationMs;
     this.tracingSessionListener.onStatus(`${RECORDING_IN_PROGRESS}${
         duration ? ' for ' + duration.toString() + ' ms' : ''}...`);

     const enableTracingRequest = new EnableTracingRequest();
     enableTracingRequest.traceConfig = config;
     const enableTracingRequestProto =
         EnableTracingRequest.encode(enableTracingRequest).finish();
     this.rpcInvoke('EnableTracing', enableTracingRequestProto);
   }

   cancel(): void {
     this.terminateConnection();
   }

   stop(): void {
     const requestProto =
         DisableTracingRequest.encode(new DisableTracingRequest()).finish();
     this.rpcInvoke('DisableTracing', requestProto);
   }

   async getTraceBufferUsage(): Promise<number> {
     if (!this.byteStream.isConnected()) {
       // TODO(octaviant): make this more in line with the other trace buffer
       //  error cases.
       return 0;
     }
     const bufferStats = await this.getBufferStats();
     let percentageUsed = -1;
     for (const buffer of bufferStats) {
       if (!Number.isFinite(buffer.bytesWritten) ||
           !Number.isFinite(buffer.bufferSize)) {
         continue;
       }
       const used = assertExists(buffer.bytesWritten);
       const total = assertExists(buffer.bufferSize);
       if (total >= 0) {
         percentageUsed = Math.max(percentageUsed, used / total);
       }
     }

     if (percentageUsed === -1) {
       return Promise.reject(new RecordingError(BUFFER_USAGE_INCORRECT_FORMAT));
     }
     return percentageUsed;
   }

   initConnection(): Promise<void> {
     // bind IPC methods
     const requestId = this.requestId++;
     const frame = new IPCFrame({
       requestId,
       msgBindService: new IPCFrame.BindService({serviceName: 'ConsumerPort'}),
     });
     this.writeFrame(frame);

     // We shouldn't bind multiple times to the service in the same tracing
     // session.
     assertFalse(!!this.resolveBindingPromise);
     this.resolveBindingPromise = defer<void>();
     return this.resolveBindingPromise;
   }

   private getBufferStats(): Promise<IBufferStats[]> {
     const getTraceStatsRequestProto =
         GetTraceStatsRequest.encode(new GetTraceStatsRequest()).finish();
     try {
       this.rpcInvoke('GetTraceStats', getTraceStatsRequestProto);
     } catch (e) {
       // GetTraceStats was introduced only on Android 10.
       this.raiseError(e);
     }

     const statsMessage = defer<IBufferStats[]>();
     this.pendingStatsMessages.push(statsMessage);
     return statsMessage;
   }

   private terminateConnection(): void {
     this.clearState();
     const requestProto =
         FreeBuffersRequest.encode(new FreeBuffersRequest()).finish();
     this.rpcInvoke('FreeBuffers', requestProto);
     this.byteStream.close();
   }

   private clearState() {
     for (const statsMessage of this.pendingStatsMessages) {
       statsMessage.reject(new RecordingError(BUFFER_USAGE_NOT_ACCESSIBLE));
     }
     this.pendingStatsMessages = [];
     this.pendingDataSources = [];
     this.pendingQssMessage = undefined;
   }

   private rpcInvoke(methodName: string, argsProto: Uint8Array): void {
     if (!this.byteStream.isConnected()) {
       return;
     }
     const method = this.availableMethods.find((m) => m.name === methodName);
     if (!method || !method.id) {
       throw new RecordingError(
           `Method ${methodName} not supported by the target`);
     }
     const requestId = this.requestId++;
     const frame = new IPCFrame({
       requestId,
       msgInvokeMethod: new IPCFrame.InvokeMethod(
           {serviceId: this.serviceId, methodId: method.id, argsProto}),
     });
     this.requestMethods.set(requestId, methodName);
     this.writeFrame(frame);
   }

   private writeFrame(frame: IPCFrame): void {
     const frameProto: Uint8Array = IPCFrame.encode(frame).finish();
     const frameLen = frameProto.length;
     const buf = new Uint8Array(WIRE_PROTOCOL_HEADER_SIZE + frameLen);
     const dv = new DataView(buf.buffer);
     dv.setUint32(0, frameProto.length, /* littleEndian */ true);
     for (let i = 0; i < frameLen; i++) {
       dv.setUint8(WIRE_PROTOCOL_HEADER_SIZE + i, frameProto[i]);
     }
     this.byteStream.write(buf);
   }

   private handleReceivedData(rawData: Uint8Array): void {
     // we parse the length of the next frame if it's available
     if (this.currentFrameLength === undefined &&
         this.canCompleteLengthHeader(rawData)) {
       const remainingFrameBytes =
           WIRE_PROTOCOL_HEADER_SIZE - this.bufferedPartLength;
       this.appendToIncomingBuffer(rawData.subarray(0, remainingFrameBytes));
       rawData = rawData.subarray(remainingFrameBytes);

       this.currentFrameLength = parseMessageSize(this.incomingBuffer);
       this.bufferedPartLength = 0;
     }

     // Parse all complete frames.
     while (this.currentFrameLength !== undefined &&
            this.bufferedPartLength + rawData.length >=
                this.currentFrameLength) {
       // Read the remaining part of this message.
       const bytesToCompleteMessage =
           this.currentFrameLength - this.bufferedPartLength;
       this.appendToIncomingBuffer(rawData.subarray(0, bytesToCompleteMessage));
       this.parseFrame(this.incomingBuffer.subarray(0, this.currentFrameLength));
       this.bufferedPartLength = 0;
       // Remove the data just parsed.
       rawData = rawData.subarray(bytesToCompleteMessage);

       if (!this.canCompleteLengthHeader(rawData)) {
         this.currentFrameLength = undefined;
         break;
       }
       this.currentFrameLength = parseMessageSize(rawData);
       rawData = rawData.subarray(WIRE_PROTOCOL_HEADER_SIZE);
     }

     // Buffer the remaining data (part of the next message).
     this.appendToIncomingBuffer(rawData);
   }

   private canCompleteLengthHeader(newData: Uint8Array): boolean {
     return newData.length + this.bufferedPartLength > WIRE_PROTOCOL_HEADER_SIZE;
   }

   private appendToIncomingBuffer(array: Uint8Array): void {
     this.incomingBuffer.set(array, this.bufferedPartLength);
     this.bufferedPartLength += array.length;
   }

   private parseFrame(frameBuffer: Uint8Array): void {
     // Get a copy of the ArrayBuffer to avoid the original being overriden.
     // See 170256902#comment21
     const frame = IPCFrame.decode(frameBuffer.slice());
     if (frame.msg === 'msgBindServiceReply') {
       const msgBindServiceReply = frame.msgBindServiceReply;
       if (msgBindServiceReply && msgBindServiceReply.methods &&
           msgBindServiceReply.serviceId) {
         assertTrue(msgBindServiceReply.success === true);
         this.availableMethods = msgBindServiceReply.methods;
         this.serviceId = msgBindServiceReply.serviceId;
         this.resolveBindingPromise.resolve();
       }
     } else if (frame.msg === 'msgInvokeMethodReply') {
       const msgInvokeMethodReply = frame.msgInvokeMethodReply;
       // We process messages without a `replyProto` field (for instance
       // `FreeBuffers` does not have `replyProto`). However, we ignore messages
       // without a valid 'success' field.
       if (!msgInvokeMethodReply || !msgInvokeMethodReply.success) {
         return;
       }

       const method = this.requestMethods.get(frame.requestId);
       if (!method) {
         this.raiseError(`${PARSING_UNKNWON_REQUEST_ID}: ${frame.requestId}`);
         return;
       }
       const decoder = decoders.get(method);
       if (decoder === undefined) {
         this.raiseError(`${PARSING_UNABLE_TO_DECODE_METHOD}: ${method}`);
         return;
       }
       const data = {...decoder(msgInvokeMethodReply.replyProto)};

       if (method === 'ReadBuffers') {
         if (data.slices) {
           for (const slice of data.slices) {
             this.partialPacket.push(slice);
             if (slice.lastSliceForPacket) {
               let bufferSize = 0;
               for (const slice of this.partialPacket) {
                 bufferSize += slice.data!.length;
               }
               const tracePacket = new Uint8Array(bufferSize);
               let written = 0;
               for (const slice of this.partialPacket) {
                 const data = slice.data!;
                 tracePacket.set(data, written);
                 written += data.length;
               }
               this.traceProtoWriter.uint32(TRACE_PACKET_PROTO_TAG);
               this.traceProtoWriter.bytes(tracePacket);
               this.partialPacket = [];
             }
           }
         }
         if (msgInvokeMethodReply.hasMore === false) {
           this.tracingSessionListener.onTraceData(
               this.traceProtoWriter.finish());
           this.terminateConnection();
         }
       } else if (method === 'EnableTracing') {
         const readBuffersRequestProto =
             ReadBuffersRequest.encode(new ReadBuffersRequest()).finish();
         this.rpcInvoke('ReadBuffers', readBuffersRequestProto);
       } else if (method === 'GetTraceStats') {
         const maybePendingStatsMessage = this.pendingStatsMessages.shift();
         if (maybePendingStatsMessage) {
           maybePendingStatsMessage.resolve(data?.traceStats?.bufferStats || []);
         }
       } else if (method === 'FreeBuffers') {
         // No action required. If we successfully read a whole trace,
         // we close the connection. Alternatively, if the tracing finishes
         // with an exception or if the user cancels it, we also close the
         // connection.
       } else if (method === 'DisableTracing') {
         // No action required. Same reasoning as for FreeBuffers.
       } else if (method === 'QueryServiceState') {
         const dataSources =
             (data as QueryServiceStateResponse)?.serviceState?.dataSources ||
             [];
         for (const dataSource of dataSources) {
           const name = dataSource?.dsDescriptor?.name;
           if (name) {
             this.pendingDataSources.push(
                 {name, descriptor: dataSource.dsDescriptor});
           }
         }
         if (msgInvokeMethodReply.hasMore === false) {
           assertExists(this.pendingQssMessage).resolve(this.pendingDataSources);
           this.pendingDataSources = [];
           this.pendingQssMessage = undefined;
         }
       } else {
         this.raiseError(`${PARSING_UNRECOGNIZED_PORT}: ${method}`);
       }
     } else {
       this.raiseError(`${PARSING_UNRECOGNIZED_MESSAGE}: ${frame.msg}`);
     }
   }

   private raiseError(message: string): void {
     this.terminateConnection();
     this.tracingSessionListener.onError(message);
   }
 }

 const decoders =
     new Map<string, Function>()
         .set('EnableTracing', EnableTracingResponse.decode)
         .set('FreeBuffers', FreeBuffersResponse.decode)
         .set('ReadBuffers', ReadBuffersResponse.decode)
         .set('DisableTracing', DisableTracingResponse.decode)
         .set('GetTraceStats', GetTraceStatsResponse.decode)
         .set('QueryServiceState', QueryServiceStateResponse.decode);
	// Copyright (C) 2022 The Android Open Source Project
	//
	// Licensed under the Apache License, Version 2.0 (the "License");
	// you may not use this file except in compliance with the License.
	// You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and
	// limitations under the License.

	import protobuf from 'protobufjs/minimal';

	import {defer, Deferred} from '../../base/deferred';
	import {assertExists, assertFalse, assertTrue} from '../../base/logging';
	import {
	DisableTracingRequest,
	DisableTracingResponse,
	EnableTracingRequest,
	EnableTracingResponse,
	FreeBuffersRequest,
	FreeBuffersResponse,
	GetTraceStatsRequest,
	GetTraceStatsResponse,
	IBufferStats,
	IMethodInfo,
	IPCFrame,
	ISlice,
	QueryServiceStateRequest,
	QueryServiceStateResponse,
	ReadBuffersRequest,
	ReadBuffersResponse,
	TraceConfig,
	} from '../protos';

	import {RecordingError} from './recording_error_handling';
	import {
	ByteStream,
	DataSource,
	TracingSession,
	TracingSessionListener,
	} from './recording_interfaces_v2';
	import {
	BUFFER_USAGE_INCORRECT_FORMAT,
	BUFFER_USAGE_NOT_ACCESSIBLE,
	PARSING_UNABLE_TO_DECODE_METHOD,
	PARSING_UNKNWON_REQUEST_ID,
	PARSING_UNRECOGNIZED_MESSAGE,
	PARSING_UNRECOGNIZED_PORT,
	RECORDING_IN_PROGRESS,
	} from './recording_utils';

	// See wire_protocol.proto for more details.
	const WIRE_PROTOCOL_HEADER_SIZE = 4;
	// See basic_types.h (kIPCBufferSize) for more details.
	const MAX_IPC_BUFFER_SIZE = 128 * 1024;

	const PROTO_LEN_DELIMITED_WIRE_TYPE = 2;
	const TRACE_PACKET_PROTO_ID = 1;
	const TRACE_PACKET_PROTO_TAG =
	(TRACE_PACKET_PROTO_ID << 3) \| PROTO_LEN_DELIMITED_WIRE_TYPE;

	function parseMessageSize(buffer: Uint8Array) {
	const dv = new DataView(buffer.buffer, buffer.byteOffset, buffer.length);
	return dv.getUint32(0, true);
	}

	// This class implements the protocol described in
	// https://perfetto.dev/docs/design-docs/api-and-abi#tracing-protocol-abi
	export class TracedTracingSession implements TracingSession {
	// Buffers received wire protocol data.
	private incomingBuffer = new Uint8Array(MAX_IPC_BUFFER_SIZE);
	private bufferedPartLength = 0;
	private currentFrameLength?: number;

	private availableMethods: IMethodInfo[] = [];
	private serviceId = -1;

	private resolveBindingPromise!: Deferred<void>;
	private requestMethods = new Map<number, string>();

	// Needed for ReadBufferResponse: all the trace packets are split into
	// several slices. \|partialPacket\| is the buffer for them. Once we receive a
	// slice with the flag \|lastSliceForPacket\|, a new packet is created.
	private partialPacket: ISlice[] = [];
	// Accumulates trace packets into a proto trace file..
	private traceProtoWriter = protobuf.Writer.create();

	// Accumulates DataSource objects from QueryServiceStateResponse,
	// which can have >1 replies for each query
	// go/codesearch/android/external/perfetto/protos/
	// perfetto/ipc/consumer_port.proto;l=243-246
	private pendingDataSources: DataSource[] = [];

	// For concurrent calls to 'QueryServiceState', we return the same value.
	private pendingQssMessage?: Deferred<DataSource[]>;

	// Wire protocol request ID. After each request it is increased. It is needed
	// to keep track of the type of request, and parse the response correctly.
	private requestId = 1;

	private pendingStatsMessages = new Array<Deferred<IBufferStats[]>>();

	// The bytestream is obtained when creating a connection with a target.
	// For instance, the AdbStream is obtained from a connection with an Adb
	// device.
	constructor(
	private byteStream: ByteStream,
	private tracingSessionListener: TracingSessionListener) {
	this.byteStream.addOnStreamDataCallback(
	(data) => this.handleReceivedData(data));
	this.byteStream.addOnStreamCloseCallback(() => this.clearState());
	}

	queryServiceState(): Promise<DataSource[]> {
	if (this.pendingQssMessage) {
	return this.pendingQssMessage;
	}

	const requestProto =
	QueryServiceStateRequest.encode(new QueryServiceStateRequest())
	.finish();
	this.rpcInvoke('QueryServiceState', requestProto);

	return this.pendingQssMessage = defer<DataSource[]>();
	}

	start(config: TraceConfig): void {
	const duration = config.durationMs;
	this.tracingSessionListener.onStatus(`${RECORDING_IN_PROGRESS}${
	duration ? ' for ' + duration.toString() + ' ms' : ''}...`);

	const enableTracingRequest = new EnableTracingRequest();
	enableTracingRequest.traceConfig = config;
	const enableTracingRequestProto =
	EnableTracingRequest.encode(enableTracingRequest).finish();
	this.rpcInvoke('EnableTracing', enableTracingRequestProto);
	}

	cancel(): void {
	this.terminateConnection();
	}

	stop(): void {
	const requestProto =
	DisableTracingRequest.encode(new DisableTracingRequest()).finish();
	this.rpcInvoke('DisableTracing', requestProto);
	}

	async getTraceBufferUsage(): Promise<number> {
	if (!this.byteStream.isConnected()) {
	// TODO(octaviant): make this more in line with the other trace buffer
	// error cases.
	return 0;
	}
	const bufferStats = await this.getBufferStats();
	let percentageUsed = -1;
	for (const buffer of bufferStats) {
	if (!Number.isFinite(buffer.bytesWritten) \|\|
	!Number.isFinite(buffer.bufferSize)) {
	continue;
	}
	const used = assertExists(buffer.bytesWritten);
	const total = assertExists(buffer.bufferSize);
	if (total >= 0) {
	percentageUsed = Math.max(percentageUsed, used / total);
	}
	}

	if (percentageUsed === -1) {
	return Promise.reject(new RecordingError(BUFFER_USAGE_INCORRECT_FORMAT));
	}
	return percentageUsed;
	}

	initConnection(): Promise<void> {
	// bind IPC methods
	const requestId = this.requestId++;
	const frame = new IPCFrame({
	requestId,
	msgBindService: new IPCFrame.BindService({serviceName: 'ConsumerPort'}),
	});
	this.writeFrame(frame);

	// We shouldn't bind multiple times to the service in the same tracing
	// session.
	assertFalse(!!this.resolveBindingPromise);
	this.resolveBindingPromise = defer<void>();
	return this.resolveBindingPromise;
	}

	private getBufferStats(): Promise<IBufferStats[]> {
	const getTraceStatsRequestProto =
	GetTraceStatsRequest.encode(new GetTraceStatsRequest()).finish();
	try {
	this.rpcInvoke('GetTraceStats', getTraceStatsRequestProto);
	} catch (e) {
	// GetTraceStats was introduced only on Android 10.
	this.raiseError(e);
	}

	const statsMessage = defer<IBufferStats[]>();
	this.pendingStatsMessages.push(statsMessage);
	return statsMessage;
	}

	private terminateConnection(): void {
	this.clearState();
	const requestProto =
	FreeBuffersRequest.encode(new FreeBuffersRequest()).finish();
	this.rpcInvoke('FreeBuffers', requestProto);
	this.byteStream.close();
	}

	private clearState() {
	for (const statsMessage of this.pendingStatsMessages) {
	statsMessage.reject(new RecordingError(BUFFER_USAGE_NOT_ACCESSIBLE));
	}
	this.pendingStatsMessages = [];
	this.pendingDataSources = [];
	this.pendingQssMessage = undefined;
	}

	private rpcInvoke(methodName: string, argsProto: Uint8Array): void {
	if (!this.byteStream.isConnected()) {
	return;
	}
	const method = this.availableMethods.find((m) => m.name === methodName);
	if (!method \|\| !method.id) {
	throw new RecordingError(
	`Method ${methodName} not supported by the target`);
	}
	const requestId = this.requestId++;
	const frame = new IPCFrame({
	requestId,
	msgInvokeMethod: new IPCFrame.InvokeMethod(
	{serviceId: this.serviceId, methodId: method.id, argsProto}),
	});
	this.requestMethods.set(requestId, methodName);
	this.writeFrame(frame);
	}

	private writeFrame(frame: IPCFrame): void {
	const frameProto: Uint8Array = IPCFrame.encode(frame).finish();
	const frameLen = frameProto.length;
	const buf = new Uint8Array(WIRE_PROTOCOL_HEADER_SIZE + frameLen);
	const dv = new DataView(buf.buffer);
	dv.setUint32(0, frameProto.length, /* littleEndian */ true);
	for (let i = 0; i < frameLen; i++) {
	dv.setUint8(WIRE_PROTOCOL_HEADER_SIZE + i, frameProto[i]);
	}
	this.byteStream.write(buf);
	}

	private handleReceivedData(rawData: Uint8Array): void {
	// we parse the length of the next frame if it's available
	if (this.currentFrameLength === undefined &&
	this.canCompleteLengthHeader(rawData)) {
	const remainingFrameBytes =
	WIRE_PROTOCOL_HEADER_SIZE - this.bufferedPartLength;
	this.appendToIncomingBuffer(rawData.subarray(0, remainingFrameBytes));
	rawData = rawData.subarray(remainingFrameBytes);

	this.currentFrameLength = parseMessageSize(this.incomingBuffer);
	this.bufferedPartLength = 0;
	}

	// Parse all complete frames.
	while (this.currentFrameLength !== undefined &&
	this.bufferedPartLength + rawData.length >=
	this.currentFrameLength) {
	// Read the remaining part of this message.
	const bytesToCompleteMessage =
	this.currentFrameLength - this.bufferedPartLength;
	this.appendToIncomingBuffer(rawData.subarray(0, bytesToCompleteMessage));
	this.parseFrame(this.incomingBuffer.subarray(0, this.currentFrameLength));
	this.bufferedPartLength = 0;
	// Remove the data just parsed.
	rawData = rawData.subarray(bytesToCompleteMessage);

	if (!this.canCompleteLengthHeader(rawData)) {
	this.currentFrameLength = undefined;
	break;
	}
	this.currentFrameLength = parseMessageSize(rawData);
	rawData = rawData.subarray(WIRE_PROTOCOL_HEADER_SIZE);
	}

	// Buffer the remaining data (part of the next message).
	this.appendToIncomingBuffer(rawData);
	}

	private canCompleteLengthHeader(newData: Uint8Array): boolean {
	return newData.length + this.bufferedPartLength > WIRE_PROTOCOL_HEADER_SIZE;
	}

	private appendToIncomingBuffer(array: Uint8Array): void {
	this.incomingBuffer.set(array, this.bufferedPartLength);
	this.bufferedPartLength += array.length;
	}

	private parseFrame(frameBuffer: Uint8Array): void {
	// Get a copy of the ArrayBuffer to avoid the original being overriden.
	// See 170256902#comment21
	const frame = IPCFrame.decode(frameBuffer.slice());
	if (frame.msg === 'msgBindServiceReply') {
	const msgBindServiceReply = frame.msgBindServiceReply;
	if (msgBindServiceReply && msgBindServiceReply.methods &&
	msgBindServiceReply.serviceId) {
	assertTrue(msgBindServiceReply.success === true);
	this.availableMethods = msgBindServiceReply.methods;
	this.serviceId = msgBindServiceReply.serviceId;
	this.resolveBindingPromise.resolve();
	}
	} else if (frame.msg === 'msgInvokeMethodReply') {
	const msgInvokeMethodReply = frame.msgInvokeMethodReply;
	// We process messages without a `replyProto` field (for instance
	// `FreeBuffers` does not have `replyProto`). However, we ignore messages
	// without a valid 'success' field.
	if (!msgInvokeMethodReply \|\| !msgInvokeMethodReply.success) {
	return;
	}

	const method = this.requestMethods.get(frame.requestId);
	if (!method) {
	this.raiseError(`${PARSING_UNKNWON_REQUEST_ID}: ${frame.requestId}`);
	return;
	}
	const decoder = decoders.get(method);
	if (decoder === undefined) {
	this.raiseError(`${PARSING_UNABLE_TO_DECODE_METHOD}: ${method}`);
	return;
	}
	const data = {...decoder(msgInvokeMethodReply.replyProto)};

	if (method === 'ReadBuffers') {
	if (data.slices) {
	for (const slice of data.slices) {
	this.partialPacket.push(slice);
	if (slice.lastSliceForPacket) {
	let bufferSize = 0;
	for (const slice of this.partialPacket) {
	bufferSize += slice.data!.length;
	}
	const tracePacket = new Uint8Array(bufferSize);
	let written = 0;
	for (const slice of this.partialPacket) {
	const data = slice.data!;
	tracePacket.set(data, written);
	written += data.length;
	}
	this.traceProtoWriter.uint32(TRACE_PACKET_PROTO_TAG);
	this.traceProtoWriter.bytes(tracePacket);
	this.partialPacket = [];
	}
	}
	}
	if (msgInvokeMethodReply.hasMore === false) {
	this.tracingSessionListener.onTraceData(
	this.traceProtoWriter.finish());
	this.terminateConnection();
	}
	} else if (method === 'EnableTracing') {
	const readBuffersRequestProto =
	ReadBuffersRequest.encode(new ReadBuffersRequest()).finish();
	this.rpcInvoke('ReadBuffers', readBuffersRequestProto);
	} else if (method === 'GetTraceStats') {
	const maybePendingStatsMessage = this.pendingStatsMessages.shift();
	if (maybePendingStatsMessage) {
	maybePendingStatsMessage.resolve(data?.traceStats?.bufferStats \|\| []);
	}
	} else if (method === 'FreeBuffers') {
	// No action required. If we successfully read a whole trace,
	// we close the connection. Alternatively, if the tracing finishes
	// with an exception or if the user cancels it, we also close the
	// connection.
	} else if (method === 'DisableTracing') {
	// No action required. Same reasoning as for FreeBuffers.
	} else if (method === 'QueryServiceState') {
	const dataSources =
	(data as QueryServiceStateResponse)?.serviceState?.dataSources \|\|
	[];
	for (const dataSource of dataSources) {
	const name = dataSource?.dsDescriptor?.name;
	if (name) {
	this.pendingDataSources.push(
	{name, descriptor: dataSource.dsDescriptor});
	}
	}
	if (msgInvokeMethodReply.hasMore === false) {
	assertExists(this.pendingQssMessage).resolve(this.pendingDataSources);
	this.pendingDataSources = [];
	this.pendingQssMessage = undefined;
	}
	} else {
	this.raiseError(`${PARSING_UNRECOGNIZED_PORT}: ${method}`);
	}
	} else {
	this.raiseError(`${PARSING_UNRECOGNIZED_MESSAGE}: ${frame.msg}`);
	}
	}

	private raiseError(message: string): void {
	this.terminateConnection();
	this.tracingSessionListener.onError(message);
	}
	}

	const decoders =
	new Map<string, Function>()
	.set('EnableTracing', EnableTracingResponse.decode)
	.set('FreeBuffers', FreeBuffersResponse.decode)
	.set('ReadBuffers', ReadBuffersResponse.decode)
	.set('DisableTracing', DisableTracingResponse.decode)
	.set('GetTraceStats', GetTraceStatsResponse.decode)
	.set('QueryServiceState', QueryServiceStateResponse.decode);