Skip to content

Commit

Permalink
Add ability to manually shut down channels
Browse files Browse the repository at this point in the history
Addresses grpc#198.

It's currently not possible to manually shut down a gRPC channel, which means that the only way to shut down a channel is to deallocate it. This requirement is a bit fragile for cases where a consumer wants to manually shut down a channel, since it requires confidence that nothing else is holding a strong reference to the channel.

This PR:
- Adds a `shutdown` function to `Channel`, allowing consumers to arbitrarily shut down connections
- Validates that existing calls will throw errors when attempting to read/write from a previously shut down channel
- Ensures creating new calls using previously shut down channels will result in the initializer throwing (already handled by code generators)

This change is increasingly relevant because consumers will need to shut down channels and restart them to help mitigate issues when switching between wifi/cellular as mentioned here: grpc#337.
  • Loading branch information
rebello95 committed Feb 26, 2019
1 parent 10aff09 commit 0a159d5
Show file tree
Hide file tree
Showing 7 changed files with 132 additions and 27 deletions.
17 changes: 9 additions & 8 deletions Examples/SimpleXcode/Simple/Document.swift
Original file line number Diff line number Diff line change
Expand Up @@ -144,16 +144,17 @@ class Document: NSDocument {
if !self.isRunning() {
break
}
let method = (i < steps) ? "/hello" : "/quit"
let call = self.channel.makeCall(method)

let metadata = try! Metadata([
"x": "xylophone",
"y": "yu",
"z": "zither"
])

do {
let method = (i < steps) ? "/hello" : "/quit"
let call = try self.channel.makeCall(method)

let metadata = try Metadata([
"x": "xylophone",
"y": "yu",
"z": "zither"
])

try call.start(.unary,
metadata: metadata,
message: messageData) { callResult in
Expand Down
2 changes: 1 addition & 1 deletion Sources/Examples/Simple/main.swift
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ func client() throws {

let method = (i < steps - 1) ? "/hello" : "/quit"
print("calling " + method)
let call = c.makeCall(method)
let call = try! c.makeCall(method)

let metadata = try Metadata([
"x": "xylophone",
Expand Down
39 changes: 32 additions & 7 deletions Sources/SwiftGRPC/Core/Channel.swift
Original file line number Diff line number Diff line change
Expand Up @@ -27,13 +27,23 @@ public class Channel {
private let completionQueue: CompletionQueue
/// Observer for connectivity state changes. Created lazily if needed
private var connectivityObserver: ConnectivityObserver?
/// Whether the gRPC channel has been shut down
private var hasBeenShutdown = false

/// Timeout for new calls
public var timeout: TimeInterval = 600.0

/// Default host to use for new calls
public var host: String

/// Errors that may be thrown by the channel.
enum Error: Swift.Error {
/// Action cannot be performed because the channel has already been shut down.
case alreadyShutdown
/// Failed to create a new call within the gRPC stack.
case callCreationFailed
}

/// Initializes a gRPC channel
///
/// - Parameter address: the address of the server to be called
Expand Down Expand Up @@ -94,12 +104,20 @@ public class Channel {
completionQueue.run() // start a loop that watches the channel's completion queue
}

deinit {
/// Shut down the channel. No new calls may be made using this channel after it is shut down
public func shutdown() {
self.mutex.synchronize {
guard !self.hasBeenShutdown else { return }

self.hasBeenShutdown = true
self.connectivityObserver?.shutdown()
cgrpc_channel_destroy(self.underlyingChannel)
self.completionQueue.shutdown()
}
cgrpc_channel_destroy(self.underlyingChannel)
self.completionQueue.shutdown()
}

deinit {
self.shutdown()
}

/// Constructs a Call object to make a gRPC API call
Expand All @@ -108,10 +126,17 @@ public class Channel {
/// - Parameter host: the gRPC host name for the call. If unspecified, defaults to the Client host
/// - Parameter timeout: a timeout value in seconds
/// - Returns: a Call object that can be used to perform the request
public func makeCall(_ method: String, host: String = "", timeout: TimeInterval? = nil) -> Call {
let host = host.isEmpty ? self.host : host
let timeout = timeout ?? self.timeout
let underlyingCall = cgrpc_channel_create_call(underlyingChannel, method, host, timeout)!
public func makeCall(_ method: String, host: String? = nil, timeout: TimeInterval? = nil) throws -> Call {
guard (self.mutex.synchronize { !self.hasBeenShutdown }) else {
throw Error.alreadyShutdown
}

guard let underlyingCall = cgrpc_channel_create_call(
self.underlyingChannel, method, host ?? self.host, timeout ?? self.timeout) else
{
throw Error.callCreationFailed
}

return Call(underlyingCall: underlyingCall, owned: true, completionQueue: completionQueue)
}

Expand Down
15 changes: 7 additions & 8 deletions Sources/SwiftGRPC/Runtime/ClientCall.swift
Original file line number Diff line number Diff line change
Expand Up @@ -13,14 +13,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/

import Dispatch
import Foundation
import SwiftProtobuf

public protocol ClientCall: class {
static var method: String { get }

/// Cancel the call.
func cancel()
}
Expand All @@ -31,9 +28,11 @@ open class ClientCallBase: ClientCall {
public let call: Call

/// Create a call.
public init(_ channel: Channel) {
call = channel.makeCall(type(of: self).method)
public init(_ channel: Channel) throws {
self.call = try channel.makeCall(type(of: self).method)
}

public func cancel() {
self.call.cancel()
}

public func cancel() { call.cancel() }
}
1 change: 1 addition & 0 deletions Tests/LinuxMain.swift
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ XCTMain([
testCase(gRPCTests.allTests),
testCase(ChannelArgumentTests.allTests),
testCase(ChannelConnectivityTests.allTests),
testCase(ChannelShutdownTests.allTests),
testCase(ClientCancellingTests.allTests),
testCase(ClientTestExample.allTests),
testCase(ClientTimeoutTests.allTests),
Expand Down
79 changes: 79 additions & 0 deletions Tests/SwiftGRPCTests/ChannelShutdownTests.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
/*
* Copyright 2018, gRPC Authors All rights reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
@testable import SwiftGRPC
import XCTest

final class ChannelShutdownTests: BasicEchoTestCase {
static var allTests: [(String, (ChannelShutdownTests) -> () throws -> Void)] {
return [
("testThrowsWhenCreatingCallWithAlreadyShutDownChannel", testThrowsWhenCreatingCallWithAlreadyShutDownChannel),
("testCallReceiveThrowsWhenChannelIsShutDown", testCallReceiveThrowsWhenChannelIsShutDown),
("testCallCloseThrowsWhenChannelIsShutDown", testCallCloseThrowsWhenChannelIsShutDown),
("testCallCloseAndReceiveThrowsWhenChannelIsShutDown", testCallCloseAndReceiveThrowsWhenChannelIsShutDown),
("testCallSendThrowsWhenChannelIsShutDown", testCallSendThrowsWhenChannelIsShutDown),
]
}
}

extension ChannelShutdownTests {
func testThrowsWhenCreatingCallWithAlreadyShutDownChannel() {
self.client.channel.shutdown()

XCTAssertThrowsError(try self.client.channel.makeCall("foobar")) { error in
XCTAssertEqual(.alreadyShutdown, error as? Channel.Error)
}
}

func testCallReceiveThrowsWhenChannelIsShutDown() {
let call = try! self.client.channel.makeCall("foo")
self.client.channel.shutdown()

XCTAssertThrowsError(try call.receiveMessage(completion: { _ in })) { error in
XCTAssertEqual(.completionQueueShutdown, error as? CallError)
XCTAssertNotNil(self.client.channel)
}
}

func testCallCloseThrowsWhenChannelIsShutDown() {
let call = try! self.client.channel.makeCall("foo")
self.client.channel.shutdown()

XCTAssertThrowsError(try call.close()) { error in
XCTAssertEqual(.completionQueueShutdown, error as? CallError)
XCTAssertNotNil(self.client.channel)
}
}

func testCallCloseAndReceiveThrowsWhenChannelIsShutDown() {
let call = try! self.client.channel.makeCall("foo")
self.client.channel.shutdown()

XCTAssertThrowsError(try call.closeAndReceiveMessage(completion: { _ in })) { error in
XCTAssertEqual(.completionQueueShutdown, error as? CallError)
XCTAssertNotNil(self.client.channel)
}
}

func testCallSendThrowsWhenChannelIsShutDown() {
let call = try! self.client.channel.makeCall("foo")
self.client.channel.shutdown()

XCTAssertThrowsError(try call.sendMessage(data: Data())) { error in
XCTAssertEqual(.completionQueueShutdown, error as? CallError)
XCTAssertNotNil(self.client.channel)
}
}
}
6 changes: 3 additions & 3 deletions Tests/SwiftGRPCTests/GRPCTests.swift
Original file line number Diff line number Diff line change
Expand Up @@ -177,7 +177,7 @@ func callUnary(channel: Channel) throws {
func callUnaryIndividual(channel: Channel, message: Data, shouldSucceed: Bool) throws {
let sem = DispatchSemaphore(value: 0)
let method = hello
let call = channel.makeCall(method)
let call = try channel.makeCall(method)
let metadata = try Metadata(initialClientMetadata)
try call.start(.unary, metadata: metadata, message: message) {
response in
Expand Down Expand Up @@ -228,7 +228,7 @@ func callServerStream(channel: Channel) throws {

let sem = DispatchSemaphore(value: 0)
let method = helloServerStream
let call = channel.makeCall(method)
let call = try channel.makeCall(method)
try call.start(.serverStreaming, metadata: metadata, message: message) {
response in

Expand Down Expand Up @@ -270,7 +270,7 @@ func callBiDiStream(channel: Channel) throws {

let sem = DispatchSemaphore(value: 0)
let method = helloBiDiStream
let call = channel.makeCall(method)
let call = try channel.makeCall(method)
try call.start(.bidiStreaming, metadata: metadata, message: nil) {
response in

Expand Down

0 comments on commit 0a159d5

Please sign in to comment.