Add an internal pausable writer #1245

glbrntt · 2021-08-20T14:24:24Z

Motivation:

Async RPCs which stream outbound messages (i.e. requests on the client,
responses on the server) will be provided with a means to asynchronously
send messages. This object should suspend if the underlying channel is
not currently writable.

Modifications:

Add an 'AsyncWriter' as the underlying type for these objects which
suspends if the writer is 'paused' and resumes when the writer is
resumed.

Result:

We have a writer capable of suspending writes.

glbrntt · 2021-08-20T14:27:18Z

cc @simonjbeaumont

Sources/GRPC/Concurrency/AsyncWriter.swift

fabianfett · 2021-08-20T15:22:47Z

Sources/GRPC/Concurrency/AsyncWriter.swift

+@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
+fileprivate final class AsyncWriterStorage<
+  Element,
+  End,
+  Delegate: AsyncWriterDelegate
+> where Delegate.Element == Element, Delegate.End == End {
+  /// A value pending a write.


Not a hill I will die on, but if we put the Storage into the AsyncWriter, we don't need to redefine the types here, since they would be inherited from their enclosing struct.

It'd rather have this than the extra level of indentation tbh!

Okay, In this case we should still remove the Element and End from the generic parameters. They are implicit in the Delegate.

Agh, yes, missed this one.

FWIW I think this change wouldn't be great for public APIs since Xcode suggests the "full" type name rather than any type aliases, e.g.: "AsyncWriter<DelegateImplType<String, Int>>.Element" rather than "Element".

Sources/GRPC/Concurrency/AsyncWriter.swift

fabianfett · 2021-08-23T09:12:22Z

Sources/GRPC/Concurrency/AsyncWriter.swift

+}
+
+@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
+fileprivate final class AsyncWriterStorage<


Did you consider making this an actor?

No -- pause and resume won't be called from asynchronous contexts.

Sources/GRPC/Concurrency/AsyncWriter.swift

fabianfett · 2021-08-24T11:26:15Z

Sources/GRPC/Concurrency/AsyncWriter.swift

+  }
+
+  /// A lock which must be help when accessing or modifying mutable state held by this storage.
+  private var lock: Lock


Why is this a var?

fabianfett · 2021-08-24T11:30:14Z

Sources/GRPC/Concurrency/AsyncWriter.swift

+      case let .element(pendingElement):
+        self.delegate.write(pendingElement.value)
+        pendingElement.continuation.resume()


I'm not sure about this. Let's consider we have a buffer full with 16 writes waiting. After we have written the first element, do we want to allow more writes from the users side? There are still 15 writes pending...

Further this is in a non detached loop. This means we will def. write all 16 items, and we will signal to everyone waiting, you can write again...

If this was an actor, we could consider using await Task.yield to check after a number of writes if the target is still writable and only in those cases to start succeed the continuations.

fabianfett · 2021-08-24T11:45:53Z

Sources/GRPC/Concurrency/Result+AsyncAwait.swift

+#if compiler(>=5.5)
+
+extension Result where Failure == Error {
+  @available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
+  internal init(catching operation: @escaping () async throws -> Success) async {
+    do {
+      let success = try await operation()
+      self = .success(success)
+    } catch {
+      self = .failure(error)
+    }
+  }
+}
+
+#endif // compiler(>=5.5)


Is this used anywhere? CMD+F doesn't show me a use.

fabianfett · 2021-08-24T11:48:12Z

Tests/GRPCTests/Concurrency/AsyncWriterTests.swift

+      writer.resume()
+      try await written1
+      try await written2
+      XCTAssertEqual(delegate.elements.sorted(), ["wunch", "meat"].sorted())


Why do we have sorted() here? I would expect that we check that the elements arrive in order here.

fabianfett · 2021-08-24T11:49:12Z

Tests/GRPCTests/Concurrency/AsyncWriterTests.swift

+      writer.pause()
+
+      async let w1: Void = writer.write("hitchcock")
+      await Task.yield()


why do we have the Task.yield() here?

