grpc
/
grpc-node
mirror of https://github.com/grpc/grpc-node.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1580 from murgatroid99/grpc-js_event_loop_liveness 

grpc-js: Add a timer that holds the event loop open while there are pending calls
This commit is contained in:
Michael Lumish 2020-09-17 14:38:34 -07:00 committed by GitHub
parent 2445875e08 a1600812d1
commit b51afd3852
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 28 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
packages/grpc-js/package.json
View File

@ -1,6 +1,6 @@
{
"name": "@grpc/grpc-js",
"version": "1.1.6",
"version": "1.1.7",
"description": "gRPC Library for Node - pure JS implementation",
"homepage": "https://grpc.io/",
"repository": "https://github.com/grpc/grpc-node/tree/master/packages/grpc-js",

30
packages/grpc-js/src/channel.ts
View File

@ -48,6 +48,11 @@ export enum ConnectivityState {
SHUTDOWN,
}
/**
* See https://nodejs.org/api/timers.html#timers_setinterval_callback_delay_args
*/
const MAX_TIMEOUT_TIME = 2147483647;
let nextCallNumber = 0;
function getNewCallNumber(): number {
@ -137,6 +142,14 @@ export class ChannelImplementation implements Channel {
private defaultAuthority: string;
private filterStackFactory: FilterStackFactory;
private target: GrpcUri;
/**
* This timer does not do anything on its own. Its purpose is to hold the
* event loop open while there are any pending calls for the channel that
* have not yet been assigned to specific subchannels. In other words,
* the invariant is that callRefTimer is reffed if and only if pickQueue
* is non-empty.
*/
private callRefTimer: NodeJS.Timer;
constructor(
target: string,
private readonly credentials: ChannelCredentials,
@ -177,6 +190,10 @@ export class ChannelImplementation implements Channel {
`Could not find a default scheme for target name "${target}"`
);
}
this.callRefTimer = setInterval(() => {}, MAX_TIMEOUT_TIME);
this.callRefTimer.unref?.();
if (this.options['grpc.default_authority']) {
this.defaultAuthority = this.options['grpc.default_authority'] as string;
} else {
@ -206,6 +223,7 @@ export class ChannelImplementation implements Channel {
updateState: (connectivityState: ConnectivityState, picker: Picker) => {
this.currentPicker = picker;
const queueCopy = this.pickQueue.slice();
this.callRefTimer.unref?.();
this.pickQueue = [];
for (const { callStream, callMetadata } of queueCopy) {
this.tryPick(callStream, callMetadata);
@ -232,6 +250,11 @@ export class ChannelImplementation implements Channel {
]);
}
private pushPick(callStream: Http2CallStream, callMetadata: Metadata) {
this.callRefTimer.ref?.();
this.pickQueue.push({ callStream, callMetadata });
}
/**
* Check the picker output for the given call and corresponding metadata,
* and take any relevant actions. Should not be called while iterating
@ -276,7 +299,7 @@ export class ChannelImplementation implements Channel {
' has state ' +
ConnectivityState[pickResult.subchannel!.getConnectivityState()]
);
this.pickQueue.push({ callStream, callMetadata });
this.pushPick(callStream, callMetadata);
break;
}
/* We need to clone the callMetadata here because the transparent
@ -367,11 +390,11 @@ export class ChannelImplementation implements Channel {
}
break;
case PickResultType.QUEUE:
this.pickQueue.push({ callStream, callMetadata });
this.pushPick(callStream, callMetadata);
break;
case PickResultType.TRANSIENT_FAILURE:
if (callMetadata.getOptions().waitForReady) {
this.pickQueue.push({ callStream, callMetadata });
this.pushPick(callStream, callMetadata);
} else {
callStream.cancelWithStatus(
pickResult.status!.code,
@ -433,6 +456,7 @@ export class ChannelImplementation implements Channel {
close() {
this.resolvingLoadBalancer.destroy();
this.updateState(ConnectivityState.SHUTDOWN);
clearInterval(this.callRefTimer);
this.subchannelPool.unrefUnusedSubchannels();
}