channel.shutdown() doesn't work if stream is cancelled before its end

[a14n]

Cancelling a resulting stream that is not done prevent channel.shutdown() to free resources on client side and let the program stop.

The issue exists in 1.0.3 and master (91564ff7aa33214b55d050bed44be1b875a6713b)

Repro steps

// service.proto
syntax = "proto3";
service Service {
  rpc Method (Empty) returns (stream Empty);
}
message Empty {}


// server.dart
import 'dart:async';

import 'package:grpc/grpc.dart';
import 'package:issue_shutdown/src/generated/service.pbgrpc.dart';

main(List<String> args) async {
  await Server([Service()]).serve(port: 8022);
}

class Service extends ServiceBase {
  @override
  Stream<Empty> method(ServiceCall call, Empty request) async* {
    yield Empty();
    await Future.delayed(Duration(seconds: 1));
    yield Empty();
  }
}


// client.dart
import 'package:grpc/grpc.dart';
import 'package:issue_shutdown/src/generated/service.pbgrpc.dart';

main(List<String> args) async {
  final channel = ClientChannel(
    'localhost',
    port: 8022,
    options: const ChannelOptions(credentials: ChannelCredentials.insecure()),
  );
  final client = ServiceClient(channel);

  print(await client.method(Empty()).take(1).length);
  await channel.shutdown();
}

• Run server.dart
• Run client.dart

Expected result: client.dart stops running

Actual result: client.dart is still running

Details

client.dart correctly stops if you consume the stream entierely (by removing take(1) for instance). But .take(1) cancels the stream before its end and it seems to be the reason the client program don't stop.

[sigurdm]

I believe this is (at least partially) working as intended (even though it a somewhat confusing design) Canceling the StreamSubcription does not signal cancellation of the call (it only stops listening to it). You have to use ResponseStream.cancel().

If you replace the code with:

main(List<String> args) async {
  final channel = ClientChannel(
    'localhost',
    port: 8022,
    options: const ChannelOptions(credentials: ChannelCredentials.insecure()),
  );
  final client = ServiceClient(channel);

  ResponseStream results = client.method(Empty());
  StreamIterator i = StreamIterator(results);
  await i.moveNext();
  results.cancel();
  await channel.shutdown();
}

You get the intended behavior.

Still my immediate thought is that the program should stop after the server finishes the call. @jonasfj do you have thoughts?

[a14n]

It's really strange to face 2 different behaviours in the following samples:

  // shutdown() works and stop the program
  print(await client.method(Empty()).length);
  await channel.shutdown();

  // shutdown() doesn't work and the program stays alive
  print(await client.method(Empty()).take(1).length);
  await channel.shutdown();

[jonasfj]

I recall we went down this rabbit hole and explored why this was the way it is.. But I don't recall what we found :)

But whether intended or not, I agree with @sigurdm that this is a somewhat confusing design. That said it's probably not trivial to fix.

I think result.cancel() exists because you can cancel before the connection is established. But given that it's a single subscription stream, stopping to listen should perhaps also cancel the request.

[sigurdm]

@mkustermann I remember you have been arguing that cancelling the subscription should not cancel the request. But since it is single subscription, and given the documentation for StreamSubscription.cancel says that the stream can clean up, I am getting convinced we should change this. Do you agree?

[mkustermann]

I remember you have been arguing that cancelling the subscription should not cancel the request.

I think this was in the context of bi-directional streams: If you cancel the incoming stream (not interested in receiving more messages from the server), it doesn't necessarily imply that you don't want to send more messages to the server.

If the stream is one-directional only, and you are not interested in any more messages, then I agree that there is no point in keeping the grpc request (and it's underlying http/2 stream) alive anymore.

[morgwai]

just stumbled upon this also:

I believe this is (at least partially) working as intended (even though it a somewhat confusing design) Canceling the StreamSubcription does not signal cancellation of the call (it only stops listening to it).

correct, however channel.shutdown() should forcibly cancel all outstanding RPCs, I think. At least that's what java does. Therefore ResponseStream.cancel() should not be necessary IMO for channel.shutdown() to work as expected.

[morgwai]

FYI: I cannot reproduce this anymore with version grpc-3.0.0 (before I was running some really old package, not sure even what version it was using. After rebuilding with grpc-3.0.0, it seems to work as intended for me)