FlatMap

[epatey]

I'm taking a stab at FlatMap. In support of operators that need serialization of sending values from multiple upstream/child subscriptions, I've implemented a simple/fast SerializedWorkQueue that adheres to the Combine threading rules and avoids unnecessary context switches.

Remaining work:

• Clean up and ensure all completion/failures paths are correct.
  • Clean up and complete with .failed if either upstream or any children fail.
  • Complete with .finished only after upstream and all children have completed (i.e. add refcounting of active subscriptions) AND all buffer values have been sent.
• Figure out and add support for maxPublishers
• Figure out non-trivial handling of demand for child subscriptions and implement it.
  • Basic child demand scenarios work
  • Add support for downstream demand going from .unlimited to .max(x)
  • Add support for downstream demand going from .max(x) to .unlimited
• Add unit tests
• Factor out common base class from Inner class to be shared with all merge like operators.

[epatey]

I'm now reasoning through how to do demand management for the child subscriptions. (demand for upstream was trivial since it's basically set to maxPublishers and never changed)

Experimenting with the Apple code, it seems to request 1 from all children, and then make some decisions about what to do next when it receives a value from any child.

I think it will be necessarily complex to deal with buffering of one extra value per child subscription. Imagine the following flow:

• downstream requests initial demand of .max(1) of the flatMap
• upstream sends 3 values resulting in subscription to 3 children publishers
• each of those children is given a demand of .max(1)
• child[0] sends a value. flatMap passes it on to downstream which satisfies downstream's demand
• child[1] sends a value

At this point, flatMap has to buffer that value so that it can be passed to downstream if a later demand request is made.

continuing to experiment with Apple code, but I think it's unfortunately non-trivial

[broadwaylamb]

@epatey hmm, that seems… unexpected. Yeah, FlatMap is probably the most nontrivial part of Combine.

[epatey]

So, here's a unit test (that passes in apple combine mode), that helps explain more precisely the behavior. It's essentially that FlatMap puts .max(1) demand on all children until it has one buffered value ready to send downstream.

func testChildDemand() {
    let upstreamPublisher = PassthroughSubject<Void, Never>()

    var childDemand: Subscribers.Demand?
    let childSubscription = CustomSubscription(onRequest: { childDemand = $0 })
    let childPublisher = CustomPublisherBase<Int, Never>(
        subscription: childSubscription)

    let flatMap = upstreamPublisher.flatMap { _ in childPublisher }

    var downstreamSubscription: Subscription?
    let downstreamSubscriber = TrackingSubscriberBase<Int, Never>(
        receiveSubscription: {
            downstreamSubscription = $0
            $0.request(.max(2))
    })

    flatMap.subscribe(downstreamSubscriber)

    upstreamPublisher.send(())

    // It seems that Apple creates enough demand on each child such that it can
    // have one extra/buffered value after the downstream demand is satisfied.
    // A demand of 1 on each child is created
    XCTAssertEqual(childDemand, .max(1))

    // Downstream demand is 2, so:
    //  - this value gets sent
    //  - downstream demand goes down to 1
    //  - child is asked for 1 more
    XCTAssertEqual(childPublisher.send(666), .max(1))
    XCTAssertEqual(downstreamSubscriber.history, [.subscription("FlatMap"),
                                                  .value(666)])

    // Downstream demand is 1, so:
    //  - this value gets sent
    //  - downstream demand goes down to 0, but still need a buffered value
    //  - child is asked for 1 more
    XCTAssertEqual(childPublisher.send(777), .max(1))
    XCTAssertEqual(downstreamSubscriber.history, [.subscription("FlatMap"),
                                                  .value(666),
                                                  .value(777)])

    // Downstream demand is 0, so:
    //  - this value is buffered and NOT sent
    //  - downstream demand is 0 and there's a buffered value
    //  - child is asked for 0 more
    XCTAssertEqual(childPublisher.send(888), .max(0))
    XCTAssertEqual(downstreamSubscriber.history, [.subscription("FlatMap"),
                                                  .value(666),
                                                  .value(777)])

    childDemand = nil
    downstreamSubscription!.request(.max(2))
    XCTAssertEqual(childDemand, .max(1))
    XCTAssertEqual(downstreamSubscriber.history, [.subscription("FlatMap"),
                                                  .value(666),
                                                  .value(777),
                                                  .value(888)])
}