fabianfett · 2021-08-24T11:50:03Z

Tests/GRPCTests/Concurrency/AsyncWriterTests.swift

+      }
+
+      writer.resume()
+      try await writer.finish(0)


Maybe check delegate.element.isEmpty

fabianfett · 2021-08-24T11:50:27Z

Tests/GRPCTests/Concurrency/AsyncWriterTests.swift

+        try await writer.write("cheddar")
+      } verify: { error in
+        XCTAssertEqual(error as? AsyncWriterError, .alreadyFinished)
+      }


Maybe check that write was not forwarded to delegate

fabianfett · 2021-08-24T11:51:26Z

Tests/GRPCTests/Concurrency/XCTAwait.swift

+
+@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
+internal func XCTAssertThrowsError<T>(
+  _ expression: @escaping () async throws -> T,


just out of interest for AHC, can we make this an @autoclosure?

It wasn't supported when I implemented this but seems to work now!

fabianfett · 2021-08-27T10:53:14Z

Sources/GRPC/AsyncAwaitSupport/AsyncWriter.swift

+
+      // `writes` will never exceed `maxWritesBeforeYield` so unchecked arithmetic is okay here.
+      writes &+= 1
+      if writes == self.maxWritesBeforeYield {


I think, we also need to check, if the queue is empty? You want to reset your counter to 0 once you have written all elements.

writes is local to this function so doesn't need to be reset?

Although I suppose if writes == self.maxWritesBeforeYield and self.pendingElements.isEmpty then we'll yield even though we're about to exit the function anyway. Maybe that doesn't matter so much though.

Sources/GRPC/AsyncAwaitSupport/AsyncWriter.swift

fabianfett · 2021-08-27T10:58:40Z

Tests/GRPCTests/AsyncAwaitSupport/AsyncWriterTests.swift

+      // Still 0.
+      XCTAssertEqual(delegate.end, 0)
+    }
+  }


needs

testCallToFinishWhilePending()

testTooManyCallsToFinishWhilePending()

Motivation: Async RPCs which stream outbound messages (i.e. requests on the client, responses on the server) will be provided with a means to asynchronously send messages. This object should suspend if the underlying channel is not currently writable. Modifications: Add an 'AsyncWriter' as the underlying type for these objects which suspends if the writer is 'paused' and resumes when the writer is resumed. Result: We have a writer capable of suspending writes.

fabianfett

Great! Really like how it turned out! Thanks!

Motivation: Async RPCs which stream outbound messages (i.e. requests on the client, responses on the server) will be provided with a means to asynchronously send messages. This object should suspend if the underlying channel is not currently writable. Modifications: Add an 'AsyncWriter' as the underlying type for these objects which suspends if the writer is 'paused' and resumes when the writer is resumed. Result: We have a writer capable of suspending writes.

This commit implements some of the types required by the proposal for async/await support, added in #1231. To aid reviewing, only the types required for the server are included. They have been pulled in from the proof-of-concept implementation linked from the proposal PR. It is a complimentary PR to #1243 ("Async-await: Base types for client implementation"). It provides a unified `AsyncServerHandler` for all of the RPC types which avoids a substantial amount of code duplication that is found in the existing handlers. Wrappers are provided for the four RPC types. Otherwise it is analogous to the existing `BidirectionalStreamingServerHandler`. It's worth calling out that this PR makes use of some placeholder types which are not intended to be final. Specifically: * `AsyncResponseStreamWriter` is expected to be superseded by the `AsyncWriter` from #1245. * `AsyncServerCallContext` conformance has been added to the existing `ServerCallContextBase`. It is expected that we will provide a new implementation of `AsyncServerCallContext` that is independent from the existing call context types.

Motivation: Async RPCs which stream outbound messages (i.e. requests on the client, responses on the server) will be provided with a means to asynchronously send messages. This object should suspend if the underlying channel is not currently writable. Modifications: Add an 'AsyncWriter' as the underlying type for these objects which suspends if the writer is 'paused' and resumes when the writer is resumed. Result: We have a writer capable of suspending writes.

