Implementations for operators and async functions on AsyncSequence per SE-0298 (#35740)

* Implementations for operators and async functions on AsyncSequence per SE-0298

* Add trailing new lines to AsyncSequence adopter files

* Remove extra // from preambles

* Transition sequence properties and sequence initializers to internal

* Rename Upstream to Base

* Rename "Failable" asynchronous sequences to "Throwing"

* Make iteration consistent with closure lifetimes and track failure via state

* Use for await syntax for async function extensions on AsyncSequence

* Work-around rethrows calculations being incorrect for conformance + closure rethrowing

* Update testing names to reflect Throwing versus Failable

* Incorperate runAsyncAndBlock into the test function for suite calls

* Remove @Frozen from AsyncSequence operators

* Back out reduce family functions for now due to IRGen bug

* Additional corrections for throwing cases and fix up potential linux crasher

* Add a upstream and throwing tests and correct flatMap's throwing behavior

* Unify parameter names to count

* Add a more helpful precondition on prefix

* Add tests for throwing cases of non closure async functions on AsyncSequence

* Correct missing count transition in dropFirst

* Correct missing , in assertion

* [NFC] add commentary on optimization in dropFirst on dropFirst

* Disable collect tests for non macOS platforms for now

* Disable all async sequence tests for non macOS platforms for now

Author: phausler

stdlib/public/Concurrency/AsyncCompactMapSequence.swift

@@ -0,0 +1,81 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift.org open source project
+//
+// Copyright (c) 2021 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+// See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
+//
+//===----------------------------------------------------------------------===//
+
+import Swift
+
+extension AsyncSequence {
+  @inlinable
+  public __consuming func compactMap<ElementOfResult>(
+    _ transform: @escaping (Element) async -> ElementOfResult?
+  ) -> AsyncCompactMapSequence<Self, ElementOfResult> {
+    return AsyncCompactMapSequence(self, transform: transform)
+  }
+}
+
+public struct AsyncCompactMapSequence<Base: AsyncSequence, ElementOfResult> {
+  @usableFromInline
+  let base: Base
+
+  @usableFromInline
+  let transform: (Base.Element) async -> ElementOfResult?
+
+  @usableFromInline
+  init(
+    _ base: Base, 
+    transform: @escaping (Base.Element) async -> ElementOfResult?
+  ) {
+    self.base = base
+    self.transform = transform
+  }
+}
+
+extension AsyncCompactMapSequence: AsyncSequence {
+  public typealias Element = ElementOfResult
+  public typealias AsyncIterator = Iterator
+
+  public struct Iterator: AsyncIteratorProtocol {
+    public typealias Element = ElementOfResult
+
+    @usableFromInline
+    var baseIterator: Base.AsyncIterator
+
+    @usableFromInline
+    let transform: (Base.Element) async -> ElementOfResult?
+
+    @usableFromInline
+    init(
+      _ baseIterator: Base.AsyncIterator, 
+      transform: @escaping (Base.Element) async -> ElementOfResult?
+    ) {
+      self.baseIterator = baseIterator
+      self.transform = transform
+    }
+
+    @inlinable
+    public mutating func next() async rethrows -> ElementOfResult? {
+      while true {
+        guard let element = try await baseIterator.next() else {
+          return nil
+        }
+
+        if let transformed = await transform(element) {
+          return transformed
+        }
+      }
+    }
+  }
+
+  @inlinable
+  public __consuming func makeAsyncIterator() -> Iterator {
+    return Iterator(base.makeAsyncIterator(), transform: transform)
+  }
+}

stdlib/public/Concurrency/AsyncDropFirstSequence.swift

@@ -0,0 +1,90 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift.org open source project
+//
+// Copyright (c) 2021 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+// See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
+//
+//===----------------------------------------------------------------------===//
+
+import Swift
+
+extension AsyncSequence {
+  @inlinable
+  public __consuming func dropFirst(
+    _ count: Int = 1
+  ) -> AsyncDropFirstSequence<Self> {
+    precondition(count >= 0, 
+      "Can't drop a negative number of elements from an async sequence")
+    return AsyncDropFirstSequence(self, dropping: count)
+  }
+}
+
+public struct AsyncDropFirstSequence<Base: AsyncSequence> {
+  @usableFromInline
+  let base: Base
+
+  @usableFromInline
+  let count: Int
+  
+  @usableFromInline 
+  init(_ base: Base, dropping count: Int) {
+    self.base = base
+    self.count = count
+  }
+}
+
+extension AsyncDropFirstSequence: AsyncSequence {
+  public typealias Element = Base.Element
+  public typealias AsyncIterator = Iterator
+
+  public struct Iterator: AsyncIteratorProtocol {
+    @usableFromInline
+    var baseIterator: Base.AsyncIterator
+    
+    @usableFromInline
+    var count: Int
+
+    @usableFromInline
+    init(_ baseIterator: Base.AsyncIterator, count: Int) {
+      self.baseIterator = baseIterator
+      self.count = count
+    }
+
+    @inlinable
+    public mutating func next() async rethrows -> Base.Element? {
+      var remainingToDrop = count
+      while remainingToDrop > 0 {
+        guard try await baseIterator.next() != nil else {
+          count = 0
+          return nil
+        }
+        remainingToDrop -= 1
+      }
+      count = 0
+      return try await baseIterator.next()
+    }
+  }
+
+  @inlinable
+  public __consuming func makeAsyncIterator() -> Iterator {
+    return Iterator(base.makeAsyncIterator(), count: count)
+  }
+}
+
+extension AsyncDropFirstSequence {
+  @inlinable
+  public __consuming func dropFirst(
+    _ count: Int = 1
+  ) -> AsyncDropFirstSequence<Base> {
+    // If this is already a AsyncDropFirstSequence, we can just sum the current 
+    // drop count and additional drop count.
+    precondition(count >= 0, 
+      "Can't drop a negative number of elements from an async sequence")
+    return AsyncDropFirstSequence(base, dropping: self.count + count)
+  }
+}
+

stdlib/public/Concurrency/AsyncDropWhileSequence.swift

@@ -0,0 +1,80 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift.org open source project
+//
+// Copyright (c) 2021 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+// See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
+//
+//===----------------------------------------------------------------------===//
+
+import Swift
+
+extension AsyncSequence {
+  @inlinable
+  public __consuming func drop(
+    while predicate: @escaping (Element) async -> Bool
+  ) -> AsyncDropWhileSequence<Self> {
+    AsyncDropWhileSequence(self, predicate: predicate)
+  }
+}
+
+public struct AsyncDropWhileSequence<Base: AsyncSequence> {
+  @usableFromInline
+  let base: Base
+
+  @usableFromInline
+  let predicate: (Base.Element) async -> Bool
+
+  @usableFromInline
+  init(
+    _ base: Base, 
+    predicate: @escaping (Base.Element) async -> Bool
+  ) {
+    self.base = base
+    self.predicate = predicate
+  }
+}
+
+extension AsyncDropWhileSequence: AsyncSequence {
+  public typealias Element = Base.Element
+  public typealias AsyncIterator = Iterator
+
+  public struct Iterator: AsyncIteratorProtocol {
+    @usableFromInline
+    var baseIterator: Base.AsyncIterator
+
+    @usableFromInline
+    var predicate: ((Base.Element) async -> Bool)?
+
+    @usableFromInline
+    init(
+      _ baseIterator: Base.AsyncIterator, 
+      predicate: @escaping (Base.Element) async -> Bool
+    ) {
+      self.baseIterator = baseIterator
+      self.predicate = predicate
+    }
+
+    @inlinable
+    public mutating func next() async rethrows -> Base.Element? {
+      while let predicate = self.predicate {
+        guard let element = try await baseIterator.next() else {
+          return nil
+        }
+        if await predicate(element) == false {
+          self.predicate = nil
+          return element
+        }
+      }
+      return try await baseIterator.next()
+    }
+  }
+
+  @inlinable
+  public __consuming func makeAsyncIterator() -> Iterator {
+    return Iterator(base.makeAsyncIterator(), predicate: predicate)
+  }
+}

stdlib/public/Concurrency/AsyncFilterSequence.swift

@@ -0,0 +1,78 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift.org open source project
+//
+// Copyright (c) 2021 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+// See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
+//
+//===----------------------------------------------------------------------===//
+
+import Swift
+
+extension AsyncSequence {
+  @inlinable
+  public __consuming func filter(
+    _ isIncluded: @escaping (Element) async -> Bool
+  ) -> AsyncFilterSequence<Self> {
+    return AsyncFilterSequence(self, isIncluded: isIncluded)
+  }
+}
+
+public struct AsyncFilterSequence<Base: AsyncSequence> {
+  @usableFromInline
+  let base: Base
+
+  @usableFromInline
+  let isIncluded: (Element) async -> Bool
+
+  @usableFromInline
+  init(
+    _ base: Base, 
+    isIncluded: @escaping (Base.Element) async -> Bool
+  ) {
+    self.base = base
+    self.isIncluded = isIncluded
+  }
+}
+
+extension AsyncFilterSequence: AsyncSequence {
+  public typealias Element = Base.Element
+  public typealias AsyncIterator = Iterator
+
+  public struct Iterator: AsyncIteratorProtocol {
+    @usableFromInline
+    var baseIterator: Base.AsyncIterator
+
+    @usableFromInline
+    let isIncluded: (Base.Element) async -> Bool
+
+    @usableFromInline
+    init(
+      _ baseIterator: Base.AsyncIterator,
+      isIncluded: @escaping (Base.Element) async -> Bool
+    ) {
+      self.baseIterator = baseIterator
+      self.isIncluded = isIncluded
+    }
+
+    @inlinable
+    public mutating func next() async rethrows -> Base.Element? {
+      while true {
+        guard let element = try await baseIterator.next() else {
+          return nil
+        }
+        if await isIncluded(element) {
+          return element
+        }
+      }
+    }
+  }
+
+  @inlinable
+  public __consuming func makeAsyncIterator() -> Iterator {
+    return Iterator(base.makeAsyncIterator(), isIncluded: isIncluded)
+  }
+}