src/vs/base/parts/ipc/common/ipc.net.ts
1232 LOC · 1092 covered · 140 uncovered · 218 ranges · 90 concepts · 32 introducers · 52 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
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 related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
>
/*---------------------------------------------------------------------------------------------
ipc.net.ts ×73
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);
72
socketIds.set(nativeObject, id);
73
}
74
return socketIds.get(nativeObject)!;
75
}
77
>
export function traceSocketEvent(nativeObject: unknown, socketDebugLabel: string, type: SocketDiagnosticsEventType, data?: VSBuffer | Uint8Array | ArrayBuffer | ArrayBufferView | any): void {
79
>
return;
80
>
}
81
const id = getSocketId(nativeObject, socketDebugLabel);
82
83
>
if (data instanceof VSBuffer || data instanceof Uint8Array || data instanceof ArrayBuffer || ArrayBuffer.isView(data)) {
ipc.net.ts ×4
84
const copiedData = VSBuffer.alloc(data.byteLength);
85
copiedData.set(data);
86
records.push({ timestamp: Date.now(), id, label: socketDebugLabel, type, buff: copiedData });
87
} else {
88
// data is a custom object
89
records.push({ timestamp: Date.now(), id, label: socketDebugLabel, type, data: data });
90
}
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;
164
>
if (!emptyBuffer) {
165
>
emptyBuffer = VSBuffer.alloc(0);
166
>
}
167
>
return emptyBuffer;
168
>
}
170
>
export class ChunkStream {
171
>
172
>
private _chunks: VSBuffer[];
173
>
private _totalLength: number;
174
>
175
>
public get byteLength() {
177
>
}
179
>
constructor() {
181
>
this._totalLength = 0;
182
>
}
184
>
public acceptChunk(buff: VSBuffer) {
186
>
this._totalLength += buff.byteLength;
187
>
}
189
>
public read(byteCount: number): VSBuffer {
191
>
}
193
>
public peek(byteCount: number): VSBuffer {
195
>
}
197
>
private _read(byteCount: number, advance: boolean): VSBuffer {
199
>
if (byteCount === 0) {
201
>
}
203
>
if (byteCount > this._totalLength) {
204
throw new Error(`Cannot read so many bytes!`);
205
}
207
>
if (this._chunks[0].byteLength === byteCount) {
209
>
const result = this._chunks[0];
210
>
if (advance) {
211
>
this._chunks.shift();
212
>
this._totalLength -= byteCount;
213
>
}
214
>
return result;
215
>
}
217
>
if (this._chunks[0].byteLength > byteCount) {
218
>
// fast path, the reading is entirely within the first chunk
219
>
const result = this._chunks[0].slice(0, byteCount);
220
>
if (advance) {
221
>
this._chunks[0] = this._chunks[0].slice(byteCount);
222
>
this._totalLength -= byteCount;
223
>
}
224
>
return result;
225
>
}
227
>
const result = VSBuffer.alloc(byteCount);
228
>
let resultOffset = 0;
229
>
let chunkIndex = 0;
230
>
while (byteCount > 0) {
231
>
const chunk = this._chunks[chunkIndex];
232
>
if (chunk.byteLength > byteCount) {
233
// this chunk will survive
234
const chunkPart = chunk.slice(0, byteCount);
235
result.set(chunkPart, resultOffset);
236
resultOffset += byteCount;
237
238
if (advance) {
239
this._chunks[chunkIndex] = chunk.slice(byteCount);
240
this._totalLength -= byteCount;
241
}
242
243
byteCount -= byteCount;
245
>
// this chunk will be entirely read
246
>
result.set(chunk, resultOffset);
247
>
resultOffset += chunk.byteLength;
248
>
249
>
if (advance) {
250
>
this._chunks.shift();
251
>
this._totalLength -= chunk.byteLength;
252
>
} else {
253
chunkIndex++;
254
}
256
>
byteCount -= chunk.byteLength;
257
>
}
258
>
}
259
>
return result;
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
>
276
>
switch (messageType) {
277
>
case ProtocolMessageType.None: return 'None';
278
>
case ProtocolMessageType.Regular: return 'Regular';
279
>
case ProtocolMessageType.Control: return 'Control';
280
>
case ProtocolMessageType.Ack: return 'Ack';
281
>
case ProtocolMessageType.Disconnect: return 'Disconnect';
282
>
case ProtocolMessageType.ReplayRequest: return 'ReplayRequest';
283
>
case ProtocolMessageType.Pause: return 'PauseWriting';
284
>
case ProtocolMessageType.Resume: return 'ResumeWriting';
285
>
case ProtocolMessageType.KeepAlive: return 'KeepAlive';
286
>
}
287
>
}
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(
321
>
public readonly id: number,
322
>
public readonly ack: number,
323
>
public readonly data: VSBuffer
324
>
) {
325
>
this.writtenTime = 0;
326
>
}
328
>
public get size(): number {
329
return this.data.byteLength;
330
}
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) {
353
>
this._socket = socket;
354
>
this._isDisposed = false;
355
>
this._incomingData = new ChunkStream();
356
>
this._register(this._socket.onData(data => this.acceptChunk(data)));
357
>
this.lastReadTime = Date.now();
358
>
}
360
>
public acceptChunk(data: VSBuffer | null): void {
363
>
}
365
>
this.lastReadTime = Date.now();
366
>
367
>
this._incomingData.acceptChunk(data);
368
>
369
>
while (this._incomingData.byteLength >= this._state.readLen) {
370
>
371
>
const buff = this._incomingData.read(this._state.readLen);
372
>
373
>
if (this._state.readHead) {
374
>
// buff is the header
375
>
376
>
// save new state => next time will read the body
377
>
this._state.readHead = false;
378
>
this._state.readLen = buff.readUInt32BE(9);
379
>
this._state.messageType = buff.readUInt8(0);
380
>
this._state.id = buff.readUInt32BE(1);
381
>
this._state.ack = buff.readUInt32BE(5);
382
>
383
>
this._socket.traceSocketEvent(SocketDiagnosticsEventType.ProtocolHeaderRead, { messageType: protocolMessageTypeToString(this._state.messageType), id: this._state.id, ack: this._state.ack, messageSize: this._state.readLen });
384
>
385
>
} else {
386
>
// buff is the body
387
>
const messageType = this._state.messageType;
388
>
const id = this._state.id;
389
>
const ack = this._state.ack;
390
>
391
>
// save new state => next time will read the header
392
>
this._state.readHead = true;
393
>
this._state.readLen = ProtocolConstants.HeaderLength;
394
>
this._state.messageType = ProtocolMessageType.None;
395
>
this._state.id = 0;
396
>
this._state.ack = 0;
397
>
398
>
this._socket.traceSocketEvent(SocketDiagnosticsEventType.ProtocolMessageRead, buff);
399
>
400
>
this._onMessage.fire(new ProtocolMessage(messageType, id, ack, buff));
401
>
402
>
if (this._isDisposed) {
403
// check if an event listener lead to our disposal
404
break;
405
}
407
>
}
408
>
}
410
>
public readEntireBuffer(): VSBuffer {
411
return this._incomingData.read(this._incomingData.byteLength);
412
}
414
>
public override dispose(): void {
416
>
super.dispose();
417
>
}
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) {
431
>
this._isPaused = false;
432
>
this._socket = socket;
433
>
this._data = [];
434
>
this._totalLength = 0;
435
>
this.lastWriteTime = 0;
436
>
}
438
>
public dispose(): void {
440
>
this.flush();
441
>
} catch (err) {
442
// ignore error, since the socket could be already closed
443
}
445
>
}
447
>
public drain(): Promise<void> {
448
this.flush();
449
return this._socket.drain();
450
}
452
>
public flush(): void {
454
>
this._writeNow();
455
>
}
457
>
public pause(): void {
459
>
}
461
>
public resume(): void {
463
>
this._scheduleWriting();
464
>
}
466
>
public write(msg: ProtocolMessage) {
468
// ignore: there could be left-over promises which complete and then
469
// decide to write a response, etc...
470
return;
471
}
473
>
this.lastWriteTime = Date.now();
474
>
const header = VSBuffer.alloc(ProtocolConstants.HeaderLength);
475
>
header.writeUInt8(msg.type, 0);
476
>
header.writeUInt32BE(msg.id, 1);
477
>
header.writeUInt32BE(msg.ack, 5);
478
>
header.writeUInt32BE(msg.data.byteLength, 9);
479
>
480
>
this._socket.traceSocketEvent(SocketDiagnosticsEventType.ProtocolHeaderWrite, { messageType: protocolMessageTypeToString(msg.type), id: msg.id, ack: msg.ack, messageSize: msg.data.byteLength });
481
>
this._socket.traceSocketEvent(SocketDiagnosticsEventType.ProtocolMessageWrite, msg.data);
482
>
483
>
this._writeSoon(header, msg.data);
484
>
}
486
>
private _bufferAdd(head: VSBuffer, body: VSBuffer): boolean {
488
>
this._data.push(head, body);
489
>
this._totalLength += head.byteLength + body.byteLength;
490
>
return wasEmpty;
491
>
}
493
>
private _bufferTake(): VSBuffer {
495
>
this._data.length = 0;
496
>
this._totalLength = 0;
497
>
return ret;
498
>
}
500
>
private _writeSoon(header: VSBuffer, data: VSBuffer): void {
502
>
this._scheduleWriting();
503
>
}
504
>
}
506
>
private _writeNowTimeout: Timeout | null = null;
507
>
private _scheduleWriting(): void {
509
return;
510
}
512
>
this._writeNowTimeout = null;
513
>
this._writeNow();
514
>
});
515
>
}
517
>
private _writeNow(): void {
519
>
return;
520
>
}
523
>
}
525
>
this._socket.traceSocketEvent(SocketDiagnosticsEventType.ProtocolWrite, { byteLength: data.byteLength });
526
>
this._socket.write(data);
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) {
561
>
this._socket = socket;
562
>
this._socketWriter = this._register(new ProtocolWriter(this._socket));
563
>
this._socketReader = this._register(new ProtocolReader(this._socket));
564
>
565
>
this._register(this._socketReader.onMessage((msg) => {
567
>
this._onMessage.fire(msg.data);
568
>
}
570
>
571
>
this._register(this._socket.onClose(() => this._onDidDispose.fire()));
572
>
}
574
>
drain(): Promise<void> {
575
return this._socketWriter.drain();
576
}
578
>
getSocket(): ISocket {
579
return this._socket;
580
}
582
>
sendDisconnect(): void {
583
// Nothing to do...
584
}
586
>
send(buffer: VSBuffer): void {
587
>
this._socketWriter.write(new ProtocolMessage(ProtocolMessageType.Regular, 0, 0, buffer));
ipc.net.ts ×2
588
>
}
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
}
603
>
override dispose(): void {
604
super.dispose();
605
const socket = this.protocol.getSocket();
606
// should be sent gracefully with a .flush(), but try to send it out as a
607
// last resort here if nothing else:
608
this.protocol.sendDisconnect();
609
this.protocol.dispose();
610
socket.end();
611
}
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() {
627
>
onWillAddFirstListener: () => {
628
>
this._hasListeners = true;
629
>
// it is important to deliver these messages after this call, but before
630
>
// other messages have a chance to be received (to guarantee in order delivery)
631
>
// that's why we're using here queueMicrotask and not other types of timeouts
632
>
queueMicrotask(() => this._deliverMessages());
633
>
},
634
>
onDidRemoveLastListener: () => {
635
>
this._hasListeners = false;
636
>
}
637
>
});
638
>
639
>
this.event = this._emitter.event;
640
>
}
642
>
private _deliverMessages(): void {
644
return;
645
}
647
>
while (this._hasListeners && this._bufferedMessages.length > 0) {
648
this._emitter.fire(this._bufferedMessages.shift()!);
649
}
651
>
}
653
>
public fire(event: T): void {
655
>
if (this._bufferedMessages.length > 0) {
656
this._bufferedMessages.push(event);
658
>
this._emitter.fire(event);
659
>
}
660
>
} else {
662
>
}
665
>
public flushBuffer(): void {
667
>
}
669
>
670
>
class QueueElement<T> {
671
>
public readonly data: T;
672
>
public next: QueueElement<T> | null;
673
>
674
>
constructor(data: T) {
676
>
this.next = null;
677
>
}
679
>
680
>
class Queue<T> {
681
>
682
>
private _first: QueueElement<T> | null;
683
>
private _last: QueueElement<T> | null;
684
>
685
>
constructor() {
687
>
this._last = null;
688
>
}
690
>
public length(): number {
692
>
let current = this._first;
693
>
while (current) {
695
>
result++;
696
>
}
698
>
}
700
>
public peek(): T | null {
703
>
}
705
>
}
707
>
public toArray(): T[] {
709
>
let resultLen = 0;
710
>
let it = this._first;
711
>
while (it) {
712
>
result[resultLen++] = it.data;
713
>
it = it.next;
714
>
}
715
>
return result;
716
>
}
718
>
public pop(): void {
720
return;
721
}
723
>
this._first = null;
724
>
this._last = null;
725
>
return;
726
>
}
730
>
public push(item: T): void {
732
>
if (!this._first) {
733
>
this._first = element;
734
>
this._last = element;
735
>
return;
736
>
}
738
>
this._last = element;
741
>
742
>
export class LoadEstimator {
743
>
744
>
private static _HISTORY_LENGTH = 10;
745
>
private static _INSTANCE: LoadEstimator | null = null;
746
>
public static getInstance(): LoadEstimator {
748
>
LoadEstimator._INSTANCE = new LoadEstimator();
749
>
}
750
>
return LoadEstimator._INSTANCE;
751
>
}
753
>
private lastRuns: number[];
754
>
755
>
constructor() {
757
>
const now = Date.now();
758
>
for (let i = 0; i < LoadEstimator._HISTORY_LENGTH; i++) {
759
>
this.lastRuns[i] = now - 1000 * i;
760
>
}
761
>
setInterval(() => {
762
for (let i = LoadEstimator._HISTORY_LENGTH; i >= 1; i--) {
763
this.lastRuns[i] = this.lastRuns[i - 1];
764
}
765
this.lastRuns[0] = Date.now();
767
>
}
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;
775
let score = 0;
776
for (let i = 0; i < LoadEstimator._HISTORY_LENGTH; i++) {
777
if (now - this.lastRuns[i] <= historyLimit) {
778
score++;
779
}
780
}
781
return 1 - score / LoadEstimator._HISTORY_LENGTH;
782
}
784
>
public hasHighLoad(): boolean {
785
return this.load() >= 0.5;
786
}
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) {
866
>
this._shouldSendKeepAlive = opts.sendKeepAlive ?? true;
867
>
this._isReconnecting = false;
868
>
this._outgoingUnackMsg = new Queue<ProtocolMessage>();
869
>
this._outgoingMsgId = 0;
870
>
this._outgoingAckId = 0;
871
>
this._outgoingAckTimeout = null;
872
>
873
>
this._incomingMsgId = 0;
874
>
this._incomingAckId = 0;
875
>
this._incomingMsgLastTime = 0;
876
>
this._incomingAckTimeout = null;
877
>
878
>
this._lastReplayRequestTime = 0;
879
>
this._lastSocketTimeoutTime = Date.now();
880
>
881
>
this._socketDisposables = new DisposableStore();
882
>
this._socket = opts.socket;
883
>
this._socketWriter = this._socketDisposables.add(new ProtocolWriter(this._socket));
884
>
this._socketReader = this._socketDisposables.add(new ProtocolReader(this._socket));
885
>
this._socketDisposables.add(this._socketReader.onMessage(msg => this._receiveMessage(msg)));
886
>
this._socketDisposables.add(this._socket.onClose(e => this._onSocketClose.fire(e)));
887
>
888
>
if (opts.initialChunk) {
889
this._socketReader.acceptChunk(opts.initialChunk);
890
}
892
>
if (this._shouldSendKeepAlive) {
898
>
}
901
>
dispose(): void {
904
>
this._outgoingAckTimeout = null;
905
>
}
908
>
this._incomingAckTimeout = null;
909
>
}
912
>
this._keepAliveInterval = null;
913
>
}
915
>
}
917
>
drain(): Promise<void> {
918
return this._socketWriter.drain();
919
}
921
>
sendDisconnect(): void {
922
if (!this._didSendDisconnect) {
923
this._didSendDisconnect = true;
924
const msg = new ProtocolMessage(ProtocolMessageType.Disconnect, 0, 0, getEmptyBuffer());
925
this._socketWriter.write(msg);
926
this._socketWriter.flush();
927
}
928
}
930
>
sendPause(): void {
931
>
const msg = new ProtocolMessage(ProtocolMessageType.Pause, 0, 0, getEmptyBuffer());
ipc.net.ts ×5
932
>
this._socketWriter.write(msg);
933
>
}
935
>
sendResume(): void {
936
>
const msg = new ProtocolMessage(ProtocolMessageType.Resume, 0, 0, getEmptyBuffer());
ipc.net.ts ×5
937
>
this._socketWriter.write(msg);
938
>
}
940
>
pauseSocketWriting() {
942
>
}
944
>
public getSocket(): ISocket {
945
return this._socket;
946
}
948
>
public getMillisSinceLastIncomingData(): number {
949
return Date.now() - this._socketReader.lastReadTime;
950
}
952
>
public beginAcceptReconnection(socket: ISocket, initialDataChunk: VSBuffer | null): void {
954
>
955
>
this._socketDisposables.dispose();
956
>
this._socketDisposables = new DisposableStore();
957
>
this._onControlMessage.flushBuffer();
958
>
this._onSocketClose.flushBuffer();
959
>
this._onSocketTimeout.flushBuffer();
960
>
this._socket.dispose();
961
>
962
>
this._lastReplayRequestTime = 0;
963
>
this._lastSocketTimeoutTime = Date.now();
964
>
965
>
this._socket = socket;
966
>
this._socketWriter = this._socketDisposables.add(new ProtocolWriter(this._socket));
967
>
this._socketReader = this._socketDisposables.add(new ProtocolReader(this._socket));
968
>
this._socketDisposables.add(this._socketReader.onMessage(msg => this._receiveMessage(msg)));
969
>
this._socketDisposables.add(this._socket.onClose(e => this._onSocketClose.fire(e)));
970
>
971
>
this._socketReader.acceptChunk(initialDataChunk);
972
>
}
974
>
public endAcceptReconnection(): void {
976
>
977
>
// After a reconnection, let the other party know (again) which messages have been received.
978
>
// (perhaps the other party didn't receive a previous ACK)
979
>
this._incomingAckId = this._incomingMsgId;
980
>
const msg = new ProtocolMessage(ProtocolMessageType.Ack, 0, this._incomingAckId, getEmptyBuffer());
981
>
this._socketWriter.write(msg);
982
>
983
>
// Send again all unacknowledged messages
984
>
const toSend = this._outgoingUnackMsg.toArray();
985
>
for (let i = 0, len = toSend.length; i < len; i++) {
986
>
this._socketWriter.write(toSend[i]);
987
>
}
988
>
this._recvAckCheck();
989
>
}
991
>
public acceptDisconnect(): void {
992
this._onDidDispose.fire();
993
}
995
>
private _receiveMessage(msg: ProtocolMessage): void {
998
>
do {
999
>
const first = this._outgoingUnackMsg.peek();
1000
>
if (first && first.id <= msg.ack) {
1001
>
// this message has been confirmed, remove it
1002
>
this._outgoingUnackMsg.pop();
1003
>
} else {
1004
>
break;
1005
>
}
1006
>
} while (true);
1007
>
}
1009
>
switch (msg.type) {
1010
>
case ProtocolMessageType.None: {
1011
// N/A
1012
break;
1013
}
1015
>
if (msg.id > this._incomingMsgId) {
1016
>
if (msg.id !== this._incomingMsgId + 1) {
1017
// in case we missed some messages we ask the other party to resend them
1018
const now = Date.now();
1019
if (now - this._lastReplayRequestTime > 10000) {
1020
// send a replay request at most once every 10s
1021
this._lastReplayRequestTime = now;
1022
this._socketWriter.write(new ProtocolMessage(ProtocolMessageType.ReplayRequest, 0, 0, getEmptyBuffer()));
1023
}
1025
>
this._incomingMsgId = msg.id;
1026
>
this._incomingMsgLastTime = Date.now();
1027
>
this._sendAckCheck();
1028
>
this._onMessage.fire(msg.data);
1029
>
}
1030
>
}
1031
>
break;
1032
>
}
1033
>
case ProtocolMessageType.Control: {
1034
this._onControlMessage.fire(msg.data);
1035
break;
1036
}
1039
>
break;
1040
>
}
1042
this._onDidDispose.fire();
1043
break;
1044
}
1046
// Send again all unacknowledged messages
1047
const toSend = this._outgoingUnackMsg.toArray();
1048
for (let i = 0, len = toSend.length; i < len; i++) {
1049
this._socketWriter.write(toSend[i]);
1050
}
1051
this._recvAckCheck();
1052
break;
1053
}
1056
>
break;
1057
>
}
1060
>
break;
1061
>
}
1064
>
break;
1065
>
}
1067
>
}
1069
>
readEntireBuffer(): VSBuffer {
1070
return this._socketReader.readEntireBuffer();
1071
}
1073
>
flush(): void {
1074
this._socketWriter.flush();
1075
}
1077
>
send(buffer: VSBuffer): void {
1079
>
this._incomingAckId = this._incomingMsgId;
1080
>
const msg = new ProtocolMessage(ProtocolMessageType.Regular, myId, this._incomingAckId, buffer);
1081
>
this._outgoingUnackMsg.push(msg);
1082
>
if (!this._isReconnecting) {
1083
>
this._socketWriter.write(msg);
1084
>
this._recvAckCheck();
1085
>
}
1086
>
}
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
}
1097
>
private _sendAckCheck(): void {
1100
>
return;
1101
>
}
1103
>
if (this._incomingAckTimeout) {
1105
>
return;
1106
>
}
1108
>
const timeSinceLastIncomingMsg = Date.now() - this._incomingMsgLastTime;
1109
>
if (timeSinceLastIncomingMsg >= ProtocolConstants.AcknowledgeTime) {
1111
>
// and no message from our side needed to be sent in the meantime,
1112
>
// so we will send a message containing only an ack.
1113
>
this._sendAck();
1114
>
return;
1115
>
}
1117
>
this._incomingAckTimeout = setTimeout(() => {
1119
>
this._sendAckCheck();
1121
>
}
1123
>
private _recvAckCheck(): void {
1126
>
return;
1127
>
}
1129
>
if (this._outgoingAckTimeout) {
1131
>
return;
1132
>
}
1134
>
if (this._isReconnecting) {
1136
>
// because messages will not be actually written until `endAcceptReconnection`
1137
>
return;
1138
>
}
1140
>
const oldestUnacknowledgedMsg = this._outgoingUnackMsg.peek()!;
1141
>
const timeSinceOldestUnacknowledgedMsg = Date.now() - oldestUnacknowledgedMsg.writtenTime;
1142
>
const timeSinceLastReceivedSomeData = Date.now() - this._socketReader.lastReadTime;
1143
>
const timeSinceLastTimeout = Date.now() - this._lastSocketTimeoutTime;
1144
>
1145
>
if (
1146
>
timeSinceOldestUnacknowledgedMsg >= ProtocolConstants.TimeoutTime
1148
>
&& timeSinceLastTimeout >= ProtocolConstants.TimeoutTime
1151
>
// and a long time since we received some data
1152
>
1153
>
// But this might be caused by the event loop being busy and failing to read messages
1154
>
if (!this._loadEstimator.hasHighLoad()) {
1155
>
// Trash the socket
1156
>
this._lastSocketTimeoutTime = Date.now();
1157
>
this._onSocketTimeout.fire({
1158
>
reason: SocketTimeoutReason.UNACKNOWLEDGED_MESSAGE,
1159
>
unacknowledgedMsgCount: this._outgoingUnackMsg.length(),
1160
>
timeSinceOldestUnacknowledgedMsg,
1161
>
timeSinceLastReceivedSomeData
1162
>
});
1163
>
return;
1164
>
}
1165
>
}
1167
>
const minimumTimeUntilTimeout = Math.max(
1168
>
ProtocolConstants.TimeoutTime - timeSinceOldestUnacknowledgedMsg,
1169
>
ProtocolConstants.TimeoutTime - timeSinceLastReceivedSomeData,
1170
>
ProtocolConstants.TimeoutTime - timeSinceLastTimeout,
1171
>
500
1172
>
);
1173
>
1174
>
this._outgoingAckTimeout = setTimeout(() => {
1176
>
this._recvAckCheck();
1178
>
}
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 {
1189
return;
1190
}
1192
>
const now = Date.now();
1193
>
const timeSinceLastReceivedSomeData = now - this._socketReader.lastReadTime;
1194
>
const timeSinceLastTimeout = now - this._lastSocketTimeoutTime;
1195
>
1196
>
if (
1197
>
timeSinceLastReceivedSomeData >= ProtocolConstants.TimeoutTime
1200
>
// But this might be caused by the event loop being busy and failing to read messages
ipc.net.ts ×1
1201
>
if (!this._loadEstimator.hasHighLoad()) {
1202
>
this._lastSocketTimeoutTime = now;
1203
>
const unacknowledgedMsgCount = this._outgoingUnackMsg.length();
1204
>
const oldestUnacknowledgedMsg = this._outgoingUnackMsg.peek();
1205
>
this._onSocketTimeout.fire({
1206
>
reason: SocketTimeoutReason.KEEP_ALIVE,
1207
>
unacknowledgedMsgCount,
1208
>
timeSinceOldestUnacknowledgedMsg: oldestUnacknowledgedMsg ? now - oldestUnacknowledgedMsg.writtenTime : undefined,
1209
>
timeSinceLastReceivedSomeData
1210
>
});
1211
>
}
1212
>
}
1215
>
private _sendAck(): void {
1217
// nothink to acknowledge
1218
return;
1219
}
1221
>
this._incomingAckId = this._incomingMsgId;
1222
>
const msg = new ProtocolMessage(ProtocolMessageType.Ack, 0, this._incomingAckId, getEmptyBuffer());
1223
>
this._socketWriter.write(msg);
1224
>
}
1226
>
private _sendKeepAlive(): void {
1228
>
const msg = new ProtocolMessage(ProtocolMessageType.KeepAlive, 0, this._incomingAckId, getEmptyBuffer());
1229
>
this._socketWriter.write(msg);
1230
>
this._keepAliveTimeoutCheck();
1231
>
}