This commit implements some of the types required by the proposal for async/await support, added in grpc#1231. To aid reviewing, only the types required for the server are included. They have been pulled in from the proof-of-concept implementation linked from the proposal PR. It is a complimentary PR to grpc#1243 ("Async-await: Base types for client implementation"). It provides a unified `AsyncServerHandler` for all of the RPC types which avoids a substantial amount of code duplication that is found in the existing handlers. Wrappers are provided for the four RPC types. Otherwise it is analogous to the existing `BidirectionalStreamingServerHandler`. It's worth calling out that this PR makes use of some placeholder types which are not intended to be final. Specifically: * `AsyncResponseStreamWriter` is expected to be superseded by the `AsyncWriter` from grpc#1245. * `AsyncServerCallContext` conformance has been added to the existing `ServerCallContextBase`. It is expected that we will provide a new implementation of `AsyncServerCallContext` that is independent from the existing call context types.

Motivation: Async RPCs which stream outbound messages (i.e. requests on the client, responses on the server) will be provided with a means to asynchronously send messages. This object should suspend if the underlying channel is not currently writable. Modifications: Add an 'AsyncWriter' as the underlying type for these objects which suspends if the writer is 'paused' and resumes when the writer is resumed. Result: We have a writer capable of suspending writes.

This commit implements some of the types required by the proposal for async/await support, added in grpc#1231. To aid reviewing, only the types required for the server are included. They have been pulled in from the proof-of-concept implementation linked from the proposal PR. It is a complimentary PR to grpc#1243 ("Async-await: Base types for client implementation"). It provides a unified `AsyncServerHandler` for all of the RPC types which avoids a substantial amount of code duplication that is found in the existing handlers. Wrappers are provided for the four RPC types. Otherwise it is analogous to the existing `BidirectionalStreamingServerHandler`. It's worth calling out that this PR makes use of some placeholder types which are not intended to be final. Specifically: * `AsyncResponseStreamWriter` is expected to be superseded by the `AsyncWriter` from grpc#1245. * `AsyncServerCallContext` conformance has been added to the existing `ServerCallContextBase`. It is expected that we will provide a new implementation of `AsyncServerCallContext` that is independent from the existing call context types.

Motivation: Async RPCs which stream outbound messages (i.e. requests on the client, responses on the server) will be provided with a means to asynchronously send messages. This object should suspend if the underlying channel is not currently writable. Modifications: Add an 'AsyncWriter' as the underlying type for these objects which suspends if the writer is 'paused' and resumes when the writer is resumed. Result: We have a writer capable of suspending writes.

This commit implements some of the types required by the proposal for async/await support, added in #1231. To aid reviewing, only the types required for the server are included. They have been pulled in from the proof-of-concept implementation linked from the proposal PR. It is a complimentary PR to #1243 ("Async-await: Base types for client implementation"). It provides a unified `AsyncServerHandler` for all of the RPC types which avoids a substantial amount of code duplication that is found in the existing handlers. Wrappers are provided for the four RPC types. Otherwise it is analogous to the existing `BidirectionalStreamingServerHandler`. It's worth calling out that this PR makes use of some placeholder types which are not intended to be final. Specifically: * `AsyncResponseStreamWriter` is expected to be superseded by the `AsyncWriter` from #1245. * `AsyncServerCallContext` conformance has been added to the existing `ServerCallContextBase`. It is expected that we will provide a new implementation of `AsyncServerCallContext` that is independent from the existing call context types.

Motivation: Async RPCs which stream outbound messages (i.e. requests on the client, responses on the server) will be provided with a means to asynchronously send messages. This object should suspend if the underlying channel is not currently writable. Modifications: Add an 'AsyncWriter' as the underlying type for these objects which suspends if the writer is 'paused' and resumes when the writer is resumed. Result: We have a writer capable of suspending writes.

