Add some test coverage for TSFFutures

Author: dmbryson

Package.swift

@@ -32,6 +32,12 @@ let package = Package(
                 "NIO", "SwiftToolsSupport-auto",
             ]
         ),
+        .testTarget(
+            name: "TSFFuturesTests",
+            dependencies: [
+                "TSFFutures",
+            ]
+        ),
         .target(
             name: "TSFUtility",
             dependencies: [

Sources/TSFFutures/Canceller.swift

@@ -83,6 +83,7 @@ public final class LLBCanceller {
 
         guard finalReason_ == nil else {
             // Already cancelled or abandoned.
+            mutex_.unlock()
             return
         }
 

Sources/TSFFutures/FutureDeduplicator.swift

@@ -148,4 +148,92 @@ public class LLBFutureDeduplicator<Key: Hashable, Value> {
 
         return promise.futureResult
     }
+
+    @inlinable
+    public func values(for keys: [Key], with resolver: ([Key]) -> LLBFuture<[Value]>) -> [LLBFuture<Value>] {
+        let (rv, created, promises): ([LLBFuture<Value>], [Key], [LLBPromise<Value>]) = lock.withLock {
+            let now = DispatchTime.now()
+
+            var rv = [LLBFuture<Value>]()
+            var created = [Key]()
+            var promises = [LLBPromise<Value>]()
+            let maxResultCount = keys.count
+            rv.reserveCapacity(maxResultCount)
+            created.reserveCapacity(maxResultCount)
+            promises.reserveCapacity(maxResultCount)
+
+            for key in keys {
+                if let valueFuture = lockedCacheGet(key: key) {
+                    rv.append(valueFuture)
+                    continue
+                }
+
+                if let (resultPromise, expires) = inFlightRequests[key] {
+                    if expires >= now {
+                        rv.append(resultPromise.futureResult)
+                        continue
+                    }
+                }
+
+                // Stop deduplicating requests after a long-ish timeout,
+                // just in case there is a particularly long CAS lag.
+                let expires = now + partialResultExpiration
+
+                // Cache for future use.
+                let promise = self.group.next().makePromise(of: Value.self)
+                self.inFlightRequests[key] = (promise, expires)
+                rv.append(promise.futureResult)
+                created.append(key)
+                promises.append(promise)
+            }
+
+            return (rv, created, promises)
+        }
+
+        // If the promise was created, fetch from the database.
+        guard created.count > 0 else {
+            return rv
+        }
+
+        resolver(created).map { results in
+            self.lock.withLockVoid {
+                for idx in 0..<created.count {
+                    let key = created[idx]
+                    let promise = promises[idx]
+                    guard self.inFlightRequests[key]?.result.futureResult === promise.futureResult else {
+                        continue
+                    }
+                    // Resolved, done.
+                    self.inFlightRequests[key] = nil
+                }
+                for (key, promise) in zip(created, promises) {
+                    self.lockedCacheSet(key, promise.futureResult)
+                }
+            }
+            for (promise, result) in zip(promises, results) {
+                promise.succeed(result)
+            }
+        }.whenFailure { error in
+            let now = DispatchTime.now()
+            let expires: DispatchTimeInterval? = self.expirationInterval(error)
+            self.lock.withLockVoid {
+                for (key, promise) in zip(created, promises) {
+                    guard self.inFlightRequests[key]?.result.futureResult === promise.futureResult else {
+                        continue
+                    }
+                    if let interval = expires {
+                        self.inFlightRequests[key] = (promise, now + interval)
+                    } else {
+                        self.inFlightRequests[key] = nil
+                    }
+                }
+            }
+
+            for promise in promises {
+                promise.fail(error)
+            }
+        }
+
+        return rv
+    }
 }

Sources/TSFFutures/OrderManager.swift

@@ -0,0 +1,181 @@
+// This source file is part of the Swift.org open source project
+//
+// Copyright (c) 2020 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See http://swift.org/LICENSE.txt for license information
+// See http://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
+
+import Dispatch
+
+import NIO
+import NIOConcurrencyHelpers
+
+
+/// The `OrderManager` allows explicitly specify dependencies between
+/// various callbacks. This is necessary to avoid or induce race conditions
+/// in otherwise timing-dependent code, making such code deterministic.
+/// The semantics is as follows: the `OrderManager` invokes callbacks
+/// specified as arguments to the `order(_:_)` functions starting with 1.
+/// The callbacks of order `n` are guaranteed to run to completion before
+/// callbacks of order `n+1` are run. If there's a gap in the callbacks order
+/// sequence, the callbacks are suspended until the missing callback is
+/// registered, `reset()` is called, or global timeout occurs.
+/// In addition to that, `reset()` restarts the global timeout.
+///
+/// Example:
+///
+///     let manager = OrderManager(on: ...)
+///     manager.order(3, { print("3") })
+///     manager.order(1, { print("1") })
+///     manager.order(2, { print("2") })
+///     try manager.order(4).wait()
+///
+/// The following will be printed out:
+///
+///     1
+///     2
+///     3
+///
+
+public class LLBOrderManager {
+
+    // A safety timer, not to be exceeded.
+    private let cancelTimer = DispatchSource.makeTimerSource()
+    private let timeout: DispatchTimeInterval
+
+    private typealias WaitListElement = (order: Int, promise: LLBPromise<Void>, file: String, line: Int)
+    private let lock = NIOConcurrencyHelpers.Lock()
+    private var waitlist = [WaitListElement]()
+    private var nextToRun = 1
+
+    private var eventLoop: EventLoop {
+        lock.withLock {
+            switch groupDesignator {
+            case let .managedGroup(group):
+                return group.next()
+            case let .externallySuppliedGroup(group):
+                return group.next()
+            }
+        }
+    }
+
+    private enum GroupDesignator {
+    case managedGroup(LLBFuturesDispatchGroup)
+    case externallySuppliedGroup(LLBFuturesDispatchGroup)
+    }
+    private var groupDesignator: GroupDesignator
+
+    public enum Error: Swift.Error {
+    case orderManagerReset(file: String, line: Int)
+    }
+
+    public init(timeout: DispatchTimeInterval = .seconds(60)) {
+        let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+        self.groupDesignator = GroupDesignator.managedGroup(group)
+        self.timeout = timeout
+        restartInactivityTimer()
+        cancelTimer.setEventHandler { [weak self] in
+            guard let self = self else { return }
+            _ = self.reset()
+            self.cancelTimer.cancel()
+        }
+        cancelTimer.resume()
+    }
+
+    public init(on loop: EventLoop, timeout: DispatchTimeInterval = .seconds(60)) {
+        self.groupDesignator = GroupDesignator.externallySuppliedGroup(loop)
+        self.timeout = timeout
+        restartInactivityTimer()
+        cancelTimer.setEventHandler { [weak self] in
+            guard let self = self else { return }
+            _ = self.reset()
+            self.cancelTimer.cancel()
+        }
+        cancelTimer.resume()
+    }
+
+    private func restartInactivityTimer() {
+        cancelTimer.schedule(deadline: DispatchTime.now() + timeout, repeating: .never)
+
+    }
+
+    /// Run a specified callback in a particular order.
+    @discardableResult
+    public func order<T>(_ n: Int, file: String = #file, line: Int = #line, _ callback: @escaping () throws -> T) -> EventLoopFuture<T> {
+        let promise = eventLoop.makePromise(of: Void.self)
+
+        lock.withLockVoid {
+            waitlist.append((order: n, promise: promise, file: file, line: line))
+        }
+
+        let future = promise.futureResult.flatMapThrowing {
+            try callback()
+        }
+
+        future.whenComplete { _ in
+            self.lock.withLockVoid {
+                if n == self.nextToRun {
+                    self.nextToRun += 1
+                }
+            }
+            self.unblockWaiters()
+        }
+
+        unblockWaiters()
+        return future
+    }
+
+    @discardableResult
+    public func order(_ n: Int, file: String = #file, line: Int = #line) -> EventLoopFuture<Void> {
+        return order(n, file: file, line: line, {})
+    }
+
+    private func unblockWaiters() {
+        let wakeup: [EventLoopPromise<Void>] = lock.withLock {
+            let wakeupPromises = waitlist
+                                .filter({$0.order <= nextToRun})
+                                .map({$0.promise})
+            waitlist = waitlist.filter({$0.order > nextToRun})
+            return wakeupPromises
+        }
+        wakeup.forEach { $0.succeed(()) }
+    }
+
+    /// Fail all ordered callbacks. Not calling the callback functions
+    /// specified as argument to order(_:_), but failing the outcome.
+    public func reset(file: String = #file, line: Int = #line) -> EventLoopFuture<Void> {
+        restartInactivityTimer()
+        let lock = self.lock
+        let toCancel: [WaitListElement] = lock.withLock {
+            let cancelList = waitlist
+            waitlist = []
+            nextToRun = Int.max
+            return cancelList
+        }
+        let error = Error.orderManagerReset(file: file, line: line)
+        let futures: [EventLoopFuture<Void>] = toCancel.sorted(by: {$0.order < $1.order}).map {
+            $0.promise.fail(error)
+            return $0.promise.futureResult.flatMapErrorThrowing { _ in () }
+        }
+
+        return EventLoopFuture.whenAllSucceed(futures, on: eventLoop).map { [weak self] _ in
+            guard let self = self else { return }
+            lock.withLockVoid {
+                assert(self.waitlist.isEmpty)
+                self.nextToRun = 1
+            }
+        }
+    }
+
+    deinit {
+        let designator = groupDesignator
+        reset().whenSuccess {
+            guard case let .managedGroup(group) = designator else { return }
+            let q = DispatchQueue(label: "tsf.OrderManager")
+            q.async { try! group.syncShutdownGracefully() }
+        }
+        cancelTimer.setEventHandler { }
+        cancelTimer.cancel()
+    }
+}