Merge pull request microsoft#138287 from microsoft/alex/connection-improvements

Improve remote connection

Author: alexdima

src/vs/base/parts/ipc/common/ipc.net.ts

@@ -778,6 +778,7 @@ export class PersistentProtocol implements IMessagePassingProtocol {
 	private _incomingAckTimeout: any | null;
 
 	private _lastReplayRequestTime: number;
+	private _lastSocketTimeoutTime: number;
 
 	private _socket: ISocket;
 	private _socketWriter: ProtocolWriter;
@@ -819,6 +820,7 @@ export class PersistentProtocol implements IMessagePassingProtocol {
 		this._incomingAckTimeout = null;
 
 		this._lastReplayRequestTime = 0;
+		this._lastSocketTimeoutTime = Date.now();
 
 		this._socketDisposables = [];
 		this._socket = socket;
@@ -887,6 +889,7 @@ export class PersistentProtocol implements IMessagePassingProtocol {
 		this._socket.dispose();
 
 		this._lastReplayRequestTime = 0;
+		this._lastSocketTimeoutTime = Date.now();
 
 		this._socket = socket;
 		this._socketWriter = new ProtocolWriter(this._socket);
@@ -901,6 +904,12 @@ export class PersistentProtocol implements IMessagePassingProtocol {
 	public endAcceptReconnection(): void {
 		this._isReconnecting = false;
 
+		// After a reconnection, let the other party know (again) which messages have been received.
+		// (perhaps the other party didn't receive a previous ACK)
+		this._incomingAckId = this._incomingMsgId;
+		const msg = new ProtocolMessage(ProtocolMessageType.Ack, 0, this._incomingAckId, getEmptyBuffer());
+		this._socketWriter.write(msg);
+
 		// Send again all unacknowledged messages
 		const toSend = this._outgoingUnackMsg.toArray();
 		for (let i = 0, len = toSend.length; i < len; i++) {
@@ -1057,26 +1066,36 @@ export class PersistentProtocol implements IMessagePassingProtocol {
 		const oldestUnacknowledgedMsg = this._outgoingUnackMsg.peek()!;
 		const timeSinceOldestUnacknowledgedMsg = Date.now() - oldestUnacknowledgedMsg.writtenTime;
 		const timeSinceLastReceivedSomeData = Date.now() - this._socketReader.lastReadTime;
+		const timeSinceLastTimeout = Date.now() - this._lastSocketTimeoutTime;
 
 		if (
 			timeSinceOldestUnacknowledgedMsg >= ProtocolConstants.TimeoutTime
 			&& timeSinceLastReceivedSomeData >= ProtocolConstants.TimeoutTime
+			&& timeSinceLastTimeout >= ProtocolConstants.TimeoutTime
 		) {
 			// It's been a long time since our sent message was acknowledged
 			// and a long time since we received some data
 
 			// But this might be caused by the event loop being busy and failing to read messages
 			if (!this._loadEstimator.hasHighLoad()) {
 				// Trash the socket
+				this._lastSocketTimeoutTime = Date.now();
 				this._onSocketTimeout.fire(undefined);
 				return;
 			}
 		}
 
+		const minimumTimeUntilTimeout = Math.max(
+			ProtocolConstants.TimeoutTime - timeSinceOldestUnacknowledgedMsg,
+			ProtocolConstants.TimeoutTime - timeSinceLastReceivedSomeData,
+			ProtocolConstants.TimeoutTime - timeSinceLastTimeout,
+			500
+		);
+
 		this._outgoingAckTimeout = setTimeout(() => {
 			this._outgoingAckTimeout = null;
 			this._recvAckCheck();
-		}, Math.max(ProtocolConstants.TimeoutTime - timeSinceOldestUnacknowledgedMsg, 500));
+		}, minimumTimeUntilTimeout);
 	}
 
 	private _sendAck(): void {

src/vs/base/parts/ipc/test/node/ipc.net.test.ts

@@ -9,7 +9,7 @@ import { createServer, Socket } from 'net';
 import { tmpdir } from 'os';
 import { Barrier, timeout } from 'vs/base/common/async';
 import { VSBuffer } from 'vs/base/common/buffer';
-import { Emitter } from 'vs/base/common/event';
+import { Emitter, Event } from 'vs/base/common/event';
 import { Disposable, DisposableStore } from 'vs/base/common/lifecycle';
 import { ILoadEstimator, PersistentProtocol, Protocol, ProtocolConstants, SocketCloseEvent, SocketDiagnosticsEventType } from 'vs/base/parts/ipc/common/ipc.net';
 import { createRandomIPCHandle, createStaticIPCHandle, NodeSocket, WebSocketNodeSocket } from 'vs/base/parts/ipc/node/ipc.net';
@@ -90,21 +90,25 @@ class Ether {
 		return <any>this._b;
 	}
 
-	constructor() {
+	constructor(
+		private readonly _wireLatency = 0
+	) {
 		this._a = new EtherStream(this, 'a');
 		this._b = new EtherStream(this, 'b');
 		this._ab = [];
 		this._ba = [];
 	}
 
 	public write(from: 'a' | 'b', data: Buffer): void {
-		if (from === 'a') {
-			this._ab.push(data);
-		} else {
-			this._ba.push(data);
-		}
+		setTimeout(() => {
+			if (from === 'a') {
+				this._ab.push(data);
+			} else {
+				this._ba.push(data);
+			}
 
-		setTimeout(() => this._deliver(), 0);
+			setTimeout(() => this._deliver(), 0);
+		}, this._wireLatency);
 	}
 
 	private _deliver(): void {
@@ -365,6 +369,120 @@ suite('PersistentProtocol reconnection', () => {
 		);
 	});
 
+	test('acks are always sent after a reconnection', async () => {
+		await runWithFakedTimers(
+			{
+				useFakeTimers: true,
+				useSetImmediate: true,
+				maxTaskCount: 1000
+			},
+			async () => {
+
+				const loadEstimator: ILoadEstimator = {
+					hasHighLoad: () => false
+				};
+				const wireLatency = 1000;
+				const ether = new Ether(wireLatency);
+				const aSocket = new NodeSocket(ether.a);
+				const a = new PersistentProtocol(aSocket, null, loadEstimator);
+				const aMessages = new MessageStream(a);
+				const bSocket = new NodeSocket(ether.b);
+				const b = new PersistentProtocol(bSocket, null, loadEstimator);
+				const bMessages = new MessageStream(b);
+
+				// send message a1 to have something unacknowledged
+				a.send(VSBuffer.fromString('a1'));
+				assert.strictEqual(a.unacknowledgedCount, 1);
+				assert.strictEqual(b.unacknowledgedCount, 0);
+
+				// read message a1 at B
+				const a1 = await bMessages.waitForOne();
+				assert.strictEqual(a1.toString(), 'a1');
+				assert.strictEqual(a.unacknowledgedCount, 1);
+				assert.strictEqual(b.unacknowledgedCount, 0);
+
+				// wait for B to send an ACK message,
+				// but resume before A receives it
+				await timeout(ProtocolConstants.AcknowledgeTime + wireLatency / 2);
+				assert.strictEqual(a.unacknowledgedCount, 1);
+				assert.strictEqual(b.unacknowledgedCount, 0);
+
+				// simulate complete reconnection
+				aSocket.dispose();
+				bSocket.dispose();
+				const ether2 = new Ether(wireLatency);
+				const aSocket2 = new NodeSocket(ether2.a);
+				const bSocket2 = new NodeSocket(ether2.b);
+				b.beginAcceptReconnection(bSocket2, null);
+				b.endAcceptReconnection();
+				a.beginAcceptReconnection(aSocket2, null);
+				a.endAcceptReconnection();
+
+				// wait for quite some time
+				await timeout(2 * ProtocolConstants.AcknowledgeTime + wireLatency);
+				assert.strictEqual(a.unacknowledgedCount, 0);
+				assert.strictEqual(b.unacknowledgedCount, 0);
+
+				aMessages.dispose();
+				bMessages.dispose();
+				a.dispose();
+				b.dispose();
+			}
+		);
+	});
+
+	test('onSocketTimeout is emitted at most once every 20s', async () => {
+		await runWithFakedTimers(
+			{
+				useFakeTimers: true,
+				useSetImmediate: true,
+				maxTaskCount: 1000
+			},
+			async () => {
+
+				const loadEstimator: ILoadEstimator = {
+					hasHighLoad: () => false
+				};
+				const ether = new Ether();
+				const aSocket = new NodeSocket(ether.a);
+				const a = new PersistentProtocol(aSocket, null, loadEstimator);
+				const aMessages = new MessageStream(a);
+				const bSocket = new NodeSocket(ether.b);
+				const b = new PersistentProtocol(bSocket, null, loadEstimator);
+				const bMessages = new MessageStream(b);
+
+				// never receive acks
+				b.pauseSocketWriting();
+
+				// send message a1 to have something unacknowledged
+				a.send(VSBuffer.fromString('a1'));
+
+				// wait for the first timeout to fire
+				await Event.toPromise(a.onSocketTimeout);
+
+				let timeoutFiredAgain = false;
+				const timeoutListener = a.onSocketTimeout(() => {
+					timeoutFiredAgain = true;
+				});
+
+				// send more messages
+				a.send(VSBuffer.fromString('a2'));
+				a.send(VSBuffer.fromString('a3'));
+
+				// wait for 10s
+				await timeout(ProtocolConstants.TimeoutTime / 2);
+
+				assert.strictEqual(timeoutFiredAgain, false);
+
+				timeoutListener.dispose();
+				aMessages.dispose();
+				bMessages.dispose();
+				a.dispose();
+				b.dispose();
+			}
+		);
+	});
+
 	test('writing can be paused', async () => {
 		await runWithFakedTimers({ useFakeTimers: true, maxTaskCount: 100 }, async () => {
 			const loadEstimator: ILoadEstimator = {

src/vs/platform/remote/common/remoteAgentConnection.ts

@@ -573,7 +573,7 @@ export abstract class PersistentConnection extends Disposable {
 		}));
 		this._register(protocol.onSocketTimeout(() => {
 			const logPrefix = commonLogPrefix(this._connectionType, this.reconnectionToken, true);
-			this._options.logService.trace(`${logPrefix} received socket timeout event.`);
+			this._options.logService.info(`${logPrefix} received socket timeout event.`);
 			this._beginReconnecting();
 		}));
 

src/vs/platform/remote/node/nodeSocketFactory.ts

@@ -21,7 +21,7 @@ export const nodeSocketFactory = new class implements ISocketFactory {
 			}
 			const nonce = buffer.toString('base64');
 
-			let headers = [
+			const headers = [
 				`GET ws://${/:/.test(host) ? `[${host}]` : host}:${port}/?${query}&skipWebSocketFrames=true HTTP/1.1`,
 				`Connection: Upgrade`,
 				`Upgrade: websocket`,
@@ -39,6 +39,8 @@ export const nodeSocketFactory = new class implements ISocketFactory {
 			};
 			socket.on('data', onData);
 		});
+		// Disable Nagle's algorithm.
+		socket.setNoDelay(true);
 		socket.once('error', errorListener);
 	}
 };

src/vs/server/remoteExtensionHostAgentServer.ts

@@ -497,6 +497,8 @@ export class RemoteExtensionHostAgentServer extends Disposable {
 
 		// Never timeout this socket due to inactivity!
 		socket.setTimeout(0);
+		// Disable Nagle's algorithm
+		socket.setNoDelay(true);
 		// Finally!
 
 		if (skipWebSocketFrames) {

src/vs/workbench/services/extensions/node/extensionHostProcessSetup.ts

@@ -120,6 +120,10 @@ function _createExtHostProtocol(): Promise<PersistentProtocol> {
 
 			process.on('message', (msg: IExtHostSocketMessage | IExtHostReduceGraceTimeMessage, handle: net.Socket) => {
 				if (msg && msg.type === 'VSCODE_EXTHOST_IPC_SOCKET') {
+					// Disable Nagle's algorithm. We also do this on the server process,
+					// but nodejs doesn't document if this option is transferred with the socket
+					handle.setNoDelay(true);
+
 					const initialDataChunk = VSBuffer.wrap(Buffer.from(msg.initialDataChunk, 'base64'));
 					let socket: NodeSocket | WebSocketNodeSocket;
 					if (msg.skipWebSocketFrames) {