# Communicating between two concurrent tasks

**URL:** <https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240>\
**Category:** Using Swift\
**Tags:** concurrency\
**Created:** [December 23, 2021, 6:39pm UTC](https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240 "2021-12-23T18:39:11Z")\
**Posts on this page:** 10\
**Page:** 1

<div class="post-metadata">

**Author:** ![ole](https://sea2.discourse-cdn.com/swift/user_avatar/forums.swift.org/ole/32/18720_2.png) [@ole](https://forums.swift.org/u/ole)\
**Post date:** [December 23, 2021, 6:39pm UTC](https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240/1 "2021-12-23T18:39:11Z")

</div>

[The full code is here](https://gist.github.com/ole/2ebb2a83c65b5a243ab33eacf4d96867), ready to be pasted into an Xcode playground.

## Setup

Suppose I have an actor that protects access to a buffer of values (simplified example):

```swift
actor Buffer {
  var elements: [Int] = []

  func enqueue(_ value: Int) {
    elements.append(value)
  }

  func dequeue() async -> Int {
    // If the buffer is empty, wait for the next enqueued value
    // ???
  }
}

```

Now I want to have two concurrent tasks, one that fills the buffer and one that empties it:

```swift
let buffer = Buffer()

// Producer task
Task {
  await buffer.enqueue(1)
  try! await Task.sleep(nanoseconds: 400_000_000)
  await buffer.enqueue(2)
  await buffer.enqueue(3)
}

// Consumer task
Task {
  while true {
    let next = await buffer.dequeue()
    print(next)
    try! await Task.sleep(nanoseconds: 100_000_000)
  }
}

```

(For simplicity, I omitted API to end the stream. As is, the consumer task will never complete.)

## Question

My question is how to implement the `dequeue` method with my desired semantics:

- If there is at least one element in the buffer, remove and return it.
- If the buffer is currently empty, suspend until another value has been enqueued.

## Approach 1: spin in a loop

We could do this by spinning in an infinite loop (taking care to yield to give the other task a chance to call `enqueue`):

```swift
  func dequeue() async -> Int {
    while true {
      if !elements.isEmpty {
        return elements.removeFirst()
      } else {
        await Task.yield()
      }
    }
  }

```

That's obviously not ideal.

## Approach 2: use a CheckedContinuation for signaling

What we need is a way for the `enqueue` method to signal a suspended `dequeue` method that it can resume. My idea was to use a `CheckedContinuation` for this:

- When `dequeue` needs to suspend, it creates a continuation and stores it in a property.
- When `enqueue` runs, it resumes the continuation, thus resuming `dequeue`.

```swift
actor Buffer {
  var elements: [Int] = []
  private var isNotEmpty: CheckedContinuation<Void, Never>? = nil

  deinit {
    // TODO: If the continuation is not nil,
    // resume it by throwing cancellation error?
  }

  func enqueue(_ value: Int) {
    elements.append(value)
    // Signal dequeue if needed.
    isNotEmpty?.resume()
    isNotEmpty = nil
  }

  func dequeue() async -> Int {
    while true {
      if !elements.isEmpty {
        return elements.removeFirst()
      } else {
        // Suspend until another element is enqueued.
        print("dequeue waiting")
        await withCheckedContinuation { cont in
          self.isNotEmpty = cont
        }
      }
    }
  }
}

```

In a more complex example, you could add a maximum buffer size to the `Buffer` actor and use another continuation to have `enqueue` suspend when the buffer is full.

## Is this a valid use case for continuations?

Is this kind of intra-task communication — like a condition variable — a valid use case for continuations? Or are continuations supposed to only be used for interop with legacy code?

[The full code is here](https://gist.github.com/ole/2ebb2a83c65b5a243ab33eacf4d96867), ready to be pasted into an Xcode playground.

---

<div class="post-metadata">

**Author:** ![Alejandro\_Martinez](https://sea2.discourse-cdn.com/swift/user_avatar/forums.swift.org/alejandro_martinez/32/821_2.png) [@Alejandro\_Martinez](https://forums.swift.org/u/Alejandro_Martinez)\
**Post date:** [December 23, 2021, 7:43pm UTC](https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240/2 "2021-12-23T19:43:09Z")

</div>

I actually had a very similar use case to this. A "Storage" type that I wanted to query via a key and get a value back, but the values were filled by another producer. Funny enough I also started with a spin loop but quickly dismissed it when I realised that task.yield doesn't stop the loop from going at maximum speed ^^

What I ended up is making a `AsyncFuture` type that basically wraps a continuation. With that the `dequeue` can await to get a value from the "future" while the enqueue can add the value to it. It's kind of a Combine PasstroughSubject but just using Swift concurrency.

I see that the difference is that you just used the continuation to signal while keeping your own storage while I actually made my wrapper "future" hold and pass the value trough.

---

<div class="post-metadata">

**Author:** ![John\_McCall](https://sea2.discourse-cdn.com/swift/user_avatar/forums.swift.org/john_mccall/32/1351_2.png) [@John\_McCall](https://forums.swift.org/u/John_McCall)\
**Post date:** [December 23, 2021, 7:46pm UTC](https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240/3 "2021-12-23T19:46:59Z")

</div>

Have you considered using `AsyncStream`, or at least checking out the implementation? It does exactly this kind of continuation management.

---

<div class="post-metadata">

**Author:** ![Alejandro\_Martinez](https://sea2.discourse-cdn.com/swift/user_avatar/forums.swift.org/alejandro_martinez/32/821_2.png) [@Alejandro\_Martinez](https://forums.swift.org/u/Alejandro_Martinez)\
**Post date:** [December 23, 2021, 7:50pm UTC](https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240/4 "2021-12-23T19:50:38Z")

</div>

I personally did and based my implementation on that (\<3 open source!).

I wondered how safe it is to make the `AsyncStream.Continuation` escape the init. (to be honest same for the normal continuation API).

---

<div class="post-metadata">

**Author:** ![tkrajacic](https://sea2.discourse-cdn.com/swift/user_avatar/forums.swift.org/tkrajacic/32/1173_2.png) [@tkrajacic](https://forums.swift.org/u/tkrajacic)\
**Post date:** [December 23, 2021, 7:51pm UTC](https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240/5 "2021-12-23T19:51:08Z")

</div>

How about something like this:

```swift
final class Buffer2 {
    private var stream: AsyncStream<Int>!
    private var continuation: AsyncStream<Int>.Continuation!
    private var iterator: AsyncStream<Int>.Iterator!
    
    public init(initialValues: [Int] = []) {
        self.stream = .init { self.continuation = $0 }
        self.iterator = self.stream.makeAsyncIterator()
        for value in initialValues { self.continuation.yield(value) }
    }
    
    public func enqueue(_ value: Int) async {
        self.continuation.yield(value)
    }
    
    public func dequeue() async throws -> Int {
        guard let value = await self.iterator.next() else { throw CancellationError() }
        return value
    }
    
    public func cancel() {
        continuation.finish()
    }
}

```

You don't see when you are waiting though, and you can't make it an actor, as the iterator is being mutated in the public method.

---

<div class="post-metadata">

**Author:** ![taylorswift](https://sea2.discourse-cdn.com/swift/user_avatar/forums.swift.org/taylorswift/32/43413_2.png) [@taylorswift](https://forums.swift.org/u/taylorswift)\
**Post date:** [December 23, 2021, 11:03pm UTC](https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240/6 "2021-12-23T23:03:27Z")

</div>

> [@Alejandro\_Martinez](#):
>
> I wondered how safe it is to make the `AsyncStream.Continuation` escape the init. (to be honest same for the normal continuation API).

`AsyncStream<T>.Continuation` is supposed to escape the `init`. the typical use case is to kick off the (concurrent) producer task from inside the `init`, while the consumer operates from outside the `init`. i suspect that the intention of this API was to prevent a potential deadlock condition when you start iterating the stream before starting up the producer task. however, (after having written over 100k lines of `async`/`await` this year) i am firmly convinced this was a mistake and does not actually provide much value.

moreover, something seems to have been broken in the typechecker for the past month or so where `AsyncStream` instances cannot be constructed in the first place. `AsyncStream` used to work well though.

---

<div class="post-metadata">

**Author:** ![ole](https://sea2.discourse-cdn.com/swift/user_avatar/forums.swift.org/ole/32/18720_2.png) [@ole](https://forums.swift.org/u/ole)\
**Post date:** [December 24, 2021, 4:43pm UTC](https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240/7 "2021-12-24T16:43:22Z")

</div>

> [@John\_McCall](#):
>
> Have you considered using `AsyncStream` , or at least checking out the implementation? It does exactly this kind of continuation management.

I did consider it briefly, but from reading the docs I didn't get the impression it was the right tool for this job. In particular, I deemed it "too big" for my use case when a plain continuation also does the job. Thank you, I'll check out how it's implemented.

---

<div class="post-metadata">

**Author:** ![ole](https://sea2.discourse-cdn.com/swift/user_avatar/forums.swift.org/ole/32/18720_2.png) [@ole](https://forums.swift.org/u/ole)\
**Post date:** [December 24, 2021, 4:46pm UTC](https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240/8 "2021-12-24T16:46:26Z")

</div>

> [@Alejandro\_Martinez](#):
>
> I wondered how safe it is to make the `AsyncStream.Continuation` escape the init. (to be honest same for the normal continuation API).

"Normal" continuations are also allowed/expected to escape, [confirmed here by John](https://forums.swift.org/t/unlike-other-continuations-in-swift-asyncstream-continuation-supports-escaping/53254/3).

---

<div class="post-metadata">

**Author:** ![taylorswift](https://sea2.discourse-cdn.com/swift/user_avatar/forums.swift.org/taylorswift/32/43413_2.png) [@taylorswift](https://forums.swift.org/u/taylorswift)\
**Post date:** [December 24, 2021, 8:28pm UTC](https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240/9 "2021-12-24T20:28:50Z")

</div>

> [@ole](#):
>
> In particular, I deemed it "too big" for my use case when a plain continuation also does the job. Thank you, I'll check out how it's implemented.

anybody know how much overhead the `AsyncStream` implementation has? personally, i have found great use of `AsyncStream<Void>` with a buffering capacity of `1` as a signaling mechanism and i’m wondering what the performance implications of that are…

---

<div class="post-metadata">

**Author:** ![Alejandro\_Martinez](https://sea2.discourse-cdn.com/swift/user_avatar/forums.swift.org/alejandro_martinez/32/821_2.png) [@Alejandro\_Martinez](https://forums.swift.org/u/Alejandro_Martinez)\
**Post date:** [December 28, 2021, 7:41pm UTC](https://forums.swift.org/t/communicating-between-two-concurrent-tasks/54240/10 "2021-12-28T19:41:28Z")

</div>

For completeness I'm leaving a link here to what I was mentioning the other day

> [@Alejandro\_Martinez](#):
>
> What I ended up is making a `AsyncFuture` type that basically wraps a continuation.

which you can find on [github alexito4/AsyncChannel](https://github.com/alexito4/AsyncChannel). There is also a blog post linked.

But I'm still curious about

> [@taylorswift](#):
>
> anybody know how much overhead the `AsyncStream` implementation has?

because is what I will probably use in production systems.