[Channel] re-implement using a state machine

[twittemb]

Hi

This PR suggests a new implementation for AsyncChannel. The idea was to:

• streamline the implementation using the same paradigm (state machine/storage) as the recently remade operators (merge, zip, ...)
• share some code between AsyncChannel and AsyncThrowingChannel since the state machine is the same
• eventually improve the performance (I've seen a 20% increase in throughput)

Some remarks:

• I've added a throughput measurement
• I think there is a bug in the old/current implementation of AsyncChannel because when I tried to run it with the new throughput measurement I had a crash from time to time
• with this new implementation we could imagine to split the producer from the consumer (same as what @FranzBusch pitched here https://forums.swift.org/t/pitch-convenience-async-throwing-stream-makestream-methods/61030 for AsyncStream). We just have to reference the same storage from an hypothetical AsyncChannel.Continuation and the AsyncChannel it self.

@phausler @FranzBusch what do you think ?

[phausler]

The perf sounds fantastic! I will make sure to go over this with a fine tooth comb Monday.

[twittemb]

For the record, I've used ad-hoc properties (hidden behind #if DEBUG instructions) so we can test the suspend/resume mechanism more precisely, particularly for task cancellation. I know this is not an ideal solution but I can't see anything better right now.

I guess this is a part of the resolution of #148

[FranzBusch]

Currently on vacation but will try to give this a look soon! I think @phausler has some ideas around testing that doesn’t involve scattering debug only code.

[eaigner]

Just FYI, your bug is probably that you increment generation += 1. In swift this does not automatically overflow, but will crash if you reach Int.max.

[phausler]

Cancellation complicates things a decent amount; accounting for all the states and their cancellation counterparts ends up being roughly similar. The advantage with state machines is that the states are well known about which transition it can be and explicitly calls those out - so for subtile state manipulation that can cross boundaries of tasks it is useful to have a bit more # of lines of code (without much perf impact) to build known state transition clarity.

[FranzBusch]

Agree with @phausler here LoC is not a good measurement here. AsyncChannel has some inherent complexities and state machines just surface the various transition edges way better.

[FranzBusch]

Left some comments here! In general, I like the switch to a state machine it makes it very clear what is going on

[twittemb]

Left some comments here! In general, I like the switch to a state machine it makes it very clear what is going on

Thanks for your review … I‘ll check it ASAP

[twittemb]

Left some comments here! In general, I like the switch to a state machine it makes it very clear what is going on

Hi @FranzBusch I've pushed a version with pretty much all the comments addressed.

[FranzBusch]

Looks really good already. Left some more comments inline but we are getting close here! Thanks for all the work

[FranzBusch]

I don't think that this is aligned with the rule that next() should throw an error once and after that it should return nil. This code right now throws the same error over and over again. Implementing the expected behaviour is going to get interesting since it means that we have to keep track of each iterator and what they saw. Maybe we can achieve this by creating a single ID for an iterator that we keep inside the iterator and use for every next() call on the state machine instead of generating an ID per next call.

[twittemb]

It is not exactly what happens. When the state is .finished(.failure) and there are some suspended consumers then they all receive the failure and the state is set to .terminated(.finished). The subsequent call to next will receive “nil”.

[twittemb]

@FranzBusch, @phausler before making any changes I'd like to be sure of the expected behaviour. I think that what I have now is aligned with the previous implementation of AsyncChannel.

[FranzBusch]

Maybe I am misreading the code here as well. If the behaviour is the same then it should be good!

[twittemb]

this is the unit test that checks that:

  func test_asyncThrowingChannel_resumes_producers_and_discards_additional_elements_when_fail_is_called() async throws {
 // Given: an AsyncThrowingChannel
 let sut = AsyncThrowingChannel<Int, Error>()

 // Given: 2 suspended send operations
 let task1 = Task {
   await sut.send(1)
 }

 let task2 = Task {
   await sut.send(2)
 }

 // When: failing the channel
 sut.fail(Failure())

 // Then: the send operations are resumed
 _ = await (task1.value, task2.value)

 // When: sending an extra value
 await sut.send(3)

 // Then: the send operation is resumed
 // Then: the iteration is resumed with a failure
 var collected = [Int]()
 do {
   for try await element in sut {
     collected.append(element)
   }
 } catch {
   XCTAssertTrue(collected.isEmpty)
   XCTAssertEqual(error as? Failure, Failure())
 }

 // When: requesting a next value
 var iterator = sut.makeAsyncIterator()
 let pastFailure = try await iterator.next()

 // Then: the past failure is nil
 XCTAssertNil(pastFailure)
}

[twittemb]

yep

func test_asyncThrowingChannel_resumes_consumers_when_fail_is_called() async throws {
  // Given: an AsyncThrowingChannel
  let sut = AsyncThrowingChannel<Int, Error>()

  // Given: 2 suspended iterations
  let task1 = Task<Int?, Error> {
    var iterator = sut.makeAsyncIterator()
    return try await iterator.next()
  }

  let task2 = Task<Int?, Error> {
    var iterator = sut.makeAsyncIterator()
    return try await iterator.next()
  }

  // When: failing the channel
  sut.fail(Failure())

  // Then: the iterations are resumed with the error
  do {
    _ = try await (task1.value, task2.value)
  } catch {
    XCTAssertEqual(error as? Failure, Failure())
  }

  // When: requesting a next value
  var iterator = sut.makeAsyncIterator()
  let pastFailure = try await iterator.next()

  // Then: the past failure is nil
  XCTAssertNil(pastFailure)
}

[UPDATE] although I could improve it a bit by asserting both task1 and task2 will fail independently ! hold on ...

[FranzBusch]

Hm that's not really testing the same. I would like to see that test but with the same iterator calling next twice. Like this

  let task1 = Task<Int?, Error> {
    var iterator = sut.makeAsyncIterator()
    try await iterator.next() // Need to catch the values of both next and return them in an array to assert
    try await iterator.next()
  }

[twittemb]

ok.

Is that ok?

  func test_asyncThrowingChannel_resumes_consumer_when_fail_is_called() async throws {
    // Given: an AsyncThrowingChannel
    let sut = AsyncThrowingChannel<Int, Error>()

    // Given: suspended iteration
    let task = Task<Int?, Error> {
      var iterator = sut.makeAsyncIterator()

      do {
        _ = try await iterator.next()
      } catch {
        XCTAssertEqual(error as? Failure, Failure())
      }

      return try await iterator.next()
    }

    // When: failing the channel
    sut.fail(Failure())

    // Then: the iterations are resumed with the error and the next element is nil
    do {
      let collected = try await task.value
      XCTAssertNil(collected)
    } catch {
      XCTFail("The task should not fail, the past failure element should be nil, not a failure.")
    }
  }

[FranzBusch]

I think almost, just one slight change

  func test_asyncThrowingChannel_resumes_consumer_when_fail_is_called() async throws {
    // Given: an AsyncThrowingChannel
    let sut = AsyncThrowingChannel<Int, Error>()

    // Given: suspended iteration
    let task = Task<Int?, Error> {
      var iterator = sut.makeAsyncIterator()

      do {
        _ = try await iterator.next()
        XCTFail("We expect the above call to throw")
      } catch {
        XCTAssertEqual(error as? Failure, Failure())
      }

      return try await iterator.next()
    }

    // When: failing the channel
    sut.fail(Failure())

    // Then: the iterations are resumed with the error and the next element is nil
    do {
      let collected = try await task.value
      XCTAssertNil(collected)
    } catch {
      XCTFail("The task should not fail, the past failure element should be nil, not a failure.")
    }
  }

Would be great if you could add that test

[twittemb]

I had to do a few tweaks in the state machine to make it pass ... nice catch.
I've pushed the updated version.

[twittemb]

Looks really good already. Left some more comments inline but we are getting close here! Thanks for all the work

Thanks for the review. I'll address the comments later today or tomorrow.

[phausler]

Overall this looks pretty good; let me know when you are ready to land it.

[twittemb]

Overall this looks pretty good; let me know when you are ready to land it.

Thanks for the review, I think I'll push something tomorrow.

[twittemb]

Overall this looks pretty good; let me know when you are ready to land it.

I've addressed all the comments (minus one for the throwing behaviour).

In the end I can confirm a 20% increase in perfs + a more reliable implementation (the current one crashes from time to time during the throughput measurement)

[FranzBusch]

One last nit, but looks good to me now! Thanks for all the work!

[twittemb]

One last nit, but looks good to me now! Thanks for all the work!

Thanks a lot for your review.

@phausler ready to merge 👍.