Implement connectivity state property and observers #196

cvanderschuere · 2018-03-26T23:14:43Z

Implementation for #186

The goal was to copy the interface that the python client exposes.

    def subscribe(self, callback, try_to_connect=None)

MrMage

Thanks for this! Would it be possible to add a simple test, at least to ensure that this works for some state changes?

MrMage · 2018-03-27T07:42:52Z

Sources/CgRPC/shim/cgrpc.h

@@ -147,6 +147,11 @@ cgrpc_completion_queue *cgrpc_channel_completion_queue(cgrpc_channel *channel);

 grpc_connectivity_state cgrpc_channel_check_connectivity_state(
    cgrpc_channel *channel, int try_to_connect);
+void cgrpc_channel_watch_connectivity_state(cgrpc_channel *channel,
+                                            cgrpc_completion_queue * completion_queue,


Nit: Please remove the space after the asterisk.

MrMage · 2018-03-27T07:43:03Z

Sources/CgRPC/shim/channel.c

@@ -103,3 +103,8 @@ cgrpc_completion_queue *cgrpc_channel_completion_queue(cgrpc_channel *channel) {
 grpc_connectivity_state cgrpc_channel_check_connectivity_state(cgrpc_channel *channel, int try_to_connect) {
  return grpc_channel_check_connectivity_state(channel->channel, try_to_connect);
 }
+
+void cgrpc_channel_watch_connectivity_state(cgrpc_channel *channel, cgrpc_completion_queue * completion_queue, grpc_connectivity_state last_observed_state, double deadline, void *tag) {


Nit: Please remove the space after the asterisk.

MrMage · 2018-03-27T07:44:31Z

Sources/SwiftGRPC/Core/Channel.swift

-  }
+
+  /// Connectivity state observers
+  private var observers: [ConnectivityObserver] = []


Nit: rename this to connectivityObservers?

MrMage · 2018-03-27T07:44:58Z

Sources/SwiftGRPC/Core/Channel.swift

+  }
+
+  public func subscribe(sourceState: ConnectivityState, tryToConnect: Bool = false, callback: @escaping (ConnectivityState) -> ()) {
+    var observer = observers.first(where: { $0.state == sourceState })


I'd suggest using a trailing closure here.

MrMage · 2018-03-27T07:45:24Z

Sources/SwiftGRPC/Core/Channel.swift

+            defer { self.lastState = newState }
+
+            if self.lastState == self.state {
+              self.callbacks.forEach({ $0(newState) })


Nit: Please use a trailing closure here.

MrMage · 2018-03-27T07:57:22Z

Sources/SwiftGRPC/Core/Channel.swift

+    return ConnectivityState.connectivityState(cgrpc_channel_check_connectivity_state(underlyingChannel, tryToConnect ? 1 : 0))
+  }
+
+  public func subscribe(sourceState: ConnectivityState, tryToConnect: Bool = false, callback: @escaping (ConnectivityState) -> ()) {


I would suggest removing tryToConnect here — it doesn't work anyway if subscribe is called with tryToConnect = true while a subscriber with that sourceState already exists.

MrMage · 2018-03-27T08:01:32Z

Sources/SwiftGRPC/Core/Channel.swift

+    return ConnectivityState.connectivityState(cgrpc_channel_check_connectivity_state(underlyingChannel, tryToConnect ? 1 : 0))
+  }
+
+  public func subscribe(sourceState: ConnectivityState, tryToConnect: Bool = false, callback: @escaping (ConnectivityState) -> ()) {


Would you mind adding a comment to this method, especially to explain what sourceState is for? It looks to me like the callback is only fired if the channel's state changes to sourceState, is that correct? Why not subscribe to all state changes and have the caller exclude unwanted ones themselves in the callback (could be done with a simple if state != sourceState { return } in the callback)? That would also avoid spinning up a separate thread for each source state.

MrMage · 2018-03-27T08:09:00Z

Sources/SwiftGRPC/Core/Channel.swift

+          defer { self.tryToConnect = false }
+
+          let deadline: TimeInterval = 0.2
+          cgrpc_channel_watch_connectivity_state(self.underlyingChannel, self.underlyingCompletionQueue, underlyingState, deadline, nil)


Shouldn't underlyingState be the last observed state? At the moment this is only updated once, at the very beginning of the spinloop, so this will fire for any states that are different from the state when the loop was started, instead of simply firing every time we detect any change.

MrMage · 2018-03-27T08:09:17Z

Sources/SwiftGRPC/Core/Channel.swift

+    }
+
+    deinit {
+      queue.shutdown()


It looks to me like this will never get deallocated, as you are acquiring a strong reference to self right before the spinloop starts and never let go of it until the end of the spinloop. I would suggest shutting down all observers in Channel.deinit as well. Let me know if I'm missing something.

MrMage · 2018-03-27T08:09:58Z

Sources/SwiftGRPC/Core/Channel.swift

+}
+
+extension Channel {
+  public enum ConnectivityState {


Thanks! This is much better than what I had added before.

MrMage · 2018-03-27T09:07:38Z

Sources/SwiftGRPC/Core/Channel.swift

+    let underlyingChannel: UnsafeMutableRawPointer
+    let underlyingCompletionQueue: UnsafeMutableRawPointer
+    private(set) var tryToConnect: Bool
+    var callbacks: [(ConnectivityState) -> ()] = []


Access to callbacks definitely needs to be synchronized.

cvanderschuere · 2018-04-02T18:32:32Z

Update: haven't had a chance to get to these fixes. I plan on coming back to it by the end of the week.

MrMage · 2018-04-03T08:37:34Z

No worries! Most of these are just nits, anyway; but the possible thread leak and synchronization issues should definitely be investigated — maybe there actually no problems, but it's still worth checking.

timburks · 2018-04-04T18:32:29Z

@cvanderschuere Looks great! We'll merge after those requested changes. Thanks for this!

timburks · 2018-04-17T15:14:00Z

@cvanderschuere @MrMage Is there more coming in this PR?

If not, I'll merge it and we can address anything remaining in future PRs.

cvanderschuere · 2018-04-18T04:01:29Z

I don’t think this should be merged until the comments have been addressed.

I have not had chance get back to this, but do plan on it eventually. I would also not be opposed to someone taking over the PR and making the necessary changes.

SebastianThiebaud · 2018-05-21T22:06:44Z

I changed a few things. Let me know @MrMage if it's more inline with what you have in mind.

MrMage

Thank you for the changes! I have much more trust in this now, but there's one more big request/suggestion:

Please handle shutdown events in the spinloop inside run (I think there should be an example of this in CompletionQueue.swift). If you do that, we might get rid of the polling variable altogether (and instead just call observer.shutdown() from Channel.deinit), and in that case you can disregard my comments about serializing access to polling. If you decide to keep the polling variable, however, all accesses to it should be guarded by a lock (but the operation queue should be unnecessary).

What do you think?

MrMage · 2018-05-22T06:33:56Z

Sources/SwiftGRPC/Core/Channel.swift

    private var lastState: ConnectivityState
-
+    private let queue: OperationQueue


Please just use SwiftGRPC.Mutex to guard all accesses to polling (including the loop below and connectivityObservers.forEach { $0.polling = false } above), an operation queue is overkill here. (Can be disregarded if we eliminate polling, but please still remove the queue variable in that case.)

Got rid of polling.

MrMage · 2018-05-22T06:36:17Z

Sources/SwiftGRPC/Core/Channel.swift

    }

    deinit {
-      queue.shutdown()
+      shutdown()
    }

    private func run() {
      DispatchQueue.global().async { [weak self] in


Please use the thread-spawning pattern now used in e.g.

grpc-swift/Sources/SwiftGRPC/Core/CompletionQueue.swift

Line 118 in 8941a85

var threadLabel = "SwiftGRPC.CompletionQueue.runToCompletion.spinloopThread"

here now instead of dispatching to the global queue.

MrMage · 2018-05-22T06:36:44Z

Sources/SwiftGRPC/Core/Channel.swift

    }

    private func run() {
      DispatchQueue.global().async { [weak self] in
-        guard let `self` = self, let underlyingState = self.lastState.underlyingState else { return }
+        guard let `self` = self else { return }

        while self.polling {


Replace self.polling with (lock.synchronize { self.polling }). (Can be disregarded if we get rid of polling.)

Got rid of polling.

MrMage · 2018-05-22T06:41:08Z

Sources/SwiftGRPC/Core/Channel.swift

-            if self.lastState == self.state {
-              self.callbacks.forEach({ $0(newState) })
-            }
+            self.callback(newState)
          }


I'd suggest to handle queue shutdown at this point as well and exit the loop in that case. FYI, this loop retains self, which means that deinit (and thus shutdown) will never get called by itself — but shutdown will be indirectly called by Channel.deinit, so we should be fine here.

I might miss something here, but I don't think this loop retains self. It's been weakened.

MrMage · 2018-05-22T06:45:20Z

Sources/SwiftGRPC/Core/Channel.swift

-          if event.success == 1 || self.tryToConnect {
-            let newState = ConnectivityState.connectivityState(cgrpc_channel_check_connectivity_state(self.underlyingChannel, self.tryToConnect ? 1 : 0))
+          if event.success == 1 {
+            let newState = ConnectivityState.connectivityState(cgrpc_channel_check_connectivity_state(self.underlyingChannel, 1))


Are you sure that this should always try to connect? This would make the connectivity observer reconnect all the time, even though the actual channel is currently not used by the client code.

You're right, I missed that one.

MrMage · 2018-05-22T06:49:15Z

Sources/SwiftGRPC/Core/Channel.swift

-
-          defer { self.tryToConnect = false }
-
+          guard let underlyingState = self.lastState.underlyingState else { return }


Log an error in this case, as this would be very unexpected?

MrMage · 2018-05-22T06:51:17Z

Sources/SwiftGRPC/Core/Channel.swift

      self.underlyingCompletionQueue = cgrpc_completion_queue_create_for_next()
-      self.queue = CompletionQueue(underlyingCompletionQueue: self.underlyingCompletionQueue, name: "Connectivity State")
+      self.completionQueue = CompletionQueue(underlyingCompletionQueue: self.underlyingCompletionQueue, name: "Connectivity State")
+      self.callback = callback
      self.lastState = ConnectivityState.connectivityState(cgrpc_channel_check_connectivity_state(self.underlyingChannel, 0))


nit: I would add a currentState argument to init and use the result of that here. The initializer call in Channel would become

let observer = ConnectivityObserver(underlyingChannel: underlyingChannel, currentState: self.connectivityState(), callback: callback)

MrMage · 2018-05-22T06:54:32Z

Sources/SwiftGRPC/Core/Channel.swift

    var polling: Bool = false {
      didSet {
-        if polling == true && oldValue == false {
-          run()
+        queue.addOperation { [weak self] in


It looks to me like polling will only set to true once, right after initializing. So how about calling self.run() at the end of init(), and make shutdown available to Channel.deinit instead? In that case, we might even get by without a polling variable altogether if you add support for handling shutdown events in the spinloop.

Got rid of polling

MrMage · 2018-05-22T07:02:18Z

Sources/SwiftGRPC/Core/Channel.swift

    completionQueue.run() // start a loop that watches the channel's completion queue
  }

  deinit {
+    connectivityObservers.forEach { $0.polling = false }


Either use a lock for setting polling, or use $0.shutdown() (see the other comments).

Got rid of polling.

MrMage · 2018-05-22T07:03:44Z

Sources/SwiftGRPC/Core/Channel.swift

+  public func subscribe(callback: @escaping (ConnectivityState) -> ()) {
+    let observer = ConnectivityObserver(underlyingChannel: underlyingChannel, callback: callback)
+    observer.polling = true
+    connectivityObservers.append(observer)


In theory there could be race conditions here of subscribe were to be called from different threads, but I think we can ignore that possibility here.

SebastianThiebaud · 2018-05-22T18:18:56Z

You're right and I got rid of polling variable, which makes things way easier to handle and read.

MrMage

Looks much better, thank you! Just a few more smaller things.

MrMage · 2018-05-22T18:19:21Z

Sources/SwiftGRPC/Core/Channel.swift

@@ -73,7 +73,6 @@ public class Channel {
  }

  deinit {
-    connectivityObservers.forEach { $0.polling = false }


Please keep this line (see below).

FYI, I think this has not been resolved yet (1).

MrMage · 2018-05-22T18:19:36Z

Sources/SwiftGRPC/Core/Channel.swift

-    }
-
-    init(underlyingChannel: UnsafeMutableRawPointer, callback: @escaping (ConnectivityState) -> ()) {
+    init(underlyingChannel: UnsafeMutableRawPointer, currentState: ConnectivityState, callback: @escaping (ConnectivityState) -> ()) {


nit: (ConnectivityState) -> Void

MrMage · 2018-05-22T18:23:42Z

Sources/SwiftGRPC/Core/Channel.swift

+            break spinloop
+          case .queueTimeout:
+            continue spinloop
+          default:


print an error here as well and/or continue (see above)?

MrMage · 2018-05-22T18:24:15Z

Sources/SwiftGRPC/Core/Channel.swift

-          guard let underlyingState = self.lastState.underlyingState else { return }
+        spinloop: while true  {
+          guard let underlyingState = self.lastState.underlyingState else {
+            print("Couldn't retrieve `underlyingState`")


nit: I prefer to also include which class printed the error message. How about

print("ERROR: `Channel.ConnectivityObserver.lastState.underlyingState` nil. This should never happen!")

MrMage · 2018-05-22T18:25:25Z

Sources/SwiftGRPC/Core/Channel.swift

+          case .queueShutdown:
+            break spinloop
+          case .queueTimeout:
+            continue spinloop


Could have just break without spinloop here to exit the switch; up to you.

You might also get rid of the spinloop: loop label by replacing break spinloop above with return.

MrMage · 2018-05-22T18:32:08Z

Sources/SwiftGRPC/Core/Channel.swift

+
+            self.lastState = newState
+          case .queueShutdown:
+            break spinloop


Should we call the observer callback with a nil argument once we exit the loop, to indicate that no further calls are expected?

I'd say no. If I subscribe to get connectivity state changes, I'd assume I will get a callback only when there is a connectivity state change. If the queue gets shut down, that means, somehow the channel has been deallocated and at that point, I would not expect to get a connectivity state change, especially since it's most likely I'm the one initiating this event.

I also dislike introducing optionals where they're not really needed.

MrMage · 2018-05-22T18:32:32Z

Sources/SwiftGRPC/Core/Channel.swift

-      DispatchQueue.global().async { [weak self] in
+      let spinloopThreadQueue = DispatchQueue(label: "SwiftGRPC.ConnectivityObserver.run.spinloopThread")
+
+      spinloopThreadQueue.async { [weak self] in
        guard let `self` = self else { return }


This is the line that retains self by acquiring a strong reference to it. It does not make sense to try to get rid of it; instead avoid the [weak self] in guard let self = self altogether and keep the forEach { $0.shutdown() } line above. So I think at the moment your code still leaks threads.

To be honest, I'd like to have a test that spins up 100 observers to ensure all their spinloop threads get spun down once the channel closes (i.e. ensure no leaks are happening), but I can understand if you don't want to add one (should be fairly easy, though).

FYI, I think this has not been resolved yet (2).

MrMage · 2018-05-22T18:33:13Z

Sources/SwiftGRPC/Core/Channel.swift

-    let observer = ConnectivityObserver(underlyingChannel: underlyingChannel, callback: callback)
-    observer.polling = true
+  public func subscribe(callback: @escaping (ConnectivityState) -> Void) {
+    let observer = ConnectivityObserver(underlyingChannel: underlyingChannel, currentState: connectivityState(), callback: callback)


ultra-nit: observer could be inlined now.

MrMage · 2018-05-22T18:33:55Z

Sources/SwiftGRPC/Core/Channel.swift

          }
        }
      }
    }
-
-    private func shutdown() {


I would have kept this to call in Channel.shutdown (see above)

FYI, I think this has not been resolved yet (3).

MrMage · 2018-05-22T19:35:42Z

Sources/SwiftGRPC/Core/Channel.swift

          default:
-            break spinloop
+            print("Event's completion type is `unknown`")


Please also improve the formulation of this error as above.

To be honest, I'm not fan of having print statements all over the place. Especially in libraries. I'd rather not have these in here. I'm going to remove them and feel free to put them back in another PR if you really want them.

Fair enough. For what it's worth, I have a task in #189 to add a proper logging class. At that point, it should become much easier to properly detect and notice such errors (and possibly investigate why they happen).

SebastianThiebaud · 2018-05-25T03:16:03Z

@MrMage Just in case you missed the last commit.

MrMage · 2018-05-25T05:46:45Z

Sources/SwiftGRPC/Core/Channel.swift

@@ -124,41 +124,43 @@ private extension Channel {
    }

    deinit {
-      completionQueue.shutdown()
+      shutdown()
    }

    private func run() {
      let spinloopThreadQueue = DispatchQueue(label: "SwiftGRPC.ConnectivityObserver.run.spinloopThread")

      spinloopThreadQueue.async { [weak self] in


No need to use [weak self] here, as mentioned before. Shutdowns are initiated by Channel.

MrMage

LGTM, but please remove [weak self] in the block, as it doesn't add value.

Thanks for all the changes!

SebastianThiebaud · 2018-05-25T20:47:05Z

Done :)

timburks

Thanks for adding this! This looks good to me. For reference, I note that https://github.com/grpc/grpc/blob/master/doc/connectivity-semantics-and-api.md says that "Since state changes can be rapid and race with any such notification, the notification should just inform the user that some state change has happened, leaving it to the user to poll the channel for the current state." The Swift callback passes state, but that state is from an immediate call to grpc_channel_check_connectivity_state, so it seems as current as state that might be obtained from a call made within the callback.

cvanderschuere and others added 2 commits March 26, 2018 16:12

Implement connectivity state property and observers (#8)

0bd14d8

Explicitly import dispatch

76faf09

MrMage requested changes Mar 27, 2018

View reviewed changes

Sebastian Thiebaud and others added 2 commits May 21, 2018 14:12

PR changes

255ce88

Merge branch 'master' into connectivity_state

4e94ef4

MrMage requested changes May 22, 2018

View reviewed changes

PR changes

0891482

Add comments

e853032

MrMage requested changes May 22, 2018

View reviewed changes

PR changes

4ca615a

MrMage reviewed May 22, 2018

View reviewed changes

Remove print statements

ad6e022

MrMage reviewed May 25, 2018

View reviewed changes

MrMage approved these changes May 25, 2018

View reviewed changes

PR changes

0d25ef8

timburks approved these changes May 26, 2018

View reviewed changes

timburks merged commit b1bba67 into grpc:master May 26, 2018

SebastianThiebaud deleted the connectivity_state branch May 27, 2018 21:40

SebastianThiebaud mentioned this pull request May 27, 2018

Thoughts on adding support for connectivity state #186

Closed

		private var lastState: ConnectivityState

		private let queue: OperationQueue


		defer { self.tryToConnect = false }

		guard let underlyingState = self.lastState.underlyingState else { return }

Implement connectivity state property and observers #196

Implement connectivity state property and observers #196

Conversation

cvanderschuere commented Mar 26, 2018 • edited Loading

MrMage left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cvanderschuere commented Apr 2, 2018

MrMage commented Apr 3, 2018

timburks commented Apr 4, 2018

timburks commented Apr 17, 2018

cvanderschuere commented Apr 18, 2018 • edited Loading

SebastianThiebaud commented May 21, 2018

MrMage left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SebastianThiebaud commented May 22, 2018

MrMage left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SebastianThiebaud commented May 25, 2018

Choose a reason for hiding this comment

MrMage left a comment

Choose a reason for hiding this comment

SebastianThiebaud commented May 25, 2018

timburks left a comment

Choose a reason for hiding this comment

cvanderschuere commented Mar 26, 2018 •

edited

Loading

cvanderschuere commented Apr 18, 2018 •

edited

Loading