fix EventLoopFuture.and's serious threading issues

[weissi]

another 🙈
alternative in #176

Motivation:

EventLoopFuture.and had serious threading issues if the EventLoops
weren't the same.

Modifications:

Fixed the threading issues and tested them properly.

Result:

Hopefully and and andAll now don't crash if you use them across
EventLoops.

[Lukasa]

This fix feels wrong: the use of a Lock here in particular seems like a sign that this approach is not really right. We really should be able to fix this using the semantics of EventLoopFuture/EventLoopPromise alone.

f1.and(f2) has the semantic of producing a third future, f3, tied to the loop of f1, that returns either a tuple of the result of f1 and f2 or the first error that occurs. Given that f3 necessarily fires on the loop of f1, any callback result of f1 is definitionally thread-safe so we don't need to worry about it. The problem is the callback on f2, which can execute on a parallel thread

The simplest way to handle that is probably to cascade the result of f2 onto an intermediate future that belongs to f1's event loop, and handle the callback there. That will use the standard thread-hopping semantics of EventLoopFuture to process the callback, at the cost of an extra few allocations of the new promise.

Another option is to eventLoop.execute the callbacks on f2, avoiding the extra allocation of a promise and future. Those callbacks then appropriately execute on the right thread.

If we use the intermediate promise for a minute, I think our code can become this:

    public func and<U>(_ other: EventLoopFuture<U>, file: StaticString = #file, line: UInt = #line) -> EventLoopFuture<(T,U)> {
        let promise = EventLoopPromise<(T,U)>(eventLoop: eventLoop, file: file, line: line)
        var tvalue: T?
        var uvalue: U?

        let otherFuture: EventLoopFuture<U>
        if other.eventLoop.inEventLoop {
            otherFuture = other
        } else {
            let threadHoppingPromise: EventLoopPromise<U> = self.eventLoop.newPromise()
            other.cascade(promise: threadHoppingPromise)
            otherFuture = threadHoppingPromise.futureResult
        }

        _whenComplete { () -> CallbackList in
            switch self.value! {
            case .failure(let error):
                return promise._setValue(value: .failure(error))
            case .success(let t):
                if let u = uvalue {
                    return promise._setValue(value: .success((t, u)))
                } else {
                    tvalue = t
                }
            }
            return CallbackList()
        }

        otherFuture._whenComplete { () -> CallbackList in
            switch otherFuture.value! {
            case .failure(let error):
                return promise._setValue(value: .failure(error))
            case .success(let u):
                if let t = tvalue {
                    return promise._setValue(value: .success((t, u)))
                } else {
                    uvalue = u
                }
            }
            return CallbackList()
        }

        return promise.futureResult
    }

This is not actually a sufficient fix because, insanely, this uses the internal _whenComplete and _setValue methods, which seems utterly pointless, so we'd probably want to refactor completely to use the actual public methods on this type, which vastly simplifies the code:

    public func and<U>(_ other: EventLoopFuture<U>, file: StaticString = #file, line: UInt = #line) -> EventLoopFuture<(T,U)> {
        let promise = EventLoopPromise<(T,U)>(eventLoop: self.eventLoop, file: file, line: line)
        var tvalue: T?
        var uvalue: U?

        let otherFuture: EventLoopFuture<U>
        if other.eventLoop.inEventLoop {
            otherFuture = other
        } else {
            let threadHoppingPromise: EventLoopPromise<U> = self.eventLoop.newPromise()
            other.cascade(promise: threadHoppingPromise)
            otherFuture = threadHoppingPromise.futureResult
        }

        self.map { t in
            if let u = uvalue {
                promise.succeed(result: (t, u))
            } else {
                tvalue = t
            }
        }.cascadeFailure(promise: promise)

        otherFuture.map { u in
            if let t = tvalue {
                promise.succeed(result: (t, u))
            } else {
                uvalue = u
            }
        }.cascadeFailure(promise: promise)

        return promise.futureResult
    }

This now seems like code we could feasibly debug, doesn't it? Have I missed something here?

[Lukasa]

As a side note, the "thread hopping promise" pattern is not unique to this method, it's present in the AcceptHandler too: it appeared when we refactored it to properly manage its threading logic. I wonder if the utility of this deserves a convenience method: futureResult<T>.futureOn(eventLoop: EventLoop) -> EventLoopFuture<T>.

[weissi]

@Lukasa I totally agree with you, I just wanted to do the smallest working fix. Maybe that was misguided, I can change that if you prefer (you do :) )

[Lukasa]

I definitely prefer changing it. This method is pretty heinous, so now that we're in here messing about with it I think we should burn it back and fix it. I'm working on a PR for the event loop hopping promise real quick, because that's a small patch in its own right.

[normanmaurer]

I agree with @Lukasa here... let us fix completely once and move on ;)

[weissi]

@Lukasa besides that, this

        self.map { t in
            if let u = uvalue {
                promise.succeed(result: (t, u))
            } else {
                tvalue = t
            }
        }.cascadeFailure(promise: promise)

        otherFuture.map { u in
            if let t = tvalue {
                promise.succeed(result: (t, u))
            } else {
                uvalue = u
            }
        }.cascadeFailure(promise: promise)

afaik isn't threading correct either as you read uvalue on self.futureResult.eventLoop but you write it on other.futureResult.eventLoop

EDIT: not quite right, otherFuture is specifically created to be on the right thread, sorry!

[weissi]

proposed alternative in #176

[Lukasa]

Closing in favour of #176.