This commit implements some of the types required by the proposal for async/await support, added in #1231. To aid reviewing, only the types required for the server are included. They have been pulled in from the proof-of-concept implementation linked from the proposal PR. It is a complimentary PR to #1243 ("Async-await: Base types for client implementation"). It provides a unified `AsyncServerHandler` for all of the RPC types which avoids a substantial amount of code duplication that is found in the existing handlers. Wrappers are provided for the four RPC types. Otherwise it is analogous to the existing `BidirectionalStreamingServerHandler`. It's worth calling out that this PR makes use of some placeholder types which are not intended to be final. Specifically: * `AsyncResponseStreamWriter` is expected to be superseded by the `AsyncWriter` from #1245. * `AsyncServerCallContext` conformance has been added to the existing `ServerCallContextBase`. It is expected that we will provide a new implementation of `AsyncServerCallContext` that is independent from the existing call context types.

glbrntt added the 🔨 semver/patch No public API change. label Aug 20, 2021

glbrntt requested a review from fabianfett August 20, 2021 14:27

fabianfett reviewed Aug 20, 2021

View reviewed changes

Sources/GRPC/Concurrency/AsyncWriter.swift Outdated Show resolved Hide resolved

fabianfett reviewed Aug 20, 2021

View reviewed changes

Sources/GRPC/Concurrency/AsyncWriter.swift Outdated Show resolved Hide resolved

fabianfett reviewed Aug 23, 2021

View reviewed changes

Sources/GRPC/Concurrency/AsyncWriter.swift Outdated Show resolved Hide resolved

glbrntt requested a review from fabianfett August 24, 2021 10:46

fabianfett reviewed Aug 24, 2021

View reviewed changes

simonjbeaumont mentioned this pull request Aug 25, 2021

[async-await] Base types for server implementation #1249

Merged

glbrntt requested a review from fabianfett August 26, 2021 14:11

glbrntt mentioned this pull request Aug 27, 2021

Add an async throwing source #1252

Merged

fabianfett reviewed Aug 27, 2021

View reviewed changes

Sources/GRPC/AsyncAwaitSupport/AsyncWriter.swift Show resolved Hide resolved

fabianfett reviewed Aug 27, 2021

View reviewed changes

Sources/GRPC/AsyncAwaitSupport/AsyncWriter.swift Show resolved Hide resolved

fabianfett reviewed Aug 27, 2021

View reviewed changes

glbrntt force-pushed the gb-async-writer branch from 0aaff3b to 7ad0403 Compare September 1, 2021 16:04

glbrntt requested a review from fabianfett September 1, 2021 16:05

glbrntt added 2 commits September 2, 2021 15:30

Add "cancellation" to drop all pending writes

85aa1d9

Remove unnecessary dependency

4813564

fabianfett approved these changes Sep 2, 2021

View reviewed changes

glbrntt merged commit bae3349 into grpc:async-await Sep 2, 2021

glbrntt deleted the gb-async-writer branch September 2, 2021 15:45

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add an internal pausable writer #1245

Add an internal pausable writer #1245

glbrntt commented Aug 20, 2021

glbrntt commented Aug 20, 2021

fabianfett Aug 20, 2021

glbrntt Aug 20, 2021

fabianfett Aug 23, 2021

glbrntt Aug 24, 2021

glbrntt Aug 24, 2021

fabianfett Aug 23, 2021

glbrntt Aug 24, 2021

fabianfett Aug 24, 2021

fabianfett Aug 24, 2021

fabianfett Aug 24, 2021

fabianfett Aug 24, 2021

fabianfett Aug 24, 2021

fabianfett Aug 24, 2021

fabianfett Aug 24, 2021

fabianfett Aug 24, 2021

glbrntt Aug 26, 2021

fabianfett Aug 27, 2021

glbrntt Aug 27, 2021 •

edited

Loading

fabianfett Aug 27, 2021

fabianfett left a comment

Add an internal pausable writer #1245

Add an internal pausable writer #1245

Conversation

glbrntt commented Aug 20, 2021

glbrntt commented Aug 20, 2021

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

glbrntt Aug 27, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fabianfett left a comment

Choose a reason for hiding this comment

glbrntt Aug 27, 2021 •

edited

Loading