[codecov]

Codecov Report

Merging #45 into master will increase coverage by 0.17%.
The diff coverage is 98.42%.

@@            Coverage Diff             @@
##           master      #45      +/-   ##
==========================================
+ Coverage   97.74%   97.91%   +0.17%     
==========================================
  Files          43       44       +1     
  Lines        1727     1918     +191     
==========================================
+ Hits         1688     1878     +190     
- Misses         39       40       +1

Impacted Files	Coverage Δ
...es/OpenCombine/Publishers/Publishers.FlatMap.swift	98.42% <98.42%> (ø)
Sources/OpenCombine/Subject.swift	100% <0%> (+100%)	⬆️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 07c7a98...305b420. Read the comment docs.

[epatey]

Back pressure/demand makes things WAY WAY more complex. I really wonder if it was worth it vs Rx's decision to not have the feature. I've honestly never felt the need for it, but maybe I'm missing something. Getting close to complete support, though.

[epatey]

@broadwaylamb , I think I'm finally ready for a review. Here are a couple of thoughts:

• I factored out the code that deals with all of the complexity of the child subscriptions. By child subscriptions, I mean subscriptions of things that are not Upstream. This is where all of the complex demand, buffering logic is.
• I'm not in love with the name, ChildManagingInnerBase, but I wanted to start with a name that was precise. Feel free to propose something else.
• Although so far only FlatMap's Inner derives from that base class, I built it with Merge in mind. It's interesting that Merge is different than FlatMap because there is no Upstream for Merge.

[epatey]

I'm thinking about iterating on ChildManagingInnerBase and recasting it as MultiUpstreamBase. In the case of FlatMap, I had been thinking about the upstream publisher and the child publishers. I think it would be a bit simpler if I just thought of them all as upstreams. Subclasses, i.e. FlatMap.Inner, could easily keep track of special upstreams, but the base class really doesn't need to know or care about the distinction.

The challenge is that, in the case of FlatMap or CombineLatest, the different upstreams can be of completely different types. That's why I'm still just thinking about it. Suggestions are welcome.

[epatey]

Ok, so my thinking has evolved a little. I think what I built with ChildManagingInnerBase is a class to help operators that deal with a dynamic list of multiple upstream publishers. By that I mean where the list is not fixed at compile time. It's possible that FlatMap is the only operator with that characteristic. Merge and CombineLatest have a fixed number of children.

If this proves to be correct, I'll undo the factoring I did since that base class adds complexity but no value.

[broadwaylamb]

Thank you for your hard work, Eric!
You've done a really amazing job. I like that you've put a lot of comments, since the implementation is nontrivial indeed. I also like the tests. I have some suggestions and questions though.

I'd like to ask you to not force-push for now as it makes reviewing a little bit harder. After it's complete, you can squash the commits to keep the history clear.

[OpenCombineBot]

LGTM

Generated by 🚫 Danger Swift against 305b420

[epatey]

@broadwaylamb, I've gotten rid of the base class in my last commit. As I feared, there really are no other operators that have to manage a dynamic set of child subscriptions. Unfortunately, that commit moved a whole lot of code, so you may want to review the previous commits individually before looking at that big one to make understanding my reaction to your feedback easier.

[epatey]

@broadwaylamb, let me know if/when you'd like me to rebase, squash and merge this - or if you'd like additional changes.

[broadwaylamb]

XCTestManifests.swift is not needed anymore, can you synchronize with master and remove this file?

[epatey]

@broadwaylamb, it looks like the swift 5.0 build for this pr is failing, but the only thing I see in the logs here is a problem with the swiftlint configuration. I'm guessing that's the reason it's failing, but I'm not sure. Any thoughts?

[broadwaylamb]

@epatey oh. My bad, I removed XCTestManifests.swift, but forgot to remove the Danger rule for that file. Should be fine now.

[broadwaylamb]

Thank you so much for this PR!