Vihren Atlas

ipc.net.ts ×73

Code frontier

unlabeled · c_422473d50e6f

52 tests · 8774 LOC · 40 files · introduces 0 tests · 802 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
124 ranges802 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
1257 ranges8774 lines · 40 files · Browse complete extent
All tests (intent)
52 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

2 files ranked by introduced lines: 802 introduced LOC across 124 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

src/vs/base/parts/ipc/common/ipc.net.ts 517 introduced LOC · 73 ranges

Open complete file

1 > /*--------------------------------------------------------------------------------------------- ipc.net.ts
2 > * Copyright (c) Microsoft Corporation. All rights reserved.
3 > * Licensed under the MIT License. See License.txt in the project root for license information.
4 > *--------------------------------------------------------------------------------------------*/
5 >
6 > import { VSBuffer } from '../../../common/buffer.js';
7 > import { Emitter, Event } from '../../../common/event.js';
8 > import { Disposable, DisposableStore, IDisposable } from '../../../common/lifecycle.js';
9 > import { IIPCLogger, IMessagePassingProtocol, IPCClient } from './ipc.js';
10 >
11 > export const enum SocketDiagnosticsEventType {
12 > Created = 'created',
13 > Read = 'read',
14 > Write = 'write',
15 > Open = 'open',
16 > Error = 'error',
17 > Close = 'close',
18 >
19 > BrowserWebSocketBlobReceived = 'browserWebSocketBlobReceived',
20 >
21 > NodeEndReceived = 'nodeEndReceived',
22 > NodeEndSent = 'nodeEndSent',
23 > NodeDrainBegin = 'nodeDrainBegin',
24 > NodeDrainEnd = 'nodeDrainEnd',
25 >
26 > zlibInflateError = 'zlibInflateError',
27 > zlibInflateData = 'zlibInflateData',
28 > zlibInflateInitialWrite = 'zlibInflateInitialWrite',
29 > zlibInflateInitialFlushFired = 'zlibInflateInitialFlushFired',
30 > zlibInflateWrite = 'zlibInflateWrite',
31 > zlibInflateFlushFired = 'zlibInflateFlushFired',
32 > zlibDeflateError = 'zlibDeflateError',
33 > zlibDeflateData = 'zlibDeflateData',
34 > zlibDeflateWrite = 'zlibDeflateWrite',
35 > zlibDeflateFlushFired = 'zlibDeflateFlushFired',
36 >
37 > WebSocketNodeSocketWrite = 'webSocketNodeSocketWrite',
38 > WebSocketNodeSocketPeekedHeader = 'webSocketNodeSocketPeekedHeader',
39 > WebSocketNodeSocketReadHeader = 'webSocketNodeSocketReadHeader',
40 > WebSocketNodeSocketReadData = 'webSocketNodeSocketReadData',
41 > WebSocketNodeSocketUnmaskedData = 'webSocketNodeSocketUnmaskedData',
42 > WebSocketNodeSocketDrainBegin = 'webSocketNodeSocketDrainBegin',
43 > WebSocketNodeSocketDrainEnd = 'webSocketNodeSocketDrainEnd',
44 >
45 > ProtocolHeaderRead = 'protocolHeaderRead',
46 > ProtocolMessageRead = 'protocolMessageRead',
47 > ProtocolHeaderWrite = 'protocolHeaderWrite',
48 > ProtocolMessageWrite = 'protocolMessageWrite',
49 > ProtocolWrite = 'protocolWrite',
50 > }
51 >
52 > export namespace SocketDiagnostics {
53 >
54 > export const enableDiagnostics = false;
55 >
56 > export interface IRecord {
57 > timestamp: number;
58 > id: string;
59 > label: string;
60 > type: SocketDiagnosticsEventType;
61 > buff?: VSBuffer;
62 > data?: any;
63 > }
64 >
65 > export const records: IRecord[] = [];
66 > const socketIds = new WeakMap<any, string>();
67 > let lastUsedSocketId = 0;
68 >
69 > function getSocketId(nativeObject: unknown, label: string): string {
70 if (!socketIds.has(nativeObject)) {
71 const id = String(++lastUsedSocketId);
74 return socketIds.get(nativeObject)!;
75 }
76 > ipc.net.ts
77 > export function traceSocketEvent(nativeObject: unknown, socketDebugLabel: string, type: SocketDiagnosticsEventType, data?: VSBuffer | Uint8Array | ArrayBuffer | ArrayBufferView | any): void {
78 if (!enableDiagnostics) {
79 return;
90 }
91 }
92 > } ipc.net.ts
93 >
94 > export const enum SocketCloseEventType {
95 > NodeSocketCloseEvent = 0,
96 > WebSocketCloseEvent = 1
97 > }
98 >
99 > export interface NodeSocketCloseEvent {
100 > /**
101 > * The type of the event
102 > */
103 > readonly type: SocketCloseEventType.NodeSocketCloseEvent;
104 > /**
105 > * `true` if the socket had a transmission error.
106 > */
107 > readonly hadError: boolean;
108 > /**
109 > * Underlying error.
110 > */
111 > readonly error: Error | undefined;
112 > }
113 >
114 > export interface WebSocketCloseEvent {
115 > /**
116 > * The type of the event
117 > */
118 > readonly type: SocketCloseEventType.WebSocketCloseEvent;
119 > /**
120 > * Returns the WebSocket connection close code provided by the server.
121 > */
122 > readonly code: number;
123 > /**
124 > * Returns the WebSocket connection close reason provided by the server.
125 > */
126 > readonly reason: string;
127 > /**
128 > * Returns true if the connection closed cleanly; false otherwise.
129 > */
130 > readonly wasClean: boolean;
131 > /**
132 > * Underlying event.
133 > */
134 > readonly event: any | undefined;
135 > }
136 >
137 > export type SocketCloseEvent = NodeSocketCloseEvent | WebSocketCloseEvent | undefined;
138 >
139 > export const enum SocketTimeoutReason {
140 > UNACKNOWLEDGED_MESSAGE = 'unacknowledgedMessage',
141 > KEEP_ALIVE = 'keepAlive',
142 > }
143 >
144 > export interface SocketTimeoutEvent {
145 > readonly reason: SocketTimeoutReason;
146 > readonly unacknowledgedMsgCount: number;
147 > readonly timeSinceOldestUnacknowledgedMsg?: number;
148 > readonly timeSinceLastReceivedSomeData: number;
149 > }
150 >
151 > export interface ISocket extends IDisposable {
152 > onData(listener: (e: VSBuffer) => void): IDisposable;
153 > onClose(listener: (e: SocketCloseEvent) => void): IDisposable;
154 > onEnd(listener: () => void): IDisposable;
155 > write(buffer: VSBuffer): void;
156 > end(): void;
157 > drain(): Promise<void>;
158 >
159 > traceSocketEvent(type: SocketDiagnosticsEventType, data?: VSBuffer | Uint8Array | ArrayBuffer | ArrayBufferView | any): void;
160 > }
161 >
162 > let emptyBuffer: VSBuffer | null = null;
163 function getEmptyBuffer(): VSBuffer {
164 if (!emptyBuffer) {
167 return emptyBuffer;
168 }
169 > ipc.net.ts
170 > export class ChunkStream {
171 >
172 > private _chunks: VSBuffer[];
173 > private _totalLength: number;
174 >
175 > public get byteLength() {
176 return this._totalLength;
177 }
178 > ipc.net.ts
179 > constructor() {
180 this._chunks = [];
181 this._totalLength = 0;
182 }
183 > ipc.net.ts
184 > public acceptChunk(buff: VSBuffer) {
185 this._chunks.push(buff);
186 this._totalLength += buff.byteLength;
187 }
188 > ipc.net.ts
189 > public read(byteCount: number): VSBuffer {
190 return this._read(byteCount, true);
191 }
192 > ipc.net.ts
193 > public peek(byteCount: number): VSBuffer {
194 return this._read(byteCount, false);
195 }
196 > ipc.net.ts
197 > private _read(byteCount: number, advance: boolean): VSBuffer {
198
199 if (byteCount === 0) {
259 return result;
260 }
261 > } ipc.net.ts
262 >
263 > const enum ProtocolMessageType {
264 > None = 0,
265 > Regular = 1,
266 > Control = 2,
267 > Ack = 3,
268 > Disconnect = 5,
269 > ReplayRequest = 6,
270 > Pause = 7,
271 > Resume = 8,
272 > KeepAlive = 9
273 > }
274 >
275 function protocolMessageTypeToString(messageType: ProtocolMessageType) {
276 switch (messageType) {
286 }
287 }
288 > ipc.net.ts
289 > export const enum ProtocolConstants {
290 > HeaderLength = 13,
291 > /**
292 > * Send an Acknowledge message at most 2 seconds later...
293 > */
294 > AcknowledgeTime = 2000, // 2 seconds
295 > /**
296 > * If there is a sent message that has been unacknowledged for 20 seconds,
297 > * and we didn't see any incoming server data in the past 20 seconds,
298 > * then consider the connection has timed out.
299 > */
300 > TimeoutTime = 20000, // 20 seconds
301 > /**
302 > * If there is no reconnection within this time-frame, consider the connection permanently closed...
303 > */
304 > ReconnectionGraceTime = 3 * 60 * 60 * 1000, // 3hrs
305 > /**
306 > * Maximal grace time between the first and the last reconnection...
307 > */
308 > ReconnectionShortGraceTime = 5 * 60 * 1000, // 5min
309 > /**
310 > * Send a message every 5 seconds to avoid that the connection is closed by the OS.
311 > */
312 > KeepAliveSendTime = 5000, // 5 seconds
313 > }
314 >
315 > class ProtocolMessage {
316 >
317 > public writtenTime: number;
318 >
319 > constructor(
320 public readonly type: ProtocolMessageType,
321 public readonly id: number,
325 this.writtenTime = 0;
326 }
327 > ipc.net.ts
328 > public get size(): number {
329 return this.data.byteLength;
330 }
331 > } ipc.net.ts
332 >
333 > class ProtocolReader extends Disposable {
334 >
335 > private readonly _socket: ISocket;
336 > private _isDisposed: boolean;
337 > private readonly _incomingData: ChunkStream;
338 > public lastReadTime: number;
339 >
340 > private readonly _onMessage = this._register(new Emitter<ProtocolMessage>());
341 > public readonly onMessage: Event<ProtocolMessage> = this._onMessage.event;
342 >
343 > private readonly _state = {
344 > readHead: true,
345 > readLen: ProtocolConstants.HeaderLength,
346 > messageType: ProtocolMessageType.None,
347 > id: 0,
348 > ack: 0
349 > };
350 >
351 > constructor(socket: ISocket) {
352 super();
353 this._socket = socket;
357 this.lastReadTime = Date.now();
358 }
359 > ipc.net.ts
360 > public acceptChunk(data: VSBuffer | null): void {
361 if (!data || data.byteLength === 0) {
362 return;
407 }
408 }
409 > ipc.net.ts
410 > public readEntireBuffer(): VSBuffer {
411 return this._incomingData.read(this._incomingData.byteLength);
412 }
413 > ipc.net.ts
414 > public override dispose(): void {
415 this._isDisposed = true;
416 super.dispose();
417 }
418 > } ipc.net.ts
419 >
420 > class ProtocolWriter {
421 >
422 > private _isDisposed: boolean;
423 > private _isPaused: boolean;
424 > private readonly _socket: ISocket;
425 > private _data: VSBuffer[];
426 > private _totalLength: number;
427 > public lastWriteTime: number;
428 >
429 > constructor(socket: ISocket) {
430 this._isDisposed = false;
431 this._isPaused = false;
435 this.lastWriteTime = 0;
436 }
437 > ipc.net.ts
438 > public dispose(): void {
439 try {
440 this.flush();
444 this._isDisposed = true;
445 }
446 > ipc.net.ts
447 > public drain(): Promise<void> {
448 this.flush();
449 return this._socket.drain();
450 }
451 > ipc.net.ts
452 > public flush(): void {
453 // flush
454 this._writeNow();
455 }
456 > ipc.net.ts
457 > public pause(): void {
458 this._isPaused = true;
459 }
460 > ipc.net.ts
461 > public resume(): void {
462 this._isPaused = false;
463 this._scheduleWriting();
464 }
465 > ipc.net.ts
466 > public write(msg: ProtocolMessage) {
467 if (this._isDisposed) {
468 // ignore: there could be left-over promises which complete and then
483 this._writeSoon(header, msg.data);
484 }
485 > ipc.net.ts
486 > private _bufferAdd(head: VSBuffer, body: VSBuffer): boolean {
487 const wasEmpty = this._totalLength === 0;
488 this._data.push(head, body);
490 return wasEmpty;
491 }
492 > ipc.net.ts
493 > private _bufferTake(): VSBuffer {
494 const ret = VSBuffer.concat(this._data, this._totalLength);
495 this._data.length = 0;
497 return ret;
498 }
499 > ipc.net.ts
500 > private _writeSoon(header: VSBuffer, data: VSBuffer): void {
501 if (this._bufferAdd(header, data)) {
502 this._scheduleWriting();
503 }
504 }
505 > ipc.net.ts
506 > private _writeNowTimeout: Timeout | null = null;
507 > private _scheduleWriting(): void {
508 if (this._writeNowTimeout) {
509 return;
514 });
515 }
516 > ipc.net.ts
517 > private _writeNow(): void {
518 if (this._totalLength === 0) {
519 return;
526 this._socket.write(data);
527 }
528 > } ipc.net.ts
529 >
530 > /**
531 > * A message has the following format:
532 > * ```
533 > * /-------------------------------|------\
534 > * | HEADER | |
535 > * |-------------------------------| DATA |
536 > * | TYPE | ID | ACK | DATA_LENGTH | |
537 > * \-------------------------------|------/
538 > * ```
539 > * The header is 9 bytes and consists of:
540 > * - TYPE is 1 byte (ProtocolMessageType) - the message type
541 > * - ID is 4 bytes (u32be) - the message id (can be 0 to indicate to be ignored)
542 > * - ACK is 4 bytes (u32be) - the acknowledged message id (can be 0 to indicate to be ignored)
543 > * - DATA_LENGTH is 4 bytes (u32be) - the length in bytes of DATA
544 > *
545 > * Only Regular messages are counted, other messages are not counted, nor acknowledged.
546 > */
547 > export class Protocol extends Disposable implements IMessagePassingProtocol {
548 >
549 > private _socket: ISocket;
550 > private _socketWriter: ProtocolWriter;
551 > private _socketReader: ProtocolReader;
552 >
553 > private readonly _onMessage = this._register(new Emitter<VSBuffer>());
554 > readonly onMessage: Event<VSBuffer> = this._onMessage.event;
555 >
556 > private readonly _onDidDispose = this._register(new Emitter<void>());
557 > readonly onDidDispose: Event<void> = this._onDidDispose.event;
558 >
559 > constructor(socket: ISocket) {
560 super();
561 this._socket = socket;
571 this._register(this._socket.onClose(() => this._onDidDispose.fire()));
572 }
573 > ipc.net.ts
574 > drain(): Promise<void> {
575 return this._socketWriter.drain();
576 }
577 > ipc.net.ts
578 > getSocket(): ISocket {
579 return this._socket;
580 }
581 > ipc.net.ts
582 > sendDisconnect(): void {
583 // Nothing to do...
584 }
585 > ipc.net.ts
586 > send(buffer: VSBuffer): void {
587 this._socketWriter.write(new ProtocolMessage(ProtocolMessageType.Regular, 0, 0, buffer));
588 }
589 > } ipc.net.ts
590 >
591 > export class Client<TContext = string> extends IPCClient<TContext> {
592 >
593 > static fromSocket<TContext = string>(socket: ISocket, id: TContext): Client<TContext> {
594 > return new Client(new Protocol(socket), id);
595 > }
596 >
597 > get onDidDispose(): Event<void> { return this.protocol.onDidDispose; }
598 >
599 > constructor(private protocol: Protocol | PersistentProtocol, id: TContext, ipcLogger: IIPCLogger | null = null) {
600 super(protocol, id, ipcLogger);
601 }
602 > ipc.net.ts
603 > override dispose(): void {
604 super.dispose();
605 const socket = this.protocol.getSocket();
610 socket.end();
611 }
612 > } ipc.net.ts
613 >
614 > /**
615 > * Will ensure no messages are lost if there are no event listeners.
616 > */
617 > export class BufferedEmitter<T> {
618 > private _emitter: Emitter<T>;
619 > public readonly event: Event<T>;
620 >
621 > private _hasListeners = false;
622 > private _isDeliveringMessages = false;
623 > private _bufferedMessages: T[] = [];
624 >
625 > constructor() {
626 this._emitter = new Emitter<T>({
627 onWillAddFirstListener: () => {
639 this.event = this._emitter.event;
640 }
641 > ipc.net.ts
642 > private _deliverMessages(): void {
643 if (this._isDeliveringMessages) {
644 return;
650 this._isDeliveringMessages = false;
651 }
652 > ipc.net.ts
653 > public fire(event: T): void {
654 if (this._hasListeners) {
655 if (this._bufferedMessages.length > 0) {
662 }
663 }
664 > ipc.net.ts
665 > public flushBuffer(): void {
666 this._bufferedMessages = [];
667 }
668 > } ipc.net.ts
669 >
670 > class QueueElement<T> {
671 > public readonly data: T;
672 > public next: QueueElement<T> | null;
673 >
674 > constructor(data: T) {
675 this.data = data;
676 this.next = null;
677 }
678 > } ipc.net.ts
679 >
680 > class Queue<T> {
681 >
682 > private _first: QueueElement<T> | null;
683 > private _last: QueueElement<T> | null;
684 >
685 > constructor() {
686 this._first = null;
687 this._last = null;
688 }
689 > ipc.net.ts
690 > public length(): number {
691 let result = 0;
692 let current = this._first;
697 return result;
698 }
699 > ipc.net.ts
700 > public peek(): T | null {
701 if (!this._first) {
702 return null;
704 return this._first.data;
705 }
706 > ipc.net.ts
707 > public toArray(): T[] {
708 const result: T[] = [];
709 let resultLen = 0;
715 return result;
716 }
717 > ipc.net.ts
718 > public pop(): void {
719 if (!this._first) {
720 return;
727 this._first = this._first.next;
728 }
729 > ipc.net.ts
730 > public push(item: T): void {
731 const element = new QueueElement(item);
732 if (!this._first) {
738 this._last = element;
739 }
740 > } ipc.net.ts
741 >
742 > export class LoadEstimator {
743 >
744 > private static _HISTORY_LENGTH = 10;
745 > private static _INSTANCE: LoadEstimator | null = null;
746 > public static getInstance(): LoadEstimator {
747 if (!LoadEstimator._INSTANCE) {
748 LoadEstimator._INSTANCE = new LoadEstimator();
750 return LoadEstimator._INSTANCE;
751 }
752 > ipc.net.ts
753 > private lastRuns: number[];
754 >
755 > constructor() {
756 this.lastRuns = [];
757 const now = Date.now();
766 }, 1000);
767 }
768 > ipc.net.ts
769 > /**
770 > * returns an estimative number, from 0 (low load) to 1 (high load)
771 > */
772 > private load(): number {
773 const now = Date.now();
774 const historyLimit = (1 + LoadEstimator._HISTORY_LENGTH) * 1000;
781 return 1 - score / LoadEstimator._HISTORY_LENGTH;
782 }
783 > ipc.net.ts
784 > public hasHighLoad(): boolean {
785 return this.load() >= 0.5;
786 }
787 > } ipc.net.ts
788 >
789 > export interface ILoadEstimator {
790 > hasHighLoad(): boolean;
791 > }
792 >
793 > export interface PersistentProtocolOptions {
794 > /**
795 > * The socket to use.
796 > */
797 > socket: ISocket;
798 > /**
799 > * The initial chunk of data that has already been received from the socket.
800 > */
801 > initialChunk?: VSBuffer | null;
802 > /**
803 > * The CPU load estimator to use.
804 > */
805 > loadEstimator?: ILoadEstimator;
806 > /**
807 > * Whether to send keep alive messages. Defaults to true.
808 > */
809 > sendKeepAlive?: boolean;
810 > }
811 >
812 > /**
813 > * Same as Protocol, but will actually track messages and acks.
814 > * Moreover, it will ensure no messages are lost if there are no event listeners.
815 > */
816 > export class PersistentProtocol implements IMessagePassingProtocol {
817 >
818 > private _isReconnecting: boolean;
819 > private _didSendDisconnect?: boolean;
820 >
821 > private _outgoingUnackMsg: Queue<ProtocolMessage>;
822 > private _outgoingMsgId: number;
823 > private _outgoingAckId: number;
824 > private _outgoingAckTimeout: Timeout | null;
825 >
826 > private _incomingMsgId: number;
827 > private _incomingAckId: number;
828 > private _incomingMsgLastTime: number;
829 > private _incomingAckTimeout: Timeout | null;
830 >
831 > private _keepAliveInterval: Timeout | null;
832 >
833 > private _lastReplayRequestTime: number;
834 > private _lastSocketTimeoutTime: number;
835 >
836 > private _socket: ISocket;
837 > private _socketWriter: ProtocolWriter;
838 > private _socketReader: ProtocolReader;
839 > // eslint-disable-next-line local/code-no-potentially-unsafe-disposables
840 > private _socketDisposables: DisposableStore;
841 >
842 > private readonly _loadEstimator: ILoadEstimator;
843 > private readonly _shouldSendKeepAlive: boolean;
844 >
845 > private readonly _onControlMessage = new BufferedEmitter<VSBuffer>();
846 > readonly onControlMessage: Event<VSBuffer> = this._onControlMessage.event;
847 >
848 > private readonly _onMessage = new BufferedEmitter<VSBuffer>();
849 > readonly onMessage: Event<VSBuffer> = this._onMessage.event;
850 >
851 > private readonly _onDidDispose = new BufferedEmitter<void>();
852 > readonly onDidDispose: Event<void> = this._onDidDispose.event;
853 >
854 > private readonly _onSocketClose = new BufferedEmitter<SocketCloseEvent>();
855 > readonly onSocketClose: Event<SocketCloseEvent> = this._onSocketClose.event;
856 >
857 > private readonly _onSocketTimeout = new BufferedEmitter<SocketTimeoutEvent>();
858 > readonly onSocketTimeout: Event<SocketTimeoutEvent> = this._onSocketTimeout.event;
859 >
860 > public get unacknowledgedCount(): number {
861 > return this._outgoingMsgId - this._outgoingAckId;
862 > }
863 >
864 > constructor(opts: PersistentProtocolOptions) {
865 this._loadEstimator = opts.loadEstimator ?? LoadEstimator.getInstance();
866 this._shouldSendKeepAlive = opts.sendKeepAlive ?? true;
898 }
899 }
900 > ipc.net.ts
901 > dispose(): void {
902 if (this._outgoingAckTimeout) {
903 clearTimeout(this._outgoingAckTimeout);
914 this._socketDisposables.dispose();
915 }
916 > ipc.net.ts
917 > drain(): Promise<void> {
918 return this._socketWriter.drain();
919 }
920 > ipc.net.ts
921 > sendDisconnect(): void {
922 if (!this._didSendDisconnect) {
923 this._didSendDisconnect = true;
927 }
928 }
929 > ipc.net.ts
930 > sendPause(): void {
931 const msg = new ProtocolMessage(ProtocolMessageType.Pause, 0, 0, getEmptyBuffer());
932 this._socketWriter.write(msg);
933 }
934 > ipc.net.ts
935 > sendResume(): void {
936 const msg = new ProtocolMessage(ProtocolMessageType.Resume, 0, 0, getEmptyBuffer());
937 this._socketWriter.write(msg);
938 }
939 > ipc.net.ts
940 > pauseSocketWriting() {
941 this._socketWriter.pause();
942 }
943 > ipc.net.ts
944 > public getSocket(): ISocket {
945 return this._socket;
946 }
947 > ipc.net.ts
948 > public getMillisSinceLastIncomingData(): number {
949 return Date.now() - this._socketReader.lastReadTime;
950 }
951 > ipc.net.ts
952 > public beginAcceptReconnection(socket: ISocket, initialDataChunk: VSBuffer | null): void {
953 this._isReconnecting = true;
954
971 this._socketReader.acceptChunk(initialDataChunk);
972 }
973 > ipc.net.ts
974 > public endAcceptReconnection(): void {
975 this._isReconnecting = false;
976
988 this._recvAckCheck();
989 }
990 > ipc.net.ts
991 > public acceptDisconnect(): void {
992 this._onDidDispose.fire();
993 }
994 > ipc.net.ts
995 > private _receiveMessage(msg: ProtocolMessage): void {
996 if (msg.ack > this._outgoingAckId) {
997 this._outgoingAckId = msg.ack;
1066 }
1067 }
1068 > ipc.net.ts
1069 > readEntireBuffer(): VSBuffer {
1070 return this._socketReader.readEntireBuffer();
1071 }
1072 > ipc.net.ts
1073 > flush(): void {
1074 this._socketWriter.flush();
1075 }
1076 > ipc.net.ts
1077 > send(buffer: VSBuffer): void {
1078 const myId = ++this._outgoingMsgId;
1079 this._incomingAckId = this._incomingMsgId;
1085 }
1086 }
1087 > ipc.net.ts
1088 > /**
1089 > * Send a message which will not be part of the regular acknowledge flow.
1090 > * Use this for early control messages which are repeated in case of reconnection.
1091 > */
1092 > sendControl(buffer: VSBuffer): void {
1093 const msg = new ProtocolMessage(ProtocolMessageType.Control, 0, 0, buffer);
1094 this._socketWriter.write(msg);
1095 }
1096 > ipc.net.ts
1097 > private _sendAckCheck(): void {
1098 if (this._incomingMsgId <= this._incomingAckId) {
1099 // nothink to acknowledge
1120 }, ProtocolConstants.AcknowledgeTime - timeSinceLastIncomingMsg + 5);
1121 }
1122 > ipc.net.ts
1123 > private _recvAckCheck(): void {
1124 if (this._outgoingMsgId <= this._outgoingAckId) {
1125 // everything has been acknowledged
1177 }, minimumTimeUntilTimeout);
1178 }
1179 > ipc.net.ts
1180 > /**
1181 > * Called after sending a keepalive. Both sides of this protocol send
1182 > * keepalives every KeepAliveSendTime (5s), so receiving no data for
1183 > * TimeoutTime (20s) means the connection is dead. This catches silent
1184 > * connection deaths that _recvAckCheck cannot detect because there are
1185 > * no unacknowledged regular messages.
1186 > */
1187 > private _keepAliveTimeoutCheck(): void {
1188 if (this._isReconnecting) {
1189 return;
1212 }
1213 }
1214 > ipc.net.ts
1215 > private _sendAck(): void {
1216 if (this._incomingMsgId <= this._incomingAckId) {
1217 // nothink to acknowledge
1223 this._socketWriter.write(msg);
1224 }
1225 > ipc.net.ts
1226 > private _sendKeepAlive(): void {
1227 this._incomingAckId = this._incomingMsgId;
1228 const msg = new ProtocolMessage(ProtocolMessageType.KeepAlive, 0, this._incomingAckId, getEmptyBuffer());
1230 this._keepAliveTimeoutCheck();
1231 }
1232 > } ipc.net.ts
src/vs/base/parts/ipc/node/ipc.net.ts 285 introduced LOC · 51 ranges

Open complete file

1 > /*--------------------------------------------------------------------------------------------- ipc.net.ts
2 > * Copyright (c) Microsoft Corporation. All rights reserved.
3 > * Licensed under the MIT License. See License.txt in the project root for license information.
4 > *--------------------------------------------------------------------------------------------*/
5 >
6 > import { createHash } from 'crypto';
7 > import type * as http from 'http';
8 > import { Server as NetServer, Socket, createConnection, createServer } from 'net';
9 > import { tmpdir } from 'os';
10 > import { DeflateRaw, InflateRaw, ZlibOptions, createDeflateRaw, createInflateRaw } from 'zlib';
11 > import { VSBuffer } from '../../../common/buffer.js';
12 > import { onUnexpectedError } from '../../../common/errors.js';
13 > import { Emitter, Event } from '../../../common/event.js';
14 > import { Disposable, IDisposable } from '../../../common/lifecycle.js';
15 > import { join } from '../../../common/path.js';
16 > import { Platform, platform } from '../../../common/platform.js';
17 > import { generateUuid } from '../../../common/uuid.js';
18 > import { ClientConnectionEvent, IPCServer } from '../common/ipc.js';
19 > import { ChunkStream, Client, ISocket, Protocol, SocketCloseEvent, SocketCloseEventType, SocketDiagnostics, SocketDiagnosticsEventType } from '../common/ipc.net.js';
20 >
21 > export function upgradeToISocket(req: http.IncomingMessage, socket: Socket, {
22 debugLabel,
23 skipWebSocketFrames = false,
84 }
85 }
86 > ipc.net.ts
87 > /**
88 > * Maximum time to wait for a 'close' event to fire after the socket stream
89 > * ends. For unix domain sockets, the close event may not fire consistently
90 > * due to what appears to be a Node.js bug.
91 > *
92 > * @see https://github.com/microsoft/vscode/issues/211462#issuecomment-2155471996
93 > */
94 > const socketEndTimeoutMs = 30_000;
95 >
96 > export class NodeSocket implements ISocket {
97 >
98 > public readonly debugLabel: string;
99 > public readonly socket: Socket;
100 > private readonly _errorListener: (err: NodeJS.ErrnoException) => void;
101 > private readonly _closeListener: (hadError: boolean) => void;
102 > private readonly _endListener: () => void;
103 > private _endTimeoutHandle: Timeout | undefined;
104 > private _canWrite = true;
105 >
106 > public traceSocketEvent(type: SocketDiagnosticsEventType, data?: VSBuffer | Uint8Array | ArrayBuffer | ArrayBufferView | unknown): void {
107 > SocketDiagnostics.traceSocketEvent(this.socket, this.debugLabel, type, data);
108 > }
109 >
110 > constructor(socket: Socket, debugLabel = '') {
111 this.debugLabel = debugLabel;
112 this.socket = socket;
145 this.socket.on('end', this._endListener);
146 }
147 > ipc.net.ts
148 > public dispose(destroySocket = true): void {
149 if (this._endTimeoutHandle) {
150 clearTimeout(this._endTimeoutHandle);
158 }
159 }
160 > ipc.net.ts
161 > public onData(_listener: (e: VSBuffer) => void): IDisposable {
162 const listener = (buff: Buffer) => {
163 this.traceSocketEvent(SocketDiagnosticsEventType.Read, buff);
169 };
170 }
171 > ipc.net.ts
172 > public onClose(listener: (e: SocketCloseEvent) => void): IDisposable {
173 const adapter = (hadError: boolean) => {
174 listener({
183 };
184 }
185 > ipc.net.ts
186 > public onEnd(listener: () => void): IDisposable {
187 const adapter = () => {
188 listener();
193 };
194 }
195 > ipc.net.ts
196 > public write(buffer: VSBuffer): void {
197 // return early if socket has been destroyed in the meantime
198 if (this.socket.destroyed || !this._canWrite) {
234 }
235 }
236 > ipc.net.ts
237 > public end(): void {
238 this.traceSocketEvent(SocketDiagnosticsEventType.NodeEndSent);
239 this.socket.end();
240 }
241 > ipc.net.ts
242 > public drain(): Promise<void> {
243 this.traceSocketEvent(SocketDiagnosticsEventType.NodeDrainBegin);
244 return new Promise<void>((resolve, reject) => {
264 });
265 }
266 > } ipc.net.ts
267 >
268 > const enum Constants {
269 > MinHeaderByteSize = 2,
270 > /**
271 > * If we need to write a large buffer, we will split it into 256KB chunks and
272 > * send each chunk as a websocket message. This is to prevent that the sending
273 > * side is stuck waiting for the entire buffer to be compressed before writing
274 > * to the underlying socket or that the receiving side is stuck waiting for the
275 > * entire message to be received before processing the bytes.
276 > */
277 > MaxWebSocketMessageLength = 256 * 1024 // 256 KB
278 > }
279 >
280 > const enum ReadState {
281 > PeekHeader = 1,
282 > ReadHeader = 2,
283 > ReadBody = 3,
284 > Fin = 4
285 > }
286 >
287 > interface ISocketTracer {
288 > traceSocketEvent(type: SocketDiagnosticsEventType, data?: VSBuffer | Uint8Array | ArrayBuffer | ArrayBufferView | unknown): void;
289 > }
290 >
291 > interface FrameOptions {
292 > compressed: boolean;
293 > opcode: number;
294 > }
295 >
296 > /**
297 > * See https://tools.ietf.org/html/rfc6455#section-5.2
298 > */
299 > export class WebSocketNodeSocket extends Disposable implements ISocket, ISocketTracer {
300 >
301 > public readonly socket: NodeSocket;
302 > private readonly _flowManager: WebSocketFlowManager;
303 > private readonly _incomingData: ChunkStream;
304 > private readonly _onData = this._register(new Emitter<VSBuffer>());
305 > private readonly _onClose = this._register(new Emitter<SocketCloseEvent>());
306 > private readonly _maxSocketMessageLength: number;
307 > private _isEnded = false;
308 >
309 > private readonly _state = {
310 > state: ReadState.PeekHeader,
311 > readLen: Constants.MinHeaderByteSize,
312 > fin: 0,
313 > compressed: false,
314 > firstFrameOfMessage: true,
315 > mask: 0,
316 > opcode: 0
317 > };
318 >
319 > public get permessageDeflate(): boolean {
320 > return this._flowManager.permessageDeflate;
321 > }
322 >
323 > public get recordedInflateBytes(): VSBuffer {
324 return this._flowManager.recordedInflateBytes;
325 }
326 > ipc.net.ts
327 > public setRecordInflateBytes(record: boolean): void {
328 this._flowManager.setRecordInflateBytes(record);
329 }
330 > ipc.net.ts
331 > public traceSocketEvent(type: SocketDiagnosticsEventType, data?: VSBuffer | Uint8Array | ArrayBuffer | ArrayBufferView | unknown): void {
332 this.socket.traceSocketEvent(type, data);
333 }
334 > ipc.net.ts
335 > /**
336 > * Create a socket which can communicate using WebSocket frames.
337 > *
338 > * **NOTE**: When using the permessage-deflate WebSocket extension, if parts of inflating was done
339 > * in a different zlib instance, we need to pass all those bytes into zlib, otherwise the inflate
340 > * might hit an inflated portion referencing a distance too far back.
341 > *
342 > * @param socket The underlying socket
343 > * @param permessageDeflate Use the permessage-deflate WebSocket extension
344 > * @param inflateBytes "Seed" zlib inflate with these bytes.
345 > * @param recordInflateBytes Record all bytes sent to inflate
346 > */
347 > constructor(socket: NodeSocket, permessageDeflate: boolean, inflateBytes: VSBuffer | null, recordInflateBytes: boolean, enableMessageSplitting = true) {
348 super();
349 this.socket = socket;
379 }));
380 }
381 > ipc.net.ts
382 > public override dispose(): void {
383 if (this._flowManager.isProcessingWriteQueue()) {
384 // Wait for any outstanding writes to finish before disposing
391 }
392 }
393 > ipc.net.ts
394 > public onData(listener: (e: VSBuffer) => void): IDisposable {
395 return this._onData.event(listener);
396 }
397 > ipc.net.ts
398 > public onClose(listener: (e: SocketCloseEvent) => void): IDisposable {
399 return this._onClose.event(listener);
400 }
401 > ipc.net.ts
402 > public onEnd(listener: () => void): IDisposable {
403 return this.socket.onEnd(listener);
404 }
405 > ipc.net.ts
406 > public write(buffer: VSBuffer): void {
407 // If we write many logical messages (let's say 1000 messages of 100KB) during a single process tick, we do
408 // this thing where we install a process.nextTick timer and group all of them together and we then issue a
423 }
424 }
425 > ipc.net.ts
426 > private _write(buffer: VSBuffer, { compressed, opcode }: FrameOptions): void {
427 if (this._isEnded) {
428 // Avoid ERR_STREAM_WRITE_AFTER_END
467 this.socket.write(VSBuffer.concat([header, buffer]));
468 }
469 > ipc.net.ts
470 > public end(): void {
471 this._isEnded = true;
472 this.socket.end();
473 }
474 > ipc.net.ts
475 > private _acceptChunk(data: VSBuffer): void {
476 if (data.byteLength === 0) {
477 return;
571 }
572 }
573 > ipc.net.ts
574 > public async drain(): Promise<void> {
575 this.traceSocketEvent(SocketDiagnosticsEventType.WebSocketNodeSocketDrainBegin);
576 if (this._flowManager.isProcessingWriteQueue()) {
580 this.traceSocketEvent(SocketDiagnosticsEventType.WebSocketNodeSocketDrainEnd);
581 }
582 > } ipc.net.ts
583 >
584 > class WebSocketFlowManager extends Disposable {
585 >
586 > private readonly _onError = this._register(new Emitter<Error>());
587 > public readonly onError = this._onError.event;
588 >
589 > private readonly _zlibInflateStream: ZlibInflateStream | null;
590 > private readonly _zlibDeflateStream: ZlibDeflateStream | null;
591 > private readonly _writeQueue: { data: VSBuffer; options: FrameOptions }[] = [];
592 > private readonly _readQueue: { data: VSBuffer; isCompressed: boolean; isLastFrameOfMessage: boolean }[] = [];
593 >
594 > private readonly _onDidFinishProcessingReadQueue = this._register(new Emitter<void>());
595 > public readonly onDidFinishProcessingReadQueue = this._onDidFinishProcessingReadQueue.event;
596 >
597 > private readonly _onDidFinishProcessingWriteQueue = this._register(new Emitter<void>());
598 > public readonly onDidFinishProcessingWriteQueue = this._onDidFinishProcessingWriteQueue.event;
599 >
600 > public get permessageDeflate(): boolean {
601 > return Boolean(this._zlibInflateStream && this._zlibDeflateStream);
602 > }
603 >
604 > public get recordedInflateBytes(): VSBuffer {
605 if (this._zlibInflateStream) {
606 return this._zlibInflateStream.recordedInflateBytes;
608 return VSBuffer.alloc(0);
609 }
610 > ipc.net.ts
611 > public setRecordInflateBytes(record: boolean): void {
612 this._zlibInflateStream?.setRecordInflateBytes(record);
613 }
614 > ipc.net.ts
615 > constructor(
616 private readonly _tracer: ISocketTracer,
617 permessageDeflate: boolean,
635 }
636 }
637 > ipc.net.ts
638 > public writeMessage(data: VSBuffer, options: FrameOptions): void {
639 this._writeQueue.push({ data, options });
640 this._processWriteQueue();
641 }
642 > ipc.net.ts
643 > private _isProcessingWriteQueue = false;
644 > private async _processWriteQueue(): Promise<void> {
645 if (this._isProcessingWriteQueue) {
646 return;
659 this._onDidFinishProcessingWriteQueue.fire();
660 }
661 > ipc.net.ts
662 > public isProcessingWriteQueue(): boolean {
663 return (this._isProcessingWriteQueue);
664 }
665 > ipc.net.ts
666 > /**
667 > * Subsequent calls should wait for the previous `_deflateBuffer` call to complete.
668 > */
669 > private _deflateMessage(zlibDeflateStream: ZlibDeflateStream, buffer: VSBuffer): Promise<VSBuffer> {
670 return new Promise<VSBuffer>((resolve, reject) => {
671 zlibDeflateStream.write(buffer);
673 });
674 }
675 > ipc.net.ts
676 > public acceptFrame(data: VSBuffer, isCompressed: boolean, isLastFrameOfMessage: boolean): void {
677 this._readQueue.push({ data, isCompressed, isLastFrameOfMessage });
678 this._processReadQueue();
679 }
680 > ipc.net.ts
681 > private _isProcessingReadQueue = false;
682 > private async _processReadQueue(): Promise<void> {
683 if (this._isProcessingReadQueue) {
684 return;
701 this._onDidFinishProcessingReadQueue.fire();
702 }
703 > ipc.net.ts
704 > public isProcessingReadQueue(): boolean {
705 return (this._isProcessingReadQueue);
706 }
707 > ipc.net.ts
708 > /**
709 > * Subsequent calls should wait for the previous `transformRead` call to complete.
710 > */
711 > private _inflateFrame(zlibInflateStream: ZlibInflateStream, buffer: VSBuffer, isLastFrameOfMessage: boolean): Promise<VSBuffer> {
712 return new Promise<VSBuffer>((resolve, reject) => {
713 // See https://tools.ietf.org/html/rfc7692#section-7.2.2
719 });
720 }
721 > } ipc.net.ts
722 >
723 > class ZlibInflateStream extends Disposable {
724 >
725 > private readonly _onError = this._register(new Emitter<Error>());
726 > public readonly onError = this._onError.event;
727 >
728 > private readonly _zlibInflate: InflateRaw;
729 > private readonly _recordedInflateBytes: VSBuffer[] = [];
730 > private readonly _pendingInflateData: VSBuffer[] = [];
731 > private _recordInflateBytes: boolean;
732 >
733 > public get recordedInflateBytes(): VSBuffer {
734 > if (this._recordInflateBytes) {
735 > return VSBuffer.concat(this._recordedInflateBytes);
736 > }
737 > return VSBuffer.alloc(0);
738 > }
739 >
740 > constructor(
741 private readonly _tracer: ISocketTracer,
742 recordInflateBytes: boolean,
764 }
765 }
766 > ipc.net.ts
767 > public write(buffer: VSBuffer): void {
768 if (this._recordInflateBytes) {
769 this._recordedInflateBytes.push(buffer.clone());
772 this._zlibInflate.write(buffer.buffer);
773 }
774 > ipc.net.ts
775 > public setRecordInflateBytes(record: boolean): void {
776 this._recordInflateBytes = record;
777 if (!record) {
779 }
780 }
781 > ipc.net.ts
782 > public flush(callback: (data: VSBuffer) => void): void {
783 this._zlibInflate.flush(() => {
784 this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibInflateFlushFired);
788 });
789 }
790 > ipc.net.ts
791 > public override dispose(): void {
792 this._recordedInflateBytes.length = 0;
793 this._pendingInflateData.length = 0;
799 super.dispose();
800 }
801 > } ipc.net.ts
802 >
803 > class ZlibDeflateStream extends Disposable {
804 >
805 > private readonly _onError = this._register(new Emitter<Error>());
806 > public readonly onError = this._onError.event;
807 >
808 > private readonly _zlibDeflate: DeflateRaw;
809 > private readonly _pendingDeflateData: VSBuffer[] = [];
810 >
811 > constructor(
812 private readonly _tracer: ISocketTracer,
813 options: ZlibOptions
827 });
828 }
829 > ipc.net.ts
830 > public write(buffer: VSBuffer): void {
831 this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibDeflateWrite, buffer.buffer);
832 this._zlibDeflate.write(<Buffer>buffer.buffer);
833 }
834 > ipc.net.ts
835 > public flush(callback: (data: VSBuffer) => void): void {
836 // See https://zlib.net/manual.html#Constants
837 this._zlibDeflate.flush(/*Z_SYNC_FLUSH*/2, () => {
847 });
848 }
849 > ipc.net.ts
850 > public override dispose(): void {
851 this._pendingDeflateData.length = 0;
852 try {
857 super.dispose();
858 }
859 > } ipc.net.ts
860 >
861 function unmask(buffer: VSBuffer, mask: number): void {
862 if (mask === 0) {
883 }
884 }
885 > ipc.net.ts
886 > // Read this before there's any chance it is overwritten
887 > // Related to https://github.com/microsoft/vscode/issues/30624
888 > export const XDG_RUNTIME_DIR = process.env['XDG_RUNTIME_DIR'];
889 >
890 > const safeIpcPathLengths: { [platform: number]: number } = {
891 > [Platform.Linux]: 107,
892 > [Platform.Mac]: 103
893 > };
894 >
895 > export function createRandomIPCHandle(): string {
896 const randomSuffix = generateUuid();
897
921 return join(basePath, `vscode-ipc-${suffix}.sock`);
922 }
923 > ipc.net.ts
924 > export function createStaticIPCHandle(directoryPath: string, type: string, version: string): string {
925 const scope = createHash('sha256').update(directoryPath).digest('hex');
926 const scopeForSocket = scope.substr(0, 8);
954 return result;
955 }
956 > ipc.net.ts
957 function validateIPCHandleLength(handle: string): void {
958 const limit = safeIpcPathLengths[platform];
962 }
963 }
964 > ipc.net.ts
965 > export class Server extends IPCServer {
966 >
967 > private static toClientConnectionEvent(server: NetServer): Event<ClientConnectionEvent> {
968 const onConnection = Event.fromNodeEventEmitter<Socket>(server, 'connection');
969
973 }));
974 }
975 > ipc.net.ts
976 > private server: NetServer | null;
977 >
978 > constructor(server: NetServer) {
979 super(Server.toClientConnectionEvent(server));
980 this.server = server;
981 }
982 > ipc.net.ts
983 > override dispose(): void {
984 super.dispose();
985 if (this.server) {
988 }
989 }
990 > } ipc.net.ts
991 >
992 > export function serve(port: number): Promise<Server>;
993 > export function serve(namedPipe: string): Promise<Server>;
994 > export function serve(hook: number | string): Promise<Server> {
995 return new Promise<Server>((resolve, reject) => {
996 const server = createServer();
1003 });
1004 }
1005 > ipc.net.ts
1006 > export function connect(options: { host: string; port: number }, clientId: string): Promise<Client>;
1007 > export function connect(namedPipe: string, clientId: string): Promise<Client>;
1008 > export function connect(hook: { host: string; port: number } | string, clientId: string): Promise<Client> {
1009 return new Promise<Client>((resolve, reject) => {
1010 let socket: Socket;

Direct hierarchy

Parents

Children