SWIFT-1469 Provide AsyncSequence APIs for monitoring events

[rchhaya]

Need to write local tests

[kmahar]

good start! some minor comments as well as a few design-related questions

[kmahar]

we don't await the result or completion of this Task so I don't think we have any guarantee it has finished executing and making all of its assertions by the time the test finishes.
one suggestion is to have the task return i and await and assert on that value below. something like:

let eventsTask = Task { 
    // ...
    return i
}
try await client.db("admin").runCommand(["ping": 1])
let taskResult = try await eventTask.result
expect(taskResult).to(equal(4))

and assert it's the right value.

[rchhaya]

loop never exits so we never return. All relates to the fact that we dont save the continuation, which not sure how to do given the current initialization. Preliminarily pushing all minor commits and will look at this

[kmahar]

can we give this a more descriptive name like eventTypes?

I think the thread sanitizer is likely to complain about this because the callback appending to this array isn't always necessarily going to execute on the same thread and so there might be data races as a result. one way to fix this would be using a Lock that synchronizes access to the array. you could look at TestCommandMonitor in APMUtils.swift for an example of that.

[rchhaya]

Named it eventTypes. Looked into NioConcurrencyHelpers.Lock that APMUtils.swift uses and I locked the appending to the array (since you cant pass in an async function into the .withLock{ ... }) and left associated comments.

[kmahar]

I think the same concern as above applies where we are not awaiting completion of the Task.

[rchhaya]

yep will address

[kmahar]

given my comment above about never "finishing" the continuation, do we ever break out of this loop / does the task ever actually finish?

[rchhaya]

we do not, need to cache as commented above

[rchhaya]

After caching the continuation, I am able to break the loop and end the task by doing something like
let output = client.commandEventStream() ... output.finish()
Although I can only close the loop at a specific time if I call .finish() inside the loop. For example, if I want to check that 4 events happened, I'll call .finish() via if i == 4. Feels a little bit circular but I can confirm that we break out of the loop at expected time. Not sure if you had a different idea about how to test for ending the task

[kmahar]

few comments now but given the discussed design change to allow multiple streams I'll hold off on re-reviewing the rest as I think a lot of the logic is likely to change

[kmahar]

Since the client currently owns the handler, users will just get the same one next time they invoke the computed property, even if they drop it.

ah right, yeah I think I wrote that comment before realizing in my next comment below that there was only ever one of each stream.

yeah, I think overall as long as there aren't any technical limitations to it, the on-demand/factory approach would be a bit nicer so we can allow multiple streams, and can support stopping/restarting monitoring at will.

[patrickfreed]

Looks good! I mostly have little testing nits here and there

[patrickfreed]

The problem with using AsyncStream<T>.AsyncIterator directly is that if we ever want to change the internals of how CommandEventStream or SDAMEventStream are implemented, we'll still need to somehow create an AsyncStream iterator. If we have a wrapper type, we're free to change the internals of our event stream types as needed.

We also looked into having the event stream types be their own iterators, but that would prevent users from being able to iterate over the same stream from different AsyncIterators for not much benefit, so we decided to just keep it as is.

[patrickfreed]

So this statement doesn't verify the tasks have actually started, since the condition will always be true once this part is hit. To verify they've started, you'll need an atomic counter or something that each task can increment once they start executing.

Also, it's important that this assertion happens before we start generating the events (i.e. before the ping command) so as to ensure each task receives all the events.

[rchhaya]

I see - since the update happens in the loop but outside the task it basically runs synchronously. I think we have to return to using (a much simpler) actor for updating things async. Since the tasks dont kick off in the same order, using atomic counters often ends up updating an outdated version itself (ex. on the 4th iter of the loop, the load() function sees the 3rd iter-version of the counter meaning the counter never sees a value of 5). Using

actor TaskCounter {
     var taskCounter = 0
    
     func incr() {
        taskCounter += 1
     }
 }

seems to work best. Also moved commands to below the check.

[patrickfreed]

One of the neat things about atomics is that you can also increment them atomically. So instead of doing a load + store, you can do atomic.add(1), and it'll avoid the problems you were seeing.

That being said, this actor approach also works, so we can just leave it as-is.

[patrickfreed]

Using a block comment for the example code would be a bit more readable than just inline.
e.g.

for await event in client.commandEventStream() {
    print(event)
}

Ditto for the SDAM event example.

[rchhaya]

Took this opportunity to beef up the docstring and make sure all the elements are in the right sections of the documentation (summary/discussion/etc.). Docstring is now

/**
     * Provides an `AsyncSequence` API for consuming SDAM monitoring events.
     *
     * Example: printing the SDAM events out would be written as
    
     for try await event in client.sdamEventStream() {
             print(event)
     }
     
     * Wrapping in a `Task { ... }` may be desired for asynchronicity.
     * - Returns: An `SDAMEventStream` that implements `AsyncSequence`.
     * - Note: Only the most recent 100 events are stored in the stream.
     */

[patrickfreed]

So in order the code to be formatted as code, you need the triple backticks (```) like in markdown. I think you also need to keep prepending the line with *. For an existing example, check out ClientSession.swift.

[rchhaya]

As per discussion, this method works but used the backticks method for codebase consistency

[rchhaya]

Failed bc waiting on SWIFT-1604 to be merged in

[patrickfreed]

Overall looks good! I just have one minor docstring comment

[patrickfreed]

One of the neat things about atomics is that you can also increment them atomically. So instead of doing a load + store, you can do atomic.add(1), and it'll avoid the problems you were seeing.

That being said, this actor approach also works, so we can just leave it as-is.

[patrickfreed]

So in order the code to be formatted as code, you need the triple backticks (```) like in markdown. I think you also need to keep prepending the line with *. For an existing example, check out ClientSession.swift.

[kmahar]

this is looking really close!! I just have a few minor comments and questions.

[kmahar]

related to my comment above re the lock (not something we need to fix today, but for the future): with the number of places in the code we now guard some value behind a lock and just have to remember that it should only be accessed via the lock, it would be a nice improvement to introduce some generic wrapper type like Locked<T> that stores a value of type T plus its corresponding Lock and has a method that lets you do something like

myLockedThing.withLockedValue { value in
    // ... do something with value
}

The method would acquire the lock, execute your provided closure with the value it guards, then releases the lock. the locked value could be private to the Locked type so it's impossible to access it other than when you hold the lock, which would make it harder for us to forget to acquire the lock.

I filed SWIFT-1612 about this. it could be a nice little ticket to pick up if you end up having any free time waiting for reviews in the future

[rchhaya]

This is a cool idea and increasingly relevant with more async usage in the driver. Def a cool idea to pick up in some downtime.

[kmahar]

why do we need to try? here instead of just try?

[rchhaya]

As per patricks prev comment,

I think we'll want to use a try? here so that a CancellationError isn't throw, otherwise our XCTFail below won't be executed.

Basically if the task gets cancelled, we dont want to throw a cancellation error, we want to hit the continue, then break out of the while !Task.iscancelled() and return false to hit the XCT condition

[kmahar]

oooo ok. can we add a comment saying that this is to ignore CancellationErrors then?

[kmahar]

just a couple formatting suggestions and one very minor comment! LGTM otherwise

[kmahar]

oooo ok. can we add a comment saying that this is to ignore CancellationErrors then?

[kmahar]

lgtm! 🎉

[patrickfreed]

LGTM!!

[isabelatkinson]

looks good! just one question and a small typo fix

[isabelatkinson]

What's the reasoning behind the choice of 100 here? Would there be any benefit in allowing users to set a custom value?

[isabelatkinson]

lgtm after our discussion in standup (mod the one typo fix)!