#45 - Two sided performance test for merge (#114)

This PR extends `measureSequenceThroughput` to take two outputs so functions like merge/ zip etc can be tested by both sides.

Author: kingreza

Tests/AsyncAlgorithmsTests/Performance/TestThroughput.swift

@@ -15,13 +15,13 @@ import AsyncAlgorithms
 #if canImport(Darwin)
 final class TestThroughput: XCTestCase {
   func test_chain2() async {
-    await measureSequenceThroughput(output: 1) {
-      chain($0, [].async)
+    await measureSequenceThroughput(firstOutput: 1, secondOutput: 2) {
+      chain($0, $1)
     }
   }
   func test_chain3() async {
-    await measureSequenceThroughput(output: 1) {
-      chain($0, [].async, [].async)
+    await measureSequenceThroughput(firstOutput: 1, secondOutput: 2, thirdOutput: 3) {
+      chain($0, $1, $2)
     }
   }
   func test_compacted() async {
@@ -40,13 +40,13 @@ final class TestThroughput: XCTestCase {
     }
   }
   func test_merge2() async {
-    await measureSequenceThroughput(output: 1) {
-      merge($0, (0..<10).async)
+    await measureSequenceThroughput(firstOutput: 1, secondOutput: 2) {
+      merge($0, $1)
     }
   }
   func test_merge3() async {
-    await measureSequenceThroughput(output: 1) {
-      merge($0, (0..<10).async, (0..<10).async)
+    await measureSequenceThroughput(firstOutput: 1, secondOutput: 2, thirdOutput: 3) {
+      merge($0, $1, $2)
     }
   }
   func test_removeDuplicates() async {
@@ -55,13 +55,13 @@ final class TestThroughput: XCTestCase {
     }
   }
   func test_zip2() async {
-    await measureSequenceThroughput(output: 1) {
-      zip($0, Indefinite(value: 2).async)
+    await measureSequenceThroughput(firstOutput: 1, secondOutput: 2) {
+      zip($0, $1)
     }
   }
   func test_zip3() async {
-    await measureSequenceThroughput(output: 1) {
-      zip($0, Indefinite(value: 2).async, Indefinite(value: 3).async)
+    await measureSequenceThroughput(firstOutput: 1, secondOutput: 2, thirdOutput: 3) {
+      zip($0, $1, $2)
     }
   }
 }

Tests/AsyncAlgorithmsTests/Performance/ThroughputMeasurement.swift

@@ -87,6 +87,57 @@ extension XCTestCase {
     }
   }
 
+  public func measureSequenceThroughput<S: AsyncSequence, Output>(firstOutput: @autoclosure () -> Output, secondOutput: @autoclosure () -> Output, _ sequenceBuilder: (InfiniteAsyncSequence<Output>, InfiniteAsyncSequence<Output>) -> S) async where S: Sendable {
+    let metric = _ThroughputMetric()
+    let sampleTime: Double = 0.1
+    
+    measure(metrics: [metric]) {
+      let firstInfSeq = InfiniteAsyncSequence(value: firstOutput(), duration: sampleTime)
+      let secondInfSeq = InfiniteAsyncSequence(value: secondOutput(), duration: sampleTime)
+      let seq = sequenceBuilder(firstInfSeq, secondInfSeq)
+      
+      let exp = self.expectation(description: "Finished")
+      let iterTask = Task<Int, Error> {
+        var eventCount = 0
+        for try await _ in seq {
+          eventCount += 1
+        }
+        metric.eventCount = eventCount
+        exp.fulfill()
+        return eventCount
+      }
+      usleep(UInt32(sampleTime * Double(USEC_PER_SEC)))
+      iterTask.cancel()
+      self.wait(for: [exp], timeout: sampleTime * 2)
+    }
+  }
+    
+  public func measureSequenceThroughput<S: AsyncSequence, Output>(firstOutput: @autoclosure () -> Output, secondOutput: @autoclosure () -> Output, thirdOutput: @autoclosure () -> Output, _ sequenceBuilder: (InfiniteAsyncSequence<Output>, InfiniteAsyncSequence<Output>, InfiniteAsyncSequence<Output>) -> S) async where S: Sendable {
+    let metric = _ThroughputMetric()
+    let sampleTime: Double = 0.1
+    
+    measure(metrics: [metric]) {
+      let firstInfSeq = InfiniteAsyncSequence(value: firstOutput(), duration: sampleTime)
+      let secondInfSeq = InfiniteAsyncSequence(value: secondOutput(), duration: sampleTime)
+      let thirdInfSeq = InfiniteAsyncSequence(value: thirdOutput(), duration: sampleTime)
+      let seq = sequenceBuilder(firstInfSeq, secondInfSeq, thirdInfSeq)
+      
+      let exp = self.expectation(description: "Finished")
+      let iterTask = Task<Int, Error> {
+        var eventCount = 0
+        for try await _ in seq {
+          eventCount += 1
+        }
+        metric.eventCount = eventCount
+        exp.fulfill()
+        return eventCount
+      }
+      usleep(UInt32(sampleTime * Double(USEC_PER_SEC)))
+      iterTask.cancel()
+      self.wait(for: [exp], timeout: sampleTime * 2)
+    }
+}
+  
   public func measureSequenceThroughput<S: AsyncSequence, Source: AsyncSequence>( source: Source, _ sequenceBuilder: (Source) -> S) async where S: Sendable, Source: Sendable {
     let metric = _ThroughputMetric()
     let sampleTime: Double = 0.1